Skip to content

Commit 458e1a6

Browse files
Merge branch 'master' into cursor/add-s3-tables-support
2 parents 4b9e629 + fc274ee commit 458e1a6

10 files changed

Lines changed: 188 additions & 20 deletions

File tree

CHANGELOG.md

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
11
# Icebird Changelog
22

3+
## [0.8.12]
4+
- Add `scanColumn` streaming primitive to the Iceberg SQL data source
5+
- Fix commit crash when serializing metadata with snapshot ids above 2^53
6+
37
## [0.8.11]
48
- Fix pushed-down `where` combined with `limit`/`offset` silently dropping matching rows
59

package.json

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
{
22
"name": "icebird",
3-
"version": "0.8.11",
3+
"version": "0.8.12",
44
"description": "Apache Iceberg client for javascript",
55
"author": "Hyperparam",
66
"homepage": "https://hyperparam.app",
@@ -53,10 +53,10 @@
5353
"test": "vitest run"
5454
},
5555
"dependencies": {
56-
"hyparquet": "1.26.1",
56+
"hyparquet": "1.26.2",
5757
"hyparquet-compressors": "1.1.1",
5858
"hyparquet-writer": "0.16.1",
59-
"squirreling": "0.12.24"
59+
"squirreling": "0.12.25"
6060
},
6161
"peerDependencies": {
6262
"@aws-sdk/credential-providers": "^3.0.0"
@@ -68,10 +68,10 @@
6868
},
6969
"devDependencies": {
7070
"@aws-sdk/credential-providers": "3.1079.0",
71-
"@types/node": "26.0.0",
71+
"@types/node": "26.1.0",
7272
"@vitest/coverage-v8": "4.1.9",
7373
"eslint": "9.39.4",
74-
"eslint-plugin-jsdoc": "63.0.7",
74+
"eslint-plugin-jsdoc": "63.0.10",
7575
"typescript": "6.0.3",
7676
"vitest": "4.1.9"
7777
}

src/create.js

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
import { stringifyIcebergJson } from './json.js'
12
import { maxFieldId, validateSchemaForVersion } from './schema.js'
23
import { uuid4 } from './utils.js'
34
import { validatePartitionSpecForWrite } from './write/partition.js'
@@ -76,7 +77,7 @@ export async function icebergCreate({
7677
const metadataWriter = conditionalCommits
7778
? resolver.writer(metadataUrl, { ifNoneMatch: '*' })
7879
: resolver.writer(metadataUrl)
79-
const metadataBytes = new TextEncoder().encode(JSON.stringify(metadata, null, 2))
80+
const metadataBytes = new TextEncoder().encode(stringifyIcebergJson(metadata))
8081
metadataWriter.appendBytes(metadataBytes)
8182
await metadataWriter.finish()
8283

src/json.js

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -122,3 +122,43 @@ export function parseIcebergJson(text) {
122122
if (i !== text.length) throw new Error(`unexpected trailing input at ${i}`)
123123
return value
124124
}
125+
126+
/**
127+
* Serialize Iceberg metadata to JSON, emitting BigInt values as bare JSON
128+
* number literals (the inverse of `parseIcebergJson`). Plain `JSON.stringify`
129+
* throws on BigInt, and a naive replacer that returns a string would corrupt
130+
* the metadata by quoting 64-bit snapshot ids the spec requires as numbers.
131+
* Values above 2^53 are never coerced through Number, so precision is kept.
132+
*
133+
* Output matches `JSON.stringify(value, null, indent)` for the JSON value
134+
* types Iceberg metadata uses (objects, arrays, strings, numbers, booleans,
135+
* null) and additionally handles BigInt.
136+
*
137+
* @param {any} value
138+
* @param {number} [indent] - Spaces of indentation per level. Default 2.
139+
* @returns {string}
140+
*/
141+
export function stringifyIcebergJson(value, indent = 2) {
142+
const pad = ' '.repeat(indent)
143+
/**
144+
* @param {any} val
145+
* @param {number} depth
146+
* @returns {string}
147+
*/
148+
function serialize(val, depth) {
149+
if (typeof val === 'bigint') return val.toString()
150+
if (val === null || typeof val !== 'object') return JSON.stringify(val)
151+
const inner = pad.repeat(depth + 1)
152+
const outer = pad.repeat(depth)
153+
if (Array.isArray(val)) {
154+
if (val.length === 0) return '[]'
155+
const items = val.map(v => inner + serialize(v === undefined ? null : v, depth + 1))
156+
return `[\n${items.join(',\n')}\n${outer}]`
157+
}
158+
const keys = Object.keys(val).filter(k => val[k] !== undefined)
159+
if (keys.length === 0) return '{}'
160+
const items = keys.map(k => `${inner}${JSON.stringify(k)}: ${serialize(val[k], depth + 1)}`)
161+
return `{\n${items.join(',\n')}\n${outer}}`
162+
}
163+
return serialize(value, 0)
164+
}

src/metadata.js

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -86,7 +86,7 @@ export function icebergLatestVersion({ tableUrl, resolver, lister }) {
8686
})
8787
})
8888
.catch(err => {
89-
throw new Error(`failed to determine latest iceberg version: ${err.message}`)
89+
throw new Error(`failed to determine latest iceberg version of ${tableUrl}: ${err.message}`)
9090
})
9191
}
9292

