You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by ch...@apache.org on 2024/10/15 03:19:10 UTC
(incubator-gluten) branch main updated: [GLUTEN-7311][CH] Support grace aggregate algorithm in partial aggregating stages (#7322)
This is an automated email from the ASF dual-hosted git repository.
changchen pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 74c66410db [GLUTEN-7311][CH] Support grace aggregate algorithm in partial aggregating stages (#7322)
74c66410db is described below
commit 74c66410dbe0a3ea0be445fb9d7b41ac0a765773
Author: lgbo <lg...@gmail.com>
AuthorDate: Tue Oct 15 11:19:05 2024 +0800
[GLUTEN-7311][CH] Support grace aggregate algorithm in partial aggregating stages (#7322)
* support grace aggregating algorithm in partial aggregating stages
* stage
* update
* add debug logs
---
.../execution/CHHashAggregateExecTransformer.scala | 67 ++-
.../MergeTwoPhasesHashBaseAggregate.scala | 6 +-
.../metrics/HashAggregateMetricsUpdater.scala | 2 +
.../GlutenClickhouseCountDistinctSuite.scala | 6 +
cpp-ch/clickhouse.version | 2 +-
.../AggregateFunctionPartialMerge.h | 9 +-
cpp-ch/local-engine/Common/DebugUtils.cpp | 35 ++
cpp-ch/local-engine/Common/DebugUtils.h | 1 +
.../local-engine/Operator/GraceAggregatingStep.cpp | 105 ++++
.../local-engine/Operator/GraceAggregatingStep.h | 54 +++
...gatedStep.cpp => GraceAggregatingTransform.cpp} | 152 ++----
...ggregatedStep.h => GraceAggregatingTransform.h} | 56 +--
.../Operator/GraceMergingAggregatedStep.cpp | 528 +--------------------
.../Operator/GraceMergingAggregatedStep.h | 89 +---
.../Parser/AdvancedParametersParseUtil.cpp | 10 +
.../Parser/AdvancedParametersParseUtil.h | 7 +
.../Parser/RelParsers/AggregateRelParser.cpp | 113 ++++-
.../Parser/RelParsers/AggregateRelParser.h | 1 +
.../local-engine/Parser/RelParsers/RelParser.cpp | 36 --
cpp-ch/local-engine/Parser/RelParsers/RelParser.h | 5 -
.../local-engine/Parser/SerializedPlanParser.cpp | 16 +-
21 files changed, 472 insertions(+), 828 deletions(-)
diff --git a/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHHashAggregateExecTransformer.scala b/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHHashAggregateExecTransformer.scala
index d641c05cd6..b45f4cb97c 100644
--- a/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHHashAggregateExecTransformer.scala
+++ b/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHHashAggregateExecTransformer.scala
@@ -30,7 +30,9 @@ import org.apache.gluten.substrait.rel.{RelBuilder, RelNode}
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.aggregate._
import org.apache.spark.sql.execution._
+import org.apache.spark.sql.execution.adaptive._
import org.apache.spark.sql.execution.aggregate.BaseAggregateExec
+import org.apache.spark.sql.execution.exchange._
import org.apache.spark.sql.types._
import com.google.protobuf.{Any, StringValue}
@@ -53,6 +55,19 @@ object CHHashAggregateExecTransformer {
def newStructFieldId(): Long = curId.getAndIncrement()
}
+/**
+ * About aggregate modes. In general, all the modes of aggregate expressions in the same
+ * HashAggregateExec are the same. And the aggregation will be divided into two stages, partial
+ * aggregate and final merge aggregated. But there are some exceptions.
+ * - f(distinct x). This will be divided into four stages (without stages merged by
+ * `MergeTwoPhasesHashBaseAggregate`). The first two stages use `x` as a grouping key and
+ * without aggregate functions. The last two stages aggregate without `x` as a grouping key and
+ * with aggregate function `f`.
+ * - f1(distinct x), f(2). This will be divided into four stages. The first two stages use `x` as
+ * a grouping key and with partial aggregate function `f2`. The last two stages aggregate
+ * without `x` as a grouping key and with aggregate function `f1` and `f2`. The 3rd stages hase
+ * different modes at the same time. `f2` is partial merge but `f1` is partial.
+ */
case class CHHashAggregateExecTransformer(
requiredChildDistributionExpressions: Option[Seq[Expression]],
groupingExpressions: Seq[NamedExpression],
@@ -402,13 +417,59 @@ case class CHHashAggregateExecTransformer(
} else {
null
}
- val optimizationContent = s"has_required_child_distribution_expressions=" +
- s"${requiredChildDistributionExpressions.isDefined}\n"
+ val parametersStrBuf = new StringBuffer("AggregateParams:")
+ parametersStrBuf
+ .append(s"hasPrePartialAggregate=$hasPrePartialAggregate")
+ .append("\n")
+ .append(s"hasRequiredChildDistributionExpressions=" +
+ s"${requiredChildDistributionExpressions.isDefined}")
+ .append("\n")
val optimization =
BackendsApiManager.getTransformerApiInstance.packPBMessage(
- StringValue.newBuilder.setValue(optimizationContent).build)
+ StringValue.newBuilder.setValue(parametersStrBuf.toString).build)
ExtensionBuilder.makeAdvancedExtension(optimization, enhancement)
+ }
+
+ // Check that there is a partial aggregation ahead this HashAggregateExec.
+ // This is useful when aggregate expressions are empty.
+ private def hasPrePartialAggregate(): Boolean = {
+ def isSameAggregation(agg1: BaseAggregateExec, agg2: BaseAggregateExec): Boolean = {
+ val res = agg1.groupingExpressions.length == agg2.groupingExpressions.length &&
+ agg1.groupingExpressions.zip(agg2.groupingExpressions).forall {
+ case (e1, e2) =>
+ e1.toAttribute == e2.toAttribute
+ }
+ res
+ }
+ def checkChild(exec: SparkPlan): Boolean = {
+ exec match {
+ case agg: BaseAggregateExec =>
+ isSameAggregation(this, agg)
+ case shuffle: ShuffleExchangeLike =>
+ checkChild(shuffle.child)
+ case iter: InputIteratorTransformer =>
+ checkChild(iter.child)
+ case inputAdapter: ColumnarInputAdapter =>
+ checkChild(inputAdapter.child)
+ case wholeStage: WholeStageTransformer =>
+ checkChild(wholeStage.child)
+ case aqeShuffleRead: AQEShuffleReadExec =>
+ checkChild(aqeShuffleRead.child)
+ case shuffle: ShuffleQueryStageExec =>
+ checkChild(shuffle.plan)
+ case _ =>
+ false
+ }
+ }
+
+ // It's more complex when the aggregate expressions are empty. We need to iterate the plan
+ // to find whether there is a partial aggregation. `count(distinct x)` is one of these cases.
+ if (aggregateExpressions.length > 0) {
+ modes.exists(mode => mode == PartialMerge || mode == Final)
+ } else {
+ checkChild(child)
+ }
}
}
diff --git a/backends-clickhouse/src/main/scala/org/apache/gluten/extension/MergeTwoPhasesHashBaseAggregate.scala b/backends-clickhouse/src/main/scala/org/apache/gluten/extension/MergeTwoPhasesHashBaseAggregate.scala
index 43adf27b4e..63c5fe017f 100644
--- a/backends-clickhouse/src/main/scala/org/apache/gluten/extension/MergeTwoPhasesHashBaseAggregate.scala
+++ b/backends-clickhouse/src/main/scala/org/apache/gluten/extension/MergeTwoPhasesHashBaseAggregate.scala
@@ -18,6 +18,7 @@ package org.apache.gluten.extension
import org.apache.gluten.GlutenConfig
+import org.apache.spark.internal.Logging
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.expressions.aggregate.{Complete, Final, Partial}
import org.apache.spark.sql.catalyst.rules.Rule
@@ -33,7 +34,9 @@ import org.apache.spark.sql.execution.aggregate.{BaseAggregateExec, HashAggregat
* Note: this rule must be applied before the `PullOutPreProject` rule, because the
* `PullOutPreProject` rule will modify the attributes in some cases.
*/
-case class MergeTwoPhasesHashBaseAggregate(session: SparkSession) extends Rule[SparkPlan] {
+case class MergeTwoPhasesHashBaseAggregate(session: SparkSession)
+ extends Rule[SparkPlan]
+ with Logging {
val columnarConf: GlutenConfig = GlutenConfig.getConf
val scanOnly: Boolean = columnarConf.enableScanOnly
@@ -73,7 +76,6 @@ case class MergeTwoPhasesHashBaseAggregate(session: SparkSession) extends Rule[S
// convert to complete mode aggregate expressions
val completeAggregateExpressions = aggregateExpressions.map(_.copy(mode = Complete))
hashAgg.copy(
- requiredChildDistributionExpressions = None,
groupingExpressions = child.groupingExpressions,
aggregateExpressions = completeAggregateExpressions,
initialInputBufferOffset = 0,
diff --git a/backends-clickhouse/src/main/scala/org/apache/gluten/metrics/HashAggregateMetricsUpdater.scala b/backends-clickhouse/src/main/scala/org/apache/gluten/metrics/HashAggregateMetricsUpdater.scala
index b035d7a04f..2b34f8a16b 100644
--- a/backends-clickhouse/src/main/scala/org/apache/gluten/metrics/HashAggregateMetricsUpdater.scala
+++ b/backends-clickhouse/src/main/scala/org/apache/gluten/metrics/HashAggregateMetricsUpdater.scala
@@ -77,10 +77,12 @@ object HashAggregateMetricsUpdater {
"AggregatingTransform",
"StreamingAggregatingTransform",
"MergingAggregatedTransform",
+ "GraceAggregatingTransform",
"GraceMergingAggregatedTransform")
val CH_PLAN_NODE_NAME = Array(
"AggregatingTransform",
"StreamingAggregatingTransform",
"MergingAggregatedTransform",
+ "GraceAggregatingTransform",
"GraceMergingAggregatedTransform")
}
diff --git a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickhouseCountDistinctSuite.scala b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickhouseCountDistinctSuite.scala
index 3836817330..694b578c91 100644
--- a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickhouseCountDistinctSuite.scala
+++ b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickhouseCountDistinctSuite.scala
@@ -69,6 +69,12 @@ class GlutenClickhouseCountDistinctSuite extends GlutenClickHouseWholeStageTrans
}
}
+ test("test distinct with not-distinct") {
+ val sql = "select a, count(distinct(b)), sum(c) from " +
+ "values (0, null,1), (0,null,1), (1, 1,1), (2, 2, 1) ,(2,2,2) as data(a,b,c) group by a"
+ compareResultsAgainstVanillaSpark(sql, true, { _ => })
+ }
+
test("check all data types") {
spark.createDataFrame(genTestData()).createOrReplaceTempView("all_data_types")
diff --git a/cpp-ch/clickhouse.version b/cpp-ch/clickhouse.version
index 6057691214..0b8e1d7abc 100644
--- a/cpp-ch/clickhouse.version
+++ b/cpp-ch/clickhouse.version
@@ -1,3 +1,3 @@
CH_ORG=Kyligence
CH_BRANCH=rebase_ch/20241010
-CH_COMMIT=3e57eb5c17e
+CH_COMMIT=30b3ff313a9
diff --git a/cpp-ch/local-engine/AggregateFunctions/AggregateFunctionPartialMerge.h b/cpp-ch/local-engine/AggregateFunctions/AggregateFunctionPartialMerge.h
index 1f5ff9fca1..bc1f40f60e 100644
--- a/cpp-ch/local-engine/AggregateFunctions/AggregateFunctionPartialMerge.h
+++ b/cpp-ch/local-engine/AggregateFunctions/AggregateFunctionPartialMerge.h
@@ -27,7 +27,7 @@ namespace DB
{
namespace ErrorCodes
{
- extern const int ILLEGAL_TYPE_OF_ARGUMENT;
+extern const int ILLEGAL_TYPE_OF_ARGUMENT;
}
}
@@ -50,8 +50,7 @@ private:
public:
AggregateFunctionPartialMerge(const AggregateFunctionPtr & nested_, const DataTypePtr & argument, const Array & params_)
- : IAggregateFunctionHelper<AggregateFunctionPartialMerge>({argument}, params_, createResultType(nested_))
- , nested_func(nested_)
+ : IAggregateFunctionHelper<AggregateFunctionPartialMerge>({argument}, params_, createResultType(nested_)), nested_func(nested_)
{
const DataTypeAggregateFunction * data_type = typeid_cast<const DataTypeAggregateFunction *>(argument.get());
@@ -115,5 +114,9 @@ public:
bool allocatesMemoryInArena() const override { return nested_func->allocatesMemoryInArena(); }
AggregateFunctionPtr getNestedFunction() const override { return nested_func; }
+ /// If the aggregate phase is `INTEMEDIATE_TO_INTERMEDIATE`, partial merge combinator is applied. In this case, the actual result column's
+ /// representation is `xxxPartialMerge`. It will make block structure check fail somewhere, since the expected column's represiontation is
+ /// `xxx` without partial merge. The represiontaions of `xxxPartialMerge` and `xxx` are the same actually.
+ const IAggregateFunction & getBaseAggregateFunctionWithSameStateRepresentation() const override { return *nested_func; }
};
}
diff --git a/cpp-ch/local-engine/Common/DebugUtils.cpp b/cpp-ch/local-engine/Common/DebugUtils.cpp
index 27f9efdacf..c144a1384e 100644
--- a/cpp-ch/local-engine/Common/DebugUtils.cpp
+++ b/cpp-ch/local-engine/Common/DebugUtils.cpp
@@ -16,6 +16,7 @@
*/
#include "DebugUtils.h"
#include <iostream>
+#include <sstream>
#include <Columns/ColumnNullable.h>
#include <Columns/ColumnString.h>
#include <DataTypes/DataTypeDate.h>
@@ -58,6 +59,40 @@ void headBlock(const DB::Block & block, size_t count)
}
}
+String printBlock(const DB::Block & block, size_t count)
+{
+ std::ostringstream ss;
+ ss << std::string("============Block============\n");
+ ss << block.dumpStructure() << String("\n");
+ // print header
+ for (const auto & name : block.getNames())
+ ss << name << std::string("\t");
+ ss << std::string("\n");
+
+ // print rows
+ for (size_t row = 0; row < std::min(count, block.rows()); ++row)
+ {
+ for (size_t column = 0; column < block.columns(); ++column)
+ {
+ const auto type = block.getByPosition(column).type;
+ auto col = block.getByPosition(column).column;
+
+ if (column > 0)
+ ss << std::string("\t");
+ DB::WhichDataType which(type);
+ if (which.isAggregateFunction())
+ ss << std::string("Nan");
+ else if (col->isNullAt(row))
+ ss << std::string("null");
+ else
+ ss << toString((*col)[row]);
+ }
+ ss << std::string("\n");
+ }
+ return ss.str();
+}
+
+
void headColumn(const DB::ColumnPtr & column, size_t count)
{
std::cout << "============Column============" << std::endl;
diff --git a/cpp-ch/local-engine/Common/DebugUtils.h b/cpp-ch/local-engine/Common/DebugUtils.h
index 24fcd79141..cc0ecdc59d 100644
--- a/cpp-ch/local-engine/Common/DebugUtils.h
+++ b/cpp-ch/local-engine/Common/DebugUtils.h
@@ -21,6 +21,7 @@
namespace debug
{
void headBlock(const DB::Block & block, size_t count = 10);
+String printBlock(const DB::Block & block, size_t count = 10);
void headColumn(const DB::ColumnPtr & column, size_t count = 10);
}
diff --git a/cpp-ch/local-engine/Operator/GraceAggregatingStep.cpp b/cpp-ch/local-engine/Operator/GraceAggregatingStep.cpp
new file mode 100644
index 0000000000..63175657f4
--- /dev/null
+++ b/cpp-ch/local-engine/Operator/GraceAggregatingStep.cpp
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "GraceAggregatingStep.h"
+#include <Interpreters/JoinUtils.h>
+#include <Processors/Transforms/AggregatingTransform.h>
+#include <QueryPipeline/QueryPipelineBuilder.h>
+#include <Common/BitHelpers.h>
+#include <Common/CHUtil.h>
+#include <Common/CurrentThread.h>
+#include <Common/GlutenConfig.h>
+#include <Common/QueryContext.h>
+#include <Common/formatReadable.h>
+#include "GraceAggregatingTransform.h"
+
+namespace DB::ErrorCodes
+{
+extern const int LOGICAL_ERROR;
+}
+
+namespace local_engine
+{
+static DB::ITransformingStep::Traits getTraits()
+{
+ return DB::ITransformingStep::Traits{
+ {
+ .preserves_number_of_streams = false,
+ .preserves_sorting = false,
+ },
+ {
+ .preserves_number_of_rows = false,
+ }};
+}
+
+static DB::Block buildOutputHeader(const DB::Block & input_header_, const DB::Aggregator::Params params_, bool final)
+{
+ return params_.getHeader(input_header_, final);
+}
+
+GraceAggregatingStep::GraceAggregatingStep(
+ DB::ContextPtr context_, const DB::DataStream & input_stream_, DB::Aggregator::Params params_, bool no_pre_aggregated_)
+ : DB::ITransformingStep(input_stream_, buildOutputHeader(input_stream_.header, params_, false), getTraits())
+ , context(context_)
+ , params(std::move(params_))
+ , no_pre_aggregated(no_pre_aggregated_)
+{
+}
+
+void GraceAggregatingStep::transformPipeline(DB::QueryPipelineBuilder & pipeline, const DB::BuildQueryPipelineSettings &)
+{
+ if (params.max_bytes_before_external_group_by)
+ {
+ throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "max_bytes_before_external_group_by is not supported in GraceAggregatingStep");
+ }
+ auto num_streams = pipeline.getNumStreams();
+ auto transform_params = std::make_shared<DB::AggregatingTransformParams>(pipeline.getHeader(), params, false);
+ pipeline.resize(1);
+ auto build_transform = [&](DB::OutputPortRawPtrs outputs)
+ {
+ DB::Processors new_processors;
+ for (auto & output : outputs)
+ {
+ auto op
+ = std::make_shared<GraceAggregatingTransform>(pipeline.getHeader(), transform_params, context, no_pre_aggregated, false);
+ new_processors.push_back(op);
+ DB::connect(*output, op->getInputs().front());
+ }
+ return new_processors;
+ };
+ pipeline.transform(build_transform);
+ pipeline.resize(num_streams, true);
+}
+
+void GraceAggregatingStep::describeActions(DB::IQueryPlanStep::FormatSettings & settings) const
+{
+ return params.explain(settings.out, settings.offset);
+}
+
+void GraceAggregatingStep::describeActions(DB::JSONBuilder::JSONMap & map) const
+{
+ params.explain(map);
+}
+
+void GraceAggregatingStep::updateOutputStream()
+{
+ output_stream
+ = createOutputStream(input_streams.front(), buildOutputHeader(input_streams.front().header, params, false), getDataStreamTraits());
+}
+
+
+}
diff --git a/cpp-ch/local-engine/Operator/GraceAggregatingStep.h b/cpp-ch/local-engine/Operator/GraceAggregatingStep.h
new file mode 100644
index 0000000000..7bb3917c9f
--- /dev/null
+++ b/cpp-ch/local-engine/Operator/GraceAggregatingStep.h
@@ -0,0 +1,54 @@
+/*
+ * 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 <Interpreters/Aggregator.h>
+#include <Interpreters/Context.h>
+#include <Processors/IProcessor.h>
+#include <Processors/Port.h>
+#include <Processors/QueryPlan/IQueryPlanStep.h>
+#include <Processors/QueryPlan/ITransformingStep.h>
+#include <Processors/Transforms/AggregatingTransform.h>
+
+
+namespace local_engine
+{
+/**
+ * GraceAggregatingStep is used for partial aggregating stages.
+ */
+class GraceAggregatingStep : public DB::ITransformingStep
+{
+public:
+ explicit GraceAggregatingStep(
+ DB::ContextPtr context_,
+ const DB::DataStream & input_stream_,
+ DB::Aggregator::Params params_,
+ bool no_pre_aggregated_);
+ ~GraceAggregatingStep() override = default;
+
+ String getName() const override { return "GraceAggregatingStep"; }
+
+ void transformPipeline(DB::QueryPipelineBuilder & pipeline, const DB::BuildQueryPipelineSettings &) override;
+
+ void describeActions(DB::JSONBuilder::JSONMap & map) const override;
+ void describeActions(DB::IQueryPlanStep::FormatSettings & settings) const override;
+private:
+ DB::ContextPtr context;
+ DB::Aggregator::Params params;
+ bool no_pre_aggregated;
+ void updateOutputStream() override;
+};
+}
\ No newline at end of file
diff --git a/cpp-ch/local-engine/Operator/GraceMergingAggregatedStep.cpp b/cpp-ch/local-engine/Operator/GraceAggregatingTransform.cpp
similarity index 79%
copy from cpp-ch/local-engine/Operator/GraceMergingAggregatedStep.cpp
copy to cpp-ch/local-engine/Operator/GraceAggregatingTransform.cpp
index 82b498e58f..c8797d48d9 100644
--- a/cpp-ch/local-engine/Operator/GraceMergingAggregatedStep.cpp
+++ b/cpp-ch/local-engine/Operator/GraceAggregatingTransform.cpp
@@ -14,99 +14,30 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include "GraceMergingAggregatedStep.h"
-#include <Interpreters/JoinUtils.h>
-#include <Processors/Transforms/AggregatingTransform.h>
-#include <QueryPipeline/QueryPipelineBuilder.h>
+
+#include "GraceAggregatingTransform.h"
+#include <Common/BitHelpers.h>
#include <Common/CHUtil.h>
#include <Common/CurrentThread.h>
-#include <Common/formatReadable.h>
-#include <Common/BitHelpers.h>
#include <Common/GlutenConfig.h>
#include <Common/QueryContext.h>
+#include <Common/formatReadable.h>
-namespace DB
-{
-namespace ErrorCodes
-{
- extern const int LOGICAL_ERROR;
-}
-}
+#include <Common/DebugUtils.h>
-namespace local_engine
-{
-static DB::ITransformingStep::Traits getTraits()
+namespace DB::ErrorCodes
{
- return DB::ITransformingStep::Traits
- {
- {
- .preserves_number_of_streams = false,
- .preserves_sorting = false,
- },
- {
- .preserves_number_of_rows = false,
- }
- };
+extern const int LOGICAL_ERROR;
}
-static DB::Block buildOutputHeader(const DB::Block & input_header_, const DB::Aggregator::Params params_, bool final)
+namespace local_engine
{
- return params_.getHeader(input_header_, final);
-}
-
-GraceMergingAggregatedStep::GraceMergingAggregatedStep(
+GraceAggregatingTransform::GraceAggregatingTransform(
+ const DB::Block & header_,
+ DB::AggregatingTransformParamsPtr params_,
DB::ContextPtr context_,
- const DB::DataStream & input_stream_,
- DB::Aggregator::Params params_,
- bool no_pre_aggregated_)
- : DB::ITransformingStep(
- input_stream_, buildOutputHeader(input_stream_.header, params_, true), getTraits())
- , context(context_)
- , params(std::move(params_))
- , no_pre_aggregated(no_pre_aggregated_)
-{
-}
-
-void GraceMergingAggregatedStep::transformPipeline(DB::QueryPipelineBuilder & pipeline, const DB::BuildQueryPipelineSettings &)
-{
- if (params.max_bytes_before_external_group_by)
- {
- throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "max_bytes_before_external_group_by is not supported in GraceMergingAggregatedStep");
- }
- auto num_streams = pipeline.getNumStreams();
- auto transform_params = std::make_shared<DB::AggregatingTransformParams>(pipeline.getHeader(), params, true);
- pipeline.resize(1);
- auto build_transform = [&](DB::OutputPortRawPtrs outputs)
- {
- DB::Processors new_processors;
- for (auto & output : outputs)
- {
- auto op = std::make_shared<GraceMergingAggregatedTransform>(pipeline.getHeader(), transform_params, context, no_pre_aggregated);
- new_processors.push_back(op);
- DB::connect(*output, op->getInputs().front());
- }
- return new_processors;
- };
- pipeline.transform(build_transform);
- pipeline.resize(num_streams, true);
-}
-
-void GraceMergingAggregatedStep::describeActions(DB::IQueryPlanStep::FormatSettings & settings) const
-{
- return params.explain(settings.out, settings.offset);
-}
-
-void GraceMergingAggregatedStep::describeActions(DB::JSONBuilder::JSONMap & map) const
-{
- params.explain(map);
-}
-
-void GraceMergingAggregatedStep::updateOutputStream()
-{
- output_stream = createOutputStream(input_streams.front(), buildOutputHeader(input_streams.front().header, params, true), getDataStreamTraits());
-}
-
-GraceMergingAggregatedTransform::GraceMergingAggregatedTransform(const DB::Block &header_, DB::AggregatingTransformParamsPtr params_, DB::ContextPtr context_, bool no_pre_aggregated_)
+ bool no_pre_aggregated_,
+ bool final_output_)
: IProcessor({header_}, {params_->getHeader()})
, header(header_)
, params(params_)
@@ -114,8 +45,10 @@ GraceMergingAggregatedTransform::GraceMergingAggregatedTransform(const DB::Block
, key_columns(params_->params.keys_size)
, aggregate_columns(params_->params.aggregates_size)
, no_pre_aggregated(no_pre_aggregated_)
+ , final_output(final_output_)
, tmp_data_disk(std::make_unique<DB::TemporaryDataOnDisk>(context_->getTempDataOnDisk()))
{
+ output_header = params->getHeader();
auto config = GraceMergingAggregateConfig::loadFromContext(context);
max_buckets = config.max_grace_aggregate_merging_buckets;
throw_on_overflow_buckets = config.throw_on_overflow_grace_aggregate_merging_buckets;
@@ -129,7 +62,7 @@ GraceMergingAggregatedTransform::GraceMergingAggregatedTransform(const DB::Block
current_data_variants = std::make_shared<DB::AggregatedDataVariants>();
}
-GraceMergingAggregatedTransform::~GraceMergingAggregatedTransform()
+GraceAggregatingTransform::~GraceAggregatingTransform()
{
LOG_INFO(
logger,
@@ -145,7 +78,7 @@ GraceMergingAggregatedTransform::~GraceMergingAggregatedTransform()
total_scatter_time);
}
-GraceMergingAggregatedTransform::Status GraceMergingAggregatedTransform::prepare()
+GraceAggregatingTransform::Status GraceAggregatingTransform::prepare()
{
auto & output = outputs.front();
auto & input = inputs.front();
@@ -207,7 +140,7 @@ GraceMergingAggregatedTransform::Status GraceMergingAggregatedTransform::prepare
return Status::Ready;
}
-void GraceMergingAggregatedTransform::work()
+void GraceAggregatingTransform::work()
{
if (has_input)
{
@@ -235,7 +168,7 @@ void GraceMergingAggregatedTransform::work()
return;
}
- while(block_converter->hasNext())
+ while (block_converter->hasNext())
{
auto block = block_converter->next();
if (!block.rows())
@@ -253,7 +186,7 @@ void GraceMergingAggregatedTransform::work()
}
}
-bool GraceMergingAggregatedTransform::extendBuckets()
+bool GraceAggregatingTransform::extendBuckets()
{
if (!current_data_variants || current_data_variants->size() < aggregated_keys_before_extend_buckets)
return false;
@@ -270,7 +203,7 @@ bool GraceMergingAggregatedTransform::extendBuckets()
"Too many buckets, limit is {}. Please consider increate offhead size or partitoin number",
max_buckets);
else
- return false;
+ return false;
}
LOG_DEBUG(logger, "Extend buckets num from {} to {}", current_size, next_size);
for (size_t i = current_size; i < next_size; ++i)
@@ -278,7 +211,7 @@ bool GraceMergingAggregatedTransform::extendBuckets()
return true;
}
-void GraceMergingAggregatedTransform::rehashDataVariants()
+void GraceAggregatingTransform::rehashDataVariants()
{
auto before_memoery_usage = currentThreadGroupMemoryUsage();
@@ -289,7 +222,7 @@ void GraceMergingAggregatedTransform::rehashDataVariants()
no_more_keys = false;
size_t bucket_n = 0;
- while(converter->hasNext())
+ while (converter->hasNext())
{
auto block = converter->next();
if (!block.rows())
@@ -302,7 +235,11 @@ void GraceMergingAggregatedTransform::rehashDataVariants()
for (size_t i = 0; i < current_bucket_index; ++i)
{
if (scattered_blocks[i].rows())
- throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Scattered blocks should not belong to buckets with index({}) < current_bucket_index({})", i, current_bucket_index);
+ throw DB::Exception(
+ DB::ErrorCodes::LOGICAL_ERROR,
+ "Scattered blocks should not belong to buckets with index({}) < current_bucket_index({})",
+ i,
+ current_bucket_index);
}
for (size_t i = current_bucket_index + 1; i < getBucketsNum(); ++i)
{
@@ -324,7 +261,7 @@ void GraceMergingAggregatedTransform::rehashDataVariants()
ReadableSize(currentThreadGroupMemoryUsage()));
};
-DB::Blocks GraceMergingAggregatedTransform::scatterBlock(const DB::Block & block)
+DB::Blocks GraceAggregatingTransform::scatterBlock(const DB::Block & block)
{
if (!block.rows())
return {};
@@ -339,13 +276,14 @@ DB::Blocks GraceMergingAggregatedTransform::scatterBlock(const DB::Block & block
return blocks;
}
-void GraceMergingAggregatedTransform::addBlockIntoFileBucket(size_t bucket_index, const DB::Block & block, bool is_original_block)
+void GraceAggregatingTransform::addBlockIntoFileBucket(size_t bucket_index, const DB::Block & block, bool is_original_block)
{
if (!block.rows())
return;
if (roundUpToPowerOfTwoOrZero(bucket_index + 1) > static_cast<size_t>(block.info.bucket_num))
{
- throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Add invalid block with bucket_num {} into bucket {}", block.info.bucket_num, bucket_index);
+ throw DB::Exception(
+ DB::ErrorCodes::LOGICAL_ERROR, "Add invalid block with bucket_num {} into bucket {}", block.info.bucket_num, bucket_index);
}
auto & file_stream = buckets[bucket_index];
file_stream.pending_bytes += block.allocatedBytes();
@@ -360,7 +298,7 @@ void GraceMergingAggregatedTransform::addBlockIntoFileBucket(size_t bucket_index
}
}
-void GraceMergingAggregatedTransform::flushBuckets()
+void GraceAggregatingTransform::flushBuckets()
{
for (size_t i = current_bucket_index + 1; i < getBucketsNum(); ++i)
flushBucket(i);
@@ -394,7 +332,7 @@ static size_t flushBlocksInfoDisk(DB::TemporaryFileStream * file_stream, std::li
return flush_bytes;
}
-size_t GraceMergingAggregatedTransform::flushBucket(size_t bucket_index)
+size_t GraceAggregatingTransform::flushBucket(size_t bucket_index)
{
Stopwatch watch;
auto & file_stream = buckets[bucket_index];
@@ -419,7 +357,7 @@ size_t GraceMergingAggregatedTransform::flushBucket(size_t bucket_index)
return flush_bytes;
}
-std::unique_ptr<AggregateDataBlockConverter> GraceMergingAggregatedTransform::prepareBucketOutputBlocks(size_t bucket_index)
+std::unique_ptr<AggregateDataBlockConverter> GraceAggregatingTransform::prepareBucketOutputBlocks(size_t bucket_index)
{
auto & buffer_file_stream = buckets[bucket_index];
if (!current_data_variants && !buffer_file_stream.intermediate_file_stream && buffer_file_stream.intermediate_blocks.empty()
@@ -485,7 +423,7 @@ std::unique_ptr<AggregateDataBlockConverter> GraceMergingAggregatedTransform::pr
}
auto last_data_variants_size = current_data_variants->size();
- auto converter = currentDataVariantToBlockConverter(true);
+ auto converter = currentDataVariantToBlockConverter(final_output);
LOG_INFO(
logger,
"prepare to output bucket {}, aggregated result keys: {}, keys size: {}, read bytes from disk: {}, read rows: {}, time: {} ms",
@@ -498,7 +436,7 @@ std::unique_ptr<AggregateDataBlockConverter> GraceMergingAggregatedTransform::pr
return std::move(converter);
}
-std::unique_ptr<AggregateDataBlockConverter> GraceMergingAggregatedTransform::currentDataVariantToBlockConverter(bool final)
+std::unique_ptr<AggregateDataBlockConverter> GraceAggregatingTransform::currentDataVariantToBlockConverter(bool final)
{
if (!current_data_variants)
{
@@ -509,7 +447,7 @@ std::unique_ptr<AggregateDataBlockConverter> GraceMergingAggregatedTransform::cu
return std::move(converter);
}
-void GraceMergingAggregatedTransform::checkAndSetupCurrentDataVariants()
+void GraceAggregatingTransform::checkAndSetupCurrentDataVariants()
{
if (!current_data_variants)
{
@@ -518,7 +456,7 @@ void GraceMergingAggregatedTransform::checkAndSetupCurrentDataVariants()
}
}
-void GraceMergingAggregatedTransform::mergeOneBlock(const DB::Block &block, bool is_original_block)
+void GraceAggregatingTransform::mergeOneBlock(const DB::Block & block, bool is_original_block)
{
if (!block.rows())
return;
@@ -579,7 +517,8 @@ void GraceMergingAggregatedTransform::mergeOneBlock(const DB::Block &block, bool
if (is_original_block && no_pre_aggregated)
{
- params->aggregator.executeOnBlock(scattered_blocks[current_bucket_index], *current_data_variants, key_columns, aggregate_columns, no_more_keys);
+ params->aggregator.executeOnBlock(
+ scattered_blocks[current_bucket_index], *current_data_variants, key_columns, aggregate_columns, no_more_keys);
}
else
{
@@ -588,7 +527,7 @@ void GraceMergingAggregatedTransform::mergeOneBlock(const DB::Block &block, bool
}
}
-bool GraceMergingAggregatedTransform::isMemoryOverflow()
+bool GraceAggregatingTransform::isMemoryOverflow()
{
/// More greedy memory usage strategy.
if (!current_data_variants)
@@ -606,7 +545,8 @@ bool GraceMergingAggregatedTransform::isMemoryOverflow()
{
LOG_INFO(
logger,
- "Memory is overflow. current_mem_used: {}, max_mem_used: {}, per_key_memory_usage: {}, aggregator keys: {}, buckets: {}, hash table type: {}",
+ "Memory is overflow. current_mem_used: {}, max_mem_used: {}, per_key_memory_usage: {}, aggregator keys: {}, buckets: {}, "
+ "hash table type: {}",
ReadableSize(current_mem_used),
ReadableSize(max_mem_used),
ReadableSize(per_key_memory_usage),
@@ -622,7 +562,8 @@ bool GraceMergingAggregatedTransform::isMemoryOverflow()
{
LOG_INFO(
logger,
- "Memory is overflow on half of max usage. current_mem_used: {}, max_mem_used: {}, aggregator keys: {}, buckets: {}, hash table type: {}",
+ "Memory is overflow on half of max usage. current_mem_used: {}, max_mem_used: {}, aggregator keys: {}, buckets: {}, hash "
+ "table type: {}",
ReadableSize(current_mem_used),
ReadableSize(max_mem_used),
current_result_rows,
@@ -633,4 +574,5 @@ bool GraceMergingAggregatedTransform::isMemoryOverflow()
}
return false;
}
+
}
diff --git a/cpp-ch/local-engine/Operator/GraceMergingAggregatedStep.h b/cpp-ch/local-engine/Operator/GraceAggregatingTransform.h
similarity index 75%
copy from cpp-ch/local-engine/Operator/GraceMergingAggregatedStep.h
copy to cpp-ch/local-engine/Operator/GraceAggregatingTransform.h
index 788ca47dc0..50006ea7d7 100644
--- a/cpp-ch/local-engine/Operator/GraceMergingAggregatedStep.h
+++ b/cpp-ch/local-engine/Operator/GraceAggregatingTransform.h
@@ -14,62 +14,48 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include <unordered_map>
#include <Core/Block.h>
#include <Interpreters/Aggregator.h>
#include <Interpreters/Context.h>
+#include <Interpreters/JoinUtils.h>
#include <Processors/Chunk.h>
#include <Processors/IProcessor.h>
#include <Processors/Port.h>
-#include <Processors/QueryPlan/IQueryPlanStep.h>
#include <Processors/QueryPlan/ITransformingStep.h>
#include <Processors/Transforms/AggregatingTransform.h>
-#include <QueryPipeline/SizeLimits.h>
#include <Poco/Logger.h>
#include <Common/AggregateUtil.h>
#include <Common/logger_useful.h>
namespace local_engine
{
-/// It's used to merged aggregated data from intermediate aggregate stages, it's the final stage of
-/// aggregating.
-/// It support spilling data into disk when the memory usage is overflow.
-class GraceMergingAggregatedStep : public DB::ITransformingStep
-{
-public:
- explicit GraceMergingAggregatedStep(
- DB::ContextPtr context_,
- const DB::DataStream & input_stream_,
- DB::Aggregator::Params params_,
- bool no_pre_aggregated_);
- ~GraceMergingAggregatedStep() override = default;
-
- String getName() const override { return "GraceMergingAggregatedStep"; }
-
- void transformPipeline(DB::QueryPipelineBuilder & pipeline, const DB::BuildQueryPipelineSettings &) override;
-
- void describeActions(DB::JSONBuilder::JSONMap & map) const override;
- void describeActions(DB::IQueryPlanStep::FormatSettings & settings) const override;
-private:
- DB::ContextPtr context;
- DB::Aggregator::Params params;
- bool no_pre_aggregated;
- void updateOutputStream() override;
-};
-
-class GraceMergingAggregatedTransform : public DB::IProcessor
+/**
+ * GraceAggregatingTransform is use to aggregate original data or merging intermediate aggregating result. It support spilling data into disk
+ * when the memory usage is over limit.
+ * If the input data is original data, no_pre_aggregated is true. If it's intermediate aggregating result, no_pre_aggregated is false.
+ * IF the output data is final aggregating result, final is true, otherwise it false.
+ */
+class GraceAggregatingTransform : public DB::IProcessor
{
public:
using Status = DB::IProcessor::Status;
- explicit GraceMergingAggregatedTransform(const DB::Block &header_, DB::AggregatingTransformParamsPtr params_, DB::ContextPtr context_, bool no_pre_aggregated_);
- ~GraceMergingAggregatedTransform() override;
+ explicit GraceAggregatingTransform(
+ const DB::Block & header_,
+ DB::AggregatingTransformParamsPtr params_,
+ DB::ContextPtr context_,
+ bool no_pre_aggregated_,
+ bool final_output_);
+ ~GraceAggregatingTransform() override;
Status prepare() override;
void work() override;
- String getName() const override { return "GraceMergingAggregatedTransform"; }
+ String getName() const override { return "GraceAggregatingTransform"; }
+
private:
bool no_pre_aggregated;
+ bool final_output;
DB::Block header;
+ DB::Block output_header;
DB::ColumnRawPtrs key_columns;
DB::Aggregator::AggregateColumns aggregate_columns;
DB::AggregatingTransformParamsPtr params;
@@ -77,7 +63,7 @@ private:
DB::TemporaryDataOnDiskPtr tmp_data_disk;
DB::AggregatedDataVariantsPtr current_data_variants = nullptr;
size_t current_bucket_index = 0;
-
+
/// Followings are configurations defined in context config.
// max buckets number, default is 32
size_t max_buckets = 0;
@@ -120,7 +106,7 @@ private:
std::unique_ptr<AggregateDataBlockConverter> currentDataVariantToBlockConverter(bool final);
void checkAndSetupCurrentDataVariants();
/// Merge one block into current_data_variants.
- void mergeOneBlock(const DB::Block &block, bool is_original_block);
+ void mergeOneBlock(const DB::Block & block, bool is_original_block);
bool isMemoryOverflow();
bool input_finished = false;
diff --git a/cpp-ch/local-engine/Operator/GraceMergingAggregatedStep.cpp b/cpp-ch/local-engine/Operator/GraceMergingAggregatedStep.cpp
index 82b498e58f..348629f8aa 100644
--- a/cpp-ch/local-engine/Operator/GraceMergingAggregatedStep.cpp
+++ b/cpp-ch/local-engine/Operator/GraceMergingAggregatedStep.cpp
@@ -81,7 +81,7 @@ void GraceMergingAggregatedStep::transformPipeline(DB::QueryPipelineBuilder & pi
DB::Processors new_processors;
for (auto & output : outputs)
{
- auto op = std::make_shared<GraceMergingAggregatedTransform>(pipeline.getHeader(), transform_params, context, no_pre_aggregated);
+ auto op = std::make_shared<GraceAggregatingTransform>(pipeline.getHeader(), transform_params, context, no_pre_aggregated, true);
new_processors.push_back(op);
DB::connect(*output, op->getInputs().front());
}
@@ -106,531 +106,5 @@ void GraceMergingAggregatedStep::updateOutputStream()
output_stream = createOutputStream(input_streams.front(), buildOutputHeader(input_streams.front().header, params, true), getDataStreamTraits());
}
-GraceMergingAggregatedTransform::GraceMergingAggregatedTransform(const DB::Block &header_, DB::AggregatingTransformParamsPtr params_, DB::ContextPtr context_, bool no_pre_aggregated_)
- : IProcessor({header_}, {params_->getHeader()})
- , header(header_)
- , params(params_)
- , context(context_)
- , key_columns(params_->params.keys_size)
- , aggregate_columns(params_->params.aggregates_size)
- , no_pre_aggregated(no_pre_aggregated_)
- , tmp_data_disk(std::make_unique<DB::TemporaryDataOnDisk>(context_->getTempDataOnDisk()))
-{
- auto config = GraceMergingAggregateConfig::loadFromContext(context);
- max_buckets = config.max_grace_aggregate_merging_buckets;
- throw_on_overflow_buckets = config.throw_on_overflow_grace_aggregate_merging_buckets;
- aggregated_keys_before_extend_buckets = config.aggregated_keys_before_extend_grace_aggregate_merging_buckets;
- aggregated_keys_before_extend_buckets = PODArrayUtil::adjustMemoryEfficientSize(aggregated_keys_before_extend_buckets);
- max_pending_flush_blocks_per_bucket = config.max_pending_flush_blocks_per_grace_aggregate_merging_bucket;
- max_allowed_memory_usage_ratio = config.max_allowed_memory_usage_ratio_for_aggregate_merging;
- // bucket 0 is for in-memory data, it's just a placeholder.
- buckets.emplace(0, BufferFileStream());
-
- current_data_variants = std::make_shared<DB::AggregatedDataVariants>();
-}
-
-GraceMergingAggregatedTransform::~GraceMergingAggregatedTransform()
-{
- LOG_INFO(
- logger,
- "Metrics. total_input_blocks: {}, total_input_rows: {}, total_output_blocks: {}, total_output_rows: {}, total_spill_disk_bytes: "
- "{}, total_spill_disk_time: {}, total_read_disk_time: {}, total_scatter_time: {}",
- total_input_blocks,
- total_input_rows,
- total_output_blocks,
- total_output_rows,
- total_spill_disk_bytes,
- total_spill_disk_time,
- total_read_disk_time,
- total_scatter_time);
-}
-
-GraceMergingAggregatedTransform::Status GraceMergingAggregatedTransform::prepare()
-{
- auto & output = outputs.front();
- auto & input = inputs.front();
- if (output.isFinished() || isCancelled())
- {
- input.close();
- return Status::Finished;
- }
- if (has_output)
- {
- if (output.canPush())
- {
- LOG_DEBUG(
- logger,
- "Output one chunk. rows: {}, bytes: {}, current memory usage: {}",
- output_chunk.getNumRows(),
- ReadableSize(output_chunk.bytes()),
- ReadableSize(currentThreadGroupMemoryUsage()));
- total_output_rows += output_chunk.getNumRows();
- total_output_blocks++;
- output.push(std::move(output_chunk));
- has_output = false;
- }
- return Status::PortFull;
- }
-
- if (has_input)
- return Status::Ready;
-
-
- if (!input_finished)
- {
- if (input.isFinished())
- {
- input_finished = true;
- return Status::Ready;
- }
- input.setNeeded();
- if (!input.hasData())
- return Status::NeedData;
- input_chunk = input.pull(true);
- LOG_DEBUG(
- logger,
- "Input one new chunk. rows: {}, bytes: {}, current memory usage: {}",
- input_chunk.getNumRows(),
- ReadableSize(input_chunk.bytes()),
- ReadableSize(currentThreadGroupMemoryUsage()));
- total_input_rows += input_chunk.getNumRows();
- total_input_blocks++;
- has_input = true;
- return Status::Ready;
- }
-
- if (current_bucket_index >= getBucketsNum() && (!block_converter || !block_converter->hasNext()))
- {
- output.finish();
- return Status::Finished;
- }
- return Status::Ready;
-}
-
-void GraceMergingAggregatedTransform::work()
-{
- if (has_input)
- {
- assert(!input_finished);
- auto block = header.cloneWithColumns(input_chunk.detachColumns());
- mergeOneBlock(block, true);
- has_input = false;
- }
- else
- {
- assert(input_finished);
- if (!block_converter || !block_converter->hasNext())
- {
- block_converter = nullptr;
- while (current_bucket_index < getBucketsNum())
- {
- block_converter = prepareBucketOutputBlocks(current_bucket_index);
- if (block_converter)
- break;
- current_bucket_index++;
- }
- }
- if (!block_converter)
- {
- return;
- }
-
- while(block_converter->hasNext())
- {
- auto block = block_converter->next();
- if (!block.rows())
- continue;
- output_chunk = DB::Chunk(block.getColumns(), block.rows());
- has_output = true;
- break;
- }
-
- if (!block_converter->hasNext())
- {
- block_converter = nullptr;
- current_bucket_index++;
- }
- }
-}
-
-bool GraceMergingAggregatedTransform::extendBuckets()
-{
- if (!current_data_variants || current_data_variants->size() < aggregated_keys_before_extend_buckets)
- return false;
-
- auto current_size = getBucketsNum();
- auto next_size = current_size * 2;
- /// We have a soft limit on the number of buckets. When throw_on_overflow_buckets = false, we just
- /// continue to run with the current number of buckets until the executor is killed by spark scheduler.
- if (next_size > max_buckets)
- {
- if (throw_on_overflow_buckets)
- throw DB::Exception(
- DB::ErrorCodes::LOGICAL_ERROR,
- "Too many buckets, limit is {}. Please consider increate offhead size or partitoin number",
- max_buckets);
- else
- return false;
- }
- LOG_DEBUG(logger, "Extend buckets num from {} to {}", current_size, next_size);
- for (size_t i = current_size; i < next_size; ++i)
- buckets.emplace(i, BufferFileStream());
- return true;
-}
-
-void GraceMergingAggregatedTransform::rehashDataVariants()
-{
- auto before_memoery_usage = currentThreadGroupMemoryUsage();
-
- auto converter = currentDataVariantToBlockConverter(false);
- checkAndSetupCurrentDataVariants();
- size_t block_rows = 0;
- size_t block_memory_usage = 0;
- no_more_keys = false;
-
- size_t bucket_n = 0;
- while(converter->hasNext())
- {
- auto block = converter->next();
- if (!block.rows())
- continue;
- block_rows += block.rows();
- block_memory_usage += block.allocatedBytes();
- auto scattered_blocks = scatterBlock(block);
- block = {};
- /// the new scattered blocks from block will alway belongs to the buckets with index >= current_bucket_index
- for (size_t i = 0; i < current_bucket_index; ++i)
- {
- if (scattered_blocks[i].rows())
- throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Scattered blocks should not belong to buckets with index({}) < current_bucket_index({})", i, current_bucket_index);
- }
- for (size_t i = current_bucket_index + 1; i < getBucketsNum(); ++i)
- {
- addBlockIntoFileBucket(i, scattered_blocks[i], false);
- scattered_blocks[i] = {};
- }
-
- params->aggregator.mergeOnBlock(scattered_blocks[current_bucket_index], *current_data_variants, no_more_keys, is_cancelled);
- }
- if (block_rows)
- per_key_memory_usage = block_memory_usage * 1.0 / block_rows;
-
- LOG_INFO(
- logger,
- "Rehash data variants. current_bucket_index: {}, buckets num: {}, memory usage change, from {} to {}",
- current_bucket_index,
- getBucketsNum(),
- ReadableSize(before_memoery_usage),
- ReadableSize(currentThreadGroupMemoryUsage()));
-};
-
-DB::Blocks GraceMergingAggregatedTransform::scatterBlock(const DB::Block & block)
-{
- if (!block.rows())
- return {};
- Stopwatch watch;
- size_t bucket_num = getBucketsNum();
- auto blocks = DB::JoinCommon::scatterBlockByHash(params->params.keys, block, bucket_num);
- for (auto & new_block : blocks)
- {
- new_block.info.bucket_num = static_cast<Int32>(bucket_num);
- }
- total_scatter_time += watch.elapsedMilliseconds();
- return blocks;
-}
-
-void GraceMergingAggregatedTransform::addBlockIntoFileBucket(size_t bucket_index, const DB::Block & block, bool is_original_block)
-{
- if (!block.rows())
- return;
- if (roundUpToPowerOfTwoOrZero(bucket_index + 1) > static_cast<size_t>(block.info.bucket_num))
- {
- throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Add invalid block with bucket_num {} into bucket {}", block.info.bucket_num, bucket_index);
- }
- auto & file_stream = buckets[bucket_index];
- file_stream.pending_bytes += block.allocatedBytes();
- if (is_original_block && no_pre_aggregated)
- file_stream.original_blocks.push_back(block);
- else
- file_stream.intermediate_blocks.push_back(block);
- if (file_stream.pending_bytes > max_pending_flush_blocks_per_bucket)
- {
- flushBucket(bucket_index);
- file_stream.pending_bytes = 0;
- }
-}
-
-void GraceMergingAggregatedTransform::flushBuckets()
-{
- for (size_t i = current_bucket_index + 1; i < getBucketsNum(); ++i)
- flushBucket(i);
-}
-
-static size_t flushBlocksInfoDisk(DB::TemporaryFileStream * file_stream, std::list<DB::Block> & blocks)
-{
- size_t flush_bytes = 0;
- DB::Blocks tmp_blocks;
- while (!blocks.empty())
- {
- while (!blocks.empty())
- {
- if (!tmp_blocks.empty() && tmp_blocks.back().info.bucket_num != blocks.front().info.bucket_num)
- break;
- tmp_blocks.push_back(blocks.front());
- blocks.pop_front();
- }
- auto bucket = tmp_blocks.front().info.bucket_num;
- auto merged_block = BlockUtil::concatenateBlocksMemoryEfficiently(std::move(tmp_blocks));
- merged_block.info.bucket_num = bucket;
- tmp_blocks.clear();
- flush_bytes += merged_block.bytes();
- if (merged_block.rows())
- {
- file_stream->write(merged_block);
- }
- }
- if (flush_bytes)
- file_stream->flush();
- return flush_bytes;
-}
-
-size_t GraceMergingAggregatedTransform::flushBucket(size_t bucket_index)
-{
- Stopwatch watch;
- auto & file_stream = buckets[bucket_index];
- size_t flush_bytes = 0;
- if (!file_stream.original_blocks.empty())
- {
- if (!file_stream.original_file_stream)
- file_stream.original_file_stream = &tmp_data_disk->createStream(header);
- flush_bytes += flushBlocksInfoDisk(file_stream.original_file_stream, file_stream.original_blocks);
- }
- if (!file_stream.intermediate_blocks.empty())
- {
- if (!file_stream.intermediate_file_stream)
- {
- auto intermediate_header = params->aggregator.getHeader(false);
- file_stream.intermediate_file_stream = &tmp_data_disk->createStream(intermediate_header);
- }
- flush_bytes += flushBlocksInfoDisk(file_stream.intermediate_file_stream, file_stream.intermediate_blocks);
- }
- total_spill_disk_bytes += flush_bytes;
- total_spill_disk_time += watch.elapsedMilliseconds();
- return flush_bytes;
-}
-
-std::unique_ptr<AggregateDataBlockConverter> GraceMergingAggregatedTransform::prepareBucketOutputBlocks(size_t bucket_index)
-{
- auto & buffer_file_stream = buckets[bucket_index];
- if (!current_data_variants && !buffer_file_stream.intermediate_file_stream && buffer_file_stream.intermediate_blocks.empty()
- && !buffer_file_stream.original_file_stream && buffer_file_stream.original_blocks.empty())
- {
- return nullptr;
- }
-
- size_t read_bytes = 0;
- size_t read_rows = 0;
- Stopwatch watch;
-
- checkAndSetupCurrentDataVariants();
-
- if (buffer_file_stream.intermediate_file_stream)
- {
- buffer_file_stream.intermediate_file_stream->finishWriting();
- while (true)
- {
- auto block = buffer_file_stream.intermediate_file_stream->read();
- if (!block.rows())
- break;
- read_bytes += block.bytes();
- read_rows += block.rows();
- mergeOneBlock(block, false);
- block = {};
- }
- buffer_file_stream.intermediate_file_stream = nullptr;
- total_read_disk_time += watch.elapsedMilliseconds();
- }
- if (!buffer_file_stream.intermediate_blocks.empty())
- {
- for (auto & block : buffer_file_stream.intermediate_blocks)
- {
- mergeOneBlock(block, false);
- block = {};
- }
- }
-
- if (buffer_file_stream.original_file_stream)
- {
- buffer_file_stream.original_file_stream->finishWriting();
- while (true)
- {
- auto block = buffer_file_stream.original_file_stream->read();
- if (!block.rows())
- break;
- read_bytes += block.bytes();
- read_rows += block.rows();
- mergeOneBlock(block, true);
- block = {};
- }
- buffer_file_stream.original_file_stream = nullptr;
- total_read_disk_time += watch.elapsedMilliseconds();
- }
- if (!buffer_file_stream.original_blocks.empty())
- {
- for (auto & block : buffer_file_stream.original_blocks)
- {
- mergeOneBlock(block, true);
- block = {};
- }
- }
-
- auto last_data_variants_size = current_data_variants->size();
- auto converter = currentDataVariantToBlockConverter(true);
- LOG_INFO(
- logger,
- "prepare to output bucket {}, aggregated result keys: {}, keys size: {}, read bytes from disk: {}, read rows: {}, time: {} ms",
- bucket_index,
- last_data_variants_size,
- params->params.keys_size,
- ReadableSize(read_bytes),
- read_rows,
- watch.elapsedMilliseconds());
- return std::move(converter);
-}
-
-std::unique_ptr<AggregateDataBlockConverter> GraceMergingAggregatedTransform::currentDataVariantToBlockConverter(bool final)
-{
- if (!current_data_variants)
- {
- throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "current data variants is null");
- }
- auto converter = std::make_unique<AggregateDataBlockConverter>(params->aggregator, current_data_variants, final);
- current_data_variants = nullptr;
- return std::move(converter);
-}
-
-void GraceMergingAggregatedTransform::checkAndSetupCurrentDataVariants()
-{
- if (!current_data_variants)
- {
- current_data_variants = std::make_shared<DB::AggregatedDataVariants>();
- no_more_keys = false;
- }
-}
-
-void GraceMergingAggregatedTransform::mergeOneBlock(const DB::Block &block, bool is_original_block)
-{
- if (!block.rows())
- return;
-
- checkAndSetupCurrentDataVariants();
- // first to flush pending bytes into disk.
- if (isMemoryOverflow())
- flushBuckets();
- // then try to extend buckets.
- if (isMemoryOverflow() && extendBuckets())
- {
- rehashDataVariants();
- }
-
- LOG_DEBUG(
- logger,
- "merge on block, rows: {}, bytes:{}, bucket: {}. current bucket: {}, total bucket: {}, mem used: {}",
- block.rows(),
- ReadableSize(block.bytes()),
- block.info.bucket_num,
- current_bucket_index,
- getBucketsNum(),
- ReadableSize(currentThreadGroupMemoryUsage()));
-
- /// the block could be one read from disk. block.info.bucket_num stores the number of buckets when it was scattered.
- /// so if the buckets number is not changed since it was scattered, we don't need to scatter it again.
- if (block.info.bucket_num == static_cast<Int32>(getBucketsNum()) || getBucketsNum() == 1)
- {
- if (is_original_block && no_pre_aggregated)
- params->aggregator.executeOnBlock(block, *current_data_variants, key_columns, aggregate_columns, no_more_keys);
- else
- params->aggregator.mergeOnBlock(block, *current_data_variants, no_more_keys, is_cancelled);
- }
- else
- {
- auto bucket_num = block.info.bucket_num;
- auto scattered_blocks = scatterBlock(block);
- for (size_t i = 0; i < current_bucket_index; ++i)
- {
- if (scattered_blocks[i].rows())
- {
- throw DB::Exception(
- DB::ErrorCodes::LOGICAL_ERROR,
- "Scattered blocks should not belong to buckets with index({}) < current_bucket_index({}). bucket_num:{}. "
- "scattered_blocks.size: {}, total buckets: {}",
- i,
- current_bucket_index,
- bucket_num,
- scattered_blocks.size(),
- getBucketsNum());
- }
- }
- for (size_t i = current_bucket_index + 1; i < getBucketsNum(); ++i)
- {
- addBlockIntoFileBucket(i, scattered_blocks[i], is_original_block);
- }
-
- if (is_original_block && no_pre_aggregated)
- {
- params->aggregator.executeOnBlock(scattered_blocks[current_bucket_index], *current_data_variants, key_columns, aggregate_columns, no_more_keys);
- }
- else
- {
- params->aggregator.mergeOnBlock(scattered_blocks[current_bucket_index], *current_data_variants, no_more_keys, is_cancelled);
- }
- }
-}
-
-bool GraceMergingAggregatedTransform::isMemoryOverflow()
-{
- /// More greedy memory usage strategy.
- if (!current_data_variants)
- return false;
-
- auto memory_soft_limit = DB::CurrentThread::getGroup()->memory_tracker.getSoftLimit();
- if (!memory_soft_limit)
- return false;
- auto max_mem_used = static_cast<size_t>(memory_soft_limit * max_allowed_memory_usage_ratio);
- auto current_result_rows = current_data_variants->size();
- auto current_mem_used = currentThreadGroupMemoryUsage();
- if (per_key_memory_usage > 0)
- {
- if (current_mem_used + per_key_memory_usage * current_result_rows >= max_mem_used)
- {
- LOG_INFO(
- logger,
- "Memory is overflow. current_mem_used: {}, max_mem_used: {}, per_key_memory_usage: {}, aggregator keys: {}, buckets: {}, hash table type: {}",
- ReadableSize(current_mem_used),
- ReadableSize(max_mem_used),
- ReadableSize(per_key_memory_usage),
- current_result_rows,
- getBucketsNum(),
- current_data_variants->type);
- return true;
- }
- }
- else
- {
- if (current_mem_used * 2 >= max_mem_used)
- {
- LOG_INFO(
- logger,
- "Memory is overflow on half of max usage. current_mem_used: {}, max_mem_used: {}, aggregator keys: {}, buckets: {}, hash table type: {}",
- ReadableSize(current_mem_used),
- ReadableSize(max_mem_used),
- current_result_rows,
- getBucketsNum(),
- current_data_variants->type);
- return true;
- }
- }
- return false;
-}
}
diff --git a/cpp-ch/local-engine/Operator/GraceMergingAggregatedStep.h b/cpp-ch/local-engine/Operator/GraceMergingAggregatedStep.h
index 788ca47dc0..d4d4fb41ed 100644
--- a/cpp-ch/local-engine/Operator/GraceMergingAggregatedStep.h
+++ b/cpp-ch/local-engine/Operator/GraceMergingAggregatedStep.h
@@ -28,6 +28,7 @@
#include <Poco/Logger.h>
#include <Common/AggregateUtil.h>
#include <Common/logger_useful.h>
+#include "GraceAggregatingTransform.h"
namespace local_engine
{
@@ -57,93 +58,5 @@ private:
void updateOutputStream() override;
};
-class GraceMergingAggregatedTransform : public DB::IProcessor
-{
-public:
- using Status = DB::IProcessor::Status;
- explicit GraceMergingAggregatedTransform(const DB::Block &header_, DB::AggregatingTransformParamsPtr params_, DB::ContextPtr context_, bool no_pre_aggregated_);
- ~GraceMergingAggregatedTransform() override;
-
- Status prepare() override;
- void work() override;
- String getName() const override { return "GraceMergingAggregatedTransform"; }
-private:
- bool no_pre_aggregated;
- DB::Block header;
- DB::ColumnRawPtrs key_columns;
- DB::Aggregator::AggregateColumns aggregate_columns;
- DB::AggregatingTransformParamsPtr params;
- DB::ContextPtr context;
- DB::TemporaryDataOnDiskPtr tmp_data_disk;
- DB::AggregatedDataVariantsPtr current_data_variants = nullptr;
- size_t current_bucket_index = 0;
-
- /// Followings are configurations defined in context config.
- // max buckets number, default is 32
- size_t max_buckets = 0;
- // If the buckets number is overflow the max_buckets, throw exception or not.
- bool throw_on_overflow_buckets = false;
- // Even the memory usage has reached the limit, we still allow to aggregate some more keys before
- // extend the buckets.
- size_t aggregated_keys_before_extend_buckets = 8196;
- // The ratio of memory usage to the total memory usage of the whole query.
- double max_allowed_memory_usage_ratio = 0.9;
- // configured by max_pending_flush_blocks_per_grace_merging_bucket
- size_t max_pending_flush_blocks_per_bucket = 0;
- struct BufferFileStream
- {
- /// store the intermediate result blocks.
- std::list<DB::Block> intermediate_blocks;
- /// Only be used when there is no pre-aggregated step, store the original input blocks.
- std::list<DB::Block> original_blocks;
- /// store the intermediate result blocks.
- DB::TemporaryFileStream * intermediate_file_stream = nullptr;
- /// Only be used when there is no pre-aggregated step
- DB::TemporaryFileStream * original_file_stream = nullptr;
- size_t pending_bytes = 0;
- };
- std::unordered_map<size_t, BufferFileStream> buckets;
-
- size_t getBucketsNum() const { return buckets.size(); }
- bool extendBuckets();
- void rehashDataVariants();
- DB::Blocks scatterBlock(const DB::Block & block);
- /// Add a block into a bucket, if the pending bytes reaches limit, flush it into disk.
- void addBlockIntoFileBucket(size_t bucket_index, const DB::Block & block, bool is_original_block);
- void flushBuckets();
- size_t flushBucket(size_t bucket_index);
- /// Load blocks from disk and merge them into a new hash table, make a new AggregateDataBlockConverter
- /// to generate output blocks.
- std::unique_ptr<AggregateDataBlockConverter> prepareBucketOutputBlocks(size_t bucket);
- /// Pass current_final_blocks into a new AggregateDataBlockConverter to generate output blocks.
- std::unique_ptr<AggregateDataBlockConverter> currentDataVariantToBlockConverter(bool final);
- void checkAndSetupCurrentDataVariants();
- /// Merge one block into current_data_variants.
- void mergeOneBlock(const DB::Block &block, bool is_original_block);
- bool isMemoryOverflow();
-
- bool input_finished = false;
- bool has_input = false;
- DB::Chunk input_chunk;
- bool has_output = false;
- DB::Chunk output_chunk;
- DB::BlocksList current_final_blocks;
- std::unique_ptr<AggregateDataBlockConverter> block_converter = nullptr;
- bool no_more_keys = false;
-
- double per_key_memory_usage = 0;
-
- // metrics
- size_t total_input_blocks = 0;
- size_t total_input_rows = 0;
- size_t total_output_blocks = 0;
- size_t total_output_rows = 0;
- size_t total_spill_disk_bytes = 0;
- size_t total_spill_disk_time = 0;
- size_t total_read_disk_time = 0;
- size_t total_scatter_time = 0;
-
- Poco::Logger * logger = &Poco::Logger::get("GraceMergingAggregatedTransform");
-};
}
diff --git a/cpp-ch/local-engine/Parser/AdvancedParametersParseUtil.cpp b/cpp-ch/local-engine/Parser/AdvancedParametersParseUtil.cpp
index cc7738a15a..f2cc84082d 100644
--- a/cpp-ch/local-engine/Parser/AdvancedParametersParseUtil.cpp
+++ b/cpp-ch/local-engine/Parser/AdvancedParametersParseUtil.cpp
@@ -148,6 +148,16 @@ JoinOptimizationInfo JoinOptimizationInfo::parse(const String & advance)
return info;
}
+AggregateOptimizationInfo AggregateOptimizationInfo::parse(const String & advance)
+{
+ AggregateOptimizationInfo info;
+ auto kkvs = convertToKVs(advance);
+ auto & kvs = kkvs["AggregateParams"];
+ tryAssign(kvs, "hasPrePartialAggregate", info.has_pre_partial_aggregate);
+ tryAssign(kvs, "hasRequiredChildDistributionExpressions", info.has_required_child_distribution_expressions);
+ return info;
+}
+
WindowGroupOptimizationInfo WindowGroupOptimizationInfo::parse(const String & advance)
{
WindowGroupOptimizationInfo info;
diff --git a/cpp-ch/local-engine/Parser/AdvancedParametersParseUtil.h b/cpp-ch/local-engine/Parser/AdvancedParametersParseUtil.h
index fc478db33b..3028639bb3 100644
--- a/cpp-ch/local-engine/Parser/AdvancedParametersParseUtil.h
+++ b/cpp-ch/local-engine/Parser/AdvancedParametersParseUtil.h
@@ -39,6 +39,13 @@ struct JoinOptimizationInfo
static JoinOptimizationInfo parse(const String & advance);
};
+struct AggregateOptimizationInfo
+{
+ bool has_pre_partial_aggregate = false;
+ bool has_required_child_distribution_expressions = false;
+ static AggregateOptimizationInfo parse(const String & advance);
+};
+
struct WindowGroupOptimizationInfo
{
String window_function;
diff --git a/cpp-ch/local-engine/Parser/RelParsers/AggregateRelParser.cpp b/cpp-ch/local-engine/Parser/RelParsers/AggregateRelParser.cpp
index 06440b5029..1091e2470e 100644
--- a/cpp-ch/local-engine/Parser/RelParsers/AggregateRelParser.cpp
+++ b/cpp-ch/local-engine/Parser/RelParsers/AggregateRelParser.cpp
@@ -23,12 +23,15 @@
#include <Functions/FunctionFactory.h>
#include <Functions/FunctionHelpers.h>
#include <Operator/DefaultHashAggregateResult.h>
+#include <Operator/GraceAggregatingStep.h>
#include <Operator/GraceMergingAggregatedStep.h>
#include <Operator/StreamingAggregatingStep.h>
+#include <Parser/AdvancedParametersParseUtil.h>
#include <Parser/AggregateFunctionParser.h>
#include <Processors/QueryPlan/AggregatingStep.h>
#include <Processors/QueryPlan/ExpressionStep.h>
#include <Processors/QueryPlan/MergingAggregatedStep.h>
+#include <google/protobuf/wrappers.pb.h>
#include <Common/CHUtil.h>
#include <Common/GlutenConfig.h>
@@ -53,10 +56,10 @@ extern const SettingsUInt64 max_block_size;
}
namespace ErrorCodes
{
- extern const int LOGICAL_ERROR;
- extern const int BAD_ARGUMENTS;
- extern const int UNKNOWN_TYPE;
- extern const int ILLEGAL_TYPE_OF_ARGUMENT;
+extern const int LOGICAL_ERROR;
+extern const int BAD_ARGUMENTS;
+extern const int UNKNOWN_TYPE;
+extern const int ILLEGAL_TYPE_OF_ARGUMENT;
}
}
namespace local_engine
@@ -66,8 +69,10 @@ AggregateRelParser::AggregateRelParser(SerializedPlanParser * plan_paser_) : Rel
{
}
-DB::QueryPlanPtr AggregateRelParser::parse(DB::QueryPlanPtr query_plan, const substrait::Rel & rel, std::list<const substrait::Rel *> &)
+DB::QueryPlanPtr
+AggregateRelParser::parse(DB::QueryPlanPtr query_plan, const substrait::Rel & rel, std::list<const substrait::Rel *> & rel_stack_)
{
+ rel_stack = &rel_stack_;
setup(std::move(query_plan), rel);
addPreProjection();
@@ -94,6 +99,16 @@ DB::QueryPlanPtr AggregateRelParser::parse(DB::QueryPlanPtr query_plan, const su
LOG_TRACE(logger, "header after aggregating is: {}", plan->getCurrentDataStream().header.dumpStructure());
}
+ /// Add a check here to help find bugs, Don't remove it.
+ /// Thre order of result of result columns must be ordered grouping keys ++ ordered aggregate expression results
+ auto aggregation_output_header = plan->getCurrentDataStream().header;
+ for (size_t i = 0; i < grouping_keys.size(); ++i)
+ {
+ auto pos = aggregation_output_header.getPositionByName(grouping_keys[i]);
+ if (pos != i)
+ throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "The order of aggregation result columns is invalid");
+ }
+
/// If the groupings is empty, we still need to return one row with default values even if the input is empty.
if ((rel.aggregate().groupings().empty() || rel.aggregate().groupings()[0].grouping_expressions().empty())
&& (has_final_stage || has_complete_stage || rel.aggregate().measures().empty()))
@@ -122,13 +137,32 @@ void AggregateRelParser::setup(DB::QueryPlanPtr query_plan, const substrait::Rel
has_inter_stage = phase_set.contains(substrait::AggregationPhase::AGGREGATION_PHASE_INTERMEDIATE_TO_INTERMEDIATE);
has_final_stage = phase_set.contains(substrait::AggregationPhase::AGGREGATION_PHASE_INTERMEDIATE_TO_RESULT);
has_complete_stage = phase_set.contains(substrait::AggregationPhase::AGGREGATION_PHASE_INITIAL_TO_RESULT);
+ bool next_step_is_agg = false;
+ google::protobuf::StringValue raw_extra_params;
+ raw_extra_params.ParseFromString(aggregate_rel->advanced_extension().optimization().value());
+ auto extra_params = AggregateOptimizationInfo::parse(raw_extra_params.value());
if (aggregate_rel->measures().empty())
{
- /// According to planAggregateWithoutDistinct in AggUtils.scala, an aggregate without aggregate
- /// functions will only be organized in two stages, partial and final. In partial stage
- /// requiredChildDistributionExpressions is None, but in final stage it is Some(Seq[Expression]).
- auto configs = parseFormattedRelAdvancedOptimization(aggregate_rel->advanced_extension());
- has_final_stage = getStringConfig(configs, "has_required_child_distribution_expressions") == "true";
+ /// For aggregate without aggregate functions. we use other ways the determine wchich stage it is.
+ /// In general, when has_required_child_distribution_expressions is true, this aggregate stage is the final
+ /// merge aggregated step. But there are exceptions.
+ /// 1. when the query is a distinct like aggregation. for example.
+ /// select n_regionkey, count(distinct n_nationkey) from nation group by n_regionkey
+ /// There should be four aggregation steps.
+ /// step1: partial aggregation on keys (n_regionkey, n_nationkey) with empty aggregation functions.
+ /// step2: aggregation on keys (n_regionkey, n_nationkey) with empty aggregation functions after shuffle.
+ /// step3: aggregation on keys (n_regionkey) with partial aggregation count(n_nationkey)
+ /// step4: aggregation on keys (n_regionkey) with final aggregation merge count(n_nationkey) after shuffle
+ /// If step3 is followed by another aggregation, we must let has_final_stage = false. Otherwise it will make
+ /// columns position missmatch.
+ /// 2. the two aggregation stages are merged into one by rule `MergeTwoPhasesHashBaseAggregate`. In this case
+ /// we also let has_first_stage = false;
+ /// FIXME: Really don't like this implementation. It's too easy to be broken.
+ next_step_is_agg = rel_stack->empty() ? false : rel_stack->back()->rel_type_case() == substrait::Rel::RelTypeCase::kAggregate;
+ bool is_last_stage = extra_params.has_required_child_distribution_expressions && !next_step_is_agg;
+ has_first_stage = !extra_params.has_pre_partial_aggregate;
+ has_final_stage = extra_params.has_pre_partial_aggregate && is_last_stage;
+ has_complete_stage = !extra_params.has_pre_partial_aggregate && is_last_stage;
}
if (phase_set.size() > 1 && has_final_stage)
@@ -214,9 +248,12 @@ void AggregateRelParser::addPreProjection()
void AggregateRelParser::buildAggregateDescriptions(AggregateDescriptions & descriptions)
{
const auto & current_plan_header = plan->getCurrentDataStream().header;
- auto build_result_column_name = [this, current_plan_header](const String & function_name, const Array & params, const Strings & arg_names, substrait::AggregationPhase phase)
+ auto build_result_column_name
+ = [this, current_plan_header](
+ const String & function_name, const Array & params, const Strings & arg_names, substrait::AggregationPhase phase)
{
- if (phase == substrait::AggregationPhase::AGGREGATION_PHASE_INTERMEDIATE_TO_RESULT)
+ if (phase == substrait::AggregationPhase::AGGREGATION_PHASE_INTERMEDIATE_TO_RESULT
+ || phase == substrait::AggregationPhase::AGGREGATION_PHASE_INTERMEDIATE_TO_INTERMEDIATE)
{
assert(arg_names.size() == 1);
return arg_names[0];
@@ -242,7 +279,8 @@ void AggregateRelParser::buildAggregateDescriptions(AggregateDescriptions & desc
auto res = this->getUniqueName(result);
// Just a check for remining this issue.
if (current_plan_header.findByName(res))
- throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Name ({}) collision in header: {}", res, current_plan_header.dumpStructure());
+ throw DB::Exception(
+ DB::ErrorCodes::LOGICAL_ERROR, "Name ({}) collision in header: {}", res, current_plan_header.dumpStructure());
return res;
};
@@ -271,7 +309,8 @@ void AggregateRelParser::buildAggregateDescriptions(AggregateDescriptions & desc
// If the function is a state function, we don't need to apply `PartialMerge`.
// In INITIAL_TO_INTERMEDIATE or INITIAL_TO_RESULT phase, we do arguments -> xxState.
// In INTERMEDIATE_TO_RESULT phase, we do xxState -> xxState.
- if (agg_info.parser_func_info.phase == substrait::AggregationPhase::AGGREGATION_PHASE_INITIAL_TO_INTERMEDIATE || agg_info.parser_func_info.phase == substrait::AggregationPhase::AGGREGATION_PHASE_INITIAL_TO_RESULT)
+ if (agg_info.parser_func_info.phase == substrait::AggregationPhase::AGGREGATION_PHASE_INITIAL_TO_INTERMEDIATE
+ || agg_info.parser_func_info.phase == substrait::AggregationPhase::AGGREGATION_PHASE_INITIAL_TO_RESULT)
{
description.function = getAggregateFunction(agg_info.function_name, agg_info.arg_column_types, properties, agg_info.params);
}
@@ -370,7 +409,7 @@ void AggregateRelParser::addCompleteModeAggregatedStep()
/*only_merge*/ false,
settings[Setting::optimize_group_by_constant_keys],
settings[Setting::min_hit_rate_to_use_consecutive_keys_optimization],
- /*StatsCollectingParams*/{});
+ /*StatsCollectingParams*/ {});
auto merging_step = std::make_unique<GraceMergingAggregatedStep>(getContext(), plan->getCurrentDataStream(), params, true);
steps.emplace_back(merging_step.get());
plan->addStep(std::move(merging_step));
@@ -397,7 +436,7 @@ void AggregateRelParser::addCompleteModeAggregatedStep()
/*only_merge*/ false,
settings[Setting::optimize_group_by_constant_keys],
settings[Setting::min_hit_rate_to_use_consecutive_keys_optimization],
- /*StatsCollectingParams*/{});
+ /*StatsCollectingParams*/ {});
auto aggregating_step = std::make_unique<AggregatingStep>(
plan->getCurrentDataStream(),
@@ -427,6 +466,16 @@ void AggregateRelParser::addAggregatingStep()
const auto & settings = getContext()->getSettingsRef();
auto config = StreamingAggregateConfig::loadFromContext(getContext());
+ bool is_distinct_aggreate = false;
+ if (!rel_stack->empty())
+ {
+ const auto & next_rel = *(rel_stack->back());
+ if (next_rel.rel_type_case() == substrait::Rel::RelTypeCase::kAggregate)
+ {
+ is_distinct_aggreate = true;
+ }
+ }
+
if (config.enable_streaming_aggregating)
{
// Disable spilling to disk.
@@ -455,10 +504,32 @@ void AggregateRelParser::addAggregatingStep()
/*only_merge*/ false,
settings[Setting::optimize_group_by_constant_keys],
settings[Setting::min_hit_rate_to_use_consecutive_keys_optimization],
- /*StatsCollectingParams*/{});
- auto aggregating_step = std::make_unique<StreamingAggregatingStep>(getContext(), plan->getCurrentDataStream(), params);
- steps.emplace_back(aggregating_step.get());
- plan->addStep(std::move(aggregating_step));
+ /*StatsCollectingParams*/ {});
+ if (!is_distinct_aggreate)
+ {
+ auto aggregating_step = std::make_unique<StreamingAggregatingStep>(getContext(), plan->getCurrentDataStream(), params);
+ steps.emplace_back(aggregating_step.get());
+ plan->addStep(std::move(aggregating_step));
+ }
+ else
+ {
+ /// If it's an aggregation query involves distinct, for example,
+ /// select n_regionkey, count(distinct(n_name)), sum(n_nationkey) from nation group by n_regionkey
+ /// The running steps are as follow in general.
+ /// step1: partial aggregation on keys (n_regionkey, n_name) with empty aggregation functions.
+ /// step2: exchagne shuffle
+ /// step3: aggregation on keys (n_regionkey, n_name) with partial aggregation sum(n_nationkey)
+ /// step4: aggregation on keys (n_regionkey) with partial aggregation merge sum(n_nationkey), and partial aggregation count(n_name)
+ /// step5: exchange shuffle
+ /// step6: aggregation on keys (n_regionkey) with final aggregation merge sum(n_nationkey) and count(n_name)
+ /// We cannot use streaming aggregating strategy in step3. Otherwise it will generate multiple blocks with same n_name in them. This
+ /// will make the result for count(distinct(n_name)) wrong. step3 must finish all inputs before it puts any block into step4.
+ /// So we introduce GraceAggregatingStep here, it can handle mass data with high cardinality.
+ auto aggregating_step
+ = std::make_unique<GraceAggregatingStep>(getContext(), plan->getCurrentDataStream(), params, has_first_stage);
+ steps.emplace_back(aggregating_step.get());
+ plan->addStep(std::move(aggregating_step));
+ }
}
else
{
@@ -482,7 +553,7 @@ void AggregateRelParser::addAggregatingStep()
/*only_merge*/ false,
settings[Setting::optimize_group_by_constant_keys],
settings[Setting::min_hit_rate_to_use_consecutive_keys_optimization],
- /*StatsCollectingParams*/{});
+ /*StatsCollectingParams*/ {});
auto aggregating_step = std::make_unique<AggregatingStep>(
plan->getCurrentDataStream(),
diff --git a/cpp-ch/local-engine/Parser/RelParsers/AggregateRelParser.h b/cpp-ch/local-engine/Parser/RelParsers/AggregateRelParser.h
index 8b4ec9c4af..afcd354244 100644
--- a/cpp-ch/local-engine/Parser/RelParsers/AggregateRelParser.h
+++ b/cpp-ch/local-engine/Parser/RelParsers/AggregateRelParser.h
@@ -56,6 +56,7 @@ private:
bool has_final_stage = false;
bool has_complete_stage = false;
+ std::list<const substrait::Rel *> * rel_stack;
DB::QueryPlanPtr plan = nullptr;
const substrait::AggregateRel * aggregate_rel = nullptr;
std::vector<AggregateInfo> aggregates;
diff --git a/cpp-ch/local-engine/Parser/RelParsers/RelParser.cpp b/cpp-ch/local-engine/Parser/RelParsers/RelParser.cpp
index dc36df0ad0..773755db4f 100644
--- a/cpp-ch/local-engine/Parser/RelParsers/RelParser.cpp
+++ b/cpp-ch/local-engine/Parser/RelParsers/RelParser.cpp
@@ -95,42 +95,6 @@ RelParser::parse(std::vector<DB::QueryPlanPtr> & input_plans_, const substrait::
return parse(std::move(input_plans_[0]), rel, rel_stack_);
}
-std::map<std::string, std::string>
-RelParser::parseFormattedRelAdvancedOptimization(const substrait::extensions::AdvancedExtension & advanced_extension)
-{
- std::map<std::string, std::string> configs;
- if (advanced_extension.has_optimization())
- {
- google::protobuf::StringValue msg;
- advanced_extension.optimization().UnpackTo(&msg);
- Poco::StringTokenizer kvs(msg.value(), "\n");
- for (auto & kv : kvs)
- {
- if (kv.empty())
- continue;
- auto pos = kv.find('=');
- if (pos == std::string::npos)
- {
- throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Invalid optimization config:{}.", kv);
- }
- auto key = kv.substr(0, pos);
- auto value = kv.substr(pos + 1);
- configs[key] = value;
- }
- }
- return configs;
-}
-
-std::string
-RelParser::getStringConfig(const std::map<std::string, std::string> & configs, const std::string & key, const std::string & default_value)
-{
- auto it = configs.find(key);
- if (it == configs.end())
- return default_value;
- return it->second;
-}
-
-
RelParserFactory & RelParserFactory::instance()
{
static RelParserFactory factory;
diff --git a/cpp-ch/local-engine/Parser/RelParsers/RelParser.h b/cpp-ch/local-engine/Parser/RelParsers/RelParser.h
index 522dc05d39..838e521476 100644
--- a/cpp-ch/local-engine/Parser/RelParsers/RelParser.h
+++ b/cpp-ch/local-engine/Parser/RelParsers/RelParser.h
@@ -88,11 +88,6 @@ protected:
return plan_parser->toFunctionNode(action_dag, function, args);
}
- static std::map<std::string, std::string>
- parseFormattedRelAdvancedOptimization(const substrait::extensions::AdvancedExtension & advanced_extension);
- static std::string
- getStringConfig(const std::map<std::string, std::string> & configs, const std::string & key, const std::string & default_value = "");
-
SerializedPlanParser * plan_parser;
};
diff --git a/cpp-ch/local-engine/Parser/SerializedPlanParser.cpp b/cpp-ch/local-engine/Parser/SerializedPlanParser.cpp
index 52b5709555..c2bee8e799 100644
--- a/cpp-ch/local-engine/Parser/SerializedPlanParser.cpp
+++ b/cpp-ch/local-engine/Parser/SerializedPlanParser.cpp
@@ -1213,7 +1213,16 @@ std::unique_ptr<LocalExecutor> SerializedPlanParser::createExecutor(DB::QueryPla
Stopwatch stopwatch;
const Settings & settings = context->getSettingsRef();
- auto builder = buildQueryPipeline(*query_plan);
+ DB::QueryPipelineBuilderPtr builder = nullptr;
+ try
+ {
+ builder = buildQueryPipeline(*query_plan);
+ }
+ catch (...)
+ {
+ LOG_ERROR(getLogger("SerializedPlanParser"), "Invalid plan:\n{}", PlanUtil::explainPlan(*query_plan));
+ throw;
+ }
assert(s_plan.relations_size() == 1);
@@ -1224,7 +1233,10 @@ std::unique_ptr<LocalExecutor> SerializedPlanParser::createExecutor(DB::QueryPla
auto * logger = &Poco::Logger::get("SerializedPlanParser");
LOG_INFO(logger, "build pipeline {} ms", stopwatch.elapsedMicroseconds() / 1000.0);
LOG_DEBUG(
- logger, "clickhouse plan [optimization={}]:\n{}", settings[Setting::query_plan_enable_optimizations], PlanUtil::explainPlan(*query_plan));
+ logger,
+ "clickhouse plan [optimization={}]:\n{}",
+ settings[Setting::query_plan_enable_optimizations],
+ PlanUtil::explainPlan(*query_plan));
auto config = ExecutorConfig::loadFromContext(context);
return std::make_unique<LocalExecutor>(std::move(query_plan), std::move(builder), config.dump_pipeline);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org