[hotfix][table-common] Improve source and sink provider docs

is · Aug 31, 2021 · 7d8af8f · 7d8af8f
1 parent 4726878
commit 7d8af8f
Show file tree

Hide file tree

Showing 8 changed files with 43 additions and 22 deletions.
diff --git a/...va-bridge/src/main/java/org/apache/flink/table/connector/sink/DataStreamSinkProvider.java b/...va-bridge/src/main/java/org/apache/flink/table/connector/sink/DataStreamSinkProvider.java
@@ -28,8 +28,9 @@
  * DynamicTableSink}.
  *
  * <p>Note: This provider is only meant for advanced connector developers. Usually, a sink should
- * consist of a single entity expressed via {@link OutputFormatProvider} or {@link
- * SinkFunctionProvider}, or {@link SinkProvider}.
+ * consist of a single entity expressed via {@link SinkProvider}, {@link SinkFunctionProvider}, or
+ * {@link OutputFormatProvider}. When using a {@link DataStream} an implementer needs to pay
+ * attention to how changes are shuffled to not mess up the changelog per parallel subtask.
  */
 @PublicEvolving
 public interface DataStreamSinkProvider extends DynamicTableSink.SinkRuntimeProvider {

diff --git a/...java-bridge/src/main/java/org/apache/flink/table/connector/sink/SinkFunctionProvider.java b/...java-bridge/src/main/java/org/apache/flink/table/connector/sink/SinkFunctionProvider.java
@@ -23,6 +23,8 @@
 import org.apache.flink.table.connector.ParallelismProvider;
 import org.apache.flink.table.data.RowData;
 
+import javax.annotation.Nullable;
+
 import java.util.Optional;
 
 /**
@@ -39,7 +41,8 @@ static SinkFunctionProvider of(SinkFunction<RowData> sinkFunction) {
  }
 
  /** Helper method for creating a SinkFunction provider with a provided sink parallelism. */
- static SinkFunctionProvider of(SinkFunction<RowData> sinkFunction, Integer sinkParallelism) {
+ static SinkFunctionProvider of(
+ SinkFunction<RowData> sinkFunction, @Nullable Integer sinkParallelism) {
  return new SinkFunctionProvider() {
 
  @Override

diff --git a/...-bridge/src/main/java/org/apache/flink/table/connector/source/DataStreamScanProvider.java b/...-bridge/src/main/java/org/apache/flink/table/connector/source/DataStreamScanProvider.java
@@ -28,8 +28,8 @@
  * ScanTableSource}.
  *
  * <p>Note: This provider is only meant for advanced connector developers. Usually, a source should
- * consist of a single entity expressed via {@link InputFormatProvider}, {@link
- * SourceFunctionProvider}, or {@link SourceProvider}.
+ * consist of a single entity expressed via {@link SourceProvider}, {@link SourceFunctionProvider},
+ * or {@link InputFormatProvider}.
  */
 @PublicEvolving
 public interface DataStreamScanProvider extends ScanTableSource.ScanRuntimeProvider {

diff --git a/...nk-table-common/src/main/java/org/apache/flink/table/connector/sink/DynamicTableSink.java b/...nk-table-common/src/main/java/org/apache/flink/table/connector/sink/DynamicTableSink.java
@@ -22,7 +22,6 @@
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.table.catalog.ResolvedSchema;
 import org.apache.flink.table.connector.ChangelogMode;
-import org.apache.flink.table.connector.ParallelismProvider;
 import org.apache.flink.table.connector.RuntimeConverter;
 import org.apache.flink.table.connector.sink.abilities.SupportsOverwrite;
 import org.apache.flink.table.connector.sink.abilities.SupportsPartitioning;
@@ -101,10 +100,11 @@ public interface DynamicTableSink {
  * <p>The given {@link Context} offers utilities by the planner for creating runtime
  * implementation with minimal dependencies to internal data structures.
  *
- * <p>See {@code org.apache.flink.table.connector.sink.SinkFunctionProvider} in {@code
- * flink-table-api-java-bridge}.
+ * <p>{@link SinkProvider} is the recommended core interface. {@code SinkFunctionProvider} in
+ * {@code flink-table-api-java-bridge} and {@link OutputFormatProvider} are available for
+ * backwards compatibility.
  *
- * @see ParallelismProvider
+ * @see SinkProvider
  */
  SinkRuntimeProvider getSinkRuntimeProvider(Context context);
 
@@ -186,8 +186,11 @@ interface DataStructureConverter extends RuntimeConverter {
  * SinkRuntimeProvider} serves as the base interface. Concrete {@link SinkRuntimeProvider}
  * interfaces might be located in other Flink modules.
  *
- * <p>See {@code org.apache.flink.table.connector.sink.SinkFunctionProvider} in {@code
- * flink-table-api-java-bridge}.
+ * <p>{@link SinkProvider} is the recommended core interface. {@code SinkFunctionProvider} in
+ * {@code flink-table-api-java-bridge} and {@link OutputFormatProvider} are available for
+ * backwards compatibility.
+ *
+ * @see SinkProvider
  */
  interface SinkRuntimeProvider {
  // marker interface

diff --git a/.../flink-table-common/src/main/java/org/apache/flink/table/connector/sink/SinkProvider.java b/.../flink-table-common/src/main/java/org/apache/flink/table/connector/sink/SinkProvider.java
@@ -23,9 +23,16 @@
 import org.apache.flink.table.connector.ParallelismProvider;
 import org.apache.flink.table.data.RowData;
 
+import javax.annotation.Nullable;
+
 import java.util.Optional;
 
-/** Provider of a {@link Sink} instance as a runtime implementation for {@link DynamicTableSink}. */
+/**
+ * Provider of a {@link Sink} instance as a runtime implementation for {@link DynamicTableSink}.
+ *
+ * <p>{@code DataStreamSinkProvider} in {@code flink-table-api-java-bridge} is available for
+ * advanced connector developers.
+ */
 @PublicEvolving
 public interface SinkProvider extends DynamicTableSink.SinkRuntimeProvider, ParallelismProvider {
 
@@ -35,7 +42,7 @@ static SinkProvider of(Sink<RowData, ?, ?, ?> sink) {
  }
 
  /** Helper method for creating a Sink provider with a provided sink parallelism. */
- static SinkProvider of(Sink<RowData, ?, ?, ?> sink, Integer sinkParallelism) {
+ static SinkProvider of(Sink<RowData, ?, ?, ?> sink, @Nullable Integer sinkParallelism) {
  return new SinkProvider() {
 
  @Override

diff --git a/...k-table-common/src/main/java/org/apache/flink/table/connector/source/ScanTableSource.java b/...k-table-common/src/main/java/org/apache/flink/table/connector/source/ScanTableSource.java
@@ -85,8 +85,11 @@ public interface ScanTableSource extends DynamicTableSource {
  * <p>The given {@link ScanContext} offers utilities by the planner for creating runtime
  * implementation with minimal dependencies to internal data structures.
  *
- * <p>See {@code org.apache.flink.table.connector.source.SourceFunctionProvider} in {@code
- * flink-table-api-java-bridge}.
+ * <p>{@link SourceProvider} is the recommended core interface. {@code SourceFunctionProvider}
+ * in {@code flink-table-api-java-bridge} and {@link InputFormatProvider} are available for
+ * backwards compatibility.
+ *
+ * @see SourceProvider
  */
  ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext);
 
@@ -115,8 +118,9 @@ interface ScanContext extends DynamicTableSource.Context {
  * ScanRuntimeProvider} serves as the base interface. Concrete {@link ScanRuntimeProvider}
  * interfaces might be located in other Flink modules.
  *
- * <p>See {@code org.apache.flink.table.connector.source.SourceFunctionProvider} in {@code
- * flink-table-api-java-bridge}.
+ * <p>{@link SourceProvider} is the recommended core interface. {@code SourceFunctionProvider}
+ * in {@code flink-table-api-java-bridge} and {@link InputFormatProvider} are available for
+ * backwards compatibility.
  */
  interface ScanRuntimeProvider {
 

diff --git a/...nk-table-common/src/main/java/org/apache/flink/table/connector/source/SourceProvider.java b/...nk-table-common/src/main/java/org/apache/flink/table/connector/source/SourceProvider.java
@@ -25,6 +25,9 @@
 
 /**
  * Provider of a {@link Source} instance as a runtime implementation for {@link ScanTableSource}.
+ *
+ * <p>{@code DataStreamScanProvider} in {@code flink-table-api-java-bridge} is available for
+ * advanced connector developers.
  */
 @PublicEvolving
 public interface SourceProvider extends ScanTableSource.ScanRuntimeProvider {

diff --git a/.../main/java/org/apache/flink/table/connector/source/abilities/SupportsReadingMetadata.java b/.../main/java/org/apache/flink/table/connector/source/abilities/SupportsReadingMetadata.java
@@ -20,7 +20,7 @@
 
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.ResolvedSchema;
 import org.apache.flink.table.connector.format.DecodingFormat;
 import org.apache.flink.table.connector.source.ScanTableSource;
 import org.apache.flink.table.data.RowData;
@@ -120,10 +120,10 @@ public interface SupportsReadingMetadata {
  * <p>Implementations of this method must be idempotent. The planner might call this method
  * multiple times.
  *
- * <p>Note: Use the passed data type instead of {@link TableSchema#toPhysicalRowDataType()} for
- * describing the final output data type when creating {@link TypeInformation}. If the source
- * implements {@link SupportsProjectionPushDown}, the projection is already considered in the
- * given output data type.
+ * <p>Note: Use the passed data type instead of {@link ResolvedSchema#toPhysicalRowDataType()}
+ * for describing the final output data type when creating {@link TypeInformation}. If the
+ * source implements {@link SupportsProjectionPushDown}, the projection is already considered in
+ * the given output data type.
  *
  * @param metadataKeys a subset of the keys returned by {@link #listReadableMetadata()}, ordered
  * by the iteration order of returned map