felipepessoto commented on code in PR #13165:
URL: https://github.com/apache/gluten/pull/13165#discussion_r4288276334
##########
cpp/velox/compute/WholeStageResultIterator.cc:
##########
@@ -693,6 +704,15 @@ std::unordered_map<std::string, std::string>
WholeStageResultIterator::getQueryC
setIfExists(kOpTraceDirectoryCreateConfig,
velox::core::QueryConfig::kOpTraceDirectoryCreateConfig);
overwriteVeloxConf(veloxCfg_.get(), configs, kDynamicBackendConfPrefix);
+
+ // Only a plan that calls a Delta counter needs project and filter
operators to export their
+ // per-expression processed-row counts (see IncrementMetricFunction.h).
Velox counts the rows
+ // regardless; this only attaches them to the operator stats. Set after
the user overrides: a
+ // query that disabled the stats would report zero modified rows and could
suppress valid
+ // change-data-feed rows.
+ if (trackExpressionStats_) {
+ configs[velox::core::QueryConfig::kOperatorTrackExpressionStats] =
"true";
Review Comment:
[P1] Ensure the evaluator actually records the native counter rows
Could we also prevent counter-bearing plans from using simplified
evaluation, or fall back when reliable counts are unavailable? Setting
`operator_track_expression_stats=true` only enables export. The user can still
set `spark.gluten.sql.columnar.backend.velox.expression.eval_simplified=true`,
which selects `ExprSetSimplified`.
In the pinned Velox revision,
[`Expr::evalSimplifiedImpl`](https://github.com/IBM/velox/blob/6b616f0ae2599ba2aa30bc2d3fcb610a09dc92a5/velox/expression/Expr.cpp#L513-L554)
invokes the function without the `numProcessedRows` accounting performed by
normal evaluation. The counter returns the data, but its statistics remain
empty, and `ProjectMetricsUpdater` credits a missing slot as zero. For an
insert-only MERGE whose CDF is inferred from AddFiles, those zero modified-row
metrics can make batch `table_changes` omit real inserts.
Please add a Velox insert-only MERGE/CDF regression with this setting
enabled, asserting the table rows, persisted inserted-row metric, and
merge-version CDF rows. It should also assert that the counter projection is
native so fallback cannot hide the issue. Any evaluator override needs to be
applied after user overrides, as the export setting is here.
This is source-traced, not runtime-reproduced.
##########
gluten-delta/src-delta33/main/scala/org/apache/gluten/execution/DeltaProjectExecTransformer.scala:
##########
@@ -76,17 +89,83 @@ case class DeltaProjectExecTransformer(projectList:
Seq[NamedExpression], child:
override protected def withNewChildInternal(newChild: SparkPlan):
DeltaProjectExecTransformer =
copy(child = newChild)
+}
+
+object DeltaProjectExecTransformer {
- def genNewProjectList(projectList: Seq[NamedExpression]):
Seq[NamedExpression] = {
- projectList.map {
+ /**
+ * Rewrites the project list for the backend and returns it together with
the metrics it carries,
+ * keyed the way the metrics updater credits them.
+ *
+ * With `nativeCounting`, every [[IncrementMetric]] becomes an
[[IncrementMetricCall]] on one of
+ * the backend's counter slots, one slot per distinct metric of the
projection, so the backend
+ * counts the rows it is evaluated on wherever it sits. A projection with
more distinct metrics
+ * than slots falls back. Counters inside `AND` or `OR` also fall back
because native operand
+ * reordering can change their evaluation count.
+ *
+ * Without it, only the stack of [[IncrementMetric]] at the root of an alias
is removed and
+ * credited with the output row count. An [[IncrementMetric]] anywhere else
is evaluated only for
+ * some rows on Spark, so it cannot be represented by the output row count;
+ * [[org.apache.gluten.extension.OffloadDeltaProject]] keeps such projects
on Spark, and this
+ * method refuses them so validation falls back if one slips through.
+ */
+ private[gluten] def stripIncrementMetrics(
+ projectList: Seq[NamedExpression],
+ nativeCounting: Boolean): (Seq[NamedExpression], Seq[(String,
SQLMetric)]) = {
+ if (nativeCounting) {
+ if (!IncrementMetricOffload.canOffloadProject(projectList,
nativeCounting = true)) {
+ throw new
GlutenNotSupportException(IncrementMetricOffload.nativeConjunctionProjectReason)
+ }
+ return rewriteToNativeCounters(projectList)
+ }
+ val metrics = mutable.ArrayBuffer.empty[(String, SQLMetric)]
+ val stripped = projectList.map {
case alias: Alias =>
- val newChild = alias.child.transformUp {
- case im @ IncrementMetric(child, metric) =>
- extraMetrics :+= (im.prettyName, metric)
- child
+ var expr: Expression = alias.child
+ while (expr.isInstanceOf[IncrementMetric]) {
+ val increment = expr.asInstanceOf[IncrementMetric]
+ metrics += ((increment.prettyName, increment.metric))
+ expr = increment.child
+ }
+ if (containsIncrementMetricExpr(expr)) {
+ throw new
GlutenNotSupportException(IncrementMetricOffload.conditionalProjectReason)
+ }
+ if (expr eq alias.child) alias else
alias.withNewChildren(Seq(expr)).asInstanceOf[Alias]
+ case other =>
+ if (containsIncrementMetricExpr(other)) {
+ throw new
GlutenNotSupportException(IncrementMetricOffload.conditionalProjectReason)
}
- Alias(child = newChild, name = alias.name)(alias.exprId)
+ other
+ }
+ (stripped, metrics.toSeq)
+ }
+
+ private def rewriteToNativeCounters(
+ projectList: Seq[NamedExpression]): (Seq[NamedExpression], Seq[(String,
SQLMetric)]) = {
+ // One slot per distinct metric, in order of first appearance. SQLMetric
has identity equality,
+ // so the same metric reached from several places shares its slot and the
backend sums the rows
+ // of every call to it, which is how Spark increments it too.
+ val slots = mutable.LinkedHashMap.empty[SQLMetric, String]
+ def toCall(expr: Expression): Expression = expr.transformUp {
+ case increment: IncrementMetric =>
+ val functionName = slots.getOrElseUpdate(
+ increment.metric, {
+ if (slots.size >= IncrementMetricCall.maxCounters) {
+ throw new GlutenNotSupportException(
+ s"More than ${IncrementMetricCall.maxCounters} distinct
metrics in one projection " +
+ "cannot be counted natively")
+ }
+ IncrementMetricCall.functionName(slots.size)
+ }
+ )
+ IncrementMetricCall(increment.child, functionName)
Review Comment:
[P2] Lambda-body counter calls are not included in the exported statistics
Could we reject counters inside lambda bodies until their counts can be
exported? The recursive rewrite reaches an `IncrementMetric` inside a supported
`ArrayTransform` lambda, and the plan scanner recognizes it, so this shape can
be offloaded with expression-stat export enabled.
However, the pinned Velox [`ExprSet::stats`
traversal](https://github.com/IBM/velox/blob/6b616f0ae2599ba2aa30bc2d3fcb610a09dc92a5/velox/expression/Expr.cpp#L2197-L2238)
only follows `Expr::inputs()`. `LambdaExpr` keeps its body separately, so the
counter executes for array elements but its slot is absent from the exported
statistics. `ProjectMetricsUpdater` then treats it as zero. Even one array
element per input row gives a nonzero Spark count and a zero reported native
count.
This is a custom Scala-DSL placement, not one I found in the standard Delta
DML builders. A fallback guard plus a regression is sufficient here; there is
no need to implement general lambda counter support in this PR. The same
restriction should apply to the Delta 4.0 implementation.
This is source-traced, not runtime-reproduced.
##########
gluten-delta/src/main/scala/org/apache/gluten/extension/IncrementMetricOffload.scala:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.
+ */
+package org.apache.gluten.extension
+
+import org.apache.gluten.backendsapi.BackendsApiManager
+import
org.apache.gluten.extension.DeltaPostTransformRules.containsIncrementMetricExpr
+
+import org.apache.spark.sql.catalyst.expressions.{Alias, And, Expression,
Literal, NamedExpression, Or}
+import org.apache.spark.sql.types.BooleanType
+
+/**
+ * Decides whether a project or filter carrying Delta's `IncrementMetric` may
be offloaded.
+ *
+ * `IncrementMetric` adds one to a SQL metric every time it is evaluated and
returns its child. The
+ * native side never evaluates it: the Delta project and filter transformers
strip it from the plan
+ * and the metrics updaters credit the metric with the operator's output row
count once the task
+ * finishes. That is exact only when Spark would have evaluated the expression
once per output row:
+ * at the root of a projection alias, or as a filter condition that keeps
every row. An
+ * `IncrementMetric` below a conditional expression, which is where MERGE
keeps the counter of every
+ * `WHEN` clause (`CASE WHEN clause THEN increment ... ELSE
increment(copied)`), fires for the rows
+ * that take its branch only. Offloading it inflated those counters to the
full row count, which
+ * corrupted the operation metrics of every MERGE commit and, through
+ * `CDCReader.shouldSkipFileActionsInCommit`, produced phantom
change-data-feed rows for no-op
+ * merges (GLUTEN-9003).
+ *
+ * A backend that counts natively
+ *
([[org.apache.gluten.backendsapi.BackendSettingsApi.supportNativeIncrementMetric]])
evaluates
+ * each counter as a pass-through function named after its metric and reports
how many rows that
+ * function processed, which is the exact evaluation count: inside a `CASE
WHEN` branch only the
+ * rows that took the branch reach it. Counters inside `AND` or `OR` still
stay on Spark: native
+ * evaluation may reorder the operands and change how many rows reach a
counter. Other backends keep
+ * all conditional shapes on Spark.
+ */
+object IncrementMetricOffload {
+
+ /** Delta's `IncrementMetric`, matched by name so the shared module needs no
Delta class. */
+ private[gluten] def isIncrementMetric(expr: Expression): Boolean = {
+ expr.prettyName == "increment_metric" && expr.children.size == 1
+ }
+
+ /** Removes a stack of `IncrementMetric` wrappers and returns the expression
they wrapped. */
+ private[gluten] def peelIncrementMetrics(expr: Expression): Expression = {
+ if (isIncrementMetric(expr)) peelIncrementMetrics(expr.children.head) else
expr
+ }
+
+ /** Whether the backend can count an expression's evaluations natively. */
+ def nativeCounting: Boolean =
BackendsApiManager.getSettings.supportNativeIncrementMetric()
+
+ /**
+ * Native counters can be offloaded unless an enclosing `AND` or `OR` may
reorder their
+ * evaluation. Without native counting, every `IncrementMetric` must sit at
the root of its alias
+ * (a stack of them counts as the root), so crediting it with the output row
count is exact.
+ */
+ def canOffloadProject(projectList: Seq[NamedExpression], nativeCounting:
Boolean): Boolean = {
+ if (nativeCounting) {
+ !projectList.exists(_.exists {
+ case and: And => containsIncrementMetricExpr(and)
+ case or: Or => containsIncrementMetricExpr(or)
+ case _ => false
Review Comment:
[P2] Nullable ancestors can skip increments even without AND/OR
Could we keep unsafe null-propagating ancestor shapes on Spark as well? This
currently accepts a projection such as:
```scala
Alias(Add(IncrementMetric(Literal(1), metric), nullableColumn), "v")()
```
Spark evaluates the counter on the left before checking the nullable right
operand. Velox's
[`evalWithNulls`](https://github.com/IBM/velox/blob/6b616f0ae2599ba2aa30bc2d3fcb610a09dc92a5/velox/expression/Expr.cpp#L1238-L1287)
can prune rows with a null `nullableColumn` before evaluating either child.
For two input rows, one with a null right operand, Spark would increment twice
while the native counter would see only one row. Making the counter
nondeterministic and disabling its own default-null behavior does not prevent
its ancestor from pruning rows.
This is a custom Catalyst/Scala-DSL placement, not one I found in the
standard Delta DML builders. We do not need general support in this PR, but the
eligibility guard should reject this placement unless equivalent evaluation
counts can be guaranteed. Please add a nullable-input regression for the
accepted/fallback behavior.
This is source-traced, not runtime-reproduced.
--
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]
