You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2021/09/14 13:01:01 UTC
[flink] branch master updated: [FLINK-22275][table] Support random
past for timestamp types in datagen connector (#15703)
This is an automated email from the ASF dual-hosted git repository.
jark 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 f74db4b [FLINK-22275][table] Support random past for timestamp types in datagen connector (#15703)
f74db4b is described below
commit f74db4b6cce9a28f484dc9d5e6470a50b15c121a
Author: Yi Tang <ss...@gmail.com>
AuthorDate: Tue Sep 14 21:00:34 2021 +0800
[FLINK-22275][table] Support random past for timestamp types in datagen connector (#15703)
---
docs/content.zh/docs/connectors/table/datagen.md | 7 +++
docs/content/docs/connectors/table/datagen.md | 17 ++++++-
.../src/test/resources/sql/table.q | 1 +
.../datagen/table/DataGenConnectorOptions.java | 11 +++++
.../datagen/table/DataGenConnectorOptionsUtil.java | 1 +
.../datagen/table/DataGenTableSourceFactory.java | 1 +
.../datagen/table/DataGenVisitorBase.java | 22 ---------
.../datagen/table/RandomGeneratorVisitor.java | 53 ++++++++++++++++++++++
.../factories/DataGenTableSourceFactoryTest.java | 15 +++++-
9 files changed, 103 insertions(+), 25 deletions(-)
diff --git a/docs/content.zh/docs/connectors/table/datagen.md b/docs/content.zh/docs/connectors/table/datagen.md
index 287686f..96dff1c 100644
--- a/docs/content.zh/docs/connectors/table/datagen.md
+++ b/docs/content.zh/docs/connectors/table/datagen.md
@@ -125,6 +125,13 @@ CREATE TABLE datagen (
<td>随机生成器的最大值，适用于数字类型。</td>
</tr>
<tr>
+ <td><h5>fields.#.max-past</h5></td>
+ <td>可选</td>
+ <td style="word-wrap: break-word;">0</td>
+ <td>Duration</td>
+ <td>随机生成器生成相对当前时间向过去偏移的最大值，适用于 timestamp 类型。</td>
+ </tr>
+ <tr>
<td><h5>fields.#.length</h5></td>
<td>可选</td>
<td style="word-wrap: break-word;">100</td>
diff --git a/docs/content/docs/connectors/table/datagen.md b/docs/content/docs/connectors/table/datagen.md
index aff8683..d4f9432 100644
--- a/docs/content/docs/connectors/table/datagen.md
+++ b/docs/content/docs/connectors/table/datagen.md
@@ -157,12 +157,18 @@ Types
<tr>
<td>TIMESTAMP</td>
<td>random</td>
- <td>Always resolves to the current timestamp of the local machine.</td>
+ <td>
+ Resolves a past timestamp relative to the current timestamp of the local machine.
+ The max past can be specified by the 'max-past' option.
+ </td>
</tr>
<tr>
<td>TIMESTAMP_LTZ</td>
<td>random</td>
- <td>Always resolves to the current timestamp of the local machine.</td>
+ <td>
+ Resolves a past timestamp relative to the current timestamp of the local machine.
+ The max past can be specified by the 'max-past' option.
+ </td>
</tr>
<tr>
<td>INTERVAL YEAR TO MONTH</td>
@@ -254,6 +260,13 @@ Connector Options
<td>Maximum value of random generator, work for numeric types.</td>
</tr>
<tr>
+ <td><h5>fields.#.max-past</h5></td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">0</td>
+ <td>Duration</td>
+ <td>Maximum past of timestamp random generator, only works for timestamp types.</td>
+ </tr>
+ <tr>
<td><h5>fields.#.length</h5></td>
<td>optional</td>
<td style="word-wrap: break-word;">100</td>
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/table.q b/flink-table/flink-sql-client/src/test/resources/sql/table.q
index 44f0f96..fd840bb 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/table.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/table.q
@@ -350,6 +350,7 @@ fields.amount.min
fields.product.kind
fields.product.length
fields.ts.kind
+fields.ts.max-past
fields.user.kind
fields.user.max
fields.user.min
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenConnectorOptions.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenConnectorOptions.java
index e795c92..991f2fa 100644
--- a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenConnectorOptions.java
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenConnectorOptions.java
@@ -22,12 +22,15 @@ import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
+import java.time.Duration;
+
import static org.apache.flink.configuration.ConfigOptions.key;
import static org.apache.flink.connector.datagen.table.DataGenConnectorOptionsUtil.END;
import static org.apache.flink.connector.datagen.table.DataGenConnectorOptionsUtil.FIELDS;
import static org.apache.flink.connector.datagen.table.DataGenConnectorOptionsUtil.KIND;
import static org.apache.flink.connector.datagen.table.DataGenConnectorOptionsUtil.LENGTH;
import static org.apache.flink.connector.datagen.table.DataGenConnectorOptionsUtil.MAX;
+import static org.apache.flink.connector.datagen.table.DataGenConnectorOptionsUtil.MAX_PAST;
import static org.apache.flink.connector.datagen.table.DataGenConnectorOptionsUtil.MIN;
import static org.apache.flink.connector.datagen.table.DataGenConnectorOptionsUtil.ROWS_PER_SECOND_DEFAULT_VALUE;
import static org.apache.flink.connector.datagen.table.DataGenConnectorOptionsUtil.START;
@@ -77,6 +80,14 @@ public class DataGenConnectorOptions {
"Maximum value to generate for fields of kind 'random'. Maximum value possible for the type of the field.");
/** Placeholder {@link ConfigOption}. Not used for retrieving values. */
+ public static final ConfigOption<Duration> FIELD_MAX_PAST =
+ ConfigOptions.key(String.format("%s.#.%s", FIELDS, MAX_PAST))
+ .durationType()
+ .noDefaultValue()
+ .withDescription(
+ "Maximum past relative to the current timestamp of the local machine to generate for timestamp fields of kind 'random'.");
+
+ /** Placeholder {@link ConfigOption}. Not used for retrieving values. */
public static final ConfigOption<Integer> FIELD_LENGTH =
ConfigOptions.key(String.format("%s.#.%s", FIELDS, LENGTH))
.intType()
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenConnectorOptionsUtil.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenConnectorOptionsUtil.java
index c7a1ba1..db5ba1f 100644
--- a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenConnectorOptionsUtil.java
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenConnectorOptionsUtil.java
@@ -32,6 +32,7 @@ public class DataGenConnectorOptionsUtil {
public static final String END = "end";
public static final String MIN = "min";
public static final String MAX = "max";
+ public static final String MAX_PAST = "max-past";
public static final String LENGTH = "length";
public static final String SEQUENCE = "sequence";
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenTableSourceFactory.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenTableSourceFactory.java
index d6c773a..7873fb5 100644
--- a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenTableSourceFactory.java
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenTableSourceFactory.java
@@ -65,6 +65,7 @@ public class DataGenTableSourceFactory implements DynamicTableSourceFactory {
options.add(DataGenConnectorOptions.FIELD_KIND);
options.add(DataGenConnectorOptions.FIELD_MIN);
options.add(DataGenConnectorOptions.FIELD_MAX);
+ options.add(DataGenConnectorOptions.FIELD_MAX_PAST);
options.add(DataGenConnectorOptions.FIELD_LENGTH);
options.add(DataGenConnectorOptions.FIELD_START);
options.add(DataGenConnectorOptions.FIELD_END);
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenVisitorBase.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenVisitorBase.java
index ba8c281..8fbc894 100644
--- a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenVisitorBase.java
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/DataGenVisitorBase.java
@@ -24,13 +24,9 @@ import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.streaming.api.functions.source.datagen.DataGenerator;
import org.apache.flink.table.api.ValidationException;
-import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.types.logical.DateType;
-import org.apache.flink.table.types.logical.LocalZonedTimestampType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.TimeType;
-import org.apache.flink.table.types.logical.TimestampType;
-import org.apache.flink.table.types.logical.ZonedTimestampType;
import org.apache.flink.table.types.logical.utils.LogicalTypeDefaultVisitor;
import java.io.Serializable;
@@ -67,24 +63,6 @@ public abstract class DataGenVisitorBase extends LogicalTypeDefaultVisitor<DataG
}
@Override
- public DataGeneratorContainer visit(TimestampType timestampType) {
- return DataGeneratorContainer.of(
- TimeGenerator.of(() -> TimestampData.fromEpochMillis(System.currentTimeMillis())));
- }
-
- @Override
- public DataGeneratorContainer visit(ZonedTimestampType zonedTimestampType) {
- return DataGeneratorContainer.of(
- TimeGenerator.of(() -> TimestampData.fromEpochMillis(System.currentTimeMillis())));
- }
-
- @Override
- public DataGeneratorContainer visit(LocalZonedTimestampType localZonedTimestampType) {
- return DataGeneratorContainer.of(
- TimeGenerator.of(() -> TimestampData.fromEpochMillis(System.currentTimeMillis())));
- }
-
- @Override
protected DataGeneratorContainer defaultMethod(LogicalType logicalType) {
throw new ValidationException("Unsupported type: " + logicalType);
}
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/RandomGeneratorVisitor.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/RandomGeneratorVisitor.java
index 0116e8b..28f3782 100644
--- a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/RandomGeneratorVisitor.java
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/RandomGeneratorVisitor.java
@@ -31,6 +31,7 @@ import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.data.GenericArrayData;
import org.apache.flink.table.data.GenericMapData;
import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.types.logical.ArrayType;
import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.table.types.logical.BooleanType;
@@ -40,15 +41,19 @@ import org.apache.flink.table.types.logical.DecimalType;
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.LocalZonedTimestampType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.MapType;
import org.apache.flink.table.types.logical.MultisetType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.SmallIntType;
+import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.TinyIntType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.logical.YearMonthIntervalType;
+import org.apache.flink.table.types.logical.ZonedTimestampType;
+import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -69,6 +74,8 @@ public class RandomGeneratorVisitor extends DataGenVisitorBase {
private final ConfigOptions.OptionBuilder maxKey;
+ private final ConfigOptions.OptionBuilder maxPastKey;
+
public RandomGeneratorVisitor(String name, ReadableConfig config) {
super(name, config);
@@ -86,6 +93,13 @@ public class RandomGeneratorVisitor extends DataGenVisitorBase {
+ name
+ "."
+ DataGenConnectorOptionsUtil.MAX);
+ this.maxPastKey =
+ key(
+ DataGenConnectorOptionsUtil.FIELDS
+ + "."
+ + name
+ + "."
+ + DataGenConnectorOptionsUtil.MAX_PAST);
}
@Override
@@ -204,6 +218,33 @@ public class RandomGeneratorVisitor extends DataGenVisitorBase {
}
@Override
+ public DataGeneratorContainer visit(TimestampType timestampType) {
+ ConfigOption<Duration> maxPastOption =
+ maxPastKey.durationType().defaultValue(Duration.ZERO);
+
+ return DataGeneratorContainer.of(
+ getRandomPastTimestampGenerator(config.get(maxPastOption)), maxPastOption);
+ }
+
+ @Override
+ public DataGeneratorContainer visit(ZonedTimestampType zonedTimestampType) {
+ ConfigOption<Duration> maxPastOption =
+ maxPastKey.durationType().defaultValue(Duration.ZERO);
+
+ return DataGeneratorContainer.of(
+ getRandomPastTimestampGenerator(config.get(maxPastOption)), maxPastOption);
+ }
+
+ @Override
+ public DataGeneratorContainer visit(LocalZonedTimestampType localZonedTimestampType) {
+ ConfigOption<Duration> maxPastOption =
+ maxPastKey.durationType().defaultValue(Duration.ZERO);
+
+ return DataGeneratorContainer.of(
+ getRandomPastTimestampGenerator(config.get(maxPastOption)), maxPastOption);
+ }
+
+ @Override
public DataGeneratorContainer visit(ArrayType arrayType) {
ConfigOption<Integer> lenOption =
key(DataGenConnectorOptionsUtil.FIELDS
@@ -325,4 +366,16 @@ public class RandomGeneratorVisitor extends DataGenVisitorBase {
}
};
}
+
+ private static RandomGenerator<TimestampData> getRandomPastTimestampGenerator(
+ Duration maxPast) {
+ return new RandomGenerator<TimestampData>() {
+ @Override
+ public TimestampData next() {
+ long maxPastMillis = maxPast.toMillis();
+ long past = maxPastMillis > 0 ? random.nextLong(0, maxPastMillis) : 0;
+ return TimestampData.fromEpochMillis(System.currentTimeMillis() - past);
+ }
+ };
+ }
}
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 a8b3e4c..40cfa0d 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
@@ -35,6 +35,7 @@ import org.apache.flink.table.catalog.ResolvedSchema;
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.data.TimestampData;
import org.apache.flink.table.descriptors.DescriptorProperties;
import org.apache.flink.util.InstantiationUtil;
@@ -57,7 +58,8 @@ public class DataGenTableSourceFactoryTest {
ResolvedSchema.of(
Column.physical("f0", DataTypes.STRING()),
Column.physical("f1", DataTypes.BIGINT()),
- Column.physical("f2", DataTypes.BIGINT()));
+ Column.physical("f2", DataTypes.BIGINT()),
+ Column.physical("f3", DataTypes.TIMESTAMP()));
@Test
public void testDataTypeCoverage() throws Exception {
@@ -156,7 +158,16 @@ public class DataGenTableSourceFactoryTest {
descriptor.putLong(
DataGenConnectorOptionsUtil.FIELDS + ".f2." + DataGenConnectorOptionsUtil.END, 60);
+ descriptor.putString(
+ DataGenConnectorOptionsUtil.FIELDS + ".f3." + DataGenConnectorOptionsUtil.KIND,
+ DataGenConnectorOptionsUtil.RANDOM);
+ descriptor.putString(
+ DataGenConnectorOptionsUtil.FIELDS + ".f3." + DataGenConnectorOptionsUtil.MAX_PAST,
+ "5s");
+
+ final long begin = System.currentTimeMillis();
List<RowData> results = runGenerator(SCHEMA, descriptor);
+ final long end = System.currentTimeMillis();
Assert.assertEquals(11, results.size());
for (int i = 0; i < results.size(); i++) {
@@ -165,6 +176,8 @@ public class DataGenTableSourceFactoryTest {
long f1 = row.getLong(1);
Assert.assertTrue(f1 >= 10 && f1 <= 100);
Assert.assertEquals(i + 50, row.getLong(2));
+ final TimestampData f3 = row.getTimestamp(3, 3);
+ Assert.assertTrue(f3.getMillisecond() >= begin - 5000 && f3.getMillisecond() <= end);
}
}