fix: csas/ctas with timestamp column is used for output rowtime #4489

spena · 2020-02-07T22:46:30Z

Description

BREAKING-CHANGE:
This change is breaking the behavior of the CSAS/CTAS by transforming the ROWTIME of the output/sink topic to the one specified in the WITH TIMESTAMP property. Previously, the ROWTIME was derived from the input/source topics.

Although this is PR is mostly a fix to correctly write the ROWTIME In the sink topic, I flagged it as a breaking-change because users will experience different results when moving to this new behavior.

How to review:
The main change is found on the new file SinkBuilder which builds a stream using the transform method to modify the timestamp:

stream.transform(new TransformTimestamp<>(timestampColumnIndex))
        .to(topicName, Produced.with(keySerde, valueSerde));

The rest of the changes are just passing the right output timestamp column from the plan to the stream/table builder.

Testing done

Modified unit tests
Added query-tests
Verified manually

Reviewer checklist

Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
Ensure relevant issues are linked (description should include text like "Fixes #")

ksql-streams/src/main/java/io/confluent/ksql/execution/streams/SinkBuilder.java

spena · 2020-02-07T22:54:37Z

ksql-functional-tests/src/test/resources/query-validation-tests/timestampformat.json

@@ -12,7 +12,7 @@
      "name": "timestamp format",
      "statements": [
        "CREATE STREAM TEST (ID bigint, event_timestamp VARCHAR) WITH (kafka_topic='test_topic', value_format='JSON', timestamp='event_timestamp', timestamp_format='yyyy-MM-dd''T''HH:mm:ssX');",
-        "CREATE STREAM TS AS select id, stringtotimestamp(event_timestamp, 'yyyy-MM-dd''T''HH:mm:ssX') as ets from test;"
+        "CREATE STREAM TS WITH (timestamp='ets') AS select id, stringtotimestamp(event_timestamp, 'yyyy-MM-dd''T''HH:mm:ssX') as ets from test;"


I added the WITH TIMESTAMP here to use the source timestamp as the output timestamp.

big-andy-coates

Thanks @spena

I think there's some missing test cases on this new functionality (which I've called out below). I think these will highlight some holes in the implementation: specifically, this doesn't look like it will work if timestampformat is specified in the WITH clause of C*AS statements.

ksql-execution/src/main/java/io/confluent/ksql/execution/plan/StreamSink.java

ksql-execution/src/main/java/io/confluent/ksql/execution/plan/TableSink.java

...onal-tests/src/test/resources/expected_topology/0_6_0-pre/timestampformat_-_timestamp_format

ksql-functional-tests/src/test/resources/query-validation-tests/timestampformat.json

ksql-streams/src/main/java/io/confluent/ksql/execution/streams/SinkBuilder.java

ksql-streams/src/test/java/io/confluent/ksql/execution/streams/StreamSinkBuilderTest.java

big-andy-coates

Thanks @spena

Could of suggestions and questions below. Plus what looks to me like a bug in your change to the json schema file.

Otherwise, LGTM!

big-andy-coates · 2020-02-24T18:25:02Z

ksql-rest-app/src/test/resources/ksql-plan-schema/schema.json

        }
      },
      "title" : "streamSinkV1",
-      "required" : [ "@type", "properties", "source", "formats", "topicName" ]
+      "required" : [ "@type", "properties", "source", "formats", "topicName", "timestampColumn" ]


timestampColumn should be optional, right? Not required.

I think with this the way it is the JSON would fail to validate if it had no timestamp column defined.

@rodesai is there a test somewhere that tests existing plans validate against the schema?

our qtt tests should catch any existing plans that fail to deserialize. In this case I think the schema is just wrong. @spena how did you generate it?

But do we deserialize using the schema?

big-andy-coates · 2020-02-24T18:25:13Z

ksql-rest-app/src/test/resources/ksql-plan-schema/schema.json

        }
      },
      "title" : "tableSinkV1",
-      "required" : [ "@type", "properties", "source", "formats", "topicName" ]
+      "required" : [ "@type", "properties", "source", "formats", "topicName", "timestampColumn" ]


As above:

timestampColumn should be optional, right? Not required.

I think with this the way it is the JSON would fail to validate if it had no timestamp column defined.

big-andy-coates · 2020-02-24T18:35:02Z

ksql-streams/src/main/java/io/confluent/ksql/execution/streams/SinkBuilder.java

+    );
+
+    final KStream<K, GenericRow> transformed = tsTransformer
+        .map(t -> stream.transform(t))


Can we used the variant of transform that takes a Named please, so that we can give the step in the topology a more descriptive name than just KSTREAM-TRANSFORM-0003, e.g. name it Apply-timestamp or similar.

(You'll need to re-write the topologies again once you've done this)

big-andy-coates · 2020-02-24T18:48:26Z

ksql-streams/src/main/java/io/confluent/ksql/execution/streams/SinkBuilder.java

+        .map(c -> sourceSchema.findValueColumn(c).orElseThrow(IllegalStateException::new))
+        .map(Column::index)
+        .map(timestampPolicy::create)
+        .filter(te -> te instanceof AbstractColumnTimestampExtractor)


Remove this filter (and cast), by passing making TimestampExtractionPolicy.create return KsqlTimestampExtraction, where KsqlTimestampExtraction extends TimestampExtraction.

big-andy-coates · 2020-02-24T18:51:46Z

...ain/java/io/confluent/ksql/execution/streams/timestamp/AbstractColumnTimestampExtractor.java

