Skip to content

Commit

Permalink
Verify whether is_blind_append is calculated accurately
Browse files Browse the repository at this point in the history 
Without the fix in the previous commit, in the concurrency
tests for the non-blind INSERT operations it happened sometime that
the source table handles processed in `finishInsert` operation
were wrongly skipped because they were considered time travel
references of the target table which led to the situation in
which the INSERT operation was inaccurately considered blind.
  • Loading branch information
@findinpath @ebyhr
findinpath authored and ebyhr committed Apr 3, 2024
1 parent 11eedf4 commit fe82520
Show file tree
Hide file tree
Showing 2 changed files with 35 additions and 35 deletions.
16 changes: 8 additions & 8 deletions ...o-delta-lake/src/test/java/io/trino/plugin/deltalake/BaseDeltaLakeConnectorSmokeTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2406,11 +2406,11 @@ private void testConcurrentInsertsSelectingFromTheSameTable(boolean partitioned)
.map("(%d, 10)"::formatted)
.collect(joining(", ", ", ", "")));
assertQuery(
"SELECT version, operation, isolation_level, read_version FROM \"" + tableName + "$history\"",
"VALUES (0, 'CREATE TABLE AS SELECT', 'WriteSerializable', 0)" +
"SELECT version, operation, isolation_level, read_version, is_blind_append FROM \"" + tableName + "$history\"",
"VALUES (0, 'CREATE TABLE AS SELECT', 'WriteSerializable', 0, true)" +
LongStream.rangeClosed(1, successfulInsertsCount)
.boxed()
.map(version -> "(%s, 'WRITE', 'WriteSerializable', %s)".formatted(version, version - 1))
.map(version -> "(%s, 'WRITE', 'WriteSerializable', %s, false)".formatted(version, version - 1))
.collect(joining(", ", ", ", "")));
}
finally {
Expand Down Expand Up @@ -2458,13 +2458,13 @@ public void testConcurrentInsertsReconciliationForMixedInserts()
"SELECT * FROM " + tableName,
"VALUES (0, 10), (1, 10), (11, 20), (1, 20), (22, 30)");
assertQuery(
"SELECT version, operation, isolation_level, read_version FROM \"" + tableName + "$history\"",
"SELECT operation, isolation_level, is_blind_append FROM \"" + tableName + "$history\"",
"""
VALUES
(0, 'CREATE TABLE AS SELECT', 'WriteSerializable', 0),
(1, 'WRITE', 'WriteSerializable', 0),
(2, 'WRITE', 'WriteSerializable', 1),
(3, 'WRITE', 'WriteSerializable', 2)
('CREATE TABLE AS SELECT', 'WriteSerializable', true),
('WRITE', 'WriteSerializable', false),
('WRITE', 'WriteSerializable', false),
('WRITE', 'WriteSerializable', true)
""");
}
finally {
Expand Down
54 changes: 27 additions & 27 deletions ...-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeLocalConcurrentWritesTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -153,13 +153,13 @@ private void testConcurrentInsertsReconciliationForBlindInserts(boolean partitio
.forEach(MoreFutures::getDone);

assertThat(query("SELECT * FROM " + tableName)).matches("VALUES (1, 10), (11, 20), (21, 30)");
assertQuery("SELECT version, operation, isolation_level, read_version FROM \"" + tableName + "$history\"",
assertQuery("SELECT version, operation, isolation_level, read_version, is_blind_append FROM \"" + tableName + "$history\"",
"""
VALUES
(0, 'CREATE TABLE', 'WriteSerializable', 0),
(1, 'WRITE', 'WriteSerializable', 0),
(2, 'WRITE', 'WriteSerializable', 1),
(3, 'WRITE', 'WriteSerializable', 2)
(0, 'CREATE TABLE', 'WriteSerializable', 0, true),
(1, 'WRITE', 'WriteSerializable', 0, true),
(2, 'WRITE', 'WriteSerializable', 1, true),
(3, 'WRITE', 'WriteSerializable', 2, true)
""");
}
finally {
Expand Down Expand Up @@ -233,11 +233,11 @@ private void testConcurrentInsertsSelectingFromTheSameTable(boolean partitioned)
.map("(%d, 10)"::formatted)
.collect(joining(", ", ", ", "")));
assertQuery(
"SELECT version, operation, isolation_level, read_version FROM \"" + tableName + "$history\"",
"VALUES (0, 'CREATE TABLE AS SELECT', 'WriteSerializable', 0)" +
"SELECT version, operation, isolation_level, read_version, is_blind_append FROM \"" + tableName + "$history\"",
"VALUES (0, 'CREATE TABLE AS SELECT', 'WriteSerializable', 0, true)" +
LongStream.rangeClosed(1, successfulInsertsCount)
.boxed()
.map(version -> "(%s, 'WRITE', 'WriteSerializable', %s)".formatted(version, version - 1))
.map(version -> "(%s, 'WRITE', 'WriteSerializable', %s, false)".formatted(version, version - 1))
.collect(joining(", ", ", ", "")));
}
finally {
Expand Down Expand Up @@ -303,11 +303,11 @@ public void testConcurrentInsertsSelectingFromTheSamePartition()
.map("(%d, 10)"::formatted)
.collect(joining(", ", ", ", "")));
assertQuery(
"SELECT version, operation, isolation_level, read_version FROM \"" + tableName + "$history\"",
"VALUES (0, 'CREATE TABLE AS SELECT', 'WriteSerializable', 0)" +
"SELECT version, operation, isolation_level, read_version, is_blind_append FROM \"" + tableName + "$history\"",
"VALUES (0, 'CREATE TABLE AS SELECT', 'WriteSerializable', 0, true)" +
LongStream.rangeClosed(1, successfulInsertsCount)
.boxed()
.map(version -> "(%s, 'WRITE', 'WriteSerializable', %s)".formatted(version, version - 1))
.map(version -> "(%s, 'WRITE', 'WriteSerializable', %s, false)".formatted(version, version - 1))
.collect(joining(", ", ", ", "")));
}
finally {
Expand Down Expand Up @@ -356,13 +356,13 @@ public void testConcurrentInsertsReconciliationForMixedInserts()
"SELECT * FROM " + tableName,
"VALUES (0, 10), (1, 10), (11, 20), (1, 20), (22, 30)");
assertQuery(
"SELECT version, operation, isolation_level, read_version FROM \"" + tableName + "$history\"",
"SELECT operation, isolation_level, is_blind_append FROM \"" + tableName + "$history\"",
"""
VALUES
(0, 'CREATE TABLE AS SELECT', 'WriteSerializable', 0),
(1, 'WRITE', 'WriteSerializable', 0),
(2, 'WRITE', 'WriteSerializable', 1),
(3, 'WRITE', 'WriteSerializable', 2)
('CREATE TABLE AS SELECT', 'WriteSerializable', true),
('WRITE', 'WriteSerializable', false),
('WRITE', 'WriteSerializable', false),
('WRITE', 'WriteSerializable', true)
""");
}
finally {
Expand Down Expand Up @@ -409,13 +409,13 @@ public void testConcurrentInsertsSelectingFromDifferentPartitionsOfSameTable()
"SELECT * FROM " + tableName,
"VALUES (0, 10), (11, 20), (22, 30), (1, 40), (13, 40), (25, 40)");
assertQuery(
"SELECT version, operation, isolation_level, read_version FROM \"" + tableName + "$history\"",
"SELECT operation, isolation_level, is_blind_append FROM \"" + tableName + "$history\"",
"""
VALUES
(0, 'CREATE TABLE AS SELECT', 'WriteSerializable', 0),
(1, 'WRITE', 'WriteSerializable', 0),
(2, 'WRITE', 'WriteSerializable', 1),
(3, 'WRITE', 'WriteSerializable', 2)
('CREATE TABLE AS SELECT', 'WriteSerializable', true),
('WRITE', 'WriteSerializable', false),
('WRITE', 'WriteSerializable', false),
('WRITE', 'WriteSerializable', false)
""");
}
finally {
Expand Down Expand Up @@ -472,14 +472,14 @@ public void testConcurrentInsertsSelectingFromMultipleNonoverlappingPartitionsOf
(55, 60), (56,60), (57, 60), (58,60)
""");
assertQuery(
"SELECT version, operation, isolation_level, read_version FROM \"" + tableName + "$history\"",
"SELECT version, operation, isolation_level, read_version, is_blind_append FROM \"" + tableName + "$history\"",
"""
VALUES
(0, 'CREATE TABLE AS SELECT', 'WriteSerializable', 0),
(1, 'WRITE', 'WriteSerializable', 0),
(2, 'WRITE', 'WriteSerializable', 1),
(3, 'WRITE', 'WriteSerializable', 2),
(4, 'WRITE', 'WriteSerializable', 3)
(0, 'CREATE TABLE AS SELECT', 'WriteSerializable', 0, true),
(1, 'WRITE', 'WriteSerializable', 0, true),
(2, 'WRITE', 'WriteSerializable', 1, false),
(3, 'WRITE', 'WriteSerializable', 2, false),
(4, 'WRITE', 'WriteSerializable', 3, false)
""");
}
finally {
Expand Down

0 comments on commit fe82520

Please sign in to comment.