Improve performance with bloom_filter_length

Author: zhangjiashen

parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java

@@ -541,6 +541,10 @@ private void addRowGroup(ParquetMetadata parquetMetadata, List<RowGroup> rowGrou
       if (bloomFilterOffset >= 0) {
         metaData.setBloom_filter_offset(bloomFilterOffset);
       }
+      int bloomFilterLength = columnMetaData.getBloomFilterLength();
+      if (bloomFilterLength >= 0) {
+        metaData.setBloom_filter_length(bloomFilterLength);
+      }
       if (columnMetaData.getStatistics() != null && !columnMetaData.getStatistics().isEmpty()) {
         metaData.setStatistics(toParquetStatistics(columnMetaData.getStatistics(), this.statisticsTruncateLength));
       }
@@ -1595,6 +1599,9 @@ public ParquetMetadata fromParquetMetadata(FileMetaData parquetMetadata,
             if (metaData.isSetBloom_filter_offset()) {
               column.setBloomFilterOffset(metaData.getBloom_filter_offset());
             }
+            if (metaData.isSetBloom_filter_length()) {
+              column.setBloomFilterLength(metaData.getBloom_filter_length());
+            }
           }  else { // column encrypted with column key
             // Metadata will be decrypted later, if this column is accessed
             EncryptionWithColumnKey columnKeyStruct = cryptoMetaData.getENCRYPTION_WITH_COLUMN_KEY();

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java

@@ -32,6 +32,7 @@
 import static org.apache.parquet.hadoop.ParquetFileWriter.PARQUET_COMMON_METADATA_FILE;
 import static org.apache.parquet.hadoop.ParquetFileWriter.PARQUET_METADATA_FILE;
 
+import java.io.ByteArrayInputStream;
 import java.io.Closeable;
 import java.io.IOException;
 import java.io.InputStream;
@@ -1347,11 +1348,24 @@ public BloomFilter readBloomFilter(ColumnChunkMetaData meta) throws IOException
       }
     }
 
-    // Read Bloom filter data header.
+    // Seek to Bloom filter offset.
     f.seek(bloomFilterOffset);
+
+    // Read Bloom filter length.
+    int bloomFilterLength = meta.getBloomFilterLength();
+
+    // If it is set, read Bloom filter header and bitset together.
+    // Otherwise, read Bloom filter header first and then bitset.
+    InputStream in = null;
+    if (bloomFilterLength > 0) {
+      byte[] headerAndBitSet = new byte[bloomFilterLength];
+      f.readFully(headerAndBitSet);
+      in = new ByteArrayInputStream(headerAndBitSet);
+    }
+
     BloomFilterHeader bloomFilterHeader;
     try {
-      bloomFilterHeader = Util.readBloomFilterHeader(f, bloomFilterDecryptor, bloomFilterHeaderAAD);
+      bloomFilterHeader = Util.readBloomFilterHeader(in != null ? in : f, bloomFilterDecryptor, bloomFilterHeaderAAD);
     } catch (IOException e) {
       LOG.warn("read no bloom filter");
       return null;
@@ -1373,9 +1387,13 @@ public BloomFilter readBloomFilter(ColumnChunkMetaData meta) throws IOException
     byte[] bitset;
     if (null == bloomFilterDecryptor) {
       bitset = new byte[numBytes];
-      f.readFully(bitset);
+      if (in != null) {
+        in.read(bitset);
+      } else {
+        f.readFully(bitset);
+      }
     } else {
-      bitset = bloomFilterDecryptor.decrypt(f, bloomFilterBitsetAAD);
+      bitset = bloomFilterDecryptor.decrypt(in != null ? in : f, bloomFilterBitsetAAD);
       if (bitset.length != numBytes) {
         throw new ParquetCryptoRuntimeException("Wrong length of decrypted bloom filter bitset");
       }

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java

@@ -1396,6 +1396,9 @@ private static void serializeBloomFilters(
           serializedBitset = bloomFilterEncryptor.encrypt(serializedBitset, bloomFilterBitsetAAD);
         }
         out.write(serializedBitset);
+
+        int length = (int)(out.getPos() - offset);
+        column.setBloomFilterLength(length);
       }
     }
   }

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/metadata/ColumnChunkMetaData.java

