diff --git a/docs/benchmark-results/iceberg.md b/docs/benchmark-results/iceberg.md index d0f4e23ad3..81315a4bfe 100644 --- a/docs/benchmark-results/iceberg.md +++ b/docs/benchmark-results/iceberg.md @@ -215,6 +215,66 @@ To reproduce: the localhost benchmark configs live under [`internal/impl/iceberg --- +## Shredder Allocations — 2026-08-20 + +Record shredding (JSON `map[string]any` → columnar parquet values) built two maps per struct per record to support case-insensitive key matching. Case-sensitive matching is the default (`case_sensitive_columns: true`) and makes those maps redundant, so it now has a dedicated path that looks fields up directly and skips unknown-field scanning when every input key is accounted for. + +Driven by `BenchmarkShredWide` in [`internal/impl/iceberg/bench/`](../../internal/impl/iceberg/bench/) — a wide-schema shredder micro-benchmark that mirrors the profiling pipeline's record shape without standing up infrastructure. + +**Environment:** darwin/arm64, Apple M3 Pro, `GOMAXPROCS=1`, Go benchmark, `benchstat` over n=8 + +**Changed since last run:** the case-sensitive shredding path ([#4712](https://github.com/redpanda-data/connect/pull/4712)). No configuration or behaviour change. + +| metric | before | after | delta | +|-----------|---------|---------|-------------------| +| sec/op | 4.369µs | 1.472µs | **-66.3%** (p=0.000) | +| B/op | 4.312 KiB | 1.609 KiB | **-62.7%** (p=0.000) | +| allocs/op | 71 | 41 | **-42.3%** (p=0.000) | + +Per sub-benchmark, sec/op: `declared_schema=false` 4.304µs → 1.394µs (-67.6%); `declared_schema=true` 4.435µs → 1.555µs (-64.9%). + +**Observations:** + +- **This is the shredder in isolation, not a sink-level number.** Earlier 1-vCPU profiling attributed ~27% of the sink's CPU to shredding, so the end-to-end effect should be appreciable but much smaller than 66%. **It has not been measured end to end** — no throughput figure above or elsewhere in this file has been re-run for this change. +- The two `declared_schema` variants are within noise of each other both before and after, consistent with the earlier finding that the `schema_metadata` knob does not bypass decode, shredding or encode. + +To reproduce: `GOMAXPROCS=1 go test -bench BenchmarkShredWide -benchmem -run '^$' -count=8 ./internal/impl/iceberg/bench/` + +--- + +## Commit Regime — Commit Latency vs `max_in_flight` (synthetic) + +How commit coalescing responds to catalog commit latency and the number of concurrent in-flight submissions, measured by the flag-gated `TestCommitRegimeSweep` in [`internal/impl/iceberg/commit_regime_bench_test.go`](../../internal/impl/iceberg/commit_regime_bench_test.go). + +**Environment:** darwin/arm64, Apple M3 Pro; in-memory catalog with a fixed injected per-commit delay; 6s window per point; 300 records per submission + +**Caveat — read the numbers as ratios, not throughput.** Nothing here writes parquet or touches object storage, and the injected delay is not a real catalog, so the absolute rec/sec are not sink throughput figures and are not comparable with the localhost or live-catalog sections above. What the harness measures is how many submissions a commit carries, and at what latency. + +| commit latency | `max_in_flight` | rec/sec | records/commit | submissions/commit | +|---------------:|----------------:|--------:|---------------:|-------------------:| +| 50ms | 1 | 5,238 | 300 | 1.00 | +| 50ms | 4 | 10,437 | 600 | 2.00 | +| 50ms | 16 | 41,790 | 2,400 | 8.00 | +| 50ms | 64 | 166,306 | 9,600 | 32.00 | +| 200ms | 1 | 1,449 | 300 | 1.00 | +| 200ms | 4 | 2,896 | 600 | 2.00 | +| 200ms | 16 | 11,563 | 2,400 | 8.00 | +| 200ms | 64 | 46,230 | 9,600 | 32.00 | +| 500ms | 1 | 591 | 300 | 1.00 | +| 500ms | 4 | 1,187 | 600 | 2.00 | +| 500ms | 16 | 4,416 | 2,238 | 7.46 | +| 500ms | 64 | 20,354 | 10,338 | 34.46 | + +**Observations:** + +- **The commit batcher already coalesces concurrent submissions.** Submissions that arrive while a commit is in flight are merged into the next one, so records per commit scales with `max_in_flight` without any time-based batching involved. +- **At `max_in_flight: 1` records per commit is pinned to a single submission**, giving `records-per-submission / commit-latency` — 591 rec/sec at 500ms, matching the "throughput trap" regime described under Tuning Recipes. This is structural: the sole submitter is blocked inside the commit it is waiting on, so no second submission can exist to batch with. A commit-side linger cannot improve this case, and would add latency to it. +- Submissions per commit settles near `max_in_flight / 2` rather than `max_in_flight`, which suggests the batcher samples its queue before the just-released submitters have all re-queued. Whether closing that gap is worth anything is untested. + +To reproduce: `go test -run TestCommitRegimeSweep -iceberg.commit-regime -timeout 20m ./internal/impl/iceberg/` (add `-iceberg.commit-regime-realistic` for 320ms/5s/10s latencies). + +--- + ## Tuning Recipes The single most important factor for `iceberg` throughput is **records per commit**. Each catalog diff --git a/internal/impl/iceberg/bench/README.md b/internal/impl/iceberg/bench/README.md index 69c24db2c7..3f75c6badc 100644 --- a/internal/impl/iceberg/bench/README.md +++ b/internal/impl/iceberg/bench/README.md @@ -47,6 +47,36 @@ task bench:mif CORES=4 BATCH=10000 MIF=32 COUNT=1000000 |-----------|---------|-------------| | `MIF` | 4 | `max_in_flight` | +### Profiling (per-record CPU) + +`profile_config.yaml` is a profiling variant of `benchmark_config.yaml`: a ~1.2 kB +high-entropy JSON payload serialised to raw bytes in the pipeline, so the Iceberg +output performs a real JSON parse per record and the profile attributes decode, +shredding and encode separately. `profile_config_schema.yaml` is the same +pipeline with a declared schema, for measuring what `schema_metadata` buys. + +```bash +task bench:profile CORES=1 COUNT=500000 # schemaless +task bench:profile:schema CORES=1 COUNT=500000 # declared schema +``` + +| Parameter | Default | Description | +|-----------|---------|-------------| +| `CORES` | 1 | `GOMAXPROCS` — 1 isolates per-record CPU cost | +| `BATCH` | 5000 | `batching.count` | +| `COUNT` | 500000 | number of messages | + +### Shredder micro-benchmark + +Needs no infrastructure — it exercises the shredder directly: + +```bash +task bench:shredder # GOMAXPROCS=1, -count=8 +``` + +Results are recorded in +[`docs/benchmark-results/iceberg.md`](../../../../docs/benchmark-results/iceberg.md). + ### Clean run ```bash diff --git a/internal/impl/iceberg/bench/Taskfile.yaml b/internal/impl/iceberg/bench/Taskfile.yaml index 4035bdd07c..5bcc729a9c 100644 --- a/internal/impl/iceberg/bench/Taskfile.yaml +++ b/internal/impl/iceberg/bench/Taskfile.yaml @@ -85,6 +85,52 @@ tasks: --set output.iceberg.storage.aws_s3.credentials.secret={{.MINIO_PASSWORD}} \ ./benchmark_config.yaml + # Profiling run — same pipeline as `bench` but with the ~1.2kB high-entropy + # payload from profile_config.yaml, parsed per record so the sink does real + # JSON decode + shredding + encode work. Writes pprof profiles for attributing + # per-record CPU. + # Usage: task bench:profile CORES=1 BATCH=5000 COUNT=500000 + bench:profile: + desc: "Run the profiling pipeline and write CPU/heap profiles (e.g. task bench:profile CORES=1 COUNT=500000)" + vars: + CORES: '{{.CORES | default "1"}}' + BATCH: '{{.BATCH | default "5000"}}' + COUNT: '{{.COUNT | default "500000"}}' + CONFIG: '{{.CONFIG | default "./profile_config.yaml"}}' + cmds: + - | + AWS_EC2_METADATA_DISABLED=true \ + AWS_ACCESS_KEY_ID={{.MINIO_USER}} \ + AWS_SECRET_ACCESS_KEY={{.MINIO_PASSWORD}} \ + AWS_REGION={{.MINIO_REGION}} \ + GOMAXPROCS={{.CORES}} go run ../../../../cmd/redpanda-connect/main.go run \ + --set input.generate.count={{.COUNT}} \ + --set output.iceberg.batching.count={{.BATCH}} \ + --set output.iceberg.catalog.url={{.CATALOG_URL}} \ + --set output.iceberg.storage.aws_s3.endpoint={{.MINIO_ENDPOINT}} \ + --set output.iceberg.storage.aws_s3.bucket={{.MINIO_BUCKET}} \ + --set output.iceberg.storage.aws_s3.credentials.id={{.MINIO_USER}} \ + --set output.iceberg.storage.aws_s3.credentials.secret={{.MINIO_PASSWORD}} \ + {{.CONFIG}} + + # Same pipeline with a declared schema, to measure what schema_metadata buys. + # Usage: task bench:profile:schema CORES=1 COUNT=500000 + bench:profile:schema: + desc: "Run the declared-schema profiling pipeline (e.g. task bench:profile:schema CORES=1 COUNT=500000)" + cmds: + - task: bench:profile + vars: + CONFIG: ./profile_config_schema.yaml + CORES: '{{.CORES | default "1"}}' + BATCH: '{{.BATCH | default "5000"}}' + COUNT: '{{.COUNT | default "500000"}}' + + # Shredder micro-benchmark — no infrastructure required. + bench:shredder: + desc: Run the wide-schema shredder micro-benchmark (no infra needed) + cmds: + - GOMAXPROCS={{.CORES | default "1"}} go test -bench BenchmarkShredWide -benchmem -run '^$' -count={{.COUNT | default "8"}} . + bench:lag: desc: Show current consumer lag for the Redpanda Connect Iceberg sink group cmds: diff --git a/internal/impl/iceberg/bench/profile_config.yaml b/internal/impl/iceberg/bench/profile_config.yaml new file mode 100644 index 0000000000..0d990ac177 --- /dev/null +++ b/internal/impl/iceberg/bench/profile_config.yaml @@ -0,0 +1,89 @@ +# Profiling variant of benchmark_config.yaml, for attributing per-record CPU +# cost in the iceberg sink. +# +# Differences from benchmark_config.yaml: +# - ~1.2KB high-entropy JSON payload (uuid-heavy) instead of the ~150B event +# - the payload is serialised to raw bytes in the pipeline (root = content()) +# so the iceberg output performs a real JSON parse per record, matching the +# production Kafka -> iceberg path. Messages produced by `generate` are +# otherwise already structured and AsStructured() would be free. +# - batching.count 10000 so commits are amortised and per-record CPU dominates +# +# pprof endpoints are exposed at http://localhost:4195/debug/pprof/ via +# http.debug_endpoints. Capture with: +# curl -o cpu.pb.gz 'http://localhost:4195/debug/pprof/profile?seconds=60' +# curl -o allocs.pb.gz 'http://localhost:4195/debug/pprof/allocs' +# +# Run at 1 core with GOMAXPROCS=1 (taskset does not exist on darwin). + +http: + debug_endpoints: true + +input: + generate: + count: ${COUNT:0} + interval: "" + mapping: | + root.id = counter() + root.user_id = (counter() % 10000) + 1 + root.session_id = uuid_v4() + root.trace_id = uuid_v4() + root.span_id = uuid_v4() + root.request_id = uuid_v4() + root.device_id = uuid_v4() + root.correlation_id = uuid_v4() + root.event_type = ["click", "view", "purchase", "scroll", "hover"].index(counter() % 5) + root.country = ["US", "GB", "DE", "FR", "JP", "BR", "IN", "AU"].index(counter() % 8) + root.value = random_int(max: 1000000) + root.amount = random_int(max: 10000000) / 100.0 + root.score = random_int(max: 100000) / 1000.0 + root.latency_ms = random_int(max: 30000) + root.is_mobile = counter() % 2 == 0 + root.user_agent = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36" + root.url = "https://shop.example.com/products/" + uuid_v4() + "?ref=" + uuid_v4() + root.referrer = "https://www.google.com/search?q=" + uuid_v4() + root.payload_a = uuid_v4() + ":" + uuid_v4() + root.payload_b = uuid_v4() + ":" + uuid_v4() + root.payload_c = uuid_v4() + ":" + uuid_v4() + root.payload_d = uuid_v4() + ":" + uuid_v4() + root.description = "synthetic high entropy event record number " + counter().string() + " for per-record cpu profiling" + root.ts = now() + +pipeline: + processors: + # Force the message down to raw bytes so the iceberg output pays the JSON + # decode cost (AsStructured), as it does when reading from Kafka. + - mapping: 'root = content()' + - benchmark: + interval: 1s + count_bytes: true + +output: + iceberg: + catalog: + url: "${CATALOG_URL:http://localhost:8181}" + namespace: bench + table: "${TABLE:events_profile}" + max_in_flight: ${MIF:4} + storage: + aws_s3: + bucket: "${MINIO_BUCKET:warehouse}" + region: "${MINIO_REGION:us-east-1}" + endpoint: "${MINIO_ENDPOINT:http://localhost:9000}" + force_path_style_urls: true + credentials: + id: "${MINIO_USER:admin}" + secret: "${MINIO_PASSWORD:password}" + schema_evolution: + enabled: true + batching: + count: 10000 + period: 5s + +logger: + level: INFO + +metrics: + prometheus: + add_process_metrics: true + add_go_metrics: true diff --git a/internal/impl/iceberg/bench/profile_config_schema.yaml b/internal/impl/iceberg/bench/profile_config_schema.yaml new file mode 100644 index 0000000000..b44b639405 --- /dev/null +++ b/internal/impl/iceberg/bench/profile_config_schema.yaml @@ -0,0 +1,115 @@ +# Declared-schema variant of profile_config.yaml, for measuring whether a +# declared schema reduces per-record CPU cost. +# +# Identical workload, but every message carries schema metadata (the common +# schema format shared with parquet_encode's schema_metadata) in the +# `iceberg_schema` metadata field, and the output is configured with +# schema_evolution.schema_metadata: iceberg_schema. This exercises the +# declared-schema path: type resolution at table creation/evolution comes from +# the declared schema, and the shredder receives field schema metadata +# (SetFieldSchemaMetadata) instead of relying purely on dynamic inference. +# +# Declared types mirror what inference produces for the same payload so the +# resulting table is identical to the schemaless run (apples-to-apples). +# +# Run at 1 core with GOMAXPROCS=1. See profile_config.yaml for capture notes. + +http: + debug_endpoints: true + +input: + generate: + count: ${COUNT:0} + interval: "" + mapping: | + root.id = counter() + root.user_id = (counter() % 10000) + 1 + root.session_id = uuid_v4() + root.trace_id = uuid_v4() + root.span_id = uuid_v4() + root.request_id = uuid_v4() + root.device_id = uuid_v4() + root.correlation_id = uuid_v4() + root.event_type = ["click", "view", "purchase", "scroll", "hover"].index(counter() % 5) + root.country = ["US", "GB", "DE", "FR", "JP", "BR", "IN", "AU"].index(counter() % 8) + root.value = random_int(max: 1000000) + root.amount = random_int(max: 10000000) / 100.0 + root.score = random_int(max: 100000) / 1000.0 + root.latency_ms = random_int(max: 30000) + root.is_mobile = counter() % 2 == 0 + root.user_agent = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36" + root.url = "https://shop.example.com/products/" + uuid_v4() + "?ref=" + uuid_v4() + root.referrer = "https://www.google.com/search?q=" + uuid_v4() + root.payload_a = uuid_v4() + ":" + uuid_v4() + root.payload_b = uuid_v4() + ":" + uuid_v4() + root.payload_c = uuid_v4() + ":" + uuid_v4() + root.payload_d = uuid_v4() + ":" + uuid_v4() + root.description = "synthetic high entropy event record number " + counter().string() + " for per-record cpu profiling" + root.ts = now() + meta iceberg_schema = { + "type": "OBJECT", + "children": [ + {"name": "id", "type": "INT64"}, + {"name": "user_id", "type": "INT64"}, + {"name": "session_id", "type": "STRING"}, + {"name": "trace_id", "type": "STRING"}, + {"name": "span_id", "type": "STRING"}, + {"name": "request_id", "type": "STRING"}, + {"name": "device_id", "type": "STRING"}, + {"name": "correlation_id", "type": "STRING"}, + {"name": "event_type", "type": "STRING"}, + {"name": "country", "type": "STRING"}, + {"name": "value", "type": "INT64"}, + {"name": "amount", "type": "FLOAT64"}, + {"name": "score", "type": "FLOAT64"}, + {"name": "latency_ms", "type": "INT64"}, + {"name": "is_mobile", "type": "BOOLEAN"}, + {"name": "user_agent", "type": "STRING"}, + {"name": "url", "type": "STRING"}, + {"name": "referrer", "type": "STRING"}, + {"name": "payload_a", "type": "STRING"}, + {"name": "payload_b", "type": "STRING"}, + {"name": "payload_c", "type": "STRING"}, + {"name": "payload_d", "type": "STRING"}, + {"name": "description", "type": "STRING"}, + {"name": "ts", "type": "STRING"} + ] + } + +pipeline: + processors: + - mapping: 'root = content()' + - benchmark: + interval: 1s + count_bytes: true + +output: + iceberg: + catalog: + url: "${CATALOG_URL:http://localhost:8181}" + namespace: bench + table: "${TABLE:events_profile_schema}" + max_in_flight: ${MIF:4} + storage: + aws_s3: + bucket: "${MINIO_BUCKET:warehouse}" + region: "${MINIO_REGION:us-east-1}" + endpoint: "${MINIO_ENDPOINT:http://localhost:9000}" + force_path_style_urls: true + credentials: + id: "${MINIO_USER:admin}" + secret: "${MINIO_PASSWORD:password}" + schema_evolution: + enabled: true + schema_metadata: iceberg_schema + batching: + count: 10000 + period: 5s + +logger: + level: INFO + +metrics: + prometheus: + add_process_metrics: true + add_go_metrics: true diff --git a/internal/impl/iceberg/bench/shred_wide_bench_test.go b/internal/impl/iceberg/bench/shred_wide_bench_test.go new file mode 100644 index 0000000000..b7908de284 --- /dev/null +++ b/internal/impl/iceberg/bench/shred_wide_bench_test.go @@ -0,0 +1,142 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed as a Redpanda Enterprise file under the Redpanda Community +// License (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md + +// Package bench holds micro-benchmarks that mirror the end-to-end profiling +// workload (profile_config.yaml) so per-record costs can be attributed +// without standing up infrastructure. Part of the iceberg sink per-record CPU +// profiling effort. +package bench + +import ( + "fmt" + "testing" + + "github.com/apache/iceberg-go" + + "github.com/redpanda-data/benthos/v4/public/schema" + + "github.com/redpanda-data/connect/v4/internal/impl/iceberg/icebergx" + "github.com/redpanda-data/connect/v4/internal/impl/iceberg/shredder" +) + +// discardSink mirrors the shredder package's benchmark sink: no work, no +// allocation, so the benchmark isolates the shredder's own per-record cost. +type discardSink struct{} + +func (discardSink) EmitValue(shredder.ShreddedValue) error { return nil } +func (discardSink) OnNewField(icebergx.Path, string, any) {} + +// wideSchema mirrors the 24-column table created by profile_config.yaml +// (~1.2KB high-entropy JSON events). +func wideSchema() *iceberg.Schema { + names := wideFieldNames() + fields := make([]iceberg.NestedField, 0, len(names)) + for i, n := range names { + var typ iceberg.Type + switch n { + case "id", "user_id", "value", "latency_ms": + typ = iceberg.PrimitiveTypes.Int64 + case "amount", "score": + typ = iceberg.PrimitiveTypes.Float64 + case "is_mobile": + typ = iceberg.PrimitiveTypes.Bool + default: + typ = iceberg.PrimitiveTypes.String + } + fields = append(fields, iceberg.NestedField{ID: i + 1, Name: n, Type: typ}) + } + return iceberg.NewSchema(1, fields...) +} + +func wideFieldNames() []string { + return []string{ + "id", "user_id", "session_id", "trace_id", "span_id", "request_id", + "device_id", "correlation_id", "event_type", "country", "value", + "amount", "score", "latency_ms", "is_mobile", "user_agent", "url", + "referrer", "payload_a", "payload_b", "payload_c", "payload_d", + "description", "ts", + } +} + +func wideRecord() map[string]any { + return map[string]any{ + "id": int64(123456), + "user_id": int64(4212), + "session_id": "0d9c9c3e-9df6-4c4f-8a91-2e6f1a9f7f10", + "trace_id": "5c1a67aa-30e7-4a83-9b0e-fd3f9a3f6c1b", + "span_id": "e3b7c5d2-8f14-4a6e-b291-7c8e9d0f1a2b", + "request_id": "9a8b7c6d-5e4f-4a3b-8c2d-1e0f9a8b7c6d", + "device_id": "1f2e3d4c-5b6a-4978-8695-a4b3c2d1e0f9", + "correlation_id": "abcdef01-2345-4678-9abc-def012345678", + "event_type": "purchase", + "country": "GB", + "value": int64(778123), + "amount": 42421.42, + "score": 73.113, + "latency_ms": int64(2231), + "is_mobile": false, + "user_agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36", + "url": "https://shop.example.com/products/0d9c9c3e-9df6-4c4f-8a91-2e6f1a9f7f10?ref=5c1a67aa-30e7-4a83-9b0e-fd3f9a3f6c1b", + "referrer": "https://www.google.com/search?q=e3b7c5d2-8f14-4a6e-b291-7c8e9d0f1a2b", + "payload_a": "9a8b7c6d-5e4f-4a3b-8c2d-1e0f9a8b7c6d:1f2e3d4c-5b6a-4978-8695-a4b3c2d1e0f9", + "payload_b": "0d9c9c3e-9df6-4c4f-8a91-2e6f1a9f7f10:5c1a67aa-30e7-4a83-9b0e-fd3f9a3f6c1b", + "payload_c": "e3b7c5d2-8f14-4a6e-b291-7c8e9d0f1a2b:9a8b7c6d-5e4f-4a3b-8c2d-1e0f9a8b7c6d", + "payload_d": "abcdef01-2345-4678-9abc-def012345678:1f2e3d4c-5b6a-4978-8695-a4b3c2d1e0f9", + "description": "synthetic high entropy event record number 123456 for per-record cpu profiling", + "ts": "2026-08-03T16:10:00.000000000+01:00", + } +} + +// wideFieldCommons builds the per-field schema metadata that +// writer.messagesToParquet installs on the shredder when the output's +// schema_evolution.schema_metadata is configured, declaring the same types +// inference would produce. +func wideFieldCommons(s *iceberg.Schema) map[int]*schema.Common { + byID := make(map[int]*schema.Common) + for _, f := range s.Fields() { + var t schema.CommonType + switch f.Type { + case iceberg.PrimitiveTypes.Int64: + t = schema.Int64 + case iceberg.PrimitiveTypes.Float64: + t = schema.Float64 + case iceberg.PrimitiveTypes.Bool: + t = schema.Boolean + default: + t = schema.String + } + byID[f.ID] = &schema.Common{Name: f.Name, Type: t, Optional: true} + } + return byID +} + +// BenchmarkShredWide measures per-record shred cost for the 24-column +// profiling payload, with and without declared field schema metadata +// (the shredder-side effect of the output's declared-schema path). Run: +// +// GOMAXPROCS=1 go test -bench BenchmarkShredWide -benchmem -run '^$' ./internal/impl/iceberg/bench/ +func BenchmarkShredWide(b *testing.B) { + for _, declared := range []bool{false, true} { + b.Run(fmt.Sprintf("declared_schema=%v", declared), func(b *testing.B) { + rs := shredder.NewRecordShredder(wideSchema(), true) + if declared { + rs.SetFieldSchemaMetadata(wideFieldCommons(wideSchema())) + } + record := wideRecord() + sink := discardSink{} + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if err := rs.Shred(record, sink); err != nil { + b.Fatal(err) + } + } + }) + } +} diff --git a/internal/impl/iceberg/commit_regime_bench_test.go b/internal/impl/iceberg/commit_regime_bench_test.go new file mode 100644 index 0000000000..d11e61650a --- /dev/null +++ b/internal/impl/iceberg/commit_regime_bench_test.go @@ -0,0 +1,294 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed as a Redpanda Enterprise file under the Redpanda Community +// License (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md + +package iceberg + +import ( + "context" + "errors" + "flag" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/google/uuid" + "github.com/stretchr/testify/require" + + "github.com/redpanda-data/benthos/v4/public/service" +) + +// This harness isolates the COMMIT REGIME: how sink throughput responds to +// catalog commit latency, the number of concurrent in-flight submissions +// (max_in_flight), and the records carried per submission. It exists because +// the sink's headline throughput problem is latency-bound rather than +// CPU-bound, and the fix under evaluation (a commit linger) is a property of +// the commit batcher alone. +// +// It deliberately does NOT write parquet or touch object storage: files are +// synthesised metadata-only, so nothing here is confounded by encode or upload +// cost. Per-record CPU is measured separately by the bench/ package. +// +// Why a local latency injection is a faithful stand-in for a live catalog: +// measurement against a live Unity Catalog found commit latency near-flat +// across a 667x range of batch sizes (~5.2s at 300 records/commit rising only +// to ~9.7s at 200,000). Commit latency therefore behaves as a near-constant +// independent of batch size, which is exactly what a fixed injected delay +// models — with the advantage that the delay can be swept across regimes +// (a fast catalog at ~320ms, a slow engine-backed one at 5-10s) instead of +// being pinned to whatever one hosted service happens to do. + +var ( + regimeSweep = flag.Bool("iceberg.commit-regime", false, + "run the commit-regime sweep (takes minutes of wall time; prints a table)") + regimeRealistic = flag.Bool("iceberg.commit-regime-realistic", false, + "sweep at real engine-backed catalog latencies (5s/10s) instead of scaled-down ones") +) + +// latentCatalog wraps memCatalog with a fixed per-commit delay and counts +// commits, standing in for a catalog whose commit path costs real wall time +// (credential vending, metadata write, engine-side validation). +// +// The committer serialises all commits under commitMu, so CommitTable is never +// called concurrently; the counter is atomic only so readers can sample it +// while the sweep runs. +type latentCatalog struct { + *memCatalog + delay time.Duration + commits atomic.Int64 +} + +func (c *latentCatalog) CommitTable(ctx context.Context, ident table.Identifier, reqs []table.Requirement, updates []table.Update) (table.Metadata, string, error) { + if c.delay > 0 { + select { + case <-time.After(c.delay): + case <-ctx.Done(): + return nil, "", ctx.Err() + } + } + c.commits.Add(1) + return c.memCatalog.CommitTable(ctx, ident, reqs, updates) +} + +// regimeParams describes one point in the sweep. +type regimeParams struct { + commitLatency time.Duration // injected per-commit catalog cost + inFlight int // concurrent submitters, i.e. max_in_flight + recordsPerSubmit int // records carried by each submission + window time.Duration // measurement window +} + +// regimeResult is what one point measured. +type regimeResult struct { + params regimeParams + records int64 + submissions int64 + commits int64 + elapsed time.Duration + recordsPerSecond float64 + recordsPerCommit float64 + submitsPerCommit float64 +} + +func (r regimeResult) String() string { + return fmt.Sprintf("latency=%-6v in_flight=%-3d rec/submit=%-7d | rec/s=%-10.0f rec/commit=%-10.0f submits/commit=%-5.2f commits=%d", + r.params.commitLatency, r.params.inFlight, r.params.recordsPerSubmit, + r.recordsPerSecond, r.recordsPerCommit, r.submitsPerCommit, r.commits) +} + +// runRegime drives `inFlight` concurrent submitters against a committer whose +// catalog costs `commitLatency` per commit, for `window` of wall time, and +// reports what got through. Each submitter models one in-flight WriteBatch: +// build files, submit, block until the commit that carries them returns. +func runRegime(tb testing.TB, p regimeParams) regimeResult { + tb.Helper() + ctx := tb.Context() + + tbl, mem := newTestTable(tb) + cat := &latentCatalog{memCatalog: mem, delay: p.commitLatency} + + c, err := NewCommitter(tbl, cat, CommitConfig{ + ManifestMergeEnabled: false, + MaxRetries: 1, + }, func(context.Context) (*table.Table, error) { return cat.snapshot(), nil }, + service.MockResources().Logger()) + require.NoError(tb, err) + defer c.Close() + + var records, submissions atomic.Int64 + deadline := time.Now().Add(p.window) + schemaID := c.currentSchemaID() + + // Submitters cannot assert: require's FailNow is only valid on the test + // goroutine. Each records its first error for the test goroutine to check. + errs := make([]error, p.inFlight) + + var wg sync.WaitGroup + start := time.Now() + for i := range p.inFlight { + wg.Go(func() { + for time.Now().Before(deadline) { + df, err := recordCountDataFile(tbl.Spec(), + fmt.Sprintf("%s/data/%s.parquet", tbl.Location(), uuid.New()), + int64(p.recordsPerSubmit)) + if err != nil { + errs[i] = err + return + } + if err := c.Commit(ctx, CommitInput{Files: []iceberg.DataFile{df}, SchemaID: schemaID}); err != nil { + errs[i] = err + return + } + submissions.Add(1) + records.Add(int64(p.recordsPerSubmit)) + } + }) + } + wg.Wait() + elapsed := time.Since(start) + + // Assert before deriving any rate from a possibly half-completed run. + require.NoError(tb, errors.Join(errs...), "submitter(s) failed") + + commits := cat.commits.Load() + res := regimeResult{ + params: p, + records: records.Load(), + submissions: submissions.Load(), + commits: commits, + elapsed: elapsed, + } + res.recordsPerSecond = float64(res.records) / elapsed.Seconds() + if commits > 0 { + res.recordsPerCommit = float64(res.records) / float64(commits) + res.submitsPerCommit = float64(res.submissions) / float64(commits) + } + return res +} + +// recordCountDataFile is synthDataFile with a caller-chosen record count, so a +// submission can represent a realistic batch rather than a single row. +// +// It returns an error rather than asserting, because the submitters that call it +// run on their own goroutines: require's FailNow is only valid on the goroutine +// running the test, so a failure here has to be carried back and asserted after +// the submitters have been waited on. +func recordCountDataFile(spec iceberg.PartitionSpec, path string, records int64) (iceberg.DataFile, error) { + b, err := iceberg.NewDataFileBuilder( + spec, + iceberg.EntryContentData, + path, + iceberg.ParquetFile, + nil, nil, nil, + records, records*64, + ) + if err != nil { + return nil, err + } + return b.Build(), nil +} + +// TestCommitRegimeSweep characterises throughput across the commit regime. +// Flag-gated: it spends real wall time on purpose. +// +// The question it answers: does the existing batcher already coalesce +// concurrent submissions during a slow commit (in which case max_in_flight is +// the lever and a linger adds nothing), or does each commit carry only one +// submission (in which case a time-based linger is the fix)? +func TestCommitRegimeSweep(t *testing.T) { + if !*regimeSweep { + t.Skip("set -iceberg.commit-regime to run the commit-regime sweep") + } + + latencies := []time.Duration{50 * time.Millisecond, 200 * time.Millisecond, 500 * time.Millisecond} + window := 6 * time.Second + if *regimeRealistic { + latencies = []time.Duration{320 * time.Millisecond, 5 * time.Second, 10 * time.Second} + window = 60 * time.Second + } + + var results []regimeResult + for _, latency := range latencies { + for _, inFlight := range []int{1, 4, 16, 64} { + res := runRegime(t, regimeParams{ + commitLatency: latency, + inFlight: inFlight, + recordsPerSubmit: 300, // the observed "throughput trap" batch size + window: window, + }) + results = append(results, res) + t.Log(res.String()) + } + } + + t.Log("=== commit regime sweep (records/submit = 300) ===") + for _, r := range results { + t.Log(r.String()) + } +} + +// TestCommitCoalescesConcurrentSubmissions pins the mechanism the sweep +// explores, cheaply enough to run in CI: when several submissions are in +// flight while a slow commit is running, they must be merged into ONE +// subsequent commit rather than committed one at a time. +// +// This is the property any linger implementation must preserve — and the +// reason a linger cannot help at max_in_flight=1, where there is never a +// second submission to coalesce with. +func TestCommitCoalescesConcurrentSubmissions(t *testing.T) { + const ( + inFlight = 8 + latency = 300 * time.Millisecond + ) + ctx := t.Context() + + tbl, mem := newTestTable(t) + cat := &latentCatalog{memCatalog: mem, delay: latency} + + c, err := NewCommitter(tbl, cat, CommitConfig{ + ManifestMergeEnabled: false, + MaxRetries: 1, + }, func(context.Context) (*table.Table, error) { return cat.snapshot(), nil }, + service.MockResources().Logger()) + require.NoError(t, err) + defer c.Close() + + schemaID := c.currentSchemaID() + + // Build the data files up front, on the test goroutine, so the submitters + // below only have to commit — and so nothing in them needs to assert. + files := make([]iceberg.DataFile, inFlight) + for i := range files { + df, err := recordCountDataFile(tbl.Spec(), + fmt.Sprintf("%s/data/coalesce-%d-%s.parquet", tbl.Location(), i, uuid.New()), 300) + require.NoError(t, err) + files[i] = df + } + + // Occupy the committer so the rest of the submissions queue behind it. + errs := make([]error, inFlight) + var wg sync.WaitGroup + for i := range inFlight { + wg.Go(func() { + errs[i] = c.Commit(ctx, CommitInput{Files: []iceberg.DataFile{files[i]}, SchemaID: schemaID}) + }) + } + wg.Wait() + require.NoError(t, errors.Join(errs...), "submitter(s) failed") + + commits := cat.commits.Load() + require.Positive(t, commits, "expected at least one commit") + require.Less(t, commits, int64(inFlight), + "expected %d concurrent submissions to coalesce into fewer than %d commits, got %d", + inFlight, inFlight, commits) + t.Logf("%d concurrent submissions coalesced into %d commits (%.2f submissions/commit)", + inFlight, commits, float64(inFlight)/float64(commits)) +} diff --git a/internal/impl/iceberg/e2e/databricks/throughput_bench_test.go b/internal/impl/iceberg/e2e/databricks/throughput_bench_test.go new file mode 100644 index 0000000000..4cdeaa435f --- /dev/null +++ b/internal/impl/iceberg/e2e/databricks/throughput_bench_test.go @@ -0,0 +1,295 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed as a Redpanda Enterprise file under the Redpanda Community +// License (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md + +package databrickse2e + +import ( + "bytes" + "flag" + "fmt" + "log/slog" + "math/rand" + "slices" + "strings" + "sync" + "testing" + "time" + + "github.com/apache/iceberg-go" + "github.com/stretchr/testify/require" + + "github.com/redpanda-data/benthos/v4/public/service" + + icebergimpl "github.com/redpanda-data/connect/v4/internal/impl/iceberg" +) + +// TestDatabricksThroughput characterizes the "small-commit +// throughput trap" against a live Unity Catalog: plain APPEND commits (zero +// RowOpConfig — NOT copy-on-write) at a sweep of records-per-Route batch +// sizes, each driven for a fixed wall window. Because the router commits +// synchronously per Route call, per-call latency IS per-commit latency, and +// sustained throughput is records/window. The smallest point (300) exposes +// UC's floor commit latency for pure appends, comparable against AWS Glue's +// ~320ms and the COW bench's ~7-10s overwrites. +// +// Gated behind -databricks.throughput because it holds a live catalog busy +// for ~15 minutes of real time. + +var dbxThroughput = flag.Bool("databricks.throughput", false, "run the append throughput bench (drives the live catalog for ~15 minutes)") + +const ( + // throughputWindow is the wall-clock measurement window per batch-size + // point. The last Route call may overshoot it; throughput uses the actual + // elapsed time at that call's completion. + throughputWindow = 3 * time.Minute + // bigRouteCutoff: if the first measured Route call of a point exceeds + // this, the point is truncated to two calls total instead of the full + // window (per the run plan for the largest batch point). + bigRouteCutoff = 90 * time.Second + // payloadLen makes each JSON record ~1.2KB, mirroring the earlier + // benchmark methodology (id + seq + ~1.1KB string payload + JSON framing). + payloadLen = 1100 +) + +// syncBuffer is a concurrency-safe bytes.Buffer for the capturing logger — +// the committer may log from goroutines. +type syncBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *syncBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *syncBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + +// newCapturingRouter mirrors newRouter but wires a capturing slog logger (the +// output_iceberg_test.go seam) so committer warnings — commit retries, +// unknown-state, prohibited-key stripping — are collectable evidence. +func newCapturingRouter(t *testing.T, namespace, tableName string, rowOp icebergimpl.RowOpConfig) (*icebergimpl.Router, *syncBuffer) { + t.Helper() + namespaceStr, err := service.NewInterpolatedString(namespace) + require.NoError(t, err) + tableStr, err := service.NewInterpolatedString(tableName) + require.NoError(t, err) + + sb := &syncBuffer{} + logger := service.NewLoggerFromSlog(slog.New(slog.NewTextHandler(sb, &slog.HandlerOptions{ + Level: slog.LevelWarn, + }))) + + commitCfg := icebergimpl.CommitConfig{ + ManifestMergeEnabled: true, + MaxSnapshotAge: 24 * time.Hour, + MaxRetries: 3, + } + router := icebergimpl.NewRouter(buildCatalogConfig(), namespaceStr, tableStr, true, + icebergimpl.SchemaEvolutionConfig{Enabled: true}, commitCfg, rowOp, nil, logger) + t.Cleanup(func() { router.Close() }) + return router, sb +} + +// benchBatch builds a batch of structured append messages (~1.2KB of JSON +// each). Payload strings are zero-copy slices of a shared random pool so +// generation cost stays negligible next to multi-second commits; startID +// advances per call so ids stay unique and content varies across commits. +func benchBatch(pool string, startID int64, size int) service.MessageBatch { + rng := rand.New(rand.NewSource(startID)) //nolint:gosec // bench entropy, not crypto + msgs := make(service.MessageBatch, size) + for i := range msgs { + off := rng.Intn(len(pool) - payloadLen) + m := service.NewMessage(nil) + m.SetStructured(map[string]any{ + "id": startID + int64(i), + "seq": int64(i), + "payload": pool[off : off+payloadLen], + }) + msgs[i] = m + } + return msgs +} + +// warnEvidence tallies committer warning lines relevant to the parked #4591 +// throttle/5xx hypothesis and keeps a few (redacted) samples. +type warnEvidence struct { + commitRetries int + prohibitedKeys int + unknownState int + otherWarnings int + samples []string +} + +func collectWarnings(logged string) warnEvidence { + var ev warnEvidence + for line := range strings.SplitSeq(logged, "\n") { + if line == "" { + continue + } + lower := strings.ToLower(line) + switch { + case strings.Contains(line, "Commit attempt"): + ev.commitRetries++ + case strings.Contains(lower, "prohibit"): + ev.prohibitedKeys++ + case strings.Contains(lower, "unknown state"): + ev.unknownState++ + case strings.Contains(lower, "level=warn"), strings.Contains(lower, "level=error"): + ev.otherWarnings++ + default: + continue + } + if len(ev.samples) < 5 { + ev.samples = append(ev.samples, redact(line)) + } + } + return ev +} + +func percentile(sorted []time.Duration, p float64) time.Duration { + if len(sorted) == 0 { + return 0 + } + idx := int(p * float64(len(sorted)-1)) + return sorted[idx] +} + +type throughputPoint struct { + batchSize int + commits int + records int64 + elapsed time.Duration + p50, p95, max time.Duration + failures int + warnings warnEvidence +} + +func (pt throughputPoint) String() string { + perMin := float64(pt.commits) / pt.elapsed.Minutes() + return fmt.Sprintf("batch=%d commits=%d records=%d window=%v rec/s=%.0f commit_p50=%v p95=%v max=%v commits/min=%.1f failures=%d retries=%d prohibited=%d unknown=%d otherWarn=%d", + pt.batchSize, pt.commits, pt.records, pt.elapsed.Round(time.Second), + float64(pt.records)/pt.elapsed.Seconds(), + pt.p50.Round(time.Millisecond), pt.p95.Round(time.Millisecond), pt.max.Round(time.Millisecond), + perMin, pt.failures, + pt.warnings.commitRetries, pt.warnings.prohibitedKeys, pt.warnings.unknownState, pt.warnings.otherWarnings) +} + +func TestDatabricksThroughput(t *testing.T) { + skipIfNotConfigured(t) + if !*dbxThroughput { + t.Skip("set -databricks.throughput to run the append throughput bench") + } + ctx := t.Context() + + // Shared random pool for payload slicing (allocated once for the run). + poolRng := rand.New(rand.NewSource(42)) //nolint:gosec // bench entropy, not crypto + const alnum = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" + poolBytes := make([]byte, 64*1024) + for i := range poolBytes { + poolBytes[i] = alnum[poolRng.Intn(len(alnum))] + } + pool := string(poolBytes) + + var results []throughputPoint + + for _, batchSize := range []int{300, 5000, 50000, 200000} { + t.Run(fmt.Sprintf("batch_%d", batchSize), func(t *testing.T) { + tableName := uniqueTableName(fmt.Sprintf("tput_%d", batchSize)) + t.Cleanup(func() { dropTable(t, tableName) }) + + // Pre-create the table so the measured window contains only + // append commits — no CREATE TABLE inside the measurement. + client := newCatalogClient(t, ctx) + _, err := client.CreateTable(ctx, tableName, iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.Int64Type{}}, + iceberg.NestedField{ID: 2, Name: "seq", Type: iceberg.Int64Type{}}, + iceberg.NestedField{ID: 3, Name: "payload", Type: iceberg.StringType{}}, + )) + require.NoError(t, err) + + // Zero RowOpConfig => every message is a plain append. + router, logBuf := newCapturingRouter(t, *dbxSchema, tableName, icebergimpl.RowOpConfig{}) + + // Uncounted warmup: absorbs the first-write bootstrap (table + // load + timestamp-encoding property stamp) so measured calls + // are steady-state appends. + nextID := int64(1) + require.NoError(t, router.Route(ctx, benchBatch(pool, nextID, 10))) + nextID += 10 + + var ( + latencies []time.Duration + records int64 + failures int + ) + start := time.Now() + var elapsed time.Duration + for { + batch := benchBatch(pool, nextID, batchSize) + nextID += int64(batchSize) + + callStart := time.Now() + routeErr := router.Route(ctx, batch) + callDur := time.Since(callStart) + elapsed = time.Since(start) + + if routeErr != nil { + failures++ + t.Logf("batch=%d commit %d FAILED after %v: %v", batchSize, len(latencies)+failures, callDur.Round(time.Millisecond), redact(routeErr.Error())) + if failures >= 3 { + t.Logf("batch=%d: aborting point after %d consecutive-ish failures", batchSize, failures) + break + } + } else { + latencies = append(latencies, callDur) + records += int64(batchSize) + t.Logf("batch=%d commit %d: %v (%.0f rec/s within call)", batchSize, len(latencies), callDur.Round(time.Millisecond), float64(batchSize)/callDur.Seconds()) + } + + if elapsed >= throughputWindow { + break + } + // Very large batches: cap at two calls if a single call + // blows past the cutoff (keeps live time bounded). + if len(latencies)+failures >= 2 && callDur > bigRouteCutoff { + t.Logf("batch=%d: truncating point to %d calls (single Route exceeded %v)", batchSize, len(latencies)+failures, bigRouteCutoff) + break + } + } + + slices.Sort(latencies) + pt := throughputPoint{ + batchSize: batchSize, + commits: len(latencies), + records: records, + elapsed: elapsed, + p50: percentile(latencies, 0.50), + p95: percentile(latencies, 0.95), + max: percentile(latencies, 1.0), + failures: failures, + warnings: collectWarnings(logBuf.String()), + } + results = append(results, pt) + t.Logf("POINT RESULT: %s", pt) + for _, s := range pt.warnings.samples { + t.Logf(" warning sample: %s", s) + } + }) + } + + t.Log("=== THROUGHPUT SWEEP SUMMARY (append mode, live Unity Catalog) ===") + for _, pt := range results { + t.Logf(" %s", pt) + } +} diff --git a/internal/impl/iceberg/shredder/shredder.go b/internal/impl/iceberg/shredder/shredder.go index cadc88eca0..2034125572 100644 --- a/internal/impl/iceberg/shredder/shredder.go +++ b/internal/impl/iceberg/shredder/shredder.go @@ -152,6 +152,33 @@ func (rs *RecordShredder) shredStruct( path icebergx.Path, repLevel, defLevel, maxRepLevel int, sink Sink, +) error { + // Case-sensitive matching makes the match-key the identity function, so + // schema fields can be looked up directly in the input map and neither of + // the two per-record maps below is needed. That path is split out because + // this function runs once per record (and once per nested struct within + // it), and a 1-vCPU allocation profile attributed a material share of the + // sink's total allocations to those two maps. + if rs.caseSensitive { + return rs.shredStructExact(fields, value, path, repLevel, defLevel, maxRepLevel, sink) + } + + return rs.shredStructFolded(fields, value, path, repLevel, defLevel, maxRepLevel, sink) +} + +// shredStructFolded is the case-insensitive path: input keys are matched against +// schema field names by their folded (lowercased) form, which means several +// distinct input keys can collide on one schema field and that collision has to +// be reported rather than silently resolved. +// +// shredStructExact must stay behaviourally identical to this for input that +// happens to match exactly; TestShredStructPathsAgree pins that. +func (rs *RecordShredder) shredStructFolded( + fields []iceberg.NestedField, + value map[string]any, + path icebergx.Path, + repLevel, defLevel, maxRepLevel int, + sink Sink, ) error { // Build an index of input keys by their match-key (the original key in // case-sensitive mode, or its lowercase form in case-insensitive mode). @@ -227,6 +254,83 @@ func (rs *RecordShredder) shredStruct( return nil } +// shredStructExact is shredStruct's case-sensitive equivalent: input keys must +// match schema field names byte-for-byte, so a field's value is just +// value[field.Name] and case-collision ambiguity is impossible (Go map keys are +// themselves case-sensitive). +// +// It must stay behaviourally identical to the general path for case-sensitive +// shredders — same required-field errors, same null handling, same +// OnNewField notifications, same traversal order of fields. +func (rs *RecordShredder) shredStructExact( + fields []iceberg.NestedField, + value map[string]any, + path icebergx.Path, + repLevel, defLevel, maxRepLevel int, + sink Sink, +) error { + // Count how many input keys were claimed by a schema field, so unknown-field + // detection below can usually be skipped without tracking a set. + matchedKeys := 0 + + for _, field := range fields { + fieldValue, exists := value[field.Name] + if exists { + matchedKeys++ + } + + // Validate required fields. + if field.Required && (!exists || fieldValue == nil) { + return &RequiredFieldNullError{field, path} + } + + // Compute this field's definition level contribution. + fieldDefLevel := defLevel + if !field.Required { + fieldDefLevel++ // Optional field adds to max def level. + } + + // Build path for this field. The schema's casing is the input's casing + // here, so no canonicalisation is needed. + fieldPath := append(path, icebergx.PathSegment{Kind: icebergx.PathField, Name: field.Name}) + + if !exists || fieldValue == nil { + // Field is null or missing - emit null for all leaf descendants. + if err := rs.shredNull(field.Type, field.ID, repLevel, defLevel, sink); err != nil { + return err + } + continue + } + + if err := rs.shredValue(field.Type, field.ID, fieldValue, fieldPath, repLevel, fieldDefLevel, maxRepLevel, sink); err != nil { + return fmt.Errorf("field %q: %w", field.Name, err) + } + } + + // Detect unknown fields in input. Field names are unique within a struct, + // so each matched field claimed exactly one distinct input key: when the + // counts agree, every key is accounted for and there is nothing to report. + // That is the steady state once a schema has stabilised, and it makes the + // common case allocation-free. (A count above len(value) is impossible for + // a well-formed schema, and would simply fall through to the scan.) + if matchedKeys == len(value) { + return nil + } + + // Something is unknown: pay for a lookup set now, on the rare path only. + known := make(map[string]struct{}, len(fields)) + for _, field := range fields { + known[field.Name] = struct{}{} + } + for key, val := range value { + if _, ok := known[key]; !ok { + sink.OnNewField(slices.Clone(path), key, val) + } + } + + return nil +} + // matchKey returns the lookup key used to compare an input record key against // a schema field name. In case-sensitive mode it is the identity; in // case-insensitive mode it folds to lowercase. diff --git a/internal/impl/iceberg/shredder/shredder_paths_agree_test.go b/internal/impl/iceberg/shredder/shredder_paths_agree_test.go new file mode 100644 index 0000000000..c0fb31cbae --- /dev/null +++ b/internal/impl/iceberg/shredder/shredder_paths_agree_test.go @@ -0,0 +1,214 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed as a Redpanda Enterprise file under the Redpanda Community +// License (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md + +package shredder + +import ( + "fmt" + "sort" + "testing" + + "github.com/apache/iceberg-go" + "github.com/stretchr/testify/require" +) + +// TestShredStructPathsAgree pins the case-sensitive fast path +// (shredStructExact) to the general folded path (shredStructFolded). +// +// shredStructExact exists purely to avoid two per-record map allocations when +// key matching is exact — which is the default (case_sensitive_columns: true) +// and therefore the path almost all traffic takes. It is an optimisation, so it +// must not change observable behaviour by even one emitted value, new-field +// notification, or error. +// +// The comparison goes through the public Shred entry point with two shredders +// rather than calling the two helpers directly, because shredStruct dispatches +// on rs.caseSensitive on *every* recursion: a single case-sensitive shredder +// routes nested structs to shredStructExact no matter which helper was called +// at the top level, so calling the helpers directly would compare the exact +// path against itself below the root and silently pass on any nested +// divergence. Two shredders makes one run exact all the way down and the other +// folded all the way down. +// +// That comparison is only legitimate for a case-unambiguous corpus, so every +// schema field name and record key below is lower-case: folding then maps each +// key to itself and the two modes are *required* to agree. Inputs that differ +// in case are exactly where the modes are meant to diverge, and those belong in +// the case-sensitivity tests instead. +// +// It covers the cases the optimisation actually reasons about: every key +// matching (the allocation-free steady state), extra unknown keys (the fallback +// scan), missing fields, explicit nulls, required-field violations, and nesting +// — plus an empty record and an empty schema, where the matched-count shortcut +// is most likely to be wrong. +func TestShredStructPathsAgree(t *testing.T) { + nested := iceberg.NestedField{ + ID: 10, + Name: "inner", + Type: &iceberg.StructType{FieldList: []iceberg.NestedField{ + {ID: 11, Name: "a", Type: iceberg.PrimitiveTypes.Int64, Required: false}, + {ID: 12, Name: "b", Type: iceberg.PrimitiveTypes.String, Required: false}, + }}, + Required: false, + } + + // deep carries a REQUIRED leaf two levels down, so nested required-field + // errors and nested unknown-field notifications are both reachable — the + // divergences the earlier version of this test could not have seen. + deep := iceberg.NestedField{ + ID: 20, + Name: "deep", + Type: &iceberg.StructType{FieldList: []iceberg.NestedField{ + {ID: 21, Name: "mid", Type: &iceberg.StructType{FieldList: []iceberg.NestedField{ + {ID: 22, Name: "leaf", Type: iceberg.PrimitiveTypes.String, Required: true}, + }}, Required: false}, + }}, + Required: false, + } + + schema := iceberg.NewSchema(1, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + iceberg.NestedField{ID: 2, Name: "name", Type: iceberg.PrimitiveTypes.String, Required: false}, + iceberg.NestedField{ID: 3, Name: "flag", Type: iceberg.PrimitiveTypes.Bool, Required: false}, + nested, + deep, + ) + + emptySchema := iceberg.NewSchema(2) + + cases := []struct { + name string + schema *iceberg.Schema + record map[string]any + }{ + { + name: "all fields present", + schema: schema, + record: map[string]any{ + "id": int64(1), "name": "a", "flag": true, + "inner": map[string]any{"a": int64(2), "b": "c"}, + }, + }, + { + name: "exact match, no unknowns (allocation-free steady state)", + schema: schema, + record: map[string]any{"id": int64(1), "name": "a", "flag": false, "inner": nil}, + }, + { + name: "one unknown key", + schema: schema, + record: map[string]any{"id": int64(1), "surprise": "x"}, + }, + { + name: "several unknown keys", + schema: schema, + record: map[string]any{"id": int64(1), "x": 1, "y": "two", "z": nil}, + }, + { + name: "unknown key nested inside a known struct", + schema: schema, + record: map[string]any{ + "id": int64(1), + "inner": map[string]any{"a": int64(2), "nope": "surprise"}, + }, + }, + { + name: "unknown key in a doubly-nested struct", + schema: schema, + record: map[string]any{ + "id": int64(1), + "deep": map[string]any{"mid": map[string]any{"leaf": "ok", "extra": 1}}, + }, + }, + { + name: "required leaf missing two levels down (nested error path)", + schema: schema, + record: map[string]any{ + "id": int64(1), + "deep": map[string]any{"mid": map[string]any{}}, + }, + }, + { + name: "required leaf explicitly null two levels down", + schema: schema, + record: map[string]any{ + "id": int64(1), + "deep": map[string]any{"mid": map[string]any{"leaf": nil}}, + }, + }, + { + name: "missing optional fields", + schema: schema, + record: map[string]any{"id": int64(7)}, + }, + { + name: "explicit nulls", + schema: schema, + record: map[string]any{"id": int64(7), "name": nil, "flag": nil, "inner": nil}, + }, + { + name: "required field missing (error path)", + schema: schema, + record: map[string]any{"name": "no id here"}, + }, + { + name: "required field explicitly null (error path)", + schema: schema, + record: map[string]any{"id": nil}, + }, + { + name: "empty record", + schema: schema, + record: map[string]any{}, + }, + { + name: "empty schema, empty record", + schema: emptySchema, + record: map[string]any{}, + }, + { + name: "empty schema, all keys unknown", + schema: emptySchema, + record: map[string]any{"a": 1, "b": 2}, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + exactSink := &testSink{} + exactErr := NewRecordShredder(tc.schema, true).Shred(tc.record, exactSink) + + foldedSink := &testSink{} + foldedErr := NewRecordShredder(tc.schema, false).Shred(tc.record, foldedSink) + + if foldedErr != nil { + require.EqualError(t, exactErr, foldedErr.Error(), + "fast path must fail exactly as the general path does") + return + } + require.NoError(t, exactErr, "fast path errored where the general path did not") + + require.Equal(t, foldedSink.values, exactSink.values, + "emitted values must be identical (including order)") + + // New-field notification order follows Go map iteration, which is + // randomised, so compare as sets. + require.Equal(t, sortedNewFields(foldedSink.newFields), sortedNewFields(exactSink.newFields), + "new-field notifications must be identical") + }) + } +} + +func sortedNewFields(in []newFieldRecord) []string { + out := make([]string, 0, len(in)) + for _, nf := range in { + out = append(out, fmt.Sprintf("path=%v name=%s value=%v", nf.path, nf.name, nf.value)) + } + sort.Strings(out) + return out +}