This is an automated email from the ASF dual-hosted git repository.
zhouyuan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 4fbd758dee [GLUTEN-9003][VL] Add native counter functions for Delta's
IncrementMetric (#13357)
4fbd758dee is described below
commit 4fbd758dee9424ffd1c1c1c53b3eb00679d044bd
Author: Mohammad Linjawi <[email protected]>
AuthorDate: Fri Nov 6 15:36:55 2026 +0300
[GLUTEN-9003][VL] Add native counter functions for Delta's IncrementMetric
(#13357)
* [GLUTEN-9003][VL] Add native counter functions for Delta's IncrementMetric
---
cpp/velox/CMakeLists.txt | 1 +
cpp/velox/compute/WholeStageResultIterator.cc | 27 +++-
cpp/velox/compute/WholeStageResultIterator.h | 3 +
.../operators/functions/IncrementMetricFunction.cc | 153 +++++++++++++++++++
.../operators/functions/IncrementMetricFunction.h | 75 +++++++++
.../functions/RegistrationAllFunctions.cc | 3 +
cpp/velox/substrait/SubstraitToVeloxPlan.cc | 8 +-
cpp/velox/tests/CMakeLists.txt | 2 +-
cpp/velox/tests/IncrementMetricFunctionTest.cc | 170 +++++++++++++++++++++
9 files changed, 436 insertions(+), 6 deletions(-)
diff --git a/cpp/velox/CMakeLists.txt b/cpp/velox/CMakeLists.txt
index 23dee4ac72..6500d5c446 100644
--- a/cpp/velox/CMakeLists.txt
+++ b/cpp/velox/CMakeLists.txt
@@ -177,6 +177,7 @@ set(VELOX_SRCS
operators/functions/delta/DeltaBitmapAggregator.cc
operators/functions/overlay/GetMapValue.cc
operators/functions/overlay/RegisterFunctionOverlay.cc
+ operators/functions/IncrementMetricFunction.cc
operators/functions/RowConstructorWithNull.cc
operators/functions/SparkExprToSubfieldFilterParser.cc
operators/plannodes/RowVectorStream.cc
diff --git a/cpp/velox/compute/WholeStageResultIterator.cc
b/cpp/velox/compute/WholeStageResultIterator.cc
index 71e0d20a94..0fae61ec00 100644
--- a/cpp/velox/compute/WholeStageResultIterator.cc
+++ b/cpp/velox/compute/WholeStageResultIterator.cc
@@ -27,6 +27,7 @@
#include "compute/delta/DeltaSplit.h"
#include "compute/delta/DeltaSplitInfo.h"
#include "config/VeloxConfig.h"
+#include "operators/functions/IncrementMetricFunction.h"
#include "utils/ConfigExtractor.h"
#include "velox/connectors/hive/HiveConfig.h"
#include "velox/connectors/hive/HiveConnectorSplit.h"
@@ -106,6 +107,7 @@ WholeStageResultIterator::WholeStageResultIterator(
taskInfo_(taskInfo),
executor_(executor),
veloxPlan_(planNode),
+ usesIncrementMetricFunctions_(usesIncrementMetricFunctions(*planNode)),
spillExecutor_(spillExecutor),
connectorIds_(std::move(connectorIds)),
scanNodeIds_(scanNodeIds),
@@ -511,8 +513,7 @@ void WholeStageResultIterator::collectMetrics() {
customStats[customMetric.first] = folly::dynamic::object("sum",
customMetric.second.sum)(
"count", customMetric.second.count)("min",
customMetric.second.min)("max", customMetric.second.max);
}
-
- operatorStats.push_back(folly::dynamic::object("inputRows",
opStats->inputRows)(
+ folly::dynamic operatorJson = folly::dynamic::object("inputRows",
opStats->inputRows)(
"inputVectors", opStats->inputVectors)("inputBytes",
opStats->inputBytes)(
"rawInputRows", opStats->rawInputRows)("rawInputBytes",
opStats->rawInputBytes)(
"outputRows", opStats->outputRows)("outputVectors",
opStats->outputVectors)(
@@ -521,7 +522,19 @@ void WholeStageResultIterator::collectMetrics() {
"numMemoryAllocations",
opStats->numMemoryAllocations)("spilledInputBytes", opStats->spilledInputBytes)(
"spilledBytes", opStats->spilledBytes)("spilledRows",
opStats->spilledRows)(
"spilledPartitions", opStats->spilledPartitions)("spilledFiles",
opStats->spilledFiles)(
- "physicalWrittenBytes",
opStats->physicalWrittenBytes)("customStats", customStats));
+ "physicalWrittenBytes",
opStats->physicalWrittenBytes)("customStats", customStats);
+ // Rows processed per Delta counter function, keyed by its slot name.
Only FilterProject
+ // operators export these, and only when the plan calls a counter, so
other operators get no key.
+ folly::dynamic expressionStats = folly::dynamic::object();
+ for (const auto& exprStats : opStats->expressionStats) {
+ if (isIncrementMetricFunction(exprStats.first)) {
+ expressionStats[exprStats.first] = exprStats.second.numProcessedRows;
+ }
+ }
+ if (!expressionStats.empty()) {
+ operatorJson["expressionStats"] = std::move(expressionStats);
+ }
+ operatorStats.push_back(std::move(operatorJson));
}
statsNum += static_cast<unsigned int>(operatorStats.size());
@@ -711,6 +724,14 @@ 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 reads its counts from the
per-expression processed-row
+ // stats (see IncrementMetricFunction.h). Set after the user overrides: a
query that disabled
+ // the stats, or switched to simplified evaluation, which does not count,
would report zero
+ // modified rows and could suppress valid change-data-feed rows.
+ if (usesIncrementMetricFunctions_) {
+ setIncrementMetricQueryConfigs(configs);
+ }
} catch (const std::invalid_argument& err) {
std::string errDetails = err.what();
throw std::runtime_error("Invalid conf arg: " + errDetails);
diff --git a/cpp/velox/compute/WholeStageResultIterator.h
b/cpp/velox/compute/WholeStageResultIterator.h
index e7febfdadf..d1633bbb0b 100644
--- a/cpp/velox/compute/WholeStageResultIterator.h
+++ b/cpp/velox/compute/WholeStageResultIterator.h
@@ -133,6 +133,9 @@ class WholeStageResultIterator : public
SplitAwareColumnarBatchIterator {
std::unique_ptr<facebook::velox::exec::TaskCursor> cursor_;
facebook::velox::exec::Task* task_ = nullptr;
std::shared_ptr<const facebook::velox::core::PlanNode> veloxPlan_;
+ /// Whether the plan calls a Delta counter and so needs the query configs
set by
+ /// setIncrementMetricQueryConfigs().
+ const bool usesIncrementMetricFunctions_;
/// Spill.
std::string spillStrategy_;
diff --git a/cpp/velox/operators/functions/IncrementMetricFunction.cc
b/cpp/velox/operators/functions/IncrementMetricFunction.cc
new file mode 100644
index 0000000000..509ac0b5a3
--- /dev/null
+++ b/cpp/velox/operators/functions/IncrementMetricFunction.cc
@@ -0,0 +1,153 @@
+/*
+ * 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.
+ */
+
+#include "operators/functions/IncrementMetricFunction.h"
+
+#include <charconv>
+#include <memory>
+#include <optional>
+#include <string_view>
+#include <system_error>
+#include <unordered_set>
+#include <vector>
+
+#include "velox/core/Expressions.h"
+#include "velox/core/QueryConfig.h"
+#include "velox/expression/EvalCtx.h"
+#include "velox/expression/FunctionSignature.h"
+#include "velox/expression/VectorFunction.h"
+
+namespace gluten {
+namespace {
+
+using namespace facebook::velox;
+
+/// Returns its only argument. See the header for why the name, not the body,
carries the meaning.
+class IncrementMetricFunction : public exec::VectorFunction {
+ public:
+ void apply(
+ const SelectivityVector& rows,
+ std::vector<VectorPtr>& args,
+ const TypePtr& /*outputType*/,
+ exec::EvalCtx& context,
+ VectorPtr& result) const override {
+ VELOX_CHECK_EQ(args.size(), 1);
+ context.moveOrCopyResult(args[0], rows, result);
+ }
+};
+
+void collectIncrementMetricCalls(const core::ITypedExpr& expr,
std::unordered_set<std::string>& names) {
+ if (const auto* call = dynamic_cast<const core::CallTypedExpr*>(&expr)) {
+ if (isIncrementMetricFunction(call->name())) {
+ names.insert(call->name());
+ }
+ }
+ if (const auto* lambda = dynamic_cast<const core::LambdaTypedExpr*>(&expr)) {
+ if (lambda->body() != nullptr) {
+ collectIncrementMetricCalls(*lambda->body(), names);
+ }
+ return;
+ }
+ for (const auto& input : expr.inputs()) {
+ if (input != nullptr) {
+ collectIncrementMetricCalls(*input, names);
+ }
+ }
+}
+
+void collectIncrementMetricCallsInProjects(const core::PlanNode& plan,
std::unordered_set<std::string>& names) {
+ if (const auto* project = dynamic_cast<const core::ProjectNode*>(&plan)) {
+ for (const auto& projection : project->projections()) {
+ if (projection != nullptr) {
+ collectIncrementMetricCalls(*projection, names);
+ }
+ }
+ }
+ for (const auto& source : plan.sources()) {
+ if (source != nullptr) {
+ collectIncrementMetricCallsInProjects(*source, names);
+ }
+ }
+}
+
+} // namespace
+
+std::string incrementMetricFunctionName(int32_t slot) {
+ return std::string(kIncrementMetricFunctionPrefix) + std::to_string(slot);
+}
+
+bool isIncrementMetricFunction(const std::string& functionName) {
+ const std::string_view prefix{kIncrementMetricFunctionPrefix};
+ if (functionName.size() <= prefix.size() || functionName.compare(0,
prefix.size(), prefix) != 0) {
+ return false;
+ }
+ int32_t slot = -1;
+ const char* const end = functionName.data() + functionName.size();
+ const auto [parsedEnd, error] = std::from_chars(functionName.data() +
prefix.size(), end, slot);
+ // The last check rejects spellings that parse to a slot but are not its
name, such as a leading
+ // zero.
+ return error == std::errc() && parsedEnd == end && slot >= 0 && slot <
kIncrementMetricFunctionSlots &&
+ functionName == incrementMetricFunctionName(slot);
+}
+
+void registerIncrementMetricFunctions() {
+ // Not deterministic: keeps Velox from constant-folding or
dictionary-peeling the call, either of
+ // which would change how many rows it is evaluated on. No default null
behavior: a null input
+ // still counts, as it does on Spark.
+ const auto metadata =
exec::VectorFunctionMetadataBuilder().deterministic(false).defaultNullBehavior(false).build();
+ for (int32_t slot = 0; slot < kIncrementMetricFunctionSlots; ++slot) {
+ exec::registerVectorFunction(
+ incrementMetricFunctionName(slot),
+
{exec::FunctionSignatureBuilder().typeVariable("T").returnType("T").argumentType("T").build()},
+ std::make_unique<IncrementMetricFunction>(),
+ metadata);
+ }
+}
+
+bool usesIncrementMetricFunctions(const core::PlanNode& plan) {
+ std::unordered_set<std::string> names;
+ collectIncrementMetricCallsInProjects(plan, names);
+ return !names.empty();
+}
+
+void checkIncrementMetricFunctionsInProjects(
+ const std::unordered_map<uint64_t, std::string>& functionMap,
+ const core::PlanNode& plan) {
+ std::optional<std::unordered_set<std::string>> calledInProjects;
+ for (const auto& entry : functionMap) {
+ // Declarations are spelled `<name>:<argument types>`.
+ const auto name = entry.second.substr(0, entry.second.find(':'));
+ if (!isIncrementMetricFunction(name)) {
+ continue;
+ }
+ if (!calledInProjects.has_value()) {
+ calledInProjects.emplace();
+ collectIncrementMetricCallsInProjects(plan, *calledInProjects);
+ }
+ VELOX_CHECK(
+ calledInProjects->count(name) > 0,
+ "Delta counter {} is declared but no projection calls it, and the JVM
credits counters only from projections",
+ name);
+ }
+}
+
+void setIncrementMetricQueryConfigs(std::unordered_map<std::string,
std::string>& configs) {
+ configs[core::QueryConfig::kOperatorTrackExpressionStats] = "true";
+ configs[core::QueryConfig::kExprEvalSimplified] = "false";
+}
+
+} // namespace gluten
diff --git a/cpp/velox/operators/functions/IncrementMetricFunction.h
b/cpp/velox/operators/functions/IncrementMetricFunction.h
new file mode 100644
index 0000000000..618f2d2a27
--- /dev/null
+++ b/cpp/velox/operators/functions/IncrementMetricFunction.h
@@ -0,0 +1,75 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <string>
+#include <unordered_map>
+
+#include "velox/core/PlanNode.h"
+
+namespace gluten {
+
+/// Delta's `IncrementMetric` counts how many rows an expression is evaluated
on. Velox already
+/// keeps that count per expression (`ExprStats::numProcessedRows`) on its
regular evaluation path
+/// and the FilterProject operator exports it per function name when asked to,
so a counter is a
+/// pass-through function whose name is the key the JVM credits the metric
from. Inside a CASE WHEN
+/// branch that is the number of rows that took the branch, at the top of a
projection it is every
+/// row, which is exactly how Spark evaluates `IncrementMetric`.
+///
+/// Counters only appear in projections: the JVM emits them from Delta's
project transformer alone
+/// (the filter transformer strips its counters and credits the filter's
output rows), and a
+/// Substrait ProjectRel always becomes a ProjectNode.
checkIncrementMetricFunctionsInProjects()
+/// enforces this.
+///
+/// The names are fixed slots, `increment_metric_<slot>`, registered once at
startup. The JVM
+/// assigns one slot per distinct metric of a projection, so a metric needs no
name of its own and
+/// nothing is registered while queries run.
+constexpr const char* kIncrementMetricFunctionPrefix = "increment_metric_";
+
+/// Number of counter slots registered. Mirrored by
`IncrementMetricCall.maxCounters` on the JVM.
+constexpr int32_t kIncrementMetricFunctionSlots = 32;
+
+/// Name of the counter function for `slot`.
+std::string incrementMetricFunctionName(int32_t slot);
+
+/// True if `functionName` is one of the registered slots,
`increment_metric_0` to
+/// `increment_metric_<kIncrementMetricFunctionSlots - 1>`.
+bool isIncrementMetricFunction(const std::string& functionName);
+
+/// Registers every counter slot. Called from registerAllFunctions().
+void registerIncrementMetricFunctions();
+
+/// True if a projection in `plan` calls a counter, which is when the query
needs
+/// setIncrementMetricQueryConfigs().
+bool usesIncrementMetricFunctions(const facebook::velox::core::PlanNode& plan);
+
+/// Throws unless every counter declared in `functionMap`, the Substrait
plan's function
+/// declarations, is called by a projection of `plan`. The JVM credits
counters from projections
+/// only, so a counter anywhere else would leave its metric wrong.
+void checkIncrementMetricFunctionsInProjects(
+ const std::unordered_map<uint64_t, std::string>& functionMap,
+ const facebook::velox::core::PlanNode& plan);
+
+/// Sets the query configs a plan that calls a counter depends on, over any
user override:
+/// FilterProject operators export their expression statistics, and
expressions use the regular
+/// evaluation path, since simplified evaluation
(`expression.eval_simplified`) applies functions
+/// without counting their rows.
+void setIncrementMetricQueryConfigs(std::unordered_map<std::string,
std::string>& configs);
+
+} // namespace gluten
diff --git a/cpp/velox/operators/functions/RegistrationAllFunctions.cc
b/cpp/velox/operators/functions/RegistrationAllFunctions.cc
index 25ae12c19b..1d91e42263 100644
--- a/cpp/velox/operators/functions/RegistrationAllFunctions.cc
+++ b/cpp/velox/operators/functions/RegistrationAllFunctions.cc
@@ -16,6 +16,7 @@
*/
#include "operators/functions/RegistrationAllFunctions.h"
+#include "operators/functions/IncrementMetricFunction.h"
#include "operators/functions/RowConstructorWithNull.h"
#include "operators/functions/RowFunctionWithNull.h"
#include "operators/functions/delta/DeltaBitmapAggregator.h"
@@ -86,6 +87,8 @@ void registerAllFunctions() {
velox::functions::iceberg::registerFunctions();
registerDeltaBitmapAggregator();
+ // Fixed counter slots for Delta's IncrementMetric; see
IncrementMetricFunction.h.
+ registerIncrementMetricFunctions();
// Gluten-managed function implementations. Registered last so they take
// precedence over same-name, same-signature Velox functions.
diff --git a/cpp/velox/substrait/SubstraitToVeloxPlan.cc
b/cpp/velox/substrait/SubstraitToVeloxPlan.cc
index df076d0f6f..bc52611bb9 100644
--- a/cpp/velox/substrait/SubstraitToVeloxPlan.cc
+++ b/cpp/velox/substrait/SubstraitToVeloxPlan.cc
@@ -23,6 +23,7 @@
#include "compute/delta/DeltaSplitInfo.h"
#include "compute/iceberg/IcebergPlanConverter.h"
#include "jni/JniHashTable.h"
+#include "operators/functions/IncrementMetricFunction.h"
#include "operators/hashjoin/HashTableBuilder.h"
#include "operators/plannodes/RowVectorStream.h"
#ifdef ENABLE_KAFKA
@@ -1951,13 +1952,16 @@ core::PlanNodePtr
SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
// In fact, only one RelRoot or Rel is expected here.
VELOX_CHECK_EQ(substraitPlan.relations_size(), 1);
const auto& rel = substraitPlan.relations(0);
+ core::PlanNodePtr plan;
if (rel.has_root()) {
- return toVeloxPlan(rel.root());
+ plan = toVeloxPlan(rel.root());
} else if (rel.has_rel()) {
- return toVeloxPlan(rel.rel());
+ plan = toVeloxPlan(rel.rel());
} else {
VELOX_FAIL("RelRoot or Rel is expected in Plan.");
}
+ checkIncrementMetricFunctionsInProjects(functionMap_, *plan);
+ return plan;
}
std::string SubstraitToVeloxPlanConverter::nextPlanNodeId() {
diff --git a/cpp/velox/tests/CMakeLists.txt b/cpp/velox/tests/CMakeLists.txt
index 82af2bacbc..a45420d395 100644
--- a/cpp/velox/tests/CMakeLists.txt
+++ b/cpp/velox/tests/CMakeLists.txt
@@ -154,7 +154,7 @@ add_velox_test(
VeloxSubstraitSignatureTest.cc
VeloxToSubstraitTypeTest.cc)
add_velox_test(spark_functions_test SOURCES SparkFunctionTest.cc
- FunctionTest.cc)
+ FunctionTest.cc IncrementMetricFunctionTest.cc)
add_velox_test(runtime_test SOURCES RuntimeTest.cc)
add_velox_test(velox_backend_test SOURCES VeloxBackendTest.cc)
add_velox_test(velox_memory_test SOURCES MemoryManagerTest.cc)
diff --git a/cpp/velox/tests/IncrementMetricFunctionTest.cc
b/cpp/velox/tests/IncrementMetricFunctionTest.cc
new file mode 100644
index 0000000000..e68cdfa769
--- /dev/null
+++ b/cpp/velox/tests/IncrementMetricFunctionTest.cc
@@ -0,0 +1,170 @@
+/*
+ * 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.
+ */
+
+#include <optional>
+#include <unordered_map>
+
+#include "operators/functions/IncrementMetricFunction.h"
+#include "operators/functions/RegistrationAllFunctions.h"
+#include "velox/common/base/tests/GTestUtils.h"
+#include "velox/core/QueryConfig.h"
+#include "velox/exec/tests/utils/PlanBuilder.h"
+#include "velox/functions/sparksql/tests/SparkFunctionBaseTest.h"
+
+using namespace facebook::velox;
+using namespace facebook::velox::functions::sparksql::test;
+
+namespace gluten {
+
+class IncrementMetricFunctionTest : public SparkFunctionBaseTest {
+ public:
+ IncrementMetricFunctionTest() {
+ registerAllFunctions();
+ }
+};
+
+TEST_F(IncrementMetricFunctionTest, slotNames) {
+ ASSERT_EQ(incrementMetricFunctionName(0), "increment_metric_0");
+ ASSERT_EQ(incrementMetricFunctionName(31), "increment_metric_31");
+ ASSERT_TRUE(isIncrementMetricFunction("increment_metric_7"));
+ ASSERT_FALSE(isIncrementMetricFunction("plus"));
+ ASSERT_FALSE(isIncrementMetricFunction("my_increment_metric_7"));
+ ASSERT_TRUE(isIncrementMetricFunction("increment_metric_0"));
+
ASSERT_TRUE(isIncrementMetricFunction(incrementMetricFunctionName(kIncrementMetricFunctionSlots
- 1)));
+ // Only the registered slots count.
+ ASSERT_FALSE(isIncrementMetricFunction("increment_metric_"));
+
ASSERT_FALSE(isIncrementMetricFunction(incrementMetricFunctionName(kIncrementMetricFunctionSlots)));
+ ASSERT_FALSE(isIncrementMetricFunction("increment_metric_-1"));
+ ASSERT_FALSE(isIncrementMetricFunction("increment_metric_07"));
+ ASSERT_FALSE(isIncrementMetricFunction("increment_metric_7x"));
+ ASSERT_FALSE(isIncrementMetricFunction("increment_metric_foo"));
+}
+
+TEST_F(IncrementMetricFunctionTest, passesInputThroughIncludingNulls) {
+ auto input = makeNullableFlatVector<int64_t>({1, std::nullopt, 3});
+ auto result = evaluate("increment_metric_0(c0)", makeRowVector({input}));
+ facebook::velox::test::assertEqualVectors(input, result);
+
+ auto last = incrementMetricFunctionName(kIncrementMetricFunctionSlots - 1);
+ auto strings = makeFlatVector<StringView>({"a", "bb", "ccc"});
+ auto stringResult = evaluate(last + "(c0)", makeRowVector({strings}));
+ facebook::velox::test::assertEqualVectors(strings, stringResult);
+}
+
+TEST_F(IncrementMetricFunctionTest, countsOnlyRowsThatReachTheCall) {
+ // The shape of a MERGE clause: rows 2, 3 and 4 take the branch, row 1 does
not, so each counter
+ // sees the rows of its own branch, exactly as Spark evaluates
IncrementMetric. The literals are
+ // what Delta wraps; they must not be folded into constants, which would
skip the count.
+ // Spark function names: the parser's `>` maps to a Presto name that Gluten
does not register.
+ auto data = makeRowVector({makeFlatVector<int32_t>({1, 2, 3, 4})});
+ auto exprSet = compileExpression(
+ "if(greaterthan(c0, 1), increment_metric_1(true),
increment_metric_2(false))", asRowType(data->type()));
+ auto result = evaluate(*exprSet, data);
+ facebook::velox::test::assertEqualVectors(makeFlatVector<bool>({false, true,
true, true}), result);
+ auto stats = exprSet->stats();
+ ASSERT_EQ(stats.at("increment_metric_1").numProcessedRows, 3);
+ ASSERT_EQ(stats.at("increment_metric_2").numProcessedRows, 1);
+
+ // A counter over a literal at the root of a projection counts every row.
+ exprSet = compileExpression("increment_metric_3(true)",
asRowType(data->type()));
+ evaluate(*exprSet, data);
+ ASSERT_EQ(exprSet->stats().at("increment_metric_3").numProcessedRows, 4);
+}
+
+TEST_F(IncrementMetricFunctionTest, countsEveryRowOfAnEncodedInput) {
+ // Peeling the dictionary would evaluate the counter on its 2 distinct
values instead of 6 rows.
+ auto base = makeFlatVector<int32_t>({10, 20});
+ auto input = wrapInDictionary(makeIndices({0, 1, 0, 1, 0, 1}), 6, base);
+ auto data = makeRowVector({input});
+ auto exprSet = compileExpression("increment_metric_5(c0)",
asRowType(data->type()));
+ auto result = evaluate(*exprSet, data);
+ facebook::velox::test::assertEqualVectors(input, result);
+ ASSERT_EQ(exprSet->stats().at("increment_metric_5").numProcessedRows, 6);
+}
+
+TEST_F(IncrementMetricFunctionTest, nullInputStillCounts) {
+ auto data = makeRowVector({makeNullableFlatVector<int64_t>({1, std::nullopt,
std::nullopt})});
+ auto exprSet = compileExpression("increment_metric_4(c0)",
asRowType(data->type()));
+ evaluate(*exprSet, data);
+ ASSERT_EQ(exprSet->stats().at("increment_metric_4").numProcessedRows, 3);
+}
+
+TEST_F(IncrementMetricFunctionTest, detectsCountersInAPlan) {
+ auto data = makeRowVector({makeFlatVector<int32_t>({1, 2})});
+ auto withCounter =
exec::test::PlanBuilder().values({data}).project({"increment_metric_5(c0) AS
c0"}).planNode();
+ ASSERT_TRUE(usesIncrementMetricFunctions(*withCounter));
+
+ auto nestedCounter = exec::test::PlanBuilder()
+ .values({data})
+ .project({"if(greaterthan(c0, 1),
increment_metric_6(c0), c0) AS c0"})
+ .filter("greaterthan(c0, 0)")
+ .planNode();
+ ASSERT_TRUE(usesIncrementMetricFunctions(*nestedCounter));
+
+ auto plain =
+ exec::test::PlanBuilder().values({data}).project({"add(c0, 1) AS
c0"}).filter("greaterthan(c0, 0)").planNode();
+ ASSERT_FALSE(usesIncrementMetricFunctions(*plain));
+}
+
+TEST_F(IncrementMetricFunctionTest, rejectsCountersOutsideProjections) {
+ auto data = makeRowVector({makeFlatVector<int32_t>({1, 2})});
+ // Substrait declarations, `<name>:<argument types>`, as the plan converter
receives them.
+ const std::unordered_map<uint64_t, std::string> functionMap{
+ {0, "increment_metric_5:i32"}, {1, "greaterthan:i32_i32"}};
+
+ auto inProject = exec::test::PlanBuilder()
+ .values({data})
+ .project({"increment_metric_5(c0) AS c0"})
+ .filter("greaterthan(c0, 0)")
+ .planNode();
+ checkIncrementMetricFunctionsInProjects(functionMap, *inProject);
+
+ // The JVM credits counters from projections only, so a counter in a filter
is rejected.
+ auto inFilter =
exec::test::PlanBuilder().values({data}).filter("greaterthan(increment_metric_5(c0),
0)").planNode();
+ ASSERT_FALSE(usesIncrementMetricFunctions(*inFilter));
+ VELOX_ASSERT_THROW(
+ checkIncrementMetricFunctionsInProjects(functionMap, *inFilter),
+ "Delta counter increment_metric_5 is declared but no projection calls
it");
+
+ // Plans without counters are not inspected.
+ checkIncrementMetricFunctionsInProjects({{1, "greaterthan:i32_i32"}},
*inFilter);
+}
+
+TEST_F(IncrementMetricFunctionTest,
queryConfigsKeepCountingUnderUserOverrides) {
+ // A counter at the root of a projection, like the single clause of an
insert-only MERGE.
+ auto data = makeRowVector({makeFlatVector<int32_t>({1, 2, 3, 4})});
+ // Overrides a user can pass through
spark.gluten.sql.columnar.backend.velox.*: simplified
+ // evaluation applies the counter without counting the rows it processes,
and without the export
+ // FilterProject operators report no counts.
+ std::unordered_map<std::string, std::string> configs{
+ {core::QueryConfig::kExprEvalSimplified, "true"},
{core::QueryConfig::kOperatorTrackExpressionStats, "false"}};
+ setIncrementMetricQueryConfigs(configs);
+ queryCtx_->testingOverrideConfigUnsafe(std::move(configs));
+ ASSERT_FALSE(queryCtx_->queryConfig().exprEvalSimplified());
+ ASSERT_TRUE(queryCtx_->queryConfig().operatorTrackExpressionStats());
+
+ // FilterProject operators pick their evaluator from the query config the
same way.
+ auto exprSet =
+ exec::makeExprSetFromFlag({makeTypedExpr("increment_metric_0(c0)",
asRowType(data->type()))}, &execCtx_);
+ auto result = evaluate(*exprSet, data);
+ facebook::velox::test::assertEqualVectors(data->childAt(0), result);
+ const auto stats = exprSet->stats();
+ ASSERT_EQ(stats.count("increment_metric_0"), 1);
+ ASSERT_EQ(stats.at("increment_metric_0").numProcessedRows, 4);
+}
+
+} // namespace gluten
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
