manifest

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: 17 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func AppendFile 

func AppendFile(ctx context.Context, st Store, path string, schemaID int16, entry FileEntry) error

AppendFile adds a FileEntry to the manifest and saves it. Uses optimistic locking via etag to handle concurrent updates.

func AppendFiles 

func AppendFiles(ctx context.Context, st Store, path string, schemaID int16, entries []FileEntry) error

AppendFiles adds multiple FileEntry items to the manifest and saves it. Uses optimistic locking via etag to handle concurrent updates.

func IsNotFound added in v0.2.0 

func IsNotFound(err error) bool

IsNotFound reports whether a Store.Load error means the object does not exist. Most S3-compatible stores return an error containing "NoSuchKey" or "not found".

func ListPaths 

func ListPaths(m *Manifest, tier string) []string

ListPaths returns paths for the given tier.

func ReplaceTierFiles added in v0.2.0 

func ReplaceTierFiles(ctx context.Context, st Store, path string, schemaID int16, tier string, entries []FileEntry) error

ReplaceTierFiles replaces every manifest entry of the given tier with the provided entries, preserving entries of other tiers in their original order. cdc-init is a full re-export of a schema's live rows, so after a run the base tier's canonical inventory IS that run's output: stale entries from earlier runs (different batch ranges) and historical duplicates must not survive (#176). Compaction-promoted base entries are replaced too — after a full re-export their live content is subsumed by the new base files; their S3 objects remain for glob-based readers, and object-level reconciliation is #203. Uses optimistic locking via etag.

func S3ExistsProbe added in v0.2.0 

func S3ExistsProbe(client S3ProbeClient, bucket string) func(ctx context.Context, key string) (bool, error)

S3ExistsProbe returns a QuerySource.Exists probe backed by HeadObject. Only a confirmed 404 answers (false, nil): the SDK's *types.NotFound, or an HTTP response error with status 404 for stores whose bodyless HEAD reply the SDK cannot model as NotFound. Every other failure propagates wrapped — a probe that merely failed does not prove the object is gone, and misreading it would fabricate a data-loss report (see QuerySource.MissingIn).

func S3FallbackGlob added in v0.2.0 

func S3FallbackGlob(bucket, dataPrefix string) func(schemaID int16) string

S3FallbackGlob returns the legacy per-schema parquet glob used for schemas with no manifest entries yet, preserving pre-manifest read behavior. The single `*` does not cross `/`, so in-flight `_tmp/` objects under the schema prefix stay excluded — it must never be widened to `**`.

Writer parity is required: the CDC writers canonicalize the same prefix with strings.TrimSuffix(prefix, "/") before joining (internal/cdc BuildDeltaPath / BuildBasePath), so a configured "delta/" writes objects at "delta/{schemaID}/...". Trimming identically here is what keeps the glob pointed at the objects that were actually written; without it the extra separator produces an empty path segment and the fallback matches nothing — a silently empty cold tier. The trim is a single TrimSuffix, not TrimRight, precisely because that is what the writers do: mirroring their behavior matters more than being maximally forgiving.

A prefix that canonicalizes to empty ("" or "/") returns a nil func, which QuerySource.Paths reads as "no fallback" — a glob built from an empty prefix would scan the bucket root.

func Save 

func Save(ctx context.Context, st Store, path string, m *Manifest, etag string) (string, error)

Save writes manifest with updated timestamp and optional optimistic etag.

func SpliceTierFiles added in v0.2.0 

func SpliceTierFiles(m *Manifest, tier string, entries []FileEntry)

SpliceTierFiles replaces every entry of the given tier on the in-memory manifest with the provided entries, preserving other tiers in their original order. Extracted from ReplaceTierFiles so callers that manage their own etag/save cycle — manifest-reconcile's 412-retried init promotion (#292) — reuse the exact replacement semantics.

Types

type FSStore 

type FSStore struct {
	Root fs.FS
}

FSStore implements Store on local filesystem.

func (*FSStore) Load 

