Repository: apache/iceberg (language: python)

Issue:
Flink: Apply DeleteGranularity for writes
This PR makes sure that the DeleteGranularity table configuration option is honoured by Flink.
If DeleteGranularity.FILE is used, then compaction commits will not conflict with Flink streaming writes.
Flink: Apply DeleteGranularity for writes
This PR makes sure that the DeleteGranularity table configuration option is honoured by Flink.
If DeleteGranularity.FILE is used, then compaction commits will not conflict with Flink streaming writes.


Discussion / hints:
@chenjunjiedada you might be interested. This is to leverage the targeted position delete file that @aokolnychyi implemented from PR #9384.

the background compaction/rewrite job should use the starting sequence number.  Then we can avoid the commit conflict btw Flink streaming and compaction jobs. 

```
  /**
   * If the compaction should use the sequence number of the snapshot at compaction start time for
   * new data files, instead of using the sequence number of the newly produced snapshot.
   *
   * <p>This avoids commit conflicts with updates that add newer equality deletes at a higher
   * sequence number.
   *
   * @param useStarting use starting sequence number if set to true
   * @return this for method chaining
   */
  public BaseRewriteDataFilesAction<ThisT> useStartingSequenceNumber(boolean useStarting) 
```
Getting to this PR today/tomorrow.
The markdown is flaky, and not related to the change at all.
After taking a closer look at `BaseTaskWriter`, I think we may have a correctness issue when encoding changes if the table contains multiple specs. Our current implementation of `BaseTaskWriter` assumes all writes are encoded against the current spec. However, what if there are some matching keys in other specs? Written deletes will be scoped to the current partition spec and will not apply to data in other specs, potentially missing to upsert some records. I think we would want to eventually migrate Flink to new writers that inherit `PartitioningWriter`. This is out of scope of this PR, however.

I am also not sure we need `ContinuousFileScopedPositionDeleteWriter`. I understand we want to solve the companion issue before Flink migrates to `PartitioningWriter` so we have to come up with a fix for `TaskWriter`. What about directly using `SortingPositionOnlyDeleteWriter` with file granularity in `BaseEqualityDeltaWriter`? We only need to pass a closure to create new position delete writers and that class should already sort deletes for us on the fly. We never needed to persist deleted rows in position deletes so no behavior change.
> After taking a closer look at `BaseTaskWriter`, I think we may have a correctness issue when encoding changes if the table contains multiple specs. Our current implementation of `BaseTaskWriter` assumes all writes are encoded against the current spec. However, what if there are some matching keys in other specs? Written deletes will be scoped to the current partition spec and will not apply to data in other specs, potentially missing to upsert some records. I think we would want to eventually migrate Flink to new writers that inherit `PartitioningWriter`. This is out of scope of this PR, however.

I agree, that we need to fix this. Also agree, that in a different PR.

> I am also not sure we need `ContinuousFileScopedPositionDeleteWriter`. I understand we want to solve the companion issue before Flink migrates to `PartitioningWriter` so we have to come up with a fix for `TaskWriter`. What about directly using `SortingPositionOnlyDeleteWriter` with file granularity in `BaseEqualityDeltaWriter`? We only need to pass a closure to create new position delete writers and that class should already sort deletes for us on the fly. We never needed to persist deleted rows in position deletes so no behaviour change.

Updated to use the `SortingPositionOnlyDeleteWriter`. Needed a small "fix" to prevent writing empty delete file, but I would like to ask you to double check this.

Also there is a behavioural change that the previous write rolled deletes on `DEFAULT_RECORDS_NUM_THRESHOLD`=100_000L threshold (no config, just a hard coded value). We can't do this with the new writer, but I think it is highly unlikely that anyone reached this in a production scenario. If we need this feature back, either we need to use the `SortedPosDeleteWriter`, or we can provide similar feature by `RollingFileWriter`. The issue is that `RollingFileWriter` uses a size threshold instead of a RECORDS_NUM threshold, and we need to rewrite/enhance the `RollingFileWriter` to use a closure to create a writer instead of the factory (as it does today).

