diff --git a/cpp/velox/compute/iceberg/IcebergPlanConverter.cc b/cpp/velox/compute/iceberg/IcebergPlanConverter.cc index 07c40e6e1c7..5212381623a 100644 --- a/cpp/velox/compute/iceberg/IcebergPlanConverter.cc +++ b/cpp/velox/compute/iceberg/IcebergPlanConverter.cc @@ -19,6 +19,24 @@ namespace gluten { +namespace { + +using SubstraitDeleteBoundsMap = ::substrait::ReadRel_LocalFiles_FileOrFiles::IcebergReadOptions::DeleteFile::Map; + +std::unordered_map parseBounds(const SubstraitDeleteBoundsMap& bounds) { + std::unordered_map parsed; + parsed.reserve(bounds.key_values_size()); + + for (int i = 0; i < bounds.key_values_size(); ++i) { + const auto& kv = bounds.key_values(i); + parsed.emplace(kv.key(), kv.value()); + } + + return parsed; +} + +} // namespace + std::shared_ptr IcebergPlanConverter::parseIcebergSplitInfo( substrait::ReadRel_LocalFiles_FileOrFiles file, std::shared_ptr splitInfo) { @@ -70,7 +88,14 @@ std::shared_ptr IcebergPlanConverter::parseIcebergSplitInfo( break; } deletes.emplace_back(IcebergDeleteFile( - fileContent, deleteFile.filepath(), format, deleteFile.recordcount(), deleteFile.filesize())); + fileContent, + deleteFile.filepath(), + format, + deleteFile.recordcount(), + deleteFile.filesize(), + {}, + parseBounds(deleteFile.lowerbounds()), + parseBounds(deleteFile.upperbounds()))); } icebergSplitInfo->deleteFilesVec.emplace_back(deletes); } else { diff --git a/gluten-iceberg/pom.xml b/gluten-iceberg/pom.xml index 056535b9405..2cb694f8e9a 100644 --- a/gluten-iceberg/pom.xml +++ b/gluten-iceberg/pom.xml @@ -101,6 +101,11 @@ scalatest_${scala.binary.version} test + + junit + junit + test + diff --git a/gluten-iceberg/src/main/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNode.java b/gluten-iceberg/src/main/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNode.java index cca3a4f79db..cfae06a23aa 100644 --- a/gluten-iceberg/src/main/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNode.java +++ b/gluten-iceberg/src/main/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNode.java @@ -19,10 +19,8 @@ import io.substrait.proto.ReadRel; import org.apache.iceberg.DeleteFile; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.nio.ByteBuffer; +import java.util.*; public class IcebergLocalFilesNode extends LocalFilesNode { private final List> deleteFilesList; @@ -96,6 +94,8 @@ protected void processFileBuilder(ReadRel.LocalFiles.FileOrFiles.Builder fileBui deleteFileBuilder.setFilePath(delete.path().toString()); deleteFileBuilder.setFileSize(delete.fileSizeInBytes()); deleteFileBuilder.setRecordCount(delete.recordCount()); + deleteFileBuilder.setLowerBounds(encodeBounds(delete.lowerBounds())); + deleteFileBuilder.setUpperBounds(encodeBounds(delete.upperBounds())); switch (delete.format()) { case PARQUET: ReadRel.LocalFiles.FileOrFiles.ParquetReadOptions parquetReadOptions = @@ -118,4 +118,32 @@ protected void processFileBuilder(ReadRel.LocalFiles.FileOrFiles.Builder fileBui } fileBuilder.setIceberg(icebergBuilder); } + + private static ReadRel.LocalFiles.FileOrFiles.IcebergReadOptions.DeleteFile.Map encodeBounds( + Map bounds) { + ReadRel.LocalFiles.FileOrFiles.IcebergReadOptions.DeleteFile.Map.Builder builder = + ReadRel.LocalFiles.FileOrFiles.IcebergReadOptions.DeleteFile.Map.newBuilder(); + + if (bounds == null || bounds.isEmpty()) { + return builder.build(); + } + + for (Map.Entry entry : bounds.entrySet()) { + if (entry.getValue() == null) { + continue; + } + + ByteBuffer duplicate = entry.getValue().asReadOnlyBuffer(); + byte[] bytes = new byte[duplicate.remaining()]; + duplicate.get(bytes); + + builder.addKeyValues( + ReadRel.LocalFiles.FileOrFiles.IcebergReadOptions.DeleteFile.Map.KeyValue.newBuilder() + .setKey(entry.getKey()) + .setValue(Base64.getEncoder().encodeToString(bytes)) + .build()); + } + + return builder.build(); + } } diff --git a/gluten-iceberg/src/test/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNodeBoundsTest.java b/gluten-iceberg/src/test/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNodeBoundsTest.java new file mode 100644 index 00000000000..01fc50035cb --- /dev/null +++ b/gluten-iceberg/src/test/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNodeBoundsTest.java @@ -0,0 +1,214 @@ +/* + * 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.gluten.substrait.rel; + +import io.substrait.proto.ReadRel; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.StructLike; +import org.junit.Assert; +import org.junit.Test; + +import java.nio.ByteBuffer; +import java.util.Base64; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +public class IcebergLocalFilesNodeBoundsTest { + private static final int ICEBERG_POS_FIELD_ID = 2147483545; + + @Test + public void serializesDeleteFileLowerAndUpperBounds() { + byte[] lowerBytes = new byte[] {1, 2, 3, 4}; + byte[] upperBytes = new byte[] {5, 6, 7, 8}; + + FakeDeleteFile deleteFile = + new FakeDeleteFile( + FileContent.POSITION_DELETES, + FileFormat.PARQUET, + "/tmp/delete.parquet", + 2L, + 123L, + Collections.emptyList(), + Collections.singletonMap(ICEBERG_POS_FIELD_ID, ByteBuffer.wrap(lowerBytes)), + Collections.singletonMap(ICEBERG_POS_FIELD_ID, ByteBuffer.wrap(upperBytes)), + null); + IcebergLocalFilesNode node = + new IcebergLocalFilesNode( + 0, + Collections.singletonList("/tmp/data.parquet"), + Collections.singletonList(0L), + Collections.singletonList(100L), + Collections.singletonList(Collections.emptyMap()), + LocalFilesNode.ReadFileFormat.ParquetReadFormat, + Collections.emptyList(), + Collections.singletonList(Collections.singletonList(deleteFile)), + Collections.singletonList(Collections.emptyMap())); + + ReadRel.LocalFiles.FileOrFiles.Builder fileBuilder = + ReadRel.LocalFiles.FileOrFiles.newBuilder(); + + node.processFileBuilder(fileBuilder, 0); + + ReadRel.LocalFiles.FileOrFiles.IcebergReadOptions.DeleteFile actualDelete = + fileBuilder.getIceberg().getDeleteFiles(0); + + Assert.assertEquals(1, actualDelete.getLowerBounds().getKeyValuesCount()); + Assert.assertEquals( + ICEBERG_POS_FIELD_ID, actualDelete.getLowerBounds().getKeyValues(0).getKey()); + Assert.assertEquals( + Base64.getEncoder().encodeToString(lowerBytes), + actualDelete.getLowerBounds().getKeyValues(0).getValue()); + + Assert.assertEquals(1, actualDelete.getUpperBounds().getKeyValuesCount()); + Assert.assertEquals( + ICEBERG_POS_FIELD_ID, actualDelete.getUpperBounds().getKeyValues(0).getKey()); + Assert.assertEquals( + Base64.getEncoder().encodeToString(upperBytes), + actualDelete.getUpperBounds().getKeyValues(0).getValue()); + } + + private static final class FakeDeleteFile implements DeleteFile { + private final FileContent content; + private final FileFormat format; + private final String path; + private final long recordCount; + private final long fileSizeInBytes; + private final List equalityFieldIds; + private final Map lowerBounds; + private final Map upperBounds; + private final Long dataSequenceNumber; + + private FakeDeleteFile( + FileContent content, + FileFormat format, + String path, + long recordCount, + long fileSizeInBytes, + List equalityFieldIds, + Map lowerBounds, + Map upperBounds, + Long dataSequenceNumber) { + this.content = content; + this.format = format; + this.path = path; + this.recordCount = recordCount; + this.fileSizeInBytes = fileSizeInBytes; + this.equalityFieldIds = equalityFieldIds; + this.lowerBounds = lowerBounds; + this.upperBounds = upperBounds; + this.dataSequenceNumber = dataSequenceNumber; + } + + @Override + public Long pos() { + return 0L; + } + + @Override + public int specId() { + return 0; + } + + @Override + public FileContent content() { + return content; + } + + @Override + public CharSequence path() { + return path; + } + + @Override + public FileFormat format() { + return format; + } + + @Override + public StructLike partition() { + return null; + } + + @Override + public long recordCount() { + return recordCount; + } + + @Override + public long fileSizeInBytes() { + return fileSizeInBytes; + } + + @Override + public Map columnSizes() { + return null; + } + + @Override + public Map valueCounts() { + return null; + } + + @Override + public Map nullValueCounts() { + return null; + } + + @Override + public Map nanValueCounts() { + return null; + } + + @Override + public Map lowerBounds() { + return lowerBounds; + } + + @Override + public Map upperBounds() { + return upperBounds; + } + + @Override + public ByteBuffer keyMetadata() { + return null; + } + + @Override + public List equalityFieldIds() { + return equalityFieldIds; + } + + @Override + public Long dataSequenceNumber() { + return dataSequenceNumber; + } + + @Override + public DeleteFile copy() { + return this; + } + + @Override + public DeleteFile copyWithoutStats() { + return this; + } + } +}