feat: allow expressions on left table columns in FK-joins #7904

mjsax · 2021-07-27T23:13:50Z

Description

Currently, FK-join require to use a plain column reference for the left join expression. We want to lift this restriction and allow for actual expression in the left join expression.

Testing done

Unit tests and new QTT tests.

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-07-27T23:17:52Z

ksqldb-engine/src/main/java/io/confluent/ksql/planner/LogicalPlanner.java

+              final QualifiedColumnReferenceExp node,
+              final Context<Void> ctx
+          ) {
+            return Optional.of(new UnqualifiedColumnReferenceExp(node.getColumnName()));


The ExpressionEvaluator cannot handle fully qualified column references, thus we rewrite the given expression accordingly.

I had a sync with @mjsax offline about this specifically, and our thoughts are that since the codeGenRunner is used for physical plan generation as well at which stage there's no qualified column refs since we would just prefix the names as table/stream__column, is that right @vcrfxia ??

If yes, just as a general note here I'm feeling that we should use a separate class than codeGenRunner for traversals in the logical/physical plan generation respectively, than reusing it, since for logical planning all we really need are return type / param types etc, while nevertheless today we always run the same full-fledged traversal multiple times at different stages. Is my understanding reasonable?

mjsax · 2021-07-27T23:22:39Z

...b-execution/src/test/java/io/confluent/ksql/execution/plan/ForeignKeyTableTableJoinTest.java

@@ -54,26 +55,26 @@
  public void shouldImplementEquals() {
    new EqualsTester()
        .addEqualityGroup(
-            new ForeignKeyTableTableJoin<>(props1, INNER, JOIN_COLUMN_NAME, formats1, left1, right1),
-            new ForeignKeyTableTableJoin<>(props1, INNER, JOIN_COLUMN_NAME, formats1, left1, right1)
+            new ForeignKeyTableTableJoin<>(props1, INNER, JOIN_COLUMN_NAME, formats1, left1, right1, null),


Open TODO...

mjsax · 2021-07-27T23:23:22Z

...reams/src/test/java/io/confluent/ksql/execution/streams/ForeignKeyJoinParamsFactoryTest.java

@@ -48,7 +51,7 @@ public void shouldBuildCorrectKeyedSchema() {

    // When:
    final ForeignKeyJoinParams<String> joinParams =
-        ForeignKeyJoinParamsFactory.create(leftJoinColumnName, LEFT_SCHEMA, RIGHT_SCHEMA);
+        ForeignKeyJoinParamsFactory.create(Optional.of(leftJoinColumnName), Optional.empty(), LEFT_SCHEMA, RIGHT_SCHEMA, null);


mjsax · 2021-07-27T23:23:30Z

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

@@ -136,26 +136,26 @@ public void init() {
    );
  }

-  @Test


mjsax · 2021-07-27T23:28:24Z

ksqldb-engine/src/main/java/io/confluent/ksql/planner/LogicalPlanner.java

-            columnRef.maybeQualifier().get(),
-            columnRef.getColumnName())
-        : columnRef.getColumnName();
+    final VisitParentExpressionVisitor<Optional<Expression>, Context<Void>> aliasRewritter =


Similar to the code we replace, we need to rewrite the expression with the alias prefix...

mjsax · 2021-07-28T03:51:39Z

Needed to update the generated schemas... I think it's safe. Do you agree @vcrfxia ?

vcrfxia

Thanks @mjsax -- LGTM with some questions/comments inline. It'd also be nice to add a test for mismatched types involving expressions (ensure that the error message is sensible) and another test with expressions including qualified columns.

vcrfxia · 2021-07-28T07:16:15Z

ksqldb-engine/src/main/java/io/confluent/ksql/structured/SchemaKTable.java

      final Stacker contextStacker,
-      final FormatInfo valueFormatInfo
+      final FormatInfo valueFormatInfo,
+      final Optional<Expression> expression


nit: rename to leftJoinExpression to be more informative (and parallel leftJoinColumnName above)? Same for the other occurrences.

vcrfxia · 2021-07-28T17:41:08Z

ksqldb-engine/src/main/java/io/confluent/ksql/planner/LogicalPlanner.java

+              final Context<Void> ctx
+          ) {
+            return Optional.of(new UnqualifiedColumnReferenceExp(
+                ColumnNames.generatedJoinColumnAlias(node.getQualifier(), node.getColumnName())


I'm not following the logic with qualifiers here. The old code only sometimes generated the join column alias (if a qualifier is present) but this new code seems to always generate the join column alias (before rewriting the alias to be unqualified). Is this a change in behavior?

I think it not a behavior change, because we use the visiter pattern and this method is only executed for qualified column references (its name is visitQualifiedColumnReference). Thus unqualified column references should be untouched, as before.

vcrfxia · 2021-07-28T17:41:39Z

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

  ) {
    this.properties = requireNonNull(props, "props");
    this.joinType = requireNonNull(joinType, "joinType");
    if (joinType == JoinType.OUTER) {
      throw new IllegalArgumentException("OUTER join not supported.");
    }
-    this.leftJoinColumnName = requireNonNull(leftJoinColumnName, "leftJoinColumnName");


We can still require non-null for Optional types. It'd be nice to add this back, and also add it for leftJoinExpression below.

vcrfxia · 2021-07-28T17:41:53Z

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

          final Formats formats,
          final ExecutionStep<KTableHolder<KLeftT>> left,
-          final ExecutionStep<KTableHolder<KRightT>> right
+          final ExecutionStep<KTableHolder<KRightT>> right,
+          final Optional<Expression> expression


nit: rename to leftJoinExpression to be more informative?

vcrfxia · 2021-07-28T17:43:13Z

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

+          )
+      );
+    } else {
+      expressionEvaluator = Optional.empty();


Instead of having two separate cases here, can we combine them by always passing an expression evaluator in order to simplify the code? The expression evaluator for the single column case would be simple, just extracting the appropriate column.

vcrfxia · 2021-07-28T17:43:32Z

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

  KsqlKeyExtractor(final int leftJoinColumnIndex) {
    checkArgument(
        leftJoinColumnIndex >= 0,
        "leftJoinColumnIndex negative: " + leftJoinColumnIndex
    );

    this.leftJoinColumnIndex = leftJoinColumnIndex;
+    expressionEvaluator = Optional.empty();


As above, it'd be nice if we could simplify this code by always using an expression evaluator.

vcrfxia · 2021-07-28T17:44:54Z

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

-    );
+    if (leftJoinColumnName.isPresent()) {
+      return new ForeignKeyJoinParams<>(
+          createKeyExtractor(leftSchema, leftJoinColumnName.get()),


nit: I think it's slightly more readable to create the foreign key extractor in an if-statement and then have a single return statement for building the ForeignKeyJoinParams. Also less code duplication.

mjsax · 2021-07-28T21:05:26Z

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

    // Then:
    verify(leftKTable).leftJoin(
        same(rightKTable),
-        eq(new KsqlKeyExtractor<>(1)),
+        any(KsqlKeyExtractor.class),


@vcrfxia After the refactoring to always use ExpressionEvaluator, it pretty hard to verify that KsqlKeyExtractor is created correctly now... I just simplified this test to make it pass for now. But this change kinda make the test useless... Thoughts?

If we really want to test this, we can use an ArgumentCaptor to capture the KsqlKeyExtractor, add a getter for the expression evaluator (visible for testing only), get the expression from the expression evaluator, and verify that the expression is a column reference expression for the appropriate column. I don't think this is too much effort if we wrap it in a helper method.

As you pointed out in the first version of this PR, test coverage for this (and some of the other unit tests) is lacking at the moment since we only test single columns for the left join expression, and not any non-column expressions. If we want to extend test coverage to include non-column expressions, I think we'll have to do something similar anyway. (I'm happy deferring this to a follow-up PR if you prefer since I know you've got some time constraints.)

Oh nice! #gold

vcrfxia

Thanks @mjsax ! The new QTT coverage is great. Love the refactor too 🎉

vcrfxia · 2021-07-28T22:52:43Z

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

    // Then:
    verify(leftKTable).leftJoin(
        same(rightKTable),
-        eq(new KsqlKeyExtractor<>(1)),
+        any(KsqlKeyExtractor.class),


If we really want to test this, we can use an ArgumentCaptor to capture the KsqlKeyExtractor, add a getter for the expression evaluator (visible for testing only), get the expression from the expression evaluator, and verify that the expression is a column reference expression for the appropriate column. I don't think this is too much effort if we wrap it in a helper method.

As you pointed out in the first version of this PR, test coverage for this (and some of the other unit tests) is lacking at the moment since we only test single columns for the left join expression, and not any non-column expressions. If we want to extend test coverage to include non-column expressions, I think we'll have to do something similar anyway. (I'm happy deferring this to a follow-up PR if you prefer since I know you've got some time constraints.)

vcrfxia · 2021-07-28T22:52:58Z

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

+  KsqlKeyExtractor(final ExpressionEvaluator expressionEvaluator,
+                   final ProcessingLogger processingLogger) {
+    this.expressionEvaluator = requireNonNull(expressionEvaluator);
+    this.processingLogger = processingLogger;


Any reason we don't require this to be non-null?

We only "forward" it to expressionEvaluator.evaluate() but don't use it ourselves, so I though we don't need this (makes testing simpler as we can pass null :))

vcrfxia · 2021-07-28T22:55:57Z

Needed to update the generated schemas... I think it's safe. Do you agree @vcrfxia ?

Assuming we're talking about the execution step/plan schemas, I agree this is safe. The updates reflect the changes to the ForeignKeyTableTableJoin execution step.

guozhangwang

I do not have any major comments, just a question about the tech debt we have for using the same codeGenRunner for logical / physical planning.

guozhangwang · 2021-07-28T22:35:55Z

ksqldb-engine/src/main/java/io/confluent/ksql/structured/SchemaKTable.java

@@ -321,9 +321,10 @@ public SchemaKGroupedTable groupBy(

  public <KRightT> SchemaKTable<K> foreignKeyInnerJoin(
      final SchemaKTable<KRightT> schemaKTable,
-      final ColumnName leftJoinColumnName,
+      final Optional<ColumnName> leftJoinColumnName,


nit: move the new param as the third param? Ditto below.

guozhangwang · 2021-07-28T22:38:35Z

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

@@ -643,10 +645,10 @@ static JoinKey syntheticColumn() {
    }

    static JoinKey foreignKeyColumn(
-        final ColumnName foreignKeyColumn,
+        final Expression foreignKeyExpression,


Should we just name the function as joinKey(..)?

Let's got with foreignKey()

guozhangwang · 2021-07-28T22:47:20Z

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

    private final ImmutableList<? extends ColumnReferenceExp> leftSourceKeyColumns;

    static JoinKey of(final ColumnName foreignKeyColumn,
                      final Collection<QualifiedColumnReferenceExp> leftSourceKeyColumns) {
-      return new ForeignJoinKey(foreignKeyColumn, leftSourceKeyColumns);
+      return new ForeignJoinKey(


This static construct seems not used any more.

Good catch!

guozhangwang · 2021-07-28T22:57:07Z

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

      @JsonProperty(value = "formats", required = true)
      final Formats formats,
      @JsonProperty(value = "leftSource", required = true)
      final ExecutionStep<KTableHolder<KLeftT>> leftSource,
      @JsonProperty(value = "rightSource", required = true)
-      final ExecutionStep<KTableHolder<KRightT>> rightSource
+      final ExecutionStep<KTableHolder<KRightT>> rightSource,


Also pretty nit: could we add the new param after leftJoinColumnName to group them together? Similar for other classes with the new param in constructors.

guozhangwang · 2021-07-28T23:08:32Z

...in_-_Should_support_inner_join_with_left_key-column_expression/7.1.0_1627438328984/plan.json

+    }
+  }, {
+    "@type" : "ksqlPlanV1",
+    "statementText" : "CREATE TABLE OUTPUT AS SELECT\n  LEFT_TABLE.L_ID L_ID,\n  LEFT_TABLE.L_ID_2_FOREIGN_KEY L_ID_2_FOREIGN_KEY,\n  RIGHT_TABLE.R_ID R_ID,\n  LEFT_TABLE.NAME NAME,\n  RIGHT_TABLE.F1 F1\nFROM LEFT_TABLE LEFT_TABLE\nINNER JOIN RIGHT_TABLE RIGHT_TABLE ON ((LEFT_TABLE.L_ID_2_FOREIGN_KEY = RIGHT_TABLE.R_ID))\nEMIT CHANGES",


Seems the left key is not an expression? Did I miss anything?

Those are historic plans that are auto-generated...

guozhangwang · 2021-07-28T23:12:45Z

..._-_Should_support_inner_join_with_left_value-column_expression/7.1.0_1627438328037/plan.json

+    }
+  }, {
+    "@type" : "ksqlPlanV1",
+    "statementText" : "CREATE TABLE OUTPUT AS SELECT\n  LEFT_TABLE.L_ID L_ID,\n  RIGHT_TABLE.R_ID R_ID,\n  LEFT_TABLE.NAME NAME,\n  RIGHT_TABLE.F1 F1\nFROM LEFT_TABLE LEFT_TABLE\nINNER JOIN RIGHT_TABLE RIGHT_TABLE ON ((LEFT_TABLE.FOREIGN_KEY = RIGHT_TABLE.R_ID))\nEMIT CHANGES",


Same here..?

guozhangwang · 2021-07-29T00:04:04Z

ksqldb-engine/src/main/java/io/confluent/ksql/planner/LogicalPlanner.java

+              final QualifiedColumnReferenceExp node,
+              final Context<Void> ctx
+          ) {
+            return Optional.of(new UnqualifiedColumnReferenceExp(node.getColumnName()));


I had a sync with @mjsax offline about this specifically, and our thoughts are that since the codeGenRunner is used for physical plan generation as well at which stage there's no qualified column refs since we would just prefix the names as table/stream__column, is that right @vcrfxia ??

If yes, just as a general note here I'm feeling that we should use a separate class than codeGenRunner for traversals in the logical/physical plan generation respectively, than reusing it, since for logical planning all we really need are return type / param types etc, while nevertheless today we always run the same full-fledged traversal multiple times at different stages. Is my understanding reasonable?

mjsax · 2021-07-29T00:59:28Z

Merging this PR as-is, and do a follow up PR to address the open comments.

Follow up to confluentinc#7904

mjsax · 2021-07-29T01:07:49Z

Follow up PR: #7916

Follow up to #7904

mjsax requested review from spena, guozhangwang and vcrfxia July 27, 2021 23:13

mjsax requested a review from a team as a code owner July 27, 2021 23:13

mjsax commented Jul 27, 2021

View reviewed changes

mjsax added 3 commits July 27, 2021 17:37

feat: allow expressions on left table columns in FK-joins

675f27f

Move code

68dedab

foo

67cb736

mjsax force-pushed the fk-join-expression branch from 4db756a to 67cb736 Compare July 28, 2021 01:22

mjsax added 2 commits July 27, 2021 19:23

add historic plans

31b6634

update generated schemas

a94f92a

vcrfxia reviewed Jul 28, 2021

View reviewed changes

mjsax added 2 commits July 28, 2021 11:48

Github comments

36446c8

Github comments

8a60032

mjsax commented Jul 28, 2021

View reviewed changes

mjsax added 4 commits July 28, 2021 14:10

improve equals impl

fe74e4f

fix test

4ae9ecf

fix spotbug warning

2afe4dd

fix spotbug warning

e6554a2

vcrfxia approved these changes Jul 28, 2021

View reviewed changes

guozhangwang approved these changes Jul 29, 2021

View reviewed changes

mjsax merged commit a9668de into confluentinc:master Jul 29, 2021

mjsax deleted the fk-join-expression branch July 29, 2021 01:05

mjsax added a commit to mjsax/ksql that referenced this pull request Jul 29, 2021

chore: address open PR comments

372ea54

Follow up to confluentinc#7904

mjsax mentioned this pull request Jul 29, 2021

chore: FK join expression support follow-ups #7916

Merged

2 tasks

vcrfxia pushed a commit that referenced this pull request Jul 29, 2021

chore: FK join expression support follow-ups (#7916)

acfc72d

Follow up to #7904

@@ @@ -136,26 +136,26 @@ public void init() { @@
                   );
                 }
-                @Test

feat: allow expressions on left table columns in FK-joins #7904

feat: allow expressions on left table columns in FK-joins #7904

Conversation

mjsax commented Jul 27, 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

mjsax commented Jul 28, 2021

vcrfxia left a comment • edited

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

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

mjsax Jul 28, 2021 • edited

Choose a reason for hiding this comment

vcrfxia commented Jul 28, 2021

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

mjsax commented Jul 29, 2021

mjsax commented Jul 29, 2021

vcrfxia left a comment •

edited

mjsax Jul 28, 2021 •

edited