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 0e35f95dc8 [GLUTEN-12597][CORE] Migrate WindowRel to
ConsistentPartitionWindowRel + BoundsType (Substrait 0.98) (#12727)
0e35f95dc8 is described below
commit 0e35f95dc839aa2cbcd48788d830577bde8a8dbd
Author: Niels Pardon <[email protected]>
AuthorDate: Wed Aug 19 15:27:34 2026 +0200
[GLUTEN-12597][CORE] Migrate WindowRel to ConsistentPartitionWindowRel +
BoundsType (Substrait 0.98) (#12727)
---
.../substrait-plans/tpcds-q47-wholestage-9.json | 84 +++++++++----------
.../local-engine/Parser/AggregateFunctionParser.h | 13 +--
.../Parser/RelParsers/WindowRelParser.cpp | 54 ++++++------
.../Parser/RelParsers/WindowRelParser.h | 10 +--
cpp/velox/substrait/SubstraitToVeloxPlan.cc | 41 ++++-----
cpp/velox/substrait/SubstraitToVeloxPlan.h | 6 +-
.../substrait/SubstraitToVeloxPlanValidator.cc | 28 +++----
.../substrait/SubstraitToVeloxPlanValidator.h | 2 +-
.../substrait/expression/WindowFunctionNode.java | 30 ++++---
.../apache/gluten/substrait/rel/WindowRelNode.java | 8 +-
.../substrait/proto/substrait/algebra.proto | 69 ++++++++++++----
.../apache/gluten/utils/WindowRelProtoSuite.scala | 96 ++++++++++++++++++++++
12 files changed, 282 insertions(+), 159 deletions(-)
diff --git
a/backends-clickhouse/src/test/resources/substrait-plans/tpcds-q47-wholestage-9.json
b/backends-clickhouse/src/test/resources/substrait-plans/tpcds-q47-wholestage-9.json
index 946cb1e476..cc68e2e45b 100644
---
a/backends-clickhouse/src/test/resources/substrait-plans/tpcds-q47-wholestage-9.json
+++
b/backends-clickhouse/src/test/resources/substrait-plans/tpcds-q47-wholestage-9.json
@@ -207,23 +207,22 @@
}]
}
},
- "measures": [{
- "measure": {
- "upperBound": {
- "currentRow": {
- }
- },
- "lowerBound": {
- "unboundedPreceding": {
- }
- },
- "outputType": {
- "i32": {
- "nullability": "NULLABILITY_REQUIRED"
- }
- },
- "columnName": "rn_510"
- }
+ "windowFunctions": [{
+ "upperBound": {
+ "currentRow": {
+ }
+ },
+ "lowerBound": {
+ "unbounded": {
+ }
+ },
+ "outputType": {
+ "i32": {
+ "nullability": "NULLABILITY_REQUIRED"
+ }
+ },
+ "boundsType": "BOUNDS_TYPE_ROWS",
+ "columnName": "rn_510"
}],
"partitionExpressions": [{
"selection": {
@@ -345,35 +344,34 @@
}
}
},
- "measures": [{
- "measure": {
- "functionReference": 4,
- "upperBound": {
- "unboundedFollowing": {
- }
- },
- "lowerBound": {
- "unboundedPreceding": {
- }
- },
- "outputType": {
- "fp64": {
- "nullability": "NULLABILITY_NULLABLE"
- }
- },
- "arguments": [{
- "value": {
- "selection": {
- "directReference": {
- "structField": {
- "field": 7
- }
+ "windowFunctions": [{
+ "functionReference": 4,
+ "upperBound": {
+ "unbounded": {
+ }
+ },
+ "lowerBound": {
+ "unbounded": {
+ }
+ },
+ "outputType": {
+ "fp64": {
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ },
+ "arguments": [{
+ "value": {
+ "selection": {
+ "directReference": {
+ "structField": {
+ "field": 7
}
}
}
- }],
- "columnName": "avg_monthly_sales_509"
- }
+ }
+ }],
+ "boundsType": "BOUNDS_TYPE_ROWS",
+ "columnName": "avg_monthly_sales_509"
}],
"partitionExpressions": [{
"selection": {
diff --git a/cpp-ch/local-engine/Parser/AggregateFunctionParser.h
b/cpp-ch/local-engine/Parser/AggregateFunctionParser.h
index 3c2463bf92..5c8fe59dcf 100644
--- a/cpp-ch/local-engine/Parser/AggregateFunctionParser.h
+++ b/cpp-ch/local-engine/Parser/AggregateFunctionParser.h
@@ -53,13 +53,14 @@ public:
CommonFunctionInfo() { function_ref = -1; }
- CommonFunctionInfo(const substrait::WindowRel::Measure & win_measure)
- : function_ref(win_measure.measure().function_reference())
- , arguments(win_measure.measure().arguments())
- , output_type(win_measure.measure().output_type())
- , phase(win_measure.measure().phase())
- , sort_fields(win_measure.measure().sorts())
+ CommonFunctionInfo(const
substrait::ConsistentPartitionWindowRel::WindowRelFunction & win_function)
+ : function_ref(win_function.function_reference())
+ , arguments(win_function.arguments())
+ , output_type(win_function.output_type())
+ , phase(win_function.phase())
{
+ // Substrait 0.98's WindowRelFunction has no per-function sorts
(they are hoisted to the rel);
+ // Gluten always left them empty, so sort_fields stays
default-constructed.
is_in_window = true;
is_aggregate_function = true;
}
diff --git a/cpp-ch/local-engine/Parser/RelParsers/WindowRelParser.cpp
b/cpp-ch/local-engine/Parser/RelParsers/WindowRelParser.cpp
index 4fb9feafff..e8baaa7ea5 100644
--- a/cpp-ch/local-engine/Parser/RelParsers/WindowRelParser.cpp
+++ b/cpp-ch/local-engine/Parser/RelParsers/WindowRelParser.cpp
@@ -64,9 +64,8 @@ WindowRelParser::parse(DB::QueryPlanPtr current_plan_, const
substrait::Rel & re
// The output header is : original columns ++ window columns
output_header = *current_plan->getCurrentHeader();
- for (const auto & measure : win_rel_pb.measures())
+ for (const auto & win_function : win_rel_pb.window_functions())
{
- const auto & win_function = measure.measure();
ColumnWithTypeAndName named_col;
named_col.name = win_function.column_name();
named_col.type = TypeParser::parseType(win_function.output_type());
@@ -120,8 +119,7 @@ std::unordered_map<String, WindowDescription>
WindowRelParser::parseWindowDescri
for (size_t i = 0; i < win_infos.size(); ++i)
{
auto & win_info = win_infos[i];
- const auto & measure = *win_info.measure;
- const auto & win_function = measure.measure();
+ const auto & win_function = *win_info.window_function;
auto win_description = parseWindowDescription(win_info);
/// Check whether there is already a window description with the same
name
@@ -148,7 +146,7 @@ DB::WindowFrame WindowRelParser::parseWindowFrame(const
WindowInfo & win_info)
{
DB::WindowFrame win_frame;
const auto & signature_function_name = win_info.signature_function_name;
- const auto & window_function = win_info.measure->measure();
+ const auto & window_function = *win_info.window_function;
win_frame.type = parseWindowFrameType(signature_function_name,
window_function);
parseBoundType(window_function.lower_bound(), true, win_frame.begin_type,
win_frame.begin_offset, win_frame.begin_preceding);
parseBoundType(window_function.upper_bound(), false, win_frame.end_type,
win_frame.end_offset, win_frame.end_preceding);
@@ -162,25 +160,27 @@ DB::WindowFrame WindowRelParser::parseWindowFrame(const
WindowInfo & win_info)
return win_frame;
}
-DB::WindowFrame::FrameType
-WindowRelParser::parseWindowFrameType(const std::string & function_name, const
substrait::Expression::WindowFunction & window_function)
+DB::WindowFrame::FrameType WindowRelParser::parseWindowFrameType(
+ const std::string & function_name, const
substrait::ConsistentPartitionWindowRel::WindowRelFunction & window_function)
{
// It's weird! The frame type only could be rows in spark for rank(). But
in clickhouse
// it's should be range. If run rank() over rows frame, the result is
different. The rank number
// is different for the same values.
- static const std::unordered_map<std::string, substrait::WindowType>
special_function_frame_type
- = {{"rank", substrait::RANGE}, {"dense_rank", substrait::RANGE},
{"percent_rank", substrait::RANGE}};
+ static const std::unordered_map<std::string,
substrait::Expression_WindowFunction_BoundsType> special_function_frame_type
+ = {{"rank",
substrait::Expression_WindowFunction_BoundsType_BOUNDS_TYPE_RANGE},
+ {"dense_rank",
substrait::Expression_WindowFunction_BoundsType_BOUNDS_TYPE_RANGE},
+ {"percent_rank",
substrait::Expression_WindowFunction_BoundsType_BOUNDS_TYPE_RANGE}};
- substrait::WindowType frame_type;
+ substrait::Expression_WindowFunction_BoundsType frame_type;
auto iter = special_function_frame_type.find(function_name);
if (iter != special_function_frame_type.end())
frame_type = iter->second;
else
- frame_type = window_function.window_type();
+ frame_type = window_function.bounds_type();
- if (frame_type == substrait::ROWS)
+ if (frame_type ==
substrait::Expression_WindowFunction_BoundsType_BOUNDS_TYPE_ROWS)
return DB::WindowFrame::FrameType::ROWS;
- else if (frame_type == substrait::RANGE)
+ else if (frame_type ==
substrait::Expression_WindowFunction_BoundsType_BOUNDS_TYPE_RANGE)
return DB::WindowFrame::FrameType::RANGE;
else
throw DB::Exception(DB::ErrorCodes::UNKNOWN_TYPE, "Unknow window frame
type:{}", frame_type);
@@ -222,17 +222,13 @@ void WindowRelParser::parseBoundType(
offset = 0;
preceding_direction = is_begin_or_end;
}
- else if (bound.has_unbounded_preceding())
+ else if (bound.has_unbounded())
{
+ // Substrait 0.98 collapses unbounded_preceding/unbounded_following
into a single `unbounded`;
+ // the direction is inferred from position (begin bound = preceding,
end bound = following).
bound_type = DB::WindowFrame::BoundaryType::Unbounded;
offset = 0;
- preceding_direction = true;
- }
- else if (bound.has_unbounded_following())
- {
- bound_type = DB::WindowFrame::BoundaryType::Unbounded;
- offset = 0;
- preceding_direction = false;
+ preceding_direction = is_begin_or_end;
}
else
{
@@ -242,7 +238,7 @@ void WindowRelParser::parseBoundType(
WindowFunctionDescription WindowRelParser::parseWindowFunctionDescription(
const String & ch_function_name,
- const substrait::Expression::WindowFunction & window_function,
+ const substrait::ConsistentPartitionWindowRel::WindowRelFunction &
window_function,
const DB::Names & arg_names,
const DB::DataTypes & arg_types,
const DB::Array & params)
@@ -259,16 +255,16 @@ WindowFunctionDescription
WindowRelParser::parseWindowFunctionDescription(
return description;
}
-void WindowRelParser::initWindowsInfos(const substrait::WindowRel & win_rel)
+void WindowRelParser::initWindowsInfos(const
substrait::ConsistentPartitionWindowRel & win_rel)
{
- win_infos.reserve(win_rel.measures_size());
- for (const auto & measure : win_rel.measures())
+ win_infos.reserve(win_rel.window_functions_size());
+ for (const auto & win_function : win_rel.window_functions())
{
WindowInfo win_info;
- win_info.result_column_name = measure.measure().column_name();
- win_info.measure = &measure;
- win_info.signature_function_name =
*parseSignatureFunctionName(measure.measure().function_reference());
- win_info.parser_func_info =
AggregateFunctionParser::CommonFunctionInfo(measure);
+ win_info.result_column_name = win_function.column_name();
+ win_info.window_function = &win_function;
+ win_info.signature_function_name =
*parseSignatureFunctionName(win_function.function_reference());
+ win_info.parser_func_info =
AggregateFunctionParser::CommonFunctionInfo(win_function);
win_info.function_parser =
AggregateFunctionParserFactory::instance().get(win_info.signature_function_name,
parser_context);
win_info.function_name =
win_info.function_parser->getCHFunctionName(win_info.parser_func_info);
win_info.partition_exprs = win_rel.partition_expressions();
diff --git a/cpp-ch/local-engine/Parser/RelParsers/WindowRelParser.h
b/cpp-ch/local-engine/Parser/RelParsers/WindowRelParser.h
index 0c3b76947e..31eee0fc36 100644
--- a/cpp-ch/local-engine/Parser/RelParsers/WindowRelParser.h
+++ b/cpp-ch/local-engine/Parser/RelParsers/WindowRelParser.h
@@ -40,7 +40,7 @@ public:
private:
struct WindowInfo
{
- const substrait::WindowRel::Measure * measure = nullptr;
+ const substrait::ConsistentPartitionWindowRel::WindowRelFunction *
window_function = nullptr;
String result_column_name;
DB::Strings arg_column_names;
DB::DataTypes arg_column_types;
@@ -69,8 +69,8 @@ private:
// function may have different window frame in CH and spark.
DB::WindowDescription parseWindowDescription(const WindowInfo & win_info);
DB::WindowFrame parseWindowFrame(const WindowInfo & win_info);
- DB::WindowFrame::FrameType
- parseWindowFrameType(const std::string & function_name, const
substrait::Expression::WindowFunction & window_function);
+ DB::WindowFrame::FrameType parseWindowFrameType(
+ const std::string & function_name, const
substrait::ConsistentPartitionWindowRel::WindowRelFunction & window_function);
static void parseBoundType(
const substrait::Expression::WindowFunction::Bound & bound,
bool is_begin_or_end,
@@ -79,12 +79,12 @@ private:
bool & preceding);
DB::WindowFunctionDescription parseWindowFunctionDescription(
const String & ch_function_name,
- const substrait::Expression::WindowFunction & window_function,
+ const substrait::ConsistentPartitionWindowRel::WindowRelFunction &
window_function,
const DB::Names & arg_names,
const DB::DataTypes & arg_types,
const DB::Array & params);
- void initWindowsInfos(const substrait::WindowRel & win_rel);
+ void initWindowsInfos(const substrait::ConsistentPartitionWindowRel &
win_rel);
void tryAddProjectionBeforeWindow();
void tryAddProjectionAfterWindow();
};
diff --git a/cpp/velox/substrait/SubstraitToVeloxPlan.cc
b/cpp/velox/substrait/SubstraitToVeloxPlan.cc
index 888ab8d6ce..e648be8003 100644
--- a/cpp/velox/substrait/SubstraitToVeloxPlan.cc
+++ b/cpp/velox/substrait/SubstraitToVeloxPlan.cc
@@ -1097,14 +1097,14 @@ core::PlanNodePtr
SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
const core::WindowNode::Frame SubstraitToVeloxPlanConverter::createWindowFrame(
const ::substrait::Expression_WindowFunction_Bound& lower_bound,
const ::substrait::Expression_WindowFunction_Bound& upper_bound,
- const ::substrait::WindowType& type,
+ const ::substrait::Expression_WindowFunction_BoundsType& type,
const RowTypePtr& inputType) {
core::WindowNode::Frame frame;
switch (type) {
- case ::substrait::WindowType::ROWS:
+ case ::substrait::Expression_WindowFunction_BoundsType_BOUNDS_TYPE_ROWS:
frame.type = core::WindowNode::WindowType::kRows;
break;
- case ::substrait::WindowType::RANGE:
+ case ::substrait::Expression_WindowFunction_BoundsType_BOUNDS_TYPE_RANGE:
frame.type = core::WindowNode::WindowType::kRange;
break;
default:
@@ -1125,14 +1125,17 @@ const core::WindowNode::Frame
SubstraitToVeloxPlanConverter::createWindowFrame(
}
};
- auto boundTypeConversion = [&](::substrait::Expression_WindowFunction_Bound
boundType)
- -> std::tuple<core::WindowNode::BoundType, core::TypedExprPtr> {
+ auto boundTypeConversion = [&](::substrait::Expression_WindowFunction_Bound
boundType,
+ bool isLowerBound) ->
std::tuple<core::WindowNode::BoundType, core::TypedExprPtr> {
if (boundType.has_current_row()) {
return std::make_tuple(core::WindowNode::BoundType::kCurrentRow,
nullptr);
- } else if (boundType.has_unbounded_following()) {
- return std::make_tuple(core::WindowNode::BoundType::kUnboundedFollowing,
nullptr);
- } else if (boundType.has_unbounded_preceding()) {
- return std::make_tuple(core::WindowNode::BoundType::kUnboundedPreceding,
nullptr);
+ } else if (boundType.has_unbounded()) {
+ // Substrait 0.98 uses a single `unbounded` bound; the direction is
inferred from position:
+ // an unbounded lower bound is the start of the partition, an unbounded
upper bound is the end.
+ return std::make_tuple(
+ isLowerBound ? core::WindowNode::BoundType::kUnboundedPreceding
+ : core::WindowNode::BoundType::kUnboundedFollowing,
+ nullptr);
} else if (boundType.has_following()) {
auto following = boundType.following();
return std::make_tuple(
@@ -1147,29 +1150,29 @@ const core::WindowNode::Frame
SubstraitToVeloxPlanConverter::createWindowFrame(
VELOX_FAIL("The BoundType is not supported.");
}
};
- std::tie(frame.startType, frame.startValue) =
boundTypeConversion(lower_bound);
- std::tie(frame.endType, frame.endValue) = boundTypeConversion(upper_bound);
+ std::tie(frame.startType, frame.startValue) =
boundTypeConversion(lower_bound, /*isLowerBound=*/true);
+ std::tie(frame.endType, frame.endValue) = boundTypeConversion(upper_bound,
/*isLowerBound=*/false);
return frame;
}
-core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const
::substrait::WindowRel& windowRel) {
+core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(
+ const ::substrait::ConsistentPartitionWindowRel& windowRel) {
core::PlanNodePtr childNode;
if (windowRel.has_input()) {
childNode = toVeloxPlan(windowRel.input());
} else {
- VELOX_FAIL("Child Rel is expected in WindowRel.");
+ VELOX_FAIL("Child Rel is expected in ConsistentPartitionWindowRel.");
}
const auto& inputType = childNode->outputType();
- // Parse measures and get the window expressions.
- // Each measure represents one window expression.
+ // Parse window functions and get the window expressions.
+ // Each window function represents one window expression.
std::vector<core::WindowNode::Function> windowNodeFunctions;
std::vector<std::string> windowColumnNames;
- windowNodeFunctions.reserve(windowRel.measures().size());
- for (const auto& smea : windowRel.measures()) {
- const auto& windowFunction = smea.measure();
+ windowNodeFunctions.reserve(windowRel.window_functions().size());
+ for (const auto& windowFunction : windowRel.window_functions()) {
std::string funcName = SubstraitParser::findVeloxFunction(functionMap_,
windowFunction.function_reference());
std::vector<core::TypedExprPtr> windowParams;
auto& argumentList = windowFunction.arguments();
@@ -1189,7 +1192,7 @@ core::PlanNodePtr
SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
windowVeloxType, std::move(windowParams),
exec::sanitizeName(funcName));
auto upperBound = windowFunction.upper_bound();
auto lowerBound = windowFunction.lower_bound();
- auto type = windowFunction.window_type();
+ auto type = windowFunction.bounds_type();
windowColumnNames.push_back(windowFunction.column_name());
diff --git a/cpp/velox/substrait/SubstraitToVeloxPlan.h
b/cpp/velox/substrait/SubstraitToVeloxPlan.h
index 5ced0b01d2..b0cf76fff3 100644
--- a/cpp/velox/substrait/SubstraitToVeloxPlan.h
+++ b/cpp/velox/substrait/SubstraitToVeloxPlan.h
@@ -104,8 +104,8 @@ class SubstraitToVeloxPlanConverter {
/// Used to convert Substrait GenerateRel into Velox PlanNode.
core::PlanNodePtr toVeloxPlan(const ::substrait::GenerateRel& generateRel);
- /// Used to convert Substrait WindowRel into Velox PlanNode.
- core::PlanNodePtr toVeloxPlan(const ::substrait::WindowRel& windowRel);
+ /// Used to convert Substrait ConsistentPartitionWindowRel into Velox
PlanNode.
+ core::PlanNodePtr toVeloxPlan(const
::substrait::ConsistentPartitionWindowRel& windowRel);
/// Used to convert Substrait WindowGroupLimitRel into Velox PlanNode.
core::PlanNodePtr toVeloxPlan(const ::substrait::WindowGroupLimitRel&
windowGroupLimitRel);
@@ -277,7 +277,7 @@ class SubstraitToVeloxPlanConverter {
const core::WindowNode::Frame createWindowFrame(
const ::substrait::Expression_WindowFunction_Bound& lower_bound,
const ::substrait::Expression_WindowFunction_Bound& upper_bound,
- const ::substrait::WindowType& type,
+ const ::substrait::Expression_WindowFunction_BoundsType& type,
const RowTypePtr& inputType);
/// The unique identification for each PlanNode.
diff --git a/cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc
b/cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc
index c041e580db..638fcb4d8d 100644
--- a/cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc
+++ b/cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc
@@ -637,8 +637,7 @@ bool SubstraitToVeloxPlanValidator::validate(const
::substrait::ExpandRel& expan
bool validateBoundType(::substrait::Expression_WindowFunction_Bound boundType)
{
switch (boundType.kind_case()) {
- case ::substrait::Expression_WindowFunction_Bound::kUnboundedFollowing:
- case ::substrait::Expression_WindowFunction_Bound::kUnboundedPreceding:
+ case ::substrait::Expression_WindowFunction_Bound::kUnbounded:
case ::substrait::Expression_WindowFunction_Bound::kCurrentRow:
case ::substrait::Expression_WindowFunction_Bound::kFollowing:
case ::substrait::Expression_WindowFunction_Bound::kPreceding:
@@ -649,28 +648,28 @@ bool
validateBoundType(::substrait::Expression_WindowFunction_Bound boundType) {
return true;
}
-bool SubstraitToVeloxPlanValidator::validate(const ::substrait::WindowRel&
windowRel) {
+bool SubstraitToVeloxPlanValidator::validate(const
::substrait::ConsistentPartitionWindowRel& windowRel) {
if (windowRel.has_input() && !validate(windowRel.input())) {
- LOG_VALIDATION_MSG("WindowRel input fails to validate.");
+ LOG_VALIDATION_MSG("ConsistentPartitionWindowRel input fails to
validate.");
return false;
}
// Get and validate the input types from extension.
if (!windowRel.has_advanced_extension()) {
- LOG_VALIDATION_MSG("Input types are expected in WindowRel.");
+ LOG_VALIDATION_MSG("Input types are expected in
ConsistentPartitionWindowRel.");
return false;
}
const auto& extension = windowRel.advanced_extension();
TypePtr inputRowType;
std::vector<TypePtr> types;
if (!parseVeloxType(extension, inputRowType) ||
!flattenSingleLevel(inputRowType, types)) {
- LOG_VALIDATION_MSG("Validation failed for input types in WindowRel.");
+ LOG_VALIDATION_MSG("Validation failed for input types in
ConsistentPartitionWindowRel.");
return false;
}
if (types.empty()) {
// See: https://github.com/apache/gluten/issues/7600.
- LOG_VALIDATION_MSG("Validation failed for empty input schema in
WindowRel.");
+ LOG_VALIDATION_MSG("Validation failed for empty input schema in
ConsistentPartitionWindowRel.");
return false;
}
@@ -684,9 +683,8 @@ bool SubstraitToVeloxPlanValidator::validate(const
::substrait::WindowRel& windo
// Validate WindowFunction
std::vector<std::string> funcSpecs;
- funcSpecs.reserve(windowRel.measures().size());
- for (const auto& smea : windowRel.measures()) {
- const auto& windowFunction = smea.measure();
+ funcSpecs.reserve(windowRel.window_functions().size());
+ for (const auto& windowFunction : windowRel.window_functions()) {
funcSpecs.emplace_back(planConverter_->findFuncSpec(windowFunction.function_reference()));
SubstraitParser::parseType(windowFunction.output_type());
for (const auto& arg : windowFunction.arguments()) {
@@ -701,14 +699,14 @@ bool SubstraitToVeloxPlanValidator::validate(const
::substrait::WindowRel& windo
}
}
// Validate BoundType and Frame Type
- switch (windowFunction.window_type()) {
- case ::substrait::WindowType::ROWS:
- case ::substrait::WindowType::RANGE:
+ switch (windowFunction.bounds_type()) {
+ case ::substrait::Expression_WindowFunction_BoundsType_BOUNDS_TYPE_ROWS:
+ case ::substrait::Expression_WindowFunction_BoundsType_BOUNDS_TYPE_RANGE:
break;
default:
LOG_VALIDATION_MSG(
- "the window type only support ROWS and RANGE, and the input type
is " +
- std::to_string(windowFunction.window_type()));
+ "the bounds type only support ROWS and RANGE, and the input type
is " +
+ std::to_string(windowFunction.bounds_type()));
return false;
}
diff --git a/cpp/velox/substrait/SubstraitToVeloxPlanValidator.h
b/cpp/velox/substrait/SubstraitToVeloxPlanValidator.h
index 51c58b6afb..3f83c4c32c 100644
--- a/cpp/velox/substrait/SubstraitToVeloxPlanValidator.h
+++ b/cpp/velox/substrait/SubstraitToVeloxPlanValidator.h
@@ -87,7 +87,7 @@ class SubstraitToVeloxPlanValidator {
bool validate(const ::substrait::SortRel& sortRel);
/// Used to validate whether the computing of this Window is supported.
- bool validate(const ::substrait::WindowRel& windowRel);
+ bool validate(const ::substrait::ConsistentPartitionWindowRel& windowRel);
/// Used to validate whether the computing of this WindowGroupLimit is
supported.
bool validate(const ::substrait::WindowGroupLimitRel& windowGroupLimitRel);
diff --git
a/gluten-substrait/src/main/java/org/apache/gluten/substrait/expression/WindowFunctionNode.java
b/gluten-substrait/src/main/java/org/apache/gluten/substrait/expression/WindowFunctionNode.java
index a114c6050a..d4be6c6ba2 100644
---
a/gluten-substrait/src/main/java/org/apache/gluten/substrait/expression/WindowFunctionNode.java
+++
b/gluten-substrait/src/main/java/org/apache/gluten/substrait/expression/WindowFunctionNode.java
@@ -21,10 +21,10 @@ import org.apache.gluten.expression.ExpressionConverter;
import org.apache.gluten.substrait.SubstraitContext;
import org.apache.gluten.substrait.type.TypeNode;
+import io.substrait.proto.ConsistentPartitionWindowRel;
import io.substrait.proto.Expression;
import io.substrait.proto.FunctionArgument;
import io.substrait.proto.FunctionOption;
-import io.substrait.proto.WindowType;
import org.apache.spark.sql.catalyst.expressions.Attribute;
import org.apache.spark.sql.catalyst.expressions.PreComputeRangeFrameBound;
@@ -82,14 +82,11 @@ public class WindowFunctionNode implements Serializable {
builder.setCurrentRow(currentRowBuilder.build());
break;
case ("UNBOUNDED PRECEDING"):
- Expression.WindowFunction.Bound.Unbounded_Preceding.Builder
precedingBuilder =
- Expression.WindowFunction.Bound.Unbounded_Preceding.newBuilder();
- builder.setUnboundedPreceding(precedingBuilder.build());
- break;
case ("UNBOUNDED FOLLOWING"):
- Expression.WindowFunction.Bound.Unbounded_Following.Builder
followingBuilder =
- Expression.WindowFunction.Bound.Unbounded_Following.newBuilder();
- builder.setUnboundedFollowing(followingBuilder.build());
+ // Substrait 0.98 collapses the two unbounded bounds into a single
`unbounded`; the
+ // direction is inferred from position (lower bound = start of the
partition, upper
+ // bound = end of the partition).
+
builder.setUnbounded(Expression.WindowFunction.Bound.Unbounded.newBuilder().build());
break;
default:
if (boundType instanceof PreComputeRangeFrameBound) {
@@ -141,23 +138,24 @@ public class WindowFunctionNode implements Serializable {
return builder;
}
- private WindowType getWindowType(String type) {
- WindowType windowType;
+ private Expression.WindowFunction.BoundsType getBoundsType(String type) {
+ Expression.WindowFunction.BoundsType boundsType;
switch (type) {
case ("ROWS"):
- windowType = WindowType.forNumber(0);
+ boundsType = Expression.WindowFunction.BoundsType.BOUNDS_TYPE_ROWS;
break;
case ("RANGE"):
- windowType = WindowType.forNumber(1);
+ boundsType = Expression.WindowFunction.BoundsType.BOUNDS_TYPE_RANGE;
break;
default:
throw new UnsupportedOperationException("Only support ROWS and RANGE
Frame type.");
}
- return windowType;
+ return boundsType;
}
- public Expression.WindowFunction toProtobuf() {
- Expression.WindowFunction.Builder windowBuilder =
Expression.WindowFunction.newBuilder();
+ public ConsistentPartitionWindowRel.WindowRelFunction toProtobuf() {
+ ConsistentPartitionWindowRel.WindowRelFunction.Builder windowBuilder =
+ ConsistentPartitionWindowRel.WindowRelFunction.newBuilder();
windowBuilder.setFunctionReference(functionId);
if (ignoreNulls) {
FunctionOption option =
FunctionOption.newBuilder().setName("ignoreNulls").build();
@@ -179,7 +177,7 @@ public class WindowFunctionNode implements Serializable {
Expression.WindowFunction.Bound.newBuilder();
windowBuilder.setLowerBound(setBound(lowerBoundBuilder,
lowerBound).build());
windowBuilder.setUpperBound(setBound(upperBoundBuilder,
upperBound).build());
- windowBuilder.setWindowType(getWindowType(frameType));
+ windowBuilder.setBoundsType(getBoundsType(frameType));
return windowBuilder.build();
}
}
diff --git
a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/WindowRelNode.java
b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/WindowRelNode.java
index 5e619cc647..3a47da8b16 100644
---
a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/WindowRelNode.java
+++
b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/WindowRelNode.java
@@ -20,10 +20,10 @@ import
org.apache.gluten.substrait.expression.ExpressionNode;
import org.apache.gluten.substrait.expression.WindowFunctionNode;
import org.apache.gluten.substrait.extensions.AdvancedExtensionNode;
+import io.substrait.proto.ConsistentPartitionWindowRel;
import io.substrait.proto.Rel;
import io.substrait.proto.RelCommon;
import io.substrait.proto.SortField;
-import io.substrait.proto.WindowRel;
import java.io.Serializable;
import java.util.ArrayList;
@@ -67,16 +67,14 @@ public class WindowRelNode implements RelNode, Serializable
{
RelCommon.Builder relCommonBuilder = RelCommon.newBuilder();
relCommonBuilder.setDirect(RelCommon.Direct.newBuilder());
- WindowRel.Builder windowBuilder = WindowRel.newBuilder();
+ ConsistentPartitionWindowRel.Builder windowBuilder =
ConsistentPartitionWindowRel.newBuilder();
windowBuilder.setCommon(relCommonBuilder.build());
if (input != null) {
windowBuilder.setInput(input.toProtobuf());
}
for (WindowFunctionNode windowFunctionNode : windowFunctionNodes) {
- WindowRel.Measure.Builder measureBuilder =
WindowRel.Measure.newBuilder();
- measureBuilder.setMeasure(windowFunctionNode.toProtobuf());
- windowBuilder.addMeasures(measureBuilder.build());
+ windowBuilder.addWindowFunctions(windowFunctionNode.toProtobuf());
}
for (int i = 0; i < partitionExpressions.size(); 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 6647fe38e8..0015182f29 100644
---
a/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto
+++
b/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto
@@ -411,16 +411,41 @@ message SortRel {
substrait.extensions.AdvancedExtension advanced_extension = 10;
}
-message WindowRel {
+message ConsistentPartitionWindowRel {
RelCommon common = 1;
Rel input = 2;
- repeated Measure measures = 3;
+ repeated WindowRelFunction window_functions = 3;
repeated Expression partition_expressions = 4;
repeated SortField sorts = 5;
+
substrait.extensions.AdvancedExtension advanced_extension = 10;
- message Measure {
- Expression.WindowFunction measure = 1;
+ // This message mirrors the `WindowFunction` message but removes the fields
defining the partition,
+ // sorts, and bounds, since those must be consistent across the various
functions in this rel. Refer
+ // to the `WindowFunction` message for a description of these fields.
+ message WindowRelFunction {
+ // References a function_anchor defined in this plan.
+ uint32 function_reference = 1;
+
+ repeated FunctionArgument arguments = 9;
+
+ repeated FunctionOption options = 11;
+
+ Type output_type = 7;
+
+ AggregationPhase phase = 6;
+
+ AggregateFunction.AggregationInvocation invocation = 10;
+
+ Expression.WindowFunction.Bound lower_bound = 5;
+
+ Expression.WindowFunction.Bound upper_bound = 4;
+
+ Expression.WindowFunction.BoundsType bounds_type = 12;
+
+ // Gluten fork: output column name for this window function, consumed by
the
+ // native backends to name the resulting window output column.
+ string column_name = 13;
}
}
@@ -596,7 +621,7 @@ message Rel {
HashJoinRel hash_join = 13;
MergeJoinRel merge_join = 14;
ExpandRel expand = 15;
- WindowRel window = 16;
+ ConsistentPartitionWindowRel window = 16;
GenerateRel generate = 17;
WriteRel write = 19;
TopNRel top_n = 23;
@@ -1068,8 +1093,8 @@ message Expression {
// Optional; defaults to the start of the partition.
Bound lower_bound = 5;
- string column_name = 12;
- WindowType window_type = 13;
+ // Defines the bounds type: ROWS, RANGE
+ BoundsType bounds_type = 12;
// Defines the record relative to the current record up to which the window
// extends. The bound is inclusive. If the upper bound indexes a record
@@ -1081,6 +1106,18 @@ message Expression {
// Deprecated; use arguments instead.
repeated Expression args = 8 [deprecated = true];
+ enum BoundsType {
+ BOUNDS_TYPE_UNSPECIFIED = 0;
+ // The lower and upper bound specify how many rows before and after the
current row
+ // the window should extend.
+ BOUNDS_TYPE_ROWS = 1;
+ // The lower and upper bound describe a range of values. The window
should include all rows
+ // where the value of the ordering column is greater than or equal to
(current_value - lower bound)
+ // and less than or equal to (current_value + upper bound). This bounds
type is only valid if there
+ // is a single ordering column.
+ BOUNDS_TYPE_RANGE = 2;
+ }
+
// Defines one of the two boundaries for the window of a window function.
message Bound {
// Defines that the bound extends this far back from the current record.
@@ -1112,9 +1149,10 @@ message Expression {
// Defines that the bound extends to or from the current record.
message CurrentRow {}
- message Unbounded_Preceding {}
-
- message Unbounded_Following {}
+ // Defines an "unbounded bound": for lower bounds this means the start
+ // of the partition, and for upper bounds this means the end of the
+ // partition.
+ message Unbounded {}
oneof kind {
// The bound extends some number of records behind the current record.
@@ -1127,8 +1165,10 @@ message Expression {
// The bound extends to the current record.
CurrentRow current_row = 3;
- Unbounded_Preceding unbounded_preceding = 4;
- Unbounded_Following unbounded_following = 5;
+ // The bound extends to the start of the partition or the end of the
+ // partition, depending on whether this represents the upper or lower
+ // bound.
+ Unbounded unbounded = 4;
}
}
}
@@ -1477,11 +1517,6 @@ enum AggregationPhase {
AGGREGATION_PHASE_INTERMEDIATE_TO_RESULT = 4;
}
-enum WindowType {
- ROWS = 0;
- RANGE = 1;
-}
-
// An aggregate function.
message AggregateFunction {
// Points to a function_anchor defined in this plan, which must refer
diff --git
a/gluten-substrait/src/test/scala/org/apache/gluten/utils/WindowRelProtoSuite.scala
b/gluten-substrait/src/test/scala/org/apache/gluten/utils/WindowRelProtoSuite.scala
new file mode 100644
index 0000000000..33e94382d4
--- /dev/null
+++
b/gluten-substrait/src/test/scala/org/apache/gluten/utils/WindowRelProtoSuite.scala
@@ -0,0 +1,96 @@
+/*
+ * 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.utils
+
+import org.apache.gluten.substrait.`type`.TypeBuilder
+import org.apache.gluten.substrait.SubstraitContext
+import org.apache.gluten.substrait.expression.{ExpressionBuilder,
ExpressionNode}
+import org.apache.gluten.substrait.rel.RelBuilder
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, CurrentRow,
UnboundedFollowing, UnboundedPreceding}
+
+import io.substrait.proto.{Expression, SortField}
+import org.scalatest.funsuite.AnyFunSuite
+
+import java.util.{Arrays, Collections}
+
+/**
+ * Round-trip coverage for the Substrait 0.98 windowing migration (`WindowRel`
->
+ * `ConsistentPartitionWindowRel`, `WindowType` -> `BoundsType`, and the
+ * `unbounded_preceding`/`unbounded_following` -> single `unbounded`
collapse). These assertions pin
+ * the silent risks of the migration: the frame-type enum value remap and the
unbounded-bound
+ * collapse, neither of which a plain compile would catch.
+ */
+class WindowRelProtoSuite extends AnyFunSuite {
+
+ private def windowFunctionProto(
+ upperBound: org.apache.spark.sql.catalyst.expressions.Expression,
+ lowerBound: org.apache.spark.sql.catalyst.expressions.Expression,
+ frameType: String,
+ columnName: String)
+ : io.substrait.proto.ConsistentPartitionWindowRel.WindowRelFunction = {
+ val windowFunction = ExpressionBuilder.makeWindowFunction(
+ Integer.valueOf(0),
+ Collections.emptyList[ExpressionNode](),
+ columnName,
+ TypeBuilder.makeI32(false),
+ upperBound,
+ lowerBound,
+ frameType,
+ Collections.emptyList[Attribute]()
+ )
+ val rel = RelBuilder.makeWindowRel(
+ null,
+ Arrays.asList(windowFunction),
+ Collections.emptyList[ExpressionNode](),
+ Collections.emptyList[SortField](),
+ new SubstraitContext(),
+ 0L)
+ val proto = rel.toProtobuf
+ assert(proto.hasWindow, "Rel oneof should carry a
ConsistentPartitionWindowRel")
+ val window = proto.getWindow
+ assert(window.getWindowFunctionsCount === 1)
+ window.getWindowFunctions(0)
+ }
+
+ test("ROWS frame maps to BOUNDS_TYPE_ROWS (value remap) and keeps the
column_name graft") {
+ val fn = windowFunctionProto(CurrentRow, UnboundedPreceding, "ROWS",
"rows_col")
+ // Fork WindowType.ROWS was 0; 0.98 BoundsType.BOUNDS_TYPE_ROWS is 1
(UNSPECIFIED took slot 0).
+ assert(fn.getBoundsType ===
Expression.WindowFunction.BoundsType.BOUNDS_TYPE_ROWS)
+ assert(fn.getBoundsType.getNumber === 1)
+ // column_name is a Gluten graft on WindowRelFunction (0.98 has none).
+ assert(fn.getColumnName === "rows_col")
+ // UNBOUNDED PRECEDING collapses to the single `unbounded` bound; CURRENT
ROW is unchanged.
+ assert(fn.getLowerBound.hasUnbounded)
+ assert(fn.getUpperBound.hasCurrentRow)
+ }
+
+ test("RANGE frame maps to BOUNDS_TYPE_RANGE (value remap)") {
+ val fn = windowFunctionProto(CurrentRow, UnboundedPreceding, "RANGE",
"range_col")
+ // Fork WindowType.RANGE was 1; 0.98 BoundsType.BOUNDS_TYPE_RANGE is 2.
+ assert(fn.getBoundsType ===
Expression.WindowFunction.BoundsType.BOUNDS_TYPE_RANGE)
+ assert(fn.getBoundsType.getNumber === 2)
+ }
+
+ test("UNBOUNDED FOLLOWING collapses to the single unbounded bound") {
+ val fn = windowFunctionProto(UnboundedFollowing, UnboundedPreceding,
"ROWS", "both_unbounded")
+ // Both the lower (UNBOUNDED PRECEDING) and upper (UNBOUNDED FOLLOWING)
bounds collapse to
+ // `unbounded`; direction is inferred from position by the consumers.
+ assert(fn.getLowerBound.hasUnbounded)
+ assert(fn.getUpperBound.hasUnbounded)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
