Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: avoid spurious tombstones in table output #6405

Merged
merged 5 commits into from
Oct 12, 2020
Merged

feat: avoid spurious tombstones in table output #6405

Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a10e5eb
chore: fix master build
big-andy-coates Oct 12, 2020
cc7f1da
feat: avoid supurious tombstones
big-andy-coates Oct 12, 2020
fcee2d4
test: updated test & historical plans
big-andy-coates Oct 12, 2020
8bc96c1
test: temp disable historical plans that fail
big-andy-coates Oct 12, 2020
64480bf
test: disable _correct_ tests ;)
big-andy-coates Oct 12, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 33 additions & 10 deletions ...e/src/test/java/io/confluent/ksql/materialization/ks/KsMaterializationFunctionalTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -619,8 +619,9 @@ public void shouldQueryMaterializedTableWithMultipleAggregationColumns() {
}

@Test
public void shouldIgnoreHavingClause() {
// Note: HAVING clause are handled centrally by KsqlMaterialization
public void shouldHandleHavingClause() {
// Note: HAVING clause are handled centrally by KsqlMaterialization. This logic will have been
// installed as part of building the below statement:
Comment on lines +622 to +624
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.


// Given:
final PersistentQueryMetadata query = executeQuery(
Expand All @@ -632,7 +633,11 @@ public void shouldIgnoreHavingClause() {

final LogicalSchema schema = schema("COUNT", SqlTypes.BIGINT);

final Map<String, GenericRow> rows = waitForUniqueUserRows(STRING_DESERIALIZER, schema);
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);
Comment on lines +636 to +640
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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


// When:
final Materialization materialization = query.getMaterialization(queryId, contextStacker).get();
Expand All @@ -641,16 +646,22 @@ public void shouldIgnoreHavingClause() {
final MaterializedTable table = materialization.nonWindowed();

rows.forEach((rowKey, value) -> {
// Rows passing the HAVING clause:
final Struct key = asKeyStruct(rowKey, query.getPhysicalSchema());

final Optional<Row> expected = Optional.ofNullable(value)
.map(v -> Row.of(schema, key, v, -1L));

final Optional<Row> row = withRetry(() -> table.get(key));
assertThat(row.map(Row::schema), is(expected.map(Row::schema)));
assertThat(row.map(Row::key), is(expected.map(Row::key)));
assertThat(row.map(Row::value), is(expected.map(Row::value)));
assertThat(row.map(Row::schema), is(Optional.of(schema)));
assertThat(row.map(Row::key), is(Optional.of(key)));
assertThat(row.map(Row::value), is(Optional.of(value)));
});

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()));
});
Comment on lines +658 to +664
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

}

private static void verifyRetainedWindows(
Expand All @@ -677,10 +688,22 @@ private static void verifyRetainedWindows(
private <T> Map<T, GenericRow> waitForUniqueUserRows(
final Deserializer<T> keyDeserializer,
final LogicalSchema aggregateSchema
) {
return waitForUniqueUserRows(
USER_DATA_PROVIDER.data().size(),
keyDeserializer,
aggregateSchema
);
}

private <T> Map<T, GenericRow> waitForUniqueUserRows(
final int count,
final Deserializer<T> keyDeserializer,
final LogicalSchema aggregateSchema
) {
return TEST_HARNESS.verifyAvailableUniqueRows(
output.toUpperCase(),
USER_DATA_PROVIDER.data().size(),
count,
VALUE_FORMAT,
PhysicalSchema.from(aggregateSchema, SerdeFeatures.of(), SerdeFeatures.of()),
keyDeserializer
Expand Down