batchflow

package module
v2.0.0-rc.2 Latest Latest
Warning

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

 Go to latest Published: Jul 6, 2026 License: MIT Imports: 15 Imported by: 0

README

BatchFlow

Release Go Reference Go Report Card License

BatchFlow is a Go ingestion runtime built on go-pipeline. It provides one batching model for SQL databases, Redis, PostgreSQL/Hologres COPY FROM, and custom batch sinks: enqueue records, flush asynchronously, execute through pluggable backends, apply retry/concurrency controls, and expose production diagnostics.

Chinese documentation: README.zh-CN.md.

Current RC2 API

The public module path is:

github.com/rushairer/batchflow/v2

Because v2 has not been formally released yet, the public runtime API intentionally uses clean names instead of version-decorated names:

cfg := batchflow.DefaultConfig(executor)
flow, err := batchflow.New(ctx, cfg)

Versioning lives in the module path and release tag, not in type or function names.

Features

  • Clean runtime API: Config, RuntimeConfig, Flow, DefaultConfig, and New.
  • Unified executor model for SQL, Redis, COPY FROM, and custom BatchExecutor implementations.
  • Runtime sharding with hash, round-robin, and least-loaded routing.
  • Runtime backpressure with block, reject, and timeout modes.
  • Estimated queue memory limiter for OOM protection.
  • Adaptive tuning policy engine for flush-size and latency recommendations.
  • SQL upsert controls for explicit conflict keys, update columns, and in-batch duplicate-key coalescing.
  • PostgreSQL/Hologres COPY FROM fast path through CopyFromExecutor and optional adapters/pgxcopy.
  • Retry, timeout, concurrency limit, structured error classification, metrics, and safe diagnostics.
  • Complete lifecycle controls with Close(), Wait(), and Done().

Install

go get github.com/rushairer/batchflow/v2@v2.0.0-rc.2

Quick Start: SQL runtime

package main

import (
	"context"
	"database/sql"
	"log"
	"time"

	_ "github.com/go-sql-driver/mysql"
	batchflow "github.com/rushairer/batchflow/v2"
)

func main() {
	ctx := context.Background()

	db, err := sql.Open("mysql", "user:password@tcp(localhost:3306)/testdb?parseTime=true")
	if err != nil {
		log.Fatal(err)
	}
	defer db.Close()

	executor := batchflow.NewSQLThrottledBatchExecutorWithDriver(db, batchflow.DefaultMySQLDriver).
		WithConcurrencyLimit(8).
		WithRetryConfig(batchflow.RetryConfig{
			Enabled:     true,
			MaxAttempts: 3,
			BackoffBase: 20 * time.Millisecond,
			MaxBackoff:  500 * time.Millisecond,
		})

	cfg := batchflow.DefaultConfig(executor)
	cfg.Pipeline.BufferSize = 10000
	cfg.Pipeline.FlushSize = 1000
	cfg.Pipeline.FlushInterval = 50 * time.Millisecond
	cfg.Runtime.ShardCount = 4
	cfg.Runtime.Routing = batchflow.ShardRoutingHash
	cfg.Runtime.Backpressure = batchflow.BackpressureConfig{
		Enabled:       true,
		Mode:          batchflow.BackpressureTimeout,
		HighWatermark: 8000,
		Timeout:       500 * time.Millisecond,
	}
	cfg.Runtime.MemoryLimit = batchflow.MemoryLimitConfig{
		Enabled:         true,
		MaxQueueBytes:   512 << 20,
		AvgRequestBytes: 512,
		Mode:            batchflow.BackpressureTimeout,
		Timeout:         500 * time.Millisecond,
	}

	flow, err := batchflow.New(ctx, cfg)
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := flow.Close(); err != nil {
			log.Printf("batchflow close: %v", err)
		}
	}()

	schema := batchflow.NewSQLSchema(
		"users",
		batchflow.ConflictUpdateOperationConfig.
			WithConflictColumns("id").
			WithUpdateColumns("name", "email"),
		"id", "name", "email",
	)

	req := batchflow.NewRequest(schema).
		SetUint64("id", 1).
		SetString("name", "alice").
		SetString("email", "alice@example.com")

	if err := flow.Submit(ctx, req); err != nil {
		log.Fatal(err)
	}
}

COPY FROM fast path

For PostgreSQL/Hologres append-only ingestion, use CopyFromExecutor. The root module only depends on the minimal CopyFromClient interface. The pgx implementation lives in the optional adapter module:

go get github.com/rushairer/batchflow/adapters/pgxcopy@v2.0.0-rc.2

import (
	batchflow "github.com/rushairer/batchflow/v2"
	"github.com/rushairer/batchflow/adapters/pgxcopy"
)

copyExecutor := pgxcopy.NewExecutor(pool)

cfg := batchflow.DefaultConfig(copyExecutor)
cfg.Pipeline.BufferSize = 50000
cfg.Pipeline.FlushSize = 5000
cfg.Pipeline.FlushInterval = 20 * time.Millisecond
cfg.Runtime.ShardCount = 8
cfg.Runtime.Backpressure = batchflow.BackpressureConfig{
	Enabled:       true,
	Mode:          batchflow.BackpressureTimeout,
	HighWatermark: 40000,
	Timeout:       time.Second,
}
cfg.Runtime.MemoryLimit = batchflow.MemoryLimitConfig{
	Enabled:         true,
	MaxQueueBytes:   1 << 30,
	AvgRequestBytes: 512,
	Mode:            batchflow.BackpressureTimeout,
	Timeout:         time.Second,
}

flow, err := batchflow.New(ctx, cfg)

COPY FROM is intentionally limited to append-only schemas. Upsert/update/replace flows should use the SQL executor path.

Core Semantics

Lifecycle
  • New(ctx, cfg) starts the runtime immediately.
  • Submit(ctx, req) enqueues data and may reject/block/timeout according to runtime backpressure and memory limits.
  • Close() stops accepting new records, closes input, triggers final flush, and waits for shutdown.
  • Wait() waits for shutdown without closing input.
  • Done() returns a read-only channel closed when the background runtime exits.

Always call Close() during application shutdown:

if err := flow.Close(); err != nil {
	return err
}
Batching
  • The pipeline groups requests by FlushSize or FlushInterval.
  • Each flush is split by SchemaInterface.
  • Each schema group calls BatchExecutor.ExecuteBatch(...) once.
  • ObserveBatchSize(n) reports the size of one schema execution batch, not the whole flush input size.
Requests

Use typed setters when available:

req := batchflow.NewRequest(schema).
	SetInt("retry_count", 3).
	SetUint64("id", 42).
	SetString("email", "alice@example.com").
	SetBool("enabled", true)

Other values can be set with Set(name, value) or SetNull(name).

SQL Update and Replace

Use explicit conflict keys for PostgreSQL/MySQL upserts:

schema := batchflow.NewSQLSchema(
	"users",
	batchflow.ConflictUpdateOperationConfig.
		WithConflictColumns("tenant_id", "user_id").
		WithUpdateColumns("name", "email"),
	"tenant_id", "user_id", "name", "email", "updated_at",
)

