feat: avoid spurious tombstones in table output #6405

big-andy-coates · 2020-10-12T12:02:44Z

Description

AK commit apache/kafka#9156 enhances Kafka Streams so that filters on tables now avoid emitting spurious tombstones. ksqlDB now benefits from this. Tombstones are no longer emitted to the sink topic when a HAVING clause excludes a row from the result that has never been in the result table.

BREAKING CHANGE: This change fixes a bug where unnecessary tombstones where being emitted when a HAVING clause filtered out a row from the source that is not in the output table

For example, given:

-- source stream:
CREATE STREAM FOO (ID INT KEY, VAL INT) WITH (...);

-- aggregate into a table:
CREATE TABLE BAR AS
    SELECT ID, SUM(VAL) AS SUM
    FROM FOO
    GROUP BY ID
    HAVING SUM(VAL) > 0;


-- insert some values into the stream:
INSERT INTO FOO VALUES(1, -5); 
INSERT INTO FOO VALUES(1, 6); 
INSERT INTO FOO VALUES(1, -2); 
INSERT INTO FOO VALUES(1, -1);

Where previously the contents of the sink topic BAR would have contained records:

Key	Value	Notes
1.	null.	Spurious tombstone: the table does not contain a row with key `1`, so no tombstone is required.
1.	{sum=1}	Row added as HAVING criteria now met
1.	null.	Row deleted as HAVING criteria now not met
1.	null.	Spurious tombstone: the table does not contain a row with key `1`, so no tombstone is required.

Note: the first record in the tom

The topic will now contain:

Key	Value
1.	{sum=1}
1.	null.

Note: two historical tests are currently disabled. These need an upstream fix. See #6406 for details.

Testing done

usual

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

AK commit apache/kafka#9156 avoids filters emitting spurious tombstones. This means the sink topic now only receives the records for the two rows that pass the filter, not the other three rows. Hence the `waitForUniqueUserRows` call now only waits for the two records to be produced before running the test. Additionally, the name of the test was actually misleading as the logic in `KsqlMaterialization` to filter any records not passing the HAVING clause is actually installed as part of running the SQL in the test case, so those records are filtered from any pull request anyway.

vcrfxia

LGTM. Thanks for the fix!

big-andy-coates · 2020-10-12T13:05:37Z

More needed. Doing it now...

fixes: fixes: confluentinc#3558

big-andy-coates · 2020-10-12T13:41:28Z

...gine/src/test/java/io/confluent/ksql/materialization/ks/KsMaterializationFunctionalTest.java

