aokolnychyi commented on code in PR #7637:
URL: https://github.com/apache/iceberg/pull/7637#discussion_r1201187215
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java:
##########
@@ -262,7 +278,50 @@ private DistributionMode defaultWriteDistributionMode() {
}
}
- public DistributionMode deleteDistributionMode() {
+ public SparkWriteRequirements copyOnWriteRequirements(Command command) {
+ if (ignoreTableDistributionAndOrdering()) {
+ LOG.info("Skipping distribution/ordering: disabled per job
configuration");
+ return SparkWriteRequirements.EMPTY;
+ }
+
+ return SparkWriteUtil.copyOnWriteRequirements(
+ table, command, copyOnWriteDistributionMode(command),
fanoutWriterEnabled());
+ }
+
+ @VisibleForTesting
+ DistributionMode copyOnWriteDistributionMode(Command command) {
+ switch (command) {
+ case DELETE:
+ return deleteDistributionMode();
+ case UPDATE:
+ return updateDistributionMode();
+ case MERGE:
+ return copyOnWriteMergeDistributionMode();
+ default:
+ throw new IllegalArgumentException("Unexpected command: " + command);
+ }
+ }
+
+ public SparkWriteRequirements positionDeltaRequirements(Command command) {
+ return SparkWriteUtil.positionDeltaRequirements(
Review Comment:
I have a follow-up PR that adds a fanout position delete writer. It buffers
deletes into a bitmap and then produces a sorted file when closed. In the
future, we either remove the requirement for position deletes to be sorted or
add support for Puffin delete files that would persist bitmaps.
##########
spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java:
##########
@@ -262,7 +278,50 @@ private DistributionMode defaultWriteDistributionMode() {
}
}
- public DistributionMode deleteDistributionMode() {
+ public SparkWriteRequirements copyOnWriteRequirements(Command command) {
+ if (ignoreTableDistributionAndOrdering()) {
+ LOG.info("Skipping distribution/ordering: disabled per job
configuration");
+ return SparkWriteRequirements.EMPTY;
+ }
+
+ return SparkWriteUtil.copyOnWriteRequirements(
+ table, command, copyOnWriteDistributionMode(command),
fanoutWriterEnabled());
+ }
+
+ @VisibleForTesting
+ DistributionMode copyOnWriteDistributionMode(Command command) {
+ switch (command) {
+ case DELETE:
+ return deleteDistributionMode();
+ case UPDATE:
+ return updateDistributionMode();
+ case MERGE:
+ return copyOnWriteMergeDistributionMode();
+ default:
+ throw new IllegalArgumentException("Unexpected command: " + command);
+ }
+ }
+
+ public SparkWriteRequirements positionDeltaRequirements(Command command) {
+ return SparkWriteUtil.positionDeltaRequirements(
Review Comment:
I have a follow-up PR that adds a fanout position delete writer. It buffers
deletes into a bitmap and then produces a sorted file when closed. In the
future, we either remove the requirement for position deletes to be sorted or
add support for Puffin delete files that would persist bitmaps. I am already
looking into that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
