Spark 3.3: Dataset writes for position deletes #7029

szehon-ho · 2023-03-06T20:39:08Z

This is the last pre-requisite for implementing RewriteDeleteFiles.

It allows dataset writes to the position_deletes metadata table, on condition of rewritte_file_set_id being set (ie, it comes from Iceberg's internal use).

Part of this pr, which is simple refactoring, is already split out into: #6924

szehon-ho · 2023-03-06T20:42:28Z

core/src/main/java/org/apache/iceberg/io/DeleteSchemaUtil.java

@@ -29,7 +29,7 @@ private static Schema pathPosSchema(Schema rowSchema) {
    return new Schema(
        MetadataColumns.DELETE_FILE_PATH,
        MetadataColumns.DELETE_FILE_POS,
-        Types.NestedField.required(
+        Types.NestedField.optional(


This was necessary for the writer to allow writing position deletes with "row", but still be ok when there is null "row".

Currently, the writer code either uses a schema with required "row" field as is here, or a schema without the "row" field (see posPathSchema method just below). This one with required row field is actually not used, so changing to optional should have no impact.

This is actually more in line with the position-delete schema in the spec, where "row" is optional.

Update: Looks like a few GenericWriter depend on this, to throw exception if null rows passed in. This will thus be a change of behavior , but backward compatible.

I have some concerns about this change, the Delete Formats spec says the reason why this column type should be required is to make sure the statistics of the deleted row values is accurate. I think the reason to make sure the statistics are accurate is because of the manifest reader will use them to filter delete files:

So I think if this type is changed to optional, the statistics will become unreliable, which may cause the delete manifest entry to be incorrectly filtered? This is just my understanding of the spec, but I'm not sure.

Yea you are right, this is tricky. It says this in spec:

2147483544 row required struct<...> [1] Deleted row values. Omit the column when not storing deleted rows.

When present in the delete file, row is required because all delete entries must include the row values.

So either, entire position delete file has 'row', or entire file does not have 'row'. (Currently it seems Spark does not set 'row' at all, ref: https://github.com/apache/iceberg/blob/master/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java#L436)

I somehow need a way, when compacting delete files, to know whether the original position file all have rows or not. I am not sure at the moment how to get this

I updated the pr with a fix for this , the idea from chatting with @RussellSpitzer offline. I made SparkWrite.DeleteWriter now a fan-out that can redirect deletes to two files, one that either has 'row' as required struct, or no 'row' at all. In most case only one will be chosen. Thanks for the initial comment.

szehon-ho · 2023-03-09T01:22:13Z

Rebase on updated version of #6924

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/ScanTaskSetManager.java

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/AbstractFileRewriteCoordinator.java

aokolnychyi · 2023-03-15T05:37:25Z

I went through the change. Let me do a detailed review round with fresh eyes tomorrow.

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkWriteBuilder.java

szehon-ho · 2023-03-17T01:44:29Z

Made suggested changes (refactored to new classe SparkPositionDeletesRewrite from SparkWrite).

Notes, new classes drop some unused codes from the previous path, like reportMetrics method and cleanupOnAbort flag to control abort behavior. I assume we can go back to this when we implement the commit manager part, as of now its not clear whether we need this or not.

aokolnychyi

This is getting close. I did a detailed round. Will check tests with fresh eyes tomorrow.

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/AbstractFileRewriteCoordinator.java

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeletesRewrite.java

.../spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeletesRewriteBuilder.java

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java

aokolnychyi

Almost there.

core/src/main/java/org/apache/iceberg/PartitionsTable.java

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/BaseFileRewriteCoordinator.java

.../spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeletesRewriteBuilder.java

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeletesRewrite.java

aokolnychyi

LGTM, I left a few minor comments. Feel free to merge whenever you are ready, @szehon-ho.
Nice work!

aokolnychyi · 2023-03-31T05:41:29Z

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/BaseFileRewriteCoordinator.java

+
+abstract class BaseFileRewriteCoordinator<F extends ContentFile<F>> {
+
+  private static final Logger LOG = LoggerFactory.getLogger(FileRewriteCoordinator.class);


I think we are using a wrong class for logging. It should be BaseFileRewriteCoordinator.

Good catch, fixed

aokolnychyi · 2023-03-31T06:00:09Z

.../spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeletesRewriteBuilder.java

+    Preconditions.checkArgument(
+        fileSetId != null, "position_deletes table can only be written by RewriteDeleteFiles");
+    Preconditions.checkArgument(
+        writeConf.handleTimestampWithoutZone()


I think this part would be easier to read if we define fileSetId and handleTimestampWithoutZone instance variables, similar to what we have in SparkWriteBuilder.

I think its a bit harder to read if we define in Constructor, like SparkWriteBuilder, as its a bit detached from this code. I rewrote to define the vars at beginning of the method.

aokolnychyi · 2023-03-31T06:01:14Z

.../spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeletesRewriteBuilder.java

+    partitions.addAll(tasks.stream().map(ContentScanTask::partition).collect(Collectors.toList()));
+    Preconditions.checkArgument(
+        partitions.size() == 1,
+        "All scan tasks of %s are expected to have the same partition",


Did we miss , but got %s at the end to include partitions?

Good catch, done

aokolnychyi · 2023-03-31T06:12:17Z

.../spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeletesRewriteBuilder.java

+    String fileSetId = writeConf.rewrittenFileSetId();
+
+    Preconditions.checkArgument(
+        fileSetId != null, "position_deletes table can only be written by RewriteDeleteFiles");


I don't think there is RewriteDeleteFiles.
What about a more generic message like Can only write to %s via actions", table.name()?

aokolnychyi · 2023-03-31T06:19:04Z

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/FileRewriteCoordinator.java

-    resultMap.remove(id);
-  }
-
-  public Set<String> fetchSetIDs(Table table) {


I believe we renamed it to be fetchSetIds instead of fetchSetIDs, so have to keep the old method for now.

aokolnychyi · 2023-03-31T06:22:18Z

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeletesRewrite.java

+
+      SparkFileWriterFactory writerFactoryWithRow =
+          SparkFileWriterFactory.builderFor(table)
+              .dataSchema(writeSchema)


I am not sure we need to set these dataXXX methods since we are not writing any data (here).

aokolnychyi · 2023-03-31T06:25:25Z

.../spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeletesRewriteBuilder.java

+
+  private StructLike partition(String fileSetId, List<PositionDeletesScanTask> tasks) {
+    StructLikeSet partitions = StructLikeSet.create(tasks.get(0).spec().partitionType());
+    partitions.addAll(tasks.stream().map(ContentScanTask::partition).collect(Collectors.toList()));


nit: I think you can use forEach instead of a temp list.

tasks.stream().map(ContentScanTask::partition).forEach(partitions::add);

In any case, I like what you did here.

szehon-ho · 2023-04-05T17:28:26Z

Merged, thanks @aokolnychyi for detailed review, and @zhongyujiang @amogh-jahagirdar for initial reviews

github-actions bot added core spark labels Mar 6, 2023

szehon-ho changed the title ~~Allow writes for position deletes~~ Spark 3.3: Dataset writes for position deletes Mar 6, 2023

szehon-ho commented Mar 6, 2023

View reviewed changes

szehon-ho force-pushed the position_delete_write_master branch from 8775b2a to 1364baf Compare March 6, 2023 21:59

github-actions bot added data ORC and removed ORC labels Mar 6, 2023

szehon-ho force-pushed the position_delete_write_master branch from 9fecbba to 38cc095 Compare March 9, 2023 01:21

amogh-jahagirdar reviewed Mar 13, 2023

View reviewed changes

szehon-ho force-pushed the position_delete_write_master branch from 38cc095 to fb29954 Compare March 13, 2023 17:33

szehon-ho commented Mar 13, 2023

View reviewed changes

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/AbstractFileRewriteCoordinator.java Outdated Show resolved Hide resolved

aokolnychyi reviewed Mar 15, 2023

View reviewed changes

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java Outdated Show resolved Hide resolved

aokolnychyi reviewed Mar 15, 2023

View reviewed changes

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java Outdated Show resolved Hide resolved

aokolnychyi reviewed Mar 15, 2023

View reviewed changes

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java Outdated Show resolved Hide resolved

aokolnychyi reviewed Mar 15, 2023

View reviewed changes

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java Outdated Show resolved Hide resolved

aokolnychyi reviewed Mar 15, 2023

View reviewed changes

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkWriteBuilder.java Outdated Show resolved Hide resolved

szehon-ho mentioned this pull request Mar 16, 2023

iceberg v2 table cannot expire delete files after rewrite datafile action #5058

Closed

aokolnychyi reviewed Mar 21, 2023

View reviewed changes

szehon-ho force-pushed the position_delete_write_master branch 2 times, most recently from 29788fa to b02c011 Compare March 24, 2023 17:39

szehon-ho added 3 commits March 24, 2023 10:43

Support writing position deletes

ae6ca80

Review comments

ba03021

Review comments

652f37f

szehon-ho force-pushed the position_delete_write_master branch from b02c011 to 652f37f Compare March 24, 2023 18:13

szehon-ho added this to In progress in [Priority 1] Maintenance: Delete file compaction via automation Mar 24, 2023

aokolnychyi reviewed Mar 29, 2023

View reviewed changes

szehon-ho added 2 commits March 29, 2023 22:49

Review comments

c621f1f

More review comments

e0b75ad

aokolnychyi approved these changes Mar 31, 2023

View reviewed changes

[Priority 1] Maintenance: Delete file compaction automation moved this from In progress to Reviewer approved Mar 31, 2023

szehon-ho added 3 commits April 4, 2023 16:08

Address comments

d60fe12

More review comments

134149a

Change deprecation version now that 1.2 is released

116df4d

szehon-ho merged commit 22d29a5 into apache:master Apr 5, 2023
32 checks passed

[Priority 1] Maintenance: Delete file compaction automation moved this from Reviewer approved to Done Apr 5, 2023

ericlgoodman pushed a commit to ericlgoodman/iceberg that referenced this pull request Apr 12, 2023

Spark 3.3: Dataset writes for position deletes (apache#7029)

ad207c4

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Spark 3.3: Dataset writes for position deletes #7029

Spark 3.3: Dataset writes for position deletes #7029

szehon-ho commented Mar 6, 2023 •

edited

szehon-ho Mar 6, 2023 •

edited

szehon-ho Mar 6, 2023 •

edited

zhongyujiang Mar 7, 2023

szehon-ho Mar 7, 2023

szehon-ho Mar 8, 2023

szehon-ho commented Mar 9, 2023

aokolnychyi commented Mar 15, 2023

szehon-ho commented Mar 17, 2023

aokolnychyi left a comment

aokolnychyi left a comment

aokolnychyi left a comment

aokolnychyi Mar 31, 2023

szehon-ho Apr 4, 2023

aokolnychyi Mar 31, 2023

szehon-ho Apr 4, 2023

aokolnychyi Mar 31, 2023

szehon-ho Apr 4, 2023

aokolnychyi Mar 31, 2023

szehon-ho Apr 4, 2023

aokolnychyi Mar 31, 2023 •

edited

aokolnychyi Mar 31, 2023 •

edited

szehon-ho Apr 4, 2023

aokolnychyi Mar 31, 2023

szehon-ho Apr 4, 2023

szehon-ho commented Apr 5, 2023


		abstract class BaseFileRewriteCoordinator<F extends ContentFile<F>> {

		private static final Logger LOG = LoggerFactory.getLogger(FileRewriteCoordinator.class);

Spark 3.3: Dataset writes for position deletes #7029

Spark 3.3: Dataset writes for position deletes #7029

Conversation

szehon-ho commented Mar 6, 2023 • edited

szehon-ho Mar 6, 2023 • edited

Choose a reason for hiding this comment

szehon-ho Mar 6, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

szehon-ho commented Mar 9, 2023

aokolnychyi commented Mar 15, 2023

szehon-ho commented Mar 17, 2023

aokolnychyi left a comment

Choose a reason for hiding this comment

aokolnychyi left a comment

Choose a reason for hiding this comment

aokolnychyi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi Mar 31, 2023 • edited

Choose a reason for hiding this comment

aokolnychyi Mar 31, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

szehon-ho commented Apr 5, 2023

szehon-ho commented Mar 6, 2023 •

edited

szehon-ho Mar 6, 2023 •

edited

szehon-ho Mar 6, 2023 •

edited

aokolnychyi Mar 31, 2023 •

edited

aokolnychyi Mar 31, 2023 •

edited