Help us shape the future of Go! Take the 2026 Go Developer Survey

dorisstreamload

package module
v1.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 2, 2026 License: Apache-2.0 Imports: 31 Imported by: 0

README

Doris Stream Load Go SDK

dorisstreamload is a Go SDK for sending CSV rows or JSON objects to Apache Doris Stream Load.

It keeps the public model small:

  • ModeCSV and ModeJSON
  • string input only
  • Send(...) for one item
  • SendBatch(...) for many items
  • batching, queueing, retry, label polling, callback, handle, and stats are handled inside the SDK

Requirements

Go 1.26.4 or newer.

Install

go get github.com/wushilin/doris_go_stream_load@v1.1.0

Import path:

import dorisstreamload "github.com/wushilin/doris_go_stream_load"

For v1.x releases, keep the import path as github.com/wushilin/doris_go_stream_load; Go modules do not add /v1 to the path.

FakeSend Quick Start

This example is local runnable. It does not require a Doris cluster because FakeSend: true bypasses real HTTP upload and returns a successful Stream Load result.

mkdir dorisstreamload-quickstart
cd dorisstreamload-quickstart
go mod init quickstart
go get github.com/wushilin/doris_go_stream_load@v1.1.0

Create main.go:

package main

import (
	"fmt"
	"time"

	dorisstreamload "github.com/wushilin/doris_go_stream_load"
)

func main() {
	client, err := dorisstreamload.NewClient(dorisstreamload.Config{
		StreamLoadURL:             "http://example.invalid/api/demo/events/_stream_load",
		Columns:                   []string{"event_time", "user_id", "event_name"},
		Mode:                      dorisstreamload.ModeCSV,
		Validation:                dorisstreamload.ValidateSyntax,
		BatchBytes:                1024 * 1024,
		Linger:                    10 * time.Millisecond,
		DorisUploadWorkers:        1,
		DorisUploadRequestTimeout: 30 * time.Second,
		DorisUploadTimeout:        30 * time.Second,
		FakeSend:                  true,
		FakeSendDelay:             20 * time.Millisecond,
		FakeSendDelaySet:          true,
	})
	if err != nil {
		panic(err)
	}
	defer client.Close()

	handle, err := client.SendBatch([]string{
		"2026-05-01T10:00:00Z,1,login",
		"2026-05-01T10:00:01Z,2,logout",
	})
	if err != nil {
		panic(err)
	}

	result := handle.Wait()
	if result.Err != nil {
		panic(result.Err)
	}

	fmt.Printf("success label=%s attempts=%d records=%d\n",
		result.Response.Label,
		result.Attempts,
		client.Stats().RecordsSent,
	)
}

Run it:

go run .

The repository also contains:

  • example/: simple runnable FakeSend examples for CSV, JSON, callbacks, handles, and stats
  • benchmark_demo.go plus benchmark_demo.conf: a larger benchmark-style demo for throughput, wait modes, batching, callbacks, and periodic stats
  • normal_demo.go: a plain Doris Stream Load lifecycle demo without this SDK

Real Doris Cluster

For a real cluster, create a table first. The following DDL assumes append-only event data, common time-range queries, and a small/general-purpose cluster. Tune partitions, buckets, and replication for your own data volume and cluster size.

CREATE DATABASE IF NOT EXISTS demo;

CREATE TABLE IF NOT EXISTS demo.events (
    event_time DATETIME NOT NULL,
    user_id BIGINT NOT NULL,
    event_name VARCHAR(64) NOT NULL
)
DUPLICATE KEY(event_time, user_id)
PARTITION BY RANGE(event_time) ()
DISTRIBUTED BY HASH(user_id) BUCKETS 8
PROPERTIES (
    "dynamic_partition.enable" = "true",
    "dynamic_partition.time_unit" = "DAY",
    "dynamic_partition.start" = "-30",
    "dynamic_partition.end" = "3",
    "dynamic_partition.prefix" = "p",
    "replication_num" = "1",
    "compression" = "zstd"
);

Then configure the SDK with either a full Stream Load URL:

