You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by li...@apache.org on 2024/01/26 13:33:39 UTC
(flink) branch master updated: [FLINK-33264][table] Support source parallelism setting for DataGen connector
This is an automated email from the ASF dual-hosted git repository.
libenchao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 862a7129d27 [FLINK-33264][table] Support source parallelism setting for DataGen connector
862a7129d27 is described below
commit 862a7129d2730b4c70a21826a5b858fc541a4470
Author: Zhanghao Chen <m1...@outlook.com>
AuthorDate: Thu Jan 18 21:26:42 2024 +0800
[FLINK-33264][table] Support source parallelism setting for DataGen connector
Close apache/flink#24133
---
docs/content.zh/docs/connectors/table/datagen.md | 7 ++++
docs/content/docs/connectors/table/datagen.md | 7 ++++
.../f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e | 4 +-
.../src/test/resources/sql/table.q | 1 +
.../src/test/resources/sql/table.q | 1 +
.../datagen/table/DataGenConnectorOptions.java | 3 ++
.../datagen/table/DataGenTableSource.java | 11 ++++--
.../datagen/table/DataGenTableSourceFactory.java | 5 ++-
.../factories/DataGenTableSourceFactoryTest.java | 46 ++++++++++++++++++++++
.../stream/table/DataGeneratorConnectorITCase.java | 25 ++++++++++++
10 files changed, 104 insertions(+), 6 deletions(-)
diff --git a/docs/content.zh/docs/connectors/table/datagen.md b/docs/content.zh/docs/connectors/table/datagen.md
index 642382a7768..b90183fcc8c 100644
--- a/docs/content.zh/docs/connectors/table/datagen.md
+++ b/docs/content.zh/docs/connectors/table/datagen.md
@@ -266,6 +266,13 @@ CREATE TABLE Orders (
<td>Long</td>
<td>生成数据的总行数。默认情况下，该表是无界的。</td>
</tr>
+ <tr>
+ <td><h5>scan.parallelism</h5></td>
+ <td>可选</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>定义算子并行度。不设置将使用全局默认并发。</td>
+ </tr>
<tr>
<td><h5>fields.#.kind</h5></td>
<td>可选</td>
diff --git a/docs/content/docs/connectors/table/datagen.md b/docs/content/docs/connectors/table/datagen.md
index 70253786bff..1f105076897 100644
--- a/docs/content/docs/connectors/table/datagen.md
+++ b/docs/content/docs/connectors/table/datagen.md
@@ -270,6 +270,13 @@ Connector Options
<td>Long</td>
<td>The total number of rows to emit. By default, the table is unbounded.</td>
</tr>
+ <tr>
+ <td><h5>scan.parallelism</h5></td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>Defines the parallelism of the source. If not set, the global default parallelism is used.</td>
+ </tr>
<tr>
<td><h5>fields.#.kind</h5></td>
<td>optional</td>
diff --git a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e
index 4e44bbf3505..0fd86b29430 100644
--- a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e
+++ b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e
@@ -76,8 +76,8 @@ Constructor <org.apache.flink.connector.datagen.source.GeneratingIteratorSourceR
Constructor <org.apache.flink.connector.datagen.source.GeneratorSourceReaderFactory.<init>(org.apache.flink.connector.datagen.source.GeneratorFunction, org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (GeneratorSourceReaderFactory.java:54)
Constructor <org.apache.flink.connector.datagen.source.GeneratorSourceReaderFactory.<init>(org.apache.flink.connector.datagen.source.GeneratorFunction, org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (GeneratorSourceReaderFactory.java:55)
Constructor <org.apache.flink.connector.datagen.source.GeneratorSourceReaderFactory.<init>(org.apache.flink.connector.datagen.source.GeneratorFunction, org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy)> has parameter of type <org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy> in (GeneratorSourceReaderFactory.java:0)
-Constructor <org.apache.flink.connector.datagen.table.DataGenTableSource.<init>([Lorg.apache.flink.streaming.api.functions.source.datagen.DataGenerator;, java.lang.String, org.apache.flink.table.types.DataType, long, java.lang.Long)> depends on component type <org.apache.flink.streaming.api.functions.source.datagen.DataGenerator> in (DataGenTableSource.java:0)
-Constructor <org.apache.flink.connector.datagen.table.DataGenTableSource.<init>([Lorg.apache.flink.streaming.api.functions.source.datagen.DataGenerator;, java.lang.String, org.apache.flink.table.types.DataType, long, java.lang.Long)> has parameter of type <[Lorg.apache.flink.streaming.api.functions.source.datagen.DataGenerator;> in (DataGenTableSource.java:0)
+Constructor <org.apache.flink.connector.datagen.table.DataGenTableSource.<init>([Lorg.apache.flink.streaming.api.functions.source.datagen.DataGenerator;, java.lang.String, org.apache.flink.table.types.DataType, long, java.lang.Long, java.lang.Integer)> depends on component type <org.apache.flink.streaming.api.functions.source.datagen.DataGenerator> in (DataGenTableSource.java:0)
+Constructor <org.apache.flink.connector.datagen.table.DataGenTableSource.<init>([Lorg.apache.flink.streaming.api.functions.source.datagen.DataGenerator;, java.lang.String, org.apache.flink.table.types.DataType, long, java.lang.Long, java.lang.Integer)> has parameter of type <[Lorg.apache.flink.streaming.api.functions.source.datagen.DataGenerator;> in (DataGenTableSource.java:0)
Constructor <org.apache.flink.connector.datagen.table.DataGenVisitorBase.<init>(java.lang.String, org.apache.flink.configuration.ReadableConfig)> calls constructor <org.apache.flink.table.types.logical.utils.LogicalTypeDefaultVisitor.<init>()> in (DataGenVisitorBase.java:49)
Constructor <org.apache.flink.connector.datagen.table.DataGeneratorContainer.<init>(org.apache.flink.streaming.api.functions.source.datagen.DataGenerator, java.util.Set)> has parameter of type <org.apache.flink.streaming.api.functions.source.datagen.DataGenerator> in (DataGeneratorContainer.java:0)
Constructor <org.apache.flink.connector.datagen.table.RandomGeneratorVisitor$1.<init>(int)> calls constructor <org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator.<init>()> in (RandomGeneratorVisitor.java:458)
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/table.q b/flink-table/flink-sql-client/src/test/resources/sql/table.q
index f5a23835530..57452546242 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/table.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/table.q
@@ -455,6 +455,7 @@ fields.user.min
fields.user.null-rate
number-of-rows
rows-per-second
+scan.parallelism
!error
# ==========================================================================
diff --git a/flink-table/flink-sql-gateway/src/test/resources/sql/table.q b/flink-table/flink-sql-gateway/src/test/resources/sql/table.q
index f94a6ef379d..597281a6e33 100644
--- a/flink-table/flink-sql-gateway/src/test/resources/sql/table.q
+++ b/flink-table/flink-sql-gateway/src/test/resources/sql/table.q
@@ -410,6 +410,7 @@ fields.user.min
fields.user.null-rate
number-of-rows
rows-per-second
+scan.parallelism
!error
# ==========================================================================
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenConnectorOptions.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenConnectorOptions.java
index aa900754111..ef696b20045 100644
--- a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenConnectorOptions.java
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenConnectorOptions.java
@@ -21,6 +21,7 @@ package org.apache.flink.connector.datagen.table;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.table.factories.FactoryUtil;
import java.time.Duration;
@@ -53,6 +54,8 @@ public class DataGenConnectorOptions {
.withDescription(
"Total number of rows to emit. By default, the source is unbounded.");
+ public static final ConfigOption<Integer> SOURCE_PARALLELISM = FactoryUtil.SOURCE_PARALLELISM;
+
// --------------------------------------------------------------------------------------------
// Placeholder options
// --------------------------------------------------------------------------------------------
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenTableSource.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenTableSource.java
index 9163c2a63f2..93fa9a9a8d8 100644
--- a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenTableSource.java
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenTableSource.java
@@ -32,6 +32,8 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.sources.StreamTableSource;
import org.apache.flink.table.types.DataType;
+import javax.annotation.Nullable;
+
/** A {@link StreamTableSource} that emits generated data rows. */
@Internal
public class DataGenTableSource implements ScanTableSource, SupportsLimitPushDown {
@@ -41,24 +43,27 @@ public class DataGenTableSource implements ScanTableSource, SupportsLimitPushDow
private final DataType rowDataType;
private final long rowsPerSecond;
private Long numberOfRows;
+ private final @Nullable Integer parallelism;
public DataGenTableSource(
DataGenerator<?>[] fieldGenerators,
String tableName,
DataType rowDataType,
long rowsPerSecond,
- Long numberOfRows) {
+ Long numberOfRows,
+ Integer parallelism) {
this.fieldGenerators = fieldGenerators;
this.tableName = tableName;
this.rowDataType = rowDataType;
this.rowsPerSecond = rowsPerSecond;
this.numberOfRows = numberOfRows;
+ this.parallelism = parallelism;
}
@Override
public ScanRuntimeProvider getScanRuntimeProvider(ScanContext context) {
boolean isBounded = numberOfRows != null;
- return SourceFunctionProvider.of(createSource(), isBounded);
+ return SourceFunctionProvider.of(createSource(), isBounded, parallelism);
}
@VisibleForTesting
@@ -72,7 +77,7 @@ public class DataGenTableSource implements ScanTableSource, SupportsLimitPushDow
@Override
public DynamicTableSource copy() {
return new DataGenTableSource(
- fieldGenerators, tableName, rowDataType, rowsPerSecond, numberOfRows);
+ fieldGenerators, tableName, rowDataType, rowsPerSecond, numberOfRows, parallelism);
}
@Override
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenTableSourceFactory.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenTableSourceFactory.java
index 72398c34b96..fc55764fddf 100644
--- a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenTableSourceFactory.java
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenTableSourceFactory.java
@@ -64,6 +64,7 @@ public class DataGenTableSourceFactory implements DynamicTableSourceFactory {
Set<ConfigOption<?>> options = new HashSet<>();
options.add(DataGenConnectorOptions.ROWS_PER_SECOND);
options.add(DataGenConnectorOptions.NUMBER_OF_ROWS);
+ options.add(DataGenConnectorOptions.SOURCE_PARALLELISM);
// Placeholder options
options.add(DataGenConnectorOptions.FIELD_KIND);
@@ -116,6 +117,7 @@ public class DataGenTableSourceFactory implements DynamicTableSourceFactory {
consumedOptionKeys.add(CONNECTOR.key());
consumedOptionKeys.add(DataGenConnectorOptions.ROWS_PER_SECOND.key());
consumedOptionKeys.add(DataGenConnectorOptions.NUMBER_OF_ROWS.key());
+ consumedOptionKeys.add(DataGenConnectorOptions.SOURCE_PARALLELISM.key());
optionalOptions.stream().map(ConfigOption::key).forEach(consumedOptionKeys::add);
FactoryUtil.validateUnconsumedKeys(
factoryIdentifier(), options.keySet(), consumedOptionKeys);
@@ -126,7 +128,8 @@ public class DataGenTableSourceFactory implements DynamicTableSourceFactory {
name,
rowDataType,
options.get(DataGenConnectorOptions.ROWS_PER_SECOND),
- options.get(DataGenConnectorOptions.NUMBER_OF_ROWS));
+ options.get(DataGenConnectorOptions.NUMBER_OF_ROWS),
+ options.getOptional(DataGenConnectorOptions.SOURCE_PARALLELISM).orElse(null));
}
private DataGeneratorContainer createContainer(
diff --git a/flink-table/flink-table-api-java-bridge/src/test/java/org/apache/flink/table/factories/DataGenTableSourceFactoryTest.java b/flink-table/flink-table-api-java-bridge/src/test/java/org/apache/flink/table/factories/DataGenTableSourceFactoryTest.java
index 882ec105f01..687d479839c 100644
--- a/flink-table/flink-table-api-java-bridge/src/test/java/org/apache/flink/table/factories/DataGenTableSourceFactoryTest.java
+++ b/flink-table/flink-table-api-java-bridge/src/test/java/org/apache/flink/table/factories/DataGenTableSourceFactoryTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.factories;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.datagen.table.DataGenConnectorOptions;
import org.apache.flink.connector.datagen.table.DataGenConnectorOptionsUtil;
import org.apache.flink.connector.datagen.table.DataGenTableSource;
@@ -34,9 +35,13 @@ import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.SourceFunctionProvider;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.util.InstantiationUtil;
import org.junit.jupiter.api.Test;
@@ -45,8 +50,10 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
@@ -581,6 +588,27 @@ class DataGenTableSourceFactoryTest {
anyCauseMatches("Could not parse value 'Wrong' for key 'fields.f0.start'"));
}
+ @Test
+ void testWithParallelism() {
+ ResolvedSchema schema = ResolvedSchema.of(Column.physical("f0", DataTypes.CHAR(1)));
+
+ Map<String, String> options = new HashMap<>();
+ options.put(FactoryUtil.CONNECTOR.key(), "datagen");
+ options.put(DataGenConnectorOptions.SOURCE_PARALLELISM.key(), "10");
+
+ DynamicTableSource source = createTableSource(schema, options);
+ assertThat(source).isInstanceOf(DataGenTableSource.class);
+
+ DataGenTableSource dataGenTableSource = (DataGenTableSource) source;
+ ScanTableSource.ScanRuntimeProvider scanRuntimeProvider =
+ dataGenTableSource.getScanRuntimeProvider(new TestScanContext());
+ assertThat(scanRuntimeProvider).isInstanceOf(SourceFunctionProvider.class);
+
+ SourceFunctionProvider sourceFunctionProvider =
+ (SourceFunctionProvider) scanRuntimeProvider;
+ assertThat(sourceFunctionProvider.getParallelism()).hasValue(10);
+ }
+
private void assertException(
ResolvedSchema schema,
DescriptorProperties descriptor,
@@ -649,4 +677,22 @@ class DataGenTableSourceFactoryTest {
@Override
public void close() {}
}
+
+ private static class TestScanContext implements ScanTableSource.ScanContext {
+ @Override
+ public <T> TypeInformation<T> createTypeInformation(DataType producedDataType) {
+ return null;
+ }
+
+ @Override
+ public <T> TypeInformation<T> createTypeInformation(LogicalType producedLogicalType) {
+ return null;
+ }
+
+ @Override
+ public DynamicTableSource.DataStructureConverter createDataStructureConverter(
+ DataType producedDataType) {
+ return null;
+ }
+ }
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/table/DataGeneratorConnectorITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/table/DataGeneratorConnectorITCase.java
index c017d1f6b5d..f62d161affe 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/table/DataGeneratorConnectorITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/table/DataGeneratorConnectorITCase.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.planner.runtime.stream.table;
import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.ExplainDetail;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.planner.runtime.utils.BatchTestBase;
@@ -102,4 +103,28 @@ class DataGeneratorConnectorITCase extends BatchTestBase {
.as("Unexpected number of results")
.hasSize(5);
}
+
+ @Test
+ void testWithParallelism() {
+ final TestingTableEnvironment env =
+ TestingTableEnvironment.create(
+ EnvironmentSettings.newInstance().inStreamingMode().build(),
+ null,
+ TableConfig.getDefault());
+
+ env.executeSql(
+ "CREATE TABLE datagen_t (\n"
+ + " f0 CHAR(1)\n"
+ + ") WITH ("
+ + " 'connector' = 'datagen',"
+ + " 'scan.parallelism' = '2'"
+ + ")");
+
+ final Table table = env.sqlQuery("select * from datagen_t");
+ final String explain = table.explain(ExplainDetail.JSON_EXECUTION_PLAN);
+ final String expectedPhysicalExecutionPlanFragment =
+ "table=[[default_catalog, default_database, datagen_t]], fields=[f0])\",\n"
+ + " \"parallelism\" : 2";
+ assertThat(explain).contains(expectedPhysicalExecutionPlanFragment);
+ }
}