feat: build materializations from the physical plan #3494

rodesai · 2019-10-08T08:32:38Z

This patch changes up how we build materializations to use the
physical plan instead of building from the logical plan. A
materialization is the data in the state store with some additional
transformations applied. These transformations should be the same
as those applied by the steps in the physical plan following the
aggregation step. This patch builds up the transformation as we
build the streams app from the physical plan.

Transformations are described in MaterializationInfo, and
accumulated by a Builder class. A transformation allows mapping
a key/row pair to an optional row. The row is optional to support
filters. We currently support 3 types of transformations:

aggregate result map: mapping the results of complex aggregates
(e.g. avg)
projection
filter

Finally, the materialization code has been moved to the ksql-streams
module so that it can be used by the streams app builder.

How to review:

Look at MaterializationInfo, which has been changed to support
generic transformations, and includes the 3 types of transforms.
Look at KsqlMaterializationFactory to see how the transform descriptors
are built into a list of transforms.
Look at KsqlMaterialization to see how the transforms are applied.
Look at KsqlPlanBuilder to see how we build up the transforms as we
build the streams app from the physical plan.

big-andy-coates

Thanks @rodesai, this is awesome.

Few nits and suggestions below.

big-andy-coates · 2019-10-08T12:59:41Z

...execution/src/main/java/io/confluent/ksql/execution/materialization/MaterializationInfo.java

+  ) {
+    this.stateStoreName = requireNonNull(stateStoreName, "stateStoreName");
+    this.stateStoreSchema = requireNonNull(stateStoreSchema, "stateStoreSchema");
+    this.transforms = requireNonNull(transforms, "transforms");


defensive copy:

Suggested change

this.transforms = requireNonNull(transforms, "transforms");

this.transforms = ImmutableList.copyOf(requireNonNull(transforms, "transforms"));

big-andy-coates · 2019-10-08T13:10:11Z

ksql-engine/src/main/java/io/confluent/ksql/query/QueryExecutor.java

@@ -181,18 +182,24 @@ public TransientQueryMetadata buildTransientQuery(
    );
  }

