Skip to content

Commit 304d6ff

Browse files
raghav-regloberaghav-reglobe
authored
Parquet: add adaptive bloom filter sizing (PARQUET-2254) (#16363)
1 parent 04fd025 commit 304d6ff

4 files changed

Lines changed: 73 additions & 1 deletion

File tree

‎core/src/main/java/org/apache/iceberg/TableProperties.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -184,6 +184,10 @@ private TableProperties() {}
184184
"write.parquet.bloom-filter-max-bytes";
185185
public static final int PARQUET_BLOOM_FILTER_MAX_BYTES_DEFAULT = 1024 * 1024;
186186

187+
public static final String PARQUET_BLOOM_FILTER_ADAPTIVE_ENABLED =
188+
"write.parquet.bloom-filter-adaptive-enabled";
189+
public static final boolean PARQUET_BLOOM_FILTER_ADAPTIVE_ENABLED_DEFAULT = false;
190+
187191
public static final String PARQUET_BLOOM_FILTER_COLUMN_FPP_PREFIX =
188192
"write.parquet.bloom-filter-fpp.column.";
189193
public static final double PARQUET_BLOOM_FILTER_COLUMN_FPP_DEFAULT = 0.01;

‎docs/docs/configuration.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,7 @@ Iceberg tables support table properties to configure table behavior, like the de
5757
| write.parquet.bloom-filter-max-bytes | 1048576 (1 MB) | The maximum number of bytes for a bloom filter bitset |
5858
| write.parquet.bloom-filter-fpp.column.col1 | 0.01 | The false positive probability for a bloom filter applied to 'col1' (must > 0.0 and < 1.0) |
5959
| write.parquet.bloom-filter-ndv.column.col1 | (not set) | The expected number of distinct values for a bloom filter applied to 'col1' (must > 0) |
60+
| write.parquet.bloom-filter-adaptive-enabled | false | Use Parquet adaptive bloom filter sizing, which evaluates candidate filters and picks the smallest that satisfies the actual NDV at the configured FPP, bounded by 'write.parquet.bloom-filter-max-bytes', instead of always allocating the full max-bytes buffer. Ignored for columns where 'write.parquet.bloom-filter-ndv.column.col1' is set |
6061
| write.parquet.stats-enabled.column.col1 | (not set) | Controls whether to collect parquet column statistics for column 'col1' |
6162
| write.parquet.dict-encoding-enabled.column.col1 | (not set) | Controls whether to use parquet dictionary encoding for column 'col1' |
6263
| write.avro.compression-codec | gzip | Avro compression codec: gzip(deflate with 9 level), zstd, snappy, uncompressed |

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

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@
2727
import static org.apache.iceberg.TableProperties.DELETE_PARQUET_ROW_GROUP_CHECK_MAX_RECORD_COUNT;
2828
import static org.apache.iceberg.TableProperties.DELETE_PARQUET_ROW_GROUP_CHECK_MIN_RECORD_COUNT;
2929
import static org.apache.iceberg.TableProperties.DELETE_PARQUET_ROW_GROUP_SIZE_BYTES;
30+
import static org.apache.iceberg.TableProperties.PARQUET_BLOOM_FILTER_ADAPTIVE_ENABLED;
31+
import static org.apache.iceberg.TableProperties.PARQUET_BLOOM_FILTER_ADAPTIVE_ENABLED_DEFAULT;
3032
import static org.apache.iceberg.TableProperties.PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX;
3133
import static org.apache.iceberg.TableProperties.PARQUET_BLOOM_FILTER_COLUMN_FPP_PREFIX;
3234
import static org.apache.iceberg.TableProperties.PARQUET_BLOOM_FILTER_COLUMN_NDV_PREFIX;
@@ -470,7 +472,8 @@ public <D> FileAppender<D> build() throws IOException {
470472
.withDictionaryPageSize(dictionaryPageSize)
471473
.withMinRowCountForPageSizeCheck(rowGroupCheckMinRecordCount)
472474
.withMaxRowCountForPageSizeCheck(rowGroupCheckMaxRecordCount)
473-
.withMaxBloomFilterBytes(bloomFilterMaxBytes);
475+
.withMaxBloomFilterBytes(bloomFilterMaxBytes)
476+
.withAdaptiveBloomFilterEnabled(context.adaptiveBloomFilterEnabled());
474477

475478
setBloomFilterConfig(
476479
context,
@@ -545,6 +548,7 @@ static class Context {
545548
private final int rowGroupCheckMinRecordCount;
546549
private final int rowGroupCheckMaxRecordCount;
547550
private final int bloomFilterMaxBytes;
551+
private final boolean adaptiveBloomFilterEnabled;
548552
private final Map<String, String> columnBloomFilterFpp;
549553
private final Map<String, String> columnBloomFilterNdv;
550554
private final Map<String, String> columnBloomFilterEnabled;
@@ -564,6 +568,7 @@ private Context(
564568
int rowGroupCheckMinRecordCount,
565569
int rowGroupCheckMaxRecordCount,
566570
int bloomFilterMaxBytes,
571+
boolean adaptiveBloomFilterEnabled,
567572
Map<String, String> columnBloomFilterFpp,
568573
Map<String, String> columnBloomFilterNdv,
569574
Map<String, String> columnBloomFilterEnabled,
@@ -581,6 +586,7 @@ private Context(
581586
this.rowGroupCheckMinRecordCount = rowGroupCheckMinRecordCount;
582587
this.rowGroupCheckMaxRecordCount = rowGroupCheckMaxRecordCount;
583588
this.bloomFilterMaxBytes = bloomFilterMaxBytes;
589+
this.adaptiveBloomFilterEnabled = adaptiveBloomFilterEnabled;
584590
this.columnBloomFilterFpp = columnBloomFilterFpp;
585591
this.columnBloomFilterNdv = columnBloomFilterNdv;
586592
this.columnBloomFilterEnabled = columnBloomFilterEnabled;
@@ -646,6 +652,12 @@ static Context dataContext(Map<String, String> config) {
646652
config, PARQUET_BLOOM_FILTER_MAX_BYTES, PARQUET_BLOOM_FILTER_MAX_BYTES_DEFAULT);
647653
Preconditions.checkArgument(bloomFilterMaxBytes > 0, "bloom Filter Max Bytes must be > 0");
648654

655+
boolean adaptiveBloomFilterEnabled =
656+
PropertyUtil.propertyAsBoolean(
657+
config,
658+
PARQUET_BLOOM_FILTER_ADAPTIVE_ENABLED,
659+
PARQUET_BLOOM_FILTER_ADAPTIVE_ENABLED_DEFAULT);
660+
649661
Map<String, String> columnBloomFilterFpp =
650662
PropertyUtil.propertiesWithPrefix(config, PARQUET_BLOOM_FILTER_COLUMN_FPP_PREFIX);
651663

@@ -681,6 +693,7 @@ static Context dataContext(Map<String, String> config) {
681693
rowGroupCheckMinRecordCount,
682694
rowGroupCheckMaxRecordCount,
683695
bloomFilterMaxBytes,
696+
adaptiveBloomFilterEnabled,
684697
columnBloomFilterFpp,
685698
columnBloomFilterNdv,
686699
columnBloomFilterEnabled,
@@ -760,6 +773,7 @@ static Context deleteContext(Map<String, String> config) {
760773
rowGroupCheckMinRecordCount,
761774
rowGroupCheckMaxRecordCount,
762775
PARQUET_BLOOM_FILTER_MAX_BYTES_DEFAULT,
776+
PARQUET_BLOOM_FILTER_ADAPTIVE_ENABLED_DEFAULT,
763777
ImmutableMap.of(),
764778
ImmutableMap.of(),
765779
ImmutableMap.of(),
@@ -826,6 +840,10 @@ int bloomFilterMaxBytes() {
826840
return bloomFilterMaxBytes;
827841
}
828842

843+
boolean adaptiveBloomFilterEnabled() {
844+
return adaptiveBloomFilterEnabled;
845+
}
846+
829847
Map<String, String> columnBloomFilterFpp() {
830848
return columnBloomFilterFpp;
831849
}

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

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,9 @@
2020

2121
import static java.nio.charset.StandardCharsets.UTF_8;
2222
import static org.apache.iceberg.Files.localInput;
23+
import static org.apache.iceberg.TableProperties.PARQUET_BLOOM_FILTER_ADAPTIVE_ENABLED;
24+
import static org.apache.iceberg.TableProperties.PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX;
25+
import static org.apache.iceberg.TableProperties.PARQUET_BLOOM_FILTER_MAX_BYTES;
2326
import static org.apache.iceberg.TableProperties.PARQUET_COLUMN_STATS_ENABLED_PREFIX;
2427
import static org.apache.iceberg.TableProperties.PARQUET_DICT_ENCODING_ENABLED_COLUMN_PREFIX;
2528
import static org.apache.iceberg.TableProperties.PARQUET_ROW_GROUP_CHECK_MAX_RECORD_COUNT;
@@ -562,6 +565,52 @@ public void testAvroWriterRejectsVariantType() {
562565
.hasMessage("Avro writer does not support variant types");
563566
}
564567

568+
@Test
569+
public void adaptiveBloomFilterSizingShrinksFile() throws IOException {
570+
// when PARQUET_BLOOM_FILTER_ADAPTIVE_ENABLED is not set (the default), the writer
571+
// allocates the full PARQUET_BLOOM_FILTER_MAX_BYTES buffer (4 MiB here) regardless
572+
// of the number of distinct values written
573+
long sizeWithoutAdaptive = writeWithBloomFilter(null);
574+
assertThat(sizeWithoutAdaptive)
575+
.as("non-adaptive file should pad the bloom filter to PARQUET_BLOOM_FILTER_MAX_BYTES")
576+
.isGreaterThan(3_500_000L);
577+
578+
// with PARQUET_BLOOM_FILTER_ADAPTIVE_ENABLED, the writer picks the smallest candidate
579+
// bloom filter that satisfies the actual number of distinct values (5) at the
580+
// configured FPP
581+
long sizeWithAdaptive = writeWithBloomFilter(true);
582+
assertThat(sizeWithAdaptive)
583+
.as("adaptive file should be at least 2x smaller than the non-adaptive file")
584+
.isLessThan(sizeWithoutAdaptive / 2);
585+
}
586+
587+
private long writeWithBloomFilter(Boolean adaptiveEnabled) throws IOException {
588+
Schema schema = new Schema(required(1, "id", Types.LongType.get()));
589+
590+
ImmutableMap.Builder<String, String> propsBuilder =
591+
ImmutableMap.<String, String>builder()
592+
.put(PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX + "id", "true")
593+
.put(PARQUET_BLOOM_FILTER_MAX_BYTES, "4194304");
594+
if (adaptiveEnabled != null) {
595+
propsBuilder.put(PARQUET_BLOOM_FILTER_ADAPTIVE_ENABLED, adaptiveEnabled.toString());
596+
}
597+
598+
List<GenericData.Record> records = Lists.newArrayListWithCapacity(5);
599+
org.apache.avro.Schema avroSchema = AvroSchemaUtil.convert(schema.asStruct());
600+
for (long i = 0; i < 5; i++) {
601+
GenericData.Record record = new GenericData.Record(avroSchema);
602+
record.put("id", i);
603+
records.add(record);
604+
}
605+
606+
return write(
607+
createTempFile(temp),
608+
schema,
609+
propsBuilder.buildOrThrow(),
610+
ParquetAvroWriter::buildWriter,
611+
records.toArray(new GenericData.Record[] {}));
612+
}
613+
565614
private Pair<File, Long> generateFile(
566615
Function<MessageType, ParquetValueWriter<?>> createWriterFunc,
567616
int desiredRecordCount,

0 commit comments

Comments
 (0)