Rules:

  • If ConflictColumns is omitted, BatchFlow keeps the legacy fallback and uses the first schema column. Treat this as compatibility only.
  • ConflictUpdate updates non-conflict columns by default, or only UpdateColumns when configured.
  • PostgreSQL ConflictReplace means upsert overwrite with ON CONFLICT (...) DO UPDATE SET ....
  • MySQL ConflictReplace keeps native REPLACE INTO semantics.
  • In-batch duplicate conflict keys are coalesced before SQL generation to avoid PostgreSQL errors such as "cannot affect row a second time".

Dry-run the final SQL before production rollout:

preview, err := batchflow.GenerateSQLPreview(ctx, batchflow.DefaultPostgreSQLDriver, schema, rows)
if err != nil {
	return err
}
log.Printf("sql=%s fingerprint=%s args=%d input=%d output=%d dedup=%d",
	preview.SQL,
	preview.Fingerprint,
	preview.ArgsCount,
	preview.DedupStats.InputRows,
	preview.DedupStats.OutputRows,
	preview.DedupStats.DeduplicatedRows,
)

Do not log preview.Args in production unless the values are known to be safe.

Legacy convenience constructors

Legacy constructors remain available for simple migrations and quick tests:

flow := batchflow.NewMySQLBatchFlow(ctx, db, batchflow.PipelineConfig{
	BufferSize:    1000,
	FlushSize:     200,
	FlushInterval: 100 * time.Millisecond,
})

For new production code, prefer DefaultConfig(executor) plus New(ctx, cfg) so runtime sharding, memory limiter, and backpressure are explicit.

Documentation

Development

make fmt
make test
make lint
make docs-check

Release validation:

go test ./...
go test ./... -race
go test ./benchmark -bench=. -benchmem
cd adapters/pgxcopy && go test ./...

Documentation

Index

Examples

Constants

View Source 
const (
	ErrorReasonUnknown         = "unknown"
	ErrorReasonContextCanceled = "context_canceled"
	ErrorReasonContextDeadline = "context_deadline"
	ErrorReasonDuplicateKey    = "duplicate_key"
	ErrorReasonDeadlock        = "deadlock"
	ErrorReasonLockTimeout     = "lock_timeout"
	ErrorReasonTimeout         = "timeout"
	ErrorReasonConnection      = "connection"
	ErrorReasonIO              = "io"
	ErrorReasonSyntax          = "syntax"
	ErrorReasonNonRetryable    = "non_retryable"
)
View Source 
const (
	BackendSQL    = "sql"
	BackendRedis  = "redis"
	BackendCustom = "custom"

	OperationInsert  = "insert"
	OperationUpsert  = "upsert"
	OperationCommand = "command"
	OperationCustom  = "custom"

	BatchStageValidate = "validate"
	BatchStageGenerate = "generate"
	BatchStageExecute  = "execute"
	BatchStageRetry    = "retry"
	BatchStageFinal    = "final"
)

Variables

View Source 
var (
	// ErrEmptyRequest 空请求错误
	ErrEmptyRequest = errors.New("empty request")

	// ErrContextCanceled 上下文被取消错误
	ErrContextCanceled = errors.New("context canceled")

	// ErrInvalidSchema 无效的 schema 错误
	ErrInvalidSchema = errors.New("invalid schema")

	// ErrMissingColumn 缺少列错误
	ErrMissingColumn = errors.New("missing required column")

	// ErrInvalidColumnType 无效的列类型错误
	ErrInvalidColumnType = errors.New("invalid column type")

	// ErrEmptyBatch 空批次错误
	ErrEmptyBatch = errors.New("empty batch")

	// ErrEmptySchemaName 空表名错误
	ErrEmptySchemaName = errors.New("empty schema name")

	// ErrBackpressure 表示队列达到反压阈值且策略拒绝或等待超时
	ErrBackpressure = errors.New("batchflow backpressure limit reached")
)
View Source 
var ConflictIgnoreOperationConfig = SQLOperationConfig{
	ConflictStrategy: ConflictIgnore,
}
View Source 
var ConflictReplaceOperationConfig = SQLOperationConfig{
	ConflictStrategy: ConflictReplace,
}
View Source 
var ConflictUpdateOperationConfig = SQLOperationConfig{
	ConflictStrategy: ConflictUpdate,
}
View Source 
var DefaultMySQLDriver = NewMySQLDriver()
View Source 
var DefaultOperationConfig = SQLOperationConfig{
	ConflictStrategy: ConflictIgnore,
}
View Source 
var DefaultPostgreSQLDriver = NewPostgreSQLDriver()
View Source 
var DefaultRedisPipelineDriver = NewRedisPipelineDriver()
View Source 
var DefaultSQLiteDriver = NewSQLiteDriver()
View Source 
var ErrCopyFromUnsupportedOperation = errors.New("copy from supports append-only SQL schemas only")
View Source 
var ErrMemoryLimitExceeded = errors.New("batchflow memory limit exceeded")

Functions

func ClassifyError 

func ClassifyError(err error) (retryable bool, reason string)

ClassifyError returns a low-cardinality retry decision and reason label for logs and metrics. It unwraps BatchError and SQLError before classifying the underlying cause.

func FingerprintSQL 

func FingerprintSQL(sqlText string) string

func FingerprintText 

func FingerprintText(text string) string

func NewBatchFlowWithMock 

func NewBatchFlowWithMock(ctx context.Context, config PipelineConfig) (*BatchFlow, *MockExecutor)

NewBatchFlowWithMock 使用模拟执行器创建 BatchFlow 实例（用于测试） 内部架构：BatchFlow -> MockExecutor（直接实现BatchExecutor，无真实数据库操作） 适用于单元测试，不依赖真实数据库连接

func NewBatchFlowWithMockDriver 

func NewBatchFlowWithMockDriver(ctx context.Context, config PipelineConfig, sqlDriver SQLDriver) (*BatchFlow, *MockExecutor)

NewBatchFlowWithMockDriver 使用模拟执行器创建 BatchFlow 实例（测试特定SQLDriver） 内部架构：BatchFlow -> MockExecutor（模拟ThrottledBatchExecutor行为，测试SQLDriver逻辑） 适用于测试自定义SQLDriver的SQL生成逻辑

func OperationFingerprint 

func OperationFingerprint(parts ...string) string

func RegisterErrorClassifier 

func RegisterErrorClassifier(classifier ErrorClassifier) func()

RegisterErrorClassifier adds a custom classifier after built-in structured classifiers and before the string fallback. The returned function unregisters it.

Types

type AdaptiveTuner 

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

AdaptiveTuner is a stateless policy engine.

func NewAdaptiveTuner 

func NewAdaptiveTuner(cfg AdaptiveTuningConfig) *AdaptiveTuner

func (*AdaptiveTuner) Tune 

func (t *AdaptiveTuner) Tune(s TuningSignal) TuningRecommendation

type AdaptiveTuningConfig 

type AdaptiveTuningConfig struct {
	Enabled bool

	MinFlushSize uint32
	MaxFlushSize uint32

	MinFlushInterval time.Duration
	MaxFlushInterval time.Duration

	ScaleUpQueueDepth   int
	ScaleDownQueueDepth int

	TargetLatency time.Duration
}

