fix(streaming): refuse Iceberg partition transforms that Flink's DDL cannot create

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

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