client, err := dorisstreamload.NewClient(dorisstreamload.Config{
	StreamLoadURL:       "http://127.0.0.1:8030/api/demo/events/_stream_load",
	Columns:             []string{"event_time", "user_id", "event_name"},
	Mode:                dorisstreamload.ModeCSV,
	AuthenticationType:  dorisstreamload.AuthenticationBasic,
	AuthenticationToken: "root:password",
})

Or configure endpoint, database, and table separately:

client, err := dorisstreamload.NewClient(dorisstreamload.Config{
	Endpoint:            "http://127.0.0.1:8030",
	Database:            "demo",
	Table:               "events",
	Columns:             []string{"event_time", "user_id", "event_name"},
	Mode:                dorisstreamload.ModeJSON,
	AuthenticationType:  dorisstreamload.AuthenticationBasic,
	AuthenticationToken: "root:password",
})

For basic auth, use AuthenticationToken: "user:password". If Doris does not require auth, leave AuthenticationType and AuthenticationToken empty.

CSV And JSON

Each submitted string is one logical item.

In ModeCSV, one string is one CSV row:

client.Send("2026-05-01T10:00:00Z,1,login")
client.SendBatch([]string{
	"2026-05-01T10:00:01Z,2,logout",
	"2026-05-01T10:00:02Z,3,purchase",
})

When several CSV items are coalesced into one outbound Doris request, the body is newline joined:

row1
row2
row3

In ModeJSON, one string is one JSON object:

client.Send(`{"event_time":"2026-05-01T10:00:00Z","user_id":1,"event_name":"login"}`)
client.SendBatch([]string{
	`{"event_time":"2026-05-01T10:00:01Z","user_id":2,"event_name":"logout"}`,
	`{"event_time":"2026-05-01T10:00:02Z","user_id":3,"event_name":"purchase"}`,
})

When several JSON items are coalesced into one outbound Doris request, the body is one JSON array:

[{"id":1},{"id":2},{"id":3}]

Validation is controlled by Config.Validation:

Value CSV behavior JSON behavior
ValidateNone No parsing before queue admission No parsing before queue admission
ValidateSyntax Non-blank, parses as one row, field count matches Columns Non-blank, valid JSON object
ValidateStrict Same as syntax today Valid object, every configured column present, no extra keys

CSV formatting defaults to separator , and quote ". Override with CSVSeparator and CSVQuote.

Callback, Handle, Stats

Every accepted send returns a Handle. A submitted batch has one shared handle because it concludes as one Doris load outcome.

handle, err := client.SendBatch([]string{
	"2026-05-01T10:00:00Z,1,login",
	"2026-05-01T10:00:01Z,2,logout",
})
if err != nil {
	return err
}

result := handle.Wait()
if result.Err != nil {
	return result.Err
}

Supported handle methods:

  • Wait() DeliveryResult
  • WaitContext(ctx context.Context) (DeliveryResult, error)
  • Done() <-chan struct{}
  • IsDone() bool
  • Result() (DeliveryResult, bool)

Callbacks run once per submitted batch, not once per row:

handle, err := client.SendBatchWithCallback(func(result dorisstreamload.DeliveryResult) {
	if result.Err != nil {
		fmt.Printf("delivery failed attempts=%d err=%v\n", result.Attempts, result.Err)
		return
	}
	fmt.Printf("delivered label=%s\n", result.Response.Label)
}, []string{
	"2026-05-01T10:00:00Z,1,login",
	"2026-05-01T10:00:01Z,2,logout",
})

DeliveryResult includes Err, Attempts, StatusCode, Response, StartedAt, and FinishedAt.

Use client.Stats() for a lifetime snapshot:

stats := client.Stats()
fmt.Printf("jobs=%d errors=%d records=%d bytes=%d p99=%s\n",
	stats.TotalLoadJobs,
	stats.ErrorJobs,
	stats.RecordsSent,
	stats.TotalBytesSent,
	stats.P99LoadTime,
)

Stats include worker counts, job counts, error rate, retry average, total upload attempts, bytes/records sent, average rates, and p50/p90/p99/p999 load time.

