feat: build physical plan for FK table-table joins #7517

mjsax · 2021-05-12T21:53:41Z

Description

Add step to add FK-join to the physical plan.

Testing done

Added some unit testing. Later we need to extend QTT tests when we actually use this code (that is currently "deal code").

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

mjsax · 2021-05-12T21:56:13Z

ksqldb-streams/src/test/java/io/confluent/ksql/execution/streams/JoinParamsFactoryTest.java

@@ -28,15 +28,13 @@
      .valueColumn(ColumnName.of("R_K"), SqlTypes.STRING) // Copy of key in value
      .build();

-  private JoinParams joinParams;


side cleanup to inline variable.

mjsax · 2021-05-12T21:57:47Z

ksqldb-streams/src/test/java/io/confluent/ksql/execution/streams/TableTableJoinBuilderTest.java

@@ -193,7 +193,12 @@ public void shouldReturnCorrectSchema() {
    // Then:
    assertThat(
        result.getSchema(),
-        is(JoinParamsFactory.create(R_KEY, LEFT_SCHEMA, RIGHT_SCHEMA).getSchema())


Side cleanup: We should not use JoinParamsFactory to compute the expected result, because JoinParamsFactory is use to compute the actual result within join.build(...). Otherwise we have a cyclic tes "dependency"...

Same below.

spena · 2021-05-13T14:56:32Z

ksqldb-execution/src/main/java/io/confluent/ksql/execution/plan/ForeignKeyTableTableJoin.java

+  public ForeignKeyTableTableJoin(
+      final ExecutionStepPropertiesV1 props,
+      final JoinType joinType,
+      final ColumnName leftJoinColumnName,
+      final ExecutionStep<KTableHolder<KLeftT>> leftSource,
+      final ExecutionStep<KTableHolder<KRightT>> rightSource
+  ) {


I'm not sure if we should have two classes for each time of join. They have repeated code except the extra left column name.

Also, I'm not sure if the TableTableJoin info is written in the plan. I see that class has a @JsonCreator and it is also referenced in PlanSummary.java as .put(TableTableJoin.class, "JOIN"). Did you verify this part?

The TableTableJoin accepts an optional keyColumn. Perhaps we can add another parameter to accept an optional nonKeyColumn too?

I'm not sure if we should have two classes for each time of join. They have repeated code except the extra left column name.

I think it makes sense to have a separate execution step for FK joins since the implementation is different. Execution steps are meant to be fine-grained so we can evolve them separately as needed.

Also, I'm not sure if the TableTableJoin info is written in the plan. I see that class has a @JsonCreator and it is also referenced in PlanSummary.java as .put(TableTableJoin.class, "JOIN"). Did you verify this part?

PlanSummary.java, ExecutionStep.java, and StepSchemaResolver.java all need to be updated to include the new execution step. Let's add these changes into this PR so we don't hit confusing failures down the line when wiring up the functionality.

The TableTableJoin accepts an optional keyColumn. Perhaps we can add another parameter to accept an optional nonKeyColumn too?

Unless I'm mistaken, the optional keyColumn in TableTableJoin specifies the name of the output key column. That is not necessary for FK joins because the output key always matches the left source key.

spena · 2021-05-13T14:58:09Z

ksqldb-streams/src/main/java/io/confluent/ksql/execution/streams/ExecutionStepFactory.java

+  public static <KLeftT, KRightT> ForeignKeyTableTableJoin<KLeftT, KRightT>
+      foreignKeyTableTableJoin(final QueryContext.Stacker stacker,
+                               final JoinType joinType,
+                               final ColumnName leftJoinColumnName,


I think we could reuse the current tableTableJoin method by passing the new leftJoinColumnName as optional, don't you think?

If we're going to have a separate execution step, we should have a separate method for creating the execution step. (See above comment.)

spena · 2021-05-13T15:00:19Z

...reams/src/main/java/io/confluent/ksql/execution/streams/ForeignKeyTableTableJoinBuilder.java

+  public static <KLeftT, KRightT> KTableHolder<KLeftT> build(
+      final KTableHolder<KLeftT> left,
+      final KTableHolder<KRightT> right,
+      final ForeignKeyTableTableJoin<KLeftT, KRightT> join
+  ) {


What if we add info to the current TableTableJoin where we could check if the join is on a foreign key? Maybe we can reuse the TableTableJoinBuilder, and just change the impl. to call a different builder for foregign keys and another that has the current code.

We could, but it seems cleaner to have a dedicated node for FK-joins in the physical plan, as mentioned by Victory above.

vcrfxia

Thanks @mjsax . LGTM minus the comments inline.

vcrfxia · 2021-05-13T17:35:43Z

ksqldb-execution/src/main/java/io/confluent/ksql/execution/plan/ForeignKeyTableTableJoin.java

+  public ForeignKeyTableTableJoin(
+      final ExecutionStepPropertiesV1 props,
+      final JoinType joinType,
+      final ColumnName leftJoinColumnName,
+      final ExecutionStep<KTableHolder<KLeftT>> leftSource,
+      final ExecutionStep<KTableHolder<KRightT>> rightSource
+  ) {


I'm not sure if we should have two classes for each time of join. They have repeated code except the extra left column name.

I think it makes sense to have a separate execution step for FK joins since the implementation is different. Execution steps are meant to be fine-grained so we can evolve them separately as needed.

Also, I'm not sure if the TableTableJoin info is written in the plan. I see that class has a @JsonCreator and it is also referenced in PlanSummary.java as .put(TableTableJoin.class, "JOIN"). Did you verify this part?

PlanSummary.java, ExecutionStep.java, and StepSchemaResolver.java all need to be updated to include the new execution step. Let's add these changes into this PR so we don't hit confusing failures down the line when wiring up the functionality.

The TableTableJoin accepts an optional keyColumn. Perhaps we can add another parameter to accept an optional nonKeyColumn too?

Unless I'm mistaken, the optional keyColumn in TableTableJoin specifies the name of the output key column. That is not necessary for FK joins because the output key always matches the left source key.

vcrfxia · 2021-05-13T17:36:16Z

ksqldb-streams/src/main/java/io/confluent/ksql/execution/streams/ExecutionStepFactory.java

+  public static <KLeftT, KRightT> ForeignKeyTableTableJoin<KLeftT, KRightT>
+      foreignKeyTableTableJoin(final QueryContext.Stacker stacker,
+                               final JoinType joinType,
+                               final ColumnName leftJoinColumnName,


If we're going to have a separate execution step, we should have a separate method for creating the execution step. (See above comment.)

vcrfxia · 2021-05-13T17:38:30Z

...b-streams/src/main/java/io/confluent/ksql/execution/streams/ForeignKeyJoinParamsFactory.java

+  public static LogicalSchema createSchema(
+      final LogicalSchema leftSchema,
+      final LogicalSchema rightSchema
+  ) {


We need to validate that the join expression types match. If we follow the pattern of the existing JoinParamsFactory, we can do that here.

JoinParamsFactory also validates that the join expressions represent the complete key, which is a requirement for the right side of FK joins as well. I think we don't need this extra validation here in light of the recent updates to LogicalPlanner to check this condition up front, but it may make sense to have a duplicate check as a second guard in the physical layer here as well.

We need to validate that the join expression types match. If we follow the pattern of the existing JoinParamsFactory, we can do that here.

Yes, but this should be done in the logical planner IMHO.

Happy to add a second guard.

Where are we currently validating that the expression types match? I don't see the check in either the logical or the physical layer right now.

For table-table joins, it happens in JoinParamsFactory.throwOnKeyMismatch (but I think it's wrong to do this check when building the physical plan, but it should be done earlier.

Thus, I did not add it to ForeignKeyJoinParamsFactory and I think @spena current WIP PR should add this check. (And we should move the existing check for PK-joins in a follow up PR into the logical planner.)

JoinParamsFactory.throwOnKeyMismatch() is called from JoinParamsFactory.createSchema(), which is called from JoinNode.buildJoinSchema(), which is called as part of attempting to build the join from the logical planner. In other words, this check is already happening in the logical planner (for table-table joins today). Where are you proposing that the check should happen instead?

That is quite convoluted code than, because JoinParamsFactory is in package io.confluent.ksql.execution.streams -- the logical planner should not use helpers from the physical planner IMHO. Maybe the solution is to move the whole factory class into the logical planner package?

The method is also called from the physical layer when executing the execution step. Ideally this method would be in a separate package clearly meant to be shared between the two, rather than entirely in one or the other.

Interesting. Think we don't need to resolve this in this PR. We only need to make sure that @spena PR's adds the type check.

vcrfxia · 2021-05-13T17:40:23Z

...s/src/test/java/io/confluent/ksql/execution/streams/ForeignKeyTableTableJoinBuilderTest.java

+  }
+
+  @Test
+  public void shouldDoLeftJoinOnKey() {


What is the purpose of this test? Isn't this a primary key join, not a foreign key join?

I guess you are right -- seems I overshot on corner case testing. Even if we could compile a PK_PK-join as a FK-join and it should still work -- it would be less efficient though I guess.

This actually raises a question: if we have two tables with different partition count and/or different key format, could we implement a PK-PK-join without repartitioning one input table (or even rejecting the join do the missing co-partitioning), but compiling it down as FK-join?

Thus, I am wondering if we should just keep this test, because from a physical plan point of view, this should work, even if we don't use it when translating the logical plan into a physical plan atm -- but maybe we would do this in the future?

if we have two tables with different partition count and/or different key format, could we implement a PK-PK-join without repartitioning one input table (or even rejecting the join do the missing co-partitioning), but compiling it down as FK-join?

I suppose we could but it seems odd to do this by default, as it allows users to inadvertently write queries with poorer performance than expected. I'm not convinced we need these unit tests until (and if) we decide to do this one day, but I don't feel strongly. If we're going to leave them in, can we at least add a comment explaining that these are actually primary key joins which won't reach this code today, but we've left it in the future? Otherwise future readers may be similarly confused.

Agreed. Also not sure if we should do it, but from a unit-test point of view, it seem ok to keep the tests, as unit testing the physical plan should not know anything about the logical plan layer.

Will add a comment to clarify.

vcrfxia · 2021-05-13T17:40:30Z

...s/src/test/java/io/confluent/ksql/execution/streams/ForeignKeyTableTableJoinBuilderTest.java

+  }
+
+  @Test
+  public void shouldDoInnerJoinOnKey() {


Same question as above: What is the purpose of this test? Isn't this a primary key join, not a foreign key join?

vcrfxia · 2021-05-13T17:41:25Z

...s/src/test/java/io/confluent/ksql/execution/streams/ForeignKeyTableTableJoinBuilderTest.java

+    );
+  }
+
+  // not sure what "legacy schema" is and if we need to support it?


I don't think this is necessary. LEGACY_KEY_COL refers to the old default output key column name for joins. For FK joins, the output key columns always match the left source key columns.

vcrfxia · 2021-05-13T17:41:34Z

ksqldb-execution/src/main/java/io/confluent/ksql/execution/plan/ForeignKeyTableTableJoin.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2019 Confluent Inc.


nit: update year (and similarly for all the other new files in this PR)

vcrfxia · 2021-05-13T18:47:02Z

...b-streams/src/main/java/io/confluent/ksql/execution/streams/ForeignKeyJoinParamsFactory.java

+      final LogicalSchema rightSchema
+  ) {
+    final Builder builder = LogicalSchema.builder()
+        .keyColumns(leftSchema.key())


It occurs to me now that these column names might be aliased, which we probably don't want. Can we verify? Can also defer to a later PR when we wire everything together but my current suspicion is that this will be a problem.

Not 100% sure atm -- leftSchema.key()returns Column that only has a ColumnName but no "alias".

But yes, we need to keep aliases in mind... I think it might be best to defer to later.

guozhangwang

Made a pass over the non-testing part.

guozhangwang · 2021-05-13T21:04:45Z

...b-streams/src/main/java/io/confluent/ksql/execution/streams/ForeignKeyJoinParamsFactory.java

+      final LogicalSchema leftSchema,
+      final ColumnName leftJoinColumnName) {
+
+    final Optional<Column> leftJoinColumn = leftSchema.findValueColumn(leftJoinColumnName);


Just for my understanding, is it possible that due to alias we would not find a match here? I.e. if we do

... SELECT A.a AS a1, ... WHERE A.a = B.b

would we try to find a in a schema with a1 then?

Cf https://github.com/confluentinc/ksql/pull/7517/files#r632182509

guozhangwang · 2021-05-13T21:07:40Z

...reams/src/main/java/io/confluent/ksql/execution/streams/ForeignKeyTableTableJoinBuilder.java

+        throw new IllegalStateException("invalid join type");
+    }
+
+    return KTableHolder.unmaterialized(


Is it intentional to not materialize the join results?

It's was kinda intentional, because TableTableJoinBuilder() does the same thing. Not sure what the status of materializing table-table joins result is. \cc @cprasad1

We skip materializing the Table-Table join results immediately. We materialize at the TableSelectBuilder which comes after the TableTableJoinBuilder(). It makes sense to follow the same pattern and skip materializing here.

Specifically, we materialize here:

ksql/ksqldb-streams/src/main/java/io/confluent/ksql/execution/streams/TableSelectBuilder.java

Line 114 in 100767d

return KTableHolder.materialized(

Got it, thanks @cprasad1 for confirming!

guozhangwang · 2021-05-13T21:08:32Z

...reams/src/main/java/io/confluent/ksql/execution/streams/ForeignKeyTableTableJoinBuilder.java

+        );
+        break;
+      default:
+        throw new IllegalStateException("invalid join type");


nit: also include the type enum?

guozhangwang · 2021-05-13T21:13:18Z

ksqldb-streams/src/main/java/io/confluent/ksql/execution/streams/KsqlKeyExtractor.java

+import java.util.Objects;
+import java.util.function.Function;
+
+public final class KsqlKeyExtractor<KRightT> implements Function<GenericRow, KRightT> {


Just a question here: I saw different Extractor classes in the repo, and some with a Ksql prefix in its name and some do not. Is there a convention for the naming at the moment? cc @agavra @vcrfxia

🔥 ⚠️ - I really don't like Ksql prefix. I don't think we have a convention, so it's really just do what you like.

vcrfxia

LGTM besides the couple minor comments I just added. Thanks @mjsax !

mjsax added 2 commits May 11, 2021 22:16

feat: build physical plan for FK table-table joins

513e178

adding tests

43cd9fc

mjsax requested a review from a team as a code owner May 12, 2021 21:53

mjsax commented May 12, 2021

View reviewed changes

mjsax changed the title ~~Add fk joins~~ feat: build physical plan for FK table-table joins May 12, 2021

mjsax mentioned this pull request May 12, 2021

Support non-key joins. #4424

Closed

spena reviewed May 13, 2021

View reviewed changes

vcrfxia reviewed May 13, 2021

View reviewed changes

guozhangwang reviewed May 13, 2021

View reviewed changes

mjsax added 3 commits May 13, 2021 17:35

Github comments

44d8d65

formatting

267d59f

Github comment

b462cb7

vcrfxia approved these changes May 14, 2021

View reviewed changes

mjsax added 2 commits May 14, 2021 10:49

Github comments

78c9ff9

minor fix

81f0823

vcrfxia mentioned this pull request May 14, 2021

test: adds tests for foreign key joins #7528

Closed

2 tasks

mjsax merged commit 744fa36 into confluentinc:master May 14, 2021

mjsax deleted the add-fk-joins branch May 14, 2021 20:42

feat: build physical plan for FK table-table joins #7517

feat: build physical plan for FK table-table joins #7517

Conversation

mjsax commented May 12, 2021

Description

Testing done

Reviewer checklist

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

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

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

guozhangwang 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

vcrfxia left a comment

Choose a reason for hiding this comment