add test case

Author: jaystarshot

velox/dwio/parquet/reader/Metadata.cpp

@@ -220,14 +220,17 @@ ColumnChunkMetaDataPtr::getColumnStatistics(
 };
 
 bool ColumnChunkMetaDataPtr::hasEncodingStats() const {
-  return hasMetadata() && thriftColumnChunkPtr(ptr_)->meta_data.__isset.encoding_stats;
+  return hasMetadata() &&
+      thriftColumnChunkPtr(ptr_)->meta_data.__isset.encoding_stats;
 }
 
-const std::vector<thrift::PageEncodingStats>& ColumnChunkMetaDataPtr::getEncodingStats() const {
+const std::vector<thrift::PageEncodingStats>&
+ColumnChunkMetaDataPtr::getEncodingStats() const {
   return thriftColumnChunkPtr(ptr_)->meta_data.encoding_stats;
 }
 
-const std::vector<thrift::Encoding::type>& ColumnChunkMetaDataPtr::getEncoding() const {
+const std::vector<thrift::Encoding::type>& ColumnChunkMetaDataPtr::getEncoding()
+    const {
   return thriftColumnChunkPtr(ptr_)->meta_data.encodings;
 }
 

velox/dwio/parquet/reader/Metadata.h

@@ -44,7 +44,6 @@ class ColumnChunkMetaDataPtr {
 
   const std::vector<thrift::Encoding::type>& getEncoding() const;
 
-
   /// Return the ColumnChunk statistics.
   std::unique_ptr<dwio::common::ColumnStatistics> getColumnStatistics(
       const TypePtr type,

velox/dwio/parquet/reader/PageReader.h

@@ -325,7 +325,6 @@ 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) {

velox/dwio/parquet/reader/ParquetData.cpp

@@ -28,8 +28,8 @@ namespace facebook::velox::parquet {
 std::unique_ptr<dwio::common::FormatData> ParquetParams::toFormatData(
     const std::shared_ptr<const dwio::common::TypeWithId>& type,
     const common::ScanSpec& /*scanSpec*/) {
-  auto parquetData = std::make_unique<ParquetData>(
-      type, metaData_, pool(), sessionTimezone_);
+  auto parquetData =
+      std::make_unique<ParquetData>(type, metaData_, pool(), sessionTimezone_);
   // Set the BufferedInput if available
   if (bufferedInput_) {
     parquetData->setBufferedInput(bufferedInput_);
@@ -95,27 +95,27 @@ bool ParquetData::rowGroupMatches(
   if (columnChunk.hasStatistics()) {
     auto columnStats =
         columnChunk.getColumnStatistics(type, rowGroup.numRows());
-    bool statisticsResult = testFilter(filter, columnStats.get(), rowGroup.numRows(), type);
+    bool statisticsResult =
+        testFilter(filter, columnStats.get(), rowGroup.numRows(), type);
     if (!statisticsResult) {
       return false;
     }
   }
-
-
-    bool canUseDictionaryFiltering = isOnlyDictionaryEncodingPagesImpl(columnChunk);
-    if (canUseDictionaryFiltering) {
-      bool dictionaryResult = testFilterAgainstDictionary(rowGroupId, filter, columnChunk);
-      if (!dictionaryResult) {
-        return false;
-      }
+  bool canUseDictionaryFiltering =
+      isOnlyDictionaryEncodingPagesImpl(columnChunk);
+  if (canUseDictionaryFiltering) {
+    bool dictionaryResult =
+        testFilterAgainstDictionary(rowGroupId, filter, columnChunk);
+    if (!dictionaryResult) {
+      return false;
     }
-    return true;
+  }
+  return true;
 }
 
 void ParquetData::enqueueRowGroup(
     uint32_t index,
     dwio::common::BufferedInput& input) {
-  // Store the BufferedInput reference for creating streams on demand
   bufferedInput_ = &input;
 
   auto chunk = fileMetaDataPtr_.rowGroup(index).columnChunk(type_->column());
@@ -175,9 +175,11 @@ std::pair<int64_t, int64_t> ParquetData::getRowGroupRegion(
 }
 
 // Presto's exact isOnlyDictionaryEncodingPages function from PR #4779
-bool ParquetData::isOnlyDictionaryEncodingPagesImpl(const ColumnChunkMetaDataPtr& columnChunk) {
-  // Files written with newer versions of Parquet libraries (e.g. parquet-mr 1.9.0) will have EncodingStats available
-  // Otherwise, fallback to v1 logic
+bool ParquetData::isOnlyDictionaryEncodingPagesImpl(
+    const ColumnChunkMetaDataPtr& columnChunk) {
+  // Files written with newer versions of Parquet libraries (e.g.
+  // parquet-mr 1.9.0) will have EncodingStats available Otherwise, fallback to
+  // v1 logic
 
   // Check for EncodingStats when available (newer Parquet files)
   if (columnChunk.hasEncodingStats()) {
@@ -187,19 +189,22 @@ bool ParquetData::isOnlyDictionaryEncodingPagesImpl(const ColumnChunkMetaDataPtr
 
   // Fallback to v1 logic
   auto encodings = columnChunk.getEncoding();
-  std::set<thrift::Encoding::type> encodingSet(encodings.begin(), encodings.end());
+  std::set<thrift::Encoding::type> encodingSet(
+      encodings.begin(), encodings.end());
 
   if (encodingSet.count(thrift::Encoding::PLAIN_DICTIONARY)) {
     // PLAIN_DICTIONARY was present, which means at least one page was
     // dictionary-encoded and 1.0 encodings are used
-    // The only other allowed encodings are RLE and BIT_PACKED which are used for repetition or definition levels
+    // The only other allowed encodings are RLE and BIT_PACKED which are used
+    // for repetition or definition levels
     std::set<thrift::Encoding::type> allowedEncodings = {
-      thrift::Encoding::PLAIN_DICTIONARY,
-      thrift::Encoding::RLE,           // For repetition/definition levels
-      thrift::Encoding::BIT_PACKED     // For repetition/definition levels
+        thrift::Encoding::PLAIN_DICTIONARY,
+        thrift::Encoding::RLE, // For repetition/definition levels
+        thrift::Encoding::BIT_PACKED // For repetition/definition levels
     };
 
-    // Check if there are any disallowed encodings (equivalent to Sets.difference in Java)
+    // Check if there are any disallowed encodings (equivalent to
+    // Sets.difference in Java)
     for (const auto& encoding : encodings) {
       if (allowedEncodings.find(encoding) == allowedEncodings.end()) {
         return false;
@@ -212,7 +217,8 @@ bool ParquetData::isOnlyDictionaryEncodingPagesImpl(const ColumnChunkMetaDataPtr
 }
 
 // Helper methods for EncodingStats analysis (like Java Presto)
-bool ParquetData::hasDictionaryPages(const std::vector<thrift::PageEncodingStats>& stats) {
+bool ParquetData::hasDictionaryPages(
+    const std::vector<thrift::PageEncodingStats>& stats) {
   for (const auto& pageStats : stats) {
     if (pageStats.page_type == thrift::PageType::DICTIONARY_PAGE) {
       return true;
@@ -221,7 +227,8 @@ bool ParquetData::hasDictionaryPages(const std::vector<thrift::PageEncodingStats
   return false;
 }
 
-bool ParquetData::hasNonDictionaryEncodedPages(const std::vector<thrift::PageEncodingStats>& stats) {
+bool ParquetData::hasNonDictionaryEncodedPages(
+    const std::vector<thrift::PageEncodingStats>& stats) {
   for (const auto& pageStats : stats) {
     if (pageStats.page_type == thrift::PageType::DATA_PAGE ||
         pageStats.page_type == thrift::PageType::DATA_PAGE_V2) {
@@ -239,17 +246,6 @@ bool ParquetData::testFilterAgainstDictionary(
     uint32_t rowGroupId,
     const common::Filter* filter,
     const ColumnChunkMetaDataPtr& columnChunk) {
-    if (!filter) {
-      VLOG(3) << "    [DICT-TEST] No filter provided, dictionary test passes";
-    }
-
-  // Log file path for debugging
-  std::string filePath = "unknown";
-  if (bufferedInput_ && bufferedInput_->getReadFile()) {
-    filePath = bufferedInput_->getReadFile()->getName();
-  }
-
-  // Special handling for IsNull filters - conservative include before dictionary parsing
   if (filter->kind() == common::FilterKind::kIsNull) {
     return true; // Conservative include for IsNull filters
   }
@@ -259,48 +255,54 @@ bool ParquetData::testFilterAgainstDictionary(
     return true;
   }
 
-  // Use the dictionary directly - no temp storage needed
   auto numValues = dictionaryPtr->numValues;
   const void* dictPtr = dictionaryPtr->values->as<void>();
 
   // Test if any dictionary value passes the filter
   auto testDict = [&]<typename T>() {
     const T* dict = reinterpret_cast<const T*>(dictPtr);
 
-    // For larger dictionaries, we could use SIMD testValues() for better performance
-    // For now, use simple scalar approach which is sufficient for typical dictionary sizes
+    // For larger dictionaries, we could use SIMD testValues() for better
+    // performance For now, use simple scalar approach which is sufficient for
+    // typical dictionary sizes
     for (int32_t i = 0; i < numValues; ++i) {
-      if (common::applyFilter(*filter, dict[i])) return true;
+      if (common::applyFilter(*filter, dict[i]))
+        return true;
     }
     return false;
   };
 
   bool anyValuePasses = [&] {
     switch (type_->type()->kind()) {
-      case TypeKind::BIGINT:    return testDict.operator()<int64_t>();
-      case TypeKind::INTEGER:   return testDict.operator()<int32_t>();
+      case TypeKind::BIGINT:
+        return testDict.operator()<int64_t>();
+      case TypeKind::INTEGER:
+        return testDict.operator()<int32_t>();
       case TypeKind::VARCHAR:
-      case TypeKind::VARBINARY: return testDict.operator()<StringView>();
-      case TypeKind::REAL:      return testDict.operator()<float>();
-      case TypeKind::DOUBLE:    return testDict.operator()<double>();
-      case TypeKind::BOOLEAN:   return testDict.operator()<bool>();
-      default: return true;  // Conservative fallback
+      case TypeKind::VARBINARY:
+        return testDict.operator()<StringView>();
+      case TypeKind::REAL:
+        return testDict.operator()<float>();
+      case TypeKind::DOUBLE:
+        return testDict.operator()<double>();
+      case TypeKind::BOOLEAN:
+        return testDict.operator()<bool>();
+      default:
+        return true; // Conservative fallback
     }
   }();
-
-  // If no dictionary values pass the filter, but the filter accepts NULLs,
-  // we must conservatively include because the row group might contain NULLs that would match
   if (!anyValuePasses && filter->testNull()) {
     anyValuePasses = true;
   }
   return anyValuePasses;
 }
 
-// Read dictionary page directly for row group filtering (like Presto's dictionaryPredicatesMatch)
-std::unique_ptr<dwio::common::DictionaryValues> ParquetData::readDictionaryPageForFiltering(
+// Read dictionary page directly for row group filtering (like Presto's
+// dictionaryPredicatesMatch)
+std::unique_ptr<dwio::common::DictionaryValues>
+ParquetData::readDictionaryPageForFiltering(
     uint32_t rowGroupId,
     const ColumnChunkMetaDataPtr& columnChunk) {
-
   // Create input stream for the column chunk
   auto inputStream = getInputStream(rowGroupId, columnChunk);
   if (!inputStream) {
@@ -333,7 +335,8 @@ std::unique_ptr<dwio::common::DictionaryValues> ParquetData::readDictionaryPageF
 }
 // Helper method to get input stream for column chunk
 std::unique_ptr<dwio::common::SeekableInputStream> ParquetData::getInputStream(
-    uint32_t rowGroupId, const ColumnChunkMetaDataPtr& columnChunk) {
+    uint32_t rowGroupId,
+    const ColumnChunkMetaDataPtr& columnChunk) {
   // Use existing stream if available
   if (rowGroupId < streams_.size() && streams_[rowGroupId]) {
     return std::move(streams_[rowGroupId]);
@@ -346,12 +349,13 @@ std::unique_ptr<dwio::common::SeekableInputStream> ParquetData::getInputStream(
 
   // Calculate read parameters (same as enqueueRowGroup)
   uint64_t chunkReadOffset = columnChunk.dataPageOffset();
-  if (columnChunk.hasDictionaryPageOffset() && columnChunk.dictionaryPageOffset() >= 4) {
+  if (columnChunk.hasDictionaryPageOffset() &&
+      columnChunk.dictionaryPageOffset() >= 4) {
     chunkReadOffset = columnChunk.dictionaryPageOffset();
   }
 
-  uint64_t readSize =
-      (columnChunk.compression() == common::CompressionKind::CompressionKind_NONE)
+  uint64_t readSize = (columnChunk.compression() ==
+                       common::CompressionKind::CompressionKind_NONE)
       ? columnChunk.totalUncompressedSize()
       : columnChunk.totalCompressedSize();
 

velox/dwio/parquet/reader/ParquetData.h

@@ -76,7 +76,8 @@ class ParquetData : public dwio::common::FormatData {
         rowsInRowGroup_(-1),
         sessionTimezone_(sessionTimezone) {}
 
-  /// Sets the BufferedInput for creating streams on demand (called before filtering)
+  /// Sets the BufferedInput for creating streams on demand (called before
+  /// filtering)
   void setBufferedInput(dwio::common::BufferedInput* input) {
     bufferedInput_ = input;
   }
@@ -217,27 +218,33 @@ class ParquetData : public dwio::common::FormatData {
   /// stats in 'rowGroup'.
   bool rowGroupMatches(uint32_t rowGroupId, const common::Filter* filter);
 
-  /// Dictionary-based row group filtering (like Java Presto's dictionaryPredicatesMatch)
+  /// Dictionary-based row group filtering (like Java Presto's
+  /// dictionaryPredicatesMatch)
   bool testFilterAgainstDictionary(
       uint32_t rowGroupId,
       const common::Filter* filter,
       const ColumnChunkMetaDataPtr& columnChunk);
 
   /// Presto's exact isOnlyDictionaryEncodingPages function from PR #4779
-  bool isOnlyDictionaryEncodingPagesImpl(const ColumnChunkMetaDataPtr& columnChunk);
+  bool isOnlyDictionaryEncodingPagesImpl(
+      const ColumnChunkMetaDataPtr& columnChunk);
 
   /// Helper methods for EncodingStats analysis (like Java Presto)
   bool hasDictionaryPages(const std::vector<thrift::PageEncodingStats>& stats);
-  bool hasNonDictionaryEncodedPages(const std::vector<thrift::PageEncodingStats>& stats);
+  bool hasNonDictionaryEncodedPages(
+      const std::vector<thrift::PageEncodingStats>& stats);
 
-  /// Read dictionary page directly for row group filtering (like Presto's approach)
-  std::unique_ptr<dwio::common::DictionaryValues> readDictionaryPageForFiltering(
+  /// Read dictionary page directly for row group filtering (like Presto's
+  /// approach)
+  std::unique_ptr<dwio::common::DictionaryValues>
+  readDictionaryPageForFiltering(
       uint32_t rowGroupId,
       const ColumnChunkMetaDataPtr& columnChunk);
 
   /// Helper methods for dictionary page reading
   std::unique_ptr<dwio::common::SeekableInputStream> getInputStream(
-      uint32_t rowGroupId, const ColumnChunkMetaDataPtr& columnChunk);
+      uint32_t rowGroupId,
+      const ColumnChunkMetaDataPtr& columnChunk);
 
  protected:
   memory::MemoryPool& pool_;

velox/dwio/parquet/reader/ParquetReader.cpp

@@ -18,9 +18,9 @@
 
 #include <boost/algorithm/string.hpp>
 #include <glog/logging.h>
-#include <unordered_map>
 #include <thrift/TApplicationException.h>
 #include <thrift/protocol/TCompactProtocol.h> //@manual
+#include <unordered_map>
 
 #include "velox/dwio/parquet/reader/ParquetColumnReader.h"
 #include "velox/dwio/parquet/reader/StructColumnReader.h"
@@ -1237,7 +1237,8 @@ class ParquetRowReader::Impl {
     firstRowOfRowGroup_.reserve(rowGroups_.size());
 
     // Use regular row group filtering for now
-    // Dictionary-based filtering will be handled when dictionaries are naturally loaded during page reading
+    // Dictionary-based filtering will be handled when dictionaries are
+    // naturally loaded during page reading
     ParquetData::FilterRowGroupsResult res;
     columnReader_->filterRowGroups(0, parquetStatsContext_, res);