iceberg: repair non-compliant manifests at commit (#10641)

* iceberg: stamp a default name mapping on new tables

* iceberg: repair non-compliant manifests at commit

* s3tables: verify ClickHouse writes read back through PyIceberg

* iceberg: carry the manifest-list content into repaired manifests

* iceberg: refresh the default name mapping on schema evolution

* iceberg: merge historical names into the refreshed name mapping

* iceberg: never fail a commit on repair fallout

* iceberg: harden manifest repair against writer dialects

* s3tables: keep PyIceberg reader stderr out of row data

* iceberg: keep name mappings unambiguous across field id reassignment

* iceberg: align existing manifest content metadata with the list entry

test/s3tables/catalog_clickhouse/Dockerfile.writer

@@ -7,6 +7,6 @@ WORKDIR /app
 
 RUN pip install --no-cache-dir "pyiceberg[s3fs]==0.11.1" "pyarrow==25.0.0"
 
-COPY append_rows.py /app/
+COPY append_rows.py read_rows.py /app/
 
 ENTRYPOINT ["python3", "/app/append_rows.py"]

test/s3tables/catalog_clickhouse/README.md

@@ -32,6 +32,12 @@ database engine.
    - `ReadWrittenDataCount` and `ReadWrittenDataValues`: ClickHouse reads back
      the three PyIceberg-appended rows and the values match. This exercises the
      actual data path (parquet reads via S3), not just metadata.
+   - `WriteReadBack`: ClickHouse inserts rows with its experimental Iceberg
+     write support using default settings, which produces manifests without
+     avro field-ids, bucket-relative paths, and parquet without field ids. The
+     SeaweedFS catalog repairs the manifests at commit time and stamps a
+     default name mapping on the table, so PyIceberg (`read_rows.py`, a strict
+     reader) must return the rows ClickHouse wrote.
 
 Queries go through ClickHouse's HTTP interface (port 8123, mapped to a
 dynamically allocated host port), so the test needs no ClickHouse client

test/s3tables/catalog_clickhouse/clickhouse_catalog_test.go

@@ -105,6 +105,10 @@ func TestClickHouseIcebergCatalog(t *testing.T) {
 	buildClickHouseWriterImage(t)
 	writeIcebergRows(t, env, tableBucket, []string{namespace}, populatedTable)
 
+	// Empty table that ClickHouse writes into during the WriteReadBack subtest.
+	writeTable := "chwrite_" + randomString(6)
+	createIcebergTable(t, env, icebergToken, tableBucket, namespace, writeTable)
+
 	env.startClickHouseContainer(t)
 	env.waitForClickHouse(t, clickhouseStartTimeout)
 
@@ -169,6 +173,28 @@ func TestClickHouseIcebergCatalog(t *testing.T) {
 			t.Fatalf("SELECT id, label FROM %s = %q, want %q", populatedRef, out, want)
 		}
 	})
+
+	// ClickHouse's experimental Iceberg writes produce manifests without avro
+	// field-ids, bucket-relative paths, and parquet without field ids. The
+	// catalog repairs the manifests at commit and stamps a name mapping on the
+	// table, so a strict reader (PyIceberg) must see ClickHouse's rows.
+	t.Run("WriteReadBack", func(t *testing.T) {
+		writeRef := fmt.Sprintf("%s.`%s.%s`", clickhouseDatabase, namespace, writeTable)
+		insert := fmt.Sprintf("INSERT INTO %s (id, label) VALUES (1, 'alpha'), (2, 'beta')", writeRef)
+		if _, err := env.query(insert, map[string]string{"allow_experimental_insert_into_iceberg": "1"}); err != nil {
+			t.Fatalf("%s: %v\nContainer logs:\n%s", insert, err, clickhouseContainerLogs(env.clickhouseContainer))
+		}
+
+		out := env.mustQuery(t, fmt.Sprintf("SELECT id, label FROM %s ORDER BY id", writeRef))
+		if want := "1\talpha\n2\tbeta"; out != want {
+			t.Fatalf("ClickHouse read-back = %q, want %q", out, want)
+		}
+
+		rows := readIcebergRows(t, env, tableBucket, []string{namespace}, writeTable)
+		if want := "1,alpha\n2,beta"; rows != want {
+			t.Fatalf("PyIceberg read of ClickHouse-written table = %q, want %q", rows, want)
+		}
+	})
 }
 
 // NewTestEnvironment allocates ports and returns an environment for the test.
