Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

stevenzwu commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1585251498 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

stevenzwu commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1585251498 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

stevenzwu commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1585251498 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1584738269 ## core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java: ## @@ -109,18 +112,34 @@ protected abstract class BaseEqualityDeltaWriter implements Closeable {

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1584405398 ## flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java: ## @@ -61,6 +67,8 @@ public class TestFlinkIcebergSinkV2 extends

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1584407163 ## flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java: ## @@ -61,6 +67,8 @@ public class TestFlinkIcebergSinkV2 extends

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1584405398 ## flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java: ## @@ -61,6 +67,8 @@ public class TestFlinkIcebergSinkV2 extends

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

aokolnychyi commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1583671212 ## docs/docs/flink-configuration.md: ## @@ -124,8 +124,9 @@ env.getConfig() | max-planning-snapshot-count | connector.iceberg.max-planning-snapshot-count |

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

aokolnychyi commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1583669332 ## core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java: ## @@ -109,18 +112,34 @@ protected abstract class BaseEqualityDeltaWriter implements Closeable

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

stevenzwu commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1583309420 ## core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java: ## @@ -109,18 +112,34 @@ protected abstract class BaseEqualityDeltaWriter implements Closeable {

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

RussellSpitzer commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1583299839 ## core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java: ## @@ -109,18 +112,34 @@ protected abstract class BaseEqualityDeltaWriter implements

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

aokolnychyi commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1581637078 ## core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java: ## @@ -109,18 +112,34 @@ protected abstract class BaseEqualityDeltaWriter implements Closeable

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

aokolnychyi commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1581637078 ## core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java: ## @@ -109,18 +112,34 @@ protected abstract class BaseEqualityDeltaWriter implements Closeable

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1580859044 ## core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java: ## @@ -109,18 +112,34 @@ protected abstract class BaseEqualityDeltaWriter implements Closeable {

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1580858492 ## core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java: ## @@ -140,12 +159,18 @@ public void write(T row) throws IOException { PathOffset previous =

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

aokolnychyi commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1580294047 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java: ## @@ -56,7 +58,8 @@ abstract class BaseDeltaTaskWriter extends

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

aokolnychyi commented on PR #10200: URL: https://github.com/apache/iceberg/pull/10200#issuecomment-2078403034 > 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

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

aokolnychyi commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1580291026 ## core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java: ## @@ -109,18 +112,34 @@ protected abstract class BaseEqualityDeltaWriter implements Closeable

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on PR #10200: URL: https://github.com/apache/iceberg/pull/10200#issuecomment-2076931744 > 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

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1579259488 ## core/src/main/java/org/apache/iceberg/io/SortedPosDeleteWriter.java: ## @@ -174,20 +208,21 @@ private void flushDeletes() { } catch (IOException e) {

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1579255478 ## core/src/main/java/org/apache/iceberg/deletes/SortingPositionOnlyDeleteWriter.java: ## @@ -118,6 +118,10 @@ private DeleteWriteResult writeFileDeletes() throws

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1579208675 ## core/src/main/java/org/apache/iceberg/io/SortedPosDeleteWriter.java: ## @@ -59,20 +60,58 @@ class SortedPosDeleteWriter implements FileWriter, DeleteWr

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1579204149 ## core/src/main/java/org/apache/iceberg/io/SortedPosDeleteWriter.java: ## @@ -174,20 +208,21 @@ private void flushDeletes() { } catch (IOException e) {

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1579107619 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws IOException

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

aokolnychyi commented on PR #10200: URL: https://github.com/apache/iceberg/pull/10200#issuecomment-2076189673 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

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

stevenzwu commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1578676203 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1577755913 ## core/src/main/java/org/apache/iceberg/deletes/ContinuousFileScopedPositionDeleteWriter.java: ## @@ -0,0 +1,92 @@ +/* + * Licensed to the Apache Software Foundation

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1577753275 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java: ## @@ -56,7 +58,8 @@ abstract class BaseDeltaTaskWriter extends

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1577749819 ## core/src/main/java/org/apache/iceberg/deletes/ContinuousFileScopedPositionDeleteWriter.java: ## @@ -0,0 +1,92 @@ +/* + * Licensed to the Apache Software Foundation

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

aokolnychyi commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1577030981 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java: ## @@ -56,7 +58,8 @@ abstract class BaseDeltaTaskWriter extends

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

aokolnychyi commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1577029896 ## core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java: ## @@ -109,18 +112,30 @@ protected abstract class BaseEqualityDeltaWriter implements Closeable

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

aokolnychyi commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1577030203 ## core/src/main/java/org/apache/iceberg/deletes/ContinuousFileScopedPositionDeleteWriter.java: ## @@ -0,0 +1,92 @@ +/* + * Licensed to the Apache Software

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

aokolnychyi commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1577029896 ## core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java: ## @@ -109,18 +112,30 @@ protected abstract class BaseEqualityDeltaWriter implements Closeable

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

aokolnychyi commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1577028939 ## core/src/main/java/org/apache/iceberg/deletes/ContinuousFileScopedPositionDeleteWriter.java: ## @@ -0,0 +1,92 @@ +/* + * Licensed to the Apache Software

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

chenjunjiedada commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1576141345 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java: ## @@ -167,6 +168,18 @@ public DistributionMode distributionMode() {

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575985174 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws IOException

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575982248 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws IOException

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575983673 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/RowDataRewriter.java: ## @@ -84,7 +85,8 @@ public RowDataRewriter( format,

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575706611 ## core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java: ## @@ -140,7 +155,9 @@ public void write(T row) throws IOException { PathOffset previous =

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

chenjunjiedada commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575698957 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575700012 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws IOException

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575698414 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws IOException

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575676660 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/RowDataRewriter.java: ## @@ -84,7 +85,8 @@ public RowDataRewriter( format,

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575668667 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws IOException

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575666452 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws IOException

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

pvary commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575666321 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws IOException

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

stevenzwu commented on PR #10200: URL: https://github.com/apache/iceberg/pull/10200#issuecomment-2070951618 @chenjunjiedada you might be interested. This is to leverage the targeted position delete file that @aokolnychyi implemented from PR #9384 -- This is an automated message from the

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

stevenzwu commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575167955 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

stevenzwu commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575160742 ## core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java: ## @@ -140,7 +155,9 @@ public void write(T row) throws IOException { PathOffset previous =

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

stevenzwu commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575145666 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

stevenzwu commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575094495 ## flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/RowDataRewriter.java: ## @@ -84,7 +85,8 @@ public RowDataRewriter( format,

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

stevenzwu commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575089122 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

stevenzwu commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575081217 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws

Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]

stevenzwu commented on code in PR #10200: URL: https://github.com/apache/iceberg/pull/10200#discussion_r1575081217 ## data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java: ## @@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws