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 5107bee3176a test(spark): add repeatable SDP phase one acceptance
scenario (#20493)
5107bee3176a is described below
commit 5107bee3176a9c363c788985e667fe5d25ba26a7
Author: Danny Chan <[email protected]>
AuthorDate: Wed Nov 4 10:14:00 2026 +0800
test(spark): add repeatable SDP phase one acceptance scenario (#20493)
* test(spark): add repeatable SDP phase one acceptance scenario
* test(spark): cover MOR compaction and metadata in SDP acceptance
---
hudi-integ-test/spark-pipelines/ACCEPTANCE.md | 68 ++++++++
.../spark-pipelines/phase1_acceptance.py | 180 +++++++++++++++++++++
2 files changed, 248 insertions(+)
diff --git a/hudi-integ-test/spark-pipelines/ACCEPTANCE.md
b/hudi-integ-test/spark-pipelines/ACCEPTANCE.md
new file mode 100644
index 000000000000..46143eae9bee
--- /dev/null
+++ b/hudi-integ-test/spark-pipelines/ACCEPTANCE.md
@@ -0,0 +1,68 @@
+<!--
+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.
+-->
+
+# RFC-112 Phase 1 acceptance
+
+`phase1_acceptance.py` runs the real Spark 4.1.1 `spark-pipelines` CLI
+against a separate Connect server using the harness from PR #20453.
+It can run before that harness lands: pass its checkout's directory explicitly.
+Install that harness's pinned `requirements.txt` in a dedicated virtual
environment
+and build the candidate Spark 4.1 bundle using JDK 17:
+
+```sh
+mvn package -Pjava17 -Dspark4.1 -Dscala-2.13 -DskipTests -Dgpg.skip \
+ -Djacoco.skip -am -pl packaging/hudi-spark-bundle
+python hudi-integ-test/spark-pipelines/phase1_acceptance.py \
+ --harness-dir /path/to/harness-checkout/hudi-integ-test/spark-pipelines \
+ --bundle
"$PWD/packaging/hudi-spark-bundle/target/hudi-spark4.1-bundle_2.13-1.3.0-SNAPSHOT.jar"
+```
+
+The deterministic JSON source supplies duplicate keys, then new versions in a
+second file. Independent insert and upsert flows capture `spark.hoodie.*`
+operation settings through SQL SET. Exact rows distinguish their behavior after
+each run. An unchanged restart checks exact rows and equality of completed Hudi
+commit filenames; `restart-commits.json` preserves the before/after evidence.
+Finally, a null-key flow must fail while the same Connect server remains
available
+for a new remote query. Every client and CLI invocation is bounded by the
harness.
+Logs remain under `target/phase1-acceptance-logs` after temporary data is
removed.
+
+Use `--table-type cow` (default) or `--table-type mor` to run the same
assertions
+against either table type. MOR enables async compaction with a one-delta-commit
+threshold and requires a completed compaction commit on the upsert table before
+the unchanged restart. Each invocation uses an isolated server and filesystem.
+
+Use `--metadata enabled` to exercise the default metadata-enabled
configuration.
+The default `--metadata disabled` isolates the sink assertions from the known
+metadata-writer lifecycle caveat tracked by #20491; a passing disabled run does
+not establish default-configuration coverage. Run both explicitly, for example:
+
+```sh
+# Append these options to the command above for separate runs:
+--table-type mor --metadata disabled
+--table-type cow --metadata enabled
+--table-type mor --metadata enabled
+```
+
+Validated with Spark 4.1.1 / Scala 2.13 / JDK 17 and the candidate bundle built
+from master `b0d16f94b3b`: all four COW/MOR and metadata disabled/enabled
+combinations passed. Both MOR runs completed upsert compaction, and all four
+restart timelines remained unchanged. These bounded runs do not resolve #20491.
+
+This scenario uses one writer per target. It does not claim coverage of
materialized views, full refresh,
+metadata-writer lifetime, fan-in, chaining, or commit/checkpoint failure
injection.
+The generated source, SQL and pipeline specification are retained in the logs
as
+runnable examples within the harness's isolated filesystem layout.
diff --git a/hudi-integ-test/spark-pipelines/phase1_acceptance.py
b/hudi-integ-test/spark-pipelines/phase1_acceptance.py
new file mode 100644
index 000000000000..79ea61edd913
--- /dev/null
+++ b/hudi-integ-test/spark-pipelines/phase1_acceptance.py
@@ -0,0 +1,180 @@
+# 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.
+
+"""Run RFC-112 Phase 1 acceptance using the #20453 Spark CLI/Connect
harness."""
+
+import argparse
+import json
+from pathlib import Path
+import signal
+import sys
+import tempfile
+
+
+SOURCE = '''from pyspark import pipelines as dp
+from pyspark.sql import SparkSession
+
+spark = SparkSession.active()
+
[email protected]_view(name="input_events")
+def input_events():
+ return spark.readStream.schema("id INT, ts LONG, value
STRING").json(spark.conf.get("hudi.test.input"))
+'''
+
+# Only immutable identity belongs in the table properties. The operation is
captured per flow.
+FLOWS = '''SET spark.hoodie.datasource.write.operation=insert;
+CREATE STREAMING TABLE insert_events USING hudi
+TBLPROPERTIES ('type'='{table_type}', 'primaryKey'='id',
'preCombineField'='ts')
+AS SELECT id, ts, value FROM STREAM input_events;
+
+SET spark.hoodie.datasource.write.operation=upsert;
+CREATE STREAMING TABLE upsert_events USING hudi
+TBLPROPERTIES ('type'='{table_type}', 'primaryKey'='id',
'preCombineField'='ts')
+AS SELECT id, ts, value FROM STREAM input_events;
+'''
+
+
+def verify(mode, remote, root):
+ from pyspark.sql import SparkSession
+
+ spark = SparkSession.builder.remote(remote).getOrCreate()
+ try:
+ assert spark.conf.get("spark.hudi.test.run.id") == root.name
+ if mode == "healthy":
+ assert spark.range(3).count() == 3
+ print("PASS: Connect server accepts queries after the failing
flow")
+ return
+ old = (1, 1, "old")
+ new = (1, 2, "new")
+ for table, expected in (
+ ("insert_events", [old, old] if mode == "first" else [old, old,
new]),
+ ("upsert_events", [old] if mode == "first" else [new]),
+ ):
+ actual = [tuple(row) for row in spark.table(f"sdp_test.{table}")
+ .select("id", "ts", "value").orderBy("ts").collect()]
+ assert actual == expected, (table, actual, expected)
+ properties = dict(spark.sql(f"SHOW TBLPROPERTIES
sdp_test.{table}").collect())
+ assert not any("write.operation" in key for key in properties),
properties
+ print(f"Verified {mode}: independent insert/upsert flows through SQL
SET and Spark Connect")
+ finally:
+ spark.stop()
+
+
+def scenario(harness, table_type, metadata):
+ flows = FLOWS.format(table_type=table_type)
+ input_dir = harness.root / "input"
+ input_dir.mkdir()
+ (harness.root / "source.py").write_text(SOURCE)
+ (harness.root / "flows.sql").write_text(flows)
+ spec = {
+ "name": "hudi_flow_settings",
+ "storage": (harness.root / "checkpoints").as_uri(),
+ "catalog": "spark_catalog",
+ "database": "sdp_test",
+ "libraries": [{"glob": {"include": "source.py"}}, {"glob": {"include":
"flows.sql"}}],
+ "configuration": {
+ "hudi.test.input": input_dir.as_uri(),
+ "spark.sql.shuffle.partitions": "2",
+ "spark.sql.pipelines.maxFlowRetryAttempts": "0",
+ "spark.hoodie.metadata.enable": str(metadata).lower(),
+ "spark.hoodie.datasource.compaction.async.enable": str(table_type
== "mor").lower(),
+ "spark.hoodie.compact.inline.max.delta.commits": "1",
+ },
+ }
+ spec_path = harness.root / "spark-pipeline.yml"
+ spec_path.write_text(json.dumps(spec, indent=2))
+ for name in ("source.py", "flows.sql", "spark-pipeline.yml"):
+ (harness.logs / name).write_text((harness.root / name).read_text())
+ for name, timestamp, value in (("first", 1, "old"), ("second", 2, "new")):
+ row = json.dumps({"id": 1, "ts": timestamp, "value": value}) + "\n"
+ (input_dir / f"{name}.json").write_text(row * (2 if name == "first"
else 1))
+ harness.run_pipeline(spec_path, name=f"pipeline-{name}")
+ harness.client(Path(__file__).resolve(), name)
+ def commits():
+ result = {}
+ for table in ("insert_events", "upsert_events"):
+ timeline = harness.root / "warehouse" / "sdp_test.db" / table /
".hoodie" / "timeline"
+ result[table] = sorted(p.name for p in timeline.iterdir()
+ if p.suffix in (".commit", ".deltacommit",
".replacecommit"))
+ assert result[table], f"No completed commits found for {table}"
+ return result
+
+ before = commits()
+ if table_type == "mor":
+ # The upsert update creates a log file and must actually exercise
async compaction.
+ assert any(name.endswith(".commit") for name in
before["upsert_events"]), before
+
+ # Restart with no new input must leave both outputs and the completed
timeline unchanged.
+ harness.run_pipeline(spec_path, name="pipeline-restart")
+ harness.client(Path(__file__).resolve(), "restart")
+ after = commits()
+ assert before == after, (before, after)
+ (harness.logs / "restart-commits.json").write_text(json.dumps({"before":
before, "after": after}, indent=2))
+ # A null record key fails inside the streaming sink, after table creation.
+ failure_input = harness.root / "failure-input"
+ failure_input.mkdir()
+ (failure_input / "bad.json").write_text('{"id": null, "ts": 1, "value":
"bad"}\n')
+ (harness.root / "flows.sql").write_text(flows.split("SET
spark.hoodie.datasource.write.operation=upsert;")[1]
+ .replace("upsert_events",
"failure_events"))
+ spec["name"] = "hudi_failure_isolation"
+ spec["storage"] = (harness.root / "failure-checkpoints").as_uri()
+ spec["configuration"]["hudi.test.input"] = failure_input.as_uri()
+ spec_path.write_text(json.dumps(spec, indent=2))
+ (harness.logs / "failure-flows.sql").write_text((harness.root /
"flows.sql").read_text())
+ (harness.logs / "failure-pipeline.yml").write_text(spec_path.read_text())
+ try:
+ harness.run_pipeline(spec_path, name="pipeline-expected-failure")
+ except RuntimeError:
+ assert harness.server.poll() is None, "Failing flow killed the Connect
server"
+ failure_log = (harness.logs /
"pipeline-expected-failure.log").read_text()
+ assert "recordKey" in failure_log or "record key" in
failure_log.lower(), failure_log[-4000:]
+ else:
+ raise AssertionError("Null record key unexpectedly succeeded")
+ harness.client(Path(__file__).resolve(), "healthy")
+ print("PASS: exact insert/upsert rows, duplicate deduplication, unchanged
restart timeline, failure isolation")
+
+
+def main():
+ if len(sys.argv) == 4 and sys.argv[1] in ("first", "second", "restart",
"healthy"):
+ verify(sys.argv[1], sys.argv[2], Path(sys.argv[3]))
+ return
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("--harness-dir", type=Path,
default=Path(__file__).resolve().parent)
+ parser.add_argument("--bundle", type=Path, required=True)
+ parser.add_argument("--logs", type=Path,
default=Path("target/phase1-acceptance-logs"))
+ parser.add_argument("--table-type", choices=("cow", "mor"), default="cow")
+ parser.add_argument("--metadata", choices=("enabled", "disabled"),
default="disabled")
+ args = parser.parse_args()
+ if not (args.harness_dir / "harness.py").is_file():
+ parser.error("The #20453 harness is required; pass its spark-pipelines
directory with --harness-dir")
+ 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"Phase 1 acceptance logs: {logs}", flush=True)
+
+ def interrupted(signum, frame):
+ raise KeyboardInterrupt(f"Received signal {signum}")
+
+ signal.signal(signal.SIGTERM, interrupted)
+ with ConnectHarness(args.bundle, logs) as harness:
+ scenario(harness, args.table_type, args.metadata == "enabled")
+
+
+if __name__ == "__main__":
+ main()
