pedrumj2 commented on PR #13274: URL: https://github.com/apache/gluten/pull/13274#issuecomment-6200850778
> LGTM. Thanks. > > Just to confirm this fix also applies to `spark.gluten.sql.columnar.backend.velox.nativeUDF.bypassRegistration=false` right? Thanks for the review @marin-ma. Yes it I believe it would also fix the `bypassRegistration=false` path as well. Repro below #### [step1] Get the source and Spark ```bash git clone https://github.com/apache/gluten.git git -C gluten fetch origin pull/13274/head:pr curl -LO https://archive.apache.org/dist/spark/spark-3.5.5/spark-3.5.5-bin-hadoop3.tgz tar -xzf spark-3.5.5-bin-hadoop3.tgz ``` #### [step2] Add a Hive UDF, its native implementation registered under the Hive UDF's class name, and the query ```bash mkdir -p hiveudf/com/example/udf hiveudf/classes cat > hiveudf/com/example/udf/SumRow.java <<'EOF' package com.example.udf; import org.apache.hadoop.hive.ql.exec.UDFArgumentException; import org.apache.hadoop.hive.ql.metadata.HiveException; import org.apache.hadoop.hive.ql.udf.generic.GenericUDF; import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector; import org.apache.hadoop.hive.serde2.objectinspector.StructField; import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector; import org.apache.hadoop.hive.serde2.objectinspector.primitive.LongObjectInspector; import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory; public class SumRow extends GenericUDF { private StructObjectInspector rowOI; private StructField a; private StructField b; @Override public ObjectInspector initialize(ObjectInspector[] args) throws UDFArgumentException { rowOI = (StructObjectInspector) args[0]; a = rowOI.getStructFieldRef("a"); b = rowOI.getStructFieldRef("b"); return PrimitiveObjectInspectorFactory.javaLongObjectInspector; } @Override public Object evaluate(DeferredObject[] args) throws HiveException { Object row = args[0].get(); if (row == null) { return null; } Object av = rowOI.getStructFieldData(row, a); Object bv = rowOI.getStructFieldData(row, b); if (av == null || bv == null) { return null; } return ((LongObjectInspector) a.getFieldObjectInspector()).get(av) + ((LongObjectInspector) b.getFieldObjectInspector()).get(bv); } @Override public String getDisplayString(String[] children) { return "sum_row(" + children[0] + ")"; } } EOF cat > gluten/cpp/velox/udf/examples/HiveStructUdf.cc <<'EOF' #include <velox/functions/Macros.h> #include <velox/functions/Registerer.h> #include "udf/Udf.h" using namespace facebook::velox; namespace { template <typename T> struct SumRowFunction { VELOX_DEFINE_FUNCTION_TYPES(T); FOLLY_ALWAYS_INLINE bool callNullable(int64_t& result, const arg_type<Row<int64_t, int64_t>>* row) { if (row == nullptr) { return false; } const auto a = row->template at<0>(); const auto b = row->template at<1>(); if (!a.has_value() || !b.has_value()) { return false; } result = a.value() + b.value(); return true; } }; const char* kName = "com.example.udf.SumRow"; const char* kArgs[1] = {"struct<a:bigint,b:bigint>"}; } // namespace DEFINE_GET_NUM_UDF { return 1; } DEFINE_GET_UDF_ENTRIES { udfEntries[0] = {kName, "bigint", 1, kArgs, false, false}; } DEFINE_REGISTER_UDF { registerFunction<SumRowFunction, int64_t, Row<int64_t, int64_t>>({kName}); } EOF cat >> gluten/cpp/velox/udf/examples/CMakeLists.txt <<'EOF' add_library(hivestructudf SHARED "HiveStructUdf.cc") target_link_libraries(hivestructudf velox) EOF cat > repro.sql <<'EOF' CREATE TEMPORARY FUNCTION sum_row AS 'com.example.udf.SumRow'; SELECT sum_row(named_struct('a', id, 'b', id + 1)) AS sum_ab FROM range(1, 4, 2); EXPLAIN SELECT sum_row(named_struct('a', id, 'b', id + 1)) AS sum_ab FROM range(1, 4, 2); EOF docker run --rm -v "$PWD":/work:z -w /work apache/gluten:centos-9-jdk8 bash -c ' javac -cp "spark-3.5.5-bin-hadoop3/jars/*" -d hiveudf/classes hiveudf/com/example/udf/SumRow.java jar cf hiveudf.jar -C hiveudf/classes .' ``` #### [step3] Build Gluten with the UDF examples ```bash docker run --rm -v "$PWD":/work:z -w /work/gluten apache/gluten:centos-9-jdk8 bash -c ' source /opt/rh/gcc-toolset-12/enable export VELOX_BUILD_SHARED=ON ./dev/builddeps-veloxbe.sh --build_examples=ON --build_arrow=OFF --run_setup_script=OFF ./build/mvn clean install -Pbackends-velox -Pspark-3.5 -DskipTests' ``` #### [step4] Run the query with `bypassRegistration=false` ```bash docker run --rm -v "$PWD":/work:z -w /work apache/gluten:centos-9-jdk8 \ spark-3.5.5-bin-hadoop3/bin/spark-sql --master 'local[2]' \ --jars gluten/package/target/gluten-velox-bundle-spark3.5_2.12-linux_amd64-1.8.0-SNAPSHOT.jar,hiveudf.jar \ --driver-class-path gluten/package/target/gluten-velox-bundle-spark3.5_2.12-linux_amd64-1.8.0-SNAPSHOT.jar:hiveudf.jar \ --conf spark.plugins=org.apache.gluten.GlutenPlugin \ --conf spark.memory.offHeap.enabled=true \ --conf spark.memory.offHeap.size=2g \ --conf spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager \ --conf spark.gluten.sql.columnar.backend.velox.udfLibraryPaths=file:///work/gluten/cpp/build/velox/udf/examples/libhivestructudf.so \ --conf spark.gluten.sql.columnar.backend.velox.driver.udfLibraryPaths=file:///work/gluten/cpp/build/velox/udf/examples/libhivestructudf.so \ --conf spark.gluten.sql.columnar.backend.velox.nativeUDF.bypassRegistration=false \ --hiveconf hive.cli.errors.ignore=true \ -f repro.sql ``` #### Before: the base commit ```bash git -C gluten checkout --detach pr~1 ``` Then [step2], [step3] and [step4]: ``` 26/10/15 19:35:03 WARN GlutenFallbackReporter: Validation failed for plan: Project[QueryId=1], due to: - Validation failed with exception from: ProjectExecTransformer, reason: UDF com.example.udf.SumRow -> struct<a:bigint,b:bigint> is not registered. 3 7 ... == Physical Plan == *(1) Project [HiveGenericUDF#com.example.udf.SumRow(named_struct(a, id#9L, b, (id#9L + 1))) AS sum_ab#3L] +- *(1) ColumnarToRow +- ColumnarRange 1, 4, 2, 2, 2, [id#9L] ``` #### After: this pull request ```bash git -C gluten checkout pr ``` Then [step3] and [step4]: ``` 3 7 ... == Physical Plan == VeloxColumnarToRow +- ^(1) ProjectExecTransformer [HiveGenericUDF#com.example.udf.SumRow(named_struct(a, id#9L, b, (id#9L + 1))) AS sum_ab#3L] +- ^(1) InputIteratorTransformer[id#9L] +- ArrowColumnarToVeloxColumnar +- OffloadArrowData +- ColumnarRange 1, 4, 2, 2, 2, [id#9L] ``` -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