+public abstract class AbstractColumnTimestampExtractor implements TimestampExtractor {
+  protected final int timetampColumnIndex;
+
+  AbstractColumnTimestampExtractor(final int timestampColumnindex) {
+    Preconditions.checkArgument(timestampColumnindex >= 0, "timestampColumnindex must be >= 0");
+    this.timetampColumnIndex = timestampColumnindex;
+  }
+
+  @Override
+  public long extract(final ConsumerRecord<Object, Object> record, final long previousTimestamp) {
+    if (record.value() instanceof GenericRow) {
+      try {
+        return extract((GenericRow) record.value());
+      } catch (final Exception e) {
+        throw new KsqlException("Unable to extract timestamp from record."
+            + " record=" + record,
+            e);
+      }
+    }
+
+    return 0;
+  }
+
+  public abstract long extract(GenericRow row);
+}


I'd consider just making this an interface:

public interface KsqlTimestampExtractor implements TimestampExtractor { long extract(GenericRow row); default long extract(final ConsumerRecord<Object, Object> record, final long previousTimestamp) { return extract((GenericRow) record.value()); } }

This is because the base class can't actually handle getting the column at index timetampColumnIndex out of the row for the subclasses, so the abstract class isn't adding much.

Note:

This impl throws if the value is not a GenericRow - as it should always be a GenericRow and it should fail if its not. Return a timestamp of 0 just doesn't make any sense.

I've removed the common exception handling. Why? Because we'd want the same exception handling when calling extreact(row) directly, so it seems to me we still need the exception handling in the subclasses.

big-andy-coates · 2020-02-24T18:54:47Z

ksql-streams/src/main/java/io/confluent/ksql/execution/streams/SinkBuilder.java

+          processorContext.forward(
+              key,
+              row,
+              To.all().withTimestamp(timestampExtractor.extract(row))


If this throws.... does the exception get logged? Does it kill the streams task?

@rodesai what do you think should be the behaviour here? I guess it should be inline with other parts of the processing. What happens with (de)serialization errors?

big-andy-coates · 2020-02-24T18:56:05Z

ksql-streams/src/test/java/io/confluent/ksql/execution/streams/SinkBuilderTest.java

+  }
+
+  @Test(expected = NullPointerException.class)
+  public void shouldThrowIfNegativeProcessorContext() {


Suggested change

public void shouldThrowIfNegativeProcessorContext() {

public void shouldThrowOnNullProcessorContext() {

spena requested review from rodesai, big-andy-coates and a team February 7, 2020 22:46

spena assigned big-andy-coates Feb 7, 2020

spena commented Feb 7, 2020

View reviewed changes

ksql-streams/src/main/java/io/confluent/ksql/execution/streams/SinkBuilder.java Outdated Show resolved Hide resolved

spena requested a review from mjsax February 7, 2020 22:48

spena commented Feb 7, 2020

View reviewed changes

spena force-pushed the fix_timestamp_sink branch 2 times, most recently from 95d22bf to 2a587f1 Compare February 11, 2020 14:05

big-andy-coates reviewed Feb 18, 2020

View reviewed changes

spena force-pushed the fix_timestamp_sink branch 3 times, most recently from 0e10dcb to d4989d2 Compare February 24, 2020 17:38

big-andy-coates approved these changes Feb 24, 2020

View reviewed changes

spena force-pushed the fix_timestamp_sink branch 3 times, most recently from 5df1126 to 2eece42 Compare February 25, 2020 23:32

spena added 8 commits February 26, 2020 10:10

feat: csas/ctas with timestamp column is used for output rowtime

4ca53c1

fix: return record when WITH TIMESTAMP is not set

89aecd6

fix: address Andy's feedback

6b26d66

fix: address Andy's feedback (run TopologyFileWriter)

66f027c

fix: support string/long/any timestamp, add new QTT tests

965dace

fix: send transform forward errors to processing log

489cc33

fix: convert AbstractColumnTimestampExtractor to KsqlTimestampExtractor

5b4058c

fix: regenerate query plans for join-with-custom-timestamp

aa40d7b

spena force-pushed the fix_timestamp_sink branch from 2eece42 to aa40d7b Compare February 26, 2020 16:28

spena merged commit ddddf92 into confluentinc:master Feb 26, 2020

spena deleted the fix_timestamp_sink branch February 26, 2020 17:28

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix: csas/ctas with timestamp column is used for output rowtime #4489

fix: csas/ctas with timestamp column is used for output rowtime #4489

spena commented Feb 7, 2020

spena Feb 7, 2020

big-andy-coates left a comment

big-andy-coates left a comment

big-andy-coates Feb 24, 2020

big-andy-coates Feb 24, 2020

rodesai Feb 24, 2020

big-andy-coates Feb 25, 2020

big-andy-coates Feb 24, 2020

big-andy-coates Feb 24, 2020

big-andy-coates Feb 24, 2020

big-andy-coates Feb 24, 2020

big-andy-coates Feb 24, 2020 •

edited

Loading

big-andy-coates Feb 24, 2020

	public void shouldThrowIfNegativeProcessorContext() {
	public void shouldThrowOnNullProcessorContext() {

fix: csas/ctas with timestamp column is used for output rowtime #4489

fix: csas/ctas with timestamp column is used for output rowtime #4489

Conversation

spena commented Feb 7, 2020

Description

Testing done

Reviewer checklist

Choose a reason for hiding this comment

big-andy-coates left a comment

Choose a reason for hiding this comment

big-andy-coates left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

big-andy-coates Feb 24, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

big-andy-coates Feb 24, 2020 •

edited

Loading