Flink 源码阅读笔记（18）- Flink SQL 中的流和动态表

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

trait StreamPhysicalRel extends FlinkPhysicalRel {
  /**
    * Whether the [[StreamPhysicalRel]] produces update and delete changes.
    */
  def producesUpdates: Boolean

  /**
    * Whether the [[StreamPhysicalRel]] produces retraction messages.
    */
  def producesRetractions: Boolean

  /**
    * Whether the [[StreamPhysicalRel]] requires retraction messages or not.
    */
  def needsUpdatesAsRetraction(input: RelNode): Boolean

  /**
    * Whether the [[StreamPhysicalRel]] consumes retraction messages instead of forwarding them.
    * The node might or might not produce new retraction messages.
    */
  def consumesRetractions: Boolean
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

class UpdateAsRetractionTrait(updateAsRetraction: Boolean) extends RelTrait {
}

object UpdateAsRetractionTrait {
  def apply(updateAsRetraction: Boolean): UpdateAsRetractionTrait = {
    new UpdateAsRetractionTrait(updateAsRetraction)
  }

  //默认不需要作为 RETRACT 输出
  val DEFAULT = new UpdateAsRetractionTrait(false)
}

class AccModeTrait(accMode: AccMode) extends RelTrait {
}

/**
  * The [[AccMode]] determines how insert, update, and delete changes of tables are encoded
  * by the messages that an operator emits.
  */
object AccMode extends Enumeration {
    /**
    * unknown acc mode
    */
  val UNKNOWN = Value

  /**
    * An operator in [[Acc]] mode emits change messages as
    * [[org.apache.flink.table.dataformat.BaseRow]] which encode a data row with header info,
    * logically equivalent to (boolean, row).
    *
    * An operator in [[Acc]] mode may only produce update and delete messages, if the table has
    * a unique key and all key attributes are contained in the Row.
    *
    * Changes are encoded as follows:
    * - insert: (true, NewRow)
    * - update: (true, NewRow) // the Row includes the full unique key to identify the row to update
    * - delete: (false, OldRow) // the Row includes the full unique key to identify the row to delete
    */
  val Acc = Value

  /**
    * * An operator in [[AccRetract]] mode emits change messages as
    * [[org.apache.flink.table.dataformat.BaseRow]] which encode a data row with header info,
    * logically equivalent to (boolean, row).
    *
    * Changes are encoded as follows:
    * - insert: (true, NewRow)
    * - update: (false, OldRow), (true, NewRow) // updates are encoded in two messages!
    * - delete: (false, OldRow)
    */
  val AccRetract = Value
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

class FlinkRelMdUniqueKeys private extends MetadataHandler[BuiltInMetadata.UniqueKeys] {
  def getUniqueKeys(
      rel: Aggregate,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): JSet[ImmutableBitSet] = {
    getUniqueKeysOnAggregate(rel.getGroupSet.toArray, mq, ignoreNulls)
  }

  def getUniqueKeys(
      rel: StreamExecGroupAggregate,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): JSet[ImmutableBitSet] = {
    getUniqueKeysOnAggregate(rel.grouping, mq, ignoreNulls)
  }