Parameters

Required:

Field Description
Columns Doris target columns in record order
Mode ModeCSV or ModeJSON; defaults to ModeCSV when unset
StreamLoadURL Full URL like http://host:8030/api/db/table/_stream_load
Endpoint + Database + Table Alternative to StreamLoadURL

Connection and auth:

Field Default Description
AuthenticationType AuthenticationNone Use AuthenticationBasic for basic auth
AuthenticationToken empty For basic auth, user:password
Headers empty Extra HTTP headers sent to Doris
TLSSkipVerify false Skip TLS certificate verification
TLSCACertPath empty Custom CA certificate path
HTTPClient SDK-created client Optional custom HTTP client
CompressionType none Compress CSV or JSON request bodies with gz, lzo, bz2, lz4, lzop, or deflate. Doris decompresses the body during Stream Load.

For example, enable gzip compression in the client configuration:

client, err := dorisstreamload.NewClient(dorisstreamload.Config{
	StreamLoadURL:   "http://fe:8030/api/db/table/_stream_load",
	Columns:         []string{"id", "payload"},
	Mode:            dorisstreamload.ModeCSV,
	CompressionType: dorisstreamload.CompressionGzip,
})
if err != nil {
	return err
}
defer client.Close()

The equivalent LoaderConfig JSON field is "compression_type": "gz". The same codecs are accepted with ModeJSON. Leave CompressionType unset (or use CompressionNone) to send an uncompressed request body.

Batching and queueing:

Field Default Description
BatchBytes 90 MiB Max outbound request body size; also the per-send admission limit
Linger 5ms How long an open batch waits for more records before it is offered to an idle worker
MaxQueueSize 100000 Max submitted batches in the intake queue
MaxQueueWaitTime 0 How long Send waits for queue space; 0 waits indefinitely
MaxUploadQueueSize max(1, DorisUploadWorkers) How many full batches can wait for a worker before the batcher blocks
DorisUploadWorkers 1 Concurrent upload goroutines

Retry and timing:

Field Default Description
DorisUploadRequestTimeout 300s HTTP deadline for one upload or label-poll request; minimum 10s
DorisUploadTimeout 300s Total retry budget for one batch after its first failed upload
StatusPollTimeout 300s Max time spent polling a label after a failed upload
CallbackTimeout 100ms Reserved callback timing budget in config
SlowCallbackWarn 10ms Slow callback warning threshold

Behavior:

Field Default Description
Validation ValidateSyntax ValidateNone, ValidateSyntax, or ValidateStrict
LabelPrefix go_stream_load Prefix for generated Doris labels
FakeSend false Bypass real HTTP upload and return fake success
FakeSendDelay 500ms Artificial fake-send delay
FakeSendDelaySet false Set true to make explicit 0 fake-send delay distinct from unset
Logger nil Optional logger with Printf
LogLevel LogLevelInfo LogLevelError, LogLevelInfo, or LogLevelDebug
LogLevelSet false Set true when explicitly configuring LogLevelError, because error is the zero value

Compression is off by default. CSV and JSON support the configured codecs. We verified all six non-none codecs with JSON Stream Load requests against the configured Doris server; this behavior is broader than the current Stream Load docs, which describe compress_type as CSV-only. The equivalent LoaderConfig field is "compression_type": "gz".

BatchBytes and Linger work together like Kafka batch.size and linger.ms, with one addition that keeps uploads large under load:

  • A batch that reaches BatchBytes is always dispatched. If every worker is busy and MaxUploadQueueSize full batches are already waiting, the batcher blocks until a worker takes one.
  • A batch that only reaches Linger is handed over immediately when a worker is idle. If all workers are busy, the batcher keeps accumulating for another Linger and tries again, until a worker frees up or the batch fills. Busy workers therefore receive fewer, larger uploads instead of a stream of tiny ones.

Retry Semantics

Every failed upload whose request may have reached Doris is followed by a get_load_state check of its label, and the label state decides what happens next:

