flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Dawid Wysakowicz (Jira)" <j...@apache.org>
Subject [jira] [Updated] (FLINK-15801) Timestamp extractor created from properties does not work for some physical fields
Date Wed, 29 Jan 2020 09:49:00 GMT

     [ https://issues.apache.org/jira/browse/FLINK-15801?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]

Dawid Wysakowicz updated FLINK-15801:
-------------------------------------
    Description: 
If a timestamp extractor is created from properties it can not use a physical field if the
name of that field is equal to the logical field of the rowtime field.

The code below fails:
{code}
		StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
		streamExecutionEnvironment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
		StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(streamExecutionEnvironment);
		String allEventsTable = "allEventsTable";
		fsTableEnv.connect(new Kafka()
			.version("universal")
			.topic("events")
			.property("zookeeper.connect", "")
			.property("bootstrap.servers", "localhost:9092")
			.property("group.id", "dummyquery").startFromLatest())
			.withSchema(new Schema()
				.field("rule_id", Types.INT)
				.field("sourceAddress", Types.STRING)
				.field("deviceProduct", Types.STRING)
				.field("destHost", Types.STRING)
				.field("extra", Types.STRING)
				.field("rowtime", Types.SQL_TIMESTAMP)
				.rowtime(new Rowtime().timestampsFromField("rowtime").watermarksPeriodicBounded(2000))

			)
			.withFormat(new Json().failOnMissingField(false).deriveSchema())
			.inAppendMode()
			.registerTableSource(allEventsTable);

		Table result = fsTableEnv.sqlQuery("select * from allEventsTable where sourceAddress='12345431'");

		DataStream alert = fsTableEnv.toAppendStream(result, Row.class);
		alert.print();
{code}

with exception:
{code}
Exception in thread "main" org.apache.flink.table.api.ValidationException: Field 'rowtime'
could not be resolved by the field mapping.
    at org.apache.flink.table.sources.TableSourceValidation.resolveField(TableSourceValidation.java:245)
    at org.apache.flink.table.sources.TableSourceValidation.lambda$validateTimestampExtractorArguments$6(TableSourceValidation.java:202)
    at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
    at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
    at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
    at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
    at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:545)
    at java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260)
    at java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:438)
{code}

The problem is that the field is removed from the {{FieldMapping}} in {{org.apache.flink.table.descriptors.SchemaValidator#deriveFieldMapping}}.


One possible solution could be to add: 
{code}
				if (isRowtime) {
					Optional<String> timestampSource = properties.getOptionalString(SCHEMA + "." +
i + "." + ROWTIME_TIMESTAMPS_FROM);
					timestampSource.ifPresent(s -> mapping.put(s, s));
				}
{code}

We should also consider the case what happens if we compute generated columns on fields that
were pruned in a similar way.

Reported by a user: https://stackoverflow.com/questions/59857057/how-to-define-an-apache-flink-table-with-row-time-attribute

  was:
If a timestamp extractor is created from properties it can not use a physical field if the
name of that field is equal to the logical field of the rowtime field.

The code below fails:
{code}
		StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
		streamExecutionEnvironment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
		StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(streamExecutionEnvironment);
		String allEventsTable = "allEventsTable";
		fsTableEnv.connect(new Kafka()
			.version("universal")
			.topic("events")
			.property("zookeeper.connect", "")
			.property("bootstrap.servers", "localhost:9092")
			.property("group.id", "dummyquery").startFromLatest())
			.withSchema(new Schema()
				.field("rule_id", Types.INT)
				.field("sourceAddress", Types.STRING)
				.field("deviceProduct", Types.STRING)
				.field("destHost", Types.STRING)
				.field("extra", Types.STRING)
				.field("rowtime", Types.SQL_TIMESTAMP)
				.rowtime(new Rowtime().timestampsFromField("rowtime").watermarksPeriodicBounded(2000))

			)
			.withFormat(new Json().failOnMissingField(false).deriveSchema())
			.inAppendMode()
			.registerTableSource(allEventsTable);

		Table result = fsTableEnv.sqlQuery("select * from allEventsTable where sourceAddress='12345431'");

		DataStream alert = fsTableEnv.toAppendStream(result, Row.class);
		alert.print();
{code}

with exception:
{code}
Exception in thread "main" org.apache.flink.table.api.ValidationException: Field 'rowtime'
could not be resolved by the field mapping.
    at org.apache.flink.table.sources.TableSourceValidation.resolveField(TableSourceValidation.java:245)
    at org.apache.flink.table.sources.TableSourceValidation.lambda$validateTimestampExtractorArguments$6(TableSourceValidation.java:202)
    at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
    at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
    at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
    at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
    at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:545)
    at java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260)
    at java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:438)
{code}

