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):