You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "nastra (via GitHub)" <gi...@apache.org> on 2023/11/13 08:18:40 UTC
[PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
nastra opened a new pull request, #9050:
URL: https://github.com/apache/iceberg/pull/9050
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1393529465
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
The validation on the Iceberg side works correctly. The open question is whether the optimizer rule is safe to apply under these conditions and whether our Spark planning rule behaves correctly. I'll need to check with fresh eyes tomorrow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1408616206
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
The proper way to ensure the MERGE statement above is serializable is to perform the following checks:
- Validate no new data matching `ref(name="id") == 1005` is added to the target table.
- Validate no new data matching `class = 1` is added to the source table.
We have the first part. I wonder if there is a workaround to validate the second part, assuming it is also an Iceberg table, without a proper transaction API in Spark. I'll need to think whether we can inject a custom rule to catch this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1394691519
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
In that sense, you are right that the transaction as a whole would not be serializable as it involves multiple tables. One may argue that if the source is idempotent, the operation will be considered serializable under those conflict detection conditions.
As a data source, can we actually achieve serializability for the transactions involving reads from multiple tables without proper transaction APIs in query engines? If we modify the example to use another Iceberg table or another data source, using `true` as the conflict detection filter will not help us catch concurrent changes in other tables.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1393318722
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
Hm, I am not sure how it helps. Isn't the filter already like not null here?
@nastra, could you check if the issue is related to runtime filtering by disabling it? The source relation must be idempotent (must return the same output), which is clearly not the case here. The source may be scanned multiple times.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1392294340
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
this is just a workaround for now so that we don't violate serializable isolation. I think that's actually a viable fallback to maintain serializable isolation. The test fails without this when executed with Spark 3.4
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1394691519
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
In that sense, you are right that the transaction as a whole would not be serializable as it involves multiple tables. One may argue that if the source is idempotent, the operation will be considered serializable under those conflict detection conditions in face of concurrent modifications to the target table.
As a data source, can we actually achieve serializability for transactions involving reads from multiple tables without proper transaction APIs in query engines? If we modify the example to use another Iceberg table or another data source, using `true` as the conflict detection filter will not help us catch concurrent changes in other tables.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1392534466
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMerge.java:
##########
@@ -2944,4 +2945,77 @@ private RowLevelOperationMode mode(Table table) {
String modeName = table.properties().getOrDefault(MERGE_MODE, MERGE_MODE_DEFAULT);
return RowLevelOperationMode.fromName(modeName);
}
+
+ @Test
+ public void multipleMergeIntoWithSerializableIsolation() throws InterruptedException {
+ // cannot run tests with concurrency for Hadoop tables without atomic renames
+ Assumptions.assumeThat(catalogName).isNotEqualToIgnoringCase("testhadoop");
+ // if caching is off, the table is eagerly refreshed during runtime filtering
+ // this can cause a validation exception as concurrent changes would be visible
+ Assumptions.assumeThat(cachingCatalogEnabled()).isTrue();
+
+ createAndInitTable("id INT, class INT, value INT");
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
+ tableName, MERGE_ISOLATION_LEVEL, "serializable");
+
+ sql("INSERT INTO TABLE %s VALUES (1001, 1, 10)", tableName);
+ sql("INSERT INTO TABLE %s VALUES (1002, 1, 20)", tableName);
+ sql("INSERT INTO TABLE %s VALUES (1003, 2, 100)", tableName);
+ sql("INSERT INTO TABLE %s VALUES (1004, 2, 200)", tableName);
+ createBranchIfNeeded();
+
+ Thread t1 =
+ new Thread(
+ () ->
+ sql(
+ "MERGE INTO %s t USING (\n"
+ + " SELECT 1005 as id, 2 as class, SUM(value) as value FROM %s WHERE class = 1\n"
+ + ") u \n"
+ + "ON t.id = u.id\n"
+ + "WHEN MATCHED THEN UPDATE SET t.value = u.value\n"
+ + "WHEN NOT MATCHED THEN INSERT (id, class, value) VALUES (u.id, u.class, u.value);",
+ commitTarget(), commitTarget()));
+
+ Thread t2 =
+ new Thread(
+ () ->
+ sql(
+ "MERGE INTO %s t USING (\n"
Review Comment:
MOR is tested by this test as well, since `TestMerge` is just an abstract base class that is extended for COW and MOR testing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1393448207
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
Okay, the reason why we infer that predicate in Spark 3.4 is `FoldablePropagation` rule. Now let me see if we generated a wrong plan or if that rule has issues.
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
The reason why we infer that predicate in Spark 3.4 is `FoldablePropagation` rule. Now let me see if we generated a wrong plan or if that rule has issues.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1393606348
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
Here is the primary question. Are these statements equivalent?
```
MERGE INTO t USING (SELECT 1005 as id, 2 as class, SUM(value) as value FROM t WHERE class = 1) AS u
ON t.id = u.id
WHEN MATCHED THEN UPDATE SET t.value = u.value
WHEN NOT MATCHED THEN INSERT (id, class, value) VALUES (u.id, u.class, u.value)
```
```
MERGE INTO t USING (2 as class, SUM(value) as value FROM t WHERE class = 1) AS u
ON t.id = 1005
WHEN MATCHED THEN UPDATE SET t.value = u.value
WHEN NOT MATCHED THEN INSERT (id, class, value) VALUES (1005, u.class, u.value)
```
I think they would produce the same output, which probably makes them logically equivalent. If they are equivalent, does it mean it is a valid optimizer transform?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1393606348
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
Here is the primary question. Are these statements equivalent?
```
MERGE INTO t USING (SELECT 1005 as id, 2 as class, SUM(value) as value FROM t WHERE class = 1) AS u
ON t.id = u.id
WHEN MATCHED THEN UPDATE SET t.value = u.value
WHEN NOT MATCHED THEN INSERT (id, class, value) VALUES (u.id, u.class, u.value)
```
```
MERGE INTO t USING (2 as class, SUM(value) as value FROM t WHERE class = 1) AS u
ON t.id = 1005
WHEN MATCHED THEN UPDATE SET t.value = u.value
WHEN NOT MATCHED THEN INSERT (id, class, value) VALUES (1005, u.class, u.value)
```
I think they would produce the same output.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1401275467
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
I spent a bit more time thinking about this and I am not fully convinced using `true` as the conflict detection filter is the right approach, while I admit it works in the given example. The whole idea of the conflict detection filter is to control what is considered a serializable action against the target table (not the transaction as a whole) if there are concurrent changes to that table. In case of MERGE, we are dealing with a transaction that reads from multiple tables, so that transaction can only be serializable if we have control over both sides and can validate concurrent changes. Unfortunately, we don't have that in Spark right now as there is no transaction API.
In other words, the MERGE breaks serializability not because the conflict detection condition is wrong but because we have no way to validate concurrent changes to the other side.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1401516134
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
I'm 100% with you on this one @aokolnychyi, so I'm considering `true` as the conflict detection filter only as an intermediate approach. Long term we would want to fix this differently
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1393448207
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
The same rule applies in Spark 3.5 but the planning works differently. In Spark 3.4, we push the join condition that was already replaced. Therefore, `ref(name="id") == 1005` and `ref(name="id") == 1006` become the conflict detection conditions .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "ajantha-bhat (via GitHub)" <gi...@apache.org>.
ajantha-bhat commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1392502584
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMerge.java:
##########
@@ -2944,4 +2945,77 @@ private RowLevelOperationMode mode(Table table) {
String modeName = table.properties().getOrDefault(MERGE_MODE, MERGE_MODE_DEFAULT);
return RowLevelOperationMode.fromName(modeName);
}
+
+ @Test
+ public void multipleMergeIntoWithSerializableIsolation() throws InterruptedException {
+ // cannot run tests with concurrency for Hadoop tables without atomic renames
+ Assumptions.assumeThat(catalogName).isNotEqualToIgnoringCase("testhadoop");
+ // if caching is off, the table is eagerly refreshed during runtime filtering
+ // this can cause a validation exception as concurrent changes would be visible
+ Assumptions.assumeThat(cachingCatalogEnabled()).isTrue();
+
+ createAndInitTable("id INT, class INT, value INT");
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
+ tableName, MERGE_ISOLATION_LEVEL, "serializable");
+
+ sql("INSERT INTO TABLE %s VALUES (1001, 1, 10)", tableName);
+ sql("INSERT INTO TABLE %s VALUES (1002, 1, 20)", tableName);
+ sql("INSERT INTO TABLE %s VALUES (1003, 2, 100)", tableName);
+ sql("INSERT INTO TABLE %s VALUES (1004, 2, 200)", tableName);
+ createBranchIfNeeded();
+
+ Thread t1 =
+ new Thread(
+ () ->
+ sql(
+ "MERGE INTO %s t USING (\n"
+ + " SELECT 1005 as id, 2 as class, SUM(value) as value FROM %s WHERE class = 1\n"
+ + ") u \n"
+ + "ON t.id = u.id\n"
+ + "WHEN MATCHED THEN UPDATE SET t.value = u.value\n"
+ + "WHEN NOT MATCHED THEN INSERT (id, class, value) VALUES (u.id, u.class, u.value);",
+ commitTarget(), commitTarget()));
+
+ Thread t2 =
+ new Thread(
+ () ->
+ sql(
+ "MERGE INTO %s t USING (\n"
Review Comment:
nit: we can also have a test for update and delete case of copy-on-write?
Also I didn't see the isolation level usage for `merge-on-read`. I need to dig deeper on that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#issuecomment-2401027327
This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #9050: Spark: Add serialzable isolation test for concurrent MERGE INTOs
URL: https://github.com/apache/iceberg/pull/9050
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1394691519
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
In that sense, you are right that the transaction as a whole would not be serializable as it involves multiple tables. One may argue that if the source is idempotent, the operation will be considered serializable under those conflict detection conditions in face of concurrent modifications to the target table.
As a data source, can we actually achieve serializability for the transactions involving reads from multiple tables without proper transaction APIs in query engines? If we modify the example to use another Iceberg table or another data source, using `true` as the conflict detection filter will not help us catch concurrent changes in other tables.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1394354418
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
> The same rule applies in Spark 3.5 but the planning works differently. In Spark 3.4, we push the join condition that was already replaced. Therefore, `ref(name="id") == 1005` and `ref(name="id") == 1006` become the conflict detection conditions. According to those conditions, the operations are serializable. Therefore, both are committed.
I actually think that having `ref(name="id") == 1005` as a conflict detection condition in the one statement and `ref(name="id") == 1006` in the other statement and the way we detect conflicts isn't enough, because it can't prevent **write skew**.
In order to avoid **write skew** (a transaction reads some data, examines the results, and takes some action based the results it saw - thus there's a causal dependency between the queries and the writes of a transaction), serializable isolation needs to detect such a case, where a TX may have acted on an outdated premise (aka the data it read may have been updated in the meantime by another TX).
What I saw in the case with Spark 3.5. is that the conflict detection filter that was pushed down to Iceberg was empty, thus we ended up using `Expressions.alwaysTrue()` and correctly detected that there's a conflict. Using `Expressions.alwaysTrue()` isn't ideal for conflict detection, but it's better than violating serializable isolation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1393529465
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
The validation on the Iceberg side works correctly. The open question is whether the optimizer rule is safe to apply under these conditions and whether our Spark planning rule behaves correctly. I'll need to check with fresh eyes tomorrow.
From what I see, the scope of this (potential) issue is very limited. We must trigger constant folding in the ON condition and push that predicate to the target table.
@nastra, do you think you will have some time to test whether other engines perform this kind of replacement in ON conditions? I tested Postgres and it did not do that. However, that also depends on how things are rewritten.
```
QUERY PLAN
--------------------------------------------------------------------------------------------
Merge on target_table t (cost=35.56..71.42 rows=0 width=0)
-> Hash Right Join (cost=35.56..71.42 rows=10 width=22)
Hash Cond: (t.id = u.id)
-> Seq Scan on target_table t (cost=0.00..30.40 rows=2040 width=10)
-> Hash (cost=35.54..35.54 rows=1 width=16)
-> Subquery Scan on u (cost=35.52..35.54 rows=1 width=16)
-> Aggregate (cost=35.52..35.53 rows=1 width=16)
-> Seq Scan on target_table (cost=0.00..35.50 rows=10 width=4)
Filter: (class = 1)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1393529465
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
The validation on the Iceberg side works correctly. The open question is whether the optimizer rule is safe to apply under these conditions and whether our Spark planning rule behaves correctly. I'll need to check with fresh eyes tomorrow.
From what I see, the scope of this (potential) issue is very limited. We must trigger constant folding in the ON condition and push that predicate to the target table.
@nastra, do you think you will have some time to test whether other engines perform this kind of replacement in ON conditions? I tested Postgres and it did not do that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1392294340
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
this is just a workaround for now so that we don't violate serializable isolation. The test fails without this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "ajantha-bhat (via GitHub)" <gi...@apache.org>.
ajantha-bhat commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1392489081
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -401,7 +401,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
If we are going with this approach, it is better to check the isolation level early before line 395 and return?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "ajantha-bhat (via GitHub)" <gi...@apache.org>.
ajantha-bhat commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1392528464
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMerge.java:
##########
@@ -2944,4 +2945,77 @@ private RowLevelOperationMode mode(Table table) {
String modeName = table.properties().getOrDefault(MERGE_MODE, MERGE_MODE_DEFAULT);
return RowLevelOperationMode.fromName(modeName);
}
+
+ @Test
+ public void multipleMergeIntoWithSerializableIsolation() throws InterruptedException {
+ // cannot run tests with concurrency for Hadoop tables without atomic renames
+ Assumptions.assumeThat(catalogName).isNotEqualToIgnoringCase("testhadoop");
+ // if caching is off, the table is eagerly refreshed during runtime filtering
+ // this can cause a validation exception as concurrent changes would be visible
+ Assumptions.assumeThat(cachingCatalogEnabled()).isTrue();
+
+ createAndInitTable("id INT, class INT, value INT");
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
+ tableName, MERGE_ISOLATION_LEVEL, "serializable");
+
+ sql("INSERT INTO TABLE %s VALUES (1001, 1, 10)", tableName);
+ sql("INSERT INTO TABLE %s VALUES (1002, 1, 20)", tableName);
+ sql("INSERT INTO TABLE %s VALUES (1003, 2, 100)", tableName);
+ sql("INSERT INTO TABLE %s VALUES (1004, 2, 200)", tableName);
+ createBranchIfNeeded();
+
+ Thread t1 =
+ new Thread(
+ () ->
+ sql(
+ "MERGE INTO %s t USING (\n"
+ + " SELECT 1005 as id, 2 as class, SUM(value) as value FROM %s WHERE class = 1\n"
+ + ") u \n"
+ + "ON t.id = u.id\n"
+ + "WHEN MATCHED THEN UPDATE SET t.value = u.value\n"
+ + "WHEN NOT MATCHED THEN INSERT (id, class, value) VALUES (u.id, u.class, u.value);",
+ commitTarget(), commitTarget()));
+
+ Thread t2 =
+ new Thread(
+ () ->
+ sql(
+ "MERGE INTO %s t USING (\n"
Review Comment:
> Also I didn't see the isolation level usage for `merge-on-read`. I need to dig deeper on that.
ok. I found the code for `merge-on-read` here (only position deletes since spark doesn't support equality delete writing)
https://github.com/apache/iceberg/blob/188847428309701c2ec0debf9e94da94b39058b6/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java#L223-L225
we can have the testcases for that too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1393448207
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
The same rule applies in Spark 3.5 but the planning works differently. In Spark 3.4, we push the join condition that was already replaced. Therefore, `ref(name="id") == 1005` and `ref(name="id") == 1006` become the conflict detection conditions. According to those conditions, the operations are serializable. Therefore, both are committed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1393448207
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
The same rule applies in Spark 3.5 but the planning works differently. In Spark 3.4, we push the join condition that was already replaced. Therefore, `ref(name="id") == 1005` and `ref(name="id") == 1006` become the conflict detection conditions. According to those conditions, the operations are serializable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1393318722
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
Hm, I am not sure how it helps. Isn't the filter already true here?
@nastra, could you check if the issue is related to runtime filtering by disabling it? The source relation must be idempotent (must return the same output), which is clearly not the case here. The source may be scanned multiple times.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1393418204
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
Okay, I see that Spark 3.4 infers `ref(name="id") == 1005` and `ref(name="id") == 1006`. Let me think.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1408616725
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
If we implement that, we can have truly serializable MERGE transactions if they touch Iceberg tables.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#issuecomment-1809442786
I will check tomorrow morning.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "ajantha-bhat (via GitHub)" <gi...@apache.org>.
ajantha-bhat commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1392489081
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -401,7 +401,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
If we are going with this approach, it is better to check the isolation level early before line 395 and return.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1393448207
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
The same rule applies in Spark 3.5 but the planning works differently. In Spark 3.4, we push the join condition that was already replaced. Therefore, `ref(name="id") == 1005` and `ref(name="id") == 1006` become the conflict detection conditions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1393529465
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
The validation on the Iceberg side works correctly. The open question is whether the optimizer rule is safe to apply under these conditions and whether our Spark planning rule behaves correctly. I'll need to check with fresh eyes tomorrow.
From what I see, the scope of this (potential) issue is very limited. We must trigger constant folding in the ON condition and push that predicate to the target table.
@nastra, do you think you will have some time to test whether other engines perform this kind of replacement in ON conditions? I tested Postgres and it did not do that. However, that also depends on how things are rewritten.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "ajantha-bhat (via GitHub)" <gi...@apache.org>.
ajantha-bhat commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1392502584
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMerge.java:
##########
@@ -2944,4 +2945,77 @@ private RowLevelOperationMode mode(Table table) {
String modeName = table.properties().getOrDefault(MERGE_MODE, MERGE_MODE_DEFAULT);
return RowLevelOperationMode.fromName(modeName);
}
+
+ @Test
+ public void multipleMergeIntoWithSerializableIsolation() throws InterruptedException {
+ // cannot run tests with concurrency for Hadoop tables without atomic renames
+ Assumptions.assumeThat(catalogName).isNotEqualToIgnoringCase("testhadoop");
+ // if caching is off, the table is eagerly refreshed during runtime filtering
+ // this can cause a validation exception as concurrent changes would be visible
+ Assumptions.assumeThat(cachingCatalogEnabled()).isTrue();
+
+ createAndInitTable("id INT, class INT, value INT");
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES('%s' '%s')",
+ tableName, MERGE_ISOLATION_LEVEL, "serializable");
+
+ sql("INSERT INTO TABLE %s VALUES (1001, 1, 10)", tableName);
+ sql("INSERT INTO TABLE %s VALUES (1002, 1, 20)", tableName);
+ sql("INSERT INTO TABLE %s VALUES (1003, 2, 100)", tableName);
+ sql("INSERT INTO TABLE %s VALUES (1004, 2, 200)", tableName);
+ createBranchIfNeeded();
+
+ Thread t1 =
+ new Thread(
+ () ->
+ sql(
+ "MERGE INTO %s t USING (\n"
+ + " SELECT 1005 as id, 2 as class, SUM(value) as value FROM %s WHERE class = 1\n"
+ + ") u \n"
+ + "ON t.id = u.id\n"
+ + "WHEN MATCHED THEN UPDATE SET t.value = u.value\n"
+ + "WHEN NOT MATCHED THEN INSERT (id, class, value) VALUES (u.id, u.class, u.value);",
+ commitTarget(), commitTarget()));
+
+ Thread t2 =
+ new Thread(
+ () ->
+ sql(
+ "MERGE INTO %s t USING (\n"
Review Comment:
nit: we can also have a test for `update` and `delete` case of `copy-on-write`?
Also I didn't see the isolation level usage for `merge-on-read`. I need to dig deeper on that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Add serialzable isolation test for concurrent MERGE INTOs [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #9050:
URL: https://github.com/apache/iceberg/pull/9050#discussion_r1392294340
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java:
##########
@@ -390,7 +390,7 @@ private Expression conflictDetectionFilter() {
filter = Expressions.and(filter, expr);
}
- return filter;
+ return SERIALIZABLE == isolationLevel ? Expressions.alwaysTrue() : filter;
Review Comment:
this is just a workaround for now so that we don't violate serializable isolation. The test fails without this when executed with Spark 3.4
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org