+  public void shouldHandleHavingClause() {
+    // Note: HAVING clause are handled centrally by KsqlMaterialization. This logic will have been
+    // installed as part of building the below statement:


Note: test name and comments were misleading as the extra steps KsqlMaterialization adds to handle the HAVING clause are installed as part of this test.

big-andy-coates · 2020-10-12T13:42:01Z

...gine/src/test/java/io/confluent/ksql/materialization/ks/KsMaterializationFunctionalTest.java

+    final int matches = (int) USER_DATA_PROVIDER.data().values().stream()
+        .filter(row -> ((Long) row.get(0)) > 2)
+        .count();
+
+    final Map<String, GenericRow> rows = waitForUniqueUserRows(matches, STRING_DESERIALIZER, schema);


The number of expected rows is now reduced as we no longer produce spurious tombstones.

big-andy-coates · 2020-10-12T13:42:39Z

...gine/src/test/java/io/confluent/ksql/materialization/ks/KsMaterializationFunctionalTest.java

+    USER_DATA_PROVIDER.data().entries().stream()
+        .filter(e -> !rows.containsKey(e.getKey().getString("USERID")))
+        .forEach(e -> {
+          // Rows filtered by the HAVING clause:
+          final Optional<Row> row = withRetry(() -> table.get(e.getKey()));
+          assertThat(row, is(Optional.empty()));
+        });


Get's against the table for filtered out rows should return nothing.

big-andy-coates · 2020-10-12T13:44:56Z

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

@@ -52,9 +52,11 @@ public KudafAggregator(

  @Override
  public GenericRow apply(final K k, final GenericRow rowValue, final GenericRow aggRowValue) {
+    final GenericRow result = GenericRow.fromList(aggRowValue.values());


Kafka Streams does not expect the aggregator to mutate its parameters. The streams code is passing in the "old value", which ksqlDB was then mutating and returning as the "new value". This meant, when then function returned, the old and new values matched. This is obviously bad!

Code now takes a copy and mutates that. There is a perf hit, obviously, but it's unavoidable.

Not sure I understand -- why did the old code work, in that case? Or did something change on the Streams side recently?

The old code works because we were never enabling the sending of old values. We now do, to avoid the spurious tombstones.

Sorry, still not understanding. What was being sent before, if not the old values? Was this method even being called, previously?

The processing nodes in the streams topology can optionally include the old/previous value, as well as the new/current value, to child nodes. This is not on by default. An upstream change to how table filters is handled means this is now turned on.

The streams code for aggregation looks something like:

V process(K key, Change<V> change) { // Get old value from store: final V oldAgg = store.get(key); // Undo any previous value: final T intermediateAgg = value.oldValue != null && oldAgg != null ? remove.apply(key, value.oldValue, oldAgg) : oldAgg; // Then add the new value final T newAgg; if (value.newValue != null) { final T initializedAgg = intermediateAgg == null ? initializer.apply(); : intermediateAgg; newAgg = add.apply(key, value.newValue, initializedAgg); } else { newAgg = intermediateAgg; } // update the store with the new value. & forard store.put(key, newAgg); tupleForwarder.maybeForward(key, newAgg, sendOldValues ? oldAgg : null); }

The two calls: remove.apply(key, value.oldValue, oldAgg) and add.apply(key, value.newValue, initializedAgg) are calling out to ksqlDB code. If these calls directly mutate the oldAgg or initializedAgg parameters passed, rather than creating copies, then the old and new values forwarded to child nodes will match. i.e. in tupleForwarder.maybeForward(key, newAgg, sendOldValues ? oldAgg : null), the parameters newAgg and oldAgg will have the same updated value, rather than oldAgg holding the previous value. This breaks downstream processes, which expect the old and new value.

Previously the nodes weren't configured to send old values, so where just sending null for the old value and downstream could handle this correctly.

big-andy-coates · 2020-10-12T13:45:03Z

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

@@ -51,17 +51,19 @@ public KudafUndoAggregator(
  @SuppressWarnings({"unchecked", "rawtypes"})
  @Override
  public GenericRow apply(final Struct k, final GenericRow rowValue, final GenericRow aggRowValue) {
+    final GenericRow result = GenericRow.fromList(aggRowValue.values());


vcrfxia

LGTM besides the question inline, and the test failures in the build. Thanks!

vcrfxia · 2020-10-12T16:06:33Z

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

@@ -52,9 +52,11 @@ public KudafAggregator(

  @Override
  public GenericRow apply(final K k, final GenericRow rowValue, final GenericRow aggRowValue) {
+    final GenericRow result = GenericRow.fromList(aggRowValue.values());


Not sure I understand -- why did the old code work, in that case? Or did something change on the Streams side recently?

big-andy-coates requested a review from a team as a code owner October 12, 2020 12:02

big-andy-coates mentioned this pull request Oct 12, 2020

chore: fix build by pinning AK to 6.1.0-23-ccs #6396

Merged

2 tasks

vcrfxia approved these changes Oct 12, 2020

View reviewed changes

big-andy-coates changed the title ~~chore: fix master build~~ feat: avoid spurious tombstones in table output Oct 12, 2020

big-andy-coates added 2 commits October 12, 2020 14:31

feat: avoid supurious tombstones

cc7f1da

fixes: fixes: confluentinc#3558

test: updated test & historical plans

fcee2d4

big-andy-coates requested a review from vcrfxia October 12, 2020 13:40

big-andy-coates commented Oct 12, 2020

View reviewed changes

big-andy-coates requested a review from a team October 12, 2020 13:45

vcrfxia approved these changes Oct 12, 2020

View reviewed changes

big-andy-coates added 2 commits October 12, 2020 17:25

test: temp disable historical plans that fail

8bc96c1

test: disable _correct_ tests ;)

64480bf

big-andy-coates merged commit 4c7c9b5 into confluentinc:master Oct 12, 2020

big-andy-coates deleted the ifx_master branch October 12, 2020 19:54

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: avoid spurious tombstones in table output #6405

feat: avoid spurious tombstones in table output #6405

big-andy-coates commented Oct 12, 2020 •

edited

Loading

vcrfxia left a comment

big-andy-coates commented Oct 12, 2020

big-andy-coates Oct 12, 2020

big-andy-coates Oct 12, 2020

big-andy-coates Oct 12, 2020

big-andy-coates Oct 12, 2020

vcrfxia Oct 12, 2020

big-andy-coates Oct 12, 2020

vcrfxia Oct 12, 2020

big-andy-coates Oct 12, 2020

big-andy-coates Oct 12, 2020

vcrfxia left a comment

vcrfxia Oct 12, 2020

feat: avoid spurious tombstones in table output #6405

feat: avoid spurious tombstones in table output #6405

Conversation

big-andy-coates commented Oct 12, 2020 • edited Loading

Description

Testing done

Reviewer checklist

vcrfxia left a comment

Choose a reason for hiding this comment

big-andy-coates commented Oct 12, 2020

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

big-andy-coates commented Oct 12, 2020 •

edited

Loading