feat(sinks): partition iceberg exports by day with conversation sort (#91)

* feat(sinks): partition iceberg exports by day with conversation sort

Lay out @hypaware/format-iceberg exports for an archive's job, not the
cache's: partition by day(primaryTimestampColumn) — a writer-owned
default, not the cache's conversation_id-identity cachePartitioning,
which sets an unbounded ~1-file-per-conversation floor compaction can't
beat — and sort each day partition by the dataset's lookup columns
(conversation_id-led) so a conversation lookup prunes row groups by
min/max instead of needing a partition per conversation.

- Promote partitionSpecForDeclaration + validatePartitionSpecStability
  (and the declaration type) from src/core/cache/iceberg to a shared
  src/core/iceberg home, re-exported from src/core/index.js: they are
  core surface consumed by the registry, cache, plugin types, and now
  the export (LLP 0003).
- format-iceberg derives the day grain + sort order per dataset at
  commit time, creates the table with both, and rejects partition-spec
  drift on append (iceberg_partition_spec_drift). Emits hyp_partition_spec
  and hyp_sort_order on commit spans.
- Reframe maintenance compaction: available via icebergRewrite but not
  run in-daemon and not needed for a day grain (was "blocked by icebird").

Spec: LLP 0022 (rewritten from the abandoned cache-parity decision);
xrefs in LLP 0014 and 0003. Tests: 10 (derivation + drift through the
real icebird write path) plus a passing iceberg_export_partitioned_local_fs
smoke asserting the layout and hyp_partition_spec.

Clustering (icebird #22) and read pruning (#20/#21) require a published
icebird containing commit 3edb15b; the package.json pin must move off
0.8.5 before those benefits land. The code degrades gracefully on 0.8.5
— partitioning and drift work; the sort order is recorded but inert.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* chore(deps): bump icebird 0.8.5 -> 0.8.9 (sort-on-write + scan pruning)

Activates the conversation sort within day partitions and read-side
scan pruning that the partitioned export records in metadata. Clears
the merge blocker: 0.8.9 contains icebird 3edb15b.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(sinks): address PR #91 review — reverse drift guard, on-disk sort assertions, CLI text

- commitBatch now rejects reverse partition-spec drift: appending with no
  derived partitioning onto an already-partitioned table throws
  iceberg_partition_spec_drift instead of silently skipping the guard
  (and mislabeling spans as unpartitioned). LLP 0022#drift-rejection
  updated to record the guard as bidirectional; new test alongside the
  forward-drift case.
- The conversation sort is now asserted on disk, not just in metadata:
  both the commitBatch integration test and the partitioned smoke read a
  day-partition parquet file back with hyparquet and assert conversation_id
  row order — fails on an icebird that records the sort order but writes
  unsorted. Verified against icebird 0.8.9.
- hyp sink maintain CLI text reframed to match maintenance.js / LLP 0022:
  compaction is not run by this sink (out-of-band via icebergRewrite), not
  'unsupported by icebird'; action label is now compaction_out_of_band.
- Removed stray </content> artifact from the end of LLP 0022.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: philcunliffe

collectivus-plugin-kernel-types.d.ts

@@ -16,7 +16,7 @@
  */
 
 import type { AsyncDataSource, ScanOptions, ScanResults } from 'squirreling'
-import type { CachePartitioningDeclaration } from './src/core/cache/types.d.ts'
+import type { CachePartitioningDeclaration } from './src/core/iceberg/types.d.ts'
 
 export type { AsyncDataSource, ScanOptions, ScanResults }
 

hypaware-core/plugins-workspace/format-iceberg/src/commit.js

@@ -7,12 +7,14 @@ import {
   loadLatestFileCatalogMetadata,
 } from 'icebird'
 
+import { validatePartitionSpecStability } from '../../../../src/core/iceberg/partition-spec.js'
+
 import { icebergSchemaForColumns, mergeFieldIdsFromTable, rowsToIcebergRecords } from './schema.js'
 
 /**
  * @import { ColumnSpec } from '../../../../collectivus-plugin-kernel-types.d.ts'
- * @import { CommitInput, CommitResult, TableState } from './types.d.ts'
- * @import { Lister, Resolver, Snapshot, TableMetadata } from 'icebird/src/types.js'
+ * @import { CommitInput, CommitResult, DatasetPartitioning, TableState } from './types.d.ts'
+ * @import { Lister, PartitionSpec, Resolver, Snapshot, TableMetadata } from 'icebird/src/types.js'
  */
 
 /**
@@ -80,15 +82,46 @@ export async function commitBatch(input, priorState) {
 
   if (!priorState.exists) {
     try {
+      // @ref LLP 0022#partition-derivation — create with the writer-owned
+      // day-grain partitionSpec + conversation sort order. Both default to
+      // unpartitioned/unsorted in icebird when `partitioning` is absent.
       await icebergCreateTable({
         catalog,
         tableUrl: input.tableUrl,
         schema: targetSchema,
         formatVersion: 3,
+        partitionSpec: input.partitioning?.partitionSpec,
+        sortOrder: input.partitioning?.sortOrder,
       })
     } catch (err) {
       throw wrapCommitError(err, 'iceberg_commit_failed', `create table failed at '${input.tableUrl}'`)
     }
+  } else if (priorState.metadata) {
+    // @ref LLP 0022#drift-rejection — an existing table whose partition spec no
+    // longer matches the dataset's derived day grain is rejected; the export
+    // cannot retroactively repartition object-store data files. [constrained-by]
+    const existingSpec = currentPartitionSpec(priorState.metadata) ?? { 'spec-id': 0, fields: [] }
+    if (input.partitioning) {
+      try {
+        validatePartitionSpecStability(input.partitioning.declaration, existingSpec, targetSchema)
+      } catch (err) {
+        const message = err instanceof Error ? err.message : String(err)
+        throw newError(
+          'iceberg_partition_spec_drift',
+          `iceberg-format: partition spec drift at '${input.tableUrl}': ${message}`
+        )
+      }
+    } else if (existingSpec.fields.length > 0) {
+      // Reverse drift: the dataset stopped deriving partitioning but the
+      // table on disk is partitioned. The append itself would succeed (icebird
+      // keeps routing rows through the existing spec) while the sink reports
+      // `unpartitioned` — reject rather than let layout and telemetry diverge.
+      const existingLabel = existingSpec.fields.map((f) => `${f.transform}(${f.name})`).join(',')
+      throw newError(
+        'iceberg_partition_spec_drift',
+        `iceberg-format: partition spec drift at '${input.tableUrl}': dataset derives no partitioning but the existing table is partitioned (${existingLabel})`
+      )
+    }
   }
 
   /** @type {TableMetadata} */
@@ -142,6 +175,7 @@ const DEFAULT_STREAM_ROW_LIMIT = 100_000
  *   rows: AsyncIterable<Record<string, unknown>>,
  *   resolver: Resolver,
  *   lister: Lister,
+ *   partitioning?: DatasetPartitioning | null,
  * }} input
  * @param {{ exists: boolean, metadata: TableMetadata | null }} priorState
  * @param {{ batchByteLimit?: number, batchRowLimit?: number }} [opts]
