feat(parquet): add dictionary row-group pruning

src/iceberg/CMakeLists.txt

@@ -326,11 +326,13 @@ if(ICEBERG_BUILD_BUNDLE)
       avro/avro_schema_util.cc
       avro/avro_stream_internal.cc
       parquet/parquet_data_util.cc
+      parquet/parquet_dictionary_row_group_filter.cc
       parquet/parquet_metrics.cc
       parquet/parquet_metrics_row_group_filter.cc
       parquet/parquet_reader.cc
       parquet/parquet_register.cc
       parquet/parquet_schema_util.cc
+      parquet/parquet_value_util.cc
       parquet/parquet_writer.cc
       parquet/row_number_reader.cc)
 

src/iceberg/file_reader.h

@@ -88,7 +88,7 @@ class ICEBERG_EXPORT ReaderProperties : public ConfigBase<ReaderProperties> {
   /// \brief Validate Arrow batches before merge-on-read projection.
   /// Default: true. Disable only when the reader is trusted to produce valid arrays.
   inline static Entry<bool> kArrowValidateBatch{"read.arrow.validate-batch", true};
-  /// \brief Use footer statistics to prune Parquet row groups.
+  /// \brief Use footer statistics and dictionaries to prune Parquet row groups.
   inline static Entry<bool> kParquetRowGroupFilter{
       "read.parquet.row-group-filter.enabled", true};
   /// \brief Skip GenericDatum in Avro reader for better performance.

src/iceberg/parquet/parquet_dictionary_row_group_filter_internal.h

@@ -0,0 +1,56 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <unordered_map>
+
+#include <parquet/arrow/schema.h>
+#include <parquet/file_reader.h>
+
+#include "iceberg/iceberg_bundle_export.h"
+#include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg::parquet {
+
+// Inclusive dictionary filtering, following Java's
+// ParquetDictionaryRowGroupFilter. False proves that the row group cannot match;
+// true still requires evaluating the residual predicate on the returned rows.
+class ICEBERG_BUNDLE_EXPORT ParquetDictionaryRowGroupFilter {
+ public:
+  static Result<std::unique_ptr<ParquetDictionaryRowGroupFilter>> Make(
+      const std::shared_ptr<Expression>& filter,
+      const ::parquet::SchemaDescriptor& file_schema);
+
+  // Reads only dictionary-page byte ranges. Input and properties must match metadata.
+  Result<bool> ShouldRead(const ::parquet::arrow::SchemaManifest& manifest,
+                          const ::parquet::FileMetaData& metadata, int row_group_index,
+                          const std::shared_ptr<::arrow::io::RandomAccessFile>& input,
+                          const ::parquet::ReaderProperties& properties) const;
+
+ private:
+  ParquetDictionaryRowGroupFilter() = default;
+  std::shared_ptr<Expression> bound_;
+  std::unordered_map<int32_t, int> field_id_to_column_idx_;
+};
+
+}  // namespace iceberg::parquet

src/iceberg/parquet/parquet_metrics.cc

@@ -34,13 +34,12 @@
 
 #include "iceberg/expression/literal.h"
 #include "iceberg/parquet/parquet_metrics_internal.h"
+#include "iceberg/parquet/parquet_value_util_internal.h"
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
 #include "iceberg/type.h"
 #include "iceberg/util/checked_cast.h"
-#include "iceberg/util/decimal.h"
 #include "iceberg/util/truncate_util.h"