AdaptiveTuningConfig controls runtime auto tuning behavior.

type BackpressureConfig 

type BackpressureConfig struct {
	Enabled       bool
	Mode          BackpressureMode
	HighWatermark int
	CheckInterval time.Duration
	Timeout       time.Duration
}

BackpressureConfig controls queue pressure.

type BackpressureMode 

type BackpressureMode uint8

BackpressureMode controls queue pressure behavior. 

const (
	BackpressureBlock BackpressureMode = iota
	BackpressureReject
	BackpressureTimeout
)

type Batch 

type Batch = []Record

Batch is the backend-neutral batch payload passed through BatchFlow.

type BatchError 

type BatchError struct {
	Stage       string
	Backend     string
	Schema      string
	BatchSize   int
	Fingerprint string
	Attributes  map[string]any
	Cause       error
}

BatchError wraps backend-neutral batch failures with safe diagnostic metadata.

func (*BatchError) Error 

func (e *BatchError) Error() string

func (*BatchError) Unwrap 

func (e *BatchError) Unwrap() error

type BatchEvent 

type BatchEvent struct {
	Stage       string
	Backend     string
	Operation   string
	Schema      string
	Status      string
	Attempt     int
	BatchSize   int
	InputItems  int
	OutputItems int
	ArgCount    int
	Duration    time.Duration
	Fingerprint string
	Reason      string
	Attributes  map[string]any
	Err         error
}

type BatchExecutor 

type BatchExecutor interface {
	// ExecuteBatch 执行批量操作
	ExecuteBatch(ctx context.Context, schema SchemaInterface, data []map[string]any) error
}

BatchExecutor 批量执行器接口 - 所有数据库驱动的统一入口

type BatchFlow 

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

BatchFlow 批量处理管道 核心组件，整合 go-pipeline 和 BatchExecutor，提供统一的批量处理接口

架构层次： Application -> BatchFlow -> gopipeline -> BatchExecutor -> Database

支持的 BatchExecutor 实现： - SQL 数据库：ThrottledBatchExecutor + SQLBatchProcessor + SQLDriver - NoSQL 数据库：ThrottledBatchExecutor + RedisBatchProcessor + RedisDriver（直接生成/执行命令） - 测试环境：MockExecutor（直接实现 BatchExecutor） 可选能力： - WithConcurrencyLimit：通过信号量限制 ExecuteBatch 并发，避免攒批后同时冲击数据库（limit <= 0 等价于不限流）

func NewBatchFlow 

func NewBatchFlow(ctx context.Context, buffSize uint32, flushSize uint32, flushInterval time.Duration, executor BatchExecutor) *BatchFlow

NewBatchFlow 创建 BatchFlow 实例 这是最底层的构造函数，接受任何实现了BatchExecutor接口的执行器 通常不直接使用，而是通过具体数据库的工厂方法创建

func NewBatchFlowWithConfig 

func NewBatchFlowWithConfig(ctx context.Context, config BatchFlowConfig) (*BatchFlow, error)

func NewMySQLBatchFlow 

func NewMySQLBatchFlow(ctx context.Context, db *sql.DB, config PipelineConfig) *BatchFlow

NewMySQLBatchFlow 创建MySQL BatchFlow实例（使用默认Driver）

内部架构：BatchFlow -> ThrottledBatchExecutor -> SQLBatchProcessor -> MySQLDriver -> MySQL

这是推荐的使用方式，使用MySQL优化的默认配置

func NewPostgreSQLBatchFlow 

func NewPostgreSQLBatchFlow(ctx context.Context, db *sql.DB, config PipelineConfig) *BatchFlow

NewPostgreSQLBatchFlow 创建PostgreSQL BatchFlow实例（使用默认Driver）

func NewRedisBatchFlow 

func NewRedisBatchFlow(ctx context.Context, db *redisV9.Client, config PipelineConfig) *BatchFlow

NewRedisBatchFlow 创建Redis BatchFlow实例

内部架构（NoSQL）：BatchFlow -> ThrottledBatchExecutor -> RedisBatchProcessor -> RedisDriver -> Redis 说明：NoSQL 路径不使用 SQL 抽象层，直接生成并执行 Redis 命令；仍可启用 WithConcurrencyLimit 控制批次并发。

func NewRedisBatchFlowWithDriver 

func NewRedisBatchFlowWithDriver(ctx context.Context, db *redisV9.Client, config PipelineConfig, driver RedisDriver) *BatchFlow

func NewSQLBatchFlowWithDriver 

func NewSQLBatchFlowWithDriver(ctx context.Context, db *sql.DB, config PipelineConfig, driver SQLDriver) *BatchFlow

NewSQLBatchFlow 创建SQL BatchFlow实例（使用自定义Driver）

func NewSQLiteBatchFlow 

func NewSQLiteBatchFlow(ctx context.Context, db *sql.DB, config PipelineConfig) *BatchFlow

NewSQLiteBatchFlow 创建SQLite BatchFlow实例（使用默认Driver）

func (*BatchFlow) Close 

func (b *BatchFlow) Close() error

Close 停止接收新请求，触发最终 flush，并等待后台 pipeline 退出。 它是幂等的；首次调用会关闭内部数据通道，后续调用仅等待同一个退出结果。

Example 
package main

import (
	"context"
	"time"

	"github.com/rushairer/batchflow/v2"
)

func main() {
	ctx := context.Background()
	flow, _ := batchflow.NewBatchFlowWithMock(ctx, batchflow.PipelineConfig{
		BufferSize:    16,
		FlushSize:     100,
		FlushInterval: time.Second,
	})

	schema := batchflow.NewSQLSchema("users", batchflow.ConflictIgnoreOperationConfig, "id", "name")
	req := batchflow.NewRequest(schema).
		SetUint64("id", 1).
		SetString("name", "alice")

	if err := flow.Submit(ctx, req); err != nil {
		panic(err)
	}

	if err := flow.Close(); err != nil {
		panic(err)
	}
}

func (*BatchFlow) Done 

func (b *BatchFlow) Done() <-chan struct{}

Done 返回一个在 BatchFlow 后台 goroutine 退出时关闭的信号。

Example 
package main

import (
	"context"
	"time"

	"github.com/rushairer/batchflow/v2"
)

func main() {
	ctx := context.Background()
	flow, _ := batchflow.NewBatchFlowWithMock(ctx, batchflow.PipelineConfig{
		BufferSize:    16,
		FlushSize:     100,
		FlushInterval: time.Second,
	})

	stopped := make(chan struct{})
	go func() {
		<-flow.Done()
		close(stopped)
	}()

	if err := flow.Close(); err != nil {
		panic(err)
	}

	<-stopped
}

func (*BatchFlow) ErrorChan 

func (b *BatchFlow) ErrorChan(size int) <-chan error

ErrorChan 获取错误通道

func (*BatchFlow) Submit 

func (b *BatchFlow) Submit(ctx context.Context, request *Request) error

Submit 提交请求到批量处理管道

func (*BatchFlow) Wait 

func (b *BatchFlow) Wait() error

Wait 等待后台 pipeline 退出并返回最终运行结果。

