malinjawi commented on code in PR #13165:
URL: https://github.com/apache/gluten/pull/13165#discussion_r4306296355
##########
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:
Confirmed, thanks. `ExprSetSimplified` reaches the counter through
`evalSimplifiedImpl` → `invokeApplyWithListeners`, which skips the
`numProcessedRows` update in `applyFunction`. So a counter at an alias root
(the single-clause insert-only MERGE's `numTargetRowsInserted`, and
`numSourceRows`) read 0. Counters inside `CASE WHEN` still counted, because
`SwitchExpr` evaluates its branches with the regular `eval`.
Fixed in 1b486cca6. For a plan that calls a counter,
`setIncrementMetricQueryConfigs` now forces `expression.eval_simplified=false`
together with `operator_track_expression_stats=true`, after the user overrides.
I chose this over a JVM fallback for two reasons: every
`spark.gluten.sql.columnar.backend.velox.*` override goes through
`overwriteVeloxConf`, and the MERGE write projection stays native.
Tests:
- The `DeltaSuite` regression you described: insert-only MERGE with CDF on,
run with `eval_simplified` false and true. It asserts the table rows,
`numTargetRowsInserted`, `table_changes(t, v, v)`, and that the counter
projection is a `DeltaProjectExecTransformer`. Delta 4.1 counts these inserts
with a filter instead, so the plan check is skipped there.
- A C++ Task test that runs a root counter under `eval_simplified=true` and
`operator_track_expression_stats=false`, and checks the exported count.
##########
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:
Confirmed. With `c = [1, null, 3, null]`, Spark counts 4 and Velox counts 2.
`add` is marked null-propagating, so `evalWithNulls` drops the null rows before
either argument runs.
Nullability alone is not the right test, though. `Add(c, counter)` agrees on
both engines, because Spark also skips the right operand. And `Divide(counter,
d)` disagrees on a zero divisor with no nulls involved. So in 7f5c7a3a0 the
AND/OR denylist became an allowlist of the parents Delta 3.3/4.0 DML actually
places above a counter: `Alias`, `IncrementMetric`, `IF`/`CASE WHEN` values
(not conditions), `named_struct`, `array` and `NOT`. Any other parent keeps the
projection on Spark. `IncrementMetricOffloadSuite` covers your shape, the
same-column variant, `divide`, `coalesce` and counters in conditions, plus
positive cases for each Delta DML shape.
##########
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:
Confirmed. `LambdaExpr` gives `SpecialForm` an empty `inputs()` and keeps
the body in `body_`, so `ExprSet::stats` never reaches the counter.
The allowlist in 7f5c7a3a0 covers this too: a lambda is not an accepted
parent, so a counter inside a `transform` or `filter` body falls back. The
check is shared, so it applies to both Delta 3.3 and 4.0. There are tests for
`transform` and `filter`, plus one for a lambda below a counter, which stays
offloadable because only ancestors are restricted.
--
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]
