You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "amogh-jahagirdar (via GitHub)" <gi...@apache.org> on 2024/11/15 17:11:49 UTC
[PR] (WIP) Write DVs in Spark for V3 tables [iceberg]
amogh-jahagirdar opened a new pull request, #11561:
URL: https://github.com/apache/iceberg/pull/11561
(no 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] Spark: Write DVs for V3 MoR tables [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1863716254
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java:
##########
@@ -483,6 +493,14 @@ protected PartitioningWriter<PositionDelete<InternalRow>, DeleteWriteResult> new
SparkFileWriterFactory writers,
OutputFileFactory files,
Context context) {
+ TableOperations ops = ((HasTableOperations) table).operations();
+ Function<CharSequence, PositionDeleteIndex> previousDeleteLoader =
+ rewritableDeletes != null
+ ? new PreviousDeleteLoader(table, rewritableDeletes)
+ : path -> null;
+ if (ops.current().formatVersion() >= 3) {
Review Comment:
I did the co-location and reused the context variable (went with using context directly though in the useDVs check)
--
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] Spark: Write DVs for V3 MoR tables [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1863707596
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java:
##########
@@ -168,7 +168,7 @@ protected Map<String, DeleteFileSet> rewritableDeletes() {
for (ScanTask task : tasks()) {
FileScanTask fileScanTask = task.asFileScanTask();
for (DeleteFile deleteFile : fileScanTask.deletes()) {
- if (ContentFileUtil.isFileScoped(deleteFile)) {
+ if (ContentFileUtil.isFileScoped(deleteFile) || ContentFileUtil.isDV(deleteFile)) {
Review Comment:
Great catch, yes to both points. the file scoped Util correctly handles the DV case by checking the referenced data file, and we for DVs we do need to merge all existing position deletes, regardless of scope. I've fixed that and added a test case for update/delete/merge which will create a mix of file scoped and partition scoped deletes, and then finally produce a delete which has the positions merged from these existing deletes.
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java:
##########
@@ -168,7 +168,7 @@ protected Map<String, DeleteFileSet> rewritableDeletes() {
for (ScanTask task : tasks()) {
FileScanTask fileScanTask = task.asFileScanTask();
for (DeleteFile deleteFile : fileScanTask.deletes()) {
- if (ContentFileUtil.isFileScoped(deleteFile)) {
+ if (ContentFileUtil.isFileScoped(deleteFile) || ContentFileUtil.isDV(deleteFile)) {
Review Comment:
Great catch, yes to both points. the file scoped Util correctly handles the DV case by checking the referenced data file, and we for DVs we do need to merge all existing position deletes, regardless of scope. I've fixed that and added a test case for update/delete/merge which will create a mix of file scoped and partition scoped deletes, and then finally produce a DV which has the positions merged from these existing 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] Spark: Write DVs for V3 MoR tables [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1851471663
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java:
##########
@@ -521,7 +528,10 @@ public void deleteSingleRecordProducesDeleteOperation() throws NoSuchTableExcept
} else {
// this is a RowDelta that produces a "delete" instead of "overwrite"
validateMergeOnRead(currentSnapshot, "1", "1", null);
- validateProperty(currentSnapshot, ADD_POS_DELETE_FILES_PROP, "1");
+ TableOperations ops = ((HasTableOperations) table).operations();
Review Comment:
I don't think we need `ops` to get the version. The super class already has `formatVersion` defined, so we can just examine that field
--
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] Spark: Write DVs for V3 MoR tables [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1873681281
##########
core/src/main/java/org/apache/iceberg/io/PartitioningDVWriter.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * 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.io;
+
+import java.io.IOException;
+import java.util.function.Function;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * PartitioningDVWriter is a PartitioningWriter implementation that writes DVs for a file position
Review Comment:
Added a `.` since class level javadocs tend to have that, and rephrased a bit
--
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] Spark: Write DVs for V3 MoR tables [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar merged PR #11561:
URL: https://github.com/apache/iceberg/pull/11561
--
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] Spark: Write DVs for V3 MoR tables [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1854358716
##########
core/src/main/java/org/apache/iceberg/io/PartitioningDVWriter.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.io;
+
+import java.io.IOException;
+import java.util.function.Function;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * PartitioningDVWriter is a PartitioningWriter implementation which writes DVs for a given file
+ * position
Review Comment:
I'll fix this to avoid the awkward formatting
--
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] (WIP) Write DVs in Spark for V3 tables [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#issuecomment-2479469983
Still some failing tests, and figuring out a good pattern to extend the existing Delete/Merge/Update tests to run against DVs. Something also worth thinking about for V3 is preventing partition granularity from being set since it's at odds with DVs.
We could also take a stance that when an upgrade to V3 is performed, we invalidate the delete granularity property completely since in V3 we should only be writing DVs and not position deletes (the granularity property is really only relevant for V2 pos 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] Spark: Write DVs for V3 MoR tables [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1851471897
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java:
##########
@@ -521,7 +528,10 @@ public void deleteSingleRecordProducesDeleteOperation() throws NoSuchTableExcept
} else {
// this is a RowDelta that produces a "delete" instead of "overwrite"
validateMergeOnRead(currentSnapshot, "1", "1", null);
- validateProperty(currentSnapshot, ADD_POS_DELETE_FILES_PROP, "1");
+ TableOperations ops = ((HasTableOperations) table).operations();
+ String property =
+ ops.current().formatVersion() >= 3 ? ADDED_DVS_PROP : ADD_POS_DELETE_FILES_PROP;
Review Comment:
```suggestion
formatVersion >= 3 ? ADDED_DVS_PROP : ADD_POS_DELETE_FILES_PROP;
```
--
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] Spark: Write DVs for V3 MoR tables [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1873680851
##########
core/src/main/java/org/apache/iceberg/io/PartitioningDVWriter.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * 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.io;
+
+import java.io.IOException;
+import java.util.function.Function;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * PartitioningDVWriter is a PartitioningWriter implementation that writes DVs for a file position
+ */
+public class PartitioningDVWriter<T>
+ implements PartitioningWriter<PositionDelete<T>, DeleteWriteResult> {
+ private final DVFileWriter fileWriter;
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] (WIP) Write DVs in Spark for V3 tables [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1848169470
##########
core/src/main/java/org/apache/iceberg/io/PartitioningDVWriter.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.io;
+
+import java.io.IOException;
+import java.util.function.Function;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * PartitioningDVWriter is a PartitioningWriter implementation which writes DVs for a given file
+ * position
+ */
+public class PartitioningDVWriter<T>
Review Comment:
I think having this class here makes a lot of sense and I've reused it for https://github.com/apache/iceberg/pull/11545
--
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] Spark: Write DVs for V3 MoR tables [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1851486037
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java:
##########
@@ -397,4 +450,9 @@ protected void assertAllBatchScansVectorized(SparkPlan plan) {
List<SparkPlan> batchScans = SparkPlanUtil.collectBatchScans(plan);
assertThat(batchScans).hasSizeGreaterThan(0).allMatch(SparkPlan::supportsColumnar);
}
+
+ protected int formatVersion(Table table) {
Review Comment:
I don't think this is necessary anymore
--
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] (WIP) Write DVs in Spark for V3 tables [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1849847923
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java:
##########
@@ -267,31 +271,47 @@ private Dataset<Row> toDS(String schema, String jsonData) {
}
protected void validateDelete(
- Snapshot snapshot, String changedPartitionCount, String deletedDataFiles) {
- validateSnapshot(snapshot, DELETE, changedPartitionCount, deletedDataFiles, null, null);
+ int formatVersion, Snapshot snapshot, String changedPartitionCount, String deletedDataFiles) {
+ validateSnapshot(
+ formatVersion, snapshot, DELETE, changedPartitionCount, deletedDataFiles, null, null);
Review Comment:
rather than passing around the format version, would it make sense to parameterize `SparkRowLevelOperationsTestBase` and add `formatVersion` as another parameter?
--
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] Spark: Write DVs for V3 MoR tables [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1863714851
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java:
##########
@@ -483,6 +493,14 @@ protected PartitioningWriter<PositionDelete<InternalRow>, DeleteWriteResult> new
SparkFileWriterFactory writers,
Review Comment:
Done!
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java:
##########
@@ -483,6 +493,14 @@ protected PartitioningWriter<PositionDelete<InternalRow>, DeleteWriteResult> new
SparkFileWriterFactory writers,
OutputFileFactory files,
Context context) {
+ TableOperations ops = ((HasTableOperations) table).operations();
+ Function<CharSequence, PositionDeleteIndex> previousDeleteLoader =
Review Comment:
I like the idea of a static util, added!
--
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] Spark: Write DVs for V3 MoR tables [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1872486092
##########
core/src/main/java/org/apache/iceberg/io/PartitioningDVWriter.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * 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.io;
+
+import java.io.IOException;
+import java.util.function.Function;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * PartitioningDVWriter is a PartitioningWriter implementation that writes DVs for a file position
Review Comment:
Do we need `.` at the end? I am also not sure I understand `writes DVs for a file position`. Is there a way to rephrase the description a bit?
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java:
##########
@@ -182,15 +183,20 @@ public DeltaWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
}
private Broadcast<Map<String, DeleteFileSet>> broadcastRewritableDeletes() {
- if (context.deleteGranularity() == DeleteGranularity.FILE && scan != null) {
- Map<String, DeleteFileSet> rewritableDeletes = scan.rewritableDeletes();
+ if (scan != null && shouldRewriteDeletes()) {
+ Map<String, DeleteFileSet> rewritableDeletes = scan.rewritableDeletes(context.useDVs());
if (rewritableDeletes != null && !rewritableDeletes.isEmpty()) {
return sparkContext.broadcast(rewritableDeletes);
}
}
return null;
}
+ private boolean shouldRewriteDeletes() {
+ // Rewritable deletes can either be DVs or file scoped position deletes
Review Comment:
Minor: Other comments seem to start with lower case letters.
##########
core/src/main/java/org/apache/iceberg/io/PartitioningDVWriter.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * 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.io;
+
+import java.io.IOException;
+import java.util.function.Function;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * PartitioningDVWriter is a PartitioningWriter implementation that writes DVs for a file position
+ */
+public class PartitioningDVWriter<T>
+ implements PartitioningWriter<PositionDelete<T>, DeleteWriteResult> {
+ private final DVFileWriter fileWriter;
Review Comment:
Optional: What about just `writer` and an empty line before to separate the block with fields as the class definition is split on multiple lines?
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java:
##########
@@ -162,13 +162,14 @@ public void filter(Predicate[] predicates) {
}
}
- protected Map<String, DeleteFileSet> rewritableDeletes() {
+ protected Map<String, DeleteFileSet> rewritableDeletes(boolean forDVs) {
Map<String, DeleteFileSet> rewritableDeletes = Maps.newHashMap();
for (ScanTask task : tasks()) {
FileScanTask fileScanTask = task.asFileScanTask();
for (DeleteFile deleteFile : fileScanTask.deletes()) {
- if (ContentFileUtil.isFileScoped(deleteFile)) {
+ // Both file-scoped and partition-scoped position deletes must be rewritten for DVs
+ if (forDVs || ContentFileUtil.isFileScoped(deleteFile)) {
Review Comment:
It would be nice to also cover this with tests.
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java:
##########
@@ -162,13 +162,14 @@ public void filter(Predicate[] predicates) {
}
}
- protected Map<String, DeleteFileSet> rewritableDeletes() {
+ protected Map<String, DeleteFileSet> rewritableDeletes(boolean forDVs) {
Map<String, DeleteFileSet> rewritableDeletes = Maps.newHashMap();
for (ScanTask task : tasks()) {
FileScanTask fileScanTask = task.asFileScanTask();
for (DeleteFile deleteFile : fileScanTask.deletes()) {
- if (ContentFileUtil.isFileScoped(deleteFile)) {
+ // Both file-scoped and partition-scoped position deletes must be rewritten for DVs
Review Comment:
Minor: Other comments in this class start with lower case letters.
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java:
##########
@@ -182,15 +183,20 @@ public DeltaWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
}
private Broadcast<Map<String, DeleteFileSet>> broadcastRewritableDeletes() {
- if (context.deleteGranularity() == DeleteGranularity.FILE && scan != null) {
- Map<String, DeleteFileSet> rewritableDeletes = scan.rewritableDeletes();
+ if (scan != null && shouldRewriteDeletes()) {
+ Map<String, DeleteFileSet> rewritableDeletes = scan.rewritableDeletes(context.useDVs());
if (rewritableDeletes != null && !rewritableDeletes.isEmpty()) {
return sparkContext.broadcast(rewritableDeletes);
}
}
return null;
}
+ private boolean shouldRewriteDeletes() {
+ // Rewritable deletes can either be DVs or file scoped position deletes
Review Comment:
I think a more accurate description would be that we have to rewrite deletes when dealing with DVs and file-scoped deletes.
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java:
##########
@@ -162,13 +162,14 @@ public void filter(Predicate[] predicates) {
}
}
- protected Map<String, DeleteFileSet> rewritableDeletes() {
+ protected Map<String, DeleteFileSet> rewritableDeletes(boolean forDVs) {
Map<String, DeleteFileSet> rewritableDeletes = Maps.newHashMap();
for (ScanTask task : tasks()) {
FileScanTask fileScanTask = task.asFileScanTask();
for (DeleteFile deleteFile : fileScanTask.deletes()) {
- if (ContentFileUtil.isFileScoped(deleteFile)) {
+ // Both file-scoped and partition-scoped position deletes must be rewritten for DVs
+ if (forDVs || ContentFileUtil.isFileScoped(deleteFile)) {
Review Comment:
What about equality deletes? Those should be ignored. If so, what about `shouldRewrite` method?
```
for (DeleteFile deleteFile : fileScanTask.deletes()) {
if (shouldRewrite(deleteFile, forDVs)) {
...
}
}
```
```
private boolean shouldRewrite(DeleteFile deleteFile, boolean dvsEnabled) {
...
}
```
--
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] (WIP) Write DVs in Spark for V3 tables [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1844211908
##########
core/src/main/java/org/apache/iceberg/io/PartitioningDVWriter.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.io;
+
+import java.io.IOException;
+import java.util.function.Function;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * PartitioningDVWriter is a PartitioningWriter implementation which writes DVs for a given file
+ * position
+ */
+public class PartitioningDVWriter<T>
Review Comment:
I wrote this to avoid changing quite a bit of machinery in SparkPositionDeltaWrite which relies on PartitioningWriter. With this adapter we can just return this writer at a single point, and the existing delegate logic in SparkPositionDeltaWrite just works (line 495 in SparkPositionDeltaWRite). Another advantage is PartitioningWriter is a long standing interface which user's existing code could've been written against and so this implementation could just fit that interface.
--
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] (WIP) Write DVs in Spark for V3 tables [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1844213414
##########
core/src/main/java/org/apache/iceberg/io/PartitioningDVWriter.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.io;
+
+import java.io.IOException;
+import java.util.function.Function;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * PartitioningDVWriter is a PartitioningWriter implementation which writes DVs for a given file
+ * position
+ */
+public class PartitioningDVWriter<T>
Review Comment:
Alternatively we'd have to update SparkPositioNDeltaWrite DeleteOnlyDataWriter and DataAndDelteWriter to take in a function as a delegate (instead of a writer as a delegate) and the function either just uses the existing DV writer for V3 tables or goes to the Fanout writers. But this just makes it more complicated
--
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] (WIP) Write DVs in Spark for V3 tables [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1850541469
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java:
##########
@@ -267,31 +271,47 @@ private Dataset<Row> toDS(String schema, String jsonData) {
}
protected void validateDelete(
- Snapshot snapshot, String changedPartitionCount, String deletedDataFiles) {
- validateSnapshot(snapshot, DELETE, changedPartitionCount, deletedDataFiles, null, null);
+ int formatVersion, Snapshot snapshot, String changedPartitionCount, String deletedDataFiles) {
+ validateSnapshot(
+ formatVersion, snapshot, DELETE, changedPartitionCount, deletedDataFiles, null, null);
Review Comment:
Yeah after writing this all out, I think so. I originally wanted to avoid an exploding number of tests that could be caused by parameterizing the tests but after seeing which tests explicitly we should skip becaujse they're not relevant for DVs, we can parameterize the format and just add assume calls for those tests.
--
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] Spark: Write DVs for V3 MoR tables [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1859282133
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java:
##########
@@ -168,7 +168,7 @@ protected Map<String, DeleteFileSet> rewritableDeletes() {
for (ScanTask task : tasks()) {
FileScanTask fileScanTask = task.asFileScanTask();
for (DeleteFile deleteFile : fileScanTask.deletes()) {
- if (ContentFileUtil.isFileScoped(deleteFile)) {
+ if (ContentFileUtil.isFileScoped(deleteFile) || ContentFileUtil.isDV(deleteFile)) {
Review Comment:
Aren't DVs considered file-scoped? I think `isFileScoped` will always be true for DVs (please check).
##########
core/src/main/java/org/apache/iceberg/io/PartitioningDVWriter.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.io;
+
+import java.io.IOException;
+import java.util.function.Function;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * PartitioningDVWriter is a PartitioningWriter implementation that writes DVs for a file position
+ */
+public class PartitioningDVWriter<T>
+ implements PartitioningWriter<PositionDelete<T>, DeleteWriteResult> {
+ private final DVFileWriter fileWriter;
+ private DeleteWriteResult result;
+
+ public PartitioningDVWriter(
+ OutputFileFactory fileFactory,
+ Function<CharSequence, PositionDeleteIndex> loadPreviousDeletes) {
+ this.fileWriter = new BaseDVFileWriter(fileFactory, loadPreviousDeletes::apply);
+ }
+
+ @Override
+ public void write(PositionDelete<T> row, PartitionSpec spec, StructLike partition) {
+ fileWriter.delete(row.path().toString(), row.pos(), spec, partition);
+ }
+
+ @Override
+ public DeleteWriteResult result() {
+ Preconditions.checkState(result != null, "Cannot get result from unclosed writer");
+ return result;
+ }
+
+ @Override
+ public void close() throws IOException {
+ fileWriter.close();
Review Comment:
I know we usually check if `result` is null to see if the writer was closed and we are not re-assigning the result object. It probably works in this case (knowing the underlying implementation), but I wonder whether making this check explicit will make the behavior of this method clear even without checking how the delegate writer works.
```
if (result == null) {
fileWriter.close();
this.result = fileWriter.result();
}
```
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java:
##########
@@ -483,6 +493,14 @@ protected PartitioningWriter<PositionDelete<InternalRow>, DeleteWriteResult> new
SparkFileWriterFactory writers,
Review Comment:
Shall we update the comment above this method to mention DVs?
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java:
##########
@@ -191,6 +195,12 @@ private Broadcast<Map<String, DeleteFileSet>> broadcastRewritableDeletes() {
return null;
}
+ private boolean shouldBroadcastRewritableDeletes(TableOperations ops) {
Review Comment:
What about `shouldRewriteDeletes` as a shorter name? Also, I'd consider moving the logic for detecting whether DVs should be used to `SparkWriteConf`, where we can eventually call Eduard's utility method. We can then propagate this flag to the `Context` object within `SparkPositionDeltaWrite`.
```
private boolean shouldRewriteDeletes() {
return context.useDVs() || context.deleteGranularity() == DeleteGranularity.FILE;
}
```
With a method in `SparkWriteConf`:
```
public boolean useDVs() {
// call a utility method on Table
}
```
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java:
##########
@@ -483,6 +493,14 @@ protected PartitioningWriter<PositionDelete<InternalRow>, DeleteWriteResult> new
SparkFileWriterFactory writers,
OutputFileFactory files,
Context context) {
+ TableOperations ops = ((HasTableOperations) table).operations();
+ Function<CharSequence, PositionDeleteIndex> previousDeleteLoader =
+ rewritableDeletes != null
+ ? new PreviousDeleteLoader(table, rewritableDeletes)
+ : path -> null;
+ if (ops.current().formatVersion() >= 3) {
Review Comment:
We can reuse `context.useDVs()` here as well. I'd also co-locate all return statements into one if-else block.
```
FileIO io = table.io();
boolean inputOrdered = context.inputOrdered();
long targetFileSize = context.targetDeleteFileSize();
DeleteGranularity deleteGranularity = context.deleteGranularity();
boolean useDVs = context.useDVs();
if (useDVs) {
return new PartitioningDVWriter<>(...);
} else if (inputOrdered && rewritableDeletes == null) {
return new ClusteredPositionDeleteWriter<>(...);
} else {
return new FanoutPositionOnlyDeleteWriter<>(...);
}
```
If you think that extra variables make things more complicated, we can remove them and use `context` directly. Up to you.
##########
core/src/main/java/org/apache/iceberg/io/PartitioningDVWriter.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.io;
+
+import java.io.IOException;
+import java.util.function.Function;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * PartitioningDVWriter is a PartitioningWriter implementation which writes DVs for a given file
+ * position
+ */
+public class PartitioningDVWriter<T>
Review Comment:
I think this class more or less fits what we assume by `PartitioningWriter` (the ability to write to multiple specs and partitions). One downside is the need to wrap and unwrap deletes into `PositionDelete`, which is probably minor. Let me see how we consume this API in Spark.
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java:
##########
@@ -483,6 +493,14 @@ protected PartitioningWriter<PositionDelete<InternalRow>, DeleteWriteResult> new
SparkFileWriterFactory writers,
OutputFileFactory files,
Context context) {
+ TableOperations ops = ((HasTableOperations) table).operations();
+ Function<CharSequence, PositionDeleteIndex> previousDeleteLoader =
Review Comment:
What about a helper method like `createPreviousDeleteLoader` or `PreviousDeleteLoader.create`?
##########
core/src/main/java/org/apache/iceberg/io/PartitioningDVWriter.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.io;
+
+import java.io.IOException;
+import java.util.function.Function;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * PartitioningDVWriter is a PartitioningWriter implementation which writes DVs for a given file
+ * position
+ */
+public class PartitioningDVWriter<T>
Review Comment:
This generally makes sense to me and like that it simplifies the Spark side. The alternative would be to expose a custom implementation of `PositionDeltaWriter` for DVs. While it would not need `PositionDelete` wrappers, it will require more changes in Spark. Let's stick to what this PR has, it should be a simpler option.
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java:
##########
@@ -168,7 +168,7 @@ protected Map<String, DeleteFileSet> rewritableDeletes() {
for (ScanTask task : tasks()) {
FileScanTask fileScanTask = task.asFileScanTask();
for (DeleteFile deleteFile : fileScanTask.deletes()) {
- if (ContentFileUtil.isFileScoped(deleteFile)) {
+ if (ContentFileUtil.isFileScoped(deleteFile) || ContentFileUtil.isDV(deleteFile)) {
Review Comment:
Actually, if DVs are enabled, we have to include all position deletes, not just file-scoped. The first ever produced DV must include all previous deletes. The only difference is that we can drop file-scoped from the table state (decided in the writer) and still have to keep partition-scoped deletes as they may apply to other data files.
Can we please test migration to DVs for tables that have a mix of file-scoped and partition-scoped position deletes?
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java:
##########
@@ -191,6 +195,12 @@ private Broadcast<Map<String, DeleteFileSet>> broadcastRewritableDeletes() {
return null;
}
+ private boolean shouldBroadcastRewritableDeletes(TableOperations ops) {
Review Comment:
The point I am trying to make is that accessing `TableOperations` here is too low-level.
--
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] Spark: Write DVs for V3 MoR tables [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1853989223
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java:
##########
@@ -521,7 +528,10 @@ public void deleteSingleRecordProducesDeleteOperation() throws NoSuchTableExcept
} else {
// this is a RowDelta that produces a "delete" instead of "overwrite"
validateMergeOnRead(currentSnapshot, "1", "1", null);
- validateProperty(currentSnapshot, ADD_POS_DELETE_FILES_PROP, "1");
+ TableOperations ops = ((HasTableOperations) table).operations();
+ String property =
+ ops.current().formatVersion() >= 3 ? ADDED_DVS_PROP : ADD_POS_DELETE_FILES_PROP;
Review Comment:
Oh yes good catch!
--
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] Spark: Write DVs for V3 MoR tables [iceberg]
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #11561:
URL: https://github.com/apache/iceberg/pull/11561#discussion_r1873686457
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java:
##########
@@ -162,13 +162,14 @@ public void filter(Predicate[] predicates) {
}
}
- protected Map<String, DeleteFileSet> rewritableDeletes() {
+ protected Map<String, DeleteFileSet> rewritableDeletes(boolean forDVs) {
Map<String, DeleteFileSet> rewritableDeletes = Maps.newHashMap();
for (ScanTask task : tasks()) {
FileScanTask fileScanTask = task.asFileScanTask();
for (DeleteFile deleteFile : fileScanTask.deletes()) {
- if (ContentFileUtil.isFileScoped(deleteFile)) {
+ // Both file-scoped and partition-scoped position deletes must be rewritten for DVs
+ if (forDVs || ContentFileUtil.isFileScoped(deleteFile)) {
Review Comment:
Ah I fixed to avoid broadcasting eq. deletes unnecessarily but need to figure out how to add a test which would inspect the broadcast contents, will a publish a separate PR for that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: 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