type BatchFlowConfig 

type BatchFlowConfig struct {
	Pipeline PipelineConfig
	Executor BatchExecutor
}

BatchFlowConfig is the v2 constructor config for a fully assembled BatchFlow.

func DefaultBatchFlowConfig 

func DefaultBatchFlowConfig(executor BatchExecutor) BatchFlowConfig

type BatchFlowMetricsReporter 

type BatchFlowMetricsReporter interface {
	// IncSubmitRejected 记录 Submit 被拒绝的次数与原因。
	IncSubmitRejected(reason string)
	// ObservePipelineFlushSize 记录一次 pipeline flush 收到的请求数。
	ObservePipelineFlushSize(n int)
	// ObserveSchemaGroupsPerFlush 记录一次 flush 被拆成的 schema 组数。
	ObserveSchemaGroupsPerFlush(n int)
}

BatchFlowMetricsReporter 是 BatchFlow 自身流程指标的可选扩展接口。 这些指标不属于执行器层，也不依赖 go-pipeline 的原生 MetricsHook， 仅在实现方显式选择时才会上报。

type BatchProcessor 

type BatchProcessor interface {
	// GenerateOperations 生成批量操作
	GenerateOperations(ctx context.Context, schema SchemaInterface, data []map[string]any) (operations Operations, err error)

	// ExecuteOperations 执行批量操作
	ExecuteOperations(ctx context.Context, operations Operations) error
}

BatchProcessor 批量处理器接口 - SQL数据库的核心处理逻辑

type CoalesceResult 

type CoalesceResult struct {
	Batch             Batch
	InputItems        int
	OutputItems       int
	DeduplicatedItems int
	MergedItems       int
}

CoalesceResult contains the coalesced batch and low-cardinality statistics.

func NewCoalesceResult 

func NewCoalesceResult(batch Batch) CoalesceResult

type CoalesceStrategy 

type CoalesceStrategy string

CoalesceStrategy controls how duplicate keys are handled inside one batch. 

const (
	CoalesceDisabled           CoalesceStrategy = "disabled"
	CoalesceKeepFirst          CoalesceStrategy = "keep_first"
	CoalesceKeepLast           CoalesceStrategy = "keep_last"
	CoalesceMergePresentFields CoalesceStrategy = "merge_present_fields"
)

type Coalescer 

type Coalescer interface {
	Coalesce(ctx context.Context, schema SchemaInterface, batch Batch) (CoalesceResult, error)
}

Coalescer merges or removes duplicate items before operation generation.

type CoalescerFunc 

type CoalescerFunc func(context.Context, SchemaInterface, Batch) (CoalesceResult, error)

func (CoalescerFunc) Coalesce 

func (f CoalescerFunc) Coalesce(ctx context.Context, schema SchemaInterface, batch Batch) (CoalesceResult, error)

type ConcurrencyCapable 

type ConcurrencyCapable[T any] interface {
	WithConcurrencyLimit(int) T
}

type Config 

type Config struct {
	Pipeline PipelineConfig
	Runtime  RuntimeConfig
	Executor BatchExecutor
}

Config is the stable public constructor config for the v2 module.

func DefaultConfig 

func DefaultConfig(executor BatchExecutor) Config

type ConfigError 

type ConfigError struct {
	Field string
	Cause error
}

func (*ConfigError) Error 

func (e *ConfigError) Error() string

func (*ConfigError) Unwrap 

func (e *ConfigError) Unwrap() error

type ConflictStrategy 

type ConflictStrategy uint8

ConflictStrategy 冲突处理策略 

const (
	ConflictIgnore ConflictStrategy = iota
	ConflictReplace
	ConflictUpdate
)

type CopyFromClient 

type CopyFromClient interface {
	CopyFrom(ctx context.Context, table string, columns []string, rows [][]any) (int64, error)
}

CopyFromClient is the minimal adapter surface required by CopyFromExecutor. It intentionally avoids importing pgx in the root module.

type CopyFromExecutor 

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

CopyFromExecutor is a zero-SQL-generation backend for append-only ingestion.

func NewCopyFromExecutor 

func NewCopyFromExecutor(client CopyFromClient) *CopyFromExecutor

func (*CopyFromExecutor) ExecuteBatch 

func (e *CopyFromExecutor) ExecuteBatch(ctx context.Context, schema SchemaInterface, data []map[string]any) error

func (*CopyFromExecutor) WithTimeout 

func (e *CopyFromExecutor) WithTimeout(timeout time.Duration) *CopyFromExecutor

type ErrorClassifier 

type ErrorClassifier interface {
	Classify(error) (retryable bool, reason string, ok bool)
}

ErrorClassifier recognizes backend-specific errors and returns a low-cardinality reason.

type ErrorClassifierFunc 

type ErrorClassifierFunc func(error) (retryable bool, reason string, ok bool)

func (ErrorClassifierFunc) Classify 

func (f ErrorClassifierFunc) Classify(err error) (retryable bool, reason string, ok bool)

type Flow 

type Flow = RuntimeEngine

Flow is the stable public runtime surface for the v2 module.

func New 

func New(ctx context.Context, cfg Config) (*Flow, error)

New creates the converged runtime flow.

type KeyCoalescer 

type KeyCoalescer struct {
	Strategy   CoalesceStrategy
	KeyColumns []string
	KeyFunc    KeyFunc
}

KeyCoalescer is a general key-based coalescer for custom backends.

func NewKeyCoalescer 

func NewKeyCoalescer(strategy CoalesceStrategy, keyColumns ...string) *KeyCoalescer

NewKeyCoalescer creates a key-based coalescer using schema field names.

func (*KeyCoalescer) Coalesce 

func (c *KeyCoalescer) Coalesce(ctx context.Context, schema SchemaInterface, batch Batch) (CoalesceResult, error)

func (*KeyCoalescer) WithKeyFunc 

func (c *KeyCoalescer) WithKeyFunc(fn KeyFunc) *KeyCoalescer

WithKeyFunc overrides column-based key construction.

type KeyFunc 

type KeyFunc func(schema SchemaInterface, record Record) (key string, ok bool)

KeyFunc returns a stable duplicate key for a record.

type MemoryLimitConfig 

type MemoryLimitConfig struct {
	Enabled         bool
	MaxQueueBytes   int64
	AvgRequestBytes int64
	Mode            BackpressureMode
	CheckInterval   time.Duration
	Timeout         time.Duration
}

MemoryLimitConfig provides OOM protection for runtime queue.

type MetricsCapable 

type MetricsCapable[T any] interface {
	// WithMetricsReporter 设置性能监控报告器（返回实现者类型以支持链式）
	WithMetricsReporter(MetricsReporter) T
	// MetricsReporter 返回当前性能监控报告器（可能为 nil）
	MetricsReporter() MetricsReporter
}