@@ -620,6 +646,42 @@ func writeIcebergRows(t *testing.T, env *TestEnvironment, bucketName string, nam
 	t.Logf("PyIceberg writer output: %s", strings.TrimSpace(string(out)))
 }
 
+// readIcebergRows scans a table with PyIceberg through the REST catalog and
+// returns its "id,label" lines, ordered by id.
+func readIcebergRows(t *testing.T, env *TestEnvironment, bucketName string, namespace []string, tableName string) string {
+	t.Helper()
+
+	args := []string{
+		"run", "--rm",
+		"--add-host", "host.docker.internal:host-gateway",
+		"--entrypoint", "python3",
+		clickhouseWriterImage,
+		"/app/read_rows.py",
+		"--catalog-url", fmt.Sprintf("http://host.docker.internal:%d", env.icebergPort),
+		"--warehouse", "s3://" + bucketName,
+		"--prefix", bucketName,
+		"--s3-endpoint", fmt.Sprintf("http://host.docker.internal:%d", env.s3Port),
+		"--access-key", env.accessKey,
+		"--secret-key", env.secretKey,
+		"--region", "us-west-2",
+		"--table", tableName,
+	}
+	for _, level := range namespace {
+		args = append(args, "--namespace", level)
+	}
+
+	// Keep stdout separate: the caller compares it exactly, and warnings on
+	// stderr from the python stack must not pollute the row data.
+	cmd := exec.Command("docker", args...)
+	var stdout, stderr bytes.Buffer
+	cmd.Stdout = &stdout
+	cmd.Stderr = &stderr
+	if err := cmd.Run(); err != nil {
+		t.Fatalf("PyIceberg reader failed: %v\nstdout:\n%s\nstderr:\n%s", err, stdout.String(), stderr.String())
+	}
+	return strings.TrimSpace(stdout.String())
+}
+
 // doIcebergJSONRequest issues an authenticated JSON request to the Iceberg
 // REST endpoint and returns the response body. It fails the test unless the
 // response status matches one of expectedStatuses.

test/s3tables/catalog_clickhouse/read_rows.py

@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+"""Read all rows of an Iceberg table via the SeaweedFS REST catalog.
+
+Used by the ClickHouse integration test to prove that rows written by
+ClickHouse are readable by another engine: PyIceberg is a strict reader that
+requires spec-compliant manifests and either parquet field ids or a name
+mapping. Prints one "id,label" line per row, ordered by id.
+"""
+
+import argparse
+import sys
+
+from pyiceberg.catalog import load_catalog
+
+
+def main() -> int:
+    p = argparse.ArgumentParser()
+    p.add_argument("--catalog-url", required=True)
+    p.add_argument("--warehouse", required=True)
+    p.add_argument("--prefix", required=True)
+    p.add_argument("--s3-endpoint", required=True)
+    p.add_argument("--access-key", required=True)
+    p.add_argument("--secret-key", required=True)
+    p.add_argument("--region", default="us-east-1")
+    p.add_argument("--namespace", action="append", required=True)
+    p.add_argument("--table", required=True)
+    args = p.parse_args()
+
+    catalog = load_catalog(
+        "rest",
+        **{
+            "type": "rest",
+            "uri": args.catalog_url,
+            "warehouse": args.warehouse,
+            "prefix": args.prefix,
+            "credential": f"{args.access_key}:{args.secret_key}",
+            "s3.access-key-id": args.access_key,
+            "s3.secret-access-key": args.secret_key,
+            "s3.endpoint": args.s3_endpoint,
+            "s3.region": args.region,
+            "s3.path-style-access": "true",
+        },
+    )
+
+    table = catalog.load_table(tuple(args.namespace) + (args.table,))
+    data = table.scan().to_arrow().to_pydict()
+    rows = sorted(zip(data["id"], data["label"]))
+    for row_id, label in rows:
+        print(f"{row_id},{label}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())