Documentation
¶
Overview ¶
Package compaction merges small Parquet files within a partition into larger files, reducing S3 list overhead and scan file-open costs.
Index ¶
- Constants
- type BackgroundCompactor
- type BackgroundConfig
- type CompactionFailed
- type Compactor
- func (c *Compactor) CompactTable(ctx context.Context, tableName string) (*Result, error)
- func (c *Compactor) FlushDeferredDeletes(ctx context.Context) int
- func (c *Compactor) ForceCompactFile(ctx context.Context, tableName string, filePath string, ...) error
- func (c *Compactor) PublicationConflicts() int64
- func (c *Compactor) RewriteTable(ctx context.Context, tableName string) (*Result, error)
- type Config
- type PartitionFailure
- type Result
Constants ¶
const DefaultCompactionInterval = 5 * time.Minute
const DefaultDeleteGrace = 30 * time.Minute
DefaultDeleteGrace keeps compacted-away bytes alive long enough for any in-flight query dispatched against the old manifest to finish reading them.
const DefaultGCMinAge = 30 * time.Minute
DefaultGCMinAge is the minimum age before delete markers are eligible for GC. Set to 30 minutes to safely exceed the duration of long-running analytical queries, preventing GC from rewriting files that are being scanned.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type BackgroundCompactor ¶
type BackgroundCompactor struct {
// contains filtered or unexported fields
}
BackgroundCompactor runs periodic compaction sweeps across all tables.
func NewBackgroundCompactor ¶
func NewBackgroundCompactor(cat *catalog.Catalog, cfg BackgroundConfig, logger *slog.Logger) *BackgroundCompactor
NewBackgroundCompactor creates a background compactor.
func (*BackgroundCompactor) Start ¶
func (bc *BackgroundCompactor) Start(ctx context.Context)
Start launches the background compaction loop.
type BackgroundConfig ¶
type BackgroundConfig struct {
// Enabled controls whether background compaction runs. Default: true.
Enabled bool
// Interval between compaction sweeps. Zero uses DefaultCompactionInterval.
Interval time.Duration
// Compaction controls compaction trigger thresholds.
Compaction Config
// GCMinAge is the minimum age before delete markers are garbage collected
// and their files are force-rewritten. Zero uses DefaultGCMinAge.
GCMinAge time.Duration
// DropGrace is the minimum age before a dropped table's data files are
// physically deleted (catalog.Catalog.FlushDroppedTableFiles). Zero
// uses catalog.DefaultDropTableGrace. Only consulted when
// ReclaimDroppedTables is true.
DropGrace time.Duration
// ReclaimDroppedTables controls whether the sweep calls
// catalog.Catalog.FlushDroppedTableFiles at all. Default: false.
//
// Deliberately opt-in rather than tied to Enabled (#494 review): this
// *Catalog is not the only one a DROP can go through. An embedded
// wadjet.DB and a standalone pgwire DB each own a separate *Catalog
// from the one a BackgroundCompactor sweeps (cmd/wadjet/main.go's
// standalone mode is the concrete case — its pgwire server opens its
// own wadjet.DB), so turning this on unconditionally would reclaim a
// DROPped table's files for queries issued through the compactor's
// catalog while silently never reclaiming ones issued through psql or
// the embedded API against the others. An explicit, honest default of
// "not reclaimed anywhere yet" beats an inconsistent "reclaimed here,
// not there" that looks like a bug until someone reads the wiring.
// Turning it on is safe wherever it runs: FlushDroppedTableFiles's
// live-manifest guard and table-prefix scoping hold regardless of
// which *Catalog instance calls it.
ReclaimDroppedTables bool
}
BackgroundConfig controls the periodic compaction loop.
type CompactionFailed ¶ added in v0.18.1
type CompactionFailed struct {
Table string
Failures []PartitionFailure
Compacted int
}
CompactionFailed is the aggregate error CompactTable and RewriteTable return when at least one partition's merge failed.
One error for the whole table, not the first failure, because a merge failure is scoped to the partition whose files could not be read: #435 correctly made a failed merge visible to the caller, but by RETURNING at the first one, so a single drifted partition froze compaction of every other partition in the table — and, since the background sweep `continue`s to the next table on an error from here, the table's delete-marker GC with it.
Compacted lets a caller tell a partial run from a total one without inspecting the Result: zero means nothing in this table compacted, non-zero means the named partitions failed while the rest went through.
func (*CompactionFailed) Error ¶ added in v0.18.1
func (e *CompactionFailed) Error() string
func (*CompactionFailed) Partial ¶ added in v0.18.1
func (e *CompactionFailed) Partial() bool
Partial reports whether some partitions compacted despite the failures.
func (*CompactionFailed) Unwrap ¶ added in v0.18.1
func (e *CompactionFailed) Unwrap() []error
Unwrap exposes the individual causes to errors.Is and errors.As.
type Compactor ¶
type Compactor struct {
// contains filtered or unexported fields
}
Compactor merges small Parquet files per partition.
func (*Compactor) CompactTable ¶
CompactTable runs compaction for all partitions of a table. When a partition has more files than can be merged in one pass, multiple passes run back-to-back until the partition is fully compacted (no 5-minute wait between passes).
A partition whose merge fails does not stop the others: the failure is recorded in Result.Failed, the partition is skipped for the rest of the call, and the aggregate *CompactionFailed comes back at the end. See that type for why the first-failure return this replaced was too blunt.
func (*Compactor) FlushDeferredDeletes ¶
FlushDeferredDeletes physically retires every pending file older than DeleteGrace. Called from the background sweep; safe to call any time. Returns the number of files deleted.
The queue is a list of paths THIS table stopped referencing, which is not the same claim as "nothing references these bytes" — #896 is the gap between the two. Its only guard used to be the object's LastModified, and registering an unchanged object into ANOTHER live table does not move LastModified: `archive` registered one of `events`'s compacted-away sources during the grace, the queue deleted it, and `archive`'s manifest was left naming an object that no longer exists.
So eligibility is established by catalog.Catalog.RetireObjects instead: a live-manifest reference check across every table, taken under a retirement mark that refuses a racing registration, with the recreated-object check folded in. A path it cannot prove anything about goes BACK on the queue — doubt preserves bytes. A path some live table references is dropped from the queue, because that reference is not going to go away because we waited.
func (*Compactor) ForceCompactFile ¶
func (c *Compactor) ForceCompactFile(ctx context.Context, tableName string, filePath string, gcIndices map[int64]bool) error
ForceCompactFile rewrites a single data file, applying the delete markers the manifest holds for it. Used by delete-marker GC to physically purge deleted rows from files whose markers have aged out.
Safety invariants:
- Write-before-delete: the new file is written to the object store before the old file leaves the manifest. On partial failure the new file may become an orphan in S3, but data is never lost.
- ALL of the file's markers or none. The rewrite applies exactly the marker set the manifest held when it was read, and the publication (catalog.CommitCompaction, via SwapFileForGC) refuses if that set has moved since. The old contract — apply the GC-scanned indices, leave any that arrived since — was #894: a surviving marker names a row in a file that no longer exists, so no reader can apply it, the next sweep drops it as an orphan, and the replacement carries the deleted row for good. Removing a marker cannot remove a row from a file that already has it.
- One conditional publication: the old file's removal, the replacement's addition, and the marker cleanup are a single validated CAS.
- Per-file lock: prevents a double GC rewrite when two sweeps of THIS compactor overlap. It cannot exclude an independent compactor — that is what the commit-time input check is for (#895).
gcIndices is the GC scan's trigger, not the authority: it says this file has aged markers worth rewriting. What actually gets applied is the manifest's current marker set for the file, which is a superset when a DELETE landed since the scan — and applying that newer delete is the right answer, not a TOCTOU hazard.
A conflict is not an error to the caller: another writer got to this file first, or a DELETE committed while the rewrite was being written. The output is discarded and the rewrite is retried against the newer snapshot; past maxGCRewriteAttempts it is left for the next GC sweep, which re-scans from scratch. Compactor.PublicationConflicts counts those.
func (*Compactor) PublicationConflicts ¶ added in v0.18.45
PublicationConflicts reports how many compaction replacements this compactor wrote and then had REFUSED at publication because the snapshot they were cut from was no longer the table's state. See the field.
func (*Compactor) RewriteTable ¶ added in v0.18.1
RewriteTable rewrites EVERY file of every partition of a table exactly once, through the current writer, and replaces the originals.
This is the format-migration mode, and it is deliberately not compaction. shouldCompact's floors — two files, MinFiles, an average size under MaxFileSizeBytes — all answer "is this partition worth merging", which is the right question for a background sweep and the wrong one for a migration: a partition holding ONE 512 MB file is exactly the file that has to be rewritten, and it is the one shape compaction will never touch. So a rewrite is exempt from the floors and admits a 1 -> 1 pass.
It terminates structurally rather than by CompactTable's progress rule. The file list is read from the manifest ONCE, split into memory-bounded groups, and each group is written once; nothing re-reads the manifest, so no output of this call can become an input to it. "1 removed, 1 created" is progress here, which is precisely why the progress rule cannot apply.
Its use is ADR-0018's DECIMAL(p > 18) migration: files written before #429 annotate a wide DECIMAL over an INT64 leaf, and no reader outside wadjet can open them. One rewrite through the current writer produces a FLBA(16) leaf with byte-identical unscaled values. Every other type round-trips unchanged (that is the compaction gate's property), so running it over a table that needs nothing costs the rewrite and changes no value.
Like CompactTable, a partition whose merge fails does not stop the others; the aggregate is *CompactionFailed.
type Config ¶
type Config struct {
// MinFiles is the minimum file count per partition to trigger compaction.
MinFiles int
// MaxFileSizeBytes is the average size below which compaction triggers.
MaxFileSizeBytes int64
// MaxFilesPerPass caps the number of files merged in one compaction pass
// to bound memory usage.
MaxFilesPerPass int
// DeleteGrace is how long a compacted-away file stays physically present
// in the object store after its manifest entry is removed. In-flight
// tasks hold file lists resolved at DISPATCH time; deleting the bytes
// the instant the manifest swaps races every running query against the
// compactor (observed 2026-06-11: first successful mid-benchmark
// compaction at SF10 deleted chunks under three dispatched scan tasks →
// "object not found" ×5 → circuit breaker open → every later query
// failed). Mirrors DefaultGCMinAge's reasoning. Zero uses
// DefaultDeleteGrace; negative deletes immediately (tests).
DeleteGrace time.Duration
}
Config controls compaction trigger thresholds and limits.
type PartitionFailure ¶ added in v0.18.1
PartitionFailure is one partition whose merge failed, and why.
type Result ¶
type Result struct {
Table string
// PartitionsCompacted counts successful MERGES, not distinct partitions:
// a partition compacted over three passes, or rewritten as three
// memory-bounded groups, counts three times.
PartitionsCompacted int
FilesRemoved int
FilesCreated int
RowsMerged int64
BytesBefore int64
BytesAfter int64
Failed []PartitionFailure
// PublicationConflicts counts merges whose replacement could not be
// published because another writer had moved the partition's files or
// its delete markers since the manifest was read
// (catalog.ErrCompactionConflict). It is a counter on a SUCCESSFUL
// result, not a failure: nothing was written, the output object was
// deleted, and the pass loop replanned from the manifest that replaced
// the one it read.
//
// It is here because rows alone cannot tell "this compaction ran and
// changed nothing" from "this compaction was refused and something else
// did the work" — the same reason the DAG's routing counters sit beside
// the rows. A gate for #894/#895 asserts this, not just the final row
// set.
PublicationConflicts int
// PassLimitReached reports that the multi-pass loop stopped at
// maxCompactionPasses with the table still shrinking. It is a flag on a
// SUCCESSFUL result rather than an error: every one of those passes had
// to remove more files than it created to get there (the progress rule
// in CompactTable), so the work is real and committed, and calling the
// call a failure would discard a correct result and — through the
// background sweep's `continue` on error — skip the table's
// delete-marker GC as well. The next sweep picks the table up where
// this one left off.
PassLimitReached bool
}
Result summarizes one compaction pass for a table.
Failed names the partitions whose merge did not run. It is part of the result rather than only of the log because the counters below count only the partitions that DID compact, so a caller reading them alone cannot tell a clean run from one where a partition failed on every pass (#435).
func (*Result) Summary ¶ added in v0.18.45
Summary renders the lines a caller reports to an operator, in order. It is the one place that decides what a compaction run SAYS about itself, so the CLI cannot print a subset of it by omission.
It exists because PublicationConflicts was unreportable without it. A `wadjet compact --rewrite` whose only group lost a publication race returns a nil error, an empty Failed, PassLimitReached false and PartitionsCompacted zero — so the CLI printed
table events: 0 merges, 0 files removed, 0 created, 0 rows, 0 -> 0 bytes
which is character for character what an already-migrated table prints. The operator concludes the format migration is done. It is not: RewriteTable reads its file list exactly once, by construction, so a skipped group is not retried inside the call and only a re-run picks it up. That is the same reason PassLimitReached earns a line — a counter nobody prints cannot tell an operator anything, which is precisely the argument ADR-0020's amendment makes for having the counter at all.
The "run again" half is conditioned on this run having published NOTHING, rather than on the conflict count alone: CompactTable replans after a refusal, so a run that conflicted once and then compacted the partition has finished its work and must not be reported as unfinished.
Failed is deliberately not here. Those go to stderr, one per partition, and mixing streams in one list would decide that for the caller.
Source Files