The problem is that the field is removed from the {{FieldMapping}} in {{org.apache.flink.table.descriptors.SchemaValidator#deriveFieldMapping}}.


One possible solution could be to add: 
{code}
				if (isRowtime) {
					Optional<String> timestampSource = properties.getOptionalString(SCHEMA + "." +
i + "." + ROWTIME_TIMESTAMPS_FROM);
					timestampSource.ifPresent(s -> mapping.put(s, s));
				}
{code}

We should also consider the case what happens if we compute generated columns on fields that
were pruned in a similar way.


> Timestamp extractor created from properties does not work for some physical fields
> ----------------------------------------------------------------------------------
>
>                 Key: FLINK-15801
>                 URL: https://issues.apache.org/jira/browse/FLINK-15801
>             Project: Flink
>          Issue Type: Bug
>          Components: Table SQL / API
>    Affects Versions: 1.9.0, 1.10.0
>            Reporter: Dawid Wysakowicz
>            Priority: Major
>             Fix For: 1.11.0, 1.9.3, 1.10.1
>
>
> If a timestamp extractor is created from properties it can not use a physical field if
the name of that field is equal to the logical field of the rowtime field.
> The code below fails:
> {code}
> 		StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
> 		streamExecutionEnvironment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
> 		StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(streamExecutionEnvironment);
> 		String allEventsTable = "allEventsTable";
> 		fsTableEnv.connect(new Kafka()
> 			.version("universal")
> 			.topic("events")
> 			.property("zookeeper.connect", "")
> 			.property("bootstrap.servers", "localhost:9092")
> 			.property("group.id", "dummyquery").startFromLatest())
> 			.withSchema(new Schema()
> 				.field("rule_id", Types.INT)
> 				.field("sourceAddress", Types.STRING)
> 				.field("deviceProduct", Types.STRING)
> 				.field("destHost", Types.STRING)
> 				.field("extra", Types.STRING)
> 				.field("rowtime", Types.SQL_TIMESTAMP)
> 				.rowtime(new Rowtime().timestampsFromField("rowtime").watermarksPeriodicBounded(2000))
> 			)
> 			.withFormat(new Json().failOnMissingField(false).deriveSchema())
> 			.inAppendMode()
> 			.registerTableSource(allEventsTable);
> 		Table result = fsTableEnv.sqlQuery("select * from allEventsTable where sourceAddress='12345431'");
> 		DataStream alert = fsTableEnv.toAppendStream(result, Row.class);
> 		alert.print();
> {code}
> with exception:
> {code}
> Exception in thread "main" org.apache.flink.table.api.ValidationException: Field 'rowtime'
could not be resolved by the field mapping.
>     at org.apache.flink.table.sources.TableSourceValidation.resolveField(TableSourceValidation.java:245)
>     at org.apache.flink.table.sources.TableSourceValidation.lambda$validateTimestampExtractorArguments$6(TableSourceValidation.java:202)
>     at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
>     at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
>     at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
>     at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
>     at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:545)
>     at java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260)
>     at java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:438)
> {code}
> The problem is that the field is removed from the {{FieldMapping}} in {{org.apache.flink.table.descriptors.SchemaValidator#deriveFieldMapping}}.

> One possible solution could be to add: 
> {code}
> 				if (isRowtime) {
> 					Optional<String> timestampSource = properties.getOptionalString(SCHEMA + "."
+ i + "." + ROWTIME_TIMESTAMPS_FROM);
> 					timestampSource.ifPresent(s -> mapping.put(s, s));
> 				}
> {code}
> We should also consider the case what happens if we compute generated columns on fields
that were pruned in a similar way.
> Reported by a user: https://stackoverflow.com/questions/59857057/how-to-define-an-apache-flink-table-with-row-time-attribute



--
This message was sent by Atlassian Jira
(v8.3.4#803005)

Mime
View raw message