You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2019/06/04 09:56:47 UTC
[flink] branch master updated (0f41719 -> 88c7d82)
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 0f41719 [FLINK-12601][table] Wrapping DataStream & DataSet as TableOperations.
new 99d4fd3 [hotfix][table-api] Port TableSource specific descriptors
new 668259c [hotfix][table-planner] Fix getting external catalog
new 88c7d82 [FLINK-12604][table] Register a TableSource/Sink directly as CatalogTables
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/flink/table/catalog/CatalogManager.java | 80 +++++--
.../flink/table/catalog/ConnectorCatalogTable.java | 163 ++++++++++++++
.../table/catalog/TableOperationCatalogView.java | 2 +-
.../operations/TableOperationDefaultVisitor.java | 5 +
.../table/operations/TableOperationVisitor.java | 2 +
...eration.java => TableSourceTableOperation.java} | 43 ++--
.../flink/table/typeutils/FieldInfoUtils.java | 23 +-
.../table/sources/DefinedProctimeAttribute.java | 23 +-
.../DefinedRowtimeAttributes.java} | 30 +--
.../table/sources/RowtimeAttributeDescriptor.java | 77 +++++++
.../flink/table/catalog/CalciteCatalogTable.java | 88 --------
.../flink/table/catalog/DatabaseCalciteSchema.java | 16 +-
.../flink/table/plan/TableOperationConverter.java | 34 +++
.../apache/flink/table/api/BatchTableEnvImpl.scala | 153 +------------
.../flink/table/api/StreamTableEnvImpl.scala | 142 +-----------
.../org/apache/flink/table/api/TableEnvImpl.scala | 244 ++++++++++++++-------
.../flink/table/calcite/FlinkRelBuilder.scala | 8 +-
.../table/catalog/ExternalCatalogSchema.scala | 2 +-
.../flink/table/catalog/ExternalTableUtil.scala | 48 ++--
.../table/plan/nodes/PhysicalTableSourceScan.scala | 14 +-
.../plan/nodes/dataset/BatchTableSourceScan.scala | 2 +-
.../nodes/datastream/StreamTableSourceScan.scala | 2 +-
.../logical/FlinkLogicalTableSourceScan.scala | 21 +-
.../rules/dataSet/BatchTableSourceScanRule.scala | 10 +-
.../datastream/StreamTableSourceScanRule.scala | 10 +-
.../table/plan/schema/BatchTableSourceTable.scala | 43 ----
.../table/plan/schema/StreamTableSourceTable.scala | 42 ----
.../flink/table/plan/schema/TableSinkTable.scala | 49 -----
.../table/plan/schema/TableSourceSinkTable.scala | 67 ------
.../flink/table/plan/schema/TableSourceTable.scala | 32 ++-
.../table/sources/definedTimeAttributes.scala | 95 --------
.../table/catalog/CatalogStructureBuilder.java | 45 ++--
.../flink/table/catalog/PathResolutionTest.java | 3 +-
.../batch/table/JavaTableEnvironmentITCase.java | 34 ++-
.../table/catalog/ExternalCatalogSchemaTest.scala | 11 +-
35 files changed, 708 insertions(+), 955 deletions(-)
create mode 100644 flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ConnectorCatalogTable.java
copy flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/{CatalogTableOperation.java => TableSourceTableOperation.java} (56%)
copy flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/hadoop/mapred/wrapper/HadoopDummyProgressable.java => flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/DefinedProctimeAttribute.java (59%)
copy flink-table/flink-table-common/src/main/java/org/apache/flink/table/{expressions/Expression.java => sources/DefinedRowtimeAttributes.java} (57%)
create mode 100644 flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/RowtimeAttributeDescriptor.java
delete mode 100644 flink-table/flink-table-planner/src/main/java/org/apache/flink/table/catalog/CalciteCatalogTable.java
delete mode 100644 flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/BatchTableSourceTable.scala
delete mode 100644 flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/StreamTableSourceTable.scala
delete mode 100644 flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/TableSinkTable.scala
delete mode 100644 flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/TableSourceSinkTable.scala
delete mode 100644 flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sources/definedTimeAttributes.scala
[flink] 03/03: [FLINK-12604][table] Register a TableSource/Sink
directly as CatalogTables
Posted by dw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 88c7d82a6148a491d9c11f8d4c7eedc1722ee7a9
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Fri May 24 14:10:29 2019 +0200
[FLINK-12604][table] Register a TableSource/Sink directly as CatalogTables
This closes #8549
---
.../apache/flink/table/catalog/CatalogManager.java | 80 +++++--
.../flink/table/catalog/ConnectorCatalogTable.java | 163 ++++++++++++++
.../table/catalog/TableOperationCatalogView.java | 2 +-
.../operations/TableOperationDefaultVisitor.java | 5 +
.../table/operations/TableOperationVisitor.java | 2 +
.../operations/TableSourceTableOperation.java | 75 +++++++
.../flink/table/catalog/CalciteCatalogTable.java | 88 --------
.../flink/table/catalog/DatabaseCalciteSchema.java | 16 +-
.../flink/table/plan/TableOperationConverter.java | 34 +++
.../apache/flink/table/api/BatchTableEnvImpl.scala | 153 +------------
.../flink/table/api/StreamTableEnvImpl.scala | 142 +-----------
.../org/apache/flink/table/api/TableEnvImpl.scala | 239 ++++++++++++++-------
.../flink/table/calcite/FlinkRelBuilder.scala | 8 +-
.../table/catalog/ExternalCatalogSchema.scala | 2 +-
.../flink/table/catalog/ExternalTableUtil.scala | 48 ++---
.../table/plan/nodes/PhysicalTableSourceScan.scala | 14 +-
.../plan/nodes/dataset/BatchTableSourceScan.scala | 2 +-
.../nodes/datastream/StreamTableSourceScan.scala | 2 +-
.../logical/FlinkLogicalTableSourceScan.scala | 21 +-
.../rules/dataSet/BatchTableSourceScanRule.scala | 10 +-
.../datastream/StreamTableSourceScanRule.scala | 10 +-
.../flink/table/plan/schema/TableSinkTable.scala | 49 -----
.../table/plan/schema/TableSourceSinkTable.scala | 67 ------
.../flink/table/plan/schema/TableSourceTable.scala | 32 +--
.../table/catalog/CatalogStructureBuilder.java | 45 ++--
.../flink/table/catalog/PathResolutionTest.java | 3 +-
.../batch/table/JavaTableEnvironmentITCase.java | 34 ++-
.../table/catalog/ExternalCatalogSchemaTest.scala | 11 +-
28 files changed, 634 insertions(+), 723 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
index 975c87c..5704043 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
@@ -24,7 +24,6 @@ import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
import org.apache.flink.table.factories.TableFactoryUtil;
-import org.apache.flink.table.operations.CatalogTableOperation;
import org.apache.flink.table.sinks.TableSink;
import org.apache.flink.util.StringUtils;
@@ -64,6 +63,57 @@ public class CatalogManager {
private String currentDatabaseName;
+ /**
+ * Temporary solution to handle both {@link CatalogBaseTable} and
+ * {@link ExternalCatalogTable} in a single call.
+ */
+ public static class ResolvedTable {
+ private final ExternalCatalogTable externalCatalogTable;
+ private final CatalogBaseTable catalogTable;
+ private final TableSchema tableSchema;
+ private final List<String> tablePath;
+
+ static ResolvedTable externalTable(
+ List<String> tablePath,
+ ExternalCatalogTable table,
+ TableSchema tableSchema) {
+ return new ResolvedTable(table, null, tableSchema, tablePath);
+ }
+
+ static ResolvedTable catalogTable(
+ List<String> tablePath,
+ CatalogBaseTable table) {
+ return new ResolvedTable(null, table, table.getSchema(), tablePath);
+ }
+
+ private ResolvedTable(
+ ExternalCatalogTable externalCatalogTable,
+ CatalogBaseTable catalogTable,
+ TableSchema tableSchema,
+ List<String> tablePath) {
+ this.externalCatalogTable = externalCatalogTable;
+ this.catalogTable = catalogTable;
+ this.tableSchema = tableSchema;
+ this.tablePath = tablePath;
+ }
+
+ public Optional<ExternalCatalogTable> getExternalCatalogTable() {
+ return Optional.ofNullable(externalCatalogTable);
+ }
+
+ public Optional<CatalogBaseTable> getCatalogTable() {
+ return Optional.ofNullable(catalogTable);
+ }
+
+ public TableSchema getTableSchema() {
+ return tableSchema;
+ }
+
+ public List<String> getTablePath() {
+ return tablePath;
+ }
+ }
+
public CatalogManager(String defaultCatalogName, Catalog defaultCatalog) {
checkArgument(
!StringUtils.isNullOrWhitespaceOnly(defaultCatalogName),
@@ -243,8 +293,8 @@ public class CatalogManager {
}
/**
- * Tries to resolve a table path to a {@link CatalogTableOperation}. The algorithm looks for requested table
- * in following paths in that order:
+ * Tries to resolve a table path to a {@link ResolvedTable}. The algorithm looks for requested table
+ * in the following paths in that order:
* <ol>
* <li>{@code [current-catalog].[current-database].[tablePath]}</li>
* <li>{@code [current-catalog].[tablePath]}</li>
@@ -252,10 +302,10 @@ public class CatalogManager {
* </ol>
*
* @param tablePath table path to look for
- * @return {@link CatalogTableOperation} containing both fully qualified table identifier and its
- * {@link TableSchema}.
+ * @return {@link ResolvedTable} wrapping original table with additional information about table path and
+ * unified access to {@link TableSchema}.
*/
- public Optional<CatalogTableOperation> resolveTable(String... tablePath) {
+ public Optional<ResolvedTable> resolveTable(String... tablePath) {
checkArgument(tablePath != null && tablePath.length != 0, "Table path must not be null or empty.");
List<String> userPath = asList(tablePath);
@@ -267,7 +317,7 @@ public class CatalogManager {
);
for (List<String> prefix : prefixes) {
- Optional<CatalogTableOperation> potentialTable = lookupPath(prefix, userPath);
+ Optional<ResolvedTable> potentialTable = lookupPath(prefix, userPath);
if (potentialTable.isPresent()) {
return potentialTable;
}
@@ -276,12 +326,12 @@ public class CatalogManager {
return Optional.empty();
}
- private Optional<CatalogTableOperation> lookupPath(List<String> prefix, List<String> userPath) {
+ private Optional<ResolvedTable> lookupPath(List<String> prefix, List<String> userPath) {
try {
List<String> path = new ArrayList<>(prefix);
path.addAll(userPath);
- Optional<CatalogTableOperation> potentialTable = lookupCatalogTable(path);
+ Optional<ResolvedTable> potentialTable = lookupCatalogTable(path);
if (!potentialTable.isPresent()) {
potentialTable = lookupExternalTable(path);
@@ -292,7 +342,7 @@ public class CatalogManager {
}
}
- private Optional<CatalogTableOperation> lookupCatalogTable(List<String> path) throws TableNotExistException {
+ private Optional<ResolvedTable> lookupCatalogTable(List<String> path) throws TableNotExistException {
if (path.size() == 3) {
Catalog currentCatalog = catalogs.get(path.get(0));
String currentDatabaseName = path.get(1);
@@ -300,22 +350,22 @@ public class CatalogManager {
ObjectPath objectPath = new ObjectPath(currentDatabaseName, tableName);
if (currentCatalog != null && currentCatalog.tableExists(objectPath)) {
- TableSchema tableSchema = currentCatalog.getTable(objectPath).getSchema();
- return Optional.of(new CatalogTableOperation(
+ CatalogBaseTable table = currentCatalog.getTable(objectPath);
+ return Optional.of(ResolvedTable.catalogTable(
asList(path.get(0), currentDatabaseName, tableName),
- tableSchema));
+ table));
}
}
return Optional.empty();
}
- private Optional<CatalogTableOperation> lookupExternalTable(List<String> path) {
+ private Optional<ResolvedTable> lookupExternalTable(List<String> path) {
ExternalCatalog currentCatalog = externalCatalogs.get(path.get(0));
return Optional.ofNullable(currentCatalog)
.flatMap(externalCatalog -> extractPath(externalCatalog, path.subList(1, path.size() - 1)))
.map(finalCatalog -> finalCatalog.getTable(path.get(path.size() - 1)))
- .map(table -> new CatalogTableOperation(path, getTableSchema(table)));
+ .map(table -> ResolvedTable.externalTable(path, table, getTableSchema(table)));
}
private Optional<ExternalCatalog> extractPath(ExternalCatalog rootExternalCatalog, List<String> path) {
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ConnectorCatalogTable.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ConnectorCatalogTable.java
new file mode 100644
index 0000000..ce86084
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ConnectorCatalogTable.java
@@ -0,0 +1,163 @@
+/*
+ * 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.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.sinks.TableSink;
+import org.apache.flink.table.sources.DefinedProctimeAttribute;
+import org.apache.flink.table.sources.DefinedRowtimeAttributes;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * A {@link CatalogTable} that wraps a {@link TableSource} and/or {@link TableSink}.
+ * This allows registering those in a {@link Catalog}. It can not be persisted as the
+ * source and/or sink might be inline implementations and not be representable in a
+ * property based form.
+ *
+ * @param <T1> type of the produced elements by the {@link TableSource}
+ * @param <T2> type of the expected elements by the {@link TableSink}
+ */
+@Internal
+public class ConnectorCatalogTable<T1, T2> extends AbstractCatalogTable {
+ private final TableSource<T1> tableSource;
+ private final TableSink<T2> tableSink;
+ private final boolean isBatch;
+
+ public static <T1> ConnectorCatalogTable source(TableSource<T1> source, boolean isBatch) {
+ final TableSchema tableSchema = calculateSourceSchema(source, isBatch);
+ return new ConnectorCatalogTable<>(source, null, tableSchema, isBatch);
+ }
+
+ public static <T2> ConnectorCatalogTable sink(TableSink<T2> sink, boolean isBatch) {
+ TableSchema tableSchema = new TableSchema(sink.getFieldNames(), sink.getFieldTypes());
+ return new ConnectorCatalogTable<>(null, sink, tableSchema, isBatch);
+ }
+
+ public static <T1, T2> ConnectorCatalogTable sourceAndSink(
+ TableSource<T1> source,
+ TableSink<T2> sink,
+ boolean isBatch) {
+ TableSchema tableSchema = calculateSourceSchema(source, isBatch);
+ return new ConnectorCatalogTable<>(source, sink, tableSchema, isBatch);
+ }
+
+ @VisibleForTesting
+ protected ConnectorCatalogTable(
+ TableSource<T1> tableSource,
+ TableSink<T2> tableSink,
+ TableSchema tableSchema,
+ boolean isBatch) {
+ super(tableSchema, Collections.emptyMap(), "");
+ this.tableSource = tableSource;
+ this.tableSink = tableSink;
+ this.isBatch = isBatch;
+ }
+
+ public Optional<TableSource<T1>> getTableSource() {
+ return Optional.ofNullable(tableSource);
+ }
+
+ public Optional<TableSink<T2>> getTableSink() {
+ return Optional.ofNullable(tableSink);
+ }
+
+ public boolean isBatch() {
+ return isBatch;
+ }
+
+ @Override
+ public Map<String, String> toProperties() {
+ // This effectively makes sure the table cannot be persisted in a catalog.
+ throw new UnsupportedOperationException("ConnectorCatalogTable cannot be converted to properties");
+ }
+
+ @Override
+ public CatalogBaseTable copy() {
+ return this;
+ }
+
+ @Override
+ public Optional<String> getDescription() {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<String> getDetailedDescription() {
+ return Optional.empty();
+ }
+
+ private static <T1> TableSchema calculateSourceSchema(TableSource<T1> source, boolean isBatch) {
+ TableSchema tableSchema = source.getTableSchema();
+ if (isBatch) {
+ return tableSchema;
+ }
+
+ TypeInformation[] types = Arrays.copyOf(tableSchema.getFieldTypes(), tableSchema.getFieldCount());
+ String[] fieldNames = tableSchema.getFieldNames();
+ if (source instanceof DefinedRowtimeAttributes) {
+ updateRowtimeIndicators((DefinedRowtimeAttributes) source, fieldNames, types);
+ }
+ if (source instanceof DefinedProctimeAttribute) {
+ updateProctimeIndicator((DefinedProctimeAttribute) source, fieldNames, types);
+ }
+ return new TableSchema(fieldNames, types);
+ }
+
+ private static void updateRowtimeIndicators(
+ DefinedRowtimeAttributes source,
+ String[] fieldNames,
+ TypeInformation[] types) {
+ List<String> rowtimeAttributes = source.getRowtimeAttributeDescriptors()
+ .stream()
+ .map(RowtimeAttributeDescriptor::getAttributeName)
+ .collect(Collectors.toList());
+
+ for (int i = 0; i < fieldNames.length; i++) {
+ if (rowtimeAttributes.contains(fieldNames[i])) {
+ types[i] = TimeIndicatorTypeInfo.ROWTIME_INDICATOR;
+ }
+ }
+ }
+
+ private static void updateProctimeIndicator(
+ DefinedProctimeAttribute source,
+ String[] fieldNames,
+ TypeInformation[] types) {
+ String proctimeAttribute = source.getProctimeAttribute();
+
+ for (int i = 0; i < fieldNames.length; i++) {
+ if (fieldNames[i].equals(proctimeAttribute)) {
+ types[i] = TimeIndicatorTypeInfo.PROCTIME_INDICATOR;
+ break;
+ }
+ }
+ }
+}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/TableOperationCatalogView.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/TableOperationCatalogView.java
index 0e9aafe..89596e8 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/TableOperationCatalogView.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/TableOperationCatalogView.java
@@ -25,7 +25,7 @@ import java.util.HashMap;
import java.util.Optional;
/**
- * A view created from {@link TableOperation} via operations on {@link org.apache.flink.table.api.Table}.
+ * A view created from a {@link TableOperation} via operations on {@link org.apache.flink.table.api.Table}.
*/
@Internal
public class TableOperationCatalogView extends AbstractCatalogView {
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TableOperationDefaultVisitor.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TableOperationDefaultVisitor.java
index 303673e..0773d30 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TableOperationDefaultVisitor.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TableOperationDefaultVisitor.java
@@ -79,6 +79,11 @@ public abstract class TableOperationDefaultVisitor<T> implements TableOperationV
}
@Override
+ public <U> T visitTableSourceTable(TableSourceTableOperation<U> tableSourceTable) {
+ return defaultMethod(tableSourceTable);
+ }
+
+ @Override
public T visitOther(TableOperation other) {
return defaultMethod(other);
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TableOperationVisitor.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TableOperationVisitor.java
index 82c8065..c7c7618 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TableOperationVisitor.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TableOperationVisitor.java
@@ -47,5 +47,7 @@ public interface TableOperationVisitor<T> {
T visitCatalogTable(CatalogTableOperation catalogTable);
+ <U> T visitTableSourceTable(TableSourceTableOperation<U> tableSourceTable);
+
T visitOther(TableOperation other);
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TableSourceTableOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TableSourceTableOperation.java
new file mode 100644
index 0000000..acd12fb
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/TableSourceTableOperation.java
@@ -0,0 +1,75 @@
+/*
+ * 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.operations;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.sources.TableSource;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Inline scan of a {@link TableSource}. Used only when a {@link org.apache.flink.table.api.Table} was created
+ * from {@link org.apache.flink.table.api.TableEnvironment#fromTableSource(TableSource)}.
+ */
+@Internal
+public class TableSourceTableOperation<T> extends TableOperation {
+
+ private final TableSource<T> tableSource;
+ private final boolean isBatch;
+
+ public TableSourceTableOperation(TableSource<T> tableSource, boolean isBatch) {
+ this.tableSource = tableSource;
+ this.isBatch = isBatch;
+ }
+
+ @Override
+ public TableSchema getTableSchema() {
+ return tableSource.getTableSchema();
+ }
+
+ @Override
+ public String asSummaryString() {
+ Map<String, Object> args = new HashMap<>();
+ args.put("fields", tableSource.getTableSchema().getFieldNames());
+
+ return formatWithChildren("TableSource", args);
+ }
+
+ public TableSource<T> getTableSource() {
+ return tableSource;
+ }
+
+ public boolean isBatch() {
+ return isBatch;
+ }
+
+ @Override
+ public List<TableOperation> getChildren() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public <R> R accept(TableOperationVisitor<R> visitor) {
+ return visitor.visitTableSourceTable(this);
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/catalog/CalciteCatalogTable.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/catalog/CalciteCatalogTable.java
deleted file mode 100644
index d6a91bc..0000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/catalog/CalciteCatalogTable.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * 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.annotation.Internal;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.table.api.TableSchema;
-import org.apache.flink.table.calcite.FlinkTypeFactory;
-
-import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.schema.Table;
-
-import java.util.Collections;
-import java.util.Map;
-import java.util.Optional;
-
-/**
- * Thin wrapper around Calcite specific {@link Table}, this is a temporary solution
- * that allows to register those tables in the {@link CatalogManager}.
- * TODO remove once we decouple TableEnvironment from Calcite.
- */
-@Internal
-public class CalciteCatalogTable implements CatalogBaseTable {
- private final Table table;
- private final FlinkTypeFactory typeFactory;
-
- public CalciteCatalogTable(Table table, FlinkTypeFactory typeFactory) {
- this.table = table;
- this.typeFactory = typeFactory;
- }
-
- public Table getTable() {
- return table;
- }
-
- @Override
- public Map<String, String> getProperties() {
- return Collections.emptyMap();
- }
-
- @Override
- public TableSchema getSchema() {
- RelDataType relDataType = table.getRowType(typeFactory);
-
- String[] fieldNames = relDataType.getFieldNames().toArray(new String[0]);
- TypeInformation[] fieldTypes = relDataType.getFieldList()
- .stream()
- .map(field -> FlinkTypeFactory.toTypeInfo(field.getType())).toArray(TypeInformation[]::new);
-
- return new TableSchema(fieldNames, fieldTypes);
- }
-
- @Override
- public String getComment() {
- return null;
- }
-
- @Override
- public CatalogBaseTable copy() {
- return this;
- }
-
- @Override
- public Optional<String> getDescription() {
- return Optional.empty();
- }
-
- @Override
- public Optional<String> getDetailedDescription() {
- return Optional.empty();
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/catalog/DatabaseCalciteSchema.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/catalog/DatabaseCalciteSchema.java
index 7d93b3a..6584068 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/catalog/DatabaseCalciteSchema.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/catalog/DatabaseCalciteSchema.java
@@ -22,6 +22,8 @@ import org.apache.flink.table.api.TableException;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.plan.schema.TableSourceTable;
+import org.apache.flink.table.plan.stats.FlinkStatistic;
import org.apache.calcite.linq4j.tree.Expression;
import org.apache.calcite.rel.type.RelProtoDataType;
@@ -65,10 +67,16 @@ class DatabaseCalciteSchema implements Schema {
CatalogBaseTable table = catalog.getTable(tablePath);
- if (table instanceof CalciteCatalogTable) {
- return ((CalciteCatalogTable) table).getTable();
- } else if (table instanceof TableOperationCatalogView) {
+ if (table instanceof TableOperationCatalogView) {
return TableOperationCatalogViewTable.createCalciteTable(((TableOperationCatalogView) table));
+ } else if (table instanceof ConnectorCatalogTable) {
+ ConnectorCatalogTable<?, ?> connectorTable = (ConnectorCatalogTable<?, ?>) table;
+ return connectorTable.getTableSource()
+ .map(tableSource -> new TableSourceTable<>(
+ tableSource,
+ !connectorTable.isBatch(),
+ FlinkStatistic.UNKNOWN()))
+ .orElseThrow(() -> new TableException("Cannot query a sink only table."));
} else {
throw new TableException("Unsupported table type: " + table);
}
@@ -76,7 +84,7 @@ class DatabaseCalciteSchema implements Schema {
// TableNotExistException should never happen, because we are checking it exists
// via catalog.tableExists
throw new TableException(format(
- "A failure occured when accesing table. Table path [%s, %s, %s]",
+ "A failure occurred when accessing table. Table path [%s, %s, %s]",
catalogName,
databaseName,
tableName), e);
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/plan/TableOperationConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/plan/TableOperationConverter.java
index f3aaac8..01fa8c6 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/plan/TableOperationConverter.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/plan/TableOperationConverter.java
@@ -24,6 +24,7 @@ import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.calcite.FlinkRelBuilder;
import org.apache.flink.table.calcite.FlinkTypeFactory;
+import org.apache.flink.table.catalog.CatalogReader;
import org.apache.flink.table.expressions.AggFunctionCall;
import org.apache.flink.table.expressions.Aggregation;
import org.apache.flink.table.expressions.CallExpression;
@@ -53,6 +54,7 @@ import org.apache.flink.table.operations.SortTableOperation;
import org.apache.flink.table.operations.TableOperation;
import org.apache.flink.table.operations.TableOperationDefaultVisitor;
import org.apache.flink.table.operations.TableOperationVisitor;
+import org.apache.flink.table.operations.TableSourceTableOperation;
import org.apache.flink.table.operations.WindowAggregateTableOperation;
import org.apache.flink.table.operations.WindowAggregateTableOperation.ResolvedGroupWindow;
import org.apache.flink.table.plan.logical.LogicalWindow;
@@ -62,15 +64,21 @@ import org.apache.flink.table.plan.logical.TumblingGroupWindow;
import org.apache.flink.table.plan.nodes.FlinkConventions;
import org.apache.flink.table.plan.nodes.logical.FlinkLogicalDataSetScan;
import org.apache.flink.table.plan.nodes.logical.FlinkLogicalDataStreamScan;
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalTableSourceScan;
import org.apache.flink.table.plan.schema.FlinkTableFunctionImpl;
import org.apache.flink.table.plan.schema.RowSchema;
+import org.apache.flink.table.plan.schema.TableSourceTable;
+import org.apache.flink.table.plan.stats.FlinkStatistic;
+import org.apache.calcite.prepare.RelOptTableImpl;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.CorrelationId;
import org.apache.calcite.rel.core.JoinRelType;
import org.apache.calcite.rel.logical.LogicalTableFunctionScan;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.schema.Schemas;
+import org.apache.calcite.schema.Table;
import org.apache.calcite.tools.RelBuilder.AggCall;
import org.apache.calcite.tools.RelBuilder.GroupKey;
@@ -79,6 +87,7 @@ import java.util.List;
import java.util.Set;
import java.util.stream.IntStream;
+import scala.Option;
import scala.Some;
import static java.util.Arrays.asList;
@@ -264,6 +273,31 @@ public class TableOperationConverter extends TableOperationDefaultVisitor<RelNod
throw new TableException("Unknown table operation: " + other);
}
+ @Override
+ public <U> RelNode visitTableSourceTable(TableSourceTableOperation<U> tableSourceTable) {
+ final Table relTable = new TableSourceTable<>(
+ tableSourceTable.getTableSource(),
+ !tableSourceTable.isBatch(),
+ FlinkStatistic.UNKNOWN());
+
+ CatalogReader catalogReader = (CatalogReader) relBuilder.getRelOptSchema();
+
+ // TableSourceScan requires a unique name of a Table for computing a digest.
+ // We are using the identity hash of the TableSource object.
+ String refId = "unregistered_" + System.identityHashCode(tableSourceTable.getTableSource());
+ return new FlinkLogicalTableSourceScan(
+ relBuilder.getCluster(),
+ relBuilder.getCluster().traitSet().replace(FlinkConventions.LOGICAL()),
+ RelOptTableImpl.create(
+ catalogReader,
+ relTable.getRowType(relBuilder.getTypeFactory()),
+ relTable,
+ Schemas.path(catalogReader.getRootSchema(), Collections.singletonList(refId))),
+ tableSourceTable.getTableSource(),
+ Option.empty()
+ );
+ }
+
private RelNode convertToDataStreamScan(DataStreamTableOperation<?> tableOperation) {
RelDataType logicalRowType = relBuilder.getTypeFactory()
.buildLogicalRowType(tableOperation.getTableSchema());
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/BatchTableEnvImpl.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/BatchTableEnvImpl.scala
index 6dad03a..66a1d83 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/BatchTableEnvImpl.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/BatchTableEnvImpl.scala
@@ -18,8 +18,6 @@
package org.apache.flink.table.api
-import _root_.java.util.concurrent.atomic.AtomicInteger
-
import org.apache.calcite.plan.RelOptUtil
import org.apache.calcite.rel.RelNode
import org.apache.calcite.rel.`type`.RelDataType
@@ -42,7 +40,7 @@ import org.apache.flink.table.plan.rules.FlinkRuleSets
import org.apache.flink.table.plan.schema._
import org.apache.flink.table.runtime.MapRunner
import org.apache.flink.table.sinks._
-import org.apache.flink.table.sources.{BatchTableSource, TableSource}
+import org.apache.flink.table.sources.{BatchTableSource, TableSource, TableSourceUtil}
import org.apache.flink.table.typeutils.FieldInfoUtils.{calculateTableSchema, getFieldsInfo, validateInputTypeInfo}
import org.apache.flink.types.Row
@@ -58,160 +56,31 @@ abstract class BatchTableEnvImpl(
catalogManager: CatalogManager)
extends TableEnvImpl(config, catalogManager) {
- // a counter for unique table names.
- private val nameCntr: AtomicInteger = new AtomicInteger(0)
-
- // the naming pattern for internally registered tables.
- private val internalNamePattern = "^_DataSetTable_[0-9]+$".r
-
override def queryConfig: BatchQueryConfig = new BatchQueryConfig
/**
- * 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 =
- "_DataSetTable_" + nameCntr.getAndIncrement()
-
- /**
* Registers an internal [[BatchTableSource]] in this [[TableEnvImpl]]'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.
*/
- override protected def registerTableSourceInternal(
- name: String,
- tableSource: TableSource[_])
- : Unit = {
-
- tableSource match {
+ override protected def validateTableSource(tableSource: TableSource[_]): Unit = {
+ TableSourceUtil.validateTableSource(tableSource)
- // check for proper batch table source
- case batchTableSource: BatchTableSource[_] =>
- // check if a table (source or sink) is registered
- getTable(defaultCatalogName, defaultDatabaseName, name) match {
-
- // table source and/or sink is registered
- case Some(table: TableSourceSinkTable[_, _]) => table.tableSourceTable match {
-
- // wrapper contains source
- case Some(_: TableSourceTable[_]) =>
- throw new TableException(s"Table '$name' already exists. " +
- s"Please choose a different name.")
-
- // wrapper contains only sink (not source)
- case _ =>
- val enrichedTable = new TableSourceSinkTable(
- Some(new BatchTableSourceTable(batchTableSource)),
- table.tableSinkTable)
- replaceRegisteredTableSourceSinkInternal(name, enrichedTable)
- }
-
- // no table is registered
- case _ =>
- val newTable = new TableSourceSinkTable(
- Some(new BatchTableSourceTable(batchTableSource)),
- None)
- registerTableSourceSinkInternal(name, newTable)
- }
-
- // not a batch table source
- case _ =>
- throw new TableException("Only BatchTableSource can be registered in " +
- "BatchTableEnvironment.")
+ if (!tableSource.isInstanceOf[BatchTableSource[_]]) {
+ throw new TableException("Only BatchTableSource can be registered in " +
+ "BatchTableEnvironment.")
}
}
- def connect(connectorDescriptor: ConnectorDescriptor): BatchTableDescriptor = {
- new BatchTableDescriptor(this, connectorDescriptor)
- }
-
- def registerTableSink(
- name: String,
- fieldNames: Array[String],
- fieldTypes: Array[TypeInformation[_]],
- tableSink: TableSink[_]): Unit = {
- // validate
- checkValidTableName(name)
- if (fieldNames == null) throw new TableException("fieldNames must not be null.")
- if (fieldTypes == null) throw new TableException("fieldTypes must not be null.")
- if (fieldNames.length == 0) throw new TableException("fieldNames must not be empty.")
- if (fieldNames.length != fieldTypes.length) {
- throw new TableException("Same number of field names and types required.")
+ override protected def validateTableSink(configuredSink: TableSink[_]): Unit = {
+ if (!configuredSink.isInstanceOf[BatchTableSink[_]]) {
+ throw new TableException("Only BatchTableSink can be registered in BatchTableEnvironment.")
}
-
- // configure and register
- val configuredSink = tableSink.configure(fieldNames, fieldTypes)
- registerTableSinkInternal(name, configuredSink)
}
- def registerTableSink(name: String, configuredSink: TableSink[_]): Unit = {
- registerTableSinkInternal(name, configuredSink)
- }
-
- private def registerTableSinkInternal(name: String, configuredSink: TableSink[_]): Unit = {
- // validate
- checkValidTableName(name)
- if (configuredSink.getFieldNames == null || configuredSink.getFieldTypes == null) {
- throw new TableException("Table sink is not configured.")
- }
- if (configuredSink.getFieldNames.length == 0) {
- throw new TableException("Field names must not be empty.")
- }
- if (configuredSink.getFieldNames.length != configuredSink.getFieldTypes.length) {
- throw new TableException("Same number of field names and types required.")
- }
-
- // register
- configuredSink match {
-
- // check for proper batch table sink
- case _: BatchTableSink[_] =>
-
- // check if a table (source or sink) is registered
- getTable(name) match {
-
- // table source and/or sink is registered
- case Some(table: TableSourceSinkTable[_, _]) => table.tableSinkTable match {
-
- // wrapper contains sink
- case Some(_: TableSinkTable[_]) =>
- throw new TableException(s"Table '$name' already exists. " +
- s"Please choose a different name.")
-
- // wrapper contains only source (not sink)
- case _ =>
- val enrichedTable = new TableSourceSinkTable(
- table.tableSourceTable,
- Some(new TableSinkTable(configuredSink)))
- replaceRegisteredTableSourceSinkInternal(name, enrichedTable)
- }
-
- // no table is registered
- case _ =>
- val newTable = new TableSourceSinkTable(
- None,
- Some(new TableSinkTable(configuredSink)))
- registerTableSourceSinkInternal(name, newTable)
- }
-
- // not a batch table sink
- case _ =>
- throw new TableException("Only BatchTableSink can be registered in BatchTableEnvironment.")
- }
+ def connect(connectorDescriptor: ConnectorDescriptor): BatchTableDescriptor = {
+ new BatchTableDescriptor(this, connectorDescriptor)
}
/**
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/StreamTableEnvImpl.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/StreamTableEnvImpl.scala
index 5c96e7a..32b3f8c 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/StreamTableEnvImpl.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/StreamTableEnvImpl.scala
@@ -19,7 +19,6 @@
package org.apache.flink.table.api
import _root_.java.lang.{Boolean => JBool}
-import _root_.java.util.concurrent.atomic.AtomicInteger
import org.apache.calcite.plan.RelOptUtil
import org.apache.calcite.plan.hep.HepMatchOrder
@@ -70,45 +69,17 @@ abstract class StreamTableEnvImpl(
catalogManager: CatalogManager)
extends TableEnvImpl(config, catalogManager) {
- // a counter for unique table names
- private val nameCntr: AtomicInteger = new AtomicInteger(0)
-
- // the naming pattern for internally registered tables.
- private val internalNamePattern = "^_DataStreamTable_[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 =
- "_DataStreamTable_" + nameCntr.getAndIncrement()
-
- /**
* Registers an internal [[StreamTableSource]] in this [[TableEnvImpl]]'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.
*/
- override protected def registerTableSourceInternal(
- name: String,
- tableSource: TableSource[_])
- : Unit = {
+ override protected def validateTableSource(tableSource: TableSource[_]): Unit = {
+ TableSourceUtil.validateTableSource(tableSource)
tableSource match {
// check for proper stream table source
@@ -121,33 +92,6 @@ abstract class StreamTableEnvImpl(
s"environment. But is: ${execEnv.getStreamTimeCharacteristic}")
}
- // register
- getTable(defaultCatalogName, defaultDatabaseName, name) match {
-
- // check if a table (source or sink) is registered
- case Some(table: TableSourceSinkTable[_, _]) => table.tableSourceTable match {
-
- // wrapper contains source
- case Some(_: TableSourceTable[_]) =>
- throw new TableException(s"Table '$name' already exists. " +
- s"Please choose a different name.")
-
- // wrapper contains only sink (not source)
- case _ =>
- val enrichedTable = new TableSourceSinkTable(
- Some(new StreamTableSourceTable(streamTableSource)),
- table.tableSinkTable)
- replaceRegisteredTableSourceSinkInternal(name, enrichedTable)
- }
-
- // no table is registered
- case _ =>
- val newTable = new TableSourceSinkTable(
- Some(new StreamTableSourceTable(streamTableSource)),
- None)
- registerTableSourceSinkInternal(name, newTable)
- }
-
// not a stream table source
case _ =>
throw new TableException("Only StreamTableSource can be registered in " +
@@ -155,84 +99,16 @@ abstract class StreamTableEnvImpl(
}
}
- def connect(connectorDescriptor: ConnectorDescriptor): StreamTableDescriptor = {
- new StreamTableDescriptor(this, connectorDescriptor)
- }
-
- def registerTableSink(
- name: String,
- fieldNames: Array[String],
- fieldTypes: Array[TypeInformation[_]],
- tableSink: TableSink[_]): Unit = {
-
- checkValidTableName(name)
- if (fieldNames == null) throw new TableException("fieldNames must not be null.")
- if (fieldTypes == null) throw new TableException("fieldTypes must not be null.")
- if (fieldNames.length == 0) throw new TableException("fieldNames must not be empty.")
- if (fieldNames.length != fieldTypes.length) {
- throw new TableException("Same number of field names and types required.")
+ override protected def validateTableSink(configuredSink: TableSink[_]): Unit = {
+ if (!configuredSink.isInstanceOf[StreamTableSink[_]]) {
+ throw new TableException(
+ "Only AppendStreamTableSink, UpsertStreamTableSink, and RetractStreamTableSink can be " +
+ "registered in StreamTableEnvironment.")
}
-
- val configuredSink = tableSink.configure(fieldNames, fieldTypes)
- registerTableSinkInternal(name, configuredSink)
- }
-
- def registerTableSink(name: String, configuredSink: TableSink[_]): Unit = {
- registerTableSinkInternal(name, configuredSink)
}
- private def registerTableSinkInternal(name: String, configuredSink: TableSink[_]): Unit = {
- // validate
- checkValidTableName(name)
- if (configuredSink.getFieldNames == null || configuredSink.getFieldTypes == null) {
- throw new TableException("Table sink is not configured.")
- }
- if (configuredSink.getFieldNames.length == 0) {
- throw new TableException("Field names must not be empty.")
- }
- if (configuredSink.getFieldNames.length != configuredSink.getFieldTypes.length) {
- throw new TableException("Same number of field names and types required.")
- }
-
- // register
- configuredSink match {
-
- // check for proper batch table sink
- case _: StreamTableSink[_] =>
-
- // check if a table (source or sink) is registered
- getTable(name) match {
-
- // table source and/or sink is registered
- case Some(table: TableSourceSinkTable[_, _]) => table.tableSinkTable match {
-
- // wrapper contains sink
- case Some(_: TableSinkTable[_]) =>
- throw new TableException(s"Table '$name' already exists. " +
- s"Please choose a different name.")
-
- // wrapper contains only source (not sink)
- case _ =>
- val enrichedTable = new TableSourceSinkTable(
- table.tableSourceTable,
- Some(new TableSinkTable(configuredSink)))
- replaceRegisteredTableSourceSinkInternal(name, enrichedTable)
- }
-
- // no table is registered
- case _ =>
- val newTable = new TableSourceSinkTable(
- None,
- Some(new TableSinkTable(configuredSink)))
- registerTableSourceSinkInternal(name, newTable)
- }
-
- // not a stream table sink
- case _ =>
- throw new TableException(
- "Only AppendStreamTableSink, UpsertStreamTableSink, and RetractStreamTableSink can be " +
- "registered in StreamTableEnvironment.")
- }
+ def connect(connectorDescriptor: ConnectorDescriptor): StreamTableDescriptor = {
+ new StreamTableDescriptor(this, connectorDescriptor)
}
/**
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/TableEnvImpl.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/TableEnvImpl.scala
index 6db31ee..f281829 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/TableEnvImpl.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/TableEnvImpl.scala
@@ -22,13 +22,11 @@ import _root_.java.util.Optional
import _root_.java.util.concurrent.atomic.AtomicInteger
import com.google.common.collect.ImmutableList
-import org.apache.calcite.jdbc.CalciteSchema
import org.apache.calcite.jdbc.CalciteSchemaBuilder.asRootSchema
import org.apache.calcite.plan.RelOptPlanner.CannotPlanException
import org.apache.calcite.plan._
import org.apache.calcite.plan.hep.{HepMatchOrder, HepPlanner, HepProgram, HepProgramBuilder}
import org.apache.calcite.rel.RelNode
-import org.apache.calcite.schema.SchemaPlus
import org.apache.calcite.sql._
import org.apache.calcite.sql.parser.SqlParser
import org.apache.calcite.tools._
@@ -40,12 +38,13 @@ import org.apache.flink.table.calcite._
import org.apache.flink.table.catalog._
import org.apache.flink.table.codegen.{FunctionCodeGenerator, GeneratedFunction}
import org.apache.flink.table.expressions._
+import org.apache.flink.table.factories.{TableFactoryService, TableFactoryUtil, TableSinkFactory}
import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils._
import org.apache.flink.table.functions.{AggregateFunction, ScalarFunction, TableFunction, UserDefinedAggregateFunction}
-import org.apache.flink.table.operations.{CatalogTableOperation, OperationTreeBuilder, PlannerTableOperation}
+import org.apache.flink.table.operations.{CatalogTableOperation, OperationTreeBuilder, PlannerTableOperation, TableSourceTableOperation}
import org.apache.flink.table.plan.nodes.FlinkConventions
import org.apache.flink.table.plan.rules.FlinkRuleSets
-import org.apache.flink.table.plan.schema.{RowSchema, TableSourceSinkTable}
+import org.apache.flink.table.plan.schema.RowSchema
import org.apache.flink.table.planner.PlanningConfigurationBuilder
import org.apache.flink.table.sinks.TableSink
import org.apache.flink.table.sources.TableSource
@@ -53,9 +52,9 @@ import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo
import org.apache.flink.table.util.JavaScalaConversionUtil
import org.apache.flink.table.validate.FunctionCatalog
import org.apache.flink.types.Row
+import org.apache.flink.util.StringUtils
import _root_.scala.collection.JavaConverters._
-import _root_.scala.collection.mutable
/**
* The abstract base class for the implementation of batch and stream TableEnvironments.
@@ -73,9 +72,6 @@ abstract class TableEnvImpl(
protected val defaultCatalogName: String = config.getBuiltInCatalogName
protected val defaultDatabaseName: String = config.getBuiltInDatabaseName
- private val internalSchema: CalciteSchema =
- asRootSchema(new CatalogManagerCalciteSchema(catalogManager, isBatch))
-
// temporary bridge between API and planner
private[flink] val expressionBridge: ExpressionBridge[PlannerExpression] =
new ExpressionBridge[PlannerExpression](functionCatalog, PlannerExpressionConverter.INSTANCE)
@@ -89,7 +85,7 @@ abstract class TableEnvImpl(
new PlanningConfigurationBuilder(
config,
functionCatalog,
- internalSchema,
+ asRootSchema(new CatalogManagerCalciteSchema(catalogManager, isBatch)),
expressionBridge)
protected def calciteConfig: CalciteConfig = config.getPlannerConfig
@@ -425,23 +421,138 @@ abstract class TableEnvImpl(
"Only tables that belong to this TableEnvironment can be registered.")
}
- checkValidTableName(name)
-
val tableTable = new TableOperationCatalogView(table.getTableOperation)
registerTableInternal(name, tableTable)
}
override def registerTableSource(name: String, tableSource: TableSource[_]): Unit = {
+ validateTableSource(tableSource)
registerTableSourceInternal(name, tableSource)
}
+ override def registerTableSink(
+ name: String,
+ fieldNames: Array[String],
+ fieldTypes: Array[TypeInformation[_]],
+ tableSink: TableSink[_]): Unit = {
+
+ if (fieldNames == null) {
+ throw new TableException("fieldNames must not be null.")
+ }
+ if (fieldTypes == null) {
+ throw new TableException("fieldTypes must not be null.")
+ }
+ if (fieldNames.length == 0) {
+ throw new TableException("fieldNames must not be empty.")
+ }
+ if (fieldNames.length != fieldTypes.length) {
+ throw new TableException("Same number of field names and types required.")
+ }
+
+ val configuredSink = tableSink.configure(fieldNames, fieldTypes)
+ registerTableSinkInternal(name, configuredSink)
+ }
+
+ override def registerTableSink(name: String, configuredSink: TableSink[_]): Unit = {
+ // validate
+ if (configuredSink.getFieldNames == null || configuredSink.getFieldTypes == null) {
+ throw new TableException("Table sink is not configured.")
+ }
+ if (configuredSink.getFieldNames.length == 0) {
+ throw new TableException("Field names must not be empty.")
+ }
+ if (configuredSink.getFieldNames.length != configuredSink.getFieldTypes.length) {
+ throw new TableException("Same number of field names and types required.")
+ }
+
+ validateTableSink(configuredSink)
+ registerTableSinkInternal(name, configuredSink)
+ }
+
override def fromTableSource(source: TableSource[_]): Table = {
- val name = createUniqueTableName()
- registerTableSourceInternal(name, source)
- scan(name)
+ new TableImpl(this, new TableSourceTableOperation(source, isBatch))
+ }
+
+ /**
+ * Perform batch or streaming specific validations of the [[TableSource]].
+ * This method should throw [[ValidationException]] if the [[TableSource]] cannot be used
+ * in this [[TableEnvironment]].
+ *
+ * @param tableSource table source to validate
+ */
+ protected def validateTableSource(tableSource: TableSource[_]): Unit
+
+ /**
+ * Perform batch or streaming specific validations of the [[TableSink]].
+ * This method should throw [[ValidationException]] if the [[TableSink]] cannot be used
+ * in this [[TableEnvironment]].
+ *
+ * @param tableSink table source to validate
+ */
+ protected def validateTableSink(tableSink: TableSink[_]): Unit
+
+ private def registerTableSourceInternal(
+ name: String,
+ tableSource: TableSource[_])
+ : Unit = {
+ // register
+ getCatalogTable(defaultCatalogName, defaultDatabaseName, name) match {
+
+ // check if a table (source or sink) is registered
+ case Some(table: ConnectorCatalogTable[_, _]) =>
+ if (table.getTableSource.isPresent) {
+ // wrapper contains source
+ throw new TableException(s"Table '$name' already exists. " +
+ s"Please choose a different name.")
+ } else {
+ // wrapper contains only sink (not source)
+ replaceTableInternal(
+ name,
+ ConnectorCatalogTable
+ .sourceAndSink(tableSource, table.getTableSink.get, isBatch))
+ }
+
+ // no table is registered
+ case _ =>
+ registerTableInternal(name, ConnectorCatalogTable.source(tableSource, isBatch))
+ }
+ }
+
+ private def registerTableSinkInternal(
+ name: String,
+ tableSink: TableSink[_])
+ : Unit = {
+ // check if a table (source or sink) is registered
+ getCatalogTable(defaultCatalogName, defaultDatabaseName, name) match {
+
+ // table source and/or sink is registered
+ case Some(table: ConnectorCatalogTable[_, _]) =>
+ if (table.getTableSink.isPresent) {
+ // wrapper contains sink
+ throw new TableException(s"Table '$name' already exists. " +
+ s"Please choose a different name.")
+ } else {
+ // wrapper contains only source (not sink)
+ replaceTableInternal(
+ name,
+ ConnectorCatalogTable
+ .sourceAndSink(table.getTableSource.get, tableSink, isBatch))
+ }
+
+ // no table is registered
+ case _ =>
+ registerTableInternal(name, ConnectorCatalogTable.sink(tableSink, isBatch))
+ }
+ }
+
+ private def checkValidTableName(name: String) = {
+ if (StringUtils.isNullOrWhitespaceOnly(name)) {
+ throw new ValidationException("A table name cannot be null or consist of only whitespaces.")
+ }
}
protected def registerTableInternal(name: String, table: CatalogBaseTable): Unit = {
+ checkValidTableName(name)
val path = new ObjectPath(defaultDatabaseName, name)
JavaScalaConversionUtil.toScala(catalogManager.getCatalog(defaultCatalogName)) match {
case Some(catalog) =>
@@ -453,27 +564,14 @@ abstract class TableEnvImpl(
}
}
- protected def registerTableSourceInternal(name: String, tableSource: TableSource[_]): Unit
-
- protected def registerTableSourceSinkInternal[T1, T2](
- name: String,
- table: TableSourceSinkTable[T1, T2])
- : Unit = {
- registerTableInternal(
- name,
- new CalciteCatalogTable(table, planningConfigurationBuilder.getTypeFactory))
- }
-
- protected def replaceRegisteredTableSourceSinkInternal[T1, T2](
- name: String,
- table: TableSourceSinkTable[T1, T2])
- : Unit = {
+ protected def replaceTableInternal(name: String, table: CatalogBaseTable): Unit = {
+ checkValidTableName(name)
val path = new ObjectPath(defaultDatabaseName, name)
JavaScalaConversionUtil.toScala(catalogManager.getCatalog(defaultCatalogName)) match {
case Some(catalog) =>
catalog.alterTable(
path,
- new CalciteCatalogTable(table, planningConfigurationBuilder.getTypeFactory),
+ table,
false)
case None => throw new TableException("The default catalog does not exist.")
}
@@ -488,7 +586,8 @@ abstract class TableEnvImpl(
}
private[flink] def scanInternal(tablePath: Array[String]): Option[CatalogTableOperation] = {
- JavaScalaConversionUtil.toScala(catalogManager.resolveTable(tablePath : _*))
+ JavaScalaConversionUtil.toScala(catalogManager.resolveTable(tablePath: _*))
+ .map(t => new CatalogTableOperation(t.getTablePath, t.getTableSchema))
}
override def listTables(): Array[String] = {
@@ -578,16 +677,19 @@ abstract class TableEnvImpl(
private[flink] def insertInto(table: Table, conf: QueryConfig, sinkTablePath: String*): Unit = {
// check that sink table exists
- if (null == sinkTablePath) throw new TableException("Name of TableSink must not be null.")
- if (sinkTablePath.isEmpty) throw new TableException("Name of TableSink must not be empty.")
+ if (null == sinkTablePath) {
+ throw new TableException("Name of TableSink must not be null.")
+ }
+ if (sinkTablePath.isEmpty) {
+ throw new TableException("Name of TableSink must not be empty.")
+ }
- getTable(sinkTablePath: _*) match {
+ getTableSink(sinkTablePath: _*) match {
case None =>
throw new TableException(s"No table was registered under the name $sinkTablePath.")
- case Some(s: TableSourceSinkTable[_, _]) if s.tableSinkTable.isDefined =>
- val tableSink = s.tableSinkTable.get.tableSink
+ case Some(tableSink) =>
// validate schema of source table and table sink
val srcFieldTypes = table.getSchema.getFieldTypes
val sinkFieldTypes = tableSink.getFieldTypes
@@ -614,58 +716,35 @@ abstract class TableEnvImpl(
}
// emit the table to the configured table sink
writeToSink(table, tableSink, conf)
-
- case Some(_) =>
- throw new TableException(s"The table registered as $sinkTablePath is not a TableSink. " +
- s"You can only emit query results to a registered TableSink.")
}
}
- /** Returns a unique table name according to the internal naming pattern. */
- protected def createUniqueTableName(): String
+ private def getTableSink(name: String*): Option[TableSink[_]] = {
+ JavaScalaConversionUtil.toScala(catalogManager.resolveTable(name: _*)) match {
+ case Some(s) if s.getExternalCatalogTable.isPresent =>
- /**
- * Checks if the chosen table name is valid.
- *
- * @param name The table name to check.
- */
- protected def checkValidTableName(name: String): Unit
+ Option(TableFactoryUtil.findAndCreateTableSink(s.getExternalCatalogTable.get()))
- /**
- * 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
- }
- }
+ case Some(s) if JavaScalaConversionUtil.toScala(s.getCatalogTable)
+ .exists(_.isInstanceOf[ConnectorCatalogTable[_, _]]) =>
+
+ JavaScalaConversionUtil
+ .toScala(s.getCatalogTable.get().asInstanceOf[ConnectorCatalogTable[_, _]].getTableSink)
+
+ case Some(s) if JavaScalaConversionUtil.toScala(s.getCatalogTable)
+ .exists(_.isInstanceOf[CatalogTable]) =>
+
+ val sinkProperties = s.getCatalogTable.get().asInstanceOf[CatalogTable].toProperties
+ Option(TableFactoryService.find(classOf[TableSinkFactory[_]], sinkProperties)
+ .createTableSink(sinkProperties))
+
+ case _ => None
}
+ }
+ protected def getCatalogTable(name: String*): Option[CatalogBaseTable] = {
JavaScalaConversionUtil.toScala(catalogManager.resolveTable(name: _*))
- .flatMap(t =>
- getTableFromSchema(internalSchema.plus(), t.getTablePath.asScala.toList)
- )
+ .flatMap(t => JavaScalaConversionUtil.toScala(t.getCatalogTable))
}
/** Returns a unique temporary attribute name. */
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/calcite/FlinkRelBuilder.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/calcite/FlinkRelBuilder.scala
index 54962f5..eae99c1 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/calcite/FlinkRelBuilder.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/calcite/FlinkRelBuilder.scala
@@ -19,16 +19,12 @@
package org.apache.flink.table.calcite
import java.lang.Iterable
-import java.util.{Collections, List => JList}
+import java.util.{List => JList}
-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.rel.logical.LogicalAggregate
-import org.apache.calcite.rex.RexBuilder
+import org.apache.calcite.tools.RelBuilder
import org.apache.calcite.tools.RelBuilder.{AggCall, GroupKey}
-import org.apache.calcite.tools.{FrameworkConfig, RelBuilder}
import org.apache.flink.table.api.TableException
import org.apache.flink.table.expressions.{Alias, ExpressionBridge, PlannerExpression, WindowProperty}
import org.apache.flink.table.operations.TableOperation
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogSchema.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogSchema.scala
index 5017727..4b2287f 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogSchema.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/catalog/ExternalCatalogSchema.scala
@@ -81,7 +81,7 @@ class ExternalCatalogSchema(
*/
override def getTable(name: String): Table = try {
val externalCatalogTable = catalog.getTable(name)
- ExternalTableUtil.fromExternalCatalogTable(isBatch, externalCatalogTable)
+ ExternalTableUtil.fromExternalCatalogTable(isBatch, externalCatalogTable).orNull
} catch {
case _: TableNotExistException => {
LOG.warn(s"Table $name does not exist in externalCatalog $catalogIdentifier")
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/catalog/ExternalTableUtil.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/catalog/ExternalTableUtil.scala
index f91f309..4ac24dd 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/catalog/ExternalTableUtil.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/catalog/ExternalTableUtil.scala
@@ -22,13 +22,12 @@ import org.apache.flink.table.api._
import org.apache.flink.table.factories._
import org.apache.flink.table.plan.schema._
import org.apache.flink.table.plan.stats.FlinkStatistic
-import org.apache.flink.table.sinks.{BatchTableSink, StreamTableSink}
-import org.apache.flink.table.sources.{BatchTableSource, StreamTableSource}
+import org.apache.flink.table.sources.TableSource
import org.apache.flink.table.util.JavaScalaConversionUtil.toScala
import org.apache.flink.table.util.Logging
/**
- * The utility class is used to convert [[ExternalCatalogTable]] to [[TableSourceSinkTable]].
+ * The utility class is used to convert [[ExternalCatalogTable]].
*
* It uses [[TableFactoryService]] for discovering.
*/
@@ -40,24 +39,16 @@ object ExternalTableUtil extends Logging {
* @param externalTable the [[ExternalCatalogTable]] instance which to convert
* @return converted [[TableSourceTable]] instance from the input catalog table
*/
- def fromExternalCatalogTable[T1, T2](isBatch: Boolean, externalTable: ExternalCatalogTable)
- : TableSourceSinkTable[T1, T2] = {
+ def fromExternalCatalogTable[T](isBatch: Boolean, externalTable: ExternalCatalogTable)
+ : Option[TableSourceTable[T]] = {
val statistics = new FlinkStatistic(toScala(externalTable.getTableStats))
- val source: Option[TableSourceTable[T1]] = if (externalTable.isTableSource) {
+ if (externalTable.isTableSource) {
Some(createTableSource(isBatch, externalTable, statistics))
} else {
None
}
-
- val sink: Option[TableSinkTable[T2]] = if (externalTable.isTableSink) {
- Some(createTableSink(isBatch, externalTable, statistics))
- } else {
- None
- }
-
- new TableSourceSinkTable[T1, T2](source, sink)
}
private def createTableSource[T](
@@ -65,33 +56,20 @@ object ExternalTableUtil extends Logging {
externalTable: ExternalCatalogTable,
statistics: FlinkStatistic)
: TableSourceTable[T] = {
- if (isBatch && externalTable.isBatchTable) {
- val source = TableFactoryUtil.findAndCreateTableSource(externalTable)
- new BatchTableSourceTable[T](source.asInstanceOf[BatchTableSource[T]], statistics)
- } else if (!isBatch && externalTable.isStreamTable) {
- val source = TableFactoryUtil.findAndCreateTableSource(externalTable)
- new StreamTableSourceTable[T](source.asInstanceOf[StreamTableSource[T]], statistics)
+ val source = if (isModeCompatibleWithTable(isBatch, externalTable)) {
+ TableFactoryUtil.findAndCreateTableSource(externalTable)
} else {
throw new ValidationException(
"External catalog table does not support the current environment for a table source.")
}
+
+ new TableSourceTable[T](source.asInstanceOf[TableSource[T]], !isBatch, statistics)
}
- private def createTableSink[T](
+ private def isModeCompatibleWithTable[T](
isBatch: Boolean,
- externalTable: ExternalCatalogTable,
- statistics: FlinkStatistic)
- : TableSinkTable[T] = {
-
- if (isBatch && externalTable.isBatchTable) {
- val sink = TableFactoryUtil.findAndCreateTableSink(externalTable)
- new TableSinkTable[T](sink.asInstanceOf[BatchTableSink[T]], statistics)
- } else if (!isBatch && externalTable.isStreamTable) {
- val sink = TableFactoryUtil.findAndCreateTableSink(externalTable)
- new TableSinkTable[T](sink.asInstanceOf[StreamTableSink[T]], statistics)
- } else {
- throw new ValidationException(
- "External catalog table does not support the current environment for a table sink.")
- }
+ externalTable: ExternalCatalogTable)
+ : Boolean = {
+ isBatch && externalTable.isBatchTable || !isBatch && externalTable.isStreamTable
}
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/PhysicalTableSourceScan.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/PhysicalTableSourceScan.scala
index 03b442b..4aad828 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/PhysicalTableSourceScan.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/PhysicalTableSourceScan.scala
@@ -23,8 +23,7 @@ import org.apache.calcite.rel.RelWriter
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.core.TableScan
import org.apache.flink.table.calcite.FlinkTypeFactory
-import org.apache.flink.table.plan.schema.TableSourceSinkTable
-import org.apache.flink.table.sources.{TableSource, TableSourceUtil}
+import org.apache.flink.table.sources.{StreamTableSource, TableSource, TableSourceUtil}
import scala.collection.JavaConverters._
@@ -32,19 +31,16 @@ abstract class PhysicalTableSourceScan(
cluster: RelOptCluster,
traitSet: RelTraitSet,
table: RelOptTable,
- val tableSource: TableSource[_],
+ tableSource: TableSource[_],
val selectedFields: Option[Array[Int]])
extends TableScan(cluster, traitSet, table) {
override def deriveRowType(): RelDataType = {
val flinkTypeFactory = cluster.getTypeFactory.asInstanceOf[FlinkTypeFactory]
- val streamingTable = table.unwrap(classOf[TableSourceSinkTable[_, _]]) match {
- case t: TableSourceSinkTable[_, _] if t.isStreamSourceTable => true
- // null
- case _ => false
- }
+ val streamingTable = tableSource.isInstanceOf[StreamTableSource[_]]
- TableSourceUtil.getRelDataType(tableSource, selectedFields, streamingTable, flinkTypeFactory)
+ TableSourceUtil
+ .getRelDataType(tableSource, selectedFields, streamingTable, flinkTypeFactory)
}
override def explainTerms(pw: RelWriter): RelWriter = {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/dataset/BatchTableSourceScan.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/dataset/BatchTableSourceScan.scala
index 0b91ab7..862523b 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/dataset/BatchTableSourceScan.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/dataset/BatchTableSourceScan.scala
@@ -73,7 +73,7 @@ class BatchTableSourceScan(
cluster,
traitSet,
getTable,
- newTableSource.asInstanceOf[BatchTableSource[_]],
+ tableSource,
selectedFields
)
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamTableSourceScan.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamTableSourceScan.scala
index 7e45109..20e2234 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamTableSourceScan.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamTableSourceScan.scala
@@ -32,7 +32,7 @@ import org.apache.flink.table.plan.nodes.PhysicalTableSourceScan
import org.apache.flink.table.plan.schema.RowSchema
import org.apache.flink.table.runtime.types.CRow
import org.apache.flink.table.sources._
-import org.apache.flink.table.sources.wmstrategies.{PeriodicWatermarkAssigner, PunctuatedWatermarkAssigner, PreserveWatermarks}
+import org.apache.flink.table.sources.wmstrategies.{PeriodicWatermarkAssigner, PreserveWatermarks, PunctuatedWatermarkAssigner}
import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo
/** Flink RelNode to read data from an external source defined by a [[StreamTableSource]]. */
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/logical/FlinkLogicalTableSourceScan.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/logical/FlinkLogicalTableSourceScan.scala
index 0c3d25a..89aff89 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/logical/FlinkLogicalTableSourceScan.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/logical/FlinkLogicalTableSourceScan.scala
@@ -25,10 +25,9 @@ import org.apache.calcite.rel.core.TableScan
import org.apache.calcite.rel.logical.LogicalTableScan
import org.apache.calcite.rel.metadata.RelMetadataQuery
import org.apache.calcite.rel.{RelNode, RelWriter}
-import org.apache.flink.table.api.TableException
import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.plan.nodes.FlinkConventions
-import org.apache.flink.table.plan.schema.TableSourceSinkTable
+import org.apache.flink.table.plan.schema.TableSourceTable
import org.apache.flink.table.sources.{FilterableTableSource, TableSource, TableSourceUtil}
import scala.collection.JavaConverters._
@@ -51,11 +50,7 @@ class FlinkLogicalTableSourceScan(
override def deriveRowType(): RelDataType = {
val flinkTypeFactory = cluster.getTypeFactory.asInstanceOf[FlinkTypeFactory]
- val streamingTable = table.unwrap(classOf[TableSourceSinkTable[_, _]]) match {
- case t: TableSourceSinkTable[_, _] if t.isStreamSourceTable => true
- // null
- case _ => false
- }
+ val streamingTable = table.unwrap(classOf[TableSourceTable[_]]).isStreaming
TableSourceUtil.getRelDataType(tableSource, selectedFields, streamingTable, flinkTypeFactory)
}
@@ -111,26 +106,18 @@ class FlinkLogicalTableSourceScanConverter
override def matches(call: RelOptRuleCall): Boolean = {
val scan = call.rel[TableScan](0)
- scan.getTable.unwrap(classOf[TableSourceSinkTable[_, _]]) match {
- case t: TableSourceSinkTable[_, _] if t.isSourceTable => true
- // null
- case _ => false
- }
+ scan.getTable.unwrap(classOf[TableSourceTable[_]]) != null
}
def convert(rel: RelNode): RelNode = {
val scan = rel.asInstanceOf[TableScan]
val traitSet = rel.getTraitSet.replace(FlinkConventions.LOGICAL)
- val tableSource = scan.getTable.unwrap(classOf[TableSourceSinkTable[_, _]])
- .tableSourceTable
- .map(_.tableSource)
- .getOrElse(throw new TableException("Table source expected."))
new FlinkLogicalTableSourceScan(
rel.getCluster,
traitSet,
scan.getTable,
- tableSource,
+ scan.getTable.unwrap(classOf[TableSourceTable[_]]).tableSource,
None
)
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/dataSet/BatchTableSourceScanRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/dataSet/BatchTableSourceScanRule.scala
index cee6eef..8ce97ed 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/dataSet/BatchTableSourceScanRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/dataSet/BatchTableSourceScanRule.scala
@@ -25,7 +25,7 @@ import org.apache.calcite.rel.core.TableScan
import org.apache.flink.table.plan.nodes.FlinkConventions
import org.apache.flink.table.plan.nodes.dataset.BatchTableSourceScan
import org.apache.flink.table.plan.nodes.logical.FlinkLogicalTableSourceScan
-import org.apache.flink.table.plan.schema.TableSourceSinkTable
+import org.apache.flink.table.plan.schema.TableSourceTable
import org.apache.flink.table.sources.BatchTableSource
class BatchTableSourceScanRule
@@ -38,11 +38,9 @@ class BatchTableSourceScanRule
/** Rule must only match if TableScan targets a [[BatchTableSource]] */
override def matches(call: RelOptRuleCall): Boolean = {
val scan: TableScan = call.rel(0).asInstanceOf[TableScan]
- scan.getTable.unwrap(classOf[TableSourceSinkTable[_, _]]) match {
- case t: TableSourceSinkTable[_, _] if t.isBatchSourceTable => true
- // null
- case _ => false
- }
+
+ val sourceTable = scan.getTable.unwrap(classOf[TableSourceTable[_]])
+ sourceTable != null && !sourceTable.isStreaming
}
def convert(rel: RelNode): RelNode = {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/datastream/StreamTableSourceScanRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/datastream/StreamTableSourceScanRule.scala
index e99118f..a23830c 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/datastream/StreamTableSourceScanRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/datastream/StreamTableSourceScanRule.scala
@@ -25,7 +25,7 @@ import org.apache.calcite.rel.core.TableScan
import org.apache.flink.table.plan.nodes.FlinkConventions
import org.apache.flink.table.plan.nodes.datastream.StreamTableSourceScan
import org.apache.flink.table.plan.nodes.logical.FlinkLogicalTableSourceScan
-import org.apache.flink.table.plan.schema.TableSourceSinkTable
+import org.apache.flink.table.plan.schema.TableSourceTable
import org.apache.flink.table.sources.StreamTableSource
class StreamTableSourceScanRule
@@ -39,11 +39,9 @@ class StreamTableSourceScanRule
/** Rule must only match if TableScan targets a [[StreamTableSource]] */
override def matches(call: RelOptRuleCall): Boolean = {
val scan: TableScan = call.rel(0).asInstanceOf[TableScan]
- scan.getTable.unwrap(classOf[TableSourceSinkTable[_, _]]) match {
- case t: TableSourceSinkTable[_, _] if t.isStreamSourceTable => true
- // null
- case _ => false
- }
+
+ val sourceTable = scan.getTable.unwrap(classOf[TableSourceTable[_]])
+ sourceTable != null && sourceTable.isStreaming
}
def convert(rel: RelNode): RelNode = {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/TableSinkTable.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/TableSinkTable.scala
deleted file mode 100644
index 75ce3da..0000000
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/TableSinkTable.scala
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * 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.rel.`type`.{RelDataType, RelDataTypeFactory}
-import org.apache.calcite.schema.Statistic
-import org.apache.calcite.schema.impl.AbstractTable
-import org.apache.flink.table.calcite.FlinkTypeFactory
-import org.apache.flink.table.plan.stats.FlinkStatistic
-import org.apache.flink.table.sinks.TableSink
-
-/** Class which implements the logic to convert a [[TableSink]] to Calcite Table */
-class TableSinkTable[T](
- val tableSink: TableSink[T],
- val statistic: FlinkStatistic = FlinkStatistic.UNKNOWN) {
-
- /** Returns the row type of the table with this tableSink.
- *
- * @param typeFactory Type factory with which to create the type
- * @return Row type
- */
- def getRowType(typeFactory: RelDataTypeFactory): RelDataType = {
- val flinkTypeFactory = typeFactory.asInstanceOf[FlinkTypeFactory]
- flinkTypeFactory.buildLogicalRowType(tableSink.getFieldNames, tableSink.getFieldTypes)
- }
-
- /**
- * Returns statistics of current table
- *
- * @return statistics of current table
- */
- def getStatistic: Statistic = statistic
-}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/TableSourceSinkTable.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/TableSourceSinkTable.scala
deleted file mode 100644
index 923d82c..0000000
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/TableSourceSinkTable.scala
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * 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.rel.`type`.{RelDataType, RelDataTypeFactory}
-import org.apache.calcite.schema.Statistic
-import org.apache.calcite.schema.impl.AbstractTable
-import org.apache.flink.table.api.TableException
-
-/**
- * Wrapper for both a [[TableSourceTable]] and [[TableSinkTable]] under a common name.
- *
- * @param tableSourceTable table source table (if available)
- * @param tableSinkTable table sink table (if available)
- * @tparam T1 type of the table source table
- * @tparam T2 type of the table sink table
- */
-class TableSourceSinkTable[T1, T2](
- val tableSourceTable: Option[TableSourceTable[T1]],
- val tableSinkTable: Option[TableSinkTable[T2]])
- extends AbstractTable {
-
- // In the streaming case, the table schema of source and sink can differ because of extra
- // rowtime/proctime fields. We will always return the source table schema if tableSourceTable
- // is not None, otherwise return the sink table schema. We move the Calcite validation logic of
- // the sink table schema into Flink. This allows us to have different schemas as source and sink
- // of the same table.
- override def getRowType(typeFactory: RelDataTypeFactory): RelDataType = {
- tableSourceTable.map(_.getRowType(typeFactory))
- .orElse(tableSinkTable.map(_.getRowType(typeFactory)))
- .getOrElse(throw new TableException("Unable to get row type of table source sink table."))
- }
-
- override def getStatistic: Statistic = {
- tableSourceTable.map(_.getStatistic)
- .orElse(tableSinkTable.map(_.getStatistic))
- .getOrElse(throw new TableException("Unable to get statistics of table source sink table."))
- }
-
- def isSourceTable: Boolean = tableSourceTable.isDefined
-
- def isStreamSourceTable: Boolean = tableSourceTable match {
- case Some(_: StreamTableSourceTable[_]) => true
- case _ => false
- }
-
- def isBatchSourceTable: Boolean = tableSourceTable match {
- case Some(_: BatchTableSourceTable[_]) => true
- case _ => false
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/TableSourceTable.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/TableSourceTable.scala
index 26224b6..9dab249 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/TableSourceTable.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/TableSourceTable.scala
@@ -20,27 +20,35 @@ package org.apache.flink.table.plan.schema
import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory}
import org.apache.calcite.schema.Statistic
+import org.apache.calcite.schema.impl.AbstractTable
+import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.plan.stats.FlinkStatistic
-import org.apache.flink.table.sources.TableSource
+import org.apache.flink.table.sources.{TableSource, TableSourceUtil}
-/** Abstract class which define the interfaces required to convert a [[TableSource]] to
- * a Calcite Table */
-abstract class TableSourceTable[T](
+/**
+ * Abstract class which define the interfaces required to convert a [[TableSource]] to
+ * a Calcite Table.
+ */
+class TableSourceTable[T](
val tableSource: TableSource[T],
- val statistic: FlinkStatistic) {
+ val isStreaming: Boolean,
+ val statistic: FlinkStatistic)
+ extends AbstractTable {
- /** Returns the row type of the table with this tableSource.
- *
- * @param typeFactory Type factory with which to create the type
- * @return Row type
- */
- def getRowType(typeFactory: RelDataTypeFactory): RelDataType
+ TableSourceUtil.validateTableSource(tableSource)
/**
* Returns statistics of current table
*
* @return statistics of current table
*/
- def getStatistic: Statistic = statistic
+ override def getStatistic: Statistic = statistic
+ def getRowType(typeFactory: RelDataTypeFactory): RelDataType = {
+ TableSourceUtil.getRelDataType(
+ tableSource,
+ None,
+ isStreaming,
+ typeFactory.asInstanceOf[FlinkTypeFactory])
+ }
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/catalog/CatalogStructureBuilder.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/catalog/CatalogStructureBuilder.java
index eb247a4..8787e94 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/catalog/CatalogStructureBuilder.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/catalog/CatalogStructureBuilder.java
@@ -22,20 +22,12 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableSchema;
-import org.apache.flink.table.calcite.FlinkTypeFactory;
-import org.apache.flink.table.calcite.FlinkTypeSystem;
-import org.apache.flink.table.plan.schema.StreamTableSourceTable;
-import org.apache.flink.table.plan.schema.TableSourceSinkTable;
-import org.apache.flink.table.plan.stats.FlinkStatistic;
import org.apache.flink.table.sources.StreamTableSource;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
-import scala.Option;
-import scala.Some;
-
import static org.apache.flink.table.descriptors.ConnectorDescriptorValidator.CONNECTOR_TYPE;
/**
@@ -273,33 +265,28 @@ public class CatalogStructureBuilder {
}
}
- private static class TestTable extends CalciteCatalogTable {
-
+ private static class TestTable extends ConnectorCatalogTable<Object, Object> {
private final String fullyQualifiedPath;
- private static final StreamTableSourceTable<Object> tableSourceTable = new StreamTableSourceTable<>(
- new StreamTableSource<Object>() {
- @Override
- public DataStream<Object> getDataStream(StreamExecutionEnvironment execEnv) {
- return null;
- }
+ private static final StreamTableSource<Object> tableSource = new StreamTableSource<Object>() {
+ @Override
+ public DataStream<Object> getDataStream(StreamExecutionEnvironment execEnv) {
+ return null;
+ }
- @Override
- public TypeInformation<Object> getReturnType() {
- return null;
- }
+ @Override
+ public TypeInformation<Object> getReturnType() {
+ return null;
+ }
- @Override
- public TableSchema getTableSchema() {
- return new TableSchema(new String[] {}, new TypeInformation[] {});
- }
- }, FlinkStatistic.UNKNOWN());
+ @Override
+ public TableSchema getTableSchema() {
+ return TableSchema.builder().build();
+ }
+ };
private TestTable(String fullyQualifiedPath) {
- super(new TableSourceSinkTable<>(
- new Some<>(tableSourceTable),
- Option.empty()
- ), new FlinkTypeFactory(new FlinkTypeSystem()));
+ super(tableSource, null, tableSource.getTableSchema(), false);
this.fullyQualifiedPath = fullyQualifiedPath;
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/catalog/PathResolutionTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/catalog/PathResolutionTest.java
index c9fc524..f68aa50 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/catalog/PathResolutionTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/catalog/PathResolutionTest.java
@@ -19,7 +19,6 @@
package org.apache.flink.table.catalog;
import org.apache.flink.table.api.java.StreamTableEnvImpl;
-import org.apache.flink.table.operations.CatalogTableOperation;
import org.apache.flink.table.utils.StreamTableTestUtil;
import org.apache.flink.util.Preconditions;
@@ -202,7 +201,7 @@ public class PathResolutionTest {
testSpec.getDefaultCatalog().ifPresent(catalogManager::setCurrentCatalog);
testSpec.getDefaultDatabase().ifPresent(catalogManager::setCurrentDatabase);
- CatalogTableOperation tab = catalogManager.resolveTable(lookupPath.toArray(new String[0])).get();
+ CatalogManager.ResolvedTable tab = catalogManager.resolveTable(lookupPath.toArray(new String[0])).get();
assertThat(tab.getTablePath(), CoreMatchers.equalTo(testSpec.getExpectedPath()));
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/runtime/batch/table/JavaTableEnvironmentITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/runtime/batch/table/JavaTableEnvironmentITCase.java
index 78d723d..ffe40f6 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/runtime/batch/table/JavaTableEnvironmentITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/runtime/batch/table/JavaTableEnvironmentITCase.java
@@ -31,6 +31,7 @@ import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.table.api.PlannerConfig;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.table.calcite.CalciteConfigBuilder;
import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
@@ -71,6 +72,28 @@ public class JavaTableEnvironmentITCase extends TableProgramsCollectionTestBase
});
}
+ @Test(expected = ValidationException.class)
+ public void testIllegalEmptyName() throws Exception {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ BatchTableEnvironment tableEnv = BatchTableEnvironment.create(env, config());
+
+ DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
+ Table t = tableEnv.fromDataSet(ds);
+ // Must fail. Table is empty
+ tableEnv.registerTable("", t);
+ }
+
+ @Test(expected = ValidationException.class)
+ public void testIllegalWhitespaceOnlyName() throws Exception {
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ BatchTableEnvironment tableEnv = BatchTableEnvironment.create(env, config());
+
+ DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
+ Table t = tableEnv.fromDataSet(ds);
+ // Must fail. Table is empty
+ tableEnv.registerTable(" ", t);
+ }
+
@Test
public void testSimpleRegister() throws Exception {
final String tableName = "MyTable";
@@ -157,17 +180,6 @@ public class JavaTableEnvironmentITCase extends TableProgramsCollectionTestBase
}
@Test(expected = TableException.class)
- public void testIllegalName() throws Exception {
- ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- BatchTableEnvironment tableEnv = BatchTableEnvironment.create(env, config());
-
- DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
- Table t = tableEnv.fromDataSet(ds);
- // Must fail. Table name matches internal name pattern.
- tableEnv.registerTable("_DataSetTable_42", t);
- }
-
- @Test(expected = TableException.class)
public void testRegisterTableFromOtherEnv() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv1 = BatchTableEnvironment.create(env, config());
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/catalog/ExternalCatalogSchemaTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/catalog/ExternalCatalogSchemaTest.scala
index cec23b0..c49cd9a 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/catalog/ExternalCatalogSchemaTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/catalog/ExternalCatalogSchemaTest.scala
@@ -27,7 +27,7 @@ import org.apache.calcite.prepare.CalciteCatalogReader
import org.apache.calcite.schema.SchemaPlus
import org.apache.calcite.sql.validate.SqlMonikerType
import org.apache.flink.table.calcite.{FlinkTypeFactory, FlinkTypeSystem}
-import org.apache.flink.table.plan.schema.{TableSourceSinkTable, TableSourceTable}
+import org.apache.flink.table.plan.schema.TableSourceTable
import org.apache.flink.table.runtime.utils.CommonTestData
import org.apache.flink.table.sources.CsvTableSource
import org.apache.flink.table.utils.TableTestBase
@@ -77,13 +77,8 @@ class ExternalCatalogSchemaTest extends TableTestBase {
def testGetTable(): Unit = {
val relOptTable = calciteCatalogReader.getTable(Lists.newArrayList(schemaName, db, tb))
assertNotNull(relOptTable)
- val tableSourceSinkTable = relOptTable.unwrap(classOf[TableSourceSinkTable[_, _]])
- tableSourceSinkTable.tableSourceTable match {
- case Some(tst: TableSourceTable[_]) =>
- assertTrue(tst.tableSource.isInstanceOf[CsvTableSource])
- case _ =>
- fail("unexpected table type!")
- }
+ val tableSourceTable = relOptTable.unwrap(classOf[TableSourceTable[_]])
+ assertTrue(tableSourceTable.tableSource.isInstanceOf[CsvTableSource])
}
@Test
[flink] 02/03: [hotfix][table-planner] Fix getting external catalog
Posted by dw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 668259c70c3adfa3f073dfedabb1921f620b3177
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Fri May 24 13:41:21 2019 +0200
[hotfix][table-planner] Fix getting external catalog
---
.../src/main/scala/org/apache/flink/table/api/TableEnvImpl.scala | 5 +----
1 file changed, 1 insertion(+), 4 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/TableEnvImpl.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/TableEnvImpl.scala
index 7f329db..6db31ee 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/TableEnvImpl.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/TableEnvImpl.scala
@@ -83,9 +83,6 @@ abstract class TableEnvImpl(
// a counter for unique attribute names
private[flink] val attrNameCntr: AtomicInteger = new AtomicInteger(0)
- // registered external catalog names -> catalog
- private val externalCatalogs = new mutable.HashMap[String, ExternalCatalog]
-
private[flink] val operationTreeBuilder = new OperationTreeBuilder(this)
private val planningConfigurationBuilder: PlanningConfigurationBuilder =
@@ -329,7 +326,7 @@ abstract class TableEnvImpl(
}
override def getRegisteredExternalCatalog(name: String): ExternalCatalog = {
- this.externalCatalogs.get(name) match {
+ JavaScalaConversionUtil.toScala(catalogManager.getExternalCatalog(name)) match {
case Some(catalog) => catalog
case None => throw new ExternalCatalogNotExistException(name)
}
[flink] 01/03: [hotfix][table-api] Port TableSource specific
descriptors
Posted by dw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 99d4fd39f2b44630130d47b16ebcd69923f09e9a
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Fri May 24 12:55:48 2019 +0200
[hotfix][table-api] Port TableSource specific descriptors
---
.../flink/table/typeutils/FieldInfoUtils.java | 23 ++++--
.../table/sources/DefinedProctimeAttribute.java} | 38 ++++-----
.../table/sources/DefinedRowtimeAttributes.java} | 41 +++++-----
.../table/sources/RowtimeAttributeDescriptor.java | 77 ++++++++++++++++++
.../table/sources/definedTimeAttributes.scala | 95 ----------------------
5 files changed, 133 insertions(+), 141 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/typeutils/FieldInfoUtils.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/typeutils/FieldInfoUtils.java
index a0a3cc6..97c341c 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/typeutils/FieldInfoUtils.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/typeutils/FieldInfoUtils.java
@@ -49,7 +49,7 @@ import static java.lang.String.format;
import static org.apache.flink.table.expressions.BuiltInFunctionDefinitions.TIME_ATTRIBUTES;
/**
- * Utility classes for extracting names and indices of fields from different {@link TypeInformation}s.
+ * Utility methods for extracting names and indices of fields from different {@link TypeInformation}s.
*/
public class FieldInfoUtils {
@@ -234,8 +234,19 @@ public class FieldInfoUtils {
return fieldTypes;
}
- public static <T> TableSchema calculateTableSchema(
- TypeInformation<T> typeInfo,
+ /**
+ * Derives {@link TableSchema} out of a {@link TypeInformation}. It is complementary to other
+ * methods in this class. This also performs translation from time indicator markers such as
+ * {@link TimeIndicatorTypeInfo#ROWTIME_STREAM_MARKER} etc. to a corresponding
+ * {@link TimeIndicatorTypeInfo}.
+ *
+ * @param typeInfo input type info to calculate fields type infos from
+ * @param fieldIndexes indices within the typeInfo of the resulting Table schema
+ * @param fieldNames names of the fields of the resulting schema
+ * @return calculates resulting schema
+ */
+ public static TableSchema calculateTableSchema(
+ TypeInformation<?> typeInfo,
int[] fieldIndexes,
String[] fieldNames) {
@@ -292,6 +303,8 @@ public class FieldInfoUtils {
return new TableSchema(fieldNames, types);
}
+ /* Utility methods */
+
private static Optional<TypeInformation<?>> extractTimeMarkerType(int idx) {
switch (idx) {
case TimeIndicatorTypeInfo.ROWTIME_STREAM_MARKER:
@@ -306,10 +319,6 @@ public class FieldInfoUtils {
}
}
-
-
- /* Utility methods */
-
private static Set<FieldInfo> extractFieldInfoFromAtomicType(Expression[] exprs) {
boolean referenced = false;
FieldInfo fieldInfo = null;
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/StreamTableSourceTable.scala b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/DefinedProctimeAttribute.java
similarity index 51%
rename from flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/StreamTableSourceTable.scala
rename to flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/DefinedProctimeAttribute.java
index 38b7df8..803ca6a 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/StreamTableSourceTable.scala
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/DefinedProctimeAttribute.java
@@ -16,27 +16,27 @@
* limitations under the License.
*/
-package org.apache.flink.table.plan.schema
+package org.apache.flink.table.sources;
-import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory}
-import org.apache.flink.table.calcite.FlinkTypeFactory
-import org.apache.flink.table.plan.stats.FlinkStatistic
-import org.apache.flink.table.sources.{StreamTableSource, TableSourceUtil}
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.table.api.TableSchema;
-class StreamTableSourceTable[T](
- tableSource: StreamTableSource[T],
- statistic: FlinkStatistic = FlinkStatistic.UNKNOWN)
- extends TableSourceTable[T](
- tableSource,
- statistic) {
+import javax.annotation.Nullable;
- TableSourceUtil.validateTableSource(tableSource)
+/**
+ * Extends a {@link TableSource} to specify a processing time attribute.
+ */
+@PublicEvolving
+public interface DefinedProctimeAttribute {
- def getRowType(typeFactory: RelDataTypeFactory): RelDataType = {
- TableSourceUtil.getRelDataType(
- tableSource,
- None,
- streaming = true,
- typeFactory.asInstanceOf[FlinkTypeFactory])
- }
+ /**
+ * Returns the name of a processing time attribute or null if no processing time attribute is
+ * present.
+ *
+ * <p>The referenced attribute must be present in the {@link TableSchema} of the {@link TableSource} and of
+ * type {@link Types#SQL_TIMESTAMP}.
+ */
+ @Nullable
+ String getProctimeAttribute();
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/BatchTableSourceTable.scala b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/DefinedRowtimeAttributes.java
similarity index 50%
rename from flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/BatchTableSourceTable.scala
rename to flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/DefinedRowtimeAttributes.java
index 14e8cb1..5238084 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/schema/BatchTableSourceTable.scala
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/DefinedRowtimeAttributes.java
@@ -16,28 +16,29 @@
* limitations under the License.
*/
-package org.apache.flink.table.plan.schema
+package org.apache.flink.table.sources;
-import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory}
-import org.apache.flink.table.calcite.FlinkTypeFactory
-import org.apache.flink.table.plan.stats.FlinkStatistic
-import org.apache.flink.table.sources.{BatchTableSource, TableSourceUtil}
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.table.api.TableSchema;
-class BatchTableSourceTable[T](
- tableSource: BatchTableSource[T],
- statistic: FlinkStatistic = FlinkStatistic.UNKNOWN)
- extends TableSourceTable[T](
- tableSource,
- statistic) {
+import java.util.List;
- TableSourceUtil.validateTableSource(tableSource)
+/**
+ * Extends a {@link TableSource} to specify rowtime attributes via a
+ * {@link RowtimeAttributeDescriptor}.
+ */
+@PublicEvolving
+public interface DefinedRowtimeAttributes {
- override def getRowType(typeFactory: RelDataTypeFactory): RelDataType = {
- TableSourceUtil.getRelDataType(
- tableSource,
- None,
- streaming = false,
- typeFactory.asInstanceOf[FlinkTypeFactory])
- }
+ /**
+ * Returns a list of {@link RowtimeAttributeDescriptor} for all rowtime
+ * attributes of the table.
+ *
+ * <p>All referenced attributes must be present in the {@link TableSchema}
+ * of the {@link TableSource} and of type {@link Types#SQL_TIMESTAMP}.
+ *
+ * @return A list of {@link RowtimeAttributeDescriptor}.
+ */
+ List<RowtimeAttributeDescriptor> getRowtimeAttributeDescriptors();
}
-
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/RowtimeAttributeDescriptor.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/RowtimeAttributeDescriptor.java
new file mode 100644
index 0000000..6851dd2
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/sources/RowtimeAttributeDescriptor.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.sources;
+
+import org.apache.flink.table.sources.tsextractors.TimestampExtractor;
+import org.apache.flink.table.sources.wmstrategies.WatermarkStrategy;
+
+import java.util.Objects;
+
+/**
+ * Describes a rowtime attribute of a {@link TableSource}.
+ */
+public final class RowtimeAttributeDescriptor {
+
+ private final String attributeName;
+ private final TimestampExtractor timestampExtractor;
+ private final WatermarkStrategy watermarkStrategy;
+
+ public RowtimeAttributeDescriptor(
+ String attributeName,
+ TimestampExtractor timestampExtractor,
+ WatermarkStrategy watermarkStrategy) {
+ this.attributeName = attributeName;
+ this.timestampExtractor = timestampExtractor;
+ this.watermarkStrategy = watermarkStrategy;
+ }
+
+ /** Returns the name of the rowtime attribute. */
+ public String getAttributeName() {
+ return attributeName;
+ }
+
+ /** Returns the [[TimestampExtractor]] for the attribute. */
+ public TimestampExtractor getTimestampExtractor() {
+ return timestampExtractor;
+ }
+
+ /** Returns the [[WatermarkStrategy]] for the attribute. */
+ public WatermarkStrategy getWatermarkStrategy() {
+ return watermarkStrategy;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ RowtimeAttributeDescriptor that = (RowtimeAttributeDescriptor) o;
+ return Objects.equals(attributeName, that.attributeName) &&
+ Objects.equals(timestampExtractor, that.timestampExtractor) &&
+ Objects.equals(watermarkStrategy, that.watermarkStrategy);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(attributeName, timestampExtractor, watermarkStrategy);
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sources/definedTimeAttributes.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sources/definedTimeAttributes.scala
deleted file mode 100644
index b144312..0000000
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/sources/definedTimeAttributes.scala
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * 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.sources
-
-import java.util
-import java.util.Objects
-import javax.annotation.Nullable
-
-import org.apache.flink.table.api.TableSchema
-import org.apache.flink.table.api.Types
-import org.apache.flink.table.sources.tsextractors.TimestampExtractor
-import org.apache.flink.table.sources.wmstrategies.WatermarkStrategy
-
-/**
- * Extends a [[TableSource]] to specify a processing time attribute.
- */
-trait DefinedProctimeAttribute {
-
- /**
- * Returns the name of a processing time attribute or null if no processing time attribute is
- * present.
- *
- * The referenced attribute must be present in the [[TableSchema]] of the [[TableSource]] and of
- * type [[Types.SQL_TIMESTAMP]].
- */
- @Nullable
- def getProctimeAttribute: String
-}
-
-/**
- * Extends a [[TableSource]] to specify rowtime attributes via a
- * [[RowtimeAttributeDescriptor]].
- */
-trait DefinedRowtimeAttributes {
-
- /**
- * Returns a list of [[RowtimeAttributeDescriptor]] for all rowtime attributes of the table.
- *
- * All referenced attributes must be present in the [[TableSchema]] of the [[TableSource]] and of
- * type [[Types.SQL_TIMESTAMP]].
- *
- * @return A list of [[RowtimeAttributeDescriptor]].
- */
- def getRowtimeAttributeDescriptors: util.List[RowtimeAttributeDescriptor]
-}
-
-/**
- * Describes a rowtime attribute of a [[TableSource]].
- *
- * @param attributeName The name of the rowtime attribute.
- * @param timestampExtractor The timestamp extractor to derive the values of the attribute.
- * @param watermarkStrategy The watermark strategy associated with the attribute.
- */
-class RowtimeAttributeDescriptor(
- val attributeName: String,
- val timestampExtractor: TimestampExtractor,
- val watermarkStrategy: WatermarkStrategy) {
-
- /** Returns the name of the rowtime attribute. */
- def getAttributeName: String = attributeName
-
- /** Returns the [[TimestampExtractor]] for the attribute. */
- def getTimestampExtractor: TimestampExtractor = timestampExtractor
-
- /** Returns the [[WatermarkStrategy]] for the attribute. */
- def getWatermarkStrategy: WatermarkStrategy = watermarkStrategy
-
- override def equals(other: Any): Boolean = other match {
- case that: RowtimeAttributeDescriptor =>
- Objects.equals(attributeName, that.attributeName) &&
- Objects.equals(timestampExtractor, that.timestampExtractor) &&
- Objects.equals(watermarkStrategy, that.watermarkStrategy)
- case _ => false
- }
-
- override def hashCode(): Int = {
- Objects.hash(attributeName, timestampExtractor, watermarkStrategy)
- }
-}