Add icebergUpdateSchema for schema evolution (#31)

Expose the existing add-schema / set-current-schema commit logic through a
stable public entry point so consumers can add nullable columns (or rename
columns, promote types) without recreating the table.

- icebergStageUpdateSchema: pure metadata-only stage with CAS requirements
  (assert-current-schema-id, assert-last-assigned-field-id), validated
  eagerly via applyUpdates
- icebergUpdateSchema: top-level commit through commitWithRetry, works for
  file and REST catalogs
- StagedCommit supertype with optional snapshot for metadata-only commits

Author: platypii

README.md

@@ -144,6 +144,7 @@ import {
   icebergExpireSnapshots,
   icebergRewrite,
   icebergSetRef,
+  icebergUpdateSchema,
 } from 'icebird'
 
 // `urlResolver()` ships with a `writer` (HTTP PUT) and `deleter` (HTTP DELETE);
@@ -173,6 +174,21 @@ await icebergDelete({
 // snapshot management
 await icebergSetRef({ catalog, tableUrl, ref: 'main', snapshotId })
 await icebergExpireSnapshots({ catalog, tableUrl, snapshotIds: [oldSnapshotId] })
+
+// schema evolution — pass the complete evolved schema; existing columns keep
+// their field ids, new columns use ids above the table's `last-column-id`.
+// Metadata-only: existing data files read the new column as `null`.
+await icebergUpdateSchema({
+  catalog, tableUrl,
+  schema: {
+    type: 'struct',
+    'schema-id': 0, // ignored; the next schema id is assigned at commit
+    fields: [
+      ...schema.fields,
+      { id: 3, name: 'score', required: false, type: 'double' },
+    ],
+  },
+})
 ```
 
 If the table is created with a `sortOrder`, `icebergAppend` orders the rows in each written file by that order (tightening per-file column bounds for scan pruning). `icebergRewrite` compacts the current snapshot — reading every live row (deletes applied), sorting globally, and rewriting into consolidated, non-overlapping files via a `replace` snapshot (v2 tables):

src/index.js

@@ -1,4 +1,4 @@
-export { IcebergTransactionConflictError, icebergAppend, icebergCreateTable, icebergDelete, icebergDropTable, icebergExpireSnapshots, icebergRewrite, icebergSetRef, icebergTransaction } from './write/write.js'
+export { IcebergTransactionConflictError, icebergAppend, icebergCreateTable, icebergDelete, icebergDropTable, icebergExpireSnapshots, icebergRewrite, icebergSetRef, icebergTransaction, icebergUpdateSchema } from './write/write.js'
 export { icebergCreate } from './create.js'
 export { fileCatalog } from './catalog/file.js'
 export { restCatalogConnect, restCatalogCreateNamespace, restCatalogDropNamespace, restCatalogListNamespaces, restCatalogListTables, restCatalogLoadCredentials, restCatalogLoadTable, restCatalogRegisterTable, restCatalogRenameTable } from './catalog/rest.js'

src/types.d.ts

@@ -318,17 +318,27 @@ export interface IcebergTransaction {
 }
 
 /**
- * Output of an `icebergStage*` call: the snapshot just produced, the CAS
- * preconditions and updates a catalog must apply, and the data/manifest files
- * already written to storage (useful for cleanup on commit failure).
+ * Input to a commit function (`fileCatalogCommit`, `restCatalogUpdateTable`):
+ * the CAS preconditions and updates a catalog must apply, and the
+ * data/manifest files already written to storage (useful for cleanup on
+ * commit failure). `snapshot` is absent for metadata-only updates like a
+ * schema change.
  */
-export interface StagedUpdate {
-  snapshot: Snapshot
+export interface StagedCommit {
+  snapshot?: Snapshot
   requirements: TableRequirement[]
   updates: TableUpdate[]
   writtenFiles: string[]
 }
 
+/**
+ * Output of an `icebergStage*` call that produces a snapshot (append, delete,
+ * rewrite, setRef, expireSnapshots).
+ */
+export interface StagedUpdate extends StagedCommit {
+  snapshot: Snapshot
+}
+
 /**
  * Output of `prepareAppend`. Captures everything that does NOT depend on the
  * eventually-committed snapshot's sequence number or parent: the data files,

src/write/commit.js

@@ -4,11 +4,11 @@ import { parseDecimalType } from './conversions.js'
 import { validatePartitionSpecForWrite } from './partition.js'
 
 /**
- * @import {Field, IcebergType, PartitionSpec, Resolver, Schema, SnapshotRef, SortOrder, StagedUpdate, TableMetadata, TableRequirement, TableUpdate} from '../../src/types.js'
+ * @import {Field, IcebergType, PartitionSpec, Resolver, Schema, SnapshotRef, SortOrder, StagedCommit, TableMetadata, TableRequirement, TableUpdate} from '../../src/types.js'
  */
 
 /**
- * Commit a `StagedUpdate` against a file-based catalog: verify requirements
+ * Commit a `StagedCommit` against a file-based catalog: verify requirements
  * against the current metadata, apply updates, and write the next
  * `vN.metadata.json` and `version-hint.text`.
  *
@@ -30,7 +30,7 @@ import { validatePartitionSpecForWrite } from './partition.js'
  *   prior version so rollback / log walks land on a real file even when the
  *   prior writer used `NNNNN-<uuid>.metadata.json` instead of `vN.metadata.json`.
  * @param {number} [options.currentVersion] - If known, the on-disk version of `metadata`. Bypasses deriving from `metadata-log`, which can be empty/stale on foreign-written tables.
- * @param {StagedUpdate} options.staged
+ * @param {StagedCommit} options.staged
  * @param {Resolver} options.resolver
  * @param {boolean} [options.conditionalCommits] - When true, write the metadata file with `ifNoneMatch: '*'`.
  * @returns {Promise<TableMetadata>} The new metadata, already persisted.

src/write/stage.js

@@ -1,5 +1,6 @@
 import { validateSchemaForVersion } from '../schema.js'
 import { uuid4 } from '../utils.js'
+import { applyUpdates } from './commit.js'
 import { writeParquet } from './parquet.js'
 import { writeDataManifest } from './manifest.js'
 import { groupByPartition } from './partition.js'
@@ -13,7 +14,7 @@ import { computeColumnStats } from './stats.js'
 
 /**
  * @import {CompressionCodec} from 'hyparquet'
- * @import {FieldSummary, Manifest, PreparedAppend, Resolver, Snapshot, StagedUpdate, TableMetadata, TableRequirement, TableUpdate} from '../../src/types.js'
+ * @import {FieldSummary, Manifest, PreparedAppend, Resolver, Schema, Snapshot, StagedCommit, StagedUpdate, TableMetadata, TableRequirement, TableUpdate} from '../../src/types.js'
  */
 
 /**
@@ -367,10 +368,8 @@ export function icebergStageExpireSnapshots({ metadata, snapshotIds }) {
   /** @type {TableUpdate} */
   const update = { action: 'remove-snapshots', 'snapshot-ids': snapshotIds }
 
-  // The snapshot field on StagedUpdate is non-optional; surface the current
-  // snapshot so callers reading `staged.snapshot` after an expire still see
-  // the live tip. Synthesize a minimal placeholder for tables with no
-  // snapshots left after the operation.
+  // Surface the current snapshot so callers reading `staged.snapshot` after
+  // an expire (e.g. `icebergTransaction`) still see the live tip.
   const tip = currentSnapshot(metadata) ?? snapshots[0]
   if (!tip) throw new Error('cannot expire snapshots from a table with no snapshots')
 
@@ -382,6 +381,56 @@ export function icebergStageExpireSnapshots({ metadata, snapshotIds }) {
   }
 }
 
+/**
+ * Stage a schema update: add `schema` as a new schema and make it current.
+ * This is the schema-evolution primitive — add a column, rename a column,
+ * promote a type — expressed as the full evolved schema.
+ *
+ * The caller supplies the complete new schema with field ids assigned:
+ * existing columns keep their ids, new columns use ids above
+ * `last-column-id`. The `schema-id` is assigned at commit time via the spec
+ * sentinel `-1`, so any `schema-id` on the input is ignored. Evolution rules
+ * (no field-id reuse, valid type promotions, immutable `initial-default`,
+ * required new fields need defaults) are validated here against the loaded
+ * metadata and again at commit.
+ *
+ * Pure: produces a metadata-only `StagedCommit` (no snapshot, no files) to
+ * pass into a commit function (`fileCatalogCommit`, `restCatalogUpdateTable`).
+ *
+ * @param {object} options
+ * @param {TableMetadata} options.metadata - Current table metadata.
+ * @param {Schema} options.schema - The complete evolved schema.
+ * @returns {StagedCommit}
+ */
+export function icebergStageUpdateSchema({ metadata, schema }) {
+  if (!schema || schema.type !== 'struct' || !Array.isArray(schema.fields)) {
+    throw new Error('schema must be a struct with a fields array')
+  }
+
+  /** @type {TableRequirement[]} */
+  const requirements = [
+    { type: 'assert-table-uuid', uuid: metadata['table-uuid'] },
+    { type: 'assert-current-schema-id', 'current-schema-id': metadata['current-schema-id'] },
+    { type: 'assert-last-assigned-field-id', 'last-assigned-field-id': metadata['last-column-id'] },
+  ]
+
+  /** @type {TableUpdate[]} */
+  const updates = [
+    { action: 'add-schema', schema: { ...schema, 'schema-id': -1 } },
+    { action: 'set-current-schema', 'schema-id': -1 },
+  ]
+
+  // Fail fast at stage time: applyUpdates enforces the schema-evolution rules
+  // that would otherwise only surface at commit (server-side for REST).
+  applyUpdates(metadata, updates)
+
+  return {
+    requirements,
+    updates,
+    writtenFiles: [],
+  }
+}
+
 /**
  * Reject `write.format.default` values other than parquet. Iceberg also
  * defines `avro` and `orc`, but Icebird only writes parquet today.

src/write/write.js

@@ -5,11 +5,11 @@ import { loadLatestFileCatalogMetadata } from '../metadata.js'
 import { applyUpdates, fileCatalogCommit } from './commit.js'
 import { icebergStageDeletionVector } from './stage-deletion-vector.js'
 import { icebergStagePositionDelete } from './stage-position-delete.js'
-import { icebergStageAppend, icebergStageExpireSnapshots, icebergStageSetRef, prepareAppend, stageSnapshotForAppend } from './stage.js'
+import { icebergStageAppend, icebergStageExpireSnapshots, icebergStageSetRef, icebergStageUpdateSchema, prepareAppend, stageSnapshotForAppend } from './stage.js'
 import { icebergStageRewrite } from './rewrite.js'
 
 /**
- * @import {Catalog, IcebergTransaction, Lister, PartitionSpec, Resolver, Schema, Snapshot, SortOrder, StagedUpdate, TableMetadata, TableRequirement, TableUpdate} from '../../src/types.js'
+ * @import {Catalog, IcebergTransaction, Lister, PartitionSpec, Resolver, Schema, Snapshot, SortOrder, StagedCommit, StagedUpdate, TableMetadata, TableRequirement, TableUpdate} from '../../src/types.js'
  */
 
 const DEFAULT_RETRY = Object.freeze({
@@ -177,6 +177,33 @@ export async function icebergSetRef({
   })
 }
 
+/**
+ * Evolve the table schema: add `schema` as a new schema version and make it
+ * current. Use it to add columns, rename columns, or promote types — pass the
+ * complete evolved schema with existing field ids preserved and new columns
+ * using ids above the table's `last-column-id`.
+ *
+ * Metadata-only: no data files are rewritten. Existing data files read the
+ * new columns as `null` (or their `initial-default`), and subsequent appends
+ * write with the evolved schema.
+ *
+ * @param {object} options
+ * @param {Catalog} options.catalog
+ * @param {string | string[]} [options.namespace] - REST catalog only.
+ * @param {string} [options.table] - REST catalog only.
+ * @param {string} [options.tableUrl] - File catalog only.
+ * @param {Resolver} [options.resolver]
+ * @param {Schema} options.schema - The complete evolved schema.
+ * @returns {Promise<TableMetadata>}
+ */
+export async function icebergUpdateSchema({ catalog, namespace, table, tableUrl, resolver, schema }) {
+  const ctx = await loadTable({ catalog, namespace, table, tableUrl, resolver })
+  return await commitWithRetry({
+    catalog, target: { namespace, table }, ctx,
+    stage: workingCtx => icebergStageUpdateSchema({ metadata: workingCtx.metadata, schema }),
+  })
+}
+
 /**
  * Expire one or more snapshots from a table. Data files are not removed from
  * storage; that is a separate maintenance pass.
@@ -499,7 +526,7 @@ function requireResolver(resolver, caller) {
  * @param {Catalog} catalog
  * @param {{namespace?: string | string[], table?: string}} target
  * @param {{metadata: TableMetadata, metadataFileName: string | undefined, version?: number, tableUrl: string, resolver: Resolver | undefined}} ctx
- * @param {StagedUpdate} staged
+ * @param {StagedCommit} staged
  * @returns {Promise<TableMetadata>}
  */
 async function commitStaged(catalog, target, ctx, staged) {
@@ -552,7 +579,7 @@ async function commitStaged(catalog, target, ctx, staged) {
  * @param {Catalog} options.catalog
  * @param {{namespace?: string | string[], table?: string}} options.target
  * @param {{metadata: TableMetadata, metadataFileName: string | undefined, version?: number, tableUrl: string, resolver: Resolver | undefined}} options.ctx - The initial loaded ctx; refreshed on retry.
- * @param {(workingCtx: {metadata: TableMetadata, metadataFileName: string | undefined, version?: number, tableUrl: string, resolver: Resolver | undefined}) => Promise<StagedUpdate> | StagedUpdate} options.stage
+ * @param {(workingCtx: {metadata: TableMetadata, metadataFileName: string | undefined, version?: number, tableUrl: string, resolver: Resolver | undefined}) => Promise<StagedCommit> | StagedCommit} options.stage
  * @returns {Promise<TableMetadata>}
  */
 async function commitWithRetry({ catalog, target, ctx, stage }) {