@@ -167,7 +201,7 @@ export async function commitRowStream(input, priorState, opts = {}) {
   async function flushBatch() {
     if (batch.length === 0) return
     const result = await commitBatch(
-      { tableUrl: input.tableUrl, columns: input.columns, rows: batch, resolver: input.resolver, lister: input.lister },
+      { tableUrl: input.tableUrl, columns: input.columns, rows: batch, resolver: input.resolver, lister: input.lister, partitioning: input.partitioning },
       state
     )
     state = { exists: true, metadata: result.metadata }
@@ -225,6 +259,23 @@ function schemaFromExistingMetadata(columns, metadata) {
   return mergeFieldIdsFromTable(columns, existing)
 }
 
+/**
+ * Resolve the table's current `PartitionSpec` from metadata (the default spec,
+ * falling back to the last). Used for the on-append drift check.
+ *
+ * @param {TableMetadata} metadata
+ * @returns {PartitionSpec | undefined}
+ */
+function currentPartitionSpec(metadata) {
+  const specId = metadata['default-spec-id']
+  const specs = metadata['partition-specs']
+  if (specs?.length) {
+    const match = specs.find((s) => s['spec-id'] === specId)
+    return match ?? specs[specs.length - 1]
+  }
+  return undefined
+}
+
 /**
  * @param {unknown} value
  */

hypaware-core/plugins-workspace/format-iceberg/src/maintenance.js

@@ -117,9 +117,12 @@ export async function discoverExportDatasets(blobStore, prefix) {
  * Run export maintenance on all datasets under a prefix: snapshot
  * expiration per dataset, plus a compaction status report.
  *
- * icebird V1 does not expose `rewrite-data-files` or `delete-data-files`,
- * so compaction is not supported.  The report signals
- * `compactionSupported: false` so the CLI can surface a clear message.
+ * @ref LLP 0022#compaction — icebird now exposes `icebergRewrite`
+ * (read-rewrite compaction), but the export deliberately does not run it:
+ * day-grain partitioning already yields large files, and an in-daemon
+ * read-rewrite risks the OOM/blocking failure seen with the parquet sink.
+ * `compactionSupported: false` here means "not run by this sink" (out-of-band
+ * only), not "impossible".
  *
  * @param {{
  *   blobStore: BlobStore

hypaware-core/plugins-workspace/format-iceberg/src/partitioning.js

@@ -0,0 +1,98 @@
+// @ts-check
+
+import { partitionSpecForDeclaration } from '../../../../src/core/iceberg/partition-spec.js'
+
+import { icebergSchemaForColumns } from './schema.js'
+
+/**
+ * @import { ColumnSpec, DatasetRegistration } from '../../../../collectivus-plugin-kernel-types.d.ts'
+ * @import { CachePartitioningDeclaration } from '../../../../src/core/iceberg/types.d.ts'
+ * @import { Schema, SortField, SortOrder } from 'icebird/src/types.js'
+ * @import { DatasetPartitioning } from './types.d.ts'
+ */
+
+// @ref LLP 0022#partition-derivation — the export partitions by a writer-owned
+// day grain on the dataset's primaryTimestampColumn, derived independently of
+// the cache's `cachePartitioning` (which would impose an unbounded
+// per-conversation file count on an archive). [implements]
+/**
+ * Derive the export table's layout for a dataset: a `day(primaryTimestampColumn)`
+ * partition plus a within-partition sort on the dataset's lookup columns.
+ * Returns `null` when the dataset declares no `primaryTimestampColumn` present
+ * in its schema — that dataset exports unpartitioned (V1 behavior unchanged).
+ *
+ * @param {DatasetRegistration | undefined} reg
+ * @param {readonly ColumnSpec[]} columns
+ * @returns {DatasetPartitioning | null}
+ */
+export function derivePartitioning(reg, columns) {
+  if (!reg) return null
+  const tsColumn = typeof reg.primaryTimestampColumn === 'string' ? reg.primaryTimestampColumn : ''
+  if (!tsColumn) return null
+  // A primaryTimestampColumn that isn't in the exported schema can't anchor a
+  // day grain; fall back to unpartitioned rather than synthesize a bad spec.
+  if (!columns.some((c) => c.name === tsColumn)) return null
+
+  const schema = icebergSchemaForColumns(columns)
+  /** @type {CachePartitioningDeclaration} */
+  const declaration = {
+    source: { columns: [tsColumn] },
+    iceberg: { fields: [{ column: tsColumn, transform: 'day', required: true }] },
+  }
+  const partitionSpec = partitionSpecForDeclaration(declaration, schema)
+  const sortOrder = sortOrderForLookup(reg, schema)
+  return {
+    declaration,
+    partitionSpec,
+    sortOrder,
+    partitionSpecLabel: `day(${tsColumn})`,
+    sortOrderLabel: sortOrder.fields.map((f) => nameForSourceId(schema, f)).join(','),
+  }
+}
+
+// @ref LLP 0022#within-partition-sort — cluster each day partition by the
+// dataset's declared identity (lookup) columns so a conversation lookup prunes
+// row groups by min/max, without the file-count cost of partitioning on it.
+// This is the one place the export reads `cachePartitioning` — sort axis only.
+// [implements]
+/**
+ * Build a sort order from the dataset's declared identity columns
+ * (`cachePartitioning.iceberg.fields`, transform `identity`), in declared order.
+ * Returns an empty (unsorted) order when none apply — icebird treats that as a
+ * no-op, so an undeclared dataset is day-partitioned but unsorted.
+ *
+ * @param {DatasetRegistration} reg
+ * @param {Schema} schema
+ * @returns {SortOrder}
+ */
+function sortOrderForLookup(reg, schema) {
+  /** @type {Map<string, number>} */
+  const idByName = new Map(schema.fields.map((f) => [f.name, f.id]))
+  const declared = reg.cachePartitioning?.iceberg?.fields ?? []
+  /** @type {SortField[]} */
+  const fields = []
+  for (const f of declared) {
+    if (f.transform !== 'identity') continue
+    const id = idByName.get(f.column)
+    if (id === undefined) continue
+    fields.push({
+      'source-id': id,
+      transform: 'identity',
+      direction: 'asc',
+      'null-order': 'nulls-last',
+    })
+  }
+  // order-id 0 is conventionally "unsorted"; a real order uses 1.
+  return fields.length > 0 ? { 'order-id': 1, fields } : { 'order-id': 0, fields: [] }
+}
+
+/**
+ * @param {Schema} schema
+ * @param {SortField} field
+ * @returns {string}
+ */
+function nameForSourceId(schema, field) {
+  const id = field['source-id']
+  const match = schema.fields.find((f) => f.id === id)
+  return match ? match.name : String(id)
+}

hypaware-core/plugins-workspace/format-iceberg/src/table-format.js

@@ -5,6 +5,7 @@ import { getTracer, SpanStatusCode } from '../../../../src/core/observability/in
 import { createBlobStoreIO, pathToKey, tableUrlForBlobPrefix } from './blob-io.js'
 import { commitBatch, commitRowStream, probeTable } from './commit.js'
 import { expireExportSnapshots, normalizeExportRetentionConfig } from './maintenance.js'
+import { derivePartitioning } from './partitioning.js'
 import { loadMarker, markerKey, markerSubsumedBySnapshot, writeMarker } from './state.js'
 
 /**
@@ -195,6 +196,11 @@ async function exportDataset({ ctx, batch, dataset, partitions, prefix, log, mai
     return { partitionsExported: partitions.length, bytesWritten: 0, status: 'skipped' }
   }
 
+  // @ref LLP 0022#partition-derivation — derived per dataset at commit time
+  // because `createSink` runs once for a sink that exports many datasets, so
+  // the spec cannot be resolved up front. [implements]
+  const partitioning = derivePartitioning(ctx.query.getDataset(dataset), columns)
+
   const blobPrefix = joinKeys(pathToKey(prefix), sanitizeDataset(dataset))
   const tableUrl = tableUrlForBlobPrefix(blobPrefix)
   // Track the most recent metadata.json write so the commit span can
@@ -283,14 +289,18 @@ async function exportDataset({ ctx, batch, dataset, partitions, prefix, log, mai
         hyp_dataset: dataset,
         hyp_batch_id: batch.batchId,
         encoder_format: ctx.encoder.format,
+        // @ref LLP 0022#observability — surface the resolved layout so a smoke
+        // can assert what was written, not just that rows landed.
+        hyp_partition_spec: partitioning?.partitionSpecLabel ?? 'unpartitioned',
+        hyp_sort_order: partitioning?.sortOrderLabel ?? '',
         status: 'ok',
         ...destinationAttrs,
       },
     },
     async (span) => {
       try {
         const result = await commitRowStream(
-          { tableUrl, columns, rows: rowStream(), resolver, lister },
+          { tableUrl, columns, rows: rowStream(), resolver, lister, partitioning },
           { exists: priorState.exists, metadata: priorState.metadata }
         )
         span.setAttribute('snapshot_id', result.snapshotId)

hypaware-core/plugins-workspace/format-iceberg/src/types.d.ts

@@ -1,5 +1,6 @@
 import type { ColumnSpec } from '../../../../collectivus-plugin-kernel-types.d.ts'
-import type { TableMetadata, Resolver, Lister } from 'icebird/src/types.js'
+import type { TableMetadata, Resolver, Lister, PartitionSpec, SortOrder } from 'icebird/src/types.js'
+import type { CachePartitioningDeclaration } from '../../../../src/core/iceberg/types.d.ts'
 
 export interface TableState {
   /** True when at least one metadata file is visible. */
@@ -8,6 +9,24 @@ export interface TableState {
   currentSnapshotId: string | undefined
 }
 
+/**
+ * Writer-owned export layout for one dataset (LLP 0022): a day-grain partition
+ * derived from `primaryTimestampColumn`, plus a within-partition sort on the
+ * dataset's declared identity (lookup) columns.
+ */
+export interface DatasetPartitioning {
+  /** Synthesized day-grain declaration — kept for the on-append drift check. */
+  declaration: CachePartitioningDeclaration
+  /** Iceberg partition spec passed to `icebergCreateTable`. */
+  partitionSpec: PartitionSpec
+  /** Within-partition sort order; an empty order means unsorted (no-op). */
+  sortOrder: SortOrder
+  /** Span label, e.g. `day(message_created_at)`. */
+  partitionSpecLabel: string
+  /** Span label, e.g. `conversation_id,cwd,date` (empty when unsorted). */
+  sortOrderLabel: string
+}
+
 export interface CommitInput {
   /** Table URL the resolver understands. */
   tableUrl: string
@@ -17,6 +36,8 @@ export interface CommitInput {
   rows: readonly Record<string, unknown>[]
   resolver: Resolver
   lister: Lister
+  /** Day-grain partition + sort layout; absent ⇒ unpartitioned table. */
+  partitioning?: DatasetPartitioning | null
 }
 
 export interface CommitResult {

hypaware-core/smoke/flows/iceberg_export_local_fs.js

@@ -272,7 +272,7 @@ export async function run({ harness, expect }) {
     (ds) => Array.isArray(ds) && ds.some((d) => d.dataset === DATASET)
   )
   expect.that(
-    'maintain: compactionSupported is false (icebird V1 limitation)',
+    'maintain: compactionSupported is false (not run by this sink — out-of-band only, LLP 0022)',
     maintainReport.compactionSupported,
     (v) => v === false
   )