  def getUniqueKeysOnAggregate(
      grouping: Array[Int],
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): util.Set[ImmutableBitSet] = {
    // group by keys form a unique key
    ImmutableSet.of(ImmutableBitSet.of(grouping.indices: _*))
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

class FlinkRelMdUniqueKeys private extends MetadataHandler[BuiltInMetadata.UniqueKeys] {
  def getUniqueKeys(
      join: Join,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): JSet[ImmutableBitSet] = {
    join.getJoinType match {
      case JoinRelType.SEMI | JoinRelType.ANTI =>
        // only return the unique keys from the LHS since a SEMI/ANTI join only
        // returns the LHS
        mq.getUniqueKeys(join.getLeft, ignoreNulls)
      case _ =>
        getJoinUniqueKeys(
          join.analyzeCondition(), join.getJoinType, join.getLeft, join.getRight, mq, ignoreNulls)
    }
  }

  private def getJoinUniqueKeys(
      joinInfo: JoinInfo,
      joinRelType: JoinRelType,
      left: RelNode,
      right: RelNode,
      mq: RelMetadataQuery,
      ignoreNulls: Boolean): JSet[ImmutableBitSet] = {
    val leftUniqueKeys = mq.getUniqueKeys(left, ignoreNulls)
    val rightUniqueKeys = mq.getUniqueKeys(right, ignoreNulls)
    getJoinUniqueKeys(
      joinInfo, joinRelType, left.getRowType, leftUniqueKeys, rightUniqueKeys,
      mq.areColumnsUnique(left, joinInfo.leftSet, ignoreNulls),
      mq.areColumnsUnique(right, joinInfo.rightSet, ignoreNulls),
      mq)
  }
}

1
2
3
4
5

class StreamExecSink[T] {
  //如果是RetractStreamTableSink，则要求子节点必须以 RETRACT 形式发送消息
  override def needsUpdatesAsRetraction(input: RelNode): Boolean =
    sink.isInstanceOf[RetractStreamTableSink[_]]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

public class OutputConversionModifyOperation implements ModifyOperation {
  /**
	 * Should the output type contain the change flag, and what should the
	 * flag represent (retraction or deletion).
	 */
	public enum UpdateMode {
		APPEND,
		RETRACT,
		UPSERT
	}
}

1
2
3
4
5
6
7

class DataStreamTableSink[T](
    queryOperation: QueryOperation,
    outputType: TypeInformation[T],
    val updatesAsRetraction: Boolean, 
    val withChangeFlag: Boolean
  ) extends TableSink[T] {
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

abstract class PlannerBase {
  private[flink] def translateToRel(modifyOperation: ModifyOperation): RelNode = {
    modifyOperation match {
      case outputConversion: OutputConversionModifyOperation =>
        val input = getRelBuilder.queryOperation(outputConversion.getChild).build()
        val (updatesAsRetraction, withChangeFlag) = outputConversion.getUpdateMode match {
          case UpdateMode.RETRACT => (true, true)
          case UpdateMode.APPEND => (false, false)
          case UpdateMode.UPSERT => (false, true)
        }
        val typeInfo = LegacyTypeInfoDataTypeConverter.toLegacyTypeInfo(outputConversion.getType)
        val tableSink = new DataStreamTableSink(
          outputConversion.getChild, typeInfo, updatesAsRetraction, withChangeFlag)
        LogicalSink.create(input, tableSink, "DataStreamTableSink")
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

class FlinkUpdateAsRetractionTraitInitProgram extends FlinkOptimizeProgram[StreamOptimizeContext] {

  override def optimize(root: RelNode, context: StreamOptimizeContext): RelNode = {
    if (context.updateAsRetraction) {
      val newTraitSet = root.getTraitSet.plus(UpdateAsRetractionTrait(true))
      root.copy(newTraitSet, root.getInputs)
    } else {
      root
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

class StreamCommonSubGraphBasedOptimizer(planner: StreamPlanner)
  extends CommonSubGraphBasedOptimizer {
  
  override protected def doOptimize(roots: Seq[RelNode]): Seq[RelNodeBlock] = {
    val config = planner.getTableConfig
    // build RelNodeBlock plan
    val sinkBlocks = RelNodeBlockPlanBuilder.buildRelNodeBlockPlan(roots, config)
    // infer updateAsRetraction property for sink block
    sinkBlocks.foreach { sinkBlock =>
      val retractionFromRoot = sinkBlock.outputNode match {
        case n: Sink =>
          n.sink match {
            case _: RetractStreamTableSink[_] => true
            case s: DataStreamTableSink[_] => s.updatesAsRetraction
            case _ => false
          }
        case o =>
          o.getTraitSet.getTrait(UpdateAsRetractionTraitDef.INSTANCE).sendsUpdatesAsRetractions
      }
      sinkBlock.setUpdateAsRetraction(retractionFromRoot)

      .......
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

public final class BaseRowUtil {
	/**
	 * Indicates the row as an accumulate message.
	 */
	public static final byte ACCUMULATE_MSG = 0;

	/**
	 * Indicates the row as a retraction message.
	 */
	public static final byte RETRACT_MSG = 1;

	public static boolean isAccumulateMsg(BaseRow baseRow) {
		return baseRow.getHeader() == ACCUMULATE_MSG;
	}

	public static boolean isRetractMsg(BaseRow baseRow) {
		return baseRow.getHeader() == RETRACT_MSG;
	}
}