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 d93ac7418f [GLUTEN-12597][CORE] Migrate TopNRel to Expression count +
FetchMode (Substrait 0.98) (#12728)
d93ac7418f is described below
commit d93ac7418feee88c82354e07bd0961de6b1445cb
Author: Niels Pardon <[email protected]>
AuthorDate: Wed Aug 19 15:27:54 2026 +0200
[GLUTEN-12597][CORE] Migrate TopNRel to Expression count + FetchMode
(Substrait 0.98) (#12728)
---
cpp/velox/substrait/SubstraitParser.cc | 12 ++++
cpp/velox/substrait/SubstraitParser.h | 8 +++
cpp/velox/substrait/SubstraitToVeloxPlan.cc | 13 +++-
.../substrait/SubstraitToVeloxPlanValidator.cc | 23 ++++++-
.../apache/gluten/substrait/rel/RelBuilder.java | 8 +--
.../org/apache/gluten/substrait/rel/TopNNode.java | 10 +++-
.../substrait/proto/substrait/algebra.proto | 32 ++++++++--
.../gluten/substrait/rel/TopNRelProtoSuite.scala | 70 ++++++++++++++++++++++
8 files changed, 164 insertions(+), 12 deletions(-)
diff --git a/cpp/velox/substrait/SubstraitParser.cc
b/cpp/velox/substrait/SubstraitParser.cc
index a57b3f69cc..e31ea8fa4e 100644
--- a/cpp/velox/substrait/SubstraitParser.cc
+++ b/cpp/velox/substrait/SubstraitParser.cc
@@ -16,6 +16,7 @@
*/
#include "SubstraitParser.h"
+#include <limits>
#include "TypeUtils.h"
#include "VeloxSubstraitSignature.h"
#include "velox/common/base/Exceptions.h"
@@ -316,6 +317,17 @@ std::vector<TypePtr> SubstraitParser::sigToTypes(const
std::string& signature) {
return types;
}
+std::optional<int32_t> SubstraitParser::getRowCount(const
::substrait::Expression& expression) {
+ if (!expression.has_literal() || !expression.literal().has_i64()) {
+ return std::nullopt;
+ }
+ const int64_t count = expression.literal().i64();
+ if (count <= 0 || count > std::numeric_limits<int32_t>::max()) {
+ return std::nullopt;
+ }
+ return static_cast<int32_t>(count);
+}
+
template <typename T>
T SubstraitParser::getLiteralValue(const ::substrait::Expression::Literal& /*
literal */) {
VELOX_NYI();
diff --git a/cpp/velox/substrait/SubstraitParser.h
b/cpp/velox/substrait/SubstraitParser.h
index 1122f3dc9b..5097783f9b 100644
--- a/cpp/velox/substrait/SubstraitParser.h
+++ b/cpp/velox/substrait/SubstraitParser.h
@@ -24,6 +24,8 @@
#include <google/protobuf/wrappers.pb.h>
+#include <optional>
+
#include "velox/connectors/hive/TableHandle.h"
#include "velox/type/Type.h"
@@ -103,6 +105,12 @@ class SubstraitParser {
template <typename T>
static T getLiteralValue(const ::substrait::Expression::Literal& /* literal
*/);
+ /// Extract a row count that Substrait models as an Expression, e.g.
TopNRel::count. Velox row
+ /// counts are positive int32 values, so std::nullopt is returned when the
expression is unset,
+ /// null, not an i64 literal, or out of the int32 range. Callers are
expected to reject the plan
+ /// in that case rather than silently substituting a count.
+ static std::optional<int32_t> getRowCount(const ::substrait::Expression&
expression);
+
private:
/// A map used for mapping Substrait function keywords into Velox functions'
/// keywords. Key: the Substrait function keyword, Value: the Velox function
diff --git a/cpp/velox/substrait/SubstraitToVeloxPlan.cc
b/cpp/velox/substrait/SubstraitToVeloxPlan.cc
index e648be8003..2cff687dd2 100644
--- a/cpp/velox/substrait/SubstraitToVeloxPlan.cc
+++ b/cpp/velox/substrait/SubstraitToVeloxPlan.cc
@@ -1401,10 +1401,21 @@ core::PlanNodePtr
SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
}
core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const
::substrait::TopNRel& topNRel) {
+ // validate(TopNRel) rejects everything checked below, but native validation
can be turned off
+ // (spark.gluten.sql.enable.native.validation), and this converter is also
reachable from the
+ // JSON-plan test and benchmark paths. Fail loudly rather than executing a
plan whose OFFSET or
+ // WITH TIES semantics Velox's TopN cannot express and would silently drop.
+ VELOX_USER_CHECK(topNRel.mode() == ::substrait::FETCH_MODE_ROWS_ONLY,
"TopNRel only supports FETCH_MODE_ROWS_ONLY.");
+ VELOX_USER_CHECK(!topNRel.has_offset(), "TopNRel does not support an
offset.");
+ // TopNRel models the row limit as an `Expression count`; Gluten's producer
always emits a
+ // positive i64 literal.
+ const auto count = SubstraitParser::getRowCount(topNRel.count());
+ VELOX_USER_CHECK(count.has_value(), "TopNRel count must be an i64 literal in
the range [1, INT32_MAX].");
+
auto childNode = convertSingleInput<::substrait::TopNRel>(topNRel);
auto [sortingKeys, sortingOrders] = processSortField(topNRel.sorts(),
childNode->outputType());
return std::make_shared<core::TopNNode>(
- nextPlanNodeId(), sortingKeys, sortingOrders,
static_cast<int32_t>(topNRel.n()), false /*isPartial*/, childNode);
+ nextPlanNodeId(), sortingKeys, sortingOrders, count.value(), false
/*isPartial*/, childNode);
}
core::PlanNodePtr SubstraitToVeloxPlanConverter::constructValueStreamNode(
diff --git a/cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc
b/cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc
index 638fcb4d8d..f1bd84b671 100644
--- a/cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc
+++ b/cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc
@@ -515,8 +515,27 @@ bool SubstraitToVeloxPlanValidator::validate(const
::substrait::TopNRel& topNRel
rowType = std::make_shared<RowType>(std::move(names), std::move(types));
}
- if (topNRel.n() < 0) {
- LOG_VALIDATION_MSG("N should be valid in TopNRel.");
+ // Velox's TopN supports neither an OFFSET nor WITH TIES, and its row count
is a positive int32.
+ // Reject anything else here so the query falls back instead of throwing
during plan conversion.
+ // The mode is checked against an allow list: FETCH_MODE_UNSPECIFIED is the
proto3 default and is
+ // not a valid producer choice, and a future mode must not be silently
treated as ROWS_ONLY.
+ if (topNRel.mode() != ::substrait::FETCH_MODE_ROWS_ONLY) {
+ LOG_VALIDATION_MSG("Only FETCH_MODE_ROWS_ONLY is supported in TopNRel.");
+ return false;
+ }
+ if (topNRel.has_offset()) {
+ LOG_VALIDATION_MSG("Offset is not supported in TopNRel.");
+ return false;
+ }
+ if (!SubstraitParser::getRowCount(topNRel.count()).has_value()) {
+ LOG_VALIDATION_MSG("Count should be an i64 literal in the range [1,
INT32_MAX] in TopNRel.");
+ return false;
+ }
+ // Substrait requires at least one sort field, and core::TopNNode asserts on
an empty key list.
+ // The duplicate-key loop below is a no-op for an empty list, so reject here
to fall back instead
+ // of throwing during plan conversion.
+ if (topNRel.sorts_size() == 0) {
+ LOG_VALIDATION_MSG("At least one sort field is required in TopNRel.");
return false;
}
diff --git
a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/RelBuilder.java
b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/RelBuilder.java
index e7d0640b7d..46dfbb0e28 100644
---
a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/RelBuilder.java
+++
b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/RelBuilder.java
@@ -285,20 +285,20 @@ public class RelBuilder {
}
public static RelNode makeTopNRel(
- RelNode input, Long n, List<SortField> sorts, SubstraitContext context,
Long operatorId) {
+ RelNode input, Long count, List<SortField> sorts, SubstraitContext
context, Long operatorId) {
context.registerRelToOperator(operatorId);
- return new TopNNode(input, n, sorts);
+ return new TopNNode(input, count, sorts);
}
public static RelNode makeTopNRel(
RelNode input,
- Long n,
+ Long count,
List<SortField> sorts,
AdvancedExtensionNode extensionNode,
SubstraitContext context,
Long operatorId) {
context.registerRelToOperator(operatorId);
- return new TopNNode(input, n, sorts, extensionNode);
+ return new TopNNode(input, count, sorts, extensionNode);
}
public static RelNode makeWindowRel(
diff --git
a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/TopNNode.java
b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/TopNNode.java
index dbc325674f..c23748b12a 100644
---
a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/TopNNode.java
+++
b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/TopNNode.java
@@ -16,8 +16,10 @@
*/
package org.apache.gluten.substrait.rel;
+import org.apache.gluten.substrait.expression.ExpressionBuilder;
import org.apache.gluten.substrait.extensions.AdvancedExtensionNode;
+import io.substrait.proto.FetchMode;
import io.substrait.proto.Rel;
import io.substrait.proto.RelCommon;
import io.substrait.proto.SortField;
@@ -55,12 +57,18 @@ public class TopNNode implements RelNode, Serializable {
relCommonBuilder.setDirect(RelCommon.Direct.newBuilder());
TopNRel.Builder topNBuilder = TopNRel.newBuilder();
+ topNBuilder.setCommon(relCommonBuilder.build());
if (input != null) {
topNBuilder.setInput(input.toProtobuf());
}
- topNBuilder.setN(count);
+ // Substrait's TopNRel expresses the row limit as an `Expression count`;
the only Gluten
+ // producer (TakeOrderedAndProject) supplies a literal Long, so wrap it as
an i64 literal.
+ // FetchMode has no meaningful proto3 default (FETCH_MODE_UNSPECIFIED is
not a valid producer
+ // choice), so set it explicitly; Gluten never emits WITH TIES.
+
topNBuilder.setCount(ExpressionBuilder.makeLongLiteral(count).toProtobuf());
+ topNBuilder.setMode(FetchMode.FETCH_MODE_ROWS_ONLY);
for (int i = 0; i < sorts.size(); i++) {
topNBuilder.addSorts(i, sorts.get(i));
diff --git
a/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto
b/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto
index 0015182f29..3d906a4e24 100644
---
a/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto
+++
b/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto
@@ -351,13 +351,37 @@ message FetchRel {
substrait.extensions.AdvancedExtension advanced_extension = 10;
}
-// The relational operator representing TOP N calculation
+// The top-N relational operator. A combination of a sort and fetch that only
+// maintains the number of records required to ensure a limited output.
message TopNRel {
RelCommon common = 1;
Rel input = 2;
- int64 n = 3;
- repeated SortField sorts = 4;
- substrait.extensions.AdvancedExtension advanced_extension = 10;
+
+ // List of one or more fields to sort by. At least one is required.
+ repeated SortField sorts = 3;
+
+ // Expression evaluating to a non-negative integer specifying the number of
+ // records to skip. Null is treated as 0. Recommended type is int64.
+ Expression offset = 4;
+
+ // Expression evaluating to a non-negative integer specifying the number of
+ // records to return. Null signals ALL. Recommended type is int64.
+ Expression count = 5;
+
+ // Determines how to handle rows tied with the last returned row.
+ FetchMode mode = 6;
+
+ substrait.extensions.AdvancedExtension advanced_extension = 7;
+}
+
+// Determines how a fetch operation handles rows tied with the last returned
row.
+enum FetchMode {
+ // Unspecified. Producers must set one of the defined modes.
+ FETCH_MODE_UNSPECIFIED = 0;
+ // Only return the requested number of rows.
+ FETCH_MODE_ROWS_ONLY = 1;
+ // Include additional rows tied with the last row (per the sort fields).
+ FETCH_MODE_WITH_TIES = 2;
}
// The relational operator representing a GROUP BY Aggregate
diff --git
a/gluten-substrait/src/test/scala/org/apache/gluten/substrait/rel/TopNRelProtoSuite.scala
b/gluten-substrait/src/test/scala/org/apache/gluten/substrait/rel/TopNRelProtoSuite.scala
new file mode 100644
index 0000000000..448364b85e
--- /dev/null
+++
b/gluten-substrait/src/test/scala/org/apache/gluten/substrait/rel/TopNRelProtoSuite.scala
@@ -0,0 +1,70 @@
+/*
+ * 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.substrait.rel
+
+import org.apache.gluten.substrait.SubstraitContext
+
+import io.substrait.proto.{FetchMode, SortField, TopNRel}
+import org.scalatest.funsuite.AnyFunSuite
+
+import java.util.Collections
+
+/**
+ * Locks the TopNRel producer contract after adopting upstream Substrait's
`TopNRel`. Gluten's
+ * vendored copy previously carried a local `int64 n = 3` field with `sorts =
4`. Upstream instead
+ * models the limit as `Expression count = 5`, alongside `Expression offset =
4` and
+ * `FetchMode mode = 6`, and puts `sorts` on tag 3. Gluten's only TopNRel
producer (Spark
+ * `TakeOrderedAndProject`) supplies a literal `Long` limit with no offset and
no ties, so the
+ * producer wraps the limit into an i64-literal `Expression`, sets
`FETCH_MODE_ROWS_ONLY` explicitly
+ * (`FETCH_MODE_UNSPECIFIED` is merely the proto3 default and not a valid
producer choice), and
+ * leaves `offset` unset. These assertions pin that contract, none of which a
plain compile catches.
+ */
+class TopNRelProtoSuite extends AnyFunSuite {
+
+ test("makeTopNRel emits count as an i64 literal expression with ROWS_ONLY
mode and no offset") {
+ val context = new SubstraitContext
+ val sorts = Collections.singletonList(SortField.getDefaultInstance)
+ val rel = RelBuilder.makeTopNRel(null, 10L, sorts, context, 0L)
+ val topNRel = rel.toProtobuf.getTopN
+
+ // The limit travels as an i64-literal `Expression`, not a scalar field.
+ assert(topNRel.hasCount)
+ assert(topNRel.getCount.getLiteral.getI64 === 10L)
+ // Fetch mode must be set explicitly rather than left at the proto3
default.
+ assert(topNRel.getMode === FetchMode.FETCH_MODE_ROWS_ONLY)
+ // Gluten never emits an offset (an unset offset is treated as 0).
+ assert(!topNRel.hasOffset)
+ assert(topNRel.getSortsCount === 1)
+ // RelCommon carries the direct output mapping, as it does for every other
rel node.
+ assert(topNRel.hasCommon)
+ assert(topNRel.getCommon.hasDirect)
+ }
+
+ test("TopNRel field numbers match upstream Substrait") {
+ // A round trip through the generated class cannot detect a renumber,
because producer and
+ // consumer share one schema. Assert on the descriptor so the wire tags
are actually pinned:
+ // the native backend and any other Substrait consumer decode by field
number, not by name.
+ val descriptor = TopNRel.getDescriptor
+ assert(descriptor.findFieldByName("sorts").getNumber === 3)
+ assert(descriptor.findFieldByName("offset").getNumber === 4)
+ assert(descriptor.findFieldByName("count").getNumber === 5)
+ assert(descriptor.findFieldByName("mode").getNumber === 6)
+ assert(descriptor.findFieldByName("advanced_extension").getNumber === 7)
+ // The Gluten-local `int64 n` field is gone; tag 3 now carries `sorts`.
+ assert(descriptor.findFieldByName("n") === null)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