@@ -211,6 +211,7 @@ protected static boolean positiveLongFitsInAnInt(long value) {
   private IndexReference offsetIndexReference;
 
   private long bloomFilterOffset = -1;
+  private int bloomFilterLength = -1;
 
   protected ColumnChunkMetaData(ColumnChunkProperties columnChunkProperties) {
     this(null, columnChunkProperties);
@@ -332,6 +333,15 @@ public void setBloomFilterOffset(long bloomFilterOffset) {
     this.bloomFilterOffset = bloomFilterOffset;
   }
 
+  /**
+   * @param bloomFilterLength
+   *          the reference to the Bloom filter
+   */
+  @Private
+  public void setBloomFilterLength(int bloomFilterLength) {
+    this.bloomFilterLength = bloomFilterLength;
+  }
+
   /**
    * @return the offset to the Bloom filter or {@code -1} if there is no bloom filter for this column chunk
    */
@@ -341,6 +351,15 @@ public long getBloomFilterOffset() {
     return bloomFilterOffset;
   }
 
+  /**
+   * @return the length to the Bloom filter or {@code -1} if there is no bloom filter for this column chunk
+   */
+  @Private
+  public int getBloomFilterLength() {
+    decryptIfNeeded();
+    return bloomFilterLength;
+  }
+
   /**
    * @return all the encodings used in this column
    */
@@ -630,6 +649,9 @@ protected void decryptIfNeeded() {
     if (metaData.isSetBloom_filter_offset()) {
       setBloomFilterOffset(metaData.getBloom_filter_offset());
     }
+    if (metaData.isSetBloom_filter_length()) {
+      setBloomFilterLength(metaData.getBloom_filter_length());
+    }
   }
 
   @Override

parquet-hadoop/src/test/java/org/apache/parquet/format/converter/TestParquetMetadataConverter.java

@@ -274,6 +274,26 @@ public void testBloomFilterOffset() throws IOException {
     assertEquals(1234, convertedMetaData.getBlocks().get(0).getColumns().get(0).getBloomFilterOffset());
   }
 
+  @Test
+  public void testBloomFilterLength() throws IOException {
+    ParquetMetadata origMetaData = createParquetMetaData(null, Encoding.PLAIN);
+    ParquetMetadataConverter converter = new ParquetMetadataConverter();
+
+    // Without bloom filter length
+    FileMetaData footer = converter.toParquetMetadata(1, origMetaData);
+    assertFalse(footer.getRow_groups().get(0).getColumns().get(0).getMeta_data().isSetBloom_filter_length());
+    ParquetMetadata convertedMetaData = converter.fromParquetMetadata(footer);
+    assertTrue(convertedMetaData.getBlocks().get(0).getColumns().get(0).getBloomFilterLength() < 0);
+
+    // With bloom filter length
+    origMetaData.getBlocks().get(0).getColumns().get(0).setBloomFilterLength(1024);
+    footer = converter.toParquetMetadata(1, origMetaData);
+    assertTrue(footer.getRow_groups().get(0).getColumns().get(0).getMeta_data().isSetBloom_filter_length());
+    assertEquals(1024, footer.getRow_groups().get(0).getColumns().get(0).getMeta_data().getBloom_filter_length());
+    convertedMetaData = converter.fromParquetMetadata(footer);
+    assertEquals(1024, convertedMetaData.getBlocks().get(0).getColumns().get(0).getBloomFilterLength());
+  }
+
   @Test
   public void testLogicalTypesBackwardCompatibleWithConvertedTypes() {
     ParquetMetadataConverter parquetMetadataConverter = new ParquetMetadataConverter();

pom.xml

@@ -71,7 +71,7 @@
     <japicmp.version>0.18.1</japicmp.version>
     <shade.prefix>shaded.parquet</shade.prefix>
     <hadoop.version>3.3.5</hadoop.version>
-    <parquet.format.version>2.9.0</parquet.format.version>
+    <parquet.format.version>2.10.0-SNAPSHOT</parquet.format.version>
     <previous.version>1.13.0</previous.version>
     <thrift.executable>thrift</thrift.executable>
     <format.thrift.executable>${thrift.executable}</format.thrift.executable>