Label state Meaning Action
VISIBLE / COMMITTED Data was loaded despite the error (for example a 5xx from a proxy) Success, no resend
ABORTED / UNKNOWN Data was not loaded (for example too many versions, -235, or a rejected request) Retry with a new label after backoff, within DorisUploadTimeout
PREPARE / PRECOMMITTED until StatusPollTimeout Load still in progress Failure, no resend, since a resend could load the data twice
poll rejected (HTTP 4xx, or no database configured) Label cannot be checked Failure if the upload outcome was ambiguous; otherwise retry with a new label, like the Flink connector

A request that never reached Doris (TCP dial failure) is retried directly with the same label. Retries back off from 1s to 4s. When DorisUploadTimeout runs out the batch fails with an error that includes the last upload error.

Because retry is decided by the label state rather than the HTTP status, permanent mistakes such as a wrong password or a bad column list are retried until DorisUploadTimeout expires. Lower it if you want such failures surfaced faster.

LoaderConfig

LoaderConfig is the JSON-serializable counterpart of Config.

lc, err := dorisstreamload.LoadLoaderConfig("loader.json")
if err != nil {
	return err
}
cfg, err := lc.Config()
if err != nil {
	return err
}
client, err := dorisstreamload.NewClient(cfg)

Duration fields use Go duration strings such as "500ms", "2s", and "5m". JSON logging also supports log_level_set; set it to true when log_level is 0 and you mean LogLevelError.

Common Errors

Send(...) and SendBatch(...) can fail before data enters the SDK queue:

Error Meaning Typical action
ErrClientClosed The client is closed Stop sending or create a new client
ErrQueueFull The intake queue stayed full longer than MaxQueueWaitTime Increase workers, queue size, or timeout; reduce producer rate
ErrSendTooLarge One submitted item or batch is larger than BatchBytes Split the caller-side batch or raise BatchBytes up to the 90 MiB limit
validation error CSV/JSON failed configured validation Fix the row/object or loosen Validation

Delivery can fail after queue admission; check DeliveryResult.Err from the handle or callback:

Failure Meaning Typical action
upload timeout The label was never loaded within DorisUploadTimeout; the error includes the last Doris response Inspect the Doris message (auth, table, schema, data format, too many versions); raise DorisUploadTimeout for transient cluster issues
ambiguous transport error Request may have reached Doris but response was lost The SDK polls the load label to decide whether it became visible
status poll timeout Doris did not reach a final visible/failed state in time Increase StatusPollTimeout or inspect Doris load jobs
label state could not be checked get_load_state was rejected after an ambiguous outcome Point StreamLoadURL/Endpoint at an FE and make sure Database (or the URL path) names the database
callback too slow Callback exceeded SlowCallbackWarn and produced an info log Keep callbacks small; hand work to another goroutine if needed

Shutdown

Close() stops intake, drains accepted work, and waits for in-flight deliveries. After Close(), new sends return ErrClientClosed; already accepted handles still complete.

Documentation

Index

Constants

View Source 
const (
	ModeCSV  Mode = "csv"
	ModeJSON Mode = "json"

	CompressionNone    CompressionType = "none"
	CompressionGzip    CompressionType = "gz"
	CompressionLZO     CompressionType = "lzo"
	CompressionBZip2   CompressionType = "bz2"
	CompressionLZ4     CompressionType = "lz4"
	CompressionLZOP    CompressionType = "lzop"
	CompressionDeflate CompressionType = "deflate"

	ValidateNone   ValidationMode = "none"
	ValidateSyntax ValidationMode = "syntax"
	ValidateStrict ValidationMode = "strict"

	AuthenticationNone  AuthenticationType = ""
	AuthenticationBasic AuthenticationType = "basic"
)

Variables

View Source 
var (
	ErrClientClosed = errors.New("client is closed")
	ErrQueueFull    = errors.New("queue is full")
	ErrSendTooLarge = errors.New("send exceeds batch bytes limit")
)

Functions

func SaveLoaderConfig 

func SaveLoaderConfig(path string, cfg LoaderConfig) error

Types

type AuthenticationType 

type AuthenticationType string

