[GitHub] dianfu commented on a change in pull request #7177: [FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE
dianfu commented on a change in pull request #7177: [FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE URL: https://github.com/apache/flink/pull/7177#discussion_r237724165 ## File path: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala ## @@ -42,6 +50,90 @@ import scala.collection.mutable /** * A code generator for generating CEP related functions. * + * Aggregates are generated as follows: + * 1. all aggregate [[RexCall]]s are grouped by corresponding pattern variable + * 2. even if the same aggregation is used multiple times in an expression + *(e.g. SUM(A.price) > SUM(A.price) + 1) it will be calculated once. To do so [[AggBuilder]] + *keeps set of already seen different aggregation calls, and reuses the code to access + *appropriate field of aggregation result + * 3. after translating every expression (either in [[generateCondition]] or in + *[[generateOneRowPerMatchExpression]]) there will be generated code for + * * [[GeneratedFunction]], which will be an inner class Review comment: code format This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] dianfu commented on a change in pull request #7177: [FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE
dianfu commented on a change in pull request #7177: [FLINK-7599] [table]
Support for aggregates in MATCH_RECOGNIZE
URL: https://github.com/apache/flink/pull/7177#discussion_r237726170
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala
##
@@ -442,4 +582,245 @@ class MatchCodeGenerator(
generateFieldAccess(patternVariableRef.copy(code = NO_CODE),
fieldRef.getIndex)
}
+
+ class AggBuilder(variable: String) {
+
+private val aggregates = new mutable.ListBuffer[RexCall]()
+
+private val variableUID = newName("variable")
+
+private val resultRowTerm = newName(s"aggRow_$variableUID")
+
+private val rowTypeTerm = "org.apache.flink.types.Row"
+
+def generateAggAccess(aggCall: RexCall): GeneratedExpression = {
+ reusableInputUnboxingExprs.get((aggCall.toString, 0)) match {
+case Some(expr) =>
+ expr
+
+case None =>
+ val exp: GeneratedExpression = doGenerateAggAccess(aggCall)
+ aggregates += aggCall
+ reusableInputUnboxingExprs((aggCall.toString, 0)) = exp
+ exp.copy(code = NO_CODE)
+ }
+}
+
+private def doGenerateAggAccess(call: RexCall) = {
Review comment:
rename to aggCall
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] dianfu commented on a change in pull request #7177: [FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE
dianfu commented on a change in pull request #7177: [FLINK-7599] [table]
Support for aggregates in MATCH_RECOGNIZE
URL: https://github.com/apache/flink/pull/7177#discussion_r237737335
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala
##
@@ -442,4 +582,245 @@ class MatchCodeGenerator(
generateFieldAccess(patternVariableRef.copy(code = NO_CODE),
fieldRef.getIndex)
}
+
+ class AggBuilder(variable: String) {
+
+private val aggregates = new mutable.ListBuffer[RexCall]()
+
+private val variableUID = newName("variable")
+
+private val resultRowTerm = newName(s"aggRow_$variableUID")
+
+private val rowTypeTerm = "org.apache.flink.types.Row"
+
+def generateAggAccess(aggCall: RexCall): GeneratedExpression = {
+ reusableInputUnboxingExprs.get((aggCall.toString, 0)) match {
+case Some(expr) =>
+ expr
+
+case None =>
+ val exp: GeneratedExpression = doGenerateAggAccess(aggCall)
+ aggregates += aggCall
+ reusableInputUnboxingExprs((aggCall.toString, 0)) = exp
+ exp.copy(code = NO_CODE)
+ }
+}
+
+private def doGenerateAggAccess(call: RexCall) = {
+ val singleResultTerm = newName("result")
Review comment:
What about removing the prefix `single`?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] dianfu commented on a change in pull request #7177: [FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE
dianfu commented on a change in pull request #7177: [FLINK-7599] [table]
Support for aggregates in MATCH_RECOGNIZE
URL: https://github.com/apache/flink/pull/7177#discussion_r237738914
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala
##
@@ -442,4 +582,245 @@ class MatchCodeGenerator(
generateFieldAccess(patternVariableRef.copy(code = NO_CODE),
fieldRef.getIndex)
}
+
+ class AggBuilder(variable: String) {
+
+private val aggregates = new mutable.ListBuffer[RexCall]()
+
+private val variableUID = newName("variable")
+
+private val resultRowTerm = newName(s"aggRow_$variableUID")
+
+private val rowTypeTerm = "org.apache.flink.types.Row"
+
+def generateAggAccess(aggCall: RexCall): GeneratedExpression = {
+ reusableInputUnboxingExprs.get((aggCall.toString, 0)) match {
+case Some(expr) =>
+ expr
+
+case None =>
+ val exp: GeneratedExpression = doGenerateAggAccess(aggCall)
+ aggregates += aggCall
+ reusableInputUnboxingExprs((aggCall.toString, 0)) = exp
+ exp.copy(code = NO_CODE)
+ }
+}
+
+private def doGenerateAggAccess(call: RexCall) = {
+ val singleResultTerm = newName("result")
+ val singleResultNullTerm = newName("nullTerm")
+ val singleResultType = FlinkTypeFactory.toTypeInfo(call.`type`)
+ val primitiveSingleResultTypeTerm =
primitiveTypeTermForTypeInfo(singleResultType)
+ val boxedSingleResultTypeTerm =
boxedTypeTermForTypeInfo(singleResultType)
+
+ val patternName = findEventsByPatternName(variable)
+
+ val codeForAgg =
+j"""
+ |$rowTypeTerm $resultRowTerm =
calculateAgg_$variableUID(${patternName.resultTerm});
+ |""".stripMargin
+
+ reusablePerRecordStatements += codeForAgg
+
+ val defaultValue = primitiveDefaultValue(singleResultType)
+ val codeForSingleAgg = if (nullCheck) {
+j"""
+ |boolean $singleResultNullTerm;
+ |$primitiveSingleResultTypeTerm $singleResultTerm;
+ |if ($resultRowTerm.getField(${aggregates.size}) != null) {
+ | $singleResultTerm = ($boxedSingleResultTypeTerm) $resultRowTerm
+ |.getField(${aggregates.size});
+ | $singleResultNullTerm = false;
+ |} else {
+ | $singleResultNullTerm = true;
+ | $singleResultTerm = $defaultValue;
+ |}
+ |""".stripMargin
+ } else {
+j"""
+ |boolean $singleResultNullTerm = false;
+ |$primitiveSingleResultTypeTerm $singleResultTerm =
+ |($boxedSingleResultTypeTerm)
$resultRowTerm.getField(${aggregates.size});
+ |""".stripMargin
+ }
+
+ reusablePerRecordStatements += codeForSingleAgg
+
+ GeneratedExpression(singleResultTerm, singleResultNullTerm, NO_CODE,
singleResultType)
+}
+
+def generateAggFunction() : Unit = {
Review comment:
Remove the whitespace before the colon
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] dianfu commented on a change in pull request #7177: [FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE
dianfu commented on a change in pull request #7177: [FLINK-7599] [table]
Support for aggregates in MATCH_RECOGNIZE
URL: https://github.com/apache/flink/pull/7177#discussion_r237737176
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala
##
@@ -442,4 +582,245 @@ class MatchCodeGenerator(
generateFieldAccess(patternVariableRef.copy(code = NO_CODE),
fieldRef.getIndex)
}
+
+ class AggBuilder(variable: String) {
+
+private val aggregates = new mutable.ListBuffer[RexCall]()
+
+private val variableUID = newName("variable")
+
+private val resultRowTerm = newName(s"aggRow_$variableUID")
+
+private val rowTypeTerm = "org.apache.flink.types.Row"
+
+def generateAggAccess(aggCall: RexCall): GeneratedExpression = {
+ reusableInputUnboxingExprs.get((aggCall.toString, 0)) match {
+case Some(expr) =>
+ expr
+
+case None =>
+ val exp: GeneratedExpression = doGenerateAggAccess(aggCall)
+ aggregates += aggCall
+ reusableInputUnboxingExprs((aggCall.toString, 0)) = exp
+ exp.copy(code = NO_CODE)
+ }
+}
+
+private def doGenerateAggAccess(call: RexCall) = {
+ val singleResultTerm = newName("result")
+ val singleResultNullTerm = newName("nullTerm")
+ val singleResultType = FlinkTypeFactory.toTypeInfo(call.`type`)
+ val primitiveSingleResultTypeTerm =
primitiveTypeTermForTypeInfo(singleResultType)
+ val boxedSingleResultTypeTerm =
boxedTypeTermForTypeInfo(singleResultType)
+
+ val patternName = findEventsByPatternName(variable)
+
+ val codeForAgg =
+j"""
+ |$rowTypeTerm $resultRowTerm =
calculateAgg_$variableUID(${patternName.resultTerm});
Review comment:
Define `calculateAgg_$variableUID` as a variable such as `val
calculateAggFuncName = calculateAgg_$variableUID`
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] dianfu commented on a change in pull request #7177: [FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE
dianfu commented on a change in pull request #7177: [FLINK-7599] [table]
Support for aggregates in MATCH_RECOGNIZE
URL: https://github.com/apache/flink/pull/7177#discussion_r237472659
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala
##
@@ -442,4 +498,223 @@ class MatchCodeGenerator(
generateFieldAccess(patternVariableRef.copy(code = NO_CODE),
fieldRef.getIndex)
}
+
+ class AggBuilder(variable: String) {
+
+private val aggregates = new mutable.ListBuffer[RexCall]()
+
+private val variableUID = newName("variable")
+
+private val resultRowTerm = newName(s"aggRow_$variableUID")
+
+private val rowTypeTerm = "org.apache.flink.types.Row"
+
+def getOrAddAggregation(call: RexCall): GeneratedExpression = {
+ reusableInputUnboxingExprs.get((call.toString, 0)) match {
+case Some(expr) =>
+ expr
+
+case None =>
+ val exp: GeneratedExpression = generateAggAccess(call)
+ aggregates += call
+ reusableInputUnboxingExprs((call.toString, 0)) = exp
+ exp.copy(code = NO_CODE)
+ }
+}
+
+private def generateAggAccess(call: RexCall) = {
Review comment:
What about renaming to generateAggregation?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] dianfu commented on a change in pull request #7177: [FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE
dianfu commented on a change in pull request #7177: [FLINK-7599] [table]
Support for aggregates in MATCH_RECOGNIZE
URL: https://github.com/apache/flink/pull/7177#discussion_r237727039
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala
##
@@ -442,4 +582,245 @@ class MatchCodeGenerator(
generateFieldAccess(patternVariableRef.copy(code = NO_CODE),
fieldRef.getIndex)
}
+
+ class AggBuilder(variable: String) {
+
+private val aggregates = new mutable.ListBuffer[RexCall]()
+
+private val variableUID = newName("variable")
+
+private val resultRowTerm = newName(s"aggRow_$variableUID")
+
+private val rowTypeTerm = "org.apache.flink.types.Row"
+
+def generateAggAccess(aggCall: RexCall): GeneratedExpression = {
+ reusableInputUnboxingExprs.get((aggCall.toString, 0)) match {
+case Some(expr) =>
+ expr
+
+case None =>
+ val exp: GeneratedExpression = doGenerateAggAccess(aggCall)
+ aggregates += aggCall
+ reusableInputUnboxingExprs((aggCall.toString, 0)) = exp
+ exp.copy(code = NO_CODE)
+ }
+}
+
+private def doGenerateAggAccess(call: RexCall) = {
+ val singleResultTerm = newName("result")
+ val singleResultNullTerm = newName("nullTerm")
+ val singleResultType = FlinkTypeFactory.toTypeInfo(call.`type`)
+ val primitiveSingleResultTypeTerm =
primitiveTypeTermForTypeInfo(singleResultType)
+ val boxedSingleResultTypeTerm =
boxedTypeTermForTypeInfo(singleResultType)
+
+ val patternName = findEventsByPatternName(variable)
Review comment:
`patternName` doesn't reflects what this variable holds. What about rename
`patternName` to `eventsExpr` or something else?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] dianfu commented on a change in pull request #7177: [FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE
dianfu commented on a change in pull request #7177: [FLINK-7599] [table]
Support for aggregates in MATCH_RECOGNIZE
URL: https://github.com/apache/flink/pull/7177#discussion_r237739790
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala
##
@@ -442,4 +582,245 @@ class MatchCodeGenerator(
generateFieldAccess(patternVariableRef.copy(code = NO_CODE),
fieldRef.getIndex)
}
+
+ class AggBuilder(variable: String) {
+
+private val aggregates = new mutable.ListBuffer[RexCall]()
+
+private val variableUID = newName("variable")
+
+private val resultRowTerm = newName(s"aggRow_$variableUID")
+
+private val rowTypeTerm = "org.apache.flink.types.Row"
+
+def generateAggAccess(aggCall: RexCall): GeneratedExpression = {
+ reusableInputUnboxingExprs.get((aggCall.toString, 0)) match {
+case Some(expr) =>
+ expr
+
+case None =>
+ val exp: GeneratedExpression = doGenerateAggAccess(aggCall)
+ aggregates += aggCall
+ reusableInputUnboxingExprs((aggCall.toString, 0)) = exp
+ exp.copy(code = NO_CODE)
+ }
+}
+
+private def doGenerateAggAccess(call: RexCall) = {
+ val singleResultTerm = newName("result")
+ val singleResultNullTerm = newName("nullTerm")
+ val singleResultType = FlinkTypeFactory.toTypeInfo(call.`type`)
+ val primitiveSingleResultTypeTerm =
primitiveTypeTermForTypeInfo(singleResultType)
+ val boxedSingleResultTypeTerm =
boxedTypeTermForTypeInfo(singleResultType)
+
+ val patternName = findEventsByPatternName(variable)
+
+ val codeForAgg =
+j"""
+ |$rowTypeTerm $resultRowTerm =
calculateAgg_$variableUID(${patternName.resultTerm});
+ |""".stripMargin
+
+ reusablePerRecordStatements += codeForAgg
+
+ val defaultValue = primitiveDefaultValue(singleResultType)
+ val codeForSingleAgg = if (nullCheck) {
+j"""
+ |boolean $singleResultNullTerm;
+ |$primitiveSingleResultTypeTerm $singleResultTerm;
+ |if ($resultRowTerm.getField(${aggregates.size}) != null) {
+ | $singleResultTerm = ($boxedSingleResultTypeTerm) $resultRowTerm
+ |.getField(${aggregates.size});
+ | $singleResultNullTerm = false;
+ |} else {
+ | $singleResultNullTerm = true;
+ | $singleResultTerm = $defaultValue;
+ |}
+ |""".stripMargin
+ } else {
+j"""
+ |boolean $singleResultNullTerm = false;
+ |$primitiveSingleResultTypeTerm $singleResultTerm =
+ |($boxedSingleResultTypeTerm)
$resultRowTerm.getField(${aggregates.size});
+ |""".stripMargin
+ }
+
+ reusablePerRecordStatements += codeForSingleAgg
+
+ GeneratedExpression(singleResultTerm, singleResultNullTerm, NO_CODE,
singleResultType)
+}
+
+def generateAggFunction() : Unit = {
+ val matchAgg = extractAggregatesAndExpressions
+
+ val aggGenerator = new AggregationCodeGenerator(config, false, input,
None)
+
+ val aggFunc = aggGenerator.generateAggregations(
+s"AggFunction_$variableUID",
+matchAgg.inputExprs.map(r => FlinkTypeFactory.toTypeInfo(r.getType)),
+matchAgg.aggregations.map(_.aggFunction).toArray,
+matchAgg.aggregations.map(_.inputIndices).toArray,
+matchAgg.aggregations.indices.toArray,
+Array.fill(matchAgg.aggregations.size)(false),
+isStateBackedDataViews = false,
+partialResults = false,
+Array.emptyIntArray,
+None,
+matchAgg.aggregations.size,
+needRetract = false,
+needMerge = false,
+needReset = false,
+None
+ )
+
+ reusableMemberStatements.add(aggFunc.code)
+
+ val transformFuncName = s"transformRowForAgg_$variableUID"
+ val inputTransform: String = generateAggInputExprEvaluation(
+matchAgg.inputExprs,
+transformFuncName)
+
+ generateAggCalculation(aggFunc, transformFuncName, inputTransform)
+}
+
+private case class LogicalMatchAggCall(
+ function: SqlAggFunction,
+ inputTypes: Seq[RelDataType],
+ exprIndices: Seq[Int]
+)
+
+private case class MatchAggCall(
+ aggFunction: TableAggregateFunction[_, _],
+ inputIndices: Array[Int],
+ dataViews: Seq[DataViewSpec[_]]
+)
+
+private case class MatchAgg(
+ aggregations: Seq[MatchAggCall],
+ inputExprs: Seq[RexNode]
+)
+
+private def extractAggregatesAndExpressions: MatchAgg = {
+ val inputRows = new mutable.LinkedHashMap[String, (RexNode, Int)]
+
+ val logicalAggregates = aggregates.map(rexAggCall => {
Review comment:
rexAggCall renamed to aggCall?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and
[GitHub] dianfu commented on a change in pull request #7177: [FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE
dianfu commented on a change in pull request #7177: [FLINK-7599] [table]
Support for aggregates in MATCH_RECOGNIZE
URL: https://github.com/apache/flink/pull/7177#discussion_r237472868
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala
##
@@ -442,4 +498,223 @@ class MatchCodeGenerator(
generateFieldAccess(patternVariableRef.copy(code = NO_CODE),
fieldRef.getIndex)
}
+
+ class AggBuilder(variable: String) {
+
+private val aggregates = new mutable.ListBuffer[RexCall]()
+
+private val variableUID = newName("variable")
+
+private val resultRowTerm = newName(s"aggRow_$variableUID")
+
+private val rowTypeTerm = "org.apache.flink.types.Row"
+
+def getOrAddAggregation(call: RexCall): GeneratedExpression = {
Review comment:
getOrComputeAggregation?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] dianfu commented on a change in pull request #7177: [FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE
dianfu commented on a change in pull request #7177: [FLINK-7599] [table]
Support for aggregates in MATCH_RECOGNIZE
URL: https://github.com/apache/flink/pull/7177#discussion_r237731718
##
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala
##
@@ -442,4 +582,245 @@ class MatchCodeGenerator(
generateFieldAccess(patternVariableRef.copy(code = NO_CODE),
fieldRef.getIndex)
}
+
+ class AggBuilder(variable: String) {
+
+private val aggregates = new mutable.ListBuffer[RexCall]()
+
+private val variableUID = newName("variable")
+
+private val resultRowTerm = newName(s"aggRow_$variableUID")
+
+private val rowTypeTerm = "org.apache.flink.types.Row"
+
+def generateAggAccess(aggCall: RexCall): GeneratedExpression = {
+ reusableInputUnboxingExprs.get((aggCall.toString, 0)) match {
+case Some(expr) =>
+ expr
+
+case None =>
+ val exp: GeneratedExpression = doGenerateAggAccess(aggCall)
+ aggregates += aggCall
+ reusableInputUnboxingExprs((aggCall.toString, 0)) = exp
+ exp.copy(code = NO_CODE)
+ }
+}
+
+private def doGenerateAggAccess(call: RexCall) = {
+ val singleResultTerm = newName("result")
+ val singleResultNullTerm = newName("nullTerm")
+ val singleResultType = FlinkTypeFactory.toTypeInfo(call.`type`)
+ val primitiveSingleResultTypeTerm =
primitiveTypeTermForTypeInfo(singleResultType)
+ val boxedSingleResultTypeTerm =
boxedTypeTermForTypeInfo(singleResultType)
+
+ val patternName = findEventsByPatternName(variable)
+
+ val codeForAgg =
+j"""
+ |$rowTypeTerm $resultRowTerm =
calculateAgg_$variableUID(${patternName.resultTerm});
+ |""".stripMargin
+
+ reusablePerRecordStatements += codeForAgg
+
+ val defaultValue = primitiveDefaultValue(singleResultType)
+ val codeForSingleAgg = if (nullCheck) {
+j"""
+ |boolean $singleResultNullTerm;
+ |$primitiveSingleResultTypeTerm $singleResultTerm;
+ |if ($resultRowTerm.getField(${aggregates.size}) != null) {
+ | $singleResultTerm = ($boxedSingleResultTypeTerm) $resultRowTerm
+ |.getField(${aggregates.size});
+ | $singleResultNullTerm = false;
+ |} else {
+ | $singleResultNullTerm = true;
+ | $singleResultTerm = $defaultValue;
+ |}
+ |""".stripMargin
+ } else {
+j"""
+ |boolean $singleResultNullTerm = false;
+ |$primitiveSingleResultTypeTerm $singleResultTerm =
+ |($boxedSingleResultTypeTerm)
$resultRowTerm.getField(${aggregates.size});
+ |""".stripMargin
+ }
+
+ reusablePerRecordStatements += codeForSingleAgg
+
+ GeneratedExpression(singleResultTerm, singleResultNullTerm, NO_CODE,
singleResultType)
+}
+
+def generateAggFunction() : Unit = {
+ val matchAgg = extractAggregatesAndExpressions
+
+ val aggGenerator = new AggregationCodeGenerator(config, false, input,
None)
+
+ val aggFunc = aggGenerator.generateAggregations(
+s"AggFunction_$variableUID",
+matchAgg.inputExprs.map(r => FlinkTypeFactory.toTypeInfo(r.getType)),
+matchAgg.aggregations.map(_.aggFunction).toArray,
+matchAgg.aggregations.map(_.inputIndices).toArray,
+matchAgg.aggregations.indices.toArray,
+Array.fill(matchAgg.aggregations.size)(false),
+isStateBackedDataViews = false,
+partialResults = false,
+Array.emptyIntArray,
+None,
+matchAgg.aggregations.size,
+needRetract = false,
+needMerge = false,
+needReset = false,
+None
+ )
+
+ reusableMemberStatements.add(aggFunc.code)
+
+ val transformFuncName = s"transformRowForAgg_$variableUID"
+ val inputTransform: String = generateAggInputExprEvaluation(
+matchAgg.inputExprs,
+transformFuncName)
+
+ generateAggCalculation(aggFunc, transformFuncName, inputTransform)
+}
+
+private case class LogicalMatchAggCall(
+ function: SqlAggFunction,
+ inputTypes: Seq[RelDataType],
+ exprIndices: Seq[Int]
+)
+
+private case class MatchAggCall(
+ aggFunction: TableAggregateFunction[_, _],
+ inputIndices: Array[Int],
+ dataViews: Seq[DataViewSpec[_]]
+)
+
+private case class MatchAgg(
+ aggregations: Seq[MatchAggCall],
+ inputExprs: Seq[RexNode]
+)
+
+private def extractAggregatesAndExpressions: MatchAgg = {
+ val inputRows = new mutable.LinkedHashMap[String, (RexNode, Int)]
+
+ val logicalAggregates = aggregates.map(rexAggCall => {
+val callsWithIndices = rexAggCall.operands.asScala.map(innerCall => {
+ inputRows.get(innerCall.toString) match {
+case Some(x) =>
+ x
+
+case None =>
+ val
[GitHub] dianfu commented on a change in pull request #7177: [FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE
dianfu commented on a change in pull request #7177: [FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE URL: https://github.com/apache/flink/pull/7177#discussion_r237466962 ## File path: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala ## @@ -71,12 +78,28 @@ class MatchCodeGenerator( private var offset: Int = 0 private var first : Boolean = false + /** +* Flags that tells if we generate expressions inside an aggregate. It tells how to access input Review comment: Flags -> Flag This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
