wombatu-kun commented on code in PR #20496: URL: https://github.com/apache/hudi/pull/20496#discussion_r4439995520
########## hudi-integ-test/spark-pipelines/keyed_flows.py: ########## @@ -0,0 +1,133 @@ +# 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 +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'" Review Comment: `PROPERTIES` pins `'type'='cow'`, so a late update or delete marker merged from MOR log files stays uncovered, while RFC-112's keyed-upsert example is `'type' = 'mor'`. Can `TABLES` take a MOR target here, or should KEYED_FLOWS.md list MOR as deferred? ########## hudi-integ-test/spark-pipelines/keyed_flows.py: ########## @@ -0,0 +1,133 @@ +# 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 +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 client(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 == "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()] + assert 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 = "SET spark.hoodie.datasource.write.operation=upsert;\n" + "\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, 1, "a", "old", False), (1, 2, "a", "new", False), (2, 1, "b", "keep", False)], Review Comment: Stage 1 writes the duplicate in ascending `ts`, so keeping the last arrival gives the same row, and no stage sends a lower-order delete marker. Can stage 1 put the higher `ts` first, and a new stage send a late `_hoodie_is_deleted=true` that must leave the newer row in place? ########## hudi-integ-test/spark-pipelines/keyed_flows.py: ########## @@ -0,0 +1,133 @@ +# 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 +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 client(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 == "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()] + assert 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 = "SET spark.hoodie.datasource.write.operation=upsert;\n" + "\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, 1, "a", "old", False), (1, 2, "a", "new", 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, 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, "missing_key"), + ("missing_order", PROPERTIES.replace("'preCombineField'='ts'", "'preCombineField'='missing_order'"), "part", SOURCE, "missing_order"), + ("missing_partition", PROPERTIES, "missing_partition", SOURCE, "missing_partition"), + ("invalid_marker", PROPERTIES, "part", SOURCE.replace("_hoodie_is_deleted BOOLEAN", "_hoodie_is_deleted STRING"), "_hoodie_is_deleted"), + ("changed_identity", PROPERTIES.replace("'primaryKey'='id'", "'primaryKey'='value'"), "part", SOURCE, "recordkey")) + 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() + assert error_text.lower() in log.lower(), (name, log[-5000:]) + assert 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" + assert 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) + with ConnectHarness(args.bundle, logs) as harness: Review Comment: SIGTERM kills this driver without running `ConnectHarness.__exit__`, so the Connect server (in its own session) and the `hudi-sdp-*` temp directory stay behind. Install the same SIGTERM-to-`KeyboardInterrupt` handler as `harness.main` before entering this block. ########## hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala: ########## @@ -354,6 +354,13 @@ 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))) { Review Comment: `PARTITIONED BY (PART)` over a column `part` passes this resolver check, but `newPartitionColumns` keeps `PART`, so hoodie.properties records a field no record has and the flow still writes to the default partition before the read fails (traced from source, not run). Can `newPartitionColumns` take the schema's spelling, for example through `CatalogUtils.normalizePartCols`, with a differently-cased column in the new test? ########## hudi-integ-test/spark-pipelines/keyed_flows.py: ########## @@ -0,0 +1,133 @@ +# 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 +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 client(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 == "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()] + assert 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 = "SET spark.hoodie.datasource.write.operation=upsert;\n" + "\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, 1, "a", "old", False), (1, 2, "a", "new", 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, 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, "missing_key"), + ("missing_order", PROPERTIES.replace("'preCombineField'='ts'", "'preCombineField'='missing_order'"), "part", SOURCE, "missing_order"), + ("missing_partition", PROPERTIES, "missing_partition", SOURCE, "missing_partition"), + ("invalid_marker", PROPERTIES, "part", SOURCE.replace("_hoodie_is_deleted BOOLEAN", "_hoodie_is_deleted STRING"), "_hoodie_is_deleted"), + ("changed_identity", PROPERTIES.replace("'primaryKey'='id'", "'primaryKey'='value'"), "part", SOURCE, "recordkey")) + 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() + assert error_text.lower() in log.lower(), (name, log[-5000:]) Review Comment: Each `missing_*` `error_text` is a substring of the target name `bad_{name}`, so this assert passes on any failure log that names the table, not only on the rejection it is meant to check. Assert the messages themselves, for example `Can't find primaryKey`, `Can't find ordering fields`, `Partition column 'missing_partition'` and `has to be BOOLEAN`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
