feat: move aggregations to plan builder #3391

rodesai · 2019-09-20T17:30:17Z

Description

This patch moves the code for building aggregations into a plan builder:

To express windowed aggregates, I've added a new execution step type
called StreamWindowedAggregate. Adding a new type ensures that when we
implement the visitor that builds the streams app, the handler for windowed
aggregations is type-safe.

I've moved the windowing pojos into KSQL execution, and split them off
from the AST. This way we can serialize these into the aggregation plan
nodes to express windows. I've also removed the code that builds
aggregations from these pojos and moved it into a visitor inside
StreamAggregateBuilder.

To implement window start/end projections, I had to move
WindowSelectMapper to ksql-execution. This also requires having
WindowSelectMapper own the start/end udaf names.

This patch also includes a refactor of AggregateNode to pass down the
aggregation function call expressions rather than the resolved aggregation
functions. The code for resolving the function call expressions against the
internal schema and building the aggregators, initializers, and undo
aggregators has been moved into a class called AggregateParams, which
the aggregate builders use to make the aggregation calls in streams.

The rest of the patch implements the actual aggregation from step builders.

How to review:

look at the changes to the windowing POJOs moved from the ast to ksql-execution
look at the changes to the windowstart and windowend udafs, which have been moved under ksql-execution
look at the changes to AggregateNode, which has been refactored. It now computes a list of Function expressions resolved to the internal schema rather than a map of index to our internal function class.
look at the new builders for converting from our plan to a streams topology for aggregates. These take the list of function expressions and convert them to the map expected by KudafAggregator. The builder for stream aggregates (StreamAggregateBuilder) also implements a visitor for computing windowed aggregates from a KsqlWindowExpression.

Testing done

Added unit tests for the new plan builders and AggregateParams.

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-20T17:31:52Z

ksql-execution/src/main/java/io/confluent/ksql/execution/windows/KsqlWindowExpression.java