Metrics 相关接口设计说明

  • BatchExecutor：仅负责“执行”职责，保持最小接口。
  • MetricsCapable[T]（泛型）：提供“读 + 写”的度量能力，方法返回自类型 T，便于在具体类型或已实例化接口上进行链式配置。 注意：泛型接口在运行时类型断言时必须使用具体的类型实参（如 MetricsCapable[*ThrottledBatchExecutor]）。
  • 兼容性与运行时探测： 在 BatchFlow 等仅持有 BatchExecutor 的位置，无法统一断言到 MetricsCapable[T]， 因此使用一个非泛型只读探测接口（即仅调用 MetricsReporter()）来判断是否已有 Reporter； 若返回 nil，则在本地使用 Noop 兜底，不强制写回（写回需要具体类型 T）。

这允许： - 在需要链式的调用点：以具体类型或已实例化能力接口使用 Set/With 风格链式； - 在框架内部通用路径：通过只读探测保证安全、零开销的观测兜底。

MetricsCapable 扩展接口：支持性能监控报告器（自类型泛型）

type MetricsReporter 

type MetricsReporter interface {
	// 阶段耗时
	ObserveEnqueueLatency(d time.Duration) // Submit -> 入队
	ObserveBatchAssemble(d time.Duration)  // 攒批/组装
	// ObserveExecuteDuration reports execution duration for a schema/table batch.
	// The first argument is historically named table in implementations; non-SQL
	// backends should pass the schema name.
	ObserveExecuteDuration(table string, n int, d time.Duration, status string)

	// 其他观测
	ObserveBatchSize(n int)
	IncError(table string, typ string)
	SetConcurrency(n int)
	SetQueueLength(n int)
	// 在途批次数（不限流也可观察执行压力）
	IncInflight()
	DecInflight()
}

MetricsReporter 统一指标接口（默认 Noop 实现，避免启用前引入开销）

type MockDriver 

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

func NewMockDriver 

func NewMockDriver(databaseType string) *MockDriver

func (*MockDriver) GenerateInsertSQL 

func (d *MockDriver) GenerateInsertSQL(ctx context.Context, schema *SQLSchema, data []map[string]any) (string, []any, error)

GenerateInsertSQL 生成模拟SQL（默认MySQL语法）

type MockExecutor 

type MockExecutor struct {
	ExecutedBatches [][]map[string]any
	// contains filtered or unexported fields
}

Executor 模拟批量执行器（用于测试）

func NewMockExecutor 

func NewMockExecutor() *MockExecutor

NewMockExecutor 创建模拟批量执行器（使用默认Driver）

func NewMockExecutorWithDriver 

func NewMockExecutorWithDriver(driver SQLDriver) *MockExecutor

NewMockExecutorWithDriver 创建模拟批量执行器（使用自定义Driver）

func (*MockExecutor) ExecuteBatch 

func (e *MockExecutor) ExecuteBatch(ctx context.Context, schema SchemaInterface, data []map[string]any) error

ExecuteBatch 模拟执行批量操作

func (*MockExecutor) SnapshotExecutedBatches 

func (e *MockExecutor) SnapshotExecutedBatches() [][]map[string]any

SnapshotExecutedBatches 返回一次性快照，避免并发读写竞态

func (*MockExecutor) SnapshotResults 

func (e *MockExecutor) SnapshotResults() map[string]map[string]int64

SnapshotResults 返回只读快照（拷贝），用于测试收尾输出或断言

type MySQLDriver 

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

func NewMySQLDriver 

func NewMySQLDriver() *MySQLDriver

func (*MySQLDriver) GenerateInsertSQL 

func (d *MySQLDriver) GenerateInsertSQL(ctx context.Context, schema *SQLSchema, data []map[string]any) (string, []any, error)

GenerateInsertSQL 生成MySQL批量插入SQL

type NoopMetricsReporter 

type NoopMetricsReporter struct{}

NoopMetricsReporter 默认关闭时的无操作实现（零开销路径）

func NewNoopMetricsReporter 

func NewNoopMetricsReporter() *NoopMetricsReporter

func (*NoopMetricsReporter) DecInflight 

func (*NoopMetricsReporter) DecInflight()

func (*NoopMetricsReporter) IncError 

func (*NoopMetricsReporter) IncError(string, string)

func (*NoopMetricsReporter) IncInflight 

func (*NoopMetricsReporter) IncInflight()

func (*NoopMetricsReporter) ObserveBatchAssemble 

func (*NoopMetricsReporter) ObserveBatchAssemble(time.Duration)

func (*NoopMetricsReporter) ObserveBatchSize 

func (*NoopMetricsReporter) ObserveBatchSize(int)

func (*NoopMetricsReporter) ObserveEnqueueLatency 

func (*NoopMetricsReporter) ObserveEnqueueLatency(time.Duration)

func (*NoopMetricsReporter) ObserveExecuteDuration 

func (*NoopMetricsReporter) ObserveExecuteDuration(string, int, time.Duration, string)

func (*NoopMetricsReporter) SetConcurrency 

func (*NoopMetricsReporter) SetConcurrency(int)

func (*NoopMetricsReporter) SetQueueLength 

func (*NoopMetricsReporter) SetQueueLength(int)

type ObservabilityConfig 

type ObservabilityConfig struct {
	Observer           Observer
	Logger             *slog.Logger
	Sampler            Sampler
	Redactor           Redactor
	SlowBatchThreshold time.Duration
}

type Observer 

type Observer interface {
	OnBatchEvent(ctx context.Context, event BatchEvent)
}

func NewSlogObserver 

func NewSlogObserver(logger *slog.Logger, sampler Sampler, redactor Redactor, slowBatchThreshold time.Duration) Observer

type ObserverFunc 

type ObserverFunc func(context.Context, BatchEvent)

func (ObserverFunc) OnBatchEvent 

func (f ObserverFunc) OnBatchEvent(ctx context.Context, event BatchEvent)

type OperationMetricsReporter 

type OperationMetricsReporter interface {
	ObserveOperationGenerated(preview OperationPreview)
	IncOperationError(schema string, backend string, stage string, reason string)
}

OperationMetricsReporter is the preferred backend-neutral extension for generated operation diagnostics. Implementations should keep labels low-cardinality and never use raw payloads as labels.

type OperationPreview 

type OperationPreview struct {
	Backend     string
	Operation   string
	Schema      string
	InputItems  int
	OutputItems int
	ArgCount    int
	Fingerprint string
	Attributes  map[string]any
}

OperationPreview is a backend-neutral dry-run summary of generated work. Attributes must stay small and low-cardinality. Raw payloads, SQL args, Redis keys, API bodies, and other sensitive values should not be stored here.

type OperationPreviewer 

type OperationPreviewer interface {
	GenerateOperationPreview(ctx context.Context, schema SchemaInterface, data []map[string]any) (Operations, OperationPreview, error)
}

OperationPreviewer lets processors generate operations and a diagnostic preview together. Custom processors can implement this interface to participate in logs, metrics, and tracing.

type Operations 

type Operations []any

type PipelineConfig 

