Skip to content

Commit e07d1aa

Browse files
committed
docs: add Iceberg row-level operations section and version metadata
Two follow-ups from the review: - The new row_operation and identifier_fields fields linked to a 'Row-level operations' section that didn't exist on the iceberg output page, so those cross-references were broken. Add the section (insert/upsert/delete semantics, identifier field rules, batching and ordering, and a change-data-capture example) so the links resolve and the feature from DOC-2256 is actually documented. - Tag global_table, global_table_replicas, row_operation, and identifier_fields with version 4.99.0 in the overrides. The upstream source only annotated snapshot_mode, so these four rendered without a version note. The generator picks the version up from the override on regeneration.
1 parent ef40018 commit e07d1aa

2 files changed

Lines changed: 85 additions & 2 deletions

File tree

docs-data/overrides.json

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -418,7 +418,13 @@
418418
},
419419
{
420420
"name": "global_table",
421-
"description": "Provision the checkpoint table as a DynamoDB Global Table (v2) so checkpoints replicate across regions. Requires `global_table_replicas`. When the table is auto-created it is created as a global table; when it already exists, its replicas are reconciled (missing regions are added by calling `UpdateTable`). The existing table must have been created in global mode (`TableId` hash key). Enabling this against a pre-existing non-global checkpoint table fails fast with a clear error."
421+
"description": "Provision the checkpoint table as a DynamoDB Global Table (v2) so checkpoints replicate across regions. Requires `global_table_replicas`. When the table is auto-created it is created as a global table; when it already exists, its replicas are reconciled (missing regions are added by calling `UpdateTable`). The existing table must have been created in global mode (`TableId` hash key). Enabling this against a pre-existing non-global checkpoint table fails fast with a clear error.",
422+
"version": "4.99.0"
423+
},
424+
{
425+
"name": "global_table_replicas",
426+
"description": "Regions other than this pipeline's own region to replicate the checkpoint table to. The pipeline's own region is always included. Required when `global_table` is true. Applied both when the checkpoint table is created and, for an existing global table, when reconciling replicas (missing regions are added; this list is not used to remove regions).",
427+
"version": "4.99.0"
422428
}
423429
]
424430
}
@@ -4002,7 +4008,13 @@
40024008
},
40034009
{
40044010
"name": "row_operation",
4005-
"description": "The row-level operation to apply for each message: `insert` (append), `upsert` (replace rows matching `identifier_fields`, then append), or `delete` (remove rows matching `identifier_fields`). Supports interpolation so the operation can be driven by the data, such as a change-data-capture stream's operation field. Defaults to `insert`, preserving the original append-only behavior.\n\nSee the <<row-level-operations,Row-level operations>> section above for the full semantics, the format-version-2 upgrade, batching behavior, and important caveats."
4011+
"description": "The row-level operation to apply for each message: `insert` (append), `upsert` (replace rows matching `identifier_fields`, then append), or `delete` (remove rows matching `identifier_fields`). Supports interpolation so the operation can be driven by the data, such as a change-data-capture stream's operation field. Defaults to `insert`, preserving the original append-only behavior.\n\nSee the <<row-level-operations,Row-level operations>> section above for the full semantics, the format-version-2 upgrade, batching behavior, and important caveats.",
4012+
"version": "4.99.0"
4013+
},
4014+
{
4015+
"name": "identifier_fields",
4016+
"description": "The columns forming the row identity (the Iceberg identifier fields, or equality-delete key) used by `upsert` and `delete`. Required when `row_operation` can evaluate to `upsert` or `delete`, and must reference existing table columns of a primitive, non-floating-point type.\n\nSee the <<row-level-operations,Row-level operations>> section above for the full constraints, including the temporal-type and partitioning rules and when the requirement is enforced.",
4017+
"version": "4.99.0"
40064018
},
40074019
{
40084020
"name": "catalog",

modules/components/pages/outputs/iceberg.adoc

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -106,6 +106,77 @@ object:struct
106106
array:list
107107
|===
108108

109+
[#row-level-operations]
110+
== Row-level operations
111+
112+
By default, this output is append-only: every message becomes a new row (`row_operation: insert`), and existing configurations are unaffected. Set `row_operation` to apply a per-message operation, with `identifier_fields` defining the row identity:
113+
114+
* `insert`: Append the row. This is an unconditional append, and is not keyed or deduplicated.
115+
* `upsert`: Replace any existing rows matching `identifier_fields`, then append this row. This is equivalent to Iceberg's Flink UPSERT mode.
116+
* `delete`: Remove rows matching `identifier_fields`.
117+
118+
`row_operation` supports interpolation, so the operation can be driven by the data itself, for example by mapping a change-data-capture stream's operation field. No CDC-specific format is assumed. The field is named `row_operation` to distinguish it from Iceberg's snapshot-level operation.
119+
120+
`upsert` and `delete` require `identifier_fields` and use Iceberg merge-on-read equality deletes, which require table format version 2. A version-1 table is automatically upgraded to version 2 on the first `upsert` or `delete`. This upgrade is irreversible.
121+
122+
=== Identifier fields
123+
124+
`identifier_fields` must reference existing table columns of a primitive, non-floating-point type. A static `upsert` or `delete` is validated at startup. An interpolated `row_operation` is validated for each message at write time, so an empty `identifier_fields` is not caught until the first `upsert` or `delete` message arrives.
125+
126+
Identifier columns of a temporal type (`timestamp`, `timestamptz`, `date`, `time`) must arrive as time values, not bare numbers. A numeric epoch is ambiguous as a delete key and is rejected at write time, so convert it to a timestamp upstream. If the table is partitioned, every partition source column must be one of the `identifier_fields`, because equality deletes are partition-scoped.
127+
128+
When this output auto-creates a table through `schema_evolution`, the `identifier_fields` columns are created as required and registered as the table's Iceberg identifier-field-ids, so downstream engines and other writers see the primary key. As a result, a null or missing value in an identifier column is rejected on write, even for `insert`. Identifier columns must therefore be present at creation, either in the first message or declared through `schema_metadata`. Pre-existing tables are never modified.
129+
130+
=== Batching and ordering
131+
132+
Within a single batch, the last `upsert` or `delete` for each `identifier_fields` key wins. Each batch containing an `upsert` or `delete` is committed as its own snapshot. These commits are never coalesced, which is required for correctness, so a high-throughput mutation workload produces one snapshot per batch. Size batches accordingly, and run regular table maintenance (snapshot expiry and compaction) to keep metadata manageable. Pure `insert`-only batches keep the original append fast path, which does coalesce commits.
133+
134+
Ordering only holds within a batch. With more than one batch in flight, concurrent batches can commit out of order, so a stale `upsert` might overwrite a newer one for the same key. Set `max_in_flight: 1` for keyed (change-data-capture) workloads to preserve per-key order. This is enforced by config linting whenever `row_operation` is anything other than a static `insert`.
135+
136+
[CAUTION]
137+
====
138+
`insert` is an unconditional append, and is not keyed or deduplicated. For keyed data, including change-data-capture, map create and read events to `upsert`, never `insert`. Mixing `insert` with `upsert` or `delete` on the same key in one batch produces duplicate rows.
139+
====
140+
141+
=== Change-data-capture example
142+
143+
Materialize a change-data-capture stream into an Iceberg table. The mapping derives the row operation from the source's operation field (here Debezium's `op`, where `c`, `r`, and `u` map to `upsert`, and `d` maps to `delete`) and selects the row image, while `identifier_fields` is the primary key:
144+
145+
[source,yaml]
146+
----
147+
input:
148+
redpanda:
149+
seed_brokers: [ localhost:9092 ]
150+
topics: [ dbserver.inventory.customers ]
151+
consumer_group: iceberg_sink
152+
153+
pipeline:
154+
processors:
155+
- mapping: |
156+
meta op = match this.op {
157+
"d" => "delete",
158+
_ => "upsert",
159+
}
160+
# Debezium puts the row image in 'after', or 'before' for deletes.
161+
root = this.after | this.before
162+
163+
output:
164+
iceberg:
165+
catalog:
166+
url: http://localhost:8181/api/catalog
167+
namespace: inventory
168+
table: customers
169+
row_operation: ${! metadata("op") }
170+
identifier_fields: [ id ]
171+
# Keyed writes must stay ordered. A single batch in flight prevents
172+
# concurrent batches from committing a stale update over a newer one.
173+
max_in_flight: 1
174+
storage:
175+
aws_s3:
176+
bucket: my-iceberg-data
177+
region: us-east-1
178+
----
179+
109180
include::connect:components:partial$fields/outputs/iceberg.adoc[]
110181

111182
// end::single-source[]

0 commit comments

Comments
 (0)