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: [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]
