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; + } +}