type Client 

type Client struct {
	// contains filtered or unexported fields
}

func NewClient 

func NewClient(cfg Config) (*Client, error)

func (*Client) BufferedRecords 

func (c *Client) BufferedRecords() int

func (*Client) Close 

func (c *Client) Close() error

func (*Client) Closed 

func (c *Client) Closed() <-chan struct{}

func (*Client) Send 

func (c *Client) Send(record string) (*Handle, error)

func (*Client) SendBatch 

func (c *Client) SendBatch(records []string) (*Handle, error)

func (*Client) SendBatchContext 

func (c *Client) SendBatchContext(ctx context.Context, records []string) (*Handle, error)

func (*Client) SendBatchWithCallback 

func (c *Client) SendBatchWithCallback(callback DeliveryCallback, records []string) (*Handle, error)

func (*Client) SendRecord 

func (c *Client) SendRecord(records ...string) (*Handle, error)

func (*Client) SendRecordContext 

func (c *Client) SendRecordContext(ctx context.Context, records ...string) (*Handle, error)

func (*Client) SendRecordWithCallback 

func (c *Client) SendRecordWithCallback(callback DeliveryCallback, records ...string) (*Handle, error)

func (*Client) SendWithCallback 

func (c *Client) SendWithCallback(callback DeliveryCallback, record string) (*Handle, error)

func (*Client) Stats 

func (c *Client) Stats() ClientStats

type ClientStats 

type ClientStats struct {
	StartedAt time.Time

	TotalWorkers int
	IdleWorkers  int
	BusyWorkers  int

	TotalLoadJobs int64
	ErrorJobs     int64
	ErrorRate     float64

	AverageLoadTime time.Duration
	P50LoadTime     time.Duration
	P90LoadTime     time.Duration
	P99LoadTime     time.Duration
	P999LoadTime    time.Duration

	AverageRetries float64

	TotalBytesSent   int64
	AverageLoadSize  float64
	AverageBytesRate float64

	RecordsSent        int64
	AverageRecordsRate float64

	TotalUploadAttempts int64
}

type CompressionType added in v1.1.0 

type CompressionType string

type Config 

type Config struct {
	Endpoint      string
	Database      string
	Table         string
	StreamLoadURL string

	Columns         []string
	Headers         http.Header
	Mode            Mode
	CompressionType CompressionType

	AuthenticationType  AuthenticationType
	AuthenticationToken string

	MaxQueueSize       int
	MaxUploadQueueSize int
	BatchBytes         int
	Linger             time.Duration
	MaxQueueWaitTime   time.Duration
	DorisUploadWorkers int
	Validation         ValidationMode

	DorisUploadTimeout        time.Duration
	DorisUploadRequestTimeout time.Duration
	CallbackTimeout           time.Duration
	SlowCallbackWarn          time.Duration
	LabelPrefix               string
	StatusPollTimeout         time.Duration
	FakeSend                  bool
	FakeSendDelay             time.Duration
	FakeSendDelaySet          bool

	CSVSeparator string
	CSVQuote     string

	TLSSkipVerify bool
	TLSCACertPath string

	Logger   Logger
	LogLevel LogLevel
	// LogLevelSet distinguishes an explicit LogLevelError from the zero-value
	// default path. Non-zero log levels are treated as explicitly configured.
	LogLevelSet bool

	HTTPClient *http.Client
}

type DeliveryCallback 

type DeliveryCallback func(DeliveryResult)

type DeliveryResult 

type DeliveryResult struct {
	Err        error
	Attempts   int
	StatusCode int
	Response   *StreamLoadResponse
	StartedAt  time.Time
	FinishedAt time.Time
}

func (DeliveryResult) Success 

func (r DeliveryResult) Success() bool

type Handle 

type Handle struct {
	// contains filtered or unexported fields
}

func (*Handle) Done 

func (h *Handle) Done() <-chan struct{}

func (*Handle) IsDone 

func (h *Handle) IsDone() bool

func (*Handle) Result 

func (h *Handle) Result() (DeliveryResult, bool)

func (*Handle) Wait 

