Reuse manifests within each data source

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<Manifest, Promise<ManifestEntry[]>>} */
+  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,

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<typeof fileCatalog>, 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<Record<string, import('squirreling').SqlPrimitive>[]>}
+     */
+    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<Record<string, import('squirreling').SqlPrimitive>[]>} */
+    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 })

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 })