[FLINK-18735][table] Add support for more types to DataGen source #13010

sjwiesman · 2020-07-28T17:09:43Z

What is the purpose of the change

Updates the DataGen source to support most types. This enables using DataGen in conjunction with Flink's LIKE clause as it currently does not support overwriting physical columns with computed columns.

Brief change log

I refactored the table factory and added support for all types except RAW and STRUCTURED. See commit messages for a full list of changes. I also added support for creating bounded tables which is useful when prototyping a query.

Verifying this change

New unit tests

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

…tern

flinkbot · 2020-07-28T17:12:40Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 60b2d39 (Tue Jul 28 17:12:39 UTC 2020)

Warnings:

Documentation files were touched, but no .zh.md files: Update Chinese documentation or file Jira ticket.
Invalid pull request title: No valid Jira ID provided

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

Details

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2020-07-28T17:31:53Z

CI report:

cbe7338 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

JingsongLi

Thanks @sjwiesman , sorry for late response.
Now data generator source can be a bounded source, you can implement a ITCase for it, like BlackHoleConnectorITCase.

JingsongLi · 2020-08-11T03:31:22Z

...c/main/java/org/apache/flink/streaming/api/functions/source/datagen/DataGeneratorSource.java

+				if (isRunning && generator.hasNext() && (numberOfRows == null || outputSoFar < toOutput)) {
 					synchronized (ctx.getCheckpointLock()) {
+						if (numberOfRows != null) {
+							outputSoFar++;


I have an idea: print outputSoFar in close. It is good for debugging.

JingsongLi · 2020-08-11T03:31:30Z

...c/main/java/org/apache/flink/streaming/api/functions/source/datagen/DataGeneratorSource.java

-				if (isRunning && generator.hasNext()) {
+				if (isRunning && generator.hasNext() && (numberOfRows == null || outputSoFar < toOutput)) {
 					synchronized (ctx.getCheckpointLock()) {
+						if (numberOfRows != null) {


Remove this if?

JingsongLi · 2020-08-11T03:47:18Z

...va-bridge/src/main/java/org/apache/flink/table/factories/datagen/RandomGeneratorVisitor.java

+		Set<ConfigOption<?>> options = keyContainer.getOptions();
+		options.addAll(valContainer.getOptions());
+
+		return DataGeneratorContainer.of(


Take a look to the comments of RowData. The data strutures should be:

* +--------------------------------+-----------------------------------------+ * | Row | {@link RowData} | * +--------------------------------+-----------------------------------------+ * | ARRAY | {@link ArrayData} | * +--------------------------------+-----------------------------------------+ * | MAP / MULTISET | {@link MapData} | * +--------------------------------+-----------------------------------------+

JingsongLi · 2020-08-11T03:49:37Z

docs/dev/table/connectors/datagen.md


- Random generator is the default generator, you can specify random max and min values. For char/varchar/string, the length can be specified. It is a unbounded generator.
- Sequence generator, you can specify sequence start and end values. It is a bounded generator, when the sequence number reaches the end value, the reading ends.
+Time types are always the local machines current system time.


Maybe we can have a table to show all types.
Display the generation strategies they support, and the required parameters?

sjwiesman · 2020-08-11T21:21:25Z

@JingsongLi thanks for taking a look. I've updated my PR accordingly.

sjwiesman · 2020-09-03T14:00:08Z

@JingsongLi what do you think of the status of this PR?

JingsongLi

I am very sorry for the late response. Left some comments. Feel free to ping me...

JingsongLi · 2020-09-07T03:19:35Z

...i-java-bridge/src/main/java/org/apache/flink/table/factories/datagen/DataGenVisitorBase.java

+		throw new ValidationException("Unsupported type: " + logicalType);
+	}
+
+	private interface SerializableSupplier<T> extends Supplier<T>, Serializable { }


I think lambda is already Serializable.

Its not, the source gets caught in the closure cleaner without this.

I meant we could do something like return (Supplier<T> & Serializable) () -> {...}, but it is OK for now.

JingsongLi · 2020-09-07T03:19:52Z

...c/main/java/org/apache/flink/streaming/api/functions/source/datagen/DataGeneratorSource.java

-	public DataGeneratorSource(DataGenerator<T> generator, long rowsPerSecond) {
+	public DataGeneratorSource(DataGenerator<T> generator, String name, long rowsPerSecond, Long numberOfRows) {
 		this.generator = generator;
+		this.name = name;


Do we need to add name?

that's a remnant from an old commit, will remove.

JingsongLi · 2020-09-07T03:24:07Z

...bridge/src/main/java/org/apache/flink/table/factories/datagen/types/DataGeneratorMapper.java

+		return mapper.apply(generator.next());
+	}
+
+	public interface SerializableFunction<A, B> extends Function<A, B>, Serializable {}


same, we need this.

JingsongLi · 2020-09-07T03:29:43Z

...va-bridge/src/main/java/org/apache/flink/table/factories/datagen/types/RowDataGenerator.java

+
+	private static final long serialVersionUID = 1L;
+
+	private final DataGenerator[] fieldGenerators;


Can we add ? to DataGenerator like DataGenerator<?>[]? At least, The compiler will have no warning.

JingsongLi · 2020-09-07T03:41:14Z

...t/java/org/apache/flink/table/planner/runtime/stream/table/DataGeneratorConnectorITCase.java

+
+import org.junit.Test;
+
+public class DataGeneratorConnectorITCase extends StreamingTestBase {


You can use BatchTestBase.

JingsongLi · 2020-09-07T03:42:06Z

...i-java-bridge/src/main/java/org/apache/flink/table/factories/datagen/DataGenTableSource.java

+
+	@Override
+	public ScanRuntimeProvider getScanRuntimeProvider(ScanContext context) {
+		boolean isBounded = numberOfRows == null;


Looks like the opposite?

JingsongLi · 2020-09-07T03:42:45Z

...t/java/org/apache/flink/table/planner/runtime/stream/table/DataGeneratorConnectorITCase.java

+	public void testTypes() {
+		tEnv().executeSql(TABLE);
+		tEnv().executeSql(SINK);
+		tEnv().from("datagen_t").executeInsert("sink");


You should use: Lists.newArrayList(tEnv().executeSql("select * from datagen_t").collect()).
In this way, the types can be verified.

JingsongLi · 2020-09-07T03:43:47Z

...-bridge/src/main/java/org/apache/flink/table/factories/datagen/SequenceGeneratorVisitor.java

+	@Override
+	public DataGeneratorContainer visit(DecimalType decimalType) {
+		return DataGeneratorContainer.of(
+			SequenceGenerator.bigDecimalGenerator(


Looks like the structures of Decimal, Times are wrong, you can take a look to the comments of RowData.

sjwiesman · 2020-09-09T17:39:33Z

@JingsongLi thank you for the thorough. Review. I've corrected the types and updated the test to validate them. If you don't have any other comments I'll merge on green.

JingsongLi

Thanks for update, Looks good to me except time type generator, and left some minor comments.

JingsongLi · 2020-09-10T03:56:27Z

...c/main/java/org/apache/flink/streaming/api/functions/source/datagen/DataGeneratorSource.java

 	private final long rowsPerSecond;

+	@Nullable
+	private Long numberOfRows;


JingsongLi · 2020-09-10T03:57:01Z

...c/main/java/org/apache/flink/streaming/api/functions/source/datagen/DataGeneratorSource.java

+	@Nullable
+	private Long numberOfRows;
+
+	private int outputSoFar;


JingsongLi · 2020-09-10T03:57:04Z

...c/main/java/org/apache/flink/streaming/api/functions/source/datagen/DataGeneratorSource.java

+
+	private int outputSoFar;
+
+	private int toOutput;


JingsongLi · 2020-09-10T03:59:27Z

...i-java-bridge/src/main/java/org/apache/flink/table/factories/datagen/DataGenVisitorBase.java

+
+	@Override
+	public DataGeneratorContainer visit(TimeType timeType) {
+		return DataGeneratorContainer.of(TimeGenerator.of(() -> LocalTime.now().get(MILLI_OF_SECOND)));


Shouldn't this be MILLI_OF_DAY?

JingsongLi · 2020-09-10T04:00:36Z

...va-bridge/src/main/java/org/apache/flink/table/factories/datagen/RandomGeneratorVisitor.java

+		};
+	}
+
+	private static class BigDecimalRandomGenerator implements DataGenerator<DecimalData> {


I think it can be named DecimalDataRandomGenerator
BigDecimal makes confused

sjwiesman · 2020-09-10T20:35:04Z

Sorry for the long back and forth, just pushed a fix addressing your comments.

sjwiesman added 2 commits July 27, 2020 11:27

[hotfix][table] Refactor DataGen source to use LogicalTypeVisitor pat…

c7227e2

…tern

[FLINK-18735][table] Support bounded datagen tables

3a9518c

sjwiesman requested review from JingsongLi, fhueske and knaufk July 28, 2020 17:09

rmetzger added the review=description? label Jul 28, 2020

sjwiesman changed the title ~~Datagen~~ [FLINK-18735][table] Add support for more types to DataGen source Jul 28, 2020

rmetzger added the component=TableSQL/Ecosystem label Jul 28, 2020

sjwiesman added 2 commits July 28, 2020 14:27

[FLINK-18735][table] Add support for more types to DataGen source

9155299

[FLINK-18735][table] Update documentation of DataGen source

920c6c0

sjwiesman force-pushed the datagen branch from 60b2d39 to 920c6c0 Compare July 28, 2020 19:27

JingsongLi requested changes Aug 11, 2020

View reviewed changes

JingsongLi reviewed Aug 11, 2020

View reviewed changes

sjwiesman added 3 commits August 11, 2020 08:40

[FLINK-18735][table] Add additional logging to DataGen source

96e1f28

fixup! [FLINK-18735][table] Add support for more types to DataGen source

846de4a

fixup! [FLINK-18735][table] Update documentation of DataGen source

125b5ad

JingsongLi requested changes Sep 7, 2020

View reviewed changes

fixup! [FLINK-18735][table] Add support for more types to DataGen source

a969482

JingsongLi approved these changes Sep 10, 2020

View reviewed changes

fixup! [FLINK-18735][table] Add support for more types to DataGen source

cbe7338

sjwiesman closed this in ef01fec Sep 14, 2020

snuyanzin mentioned this pull request Dec 9, 2024

[FLINK-18704][table] Add DECIMAL type in datagen #13154

Closed


		private static final long serialVersionUID = 1L;

		private final DataGenerator[] fieldGenerators;


		import org.junit.Test;

		public class DataGeneratorConnectorITCase extends StreamingTestBase {

[FLINK-18735][table] Add support for more types to DataGen source #13010

[FLINK-18735][table] Add support for more types to DataGen source #13010

Uh oh!

Conversation

sjwiesman commented Jul 28, 2020

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Jul 28, 2020

Automated Checks

Review Progress

Uh oh!

flinkbot commented Jul 28, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

JingsongLi left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sjwiesman commented Aug 11, 2020

Uh oh!

sjwiesman commented Sep 3, 2020

Uh oh!

JingsongLi left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JingsongLi Sep 10, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sjwiesman commented Sep 9, 2020

Uh oh!

JingsongLi left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

flinkbot commented Jul 28, 2020 •

edited

Loading

JingsongLi Sep 10, 2020 •

edited

Loading