序
本文主要研究一下flink Table的select操作
Table.select
flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/api/table.scala
class Table( private[flink] val tableEnv: TableEnvironment, private[flink] val logicalPlan: LogicalNode) { //...... def select(fields: String): Table = { val fieldExprs = ExpressionParser.parseExpressionList(fields) //get the correct expression for AggFunctionCall val withResolvedAggFunctionCall = fieldExprs.map(replaceAggFunctionCall(_, tableEnv)) select(withResolvedAggFunctionCall: _*) } def replaceAggFunctionCall(field: Expression, tableEnv: TableEnvironment): Expression = { field match { case l: LeafExpression => l case u: UnaryExpression => val c = replaceAggFunctionCall(u.child, tableEnv) u.makeCopy(Array(c)) case b: BinaryExpression => val l = replaceAggFunctionCall(b.left, tableEnv) val r = replaceAggFunctionCall(b.right, tableEnv) b.makeCopy(Array(l, r)) // Functions calls case c @ Call(name, args) => val function = tableEnv.getFunctionCatalog.lookupFunction(name, args) function match { case a: AggFunctionCall => a case a: Aggregation => a case p: AbstractWindowProperty => p case _ => val newArgs = args.map( (exp: Expression) => replaceAggFunctionCall(exp, tableEnv)) c.makeCopy(Array(name, newArgs)) } // Scala functions case sfc @ ScalarFunctionCall(clazz, args) => val newArgs: Seq[Expression] = args.map( (exp: Expression) => replaceAggFunctionCall(exp, tableEnv)) sfc.makeCopy(Array(clazz, newArgs)) // Array constructor case c @ ArrayConstructor(args) => val newArgs = c.elements .map((exp: Expression) => replaceAggFunctionCall(exp, tableEnv)) c.makeCopy(Array(newArgs)) // Other expressions case e: Expression => e } } def select(fields: Expression*): Table = { val expandedFields = expandProjectList(fields, logicalPlan, tableEnv) val (aggNames, propNames) = extractAggregationsAndProperties(expandedFields, tableEnv) if (propNames.nonEmpty) { throw new ValidationException("Window properties can only be used on windowed tables.") } if (aggNames.nonEmpty) { val projectsOnAgg = replaceAggregationsAndProperties( expandedFields, tableEnv, aggNames, propNames) val projectFields = extractFieldReferences(expandedFields) new Table(tableEnv, Project(projectsOnAgg, Aggregate(Nil, aggNames.map(a => Alias(a._1, a._2)).toSeq, Project(projectFields, logicalPlan).validate(tableEnv) ).validate(tableEnv) ).validate(tableEnv) ) } else { new Table(tableEnv, Project(expandedFields.map(UnresolvedAlias), logicalPlan).validate(tableEnv)) } } //......} - Table提供了两个select方法，一个接收String参数，一个接收Expression参数
- String参数的select内部先调用ExpressionParser.parseExpressionList解析String，之后再通过replaceAggFunctionCall替换UDAGG function，最后再调用Expression参数的select方法
- Expression参数的select方法会使用Project重新创建Table，如果有aggregate的话，会创建Aggregate，然后再通过Project包装
Expression
flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/expressions/Expression.scala
abstract class Expression extends TreeNode[Expression] { /** * Returns the [[TypeInformation]] for evaluating this expression. * It is sometimes not available until the expression is valid. */ private[flink] def resultType: TypeInformation[_] /** * One pass validation of the expression tree in post order. */ private[flink] lazy val valid: Boolean = childrenValid && validateInput().isSuccess private[flink] def childrenValid: Boolean = children.forall(_.valid) /** * Check input data types, inputs number or other properties specified by this expression. * Return `ValidationSuccess` if it pass the check, * or `ValidationFailure` with supplement message explaining the error. * Note: we should only call this method until `childrenValid == true` */ private[flink] def validateInput(): ValidationResult = ValidationSuccess /** * Convert Expression to its counterpart in Calcite, i.e. RexNode */ private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = throw new UnsupportedOperationException( s"${this.getClass.getName} cannot be transformed to RexNode" ) private[flink] def checkEquals(other: Expression): Boolean = { if (this.getClass != other.getClass) { false } else { def checkEquality(elements1: Seq[Any], elements2: Seq[Any]): Boolean = { elements1.length == elements2.length && elements1.zip(elements2).forall { case (e1: Expression, e2: Expression) => e1.checkEquals(e2) case (t1: Seq[_], t2: Seq[_]) => checkEquality(t1, t2) case (i1, i2) => i1 == i2 } } val elements1 = this.productIterator.toSeq val elements2 = other.productIterator.toSeq checkEquality(elements1, elements2) } }}abstract class BinaryExpression extends Expression { private[flink] def left: Expression private[flink] def right: Expression private[flink] def children = Seq(left, right)}abstract class UnaryExpression extends Expression { private[flink] def child: Expression private[flink] def children = Seq(child)}abstract class LeafExpression extends Expression { private[flink] val children = Nil} - Expression继承了TreeNode，它有三个抽象子类分别是BinaryExpression、UnaryExpression、LeafExpression
Project
flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/plan/logical/operators.scala
case class Project( projectList: Seq[NamedExpression], child: LogicalNode, explicitAlias: Boolean = false) extends UnaryNode { override def output: Seq[Attribute] = projectList.map(_.toAttribute) override def resolveExpressions(tableEnv: TableEnvironment): LogicalNode = { val afterResolve = super.resolveExpressions(tableEnv).asInstanceOf[Project] val newProjectList = afterResolve.projectList.zipWithIndex.map { case (e, i) => e match { case u @ UnresolvedAlias(c) => c match { case ne: NamedExpression => ne case expr if !expr.valid => u case c @ Cast(ne: NamedExpression, tp) => Alias(c, s"${ne.name}-$tp") case gcf: GetCompositeField => Alias(gcf, gcf.aliasName().getOrElse(s"_c$i")) case other => Alias(other, s"_c$i") } case _ => throw new RuntimeException("This should never be called and probably points to a bug.") } } Project(newProjectList, child, explicitAlias) } override def validate(tableEnv: TableEnvironment): LogicalNode = { val resolvedProject = super.validate(tableEnv).asInstanceOf[Project] val names: mutable.Set[String] = mutable.Set() def checkName(name: String): Unit = { if (names.contains(name)) { failValidation(s"Duplicate field name $name.") } else { names.add(name) } } resolvedProject.projectList.foreach { case n: Alias => // explicit name checkName(n.name) case r: ResolvedFieldReference => // simple field forwarding checkName(r.name) case _ => // Do nothing } resolvedProject } override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = { child.construct(relBuilder) val exprs = if (explicitAlias) { projectList } else { // remove AS expressions, according to Calcite they should not be in a final RexNode projectList.map { case Alias(e: Expression, _, _) => e case e: Expression => e } } relBuilder.project( exprs.map(_.toRexNode(relBuilder)).asJava, projectList.map(_.name).asJava, true) }} - Project继承了UnaryNode，它构造器接收Seq[NamedExpression]、LogicalNode、explicitAlias三个参数，其中explicitAlias可选，默认为false
Aggregate
flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/plan/logical/operators.scala
case class Aggregate( groupingExpressions: Seq[Expression], aggregateExpressions: Seq[NamedExpression], child: LogicalNode) extends UnaryNode { override def output: Seq[Attribute] = { (groupingExpressions ++ aggregateExpressions) map { case ne: NamedExpression => ne.toAttribute case e => Alias(e, e.toString).toAttribute } } override protected[logical] def construct(relBuilder: RelBuilder): RelBuilder = { child.construct(relBuilder) relBuilder.aggregate( relBuilder.groupKey(groupingExpressions.map(_.toRexNode(relBuilder)).asJava), aggregateExpressions.map { case Alias(agg: Aggregation, name, _) => agg.toAggCall(name)(relBuilder) case _ => throw new RuntimeException("This should never happen.") }.asJava) } override def validate(tableEnv: TableEnvironment): LogicalNode = { implicit val relBuilder: RelBuilder = tableEnv.getRelBuilder val resolvedAggregate = super.validate(tableEnv).asInstanceOf[Aggregate] val groupingExprs = resolvedAggregate.groupingExpressions val aggregateExprs = resolvedAggregate.aggregateExpressions aggregateExprs.foreach(validateAggregateExpression) groupingExprs.foreach(validateGroupingExpression) def validateAggregateExpression(expr: Expression): Unit = expr match { case distinctExpr: DistinctAgg => distinctExpr.child match { case _: DistinctAgg => failValidation( "Chained distinct operators are not supported!") case aggExpr: Aggregation => validateAggregateExpression(aggExpr) case _ => failValidation( "Distinct operator can only be applied to aggregation expressions!") } // check aggregate function case aggExpr: Aggregation if aggExpr.getSqlAggFunction.requiresOver => failValidation(s"OVER clause is necessary for window functions: [${aggExpr.getClass}].") // check no nested aggregation exists. case aggExpr: Aggregation => aggExpr.children.foreach { child => child.preOrderVisit { case agg: Aggregation => failValidation( "It's not allowed to use an aggregate function as " + "input of another aggregate function") case _ => // OK } } case a: Attribute if !groupingExprs.exists(_.checkEquals(a)) => failValidation( s"expression '$a' is invalid because it is neither" + " present in group by nor an aggregate function") case e if groupingExprs.exists(_.checkEquals(e)) => // OK case e => e.children.foreach(validateAggregateExpression) } def validateGroupingExpression(expr: Expression): Unit = { if (!expr.resultType.isKeyType) { failValidation( s"expression $expr cannot be used as a grouping expression " + "because it's not a valid key type which must be hashable and comparable") } } resolvedAggregate }} - Aggregate继承了UnaryNode，它构造器接收Seq[Expression]、Seq[NamedExpression]、LogicalNode三个参数
LogicalNode
flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/plan/logical/LogicalNode.scala
abstract class LogicalNode extends TreeNode[LogicalNode] { def output: Seq[Attribute] def resolveExpressions(tableEnv: TableEnvironment): LogicalNode = { // resolve references and function calls val exprResolved = expressionPostOrderTransform { case u @ UnresolvedFieldReference(name) => // try resolve a field resolveReference(tableEnv, name).getOrElse(u) case c @ Call(name, children) if c.childrenValid => tableEnv.getFunctionCatalog.lookupFunction(name, children) } exprResolved.expressionPostOrderTransform { case ips: InputTypeSpec if ips.childrenValid => var changed: Boolean = false val newChildren = ips.expectedTypes.zip(ips.children).map { case (tpe, child) => val childType = child.resultType if (childType != tpe && TypeCoercion.canSafelyCast(childType, tpe)) { changed = true Cast(child, tpe) } else { child } }.toArray[AnyRef] if (changed) ips.makeCopy(newChildren) else ips } } final def toRelNode(relBuilder: RelBuilder): RelNode = construct(relBuilder).build() protected[logical] def construct(relBuilder: RelBuilder): RelBuilder def validate(tableEnv: TableEnvironment): LogicalNode = { val resolvedNode = resolveExpressions(tableEnv) resolvedNode.expressionPostOrderTransform { case a: Attribute if !a.valid => val from = children.flatMap(_.output).map(_.name).mkString(", ") // give helpful error message for null literals if (a.name == "null") { failValidation(s"Cannot resolve field [${a.name}] given input [$from]. If you want to " + s"express a null literal, use 'Null(TYPE)' for typed null expressions. " + s"For example: Null(INT)") } else { failValidation(s"Cannot resolve field [${a.name}] given input [$from].") } case e: Expression if e.validateInput().isFailure => failValidation(s"Expression $e failed on input check: " + s"${e.validateInput().asInstanceOf[ValidationFailure].message}") } } /** * Resolves the given strings to a [[NamedExpression]] using the input from all child * nodes of this LogicalPlan. */ def resolveReference(tableEnv: TableEnvironment, name: String): Option[NamedExpression] = { // try to resolve a field val childrenOutput = children.flatMap(_.output) val fieldCandidates = childrenOutput.filter(_.name.equalsIgnoreCase(name)) if (fieldCandidates.length > 1) { failValidation(s"Reference $name is ambiguous.") } else if (fieldCandidates.nonEmpty) { return Some(fieldCandidates.head.withName(name)) } // try to resolve a table tableEnv.scanInternal(Array(name)) match { case Some(table) => Some(TableReference(name, table)) case None => None } } /** * Runs [[postOrderTransform]] with `rule` on all expressions present in this logical node. * * @param rule the rule to be applied to every expression in this logical node. */ def expressionPostOrderTransform(rule: PartialFunction[Expression, Expression]): LogicalNode = { var changed = false def expressionPostOrderTransform(e: Expression): Expression = { val newExpr = e.postOrderTransform(rule) if (newExpr.fastEquals(e)) { e } else { changed = true newExpr } } val newArgs = productIterator.map { case e: Expression => expressionPostOrderTransform(e) case Some(e: Expression) => Some(expressionPostOrderTransform(e)) case seq: Traversable[_] => seq.map { case e: Expression => expressionPostOrderTransform(e) case other => other } case r: Resolvable[_] => r.resolveExpressions(e => expressionPostOrderTransform(e)) case other: AnyRef => other }.toArray if (changed) makeCopy(newArgs) else this } protected def failValidation(msg: String): Nothing = { throw new ValidationException(msg) }}abstract class LeafNode extends LogicalNode { override def children: Seq[LogicalNode] = Nil}abstract class UnaryNode extends LogicalNode { def child: LogicalNode override def children: Seq[LogicalNode] = child :: Nil}abstract class BinaryNode extends LogicalNode { def left: LogicalNode def right: LogicalNode override def children: Seq[LogicalNode] = left :: right :: Nil} - LogicalNode跟Expression一样，也继承了TreeNode；LogicalNode有三个抽象子类，分别是BinaryNode、UnaryNode、LeafNode
小结
- Table提供了两个select方法，一个接收String参数，一个接收Expression参数；String参数的select内部先调用ExpressionParser.parseExpressionList解析String，之后再通过replaceAggFunctionCall替换UDAGG function，最后再调用Expression参数的select方法
- Expression参数的select方法会使用Project重新创建Table，如果有aggregate的话，会创建Aggregate，然后在通过Project包装
- Project及Aggregate都是case class，它们都继承了UnaryNode，UnaryNode是LogicalNode的子类；LogicalNode跟Expression一样，也继承了TreeNode；Expression有三个抽象子类分别是BinaryExpression、UnaryExpression、LeafExpression；LogicalNode也有三个抽象子类，分别是BinaryNode、UnaryNode、LeafNode