func (f *FSStore) Load(ctx context.Context, path string) ([]byte, string, error)

func (*FSStore) Save 

func (f *FSStore) Save(ctx context.Context, path string, data []byte, etag string) (string, error)

type FileEntry 

type FileEntry struct {
	Tier       string `json:"tier"`
	Path       string `json:"path"`
	RowIDMin   string `json:"row_id_min"`
	RowIDMax   string `json:"row_id_max"`
	CreatedMin int64  `json:"created_min"`
	CreatedMax int64  `json:"created_max"`
	SizeBytes  int64  `json:"size_bytes"`
	RowCount   int64  `json:"row_count"`
	Checksum   string `json:"checksum,omitempty"`
	// Columns records the file's parquet footer schema (column name → DuckDB
	// type), stamped by the writer from a DESCRIBE of the object it just
	// wrote (#256). Readers use it to validate the #189 system-column
	// invariant and build the #255 column union without a footer probe.
	// Nil/absent means the entry predates stamping: readers fall back to
	// probing, so no manifest migration or version bump is needed — field
	// presence is the format version signal.
	Columns map[string]string `json:"columns,omitempty"`
}

FileEntry describes a single parquet file tracked by the manifest. Tier is typically "base" or "delta".

func FilterByTier 

func FilterByTier(m *Manifest, tier string) []FileEntry

FilterBySchema returns files matching the tier (use "" for all tiers).

type Manifest 

type Manifest struct {
	SchemaID    int16       `json:"schema_id"`
	Version     int64       `json:"version"`
	UpdatedAtMs int64       `json:"updated_at_ms"`
	Files       []FileEntry `json:"files"`
}

Manifest holds per-schema parquet inventory.

func Decode 

func Decode(r io.Reader) (*Manifest, error)

Decode reads from reader.

func Load 

func Load(ctx context.Context, st Store, path string) (*Manifest, string, error)

Load reads manifest from store.

func LoadOrCreate 

func LoadOrCreate(ctx context.Context, st Store, path string, schemaID int16) (*Manifest, string, error)

LoadOrCreate loads an existing manifest or creates a new empty one for the schema. Returns the manifest, etag (empty if new), and any error.

func Parse 

func Parse(data []byte) (*Manifest, error)

Parse decodes manifest JSON bytes.

type PathResolver 

type PathResolver struct {
	Prefix       string
	PathTemplate string
}

PathResolver builds manifest paths per schema using a template. Defaults to "manifest/{{.SchemaID}}.json" under an optional prefix.

func (PathResolver) Resolve 

func (r PathResolver) Resolve(schemaID int16) (string, error)

Resolve returns the full path for a schema manifest.

type QuerySource added in v0.2.0 

type QuerySource struct {
	Store    Store
	Resolver PathResolver
	// Bucket prefixes manifest FileEntry paths (bucket-relative keys) into
	// full s3:// URIs for DuckDB.
	Bucket string
	// Exists probes one bucket-relative key for existence (e.g. HeadObject).
	// Nil disables missing-key classification (MissingKeys reports none).
	Exists func(ctx context.Context, key string) (bool, error)
	// Fallback, when set, supplies the path set for schemas whose manifest
	// is missing or empty — typically the legacy per-schema glob, preserving
	// pre-manifest read behavior for never-flushed schemas.
	Fallback func(schemaID int16) string
}

QuerySource resolves a schema's parquet object set for federated reads (#187): Paths returns the manifest-listed objects as s3:// URIs and MissingKeys reports listed keys absent from storage, so the read path can classify a failed scan as manifest inconsistency by storage state instead of driver message text.

Direction contract: manifest ⊆ live objects is required — a listed key missing from storage is loud (the read fails and classifies). Extra unlisted objects are tolerated and invisible to reads; object-level reconciliation is #203's scope. That one-directional contract is what makes this safe against the CDC write windows: both the flusher and init copy the final object to storage before listing it in the manifest, so a listed-but-absent key can only mean loss, never in-flight publication.

func NewS3QuerySource added in v0.2.0 