type PipelineConfig struct {
	BufferSize               uint32
	FlushSize                uint32
	FlushInterval            time.Duration
	MaxConcurrentFlushes     uint32
	DrainOnCancel            bool
	DrainGracePeriod         time.Duration
	FinalFlushOnCloseTimeout time.Duration

	// 可选重试配置（零值=关闭，向后兼容）
	Retry RetryConfig

	// 可选超时配置（零值=关闭，向后兼容）
	Timeout time.Duration

	// 可选指标上报器（零值=关闭，向后兼容）
	MetricsReporter MetricsReporter

	// 可选通用观测配置（结构化日志、采样、脱敏、trace hook）
	Observability ObservabilityConfig

	// 可选并发限制（零值=无限制，向后兼容）
	ConcurrencyLimit int

	// 可选批内合并/去重策略。SQL 默认仍使用 SQLOperationConfig 的 conflict-key 合并。
	Coalescer Coalescer
}

PipelineConfig 管道配置

func DefaultPipelineConfig 

func DefaultPipelineConfig() PipelineConfig

func (PipelineConfig) Validate 

func (c PipelineConfig) Validate() error

type PipelineMetricsReporter 

type PipelineMetricsReporter interface {
	// 出队/取用等待时长（元素在队列中的等待时间）
	// 当前由 BatchFlow 在入队成功后自采样并在 flush 开始时上报，
	// 不依赖 go-pipeline 的 MetricsHook。
	ObserveDequeueLatency(d time.Duration)
	// 管道处理总耗时（与执行器/数据库层的 ObserveExecuteDuration 区分）
	ObserveProcessDuration(d time.Duration, status string)
	// 丢弃/拒绝计数（如队列满、背压拒绝等）
	IncDropped(reason string)
}

PipelineMetricsReporter 是对 go-pipeline v2.2.0 WithMetrics 的可选扩展接口。 - 若实现该接口，框架将把管道级指标事件（通过 pipeline.WithMetrics）桥接到以下方法； - 若未实现，则回退到现有 MetricsReporter 的近似指标或直接忽略（保持向后兼容）。

type PostgreSQLDriver 

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

func NewPostgreSQLDriver 

func NewPostgreSQLDriver() *PostgreSQLDriver

func (*PostgreSQLDriver) GenerateInsertSQL 

func (d *PostgreSQLDriver) GenerateInsertSQL(ctx context.Context, schema *SQLSchema, data []map[string]any) (string, []any, error)

GenerateInsertSQL 生成PostgreSQL批量插入SQL

type Record 

type Record = map[string]any

Record is a single batch item keyed by field or column name.

type Redactor 

type Redactor interface {
	Redact(key string, value any) any
}

func DefaultRedactor 

func DefaultRedactor() Redactor

func NewFieldNameRedactor 

func NewFieldNameRedactor(fields ...string) Redactor

type RedactorFunc 

type RedactorFunc func(key string, value any) any

func (RedactorFunc) Redact 

func (f RedactorFunc) Redact(key string, value any) any

type RedisBatchProcessor 

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

RedisBatchProcessor Redis批量处理器 实现 BatchProcessor 接口，专注于Redis的核心处理逻辑

func NewRedisBatchProcessor 

func NewRedisBatchProcessor(client *redis.Client, driver RedisDriver) *RedisBatchProcessor

NewRedisBatchProcessor 创建Redis批量处理器 参数： - client: Redis客户端连接 - driver: Redis操作生成器

func (*RedisBatchProcessor) ExecuteOperations 

func (rp *RedisBatchProcessor) ExecuteOperations(ctx context.Context, operations Operations) error

Redis 执行与快速退出： - 在设置了 rp.timeout 时，使用 context.WithTimeoutCause 限定执行时限。 - 大批量 operations 时，在循环内检查 ctx（可每次或每 N 次）以快速响应取消/超时，避免无谓迭代开销。 - Pipeline 在本函数内构建并执行，不跨越函数生命周期，defer cancel() 安全。

func (*RedisBatchProcessor) GenerateOperationPreview 

func (rp *RedisBatchProcessor) GenerateOperationPreview(ctx context.Context, schema SchemaInterface, data []map[string]any) (Operations, OperationPreview, error)

func (*RedisBatchProcessor) GenerateOperations 

func (rp *RedisBatchProcessor) GenerateOperations(ctx context.Context, schema SchemaInterface, data []map[string]any) (operations Operations, err error)

GenerateOperations 执行批量操作

func (*RedisBatchProcessor) WithTimeout 

func (rp *RedisBatchProcessor) WithTimeout(timeout time.Duration) *RedisBatchProcessor

type RedisCmd 

type RedisCmd []any

type RedisDriver 

type RedisDriver interface {
	GenerateCmds(ctx context.Context, schema SchemaInterface, data []map[string]any) ([]RedisCmd, error)
}

type RedisPipelineDriver 

type RedisPipelineDriver struct{}

func NewRedisPipelineDriver 

func NewRedisPipelineDriver() *RedisPipelineDriver

func (*RedisPipelineDriver) GenerateCmds 

func (d *RedisPipelineDriver) GenerateCmds(ctx context.Context, schema SchemaInterface, data []map[string]any) ([]RedisCmd, error)

type Request 

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

用来存储请求的数据的各种字段信息和对应的schema

func NewRequest 

func NewRequest(schema SchemaInterface) *Request

func (*Request) Columns 

func (r *Request) Columns() map[string]any

Columns 获取所有列数据

func (*Request) GetBool 

func (r *Request) GetBool(colName string) (bool, error)

func (*Request) GetFloat64 

func (r *Request) GetFloat64(colName string) (float64, error)

func (*Request) GetInt32 

func (r *Request) GetInt32(colName string) (int32, error)

类型化的获取方法

func (*Request) GetInt64 

func (r *Request) GetInt64(colName string) (int64, error)

func (*Request) GetOrderedValues 

func (r *Request) GetOrderedValues() []any

GetOrderedValues 按照 schema 中定义的列顺序返回值

func (*Request) GetString 

func (r *Request) GetString(colName string) (string, error)

func (*Request) GetTime 

func (r *Request) GetTime(colName string) (time.Time, error)

func (*Request) Schema 

func (r *Request) Schema() SchemaInterface

Schema 获取请求的 schema

func (*Request) Set 

func (r *Request) Set(colName string, value any) *Request

通用设置方法

func (*Request) SetBool 

func (r *Request) SetBool(colName string, value bool) *Request

func (*Request) SetBytes 

func (r *Request) SetBytes(colName string, value []byte) *Request

func (*Request) SetFloat32 

func (r *Request) SetFloat32(colName string, value float32) *Request

func (*Request) SetFloat64 

func (r *Request) SetFloat64(colName string, value float64) *Request

func (*Request) SetInt 

func (r *Request) SetInt(colName string, value int) *Request

类型化的设置方法

func (*Request) SetInt8 

func (r *Request) SetInt8(colName string, value int8) *Request

func (*Request) SetInt16 

func (r *Request) SetInt16(colName string, value int16) *Request

func (*Request) SetInt32 

func (r *Request) SetInt32(colName string, value int32) *Request

func (*Request) SetInt64 

func (r *Request) SetInt64(colName string, value int64) *Request

func (*Request) SetNull 

func (r *Request) SetNull(colName string) *Request

func (*Request) SetString 

func (r *Request) SetString(colName string, value string) *Request

func (*Request) SetTime 

func (r *Request) SetTime(colName string, value time.Time) *Request

func (*Request) SetUint 

func (r *Request) SetUint(colName string, value uint) *Request

