Skip to content

feat(parquet): add statistics-based row group filtering #956

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
YinZheng-Sun wants to merge 2 commits into
base: main
Choose a base branch
from
+1,516 −74
Open

feat(parquet): add statistics-based row group filtering #956

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2e387ad
feat(parquet): prune row groups with footer statistics
Oct 15, 2026
3835dfe
refactor(parquet): simplify split row group selection
Oct 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions src/iceberg/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -319,6 +319,7 @@ if(ICEBERG_BUILD_BUNDLE)
avro/avro_stream_internal.cc
parquet/parquet_data_util.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
Expand Down
12 changes: 8 additions & 4 deletions src/iceberg/data/file_scan_task_reader.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
#include <memory>
#include <optional>
#include <utility>
#include <vector>

#include "iceberg/arrow_c_data_guard_internal.h"
#include "iceberg/arrow_c_data_util_internal.h"
Expand All @@ -41,7 +40,7 @@ namespace {

ReaderOptions MakeReaderOptions(const DataFile& data_file, std::shared_ptr<FileIO> io,
std::shared_ptr<Schema> projection,
std::shared_ptr<Expression> filter,
std::shared_ptr<Expression> filter, bool case_sensitive,
std::shared_ptr<NameMapping> name_mapping,
ReaderProperties properties,
std::optional<int64_t> first_row_id,
Expand All @@ -52,6 +51,7 @@ ReaderOptions MakeReaderOptions(const DataFile& data_file, std::shared_ptr<FileI
.io = std::move(io),
.projection = std::move(projection),
.filter = std::move(filter),
.case_sensitive = case_sensitive,
.name_mapping = std::move(name_mapping),
.first_row_id = first_row_id,
.data_sequence_number = data_sequence_number,
Expand Down Expand Up @@ -170,9 +170,11 @@ class FileScanTaskReader::Impl {
data_file->file_path, task.start(), task.length());
}

auto filter = task.residual_filter();

if (task.delete_files().empty()) {
auto options = MakeReaderOptions(
*data_file, io_, projected_schema_, task.residual_filter(), name_mapping_,
*data_file, io_, projected_schema_, filter, case_sensitive_, name_mapping_,
properties_, data_file->first_row_id, data_file->data_sequence_number);
ICEBERG_ASSIGN_OR_RAISE(
auto reader, ReaderFactoryRegistry::Open(data_file->file_format, options));
Expand All @@ -198,7 +200,7 @@ class FileScanTaskReader::Impl {
project_batch_function));

auto options = MakeReaderOptions(
*data_file, io_, required_schema, task.residual_filter(), name_mapping_,
*data_file, io_, required_schema, filter, case_sensitive_, name_mapping_,
properties_, data_file->first_row_id, data_file->data_sequence_number);
ICEBERG_ASSIGN_OR_RAISE(auto reader,
ReaderFactoryRegistry::Open(data_file->file_format, options));
Expand All @@ -215,6 +217,7 @@ class FileScanTaskReader::Impl {
: io_(std::move(options.io)),
schemas_(std::move(options.schemas)),
projected_schema_(std::move(options.projected_schema)),
case_sensitive_(options.case_sensitive),
name_mapping_(std::move(options.name_mapping)),
properties_(ReaderProperties::FromMap(options.properties)),
field_lookup_(std::move(field_lookup)),
Expand All @@ -223,6 +226,7 @@ class FileScanTaskReader::Impl {
std::shared_ptr<FileIO> io_;
std::vector<std::shared_ptr<Schema>> schemas_;
std::shared_ptr<Schema> projected_schema_;
bool case_sensitive_;
std::shared_ptr<NameMapping> name_mapping_;
ReaderProperties properties_;
DeleteFilter::FieldLookup field_lookup_;
Expand Down
2 changes: 2 additions & 0 deletions src/iceberg/data/file_scan_task_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,8 @@ class ICEBERG_DATA_EXPORT FileScanTaskReader {
/// The output schema for the returned ArrowArrayStream. Must be a
/// projection of table_schema.
std::shared_ptr<Schema> projected_schema;
/// Case sensitivity for field name resolution. Pass TableScan::is_case_sensitive().
bool case_sensitive = true;
/// Optional name mapping for files written without field IDs.
std::shared_ptr<NameMapping> name_mapping;
/// Format-specific or implementation-specific options for data readers.
Expand Down
2 changes: 2 additions & 0 deletions src/iceberg/expression/evaluator.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@

#include "iceberg/expression/binder.h"
#include "iceberg/expression/expression_visitor.h"
#include "iceberg/expression/rewrite_not.h"
#include "iceberg/schema.h"
#include "iceberg/util/macros.h"

Expand Down Expand Up @@ -140,6 +141,7 @@ Result<std::unique_ptr<Evaluator>> Evaluator::Make(const Schema& schema,
std::shared_ptr<Expression> unbound,
bool case_sensitive) {
ICEBERG_ASSIGN_OR_RAISE(auto bound_expr, Binder::Bind(schema, unbound, case_sensitive));
ICEBERG_ASSIGN_OR_RAISE(bound_expr, RewriteNot::Visit(bound_expr));

@wgtmac wgtmac Oct 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These two evaluators now disagree on the same row.

Suppose a file uses identity(key) partitioning. Its partition value is null, and it contains this row:

key value
null 7

The filter is NOT(key < 100) OR value = 5.

Evaluator rewrites the filter to key >= 100 OR value = 5. Both checks return false for this row, so the row does not match.

ResidualEvaluator still checks the original filter:

  1. key < 100 returns false for the null partition value.
  2. NOT(false) becomes true.
  3. true OR value = 5 becomes true for the whole file.

This drops the value = 5 condition. If file statistics cannot reject the file, the reader gets a true residual. Any later filter using that residual will keep the row, even though Evaluator rejects it.

@YinZheng-Sun YinZheng-Sun Oct 13, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch, thanks!！

@advancedxy advancedxy Oct 15, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hmm. the rewrite expr doesn't seem to be the right place.

The evaluator class is the general expression evaluator, it should not rewrite not unconditionally? I think in the java side, the Evaluator just accept the unbound expr as it is, the corresponding caller, or metrics evaluator or xx evaluator rewrite not necessary.

@wgtmac wgtmac Oct 16, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For non-equal UUIDs, Literal::operator<=> returns unordered. This makes both u < A and u >= A false.

So for u = B where B > A, NOT(u < A) should be true, but this rewrite returns false. UUIDs should use unsigned byte-wise ordering before applying this rewrite. The same issue affects ResidualEvaluator.

return std::unique_ptr<Evaluator>(new Evaluator(std::move(bound_expr)));
}

Expand Down
2 changes: 2 additions & 0 deletions src/iceberg/expression/residual_evaluator.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
#include "iceberg/expression/expression.h"
#include "iceberg/expression/expression_visitor.h"
#include "iceberg/expression/predicate.h"
#include "iceberg/expression/rewrite_not.h"
#include "iceberg/partition_spec.h"
#include "iceberg/row/struct_like.h"
#include "iceberg/schema.h"
Expand Down Expand Up @@ -328,6 +329,7 @@ Result<std::unique_ptr<ResidualEvaluator>> ResidualEvaluator::Unpartitioned(
Result<std::unique_ptr<ResidualEvaluator>> ResidualEvaluator::Make(
std::shared_ptr<Expression> expr, const PartitionSpec& spec, const Schema& schema,
bool case_sensitive) {
ICEBERG_ASSIGN_OR_RAISE(expr, RewriteNot::Visit(std::move(expr)));
if (spec.fields().empty()) {
return Unpartitioned(std::move(expr));
}
Expand Down
5 changes: 5 additions & 0 deletions src/iceberg/file_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,9 @@ class ICEBERG_EXPORT ReaderProperties : public ConfigBase<ReaderProperties> {
/// Only the Parquet reader honors this option; other readers ignore it.
/// Default: false (use 32-bit offset list).
inline static Entry<bool> kArrowUseLargeList{"read.arrow.use-large-list", false};
/// \brief Use footer statistics 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.
/// When true, decode directly from Avro to Arrow without GenericDatum intermediate.
/// Default: true (skip GenericDatum for better performance).
Expand Down Expand Up @@ -107,6 +110,8 @@ struct ICEBERG_EXPORT ReaderOptions {
/// \brief The filter to apply to the data. Reader implementations may ignore this if
/// the file format does not support filtering.
std::shared_ptr<class Expression> filter;
/// \brief Case sensitivity inherited from the scan when resolving field names.
bool case_sensitive = true;
/// \brief Name mapping for schema evolution compatibility. Used when reading files
/// that may have different field names than the current schema.
std::shared_ptr<class NameMapping> name_mapping;
Expand Down
90 changes: 45 additions & 45 deletions src/iceberg/parquet/parquet_metrics.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,47 +191,6 @@ bool NeedsBoundTruncation(const PrimitiveType& type) {
return type.type_id() == TypeId::kString || type.type_id() == TypeId::kBinary;
}

Result<Literal> StatsValueToLiteral(const ::parquet::ColumnDescriptor& column,
const PrimitiveType& iceberg_type,
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); });
case ::parquet::Type::INT32:
return TypedStatsLiteral<::parquet::Int32Statistics>(
stats, is_min,
[&](int32_t value) { return Int32StatsLiteral(value, iceberg_type); });
case ::parquet::Type::INT64:
return TypedStatsLiteral<::parquet::Int64Statistics>(
stats, is_min,
[&](int64_t value) { return Int64StatsLiteral(value, iceberg_type); });
case ::parquet::Type::FLOAT:
return TypedStatsLiteral<::parquet::FloatStatistics>(
stats, is_min,
[&](float value) { return FloatStatsLiteral(value, iceberg_type); });
case ::parquet::Type::DOUBLE:
return TypedStatsLiteral<::parquet::DoubleStatistics>(
stats, is_min, [](double value) { return Literal::Double(value); });
case ::parquet::Type::BYTE_ARRAY:
return TypedStatsLiteral<::parquet::ByteArrayStatistics>(
stats, is_min, [&](const ::parquet::ByteArray& value) {
return BinaryStatsLiteral(BytesFromByteArray(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);
});
case ::parquet::Type::INT96:
case ::parquet::Type::UNDEFINED:
return NotSupported("Cannot convert Parquet statistics for physical type {}",
static_cast<int>(column.physical_type()));
}
std::unreachable();
}

/// \brief Collect counts (value count and null count) from footer statistics.
/// \param field_id The Iceberg field ID.
/// \param metadata The Parquet file metadata.
Expand Down Expand Up @@ -288,15 +247,15 @@ Result<std::optional<FieldMetrics>> CollectBounds(
value_count += column_chunk->num_values();

if (stats->HasMinMax()) {
ICEBERG_ASSIGN_OR_RAISE(auto min_value,
StatsValueToLiteral(*column_desc, *iceberg_type, *stats,
ICEBERG_ASSIGN_OR_RAISE(auto min_value, ParquetMetrics::StatsValueToLiteral(
*column_desc, *iceberg_type, *stats,
/*is_min=*/true));
if (!lower_bound.has_value() || min_value < lower_bound.value()) {
lower_bound = std::move(min_value);
}

ICEBERG_ASSIGN_OR_RAISE(auto max_value,
StatsValueToLiteral(*column_desc, *iceberg_type, *stats,
ICEBERG_ASSIGN_OR_RAISE(auto max_value, ParquetMetrics::StatsValueToLiteral(
*column_desc, *iceberg_type, *stats,
/*is_min=*/false));
if (!upper_bound.has_value() || max_value > upper_bound.value()) {
upper_bound = std::move(max_value);
Expand Down Expand Up @@ -504,6 +463,47 @@ class CollectMetricsVisitor {

} // namespace

Result<Literal> ParquetMetrics::StatsValueToLiteral(
const ::parquet::ColumnDescriptor& column, const PrimitiveType& iceberg_type,
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); });
case ::parquet::Type::INT32:
return TypedStatsLiteral<::parquet::Int32Statistics>(
stats, is_min,
[&](int32_t value) { return Int32StatsLiteral(value, iceberg_type); });
case ::parquet::Type::INT64:
return TypedStatsLiteral<::parquet::Int64Statistics>(
stats, is_min,
[&](int64_t value) { return Int64StatsLiteral(value, iceberg_type); });
case ::parquet::Type::FLOAT:
return TypedStatsLiteral<::parquet::FloatStatistics>(
stats, is_min,
[&](float value) { return FloatStatsLiteral(value, iceberg_type); });
case ::parquet::Type::DOUBLE:
return TypedStatsLiteral<::parquet::DoubleStatistics>(
stats, is_min, [](double value) { return Literal::Double(value); });
case ::parquet::Type::BYTE_ARRAY:
return TypedStatsLiteral<::parquet::ByteArrayStatistics>(
stats, is_min, [&](const ::parquet::ByteArray& value) {
return BinaryStatsLiteral(BytesFromByteArray(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);
});
case ::parquet::Type::INT96:
case ::parquet::Type::UNDEFINED:
return NotSupported("Cannot convert Parquet statistics for physical type {}",
static_cast<int>(column.physical_type()));
}
std::unreachable();
}

Result<Metrics> ParquetMetrics::GetMetrics(
const Schema& schema, const ::parquet::SchemaDescriptor& parquet_schema,
const MetricsConfig& metrics_config, const ::parquet::FileMetaData& metadata,
Expand Down
6 changes: 6 additions & 0 deletions src/iceberg/parquet/parquet_metrics_internal.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,12 @@ class ParquetMetrics {
public:
ParquetMetrics() = delete;

/// Convert one footer bound to an Iceberg literal, including supported promotions.
static Result<Literal> StatsValueToLiteral(const ::parquet::ColumnDescriptor& column,
const PrimitiveType& iceberg_type,
const ::parquet::Statistics& stats,
bool is_min);

/// \brief Compute file-level metrics from Parquet file metadata.
///
/// This function extracts metrics including row count, column sizes, value counts,
Expand Down
Loading
Loading