voonhous commented on code in PR #19674: URL: https://github.com/apache/hudi/pull/19674#discussion_r3834824394
########## packaging/bundle-validation/native_spark/validate.scala: ########## @@ -0,0 +1,85 @@ +/* + * 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. + */ + +import org.apache.spark.sql.SaveMode._ +import org.apache.hudi.DataSourceWriteOptions._ +import org.apache.hudi.config.HoodieWriteConfig._ + +val outputDir = "/tmp/native-spark-bundle" + +// Force a real join rather than a broadcast, so the plan exercises Comet's join, shuffle and sort. +spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1") + +// Deterministic input, so the query result can be asserted exactly. 300 rows spread evenly over +// three partitions, fare equal to the row id. +def rows(from: Int, to: Int) = spark.range(from, to).selectExpr( + "concat('id-', cast(id as string)) as uuid", + "cast(id % 3 as string) as partitionpath", + "cast(id as double) as fare", + "id as ts") + +def write(name: String, tableType: String, mode: org.apache.spark.sql.SaveMode, + df: org.apache.spark.sql.DataFrame): String = { + val path = "file:///tmp/hudi-bundles/tests/" + name + df.write.format("hudi"). + option(PRECOMBINE_FIELD_OPT_KEY, "ts"). + option(RECORDKEY_FIELD_OPT_KEY, "uuid"). + option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath"). + option(TABLE_TYPE_OPT_KEY, tableType). + option(TABLE_NAME, name). + mode(mode). + save(path) + path +} + +// Each partition holds 100 rows per side, so the join emits 100 * 100 rows per partition and each +// left fare is summed 100 times. Aggregate a data column, not just the partition column: a scan +// projecting no data columns reads as ReadSchema struct<> and Comet declines to bridge it. +def probe(label: String, leftPath: String, rightPath: String): Unit = { + spark.read.format("hudi").load(leftPath).createOrReplaceTempView("t1") + spark.read.format("hudi").load(rightPath).createOrReplaceTempView("t2") + val query = spark.sql( + "select t1.partitionpath, count(*) as c, sum(t1.fare) as s from t1 " + + "join t2 on t1.partitionpath = t2.partitionpath group by t1.partitionpath order by t1.partitionpath") + val result = query.collect().map(r => s"${r.get(0)},${r.get(1)},${r.get(2)}") + result.foreach(r => println(s"::warning::native bundle $label row $r")) + sc.parallelize(result, 1).saveAsTextFile(s"$outputDir/${label}_rows") + + // Comet does not recognize Hudi's file format and leaves the scan to Spark, but with + // spark.comet.convert.parquet.enabled it bridges the scan's output into Arrow and runs everything + // above it natively. Copy-on-write keeps the vectorized read and bridges columnar to columnar; + // merge-on-read reads row by row because file group merging is row level. Asserting on the plan + // matters because Comet degrades silently: a mis-relocated Comet or a libcomet.so that failed to + // load still returns correct results. + val plan = query.queryExecution.executedPlan.toString + println(s"::warning::native bundle $label executed plan\n" + plan) + sc.parallelize(Seq(plan), 1).saveAsTextFile(s"$outputDir/${label}_plan") +} + +probe("cow", write("native_cow_1", "COPY_ON_WRITE", Overwrite, rows(0, 300)), + write("native_cow_2", "COPY_ON_WRITE", Overwrite, rows(0, 300))) + +// Merge-on-read with a second commit, so the snapshot read merges base files with log files. +val morLeft = write("native_mor_1", "MERGE_ON_READ", Overwrite, rows(0, 300)) +write("native_mor_1", "MERGE_ON_READ", Append, rows(0, 150)) Review Comment: **major** The append re-writes the same `fare` and `ts` as the base commit, so a snapshot read that ignored log files entirely still returns the copy-on-write numbers -- `expectedRows` is reused verbatim for MOR at `validate.sh:121-127`. The check catches duplication but not a broken base+log merge. Could the append write `cast(id + 1000 as double) as fare` for ids 0-149, with the MOR expectation adjusted, so a log-less read fails? ########## packaging/bundle-validation/validate.sh: ########## @@ -69,6 +75,75 @@ use_default_java_runtime () { export JAVA_HOME=${DEFAULT_JAVA_HOME} } +## +# Function to test the native spark bundle, which carries Apache DataFusion Comet. +# +# Comet ships class file version 61 bytecode and a glibc linked libcomet.so, so this only runs on +# the Java 17 pass. +# +# env vars (defined in container): +# SPARK_HOME: path to the spark directory +## +test_native_spark_bundle () { + local outputDir=/tmp/native-spark-bundle + rm -rf $outputDir + change_java_runtime_version + echo "::warning::validate.sh Writing and querying Hudi tables with Comet enabled" + $SPARK_HOME/bin/spark-shell --jars $JARS_DIR/native-spark.jar \ + --conf 'spark.plugins=org.apache.spark.CometPlugin' \ + --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension,org.apache.comet.CometSparkSessionExtensions' \ + --conf 'spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager' \ + --conf 'spark.comet.enabled=true' \ + --conf 'spark.comet.exec.enabled=true' \ + --conf 'spark.comet.convert.parquet.enabled=true' \ + --conf 'spark.comet.explain.fallback.enabled=true' \ + --conf 'spark.comet.metrics.enabled=true' \ + --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \ + --conf 'spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar' \ + --conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog' < $WORKDIR/native_spark/validate.scala Review Comment: **blocker** `test_spark_hadoop_mr_bundles` ends by killing Derby/Hive (`:201`) and this block runs right after it, so `spark_catalog=HoodieCatalog` fails the first `spark.sql()` on metastore init. That is what reddens the current run, before Comet ever matters: the log shows `ConnectException localhost:1527`, then `cat: can't open .../cow_rows/part-*` and a blank `actual`. The glibc image alone will not turn this lane green. `validate.scala` only does path reads and temp views. Could we drop this conf, or move the block above `test_spark_hadoop_mr_bundles`? ########## packaging/bundle-validation/native_spark/validate.sh: ########## @@ -0,0 +1,72 @@ +#!/bin/bash +# +# 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. +# + +## +# Validates the native spark bundle by writing and querying Hudi tables with Comet enabled. +# +# Must run on glibc with Java 17. Comet's libcomet.so is glibc-linked and its bytecode is class +# file version 61, so this cannot run in the Alpine based bundle-validation image; the caller is +# responsible for providing a suitable environment. +# +# env vars: +# SPARK_HOME: path to the spark directory +# NATIVE_BUNDLE_JAR: path to the hudi native spark bundle jar +## +set -o errexit +set -o nounset + +WORKDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +outputDir=/tmp/native-spark-bundle +rm -rf $outputDir + +echo "::warning::validating native spark bundle with $NATIVE_BUNDLE_JAR" +$SPARK_HOME/bin/spark-shell --jars "$NATIVE_BUNDLE_JAR" \ + --conf 'spark.plugins=org.apache.spark.CometPlugin' \ + --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension,org.apache.comet.CometSparkSessionExtensions' \ + --conf 'spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager' \ + --conf 'spark.comet.enabled=true' \ + --conf 'spark.comet.exec.enabled=true' \ + --conf 'spark.comet.convert.parquet.enabled=true' \ + --conf 'spark.comet.explain.fallback.enabled=true' \ + --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \ + --conf 'spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar' \ + --conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog' \ + < "$WORKDIR/validate.scala" + +# 300 rows per table over three partitions: 100 * 100 joined rows per partition, and each t1 fare +# summed 100 times (100 * 14850, 100 * 14950, 100 * 15050). +expectedRows='0,10000,1485000.0 +1,10000,1495000.0 +2,10000,1505000.0' +actualRows=$(cat $outputDir/rows/part-*) Review Comment: **major** This file has had no caller since `native_spark/ci_run.sh` was deleted, and it has drifted: it reads `$outputDir/rows` and `$outputDir/plan`, while `validate.scala` now writes `cow_rows`/`mor_rows` and `cow_plan`/`mor_plan`. It still ships in the source release and the validation image, next to the `validate.scala` that is used. Could it be deleted along with its two siblings? ########## packaging/bundle-validation/validate.sh: ########## @@ -377,6 +452,17 @@ if [ "$?" -ne 0 ]; then fi echo "::warning::validate.sh done validating spark & hadoop-mr bundle" +if [ -e $JARS_DIR/native-spark.jar ] && [[ ${JAVA_RUNTIME_VERSION} == 'openjdk17' ]]; then Review Comment: **blocker** The function `exit 1`s on failure, so while the native check is red the whole Java 17 pass stops here: cli (`:469`), utilities (`:481`), utilities-slim (`:491`), flink (`:499`), kafka-connect (`:508`) and metaserver (`:515`) never run. In the current run the Java 11 pass validated cli/utilities/slim/deltastreamer; the Java 17 pass validated none of them. Could this block move to the end of the sequence, or set a deferred failure flag instead of exiting mid-way? ########## packaging/bundle-validation/validate.sh: ########## @@ -69,6 +75,75 @@ use_default_java_runtime () { export JAVA_HOME=${DEFAULT_JAVA_HOME} } +## +# Function to test the native spark bundle, which carries Apache DataFusion Comet. +# +# Comet ships class file version 61 bytecode and a glibc linked libcomet.so, so this only runs on +# the Java 17 pass. +# +# env vars (defined in container): +# SPARK_HOME: path to the spark directory +## +test_native_spark_bundle () { + local outputDir=/tmp/native-spark-bundle + rm -rf $outputDir + change_java_runtime_version + echo "::warning::validate.sh Writing and querying Hudi tables with Comet enabled" + $SPARK_HOME/bin/spark-shell --jars $JARS_DIR/native-spark.jar \ + --conf 'spark.plugins=org.apache.spark.CometPlugin' \ + --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension,org.apache.comet.CometSparkSessionExtensions' \ + --conf 'spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager' \ + --conf 'spark.comet.enabled=true' \ + --conf 'spark.comet.exec.enabled=true' \ + --conf 'spark.comet.convert.parquet.enabled=true' \ + --conf 'spark.comet.explain.fallback.enabled=true' \ + --conf 'spark.comet.metrics.enabled=true' \ + --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \ + --conf 'spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar' \ + --conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog' < $WORKDIR/native_spark/validate.scala + + # 300 rows per table over three partitions: 100 * 100 joined rows per partition, and each cow + # fare summed 100 times (100 * 14850, 100 * 14950, 100 * 15050). + local expectedRows='0,10000,1485000.0 +1,10000,1495000.0 +2,10000,1505000.0' + local actualRows + actualRows=$(cat $outputDir/cow_rows/part-*) + if [ "$actualRows" != "$expectedRows" ]; then + echo "::error::validate.sh native spark bundle copy-on-write query returned unexpected results" + echo "expected:"; echo "$expectedRows" + echo "actual:"; echo "$actualRows" + exit 1 + fi + + # Merge-on-read with log files reads row by row, so the same query must still be correct. + local actualMorRows + actualMorRows=$(cat $outputDir/mor_rows/part-*) + if [ "$actualMorRows" != "$expectedRows" ]; then + echo "::error::validate.sh native spark bundle merge-on-read query returned unexpected results" + echo "expected:"; echo "$expectedRows" + echo "actual:"; echo "$actualMorRows" + exit 1 + fi + + # Comet declines what it cannot accelerate and hands it back to Spark, so correct results on + # their own would still pass with a mis-relocated Comet or a libcomet.so that failed to load. + # Copy-on-write keeps the vectorized read and bridges columnar to columnar; merge-on-read reads + # row by row because file group merging is row level, and bridges through a row conversion. + if ! grep -q 'CometSortMergeJoin' $outputDir/cow_plan/part-*; then Review Comment: **major** The copy-on-write branch asserts only `CometSortMergeJoin`, so a COW scan silently dropping from `CometSparkColumnarToColumnar` to the row bridge still passes -- the columnar-vs-row distinction the module README calls out as this bundle's value. The MOR branch does assert its bridge. `supportBatch` on the file group reader has shipped wrong before (HUDI-7068, #10043). Could `cow_plan` also require `CometSparkColumnarToColumnar`? ########## azure-pipelines-20230430.yml: ########## @@ -96,6 +96,7 @@ parameters: - '!packaging/hudi-metaserver-server-bundle' - '!packaging/hudi-presto-bundle' - '!packaging/hudi-spark-bundle' + - '!packaging/hudi-native-spark-bundle' Review Comment: **major** This feeds only `JACOCO_MODULES` (`:123`), which is consumed at `:601` by `-Pcopy-files-for-jacoco`. The reactor install at `:535` has no `-pl`, and `MVN_OPTS_INSTALL` carries `BUILD_PROFILES=-Dspark3.5`, so the Comet uber jar is still shaded on every PR run of UT_FT_10 -- the job capped to `-T 1` for shade OOM in #19119. Verified: `mvn help:evaluate -Dexpression=hudi.native.bundle.skip -Dspark3.5` returns `false`. Could `:535` pass `-Dhudi.native.bundle.skip=true`? -- 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]
