You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/03/30 07:57:53 UTC
[GitHub] [flink] twalthr opened a new pull request #15428: [FLINK-19977][table] Add new Add new StreamTableEnvironment.fromDataStream
twalthr opened a new pull request #15428:
URL: https://github.com/apache/flink/pull/15428
## What is the purpose of the change
This reworks `StreamTableEnvironment.fromDataStream` as mentioned in FLIP-136. The new method is deeply integrated into the API and behaves similar to a `DynamicTableSource`. Similar to sources, it is possible to define a `Schema`, read metadata, and declare computed columns or watermarks including `SOURCE_WATERMARK()`. Type information is converted to the new type system without any loss of information. All conversion classes are supported.
```
final DataStream<Tuple3<Long, Integer, String>> dataStream = getWatermarkedDataStream();
final Table table =
tableEnv.fromDataStream(
dataStream,
Schema.newBuilder()
.columnByMetadata("rowtime", "TIMESTAMP(3)")
.watermark("rowtime", "SOURCE_WATERMARK()")
.build());
```
## Brief change log
- Add a dedicated `DynamicTableSource` for internal translation
- Add a `TransformationScanProvider` for internal translation
- Add an `InputConversionOperator` for converting to internal data structures
- Exposes the new logic via `StreamTableEnvironment.fromDataStream`
- Aims to keep most of the old tests running
## Verifying this change
This change is already covered by existing tests, such as `InputConversionOperatorTest`, `DataStreamJavaITCase`.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: yes
- The serializers: no
- The runtime per-record code paths (performance sensitive): yes
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? JavaDocs more docs will follow
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15428: [FLINK-19977][table] Add new StreamTableEnvironment.fromDataStream
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15428:
URL: https://github.com/apache/flink/pull/15428#issuecomment-810011468
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3d05b73ba6f0a6a177ed3493caf43d1a39238904",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15772",
"triggerID" : "3d05b73ba6f0a6a177ed3493caf43d1a39238904",
"triggerType" : "PUSH"
}, {
"hash" : "39a623bb95d1363bcedf93b481493a52d47ec2e8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "39a623bb95d1363bcedf93b481493a52d47ec2e8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3d05b73ba6f0a6a177ed3493caf43d1a39238904 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15772)
* 39a623bb95d1363bcedf93b481493a52d47ec2e8 UNKNOWN
<details>
<summary>Bot commands</summary>
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
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] twalthr commented on pull request #15428: [FLINK-19977][table] Add new StreamTableEnvironment.fromDataStream
Posted by GitBox <gi...@apache.org>.
twalthr commented on pull request #15428:
URL: https://github.com/apache/flink/pull/15428#issuecomment-811150482
Thanks @Airblader. I hope I could address all your comments. I will merge this once the build is green.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15428: [FLINK-19977][table] Add new StreamTableEnvironment.fromDataStream
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15428:
URL: https://github.com/apache/flink/pull/15428#issuecomment-810011468
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3d05b73ba6f0a6a177ed3493caf43d1a39238904",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15772",
"triggerID" : "3d05b73ba6f0a6a177ed3493caf43d1a39238904",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3d05b73ba6f0a6a177ed3493caf43d1a39238904 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15772)
<details>
<summary>Bot commands</summary>
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
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15428: [FLINK-19977][table] Add new StreamTableEnvironment.fromDataStream
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15428:
URL: https://github.com/apache/flink/pull/15428#issuecomment-810011468
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3d05b73ba6f0a6a177ed3493caf43d1a39238904",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15772",
"triggerID" : "3d05b73ba6f0a6a177ed3493caf43d1a39238904",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3d05b73ba6f0a6a177ed3493caf43d1a39238904 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15772)
<details>
<summary>Bot commands</summary>
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
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #15428: [FLINK-19977][table] Add new StreamTableEnvironment.fromDataStream
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #15428:
URL: https://github.com/apache/flink/pull/15428#issuecomment-810005195
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 3d05b73ba6f0a6a177ed3493caf43d1a39238904 (Tue Mar 30 07:59:40 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## 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](https://flink.apache.org/contributing/reviewing-prs.html) 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 <summary>Bot commands</summary>
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
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] twalthr closed pull request #15428: [FLINK-19977][table] Add new StreamTableEnvironment.fromDataStream
Posted by GitBox <gi...@apache.org>.
twalthr closed pull request #15428:
URL: https://github.com/apache/flink/pull/15428
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #15428: [FLINK-19977][table] Add new StreamTableEnvironment.fromDataStream
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #15428:
URL: https://github.com/apache/flink/pull/15428#issuecomment-810011468
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3d05b73ba6f0a6a177ed3493caf43d1a39238904",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3d05b73ba6f0a6a177ed3493caf43d1a39238904",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3d05b73ba6f0a6a177ed3493caf43d1a39238904 UNKNOWN
<details>
<summary>Bot commands</summary>
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
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] twalthr commented on a change in pull request #15428: [FLINK-19977][table] Add new StreamTableEnvironment.fromDataStream
Posted by GitBox <gi...@apache.org>.
twalthr commented on a change in pull request #15428:
URL: https://github.com/apache/flink/pull/15428#discussion_r604981686
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/sources/DynamicSourceUtils.java
##########
@@ -155,6 +245,117 @@ public static boolean isSourceChangeEventsDuplicate(
// --------------------------------------------------------------------------------------------
+ /** Creates a specialized node for assigning watermarks. */
+ private static void pushWatermarkAssigner(FlinkRelBuilder relBuilder, ResolvedSchema schema) {
+ final ExpressionConverter converter = new ExpressionConverter(relBuilder);
+ final RelDataType inputRelDataType = relBuilder.peek().getRowType();
+
+ final WatermarkSpec watermarkSpec = schema.getWatermarkSpecs().get(0);
Review comment:
This is checked during schema resolution in `org.apache.flink.table.catalog.DefaultSchemaResolver#resolveWatermarkSpecs`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Airblader commented on a change in pull request #15428: [FLINK-19977][table] Add new StreamTableEnvironment.fromDataStream
Posted by GitBox <gi...@apache.org>.
Airblader commented on a change in pull request #15428:
URL: https://github.com/apache/flink/pull/15428#discussion_r603916451
##########
File path: flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImpl.java
##########
@@ -214,10 +227,71 @@ private static Executor lookupExecutor(
@Override
public <T> Table fromDataStream(DataStream<T> dataStream) {
- JavaDataStreamQueryOperation<T> queryOperation =
- asQueryOperation(dataStream, Optional.empty());
+ return fromDataStreamInternal(dataStream, null, null);
+ }
- return createTable(queryOperation);
+ @Override
+ public <T> Table fromDataStream(DataStream<T> dataStream, Schema schema) {
+ Preconditions.checkNotNull(schema, "Schema must not be null.");
+ return fromDataStreamInternal(dataStream, schema, null);
+ }
+
+ @Override
+ public <T> void createTemporaryView(String path, DataStream<T> dataStream) {
+ createTemporaryView(path, fromDataStreamInternal(dataStream, null, path));
+ }
+
+ @Override
+ public <T> void createTemporaryView(String path, DataStream<T> dataStream, Schema schema) {
+ createTemporaryView(path, fromDataStreamInternal(dataStream, schema, path));
+ }
+
+ private <T> Table fromDataStreamInternal(
+ DataStream<T> dataStream, @Nullable Schema schema, @Nullable String viewPath) {
+ Preconditions.checkNotNull(dataStream, "Data stream must not be null.");
+ final CatalogManager catalogManager = getCatalogManager();
+ final SchemaResolver schemaResolver = catalogManager.getSchemaResolver();
+ final OperationTreeBuilder operationTreeBuilder = getOperationTreeBuilder();
+
+ final UnresolvedIdentifier unresolvedIdentifier;
+ if (viewPath != null) {
+ unresolvedIdentifier = getParser().parseIdentifier(viewPath);
+ } else {
+ unresolvedIdentifier =
+ UnresolvedIdentifier.of("Unregistered_DataStream_" + dataStream.getId());
+ }
+ final ObjectIdentifier objectIdentifier =
+ catalogManager.qualifyIdentifier(unresolvedIdentifier);
+
+ final ExternalSchemaTranslator.InputResult schemaTranslationResult =
+ ExternalSchemaTranslator.fromExternal(
+ catalogManager.getDataTypeFactory(), dataStream.getType(), schema);
+
+ final ResolvedSchema resolvedSchema =
+ schemaTranslationResult.getSchema().resolve(schemaResolver);
+
+ final QueryOperation scanOperation =
+ new JavaExternalQueryOperation<>(
+ objectIdentifier,
+ dataStream,
+ schemaTranslationResult.getPhysicalDataType(),
+ schemaTranslationResult.isTopLevelRecord(),
+ ChangelogMode.insertOnly(),
+ resolvedSchema);
+
+ final List<String> projections = schemaTranslationResult.getProjections();
+ if (projections == null) {
Review comment:
`getProjections` is declared nullable, but at least `fromExternal` doesn't seem to make use of that. Does it make sense to include emptiness in this condition to avoid wrapping it unnecessarily?
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/sources/DynamicSourceUtils.java
##########
@@ -155,6 +245,117 @@ public static boolean isSourceChangeEventsDuplicate(
// --------------------------------------------------------------------------------------------
+ /** Creates a specialized node for assigning watermarks. */
+ private static void pushWatermarkAssigner(FlinkRelBuilder relBuilder, ResolvedSchema schema) {
+ final ExpressionConverter converter = new ExpressionConverter(relBuilder);
+ final RelDataType inputRelDataType = relBuilder.peek().getRowType();
+
+ final WatermarkSpec watermarkSpec = schema.getWatermarkSpecs().get(0);
Review comment:
Should we validate here somehow that only one watermark spec exists (since only one is supported)?
##########
File path: flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/StreamTableEnvironment.java
##########
@@ -202,15 +209,152 @@ static StreamTableEnvironment create(
/**
* Converts the given {@link DataStream} into a {@link Table}.
*
- * <p>The field names of the {@link Table} are automatically derived from the type of the {@link
- * DataStream}.
+ * <p>Column names and types of the {@link Table} are automatically derived from the {@link
+ * TypeInformation} of the {@link DataStream}. If the outermost record's {@link TypeInformation}
+ * is a {@link CompositeType}, it will be flattened in the first level. {@link TypeInformation}
+ * that cannot be represented as one of the listed {@link DataTypes} will be treated as a
+ * black-box {@link DataTypes#RAW(Class, TypeSerializer)} type. Thus, composite nested fields
+ * will not be accessible.
+ *
+ * <p>Since the DataStream API does not support changelog processing natively, this method
+ * assumes append-only/insert-only semantics during the stream-to-table conversion. Records of
+ * type {@link Row} must describe {@link RowKind#INSERT} changes.
+ *
+ * <p>By default, the stream record's timestamp and watermarks are not propagated unless
Review comment:
I assume you're just documenting the existing behavior here, but the saner default would be to propagate them, right? I would assume this detail makes this signature significantly less useful.
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/sources/ExternalDynamicSource.java
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.planner.sources;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+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.abilities.SupportsReadingMetadata;
+import org.apache.flink.table.connector.source.abilities.SupportsSourceWatermark;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.operators.source.InputConversionOperator;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.types.RowKind;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/** Table source for connecting to the external {@link DataStream} API. */
+@Internal
+final class ExternalDynamicSource<E>
+ implements ScanTableSource, SupportsReadingMetadata, SupportsSourceWatermark {
+
+ private static final String ROWTIME_METADATA_KEY = "rowtime";
+
+ private static final DataType ROWTIME_METADATA_DATA_TYPE = DataTypes.TIMESTAMP_LTZ(3).notNull();
+
+ private final ObjectIdentifier identifier;
+
+ private final DataStream<E> dataStream;
+
+ private final DataType physicalDataType;
+
+ private final boolean isTopLevelRecord;
+
+ private final ChangelogMode changelogMode;
+
+ // mutable attributes
+
+ private boolean attachRowtime;
+
+ private boolean propagateWatermark;
+
+ ExternalDynamicSource(
+ ObjectIdentifier identifier,
+ DataStream<E> dataStream,
+ DataType physicalDataType,
+ boolean isTopLevelRecord,
+ ChangelogMode changelogMode) {
+ this.identifier = identifier;
+ this.dataStream = dataStream;
+ this.physicalDataType = physicalDataType;
+ this.isTopLevelRecord = isTopLevelRecord;
+ this.changelogMode = changelogMode;
+ }
+
+ @Override
+ public DynamicTableSource copy() {
+ final ExternalDynamicSource<E> copy =
+ new ExternalDynamicSource<>(
+ identifier, dataStream, physicalDataType, isTopLevelRecord, changelogMode);
+ copy.attachRowtime = attachRowtime;
+ copy.propagateWatermark = propagateWatermark;
+ return copy;
+ }
+
+ @Override
+ public String asSummaryString() {
+ return null;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return changelogMode;
+ }
+
+ @Override
+ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
+ final DataStructureConverter physicalConverter =
+ runtimeProviderContext.createDataStructureConverter(physicalDataType);
+
+ final Transformation<E> externalTransformation = dataStream.getTransformation();
+
+ final Transformation<RowData> conversionTransformation =
+ new OneInputTransformation<>(
+ externalTransformation,
+ generateOperatorName(),
+ new InputConversionOperator<>(
+ physicalConverter,
+ !isTopLevelRecord,
+ attachRowtime,
+ propagateWatermark,
+ changelogMode.containsOnly(RowKind.INSERT)),
+ null, // will be filled by the framework
+ externalTransformation.getParallelism());
+
+ return TransformationScanProvider.of(conversionTransformation, false);
+ }
+
+ private String generateOperatorName() {
+ return String.format(
+ "DataSteamToTable(stream=%s, type=%s, rowtime=%s, watermark=%s)",
+ identifier.asSummaryString(),
+ physicalDataType.toString(),
+ attachRowtime,
+ propagateWatermark);
+ }
+
+ @Override
+ public Map<String, DataType> listReadableMetadata() {
+ return Collections.singletonMap(ROWTIME_METADATA_KEY, ROWTIME_METADATA_DATA_TYPE);
+ }
+
+ @Override
+ public void applyReadableMetadata(List<String> metadataKeys, DataType producedDataType) {
+ if (metadataKeys.contains(ROWTIME_METADATA_KEY)) {
Review comment:
Maybe this is more for my understanding, but can this be called multiple times and if so, do we intentionally do it such that `attachRowtime` will never be unset again?
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/sources/ExternalDynamicSource.java
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.planner.sources;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+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.abilities.SupportsReadingMetadata;
+import org.apache.flink.table.connector.source.abilities.SupportsSourceWatermark;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.operators.source.InputConversionOperator;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.types.RowKind;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/** Table source for connecting to the external {@link DataStream} API. */
+@Internal
+final class ExternalDynamicSource<E>
+ implements ScanTableSource, SupportsReadingMetadata, SupportsSourceWatermark {
+
+ private static final String ROWTIME_METADATA_KEY = "rowtime";
+
+ private static final DataType ROWTIME_METADATA_DATA_TYPE = DataTypes.TIMESTAMP_LTZ(3).notNull();
+
+ private final ObjectIdentifier identifier;
+
+ private final DataStream<E> dataStream;
+
+ private final DataType physicalDataType;
+
+ private final boolean isTopLevelRecord;
+
+ private final ChangelogMode changelogMode;
+
+ // mutable attributes
+
+ private boolean attachRowtime;
+
+ private boolean propagateWatermark;
+
+ ExternalDynamicSource(
+ ObjectIdentifier identifier,
+ DataStream<E> dataStream,
+ DataType physicalDataType,
+ boolean isTopLevelRecord,
+ ChangelogMode changelogMode) {
+ this.identifier = identifier;
+ this.dataStream = dataStream;
+ this.physicalDataType = physicalDataType;
+ this.isTopLevelRecord = isTopLevelRecord;
+ this.changelogMode = changelogMode;
+ }
+
+ @Override
+ public DynamicTableSource copy() {
+ final ExternalDynamicSource<E> copy =
+ new ExternalDynamicSource<>(
+ identifier, dataStream, physicalDataType, isTopLevelRecord, changelogMode);
+ copy.attachRowtime = attachRowtime;
+ copy.propagateWatermark = propagateWatermark;
+ return copy;
+ }
+
+ @Override
+ public String asSummaryString() {
+ return null;
Review comment:
Missing summary :-)
##########
File path: flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImpl.java
##########
@@ -214,10 +227,71 @@ private static Executor lookupExecutor(
@Override
public <T> Table fromDataStream(DataStream<T> dataStream) {
- JavaDataStreamQueryOperation<T> queryOperation =
- asQueryOperation(dataStream, Optional.empty());
+ return fromDataStreamInternal(dataStream, null, null);
+ }
- return createTable(queryOperation);
+ @Override
+ public <T> Table fromDataStream(DataStream<T> dataStream, Schema schema) {
+ Preconditions.checkNotNull(schema, "Schema must not be null.");
+ return fromDataStreamInternal(dataStream, schema, null);
+ }
+
+ @Override
+ public <T> void createTemporaryView(String path, DataStream<T> dataStream) {
+ createTemporaryView(path, fromDataStreamInternal(dataStream, null, path));
+ }
+
+ @Override
+ public <T> void createTemporaryView(String path, DataStream<T> dataStream, Schema schema) {
+ createTemporaryView(path, fromDataStreamInternal(dataStream, schema, path));
+ }
+
+ private <T> Table fromDataStreamInternal(
+ DataStream<T> dataStream, @Nullable Schema schema, @Nullable String viewPath) {
+ Preconditions.checkNotNull(dataStream, "Data stream must not be null.");
+ final CatalogManager catalogManager = getCatalogManager();
+ final SchemaResolver schemaResolver = catalogManager.getSchemaResolver();
+ final OperationTreeBuilder operationTreeBuilder = getOperationTreeBuilder();
+
+ final UnresolvedIdentifier unresolvedIdentifier;
+ if (viewPath != null) {
+ unresolvedIdentifier = getParser().parseIdentifier(viewPath);
+ } else {
+ unresolvedIdentifier =
+ UnresolvedIdentifier.of("Unregistered_DataStream_" + dataStream.getId());
+ }
+ final ObjectIdentifier objectIdentifier =
+ catalogManager.qualifyIdentifier(unresolvedIdentifier);
+
+ final ExternalSchemaTranslator.InputResult schemaTranslationResult =
+ ExternalSchemaTranslator.fromExternal(
+ catalogManager.getDataTypeFactory(), dataStream.getType(), schema);
+
+ final ResolvedSchema resolvedSchema =
+ schemaTranslationResult.getSchema().resolve(schemaResolver);
+
+ final QueryOperation scanOperation =
+ new JavaExternalQueryOperation<>(
+ objectIdentifier,
+ dataStream,
+ schemaTranslationResult.getPhysicalDataType(),
+ schemaTranslationResult.isTopLevelRecord(),
+ ChangelogMode.insertOnly(),
+ resolvedSchema);
+
+ final List<String> projections = schemaTranslationResult.getProjections();
+ if (projections == null) {
Review comment:
Ah, never mind, you also updated `fromExternal` to actually return `null` now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15428: [FLINK-19977][table] Add new StreamTableEnvironment.fromDataStream
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15428:
URL: https://github.com/apache/flink/pull/15428#issuecomment-810011468
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "3d05b73ba6f0a6a177ed3493caf43d1a39238904",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15772",
"triggerID" : "3d05b73ba6f0a6a177ed3493caf43d1a39238904",
"triggerType" : "PUSH"
}, {
"hash" : "39a623bb95d1363bcedf93b481493a52d47ec2e8",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15921",
"triggerID" : "39a623bb95d1363bcedf93b481493a52d47ec2e8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3d05b73ba6f0a6a177ed3493caf43d1a39238904 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15772)
* 39a623bb95d1363bcedf93b481493a52d47ec2e8 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15921)
<details>
<summary>Bot commands</summary>
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
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] twalthr commented on a change in pull request #15428: [FLINK-19977][table] Add new StreamTableEnvironment.fromDataStream
Posted by GitBox <gi...@apache.org>.
twalthr commented on a change in pull request #15428:
URL: https://github.com/apache/flink/pull/15428#discussion_r604985778
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/sources/ExternalDynamicSource.java
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.planner.sources;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+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.abilities.SupportsReadingMetadata;
+import org.apache.flink.table.connector.source.abilities.SupportsSourceWatermark;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.operators.source.InputConversionOperator;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.types.RowKind;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/** Table source for connecting to the external {@link DataStream} API. */
+@Internal
+final class ExternalDynamicSource<E>
+ implements ScanTableSource, SupportsReadingMetadata, SupportsSourceWatermark {
+
+ private static final String ROWTIME_METADATA_KEY = "rowtime";
+
+ private static final DataType ROWTIME_METADATA_DATA_TYPE = DataTypes.TIMESTAMP_LTZ(3).notNull();
+
+ private final ObjectIdentifier identifier;
+
+ private final DataStream<E> dataStream;
+
+ private final DataType physicalDataType;
+
+ private final boolean isTopLevelRecord;
+
+ private final ChangelogMode changelogMode;
+
+ // mutable attributes
+
+ private boolean attachRowtime;
+
+ private boolean propagateWatermark;
+
+ ExternalDynamicSource(
+ ObjectIdentifier identifier,
+ DataStream<E> dataStream,
+ DataType physicalDataType,
+ boolean isTopLevelRecord,
+ ChangelogMode changelogMode) {
+ this.identifier = identifier;
+ this.dataStream = dataStream;
+ this.physicalDataType = physicalDataType;
+ this.isTopLevelRecord = isTopLevelRecord;
+ this.changelogMode = changelogMode;
+ }
+
+ @Override
+ public DynamicTableSource copy() {
+ final ExternalDynamicSource<E> copy =
+ new ExternalDynamicSource<>(
+ identifier, dataStream, physicalDataType, isTopLevelRecord, changelogMode);
+ copy.attachRowtime = attachRowtime;
+ copy.propagateWatermark = propagateWatermark;
+ return copy;
+ }
+
+ @Override
+ public String asSummaryString() {
+ return null;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return changelogMode;
+ }
+
+ @Override
+ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
+ final DataStructureConverter physicalConverter =
+ runtimeProviderContext.createDataStructureConverter(physicalDataType);
+
+ final Transformation<E> externalTransformation = dataStream.getTransformation();
+
+ final Transformation<RowData> conversionTransformation =
+ new OneInputTransformation<>(
+ externalTransformation,
+ generateOperatorName(),
+ new InputConversionOperator<>(
+ physicalConverter,
+ !isTopLevelRecord,
+ attachRowtime,
+ propagateWatermark,
+ changelogMode.containsOnly(RowKind.INSERT)),
+ null, // will be filled by the framework
+ externalTransformation.getParallelism());
+
+ return TransformationScanProvider.of(conversionTransformation, false);
+ }
+
+ private String generateOperatorName() {
+ return String.format(
+ "DataSteamToTable(stream=%s, type=%s, rowtime=%s, watermark=%s)",
+ identifier.asSummaryString(),
+ physicalDataType.toString(),
+ attachRowtime,
+ propagateWatermark);
+ }
+
+ @Override
+ public Map<String, DataType> listReadableMetadata() {
+ return Collections.singletonMap(ROWTIME_METADATA_KEY, ROWTIME_METADATA_DATA_TYPE);
+ }
+
+ @Override
+ public void applyReadableMetadata(List<String> metadataKeys, DataType producedDataType) {
+ if (metadataKeys.contains(ROWTIME_METADATA_KEY)) {
Review comment:
I will make it idempotent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] twalthr commented on a change in pull request #15428: [FLINK-19977][table] Add new StreamTableEnvironment.fromDataStream
Posted by GitBox <gi...@apache.org>.
twalthr commented on a change in pull request #15428:
URL: https://github.com/apache/flink/pull/15428#discussion_r604982067
##########
File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/sources/DynamicSourceUtils.java
##########
@@ -155,6 +245,117 @@ public static boolean isSourceChangeEventsDuplicate(
// --------------------------------------------------------------------------------------------
+ /** Creates a specialized node for assigning watermarks. */
+ private static void pushWatermarkAssigner(FlinkRelBuilder relBuilder, ResolvedSchema schema) {
+ final ExpressionConverter converter = new ExpressionConverter(relBuilder);
+ final RelDataType inputRelDataType = relBuilder.peek().getRowType();
+
+ final WatermarkSpec watermarkSpec = schema.getWatermarkSpecs().get(0);
Review comment:
I will add a comment though.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15428: [FLINK-19977][table] Add new StreamTableEnvironment.fromDataStream
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15428:
URL: https://github.com/apache/flink/pull/15428#issuecomment-810005195
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 39a623bb95d1363bcedf93b481493a52d47ec2e8 (Sat Aug 28 11:11:54 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## 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](https://flink.apache.org/contributing/reviewing-prs.html) 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 <summary>Bot commands</summary>
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
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] twalthr commented on a change in pull request #15428: [FLINK-19977][table] Add new StreamTableEnvironment.fromDataStream
Posted by GitBox <gi...@apache.org>.
twalthr commented on a change in pull request #15428:
URL: https://github.com/apache/flink/pull/15428#discussion_r604979251
##########
File path: flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/StreamTableEnvironment.java
##########
@@ -202,15 +209,152 @@ static StreamTableEnvironment create(
/**
* Converts the given {@link DataStream} into a {@link Table}.
*
- * <p>The field names of the {@link Table} are automatically derived from the type of the {@link
- * DataStream}.
+ * <p>Column names and types of the {@link Table} are automatically derived from the {@link
+ * TypeInformation} of the {@link DataStream}. If the outermost record's {@link TypeInformation}
+ * is a {@link CompositeType}, it will be flattened in the first level. {@link TypeInformation}
+ * that cannot be represented as one of the listed {@link DataTypes} will be treated as a
+ * black-box {@link DataTypes#RAW(Class, TypeSerializer)} type. Thus, composite nested fields
+ * will not be accessible.
+ *
+ * <p>Since the DataStream API does not support changelog processing natively, this method
+ * assumes append-only/insert-only semantics during the stream-to-table conversion. Records of
+ * type {@link Row} must describe {@link RowKind#INSERT} changes.
+ *
+ * <p>By default, the stream record's timestamp and watermarks are not propagated unless
Review comment:
True, but for clear semantics the current default makes more sense. Otherwise we don't know how to name and where to put the time attribute. Also, not every data stream has a time attribute. We cannot know this in advance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org