From 7ab169e75cad4dd965bd950aadd20f6b54b3a3b9 Mon Sep 17 00:00:00 2001 From: Denys Kuzmenko Date: Sun, 30 Aug 2026 13:05:03 +0300 Subject: [PATCH] HIVE-29737: Prune Parquet row groups with bloom filters in the vectorized reader Parquet files can carry bloom filters, and Iceberg tables ask for them through the write.parquet.bloom-filter-enabled.column.* table properties, but the vectorized reader never consulted them. VectorizedParquetRecordReader builds its reader from the original job conf, where ParquetInputFormat.getFilter yields a NoOpFilter. Row groups are therefore pruned only by ParquetRecordReaderBase#getSplit, which asked RowGroupFilter for statistics. A value inside a column's min/max range but absent from the data was never skipped. The non-vectorized reader was unaffected: it builds its context from the conf that carries the pushed down predicate, so parquet applies all filter levels there already. This sits under MapredParquetInputFormat's vectorized branch as much as under Iceberg's reader, so it applies to any Parquet table read vectorized, whoever wrote the file. Run the bloom filter level over the row groups that survive statistics. Bloom filters live in the data file rather than the footer, so this needs an open reader, and it is opened only when a surviving row group carries a filter for a column the predicate can prune on. A bloom filter only proves a value absent, so that is the columns under equality and set membership, which is what BloomFilterImpl reads; an OR contributes only when both of its sides do. The reader is built without a record filter, since parquet would otherwise repeat the whole filtering pass in the constructor. Parquet's own parquet.filter.bloom.enabled turns the level off. Under LLAP the reader otherwise reaches past the cache for those bytes: only the footer was cached, so every reader re-read the filters from the file. The column data cache cannot hold them either, as it is indexed by column chunk while bloom filters sit outside every chunk, and its reads must start at a registered chunk. Cache them the way ORC caches the metadata its bloom indexes ride in. MetadataCache gains get/putParquetBloomFilters, keyed by file and by the offset the footer records for the column chunk. Keying per filter rather than per file keeps an entry describing exactly the bytes it holds, and keeps its size to one filter: parquet sizes these from its own defaults, roughly a megabyte per column per row group, so a file wide entry would pull every row group's filters on the first miss into a cache that otherwise holds footers of a few hundred bytes. A reader asks for the filters of its own row groups in one call, which opens the file once for whatever is missing, and presents them to parquet through ParquetFilterDataFromCache, a sparse InputFile serving cached ranges at the offsets the file stores them at. The buffers stay locked until the filtering pass is done, as the cache is what the allocations for the remaining filters evict from. Without LLAP, or when a column chunk records no bloom filter length, the plain reader reads the file as before. TestParquetRowGroupFilter covers the pruning rules over plain Parquet files, including the predicate shapes that must not open the file, and TestParquetFilterDataFromCache covers serving several cached ranges at their own offsets. TestMetadataCache covers evicting one filter while the file's footer and its other filters stay cached. iceberg_parquet_bloom_filter.q covers the write path and end to end pruning, and llap_iceberg_bloom_filter.q covers the LLAP path over both a single row group and a file of several, where hive.llap.io.cache.only makes the repeated queries fail unless the filters come from the cache. --- .../TestHiveIcebergParquetBloomFilter.java | 124 +++++++ .../positive/iceberg_parquet_bloom_filter.q | 29 ++ .../positive/llap_iceberg_bloom_filter.q | 56 ++++ .../iceberg_parquet_bloom_filter.q.out | 136 ++++++++ .../llap/llap_iceberg_bloom_filter.q.out | 159 +++++++++ .../resources/testconfiguration.properties | 2 + .../hadoop/hive/llap/io/api/LlapIo.java | 16 + .../hive/llap/io/api/impl/LlapIoImpl.java | 57 +++- .../hive/llap/io/metadata/MetadataCache.java | 59 +++- .../hive/llap/cache/TestMetadataCache.java | 77 +++++ .../TestParquetBloomFilterBufferRelease.java | 180 +++++++++++ .../impl/TestLlapParquetBloomFilterCache.java | 306 ++++++++++++++++++ .../io/parquet/ParquetRecordReaderBase.java | 11 +- .../vector/ParquetFilterDataFromCache.java | 198 ++++++++++++ .../vector/VectorizedParquetRecordReader.java | 181 +++++++++++ .../io/parquet/AbstractTestParquetDirect.java | 46 ++- .../io/parquet/TestParquetRowGroupFilter.java | 235 ++++++++++++++ .../TestParquetFilterDataFromCache.java | 171 ++++++++++ .../TestVectorizedParquetBloomFilters.java | 145 +++++++++ 19 files changed, 2165 insertions(+), 23 deletions(-) create mode 100644 iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergParquetBloomFilter.java create mode 100644 iceberg/iceberg-handler/src/test/queries/positive/iceberg_parquet_bloom_filter.q create mode 100644 iceberg/iceberg-handler/src/test/queries/positive/llap_iceberg_bloom_filter.q create mode 100644 iceberg/iceberg-handler/src/test/results/positive/iceberg_parquet_bloom_filter.q.out create mode 100644 iceberg/iceberg-handler/src/test/results/positive/llap/llap_iceberg_bloom_filter.q.out create mode 100644 llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestParquetBloomFilterBufferRelease.java create mode 100644 llap-server/src/test/org/apache/hadoop/hive/llap/io/api/impl/TestLlapParquetBloomFilterCache.java create mode 100644 ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/ParquetFilterDataFromCache.java create mode 100644 ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/TestParquetFilterDataFromCache.java create mode 100644 ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/TestVectorizedParquetBloomFilters.java diff --git a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergParquetBloomFilter.java b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergParquetBloomFilter.java new file mode 100644 index 000000000000..f500e9a5871b --- /dev/null +++ b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergParquetBloomFilter.java @@ -0,0 +1,124 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iceberg.mr.hive; + +import java.io.IOException; +import java.util.Collection; +import java.util.List; +import org.apache.hadoop.fs.Path; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableProperties; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.mr.hive.test.TestTables.TestTableType; +import org.apache.iceberg.parquet.ParquetBloomRowGroupFilter; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.types.Types; +import org.apache.parquet.column.values.bloomfilter.BloomFilter; +import org.apache.parquet.hadoop.BloomFilterReader; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.metadata.BlockMetaData; +import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; +import org.apache.parquet.hadoop.util.HadoopInputFile; +import org.apache.parquet.schema.MessageType; +import org.junit.Assert; +import org.junit.Test; +import org.junit.runners.Parameterized.Parameters; + +import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.apache.iceberg.types.Types.NestedField.required; + +/** + * Verifies that Hive inserts into Parquet Iceberg tables honor the + * {@code write.parquet.bloom-filter-enabled.column.*} table properties: the written files must contain working + * bloom filters that Iceberg's read-side row group filter can prune on. + */ +public class TestHiveIcebergParquetBloomFilter extends HiveIcebergStorageHandlerWithEngineBase { + + private static final long PRESENT_ID = 42L; + private static final long ABSENT_ID = 12345678L; + + @Parameters(name = "fileFormat={0}, catalog={1}, isVectorized={2}, formatVersion={3}") + public static Collection parameters() { + return HiveIcebergStorageHandlerWithEngineBase.getParameters(p -> + p.fileFormat() == FileFormat.PARQUET && p.testTableType() == TestTableType.HIVE_CATALOG && + p.formatVersion() == 2); + } + + @Test + public void testBloomFilterWrittenByHiveInsert() throws IOException { + Schema schema = new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "name", Types.StringType.get())); + + testTables.createTable(shell, "bloom_test", schema, fileFormat, ImmutableList.of(), formatVersion, + ImmutableMap.of( + TableProperties.PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX + "id", "true", + TableProperties.PARQUET_BLOOM_FILTER_COLUMN_FPP_PREFIX + "id", "0.01")); + + shell.executeStatement("INSERT INTO bloom_test VALUES (1, 'a'), (" + PRESENT_ID + ", 'b'), (100, 'c')"); + + Table table = testTables.loadTable(TableIdentifier.of("default", "bloom_test")); + List dataFiles = Lists.newArrayList(table.currentSnapshot().addedDataFiles(table.io())); + Assert.assertEquals(1, dataFiles.size()); + + HadoopInputFile inputFile = HadoopInputFile.fromPath(new Path(dataFiles.get(0).location()), shell.getHiveConf()); + try (ParquetFileReader reader = ParquetFileReader.open(inputFile)) { + MessageType fileSchema = reader.getFooter().getFileMetaData().getSchema(); + List rowGroups = reader.getFooter().getBlocks(); + Assert.assertFalse(rowGroups.isEmpty()); + + for (BlockMetaData rowGroup : rowGroups) { + BloomFilterReader bloomReader = reader.getBloomFilterDataReader(rowGroup); + + BloomFilter bloom = bloomReader.readBloomFilter(columnChunk(rowGroup, "id")); + Assert.assertNotNull("Bloom filter should be written for the enabled column", bloom); + Assert.assertTrue(bloom.findHash(bloom.hash(PRESENT_ID))); + Assert.assertFalse(bloom.findHash(bloom.hash(ABSENT_ID))); + + Assert.assertNull("Bloom filter should not be written for a column where it was not enabled", + bloomReader.readBloomFilter(columnChunk(rowGroup, "name"))); + + Assert.assertTrue(new ParquetBloomRowGroupFilter(schema, Expressions.equal("id", PRESENT_ID)) + .shouldRead(fileSchema, rowGroup, bloomReader)); + Assert.assertFalse("Row group should be prunable for a value not in the bloom filter", + new ParquetBloomRowGroupFilter(schema, Expressions.equal("id", ABSENT_ID)) + .shouldRead(fileSchema, rowGroup, bloomReader)); + } + } + + List rows = shell.executeStatement("SELECT name FROM bloom_test WHERE id = " + PRESENT_ID); + Assert.assertEquals(1, rows.size()); + Assert.assertEquals("b", rows.get(0)[0]); + Assert.assertTrue(shell.executeStatement("SELECT * FROM bloom_test WHERE id = " + ABSENT_ID).isEmpty()); + } + + private static ColumnChunkMetaData columnChunk(BlockMetaData rowGroup, String columnName) { + return rowGroup.getColumns().stream() + .filter(column -> column.getPath().toDotString().equals(columnName)) + .findAny() + .orElseThrow(); + } +} diff --git a/iceberg/iceberg-handler/src/test/queries/positive/iceberg_parquet_bloom_filter.q b/iceberg/iceberg-handler/src/test/queries/positive/iceberg_parquet_bloom_filter.q new file mode 100644 index 000000000000..bc2fb20c77a2 --- /dev/null +++ b/iceberg/iceberg-handler/src/test/queries/positive/iceberg_parquet_bloom_filter.q @@ -0,0 +1,29 @@ +-- Mask random uuid +--! qt:replace:/(\s+'uuid'=')\S+('\s*)/$1#Masked#$2/ + +-- Parquet bloom filter write properties on Iceberg tables: verifies the properties are accepted, +-- survive in HMS, and inserts/point lookups work with bloom filters enabled. +-- Bloom filter presence in the data files is asserted by TestHiveIcebergParquetBloomFilter. + +drop table if exists tbl_bloom; +create external table tbl_bloom(id bigint, name string) stored by iceberg stored as parquet +tblproperties ('format-version'='2', + 'write.parquet.bloom-filter-enabled.column.id'='true', + 'write.parquet.bloom-filter-fpp.column.id'='0.05'); + +show create table tbl_bloom; + +insert into tbl_bloom values (1, 'one'), (42, 'answer'), (100, 'hundred'), (12345678, 'big'); + +select name from tbl_bloom where id = 42; +select count(*) from tbl_bloom where id = 43; +select * from tbl_bloom order by id; + +-- enable bloom filter on another column, subsequent writes pick it up +alter table tbl_bloom set tblproperties ('write.parquet.bloom-filter-enabled.column.name'='true'); +insert into tbl_bloom values (200, 'two hundred'); + +select id from tbl_bloom where name = 'two hundred'; +select count(*) from tbl_bloom; + +drop table tbl_bloom; diff --git a/iceberg/iceberg-handler/src/test/queries/positive/llap_iceberg_bloom_filter.q b/iceberg/iceberg-handler/src/test/queries/positive/llap_iceberg_bloom_filter.q new file mode 100644 index 000000000000..8882ee0bf18a --- /dev/null +++ b/iceberg/iceberg-handler/src/test/queries/positive/llap_iceberg_bloom_filter.q @@ -0,0 +1,56 @@ +-- Parquet bloom filter pruning under vectorized LLAP execution, where the filters are served from the +-- LLAP metadata cache after the first read of a file. +set hive.llap.io.enabled=true; +set hive.vectorized.execution.enabled=true; + +DROP TABLE IF EXISTS llap_bloom_parquet PURGE; + +CREATE EXTERNAL TABLE llap_bloom_parquet (id bigint, name string) +STORED BY ICEBERG STORED AS PARQUET +TBLPROPERTIES ('format-version'='2', 'write.parquet.bloom-filter-enabled.column.id'='true'); + +INSERT INTO llap_bloom_parquet VALUES +(2, 'two'), (4, 'four'), (6, 'six'), (8, 'eight'), (10, 'ten'); + +-- absent from the bloom filter but inside the min/max range, so only the bloom filter can prune it; +-- this first read fills the cache +SELECT count(*) FROM llap_bloom_parquet WHERE id = 5; + +-- under cache.only the reader may not fall back to the file, so this answers only if the bloom filter +-- itself came from the cache +set hive.llap.io.cache.only=true; +SELECT count(*) FROM llap_bloom_parquet WHERE id = 5; +set hive.llap.io.cache.only=false; + +-- a value the bloom filter does contain must survive pruning; count(*) keeps this off the fetch-task +-- path, which runs in the client JVM and would never reach the LLAP reader +SELECT count(*) FROM llap_bloom_parquet WHERE id = 6; + +DROP TABLE llap_bloom_parquet PURGE; + +-- A file of several row groups, where statistics leave a different row group standing per predicate. Each +-- filter is cached under its own offset, so serving one never depends on which query cached it. +DROP TABLE IF EXISTS llap_bloom_multi PURGE; + +CREATE EXTERNAL TABLE llap_bloom_multi (id bigint, name string) +STORED BY ICEBERG STORED AS PARQUET +TBLPROPERTIES ('format-version'='2', 'write.parquet.bloom-filter-enabled.column.id'='true', + 'write.parquet.bloom-filter-max-bytes'='1024', 'write.parquet.row-group-size-bytes'='1024'); + +INSERT INTO llap_bloom_multi +SELECT pos * 2, concat('n', pos) FROM (SELECT 1) x LATERAL VIEW posexplode(split(space(399), ' ')) e AS pos, val; + +-- odd ids are absent everywhere, and each lands in a different row group +SELECT count(*) FROM llap_bloom_multi WHERE id = 51; +SELECT count(*) FROM llap_bloom_multi WHERE id = 651; + +set hive.llap.io.cache.only=true; +SELECT count(*) FROM llap_bloom_multi WHERE id = 51; +SELECT count(*) FROM llap_bloom_multi WHERE id = 651; +set hive.llap.io.cache.only=false; + +-- even ids are present, and must survive pruning against filters served from the cache +SELECT count(*) FROM llap_bloom_multi WHERE id = 4; +SELECT count(*) FROM llap_bloom_multi WHERE id = 700; + +DROP TABLE llap_bloom_multi PURGE; diff --git a/iceberg/iceberg-handler/src/test/results/positive/iceberg_parquet_bloom_filter.q.out b/iceberg/iceberg-handler/src/test/results/positive/iceberg_parquet_bloom_filter.q.out new file mode 100644 index 000000000000..7e30c536ddb7 --- /dev/null +++ b/iceberg/iceberg-handler/src/test/results/positive/iceberg_parquet_bloom_filter.q.out @@ -0,0 +1,136 @@ +PREHOOK: query: drop table if exists tbl_bloom +PREHOOK: type: DROPTABLE +PREHOOK: Output: database:default +POSTHOOK: query: drop table if exists tbl_bloom +POSTHOOK: type: DROPTABLE +POSTHOOK: Output: database:default +PREHOOK: query: create external table tbl_bloom(id bigint, name string) stored by iceberg stored as parquet +tblproperties ('format-version'='2', + 'write.parquet.bloom-filter-enabled.column.id'='true', + 'write.parquet.bloom-filter-fpp.column.id'='0.05') +PREHOOK: type: CREATETABLE +PREHOOK: Output: database:default +PREHOOK: Output: default@tbl_bloom +POSTHOOK: query: create external table tbl_bloom(id bigint, name string) stored by iceberg stored as parquet +tblproperties ('format-version'='2', + 'write.parquet.bloom-filter-enabled.column.id'='true', + 'write.parquet.bloom-filter-fpp.column.id'='0.05') +POSTHOOK: type: CREATETABLE +POSTHOOK: Output: database:default +POSTHOOK: Output: default@tbl_bloom +PREHOOK: query: show create table tbl_bloom +PREHOOK: type: SHOW_CREATETABLE +PREHOOK: Input: default@tbl_bloom +POSTHOOK: query: show create table tbl_bloom +POSTHOOK: type: SHOW_CREATETABLE +POSTHOOK: Input: default@tbl_bloom +CREATE EXTERNAL TABLE `tbl_bloom`( + `id` bigint, + `name` string) +ROW FORMAT SERDE + 'org.apache.iceberg.mr.hive.HiveIcebergSerDe' +STORED BY + 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler' + +LOCATION + 'hdfs://### HDFS PATH ###' +TBLPROPERTIES ( + 'bucketing_version'='2', + 'current-schema'='{"type":"struct","schema-id":0,"fields":[{"id":1,"name":"id","required":false,"type":"long"},{"id":2,"name":"name","required":false,"type":"string"}]}', + 'format-version'='2', + 'metadata_location'='hdfs://### HDFS PATH ###', + 'parquet.compression'='zstd', + 'serialization.format'='1', + 'snapshot-count'='0', + 'table_type'='ICEBERG', +#### A masked pattern was here #### + 'uuid'='#Masked#', + 'write.delete.mode'='merge-on-read', + 'write.format.default'='parquet', + 'write.merge.mode'='merge-on-read', + 'write.metadata.delete-after-commit.enabled'='true', + 'write.parquet.bloom-filter-enabled.column.id'='true', + 'write.parquet.bloom-filter-fpp.column.id'='0.05', + 'write.update.mode'='merge-on-read') +PREHOOK: query: insert into tbl_bloom values (1, 'one'), (42, 'answer'), (100, 'hundred'), (12345678, 'big') +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@tbl_bloom +POSTHOOK: query: insert into tbl_bloom values (1, 'one'), (42, 'answer'), (100, 'hundred'), (12345678, 'big') +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@tbl_bloom +PREHOOK: query: select name from tbl_bloom where id = 42 +PREHOOK: type: QUERY +PREHOOK: Input: default@tbl_bloom +PREHOOK: Output: hdfs://### HDFS PATH ### +POSTHOOK: query: select name from tbl_bloom where id = 42 +POSTHOOK: type: QUERY +POSTHOOK: Input: default@tbl_bloom +POSTHOOK: Output: hdfs://### HDFS PATH ### +answer +PREHOOK: query: select count(*) from tbl_bloom where id = 43 +PREHOOK: type: QUERY +PREHOOK: Input: default@tbl_bloom +PREHOOK: Output: hdfs://### HDFS PATH ### +POSTHOOK: query: select count(*) from tbl_bloom where id = 43 +POSTHOOK: type: QUERY +POSTHOOK: Input: default@tbl_bloom +POSTHOOK: Output: hdfs://### HDFS PATH ### +0 +PREHOOK: query: select * from tbl_bloom order by id +PREHOOK: type: QUERY +PREHOOK: Input: default@tbl_bloom +PREHOOK: Output: hdfs://### HDFS PATH ### +POSTHOOK: query: select * from tbl_bloom order by id +POSTHOOK: type: QUERY +POSTHOOK: Input: default@tbl_bloom +POSTHOOK: Output: hdfs://### HDFS PATH ### +1 one +42 answer +100 hundred +12345678 big +PREHOOK: query: alter table tbl_bloom set tblproperties ('write.parquet.bloom-filter-enabled.column.name'='true') +PREHOOK: type: ALTERTABLE_PROPERTIES +PREHOOK: Input: default@tbl_bloom +PREHOOK: Output: default@tbl_bloom +POSTHOOK: query: alter table tbl_bloom set tblproperties ('write.parquet.bloom-filter-enabled.column.name'='true') +POSTHOOK: type: ALTERTABLE_PROPERTIES +POSTHOOK: Input: default@tbl_bloom +POSTHOOK: Output: default@tbl_bloom +PREHOOK: query: insert into tbl_bloom values (200, 'two hundred') +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@tbl_bloom +POSTHOOK: query: insert into tbl_bloom values (200, 'two hundred') +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@tbl_bloom +PREHOOK: query: select id from tbl_bloom where name = 'two hundred' +PREHOOK: type: QUERY +PREHOOK: Input: default@tbl_bloom +PREHOOK: Output: hdfs://### HDFS PATH ### +POSTHOOK: query: select id from tbl_bloom where name = 'two hundred' +POSTHOOK: type: QUERY +POSTHOOK: Input: default@tbl_bloom +POSTHOOK: Output: hdfs://### HDFS PATH ### +200 +PREHOOK: query: select count(*) from tbl_bloom +PREHOOK: type: QUERY +PREHOOK: Input: default@tbl_bloom +PREHOOK: Output: hdfs://### HDFS PATH ### +POSTHOOK: query: select count(*) from tbl_bloom +POSTHOOK: type: QUERY +POSTHOOK: Input: default@tbl_bloom +POSTHOOK: Output: hdfs://### HDFS PATH ### +5 +PREHOOK: query: drop table tbl_bloom +PREHOOK: type: DROPTABLE +PREHOOK: Input: default@tbl_bloom +PREHOOK: Output: database:default +PREHOOK: Output: default@tbl_bloom +POSTHOOK: query: drop table tbl_bloom +POSTHOOK: type: DROPTABLE +POSTHOOK: Input: default@tbl_bloom +POSTHOOK: Output: database:default +POSTHOOK: Output: default@tbl_bloom diff --git a/iceberg/iceberg-handler/src/test/results/positive/llap/llap_iceberg_bloom_filter.q.out b/iceberg/iceberg-handler/src/test/results/positive/llap/llap_iceberg_bloom_filter.q.out new file mode 100644 index 000000000000..db9ed08ab422 --- /dev/null +++ b/iceberg/iceberg-handler/src/test/results/positive/llap/llap_iceberg_bloom_filter.q.out @@ -0,0 +1,159 @@ +PREHOOK: query: DROP TABLE IF EXISTS llap_bloom_parquet PURGE +PREHOOK: type: DROPTABLE +PREHOOK: Output: database:default +POSTHOOK: query: DROP TABLE IF EXISTS llap_bloom_parquet PURGE +POSTHOOK: type: DROPTABLE +POSTHOOK: Output: database:default +PREHOOK: query: CREATE EXTERNAL TABLE llap_bloom_parquet (id bigint, name string) +STORED BY ICEBERG STORED AS PARQUET +TBLPROPERTIES ('format-version'='2', 'write.parquet.bloom-filter-enabled.column.id'='true') +PREHOOK: type: CREATETABLE +PREHOOK: Output: database:default +PREHOOK: Output: default@llap_bloom_parquet +POSTHOOK: query: CREATE EXTERNAL TABLE llap_bloom_parquet (id bigint, name string) +STORED BY ICEBERG STORED AS PARQUET +TBLPROPERTIES ('format-version'='2', 'write.parquet.bloom-filter-enabled.column.id'='true') +POSTHOOK: type: CREATETABLE +POSTHOOK: Output: database:default +POSTHOOK: Output: default@llap_bloom_parquet +PREHOOK: query: INSERT INTO llap_bloom_parquet VALUES +(2, 'two'), (4, 'four'), (6, 'six'), (8, 'eight'), (10, 'ten') +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@llap_bloom_parquet +POSTHOOK: query: INSERT INTO llap_bloom_parquet VALUES +(2, 'two'), (4, 'four'), (6, 'six'), (8, 'eight'), (10, 'ten') +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@llap_bloom_parquet +PREHOOK: query: SELECT count(*) FROM llap_bloom_parquet WHERE id = 5 +PREHOOK: type: QUERY +PREHOOK: Input: default@llap_bloom_parquet +#### A masked pattern was here #### +POSTHOOK: query: SELECT count(*) FROM llap_bloom_parquet WHERE id = 5 +POSTHOOK: type: QUERY +POSTHOOK: Input: default@llap_bloom_parquet +#### A masked pattern was here #### +0 +PREHOOK: query: SELECT count(*) FROM llap_bloom_parquet WHERE id = 5 +PREHOOK: type: QUERY +PREHOOK: Input: default@llap_bloom_parquet +#### A masked pattern was here #### +POSTHOOK: query: SELECT count(*) FROM llap_bloom_parquet WHERE id = 5 +POSTHOOK: type: QUERY +POSTHOOK: Input: default@llap_bloom_parquet +#### A masked pattern was here #### +0 +PREHOOK: query: SELECT count(*) FROM llap_bloom_parquet WHERE id = 6 +PREHOOK: type: QUERY +PREHOOK: Input: default@llap_bloom_parquet +#### A masked pattern was here #### +POSTHOOK: query: SELECT count(*) FROM llap_bloom_parquet WHERE id = 6 +POSTHOOK: type: QUERY +POSTHOOK: Input: default@llap_bloom_parquet +#### A masked pattern was here #### +1 +PREHOOK: query: DROP TABLE llap_bloom_parquet PURGE +PREHOOK: type: DROPTABLE +PREHOOK: Input: default@llap_bloom_parquet +PREHOOK: Output: database:default +PREHOOK: Output: default@llap_bloom_parquet +POSTHOOK: query: DROP TABLE llap_bloom_parquet PURGE +POSTHOOK: type: DROPTABLE +POSTHOOK: Input: default@llap_bloom_parquet +POSTHOOK: Output: database:default +POSTHOOK: Output: default@llap_bloom_parquet +PREHOOK: query: DROP TABLE IF EXISTS llap_bloom_multi PURGE +PREHOOK: type: DROPTABLE +PREHOOK: Output: database:default +POSTHOOK: query: DROP TABLE IF EXISTS llap_bloom_multi PURGE +POSTHOOK: type: DROPTABLE +POSTHOOK: Output: database:default +PREHOOK: query: CREATE EXTERNAL TABLE llap_bloom_multi (id bigint, name string) +STORED BY ICEBERG STORED AS PARQUET +TBLPROPERTIES ('format-version'='2', 'write.parquet.bloom-filter-enabled.column.id'='true', + 'write.parquet.bloom-filter-max-bytes'='1024', 'write.parquet.row-group-size-bytes'='1024') +PREHOOK: type: CREATETABLE +PREHOOK: Output: database:default +PREHOOK: Output: default@llap_bloom_multi +POSTHOOK: query: CREATE EXTERNAL TABLE llap_bloom_multi (id bigint, name string) +STORED BY ICEBERG STORED AS PARQUET +TBLPROPERTIES ('format-version'='2', 'write.parquet.bloom-filter-enabled.column.id'='true', + 'write.parquet.bloom-filter-max-bytes'='1024', 'write.parquet.row-group-size-bytes'='1024') +POSTHOOK: type: CREATETABLE +POSTHOOK: Output: database:default +POSTHOOK: Output: default@llap_bloom_multi +PREHOOK: query: INSERT INTO llap_bloom_multi +SELECT pos * 2, concat('n', pos) FROM (SELECT 1) x LATERAL VIEW posexplode(split(space(399), ' ')) e AS pos, val +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@llap_bloom_multi +POSTHOOK: query: INSERT INTO llap_bloom_multi +SELECT pos * 2, concat('n', pos) FROM (SELECT 1) x LATERAL VIEW posexplode(split(space(399), ' ')) e AS pos, val +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@llap_bloom_multi +PREHOOK: query: SELECT count(*) FROM llap_bloom_multi WHERE id = 51 +PREHOOK: type: QUERY +PREHOOK: Input: default@llap_bloom_multi +#### A masked pattern was here #### +POSTHOOK: query: SELECT count(*) FROM llap_bloom_multi WHERE id = 51 +POSTHOOK: type: QUERY +POSTHOOK: Input: default@llap_bloom_multi +#### A masked pattern was here #### +0 +PREHOOK: query: SELECT count(*) FROM llap_bloom_multi WHERE id = 651 +PREHOOK: type: QUERY +PREHOOK: Input: default@llap_bloom_multi +#### A masked pattern was here #### +POSTHOOK: query: SELECT count(*) FROM llap_bloom_multi WHERE id = 651 +POSTHOOK: type: QUERY +POSTHOOK: Input: default@llap_bloom_multi +#### A masked pattern was here #### +0 +PREHOOK: query: SELECT count(*) FROM llap_bloom_multi WHERE id = 51 +PREHOOK: type: QUERY +PREHOOK: Input: default@llap_bloom_multi +#### A masked pattern was here #### +POSTHOOK: query: SELECT count(*) FROM llap_bloom_multi WHERE id = 51 +POSTHOOK: type: QUERY +POSTHOOK: Input: default@llap_bloom_multi +#### A masked pattern was here #### +0 +PREHOOK: query: SELECT count(*) FROM llap_bloom_multi WHERE id = 651 +PREHOOK: type: QUERY +PREHOOK: Input: default@llap_bloom_multi +#### A masked pattern was here #### +POSTHOOK: query: SELECT count(*) FROM llap_bloom_multi WHERE id = 651 +POSTHOOK: type: QUERY +POSTHOOK: Input: default@llap_bloom_multi +#### A masked pattern was here #### +0 +PREHOOK: query: SELECT count(*) FROM llap_bloom_multi WHERE id = 4 +PREHOOK: type: QUERY +PREHOOK: Input: default@llap_bloom_multi +#### A masked pattern was here #### +POSTHOOK: query: SELECT count(*) FROM llap_bloom_multi WHERE id = 4 +POSTHOOK: type: QUERY +POSTHOOK: Input: default@llap_bloom_multi +#### A masked pattern was here #### +1 +PREHOOK: query: SELECT count(*) FROM llap_bloom_multi WHERE id = 700 +PREHOOK: type: QUERY +PREHOOK: Input: default@llap_bloom_multi +#### A masked pattern was here #### +POSTHOOK: query: SELECT count(*) FROM llap_bloom_multi WHERE id = 700 +POSTHOOK: type: QUERY +POSTHOOK: Input: default@llap_bloom_multi +#### A masked pattern was here #### +1 +PREHOOK: query: DROP TABLE llap_bloom_multi PURGE +PREHOOK: type: DROPTABLE +PREHOOK: Input: default@llap_bloom_multi +PREHOOK: Output: database:default +PREHOOK: Output: default@llap_bloom_multi +POSTHOOK: query: DROP TABLE llap_bloom_multi PURGE +POSTHOOK: type: DROPTABLE +POSTHOOK: Input: default@llap_bloom_multi +POSTHOOK: Output: database:default +POSTHOOK: Output: default@llap_bloom_multi diff --git a/itests/src/test/resources/testconfiguration.properties b/itests/src/test/resources/testconfiguration.properties index 5b077e8f5bf5..18d63420319c 100644 --- a/itests/src/test/resources/testconfiguration.properties +++ b/itests/src/test/resources/testconfiguration.properties @@ -410,6 +410,7 @@ iceberg.llap.query.files=\ iceberg_create_locally_zordered_table.q,\ iceberg_merge_delete_files.q,\ iceberg_merge_files.q,\ + llap_iceberg_bloom_filter.q,\ llap_iceberg_read_orc.q,\ llap_iceberg_read_parquet.q,\ puffin_col_stats_with_time_travel.q,\ @@ -468,6 +469,7 @@ iceberg.llap.only.query.files=\ iceberg_create_locally_zordered_table.q,\ iceberg_merge_delete_files.q,\ iceberg_merge_files.q,\ + llap_iceberg_bloom_filter.q,\ llap_iceberg_read_orc.q,\ llap_iceberg_read_parquet.q,\ puffin_col_stats_with_time_travel.q diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/io/api/LlapIo.java b/llap-client/src/java/org/apache/hadoop/hive/llap/io/api/LlapIo.java index 84b562156e43..a68db15cd27d 100644 --- a/llap-client/src/java/org/apache/hadoop/hive/llap/io/api/LlapIo.java +++ b/llap-client/src/java/org/apache/hadoop/hive/llap/io/api/LlapIo.java @@ -20,6 +20,8 @@ package org.apache.hadoop.hive.llap.io.api; import java.io.IOException; +import java.util.Map; +import java.util.SortedMap; import java.util.List; import org.apache.hadoop.conf.Configuration; @@ -80,6 +82,20 @@ InputFormat getInputFormat( MemoryBufferOrBuffers getParquetFooterBuffersFromCache(Path path, JobConf conf, @Nullable Object fileKey) throws IOException; + /** + * Returns the buffers holding the given bloom filters of the Parquet file on the given path, reading and + * caching the ones not held yet. The buffers come back locked, so the caller releases each of them with + * {@code decRefBuffer} once it has read them. + * + * @param path Parquet file path + * @param conf job conf + * @param fileKey fileId of the Parquet file (either the Long fileId of HDFS or the SyntheticFileId) + * @param ranges offset to length of every wanted bloom filter, as the footer records them + * @return the buffers by offset, or null if there is nothing to serve them for + */ + Map getParquetBloomFilterBuffersFromCache(Path path, JobConf conf, + @Nullable Object fileKey, SortedMap ranges) throws IOException; + /** * Handles request to evict entities specified in the request object. * @param protoRequest lists Hive entities (DB, table, etc..) whose LLAP buffers should be evicted. diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java index e7927eb8e02d..82084cc7ba93 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapIoImpl.java @@ -20,6 +20,10 @@ package org.apache.hadoop.hive.llap.io.api.impl; import java.io.IOException; +import java.util.TreeMap; +import java.util.SortedMap; +import java.util.Map; +import java.util.HashMap; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -129,7 +133,7 @@ public class LlapIoImpl implements LlapIo, LlapIoDebugDump { private final boolean useLowLevelCache; private ObjectName buddyAllocatorMXBean; private final Allocator allocator; - private final FileMetadataCache fileMetadataCache; + private final MetadataCache fileMetadataCache; private final LowLevelCache dataCache; private final SerDeLowLevelCacheImpl serdeCache; private final BufferUsageManager bufferManager; @@ -437,7 +441,7 @@ public void debugDumpShort(StringBuilder sb) { @Override public OrcTail getOrcTailFromCache(Path path, Configuration jobConf, CacheTag tag, Object fileKey) throws IOException { - return OrcEncodedDataReader.getOrcTailForPath(path, jobConf, tag, daemonConf, (MetadataCache) fileMetadataCache, fileKey); + return OrcEncodedDataReader.getOrcTailForPath(path, jobConf, tag, daemonConf, fileMetadataCache, fileKey); } @Override @@ -517,6 +521,55 @@ public MemoryBufferOrBuffers getParquetFooterBuffersFromCache(Path path, JobConf } } + @Override + public Map getParquetBloomFilterBuffersFromCache(Path path, JobConf conf, + Object fileKey, SortedMap ranges) throws IOException { + + Preconditions.checkNotNull(fileMetadataCache, "Metadata cache must not be null"); + if (fileKey == null || ranges.isEmpty()) { + return null; + } + + Map bloomFilters = new HashMap<>(ranges.size()); + SortedMap missing = new TreeMap<>(); + boolean done = false; + try { + for (Map.Entry range : ranges.entrySet()) { + MemoryBufferOrBuffers cached = fileMetadataCache.getParquetBloomFilters(fileKey, range.getKey()); + if (cached != null) { + LOG.debug("Serving {} bytes of bloom filter at {} for {} from cache", range.getValue(), + range.getKey(), fileKey); + bloomFilters.put(range.getKey(), cached); + } else { + missing.put(range.getKey(), range.getValue()); + } + } + if (!missing.isEmpty()) { + throwIfCacheOnlyRead(HiveConf.getBoolVar(conf, ConfVars.LLAP_IO_CACHE_ONLY)); + CacheTag tag = VectorizedParquetRecordReader.cacheTagOfParquetFile(path, daemonConf, conf); + final FileSystem fs = path.getFileSystem(conf); + // One open serves every filter this reader is missing. + try (SeekableInputStream stream = HadoopStreams.wrap(fs.open(path))) { + for (Map.Entry range : missing.entrySet()) { + stream.seek(range.getKey()); + LOG.debug("Caching {} bytes of bloom filter at {} for {}", range.getValue(), range.getKey(), + fileKey); + // Note: we don't pass in isStopped here - this is not on an IO thread. + bloomFilters.put(range.getKey(), fileMetadataCache.putParquetBloomFilters(fileKey, + range.getKey(), range.getValue(), stream, tag, null)); + } + } + } + done = true; + return bloomFilters; + } finally { + if (!done) { + // Nothing will read these, so do not leave them locked for the life of the daemon. + bloomFilters.values().forEach(fileMetadataCache::decRefBuffer); + } + } + } + @Override public LlapDaemonProtocolProtos.CacheEntryList fetchCachedContentInfo() { if (useLowLevelCache) { diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/MetadataCache.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/MetadataCache.java index 1f0543910b0d..5b8ba9704e17 100644 --- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/MetadataCache.java +++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/metadata/MetadataCache.java @@ -185,6 +185,24 @@ public LlapBufferOrBuffers getStripeTail(OrcBatchKey stripeKey) { return getInternal(new StripeKey(stripeKey.fileKey, stripeKey.stripeIx)); } + /** + * One bloom filter of a Parquet file, keyed by the offset the footer records for its column chunk. + */ + public LlapBufferOrBuffers getParquetBloomFilters(Object fileKey, long offset) { + return getInternal(new ParquetBloomFilterKey(fileKey, offset)); + } + + /** + * Cached at NORMAL rather than the HIGH the footers use. A footer is a few hundred bytes and the file + * cannot be read without it, while a bloom filter runs to a megabyte and only saves work, so it does not + * deserve the priority boost that would have it evict column data. + */ + public LlapBufferOrBuffers putParquetBloomFilters(Object fileKey, long offset, int length, InputStream is, + CacheTag tag, AtomicBoolean isStopped) throws IOException { + return putInternal(new ParquetBloomFilterKey(fileKey, offset), length, is, tag, isStopped, + Priority.NORMAL); + } + private LlapBufferOrBuffers getInternal(Object key) { LlapBufferOrBuffers result = metadata.get(key); if (result == null) return null; @@ -235,32 +253,41 @@ public LlapBufferOrBuffers putFileMetadata(Object fileKey, @Override public LlapBufferOrBuffers putFileMetadata(Object fileKey, int length, InputStream is, CacheTag tag, AtomicBoolean isStopped) throws IOException { + return putInternal(fileKey, length, is, tag, isStopped, Priority.HIGH); + } + + /** + * @param key what the entry is stored under. The buffers carry it too, as eviction removes the entry by + * the key its buffers hold. + */ + private LlapBufferOrBuffers putInternal(T key, int length, InputStream is, + CacheTag tag, AtomicBoolean isStopped, Priority priority) throws IOException { LlapBufferOrBuffers result = null; while (true) { // Overwhelmingly executes once, or maybe twice (replacing stale value). - LlapBufferOrBuffers oldVal = metadata.get(fileKey); + LlapBufferOrBuffers oldVal = metadata.get(key); if (oldVal == null) { - result = wrapBbForFile(result, fileKey, length, is, tag, isStopped); + result = wrapBbForFile(result, key, length, is, tag, isStopped); if (!lockBuffer(result, false)) { throw new AssertionError("Cannot lock a newly created value " + result); } - oldVal = metadata.putIfAbsent(fileKey, result); + oldVal = metadata.putIfAbsent(key, result); if (oldVal == null) { - cacheInPolicy(result); // Cached successfully, add to policy. + cacheInPolicy(result, priority); // Cached successfully, add to policy. return result; } } - if (lockOldVal(fileKey, result, oldVal)) { + if (lockOldVal(key, result, oldVal)) { return oldVal; } // We found some old value but couldn't incRef it; remove it. - metadata.remove(fileKey, oldVal); + metadata.remove(key, oldVal); } } @SuppressWarnings("unchecked") private LlapBufferOrBuffers wrapBbForFile(LlapBufferOrBuffers result, - Object fileKey, int length, InputStream stream, CacheTag tag, AtomicBoolean isStopped) throws IOException { + Object key, int length, InputStream stream, CacheTag tag, AtomicBoolean isStopped) throws IOException { if (result != null) { return result; } @@ -269,7 +296,7 @@ private LlapBufferOrBuffers wrapBbForFile(LlapBufferOrBuffers result, // allocated if a later read or allocation throws - otherwise it leaks (nothing else reclaims it). if (length <= maxAlloc) { // The whole footer fits in a single buffer - the overwhelmingly common case. - LlapMetadataBuffer buffer = new LlapMetadataBuffer<>(fileKey, tag); + LlapMetadataBuffer buffer = new LlapMetadataBuffer<>(key, tag); allocator.allocateMultiple(new MemoryBuffer[] { buffer }, length, null, isStopped); boolean done = false; try { @@ -285,7 +312,7 @@ private LlapBufferOrBuffers wrapBbForFile(LlapBufferOrBuffers result, // Larger footers are split across maxAlloc-sized chunks, the last one holding the remainder. LlapMetadataBuffer[] largeBuffers = new LlapMetadataBuffer[length / maxAlloc]; for (int i = 0; i < largeBuffers.length; ++i) { - largeBuffers[i] = new LlapMetadataBuffer<>(fileKey, tag); + largeBuffers[i] = new LlapMetadataBuffer<>(key, tag); } // allocateMultiple is all-or-nothing: on success every chunk is allocated; on failure it // releases whatever it reserved. @@ -303,7 +330,7 @@ private LlapBufferOrBuffers wrapBbForFile(LlapBufferOrBuffers result, return new LlapMetadataBuffers<>(largeBuffers); } // Allocate the remainder only; the last chunk is smaller than maxAlloc. - LlapMetadataBuffer remainder = new LlapMetadataBuffer<>(fileKey, tag); + LlapMetadataBuffer remainder = new LlapMetadataBuffer<>(key, tag); allocator.allocateMultiple(new MemoryBuffer[] { remainder }, smallSize, null, isStopped); smallBuffer = remainder; // Registered for cleanup only now that allocation succeeded. readIntoCacheBuffer(stream, smallSize, remainder); @@ -358,7 +385,7 @@ private LlapBufferOrBuffers putInternal(T key, ByteBuffer tailBuffer, CacheT result = wrapBb(result, key, tailBuffer, tag, isStopped); oldVal = metadata.putIfAbsent(key, result); if (oldVal == null) { - cacheInPolicy(result); // Cached successfully, add to policy. + cacheInPolicy(result, Priority.HIGH); // Cached successfully, add to policy. return result; } } @@ -370,14 +397,14 @@ private LlapBufferOrBuffers putInternal(T key, ByteBuffer tailBuffer, CacheT } } - private void cacheInPolicy(LlapBufferOrBuffers buffers) { + private void cacheInPolicy(LlapBufferOrBuffers buffers, Priority priority) { LlapAllocatorBuffer singleBuffer = buffers.getSingleLlapBuffer(); if (singleBuffer != null) { - policy.cache(singleBuffer, Priority.HIGH); + policy.cache(singleBuffer, priority); return; } for (LlapAllocatorBuffer buffer : buffers.getMultipleLlapBuffers()) { - policy.cache(buffer, Priority.HIGH); + policy.cache(buffer, priority); } } @@ -549,6 +576,10 @@ private void unlockSingleBuffer(LlapAllocatorBuffer buffer, boolean isCached) { metrics.decrCacheNumLockedBuffers(); } + /** Distinguishes a file's cached bloom filters from its footer, which uses the file key itself. */ + private record ParquetBloomFilterKey(Object fileKey, long offset) { + } + private final static class StripeKey { private final Object fileKey; private final int stripeIx; diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestMetadataCache.java b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestMetadataCache.java index 96d621f9e37f..8591b3b4cd9e 100644 --- a/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestMetadataCache.java +++ b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestMetadataCache.java @@ -25,6 +25,8 @@ import java.io.EOFException; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; import java.util.Random; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Predicate; @@ -60,9 +62,11 @@ public class TestMetadataCache { private static class DummyCachePolicy implements LowLevelCachePolicy { int lockCount = 0, unlockCount = 0; + final List priorities = new ArrayList<>(); public void cache(LlapCacheableBuffer buffer, Priority pri) { ++lockCount; + priorities.add(pri); } public void notifyLock(LlapCacheableBuffer buffer) { @@ -247,6 +251,61 @@ private void assertFooterReadFailureReleasesBuffers(int length) throws IOExcepti alloc.allocateMultiple(wholeArena, MAX_ALLOC); } + /** + * A file's bloom filters are cached under a wrapper key while its footer keeps the bare file key. + * Eviction removes an entry by the key its buffers carry, so a bloom filter whose buffers carried the + * file key would take the footer's entry with it - and bloom filters, about a megabyte each against a + * footer of a few hundred bytes, are exactly what gets evicted. + */ + @Test + public void testEvictingBloomFilterKeepsFooter() throws Exception { + MetadataCache cache = newMetadataCache(); + Object fileKey = new Object(); + byte[] footer = new byte[MAX_ALLOC - 1]; + byte[] bloom = new byte[MAX_ALLOC - 1]; + new Random(0).nextBytes(footer); + java.util.Arrays.fill(bloom, (byte) 7); + + LlapBufferOrBuffers footerBuffers = + cache.putFileMetadata(fileKey, footer.length, new ByteArrayInputStream(footer), null, null); + cache.decRefBuffer(footerBuffers); + LlapBufferOrBuffers bloomBuffers = cache.putParquetBloomFilters(fileKey, 4096, bloom.length, + new ByteArrayInputStream(bloom), null, null); + cache.decRefBuffer(bloomBuffers); + + cache.notifyEvicted((LlapMetadataBuffer) bloomBuffers.getSingleBuffer()); + + LlapBufferOrBuffers footerAfter = cache.getFileMetadata(fileKey); + assertNotNull("evicting a bloom filter must not remove the file's footer", footerAfter); + cache.decRefBuffer(footerAfter); + assertNull("the evicted bloom filter must be gone", cache.getParquetBloomFilters(fileKey, 4096)); + } + + /** + * Each bloom filter of a file is cached on its own, so evicting one must leave the others alone. + */ + @Test + public void testEvictingBloomFilterKeepsTheOthers() throws Exception { + MetadataCache cache = newMetadataCache(); + Object fileKey = new Object(); + byte[] bloom = new byte[MAX_ALLOC - 1]; + java.util.Arrays.fill(bloom, (byte) 3); + + LlapBufferOrBuffers first = cache.putParquetBloomFilters(fileKey, 1024, bloom.length, + new ByteArrayInputStream(bloom), null, null); + cache.decRefBuffer(first); + LlapBufferOrBuffers second = cache.putParquetBloomFilters(fileKey, 8192, bloom.length, + new ByteArrayInputStream(bloom), null, null); + cache.decRefBuffer(second); + + cache.notifyEvicted((LlapMetadataBuffer) first.getSingleBuffer()); + + assertNull(cache.getParquetBloomFilters(fileKey, 1024)); + LlapBufferOrBuffers kept = cache.getParquetBloomFilters(fileKey, 8192); + assertNotNull("evicting one bloom filter must not remove another", kept); + cache.decRefBuffer(kept); + } + private MetadataCache newMetadataCache() { return newMetadataCache(4096); } @@ -460,4 +519,22 @@ public void verifyResult(DiskRangeList result, long... vals) { } assertNull(result); } + + @Test + public void testABloomFilterIsCachedBelowTheFooterItBelongsTo() throws Exception { + // a footer has to be read before anything else of the file can be, so it outranks a filter that + // only saves reading data + DummyCachePolicy policy = new DummyCachePolicy(); + MetadataCache cache = newMetadataCache(MAX_ALLOC, policy); + Object fileKey = new Object(); + byte[] bytes = new byte[MAX_ALLOC - 1]; + + cache.decRefBuffer(cache.putFileMetadata(fileKey, bytes.length, + new ByteArrayInputStream(bytes), null, null)); + cache.decRefBuffer(cache.putParquetBloomFilters(fileKey, 4096, bytes.length, + new ByteArrayInputStream(bytes), null, null)); + + assertEquals("a footer is cached high and a bloom filter below it", + List.of(Priority.HIGH, Priority.NORMAL), policy.priorities); + } } diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestParquetBloomFilterBufferRelease.java b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestParquetBloomFilterBufferRelease.java new file mode 100644 index 000000000000..2495bc38ae11 --- /dev/null +++ b/llap-server/src/test/org/apache/hadoop/hive/llap/cache/TestParquetBloomFilterBufferRelease.java @@ -0,0 +1,180 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.hadoop.hive.llap.cache; + +import java.util.Arrays; +import java.util.List; +import java.util.SortedMap; +import java.util.TreeMap; +import java.util.Map; + +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.llap.LlapHiveUtils; +import org.apache.hadoop.hive.common.io.encoded.MemoryBufferOrBuffers; +import org.apache.hadoop.hive.llap.io.api.LlapProxy; +import org.apache.hadoop.hive.ql.exec.SerializationUtilities; +import org.apache.hadoop.hive.ql.exec.Utilities; +import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatchCtx; +import org.apache.hadoop.hive.ql.io.parquet.AbstractTestParquetDirect; +import org.apache.hadoop.hive.ql.io.parquet.VectorizedParquetInputFormat; +import org.apache.hadoop.hive.ql.io.parquet.serde.ArrayWritableObjectInspector; +import org.apache.hadoop.hive.ql.io.parquet.vector.VectorizedParquetRecordReader; +import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc; +import org.apache.hadoop.hive.ql.plan.ExprNodeConstantDesc; +import org.apache.hadoop.hive.ql.plan.ExprNodeDesc; +import org.apache.hadoop.hive.ql.plan.ExprNodeGenericFuncDesc; +import org.apache.hadoop.hive.ql.plan.MapWork; +import org.apache.hadoop.hive.ql.plan.TableScanDesc; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPEqual; +import org.apache.hadoop.hive.serde2.ColumnProjectionUtils; +import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector; +import org.apache.hadoop.hive.serde2.typeinfo.StructTypeInfo; +import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory; +import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils; +import org.apache.hadoop.mapred.FileSplit; +import org.apache.hadoop.mapred.JobConf; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.metadata.ParquetMetadata; +import org.apache.parquet.hadoop.util.HadoopInputFile; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; + +import com.google.common.collect.Lists; + +/** + * That a read lets go of the bloom filter buffers it locks in the metadata cache. A buffer left locked is + * never evicted, so a leak fills a daemon's cache with entries nothing can reclaim - and because a locked + * buffer is still readable, every other test of this feature would keep passing. + * + *

