Skip to content

Commit 70ef0f0

Browse files
infvginfvg
committed
[VL] feat(iceberg): Add delete file bounds propagation
1 parent 2ee5670 commit 70ef0f0

3 files changed

Lines changed: 254 additions & 5 deletions

File tree

‎cpp/velox/compute/iceberg/IcebergPlanConverter.cc‎

Lines changed: 21 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,26 @@
1919

2020
namespace gluten {
2121

22+
namespace {
23+
24+
using SubstraitDeleteBoundsMap =
25+
::substrait::ReadRel_LocalFiles_FileOrFiles::IcebergReadOptions::DeleteFile::Map;
26+
27+
std::unordered_map<int32_t, std::string> parseBounds(
28+
const SubstraitDeleteBoundsMap& bounds) {
29+
std::unordered_map<int32_t, std::string> parsed;
30+
parsed.reserve(bounds.key_values_size());
31+
32+
for (int i = 0; i < bounds.key_values_size(); ++i) {
33+
const auto& kv = bounds.key_values(i);
34+
parsed.emplace(kv.key(), kv.value());
35+
}
36+
37+
return parsed;
38+
}
39+
40+
} // namespace
41+
2242
std::shared_ptr<IcebergSplitInfo> IcebergPlanConverter::parseIcebergSplitInfo(
2343
substrait::ReadRel_LocalFiles_FileOrFiles file,
2444
std::shared_ptr<SplitInfo> splitInfo) {
@@ -70,7 +90,7 @@ std::shared_ptr<IcebergSplitInfo> IcebergPlanConverter::parseIcebergSplitInfo(
7090
break;
7191
}
7292
deletes.emplace_back(IcebergDeleteFile(
73-
fileContent, deleteFile.filepath(), format, deleteFile.recordcount(), deleteFile.filesize()));
93+
fileContent, deleteFile.filepath(), format, deleteFile.recordcount(), deleteFile.filesize(), {}, parseBounds(deleteFile.lowerbounds()), parseBounds(deleteFile.upperbounds())));
7494
}
7595
icebergSplitInfo->deleteFilesVec.emplace_back(deletes);
7696
} else {

‎gluten-iceberg/src/main/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNode.java‎

Lines changed: 33 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,8 @@
1919
import io.substrait.proto.ReadRel;
2020
import org.apache.iceberg.DeleteFile;
2121

22-
import java.util.ArrayList;
23-
import java.util.HashMap;
24-
import java.util.List;
25-
import java.util.Map;
22+
import java.nio.ByteBuffer;
23+
import java.util.*;
2624

2725
public class IcebergLocalFilesNode extends LocalFilesNode {
2826
private final List<List<DeleteFile>> deleteFilesList;
@@ -96,6 +94,8 @@ protected void processFileBuilder(ReadRel.LocalFiles.FileOrFiles.Builder fileBui
9694
deleteFileBuilder.setFilePath(delete.path().toString());
9795
deleteFileBuilder.setFileSize(delete.fileSizeInBytes());
9896
deleteFileBuilder.setRecordCount(delete.recordCount());
97+
deleteFileBuilder.setLowerBounds(encodeBounds(delete.lowerBounds()));
98+
deleteFileBuilder.setUpperBounds(encodeBounds(delete.upperBounds()));
9999
switch (delete.format()) {
100100
case PARQUET:
101101
ReadRel.LocalFiles.FileOrFiles.ParquetReadOptions parquetReadOptions =
@@ -118,4 +118,33 @@ protected void processFileBuilder(ReadRel.LocalFiles.FileOrFiles.Builder fileBui
118118
}
119119
fileBuilder.setIceberg(icebergBuilder);
120120
}
121+
122+
private static ReadRel.LocalFiles.FileOrFiles.IcebergReadOptions.DeleteFile.Map encodeBounds(
123+
Map<Integer, ByteBuffer> bounds) {
124+
ReadRel.LocalFiles.FileOrFiles.IcebergReadOptions.DeleteFile.Map.Builder builder =
125+
ReadRel.LocalFiles.FileOrFiles.IcebergReadOptions.DeleteFile.Map.newBuilder();
126+
127+
if (bounds == null || bounds.isEmpty()) {
128+
return builder.build();
129+
}
130+
131+
for (Map.Entry<Integer, ByteBuffer> entry : bounds.entrySet()) {
132+
if (entry.getValue() == null) {
133+
continue;
134+
}
135+
136+
ByteBuffer duplicate = entry.getValue().asReadOnlyBuffer();
137+
byte[] bytes = new byte[duplicate.remaining()];
138+
duplicate.get(bytes);
139+
140+
builder.addKeyValues(
141+
ReadRel.LocalFiles.FileOrFiles.IcebergReadOptions.DeleteFile.Map.KeyValue
142+
.newBuilder()
143+
.setKey(entry.getKey())
144+
.setValue(Base64.getEncoder().encodeToString(bytes))
145+
.build());
146+
}
147+
148+
return builder.build();
149+
}
121150
}

‎gluten-iceberg/src/test/java/org/apache/gluten/substrait/rel/IcebergLocalFilesNodeBoundsTest.java‎

Lines changed: 200 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,200 @@
1+
package org.apache.gluten.substrait.rel;
2+
3+
import io.substrait.proto.ReadRel;
4+
import org.apache.iceberg.DeleteFile;
5+
import org.apache.iceberg.FileContent;
6+
import org.apache.iceberg.FileFormat;
7+
import org.apache.iceberg.StructLike;
8+
import org.junit.Assert;
9+
import org.junit.Test;
10+
11+
import java.nio.ByteBuffer;
12+
import java.util.Base64;
13+
import java.util.Collections;
14+
import java.util.List;
15+
import java.util.Map;
16+
17+
public class IcebergLocalFilesNodeBoundsTest {
18+
private static final int ICEBERG_POS_FIELD_ID = 2147483545;
19+
20+
@Test
21+
public void serializesDeleteFileLowerAndUpperBounds() {
22+
byte[] lowerBytes = new byte[] {1, 2, 3, 4};
23+
byte[] upperBytes = new byte[] {5, 6, 7, 8};
24+
25+
FakeDeleteFile deleteFile =
26+
new FakeDeleteFile(
27+
FileContent.POSITION_DELETES,
28+
FileFormat.PARQUET,
29+
"/tmp/delete.parquet",
30+
2L,
31+
123L,
32+
Collections.emptyList(),
33+
Collections.singletonMap(ICEBERG_POS_FIELD_ID, ByteBuffer.wrap(lowerBytes)),
34+
Collections.singletonMap(ICEBERG_POS_FIELD_ID, ByteBuffer.wrap(upperBytes)),
35+
null);
36+
IcebergLocalFilesNode node =
37+
new IcebergLocalFilesNode(
38+
0,
39+
Collections.singletonList("/tmp/data.parquet"),
40+
Collections.singletonList(0L),
41+
Collections.singletonList(100L),
42+
Collections.singletonList(Collections.emptyMap()),
43+
LocalFilesNode.ReadFileFormat.ParquetReadFormat,
44+
Collections.emptyList(),
45+
Collections.singletonList(Collections.singletonList(deleteFile)),
46+
Collections.singletonList(Collections.emptyMap()));
47+
48+
ReadRel.LocalFiles.FileOrFiles.Builder fileBuilder =
49+
ReadRel.LocalFiles.FileOrFiles.newBuilder();
50+
51+
node.processFileBuilder(fileBuilder, 0);
52+
53+
ReadRel.LocalFiles.FileOrFiles.IcebergReadOptions.DeleteFile actualDelete =
54+
fileBuilder.getIceberg().getDeleteFiles(0);
55+
56+
Assert.assertEquals(1, actualDelete.getLowerBounds().getKeyValuesCount());
57+
Assert.assertEquals(
58+
ICEBERG_POS_FIELD_ID,
59+
actualDelete.getLowerBounds().getKeyValues(0).getKey());
60+
Assert.assertEquals(
61+
Base64.getEncoder().encodeToString(lowerBytes),
62+
actualDelete.getLowerBounds().getKeyValues(0).getValue());
63+
64+
Assert.assertEquals(1, actualDelete.getUpperBounds().getKeyValuesCount());
65+
Assert.assertEquals(
66+
ICEBERG_POS_FIELD_ID,
67+
actualDelete.getUpperBounds().getKeyValues(0).getKey());
68+
Assert.assertEquals(
69+
Base64.getEncoder().encodeToString(upperBytes),
70+
actualDelete.getUpperBounds().getKeyValues(0).getValue());
71+
}
72+
73+
private static final class FakeDeleteFile implements DeleteFile {
74+
private final FileContent content;
75+
private final FileFormat format;
76+
private final String path;
77+
private final long recordCount;
78+
private final long fileSizeInBytes;
79+
private final List<Integer> equalityFieldIds;
80+
private final Map<Integer, ByteBuffer> lowerBounds;
81+
private final Map<Integer, ByteBuffer> upperBounds;
82+
private final Long dataSequenceNumber;
83+
84+
private FakeDeleteFile(
85+
FileContent content,
86+
FileFormat format,
87+
String path,
88+
long recordCount,
89+
long fileSizeInBytes,
90+
List<Integer> equalityFieldIds,
91+
Map<Integer, ByteBuffer> lowerBounds,
92+
Map<Integer, ByteBuffer> upperBounds,
93+
Long dataSequenceNumber) {
94+
this.content = content;
95+
this.format = format;
96+
this.path = path;
97+
this.recordCount = recordCount;
98+
this.fileSizeInBytes = fileSizeInBytes;
99+
this.equalityFieldIds = equalityFieldIds;
100+
this.lowerBounds = lowerBounds;
101+
this.upperBounds = upperBounds;
102+
this.dataSequenceNumber = dataSequenceNumber;
103+
}
104+
105+
@Override
106+
public Long pos() {
107+
return 0L;
108+
}
109+
110+
@Override
111+
public int specId() {
112+
return 0;
113+
}
114+
115+
@Override
116+
public FileContent content() {
117+
return content;
118+
}
119+
120+
@Override
121+
public CharSequence path() {
122+
return path;
123+
}
124+
125+
@Override
126+
public FileFormat format() {
127+
return format;
128+
}
129+
130+
@Override
131+
public StructLike partition() {
132+
return null;
133+
}
134+
135+
@Override
136+
public long recordCount() {
137+
return recordCount;
138+
}
139+
140+
@Override
141+
public long fileSizeInBytes() {
142+
return fileSizeInBytes;
143+
}
144+
145+
@Override
146+
public Map<Integer, Long> columnSizes() {
147+
return null;
148+
}
149+
150+
@Override
151+
public Map<Integer, Long> valueCounts() {
152+
return null;
153+
}
154+
155+
@Override
156+
public Map<Integer, Long> nullValueCounts() {
157+
return null;
158+
}
159+
160+
@Override
161+
public Map<Integer, Long> nanValueCounts() {
162+
return null;
163+
}
164+
165+
@Override
166+
public Map<Integer, ByteBuffer> lowerBounds() {
167+
return lowerBounds;
168+
}
169+
170+
@Override
171+
public Map<Integer, ByteBuffer> upperBounds() {
172+
return upperBounds;
173+
}
174+
175+
@Override
176+
public ByteBuffer keyMetadata() {
177+
return null;
178+
}
179+
180+
@Override
181+
public List<Integer> equalityFieldIds() {
182+
return equalityFieldIds;
183+
}
184+
185+
@Override
186+
public Long dataSequenceNumber() {
187+
return dataSequenceNumber;
188+
}
189+
190+
@Override
191+
public DeleteFile copy() {
192+
return this;
193+
}
194+
195+
@Override
196+
public DeleteFile copyWithoutStats() {
197+
return this;
198+
}
199+
}
200+
}

0 commit comments

Comments
 (0)