You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sj...@apache.org on 2020/09/14 17:48:35 UTC
[flink] 01/05: [hotfix][table] Refactor DataGen source to use
LogicalTypeVisitor pattern
This is an automated email from the ASF dual-hosted git repository.
sjwiesman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 37e128c637746c9c93e2199f85517a29844cfb63
Author: Seth Wiesman <sj...@gmail.com>
AuthorDate: Mon Jul 27 11:27:42 2020 -0500
[hotfix][table] Refactor DataGen source to use LogicalTypeVisitor pattern
---
.../table/factories/DataGenTableSourceFactory.java | 293 +--------------------
.../factories/datagen/DataGenTableSource.java | 131 +++++++++
.../factories/datagen/DataGeneratorContainer.java | 59 +++++
.../factories/datagen/RandomGeneratorVisitor.java | 166 ++++++++++++
.../datagen/SequenceGeneratorVisitor.java | 175 ++++++++++++
.../factories/DataGenTableSourceFactoryTest.java | 2 +-
6 files changed, 540 insertions(+), 286 deletions(-)
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/factories/DataGenTableSourceFactory.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/factories/DataGenTableSourceFactory.java
index 4a6c3a8..e10fd52 100644
--- a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/factories/DataGenTableSourceFactory.java
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/factories/DataGenTableSourceFactory.java
@@ -19,33 +19,20 @@
package org.apache.flink.table.factories;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.configuration.ConfigOptions.OptionBuilder;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
-import org.apache.flink.runtime.state.FunctionInitializationContext;
-import org.apache.flink.runtime.state.FunctionSnapshotContext;
-import org.apache.flink.streaming.api.functions.source.StatefulSequenceSource;
import org.apache.flink.streaming.api.functions.source.datagen.DataGenerator;
-import org.apache.flink.streaming.api.functions.source.datagen.DataGeneratorSource;
-import org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator;
-import org.apache.flink.streaming.api.functions.source.datagen.SequenceGenerator;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.api.ValidationException;
-import org.apache.flink.table.connector.ChangelogMode;
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.data.StringData;
-import org.apache.flink.table.sources.StreamTableSource;
+import org.apache.flink.table.factories.datagen.DataGenTableSource;
+import org.apache.flink.table.factories.datagen.DataGeneratorContainer;
+import org.apache.flink.table.factories.datagen.RandomGeneratorVisitor;
+import org.apache.flink.table.factories.datagen.SequenceGeneratorVisitor;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.utils.TableSchemaUtils;
-import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
@@ -60,7 +47,6 @@ public class DataGenTableSourceFactory implements DynamicTableSourceFactory {
public static final String IDENTIFIER = "datagen";
public static final Long ROWS_PER_SECOND_DEFAULT_VALUE = 10000L;
- public static final int RANDOM_STRING_LENGTH_DEFAULT = 100;
public static final ConfigOption<Long> ROWS_PER_SECOND = key("rows-per-second")
.longType()
@@ -111,10 +97,10 @@ public class DataGenTableSourceFactory implements DynamicTableSourceFactory {
ConfigOption<String> kind = key(FIELDS + "." + name + "." + KIND)
.stringType().defaultValue(RANDOM);
DataGeneratorContainer container = createContainer(name, type, options.get(kind), options);
- fieldGenerators[i] = container.generator;
+ fieldGenerators[i] = container.getGenerator();
optionalOptions.add(kind);
- optionalOptions.addAll(container.options);
+ optionalOptions.addAll(container.getOptions());
}
FactoryUtil.validateFactoryOptions(requiredOptions(), optionalOptions, options);
@@ -132,274 +118,11 @@ public class DataGenTableSourceFactory implements DynamicTableSourceFactory {
String name, DataType type, String kind, ReadableConfig options) {
switch (kind) {
case RANDOM:
- return createRandomContainer(name, type, options);
+ return type.getLogicalType().accept(new RandomGeneratorVisitor(name, type, options));
case SEQUENCE:
- return createSequenceContainer(name, type, options);
+ return type.getLogicalType().accept(new SequenceGeneratorVisitor(name, type, options));
default:
throw new ValidationException("Unsupported generator kind: " + kind);
}
}
-
- private DataGeneratorContainer createRandomContainer(String name, DataType type, ReadableConfig config) {
- OptionBuilder minKey = key(FIELDS + "." + name + "." + MIN);
- OptionBuilder maxKey = key(FIELDS + "." + name + "." + MAX);
- switch (type.getLogicalType().getTypeRoot()) {
- case BOOLEAN: {
- return DataGeneratorContainer.of(RandomGenerator.booleanGenerator());
- }
- case CHAR:
- case VARCHAR: {
- ConfigOption<Integer> lenOption = key(FIELDS + "." + name + "." + LENGTH)
- .intType()
- .defaultValue(RANDOM_STRING_LENGTH_DEFAULT);
- return DataGeneratorContainer.of(getRandomStringGenerator(config.get(lenOption)), lenOption);
- }
- case TINYINT: {
- ConfigOption<Integer> min = minKey.intType().defaultValue((int) Byte.MIN_VALUE);
- ConfigOption<Integer> max = maxKey.intType().defaultValue((int) Byte.MAX_VALUE);
- return DataGeneratorContainer.of(
- RandomGenerator.byteGenerator(
- config.get(min).byteValue(), config.get(max).byteValue()),
- min, max);
- }
- case SMALLINT: {
- ConfigOption<Integer> min = minKey.intType().defaultValue((int) Short.MIN_VALUE);
- ConfigOption<Integer> max = maxKey.intType().defaultValue((int) Short.MAX_VALUE);
- return DataGeneratorContainer.of(
- RandomGenerator.shortGenerator(
- config.get(min).shortValue(),
- config.get(max).shortValue()),
- min, max);
- }
- case INTEGER: {
- ConfigOption<Integer> min = minKey.intType().defaultValue(Integer.MIN_VALUE);
- ConfigOption<Integer> max = maxKey.intType().defaultValue(Integer.MAX_VALUE);
- return DataGeneratorContainer.of(
- RandomGenerator.intGenerator(
- config.get(min), config.get(max)),
- min, max);
- }
- case BIGINT: {
- ConfigOption<Long> min = minKey.longType().defaultValue(Long.MIN_VALUE);
- ConfigOption<Long> max = maxKey.longType().defaultValue(Long.MAX_VALUE);
- return DataGeneratorContainer.of(
- RandomGenerator.longGenerator(
- config.get(min), config.get(max)),
- min, max);
- }
- case FLOAT: {
- ConfigOption<Float> min = minKey.floatType().defaultValue(Float.MIN_VALUE);
- ConfigOption<Float> max = maxKey.floatType().defaultValue(Float.MAX_VALUE);
- return DataGeneratorContainer.of(
- RandomGenerator.floatGenerator(
- config.get(min), config.get(max)),
- min, max);
- }
- case DOUBLE: {
- ConfigOption<Double> min = minKey.doubleType().defaultValue(Double.MIN_VALUE);
- ConfigOption<Double> max = maxKey.doubleType().defaultValue(Double.MAX_VALUE);
- return DataGeneratorContainer.of(
- RandomGenerator.doubleGenerator(
- config.get(min), config.get(max)),
- min, max);
- }
- default:
- throw new ValidationException("Unsupported type: " + type);
- }
- }
-
- private DataGeneratorContainer createSequenceContainer(String name, DataType type, ReadableConfig config) {
- String startKeyStr = FIELDS + "." + name + "." + START;
- String endKeyStr = FIELDS + "." + name + "." + END;
- OptionBuilder startKey = key(startKeyStr);
- OptionBuilder endKey = key(endKeyStr);
-
- config.getOptional(startKey.stringType().noDefaultValue()).orElseThrow(
- () -> new ValidationException(
- "Could not find required property '" + startKeyStr + "' for sequence generator."));
- config.getOptional(endKey.stringType().noDefaultValue()).orElseThrow(
- () -> new ValidationException(
- "Could not find required property '" + endKeyStr + "' for sequence generator."));
-
- ConfigOption<Integer> intStart = startKey.intType().noDefaultValue();
- ConfigOption<Integer> intEnd = endKey.intType().noDefaultValue();
- ConfigOption<Long> longStart = startKey.longType().noDefaultValue();
- ConfigOption<Long> longEnd = endKey.longType().noDefaultValue();
- switch (type.getLogicalType().getTypeRoot()) {
- case CHAR:
- case VARCHAR:
- return DataGeneratorContainer.of(
- getSequenceStringGenerator(
- config.get(longStart), config.get(longEnd)),
- longStart, longEnd);
- case TINYINT:
- return DataGeneratorContainer.of(
- SequenceGenerator.byteGenerator(
- config.get(intStart).byteValue(),
- config.get(intEnd).byteValue()),
- intStart, intEnd);
- case SMALLINT:
- return DataGeneratorContainer.of(
- SequenceGenerator.shortGenerator(
- config.get(intStart).shortValue(),
- config.get(intEnd).shortValue()),
- intStart, intEnd);
- case INTEGER:
- return DataGeneratorContainer.of(
- SequenceGenerator.intGenerator(
- config.get(intStart), config.get(intEnd)),
- intStart, intEnd);
- case BIGINT:
- return DataGeneratorContainer.of(
- SequenceGenerator.longGenerator(
- config.get(longStart), config.get(longEnd)),
- longStart, longEnd);
- case FLOAT:
- return DataGeneratorContainer.of(
- SequenceGenerator.floatGenerator(
- config.get(intStart).shortValue(),
- config.get(intEnd).shortValue()),
- intStart, intEnd);
- case DOUBLE:
- return DataGeneratorContainer.of(
- SequenceGenerator.doubleGenerator(
- config.get(intStart), config.get(intEnd)),
- intStart, intEnd);
- default:
- throw new ValidationException("Unsupported type: " + type);
- }
- }
-
- private static SequenceGenerator<StringData> getSequenceStringGenerator(long start, long end) {
- return new SequenceGenerator<StringData>(start, end) {
- @Override
- public StringData next() {
- return StringData.fromString(valuesToEmit.poll().toString());
- }
- };
- }
-
- private static RandomGenerator<StringData> getRandomStringGenerator(int length) {
- return new RandomGenerator<StringData>() {
- @Override
- public StringData next() {
- return StringData.fromString(random.nextHexString(length));
- }
- };
- }
-
- // -------------------------------- Help Classes -------------------------------
-
- private static class DataGeneratorContainer {
-
- private DataGenerator generator;
-
- /**
- * Generator config options, for validation.
- */
- private Set<ConfigOption<?>> options;
-
- private DataGeneratorContainer(DataGenerator generator, Set<ConfigOption<?>> options) {
- this.generator = generator;
- this.options = options;
- }
-
- private static DataGeneratorContainer of(
- DataGenerator generator, ConfigOption<?>... options) {
- return new DataGeneratorContainer(generator, new HashSet<>(Arrays.asList(options)));
- }
- }
-
- /**
- * A {@link StreamTableSource} that emits each number from a given interval exactly once,
- * possibly in parallel. See {@link StatefulSequenceSource}.
- */
- static class DataGenTableSource implements ScanTableSource {
-
- private final DataGenerator[] fieldGenerators;
- private final TableSchema schema;
- private final long rowsPerSecond;
-
- private DataGenTableSource(DataGenerator[] fieldGenerators, TableSchema schema, long rowsPerSecond) {
- this.fieldGenerators = fieldGenerators;
- this.schema = schema;
- this.rowsPerSecond = rowsPerSecond;
- }
-
- @Override
- public ScanRuntimeProvider getScanRuntimeProvider(ScanContext context) {
- return SourceFunctionProvider.of(createSource(), false);
- }
-
- @VisibleForTesting
- DataGeneratorSource<RowData> createSource() {
- return new DataGeneratorSource<>(
- new RowGenerator(fieldGenerators, schema.getFieldNames()),
- rowsPerSecond);
- }
-
- @Override
- public DynamicTableSource copy() {
- return new DataGenTableSource(fieldGenerators, schema, rowsPerSecond);
- }
-
- @Override
- public String asSummaryString() {
- return "DataGenTableSource";
- }
-
- @Override
- public ChangelogMode getChangelogMode() {
- return ChangelogMode.insertOnly();
- }
- }
-
- private static class RowGenerator implements DataGenerator<RowData> {
-
- private static final long serialVersionUID = 1L;
-
- private final DataGenerator[] fieldGenerators;
- private final String[] fieldNames;
-
- private RowGenerator(DataGenerator[] fieldGenerators, String[] fieldNames) {
- this.fieldGenerators = fieldGenerators;
- this.fieldNames = fieldNames;
- }
-
- @Override
- public void open(
- String name,
- FunctionInitializationContext context,
- RuntimeContext runtimeContext) throws Exception {
- for (int i = 0; i < fieldGenerators.length; i++) {
- fieldGenerators[i].open(fieldNames[i], context, runtimeContext);
- }
- }
-
- @Override
- public void snapshotState(FunctionSnapshotContext context) throws Exception {
- for (DataGenerator generator : fieldGenerators) {
- generator.snapshotState(context);
- }
- }
-
- @Override
- public boolean hasNext() {
- for (DataGenerator generator : fieldGenerators) {
- if (!generator.hasNext()) {
- return false;
- }
- }
- return true;
- }
-
- @Override
- public RowData next() {
- GenericRowData row = new GenericRowData(fieldNames.length);
- for (int i = 0; i < fieldGenerators.length; i++) {
- row.setField(i, fieldGenerators[i].next());
- }
- return row;
- }
- }
}
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/factories/datagen/DataGenTableSource.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/factories/datagen/DataGenTableSource.java
new file mode 100644
index 0000000..634f275
--- /dev/null
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/factories/datagen/DataGenTableSource.java
@@ -0,0 +1,131 @@
+/*
+ * 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.factories.datagen;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.functions.source.StatefulSequenceSource;
+import org.apache.flink.streaming.api.functions.source.datagen.DataGenerator;
+import org.apache.flink.streaming.api.functions.source.datagen.DataGeneratorSource;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.connector.ChangelogMode;
+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.sources.StreamTableSource;
+
+/**
+ * A {@link StreamTableSource} that emits each number from a given interval exactly once,
+ * possibly in parallel. See {@link StatefulSequenceSource}.
+ */
+@Internal
+public class DataGenTableSource implements ScanTableSource {
+
+ private final DataGenerator[] fieldGenerators;
+ private final TableSchema schema;
+ private final long rowsPerSecond;
+
+ public DataGenTableSource(DataGenerator[] fieldGenerators, TableSchema schema, long rowsPerSecond) {
+ this.fieldGenerators = fieldGenerators;
+ this.schema = schema;
+ this.rowsPerSecond = rowsPerSecond;
+ }
+
+ @Override
+ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext context) {
+ return SourceFunctionProvider.of(createSource(), false);
+ }
+
+ @VisibleForTesting
+ public DataGeneratorSource<RowData> createSource() {
+ return new DataGeneratorSource<>(
+ new RowGenerator(fieldGenerators, schema.getFieldNames()),
+ rowsPerSecond);
+ }
+
+ @Override
+ public DynamicTableSource copy() {
+ return new DataGenTableSource(fieldGenerators, schema, rowsPerSecond);
+ }
+
+ @Override
+ public String asSummaryString() {
+ return "DataGenTableSource";
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.insertOnly();
+ }
+
+ private static class RowGenerator implements DataGenerator<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final DataGenerator[] fieldGenerators;
+ private final String[] fieldNames;
+
+ private RowGenerator(DataGenerator[] fieldGenerators, String[] fieldNames) {
+ this.fieldGenerators = fieldGenerators;
+ this.fieldNames = fieldNames;
+ }
+
+ @Override
+ public void open(
+ String name,
+ FunctionInitializationContext context,
+ RuntimeContext runtimeContext) throws Exception {
+ for (int i = 0; i < fieldGenerators.length; i++) {
+ fieldGenerators[i].open(fieldNames[i], context, runtimeContext);
+ }
+ }
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) throws Exception {
+ for (DataGenerator generator : fieldGenerators) {
+ generator.snapshotState(context);
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ for (DataGenerator generator : fieldGenerators) {
+ if (!generator.hasNext()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public RowData next() {
+ GenericRowData row = new GenericRowData(fieldNames.length);
+ for (int i = 0; i < fieldGenerators.length; i++) {
+ row.setField(i, fieldGenerators[i].next());
+ }
+ return row;
+ }
+ }
+}
+
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/factories/datagen/DataGeneratorContainer.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/factories/datagen/DataGeneratorContainer.java
new file mode 100644
index 0000000..24b2307
--- /dev/null
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/factories/datagen/DataGeneratorContainer.java
@@ -0,0 +1,59 @@
+/*
+ * 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.factories.datagen;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.streaming.api.functions.source.datagen.DataGenerator;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Container class for wrapping a {@link DataGenerator with its configuration options}.
+ */
+@Internal
+public class DataGeneratorContainer {
+
+ private final DataGenerator generator;
+
+ /**
+ * Generator config options, for validation.
+ */
+ private final Set<ConfigOption<?>> options;
+
+ private DataGeneratorContainer(DataGenerator generator, Set<ConfigOption<?>> options) {
+ this.generator = generator;
+ this.options = options;
+ }
+
+ public static DataGeneratorContainer of(
+ DataGenerator generator, ConfigOption<?>... options) {
+ return new DataGeneratorContainer(generator, new HashSet<>(Arrays.asList(options)));
+ }
+
+ public DataGenerator getGenerator() {
+ return generator;
+ }
+
+ public Set<ConfigOption<?>> getOptions() {
+ return options;
+ }
+}
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/factories/datagen/RandomGeneratorVisitor.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/factories/datagen/RandomGeneratorVisitor.java
new file mode 100644
index 0000000..c203114
--- /dev/null
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/factories/datagen/RandomGeneratorVisitor.java
@@ -0,0 +1,166 @@
+/*
+ * 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.factories.datagen;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.BooleanType;
+import org.apache.flink.table.types.logical.CharType;
+import org.apache.flink.table.types.logical.DoubleType;
+import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.SmallIntType;
+import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeDefaultVisitor;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+import static org.apache.flink.table.factories.DataGenTableSourceFactory.*;
+
+/**
+ * Creates a random {@link DataGeneratorContainer} for a particular logical type.
+ */
+@Internal
+public class RandomGeneratorVisitor extends LogicalTypeDefaultVisitor<DataGeneratorContainer> {
+
+ public static final int RANDOM_STRING_LENGTH_DEFAULT = 100;
+
+ private final String name;
+
+ private final DataType type;
+
+ private final ReadableConfig config;
+
+ private final ConfigOptions.OptionBuilder minKey;
+
+ private final ConfigOptions.OptionBuilder maxKey;
+
+ public RandomGeneratorVisitor(String name, DataType type, ReadableConfig config) {
+ this.name = name;
+ this.type = type;
+ this.config = config;
+
+ this.minKey = key(FIELDS + "." + name + "." + MIN);
+ this.maxKey = key(FIELDS + "." + name + "." + MAX);
+ }
+
+ @Override
+ public DataGeneratorContainer visit(BooleanType booleanType) {
+ return DataGeneratorContainer.of(RandomGenerator.booleanGenerator());
+ }
+
+ @Override
+ public DataGeneratorContainer visit(CharType booleanType) {
+ ConfigOption<Integer> lenOption = key(FIELDS + "." + name + "." + LENGTH)
+ .intType()
+ .defaultValue(RANDOM_STRING_LENGTH_DEFAULT);
+ return DataGeneratorContainer.of(getRandomStringGenerator(config.get(lenOption)), lenOption);
+ }
+
+ @Override
+ public DataGeneratorContainer visit(VarCharType booleanType) {
+ ConfigOption<Integer> lenOption = key(FIELDS + "." + name + "." + LENGTH)
+ .intType()
+ .defaultValue(RANDOM_STRING_LENGTH_DEFAULT);
+ return DataGeneratorContainer.of(getRandomStringGenerator(config.get(lenOption)), lenOption);
+ }
+
+ @Override
+ public DataGeneratorContainer visit(TinyIntType booleanType) {
+ ConfigOption<Integer> min = minKey.intType().defaultValue((int) Byte.MIN_VALUE);
+ ConfigOption<Integer> max = maxKey.intType().defaultValue((int) Byte.MAX_VALUE);
+ return DataGeneratorContainer.of(
+ RandomGenerator.byteGenerator(
+ config.get(min).byteValue(), config.get(max).byteValue()),
+ min, max);
+ }
+
+ @Override
+ public DataGeneratorContainer visit(SmallIntType booleanType) {
+ ConfigOption<Integer> min = minKey.intType().defaultValue((int) Short.MIN_VALUE);
+ ConfigOption<Integer> max = maxKey.intType().defaultValue((int) Short.MAX_VALUE);
+ return DataGeneratorContainer.of(
+ RandomGenerator.shortGenerator(
+ config.get(min).shortValue(),
+ config.get(max).shortValue()),
+ min, max);
+ }
+
+ @Override
+ public DataGeneratorContainer visit(IntType integerType) {
+ ConfigOption<Integer> min = minKey.intType().defaultValue(Integer.MIN_VALUE);
+ ConfigOption<Integer> max = maxKey.intType().defaultValue(Integer.MAX_VALUE);
+ return DataGeneratorContainer.of(
+ RandomGenerator.intGenerator(
+ config.get(min), config.get(max)),
+ min, max);
+ }
+
+ @Override
+ public DataGeneratorContainer visit(BigIntType bigIntType) {
+ ConfigOption<Long> min = minKey.longType().defaultValue(Long.MIN_VALUE);
+ ConfigOption<Long> max = maxKey.longType().defaultValue(Long.MAX_VALUE);
+ return DataGeneratorContainer.of(
+ RandomGenerator.longGenerator(
+ config.get(min), config.get(max)),
+ min, max);
+ }
+
+ @Override
+ public DataGeneratorContainer visit(FloatType floatType) {
+ ConfigOption<Float> min = minKey.floatType().defaultValue(Float.MIN_VALUE);
+ ConfigOption<Float> max = maxKey.floatType().defaultValue(Float.MAX_VALUE);
+ return DataGeneratorContainer.of(
+ RandomGenerator.floatGenerator(
+ config.get(min), config.get(max)),
+ min, max);
+ }
+
+ @Override
+ public DataGeneratorContainer visit(DoubleType doubleType) {
+ ConfigOption<Double> min = minKey.doubleType().defaultValue(Double.MIN_VALUE);
+ ConfigOption<Double> max = maxKey.doubleType().defaultValue(Double.MAX_VALUE);
+ return DataGeneratorContainer.of(
+ RandomGenerator.doubleGenerator(
+ config.get(min), config.get(max)),
+ min, max);
+ }
+
+ @Override
+ protected DataGeneratorContainer defaultMethod(LogicalType logicalType) {
+ throw new ValidationException("Unsupported type: " + type);
+ }
+
+ private static RandomGenerator<StringData> getRandomStringGenerator(int length) {
+ return new RandomGenerator<StringData>() {
+ @Override
+ public StringData next() {
+ return StringData.fromString(random.nextHexString(length));
+ }
+ };
+ }
+}
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/factories/datagen/SequenceGeneratorVisitor.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/factories/datagen/SequenceGeneratorVisitor.java
new file mode 100644
index 0000000..2fa00d7
--- /dev/null
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/factories/datagen/SequenceGeneratorVisitor.java
@@ -0,0 +1,175 @@
+/*
+ * 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.factories.datagen;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator;
+import org.apache.flink.streaming.api.functions.source.datagen.SequenceGenerator;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.BooleanType;
+import org.apache.flink.table.types.logical.CharType;
+import org.apache.flink.table.types.logical.DoubleType;
+import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.SmallIntType;
+import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeDefaultVisitor;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+import static org.apache.flink.table.factories.DataGenTableSourceFactory.*;
+
+/**
+ * Creates a sequential {@link DataGeneratorContainer} for a particular logical type.
+ */
+@Internal
+public class SequenceGeneratorVisitor extends LogicalTypeDefaultVisitor<DataGeneratorContainer> {
+
+ private final DataType type;
+
+ private final ReadableConfig config;
+
+ private final String startKeyStr;
+
+ private final String endKeyStr;
+
+ private final ConfigOption<Integer> intStart;
+
+ private final ConfigOption<Integer> intEnd;
+
+ private final ConfigOption<Long> longStart;
+
+ private final ConfigOption<Long> longEnd;
+
+ public SequenceGeneratorVisitor(String name, DataType type, ReadableConfig config) {
+ this.type = type;
+ this.config = config;
+
+ this.startKeyStr = FIELDS + "." + name + "." + START;
+ this.endKeyStr = FIELDS + "." + name + "." + END;
+
+ ConfigOptions.OptionBuilder startKey = key(startKeyStr);
+ ConfigOptions.OptionBuilder endKey = key(endKeyStr);
+
+ config.getOptional(startKey.stringType().noDefaultValue()).orElseThrow(
+ () -> new ValidationException(
+ "Could not find required property '" + startKeyStr + "' for sequence generator."));
+ config.getOptional(endKey.stringType().noDefaultValue()).orElseThrow(
+ () -> new ValidationException(
+ "Could not find required property '" + endKeyStr + "' for sequence generator."));
+
+ this.intStart = startKey.intType().noDefaultValue();
+ this.intEnd = endKey.intType().noDefaultValue();
+ this.longStart = startKey.longType().noDefaultValue();
+ this.longEnd = endKey.longType().noDefaultValue();
+ }
+
+ @Override
+ public DataGeneratorContainer visit(BooleanType booleanType) {
+ return DataGeneratorContainer.of(RandomGenerator.booleanGenerator());
+ }
+
+ @Override
+ public DataGeneratorContainer visit(CharType booleanType) {
+ return DataGeneratorContainer.of(
+ getSequenceStringGenerator(
+ config.get(longStart), config.get(longEnd)),
+ longStart, longEnd);
+ }
+
+ @Override
+ public DataGeneratorContainer visit(VarCharType booleanType) {
+ return DataGeneratorContainer.of(
+ getSequenceStringGenerator(
+ config.get(longStart), config.get(longEnd)),
+ longStart, longEnd);
+ }
+
+ @Override
+ public DataGeneratorContainer visit(TinyIntType booleanType) {
+ return DataGeneratorContainer.of(
+ SequenceGenerator.byteGenerator(
+ config.get(intStart).byteValue(),
+ config.get(intEnd).byteValue()),
+ intStart, intEnd);
+ }
+
+ @Override
+ public DataGeneratorContainer visit(SmallIntType booleanType) {
+ return DataGeneratorContainer.of(
+ SequenceGenerator.shortGenerator(
+ config.get(intStart).shortValue(),
+ config.get(intEnd).shortValue()),
+ intStart, intEnd);
+ }
+
+ @Override
+ public DataGeneratorContainer visit(IntType integerType) {
+ return DataGeneratorContainer.of(
+ SequenceGenerator.intGenerator(
+ config.get(intStart), config.get(intEnd)),
+ intStart, intEnd);
+ }
+
+ @Override
+ public DataGeneratorContainer visit(BigIntType bigIntType) {
+ return DataGeneratorContainer.of(
+ SequenceGenerator.longGenerator(
+ config.get(longStart), config.get(longEnd)),
+ longStart, longEnd);
+ }
+
+ @Override
+ public DataGeneratorContainer visit(FloatType floatType) {
+ return DataGeneratorContainer.of(
+ SequenceGenerator.floatGenerator(
+ config.get(intStart).shortValue(),
+ config.get(intEnd).shortValue()),
+ intStart, intEnd);
+ }
+
+ @Override
+ public DataGeneratorContainer visit(DoubleType doubleType) {
+ return DataGeneratorContainer.of(
+ SequenceGenerator.doubleGenerator(
+ config.get(intStart), config.get(intEnd)),
+ intStart, intEnd);
+ }
+
+ @Override
+ protected DataGeneratorContainer defaultMethod(LogicalType logicalType) {
+ throw new ValidationException("Unsupported type: " + type);
+ }
+
+ private static SequenceGenerator<StringData> getSequenceStringGenerator(long start, long end) {
+ return new SequenceGenerator<StringData>(start, end) {
+ @Override
+ public StringData next() {
+ return StringData.fromString(valuesToEmit.poll().toString());
+ }
+ };
+ }
+}
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 23e96f5..c5a547e 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
@@ -34,7 +34,7 @@ import org.apache.flink.table.connector.source.DynamicTableSource;
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.factories.DataGenTableSourceFactory.DataGenTableSource;
+import org.apache.flink.table.factories.datagen.DataGenTableSource;
import org.apache.flink.util.InstantiationUtil;
import org.junit.Assert;