Copilot commented on code in PR #12278:
URL: https://github.com/apache/gluten/pull/12278#discussion_r3409726816
##########
backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/stats/GlutenDeltaJobStatsTracker.scala:
##########
@@ -99,6 +110,48 @@ object GlutenDeltaJobStatsTracker extends Logging {
new GlutenDeltaJobStatsFallbackTracker(tracker)
}
+ /**
+ * Returns whether the Delta per-file statistics aggregation can be
offloaded to a Velox
+ * whole-stage transformer. This mirrors the plan that
[[GlutenDeltaTaskStatsTracker]] builds on
+ * the executors: if the aggregation/projection is not supported by Velox it
stays a vanilla
+ * [[ProjectExec]] (i.e. does not collapse into a
[[WholeStageTransformer]]), in which case the
+ * native stats tracker must not be used. Evaluated once on the driver so
the executors never
+ * allocate native resources for a plan that cannot run.
+ */
+ private def canOffloadStats(dataCols: Seq[Attribute], statsColExpr:
Expression): Boolean = {
+ try {
+ val aggregates = statsColExpr.collect {
+ case ae: AggregateExpression if
ae.aggregateFunction.isInstanceOf[DeclarativeAggregate] =>
+ ae
+ }
+ val statsAttrs =
aggregates.flatMap(_.aggregateFunction.aggBufferAttributes)
+ val statsResultAttrs =
aggregates.flatMap(_.aggregateFunction.inputAggBufferAttributes)
+ val aggOp = SortAggregateExec(
+ None,
+ isStreaming = false,
+ None,
+ Seq.empty,
+ aggregates,
+ statsAttrs,
+ 0,
+ statsResultAttrs,
+ StatisticsInputNode(dataCols))
+ val projOp = ProjectExec(statsResultAttrs, aggOp)
+ val offloads = Seq(OffloadOthers()).map(_.toStrcitRule())
Review Comment:
`toStrcitRule()` looks like a misspelling (\"Strcit\"). If the intended
helper is `toStrictRule()` (or similar), this will fail to compile and break
the driver-side gating logic. Please confirm the correct method name and rename
the call (and any related definitions) accordingly.
##########
backends-velox/src-delta40/main/scala/org/apache/spark/sql/delta/stats/GlutenDeltaJobStatsTracker.scala:
##########
@@ -99,6 +110,48 @@ object GlutenDeltaJobStatsTracker extends Logging {
new GlutenDeltaJobStatsFallbackTracker(tracker)
}
+ /**
+ * Returns whether the Delta per-file statistics aggregation can be
offloaded to a Velox
+ * whole-stage transformer. This mirrors the plan that
[[GlutenDeltaTaskStatsTracker]] builds on
+ * the executors: if the aggregation/projection is not supported by Velox it
stays a vanilla
+ * [[ProjectExec]] (i.e. does not collapse into a
[[WholeStageTransformer]]), in which case the
+ * native stats tracker must not be used. Evaluated once on the driver so
the executors never
+ * allocate native resources for a plan that cannot run.
+ */
+ private def canOffloadStats(dataCols: Seq[Attribute], statsColExpr:
Expression): Boolean = {
+ try {
+ val aggregates = statsColExpr.collect {
+ case ae: AggregateExpression if
ae.aggregateFunction.isInstanceOf[DeclarativeAggregate] =>
+ ae
+ }
+ val statsAttrs =
aggregates.flatMap(_.aggregateFunction.aggBufferAttributes)
+ val statsResultAttrs =
aggregates.flatMap(_.aggregateFunction.inputAggBufferAttributes)
+ val aggOp = SortAggregateExec(
+ None,
+ isStreaming = false,
+ None,
+ Seq.empty,
+ aggregates,
+ statsAttrs,
+ 0,
+ statsResultAttrs,
+ StatisticsInputNode(dataCols))
+ val projOp = ProjectExec(statsResultAttrs, aggOp)
+ val offloads = Seq(OffloadOthers()).map(_.toStrcitRule())
Review Comment:
`toStrcitRule()` looks like a misspelling (\"Strcit\"). If the intended
helper is `toStrictRule()` (or similar), this will fail to compile and break
the driver-side gating logic. Please confirm the correct method name and rename
the call (and any related definitions) accordingly.
##########
cpp/velox/substrait/SubstraitToVeloxExpr.cc:
##########
@@ -216,13 +216,29 @@ std::shared_ptr<const core::FieldAccessTypedExpr>
SubstraitVeloxExprConverter::t
auto inputColumnType = inputType;
for (;;) {
auto idx = tmp->field();
- fieldAccess = makeFieldAccessExpr(inputColumnType->nameOf(idx),
inputColumnType->childAt(idx), fieldAccess);
+ VELOX_USER_CHECK(
+ idx >= 0 && static_cast<uint32_t>(idx) < inputColumnType->size(),
+ "Field reference index {} is out of range for the {}-field row
type.",
+ idx,
+ inputColumnType->size());
+ const TypePtr childType = inputColumnType->childAt(idx);
+ fieldAccess = makeFieldAccessExpr(inputColumnType->nameOf(idx),
childType, fieldAccess);
if (!tmp->has_child()) {
break;
}
- inputColumnType = asRowType(inputColumnType->childAt(idx));
+ // Descending into a nested field is only valid when the current child
is
+ // itself a struct/row. For array/map/primitive children (e.g. a field
+ // nested under an array, as in Delta's "updating array type" case)
+ // asRowType() returns null; previously the next loop iteration
+ // dereferenced that null RowType and crashed the process with a
SIGSEGV.
+ // Throw a user error instead so plan validation fails cleanly and the
+ // query falls back to vanilla execution.
+ inputColumnType = asRowType(childType);
+ VELOX_USER_CHECK_NOT_NULL(
+ inputColumnType,
+ "Nested field reference into a non-struct type (e.g. an array or
map element) is not supported.");
Review Comment:
This error message doesn’t include the actual offending type, which makes
debugging plan issues harder (especially for deeply nested schemas). Consider
including the child type (e.g. `childType->toString()`) and/or the field
index/path in the message so users can identify which reference is unsupported.
##########
.github/workflows/delta_spark_ut.yml:
##########
@@ -0,0 +1,502 @@
+# 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.
+
+# Runs Delta Lake's `spark` sbt module unit tests against a Gluten Velox bundle
+# that is built from the source in this repository. The pipeline:
+#
+# 1. Builds the Velox/Gluten native libraries (centos-7 + vcpkg, x86_64).
+# 2. Builds the Gluten Java/Scala jars and assembles the
+# `gluten-velox-bundle-spark<spark>_<scala>-linux_amd64-<version>.jar`
+# fat jar for Spark 4.1 + Scala 2.13 + Java 17 with the Delta profile.
+# 3. Clones delta-io/delta at the requested release tag (default `v4.2.0`),
+# drops the bundle jar into `spark-unified/lib/` only (NOT `spark/lib/`
+# -- see setup-delta.sh for the unmanagedJars scoping rationale),
+# patches Delta's `DeltaSQLCommandTest` to register the Gluten plugin,
+# and runs `sbt spark/test` sharded across the matrix.
+#
+# Limited to Velox + x86 to keep the matrix simple, per the pipeline's purpose
+# of validating Gluten changes against the latest Delta release.
+
+name: Delta Spark UT (Gluten)
+
+on:
+ workflow_dispatch:
+ inputs:
+ delta_ref:
+ description: 'delta-io/delta git ref (tag/branch/SHA) to test against'
+ required: true
+ default: 'v4.2.0'
+ spark_version:
+ description: 'Delta `-DsparkVersion` value (must match the Gluten -P
profile below)'
+ required: true
+ default: '4.1'
+ test_parallelism:
+ description: 'Forked test JVMs per shard (TEST_PARALLELISM_COUNT)'
+ required: true
+ default: '1'
+ update_baseline:
+ description: 'Seed/refresh the known-failures baseline instead of
enforcing it'
+ type: boolean
+ required: false
+ default: false
+ fail_on_fixed:
+ description: 'Fail when a baseline test now passes (keeps the baseline
honest)'
+ type: boolean
+ required: false
+ default: true
+ pull_request:
+ paths:
+ - '.github/workflows/delta_spark_ut.yml'
+ - '.github/workflows/util/delta-spark-ut/**'
+ - 'gluten-delta/**'
+ - 'backends-velox/src-delta40/**/DeltaSQLCommandTest.scala'
+
+env:
+ ACTIONS_ALLOW_USE_UNSECURE_NODE_VERSION: true
Review Comment:
Setting `ACTIONS_ALLOW_USE_UNSECURE_NODE_VERSION: true` weakens the
workflow’s security posture by allowing deprecated/unsecure Node runtimes for
Actions. If this isn’t strictly required (the workflow uses `actions/*@v4`,
which are Node20-based), please remove it; if it is required for another
action, scope it as narrowly as possible (job/step-level) and document the
dependency.
##########
backends-velox/src-delta40/main/scala/org/apache/spark/sql/delta/stats/GlutenDeltaJobStatsTracker.scala:
##########
@@ -99,6 +110,48 @@ object GlutenDeltaJobStatsTracker extends Logging {
new GlutenDeltaJobStatsFallbackTracker(tracker)
}
+ /**
+ * Returns whether the Delta per-file statistics aggregation can be
offloaded to a Velox
+ * whole-stage transformer. This mirrors the plan that
[[GlutenDeltaTaskStatsTracker]] builds on
+ * the executors: if the aggregation/projection is not supported by Velox it
stays a vanilla
+ * [[ProjectExec]] (i.e. does not collapse into a
[[WholeStageTransformer]]), in which case the
+ * native stats tracker must not be used. Evaluated once on the driver so
the executors never
+ * allocate native resources for a plan that cannot run.
+ */
+ private def canOffloadStats(dataCols: Seq[Attribute], statsColExpr:
Expression): Boolean = {
+ try {
+ val aggregates = statsColExpr.collect {
+ case ae: AggregateExpression if
ae.aggregateFunction.isInstanceOf[DeclarativeAggregate] =>
+ ae
+ }
+ val statsAttrs =
aggregates.flatMap(_.aggregateFunction.aggBufferAttributes)
+ val statsResultAttrs =
aggregates.flatMap(_.aggregateFunction.inputAggBufferAttributes)
Review Comment:
The `DeclarativeAggregate` filter can cause `canOffloadStats` to build a
plan that does not match the real stats aggregation when `statsColExpr`
contains non-declarative aggregates. That can produce a false-positive offload
decision and reintroduce the crash you’re trying to avoid (native tracker
selected even though the real plan can’t collapse into a
`WholeStageTransformer`). Recommendation: either (mandatory) return `false` if
*any* aggregate in `statsColExpr` is non-declarative, or (preferred) collect
all `AggregateExpression` instances and let validators determine
support.\n\nAlso, `inputAggBufferAttributes` are buffer-input attributes, not
necessarily the aggregation outputs. Using them as `resultExpressions`/project
list risks constructing an unrealistic plan and misclassifying offloadability.
Prefer using the actual `AggregateExpression` `resultAttribute`s /
`aggregateAttributes` / or mirror the exact `SparkPlan` shape produced in
`GlutenDeltaTaskStatsTracker` so this
driver-side check is faithful.
##########
backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/stats/GlutenDeltaJobStatsTracker.scala:
##########
@@ -99,6 +110,48 @@ object GlutenDeltaJobStatsTracker extends Logging {
new GlutenDeltaJobStatsFallbackTracker(tracker)
}
+ /**
+ * Returns whether the Delta per-file statistics aggregation can be
offloaded to a Velox
+ * whole-stage transformer. This mirrors the plan that
[[GlutenDeltaTaskStatsTracker]] builds on
+ * the executors: if the aggregation/projection is not supported by Velox it
stays a vanilla
+ * [[ProjectExec]] (i.e. does not collapse into a
[[WholeStageTransformer]]), in which case the
+ * native stats tracker must not be used. Evaluated once on the driver so
the executors never
+ * allocate native resources for a plan that cannot run.
+ */
+ private def canOffloadStats(dataCols: Seq[Attribute], statsColExpr:
Expression): Boolean = {
+ try {
+ val aggregates = statsColExpr.collect {
+ case ae: AggregateExpression if
ae.aggregateFunction.isInstanceOf[DeclarativeAggregate] =>
+ ae
+ }
+ val statsAttrs =
aggregates.flatMap(_.aggregateFunction.aggBufferAttributes)
+ val statsResultAttrs =
aggregates.flatMap(_.aggregateFunction.inputAggBufferAttributes)
Review Comment:
The `DeclarativeAggregate` filter can cause `canOffloadStats` to build a
plan that does not match the real stats aggregation when `statsColExpr`
contains non-declarative aggregates. That can produce a false-positive offload
decision and reintroduce the crash you’re trying to avoid (native tracker
selected even though the real plan can’t collapse into a
`WholeStageTransformer`). Recommendation: either (mandatory) return `false` if
*any* aggregate in `statsColExpr` is non-declarative, or (preferred) collect
all `AggregateExpression` instances and let validators determine
support.\n\nAlso, `inputAggBufferAttributes` are buffer-input attributes, not
necessarily the aggregation outputs. Using them as `resultExpressions`/project
list risks constructing an unrealistic plan and misclassifying offloadability.
Prefer using the actual `AggregateExpression` `resultAttribute`s /
`aggregateAttributes` / or mirror the exact `SparkPlan` shape produced in
`GlutenDeltaTaskStatsTracker` so this
driver-side check is faithful.
##########
.github/workflows/util/delta-spark-ut/known-failures.txt:
##########
@@ -0,0 +1,909 @@
+# Known Delta-on-Gluten unit test failures.
+#
+# Baseline of delta-io/delta `spark` ScalaTest tests EXPECTED to fail under the
+# Gluten Velox bundle. The Delta Spark UT (Gluten) workflow enforces this list:
+# a failing test NOT listed here is a regression (fails CI); a listed test that
+# now passes should be removed. Format: <fully.qualified.SuiteName>#<test
name>.
+# Lines starting with '#' are comments. See README.md in this directory.
+#
+# ---------------------------------------------------------------------------
+# PARTIAL BASELINE -- seeded from 15 of 16 shards of run 27490052632
+# (894 known failures across 18003 tests run). Shard 2 hung and is NOT yet
+# included, so its tests will surface as regressions until shard 2 is collected
+# and merged (see MANIFEST/README). Re-run shard 2, then aggregate all 16 and
+# replace this file.
Review Comment:
This baseline is explicitly incomplete, but the workflow defaults to
enforcement (`update_baseline` default false). That means a large set of
*expected* failures (from the missing shard) will be treated as regressions and
will fail CI, making the gate unusable until the baseline is
completed.\n\nMandatory fix before merging/enabling enforcement: either (a)
commit a complete baseline covering all shards, or (b) temporarily change the
workflow defaults for PR runs to seed mode until the baseline is complete, then
flip back to enforcement once the baseline is ready.
##########
.github/workflows/util/delta-spark-ut/setup-delta.sh:
##########
@@ -0,0 +1,141 @@
+#!/usr/bin/env 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.
+
+#
+# Prepares a delta-io/delta clone for running its `spark` module tests with the
+# Gluten (Velox) bundle jar on the classpath.
+#
+# Usage:
+# setup-delta.sh <delta_ref> <delta_dir> <gluten_bundle_jar>
<gluten_repo_root>
+#
+# Arguments:
+# delta_ref - git ref (tag/branch/sha) to check out (e.g. v4.2.0)
+# delta_dir - destination directory for the Delta clone
+# gluten_bundle_jar - path to the gluten-velox-bundle fat jar
+# gluten_repo_root - path to the Gluten repository root (used to locate
+#
backends-velox/src-delta40/.../DeltaSQLCommandTest.scala)
+#
+
+set -euo pipefail
+
+if [ "$#" -ne 4 ]; then
+ echo "Usage: $0 <delta_ref> <delta_dir> <gluten_bundle_jar>
<gluten_repo_root>" >&2
+ exit 1
+fi
+
+DELTA_REF="$1"
+DELTA_DIR="$2"
+GLUTEN_BUNDLE_JAR="$3"
+GLUTEN_ROOT="$4"
+
+if [ ! -f "$GLUTEN_BUNDLE_JAR" ]; then
+ echo "Gluten bundle jar not found: $GLUTEN_BUNDLE_JAR" >&2
+ exit 1
+fi
+
+# Reuse the existing DeltaSQLCommandTest from Gluten's backends-velox module
+# rather than maintaining a separate copy. This file is compiled as part of the
+# unified `spark` project's Test scope, which has the Gluten bundle on its
+# classpath (via spark-unified/lib/), so the typed GlutenConfig /
VeloxDeltaConfig
+# imports resolve correctly.
+PATCH_SOURCE="$GLUTEN_ROOT/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/test/DeltaSQLCommandTest.scala"
+if [ ! -f "$PATCH_SOURCE" ]; then
+ echo "Gluten DeltaSQLCommandTest not found: $PATCH_SOURCE" >&2
+ exit 1
+fi
+
+echo "::group::Cloning delta-io/delta @ ${DELTA_REF}"
+# Shallow clone the requested tag/branch. Fall back to full clone when the ref
is a SHA.
+if ! git clone --depth 1 --branch "$DELTA_REF"
https://github.com/delta-io/delta.git "$DELTA_DIR"; then
+ echo "Shallow clone of ref '${DELTA_REF}' failed, falling back to full
clone."
+ rm -rf "$DELTA_DIR"
+ git clone https://github.com/delta-io/delta.git "$DELTA_DIR"
+ git -C "$DELTA_DIR" checkout "$DELTA_REF"
+fi
+git -C "$DELTA_DIR" --no-pager log -1 --oneline
+echo "::endgroup::"
+
+echo "::group::Injecting Gluten bundle jar onto the spark project's TEST
classpath"
+# The Gluten bundle jar must be on the spark project's TEST runtime classpath
+# (so DeltaSQLCommandTest can load org.apache.gluten.GlutenPlugin by name) but
+# NOT on the COMPILE classpath of `sparkV1`, which is the project that holds
+# Delta's main sources. The bundle's transitive contents include extra symbols
+# under `org.apache.spark.sql` that collide with Delta's main sources -- e.g.
+# MergeOutputGeneration.scala imports both `org.apache.spark.sql._` and
+# `org.apache.spark.sql.delta.ClassicColumnConversions._`, and would then fail
+# with `reference to expression is ambiguous`.
+#
+# sbt auto-scans `<baseDirectory>/lib` via `unmanagedBase`. Two relevant
+# projects in Delta v4.2.0 have a `lib/` baseDirectory:
+# - sparkV1: `project in file("spark")` -> spark/lib
+# - spark : `project in file("spark-unified")` -> spark-unified/lib
+# unmanagedJars are project-scoped (NOT inherited by dependents), so dropping
+# the bundle into spark-unified/lib/ adds it to the unified `spark` project's
+# Compile *and* Test classpaths -- but NOT to sparkV1's. That's exactly what
+# we want:
+# * sparkV1/Compile sees ONLY Delta's regular deps -> Delta main compiles.
+# * spark/Test/fullClasspath sees the bundle -> tests load GlutenPlugin.
+# (Verified empirically: with bundle only in spark-unified/lib/, sbt's
+# `show sparkV1/Compile/dependencyClasspath` excludes the bundle and
+# `show spark/Test/fullClasspath` includes it.)
+#
+# We deliberately do NOT also drop the bundle into spark/lib/, which is what
+# caused the previous compile failure: spark/lib/ is sparkV1's unmanagedBase,
+# and putting the bundle there would re-introduce the ambiguity errors.
+SPARK_UNIFIED_LIB="$DELTA_DIR/spark-unified/lib"
+mkdir -p "$SPARK_UNIFIED_LIB"
+cp "$GLUTEN_BUNDLE_JAR" "$SPARK_UNIFIED_LIB/gluten-velox-bundle.jar"
+ls -lh "$SPARK_UNIFIED_LIB"
+echo "::endgroup::"
+
+echo "::group::Patching DeltaSQLCommandTest to enable Gluten plugin"
+TARGET="$DELTA_DIR/spark/src/test/scala/org/apache/spark/sql/delta/test/DeltaSQLCommandTest.scala"
+if [ ! -f "$TARGET" ]; then
+ echo "Expected file not found in Delta clone: $TARGET" >&2
+ echo "The Delta directory layout for ref '${DELTA_REF}' may have changed."
+ exit 1
+fi
+cp "$PATCH_SOURCE" "$TARGET"
+echo "Patched $TARGET"
+echo "--- diff vs. upstream ---"
+git -C "$DELTA_DIR" --no-pager diff --
"spark/src/test/scala/org/apache/spark/sql/delta/test/DeltaSQLCommandTest.scala"
|| true
+echo "::endgroup::"
+
+echo "::group::Disabling Delta scalastyle HeaderMatchesChecker"
+# Our reused DeltaSQLCommandTest carries Gluten's ASF-only license header,
which
+# does not match Delta's HeaderMatchesChecker regex (the regex expects either a
+# Delta copyright block, or the ASF header followed by a Spark-modifications
+# block and the Delta copyright block). HeaderMatchesChecker is a file-level
+# checker that does NOT honor `// scalastyle:off` directives, so we instead
+# disable it globally in Delta's shared scalastyle-config.xml. The config is
+# applied via `ThisBuild / scalastyleConfig` in project/Checkstyle.scala, so a
+# single edit covers every sbt sub-project.
+SCALASTYLE_CONFIG="$DELTA_DIR/scalastyle-config.xml"
+if [ ! -f "$SCALASTYLE_CONFIG" ]; then
+ echo "Expected scalastyle config not found: $SCALASTYLE_CONFIG" >&2
+ exit 1
+fi
+sed -i \
+ 's|<check level="error" class="org.scalastyle.file.HeaderMatchesChecker"
enabled="true">|<check level="error"
class="org.scalastyle.file.HeaderMatchesChecker" enabled="false">|' \
+ "$SCALASTYLE_CONFIG"
Review Comment:
`sed -i` behavior differs between GNU sed and BSD sed (macOS). If this
script is expected to be runnable outside CI, consider making the in-place edit
portable (e.g. detecting sed flavor or using a temporary file + move) to avoid
local developer breakage.
--
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]