This lives beside the cache rather than beside the reader because the reference count it asserts on is + * visible only within this package. + */ +public class TestParquetBloomFilterBufferRelease extends AbstractTestParquetDirect { + + @BeforeClass + public static void startLlapIo() throws Exception { + HiveConf daemonConf = new HiveConf(); + HiveConf.setVar(daemonConf, HiveConf.ConfVars.LLAP_IO_MEMORY_MAX_SIZE, "64Mb"); + HiveConf.setBoolVar(daemonConf, HiveConf.ConfVars.LLAP_ALLOCATOR_DIRECT, false); + HiveConf.setBoolVar(daemonConf, HiveConf.ConfVars.LLAP_ALLOCATOR_PREALLOCATE, false); + HiveConf.setIntVar(daemonConf, HiveConf.ConfVars.LLAP_ALLOCATOR_ARENA_COUNT, 1); + HiveConf.setBoolVar(daemonConf, HiveConf.ConfVars.LLAP_TRACK_CACHE_USAGE, false); + LlapProxy.setDaemon(true); + LlapProxy.initializeLlapIo(daemonConf); + Assert.assertTrue(LlapProxy.getIo().usingLowLevelCache()); + } + + @AfterClass + public static void stopLlapIo() { + LlapProxy.close(); + } + + @Test + public void testAReadLetsGoOfTheFiltersItLocks() throws Exception { + JobConf conf = new JobConf(); + Path file = writeBloomFilterFile(); + + // reading with an equality predicate caches the filter it prunes by, and should leave it unlocked + Assert.assertEquals("the row group holds no odd value, so the bloom filter drops it", + 0, filteredBlocks(file, conf, 51).size()); + + long offset = bloomOffsetOf(file, conf); + SortedMap ranges = new TreeMap<>(); + ranges.put(offset, bloomLengthOf(file, conf)); + Map served = LlapProxy.getIo() + .getParquetBloomFilterBuffersFromCache(file, conf, fileKeyOf(file, conf), ranges); + Assert.assertNotNull("the read should have cached the filter it used", served); + + // this fetch is the only lock standing: a read that kept its own would make it two + LlapAllocatorBuffer buffer = (LlapAllocatorBuffer) served.get(offset).getSingleBuffer(); + Assert.assertEquals("the read let go of the filter it locked", 1, buffer.getRefCount()); + buffer.decRef(); + } + + private static Object fileKeyOf(Path file, JobConf conf) throws Exception { + return LlapHiveUtils.createFileIdUsingFS(file.getFileSystem(conf), file, conf); + } + + private static long bloomOffsetOf(Path file, JobConf conf) throws Exception { + try (ParquetFileReader reader = ParquetFileReader.open(HadoopInputFile.fromPath(file, conf))) { + return reader.getFooter().getBlocks().get(0).getColumns().get(0).getBloomFilterOffset(); + } + } + + private static int bloomLengthOf(Path file, JobConf conf) throws Exception { + try (ParquetFileReader reader = ParquetFileReader.open(HadoopInputFile.fromPath(file, conf))) { + return reader.getFooter().getBlocks().get(0).getColumns().get(0).getBloomFilterLength(); + } + } + + private Path writeBloomFilterFile() throws Exception { + MessageType fileSchema = MessageTypeParser.parseMessageType( + "message hive_schema {\n optional int32 intCol;\n}\n"); + return writeDirect("BloomFilterBufferRelease", fileSchema, + consumer -> { + for (int i = 0; i < 100; i++) { + consumer.startMessage(); + consumer.startField("intCol", 0); + consumer.addInteger(i * 2); + consumer.endField("intCol", 0); + consumer.endMessage(); + } + }, + builder -> builder.withBloomFilterEnabled("intCol", true)); + } + + private List filteredBlocks( + Path file, JobConf conf, int value) throws Exception { + StructTypeInfo rowTypeInfo = (StructTypeInfo) TypeInfoFactory.getStructTypeInfo( + Arrays.asList("intCol"), TypeInfoUtils.getTypeInfosFromTypeString("int")); + StructObjectInspector inspector = new ArrayWritableObjectInspector(rowTypeInfo); + + conf.set(ColumnProjectionUtils.READ_COLUMN_NAMES_CONF_STR, "intCol"); + conf.set("columns", "intCol"); + conf.set("columns.types", "int"); + + List children = Lists.newArrayList( + new ExprNodeColumnDesc(Integer.class, "intCol", "T", false), new ExprNodeConstantDesc(value)); + ExprNodeGenericFuncDesc predicate = + new ExprNodeGenericFuncDesc(inspector, new GenericUDFOPEqual(), children); + conf.set(TableScanDesc.FILTER_EXPR_CONF_STR, SerializationUtilities.serializeExpression(predicate)); + + MapWork mapWork = new MapWork(); + VectorizedRowBatchCtx rbCtx = new VectorizedRowBatchCtx(); + rbCtx.init(inspector, new String[0]); + mapWork.setVectorMode(true); + mapWork.setVectorizedRowBatchCtx(rbCtx); + HiveConf.setBoolVar(conf, HiveConf.ConfVars.HIVE_VECTORIZATION_ENABLED, true); + HiveConf.setVar(conf, HiveConf.ConfVars.PLAN, "//tmp"); + Utilities.setMapWork(conf, mapWork); + + VectorizedParquetInputFormat inputFormat = new VectorizedParquetInputFormat(); + LlapProxy.getIo().initCacheOnlyInputFormat(inputFormat); + FileSplit split = new FileSplit(file, 0, fileLength(file), (String[]) null); + try (VectorizedParquetRecordReader reader = + (VectorizedParquetRecordReader) inputFormat.getRecordReader(split, conf, null)) { + return reader.getFilteredBlocks(); + } + } +} diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/io/api/impl/TestLlapParquetBloomFilterCache.java b/llap-server/src/test/org/apache/hadoop/hive/llap/io/api/impl/TestLlapParquetBloomFilterCache.java new file mode 100644 index 000000000000..1d4c5410ed9e --- /dev/null +++ b/llap-server/src/test/org/apache/hadoop/hive/llap/io/api/impl/TestLlapParquetBloomFilterCache.java @@ -0,0 +1,306 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.hadoop.hive.llap.io.api.impl; + +import java.util.Arrays; +import java.util.List; + +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.llap.io.api.LlapProxy; +import org.apache.hadoop.hive.ql.exec.SerializationUtilities; +import org.apache.hadoop.hive.ql.exec.Utilities; +import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatchCtx; +import org.apache.hadoop.hive.ql.io.parquet.AbstractTestParquetDirect; +import org.apache.hadoop.hive.ql.io.parquet.VectorizedParquetInputFormat; +import org.apache.hadoop.hive.ql.io.parquet.serde.ArrayWritableObjectInspector; +import org.apache.hadoop.hive.ql.io.parquet.vector.VectorizedParquetRecordReader; +import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc; +import org.apache.hadoop.hive.ql.plan.ExprNodeConstantDesc; +import org.apache.hadoop.hive.ql.plan.ExprNodeDesc; +import org.apache.hadoop.hive.ql.plan.ExprNodeGenericFuncDesc; +import org.apache.hadoop.hive.ql.plan.MapWork; +import org.apache.hadoop.hive.ql.plan.TableScanDesc; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDF; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPEqual; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPGreaterThan; +import org.apache.hadoop.hive.serde2.ColumnProjectionUtils; +import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector; +import org.apache.hadoop.hive.serde2.typeinfo.StructTypeInfo; +import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory; +import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils; +import org.apache.hadoop.mapred.FileSplit; +import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.conf.Configuration; +import org.apache.parquet.hadoop.ParquetWriter; +import org.apache.parquet.hadoop.api.WriteSupport; +import org.apache.parquet.hadoop.metadata.BlockMetaData; +import org.apache.parquet.io.api.RecordConsumer; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +import com.google.common.collect.Lists; + +/** + * That the bloom filters a vectorized Parquet read prunes by are served from the LLAP metadata cache. + * The query results are the same whether they come from the cache or the file, so what shows the cache + * was used is a read that is not allowed to touch the file: it can only answer from what was cached. + */ +public class TestLlapParquetBloomFilterCache extends AbstractTestParquetDirect { + + private static final String COLUMN_NAMES = "intCol"; + private static final String COLUMN_TYPES = "int"; + + private JobConf conf; + + @BeforeClass + public static void startLlapIo() throws Exception { + HiveConf daemonConf = new HiveConf(); + // a real cache, sized for a test rather than for a daemon + HiveConf.setVar(daemonConf, HiveConf.ConfVars.LLAP_IO_MEMORY_MAX_SIZE, "64Mb"); + HiveConf.setBoolVar(daemonConf, HiveConf.ConfVars.LLAP_ALLOCATOR_DIRECT, false); + HiveConf.setBoolVar(daemonConf, HiveConf.ConfVars.LLAP_ALLOCATOR_PREALLOCATE, false); + HiveConf.setIntVar(daemonConf, HiveConf.ConfVars.LLAP_ALLOCATOR_ARENA_COUNT, 1); + HiveConf.setBoolVar(daemonConf, HiveConf.ConfVars.LLAP_TRACK_CACHE_USAGE, false); + + LlapProxy.setDaemon(true); + LlapProxy.initializeLlapIo(daemonConf); + Assert.assertTrue("these filters are held by the low level cache", + LlapProxy.getIo().usingLowLevelCache()); + } + + @AfterClass + public static void stopLlapIo() { + LlapProxy.close(); + } + + @Before + public void initConf() { + conf = new JobConf(); + } + + @Test + public void testTheFiltersPrunedByComeFromTheCache() throws Exception { + StructObjectInspector inspector = objectInspector(); + Path first = writeBloomFilterFile("LlapBloomFirst"); + Path second = writeBloomFilterFile("LlapBloomSecond"); + + // this read is here to fill the cache; the pruning itself is pinned by TestParquetRowGroupFilter + Assert.assertEquals("the row group holds no odd value, so the bloom filter drops it", + 0, filteredBlocks(first, inspector, 51, new GenericUDFOPEqual()).size()); + + // the second file is read by a predicate no bloom filter answers, so its footer is cached and + // its filters are not: what the cache holds of the two files differs only in the filters + Assert.assertEquals("a greater-than keeps the row group", + 1, filteredBlocks(second, inspector, 5, new GenericUDFOPGreaterThan()).size()); + + // from here neither file may be read, so anything answered comes from the cache + HiveConf.setBoolVar(conf, HiveConf.ConfVars.LLAP_IO_CACHE_ONLY, true); + + Assert.assertEquals("the filter cached by the first read still drops the row group", + 0, filteredBlocks(first, inspector, 51, new GenericUDFOPEqual()).size()); + + Assert.assertEquals("a value the row group holds keeps it, so the cached filter is the right one", + 1, filteredBlocks(first, inspector, 50, new GenericUDFOPEqual()).size()); + + // proves the second file's footer is in the cache: a predicate needing no filter reads fine under + // cache only, so the refusal below can only be about the filters + Assert.assertEquals("the second file's footer is cached, so a greater-than answers from the cache", + 1, filteredBlocks(second, inspector, 5, new GenericUDFOPGreaterThan()).size()); + + // the second file's footer is cached but its filters are not, so the filters cannot be had + try { + filteredBlocks(second, inspector, 51, new GenericUDFOPEqual()); + Assert.fail("a filter that was never cached must not be read from the file under cache only"); + } catch (RuntimeException e) { + Assert.assertTrue("expected the cache only refusal, got: " + rootCause(e), + rootCause(e).contains(HiveConf.ConfVars.LLAP_IO_CACHE_ONLY.varname)); + } + } + + private static String rootCause(Throwable t) { + Throwable cause = t; + while (cause.getCause() != null) { + cause = cause.getCause(); + } + return String.valueOf(cause.getMessage()); + } + + /** One row group of even values, with a bloom filter over the column. */ + private Path writeBloomFilterFile(String name) throws Exception { + MessageType fileSchema = MessageTypeParser.parseMessageType( + "message hive_schema {\n optional int32 intCol;\n}\n"); + return writeDirect(name, fileSchema, + consumer -> { + for (int i = 0; i < 100; i++) { + consumer.startMessage(); + consumer.startField("intCol", 0); + consumer.addInteger(i * 2); + consumer.endField("intCol", 0); + consumer.endMessage(); + } + }, + builder -> builder.withBloomFilterEnabled("intCol", true)); + } + + private List filteredBlocks(Path file, StructObjectInspector inspector, int value, + GenericUDF comparison) throws Exception { + conf.set(ColumnProjectionUtils.READ_COLUMN_NAMES_CONF_STR, COLUMN_NAMES); + conf.set("columns", COLUMN_NAMES); + conf.set("columns.types", COLUMN_TYPES); + + List children = Lists.newArrayList( + new ExprNodeColumnDesc(Integer.class, "intCol", "T", false), new ExprNodeConstantDesc(value)); + ExprNodeGenericFuncDesc predicate = + new ExprNodeGenericFuncDesc(inspector, comparison, children); + conf.set(TableScanDesc.FILTER_EXPR_CONF_STR, SerializationUtilities.serializeExpression(predicate)); + + MapWork mapWork = new MapWork(); + VectorizedRowBatchCtx rbCtx = new VectorizedRowBatchCtx(); + rbCtx.init(inspector, new String[0]); + mapWork.setVectorMode(true); + mapWork.setVectorizedRowBatchCtx(rbCtx); + HiveConf.setBoolVar(conf, HiveConf.ConfVars.HIVE_VECTORIZATION_ENABLED, true); + HiveConf.setVar(conf, HiveConf.ConfVars.PLAN, "//tmp"); + Utilities.setMapWork(conf, mapWork); + + // the caches reach the reader the way they do in a daemon, through the input format + VectorizedParquetInputFormat inputFormat = new VectorizedParquetInputFormat(); + LlapProxy.getIo().initCacheOnlyInputFormat(inputFormat); + FileSplit split = new FileSplit(file, 0, fileLength(file), (String[]) null); + try (VectorizedParquetRecordReader reader = + (VectorizedParquetRecordReader) inputFormat.getRecordReader(split, conf, null)) { + return reader.getFilteredBlocks(); + } + } + + private ArrayWritableObjectInspector objectInspector() { + StructTypeInfo rowTypeInfo = (StructTypeInfo) TypeInfoFactory.getStructTypeInfo( + Arrays.asList(COLUMN_NAMES.split(",")), + TypeInfoUtils.getTypeInfosFromTypeString(COLUMN_TYPES)); + return new ArrayWritableObjectInspector(rowTypeInfo); + } + + /** + * A file of two row groups, where a value held by one is inside the other's statistics: only the filters + * tell them apart, so each row group has to be served the filter that belongs to it. The first read leaves + * one of the two filters cached, so the second is served from a cache that holds part of what it asks for. + */ + @Test + public void testEachRowGroupIsServedItsOwnFilter() throws Exception { + StructObjectInspector inspector = objectInspector(); + Path file = writeTwoRowGroups("LlapBloomTwoRowGroups"); + + // a predicate no filter answers, so this reads the row groups the statistics keep: both of them + List both = filteredBlocks(file, inspector, -1, new GenericUDFOPGreaterThan()); + Assert.assertEquals("the file should hold two row groups for the rest of this to mean anything", + 2, both.size()); + long firstRowGroup = both.get(0).getStartingPos(); + long secondRowGroup = both.get(1).getStartingPos(); + + // 0 is below everything the second row group holds, so the statistics leave one row group to ask + // about and only that one's filter is read and cached + Assert.assertEquals("the first row group holds 0", + 1, filteredBlocks(file, inspector, 0, new GenericUDFOPEqual()).size()); + + // both row groups' statistics span 50, so both filters are wanted and only one of them is cached + List holdingFifty = filteredBlocks(file, inspector, 50, new GenericUDFOPEqual()); + Assert.assertEquals("50 is held by one row group", 1, holdingFifty.size()); + Assert.assertEquals("and it is the first, so it was served its own filter", + firstRowGroup, holdingFifty.get(0).getStartingPos()); + + List holdingFiftyOne = filteredBlocks(file, inspector, 51, new GenericUDFOPEqual()); + Assert.assertEquals("51 is held by one row group", 1, holdingFiftyOne.size()); + Assert.assertEquals("and it is the second, so the two filters were not swapped", + secondRowGroup, holdingFiftyOne.get(0).getStartingPos()); + + // both filters are cached by now, so the same answers must come from the cache alone + HiveConf.setBoolVar(conf, HiveConf.ConfVars.LLAP_IO_CACHE_ONLY, true); + Assert.assertEquals("the cached filter of the first row group still holds 50", + firstRowGroup, filteredBlocks(file, inspector, 50, new GenericUDFOPEqual()).get(0).getStartingPos()); + Assert.assertEquals("the cached filter of the second row group still holds 51", + secondRowGroup, filteredBlocks(file, inspector, 51, new GenericUDFOPEqual()).get(0).getStartingPos()); + } + + /** + * Evens then odds, a row group each, so a value of either sits inside the other's statistics. Written a + * record at a time because Parquet decides where a row group ends by counting the records it is handed. + */ + private Path writeTwoRowGroups(String name) throws Exception { + java.io.File temp = tempDir.newFile(name + ".parquet"); + temp.delete(); + Path path = new Path(temp.getPath()); + MessageType schema = MessageTypeParser.parseMessageType( + "message hive_schema {\n optional int32 intCol;\n}\n"); + try (ParquetWriter writer = new IntBuilder(path, schema) + .withBloomFilterEnabled("intCol", true) + .withRowGroupSize(128L) + .build()) { + for (int i = 0; i < 200; i++) { + writer.write(i < 100 ? i * 2 : (i - 100) * 2 + 1); + } + } + return path; + } + + private static final class IntBuilder extends ParquetWriter.Builder { + private final MessageType schema; + + private IntBuilder(Path file, MessageType schema) { + super(file); + this.schema = schema; + } + + @Override + protected IntBuilder self() { + return this; + } + + @Override + protected WriteSupport getWriteSupport(Configuration conf) { + return new WriteSupport() { + private RecordConsumer consumer; + + @Override + public WriteContext init(Configuration configuration) { + return new WriteContext(schema, new java.util.HashMap<>()); + } + + @Override + public void prepareForWrite(RecordConsumer recordConsumer) { + this.consumer = recordConsumer; + } + + @Override + public void write(Integer value) { + consumer.startMessage(); + consumer.startField("intCol", 0); + consumer.addInteger(value); + consumer.endField("intCol", 0); + consumer.endMessage(); + } + }; + } + } +} diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/ParquetRecordReaderBase.java b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/ParquetRecordReaderBase.java index 50c30e2941a3..24147df3dcca 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/ParquetRecordReaderBase.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/ParquetRecordReaderBase.java @@ -143,7 +143,7 @@ protected ParquetInputSplit getSplit( FilterCompat.Filter filter = setFilter(jobConf, fileMetaData.getSchema()); if (filter != null) { - filteredBlocks = RowGroupFilter.filterRowGroups(filter, splitGroup, fileMetaData.getSchema()); + filteredBlocks = filterRowGroups(filter, splitGroup, fileMetaData); if (filteredBlocks.isEmpty()) { LOG.debug("All row groups are dropped due to filter predicates"); return null; @@ -181,6 +181,15 @@ protected ParquetInputSplit getSplit( return split; } + /** + * Prunes the row groups of this split with the pushed down predicate, using the statistics that the footer + * already holds. + */ + protected List filterRowGroups(FilterCompat.Filter filter, List splitGroup, + FileMetaData fileMetaData) throws IOException { + return RowGroupFilter.filterRowGroups(filter, splitGroup, fileMetaData.getSchema()); + } + @SuppressWarnings("deprecation") protected ParquetMetadata getParquetMetadata(Path path, JobConf conf) throws IOException { return ParquetFileReader.readFooter(jobConf, path); diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/ParquetFilterDataFromCache.java b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/ParquetFilterDataFromCache.java new file mode 100644 index 000000000000..fe8a9018a399 --- /dev/null +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/ParquetFilterDataFromCache.java @@ -0,0 +1,198 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hive.ql.io.parquet.vector; + +import java.io.EOFException; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.List; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.common.io.encoded.MemoryBuffer; +import org.apache.hadoop.hive.common.io.encoded.MemoryBufferOrBuffers; +import org.apache.parquet.hadoop.util.HadoopInputFile; +import org.apache.parquet.io.InputFile; +import org.apache.parquet.io.SeekableInputStream; + +/** + * Serves cached byte ranges of a Parquet file at the offsets the file itself uses. + * + *

Parquet seeks to the offset the footer records for a column chunk to read a bloom filter, so unlike + * {@link ParquetFooterInputFromCache}, which presents the footer as a small file of its own, this keeps the + * original offsets and backs only the ranges that were cached. + * + *

Ranges are disjoint and a read stays inside the one it started in, as every structure served here is + * self contained at its offset. + */ +public final class ParquetFilterDataFromCache + extends SeekableInputStream implements InputFile { + + /** A cached range of the file, held at the offset the file stores it at. */ + public record Range(long offset, int length, MemoryBufferOrBuffers data) { + } + + private final long[] starts; + private final long[] ends; + private final MemoryBuffer[][] buffers; + private final Path path; + private final Configuration conf; + + private long fileLength = -1; + private long position; + private int rangeIx = -1; + private int bufferIx; + private int bufferPos; + + public ParquetFilterDataFromCache(List ranges, Path path, Configuration conf) { + this.path = path; + this.conf = conf; + starts = new long[ranges.size()]; + ends = new long[ranges.size()]; + buffers = new MemoryBuffer[ranges.size()][]; + for (int i = 0; i < ranges.size(); ++i) { + Range range = ranges.get(i); + MemoryBuffer single = range.data().getSingleBuffer(); + buffers[i] = (single != null) ? new MemoryBuffer[] { single } : range.data().getMultipleBuffers(); + starts[i] = range.offset(); + ends[i] = range.offset() + range.length(); + } + position = (starts.length == 0) ? 0 : starts[0]; + } + + @Override + public long getLength() throws IOException { + // Only selected ranges are backed here, so the length comes from the file rather than from the last + // range end, which would be short. Parquet does not ask for it when the footer is supplied, so this + // stats the file at most once and usually never. + if (fileLength < 0) { + fileLength = HadoopInputFile.fromPath(path, conf).getLength(); + } + return fileLength; + } + + @Override + public SeekableInputStream newStream() { + return this; + } + + @Override + public long getPos() { + return position; + } + + @Override + public void seek(long targetPos) throws IOException { + for (int i = 0; i < starts.length; ++i) { + if (targetPos >= starts[i] && targetPos < ends[i]) { + position = targetPos; + rangeIx = i; + long relative = targetPos - starts[i]; + for (bufferIx = 0; bufferIx < buffers[i].length; ++bufferIx) { + int size = buffers[i][bufferIx].getByteBufferRaw().remaining(); + if (relative < size) { + bufferPos = (int) relative; + return; + } + relative -= size; + } + bufferPos = 0; + return; + } + } + throw new IOException("Seek to " + targetPos + " outside the cached ranges " + describeRanges()); + } + + private String describeRanges() { + StringBuilder sb = new StringBuilder("["); + for (int i = 0; i < starts.length; ++i) { + sb.append(i == 0 ? "" : ", ").append('[').append(starts[i]).append(", ").append(ends[i]).append(')'); + } + return sb.append(']').toString(); + } + + private int readInternal(byte[] b, int offset, int len) { + if (rangeIx < 0) { + return 0; + } + int argPos = offset; + int argEnd = offset + len; + MemoryBuffer[] rangeBuffers = buffers[rangeIx]; + while (argPos < argEnd) { + if (bufferIx >= rangeBuffers.length) { + return argPos - offset; + } + ByteBuffer data = rangeBuffers[bufferIx].getByteBufferDup(); + int available = data.remaining() - bufferPos; + if (available <= 0) { + ++bufferIx; + bufferPos = 0; + continue; + } + int toConsume = Math.min(argEnd - argPos, available); + data.position(data.position() + bufferPos); + data.get(b, argPos, toConsume); + bufferPos += toConsume; + argPos += toConsume; + position += toConsume; + } + return len; + } + + @Override + public void readFully(byte[] b, int offset, int len) throws IOException { + if (readInternal(b, offset, len) != len) { + throw new EOFException(); + } + } + + @Override + public void readFully(byte[] b) throws IOException { + readFully(b, 0, b.length); + } + + @Override + public int read(byte[] b, int offset, int len) { + int read = readInternal(b, offset, len); + return (read == 0 && len > 0) ? -1 : read; + } + + @Override + public int read() throws IOException { + byte[] one = new byte[1]; + return (readInternal(one, 0, 1) == 1) ? (one[0] & 0xFF) : -1; + } + + @Override + public int read(ByteBuffer bb) throws IOException { + byte[] buffer = new byte[bb.remaining()]; + int read = readInternal(buffer, 0, buffer.length); + if (read <= 0) { + return -1; + } + bb.put(buffer, 0, read); + return read; + } + + @Override + public void readFully(ByteBuffer bb) throws IOException { + byte[] buffer = new byte[bb.remaining()]; + readFully(buffer, 0, buffer.length); + bb.put(buffer); + } +} diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedParquetRecordReader.java b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedParquetRecordReader.java index 236f6f3095f0..f74d08a0da6a 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedParquetRecordReader.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedParquetRecordReader.java @@ -58,13 +58,21 @@ import org.apache.hadoop.mapred.RecordReader; import org.apache.parquet.ParquetRuntimeException; import org.apache.parquet.column.ColumnDescriptor; +import org.apache.parquet.HadoopReadOptions; +import org.apache.parquet.ParquetReadOptions; +import org.apache.parquet.filter2.compat.FilterCompat; +import org.apache.parquet.filter2.compat.RowGroupFilter; +import org.apache.parquet.filter2.predicate.FilterPredicate; +import org.apache.parquet.filter2.predicate.Operators; import org.apache.parquet.column.page.PageReadStore; import org.apache.parquet.format.converter.ParquetMetadataConverter.MetadataFilter; import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.ParquetInputFormat; import org.apache.parquet.hadoop.ParquetInputSplit; import org.apache.parquet.hadoop.metadata.BlockMetaData; import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; import org.apache.parquet.hadoop.metadata.ColumnPath; +import org.apache.parquet.hadoop.metadata.FileMetaData; import org.apache.parquet.hadoop.metadata.ParquetMetadata; import org.apache.parquet.hadoop.util.HadoopStreams; import org.apache.parquet.io.InputFile; @@ -79,12 +87,15 @@ import java.io.IOException; import java.time.ZoneId; +import com.google.common.annotations.VisibleForTesting; import java.util.ArrayList; +import java.util.Collections; import java.util.Arrays; import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.SortedMap; import java.util.Optional; import java.util.Set; import java.util.TreeMap; @@ -283,6 +294,176 @@ public void initialize( configuration, parquetMetadata.getFileMetaData(), path, blocks, requestedSchema.getColumns()); } + /** + * Prunes the row groups of this split with the pushed down predicate. + * + *

Statistics come from the footer, which is already in memory. Bloom filters live in the data file and + * need an open reader. + */ + @Override + protected List filterRowGroups(FilterCompat.Filter filter, List splitGroup, + FileMetaData fileMetaData) throws IOException { + List statsFiltered = super.filterRowGroups(filter, splitGroup, fileMetaData); + if (statsFiltered.isEmpty() || !jobConf.getBoolean(ParquetInputFormat.BLOOM_FILTERING_ENABLED, true)) { + return statsFiltered; + } + BloomFilterPlan plan = bloomFilterPlan(filter); + if (!plan.canPrune()) { + return statsFiltered; + } + SortedMap ranges = bloomFilterRanges(plan.columnsRead(), statsFiltered); + if (ranges != null && ranges.isEmpty()) { + return statsFiltered; + } + if (ranges == null && isCacheOnlyRead()) { + // the filters cannot be served from the cache, and the file may not be read to get them + return statsFiltered; + } + + Map cached = null; + try { + cached = (ranges == null) ? null : cachedBloomFilters(ranges); + try (ParquetFileReader bloomFilterReader = openBloomFilterReader(fileMetaData, statsFiltered, ranges, + cached)) { + return RowGroupFilter.filterRowGroups( + Collections.singletonList(RowGroupFilter.FilterLevel.BLOOMFILTER), filter, statsFiltered, + bloomFilterReader); + } + } catch (Exception e) { + if (isCacheOnlyRead()) { + // a miss is the answer the caller asked for, as it is of every other read this mode makes + throw e; + } + // the row groups these would have dropped are read instead, which costs time and not correctness + LOG.warn("Skipping the bloom filters of " + filePath + ", reading the row groups statistics kept", e); + return statsFiltered; + } finally { + if (cached != null) { + cached.values().forEach(metadataCache::decRefBuffer); + } + } + } + + /** + * A reader over the bloom filters this predicate needs, served from the LLAP metadata cache where there + * is one and read from the file where there is not. + */ + private ParquetFileReader openBloomFilterReader(FileMetaData fileMetaData, List blocks, + SortedMap ranges, Map cached) throws IOException { + ParquetMetadata metadata = new ParquetMetadata(fileMetaData, blocks); + // Without a record filter: parquet filters row groups in the constructor otherwise, reading every + // bloom filter a second time and seeking to chunks that are not among the cached ranges. + ParquetReadOptions options = HadoopReadOptions.builder(jobConf, filePath) + .withRecordFilter(FilterCompat.NOOP).build(); + if (cached == null) { + return new ParquetFileReader(jobConf, filePath, metadata, options); + } + List cachedRanges = new ArrayList<>(ranges.size()); + for (Map.Entry range : ranges.entrySet()) { + cachedRanges.add(new ParquetFilterDataFromCache.Range(range.getKey(), range.getValue(), + cached.get(range.getKey()))); + } + ParquetFilterDataFromCache input = new ParquetFilterDataFromCache(cachedRanges, filePath, jobConf); + return ParquetFileReader.open(input, metadata, options, input); + } + + private Map cachedBloomFilters(SortedMap ranges) + throws IOException { + if (cacheKey == null || metadataCache == null) { + return null; + } + return LlapProxy.getIo().getParquetBloomFilterBuffersFromCache(filePath, jobConf, cacheKey, ranges); + } + + /** + * The bloom filters of these row groups that the predicate could prune on, as offset to length. Empty + * when none of the usable columns carries one, so the level can be skipped. Null when one records no + * length, as files written before Parquet stored it do: those are read from the file rather than cached. + */ + @VisibleForTesting + static SortedMap bloomFilterRanges(Set columns, List blocks) { + SortedMap ranges = new TreeMap<>(); + for (BlockMetaData block : blocks) { + for (ColumnChunkMetaData column : block.getColumns()) { + // The path is tested first because reading the offset of an encrypted chunk decrypts its metadata, + // which throws when the query holds no key for that column. + if (!columns.contains(column.getPath())) { + continue; + } + long offset = column.getBloomFilterOffset(); + if (offset <= 0) { + continue; + } + int length = column.getBloomFilterLength(); + if (length <= 0) { + return null; + } + ranges.put(offset, length); + } + } + return ranges; + } + + /** + * What the bloom filter level can do with a predicate: the columns Parquet may read a filter for, and + * whether dropping a row group is possible at all. The two differ, and conflating them leaves a read + * unserved: an OR only drops a row group when both of its sides do, but BloomFilterImpl still reads the + * filter of a side that can prune, so its column has to be fetched even when the OR itself cannot prune. + */ + record BloomFilterPlan(Set columnsRead, boolean canPrune) { + private static final BloomFilterPlan NONE = new BloomFilterPlan(Set.of(), false); + } + + static BloomFilterPlan bloomFilterPlan(FilterCompat.Filter filter) { + if (!(filter instanceof FilterCompat.FilterPredicateCompat predicateFilter)) { + return BloomFilterPlan.NONE; + } + Set columnsRead = new HashSet<>(); + boolean canPrune = prunesWithBloomFilter(predicateFilter.getFilterPredicate(), columnsRead); + return new BloomFilterPlan(columnsRead, canPrune); + } + + /** + * Answers whether this predicate can drop a row group, adding the columns Parquet reads a filter for. + * A bloom filter only proves a value absent, so it serves equality and set membership and nothing else. + */ + private static boolean prunesWithBloomFilter(FilterPredicate predicate, Set columnsRead) { + switch (predicate) { + // eq(col, null) asks for nulls, which a bloom filter says nothing about + case Operators.Eq eq -> { + if (eq.getValue() == null) { + return false; + } + columnsRead.add(eq.getColumn().getColumnPath()); + return true; + } + case Operators.In in -> { + columnsRead.add(in.getColumn().getColumnPath()); + return true; + } + // both sides are walked whatever they answer, since a side that can prune has its filter read + // even where the node above it cannot + case Operators.And and -> { + boolean left = prunesWithBloomFilter(and.getLeft(), columnsRead); + boolean right = prunesWithBloomFilter(and.getRight(), columnsRead); + return left || right; + } + case Operators.Or or -> { + boolean left = prunesWithBloomFilter(or.getLeft(), columnsRead); + boolean right = prunesWithBloomFilter(or.getRight(), columnsRead); + return left && right; + } + default -> { + return false; + } + } + } + + private boolean isCacheOnlyRead() { + return cacheKey != null && metadataCache != null + && HiveConf.getBoolVar(jobConf, HiveConf.ConfVars.LLAP_IO_CACHE_ONLY); + } + private Path wrapPathForCache(Path path, Object fileKey, JobConf configuration, List blocks, CacheTag tag) throws IOException { if (fileKey == null || cache == null) { diff --git a/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/AbstractTestParquetDirect.java b/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/AbstractTestParquetDirect.java index e99460f13f64..254eff566c22 100644 --- a/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/AbstractTestParquetDirect.java +++ b/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/AbstractTestParquetDirect.java @@ -28,6 +28,7 @@ import java.util.List; import java.util.Map; import java.util.Properties; +import java.util.function.Consumer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; @@ -94,22 +95,55 @@ public void write(Void record) { } } - public Path writeDirect(String name, MessageType type, DirectWriter writer) - throws IOException { + /** + * Writes a Parquet file whose writer can be configured, so tests can turn on features that are off by + * default, such as bloom filters. + */ + public Path writeDirect(String name, MessageType type, DirectWriter writer, + Consumer> configurer) throws IOException { File temp = tempDir.newFile(name + ".parquet"); temp.deleteOnExit(); temp.delete(); Path path = new Path(temp.getPath()); - ParquetWriter parquetWriter = new ParquetWriter(path, - new DirectWriteSupport(type, writer, new HashMap())); - parquetWriter.write(null); - parquetWriter.close(); + DirectBuilder builder = new DirectBuilder(path, type, writer, new HashMap()); + configurer.accept(builder); + try (ParquetWriter parquetWriter = builder.build()) { + parquetWriter.write(null); + } return path; } + private static class DirectBuilder extends ParquetWriter.Builder { + private final MessageType type; + private final DirectWriter writer; + private final Map metadata; + + private DirectBuilder(Path file, MessageType type, DirectWriter writer, Map metadata) { + super(file); + this.type = type; + this.writer = writer; + this.metadata = metadata; + } + + @Override + protected DirectBuilder self() { + return this; + } + + @Override + protected WriteSupport getWriteSupport(Configuration conf) { + return new DirectWriteSupport(type, writer, metadata); + } + } + + public Path writeDirect(String name, MessageType type, DirectWriter writer) + throws IOException { + return writeDirect(name, type, writer, builder -> { }); + } + public static ArrayWritable record(Writable... fields) { return new ArrayWritable(Writable.class, fields); } diff --git a/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/TestParquetRowGroupFilter.java b/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/TestParquetRowGroupFilter.java index 60872b3f0817..546ebb80dd17 100644 --- a/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/TestParquetRowGroupFilter.java +++ b/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/TestParquetRowGroupFilter.java @@ -24,16 +24,24 @@ import java.util.List; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.ql.exec.SerializationUtilities; +import org.apache.hadoop.hive.ql.exec.Utilities; +import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatchCtx; import org.apache.hadoop.hive.ql.io.parquet.read.ParquetRecordReaderWrapper; import org.apache.hadoop.hive.ql.io.parquet.serde.ArrayWritableObjectInspector; +import org.apache.hadoop.hive.ql.io.parquet.vector.VectorizedParquetRecordReader; import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc; import org.apache.hadoop.hive.ql.plan.ExprNodeConstantDesc; import org.apache.hadoop.hive.ql.plan.ExprNodeDesc; import org.apache.hadoop.hive.ql.plan.ExprNodeGenericFuncDesc; +import org.apache.hadoop.hive.ql.plan.MapWork; import org.apache.hadoop.hive.ql.plan.TableScanDesc; import org.apache.hadoop.hive.ql.udf.generic.GenericUDF; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPAnd; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPEqual; import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPGreaterThan; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPOr; import org.apache.hadoop.hive.serde2.ColumnProjectionUtils; import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector; import org.apache.hadoop.hive.serde2.typeinfo.StructTypeInfo; @@ -42,6 +50,8 @@ import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils; import org.apache.hadoop.mapred.FileSplit; import org.apache.hadoop.mapred.JobConf; +import org.apache.parquet.hadoop.ParquetInputFormat; +import org.apache.parquet.hadoop.metadata.BlockMetaData; import org.apache.parquet.io.api.RecordConsumer; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.MessageTypeParser; @@ -56,6 +66,7 @@ public class TestParquetRowGroupFilter extends AbstractTestParquetDirect { JobConf conf; String columnNames; String columnTypes; + private Path bloomTestPath; @Before public void initConf() throws Exception { @@ -125,6 +136,230 @@ public void write(RecordConsumer consumer) { Assert.assertEquals("row group is not filtered correctly", 0, recordReader.getFilteredBlocks().size()); } + /** + * A row group whose bloom filter proves the value is absent must be dropped even when the value sits + * inside the column's min/max range, where statistics alone cannot prune anything. + */ + @Test + public void testBloomFilterRowGroupFilterTakeEffect() throws Exception { + StructObjectInspector inspector = writeBloomFilterFile(); + + // present value, and one that statistics cannot rule out because it lies inside [0, 198] + Assert.assertEquals("row group with a value present in the bloom filter must be read", + 1, filteredBlocksForEquals(inspector, 50).size()); + Assert.assertEquals("row group must be dropped by the bloom filter", + 0, filteredBlocksForEquals(inspector, 51).size()); + + // the same predicate keeps the row group once bloom filtering is switched off + conf.setBoolean(ParquetInputFormat.BLOOM_FILTERING_ENABLED, false); + Assert.assertEquals("row group must survive on statistics alone", + 1, filteredBlocksForEquals(inspector, 51).size()); + } + + /** + * A bloom filter on a column the predicate never mentions cannot prune anything, so the row group must + * survive on statistics and the data file must not be consulted for it. + */ + @Test + public void testBloomFilterOnUnrelatedColumnDoesNotPrune() throws Exception { + columnNames = "intCol,otherCol"; + columnTypes = "int,int"; + StructObjectInspector inspector = getObjectInspector(columnNames, columnTypes); + MessageType fileSchema = MessageTypeParser.parseMessageType( + "message hive_schema {\n" + + " optional int32 intCol;\n" + + " optional int32 otherCol;\n" + + "}\n" + ); + + conf.set(ColumnProjectionUtils.READ_COLUMN_NAMES_CONF_STR, "intCol,otherCol"); + conf.set("columns", "intCol,otherCol"); + conf.set("columns.types", "int,int"); + + // bloom filter only on intCol; otherCol holds the same even values + bloomTestPath = writeDirect("BloomFilterOnUnrelatedColumn", fileSchema, + consumer -> { + for (int i = 0; i < 100; i++) { + consumer.startMessage(); + consumer.startField("intCol", 0); + consumer.addInteger(i * 2); + consumer.endField("intCol", 0); + consumer.startField("otherCol", 1); + consumer.addInteger(i * 2); + consumer.endField("otherCol", 1); + consumer.endMessage(); + } + }, + builder -> builder.withBloomFilterEnabled("intCol", true)); + + List children = Lists.newArrayList( + new ExprNodeColumnDesc(Integer.class, "otherCol", "T", false), new ExprNodeConstantDesc(51)); + ExprNodeGenericFuncDesc predicate = + new ExprNodeGenericFuncDesc(inspector, new GenericUDFOPEqual(), children); + conf.set(TableScanDesc.FILTER_EXPR_CONF_STR, SerializationUtilities.serializeExpression(predicate)); + + Assert.assertEquals("a bloom filter on another column must not prune this row group", + 1, filteredBlocksVectorized(inspector).size()); + } + + /** Writes a single-row-group file holding only even values 0..198, with a bloom filter on intCol. */ + private StructObjectInspector writeBloomFilterFile() throws Exception { + columnNames = "intCol"; + columnTypes = "int"; + StructObjectInspector inspector = getObjectInspector(columnNames, columnTypes); + MessageType fileSchema = MessageTypeParser.parseMessageType( + "message hive_schema {\n" + + " optional int32 intCol;\n" + + "}\n" + ); + + conf.set(ColumnProjectionUtils.READ_COLUMN_NAMES_CONF_STR, "intCol"); + conf.set("columns", "intCol"); + conf.set("columns.types", "int"); + + bloomTestPath = writeDirect("BloomFilterRowGroupFilterTakeEffect", fileSchema, + consumer -> { + for (int i = 0; i < 100; i++) { + consumer.startMessage(); + consumer.startField("intCol", 0); + consumer.addInteger(i * 2); + consumer.endField("intCol", 0); + consumer.endMessage(); + } + }, + builder -> builder.withBloomFilterEnabled("intCol", true)); + return inspector; + } + + /** + * The JIRA is about the vectorized reader, which is the only one parquet did not already apply the bloom + * filter level for, so assert the pruning on that reader and not just on the mapred one. + */ + @Test + public void testBloomFilterRowGroupFilterVectorized() throws Exception { + StructObjectInspector inspector = writeBloomFilterFile(); + + setEqualsPredicate(inspector, 50); + Assert.assertEquals("row group with a value present in the bloom filter must be read", + 1, filteredBlocksVectorized(inspector).size()); + + setEqualsPredicate(inspector, 51); + Assert.assertEquals("vectorized reader must drop the row group on the bloom filter", + 0, filteredBlocksVectorized(inspector).size()); + } + + /** + * A conjunction is dropped when either side drops it, so a range term alongside an equality term must not + * stop the bloom filter from being consulted. + */ + @Test + public void testBloomFilterUsedForConjunctionContainingEquality() throws Exception { + StructObjectInspector inspector = writeBloomFilterFile(); + + List equals = Lists.newArrayList( + new ExprNodeColumnDesc(Integer.class, "intCol", "T", false), new ExprNodeConstantDesc(51)); + List greater = Lists.newArrayList( + new ExprNodeColumnDesc(Integer.class, "intCol", "T", false), new ExprNodeConstantDesc(0)); + ExprNodeGenericFuncDesc conjunction = new ExprNodeGenericFuncDesc(inspector, new GenericUDFOPAnd(), + Lists.newArrayList( + new ExprNodeGenericFuncDesc(inspector, new GenericUDFOPEqual(), equals), + new ExprNodeGenericFuncDesc(inspector, new GenericUDFOPGreaterThan(), greater))); + conf.set(TableScanDesc.FILTER_EXPR_CONF_STR, SerializationUtilities.serializeExpression(conjunction)); + + Assert.assertEquals("bloom filter must still prune when the equality sits inside a conjunction", + 0, filteredBlocksVectorized(inspector).size()); + } + + /** + * IN reaches Parquet as a chain of ORed equalities, and a disjunction only drops a row group when every + * branch drops it. This is the case bloom filters are most useful for, so pin it. + */ + @Test + public void testBloomFilterUsedForInList() throws Exception { + StructObjectInspector inspector = writeBloomFilterFile(); + + Assert.assertEquals("row group must be dropped when no value of the IN list is in the bloom filter", + 0, filteredBlocksForIn(inspector, 51, 53).size()); + Assert.assertEquals("row group must be read when one value of the IN list is present", + 1, filteredBlocksForIn(inspector, 50, 51).size()); + } + + private List filteredBlocksForIn(StructObjectInspector inspector, int... values) + throws Exception { + List disjuncts = Lists.newArrayList(); + for (int value : values) { + disjuncts.add(new ExprNodeGenericFuncDesc(inspector, new GenericUDFOPEqual(), Lists.newArrayList( + new ExprNodeColumnDesc(Integer.class, "intCol", "T", false), new ExprNodeConstantDesc(value)))); + } + ExprNodeDesc predicate = disjuncts.get(0); + for (int i = 1; i < disjuncts.size(); i++) { + predicate = new ExprNodeGenericFuncDesc(inspector, new GenericUDFOPOr(), + Lists.newArrayList(predicate, disjuncts.get(i))); + } + conf.set(TableScanDesc.FILTER_EXPR_CONF_STR, + SerializationUtilities.serializeExpression((ExprNodeGenericFuncDesc) predicate)); + + return filteredBlocksVectorized(inspector); + } + + /** + * A range-only predicate cannot be served by a bloom filter; the row group must survive on statistics. + */ + @Test + public void testRangeOnlyPredicateKeepsRowGroup() throws Exception { + StructObjectInspector inspector = writeBloomFilterFile(); + + List children = Lists.newArrayList( + new ExprNodeColumnDesc(Integer.class, "intCol", "T", false), new ExprNodeConstantDesc(50)); + ExprNodeGenericFuncDesc predicate = + new ExprNodeGenericFuncDesc(inspector, new GenericUDFOPGreaterThan(), children); + conf.set(TableScanDesc.FILTER_EXPR_CONF_STR, SerializationUtilities.serializeExpression(predicate)); + + Assert.assertEquals("range predicate must be answered by statistics alone", + 1, filteredBlocksVectorized(inspector).size()); + } + + /** + * Bloom filter pruning only runs on the vectorized reader: Parquet applies every filter level itself for + * the mapred reader, whose context carries the pushed down predicate. So the bloom assertions have to go + * through VectorizedParquetRecordReader. + */ + private List filteredBlocksVectorized(StructObjectInspector inspector) throws Exception { + MapWork mapWork = new MapWork(); + VectorizedRowBatchCtx rbCtx = new VectorizedRowBatchCtx(); + rbCtx.init(inspector, new String[0]); + mapWork.setVectorMode(true); + mapWork.setVectorizedRowBatchCtx(rbCtx); + HiveConf.setBoolVar(conf, HiveConf.ConfVars.HIVE_VECTORIZATION_ENABLED, true); + HiveConf.setVar(conf, HiveConf.ConfVars.PLAN, "//tmp"); + Utilities.setMapWork(conf, mapWork); + + try (VectorizedParquetRecordReader reader = new VectorizedParquetRecordReader( + new FileSplit(bloomTestPath, 0, fileLength(bloomTestPath), (String[]) null), new JobConf(conf))) { + return reader.getFilteredBlocks(); + } + } + + private void setEqualsPredicate(StructObjectInspector inspector, int value) { + List children = Lists.newArrayList( + new ExprNodeColumnDesc(Integer.class, "intCol", "T", false), new ExprNodeConstantDesc(value)); + ExprNodeGenericFuncDesc predicate = + new ExprNodeGenericFuncDesc(inspector, new GenericUDFOPEqual(), children); + conf.set(TableScanDesc.FILTER_EXPR_CONF_STR, SerializationUtilities.serializeExpression(predicate)); + } + + private List filteredBlocksForEquals(StructObjectInspector inspector, int value) + throws Exception { + List children = Lists.newArrayList( + new ExprNodeColumnDesc(Integer.class, "intCol", "T", false), + new ExprNodeConstantDesc(value)); + ExprNodeGenericFuncDesc predicate = + new ExprNodeGenericFuncDesc(inspector, new GenericUDFOPEqual(), children); + conf.set(TableScanDesc.FILTER_EXPR_CONF_STR, SerializationUtilities.serializeExpression(predicate)); + + return filteredBlocksVectorized(inspector); + } + private ArrayWritableObjectInspector getObjectInspector(final String columnNames, final String columnTypes) { List columnTypeList = createHiveTypeInfoFrom(columnTypes); List columnNameList = createHiveColumnsFrom(columnNames); diff --git a/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/TestParquetFilterDataFromCache.java b/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/TestParquetFilterDataFromCache.java new file mode 100644 index 000000000000..fa0115f53216 --- /dev/null +++ b/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/TestParquetFilterDataFromCache.java @@ -0,0 +1,171 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hive.ql.io.parquet.vector; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.EOFException; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.List; + +import org.apache.hadoop.hive.common.io.encoded.MemoryBuffer; +import org.apache.hadoop.hive.common.io.encoded.MemoryBufferOrBuffers; +import org.junit.Test; + +/** + * The cached ranges are served at the offsets the file stores them at, so an error in the range or buffer + * bookkeeping hands Parquet a structurally valid but wrong bloom filter, which prunes the wrong row groups + * and silently drops rows. + */ +public class TestParquetFilterDataFromCache { + + private static final long FIRST_OFFSET = 1_000L; + private static final long SECOND_OFFSET = 50_000L; + private static final long THIRD_OFFSET = 90_000L; + + @Test + public void testEachRangeIsServedAtItsOwnOffset() throws IOException { + byte[] first = filled(16, (byte) 1); + byte[] second = filled(24, (byte) 2); + byte[] third = filled(8, (byte) 3); + ParquetFilterDataFromCache input = input( + range(FIRST_OFFSET, first), range(SECOND_OFFSET, second), range(THIRD_OFFSET, third)); + + assertArrayEquals(third, readAt(input, THIRD_OFFSET, third.length)); + assertArrayEquals(first, readAt(input, FIRST_OFFSET, first.length)); + assertArrayEquals(second, readAt(input, SECOND_OFFSET, second.length)); + } + + @Test + public void testReadCrossesTheBuffersOfOneRange() throws IOException { + byte[] whole = new byte[40]; + for (int i = 0; i < whole.length; ++i) { + whole[i] = (byte) i; + } + // one range the cache split across three buffers, as it does for anything over the max allocation + ParquetFilterDataFromCache input = input(range(FIRST_OFFSET, + Arrays.copyOfRange(whole, 0, 16), Arrays.copyOfRange(whole, 16, 32), Arrays.copyOfRange(whole, 32, 40))); + + assertArrayEquals(whole, readAt(input, FIRST_OFFSET, whole.length)); + // starting inside the second buffer and running into the third + assertArrayEquals(Arrays.copyOfRange(whole, 20, 36), readAt(input, FIRST_OFFSET + 20, 16)); + } + + @Test + public void testPositionTracksTheReads() throws IOException { + ParquetFilterDataFromCache input = input(range(FIRST_OFFSET, filled(16, (byte) 1))); + input.seek(FIRST_OFFSET + 4); + assertEquals(FIRST_OFFSET + 4, input.getPos()); + input.readFully(new byte[8], 0, 8); + assertEquals(FIRST_OFFSET + 12, input.getPos()); + } + + @Test + public void testSeekOutsideEveryRangeFails() throws IOException { + ParquetFilterDataFromCache input = + input(range(FIRST_OFFSET, filled(16, (byte) 1)), range(SECOND_OFFSET, filled(16, (byte) 2))); + for (long offset : new long[] { 0L, FIRST_OFFSET - 1, FIRST_OFFSET + 16, SECOND_OFFSET + 16 }) { + try { + input.seek(offset); + fail("seek to " + offset + " is outside every cached range and must not be served"); + } catch (IOException expected) { + assertTrue(expected.getMessage(), expected.getMessage().contains("outside the cached ranges")); + } + } + } + + @Test + public void testReadPastTheEndOfARangeDoesNotRunIntoTheNext() throws IOException { + ParquetFilterDataFromCache input = + input(range(FIRST_OFFSET, filled(16, (byte) 1)), range(SECOND_OFFSET, filled(16, (byte) 2))); + input.seek(FIRST_OFFSET); + try { + input.readFully(new byte[24], 0, 24); + fail("a read may not continue past the range it started in"); + } catch (EOFException expected) { + // the ranges are disjoint regions of the file, so the bytes after one are not the next one's + } + } + + private static byte[] readAt(ParquetFilterDataFromCache input, long offset, int length) + throws IOException { + input.seek(offset); + byte[] read = new byte[length]; + input.readFully(read, 0, length); + return read; + } + + private static byte[] filled(int length, byte value) { + byte[] bytes = new byte[length]; + Arrays.fill(bytes, value); + return bytes; + } + + private static ParquetFilterDataFromCache input(ParquetFilterDataFromCache.Range... ranges) { + return new ParquetFilterDataFromCache(List.of(ranges), null, null); + } + + private static ParquetFilterDataFromCache.Range range(long offset, byte[]... chunks) { + int length = 0; + for (byte[] chunk : chunks) { + length += chunk.length; + } + return new ParquetFilterDataFromCache.Range(offset, length, buffers(chunks)); + } + + private static MemoryBufferOrBuffers buffers(byte[]... chunks) { + MemoryBuffer[] wrapped = new MemoryBuffer[chunks.length]; + for (int i = 0; i < chunks.length; ++i) { + wrapped[i] = new HeapBuffer(chunks[i]); + } + return new MemoryBufferOrBuffers() { + @Override + public MemoryBuffer getSingleBuffer() { + return (wrapped.length == 1) ? wrapped[0] : null; + } + + @Override + public MemoryBuffer[] getMultipleBuffers() { + return (wrapped.length == 1) ? null : wrapped; + } + }; + } + + private static final class HeapBuffer implements MemoryBuffer { + private final ByteBuffer data; + + HeapBuffer(byte[] bytes) { + this.data = ByteBuffer.wrap(bytes); + } + + @Override + public ByteBuffer getByteBufferRaw() { + return data; + } + + @Override + public ByteBuffer getByteBufferDup() { + return data.duplicate(); + } + } +} diff --git a/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/TestVectorizedParquetBloomFilters.java b/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/TestVectorizedParquetBloomFilters.java new file mode 100644 index 000000000000..9e730085250d --- /dev/null +++ b/ql/src/test/org/apache/hadoop/hive/ql/io/parquet/vector/TestVectorizedParquetBloomFilters.java @@ -0,0 +1,145 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hive.ql.io.parquet.vector; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.parquet.filter2.compat.FilterCompat; +import org.apache.parquet.filter2.predicate.FilterApi; +import org.apache.parquet.filter2.predicate.FilterPredicate; +import org.apache.parquet.filter2.predicate.Operators.IntColumn; +import org.apache.parquet.hadoop.metadata.BlockMetaData; +import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; +import org.apache.parquet.hadoop.metadata.ColumnPath; +import org.junit.Test; + +/** + * The gate deciding which bloom filters are worth reading, and so whether the data file is worth opening + * at all. The block count alone cannot see an error here: Parquet answers a filter read for a column that + * has none with null and leaves the row group standing, so a predicate collected too widely prunes exactly + * as much while reading megabytes it did not need. + */ +public class TestVectorizedParquetBloomFilters { + + private static final ColumnPath INT_COL = ColumnPath.get("intCol"); + private static final ColumnPath OTHER_COL = ColumnPath.get("otherCol"); + private static final IntColumn INT_COLUMN = FilterApi.intColumn("intCol"); + private static final IntColumn OTHER_COLUMN = FilterApi.intColumn("otherCol"); + private static final ColumnPath THIRD_COL = ColumnPath.get("thirdCol"); + private static final IntColumn THIRD_COLUMN = FilterApi.intColumn("thirdCol"); + + @Test + public void testEqualityIsCollectedAndRangesAreNot() { + assertEquals(Set.of(INT_COL), columnsRead(FilterApi.eq(INT_COLUMN, 51))); + // a bloom filter only proves a value absent, so it says nothing about ranges + assertEquals(Set.of(), columnsRead(FilterApi.lt(INT_COLUMN, 51))); + assertEquals(Set.of(), columnsRead(FilterApi.gtEq(INT_COLUMN, 51))); + } + + @Test + public void testNullEqualityIsNotCollected() { + // eq(col, null) asks for nulls, which a bloom filter says nothing about + assertEquals(Set.of(), columnsRead(FilterApi.eq(INT_COLUMN, null))); + } + + @Test + public void testAndCollectsEitherSide() { + // an AND drops a row group when either side proves absence, so both sides are worth reading + assertEquals(Set.of(INT_COL, OTHER_COL), + columnsRead(FilterApi.and(FilterApi.eq(INT_COLUMN, 51), FilterApi.eq(OTHER_COLUMN, 7)))); + assertEquals(Set.of(INT_COL), + columnsRead(FilterApi.and(FilterApi.eq(INT_COLUMN, 51), FilterApi.lt(OTHER_COLUMN, 7)))); + } + + @Test + public void testOrPrunesOnlyWhenBothSidesDo() { + assertTrue(canPrune(FilterApi.or(FilterApi.eq(INT_COLUMN, 51), FilterApi.eq(OTHER_COLUMN, 7)))); + // one side a bloom filter cannot answer makes the OR itself unable to drop a row group + assertFalse(canPrune(FilterApi.or(FilterApi.eq(INT_COLUMN, 51), FilterApi.lt(OTHER_COLUMN, 7)))); + } + + /** + * Parquet reads the filter of a prunable side of an OR even when the OR cannot prune, so that column + * still has to be fetched. Collecting only what can prune leaves that read unserved. + */ + @Test + public void testUnprunableOrStillContributesItsColumns() { + FilterPredicate unprunableOr = FilterApi.or(FilterApi.eq(INT_COLUMN, 51), FilterApi.lt(OTHER_COLUMN, 7)); + assertEquals(Set.of(INT_COL), columnsRead(unprunableOr)); + + // the AND can prune through its other side, so the level runs and the OR's column is read + FilterPredicate predicate = FilterApi.and(unprunableOr, FilterApi.eq(THIRD_COLUMN, 3)); + assertTrue(canPrune(predicate)); + assertEquals(Set.of(INT_COL, THIRD_COL), columnsRead(predicate)); + } + + private static Set columnsRead(org.apache.parquet.filter2.predicate.FilterPredicate predicate) { + return VectorizedParquetRecordReader.bloomFilterPlan(FilterCompat.get(predicate)).columnsRead(); + } + + private static boolean canPrune(org.apache.parquet.filter2.predicate.FilterPredicate predicate) { + return VectorizedParquetRecordReader.bloomFilterPlan(FilterCompat.get(predicate)).canPrune(); + } + + @Test + public void testAChunkStatingNoFilterLengthLeavesTheWholeFileToThePlainReader() { + // a file written before the length was recorded states an offset it cannot say the extent of, and + // the cache is served by extent, so none of the file's filters can be served from it + assertNull("a chunk of unknown length gives up the file", + VectorizedParquetRecordReader.bloomFilterRanges(Set.of(INT_COL), + List.of(blockOf(chunk(INT_COL, 1024L, -1))))); + } + + @Test + public void testAChunkWithNoFilterIsPassedOver() { + // no filter was written for this column, which is not a reason to give up the ones that were + assertEquals("a chunk holding no filter contributes no range", + Map.of(2048L, 64), VectorizedParquetRecordReader.bloomFilterRanges(Set.of(INT_COL, OTHER_COL), + List.of(blockOf(chunk(INT_COL, -1L, -1), chunk(OTHER_COL, 2048L, 64))))); + } + + @Test + public void testOnlyTheColumnsAskedAboutContributeRanges() { + assertEquals("a column the predicate never named is not read", + Map.of(1024L, 32), VectorizedParquetRecordReader.bloomFilterRanges(Set.of(INT_COL), + List.of(blockOf(chunk(INT_COL, 1024L, 32), chunk(OTHER_COL, 4096L, 64))))); + } + + private static ColumnChunkMetaData chunk(ColumnPath path, long bloomOffset, int bloomLength) { + ColumnChunkMetaData chunk = mock(ColumnChunkMetaData.class); + when(chunk.getPath()).thenReturn(path); + when(chunk.getBloomFilterOffset()).thenReturn(bloomOffset); + when(chunk.getBloomFilterLength()).thenReturn(bloomLength); + return chunk; + } + + private static BlockMetaData blockOf(ColumnChunkMetaData... chunks) { + BlockMetaData block = mock(BlockMetaData.class); + when(block.getColumns()).thenReturn(List.of(chunks)); + return block; + } +}