Skip to content

Commit 5be2f64

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

3 files changed

Lines changed: 272 additions & 5 deletions

File tree

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

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

2020
namespace gluten {
2121

22+
namespace {
23+
24+
using SubstraitDeleteBoundsMap = ::substrait::ReadRel_LocalFiles_FileOrFiles::IcebergReadOptions::DeleteFile::Map;
25+
26+
std::unordered_map<int32_t, std::string> parseBounds(const SubstraitDeleteBoundsMap& bounds) {
27+
std::unordered_map<int32_t, std::string> parsed;
28+
parsed.reserve(bounds.key_values_size());
29+
30+
for (int i = 0; i < bounds.key_values_size(); ++i) {
31+
const auto& kv = bounds.key_values(i);
32+
parsed.emplace(kv.key(), kv.value());
33+
}
34+
35+
return parsed;
36+
}
37+
38+
} // namespace
39+
2240
std::shared_ptr<IcebergSplitInfo> IcebergPlanConverter::parseIcebergSplitInfo(
2341
substrait::ReadRel_LocalFiles_FileOrFiles file,
2442
std::shared_ptr<SplitInfo> splitInfo) {
@@ -70,7 +88,14 @@ std::shared_ptr<IcebergSplitInfo> IcebergPlanConverter::parseIcebergSplitInfo(
7088
break;
7189
}
7290
deletes.emplace_back(IcebergDeleteFile(
73-
fileContent, deleteFile.filepath(), format, deleteFile.recordcount(), deleteFile.filesize()));
91+
fileContent,
92+
deleteFile.filepath(),
93+
format,
94+
deleteFile.recordcount(),
95+
deleteFile.filesize(),
96+
{},
97+
parseBounds(deleteFile.lowerbounds()),
98+
parseBounds(deleteFile.upperbounds())));
7499
}
75100
icebergSplitInfo->deleteFilesVec.emplace_back(deletes);
76101
} else {

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

Lines changed: 32 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,32 @@ 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.newBuilder()
142+
.setKey(entry.getKey())
143+
.setValue(Base64.getEncoder().encodeToString(bytes))
144+
.build());
145+
}
146+
147+
return builder.build();
148+
}
121149
}

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

