You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2019/03/08 08:40:22 UTC
[flink] branch master updated: [FLINK-11703][table-planner-blink]
Introduce TableEnvironments and support registerDataStream and sqlQuery
This is an automated email from the ASF dual-hosted git repository.
jark 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 586b890 [FLINK-11703][table-planner-blink] Introduce TableEnvironments and support registerDataStream and sqlQuery
586b890 is described below
commit 586b890237bc2516fa202bfb8c89f5dea4209143
Author: Jark Wu <im...@gmail.com>
AuthorDate: Sun Mar 3 23:58:39 2019 +0800
[FLINK-11703][table-planner-blink] Introduce TableEnvironments and support registerDataStream and sqlQuery
This closes #7923
---
.../flink/table/api/StreamTableEnvironment.scala | 204 ++++++++
.../scala/org/apache/flink/table/api/Table.scala | 41 ++
.../apache/flink/table/api/TableEnvironment.scala | 549 +++++++++++++++++++++
.../table/api/java/StreamTableEnvironment.scala | 176 +++++++
.../table/api/scala/DataStreamConversions.scala | 63 +++
.../table/api/scala/StreamTableEnvironment.scala | 176 +++++++
.../org/apache/flink/table/api/scala/package.scala | 66 +++
.../table/calcite/FlinkCalciteSqlValidator.scala | 94 ++++
.../flink/table/calcite/FlinkPlannerImpl.scala | 224 +++++++++
.../flink/table/calcite/FlinkRelBuilder.scala | 80 +++
.../table/calcite/FlinkRelOptClusterFactory.scala | 41 ++
.../apache/flink/table/plan/schema/RelTable.scala | 46 ++
.../table/typeutils/TimeIndicatorTypeInfo.scala | 59 +++
.../flink/table/api/TableEnvironmentTest.scala | 74 +++
14 files changed, 1893 insertions(+)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
new file mode 100644
index 0000000..38319f0
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
@@ -0,0 +1,204 @@
+/*
+ * 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.api
+
+import org.apache.flink.streaming.api.TimeCharacteristic
+import org.apache.flink.streaming.api.datastream.DataStream
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
+import org.apache.flink.table.plan.schema._
+import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo
+
+import _root_.java.util.concurrent.atomic.AtomicInteger
+
+/**
+ * The base class for stream TableEnvironments.
+ *
+ * A TableEnvironment can be used to:
+ * - convert [[DataStream]] to a [[Table]]
+ * - register a [[DataStream]] as a table in the catalog
+ * - register a [[Table]] in the catalog
+ * - scan a registered table to obtain a [[Table]]
+ * - specify a SQL query on registered tables to obtain a [[Table]]
+ * - convert a [[Table]] into a [[DataStream]]
+ *
+ * @param execEnv The [[StreamExecutionEnvironment]] which is wrapped in this
+ * [[StreamTableEnvironment]].
+ * @param config The [[TableConfig]] of this [[StreamTableEnvironment]].
+ */
+abstract class StreamTableEnvironment(
+ private[flink] val execEnv: StreamExecutionEnvironment,
+ config: TableConfig)
+ extends TableEnvironment(config) {
+
+ // a counter for unique table names
+ private val nameCntr: AtomicInteger = new AtomicInteger(0)
+
+ // the naming pattern for internally registered tables.
+ private val internalNamePattern = "^_StreamTable_[0-9]+$".r
+
+ override def queryConfig: StreamQueryConfig = new StreamQueryConfig
+
+ /**
+ * Checks if the chosen table name is valid.
+ *
+ * @param name The table name to check.
+ */
+ override protected def checkValidTableName(name: String): Unit = {
+ val m = internalNamePattern.findFirstIn(name)
+ m match {
+ case Some(_) =>
+ throw new TableException(s"Illegal Table name. " +
+ s"Please choose a name that does not contain the pattern $internalNamePattern")
+ case None =>
+ }
+ }
+
+ /** Returns a unique table name according to the internal naming pattern. */
+ override protected def createUniqueTableName(): String =
+ "_StreamTable_" + nameCntr.getAndIncrement()
+
+
+ /**
+ * Registers a [[DataStream]] as a table under a given name in the [[TableEnvironment]]'s
+ * catalog.
+ *
+ * @param name The name under which the table is registered in the catalog.
+ * @param dataStream The [[DataStream]] to register as table in the catalog.
+ * @tparam T the type of the [[DataStream]].
+ */
+ protected def registerDataStreamInternal[T](
+ name: String,
+ dataStream: DataStream[T]): Unit = {
+
+ val (fieldNames, fieldIndexes) = getFieldInfo[T](dataStream.getType)
+ val dataStreamTable = new DataStreamTable[T](
+ dataStream,
+ fieldIndexes,
+ fieldNames
+ )
+ registerTableInternal(name, dataStreamTable)
+ }
+
+ /**
+ * Registers a [[DataStream]] as a table under a given name with field names as specified by
+ * field expressions in the [[TableEnvironment]]'s catalog.
+ *
+ * @param name The name under which the table is registered in the catalog.
+ * @param dataStream The [[DataStream]] to register as table in the catalog.
+ * @param fields The field expressions to define the field names of the table.
+ * @tparam T The type of the [[DataStream]].
+ */
+ protected def registerDataStreamInternal[T](
+ name: String,
+ dataStream: DataStream[T],
+ fields: Array[String])
+ : Unit = {
+
+ val streamType = dataStream.getType
+
+ // get field names and types for all non-replaced fields
+ val (fieldNames, fieldIndexes) = getFieldInfo[T](streamType, fields)
+
+ // TODO: validate and extract time attributes after we introduce [Expression],
+ // return None currently
+ val (rowtime, proctime) = (None, None)
+
+ // check if event-time is enabled
+ if (rowtime.isDefined && execEnv.getStreamTimeCharacteristic != TimeCharacteristic.EventTime) {
+ throw new TableException(
+ s"A rowtime attribute requires an EventTime time characteristic in stream environment. " +
+ s"But is: ${execEnv.getStreamTimeCharacteristic}")
+ }
+
+ // adjust field indexes and field names
+ val indexesWithIndicatorFields = adjustFieldIndexes(fieldIndexes, rowtime, proctime)
+ val namesWithIndicatorFields = adjustFieldNames(fieldNames, rowtime, proctime)
+
+ val dataStreamTable = new DataStreamTable[T](
+ dataStream,
+ indexesWithIndicatorFields,
+ namesWithIndicatorFields
+ )
+ registerTableInternal(name, dataStreamTable)
+ }
+
+
+ /**
+ * Injects markers for time indicator fields into the field indexes.
+ *
+ * @param fieldIndexes The field indexes into which the time indicators markers are injected.
+ * @param rowtime An optional rowtime indicator
+ * @param proctime An optional proctime indicator
+ * @return An adjusted array of field indexes.
+ */
+ private def adjustFieldIndexes(
+ fieldIndexes: Array[Int],
+ rowtime: Option[(Int, String)],
+ proctime: Option[(Int, String)]): Array[Int] = {
+
+ // inject rowtime field
+ val withRowtime = rowtime match {
+ case Some(rt) =>
+ fieldIndexes.patch(rt._1, Seq(TimeIndicatorTypeInfo.ROWTIME_STREAM_MARKER), 0)
+ case _ =>
+ fieldIndexes
+ }
+
+ // inject proctime field
+ val withProctime = proctime match {
+ case Some(pt) =>
+ withRowtime.patch(pt._1, Seq(TimeIndicatorTypeInfo.PROCTIME_STREAM_MARKER), 0)
+ case _ =>
+ withRowtime
+ }
+
+ withProctime
+ }
+
+ /**
+ * Injects names of time indicator fields into the list of field names.
+ *
+ * @param fieldNames The array of field names into which the time indicator field names are
+ * injected.
+ * @param rowtime An optional rowtime indicator
+ * @param proctime An optional proctime indicator
+ * @return An adjusted array of field names.
+ */
+ private def adjustFieldNames(
+ fieldNames: Array[String],
+ rowtime: Option[(Int, String)],
+ proctime: Option[(Int, String)]): Array[String] = {
+
+ // inject rowtime field
+ val withRowtime = rowtime match {
+ case Some(rt) => fieldNames.patch(rt._1, Seq(rowtime.get._2), 0)
+ case _ => fieldNames
+ }
+
+ // inject proctime field
+ val withProctime = proctime match {
+ case Some(pt) => withRowtime.patch(pt._1, Seq(proctime.get._2), 0)
+ case _ => withRowtime
+ }
+
+ withProctime
+ }
+
+}
+
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/Table.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/Table.scala
new file mode 100644
index 0000000..8be880f
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/Table.scala
@@ -0,0 +1,41 @@
+/*
+ * 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.api
+
+import org.apache.calcite.rel.RelNode
+
+/**
+ * A Table is the core component of the Table API.
+ * Similar to how the batch and streaming APIs have DataSet and DataStream,
+ * the Table API is built around [[Table]].
+ *
+ * NOTE: This class is only a placeholder to support end-to-end tests for Blink planner.
+ * Will be removed when [[Table]] is moved to "flink-table-api-java" module.
+ *
+ * @param tableEnv The [[TableEnvironment]] to which the table is bound.
+ * @param relNode The Calcite RelNode representation
+ */
+class Table(val tableEnv: TableEnvironment, relNode: RelNode) {
+
+ /**
+ * Returns the Calcite RelNode represent this Table.
+ */
+ def getRelNode: RelNode = relNode
+
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
new file mode 100644
index 0000000..97cf376
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
@@ -0,0 +1,549 @@
+/*
+ * 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.api
+
+import org.apache.calcite.config.Lex
+import org.apache.calcite.jdbc.CalciteSchema
+import org.apache.calcite.plan.RelOptPlanner
+import org.apache.calcite.schema.SchemaPlus
+import org.apache.calcite.schema.impl.AbstractTable
+import org.apache.calcite.sql._
+import org.apache.calcite.sql.parser.SqlParser
+import org.apache.calcite.sql2rel.SqlToRelConverter
+import org.apache.calcite.tools._
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.common.typeutils.CompositeType
+import org.apache.flink.api.java.typeutils.{RowTypeInfo, _}
+import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
+import org.apache.flink.table.calcite.{FlinkPlannerImpl, FlinkRelBuilder, FlinkTypeFactory, FlinkTypeSystem}
+import org.apache.flink.table.plan.cost.FlinkCostFactory
+import org.apache.flink.table.plan.schema.RelTable
+import org.apache.flink.types.Row
+
+import _root_.java.lang.reflect.Modifier
+import _root_.java.util.concurrent.atomic.AtomicInteger
+import _root_.java.util.{Arrays => JArrays}
+
+import _root_.scala.annotation.varargs
+import _root_.scala.collection.JavaConverters._
+
+/**
+ * The abstract base class for batch and stream TableEnvironments.
+ *
+ * @param config The configuration of the TableEnvironment
+ */
+abstract class TableEnvironment(val config: TableConfig) {
+
+ // the catalog to hold all registered and translated tables
+ // we disable caching here to prevent side effects
+ private val internalSchema: CalciteSchema = CalciteSchema.createRootSchema(false, false)
+ private val rootSchema: SchemaPlus = internalSchema.plus()
+
+ // the configuration to create a Calcite planner
+ private lazy val frameworkConfig: FrameworkConfig = Frameworks
+ .newConfigBuilder
+ .defaultSchema(rootSchema)
+ .parserConfig(getSqlParserConfig)
+ .costFactory(new FlinkCostFactory)
+ .typeSystem(new FlinkTypeSystem)
+ .sqlToRelConverterConfig(getSqlToRelConverterConfig)
+ // TODO: introduce ExpressionReducer after codegen
+ // set the executor to evaluate constant expressions
+ // .executor(new ExpressionReducer(config))
+ .build
+
+ // the builder for Calcite RelNodes, Calcite's representation of a relational expression tree.
+ protected lazy val relBuilder: FlinkRelBuilder = FlinkRelBuilder.create(frameworkConfig)
+
+ // the planner instance used to optimize queries of this TableEnvironment
+ private lazy val planner: RelOptPlanner = relBuilder.getPlanner
+
+ private lazy val typeFactory: FlinkTypeFactory = relBuilder.getTypeFactory
+
+ // a counter for unique attribute names
+ private[flink] val attrNameCntr: AtomicInteger = new AtomicInteger(0)
+
+ /** Returns the table config to define the runtime behavior of the Table API. */
+ def getConfig: TableConfig = config
+
+ /** Returns the [[QueryConfig]] depends on the concrete type of this TableEnvironment. */
+ private[flink] def queryConfig: QueryConfig
+
+ /**
+ * Returns the SqlToRelConverter config.
+ */
+ protected def getSqlToRelConverterConfig: SqlToRelConverter.Config = {
+ SqlToRelConverter.configBuilder()
+ .withTrimUnusedFields(false)
+ .withConvertTableAccess(false)
+ .withInSubQueryThreshold(Integer.MAX_VALUE)
+ .build()
+ }
+
+ /**
+ * Returns the SQL parser config for this environment including a custom Calcite configuration.
+ */
+ protected def getSqlParserConfig: SqlParser.Config = {
+ // we use Java lex because back ticks are easier than double quotes in programming
+ // and cases are preserved
+ SqlParser
+ .configBuilder()
+ .setLex(Lex.JAVA)
+ .build()
+ }
+
+ /**
+ * Registers a [[Table]] under a unique name in the TableEnvironment's catalog.
+ * Registered tables can be referenced in SQL queries.
+ *
+ * @param name The name under which the table will be registered.
+ * @param table The table to register.
+ */
+ def registerTable(name: String, table: Table): Unit = {
+
+ // check that table belongs to this table environment
+ if (table.tableEnv != this) {
+ throw new TableException(
+ "Only tables that belong to this TableEnvironment can be registered.")
+ }
+
+ checkValidTableName(name)
+ val tableTable = new RelTable(table.getRelNode)
+ registerTableInternal(name, tableTable)
+ }
+
+ /**
+ * Replaces a registered Table with another Table under the same name.
+ * We use this method to replace a [[org.apache.flink.table.plan.schema.DataStreamTable]]
+ * with a [[org.apache.calcite.schema.TranslatableTable]].
+ *
+ * @param name Name of the table to replace.
+ * @param table The table that replaces the previous table.
+ */
+ protected def replaceRegisteredTable(name: String, table: AbstractTable): Unit = {
+ if (isRegistered(name)) {
+ rootSchema.add(name, table)
+ } else {
+ throw new TableException(s"Table \'$name\' is not registered.")
+ }
+ }
+
+ /**
+ * Scans a registered table and returns the resulting [[Table]].
+ *
+ * A table to scan must be registered in the TableEnvironment. It can be either directly
+ * registered as bounded or unbounded DataStream, or Table.
+ *
+ * Examples:
+ *
+ * - Scanning a directly registered table
+ * {{{
+ * val tab: Table = tableEnv.scan("tableName")
+ * }}}
+ *
+ * - Scanning a table from a registered catalog
+ * {{{
+ * val tab: Table = tableEnv.scan("catalogName", "dbName", "tableName")
+ * }}}
+ *
+ * @param tablePath The path of the table to scan.
+ * @throws TableException if no table is found using the given table path.
+ * @return The resulting [[Table]].
+ */
+ @throws[TableException]
+ @varargs
+ def scan(tablePath: String*): Table = {
+ scanInternal(tablePath.toArray) match {
+ case Some(table) => table
+ case None => throw new TableException(s"Table '${tablePath.mkString(".")}' was not found.")
+ }
+ }
+
+ private[flink] def scanInternal(tablePath: Array[String]): Option[Table] = {
+ require(tablePath != null && !tablePath.isEmpty, "tablePath must not be null or empty.")
+ val schemaPaths = tablePath.slice(0, tablePath.length - 1)
+ val schema = getSchema(schemaPaths)
+ if (schema != null) {
+ val tableName = tablePath(tablePath.length - 1)
+ val table = schema.getTable(tableName)
+ if (table != null) {
+ val scan = relBuilder.scan(JArrays.asList(tablePath: _*)).build()
+ return Some(new Table(this, scan))
+ }
+ }
+ None
+ }
+
+ private def getSchema(schemaPath: Array[String]): SchemaPlus = {
+ var schema = rootSchema
+ for (schemaName <- schemaPath) {
+ schema = schema.getSubSchema(schemaName)
+ if (schema == null) {
+ return schema
+ }
+ }
+ schema
+ }
+
+ /**
+ * Gets the names of all tables registered in this environment.
+ *
+ * @return A list of the names of all registered tables.
+ */
+ def listTables(): Array[String] = {
+ rootSchema.getTableNames.asScala.toArray
+ }
+
+ /**
+ * Returns completion hints for the given statement at the given cursor position.
+ * The completion happens case insensitively.
+ *
+ * @param statement Partial or slightly incorrect SQL statement
+ * @param position cursor position
+ * @return completion hints that fit at the current cursor position
+ */
+ def getCompletionHints(statement: String, position: Int): Array[String] = {
+ val planner = new FlinkPlannerImpl(
+ getFrameworkConfig,
+ getPlanner,
+ getTypeFactory)
+ planner.getCompletionHints(statement, position)
+ }
+
+ /**
+ * Evaluates a SQL query on registered tables and retrieves the result as a [[Table]].
+ *
+ * All tables referenced by the query must be registered in the TableEnvironment.
+ * A [[Table]] is automatically registered when its [[toString]] method is called, for example
+ * when it is embedded into a String.
+ * Hence, SQL queries can directly reference a [[Table]] as follows:
+ *
+ * {{{
+ * val table: Table = ...
+ * // the table is not registered to the table environment
+ * tEnv.sqlQuery(s"SELECT * FROM $table")
+ * }}}
+ *
+ * @param query The SQL query to evaluate.
+ * @return The result of the query as Table
+ */
+ def sqlQuery(query: String): Table = {
+ val planner = new FlinkPlannerImpl(getFrameworkConfig, getPlanner, getTypeFactory)
+ // parse the sql query
+ val parsed = planner.parse(query)
+ if (null != parsed && parsed.getKind.belongsTo(SqlKind.QUERY)) {
+ // validate the sql query
+ val validated = planner.validate(parsed)
+ // transform to a relational tree
+ val relational = planner.rel(validated)
+ new Table(this, relational.rel)
+ } else {
+ throw new TableException(
+ "Unsupported SQL query! sqlQuery() only accepts SQL queries of type " +
+ "SELECT, UNION, INTERSECT, EXCEPT, VALUES, and ORDER_BY.")
+ }
+ }
+
+ /**
+ * Registers a Calcite [[AbstractTable]] in the TableEnvironment's catalog.
+ *
+ * @param name The name under which the table will be registered.
+ * @param table The table to register in the catalog
+ * @throws TableException if another table is registered under the provided name.
+ */
+ @throws[TableException]
+ protected def registerTableInternal(name: String, table: AbstractTable): Unit = {
+ if (isRegistered(name)) {
+ throw new TableException(s"Table \'$name\' already exists. " +
+ s"Please, choose a different name.")
+ } else {
+ rootSchema.add(name, table)
+ }
+ }
+
+ /** Returns a unique table name according to the internal naming pattern. */
+ protected def createUniqueTableName(): String
+
+ /**
+ * Checks if the chosen table name is valid.
+ *
+ * @param name The table name to check.
+ */
+ protected def checkValidTableName(name: String): Unit
+
+ /**
+ * Checks if a table is registered under the given name.
+ *
+ * @param name The table name to check.
+ * @return true, if a table is registered under the name, false otherwise.
+ */
+ protected[flink] def isRegistered(name: String): Boolean = {
+ rootSchema.getTableNames.contains(name)
+ }
+
+ /**
+ * Get a table from either internal or external catalogs.
+ *
+ * @param name The name of the table.
+ * @return The table registered either internally or externally, None otherwise.
+ */
+ protected def getTable(name: String): Option[org.apache.calcite.schema.Table] = {
+
+ // recursively fetches a table from a schema.
+ def getTableFromSchema(
+ schema: SchemaPlus,
+ path: List[String]): Option[org.apache.calcite.schema.Table] = {
+
+ path match {
+ case tableName :: Nil =>
+ // look up table
+ Option(schema.getTable(tableName))
+ case subschemaName :: remain =>
+ // look up subschema
+ val subschema = Option(schema.getSubSchema(subschemaName))
+ subschema match {
+ case Some(s) =>
+ // search for table in subschema
+ getTableFromSchema(s, remain)
+ case None =>
+ // subschema does not exist
+ None
+ }
+ }
+ }
+
+ val pathNames = name.split('.').toList
+ getTableFromSchema(rootSchema, pathNames)
+ }
+
+ /** Returns a unique temporary attribute name. */
+ private[flink] def createUniqueAttributeName(): String = {
+ "TMP_" + attrNameCntr.getAndIncrement()
+ }
+
+ /** Returns the [[FlinkRelBuilder]] of this TableEnvironment. */
+ private[flink] def getRelBuilder: FlinkRelBuilder = {
+ relBuilder
+ }
+
+ /** Returns the Calcite [[org.apache.calcite.plan.RelOptPlanner]] of this TableEnvironment. */
+ private[flink] def getPlanner: RelOptPlanner = {
+ planner
+ }
+
+ /** Returns the [[FlinkTypeFactory]] of this TableEnvironment. */
+ private[flink] def getTypeFactory: FlinkTypeFactory = {
+ typeFactory
+ }
+
+ /** Returns the Calcite [[FrameworkConfig]] of this TableEnvironment. */
+ private[flink] def getFrameworkConfig: FrameworkConfig = {
+ frameworkConfig
+ }
+
+ /**
+ * Reference input fields by name:
+ * All fields in the schema definition are referenced by name
+ * (and possibly renamed using an alias (as). In this mode, fields can be reordered and
+ * projected out. Moreover, we can define proctime and rowtime attributes at arbitrary
+ * positions using arbitrary names (except those that exist in the result schema). This mode
+ * can be used for any input type, including POJOs.
+ *
+ * Reference input fields by position:
+ * In this mode, fields are simply renamed. Event-time attributes can
+ * replace the field on their position in the input data (if it is of correct type) or be
+ * appended at the end. Proctime attributes must be appended at the end. This mode can only be
+ * used if the input type has a defined field order (tuple, case class, Row) and no of fields
+ * references a field of the input type.
+ */
+ // TODO: we should support Expression fields after we introduce [Expression]
+ protected def isReferenceByPosition(ct: CompositeType[_], fields: Array[String]): Boolean = {
+ if (!ct.isInstanceOf[TupleTypeInfoBase[_]]) {
+ return false
+ }
+
+ val inputNames = ct.getFieldNames
+
+ // Use the by-position mode if no of the fields exists in the input.
+ // This prevents confusing cases like ('f2, 'f0, 'myName) for a Tuple3 where fields are renamed
+ // by position but the user might assume reordering instead of renaming.
+ fields.forall(!inputNames.contains(_))
+ }
+
+ /**
+ * Returns field names and field positions for a given [[TypeInformation]].
+ *
+ * @param inputType The TypeInformation extract the field names and positions from.
+ * @tparam A The type of the TypeInformation.
+ * @return A tuple of two arrays holding the field names and corresponding field positions.
+ */
+ protected[flink] def getFieldInfo[A](inputType: TypeInformation[A]):
+ (Array[String], Array[Int]) = {
+
+ if (inputType.isInstanceOf[GenericTypeInfo[A]] && inputType.getTypeClass == classOf[Row]) {
+ throw new TableException(
+ "An input of GenericTypeInfo<Row> cannot be converted to Table. " +
+ "Please specify the type of the input with a RowTypeInfo.")
+ } else {
+ (TableEnvironment.getFieldNames(inputType), TableEnvironment.getFieldIndices(inputType))
+ }
+ }
+
+ /**
+ * Returns field names and field positions for a given [[TypeInformation]] and [[Array]] of
+ * field names. It does not handle time attributes.
+ *
+ * @param inputType The [[TypeInformation]] against which the field names are referenced.
+ * @param fields The fields that define the field names.
+ * @tparam A The type of the TypeInformation.
+ * @return A tuple of two arrays holding the field names and corresponding field positions.
+ */
+ // TODO: we should support Expression fields after we introduce [Expression]
+ protected def getFieldInfo[A](
+ inputType: TypeInformation[A],
+ fields: Array[String])
+ : (Array[String], Array[Int]) = {
+
+ TableEnvironment.validateType(inputType)
+
+ def referenceByName(name: String, ct: CompositeType[_]): Option[Int] = {
+ val inputIdx = ct.getFieldIndex(name)
+ if (inputIdx < 0) {
+ throw new TableException(s"$name is not a field of type $ct. " +
+ s"Expected: ${ct.getFieldNames.mkString(", ")}")
+ } else {
+ Some(inputIdx)
+ }
+ }
+
+ val indexedNames: Array[(Int, String)] = inputType match {
+
+ case g: GenericTypeInfo[A] if g.getTypeClass == classOf[Row] =>
+ throw new TableException(
+ "An input of GenericTypeInfo<Row> cannot be converted to Table. " +
+ "Please specify the type of the input with a RowTypeInfo.")
+
+ case t: TupleTypeInfoBase[A] if t.isInstanceOf[TupleTypeInfo[A]] ||
+ t.isInstanceOf[CaseClassTypeInfo[A]] || t.isInstanceOf[RowTypeInfo] =>
+
+ // determine schema definition mode (by position or by name)
+ val isRefByPos = isReferenceByPosition(t, fields)
+
+ fields.zipWithIndex flatMap { case (name, idx) =>
+ if (isRefByPos) {
+ Some((idx, name))
+ } else {
+ referenceByName(name, t).map((_, name))
+ }
+ }
+
+ case p: PojoTypeInfo[A] =>
+ fields flatMap { name =>
+ referenceByName(name, p).map((_, name))
+ }
+
+ case _: TypeInformation[_] => // atomic or other custom type information
+ if (fields.length > 1) {
+ // only accept the first field for an atomic type
+ throw new TableException("Only accept one field to reference an atomic type.")
+ }
+ // first field reference is mapped to atomic type
+ Array((0, fields(0)))
+ }
+
+ val (fieldIndexes, fieldNames) = indexedNames.unzip
+
+ if (fieldNames.contains("*")) {
+ throw new TableException("Field name can not be '*'.")
+ }
+
+ (fieldNames, fieldIndexes)
+ }
+
+}
+
+/**
+ * Object to instantiate a [[TableEnvironment]] depending on the batch or stream execution
+ * environment.
+ */
+object TableEnvironment {
+
+ /**
+ * Returns field names for a given [[TypeInformation]].
+ *
+ * @param inputType The TypeInformation extract the field names.
+ * @tparam A The type of the TypeInformation.
+ * @return An array holding the field names
+ */
+ def getFieldNames[A](inputType: TypeInformation[A]): Array[String] = {
+ validateType(inputType)
+
+ val fieldNames: Array[String] = inputType match {
+ case t: CompositeType[_] => t.getFieldNames
+ case _: TypeInformation[_] => Array("f0")
+ }
+
+ if (fieldNames.contains("*")) {
+ throw new TableException("Field name can not be '*'.")
+ }
+
+ fieldNames
+ }
+
+ /**
+ * Validate if class represented by the typeInfo is static and globally accessible
+ * @param typeInfo type to check
+ * @throws TableException if type does not meet these criteria
+ */
+ def validateType(typeInfo: TypeInformation[_]): Unit = {
+ val clazz = typeInfo.getTypeClass
+ if ((clazz.isMemberClass && !Modifier.isStatic(clazz.getModifiers)) ||
+ !Modifier.isPublic(clazz.getModifiers) ||
+ clazz.getCanonicalName == null) {
+ throw new TableException(
+ s"Class '$clazz' described in type information '$typeInfo' must be " +
+ s"static and globally accessible.")
+ }
+ }
+
+ /**
+ * Returns field indexes for a given [[TypeInformation]].
+ *
+ * @param inputType The TypeInformation extract the field positions from.
+ * @return An array holding the field positions
+ */
+ def getFieldIndices(inputType: TypeInformation[_]): Array[Int] = {
+ getFieldNames(inputType).indices.toArray
+ }
+
+ /**
+ * Returns field types for a given [[TypeInformation]].
+ *
+ * @param inputType The TypeInformation to extract field types from.
+ * @return An array holding the field types.
+ */
+ def getFieldTypes(inputType: TypeInformation[_]): Array[TypeInformation[_]] = {
+ validateType(inputType)
+
+ inputType match {
+ case ct: CompositeType[_] => 0.until(ct.getArity).map(i => ct.getTypeAt(i)).toArray
+ case t: TypeInformation[_] => Array(t.asInstanceOf[TypeInformation[_]])
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/java/StreamTableEnvironment.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/java/StreamTableEnvironment.scala
new file mode 100644
index 0000000..b47ff69
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/java/StreamTableEnvironment.scala
@@ -0,0 +1,176 @@
+/*
+ * 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.api.java
+
+import org.apache.flink.streaming.api.datastream.DataStream
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
+import org.apache.flink.table.api._
+
+/**
+ * The [[TableEnvironment]] for a Java [[StreamExecutionEnvironment]] that works with
+ * [[DataStream]]s.
+ *
+ * A TableEnvironment can be used to:
+ * - convert a [[DataStream]] to a [[Table]]
+ * - register a [[DataStream]] in the [[TableEnvironment]]'s catalog
+ * - register a [[Table]] in the [[TableEnvironment]]'s catalog
+ * - scan a registered table to obtain a [[Table]]
+ * - specify a SQL query on registered tables to obtain a [[Table]]
+ * - convert a [[Table]] into a [[DataStream]]
+ * - explain the AST and execution plan of a [[Table]]
+ *
+ * @param execEnv The Java [[StreamExecutionEnvironment]] of the TableEnvironment.
+ * @param config The configuration of the TableEnvironment.
+ *
+ * @deprecated This constructor will be removed. Use StreamTableEnvironment.create() instead.
+ */
+class StreamTableEnvironment @Deprecated() (
+ execEnv: StreamExecutionEnvironment,
+ config: TableConfig)
+ extends org.apache.flink.table.api.StreamTableEnvironment(execEnv, config) {
+
+ /**
+ * Converts the given [[DataStream]] into a [[Table]].
+ *
+ * The field names of the [[Table]] are automatically derived from the type of the
+ * [[DataStream]].
+ *
+ * @param dataStream The [[DataStream]] to be converted.
+ * @tparam T The type of the [[DataStream]].
+ * @return The converted [[Table]].
+ */
+ def fromDataStream[T](dataStream: DataStream[T]): Table = {
+ val name = createUniqueTableName()
+ registerDataStreamInternal(name, dataStream)
+ scan(name)
+ }
+
+ /**
+ * Converts the given [[DataStream]] into a [[Table]] with specified field names.
+ *
+ * Example:
+ *
+ * {{{
+ * DataStream<Tuple2<String, Long>> stream = ...
+ * Table tab = tableEnv.fromDataStream(stream, "a, b")
+ * }}}
+ *
+ * @param dataStream The [[DataStream]] to be converted.
+ * @param fields The field names of the resulting [[Table]].
+ * @tparam T The type of the [[DataStream]].
+ * @return The converted [[Table]].
+ */
+ def fromDataStream[T](dataStream: DataStream[T], fields: String): Table = {
+ // TODO: use ExpressionParser instead after we introduce [Expression]
+ val exprs = fields.split(",").map(_.trim)
+
+ val name = createUniqueTableName()
+ registerDataStreamInternal(name, dataStream, exprs)
+ scan(name)
+ }
+
+ /**
+ * Registers the given [[DataStream]] as table in the
+ * [[TableEnvironment]]'s catalog.
+ * Registered tables can be referenced in SQL queries.
+ *
+ * The field names of the [[Table]] are automatically derived
+ * from the type of the [[DataStream]].
+ *
+ * @param name The name under which the [[DataStream]] is registered in the catalog.
+ * @param dataStream The [[DataStream]] to register.
+ * @tparam T The type of the [[DataStream]] to register.
+ */
+ def registerDataStream[T](name: String, dataStream: DataStream[T]): Unit = {
+
+ checkValidTableName(name)
+ registerDataStreamInternal(name, dataStream)
+ }
+
+ /**
+ * Registers the given [[DataStream]] as table with specified field names in the
+ * [[TableEnvironment]]'s catalog.
+ * Registered tables can be referenced in SQL queries.
+ *
+ * Example:
+ *
+ * {{{
+ * DataStream<Tuple2<String, Long>> set = ...
+ * tableEnv.registerDataStream("myTable", set, "a, b")
+ * }}}
+ *
+ * @param name The name under which the [[DataStream]] is registered in the catalog.
+ * @param dataStream The [[DataStream]] to register.
+ * @param fields The field names of the registered table.
+ * @tparam T The type of the [[DataStream]] to register.
+ */
+ def registerDataStream[T](name: String, dataStream: DataStream[T], fields: String): Unit = {
+ // TODO: use ExpressionParser instead after we introduce [Expression]
+ val exprs = fields.split(",").map(_.trim)
+
+ checkValidTableName(name)
+ registerDataStreamInternal(name, dataStream, exprs)
+ }
+
+}
+
+object StreamTableEnvironment {
+
+ /**
+ * The [[TableEnvironment]] for a Java [[StreamExecutionEnvironment]] that works with
+ * [[DataStream]]s.
+ *
+ * A TableEnvironment can be used to:
+ * - convert a [[DataStream]] to a [[Table]]
+ * - register a [[DataStream]] in the [[TableEnvironment]]'s catalog
+ * - register a [[Table]] in the [[TableEnvironment]]'s catalog
+ * - scan a registered table to obtain a [[Table]]
+ * - specify a SQL query on registered tables to obtain a [[Table]]
+ * - convert a [[Table]] into a [[DataStream]]
+ * - explain the AST and execution plan of a [[Table]]
+ *
+ * @param executionEnvironment The Java [[StreamExecutionEnvironment]] of the TableEnvironment.
+ */
+ def create(executionEnvironment: StreamExecutionEnvironment):
+ StreamTableEnvironment = {
+ new StreamTableEnvironment(executionEnvironment, new TableConfig())
+ }
+
+ /**
+ * The [[TableEnvironment]] for a Java [[StreamExecutionEnvironment]] that works with
+ * [[DataStream]]s.
+ *
+ * A TableEnvironment can be used to:
+ * - convert a [[DataStream]] to a [[Table]]
+ * - register a [[DataStream]] in the [[TableEnvironment]]'s catalog
+ * - register a [[Table]] in the [[TableEnvironment]]'s catalog
+ * - scan a registered table to obtain a [[Table]]
+ * - specify a SQL query on registered tables to obtain a [[Table]]
+ * - convert a [[Table]] into a [[DataStream]]
+ * - explain the AST and execution plan of a [[Table]]
+ *
+ * @param executionEnvironment The Java [[StreamExecutionEnvironment]] of the TableEnvironment.
+ * @param tableConfig The configuration of the TableEnvironment.
+ */
+ def create(
+ executionEnvironment: StreamExecutionEnvironment,
+ tableConfig: TableConfig): StreamTableEnvironment = {
+
+ new StreamTableEnvironment(executionEnvironment, tableConfig)
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/scala/DataStreamConversions.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/scala/DataStreamConversions.scala
new file mode 100644
index 0000000..881330f
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/scala/DataStreamConversions.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.api.scala
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.streaming.api.scala.DataStream
+import org.apache.flink.table.api.Table
+
+/**
+ * Holds methods to convert a [[DataStream]] into a [[Table]].
+ *
+ * @param dataStream The [[DataStream]] to convert.
+ * @param inputType The [[TypeInformation]] for the type of the [[DataStream]].
+ * @tparam T The type of the [[DataStream]].
+ */
+class DataStreamConversions[T](dataStream: DataStream[T], inputType: TypeInformation[T]) {
+
+ /**
+ * Converts the [[DataStream]] into a [[Table]].
+ *
+ * The field name of the new [[Table]] can be specified like this:
+ *
+ * {{{
+ * val env = StreamExecutionEnvironment.getExecutionEnvironment
+ * val tEnv = TableEnvironment.getTableEnvironment(env)
+ *
+ * val stream: DataStream[(String, Int)] = ...
+ * val table = stream.toTable(tEnv, 'name, 'amount)
+ * }}}
+ *
+ * If not explicitly specified, field names are automatically extracted from the type of
+ * the [[DataStream]].
+ *
+ * @param tableEnv The [[StreamTableEnvironment]] in which the new [[Table]] is created.
+ * @param fields The field names of the new [[Table]] (optional).
+ * @return The resulting [[Table]].
+ */
+ // TODO: Change fields type to `Expression*` after introducing [Expression]
+ def toTable(tableEnv: StreamTableEnvironment, fields: Symbol*): Table = {
+ if (fields.isEmpty) {
+ tableEnv.fromDataStream(dataStream)
+ } else {
+ tableEnv.fromDataStream(dataStream, fields: _*)
+ }
+ }
+
+}
+
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/scala/StreamTableEnvironment.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/scala/StreamTableEnvironment.scala
new file mode 100644
index 0000000..04df0cf
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/scala/StreamTableEnvironment.scala
@@ -0,0 +1,176 @@
+/*
+ * 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.api.scala
+
+import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
+import org.apache.flink.table.api.{Table, TableConfig, TableEnvironment}
+
+/**
+ * The [[TableEnvironment]] for a Scala [[StreamExecutionEnvironment]] that works with
+ * [[DataStream]]s.
+ *
+ * A TableEnvironment can be used to:
+ * - convert a [[DataStream]] to a [[Table]]
+ * - register a [[DataStream]] in the [[TableEnvironment]]'s catalog
+ * - register a [[Table]] in the [[TableEnvironment]]'s catalog
+ * - scan a registered table to obtain a [[Table]]
+ * - specify a SQL query on registered tables to obtain a [[Table]]
+ * - convert a [[Table]] into a [[DataStream]]
+ * - explain the AST and execution plan of a [[Table]]
+ *
+ * @param execEnv The Scala [[StreamExecutionEnvironment]] of the TableEnvironment.
+ * @param config The configuration of the TableEnvironment.
+ */
+class StreamTableEnvironment @deprecated(
+ "This constructor will be removed. Use StreamTableEnvironment.create() instead.",
+ "1.8.0") (
+ execEnv: StreamExecutionEnvironment,
+ config: TableConfig)
+ extends org.apache.flink.table.api.StreamTableEnvironment(
+ execEnv.getWrappedStreamExecutionEnvironment,
+ config) {
+
+ /**
+ * Converts the given [[DataStream]] into a [[Table]].
+ *
+ * The field names of the [[Table]] are automatically derived from the type of the
+ * [[DataStream]].
+ *
+ * @param dataStream The [[DataStream]] to be converted.
+ * @tparam T The type of the [[DataStream]].
+ * @return The converted [[Table]].
+ */
+ def fromDataStream[T](dataStream: DataStream[T]): Table = {
+
+ val name = createUniqueTableName()
+ registerDataStreamInternal(name, dataStream.javaStream)
+ scan(name)
+ }
+
+ /**
+ * Converts the given [[DataStream]] into a [[Table]] with specified field names.
+ *
+ * Example:
+ *
+ * {{{
+ * val stream: DataStream[(String, Long)] = ...
+ * val tab: Table = tableEnv.fromDataStream(stream, 'a, 'b)
+ * }}}
+ *
+ * @param dataStream The [[DataStream]] to be converted.
+ * @param fields The field names of the resulting [[Table]].
+ * @tparam T The type of the [[DataStream]].
+ * @return The converted [[Table]].
+ */
+ // TODO: Change fields type to `Expression*` after introducing [Expression]
+ def fromDataStream[T](dataStream: DataStream[T], fields: Symbol*): Table = {
+ val exprs = fields.map(_.name).toArray
+ val name = createUniqueTableName()
+ registerDataStreamInternal(name, dataStream.javaStream, exprs)
+ scan(name)
+ }
+
+ /**
+ * Registers the given [[DataStream]] as table in the
+ * [[TableEnvironment]]'s catalog.
+ * Registered tables can be referenced in SQL queries.
+ *
+ * The field names of the [[Table]] are automatically derived
+ * from the type of the [[DataStream]].
+ *
+ * @param name The name under which the [[DataStream]] is registered in the catalog.
+ * @param dataStream The [[DataStream]] to register.
+ * @tparam T The type of the [[DataStream]] to register.
+ */
+ def registerDataStream[T](name: String, dataStream: DataStream[T]): Unit = {
+
+ checkValidTableName(name)
+ registerDataStreamInternal(name, dataStream.javaStream)
+ }
+
+ /**
+ * Registers the given [[DataStream]] as table with specified field names in the
+ * [[TableEnvironment]]'s catalog.
+ * Registered tables can be referenced in SQL queries.
+ *
+ * Example:
+ *
+ * {{{
+ * val set: DataStream[(String, Long)] = ...
+ * tableEnv.registerDataStream("myTable", set, 'a, 'b)
+ * }}}
+ *
+ * @param name The name under which the [[DataStream]] is registered in the catalog.
+ * @param dataStream The [[DataStream]] to register.
+ * @param fields The field names of the registered table.
+ * @tparam T The type of the [[DataStream]] to register.
+ */
+ // TODO: Change fields type to `Expression*` after introducing [Expression]
+ def registerDataStream[T](name: String, dataStream: DataStream[T], fields: Symbol*): Unit = {
+ val exprs = fields.map(_.name).toArray
+ checkValidTableName(name)
+ registerDataStreamInternal(name, dataStream.javaStream, exprs)
+ }
+
+
+}
+
+object StreamTableEnvironment {
+
+ /**
+ * The [[TableEnvironment]] for a Scala [[StreamExecutionEnvironment]] that works with
+ * [[DataStream]]s.
+ *
+ * A TableEnvironment can be used to:
+ * - convert a [[DataStream]] to a [[Table]]
+ * - register a [[DataStream]] in the [[TableEnvironment]]'s catalog
+ * - register a [[Table]] in the [[TableEnvironment]]'s catalog
+ * - scan a registered table to obtain a [[Table]]
+ * - specify a SQL query on registered tables to obtain a [[Table]]
+ * - convert a [[Table]] into a [[DataStream]]
+ * - explain the AST and execution plan of a [[Table]]
+ *
+ * @param executionEnvironment The Scala [[StreamExecutionEnvironment]] of the TableEnvironment.
+ */
+ def create(executionEnvironment: StreamExecutionEnvironment): StreamTableEnvironment = {
+ new StreamTableEnvironment(executionEnvironment, new TableConfig())
+ }
+
+ /**
+ * The [[TableEnvironment]] for a Scala [[StreamExecutionEnvironment]] that works with
+ * [[DataStream]]s.
+ *
+ * A TableEnvironment can be used to:
+ * - convert a [[DataStream]] to a [[Table]]
+ * - register a [[DataStream]] in the [[TableEnvironment]]'s catalog
+ * - register a [[Table]] in the [[TableEnvironment]]'s catalog
+ * - scan a registered table to obtain a [[Table]]
+ * - specify a SQL query on registered tables to obtain a [[Table]]
+ * - convert a [[Table]] into a [[DataStream]]
+ * - explain the AST and execution plan of a [[Table]]
+ *
+ * @param executionEnvironment The Scala [[StreamExecutionEnvironment]] of the TableEnvironment.
+ * @param tableConfig The configuration of the TableEnvironment.
+ */
+ def create(
+ executionEnvironment: StreamExecutionEnvironment,
+ tableConfig: TableConfig): StreamTableEnvironment = {
+
+ new StreamTableEnvironment(executionEnvironment, tableConfig)
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/scala/package.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/scala/package.scala
new file mode 100644
index 0000000..20fbe8e
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/scala/package.scala
@@ -0,0 +1,66 @@
+/*
+ * 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.api
+
+import org.apache.flink.streaming.api.scala.DataStream
+
+import _root_.scala.language.implicitConversions
+
+/**
+ * == Table API (Scala) ==
+ *
+ * Importing this package with:
+ *
+ * {{{
+ * import org.apache.flink.table.api.scala._
+ * }}}
+ *
+ * imports implicit conversions for converting a [[DataStream]] to a
+ * [[Table]]. This can be used to perform SQL-like queries on data. Please have
+ * a look at [[Table]] to see which operations are supported to see how an
+ * expression can be specified.
+ *
+ * When writing a query you can use Scala Symbols to refer to field names. One would
+ * refer to field `a` by writing `'a`. Sometimes it is necessary to manually convert a
+ * Scala literal to an expression literal, in those cases use `.toExpr`, as in `3.toExpr`.
+ *
+ * Example:
+ *
+ * {{{
+ * import org.apache.flink.api.scala._
+ * import org.apache.flink.table.api.scala._
+ *
+ * val env = ExecutionEnvironment.getExecutionEnvironment
+ * val tEnv = TableEnvironment.getTableEnvironment(env)
+ *
+ * val input: DataSet[(String, Int)] = env.fromElements(("Hello", 2), ("Hello", 5), ("Ciao", 3))
+ * val result = input
+ * .toTable(tEnv, 'word, 'count)
+ * .groupBy('word)
+ * .select('word, 'count.avg)
+ *
+ * result.print()
+ * }}}
+ *
+ */
+package object scala {
+
+ implicit def dataStream2DataStreamConversions[T](set: DataStream[T]): DataStreamConversions[T] = {
+ new DataStreamConversions[T](set, set.dataType)
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkCalciteSqlValidator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkCalciteSqlValidator.scala
new file mode 100644
index 0000000..4553fb4
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkCalciteSqlValidator.scala
@@ -0,0 +1,94 @@
+/*
+ * 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.calcite.adapter.java.JavaTypeFactory
+import org.apache.calcite.prepare.CalciteCatalogReader
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.sql._
+import org.apache.calcite.sql.`type`.SqlTypeName
+import org.apache.calcite.sql.validate.{SqlConformanceEnum, SqlValidatorImpl, SqlValidatorScope}
+import org.apache.calcite.util.Static
+import org.apache.flink.table.api.ValidationException
+
+/**
+ * This is a copy of Calcite's CalciteSqlValidator to use with [[FlinkPlannerImpl]].
+ */
+class FlinkCalciteSqlValidator(
+ opTab: SqlOperatorTable,
+ catalogReader: CalciteCatalogReader,
+ factory: JavaTypeFactory)
+ extends SqlValidatorImpl(
+ opTab,
+ catalogReader,
+ factory,
+ SqlConformanceEnum.DEFAULT) {
+
+ override def getLogicalSourceRowType(
+ sourceRowType: RelDataType,
+ insert: SqlInsert): RelDataType = {
+ typeFactory.asInstanceOf[JavaTypeFactory].toSql(sourceRowType)
+ }
+
+ override def getLogicalTargetRowType(
+ targetRowType: RelDataType,
+ insert: SqlInsert): RelDataType = {
+ typeFactory.asInstanceOf[JavaTypeFactory].toSql(targetRowType)
+ }
+
+ override def validateLiteral(literal: SqlLiteral): Unit = {
+ literal.getTypeName match {
+ case SqlTypeName.DECIMAL =>
+ val bd = literal.getValue.asInstanceOf[java.math.BigDecimal]
+ // TODO: use Decimal.MAX_PS instead when we introduce [Decimal]
+ if (bd.precision > 38) {
+ throw newValidationError(
+ literal,
+ Static.RESOURCE.numberLiteralOutOfRange(bd.toString))
+ }
+ case _ => super.validateLiteral(literal)
+ }
+ }
+
+ override def validateJoin(join: SqlJoin, scope: SqlValidatorScope): Unit = {
+ // Due to the improper translation of lateral table left outer join in Calcite, we need to
+ // temporarily forbid the common predicates until the problem is fixed (see FLINK-7865).
+ if (join.getJoinType == JoinType.LEFT &&
+ isCollectionTable(join.getRight)) {
+ join.getCondition match {
+ case c: SqlLiteral if c.booleanValue() && c.getValue.asInstanceOf[Boolean] =>
+ // We accept only literal true
+ case c if null != c =>
+ throw new ValidationException(
+ s"Left outer joins with a table function do not accept a predicate such as $c. " +
+ s"Only literal TRUE is accepted.")
+ }
+ }
+ super.validateJoin(join, scope)
+ }
+
+ private def isCollectionTable(node: SqlNode): Boolean = {
+ // TABLE (`func`(`foo`)) AS bar
+ node match {
+ case n: SqlCall if n.getKind == SqlKind.AS =>
+ n.getOperandList.get(0).getKind == SqlKind.COLLECTION_TABLE
+ case _ => false
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkPlannerImpl.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkPlannerImpl.scala
new file mode 100644
index 0000000..951bc05
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkPlannerImpl.scala
@@ -0,0 +1,224 @@
+/*
+ * 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 java.util
+import java.util.Properties
+
+import com.google.common.collect.ImmutableList
+import org.apache.calcite.config.{CalciteConnectionConfigImpl, CalciteConnectionProperty}
+import org.apache.calcite.jdbc.CalciteSchema
+import org.apache.calcite.plan.RelOptTable.ViewExpander
+import org.apache.calcite.plan._
+import org.apache.calcite.prepare.CalciteCatalogReader
+import org.apache.calcite.rel.RelRoot
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rex.RexBuilder
+import org.apache.calcite.schema.SchemaPlus
+import org.apache.calcite.sql.advise.{SqlAdvisor, SqlAdvisorValidator}
+import org.apache.calcite.sql.parser.{SqlParser, SqlParseException => CSqlParseException}
+import org.apache.calcite.sql.validate.SqlValidator
+import org.apache.calcite.sql.{SqlNode, SqlOperatorTable}
+import org.apache.calcite.sql2rel.{RelDecorrelator, SqlRexConvertletTable, SqlToRelConverter}
+import org.apache.calcite.tools.{FrameworkConfig, RelConversionException}
+import org.apache.flink.table.api.{SqlParserException, TableException, ValidationException}
+
+import scala.collection.JavaConversions._
+
+/**
+ * NOTE: this is heavily inspired by Calcite's PlannerImpl.
+ * We need it in order to share the planner between the Table API relational plans
+ * and the SQL relation plans that are created by the Calcite parser.
+ * The main difference is that we do not create a new RelOptPlanner in the ready() method.
+ */
+class FlinkPlannerImpl(
+ config: FrameworkConfig,
+ planner: RelOptPlanner,
+ typeFactory: FlinkTypeFactory) {
+
+ val operatorTable: SqlOperatorTable = config.getOperatorTable
+ /** Holds the trait definitions to be registered with planner. May be null. */
+ val traitDefs: ImmutableList[RelTraitDef[_ <: RelTrait]] = config.getTraitDefs
+ val parserConfig: SqlParser.Config = config.getParserConfig
+ val convertletTable: SqlRexConvertletTable = config.getConvertletTable
+ val defaultSchema: SchemaPlus = config.getDefaultSchema
+ val sqlToRelConverterConfig: SqlToRelConverter.Config = config.getSqlToRelConverterConfig
+
+ var validator: FlinkCalciteSqlValidator = _
+ var root: RelRoot = _
+
+ private def ready() {
+ if (this.traitDefs != null) {
+ planner.clearRelTraitDefs()
+ for (traitDef <- this.traitDefs) {
+ planner.addRelTraitDef(traitDef)
+ }
+ }
+ }
+
+ def getCompletionHints(sql: String, cursor: Int): Array[String] = {
+ val advisorValidator = new SqlAdvisorValidator(
+ operatorTable,
+ createCatalogReader(true), // ignore cases for lenient completion
+ typeFactory,
+ config.getParserConfig.conformance())
+ val advisor = new SqlAdvisor(advisorValidator)
+ val replaced = Array[String](null)
+ val hints = advisor.getCompletionHints(sql, cursor, replaced)
+ .map(item => item.toIdentifier.toString)
+ hints.toArray
+ }
+
+ def parse(sql: String): SqlNode = {
+ try {
+ ready()
+ val parser: SqlParser = SqlParser.create(sql, parserConfig)
+ val sqlNode: SqlNode = parser.parseStmt
+ sqlNode
+ } catch {
+ case e: CSqlParseException =>
+ throw new SqlParserException(s"SQL parse failed. ${e.getMessage}", e)
+ }
+ }
+
+ def validate(sqlNode: SqlNode): SqlNode = {
+ validator = new FlinkCalciteSqlValidator(
+ operatorTable,
+ createCatalogReader(false),
+ typeFactory)
+ validator.setIdentifierExpansion(true)
+ try {
+ validator.validate(sqlNode)
+ }
+ catch {
+ case e: RuntimeException =>
+ throw new ValidationException(s"SQL validation failed. ${e.getMessage}", e)
+ }
+ }
+
+ def rel(validatedSqlNode: SqlNode): RelRoot = {
+ try {
+ assert(validatedSqlNode != null)
+ val rexBuilder: RexBuilder = createRexBuilder
+ val cluster: RelOptCluster = FlinkRelOptClusterFactory.create(planner, rexBuilder)
+ val sqlToRelConverter: SqlToRelConverter = new SqlToRelConverter(
+ new ViewExpanderImpl,
+ validator,
+ createCatalogReader(false),
+ cluster,
+ convertletTable,
+ sqlToRelConverterConfig)
+ root = sqlToRelConverter.convertQuery(validatedSqlNode, false, true)
+ // we disable automatic flattening in order to let composite types pass without modification
+ // we might enable it again once Calcite has better support for structured types
+ // root = root.withRel(sqlToRelConverter.flattenTypes(root.rel, true))
+
+ // TableEnvironment.optimize will execute the following
+ // root = root.withRel(RelDecorrelator.decorrelateQuery(root.rel))
+ // convert time indicators
+ // root = root.withRel(RelTimeIndicatorConverter.convert(root.rel, rexBuilder))
+ root
+ } catch {
+ case e: RelConversionException => throw new TableException(e.getMessage)
+ }
+ }
+
+ /** Implements [[org.apache.calcite.plan.RelOptTable.ViewExpander]]
+ * interface for [[org.apache.calcite.tools.Planner]]. */
+ class ViewExpanderImpl extends ViewExpander {
+
+ override def expandView(
+ rowType: RelDataType,
+ queryString: String,
+ schemaPath: util.List[String],
+ viewPath: util.List[String]): RelRoot = {
+
+ val parser: SqlParser = SqlParser.create(queryString, parserConfig)
+ var sqlNode: SqlNode = null
+ try {
+ sqlNode = parser.parseQuery
+ }
+ catch {
+ case e: CSqlParseException =>
+ throw new SqlParserException(s"SQL parse failed. ${e.getMessage}", e)
+ }
+ val catalogReader: CalciteCatalogReader = createCatalogReader(false)
+ .withSchemaPath(schemaPath)
+ val validator: SqlValidator =
+ new FlinkCalciteSqlValidator(operatorTable, catalogReader, typeFactory)
+ validator.setIdentifierExpansion(true)
+ val validatedSqlNode: SqlNode = validator.validate(sqlNode)
+ val rexBuilder: RexBuilder = createRexBuilder
+ val cluster: RelOptCluster = FlinkRelOptClusterFactory.create(planner, rexBuilder)
+ val sqlToRelConverter: SqlToRelConverter = new SqlToRelConverter(
+ new ViewExpanderImpl,
+ validator,
+ catalogReader,
+ cluster,
+ convertletTable,
+ sqlToRelConverterConfig)
+ root = sqlToRelConverter.convertQuery(validatedSqlNode, true, false)
+ root = root.withRel(sqlToRelConverter.flattenTypes(root.rel, true))
+ root = root.withRel(RelDecorrelator.decorrelateQuery(root.rel))
+ FlinkPlannerImpl.this.root
+ }
+ }
+
+ private def createCatalogReader(lenientCaseSensitivity: Boolean): CalciteCatalogReader = {
+ val rootSchema: SchemaPlus = FlinkPlannerImpl.rootSchema(defaultSchema)
+
+ val caseSensitive = if (lenientCaseSensitivity) {
+ false
+ } else {
+ this.parserConfig.caseSensitive()
+ }
+
+ val parserConfig = SqlParser.configBuilder(this.parserConfig)
+ .setCaseSensitive(caseSensitive)
+ .build()
+
+ val prop = new Properties()
+ prop.setProperty(
+ CalciteConnectionProperty.CASE_SENSITIVE.camelName,
+ String.valueOf(parserConfig.caseSensitive))
+ val connectionConfig = new CalciteConnectionConfigImpl(prop)
+ new CalciteCatalogReader(
+ CalciteSchema.from(rootSchema),
+ CalciteSchema.from(defaultSchema).path(null),
+ typeFactory,
+ connectionConfig
+ )
+ }
+
+ private def createRexBuilder: RexBuilder = {
+ new RexBuilder(typeFactory)
+ }
+
+}
+
+object FlinkPlannerImpl {
+ private def rootSchema(schema: SchemaPlus): SchemaPlus = {
+ if (schema.getParentSchema == null) {
+ schema
+ }
+ else {
+ rootSchema(schema.getParentSchema)
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkRelBuilder.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkRelBuilder.scala
new file mode 100644
index 0000000..d7afcd0
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkRelBuilder.scala
@@ -0,0 +1,80 @@
+/*
+ * 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 java.util.{Collections, Properties}
+
+import org.apache.calcite.config.{CalciteConnectionConfigImpl, CalciteConnectionProperty}
+import org.apache.calcite.jdbc.CalciteSchema
+import org.apache.calcite.plan._
+import org.apache.calcite.plan.volcano.VolcanoPlanner
+import org.apache.calcite.prepare.CalciteCatalogReader
+import org.apache.calcite.rex.RexBuilder
+import org.apache.calcite.tools.{FrameworkConfig, RelBuilder}
+
+/**
+ * Flink specific [[RelBuilder]] that changes the default type factory to a [[FlinkTypeFactory]].
+ */
+class FlinkRelBuilder(
+ context: Context,
+ relOptCluster: RelOptCluster,
+ relOptSchema: RelOptSchema)
+ extends RelBuilder(
+ context,
+ relOptCluster,
+ relOptSchema) {
+
+ def getRelOptSchema: RelOptSchema = relOptSchema
+
+ def getPlanner: RelOptPlanner = cluster.getPlanner
+
+ def getCluster: RelOptCluster = relOptCluster
+
+ override def getTypeFactory: FlinkTypeFactory =
+ super.getTypeFactory.asInstanceOf[FlinkTypeFactory]
+}
+
+object FlinkRelBuilder {
+
+ def create(config: FrameworkConfig): FlinkRelBuilder = {
+
+ // create Flink type factory
+ val typeSystem = config.getTypeSystem
+ val typeFactory = new FlinkTypeFactory(typeSystem)
+
+ // create context instances with Flink type factory
+ val planner = new VolcanoPlanner(config.getCostFactory, Contexts.empty())
+ planner.setExecutor(config.getExecutor)
+ planner.addRelTraitDef(ConventionTraitDef.INSTANCE)
+ val cluster = FlinkRelOptClusterFactory.create(planner, new RexBuilder(typeFactory))
+ val calciteSchema = CalciteSchema.from(config.getDefaultSchema)
+ val prop = new Properties()
+ prop.setProperty(
+ CalciteConnectionProperty.CASE_SENSITIVE.camelName,
+ String.valueOf(config.getParserConfig.caseSensitive))
+ val connectionConfig = new CalciteConnectionConfigImpl(prop)
+ val relOptSchema = new CalciteCatalogReader(
+ calciteSchema,
+ Collections.emptyList(),
+ typeFactory,
+ connectionConfig)
+
+ new FlinkRelBuilder(config.getContext, cluster, relOptSchema)
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkRelOptClusterFactory.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkRelOptClusterFactory.scala
new file mode 100644
index 0000000..7bd6992
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkRelOptClusterFactory.scala
@@ -0,0 +1,41 @@
+/*
+ * 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.calcite.plan.{RelOptCluster, RelOptPlanner}
+import org.apache.calcite.rel.metadata.{DefaultRelMetadataProvider, JaninoRelMetadataProvider, RelMetadataQuery}
+import org.apache.calcite.rex.RexBuilder
+
+/**
+ * The utility class is to create special [[RelOptCluster]] instance.
+ */
+object FlinkRelOptClusterFactory {
+
+ def create(planner: RelOptPlanner, rexBuilder: RexBuilder): RelOptCluster = {
+ val cluster = RelOptCluster.create(planner, rexBuilder)
+ // TODO: create a Flink specific metadata provider [FlinkDefaultRelMetadataProvider]
+ cluster.setMetadataProvider(DefaultRelMetadataProvider.INSTANCE)
+ // just set metadataProvider is not enough, see
+ // https://www.mail-archive.com/dev@calcite.apache.org/msg00930.html
+ RelMetadataQuery.THREAD_PROVIDERS.set(
+ JaninoRelMetadataProvider.of(cluster.getMetadataProvider))
+ cluster
+ }
+
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/RelTable.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/RelTable.scala
new file mode 100644
index 0000000..57c725c
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/schema/RelTable.scala
@@ -0,0 +1,46 @@
+/*
+ * 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.calcite.plan.RelOptTable
+import org.apache.calcite.plan.RelOptTable.ToRelContext
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory}
+import org.apache.calcite.schema.Schema.TableType
+import org.apache.calcite.schema.impl.AbstractTable
+import org.apache.calcite.schema.TranslatableTable
+
+/**
+ * A [[org.apache.calcite.schema.Table]] implementation for registering
+ * Table API Tables in the Calcite schema to be used by Flink SQL.
+ * It implements [[TranslatableTable]] so that its logical scan
+ * can be converted to a relational expression.
+ *
+ * @see [[DataStreamTable]]
+ */
+class RelTable(relNode: RelNode) extends AbstractTable with TranslatableTable {
+
+ override def getJdbcTableType: TableType = ???
+
+ override def getRowType(typeFactory: RelDataTypeFactory): RelDataType = relNode.getRowType
+
+ override def toRel(context: ToRelContext, relOptTable: RelOptTable): RelNode = {
+ relNode
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/typeutils/TimeIndicatorTypeInfo.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/typeutils/TimeIndicatorTypeInfo.scala
new file mode 100644
index 0000000..ad82d52
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/typeutils/TimeIndicatorTypeInfo.scala
@@ -0,0 +1,59 @@
+/*
+ * 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.typeutils
+
+import java.sql.Timestamp
+
+import org.apache.flink.api.common.ExecutionConfig
+import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo
+import org.apache.flink.api.common.typeutils.{TypeComparator, TypeSerializer}
+import org.apache.flink.api.common.typeutils.base.{LongSerializer, SqlTimestampComparator, SqlTimestampSerializer}
+
+/**
+ * Type information for indicating event or processing time. However, it behaves like a
+ * regular SQL timestamp but is serialized as Long.
+ */
+class TimeIndicatorTypeInfo(val isEventTime: Boolean)
+ extends SqlTimeTypeInfo[Timestamp](
+ classOf[Timestamp],
+ SqlTimestampSerializer.INSTANCE,
+ classOf[SqlTimestampComparator].asInstanceOf[Class[TypeComparator[Timestamp]]]) {
+
+ // this replaces the effective serializer by a LongSerializer
+ // it is a hacky but efficient solution to keep the object creation overhead low but still
+ // be compatible with the corresponding SqlTimestampTypeInfo
+ override def createSerializer(executionConfig: ExecutionConfig): TypeSerializer[Timestamp] =
+ LongSerializer.INSTANCE.asInstanceOf[TypeSerializer[Timestamp]]
+
+ override def toString: String =
+ s"TimeIndicatorTypeInfo(${if (isEventTime) "rowtime" else "proctime" })"
+}
+
+object TimeIndicatorTypeInfo {
+
+ val ROWTIME_STREAM_MARKER: Int = -1
+ val PROCTIME_STREAM_MARKER: Int = -2
+
+ val ROWTIME_BATCH_MARKER: Int = -3
+ val PROCTIME_BATCH_MARKER: Int = -4
+
+ val ROWTIME_INDICATOR = new TimeIndicatorTypeInfo(true)
+ val PROCTIME_INDICATOR = new TimeIndicatorTypeInfo(false)
+
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala
new file mode 100644
index 0000000..e5981ef
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala
@@ -0,0 +1,74 @@
+/*
+ * 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.api
+
+import org.apache.calcite.plan.RelOptUtil
+import org.apache.flink.api.scala._
+import org.apache.flink.streaming.api.environment.LocalStreamEnvironment
+import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.api.scala.StreamTableEnvironment
+import org.junit.{Rule, Test}
+import org.junit.Assert.assertEquals
+import org.junit.rules.ExpectedException
+
+class TableEnvironmentTest {
+
+ // used for accurate exception information checking.
+ val expectedException: ExpectedException = ExpectedException.none()
+
+ @Rule
+ def thrown: ExpectedException = expectedException
+
+ val env = new StreamExecutionEnvironment(new LocalStreamEnvironment())
+ val tableEnv: scala.StreamTableEnvironment = StreamTableEnvironment.create(env)
+
+ @Test
+ def testScanNonExistTable(): Unit = {
+ thrown.expect(classOf[TableException])
+ thrown.expectMessage("Table 'MyTable' was not found")
+ tableEnv.scan("MyTable")
+ }
+
+ @Test
+ def testRegisterDataStream(): Unit = {
+ val table = env.fromElements[(Int, Long, String, Boolean)]().toTable(tableEnv, 'a, 'b, 'c, 'd)
+ tableEnv.registerTable("MyTable", table)
+ val scanTable = tableEnv.scan("MyTable")
+ val actual = RelOptUtil.toString(scanTable.getRelNode)
+ val expected = "LogicalTableScan(table=[[MyTable]])\n"
+ assertEquals(expected, actual)
+
+ // register on a conflict name
+ thrown.expect(classOf[TableException])
+ thrown.expectMessage("Table 'MyTable' already exists")
+ tableEnv.registerDataStream("MyTable", env.fromElements[(Int, Long)]())
+ }
+
+ @Test
+ def testSimpleQuery(): Unit = {
+ val table = env.fromElements[(Int, Long, String, Boolean)]().toTable(tableEnv, 'a, 'b, 'c, 'd)
+ tableEnv.registerTable("MyTable", table)
+ val queryTable = tableEnv.sqlQuery("SELECT a, c, d FROM MyTable")
+ val actual = RelOptUtil.toString(queryTable.getRelNode)
+ val expected = "LogicalProject(a=[$0], c=[$2], d=[$3])\n" +
+ " LogicalTableScan(table=[[MyTable]])\n"
+ assertEquals(expected, actual)
+ }
+}