You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2019/03/04 03:24:54 UTC
[flink] branch master updated: [FLINK-11795][table-planner-blink]
Introduce DataStream nodes and converter rules for batch and stream
This is an automated email from the ASF dual-hosted git repository.
kurt pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 6918d62 [FLINK-11795][table-planner-blink] Introduce DataStream nodes and converter rules for batch and stream
6918d62 is described below
commit 6918d62bac9e421dd06611563ecf4c964fbf2448
Author: godfreyhe <go...@163.com>
AuthorDate: Sun Mar 3 19:59:34 2019 +0800
[FLINK-11795][table-planner-blink] Introduce DataStream nodes and converter rules for batch and stream
This closes #7881
---
.../flink/table/calcite/FlinkTypeFactory.scala | 244 +++++++++++++++++++++
.../logical/FlinkLogicalDataStreamTableScan.scala | 77 +++++++
.../batch/BatchExecBoundedStreamScan.scala | 63 ++++++
.../physical/stream/StreamExecDataStreamScan.scala | 68 ++++++
.../physical/batch/BatchExecScanTableRule.scala | 54 +++++
.../rules/physical/stream/StreamExecScanRule.scala | 58 +++++
.../flink/table/plan/schema/DataStreamTable.scala | 50 +++++
.../flink/table/plan/schema/FlinkTable.scala | 43 ++++
.../flink/table/plan/schema/InlineTable.scala | 101 +++++++++
.../flink/table/plan/stats/FlinkStatistic.scala | 73 ++++++
10 files changed, 831 insertions(+)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
new file mode 100644
index 0000000..ba52152
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
@@ -0,0 +1,244 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.calcite
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo._
+import org.apache.flink.api.common.typeinfo._
+import org.apache.flink.api.java.typeutils.ValueTypeInfo._
+import org.apache.flink.table.`type`.DecimalType
+import org.apache.flink.table.api.TableException
+import org.apache.flink.table.calcite.FlinkTypeFactory.typeInfoToSqlTypeName
+import org.apache.flink.table.typeutils._
+
+import org.apache.calcite.avatica.util.TimeUnit
+import org.apache.calcite.jdbc.JavaTypeFactoryImpl
+import org.apache.calcite.rel.`type`._
+import org.apache.calcite.sql.SqlIntervalQualifier
+import org.apache.calcite.sql.`type`.SqlTypeName
+import org.apache.calcite.sql.`type`.SqlTypeName._
+import org.apache.calcite.sql.parser.SqlParserPos
+
+import java.util
+
+import scala.collection.JavaConverters._
+
+/**
+ * Flink specific type factory that represents the interface between Flink's [[TypeInformation]]
+ * and Calcite's [[RelDataType]].
+ */
+class FlinkTypeFactory(typeSystem: RelDataTypeSystem) extends JavaTypeFactoryImpl(typeSystem) {
+
+ // NOTE: for future data types it might be necessary to
+ // override more methods of RelDataTypeFactoryImpl
+
+ def isAdvanced(dataType: TypeInformation[_]): Boolean = dataType match {
+ case PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO => false
+ case _: BasicTypeInfo[_] => false
+ case _: SqlTimeTypeInfo[_] => false
+ case _: TimeIntervalTypeInfo[_] => false
+ case _ => true
+ }
+
+ def createTypeFromTypeInfo(
+ typeInfo: TypeInformation[_],
+ isNullable: Boolean): RelDataType = {
+
+ // we cannot use seenTypes for simple types,
+ // because time indicators and timestamps would be the same
+
+ val relType = if (!isAdvanced(typeInfo)) {
+ // simple types can be converted to SQL types and vice versa
+ val sqlType = typeInfoToSqlTypeName(typeInfo)
+ sqlType match {
+
+ case INTERVAL_YEAR_MONTH =>
+ createSqlIntervalType(
+ new SqlIntervalQualifier(TimeUnit.YEAR, TimeUnit.MONTH, SqlParserPos.ZERO))
+
+ case INTERVAL_DAY_SECOND =>
+ createSqlIntervalType(
+ new SqlIntervalQualifier(TimeUnit.DAY, TimeUnit.SECOND, SqlParserPos.ZERO))
+
+ case _ =>
+ createSqlType(sqlType)
+ }
+ } else {
+ throw new TableException("Unsupported now")
+ }
+
+ createTypeWithNullability(relType, isNullable)
+ }
+
+ /**
+ * Creates a struct type with the input fieldNames and input fieldTypes using FlinkTypeFactory
+ *
+ * @param fieldNames field names
+ * @param fieldTypes field types, every element is Flink's [[TypeInformation]]
+ * @return a struct type with the input fieldNames, input fieldTypes, and system fields
+ */
+ def buildLogicalRowType(
+ fieldNames: Seq[String],
+ fieldTypes: Seq[TypeInformation[_]]): RelDataType = {
+ buildLogicalRowType(
+ fieldNames,
+ fieldTypes,
+ fieldTypes.map(_ => true))
+ }
+
+ /**
+ * Creates a struct type with the input fieldNames, input fieldTypes and input fieldNullables
+ * using FlinkTypeFactory
+ *
+ * @param fieldNames field names
+ * @param fieldTypes field types, every element is Flink's [[TypeInformation]]
+ * @param fieldNullables field nullable properties
+ * @return a struct type with the input fieldNames, input fieldTypes, and system fields
+ */
+ def buildLogicalRowType(
+ fieldNames: Seq[String],
+ fieldTypes: Seq[TypeInformation[_]],
+ fieldNullables: Seq[Boolean]): RelDataType = {
+ val logicalRowTypeBuilder = builder
+ val fields = fieldNames.zip(fieldTypes).zip(fieldNullables)
+ fields foreach {
+ case ((fieldName, fieldType), fieldNullable) =>
+ logicalRowTypeBuilder.add(fieldName, createTypeFromTypeInfo(fieldType, fieldNullable))
+ }
+ logicalRowTypeBuilder.build
+ }
+
+ // ----------------------------------------------------------------------------------------------
+
+ override def getJavaClass(`type`: RelDataType): java.lang.reflect.Type = {
+ if (`type`.getSqlTypeName == FLOAT) {
+ if (`type`.isNullable) {
+ classOf[java.lang.Float]
+ } else {
+ java.lang.Float.TYPE
+ }
+ } else {
+ super.getJavaClass(`type`)
+ }
+ }
+
+ override def createSqlType(typeName: SqlTypeName, precision: Int): RelDataType = {
+ // it might happen that inferred VARCHAR types overflow as we set them to Int.MaxValue
+ // Calcite will limit the length of the VARCHAR type to 65536.
+ if (typeName == VARCHAR && precision < 0) {
+ createSqlType(typeName, getTypeSystem.getDefaultPrecision(typeName))
+ } else {
+ super.createSqlType(typeName, precision)
+ }
+ }
+
+ override def createSqlType(typeName: SqlTypeName): RelDataType = {
+ if (typeName == DECIMAL) {
+ // if we got here, the precision and scale are not specified, here we
+ // keep precision/scale in sync with our type system's default value,
+ // see DecimalType.USER_DEFAULT.
+ createSqlType(typeName, DecimalType.USER_DEFAULT.precision(),
+ DecimalType.USER_DEFAULT.scale())
+ } else {
+ super.createSqlType(typeName)
+ }
+ }
+
+ override def createTypeWithNullability(
+ relDataType: RelDataType,
+ isNullable: Boolean): RelDataType = {
+
+ // nullability change not necessary
+ if (relDataType.isNullable == isNullable) {
+ return canonize(relDataType)
+ }
+
+ // change nullability
+ val newType = super.createTypeWithNullability(relDataType, isNullable)
+
+ canonize(newType)
+ }
+
+ override def leastRestrictive(types: util.List[RelDataType]): RelDataType = {
+ val type0 = types.get(0)
+ if (type0.getSqlTypeName != null) {
+ val resultType = resolveAllIdenticalTypes(types)
+ if (resultType.isDefined) {
+ // result type for identical types
+ return resultType.get
+ }
+ }
+ // fall back to super
+ super.leastRestrictive(types)
+ }
+
+ private def resolveAllIdenticalTypes(types: util.List[RelDataType]): Option[RelDataType] = {
+ val allTypes = types.asScala
+
+ val head = allTypes.head
+ // check if all types are the same
+ if (allTypes.forall(_ == head)) {
+ // types are the same, check nullability
+ val nullable = allTypes
+ .exists(sqlType => sqlType.isNullable || sqlType.getSqlTypeName == SqlTypeName.NULL)
+ // return type with nullability
+ Some(createTypeWithNullability(head, nullable))
+ } else {
+ // types are not all the same
+ if (allTypes.exists(_.getSqlTypeName == SqlTypeName.ANY)) {
+ // one of the type was ANY.
+ // we cannot generate a common type if it differs from other types.
+ throw new TableException("Generic ANY types must have a common type information.")
+ } else {
+ // cannot resolve a common type for different input types
+ None
+ }
+ }
+ }
+
+}
+
+object FlinkTypeFactory {
+
+ private[flink] def typeInfoToSqlTypeName(typeInfo: TypeInformation[_]): SqlTypeName =
+ typeInfo match {
+ case BOOLEAN_TYPE_INFO => BOOLEAN
+ case BYTE_TYPE_INFO => TINYINT
+ case SHORT_TYPE_INFO => SMALLINT
+ case INT_TYPE_INFO => INTEGER
+ case LONG_TYPE_INFO => BIGINT
+ case FLOAT_TYPE_INFO => FLOAT
+ case DOUBLE_TYPE_INFO => DOUBLE
+ case STRING_TYPE_INFO => VARCHAR
+
+ // temporal types
+ case SqlTimeTypeInfo.DATE => DATE
+ case SqlTimeTypeInfo.TIME => TIME
+ case SqlTimeTypeInfo.TIMESTAMP => TIMESTAMP
+ case TimeIntervalTypeInfo.INTERVAL_MONTHS => INTERVAL_YEAR_MONTH
+ case TimeIntervalTypeInfo.INTERVAL_MILLIS => INTERVAL_DAY_SECOND
+
+ case PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO => VARBINARY
+
+ case CHAR_TYPE_INFO | CHAR_VALUE_TYPE_INFO =>
+ throw new TableException("Character type is not supported.")
+
+ case _@t =>
+ throw new TableException(s"Type is not supported: $t")
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/logical/FlinkLogicalDataStreamTableScan.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/logical/FlinkLogicalDataStreamTableScan.scala
new file mode 100644
index 0000000..b16f629
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/logical/FlinkLogicalDataStreamTableScan.scala
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.plan.nodes.logical
+
+import org.apache.flink.table.plan.nodes.FlinkConventions
+import org.apache.flink.table.plan.schema.DataStreamTable
+
+import org.apache.calcite.plan._
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.core.TableScan
+import org.apache.calcite.rel.logical.LogicalTableScan
+import org.apache.calcite.rel.metadata.RelMetadataQuery
+
+import java.util
+
+class FlinkLogicalDataStreamTableScan(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ table: RelOptTable)
+ extends TableScan(cluster, traitSet, table)
+ with FlinkLogicalRel {
+
+ override def copy(traitSet: RelTraitSet, inputs: util.List[RelNode]): RelNode = {
+ new FlinkLogicalDataStreamTableScan(cluster, traitSet, getTable)
+ }
+
+ override def computeSelfCost(planner: RelOptPlanner, mq: RelMetadataQuery): RelOptCost = {
+ val rowCnt = mq.getRowCount(this)
+ val rowSize = mq.getAverageRowSize(this)
+ planner.getCostFactory.makeCost(rowCnt, rowCnt, rowCnt * rowSize)
+ }
+}
+
+class FlinkLogicalDataStreamTableScanConverter
+ extends ConverterRule(
+ classOf[LogicalTableScan],
+ Convention.NONE,
+ FlinkConventions.LOGICAL,
+ "FlinkLogicalDataStreamTableScanConverter") {
+
+ override def matches(call: RelOptRuleCall): Boolean = {
+ val scan: TableScan = call.rel(0)
+ val dataStreamTable = scan.getTable.unwrap(classOf[DataStreamTable[_]])
+ dataStreamTable != null
+ }
+
+ def convert(rel: RelNode): RelNode = {
+ val scan = rel.asInstanceOf[TableScan]
+ val traitSet = rel.getTraitSet.replace(FlinkConventions.LOGICAL)
+ new FlinkLogicalDataStreamTableScan(
+ rel.getCluster,
+ traitSet,
+ scan.getTable
+ )
+ }
+}
+
+object FlinkLogicalDataStreamTableScan {
+ val CONVERTER = new FlinkLogicalDataStreamTableScanConverter
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecBoundedStreamScan.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecBoundedStreamScan.scala
new file mode 100644
index 0000000..2efc468
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecBoundedStreamScan.scala
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.plan.nodes.physical.batch
+
+import org.apache.flink.table.plan.schema.DataStreamTable
+
+import org.apache.calcite.plan._
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.core.TableScan
+import org.apache.calcite.rel.metadata.RelMetadataQuery
+import org.apache.calcite.rel.{RelNode, RelWriter}
+
+import scala.collection.JavaConverters._
+
+/**
+ * Flink RelNode which matches along with StreamTransformation.
+ * It ensures that types without deterministic field order (e.g. POJOs) are not part of
+ * the plan translation.
+ */
+class BatchExecBoundedStreamScan(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ table: RelOptTable,
+ rowRelDataType: RelDataType)
+ extends TableScan(cluster, traitSet, table)
+ with BatchPhysicalRel {
+
+ val boundedStreamTable: DataStreamTable[Any] = getTable.unwrap(classOf[DataStreamTable[Any]])
+
+ override def deriveRowType(): RelDataType = rowRelDataType
+
+ override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+ new BatchExecBoundedStreamScan(cluster, traitSet, getTable, getRowType)
+ }
+
+ override def computeSelfCost(planner: RelOptPlanner, mq: RelMetadataQuery): RelOptCost = {
+ val rowCnt = mq.getRowCount(this)
+ val rowSize = mq.getAverageRowSize(this)
+ planner.getCostFactory.makeCost(rowCnt, rowCnt, rowCnt * rowSize)
+ }
+
+ override def explainTerms(pw: RelWriter): RelWriter = {
+ super.explainTerms(pw)
+ .item("fields", getRowType.getFieldNames.asScala.mkString(", "))
+ }
+
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecDataStreamScan.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecDataStreamScan.scala
new file mode 100644
index 0000000..481b7b5
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecDataStreamScan.scala
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.plan.nodes.physical.stream
+
+import org.apache.flink.table.plan.schema.DataStreamTable
+
+import org.apache.calcite.plan._
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.core.TableScan
+import org.apache.calcite.rel.metadata.RelMetadataQuery
+import org.apache.calcite.rel.{RelNode, RelWriter}
+
+import scala.collection.JavaConverters._
+
+/**
+ * Flink RelNode which matches along with DataStreamSource.
+ * It ensures that types without deterministic field order (e.g. POJOs) are not part of
+ * the plan translation.
+ */
+class StreamExecDataStreamScan(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ table: RelOptTable,
+ relDataType: RelDataType)
+ extends TableScan(cluster, traitSet, table)
+ with StreamPhysicalRel {
+
+ val dataStreamTable: DataStreamTable[Any] = getTable.unwrap(classOf[DataStreamTable[Any]])
+
+ def isAccRetract: Boolean = getTable.unwrap(classOf[DataStreamTable[Any]]).isAccRetract
+
+ override def producesUpdates: Boolean = dataStreamTable.producesUpdates
+
+ override def producesRetractions: Boolean = producesUpdates && isAccRetract
+
+ override def deriveRowType(): RelDataType = relDataType
+
+ override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+ new StreamExecDataStreamScan(cluster, traitSet, getTable, getRowType)
+ }
+
+ override def computeSelfCost(planner: RelOptPlanner, mq: RelMetadataQuery): RelOptCost = {
+ val rowCnt = mq.getRowCount(this)
+ val rowSize = mq.getAverageRowSize(this)
+ planner.getCostFactory.makeCost(rowCnt, rowCnt, rowCnt * rowSize)
+ }
+
+ override def explainTerms(pw: RelWriter): RelWriter = {
+ super.explainTerms(pw)
+ .item("fields", getRowType.getFieldNames.asScala.mkString(", "))
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/batch/BatchExecScanTableRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/batch/BatchExecScanTableRule.scala
new file mode 100644
index 0000000..7acf0b7
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/batch/BatchExecScanTableRule.scala
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.plan.rules.physical.batch
+
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.table.plan.nodes.FlinkConventions
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalDataStreamTableScan
+import org.apache.flink.table.plan.nodes.physical.batch.BatchExecBoundedStreamScan
+import org.apache.flink.table.plan.schema.DataStreamTable
+
+class BatchExecScanTableRule
+ extends ConverterRule(
+ classOf[FlinkLogicalDataStreamTableScan],
+ FlinkConventions.LOGICAL,
+ FlinkConventions.BATCH_PHYSICAL,
+ "BatchExecScanTableRule") {
+
+ /**
+ * If the input is not a DataStreamTable, we want the TableScanRule to match instead
+ */
+ override def matches(call: RelOptRuleCall): Boolean = {
+ val scan: FlinkLogicalDataStreamTableScan = call.rel(0)
+ val dataStreamTable = scan.getTable.unwrap(classOf[DataStreamTable[Any]])
+ dataStreamTable != null
+ }
+
+ def convert(rel: RelNode): RelNode = {
+ val scan = rel.asInstanceOf[FlinkLogicalDataStreamTableScan]
+ val newTrait = rel.getTraitSet.replace(FlinkConventions.BATCH_PHYSICAL)
+ new BatchExecBoundedStreamScan(rel.getCluster, newTrait, scan.getTable, rel.getRowType)
+ }
+}
+
+object BatchExecScanTableRule {
+ val INSTANCE: RelOptRule = new BatchExecScanTableRule
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/stream/StreamExecScanRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/stream/StreamExecScanRule.scala
new file mode 100644
index 0000000..394be4a
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/stream/StreamExecScanRule.scala
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.plan.rules.physical.stream
+
+import org.apache.flink.table.plan.nodes.FlinkConventions
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalDataStreamTableScan
+import org.apache.flink.table.plan.nodes.physical.stream.StreamExecDataStreamScan
+import org.apache.flink.table.plan.schema.DataStreamTable
+
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+
+class StreamExecScanRule
+ extends ConverterRule(
+ classOf[FlinkLogicalDataStreamTableScan],
+ FlinkConventions.LOGICAL,
+ FlinkConventions.STREAM_PHYSICAL,
+ "StreamExecScanRule") {
+
+ override def matches(call: RelOptRuleCall): Boolean = {
+ val scan: FlinkLogicalDataStreamTableScan = call.rel(0)
+ val dataStreamTable = scan.getTable.unwrap(classOf[DataStreamTable[Any]])
+ dataStreamTable != null
+ }
+
+ def convert(rel: RelNode): RelNode = {
+ val scan: FlinkLogicalDataStreamTableScan = rel.asInstanceOf[FlinkLogicalDataStreamTableScan]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConventions.STREAM_PHYSICAL)
+
+ new StreamExecDataStreamScan(
+ rel.getCluster,
+ traitSet,
+ scan.getTable,
+ rel.getRowType
+ )
+ }
+}
+
+object StreamExecScanRule {
+ val INSTANCE: RelOptRule = new StreamExecScanRule
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/DataStreamTable.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/DataStreamTable.scala
new file mode 100644
index 0000000..2fe6f3b
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/DataStreamTable.scala
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.plan.schema
+
+import org.apache.flink.streaming.api.datastream.DataStream
+import org.apache.flink.table.plan.stats.FlinkStatistic
+
+class DataStreamTable[T](
+ val dataStream: DataStream[T],
+ val producesUpdates: Boolean,
+ val isAccRetract: Boolean,
+ override val fieldIndexes: Array[Int],
+ override val fieldNames: Array[String],
+ statistic: FlinkStatistic = FlinkStatistic.UNKNOWN)
+ extends InlineTable[T](dataStream.getType, fieldIndexes, fieldNames, statistic) {
+
+ // This is only used for bounded stream now, we supply default statistic.
+ def this(
+ stream: DataStream[T],
+ fieldIndexes: Array[Int],
+ fieldNames: Array[String]) {
+ this(stream, false, false, fieldIndexes, fieldNames)
+ }
+
+ /**
+ * Creates a copy of this table, changing statistic.
+ *
+ * @param statistic A new FlinkStatistic.
+ * @return Copy of this table, substituting statistic.
+ */
+ override def copy(statistic: FlinkStatistic): FlinkTable =
+ new DataStreamTable(
+ dataStream, producesUpdates, isAccRetract, fieldIndexes, fieldNames, statistic)
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/FlinkTable.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/FlinkTable.scala
new file mode 100644
index 0000000..9c0c3df
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/FlinkTable.scala
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.plan.schema
+
+import org.apache.flink.table.plan.stats.FlinkStatistic
+
+import org.apache.calcite.schema.impl.AbstractTable
+
+/**
+ * Base class for flink table.
+ */
+abstract class FlinkTable extends AbstractTable {
+
+ /**
+ * Restrict return type of statistic to FlinkStatistic.
+ */
+ override def getStatistic: FlinkStatistic = ???
+
+ /**
+ * Creates a copy of this table, changing statistic.
+ *
+ * @param statistic A new FlinkStatistic.
+ * @return Copy of this table, substituting statistic.
+ */
+ def copy(statistic: FlinkStatistic): FlinkTable
+
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/InlineTable.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/InlineTable.scala
new file mode 100644
index 0000000..fce6bcf
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/InlineTable.scala
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.plan.schema
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.common.typeutils.CompositeType
+import org.apache.flink.table.api.TableException
+import org.apache.flink.table.calcite.FlinkTypeFactory
+import org.apache.flink.table.plan.stats.FlinkStatistic
+
+import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory}
+
+abstract class InlineTable[T](
+ val typeInfo: TypeInformation[T],
+ val fieldIndexes: Array[Int],
+ val fieldNames: Array[String],
+ val statistic: FlinkStatistic)
+ extends FlinkTable {
+
+ if (fieldIndexes.length != fieldNames.length) {
+ throw new TableException(
+ s"Number of field names and field indexes must be equal.\n" +
+ s"Number of names is ${fieldNames.length}, number of indexes is ${fieldIndexes.length}.\n" +
+ s"List of column names: ${fieldNames.mkString("[", ", ", "]")}.\n" +
+ s"List of column indexes: ${fieldIndexes.mkString("[", ", ", "]")}.")
+ }
+
+ // check uniqueness of field names
+ if (fieldNames.length != fieldNames.toSet.size) {
+ val duplicateFields = fieldNames
+ // count occurrences of field names
+ .groupBy(identity).mapValues(_.length)
+ // filter for occurrences > 1 and map to field name
+ .filter(g => g._2 > 1).keys
+
+ throw new TableException(
+ s"Field names must be unique.\n" +
+ s"List of duplicate fields: ${duplicateFields.mkString("[", ", ", "]")}.\n" +
+ s"List of all fields: ${fieldNames.mkString("[", ", ", "]")}.")
+ }
+
+ val fieldTypes: Array[TypeInformation[_]] =
+ typeInfo match {
+
+ case ct: CompositeType[_] =>
+ // it is ok to leave out fields
+ if (fieldIndexes.count(_ >= 0) > ct.getArity) {
+ throw new TableException(
+ s"Arity of type (" + ct.getFieldNames.deep + ") " +
+ "must not be greater than number of field names " + fieldNames.deep + ".")
+ }
+ fieldIndexes.map(i => ct.getTypeAt(i).asInstanceOf[TypeInformation[_]])
+
+ case t: TypeInformation[_] =>
+ val cnt = fieldIndexes.length
+ val types = fieldIndexes.map(_.asInstanceOf[TypeInformation[_]])
+ // ensure that the atomic type is matched at most once.
+ if (cnt > 1) {
+ throw new TableException(
+ "Non-composite input type may have only a single field and its index must be 0.")
+ } else {
+ types
+ }
+ }
+
+ override def getRowType(typeFactory: RelDataTypeFactory): RelDataType = {
+ val flinkTypeFactory = typeFactory.asInstanceOf[FlinkTypeFactory]
+ flinkTypeFactory.buildLogicalRowType(fieldNames, fieldTypes)
+ }
+
+ /**
+ * Returns statistics of current table
+ *
+ * @return statistics of current table
+ */
+ override def getStatistic: FlinkStatistic = statistic
+
+ /**
+ * Creates a copy of this table, changing statistic.
+ *
+ * @param statistic A new FlinkStatistic.
+ * @return Copy of this table, substituting statistic.
+ */
+ def copy(statistic: FlinkStatistic): FlinkTable
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/stats/FlinkStatistic.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/stats/FlinkStatistic.scala
new file mode 100644
index 0000000..86833bc
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/stats/FlinkStatistic.scala
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.plan.stats
+
+import org.apache.calcite.rel.{RelCollation, RelDistribution, RelReferentialConstraint}
+import org.apache.calcite.schema.Statistic
+import org.apache.calcite.util.ImmutableBitSet
+
+import java.lang.Double
+import java.util
+
+/**
+ * The class provides statistics for a [[org.apache.flink.table.plan.schema.FlinkTable]].
+ *
+ * TODO: Introduce TableStats after https://github.com/apache/flink/pull/7642 finished
+ */
+class FlinkStatistic extends Statistic {
+
+ /**
+ * Returns the number of rows of the table.
+ *
+ * @return The number of rows of the table.
+ */
+ override def getRowCount: Double = null
+
+ override def getCollations: util.List[RelCollation] = util.Collections.emptyList()
+
+ /**
+ * Returns whether the given columns are a key or a superset of a unique key
+ * of this table.
+ *
+ * Note: Do not call this method!
+ * Use [[org.apache.calcite.rel.metadata.RelMetadataQuery]].areRowsUnique if need.
+ * Because columns in original uniqueKey may not exist in RowType after project pushDown, however
+ * the RowType cannot be available here.
+ *
+ * @param columns Ordinals of key columns
+ * @return if bit mask represents a unique column set; false if not (or
+ * if no metadata is available).
+ */
+ override def isKey(columns: ImmutableBitSet): Boolean = false
+
+ override def getDistribution: RelDistribution = null
+
+ override def getReferentialConstraints: util.List[RelReferentialConstraint] =
+ util.Collections.emptyList()
+}
+
+/**
+ * Methods to create FlinkStatistic.
+ */
+object FlinkStatistic {
+
+ /** Represents a FlinkStatistic that knows nothing about a table */
+ val UNKNOWN: FlinkStatistic = new FlinkStatistic()
+
+}