From 7b32699cc9c485e6f8f4c61f0a2ab4be7b515269 Mon Sep 17 00:00:00 2001 From: Kenny Daniel Date: Wed, 7 Oct 2026 14:16:14 -0700 Subject: [PATCH] Reuse manifests within each data source --- src/sql/icebergDataSource.js | 24 +++++++-- test/sql/manifestPruning.test.js | 85 ++++++++++++++++++++++++++++++++ test/write/merge.test.js | 71 +++++++++++++++++++++++++- 3 files changed, 175 insertions(+), 5 deletions(-) diff --git a/src/sql/icebergDataSource.js b/src/sql/icebergDataSource.js index 8b2c816..4768690 100644 --- a/src/sql/icebergDataSource.js +++ b/src/sql/icebergDataSource.js @@ -35,9 +35,10 @@ import { pruneTopKFiles, scanTopKFiles } from './topK.js' * rows for compatibility. * * Metadata, the manifest list, delete manifests, schema, and delete maps are - * resolved once at construction. Each scan fetches the data manifests it - * needs and walks their data files in manifest order, yielding rows on - * demand. Pushdowns: + * resolved once at construction. Data manifests are fetched lazily and shared + * by scans of this source's fixed snapshot, including concurrent scans. They + * are retained only for this source's lifetime; failed reads can be retried. + * Each scan walks its selected data files in manifest order. Pushdowns: * - WHERE skips whole data manifests before they are fetched, using the * manifest list's per-partition-field summaries (Java `ManifestEvaluator`). * This pays off on tables whose manifests are clustered by partition (see @@ -118,6 +119,11 @@ export async function icebergDataSource({ tableUrl, metadataFileName, metadata, // Pre-fetch delete maps once; reused by every scan. const deleteMapsPromise = fetchDeleteMaps(deleteEntries, fetchResolver) + // The source pins its manifest list at construction. Retain decoded entries + // alongside it, with no sharing or invalidation across source instances. + /** @type {Map>} */ + const manifestEntries = new Map() + /** * Live data entries of the manifests that might match `filter`, in * manifest-list order. @@ -129,7 +135,17 @@ export async function icebergDataSource({ tableUrl, metadataFileName, metadata, const selected = filter ? dataManifests.filter(m => manifestMightMatch(filter, m, scanSchema, tableMetadata)) : dataManifests - return liveEntries(await Promise.all(selected.map(m => fetchManifestEntries(m, fetchResolver)))) + return liveEntries(await Promise.all(selected.map(m => { + let entries = manifestEntries.get(m) + if (!entries) { + entries = fetchManifestEntries(m, fetchResolver).catch(err => { + manifestEntries.delete(m) + throw err + }) + manifestEntries.set(m, entries) + } + return entries + }))) } /** * Live rows the data files of `selected` manifests hold, before deletes, diff --git a/test/sql/manifestPruning.test.js b/test/sql/manifestPruning.test.js index 633c9f5..63389a5 100644 --- a/test/sql/manifestPruning.test.js +++ b/test/sql/manifestPruning.test.js @@ -3,6 +3,7 @@ import { describe, expect, it, vi } from 'vitest' import { fileCatalog } from '../../src/catalog/file.js' import { icebergManifests } from '../../src/manifest.js' import { manifestMightMatch } from '../../src/prune.js' +import { icebergDataSource } from '../../src/sql/icebergDataSource.js' import { icebergQuery } from '../../src/sql/icebergQuery.js' import { serializeValue } from '../../src/write/serde.js' import { icebergAppend, icebergCreateTable, icebergRewriteManifests, icebergUpdateSchema } from '../../src/write/write.js' @@ -85,7 +86,91 @@ async function query(resolver, tableUrl, where) { return rows.map(r => /** @type {bigint} */ (r.id)).sort((a, b) => Number(a) - Number(b)) } +/** + * Two independently prunable manifests for source-lifetime tests. + * + * @returns {Promise<{ resolver: Resolver, catalog: ReturnType, tableUrl: string }>} + */ +async function smallPartitionedTable() { + const { resolver, lister } = memResolver() + const catalog = fileCatalog({ resolver, lister, conditionalCommits: true }) + const tableUrl = 'http://test/source-manifests' + await icebergCreateTable({ + catalog, tableUrl, schema, + partitionSpec: { + 'spec-id': 0, + fields: [{ 'source-id': 3, 'field-id': 1000, name: 'kind', transform: 'identity' }], + }, + }) + for (let i = 0; i < 2; i++) { + await icebergAppend({ catalog, tableUrl, records: [{ id: BigInt(i), kind: `k${i}` }] }) + } + return { resolver, catalog, tableUrl } +} + describe('manifest list pruning', () => { + it('loads manifests lazily and shares them across concurrent and repeated scans', async () => { + const { resolver, tableUrl } = await smallPartitionedTable() + const counting = countingResolver(resolver) + const source = await icebergDataSource({ tableUrl, resolver: counting.resolver }) + expect(counting.manifestsRead()).toBe(0) + /** + * @param {string} where + * @returns {Promise[]>} + */ + async function scan(where) { + return collect(await icebergQuery({ query: `SELECT id FROM t ${where} ORDER BY id`, tables: { t: source } })) + } + const samePartition = await Promise.all([scan('WHERE kind = \'k0\''), scan('WHERE kind = \'k0\'')]) + expect(samePartition).toEqual([[{ id: 0n }], [{ id: 0n }]]) + expect(counting.manifestsRead()).toBe(1) + expect(await scan('WHERE kind = \'k0\'')).toEqual([{ id: 0n }]) + expect(counting.manifestsRead()).toBe(1) + expect(await scan('')).toEqual([{ id: 0n }, { id: 1n }]) + expect(counting.manifestsRead()).toBe(2) + expect(await scan('WHERE kind = \'k1\'')).toEqual([{ id: 1n }]) + expect(counting.manifestsRead()).toBe(2) + }) + + it('keeps retained manifests private to each source and its snapshot', async () => { + const { resolver, catalog, tableUrl } = await smallPartitionedTable() + const counting = countingResolver(resolver) + const old = await icebergDataSource({ tableUrl, resolver: counting.resolver }) + const sql = 'SELECT id FROM t ORDER BY id' + expect(await collect(await icebergQuery({ query: sql, tables: { t: old } }))).toEqual([{ id: 0n }, { id: 1n }]) + expect(counting.manifestsRead()).toBe(2) + await icebergAppend({ catalog, tableUrl, records: [{ id: 2n, kind: 'k2' }] }) + const fresh = await icebergDataSource({ tableUrl, resolver: counting.resolver }) + expect(await collect(await icebergQuery({ query: sql, tables: { t: fresh } }))).toEqual([{ id: 0n }, { id: 1n }, { id: 2n }]) + expect(counting.manifestsRead()).toBe(5) + expect(await collect(await icebergQuery({ query: sql, tables: { t: old } }))).toEqual([{ id: 0n }, { id: 1n }]) + expect(counting.manifestsRead()).toBe(5) + }) + + it('retries a failed manifest read on the same source', async () => { + const { resolver, tableUrl } = await smallPartitionedTable() + let attempts = 0 + /** @type {Resolver} */ + const flaky = { + reader(url, length) { + if (url.endsWith('.avro') && !url.includes('/snap-') && ++attempts === 1) { + throw new Error('temporary manifest failure') + } + return resolver.reader(url, length) + }, + } + const source = await icebergDataSource({ tableUrl, resolver: flaky }) + /** @returns {Promise[]>} */ + async function scan() { + return collect(await icebergQuery({ query: 'SELECT id FROM t WHERE kind = \'k0\'', tables: { t: source } })) + } + await expect(scan()).rejects.toThrow('temporary manifest failure') + expect(await scan()).toEqual([{ id: 0n }]) + expect(attempts).toBe(2) + expect(await scan()).toEqual([{ id: 0n }]) + expect(attempts).toBe(2) + }) + it('keeps identity dates before a timestamp later on the same day', async () => { const { resolver, lister } = memResolver() const catalog = fileCatalog({ resolver, lister, conditionalCommits: true }) diff --git a/test/write/merge.test.js b/test/write/merge.test.js index 6c20a45..6ec23c6 100644 --- a/test/write/merge.test.js +++ b/test/write/merge.test.js @@ -11,7 +11,10 @@ import { loadLatestFileCatalogMetadata } from '../../src/metadata.js' import { writeCarriedManifest } from '../../src/write/manifest.js' import { writeManifestList } from '../../src/write/manifest-list.js' import { manifestMergeConfig, mergeManifests, packManifests } from '../../src/write/merge.js' -import { icebergAppend, icebergCreateTable, icebergDelete, icebergRewrite, icebergUpdateSchema } from '../../src/write/write.js' +import { writeParquet } from '../../src/write/parquet.js' +import { writeManifestFile } from '../../src/write/rewrite-manifests.js' +import { buildSnapshotUpdate } from '../../src/write/snapshot.js' +import { icebergAppend, icebergCreateTable, icebergDelete, icebergRewrite, icebergRewriteManifests, icebergUpdateSchema } from '../../src/write/write.js' import { memResolver } from '../helpers.js' /** @@ -134,6 +137,72 @@ describe('packManifests', () => { }) describe('merge on commit', () => { + it.each([false, true])('preserves equality deletes and newer matching rows (partitioned=%s)', async partitioned => { + const { resolver, lister } = memResolver() + const catalog = fileCatalog({ resolver, lister, conditionalCommits: true }) + const tableUrl = 'http://test/merge-equality' + /** @type {PartitionSpec} */ + const spec = { + 'spec-id': 0, + fields: partitioned ? [{ 'source-id': 2, 'field-id': 1000, name: 'name', transform: 'identity' }] : [], + } + await icebergCreateTable({ + catalog, tableUrl, schema, partitionSpec: spec, + properties: { 'commit.manifest.min-count-to-merge': '2' }, + }) + let metadata = await icebergAppend({ catalog, tableUrl, records: [ + { id: 1n, name: 'a' }, { id: 2n, name: 'a' }, { id: 1n, name: 'b' }, + ] }) + const initial = metadata + // Two delete commits force the delete manifests to merge, too. A new + // matching row appended after each delete must remain visible. + for (const id of [1n, 2n]) { + const sequenceNumber = BigInt(metadata['last-sequence-number']) + 1n + const snapshotId = BigInt(metadata['current-snapshot-id'] ?? 0) + 1n + if (!resolver.writer) throw new Error('writer required') + const filePath = `${tableUrl}/data/eq-${id}.parquet` + const writer = resolver.writer(filePath) + await writeParquet({ writer, schema: { ...schema, fields: [schema.fields[0]] }, records: [{ id }] }) + const manifestPath = `${tableUrl}/metadata/eq-${id}.avro` + const manifest = await writeManifestFile({ + resolver, manifestPath, schema, spec, content: 1, snapshotId, sequenceNumber, formatVersion: 2, + entries: [{ + status: 1, snapshot_id: snapshotId, sequence_number: sequenceNumber, file_sequence_number: sequenceNumber, + data_file: { + content: 2, file_path: filePath, file_format: 'parquet', + partition: partitioned ? { name: 'a' } : {}, equality_ids: [1], + record_count: 1n, file_size_in_bytes: BigInt(writer.offset), + }, + }], + }) + const staged = await buildSnapshotUpdate({ + tableUrl, metadata, resolver, snapshotId, sequenceNumber, manifestUuid: `eq-${id}`, + timestampMs: Date.now(), formatVersion: 2, newManifests: [manifest], + summary: { operation: 'delete' }, writtenFiles: [filePath, manifestPath], + }) + await fileCatalogCommit({ tableUrl, metadata, staged, resolver, conditionalCommits: true }) + metadata = await icebergAppend({ catalog, tableUrl, records: [{ id, name: 'a' }] }) + } + const list = await manifestList(metadata, resolver) + expect(list.filter(m => m.content === 1)).toHaveLength(1) + expect(list.filter(m => m.content === 0)).toHaveLength(1) + const expected = partitioned + ? [{ id: 1n, name: 'b' }, { id: 1n, name: 'a' }, { id: 2n, name: 'a' }] + : [{ id: 1n, name: 'a' }, { id: 2n, name: 'a' }] + expect(await icebergRead({ tableUrl, metadata, resolver })).toEqual(expected) + // A rename verifies that equality IDs survive independently of names. + await icebergUpdateSchema({ catalog, tableUrl, schema: { + ...schema, fields: [{ ...schema.fields[0], name: 'renamed_id' }, schema.fields[1]], + } }) + const rewritten = await icebergRewriteManifests({ catalog, tableUrl, targetSizeBytes: 1 }) + const actual = await icebergRead({ tableUrl, metadata: rewritten, resolver }) + expect(actual).toHaveLength(expected.length) + expect(actual).toEqual(expect.arrayContaining(expected.map(({ id, name }) => ({ renamed_id: id, name })))) + expect(await icebergRead({ tableUrl, metadata: initial, resolver })).toEqual([ + { id: 1n, name: 'a' }, { id: 2n, name: 'a' }, { id: 1n, name: 'b' }, + ]) + }) + it.each([-1, 0, 20000])('preserves deletes when only data day partitions are normalized (%s)', async day => { const { resolver, lister } = memResolver() const catalog = fileCatalog({ resolver, lister, conditionalCommits: true })