feat: move groupBy into plan builders #3359

rodesai · 2019-09-16T18:30:13Z

Description

This patch moves the code for regrouping streams/tables into plan
builders. This also required adding a new execution step for
groupByKey, which we missed the first go-round.

Testing done

Adds tests for the new grouped step builders.

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

big-andy-coates

Thanks @rodesai

As per other PRs there are issues with the generics and use of raw types in this that I'd like to see cleaned up before I can approve.

big-andy-coates · 2019-09-17T11:37:25Z

ksql-execution/src/main/java/io/confluent/ksql/execution/builder/KsqlQueryBuilder.java

@@ -152,6 +153,28 @@ public KsqlQueryBuilder withKsqlConfig(final KsqlConfig newConfig) {
    );
  }

+  @SuppressWarnings("unchecked")
+  public KeySerde<Object> buildKeySerde(


As per other PRs - the generics are incorrect

big-andy-coates · 2019-09-17T11:38:13Z

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

+import java.util.Objects;
+
+@Immutable
+public class StreamGroupByKey<S, G> implements ExecutionStep<G> {


Would be good to get some javadocs on what the generic parameters are for each step type.

big-andy-coates · 2019-09-17T11:39:58Z

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

+import java.util.Objects;
+
+@Immutable
+public class StreamGroupByKey<S, G> implements ExecutionStep<G> {


If we're grouping by the existing key, would not S and G be the same by definition?

These were originally meant to be the full stream and table types. I've switched it to just be key types now (in this PR for just the grouped steps, but I'll do that for all the steps). So this will go away.

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

big-andy-coates · 2019-09-17T11:45:13Z

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

 import org.apache.kafka.streams.kstream.KeyValueMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

-class GroupByMapper implements KeyValueMapper<Object, GenericRow, Struct> {
+class GroupByMapper implements KeyValueMapper<Object, GenericRow, Object> {


Why are we changing the return value on this mapper? It's always going to be a Struct...

big-andy-coates · 2019-09-17T12:04:20Z

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

+      .withMetaAndKeyColsInValue();
+  private static final PhysicalSchema PHYSICAL_SCHEMA = PhysicalSchema.from(SCHEMA, SerdeOption.none());
+
+  private final List<Expression> groupByExpressions = ImmutableList.of(


nit: could be static.

Suggested change

private final List<Expression> groupByExpressions = ImmutableList.of(

private static final List<Expression> GROUP_BY_EXPRESSIONS = ImmutableList.of(

Likely others could too.

big-andy-coates · 2019-09-17T12:05:06Z

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

+  @Mock
+  private FunctionRegistry functionRegistry;
+  @Mock
+  private GroupedFactory groupedFactory;


Use of raw types - can we please use generic types.

big-andy-coates · 2019-09-17T12:05:23Z

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

+  }
+
+  @Test
+  @SuppressWarnings("unchecked")


consider moving to class scope?

big-andy-coates · 2019-09-17T12:05:45Z

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

+      .withMetaAndKeyColsInValue();
+  private static final PhysicalSchema PHYSICAL_SCHEMA = PhysicalSchema.from(SCHEMA, SerdeOption.none());
+
+  private final List<Expression> groupByExpressions = ImmutableList.of(


nit: can be static, likely others too.

big-andy-coates · 2019-09-17T12:05:59Z

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

+  @Mock
+  private FunctionRegistry functionRegistry;
+  @Mock
+  private GroupedFactory groupedFactory;


raw types - can we please use generic type.s

raw types - can we please use generic types. Same for ExecutionStep

This patch moves the code for regrouping streams/tables into plan builders. This also required adding a new execution step for groupByKey, which we missed the first go-round.

big-andy-coates

Thanks @rodesai LGTM 'cept a few nits

big-andy-coates · 2019-09-19T16:15:37Z

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

@@ -66,7 +78,7 @@ public boolean equals(final Object o) {
    if (o == null || getClass() != o.getClass()) {
      return false;
    }
-    final StreamGroupBy<?, ?> that = (StreamGroupBy<?, ?>) o;
+    final StreamGroupBy that = (StreamGroupBy) o;


raw type...

big-andy-coates · 2019-09-19T16:15:55Z

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

+
+  @Override
+  public int hashCode() {
+


super nit: blank line

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

big-andy-coates · 2019-09-19T16:17:42Z

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


 @Immutable
-public class TableGroupBy<T, G> implements ExecutionStep<G> {
+public class TableGroupBy<K> implements ExecutionStep<KGroupedTable<Struct, GenericRow>> {


Like others, I think we can drop the generic K from this type and just use ?. K is the generic type of the source for this step, it doesn't make sense to have this on the type itself.

big-andy-coates · 2019-09-19T16:18:52Z

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

+      .withMetaAndKeyColsInValue();
+  private static final PhysicalSchema PHYSICAL_SCHEMA = PhysicalSchema.from(SCHEMA, SerdeOption.none());
+
+  private final List<Expression> groupByExpressions = ImmutableList.of(


nit: can be static, likely others too.

big-andy-coates · 2019-09-19T16:19:09Z

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

+  @Mock
+  private FunctionRegistry functionRegistry;
+  @Mock
+  private GroupedFactory groupedFactory;


raw types - can we please use generic types. Same for ExecutionStep

rodesai requested a review from a team as a code owner September 16, 2019 18:30

rodesai force-pushed the move-grouped-to-plan-builder-rebase branch from e39dfc9 to f6212f9 Compare September 16, 2019 18:32

big-andy-coates suggested changes Sep 17, 2019

View reviewed changes

rodesai force-pushed the move-grouped-to-plan-builder-rebase branch from ea95693 to 2ab4aa3 Compare September 18, 2019 07:46

rodesai requested a review from big-andy-coates September 18, 2019 07:54

rodesai force-pushed the move-grouped-to-plan-builder-rebase branch 3 times, most recently from 6c6fefc to 9b7d2a2 Compare September 19, 2019 04:51

feat: move groupBy into plan builders

d7579a0

This patch moves the code for regrouping streams/tables into plan builders. This also required adding a new execution step for groupByKey, which we missed the first go-round.

rodesai force-pushed the move-grouped-to-plan-builder-rebase branch from 9b7d2a2 to d7579a0 Compare September 19, 2019 05:04

big-andy-coates requested a review from a team September 19, 2019 15:55

big-andy-coates approved these changes Sep 19, 2019

View reviewed changes

feedback

f2fef45

rodesai merged commit 730c913 into confluentinc:master Sep 19, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: move groupBy into plan builders #3359

feat: move groupBy into plan builders #3359

rodesai commented Sep 16, 2019

big-andy-coates left a comment

big-andy-coates Sep 17, 2019

big-andy-coates Sep 17, 2019

big-andy-coates Sep 17, 2019

rodesai Sep 18, 2019

big-andy-coates Sep 17, 2019

big-andy-coates Sep 17, 2019

big-andy-coates Sep 17, 2019

big-andy-coates Sep 17, 2019

big-andy-coates Sep 17, 2019

big-andy-coates Sep 19, 2019

big-andy-coates Sep 17, 2019

big-andy-coates Sep 19, 2019

big-andy-coates left a comment

big-andy-coates Sep 19, 2019

big-andy-coates Sep 19, 2019

big-andy-coates Sep 19, 2019

big-andy-coates Sep 19, 2019

big-andy-coates Sep 19, 2019

	private final List<Expression> groupByExpressions = ImmutableList.of(
	private static final List<Expression> GROUP_BY_EXPRESSIONS = ImmutableList.of(

feat: move groupBy into plan builders #3359

feat: move groupBy into plan builders #3359

Conversation

rodesai commented Sep 16, 2019

Description

Testing done

Reviewer checklist

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

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 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