func (*Request) SetUint8 

func (r *Request) SetUint8(colName string, value uint8) *Request

func (*Request) SetUint16 

func (r *Request) SetUint16(colName string, value uint16) *Request

func (*Request) SetUint32 

func (r *Request) SetUint32(colName string, value uint32) *Request

func (*Request) SetUint64 

func (r *Request) SetUint64(colName string, value uint64) *Request

func (*Request) Validate 

func (r *Request) Validate() error

验证请求是否包含所有必需的列

type RetryConfig 

type RetryConfig struct {
	Enabled     bool
	MaxAttempts int           // 总尝试次数（含首轮），建议 2~3
	BackoffBase time.Duration // 退避基值（指数退避起点）
	MaxBackoff  time.Duration // 最大退避时长（上限）
	// 自定义错误分类（可选）；返回是否可重试与原因标签
	Classifier func(error) (retryable bool, reason string)
}

RetryConfig 可选重试配置（零值关闭）

type RuntimeConfig 

type RuntimeConfig struct {
	ShardCount   uint32
	Routing      ShardRoutingPolicy
	ShardKeyFunc ShardKeyFunc

	Backpressure BackpressureConfig
	MemoryLimit  MemoryLimitConfig
	Adaptive     AdaptiveTuningConfig
}

RuntimeConfig contains runtime controls separate from pipeline and executor settings.

func DefaultRuntimeConfig 

func DefaultRuntimeConfig() RuntimeConfig

type RuntimeEngine 

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

RuntimeEngine owns sharding, routing, backpressure and memory protection.

func NewRuntimeEngine 

func NewRuntimeEngine(ctx context.Context, cfg Config) (*RuntimeEngine, error)

func (*RuntimeEngine) Close 

func (e *RuntimeEngine) Close() error

func (*RuntimeEngine) Done 

func (e *RuntimeEngine) Done() <-chan struct{}

func (*RuntimeEngine) ErrorChan 

func (e *RuntimeEngine) ErrorChan(size int) <-chan error

func (*RuntimeEngine) Submit 

func (e *RuntimeEngine) Submit(ctx context.Context, req *Request) error

func (*RuntimeEngine) Wait 

func (e *RuntimeEngine) Wait() error

type SQLBatchProcessor 

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

SQLBatchProcessor SQL数据库批量处理器 实现 BatchProcessor 接口，专注于SQL数据库的核心处理逻辑

func NewSQLBatchProcessor 

func NewSQLBatchProcessor(db *sql.DB, driver SQLDriver) *SQLBatchProcessor

NewSQLBatchProcessor 创建SQL批量处理器 参数： - db: 数据库连接（用户管理连接池） - driver: 数据库特定的SQL生成器

func (*SQLBatchProcessor) ExecuteOperations 

func (bp *SQLBatchProcessor) ExecuteOperations(ctx context.Context, operations Operations) error

SQL 执行语义：

  • 在设置了 bp.timeout 时，使用 context.WithTimeoutCause 派生子 ctx（具体 cause 如 "execute batch timeout"）。
  • 当子 ctx 达到超时时，驱动通常返回 context.DeadlineExceeded；本处理器会读取 context.Cause(ctx) 并原样返回该 cause， 以便上层执行器的重试分类器可以区分“处理器内部超时”，按需实施重试与退避。
  • 安全性：在执行前校验空 operations，避免越界；不持久化/返回子 ctx，defer cancel() 安全。

func (*SQLBatchProcessor) GenerateOperationPreview 

func (bp *SQLBatchProcessor) GenerateOperationPreview(ctx context.Context, schema SchemaInterface, data []map[string]any) (Operations, OperationPreview, error)

func (*SQLBatchProcessor) GenerateOperations 

func (bp *SQLBatchProcessor) GenerateOperations(ctx context.Context, schema SchemaInterface, data []map[string]any) (operations Operations, err error)

func (*SQLBatchProcessor) GenerateSQLPreview 

func (bp *SQLBatchProcessor) GenerateSQLPreview(ctx context.Context, schema *SQLSchema, data []map[string]any) (SQLPreview, error)

func (*SQLBatchProcessor) WithTimeout 

func (bp *SQLBatchProcessor) WithTimeout(timeout time.Duration) *SQLBatchProcessor

type SQLDedupStats 

type SQLDedupStats struct {
	InputRows        int
	OutputRows       int
	DeduplicatedRows int
	MergedRows       int
}

SQLDedupStats describes client-side duplicate conflict-key handling before SQL generation.

type SQLDriver 

type SQLDriver interface {
	GenerateInsertSQL(ctx context.Context, schema *SQLSchema, data []map[string]any) (sql string, args []any, err error)
}

SQLDriver 数据库特定的SQL生成器接口

type SQLError 

type SQLError struct {
	Stage            SQLStage
	Table            string
	BatchSize        int
	ConflictStrategy ConflictStrategy
	ConflictColumns  []string
	UpdateColumns    []string
	SQLFingerprint   string
	ArgsCount        int
	Cause            error
}

SQLError wraps SQL batch failures with safe diagnostic metadata.

func (*SQLError) Error 

func (e *SQLError) Error() string

func (*SQLError) Unwrap 

func (e *SQLError) Unwrap() error

type SQLMetricsReporter 

type SQLMetricsReporter interface {
	ObserveSQLGenerated(table string, inputRows, outputRows, argsCount int)
	ObserveSQLDeduplicated(table string, strategy ConflictStrategy, deduplicatedRows, mergedRows int)
	IncSQLError(table string, stage SQLStage, reason string)
}

SQLMetricsReporter is an optional SQL-only detail extension kept for SQL diagnostics and backward-compatible Prometheus examples. Backend-neutral integrations should prefer OperationMetricsReporter. Implementations must keep labels low-cardinality; do not use raw SQL as a label.

type SQLOperationConfig 

type SQLOperationConfig struct {
	ConflictStrategy ConflictStrategy
	// ConflictColumns defines the conflict target used by upsert-style writes.
	// When empty, drivers fall back to the first schema column for backward compatibility.
	ConflictColumns []string
	// UpdateColumns limits columns updated by ConflictUpdate. When empty,
	// drivers update all non-conflict columns.
	UpdateColumns []string
	// DeduplicateByConflictColumns controls client-side merge of duplicate
	// conflict keys before generating SQL. The default is true; use
	// WithDeduplicateByConflictColumns(false) to disable it.
	DeduplicateByConflictColumns bool
	// contains filtered or unexported fields
}

操作配置

func (SQLOperationConfig) WithConflictColumns 

func (c SQLOperationConfig) WithConflictColumns(cols ...string) SQLOperationConfig

func (SQLOperationConfig) WithDeduplicateByConflictColumns 

func (c SQLOperationConfig) WithDeduplicateByConflictColumns(enabled bool) SQLOperationConfig

func (SQLOperationConfig) WithUpdateColumns 

func (c SQLOperationConfig) WithUpdateColumns(cols ...string) SQLOperationConfig

type SQLPreview 

type SQLPreview struct {
	Table            string
	SQL              string
	Args             []any
	ArgsCount        int
	Columns          []string
	ConflictStrategy ConflictStrategy
	ConflictColumns  []string
	UpdateColumns    []string
	DedupStats       SQLDedupStats
	Fingerprint      string
}

