Carry forward v1 inline manifests when appending

Author: platypii

src/manifest.js

@@ -1,11 +1,13 @@
+import { avroMetadata } from './avro/avro.metadata.js'
+import { avroRead } from './avro/avro.read.js'
 import { fetchAvroRecords, urlResolver } from './fetch.js'
 
 /**
  * Returns manifest entries for a snapshot. Defaults to the current snapshot;
  * pass `snapshotId` to time-travel to a prior snapshot in the metadata's
  * snapshot log.
  *
- * @import {Resolver, TableMetadata, Manifest, ManifestEntry} from '../src/types.js'
+ * @import {Resolver, TableMetadata, Manifest, ManifestEntry, Snapshot} from '../src/types.js'
  * @typedef {{ url: string, entries: ManifestEntry[] }[]} ManifestList
  * @param {object} options
  * @param {TableMetadata} options.metadata
@@ -36,15 +38,56 @@ export async function icebergManifests({ metadata, resolver, snapshotId }) {
     const manifestListUrl = snapshot['manifest-list']
     manifests = /** @type {Manifest[]} */ (await fetchAvroRecords(manifestListUrl, resolver))
   } else if (snapshot.manifests) {
-    // Use manifest URLs directly from snapshot
-    manifests = snapshot.manifests
+    // v1 snapshots list manifests inline instead of pointing at a manifest list
+    manifests = await resolveInlineManifests(snapshot, resolver)
   } else {
     throw new Error('No manifest information found in snapshot')
   }
 
   return await fetchManifests(manifests, resolver)
 }
 
