| /* |
| * 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.annotation.{Experimental, Internal, VisibleForTesting} |
| import org.apache.flink.api.common.JobExecutionResult |
| import org.apache.flink.api.common.time.Time |
| import org.apache.flink.api.common.typeinfo.TypeInformation |
| import org.apache.flink.api.java.typeutils._ |
| import org.apache.flink.streaming.api.environment.{StreamExecutionEnvironment => JavaStreamExecEnv} |
| import org.apache.flink.streaming.api.graph.StreamGraph |
| import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment => ScalaStreamExecEnv} |
| import org.apache.flink.streaming.api.transformations.StreamTransformation |
| import org.apache.flink.table.api.functions.{AggregateFunction, ScalarFunction, TableFunction} |
| import org.apache.flink.table.api.java.{BatchTableEnvironment => JavaBatchTableEnvironment, StreamTableEnvironment => JavaStreamTableEnv} |
| import org.apache.flink.table.api.scala.{BatchTableEnvironment => ScalaBatchTableEnvironment, StreamTableEnvironment => ScalaStreamTableEnv, _} |
| import org.apache.flink.table.api.types._ |
| import org.apache.flink.table.calcite._ |
| import org.apache.flink.table.catalog._ |
| import org.apache.flink.table.codegen._ |
| import org.apache.flink.table.descriptors.{ConnectorDescriptor, TableDescriptor} |
| import org.apache.flink.table.errorcode.TableErrors |
| import org.apache.flink.table.expressions.{Alias, Expression, TimeAttribute, UnresolvedFieldReference} |
| import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils |
| import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils._ |
| import org.apache.flink.table.plan.cost.FlinkCostFactory |
| import org.apache.flink.table.plan.logical.{CatalogNode, LogicalNode, LogicalRelNode, SinkNode} |
| import org.apache.flink.table.plan.nodes.exec.ExecNode |
| import org.apache.flink.table.plan.schema._ |
| import org.apache.flink.table.plan.stats.{FlinkStatistic, TableStats} |
| import org.apache.flink.table.sinks._ |
| import org.apache.flink.table.sources.TableSource |
| import org.apache.flink.table.temptable.FlinkTableServiceManager |
| import org.apache.flink.table.typeutils.TypeUtils |
| import org.apache.flink.table.validate.{BuiltInFunctionCatalog, ChainedFunctionCatalog, FunctionCatalog} |
| |
| import org.apache.calcite.config.Lex |
| import org.apache.calcite.plan.{Contexts, RelOptPlanner} |
| import org.apache.calcite.rel.logical.LogicalTableModify |
| import org.apache.calcite.schema |
| import org.apache.calcite.schema.SchemaPlus |
| import org.apache.calcite.schema.impl.AbstractTable |
| import org.apache.calcite.sql.parser.SqlParser |
| import org.apache.calcite.sql.util.ChainedSqlOperatorTable |
| import org.apache.calcite.sql.{SqlIdentifier, SqlInsert, SqlOperatorTable, _} |
| import org.apache.calcite.sql2rel.SqlToRelConverter |
| import org.apache.calcite.tools._ |
| import org.apache.commons.lang3.StringUtils |
| |
| import _root_.java.lang.reflect.Modifier |
| import _root_.java.util |
| import _root_.java.util.concurrent.atomic.AtomicInteger |
| import _root_.java.util.concurrent.atomic.AtomicBoolean |
| |
| import _root_.scala.annotation.varargs |
| import _root_.scala.collection.JavaConversions._ |
| import _root_.scala.collection.JavaConverters._ |
| import _root_.scala.collection.mutable |
| import _root_.scala.collection.mutable.ArrayBuffer |
| |
| /** |
| * The abstract base class for batch and stream TableEnvironments. |
| * |
| * @param config The configuration of the TableEnvironment |
| */ |
| abstract class TableEnvironment( |
| private[flink] val execEnv: JavaStreamExecEnv, |
| val config: TableConfig) extends AutoCloseable { |
| |
| protected val DEFAULT_JOB_NAME = "Flink Exec Table Job" |
| |
| protected val catalogManager: CatalogManager = new CatalogManager() |
| private val currentSchema: SchemaPlus = catalogManager.getRootSchema |
| |
| private val typeFactory: FlinkTypeFactory = new FlinkTypeFactory(new FlinkTypeSystem) |
| |
| // Table API/SQL function catalog (built in, does not contain external functions) |
| private val functionCatalog: FunctionCatalog = BuiltInFunctionCatalog.withBuiltIns() |
| |
| // Table API/SQL function catalog built in function catalog. |
| private[flink] lazy val chainedFunctionCatalog: FunctionCatalog = |
| new ChainedFunctionCatalog(Seq(functionCatalog)) |
| |
| // the configuration to create a Calcite planner |
| protected var frameworkConfig: FrameworkConfig = createFrameworkConfig |
| |
| // the builder for Calcite RelNodes, Calcite's representation of a relational expression tree. |
| protected var relBuilder: FlinkRelBuilder = createRelBuilder |
| |
| // the planner instance used to optimize queries of this TableEnvironment |
| private var planner: RelOptPlanner = createRelOptPlanner |
| |
| // reuse flink planner |
| private var flinkPlanner: FlinkPlannerImpl = createFlinkPlanner |
| |
| // a counter for unique attribute names |
| private[flink] val attrNameCntr: AtomicInteger = new AtomicInteger(0) |
| |
| // a counter for unique table names |
| private[flink] val tableNameCntr: AtomicInteger = new AtomicInteger(0) |
| |
| private[flink] val tableNamePrefix = "_TempTable_" |
| |
| // sink nodes collection |
| private[flink] val sinkNodes = new mutable.MutableList[SinkNode] |
| |
| private[flink] val transformations = new ArrayBuffer[StreamTransformation[_]] |
| |
| protected var userClassloader: ClassLoader = null |
| |
| // a manager for table service |
| private[flink] val tableServiceManager = new FlinkTableServiceManager(this) |
| |
| private val closed: AtomicBoolean = new AtomicBoolean(false) |
| |
| // the configuration for SqlToRelConverter |
| private[flink] lazy val sqlToRelConverterConfig: SqlToRelConverter.Config = { |
| val calciteConfig = config.getCalciteConfig |
| calciteConfig.getSqlToRelConverterConfig match { |
| case Some(c) => c |
| case None => getSqlToRelConverterConfig |
| } |
| } |
| |
| /** 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 = this match { |
| case _: BatchTableEnvironment => new BatchQueryConfig |
| case _: StreamTableEnvironment => new StreamQueryConfig |
| case _ => null |
| } |
| |
| /** |
| * Compile the sink [[org.apache.flink.table.plan.logical.LogicalNode]] to |
| * [[org.apache.flink.streaming.api.transformations.StreamTransformation]]. |
| */ |
| protected def compile(): Unit = { |
| if (config.getSubsectionOptimization) { |
| // optimize rel node, and translate to node dag |
| val nodeDag = optimizeAndTranslateNodeDag(true, sinkNodes: _*) |
| // translate to transformation |
| val sinkTransformations = translate(nodeDag) |
| transformations.addAll(sinkTransformations) |
| } |
| } |
| |
| /** |
| * Optimize the RelNode tree (or DAG), and translate the result to [[ExecNode]] tree (or DAG). |
| */ |
| @VisibleForTesting |
| private[flink] def optimizeAndTranslateNodeDag( |
| dagOptimizeEnabled: Boolean, logicalNodes: LogicalNode*): Seq[ExecNode[_, _]] |
| |
| /** |
| * Translates a [[ExecNode]] DAG into a [[StreamTransformation]] DAG. |
| * |
| * @param sinks The node DAG to translate. |
| * @return The [[StreamTransformation]] DAG that corresponds to the node DAG. |
| */ |
| protected def translate(sinks: Seq[ExecNode[_, _]]): Seq[StreamTransformation[_]] |
| |
| /** |
| * Generate a [[StreamGraph]] from this table environment, this will also |
| * clear [[LogicalNode]]s. |
| * @return A [[StreamGraph]] describing the whole job. |
| */ |
| def generateStreamGraph(): StreamGraph = generateStreamGraph(DEFAULT_JOB_NAME) |
| |
| /** |
| * Generate a [[StreamGraph]] from this table environment, this will also |
| * clear [[LogicalNode]]s. |
| * @return A [[StreamGraph]] describing the whole job. |
| */ |
| def generateStreamGraph(jobName: String): StreamGraph = { |
| try { |
| if (config.getSubsectionOptimization) { |
| compile() |
| } |
| if (transformations.isEmpty) { |
| throw new TableException("No table sinks have been created yet. " + |
| "A program needs at least one sink that consumes data. ") |
| } |
| translateStreamGraph(transformations, Option.apply(jobName)) |
| } finally { |
| sinkNodes.clear() |
| } |
| } |
| |
| /** |
| * Translate a [[StreamGraph]] from Given streamingTransformations. |
| * @return A [[StreamGraph]] describing the given job. |
| */ |
| protected def translateStreamGraph( |
| streamingTransformations: ArrayBuffer[StreamTransformation[_]], |
| jobName: Option[String]): StreamGraph = ??? |
| |
| /** |
| * Returns the operator table for this environment including a custom Calcite configuration. |
| */ |
| protected def getSqlOperatorTable: SqlOperatorTable = { |
| val calciteConfig = config.getCalciteConfig |
| |
| calciteConfig.getSqlOperatorTable match { |
| case None => |
| chainedFunctionCatalog.getSqlOperatorTable |
| case Some(table) => |
| if (calciteConfig.replacesSqlOperatorTable) { |
| table |
| } else { |
| ChainedSqlOperatorTable.of(chainedFunctionCatalog.getSqlOperatorTable, table) |
| } |
| } |
| } |
| |
| /** |
| * Returns the SQL parser config for this environment including a custom Calcite configuration. |
| */ |
| protected def getSqlParserConfig: SqlParser.Config = { |
| val calciteConfig = config.getCalciteConfig |
| calciteConfig.getSqlParserConfig match { |
| |
| case None => |
| // we use Java lex because back ticks are easier than double quotes in programming |
| // and cases are preserved |
| SqlParser |
| .configBuilder() |
| .setLex(Lex.JAVA) |
| .setIdentifierMaxLength(256) |
| .build() |
| |
| case Some(sqlParserConfig) => |
| sqlParserConfig |
| } |
| } |
| |
| /** |
| * Returns the SqlToRelConverter config. |
| */ |
| @Internal |
| protected def getSqlToRelConverterConfig: SqlToRelConverter.Config = |
| SqlToRelConverter.configBuilder() |
| .withTrimUnusedFields(false) |
| .withConvertTableAccess(false) |
| .build() |
| |
| def getCatalogManager: CatalogManager = { |
| catalogManager |
| } |
| |
| /** |
| * Registers an [[ReadableCatalog]] under a unique name in the TableEnvironment's schema. |
| * All tables registered in the [[ReadableCatalog]] can be accessed. |
| * |
| * @param name The name under which the catalog will be registered |
| * @param catalog The catalog to register |
| */ |
| def registerCatalog(name: String, catalog: ReadableCatalog): Unit = { |
| registerCatalogInternal(name, catalog) |
| } |
| |
| /** |
| * Registers an [[ReadableCatalog]] under a unique name in the TableEnvironment's schema. |
| * All tables registered in the [[ReadableCatalog]] can be accessed. |
| * |
| * @param name The name under which the catalog will be registered |
| * @param catalog The catalog to register |
| * |
| */ |
| @throws[CatalogAlreadyExistException] |
| def registerCatalogInternal( |
| name: String, |
| catalog: ReadableCatalog): Unit = { |
| |
| catalogManager.registerCatalog(name, catalog) |
| } |
| |
| /** |
| * Get a registered catalog. |
| * |
| * @param catalogName |
| * @return ReadableCatalog |
| */ |
| @throws[CatalogNotExistException] |
| def getCatalog(catalogName: String): ReadableCatalog = { |
| catalogManager.getCatalog(catalogName) |
| } |
| |
| /** |
| * Get the default registered catalog. |
| * |
| * @return ReadableWritableCatalog |
| */ |
| @throws[CatalogNotExistException] |
| def getDefaultCatalog(): ReadableWritableCatalog = { |
| catalogManager.getCatalog(getDefaultCatalogName()).asInstanceOf[ReadableWritableCatalog] |
| } |
| |
| /** |
| * Get the default catalog name. |
| */ |
| def getDefaultCatalogName(): String = { |
| catalogManager.getDefaultCatalogName |
| } |
| |
| /** |
| * Get the default database name. |
| */ |
| def getDefaultDatabaseName(): String = { |
| catalogManager.getDefaultDatabaseName |
| } |
| |
| /** |
| * Set a default catalog. |
| * |
| * @param name Name of the catalog |
| */ |
| def setDefaultCatalog(name: String): Unit = { |
| catalogManager.setDefaultCatalog(name) |
| } |
| |
| /** |
| * Set a default catalog and database. |
| * |
| * @param catalogName Name of the catalog |
| * @param dbName Name of the database |
| */ |
| def setDefaultDatabase(catalogName: String, dbName: String): Unit = { |
| catalogManager.setDefaultDatabase(catalogName, dbName) |
| } |
| |
| /** |
| * Set the default database. If a catalog is not specified, the database is resolved relative |
| * to the current default catalog. |
| * Note! This method does not support setting default catalog only. |
| * |
| * @param dbPath Name or path of the database |
| */ |
| @varargs |
| def setDefaultDatabase(dbPath: String*): Unit = { |
| if (dbPath(0) == null || dbPath.length < 1 || dbPath.length > 2 || dbPath(0).isEmpty || |
| (dbPath.length == 2 && dbPath(1).isEmpty)) { |
| throw new IllegalArgumentException(String.format("Invalid database path %s", dbPath)) |
| } |
| |
| val catalogName = if (dbPath.length == 1) getDefaultCatalogName() else dbPath(0) |
| val dbName = if (dbPath.length == 1) dbPath(0) else dbPath(1) |
| |
| catalogManager.setDefaultDatabase(catalogName, dbName) |
| } |
| |
| /** |
| * Get a table catalogs. |
| * |
| * @param name The name of the table. |
| * @return The table registered either internally or externally, None otherwise. |
| */ |
| def getTable(name: String): Option[org.apache.calcite.schema.Table] = { |
| getTable(name.split('.')) |
| } |
| |
| /** |
| * Get a table from catalogs. |
| * |
| * @param paths The paths of the table. |
| * @return The table registered either internally or externally, None otherwise. |
| */ |
| def getTable(paths: Array[String]): Option[org.apache.calcite.schema.Table] = { |
| val names = catalogManager.resolveTableName(paths : _*) |
| val catalogName = names(0) |
| val dbName = names(1) |
| val tableName = names(2) |
| |
| val catalogSchema = catalogManager.getRootSchema.getSubSchema(catalogName) |
| |
| if (catalogSchema == null) { |
| Option.empty |
| } else { |
| val dbSchema = catalogSchema.getSubSchema(dbName) |
| |
| if (dbSchema == null) { |
| Option.empty |
| } else { |
| Option(dbSchema.getTable(tableName)) |
| } |
| } |
| } |
| |
| /** |
| * Registers a [[ScalarFunction]] under a unique name. Replaces already existing |
| * user-defined functions under this name. |
| */ |
| def registerFunction(name: String, function: ScalarFunction): Unit = { |
| // check if class could be instantiated |
| checkForInstantiation(function.getClass) |
| |
| // register in Table API |
| functionCatalog.registerFunction(name, function.getClass) |
| |
| // register in SQL API |
| functionCatalog.registerSqlFunction( |
| createScalarSqlFunction(name, name, function, typeFactory) |
| ) |
| } |
| |
| /** |
| * Registers an [[AggregateFunction]] under a unique name in the TableEnvironment's catalog. |
| * Registered functions can be referenced in Table API and SQL queries. |
| * |
| * @param name The name under which the function is registered. |
| * @param f The AggregateFunction to register. |
| * @tparam T The type of the output value. |
| * @tparam ACC The type of aggregate accumulator. |
| */ |
| def registerFunction[T, ACC]( |
| name: String, |
| f: AggregateFunction[T, ACC]) |
| : Unit = { |
| implicit val typeInfo: TypeInformation[T] = TypeExtractor |
| .createTypeInfo(f, classOf[AggregateFunction[T, ACC]], f.getClass, 0) |
| .asInstanceOf[TypeInformation[T]] |
| |
| implicit val accTypeInfo: TypeInformation[ACC] = TypeExtractor |
| .createTypeInfo(f, classOf[AggregateFunction[T, ACC]], f.getClass, 1) |
| .asInstanceOf[TypeInformation[ACC]] |
| |
| registerAggregateFunctionInternal[T, ACC](name, f) |
| } |
| |
| /** |
| * Registers a [[TableFunction]] under a unique name in the TableEnvironment's catalog. |
| * Registered functions can be referenced in Table API and SQL queries. |
| * |
| * @param name The name under which the function is registered. |
| * @param tf The TableFunction to register. |
| * @tparam T The type of the output row. |
| */ |
| def registerFunction[T](name: String, tf: TableFunction[T]): Unit = { |
| implicit val typeInfo: TypeInformation[T] = |
| UserDefinedFunctionUtils.getImplicitResultTypeInfo(tf) |
| registerTableFunctionInternal(name, tf) |
| } |
| |
| /** |
| * Registers a [[TableFunction]] under a unique name. Replaces already existing |
| * user-defined functions under this name. |
| */ |
| private[flink] def registerTableFunctionInternal[T: TypeInformation]( |
| name: String, function: TableFunction[T]): Unit = { |
| // check if class not Scala object |
| checkNotSingleton(function.getClass) |
| // check if class could be instantiated |
| checkForInstantiation(function.getClass) |
| val implicitResultType: DataType = |
| // we may use arguments types to infer later on. |
| if (UserDefinedFunctionUtils.getResultTypeIgnoreException(function) != null) { |
| function.getResultType(null, null) |
| } else { |
| implicitly[TypeInformation[T]] |
| } |
| |
| // register in Table API |
| functionCatalog.registerFunction(name, function.getClass) |
| |
| // register in SQL API |
| val sqlFunctions = |
| createTableSqlFunction(name, name, function, implicitResultType, typeFactory) |
| functionCatalog.registerSqlFunction(sqlFunctions) |
| } |
| |
| /** |
| * Registers an [[AggregateFunction]] under a unique name. Replaces already existing |
| * user-defined functions under this name. |
| */ |
| private[flink] def registerAggregateFunctionInternal[T: TypeInformation, ACC: TypeInformation]( |
| name: String, |
| function: AggregateFunction[T, ACC]): Unit = { |
| // check if class not Scala object |
| checkNotSingleton(function.getClass) |
| // check if class could be instantiated |
| checkForInstantiation(function.getClass) |
| |
| val resultType = getResultTypeOfAggregateFunction(function, implicitly[TypeInformation[T]]) |
| val accType = getAccumulatorTypeOfAggregateFunction(function, implicitly[TypeInformation[ACC]]) |
| |
| // register in Table API |
| functionCatalog.registerFunction(name, function.getClass) |
| |
| // register in SQL API |
| val sqlFunctions = createAggregateSqlFunction( |
| name, |
| name, |
| function, |
| resultType, |
| accType, |
| typeFactory) |
| |
| functionCatalog.registerSqlFunction(sqlFunctions) |
| } |
| |
| /** |
| * 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, replace = false) |
| } |
| |
| /** |
| * Registers a [[CatalogTable]] under a unique name in the TableEnvironment's default catalog. |
| * Registered tables can be referenced in SQL queries. |
| * |
| * @param name The name under which the table will be registered. |
| * @param catalogTable The table to register. |
| */ |
| def registerTable(name: String, catalogTable: CatalogTable): Unit = { |
| val catalog = getDefaultCatalog() |
| val path = new ObjectPath(getDefaultDatabaseName(), name) |
| catalog.createTable(path, catalogTable, false) |
| } |
| |
| /** |
| * Registers or replace 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 registerOrReplaceTable(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, replace = true) |
| } |
| |
| /** |
| * Registers or replace a [[CatalogTable]] under a unique name in TableEnvironment's default |
| * catalog. Registered tables can be referenced in SQL queries. |
| * |
| * @param name The name under which the table will be registered. |
| * @param catalogTable The table to register. |
| */ |
| def registerOrReplaceTable(name: String, catalogTable: CatalogTable): Unit = { |
| val catalog = getDefaultCatalog() |
| val path = new ObjectPath(getDefaultDatabaseName(),name) |
| |
| catalog.dropTable(path, true) |
| catalog.createTable(path, catalogTable, false) |
| } |
| |
| /** |
| * 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 TableAlreadyExistException if another table is registered under the provided name. |
| */ |
| @throws[TableAlreadyExistException] |
| private[flink] def registerTableInternal(name: String, table: AbstractTable): Unit = { |
| catalogManager.getCatalog(catalogManager.getDefaultCatalogName) |
| .asInstanceOf[ReadableWritableCatalog] |
| .createTable( |
| new ObjectPath(catalogManager.getDefaultDatabaseName, name), |
| createFlinkTempTable(table), |
| false |
| ) |
| } |
| |
| /** |
| * 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 = { |
| catalogManager.getCatalog(catalogManager.getDefaultCatalogName) |
| .asInstanceOf[ReadableWritableCatalog] |
| .alterTable( |
| new ObjectPath(catalogManager.getDefaultDatabaseName, name), |
| createFlinkTempTable(table), |
| false |
| ) |
| } |
| |
| private def createFlinkTempTable(table: AbstractTable): FlinkTempTable = { |
| val currentMillis = System.currentTimeMillis() |
| |
| return new FlinkTempTable( |
| table, |
| null, |
| null, |
| new util.HashMap[String, String](), |
| null, |
| null, |
| null, |
| new util.LinkedHashSet[String](), |
| false, |
| null, |
| null, |
| -1, |
| currentMillis, |
| currentMillis |
| ) |
| } |
| |
| /** |
| * Registers an external [[TableSource]] in this [[TableEnvironment]]'s catalog. |
| * Registered tables can be referenced in SQL queries. |
| * |
| * @param name The name under which the [[TableSource]] is registered. |
| * @param tableSource The [[TableSource]] to register. |
| */ |
| def registerTableSource(name: String, tableSource: TableSource): Unit = { |
| checkValidTableName(name) |
| registerTableSourceInternal(name, tableSource, FlinkStatistic.UNKNOWN, false) |
| } |
| |
| /** |
| * Registers or replace an external [[TableSource]] in this [[TableEnvironment]]'s catalog. |
| * Registered tables can be referenced in SQL queries. |
| * |
| * @param name The name under which the [[TableSource]] is registered. |
| * @param tableSource The [[TableSource]] to register. |
| */ |
| def registerOrReplaceTableSource(name: String, |
| tableSource: TableSource): Unit = { |
| checkValidTableName(name) |
| registerTableSourceInternal(name, tableSource, FlinkStatistic.UNKNOWN, true) |
| } |
| |
| /** |
| * Registers an internal [[TableSource]] in this [[TableEnvironment]]'s catalog without |
| * name checking. Registered tables can be referenced in SQL queries. |
| * |
| * @param name The name under which the [[TableSource]] is registered. |
| * @param tableSource The [[TableSource]] to register. |
| * @param replace Whether to replace this [[TableSource]] |
| */ |
| protected def registerTableSourceInternal(name: String, |
| tableSource: TableSource, |
| tableStats: FlinkStatistic, |
| replace: Boolean): Unit |
| |
| /** |
| * Gets the statistics of a table. |
| * Note: this function returns current statistics of the table directly, does not trigger |
| * statistics gather operation. |
| * |
| * @param tableName The table name under which the table is registered in [[TableEnvironment]]. |
| * tableName must be a single name(e.g. "MyTable") associated with a table. |
| * @return Statistics of a table if the statistics is available, else return null. |
| */ |
| @Experimental |
| def getTableStats(tableName: String): TableStats = { |
| require(tableName != null && tableName.nonEmpty, "tableName must not be null or empty.") |
| getTableStats(Array(tableName)) |
| } |
| |
| /** |
| * Gets the statistics of a table. |
| * Note: this function returns current statistics of the table directly, does not trigger |
| * statistics gather operation. |
| * |
| * @param tablePath The table name under which the table is registered in [[TableEnvironment]]. |
| * tablePath can be a single name(e.g. Array("MyTable")) associated with a |
| * table , or can be a nest names (e.g. Array("MyCatalog", "MyDb", "MyTable")) |
| * associated with a table registered as member of a [[ReadableCatalog]]. |
| * @return Statistics of a table if the statistics is available, else return null. |
| */ |
| @Experimental |
| def getTableStats(tablePath: Array[String]): TableStats = { |
| val tableOpt = getTable(tablePath) |
| if (tableOpt.isEmpty) { |
| throw new TableException(s"Table '${tablePath.mkString(".")}' was not found.") |
| } |
| val table = tableOpt.get |
| val tableName = tablePath.last |
| val stats = if (tablePath.length == 1) { |
| table match { |
| case t: FlinkTable => |
| // call statistic instead of getStatistics of FlinkTable to fetch the original statistics. |
| val statistics = t.getStatistic |
| if (statistics == null) { |
| None |
| } else { |
| Option(statistics.getTableStats) |
| } |
| // Only source table extends FlinkTable now |
| case sourceSinkTable: TableSourceSinkTable[_] if sourceSinkTable.isSourceTable => |
| val statistics = sourceSinkTable.tableSourceTable.get.getStatistic |
| if (statistics == null) { |
| None |
| } else { |
| Option(statistics.getTableStats) |
| } |
| case _ => None |
| } |
| } else { |
| // table in catalogs |
| val path = catalogManager.resolveTableName(tablePath.toList) |
| val catalog = getCatalog(path(0)) |
| |
| Option(catalog.getTable(new ObjectPath(path(1), path(2))).getTableStats) |
| } |
| stats.orNull |
| } |
| |
| /** |
| * Alters the statistics of a table. |
| * |
| * @param tablePath The table name under which the table is registered in [[TableEnvironment]]. |
| * tablePath can be a single name(e.g. Array("MyTable")) associated with a |
| * table , or can be a nest names (e.g. Array("MyCatalog", "MyDb", "MyTable")) |
| * associated with a table registered as member of a [[ReadableCatalog]]. |
| * @param tableStats The [[TableStats]] to update. |
| */ |
| def alterTableStats(tablePath: Array[String], tableStats: TableStats): Unit = { |
| alterTableStats(tablePath, Option(tableStats)) |
| } |
| |
| /** |
| * Alters the statistics of a table. |
| * |
| * @param tableName The table name under which the table is registered in [[TableEnvironment]]. |
| * tableName must be a single name(e.g. "MyTable") associated with a table. |
| * @param tableStats The [[TableStats]] to update. |
| */ |
| @Experimental |
| def alterTableStats(tableName: String, tableStats: Option[TableStats]): Unit = { |
| require(tableName != null && tableName.nonEmpty, "tableName must not be null or empty.") |
| alterTableStats(Array(tableName), tableStats) |
| } |
| |
| /** |
| * Alters the statistics of a table. |
| * |
| * @param tablePath The table name under which the table is registered in [[TableEnvironment]]. |
| * tablePath can be a single name(e.g. Array("MyTable")) associated with a |
| * table , or can be a nest names (e.g. Array("MyCatalog", "MyDb", "MyTable")) |
| * associated with a table registered as member of a [[ReadableCatalog]]. |
| * @param tableStats The [[TableStats]] to update. |
| */ |
| @Experimental |
| def alterTableStats(tablePath: Array[String], tableStats: Option[TableStats]): Unit = { |
| val tableOpt = getTable(tablePath) |
| if (tableOpt.isEmpty) { |
| throw new TableException(s"Table '${tablePath.mkString(".")}' was not found.") |
| } |
| |
| val table = tableOpt.get |
| val tableName = tablePath.last |
| if (tablePath.length == 1) { |
| // table in calcite root schema |
| val statistic = table match { |
| // call statistic instead of getStatistics of TableSourceTable |
| // to fetch the original statistics. |
| case t: TableSourceSinkTable[_] if t.isSourceTable => t.tableSourceTable.get.statistic |
| case t: FlinkTable => t.getStatistic |
| case _ => throw new TableException( |
| s"alter TableStats operation is not supported for ${table.getClass}.") |
| } |
| val oldStatistic = if (statistic == null) FlinkStatistic.UNKNOWN else statistic |
| val newStatistic = FlinkStatistic.builder.statistic(oldStatistic). |
| tableStats(tableStats.orNull).build() |
| val newTable = table.asInstanceOf[FlinkTable].copy(newStatistic) |
| replaceRegisteredTable(tableName, newTable) |
| } else { |
| // table in catalogs |
| val path = catalogManager.resolveTableName(tablePath.toList) |
| val catalog = getCatalog(path(0)) |
| |
| // TODO: [BLINK-18570617] re-enable alter catalog table stats in TableEnvironment |
| // catalog match { |
| // case c: ReadableWritableCatalog => |
| // c.alterTableStats(tableName, tableStats, ignoreIfNotExists = false) |
| // case _ => throw new TableException( |
| // s"alterTableStats operation is not supported for ${catalog.getClass}.") |
| // } |
| |
| throw new TableException( |
| s"catalogs haven't supportted alterTableStats operation yet.") |
| } |
| } |
| |
| /** |
| * Alter skew info to a table, optimizer will try to choose better plan on skewed data. |
| * 1. pick the skewed values to join separately |
| * 2. prefer to choose add local-combine aggregate before global aggregate which group by |
| * skewed data |
| * |
| * TODO: Add skewInfo on a specified set of columns later. Now only support to specify skewInfo |
| * on a singleKey, it is not enough to determines whether a specified set of columns from a |
| * specified relational expression is skew or not. |
| * |
| * @param tableName table name to alter. |
| * @param skewInfo statistics of skewedColNames and skewedColValues. |
| */ |
| def alterSkewInfo( |
| tableName: String, |
| skewInfo: util.Map[String, util.List[AnyRef]]): Unit = { |
| require(tableName != null && tableName.nonEmpty, "tableName must not be null or empty.") |
| alterSkewInfo(Array(tableName), skewInfo) |
| } |
| |
| private def alterSkewInfo( |
| tablePath: Array[String], |
| skewInfo: util.Map[String, util.List[AnyRef]]): Unit = { |
| val tableOpt = getTable(tablePath) |
| if (tableOpt.isEmpty) { |
| throw new TableException(s"Table '${tablePath.mkString(".")}' was not found.") |
| } |
| |
| val table = tableOpt.get |
| val tableName = tablePath.last |
| if (tablePath.length == 1) { |
| // table in calcite root schema |
| val statistic = table match { |
| // call statistic instead of getStatistics of TableSourceTable |
| // to fetch the original statistics. |
| case t: TableSourceSinkTable[_] if t.isSourceTable => t.tableSourceTable.get.statistic |
| case t: FlinkTable => t.getStatistic |
| case _ => throw new TableException( |
| s"alter SkewInfo operation is not supported for ${table.getClass}.") |
| } |
| val oldStatistic = if (statistic == null) FlinkStatistic.UNKNOWN else statistic |
| val newStatistic = FlinkStatistic.builder.statistic(oldStatistic).skewInfo(skewInfo).build() |
| val newTable = table.asInstanceOf[FlinkTable].copy(newStatistic) |
| replaceRegisteredTable(tableName, newTable) |
| } else { |
| throw new TableException("alterSkewInfo operation is not supported for external catalog.") |
| } |
| } |
| |
| /** |
| * Registers an external [[TableSink]] with given field names and types in this |
| * [[TableEnvironment]]'s catalog. |
| * Registered sink tables can be referenced in SQL DML statements. |
| * |
| * @param name The name under which the [[TableSink]] is registered. |
| * @param fieldNames The field names to register with the [[TableSink]]. |
| * @param fieldTypes The field types to register with the [[TableSink]]. |
| * @param tableSink The [[TableSink]] to register. |
| */ |
| def registerTableSink( |
| name: String, |
| fieldNames: Array[String], |
| fieldTypes: Array[DataType], |
| tableSink: TableSink[_]): Unit = { |
| registerTableSinkInternal(name, fieldNames, fieldTypes, tableSink, false) |
| } |
| |
| /** |
| * Registers or replace an external [[TableSink]] with given field names and types in this |
| * [[TableEnvironment]]'s catalog. |
| * Registered sink tables can be referenced in SQL DML statements. |
| * |
| * @param name The name under which the [[TableSink]] is registered. |
| * @param fieldNames The field names to register with the [[TableSink]]. |
| * @param fieldTypes The field types to register with the [[TableSink]]. |
| * @param tableSink The [[TableSink]] to register. |
| */ |
| def registerOrReplaceTableSink( |
| name: String, |
| fieldNames: Array[String], |
| fieldTypes: Array[DataType], |
| tableSink: TableSink[_]): Unit = { |
| registerTableSinkInternal(name, fieldNames, fieldTypes, tableSink, true) |
| } |
| |
| /** |
| * Registers or replace an external [[TableSink]] with given field names and types in this |
| * [[TableEnvironment]]'s catalog. |
| * Registered sink tables can be referenced in SQL DML statements. |
| * |
| * @param name The name under which the [[TableSink]] is registered. |
| * @param fieldNames The field names to register with the [[TableSink]]. |
| * @param fieldTypes The field types to register with the [[TableSink]]. |
| * @param tableSink The [[TableSink]] to register. |
| * @param replace Whether replace this [[TableSink]]. |
| */ |
| protected def registerTableSinkInternal( |
| name: String, |
| fieldNames: Array[String], |
| fieldTypes: Array[DataType], |
| tableSink: TableSink[_], |
| replace: Boolean): Unit |
| |
| /** |
| * Registers an external [[TableSink]] with already configured field names and field types in |
| * this [[TableEnvironment]]'s catalog. |
| * Registered sink tables can be referenced in SQL DML statements. |
| * |
| * @param name The name under which the [[TableSink]] is registered. |
| * @param configuredSink The configured [[TableSink]] to register. |
| */ |
| def registerTableSink(name: String, configuredSink: TableSink[_]): Unit = { |
| registerTableSinkInternal(name, configuredSink, false) |
| } |
| |
| /** |
| * Registers or replace an external [[TableSink]] with already configured field names and |
| * field types in this [[TableEnvironment]]'s catalog. |
| * Registered sink tables can be referenced in SQL DML statements. |
| * |
| * @param name The name under which the [[TableSink]] is registered. |
| * @param configuredSink The configured [[TableSink]] to register. |
| */ |
| def registerOrReplaceTableSink(name: String, configuredSink: TableSink[_]): Unit = { |
| registerTableSinkInternal(name, configuredSink, true) |
| } |
| |
| /** |
| * Registers an external [[TableSink]] with already configured field names and field types in |
| * this [[TableEnvironment]]'s catalog. |
| * Registered sink tables can be referenced in SQL DML statements. |
| * |
| * @param name The name under which the [[TableSink]] is registered. |
| * @param configuredSink The configured [[TableSink]] to register. |
| */ |
| protected def registerTableSinkInternal(name: String, |
| configuredSink: TableSink[_], |
| replace: Boolean): Unit |
| |
| |
| private[flink] def getStateTableNameForWrite(name: String): String = { |
| s"__W_$name" |
| } |
| |
| private[flink] def collect[T]( |
| table: Table, |
| sink: CollectTableSink[T], |
| jobName: Option[String]): Seq[T] = { |
| throw new TableException(s"collect is not supported.") |
| } |
| |
| /** |
| * Scans a registered table and returns the resulting [[Table]]. |
| * |
| * A table to scan must be registered in the catalog. It can be either directly |
| * registered as DataStream, DataSet, or Table or as member of an [[ReadableCatalog]]. |
| * |
| * 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(catalogManager.resolveTableName(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] = { |
| val tableOpt = getTable(tablePath) |
| if (tableOpt.nonEmpty) { |
| Some(new Table(this, CatalogNode(tablePath, tableOpt.get.getRowType(typeFactory)))) |
| } else { |
| None |
| } |
| } |
| |
| /** |
| * Creates a table source and/or table sink from a descriptor. |
| * |
| * Descriptors allow for declaring the communication to external systems in an |
| * implementation-agnostic way. The classpath is scanned for suitable table factories that match |
| * the desired configuration. |
| * |
| * The following example shows how to read from a connector using a JSON format and |
| * registering a table source as "MyTable": |
| * |
| * {{{ |
| * |
| * tableEnv |
| * .connect( |
| * new ExternalSystemXYZ() |
| * .version("0.11")) |
| * .withFormat( |
| * new Json() |
| * .jsonSchema("{...}") |
| * .failOnMissingField(false)) |
| * .withSchema( |
| * new Schema() |
| * .field("user-name", "VARCHAR").from("u_name") |
| * .field("count", "DECIMAL") |
| * .registerSource("MyTable") |
| * }}} |
| * |
| * @param connectorDescriptor connector descriptor describing the external system |
| */ |
| def connect(connectorDescriptor: ConnectorDescriptor): TableDescriptor |
| |
| private def getSchema(schemaPath: Array[String]): SchemaPlus = { |
| var schema = currentSchema |
| for (schemaName <- schemaPath) { |
| schema = schema.getSubSchema(schemaName) |
| if (schema == null) { |
| return schema |
| } |
| } |
| schema |
| } |
| |
| /** |
| * Gets the names of all catalogs registered in this environment. |
| * |
| * @return A list of the names of all registered catalogs. |
| */ |
| def listCatalogs(): Array[String] = { |
| catalogManager.getCatalogs.asScala.toArray |
| } |
| |
| /** |
| * Gets the names of all databases registered in the default catalog. |
| * |
| * @return A list of the names of all registered databases. |
| */ |
| def listDatabases(): Array[String] = { |
| catalogManager.getDefaultCatalog.listDatabases().asScala.toArray |
| } |
| |
| /** |
| * Gets the names of all tables registered in the default database. |
| * |
| * @return A list of the names of all registered tables. |
| */ |
| def listTables(): Array[String] = { |
| catalogManager.getDefaultCatalog.listTables(catalogManager.getDefaultDatabaseName) |
| .map(op => op.getObjectName) |
| .toArray |
| } |
| |
| /** |
| * Gets the names of all functions registered in this environment. |
| */ |
| def listUserDefinedFunctions(): Array[String] = { |
| chainedFunctionCatalog.getSqlOperatorTable.getOperatorList.map(e => e.getName).toArray |
| } |
| |
| /** |
| * Returns the AST of the specified Table API and SQL queries and the execution plan to compute |
| * the result of the given [[Table]]. |
| * |
| * @param table The table for which the AST and execution plan will be returned. |
| */ |
| def explain(table: Table): String |
| |
| /** |
| * Explain the whole plan only when subsection optimization is supported, and returns the AST |
| * of the specified Table API and SQL queries and the execution plan. |
| * |
| * @param extended Flag to include detailed optimizer estimates. |
| */ |
| def explain(extended: Boolean = false): String |
| |
| /** |
| * 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 = { |
| // parse the sql query |
| val parsed = flinkPlanner.parse(query) |
| if (null != parsed && parsed.getKind.belongsTo(SqlKind.QUERY)) { |
| // validate the sql query |
| val validated = flinkPlanner.validate(parsed) |
| // transform to a relational tree |
| val relational = flinkPlanner.rel(validated) |
| new Table(this, LogicalRelNode(relational.project())) |
| } else { |
| throw new TableException( |
| "Unsupported SQL query! sqlQuery() only accepts SQL queries of type " + |
| "SELECT, UNION, INTERSECT, EXCEPT, VALUES, and ORDER_BY.") |
| } |
| } |
| |
| /** |
| * Returns specific FlinkCostFactory of TableEnvironment's subclass. |
| */ |
| protected def getFlinkCostFactory: FlinkCostFactory |
| |
| /** |
| * Evaluates a SQL statement such as INSERT, UPDATE or DELETE; or a DDL statement; |
| * NOTE: Currently only SQL INSERT statements are supported. |
| * |
| * 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: |
| * |
| * {{{ |
| * // register the table sink into which the result is inserted. |
| * tEnv.registerTableSink("sinkTable", fieldNames, fieldsTypes, tableSink) |
| * val sourceTable: Table = ... |
| * // sourceTable is not registered to the table environment |
| * tEnv.sqlUpdate(s"INSERT INTO sinkTable SELECT * FROM $sourceTable") |
| * }}} |
| * |
| * @param stmt The SQL statement to evaluate. |
| */ |
| def sqlUpdate(stmt: String): Unit = { |
| // parse the sql query |
| val parsed = flinkPlanner.parse(stmt) |
| parsed match { |
| case insert: SqlInsert => |
| if (insert.getTargetTable.isInstanceOf[SqlIdentifier] && |
| insert.getTargetTable.asInstanceOf[SqlIdentifier].toString.equals("console") && |
| getTable("console").isEmpty) { |
| val source = flinkPlanner.validate(insert.getSource) |
| val queryResult = new Table(this, LogicalRelNode(flinkPlanner.rel(source).rel)) |
| val schema = queryResult.getSchema |
| val printTableSink = new PrintTableSink(getConfig.getTimeZone).configure( |
| schema.getColumnNames, schema.getTypes.asInstanceOf[Array[DataType]]) |
| writeToSink(queryResult, printTableSink, "console") |
| return |
| } |
| // validate the insert sql |
| val validated = flinkPlanner.validate(insert) |
| // transform to a relational tree |
| val relational:LogicalTableModify = flinkPlanner.rel(validated).rel |
| .asInstanceOf[LogicalTableModify] |
| |
| // get query result as Table |
| val queryResult = new Table(this, LogicalRelNode(relational.getInput(0))) |
| |
| // get name of sink table |
| val targetTable = relational.getTable |
| |
| // set emit configs |
| val emit = insert.getEmit |
| if (emit != null && this.isInstanceOf[StreamTableEnvironment]) { |
| if (emit.getBeforeDelayValue >= 0) { |
| getConfig.withEarlyFireInterval(Time.milliseconds(emit.getBeforeDelayValue)) |
| } |
| if (emit.getAfterDelayValue >= 0) { |
| getConfig.withLateFireInterval(Time.milliseconds(emit.getAfterDelayValue)) |
| } |
| } |
| |
| // insert query result into sink table |
| insertInto( |
| queryResult, |
| targetTable.unwrap(classOf[schema.Table]), |
| StringUtils.join(targetTable.getQualifiedName, ",")) |
| case _ => |
| throw new TableException( |
| "Unsupported SQL query! sqlUpdate() only accepts SQL statements of type INSERT.") |
| } |
| } |
| |
| /** |
| * Evaluates a SQL statement such as INSERT, UPDATE or DELETE; or a DDL statement; |
| * NOTE: Currently only SQL INSERT statements are supported. |
| * |
| * 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: |
| * |
| * {{{ |
| * // register the table sink into which the result is inserted. |
| * tEnv.registerTableSink("sinkTable", fieldNames, fieldsTypes, tableSink) |
| * val sourceTable: Table = ... |
| * // sourceTable is not registered to the table environment |
| * tEnv.sqlUpdate(s"INSERT INTO sinkTable SELECT * FROM $sourceTable") |
| * }}} |
| * |
| * @param stmt The SQL statement to evaluate. |
| * @param config The [[QueryConfig]] to use. |
| */ |
| def sqlUpdate(stmt: String, config: QueryConfig): Unit = { |
| config.overrideTableConfig(getConfig) |
| sqlUpdate(stmt) |
| } |
| |
| /** |
| * Writes a [[Table]] to a [[TableSink]]. |
| * |
| * @param table The [[Table]] to write. |
| * @param sink The [[TableSink]] to write the [[Table]] to. |
| * @tparam T The data type that the [[TableSink]] expects. |
| */ |
| private[table] def writeToSink[T]( |
| table: Table, |
| sink: TableSink[T], |
| sinkName: String = null): Unit |
| |
| /** |
| * Triggers the program execution. |
| */ |
| def execute(): JobExecutionResult = execute(DEFAULT_JOB_NAME) |
| |
| /** |
| * Triggers the program execution with jobName. |
| */ |
| def execute(jobName: String): JobExecutionResult |
| |
| /** |
| * Writes the [[Table]] to a [[TableSink]] that was registered under the specified name. |
| * |
| * @param table The table to write to the TableSink. |
| * @param sinkTableName The name of the registered TableSink. |
| */ |
| private[flink] def insertInto(table: Table, sinkTableName: String): Unit = { |
| |
| // check that sink table exists |
| if (null == sinkTableName || sinkTableName.isEmpty) { |
| throw new TableException(TableErrors.INST.sqlInvalidSinkTblName()) |
| } |
| |
| if (!catalogManager.getDefaultCatalog() |
| .tableExists(new ObjectPath(catalogManager.getDefaultDatabaseName, sinkTableName))) { |
| throw new TableException(TableErrors.INST.sqlTableNotRegistered(sinkTableName)) |
| } |
| val targetTable = getTable(sinkTableName).get |
| |
| insertInto(table, targetTable, sinkTableName) |
| } |
| |
| private def insertInto( |
| sourceTable: Table, |
| targetTable: schema.Table, |
| targetTableName: String) = { |
| val tableSink = targetTable match { |
| case s: CatalogCalciteTable => s.tableSink |
| case s: TableSinkTable[_] => s.tableSink |
| case s: TableSourceSinkTable[_] if s.tableSinkTable.isDefined => |
| s.tableSinkTable.get.tableSink |
| case _ => |
| throw new TableException(TableErrors.INST.sqlNotTableSinkError(targetTableName)) |
| |
| } |
| |
| // validate schema of source table and table sink |
| val srcFieldTypes = sourceTable.getSchema.getTypes |
| val sinkFieldTypes = tableSink.getFieldTypes.map(_.toInternalType) |
| |
| val srcFieldNames = sourceTable.getSchema.getColumnNames |
| val sinkFieldNames = tableSink.getFieldNames |
| |
| val srcNameTypes = srcFieldNames.zip(srcFieldTypes) |
| val sinkNameTypes = sinkFieldNames.zip(sinkFieldTypes) |
| |
| def typeMatch(t1: InternalType, t2: InternalType): Boolean = { |
| t1 == t2 || |
| (t1.isInstanceOf[DateType] && t2.isInstanceOf[DateType]) || |
| (t1.isInstanceOf[TimestampType] && t2.isInstanceOf[TimestampType]) |
| |
| } |
| |
| if (srcFieldTypes.length != sinkFieldTypes.length) { |
| // format table and table sink schema strings |
| val srcSchema = srcNameTypes |
| .map { case (n, t) => s"$n: ${TypeUtils.getExternalClassForType(t)}" } |
| .mkString("[", ", ", "]") |
| |
| val sinkSchema = sinkNameTypes |
| .map { case (n, t) => s"$n: ${TypeUtils.getExternalClassForType(t)}" } |
| .mkString("[", ", ", "]") |
| |
| throw new ValidationException( |
| TableErrors.INST.sqlInsertIntoMismatchedFieldLen( |
| targetTableName, srcSchema, sinkSchema)) |
| } else if (srcFieldTypes.zip(sinkFieldTypes) |
| .exists { |
| case (_: GenericType[_], _: GenericType[_]) => false |
| case (srcF, snkF) => !typeMatch(srcF, snkF) |
| } |
| ) { |
| val diffNameTypes = srcNameTypes.zip(sinkNameTypes) |
| .filter { |
| case ((_, srcType), (_, sinkType)) => !typeMatch(srcType, sinkType) |
| } |
| val srcDiffMsg = diffNameTypes |
| .map(_._1) |
| .map { case (n, t) => s"$n: ${TypeUtils.getExternalClassForType(t)}" } |
| .mkString("[", ", ", "]") |
| val sinkDiffMsg = diffNameTypes |
| .map(_._2) |
| .map { case (n, t) => s"$n: ${TypeUtils.getExternalClassForType(t)}" } |
| .mkString("[", ", ", "]") |
| |
| throw new ValidationException( |
| TableErrors.INST.sqlInsertIntoMismatchedFieldTypes( |
| targetTableName, srcDiffMsg, sinkDiffMsg)) |
| } |
| |
| // emit the table to the configured table sink |
| writeToSink(sourceTable, tableSink, targetTableName) |
| } |
| |
| /** |
| * 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. |
| * @param replace Whether to replace the registered table. |
| * @throws TableException if another table is registered under the provided name. |
| */ |
| @throws[TableException] |
| protected def registerTableInternal(name: String, |
| table: AbstractTable, |
| replace: Boolean): Unit = { |
| if (replace) { |
| catalogManager.getCatalog(catalogManager.getDefaultCatalogName) |
| .asInstanceOf[ReadableWritableCatalog] |
| .dropTable(new ObjectPath(catalogManager.getDefaultDatabaseName, name), true) |
| } |
| catalogManager.getCatalog(catalogManager.getDefaultCatalogName) |
| .asInstanceOf[ReadableWritableCatalog] |
| .createTable( |
| new ObjectPath(catalogManager.getDefaultDatabaseName, name), |
| createFlinkTempTable(table), |
| false |
| ) |
| } |
| |
| /** |
| * Checks if the chosen table name is valid. |
| * |
| * @param name The table name to check. |
| */ |
| protected def checkValidTableName(name: String): Unit = {} |
| |
| /** |
| * Close the table environment. This method will clean up the internal state and background |
| * services. Users should invoke this method if possible to avoid resource leak. |
| */ |
| def close(): Unit = { |
| if (closed.compareAndSet(false, true)) { |
| tableServiceManager.close() |
| } |
| } |
| |
| /** Returns a unique temporary attribute name. */ |
| private[flink] def createUniqueAttributeName(prefix: String): String = { |
| prefix + attrNameCntr.getAndIncrement() |
| } |
| |
| /** Returns a unique temporary attribute name. */ |
| private[flink] def createUniqueAttributeName(): String = { |
| "TMP_" + attrNameCntr.getAndIncrement() |
| } |
| |
| /** Returns a unique table name according to the internal naming pattern. */ |
| private[flink] def createUniqueTableName(): String = { |
| var res = tableNamePrefix + tableNameCntr.getAndIncrement() |
| while (getTable(res).nonEmpty) { |
| res = tableNamePrefix + tableNameCntr.getAndIncrement() |
| } |
| res |
| } |
| |
| /** 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 chained [[FunctionCatalog]]. */ |
| private[flink] def getFunctionCatalog: FunctionCatalog = { |
| chainedFunctionCatalog |
| } |
| |
| private def createFrameworkConfig: FrameworkConfig = { |
| Frameworks |
| .newConfigBuilder |
| .defaultSchema(currentSchema) |
| .parserConfig(getSqlParserConfig) |
| .costFactory(getFlinkCostFactory) |
| .operatorTable(getSqlOperatorTable) |
| // set the executor to evaluate constant expressions |
| .executor(new ExpressionReducer(config)) |
| .context(FlinkChainContext.chain(Contexts.of(config), Contexts.of(chainedFunctionCatalog))) |
| .build |
| } |
| |
| /** Returns the Calcite [[FrameworkConfig]] of this TableEnvironment. */ |
| private[flink] def getFrameworkConfig: FrameworkConfig = { |
| frameworkConfig |
| } |
| |
| protected def createRelBuilder: FlinkRelBuilder = { |
| FlinkRelBuilder.create( |
| frameworkConfig, config, getTypeFactory, catalogManager = catalogManager) |
| } |
| |
| private def createRelOptPlanner: RelOptPlanner = { |
| relBuilder.getPlanner |
| } |
| |
| private def createFlinkPlanner: FlinkPlannerImpl = { |
| new FlinkPlannerImpl( |
| getFrameworkConfig, |
| getPlanner, |
| getTypeFactory, |
| sqlToRelConverterConfig, |
| relBuilder.getCluster, |
| catalogManager) |
| } |
| |
| /** |
| * 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. |
| */ |
| protected def isReferenceByPosition(ct: RowType, fields: Array[Expression]): Boolean = { |
| |
| 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 { |
| case UnresolvedFieldReference(name) => !inputNames.contains(name) |
| case Alias(_, _, _) => false |
| case _ => true |
| } |
| } |
| |
| /** |
| * Returns field names and field positions for a given [[TypeInformation]]. |
| * |
| * @param inputType The DataType extract the field names and positions from. |
| * @return A tuple of two arrays holding the field names and corresponding field positions. |
| */ |
| protected[flink] def getFieldInfo(inputType: DataType): |
| (Array[String], Array[Int]) = { |
| |
| (TableEnvironment.getFieldNames(inputType), TableEnvironment.getFieldIndices(inputType)) |
| } |
| |
| /** |
| * Returns field names and field positions for a given [[TypeInformation]] and [[Array]] of |
| * [[Expression]]. It does not handle time attributes but considers them in indices. |
| * |
| * @param inputType The [[DataType]] against which the [[Expression]]s are evaluated. |
| * @param exprs The expressions that define the field names. |
| * @return A tuple of two arrays holding the field names and corresponding field positions. |
| */ |
| protected[flink] def getFieldInfo[A]( |
| inputType: DataType, |
| exprs: Array[Expression]) |
| : (Array[String], Array[Int]) = { |
| |
| TableEnvironment.validateType(inputType) |
| |
| def referenceByName(name: String, ct: RowType): 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(", ")}. " + |
| s"Make sure there is no field in physical data type referred " + |
| s"if you want to refer field by position.") |
| } else { |
| Some(inputIdx) |
| } |
| } |
| |
| val indexedNames: Array[(Int, String)] = inputType.toInternalType match { |
| |
| case t: RowType => |
| |
| val isRefByPos = isReferenceByPosition(t, exprs) |
| exprs.zipWithIndex flatMap { |
| case (UnresolvedFieldReference(name: String), idx) => |
| if (isRefByPos) { |
| Some((idx, name)) |
| } else { |
| referenceByName(name, t).map((_, name)) |
| } |
| case (Alias(UnresolvedFieldReference(origName), name: String, _), _) => |
| if (isRefByPos) { |
| throw new TableException( |
| s"Alias '$name' is not allowed if other fields are referenced by position.") |
| } else { |
| referenceByName(origName, t).map((_, name)) |
| } |
| case (_: TimeAttribute, _) => |
| None |
| case _ => throw new TableException( |
| "Field reference expression or alias on field expression expected.") |
| } |
| |
| case _: InternalType => // atomic or other custom type information |
| var referenced = false |
| exprs flatMap { |
| case _: TimeAttribute => |
| None |
| case UnresolvedFieldReference(_) if referenced => |
| // only accept the first field for an atomic type |
| throw new TableException("Only the first field can reference an atomic type.") |
| case UnresolvedFieldReference(name: String) => |
| referenced = true |
| // first field reference is mapped to atomic type |
| Some((0, name)) |
| case _ => throw new TableException( |
| "Field reference expression expected.") |
| } |
| } |
| |
| val (fieldIndexes, fieldNames) = indexedNames.unzip |
| |
| if (fieldNames.contains("*")) { |
| throw new TableException("Field name can not be '*'.") |
| } |
| |
| (fieldNames, fieldIndexes) |
| } |
| |
| def setUserClassLoader(userClassLoader: ClassLoader): Unit = { |
| this.userClassloader = userClassLoader |
| } |
| } |
| |
| /** |
| * Object to instantiate a [[TableEnvironment]] depending on the batch or stream execution |
| * environment. |
| */ |
| object TableEnvironment { |
| |
| /** |
| * The key for external catalog |
| */ |
| val DEFAULT_SCHEMA: String = "hive" |
| |
| /** |
| * Returns a [[BatchTableEnvironment]] for a Java [[JavaStreamExecEnv]]. |
| * |
| * @param executionEnvironment The Java batch ExecutionEnvironment. |
| */ |
| def getBatchTableEnvironment( |
| executionEnvironment: JavaStreamExecEnv): JavaBatchTableEnvironment = { |
| new JavaBatchTableEnvironment(executionEnvironment, new TableConfig()) |
| } |
| |
| /** |
| * Returns a [[BatchTableEnvironment]] for a Java [[JavaStreamExecEnv]] and a given |
| * [[TableConfig]]. |
| * |
| * @param executionEnvironment The Java batch ExecutionEnvironment. |
| * @param tableConfig The TableConfig for the new TableEnvironment. |
| */ |
| def getBatchTableEnvironment( |
| executionEnvironment: JavaStreamExecEnv, |
| tableConfig: TableConfig): JavaBatchTableEnvironment = { |
| new JavaBatchTableEnvironment(executionEnvironment, tableConfig) |
| } |
| |
| /** |
| * Returns a [[ScalaBatchTableEnvironment]] for a Scala stream [[ScalaStreamExecEnv]]. |
| * |
| * @param executionEnvironment The Scala StreamExecutionEnvironment. |
| */ |
| def getBatchTableEnvironment( |
| executionEnvironment: ScalaStreamExecEnv): ScalaBatchTableEnvironment = { |
| new ScalaBatchTableEnvironment(executionEnvironment, new TableConfig()) |
| } |
| |
| /** |
| * Returns a [[ScalaBatchTableEnvironment]] for a Scala stream [[ScalaStreamExecEnv]]. |
| * |
| * @param executionEnvironment The Scala StreamExecutionEnvironment. |
| * @param tableConfig The TableConfig for the new TableEnvironment. |
| */ |
| def getBatchTableEnvironment( |
| executionEnvironment: ScalaStreamExecEnv, |
| tableConfig: TableConfig): ScalaBatchTableEnvironment = { |
| |
| new ScalaBatchTableEnvironment(executionEnvironment, tableConfig) |
| } |
| |
| /** |
| * Returns a [[JavaStreamTableEnv]] for a Java [[JavaStreamExecEnv]]. |
| * |
| * @param executionEnvironment The Java StreamExecutionEnvironment. |
| */ |
| def getTableEnvironment(executionEnvironment: JavaStreamExecEnv): JavaStreamTableEnv = { |
| new JavaStreamTableEnv(executionEnvironment, new TableConfig()) |
| } |
| |
| /** |
| * Returns a [[JavaStreamTableEnv]] for a Java [[JavaStreamExecEnv]] and a given [[TableConfig]]. |
| * |
| * @param executionEnvironment The Java StreamExecutionEnvironment. |
| * @param tableConfig The TableConfig for the new TableEnvironment. |
| */ |
| def getTableEnvironment( |
| executionEnvironment: JavaStreamExecEnv, |
| tableConfig: TableConfig): JavaStreamTableEnv = { |
| |
| new JavaStreamTableEnv(executionEnvironment, tableConfig) |
| } |
| |
| /** |
| * Returns a [[ScalaStreamTableEnv]] for a Scala stream [[ScalaStreamExecEnv]]. |
| * |
| * @param executionEnvironment The Scala StreamExecutionEnvironment. |
| */ |
| def getTableEnvironment(executionEnvironment: ScalaStreamExecEnv): ScalaStreamTableEnv = { |
| new ScalaStreamTableEnv(executionEnvironment, new TableConfig()) |
| } |
| |
| /** |
| * Returns a [[ScalaStreamTableEnv]] for a Scala stream [[ScalaStreamExecEnv]]. |
| * |
| * @param executionEnvironment The Scala StreamExecutionEnvironment. |
| * @param tableConfig The TableConfig for the new TableEnvironment. |
| */ |
| def getTableEnvironment( |
| executionEnvironment: ScalaStreamExecEnv, |
| tableConfig: TableConfig): ScalaStreamTableEnv = { |
| |
| new ScalaStreamTableEnv(executionEnvironment, tableConfig) |
| } |
| |
| /** |
| * Validate if class represented by the typeInfo is static and globally accessible |
| * @param t type to check |
| * @throws TableException if type does not meet these criteria |
| */ |
| def validateType(t: DataType): Unit = { |
| val clazz = TypeUtils.getExternalClassForType(t) |
| if ((clazz.isMemberClass && !Modifier.isStatic(clazz.getModifiers)) || |
| !Modifier.isPublic(clazz.getModifiers) || |
| clazz.getCanonicalName == null) { |
| throw new TableException(s"Class '$clazz' described in type information '$t' must be " + |
| s"static and globally accessible.") |
| } |
| } |
| |
| /** |
| * Return rowType of tableSink. [[UpsertStreamTableSink]] and [[RetractStreamTableSink]] should |
| * return recordType, others return outputType. |
| * @param tableSink |
| * @tparam A |
| * @return |
| */ |
| def getRowTypeForTableSink[A](tableSink: TableSink[A]): DataType = { |
| tableSink match { |
| case u: UpsertStreamTableSink[A] => u.getRecordType |
| case r: RetractStreamTableSink[A] => r.getRecordType |
| case _ => tableSink.getOutputType |
| } |
| } |
| |
| /** |
| * Returns field names for a given [[TypeInformation]]. |
| * |
| * @param inputType The DataType extract the field names. |
| * @return An array holding the field names |
| */ |
| def getFieldNames(inputType: DataType): Array[String] = { |
| validateType(inputType) |
| |
| val fieldNames: Array[String] = inputType.toInternalType match { |
| case t: RowType => t.getFieldNames |
| case _: InternalType => Array("f0") |
| } |
| |
| if (fieldNames.contains("*")) { |
| throw new TableException("Field name can not be '*'.") |
| } |
| |
| fieldNames |
| } |
| |
| /** |
| * Returns field indexes for a given [[TypeInformation]]. |
| * |
| * @param inputType The DataType extract the field positions from. |
| * @return An array holding the field positions |
| */ |
| def getFieldIndices(inputType: DataType): Array[Int] = { |
| getFieldNames(inputType).indices.toArray |
| } |
| |
| /** |
| * Returns field types for a given [[TypeInformation]]. |
| * |
| * @param inputType The DataType to extract field types from. |
| * @return An array holding the field types. |
| */ |
| def getFieldTypes(inputType: DataType): Array[InternalType] = { |
| validateType(inputType) |
| |
| inputType.toInternalType match { |
| case ct: RowType => |
| 0.until(ct.getArity).map(i => ct.getInternalTypeAt(i).toInternalType).toArray |
| case t: InternalType => Array(t) |
| } |
| } |
| |
| } |
