diff --git a/llp/0199-maintenance-compaction-convergence.decision.md b/llp/0199-maintenance-compaction-convergence.decision.md index a0232ef9..606b4240 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 +**Extended-by:** [LLP 0207](./0207-foreign-sorted-replace-convergence.decision.md) (a second convergence source: a foreign sorted `replace` re-baselines the gate instead of triggering a rewrite) > 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 f1d97137..7080c0fe 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' @@ -117,6 +117,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: @@ -157,6 +158,7 @@ export async function maintainCache(opts) { reports.push(report) totalSnapshotsExpired += report.snapshotsExpired if (report.compacted) totalCompacted++ + if (report.rebaselined) totalRebaselined++ } if (!opts.dryRun) { @@ -167,6 +169,7 @@ export async function maintainCache(opts) { partitions: reports, totalSnapshotsExpired, totalCompacted, + totalRebaselined, dryRun: opts.dryRun ?? false, elapsedMs: Date.now() - startMs, } @@ -275,6 +278,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 @@ -283,10 +293,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 @@ -299,6 +314,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 45084d72..00d529d6 100644 --- a/src/core/cache/types.d.ts +++ b/src/core/cache/types.d.ts @@ -232,6 +232,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 c8947519..9b45b1a0 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,11 +12,13 @@ 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 { ColumnSpec } from '../../hypaware-plugin-kernel-types.js' * @import { CachePartitioningDeclaration } from '../../src/core/cache/types.js' + * @import { Span } from '../../src/core/observability/runtime.js' */ /** @param {string} prefix */ @@ -853,6 +856,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, @@ -862,6 +866,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') @@ -873,6 +878,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 @@ -998,6 +1004,128 @@ 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). + // + // A stack-based observation can go blind: the deciding frame sits ~8 frames + // below the mock, and V8's default `Error.stackTraceLimit` of 10 leaves only + // two frames of headroom. Three more frames anywhere between the mock and + // the caller (an icebird refactor, extra node:test mock internals, a wrapper + // in `resolver.js`) would drop it, and a negative "no stack mentions + // `hasResettleCandidate`" assertion passes vacuously on truncated stacks. + // Two guards keep that from happening silently: + // 1. raise `Error.stackTraceLimit` while the mock is installed (restored + // below even if the test throws), which removes the hazard outright; + // 2. assert positively that some stack names `compactGeneration`, the + // legitimate reader. It calls `scanRowsFromTable` from exactly the same + // depth as `hasResettleCandidate` does, so any truncation deep enough + // to hide the frame the negative assertion hunts for also hides this + // one, and the test fails loudly instead of going quiet. Asserting on + // `scanRowsFromTable` itself would not do: it sits one frame shallower + // and survives truncation that has already blinded the real check. + // Guard 2 also catches the `?? ''` fallback below storing an unattributable + // empty string. + const cacheRoot = await makeTmpDir('maint-scan-skip') + const originalStackTraceLimit = Error.stackTraceLimit + 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 + Error.stackTraceLimit = 50 + 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.ok( + stacks.some((s) => s.includes('compactGeneration')), + 'sanity: captured stacks must be deep enough to name the reader, or the assertion below passes vacuously' + ) + assert.deepEqual( + stacks.filter((s) => s.includes('hasResettleCandidate')), + [], + 'the resettle scan must not read the data file' + ) + } finally { + Error.stackTraceLimit = originalStackTraceLimit + 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 {