chore: force repartition on joins with SR-enabled key formats #6635

vcrfxia · 2020-11-17T20:23:11Z

Description

This PR forces repartitions on both sources of joins with SR-enabled key formats. This is in order to ensure that the same Schema Registry schema is used on both sides of the join, so the user will not experience unexpected join misses due to logically equivalent data being sent to different topic partitions because the serialized bytes differ (due to differences in schema or schema ID). By forcing repartitions, the schema generated by ksqlDB is used for data on both sides of the join, which is consistent given logical schema, key format, key format properties, and key serde features, all of which are verified to be the same on both sides of the join, else the join is rejected.

In order to support stream-table and table-table joins, this PR adds support for repartitioning tables. This functionality is hidden from the user (users still see the same error message indicating that ksqlDB does not support repartitioning tables) and is purely used for this internal use case of enabling joins with SR-enabled key formats.

This approach of forcing repartitions on both sides of the join is not resource-efficient for two main reasons:

The repartition happens even if it's not strictly necessary (e.g., if there was already another repartition upstream of the join). We can enhance the join logic to avoid this in a subsequent PR if desired.
Table repartitions currently introduce an extra state store and changelog topic (there are two rather than one) because we need to pass in custom serdes in the toTable() call, which requires passing in a Materialized object, which results in an extra state store. I don't know of a way around this at this time.

Testing done

QTT + manual.

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

vcrfxia · 2020-11-17T20:25:15Z

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

