You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2022/02/05 01:09:54 UTC
[GitHub] [iceberg] aokolnychyi opened a new pull request #4047: Spark 3.2: Implement merge-on-read MERGE
aokolnychyi opened a new pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047
This PR implements merge-on-read MERGE in Spark 3.2.
--
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
[GitHub] [iceberg] aokolnychyi commented on pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#issuecomment-1040860974
Thanks for reviewing, @rdblue @szehon-ho!
--
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
[GitHub] [iceberg] rdblue commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r805429098
##########
File path: spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMerge.java
##########
@@ -84,17 +86,22 @@ public void testMergeWithStaticPredicatePushDown() {
sql("ALTER TABLE %s ADD PARTITION FIELD dep", tableName);
- append(tableName,
- "{ \"id\": 1, \"dep\": \"software\" }\n" +
- "{ \"id\": 11, \"dep\": \"software\" }\n" +
- "{ \"id\": 1, \"dep\": \"hr\" }");
+ // add a data file to the 'software' partition
+ append(tableName, "{ \"id\": 1, \"dep\": \"software\" }");
+
+ // add a data file to the 'hr' partition
+ append(tableName, "{ \"id\": 1, \"dep\": \"hr\" }");
Table table = validationCatalog.loadTable(tableIdent);
Snapshot snapshot = table.currentSnapshot();
String dataFilesCount = snapshot.summary().get(SnapshotSummary.TOTAL_DATA_FILES_PROP);
Assert.assertEquals("Must have 2 files before MERGE", "2", dataFilesCount);
+ // remove the data file from the 'hr' partition to ensure it is not scanned
+ DataFile dataFile = Iterables.getOnlyElement(snapshot.addedFiles());
+ table.io().deleteFile(dataFile.path().toString());
Review comment:
Could you move it and move it back so that you don't remove the validation at the end of the test? Something like...
```java
withFilesUnavailable(Iterables.getOnlyElement(snapshot.addedFiles()).path().toString()) {
createOrReplaceView(...);
withSQLConf(...) { sql("MERGE INTO ...") };
}
```
You could move the files to a temp location, then copy them back at the end.
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r801000535
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is HASH -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is RANGE -> RANGE DISTRIBUTE BY _spec_id, _partition, _file, id, data +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ //
+ // PARTITIONED BY date, days(ts) UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDERED BY _spec_id, _partition, _file, _pos, date, days(ts)
+ // merge mode is HASH -> CLUSTER BY _spec_id, _partition, date, days(ts) +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, date, days(ts)
+ // merge mode is RANGE -> RANGE DISTRIBUTE BY _spec_id, _partition, _file, date, days(ts) +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, date, days(ts)
+ //
+ // PARTITIONED BY date ORDERED BY id
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDERED BY _spec_id, _partition, _file, _pos, date, id
+ // merge mode is HASH -> CLUSTER BY _spec_id, _partition, date +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, date, id
+ // merge mode is RANGE -> RANGE DISTRIBUTE BY _spec_id, _partition, _file, date, id
Review comment:
Note that we don't include `_pos` in the distribution. Just in the local ordering.
--
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
[GitHub] [iceberg] rdblue commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r805432750
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is HASH -> unspecified distribution +
Review comment:
Oh, I think I see. I was thinking about the `PARTITIONED BY, UNORDERED` case that is actually below. I concluded what you did for that case, so that's good validation!
Here, it still seems bad to me not to distribute. That's going to result in a lot of small delete files, which is really expensive and possibly worse than having a single writer for all the inserted data. It would be nice to be able to round-robin the new data... what about using something like `HASH DISTRIBUTE BY _spec, _partition, bucket(id, data, numShufflePartitions)`?
--
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
[GitHub] [iceberg] rdblue commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r805430382
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
Review comment:
I think by this you mean that we will check the write distribution mode and then use one of the following cases, right? So we can expect this to be `unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos` because all 3 modes (whether merge or write mode) have that behavior?
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r806263344
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is HASH -> unspecified distribution +
Review comment:
Well, I am not sure. I like that our merge and write logic are consistent right now. My hope was that AQE would coalesce tasks as needed to avoid a huge number of small writing tasks (and hence a huge number of delete files). I think AQE should behave better than a round-robin distribution. This case is about unpartitioned tables so we will most like produce at a single delete file per writing task (that shouldn't be that bad). As long as we don't have a huge number of writing tasks, we should be fine, I guess?
--
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
[GitHub] [iceberg] szehon-ho commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r806356103
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
Review comment:
Was also trying to compare the two code-paths. Noticed this has a 'later' in the comment and the other does not, the 'later' seems to be more clear. Actually wondering, did you consider putting some of these repeated codes/comments into methods
```
val sourceTableProj = sourceTableProj(source)
val joinPlan = joinPlan(sourceTableProj, targetTableProj, cond, notMatchedActions)
```
--
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
[GitHub] [iceberg] szehon-ho commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r806412971
##########
File path: spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkDistributionAndOrderingUtil.java
##########
@@ -171,12 +194,16 @@ private static Distribution buildPositionDeleteUpdateDistribution(DistributionMo
}
}
- public static SortOrder[] buildPositionDeltaOrdering(Table table, Command command, Distribution distribution) {
+ public static SortOrder[] buildPositionDeltaOrdering(Table table, Command command) {
// the spec requires position delete files to be sorted by file and pos
+ SortOrder[] deleteOrdering = {SPEC_ID_ORDER, PARTITION_ORDER, FILE_PATH_ORDER, ROW_POSITION_ORDER};
+
if (command == DELETE || command == UPDATE) {
- return new SortOrder[]{SPEC_ID_ORDER, PARTITION_ORDER, FILE_PATH_ORDER, ROW_POSITION_ORDER};
+ return deleteOrdering;
} else {
- throw new IllegalArgumentException("Only position deletes and updates are currently supported");
+ // all metadata columns like _spec_id, _file, _pos will be null for new data records
Review comment:
Optional: just noticed we throw IllegalArgumentException in other places where we have an unknown command, so was just wondering would it be clearer/more consistent to put this in a MERGE case block and throw exception in unknown command case? Up to you
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r806266795
##########
File path: spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMerge.java
##########
@@ -84,17 +86,22 @@ public void testMergeWithStaticPredicatePushDown() {
sql("ALTER TABLE %s ADD PARTITION FIELD dep", tableName);
- append(tableName,
- "{ \"id\": 1, \"dep\": \"software\" }\n" +
- "{ \"id\": 11, \"dep\": \"software\" }\n" +
- "{ \"id\": 1, \"dep\": \"hr\" }");
+ // add a data file to the 'software' partition
+ append(tableName, "{ \"id\": 1, \"dep\": \"software\" }");
+
+ // add a data file to the 'hr' partition
+ append(tableName, "{ \"id\": 1, \"dep\": \"hr\" }");
Table table = validationCatalog.loadTable(tableIdent);
Snapshot snapshot = table.currentSnapshot();
String dataFilesCount = snapshot.summary().get(SnapshotSummary.TOTAL_DATA_FILES_PROP);
Assert.assertEquals("Must have 2 files before MERGE", "2", dataFilesCount);
+ // remove the data file from the 'hr' partition to ensure it is not scanned
+ DataFile dataFile = Iterables.getOnlyElement(snapshot.addedFiles());
+ table.io().deleteFile(dataFile.path().toString());
Review comment:
Yeah, this will be helpful for future use cases. Let me spend some time on this one now.
--
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
[GitHub] [iceberg] aokolnychyi commented on pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#issuecomment-1039602241
@rdblue, you are right. There will be no `RowLevelOperationTable` when a MERGE gets converted into an INSERT. We will use the regular write distribution and ordering in that case.
--
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
[GitHub] [iceberg] rdblue commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r805431315
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is HASH -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is RANGE -> RANGE DISTRIBUTE BY _spec_id, _partition, _file, id, data +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ //
+ // PARTITIONED BY date, days(ts) UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDERED BY _spec_id, _partition, _file, _pos, date, days(ts)
+ // merge mode is HASH -> CLUSTER BY _spec_id, _partition, date, days(ts) +
Review comment:
I just thought about this for the case above. I think it's okay.
--
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
[GitHub] [iceberg] rdblue commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r805426755
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
+ val joinType = if (notMatchedActions.isEmpty) Inner else RightOuter
+ val joinHint = JoinHint(leftHint = Some(HintInfo(Some(NO_BROADCAST_HASH))), rightHint = None)
+ val joinPlan = Join(NoStatsUnaryNode(targetTableProj), sourceTableProj, joinType, Some(cond), joinHint)
+
+ val deleteRowValues = buildDeltaDeleteRowValues(rowAttrs, rowIdAttrs)
+ val metadataReadAttrs = readAttrs.filterNot(relation.outputSet.contains)
+
+ val matchedConditions = matchedActions.map(actionCondition)
+ val matchedOutputs = matchedActions.map { action =>
Review comment:
Nit: in the replace code, this is a one-liner. You could do the same thing here:
```scala
val matchedOutputs = matchedActions.map(deltaActionOutput(_, deleteRowValues, metadataReadAttrs))
```
Not a big deal, I'm just going through both and looking for changes so this looked a little odd. Same thing with `notMatchedOutputs`.
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
+ val joinType = if (notMatchedActions.isEmpty) Inner else RightOuter
+ val joinHint = JoinHint(leftHint = Some(HintInfo(Some(NO_BROADCAST_HASH))), rightHint = None)
+ val joinPlan = Join(NoStatsUnaryNode(targetTableProj), sourceTableProj, joinType, Some(cond), joinHint)
+
+ val deleteRowValues = buildDeltaDeleteRowValues(rowAttrs, rowIdAttrs)
+ val metadataReadAttrs = readAttrs.filterNot(relation.outputSet.contains)
+
+ val matchedConditions = matchedActions.map(actionCondition)
+ val matchedOutputs = matchedActions.map { action =>
+ deltaActionOutput(action, deleteRowValues, metadataReadAttrs)
+ }
+
+ val notMatchedConditions = notMatchedActions.map(actionCondition)
+ val notMatchedOutputs = notMatchedActions.map { action =>
+ deltaActionOutput(action, deleteRowValues, metadataReadAttrs)
+ }
+
+ val operationTypeAttr = AttributeReference(OPERATION_COLUMN, IntegerType, nullable = false)()
+ val rowFromSourceAttr = resolveAttrRef(ROW_FROM_SOURCE_REF, joinPlan)
+ val rowFromTargetAttr = resolveAttrRef(ROW_FROM_TARGET_REF, joinPlan)
+
+ // merged rows must contain values for the operation type and all read attrs
+ val mergeRowsOutput = buildMergeRowsOutput(
+ matchedOutputs,
+ notMatchedOutputs,
+ attrs = operationTypeAttr +: readAttrs)
Review comment:
This also fits inline if you want to match the replace code. (minor)
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
+ val joinType = if (notMatchedActions.isEmpty) Inner else RightOuter
+ val joinHint = JoinHint(leftHint = Some(HintInfo(Some(NO_BROADCAST_HASH))), rightHint = None)
+ val joinPlan = Join(NoStatsUnaryNode(targetTableProj), sourceTableProj, joinType, Some(cond), joinHint)
+
+ val deleteRowValues = buildDeltaDeleteRowValues(rowAttrs, rowIdAttrs)
+ val metadataReadAttrs = readAttrs.filterNot(relation.outputSet.contains)
+
+ val matchedConditions = matchedActions.map(actionCondition)
+ val matchedOutputs = matchedActions.map { action =>
+ deltaActionOutput(action, deleteRowValues, metadataReadAttrs)
+ }
+
+ val notMatchedConditions = notMatchedActions.map(actionCondition)
+ val notMatchedOutputs = notMatchedActions.map { action =>
+ deltaActionOutput(action, deleteRowValues, metadataReadAttrs)
+ }
+
+ val operationTypeAttr = AttributeReference(OPERATION_COLUMN, IntegerType, nullable = false)()
+ val rowFromSourceAttr = resolveAttrRef(ROW_FROM_SOURCE_REF, joinPlan)
+ val rowFromTargetAttr = resolveAttrRef(ROW_FROM_TARGET_REF, joinPlan)
+
+ // merged rows must contain values for the operation type and all read attrs
+ val mergeRowsOutput = buildMergeRowsOutput(
+ matchedOutputs,
+ notMatchedOutputs,
+ attrs = operationTypeAttr +: readAttrs)
+
+ val mergeRows = MergeRows(
+ isSourceRowPresent = IsNotNull(rowFromSourceAttr),
+ isTargetRowPresent = if (notMatchedActions.isEmpty) TrueLiteral else IsNotNull(rowFromTargetAttr),
+ matchedConditions = matchedConditions,
+ matchedOutputs = matchedOutputs,
+ notMatchedConditions = notMatchedConditions,
+ notMatchedOutputs = notMatchedOutputs,
+ targetOutput = Nil,
Review comment:
Would be nice to have a comment to explain this: "only needed if emitting unmatched target rows"
##########
File path: spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMerge.java
##########
@@ -84,17 +86,22 @@ public void testMergeWithStaticPredicatePushDown() {
sql("ALTER TABLE %s ADD PARTITION FIELD dep", tableName);
- append(tableName,
- "{ \"id\": 1, \"dep\": \"software\" }\n" +
- "{ \"id\": 11, \"dep\": \"software\" }\n" +
- "{ \"id\": 1, \"dep\": \"hr\" }");
+ // add a data file to the 'software' partition
+ append(tableName, "{ \"id\": 1, \"dep\": \"software\" }");
+
+ // add a data file to the 'hr' partition
+ append(tableName, "{ \"id\": 1, \"dep\": \"hr\" }");
Table table = validationCatalog.loadTable(tableIdent);
Snapshot snapshot = table.currentSnapshot();
String dataFilesCount = snapshot.summary().get(SnapshotSummary.TOTAL_DATA_FILES_PROP);
Assert.assertEquals("Must have 2 files before MERGE", "2", dataFilesCount);
+ // remove the data file from the 'hr' partition to ensure it is not scanned
+ DataFile dataFile = Iterables.getOnlyElement(snapshot.addedFiles());
+ table.io().deleteFile(dataFile.path().toString());
Review comment:
Could you move it and move it back so that you don't remove the validation at the end of the test? Something like...
```java
withFilesUnavailable(Iterables.getOnlyElement(snapshot.addedFiles()).path().toString()) {
createOrReplaceView(...);
withSQLConf(...) { sql("MERGE INTO ...") };
}
```
You could move the files to a temp location, then copy them back at the end.
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
Review comment:
I think by this you mean that we will check the write distribution mode and then use one of the following cases, right? So we can expect this to be `unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos` because all 3 modes (whether merge or write mode) have that behavior?
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
Review comment:
I think it may be worth an additional comment at the top of this section that says: `IMPORTANT: metadata columns like _spec_id and _partition are null for new rows`.
I think what this means is:
* Deletes and updates will be clustered by original partition and _file, and additionally sorted by _pos
* Inserts will effectively be distributed by id, data and sorted by id, data
This holds roughly true for all of the cases.
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
Review comment:
I think it may be worth an additional comment at the top of this section that says: `IMPORTANT: metadata columns like _spec_id and _partition are null for new rows`.
I think what this means is:
* Deletes and updates will be clustered by original partition and _file, and additionally sorted by _pos
* Inserts will effectively be distributed by id, data and sorted by id, data
This holds roughly true for all of the partitioning and sort order cases.
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is HASH -> unspecified distribution +
Review comment:
Why is the distribution unspecified when mode is hash here? Shouldn't this hash distribute by original data file (or maybe original partition), then by the new partition? Or is the fear that this will create too many small tasks by dividing original data files by the new partitioning?
If that's the case, then it seems like this is optimizing for the case where you're running a MERGE with data from an old partition spec. I'd rather optimize for the case where the partition spec matches.
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is HASH -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is RANGE -> RANGE DISTRIBUTE BY _spec_id, _partition, _file, id, data +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ //
+ // PARTITIONED BY date, days(ts) UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDERED BY _spec_id, _partition, _file, _pos, date, days(ts)
+ // merge mode is HASH -> CLUSTER BY _spec_id, _partition, date, days(ts) +
Review comment:
I just thought about this for the case above. I think it's okay.
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is HASH -> unspecified distribution +
Review comment:
Oh, I think I see. I was thinking about the `PARTITIONED BY, UNORDERED` case that is actually below. I concluded what you did for that case, so that's good validation!
Here, it still seems bad to me not to distribute. That's going to result in a lot of small delete files, which is really expensive and possibly worse than having a single writer for all the inserted data. It would be nice to be able to round-robin the new data... what about using something like `HASH DISTRIBUTE BY _spec, _partition, bucket(id, data, numShufflePartitions)`?
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is HASH -> unspecified distribution +
Review comment:
~~Why is the distribution unspecified when mode is hash here? Shouldn't this hash distribute by original data file (or maybe original partition), then by the new partition? Or is the fear that this will create too many small tasks by dividing original data files by the new partitioning?~~
~~If that's the case, then it seems like this is optimizing for the case where you're running a MERGE with data from an old partition spec. I'd rather optimize for the case where the partition spec matches.~~
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r807305902
##########
File path: spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkDistributionAndOrderingUtil.java
##########
@@ -171,12 +194,16 @@ private static Distribution buildPositionDeleteUpdateDistribution(DistributionMo
}
}
- public static SortOrder[] buildPositionDeltaOrdering(Table table, Command command, Distribution distribution) {
+ public static SortOrder[] buildPositionDeltaOrdering(Table table, Command command) {
// the spec requires position delete files to be sorted by file and pos
+ SortOrder[] deleteOrdering = {SPEC_ID_ORDER, PARTITION_ORDER, FILE_PATH_ORDER, ROW_POSITION_ORDER};
+
if (command == DELETE || command == UPDATE) {
- return new SortOrder[]{SPEC_ID_ORDER, PARTITION_ORDER, FILE_PATH_ORDER, ROW_POSITION_ORDER};
+ return deleteOrdering;
} else {
- throw new IllegalArgumentException("Only position deletes and updates are currently supported");
+ // all metadata columns like _spec_id, _file, _pos will be null for new data records
Review comment:
It looks like we usually do that in switch blocks as it's required by the syntax. I think it should be safe enough to assume we know a finite set of commands.
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r806228530
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
+ val joinType = if (notMatchedActions.isEmpty) Inner else RightOuter
+ val joinHint = JoinHint(leftHint = Some(HintInfo(Some(NO_BROADCAST_HASH))), rightHint = None)
+ val joinPlan = Join(NoStatsUnaryNode(targetTableProj), sourceTableProj, joinType, Some(cond), joinHint)
+
+ val deleteRowValues = buildDeltaDeleteRowValues(rowAttrs, rowIdAttrs)
+ val metadataReadAttrs = readAttrs.filterNot(relation.outputSet.contains)
+
+ val matchedConditions = matchedActions.map(actionCondition)
+ val matchedOutputs = matchedActions.map { action =>
Review comment:
I like it on a single line too. It is just the line limit in Spark is 100 (instead of 120 in Iceberg) and I copied the code from there. I'll make it consistent in Iceberg, at least.
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r806236672
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
Review comment:
Good idea, will add a 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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r807303645
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is HASH -> unspecified distribution +
Review comment:
Modified the behavior for HASH and RANGE modes.
##########
File path: spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMerge.java
##########
@@ -84,17 +86,22 @@ public void testMergeWithStaticPredicatePushDown() {
sql("ALTER TABLE %s ADD PARTITION FIELD dep", tableName);
- append(tableName,
- "{ \"id\": 1, \"dep\": \"software\" }\n" +
- "{ \"id\": 11, \"dep\": \"software\" }\n" +
- "{ \"id\": 1, \"dep\": \"hr\" }");
+ // add a data file to the 'software' partition
+ append(tableName, "{ \"id\": 1, \"dep\": \"software\" }");
+
+ // add a data file to the 'hr' partition
+ append(tableName, "{ \"id\": 1, \"dep\": \"hr\" }");
Table table = validationCatalog.loadTable(tableIdent);
Snapshot snapshot = table.currentSnapshot();
String dataFilesCount = snapshot.summary().get(SnapshotSummary.TOTAL_DATA_FILES_PROP);
Assert.assertEquals("Must have 2 files before MERGE", "2", dataFilesCount);
+ // remove the data file from the 'hr' partition to ensure it is not scanned
+ DataFile dataFile = Iterables.getOnlyElement(snapshot.addedFiles());
+ table.io().deleteFile(dataFile.path().toString());
Review comment:
Fixed.
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
+ val joinType = if (notMatchedActions.isEmpty) Inner else RightOuter
+ val joinHint = JoinHint(leftHint = Some(HintInfo(Some(NO_BROADCAST_HASH))), rightHint = None)
+ val joinPlan = Join(NoStatsUnaryNode(targetTableProj), sourceTableProj, joinType, Some(cond), joinHint)
+
+ val deleteRowValues = buildDeltaDeleteRowValues(rowAttrs, rowIdAttrs)
+ val metadataReadAttrs = readAttrs.filterNot(relation.outputSet.contains)
+
+ val matchedConditions = matchedActions.map(actionCondition)
+ val matchedOutputs = matchedActions.map { action =>
+ deltaActionOutput(action, deleteRowValues, metadataReadAttrs)
+ }
+
+ val notMatchedConditions = notMatchedActions.map(actionCondition)
+ val notMatchedOutputs = notMatchedActions.map { action =>
+ deltaActionOutput(action, deleteRowValues, metadataReadAttrs)
+ }
+
+ val operationTypeAttr = AttributeReference(OPERATION_COLUMN, IntegerType, nullable = false)()
+ val rowFromSourceAttr = resolveAttrRef(ROW_FROM_SOURCE_REF, joinPlan)
+ val rowFromTargetAttr = resolveAttrRef(ROW_FROM_TARGET_REF, joinPlan)
+
+ // merged rows must contain values for the operation type and all read attrs
+ val mergeRowsOutput = buildMergeRowsOutput(
+ matchedOutputs,
+ notMatchedOutputs,
+ attrs = operationTypeAttr +: readAttrs)
+
+ val mergeRows = MergeRows(
+ isSourceRowPresent = IsNotNull(rowFromSourceAttr),
+ isTargetRowPresent = if (notMatchedActions.isEmpty) TrueLiteral else IsNotNull(rowFromTargetAttr),
+ matchedConditions = matchedConditions,
+ matchedOutputs = matchedOutputs,
+ notMatchedConditions = notMatchedConditions,
+ notMatchedOutputs = notMatchedOutputs,
+ targetOutput = Nil,
Review comment:
Added.
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
+ val joinType = if (notMatchedActions.isEmpty) Inner else RightOuter
+ val joinHint = JoinHint(leftHint = Some(HintInfo(Some(NO_BROADCAST_HASH))), rightHint = None)
+ val joinPlan = Join(NoStatsUnaryNode(targetTableProj), sourceTableProj, joinType, Some(cond), joinHint)
+
+ val deleteRowValues = buildDeltaDeleteRowValues(rowAttrs, rowIdAttrs)
+ val metadataReadAttrs = readAttrs.filterNot(relation.outputSet.contains)
+
+ val matchedConditions = matchedActions.map(actionCondition)
+ val matchedOutputs = matchedActions.map { action =>
+ deltaActionOutput(action, deleteRowValues, metadataReadAttrs)
+ }
+
+ val notMatchedConditions = notMatchedActions.map(actionCondition)
+ val notMatchedOutputs = notMatchedActions.map { action =>
+ deltaActionOutput(action, deleteRowValues, metadataReadAttrs)
+ }
+
+ val operationTypeAttr = AttributeReference(OPERATION_COLUMN, IntegerType, nullable = false)()
+ val rowFromSourceAttr = resolveAttrRef(ROW_FROM_SOURCE_REF, joinPlan)
+ val rowFromTargetAttr = resolveAttrRef(ROW_FROM_TARGET_REF, joinPlan)
+
+ // merged rows must contain values for the operation type and all read attrs
+ val mergeRowsOutput = buildMergeRowsOutput(
+ matchedOutputs,
+ notMatchedOutputs,
+ attrs = operationTypeAttr +: readAttrs)
Review comment:
Fixed.
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
+ val joinType = if (notMatchedActions.isEmpty) Inner else RightOuter
+ val joinHint = JoinHint(leftHint = Some(HintInfo(Some(NO_BROADCAST_HASH))), rightHint = None)
+ val joinPlan = Join(NoStatsUnaryNode(targetTableProj), sourceTableProj, joinType, Some(cond), joinHint)
+
+ val deleteRowValues = buildDeltaDeleteRowValues(rowAttrs, rowIdAttrs)
+ val metadataReadAttrs = readAttrs.filterNot(relation.outputSet.contains)
+
+ val matchedConditions = matchedActions.map(actionCondition)
+ val matchedOutputs = matchedActions.map { action =>
Review comment:
Fixed.
--
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
[GitHub] [iceberg] szehon-ho commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r806356606
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
+ val joinType = if (notMatchedActions.isEmpty) Inner else RightOuter
+ val joinHint = JoinHint(leftHint = Some(HintInfo(Some(NO_BROADCAST_HASH))), rightHint = None)
+ val joinPlan = Join(NoStatsUnaryNode(targetTableProj), sourceTableProj, joinType, Some(cond), joinHint)
+
+ val deleteRowValues = buildDeltaDeleteRowValues(rowAttrs, rowIdAttrs)
+ val metadataReadAttrs = readAttrs.filterNot(relation.outputSet.contains)
+
+ val matchedConditions = matchedActions.map(actionCondition)
+ val matchedOutputs = matchedActions.map { action =>
+ deltaActionOutput(action, deleteRowValues, metadataReadAttrs)
+ }
+
+ val notMatchedConditions = notMatchedActions.map(actionCondition)
+ val notMatchedOutputs = notMatchedActions.map { action =>
Review comment:
I guess on the topic of oneline, this can as well?
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
Review comment:
Was also trying to compare the two code-paths. Noticed this has a 'later' and the other does not, the 'later' seems to be more clear. Actually wondering, did you consider putting some of these repeated codes/comments into methods
```
val sourceTableProj = sourceTableProj(source)
val joinPlan = joinPlan(sourceTableProj, targetTableProj, cond, notMatchedActions)
```
##########
File path: spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkDistributionAndOrderingUtil.java
##########
@@ -171,12 +194,16 @@ private static Distribution buildPositionDeleteUpdateDistribution(DistributionMo
}
}
- public static SortOrder[] buildPositionDeltaOrdering(Table table, Command command, Distribution distribution) {
+ public static SortOrder[] buildPositionDeltaOrdering(Table table, Command command) {
// the spec requires position delete files to be sorted by file and pos
+ SortOrder[] deleteOrdering = {SPEC_ID_ORDER, PARTITION_ORDER, FILE_PATH_ORDER, ROW_POSITION_ORDER};
+
if (command == DELETE || command == UPDATE) {
- return new SortOrder[]{SPEC_ID_ORDER, PARTITION_ORDER, FILE_PATH_ORDER, ROW_POSITION_ORDER};
+ return deleteOrdering;
} else {
- throw new IllegalArgumentException("Only position deletes and updates are currently supported");
+ // all metadata columns like _spec_id, _file, _pos will be null for new data records
Review comment:
Optional: just noticed we throw IllegalArgumentException in other places where we have an unknown command, so was just wondering would it be clearer/more consistent to mark this case as 'MERGE' and throw the exception like other case? Up to you
--
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
[GitHub] [iceberg] szehon-ho commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r806356103
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
Review comment:
Was also trying to compare the two code-paths. Noticed this has a 'later' in the comment and the other does not, the 'later' seems to be more clear. Actually wondering, did you consider putting some of these repeated codes/comments into single methods/scaladoc?
```
val sourceTableProj = sourceTableProj(source)
val joinPlan = joinPlan(sourceTableProj, targetTableProj, cond, notMatchedActions)
```
--
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
[GitHub] [iceberg] rdblue commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r805431271
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is HASH -> unspecified distribution +
Review comment:
Why is the distribution unspecified when mode is hash here? Shouldn't this hash distribute by original data file (or maybe original partition), then by the new partition? Or is the fear that this will create too many small tasks by dividing original data files by the new partitioning?
If that's the case, then it seems like this is optimizing for the case where you're running a MERGE with data from an old partition spec. I'd rather optimize for the case where the partition spec matches.
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r806228530
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
+ val joinType = if (notMatchedActions.isEmpty) Inner else RightOuter
+ val joinHint = JoinHint(leftHint = Some(HintInfo(Some(NO_BROADCAST_HASH))), rightHint = None)
+ val joinPlan = Join(NoStatsUnaryNode(targetTableProj), sourceTableProj, joinType, Some(cond), joinHint)
+
+ val deleteRowValues = buildDeltaDeleteRowValues(rowAttrs, rowIdAttrs)
+ val metadataReadAttrs = readAttrs.filterNot(relation.outputSet.contains)
+
+ val matchedConditions = matchedActions.map(actionCondition)
+ val matchedOutputs = matchedActions.map { action =>
Review comment:
I like it on a single line too. It is just the line limit in Spark is 100 (instead of 120 in Iceberg) and I copied the code from there. I'll make it consistent in Iceberg, at least.
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
Review comment:
Correct, I just did not want to duplicate all possible values of the write property. If merge mode is not set, use the write mode and adapt it to cover deletes.
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
Review comment:
Correct, I just did not want to duplicate all possible values of the write property. If the merge mode is not set, use the write mode and adapt it to cover deletes.
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
Review comment:
Good idea, will add a comment.
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is HASH -> unspecified distribution +
Review comment:
Well, I am not sure. I like that our merge and write logic are consistent right now. My hope was that AQE would coalesce tasks as needed to avoid a huge number of small writing tasks (and hence a huge number of delete files). I think AQE should behave better than a round-robin distribution. This case is about unpartitioned tables so we will most like produce at a single delete file per writing task (that shouldn't be that bad). As long as we don't have a huge number of writing tasks, we should be fine, I guess?
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is HASH -> unspecified distribution +
Review comment:
Well, I am not sure. I like that our merge and write logic are consistent right now. My hope was that AQE would coalesce tasks as needed to avoid a huge number of small writing tasks (and hence a huge number of delete files). I think AQE should behave better than a round-robin distribution. This case is about unpartitioned tables so we will most likely produce at a single delete file per writing task (that shouldn't be that bad). As long as we don't have a huge number of writing tasks, we should be fine, I guess?
##########
File path: spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMerge.java
##########
@@ -84,17 +86,22 @@ public void testMergeWithStaticPredicatePushDown() {
sql("ALTER TABLE %s ADD PARTITION FIELD dep", tableName);
- append(tableName,
- "{ \"id\": 1, \"dep\": \"software\" }\n" +
- "{ \"id\": 11, \"dep\": \"software\" }\n" +
- "{ \"id\": 1, \"dep\": \"hr\" }");
+ // add a data file to the 'software' partition
+ append(tableName, "{ \"id\": 1, \"dep\": \"software\" }");
+
+ // add a data file to the 'hr' partition
+ append(tableName, "{ \"id\": 1, \"dep\": \"hr\" }");
Table table = validationCatalog.loadTable(tableIdent);
Snapshot snapshot = table.currentSnapshot();
String dataFilesCount = snapshot.summary().get(SnapshotSummary.TOTAL_DATA_FILES_PROP);
Assert.assertEquals("Must have 2 files before MERGE", "2", dataFilesCount);
+ // remove the data file from the 'hr' partition to ensure it is not scanned
+ DataFile dataFile = Iterables.getOnlyElement(snapshot.addedFiles());
+ table.io().deleteFile(dataFile.path().toString());
Review comment:
Yeah, this will be helpful for future use cases. Let me spend some time on this one now.
--
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
[GitHub] [iceberg] rdblue commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r805430946
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
Review comment:
I think it may be worth an additional comment at the top of this section that says: `IMPORTANT: metadata columns like _spec_id and _partition are null for new rows`.
I think what this means is:
* Deletes and updates will be clustered by original partition and _file, and additionally sorted by _pos
* Inserts will effectively be distributed by id, data and sorted by id, data
This holds roughly true for all of the partitioning and sort order cases.
--
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
[GitHub] [iceberg] rdblue commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r805430946
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
Review comment:
I think it may be worth an additional comment at the top of this section that says: `IMPORTANT: metadata columns like _spec_id and _partition are null for new rows`.
I think what this means is:
* Deletes and updates will be clustered by original partition and _file, and additionally sorted by _pos
* Inserts will effectively be distributed by id, data and sorted by id, data
This holds roughly true for all of the cases.
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r806236038
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
Review comment:
Correct, I just did not want to duplicate all possible values of the write property. If the merge mode is not set, use the write mode and adapt it to cover deletes.
--
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
[GitHub] [iceberg] szehon-ho commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
szehon-ho commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r806356606
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
+ val joinType = if (notMatchedActions.isEmpty) Inner else RightOuter
+ val joinHint = JoinHint(leftHint = Some(HintInfo(Some(NO_BROADCAST_HASH))), rightHint = None)
+ val joinPlan = Join(NoStatsUnaryNode(targetTableProj), sourceTableProj, joinType, Some(cond), joinHint)
+
+ val deleteRowValues = buildDeltaDeleteRowValues(rowAttrs, rowIdAttrs)
+ val metadataReadAttrs = readAttrs.filterNot(relation.outputSet.contains)
+
+ val matchedConditions = matchedActions.map(actionCondition)
+ val matchedOutputs = matchedActions.map { action =>
+ deltaActionOutput(action, deleteRowValues, metadataReadAttrs)
+ }
+
+ val notMatchedConditions = notMatchedActions.map(actionCondition)
+ val notMatchedOutputs = notMatchedActions.map { action =>
Review comment:
I guess on the topic of oneline, this can as well?
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
Review comment:
Was also trying to compare the two code-paths. Noticed this has a 'later' and the other does not, the 'later' seems to be more clear. Actually wondering, did you consider putting some of these repeated codes/comments into methods
```
val sourceTableProj = sourceTableProj(source)
val joinPlan = joinPlan(sourceTableProj, targetTableProj, cond, notMatchedActions)
```
##########
File path: spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkDistributionAndOrderingUtil.java
##########
@@ -171,12 +194,16 @@ private static Distribution buildPositionDeleteUpdateDistribution(DistributionMo
}
}
- public static SortOrder[] buildPositionDeltaOrdering(Table table, Command command, Distribution distribution) {
+ public static SortOrder[] buildPositionDeltaOrdering(Table table, Command command) {
// the spec requires position delete files to be sorted by file and pos
+ SortOrder[] deleteOrdering = {SPEC_ID_ORDER, PARTITION_ORDER, FILE_PATH_ORDER, ROW_POSITION_ORDER};
+
if (command == DELETE || command == UPDATE) {
- return new SortOrder[]{SPEC_ID_ORDER, PARTITION_ORDER, FILE_PATH_ORDER, ROW_POSITION_ORDER};
+ return deleteOrdering;
} else {
- throw new IllegalArgumentException("Only position deletes and updates are currently supported");
+ // all metadata columns like _spec_id, _file, _pos will be null for new data records
Review comment:
Optional: just noticed we throw IllegalArgumentException in other places where we have an unknown command, so was just wondering would it be clearer/more consistent to mark this case as 'MERGE' and throw the exception like other case? Up to you
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
Review comment:
Was also trying to compare the two code-paths. Noticed this has a 'later' in the comment and the other does not, the 'later' seems to be more clear. Actually wondering, did you consider putting some of these repeated codes/comments into methods
```
val sourceTableProj = sourceTableProj(source)
val joinPlan = joinPlan(sourceTableProj, targetTableProj, cond, notMatchedActions)
```
##########
File path: spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkDistributionAndOrderingUtil.java
##########
@@ -171,12 +194,16 @@ private static Distribution buildPositionDeleteUpdateDistribution(DistributionMo
}
}
- public static SortOrder[] buildPositionDeltaOrdering(Table table, Command command, Distribution distribution) {
+ public static SortOrder[] buildPositionDeltaOrdering(Table table, Command command) {
// the spec requires position delete files to be sorted by file and pos
+ SortOrder[] deleteOrdering = {SPEC_ID_ORDER, PARTITION_ORDER, FILE_PATH_ORDER, ROW_POSITION_ORDER};
+
if (command == DELETE || command == UPDATE) {
- return new SortOrder[]{SPEC_ID_ORDER, PARTITION_ORDER, FILE_PATH_ORDER, ROW_POSITION_ORDER};
+ return deleteOrdering;
} else {
- throw new IllegalArgumentException("Only position deletes and updates are currently supported");
+ // all metadata columns like _spec_id, _file, _pos will be null for new data records
Review comment:
Optional: just noticed we throw IllegalArgumentException in other places where we have an unknown command, so was just wondering would it be clearer/more consistent to put this in a MERGE case block and throw exception in unknown command case? Up to you
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
Review comment:
Was also trying to compare the two code-paths. Noticed this has a 'later' in the comment and the other does not, the 'later' seems to be more clear. Actually wondering, did you consider putting some of these repeated codes/comments into single methods/scaladoc?
```
val sourceTableProj = sourceTableProj(source)
val joinPlan = joinPlan(sourceTableProj, targetTableProj, cond, notMatchedActions)
```
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r807302964
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
Review comment:
Removed "later" for consistency in comments. The comments in copy-on-write and merge-on-read are slightly different as different join type are used. I'll take a look at what extra methods we can introduce to simplify this. I did that in a few places already.
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
Review comment:
Added a 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
[GitHub] [iceberg] rdblue commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r806309925
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is HASH -> unspecified distribution +
Review comment:
We chatted offline and decided that clustering by _spec_id, _partition, and _file is a good idea to avoid a large number of delete files.
--
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
[GitHub] [iceberg] aokolnychyi merged pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi merged pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r801000095
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is HASH -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is RANGE -> RANGE DISTRIBUTE BY _spec_id, _partition, _file, id, data +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ //
+ // PARTITIONED BY date, days(ts) UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDERED BY _spec_id, _partition, _file, _pos, date, days(ts)
+ // merge mode is HASH -> CLUSTER BY _spec_id, _partition, date, days(ts) +
Review comment:
This means we may split records for a partition into a number of tasks if that partition is in the old spec and is not aligned with the current partitioning (i.e. `date, days(ts)` in this case). Not sure how to avoid this, though.
--
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
[GitHub] [iceberg] rdblue commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r805428319
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
+ val joinType = if (notMatchedActions.isEmpty) Inner else RightOuter
+ val joinHint = JoinHint(leftHint = Some(HintInfo(Some(NO_BROADCAST_HASH))), rightHint = None)
+ val joinPlan = Join(NoStatsUnaryNode(targetTableProj), sourceTableProj, joinType, Some(cond), joinHint)
+
+ val deleteRowValues = buildDeltaDeleteRowValues(rowAttrs, rowIdAttrs)
+ val metadataReadAttrs = readAttrs.filterNot(relation.outputSet.contains)
+
+ val matchedConditions = matchedActions.map(actionCondition)
+ val matchedOutputs = matchedActions.map { action =>
+ deltaActionOutput(action, deleteRowValues, metadataReadAttrs)
+ }
+
+ val notMatchedConditions = notMatchedActions.map(actionCondition)
+ val notMatchedOutputs = notMatchedActions.map { action =>
+ deltaActionOutput(action, deleteRowValues, metadataReadAttrs)
+ }
+
+ val operationTypeAttr = AttributeReference(OPERATION_COLUMN, IntegerType, nullable = false)()
+ val rowFromSourceAttr = resolveAttrRef(ROW_FROM_SOURCE_REF, joinPlan)
+ val rowFromTargetAttr = resolveAttrRef(ROW_FROM_TARGET_REF, joinPlan)
+
+ // merged rows must contain values for the operation type and all read attrs
+ val mergeRowsOutput = buildMergeRowsOutput(
+ matchedOutputs,
+ notMatchedOutputs,
+ attrs = operationTypeAttr +: readAttrs)
+
+ val mergeRows = MergeRows(
+ isSourceRowPresent = IsNotNull(rowFromSourceAttr),
+ isTargetRowPresent = if (notMatchedActions.isEmpty) TrueLiteral else IsNotNull(rowFromTargetAttr),
+ matchedConditions = matchedConditions,
+ matchedOutputs = matchedOutputs,
+ notMatchedConditions = notMatchedConditions,
+ notMatchedOutputs = notMatchedOutputs,
+ targetOutput = Nil,
Review comment:
Would be nice to have a comment to explain this: "only needed if emitting unmatched target rows"
--
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
[GitHub] [iceberg] rdblue commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r805427254
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
+ val joinType = if (notMatchedActions.isEmpty) Inner else RightOuter
+ val joinHint = JoinHint(leftHint = Some(HintInfo(Some(NO_BROADCAST_HASH))), rightHint = None)
+ val joinPlan = Join(NoStatsUnaryNode(targetTableProj), sourceTableProj, joinType, Some(cond), joinHint)
+
+ val deleteRowValues = buildDeltaDeleteRowValues(rowAttrs, rowIdAttrs)
+ val metadataReadAttrs = readAttrs.filterNot(relation.outputSet.contains)
+
+ val matchedConditions = matchedActions.map(actionCondition)
+ val matchedOutputs = matchedActions.map { action =>
+ deltaActionOutput(action, deleteRowValues, metadataReadAttrs)
+ }
+
+ val notMatchedConditions = notMatchedActions.map(actionCondition)
+ val notMatchedOutputs = notMatchedActions.map { action =>
+ deltaActionOutput(action, deleteRowValues, metadataReadAttrs)
+ }
+
+ val operationTypeAttr = AttributeReference(OPERATION_COLUMN, IntegerType, nullable = false)()
+ val rowFromSourceAttr = resolveAttrRef(ROW_FROM_SOURCE_REF, joinPlan)
+ val rowFromTargetAttr = resolveAttrRef(ROW_FROM_TARGET_REF, joinPlan)
+
+ // merged rows must contain values for the operation type and all read attrs
+ val mergeRowsOutput = buildMergeRowsOutput(
+ matchedOutputs,
+ notMatchedOutputs,
+ attrs = operationTypeAttr +: readAttrs)
Review comment:
This also fits inline if you want to match the replace code. (minor)
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r806263344
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is HASH -> unspecified distribution +
Review comment:
Well, I am not sure. I like that our merge and write logic are consistent right now. My hope was that AQE would coalesce tasks as needed to avoid a huge number of small writing tasks (and hence a huge number of delete files). I think AQE should behave better than a round-robin distribution. This case is about unpartitioned tables so we will most likely produce at a single delete file per writing task (that shouldn't be that bad). As long as we don't have a huge number of writing tasks, we should be fine, I guess?
--
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
[GitHub] [iceberg] aokolnychyi commented on pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#issuecomment-1039602241
@rdblue, you are right. There will be no `RowLevelOperationTable` when a MERGE gets converted into an INSERT. We will use the regular write distribution and ordering in that case.
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r807300575
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
+ val joinType = if (notMatchedActions.isEmpty) Inner else RightOuter
+ val joinHint = JoinHint(leftHint = Some(HintInfo(Some(NO_BROADCAST_HASH))), rightHint = None)
+ val joinPlan = Join(NoStatsUnaryNode(targetTableProj), sourceTableProj, joinType, Some(cond), joinHint)
+
+ val deleteRowValues = buildDeltaDeleteRowValues(rowAttrs, rowIdAttrs)
+ val metadataReadAttrs = readAttrs.filterNot(relation.outputSet.contains)
+
+ val matchedConditions = matchedActions.map(actionCondition)
+ val matchedOutputs = matchedActions.map { action =>
+ deltaActionOutput(action, deleteRowValues, metadataReadAttrs)
+ }
+
+ val notMatchedConditions = notMatchedActions.map(actionCondition)
+ val notMatchedOutputs = notMatchedActions.map { action =>
Review comment:
Fixed.
--
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
[GitHub] [iceberg] rdblue commented on pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#issuecomment-1040865779
Awesome work, @aokolnychyi!
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r799917712
##########
File path: spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMerge.java
##########
@@ -84,17 +86,22 @@ public void testMergeWithStaticPredicatePushDown() {
sql("ALTER TABLE %s ADD PARTITION FIELD dep", tableName);
- append(tableName,
Review comment:
I had to adapt this test and use a similar trick like in a few other places where we delete files we don't expect to query. Otherwise, I'd need to have different checks for copy-on-write and merge-on-read.
--
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
[GitHub] [iceberg] rdblue commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r805426755
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
##########
@@ -226,6 +231,82 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand {
ReplaceData(writeRelation, mergeRows, relation)
}
+ // build a rewrite plan for sources that support row deltas
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ source: LogicalPlan,
+ cond: Expression,
+ matchedActions: Seq[MergeAction],
+ notMatchedActions: Seq[MergeAction]): WriteDelta = {
+
+ // resolve all needed attrs (e.g. row ID and any required metadata attrs)
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operationTable.operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
+
+ // construct a scan relation and include all required metadata columns
+ val readRelation = buildReadRelation(relation, operationTable, metadataAttrs, rowIdAttrs)
+ val readAttrs = readRelation.output
+
+ // project an extra column to check if a target row exists after the join
+ val targetTableProjExprs = readAttrs :+ Alias(TrueLiteral, ROW_FROM_TARGET)()
+ val targetTableProj = Project(targetTableProjExprs, readRelation)
+
+ // project an extra column to check if a source row exists after the join
+ val sourceTableProjExprs = source.output :+ Alias(TrueLiteral, ROW_FROM_SOURCE)()
+ val sourceTableProj = Project(sourceTableProjExprs, source)
+
+ // use inner join if there is no NOT MATCHED action, unmatched source rows can be discarded
+ // use right outer join in all other cases, unmatched source rows may be needed
+ // also disable broadcasts for the target table to perform the cardinality check later
+ val joinType = if (notMatchedActions.isEmpty) Inner else RightOuter
+ val joinHint = JoinHint(leftHint = Some(HintInfo(Some(NO_BROADCAST_HASH))), rightHint = None)
+ val joinPlan = Join(NoStatsUnaryNode(targetTableProj), sourceTableProj, joinType, Some(cond), joinHint)
+
+ val deleteRowValues = buildDeltaDeleteRowValues(rowAttrs, rowIdAttrs)
+ val metadataReadAttrs = readAttrs.filterNot(relation.outputSet.contains)
+
+ val matchedConditions = matchedActions.map(actionCondition)
+ val matchedOutputs = matchedActions.map { action =>
Review comment:
Nit: in the replace code, this is a one-liner. You could do the same thing here:
```scala
val matchedOutputs = matchedActions.map(deltaActionOutput(_, deleteRowValues, metadataReadAttrs))
```
Not a big deal, I'm just going through both and looking for changes so this looked a little odd. Same thing with `notMatchedOutputs`.
--
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
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r806236038
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
Review comment:
Correct, I just did not want to duplicate all possible values of the write property. If merge mode is not set, use the write mode and adapt it to cover deletes.
--
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
[GitHub] [iceberg] rdblue commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r805431271
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is HASH -> unspecified distribution +
Review comment:
~~Why is the distribution unspecified when mode is hash here? Shouldn't this hash distribute by original data file (or maybe original partition), then by the new partition? Or is the fear that this will create too many small tasks by dividing original data files by the new partitioning?~~
~~If that's the case, then it seems like this is optimizing for the case where you're running a MERGE with data from an old partition spec. I'd rather optimize for the case where the partition spec matches.~~
--
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
[GitHub] [iceberg] rdblue commented on a change in pull request #4047: Spark 3.2: Implement merge-on-read MERGE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #4047:
URL: https://github.com/apache/iceberg/pull/4047#discussion_r806309925
##########
File path: spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkDistributionAndOrderingUtil.java
##########
@@ -1567,6 +1567,370 @@ public void testRangePositionDeltaUpdatePartitionedTable() {
table, UPDATE, expectedDistribution, SPEC_ID_PARTITION_FILE_POSITION_ORDERING);
}
+ // ==================================================================================
+ // Distribution and ordering for merge-on-read MERGE operations with position deletes
+ // ==================================================================================
+ //
+ // UNPARTITIONED UNORDERED
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is HASH -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ // merge mode is RANGE -> unspecified distribution + LOCALLY ORDER BY _spec_id, _partition, _file, _pos
+ //
+ // UNPARTITIONED ORDERED BY id, data
+ // -------------------------------------------------------------------------
+ // merge mode is NOT SET -> rely on write distribution and ordering as a basis
+ // merge mode is NONE -> unspecified distribution +
+ // LOCALLY ORDER BY _spec_id, _partition, _file, _pos, id, data
+ // merge mode is HASH -> unspecified distribution +
Review comment:
We chatted offline and decided that clustering by _spec_id, _partition, and _file is a good idea to avoid a large number of delete files.
--
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