feat(parquet): add statistics-based row group filtering - #956
YinZheng-Sun wants to merge 5 commits into
Conversation
wgtmac
left a comment
There was a problem hiding this comment.
I haven't finished reviewing it yet, just post my finding so far. Please let me know what you think.
|
|
||
| auto filter = task.residual_filter(); | ||
| if (filter) { | ||
| ICEBERG_ASSIGN_OR_RAISE(auto is_bound, IsBoundVisitor::IsBound(filter)); |
There was a problem hiding this comment.
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 (value.IsNaN()) { | ||
| return kRowsMightMatch; | ||
| } | ||
| const auto bound = lower_bound ? MinValue(ref) : MaxValue(ref); |
There was a problem hiding this comment.
Parquet min/max exclude NaNs. A group with NaN and 2.0 gets pruned by key > 3.0, but NaN matches in Iceberg order. We are waiting for apache/arrow#50807 (by @HuaHuaY) to land so we can leverage stats of floating types when their column orders are IEEE754_TOTAL_ORDER and ignore other orders. For now, we can just skip filtering for floating types?
| 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}; |
There was a problem hiding this comment.
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.
| if (!is_bound) { | ||
| ICEBERG_ASSIGN_OR_RAISE( | ||
| filter, | ||
| Binder::Bind(*table_schema_, filter, |
There was a problem hiding this comment.
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.
What
Adds
ParquetMetricsRowGroupFilterto evaluate Iceberg predicates against Parquet footer statistics before reading row groups.Adds tests covering split intersection, non-contiguous row groups, physical row positions, unprojected filter columns, schema evolution, missing statistics, predicate binding, and delete handling.
Why
The Parquet reader previously selected row groups only by file split. Filter predicates could not eliminate row groups whose footer statistics ruled out matching rows.
Behavior change
read.parquet.row-group-filter.enabled.Testing
Built and ran
parquet_test,data_test,avro_test, andexpression_testlocally during implementation. Rebuilt and reranparquet_testanddata_testafter the binding and field-mapping refactors; both passed.The
INregression covers 201 input literals that deduplicate to 200 values, ensuring the limit applies to the bound set.