diff --git a/llp/0199-maintenance-compaction-convergence.decision.md b/llp/0199-maintenance-compaction-convergence.decision.md index c5931349..d23d855c 100644 --- a/llp/0199-maintenance-compaction-convergence.decision.md +++ b/llp/0199-maintenance-compaction-convergence.decision.md @@ -6,7 +6,7 @@ **Author:** Kenny / Claude **Date:** 2026-08-07 **Related:** LLP 0027 -**Extended-by:** LLP 0207 (a baseline mismatch whose current snapshot is a sorted `replace` is a foreign rewrite, not growth: recognize it and re-baseline instead of compacting), LLP 0209 (the rewrite the baseline gate protects now sizes its output files by bytes written rather than by the in-memory batch estimate, so a compacted generation actually reaches `target_file_bytes`) +**Extended-by:** [LLP 0207](./0207-foreign-sorted-replace-convergence.decision.md) (a baseline mismatch whose current snapshot is a sorted `replace` is a foreign rewrite, not growth: recognize it and re-baseline instead of compacting); [LLP 0209](./0209-compaction-file-size.decision.md) (the rewrite the baseline gate protects now sizes its output files by bytes written rather than by the in-memory batch estimate, so a compacted generation actually reaches `target_file_bytes`) > Maintenance stops re-flagging already-compacted partitions: a partition is > only compaction-due when its live data-file count has moved off the count diff --git a/src/core/cache/maintenance.js b/src/core/cache/maintenance.js index 840c5d0b..d788c72f 100644 --- a/src/core/cache/maintenance.js +++ b/src/core/cache/maintenance.js @@ -10,7 +10,7 @@ import { loadLatestFileCatalogMetadata, } from 'icebird' -import { Attr, getMeter, withSpan } from '../observability/index.js' +import { Attr, getActiveSpan, getMeter, withSpan } from '../observability/index.js' import { inferColumnType } from './migrate.js' import { discoverCachePartitions, readCursorSync, tryReadCursorSync, writeCursor } from './partition.js' import { datasetsRoot } from './paths.js' @@ -119,6 +119,7 @@ export async function maintainCache(opts) { const reports = [] let totalSnapshotsExpired = 0 let totalCompacted = 0 + let totalRebaselined = 0 for (const part of partitions) { // Always work one partition before the budget can cut the tick short: @@ -173,6 +174,7 @@ export async function maintainCache(opts) { reports.push(report) totalSnapshotsExpired += report.snapshotsExpired if (report.compacted) totalCompacted++ + if (report.rebaselined) totalRebaselined++ } if (!opts.dryRun) { @@ -183,6 +185,7 @@ export async function maintainCache(opts) { partitions: reports, totalSnapshotsExpired, totalCompacted, + totalRebaselined, dryRun: opts.dryRun ?? false, elapsedMs: Date.now() - startMs, } @@ -291,6 +294,13 @@ async function maintainGeneration(r, cursor, cfg, opts, settle, snapshotsExpired // compact_avg_file_bytes), and the tick budget is burned rewriting the // same partitions while the rest of the walk starves. const grewSinceCompaction = dataFilesBefore !== resettleBaselineFiles(cursor) + // Cheap dueness check first: file-count and byte-size heuristics only, + // no metadata load and no row scan. A foreign sorted replace almost + // always lands here (its baseline mismatch alone doesn't imply the + // size heuristics fire), so gating the expensive re-settle scan behind + // this check means the common "recognized, nothing to scan for" tick + // never pays for one. + const compactionDue = opts.force || (grewSinceCompaction && needsCompaction(liveDir, cfg)) // @ref LLP 0027#re-settle-sweep: a partition holding a committed // fallback row may carry a split twin pair the flush-time settle // never collapsed; force a rewrite so the sweep can re-settle it even @@ -299,10 +309,15 @@ async function maintainGeneration(r, cursor, cfg, opts, settle, snapshotsExpired // line never lands (harness aux, wire-only reminders) - from forcing a // full rewrite every tick, and skips the attributes scan entirely when // nothing new has flushed. - const hasResettle = settle + // @ref LLP 0207#outranks-resettle [constrained-by]: when the cheap check + // above already made compaction due, the scan's answer can never + // change the outcome (recognition, tested below, still outranks it), + // so skip it: only run the scan when it might be the sole reason to + // compact. + const hasResettle = !compactionDue && settle ? grewSinceCompaction && await hasResettleCandidate(liveDir) : false - const shouldCompact = opts.force || hasResettle || (grewSinceCompaction && needsCompaction(liveDir, cfg)) + const shouldCompact = compactionDue || hasResettle if (shouldCompact) { const tableInfo = await loadCompactionTableInfo(liveDir) // @ref LLP 0207#foreign-replace [implements]: a baseline mismatch whose @@ -315,6 +330,11 @@ async function maintainGeneration(r, cursor, cfg, opts, settle, snapshotsExpired // the sorted layout every night. An explicit --force still rewrites. if (!opts.force && foreignSortedReplace(tableInfo)) { r.rebaselined = true + // The counter proves a rebaseline happened at all, but it carries + // only the dataset; tagging the enclosing maintenance.partition span + // names the partition, so a trace query finds which day re-baselined + // without cross-referencing the counter. + getActiveSpan()?.setAttribute('rebaselined', true) if (!opts.dryRun) { await writeCursor(r.path, rebaselineCursor(cursor, dataFilesBefore)) rebaselinesCounter.add(1, { [Attr.DATASET]: r.dataset }) diff --git a/src/core/cache/types.d.ts b/src/core/cache/types.d.ts index 8db78463..1e3085c8 100644 --- a/src/core/cache/types.d.ts +++ b/src/core/cache/types.d.ts @@ -301,6 +301,7 @@ export interface MaintenanceReport { partitions: MaintenancePartitionReport[] totalSnapshotsExpired: number totalCompacted: number + totalRebaselined: number dryRun: boolean elapsedMs: number } diff --git a/src/core/commands/query.js b/src/core/commands/query.js index 1cfac29d..b22afa21 100644 --- a/src/core/commands/query.js +++ b/src/core/commands/query.js @@ -327,8 +327,7 @@ export async function runQueryMaintain(argv, ctx) { ctx.stdout.write(` ${label}: ${actions.join(', ')}\n`) } } - const rebaselined = report.partitions.filter((p) => p.rebaselined).length - const rebaselineNote = rebaselined > 0 ? `, ${rebaselined} rebaselined` : '' + const rebaselineNote = report.totalRebaselined > 0 ? `, ${report.totalRebaselined} rebaselined` : '' ctx.stdout.write(`maintenance: ${report.totalSnapshotsExpired} snapshots expired, ${report.totalCompacted} partitions compacted${rebaselineNote} (${report.elapsedMs}ms)\n`) return 0 } diff --git a/test/core/cache-retention-maintenance.test.js b/test/core/cache-retention-maintenance.test.js index be8418de..f0a4f6e7 100644 --- a/test/core/cache-retention-maintenance.test.js +++ b/test/core/cache-retention-maintenance.test.js @@ -3,6 +3,7 @@ import test from 'node:test' import assert from 'node:assert/strict' import fs from 'node:fs/promises' +import fsSync from 'node:fs' import os from 'node:os' import path from 'node:path' @@ -11,12 +12,14 @@ import { maintainCache, cacheStatus, normalizeMaintenanceConfig } from '../../sr import { appendRowsToSourceTable, readCursorSync, writeCursor } from '../../src/core/cache/partition.js' import { appendRowsToTable, currentPartitionSpec, currentSchema, readRowsFromTable, sortColumnsFromMetadata, tableExists } from '../../src/core/cache/iceberg/store.js' import { createLocalIcebergIO, tableUrlForDir } from '../../src/core/cache/iceberg/resolver.js' +import { TracerProvider } from '../../src/core/observability/runtime.js' import { fileCatalog, icebergRewrite, loadLatestFileCatalogMetadata } from 'icebird' import { parquetMetadata } from 'hyparquet' /** * @import { ColumnSpec } from '../../hypaware-plugin-kernel-types.js' * @import { CachePartitioningDeclaration } from '../../src/core/cache/types.js' + * @import { Span } from '../../src/core/observability/runtime.js' */ /** @@ -885,6 +888,7 @@ test('a foreign sorted replace re-baselines the cursor instead of being rewritte // A dry run predicts the recognition without writing anything. const preview = await maintainCache({ cacheRoot, compactOnly: true, dryRun: true }) assert.equal(preview.totalCompacted, 0) + assert.equal(preview.totalRebaselined, 1, 'the report-level rebaseline count mirrors totalCompacted') assert.equal(preview.partitions[0].rebaselined, true) assert.equal( /** @type {{ resettleBaselineFiles: number }} */ (readCursorSync(partDir).compaction).resettleBaselineFiles, @@ -894,6 +898,7 @@ test('a foreign sorted replace re-baselines the cursor instead of being rewritte const first = await maintainCache({ cacheRoot, compactOnly: true }) assert.equal(first.totalCompacted, 0) + assert.equal(first.totalRebaselined, 1, 're-baselining one partition must be reflected in the report total') assert.equal(first.partitions[0].rebaselined, true) const cursor = readCursorSync(partDir) assert.equal(cursor.epoch, 0, 'no rewrite: the generation must not advance') @@ -905,6 +910,7 @@ test('a foreign sorted replace re-baselines the cursor instead of being rewritte // Converged: the baseline gate now blocks before any metadata load. const second = await maintainCache({ cacheRoot, compactOnly: true }) assert.equal(second.totalCompacted, 0) + assert.equal(second.totalRebaselined, 0, 'converged: no rebaseline happened this tick') assert.notEqual(second.partitions[0].rebaselined, true) // A late append flips the current snapshot off `replace` and moves the @@ -1030,6 +1036,102 @@ test('force still rewrites a foreign sorted replace', async () => { } }) +test('a foreign sorted replace tags the maintenance.partition span with rebaselined', async () => { + const cacheRoot = await makeTmpDir('maint-foreign-span') + /** @type {Span[]} */ + const captured = [] + const provider = new TracerProvider({ + resource: { attributes: {} }, + exporters: [{ exportBatch(/** @type {Span[]} */ spans) { captured.push(...spans) } }], + }) + provider.register() + try { + const partDir = path.join(cacheRoot, 'datasets', 'ds1', 'date=2026-08-08') + const epoch0 = path.join(partDir, 'epoch=0') + for (let i = 0; i < 3; i++) { + await appendRowsToTable(epoch0, COLUMNS, [ + { id: i, value: `v${i}`, timestamp: new Date().toISOString() }, + ], { sortOrder: [{ column: 'id', direction: 'asc' }] }) + } + await commitForeignReplace(epoch0) + await writeCursor(partDir, { + epoch: 0, + rowCount: 3, + layout: 'epoch', + compaction: { compactedAt: '2026-08-08T00:00:00.000Z', resettleBaselineFiles: 99 }, + }) + + const report = await maintainCache({ cacheRoot, compactOnly: true }) + assert.equal(report.partitions[0].rebaselined, true, 'sanity: this tick recognized the foreign replace') + + const partitionSpan = captured.find((span) => span.name === 'maintenance.partition') + assert.ok(partitionSpan, 'maintenance.partition span must be exported') + assert.equal( + partitionSpan?.attributes.rebaselined, + true, + 'the span, not just the hyp_rebaselines counter, must name which partition re-baselined' + ) + } finally { + await provider.shutdown() + await fs.rm(cacheRoot, { recursive: true, force: true }) + } +}) + +test('a partition already due for compaction skips the resettle-candidate row scan', async (t) => { + // @ref LLP 0207#outranks-resettle [tests]: once the cheap file-count/size + // check alone makes compaction due, the resettle scan's answer cannot + // change `shouldCompact`, so it must not run at all. `hasResettleCandidate` + // is module-private and its `scanRowsFromTable` is an unpatchable ESM named + // import, so there is no direct call-count hook to assert against; instead + // this mocks `readFileSync` and captures a stack trace per `.parquet` read, + // then asserts none of those stacks pass through `hasResettleCandidate`. + // That frame name survives the async boundary between the scan and the + // read, so it attributes each read to its caller instead of only counting + // reads tick-wide (a second legitimate read elsewhere in the tick, e.g. a + // footer-stats probe, would not falsely implicate the scan). + const cacheRoot = await makeTmpDir('maint-scan-skip') + try { + const partDir = path.join(cacheRoot, 'datasets', 'ds1', 'date=2026-08-08') + const epoch0 = path.join(partDir, 'epoch=0') + await appendRowsToTable(epoch0, COLUMNS, [ + { id: 1, value: 'v1', timestamp: new Date().toISOString() }, + ]) + // Never compacted: `grewSinceCompaction` is true unconditionally, so the + // only thing standing between the old code and a resettle scan is the + // `compactionDue` gate under test. + await writeCursor(partDir, { epoch: 0, rowCount: 1, compaction: null, layout: 'epoch' }) + + /** @type {string[]} */ + const stacks = [] + const original = fsSync.readFileSync + t.mock.method(fsSync, 'readFileSync', function (p, ...rest) { + if (String(p).endsWith('.parquet')) stacks.push(new Error().stack ?? '') + return original.call(this, p, ...rest) + }) + + const report = await maintainCache({ + cacheRoot, + compactOnly: true, + // compact_file_count: 0 makes `needsCompaction` (and so + // `compactionDue`) true on file count alone, with no size heuristic + // involved: dueness is settled before the resettle scan would run. + config: { compact_file_count: 0 }, + storage: /** @type {any} */ ({}), + getSettleHook: () => async (rows) => rows, + }) + assert.equal(report.totalCompacted, 1, 'sanity: compaction actually ran') + + assert.ok(stacks.length > 0, 'sanity: the data file was read at all') + assert.deepEqual( + stacks.filter((s) => s.includes('hasResettleCandidate')), + [], + 'the resettle scan must not read the data file' + ) + } finally { + await fs.rm(cacheRoot, { recursive: true, force: true }) + } +}) + test('maintenance walks partitions neediest-first, not directory order', async () => { const cacheRoot = await makeTmpDir('maint-order') try {