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 5 commits into
base: main
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
84415a5
feat(parquet): add statistics-based row group filtering
Sep 21, 2026
64cc97c
refactor
Sep 21, 2026
3938983
fix ci
Sep 22, 2026
782b503
fix ci
Sep 22, 2026
a1211cc
fix
Sep 22, 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 @@ -314,6 +314,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
24 changes: 19 additions & 5 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 @@ -28,6 +28,7 @@
#include "iceberg/arrow_c_data_guard_internal.h"
#include "iceberg/arrow_c_data_util_internal.h"
#include "iceberg/data/delete_filter.h"
#include "iceberg/expression/binder.h"
#include "iceberg/file_reader.h"
#include "iceberg/manifest/manifest_entry.h"
#include "iceberg/schema.h"
Expand Down Expand Up @@ -164,10 +165,21 @@ class FileScanTaskReader::Impl {
"Data file size must not be negative: {}",
data_file->file_size_in_bytes);

auto filter = task.residual_filter();
if (filter) {
ICEBERG_ASSIGN_OR_RAISE(auto is_bound, IsBoundVisitor::IsBound(filter));

@wgtmac wgtmac Sep 24, 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.

A normal scan can pass True as its residual, and constants can also be nested in And/Or. IsBoundVisitor currently returns an error for both constants, so these expressions fail before reading.

Java avoids this because its IsBoundVisitor returns null for constants, and Binder.BindVisitor handles constants directly. Please mirror that behavior. Simply returning true for constants is not sufficient: for And(True, unbound), that would mark the whole expression as bound and leave the unbound predicate unbound.

One concrete fix is a tri-state result:

// IsBoundVisitor
AlwaysTrue()  -> std::nullopt;  // constants only
AlwaysFalse() -> std::nullopt;

combine(left, right):
  if (!left) return right;
  if (!right) return left;
  if (*left != *right) return InvalidExpression("Found partially bound expression");
  return left;

Then Binder::Bind() handles constants and unbound predicates. Add tests for True, False, And(True, pred), and Or(False, pred).

if (!is_bound) {
ICEBERG_ASSIGN_OR_RAISE(
filter,
Binder::Bind(*table_schema_, filter,

@wgtmac wgtmac Sep 24, 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.

Please bind against the read/projection schema, not table_schema_. Java's GenericReader passes its read schema to Parquet.ReadBuilder, and ReadConf creates ParquetMetricsRowGroupFilter(expectedSchema, filter, caseSensitive). The scan projection includes filter fields through Binder.boundReferences().

To align with Java, build the read schema from projected_schema_ plus referenced filter fields, then bind against that schema. This also ensures any later residual evaluation has the required columns.

properties_.Get(ReaderProperties::kFilterCaseSensitive)));
}
}

if (task.delete_files().empty()) {
auto options = MakeReaderOptions(
*data_file, io_, projected_schema_, task.residual_filter(), name_mapping_,
properties_, data_file->first_row_id, data_file->data_sequence_number);
*data_file, io_, projected_schema_, filter, 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));
return MakeArrowArrayStream(std::move(reader));
Expand All @@ -191,9 +203,9 @@ class FileScanTaskReader::Impl {
ProjectionContext::Make(*required_schema, *projected_schema_,
project_batch_function));

auto options = MakeReaderOptions(
*data_file, io_, required_schema, task.residual_filter(), name_mapping_,
properties_, data_file->first_row_id, data_file->data_sequence_number);
auto options = MakeReaderOptions(*data_file, io_, required_schema, filter,
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 @@ -207,6 +219,7 @@ class FileScanTaskReader::Impl {
Impl(Options options, DeleteFilter::FieldLookup field_lookup,
std::shared_ptr<DeleteCounter> delete_counter)
: io_(std::move(options.io)),
table_schema_(std::move(options.table_schema)),
schemas_(std::move(options.schemas)),
projected_schema_(std::move(options.projected_schema)),
name_mapping_(std::move(options.name_mapping)),
Expand All @@ -215,6 +228,7 @@ class FileScanTaskReader::Impl {
delete_counter_(std::move(delete_counter)) {}

std::shared_ptr<FileIO> io_;
std::shared_ptr<Schema> table_schema_;
std::vector<std::shared_ptr<Schema>> schemas_;
std::shared_ptr<Schema> projected_schema_;
std::shared_ptr<NameMapping> name_mapping_;
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,11 @@ 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 Case sensitivity when binding unbound filter references.
inline static Entry<bool> kFilterCaseSensitive{"read.filter.case-sensitive", true};

@wgtmac wgtmac Sep 24, 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.

case_sensitive is scan/filter binding state, not a file-reader property. The current string property defaults to true and can diverge from a case-insensitive scan.

Please add explicit fields:

struct ReaderOptions {
  ...
  bool filter_case_sensitive = true;
};

class FileScanTaskReader {
 public:
  struct Options {
    ...
    bool filter_case_sensitive = true;
  };
};

Source it from the scan and pass it through:

FileScanTaskReader::Make({
    .io = scan->io(),
    .table_schema = scan->table()->schema(),
    .schemas = historical_schemas,
    .projected_schema = *scan->schema(),
    .filter_case_sensitive = scan->is_case_sensitive(),
});

Binder::Bind(*table_schema_, filter, options_.filter_case_sensitive);

MakeReaderOptions() should copy it into ReaderOptions, and ParquetReader::Open() should use options.filter_case_sensitive. Then remove ReaderProperties::kFilterCaseSensitive.

/// \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
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