You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "aokolnychyi (via GitHub)" <gi...@apache.org> on 2024/10/11 17:22:52 UTC
[PR] [WIP] Core: Prototype for DVs in V3 [iceberg]
aokolnychyi opened a new pull request, #11302:
URL: https://github.com/apache/iceberg/pull/11302
This PR is a prototype for DVs in V3.
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#issuecomment-2471377793
Closing this in favor of other smaller PRs.
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi closed pull request #11302: [WIP] Core: Prototype for DVs in V3
URL: https://github.com/apache/iceberg/pull/11302
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1824255751
##########
data/src/main/java/org/apache/iceberg/data/BaseDeleteLoader.java:
##########
@@ -259,4 +294,40 @@ private long estimateEqDeletesSize(DeleteFile deleteFile, Schema projection) {
private int estimateRecordSize(Schema schema) {
return schema.columns().stream().mapToInt(TypeUtil::estimateSize).sum();
}
+
+ private boolean containsDVs(Iterable<DeleteFile> deleteFiles) {
+ return Iterables.any(deleteFiles, ContentFileUtil::isDV);
+ }
+
+ private void validateDV(DeleteFile deleteFile, CharSequence filePath) {
+ Preconditions.checkArgument(deleteFile.contentOffset() != null, "DV offset is missing");
Review Comment:
nit: maybe `Invalid DV: offset missing`?
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1829979110
##########
data/src/main/java/org/apache/iceberg/data/BaseDeleteLoader.java:
##########
@@ -146,6 +151,36 @@ private <T> Iterable<T> materialize(CloseableIterable<T> iterable) {
@Override
public PositionDeleteIndex loadPositionDeletes(
Iterable<DeleteFile> deleteFiles, CharSequence filePath) {
+ if (containsDVs(deleteFiles)) {
+ DeleteFile deleteFile = Iterables.getOnlyElement(deleteFiles);
Review Comment:
I am using `Iterables.getOnlyElement()`, which will complain if there is more than one.
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1824259461
##########
data/src/main/java/org/apache/iceberg/data/BaseDeleteLoader.java:
##########
@@ -259,4 +294,40 @@ private long estimateEqDeletesSize(DeleteFile deleteFile, Schema projection) {
private int estimateRecordSize(Schema schema) {
return schema.columns().stream().mapToInt(TypeUtil::estimateSize).sum();
}
+
+ private boolean containsDVs(Iterable<DeleteFile> deleteFiles) {
+ return Iterables.any(deleteFiles, ContentFileUtil::isDV);
+ }
+
+ private void validateDV(DeleteFile deleteFile, CharSequence filePath) {
+ Preconditions.checkArgument(deleteFile.contentOffset() != null, "DV offset is missing");
+ Preconditions.checkArgument(deleteFile.contentSizeInBytes() != null, "DV length is missing");
+ Preconditions.checkArgument(
+ deleteFile.contentSizeInBytes() <= Integer.MAX_VALUE,
+ "Can't read DV larger than 2GB: %s",
+ deleteFile.contentSizeInBytes());
+ Preconditions.checkArgument(
+ filePath.toString().equals(deleteFile.referencedDataFile()),
+ "DV must reference %s, not %s",
+ filePath,
+ deleteFile.referencedDataFile());
+ }
+
+ private byte[] readBytes(InputFile inputFile, long offset, int length) {
+ try (SeekableInputStream stream = inputFile.newStream()) {
+ byte[] bytes = new byte[length];
+
+ if (stream instanceof RangeReadable) {
Review Comment:
looks like this is re-using the same functionality that `PuffinReader` has internally, so not sure if it's worth extracting that into a common method
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1824260251
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/CommitSummary.java:
##########
@@ -50,7 +51,8 @@ class CommitSummary {
.forEach(
deleteFile -> {
deleteFilesRecordCount.addAndGet(deleteFile.recordCount());
- deleteFilesByteCount.addAndGet(deleteFile.fileSizeInBytes());
+ long contentSizeInBytes = ContentFileUtil.contentSizeInBytes(deleteFile);
+ deleteFilesByteCount.addAndGet(contentSizeInBytes);
Review Comment:
do we need to fix other Flink versions too?
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1824130043
##########
core/src/main/java/org/apache/iceberg/FileMetadata.java:
##########
@@ -230,6 +252,15 @@ public DeleteFile build() {
Preconditions.checkArgument(fileSizeInBytes >= 0, "File size is required");
Preconditions.checkArgument(recordCount >= 0, "Record count is required");
+ if (format == FileFormat.PUFFIN) {
+ Preconditions.checkArgument(referencedDataFile != null, "Data file ref is required for DV");
+ Preconditions.checkArgument(contentOffset != null, "Content offset is required for DV");
+ Preconditions.checkArgument(contentSizeInBytes != null, "Content size is required for DV");
+ } else {
+ Preconditions.checkArgument(contentOffset == null, "Content offset is only for DV");
Review Comment:
nit: maybe `Content offset must only be set for DVs`? Same for content size
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1823830578
##########
core/src/main/java/org/apache/iceberg/metrics/CommitMetricsResult.java:
##########
@@ -75,6 +77,9 @@ public interface CommitMetricsResult {
@Nullable
CounterResult addedPositionalDeleteFiles();
+ @Nullable
+ CounterResult addedDVs();
Review Comment:
```suggestion
@Value.Default
default CounterResult addedDVs() {
return null;
}
```
same for the others so that we don't break APIs (and you'll be able to remove the RevAPI failures)
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1823839780
##########
.palantir/revapi.yml:
##########
@@ -1100,6 +1100,15 @@ acceptedBreaks:
- code: "java.class.removed"
old: "enum org.apache.iceberg.BaseMetastoreTableOperations.CommitStatus"
justification: "Removing deprecated code"
+ - code: "java.method.addedToInterface"
Review Comment:
I've commented further below on how to avoid these
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1824254586
##########
data/src/main/java/org/apache/iceberg/data/BaseDeleteLoader.java:
##########
@@ -146,6 +151,36 @@ private <T> Iterable<T> materialize(CloseableIterable<T> iterable) {
@Override
public PositionDeleteIndex loadPositionDeletes(
Iterable<DeleteFile> deleteFiles, CharSequence filePath) {
+ if (containsDVs(deleteFiles)) {
+ DeleteFile deleteFile = Iterables.getOnlyElement(deleteFiles);
Review Comment:
minor: should we add some validation to make sure `deleteFiles` indeed only has a single entry?
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1824174120
##########
core/src/main/java/org/apache/iceberg/deletes/DVFileWriter.java:
##########
@@ -0,0 +1,46 @@
+/*
+ * 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.Closeable;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.io.DeleteWriteResult;
+
+/** A deletion vector file writer. */
+public interface DVFileWriter extends Closeable {
+ /**
+ * Marks a position in a given data file as deleted.
+ *
+ * @param path the data file path
+ * @param pos the data file position
+ * @param spec the data file partition spec
+ * @param partition the data file partition
+ */
+ void write(CharSequence path, long pos, PartitionSpec spec, StructLike partition);
Review Comment:
```suggestion
void write(String path, long pos, PartitionSpec spec, StructLike partition);
```
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1823830578
##########
core/src/main/java/org/apache/iceberg/metrics/CommitMetricsResult.java:
##########
@@ -75,6 +77,9 @@ public interface CommitMetricsResult {
@Nullable
CounterResult addedPositionalDeleteFiles();
+ @Nullable
+ CounterResult addedDVs();
Review Comment:
```suggestion
default CounterResult addedDVs() { return null; }
```
same for the others so that we don't break APIs (and you'll be able to remove the RevAPI failures)
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1824137648
##########
core/src/main/java/org/apache/iceberg/deletes/BaseDVFileWriter.java:
##########
@@ -0,0 +1,194 @@
+/*
+ * 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.function.Function;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileMetadata;
+import org.apache.iceberg.IcebergBuild;
+import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.io.DeleteWriteResult;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.io.StructCopy;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.BlobMetadata;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.CharSequenceMap;
+import org.apache.iceberg.util.CharSequenceSet;
+import org.apache.iceberg.util.ContentFileUtil;
+
+public class BaseDVFileWriter implements DVFileWriter {
+
+ private static final String REF_DATA_FILE_KEY = "referenced-data-file";
+ private static final String CARDINALITY_KEY = "cardinality";
+
+ private final OutputFileFactory fileFactory;
+ private final Function<CharSequence, PositionDeleteIndex> loadPreviousDeletes;
+ private final CharSequenceMap<Deletes> deletesByPath = CharSequenceMap.create();
+ private final CharSequenceMap<BlobMetadata> blobsByPath = CharSequenceMap.create();
+ private DeleteWriteResult result = null;
+
+ public BaseDVFileWriter(
+ OutputFileFactory fileFactory,
+ Function<CharSequence, PositionDeleteIndex> loadPreviousDeletes) {
+ this.fileFactory = fileFactory;
+ this.loadPreviousDeletes = loadPreviousDeletes;
+ }
+
+ @Override
+ public void write(CharSequence path, long pos, PartitionSpec spec, StructLike partition) {
Review Comment:
I wonder whether `path` should rather be a `String` now, since we changed other APIs and moved them from `CharSequence` to `String` (e.g. `ContentFile` now has `location()`, which returns a `String`)
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1824260801
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java:
##########
@@ -79,7 +80,7 @@ void updateFlushResult(WriteResult result) {
Arrays.stream(result.deleteFiles())
.forEach(
deleteFile -> {
- deleteFilesSizeHistogram.update(deleteFile.fileSizeInBytes());
+ deleteFilesSizeHistogram.update(ContentFileUtil.contentSizeInBytes(deleteFile));
Review Comment:
do we need to fix/update other Flink versions too?
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1823830578
##########
core/src/main/java/org/apache/iceberg/metrics/CommitMetricsResult.java:
##########
@@ -75,6 +77,9 @@ public interface CommitMetricsResult {
@Nullable
CounterResult addedPositionalDeleteFiles();
+ @Nullable
+ CounterResult addedDVs();
Review Comment:
```suggestion
default CounterResult addedDVs() { return null; }
```
same for the others
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1798616398
##########
api/src/main/java/org/apache/iceberg/DataFile.java:
##########
@@ -98,12 +98,23 @@ public interface DataFile extends ContentFile<DataFile> {
Types.NestedField SORT_ORDER_ID =
optional(140, "sort_order_id", IntegerType.get(), "Sort order ID");
Types.NestedField SPEC_ID = optional(141, "spec_id", IntegerType.get(), "Partition spec ID");
+ Types.NestedField REFERENCED_DATA_FILE =
Review Comment:
Follows the proposed spec, reserving 142 for row lineage.
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1824161450
##########
core/src/main/java/org/apache/iceberg/deletes/BaseDVFileWriter.java:
##########
@@ -0,0 +1,194 @@
+/*
+ * 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.function.Function;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileMetadata;
+import org.apache.iceberg.IcebergBuild;
+import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.io.DeleteWriteResult;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.io.StructCopy;
+import org.apache.iceberg.puffin.Blob;
+import org.apache.iceberg.puffin.BlobMetadata;
+import org.apache.iceberg.puffin.Puffin;
+import org.apache.iceberg.puffin.PuffinWriter;
+import org.apache.iceberg.puffin.StandardBlobTypes;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.CharSequenceMap;
+import org.apache.iceberg.util.CharSequenceSet;
+import org.apache.iceberg.util.ContentFileUtil;
+
+public class BaseDVFileWriter implements DVFileWriter {
+
+ private static final String REF_DATA_FILE_KEY = "referenced-data-file";
+ private static final String CARDINALITY_KEY = "cardinality";
+
+ private final OutputFileFactory fileFactory;
+ private final Function<CharSequence, PositionDeleteIndex> loadPreviousDeletes;
+ private final CharSequenceMap<Deletes> deletesByPath = CharSequenceMap.create();
+ private final CharSequenceMap<BlobMetadata> blobsByPath = CharSequenceMap.create();
+ private DeleteWriteResult result = null;
+
+ public BaseDVFileWriter(
+ OutputFileFactory fileFactory,
+ Function<CharSequence, PositionDeleteIndex> loadPreviousDeletes) {
+ this.fileFactory = fileFactory;
+ this.loadPreviousDeletes = loadPreviousDeletes;
+ }
+
+ @Override
+ public void write(CharSequence path, long pos, PartitionSpec spec, StructLike partition) {
Review Comment:
I did some minor updates and this should work I believe
```
private final OutputFileFactory fileFactory;
- private final Function<CharSequence, PositionDeleteIndex> loadPreviousDeletes;
- private final CharSequenceMap<Deletes> deletesByPath = CharSequenceMap.create();
- private final CharSequenceMap<BlobMetadata> blobsByPath = CharSequenceMap.create();
+ private final Function<String, PositionDeleteIndex> loadPreviousDeletes;
+ private final Map<String, Deletes> deletesByPath = Maps.newHashMap();
+ private final Map<String, BlobMetadata> blobsByPath = Maps.newHashMap();
private DeleteWriteResult result = null;
public BaseDVFileWriter(
- OutputFileFactory fileFactory,
- Function<CharSequence, PositionDeleteIndex> loadPreviousDeletes) {
+ OutputFileFactory fileFactory, Function<String, PositionDeleteIndex> loadPreviousDeletes) {
this.fileFactory = fileFactory;
this.loadPreviousDeletes = loadPreviousDeletes;
}
@Override
- public void write(CharSequence path, long pos, PartitionSpec spec, StructLike partition) {
- Deletes deletes = deletesByPath.computeIfAbsent(path, () -> new Deletes(path, spec, partition));
+ public void write(String path, long pos, PartitionSpec spec, StructLike partition) {
+ Deletes deletes = deletesByPath.computeIfAbsent(path, k -> new Deletes(path, spec, partition));
PositionDeleteIndex positions = deletes.positions();
positions.delete(pos);
}
@@ -87,7 +87,7 @@ public class BaseDVFileWriter implements DVFileWriter {
try (PuffinWriter closeableWriter = writer) {
for (Deletes deletes : deletesByPath.values()) {
- CharSequence path = deletes.path();
+ String path = deletes.path();
PositionDeleteIndex positions = deletes.positions();
PositionDeleteIndex previousPositions = loadPreviousDeletes.apply(path);
if (previousPositions != null) {
@@ -108,7 +108,7 @@ public class BaseDVFileWriter implements DVFileWriter {
String puffinPath = writer.location();
long puffinFileSize = writer.fileSize();
- for (CharSequence path : deletesByPath.keySet()) {
+ for (String path : deletesByPath.keySet()) {
DeleteFile deleteFile = createDeleteFile(puffinPath, puffinFileSize, path);
deleteFiles.add(deleteFile);
}
@@ -117,8 +117,7 @@ public class BaseDVFileWriter implements DVFileWriter {
}
}
- @SuppressWarnings("CollectionUndefinedEquality")
- private DeleteFile createDeleteFile(String path, long size, CharSequence referencedDataFile) {
+ private DeleteFile createDeleteFile(String path, long size, String referencedDataFile) {
Deletes deletes = deletesByPath.get(referencedDataFile);
BlobMetadata blobMetadata = blobsByPath.get(referencedDataFile);
return FileMetadata.deleteFileBuilder(deletes.spec())
@@ -135,7 +134,7 @@ public class BaseDVFileWriter implements DVFileWriter {
}
private void write(PuffinWriter writer, Deletes deletes) {
- CharSequence path = deletes.path();
+ String path = deletes.path();
PositionDeleteIndex positions = deletes.positions();
BlobMetadata blobMetadata = writer.write(toBlob(positions, path));
blobsByPath.put(path, blobMetadata);
@@ -147,7 +146,7 @@ public class BaseDVFileWriter implements DVFileWriter {
return Puffin.write(outputFile).createdBy(ident).build();
}
- private Blob toBlob(PositionDeleteIndex positions, CharSequence path) {
+ private Blob toBlob(PositionDeleteIndex positions, String path) {
return new Blob(
StandardBlobTypes.DV_V1,
ImmutableList.of(MetadataColumns.ROW_POSITION.fieldId()),
@@ -156,26 +155,23 @@ public class BaseDVFileWriter implements DVFileWriter {
positions.serialize(),
null /* uncompressed */,
ImmutableMap.of(
- REF_DATA_FILE_KEY,
- path.toString(),
- CARDINALITY_KEY,
- String.valueOf(positions.cardinality())));
+ REF_DATA_FILE_KEY, path, CARDINALITY_KEY, String.valueOf(positions.cardinality())));
}
private static class Deletes {
- private final CharSequence path;
+ private final String path;
private final PositionDeleteIndex positions;
private final PartitionSpec spec;
private final StructLike partition;
- private Deletes(CharSequence path, PartitionSpec spec, StructLike partition) {
+ private Deletes(String path, PartitionSpec spec, StructLike partition) {
this.path = path;
this.positions = new BitmapPositionDeleteIndex();
this.spec = spec;
this.partition = StructCopy.copy(partition);
}
- public CharSequence path() {
+ public String path() {
return path;
}
```
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1824171453
##########
core/src/main/java/org/apache/iceberg/deletes/BitmapPositionDeleteIndex.java:
##########
@@ -71,21 +85,128 @@ public void merge(PositionDeleteIndex that) {
@Override
public boolean isDeleted(long position) {
- return roaring64Bitmap.contains(position);
+ return bitmap.contains(position);
}
@Override
public boolean isEmpty() {
- return roaring64Bitmap.isEmpty();
+ return bitmap.isEmpty();
}
@Override
public void forEach(LongConsumer consumer) {
- roaring64Bitmap.forEach(consumer::accept);
+ bitmap.forEach(consumer);
}
@Override
public Collection<DeleteFile> deleteFiles() {
return deleteFiles;
}
+
+ @Override
+ public long cardinality() {
+ return bitmap.cardinality();
+ }
+
+ /**
+ * Serializes the index using the following format:
+ *
+ * <ul>
+ * <li>The length of the bitmap and magic bytes stored as 4 bytes (big-endian).
+ * <li>A 4-byte magic sequence (0xD1D33964) (little-endian).
+ * <li>The bitmap serialized using the portable Roaring spec (little-endian).
+ * <li>A CRC-32 checksum of the bitmap and magic bytes as 4-bytes (big-endian).
+ * </ul>
+ *
+ * Note that the length and the checksum are computed for the bitmap data, which includes the
+ * magic bytes and the serialized bitmap for compatibility with Delta Lake.
+ */
+ @Override
+ public ByteBuffer serialize() {
+ bitmap.runLengthEncode(); // run-length encode the bitmap before serializing
+ int bitmapDataLength = computeBitmapDataLength(bitmap); // magic bytes + bitmap
+ byte[] bytes = new byte[LENGTH_SIZE_BYTES + bitmapDataLength + CRC_SIZE_BYTES];
+ ByteBuffer buffer = ByteBuffer.wrap(bytes);
+ buffer.putInt(bitmapDataLength);
+ serializeBitmapData(bytes, bitmapDataLength, bitmap);
+ int crcOffset = LENGTH_SIZE_BYTES + bitmapDataLength;
+ int crc = computeChecksum(bytes, bitmapDataLength);
+ buffer.putInt(crcOffset, crc);
+ buffer.rewind();
+ return buffer;
+ }
+
+ public static PositionDeleteIndex deserialize(byte[] bytes, DeleteFile deleteFile) {
+ ByteBuffer buffer = ByteBuffer.wrap(bytes);
+ int bitmapDataLength = readBitmapDataLength(buffer, deleteFile);
+ RoaringPositionBitmap bitmap = deserializeBitmap(bytes, bitmapDataLength, deleteFile);
+ int crc = computeChecksum(bytes, bitmapDataLength);
+ int crcOffset = LENGTH_SIZE_BYTES + bitmapDataLength;
+ int expectedCrc = buffer.getInt(crcOffset);
+ Preconditions.checkArgument(crc == expectedCrc, "Invalid CRC");
+ return new BitmapPositionDeleteIndex(bitmap, deleteFile);
+ }
+
+ // computes and validates the length of the bitmap data (magic bytes + bitmap)
+ private static int computeBitmapDataLength(RoaringPositionBitmap bitmap) {
+ long length = MAGIC_NUMBER_SIZE_BYTES + bitmap.serializedSizeInBytes();
+ long bufferSize = LENGTH_SIZE_BYTES + length + CRC_SIZE_BYTES;
+ Preconditions.checkState(bufferSize <= Integer.MAX_VALUE, "Can't serialize index > 2GB");
+ return (int) length;
+ }
+
+ // serializes the bitmap data (magic bytes + bitmap) using the little-endian byte order
+ private static void serializeBitmapData(
+ byte[] bytes, int bitmapDataLength, RoaringPositionBitmap bitmap) {
+ ByteBuffer bitmapData = pointToBitmapData(bytes, bitmapDataLength);
+ bitmapData.putInt(MAGIC_NUMBER);
+ bitmap.serialize(bitmapData);
+ }
+
+ // points to the bitmap data in the blob
+ private static ByteBuffer pointToBitmapData(byte[] bytes, int bitmapDataLength) {
+ ByteBuffer bitmapData = ByteBuffer.wrap(bytes, BITMAP_DATA_OFFSET, bitmapDataLength);
+ bitmapData.order(ByteOrder.LITTLE_ENDIAN);
+ return bitmapData;
+ }
+
+ // checks the blob size is equal to the bitmap data length + extra bytes for length and CRC
+ private static int readBitmapDataLength(ByteBuffer buffer, DeleteFile deleteFile) {
+ int length = buffer.getInt();
+ long expectedLength = deleteFile.contentSizeInBytes() - LENGTH_SIZE_BYTES - CRC_SIZE_BYTES;
+ Preconditions.checkArgument(
+ length == expectedLength,
+ "Invalid bitmap data length %s, expected %s",
Review Comment:
```suggestion
"Invalid bitmap data length: %s, expected %s",
```
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1824120356
##########
core/src/main/java/org/apache/iceberg/DeleteFileIndex.java:
##########
@@ -453,7 +490,18 @@ DeleteFileIndex build() {
globalDeletes.isEmpty() ? null : globalDeletes,
eqDeletesByPartition.isEmpty() ? null : eqDeletesByPartition,
posDeletesByPartition.isEmpty() ? null : posDeletesByPartition,
- posDeletesByPath.isEmpty() ? null : posDeletesByPath);
+ posDeletesByPath.isEmpty() ? null : posDeletesByPath,
+ dvByPath.isEmpty() ? null : dvByPath);
+ }
+
+ private void add(Map<String, DeleteFile> dvByPath, DeleteFile dv) {
+ String path = dv.referencedDataFile();
+ DeleteFile existingDV = dvByPath.putIfAbsent(path, dv);
+ if (existingDV != null) {
+ throw new ValidationException(
+ "Can't index multiple DVs for %s: %s and %s",
Review Comment:
would it make sense to add `existing` in the wording here so that readers of this failure know which one points to the existing DV?
--
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] Core: Prototype for DVs in V3 [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #11302:
URL: https://github.com/apache/iceberg/pull/11302#discussion_r1824246937
##########
core/src/test/java/org/apache/iceberg/TestSnapshotSummary.java:
##########
@@ -358,4 +358,66 @@ public void rewriteWithDeletesAndDuplicates() {
.containsEntry(SnapshotSummary.TOTAL_FILE_SIZE_PROP, "20")
.containsEntry(SnapshotSummary.TOTAL_RECORDS_PROP, "1");
}
+
+ @TestTemplate
+ public void testFileSizeSummaryWithDVs() {
+ assumeThat(formatVersion).isGreaterThanOrEqualTo(3);
+
+ DeleteFile dv1 = newDV(FILE_A);
+ table.newRowDelta().addDeletes(dv1).commit();
+
+ DeleteFile dv2 = newDV(FILE_B);
+ table.newRowDelta().addDeletes(dv2).commit();
+
+ Map<String, String> summary1 = table.currentSnapshot().summary();
+ long addedPosDeletes1 = dv2.recordCount();
+ long addedFileSize1 = dv2.contentSizeInBytes();
+ long totalPosDeletes1 = dv1.recordCount() + dv2.recordCount();
+ long totalFileSize1 = dv1.contentSizeInBytes() + dv2.contentSizeInBytes();
+ assertThat(summary1)
+ .doesNotContainKey(SnapshotSummary.ADD_POS_DELETE_FILES_PROP)
Review Comment:
minor: other tests in this class do a `hasSize()` check and check for all props (except for the `iceberg-version`). This is being done mainly to make sure the summary contains all the expected info.
I think it would be good here to add the `hasSize(12)` check and check that all of the 11 (not counting `iceberg-version`) have the right values
--
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