wuchong commented on a change in pull request #8844: [FLINK-12951][table-planner] Add logic to bridge DDL to table source(… URL: https://github.com/apache/flink/pull/8844#discussion_r299372066
########## File path: flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/catalog/CalciteCatalogTable.scala ########## @@ -0,0 +1,156 @@ +/* + * 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.catalog + +import org.apache.flink.table.api.TableSchema +import org.apache.flink.table.calcite.FlinkTypeFactory +import org.apache.flink.table.descriptors.{ConnectorDescriptor, ConnectorDescriptorValidator, DescriptorProperties, Schema} +import org.apache.flink.table.factories.TableFactoryUtil +import org.apache.flink.table.sinks.TableSink +import org.apache.flink.table.sources.TableSource + +import org.apache.calcite.config.CalciteConnectionConfig +import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory} +import org.apache.calcite.schema.{Statistic, Statistics, Table, Wrapper, Schema => CSchema} +import org.apache.calcite.sql.{SqlCall, SqlNode} + +import java.util.{Objects, Optional, ArrayList => JArrayList, HashMap => JHashMap, List => JList, Map => JMap} + +/** + * A Calcite [[org.apache.calcite.schema.Table]] which also inherits [[CatalogTable]], + * it is used to bridge the [[CatalogTable]] and + * [[org.apache.flink.table.sources.TableSource]]/[[org.apache.flink.table.sinks.TableSink]]. + * + * <p>This table usually comes from a sql DDL statement, it would be registered as + * a [[CatalogTable]] into the [[CatalogManager]] + * (See [[org.apache.flink.table.sqlexec.SqlExecutableStatements]] for details); + * + * <p> This table would be transformed to table source/sink based on the sql context where + * is it referenced: + * a.) If this table is referenced as a source table, it would be transformed to a + * [[org.apache.flink.table.plan.schema.TableSourceTable]] to use during planning phrase, + * right after we finish the sql to rel conversion; + * b.) If this table is referenced as a sink table, it would be transformed to a [[TableSink]] + * when we update through a DML, see [[org.apache.flink.table.api.TableEnvironment#sql]]. + * + * <p>Notes: + * 1. The schema of CatalogTable can be different in different execution environment since there is + * no TimeIndicator in batch table and temporal table. + * 2. The computed columns of CatalogTable are registered as virtual columns for calcite + * which cannot be update in dml query. + * + * @param logicalSchema TableSchema + * @param partitionKeys partition keys + * @param properties table properties + * @param comment table comment + */ +class CalciteCatalogTable( + val logicalSchema: TableSchema, + val partitionKeys: JList[String], + val properties: JMap[String, String], + val comment: String) + extends AbstractCatalogTable(logicalSchema.physicalSchema(), + partitionKeys, properties, comment) + with Table + with Wrapper { + + /** + * Converts a [[CalciteCatalogTable]] to a [[TableSource]]. + * + * <p>Caution that this table source has table schema excluding the virtual columns. + * + * @return converted [[TableSource]] instance from the input catalog table + */ + def toTableSource: Optional[TableSource[_]] = { + val properties = toProperties + val connectorType = properties.get(ConnectorDescriptorValidator.CONNECTOR) + Objects.requireNonNull(connectorType, + s"Option [${ConnectorDescriptorValidator.CONNECTOR}]" + + s" must be specified for with options!") + val descriptor = new SimpleDescriptor(connectorType, properties) + Optional.of(TableFactoryUtil.findAndCreateTableSource(descriptor)) + } + + /** + * Converts a [[CalciteCatalogTable]] to a [[TableSink]]. + * + * <p>Caution that this table sink has table schema excluding the virtual columns. + * + * @return converted [[TableSource]] instance from the input catalog table + */ + def toTableSink: Optional[TableSink[_]] = { + val properties = toProperties + val connectorType = properties.get(ConnectorDescriptorValidator.CONNECTOR) + Objects.requireNonNull(connectorType, + s"Option [${ConnectorDescriptorValidator.CONNECTOR}]" + + s" must be specified for with options!") + val descriptor = new SimpleDescriptor(connectorType, properties) + Optional.of(TableFactoryUtil.findAndCreateTableSink(descriptor)) + } + + override def getRowType(relDataTypeFactory: RelDataTypeFactory): RelDataType = { + relDataTypeFactory.asInstanceOf[FlinkTypeFactory].buildLogicalRowType(logicalSchema) + } + + override def copy(): CatalogBaseTable = new CalciteCatalogTable( + logicalSchema.copy(), + new JArrayList[String](getPartitionKeys), + new JHashMap[String, String](getProperties), + getComment) + + override def getDescription: Optional[String] = Optional.of(getComment) + + override def getDetailedDescription: Optional[String] = Optional.of("This is a" + Review comment: Add a space after `a` ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services
