You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2023/01/03 03:49:46 UTC
[GitHub] [flink] swuferhong opened a new pull request, #21586: [FLINK-30352][table-runtime] Introduce adaptive hash aggregate to adaptively determine whether local hash aggregate is required at runtime
swuferhong opened a new pull request, #21586:
URL: https://github.com/apache/flink/pull/21586
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
This pr is aims to introduce adaptive hash aggregate in `batch` mode to adaptively determine whether local hash aggregate is required according to the aggregation degree of local hash aggregate. `adaptive hash agg` will first sample the data, and then decide whether `local hash agg` is required according to the degree of aggregation of the sampled data. If requiring `local hash agg`, it will not take any action. If required, we will transfer the output rowType of data to same with aggregated results and pass to downstream.
## Brief change log
- Add codeGen code in `HashAggCodeGenerator`,
- Add ITCase in `HashAggITCase`.
## Verifying this change
- Add ITCase in `HashAggITCase`.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? no document now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on pull request #21586: [FLINK-30542][table-runtime] Introduce adaptive local hash aggregate to adaptively determine whether local hash aggregate is required at runtime
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on PR #21586:
URL: https://github.com/apache/flink/pull/21586#issuecomment-1409590272
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on pull request #21586: [FLINK-30542][table-runtime] Introduce adaptive local hash aggregate to adaptively determine whether local hash aggregate is required at runtime
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on PR #21586:
URL: https://github.com/apache/flink/pull/21586#issuecomment-1408474811
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #21586: [FLINK-30542][table-runtime] Introduce adaptive hash aggregate to adaptively determine whether local hash aggregate is required at runtime
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #21586:
URL: https://github.com/apache/flink/pull/21586#discussion_r1066676100
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashAggregate.scala:
##########
@@ -53,7 +53,8 @@ class BatchPhysicalHashAggregate(
grouping: Array[Int],
auxGrouping: Array[Int],
aggCallToAggFunction: Seq[(AggregateCall, UserDefinedFunction)],
- isMerge: Boolean)
+ isMerge: Boolean,
+ canProjection: Boolean)
Review Comment:
> For global hash agg, we don't need this variable, it should always be false.
Done!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe commented on a diff in pull request #21586: [FLINK-30542][table-runtime] Introduce adaptive hash aggregate to adaptively determine whether local hash aggregate is required at runtime
Posted by GitBox <gi...@apache.org>.
godfreyhe commented on code in PR #21586:
URL: https://github.com/apache/flink/pull/21586#discussion_r1073610534
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecHashAggregate.java:
##########
@@ -58,6 +58,7 @@ public class BatchExecHashAggregate extends ExecNodeBase<RowData>
private final RowType aggInputRowType;
private final boolean isMerge;
private final boolean isFinal;
+ private final boolean canDoAdaptiveHashAgg;
Review Comment:
how about `supportAdaptiveLocalHashAgg` ?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -38,32 +42,52 @@ import org.apache.calcite.tools.RelBuilder
* aggregateBuffers should be update(e.g.: setInt) in [[BinaryRowData]]. (Hash Aggregate performs
* much better than Sort Aggregate).
*/
-class HashAggCodeGenerator(
- ctx: CodeGeneratorContext,
- builder: RelBuilder,
- aggInfoList: AggregateInfoList,
- inputType: RowType,
- outputType: RowType,
- grouping: Array[Int],
- auxGrouping: Array[Int],
- isMerge: Boolean,
- isFinal: Boolean) {
+object HashAggCodeGenerator {
- private lazy val aggInfos: Array[AggregateInfo] = aggInfoList.aggInfos
+ // It is a experimental config, will may be removed later.
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED: ConfigOption[JBoolean] =
+ key("table.exec.adaptive-local-hash-agg.enabled")
+ .booleanType()
+ .defaultValue(Boolean.box(true))
+ .withDescription("Whether to enable adaptive local hash agg")
Review Comment:
Whether to enable adaptive local hash aggregation, this is only used for batch job. Default value is true.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -38,32 +42,52 @@ import org.apache.calcite.tools.RelBuilder
* aggregateBuffers should be update(e.g.: setInt) in [[BinaryRowData]]. (Hash Aggregate performs
* much better than Sort Aggregate).
*/
-class HashAggCodeGenerator(
- ctx: CodeGeneratorContext,
- builder: RelBuilder,
- aggInfoList: AggregateInfoList,
- inputType: RowType,
- outputType: RowType,
- grouping: Array[Int],
- auxGrouping: Array[Int],
- isMerge: Boolean,
- isFinal: Boolean) {
+object HashAggCodeGenerator {
- private lazy val aggInfos: Array[AggregateInfo] = aggInfoList.aggInfos
+ // It is a experimental config, will may be removed later.
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED: ConfigOption[JBoolean] =
+ key("table.exec.adaptive-local-hash-agg.enabled")
+ .booleanType()
+ .defaultValue(Boolean.box(true))
+ .withDescription("Whether to enable adaptive local hash agg")
- private lazy val functionIdentifiers: Map[AggregateFunction[_, _], String] =
- AggCodeGenHelper.getFunctionIdentifiers(aggInfos)
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLE_POINT: ConfigOption[JLong] =
+ key("table.exec.adaptive-local-hash-agg.sample-threshold")
+ .longType()
+ .defaultValue(Long.box(5000000L))
+ .withDescription("If adaptive local hash agg is enabled, "
+ + "the proportion of distinct value will be checked after reading this number of records")
Review Comment:
If adaptive local hash aggregation is enabled, this value defines how many records will be used as sampled data to calculate distinct value rate (see `TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_LIMIT_DISTINCT_RATIO`) for the local aggregate. The higher the sampling threshold, the more accurate the distinct value rate is. But as the sampling threshold increases, local aggregation is meaningless when the distinct values rate is low. The default values is 5000000.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -38,32 +42,52 @@ import org.apache.calcite.tools.RelBuilder
* aggregateBuffers should be update(e.g.: setInt) in [[BinaryRowData]]. (Hash Aggregate performs
* much better than Sort Aggregate).
*/
-class HashAggCodeGenerator(
- ctx: CodeGeneratorContext,
- builder: RelBuilder,
- aggInfoList: AggregateInfoList,
- inputType: RowType,
- outputType: RowType,
- grouping: Array[Int],
- auxGrouping: Array[Int],
- isMerge: Boolean,
- isFinal: Boolean) {
+object HashAggCodeGenerator {
- private lazy val aggInfos: Array[AggregateInfo] = aggInfoList.aggInfos
+ // It is a experimental config, will may be removed later.
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED: ConfigOption[JBoolean] =
+ key("table.exec.adaptive-local-hash-agg.enabled")
+ .booleanType()
+ .defaultValue(Boolean.box(true))
+ .withDescription("Whether to enable adaptive local hash agg")
- private lazy val functionIdentifiers: Map[AggregateFunction[_, _], String] =
- AggCodeGenHelper.getFunctionIdentifiers(aggInfos)
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLE_POINT: ConfigOption[JLong] =
+ key("table.exec.adaptive-local-hash-agg.sample-threshold")
Review Comment:
table.exec.local-hash-agg.adaptive.sampling-threshold
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProjectionCodeGenerator.scala:
##########
@@ -124,6 +129,93 @@ object ProjectionCodeGenerator {
new GeneratedProjection(className, code, ctx.references.toArray, ctx.tableConfig)
}
+ /**
+ * If adaptive hash agg takes effect, and the sample result was to suppress local hash agg. In
+ * order to ensure that the data format transmitted downstream with doing local hash agg is
+ * consistent with the data format transmitted downstream without doing local hash agg, we need to
+ * do projection for grouping function value.
+ *
+ * <p> For example, for sql statement "select a, avg(b), count(c) from T group by a", if local
+ * hash agg suppressed and a row (1, 5, "a") comes to local hash agg, we will pass (1, 5, 1, 1) to
+ * downstream.
+ */
+ def generatedAdaptiveHashAggValueProjectionCode(
Review Comment:
generateAdaptiveLocalHashAggProjectionCode
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -173,6 +214,81 @@ class HashAggCodeGenerator(
HashAggCodeGenHelper.prepareMetrics(ctx, aggregateMapTerm, if (isFinal) sorterTerm else null)
+ // Do adaptive hash aggregation
+ val outputResultForOneRowAgg = {
+ // gen code to iterating the aggregate map and output to downstream
+ val inputUnboxingCode = s"${ctx.reuseInputUnboxingCode(reuseAggBufferTerm)}"
+ val rowDataType = classOf[RowData].getCanonicalName
+ s"""
+ | // set result and output
+ |
+ | $reuseGroupKeyTerm = ($rowDataType)$currentKeyTerm;
+ | $reuseAggBufferTerm = ($rowDataType)$currentValueTerm;
+ | $inputUnboxingCode
+ | ${outputExpr.code}
+ | ${OperatorCodeGenerator.generateCollect(outputExpr.resultTerm)}
+ |
+ """.stripMargin
+ }
+ val localAggSuppressedTerm = CodeGenUtils.newName("localAggSuppressed")
+ ctx.addReusableMember(s"boolean $localAggSuppressedTerm = false;")
+ val (
+ distinctCountInCode,
+ totalCountIncCode,
+ adaptiveSamplePointCode,
+ adaptiveSuppressCode,
+ flushResultIfSuppressEnableCode) = {
Review Comment:
The field names should be adapted to the configuration names
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -173,6 +214,81 @@ class HashAggCodeGenerator(
HashAggCodeGenHelper.prepareMetrics(ctx, aggregateMapTerm, if (isFinal) sorterTerm else null)
+ // Do adaptive hash aggregation
+ val outputResultForOneRowAgg = {
+ // gen code to iterating the aggregate map and output to downstream
+ val inputUnboxingCode = s"${ctx.reuseInputUnboxingCode(reuseAggBufferTerm)}"
+ val rowDataType = classOf[RowData].getCanonicalName
+ s"""
+ | // set result and output
+ |
+ | $reuseGroupKeyTerm = ($rowDataType)$currentKeyTerm;
+ | $reuseAggBufferTerm = ($rowDataType)$currentValueTerm;
+ | $inputUnboxingCode
+ | ${outputExpr.code}
+ | ${OperatorCodeGenerator.generateCollect(outputExpr.resultTerm)}
+ |
+ """.stripMargin
+ }
+ val localAggSuppressedTerm = CodeGenUtils.newName("localAggSuppressed")
+ ctx.addReusableMember(s"boolean $localAggSuppressedTerm = false;")
+ val (
+ distinctCountInCode,
+ totalCountIncCode,
+ adaptiveSamplePointCode,
+ adaptiveSuppressCode,
+ flushResultIfSuppressEnableCode) = {
+ // from these conditions we know that it must be a distinct operation
+ if (
+ !isFinal &&
+ ctx.tableConfig.get(TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED) &&
+ canDoAdaptiveHashAgg
+ ) {
+ val adaptiveDistinctCountTerm = CodeGenUtils.newName("distinctCount")
+ val adaptiveTotalCountTerm = CodeGenUtils.newName("totalCount")
+ ctx.addReusableMember(s"long $adaptiveDistinctCountTerm = 0;")
+ ctx.addReusableMember(s"long $adaptiveTotalCountTerm = 0;")
+
+ val loggerTerm = CodeGenUtils.newName("LOG")
+ ctx.addReusableLogger(loggerTerm, className)
+
+ val samplePoint =
+ ctx.tableConfig.get(TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLE_POINT)
+ val limitDistinctRatio =
+ ctx.tableConfig.get(TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_LIMIT_DISTINCT_RATIO)
+
+ (
+ s"$adaptiveDistinctCountTerm++;",
+ s"$adaptiveTotalCountTerm++;",
+ s"""
+ |if ($adaptiveTotalCountTerm == $samplePoint) {
+ | $loggerTerm.info("Local hash agg checkpoint reached, sample point = " +
+ | $samplePoint + ", distinct = " + $adaptiveDistinctCountTerm + ", total = " +
+ | $adaptiveTotalCountTerm + ", limit distinct ratio = " + $limitDistinctRatio);
Review Comment:
ditto
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -227,7 +353,13 @@ class HashAggCodeGenerator(
|}
""".stripMargin
} else {
- s"$outputResultFromMap"
+ s"""
+ |if ($localAggSuppressedTerm) {
+ | return;
+ |} else {
+ | $outputResultFromMap
+ |}
+ |""".stripMargin
Review Comment:
s"""
|if (!$localAggSuppressedTerm) {
| $outputResultFromMap
|}
|""".stripMargin
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortAggRule.scala:
##########
@@ -108,7 +108,8 @@ class BatchPhysicalSortAggRule
auxGroupSet,
aggBufferTypes,
aggCallToAggFunction,
- isLocalHashAgg = false)
+ isLocalHashAgg = false,
+ false)
Review Comment:
supportAdaptiveLocalHashAgg = false
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/HashAggITCase.scala:
##########
@@ -17,12 +17,134 @@
*/
package org.apache.flink.table.planner.runtime.batch.sql.agg
+import org.apache.flink.api.scala._
import org.apache.flink.table.api.config.ExecutionConfigOptions
+import org.apache.flink.table.planner.codegen.agg.batch.HashAggCodeGenerator
+
+import org.junit.Test
/** AggregateITCase using HashAgg Operator. */
class HashAggITCase extends AggregateITCaseBase("HashAggregate") {
Review Comment:
I think we can use `Parameterized` test to verify the different scenarios
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProjectionCodeGenerator.scala:
##########
@@ -124,6 +129,93 @@ object ProjectionCodeGenerator {
new GeneratedProjection(className, code, ctx.references.toArray, ctx.tableConfig)
}
+ /**
+ * If adaptive hash agg takes effect, and the sample result was to suppress local hash agg. In
+ * order to ensure that the data format transmitted downstream with doing local hash agg is
+ * consistent with the data format transmitted downstream without doing local hash agg, we need to
+ * do projection for grouping function value.
+ *
+ * <p> For example, for sql statement "select a, avg(b), count(c) from T group by a", if local
+ * hash agg suppressed and a row (1, 5, "a") comes to local hash agg, we will pass (1, 5, 1, 1) to
+ * downstream.
+ */
+ def generatedAdaptiveHashAggValueProjectionCode(
+ ctx: CodeGeneratorContext,
+ inputType: RowType,
+ outClass: Class[_ <: RowData] = classOf[BinaryRowData],
+ inputTerm: String = DEFAULT_INPUT1_TERM,
+ aggInfos: Array[AggregateInfo],
+ outRecordTerm: String = DEFAULT_OUT_RECORD_TERM,
+ outRecordWriterTerm: String = DEFAULT_OUT_RECORD_WRITER_TERM): String = {
+ val fieldExprs: ArrayBuffer[GeneratedExpression] = ArrayBuffer()
+ aggInfos.map {
+ aggInfo =>
+ aggInfo.function match {
+ case _: SumAggFunction =>
+ fieldExprs += GenerateUtils.generateFieldAccess(
+ ctx,
+ inputType,
+ inputTerm,
+ aggInfo.agg.getArgList.get(0))
+ case _: MaxAggFunction | _: MinAggFunction =>
+ fieldExprs += GenerateUtils.generateFieldAccess(
+ ctx,
+ inputType,
+ inputTerm,
+ aggInfo.agg.getArgList.get(0))
Review Comment:
they can be merged into one branch
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -38,32 +42,52 @@ import org.apache.calcite.tools.RelBuilder
* aggregateBuffers should be update(e.g.: setInt) in [[BinaryRowData]]. (Hash Aggregate performs
* much better than Sort Aggregate).
*/
-class HashAggCodeGenerator(
- ctx: CodeGeneratorContext,
- builder: RelBuilder,
- aggInfoList: AggregateInfoList,
- inputType: RowType,
- outputType: RowType,
- grouping: Array[Int],
- auxGrouping: Array[Int],
- isMerge: Boolean,
- isFinal: Boolean) {
+object HashAggCodeGenerator {
- private lazy val aggInfos: Array[AggregateInfo] = aggInfoList.aggInfos
+ // It is a experimental config, will may be removed later.
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED: ConfigOption[JBoolean] =
+ key("table.exec.adaptive-local-hash-agg.enabled")
+ .booleanType()
+ .defaultValue(Boolean.box(true))
+ .withDescription("Whether to enable adaptive local hash agg")
- private lazy val functionIdentifiers: Map[AggregateFunction[_, _], String] =
- AggCodeGenHelper.getFunctionIdentifiers(aggInfos)
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLE_POINT: ConfigOption[JLong] =
+ key("table.exec.adaptive-local-hash-agg.sample-threshold")
+ .longType()
+ .defaultValue(Long.box(5000000L))
+ .withDescription("If adaptive local hash agg is enabled, "
+ + "the proportion of distinct value will be checked after reading this number of records")
- private lazy val aggBufferNames: Array[Array[String]] =
- AggCodeGenHelper.getAggBufferNames(auxGrouping, aggInfos)
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_LIMIT_DISTINCT_RATIO: ConfigOption[JDouble] =
+ key("table.exec.adaptive-local-hash-agg.limit-distinct-ratio")
Review Comment:
table.exec.local-hash-agg.adaptive.distinct-value-rate-threshold
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -38,32 +42,52 @@ import org.apache.calcite.tools.RelBuilder
* aggregateBuffers should be update(e.g.: setInt) in [[BinaryRowData]]. (Hash Aggregate performs
* much better than Sort Aggregate).
*/
-class HashAggCodeGenerator(
- ctx: CodeGeneratorContext,
- builder: RelBuilder,
- aggInfoList: AggregateInfoList,
- inputType: RowType,
- outputType: RowType,
- grouping: Array[Int],
- auxGrouping: Array[Int],
- isMerge: Boolean,
- isFinal: Boolean) {
+object HashAggCodeGenerator {
- private lazy val aggInfos: Array[AggregateInfo] = aggInfoList.aggInfos
+ // It is a experimental config, will may be removed later.
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED: ConfigOption[JBoolean] =
+ key("table.exec.adaptive-local-hash-agg.enabled")
+ .booleanType()
+ .defaultValue(Boolean.box(true))
+ .withDescription("Whether to enable adaptive local hash agg")
- private lazy val functionIdentifiers: Map[AggregateFunction[_, _], String] =
- AggCodeGenHelper.getFunctionIdentifiers(aggInfos)
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLE_POINT: ConfigOption[JLong] =
Review Comment:
TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLING_THRESHOLD
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -38,32 +42,52 @@ import org.apache.calcite.tools.RelBuilder
* aggregateBuffers should be update(e.g.: setInt) in [[BinaryRowData]]. (Hash Aggregate performs
* much better than Sort Aggregate).
*/
-class HashAggCodeGenerator(
- ctx: CodeGeneratorContext,
- builder: RelBuilder,
- aggInfoList: AggregateInfoList,
- inputType: RowType,
- outputType: RowType,
- grouping: Array[Int],
- auxGrouping: Array[Int],
- isMerge: Boolean,
- isFinal: Boolean) {
+object HashAggCodeGenerator {
- private lazy val aggInfos: Array[AggregateInfo] = aggInfoList.aggInfos
+ // It is a experimental config, will may be removed later.
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED: ConfigOption[JBoolean] =
+ key("table.exec.adaptive-local-hash-agg.enabled")
+ .booleanType()
+ .defaultValue(Boolean.box(true))
+ .withDescription("Whether to enable adaptive local hash agg")
- private lazy val functionIdentifiers: Map[AggregateFunction[_, _], String] =
- AggCodeGenHelper.getFunctionIdentifiers(aggInfos)
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLE_POINT: ConfigOption[JLong] =
+ key("table.exec.adaptive-local-hash-agg.sample-threshold")
+ .longType()
+ .defaultValue(Long.box(5000000L))
+ .withDescription("If adaptive local hash agg is enabled, "
+ + "the proportion of distinct value will be checked after reading this number of records")
- private lazy val aggBufferNames: Array[Array[String]] =
- AggCodeGenHelper.getAggBufferNames(auxGrouping, aggInfos)
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_LIMIT_DISTINCT_RATIO: ConfigOption[JDouble] =
+ key("table.exec.adaptive-local-hash-agg.limit-distinct-ratio")
+ .doubleType()
+ .defaultValue(0.5d)
+ .withDescription("If adaptive distinct is enabled, local aggregation will be suppressed " +
+ "if the ratio of distinct keys is higher than this value after sample point")
Review Comment:
The distinct value rate can be defined as the number of local aggregation result for the sampled data divided by the sampling threshold (see TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLING_THRESHOLD). If the computed result is lower than the given configuration value, the remaining input records proceed to do local aggregation, otherwise the remaining input records are subjected to simple projection which calculation cost is less than local aggregation. The default value is 0.5.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -74,6 +98,9 @@ class HashAggCodeGenerator(
// gen code to do group key projection from input
val currentKeyTerm = CodeGenUtils.newName("currentKey")
val currentKeyWriterTerm = CodeGenUtils.newName("currentKeyWriter")
+ // currentValueTerm and currentValueWriterTerm are used for value projection while canProjection is true.
Review Comment:
while `supportAdaptiveLocalHashAgg` is true.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -38,32 +42,52 @@ import org.apache.calcite.tools.RelBuilder
* aggregateBuffers should be update(e.g.: setInt) in [[BinaryRowData]]. (Hash Aggregate performs
* much better than Sort Aggregate).
*/
-class HashAggCodeGenerator(
- ctx: CodeGeneratorContext,
- builder: RelBuilder,
- aggInfoList: AggregateInfoList,
- inputType: RowType,
- outputType: RowType,
- grouping: Array[Int],
- auxGrouping: Array[Int],
- isMerge: Boolean,
- isFinal: Boolean) {
+object HashAggCodeGenerator {
- private lazy val aggInfos: Array[AggregateInfo] = aggInfoList.aggInfos
+ // It is a experimental config, will may be removed later.
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED: ConfigOption[JBoolean] =
+ key("table.exec.adaptive-local-hash-agg.enabled")
Review Comment:
table.exec.local-hash-agg.adaptive.enabled
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala:
##########
@@ -611,6 +611,71 @@ object GenerateUtils {
generateInputFieldUnboxing(ctx, inputType, inputCode, inputCode)
}
+ /**
+ * Do projection for grouping function 'max(col)' or 'min(col)' if adaptive hash agg takes effect.
+ * 'max/min(col)' will be convert to the current col value with same type.
+ */
+ def generateFiledAccessForMaxAndMin(
Review Comment:
useless code ?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/EnforceLocalSortAggRule.scala:
##########
@@ -80,7 +80,7 @@ class EnforceLocalSortAggRule
val localGrouping = agg.grouping
// create local sort
val localSort = createSort(expand, localGrouping)
- val localAgg = createLocalAgg(agg, localSort)
+ val localAgg = createLocalAgg(agg, localSort, false)
Review Comment:
use named parameter
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashAggregate.scala:
##########
@@ -161,6 +161,7 @@ class BatchPhysicalHashAggregate(
FlinkTypeFactory.toLogicalRowType(aggInputRowType),
isMerge,
true, // isFinal is always true
+ false,
Review Comment:
add "// supportAdaptiveLocalHashAgg is always false"
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/HashAggITCase.scala:
##########
@@ -17,12 +17,134 @@
*/
package org.apache.flink.table.planner.runtime.batch.sql.agg
+import org.apache.flink.api.scala._
import org.apache.flink.table.api.config.ExecutionConfigOptions
+import org.apache.flink.table.planner.codegen.agg.batch.HashAggCodeGenerator
+
+import org.junit.Test
/** AggregateITCase using HashAgg Operator. */
class HashAggITCase extends AggregateITCaseBase("HashAggregate") {
override def prepareAggOp(): Unit = {
tEnv.getConfig.set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "SortAgg")
}
+
+ @Test
+ def testAdaptiveHashAggWithHighAggregationDegree(): Unit = {
+ tEnv.getConfig.set(
+ HashAggCodeGenerator.TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED,
+ Boolean.box(true))
+ tEnv.getConfig.set(
+ HashAggCodeGenerator.TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLE_POINT,
+ Long.box(5L))
+
+ checkQuery(
+ Seq(
+ (1, 1, 1, 1),
+ (1, 1, 1, 2),
+ (1, 1, 2, 3),
+ (1, 1, 2, 2),
+ (1, 1, 3, 3),
+ (1, 2, 1, 1),
+ (1, 2, 1, 2),
+ (1, 3, 1, 1),
+ (1, 4, 1, 1),
+ (2, 1, 2, 2),
+ (2, 2, 3, 3)),
+ "SELECT f0, f1, sum(f2), max(f3), count(f3), count(*) FROM TableName GROUP BY f0, f1",
Review Comment:
add `min`
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/EnforceLocalHashAggRule.scala:
##########
@@ -69,7 +69,7 @@ class EnforceLocalHashAggRule
val agg: BatchPhysicalHashAggregate = call.rel(0)
val expand: BatchPhysicalExpand = call.rel(2)
- val localAgg = createLocalAgg(agg, expand)
+ val localAgg = createLocalAgg(agg, expand, true)
Review Comment:
why `supportAdaptiveLocalHashAgg ` is always true
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/HashAggITCase.scala:
##########
@@ -17,12 +17,134 @@
*/
package org.apache.flink.table.planner.runtime.batch.sql.agg
+import org.apache.flink.api.scala._
import org.apache.flink.table.api.config.ExecutionConfigOptions
+import org.apache.flink.table.planner.codegen.agg.batch.HashAggCodeGenerator
+
+import org.junit.Test
/** AggregateITCase using HashAgg Operator. */
class HashAggITCase extends AggregateITCaseBase("HashAggregate") {
override def prepareAggOp(): Unit = {
tEnv.getConfig.set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "SortAgg")
}
+
+ @Test
+ def testAdaptiveHashAggWithHighAggregationDegree(): Unit = {
+ tEnv.getConfig.set(
+ HashAggCodeGenerator.TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED,
+ Boolean.box(true))
+ tEnv.getConfig.set(
+ HashAggCodeGenerator.TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLE_POINT,
+ Long.box(5L))
+
+ checkQuery(
+ Seq(
+ (1, 1, 1, 1),
+ (1, 1, 1, 2),
+ (1, 1, 2, 3),
+ (1, 1, 2, 2),
+ (1, 1, 3, 3),
+ (1, 2, 1, 1),
+ (1, 2, 1, 2),
+ (1, 3, 1, 1),
+ (1, 4, 1, 1),
+ (2, 1, 2, 2),
+ (2, 2, 3, 3)),
+ "SELECT f0, f1, sum(f2), max(f3), count(f3), count(*) FROM TableName GROUP BY f0, f1",
+ Seq(
+ (1, 1, 9, 3, 5, 5),
+ (1, 2, 2, 2, 2, 2),
+ (1, 3, 1, 1, 1, 1),
+ (1, 4, 1, 1, 1, 1),
+ (2, 1, 2, 2, 1, 1),
+ (2, 2, 3, 3, 1, 1))
+ )
+ }
+
+ @Test
+ def testAdaptiveHashAggWithLowAggregationDegree(): Unit = {
+ tEnv.getConfig.set(
+ HashAggCodeGenerator.TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED,
+ Boolean.box(true))
+ tEnv.getConfig.set(
+ HashAggCodeGenerator.TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLE_POINT,
+ Long.box(5L))
+
+ checkQuery(
+ Seq(
+ (1, 1, 1, 1),
+ (1, 1, 1, 2),
+ (1, 2, 2, 3),
+ (1, 3, 2, 2),
+ (1, 4, 3, 3),
+ (1, 5, 1, 1),
+ (2, 1, 1, 2),
+ (2, 2, 1, 1),
+ (2, 3, 1, 1),
+ (2, 3, 2, 2),
+ (2, 3, 3, 3)),
+ "SELECT f0, f1, sum(f2), max(f3) FROM TableName GROUP BY f0, f1",
+ Seq(
+ (1, 1, 2, 2),
+ (1, 2, 2, 3),
+ (1, 3, 2, 2),
+ (1, 4, 3, 3),
+ (1, 5, 1, 1),
+ (2, 1, 1, 2),
+ (2, 2, 1, 1),
+ (2, 3, 6, 3))
+ )
+ }
+
+ @Test
+ def testAdaptiveHashAggWithRowLessThanSamplePoint(): Unit = {
+ tEnv.getConfig.set(
+ HashAggCodeGenerator.TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED,
+ Boolean.box(true))
+ tEnv.getConfig.set(
+ HashAggCodeGenerator.TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLE_POINT,
+ Long.box(5L))
+
+ checkQuery(
+ Seq((1, 1, 1, 1), (1, 1, 1, 2), (1, 2, 2, 3)),
+ "SELECT f0, f1, sum(f2), max(f3) FROM TableName GROUP BY f0, f1",
+ Seq((1, 1, 2, 2), (1, 2, 2, 3))
+ )
+ }
+
+ @Test
+ def testAdaptiveHashAggWithDifferentSumColType(): Unit = {
Review Comment:
this can be merged into one case
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] godfreyhe closed pull request #21586: [FLINK-30542][table-runtime] Introduce adaptive local hash aggregate to adaptively determine whether local hash aggregate is required at runtime
Posted by "godfreyhe (via GitHub)" <gi...@apache.org>.
godfreyhe closed pull request #21586: [FLINK-30542][table-runtime] Introduce adaptive local hash aggregate to adaptively determine whether local hash aggregate is required at runtime
URL: https://github.com/apache/flink/pull/21586
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on pull request #21586: [FLINK-30542][table-runtime] Introduce adaptive hash aggregate to adaptively determine whether local hash aggregate is required at runtime
Posted by GitBox <gi...@apache.org>.
swuferhong commented on PR #21586:
URL: https://github.com/apache/flink/pull/21586#issuecomment-1383049271
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on pull request #21586: [FLINK-30542][table-runtime] Introduce adaptive local hash aggregate to adaptively determine whether local hash aggregate is required at runtime
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on PR #21586:
URL: https://github.com/apache/flink/pull/21586#issuecomment-1408240235
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #21586: [FLINK-30542][table-runtime] Introduce adaptive hash aggregate to adaptively determine whether local hash aggregate is required at runtime
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #21586:
URL: https://github.com/apache/flink/pull/21586#discussion_r1066775534
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala:
##########
@@ -610,6 +610,58 @@ object GenerateUtils {
generateInputFieldUnboxing(ctx, inputType, inputCode, inputCode)
}
+ def generateFieldAccessForCountCol(
+ ctx: CodeGeneratorContext,
+ inputType: LogicalType,
+ inputTerm: String,
+ index: Int): GeneratedExpression = inputType.getTypeRoot match {
+ // ordered by type root definition
+ case ROW | STRUCTURED_TYPE =>
Review Comment:
> Do we support count `row` or `structured` type now?
Done!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lsyldliu commented on a diff in pull request #21586: [FLINK-30542][table-runtime] Introduce adaptive hash aggregate to adaptively determine whether local hash aggregate is required at runtime
Posted by GitBox <gi...@apache.org>.
lsyldliu commented on code in PR #21586:
URL: https://github.com/apache/flink/pull/21586#discussion_r1065306202
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -202,11 +299,43 @@ class HashAggCodeGenerator(
| $dealWithAggHashMapOOM
| }
|}
+ |
+ |$totalCountIncCode
+ |$adaptiveSamplePointCode
+ |
| // aggregate buffer fields access
|${ctx.reuseInputUnboxingCode(currentAggBufferTerm)}
| // do aggregate and update agg buffer
|${aggregate.code}
+ | // flush result form map if suppress is enable.
+ |$flushResultIfSuppressEnableCode
|""".stripMargin.trim
+ } else {
+ s"""
Review Comment:
We don't need this repeated code, for local and global agg, the main code is reusable. For example, the `adaptiveSuppressCode ` for global agg is "", but for local agg is the needed code.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala:
##########
@@ -610,6 +610,58 @@ object GenerateUtils {
generateInputFieldUnboxing(ctx, inputType, inputCode, inputCode)
}
+ def generateFieldAccessForCountCol(
Review Comment:
Add some annotation for these two methods?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashAggregate.scala:
##########
@@ -53,7 +53,8 @@ class BatchPhysicalHashAggregate(
grouping: Array[Int],
auxGrouping: Array[Int],
aggCallToAggFunction: Seq[(AggregateCall, UserDefinedFunction)],
- isMerge: Boolean)
+ isMerge: Boolean,
+ canProjection: Boolean)
Review Comment:
For global hash agg, we don't need this variable, it should always be false.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala:
##########
@@ -610,6 +610,58 @@ object GenerateUtils {
generateInputFieldUnboxing(ctx, inputType, inputCode, inputCode)
}
+ def generateFieldAccessForCountCol(
+ ctx: CodeGeneratorContext,
+ inputType: LogicalType,
+ inputTerm: String,
+ index: Int): GeneratedExpression = inputType.getTypeRoot match {
+ // ordered by type root definition
+ case ROW | STRUCTURED_TYPE =>
Review Comment:
Do we support count `row` or `structured` type now?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #21586: [FLINK-30542][table-runtime] Introduce adaptive hash aggregate to adaptively determine whether local hash aggregate is required at runtime
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #21586:
URL: https://github.com/apache/flink/pull/21586#discussion_r1066761596
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -202,11 +299,43 @@ class HashAggCodeGenerator(
| $dealWithAggHashMapOOM
| }
|}
+ |
+ |$totalCountIncCode
+ |$adaptiveSamplePointCode
+ |
| // aggregate buffer fields access
|${ctx.reuseInputUnboxingCode(currentAggBufferTerm)}
| // do aggregate and update agg buffer
|${aggregate.code}
+ | // flush result form map if suppress is enable.
+ |$flushResultIfSuppressEnableCode
|""".stripMargin.trim
+ } else {
+ s"""
Review Comment:
> We don't need this repeated code, for local and global agg, the main code is reusable. For example, the `adaptiveSuppressCode ` for global agg is "", but for local agg is the needed code.
Done!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lsyldliu commented on a diff in pull request #21586: [FLINK-30542][table-runtime] Introduce adaptive hash aggregate to adaptively determine whether local hash aggregate is required at runtime
Posted by GitBox <gi...@apache.org>.
lsyldliu commented on code in PR #21586:
URL: https://github.com/apache/flink/pull/21586#discussion_r1073489490
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -38,32 +42,52 @@ import org.apache.calcite.tools.RelBuilder
* aggregateBuffers should be update(e.g.: setInt) in [[BinaryRowData]]. (Hash Aggregate performs
* much better than Sort Aggregate).
*/
-class HashAggCodeGenerator(
- ctx: CodeGeneratorContext,
- builder: RelBuilder,
- aggInfoList: AggregateInfoList,
- inputType: RowType,
- outputType: RowType,
- grouping: Array[Int],
- auxGrouping: Array[Int],
- isMerge: Boolean,
- isFinal: Boolean) {
+object HashAggCodeGenerator {
- private lazy val aggInfos: Array[AggregateInfo] = aggInfoList.aggInfos
+ // It is a experimental config, will may be removed later.
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED: ConfigOption[JBoolean] =
+ key("table.exec.adaptive-local-hash-agg.enabled")
Review Comment:
Referring to the [FLIP-283](https://cwiki.apache.org/confluence/display/FLINK/FLIP-283%3A+Use+adaptive+batch+scheduler+as+default+scheduler+for+batch+jobs), I think it would better to rename the option to `table.exec.adaptive.local-hash-agg.enabled`. cc @godfreyhe
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -38,32 +42,52 @@ import org.apache.calcite.tools.RelBuilder
* aggregateBuffers should be update(e.g.: setInt) in [[BinaryRowData]]. (Hash Aggregate performs
* much better than Sort Aggregate).
*/
-class HashAggCodeGenerator(
- ctx: CodeGeneratorContext,
- builder: RelBuilder,
- aggInfoList: AggregateInfoList,
- inputType: RowType,
- outputType: RowType,
- grouping: Array[Int],
- auxGrouping: Array[Int],
- isMerge: Boolean,
- isFinal: Boolean) {
+object HashAggCodeGenerator {
- private lazy val aggInfos: Array[AggregateInfo] = aggInfoList.aggInfos
+ // It is a experimental config, will may be removed later.
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED: ConfigOption[JBoolean] =
+ key("table.exec.adaptive-local-hash-agg.enabled")
+ .booleanType()
+ .defaultValue(Boolean.box(true))
+ .withDescription("Whether to enable adaptive local hash agg")
- private lazy val functionIdentifiers: Map[AggregateFunction[_, _], String] =
- AggCodeGenHelper.getFunctionIdentifiers(aggInfos)
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLE_POINT: ConfigOption[JLong] =
+ key("table.exec.adaptive-local-hash-agg.sample-threshold")
+ .longType()
+ .defaultValue(Long.box(5000000L))
+ .withDescription("If adaptive local hash agg is enabled, "
+ + "the proportion of distinct value will be checked after reading this number of records")
Review Comment:
```suggestion
+ "the proportion of distinct value will be checked after reading this number of records.")
```
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/AggregateUtil.scala:
##########
@@ -926,6 +926,19 @@ object AggregateUtil extends Enumeration {
aggInfos.isEmpty || supportMerge
}
+ /** Return true if all aggregates can be projection in adaptive hash agg. False otherwise. */
+ def doAllAggSupportProjection(aggCalls: Seq[AggregateCall]): Boolean = {
Review Comment:
`doAllSuppportAdaptiveHashAgg`?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -173,6 +214,81 @@ class HashAggCodeGenerator(
HashAggCodeGenHelper.prepareMetrics(ctx, aggregateMapTerm, if (isFinal) sorterTerm else null)
+ // Do adaptive hash aggregation
+ val outputResultForOneRowAgg = {
+ // gen code to iterating the aggregate map and output to downstream
+ val inputUnboxingCode = s"${ctx.reuseInputUnboxingCode(reuseAggBufferTerm)}"
+ val rowDataType = classOf[RowData].getCanonicalName
Review Comment:
```
s"""
| // set result and output
| $reuseGroupKeyTerm = ($ROW_DATA)$currentKeyTerm;
| $reuseAggBufferTerm = ($ROW_DATA)$currentValueTerm;
| $inputUnboxingCode
| ${outputExpr.code}
| ${OperatorCodeGenerator.generateCollect(outputExpr.resultTerm)}
|
""".stripMargin
```
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/AggregateUtil.scala:
##########
@@ -926,6 +926,19 @@ object AggregateUtil extends Enumeration {
aggInfos.isEmpty || supportMerge
}
+ /** Return true if all aggregates can be projection in adaptive hash agg. False otherwise. */
Review Comment:
```suggestion
/** Return true if all aggregates can be projected for adaptive local hash aggregate. False otherwise. */
```
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -38,32 +42,52 @@ import org.apache.calcite.tools.RelBuilder
* aggregateBuffers should be update(e.g.: setInt) in [[BinaryRowData]]. (Hash Aggregate performs
* much better than Sort Aggregate).
*/
-class HashAggCodeGenerator(
- ctx: CodeGeneratorContext,
- builder: RelBuilder,
- aggInfoList: AggregateInfoList,
- inputType: RowType,
- outputType: RowType,
- grouping: Array[Int],
- auxGrouping: Array[Int],
- isMerge: Boolean,
- isFinal: Boolean) {
+object HashAggCodeGenerator {
- private lazy val aggInfos: Array[AggregateInfo] = aggInfoList.aggInfos
+ // It is a experimental config, will may be removed later.
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED: ConfigOption[JBoolean] =
+ key("table.exec.adaptive-local-hash-agg.enabled")
+ .booleanType()
+ .defaultValue(Boolean.box(true))
+ .withDescription("Whether to enable adaptive local hash agg")
- private lazy val functionIdentifiers: Map[AggregateFunction[_, _], String] =
- AggCodeGenHelper.getFunctionIdentifiers(aggInfos)
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLE_POINT: ConfigOption[JLong] =
Review Comment:
```suggestion
val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLE_THRESHOLD: ConfigOption[JLong] =
```
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -173,6 +214,81 @@ class HashAggCodeGenerator(
HashAggCodeGenHelper.prepareMetrics(ctx, aggregateMapTerm, if (isFinal) sorterTerm else null)
+ // Do adaptive hash aggregation
+ val outputResultForOneRowAgg = {
+ // gen code to iterating the aggregate map and output to downstream
+ val inputUnboxingCode = s"${ctx.reuseInputUnboxingCode(reuseAggBufferTerm)}"
+ val rowDataType = classOf[RowData].getCanonicalName
+ s"""
+ | // set result and output
+ |
+ | $reuseGroupKeyTerm = ($rowDataType)$currentKeyTerm;
+ | $reuseAggBufferTerm = ($rowDataType)$currentValueTerm;
+ | $inputUnboxingCode
+ | ${outputExpr.code}
+ | ${OperatorCodeGenerator.generateCollect(outputExpr.resultTerm)}
+ |
+ """.stripMargin
+ }
+ val localAggSuppressedTerm = CodeGenUtils.newName("localAggSuppressed")
+ ctx.addReusableMember(s"boolean $localAggSuppressedTerm = false;")
+ val (
+ distinctCountInCode,
+ totalCountIncCode,
+ adaptiveSamplePointCode,
+ adaptiveSuppressCode,
+ flushResultIfSuppressEnableCode) = {
+ // from these conditions we know that it must be a distinct operation
+ if (
+ !isFinal &&
+ ctx.tableConfig.get(TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED) &&
+ canDoAdaptiveHashAgg
+ ) {
+ val adaptiveDistinctCountTerm = CodeGenUtils.newName("distinctCount")
+ val adaptiveTotalCountTerm = CodeGenUtils.newName("totalCount")
+ ctx.addReusableMember(s"long $adaptiveDistinctCountTerm = 0;")
Review Comment:
```suggestion
ctx.addReusableMember(s"private long transient $adaptiveDistinctCountTerm = 0;")
```
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -173,6 +214,81 @@ class HashAggCodeGenerator(
HashAggCodeGenHelper.prepareMetrics(ctx, aggregateMapTerm, if (isFinal) sorterTerm else null)
+ // Do adaptive hash aggregation
+ val outputResultForOneRowAgg = {
+ // gen code to iterating the aggregate map and output to downstream
+ val inputUnboxingCode = s"${ctx.reuseInputUnboxingCode(reuseAggBufferTerm)}"
+ val rowDataType = classOf[RowData].getCanonicalName
+ s"""
+ | // set result and output
+ |
+ | $reuseGroupKeyTerm = ($rowDataType)$currentKeyTerm;
+ | $reuseAggBufferTerm = ($rowDataType)$currentValueTerm;
+ | $inputUnboxingCode
+ | ${outputExpr.code}
+ | ${OperatorCodeGenerator.generateCollect(outputExpr.resultTerm)}
+ |
+ """.stripMargin
+ }
+ val localAggSuppressedTerm = CodeGenUtils.newName("localAggSuppressed")
+ ctx.addReusableMember(s"boolean $localAggSuppressedTerm = false;")
+ val (
+ distinctCountInCode,
+ totalCountIncCode,
+ adaptiveSamplePointCode,
+ adaptiveSuppressCode,
+ flushResultIfSuppressEnableCode) = {
+ // from these conditions we know that it must be a distinct operation
+ if (
+ !isFinal &&
+ ctx.tableConfig.get(TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED) &&
+ canDoAdaptiveHashAgg
+ ) {
+ val adaptiveDistinctCountTerm = CodeGenUtils.newName("distinctCount")
+ val adaptiveTotalCountTerm = CodeGenUtils.newName("totalCount")
+ ctx.addReusableMember(s"long $adaptiveDistinctCountTerm = 0;")
+ ctx.addReusableMember(s"long $adaptiveTotalCountTerm = 0;")
+
+ val loggerTerm = CodeGenUtils.newName("LOG")
+ ctx.addReusableLogger(loggerTerm, className)
+
+ val samplePoint =
+ ctx.tableConfig.get(TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLE_POINT)
+ val limitDistinctRatio =
+ ctx.tableConfig.get(TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_LIMIT_DISTINCT_RATIO)
+
+ (
+ s"$adaptiveDistinctCountTerm++;",
+ s"$adaptiveTotalCountTerm++;",
+ s"""
+ |if ($adaptiveTotalCountTerm == $samplePoint) {
+ | $loggerTerm.info("Local hash agg checkpoint reached, sample point = " +
+ | $samplePoint + ", distinct = " + $adaptiveDistinctCountTerm + ", total = " +
+ | $adaptiveTotalCountTerm + ", limit distinct ratio = " + $limitDistinctRatio);
+ | if ((double) $adaptiveDistinctCountTerm / $adaptiveTotalCountTerm > $limitDistinctRatio) {
+ | $loggerTerm.info("Local hash agg suppressed");
Review Comment:
```suggestion
| $loggerTerm.info("Local hash agg is suppressed");
```
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -173,6 +214,81 @@ class HashAggCodeGenerator(
HashAggCodeGenHelper.prepareMetrics(ctx, aggregateMapTerm, if (isFinal) sorterTerm else null)
+ // Do adaptive hash aggregation
+ val outputResultForOneRowAgg = {
+ // gen code to iterating the aggregate map and output to downstream
+ val inputUnboxingCode = s"${ctx.reuseInputUnboxingCode(reuseAggBufferTerm)}"
+ val rowDataType = classOf[RowData].getCanonicalName
+ s"""
+ | // set result and output
+ |
+ | $reuseGroupKeyTerm = ($rowDataType)$currentKeyTerm;
+ | $reuseAggBufferTerm = ($rowDataType)$currentValueTerm;
+ | $inputUnboxingCode
+ | ${outputExpr.code}
+ | ${OperatorCodeGenerator.generateCollect(outputExpr.resultTerm)}
+ |
+ """.stripMargin
+ }
+ val localAggSuppressedTerm = CodeGenUtils.newName("localAggSuppressed")
+ ctx.addReusableMember(s"boolean $localAggSuppressedTerm = false;")
+ val (
+ distinctCountInCode,
+ totalCountIncCode,
+ adaptiveSamplePointCode,
+ adaptiveSuppressCode,
Review Comment:
`adaptiveHashAggCode` maybe better.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProjectionCodeGenerator.scala:
##########
@@ -124,6 +129,93 @@ object ProjectionCodeGenerator {
new GeneratedProjection(className, code, ctx.references.toArray, ctx.tableConfig)
}
+ /**
+ * If adaptive hash agg takes effect, and the sample result was to suppress local hash agg. In
+ * order to ensure that the data format transmitted downstream with doing local hash agg is
+ * consistent with the data format transmitted downstream without doing local hash agg, we need to
+ * do projection for grouping function value.
+ *
+ * <p> For example, for sql statement "select a, avg(b), count(c) from T group by a", if local
+ * hash agg suppressed and a row (1, 5, "a") comes to local hash agg, we will pass (1, 5, 1, 1) to
+ * downstream.
+ */
+ def generatedAdaptiveHashAggValueProjectionCode(
+ ctx: CodeGeneratorContext,
+ inputType: RowType,
+ outClass: Class[_ <: RowData] = classOf[BinaryRowData],
+ inputTerm: String = DEFAULT_INPUT1_TERM,
+ aggInfos: Array[AggregateInfo],
+ outRecordTerm: String = DEFAULT_OUT_RECORD_TERM,
+ outRecordWriterTerm: String = DEFAULT_OUT_RECORD_WRITER_TERM): String = {
+ val fieldExprs: ArrayBuffer[GeneratedExpression] = ArrayBuffer()
+ aggInfos.map {
+ aggInfo =>
+ aggInfo.function match {
+ case _: SumAggFunction =>
+ fieldExprs += GenerateUtils.generateFieldAccess(
+ ctx,
+ inputType,
+ inputTerm,
+ aggInfo.agg.getArgList.get(0))
+ case _: MaxAggFunction | _: MinAggFunction =>
+ fieldExprs += GenerateUtils.generateFieldAccess(
+ ctx,
+ inputType,
+ inputTerm,
+ aggInfo.agg.getArgList.get(0))
+ case _: AvgAggFunction =>
+ fieldExprs += GenerateUtils.generateFieldAccess(
+ ctx,
+ inputType,
+ inputTerm,
+ aggInfo.agg.getArgList.get(0))
+ fieldExprs += GenerateUtils.generateFieldAccessForCountSpecificCol(
+ ctx,
+ inputTerm,
+ aggInfo.agg.getArgList.get(0))
+ case _: CountAggFunction =>
+ fieldExprs += GenerateUtils.generateFieldAccessForCountSpecificCol(
+ ctx,
+ inputTerm,
+ aggInfo.agg.getArgList.get(0))
+ case _: Count1AggFunction =>
+ fieldExprs += GenerateUtils.generateFieldAccessForCountOne(ctx)
+ }
+ }
+
+ val binaryRowWriter = CodeGenUtils.className[BinaryRowWriter]
+ val typeTerm = outClass.getCanonicalName
+ ctx.addReusableMember(s"$typeTerm $outRecordTerm= new $typeTerm(${fieldExprs.size});")
+ ctx.addReusableMember(
Review Comment:
ditto
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProjectionCodeGenerator.scala:
##########
@@ -124,6 +129,93 @@ object ProjectionCodeGenerator {
new GeneratedProjection(className, code, ctx.references.toArray, ctx.tableConfig)
}
+ /**
+ * If adaptive hash agg takes effect, and the sample result was to suppress local hash agg. In
+ * order to ensure that the data format transmitted downstream with doing local hash agg is
+ * consistent with the data format transmitted downstream without doing local hash agg, we need to
+ * do projection for grouping function value.
+ *
+ * <p> For example, for sql statement "select a, avg(b), count(c) from T group by a", if local
+ * hash agg suppressed and a row (1, 5, "a") comes to local hash agg, we will pass (1, 5, 1, 1) to
+ * downstream.
+ */
+ def generatedAdaptiveHashAggValueProjectionCode(
+ ctx: CodeGeneratorContext,
+ inputType: RowType,
+ outClass: Class[_ <: RowData] = classOf[BinaryRowData],
+ inputTerm: String = DEFAULT_INPUT1_TERM,
+ aggInfos: Array[AggregateInfo],
+ outRecordTerm: String = DEFAULT_OUT_RECORD_TERM,
+ outRecordWriterTerm: String = DEFAULT_OUT_RECORD_WRITER_TERM): String = {
+ val fieldExprs: ArrayBuffer[GeneratedExpression] = ArrayBuffer()
+ aggInfos.map {
+ aggInfo =>
+ aggInfo.function match {
+ case _: SumAggFunction =>
+ fieldExprs += GenerateUtils.generateFieldAccess(
Review Comment:
Assuming the sum field type is bigint, if some element of the field value is null, we emit the result to downstream global agg is default value -1L, whether the final is corrected?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -173,6 +214,81 @@ class HashAggCodeGenerator(
HashAggCodeGenHelper.prepareMetrics(ctx, aggregateMapTerm, if (isFinal) sorterTerm else null)
+ // Do adaptive hash aggregation
+ val outputResultForOneRowAgg = {
+ // gen code to iterating the aggregate map and output to downstream
+ val inputUnboxingCode = s"${ctx.reuseInputUnboxingCode(reuseAggBufferTerm)}"
+ val rowDataType = classOf[RowData].getCanonicalName
+ s"""
+ | // set result and output
+ |
+ | $reuseGroupKeyTerm = ($rowDataType)$currentKeyTerm;
+ | $reuseAggBufferTerm = ($rowDataType)$currentValueTerm;
+ | $inputUnboxingCode
+ | ${outputExpr.code}
+ | ${OperatorCodeGenerator.generateCollect(outputExpr.resultTerm)}
+ |
+ """.stripMargin
+ }
+ val localAggSuppressedTerm = CodeGenUtils.newName("localAggSuppressed")
+ ctx.addReusableMember(s"boolean $localAggSuppressedTerm = false;")
+ val (
+ distinctCountInCode,
+ totalCountIncCode,
+ adaptiveSamplePointCode,
+ adaptiveSuppressCode,
+ flushResultIfSuppressEnableCode) = {
+ // from these conditions we know that it must be a distinct operation
+ if (
+ !isFinal &&
+ ctx.tableConfig.get(TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED) &&
+ canDoAdaptiveHashAgg
+ ) {
+ val adaptiveDistinctCountTerm = CodeGenUtils.newName("distinctCount")
+ val adaptiveTotalCountTerm = CodeGenUtils.newName("totalCount")
+ ctx.addReusableMember(s"long $adaptiveDistinctCountTerm = 0;")
+ ctx.addReusableMember(s"long $adaptiveTotalCountTerm = 0;")
+
+ val loggerTerm = CodeGenUtils.newName("LOG")
Review Comment:
We don't need to redefine this LOG variable again, the line 95 has defined it.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -173,6 +214,81 @@ class HashAggCodeGenerator(
HashAggCodeGenHelper.prepareMetrics(ctx, aggregateMapTerm, if (isFinal) sorterTerm else null)
+ // Do adaptive hash aggregation
+ val outputResultForOneRowAgg = {
+ // gen code to iterating the aggregate map and output to downstream
+ val inputUnboxingCode = s"${ctx.reuseInputUnboxingCode(reuseAggBufferTerm)}"
+ val rowDataType = classOf[RowData].getCanonicalName
+ s"""
+ | // set result and output
+ |
+ | $reuseGroupKeyTerm = ($rowDataType)$currentKeyTerm;
+ | $reuseAggBufferTerm = ($rowDataType)$currentValueTerm;
+ | $inputUnboxingCode
+ | ${outputExpr.code}
+ | ${OperatorCodeGenerator.generateCollect(outputExpr.resultTerm)}
+ |
+ """.stripMargin
+ }
+ val localAggSuppressedTerm = CodeGenUtils.newName("localAggSuppressed")
+ ctx.addReusableMember(s"boolean $localAggSuppressedTerm = false;")
Review Comment:
```suggestion
ctx.addReusableMember(s"private boolean transient $localAggSuppressedTerm = false;")
```
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -38,32 +42,52 @@ import org.apache.calcite.tools.RelBuilder
* aggregateBuffers should be update(e.g.: setInt) in [[BinaryRowData]]. (Hash Aggregate performs
* much better than Sort Aggregate).
*/
-class HashAggCodeGenerator(
- ctx: CodeGeneratorContext,
- builder: RelBuilder,
- aggInfoList: AggregateInfoList,
- inputType: RowType,
- outputType: RowType,
- grouping: Array[Int],
- auxGrouping: Array[Int],
- isMerge: Boolean,
- isFinal: Boolean) {
+object HashAggCodeGenerator {
- private lazy val aggInfos: Array[AggregateInfo] = aggInfoList.aggInfos
+ // It is a experimental config, will may be removed later.
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED: ConfigOption[JBoolean] =
+ key("table.exec.adaptive-local-hash-agg.enabled")
+ .booleanType()
+ .defaultValue(Boolean.box(true))
+ .withDescription("Whether to enable adaptive local hash agg")
- private lazy val functionIdentifiers: Map[AggregateFunction[_, _], String] =
- AggCodeGenHelper.getFunctionIdentifiers(aggInfos)
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLE_POINT: ConfigOption[JLong] =
+ key("table.exec.adaptive-local-hash-agg.sample-threshold")
+ .longType()
+ .defaultValue(Long.box(5000000L))
+ .withDescription("If adaptive local hash agg is enabled, "
+ + "the proportion of distinct value will be checked after reading this number of records")
- private lazy val aggBufferNames: Array[Array[String]] =
- AggCodeGenHelper.getAggBufferNames(auxGrouping, aggInfos)
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_LIMIT_DISTINCT_RATIO: ConfigOption[JDouble] =
+ key("table.exec.adaptive-local-hash-agg.limit-distinct-ratio")
Review Comment:
How do we understand the `limit-distinct-ration`? what does it mean?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecHashAggregate.java:
##########
@@ -58,6 +58,7 @@ public class BatchExecHashAggregate extends ExecNodeBase<RowData>
private final RowType aggInputRowType;
private final boolean isMerge;
private final boolean isFinal;
+ private final boolean canDoAdaptiveHashAgg;
Review Comment:
`supportAdaptiveHashAgg`?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -74,6 +98,9 @@ class HashAggCodeGenerator(
// gen code to do group key projection from input
val currentKeyTerm = CodeGenUtils.newName("currentKey")
val currentKeyWriterTerm = CodeGenUtils.newName("currentKeyWriter")
+ // currentValueTerm and currentValueWriterTerm are used for value projection while canProjection is true.
Review Comment:
```suggestion
// currentValueTerm and currentValueWriterTerm are used for value projection while canDoAdaptiveHashAgg is true.
```
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -38,32 +42,52 @@ import org.apache.calcite.tools.RelBuilder
* aggregateBuffers should be update(e.g.: setInt) in [[BinaryRowData]]. (Hash Aggregate performs
* much better than Sort Aggregate).
*/
-class HashAggCodeGenerator(
- ctx: CodeGeneratorContext,
- builder: RelBuilder,
- aggInfoList: AggregateInfoList,
- inputType: RowType,
- outputType: RowType,
- grouping: Array[Int],
- auxGrouping: Array[Int],
- isMerge: Boolean,
- isFinal: Boolean) {
+object HashAggCodeGenerator {
- private lazy val aggInfos: Array[AggregateInfo] = aggInfoList.aggInfos
+ // It is a experimental config, will may be removed later.
+ @Experimental
+ val TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED: ConfigOption[JBoolean] =
+ key("table.exec.adaptive-local-hash-agg.enabled")
+ .booleanType()
+ .defaultValue(Boolean.box(true))
+ .withDescription("Whether to enable adaptive local hash agg")
Review Comment:
```suggestion
.withDescription("Whether to enable adaptive local hash aggregation.")
```
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -173,6 +214,81 @@ class HashAggCodeGenerator(
HashAggCodeGenHelper.prepareMetrics(ctx, aggregateMapTerm, if (isFinal) sorterTerm else null)
+ // Do adaptive hash aggregation
+ val outputResultForOneRowAgg = {
+ // gen code to iterating the aggregate map and output to downstream
+ val inputUnboxingCode = s"${ctx.reuseInputUnboxingCode(reuseAggBufferTerm)}"
+ val rowDataType = classOf[RowData].getCanonicalName
+ s"""
+ | // set result and output
+ |
+ | $reuseGroupKeyTerm = ($rowDataType)$currentKeyTerm;
+ | $reuseAggBufferTerm = ($rowDataType)$currentValueTerm;
+ | $inputUnboxingCode
+ | ${outputExpr.code}
+ | ${OperatorCodeGenerator.generateCollect(outputExpr.resultTerm)}
+ |
+ """.stripMargin
+ }
+ val localAggSuppressedTerm = CodeGenUtils.newName("localAggSuppressed")
Review Comment:
`isLocalAggSuppressed`?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -173,6 +214,81 @@ class HashAggCodeGenerator(
HashAggCodeGenHelper.prepareMetrics(ctx, aggregateMapTerm, if (isFinal) sorterTerm else null)
+ // Do adaptive hash aggregation
+ val outputResultForOneRowAgg = {
+ // gen code to iterating the aggregate map and output to downstream
+ val inputUnboxingCode = s"${ctx.reuseInputUnboxingCode(reuseAggBufferTerm)}"
+ val rowDataType = classOf[RowData].getCanonicalName
+ s"""
+ | // set result and output
+ |
+ | $reuseGroupKeyTerm = ($rowDataType)$currentKeyTerm;
+ | $reuseAggBufferTerm = ($rowDataType)$currentValueTerm;
+ | $inputUnboxingCode
+ | ${outputExpr.code}
+ | ${OperatorCodeGenerator.generateCollect(outputExpr.resultTerm)}
+ |
+ """.stripMargin
+ }
+ val localAggSuppressedTerm = CodeGenUtils.newName("localAggSuppressed")
+ ctx.addReusableMember(s"boolean $localAggSuppressedTerm = false;")
+ val (
+ distinctCountInCode,
+ totalCountIncCode,
+ adaptiveSamplePointCode,
+ adaptiveSuppressCode,
+ flushResultIfSuppressEnableCode) = {
+ // from these conditions we know that it must be a distinct operation
+ if (
+ !isFinal &&
+ ctx.tableConfig.get(TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED) &&
+ canDoAdaptiveHashAgg
+ ) {
+ val adaptiveDistinctCountTerm = CodeGenUtils.newName("distinctCount")
+ val adaptiveTotalCountTerm = CodeGenUtils.newName("totalCount")
+ ctx.addReusableMember(s"long $adaptiveDistinctCountTerm = 0;")
+ ctx.addReusableMember(s"long $adaptiveTotalCountTerm = 0;")
Review Comment:
ditto
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -173,6 +214,81 @@ class HashAggCodeGenerator(
HashAggCodeGenHelper.prepareMetrics(ctx, aggregateMapTerm, if (isFinal) sorterTerm else null)
+ // Do adaptive hash aggregation
+ val outputResultForOneRowAgg = {
+ // gen code to iterating the aggregate map and output to downstream
+ val inputUnboxingCode = s"${ctx.reuseInputUnboxingCode(reuseAggBufferTerm)}"
+ val rowDataType = classOf[RowData].getCanonicalName
+ s"""
+ | // set result and output
+ |
+ | $reuseGroupKeyTerm = ($rowDataType)$currentKeyTerm;
+ | $reuseAggBufferTerm = ($rowDataType)$currentValueTerm;
+ | $inputUnboxingCode
+ | ${outputExpr.code}
+ | ${OperatorCodeGenerator.generateCollect(outputExpr.resultTerm)}
+ |
+ """.stripMargin
+ }
+ val localAggSuppressedTerm = CodeGenUtils.newName("localAggSuppressed")
+ ctx.addReusableMember(s"boolean $localAggSuppressedTerm = false;")
+ val (
+ distinctCountInCode,
+ totalCountIncCode,
+ adaptiveSamplePointCode,
+ adaptiveSuppressCode,
+ flushResultIfSuppressEnableCode) = {
Review Comment:
```suggestion
flushResultSuppressEnableCode) = {
```
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -173,6 +214,81 @@ class HashAggCodeGenerator(
HashAggCodeGenHelper.prepareMetrics(ctx, aggregateMapTerm, if (isFinal) sorterTerm else null)
+ // Do adaptive hash aggregation
+ val outputResultForOneRowAgg = {
Review Comment:
`outResultForAdaptiveHashAgg` maybe more clear?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -173,6 +214,81 @@ class HashAggCodeGenerator(
HashAggCodeGenHelper.prepareMetrics(ctx, aggregateMapTerm, if (isFinal) sorterTerm else null)
+ // Do adaptive hash aggregation
+ val outputResultForOneRowAgg = {
+ // gen code to iterating the aggregate map and output to downstream
+ val inputUnboxingCode = s"${ctx.reuseInputUnboxingCode(reuseAggBufferTerm)}"
+ val rowDataType = classOf[RowData].getCanonicalName
+ s"""
+ | // set result and output
+ |
+ | $reuseGroupKeyTerm = ($rowDataType)$currentKeyTerm;
+ | $reuseAggBufferTerm = ($rowDataType)$currentValueTerm;
+ | $inputUnboxingCode
+ | ${outputExpr.code}
+ | ${OperatorCodeGenerator.generateCollect(outputExpr.resultTerm)}
+ |
+ """.stripMargin
+ }
+ val localAggSuppressedTerm = CodeGenUtils.newName("localAggSuppressed")
+ ctx.addReusableMember(s"boolean $localAggSuppressedTerm = false;")
+ val (
+ distinctCountInCode,
Review Comment:
```suggestion
distinctCountIncCode,
```
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -173,6 +214,81 @@ class HashAggCodeGenerator(
HashAggCodeGenHelper.prepareMetrics(ctx, aggregateMapTerm, if (isFinal) sorterTerm else null)
+ // Do adaptive hash aggregation
+ val outputResultForOneRowAgg = {
+ // gen code to iterating the aggregate map and output to downstream
+ val inputUnboxingCode = s"${ctx.reuseInputUnboxingCode(reuseAggBufferTerm)}"
+ val rowDataType = classOf[RowData].getCanonicalName
+ s"""
+ | // set result and output
+ |
+ | $reuseGroupKeyTerm = ($rowDataType)$currentKeyTerm;
+ | $reuseAggBufferTerm = ($rowDataType)$currentValueTerm;
+ | $inputUnboxingCode
+ | ${outputExpr.code}
+ | ${OperatorCodeGenerator.generateCollect(outputExpr.resultTerm)}
+ |
+ """.stripMargin
+ }
+ val localAggSuppressedTerm = CodeGenUtils.newName("localAggSuppressed")
+ ctx.addReusableMember(s"boolean $localAggSuppressedTerm = false;")
+ val (
+ distinctCountInCode,
+ totalCountIncCode,
+ adaptiveSamplePointCode,
+ adaptiveSuppressCode,
+ flushResultIfSuppressEnableCode) = {
+ // from these conditions we know that it must be a distinct operation
+ if (
+ !isFinal &&
+ ctx.tableConfig.get(TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED) &&
+ canDoAdaptiveHashAgg
+ ) {
+ val adaptiveDistinctCountTerm = CodeGenUtils.newName("distinctCount")
+ val adaptiveTotalCountTerm = CodeGenUtils.newName("totalCount")
+ ctx.addReusableMember(s"long $adaptiveDistinctCountTerm = 0;")
+ ctx.addReusableMember(s"long $adaptiveTotalCountTerm = 0;")
+
+ val loggerTerm = CodeGenUtils.newName("LOG")
+ ctx.addReusableLogger(loggerTerm, className)
+
+ val samplePoint =
+ ctx.tableConfig.get(TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLE_POINT)
+ val limitDistinctRatio =
+ ctx.tableConfig.get(TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_LIMIT_DISTINCT_RATIO)
+
+ (
+ s"$adaptiveDistinctCountTerm++;",
+ s"$adaptiveTotalCountTerm++;",
+ s"""
+ |if ($adaptiveTotalCountTerm == $samplePoint) {
+ | $loggerTerm.info("Local hash agg checkpoint reached, sample point = " +
+ | $samplePoint + ", distinct = " + $adaptiveDistinctCountTerm + ", total = " +
+ | $adaptiveTotalCountTerm + ", limit distinct ratio = " + $limitDistinctRatio);
+ | if ((double) $adaptiveDistinctCountTerm / $adaptiveTotalCountTerm > $limitDistinctRatio) {
Review Comment:
```suggestion
| if ($adaptiveDistinctCountTerm / (1.0 * $adaptiveTotalCountTerm) > $limitDistinctRatio) {
```
We can multiply 1.0 directly instead of casting the result to double forcibly.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala:
##########
@@ -611,6 +611,71 @@ object GenerateUtils {
generateInputFieldUnboxing(ctx, inputType, inputCode, inputCode)
}
+ /**
+ * Do projection for grouping function 'max(col)' or 'min(col)' if adaptive hash agg takes effect.
+ * 'max/min(col)' will be convert to the current col value with same type.
+ */
+ def generateFiledAccessForMaxAndMin(
+ ctx: CodeGeneratorContext,
+ inputType: LogicalType,
+ inputTerm: String,
+ index: Int): GeneratedExpression = {
+ val fieldType = getFieldTypes(inputType).get(index)
+ val resultTypeTerm = primitiveTypeTermForType(fieldType)
+ val defaultValue = primitiveDefaultValue(fieldType)
+ val readCode = rowFieldReadAccess(index.toString, inputTerm, fieldType)
+ val Seq(fieldTerm, nullTerm) =
+ ctx.addReusableLocalVariables((resultTypeTerm, "field"), ("boolean", "isNull"))
+
+ val inputCode =
+ s"""
+ |$nullTerm = $inputTerm.isNullAt($index);
+ |$fieldTerm = $defaultValue;
+ |if (!$nullTerm) {
+ | $fieldTerm = $readCode;
+ |}
+ """.stripMargin.trim
+
+ GeneratedExpression(fieldTerm, nullTerm, inputCode, fieldType)
+ }
+
+ /**
+ * Do projection for grouping function 'count(col)' if adaptive hash agg takes effect.
+ * 'Count(col)' will be convert to 1L if col is not null and convert to 0L if col is null.
+ */
+ def generateFieldAccessForCountSpecificCol(
+ ctx: CodeGeneratorContext,
+ inputTerm: String,
+ index: Int): GeneratedExpression = {
+ val Seq(fieldTerm, nullTerm) =
+ ctx.addReusableLocalVariables(("long", "field"), ("boolean", "isNull"))
+
+ val inputCode =
+ s"""
+ |$nullTerm = $inputTerm.isNullAt($index);
+ |$fieldTerm = 0L;
+ |if (!$nullTerm) {
+ | $fieldTerm = 1L;
+ |}
+ """.stripMargin.trim
+
+ GeneratedExpression(fieldTerm, nullTerm, inputCode, new BigIntType())
+ }
+
+ /**
+ * Do projection for grouping function 'count(*)' or 'count()' if adaptive hash agg takes effect.
Review Comment:
```suggestion
* Do projection for grouping function 'count(*)' or 'count(1)' if adaptive hash agg takes effect.
```
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProjectionCodeGenerator.scala:
##########
@@ -124,6 +129,93 @@ object ProjectionCodeGenerator {
new GeneratedProjection(className, code, ctx.references.toArray, ctx.tableConfig)
}
+ /**
+ * If adaptive hash agg takes effect, and the sample result was to suppress local hash agg. In
Review Comment:
```suggestion
* If adaptive hash agg takes effect, local hash agg is suppressed. In
```
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala:
##########
@@ -611,6 +611,71 @@ object GenerateUtils {
generateInputFieldUnboxing(ctx, inputType, inputCode, inputCode)
}
+ /**
+ * Do projection for grouping function 'max(col)' or 'min(col)' if adaptive hash agg takes effect.
+ * 'max/min(col)' will be convert to the current col value with same type.
+ */
+ def generateFiledAccessForMaxAndMin(
Review Comment:
No one uses it now, drop this method.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProjectionCodeGenerator.scala:
##########
@@ -124,6 +129,93 @@ object ProjectionCodeGenerator {
new GeneratedProjection(className, code, ctx.references.toArray, ctx.tableConfig)
}
+ /**
+ * If adaptive hash agg takes effect, and the sample result was to suppress local hash agg. In
+ * order to ensure that the data format transmitted downstream with doing local hash agg is
Review Comment:
```suggestion
* order to ensure that the data structure transmitted downstream with doing local hash agg is
```
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProjectionCodeGenerator.scala:
##########
@@ -124,6 +129,93 @@ object ProjectionCodeGenerator {
new GeneratedProjection(className, code, ctx.references.toArray, ctx.tableConfig)
}
+ /**
+ * If adaptive hash agg takes effect, and the sample result was to suppress local hash agg. In
+ * order to ensure that the data format transmitted downstream with doing local hash agg is
+ * consistent with the data format transmitted downstream without doing local hash agg, we need to
+ * do projection for grouping function value.
+ *
+ * <p> For example, for sql statement "select a, avg(b), count(c) from T group by a", if local
+ * hash agg suppressed and a row (1, 5, "a") comes to local hash agg, we will pass (1, 5, 1, 1) to
+ * downstream.
+ */
+ def generatedAdaptiveHashAggValueProjectionCode(
+ ctx: CodeGeneratorContext,
+ inputType: RowType,
+ outClass: Class[_ <: RowData] = classOf[BinaryRowData],
+ inputTerm: String = DEFAULT_INPUT1_TERM,
+ aggInfos: Array[AggregateInfo],
+ outRecordTerm: String = DEFAULT_OUT_RECORD_TERM,
+ outRecordWriterTerm: String = DEFAULT_OUT_RECORD_WRITER_TERM): String = {
+ val fieldExprs: ArrayBuffer[GeneratedExpression] = ArrayBuffer()
+ aggInfos.map {
+ aggInfo =>
+ aggInfo.function match {
+ case _: SumAggFunction =>
+ fieldExprs += GenerateUtils.generateFieldAccess(
+ ctx,
+ inputType,
+ inputTerm,
+ aggInfo.agg.getArgList.get(0))
+ case _: MaxAggFunction | _: MinAggFunction =>
+ fieldExprs += GenerateUtils.generateFieldAccess(
+ ctx,
+ inputType,
+ inputTerm,
+ aggInfo.agg.getArgList.get(0))
+ case _: AvgAggFunction =>
+ fieldExprs += GenerateUtils.generateFieldAccess(
+ ctx,
+ inputType,
+ inputTerm,
+ aggInfo.agg.getArgList.get(0))
+ fieldExprs += GenerateUtils.generateFieldAccessForCountSpecificCol(
+ ctx,
+ inputTerm,
+ aggInfo.agg.getArgList.get(0))
+ case _: CountAggFunction =>
+ fieldExprs += GenerateUtils.generateFieldAccessForCountSpecificCol(
+ ctx,
+ inputTerm,
+ aggInfo.agg.getArgList.get(0))
+ case _: Count1AggFunction =>
+ fieldExprs += GenerateUtils.generateFieldAccessForCountOne(ctx)
+ }
+ }
+
+ val binaryRowWriter = CodeGenUtils.className[BinaryRowWriter]
+ val typeTerm = outClass.getCanonicalName
+ ctx.addReusableMember(s"$typeTerm $outRecordTerm= new $typeTerm(${fieldExprs.size});")
Review Comment:
```suggestion
ctx.addReusableMember(s"private $typeTerm $outRecordTerm= new $typeTerm(${fieldExprs.size});")
```
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/HashAggITCase.scala:
##########
@@ -17,12 +17,134 @@
*/
package org.apache.flink.table.planner.runtime.batch.sql.agg
+import org.apache.flink.api.scala._
import org.apache.flink.table.api.config.ExecutionConfigOptions
+import org.apache.flink.table.planner.codegen.agg.batch.HashAggCodeGenerator
+
+import org.junit.Test
/** AggregateITCase using HashAgg Operator. */
class HashAggITCase extends AggregateITCaseBase("HashAggregate") {
override def prepareAggOp(): Unit = {
tEnv.getConfig.set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "SortAgg")
}
+
+ @Test
+ def testAdaptiveHashAggWithHighAggregationDegree(): Unit = {
Review Comment:
I think we should cover the case that some element value is null of agg field.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on pull request #21586: [FLINK-30542][table-runtime] Introduce adaptive local hash aggregate to adaptively determine whether local hash aggregate is required at runtime
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on PR #21586:
URL: https://github.com/apache/flink/pull/21586#issuecomment-1408817568
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on pull request #21586: [FLINK-30542][table-runtime] Introduce adaptive local hash aggregate to adaptively determine whether local hash aggregate is required at runtime
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on PR #21586:
URL: https://github.com/apache/flink/pull/21586#issuecomment-1408687490
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #21586: [FLINK-30352][table-runtime] Introduce adaptive hash aggregate to adaptively determine whether local hash aggregate is required at runtime
Posted by GitBox <gi...@apache.org>.
flinkbot commented on PR #21586:
URL: https://github.com/apache/flink/pull/21586#issuecomment-1369368128
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9fe565a241df6db1a22bd552d90233e9895ff9db",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9fe565a241df6db1a22bd552d90233e9895ff9db",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9fe565a241df6db1a22bd552d90233e9895ff9db UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on pull request #21586: [FLINK-30542][table-runtime] Introduce adaptive hash aggregate to adaptively determine whether local hash aggregate is required at runtime
Posted by GitBox <gi...@apache.org>.
swuferhong commented on PR #21586:
URL: https://github.com/apache/flink/pull/21586#issuecomment-1384244507
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] swuferhong commented on a diff in pull request #21586: [FLINK-30542][table-runtime] Introduce adaptive hash aggregate to adaptively determine whether local hash aggregate is required at runtime
Posted by GitBox <gi...@apache.org>.
swuferhong commented on code in PR #21586:
URL: https://github.com/apache/flink/pull/21586#discussion_r1069126747
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala:
##########
@@ -611,6 +611,32 @@ object GenerateUtils {
generateInputFieldUnboxing(ctx, inputType, inputCode, inputCode)
}
+ def generateFieldAccessForCountCol(
Review Comment:
> generateFieldAccessForCountSpecificCol?
Done1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lsyldliu commented on a diff in pull request #21586: [FLINK-30542][table-runtime] Introduce adaptive hash aggregate to adaptively determine whether local hash aggregate is required at runtime
Posted by GitBox <gi...@apache.org>.
lsyldliu commented on code in PR #21586:
URL: https://github.com/apache/flink/pull/21586#discussion_r1067180556
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala:
##########
@@ -611,6 +611,32 @@ object GenerateUtils {
generateInputFieldUnboxing(ctx, inputType, inputCode, inputCode)
}
+ def generateFieldAccessForCountCol(
Review Comment:
generateFieldAccessForCountSpecificCol?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProjectionCodeGenerator.scala:
##########
@@ -124,6 +129,93 @@ object ProjectionCodeGenerator {
new GeneratedProjection(className, code, ctx.references.toArray, ctx.tableConfig)
}
+ /**
+ * If adaptive hash agg takes effect, and the sample result was to suppress local hash agg. In
+ * order to ensure that the data format transmitted downstream with doing local hash agg is
+ * consistent with the data format transmitted downstream without doing local hash agg, we need to
+ * do projection for grouping function value.
+ *
+ * <p> For example, for sql statement "select a, avg(b), count(c) from T group by a", if local
+ * hash agg suppressed and a row (1, 5, "a") comes to local hash agg, we will pass (1, 5, 1, 1) to
+ * downstream.
+ */
+ def generatedValueProjectionCode(
Review Comment:
generatedAdaptiveHashAggValueProjectionCode?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -173,6 +206,81 @@ class HashAggCodeGenerator(
HashAggCodeGenHelper.prepareMetrics(ctx, aggregateMapTerm, if (isFinal) sorterTerm else null)
+ // Do adaptive hash aggregation
+ val outputResultForOneRowAgg = {
+ // gen code to iterating the aggregate map and output to downstream
+ val inputUnboxingCode = s"${ctx.reuseInputUnboxingCode(reuseAggBufferTerm)}"
+ val rowDataType = classOf[RowData].getCanonicalName
+ s"""
+ | // set result and output
+ |
+ | $reuseGroupKeyTerm = ($rowDataType)$currentKeyTerm;
+ | $reuseAggBufferTerm = ($rowDataType)$currentValueTerm;
+ | $inputUnboxingCode
+ | ${outputExpr.code}
+ | ${OperatorCodeGenerator.generateCollect(outputExpr.resultTerm)}
+ |
+ """.stripMargin
+ }
+ val localAggSuppressedTerm = CodeGenUtils.newName("localAggSuppressed")
+ ctx.addReusableMember(s"boolean $localAggSuppressedTerm = false;")
+ val (
+ distinctCountInCode,
+ totalCountIncCode,
+ adaptiveSamplePointCode,
+ adaptiveSuppressCode,
+ flushResultIfSuppressEnableCode) = {
+ // from these conditions we know that it must be a distinct operation
+ if (
+ !isFinal &&
+ ctx.tableConfig.get(
+ HashAggCodeGenerator.TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED) && canDoAdaptiveHashAgg
Review Comment:
Removing the `HashAggCodeGenerator.`?
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala:
##########
@@ -173,6 +206,81 @@ class HashAggCodeGenerator(
HashAggCodeGenHelper.prepareMetrics(ctx, aggregateMapTerm, if (isFinal) sorterTerm else null)
+ // Do adaptive hash aggregation
+ val outputResultForOneRowAgg = {
+ // gen code to iterating the aggregate map and output to downstream
+ val inputUnboxingCode = s"${ctx.reuseInputUnboxingCode(reuseAggBufferTerm)}"
+ val rowDataType = classOf[RowData].getCanonicalName
+ s"""
+ | // set result and output
+ |
+ | $reuseGroupKeyTerm = ($rowDataType)$currentKeyTerm;
+ | $reuseAggBufferTerm = ($rowDataType)$currentValueTerm;
+ | $inputUnboxingCode
+ | ${outputExpr.code}
+ | ${OperatorCodeGenerator.generateCollect(outputExpr.resultTerm)}
+ |
+ """.stripMargin
+ }
+ val localAggSuppressedTerm = CodeGenUtils.newName("localAggSuppressed")
+ ctx.addReusableMember(s"boolean $localAggSuppressedTerm = false;")
+ val (
+ distinctCountInCode,
+ totalCountIncCode,
+ adaptiveSamplePointCode,
+ adaptiveSuppressCode,
+ flushResultIfSuppressEnableCode) = {
+ // from these conditions we know that it must be a distinct operation
+ if (
+ !isFinal &&
+ ctx.tableConfig.get(
+ HashAggCodeGenerator.TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_ENABLED) && canDoAdaptiveHashAgg
+ ) {
+ val adaptiveDistinctCountTerm = CodeGenUtils.newName("distinctCount")
+ val adaptiveTotalCountTerm = CodeGenUtils.newName("totalCount")
+ ctx.addReusableMember(s"long $adaptiveDistinctCountTerm = 0;")
+ ctx.addReusableMember(s"long $adaptiveTotalCountTerm = 0;")
+
+ val loggerTerm = CodeGenUtils.newName("LOG")
+ ctx.addReusableLogger(loggerTerm, className)
+
+ val samplePoint =
+ ctx.tableConfig.get(HashAggCodeGenerator.TABLE_EXEC_ADAPTIVE_LOCAL_HASH_AGG_SAMPLE_POINT)
Review Comment:
Ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org