Flink所有的内部函数都字义在类org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable中，其中有很多的函数是复用类org.apache.calcite.sql.fun.SqlStdOperatorTable中的函数。
生成对应java代码在类逻辑org.apache.flink.table.planner.codegen.calls.FunctionGenerator中

函数类型及实现

Aggregate Functions

函数的实现在org.apache.flink.table.planner.functions.aggfunctions中，在类org.apache.flink.table.planner.plan.utils.AggFunctionFactory中实现函数定义到函数实现逻辑的转化，具体代码如下
函数定义代码

  1.     // 类 org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable
  2.     // AGGREGATE OPERATORS
  3.     public static final SqlAggFunction SUM = SqlStdOperatorTable.SUM;
  4.     public static final SqlAggFunction SUM0 = SqlStdOperatorTable.SUM0;
  5.     public static final SqlAggFunction COUNT = SqlStdOperatorTable.COUNT;
  6.     public static final SqlAggFunction COLLECT = SqlStdOperatorTable.COLLECT;
  7.     public static final SqlAggFunction MIN = SqlStdOperatorTable.MIN;
  8.     public static final SqlAggFunction MAX = SqlStdOperatorTable.MAX;
  9.     public static final SqlAggFunction AVG = SqlStdOperatorTable.AVG;
  10.     public static final SqlAggFunction STDDEV = SqlStdOperatorTable.STDDEV;
  11.     public static final SqlAggFunction STDDEV_POP = SqlStdOperatorTable.STDDEV_POP;
  12.     public static final SqlAggFunction STDDEV_SAMP = SqlStdOperatorTable.STDDEV_SAMP;
  13.     public static final SqlAggFunction VARIANCE = SqlStdOperatorTable.VARIANCE;
  14.     public static final SqlAggFunction VAR_POP = SqlStdOperatorTable.VAR_POP;
  15.     public static final SqlAggFunction VAR_SAMP = SqlStdOperatorTable.VAR_SAMP;
  16.     public static final SqlAggFunction SINGLE_VALUE = SqlStdOperatorTable.SINGLE_VALUE;

函数实现代码

  1.      // 类 org.apache.flink.table.planner.plan.utils.AggFunctionFactory
  2.     /**
  3.     * The entry point to create an aggregate function from the given AggregateCall
  4.     */
  5.   def createAggFunction(call: AggregateCall, index: Int): UserDefinedFunction = {
  7.     val argTypes: Array[LogicalType] = call.getArgList
  8.       .map(inputType.getFieldList.get(_).getType)
  9.       .map(FlinkTypeFactory.toLogicalType)
  10.       .toArray
  12.     call.getAggregation match {
  13.       case a: SqlAvgAggFunction if a.kind == SqlKind.AVG => createAvgAggFunction(argTypes)
  15.       case _: SqlSumAggFunction => createSumAggFunction(argTypes, index)
  17.       case _: SqlSumEmptyIsZeroAggFunction => createSum0AggFunction(argTypes)
  19.       case a: SqlMinMaxAggFunction if a.getKind == SqlKind.MIN =>
  20.         createMinAggFunction(argTypes, index)
  22.       case a: SqlMinMaxAggFunction if a.getKind == SqlKind.MAX =>
  23.         createMaxAggFunction(argTypes, index)
  25.       case _: SqlCountAggFunction if call.getArgList.size() > 1 =>
  26.         throw new TableException("We now only support the count of one field.")
  28.       // TODO supports ApproximateCountDistinctAggFunction and CountDistinctAggFunction
  30.       case _: SqlCountAggFunction if call.getArgList.isEmpty => createCount1AggFunction(argTypes)
  32.       case _: SqlCountAggFunction => createCountAggFunction(argTypes)
  34.       case a: SqlRankFunction if a.getKind == SqlKind.ROW_NUMBER =>
  35.         createRowNumberAggFunction(argTypes)
  37.       case a: SqlRankFunction if a.getKind == SqlKind.RANK =>
  38.         createRankAggFunction(argTypes)
  40.       case a: SqlRankFunction if a.getKind == SqlKind.DENSE_RANK =>
  41.         createDenseRankAggFunction(argTypes)
  43.       case _: SqlLeadLagAggFunction =>
  44.         createLeadLagAggFunction(argTypes, index)
  46.       case _: SqlSingleValueAggFunction =>
  47.         createSingleValueAggFunction(argTypes)
  49.       case a: SqlFirstLastValueAggFunction if a.getKind == SqlKind.FIRST_VALUE =>
  50.         createFirstValueAggFunction(argTypes, index)
  52.       case a: SqlFirstLastValueAggFunction if a.getKind == SqlKind.LAST_VALUE =>
  53.         createLastValueAggFunction(argTypes, index)
  55.       case _: SqlListAggFunction if call.getArgList.size() == 1 =>
  56.         createListAggFunction(argTypes, index)
  58.       case _: SqlListAggFunction if call.getArgList.size() == 2 =>
  59.         createListAggWsFunction(argTypes, index)
  61.       // TODO supports SqlCardinalityCountAggFunction
  63.       case a: SqlAggFunction if a.getKind == SqlKind.COLLECT =>
  64.         createCollectAggFunction(argTypes)
  66.       case udagg: AggSqlFunction =>
  67.         // Can not touch the literals, Calcite make them in previous RelNode.
  68.         // In here, all inputs are input refs.
  69.         val constants = new util.ArrayList[AnyRef]()
  70.         argTypes.foreach(t => constants.add(null))
  71.         udagg.makeFunction(
  72.           constants.toArray,
  73.           argTypes)
  75.       case unSupported: SqlAggFunction =>
  76.         throw new TableException(s"Unsupported Function: '${unSupported.getName}'")
  77.     }
  78.   }

