You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2021/08/31 08:16:32 UTC
[flink] 01/03: [FLINK-24004][table-planner] Introduce
TableFactoryHarness
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit bc69c1c77ec94a57b3603c5eefcbddb8ce83ea64
Author: Ingo Bürk <in...@tngtech.com>
AuthorDate: Mon Aug 30 08:39:53 2021 +0200
[FLINK-24004][table-planner] Introduce TableFactoryHarness
---
.../apache/flink/table/api/TableDescriptor.java | 19 +-
.../table/api/internal/TableEnvironmentImpl.java | 26 +-
.../flink/table/catalog/DefaultCatalogTable.java | 4 +-
.../planner/factories/TableFactoryHarness.java | 423 +++++++++++++++++++++
.../org.apache.flink.table.factories.Factory | 1 +
5 files changed, 445 insertions(+), 28 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableDescriptor.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableDescriptor.java
index da21a8c..e60ad10 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableDescriptor.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableDescriptor.java
@@ -51,14 +51,14 @@ import java.util.stream.Collectors;
* TableEnvironment#createTemporaryTable(String, TableDescriptor)}.
*/
@PublicEvolving
-public final class TableDescriptor {
+public class TableDescriptor {
private final @Nullable Schema schema;
private final Map<String, String> options;
private final List<String> partitionKeys;
private final @Nullable String comment;
- private TableDescriptor(
+ protected TableDescriptor(
@Nullable Schema schema,
Map<String, String> options,
List<String> partitionKeys,
@@ -101,6 +101,21 @@ public final class TableDescriptor {
// ---------------------------------------------------------------------------------------------
+ /** Converts this descriptor into a {@link CatalogTable}. */
+ public CatalogTable toCatalogTable() {
+ final Schema schema =
+ getSchema()
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ "Missing schema in TableDescriptor. "
+ + "A schema is typically required. "
+ + "It can only be omitted at certain "
+ + "documented locations."));
+
+ return CatalogTable.of(schema, getComment().orElse(null), getPartitionKeys(), getOptions());
+ }
+
/** Converts this immutable instance into a mutable {@link Builder}. */
public Builder toBuilder() {
return new Builder(this);
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
index 78e1b35..892645c 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
@@ -29,7 +29,6 @@ import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.ExplainDetail;
import org.apache.flink.table.api.ResultKind;
-import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.SqlParserException;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.Table;
@@ -470,8 +469,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
private void createTemporaryTableInternal(
UnresolvedIdentifier path, TableDescriptor descriptor) {
final ObjectIdentifier tableIdentifier = catalogManager.qualifyIdentifier(path);
- final CatalogTable catalogTable = convertTableDescriptor(descriptor);
- catalogManager.createTemporaryTable(catalogTable, tableIdentifier, false);
+ catalogManager.createTemporaryTable(descriptor.toCatalogTable(), tableIdentifier, false);
}
@Override
@@ -481,27 +479,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
final ObjectIdentifier tableIdentifier =
catalogManager.qualifyIdentifier(getParser().parseIdentifier(path));
- final CatalogTable catalogTable = convertTableDescriptor(descriptor);
- catalogManager.createTable(catalogTable, tableIdentifier, false);
- }
-
- private CatalogTable convertTableDescriptor(TableDescriptor descriptor) {
- final Schema schema =
- descriptor
- .getSchema()
- .orElseThrow(
- () ->
- new ValidationException(
- "Missing schema in TableDescriptor. "
- + "A schema is typically required. "
- + "It can only be omitted at certain "
- + "documented locations."));
-
- return CatalogTable.of(
- schema,
- descriptor.getComment().orElse(null),
- descriptor.getPartitionKeys(),
- descriptor.getOptions());
+ catalogManager.createTable(descriptor.toCatalogTable(), tableIdentifier, false);
}
@Override
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DefaultCatalogTable.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DefaultCatalogTable.java
index 35b713e..c149a67 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DefaultCatalogTable.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DefaultCatalogTable.java
@@ -33,14 +33,14 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
/** Default implementation of a {@link CatalogTable}. */
@Internal
-class DefaultCatalogTable implements CatalogTable {
+public class DefaultCatalogTable implements CatalogTable {
private final Schema schema;
private final @Nullable String comment;
private final List<String> partitionKeys;
private final Map<String, String> options;
- DefaultCatalogTable(
+ protected DefaultCatalogTable(
Schema schema,
@Nullable String comment,
List<String> partitionKeys,
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java
new file mode 100644
index 0000000..9d15acf
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TableFactoryHarness.java
@@ -0,0 +1,423 @@
+/*
+ * 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.planner.factories;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableDescriptor;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.DefaultCatalogTable;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkFunctionProvider;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.LookupTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.connector.source.TableFunctionProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.testutils.junit.SharedObjects;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Provides a flexible testing harness for table factories.
+ *
+ * <p>This testing harness allows writing custom sources and sinks which can be directly
+ * instantiated from the test. This avoids having to implement a factory, and enables using the
+ * {@link SharedObjects} rule to get direct access to the underlying source/sink from the test.
+ *
+ * <p>The underlying source/sink must extend from {@link SourceBase} or {@link SinkBase}.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * public class CustomSourceTest {
+ * {@literal @}Rule public SharedObjects sharedObjects = SharedObjects.create();
+ *
+ * {@literal @}Test
+ * public void test() {
+ * SharedReference<List<Long>> appliedLimits = sharedObjects.add(new ArrayList<>());
+ * TableDescriptor sourceDescriptor =
+ * TableFactoryHarness.newBuilder()
+ * .schema(Schema.derived())
+ * .source(new CustomSource(appliedLimits))
+ * .build();
+ * tEnv.createTable("T", sourceDescriptor);
+ *
+ * tEnv.explainSql("SELECT * FROM T LIMIT 42");
+ *
+ * assertEquals(1, appliedLimits.get().size());
+ * assertEquals((Long) 42L, appliedLimits.get().get(0));
+ * }
+ *
+ * private static class CustomSource extends TableFactoryHarness.ScanSourceBase
+ * implements SupportsLimitPushDown {
+ * private final SharedReference<List<Long>> appliedLimits;
+ *
+ * CustomSource(SharedReference<List<Long>> appliedLimits) {
+ * this.appliedLimits = appliedLimits;
+ * }
+ *
+ * public void applyLimit(long limit) {
+ * appliedLimits.get().add(limit);
+ * }
+ * }
+ * }
+ * }</pre>
+ */
+public class TableFactoryHarness {
+
+ /** Factory identifier for {@link Factory}. */
+ public static final String IDENTIFIER = "harness";
+
+ // ---------------------------------------------------------------------------------------------
+
+ /**
+ * Creates a builder for a new {@link TableDescriptor} specialized for this harness.
+ *
+ * <p>Use this method to create a {@link TableDescriptor} and passing in the source / sink
+ * implementation you want to use. The descriptor can for example be used with {@link
+ * TableEnvironment#createTable(String, TableDescriptor)} to register a table.
+ */
+ public static HarnessTableDescriptor.Builder newBuilder() {
+ return new HarnessTableDescriptor.Builder();
+ }
+
+ // ---------------------------------------------------------------------------------------------
+
+ /**
+ * Factory which creates a source / sink defined in the specialized {@link HarnessCatalogTable}.
+ */
+ public static class Factory implements DynamicTableSourceFactory, DynamicTableSinkFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ final FactoryUtil.TableFactoryHelper factoryHelper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+ factoryHelper.validate();
+
+ final HarnessCatalogTable catalogTable =
+ (HarnessCatalogTable) context.getCatalogTable().getOrigin();
+ if (catalogTable.source == null) {
+ throw new ValidationException(
+ String.format(
+ "Table '%s' has no source configured.",
+ context.getObjectIdentifier()));
+ }
+
+ catalogTable.source.factoryContext = context;
+ return catalogTable.source;
+ }
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ final FactoryUtil.TableFactoryHelper factoryHelper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+ factoryHelper.validate();
+
+ final HarnessCatalogTable catalogTable =
+ (HarnessCatalogTable) context.getCatalogTable().getOrigin();
+ if (catalogTable.sink == null) {
+ throw new ValidationException(
+ String.format(
+ "Table '%s' has no sink configured.",
+ context.getObjectIdentifier()));
+ }
+
+ catalogTable.sink.factoryContext = context;
+ return catalogTable.sink;
+ }
+ }
+
+ /**
+ * Specialized version of {@link TableDescriptor} which allows passing a custom source / sink to
+ * the {@link CatalogTable} created from it.
+ */
+ private static class HarnessTableDescriptor extends TableDescriptor {
+
+ private final @Nullable SourceBase source;
+ private final @Nullable SinkBase sink;
+
+ private HarnessTableDescriptor(
+ Schema schema, @Nullable SourceBase source, @Nullable SinkBase sink) {
+ super(
+ schema,
+ Collections.singletonMap(FactoryUtil.CONNECTOR.key(), IDENTIFIER),
+ Collections.emptyList(),
+ null);
+
+ this.source = source;
+ this.sink = sink;
+ }
+
+ @Override
+ public CatalogTable toCatalogTable() {
+ return new HarnessCatalogTable(super.toCatalogTable(), source, sink);
+ }
+
+ /** Builder for {@link HarnessTableDescriptor}. */
+ public static class Builder {
+ private @Nullable Schema schema;
+ private @Nullable SourceBase source;
+ private @Nullable SinkBase sink;
+
+ /** Define the schema of the {@link TableDescriptor}. */
+ public Builder schema(Schema schema) {
+ this.schema = schema;
+ return this;
+ }
+
+ /** Use a bounded {@link ScanTableSource} which produces no data. */
+ public Builder boundedScanSource() {
+ return source(new ScanSourceBase(true) {});
+ }
+
+ /** Use an unbounded {@link ScanTableSource} which produces no data. */
+ public Builder unboundedScanSource() {
+ return source(new ScanSourceBase(false) {});
+ }
+
+ /**
+ * Use an unbounded {@link ScanTableSource} with the given {@param changelogMode} which
+ * produces no data.
+ */
+ public Builder unboundedScanSource(ChangelogMode changelogMode) {
+ return source(
+ new ScanSourceBase(false) {
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return changelogMode;
+ }
+ });
+ }
+
+ /** Use a {@link LookupTableSource} which produces no data. */
+ public Builder lookupSource() {
+ return source(new LookupSourceBase() {});
+ }
+
+ /** Use a custom {@link DynamicTableSource}. */
+ public Builder source(SourceBase source) {
+ this.source = source;
+ return this;
+ }
+
+ /** Use a {@link DynamicTableSink} which discards all data. */
+ public Builder sink() {
+ return sink(new SinkBase() {});
+ }
+
+ /** Use a custom {@link DynamicTableSink}. */
+ public Builder sink(SinkBase sink) {
+ this.sink = sink;
+ return this;
+ }
+
+ /** Builds a {@link TableDescriptor}. */
+ public TableDescriptor build() {
+ return new HarnessTableDescriptor(schema, source, sink);
+ }
+ }
+ }
+
+ /** Specialized {@link CatalogTable} which contains a custom source / sink. */
+ private static class HarnessCatalogTable extends DefaultCatalogTable {
+
+ private final @Nullable SourceBase source;
+ private final @Nullable SinkBase sink;
+
+ public HarnessCatalogTable(
+ CatalogTable parentTable, @Nullable SourceBase source, @Nullable SinkBase sink) {
+ super(
+ parentTable.getUnresolvedSchema(),
+ parentTable.getComment(),
+ parentTable.getPartitionKeys(),
+ parentTable.getOptions());
+
+ this.source = source;
+ this.sink = sink;
+ }
+
+ @Override
+ public CatalogBaseTable copy() {
+ return copy(getOptions());
+ }
+
+ @Override
+ public CatalogTable copy(Map<String, String> options) {
+ final CatalogTable parentTable =
+ CatalogTable.of(
+ getUnresolvedSchema(), getComment(), getPartitionKeys(), options);
+ return new HarnessCatalogTable(parentTable, source, sink);
+ }
+ }
+
+ // ---------------------------------------------------------------------------------------------
+
+ /**
+ * Base class for custom sources which implement {@link ScanTableSource}.
+ *
+ * <p>Most interface methods are default-implemented for convenience, but can be overridden when
+ * necessary. By default, a {@link ScanRuntimeProvider} is used which doesn't produce anything.
+ *
+ * <p>Sources derived from this base class will also be provided the {@link
+ * DynamicTableFactory.Context} of the factory which gives access to e.g. the {@link
+ * CatalogTable}.
+ */
+ public abstract static class ScanSourceBase extends SourceBase implements ScanTableSource {
+
+ private final boolean bounded;
+
+ public ScanSourceBase() {
+ this(true);
+ }
+
+ public ScanSourceBase(boolean bounded) {
+ this.bounded = bounded;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.insertOnly();
+ }
+
+ @Override
+ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
+ return SourceFunctionProvider.of(
+ new SourceFunction<RowData>() {
+ @Override
+ public void run(SourceContext<RowData> ctx) {}
+
+ @Override
+ public void cancel() {}
+ },
+ bounded);
+ }
+ }
+
+ /**
+ * Base class for custom sources which implement {@link LookupTableSource}.
+ *
+ * <p>Most interface methods are default-implemented for convenience, but can be overridden when
+ * necessary. By default, a {@link LookupRuntimeProvider} is used which doesn't produce
+ * anything.
+ *
+ * <p>Sources derived from this base class will also be provided the {@link
+ * DynamicTableFactory.Context} of the factory which gives access to e.g. the {@link
+ * CatalogTable}.
+ */
+ public abstract static class LookupSourceBase extends SourceBase implements LookupTableSource {
+ @Override
+ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
+ return TableFunctionProvider.of(new TableFunction<RowData>() {});
+ }
+ }
+
+ /**
+ * Base class for custom sinks.
+ *
+ * <p>Most interface methods are default-implemented for convenience, but can be overridden when
+ * necessary. By default, a {@link SinkRuntimeProvider} is used which does nothing.
+ *
+ * <p>Sinks derived from this base class will also be provided the {@link
+ * DynamicTableFactory.Context} of the factory which gives access to e.g. the {@link
+ * CatalogTable}.
+ */
+ public abstract static class SinkBase implements DynamicTableSink {
+
+ private DynamicTableFactory.Context factoryContext;
+
+ @Override
+ public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
+ return ChangelogMode.all();
+ }
+
+ @Override
+ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+ return SinkFunctionProvider.of(
+ new SinkFunction<RowData>() {
+ @Override
+ public void invoke(RowData value, Context context1) {}
+ });
+ }
+
+ @Override
+ public DynamicTableSink copy() {
+ return this;
+ }
+
+ @Override
+ public String asSummaryString() {
+ return "Unspecified Testing Sink";
+ }
+
+ public DynamicTableFactory.Context getFactoryContext() {
+ return factoryContext;
+ }
+ }
+
+ /** Base class for {@link ScanSourceBase} and {{@link LookupSourceBase}}. */
+ private abstract static class SourceBase implements DynamicTableSource {
+ private DynamicTableFactory.Context factoryContext;
+
+ @Override
+ public DynamicTableSource copy() {
+ return this;
+ }
+
+ @Override
+ public String asSummaryString() {
+ return "Unspecified Testing Source";
+ }
+
+ public DynamicTableFactory.Context getFactoryContext() {
+ return factoryContext;
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink-table/flink-table-planner/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory
index 05bddfe..616b708 100644
--- a/flink-table/flink-table-planner/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory
+++ b/flink-table/flink-table-planner/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -16,3 +16,4 @@
org.apache.flink.table.planner.factories.TestValuesTableFactory
org.apache.flink.table.planner.factories.TestFileFactory
org.apache.flink.table.planner.plan.rules.logical.PushProjectIntoTableSourceScanRuleTest$MetadataNoProjectionPushDownTableFactory
+org.apache.flink.table.planner.factories.TableFactoryHarness$Factory