wuchong commented on a change in pull request #14822:
URL: https://github.com/apache/flink/pull/14822#discussion_r580011204
##########
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
##########
@@ -201,6 +185,31 @@ protected CommonExecSink(
finalInputTransform = new
PartitionTransformation<>(inputTransform, partitioner);
finalInputTransform.setParallelism(parallelism);
}
+
+ final SinkFunction<RowData> sinkFunction;
+ if (runtimeProvider instanceof SinkFunctionProvider) {
+ sinkFunction = ((SinkFunctionProvider)
runtimeProvider).createSinkFunction();
+ } else if (runtimeProvider instanceof OutputFormatProvider) {
+ OutputFormat<RowData> outputFormat =
+ ((OutputFormatProvider)
runtimeProvider).createOutputFormat();
+ sinkFunction = new OutputFormatSinkFunction<>(outputFormat);
+ } else if (runtimeProvider instanceof SinkProvider) {
+ return new SinkTransformation<>(
+ finalInputTransform,
+ ((SinkProvider) runtimeProvider).createSink(),
Review comment:
This still misses to apply not-null-enforcer.
Besides, could you add an IT case for not-null-enforcer + unified sink
interface?
##########
File path:
flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/TableSinkITCase.scala
##########
@@ -838,6 +841,44 @@ class TableSinkITCase extends StreamingTestBase {
}
+ @Test
+ def testSinkProvider(): Unit = {
Review comment:
nit: what about `testUnifiedSinkInterface`? Currently, all the
SinkRuntimeProviders have keywords of `Sink` and `Provider`.
##########
File path:
flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/factories/TestFileFactory.java
##########
@@ -31,27 +33,40 @@
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkProvider;
import org.apache.flink.table.connector.source.DataStreamScanProvider;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.connector.source.ScanTableSource;
import org.apache.flink.table.connector.source.SourceProvider;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.filesystem.FileSystemOptions;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
+import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
-/** Test file source {@link DynamicTableSourceFactory}. */
-public class TestFileSourceFactory implements DynamicTableSourceFactory {
+/**
+ * Test implementation of {@link DynamicTableSourceFactory} and {@link
DynamicTableSinkFactory} that
+ * creates a file source and sink based on {@link SourceProvider} and {@link
SinkProvider}.
+ */
+public class TestFileFactory implements DynamicTableSourceFactory,
DynamicTableSinkFactory {
+
+ //
--------------------------------------------------------------------------------------------
+ // Factory
+ //
--------------------------------------------------------------------------------------------
+
+ private static final String IDENTIFIER = "file";
Review comment:
What about using `test-file` to make it more clearer it is only used for
testing.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
