You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by "malinjawi (via GitHub)" <gi...@apache.org> on 2026/09/30 08:25:26 UTC
[PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
malinjawi opened a new pull request, #13165:
URL: https://github.com/apache/gluten/pull/13165
## What changes are proposed in this pull request?
Stacked on #13158 (its three commits are included; review the last one here).
#13158 keeps Delta's conditional `IncrementMetric` counters on Spark, which is exact but leaves MERGE's write projection on the JVM. This PR evaluates the counter in Velox with exact counts, so that projection is offloaded again and MERGE's write stage is back to its pre-#13158 shape.
How it counts: Velox already tracks the rows every expression is evaluated on (`numProcessedRows`), `FilterProject` exports those counts per function name (`OperatorStats.expressionStats`, behind `operator_track_expression_stats`), and `switch` evaluates a branch only on the rows that take it. So:
- Each Delta metric gets its own pass-through Velox function, `increment_metric_<metric display name>`, registered on first sight during plan conversion (`IncrementMetricFunction`). Non-deterministic, so Velox never constant-folds, shares or dictionary-peels it; no default null behavior, so null inputs still count.
- `WholeStageResultIterator` turns the config on and serializes `expressionStats` next to `customStats`; `ProjectMetricsUpdater` credits each metric from its own name.
- On the JVM, `IncrementMetricCall` (gluten-substrait) is a `Transformable` pass-through; the Delta project transformer rewrites every `IncrementMetric` into it when the backend reports `supportNativeIncrementMetric`, and `OffloadDeltaProject` then accepts every projection shape. A metric without a usable display name, or two different metrics sharing one name in a projection, falls back. Filters keep #13158's root-stack strip (exact already, and Velox fuses a filter into the project above it, which moves its expression stats). Backends without the capability keep #13158's behavior.
No Velox change.
## How was this patch tested?
- `IncrementMetricOffloadSuite`: counter naming and the unnamed-metric fallback.
- `DeltaSuite` exact-metric tests (MERGE with delete, update and insert clauses; no-op MERGE with CDF; DELETE/UPDATE with and without DVs) now assert that MERGE's `CASE WHEN` projection runs as `DeltaProjectExecTransformer` on Velox.
- JVM compiled on the spark-3.4, spark-3.5 and spark-4.0 Delta profiles; the C++ compiles in CI (this machine cannot rebuild the native library).
- `delta_spark_ut` against the #13158 baseline: expected zero regressions and zero now-passing.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6247501854
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6372976954
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6123322358
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6249479210
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6249429146
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6272823295
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6019795133
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "malinjawi (via GitHub)" <gi...@apache.org>.
malinjawi commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6247631554
Thanks @felipepessoto, all three are fixed: 1b486cca6 for eval_simplified, 7f5c7a3a0 for the placement guard. I also added the scope statement to the description.
One more thing I noticed while reading the code, not reproduced yet: Velox can push a join filter below the numSourceRows counter, so that metric can undercount in some inner-join MERGEs. It's the same on main and isn't changed here, and table rows and CDF aren't affected. I'll raise it separately.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6349764066
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6300328699
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6016237265
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "felipepessoto (via GitHub)" <gi...@apache.org>.
felipepessoto commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6202098240
I support the native-counting approach here. Could you add a targeted regression/guard and clarify the intended scope? I don't think a general-purpose counter redesign needs to be part of this PR.
### Evaluator-mode correctness
Please add a Velox insert-only MERGE regression with CDF enabled and:
```text
spark.gluten.sql.columnar.backend.velox.expression.eval_simplified=true
```
Please assert the resulting table rows, persisted `numTargetRowsInserted`, and commit-bounded `table_changes` output. Also verify that the counter projection actually runs natively, so Spark fallback cannot hide the issue.
The concern is that [`getQueryContextConf`](https://github.com/apache/gluten/blob/0db8d15f237430aaf8072882cc5de3ecdeac31e6/cpp/velox/compute/WholeStageResultIterator.cc#L706-L715) forces expression-stat export on, but still permits simplified evaluation. 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 processed-row accounting done by normal evaluation. The counter can therefore return the correct values while its statistics remain empty, and `ProjectMetricsUpdater` treats a missing slot as zero. For insert-only MERGEs whose CDF is inferred from AddFiles, zero modified-row metrics can suppress real changes.
This is source-traced, not runtime-reproduced. Please confirm with the regression and ensure counter-bearing plans use a stats-producing evaluator after user overrides, or fall back when that guarantee is unavailable.
### Scope clarification in the PR description
Could you explicitly state that this targets the `IncrementMetric` patterns generated by Delta DML, rather than general Spark-equivalent behavior for arbitrary user-constructed Catalyst expressions? For example:
> This PR targets the IncrementMetric patterns generated by Delta DML. General Spark-equivalent behavior for arbitrary user-constructed Catalyst expression trees is outside this PR's scope.
The current eligibility check excludes AND/OR ancestors but also accepts counter-bearing lambdas and nullable-ancestor shapes. Those are not placements emitted by the standard Delta DML builders I traced, so I would treat them as support-boundary questions, not evidence that ordinary MERGE fails under default settings. Please clarify whether they are supported and why they cannot occur in supported plans; if they can reach native execution, they need safe fallback or exact accounting. The description should not promise fallback that the code does not enforce.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "malinjawi (via GitHub)" <gi...@apache.org>.
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:
Good catch, confirmed. Simplified evaluation skips the numProcessedRows update, so a counter at the alias root reported 0, which covers the insert-only MERGE case.
Fixed in 1b486cca6: when a plan has a counter, we now force eval_simplified=false after user overrides. I also added the regression you described: an insert-only MERGE with CDF, run with eval_simplified off and on, checking rows, numTargetRowsInserted, table_changes, and that the projection runs natively. That last check is skipped on Delta 4.1, which counts these inserts with a filter.
##########
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, Add(counter, nullableCol) undercounts in Velox. Nullability alone isn't quite the right check, though: Divide can diverge with no nulls at all, since Spark skips the dividend when the divisor is zero.
So in 7f5c7a3a0 I replaced the AND/OR check with an allowlist of the parents Delta DML actually generates: Alias, IncrementMetric, IF/CASE WHEN values, named_struct, array, NOT. Anything else stays on Spark. Tests cover your case and a few others.
##########
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. The allowlist in 7f5c7a3a0 covers this too: lambdas aren't an allowed parent, so a counter inside a lambda body falls back. It's the same shared check for Delta 3.3 and 4.0, and there are tests for transform and filter.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6124643389
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-5907353885
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-5907255815
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "malinjawi (via GitHub)" <gi...@apache.org>.
malinjawi commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6290437580
@felipepessoto I split this up to make it easier to review:
1. #13357: native counter functions (C++)
2. #13358: carries the counts into the SQL metrics (JVM)
3. this PR: the Delta wiring. Only the last commit is new: https://github.com/apache/gluten/pull/13165/commits/6ae58c105ca60b466aae42dc12a912a7d6a56f1f
Feel free to have another look.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "felipepessoto (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6020245146
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "felipepessoto (via GitHub)" <gi...@apache.org>.
felipepessoto commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6376783350
The initial estimate when I ran the CI with increment metric offload disabled was fixing 225 tests. The PR removed only 179, do you know if it is because my baseline became stale after Delta 4.4 update? Or do we still have some cases to fix?
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "malinjawi (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-5908359331
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-5907712184
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6275170617
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6294090243
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "malinjawi (via GitHub)" <gi...@apache.org>.
malinjawi commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6414714080
> The initial estimate when I ran the CI with increment metric offload disabled was fixing 225 tests. The PR removed only 179, do you know if it is because my baseline became stale after Delta 4.4 update? Or do we still have some cases to fix?
>
> In my experiment disabling the offload it would also avoid some other unrelated bugs
Good question @felipepessoto
Its not the Delta 4.4 update. Out of your 225:
- 157 are removed in this PR
- 54 were already fixed on main by #13198 (I dropped them from the baseline in #13252)
- 14 still fail, all MergeIntoNestedStructInMapEvolution. With offload disabled that projection runs on Spark, but here it stays in Velox, and Velox's transform_keys rejects map keys with a null field ("map key cannot be indeterminate"). That happens when schema evolution adds a field to a struct map key. Spark is fine with it, so it's one of those unrelated bugs. I'll follow up on it separately.
This PR also drops 22 that weren't in your list (20 ImplicitMergeCasting overflow tests and the 2 no-op MERGE CDC stream tests), which is how it gets to 179.
Do you think there is a mistake?
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-5912467736
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6019978391
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6020673129
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6130430897
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6232916361
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-9003][VL] Count Delta's IncrementMetric natively [gluten]
Posted by "malinjawi (via GitHub)" <gi...@apache.org>.
malinjawi commented on PR #13165:
URL: https://github.com/apache/gluten/pull/13165#issuecomment-6247504249
Thanks @felipepessoto. All three findings are confirmed and addressed:
- P1 (1b486cca6): plans that call a counter now force `expression.eval_simplified=false` after user overrides. I added the regression you described.
- Both P2s (7f5c7a3a0): the guard is now an allowlist of the parents Delta DML generates. Nullable ancestors, lambdas, conditions, AND/OR and everything else fall back.
The description now has the scope statement. While checking I found one more gap, which already exists on main and which this PR does not change. Velox can push a hash-join dynamic filter below the projection that counts `numSourceRows`, so that metric can undercount for an inner-join MERGE with a broadcast target. Only that metric is affected; table rows and CDF are not. I noted it in the description and will raise it separately.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org