Copilot commented on code in PR #12278:
URL: https://github.com/apache/gluten/pull/12278#discussion_r3411568645
##########
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
+ }
Review Comment:
This collects only `AggregateExpression`s whose function is a
`DeclarativeAggregate`, but it silently ignores any non-declarative aggregates
present in `statsColExpr`. That can produce a partial/incorrect synthetic plan
and a false-positive offload decision (leading to the same
`WholeStageTransformer`/`ClassCastException` issues you’re trying to avoid). A
safer approach is to collect all `AggregateExpression`s first and return
`false` if any aggregate function is not `DeclarativeAggregate`.
##########
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:
The method name `toStrcitRule()` looks like a misspelling (likely
`toStrictRule()`). If `toStrcitRule` doesn’t exist, this won’t compile; if it
does exist, it’s easy to call the wrong API in other call sites. Please verify
the intended API and rename/update accordingly (and keep delta33/delta40 in
sync).
##########
cpp/velox/substrait/SubstraitToVeloxExpr.cc:
##########
@@ -217,13 +217,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 enough context to debug which field/path
triggered the failure (e.g., field index/name and the actual `childType`).
Consider including the offending field index (and/or name) and the child type
string in the message so users can pinpoint the unsupported nested reference
quickly.
##########
.github/workflows/delta_spark_ut.yml:
##########
@@ -0,0 +1,622 @@
+# 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 workflow
security posture by allowing deprecated/insecure Node runtimes for Actions.
Since this workflow already uses `actions/*@v4` (Node20-based), it should be
removed unless there’s a demonstrated hard requirement; if a non-node20 action
needs it, prefer upgrading/replacing that action instead.
--
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]
