You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2020/05/15 05:48:14 UTC
[flink] branch master updated: [FLINK-17606][table] Introduce
DataGenerator connector in table
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 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 91557c8 [FLINK-17606][table] Introduce DataGenerator connector in table
91557c8 is described below
commit 91557c83447444ecfbeba3e4c58297908940834a
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Fri May 15 13:47:47 2020 +0800
[FLINK-17606][table] Introduce DataGenerator connector in table
This closes #12074
---
.../functions/source/datagen/DataGenerator.java | 53 ++++++
.../source/datagen/DataGeneratorSource.java | 102 ++++++++++
.../functions/source/datagen/RandomGenerator.java | 119 ++++++++++++
.../source/datagen/SequenceGenerator.java | 177 +++++++++++++++++
.../api/functions/StatefulSequenceSourceTest.java | 5 +-
.../datagen/DataGeneratorSourceTest.java} | 212 +++++++++------------
flink-table/flink-table-api-java-bridge/pom.xml | 13 ++
.../table/sources/datagen/DataGenTableSource.java | 108 +++++++++++
.../sources/datagen/DataGenTableSourceFactory.java | 208 ++++++++++++++++++++
.../org.apache.flink.table.factories.Factory | 16 ++
.../factories/DataGenTableSourceFactoryTest.java | 142 ++++++++++++++
11 files changed, 1036 insertions(+), 119 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/datagen/DataGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/datagen/DataGenerator.java
new file mode 100644
index 0000000..155e581
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/datagen/DataGenerator.java
@@ -0,0 +1,53 @@
+/*
+ * 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.streaming.api.functions.source.datagen;
+
+import org.apache.flink.annotation.Experimental;
+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.checkpoint.CheckpointedFunction;
+
+import java.io.Serializable;
+import java.util.Iterator;
+
+/**
+ * Stateful and re-scalable data generator.
+ */
+@Experimental
+public interface DataGenerator<T> extends Serializable, Iterator<T> {
+
+ /**
+ * Open and initialize state for {@link DataGenerator}.
+ * See {@link CheckpointedFunction#initializeState}.
+ *
+ * @param name The state of {@link DataGenerator} should related to this name, make sure
+ * the name of state is different.
+ */
+ void open(
+ String name,
+ FunctionInitializationContext context,
+ RuntimeContext runtimeContext) throws Exception;
+
+ /**
+ * Snapshot state for {@link DataGenerator}.
+ * See {@link CheckpointedFunction#snapshotState}.
+ */
+ default void snapshotState(FunctionSnapshotContext context) throws Exception {}
+}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/datagen/DataGeneratorSource.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/datagen/DataGeneratorSource.java
new file mode 100644
index 0000000..6760260
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/datagen/DataGeneratorSource.java
@@ -0,0 +1,102 @@
+/*
+ * 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.streaming.api.functions.source.datagen;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
+
+/**
+ * A data generator source that abstract data generator. It can be used to easy startup/test
+ * for streaming job and performance testing.
+ * It is stateful, re-scalable, possibly in parallel.
+ */
+@Experimental
+public class DataGeneratorSource<T> extends RichParallelSourceFunction<T> implements CheckpointedFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private final DataGenerator<T> generator;
+ private final long rowsPerSecond;
+
+ transient volatile boolean isRunning;
+
+ /**
+ * Creates a source that emits records by {@link DataGenerator} without controlling emit rate.
+ *
+ * @param generator data generator.
+ */
+ public DataGeneratorSource(DataGenerator<T> generator) {
+ this(generator, Long.MAX_VALUE);
+ }
+
+ /**
+ * Creates a source that emits records by {@link DataGenerator}.
+ *
+ * @param generator data generator.
+ * @param rowsPerSecond Control the emit rate.
+ */
+ public DataGeneratorSource(DataGenerator<T> generator, long rowsPerSecond) {
+ this.generator = generator;
+ this.rowsPerSecond = rowsPerSecond;
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ this.generator.open("DataGenerator", context, getRuntimeContext());
+ this.isRunning = true;
+ }
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) throws Exception {
+ this.generator.snapshotState(context);
+ }
+
+ @Override
+ public void run(SourceContext<T> ctx) throws Exception {
+ double taskRowsPerSecond = (double) rowsPerSecond / getRuntimeContext().getNumberOfParallelSubtasks();
+ long nextReadTime = System.currentTimeMillis();
+
+ while (isRunning) {
+ for (int i = 0; i < taskRowsPerSecond; i++) {
+ if (isRunning && generator.hasNext()) {
+ synchronized (ctx.getCheckpointLock()) {
+ ctx.collect(this.generator.next());
+ }
+ } else {
+ return;
+ }
+ }
+
+ nextReadTime += 1000;
+ long toWaitMs = nextReadTime - System.currentTimeMillis();
+ while (toWaitMs > 0) {
+ Thread.sleep(toWaitMs);
+ toWaitMs = nextReadTime - System.currentTimeMillis();
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ isRunning = false;
+ }
+}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/datagen/RandomGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/datagen/RandomGenerator.java
new file mode 100644
index 0000000..6bc795f
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/datagen/RandomGenerator.java
@@ -0,0 +1,119 @@
+/*
+ * 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.streaming.api.functions.source.datagen;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+
+import org.apache.commons.math3.random.RandomDataGenerator;
+
+/**
+ * Random generator.
+ */
+@Experimental
+public abstract class RandomGenerator<T> implements DataGenerator<T> {
+
+ protected transient RandomDataGenerator random;
+
+ @Override
+ public void open(
+ String name,
+ FunctionInitializationContext context,
+ RuntimeContext runtimeContext) throws Exception {
+ this.random = new RandomDataGenerator();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return true;
+ }
+
+ public static RandomGenerator<Long> longGenerator(long min, long max) {
+ return new RandomGenerator<Long>() {
+ @Override
+ public Long next() {
+ return random.nextLong(min, max);
+ }
+ };
+ }
+
+ public static RandomGenerator<Integer> intGenerator(int min, int max) {
+ return new RandomGenerator<Integer>() {
+ @Override
+ public Integer next() {
+ return random.nextInt(min, max);
+ }
+ };
+ }
+
+ public static RandomGenerator<Short> shortGenerator(short min, short max) {
+ return new RandomGenerator<Short>() {
+ @Override
+ public Short next() {
+ return (short) random.nextInt(min, max);
+ }
+ };
+ }
+
+ public static RandomGenerator<Byte> byteGenerator(byte min, byte max) {
+ return new RandomGenerator<Byte>() {
+ @Override
+ public Byte next() {
+ return (byte) random.nextInt(min, max);
+ }
+ };
+ }
+
+ public static RandomGenerator<Float> floatGenerator(float min, float max) {
+ return new RandomGenerator<Float>() {
+ @Override
+ public Float next() {
+ return (float) random.nextUniform(min, max);
+ }
+ };
+ }
+
+ public static RandomGenerator<Double> doubleGenerator(double min, double max) {
+ return new RandomGenerator<Double>() {
+ @Override
+ public Double next() {
+ return random.nextUniform(min, max);
+ }
+ };
+ }
+
+ public static RandomGenerator<String> stringGenerator(int len) {
+ return new RandomGenerator<String>() {
+ @Override
+ public String next() {
+ return random.nextHexString(len);
+ }
+ };
+ }
+
+ public static RandomGenerator<Boolean> booleanGenerator() {
+ return new RandomGenerator<Boolean>() {
+ @Override
+ public Boolean next() {
+ return random.nextInt(0, 1) == 0;
+ }
+ };
+ }
+}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/datagen/SequenceGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/datagen/SequenceGenerator.java
new file mode 100644
index 0000000..855937e
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/datagen/SequenceGenerator.java
@@ -0,0 +1,177 @@
+/*
+ * 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.streaming.api.functions.source.datagen;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.util.Preconditions;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+
+/**
+ * A stateful, re-scalable {@link DataGenerator} that emits each number from a given interval
+ * exactly once, possibly in parallel.
+ */
+@Experimental
+public abstract class SequenceGenerator<T> implements DataGenerator<T> {
+
+ private final long start;
+ private final long end;
+
+ private transient ListState<Long> checkpointedState;
+ protected transient Deque<Long> valuesToEmit;
+
+ /**
+ * Creates a DataGenerator that emits all numbers from the given interval exactly once.
+ *
+ * @param start Start of the range of numbers to emit.
+ * @param end End of the range of numbers to emit.
+ */
+ public SequenceGenerator(long start, long end) {
+ this.start = start;
+ this.end = end;
+ }
+
+ @Override
+ public void open(
+ String name,
+ FunctionInitializationContext context,
+ RuntimeContext runtimeContext) throws Exception {
+ Preconditions.checkState(this.checkpointedState == null,
+ "The " + getClass().getSimpleName() + " has already been initialized.");
+
+ this.checkpointedState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ name + "-sequence-state",
+ LongSerializer.INSTANCE));
+ this.valuesToEmit = new ArrayDeque<>();
+ if (context.isRestored()) {
+ // upon restoring
+
+ for (Long v : this.checkpointedState.get()) {
+ this.valuesToEmit.add(v);
+ }
+ } else {
+ // the first time the job is executed
+ final int stepSize = runtimeContext.getNumberOfParallelSubtasks();
+ final int taskIdx = runtimeContext.getIndexOfThisSubtask();
+ final long congruence = start + taskIdx;
+
+ long totalNoOfElements = Math.abs(end - start + 1);
+ final int baseSize = safeDivide(totalNoOfElements, stepSize);
+ final int toCollect = (totalNoOfElements % stepSize > taskIdx) ? baseSize + 1 : baseSize;
+
+ for (long collected = 0; collected < toCollect; collected++) {
+ this.valuesToEmit.add(collected * stepSize + congruence);
+ }
+ }
+ }
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) throws Exception {
+ Preconditions.checkState(this.checkpointedState != null,
+ "The " + getClass().getSimpleName() + " state has not been properly initialized.");
+
+ this.checkpointedState.clear();
+ for (Long v : this.valuesToEmit) {
+ this.checkpointedState.add(v);
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ return !this.valuesToEmit.isEmpty();
+ }
+
+ private static int safeDivide(long left, long right) {
+ Preconditions.checkArgument(right > 0);
+ Preconditions.checkArgument(left >= 0);
+ Preconditions.checkArgument(left <= Integer.MAX_VALUE * right);
+ return (int) (left / right);
+ }
+
+ public static SequenceGenerator<Long> longGenerator(long start, long end) {
+ return new SequenceGenerator<Long>(start, end) {
+ @Override
+ public Long next() {
+ return valuesToEmit.poll();
+ }
+ };
+ }
+
+ public static SequenceGenerator<Integer> intGenerator(int start, int end) {
+ return new SequenceGenerator<Integer>(start, end) {
+ @Override
+ public Integer next() {
+ return valuesToEmit.poll().intValue();
+ }
+ };
+ }
+
+ public static SequenceGenerator<Short> shortGenerator(short start, short end) {
+ return new SequenceGenerator<Short>(start, end) {
+ @Override
+ public Short next() {
+ return valuesToEmit.poll().shortValue();
+ }
+ };
+ }
+
+ public static SequenceGenerator<Byte> byteGenerator(byte start, byte end) {
+ return new SequenceGenerator<Byte>(start, end) {
+ @Override
+ public Byte next() {
+ return valuesToEmit.poll().byteValue();
+ }
+ };
+ }
+
+ public static SequenceGenerator<Float> floatGenerator(short start, short end) {
+ return new SequenceGenerator<Float>(start, end) {
+ @Override
+ public Float next() {
+ return valuesToEmit.poll().floatValue();
+ }
+ };
+ }
+
+ public static SequenceGenerator<Double> doubleGenerator(int start, int end) {
+ return new SequenceGenerator<Double>(start, end) {
+ @Override
+ public Double next() {
+ return valuesToEmit.poll().doubleValue();
+ }
+ };
+ }
+
+ public static SequenceGenerator<String> stringGenerator(long start, long end) {
+ return new SequenceGenerator<String>(start, end) {
+ @Override
+ public String next() {
+ return valuesToEmit.poll().toString();
+ }
+ };
+ }
+}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/StatefulSequenceSourceTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/StatefulSequenceSourceTest.java
index 03c03f2..a9d5443 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/StatefulSequenceSourceTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/StatefulSequenceSourceTest.java
@@ -183,7 +183,10 @@ public class StatefulSequenceSourceTest {
runner2.join();
}
- private static class BlockingSourceContext implements SourceFunction.SourceContext<Long> {
+ /**
+ * Test SourceContext.
+ */
+ public static class BlockingSourceContext implements SourceFunction.SourceContext<Long> {
private final String name;
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/StatefulSequenceSourceTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/source/datagen/DataGeneratorSourceTest.java
similarity index 55%
copy from flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/StatefulSequenceSourceTest.java
copy to flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/source/datagen/DataGeneratorSourceTest.java
index 03c03f2..a59bb4a 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/StatefulSequenceSourceTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/source/datagen/DataGeneratorSourceTest.java
@@ -16,12 +16,12 @@
* limitations under the License.
*/
-package org.apache.flink.streaming.api.functions;
+package org.apache.flink.streaming.api.functions.source.datagen;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.streaming.api.functions.StatefulSequenceSourceTest.BlockingSourceContext;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
-import org.apache.flink.streaming.api.functions.source.StatefulSequenceSource;
import org.apache.flink.streaming.api.operators.StreamSource;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.util.AbstractStreamOperatorTestHarness;
@@ -37,12 +37,66 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
/**
- * Tests for {@link StatefulSequenceSource}.
+ * Tests for {@link DataGeneratorSource}.
*/
-public class StatefulSequenceSourceTest {
+public class DataGeneratorSourceTest {
@Test
- public void testCheckpointRestore() throws Exception {
+ public void testRandomGenerator() throws Exception {
+ long min = 10;
+ long max = 20;
+ DataGeneratorSource<Long> source = new DataGeneratorSource<>(
+ RandomGenerator.longGenerator(min, max));
+ StreamSource<Long, DataGeneratorSource<Long>> src = new StreamSource<>(source);
+ AbstractStreamOperatorTestHarness<Long> testHarness =
+ new AbstractStreamOperatorTestHarness<>(src, 1, 1, 0);
+ testHarness.open();
+
+ int totalNumber = 1000;
+ List<Long> results = new ArrayList<>();
+
+ source.run(new SourceFunction.SourceContext<Long>() {
+
+ private Object lock = new Object();
+ private int emitNumber = 0;
+
+ @Override
+ public void collect(Long element) {
+ if (++emitNumber > totalNumber) {
+ source.isRunning = false;
+ }
+ results.add(element);
+ }
+
+ @Override
+ public void collectWithTimestamp(Long element, long timestamp) {
+ }
+
+ @Override
+ public void emitWatermark(Watermark mark) {
+ }
+
+ @Override
+ public void markAsTemporarilyIdle() {
+ }
+
+ @Override
+ public Object getCheckpointLock() {
+ return lock;
+ }
+
+ @Override
+ public void close() {
+ }
+ });
+
+ for (Long l : results) {
+ Assert.assertTrue(l >= min && l <= max);
+ }
+ }
+
+ @Test
+ public void testSequenceCheckpointRestore() throws Exception {
final int initElement = 0;
final int maxElement = 100;
final int maxParallelsim = 2;
@@ -58,49 +112,42 @@ public class StatefulSequenceSourceTest {
final OneShotLatch latchToTrigger2 = new OneShotLatch();
final OneShotLatch latchToWait2 = new OneShotLatch();
- final StatefulSequenceSource source1 = new StatefulSequenceSource(initElement, maxElement);
- StreamSource<Long, StatefulSequenceSource> src1 = new StreamSource<>(source1);
+ final DataGeneratorSource<Long> source1 = new DataGeneratorSource<>(
+ SequenceGenerator.longGenerator(initElement, maxElement));
+ StreamSource<Long, DataGeneratorSource<Long>> src1 = new StreamSource<>(source1);
final AbstractStreamOperatorTestHarness<Long> testHarness1 =
new AbstractStreamOperatorTestHarness<>(src1, maxParallelsim, 2, 0);
testHarness1.open();
- final StatefulSequenceSource source2 = new StatefulSequenceSource(initElement, maxElement);
- StreamSource<Long, StatefulSequenceSource> src2 = new StreamSource<>(source2);
+ final DataGeneratorSource<Long> source2 = new DataGeneratorSource<>(
+ SequenceGenerator.longGenerator(initElement, maxElement));
+ StreamSource<Long, DataGeneratorSource<Long>> src2 = new StreamSource<>(source2);
final AbstractStreamOperatorTestHarness<Long> testHarness2 =
new AbstractStreamOperatorTestHarness<>(src2, maxParallelsim, 2, 1);
testHarness2.open();
- final Throwable[] error = new Throwable[3];
-
// run the source asynchronously
- Thread runner1 = new Thread() {
- @Override
- public void run() {
- try {
- source1.run(new BlockingSourceContext("1", latchToTrigger1, latchToWait1, outputCollector, 21));
- }
- catch (Throwable t) {
- t.printStackTrace();
- error[0] = t;
- }
+ Thread runner1 = new Thread(() -> {
+ try {
+ source1.run(new BlockingSourceContext(
+ "1", latchToTrigger1, latchToWait1, outputCollector, 21));
+ } catch (Throwable t) {
+ t.printStackTrace();
}
- };
+ });
// run the source asynchronously
- Thread runner2 = new Thread() {
- @Override
- public void run() {
- try {
- source2.run(new BlockingSourceContext("2", latchToTrigger2, latchToWait2, outputCollector, 32));
- }
- catch (Throwable t) {
- t.printStackTrace();
- error[1] = t;
- }
+ Thread runner2 = new Thread(() -> {
+ try {
+ source2.run(new BlockingSourceContext(
+ "2", latchToTrigger2, latchToWait2, outputCollector, 32));
}
- };
+ catch (Throwable t) {
+ t.printStackTrace();
+ }
+ });
runner1.start();
runner2.start();
@@ -118,11 +165,13 @@ public class StatefulSequenceSourceTest {
testHarness2.snapshot(0L, 0L)
);
- final StatefulSequenceSource source3 = new StatefulSequenceSource(initElement, maxElement);
- StreamSource<Long, StatefulSequenceSource> src3 = new StreamSource<>(source3);
+ final DataGeneratorSource<Long> source3 = new DataGeneratorSource<>(
+ SequenceGenerator.longGenerator(initElement, maxElement));
+ StreamSource<Long, DataGeneratorSource<Long>> src3 = new StreamSource<>(source3);
- final OperatorSubtaskState initState = AbstractStreamOperatorTestHarness.repartitionOperatorState(
- snapshot, maxParallelsim, 2, 1, 0);
+ final OperatorSubtaskState initState =
+ AbstractStreamOperatorTestHarness.repartitionOperatorState(
+ snapshot, maxParallelsim, 2, 1, 0);
final AbstractStreamOperatorTestHarness<Long> testHarness3 =
new AbstractStreamOperatorTestHarness<>(src3, maxParallelsim, 1, 0);
@@ -135,18 +184,15 @@ public class StatefulSequenceSourceTest {
latchToWait3.trigger();
// run the source asynchronously
- Thread runner3 = new Thread() {
- @Override
- public void run() {
- try {
- source3.run(new BlockingSourceContext("3", latchToTrigger3, latchToWait3, outputCollector, 3));
- }
- catch (Throwable t) {
- t.printStackTrace();
- error[2] = t;
- }
+ Thread runner3 = new Thread(() -> {
+ try {
+ source3.run(new BlockingSourceContext(
+ "3", latchToTrigger3, latchToWait3, outputCollector, 3));
}
- };
+ catch (Throwable t) {
+ t.printStackTrace();
+ }
+ });
runner3.start();
runner3.join();
@@ -182,74 +228,4 @@ public class StatefulSequenceSourceTest {
runner1.join();
runner2.join();
}
-
- private static class BlockingSourceContext implements SourceFunction.SourceContext<Long> {
-
- private final String name;
-
- private final Object lock;
- private final OneShotLatch latchToTrigger;
- private final OneShotLatch latchToWait;
- private final ConcurrentHashMap<String, List<Long>> collector;
-
- private final int threshold;
- private int counter = 0;
-
- private final List<Long> localOutput;
-
- public BlockingSourceContext(String name, OneShotLatch latchToTrigger, OneShotLatch latchToWait,
- ConcurrentHashMap<String, List<Long>> output, int elemToFire) {
- this.name = name;
- this.lock = new Object();
- this.latchToTrigger = latchToTrigger;
- this.latchToWait = latchToWait;
- this.collector = output;
- this.threshold = elemToFire;
-
- this.localOutput = new ArrayList<>();
- List<Long> prev = collector.put(name, localOutput);
- if (prev != null) {
- Assert.fail();
- }
- }
-
- @Override
- public void collectWithTimestamp(Long element, long timestamp) {
- collect(element);
- }
-
- @Override
- public void collect(Long element) {
- localOutput.add(element);
- if (++counter == threshold) {
- latchToTrigger.trigger();
- try {
- if (!latchToWait.isTriggered()) {
- latchToWait.await();
- }
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- }
- }
-
- @Override
- public void emitWatermark(Watermark mark) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void markAsTemporarilyIdle() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public Object getCheckpointLock() {
- return lock;
- }
-
- @Override
- public void close() {
- }
- }
}
diff --git a/flink-table/flink-table-api-java-bridge/pom.xml b/flink-table/flink-table-api-java-bridge/pom.xml
index b25bb8c..3d79cf2 100644
--- a/flink-table/flink-table-api-java-bridge/pom.xml
+++ b/flink-table/flink-table-api-java-bridge/pom.xml
@@ -67,5 +67,18 @@ under the License.
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <type>test-jar</type>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
</dependencies>
</project>
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/sources/datagen/DataGenTableSource.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/sources/datagen/DataGenTableSource.java
new file mode 100644
index 0000000..8fc61eb
--- /dev/null
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/sources/datagen/DataGenTableSource.java
@@ -0,0 +1,108 @@
+/*
+ * 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.sources.datagen;
+
+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.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}.
+ */
+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(Context context) {
+ return SourceFunctionProvider.of(createSource(), false);
+ }
+
+ @VisibleForTesting
+ public DataGeneratorSource<RowData> createSource() {
+ return new DataGeneratorSource<>(new RowGenerator(), 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 class RowGenerator implements DataGenerator<RowData> {
+
+ @Override
+ public void open(
+ String name,
+ FunctionInitializationContext context,
+ RuntimeContext runtimeContext) throws Exception {
+ for (int i = 0; i < fieldGenerators.length; i++) {
+ fieldGenerators[i].open(schema.getFieldName(i).get(), context, runtimeContext);
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ for (DataGenerator generator : fieldGenerators) {
+ if (!generator.hasNext()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public RowData next() {
+ GenericRowData row = new GenericRowData(schema.getFieldCount());
+ 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/sources/datagen/DataGenTableSourceFactory.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/sources/datagen/DataGenTableSourceFactory.java
new file mode 100644
index 0000000..449513e
--- /dev/null
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/sources/datagen/DataGenTableSourceFactory.java
@@ -0,0 +1,208 @@
+/*
+ * 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.sources.datagen;
+
+import org.apache.flink.annotation.Experimental;
+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.streaming.api.functions.source.datagen.DataGenerator;
+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.source.DynamicTableSource;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.utils.TableSchemaUtils;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * Factory for creating configured instances of {@link DataGenTableSource} in a stream environment.
+ */
+@Experimental
+public class DataGenTableSourceFactory implements DynamicTableSourceFactory {
+
+ public static final String IDENTIFIER = "datagen";
+
+ public static final ConfigOption<Long> ROWS_PER_SECOND = key("rows-per-second")
+ .longType()
+ .defaultValue(Long.MAX_VALUE)
+ .withDescription("Rows per second to control the emit rate.");
+
+ public static final String FIELDS = "fields";
+ public static final String KIND = "kind";
+ public static final String START = "start";
+ public static final String END = "end";
+ public static final String MIN = "min";
+ public static final String MAX = "max";
+ public static final String LENGTH = "length";
+
+ public static final String SEQUENCE = "sequence";
+ public static final String RANDOM = "random";
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(ROWS_PER_SECOND);
+ return options;
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ Configuration options = new Configuration();
+ context.getCatalogTable().getOptions().forEach(options::setString);
+
+ TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
+
+ DataGenerator[] fieldGenerators = new DataGenerator[tableSchema.getFieldCount()];
+ for (int i = 0; i < fieldGenerators.length; i++) {
+ fieldGenerators[i] = createDataGenerator(
+ tableSchema.getFieldName(i).get(),
+ tableSchema.getFieldDataType(i).get(),
+ options);
+ }
+
+ return new DataGenTableSource(fieldGenerators, tableSchema, options.get(ROWS_PER_SECOND));
+ }
+
+ private DataGenerator createDataGenerator(String name, DataType type, ReadableConfig options) {
+ String genType = options.get(
+ key(FIELDS + "." + name + "." + KIND).stringType().defaultValue(RANDOM));
+ switch (genType) {
+ case RANDOM:
+ return createRandomGenerator(name, type, options);
+ case SEQUENCE:
+ return createSequenceGenerator(name, type, options);
+ default:
+ throw new ValidationException("Unsupported generator type: " + genType);
+ }
+ }
+
+ private DataGenerator createRandomGenerator(String name, DataType type, ReadableConfig options) {
+ ConfigOption<Integer> lenKey = key(FIELDS + "." + name + "." + LENGTH)
+ .intType().defaultValue(100);
+ OptionBuilder minKey = key(FIELDS + "." + name + "." + MIN);
+ OptionBuilder maxKey = key(FIELDS + "." + name + "." + MAX);
+ switch (type.getLogicalType().getTypeRoot()) {
+ case BOOLEAN:
+ return RandomGenerator.booleanGenerator();
+ case CHAR:
+ case VARCHAR:
+ int length = options.get(lenKey);
+ return new RandomGenerator<StringData>() {
+ @Override
+ public StringData next() {
+ return StringData.fromString(random.nextHexString(length));
+ }
+ };
+ case TINYINT:
+ return RandomGenerator.byteGenerator(
+ options.get(minKey.intType().defaultValue((int) Byte.MIN_VALUE)).byteValue(),
+ options.get(maxKey.intType().defaultValue((int) Byte.MAX_VALUE)).byteValue());
+ case SMALLINT:
+ return RandomGenerator.shortGenerator(
+ options.get(minKey.intType().defaultValue((int) Short.MIN_VALUE)).shortValue(),
+ options.get(maxKey.intType().defaultValue((int) Short.MAX_VALUE)).shortValue());
+ case INTEGER:
+ return RandomGenerator.intGenerator(
+ options.get(minKey.intType().defaultValue(Integer.MIN_VALUE)),
+ options.get(maxKey.intType().defaultValue(Integer.MAX_VALUE)));
+ case BIGINT:
+ return RandomGenerator.longGenerator(
+ options.get(minKey.longType().defaultValue(Long.MIN_VALUE)),
+ options.get(maxKey.longType().defaultValue(Long.MAX_VALUE)));
+ case FLOAT:
+ return RandomGenerator.floatGenerator(
+ options.get(minKey.floatType().defaultValue(Float.MIN_VALUE)),
+ options.get(maxKey.floatType().defaultValue(Float.MAX_VALUE)));
+ case DOUBLE:
+ return RandomGenerator.doubleGenerator(
+ options.get(minKey.doubleType().defaultValue(Double.MIN_VALUE)),
+ options.get(maxKey.doubleType().defaultValue(Double.MAX_VALUE)));
+ default:
+ throw new ValidationException("Unsupported type: " + type);
+ }
+ }
+
+ private DataGenerator createSequenceGenerator(String name, DataType type, ReadableConfig options) {
+ OptionBuilder startKey = key(FIELDS + "." + name + "." + START);
+ OptionBuilder endKey = key(FIELDS + "." + name + "." + END);
+
+ options.getOptional(startKey.stringType().noDefaultValue()).orElseThrow(
+ () -> new ValidationException("Could not find required property '" + startKey + "'."));
+ options.getOptional(endKey.stringType().noDefaultValue()).orElseThrow(
+ () -> new ValidationException("Could not find required property '" + endKey + "'."));
+
+ switch (type.getLogicalType().getTypeRoot()) {
+ case CHAR:
+ case VARCHAR:
+ return new SequenceGenerator<StringData>(
+ options.get(startKey.longType().noDefaultValue()),
+ options.get(endKey.longType().noDefaultValue())) {
+ @Override
+ public StringData next() {
+ return StringData.fromString(valuesToEmit.poll().toString());
+ }
+ };
+ case TINYINT:
+ return SequenceGenerator.byteGenerator(
+ options.get(startKey.intType().noDefaultValue()).byteValue(),
+ options.get(endKey.intType().noDefaultValue()).byteValue());
+ case SMALLINT:
+ return SequenceGenerator.shortGenerator(
+ options.get(startKey.intType().noDefaultValue()).shortValue(),
+ options.get(endKey.intType().noDefaultValue()).shortValue());
+ case INTEGER:
+ return SequenceGenerator.intGenerator(
+ options.get(startKey.intType().noDefaultValue()),
+ options.get(endKey.intType().noDefaultValue()));
+ case BIGINT:
+ return SequenceGenerator.longGenerator(
+ options.get(startKey.longType().noDefaultValue()),
+ options.get(endKey.longType().noDefaultValue()));
+ case FLOAT:
+ return SequenceGenerator.floatGenerator(
+ options.get(startKey.intType().noDefaultValue()).shortValue(),
+ options.get(endKey.intType().noDefaultValue()).shortValue());
+ case DOUBLE:
+ return SequenceGenerator.doubleGenerator(
+ options.get(startKey.intType().noDefaultValue()),
+ options.get(endKey.intType().noDefaultValue()));
+ default:
+ throw new ValidationException("Unsupported type: " + type);
+ }
+ }
+}
diff --git a/flink-table/flink-table-api-java-bridge/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink-table/flink-table-api-java-bridge/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
new file mode 100644
index 0000000..d610a14
--- /dev/null
+++ b/flink-table/flink-table-api-java-bridge/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.flink.table.sources.datagen.DataGenTableSourceFactory
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
new file mode 100644
index 0000000..d3cf337
--- /dev/null
+++ b/flink-table/flink-table-api-java-bridge/src/test/java/org/apache/flink/table/factories/DataGenTableSourceFactoryTest.java
@@ -0,0 +1,142 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.api.functions.source.datagen.DataGeneratorSource;
+import org.apache.flink.streaming.api.operators.StreamSource;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.util.AbstractStreamOperatorTestHarness;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogTableImpl;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.sources.datagen.DataGenTableSource;
+import org.apache.flink.table.sources.datagen.DataGenTableSourceFactory;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.sources.datagen.DataGenTableSourceFactory.END;
+import static org.apache.flink.table.sources.datagen.DataGenTableSourceFactory.FIELDS;
+import static org.apache.flink.table.sources.datagen.DataGenTableSourceFactory.KIND;
+import static org.apache.flink.table.sources.datagen.DataGenTableSourceFactory.LENGTH;
+import static org.apache.flink.table.sources.datagen.DataGenTableSourceFactory.MAX;
+import static org.apache.flink.table.sources.datagen.DataGenTableSourceFactory.MIN;
+import static org.apache.flink.table.sources.datagen.DataGenTableSourceFactory.RANDOM;
+import static org.apache.flink.table.sources.datagen.DataGenTableSourceFactory.ROWS_PER_SECOND;
+import static org.apache.flink.table.sources.datagen.DataGenTableSourceFactory.SEQUENCE;
+import static org.apache.flink.table.sources.datagen.DataGenTableSourceFactory.START;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for {@link DataGenTableSourceFactory}.
+ */
+public class DataGenTableSourceFactoryTest {
+
+ private static final TableSchema TEST_SCHEMA = TableSchema.builder()
+ .field("f0", DataTypes.STRING())
+ .field("f1", DataTypes.BIGINT())
+ .field("f2", DataTypes.BIGINT())
+ .build();
+
+ @Test
+ public void testSource() throws Exception {
+ DescriptorProperties descriptor = new DescriptorProperties();
+ descriptor.putString(FactoryUtil.CONNECTOR.key(), "datagen");
+ descriptor.putLong(ROWS_PER_SECOND.key(), 100);
+
+ descriptor.putString(FIELDS + ".f0." + KIND, RANDOM);
+ descriptor.putLong(FIELDS + ".f0." + LENGTH, 20);
+
+ descriptor.putString(FIELDS + ".f1." + KIND, RANDOM);
+ descriptor.putLong(FIELDS + ".f1." + MIN, 10);
+ descriptor.putLong(FIELDS + ".f1." + MAX, 100);
+
+ descriptor.putString(FIELDS + ".f2." + KIND, SEQUENCE);
+ descriptor.putLong(FIELDS + ".f2." + START, 50);
+ descriptor.putLong(FIELDS + ".f2." + END, 60);
+
+ DynamicTableSource source = FactoryUtil.createTableSource(
+ null,
+ ObjectIdentifier.of("", "", ""),
+ new CatalogTableImpl(TEST_SCHEMA, descriptor.asMap(), ""),
+ new Configuration(),
+ Thread.currentThread().getContextClassLoader());
+
+ assertTrue(source instanceof DataGenTableSource);
+
+ DataGenTableSource dataGenTableSource = (DataGenTableSource) source;
+ DataGeneratorSource<RowData> gen = dataGenTableSource.createSource();
+
+ StreamSource<RowData, DataGeneratorSource<RowData>> src = new StreamSource<>(gen);
+ AbstractStreamOperatorTestHarness<RowData> testHarness =
+ new AbstractStreamOperatorTestHarness<>(src, 1, 1, 0);
+ testHarness.open();
+
+ List<RowData> results = new ArrayList<>();
+
+ gen.run(new SourceFunction.SourceContext<RowData>() {
+
+ private Object lock = new Object();
+
+ @Override
+ public void collect(RowData element) {
+ results.add(element);
+ }
+
+ @Override
+ public void collectWithTimestamp(RowData element, long timestamp) {
+ }
+
+ @Override
+ public void emitWatermark(Watermark mark) {
+ }
+
+ @Override
+ public void markAsTemporarilyIdle() {
+ }
+
+ @Override
+ public Object getCheckpointLock() {
+ return lock;
+ }
+
+ @Override
+ public void close() {
+ }
+ });
+
+ Assert.assertEquals(11, results.size());
+ for (int i = 0; i < results.size(); i++) {
+ RowData row = results.get(i);
+ Assert.assertEquals(20, row.getString(0).toString().length());
+ long f1 = row.getLong(1);
+ Assert.assertTrue(f1 >= 10 && f1 <= 100);
+ Assert.assertEquals(i + 50, row.getLong(2));
+ }
+ }
+}