Skip to content

Commit ef5f89f

Browse files
authored
[Iceberg] Fix manifest bounds being padded with trailing 0x00 bytes (#38580)
* [Iceberg] Fix manifest bounds being padded with trailing 0x00 bytes * update changes.md * test: encode bound ByteBuffers via Conversions.toByteBuffer * trigger build
1 parent 63501f3 commit ef5f89f

3 files changed

Lines changed: 79 additions & 1 deletion

File tree

CHANGES.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,7 @@
9898

9999
* Fixed BigQueryEnrichmentHandler batch mode dropping earlier requests when multiple requests share the same enrichment key (Python) ([#38035](https://github.com/apache/beam/issues/38035)).
100100
* Added `max_batch_duration_secs` passthrough support in Python Enrichment BigQuery and CloudSQL handlers so batching duration can be forwarded to `BatchElements` ([#38243](https://github.com/apache/beam/issues/38243)).
101+
* Fixed IcebergIO writing manifest column bounds padded with trailing `0x00` bytes, which broke equality predicate pushdown in some query engines (Java) ([#38580](https://github.com/apache/beam/issues/38580)).
101102

102103
## Security Fixes
103104

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableDataFile.java

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -205,11 +205,23 @@ DataFile createDataFile(Map<Integer, PartitionSpec> partitionSpecs) {
205205
}
206206
Map<Integer, byte[]> output = new HashMap<>(input.size());
207207
for (Map.Entry<Integer, ByteBuffer> e : input.entrySet()) {
208-
output.put(e.getKey(), e.getValue().array());
208+
output.put(e.getKey(), toByteArray(e.getValue()));
209209
}
210210
return output;
211211
}
212212

213+
// Copy only [position, limit). ByteBuffer.array() returns the full backing
214+
// array, which is sometimes larger than the buffer's content (e.g. trailing
215+
// 0x00 bytes). Leaking those into manifest bounds shifts the lower bound
216+
// above the real min and breaks equality predicate pushdown in some query
217+
// engines.
218+
private static byte[] toByteArray(ByteBuffer buf) {
219+
ByteBuffer view = buf.duplicate();
220+
byte[] bytes = new byte[view.remaining()];
221+
view.get(bytes);
222+
return bytes;
223+
}
224+
213225
private static @Nullable Map<Integer, ByteBuffer> toByteBufferMap(
214226
@Nullable Map<Integer, byte[]> input) {
215227
if (input == null) {

sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableDataFileTest.java

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,13 +17,27 @@
1717
*/
1818
package org.apache.beam.sdk.io.iceberg;
1919

20+
import static org.junit.Assert.assertArrayEquals;
21+
import static org.junit.Assert.assertEquals;
22+
2023
import java.lang.reflect.Method;
24+
import java.nio.ByteBuffer;
25+
import java.nio.charset.StandardCharsets;
2126
import java.util.ArrayList;
2227
import java.util.Arrays;
28+
import java.util.HashMap;
2329
import java.util.List;
30+
import java.util.Map;
2431
import java.util.Set;
2532
import java.util.stream.Collectors;
2633
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet;
34+
import org.apache.iceberg.DataFile;
35+
import org.apache.iceberg.DataFiles;
36+
import org.apache.iceberg.FileFormat;
37+
import org.apache.iceberg.Metrics;
38+
import org.apache.iceberg.PartitionSpec;
39+
import org.apache.iceberg.types.Conversions;
40+
import org.apache.iceberg.types.Types;
2741
import org.junit.Test;
2842

2943
/**
@@ -73,4 +87,55 @@ public void testFieldsInEqualsMethodInSyncWithGetterFields() {
7387
+ "to this test class's FIELDS_SET.");
7488
}
7589
}
90+
91+
/**
92+
* Bounds with {@code capacity > limit} must be copied by {@code [position, limit)}, not by {@link
93+
* ByteBuffer#array()}. Otherwise trailing 0x00 bytes leak into the manifest bounds and break
94+
* equality predicate pushdown in some query engines.
95+
*/
96+
@Test
97+
public void testBoundByteBufferIsCopiedByLimitNotBackingArrayLength() {
98+
// Encode bounds the same way iceberg-parquet does in the wild — via
99+
// Conversions.toByteBuffer(STRING, value). For UTF-8 strings of 10+
100+
// characters the underlying JDK CharsetEncoder over-allocates by ~10%
101+
// and flips, producing a ByteBuffer with capacity > limit.
102+
int columnId = 3;
103+
String lowerValue = "lower_bound_str";
104+
String upperValue = "upper_bound_str";
105+
byte[] expectedLower = lowerValue.getBytes(StandardCharsets.UTF_8);
106+
byte[] expectedUpper = upperValue.getBytes(StandardCharsets.UTF_8);
107+
108+
ByteBuffer lower = Conversions.toByteBuffer(Types.StringType.get(), lowerValue);
109+
ByteBuffer upper = Conversions.toByteBuffer(Types.StringType.get(), upperValue);
110+
111+
Map<Integer, ByteBuffer> lowerBounds = new HashMap<>();
112+
lowerBounds.put(columnId, lower);
113+
Map<Integer, ByteBuffer> upperBounds = new HashMap<>();
114+
upperBounds.put(columnId, upper);
115+
116+
Metrics metrics = new Metrics(1L, null, null, null, null, lowerBounds, upperBounds);
117+
118+
DataFile dataFile =
119+
DataFiles.builder(PartitionSpec.unpartitioned())
120+
.withFormat(FileFormat.PARQUET)
121+
.withPath("gs://test-bucket/data/test-file.parquet")
122+
.withFileSizeInBytes(1024L)
123+
.withMetrics(metrics)
124+
.build();
125+
126+
SerializableDataFile serialized = SerializableDataFile.from(dataFile, "");
127+
128+
byte[] serializedLower = serialized.getLowerBounds().get(columnId);
129+
byte[] serializedUpper = serialized.getUpperBounds().get(columnId);
130+
assertEquals(
131+
"lower bound length must match content, not backing array",
132+
expectedLower.length,
133+
serializedLower.length);
134+
assertEquals(
135+
"upper bound length must match content, not backing array",
136+
expectedUpper.length,
137+
serializedUpper.length);
138+
assertArrayEquals(expectedLower, serializedLower);
139+
assertArrayEquals(expectedUpper, serializedUpper);
140+
}
76141
}

0 commit comments

Comments
 (0)