feat: build ks app from an execution plan visitor #3418

rodesai · 2019-09-26T00:08:03Z

Description

This patch implements a visitor that iterates over the execution plan
and builds the final kstreams app. In addition to defining and
implementing the visitor, this required updating the type built by
many of the plan nodes to a wrapper class that includes both a kstream/
ktable, and a factory for building key serdes.

Now that we have this visior, we no longer need the code in SchemaKX
that makes calls into kafka streams, so that's all cleaned up.

Finally, we need to actually call the visitor to build the streams app.
For now that's happening in PhysicalPlanBuilder, but that will get moved
very soon.

Testing done

Adapted execution plan builder tests to incorporate the new visitor.

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 #")

rodesai · 2019-09-26T04:18:37Z

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

+      final StreamSource<?> streamSource
+  ) {
+    if (streamSource.getFormats().getKeyFormat().isWindowed()) {
+      return (StreamWithKeySerdeFactory) buildWindowed(queryBuilder,streamSource);


This is not ideal. Planning to clean this up in a follow-on that adds a separate step type for a windowed source so we can have it strictly typed (similar to aggregates).

purplefox

Looks great! Just a few small comments, mainly on naming.

ksql-engine/src/main/java/io/confluent/ksql/physical/PhysicalPlanBuilder.java

purplefox · 2019-10-02T07:38:03Z

ksql-engine/src/main/java/io/confluent/ksql/physical/PhysicalPlanBuilder.java

    if (sourceType == DataSourceType.KTABLE) {
+      ((SchemaKTable) schemaKStream).getSourceTableStep().build(planBuilder);


purplefox · 2019-10-02T07:46:56Z

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

+import org.apache.kafka.streams.kstream.KGroupedTable;
+import org.apache.kafka.streams.kstream.Windowed;
+
+public interface PlanBuilder {


Would be nice to have a few lines of class header comment explaining what the purpose of this interface is (Not expecting full javadoc!)

purplefox · 2019-10-02T07:56:23Z

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

+import java.util.Objects;
+import org.apache.kafka.streams.kstream.KStream;
+
+public final class StreamWithKeySerdeFactory<K> {


This class doesn't seems to be a StreamWithKeySerdeFactory (i.e. it doesn't create instances of StreamWithKeySerde) but more of a holder class for a stream and a KeySerdeFactory. Perhaps it could have a better name? Maybe StreamHolder?

purplefox · 2019-10-02T08:00:25Z

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

+import java.util.Objects;
+import org.apache.kafka.streams.kstream.KTable;
+
+public final class TableWithKeySerdeFactory<K> {


Naming again. Maybe TableHolder?

purplefox · 2019-10-02T08:04:17Z

ksql-functional-tests/src/main/java/io/confluent/ksql/test/tools/TestExecutor.java

+            testCase.getGeneratedSchemas().get(0).split(System.lineSeparator()));
+        assertThat("Schemas used by topology differ "
+                + "from those used by previous versions"
+                + " of KSQL - this likely means there is a non-backwards compatible change.\n"


"... this is likely to mean..."

purplefox · 2019-10-02T08:06:18Z

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

+import org.apache.kafka.streams.kstream.KGroupedTable;
+import org.apache.kafka.streams.kstream.Windowed;
+
+public final class KSPlanBuilder implements PlanBuilder {


Just a few lines of class header comment would be awesome "This is an implementation of PlanBuilder which uses Kafka streams...."

purplefox · 2019-10-02T08:09:41Z

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

-  public static <K> KStream<K, GenericRow> build(
-      final KStream<K, GenericRow> kstream,
-      final StreamFilter<KStream<K, GenericRow>> step,
+  public static <K> StreamWithKeySerdeFactory<K> build(


This is kind of a general point. I notice that we use "build" in the codebase a lot. And often it's used where the builder pattern is not actually being used, rather, something is just being created. I'd suggest only using "build" where there's a real builder pattern in place, and using "create" in the case of a simple factory method.

Fair enough. I'll defer this to another PR though to keep the scope of the changes here at a minimum.

purplefox · 2019-10-02T08:10:35Z

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

+      final StreamSource<?> streamSource
+  ) {
+    if (streamSource.getFormats().getKeyFormat().isWindowed()) {
+      return (StreamWithKeySerdeFactory) buildWindowed(queryBuilder,streamSource);


purplefox · 2019-10-02T08:12:46Z

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

@@ -79,15 +79,20 @@ private StreamStreamJoinBuilder() {
    );
    final KsqlValueJoiner joiner = new KsqlValueJoiner(leftSchema, rightSchema);
    final JoinWindows joinWindows = JoinWindows.of(join.getBefore()).after(join.getAfter());
+    final KStream<K, GenericRow> result;
    switch (join.getJoinType()) {


Kind of tangential - but I wonder why we don't support RIGHT joins? It would be a simple matter for us to just swap left and right here, rather than forcing the user to do this.

Not sure. What's the value over just doing a left join?

This patch implements a visitor that iterates over the execution plan and builds the final kstreams app. In addition to defining and implementing the visitor, this required updating the type built by many of the plan nodes to a wrapper class that includes both a kstream/ ktable, and a factory for building key serdes. Now that we have this visior, we no longer need the code in SchemaKX that makes calls into kafka streams, so that's all cleaned up. Finally, we need to actually call the visitor to build the streams app. For now that's happening in PhysicalPlanBuilder, but that will get moved very soon.

stevenpyzhang

Overall, I think it looks good, left 2 minor comments.

stevenpyzhang · 2019-10-02T16:08:24Z

ksql-engine/src/main/java/io/confluent/ksql/physical/PhysicalPlanBuilder.java

  ) {
    final DataSourceType sourceType = (schemaKStream instanceof SchemaKTable)
        ? DataSourceType.KTABLE
        : DataSourceType.KSTREAM;

    final DataSource<?> sinkDataSource;
+    final PlanBuilder planBuilder = new KSPlanBuilder(ksqlQueryBuilder);


suggestion (nit): Since a new new KSPlanBuilder() object is created in both buildPlanForStructuredOutputNode() and buildPlanForBareQuery(), this new object could instead be created in buildPlan() and passed into both methods. I don't feel too strongly about this though.

stevenpyzhang · 2019-10-02T21:15:13Z

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

 import org.apache.kafka.streams.kstream.KeyValueMapper;

 public final class TableGroupByBuilder {
  private TableGroupByBuilder() {
  }

  public static <K> KGroupedTable<Struct, GenericRow> build(
-      final KTable<K, GenericRow> ktable,
+      final KTableHolder<K> table,


[nit] Could this be renamed to tableHolder? This goes for other holder variables in all the other builders. It seems weird to have table.getTable() and stream.getStream() (it looks like we already have a table available, why would we need to call getTable() from it?)

rodesai requested a review from a team as a code owner September 26, 2019 00:08

rodesai commented Sep 26, 2019

View reviewed changes

rodesai force-pushed the build-streams-from-visitor-rebase branch 2 times, most recently from ec682fe to d688767 Compare October 1, 2019 02:14

purplefox reviewed Oct 2, 2019

View reviewed changes

rodesai force-pushed the build-streams-from-visitor-rebase branch from d688767 to 8f41e9d Compare October 2, 2019 17:55

purplefox approved these changes Oct 2, 2019

View reviewed changes

rodesai force-pushed the build-streams-from-visitor-rebase branch from 8f41e9d to edfa8c9 Compare October 2, 2019 18:14

broken test

f2b66a9

stevenpyzhang reviewed Oct 2, 2019

View reviewed changes

rodesai merged commit b57d194 into confluentinc:master Oct 2, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: build ks app from an execution plan visitor #3418

feat: build ks app from an execution plan visitor #3418

rodesai commented Sep 26, 2019 •

edited

rodesai Sep 26, 2019

purplefox Oct 2, 2019

purplefox left a comment

purplefox Oct 2, 2019

purplefox Oct 2, 2019

purplefox Oct 2, 2019

purplefox Oct 2, 2019

purplefox Oct 2, 2019

purplefox Oct 2, 2019

purplefox Oct 2, 2019

rodesai Oct 2, 2019

purplefox Oct 2, 2019

purplefox Oct 2, 2019

rodesai Oct 2, 2019

stevenpyzhang left a comment

stevenpyzhang Oct 2, 2019

stevenpyzhang Oct 2, 2019 •

edited

		if (sourceType == DataSourceType.KTABLE) {
		((SchemaKTable) schemaKStream).getSourceTableStep().build(planBuilder);

feat: build ks app from an execution plan visitor #3418

feat: build ks app from an execution plan visitor #3418

Conversation

rodesai commented Sep 26, 2019 • edited

Description

Testing done

Reviewer checklist

Choose a reason for hiding this comment

Choose a reason for hiding this comment

purplefox 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stevenpyzhang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stevenpyzhang Oct 2, 2019 • edited

Choose a reason for hiding this comment

rodesai commented Sep 26, 2019 •

edited

stevenpyzhang Oct 2, 2019 •

edited