SQLPreview is a dry-run view of the final SQL operation that would be executed.

Args contains raw values and may include sensitive data. Prefer ArgsCount and Fingerprint for logs and metrics unless the caller explicitly needs full inspection.

func GenerateSQLPreview 

func GenerateSQLPreview(ctx context.Context, driver SQLDriver, schema *SQLSchema, data []map[string]any) (SQLPreview, error)

GenerateSQLPreview builds the same final SQL and args as SQLBatchProcessor without executing it.

func (SQLPreview) OperationPreview 

func (p SQLPreview) OperationPreview() OperationPreview

type SQLSchema 

type SQLSchema struct {
	*Schema
	// contains filtered or unexported fields
}

func NewSQLSchema 

func NewSQLSchema(name string, operationConfig SQLOperationConfig, columns ...string) *SQLSchema

func (*SQLSchema) OperationConfig 

func (s *SQLSchema) OperationConfig() any

type SQLStage 

type SQLStage string

SQLStage identifies where a SQL batch failed. 

const (
	SQLStageValidate SQLStage = "validate"
	SQLStageGenerate SQLStage = "generate"
	SQLStageExecute  SQLStage = "execute"
)

type SQLiteDriver 

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

func NewSQLiteDriver 

func NewSQLiteDriver() *SQLiteDriver

func (*SQLiteDriver) GenerateInsertSQL 

func (d *SQLiteDriver) GenerateInsertSQL(ctx context.Context, schema *SQLSchema, data []map[string]any) (string, []any, error)

GenerateInsertSQL 生成SQLite批量插入SQL

type Sampler 

type Sampler interface {
	ShouldSample(event BatchEvent) bool
}

func NewErrorAndSlowSampler 

func NewErrorAndSlowSampler(slowThreshold time.Duration) Sampler

func NewRatioSampler 

func NewRatioSampler(ratio float64) Sampler

type SamplerFunc 

type SamplerFunc func(BatchEvent) bool

func (SamplerFunc) ShouldSample 

func (f SamplerFunc) ShouldSample(event BatchEvent) bool

type Schema 

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

Schema 表结构定义

func NewSchema 

func NewSchema(
	name string,
	columns ...string,
) *Schema

NewSchema 创建新的Schema实例

func (*Schema) Columns 

func (s *Schema) Columns() []string

func (*Schema) Name 

func (s *Schema) Name() string

type SchemaInterface 

type SchemaInterface interface {
	Name() string
	Columns() []string
}

type ShardKeyFunc 

type ShardKeyFunc func(*Request) uint64

ShardKeyFunc returns routing key for a request.

type ShardRoutingPolicy 

type ShardRoutingPolicy uint8

ShardRoutingPolicy controls how a sharded BatchFlow chooses a target shard. 

const (
	ShardRoutingHash ShardRoutingPolicy = iota
	ShardRoutingRoundRobin
	ShardRoutingLeastLoaded
)

type ThrottledBatchExecutor 

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

ThrottledBatchExecutor 通用批量执行器 架构：ThrottledBatchExecutor -> BatchProcessor -> backend driver/client

设计优势： - 代码复用：SQL、Redis、自定义后端共享执行、限流、重试和观测逻辑 - 职责分离：执行控制与具体处理逻辑分离 - 易于扩展：新增后端只需实现 BatchProcessor，可选实现 OperationPreviewer

func NewRedisThrottledBatchExecutor 

func NewRedisThrottledBatchExecutor(client *redis.Client) *ThrottledBatchExecutor

func NewRedisThrottledBatchExecutorWithDriver 

func NewRedisThrottledBatchExecutorWithDriver(client *redis.Client, driver RedisDriver) *ThrottledBatchExecutor

func NewSQLThrottledBatchExecutorWithDriver 

func NewSQLThrottledBatchExecutorWithDriver(db *sql.DB, driver SQLDriver) *ThrottledBatchExecutor

NewThrottledBatchExecutorWithDriver 创建SQL数据库执行器（推荐方式） 内部使用 SQLBatchProcessor + SQLDriver 组合

func NewThrottledBatchExecutor 

func NewThrottledBatchExecutor(processor BatchProcessor) *ThrottledBatchExecutor

NewThrottledBatchExecutor 创建通用执行器（使用自定义BatchProcessor）

func (*ThrottledBatchExecutor) ExecuteBatch 

func (e *ThrottledBatchExecutor) ExecuteBatch(ctx context.Context, schema SchemaInterface, data []map[string]any) error

ExecuteBatch 执行批量操作

func (*ThrottledBatchExecutor) MetricsReporter 

func (e *ThrottledBatchExecutor) MetricsReporter() MetricsReporter

MetricsReporter 获取指标报告器

func (*ThrottledBatchExecutor) Observer 

func (e *ThrottledBatchExecutor) Observer() Observer

func (*ThrottledBatchExecutor) WithCoalescer 

func (e *ThrottledBatchExecutor) WithCoalescer(coalescer Coalescer) *ThrottledBatchExecutor

func (*ThrottledBatchExecutor) WithConcurrencyLimit 

func (e *ThrottledBatchExecutor) WithConcurrencyLimit(limit int) *ThrottledBatchExecutor

WithConcurrencyLimit 设置并发上限（limit <= 0 表示不启用限流）

func (*ThrottledBatchExecutor) WithMetricsReporter 

func (e *ThrottledBatchExecutor) WithMetricsReporter(metricsReporter MetricsReporter) *ThrottledBatchExecutor

WithMetricsReporter 设置指标报告器

func (*ThrottledBatchExecutor) WithObservability 

func (e *ThrottledBatchExecutor) WithObservability(config ObservabilityConfig) *ThrottledBatchExecutor

func (*ThrottledBatchExecutor) WithObserver 

func (e *ThrottledBatchExecutor) WithObserver(observer Observer) *ThrottledBatchExecutor

func (*ThrottledBatchExecutor) WithRetryConfig 

func (e *ThrottledBatchExecutor) WithRetryConfig(cfg RetryConfig) *ThrottledBatchExecutor

WithRetryConfig 启用/配置重试（仅对 ThrottledBatchExecutor 可用）

type TimeOutCapable 

type TimeOutCapable[T any] interface {
	WithTimeout(time.Duration) T
}

TimeOutCapable 扩展接口：支持超时设置（自类型泛型）

type TuningRecommendation 

type TuningRecommendation struct {
	FlushSize      uint32
	FlushInterval  time.Duration
	BackpressureHW int
	ShardCount     uint32
}

TuningRecommendation is output of adaptive tuner.

type TuningSignal 

type TuningSignal struct {
	LatencyAvg    time.Duration
	QueueDepth    int
	ErrorRate     float64
	ThroughputRPS float64
}

TuningSignal is runtime feedback from engine.

Source Files

View all Source files

Directories

Path Synopsis
examples
metrics/prometheus
test
integration command
sqlite/tools/benchmark command
sqlite/tools/clear-test command
sqlite/tools/config-analysis command
sqlite/tools/path-compatibility command

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.