From 3986fd59555ea03347951de37054c3f11c0606b9 Mon Sep 17 00:00:00 2001 From: Kaifei Yi Date: Thu, 4 Jul 2024 12:51:51 +0800 Subject: [PATCH 01/13] [CELEBORN] support celeborn 0.5.0 (#6264) Co-authored-by: yikaifei --- .github/workflows/velox_docker.yml | 6 ++++-- docs/get-started/ClickHouse.md | 2 +- docs/get-started/Velox.md | 2 +- tools/gluten-it/pom.xml | 6 ++++++ 4 files changed, 12 insertions(+), 4 deletions(-) diff --git a/.github/workflows/velox_docker.yml b/.github/workflows/velox_docker.yml index d07ceb93b3d9d..c4b24598ceb9a 100644 --- a/.github/workflows/velox_docker.yml +++ b/.github/workflows/velox_docker.yml @@ -532,7 +532,7 @@ jobs: fail-fast: false matrix: spark: [ "spark-3.2" ] - celeborn: [ "celeborn-0.4.1", "celeborn-0.3.2-incubating" ] + celeborn: [ "celeborn-0.5.0", "celeborn-0.4.1", "celeborn-0.3.2-incubating" ] runs-on: ubuntu-20.04 container: ubuntu:22.04 steps: @@ -563,8 +563,10 @@ jobs: - name: TPC-H SF1.0 && TPC-DS SF1.0 Parquet local spark3.2 with ${{ matrix.celeborn }} run: | EXTRA_PROFILE="" - if [ "${{ matrix.celeborn }}" = "celeborn-0.4.0" ]; then + if [ "${{ matrix.celeborn }}" = "celeborn-0.4.1" ]; then EXTRA_PROFILE="-Pceleborn-0.4" + elif [ "${{ matrix.celeborn }}" = "celeborn-0.5.0" ]; then + EXTRA_PROFILE="-Pceleborn-0.5" fi echo "EXTRA_PROFILE: ${EXTRA_PROFILE}" cd /opt && mkdir -p celeborn && \ diff --git a/docs/get-started/ClickHouse.md b/docs/get-started/ClickHouse.md index 38ce048fe0def..f0b7fc13b2975 100644 --- a/docs/get-started/ClickHouse.md +++ b/docs/get-started/ClickHouse.md @@ -629,7 +629,7 @@ public read-only account：gluten/hN2xX3uQ4m ### Celeborn support -Gluten with clickhouse backend supports [Celeborn](https://github.com/apache/celeborn) as remote shuffle service. Currently, the supported Celeborn versions are `0.3.x` and `0.4.0`. +Gluten with clickhouse backend supports [Celeborn](https://github.com/apache/celeborn) as remote shuffle service. Currently, the supported Celeborn versions are `0.3.x`, `0.4.x` and `0.5.0`. Below introduction is used to enable this feature. diff --git a/docs/get-started/Velox.md b/docs/get-started/Velox.md index 5f9ae2a46b19d..ff3b8f4b90f43 100644 --- a/docs/get-started/Velox.md +++ b/docs/get-started/Velox.md @@ -222,7 +222,7 @@ Currently there are several ways to asscess S3 in Spark. Please refer [Velox S3] ## Celeborn support -Gluten with velox backend supports [Celeborn](https://github.com/apache/celeborn) as remote shuffle service. Currently, the supported Celeborn versions are `0.3.x` and `0.4.0`. +Gluten with velox backend supports [Celeborn](https://github.com/apache/celeborn) as remote shuffle service. Currently, the supported Celeborn versions are `0.3.x`, `0.4.x` and `0.5.0`. Below introduction is used to enable this feature. diff --git a/tools/gluten-it/pom.xml b/tools/gluten-it/pom.xml index 71db637a8403e..c092a0ebb0e6e 100644 --- a/tools/gluten-it/pom.xml +++ b/tools/gluten-it/pom.xml @@ -170,5 +170,11 @@ 0.4.1 + + celeborn-0.5 + + 0.5.0 + + From 2cd7491d4050d9167514d6057aae22124cbff88c Mon Sep 17 00:00:00 2001 From: Kyligence Git Date: Thu, 4 Jul 2024 00:23:14 -0500 Subject: [PATCH 02/13] [GLUTEN-1632][CH]Daily Update Clickhouse Version (20240704) (#6327) * [GLUTEN-1632][CH]Daily Update Clickhouse Version (20240704) * Fix build due to https://github.com/ClickHouse/ClickHouse/pull/58661 --------- Co-authored-by: kyligence-git Co-authored-by: Chang Chen --- cpp-ch/clickhouse.version | 4 ++-- .../local-engine/Functions/SparkFunctionRegexpExtractAll.cpp | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/cpp-ch/clickhouse.version b/cpp-ch/clickhouse.version index 1630f57601877..0fb13497d01a5 100644 --- a/cpp-ch/clickhouse.version +++ b/cpp-ch/clickhouse.version @@ -1,4 +1,4 @@ CH_ORG=Kyligence -CH_BRANCH=rebase_ch/20240703 -CH_COMMIT=aa71be074ad +CH_BRANCH=rebase_ch/20240704 +CH_COMMIT=f617655ccea diff --git a/cpp-ch/local-engine/Functions/SparkFunctionRegexpExtractAll.cpp b/cpp-ch/local-engine/Functions/SparkFunctionRegexpExtractAll.cpp index ca4f1002059f8..68136713f59c6 100644 --- a/cpp-ch/local-engine/Functions/SparkFunctionRegexpExtractAll.cpp +++ b/cpp-ch/local-engine/Functions/SparkFunctionRegexpExtractAll.cpp @@ -77,7 +77,7 @@ namespace if (arguments.size() == 3) args.emplace_back(FunctionArgumentDescriptor{"index", static_cast(&isInteger), nullptr, "Integer"}); - validateFunctionArgumentTypes(*this, arguments, args); + validateFunctionArguments(*this, arguments, args); return std::make_shared(std::make_shared()); } From ffb693ed17e17b0c2a9c2aff4719ea8309df8e89 Mon Sep 17 00:00:00 2001 From: j7nhai <146867566+j7nhai@users.noreply.github.com> Date: Thu, 4 Jul 2024 17:22:20 +0800 Subject: [PATCH 03/13] [VL] Fix build Velox script incorrectly judged as successful when run make (#6331) --- ep/build-velox/src/build_velox.sh | 3 +++ 1 file changed, 3 insertions(+) diff --git a/ep/build-velox/src/build_velox.sh b/ep/build-velox/src/build_velox.sh index b55f65a98e9e3..47d0398c0830a 100755 --- a/ep/build-velox/src/build_velox.sh +++ b/ep/build-velox/src/build_velox.sh @@ -109,6 +109,9 @@ function compile { fi fi + # Maybe there is some set option in velox setup script. Run set command again. + set -exu + CXX_FLAGS='-Wno-missing-field-initializers' COMPILE_OPTION="-DCMAKE_CXX_FLAGS=\"$CXX_FLAGS\" -DVELOX_ENABLE_PARQUET=ON -DVELOX_BUILD_TESTING=OFF" if [ $BUILD_TEST_UTILS == "ON" ]; then From 7fc385dda81d4a659d36d287b49d232b9504c0b0 Mon Sep 17 00:00:00 2001 From: Zhichao Zhang Date: Thu, 4 Jul 2024 18:45:28 +0800 Subject: [PATCH 04/13] [GLUTEN-6334][CH] Support ntile window function (#6335) [CH] Support ntile window function Close #6334. --- .../backendsapi/clickhouse/CHBackend.scala | 4 +- .../clickhouse/CHSparkPlanExecApi.scala | 18 +++++++- ...enClickHouseTPCHSaltNullParquetSuite.scala | 16 +++++++ .../CommonAggregateFunctionParser.cpp | 3 -- .../aggregate_function_parser/NtileParser.cpp | 42 +++++++++++++++++++ .../aggregate_function_parser/NtileParser.h | 34 +++++++++++++++ 6 files changed, 111 insertions(+), 6 deletions(-) create mode 100644 cpp-ch/local-engine/Parser/aggregate_function_parser/NtileParser.cpp create mode 100644 cpp-ch/local-engine/Parser/aggregate_function_parser/NtileParser.h diff --git a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHBackend.scala b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHBackend.scala index cdca1b031a915..d369b8c1626f0 100644 --- a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHBackend.scala +++ b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHBackend.scala @@ -26,7 +26,7 @@ import org.apache.gluten.substrait.rel.LocalFilesNode.ReadFileFormat._ import org.apache.spark.SparkEnv import org.apache.spark.internal.Logging import org.apache.spark.sql.catalyst.catalog.BucketSpec -import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, DenseRank, Expression, Lag, Lead, Literal, NamedExpression, Rank, RowNumber} +import org.apache.spark.sql.catalyst.expressions._ import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, Partitioning} import org.apache.spark.sql.execution.SparkPlan @@ -237,7 +237,7 @@ object CHBackendSettings extends BackendSettingsApi with Logging { } wExpression.windowFunction match { - case _: RowNumber | _: AggregateExpression | _: Rank | _: DenseRank => + case _: RowNumber | _: AggregateExpression | _: Rank | _: DenseRank | _: NTile => allSupported = allSupported case l: Lag => checkLagOrLead(l.third) diff --git a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHSparkPlanExecApi.scala b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHSparkPlanExecApi.scala index 44aeba0215572..add82cbb591d8 100644 --- a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHSparkPlanExecApi.scala +++ b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHSparkPlanExecApi.scala @@ -704,7 +704,7 @@ class CHSparkPlanExecApi extends SparkPlanExecApi { val columnName = s"${aliasExpr.name}_${aliasExpr.exprId.id}" val wExpression = aliasExpr.child.asInstanceOf[WindowExpression] wExpression.windowFunction match { - case wf @ (RowNumber() | Rank(_) | DenseRank(_) | CumeDist() | PercentRank(_)) => + case wf @ (RowNumber() | Rank(_) | DenseRank(_)) => val aggWindowFunc = wf.asInstanceOf[AggregateWindowFunction] val frame = aggWindowFunc.frame.asInstanceOf[SpecifiedWindowFrame] val windowFunctionNode = ExpressionBuilder.makeWindowFunction( @@ -795,6 +795,22 @@ class CHSparkPlanExecApi extends SparkPlanExecApi { originalInputAttributes.asJava ) windowExpressionNodes.add(windowFunctionNode) + case wf @ NTile(buckets: Expression) => + val frame = wExpression.windowSpec.frameSpecification.asInstanceOf[SpecifiedWindowFrame] + val childrenNodeList = new JArrayList[ExpressionNode]() + val literal = buckets.asInstanceOf[Literal] + childrenNodeList.add(LiteralTransformer(literal).doTransform(args)) + val windowFunctionNode = ExpressionBuilder.makeWindowFunction( + WindowFunctionsBuilder.create(args, wf).toInt, + childrenNodeList, + columnName, + ConverterUtils.getTypeNode(wf.dataType, wf.nullable), + frame.upper, + frame.lower, + frame.frameType.sql, + originalInputAttributes.asJava + ) + windowExpressionNodes.add(windowFunctionNode) case _ => throw new GlutenNotSupportException( "unsupported window function type: " + diff --git a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseTPCHSaltNullParquetSuite.scala b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseTPCHSaltNullParquetSuite.scala index c0f37b08616ea..b0d3e1bdb8661 100644 --- a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseTPCHSaltNullParquetSuite.scala +++ b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseTPCHSaltNullParquetSuite.scala @@ -978,6 +978,22 @@ class GlutenClickHouseTPCHSaltNullParquetSuite extends GlutenClickHouseTPCHAbstr compareResultsAgainstVanillaSpark(sql, true, { _ => }) } + test("window ntile") { + val sql = + """ + | select n_regionkey, n_nationkey, + | first_value(n_nationkey) over (partition by n_regionkey order by n_nationkey) as + | first_v, + | ntile(4) over (partition by n_regionkey order by n_nationkey) as ntile_v + | from + | ( + | select n_regionkey, if(n_nationkey = 1, null, n_nationkey) as n_nationkey from nation + | ) as t + | order by n_regionkey, n_nationkey + """.stripMargin + compareResultsAgainstVanillaSpark(sql, true, { _ => }) + } + test("window first value with nulls") { val sql = """ diff --git a/cpp-ch/local-engine/Parser/aggregate_function_parser/CommonAggregateFunctionParser.cpp b/cpp-ch/local-engine/Parser/aggregate_function_parser/CommonAggregateFunctionParser.cpp index 1619c74106d1c..e7d6e1b9bd733 100644 --- a/cpp-ch/local-engine/Parser/aggregate_function_parser/CommonAggregateFunctionParser.cpp +++ b/cpp-ch/local-engine/Parser/aggregate_function_parser/CommonAggregateFunctionParser.cpp @@ -42,8 +42,5 @@ REGISTER_COMMON_AGGREGATE_FUNCTION_PARSER(LastIgnoreNull, last_ignore_null, last REGISTER_COMMON_AGGREGATE_FUNCTION_PARSER(DenseRank, dense_rank, dense_rank) REGISTER_COMMON_AGGREGATE_FUNCTION_PARSER(Rank, rank, rank) REGISTER_COMMON_AGGREGATE_FUNCTION_PARSER(RowNumber, row_number, row_number) -REGISTER_COMMON_AGGREGATE_FUNCTION_PARSER(Ntile, ntile, ntile) -REGISTER_COMMON_AGGREGATE_FUNCTION_PARSER(PercentRank, percent_rank, percent_rank) -REGISTER_COMMON_AGGREGATE_FUNCTION_PARSER(CumeDist, cume_dist, cume_dist) REGISTER_COMMON_AGGREGATE_FUNCTION_PARSER(CountDistinct, count_distinct, uniqExact) } diff --git a/cpp-ch/local-engine/Parser/aggregate_function_parser/NtileParser.cpp b/cpp-ch/local-engine/Parser/aggregate_function_parser/NtileParser.cpp new file mode 100644 index 0000000000000..49a59c6570fba --- /dev/null +++ b/cpp-ch/local-engine/Parser/aggregate_function_parser/NtileParser.cpp @@ -0,0 +1,42 @@ +/* + * 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 "NtileParser.h" +#include +#include +#include + +namespace local_engine +{ +DB::ActionsDAG::NodeRawConstPtrs +NtileParser::parseFunctionArguments(const CommonFunctionInfo & func_info, const String & /*ch_func_name*/, DB::ActionsDAGPtr & actions_dag) const +{ + if (func_info.arguments.size() != 1) + throw Exception(ErrorCodes::BAD_ARGUMENTS, "Function ntile takes exactly one argument"); + DB::ActionsDAG::NodeRawConstPtrs args; + + const auto & arg0 = func_info.arguments[0].value(); + auto [data_type, field] = parseLiteral(arg0.literal()); + if (!(DB::WhichDataType(data_type).isInt32())) + throw Exception(ErrorCodes::BAD_ARGUMENTS, "ntile's argument must be i32"); + Int32 field_index = static_cast(field.get()); + // For CH, the data type of the args[0] must be the UInt32 + const auto * index_node = addColumnToActionsDAG(actions_dag, std::make_shared(), field_index); + args.emplace_back(index_node); + return args; +} +AggregateFunctionParserRegister ntile_register; +} diff --git a/cpp-ch/local-engine/Parser/aggregate_function_parser/NtileParser.h b/cpp-ch/local-engine/Parser/aggregate_function_parser/NtileParser.h new file mode 100644 index 0000000000000..441de23532478 --- /dev/null +++ b/cpp-ch/local-engine/Parser/aggregate_function_parser/NtileParser.h @@ -0,0 +1,34 @@ +/* + * 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 + +namespace local_engine +{ +class NtileParser : public AggregateFunctionParser +{ +public: + explicit NtileParser(SerializedPlanParser * plan_parser_) : AggregateFunctionParser(plan_parser_) { } + ~NtileParser() override = default; + static constexpr auto name = "ntile"; + String getName() const override { return name; } + String getCHFunctionName(const CommonFunctionInfo &) const override { return "ntile"; } + String getCHFunctionName(DB::DataTypes &) const override { return "ntile"; } + DB::ActionsDAG::NodeRawConstPtrs parseFunctionArguments( + const CommonFunctionInfo & func_info, const String & ch_func_name, DB::ActionsDAGPtr & actions_dag) const override; +}; +} From a091ac5010968274f3f8b561d7e36054f92ed151 Mon Sep 17 00:00:00 2001 From: Mingliang Zhu Date: Thu, 4 Jul 2024 19:18:26 +0800 Subject: [PATCH 05/13] [CORE] Drop redundant partial sort which has pre-project when offloading sort agg (#6294) --- .../VeloxAggregateFunctionsSuite.scala | 26 +++++++++++++++++++ .../apache/gluten/execution/SortUtils.scala | 24 ++++++++++++++--- .../columnar/OffloadSingleNode.scala | 8 +----- 3 files changed, 47 insertions(+), 11 deletions(-) diff --git a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxAggregateFunctionsSuite.scala b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxAggregateFunctionsSuite.scala index ae6306cc0d4a1..992106d131e67 100644 --- a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxAggregateFunctionsSuite.scala +++ b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxAggregateFunctionsSuite.scala @@ -1135,6 +1135,32 @@ abstract class VeloxAggregateFunctionsSuite extends VeloxWholeStageTransformerSu df.select(max(col("txn"))).collect } + + test("drop redundant partial sort which has pre-project when offload sortAgg") { + // Spark 3.2 does not have this configuration, but it does not affect the test results. + withSQLConf("spark.sql.test.forceApplySortAggregate" -> "true") { + withTempView("t1") { + Seq((-1, 2), (-1, 3), (2, 3), (3, 4), (-3, 5), (4, 5)) + .toDF("c1", "c2") + .createOrReplaceTempView("t1") + runQueryAndCompare("select c2, sum(if(c1<0,0,c1)) from t1 group by c2") { + df => + { + assert( + getExecutedPlan(df).count( + plan => { + plan.isInstanceOf[HashAggregateExecTransformer] + }) == 2) + assert( + getExecutedPlan(df).count( + plan => { + plan.isInstanceOf[SortExecTransformer] + }) == 0) + } + } + } + } + } } class VeloxAggregateFunctionsDefaultSuite extends VeloxAggregateFunctionsSuite { diff --git a/gluten-core/src/main/scala/org/apache/gluten/execution/SortUtils.scala b/gluten-core/src/main/scala/org/apache/gluten/execution/SortUtils.scala index 2c0ad1b0a59a1..b01c71738e75a 100644 --- a/gluten-core/src/main/scala/org/apache/gluten/execution/SortUtils.scala +++ b/gluten-core/src/main/scala/org/apache/gluten/execution/SortUtils.scala @@ -23,11 +23,27 @@ import org.apache.spark.sql.execution.{ProjectExec, SortExec, SparkPlan} object SortUtils { def dropPartialSort(plan: SparkPlan): SparkPlan = plan match { case RewrittenNodeWall(p) => RewrittenNodeWall(dropPartialSort(p)) - case sort: SortExec if !sort.global => sort.child + case PartialSortLike(child) => child // from pre/post project-pulling - case ProjectExec(_, SortExec(_, false, ProjectExec(_, p), _)) - if plan.outputSet == p.outputSet => - p + case ProjectLike(PartialSortLike(ProjectLike(child))) if plan.outputSet == child.outputSet => + child + case ProjectLike(PartialSortLike(child)) => plan.withNewChildren(Seq(child)) case _ => plan } } + +object PartialSortLike { + def unapply(plan: SparkPlan): Option[SparkPlan] = plan match { + case sort: SortExecTransformer if !sort.global => Some(sort.child) + case sort: SortExec if !sort.global => Some(sort.child) + case _ => None + } +} + +object ProjectLike { + def unapply(plan: SparkPlan): Option[SparkPlan] = plan match { + case project: ProjectExecTransformer => Some(project.child) + case project: ProjectExec => Some(project.child) + case _ => None + } +} diff --git a/gluten-core/src/main/scala/org/apache/gluten/extension/columnar/OffloadSingleNode.scala b/gluten-core/src/main/scala/org/apache/gluten/extension/columnar/OffloadSingleNode.scala index 7a4222b5cb382..62c72af792e91 100644 --- a/gluten-core/src/main/scala/org/apache/gluten/extension/columnar/OffloadSingleNode.scala +++ b/gluten-core/src/main/scala/org/apache/gluten/extension/columnar/OffloadSingleNode.scala @@ -425,13 +425,7 @@ object OffloadOthers { ColumnarCoalesceExec(plan.numPartitions, plan.child) case plan: SortAggregateExec => logDebug(s"Columnar Processing for ${plan.getClass} is currently supported.") - HashAggregateExecBaseTransformer.from(plan) { - case sort: SortExecTransformer if !sort.global => - sort.child - case sort: SortExec if !sort.global => - sort.child - case other => other - } + HashAggregateExecBaseTransformer.from(plan)(SortUtils.dropPartialSort) case plan: ObjectHashAggregateExec => logDebug(s"Columnar Processing for ${plan.getClass} is currently supported.") HashAggregateExecBaseTransformer.from(plan)() From 01526e62e84b74dd48084ea36d55df3467344c8d Mon Sep 17 00:00:00 2001 From: Mingliang Zhu Date: Thu, 4 Jul 2024 19:20:02 +0800 Subject: [PATCH 06/13] [VL] RAS: Remove NoopFilter that has same output schema with child (#6324) --- .../extension/columnar/enumerated/RemoveFilter.scala | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/gluten-core/src/main/scala/org/apache/gluten/extension/columnar/enumerated/RemoveFilter.scala b/gluten-core/src/main/scala/org/apache/gluten/extension/columnar/enumerated/RemoveFilter.scala index e2b8439fd218e..8b8441e8d6ce7 100644 --- a/gluten-core/src/main/scala/org/apache/gluten/extension/columnar/enumerated/RemoveFilter.scala +++ b/gluten-core/src/main/scala/org/apache/gluten/extension/columnar/enumerated/RemoveFilter.scala @@ -41,9 +41,12 @@ object RemoveFilter extends RasRule[SparkPlan] { override def shift(node: SparkPlan): Iterable[SparkPlan] = { val filter = node.asInstanceOf[FilterExecTransformerBase] if (filter.isNoop()) { - val out = NoopFilter(filter.child, filter.output) - out.copyTagsFrom(filter) - return List(out) + if (filter.output != filter.child.output) { + val out = NoopFilter(filter.child, filter.output) + out.copyTagsFrom(filter) + return List(out) + } + return List(filter.child) } List.empty } From 663ae51f5d42415769354be465a09609441ffa01 Mon Sep 17 00:00:00 2001 From: Shuai li Date: Thu, 4 Jul 2024 20:28:27 +0800 Subject: [PATCH 07/13] [GLUTEN-6333][CH] Support rangepartitioning by timestamptype (#6336) [CH] Support rangepartitioning by timestamptype --- .../RangePartitionerBoundsGenerator.scala | 2 ++ ...tenClickHouseDatetimeExpressionSuite.scala | 22 +++++++++++++++++++ cpp-ch/local-engine/Parser/TypeParser.cpp | 3 ++- .../local-engine/Shuffle/SelectorBuilder.cpp | 5 +++++ 4 files changed, 31 insertions(+), 1 deletion(-) diff --git a/backends-clickhouse/src/main/scala/org/apache/spark/shuffle/utils/RangePartitionerBoundsGenerator.scala b/backends-clickhouse/src/main/scala/org/apache/spark/shuffle/utils/RangePartitionerBoundsGenerator.scala index 61fbc86b36cc9..87c6ae343d4c1 100644 --- a/backends-clickhouse/src/main/scala/org/apache/spark/shuffle/utils/RangePartitionerBoundsGenerator.scala +++ b/backends-clickhouse/src/main/scala/org/apache/spark/shuffle/utils/RangePartitionerBoundsGenerator.scala @@ -199,6 +199,7 @@ class RangePartitionerBoundsGenerator[K: Ordering: ClassTag, V]( case d: DecimalType => val decimal = row.getDecimal(i, d.precision, d.scale).toString() node.put("value", decimal) + case _: TimestampType => node.put("value", row.getLong(i)) case _ => throw new IllegalArgumentException( s"Unsupported data type ${ordering.dataType.toString}") @@ -244,6 +245,7 @@ object RangePartitionerBoundsGenerator { case _: StringType => true case _: DateType => true case _: DecimalType => true + case _: TimestampType => true case _ => false } } diff --git a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseDatetimeExpressionSuite.scala b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseDatetimeExpressionSuite.scala index 53416607521ef..a1749efb18b2a 100644 --- a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseDatetimeExpressionSuite.scala +++ b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseDatetimeExpressionSuite.scala @@ -162,4 +162,26 @@ class GlutenClickHouseDatetimeExpressionSuite |""".stripMargin compareResultsAgainstVanillaSpark(sql, true, { _ => }) } + + test("support range partition by timestamp") { + import testImplicits._ + val df = Seq( + (1, Timestamp.valueOf("2015-07-22 10:01:40.123456")), + (2, Timestamp.valueOf("2014-12-31 05:29:06.123456")), + (3, Timestamp.valueOf("2015-07-22 16:01:40.123456")), + (4, Timestamp.valueOf("2012-02-29 23:01:40.123456")) + ).toDF("i", "t") + + df.createOrReplaceTempView("test") + + val sql = + s""" + | select + | /** repartition(2) */ + | * + | from test + | order by t + |""".stripMargin + compareResultsAgainstVanillaSpark(sql, compareResult = true, { _ => }) + } } diff --git a/cpp-ch/local-engine/Parser/TypeParser.cpp b/cpp-ch/local-engine/Parser/TypeParser.cpp index 3ad19bb2bd733..0d5e54bb17576 100644 --- a/cpp-ch/local-engine/Parser/TypeParser.cpp +++ b/cpp-ch/local-engine/Parser/TypeParser.cpp @@ -59,7 +59,8 @@ std::unordered_map TypeParser::type_names_mapping {"FloatType", "Float32"}, {"DoubleType", "Float64"}, {"StringType", "String"}, - {"DateType", "Date32"}}; + {"DateType", "Date32"}, + {"TimestampType", "DateTime64"}}; String TypeParser::getCHTypeName(const String & spark_type_name) { diff --git a/cpp-ch/local-engine/Shuffle/SelectorBuilder.cpp b/cpp-ch/local-engine/Shuffle/SelectorBuilder.cpp index 7e3642dacd523..6804770c34c15 100644 --- a/cpp-ch/local-engine/Shuffle/SelectorBuilder.cpp +++ b/cpp-ch/local-engine/Shuffle/SelectorBuilder.cpp @@ -291,6 +291,11 @@ void RangeSelectorBuilder::initRangeBlock(Poco::JSON::Array::Ptr range_bounds) int val = field_value.convert(); col->insert(val); } + else if (const auto * timestamp = dynamic_cast(type_info.inner_type.get())) + { + auto value = field_value.convert(); + col->insert(DecimalField(value, 6)); + } else if (const auto * decimal32 = dynamic_cast *>(type_info.inner_type.get())) { auto value = decimal32->parseFromString(field_value.convert()); From f85b3d68c901d9f6b7016fd6121c1e8cacd6d4a6 Mon Sep 17 00:00:00 2001 From: zhixingheyi-tian Date: Thu, 4 Jul 2024 22:57:52 +0800 Subject: [PATCH 08/13] [VL] Support tencentos 2.4 (#5207) Co-authored-by: xiangxshen --- .../apache/gluten/backendsapi/velox/VeloxListenerApi.scala | 4 +++- dev/package.sh | 4 +++- ep/build-velox/src/build_velox.sh | 7 +++++++ ep/build-velox/src/get_velox.sh | 1 + 4 files changed, 14 insertions(+), 2 deletions(-) diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxListenerApi.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxListenerApi.scala index e1abbdd7c6b7e..e5c3cb084819d 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxListenerApi.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxListenerApi.scala @@ -78,6 +78,8 @@ class VeloxListenerApi extends ListenerApi { new SharedLibraryLoaderCentos8 } else if (systemName.contains("Anolis") && systemVersion.startsWith("7")) { new SharedLibraryLoaderCentos7 + } else if (system.contains("tencentos") && system.contains("2.4")) { + new SharedLibraryLoaderCentos7 } else if (system.contains("tencentos") && system.contains("3.2")) { new SharedLibraryLoaderCentos8 } else if (systemName.contains("Red Hat") && systemVersion.startsWith("9")) { @@ -94,7 +96,7 @@ class VeloxListenerApi extends ListenerApi { throw new GlutenException( s"Found unsupported OS($systemName, $systemVersion)! Currently, Gluten's Velox backend" + " only supports Ubuntu 20.04/22.04, CentOS 7/8, " + - "Alibaba Cloud Linux 2/3 & Anolis 7/8, tencentos 3.2, RedHat 7/8/9, " + + "Alibaba Cloud Linux 2/3 & Anolis 7/8, tencentos 2.4/3.2, RedHat 7/8, " + "Debian 11/12.") } } diff --git a/dev/package.sh b/dev/package.sh index 1b9ca85e9590c..7e7e793bdabdd 100755 --- a/dev/package.sh +++ b/dev/package.sh @@ -68,7 +68,9 @@ elif [ "$LINUX_OS" == "alinux" ]; then process_setup_centos_7 fi elif [ "$LINUX_OS" == "tencentos" ]; then - if [ "$VERSION" == "3.2" ]; then + if [ "$VERSION" == "2.4" ]; then + process_setup_centos_7 + elif [ "$VERSION" == "3.2" ]; then process_setup_centos_8 fi elif [ "$LINUX_OS" == "debian" ]; then diff --git a/ep/build-velox/src/build_velox.sh b/ep/build-velox/src/build_velox.sh index 47d0398c0830a..747eec3f8b6fb 100755 --- a/ep/build-velox/src/build_velox.sh +++ b/ep/build-velox/src/build_velox.sh @@ -265,6 +265,13 @@ function setup_linux { esac elif [[ "$LINUX_DISTRIBUTION" == "tencentos" ]]; then case "$LINUX_VERSION_ID" in + 2.4) + scripts/setup-centos7.sh + set +u + export PKG_CONFIG_PATH=/usr/local/lib64/pkgconfig:/usr/local/lib/pkgconfig:/usr/lib64/pkgconfig:/usr/lib/pkgconfig:$PKG_CONFIG_PATH + source /opt/rh/devtoolset-9/enable + set -u + ;; 3.2) scripts/setup-centos8.sh ;; *) echo "Unsupported tencentos version: $LINUX_VERSION_ID" diff --git a/ep/build-velox/src/get_velox.sh b/ep/build-velox/src/get_velox.sh index 6cd62332a3ccf..9193d7f898487 100755 --- a/ep/build-velox/src/get_velox.sh +++ b/ep/build-velox/src/get_velox.sh @@ -286,6 +286,7 @@ function setup_linux { esac elif [[ "$LINUX_DISTRIBUTION" == "tencentos" ]]; then case "$LINUX_VERSION_ID" in + 2.4) process_setup_centos7 ;; 3.2) process_setup_tencentos32 ;; *) echo "Unsupport tencentos version: $LINUX_VERSION_ID" From ff0b4733a8282476bbb4dbcca2d46458ca04b36b Mon Sep 17 00:00:00 2001 From: Gluten Performance Bot <137994563+GlutenPerfBot@users.noreply.github.com> Date: Fri, 5 Jul 2024 00:39:12 +0800 Subject: [PATCH 09/13] [VL] Daily Update Velox Version (2024_07_04) (#6328) 26f001441 by xiaoxmeng, Fix tsan race in cache fuzzer test (10391) 1705bde41 by xiaoxmeng, Add exchange request count metrics (10389) 02cbe7476 by Bikramjeet Vig, Fix NaN handling in comparison functions (10165) decd91eb5 by Kevin Wilfong, Call commitNull in SimpleFunctionAdapter on exceptions (10377) a03396845 by Kevin Wilfong, StringWriter needs to implement finalizeNull (10376) 0cb715ea3 by youxiduo, Reject duplicate sorting keys (10040) 4e39b06c7 by gaoyangxiaozhu, Add Spark raise_error function (10110) 63ccecaa2 by rui-mo, Check determinism using function name only (10241) --- .../gluten/execution/TestOperator.scala | 20 +++++++++---------- ep/build-velox/src/get_velox.sh | 2 +- .../GlutenSQLWindowFunctionSuite.scala | 2 +- .../GlutenSQLWindowFunctionSuite.scala | 2 +- .../utils/velox/VeloxTestSettings.scala | 1 + .../GlutenSQLWindowFunctionSuite.scala | 2 +- .../utils/velox/VeloxTestSettings.scala | 1 + .../GlutenSQLWindowFunctionSuite.scala | 6 +++--- 8 files changed, 19 insertions(+), 17 deletions(-) diff --git a/backends-velox/src/test/scala/org/apache/gluten/execution/TestOperator.scala b/backends-velox/src/test/scala/org/apache/gluten/execution/TestOperator.scala index c010b9128ce1c..c6e72197d6c7d 100644 --- a/backends-velox/src/test/scala/org/apache/gluten/execution/TestOperator.scala +++ b/backends-velox/src/test/scala/org/apache/gluten/execution/TestOperator.scala @@ -422,18 +422,18 @@ class TestOperator extends VeloxWholeStageTransformerSuite with AdaptiveSparkPla } // Test same partition/ordering keys. - runQueryAndCompare( - "select avg(l_partkey) over" + - " (partition by l_suppkey order by l_suppkey) from lineitem ") { - checkGlutenOperatorMatch[WindowExecTransformer] - } +// runQueryAndCompare( +// "select avg(l_partkey) over" + +// " (partition by l_suppkey order by l_suppkey) from lineitem ") { +// checkGlutenOperatorMatch[WindowExecTransformer] +// } // Test overlapping partition/ordering keys. - runQueryAndCompare( - "select avg(l_partkey) over" + - " (partition by l_suppkey order by l_suppkey, l_orderkey) from lineitem ") { - checkGlutenOperatorMatch[WindowExecTransformer] - } +// runQueryAndCompare( +// "select avg(l_partkey) over" + +// " (partition by l_suppkey order by l_suppkey, l_orderkey) from lineitem ") { +// checkGlutenOperatorMatch[WindowExecTransformer] +// } } } } diff --git a/ep/build-velox/src/get_velox.sh b/ep/build-velox/src/get_velox.sh index 9193d7f898487..07694a36102d2 100755 --- a/ep/build-velox/src/get_velox.sh +++ b/ep/build-velox/src/get_velox.sh @@ -17,7 +17,7 @@ set -exu VELOX_REPO=https://github.com/oap-project/velox.git -VELOX_BRANCH=2024_07_03 +VELOX_BRANCH=2024_07_04 VELOX_HOME="" #Set on run gluten on HDFS diff --git a/gluten-ut/spark32/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala b/gluten-ut/spark32/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala index 6665174207b09..e26c1dc41a9a2 100644 --- a/gluten-ut/spark32/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala +++ b/gluten-ut/spark32/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala @@ -47,7 +47,7 @@ class GlutenSQLWindowFunctionSuite extends SQLWindowFunctionSuite with GlutenSQL Row(95337, 12, decimal(915.61)) ) - testGluten("Literal in window partition by and sort") { + ignoreGluten("Literal in window partition by and sort") { withTable("customer") { val rdd = spark.sparkContext.parallelize(customerData) val customerDF = spark.createDataFrame(rdd, customerSchema) diff --git a/gluten-ut/spark33/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala b/gluten-ut/spark33/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala index 6665174207b09..e26c1dc41a9a2 100644 --- a/gluten-ut/spark33/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala +++ b/gluten-ut/spark33/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala @@ -47,7 +47,7 @@ class GlutenSQLWindowFunctionSuite extends SQLWindowFunctionSuite with GlutenSQL Row(95337, 12, decimal(915.61)) ) - testGluten("Literal in window partition by and sort") { + ignoreGluten("Literal in window partition by and sort") { withTable("customer") { val rdd = spark.sparkContext.parallelize(customerData) val customerDF = spark.createDataFrame(rdd, customerSchema) diff --git a/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala b/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala index 0da19922ffdaf..941f7b89b2d5c 100644 --- a/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala +++ b/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala @@ -1182,6 +1182,7 @@ class VeloxTestSettings extends BackendTestSettings { enableSuite[GlutenDataFrameToSchemaSuite] enableSuite[GlutenDatasetUnpivotSuite] enableSuite[GlutenLateralColumnAliasSuite] + .exclude("Aggregate expressions containing no aggregate or grouping expressions still resolves") enableSuite[GlutenParametersSuite] enableSuite[GlutenResolveDefaultColumnsSuite] enableSuite[GlutenSubqueryHintPropagationSuite] diff --git a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala b/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala index 6665174207b09..e26c1dc41a9a2 100644 --- a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala +++ b/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala @@ -47,7 +47,7 @@ class GlutenSQLWindowFunctionSuite extends SQLWindowFunctionSuite with GlutenSQL Row(95337, 12, decimal(915.61)) ) - testGluten("Literal in window partition by and sort") { + ignoreGluten("Literal in window partition by and sort") { withTable("customer") { val rdd = spark.sparkContext.parallelize(customerData) val customerDF = spark.createDataFrame(rdd, customerSchema) diff --git a/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala b/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala index e54aca34ec757..f3bd7ff6c752d 100644 --- a/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala +++ b/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala @@ -1197,6 +1197,7 @@ class VeloxTestSettings extends BackendTestSettings { enableSuite[GlutenDataFrameToSchemaSuite] enableSuite[GlutenDatasetUnpivotSuite] enableSuite[GlutenLateralColumnAliasSuite] + .exclude("Aggregate expressions containing no aggregate or grouping expressions still resolves") enableSuite[GlutenParametersSuite] enableSuite[GlutenResolveDefaultColumnsSuite] enableSuite[GlutenSubqueryHintPropagationSuite] diff --git a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala index 89a4351744ef4..e61c084ab7db4 100644 --- a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala +++ b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala @@ -47,7 +47,7 @@ class GlutenSQLWindowFunctionSuite extends SQLWindowFunctionSuite with GlutenSQL Row(95337, 12, decimal(915.61)) ) - testGluten("Literal in window partition by and sort") { + ignoreGluten("Literal in window partition by and sort") { withTable("customer") { val rdd = spark.sparkContext.parallelize(customerData) val customerDF = spark.createDataFrame(rdd, customerSchema) @@ -93,7 +93,7 @@ class GlutenSQLWindowFunctionSuite extends SQLWindowFunctionSuite with GlutenSQL } } - testGluten("Filter on row number") { + ignoreGluten("Filter on row number") { withTable("customer") { val rdd = spark.sparkContext.parallelize(customerData) val customerDF = spark.createDataFrame(rdd, customerSchema) @@ -137,7 +137,7 @@ class GlutenSQLWindowFunctionSuite extends SQLWindowFunctionSuite with GlutenSQL } } - testGluten("Filter on rank") { + ignoreGluten("Filter on rank") { withTable("customer") { val rdd = spark.sparkContext.parallelize(customerData) val customerDF = spark.createDataFrame(rdd, customerSchema) From 995145e93cb8c930501e69156ca57211d9932d2a Mon Sep 17 00:00:00 2001 From: lgbo Date: Fri, 5 Jul 2024 08:34:20 +0800 Subject: [PATCH 10/13] support sort_array (#6323) --- .../clickhouse/CHSparkPlanExecApi.scala | 9 + .../Functions/SparkFunctionArraySort.cpp | 223 ++++++++++++++---- .../Functions/SparkFunctionSortArray.cpp | 88 +++++++ ...onArraySort.h => SparkFunctionSortArray.h} | 14 +- .../arrayHighOrderFunctions.cpp | 144 +++++++++++ .../scalar_function_parser/sortArray.cpp | 4 +- .../gluten/backendsapi/SparkPlanExecApi.scala | 9 + .../expression/ExpressionConverter.scala | 13 + .../expression/ExpressionMappings.scala | 1 + .../gluten/expression/ExpressionNames.scala | 1 + 10 files changed, 454 insertions(+), 52 deletions(-) create mode 100644 cpp-ch/local-engine/Functions/SparkFunctionSortArray.cpp rename cpp-ch/local-engine/Functions/{SparkFunctionArraySort.h => SparkFunctionSortArray.h} (86%) diff --git a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHSparkPlanExecApi.scala b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHSparkPlanExecApi.scala index add82cbb591d8..f5feade886b96 100644 --- a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHSparkPlanExecApi.scala +++ b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHSparkPlanExecApi.scala @@ -876,6 +876,15 @@ class CHSparkPlanExecApi extends SparkPlanExecApi { GenericExpressionTransformer(substraitExprName, Seq(argument, function), expr) } + /** Transform array sort to Substrait. */ + override def genArraySortTransformer( + substraitExprName: String, + argument: ExpressionTransformer, + function: ExpressionTransformer, + expr: ArraySort): ExpressionTransformer = { + GenericExpressionTransformer(substraitExprName, Seq(argument, function), expr) + } + override def genPreProjectForGenerate(generate: GenerateExec): SparkPlan = generate override def genPostProjectForGenerate(generate: GenerateExec): SparkPlan = generate diff --git a/cpp-ch/local-engine/Functions/SparkFunctionArraySort.cpp b/cpp-ch/local-engine/Functions/SparkFunctionArraySort.cpp index 126b84eaaf95d..1371ec60e1796 100644 --- a/cpp-ch/local-engine/Functions/SparkFunctionArraySort.cpp +++ b/cpp-ch/local-engine/Functions/SparkFunctionArraySort.cpp @@ -14,75 +14,212 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -#include +#include #include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include -namespace DB +namespace DB::ErrorCodes { + extern const int TOO_FEW_ARGUMENTS_FOR_FUNCTION; + extern const int ILLEGAL_TYPE_OF_ARGUMENT; + extern const int TYPE_MISMATCH; + extern const int ILLEGAL_COLUMN; +} -namespace ErrorCodes +/// The usage of `arraySort` in CH is different from Spark's `sort_array` function. +/// We need to implement a custom function to sort arrays. +namespace local_engine { - extern const int LOGICAL_ERROR; -} -namespace +struct LambdaLess { + const DB::IColumn & column; + DB::DataTypePtr type; + const DB::ColumnFunction & lambda; + explicit LambdaLess(const DB::IColumn & column_, DB::DataTypePtr type_, const DB::ColumnFunction & lambda_) + : column(column_), type(type_), lambda(lambda_) {} + + /// May not efficient + bool operator()(size_t lhs, size_t rhs) const + { + /// The column name seems not matter. + auto left_value_col = DB::ColumnWithTypeAndName(oneRowColumn(lhs), type, "left"); + auto right_value_col = DB::ColumnWithTypeAndName(oneRowColumn(rhs), type, "right"); + auto cloned_lambda = lambda.cloneResized(1); + auto * lambda_ = typeid_cast(cloned_lambda.get()); + lambda_->appendArguments({std::move(left_value_col), std::move(right_value_col)}); + auto compare_res_col = lambda_->reduce(); + DB::Field field; + compare_res_col.column->get(0, field); + return field.get() < 0; + } +private: + ALWAYS_INLINE DB::ColumnPtr oneRowColumn(size_t i) const + { + auto res = column.cloneEmpty(); + res->insertFrom(column, i); + return std::move(res); + } +}; -template struct Less { - const IColumn & column; + const DB::IColumn & column; - explicit Less(const IColumn & column_) : column(column_) { } + explicit Less(const DB::IColumn & column_) : column(column_) { } bool operator()(size_t lhs, size_t rhs) const { - if constexpr (positive) - /* - Note: We use nan_direction_hint=-1 for ascending sort to make NULL the least value. - However, NaN is also considered the least value, - which results in different sorting results compared to Spark since Spark treats NaN as the greatest value. - For now, we are temporarily ignoring this issue because cases with NaN are rare, - and aligning with Spark would require tricky modifications to the CH underlying code. - */ - return column.compareAt(lhs, rhs, column, -1) < 0; - else - return column.compareAt(lhs, rhs, column, -1) > 0; + return column.compareAt(lhs, rhs, column, 1) < 0; } }; -} - -template -ColumnPtr SparkArraySortImpl::execute( - const ColumnArray & array, - ColumnPtr mapped, - const ColumnWithTypeAndName * fixed_arguments [[maybe_unused]]) +class FunctionSparkArraySort : public DB::IFunction { - const ColumnArray::Offsets & offsets = array.getOffsets(); +public: + static constexpr auto name = "arraySortSpark"; + static DB::FunctionPtr create(DB::ContextPtr /*context*/) { return std::make_shared(); } - size_t size = offsets.size(); - size_t nested_size = array.getData().size(); - IColumn::Permutation permutation(nested_size); + bool isVariadic() const override { return true; } + size_t getNumberOfArguments() const override { return 0; } + bool isSuitableForShortCircuitArgumentsExecution(const DB::DataTypesWithConstInfo &) const override { return true; } - for (size_t i = 0; i < nested_size; ++i) - permutation[i] = i; + bool useDefaultImplementationForNulls() const override { return false; } + bool useDefaultImplementationForLowCardinalityColumns() const override { return false; } - ColumnArray::Offset current_offset = 0; - for (size_t i = 0; i < size; ++i) + void getLambdaArgumentTypes(DB::DataTypes & arguments) const override { - auto next_offset = offsets[i]; - ::sort(&permutation[current_offset], &permutation[next_offset], Less(*mapped)); - current_offset = next_offset; + if (arguments.size() < 2) + throw DB::Exception(DB::ErrorCodes::TOO_FEW_ARGUMENTS_FOR_FUNCTION, "Function {} requires as arguments a lambda function and an array", getName()); + + if (arguments.size() > 1) + { + const auto * lambda_function_type = DB::checkAndGetDataType(arguments[0].get()); + if (!lambda_function_type || lambda_function_type->getArgumentTypes().size() != 2) + throw DB::Exception(DB::ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT, "First argument of function {} must be a lambda function with 2 arguments, found {} instead.", + getName(), arguments[0]->getName()); + auto array_nesteed_type = DB::checkAndGetDataType(arguments.back().get())->getNestedType(); + DB::DataTypes lambda_args = {array_nesteed_type, array_nesteed_type}; + arguments[0] = std::make_shared(lambda_args); + } } - return ColumnArray::create(array.getData().permute(permutation, 0), array.getOffsetsPtr()); -} + DB::DataTypePtr getReturnTypeImpl(const DB::ColumnsWithTypeAndName & arguments) const override + { + if (arguments.size() > 1) + { + const auto * lambda_function_type = checkAndGetDataType(arguments[0].type.get()); + if (!lambda_function_type) + throw DB::Exception(DB::ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT, "First argument for function {} must be a function", getName()); + } + + return arguments.back().type; + } + + DB::ColumnPtr executeImpl(const DB::ColumnsWithTypeAndName & arguments, const DB::DataTypePtr &, size_t input_rows_count) const override + { + auto array_col = arguments.back().column; + auto array_type = arguments.back().type; + DB::ColumnPtr null_map = nullptr; + if (const auto * null_col = typeid_cast(array_col.get())) + { + null_map = null_col->getNullMapColumnPtr(); + array_col = null_col->getNestedColumnPtr(); + array_type = typeid_cast(array_type.get())->getNestedType(); + } + + const auto * array_col_concrete = DB::checkAndGetColumn(array_col.get()); + if (!array_col_concrete) + { + const auto * aray_col_concrete_const = DB::checkAndGetColumnConst(array_col.get()); + if (!aray_col_concrete_const) + { + throw DB::Exception(DB::ErrorCodes::ILLEGAL_COLUMN, "Expected array column, found {}", array_col->getName()); + } + array_col = DB::recursiveRemoveLowCardinality(aray_col_concrete_const->convertToFullColumn()); + array_col_concrete = DB::checkAndGetColumn(array_col.get()); + } + auto array_nested_type = DB::checkAndGetDataType(array_type.get())->getNestedType(); + + DB::ColumnPtr sorted_array_col = nullptr; + if (arguments.size() > 1) + sorted_array_col = executeWithLambda(*array_col_concrete, array_nested_type, *checkAndGetColumn(arguments[0].column.get())); + else + sorted_array_col = executeWithoutLambda(*array_col_concrete); + + if (null_map) + { + sorted_array_col = DB::ColumnNullable::create(sorted_array_col, null_map); + } + return sorted_array_col; + } +private: + static DB::ColumnPtr executeWithLambda(const DB::ColumnArray & array_col, DB::DataTypePtr array_nested_type, const DB::ColumnFunction & lambda) + { + const auto & offsets = array_col.getOffsets(); + auto rows = array_col.size(); + + size_t nested_size = array_col.getData().size(); + DB::IColumn::Permutation permutation(nested_size); + for (size_t i = 0; i < nested_size; ++i) + permutation[i] = i; + + DB::ColumnArray::Offset current_offset = 0; + for (size_t i = 0; i < rows; ++i) + { + auto next_offset = offsets[i]; + ::sort(&permutation[current_offset], + &permutation[next_offset], + LambdaLess(array_col.getData(), + array_nested_type, + lambda)); + current_offset = next_offset; + } + auto res = DB::ColumnArray::create(array_col.getData().permute(permutation, 0), array_col.getOffsetsPtr()); + return res; + } + + static DB::ColumnPtr executeWithoutLambda(const DB::ColumnArray & array_col) + { + const auto & offsets = array_col.getOffsets(); + auto rows = array_col.size(); + + size_t nested_size = array_col.getData().size(); + DB::IColumn::Permutation permutation(nested_size); + for (size_t i = 0; i < nested_size; ++i) + permutation[i] = i; + + DB::ColumnArray::Offset current_offset = 0; + for (size_t i = 0; i < rows; ++i) + { + auto next_offset = offsets[i]; + ::sort(&permutation[current_offset], + &permutation[next_offset], + Less(array_col.getData())); + current_offset = next_offset; + } + auto res = DB::ColumnArray::create(array_col.getData().permute(permutation, 0), array_col.getOffsetsPtr()); + return res; + } + + String getName() const override + { + return name; + } + +}; REGISTER_FUNCTION(ArraySortSpark) { - factory.registerFunction(); - factory.registerFunction(); + factory.registerFunction(); } - } diff --git a/cpp-ch/local-engine/Functions/SparkFunctionSortArray.cpp b/cpp-ch/local-engine/Functions/SparkFunctionSortArray.cpp new file mode 100644 index 0000000000000..42b88fbce730d --- /dev/null +++ b/cpp-ch/local-engine/Functions/SparkFunctionSortArray.cpp @@ -0,0 +1,88 @@ +/* + * 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 +#include + +namespace DB +{ + +namespace ErrorCodes +{ + extern const int LOGICAL_ERROR; +} + +namespace +{ + +template +struct Less +{ + const IColumn & column; + + explicit Less(const IColumn & column_) : column(column_) { } + + bool operator()(size_t lhs, size_t rhs) const + { + if constexpr (positive) + /* + Note: We use nan_direction_hint=-1 for ascending sort to make NULL the least value. + However, NaN is also considered the least value, + which results in different sorting results compared to Spark since Spark treats NaN as the greatest value. + For now, we are temporarily ignoring this issue because cases with NaN are rare, + and aligning with Spark would require tricky modifications to the CH underlying code. + */ + return column.compareAt(lhs, rhs, column, -1) < 0; + else + return column.compareAt(lhs, rhs, column, -1) > 0; + } +}; + +} + +template +ColumnPtr SparkSortArrayImpl::execute( + const ColumnArray & array, + ColumnPtr mapped, + const ColumnWithTypeAndName * fixed_arguments [[maybe_unused]]) +{ + const ColumnArray::Offsets & offsets = array.getOffsets(); + + size_t size = offsets.size(); + size_t nested_size = array.getData().size(); + IColumn::Permutation permutation(nested_size); + + for (size_t i = 0; i < nested_size; ++i) + permutation[i] = i; + + ColumnArray::Offset current_offset = 0; + for (size_t i = 0; i < size; ++i) + { + auto next_offset = offsets[i]; + ::sort(&permutation[current_offset], &permutation[next_offset], Less(*mapped)); + current_offset = next_offset; + } + + return ColumnArray::create(array.getData().permute(permutation, 0), array.getOffsetsPtr()); +} + +REGISTER_FUNCTION(SortArraySpark) +{ + factory.registerFunction(); + factory.registerFunction(); +} + +} diff --git a/cpp-ch/local-engine/Functions/SparkFunctionArraySort.h b/cpp-ch/local-engine/Functions/SparkFunctionSortArray.h similarity index 86% rename from cpp-ch/local-engine/Functions/SparkFunctionArraySort.h rename to cpp-ch/local-engine/Functions/SparkFunctionSortArray.h index 9ce48f9c0baf5..18c2128c0258e 100644 --- a/cpp-ch/local-engine/Functions/SparkFunctionArraySort.h +++ b/cpp-ch/local-engine/Functions/SparkFunctionSortArray.h @@ -32,7 +32,7 @@ namespace ErrorCodes /** Sort arrays, by values of its elements, or by values of corresponding elements of calculated expression (known as "schwartzsort"). */ template -struct SparkArraySortImpl +struct SparkSortArrayImpl { static bool needBoolean() { return false; } static bool needExpression() { return false; } @@ -67,16 +67,16 @@ struct SparkArraySortImpl const ColumnWithTypeAndName * fixed_arguments [[maybe_unused]] = nullptr); }; -struct NameArraySort +struct NameSortArray { - static constexpr auto name = "arraySortSpark"; + static constexpr auto name = "sortArraySpark"; }; -struct NameArrayReverseSort +struct NameReverseSortArray { - static constexpr auto name = "arrayReverseSortSpark"; + static constexpr auto name = "reverseSortArraySpark"; }; -using SparkFunctionArraySort = FunctionArrayMapped, NameArraySort>; -using SparkFunctionArrayReverseSort = FunctionArrayMapped, NameArrayReverseSort>; +using SparkFunctionSortArray = FunctionArrayMapped, NameSortArray>; +using SparkFunctionReverseSortArray = FunctionArrayMapped, NameReverseSortArray>; } diff --git a/cpp-ch/local-engine/Parser/scalar_function_parser/arrayHighOrderFunctions.cpp b/cpp-ch/local-engine/Parser/scalar_function_parser/arrayHighOrderFunctions.cpp index 584bc0ef1e04f..3811880aea63f 100644 --- a/cpp-ch/local-engine/Parser/scalar_function_parser/arrayHighOrderFunctions.cpp +++ b/cpp-ch/local-engine/Parser/scalar_function_parser/arrayHighOrderFunctions.cpp @@ -151,4 +151,148 @@ class ArrayAggregate : public FunctionParser }; static FunctionParserRegister register_array_aggregate; +class ArraySort : public FunctionParser +{ +public: + static constexpr auto name = "array_sort"; + explicit ArraySort(SerializedPlanParser * plan_parser_) : FunctionParser(plan_parser_) {} + ~ArraySort() override = default; + String getName() const override { return name; } + String getCHFunctionName(const substrait::Expression_ScalarFunction & scalar_function) const override + { + return "arraySortSpark"; + } + const DB::ActionsDAG::Node * parse(const substrait::Expression_ScalarFunction & substrait_func, + DB::ActionsDAGPtr & actions_dag) const + { + auto ch_func_name = getCHFunctionName(substrait_func); + auto parsed_args = parseFunctionArguments(substrait_func, ch_func_name, actions_dag); + + if (parsed_args.size() != 2) + throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "array_sort function must have two arguments"); + if (isDefaultCompare(substrait_func.arguments()[1].value().scalar_function())) + { + return toFunctionNode(actions_dag, ch_func_name, {parsed_args[0]}); + } + + return toFunctionNode(actions_dag, ch_func_name, {parsed_args[1], parsed_args[0]}); + } +private: + + /// The default lambda compare function for array_sort, `array_sort(x)`. + bool isDefaultCompare(const substrait::Expression_ScalarFunction & scalar_function) const + { + String left_variable_name, right_variable_name; + auto names_types = collectLambdaArguments(*plan_parser, scalar_function); + { + auto it = names_types.begin(); + left_variable_name = it->name; + it++; + right_variable_name = it->name; + } + + auto is_function = [&](const substrait::Expression & expr, const String & function_name) { + return expr.has_scalar_function() + && *(plan_parser->getFunctionSignatureName(expr.scalar_function().function_reference())) == function_name; + }; + + auto is_variable = [&](const substrait::Expression & expr, const String & var) { + if (!is_function(expr, "namedlambdavariable")) + { + return false; + } + const auto var_expr = expr.scalar_function().arguments()[0].value(); + if (!var_expr.has_literal()) + return false; + auto [_, name] = plan_parser->parseLiteral(var_expr.literal()); + return var == name.get(); + }; + + auto is_int_value = [&](const substrait::Expression & expr, Int32 val) { + if (!expr.has_literal()) + return false; + auto [_, x] = plan_parser->parseLiteral(expr.literal()); + return val == x.get(); + }; + + auto is_variable_null = [&](const substrait::Expression & expr, const String & var) { + return is_function(expr, "is_null") && is_variable(expr.scalar_function().arguments(0).value(), var); + }; + + auto is_both_null = [&](const substrait::Expression & expr) { + return is_function(expr, "and") + && is_variable_null(expr.scalar_function().arguments(0).value(), left_variable_name) + && is_variable_null(expr.scalar_function().arguments(1).value(), right_variable_name); + }; + + auto is_left_greater_right = [&](const substrait::Expression & expr) { + if (!expr.has_if_then()) + return false; + + const auto & if_ = expr.if_then().ifs(0); + if (!is_function(if_.if_(), "gt")) + return false; + + const auto & less_args = if_.if_().scalar_function().arguments(); + return is_variable(less_args[0].value(), left_variable_name) + && is_variable(less_args[1].value(), right_variable_name) + && is_int_value(if_.then(), 1) + && is_int_value(expr.if_then().else_(), 0); + }; + + auto is_left_less_right = [&](const substrait::Expression & expr) { + if (!expr.has_if_then()) + return false; + + const auto & if_ = expr.if_then().ifs(0); + if (!is_function(if_.if_(), "lt")) + return false; + + const auto & less_args = if_.if_().scalar_function().arguments(); + return is_variable(less_args[0].value(), left_variable_name) + && is_variable(less_args[1].value(), right_variable_name) + && is_int_value(if_.then(), -1) + && is_left_greater_right(expr.if_then().else_()); + }; + + auto is_right_null_else = [&](const substrait::Expression & expr) { + if (!expr.has_if_then()) + return false; + + /// if right arg is null, return 1 + const auto & if_then = expr.if_then(); + return is_variable_null(if_then.ifs(0).if_(), right_variable_name) + && is_int_value(if_then.ifs(0).then(), -1) + && is_left_less_right(if_then.else_()); + + }; + + auto is_left_null_else = [&](const substrait::Expression & expr) { + if (!expr.has_if_then()) + return false; + + /// if left arg is null, return 1 + const auto & if_then = expr.if_then(); + return is_variable_null(if_then.ifs(0).if_(), left_variable_name) + && is_int_value(if_then.ifs(0).then(), 1) + && is_right_null_else(if_then.else_()); + }; + + auto is_if_both_null_else = [&](const substrait::Expression & expr) { + if (!expr.has_if_then()) + { + return false; + } + const auto & if_ = expr.if_then().ifs(0); + return is_both_null(if_.if_()) + && is_int_value(if_.then(), 0) + && is_left_null_else(expr.if_then().else_()); + }; + + const auto & lambda_body = scalar_function.arguments()[0].value(); + return is_if_both_null_else(lambda_body); + } +}; +static FunctionParserRegister register_array_sort; + } diff --git a/cpp-ch/local-engine/Parser/scalar_function_parser/sortArray.cpp b/cpp-ch/local-engine/Parser/scalar_function_parser/sortArray.cpp index 85416bd71864b..4fd2fd4f68004 100644 --- a/cpp-ch/local-engine/Parser/scalar_function_parser/sortArray.cpp +++ b/cpp-ch/local-engine/Parser/scalar_function_parser/sortArray.cpp @@ -52,8 +52,8 @@ class FunctionParserSortArray : public FunctionParser const auto * array_arg = parsed_args[0]; const auto * order_arg = parsed_args[1]; - const auto * sort_node = toFunctionNode(actions_dag, "arraySortSpark", {array_arg}); - const auto * reverse_sort_node = toFunctionNode(actions_dag, "arrayReverseSortSpark", {array_arg}); + const auto * sort_node = toFunctionNode(actions_dag, "sortArraySpark", {array_arg}); + const auto * reverse_sort_node = toFunctionNode(actions_dag, "reverseSortArraySpark", {array_arg}); const auto * result_node = toFunctionNode(actions_dag, "if", {order_arg, sort_node, reverse_sort_node}); return result_node; diff --git a/gluten-core/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala b/gluten-core/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala index ff7449e2d3404..a69d41d00c12d 100644 --- a/gluten-core/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala +++ b/gluten-core/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala @@ -258,6 +258,15 @@ trait SparkPlanExecApi { throw new GlutenNotSupportException("all_match is not supported") } + /** Transform array array_sort to Substrait. */ + def genArraySortTransformer( + substraitExprName: String, + argument: ExpressionTransformer, + function: ExpressionTransformer, + expr: ArraySort): ExpressionTransformer = { + throw new GlutenNotSupportException("array_sort(on array) is not supported") + } + /** Transform array exists to Substrait */ def genArrayExistsTransformer( substraitExprName: String, diff --git a/gluten-core/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala b/gluten-core/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala index b5bcb6876e4d9..805ff94900fe9 100644 --- a/gluten-core/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala +++ b/gluten-core/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala @@ -556,6 +556,19 @@ object ExpressionConverter extends SQLConfHelper with Logging { expressionsMap), arrayTransform ) + case arraySort: ArraySort => + BackendsApiManager.getSparkPlanExecApiInstance.genArraySortTransformer( + substraitExprName, + replaceWithExpressionTransformerInternal( + arraySort.argument, + attributeSeq, + expressionsMap), + replaceWithExpressionTransformerInternal( + arraySort.function, + attributeSeq, + expressionsMap), + arraySort + ) case tryEval @ TryEval(a: Add) => BackendsApiManager.getSparkPlanExecApiInstance.genTryArithmeticTransformer( substraitExprName, diff --git a/gluten-core/src/main/scala/org/apache/gluten/expression/ExpressionMappings.scala b/gluten-core/src/main/scala/org/apache/gluten/expression/ExpressionMappings.scala index e7e9c7ffe9004..51e78a97e9979 100644 --- a/gluten-core/src/main/scala/org/apache/gluten/expression/ExpressionMappings.scala +++ b/gluten-core/src/main/scala/org/apache/gluten/expression/ExpressionMappings.scala @@ -248,6 +248,7 @@ object ExpressionMappings { Sig[ArrayFilter](FILTER), Sig[ArrayForAll](FORALL), Sig[ArrayExists](EXISTS), + Sig[ArraySort](ARRAY_SORT), Sig[Shuffle](SHUFFLE), Sig[ZipWith](ZIP_WITH), Sig[Flatten](FLATTEN), diff --git a/shims/common/src/main/scala/org/apache/gluten/expression/ExpressionNames.scala b/shims/common/src/main/scala/org/apache/gluten/expression/ExpressionNames.scala index 278f119226457..e3dc3a8ab0a91 100644 --- a/shims/common/src/main/scala/org/apache/gluten/expression/ExpressionNames.scala +++ b/shims/common/src/main/scala/org/apache/gluten/expression/ExpressionNames.scala @@ -261,6 +261,7 @@ object ExpressionNames { final val ARRAY_EXCEPT = "array_except" final val ARRAY_REPEAT = "array_repeat" final val ARRAY_REMOVE = "array_remove" + final val ARRAY_SORT = "array_sort" final val ARRAYS_ZIP = "arrays_zip" final val FILTER = "filter" final val FORALL = "forall" From 6c181961e2d35fc21e608f9fa4822346b07bb1df Mon Sep 17 00:00:00 2001 From: Rong Ma Date: Fri, 5 Jul 2024 09:49:43 +0800 Subject: [PATCH 11/13] [CI] Hotfix centos7 CI failure (#6340) --- .github/workflows/velox_docker.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/velox_docker.yml b/.github/workflows/velox_docker.yml index c4b24598ceb9a..116503ad2cb0f 100644 --- a/.github/workflows/velox_docker.yml +++ b/.github/workflows/velox_docker.yml @@ -42,6 +42,7 @@ on: - 'dev/**' env: + ACTIONS_ALLOW_USE_UNSECURE_NODE_VERSION: true MVN_CMD: 'mvn -ntp' concurrency: From ef424aa3c0917fefc3c80d17ca40109cd0e14b88 Mon Sep 17 00:00:00 2001 From: Xiduo You Date: Fri, 5 Jul 2024 13:30:43 +0800 Subject: [PATCH 12/13] [VL] Deduplicate sorting keys (#6332) --- .../gluten/execution/TestOperator.scala | 26 ++++++++++++------- cpp/velox/substrait/SubstraitToVeloxPlan.cc | 16 +++++------- cpp/velox/substrait/SubstraitToVeloxPlan.h | 1 + .../GlutenSQLWindowFunctionSuite.scala | 2 +- .../GlutenSQLWindowFunctionSuite.scala | 2 +- .../utils/velox/VeloxTestSettings.scala | 1 - .../GlutenSQLWindowFunctionSuite.scala | 2 +- .../utils/velox/VeloxTestSettings.scala | 1 - .../GlutenSQLWindowFunctionSuite.scala | 6 ++--- 9 files changed, 30 insertions(+), 27 deletions(-) diff --git a/backends-velox/src/test/scala/org/apache/gluten/execution/TestOperator.scala b/backends-velox/src/test/scala/org/apache/gluten/execution/TestOperator.scala index c6e72197d6c7d..230fc565d9eb3 100644 --- a/backends-velox/src/test/scala/org/apache/gluten/execution/TestOperator.scala +++ b/backends-velox/src/test/scala/org/apache/gluten/execution/TestOperator.scala @@ -422,18 +422,18 @@ class TestOperator extends VeloxWholeStageTransformerSuite with AdaptiveSparkPla } // Test same partition/ordering keys. -// runQueryAndCompare( -// "select avg(l_partkey) over" + -// " (partition by l_suppkey order by l_suppkey) from lineitem ") { -// checkGlutenOperatorMatch[WindowExecTransformer] -// } + runQueryAndCompare( + "select avg(l_partkey) over" + + " (partition by l_suppkey order by l_suppkey) from lineitem ") { + checkGlutenOperatorMatch[WindowExecTransformer] + } // Test overlapping partition/ordering keys. -// runQueryAndCompare( -// "select avg(l_partkey) over" + -// " (partition by l_suppkey order by l_suppkey, l_orderkey) from lineitem ") { -// checkGlutenOperatorMatch[WindowExecTransformer] -// } + runQueryAndCompare( + "select avg(l_partkey) over" + + " (partition by l_suppkey order by l_suppkey, l_orderkey) from lineitem ") { + checkGlutenOperatorMatch[WindowExecTransformer] + } } } } @@ -1911,4 +1911,10 @@ class TestOperator extends VeloxWholeStageTransformerSuite with AdaptiveSparkPla val resultLength = df.collect().length assert(resultLength > 25000 && resultLength < 35000) } + + test("Deduplicate sorting keys") { + runQueryAndCompare("select * from lineitem order by l_orderkey, l_orderkey") { + checkGlutenOperatorMatch[SortExecTransformer] + } + } } diff --git a/cpp/velox/substrait/SubstraitToVeloxPlan.cc b/cpp/velox/substrait/SubstraitToVeloxPlan.cc index 73047b2f49073..34710c35a40db 100644 --- a/cpp/velox/substrait/SubstraitToVeloxPlan.cc +++ b/cpp/velox/substrait/SubstraitToVeloxPlan.cc @@ -1046,17 +1046,15 @@ SubstraitToVeloxPlanConverter::processSortField( const RowTypePtr& inputType) { std::vector sortingKeys; std::vector sortingOrders; - sortingKeys.reserve(sortFields.size()); - sortingOrders.reserve(sortFields.size()); - + std::unordered_set uniqueKeys; for (const auto& sort : sortFields) { - sortingOrders.emplace_back(toSortOrder(sort)); - - if (sort.has_expr()) { - auto expression = exprConverter_->toVeloxExpr(sort.expr(), inputType); - auto fieldExpr = std::dynamic_pointer_cast(expression); - VELOX_USER_CHECK_NOT_NULL(fieldExpr, "Sort Operator only supports field sorting key"); + GLUTEN_CHECK(sort.has_expr(), "Sort field must have expr"); + auto expression = exprConverter_->toVeloxExpr(sort.expr(), inputType); + auto fieldExpr = std::dynamic_pointer_cast(expression); + VELOX_USER_CHECK_NOT_NULL(fieldExpr, "Sort Operator only supports field sorting key"); + if (uniqueKeys.insert(fieldExpr->name()).second) { sortingKeys.emplace_back(fieldExpr); + sortingOrders.emplace_back(toSortOrder(sort)); } } return {sortingKeys, sortingOrders}; diff --git a/cpp/velox/substrait/SubstraitToVeloxPlan.h b/cpp/velox/substrait/SubstraitToVeloxPlan.h index 1535b1f85f51b..1f2f39f51ae85 100644 --- a/cpp/velox/substrait/SubstraitToVeloxPlan.h +++ b/cpp/velox/substrait/SubstraitToVeloxPlan.h @@ -198,6 +198,7 @@ class SubstraitToVeloxPlanConverter { /// Helper Function to convert Substrait sortField to Velox sortingKeys and /// sortingOrders. + /// Note that, this method would deduplicate the sorting keys which have the same field name. std::pair, std::vector> processSortField( const ::google::protobuf::RepeatedPtrField<::substrait::SortField>& sortField, const RowTypePtr& inputType); diff --git a/gluten-ut/spark32/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala b/gluten-ut/spark32/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala index e26c1dc41a9a2..6665174207b09 100644 --- a/gluten-ut/spark32/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala +++ b/gluten-ut/spark32/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala @@ -47,7 +47,7 @@ class GlutenSQLWindowFunctionSuite extends SQLWindowFunctionSuite with GlutenSQL Row(95337, 12, decimal(915.61)) ) - ignoreGluten("Literal in window partition by and sort") { + testGluten("Literal in window partition by and sort") { withTable("customer") { val rdd = spark.sparkContext.parallelize(customerData) val customerDF = spark.createDataFrame(rdd, customerSchema) diff --git a/gluten-ut/spark33/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala b/gluten-ut/spark33/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala index e26c1dc41a9a2..6665174207b09 100644 --- a/gluten-ut/spark33/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala +++ b/gluten-ut/spark33/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala @@ -47,7 +47,7 @@ class GlutenSQLWindowFunctionSuite extends SQLWindowFunctionSuite with GlutenSQL Row(95337, 12, decimal(915.61)) ) - ignoreGluten("Literal in window partition by and sort") { + testGluten("Literal in window partition by and sort") { withTable("customer") { val rdd = spark.sparkContext.parallelize(customerData) val customerDF = spark.createDataFrame(rdd, customerSchema) diff --git a/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala b/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala index 941f7b89b2d5c..0da19922ffdaf 100644 --- a/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala +++ b/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala @@ -1182,7 +1182,6 @@ class VeloxTestSettings extends BackendTestSettings { enableSuite[GlutenDataFrameToSchemaSuite] enableSuite[GlutenDatasetUnpivotSuite] enableSuite[GlutenLateralColumnAliasSuite] - .exclude("Aggregate expressions containing no aggregate or grouping expressions still resolves") enableSuite[GlutenParametersSuite] enableSuite[GlutenResolveDefaultColumnsSuite] enableSuite[GlutenSubqueryHintPropagationSuite] diff --git a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala b/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala index e26c1dc41a9a2..6665174207b09 100644 --- a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala +++ b/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala @@ -47,7 +47,7 @@ class GlutenSQLWindowFunctionSuite extends SQLWindowFunctionSuite with GlutenSQL Row(95337, 12, decimal(915.61)) ) - ignoreGluten("Literal in window partition by and sort") { + testGluten("Literal in window partition by and sort") { withTable("customer") { val rdd = spark.sparkContext.parallelize(customerData) val customerDF = spark.createDataFrame(rdd, customerSchema) diff --git a/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala b/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala index f3bd7ff6c752d..e54aca34ec757 100644 --- a/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala +++ b/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala @@ -1197,7 +1197,6 @@ class VeloxTestSettings extends BackendTestSettings { enableSuite[GlutenDataFrameToSchemaSuite] enableSuite[GlutenDatasetUnpivotSuite] enableSuite[GlutenLateralColumnAliasSuite] - .exclude("Aggregate expressions containing no aggregate or grouping expressions still resolves") enableSuite[GlutenParametersSuite] enableSuite[GlutenResolveDefaultColumnsSuite] enableSuite[GlutenSubqueryHintPropagationSuite] diff --git a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala index e61c084ab7db4..89a4351744ef4 100644 --- a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala +++ b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/GlutenSQLWindowFunctionSuite.scala @@ -47,7 +47,7 @@ class GlutenSQLWindowFunctionSuite extends SQLWindowFunctionSuite with GlutenSQL Row(95337, 12, decimal(915.61)) ) - ignoreGluten("Literal in window partition by and sort") { + testGluten("Literal in window partition by and sort") { withTable("customer") { val rdd = spark.sparkContext.parallelize(customerData) val customerDF = spark.createDataFrame(rdd, customerSchema) @@ -93,7 +93,7 @@ class GlutenSQLWindowFunctionSuite extends SQLWindowFunctionSuite with GlutenSQL } } - ignoreGluten("Filter on row number") { + testGluten("Filter on row number") { withTable("customer") { val rdd = spark.sparkContext.parallelize(customerData) val customerDF = spark.createDataFrame(rdd, customerSchema) @@ -137,7 +137,7 @@ class GlutenSQLWindowFunctionSuite extends SQLWindowFunctionSuite with GlutenSQL } } - ignoreGluten("Filter on rank") { + testGluten("Filter on rank") { withTable("customer") { val rdd = spark.sparkContext.parallelize(customerData) val customerDF = spark.createDataFrame(rdd, customerSchema) From f8e6b75120fed1fd89d21bbeede7131c1ad6bc65 Mon Sep 17 00:00:00 2001 From: Gluten Performance Bot <137994563+GlutenPerfBot@users.noreply.github.com> Date: Fri, 5 Jul 2024 13:38:24 +0800 Subject: [PATCH 13/13] [VL] Daily Update Velox Version (2024_07_05) (#6339) 92d7ca3c1 by Jimmy Lu, Fix DirectBufferedInput::read not recording runtime stats (10393) ac92be3ec by Jimmy Lu, Use retainedSize instead of estimateFlatSize in LocalExchange (10390) --- ep/build-velox/src/get_velox.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ep/build-velox/src/get_velox.sh b/ep/build-velox/src/get_velox.sh index 07694a36102d2..0ef03bd83e58c 100755 --- a/ep/build-velox/src/get_velox.sh +++ b/ep/build-velox/src/get_velox.sh @@ -17,7 +17,7 @@ set -exu VELOX_REPO=https://github.com/oap-project/velox.git -VELOX_BRANCH=2024_07_04 +VELOX_BRANCH=2024_07_05 VELOX_HOME="" #Set on run gluten on HDFS