From 923d0bd20cc5a1b185c465d82b58bd476f8c80db Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 8 Aug 2026 21:24:37 -0700 Subject: [PATCH] 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 --- .../catalog_clickhouse/Dockerfile.writer | 2 +- test/s3tables/catalog_clickhouse/README.md | 6 + .../clickhouse_catalog_test.go | 62 ++ test/s3tables/catalog_clickhouse/read_rows.py | 54 ++ weed/s3api/iceberg/commit_helpers.go | 1 + weed/s3api/iceberg/handlers_commit.go | 29 + weed/s3api/iceberg/handlers_table.go | 18 +- weed/s3api/iceberg/manifest_repair.go | 618 ++++++++++++++++++ weed/s3api/iceberg/manifest_repair_test.go | 281 ++++++++ weed/s3api/iceberg/metadata_files.go | 28 +- weed/s3api/iceberg/name_mapping.go | 142 ++++ weed/s3api/iceberg/name_mapping_test.go | 213 ++++++ .../b317f0e5-10f3-41d6-86bc-8b42f01f3e86.avro | Bin 0 -> 2845 bytes ...-51798e6c-427b-4fc2-9149-99e75d96e045.avro | Bin 0 -> 1591 bytes .../690958a2-9936-4a04-8117-d8104c977195.avro | Bin 0 -> 2829 bytes ...-48b874af-90f6-4e5e-8002-df91e59c8a0c.avro | Bin 0 -> 1575 bytes 16 files changed, 1441 insertions(+), 13 deletions(-) create mode 100644 test/s3tables/catalog_clickhouse/read_rows.py create mode 100644 weed/s3api/iceberg/manifest_repair.go create mode 100644 weed/s3api/iceberg/manifest_repair_test.go create mode 100644 weed/s3api/iceberg/name_mapping.go create mode 100644 weed/s3api/iceberg/name_mapping_test.go create mode 100644 weed/s3api/iceberg/testdata/clickhouse_abs/b317f0e5-10f3-41d6-86bc-8b42f01f3e86.avro create mode 100644 weed/s3api/iceberg/testdata/clickhouse_abs/snap-331611856-2-51798e6c-427b-4fc2-9149-99e75d96e045.avro create mode 100644 weed/s3api/iceberg/testdata/clickhouse_rel/690958a2-9936-4a04-8117-d8104c977195.avro create mode 100644 weed/s3api/iceberg/testdata/clickhouse_rel/snap-449670589-2-48b874af-90f6-4e5e-8002-df91e59c8a0c.avro diff --git a/test/s3tables/catalog_clickhouse/Dockerfile.writer b/test/s3tables/catalog_clickhouse/Dockerfile.writer index 5bda8c143..e940a1c79 100644 --- a/test/s3tables/catalog_clickhouse/Dockerfile.writer +++ b/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"] diff --git a/test/s3tables/catalog_clickhouse/README.md b/test/s3tables/catalog_clickhouse/README.md index 350a233c5..139070593 100644 --- a/test/s3tables/catalog_clickhouse/README.md +++ b/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 diff --git a/test/s3tables/catalog_clickhouse/clickhouse_catalog_test.go b/test/s3tables/catalog_clickhouse/clickhouse_catalog_test.go index e7ccff85d..b89eb6a68 100644 --- a/test/s3tables/catalog_clickhouse/clickhouse_catalog_test.go +++ b/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. diff --git a/test/s3tables/catalog_clickhouse/read_rows.py b/test/s3tables/catalog_clickhouse/read_rows.py new file mode 100644 index 000000000..3c1d5c5dc --- /dev/null +++ b/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()) diff --git a/weed/s3api/iceberg/commit_helpers.go b/weed/s3api/iceberg/commit_helpers.go index d5740ac06..2acf34044 100644 --- a/weed/s3api/iceberg/commit_helpers.go +++ b/weed/s3api/iceberg/commit_helpers.go @@ -133,6 +133,7 @@ func (s *Server) finalizeCreateOnCommit(ctx context.Context, input createOnCommi message: "Failed to apply statistics updates: " + err.Error(), } } + metadataBytes = refreshDefaultNameMapping(metadataBytes, newMetadata) // Same spec-compliance fixup we apply on create-table; ensures // v{N}.metadata.json files written through this create-on-commit path are // also readable by strict Iceberg clients reading directly from S3. diff --git a/weed/s3api/iceberg/handlers_commit.go b/weed/s3api/iceberg/handlers_commit.go index 78bd26d33..c62c774cb 100644 --- a/weed/s3api/iceberg/handlers_commit.go +++ b/weed/s3api/iceberg/handlers_commit.go @@ -66,6 +66,30 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) { return } } + // Manifest repair runs once, as soon as the table location is known; on + // commit retries the updates already reference the repaired files. Repair + // is best effort end to end: the originals parsed already, so a repair + // that fails to re-parse is discarded rather than failing the commit. + manifestsRepaired := false + repairManifests := func(location string) { + if manifestsRepaired { + return + } + manifestsRepaired = true + repaired, changed := s.repairAddSnapshotManifests(r.Context(), location, raw.Updates) + if !changed { + return + } + repairedUpdates, repairedStatistics, err := parseCommitUpdates(repaired) + if err != nil { + glog.Warningf("Iceberg: repaired updates failed to parse, keeping originals: %v", err) + return + } + raw.Updates = repaired + req.Updates = repairedUpdates + statisticsUpdates = repairedStatistics + } + maxCommitAttempts := 3 generatedLegacyUUID := uuid.New() stageCreateEnabled := isStageCreateEnabled() @@ -168,6 +192,8 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) { } } + repairManifests(location) + result, reqErr := s.finalizeCreateOnCommit(r.Context(), createOnCommitInput{ bucketARN: bucketARN, markerBucket: bucketName, @@ -232,6 +258,8 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) { } } + repairManifests(location) + builder, err := table.MetadataBuilderFromBase(currentMetadata, getResp.MetadataLocation) if err != nil { writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to create metadata builder: "+err.Error()) @@ -266,6 +294,7 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "BadRequestException", "Failed to apply statistics updates: "+err.Error()) return } + metadataBytes = refreshDefaultNameMapping(metadataBytes, newMetadata) // Same spec-compliance fixup we apply on create-table; ensures // v{N}.metadata.json files written during commit are also readable by // strict Iceberg clients reading directly from S3, and that the diff --git a/weed/s3api/iceberg/handlers_table.go b/weed/s3api/iceberg/handlers_table.go index 616a6ac25..6d9a4e7dc 100644 --- a/weed/s3api/iceberg/handlers_table.go +++ b/weed/s3api/iceberg/handlers_table.go @@ -1,6 +1,7 @@ package iceberg import ( + "bytes" "context" "encoding/json" "errors" @@ -851,5 +852,20 @@ func newTableMetadata( } // Create metadata directly using the constructor which ensures spec compliance for V2 - return table.NewMetadataWithUUID(s, pSpec, so, location, props, tableUUID) + metadata, err := table.NewMetadataWithUUID(s, pSpec, so, location, props, tableUUID) + if err != nil { + return nil, err + } + // The constructor reassigns field ids, so the default name mapping must be + // derived from the final schema rather than the request's. + raw, err := json.Marshal(metadata) + if err != nil { + return nil, err + } + if patched := refreshDefaultNameMapping(raw, metadata); !bytes.Equal(patched, raw) { + if withMapping, err := table.ParseMetadataBytes(patched); err == nil { + return withMapping, nil + } + } + return metadata, nil } diff --git a/weed/s3api/iceberg/manifest_repair.go b/weed/s3api/iceberg/manifest_repair.go new file mode 100644 index 000000000..ea629ffdd --- /dev/null +++ b/weed/s3api/iceberg/manifest_repair.go @@ -0,0 +1,618 @@ +package iceberg + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "strings" + + "github.com/linkedin/goavro/v2" + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/wdclient" +) + +// Some engines (ClickHouse's experimental Iceberg writes among them) commit +// snapshots whose manifest avro files omit the spec's field-id annotations and +// reference files by bucket-relative path. Both make the snapshot unreadable +// for other engines even though the data itself is fine. Since the catalog is +// the commit chokepoint, add-snapshot updates are repaired here: the manifest +// list and manifests are transcoded with the annotations added and the paths +// absolutized, and the snapshot is pointed at the repaired copies. + +// avroFieldIDs describes the spec-mandated ids for one avro field and the +// fields nested beneath it (through unions, arrays, and records). +type avroFieldIDs struct { + id int + elementID int + children map[string]avroFieldIDs +} + +// manifestListFieldIDs returns the fixed v2 manifest_file schema ids. +func manifestListFieldIDs() map[string]avroFieldIDs { + return map[string]avroFieldIDs{ + "manifest_path": {id: 500}, + "manifest_length": {id: 501}, + "partition_spec_id": {id: 502}, + "content": {id: 517}, + "sequence_number": {id: 515}, + "min_sequence_number": {id: 516}, + "added_snapshot_id": {id: 503}, + "added_files_count": {id: 504}, + "existing_files_count": {id: 505}, + "deleted_files_count": {id: 506}, + "added_rows_count": {id: 512}, + "existing_rows_count": {id: 513}, + "deleted_rows_count": {id: 514}, + "partitions": {id: 507, elementID: 508, children: map[string]avroFieldIDs{ + "contains_null": {id: 509}, + "contains_nan": {id: 518}, + "lower_bound": {id: 510}, + "upper_bound": {id: 511}, + }}, + "key_metadata": {id: 519}, + "first_row_id": {id: 520}, + } +} + +// manifestEntryFieldIDs returns the fixed v2 manifest_entry schema ids. The +// partition struct is table-specific; its ids come from the partition-spec +// JSON embedded in the manifest's OCF metadata. +func manifestEntryFieldIDs(partitionIDs map[string]avroFieldIDs) map[string]avroFieldIDs { + return map[string]avroFieldIDs{ + "status": {id: 0}, + "snapshot_id": {id: 1}, + "sequence_number": {id: 3}, + "file_sequence_number": {id: 4}, + "data_file": {id: 2, children: map[string]avroFieldIDs{ + "content": {id: 134}, + "file_path": {id: 100}, + "file_format": {id: 101}, + "partition": {id: 102, children: partitionIDs}, + "record_count": {id: 103}, + "file_size_in_bytes": {id: 104}, + "column_sizes": {id: 108, children: kvFieldIDs(117, 118)}, + "value_counts": {id: 109, children: kvFieldIDs(119, 120)}, + "null_value_counts": {id: 110, children: kvFieldIDs(121, 122)}, + "nan_value_counts": {id: 137, children: kvFieldIDs(138, 139)}, + "lower_bounds": {id: 125, children: kvFieldIDs(126, 127)}, + "upper_bounds": {id: 128, children: kvFieldIDs(129, 130)}, + "key_metadata": {id: 131}, + "split_offsets": {id: 132, elementID: 133}, + "equality_ids": {id: 135, elementID: 136}, + "sort_order_id": {id: 140}, + "first_row_id": {id: 142}, + "referenced_data_file": {id: 143}, + "content_offset": {id: 144}, + "content_size_in_bytes": {id: 145}, + }}, + } +} + +func kvFieldIDs(keyID, valueID int) map[string]avroFieldIDs { + return map[string]avroFieldIDs{ + "key": {id: keyID}, + "value": {id: valueID}, + } +} + +// partitionFieldIDsFromSpec extracts name -> field-id from the partition-spec +// JSON carried in a manifest's OCF metadata. +func partitionFieldIDsFromSpec(specJSON []byte) (map[string]avroFieldIDs, error) { + if len(specJSON) == 0 { + return map[string]avroFieldIDs{}, nil + } + var fields []struct { + Name string `json:"name"` + FieldID int `json:"field-id"` + } + if err := json.Unmarshal(specJSON, &fields); err != nil { + return nil, fmt.Errorf("parse partition-spec metadata: %w", err) + } + ids := make(map[string]avroFieldIDs, len(fields)) + for _, f := range fields { + ids[f.Name] = avroFieldIDs{id: f.FieldID} + } + return ids, nil +} + +// annotateAvroRecordFields adds "field-id" to every field of a parsed avro +// record schema, recursing into nested types. Returns whether anything was +// added, and ok=false when a field has no known id (the schema is left for an +// unknown writer dialect rather than half-annotated). +func annotateAvroRecordFields(record map[string]any, ids map[string]avroFieldIDs) (changed, ok bool) { + fields, _ := record["fields"].([]any) + if len(fields) == 0 { + return false, true + } + for _, f := range fields { + field, isMap := f.(map[string]any) + if !isMap { + return changed, false + } + name, _ := field["name"].(string) + spec, known := ids[name] + if !known { + return changed, false + } + if _, has := field["field-id"]; !has { + field["field-id"] = spec.id + changed = true + } + typeChanged, typeOK := annotateAvroType(field["type"], spec) + changed = changed || typeChanged + if !typeOK { + return changed, false + } + } + return changed, true +} + +// annotateAvroType annotates the schema node of a single field: unions are +// followed through their non-null branches, arrays get "element-id", and +// nested records recurse with the field's child ids. +func annotateAvroType(node any, spec avroFieldIDs) (changed, ok bool) { + switch t := node.(type) { + case []any: // union + for _, branch := range t { + branchChanged, branchOK := annotateAvroType(branch, spec) + changed = changed || branchChanged + if !branchOK { + return changed, false + } + } + return changed, true + case map[string]any: + switch t["type"] { + case "record": + childIDs := spec.children + if childIDs == nil { + childIDs = map[string]avroFieldIDs{} + } + return annotateAvroRecordFields(t, childIDs) + case "array": + if spec.elementID != 0 { + if _, has := t["element-id"]; !has { + t["element-id"] = spec.elementID + changed = true + } + } + itemChanged, itemOK := annotateAvroType(t["items"], spec) + return changed || itemChanged, itemOK + case "map": + // The spec wants key-id/value-id on genuine avro maps; this repair + // only knows the array-of-key_value encoding, so leave the file to + // its writer rather than half-annotating it. + return changed, false + default: + return false, true + } + default: // primitive name or named-type reference + return false, true + } +} + +// transcodeManifestAvro decodes an avro OCF file, annotates its schema with +// the given ids, applies mutate to each datum and fixMeta to the OCF metadata, +// and re-encodes. When nothing changes, the original bytes are returned +// unchanged. +func transcodeManifestAvro(raw []byte, ids map[string]avroFieldIDs, mutate func(map[string]any) bool, fixMeta func(map[string][]byte) bool) ([]byte, bool, error) { + reader, err := goavro.NewOCFReader(bytes.NewReader(raw)) + if err != nil { + return nil, false, fmt.Errorf("open avro: %w", err) + } + meta := reader.MetaData() + + outMeta := make(map[string][]byte, len(meta)) + for k, v := range meta { + if k == "avro.schema" || k == "avro.codec" { + continue + } + outMeta[k] = v + } + metaChanged := false + if fixMeta != nil { + metaChanged = fixMeta(outMeta) + } + + var schema map[string]any + if err := json.Unmarshal(meta["avro.schema"], &schema); err != nil { + return nil, false, fmt.Errorf("parse avro schema: %w", err) + } + schemaChanged, ok := annotateAvroRecordFields(schema, ids) + if !ok { + return nil, false, fmt.Errorf("avro schema has fields outside the Iceberg manifest spec") + } + + datumChanged := false + var datums []any + for reader.Scan() { + datum, err := reader.Read() + if err != nil { + return nil, false, fmt.Errorf("read avro datum: %w", err) + } + if record, isMap := datum.(map[string]any); isMap && mutate != nil { + if mutate(record) { + datumChanged = true + } + } + datums = append(datums, datum) + } + if err := reader.Err(); err != nil { + return nil, false, fmt.Errorf("scan avro: %w", err) + } + + if !schemaChanged && !datumChanged && !metaChanged { + return raw, false, nil + } + + annotatedSchema, err := json.Marshal(schema) + if err != nil { + return nil, false, err + } + var buf bytes.Buffer + writer, err := goavro.NewOCFWriter(goavro.OCFConfig{ + W: &buf, + Schema: string(annotatedSchema), + MetaData: outMeta, + }) + if err != nil { + return nil, false, fmt.Errorf("create avro writer: %w", err) + } + if err := writer.Append(datums); err != nil { + return nil, false, fmt.Errorf("write avro datums: %w", err) + } + return buf.Bytes(), true, nil +} + +// manifestStore abstracts reading and writing files in a table's metadata +// directory so the repair logic is testable without a filer. +type manifestStore interface { + loadFile(ctx context.Context, location string) ([]byte, error) + saveFile(ctx context.Context, location string, data []byte) error +} + +// absolutizeLocation resolves a bucket-relative path against the bucket of the +// table location. Locations that already carry a scheme pass through. +func absolutizeLocation(location, tableLocation string) string { + if location == "" || strings.Contains(location, "://") { + return location + } + bucket, _, err := parseS3Location(tableLocation) + if err != nil { + return location + } + return "s3://" + bucket + "/" + strings.TrimPrefix(location, "/") +} + +// metadataFileName returns the file name when location is a direct child of +// the table's metadata directory, which is the only place repair will read +// from or write to. +func metadataFileName(tableLocation, location string) (string, bool) { + prefix := strings.TrimSuffix(tableLocation, "/") + "/metadata/" + name := strings.TrimPrefix(location, prefix) + if name == location || name == "" || strings.Contains(name, "/") { + return "", false + } + return name, true +} + +const repairedManifestPrefix = "repaired-" + +// maxRepairableManifestSize bounds how much manifest data the repair path +// buffers in memory; anything larger fails the repair and the commit +// proceeds with the writer's original files. +const maxRepairableManifestSize = 64 << 20 + +// hasFieldIDAnnotations is the cheap compliance probe: spec-compliant writers +// annotate every manifest avro schema with field-id, and a writer that +// annotates the manifest list annotates its manifests too. +func hasFieldIDAnnotations(raw []byte) bool { + reader, err := goavro.NewOCFReader(bytes.NewReader(raw)) + if err != nil { + return false + } + return bytes.Contains(reader.MetaData()["avro.schema"], []byte(`"field-id"`)) +} + +// repairManifestList loads a snapshot's manifest list and, when it needs +// repair, rewrites it (and the manifests it references) into spec-compliant +// copies next to the originals. Returns the manifest list location the +// snapshot should reference. +func repairManifestList(ctx context.Context, store manifestStore, tableLocation, listLocation string) (string, bool, error) { + absListLocation := absolutizeLocation(listLocation, tableLocation) + listName, inMetadataDir := metadataFileName(tableLocation, absListLocation) + if !inMetadataDir { + return listLocation, false, nil + } + + listBytes, err := store.loadFile(ctx, absListLocation) + if err != nil { + return listLocation, false, fmt.Errorf("load manifest list: %w", err) + } + if absListLocation == listLocation && hasFieldIDAnnotations(listBytes) { + return listLocation, false, nil + } + + metadataDir := strings.TrimSuffix(tableLocation, "/") + "/metadata/" + var manifestErr error + repairedList, listChanged, err := transcodeManifestAvro(listBytes, manifestListFieldIDs(), func(record map[string]any) bool { + manifestPath, _ := record["manifest_path"].(string) + if manifestPath == "" || manifestErr != nil { + return false + } + absManifest := absolutizeLocation(manifestPath, tableLocation) + recordChanged := absManifest != manifestPath + if recordChanged { + record["manifest_path"] = absManifest + } + + manifestName, ok := metadataFileName(tableLocation, absManifest) + if !ok { + return recordChanged + } + manifestBytes, err := store.loadFile(ctx, absManifest) + if err != nil { + manifestErr = fmt.Errorf("load manifest %s: %w", absManifest, err) + return recordChanged + } + repaired, changed, err := repairManifest(manifestBytes, tableLocation, manifestContentValue(record)) + if err != nil { + manifestErr = fmt.Errorf("repair manifest %s: %w", absManifest, err) + return recordChanged + } + if !changed { + return recordChanged + } + repairedLocation := metadataDir + repairedManifestPrefix + manifestName + if err := store.saveFile(ctx, repairedLocation, repaired); err != nil { + manifestErr = fmt.Errorf("save repaired manifest %s: %w", repairedLocation, err) + return recordChanged + } + record["manifest_path"] = repairedLocation + record["manifest_length"] = int64(len(repaired)) + return true + }, nil) + if err != nil { + return listLocation, false, err + } + if manifestErr != nil { + return listLocation, false, manifestErr + } + if !listChanged && absListLocation == listLocation { + return listLocation, false, nil + } + + repairedListLocation := metadataDir + repairedManifestPrefix + listName + if listChanged { + if err := store.saveFile(ctx, repairedListLocation, repairedList); err != nil { + return listLocation, false, fmt.Errorf("save repaired manifest list: %w", err) + } + return repairedListLocation, true, nil + } + // Only the pointer was relative; the file itself is fine. + return absListLocation, true, nil +} + +// manifestContentValue reads the manifest-list entry's content field +// (0 = data, 1 = deletes); v1 lists have no such field and default to data. +// Union-typed values arrive from goavro as a single-branch map. +func manifestContentValue(record map[string]any) int { + value := record["content"] + if union, isUnion := value.(map[string]any); isUnion { + for _, branch := range union { + value = branch + break + } + } + switch v := value.(type) { + case int32: + return int(v) + case int64: + return int(v) + case int: + return v + } + return 0 +} + +// fixManifestOCFMetadata fixes OCF metadata keys the spec requires of +// manifests but lax writers omit or contradict: "content" must agree with the +// manifest-list entry, which is what readers trust for scan planning, and +// "partition-spec-id" (ClickHouse writes the underscore variant). +func fixManifestOCFMetadata(listContent int) func(map[string][]byte) bool { + return func(meta map[string][]byte) bool { + changed := false + content := "data" + if listContent == 1 { + content = "deletes" + } + if string(meta["content"]) != content { + meta["content"] = []byte(content) + changed = true + } + if _, ok := meta["partition-spec-id"]; !ok { + if v, ok := meta["partition_spec_id"]; ok { + meta["partition-spec-id"] = v + changed = true + } + } + return changed + } +} + +// repairManifest annotates a single manifest's schema and absolutizes the +// data file paths inside it. listContent is the content declared by the +// manifest-list entry referencing this manifest. +func repairManifest(raw []byte, tableLocation string, listContent int) ([]byte, bool, error) { + reader, err := goavro.NewOCFReader(bytes.NewReader(raw)) + if err != nil { + return nil, false, fmt.Errorf("open manifest avro: %w", err) + } + partitionIDs, err := partitionFieldIDsFromSpec(reader.MetaData()["partition-spec"]) + if err != nil { + return nil, false, err + } + return transcodeManifestAvro(raw, manifestEntryFieldIDs(partitionIDs), func(record map[string]any) bool { + dataFile, _ := record["data_file"].(map[string]any) + if dataFile == nil { + return false + } + filePath, _ := dataFile["file_path"].(string) + absPath := absolutizeLocation(filePath, tableLocation) + if absPath == filePath { + return false + } + dataFile["file_path"] = absPath + return true + }, fixManifestOCFMetadata(listContent)) +} + +// repairAddSnapshotUpdates rewrites the manifest-list references of +// add-snapshot updates whose manifests need repair. Repair is best effort: on +// any error the original update is kept and the commit proceeds as it would +// have without repair. +func repairAddSnapshotUpdates(ctx context.Context, store manifestStore, tableLocation string, rawUpdates []json.RawMessage) ([]json.RawMessage, bool) { + changedAny := false + out := make([]json.RawMessage, len(rawUpdates)) + copy(out, rawUpdates) + for i, rawUpdate := range rawUpdates { + var probe struct { + Action string `json:"action"` + } + if err := json.Unmarshal(rawUpdate, &probe); err != nil || probe.Action != "add-snapshot" { + continue + } + var update map[string]json.RawMessage + if err := json.Unmarshal(rawUpdate, &update); err != nil { + continue + } + var snapshot map[string]json.RawMessage + if err := json.Unmarshal(update["snapshot"], &snapshot); err != nil { + continue + } + var listLocation string + if err := json.Unmarshal(snapshot["manifest-list"], &listLocation); err != nil || listLocation == "" { + continue + } + + repairedLocation, changed, err := repairManifestList(ctx, store, tableLocation, listLocation) + if err != nil { + glog.Warningf("Iceberg: manifest repair for %s skipped: %v", listLocation, err) + continue + } + if !changed { + continue + } + + locationJSON, err := json.Marshal(repairedLocation) + if err != nil { + continue + } + snapshot["manifest-list"] = locationJSON + snapshotJSON, err := json.Marshal(snapshot) + if err != nil { + continue + } + update["snapshot"] = snapshotJSON + updateJSON, err := json.Marshal(update) + if err != nil { + continue + } + out[i] = updateJSON + changedAny = true + glog.V(1).Infof("Iceberg: repaired manifest list %s -> %s", listLocation, repairedLocation) + } + return out, changedAny +} + +// serverManifestStore reads and writes table metadata files through the filer. +type serverManifestStore struct { + server *Server + tableLocation string +} + +func (st *serverManifestStore) split(location string) (bucket, tablePath, fileName string, err error) { + fileName, ok := metadataFileName(st.tableLocation, location) + if !ok { + return "", "", "", fmt.Errorf("%s is outside the table metadata directory", location) + } + bucket, tablePath, err = parseS3Location(st.tableLocation) + return bucket, tablePath, fileName, err +} + +func (st *serverManifestStore) loadFile(ctx context.Context, location string) ([]byte, error) { + bucket, tablePath, fileName, err := st.split(location) + if err != nil { + return nil, err + } + return st.server.loadTableMetadataBlob(ctx, bucket, tablePath, fileName) +} + +func (st *serverManifestStore) saveFile(ctx context.Context, location string, data []byte) error { + bucket, tablePath, fileName, err := st.split(location) + if err != nil { + return err + } + return st.server.saveMetadataBlob(ctx, bucket, tablePath, fileName, data, "application/avro") +} + +// repairAddSnapshotManifests is the server entry point used by the commit +// handlers. +func (s *Server) repairAddSnapshotManifests(ctx context.Context, tableLocation string, rawUpdates []json.RawMessage) ([]json.RawMessage, bool) { + store := &serverManifestStore{server: s, tableLocation: tableLocation} + return repairAddSnapshotUpdates(ctx, store, tableLocation, rawUpdates) +} + +// lookupFileIDAdapter lets filer.StreamContent resolve volume locations +// through the iceberg server's filer client. +type lookupFileIDAdapter struct { + fn wdclient.LookupFileIdFunctionType +} + +func (a *lookupFileIDAdapter) GetLookupFileIdFunction() wdclient.LookupFileIdFunctionType { + return a.fn +} + +// loadTableMetadataBlob reads a file from the table's metadata directory, +// following chunks for files that were written through the S3 gateway rather +// than inline by the catalog itself. +func (s *Server) loadTableMetadataBlob(ctx context.Context, bucketName, tablePath, fileName string) ([]byte, error) { + var entry *filer_pb.Entry + err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{ + Directory: metadataDirPath(bucketName, tablePath), + Name: fileName, + }) + if err != nil { + return err + } + entry = resp.Entry + return nil + }) + if err != nil { + return nil, err + } + if entry == nil { + return nil, fmt.Errorf("no entry for %s", fileName) + } + if len(entry.Content) > 0 || len(entry.GetChunks()) == 0 { + return entry.Content, nil + } + if size := filer.FileSize(entry); size > maxRepairableManifestSize { + return nil, fmt.Errorf("%s is %d bytes, larger than the %d byte repair limit", fileName, size, maxRepairableManifestSize) + } + + fullClient, ok := s.filerClient.(filer_pb.FilerClient) + if !ok { + return nil, fmt.Errorf("filer client cannot resolve chunk locations") + } + var buf bytes.Buffer + lookup := &lookupFileIDAdapter{fn: filer.LookupFn(fullClient)} + if err := filer.StreamContent(lookup, &buf, entry.GetChunks(), 0, int64(filer.FileSize(entry))); err != nil { + return nil, fmt.Errorf("read %s: %w", fileName, err) + } + return buf.Bytes(), nil +} diff --git a/weed/s3api/iceberg/manifest_repair_test.go b/weed/s3api/iceberg/manifest_repair_test.go new file mode 100644 index 000000000..b13fbeca3 --- /dev/null +++ b/weed/s3api/iceberg/manifest_repair_test.go @@ -0,0 +1,281 @@ +package iceberg + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + + icebergmanifest "github.com/apache/iceberg-go" + "github.com/linkedin/goavro/v2" +) + +// The testdata fixtures were written by ClickHouse 25.8's experimental +// Iceberg insert against a SeaweedFS table bucket. clickhouse_rel holds the +// default output (bucket-relative paths, no field-id annotations); +// clickhouse_abs was written with write_full_path_in_iceberg_metadata=1 +// (absolute paths, still no field-ids). + +type fakeManifestStore struct { + files map[string][]byte + saved map[string][]byte +} + +func newFakeManifestStore() *fakeManifestStore { + return &fakeManifestStore{files: map[string][]byte{}, saved: map[string][]byte{}} +} + +func (f *fakeManifestStore) loadFile(_ context.Context, location string) ([]byte, error) { + if data, ok := f.files[location]; ok { + return data, nil + } + return nil, fmt.Errorf("no file at %s", location) +} + +func (f *fakeManifestStore) saveFile(_ context.Context, location string, data []byte) error { + f.files[location] = data + f.saved[location] = data + return nil +} + +func loadFixture(t *testing.T, parts ...string) []byte { + t.Helper() + data, err := os.ReadFile(filepath.Join(append([]string{"testdata"}, parts...)...)) + if err != nil { + t.Fatalf("read fixture: %v", err) + } + return data +} + +const ( + relTableLocation = "s3://iceberg-tables/wprobe/target_rel" + relListName = "snap-449670589-2-48b874af-90f6-4e5e-8002-df91e59c8a0c.avro" + relManifestName = "690958a2-9936-4a04-8117-d8104c977195.avro" + absTableLocation = "s3://iceberg-tables/wprobe/target" + absListName = "snap-331611856-2-51798e6c-427b-4fc2-9149-99e75d96e045.avro" + absManifestName = "b317f0e5-10f3-41d6-86bc-8b42f01f3e86.avro" +) + +// verifyRepairedManifests checks the repaired files with iceberg-go's strict +// typed readers: the manifest list parses, references the repaired manifest +// with the right length, and the manifest's data file path is absolute. +func verifyRepairedManifests(t *testing.T, store *fakeManifestStore, listLocation, tableLocation string) { + t.Helper() + + listBytes, ok := store.files[listLocation] + if !ok { + t.Fatalf("repaired manifest list %s was not saved", listLocation) + } + if !bytes.Contains(listBytes, []byte(`"field-id"`)) { + t.Fatalf("repaired manifest list schema has no field-id annotations") + } + + manifests, err := icebergmanifest.ReadManifestList(bytes.NewReader(listBytes)) + if err != nil { + t.Fatalf("iceberg-go cannot read repaired manifest list: %v", err) + } + if len(manifests) != 1 { + t.Fatalf("manifest list has %d entries, want 1", len(manifests)) + } + manifestPath := manifests[0].FilePath() + if !strings.HasPrefix(manifestPath, tableLocation+"/metadata/") { + t.Fatalf("manifest path %s is not absolute under the table location", manifestPath) + } + + manifestBytes, ok := store.files[manifestPath] + if !ok { + t.Fatalf("manifest %s not present in store", manifestPath) + } + if manifests[0].Length() != int64(len(manifestBytes)) { + t.Fatalf("manifest_length %d != stored size %d", manifests[0].Length(), len(manifestBytes)) + } + if !bytes.Contains(manifestBytes, []byte(`"field-id"`)) { + t.Fatalf("repaired manifest schema has no field-id annotations") + } + + entries, err := icebergmanifest.ReadManifest(manifests[0], bytes.NewReader(manifestBytes), false) + if err != nil { + t.Fatalf("iceberg-go cannot read repaired manifest: %v", err) + } + if len(entries) != 1 { + t.Fatalf("manifest has %d entries, want 1", len(entries)) + } + dataFile := entries[0].DataFile() + if !strings.HasPrefix(dataFile.FilePath(), "s3://iceberg-tables/") { + t.Fatalf("data file path %s is not absolute", dataFile.FilePath()) + } + if dataFile.Count() != 2 { + t.Fatalf("record count %d, want 2", dataFile.Count()) + } + if len(dataFile.LowerBoundValues()) == 0 { + t.Fatalf("lower bounds were lost in repair") + } +} + +func addSnapshotUpdateJSON(t *testing.T, manifestList string) []json.RawMessage { + t.Helper() + update := fmt.Sprintf(`{"action":"add-snapshot","snapshot":{"snapshot-id":449670589,"sequence-number":1,"timestamp-ms":1754625869000,"manifest-list":%q,"summary":{"operation":"append"},"schema-id":0}}`, manifestList) + return []json.RawMessage{json.RawMessage(update)} +} + +func TestRepairClickHouseRelativeManifests(t *testing.T) { + store := newFakeManifestStore() + store.files[relTableLocation+"/metadata/"+relListName] = loadFixture(t, "clickhouse_rel", relListName) + store.files[relTableLocation+"/metadata/"+relManifestName] = loadFixture(t, "clickhouse_rel", relManifestName) + + rawUpdates := addSnapshotUpdateJSON(t, "wprobe/target_rel/metadata/"+relListName) + repaired, changed := repairAddSnapshotUpdates(context.Background(), store, relTableLocation, rawUpdates) + if !changed { + t.Fatal("expected repair to change the update") + } + + var update struct { + Snapshot struct { + ManifestList string `json:"manifest-list"` + SnapshotID int64 `json:"snapshot-id"` + Summary struct { + Operation string `json:"operation"` + } `json:"summary"` + } `json:"snapshot"` + } + if err := json.Unmarshal(repaired[0], &update); err != nil { + t.Fatalf("unmarshal repaired update: %v", err) + } + wantList := relTableLocation + "/metadata/" + repairedManifestPrefix + relListName + if update.Snapshot.ManifestList != wantList { + t.Fatalf("manifest-list = %s, want %s", update.Snapshot.ManifestList, wantList) + } + // The rest of the snapshot survives the rewrite. + if update.Snapshot.SnapshotID != 449670589 || update.Snapshot.Summary.Operation != "append" { + t.Fatalf("snapshot fields were disturbed: %s", repaired[0]) + } + + verifyRepairedManifests(t, store, wantList, relTableLocation) +} + +func TestRepairClickHouseAbsoluteManifests(t *testing.T) { + store := newFakeManifestStore() + listLocation := absTableLocation + "/metadata/" + absListName + store.files[listLocation] = loadFixture(t, "clickhouse_abs", absListName) + store.files[absTableLocation+"/metadata/"+absManifestName] = loadFixture(t, "clickhouse_abs", absManifestName) + + repaired, changed, err := repairManifestList(context.Background(), store, absTableLocation, listLocation) + if err != nil { + t.Fatalf("repairManifestList: %v", err) + } + if !changed { + t.Fatal("expected repair for missing field-ids") + } + verifyRepairedManifests(t, store, repaired, absTableLocation) +} + +func TestRepairedManifestsAreLeftAlone(t *testing.T) { + store := newFakeManifestStore() + listLocation := absTableLocation + "/metadata/" + absListName + store.files[listLocation] = loadFixture(t, "clickhouse_abs", absListName) + store.files[absTableLocation+"/metadata/"+absManifestName] = loadFixture(t, "clickhouse_abs", absManifestName) + + firstPass, changed, err := repairManifestList(context.Background(), store, absTableLocation, listLocation) + if err != nil || !changed { + t.Fatalf("first pass: changed=%v err=%v", changed, err) + } + + store.saved = map[string][]byte{} + secondPass, changed, err := repairManifestList(context.Background(), store, absTableLocation, firstPass) + if err != nil { + t.Fatalf("second pass: %v", err) + } + if changed || secondPass != firstPass { + t.Fatalf("second pass rewrote an already-repaired list: changed=%v location=%s", changed, secondPass) + } + if len(store.saved) != 0 { + t.Fatalf("second pass saved files: %v", store.saved) + } +} + +func TestRepairKeepsDeleteManifestContent(t *testing.T) { + // The fixture lacks the OCF "content" key; the repaired copy must take the + // content declared by the manifest-list entry, not default to data. + raw := loadFixture(t, "clickhouse_rel", relManifestName) + repaired, changed, err := repairManifest(raw, relTableLocation, 1) + if err != nil || !changed { + t.Fatalf("repairManifest: changed=%v err=%v", changed, err) + } + reader, err := goavro.NewOCFReader(bytes.NewReader(repaired)) + if err != nil { + t.Fatalf("read repaired manifest: %v", err) + } + if got := string(reader.MetaData()["content"]); got != "deletes" { + t.Fatalf(`OCF content = %q, want "deletes"`, got) + } + + asData, _, err := repairManifest(raw, relTableLocation, 0) + if err != nil { + t.Fatalf("repairManifest: %v", err) + } + reader, err = goavro.NewOCFReader(bytes.NewReader(asData)) + if err != nil { + t.Fatalf("read repaired manifest: %v", err) + } + if got := string(reader.MetaData()["content"]); got != "data" { + t.Fatalf(`OCF content = %q, want "data"`, got) + } + + // An existing content value that contradicts the manifest-list entry is + // corrected, not preserved. + relabeled, changed, err := repairManifest(asData, relTableLocation, 1) + if err != nil || !changed { + t.Fatalf("repairManifest on mismatched content: changed=%v err=%v", changed, err) + } + reader, err = goavro.NewOCFReader(bytes.NewReader(relabeled)) + if err != nil { + t.Fatalf("read relabeled manifest: %v", err) + } + if got := string(reader.MetaData()["content"]); got != "deletes" { + t.Fatalf(`OCF content after relabel = %q, want "deletes"`, got) + } +} + +func TestManifestContentValue(t *testing.T) { + cases := []struct { + record map[string]any + want int + }{ + {map[string]any{"content": int32(1)}, 1}, + {map[string]any{"content": int64(1)}, 1}, + {map[string]any{"content": map[string]any{"int": int32(1)}}, 1}, + {map[string]any{"content": nil}, 0}, + {map[string]any{}, 0}, + } + for _, c := range cases { + if got := manifestContentValue(c.record); got != c.want { + t.Errorf("manifestContentValue(%v) = %d, want %d", c.record, got, c.want) + } + } +} + +func TestRepairSkipsLocationsOutsideMetadataDir(t *testing.T) { + store := newFakeManifestStore() + rawUpdates := addSnapshotUpdateJSON(t, "s3://other-bucket/elsewhere/list.avro") + _, changed := repairAddSnapshotUpdates(context.Background(), store, relTableLocation, rawUpdates) + if changed { + t.Fatal("repair touched a manifest list outside the table metadata directory") + } +} + +func TestRepairLeavesUnreadableListAlone(t *testing.T) { + store := newFakeManifestStore() + store.files[relTableLocation+"/metadata/junk.avro"] = []byte("not avro") + rawUpdates := addSnapshotUpdateJSON(t, relTableLocation+"/metadata/junk.avro") + repaired, changed := repairAddSnapshotUpdates(context.Background(), store, relTableLocation, rawUpdates) + if changed { + t.Fatal("repair claimed to change an unreadable manifest list") + } + if !bytes.Equal(repaired[0], rawUpdates[0]) { + t.Fatal("update was modified despite repair failure") + } +} diff --git a/weed/s3api/iceberg/metadata_files.go b/weed/s3api/iceberg/metadata_files.go index e18189040..1e6223f0a 100644 --- a/weed/s3api/iceberg/metadata_files.go +++ b/weed/s3api/iceberg/metadata_files.go @@ -12,9 +12,23 @@ import ( "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" ) +// metadataDirPath returns the filer directory holding a table's metadata files. +func metadataDirPath(bucketName, tablePath string) string { + dir := path.Join(s3tables.TablesPath, bucketName) + if tablePath != "" { + dir = path.Join(dir, tablePath) + } + return path.Join(dir, "metadata") +} + // saveMetadataFile saves the Iceberg metadata JSON file to the filer. // It constructs the filer path from the S3 location components. func (s *Server) saveMetadataFile(ctx context.Context, bucketName, tablePath, metadataFileName string, content []byte) error { + return s.saveMetadataBlob(ctx, bucketName, tablePath, metadataFileName, content, "application/json") +} + +// saveMetadataBlob saves a file into the table's metadata directory. +func (s *Server) saveMetadataBlob(ctx context.Context, bucketName, tablePath, metadataFileName string, content []byte, mimeType string) error { // Create context with timeout for file operations opCtx, cancel := context.WithTimeout(ctx, 30*time.Second) @@ -97,7 +111,7 @@ func (s *Server) saveMetadataFile(ctx context.Context, bucketName, tablePath, me }, Content: content, Extended: map[string][]byte{ - "Mime-Type": []byte("application/json"), + "Mime-Type": []byte(mimeType), }, }, }) @@ -118,11 +132,7 @@ func (s *Server) deleteMetadataFile(ctx context.Context, bucketName, tablePath, opCtx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() - metadataDir := path.Join(s3tables.TablesPath, bucketName) - if tablePath != "" { - metadataDir = path.Join(metadataDir, tablePath) - } - metadataDir = path.Join(metadataDir, "metadata") + metadataDir := metadataDirPath(bucketName, tablePath) return s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { return filer_pb.DoRemove(opCtx, client, metadataDir, metadataFileName, true, false, true, false, nil) }) @@ -132,11 +142,7 @@ func (s *Server) loadMetadataFile(ctx context.Context, bucketName, tablePath, me opCtx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() - metadataDir := path.Join(s3tables.TablesPath, bucketName) - if tablePath != "" { - metadataDir = path.Join(metadataDir, tablePath) - } - metadataDir = path.Join(metadataDir, "metadata") + metadataDir := metadataDirPath(bucketName, tablePath) var content []byte err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { diff --git a/weed/s3api/iceberg/name_mapping.go b/weed/s3api/iceberg/name_mapping.go new file mode 100644 index 000000000..f89f6aa5c --- /dev/null +++ b/weed/s3api/iceberg/name_mapping.go @@ -0,0 +1,142 @@ +package iceberg + +import ( + "encoding/json" + "slices" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" +) + +// nameMappingFromSchema derives the spec's field-id-to-names mapping from a +// table schema, mirroring Java's MappingUtil.create. List elements and map +// keys/values use the canonical "element"/"key"/"value" names. +func nameMappingFromSchema(schema *iceberg.Schema) iceberg.NameMapping { + return mappedFieldsOf(schema.Fields()) +} + +func mappedFieldsOf(fields []iceberg.NestedField) []iceberg.MappedField { + mapped := make([]iceberg.MappedField, 0, len(fields)) + for _, field := range fields { + id := field.ID + mapped = append(mapped, iceberg.MappedField{ + Names: []string{field.Name}, + FieldID: &id, + Fields: mappedFieldsOfType(field.Type), + }) + } + return mapped +} + +func mappedFieldsOfType(typ iceberg.Type) []iceberg.MappedField { + switch t := typ.(type) { + case *iceberg.StructType: + return mappedFieldsOf(t.FieldList) + case *iceberg.ListType: + return mappedFieldsOf([]iceberg.NestedField{t.ElementField()}) + case *iceberg.MapType: + return mappedFieldsOf([]iceberg.NestedField{t.KeyField(), t.ValueField()}) + default: + return nil + } +} + +// refreshDefaultNameMapping keeps schema.name-mapping.default in sync with +// the current schema across commits, so schema evolution does not strand +// readers of field-id-less data files on the creation-time mapping. The +// stored mapping is merged rather than replaced: per field-id, names already +// present are kept alongside the current schema name, so files written under +// a renamed column's old name (and user-added aliases) stay resolvable. +// Returns the (possibly patched) serialized metadata. +func refreshDefaultNameMapping(raw []byte, updated table.Metadata) []byte { + if updated == nil { + return raw + } + updatedSchema := updated.CurrentSchema() + if updatedSchema == nil || len(updatedSchema.Fields()) == 0 { + return raw + } + mapping := nameMappingFromSchema(updatedSchema) + + existing := updated.Properties()[table.DefaultNameMappingKey] + if existing != "" { + var existingMapping iceberg.NameMapping + if err := json.Unmarshal([]byte(existing), &existingMapping); err != nil { + // Not a mapping we can merge; leave whatever the user stored. + return raw + } + mapping = mergeMappedFields(existingMapping, mapping) + } + wantJSON, err := json.Marshal(mapping) + if err != nil { + return raw + } + want := string(wantJSON) + if existing == want { + return raw + } + + var metadata map[string]json.RawMessage + if err := json.Unmarshal(raw, &metadata); err != nil { + return raw + } + properties := map[string]json.RawMessage{} + if rawProperties, ok := metadata["properties"]; ok { + if err := json.Unmarshal(rawProperties, &properties); err != nil { + return raw + } + } + valueJSON, err := json.Marshal(want) + if err != nil { + return raw + } + properties[table.DefaultNameMappingKey] = valueJSON + propertiesJSON, err := json.Marshal(properties) + if err != nil { + return raw + } + metadata["properties"] = propertiesJSON + patched, err := json.Marshal(metadata) + if err != nil { + return raw + } + return patched +} + +// mergeMappedFields folds the names of an existing mapping into the mapping +// derived from the current schema. Fields are matched by field-id per nesting +// level; entries whose ids left the schema are dropped, since current readers +// cannot project those fields anyway. A name can belong to only one field per +// level (Java readers reject ambiguous mappings), so a historical name that +// the current schema assigns to a different field-id stays with its new owner +// instead of being duplicated onto the old one — which also keeps mappings +// clean when a commit replaces the schema with reassigned field ids. +func mergeMappedFields(existing, derived []iceberg.MappedField) []iceberg.MappedField { + byID := make(map[int]*iceberg.MappedField, len(existing)) + for i := range existing { + byID[existing[i].ID()] = &existing[i] + } + nameOwner := make(map[string]int) + for _, d := range derived { + for _, name := range d.Names { + nameOwner[name] = d.ID() + } + } + for i := range derived { + prior, ok := byID[derived[i].ID()] + if !ok { + continue + } + for _, name := range prior.Names { + if ownerID, taken := nameOwner[name]; taken && ownerID != derived[i].ID() { + continue + } + if !slices.Contains(derived[i].Names, name) { + derived[i].Names = append(derived[i].Names, name) + nameOwner[name] = derived[i].ID() + } + } + derived[i].Fields = mergeMappedFields(prior.Fields, derived[i].Fields) + } + return derived +} diff --git a/weed/s3api/iceberg/name_mapping_test.go b/weed/s3api/iceberg/name_mapping_test.go new file mode 100644 index 000000000..b3555f42f --- /dev/null +++ b/weed/s3api/iceberg/name_mapping_test.go @@ -0,0 +1,213 @@ +package iceberg + +import ( + "encoding/json" + "slices" + "testing" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/google/uuid" +) + +func TestNameMappingFromSchema(t *testing.T) { + schema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 2, Name: "label", Type: iceberg.PrimitiveTypes.String}, + iceberg.NestedField{ID: 3, Name: "tags", Type: &iceberg.ListType{ + ElementID: 4, Element: iceberg.PrimitiveTypes.String, + }}, + iceberg.NestedField{ID: 5, Name: "attrs", Type: &iceberg.MapType{ + KeyID: 6, KeyType: iceberg.PrimitiveTypes.String, + ValueID: 7, ValueType: &iceberg.StructType{FieldList: []iceberg.NestedField{ + {ID: 8, Name: "inner", Type: iceberg.PrimitiveTypes.Int32}, + }}, + }}, + ) + + mapping := nameMappingFromSchema(schema) + got, err := json.Marshal(mapping) + if err != nil { + t.Fatalf("marshal mapping: %v", err) + } + + want := `[{"names":["id"],"field-id":1},{"names":["label"],"field-id":2},` + + `{"names":["tags"],"field-id":3,"fields":[{"names":["element"],"field-id":4}]},` + + `{"names":["attrs"],"field-id":5,"fields":[{"names":["key"],"field-id":6},` + + `{"names":["value"],"field-id":7,"fields":[{"names":["inner"],"field-id":8}]}]}]` + if string(got) != want { + t.Fatalf("mapping = %s\nwant %s", got, want) + } +} + +func TestRefreshDefaultNameMappingDropsReassignedNames(t *testing.T) { + // The stored mapping was derived from a schema whose ids were later + // reassigned (id<->name swapped). The merged mapping must follow the + // current schema exactly: duplicating names across ids makes Java readers + // reject the whole mapping. + schema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 2, Name: "name", Type: iceberg.PrimitiveTypes.String}, + ) + metadata, err := newTableMetadata(uuid.New(), "s3://bucket/ns/tbl", schema, nil, nil, + iceberg.Properties{table.DefaultNameMappingKey: `[{"names":["name"],"field-id":1},{"names":["id"],"field-id":2}]`}) + if err != nil { + t.Fatalf("newTableMetadata: %v", err) + } + raw, err := json.Marshal(metadata) + if err != nil { + t.Fatalf("marshal metadata: %v", err) + } + + refreshed := refreshDefaultNameMapping(raw, metadata) + final, err := table.ParseMetadataBytes(refreshed) + if err != nil { + t.Fatalf("parse refreshed metadata: %v", err) + } + var mapping iceberg.NameMapping + if err := json.Unmarshal([]byte(final.Properties()[table.DefaultNameMappingKey]), &mapping); err != nil { + t.Fatalf("parse refreshed mapping: %v", err) + } + seen := map[string]int{} + for _, field := range mapping { + for _, name := range field.Names { + if prior, dup := seen[name]; dup { + t.Fatalf("name %q mapped to both field %d and field %d: %s", + name, prior, field.ID(), final.Properties()[table.DefaultNameMappingKey]) + } + seen[name] = field.ID() + } + } + if seen["id"] != 1 || seen["name"] != 2 { + t.Fatalf("current schema names must win: %s", final.Properties()[table.DefaultNameMappingKey]) + } +} + +// evolveMetadataSchema returns serialized metadata whose schema renamed +// "label" to "tag" and gained an "extra" column while the properties still +// carry the creation-time mapping, simulating a schema-evolution commit. +func evolveMetadataSchema(t *testing.T, base table.Metadata) []byte { + t.Helper() + + raw, err := json.Marshal(base) + if err != nil { + t.Fatalf("marshal base metadata: %v", err) + } + evolved := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 2, Name: "tag", Type: iceberg.PrimitiveTypes.String}, + iceberg.NestedField{ID: 3, Name: "extra", Type: iceberg.PrimitiveTypes.String}, + ) + schemaJSON, err := json.Marshal(evolved) + if err != nil { + t.Fatalf("marshal evolved schema: %v", err) + } + var metadata map[string]json.RawMessage + if err := json.Unmarshal(raw, &metadata); err != nil { + t.Fatalf("unmarshal metadata: %v", err) + } + metadata["schemas"] = json.RawMessage("[" + string(schemaJSON) + "]") + edited, err := json.Marshal(metadata) + if err != nil { + t.Fatalf("marshal edited metadata: %v", err) + } + return edited +} + +func TestRefreshDefaultNameMappingOnSchemaEvolution(t *testing.T) { + schema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 2, Name: "label", Type: iceberg.PrimitiveTypes.String}, + ) + base, err := newTableMetadata(uuid.New(), "s3://bucket/ns/tbl", schema, nil, nil, nil) + if err != nil { + t.Fatalf("newTableMetadata: %v", err) + } + + edited := evolveMetadataSchema(t, base) + updated, err := table.ParseMetadataBytes(edited) + if err != nil { + t.Fatalf("parse evolved metadata: %v", err) + } + + refreshed := refreshDefaultNameMapping(edited, updated) + final, err := table.ParseMetadataBytes(refreshed) + if err != nil { + t.Fatalf("parse refreshed metadata: %v", err) + } + var mapping iceberg.NameMapping + if err := json.Unmarshal([]byte(final.Properties()[table.DefaultNameMappingKey]), &mapping); err != nil { + t.Fatalf("parse refreshed mapping: %v", err) + } + if len(mapping) != 3 || mapping[2].ID() != 3 || mapping[2].Names[0] != "extra" { + t.Fatalf("mapping was not refreshed for the evolved schema: %s", final.Properties()[table.DefaultNameMappingKey]) + } + // The renamed column keeps its historical name so files written under the + // old physical name stay resolvable. + if mapping[1].ID() != 2 || !slices.Contains(mapping[1].Names, "tag") || !slices.Contains(mapping[1].Names, "label") { + t.Fatalf("renamed column lost a name: %s", final.Properties()[table.DefaultNameMappingKey]) + } +} + +func TestRefreshDefaultNameMappingKeepsUserNames(t *testing.T) { + schema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 2, Name: "label", Type: iceberg.PrimitiveTypes.String}, + ) + base, err := newTableMetadata(uuid.New(), "s3://bucket/ns/tbl", schema, nil, nil, + iceberg.Properties{table.DefaultNameMappingKey: `[{"names":["custom"],"field-id":1}]`}) + if err != nil { + t.Fatalf("newTableMetadata: %v", err) + } + + edited := evolveMetadataSchema(t, base) + updated, err := table.ParseMetadataBytes(edited) + if err != nil { + t.Fatalf("parse evolved metadata: %v", err) + } + + refreshed := refreshDefaultNameMapping(edited, updated) + final, err := table.ParseMetadataBytes(refreshed) + if err != nil { + t.Fatalf("parse refreshed metadata: %v", err) + } + var mapping iceberg.NameMapping + if err := json.Unmarshal([]byte(final.Properties()[table.DefaultNameMappingKey]), &mapping); err != nil { + t.Fatalf("parse refreshed mapping: %v", err) + } + // The user's alias survives the merge alongside the schema name. + if mapping[0].ID() != 1 || !slices.Contains(mapping[0].Names, "custom") || !slices.Contains(mapping[0].Names, "id") { + t.Fatalf("user alias was dropped: %s", final.Properties()[table.DefaultNameMappingKey]) + } +} + +func TestCreateTableSetsDefaultNameMapping(t *testing.T) { + // Request ids are deliberately non-sequential: the metadata constructor + // reassigns fresh ids, and the stamped mapping must follow the final + // schema, not the request. + schema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 7, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 3, Name: "label", Type: iceberg.PrimitiveTypes.String}, + ) + metadata, err := newTableMetadata(uuid.New(), "s3://bucket/ns/tbl", schema, nil, nil, nil) + if err != nil { + t.Fatalf("newTableMetadata: %v", err) + } + mappingJSON, ok := metadata.Properties()[table.DefaultNameMappingKey] + if !ok { + t.Fatalf("newTableMetadata did not set %s: %v", table.DefaultNameMappingKey, metadata.Properties()) + } + var mapping iceberg.NameMapping + if err := json.Unmarshal([]byte(mappingJSON), &mapping); err != nil { + t.Fatalf("parse stamped mapping: %v", err) + } + finalSchema := metadata.CurrentSchema() + if len(mapping) != len(finalSchema.Fields()) { + t.Fatalf("mapping has %d entries, schema has %d", len(mapping), len(finalSchema.Fields())) + } + for i, field := range finalSchema.Fields() { + if mapping[i].ID() != field.ID || mapping[i].Names[0] != field.Name { + t.Fatalf("mapping entry %d = %v, want id %d name %s", i, mapping[i], field.ID, field.Name) + } + } +} diff --git a/weed/s3api/iceberg/testdata/clickhouse_abs/b317f0e5-10f3-41d6-86bc-8b42f01f3e86.avro b/weed/s3api/iceberg/testdata/clickhouse_abs/b317f0e5-10f3-41d6-86bc-8b42f01f3e86.avro new file mode 100644 index 0000000000000000000000000000000000000000..74e25e41514022b0375e0303a1d747e4d17fccbf GIT binary patch literal 2845 zcmcgu&2G~`5KgUvP*fof6{j8+U%150U)%gVKqVy9A}Cy{$XYw=WWn*8-F2c+kotP{scQ>j%+eMs#2+nH}>cIMc7yRo%Rhf?hNqRadTQ4)ra z&A0OVY(W1!IkWXRW_H__%onn2m+XiR@OwZb9x$a{7HK&KdcaxORd)OIY$l>K)rkVu z)C7;TJt@uE5sg(}XqTheROZx17vM7#QP1w=rP=$0MLu(*WZ*G*4UK?@%vCp5>rzc! z)Xk2#5(8NSnagqeBGQ0Lrf*~5CA%wp!0iO!BqAVFA)>U9tV_Hx(S1hnaAFS6g>C|g zsqW7$DlItxG&A~K_*8Qdk=HMukrCH@a`=jv6tXM7XR^1aPK0BS`~v>sL<@Xx@{678r?|@^(MqnZ>=Q}p%^jgdeES7)6XjMKQ}_{ z0CUi|t^d;|Uh0G-j@KUmc0&Euov=Cps6uzZG{r%2DFb$@lW%<$hg`cN2o%%TNZItB z%C3?5w1zoK!PFS0^$qNpk-y-Nkl>E&LKaqyeU$H{*Vj%ugq$Ih3TBH{+sWpPqheTA zPK(8kozYyRflaHjmF_MV_s)t0p4oj$+T_#_keIUY4Idj^0d}>N@eS2%IzUJE5=#Oc z9OgEvX_>Gkwz;ZN(~O$T>SQjq(t+BXzvw_nJqBA*X2@{!J_IyWS2{7B@R|0+ymeA@ z5FG3(9EOf`%;a$}!QeZ6@P) ze_KB+%liEH(>M6Dtism~Rd1Ke+-I=F^&CyTkg4)0mcnCYO=XYiGS;^FarV6ZpvCHS zr_!ucoW1?My3=gXmeZt_M&qDW4T7M%JKabgzDPHd&5JE-(<*Eh@PW5SMe|lDY~F{h S%uu#xlr24i;K$~3+3|lI7JB9Y literal 0 HcmV?d00001 diff --git a/weed/s3api/iceberg/testdata/clickhouse_abs/snap-331611856-2-51798e6c-427b-4fc2-9149-99e75d96e045.avro b/weed/s3api/iceberg/testdata/clickhouse_abs/snap-331611856-2-51798e6c-427b-4fc2-9149-99e75d96e045.avro new file mode 100644 index 0000000000000000000000000000000000000000..5bb04baf91535f0da7e08762d64a9498ed031f83 GIT binary patch literal 1591 zcma)6L2lDP6iqFJH~_F>@#PJhq)D1KM3K5cv`CdID$&BGs;sg79S<7Mn3-`*sB#Z> zutens2#G5ovET^oxBwew>^OCrq(NTU^ZfpMfBt_b{+q44_sLj_?LdSW+~XNz4~usd z#F&sj4=!CjNzom;5(6Pax9M_{*!P5R8llo&L>ZdT5k(d%cX)oegh+{w4Pr*5w4{7w z-a`=>o?{UZO$CQnFOMLnI)>+`P*g{E+FT}M$VYWLCTe77LK6m75h(f|fpYLC8daSs zB~nvc(Nie~9<`ok(kRz+U8m*{jt(GG7=jj%OR5(X1%hiVepvDMv+_`Zh(N~(Rkw!F z*$;^(P3Z22$nb_Xg>N(D0eU=3e3aM97F?drP*9>EkC87_ft)BHVTfVfctVZk|0-}S za&xm#fvY;En{_XB?xL8dI7wA3^g>e?rOyzyvpOcaRJnoH-ly^ck*(|YQ)?B>WylGX zu-fF{JF2vqxMEOmr8%4-*D;kB7H%dwC~kJGqOw_1 z$WwV*O*3CAHJ{mY*QSYUR(Q4_L`pI-$7YWb6BP6BHru?@j;%P8%C{?aL@tn^p#{(3mK7jZ(foUq5EX_EiozNMZ}GvK&k;PIJaPG7a}K_48c@)2-ZV| zNX77n#8g2d<`M}cE%5l@Pp*#|dhL~v6 zCo#~3<^$-BTz0XWj$OCu4PCbdvX=|$n4{h0T{NU&IxE$zZ^i&?v3HD2hU@D5Z-f7} z>Pq~_JHfk)AZjiKF;}7ab#>b$L^QgS7uaQJa!=*g z$ZT4}9Hn4t1k?HkcFf41^G7OhNBNM2Wn&-ZyV>>iG#f(BkgEh{i&cA?&lx9);jwa9 zE_dym<{=GiTGfs0?Q(Jdv`pZe-lwEP4h;c`84FkV+29JW-Ac|kRP)jSI2lrON&RgIcv)Z|vjGqI%()Mos72SVE8@G8m;8E(;sfQHGHPE00zu01xZZe|XG zgSU!+p`#o#c^phI_zv*mF;lS&RA3^nI?>JJgd?ee(vRqg$>&Klp(_8ZUN|M>FD zI=3wA@3-IQ_u$_ueO^0>mF)3tP1OGwQiNxOZdU+aoMa&rS%8!3Nw_=DP=>SK(HgTU3Byx DVmNlh literal 0 HcmV?d00001 diff --git a/weed/s3api/iceberg/testdata/clickhouse_rel/snap-449670589-2-48b874af-90f6-4e5e-8002-df91e59c8a0c.avro b/weed/s3api/iceberg/testdata/clickhouse_rel/snap-449670589-2-48b874af-90f6-4e5e-8002-df91e59c8a0c.avro new file mode 100644 index 0000000000000000000000000000000000000000..9ee8cee5b9836346758db521c004f4857ff7a81d GIT binary patch literal 1575 zcma)6O>Wab6iy978~|9cc=Co#(j=u#lS*A63R0ztN~Ex5jX^?|c1e(1kU;;)HL%PnC zqB6`y40K9iz?`S4G$zbU*U(oXg=91tkjdn8aKZ$Wa4E5k+<53A2{}E5M%2 zZDpYXS8;4M>t5>IMX*2%Om!^HTvF$yPZ2J&CPq`LTte&Wse@2v*Jb;OvkJBtY62y! z);aW!X=6LCXw(~No=uV0F_i_2mJ{t1mAzI`nJgvq1vPWIDqT6)4W^c|%uHQPRC!v; zQ+c|YW;RrM*0YDc+a|sp;pu#!QaG{Kc8(IeC}#gJZx*%D>08+1LU9v^OlUhDj=16- zNy!O?{ZVd|1C`my9|MX>xe8)84Yr9nhnY?|L7pwlbID$Cm7*&Xeih`X)>+(+i3&!x zFE4h;#e-t+)Ztm0E{7kTogLZl;aI8!P3we|wyVlpYkXEOetPG<@8|oyN3VZ;{N;UJ z`LL3wN{*;)P#u#A6!CUK2Hk>b_XeH8MjzK&gTdooYaKi5t$w$=*^2tz&U!f5-0TiE aR^6)q)NC|7@9($YUv7C`<94I55b7V(3K=5+ literal 0 HcmV?d00001