yihua commented on code in PR #19674:
URL: https://github.com/apache/hudi/pull/19674#discussion_r4201841473
##########
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:
Fixed, though the trigger is the Hudi read rather than this conf:
`HoodieFileIndex.getConfigProperties` calls `spark.catalog.tableExists` on
every path read, so with the conf dropped the session still initialised the
Hive metastore and hit the same `ConnectException` (reproduced in a glibc
container with the Derby server down). The session now uses
`spark.sql.catalogImplementation=in-memory`, and the `HoodieCatalog` conf is
gone.
##########
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:
Done, the native check now runs last in the sequence.
##########
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:
Done, the append writes `fare` and `ts` offset by 1000 for ids 0-149 and the
merge-on-read expectation is adjusted; a `read_optimized` read of the same
tables returns the copy-on-write sums and fails the check.
##########
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:
Done, `cow_plan` now requires `CometSparkColumnarToColumnar` and rejects
`CometSparkRowToColumnar`; with the vectorized reader disabled the join still
plans as `CometSortMergeJoin` and the new check fails.
##########
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:
Done, `-Dhudi.native.bundle.skip=true` on the install at :535 (`Shading has
been skipped.`, nothing installed).
##########
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:
Done, deleted.
--
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]