func (h *Handle) Wait() DeliveryResult

func (*Handle) WaitContext 

func (h *Handle) WaitContext(ctx context.Context) (DeliveryResult, error)

type LoaderConfig 

type LoaderConfig struct {
	Endpoint        string          `json:"endpoint,omitempty"`
	Database        string          `json:"database,omitempty"`
	Table           string          `json:"table,omitempty"`
	StreamLoadURL   string          `json:"stream_load_url,omitempty"`
	Columns         []string        `json:"columns,omitempty"`
	Headers         http.Header     `json:"headers,omitempty"`
	Mode            Mode            `json:"mode,omitempty"`
	CompressionType CompressionType `json:"compression_type,omitempty"`

	AuthenticationType  AuthenticationType `json:"authentication_type,omitempty"`
	AuthenticationToken string             `json:"authentication_token,omitempty"`

	MaxQueueSize       int            `json:"max_queue_size,omitempty"`
	MaxUploadQueueSize int            `json:"max_upload_queue_size,omitempty"`
	BatchBytes         int            `json:"batch_bytes,omitempty"`
	Linger             string         `json:"linger,omitempty"`
	MaxQueueWaitTime   string         `json:"max_queue_wait_time,omitempty"`
	DorisUploadWorkers int            `json:"doris_upload_workers,omitempty"`
	Validation         ValidationMode `json:"validation,omitempty"`

	DorisUploadTimeout        string `json:"doris_upload_timeout,omitempty"`
	DorisUploadRequestTimeout string `json:"doris_upload_request_timeout,omitempty"`
	CallbackTimeout           string `json:"callback_timeout,omitempty"`
	SlowCallbackWarn          string `json:"slow_callback_warn,omitempty"`
	LabelPrefix               string `json:"label_prefix,omitempty"`
	StatusPollTimeout         string `json:"status_poll_timeout,omitempty"`
	FakeSend                  bool   `json:"fake_send,omitempty"`
	FakeSendDelay             string `json:"fake_send_delay,omitempty"`
	FakeSendDelaySet          bool   `json:"fake_send_delay_set,omitempty"`

	CSVSeparator string `json:"csv_separator,omitempty"`
	CSVQuote     string `json:"csv_quote,omitempty"`

	TLSSkipVerify bool   `json:"tls_skip_verify,omitempty"`
	TLSCACertPath string `json:"tls_ca_cert_path,omitempty"`

	LogLevel    LogLevel `json:"log_level,omitempty"`
	LogLevelSet bool     `json:"log_level_set,omitempty"`
}

func LoadLoaderConfig 

func LoadLoaderConfig(path string) (LoaderConfig, error)

func LoaderConfigFromConfig 

func LoaderConfigFromConfig(cfg Config) LoaderConfig

func (LoaderConfig) Config 

func (lc LoaderConfig) Config() (Config, error)

type LogLevel 

type LogLevel int
const (
	LogLevelError LogLevel = iota
	LogLevelInfo
	LogLevelDebug
)

type Logger 

type Logger interface {
	Printf(format string, args ...any)
}

type Mode 

type Mode string

type SendOption 

type SendOption func(*sendOptions)

func WithCallback 

func WithCallback(callback DeliveryCallback) SendOption

type StreamLoadResponse 

type StreamLoadResponse struct {
	TxnID              int64  `json:"TxnId"`
	Label              string `json:"Label"`
	Status             string `json:"Status"`
	ExistingJobStatus  string `json:"ExistingJobStatus"`
	Message            string `json:"Message"`
	ErrorURL           string `json:"ErrorURL"`
	NumberTotalRows    int64  `json:"NumberTotalRows"`
	NumberLoadedRows   int64  `json:"NumberLoadedRows"`
	NumberFilteredRows int64  `json:"NumberFilteredRows"`
	NumberUnselected   int64  `json:"NumberUnselectedRows"`
	LoadBytes          int64  `json:"LoadBytes"`
	LoadTimeMS         int64  `json:"LoadTimeMs"`
}

type ValidationMode 

type ValidationMode string

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.