@@ -92,6 +101,11 @@ public LogicalSchema getSchema() {

  @Override
  public void setKeyFormat(final FormatInfo format) {
+    // Force repartition in case of schema inference, to avoid misses due to key schema ID mismatch
+    if (FormatFactory.of(format).supportsFeature(SerdeFeature.SCHEMA_INFERENCE)) {


If we want to fix the first inefficiency mentioned in the PR description (that we currently force repartitions even if not strictly necessary), this is the bit of logic that would be updated.

might be good to link this to a GH issue and place the issue inline

vcrfxia · 2020-11-17T20:25:57Z

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

+      final String errorMsg = "Implicit repartitioning of windowed sources is not supported. "
+          + "See https://github.com/confluentinc/ksql/issues/4385.";
+      final String additionalMsg = forceRepartition
+          ? " As a result, ksqlDB does not support joins on Schema-Registry-enabled key formats "


This PR does not add support for joining on windowed sources with SR-enableld key formats. This will be a follow-up PR.

vcrfxia · 2020-11-17T20:26:40Z

ksqldb-engine/src/test/java/io/confluent/ksql/structured/SchemaKStreamTest.java


    // Then:
    assertThat(result, is(initialSchemaKStream));
  }

  @Test
-  public void shouldNotRepartitionIfRowkey() {


This test is identical to shouldNotRepartitionIfSameKeyField above, so the duplicate has been removed.

vcrfxia · 2020-11-17T20:27:44Z

ksqldb-functional-tests/src/test/resources/query-validation-tests/joins.json

      }
    },
    {
-      "name": "stream stream left join",


The test immediately above this one has the same name. I've renamed this one to indicate the difference and avoid confusion.

vcrfxia · 2020-11-17T20:28:33Z

ksqldb-functional-tests/src/test/resources/query-validation-tests/joins.json

+              "valueFormat" : {"format" : "JSON"}
+            },
+            {
+              "name" : "_confluent-ksql-some.ksql.service.idquery_CSAS_OUTPUT_0-KafkaTopic_Right-Reduce-changelog",


This is an example of the fact that we have two changelog topics here when we really only want to have one :(

vcrfxia · 2020-11-17T20:30:18Z

ksqldb-functional-tests/src/test/resources/query-validation-tests/joins.json

+        "topics" : {
+          "topics" : [
+            {
+              "name" : "_confluent-ksql-some.ksql.service.idquery_CSAS_OUTPUT_0-Join-repartition",


Ideally this topic would be named _confluent-ksql-some.ksql.service.idquery_CSAS_OUTPUT_0-Join-left-repartition (or similar) to indicate that it's the repartition topic on the left side of the join, but this would be a breaking change and I'm not sure it's worth the confusion it will save. (The reason it's unambiguous prior to this PR is because for stream-table joins, only the stream side (left side) would ever be repartitioned, but that's no longer true with this PR.)

vcrfxia · 2020-11-17T20:32:37Z

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

+      final MaterializedFactory materializedFactory,
+      final PartitionByParamsBuilder paramsBuilder
+  ) {
+    final LogicalSchema sourceSchema = table.getSchema();


I've tried to implement this new plan step in a generic way so that it can also support user table repartitions in the future, rather than baking in assumptions such as the fact that today the logical schema (and key format) never change when this step is called, as the step is only used to enable a specific join use case. However, there's no testing for the more general table rekey functionality, so maybe it's more misleading to have the step appear general, and we should be failing if any of these assumptions are violated instead?

I think it's fine to implement it generically, we should have tests that fail if a user tries to repartition a table so I think that covers that this isn't being used for evil

agavra

LGTM! Some comments inline about test coverage and a few clarifying points, otherwise I think the approach makes sense

agavra · 2020-11-17T20:38:24Z

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

@@ -92,6 +101,11 @@ public LogicalSchema getSchema() {

  @Override
  public void setKeyFormat(final FormatInfo format) {
+    // Force repartition in case of schema inference, to avoid misses due to key schema ID mismatch
+    if (FormatFactory.of(format).supportsFeature(SerdeFeature.SCHEMA_INFERENCE)) {


might be good to link this to a GH issue and place the issue inline

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

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

ksqldb-functional-tests/src/test/resources/query-validation-tests/joins.json

agavra · 2020-11-17T20:53:12Z

ksqldb-functional-tests/src/test/resources/query-validation-tests/joins.json

@@ -2298,6 +2342,159 @@
        "type": "io.confluent.ksql.util.KsqlStatementException",
        "message": "Incompatible key formats. `T1` has KAFKA while `T2` has DELIMITED.\nCorrect the key format by creating a copy of the table with the correct key format. For example:\n\tCREATE TABLE T_COPY\n\t WITH (KEY_FORMAT = <required format>, <other key format config>)\n\t AS SELECT * FROM T;"
      }
+    },
+    {
+      "name": "stream-steam key-to-key - SR-enabled key format",


we might want to introduce a pre-topics node here so that we have control over the actual schema in schema registry. I think we also have the ability to ensure the schema at the end. It would be good to have both of these in the test to make sure we don't accidentally create a backwards incompatible change if/when we optimize away unnecessary repartitions

I think it also makes sense to add tests for:

schemas are identical, should repartition anyway

schemas are logically identical but physically different

schemas are logically different, should fail (in table-table joins anyway)

Also we should make sure what happens when we have an SR-enabled value format as well (I don't imagine anything should go wrong, but it makes sure that when we piped through the value format in the table select key we don't mess things up)

🚂

Suggested change

"name": "stream-steam key-to-key - SR-enabled key format",

"name": "stream-stream key-to-key - SR-enabled key format",

we might want to introduce a pre-topics node here so that we have control over the actual schema in schema registry. I think we also have the ability to ensure the schema at the end.

Yes, I've been thinking this as well. Wanted to open the PR first in order to sanity-check the approach -- since we're on the same page I'll pursue beefing up the tests 👍

I think it also makes sense to add tests for:

schemas are identical, should repartition anyway

schemas are logically identical but physically different

schemas are logically different, should fail (in table-table joins anyway)

The new tests I added are already case (1). Tests for case (3) exist for non-SR-enabled formats, but I'll add one for an SR-enabled format. I'll add (2) and also tests with SR-enabled value formats as a sanity check.

🚂

Haha! I fell victim to copy-paste from existing tests. I made the update, which required renaming existing historic plans so this PR diff just became a lot larger in terms of number of lines.

Added the different tests discussed above. Good call on testing table-table join with nulls -- turns out we've got a bug. The newly added test won't pass until #6647 is merged.

agavra · 2020-11-17T21:10:15Z

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

@@ -54,6 +59,10 @@ public PreJoinRepartitionNode(
    this.joiningNode = source instanceof JoiningNode
        ? Optional.of((JoiningNode) source)
        : Optional.empty();
+    this.valueFormat = getLeftmostSourceNode()


we might want to move this into JoiningNode so that we can leverage it both here and in JoinNode#getValueFormatForSource (that way, we can ensure that they line up - if for some reason we decide to switch JoinNode to use the right most value format, we won't have to make that change here too)

Done -- good call!

agavra · 2020-11-17T21:52:51Z

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

+    if (FormatFactory.of(format).supportsFeature(SerdeFeature.SCHEMA_INFERENCE)) {
+      forceRepartition = true;
+    }
+
    if (requiresRepartition()) {


I'm not sure, but should we make this requiresRepartition() || forceRepartition? If not, we should add a comment about why forced repartitions don't move in here (instead of // Node is repartitioning already)

Addressed this as part of revamping the logic (see #6635 (comment)).

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

agavra · 2020-11-17T21:57:02Z

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

@@ -92,6 +101,11 @@ public LogicalSchema getSchema() {

  @Override
  public void setKeyFormat(final FormatInfo format) {
+    // Force repartition in case of schema inference, to avoid misses due to key schema ID mismatch


not your code: it bothers me a little that there's nothing "requiring" that this is called. can we have buildStream() throw an exception if setKeyFormat was never set? this will make sure that we don't refactor the code and lose this call somewhere in the refactor

That's because it wasn't required that this method be called previously :) I've revamped the logic here so it's now required, as that was the major point of confusion when reading this code as well. Also added the check you suggested.

Good call on challenging this, BTW. There was a bug where we weren't always forcing repartitions for upstream joins with SR-enableld key formats in multi-joins. I've fixed the bug and also added multi-join QTTs to cover this.

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

vcrfxia · 2020-11-20T16:58:31Z

ksqldb-functional-tests/src/test/resources/query-validation-tests/joins.json

+      }
+    },
+    {
+      "name": "table-table - SR-enabled key format - with nulls",


This test will not succeed until #6647 is merged.

agavra

💯 excellent test coverage. LGTM!

agavra · 2020-11-20T18:52:18Z

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

@@ -140,15 +144,18 @@ public SchemaKTable(
    );
  }

+  @SuppressFBWarnings("UC_USELESS_CONDITION")


😂 pah. what a useless condition.

agavra · 2020-11-20T18:53:35Z

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

  ) {
-    if (repartitionNotNeeded(ImmutableList.of(keyExpression))) {
-      return (SchemaKStream<Struct>) this;
+    if (repartitionNotNeeded(ImmutableList.of(keyExpression)) && !forceRepartition) {


Suggested change

if (repartitionNotNeeded(ImmutableList.of(keyExpression)) && !forceRepartition) {

if (!forceRepartition && repartitionNotNeeded(ImmutableList.of(keyExpression))) {

nit: pet peeve of mine 😈

vcrfxia added 3 commits November 17, 2020 12:11

chore: force repartition on joins with SR-enabled key formats

aed78b1

refactor: move common methods to new util

cb94de4

chore: historic plans

c0d8346

vcrfxia requested a review from a team as a code owner November 17, 2020 20:23

vcrfxia commented Nov 17, 2020

View reviewed changes

vcrfxia requested a review from agavra November 17, 2020 20:33

chore: findbugs

8638a61

vcrfxia mentioned this pull request Nov 17, 2020

[SR Support]: Figure out how to join on keys where the key contains the schema id #6332

Closed

chore: fix test

ad4ed3b

agavra approved these changes Nov 17, 2020

View reviewed changes

vcrfxia added 5 commits November 17, 2020 16:57

chore: fix test

94bef77

chore: tweak error message, rename qtts

755de3b

test: additional qtt

3a13511

test: add table-table join test with nulls (not passing)

12bf73f

Merge branch 'master' into join-force-repartition

062e9ac

agavra mentioned this pull request Nov 19, 2020

chore: make StreamSelectKey generic to handle window repartitions #6642

Merged

2 tasks

This was referenced Nov 19, 2020

fix: propagate null-valued records in repartition #6647

Merged

Avoid unnecessary repartitions on joins with SR-enabled key formats #6648

Open

vcrfxia added 2 commits November 19, 2020 19:06

chore: feedback

fa5050e

fix: require key format to always be set

e4ed326

vcrfxia commented Nov 20, 2020

View reviewed changes

vcrfxia requested a review from agavra November 20, 2020 16:58

This was referenced Nov 20, 2020

Repartitioning tables for joins should result in a single changelog, not two #6650

Closed

[Auto repartitioning support]: Auto repartition joins on key format mismatch #6229

Closed

chore: minor cleanup to internal repartition topic names

a7bcc09

agavra approved these changes Nov 20, 2020

View reviewed changes

vcrfxia added 2 commits November 20, 2020 12:12

chore: almog's pet peeve ;o

ffd1293

Merge branch 'master' into join-force-repartition

c9cb2d7

vcrfxia merged commit 67f13f3 into confluentinc:master Nov 20, 2020

vcrfxia deleted the join-force-repartition branch November 20, 2020 22:17

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

chore: force repartition on joins with SR-enabled key formats #6635

chore: force repartition on joins with SR-enabled key formats #6635

vcrfxia commented Nov 17, 2020

vcrfxia Nov 17, 2020

agavra Nov 17, 2020

vcrfxia Nov 17, 2020

vcrfxia Nov 17, 2020

vcrfxia Nov 17, 2020

vcrfxia Nov 17, 2020

vcrfxia Nov 17, 2020

vcrfxia Nov 17, 2020

agavra Nov 17, 2020

agavra left a comment

agavra Nov 17, 2020

agavra Nov 17, 2020

vcrfxia Nov 18, 2020

vcrfxia Nov 20, 2020

agavra Nov 17, 2020

vcrfxia Nov 20, 2020

agavra Nov 17, 2020

vcrfxia Nov 20, 2020

agavra Nov 17, 2020

vcrfxia Nov 20, 2020

vcrfxia Nov 20, 2020

vcrfxia Nov 20, 2020

agavra left a comment

agavra Nov 20, 2020

agavra Nov 20, 2020

	"name": "stream-steam key-to-key - SR-enabled key format",
	"name": "stream-stream key-to-key - SR-enabled key format",

	if (repartitionNotNeeded(ImmutableList.of(keyExpression)) && !forceRepartition) {
	if (!forceRepartition && repartitionNotNeeded(ImmutableList.of(keyExpression))) {

chore: force repartition on joins with SR-enabled key formats #6635

chore: force repartition on joins with SR-enabled key formats #6635

Conversation

vcrfxia commented Nov 17, 2020

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

Choose a reason for hiding this comment

agavra 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

agavra left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment