You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "pvary (via GitHub)" <gi...@apache.org> on 2024/04/22 09:13:09 UTC
[PR] Flink: Apply DeleteGranularity for writes [iceberg]
pvary opened a new pull request, #10200:
URL: https://github.com/apache/iceberg/pull/10200
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.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
+ deltaWriter.write(record);
+ if (i % 5 == 10) {
Review Comment:
Good catch!
Actually found an issue I was expecting, so I needed a new delete writer
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "chenjunjiedada (via GitHub)" <gi...@apache.org>.
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 IOException {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
+ deltaWriter.write(record);
+ if (i % 5 == 10) {
+ deltaWriter.delete(record);
+ } else {
+ expected.put(i, record);
+ }
+ }
+
+ // Add some deletes in the end
+ for (int i = 0; i < 199; ++i) {
+ int id = i * 10 + 1;
+ Record record = createRecord(id, "aaa" + id);
+ deltaWriter.delete(record);
+ expected.remove(id);
+ }
+
+ WriteResult result = deltaWriter.complete();
+ assertThat(result.dataFiles()).as("Should have 2 data files.").hasSize(2);
Review Comment:
I guess encoding and compression are working. But it might be more data files if we use other file formats, such as Avro.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
+ deltaWriter.write(record);
+ if (i % 5 == 10) {
+ deltaWriter.delete(record);
+ } else {
+ expected.put(i, record);
+ }
+ }
+
+ // Add some deletes in the end
+ for (int i = 0; i < 199; ++i) {
+ int id = i * 10 + 1;
+ Record record = createRecord(id, "aaa" + id);
+ deltaWriter.delete(record);
+ expected.remove(id);
+ }
+
+ WriteResult result = deltaWriter.complete();
+ assertThat(result.dataFiles()).as("Should have 2 data files.").hasSize(2);
+ assertThat(result.deleteFiles())
Review Comment:
We do not write equality delete in the tests. We use only `deltaWriter.delete(record)` to write the 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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
Review Comment:
Added the 2 iteration of deletes, to prevent the monotonically increasing positions.
After based on your comment, I have fixed the test, and found the exact issue I was expecting first, and created the writer for handling it
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
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 IOException {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
+ deltaWriter.write(record);
+ if (i % 5 == 10) {
Review Comment:
this seems wrong. it is always false
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
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 rows and positions in a set in the old writers. We are migrating to `SortingPositionOnlyDeleteWriter` that keeps track of positions in bitmaps. Even millions of position deletes would take KBs on disk.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 TestFlinkIcebergSinkV2Base {
public final HadoopCatalogResource catalogResource =
new HadoopCatalogResource(TEMPORARY_FOLDER, TestFixtures.DATABASE, TestFixtures.TABLE);
+ @Rule public final Timeout globalTimeout = Timeout.seconds(10);
Review Comment:
Added timeout, because these tests are also prone to running indefinitely
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
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 BaseTaskWriter<RowData> {
Schema schema,
RowType flinkSchema,
List<Integer> equalityFieldIds,
- boolean upsert) {
+ boolean upsert,
+ DeleteGranularity deleteGranularity) {
Review Comment:
Are we worried about the increased metadata size? If so, I don't think it should be a problem as position delete files require much less metadata and highly doubt the difference will be noticed.
##########
flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java:
##########
@@ -56,7 +58,8 @@ abstract class BaseDeltaTaskWriter extends BaseTaskWriter<RowData> {
Schema schema,
RowType flinkSchema,
List<Integer> equalityFieldIds,
- boolean upsert) {
+ boolean upsert,
+ DeleteGranularity deleteGranularity) {
Review Comment:
Are we worried about the increased metadata size? If so, I don't think it should be a problem as position delete files require much less metadata and I highly doubt the difference will be noticed.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
+ deltaWriter.write(record);
+ if (i % 5 == 10) {
+ deltaWriter.delete(record);
+ } else {
+ expected.put(i, record);
+ }
+ }
+
+ // Add some deletes in the end
+ for (int i = 0; i < 199; ++i) {
+ int id = i * 10 + 1;
+ Record record = createRecord(id, "aaa" + id);
+ deltaWriter.delete(record);
+ expected.remove(id);
+ }
+
+ WriteResult result = deltaWriter.complete();
+ assertThat(result.dataFiles()).as("Should have 2 data files.").hasSize(2);
Review Comment:
For the record, this is based on `BaseRollingWriter.ROWS_DIVISOR` which is not configurable.
Added the comment describing this.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
+ deltaWriter.write(record);
+ if (i % 5 == 10) {
+ deltaWriter.delete(record);
+ } else {
+ expected.put(i, record);
+ }
+ }
+
+ // Add some deletes in the end
+ for (int i = 0; i < 199; ++i) {
+ int id = i * 10 + 1;
+ Record record = createRecord(id, "aaa" + id);
+ deltaWriter.delete(record);
+ expected.remove(id);
+ }
+
+ WriteResult result = deltaWriter.complete();
+ assertThat(result.dataFiles()).as("Should have 2 data files.").hasSize(2);
Review Comment:
The size is only checked every 1000 records - so 2000 records, max 2 data 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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
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 `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.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #10200:
URL: https://github.com/apache/iceberg/pull/10200#discussion_r1580507541
##########
core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java:
##########
@@ -109,18 +112,34 @@ protected abstract class BaseEqualityDeltaWriter implements Closeable {
private final StructProjection structProjection;
private RollingFileWriter dataWriter;
private RollingEqDeleteWriter eqDeleteWriter;
- private SortedPosDeleteWriter<T> posDeleteWriter;
+ private FileWriter<PositionDelete<T>, DeleteWriteResult> posDeleteWriter;
private Map<StructLike, PathOffset> insertedRowMap;
protected BaseEqualityDeltaWriter(StructLike partition, Schema schema, Schema deleteSchema) {
+ this(partition, schema, deleteSchema, DeleteGranularity.PARTITION);
+ }
+
+ protected BaseEqualityDeltaWriter(
+ StructLike partition,
+ Schema schema,
+ Schema deleteSchema,
+ DeleteGranularity deleteGranularity) {
Review Comment:
What is the current experience is Spark with this opinion? Are we planning to move the Spark deletes to file granularity?
I would like to hear @stevenzwu's opinion in this regard, but I would be happy to eliminate a configuration possibility 😄
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
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 Closeable {
private final StructProjection structProjection;
private RollingFileWriter dataWriter;
private RollingEqDeleteWriter eqDeleteWriter;
- private SortedPosDeleteWriter<T> posDeleteWriter;
+ private FileWriter<PositionDelete<T>, DeleteWriteResult> posDeleteWriter;
private Map<StructLike, PathOffset> insertedRowMap;
protected BaseEqualityDeltaWriter(StructLike partition, Schema schema, Schema deleteSchema) {
+ this(partition, schema, deleteSchema, DeleteGranularity.PARTITION);
+ }
+
+ protected BaseEqualityDeltaWriter(
+ StructLike partition,
+ Schema schema,
+ Schema deleteSchema,
+ DeleteGranularity deleteGranularity) {
Review Comment:
We discussed a bit on slack. I think no-conf is good, in my mind if there was a benefit from having the two modes as an option it's something flink should be choosing at runtime rather than a user defined config. I think this is also mostly a perf difference so I wouldn't mind changing the behavior from the previous behavior in those edge cases with enough data to write multiple files in the same partition in one batch.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
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 IOException {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
Review Comment:
this still doesn't result in position deletes.
first round - `i % 5 == 0`
second round - `i * 10 + 1`
there is no overlap. both produces equality deletes? position deletes is only generated when the key is deleted multiple times in the same write session.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
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 IOException {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
Review Comment:
this still doesn't result in position deletes.
first round - `i % 5 == 0`
second round - `i * 10 + 1`
there is no overlap. both produces equality deletes? position deletes is only generated when the key is deleted multiple times in the same write session.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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,
table.properties(),
null,
- false);
+ false,
+ DeleteGranularity.FILE);
Review Comment:
Also the goal is to used `DeleteGranularity.FILE` in Flink as a general rule
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 {
private final StructProjection structProjection;
private RollingFileWriter dataWriter;
private RollingEqDeleteWriter eqDeleteWriter;
- private SortedPosDeleteWriter<T> posDeleteWriter;
+ private FileWriter<PositionDelete<T>, DeleteWriteResult> posDeleteWriter;
private Map<StructLike, PathOffset> insertedRowMap;
protected BaseEqualityDeltaWriter(StructLike partition, Schema schema, Schema deleteSchema) {
+ this(partition, schema, deleteSchema, DeleteGranularity.PARTITION);
+ }
+
+ protected BaseEqualityDeltaWriter(
+ StructLike partition,
+ Schema schema,
+ Schema deleteSchema,
+ DeleteGranularity deleteGranularity) {
Preconditions.checkNotNull(schema, "Iceberg table schema cannot be null.");
Preconditions.checkNotNull(deleteSchema, "Equality-delete schema cannot be null.");
this.structProjection = StructProjection.create(schema, deleteSchema);
this.dataWriter = new RollingFileWriter(partition);
this.eqDeleteWriter = new RollingEqDeleteWriter(partition);
this.posDeleteWriter =
- new SortedPosDeleteWriter<>(appenderFactory, fileFactory, format, partition);
+ new SortingPositionOnlyDeleteWriter<>(
+ () ->
+ appenderFactory.newPosDeleteWriter(
+ partition == null
Review Comment:
Done.
Thanks for the catch!
I just refactored the previous, but your suggestion makes sense
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary merged PR #10200:
URL: https://github.com/apache/iceberg/pull/10200
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
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 = insertedRowMap.put(copiedKey, pathOffset);
if (previous != null) {
// TODO attach the previous row if has a positional-delete row schema in appender factory.
- posDeleteWriter.delete(previous.path, previous.rowOffset, null);
+ PositionDelete<T> delete = PositionDelete.create();
Review Comment:
nit: extract these 3 lines to a method?
not directly related to this PR. wondering if the TODO comment is still applicable?
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 `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?
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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<T> implements FileWriter<PositionDelete<T>, DeleteWr
OutputFileFactory fileFactory,
FileFormat format,
StructLike partition,
- long recordsNumThreshold) {
+ long recordsNumThreshold,
+ DeleteGranularity deleteGranularity) {
this.appenderFactory = appenderFactory;
this.fileFactory = fileFactory;
this.format = format;
this.partition = partition;
this.recordsNumThreshold = recordsNumThreshold;
+ this.deleteGranularity = deleteGranularity;
+ }
+
+ SortedPosDeleteWriter(
Review Comment:
These constructors are not utilised by our code (or only by tests), but kept them for compatibility reasons.
Do we need them? Are they part of the API?
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
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 Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.deletes;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.io.DeleteWriteResult;
+import org.apache.iceberg.io.FileWriter;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.CharSequenceSet;
+
+/**
+ * A position delete writer that produces a separate delete file for each referenced data file.
+ *
+ * <p>This writer does not keep track of seen deletes and assumes all incoming records are ordered
Review Comment:
Does it actually assume the records are ordered by file and position? It seems to keep multiple file handles open until the end but assumes records are ordered by position, meaning it is in between clustered and fan-out writers.
Also, why isn't enough to use `ClusteredPositionDeleteWriter` with file granularity?
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
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 {
private final StructProjection structProjection;
private RollingFileWriter dataWriter;
private RollingEqDeleteWriter eqDeleteWriter;
- private SortedPosDeleteWriter<T> posDeleteWriter;
+ private FileWriter<PositionDelete<T>, DeleteWriteResult> posDeleteWriter;
private Map<StructLike, PathOffset> insertedRowMap;
protected BaseEqualityDeltaWriter(StructLike partition, Schema schema, Schema deleteSchema) {
+ this(partition, schema, deleteSchema, DeleteGranularity.PARTITION);
+ }
+
+ protected BaseEqualityDeltaWriter(
+ StructLike partition,
+ Schema schema,
+ Schema deleteSchema,
+ DeleteGranularity deleteGranularity) {
Preconditions.checkNotNull(schema, "Iceberg table schema cannot be null.");
Preconditions.checkNotNull(deleteSchema, "Equality-delete schema cannot be null.");
this.structProjection = StructProjection.create(schema, deleteSchema);
this.dataWriter = new RollingFileWriter(partition);
this.eqDeleteWriter = new RollingEqDeleteWriter(partition);
this.posDeleteWriter =
- new SortedPosDeleteWriter<>(appenderFactory, fileFactory, format, partition);
+ deleteGranularity.equals(DeleteGranularity.FILE)
Review Comment:
Is there a use case where we want to use partition granularity? Based on our initial discussions, we wanted to always use file granularity in Flink.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #10200:
URL: https://github.com/apache/iceberg/pull/10200#discussion_r1577744934
##########
flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java:
##########
@@ -167,6 +168,18 @@ public DistributionMode distributionMode() {
return DistributionMode.fromName(modeName);
}
+ public DeleteGranularity deleteGranularity() {
+ String modeName =
+ confParser
+ .stringConf()
+ .option(FlinkWriteOptions.DELETE_GRANULARITY.key())
+ .flinkConfig(FlinkWriteOptions.DELETE_GRANULARITY)
+ .tableProperty(TableProperties.DELETE_GRANULARITY)
+ .defaultValue(TableProperties.DELETE_GRANULARITY_DEFAULT)
Review Comment:
I have chickened out from this, as this is something that might affect some users.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 BaseTaskWriter<RowData> {
Schema schema,
RowType flinkSchema,
List<Integer> equalityFieldIds,
- boolean upsert) {
+ boolean upsert,
+ DeleteGranularity deleteGranularity) {
Review Comment:
I think that in most cases we want to use FILE granularity, but we found a case when Flink might ending up writing multiple delete files with FILE granularity - when there are big writes, and the data writer rolls. In this case we might have to create multiple delete files for a single commit. This might be undesirable for some users, so I have kept the option for 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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 IOException {
@SuppressWarnings("CollectionUndefinedEquality")
private DeleteWriteResult writeDeletes(Collection<CharSequence> paths) throws IOException {
+ if (paths.size() == 0) {
Review Comment:
Needed to add this, so we do not create an empty delete file
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
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 IOException {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
+ deltaWriter.write(record);
+ if (i % 5 == 10) {
+ deltaWriter.delete(record);
+ } else {
+ expected.put(i, record);
+ }
+ }
+
+ // Add some deletes in the end
+ for (int i = 0; i < 199; ++i) {
+ int id = i * 10 + 1;
+ Record record = createRecord(id, "aaa" + id);
+ deltaWriter.delete(record);
+ expected.remove(id);
+ }
+
+ WriteResult result = deltaWriter.complete();
+ assertThat(result.dataFiles()).as("Should have 2 data files.").hasSize(2);
+ assertThat(result.deleteFiles())
+ .as("Should have correct number of pos-delete files")
+ .hasSize(granularity.equals(DeleteGranularity.FILE) ? 2 : 1);
Review Comment:
should we also add assertion on the number of rows in the position 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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 = insertedRowMap.put(copiedKey, pathOffset);
if (previous != null) {
// TODO attach the previous row if has a positional-delete row schema in appender factory.
- posDeleteWriter.delete(previous.path, previous.rowOffset, null);
+ writePosDelete(previous);
}
dataWriter.write(row);
}
+ private void writePosDelete(PathOffset pathOffset) {
+ PositionDelete<T> delete = PositionDelete.create();
Review Comment:
Done
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 TestFlinkIcebergSinkV2Base {
public final HadoopCatalogResource catalogResource =
new HadoopCatalogResource(TEMPORARY_FOLDER, TestFixtures.DATABASE, TestFixtures.TABLE);
+ @Rule public final Timeout globalTimeout = Timeout.seconds(10);
Review Comment:
Added timeout, because these tests are also prone to running indefinitely
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
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 {
private final StructProjection structProjection;
private RollingFileWriter dataWriter;
private RollingEqDeleteWriter eqDeleteWriter;
- private SortedPosDeleteWriter<T> posDeleteWriter;
+ private FileWriter<PositionDelete<T>, DeleteWriteResult> posDeleteWriter;
private Map<StructLike, PathOffset> insertedRowMap;
protected BaseEqualityDeltaWriter(StructLike partition, Schema schema, Schema deleteSchema) {
+ this(partition, schema, deleteSchema, DeleteGranularity.PARTITION);
+ }
+
+ protected BaseEqualityDeltaWriter(
+ StructLike partition,
+ Schema schema,
+ Schema deleteSchema,
+ DeleteGranularity deleteGranularity) {
Review Comment:
Seems like everyone is in favor of not exposing the config then.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
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 IOException {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
Review Comment:
this still doesn't result in position deletes.
first round - `i % 5 == 0`
second round - `i * 10 + 1`
there is no overlap. both produces equality 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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
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 {
private final StructProjection structProjection;
private RollingFileWriter dataWriter;
private RollingEqDeleteWriter eqDeleteWriter;
- private SortedPosDeleteWriter<T> posDeleteWriter;
+ private FileWriter<PositionDelete<T>, DeleteWriteResult> posDeleteWriter;
private Map<StructLike, PathOffset> insertedRowMap;
protected BaseEqualityDeltaWriter(StructLike partition, Schema schema, Schema deleteSchema) {
+ this(partition, schema, deleteSchema, DeleteGranularity.PARTITION);
+ }
+
+ protected BaseEqualityDeltaWriter(
+ StructLike partition,
+ Schema schema,
+ Schema deleteSchema,
+ DeleteGranularity deleteGranularity) {
Review Comment:
I am also in favor of no config. It is a "good" internal impl/perf change. In the data file rollover case, one smaller delete file every data file is better than one larger delete file for multiple data 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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
+ deltaWriter.write(record);
+ if (i % 5 == 10) {
+ deltaWriter.delete(record);
+ } else {
+ expected.put(i, record);
+ }
+ }
+
+ // Add some deletes in the end
+ for (int i = 0; i < 199; ++i) {
+ int id = i * 10 + 1;
+ Record record = createRecord(id, "aaa" + id);
+ deltaWriter.delete(record);
+ expected.remove(id);
+ }
+
+ WriteResult result = deltaWriter.complete();
+ assertThat(result.dataFiles()).as("Should have 2 data files.").hasSize(2);
+ assertThat(result.deleteFiles())
+ .as("Should have correct number of pos-delete files")
+ .hasSize(granularity.equals(DeleteGranularity.FILE) ? 2 : 1);
Review Comment:
Added an assertion
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
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 {
private final StructProjection structProjection;
private RollingFileWriter dataWriter;
private RollingEqDeleteWriter eqDeleteWriter;
- private SortedPosDeleteWriter<T> posDeleteWriter;
+ private FileWriter<PositionDelete<T>, DeleteWriteResult> posDeleteWriter;
private Map<StructLike, PathOffset> insertedRowMap;
protected BaseEqualityDeltaWriter(StructLike partition, Schema schema, Schema deleteSchema) {
+ this(partition, schema, deleteSchema, DeleteGranularity.PARTITION);
+ }
+
+ protected BaseEqualityDeltaWriter(
+ StructLike partition,
+ Schema schema,
+ Schema deleteSchema,
+ DeleteGranularity deleteGranularity) {
Preconditions.checkNotNull(schema, "Iceberg table schema cannot be null.");
Preconditions.checkNotNull(deleteSchema, "Equality-delete schema cannot be null.");
this.structProjection = StructProjection.create(schema, deleteSchema);
this.dataWriter = new RollingFileWriter(partition);
this.eqDeleteWriter = new RollingEqDeleteWriter(partition);
this.posDeleteWriter =
- new SortedPosDeleteWriter<>(appenderFactory, fileFactory, format, partition);
+ deleteGranularity.equals(DeleteGranularity.FILE)
Review Comment:
Is there a use case where we want to use partition granularity? Based on our initial discussions, we wanted to always use file granularity in Flink.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
+ deltaWriter.write(record);
+ if (i % 5 == 10) {
+ deltaWriter.delete(record);
+ } else {
+ expected.put(i, record);
+ }
+ }
+
+ // Add some deletes in the end
+ for (int i = 0; i < 199; ++i) {
+ int id = i * 10 + 1;
+ Record record = createRecord(id, "aaa" + id);
+ deltaWriter.delete(record);
+ expected.remove(id);
+ }
+
+ WriteResult result = deltaWriter.complete();
+ assertThat(result.dataFiles()).as("Should have 2 data files.").hasSize(2);
Review Comment:
BaseRollingWriter checks the size on every 1000 rows (ROWS_DIVISOR)
Added a comment about this.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.deletes;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.io.DeleteWriteResult;
+import org.apache.iceberg.io.FileWriter;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.CharSequenceSet;
+
+/**
+ * A position delete writer that produces a separate delete file for each referenced data file.
+ *
+ * <p>This writer does not keep track of seen deletes and assumes all incoming records are ordered
Review Comment:
Also, I was not brave enough to change any of the existing generic writer classes, as I can not be sure how they are used.
Maybe if we enhance the `SortedPosDeleteWriter` to handle the granularity, then we would not need an extra class. For this decision, I would like to have some oversight from someone who is more familiar with the writer api design and use-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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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) {
setFailure(e);
throw new UncheckedIOException(
- "Failed to write the sorted path/pos pairs to pos-delete file: "
- + outputFile.encryptingOutputFile().location(),
Review Comment:
Adding the file location would require some confusing code, like:
- Adding a local attribute to the class `lastOutputFile`
- Updating this attribute in `createWriter`
- We need to assume that if there is an error in `FileScopedPositionDeleteWriter` it is always with the last opened writer - which is true, but I am not sure we want to build on that
I think this is more confusing than what it worth to have the filename in the exception message, but I am open to discussion
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
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 IOException {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
+ deltaWriter.write(record);
+ if (i % 5 == 10) {
+ deltaWriter.delete(record);
+ } else {
+ expected.put(i, record);
+ }
+ }
+
+ // Add some deletes in the end
+ for (int i = 0; i < 199; ++i) {
+ int id = i * 10 + 1;
+ Record record = createRecord(id, "aaa" + id);
+ deltaWriter.delete(record);
+ expected.remove(id);
+ }
+
+ WriteResult result = deltaWriter.complete();
+ assertThat(result.dataFiles()).as("Should have 2 data files.").hasSize(2);
Review Comment:
> The size is only checked every 1000 records - so 2000 records, max 2 data files
I see. that's the default Parquet writer config. it is not obvious when reading the code. might add a comment or set the parquet size check config explicitly in the code?
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 {
private final StructProjection structProjection;
private RollingFileWriter dataWriter;
private RollingEqDeleteWriter eqDeleteWriter;
- private SortedPosDeleteWriter<T> posDeleteWriter;
+ private FileWriter<PositionDelete<T>, DeleteWriteResult> posDeleteWriter;
private Map<StructLike, PathOffset> insertedRowMap;
protected BaseEqualityDeltaWriter(StructLike partition, Schema schema, Schema deleteSchema) {
+ this(partition, schema, deleteSchema, DeleteGranularity.PARTITION);
+ }
+
+ protected BaseEqualityDeltaWriter(
+ StructLike partition,
+ Schema schema,
+ Schema deleteSchema,
+ DeleteGranularity deleteGranularity) {
Review Comment:
I love this 😄
The PR is sooo much smaller!
@stevenzwu: one final round of review?
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
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 | N/A | Integer.MAX_VALUE | Max number of snapshots limited per split enumeration. Applicable only to streaming read.
|
| limit | connector.iceberg.limit | N/A | -1 | Limited output number of rows.
|
| max-allowed-planning-failures | connector.iceberg.max-allowed-planning-failures | N/A | 3 | Max allowed consecutive failures for scan planning before failing the job. Set to -1 for never failing the job for scan planing failure.
|
-| watermark-column | connector.iceberg.watermark-column | N/A | null | Specifies the watermark column to use for watermark generation. If this option is present, the `splitAssignerFactory` will be overridden with `OrderedSplitAssignerFactory`.
|
-| watermark-column-time-unit | connector.iceberg.watermark-column-time-unit | N/A | TimeUnit.MICROSECONDS | Specifies the watermark time unit to use for watermark generation. The possible values are DAYS, HOURS, MINUTES, SECONDS, MILLISECONDS, MICROSECONDS, NANOSECONDS.
|
+| watermark-column | connector.iceberg.watermark-column | N/A | null | Specifies the watermark column to use for watermark generation. If this option is present, the `splitAssignerFactory` will be overridden with `OrderedSplitAssignerFactory`.
|
+| watermark-column-time-unit | connector.iceberg.watermark-column-time-unit | N/A | TimeUnit.MICROSECONDS | Specifies the watermark time unit to use for watermark generation. The possible values are DAYS, HOURS, MINUTES, SECONDS, MILLISECONDS, MICROSECONDS, NANOSECONDS.
|
+| delete-granularity | connector.iceberg.delete-granularity | write.delete.granularity | PARTITION | Specifies the delete file granularity when writing positional deletes. The possible values are PARTITION, FILE.
|
Review Comment:
This file should be reverted if we drop the config.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #10200:
URL: https://github.com/apache/iceberg/pull/10200#issuecomment-2087256346
Merged to main.
Thanks for the reviews @aokolnychyi, @stevenzwu and @RussellSpitzer!
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 TestFlinkIcebergSinkV2Base {
public final HadoopCatalogResource catalogResource =
new HadoopCatalogResource(TEMPORARY_FOLDER, TestFixtures.DATABASE, TestFixtures.TABLE);
+ @Rule public final Timeout globalTimeout = Timeout.seconds(60);
Review Comment:
Added timeout, as these tests are also prone to run indefinitely
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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,
table.properties(),
null,
- false);
+ false,
+ DeleteGranularity.FILE);
Review Comment:
We do not write delete files, so it has no impact.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
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 IOException {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
+ deltaWriter.write(record);
+ if (i % 5 == 10) {
+ deltaWriter.delete(record);
+ } else {
+ expected.put(i, record);
+ }
+ }
+
+ // Add some deletes in the end
+ for (int i = 0; i < 199; ++i) {
+ int id = i * 10 + 1;
+ Record record = createRecord(id, "aaa" + id);
+ deltaWriter.delete(record);
+ expected.remove(id);
+ }
+
+ WriteResult result = deltaWriter.complete();
+ assertThat(result.dataFiles()).as("Should have 2 data files.").hasSize(2);
+ assertThat(result.deleteFiles())
+ .as("Should have correct number of pos-delete files")
+ .hasSize(granularity.equals(DeleteGranularity.FILE) ? 2 : 1);
+
+ commitTransaction(result);
+ assertThat(actualRowSet("*"))
+ .as("Should have an expected record")
Review Comment:
nit: "Should have expected records"
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
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 IOException {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
Review Comment:
maybe we create some position deletes via `update`. current monotonically increased index is unique record/key per iteration
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
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 IOException {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
+ deltaWriter.write(record);
+ if (i % 5 == 10) {
Review Comment:
this seems wrong
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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) {
setFailure(e);
throw new UncheckedIOException(
- "Failed to write the sorted path/pos pairs to pos-delete file: "
- + outputFile.encryptingOutputFile().location(),
Review Comment:
reverted
##########
core/src/main/java/org/apache/iceberg/io/SortedPosDeleteWriter.java:
##########
@@ -59,20 +60,58 @@ class SortedPosDeleteWriter<T> implements FileWriter<PositionDelete<T>, DeleteWr
OutputFileFactory fileFactory,
FileFormat format,
StructLike partition,
- long recordsNumThreshold) {
+ long recordsNumThreshold,
+ DeleteGranularity deleteGranularity) {
this.appenderFactory = appenderFactory;
this.fileFactory = fileFactory;
this.format = format;
this.partition = partition;
this.recordsNumThreshold = recordsNumThreshold;
+ this.deleteGranularity = deleteGranularity;
+ }
+
+ SortedPosDeleteWriter(
Review Comment:
reverted
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
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 BaseTaskWriter<RowData> {
Schema schema,
RowType flinkSchema,
List<Integer> equalityFieldIds,
- boolean upsert) {
+ boolean upsert,
+ DeleteGranularity deleteGranularity) {
Review Comment:
Is there a use case where we want to use partition granularity? Based on our initial discussions, we wanted to always use file granularity in Flink.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on PR #10200:
URL: https://github.com/apache/iceberg/pull/10200#issuecomment-2073659964
Getting to this PR today/tomorrow.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #10200:
URL: https://github.com/apache/iceberg/pull/10200#discussion_r1577745182
##########
flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java:
##########
@@ -242,6 +243,18 @@ public Builder distributionMode(DistributionMode mode) {
return this;
}
+ /**
+ * Configure the write {@link DeleteGranularity} that the flink sink will use. Currently, flink
+ * support {@link DeleteGranularity#PARTITION} and {@link DeleteGranularity#FILE}.
Review Comment:
Added the document change
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 = insertedRowMap.put(copiedKey, pathOffset);
if (previous != null) {
// TODO attach the previous row if has a positional-delete row schema in appender factory.
- posDeleteWriter.delete(previous.path, previous.rowOffset, null);
+ PositionDelete<T> delete = PositionDelete.create();
Review Comment:
Extracted the method.
I think we don't want to add to store the previous row to save memory. So we can't write it to the positional delete, but I didn't want to change an unrelated 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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "chenjunjiedada (via GitHub)" <gi...@apache.org>.
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() {
return DistributionMode.fromName(modeName);
}
+ public DeleteGranularity deleteGranularity() {
+ String modeName =
+ confParser
+ .stringConf()
+ .option(FlinkWriteOptions.DELETE_GRANULARITY.key())
+ .flinkConfig(FlinkWriteOptions.DELETE_GRANULARITY)
+ .tableProperty(TableProperties.DELETE_GRANULARITY)
+ .defaultValue(TableProperties.DELETE_GRANULARITY_DEFAULT)
Review Comment:
Do we want to change to file granularity right now?
##########
data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java:
##########
@@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws IOException {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
+ deltaWriter.write(record);
+ if (i % 5 == 10) {
+ deltaWriter.delete(record);
+ } else {
+ expected.put(i, record);
+ }
+ }
+
+ // Add some deletes in the end
+ for (int i = 0; i < 199; ++i) {
+ int id = i * 10 + 1;
+ Record record = createRecord(id, "aaa" + id);
+ deltaWriter.delete(record);
+ expected.remove(id);
+ }
+
+ WriteResult result = deltaWriter.complete();
+ assertThat(result.dataFiles()).as("Should have 2 data files.").hasSize(2);
Review Comment:
Just recall this.
##########
flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java:
##########
@@ -242,6 +243,18 @@ public Builder distributionMode(DistributionMode mode) {
return this;
}
+ /**
+ * Configure the write {@link DeleteGranularity} that the flink sink will use. Currently, flink
+ * support {@link DeleteGranularity#PARTITION} and {@link DeleteGranularity#FILE}.
Review Comment:
How about documenting the default granularity that Flink sink uses?
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
+ deltaWriter.write(record);
+ if (i % 5 == 10) {
+ deltaWriter.delete(record);
+ } else {
+ expected.put(i, record);
+ }
+ }
+
+ // Add some deletes in the end
+ for (int i = 0; i < 199; ++i) {
+ int id = i * 10 + 1;
+ Record record = createRecord(id, "aaa" + id);
+ deltaWriter.delete(record);
+ expected.remove(id);
+ }
+
+ WriteResult result = deltaWriter.complete();
+ assertThat(result.dataFiles()).as("Should have 2 data files.").hasSize(2);
+ assertThat(result.deleteFiles())
+ .as("Should have correct number of pos-delete files")
+ .hasSize(granularity.equals(DeleteGranularity.FILE) ? 2 : 1);
+
+ commitTransaction(result);
+ assertThat(actualRowSet("*"))
+ .as("Should have an expected record")
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
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 Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.deletes;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.io.DeleteWriteResult;
+import org.apache.iceberg.io.FileWriter;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.CharSequenceSet;
+
+/**
+ * A position delete writer that produces a separate delete file for each referenced data file.
+ *
+ * <p>This writer does not keep track of seen deletes and assumes all incoming records are ordered
Review Comment:
I will have to take a closer look at `BaseTaskWriter`. It has been a while.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
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 {
private final StructProjection structProjection;
private RollingFileWriter dataWriter;
private RollingEqDeleteWriter eqDeleteWriter;
- private SortedPosDeleteWriter<T> posDeleteWriter;
+ private FileWriter<PositionDelete<T>, DeleteWriteResult> posDeleteWriter;
private Map<StructLike, PathOffset> insertedRowMap;
protected BaseEqualityDeltaWriter(StructLike partition, Schema schema, Schema deleteSchema) {
+ this(partition, schema, deleteSchema, DeleteGranularity.PARTITION);
+ }
+
+ protected BaseEqualityDeltaWriter(
+ StructLike partition,
+ Schema schema,
+ Schema deleteSchema,
+ DeleteGranularity deleteGranularity) {
Preconditions.checkNotNull(schema, "Iceberg table schema cannot be null.");
Preconditions.checkNotNull(deleteSchema, "Equality-delete schema cannot be null.");
this.structProjection = StructProjection.create(schema, deleteSchema);
this.dataWriter = new RollingFileWriter(partition);
this.eqDeleteWriter = new RollingEqDeleteWriter(partition);
this.posDeleteWriter =
- new SortedPosDeleteWriter<>(appenderFactory, fileFactory, format, partition);
+ new SortingPositionOnlyDeleteWriter<>(
+ () ->
+ appenderFactory.newPosDeleteWriter(
+ partition == null
Review Comment:
If I am not mistaken, this line should check `spec.isUnpartitioned() || partition == null` to avoid weird issues we discovered in #7685. We also have to pass `spec` with `partition` below in `newOutputFile`.
I'd also consider moving this logic into a separate method for better readability. Up to you, though.
```
private EncryptedOutputFile newOutputFile(StructLike partition) {
if (spec.isUnpartitioned() || partition == null) {
return fileFactory.newOutputFile();
} else {
return fileFactory.newOutputFile(spec, partition);
}
}
```
##########
core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java:
##########
@@ -140,12 +159,18 @@ public void write(T row) throws IOException {
PathOffset previous = insertedRowMap.put(copiedKey, pathOffset);
if (previous != null) {
// TODO attach the previous row if has a positional-delete row schema in appender factory.
- posDeleteWriter.delete(previous.path, previous.rowOffset, null);
+ writePosDelete(previous);
}
dataWriter.write(row);
}
+ private void writePosDelete(PathOffset pathOffset) {
+ PositionDelete<T> delete = PositionDelete.create();
Review Comment:
We usually avoid creating an extra `PositionDelete` object for every delete, if possible. Instead, we initialize one object and reuse it. Take a look at `BasePositionDeltaWriter`.
##########
core/src/main/java/org/apache/iceberg/deletes/SortingPositionOnlyDeleteWriter.java:
##########
@@ -118,6 +118,10 @@ private DeleteWriteResult writeFileDeletes() throws IOException {
@SuppressWarnings("CollectionUndefinedEquality")
private DeleteWriteResult writeDeletes(Collection<CharSequence> paths) throws IOException {
+ if (paths.size() == 0) {
Review Comment:
Shall we use `paths.isEmpty()`?
##########
flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java:
##########
@@ -56,7 +58,8 @@ abstract class BaseDeltaTaskWriter extends BaseTaskWriter<RowData> {
Schema schema,
RowType flinkSchema,
List<Integer> equalityFieldIds,
- boolean upsert) {
+ boolean upsert,
+ DeleteGranularity deleteGranularity) {
Review Comment:
Are we worried about the increased metadata size?
##########
core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java:
##########
@@ -109,18 +112,34 @@ protected abstract class BaseEqualityDeltaWriter implements Closeable {
private final StructProjection structProjection;
private RollingFileWriter dataWriter;
private RollingEqDeleteWriter eqDeleteWriter;
- private SortedPosDeleteWriter<T> posDeleteWriter;
+ private FileWriter<PositionDelete<T>, DeleteWriteResult> posDeleteWriter;
private Map<StructLike, PathOffset> insertedRowMap;
protected BaseEqualityDeltaWriter(StructLike partition, Schema schema, Schema deleteSchema) {
+ this(partition, schema, deleteSchema, DeleteGranularity.PARTITION);
+ }
+
+ protected BaseEqualityDeltaWriter(
+ StructLike partition,
+ Schema schema,
+ Schema deleteSchema,
+ DeleteGranularity deleteGranularity) {
Review Comment:
I am not convinced we should make it configurable when writing equality deltas. Our original plan was to always generate file-scoped position deletes when writing Flink CDC. Take a look at [this](https://github.com/apache/iceberg/pull/9384#discussion_r1439722314) discussion.
##########
flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java:
##########
@@ -167,6 +168,18 @@ public DistributionMode distributionMode() {
return DistributionMode.fromName(modeName);
}
+ public DeleteGranularity deleteGranularity() {
+ String modeName =
+ confParser
+ .stringConf()
+ .option(FlinkWriteOptions.DELETE_GRANULARITY.key())
+ .flinkConfig(FlinkWriteOptions.DELETE_GRANULARITY)
+ .tableProperty(TableProperties.DELETE_GRANULARITY)
+ .defaultValue(TableProperties.DELETE_GRANULARITY_DEFAULT)
Review Comment:
The way we generate position deletes in Flink CDC causes a lot of issues. I think we should switch now and not even offer the configuration. If we think that everyone who runs Flink CDC would have to set this config, then I don't think this config should exist.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
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 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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
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,
table.properties(),
null,
- false);
+ false,
+ DeleteGranularity.FILE);
Review Comment:
is the default change intentional?
##########
data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java:
##########
@@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws IOException {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
+ deltaWriter.write(record);
+ if (i % 5 == 10) {
+ deltaWriter.delete(record);
+ } else {
+ expected.put(i, record);
+ }
+ }
+
+ // Add some deletes in the end
+ for (int i = 0; i < 199; ++i) {
+ int id = i * 10 + 1;
+ Record record = createRecord(id, "aaa" + id);
+ deltaWriter.delete(record);
+ expected.remove(id);
+ }
+
+ WriteResult result = deltaWriter.complete();
+ assertThat(result.dataFiles()).as("Should have 2 data files.").hasSize(2);
Review Comment:
target file size is 128 bytes. wondering why only 2 data files are created.
##########
data/src/test/java/org/apache/iceberg/io/TestTaskEqualityDeltaWriter.java:
##########
@@ -409,6 +421,55 @@ public void testUpsertDataWithFullRowSchema() throws IOException {
.isEqualTo(ImmutableList.of(posRecord.copy("file_path", dataFile.path(), "pos", 0L)));
}
+ @TestTemplate
+ public void testDeleteFileGranularity() throws IOException {
+ withGranularity(DeleteGranularity.FILE);
+ }
+
+ @TestTemplate
+ public void testDeletePartitionGranularity() throws IOException {
+ withGranularity(DeleteGranularity.PARTITION);
+ }
+
+ private void withGranularity(DeleteGranularity granularity) throws IOException {
+ List<Integer> eqDeleteFieldIds = Lists.newArrayList(idFieldId, dataFieldId);
+ Schema eqDeleteRowSchema = table.schema();
+
+ GenericTaskDeltaWriter deltaWriter =
+ createTaskWriter(eqDeleteFieldIds, eqDeleteRowSchema, granularity);
+
+ Map<Integer, Record> expected = Maps.newHashMapWithExpectedSize(2000);
+ // Create enough records, so we have multiple files
+ for (int i = 0; i < 2000; ++i) {
+ Record record = createRecord(i, "aaa" + i);
+ deltaWriter.write(record);
+ if (i % 5 == 10) {
+ deltaWriter.delete(record);
+ } else {
+ expected.put(i, record);
+ }
+ }
+
+ // Add some deletes in the end
+ for (int i = 0; i < 199; ++i) {
+ int id = i * 10 + 1;
+ Record record = createRecord(id, "aaa" + id);
+ deltaWriter.delete(record);
+ expected.remove(id);
+ }
+
+ WriteResult result = deltaWriter.complete();
+ assertThat(result.dataFiles()).as("Should have 2 data files.").hasSize(2);
+ assertThat(result.deleteFiles())
Review Comment:
are `deleteFiles()` only return position deletes? what about equality 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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
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 {
private final StructProjection structProjection;
private RollingFileWriter dataWriter;
private RollingEqDeleteWriter eqDeleteWriter;
- private SortedPosDeleteWriter<T> posDeleteWriter;
+ private FileWriter<PositionDelete<T>, DeleteWriteResult> posDeleteWriter;
private Map<StructLike, PathOffset> insertedRowMap;
protected BaseEqualityDeltaWriter(StructLike partition, Schema schema, Schema deleteSchema) {
+ this(partition, schema, deleteSchema, DeleteGranularity.PARTITION);
+ }
+
+ protected BaseEqualityDeltaWriter(
+ StructLike partition,
+ Schema schema,
+ Schema deleteSchema,
+ DeleteGranularity deleteGranularity) {
Review Comment:
We don't have a lot of input from the Spark side as it was shipped in 1.5 for the first time. There, the situation is also very different as the number of deletes is proposal to the number of changes in a batch.
@chenjunjiedada, you faced the concurrent compaction issue in the past. Would you have a chance to try this PR and see how it performs for your use cases?
@stevenzwu @rdblue @RussellSpitzer, would love to hear your thoughts as well.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
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 {
private final StructProjection structProjection;
private RollingFileWriter dataWriter;
private RollingEqDeleteWriter eqDeleteWriter;
- private SortedPosDeleteWriter<T> posDeleteWriter;
+ private FileWriter<PositionDelete<T>, DeleteWriteResult> posDeleteWriter;
private Map<StructLike, PathOffset> insertedRowMap;
protected BaseEqualityDeltaWriter(StructLike partition, Schema schema, Schema deleteSchema) {
+ this(partition, schema, deleteSchema, DeleteGranularity.PARTITION);
+ }
+
+ protected BaseEqualityDeltaWriter(
+ StructLike partition,
+ Schema schema,
+ Schema deleteSchema,
+ DeleteGranularity deleteGranularity) {
Review Comment:
We don't have a lot of input from the Spark side as it was shipped in 1.5 for the first time. There, the situation is also very different as the number of deletes is different.
@chenjunjiedada, you faced the concurrent compaction issue in the past. Would you have a chance to try this PR and see how it performs for your use cases?
@stevenzwu @rdblue @RussellSpitzer, would love to hear your thoughts as well.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #10200:
URL: https://github.com/apache/iceberg/pull/10200#discussion_r1580858272
##########
core/src/main/java/org/apache/iceberg/deletes/SortingPositionOnlyDeleteWriter.java:
##########
@@ -118,6 +118,10 @@ private DeleteWriteResult writeFileDeletes() throws IOException {
@SuppressWarnings("CollectionUndefinedEquality")
private DeleteWriteResult writeDeletes(Collection<CharSequence> paths) throws IOException {
+ if (paths.size() == 0) {
Review Comment:
Done
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
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 (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.deletes;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.io.DeleteWriteResult;
+import org.apache.iceberg.io.FileWriter;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.CharSequenceSet;
+
+/**
+ * A position delete writer that produces a separate delete file for each referenced data file.
+ *
+ * <p>This writer does not keep track of seen deletes and assumes all incoming records are ordered
Review Comment:
We do not have a `FileWriterFactory` at hand. What we have is a `FileAppenderFactory` which is already used by the `SortedPosDeleteWriter`.
Also, the internal writer is a `SortedPosDeleteWriter`, which takes care of the sorting for us.
--
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
Re: [PR] Flink: Apply DeleteGranularity for writes [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #10200:
URL: https://github.com/apache/iceberg/pull/10200#issuecomment-2075138226
The markdown is flaky, and not related to the change at all.
--
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