Copilot commented on code in PR #12278:
URL: https://github.com/apache/gluten/pull/12278#discussion_r3410296182
##########
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())
+ val config = GlutenConfig.get
+ val transformRule = HeuristicTransform.WithRewrites(
+ Validators.newValidator(config, offloads),
+ Seq(PullOutPreProject),
+ offloads)
Review Comment:
`toStrcitRule()` looks like a misspelling (likely `toStrictRule()`). If the
actual API is `toStrictRule`, this will not compile; please verify the method
name on the rule type returned by `OffloadOthers()` and correct it here.
##########
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())
+ val config = GlutenConfig.get
+ val transformRule = HeuristicTransform.WithRewrites(
+ Validators.newValidator(config, offloads),
+ Seq(PullOutPreProject),
+ offloads)
Review Comment:
Same issue as the Delta 4.0 variant: `toStrcitRule()` is likely a typo
(commonly `toStrictRule()`) and may break compilation. Please confirm the
correct method name and update it.
##########
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 user error doesn’t include the actual encountered child type, which can
make debugging difficult. Consider including the child type (and ideally the
field name/index) in the message (e.g., mentioning `childType->toString()` or
its kind) so it’s clear what non-struct type triggered the failure.
##########
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)
Review Comment:
This offloadability check re-constructs a Spark aggregate/projection plan
from `statsColExpr`, which risks drifting from the actual executor-side plan
shapes (and can lead to false positives/negatives as Delta/Spark change). If
there is already a canonical plan builder for the stats aggregation (e.g., the
same logic used by `GlutenDeltaTaskStatsTracker`), consider factoring that into
a shared helper and reusing it here so the driver-side check stays consistent
over time.
##########
.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
Review Comment:
`rm -rf \"$DELTA_DIR\"` is dangerous if `$DELTA_DIR` is ever empty or an
unexpected path (e.g., `/`). Even though CI likely passes a safe value, this
script is in-repo and can be reused manually; add a defensive check to refuse
empty/root (and possibly require it to be under the workspace) before deleting.
--
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]
