Skip to content

Commit e0afb38

Browse files
nssaliannssalian
authored
Parquet: Honor column metrics truncate length for variant shredded bounds (#17342)
* Parquet: Honor column metrics truncate length for variant shredded bounds * test cleanup * Bump testShreddedStringBoundsFull input above default truncation
1 parent 58d5c37 commit e0afb38

3 files changed

Lines changed: 205 additions & 26 deletions

File tree

‎parquet/src/main/java/org/apache/iceberg/parquet/ParquetMetrics.java‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -390,7 +390,8 @@ public Iterable<FieldMetrics<?>> variant(
390390

391391
List<ParquetVariantUtil.VariantMetrics> results =
392392
Lists.newArrayList(
393-
ParquetVariantVisitor.visit(variant, new MetricsVariantVisitor(currentPath())));
393+
ParquetVariantVisitor.visit(
394+
variant, new MetricsVariantVisitor(currentPath(), truncateLength(mode))));
394395

395396
if (results.isEmpty()) {
396397
return ImmutableList.of();
@@ -442,9 +443,11 @@ private class MetricsVariantVisitor
442443
extends ParquetVariantVisitor<Iterable<ParquetVariantUtil.VariantMetrics>> {
443444
private final Deque<String> fieldNames = Lists.newLinkedList();
444445
private final String[] basePath;
446+
private final int truncateLength;
445447

446-
private MetricsVariantVisitor(String[] basePath) {
448+
private MetricsVariantVisitor(String[] basePath, int truncateLength) {
447449
this.basePath = basePath;
450+
this.truncateLength = truncateLength;
448451
}
449452

450453
@Override
@@ -624,10 +627,11 @@ private <T> ParquetVariantUtil.VariantMetrics metrics(PrimitiveType primitive) {
624627
return null;
625628
}
626629

627-
if (lowerBound != null && upperBound != null) {
630+
if (lowerBound != null && upperBound != null && truncateLength > 0) {
628631
VariantValue lower = Variants.of(variantType, lowerBound);
629632
VariantValue upper = Variants.of(variantType, upperBound);
630-
return new ParquetVariantUtil.VariantMetrics(valueCount, nullCount, lower, upper);
633+
return new ParquetVariantUtil.VariantMetrics(
634+
valueCount, nullCount, lower, upper, truncateLength);
631635
} else {
632636
return new ParquetVariantUtil.VariantMetrics(valueCount, nullCount);
633637
}

‎parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantUtil.java‎

Lines changed: 13 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -307,11 +307,15 @@ static class VariantMetrics {
307307
}
308308

309309
VariantMetrics(
310-
long valueCount, long nullCount, VariantValue lowerBound, VariantValue upperBound) {
310+
long valueCount,
311+
long nullCount,
312+
VariantValue lowerBound,
313+
VariantValue upperBound,
314+
int truncateLength) {
311315
this.valueCount = valueCount;
312316
this.nullCount = nullCount;
313-
this.lowerBound = truncateLowerBound(lowerBound);
314-
this.upperBound = truncateUpperBound(upperBound);
317+
this.lowerBound = truncateLowerBound(lowerBound, truncateLength);
318+
this.upperBound = truncateUpperBound(upperBound, truncateLength);
315319
}
316320

317321
VariantMetrics prependFieldName(String name) {
@@ -344,30 +348,30 @@ public VariantValue upperBound() {
344348
return upperBound;
345349
}
346350

347-
private static VariantValue truncateLowerBound(VariantValue value) {
351+
private static VariantValue truncateLowerBound(VariantValue value, int length) {
348352
switch (value.type()) {
349353
case STRING:
350354
return Variants.of(
351355
PhysicalType.STRING,
352-
UnicodeUtil.truncateStringMin((String) value.asPrimitive().get(), 16));
356+
UnicodeUtil.truncateStringMin((String) value.asPrimitive().get(), length));
353357
case BINARY:
354358
return Variants.of(
355359
PhysicalType.BINARY,
356-
BinaryUtil.truncateBinaryMin((ByteBuffer) value.asPrimitive().get(), 16));
360+
BinaryUtil.truncateBinaryMin((ByteBuffer) value.asPrimitive().get(), length));
357361
default:
358362
return value;
359363
}
360364
}
361365

362-
private static VariantValue truncateUpperBound(VariantValue value) {
366+
private static VariantValue truncateUpperBound(VariantValue value, int length) {
363367
switch (value.type()) {
364368
case STRING:
365369
String truncatedString =
366-
UnicodeUtil.truncateStringMax((String) value.asPrimitive().get(), 16);
370+
UnicodeUtil.truncateStringMax((String) value.asPrimitive().get(), length);
367371
return truncatedString != null ? Variants.of(PhysicalType.STRING, truncatedString) : null;
368372
case BINARY:
369373
ByteBuffer truncatedBuffer =
370-
BinaryUtil.truncateBinaryMax((ByteBuffer) value.asPrimitive().get(), 16);
374+
BinaryUtil.truncateBinaryMax((ByteBuffer) value.asPrimitive().get(), length);
371375
return truncatedBuffer != null ? Variants.of(PhysicalType.BINARY, truncatedBuffer) : null;
372376
default:
373377
return value;

‎parquet/src/test/java/org/apache/iceberg/parquet/TestVariantMetrics.java‎

Lines changed: 184 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import org.apache.iceberg.Metrics;
3131
import org.apache.iceberg.MetricsConfig;
3232
import org.apache.iceberg.Schema;
33+
import org.apache.iceberg.TableProperties;
3334
import org.apache.iceberg.data.GenericRecord;
3435
import org.apache.iceberg.data.Record;
3536
import org.apache.iceberg.data.parquet.InternalWriter;
@@ -72,6 +73,16 @@ public class TestVariantMetrics {
7273

7374
private static final String ROOT_FIELD = "$";
7475

76+
private static final byte[] BINARY_20_BYTES = new byte[20];
77+
private static final byte[] BINARY_20_BYTES_ALL_FF = new byte[20];
78+
79+
static {
80+
for (int i = 0; i < 20; i += 1) {
81+
BINARY_20_BYTES[i] = (byte) (i + 1);
82+
BINARY_20_BYTES_ALL_FF[i] = (byte) 0xFF;
83+
}
84+
}
85+
7586
private static final VariantValue[] PRIMITIVES =
7687
new VariantValue[] {
7788
Variants.of(true),
@@ -226,11 +237,7 @@ public void testShreddedPrimitiveTypeMismatch(VariantValue value) throws IOExcep
226237
@Test
227238
public void testShreddedBinaryBoundsTruncation() throws IOException {
228239
// binary longer than the 16-byte truncation length so the bounds are truncated
229-
byte[] bytes = new byte[20];
230-
for (int i = 0; i < bytes.length; i += 1) {
231-
bytes[i] = (byte) (i + 1);
232-
}
233-
VariantValue value = Variants.of(ByteBuffer.wrap(bytes));
240+
VariantValue value = Variants.of(ByteBuffer.wrap(BINARY_20_BYTES));
234241

235242
Metrics metrics =
236243
writeParquet(
@@ -241,21 +248,17 @@ public void testShreddedBinaryBoundsTruncation() throws IOException {
241248

242249
assertThat(metrics.lowerBounds().get(2))
243250
.extracting(b -> Variant.from(b).value().asObject().get(ROOT_FIELD))
244-
.isEqualTo(Variants.of(BinaryUtil.truncateBinaryMin(ByteBuffer.wrap(bytes), 16)));
251+
.isEqualTo(Variants.of(BinaryUtil.truncateBinaryMin(ByteBuffer.wrap(BINARY_20_BYTES), 16)));
245252

246253
assertThat(metrics.upperBounds().get(2))
247254
.extracting(b -> Variant.from(b).value().asObject().get(ROOT_FIELD))
248-
.isEqualTo(Variants.of(BinaryUtil.truncateBinaryMax(ByteBuffer.wrap(bytes), 16)));
255+
.isEqualTo(Variants.of(BinaryUtil.truncateBinaryMax(ByteBuffer.wrap(BINARY_20_BYTES), 16)));
249256
}
250257

251258
@Test
252259
public void testShreddedBinaryUpperBoundOverflow() throws IOException {
253260
// an all-0xFF binary cannot be truncated up so the upper bound is omitted
254-
byte[] bytes = new byte[20];
255-
for (int i = 0; i < bytes.length; i += 1) {
256-
bytes[i] = (byte) 0xFF;
257-
}
258-
VariantValue value = Variants.of(ByteBuffer.wrap(bytes));
261+
VariantValue value = Variants.of(ByteBuffer.wrap(BINARY_20_BYTES_ALL_FF));
259262

260263
Metrics metrics =
261264
writeParquet(
@@ -266,13 +269,174 @@ public void testShreddedBinaryUpperBoundOverflow() throws IOException {
266269

267270
assertThat(metrics.lowerBounds().get(2))
268271
.extracting(b -> Variant.from(b).value().asObject().get(ROOT_FIELD))
269-
.isEqualTo(Variants.of(BinaryUtil.truncateBinaryMin(ByteBuffer.wrap(bytes), 16)));
272+
.isEqualTo(
273+
Variants.of(BinaryUtil.truncateBinaryMin(ByteBuffer.wrap(BINARY_20_BYTES_ALL_FF), 16)));
270274

271275
assertThat(metrics.upperBounds().get(2))
272276
.extracting(b -> Variant.from(b).value().asObject().get(ROOT_FIELD))
273277
.isNull();
274278
}
275279

280+
@Test
281+
public void testShreddedBinaryBoundsTruncateLength() throws IOException {
282+
// a per-column truncate(8) overrides the default 16-byte truncation
283+
VariantValue value = Variants.of(ByteBuffer.wrap(BINARY_20_BYTES));
284+
285+
MetricsConfig metricsConfig =
286+
MetricsConfig.from(
287+
ImmutableMap.of(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "var", "truncate(8)"),
288+
SCHEMA,
289+
null);
290+
291+
Metrics metrics =
292+
writeParquetWithMetricsConfig(
293+
(id, name) -> ParquetVariantUtil.toParquetSchema(value),
294+
metricsConfig,
295+
Variant.of(EMPTY, value),
296+
Variant.of(EMPTY, Variants.ofNull()),
297+
null);
298+
299+
assertThat(metrics.lowerBounds().get(2))
300+
.extracting(b -> Variant.from(b).value().asObject().get(ROOT_FIELD))
301+
.isEqualTo(Variants.of(BinaryUtil.truncateBinaryMin(ByteBuffer.wrap(BINARY_20_BYTES), 8)));
302+
303+
assertThat(metrics.upperBounds().get(2))
304+
.extracting(b -> Variant.from(b).value().asObject().get(ROOT_FIELD))
305+
.isEqualTo(Variants.of(BinaryUtil.truncateBinaryMax(ByteBuffer.wrap(BINARY_20_BYTES), 8)));
306+
}
307+
308+
@Test
309+
public void testShreddedBinaryBoundsFull() throws IOException {
310+
// full mode leaves the bounds untruncated
311+
VariantValue value = Variants.of(ByteBuffer.wrap(BINARY_20_BYTES));
312+
313+
MetricsConfig metricsConfig =
314+
MetricsConfig.from(
315+
ImmutableMap.of(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "var", "full"),
316+
SCHEMA,
317+
null);
318+
319+
Metrics metrics =
320+
writeParquetWithMetricsConfig(
321+
(id, name) -> ParquetVariantUtil.toParquetSchema(value),
322+
metricsConfig,
323+
Variant.of(EMPTY, value),
324+
Variant.of(EMPTY, Variants.ofNull()),
325+
null);
326+
327+
assertThat(metrics.lowerBounds().get(2))
328+
.extracting(b -> Variant.from(b).value().asObject().get(ROOT_FIELD))
329+
.isEqualTo(Variants.of(ByteBuffer.wrap(BINARY_20_BYTES)));
330+
331+
assertThat(metrics.upperBounds().get(2))
332+
.extracting(b -> Variant.from(b).value().asObject().get(ROOT_FIELD))
333+
.isEqualTo(Variants.of(ByteBuffer.wrap(BINARY_20_BYTES)));
334+
}
335+
336+
@Test
337+
public void testShreddedBinaryBoundsCounts() throws IOException {
338+
// counts mode drops shredded bounds
339+
VariantValue value = Variants.of(ByteBuffer.wrap(BINARY_20_BYTES));
340+
341+
MetricsConfig metricsConfig =
342+
MetricsConfig.from(
343+
ImmutableMap.of(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "var", "counts"),
344+
SCHEMA,
345+
null);
346+
347+
Metrics metrics =
348+
writeParquetWithMetricsConfig(
349+
(id, name) -> ParquetVariantUtil.toParquetSchema(value),
350+
metricsConfig,
351+
Variant.of(EMPTY, value),
352+
Variant.of(EMPTY, Variants.ofNull()),
353+
null);
354+
355+
assertThat(metrics.valueCounts()).containsKey(2);
356+
assertThat(metrics.lowerBounds()).doesNotContainKey(2);
357+
assertThat(metrics.upperBounds()).doesNotContainKey(2);
358+
}
359+
360+
@Test
361+
public void testShreddedStringBoundsTruncateLength() throws IOException {
362+
// a per-column truncate(8) overrides the default 16-char truncation
363+
VariantValue value = Variants.of("iceberg_variant");
364+
365+
MetricsConfig metricsConfig =
366+
MetricsConfig.from(
367+
ImmutableMap.of(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "var", "truncate(8)"),
368+
SCHEMA,
369+
null);
370+
371+
Metrics metrics =
372+
writeParquetWithMetricsConfig(
373+
(id, name) -> ParquetVariantUtil.toParquetSchema(value),
374+
metricsConfig,
375+
Variant.of(EMPTY, value),
376+
Variant.of(EMPTY, Variants.ofNull()),
377+
null);
378+
379+
assertThat(metrics.lowerBounds().get(2))
380+
.extracting(b -> Variant.from(b).value().asObject().get(ROOT_FIELD))
381+
.isEqualTo(Variants.of(UnicodeUtil.truncateStringMin("iceberg_variant", 8)));
382+
383+
assertThat(metrics.upperBounds().get(2))
384+
.extracting(b -> Variant.from(b).value().asObject().get(ROOT_FIELD))
385+
.isEqualTo(Variants.of(UnicodeUtil.truncateStringMax("iceberg_variant", 8)));
386+
}
387+
388+
@Test
389+
public void testShreddedStringBoundsFull() throws IOException {
390+
// full mode leaves the string bound untruncated
391+
VariantValue value = Variants.of("iceberg_variant_full");
392+
393+
MetricsConfig metricsConfig =
394+
MetricsConfig.from(
395+
ImmutableMap.of(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "var", "full"),
396+
SCHEMA,
397+
null);
398+
399+
Metrics metrics =
400+
writeParquetWithMetricsConfig(
401+
(id, name) -> ParquetVariantUtil.toParquetSchema(value),
402+
metricsConfig,
403+
Variant.of(EMPTY, value),
404+
Variant.of(EMPTY, Variants.ofNull()),
405+
null);
406+
407+
assertThat(metrics.lowerBounds().get(2))
408+
.extracting(b -> Variant.from(b).value().asObject().get(ROOT_FIELD))
409+
.isEqualTo(Variants.of("iceberg_variant_full"));
410+
411+
assertThat(metrics.upperBounds().get(2))
412+
.extracting(b -> Variant.from(b).value().asObject().get(ROOT_FIELD))
413+
.isEqualTo(Variants.of("iceberg_variant_full"));
414+
}
415+
416+
@Test
417+
public void testShreddedStringBoundsCounts() throws IOException {
418+
// counts mode must not truncate the shredded string bound: truncate length 0 would throw
419+
VariantValue value = Variants.of("iceberg_variant");
420+
421+
MetricsConfig metricsConfig =
422+
MetricsConfig.from(
423+
ImmutableMap.of(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "var", "counts"),
424+
SCHEMA,
425+
null);
426+
427+
Metrics metrics =
428+
writeParquetWithMetricsConfig(
429+
(id, name) -> ParquetVariantUtil.toParquetSchema(value),
430+
metricsConfig,
431+
Variant.of(EMPTY, value),
432+
Variant.of(EMPTY, Variants.ofNull()),
433+
null);
434+
435+
assertThat(metrics.valueCounts()).containsKey(2);
436+
assertThat(metrics.lowerBounds()).doesNotContainKey(2);
437+
assertThat(metrics.upperBounds()).doesNotContainKey(2);
438+
}
439+
276440
@Test
277441
public void testVariantFloatNaN() throws IOException {
278442
// NaN values are not counted because there is no ID for FieldMetrics
@@ -578,13 +742,20 @@ public void testShreddedValueColumnWithEmptyStats() throws IOException {
578742

579743
private Metrics writeParquet(VariantShreddingFunction shredding, Variant... variants)
580744
throws IOException {
745+
return writeParquetWithMetricsConfig(shredding, MetricsConfig.getDefault(), variants);
746+
}
747+
748+
private Metrics writeParquetWithMetricsConfig(
749+
VariantShreddingFunction shredding, MetricsConfig metricsConfig, Variant... variants)
750+
throws IOException {
581751
OutputFile out = new InMemoryOutputFile();
582752
GenericRecord record = GenericRecord.create(SCHEMA);
583753

584754
FileAppender<Record> writer =
585755
Parquet.write(out)
586756
.schema(SCHEMA)
587757
.variantShreddingFunc(shredding)
758+
.metricsConfig(metricsConfig)
588759
.createWriterFunc(fileSchema -> InternalWriter.create(SCHEMA.asStruct(), fileSchema))
589760
.build();
590761

0 commit comments

Comments
 (0)