This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 399a3406a582 fix(spark): validate partitions and cover keyed SDP flows
(#20496)
399a3406a582 is described below
commit 399a3406a5824d19135b04554ddfd436a2ab92e7
Author: Danny Chan <[email protected]>
AuthorDate: Fri Nov 6 14:40:37 2026 +0800
fix(spark): validate partitions and cover keyed SDP flows (#20496)
* fix(spark): validate partitions and cover keyed SDP flows
* fix(spark): normalize partitions and strengthen keyed flow checks
---
hudi-integ-test/spark-pipelines/KEYED_FLOWS.md | 54 ++++++++
hudi-integ-test/spark-pipelines/keyed_flows.py | 145 +++++++++++++++++++++
.../spark/sql/hudi/catalog/HoodieCatalog.scala | 13 +-
.../sql/hudi/catalog/TestHoodieCatalogDDL.scala | 42 +++++-
4 files changed, 252 insertions(+), 2 deletions(-)
diff --git a/hudi-integ-test/spark-pipelines/KEYED_FLOWS.md
b/hudi-integ-test/spark-pipelines/KEYED_FLOWS.md
new file mode 100644
index 000000000000..888f642483fb
--- /dev/null
+++ b/hudi-integ-test/spark-pipelines/KEYED_FLOWS.md
@@ -0,0 +1,54 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+
+# Keyed SDP flow acceptance
+
+Run the real Spark 4.1.1 CLI against Connect using the #20453 harness and a
+candidate bundle containing this PR's early partition-field validation.
+Use JDK 17 and install the harness's pinned requirements in a dedicated
environment.
+
+```sh
+python hudi-integ-test/spark-pipelines/keyed_flows.py \
+ --harness-dir /path/to/harness/hudi-integ-test/spark-pipelines \
+ --bundle /path/to/candidate-hudi-spark4.1-bundle.jar
+```
+
+The generated SQL defines partitioned COW tables with `primaryKey=id`,
+`preCombineField=ts`, and `hoodie.record.merge.mode=EVENT_TIME_ORDERING`.
+A boolean `_hoodie_is_deleted` carries deletes. The scenario checks a newly
+created target and a pre-existing catalog table, then restarts each through the
+same SDP-managed checkpoint. It verifies exact contents after intra-batch
+precombine (higher ordering value arrives first), a newer update, a late
lower-order
+update, a late delete that must preserve the newer row, deletion, reinsertion
and
+an unchanged restart. Partition `b` retains its unrelated key throughout.
+
+Negative scenarios reject nonexistent key, ordering and partition fields,
+non-boolean delete markers and a change to persisted record-key identity.
+Failed new targets must have no completed data commit; the existing valid
targets
+must retain their exact contents after every rejected scenario.
+
+Generated Python/SQL and logs remain under `target/keyed-acceptance-logs`.
+Metadata indexing is disabled. This is delete input carried by a flow, not CDC
+propagation of upstream deletes. It does not cover full refresh, table
chaining,
+fan-in or failure between commit and checkpoint (separate recovery coverage).
+
+MOR log-merge coverage remains deferred under #20408; this matrix covers COW
only.
+The flows rely on the default streaming upsert operation, without a session
override.
+The catalog regression uses `PARTITIONED BY (PART)` against schema field `part`
+to verify case-insensitive normalization. The SDP scenario uses exact spelling
+because Spark separately rejects a declared/stored partition spelling
mismatch. Land after the #20453 harness dependency.
+Checks remain active under `python -O`; SIGTERM unwinds harness cleanup.
diff --git a/hudi-integ-test/spark-pipelines/keyed_flows.py
b/hudi-integ-test/spark-pipelines/keyed_flows.py
new file mode 100644
index 000000000000..4d1d2ed6b570
--- /dev/null
+++ b/hudi-integ-test/spark-pipelines/keyed_flows.py
@@ -0,0 +1,145 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Check keyed Hudi SDP flows across ordering, deletion, reinsertion and
restart."""
+import argparse
+import json
+import signal
+from pathlib import Path
+import sys
+import tempfile
+
+SOURCE = 'from pyspark import pipelines as dp\nfrom pyspark.sql import
SparkSession\nspark =
SparkSession.active()\[email protected]_view(name="input_events")\ndef
input_events():\n return spark.readStream.schema("id INT, ts LONG, part
STRING, value STRING, _hoodie_is_deleted
BOOLEAN").json(spark.conf.get("hudi.test.input"))\n'
+PROPERTIES = "'type'='cow', 'primaryKey'='id', 'preCombineField'='ts',
'hoodie.record.merge.mode'='EVENT_TIME_ORDERING'"
+TABLES = ("keyed_new", "keyed_existing")
+
+
+def require(condition, message):
+ if not condition:
+ raise AssertionError(message)
+
+
+def interrupted(signum, _frame):
+ raise KeyboardInterrupt(f"Received signal {signum}")
+
+
+def client(mode, remote, root):
+ from pyspark.sql import SparkSession
+ spark = SparkSession.builder.remote(remote).getOrCreate()
+ try:
+ require(spark.conf.get("spark.hudi.test.run.id") == root.name,
"Connected to the wrong test server")
+ if mode == "prepare":
+ spark.sql(f"CREATE TABLE sdp_test.keyed_existing (id INT, ts
BIGINT, part STRING, value STRING, "
+ f"_hoodie_is_deleted BOOLEAN) USING hudi PARTITIONED BY
(part) TBLPROPERTIES ({PROPERTIES})").collect()
+ else:
+ expected = json.loads((root / "expected.json").read_text())
+ for table in TABLES:
+ actual = [list(row) for row in spark.table(f"sdp_test.{table}")
+ .select("id", "ts", "part",
"value").orderBy("id").collect()]
+ require(actual == expected, (table, actual, expected))
+ print(f"PASS {mode} {table}: {actual}")
+ finally:
+ spark.stop()
+
+
+def scenario(h):
+ source = h.root / "input"
+ source.mkdir()
+ (h.root / "source.py").write_text(SOURCE)
+ flows = "\n".join(
+ f"CREATE STREAMING TABLE {table} USING hudi PARTITIONED BY (part)
TBLPROPERTIES ({PROPERTIES}) "
+ "AS SELECT id, ts, part, value, _hoodie_is_deleted FROM STREAM
input_events;" for table in TABLES)
+ (h.root / "flows.sql").write_text(flows)
+ spec = {"name": "keyed_flows", "storage": (h.root /
"checkpoints").as_uri(),
+ "catalog": "spark_catalog", "database": "sdp_test",
+ "libraries": [{"glob": {"include": "source.py"}}, {"glob":
{"include": "flows.sql"}}],
+ "configuration": {"hudi.test.input": source.as_uri(),
"spark.sql.shuffle.partitions": "2",
+ "spark.hoodie.metadata.enable": "false",
+ "spark.sql.pipelines.maxFlowRetryAttempts": "0"}}
+ path = h.root / "spark-pipeline.yml"
+ path.write_text(json.dumps(spec, indent=2))
+ for name in ("source.py", "flows.sql", path.name):
+ (h.logs / name).write_text((h.root / name).read_text())
+ h.client(Path(__file__).resolve(), "prepare")
+ stages = (
+ ([(1, 2, "a", "new", False), (1, 1, "a", "old", False), (2, 1, "b",
"keep", False)],
+ [[1, 2, "a", "new"], [2, 1, "b", "keep"]]),
+ ([(1, 3, "a", "updated", False)], [[1, 3, "a", "updated"], [2, 1, "b",
"keep"]]),
+ ([(1, 2, "a", "late", False)], [[1, 3, "a", "updated"], [2, 1, "b",
"keep"]]),
+ ([(1, 2, "a", "late-delete", True)], [[1, 3, "a", "updated"], [2, 1,
"b", "keep"]]),
+ ([(1, 4, "a", "deleted", True)], [[2, 1, "b", "keep"]]),
+ ([(1, 5, "a", "reinserted", False)], [[1, 5, "a", "reinserted"], [2,
1, "b", "keep"]]))
+ for i, (rows, expected) in enumerate(stages, 1):
+ (source / f"{i}.json").write_text("".join(json.dumps(dict(zip(
+ ("id", "ts", "part", "value", "_hoodie_is_deleted"), row))) + "\n"
for row in rows))
+ (h.root / "expected.json").write_text(json.dumps(expected))
+ h.run_pipeline(path, f"stage-{i}")
+ h.client(Path(__file__).resolve(), f"verify-{i}")
+ h.run_pipeline(path, "restart")
+ h.client(Path(__file__).resolve(), "verify-restart")
+ invalid = (
+ ("missing_key", PROPERTIES.replace("'primaryKey'='id'",
"'primaryKey'='missing_key'"), "part", SOURCE, "Can't find primaryKey"),
+ ("missing_order", PROPERTIES.replace("'preCombineField'='ts'",
"'preCombineField'='missing_order'"), "part", SOURCE, "Can't find ordering
fields"),
+ ("missing_partition", PROPERTIES, "missing_partition", SOURCE,
"Partition column 'missing_partition'"),
+ ("invalid_marker", PROPERTIES, "part",
SOURCE.replace("_hoodie_is_deleted BOOLEAN", "_hoodie_is_deleted STRING"), "has
to be BOOLEAN"),
+ ("changed_identity", PROPERTIES.replace("'primaryKey'='id'",
"'primaryKey'='value'"), "part", SOURCE, "Can't set primaryKey to 'value' on
Hudi table"))
+ for name, props, partition, definition, error_text in invalid:
+ target = "keyed_existing" if name == "changed_identity" else
f"bad_{name}"
+ (h.root / "source.py").write_text(definition)
+ statement = (f"CREATE STREAMING TABLE {target} USING hudi PARTITIONED
BY ({partition}) "
+ f"TBLPROPERTIES ({props}) AS SELECT id, ts, part, value,
_hoodie_is_deleted FROM STREAM input_events;")
+ (h.root / "flows.sql").write_text(statement)
+ spec["name"] = name
+ spec["storage"] = (h.root / f"checkpoint-{name}").as_uri()
+ path.write_text(json.dumps(spec, indent=2))
+ (h.logs / f"{name}.sql").write_text(statement)
+ try:
+ h.run_pipeline(path, f"reject-{name}")
+ except RuntimeError:
+ log = (h.logs / f"reject-{name}.log").read_text()
+ require(error_text.lower() in log.lower(), (name, log[-5000:]))
+ require(h.server.poll() is None, name)
+ else:
+ raise AssertionError(f"Invalid configuration unexpectedly
succeeded: {name}")
+ if name != "changed_identity":
+ timeline = h.root / "warehouse" / "sdp_test.db" / target /
".hoodie" / "timeline"
+ require(not [p for p in timeline.glob("*") if p.suffix in
(".commit", ".deltacommit", ".replacecommit")], name)
+ h.client(Path(__file__).resolve(), f"verify-after-{name}")
+ print("PASS: keyed deduplication, updates, late values, delete,
reinsertion, restart and invalid configurations")
+
+
+def main():
+ if len(sys.argv) == 4 and (sys.argv[1] == "prepare" or
sys.argv[1].startswith("verify-")):
+ client(sys.argv[1], sys.argv[2], Path(sys.argv[3]))
+ return
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("--harness-dir", required=True, type=Path)
+ parser.add_argument("--bundle", required=True, type=Path)
+ parser.add_argument("--logs", type=Path,
default=Path("target/keyed-acceptance-logs"))
+ args = parser.parse_args()
+ sys.path.insert(0, str(args.harness_dir.resolve()))
+ from harness import ConnectHarness
+ args.logs.mkdir(parents=True, exist_ok=True)
+ logs = Path(tempfile.mkdtemp(prefix="run-", dir=args.logs.resolve()))
+ print(f"Keyed logs: {logs}", flush=True)
+ signal.signal(signal.SIGTERM, interrupted)
+ with ConnectHarness(args.bundle, logs) as harness:
+ scenario(harness)
+
+
+if __name__ == "__main__":
+ main()
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala
index 6ae0f9a91ee2..38ebd5da4497 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala
@@ -391,8 +391,19 @@ class HoodieCatalog extends DelegatingCatalogExtension
operation: TableCreationMode): Table = {
val (partitionColumns, maybeBucketSpec) =
HoodieCatalog.convertTransforms(partitions)
+ if (schema.nonEmpty) {
+ partitionColumns.foreach { column =>
+ if (!schema.fieldNames.exists(spark.sessionState.conf.resolver(_,
column))) {
+ throw new HoodieAnalysisException(s"Partition column '$column' not
found in table schema: ${schema.treeString}")
+ }
+ }
+ }
val newSchema = schema
- val newPartitionColumns = partitionColumns
+ val newPartitionColumns = if (schema.nonEmpty) {
+ partitionColumns.map(column =>
schema.fieldNames.find(spark.sessionState.conf.resolver(_, column)).get)
+ } else {
+ partitionColumns
+ }
val newBucketSpec = maybeBucketSpec
val storage =
DataSource.buildStorageFormatFromOptions(writeOptions.--(needFilterProps))
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogDDL.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogDDL.scala
index 211028bd6241..7b6f49229d3b 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogDDL.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/catalog/TestHoodieCatalogDDL.scala
@@ -25,7 +25,7 @@ import
org.apache.spark.sql.catalyst.analysis.NoSuchTableException
import org.apache.spark.sql.catalyst.catalog.HoodieCatalogTable
import org.apache.spark.sql.connector.catalog.{Identifier, Table,
TableCatalog, TableChange}
import
org.apache.spark.sql.connector.catalog.TableCapability.{ACCEPT_ANY_SCHEMA,
BATCH_READ, OVERWRITE_BY_FILTER, TRUNCATE, V1_BATCH_WRITE}
-import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.connector.expressions.{Expressions, Transform}
import org.apache.spark.sql.execution.datasources.v2.V2SessionCatalog
import org.apache.spark.sql.hudi.HoodieSqlCommonUtils
import org.apache.spark.sql.hudi.command.ShowHoodieCreateTableCommand
@@ -109,6 +109,46 @@ class TestHoodieCatalogDDL extends HoodieSparkSqlTestBase {
}
}
+ test("createTable rejects missing partition fields before creating Hudi
metadata") {
+ withTempDir { tmp =>
+ val name = generateTableName
+ val ident = Identifier.of(Array("default"), name)
+ val path = s"${tmp.getCanonicalPath}/$name"
+ val error = intercept[HoodieAnalysisException] {
+ hoodieCatalog.createTable(ident, StructType(Seq(StructField("id",
IntegerType))),
+ Array[Transform](Expressions.identity("missing_partition")),
+ Map("provider" -> "hudi", "primaryKey" -> "id", "location" ->
path).asJava)
+ }
+ assertTrue(error.getMessage.contains("missing_partition"),
error.getMessage)
+ assertFalse(hoodieCatalog.tableExists(ident))
+ assertFalse(new File(s"$path/.hoodie/hoodie.properties").exists())
+ }
+ }
+
+ test("createTable normalizes partition field spelling using the resolver") {
+ withSQLConf("spark.sql.caseSensitive" -> "false") {
+ withTempDir { tmp =>
+ val name = generateTableName
+ val ident = Identifier.of(Array("default"), name)
+ val path = s"${tmp.getCanonicalPath}/$name"
+ hoodieCatalog.createTable(ident,
+ StructType(Seq(StructField("id", IntegerType), StructField("part",
StringType))),
+ Array[Transform](Expressions.identity("PART")),
+ Map("provider" -> "hudi", "primaryKey" -> "id", "location" ->
path).asJava)
+ try {
+ spark.sql(s"INSERT INTO $name VALUES (1, 'a')")
+ checkAnswer(s"SELECT id, part FROM $name")(Seq(1, "a"))
+ val props = new java.util.Properties()
+ val stream = new
java.io.FileInputStream(s"$path/.hoodie/hoodie.properties")
+ try props.load(stream) finally stream.close()
+ assertEquals("part",
props.getProperty("hoodie.table.partition.fields"))
+ } finally {
+ hoodieCatalog.dropTable(ident)
+ }
+ }
+ }
+ }
+
Seq(false, true).foreach { operationMetadata =>
test(s"alterTable preserves metadata columns during repeated schema
reconciliation, operation=$operationMetadata") {
withSQLConf("hoodie.schema.on.read.enable" -> "true") {
