Support rowgroup skipping based on parquet dict encoding

Author: jaystarshot

velox/connectors/hive/HiveConnectorUtil.cpp

@@ -426,7 +426,6 @@ std::shared_ptr<common::ScanSpec> makeScanSpec(
       subfieldSpecs.clear();
     }
   }
-
   for (auto& pair : filters) {
     const auto name = pair.first.toString();
     // SelectiveColumnReader doesn't support constant columns with filters,
@@ -444,7 +443,6 @@ std::shared_ptr<common::ScanSpec> makeScanSpec(
     VELOX_CHECK_NULL(spec->filter());
     fieldSpec->setFilter(pair.second);
   }
-
   if (disableStatsBasedFilterReorder) {
     spec->disableStatsBasedFilterReorder();
   }

velox/connectors/hive/HiveDataSource.cpp

@@ -123,17 +123,18 @@ HiveDataSource::HiveDataSource(
     checkColumnNameLowerCase(hiveTableHandle_->subfieldFilters(), infoColumns_);
     checkColumnNameLowerCase(hiveTableHandle_->remainingFilter());
   }
-
   for (const auto& [k, v] : hiveTableHandle_->subfieldFilters()) {
     filters_.emplace(k.clone(), v);
   }
+
   double sampleRate = 1;
   auto remainingFilter = extractFiltersFromRemainingFilter(
       hiveTableHandle_->remainingFilter(),
       expressionEvaluator_,
       false,
       filters_,
       sampleRate);
+
   if (sampleRate != 1) {
     randomSkip_ = std::make_shared<random::RandomSkipTracker>(sampleRate);
   }

velox/connectors/hive/SplitReader.cpp

@@ -386,6 +386,7 @@ void SplitReader::createRowReader(
     RowTypePtr rowType,
     std::optional<bool> rowSizeTrackingEnabled) {
   VELOX_CHECK_NULL(baseRowReader_);
+
   configureRowReaderOptions(
       hiveTableHandle_->tableParameters(),
       scanSpec_,

velox/dwio/parquet/reader/Metadata.cpp

@@ -219,6 +219,18 @@ ColumnChunkMetaDataPtr::getColumnStatistics(
       thriftColumnChunkPtr(ptr_)->meta_data.statistics, *type, numRows);
 };
 
+bool ColumnChunkMetaDataPtr::hasEncodingStats() const {
+  return hasMetadata() && thriftColumnChunkPtr(ptr_)->meta_data.__isset.encoding_stats;
+}
+
+const std::vector<thrift::PageEncodingStats>& ColumnChunkMetaDataPtr::getEncodingStats() const {
+  return thriftColumnChunkPtr(ptr_)->meta_data.encoding_stats;
+}
+
+const std::vector<thrift::Encoding::type>& ColumnChunkMetaDataPtr::getEncoding() const {
+  return thriftColumnChunkPtr(ptr_)->meta_data.encodings;
+}
+
 std::string ColumnChunkMetaDataPtr::getColumnMetadataStatsMinValue() {
   VELOX_CHECK(hasStatistics());
   return thriftColumnChunkPtr(ptr_)->meta_data.statistics.min_value;
@@ -243,6 +255,18 @@ int64_t ColumnChunkMetaDataPtr::dictionaryPageOffset() const {
   return thriftColumnChunkPtr(ptr_)->meta_data.dictionary_page_offset;
 }
 
+bool ColumnChunkMetaDataPtr::hasBloomFilter() const {
+  return hasMetadata() &&
+      thriftColumnChunkPtr(ptr_)->meta_data.__isset.bloom_filter_offset;
+}
+
+std::optional<int64_t> ColumnChunkMetaDataPtr::bloom_filter_offset() const {
+  if (hasBloomFilter()) {
+    return thriftColumnChunkPtr(ptr_)->meta_data.bloom_filter_offset;
+  }
+  return std::nullopt;
+}
+
 common::CompressionKind ColumnChunkMetaDataPtr::compression() const {
   return thriftCodecToCompressionKind(
       thriftColumnChunkPtr(ptr_)->meta_data.codec);

velox/dwio/parquet/reader/Metadata.h

@@ -18,6 +18,7 @@
 
 #include "velox/dwio/common/Statistics.h"
 #include "velox/dwio/common/compression/Compression.h"
+#include "velox/dwio/parquet/thrift/ParquetThriftTypes.h"
 
 namespace facebook::velox::parquet {
 
@@ -37,6 +38,13 @@ class ColumnChunkMetaDataPtr {
   /// Check the presence of the dictionary page offset in ColumnChunk metadata.
   bool hasDictionaryPageOffset() const;
 
+  bool hasEncodingStats() const;
+
+  const std::vector<thrift::PageEncodingStats>& getEncodingStats() const;
+
+  const std::vector<thrift::Encoding::type>& getEncoding() const;
+
+
   /// Return the ColumnChunk statistics.
   std::unique_ptr<dwio::common::ColumnStatistics> getColumnStatistics(
       const TypePtr type,
@@ -61,6 +69,13 @@ class ColumnChunkMetaDataPtr {
   /// Must check for its presence using hasDictionaryPageOffset().
   int64_t dictionaryPageOffset() const;
 
+  /// Check if bloom filter is available.
+  bool hasBloomFilter() const;
+
+  /// The bloom filter offset.
+  /// Must check for its presence using hasBloomFilter().
+  std::optional<int64_t> bloom_filter_offset() const;
+
   /// The compression.
   common::CompressionKind compression() const;
 

velox/dwio/parquet/reader/PageReader.h

@@ -149,6 +149,31 @@ class PageReader {
     return sessionTimezone_;
   }
 
+  // Access the loaded dictionary for filtering purposes
+  const dwio::common::DictionaryValues& dictionary() const {
+    return dictionary_;
+  }
+
+  // Starts iterating over 'rows', which may span multiple pages. 'rows' are
+  // relative to current position, with 0 meaning the first
+  // unprocessed value in the current page, i.e. the row after the
+  // last row touched on a previous call to skip() or
+  // readWithVisitor(). This is the first row of the first data page
+  // if first call.
+  void startVisit(folly::Range<const vector_size_t*> rows);
+
+  // Seeks to page containing 'row'. Returns the number of rows to skip on
+  // the page to get to 'row'.  Clears the state and positions the stream and
+  // initializes a decoder for the found page. row kRepDefOnly means
+  // getting repdefs for the next page. If non-top level column, 'row'
+  // is interpreted in terms of leaf rows, including leaf
+  // nulls. Seeking ahead of pages covered by decodeRepDefs is not
+  // allowed for non-top level columns.
+  void seekToPage(int64_t row);
+
+  // Prepares dictionary from a dictionary page header
+  void prepareDictionary(const thrift::PageHeader& pageHeader);
+
  private:
   // Indicates that we only want the repdefs for the next page. Used when
   // prereading repdefs with seekToPage.
@@ -174,15 +199,6 @@ class PageReader {
   // 'pageData_' + 'encodedDataSize_'.
   void makedecoder();
 
-  // Reads and skips pages until finding a data page that contains
-  // 'row'. Reads and sets 'rowOfPage_' and 'numRowsInPage_' and
-  // initializes a decoder for the found page. row kRepDefOnly means
-  // getting repdefs for the next page. If non-top level column, 'row'
-  // is interpreted in terms of leaf rows, including leaf
-  // nulls. Seeking ahead of pages covered by decodeRepDefs is not
-  // allowed for non-top level columns.
-  void seekToPage(int64_t row);
-
   // Preloads the repdefs for the column chunk. To avoid preloading,
   // would need a way too clone the input stream so that one stream
   // reads ahead for repdefs and the other tracks the data. This is
@@ -202,7 +218,6 @@ class PageReader {
 
   void prepareDataPageV1(const thrift::PageHeader& pageHeader, int64_t row);
   void prepareDataPageV2(const thrift::PageHeader& pageHeader, int64_t row);
-  void prepareDictionary(const thrift::PageHeader& pageHeader);
   void makeDecoder();
 
   // For a non-top level leaf, reads the defs and sets 'leafNulls_' and
@@ -230,14 +245,6 @@ class PageReader {
     return data;
   }
 
-  // Starts iterating over 'rows', which may span multiple pages. 'rows' are
-  // relative to current position, with 0 meaning the first
-  // unprocessed value in the current page, i.e. the row after the
-  // last row touched on a previous call to skip() or
-  // readWithVisitor(). This is the first row of the first data page
-  // if first call.
-  void startVisit(folly::Range<const vector_size_t*> rows);
-
   // Seeks to the next page in a range given by startVisit().  Returns
   // true if there are unprocessed rows in the set given to
   // startVisit(). Seeks 'this' to the appropriate page and sets
@@ -318,6 +325,7 @@ class PageReader {
       }
     } else {
       if (isDictionary()) {
+        //  LOG(DEBUG) << "PageReader: Using StringDictionaryColumnVisitor for row filtering (without nulls) - RowGroup:" << rowGroupIndex_ << " PageOrdinal:" << pageOrdinal_ << " PageIndex:" << pageIndex_;
         auto dictVisitor = visitor.toStringDictionaryColumnVisitor();
         dictionaryIdDecoder_->readWithVisitor<false>(nullptr, dictVisitor);
       } else if (encoding_ == thrift::Encoding::DELTA_BYTE_ARRAY) {