@@ -115,7 +115,7 @@ export function icebergListVersions({ tableUrl, resolver, lister }) {
115115
return lister(metadataDir).then(metadataVersions)
116116
})
117117
.catch(err => {
118-
throw new Error(`failed to determine latest iceberg version: ${err.message}`)
118+
throw new Error(`failed to determine latest iceberg version of ${tableUrl}: ${err.message}`)
119119
})
120120
}
121121

src/write/commit.js

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
import { stringifyIcebergJson } from '../json.js'
12
import { maxFieldId, validateSchemaForVersion } from '../schema.js'
23
import { parseDecimalType } from './conversions.js'
34
import { validatePartitionSpecForWrite } from './partition.js'
@@ -75,7 +76,7 @@ export async function fileCatalogCommit({ tableUrl, metadata, metadataFileName,
7576
const metaWriter = conditionalCommits
7677
? resolver.writer(newMetadataPath, { ifNoneMatch: '*' })
7778
: resolver.writer(newMetadataPath)
78-
metaWriter.appendBytes(new TextEncoder().encode(JSON.stringify(newMetadata, null, 2)))
79+
metaWriter.appendBytes(new TextEncoder().encode(stringifyIcebergJson(newMetadata)))
7980
await metaWriter.finish()
8081

8182
// version-hint last so a partial write doesn't surface a torn commit.

src/write/write.js

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -187,7 +187,7 @@ export async function icebergSetRef({
187187
* @param {string} [options.table]
188188
* @param {string} [options.tableUrl]
189189
* @param {Resolver} [options.resolver]
190-
* @param {number[]} options.snapshotIds
190+
* @param {(number | bigint)[]} options.snapshotIds
191191
* @returns {Promise<TableMetadata>}
192192
*/
193193
export async function icebergExpireSnapshots({ catalog, namespace, table, tableUrl, resolver, snapshotIds }) {

test/read.test.js

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -14,13 +14,13 @@ describe.concurrent('icebergRead', () => {
1414
})
1515

1616
it('throws for fetch errors', async () => {
17-
// not found
17+
// not found; the error names the table it was looking for
1818
await expect(() => icebergRead({ tableUrl: 'https://hyperparam.app' }))
19-
.rejects.toThrow('failed to determine latest iceberg version')
19+
.rejects.toThrow('failed to determine latest iceberg version of https://hyperparam.app')
2020

2121
// invalid dns
2222
await expect(() => icebergRead({ tableUrl: 'https://nope.hyperparam.app' }))
23-
.rejects.toThrow('failed to determine latest iceberg version')
23+
.rejects.toThrow('failed to determine latest iceberg version of https://nope.hyperparam.app')
2424

2525
// with metadataFileName
2626
await expect(() => icebergRead({

test/sql/icebergDataSource.test.js

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -692,12 +692,11 @@ describe.concurrent('icebergDataSource scanColumn', () => {
692692
expect(viaHook).toHaveLength(5)
693693
})
694694

695-
it('characterizes: N aggregates on one column re-scan it N times (no coalescing)', async () => {
696-
// Current behavior, pinned for visibility — NOT an endorsement. squirreling
697-
// runs the streaming-aggregate fast path once per aggregate, so five
698-
// aggregates over one column re-scan the column five times. Coalescing them
699-
// into one pass is an upstream squirreling opportunity, not an icebird bug;
700-
// this characterization makes a future fix a visible diff here.
695+
it('characterizes: N aggregates on one column coalesce into a single scan', async () => {
696+
// Current behavior, pinned for visibility. squirreling coalesces the
697+
// streaming-aggregate fast path, so five aggregates over one column share a
698+
// single scan of that column. (Earlier squirreling releases re-scanned once
699+
// per aggregate; this characterization tracks the behavior as a visible diff.)
701700
const source = await icebergDataSource({ tableUrl, resolver, metadataFileName: 'v2.metadata.json' })
702701
const baseScanColumn = source.scanColumn
703702
if (!baseScanColumn) throw new Error('scanColumn not implemented')
@@ -718,6 +717,6 @@ describe.concurrent('icebergDataSource scanColumn', () => {
718717
query: 'SELECT COUNT("Popularity Rank") AS c, MIN("Popularity Rank") AS mn, MAX("Popularity Rank") AS mx, SUM("Popularity Rank") AS s, AVG("Popularity Rank") AS a FROM bunnies',
719718
}))
720719

721-
expect(scanColumnCalls).toBe(5)
720+
expect(scanColumnCalls).toBe(1)
722721
})
723722
})
Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,123 @@
1+
// Iceberg snapshot ids are 64-bit longs. Icebird's metadata reader parses
2+
// integers above Number.MAX_SAFE_INTEGER (2^53-1) as BigInt to avoid lossy
3+
// doubles. The write path must round-trip those BigInts: a plain
4+
// `JSON.stringify` throws "Do not know how to serialize a BigInt", and a
5+
// naive replacer that emits a quoted string would corrupt the metadata
6+
// (Iceberg expects bare JSON numbers). These tests pin the contract that a
7+
// commit re-serializing loaded metadata with a snapshot id > 2^53 succeeds
8+
// and preserves the id exactly.
9+
10+
import { describe, expect, it } from 'vitest'
11+
import { fileCatalog } from '../../src/catalog/file.js'
12+
import { loadLatestFileCatalogMetadata } from '../../src/metadata.js'
13+
import { icebergAppend, icebergCreateTable, icebergExpireSnapshots } from '../../src/write/write.js'
14+
import { memResolver } from '../helpers.js'
15+
16+
/**
17+
* @import {Schema} from '../../src/types.js'
18+
*/
19+
20+
// > 2^53 (9007199254740992) and < 2^63, so it is a valid 64-bit long that a
21+
// double cannot represent. parseIcebergJson keeps it as a BigInt.
22+
const BIG_ID = 9151314442816847871n
23+
24+
/** @type {Schema} */
25+
const schema = {
26+
type: 'struct',
27+
'schema-id': 0,
28+
fields: [
29+
{ id: 1, name: 'id', required: true, type: 'long' },
30+
{ id: 2, name: 'msg', required: false, type: 'string' },
31+
],
32+
}
33+
34+
/**
35+
* Rewrite every bare-number occurrence of `from` to `to` in the latest
36+
* on-disk metadata file, simulating a table whose snapshot id exceeds 2^53.
37+
* Operates on the raw JSON text so it does not depend on the serializer
38+
* under test. Word boundaries avoid rewriting a coincidental substring of a
39+
* larger number (e.g. a timestamp).
40+
*
41+
* @param {Map<string, Uint8Array>} files
42+
* @param {string} tableUrl
43+
* @param {bigint} from
44+
* @param {bigint} to
45+
*/
46+
function rewriteSnapshotId(files, tableUrl, from, to) {
47+
const dir = `${tableUrl}/metadata/`
48+
let latestKey = ''
49+
let latestVersion = -1
50+
for (const key of files.keys()) {
51+
const m = key.startsWith(dir) && key.match(/\/v(\d+)\.metadata\.json$/)
52+
if (m && Number(m[1]) > latestVersion) {
53+
latestVersion = Number(m[1])
54+
latestKey = key
55+
}
56+
}
57+
if (!latestKey) throw new Error('no metadata file found')
58+
const text = new TextDecoder().decode(/** @type {Uint8Array} */ (files.get(latestKey)))
59+
// Only rewrite bare JSON number values (`"key": <id>`), never digits inside
60+
// a quoted string such as the `manifest-list` path (`.../snap-<id>-...avro`),
61+
// whose physical file on disk keeps the original id in its name.
62+
const re = new RegExp(`(?<=: )${from.toString()}(?=[,\\n}\\]])`, 'g')
63+
const rewritten = text.replace(re, to.toString())
64+
files.set(latestKey, new TextEncoder().encode(rewritten))
65+
}
66+
67+
describe('commit round-trips snapshot ids above 2^53', () => {
68+
it('expireSnapshots preserves a surviving snapshot id > 2^53', async () => {
69+
const tableUrl = 'http://test/bigint-expire'
70+
const { resolver, files, lister } = memResolver()
71+
const catalog = fileCatalog({ resolver, lister })
72+
73+
await icebergCreateTable({ catalog, tableUrl, schema })
74+
await icebergAppend({ catalog, tableUrl, records: [{ id: 1n, msg: 'a' }] })
75+
await icebergAppend({ catalog, tableUrl, records: [{ id: 2n, msg: 'b' }] })
76+
77+
// Promote the current (newest) snapshot's id above 2^53 on disk.
78+
const before = await loadLatestFileCatalogMetadata({ tableUrl, resolver, lister })
79+
const currentId = before.metadata['current-snapshot-id']
80+
if (currentId == null) throw new Error('expected a current snapshot')
81+
const olderId = (before.metadata.snapshots ?? [])
82+
.map(s => s['snapshot-id'])
83+
.find(id => BigInt(id) !== BigInt(currentId))
84+
if (olderId == null) throw new Error('expected two snapshots')
85+
rewriteSnapshotId(files, tableUrl, BigInt(currentId), BIG_ID)
86+
87+
// Expiring the older snapshot re-serializes metadata that still holds the
88+
// BIG_ID snapshot. Before the fix this threw on JSON.stringify(BigInt).
89+
await icebergExpireSnapshots({ catalog, tableUrl, snapshotIds: [olderId] })
90+
91+
const after = await loadLatestFileCatalogMetadata({ tableUrl, resolver, lister })
92+
const ids = (after.metadata.snapshots ?? []).map(s => s['snapshot-id'])
93+
expect(ids).toEqual([BIG_ID])
94+
expect(after.metadata['current-snapshot-id']).toBe(BIG_ID)
95+
expect(after.metadata.refs?.main?.['snapshot-id']).toBe(BIG_ID)
96+
})
97+
98+
it('append records parent-snapshot-id > 2^53 without precision loss', async () => {
99+
const tableUrl = 'http://test/bigint-append'
100+
const { resolver, files, lister } = memResolver()
101+
const catalog = fileCatalog({ resolver, lister })
102+
103+
await icebergCreateTable({ catalog, tableUrl, schema })
104+
await icebergAppend({ catalog, tableUrl, records: [{ id: 1n, msg: 'a' }] })
105+
106+
// Promote the only snapshot's id above 2^53 on disk. A subsequent append
107+
// will reference it as parent-snapshot-id (a BigInt), which the commit
108+
// must serialize as a bare JSON number.
109+
const before = await loadLatestFileCatalogMetadata({ tableUrl, resolver, lister })
110+
const parentId = before.metadata['current-snapshot-id']
111+
if (parentId == null) throw new Error('expected a current snapshot')
112+
rewriteSnapshotId(files, tableUrl, BigInt(parentId), BIG_ID)
113+
114+
await icebergAppend({ catalog, tableUrl, records: [{ id: 2n, msg: 'b' }] })
115+
116+
const after = await loadLatestFileCatalogMetadata({ tableUrl, resolver, lister })
117+
const child = (after.metadata.snapshots ?? [])
118+
.find(s => BigInt(s['snapshot-id']) !== BIG_ID)
119+
if (!child) throw new Error('expected the appended snapshot')
120+
// Exact BigInt equality proves the parent id was not coerced to a double.
121+
expect(child['parent-snapshot-id']).toBe(BIG_ID)
122+
})
123+
})

0 commit comments

Comments
 (0)