[flink] Delta Join additional IT tests and docs improvement #2268
Conversation
|
@xuyangzhong @wuchong |
fresh-borzoni
force-pushed
the
fluss-delta-joins-tests-docs
branch
from
ac23b6b to
6035422
Compare
|
Hi, @fresh-borzoni I'm a bit busy these days, I'll try my best to take a look after next Wednesday |
xuyangzhong
left a comment
xuyangzhong
left a comment
There was a problem hiding this comment.
Thanks for driving this! I have added some comments. Furthermore, some methods for table creation could be reused.
| import static org.assertj.core.api.Assertions.assertThatThrownBy; | ||
|
|
||
| /** IT case for {@link FlinkTableSource} in Flink 2.2. */ | ||
| public class Flink22TableSourceITCase extends FlinkTableSourceITCase { |
There was a problem hiding this comment.
I’m thinking it might be better to use a separate Flink22DeltaJoinITCase to hold these Delta Join–related tests, since there are now more delta join related tests, and there’s some common logic we could factor out, for example, setting the FORCE strategy, setting the default parallelism to 2, and creating the source and sink tables. WDYT?
| + " 'table.delete.behavior' = 'IGNORE' " | ||
| + ")", | ||
| leftTableName)); | ||
| List<InternalRow> rows1 = |
There was a problem hiding this comment.
nit: Could we prepare some data with the same primary key to test it?
| "INSERT INTO %s SELECT a1, c1, a2 FROM %s LEFT JOIN %s ON c1 = c2 AND d1 = d2", | ||
| sinkTableName, leftTableName, rightTableName); | ||
|
|
||
| assertThatThrownBy(() -> tEnv.explainSql(sql)) |
There was a problem hiding this comment.
nit Merge testDeltaJoinFailsWithLeftJoin, testDeltaJoinFailsWithRightJoin and testDeltaJoinFailsWithFullOuterJoin to testDeltaJoinFailsWithOuterJoin and assert the thrown exception here three times.
|
|
||
| assertThatThrownBy(() -> tEnv.explainSql(sql)) | ||
| .isInstanceOf(ValidationException.class) | ||
| .hasMessageContaining("doesn't support to do delta join optimization"); |
There was a problem hiding this comment.
add a more test here.
// Non-equiv-cond on e1 > e2, where e1 and e2 are NOT part of the upsert key
String sql2 =
String.format(
"INSERT INTO %s SELECT * FROM %s INNER JOIN %s ON c1 = c2 AND d1 = d2 AND e1 > e2",
sinkTableName, leftTableName, rightTableName);
assertThatThrownBy(() -> tEnv.explainSql(sql2))
.isInstanceOf(ValidationException.class)
.hasMessageContaining("doesn't support to do delta join optimization");
| } | ||
|
|
||
| @Test | ||
| void testDeltaJoinFailsWithCascadeJoin() throws Exception { |
There was a problem hiding this comment.
What about adding more tests from the doc like:
- the join key includes more fields than the primary key.
- sink materializer(the sink's primary key is different from the primary keys of the two upstream sources.)
- filters or projections contain non-deterministic functions like
rand.
2 participants
Purpose
Linked issue: close #2231
Add comprehensive test coverage for Delta Join feature in Flink 2.2 and improve documentation.
Brief change log
table.delete.behavior=IGNORE(not justfirst_rowmerge engine)Tests
testDeltaJoinWithPrimaryKeyTableNoDeletes- normal PK table withdelete.behavior=IGNOREtestDeltaJoinOnBucketKey- join on bucket key onlytestDeltaJoinFailsWhenFilterOnNonUpsertKeys- filter on non-upsert-key columns failstestDeltaJoinOnBucketKey- join on bucket key only (not full PK)testDeltaJoinFailsWhenSourceHasDelete- source with DELETE records failstestDeltaJoinFailsWhenJoinKeyNotContainIndex- join key not containing index failtestDeltaJoinFailsWithLeftJoin- LEFT JOIN wouldn't be converted to DeltaJointestDeltaJoinFailsWithRightJoin- RIGHT JOIN wouldn't be converted to DeltaJointestDeltaJoinFailsWithFullOuterJoin- FULL OUTER JOIN wouldn't be converted to DeltaJointestDeltaJoinFailsWithCascadeJoin- cascade join wouldn't be converted to DeltaJoinAPI and Format
No
Documentation
Yes - updated
docs/engine-flink/delta-joins.mdin Flink 2.2 part.