cdc

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Aug 25, 2026 License: MIT Imports: 29 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	DefaultParquetCompression      = "zstd"
	DefaultParquetCompressionLevel = 3
	DefaultMinRecords              = 20000
	DefaultMaxAgeMs                = 3600000 // 1 hour
	DefaultEstimatedRowBytes       = 1024
	DefaultMaxBatchBytes           = int64(50 * 1024 * 1024)
	DefaultBatchSize               = 10000
	DefaultTargetBaseSizeMB        = 256
	DefaultTargetFileSizeMB        = 256
	DefaultMaxBatchSize            = 10000000 // 10M rows max
	DefaultMaxDeltaSizeMB          = 50
	DefaultDirtyRatioPct           = 5
	DefaultMaxRetries              = 5
	DefaultBaseBackoffMs           = 100
	DefaultMaxBackoffMs            = 10000
	DefaultPGSSLMode               = "require"
)

Default constants

View Source 
const ChecksumSHA256Prefix = "sha256:"

ChecksumSHA256Prefix tags manifest FileEntry.Checksum values produced by ObjectSHA256. Empty Checksum means the entry predates stamping (#347); field presence is the format version signal, matching the Columns rule.

Variables

View Source 
var ErrSchemaAttrCacheUnavailable = errors.New("schema attribute metadata cache unavailable")

ErrSchemaAttrCacheUnavailable marks a CDC export that cannot proceed because a schema's attribute metadata cache could not be resolved — the registry lookup failed or the cache is empty. CDC export needs it to produce parquet the federated reader can consume; without it the reader fails fast with ErrSchemaMetadataCacheRequired (#193). This is an operator-visible configuration error, not write-path validation, so it is NOT wrapped in forma.ErrInvalidInput.

View Source 
var ErrSchemaLockContended = errors.New("schema advisory lock contended")

ErrSchemaLockContended reports that another holder (flusher, cdc-init, or manifest-reconcile) owns the per-schema advisory lock.

Functions

func BuildBasePath 

func BuildBasePath(prefix string, schemaID int16, minRowID, maxRowID string) string

BuildBasePath returns the canonical base file path for a schema. Base files use min/max row_id naming to indicate the row range covered.

func BuildBaseTempPath 

func BuildBaseTempPath(prefix string, schemaID int16, fileUUID string) string

BuildBaseTempPath returns the temp path for a base file during init.

func BuildDeltaPath 

func BuildDeltaPath(prefix string, schemaID int16, fileUUID string) string

BuildDeltaPath returns the canonical delta file path for a schema.

func BuildMergedBasePath added in v0.2.0 

func BuildMergedBasePath(prefix string, schemaID int16, fileUUID string) string

BuildMergedBasePath returns the path of a compaction-rewritten base file. Merged bases are UUID-named, never {min}_{max}: repeated rewrites over the same row range would otherwise reuse a key and overwrite an object still listed in the manifest under concurrent readers (#188). The row-ID range lives in the manifest FileEntry instead.

func BuildPGDSN added in v0.2.0 

func BuildPGDSN(p PGDSNParams) string

BuildPGDSN renders a libpq keyword/value DSN with every string value quoted (single-quote wrapped, backslash and single-quote escaped) so passwords with spaces or quotes survive parsing by pgx and DuckDB's postgres scanner.

The implementation moved to internal/pgdsn so internal/federated can quote the postgres_scan DSN the same way (#301). Behaviour is unchanged; the tests in pgdsn_test.go and redact_test.go remain the contract.

func BuildTempPath 

func BuildTempPath(prefix string, schemaID int16, fileUUID string) string

BuildTempPath returns the temp path for a delta file.

func CopyTmpToFinal 

func CopyTmpToFinal(ctx context.Context, client S3ObjectClient, bucket, tmpKey, finalKey string, logger *zap.Logger) error

CopyTmpToFinal copies a parquet file from tmp key to final key and deletes tmp.

func DeleteObjectKey added in v0.2.0 

func DeleteObjectKey(ctx context.Context, client S3ObjectClient, bucket, key string) error

DeleteObjectKey deletes one S3 object.

func GetChangeLogStats 

func GetChangeLogStats(ctx context.Context, db *sql.DB, table string, schemaID int16) (int64, int64, error)

GetChangeLogStats returns count and oldest changed_at for unflushed rows.

func HeadObjectSize added in v0.2.0 

func HeadObjectSize(ctx context.Context, client S3ObjectClient, bucket, key string) (int64, error)

HeadObjectSize returns the byte size of an S3 object. Callers use it to populate manifest FileEntry.SizeBytes after a tmp->final copy; the size feeds compaction's promotion heuristic only, so callers should treat a failure as best-effort (log and keep 0) rather than failing the pipeline.

func MarkFlushedVersions added in v0.2.0 

func MarkFlushedVersions(ctx context.Context, db *sql.DB, table string, schemaID int16, rowIDs []uuid.UUID, versions map[uuid.UUID]int64, flushedAt int64) ([]uuid.UUID, error)

MarkFlushedVersions updates flushed_at for the given rows only where the slot-0 changed_at still EQUALS the version LISTED for that row at batch selection. Exact equality is what keeps export and mark consistent without a global wall-clock cutoff: any slot whose version differs from its listing was concurrently rewritten — advanced by an update/delete, or replaced by a delete→recreate — and its exported copy (if any) no longer matches slot-0, so the row must stay dirty for the next run. (`<=` is NOT safe here: a recreate overwrites the slot, and before #274's create ordering it could even land BELOW a clock-ahead tombstone's listing — a `<=` mark would clear the dirty barrier for a payload no parquet holds.) A clock-ahead listed version matches its own listing and marks normally (review round 2 P1). Returns the row_ids actually marked flushed.

func ObjectSHA256 added in v0.2.0 

func ObjectSHA256(ctx context.Context, client S3GetClient, bucket, key string) (string, error)

ObjectSHA256 streams an object and returns "sha256:<hex>" over its bytes. The hash is of what the store returns, not what the writer sent: it blesses the published state, so it detects later mutation, not upload mangling.

func ResolveStaticS3Credentials added in v0.2.0 

func ResolveStaticS3Credentials(cfg CDCConfig) (accessKeyID, secretAccessKey, sessionToken string)

ResolveStaticS3Credentials resolves the static S3 credential triple shared by every CDC credential site — the SDK S3 client and the DuckDB httpfs plumbing (#326). Explicit config wins as-is. Otherwise the environment pair applies only when BOTH halves are non-empty: a lone AWS_ACCESS_KEY_ID would build an empty-secret static provider whose only observable behavior is an opaque signing failure (#302). With no fully-set pair it returns empty strings — SDK callers then leave the default chain in place, and DuckDB inherits its own environment chain.

The session token rides the source that supplied the pair and never crosses sources (#329): a config pair carries only the config token, an environment pair carries only AWS_SESSION_TOKEN. Pairing a long-lived key with a foreign temporary token yields a combination the storage endpoint can only report as an opaque signing failure. A token with no accompanying pair is not a credential source at all and is discarded with the rest.

func RunOnce 

func RunOnce(ctx context.Context, cfg CDCConfig, s3Client S3ObjectClient, dryRun bool, logger *zap.Logger, schemaRegistry forma.SchemaRegistry) error

RunOnce performs one full pass over schemas and attempts flush where needed. Caller may provide an S3ObjectClient; when nil, AWS config will be loaded from environment (still respecting cfg.S3Region). Optional schemaRegistry enables schema-aware projections in DuckDB export.

func SelectBatchRowIDs 

func SelectBatchRowIDs(ctx context.Context, db *sql.DB, table string, schemaID int16, batchSize int) ([]uuid.UUID, map[uuid.UUID]int64, int64, error)

SelectBatchRowIDs picks up to batchSize row_ids for flushing, recording each listed row's changed_at version, and returns a snapshot cutoff (ms) for exporting. The snapshot is the MAX of the wall clock and the listed versions: per-row versions are strictly monotonic and may run ahead of the wall clock (#274 GREATEST ordering), so a wall-clock-only cutoff would exclude a clock-ahead row from both export and mark — shipping an empty delta and leaving the row dirty until the clock caught up, indefinitely under sustained lead (review round 2 P1). The versions feed MarkFlushedVersions, which marks exactly what was listed.

func TrySchemaLock added in v0.2.0 

func TrySchemaLock(ctx context.Context, db *sql.DB, schemaID int16) (bool, func(), error)

TrySchemaLock pins one physical connection and takes pg_try_advisory_lock(schemaID, schemaID). The lock is session-scoped: on a pool, acquire and release could land on different connections, and a lock released on the wrong session silently fails — so unlock runs on the same pinned conn and closes it to end the session. unlock is non-nil iff locked; it uses a background context so the lock is released even after ctx cancel.

Types

type CDCConfig 

type CDCConfig struct {
	// Table names
	ChangeLogTable  string
	EntityMainTable string
	EAVDataTable    string

	// Thresholds
	MinRecords int   // flush when unflushed rows >= MinRecords
	MaxAgeMs   int64 // flush when oldest unflushed row age >= MaxAgeMs
	BatchSize  int   // maximum rows per snapshot

	// Postgres connection
	PGHost     string
	PGPort     int
	PGUser     string
	PGPassword string
	PGDB       string
	PGUseIAM   bool
	PGSSLMode  string

	// DuckDB export options
	DuckDBPath              string        // optional on-disk duckdb; empty for :memory:
	DuckThreads             int           // PRAGMA threads
	DuckMemLimit            string        // e.g. "4GB"
	QueryTimeout            time.Duration // timeout for duckdb export
	ParquetCompression      string        // e.g. "zstd"
	ParquetCompressionLevel int           // codec level if supported
	EstimatedRowBytes       int           // rough row size estimate for batch sizing
	MaxBatchBytes           int64         // optional cap to limit batch size by bytes

	// Init-specific options (cdc-init base file export)
	TargetFileSizeMB int // target parquet file size in MB (0 = use BatchSize)
	MaxBatchSize     int // maximum rows per batch to cap memory usage

	// S3
	S3Bucket          string
	S3Prefix          string // prefix inside bucket for delta files
	S3Endpoint        string
	S3Region          string
	S3UseSSL          bool
	S3UsePath         bool   // path style addressing
	S3AccessKeyID     string // AWS access key ID; overrides environment variable AWS_ACCESS_KEY_ID
	S3SecretAccessKey string // AWS secret access key; overrides environment variable AWS_SECRET_ACCESS_KEY
	S3SessionToken    string // AWS session token for temporary credentials (STS/roles); effective only with same-source S3AccessKeyID/S3SecretAccessKey, see ResolveStaticS3Credentials (#329)

	// Manifest (optional - when set, flush updates manifest after export)
	ManifestPrefix   string // root prefix for manifests in S3
	ManifestTemplate string // path template, e.g. "manifest/{{.SchemaID}}.json"

	// BeforeExportHook, when non-nil, runs per batch after dirty-ID selection
	// (the snapshot is already captured) and before the DuckDB export. Test
	// seam for driving mutations inside the selection->export race window
	// (#182); always nil in production. A hook error aborts the batch before
	// any side effect.
	BeforeExportHook func(ctx context.Context, schemaID int16, batchIDs []uuid.UUID, snapshot int64) error
}

CDCConfig controls change_log flushing and export behavior. S3 client is injected separately via RunOnce parameter to allow callers to provide either AWS or MinIO implementations.

func (CDCConfig) WithDefaults 

func (c CDCConfig) WithDefaults() CDCConfig

WithDefaults returns a copy of CDCConfig with missing fields set to defaults.

type CompactionConfig 

type CompactionConfig struct {
	SchemaID         int16  // optional filter; 0 means all
	ManifestPath     string // s3 path or fs path to manifest JSON for the schema
	TargetBaseSizeMB int    // default 256
	// TargetBaseSizeBytes is the byte-precise promotion threshold. Zero
	// derives it from TargetBaseSizeMB; the compactor compares delta-tier
	// bytes against it directly, so sub-MB tiers are not truncated to 0 MB.
	TargetBaseSizeBytes int64
	MaxDeltaSizeMB      int           // default 50
	DirtyRatioPct       int           // default 5 (rewrite when updated rows/base rows > 5%)
	RunInterval         time.Duration // scheduler interval when used in a loop
	TempPrefix          string        // temp path prefix for rewrites
	MaxParallelFiles    int           // optional parallelism for rewrites

	// SkipInputChecksumVerify opts out of the pre-merge verification of
	// rewrite inputs (#347). The zero value verifies: a rewrite merges its
	// sources and then deletes them, so the gate is the last moment silent
	// corruption is both detectable and attributable to a named object, and
	// that has to hold for deployments that never set this struct field.
	// Setting it true trades that detection away to save one GET per stamped
	// source.
	SkipInputChecksumVerify bool

	// Backoff parameters for S3/manifest operations
	MaxRetries  int           // default 5
	BaseBackoff time.Duration // default 100ms
	MaxBackoff  time.Duration // default 10s
}

CompactionConfig controls Base/Delta maintenance.

func (CompactionConfig) WithDefaults 

func (c CompactionConfig) WithDefaults() CompactionConfig

WithDefaults returns a copy of CompactionConfig with missing fields set to defaults.

type DuckExporter 

type DuckExporter struct {
	DB     *sql.DB
	Logger *zap.Logger
}

DuckExporter handles DuckDB interactions for exporting snapshots to S3 temp path.

func NewDuckExporter 

func NewDuckExporter(ctx context.Context, cfg CDCConfig, s3AccessKey, s3Secret, s3SessionToken string, logger *zap.Logger) (*DuckExporter, error)

NewDuckExporter opens a DuckDB pool whose every physical connection self-configures (pragmas, extensions, S3 session settings) via a connector init hook — session-scoped statements issued through the pool reach only one arbitrary connection (#285, same class as #245). Init statement failures are logged and skipped, never blocking the connection; construction fails only on credential validation or ping.

The caller passes the whole key/secret/token triple because a session token is only valid with the pair it was issued alongside; resolving it here would let a token from one source pair with a key from another (#329, see ResolveStaticS3Credentials).

func (*DuckExporter) ExportBaseFileToTmp 

func (e *DuckExporter) ExportBaseFileToTmp(ctx context.Context, cfg CDCConfig, pgConnStr string, s3TmpPath string, schemaID int16, rowIDs []uuid.UUID, attrCache forma.SchemaAttributeCache) error

ExportBaseFileToTmp exports existing entity_main + eav_data rows directly to a base parquet file. Unlike delta export, this does NOT use change_log - it reads directly from entity_main. s3TmpPath is the destination like 's3://bucket/base/<schema_id>/_tmp/<tmp_uuid>.parquet'

func (*DuckExporter) ExportSnapshotToTmp 

func (e *DuckExporter) ExportSnapshotToTmp(ctx context.Context, cfg CDCConfig, pgConnStr string, s3TmpPath string, schemaID int16, snapshotTS int64, rowIDs []uuid.UUID, attrCache forma.SchemaAttributeCache) error

ExportSnapshotToTmp builds an export SQL and runs COPY to the provided s3TmpPath. s3TmpPath is the destination like 's3://bucket/prefix/<schema_id>/_tmp/<tmp_uuid>.parquet'

type InitOptions added in v0.2.0 

type InitOptions struct {
	Config               CDCConfig
	S3Client             *s3.Client
	SchemaRegistryTable  string
	SchemaIDFilter       int
	DryRun               bool
	AutoEstimateRowBytes bool
	Logger               *zap.Logger
	SchemaRegistry       forma.SchemaRegistry
}

InitOptions configures a base-file initialization export run. It mirrors the cdc-init CLI options; the CLI in cmd/tools delegates here.

type InitSummary added in v0.2.0 

type InitSummary struct {
	TotalRowsExported int64
	TotalFilesCreated int
}

InitSummary reports totals across all schemas processed by RunInit.

func RunInit added in v0.2.0 

func RunInit(ctx context.Context, opts InitOptions) (InitSummary, error)

RunInit performs the initialization export for all or a specific schema: existing non-deleted entity_main rows are batched into base parquet files on S3 and (when a manifest template is configured) recorded in the schema manifest. Extracted from cmd/tools/cdc_init.go (mechanical move, #173).

type Manifest 

type Manifest struct {
	SchemaID int16           `json:"schema_id"`
	Base     []ManifestEntry `json:"base,omitempty"`
	Delta    []ManifestEntry `json:"delta,omitempty"`
}

Manifest captures the files for a schema and their stats.

type ManifestConfig 

type ManifestConfig struct {
	Bucket       string
	Prefix       string // root prefix for manifests (e.g., manifest/<project>/)
	PathTemplate string // optional template per schema, e.g., "manifest/{{.SchemaID}}.json"
}

ManifestConfig describes where manifests are stored and how to resolve parquet paths.

type ManifestEntry 

type ManifestEntry struct {
	Path        string `json:"path"`
	MinRowID    string `json:"min_row_id,omitempty"`
	MaxRowID    string `json:"max_row_id,omitempty"`
	MinTimeSlot int64  `json:"min_time_slot,omitempty"`
	MaxTimeSlot int64  `json:"max_time_slot,omitempty"`
	SizeBytes   int64  `json:"size_bytes,omitempty"`
	Level       string `json:"level,omitempty"` // "base" or "delta"
}

ManifestEntry describes a parquet file tracked for a schema. It can represent either a base or delta file.

type PGDSNParams added in v0.2.0 

type PGDSNParams = pgdsn.Params

PGDSNParams are the inputs for a libpq keyword/value connection string.

type Runner added in v0.0.27 

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

Runner caches per-config S3 runtimes and DuckDB exporters across RunOnce calls. Each cache holds one entry per non-credential config group; a credential rotation (#329) replaces the group's entry, and a superseded DuckDB exporter is closed once no in-flight RunOnce holds it (#331) — so a long-lived Runner under STS rotation keeps exactly one exporter per group, not one per token. Two configs sharing a group but alternating *different* static credential triples therefore rebuild on every alternation: a rebuild cost, accepted in #331, never a correctness issue.

func NewRunner added in v0.0.27 

func NewRunner(logger *zap.Logger) *Runner

func (*Runner) Close added in v0.0.27 

func (r *Runner) Close() error

func (*Runner) RunOnce added in v0.0.27 

func (r *Runner) RunOnce(ctx context.Context, cfg CDCConfig, s3Client S3ObjectClient, dryRun bool, schemaRegistry forma.SchemaRegistry) error

type S3FullClient 

type S3FullClient interface {
	S3ObjectClient
	GetObject(ctx context.Context, params *s3.GetObjectInput, optFns ...func(*s3.Options)) (*s3.GetObjectOutput, error)
	PutObject(ctx context.Context, params *s3.PutObjectInput, optFns ...func(*s3.Options)) (*s3.PutObjectOutput, error)
}

S3FullClient extends S3ObjectClient with GetObject and PutObject for manifest operations.

type S3GetClient added in v0.2.0 

type S3GetClient interface {
	GetObject(ctx context.Context, params *s3.GetObjectInput, optFns ...func(*s3.Options)) (*s3.GetObjectOutput, error)
}

S3GetClient is the minimal byte-read surface content hashing needs.

type S3ObjectClient 

type S3ObjectClient interface {
	CopyObject(ctx context.Context, params *s3.CopyObjectInput, optFns ...func(*s3.Options)) (*s3.CopyObjectOutput, error)
	DeleteObject(ctx context.Context, params *s3.DeleteObjectInput, optFns ...func(*s3.Options)) (*s3.DeleteObjectOutput, error)
	HeadObject(ctx context.Context, params *s3.HeadObjectInput, optFns ...func(*s3.Options)) (*s3.HeadObjectOutput, error)
}

S3ObjectClient is a minimal interface for copy + delete + stat used by the CDC flusher.

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.