|
1 | 1 | import { describe, expect, it, vi } from 'vitest' |
2 | 2 | import { fileCatalog } from '../../src/catalog/file.js' |
| 3 | +import { fetchAvroRecords } from '../../src/fetch.js' |
3 | 4 | import { fileCatalogCommit } from '../../src/write/commit.js' |
4 | 5 | import { icebergCreate } from '../../src/create.js' |
5 | 6 | import { icebergManifests, splitManifestEntries } from '../../src/manifest.js' |
6 | 7 | import { icebergRead } from '../../src/read.js' |
7 | 8 | import { icebergAppend, icebergRewrite } from '../../src/write/write.js' |
8 | 9 | import { icebergStageAppend } from '../../src/write/stage.js' |
| 10 | +import { icebergStageDeletionVector } from '../../src/write/stage-deletion-vector.js' |
9 | 11 | import { icebergStagePositionDelete } from '../../src/write/stage-position-delete.js' |
10 | 12 | import { icebergStageRewrite } from '../../src/write/rewrite.js' |
11 | 13 | import { deserializeValue } from '../../src/write/serde.js' |
@@ -54,17 +56,27 @@ function longBound(entry, fieldId, side) { |
54 | 56 | return bytes ? deserializeValue(bytes, 'long') : undefined |
55 | 57 | } |
56 | 58 |
|
| 59 | +/** |
| 60 | + * Map each row's id to its lineage pair, for before/after comparison. |
| 61 | + * @param {Record<string, any>[]} rows |
| 62 | + * @returns {Map<any, {rowId: any, lusn: any}>} |
| 63 | + */ |
| 64 | +function lineageById(rows) { |
| 65 | + return new Map(rows.map(r => [r.id, { rowId: r._row_id, lusn: r._last_updated_sequence_number }])) |
| 66 | +} |
| 67 | + |
57 | 68 | /** |
58 | 69 | * Create a sorted, multi-file table and return its committed metadata. |
59 | 70 | * @param {object} [opts] |
60 | 71 | * @param {SortOrder} [opts.sortOrder] |
| 72 | + * @param {2 | 3} [opts.formatVersion] |
61 | 73 | * @returns {Promise<{ tableUrl: string, resolver: import('../../src/types.js').Resolver, metadata: TableMetadata }>} |
62 | 74 | */ |
63 | | -async function makeMultiFileTable({ sortOrder } = {}) { |
| 75 | +async function makeMultiFileTable({ sortOrder, formatVersion } = {}) { |
64 | 76 | vi.spyOn(Date, 'now').mockReturnValue(1700000000000) |
65 | 77 | const tableUrl = 'mem://rewrite' |
66 | 78 | const { resolver } = memResolver() |
67 | | - let metadata = await icebergCreate({ tableUrl, resolver, schema, sortOrder }) |
| 79 | + let metadata = await icebergCreate({ tableUrl, resolver, schema, sortOrder, formatVersion }) |
68 | 80 | const batches = [ |
69 | 81 | [{ id: 5n, name: 'e' }, { id: 2n, name: 'b' }], |
70 | 82 | [{ id: 1n, name: 'a' }, { id: 6n, name: 'f' }], |
@@ -209,16 +221,155 @@ describe('icebergRewrite — safety', () => { |
209 | 221 | expect(rows.map(r => r.id).sort()).toEqual([1n, 2n]) |
210 | 222 | }) |
211 | 223 |
|
212 | | - it('rejects format-version 3 (row lineage not yet handled)', async () => { |
| 224 | + it('rejects unsupported format versions', async () => { |
| 225 | + const { tableUrl, resolver, metadata } = await makeMultiFileTable() |
| 226 | + await expect(() => icebergStageRewrite({ |
| 227 | + tableUrl, metadata: { ...metadata, 'format-version': 1 }, resolver, |
| 228 | + })).rejects.toThrow(/unsupported format-version: 1/) |
| 229 | + }) |
| 230 | +}) |
| 231 | + |
| 232 | +describe('icebergRewrite — v3 row lineage', () => { |
| 233 | + it('preserves _row_id and _last_updated_sequence_number across a sorted rewrite', async () => { |
| 234 | + const { tableUrl, resolver, metadata } = await makeMultiFileTable({ sortOrder: sortById, formatVersion: 3 }) |
| 235 | + expect(metadata['next-row-id']).toBe(6) |
| 236 | + const before = await icebergRead({ tableUrl, metadata, resolver }) |
| 237 | + expect(new Set(before.map(r => r._row_id))).toEqual(new Set([0n, 1n, 2n, 3n, 4n, 5n])) |
| 238 | + |
| 239 | + const staged = await icebergStageRewrite({ tableUrl, metadata, resolver }) |
| 240 | + // A pure rewrite carries existing rows, so it assigns no new row ids. |
| 241 | + expect(staged.snapshot['first-row-id']).toBe(6) |
| 242 | + expect(staged.snapshot['added-rows']).toBe(0) |
| 243 | + expect(staged.requirements).toContainEqual({ type: 'assert-next-row-id', 'next-row-id': 6 }) |
| 244 | + |
| 245 | + const after = await fileCatalogCommit({ tableUrl, metadata, staged, resolver }) |
| 246 | + expect(after['next-row-id']).toBe(6) |
| 247 | + |
| 248 | + const rows = await icebergRead({ tableUrl, metadata: after, resolver }) |
| 249 | + expect(rows.map(r => r.id)).toEqual([1n, 2n, 3n, 4n, 5n, 6n]) |
| 250 | + // The global sort scrambled row positions, so preserved values must come |
| 251 | + // from the materialized _row_id column, not positional derivation. |
| 252 | + expect(rows.map(r => r._row_id)).not.toEqual([0n, 1n, 2n, 3n, 4n, 5n]) |
| 253 | + expect(lineageById(rows)).toEqual(lineageById(before)) |
| 254 | + |
| 255 | + // The rewritten manifest is pinned to the carried range: min(_row_id). |
| 256 | + const snapshot = after.snapshots?.find(s => s['snapshot-id'] === after['current-snapshot-id']) |
| 257 | + const manifests = await fetchAvroRecords(snapshot?.['manifest-list'] ?? '', resolver) |
| 258 | + expect(manifests.length).toBe(1) |
| 259 | + expect(manifests[0].first_row_id).toBe(0n) |
| 260 | + }) |
| 261 | + |
| 262 | + it('preserves lineage across split output files', async () => { |
| 263 | + const { tableUrl, resolver, metadata } = await makeMultiFileTable({ sortOrder: sortById, formatVersion: 3 }) |
| 264 | + const before = await icebergRead({ tableUrl, metadata, resolver }) |
| 265 | + |
| 266 | + const staged = await icebergStageRewrite({ tableUrl, metadata, resolver, targetFileRows: 2 }) |
| 267 | + const after = await fileCatalogCommit({ tableUrl, metadata, staged, resolver }) |
| 268 | + |
| 269 | + const entries = splitManifestEntries(await icebergManifests({ metadata: after, resolver })).dataEntries |
| 270 | + expect(entries.length).toBe(3) |
| 271 | + const rows = await icebergRead({ tableUrl, metadata: after, resolver }) |
| 272 | + expect(lineageById(rows)).toEqual(lineageById(before)) |
| 273 | + expect(after['next-row-id']).toBe(6) |
| 274 | + }) |
| 275 | + |
| 276 | + it('consumes deletion vectors and keeps surviving rows\' lineage', async () => { |
213 | 277 | vi.spyOn(Date, 'now').mockReturnValue(1700000000000) |
214 | | - const tableUrl = 'mem://rewrite-v3' |
| 278 | + const tableUrl = 'mem://rewrite-v3-dv' |
215 | 279 | const { resolver } = memResolver() |
216 | 280 | const created = await icebergCreate({ tableUrl, resolver, schema, formatVersion: 3 }) |
217 | | - const appended = await icebergStageAppend({ tableUrl, metadata: created, records: [{ id: 1n, name: 'a' }], resolver }) |
| 281 | + const records = [{ id: 1n, name: 'a' }, { id: 2n, name: 'b' }, { id: 3n, name: 'c' }] |
| 282 | + const appended = await icebergStageAppend({ tableUrl, metadata: created, records, resolver }) |
218 | 283 | const afterAppend = await fileCatalogCommit({ tableUrl, metadata: created, staged: appended, resolver }) |
| 284 | + const dataPath = appended.writtenFiles[0] |
219 | 285 |
|
220 | | - await expect(() => icebergStageRewrite({ tableUrl, metadata: afterAppend, resolver })) |
221 | | - .rejects.toThrow(/format-version 2 only/) |
| 286 | + const delStaged = await icebergStageDeletionVector({ |
| 287 | + tableUrl, metadata: afterAppend, deletes: [{ file_path: dataPath, pos: 1n }], resolver, |
| 288 | + }) |
| 289 | + const afterDelete = await fileCatalogCommit({ tableUrl, metadata: afterAppend, staged: delStaged, resolver }) |
| 290 | + |
| 291 | + const staged = await icebergStageRewrite({ tableUrl, metadata: afterDelete, resolver }) |
| 292 | + const afterRewrite = await fileCatalogCommit({ tableUrl, metadata: afterDelete, staged, resolver }) |
| 293 | + |
| 294 | + const rows = await icebergRead({ tableUrl, metadata: afterRewrite, resolver }) |
| 295 | + expect(rows.map(r => ({ id: r.id, _row_id: r._row_id, _last_updated_sequence_number: r._last_updated_sequence_number }))).toEqual([ |
| 296 | + { id: 1n, _row_id: 0n, _last_updated_sequence_number: 1n }, |
| 297 | + { id: 3n, _row_id: 2n, _last_updated_sequence_number: 1n }, |
| 298 | + ]) |
| 299 | + const { dataEntries, deleteEntries } = splitManifestEntries(await icebergManifests({ metadata: afterRewrite, resolver })) |
| 300 | + expect(deleteEntries.length).toBe(0) |
| 301 | + expect(dataEntries.length).toBe(1) |
| 302 | + // The deleted row's id (1n) is retired with it; next-row-id is unchanged. |
| 303 | + expect(afterRewrite['next-row-id']).toBe(3) |
| 304 | + }) |
| 305 | + |
| 306 | + it('appends after a rewrite continue from the unchanged next-row-id', async () => { |
| 307 | + const { tableUrl, resolver, metadata } = await makeMultiFileTable({ sortOrder: sortById, formatVersion: 3 }) |
| 308 | + const staged = await icebergStageRewrite({ tableUrl, metadata, resolver }) |
| 309 | + const after = await fileCatalogCommit({ tableUrl, metadata, staged, resolver }) |
| 310 | + |
| 311 | + const appended = await icebergStageAppend({ |
| 312 | + tableUrl, metadata: after, records: [{ id: 7n, name: 'g' }, { id: 8n, name: 'h' }], resolver, |
| 313 | + }) |
| 314 | + const final = await fileCatalogCommit({ tableUrl, metadata: after, staged: appended, resolver }) |
| 315 | + expect(final['next-row-id']).toBe(8) |
| 316 | + |
| 317 | + const rows = await icebergRead({ tableUrl, metadata: final, resolver }) |
| 318 | + // New rows take fresh ids 6n and 7n; no collision with the preserved 0n-5n. |
| 319 | + expect(new Set(rows.map(r => r._row_id))).toEqual(new Set([0n, 1n, 2n, 3n, 4n, 5n, 6n, 7n])) |
| 320 | + expect(rows.find(r => r.id === 7n)?._row_id).toBe(6n) |
| 321 | + expect(rows.find(r => r.id === 8n)?._row_id).toBe(7n) |
| 322 | + }) |
| 323 | + |
| 324 | + it('keeps lineage stable through a rewrite of a rewrite', async () => { |
| 325 | + const { tableUrl, resolver, metadata } = await makeMultiFileTable({ sortOrder: sortById, formatVersion: 3 }) |
| 326 | + const before = await icebergRead({ tableUrl, metadata, resolver }) |
| 327 | + |
| 328 | + const staged1 = await icebergStageRewrite({ tableUrl, metadata, resolver }) |
| 329 | + const after1 = await fileCatalogCommit({ tableUrl, metadata, staged: staged1, resolver }) |
| 330 | + // Source files now carry materialized lineage columns. |
| 331 | + const staged2 = await icebergStageRewrite({ tableUrl, metadata: after1, resolver, targetFileRows: 3 }) |
| 332 | + const after2 = await fileCatalogCommit({ tableUrl, metadata: after1, staged: staged2, resolver }) |
| 333 | + |
| 334 | + const rows = await icebergRead({ tableUrl, metadata: after2, resolver }) |
| 335 | + expect(lineageById(rows)).toEqual(lineageById(before)) |
| 336 | + expect(after2['next-row-id']).toBe(6) |
| 337 | + }) |
| 338 | + |
| 339 | + it('assigns fresh row ids when rewriting an upgraded table whose rows lack lineage', async () => { |
| 340 | + vi.spyOn(Date, 'now').mockReturnValue(1700000000000) |
| 341 | + const tableUrl = 'mem://rewrite-v3-upgrade' |
| 342 | + const { resolver } = memResolver() |
| 343 | + const created = await icebergCreate({ tableUrl, resolver, schema }) |
| 344 | + const records = [{ id: 1n, name: 'a' }, { id: 2n, name: 'b' }, { id: 3n, name: 'c' }] |
| 345 | + const appended = await icebergStageAppend({ tableUrl, metadata: created, records, resolver }) |
| 346 | + const afterAppend = await fileCatalogCommit({ tableUrl, metadata: created, staged: appended, resolver }) |
| 347 | + /** @type {TableMetadata} */ |
| 348 | + const upgraded = { ...afterAppend, 'format-version': 3, 'next-row-id': 0 } |
| 349 | + |
| 350 | + // Pre-upgrade rows read with null lineage. |
| 351 | + const before = await icebergRead({ tableUrl, metadata: upgraded, resolver }) |
| 352 | + expect(before.every(r => r._row_id === null)).toBe(true) |
| 353 | + |
| 354 | + const staged = await icebergStageRewrite({ tableUrl, metadata: upgraded, resolver }) |
| 355 | + // No carried ids, so the rewritten manifest gets a fresh range at commit. |
| 356 | + expect(staged.snapshot['first-row-id']).toBe(0) |
| 357 | + expect(staged.snapshot['added-rows']).toBe(3) |
| 358 | + |
| 359 | + const after = await fileCatalogCommit({ tableUrl, metadata: upgraded, staged, resolver }) |
| 360 | + expect(after['next-row-id']).toBe(3) |
| 361 | + const rows = await icebergRead({ tableUrl, metadata: after, resolver }) |
| 362 | + expect(rows.map(r => r._row_id)).toEqual([0n, 1n, 2n]) |
| 363 | + }) |
| 364 | + |
| 365 | + it('detects a concurrent v3 commit via assert-next-row-id', async () => { |
| 366 | + const { tableUrl, resolver, metadata } = await makeMultiFileTable({ sortOrder: sortById, formatVersion: 3 }) |
| 367 | + const staged = await icebergStageRewrite({ tableUrl, metadata, resolver }) |
| 368 | + // Committing against metadata whose next-row-id moved (a concurrent append |
| 369 | + // that somehow passed the ref check) must fail rather than risk id reuse. |
| 370 | + await expect(fileCatalogCommit({ |
| 371 | + tableUrl, metadata: { ...metadata, 'next-row-id': 8 }, staged, resolver, |
| 372 | + })).rejects.toThrow(/next-row-id expected 6, got 8/) |
222 | 373 | }) |
223 | 374 | }) |
224 | 375 |
|
|
0 commit comments