I see 3 options:
1. Accept that there is no 100_000L threshold for the positional delete file size anymore
2. Use the old SortedPosDeleteWriter for PARTITION granularity. The FILE granularity is a new feature, and most probably will not create big files (as the data files tend to be bigger, than the delete files, and when a data file is rolled, the delete file rolls automatically) - so the behaviour change will affect only very-edge cases 
3. Invest in implementing the RollingFileWriter changes

I would vote for the 1st option, and would accept the 2nd if that is the consensus. I do not think that the 3rd worth the investment.

What do you think @stevenzwu, @aokolnychyi?

> Also there is a behavioural change that the previous write rolled deletes

@pvary, I don't think that threshold makes sense in this scenario. It was needed because we kept rows and positions in a set in the old writers. We are migrating to `SortingPositionOnlyDeleteWriter` that keeps track of positions in bitmaps, which would always fit in memory. We also shouldn't be worried about the size on disk as even millions of position deletes would require KBs in Parquet.
Merged to main.
Thanks for the reviews @aokolnychyi, @stevenzwu and @RussellSpitzer!
@chenjunjiedada you might be interested. This is to leverage the targeted position delete file that @aokolnychyi implemented from PR #9384.

the background compaction/rewrite job should use the starting sequence number.  Then we can avoid the commit conflict btw Flink streaming and compaction jobs. 

```
  /**
   * If the compaction should use the sequence number of the snapshot at compaction start time for
   * new data files, instead of using the sequence number of the newly produced snapshot.
   *
   * <p>This avoids commit conflicts with updates that add newer equality deletes at a higher
   * sequence number.
   *
   * @param useStarting use starting sequence number if set to true
   * @return this for method chaining
   */
  public BaseRewriteDataFilesAction<ThisT> useStartingSequenceNumber(boolean useStarting) 
```
Getting to this PR today/tomorrow.
The markdown is flaky, and not related to the change at all.
After taking a closer look at `BaseTaskWriter`, I think we may have a correctness issue when encoding changes if the table contains multiple specs. Our current implementation of `BaseTaskWriter` assumes all writes are encoded against the current spec. However, what if there are some matching keys in other specs? Written deletes will be scoped to the current partition spec and will not apply to data in other specs, potentially missing to upsert some records. I think we would want to eventually migrate Flink to new writers that inherit `PartitioningWriter`. This is out of scope of this PR, however.

I am also not sure we need `ContinuousFileScopedPositionDeleteWriter`. I understand we want to solve the companion issue before Flink migrates to `PartitioningWriter` so we have to come up with a fix for `TaskWriter`. What about directly using `SortingPositionOnlyDeleteWriter` with file granularity in `BaseEqualityDeltaWriter`? We only need to pass a closure to create new position delete writers and that class should already sort deletes for us on the fly. We never needed to persist deleted rows in position deletes so no behavior change.
> After taking a closer look at `BaseTaskWriter`, I think we may have a correctness issue when encoding changes if the table contains multiple specs. Our current implementation of `BaseTaskWriter` assumes all writes are encoded against the current spec. However, what if there are some matching keys in other specs? Written deletes will be scoped to the current partition spec and will not apply to data in other specs, potentially missing to upsert some records. I think we would want to eventually migrate Flink to new writers that inherit `PartitioningWriter`. This is out of scope of this PR, however.

I agree, that we need to fix this. Also agree, that in a different PR.

> I am also not sure we need `ContinuousFileScopedPositionDeleteWriter`. I understand we want to solve the companion issue before Flink migrates to `PartitioningWriter` so we have to come up with a fix for `TaskWriter`. What about directly using `SortingPositionOnlyDeleteWriter` with file granularity in `BaseEqualityDeltaWriter`? We only need to pass a closure to create new position delete writers and that class should already sort deletes for us on the fly. We never needed to persist deleted rows in position deletes so no behaviour change.

Updated to use the `SortingPositionOnlyDeleteWriter`. Needed a small "fix" to prevent writing empty delete file, but I would like to ask you to double check this.