-#include "iceberg/util/uuid.h"
 #include "iceberg/util/visit_type.h"
 
 namespace iceberg::parquet {
@@ -86,99 +85,6 @@ Result<Literal> TypedStatsLiteral(const ::parquet::Statistics& stats, bool is_mi
   return converter(is_min ? typed_stats.min() : typed_stats.max());
 }
 
-std::vector<uint8_t> BytesFromByteArray(const ::parquet::ByteArray& value) {
-  return std::vector<uint8_t>{value.ptr, value.ptr + value.len};
-}
-
-std::vector<uint8_t> BytesFromFLBA(const ::parquet::FixedLenByteArray& value,
-                                   int32_t length) {
-  return std::vector<uint8_t>{value.ptr, value.ptr + length};
-}
-
-Literal DecimalLiteral(int128_t value, const PrimitiveType& iceberg_type) {
-  const auto& decimal_type = internal::checked_cast<const DecimalType&>(iceberg_type);
-  return Literal::Decimal(value, decimal_type.precision(), decimal_type.scale());
-}
-
-Result<Literal> DecimalLiteralFromBytes(std::span<const uint8_t> bytes,
-                                        const PrimitiveType& iceberg_type) {
-  ICEBERG_ASSIGN_OR_RAISE(auto decimal,
-                          Decimal::FromBigEndian(bytes.data(), bytes.size()));
-  return DecimalLiteral(decimal.value(), iceberg_type);
-}
-
-Result<Literal> BinaryStatsLiteral(std::vector<uint8_t> bytes,
-                                   const PrimitiveType& iceberg_type) {
-  switch (iceberg_type.type_id()) {
-    case TypeId::kString:
-      return Literal::String(std::string(bytes.begin(), bytes.end()));
-    case TypeId::kBinary:
-      return Literal::Binary(std::move(bytes));
-    case TypeId::kFixed:
-      return Literal::Fixed(std::move(bytes));
-    case TypeId::kUuid: {
-      ICEBERG_ASSIGN_OR_RAISE(auto uuid, Uuid::FromBytes(bytes));
-      return Literal::UUID(std::move(uuid));
-    }
-    case TypeId::kDecimal:
-      return DecimalLiteralFromBytes(bytes, iceberg_type);
-    default:
-      return InvalidArgument(
-          "Cannot convert Parquet binary statistics to Iceberg type {}",
-          iceberg_type.ToString());
-  }
-}
-
-Result<Literal> Int32StatsLiteral(int32_t value, const PrimitiveType& iceberg_type) {
-  switch (iceberg_type.type_id()) {
-    case TypeId::kInt:
-      return Literal::Int(value);
-    case TypeId::kLong:
-      return Literal::Long(value);
-    case TypeId::kDate:
-      return Literal::Date(value);
-    case TypeId::kDecimal:
-      return DecimalLiteral(value, iceberg_type);
-    default:
-      return InvalidArgument("Cannot convert Parquet INT32 statistics to Iceberg type {}",
-                             iceberg_type.ToString());
-  }
-}
-
-Result<Literal> Int64StatsLiteral(int64_t value, const PrimitiveType& iceberg_type) {
-  switch (iceberg_type.type_id()) {
-    case TypeId::kLong:
-      return Literal::Long(value);
-    case TypeId::kTime:
-      return Literal::Time(value);
-    case TypeId::kTimestamp:
-      return Literal::Timestamp(value);
-    case TypeId::kTimestampTz:
-      return Literal::TimestampTz(value);
-    case TypeId::kTimestampNs:
-      return Literal::TimestampNs(value);
-    case TypeId::kTimestampTzNs:
-      return Literal::TimestampTzNs(value);
-    case TypeId::kDecimal:
-      return DecimalLiteral(value, iceberg_type);
-    default:
-      return InvalidArgument("Cannot convert Parquet INT64 statistics to Iceberg type {}",
-                             iceberg_type.ToString());
-  }
-}
-
-Result<Literal> FloatStatsLiteral(float value, const PrimitiveType& iceberg_type) {
-  switch (iceberg_type.type_id()) {
-    case TypeId::kFloat:
-      return Literal::Float(value);
-    case TypeId::kDouble:
-      return Literal::Double(value);
-    default:
-      return InvalidArgument("Cannot convert Parquet FLOAT statistics to Iceberg type {}",
-                             iceberg_type.ToString());
-  }
-}
-
 bool IsFloatingType(const PrimitiveType& type) {
   return type.type_id() == TypeId::kFloat || type.type_id() == TypeId::kDouble;
 }
@@ -468,33 +374,34 @@ Result<Literal> ParquetMetrics::StatsValueToLiteral(
     const ::parquet::Statistics& stats, bool is_min) {
   switch (column.physical_type()) {
     case ::parquet::Type::BOOLEAN:
-      return TypedStatsLiteral<::parquet::BoolStatistics>(
-          stats, is_min, [](bool value) { return Literal::Boolean(value); });
+      return TypedStatsLiteral<::parquet::BoolStatistics>(stats, is_min, [&](bool value) {
+        return ParquetValueToLiteral(value, iceberg_type);
+      });
     case ::parquet::Type::INT32:
       return TypedStatsLiteral<::parquet::Int32Statistics>(
           stats, is_min,
-          [&](int32_t value) { return Int32StatsLiteral(value, iceberg_type); });
+          [&](int32_t value) { return ParquetValueToLiteral(value, iceberg_type); });
     case ::parquet::Type::INT64:
       return TypedStatsLiteral<::parquet::Int64Statistics>(
           stats, is_min,
-          [&](int64_t value) { return Int64StatsLiteral(value, iceberg_type); });
+          [&](int64_t value) { return ParquetValueToLiteral(value, iceberg_type); });
     case ::parquet::Type::FLOAT:
       return TypedStatsLiteral<::parquet::FloatStatistics>(
           stats, is_min,
-          [&](float value) { return FloatStatsLiteral(value, iceberg_type); });
+          [&](float value) { return ParquetValueToLiteral(value, iceberg_type); });
     case ::parquet::Type::DOUBLE:
       return TypedStatsLiteral<::parquet::DoubleStatistics>(
-          stats, is_min, [](double value) { return Literal::Double(value); });
+          stats, is_min,
+          [&](double value) { return ParquetValueToLiteral(value, iceberg_type); });
     case ::parquet::Type::BYTE_ARRAY:
       return TypedStatsLiteral<::parquet::ByteArrayStatistics>(
           stats, is_min, [&](const ::parquet::ByteArray& value) {
-            return BinaryStatsLiteral(BytesFromByteArray(value), iceberg_type);
+            return ParquetValueToLiteral(value, iceberg_type);
           });
     case ::parquet::Type::FIXED_LEN_BYTE_ARRAY:
       return TypedStatsLiteral<::parquet::FLBAStatistics>(
           stats, is_min, [&](const ::parquet::FixedLenByteArray& value) {
-            return BinaryStatsLiteral(BytesFromFLBA(value, column.type_length()),
-                                      iceberg_type);
+            return ParquetValueToLiteral(value, iceberg_type, column.type_length());
           });
     case ::parquet::Type::INT96:
     case ::parquet::Type::UNDEFINED:

