From 0f46eefc1eb370f98c5d760e398797eb6fdea2af Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 4 Nov 2026 05:40:37 +0100 Subject: [PATCH] fix(streaming): refuse Iceberg partition transforms that Flink's DDL cannot create The managed lane compiled icebergConfig.partitionSpec transforms into CREATE TABLE ... PARTITIONED BY (days(`event_time`)), and likewise for months, hours, bucket and truncate. Flink cannot parse that. Iceberg's Flink DDL partitions by plain columns only, because Iceberg's transforms are hidden partitioning. So the job failed after deploy with 'Encountered "(" at line 16'. The live lane found this; it predates the record-shapes work (#784). fluid validate now refuses those transforms: create the table with that partition spec before the job runs, and leave it out of the contract. Identity partitions compile as before. Real support needs the runner to create the table through Iceberg's catalog API, and is tracked separately. --- fluid_build/streaming/compiler.py | 16 ++++++++------ tests/streaming/test_compiler.py | 36 ++++++++++++++----------------- 2 files changed, 25 insertions(+), 27 deletions(-) diff --git a/fluid_build/streaming/compiler.py b/fluid_build/streaming/compiler.py index 4574cb30..5afa6219 100644 --- a/fluid_build/streaming/compiler.py +++ b/fluid_build/streaming/compiler.py @@ -415,13 +415,15 @@ def reject_uncompiled_rules(node: Any) -> None: transform = item["transform"] if transform == "identity": partition.append(f"`{name}`") - elif transform in {"year", "month", "day", "hour"}: - partition.append(f"{transform}s(`{name}`)") - elif transform in {"bucket", "truncate"}: - parameter = item.get("numBuckets" if transform == "bucket" else "width") - if not isinstance(parameter, int) or isinstance(parameter, bool) or parameter < 1: - raise ValueError(f"{transform} requires a positive size") - partition.append(f"{transform}({parameter}, `{name}`)") + elif transform in {"year", "month", "day", "hour", "bucket", "truncate"}: + # Iceberg's transforms are hidden partitioning, which Flink SQL cannot declare: + # Iceberg's Flink DDL partitions by plain columns only, and PARTITIONED BY + # (days(col)) does not parse, so the job would fail after it deploys. + raise ValueError( + f"Partition transform {transform!r} on {name!r}: Flink's Iceberg DDL can " + "partition only by a column (transform identity). Create the table with " + "this partition spec before the job runs, and leave it out here" + ) else: raise ValueError(f"Unsupported partition transform {transform!r}") auth, auth_errors = parse_catalog_auth(cfg.get("catalogAuth")) diff --git a/tests/streaming/test_compiler.py b/tests/streaming/test_compiler.py index 30046401..7a376354 100644 --- a/tests/streaming/test_compiler.py +++ b/tests/streaming/test_compiler.py @@ -126,11 +126,11 @@ def test_event_time_and_partition(contract): cfg = managed_config(contract["builds"][0]) cfg["source"]["eventTime"] = {"column": "event_time", "delaySeconds": 5} contract["exposes"][0]["binding"]["icebergConfig"]["partitionSpec"] = [ - {"sourceColumn": "event_time", "transform": "day"}, + {"sourceColumn": "event_id", "transform": "identity"}, ] (artifact,) = compile_contract(contract) assert "WATERMARK FOR `event_time` AS `event_time` - INTERVAL '5' SECOND" in artifact.sql - assert "PARTITIONED BY (days(`event_time`))" in artifact.sql + assert "PARTITIONED BY (`event_id`)" in artifact.sql @pytest.mark.parametrize("surface", ["root", "column", "source", "dq"]) @@ -234,32 +234,28 @@ def test_avro_source_emits_registry_format_without_json_options(contract): assert "json." not in artifact.sql -@pytest.mark.parametrize( - "partition,sql", - [ - ({"sourceColumn": "event_id", "transform": "identity"}, "`event_id`"), - ( - {"sourceColumn": "event_id", "transform": "bucket", "numBuckets": 8}, - "bucket(8, `event_id`)", - ), - ( - {"sourceColumn": "event_id", "transform": "truncate", "width": 4}, - "truncate(4, `event_id`)", - ), - ], -) -def test_partition_transforms_use_native_iceberg_ddl(contract, partition, sql): - contract["exposes"][0]["binding"]["icebergConfig"]["partitionSpec"] = [partition] +def test_an_identity_partition_uses_native_iceberg_ddl(contract): + contract["exposes"][0]["binding"]["icebergConfig"]["partitionSpec"] = [ + {"sourceColumn": "event_id", "transform": "identity"} + ] (artifact,) = compile_contract(contract) - assert f"PARTITIONED BY ({sql})" in artifact.sql + assert "PARTITIONED BY (`event_id`)" in artifact.sql @pytest.mark.parametrize( "partition,match", [ ({"sourceColumn": "missing", "transform": "identity"}, "not a target"), - ({"sourceColumn": "event_id", "transform": "bucket", "numBuckets": 0}, "positive"), ({"sourceColumn": "event_id", "transform": "unknown"}, "Unsupported partition"), + # Hidden partitioning: Flink's Iceberg DDL cannot declare it, so the job would + # fail after deploy with "Encountered "(" at line 16". + ({"sourceColumn": "event_time", "transform": "day"}, "only by a column"), + ({"sourceColumn": "event_time", "transform": "hour"}, "only by a column"), + ( + {"sourceColumn": "event_id", "transform": "bucket", "numBuckets": 8}, + "only by a column", + ), + ({"sourceColumn": "event_id", "transform": "truncate", "width": 4}, "only by a column"), ], ) def test_invalid_partition_transforms_fail(contract, partition, match):