Also there is a behavioural change that the previous write rolled deletes on `DEFAULT_RECORDS_NUM_THRESHOLD`=100_000L threshold (no config, just a hard coded value). We can't do this with the new writer, but I think it is highly unlikely that anyone reached this in a production scenario. If we need this feature back, either we need to use the `SortedPosDeleteWriter`, or we can provide similar feature by `RollingFileWriter`. The issue is that `RollingFileWriter` uses a size threshold instead of a RECORDS_NUM threshold, and we need to rewrite/enhance the `RollingFileWriter` to use a closure to create a writer instead of the factory (as it does today).

I see 3 options:
1. Accept that there is no 100_000L threshold for the positional delete file size anymore
2. Use the old SortedPosDeleteWriter for PARTITION granularity. The FILE granularity is a new feature, and most probably will not create big files (as the data files tend to be bigger, than the delete files, and when a data file is rolled, the delete file rolls automatically) - so the behaviour change will affect only very-edge cases 
3. Invest in implementing the RollingFileWriter changes

I would vote for the 1st option, and would accept the 2nd if that is the consensus. I do not think that the 3rd worth the investment.

What do you think @stevenzwu, @aokolnychyi?

> Also there is a behavioural change that the previous write rolled deletes

@pvary, I don't think that threshold makes sense in this scenario. It was needed because we kept rows and positions in a set in the old writers. We are migrating to `SortingPositionOnlyDeleteWriter` that keeps track of positions in bitmaps, which would always fit in memory. We also shouldn't be worried about the size on disk as even millions of position deletes would require KBs in Parquet.
Merged to main.
Thanks for the reviews @aokolnychyi, @stevenzwu and @RussellSpitzer!

Fix patch:
diff --git a/flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java b/flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java
index 40e0b5f2a34e..e8a46c5becd7 100644
--- a/flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java
+++ b/flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java
@@ -27,6 +27,7 @@
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.DeleteGranularity;
 import org.apache.iceberg.flink.FlinkSchemaUtil;
 import org.apache.iceberg.flink.RowDataWrapper;
 import org.apache.iceberg.flink.data.RowDataProjection;