+import java.util.Optional;
+
+@Immutable
+public abstract class KsqlWindowExpression extends Node {


Unfortunately this is showing up as a new class, but it's just KsqlWindowExpression from ksql-parser moved to ksql-execution and with the interface that builds the aggregated KTable removed.

rodesai · 2019-09-20T17:35:36Z

ksql-execution/src/main/java/io/confluent/ksql/execution/function/UdafUtil.java

+  }
+
+  @SuppressWarnings("deprecation") // Need to migrate away from Connect Schema use.
+  public static KsqlAggregateFunction resolveAggregateFunction(


This is just AggregateNode.getAggregateFunction.

vcrfxia

Thanks @rodesai ! LGTM with the usual batch of questions and nits inline.

vcrfxia · 2019-09-24T18:32:19Z

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

  private final ExecutionStepProperties properties;
-  private final ExecutionStep<G> source;


Why did this used to be generic? Looks like the new StreamWindowedAggregate has the same KGroupedStream<Struct, GenericRow>.

I changed up how the type parameters work in an earlier PR. Before, the input and output types were type parameters, and you had to specify one here (either G or T). Now we just have the step implementations implement ExecutionStep with the type parameter fully specified. For stream aggregates, both StreamAggregate and StreamWindowedAggregate take an ExecutionStep<KGroupedStream<Struct, GenericRow>> as the source. But the former implements ExecutionStep<KTable<Struct, GenericRow>> and the latter implements ExecutionStep<KTable<Windowed<Struct>, GenericRow>>.

vcrfxia · 2019-09-24T18:35:53Z

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

+        && Objects.equals(source, that.source)
+        && Objects.equals(formats, that.formats)
+        && Objects.equals(aggregations, that.aggregations)
+        && nonFuncColumnCount == that.nonFuncColumnCount;


Why are aggregationSchema and windowExpression not part of this method (or hashCode() below)?

vcrfxia · 2019-09-24T18:37:08Z

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

    this.properties = Objects.requireNonNull(properties, "properties");
    this.source = Objects.requireNonNull(source, "source");
    this.formats = Objects.requireNonNull(formats, "formats");
    this.nonFuncColumnCount = nonFuncColumnCount;
-    this.aggregations = Objects.requireNonNull(aggregations);
+    this.aggregations = Objects.requireNonNull(aggregations, "aggregations");
+    this.aggregationSchema = Objects.requireNonNull(aggregationSchema, "aggregationSchema");


Why was this not added to equals() or hashCode()? Same question for TableAggregate as well.

vcrfxia · 2019-09-24T18:39:37Z

ksql-engine/src/main/java/io/confluent/ksql/structured/SchemaKGroupedStream.java

+          windowExpression.getKsqlWindowExpression()
+      );
+      step = aggregate;
+      table = StreamAggregateBuilder.build(


nit: pull this out of the if-else block?

It needs to be it there - we build a windowed vs not windowed aggregate based on the if condition.

vcrfxia · 2019-09-24T19:10:56Z

ksql-engine/src/test/java/io/confluent/ksql/function/udaf/window/WindowSelectMapperTest.java

@@ -22,6 +22,7 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import io.confluent.ksql.GenericRow;


How come this wasn't moved to ksql-execution along with WindowSelectMapper?

vcrfxia · 2019-09-24T19:28:33Z

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

+import io.confluent.ksql.GenericRow;
+import io.confluent.ksql.execution.builder.KsqlQueryBuilder;
+import io.confluent.ksql.execution.context.QueryContext;
+import io.confluent.ksql.execution.expression.tree.FunctionCall;


(fyi: unused imports that are being caught by checkstyle in the build)

vcrfxia · 2019-09-24T19:28:53Z

ksql-parser/src/test/java/io/confluent/ksql/parser/tree/HoppingWindowExpressionTest.java

@@ -1,4 +1,4 @@
-/*
+ /*


super nit: white-space change.

vcrfxia · 2019-09-24T19:29:51Z

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

+    // When:
+    final WindowSelectMapper windowSelectMapper = aggregateParams.getWindowSelectMapper();
+    final Windowed<?> window = new Windowed<>(null, new TimeWindow(10, 20));
+    assertThat(


super nit: // Then:

big-andy-coates

Thanks @rodesai,

I've not looked in detail, but LGTM. Few nits beyond what @vcrfxia called out

big-andy-coates · 2019-09-25T07:20:17Z

ksql-engine/src/main/java/io/confluent/ksql/planner/plan/AggregateNode.java

+        .map(
+            fc -> new FunctionCall(
+                fc.getName(),
+                internalSchema.getInternalArgsExpressionList(fc.getArguments())


Ideally.... we should be looking to move the whole internal schema thing into the physical layer. The logical layer shouldn't need to know about such things. Though such internal names should be part of the serialized form of the physical.

I agree it shouldn't be in the logical plan node (like the other code that builds the execution plan). But the code that converts from the physical plan to the streams app shouldn't have to worry about it either. I think when we have a visitor that traverses the logical plan to build the physical plan, that would be the right place for it.

big-andy-coates · 2019-09-25T07:21:54Z

ksql-engine/src/main/java/io/confluent/ksql/planner/plan/AggregateNode.java

-      } else {
-        fieldType = converter.toSqlType(aggregateFunction.getReturnType());
-      }
+    for (int i = 0; i < aggregations.size(); i++) {


Maybe we could move this building of schema, and other places we build schemas, into the QueryAnalyzer / Analyzer. What do you think?

Yeah that makes sense to me for a follow-up.

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

ksql-engine/src/main/java/io/confluent/ksql/structured/SchemaKGroupedStream.java

@@ -122,54 +116,61 @@ public KeyField getKeyField() {
  @SuppressWarnings("unchecked")
  public SchemaKTable<?> aggregate(
      final LogicalSchema outputSchema,
-      final Initializer initializer,
+      final LogicalSchema aggregateSchema,


super nit: can we have aggregateSchema before outputSchema in the param list, given that the aggregation happens before the output?

big-andy-coates · 2019-09-25T12:44:24Z

ksql-engine/src/main/java/io/confluent/ksql/structured/SchemaKGroupedTable.java

@@ -109,24 +105,24 @@
  @Override
  public SchemaKTable<Struct> aggregate(
      final LogicalSchema outputSchema,
-      final Initializer initializer,
+      final LogicalSchema aggregateSchema,


as above, can we switch the param overs?

big-andy-coates · 2019-09-25T12:46:20Z

ksql-execution/src/main/java/io/confluent/ksql/execution/function/udaf/KudafInitializer.java

  private final int nonAggValSize;

-  public KudafInitializer(final int nonAggValSize) {
+  public KudafInitializer(final int nonAggValSize, final List<Supplier> aggValueSuppliers) {


nit: can we call this param and the field by what it actually is... which is the initialValuesSupplier, right?

big-andy-coates · 2019-09-25T12:50:56Z

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

+public final class AggregateParams {
+  private final KudafInitializer initializer;
+  private final int initialUdafIndex;
+  private final Map<Integer, KsqlAggregateFunction> indexToFunction;


Rather than a Map I would suggest an ordered List< KsqlAggregateFunction>.

See my PR: #3409

The indexes in the map are always sequential, starting at initialUdafIndex.

Agreed. Was trying (and failing) to keep this diff as small as possible :)

This patch moves the code for building aggregations into a plan builder: To express windowed aggregates, I've added a new execution step type called StreamWindowedAggregate. Adding a new type ensures that when we implement the visitor that builds the streams app, the handler for windowed aggregations is type-safe. I've moved the windowing pojos into KSQL execution, and split them off from the AST. This way we can serialize these into the aggregation plan nodes to express windows. I've also removed the code that builds aggregations from these pojos and moved it into a visitor inside StreamAggregateBuilder. To implement window start/end projections, I had to move WindowSelectMapper to ksql-execution. This also requires having WindowSelectMapper own the start/end udaf names. This patch also includes a refactor of AggregateNode to pass down the aggregation function call expressions rather than the resolved aggregation functions. The code for resolving the function call expressions against the internal schema and building the aggregators, initializers, and undo aggregators has been moved into a class called AggregateParams, which the aggregate builders use to make the aggregation calls in streams. The rest of the patch implements the actual aggregation from step builders.

rodesai requested a review from a team as a code owner September 20, 2019 17:30

rodesai commented Sep 20, 2019

View reviewed changes

vcrfxia approved these changes Sep 24, 2019

View reviewed changes

vcrfxia requested a review from a team September 24, 2019 19:47

rodesai force-pushed the move-aggregation-to-plan-builder-rebase branch from 08c031a to f0fead2 Compare September 24, 2019 21:23

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

View reviewed changes

rodesai force-pushed the move-aggregation-to-plan-builder-rebase branch from f0fead2 to 58cae3e Compare September 25, 2019 22:30

rodesai merged commit 3aaeb73 into confluentinc:master Sep 25, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: move aggregations to plan builder #3391

feat: move aggregations to plan builder #3391

rodesai commented Sep 20, 2019 •

edited

Loading

rodesai Sep 20, 2019

rodesai Sep 20, 2019

vcrfxia left a comment

vcrfxia Sep 24, 2019

rodesai Sep 24, 2019

vcrfxia Sep 24, 2019

vcrfxia Sep 24, 2019

vcrfxia Sep 24, 2019

rodesai Sep 24, 2019

vcrfxia Sep 24, 2019

vcrfxia Sep 24, 2019

vcrfxia Sep 24, 2019

vcrfxia Sep 24, 2019

big-andy-coates left a comment

big-andy-coates Sep 25, 2019

rodesai Sep 25, 2019

big-andy-coates Sep 25, 2019

rodesai Sep 25, 2019

big-andy-coates Sep 25, 2019

big-andy-coates Sep 25, 2019

big-andy-coates Sep 25, 2019

big-andy-coates Sep 25, 2019

rodesai Sep 25, 2019

		private final ExecutionStepProperties properties;
		private final ExecutionStep<G> source;

feat: move aggregations to plan builder #3391

feat: move aggregations to plan builder #3391

Conversation

rodesai commented Sep 20, 2019 • edited Loading

Description

Testing done

Reviewer checklist

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vcrfxia 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

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

rodesai commented Sep 20, 2019 •

edited

Loading