Lines changed: 214 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,214 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.gluten.substrait.rel;
18+
19+
import io.substrait.proto.ReadRel;
20+
import org.apache.iceberg.DeleteFile;
21+
import org.apache.iceberg.FileContent;
22+
import org.apache.iceberg.FileFormat;
23+
import org.apache.iceberg.StructLike;
24+
import org.junit.Assert;
25+
import org.junit.Test;
26+
27+
import java.nio.ByteBuffer;
28+
import java.util.Base64;
29+
import java.util.Collections;
30+
import java.util.List;
31+
import java.util.Map;
32+
33+
public class IcebergLocalFilesNodeBoundsTest {
34+
private static final int ICEBERG_POS_FIELD_ID = 2147483545;
35+
36+
@Test
37+
public void serializesDeleteFileLowerAndUpperBounds() {
38+
byte[] lowerBytes = new byte[] {1, 2, 3, 4};
39+
byte[] upperBytes = new byte[] {5, 6, 7, 8};
40+
41+
FakeDeleteFile deleteFile =
42+
new FakeDeleteFile(
43+
FileContent.POSITION_DELETES,
44+
FileFormat.PARQUET,
45+
"/tmp/delete.parquet",
46+
2L,
47+
123L,
48+
Collections.emptyList(),
49+
Collections.singletonMap(ICEBERG_POS_FIELD_ID, ByteBuffer.wrap(lowerBytes)),
50+
Collections.singletonMap(ICEBERG_POS_FIELD_ID, ByteBuffer.wrap(upperBytes)),
51+
null);
52+
IcebergLocalFilesNode node =
53+
new IcebergLocalFilesNode(
54+
0,
55+
Collections.singletonList("/tmp/data.parquet"),
56+
Collections.singletonList(0L),
57+
Collections.singletonList(100L),
58+
Collections.singletonList(Collections.emptyMap()),
59+
LocalFilesNode.ReadFileFormat.ParquetReadFormat,
60+
Collections.emptyList(),
61+
Collections.singletonList(Collections.singletonList(deleteFile)),
62+
Collections.singletonList(Collections.emptyMap()));
63+
64+
ReadRel.LocalFiles.FileOrFiles.Builder fileBuilder =
65+
ReadRel.LocalFiles.FileOrFiles.newBuilder();
66+
67+
node.processFileBuilder(fileBuilder, 0);
68+
69+
ReadRel.LocalFiles.FileOrFiles.IcebergReadOptions.DeleteFile actualDelete =
70+
fileBuilder.getIceberg().getDeleteFiles(0);
71+
72+
Assert.assertEquals(1, actualDelete.getLowerBounds().getKeyValuesCount());
73+
Assert.assertEquals(
74+
ICEBERG_POS_FIELD_ID, actualDelete.getLowerBounds().getKeyValues(0).getKey());
75+
Assert.assertEquals(
76+
Base64.getEncoder().encodeToString(lowerBytes),
77+
actualDelete.getLowerBounds().getKeyValues(0).getValue());
78+
79+
Assert.assertEquals(1, actualDelete.getUpperBounds().getKeyValuesCount());
80+
Assert.assertEquals(
81+
ICEBERG_POS_FIELD_ID, actualDelete.getUpperBounds().getKeyValues(0).getKey());
82+
Assert.assertEquals(
83+
Base64.getEncoder().encodeToString(upperBytes),
84+
actualDelete.getUpperBounds().getKeyValues(0).getValue());
85+
}
86+
87+
private static final class FakeDeleteFile implements DeleteFile {
88+
private final FileContent content;
89+
private final FileFormat format;
90+
private final String path;
91+
private final long recordCount;
92+
private final long fileSizeInBytes;
93+
private final List<Integer> equalityFieldIds;
94+
private final Map<Integer, ByteBuffer> lowerBounds;
95+
private final Map<Integer, ByteBuffer> upperBounds;
96+
private final Long dataSequenceNumber;
97+
98+
private FakeDeleteFile(
99+
FileContent content,
100+
FileFormat format,
101+
String path,
102+
long recordCount,
103+
long fileSizeInBytes,
104+
List<Integer> equalityFieldIds,
105+
Map<Integer, ByteBuffer> lowerBounds,
106+
Map<Integer, ByteBuffer> upperBounds,
107+
Long dataSequenceNumber) {
108+
this.content = content;
109+
this.format = format;
110+
this.path = path;
111+
this.recordCount = recordCount;
112+
this.fileSizeInBytes = fileSizeInBytes;
113+
this.equalityFieldIds = equalityFieldIds;
114+
this.lowerBounds = lowerBounds;
115+
this.upperBounds = upperBounds;
116+
this.dataSequenceNumber = dataSequenceNumber;
117+
}
118+
119+
@Override
120+
public Long pos() {
121+
return 0L;
122+
}
123+
124+
@Override
125+
public int specId() {
126+
return 0;
127+
}
128+
129+
@Override
130+
public FileContent content() {
131+
return content;
132+
}
133+
134+
@Override
135+
public CharSequence path() {
136+
return path;
137+
}
138+
139+
@Override
140+
public FileFormat format() {
141+
return format;
142+
}
143+
144+
@Override
145+
public StructLike partition() {
146+
return null;
147+
}
148+
149+
@Override
150+
public long recordCount() {
151+
return recordCount;
152+
}
153+
154+
@Override
155+
public long fileSizeInBytes() {
156+
return fileSizeInBytes;
157+
}
158+
159+
@Override
160+
public Map<Integer, Long> columnSizes() {
161+
return null;
162+
}
163+
164+
@Override
165+
public Map<Integer, Long> valueCounts() {
166+
return null;
167+
}
168+
169+
@Override
170+
public Map<Integer, Long> nullValueCounts() {
171+
return null;
172+
}
173+
174+
@Override
175+
public Map<Integer, Long> nanValueCounts() {
176+
return null;
177+
}
178+
179+
@Override
180+
public Map<Integer, ByteBuffer> lowerBounds() {
181+
return lowerBounds;
182+
}
183+
184+
@Override
185+
public Map<Integer, ByteBuffer> upperBounds() {
186+
return upperBounds;
187+
}
188+
189+
@Override
190+
public ByteBuffer keyMetadata() {
191+
return null;
192+
}
193+
194+
@Override
195+
public List<Integer> equalityFieldIds() {
196+
return equalityFieldIds;
197+
}
198+
199+
@Override
200+
public Long dataSequenceNumber() {
201+
return dataSequenceNumber;
202+
}
203+
204+
@Override
205+
public DeleteFile copy() {
206+
return this;
207+
}
208+
209+
@Override
210+
public DeleteFile copyWithoutStats() {
211+
return this;
212+
}
213+
}
214+
}

0 commit comments

Comments
 (0)