func NewS3QuerySource(client S3ProbeClient, cfg S3QuerySourceConfig) *QuerySource

NewS3QuerySource assembles the manifest-driven QuerySource used by federated reads (#250), centralizing the wiring the production e2e harness has been carrying inline (internal/e2e_harness/production/ engine.go parquetSource). Reads scan exactly the manifest-listed objects, listed-but-absent keys classify as an inconsistent parquet set, and never-flushed schemas fall back to the legacy per-schema glob.

func (*QuerySource) MissingIn added in v0.2.0 

func (s *QuerySource) MissingIn(ctx context.Context, scanned []string) ([]string, error)

MissingIn probes the given scanned URIs and returns the bucket-relative keys absent from storage. It deliberately probes the exact set the failed scan used rather than reloading the manifest: a concurrent flush/compaction could otherwise swap in a newer snapshot that hides the lost key or lists an unrelated one. Glob entries and URIs outside this source's bucket are skipped — their absence cannot be proven with the configured probe, and an unprovable key must not fabricate inconsistency.

func (*QuerySource) Paths added in v0.2.0 

func (s *QuerySource) Paths(ctx context.Context, schemaID int16) ([]string, map[string]map[string]string, error)

Paths returns the schema's manifest-listed parquet objects as s3:// URIs, or the Fallback glob when the schema has no manifest entries yet. The manifest format accepts both bucket-relative keys (what the CDC writers produce) and absolute s3:// URIs — absolute entries pass through unchanged instead of being double-prefixed (#249 review).

It also returns each stamped entry's write-time footer columns (#256), keyed by the SAME string returned for that entry in paths — relative keys carry the bucket prefix, absolute entries the passed-through URI — so the pre-read validator can look a stamp up by scanned path without re-deriving URIs. Entries written before stamping existed (no Columns) and the fallback glob, which names no entry at all, contribute no key; those paths probe as before. The returned maps alias the loaded manifest and must not be mutated by callers.

type S3Client added in v0.0.24 

type S3Client interface {
	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)
}

type S3ProbeClient added in v0.2.0 

type S3ProbeClient interface {
	S3Client
	HeadObject(ctx context.Context, params *s3.HeadObjectInput, optFns ...func(*s3.Options)) (*s3.HeadObjectOutput, error)
}

S3ProbeClient is the S3 surface a manifest-driven QuerySource needs: the manifest object IO of S3Client plus HeadObject for existence probes.

type S3QuerySourceConfig added in v0.2.0 

type S3QuerySourceConfig struct {
	// Bucket holds both the manifests and the parquet objects they list.
	Bucket string
	// ManifestPrefix is the root prefix for manifest objects.
	ManifestPrefix string
	// ManifestTemplate is the per-schema manifest path template (e.g.
	// "manifest/{{.SchemaID}}.json"). Callers gate on it being non-empty;
	// this package adds no default of its own.
	ManifestTemplate string
	// DataPrefix is the parquet prefix used to build the legacy fallback
	// glob for schemas with no manifest entries. Empty disables the
	// fallback entirely.
	DataPrefix string
}

S3QuerySourceConfig is the S3 wiring for a manifest-driven QuerySource. It mirrors the CDC/compaction write side: ManifestPrefix and ManifestTemplate must match the writers' manifest layout, and DataPrefix must match the writers' parquet prefix, or reads resolve a different object set than the one being written.

type S3Store 

type S3Store struct {
	Client S3Client
	Bucket string
}

S3Store implements Store using AWS S3-compatible APIs.

func (*S3Store) Load 

func (s *S3Store) Load(ctx context.Context, path string) ([]byte, string, error)

func (*S3Store) Save 

func (s *S3Store) Save(ctx context.Context, path string, data []byte, etag string) (string, error)

type Store 

type Store interface {
	Load(ctx context.Context, path string) (data []byte, etag string, err error)
	Save(ctx context.Context, path string, data []byte, etag string) (newETag string, err error)
}

Store abstracts load/save operations (could be S3 or local FS).

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.