@@ -109,7 +110,7 @@ public void write(RowData row) throws IOException {
 
   protected class RowDataDeltaWriter extends BaseEqualityDeltaWriter {
     RowDataDeltaWriter(PartitionKey partition) {
-      super(partition, schema, deleteSchema);
+      super(partition, schema, deleteSchema, DeleteGranularity.FILE);
     }
 
     @Override
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java
index 40e0b5f2a34e..e8a46c5becd7 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java
@@ -27,6 +27,7 @@
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.DeleteGranularity;
 import org.apache.iceberg.flink.FlinkSchemaUtil;
 import org.apache.iceberg.flink.RowDataWrapper;
 import org.apache.iceberg.flink.data.RowDataProjection;
@@ -109,7 +110,7 @@ public void write(RowData row) throws IOException {
 
   protected class RowDataDeltaWriter extends BaseEqualityDeltaWriter {
     RowDataDeltaWriter(PartitionKey partition) {
-      super(partition, schema, deleteSchema);
+      super(partition, schema, deleteSchema, DeleteGranularity.FILE);
     }
 
     @Override


Test patch:
diff --git a/flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java b/flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
index b5c3bcf41734..5e81c279b69b 100644
--- a/flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
+++ b/flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
@@ -18,12 +18,17 @@
  */
 package org.apache.iceberg.flink.sink;
 
+import static org.assertj.core.api.Assumptions.assumeThat;
+
 import java.util.List;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.test.util.MiniClusterWithClientResource;
 import org.apache.flink.types.Row;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetadataColumns;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.SnapshotRef;
 import org.apache.iceberg.TableProperties;
@@ -45,6 +50,7 @@
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
+import org.junit.rules.Timeout;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
@@ -61,6 +67,8 @@ public class TestFlinkIcebergSinkV2 extends TestFlinkIcebergSinkV2Base {
   public final HadoopCatalogResource catalogResource =
       new HadoopCatalogResource(TEMPORARY_FOLDER, TestFixtures.DATABASE, TestFixtures.TABLE);
 
+  @Rule public final Timeout globalTimeout = Timeout.seconds(60);
+
   @Parameterized.Parameters(
       name = "FileFormat = {0}, Parallelism = {1}, Partitioned={2}, WriteDistributionMode ={3}")
   public static Object[][] parameters() {
@@ -233,4 +241,31 @@ public void testUpsertOnDataKey() throws Exception {
   public void testUpsertOnIdDataKey() throws Exception {
     testUpsertOnIdDataKey(SnapshotRef.MAIN_BRANCH);
   }
+
+  @Test
+  public void testDeleteStats() throws Exception {
+    assumeThat(format).isNotEqualTo(FileFormat.AVRO);
+
+    List<List<Row>> elementsPerCheckpoint =
+        ImmutableList.of(
+            // Checkpoint #1
+            ImmutableList.of(row("+I", 1, "aaa"), row("-D", 1, "aaa"), row("+I", 1, "aaa")));
+
+    List<List<Record>> expectedRecords = ImmutableList.of(ImmutableList.of(record(1, "aaa")));
+
+    testChangeLogs(
+        ImmutableList.of("id", "data"),
+        row -> Row.of(row.getField(ROW_ID_POS), row.getField(ROW_DATA_POS)),
+        false,
+        elementsPerCheckpoint,
+        expectedRecords,
+        "main");
+
+    DeleteFile deleteFile = table.currentSnapshot().addedDeleteFiles(table.io()).iterator().next();
+    String fromStat =
+        new String(
+            deleteFile.lowerBounds().get(MetadataColumns.DELETE_FILE_PATH.fieldId()).array());
+    DataFile dataFile = table.currentSnapshot().addedDataFiles(table.io()).iterator().next();
+    assumeThat(fromStat).isEqualTo(dataFile.path().toString());
+  }
 }
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
index b5c3bcf41734..5e81c279b69b 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
@@ -18,12 +18,17 @@
  */
 package org.apache.iceberg.flink.sink;
 
+import static org.assertj.core.api.Assumptions.assumeThat;
+
 import java.util.List;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.test.util.MiniClusterWithClientResource;
 import org.apache.flink.types.Row;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetadataColumns;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.SnapshotRef;
 import org.apache.iceberg.TableProperties;
@@ -45,6 +50,7 @@
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
+import org.junit.rules.Timeout;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
@@ -61,6 +67,8 @@ public class TestFlinkIcebergSinkV2 extends TestFlinkIcebergSinkV2Base {
   public final HadoopCatalogResource catalogResource =
       new HadoopCatalogResource(TEMPORARY_FOLDER, TestFixtures.DATABASE, TestFixtures.TABLE);
 
+  @Rule public final Timeout globalTimeout = Timeout.seconds(60);
+
   @Parameterized.Parameters(
       name = "FileFormat = {0}, Parallelism = {1}, Partitioned={2}, WriteDistributionMode ={3}")
   public static Object[][] parameters() {
@@ -233,4 +241,31 @@ public void testUpsertOnDataKey() throws Exception {
   public void testUpsertOnIdDataKey() throws Exception {
     testUpsertOnIdDataKey(SnapshotRef.MAIN_BRANCH);
   }
+
+  @Test
+  public void testDeleteStats() throws Exception {
+    assumeThat(format).isNotEqualTo(FileFormat.AVRO);
+
+    List<List<Row>> elementsPerCheckpoint =
+        ImmutableList.of(
+            // Checkpoint #1
+            ImmutableList.of(row("+I", 1, "aaa"), row("-D", 1, "aaa"), row("+I", 1, "aaa")));
+
+    List<List<Record>> expectedRecords = ImmutableList.of(ImmutableList.of(record(1, "aaa")));
+
+    testChangeLogs(
+        ImmutableList.of("id", "data"),
+        row -> Row.of(row.getField(ROW_ID_POS), row.getField(ROW_DATA_POS)),
+        false,
+        elementsPerCheckpoint,
+        expectedRecords,
+        "main");
+
+    DeleteFile deleteFile = table.currentSnapshot().addedDeleteFiles(table.io()).iterator().next();
+    String fromStat =
+        new String(
+            deleteFile.lowerBounds().get(MetadataColumns.DELETE_FILE_PATH.fieldId()).array());
+    DataFile dataFile = table.currentSnapshot().addedDataFiles(table.io()).iterator().next();
+    assumeThat(fromStat).isEqualTo(dataFile.path().toString());
+  }
 }