+/**
+ * Turn a v1 snapshot's inline `manifests` (a list of manifest file locations)
+ * into manifest list records. Each manifest is read once to recover the
+ * length, spec id, and file counts a manifest list needs, so a table upgraded
+ * from v1 can still be read and appended to without rewriting its inherited
+ * metadata.
+ *
+ * @param {Snapshot} snapshot
+ * @param {Resolver} resolver
+ * @returns {Promise<Manifest[]>}
+ */
+export async function resolveInlineManifests(snapshot, resolver) {
+  const inline = snapshot.manifests ?? []
+  return await Promise.all(inline.map(async manifestPath => {
+    const ab = await resolver.reader(manifestPath)
+    const buffer = await ab.slice(0, ab.byteLength)
+    const reader = { view: new DataView(buffer), offset: 0 }
+    const { metadata, syncMarker } = await avroMetadata(reader)
+    const entries = /** @type {ManifestEntry[]} */ (await avroRead({ reader, metadata, syncMarker }))
+    const counts = [0, 0, 0]
+    const rows = [0n, 0n, 0n]
+    for (const entry of entries) {
+      counts[entry.status]++
+      rows[entry.status] += BigInt(entry.data_file.record_count)
+    }
+    return {
+      manifest_path: manifestPath,
+      manifest_length: BigInt(ab.byteLength),
+      partition_spec_id: Number(metadata['partition-spec-id'] ?? 0),
+      content: 0,
+      added_snapshot_id: BigInt(snapshot['snapshot-id']),
+      added_files_count: counts[1],
+      existing_files_count: counts[0],
+      deleted_files_count: counts[2],
+      added_rows_count: rows[1],
+      existing_rows_count: rows[0],
+      deleted_rows_count: rows[2],
+    }
+  }))
+}
+
 /**
  * Fetch manifest entries from a list of manifests in parallel.
  *

src/types.d.ts

@@ -201,7 +201,7 @@ export interface Snapshot {
   'sequence-number': number
   'timestamp-ms': number
   'manifest-list': string
-  manifests?: Manifest[]
+  manifests?: string[] // v1 only: manifest file locations
   summary: {
     // spec: "value of these fields should be of string type"
     operation: string

src/write/snapshot.js

@@ -1,4 +1,5 @@
 import { fetchAvroRecords } from '../fetch.js'
+import { resolveInlineManifests } from '../manifest.js'
 import { writeManifestList } from './manifest-list.js'
 import { computeFieldSummary } from './stats.js'
 import { transformResultType } from './transform.js'
@@ -40,8 +41,13 @@ export function currentSnapshot(metadata) {
  */
 export async function loadPriorManifests(metadata, resolver) {
   const snap = currentSnapshot(metadata)
-  if (!snap?.['manifest-list']) return []
-  return /** @type {Manifest[]} */ (await fetchAvroRecords(snap['manifest-list'], resolver))
+  if (!snap) return []
+  if (snap['manifest-list']) {
+    return /** @type {Manifest[]} */ (await fetchAvroRecords(snap['manifest-list'], resolver))
+  }
+  // A table upgraded from v1 keeps snapshots that list manifests inline. Carry
+  // them forward, or the next snapshot silently drops all existing data.
+  return await resolveInlineManifests(snap, resolver)
 }
 
 /**

test/manifest.test.js

@@ -71,29 +71,22 @@ describe('Iceberg Manifests', () => {
         'timestamp-ms': 0,
         'manifest-list': '',
         summary: { operation: 'append' },
-        manifests: [{
-          manifest_path: manifestPath,
-          manifest_length: BigInt(manifestLength),
-          partition_spec_id: 0,
-          content: 1,
-          added_snapshot_id: 1n,
-          added_files_count: 1,
-          existing_files_count: 0,
-          deleted_files_count: 0,
-          added_rows_count: 1n,
-          existing_rows_count: 0n,
-          deleted_rows_count: 0n,
-        }],
+        // v1 shape: manifest file locations listed inline on the snapshot
+        manifests: [manifestPath],
       }],
     }
 
     const manifests = await icebergManifests({ metadata, resolver: countingResolver })
 
     expect(manifests).toHaveLength(1)
-    expect(calls).toEqual([{ url: manifestPath, byteLength: manifestLength }])
+    // first read recovers the manifest length, second read passes it through
+    expect(calls).toEqual([
+      { url: manifestPath, byteLength: undefined },
+      { url: manifestPath, byteLength: manifestLength },
+    ])
   })
 
-  it('inherits a null entry snapshot id from the manifest list', async () => {
+  it('inherits a null entry snapshot id from a v1 snapshot', async () => {
     const { resolver: memory } = memResolver()
     const manifestPath = 'http://test/inherited-snapshot-id.avro'
     const writer = memory.writer?.(manifestPath)
@@ -117,14 +110,7 @@ describe('Iceberg Manifests', () => {
       'current-snapshot-id': 77,
       snapshots: [{
         'snapshot-id': 77,
-        manifests: [{
-          manifest_path: manifestPath,
-          manifest_length: BigInt(writer.offset),
-          partition_spec_id: 0,
-          content: 0,
-          sequence_number: 5n,
-          added_snapshot_id: 77n,
-        }],
+        manifests: [manifestPath],
       }],
     })
 

test/write/stage.test.js

@@ -77,6 +77,34 @@ describe('icebergStageAppend', () => {
     expect(read).toEqual(records)
   })
 
+  it('carries forward v1 manifest locations when appending', async () => {
+    vi.spyOn(Date, 'now').mockReturnValue(1700000000000)
+    const tableUrl = 'http://test/stage-v1-manifest-paths'
+    const { resolver } = memResolver()
+
+    const created = await icebergCreate({ tableUrl, resolver, schema })
+    const first = [{ id: 1n, name: 'alice' }]
+    const staged1 = await icebergStageAppend({ tableUrl, metadata: created, records: first, resolver })
+    const committed1 = await fileCatalogCommit({ tableUrl, metadata: created, staged: staged1, resolver })
+
+    // The spec's v1 shape: `manifests` is a list of manifest file locations.
+    const snap = committed1.snapshots?.find(s => s['snapshot-id'] === committed1['current-snapshot-id'])
+    if (!snap) throw new Error('expected current snapshot')
+    const manifests = /** @type {any[]} */ (await fetchAvroRecords(snap['manifest-list'], resolver))
+    const inline = /** @type {any} */ ({ ...snap, manifests: manifests.map(m => m.manifest_path) })
+    delete inline['manifest-list']
+    /** @type {TableMetadata} */
+    const inherited = { ...committed1, snapshots: [inline] }
+    expect(await icebergRead({ tableUrl, metadata: inherited, resolver })).toEqual(first)
+
+    const second = [{ id: 2n, name: 'bob' }]
+    const staged2 = await icebergStageAppend({ tableUrl, metadata: inherited, records: second, resolver })
+    const committed2 = await fileCatalogCommit({ tableUrl, metadata: inherited, staged: staged2, resolver })
+
+    const read = await icebergRead({ tableUrl, metadata: committed2, resolver })
+    expect(read).toEqual([...first, ...second])
+  })
+
   it('assigns row lineage for v3 appends', async () => {
     vi.spyOn(Date, 'now').mockReturnValue(1700000000000)
     const tableUrl = 'http://test/stage-v3'