src/iceberg/parquet/parquet_reader.cc

@@ -42,6 +42,7 @@
 #include "iceberg/arrow/metadata_column_util_internal.h"
 #include "iceberg/expression/binder.h"
 #include "iceberg/parquet/parquet_data_util_internal.h"
+#include "iceberg/parquet/parquet_dictionary_row_group_filter_internal.h"
 #include "iceberg/parquet/parquet_metrics_row_group_filter_internal.h"
 #include "iceberg/parquet/parquet_register.h"
 #include "iceberg/parquet/parquet_schema_util_internal.h"
@@ -133,6 +134,7 @@ class ParquetReader::Impl {
 
     read_schema_ = options.projection;
     stats_filter_.reset();
+    dictionary_filter_.reset();
 
     // Prepare reader properties
     ::parquet::ReaderProperties reader_properties(pool_);
@@ -165,9 +167,12 @@ class ParquetReader::Impl {
       }
       ICEBERG_ASSIGN_OR_RAISE(stats_filter_, ParquetMetricsRowGroupFilter::Make(
                                                  filter, *reader_->manifest().descr));
+      ICEBERG_ASSIGN_OR_RAISE(
+          dictionary_filter_,
+          ParquetDictionaryRowGroupFilter::Make(filter, *reader_->manifest().descr));
     }
 
-    ICEBERG_RETURN_UNEXPECTED(PlanRowGroups(options));
+    ICEBERG_RETURN_UNEXPECTED(PlanRowGroups(options, reader_properties));
 
     metadata_context_ = {.file_path = options.path,
                          .first_row_id = options.first_row_id,
@@ -257,7 +262,8 @@ class ParquetReader::Impl {
   }
 
  private:
-  Status PlanRowGroups(const ReaderOptions& options) {
+  Status PlanRowGroups(const ReaderOptions& options,
+                       const ::parquet::ReaderProperties& reader_properties) {
     auto metadata = reader_->parquet_reader()->metadata();
     if (options.split.has_value()) {
       const auto& split = *options.split;
@@ -300,6 +306,12 @@ class ParquetReader::Impl {
         auto row_group = metadata->RowGroup(row_group_index);
         ICEBERG_ASSIGN_OR_RAISE(
             auto should_read, stats_filter_->ShouldRead(reader_->manifest(), *row_group));
+        if (should_read && dictionary_filter_) {
+          ICEBERG_ASSIGN_OR_RAISE(
+              should_read, dictionary_filter_->ShouldRead(reader_->manifest(), *metadata,
+                                                          row_group_index, input_stream_,
+                                                          reader_properties));
+        }
         if (should_read) {
           row_group_indices_[selected_count++] = row_group_index;
         }
@@ -357,6 +369,7 @@ class ParquetReader::Impl {
   std::shared_ptr<::iceberg::Schema> read_schema_;
   // Footer-statistics filter for row group pruning, if enabled.
   std::unique_ptr<ParquetMetricsRowGroupFilter> stats_filter_;
+  std::unique_ptr<ParquetDictionaryRowGroupFilter> dictionary_filter_;
   // The projection result to apply to the read schema.
   SchemaProjection projection_;
   // The input stream to read Parquet file.