Iceberg sink: upsert rendering pass. (#34033)

The iceberg sink is a multistage dataflow comprised of the following
operators:

- Batch description minting: The batch minter picks time bounds `x` time
wide into the future which all subsequent operators use.
- Iceberg Writer Iceberg writer writes data files (parquet) with times
bounded by the batch descriptions
- Iceberg committer Coalesces the files from all writers and commits
them to the catalog

## Missing
- [ ] metrics
- [ ] tests

<!--
Describe the contents of the PR briefly but completely.

If you write detailed commit messages, it is acceptable to copy/paste
them
here, or write "see commit messages for details." If there is only one
commit
in the PR, GitHub will have already added its commit message above.
-->

### Motivation

<!--
Which of the following best describes the motivation behind this PR?

  * This PR fixes a recognized bug.

    [Ensure issue is linked somewhere.]

  * This PR adds a known-desirable feature.

    [Ensure issue is linked somewhere.]

  * This PR fixes a previously unreported bug.

    [Describe the bug in detail, as if you were filing a bug report.]

  * This PR adds a feature that has not yet been specified.

[Write a brief specification for the feature, including justification
for its inclusion in Materialize, as if you were writing the original
     feature specification.]

   * This PR refactors existing code.

[Describe what was wrong with the existing code, if it is not obvious.]
-->

### Tips for reviewer

<!--
Leave some tips for your reviewer, like:

    * The diff is much smaller if viewed with whitespace hidden.
    * [Some function/module/file] deserves extra attention.
* [Some function/module/file] is pure code movement and only needs a
skim.

Delete this section if no tips.
-->

### Checklist

- [ ] This PR has adequate test coverage / QA involvement has been duly
considered. ([trigger-ci for additional test/nightly
runs](https://trigger-ci.dev.materialize.com/))
- [ ] This PR has an associated up-to-date [design
doc](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/README.md),
is a design doc
([template](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/00000000_template.md)),
or is sufficiently small to not require a design.
  <!-- Reference the design in the description. -->
- [ ] If this PR evolves [an existing `$T ⇔ Proto$T`
mapping](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/command-and-response-binary-encoding.md)
(possibly in a backwards-incompatible way), then it is tagged with a
`T-proto` label.
- [ ] If this PR will require changes to cloud orchestration or tests,
there is a companion cloud PR to account for those changes that is
tagged with the release-blocker label
([example](MaterializeInc/cloud#5021)).
<!-- Ask in #team-cloud on Slack if you need help preparing the cloud
PR. -->
- [ ] If this PR includes major [user-facing behavior
changes](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/guide-changes.md#what-changes-require-a-release-note),
I have pinged the relevant PM to schedule a changelog post.

Author: DAlperin

Cargo.lock

@@ -349,8 +349,9 @@ async-compression = { git = "https://github.com/MaterializeInc/async-compression
 
 # Custom iceberg features for mz
 # All changes should go to the `mz_changes` branch.
-iceberg = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "c2d70cbfc4822ea826198c7787c709beb652fa31" }
-iceberg-catalog-rest = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "c2d70cbfc4822ea826198c7787c709beb652fa31" }
+iceberg = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "cf1e6e0c9de3" }
+iceberg-catalog-rest = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "cf1e6e0c9de3" }
+iceberg-catalog-s3tables = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "cf1e6e0c9de3" }
 
 
 # BEGIN LINT CONFIG

Cargo.toml

@@ -127,8 +127,8 @@ skip = [
     { name = "darling_macro", version = "0.20.11" },
     { name = "darling_core", version = "0.20.11" },
     { name = "darling", version = "0.20.11" },
-    { name = "typed-builder", version = "0.21.2" },
-    { name = "typed-builder-macro", version = "0.21.2" },
+    { name = "typed-builder", version = "0.20.1" },
+    { name = "typed-builder-macro", version = "0.20.1" },
     # chrono-tz
     { name = "phf", version = "0.11.3" },
     { name = "phf_shared", version = "0.11.3" },

deny.toml

@@ -1418,6 +1418,7 @@ impl CatalogState {
                 with_snapshot,
                 resolved_ids,
                 cluster_id: in_cluster,
+                commit_interval: sink.commit_interval,
             }),
             Plan::CreateType(CreateTypePlan { typ, .. }) => {
                 // Even if we don't need the `RelationDesc` here, error out

src/adapter/src/catalog/state.rs

@@ -2862,6 +2862,7 @@ impl Coordinator {
                                     version: sink.version,
                                     from_storage_metadata: (),
                                     to_storage_metadata: (),
+                                    commit_interval: sink.commit_interval,
                                 },
                                 instance_id: sink.cluster_id,
                             },

src/adapter/src/coord.rs

@@ -987,6 +987,7 @@ impl Coordinator {
             version: sink.version,
             from_storage_metadata: (),
             to_storage_metadata: (),
+            commit_interval: sink.commit_interval,
         };
 
         let collection_desc = CollectionDescription {

src/adapter/src/coord/ddl.rs

@@ -1105,6 +1105,7 @@ impl Coordinator {
             with_snapshot,
             resolved_ids,
             cluster_id: in_cluster,
+            commit_interval: sink.commit_interval,
         };
 
         let ops = vec![catalog::Op::CreateItem {
@@ -3413,6 +3414,7 @@ impl Coordinator {
             with_snapshot,
             resolved_ids: resolved_ids.clone(),
             cluster_id: in_cluster,
+            commit_interval: sink_plan.commit_interval,
         };
 
         let ops = vec![catalog::Op::UpdateItem {
@@ -3448,6 +3450,7 @@ impl Coordinator {
             version: sink_plan.version,
             from_storage_metadata: (),
             to_storage_metadata: (),
+            commit_interval: sink_plan.commit_interval,
         };
 
         self.controller

src/adapter/src/coord/sequencer/inner.rs

@@ -27,11 +27,52 @@ use mz_ore::cast::CastFrom;
 use mz_repr::adt::jsonb::JsonbRef;
 use mz_repr::{Datum, RelationDesc, Row, SqlScalarType};
 
+const EXTENSION_PREFIX: &str = "materialize.v1.";
+
 pub struct ArrowBuilder {
     columns: Vec<ArrowColumn>,
     /// A crude estimate of the size of the data in the builder
     /// based on the size of the rows added to it.
     row_size_bytes: usize,
+    /// The original schema, if provided. Used to preserve metadata in to_record_batch().
+    original_schema: Option<Arc<Schema>>,
+}
+
+/// Converts a RelationDesc to an Arrow Schema.
+pub fn desc_to_schema(desc: &RelationDesc) -> Result<Schema, anyhow::Error> {
+    let mut fields = vec![];
+    let mut errs = vec![];
+    let mut seen_names = BTreeMap::new();
+    for (col_name, col_type) in desc.iter() {
+        let mut col_name = col_name.to_string();
+        // If we allow columns with the same name we encounter two issues:
+        // 1. The arrow crate will accidentally reuse the same buffers for the columns
+        // 2. Many parquet readers will error when trying to read the file metadata
+        // Instead we append a number to the end of the column name for any duplicates.
+        // TODO(roshan): We should document this when writing the copy-to-s3 MZ docs.
+        seen_names
+            .entry(col_name.clone())
+            .and_modify(|e: &mut u32| {
+                *e += 1;
+                col_name += &e.to_string();
+            })
+            .or_insert(1);
+        match scalar_to_arrow_datatype(&col_type.scalar_type) {
+            Ok((data_type, extension_type_name)) => {
+                fields.push(field_with_typename(
+                    &col_name,
+                    data_type,
+                    col_type.nullable,
+                    &extension_type_name,
+                ));
+            }
+            Err(err) => errs.push(err.to_string()),
+        }
+    }
+    if !errs.is_empty() {
+        anyhow::bail!("Relation contains unimplemented arrow types: {:?}", errs);
+    }
+    Ok(Schema::new(fields))
 }
 
 impl ArrowBuilder {
@@ -60,43 +101,51 @@ impl ArrowBuilder {
         item_capacity: usize,
         data_capacity: usize,
     ) -> Result<Self, anyhow::Error> {
+        let schema = desc_to_schema(desc)?;
         let mut columns = vec![];
-        let mut errs = vec![];
-        let mut seen_names = BTreeMap::new();
-        for (col_name, col_type) in desc.iter() {
-            let mut col_name = col_name.to_string();
-            // If we allow columns with the same name we encounter two issues:
-            // 1. The arrow crate will accidentally reuse the same buffers for the columns
-            // 2. Many parquet readers will error when trying to read the file metadata
-            // Instead we append a number to the end of the column name for any duplicates.
-            // TODO(roshan): We should document this when writing the copy-to-s3 MZ docs.
-            seen_names
-                .entry(col_name.clone())
-                .and_modify(|e: &mut u32| {
-                    *e += 1;
-                    col_name += &e.to_string();
-                })
-                .or_insert(1);
-            match scalar_to_arrow_datatype(&col_type.scalar_type) {
-                Ok((data_type, extension_type_name)) => {
-                    columns.push(ArrowColumn::new(
-                        col_name,
-                        col_type.nullable,
-                        data_type,
-                        extension_type_name,
-                        item_capacity,
-                        data_capacity,
-                    )?);
-                }
-                Err(err) => errs.push(err.to_string()),
-            }
+        for field in schema.fields() {
+            columns.push(ArrowColumn::new(
+                field.name().clone(),
+                field.is_nullable(),
+                field.data_type().clone(),
+                typename_from_field(field)?,
+                item_capacity,
+                data_capacity,
+            )?);
         }
-        if !errs.is_empty() {
-            anyhow::bail!("Relation contains unimplemented arrow types: {:?}", errs);
+        Ok(Self {
+            columns,
+            row_size_bytes: 0,
+            original_schema: None,
+        })
+    }
+
+    /// Initializes a new ArrowBuilder with a pre-built Arrow Schema.
+    /// This is useful when you need to preserve schema metadata (e.g., field IDs for Iceberg).
+    /// `item_capacity` is used to initialize the capacity of each column's builder which defines
+    /// the number of values that can be appended to each column before reallocating.
+    /// `data_capacity` is used to initialize the buffer size of the string and binary builders.
+    /// Errors if the schema contains an unimplemented type.
+    pub fn new_with_schema(
+        schema: Arc<Schema>,
+        item_capacity: usize,
+        data_capacity: usize,
+    ) -> Result<Self, anyhow::Error> {
+        let mut columns = vec![];
+        for field in schema.fields() {
+            columns.push(ArrowColumn::new(
+                field.name().clone(),
+                field.is_nullable(),
+                field.data_type().clone(),
+                typename_from_field(field)?,
+                item_capacity,
+                data_capacity,
+            )?);
         }
         Ok(Self {
             columns,
             row_size_bytes: 0,
+            original_schema: Some(schema),
         })
     }
 
@@ -118,7 +167,15 @@ impl ArrowBuilder {
             arrays.push(col.finish());
             fields.push((&col).into());
         }
-        RecordBatch::try_new(Schema::new(fields).into(), arrays)
+
+        // If we have an original schema, use it to preserve metadata (e.g., field IDs)
+        let schema = if let Some(original_schema) = self.original_schema {
+            original_schema
+        } else {
+            Arc::new(Schema::new(fields))
+        };
+
+        RecordBatch::try_new(schema, arrays)
     }
 
     /// Appends a row to the builder.
@@ -437,7 +494,7 @@ fn field_with_typename(
 ) -> Field {
     Field::new(name, data_type, nullable).with_metadata(HashMap::from([(
         "ARROW:extension:name".to_string(),
-        format!("materialize.v1.{}", extension_type_name),
+        format!("{}{}", EXTENSION_PREFIX, extension_type_name),
     )]))
 }
 
@@ -447,7 +504,7 @@ fn typename_from_field(field: &Field) -> Result<String, anyhow::Error> {
     let extension_name = metadata
         .get("ARROW:extension:name")
         .ok_or_else(|| anyhow::anyhow!("Missing extension name in metadata"))?;
-    if let Some(name) = extension_name.strip_prefix("materialize.v1") {
+    if let Some(name) = extension_name.strip_prefix(EXTENSION_PREFIX) {
         Ok(name.to_string())
     } else {
         anyhow::bail!("Extension name {} does not match expected", extension_name,)
@@ -709,8 +766,8 @@ impl ArrowColumn {
                 }
                 builder.append(true).unwrap()
             }
-            (_builder, datum) => {
-                anyhow::bail!("Datum {:?} does not match builder", datum)
+            (builder, datum) => {
+                anyhow::bail!("Datum {:?} does not match builder {:?}", datum, builder)
             }
         }
         Ok(())

src/arrow-util/src/builder.rs

@@ -15,9 +15,9 @@ workspace = true
 [dependencies]
 anyhow = "1.0.100"
 async-trait = "0.1.89"
-aws-config = { version = "1.2.0", default-features = false }
-aws-credential-types = { version = "1.2.10", features = ["hardcoded-credentials"] }
-aws-sdk-secretsmanager = { version = "1.45.0", default-features = false, features = ["rt-tokio"] }
+aws-config = { version = "1.8.10", default-features = false }
+aws-credential-types = { version = "1.2.8", features = ["hardcoded-credentials"] }
+aws-sdk-secretsmanager = { version = "1.93.0", default-features = false, features = ["rt-tokio"] }
 aws-types = "1.3.9"
 futures = { version = "0.3.31" }
 mz-aws-util = { path = "../aws-util", default-features = false }

src/aws-secrets-controller/Cargo.toml

@@ -11,8 +11,8 @@ workspace = true
 
 [dependencies]
 anyhow = "1.0.100"
-aws-config = { version = "1.2.0", default-features = false }
-aws-sdk-s3 = { version = "1.48.0", default-features = false, features = [
+aws-config = { version = "1.8.10", default-features = false }
+aws-sdk-s3 = { version = "1.94.0", default-features = false, features = [
     "rt-tokio",
 ], optional = true }
 aws-smithy-runtime-api = "1.9.2"