+  private Optional<MaterializationInfo> getMaterializationInfo(final Object result) {


nit:static?

big-andy-coates · 2019-10-08T13:11:20Z

...execution/src/main/java/io/confluent/ksql/execution/materialization/MaterializationInfo.java

+   * @param schema the schema of the result of applying transformations to the data in the store
+   * @return instance.
+   */
+  public static MaterializationInfo of(


No longer used.

big-andy-coates · 2019-10-08T13:12:35Z

...execution/src/main/java/io/confluent/ksql/execution/materialization/MaterializationInfo.java

+    private final List<TransformInfo> transforms;
+    private LogicalSchema schema;
+
+    public Builder(final String stateStoreName, final LogicalSchema stateStoreSchema) {


Consider a private constructor and having a public Builder builder(final String stateStoreName, final LogicalSchema stateStoreSchema) method on MaterializationInfo.

big-andy-coates · 2019-10-08T13:15:05Z

...execution/src/main/java/io/confluent/ksql/execution/materialization/MaterializationInfo.java

+      this.schema = stateStoreSchema;
+    }
+
+    public Builder mapAggregates(final AggregatesInfo aggregatesInfo, final LogicalSchema schema) {


renaming schema param to something that doesn't clash with the schema field would make this code more readable, e.g.

Suggested change

public Builder mapAggregates(final AggregatesInfo aggregatesInfo, final LogicalSchema schema) {

public Builder mapAggregates(final AggregatesInfo aggregatesInfo, final LogicalSchema aggregatedSchema) {

big-andy-coates · 2019-10-08T13:17:43Z

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

+        Objects.requireNonNull(materializationBuilder, "materializationProvider");
+  }
+
+  public KTableHolder(


Do we need the overloaded constructor? Such constructors can make it hard for people to know which to call. Consider just inlining this constructor.

big-andy-coates · 2019-10-08T13:22:33Z

ksql-streams/pom.xml

@@ -53,6 +53,11 @@
            <artifactId>connect-runtime</artifactId>
        </dependency>

+        <dependency>
+            <groupId>io.confluent.support</groupId>
+            <artifactId>support-metrics-common</artifactId>


I'm not sure it's worth the dependency just for a DI Clock type. Maybe just use Supplier<Long> or add a Clock type to our common module...

big-andy-coates · 2019-10-08T13:31:23Z

...s/src/main/java/io/confluent/ksql/execution/streams/materialization/KsqlMaterialization.java

    this.schema = requireNonNull(schema, "schema");
+    this.transforms = Objects.requireNonNull(transforms, "transforms");


defensive copy:

Suggested change

this.transforms = Objects.requireNonNull(transforms, "transforms");

this.transforms = ImmutableList.copyOf(requireNonNull(transforms, "transforms"));

big-andy-coates · 2019-10-08T13:33:28Z

...ain/java/io/confluent/ksql/execution/streams/materialization/KsqlMaterializationFactory.java

+          info.getInfo(),
+          functionRegistry
+      );
+      return (s, g) -> Optional.of(mapper.apply(g));


nit: s and g are the key and value, which would more normally be represented with k and v, i.e. name them after what they are, not what type they are.

big-andy-coates · 2019-10-08T13:37:27Z

...execution/src/main/java/io/confluent/ksql/execution/materialization/MaterializationInfo.java

+      return this;
+    }
+
+    public Builder mapValues(


consider naming this inline with what it actually is, i.e. a projection. This then marries up with the project operation name used in the processing logger when handling this transform.

To put it another way ... this is specifically a projection, not a general mapValues, so the name should reflect that IMHO.

Maybe project or selectColumns or some such?

This patch changes up how we build materializations to use the physical plan instead of building from the logical plan. A materialization is the data in the state store with some additional transformations applied. These transformations should be the same as those applied by the steps in the physical plan following the aggregation step. This patch builds up the transformation as we build the streams app from the physical plan. Transformations are described in MaterializationInfo, and accumulated by a Builder class. A transformation allows mapping a key/row pair to an optional row. The row is optional to support filters. We currently support 3 types of transformations: - aggregate result map: mapping the results of complex aggregates (e.g. avg) - projection - filter Finally, the materialization code has been moved to the ksql-streams module so that it can be used by the streams app builder.

rodesai requested a review from a team as a code owner October 8, 2019 08:32

rodesai requested a review from big-andy-coates October 8, 2019 08:32

big-andy-coates approved these changes Oct 8, 2019

View reviewed changes

rodesai force-pushed the build-materialization-from-plan branch from 0880f66 to 44f6d86 Compare October 9, 2019 00:55

rodesai added 2 commits October 8, 2019 19:25

Review feedback

fa156eb

rodesai force-pushed the build-materialization-from-plan branch from 44f6d86 to fa156eb Compare October 9, 2019 02:25

rodesai merged commit f45d649 into confluentinc:master Oct 9, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: build materializations from the physical plan #3494

feat: build materializations from the physical plan #3494

rodesai commented Oct 8, 2019

big-andy-coates left a comment

big-andy-coates Oct 8, 2019

big-andy-coates Oct 8, 2019

big-andy-coates Oct 8, 2019

big-andy-coates Oct 8, 2019

big-andy-coates Oct 8, 2019

big-andy-coates Oct 8, 2019

big-andy-coates Oct 8, 2019

big-andy-coates Oct 8, 2019

big-andy-coates Oct 8, 2019

big-andy-coates Oct 8, 2019

	this.transforms = requireNonNull(transforms, "transforms");
	this.transforms = ImmutableList.copyOf(requireNonNull(transforms, "transforms"));

	public Builder mapAggregates(final AggregatesInfo aggregatesInfo, final LogicalSchema schema) {
	public Builder mapAggregates(final AggregatesInfo aggregatesInfo, final LogicalSchema aggregatedSchema) {

		this.schema = requireNonNull(schema, "schema");
		this.transforms = Objects.requireNonNull(transforms, "transforms");

feat: build materializations from the physical plan #3494

feat: build materializations from the physical plan #3494

Conversation

rodesai commented Oct 8, 2019

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