生成函数的code Generator的方法如下

  1. // 类 org.apache.flink.table.planner.codegen.agg.AggsHandlerCodeGenerator  
  2. private def initialAggregateInformation(aggInfoList: AggregateInfoList): Unit = {
  4.     this.accTypeInfo = RowType.of(
  5.       aggInfoList.getAccTypes.map(fromDataTypeToLogicalType): _*)
  6.     this.aggBufferSize = accTypeInfo.getFieldCount
  7.     var aggBufferOffset: Int = 0
  9.     if (mergedAccExternalTypes == null) {
  10.       mergedAccExternalTypes = aggInfoList.getAccTypes
  11.     }
  13.     val aggCodeGens = aggInfoList.aggInfos.map { aggInfo =>
  14.       val filterExpr = createFilterExpression(
  15.         aggInfo.agg.filterArg,
  16.         aggInfo.aggIndex,
  17.         aggInfo.agg.name)
  19.       val codegen = aggInfo.function match {
  20.         case _: DeclarativeAggregateFunction =>
  21.           new DeclarativeAggCodeGen(
  22.             ctx,
  23.             aggInfo,
  24.             filterExpr,
  25.             mergedAccOffset,
  26.             aggBufferOffset,
  27.             aggBufferSize,
  28.             inputFieldTypes,
  29.             constants,
  30.             relBuilder)
  31.         case _: UserDefinedAggregateFunction[_, _] =>
  32.           new ImperativeAggCodeGen(
  33.             ctx,
  34.             aggInfo,
  35.             filterExpr,
  36.             mergedAccOffset,
  37.             aggBufferOffset,
  38.             aggBufferSize,
  39.             inputFieldTypes,
  40.             constantExprs,
  41.             relBuilder,
  42.             hasNamespace,
  43.             mergedAccOnHeap,
  44.             mergedAccExternalTypes(aggBufferOffset),
  45.             copyInputField)
  46.       }
  47.       aggBufferOffset = aggBufferOffset + aggInfo.externalAccTypes.length
  48.       codegen
  49.     }
  51.     val distinctCodeGens = aggInfoList.distinctInfos.zipWithIndex.map {
  52.       case (distinctInfo, index) =>
  53.         val innerCodeGens = distinctInfo.aggIndexes.map(aggCodeGens(_)).toArray
  54.         val distinctIndex = aggCodeGens.length + index
  55.         val filterExpr = distinctInfo.filterArgs.map(
  56.           createFilterExpression(_, distinctIndex, "distinct aggregate"))
  57.         val codegen = new DistinctAggCodeGen(
  58.           ctx,
  59.           distinctInfo,
  60.           index,
  61.           innerCodeGens,
  62.           filterExpr.toArray,
  63.           mergedAccOffset,
  64.           aggBufferOffset,
  65.           aggBufferSize,
  66.           hasNamespace,
  67.           isMergeNeeded,
  68.           mergedAccOnHeap,
  69.           distinctInfo.consumeRetraction,
  70.           copyInputField,
  71.           relBuilder)
  72.         // distinct agg buffer occupies only one field
  73.         aggBufferOffset += 1
  74.         codegen
  75.     }
  77.     val distinctAggIndexes = aggInfoList.distinctInfos.flatMap(_.aggIndexes)
  78.     val nonDistinctAggIndexes = aggCodeGens.indices.filter(!distinctAggIndexes.contains(_)).toArray
  80.     this.aggBufferCodeGens = aggCodeGens ++ distinctCodeGens
  81.     this.aggActionCodeGens = nonDistinctAggIndexes.map(aggCodeGens(_)) ++ distinctCodeGens
  83.     // when input contains retractions, we inserted a count1 agg in the agg list
  84.     // the count1 agg value shouldn't be in the aggregate result
  85.     if (aggInfoList.indexOfCountStar.nonEmpty && aggInfoList.countStarInserted) {
  86.       ignoreAggValues ++= Array(aggInfoList.indexOfCountStar.get)
  87.     }
  89.     // the distinct value shouldn't be in the aggregate result
  90.     if (aggInfoList.distinctInfos.nonEmpty) {
  91.       ignoreAggValues ++= distinctCodeGens.indices.map(_ + aggCodeGens.length)
  92.     }
  93.   }

其它function

代码生成在包org.apache.flink.table.planner.codegen.calls.FunctionGenerator中