This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 187154675fe [improvement](be) Optimize projected fixed-width Parquet
predicate filtering (#65934)
187154675fe is described below
commit 187154675fec90fdfefdd93521d58a7722105fe1
Author: Gabriel <[email protected]>
AuthorDate: Thu Jul 23 13:26:03 2026 +0800
[improvement](be) Optimize projected fixed-width Parquet predicate
filtering (#65934)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #65921
Problem Summary:
When a fixed-width Parquet predicate column was also projected, Doris
materialized all selected predicate values and compacted the complete
column after filtering. The native decoder had already produced the
values needed by the raw predicate, so this added an avoidable
materialize-and-compact pass. The overhead is visible in TPC-DS Q88.
### What is changed?
Evaluate eligible predicates on decoded fixed-width values and append
only matching values to the projected column in the same decoder pass.
- Generalize the PLAIN-specific consumer and reader APIs into a
fixed-width raw filter-and-project path.
- Support PLAIN and BYTE_STREAM_SPLIT for identity-width INT32, INT64,
FLOAT, and DOUBLE values.
- Support DELTA_BINARY_PACKED for INT32 and INT64 values.
- Prevalidate advertised chunk encodings before consuming definition
levels.
- Reject an unexpected unsupported late-page encoding instead of
attempting a fallback after cursor progress.
- Keep the existing dictionary-id filter and dictionary materialization
path unchanged.
The raw decoder cannot rewind after predicate evaluation. Therefore,
when the predicate column is projected, survivors must be appended
before the encoded values and definition-level cursor are consumed.
Predicate-only columns retain their placeholder behavior. Nested
columns, converted logical types, residual/delete predicates,
unsupported expressions, and unsupported encodings continue to use the
existing materializing fallback.
### Microbenchmark
The reader microbenchmark from #65921 was run with a Release build, warm
fixture cache, CPU 8, a one-second minimum time, and 10 repetitions.
The matrix adds BYTE_STREAM_SPLIT and DELTA_BINARY_PACKED coverage with:
- 10% alternating NULLs;
- predicate-only and predicate-projected modes;
- 1%, 10%, 50%, and 90% selectivity.
All 16 combinations completed with the expected raw and selected row
counts.
Projected, 10% selectivity:
| Encoding | CPU ns/raw row, master | CPU ns/raw row, PR | Improvement |
CPU CV master / PR |
|---|---:|---:|---:|---:|
| BYTE_STREAM_SPLIT | 36.41 | 33.43 | 8.2% | 1.25% / 1.47% |
| DELTA_BINARY_PACKED | 38.75 | 36.84 | 4.9% | 5.40% / 2.94% |
The Delta baseline ran under sustained host load, so its result should
be treated as directional.
The original projected PLAIN results remain:
| Selectivity | CPU ns/raw row, master | CPU ns/raw row, PR |
Improvement |
|---:|---:|---:|---:|
| 1% | 32.80 | 30.80 | 6.1% |
| 10% | 33.95 | 31.82 | 6.3% |
| 50% | 37.76 | 35.05 | 7.2% |
| 90% | 42.06 | 37.66 | 10.5% |
### Release note
None
### Check List (For Author)
- Test: Unit Test and Manual test
- 17 targeted ASAN unit tests covering raw expression evaluation,
nullable mapping, projected PLAIN/BSS/Delta scans, unsupported-type
fallback, mixed-page safety, residual predicates, and benchmark matrix
constraints
- Release benchmark target linked successfully
- 16-case BSS/Delta reader microbenchmark matrix
- Before/after projected-filter microbenchmarks
- clang-format 16, check-format, and `git diff --check`
- clang-tidy was attempted but blocked by existing master/toolchain
errors: unmatched `NOLINTEND` in `be/src/core/types.h` and missing
`stddef.h` from the configured toolchain
- Behavior changed: No. This extends the existing raw predicate
optimization to equivalent fixed-width decoder output.
- Does this need documentation: No
---
be/benchmark/parquet/parquet_benchmark_scenarios.h | 13 ++
be/src/format_v2/parquet/parquet_profile.cpp | 12 +-
be/src/format_v2/parquet/parquet_profile.h | 8 +-
be/src/format_v2/parquet/parquet_scan.cpp | 49 ++++----
be/src/format_v2/parquet/reader/column_reader.cpp | 8 +-
be/src/format_v2/parquet/reader/column_reader.h | 14 ++-
.../parquet/reader/native/column_chunk_reader.cpp | 46 ++++---
.../parquet/reader/native/column_chunk_reader.h | 35 ++++--
.../parquet/reader/native/column_reader.cpp | 37 +++---
.../parquet/reader/native/column_reader.h | 26 ++--
.../parquet/reader/native_column_reader.cpp | 43 ++++---
.../parquet/reader/native_column_reader.h | 15 +--
be/test/format_v2/parquet/native_decoder_test.cpp | 35 +++++-
.../parquet/parquet_benchmark_scenarios_test.cpp | 19 +++
be/test/format_v2/parquet/parquet_scan_test.cpp | 133 +++++++++++++++++----
15 files changed, 355 insertions(+), 138 deletions(-)
diff --git a/be/benchmark/parquet/parquet_benchmark_scenarios.h
b/be/benchmark/parquet/parquet_benchmark_scenarios.h
index d420d9ccdd5..a01c955c6a2 100644
--- a/be/benchmark/parquet/parquet_benchmark_scenarios.h
+++ b/be/benchmark/parquet/parquet_benchmark_scenarios.h
@@ -126,6 +126,19 @@ inline std::vector<ReaderScenario> reader_scenarios() {
scenario.operation = ReaderOperation::PREDICATE_SCAN;
add(scenario);
}
+ for (const auto encoding : {Encoding::BYTE_STREAM_SPLIT,
Encoding::DELTA_BINARY_PACKED}) {
+ for (const int selectivity : {1, 10, 50, 90}) {
+ for (const auto projection :
+ {Projection::PREDICATE_ONLY,
Projection::PREDICATE_PROJECTED}) {
+ auto scenario = baseline;
+ scenario.operation = ReaderOperation::PREDICATE_SCAN;
+ scenario.encoding = encoding;
+ scenario.selectivity_percent = selectivity;
+ scenario.projection = projection;
+ add(scenario);
+ }
+ }
+ }
for (const int width : {4, 32, 128, 512}) {
for (const int predicate_position : {0, width - 1}) {
auto scenario = baseline;
diff --git a/be/src/format_v2/parquet/parquet_profile.cpp
b/be/src/format_v2/parquet/parquet_profile.cpp
index 9079b5f6060..b56a57d419e 100644
--- a/be/src/format_v2/parquet/parquet_profile.cpp
+++ b/be/src/format_v2/parquet/parquet_profile.cpp
@@ -175,10 +175,10 @@ void ParquetProfile::init(RuntimeProfile* profile) {
TUnit::BYTES,
parquet_profile, 1);
predicate_compaction_count = ADD_CHILD_COUNTER_WITH_LEVEL(profile,
"PredicateCompactionCount",
TUnit::UNIT,
parquet_profile, 1);
- plain_predicate_direct_batches = ADD_CHILD_COUNTER_WITH_LEVEL(
- profile, "PlainPredicateDirectBatches", TUnit::UNIT,
parquet_profile, 1);
- plain_predicate_direct_rows = ADD_CHILD_COUNTER_WITH_LEVEL(profile,
"PlainPredicateDirectRows",
- TUnit::UNIT,
parquet_profile, 1);
+ fixed_width_predicate_direct_batches = ADD_CHILD_COUNTER_WITH_LEVEL(
+ profile, "FixedWidthPredicateDirectBatches", TUnit::UNIT,
parquet_profile, 1);
+ fixed_width_predicate_direct_rows = ADD_CHILD_COUNTER_WITH_LEVEL(
+ profile, "FixedWidthPredicateDirectRows", TUnit::UNIT,
parquet_profile, 1);
dict_filter_rewrite_time =
ADD_CHILD_TIMER_WITH_LEVEL(profile, "DictFilterRewriteTime",
parquet_profile, 1);
dict_filter_expr_rewrite_time =
@@ -316,8 +316,8 @@ ParquetScanProfile ParquetProfile::scan_profile() const {
.predicate_compaction_time = predicate_compaction_time,
.predicate_compaction_bytes = predicate_compaction_bytes,
.predicate_compaction_count = predicate_compaction_count,
- .plain_predicate_direct_batches = plain_predicate_direct_batches,
- .plain_predicate_direct_rows = plain_predicate_direct_rows,
+ .fixed_width_predicate_direct_batches =
fixed_width_predicate_direct_batches,
+ .fixed_width_predicate_direct_rows =
fixed_width_predicate_direct_rows,
.dict_filter_rewrite_time = dict_filter_rewrite_time,
.dict_filter_expr_rewrite_time = dict_filter_expr_rewrite_time,
.dict_filter_read_dict_time = dict_filter_read_dict_time,
diff --git a/be/src/format_v2/parquet/parquet_profile.h
b/be/src/format_v2/parquet/parquet_profile.h
index f1b06be19d6..170f14b56c5 100644
--- a/be/src/format_v2/parquet/parquet_profile.h
+++ b/be/src/format_v2/parquet/parquet_profile.h
@@ -88,8 +88,8 @@ struct ParquetScanProfile {
RuntimeProfile::Counter* predicate_compaction_time = nullptr;
RuntimeProfile::Counter* predicate_compaction_bytes = nullptr;
RuntimeProfile::Counter* predicate_compaction_count = nullptr;
- RuntimeProfile::Counter* plain_predicate_direct_batches = nullptr;
- RuntimeProfile::Counter* plain_predicate_direct_rows = nullptr;
+ RuntimeProfile::Counter* fixed_width_predicate_direct_batches = nullptr;
+ RuntimeProfile::Counter* fixed_width_predicate_direct_rows = nullptr;
RuntimeProfile::Counter* dict_filter_rewrite_time = nullptr; // dictionary
rewrite time (ns)
RuntimeProfile::Counter* dict_filter_expr_rewrite_time =
nullptr; // expression/residual rewrite time (ns)
@@ -203,8 +203,8 @@ struct ParquetProfile {
RuntimeProfile::Counter* predicate_compaction_time = nullptr;
RuntimeProfile::Counter* predicate_compaction_bytes = nullptr;
RuntimeProfile::Counter* predicate_compaction_count = nullptr;
- RuntimeProfile::Counter* plain_predicate_direct_batches = nullptr;
- RuntimeProfile::Counter* plain_predicate_direct_rows = nullptr;
+ RuntimeProfile::Counter* fixed_width_predicate_direct_batches = nullptr;
+ RuntimeProfile::Counter* fixed_width_predicate_direct_rows = nullptr;
RuntimeProfile::Counter* dict_filter_rewrite_time = nullptr;
RuntimeProfile::Counter* dict_filter_expr_rewrite_time = nullptr;
RuntimeProfile::Counter* dict_filter_read_dict_time = nullptr;
diff --git a/be/src/format_v2/parquet/parquet_scan.cpp
b/be/src/format_v2/parquet/parquet_scan.cpp
index 5a4c343a5e9..47e7c4c3a68 100644
--- a/be/src/format_v2/parquet/parquet_scan.cpp
+++ b/be/src/format_v2/parquet/parquet_scan.cpp
@@ -1551,11 +1551,11 @@ Status
ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
auto read_predicate_column =
[&](ParquetColumnReader* column_reader, size_t block_position,
format::LocalColumnId local_id, const VExprContextSPtrs*
single_column_conjuncts,
- bool* used_dictionary_filter, bool* used_plain_filter) ->
Status {
+ bool* used_dictionary_filter, bool* used_fixed_width_filter)
-> Status {
DORIS_CHECK(used_dictionary_filter != nullptr);
- DORIS_CHECK(used_plain_filter != nullptr);
+ DORIS_CHECK(used_fixed_width_filter != nullptr);
*used_dictionary_filter = false;
- *used_plain_filter = false;
+ *used_fixed_width_filter = false;
DCHECK(remove_nullable(column_reader->type())
->equals(*remove_nullable(file_block->get_by_position(block_position).type)))
<< column_reader->type()->get_name() << " "
@@ -1597,8 +1597,7 @@ Status ParquetScanScheduler::read_filter_columns(int64_t
batch_rows,
}
if (single_column_conjuncts != nullptr &&
- !residual_predicate_positions.contains(block_position) &&
- request.is_predicate_only(local_id)) {
+ !residual_predicate_positions.contains(block_position)) {
VExprSPtrs direct_conjuncts;
direct_conjuncts.reserve(single_column_conjuncts->size());
std::ranges::transform(*single_column_conjuncts,
std::back_inserter(direct_conjuncts),
@@ -1607,13 +1606,19 @@ Status
ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
const uint16_t selected_rows_before = *selected_rows;
IColumn::Filter compact_filter;
bool used_filter = false;
- RETURN_IF_ERROR(column_reader->select_with_plain_filter(
+ const bool predicate_only =
request.is_predicate_only(local_id);
+ // The raw decoder cannot rewind after evaluating encoded
fixed-width values.
+ // Project survivors in that pass when output still needs the
predicate column.
+ IColumn* projected_column = predicate_only ? nullptr :
column.get();
+ RETURN_IF_ERROR(column_reader->select_with_fixed_width_filter(
*selection, *selected_rows, batch_rows,
direct_conjuncts,
- cast_set<int>(block_position), &compact_filter,
&used_filter));
+ cast_set<int>(block_position), projected_column,
&compact_filter,
+ &used_filter));
if (used_filter) {
DORIS_CHECK_EQ(compact_filter.size(),
selected_rows_before);
-
update_counter_if_not_null(_scan_profile.plain_predicate_direct_batches, 1);
-
update_counter_if_not_null(_scan_profile.plain_predicate_direct_rows,
+
update_counter_if_not_null(_scan_profile.fixed_width_predicate_direct_batches,
+ 1);
+
update_counter_if_not_null(_scan_profile.fixed_width_predicate_direct_rows,
selected_rows_before);
const uint16_t new_selected_rows =
count_selected_rows(compact_filter);
const auto filtered_rows =
static_cast<int64_t>(selected_rows_before) -
@@ -1625,15 +1630,19 @@ Status
ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
*selected_rows = apply_compact_filter_to_selection(
compact_filter, selection,
selected_rows_before);
}
- // This slot is absent from every residual/delete
conjunct, so no later
- // expression can observe its payload. Keep only the block
row-shape contract.
- auto placeholder = column->clone_empty();
- placeholder->insert_many_defaults(*selected_rows);
- file_block->replace_by_position(block_position,
std::move(placeholder));
+ if (predicate_only) {
+ // This slot is absent from every residual/delete
conjunct, so no later
+ // expression can observe its payload. Keep only the
block row-shape contract.
+ auto placeholder = column->clone_empty();
+ placeholder->insert_many_defaults(*selected_rows);
+ file_block->replace_by_position(block_position,
std::move(placeholder));
+ } else {
+ file_block->replace_by_position(block_position,
std::move(column));
+ }
read_column_positions.push_back(cast_set<uint32_t>(block_position));
remember_column_selection(cast_set<uint32_t>(block_position));
*predicate_columns_filtered = true;
- *used_plain_filter = true;
+ *used_fixed_width_filter = true;
return Status::OK();
}
}
@@ -1761,10 +1770,10 @@ Status
ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
auto position_it = request.local_positions.find(fid);
DORIS_CHECK(position_it != request.local_positions.end());
bool used_dictionary_filter = false;
- bool used_plain_filter = false;
+ bool used_fixed_width_filter = false;
RETURN_IF_ERROR(read_predicate_column(column_reader.get(),
position_it->second.value(),
fid, nullptr,
&used_dictionary_filter,
- &used_plain_filter));
+ &used_fixed_width_filter));
}
return Status::OK();
};
@@ -1803,21 +1812,21 @@ Status
ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
predicate_batch_sequence);
const int64_t start_ns = sample ? MonotonicNanos() : 0;
bool used_dictionary_filter = false;
- bool used_plain_filter = false;
+ bool used_fixed_width_filter = false;
const auto conjunct_it =
schedule.single_column_conjuncts.find(block_position);
const VExprContextSPtrs* column_conjuncts =
conjunct_it == schedule.single_column_conjuncts.end() ?
nullptr
:
&conjunct_it->second;
RETURN_IF_ERROR(read_predicate_column(reader_it->second.get(),
block_position, fid,
column_conjuncts,
&used_dictionary_filter,
- &used_plain_filter));
+ &used_fixed_width_filter));
if (*selected_rows != 0 && conjunct_it !=
schedule.single_column_conjuncts.end()) {
if (used_dictionary_filter) {
const auto residual_it =
_current_dictionary_residual_conjuncts.find(fid);
DORIS_CHECK(residual_it !=
_current_dictionary_residual_conjuncts.end());
RETURN_IF_ERROR(execute_scheduled_dictionary_residual_conjuncts_with_profile(
residual_it->second));
- } else if (!used_plain_filter) {
+ } else if (!used_fixed_width_filter) {
RETURN_IF_ERROR(execute_scheduled_conjuncts_with_profile(conjunct_it->second));
}
}
diff --git a/be/src/format_v2/parquet/reader/column_reader.cpp
b/be/src/format_v2/parquet/reader/column_reader.cpp
index f23559e5dbf..db5def75b8c 100644
--- a/be/src/format_v2/parquet/reader/column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/column_reader.cpp
@@ -81,10 +81,10 @@ Status
ParquetColumnReader::select_with_dictionary_filter(const SelectionVector&
name());
}
-Status ParquetColumnReader::select_with_plain_filter(const SelectionVector&,
uint16_t, int64_t,
- const VExprSPtrs&, int,
- IColumn::Filter*
row_filter,
- bool* used_filter) {
+Status ParquetColumnReader::select_with_fixed_width_filter(const
SelectionVector&, uint16_t,
+ int64_t, const
VExprSPtrs&, int,
+ IColumn*,
IColumn::Filter* row_filter,
+ bool* used_filter) {
DORIS_CHECK(row_filter != nullptr);
DORIS_CHECK(used_filter != nullptr);
row_filter->clear();
diff --git a/be/src/format_v2/parquet/reader/column_reader.h
b/be/src/format_v2/parquet/reader/column_reader.h
index bd4b73e3aa5..6914a5be7d1 100644
--- a/be/src/format_v2/parquet/reader/column_reader.h
+++ b/be/src/format_v2/parquet/reader/column_reader.h
@@ -61,12 +61,14 @@ public:
MutableColumnPtr& column,
IColumn::Filter* row_filter,
bool* used_filter);
- // Consume batch_rows and evaluate eligible fixed-width PLAIN values
without constructing a
- // predicate column. Implementations must leave the cursor untouched when
used_filter=false.
- virtual Status select_with_plain_filter(const SelectionVector& selection,
- uint16_t selected_rows, int64_t
batch_rows,
- const VExprSPtrs& conjuncts, int
column_id,
- IColumn::Filter* row_filter, bool*
used_filter);
+ // Consume batch_rows and evaluate eligible fixed-width values without
first constructing a
+ // complete predicate column. Append survivors when projected_column is
non-null. Implementations
+ // must leave the cursor untouched when used_filter=false.
+ virtual Status select_with_fixed_width_filter(const SelectionVector&
selection,
+ uint16_t selected_rows,
int64_t batch_rows,
+ const VExprSPtrs& conjuncts,
int column_id,
+ IColumn* projected_column,
+ IColumn::Filter* row_filter,
bool* used_filter);
// Native statistics are cumulative and can be recursively aggregated for
complex columns.
// Flush once at the scheduler batch boundary instead of snapshotting
after each operation.
diff --git a/be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp
b/be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp
index f24eab52952..c219a7aef03 100644
--- a/be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp
+++ b/be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp
@@ -653,14 +653,15 @@ Status decode_selected_nullable_values(IColumn& column,
const DataTypeSerDe& ser
return Status::OK();
}
-class PlainPredicateConsumer final : public ParquetFixedValueConsumer {
+class FixedWidthPredicateConsumer final : public ParquetFixedValueConsumer {
public:
- PlainPredicateConsumer(const VExprSPtrs& conjuncts, DataTypePtr data_type,
int column_id,
- IColumn::Filter* matches)
+ FixedWidthPredicateConsumer(const VExprSPtrs& conjuncts, DataTypePtr
data_type, int column_id,
+ IColumn::Filter* matches, IColumn*
projected_column)
: _conjuncts(conjuncts),
_data_type(std::move(data_type)),
_column_id(column_id),
- _matches(matches) {
+ _matches(matches),
+ _projected_column(projected_column) {
DORIS_CHECK(_matches != nullptr);
}
@@ -672,6 +673,23 @@ public:
_data_type,
_column_id,
_matches->data() + old_size));
}
+ if (_projected_column != nullptr) {
+ size_t row = 0;
+ while (row < num_values) {
+ while (row < num_values && (*_matches)[old_size + row] == 0) {
+ ++row;
+ }
+ const size_t run_begin = row;
+ while (row < num_values && (*_matches)[old_size + row] != 0) {
+ ++row;
+ }
+ if (row != run_begin) {
+ _projected_column->insert_many_raw_data(
+ reinterpret_cast<const char*>(values + run_begin *
value_width),
+ row - run_begin);
+ }
+ }
+ }
return Status::OK();
}
@@ -680,6 +698,7 @@ private:
DataTypePtr _data_type;
int _column_id;
IColumn::Filter* _matches;
+ IColumn* _projected_column;
};
} // namespace
@@ -1428,11 +1447,10 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::materialize_values(
}
template <bool IN_COLLECTION, bool OFFSET_INDEX>
-bool ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::can_filter_plain_values(
+bool ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::can_filter_fixed_width_values(
const VExprSPtrs& conjuncts, int column_id) const {
- if (conjuncts.empty() || _current_encoding != tparquet::Encoding::PLAIN ||
- (_metadata.type != tparquet::Type::INT32 && _metadata.type !=
tparquet::Type::INT64 &&
- _metadata.type != tparquet::Type::FLOAT && _metadata.type !=
tparquet::Type::DOUBLE)) {
+ if (conjuncts.empty() ||
+ !supports_raw_fixed_filter_encoding(_current_encoding,
_metadata.type)) {
return false;
}
const auto primitive_type =
remove_nullable(_field_schema->data_type)->get_primitive_type();
@@ -1453,17 +1471,17 @@ bool ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::can_filter_plain_values(
}
template <bool IN_COLLECTION, bool OFFSET_INDEX>
-Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::filter_plain_values(
+Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::filter_fixed_width_values(
const VExprSPtrs& conjuncts, int column_id, ColumnSelectVector&
select_vector,
- NullMap* selected_nulls, IColumn::Filter* physical_matches,
IColumn::Filter* row_filter,
- bool* used_filter) {
+ NullMap* selected_nulls, IColumn::Filter* physical_matches, IColumn*
projected_column,
+ IColumn::Filter* row_filter, bool* used_filter) {
DORIS_CHECK(selected_nulls != nullptr);
DORIS_CHECK(physical_matches != nullptr);
DORIS_CHECK(row_filter != nullptr);
DORIS_CHECK(used_filter != nullptr);
*used_filter = false;
row_filter->clear();
- if (!can_filter_plain_values(conjuncts, column_id)) {
+ if (!can_filter_fixed_width_values(conjuncts, column_id)) {
return Status::OK();
}
if (UNLIKELY(_remaining_num_values < select_vector.num_values())) {
@@ -1520,8 +1538,8 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::filter_plain_values(
if (selection.selected_values == 0) {
RETURN_IF_ERROR(_page_decoder->skip_values(selection.total_values));
} else {
- PlainPredicateConsumer consumer(conjuncts, _field_schema->data_type,
column_id,
- physical_matches);
+ FixedWidthPredicateConsumer consumer(conjuncts,
_field_schema->data_type, column_id,
+ physical_matches,
projected_column);
RETURN_IF_ERROR(_page_decoder->decode_selected_fixed_values(selection,
consumer));
DORIS_CHECK_EQ(physical_matches->size(), selection.selected_values);
}
diff --git a/be/src/format_v2/parquet/reader/native/column_chunk_reader.h
b/be/src/format_v2/parquet/reader/native/column_chunk_reader.h
index f544d4278d3..5acf544bcc9 100644
--- a/be/src/format_v2/parquet/reader/native/column_chunk_reader.h
+++ b/be/src/format_v2/parquet/reader/native/column_chunk_reader.h
@@ -166,13 +166,34 @@ public:
ParquetDecodeContext& context,
ParquetMaterializationState& state,
ColumnSelectVector& select_vector);
- // Evaluate selected fixed-width PLAIN values and return one keep byte per
selected logical
- // row. NULL comparisons are false and therefore never enter the physical
consumer.
- Status filter_plain_values(const VExprSPtrs& conjuncts, int column_id,
- ColumnSelectVector& select_vector, NullMap*
selected_nulls,
- IColumn::Filter* physical_matches,
IColumn::Filter* row_filter,
- bool* used_filter);
- bool can_filter_plain_values(const VExprSPtrs& conjuncts, int column_id)
const;
+ static bool supports_raw_fixed_filter_encoding(tparquet::Encoding::type
encoding,
+ tparquet::Type::type
physical_type) {
+ switch (encoding) {
+ case tparquet::Encoding::PLAIN:
+ return physical_type == tparquet::Type::INT32 ||
+ physical_type == tparquet::Type::INT64 ||
+ physical_type == tparquet::Type::FLOAT ||
+ physical_type == tparquet::Type::DOUBLE;
+ case tparquet::Encoding::BYTE_STREAM_SPLIT:
+ return physical_type == tparquet::Type::INT32 ||
+ physical_type == tparquet::Type::INT64 ||
+ physical_type == tparquet::Type::FLOAT ||
+ physical_type == tparquet::Type::DOUBLE;
+ case tparquet::Encoding::DELTA_BINARY_PACKED:
+ return physical_type == tparquet::Type::INT32 || physical_type ==
tparquet::Type::INT64;
+ default:
+ return false;
+ }
+ }
+
+ // Evaluate selected fixed-width values and return one keep byte per
selected logical row.
+ // NULL comparisons are false and therefore never enter the physical
consumer; non-null
+ // matches are appended to projected_column when requested.
+ Status filter_fixed_width_values(const VExprSPtrs& conjuncts, int
column_id,
+ ColumnSelectVector& select_vector,
NullMap* selected_nulls,
+ IColumn::Filter* physical_matches,
IColumn* projected_column,
+ IColumn::Filter* row_filter, bool*
used_filter);
+ bool can_filter_fixed_width_values(const VExprSPtrs& conjuncts, int
column_id) const;
// Get the repetition level decoder of current page.
LevelDecoder& rep_level_decoder() { return _rep_level_decoder; }
diff --git a/be/src/format_v2/parquet/reader/native/column_reader.cpp
b/be/src/format_v2/parquet/reader/native/column_reader.cpp
index 86c6bc39849..9417da15683 100644
--- a/be/src/format_v2/parquet/reader/native/column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/native/column_reader.cpp
@@ -1047,9 +1047,9 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::_read_nested_column(
}
template <bool IN_COLLECTION, bool OFFSET_INDEX>
-Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::_read_plain_filter_values(
+Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::_read_fixed_width_filter_values(
size_t num_values, const VExprSPtrs& conjuncts, int column_id,
FilterMap& filter_map,
- IColumn::Filter* row_filter) {
+ IColumn* projected_column, IColumn::Filter* row_filter) {
DORIS_CHECK(row_filter != nullptr);
_null_run_lengths.clear();
if (_chunk_reader->max_def_level() > 0) {
@@ -1061,7 +1061,7 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::_read_plain_filter_value
const size_t loop_read = def_decoder.get_next_run(&def_level,
num_values - has_read);
if (loop_read == 0) {
return Status::Corruption(
- "Parquet definition level stream ended while filtering
PLAIN values");
+ "Parquet definition level stream ended while filtering
fixed-width values");
}
const bool is_null = def_level < _field_schema->definition_level;
if (!(prev_is_null ^ is_null)) {
@@ -1090,9 +1090,9 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::_read_plain_filter_value
_filter_map_index));
_filter_map_index += num_values;
bool used_filter = false;
- RETURN_IF_ERROR(_chunk_reader->filter_plain_values(
- conjuncts, column_id, _select_vector, &_plain_predicate_nulls,
- &_plain_predicate_matches, row_filter, &used_filter));
+ RETURN_IF_ERROR(_chunk_reader->filter_fixed_width_values(
+ conjuncts, column_id, _select_vector,
&_fixed_width_predicate_nulls,
+ &_fixed_width_predicate_matches, projected_column, row_filter,
&used_filter));
// Chunk encodings are prevalidated before any definition level is
consumed, so a false result
// here would make a materializing fallback observe an advanced level
cursor.
DORIS_CHECK(used_filter);
@@ -1100,9 +1100,10 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::_read_plain_filter_value
}
template <bool IN_COLLECTION, bool OFFSET_INDEX>
-Status ScalarColumnReader<IN_COLLECTION, OFFSET_INDEX>::read_plain_filter(
+Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::read_fixed_width_filter(
const VExprSPtrs& conjuncts, int column_id, FilterMap& filter_map,
size_t batch_size,
- IColumn::Filter* row_filter, size_t* read_rows, bool* eof, bool*
used_filter) {
+ IColumn* projected_column, IColumn::Filter* row_filter, size_t*
read_rows, bool* eof,
+ bool* used_filter) {
DORIS_CHECK(row_filter != nullptr);
DORIS_CHECK(read_rows != nullptr);
DORIS_CHECK(eof != nullptr);
@@ -1119,15 +1120,17 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::read_plain_filter(
})) {
return Status::OK();
}
- // Levels use RLE/BIT_PACKED, while every value page must be PLAIN. Reject
mixed-encoding
- // chunks before touching either cursor so the caller can safely use the
normal expression path.
- const bool plain_only = std::ranges::all_of(
- _chunk_meta.meta_data.encodings, [](const tparquet::Encoding::type
encoding) {
- return encoding == tparquet::Encoding::PLAIN ||
+ // Validate every advertised value encoding before touching either cursor.
A late fallback
+ // cannot rewind definition levels or a previously decoded fixed-width
page.
+ const bool supported_encodings = std::ranges::all_of(
+ _chunk_meta.meta_data.encodings, [&](const
tparquet::Encoding::type encoding) {
+ return ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::
+ supports_raw_fixed_filter_encoding(encoding,
+
_chunk_meta.meta_data.type) ||
encoding == tparquet::Encoding::RLE ||
encoding == tparquet::Encoding::BIT_PACKED;
});
- if (!plain_only) {
+ if (!supported_encodings) {
return Status::OK();
}
@@ -1145,7 +1148,7 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::read_plain_filter(
} else {
RETURN_IF_ERROR(_chunk_reader->parse_page_header());
RETURN_IF_ERROR(_chunk_reader->load_page_data_idempotent());
- if (!_chunk_reader->can_filter_plain_values(conjuncts, column_id)) {
+ if (!_chunk_reader->can_filter_fixed_width_values(conjuncts,
column_id)) {
return Status::OK();
}
size_t has_read = 0;
@@ -1157,8 +1160,8 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::read_plain_filter(
const size_t values =
std::min(static_cast<size_t>(range.to() - range.from()),
batch_size - has_read);
IColumn::Filter fragment_filter;
- RETURN_IF_ERROR(_read_plain_filter_values(values, conjuncts,
column_id, filter_map,
- &fragment_filter));
+ RETURN_IF_ERROR(_read_fixed_width_filter_values(
+ values, conjuncts, column_id, filter_map,
projected_column, &fragment_filter));
row_filter->insert(row_filter->end(), fragment_filter.begin(),
fragment_filter.end());
has_read += values;
*read_rows += values;
diff --git a/be/src/format_v2/parquet/reader/native/column_reader.h
b/be/src/format_v2/parquet/reader/native/column_reader.h
index fc1c2b972d2..b9a25a8132e 100644
--- a/be/src/format_v2/parquet/reader/native/column_reader.h
+++ b/be/src/format_v2/parquet/reader/native/column_reader.h
@@ -186,11 +186,11 @@ public:
bool* eof, bool is_dict_filter,
int64_t real_column_size = -1) = 0;
- // Evaluate a predicate-only scalar directly from fixed-width PLAIN page
bytes. The default is
- // a non-consuming fallback for nested and synthetic readers.
- virtual Status read_plain_filter(const VExprSPtrs&, int, FilterMap&,
size_t,
- IColumn::Filter* row_filter, size_t*
read_rows, bool* eof,
- bool* used_filter) {
+ // Evaluate a predicate scalar directly from a supported fixed-width page
encoding. The default
+ // is a non-consuming fallback for nested and synthetic readers.
+ virtual Status read_fixed_width_filter(const VExprSPtrs&, int, FilterMap&,
size_t, IColumn*,
+ IColumn::Filter* row_filter,
size_t* read_rows,
+ bool* eof, bool* used_filter) {
DORIS_CHECK(row_filter != nullptr);
DORIS_CHECK(read_rows != nullptr);
DORIS_CHECK(eof != nullptr);
@@ -279,9 +279,10 @@ public:
const std::shared_ptr<NativeSchemaNode>& root_node,
FilterMap& filter_map, size_t batch_size, size_t*
read_rows, bool* eof,
bool is_dict_filter, int64_t real_column_size =
-1) override;
- Status read_plain_filter(const VExprSPtrs& conjuncts, int column_id,
FilterMap& filter_map,
- size_t batch_size, IColumn::Filter* row_filter,
size_t* read_rows,
- bool* eof, bool* used_filter) override;
+ Status read_fixed_width_filter(const VExprSPtrs& conjuncts, int column_id,
+ FilterMap& filter_map, size_t batch_size,
+ IColumn* projected_column, IColumn::Filter*
row_filter,
+ size_t* read_rows, bool* eof, bool*
used_filter) override;
Status read_column_levels(FilterMap& filter_map, size_t batch_size,
size_t* read_rows,
bool* eof) override;
Result<MutableColumnPtr> materialize_dictionary_values(const ColumnInt32*
dict_column,
@@ -386,8 +387,8 @@ private:
std::vector<uint16_t> _null_run_lengths;
std::unordered_set<size_t> _ancestor_null_indices;
std::vector<uint8_t> _nested_filter_map_data;
- NullMap _plain_predicate_nulls;
- IColumn::Filter _plain_predicate_matches;
+ NullMap _fixed_width_predicate_nulls;
+ IColumn::Filter _fixed_width_predicate_matches;
FilterMap _nested_filter_map;
ColumnSelectVector _select_vector;
uint8_t _oversized_scratch_idle_batches = 0;
@@ -398,8 +399,9 @@ private:
Status _skip_values(size_t num_values);
Status _read_values(size_t num_values, ColumnPtr& doris_column, const
DataTypePtr& type,
FilterMap& filter_map, bool is_dict_filter);
- Status _read_plain_filter_values(size_t num_values, const VExprSPtrs&
conjuncts, int column_id,
- FilterMap& filter_map, IColumn::Filter*
row_filter);
+ Status _read_fixed_width_filter_values(size_t num_values, const
VExprSPtrs& conjuncts,
+ int column_id, FilterMap&
filter_map,
+ IColumn* projected_column,
IColumn::Filter* row_filter);
Status _read_nested_column(ColumnPtr& doris_column, const DataTypePtr&
type,
FilterMap& filter_map, size_t batch_size,
size_t* read_rows,
bool* eof, bool is_dict_filter);
diff --git a/be/src/format_v2/parquet/reader/native_column_reader.cpp
b/be/src/format_v2/parquet/reader/native_column_reader.cpp
index 7fff27df6c8..8aa79ecb7e7 100644
--- a/be/src/format_v2/parquet/reader/native_column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/native_column_reader.cpp
@@ -337,10 +337,12 @@ Status NativeColumnReader::read_with_filter(int64_t rows,
const uint8_t* filter_
return Status::OK();
}
-Status NativeColumnReader::read_with_plain_filter(int64_t rows, const uint8_t*
filter_data,
- bool filter_all, const
VExprSPtrs& conjuncts,
- int column_id,
IColumn::Filter* row_filter,
- int64_t* rows_read, bool*
used_filter) {
+Status NativeColumnReader::read_with_fixed_width_filter(int64_t rows, const
uint8_t* filter_data,
+ bool filter_all,
+ const VExprSPtrs&
conjuncts, int column_id,
+ IColumn*
projected_column,
+ IColumn::Filter*
row_filter,
+ int64_t* rows_read,
bool* used_filter) {
DORIS_CHECK(rows >= 0);
DORIS_CHECK(row_filter != nullptr);
DORIS_CHECK(rows_read != nullptr);
@@ -361,16 +363,17 @@ Status NativeColumnReader::read_with_plain_filter(int64_t
rows, const uint8_t* f
size_t loop_rows = 0;
IColumn::Filter loop_filter;
bool loop_used = false;
- RETURN_IF_ERROR(_native_reader->read_plain_filter(
- conjuncts, column_id, filter, static_cast<size_t>(rows -
*rows_read), &loop_filter,
- &loop_rows, &eof, &loop_used));
+ RETURN_IF_ERROR(_native_reader->read_fixed_width_filter(
+ conjuncts, column_id, filter, static_cast<size_t>(rows -
*rows_read),
+ projected_column, &loop_filter, &loop_rows, &eof, &loop_used));
if (!loop_used) {
if (UNLIKELY(*rows_read != 0)) {
// Footer encoding lists are untrusted. Once a prior page
advanced the cursor, a
// typed fallback would restart the request at the wrong row,
so reject the file
// instead of terminating the BE or returning shifted results.
return Status::Corruption(
- "Parquet PLAIN predicate encoding changed after {}
rows for column {}",
+ "Parquet fixed-width predicate encoding changed after
{} rows for column "
+ "{}",
*rows_read, _name);
}
row_filter->clear();
@@ -380,7 +383,8 @@ Status NativeColumnReader::read_with_plain_filter(int64_t
rows, const uint8_t* f
if (loop_rows == 0 && !eof) {
if (++consecutive_empty_calls > _row_group_rows + 1) {
return Status::Corruption(
- "Native parquet PLAIN predicate made no progress for
column {}", _name);
+ "Native parquet fixed-width predicate made no progress
for column {}",
+ _name);
}
continue;
}
@@ -389,8 +393,8 @@ Status NativeColumnReader::read_with_plain_filter(int64_t
rows, const uint8_t* f
}
if (*rows_read != rows) {
return Status::Corruption(
- "Native parquet PLAIN predicate returned {} rows, expected {}
for {}", *rows_read,
- rows, _name);
+ "Native parquet fixed-width predicate returned {} rows,
expected {} for {}",
+ *rows_read, rows, _name);
}
*used_filter = true;
return Status::OK();
@@ -589,11 +593,10 @@ Status
NativeColumnReader::select_with_dictionary_filter(const SelectionVector&
return Status::OK();
}
-Status NativeColumnReader::select_with_plain_filter(const SelectionVector&
selection,
- uint16_t selected_rows,
int64_t batch_rows,
- const VExprSPtrs&
conjuncts, int column_id,
- IColumn::Filter*
row_filter,
- bool* used_filter) {
+Status NativeColumnReader::select_with_fixed_width_filter(
+ const SelectionVector& selection, uint16_t selected_rows, int64_t
batch_rows,
+ const VExprSPtrs& conjuncts, int column_id, IColumn* projected_column,
+ IColumn::Filter* row_filter, bool* used_filter) {
DORIS_CHECK(row_filter != nullptr);
DORIS_CHECK(used_filter != nullptr);
RETURN_IF_ERROR(validate_selected_span(batch_rows));
@@ -601,15 +604,17 @@ Status NativeColumnReader::select_with_plain_filter(const
SelectionVector& selec
const uint8_t* filter_data = nullptr;
RETURN_IF_ERROR(selection.materialize_filter(selected_rows, batch_rows,
&filter_data));
int64_t rows_read = 0;
- RETURN_IF_ERROR(read_with_plain_filter(batch_rows, filter_data,
selected_rows == 0, conjuncts,
- column_id, row_filter, &rows_read,
used_filter));
+ RETURN_IF_ERROR(read_with_fixed_width_filter(batch_rows, filter_data,
selected_rows == 0,
+ conjuncts, column_id,
projected_column, row_filter,
+ &rows_read, used_filter));
if (!*used_filter) {
return Status::OK();
}
DORIS_CHECK_EQ(rows_read, batch_rows);
if (row_filter->size() != selected_rows) {
return Status::Corruption(
- "Native parquet PLAIN predicate returned {} selected rows,
expected {} for {}",
+ "Native parquet fixed-width predicate returned {} selected
rows, expected {} for "
+ "{}",
row_filter->size(), selected_rows, _name);
}
advance_selected_span(rows_read);
diff --git a/be/src/format_v2/parquet/reader/native_column_reader.h
b/be/src/format_v2/parquet/reader/native_column_reader.h
index 4e504d3b2d7..0ffb8d05400 100644
--- a/be/src/format_v2/parquet/reader/native_column_reader.h
+++ b/be/src/format_v2/parquet/reader/native_column_reader.h
@@ -88,9 +88,10 @@ public:
const IColumn::Filter&
dictionary_filter,
MutableColumnPtr& column,
IColumn::Filter* row_filter,
bool* used_filter) override;
- Status select_with_plain_filter(const SelectionVector& selection, uint16_t
selected_rows,
- int64_t batch_rows, const VExprSPtrs&
conjuncts, int column_id,
- IColumn::Filter* row_filter, bool*
used_filter) override;
+ Status select_with_fixed_width_filter(const SelectionVector& selection,
uint16_t selected_rows,
+ int64_t batch_rows, const
VExprSPtrs& conjuncts,
+ int column_id, IColumn*
projected_column,
+ IColumn::Filter* row_filter, bool*
used_filter) override;
void flush_profile() override;
bool crossed_page_since_last_batch() override;
Result<MutableColumnPtr> dictionary_values() override;
@@ -111,10 +112,10 @@ private:
Status read_with_filter(int64_t rows, const uint8_t* filter_data, bool
filter_all,
MutableColumnPtr& column, const DataTypePtr&
output_type,
bool dictionary_ids, int64_t* rows_read);
- Status read_with_plain_filter(int64_t rows, const uint8_t* filter_data,
bool filter_all,
- const VExprSPtrs& conjuncts, int column_id,
- IColumn::Filter* row_filter, int64_t*
rows_read,
- bool* used_filter);
+ Status read_with_fixed_width_filter(int64_t rows, const uint8_t*
filter_data, bool filter_all,
+ const VExprSPtrs& conjuncts, int
column_id,
+ IColumn* projected_column,
IColumn::Filter* row_filter,
+ int64_t* rows_read, bool* used_filter);
int64_t sync_native_profile();
void record_page_fragments(int64_t page_fragments);
Status validate_selected_span(int64_t rows);
diff --git a/be/test/format_v2/parquet/native_decoder_test.cpp
b/be/test/format_v2/parquet/native_decoder_test.cpp
index a38d71dc771..1cf6fdaade7 100644
--- a/be/test/format_v2/parquet/native_decoder_test.cpp
+++ b/be/test/format_v2/parquet/native_decoder_test.cpp
@@ -175,6 +175,26 @@ TEST(ParquetV2NativeDecoderTest,
RawExprPreservesFloatNanOrdering) {
EXPECT_EQ(matches, (std::array<uint8_t, values.size()> {0, 0, 0, 1}));
}
+TEST(ParquetV2NativeDecoderTest,
RawFixedFilterSupportsIdentityWidthEncodingTypes) {
+ using Reader = ColumnChunkReader<false, false>;
+
EXPECT_TRUE(Reader::supports_raw_fixed_filter_encoding(tparquet::Encoding::BYTE_STREAM_SPLIT,
+
tparquet::Type::INT32));
+
EXPECT_TRUE(Reader::supports_raw_fixed_filter_encoding(tparquet::Encoding::BYTE_STREAM_SPLIT,
+
tparquet::Type::INT64));
+
EXPECT_TRUE(Reader::supports_raw_fixed_filter_encoding(tparquet::Encoding::BYTE_STREAM_SPLIT,
+
tparquet::Type::FLOAT));
+
EXPECT_TRUE(Reader::supports_raw_fixed_filter_encoding(tparquet::Encoding::BYTE_STREAM_SPLIT,
+
tparquet::Type::DOUBLE));
+
EXPECT_TRUE(Reader::supports_raw_fixed_filter_encoding(tparquet::Encoding::DELTA_BINARY_PACKED,
+
tparquet::Type::INT32));
+
EXPECT_TRUE(Reader::supports_raw_fixed_filter_encoding(tparquet::Encoding::DELTA_BINARY_PACKED,
+
tparquet::Type::INT64));
+
EXPECT_FALSE(Reader::supports_raw_fixed_filter_encoding(tparquet::Encoding::DELTA_BINARY_PACKED,
+
tparquet::Type::FLOAT));
+
EXPECT_FALSE(Reader::supports_raw_fixed_filter_encoding(tparquet::Encoding::RLE_DICTIONARY,
+
tparquet::Type::INT32));
+}
+
class RejectFixedConsumer final : public ParquetFixedValueConsumer {
public:
Status consume(const uint8_t* values, size_t num_values, size_t
value_width) override {
@@ -979,17 +999,28 @@ TEST(ParquetV2NativeDecoderTest,
RawExprMapsNullableSparseRowsDirectly) {
create_int32_raw_comparison(0, "lt",
TExprOpcode::LT, 13)};
NullMap selected_nulls;
IColumn::Filter physical_matches;
+ auto projected_column = make_nullable(field.data_type)->create_column();
IColumn::Filter row_filter;
bool used_filter = false;
ASSERT_TRUE(chunk_reader
- .filter_plain_values(predicates, 0, select_vector,
&selected_nulls,
- &physical_matches, &row_filter,
&used_filter)
+ .filter_fixed_width_values(predicates, 0,
select_vector, &selected_nulls,
+ &physical_matches,
projected_column.get(),
+ &row_filter, &used_filter)
.ok());
EXPECT_TRUE(used_filter);
EXPECT_EQ(selected_nulls, (NullMap {0, 0, 0, 1, 0}));
EXPECT_EQ(physical_matches, (IColumn::Filter {0, 1, 1, 0}));
EXPECT_EQ(row_filter, (IColumn::Filter {0, 1, 1, 0, 0}));
+ ASSERT_EQ(projected_column->size(), 2);
+ EXPECT_FALSE(projected_column->is_null_at(0));
+ EXPECT_FALSE(projected_column->is_null_at(1));
+ const auto& projected_values =
+ assert_cast<const ColumnInt32&>(
+ assert_cast<const
ColumnNullable&>(*projected_column).get_nested_column())
+ .get_data();
+ EXPECT_EQ(projected_values[0], 4);
+ EXPECT_EQ(projected_values[1], 7);
EXPECT_EQ(chunk_reader.remaining_num_values(), 0);
}
diff --git a/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
b/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
index c62d3566e15..490a0bde204 100644
--- a/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
+++ b/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
@@ -109,6 +109,25 @@ TEST(ParquetBenchmarkScenariosTest,
ReaderMatrixCoversOperationsEncodingsAndSche
}
}
+TEST(ParquetBenchmarkScenariosTest, ReaderMatrixCoversFixedWidthRawFilterAxes)
{
+ const auto scenarios = reader_scenarios();
+ for (const auto encoding : {Encoding::BYTE_STREAM_SPLIT,
Encoding::DELTA_BINARY_PACKED}) {
+ for (const int selectivity : {1, 10, 50, 90}) {
+ for (const auto projection :
+ {Projection::PREDICATE_ONLY,
Projection::PREDICATE_PROJECTED}) {
+ EXPECT_TRUE(std::ranges::any_of(scenarios, [&](const
ReaderScenario& scenario) {
+ return scenario.operation ==
ReaderOperation::PREDICATE_SCAN &&
+ scenario.encoding == encoding &&
scenario.null_percent == 10 &&
+ scenario.null_pattern == Pattern::ALTERNATING &&
+ scenario.selectivity_percent == selectivity &&
+ scenario.projection == projection &&
scenario.schema_width == 32 &&
+ scenario.predicate_position == 0;
+ })) << "missing fixed-width raw filter axis combination";
+ }
+ }
+ }
+}
+
TEST(ParquetBenchmarkScenariosTest,
SelectionPlanDistinguishesClusteredAndSparseRuns) {
const auto clustered = make_selection_plan(1000, 10, Pattern::CLUSTERED);
EXPECT_EQ(clustered.total_rows, 1000);
diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp
b/be/test/format_v2/parquet/parquet_scan_test.cpp
index 167e08f8c48..cc1d28238cb 100644
--- a/be/test/format_v2/parquet/parquet_scan_test.cpp
+++ b/be/test/format_v2/parquet/parquet_scan_test.cpp
@@ -567,7 +567,8 @@ std::shared_ptr<arrow::Array> build_list_array() {
void write_table(const std::string& file_path, const
std::shared_ptr<arrow::Table>& table,
int64_t row_group_size, bool enable_dictionary = false,
- bool enable_page_index = false, bool enable_statistics =
true) {
+ bool enable_page_index = false, bool enable_statistics = true,
+ std::optional<::parquet::Encoding::type> encoding =
std::nullopt) {
auto file_result = arrow::io::FileOutputStream::Open(file_path);
ASSERT_TRUE(file_result.ok()) << file_result.status();
std::shared_ptr<arrow::io::FileOutputStream> out = *file_result;
@@ -581,6 +582,9 @@ void write_table(const std::string& file_path, const
std::shared_ptr<arrow::Tabl
} else {
builder.disable_dictionary();
}
+ if (encoding.has_value()) {
+ builder.encoding(*encoding);
+ }
if (enable_page_index) {
builder.enable_write_page_index();
builder.write_batch_size(8);
@@ -620,14 +624,15 @@ void write_required_adjusted_time_parquet_file(const
std::string& file_path) {
}
void write_int_pair_parquet_file(const std::string& file_path, int64_t
row_group_size = 2,
- bool enable_statistics = true) {
+ bool enable_statistics = true,
+ std::optional<::parquet::Encoding::type>
encoding = std::nullopt) {
auto schema = arrow::schema({
arrow::field("id", arrow::int32(), false),
arrow::field("score", arrow::int32(), false),
});
auto table = arrow::Table::Make(schema, {build_int32_array({1, 2, 3, 4, 5,
6}),
build_int32_array({10, 20, 30,
40, 50, 60})});
- write_table(file_path, table, row_group_size, false, false,
enable_statistics);
+ write_table(file_path, table, row_group_size, false, false,
enable_statistics, encoding);
}
void write_uint32_pair_parquet_file(const std::string& file_path) {
@@ -665,20 +670,13 @@ void write_misdeclared_two_page_parquet_file(const
std::string& file_path) {
first_header.data_page_header_v2.__set_is_compressed(false);
auto column_bytes = serialize_test_page(first_header, first_payload);
- auto node = ::parquet::schema::PrimitiveNode::Make("id",
::parquet::Repetition::REQUIRED,
- ::parquet::Type::INT32);
- ::parquet::ColumnDescriptor descriptor(node, 0, 0);
- auto encoder = ::parquet::MakeTypedEncoder<::parquet::Int32Type>(
- ::parquet::Encoding::DELTA_BINARY_PACKED, false, &descriptor);
- const int32_t second_values[] = {3, 4};
- encoder->Put(second_values, std::size(second_values));
- auto second_buffer = encoder->FlushValues();
- std::vector<uint8_t> second_payload(second_buffer->data(),
- second_buffer->data() +
second_buffer->size());
+ // Bit width 1 followed by an RLE run of two dictionary ids. The payload
is structurally valid
+ // so the raw path reaches its late-encoding guard instead of failing
while loading the page.
+ const std::vector<uint8_t> second_payload {1, 4, 0};
tparquet::PageHeader second_header = first_header;
second_header.__set_compressed_page_size(second_payload.size());
second_header.__set_uncompressed_page_size(second_payload.size());
-
second_header.data_page_header_v2.__set_encoding(tparquet::Encoding::DELTA_BINARY_PACKED);
+
second_header.data_page_header_v2.__set_encoding(tparquet::Encoding::RLE_DICTIONARY);
auto second_page = serialize_test_page(second_header, second_payload);
column_bytes.insert(column_bytes.end(), second_page.begin(),
second_page.end());
@@ -691,7 +689,9 @@ void write_misdeclared_two_page_parquet_file(const
std::string& file_path) {
leaf.__set_repetition_type(tparquet::FieldRepetitionType::REQUIRED);
tparquet::ColumnMetaData column;
column.__set_type(tparquet::Type::INT32);
- // Deliberately omit DELTA_BINARY_PACKED to emulate untrusted/incomplete
footer metadata.
+ // Deliberately omit the second page's unsupported encoding to emulate
untrusted footer
+ // metadata. The reader must reject it before attempting a typed fallback
after partial cursor
+ // progress.
column.__set_encodings({tparquet::Encoding::PLAIN,
tparquet::Encoding::RLE});
column.__set_path_in_schema({"id"});
column.__set_codec(tparquet::CompressionCodec::UNCOMPRESSED);
@@ -1531,8 +1531,101 @@ TEST_F(ParquetScanTest,
PredicateOnlyPlainComparisonUsesPhysicalDirectPath) {
EXPECT_EQ(int32_data_column(*block.get_by_position(1).column).get_data(),
(ColumnInt32::Container {30, 40, 50, 60}));
EXPECT_EQ(block.get_by_position(0).column->size(), rows);
- EXPECT_EQ(counter_value(profile, "PlainPredicateDirectBatches"), 1);
- EXPECT_EQ(counter_value(profile, "PlainPredicateDirectRows"), 6);
+ EXPECT_EQ(counter_value(profile, "FixedWidthPredicateDirectBatches"), 1);
+ EXPECT_EQ(counter_value(profile, "FixedWidthPredicateDirectRows"), 6);
+ EXPECT_EQ(counter_value(profile, "PredicateCompactionCount"), 0);
+ EXPECT_EQ(counter_value(profile, "PredicateCompactionBytes"), 0);
+}
+
+TEST_F(ParquetScanTest,
ProjectedPlainComparisonUsesPhysicalFilterAndProjectPath) {
+ write_int_pair_parquet_file(_file_path, 6, false);
+ RuntimeProfile profile("profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok());
+
ASSERT_TRUE(request_builder.add_non_predicate_column(format::LocalColumnId(1)).ok());
+ request->conjuncts.push_back(create_int32_direct_greater_conjunct(0, 2));
+ ASSERT_TRUE(reader->open(request).ok());
+
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ ASSERT_EQ(rows, 4);
+ EXPECT_EQ(int32_data_column(*block.get_by_position(0).column).get_data(),
+ (ColumnInt32::Container {3, 4, 5, 6}));
+ EXPECT_EQ(int32_data_column(*block.get_by_position(1).column).get_data(),
+ (ColumnInt32::Container {30, 40, 50, 60}));
+ EXPECT_EQ(counter_value(profile, "FixedWidthPredicateDirectBatches"), 1);
+ EXPECT_EQ(counter_value(profile, "FixedWidthPredicateDirectRows"), 6);
+ EXPECT_EQ(counter_value(profile, "PredicateCompactionCount"), 0);
+ EXPECT_EQ(counter_value(profile, "PredicateCompactionBytes"), 0);
+}
+
+TEST_F(ParquetScanTest,
ProjectedByteStreamSplitUsesFixedWidthFilterAndProjectPath) {
+ write_int_pair_parquet_file(_file_path, 6, false,
::parquet::Encoding::BYTE_STREAM_SPLIT);
+ RuntimeProfile profile("profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok());
+
ASSERT_TRUE(request_builder.add_non_predicate_column(format::LocalColumnId(1)).ok());
+ request->conjuncts.push_back(create_int32_direct_greater_conjunct(0, 2));
+ ASSERT_TRUE(reader->open(request).ok());
+
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ ASSERT_EQ(rows, 4);
+ EXPECT_EQ(int32_data_column(*block.get_by_position(0).column).get_data(),
+ (ColumnInt32::Container {3, 4, 5, 6}));
+ EXPECT_EQ(int32_data_column(*block.get_by_position(1).column).get_data(),
+ (ColumnInt32::Container {30, 40, 50, 60}));
+ EXPECT_EQ(counter_value(profile, "FixedWidthPredicateDirectBatches"), 1);
+ EXPECT_EQ(counter_value(profile, "FixedWidthPredicateDirectRows"), 6);
+ EXPECT_EQ(counter_value(profile, "PredicateCompactionCount"), 0);
+ EXPECT_EQ(counter_value(profile, "PredicateCompactionBytes"), 0);
+}
+
+TEST_F(ParquetScanTest,
ProjectedDeltaBinaryPackedUsesFixedWidthFilterAndProjectPath) {
+ write_int_pair_parquet_file(_file_path, 6, false,
::parquet::Encoding::DELTA_BINARY_PACKED);
+ RuntimeProfile profile("profile");
+ auto reader = create_reader(0, -1, &profile);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok());
+
ASSERT_TRUE(request_builder.add_non_predicate_column(format::LocalColumnId(1)).ok());
+ request->conjuncts.push_back(create_int32_direct_greater_conjunct(0, 2));
+ ASSERT_TRUE(reader->open(request).ok());
+
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ ASSERT_EQ(rows, 4);
+ EXPECT_EQ(int32_data_column(*block.get_by_position(0).column).get_data(),
+ (ColumnInt32::Container {3, 4, 5, 6}));
+ EXPECT_EQ(int32_data_column(*block.get_by_position(1).column).get_data(),
+ (ColumnInt32::Container {30, 40, 50, 60}));
+ EXPECT_EQ(counter_value(profile, "FixedWidthPredicateDirectBatches"), 1);
+ EXPECT_EQ(counter_value(profile, "FixedWidthPredicateDirectRows"), 6);
EXPECT_EQ(counter_value(profile, "PredicateCompactionCount"), 0);
EXPECT_EQ(counter_value(profile, "PredicateCompactionBytes"), 0);
}
@@ -1565,10 +1658,10 @@ TEST_F(ParquetScanTest,
PredicateOnlyUint32FallsBackBeforeRawPlainDecode) {
ASSERT_EQ(rows, 2);
EXPECT_EQ(int32_data_column(*block.get_by_position(1).column).get_data(),
(ColumnInt32::Container {20, 30}));
- EXPECT_EQ(counter_value(profile, "PlainPredicateDirectBatches"), 0);
+ EXPECT_EQ(counter_value(profile, "FixedWidthPredicateDirectBatches"), 0);
}
-TEST_F(ParquetScanTest,
PlainPredicateReportsFooterEncodingMismatchAfterProgress) {
+TEST_F(ParquetScanTest,
FixedWidthPredicateReportsUnsupportedEncodingAfterProgress) {
write_misdeclared_two_page_parquet_file(_file_path);
RuntimeProfile profile("profile");
auto reader = create_reader(0, -1, &profile);
@@ -1619,7 +1712,7 @@ TEST_F(ParquetScanTest,
PlainDirectPathKeepsPayloadForMultiColumnResidual) {
(ColumnInt32::Container {30}));
EXPECT_EQ(block.get_by_position(0).column->size(), rows);
// A residual expression still consumes id, so raw filtering must leave
its payload available.
- EXPECT_EQ(counter_value(profile, "PlainPredicateDirectBatches"), 0);
+ EXPECT_EQ(counter_value(profile, "FixedWidthPredicateDirectBatches"), 0);
}
// Scenario: every physical batch in every row group is rejected. Predicate
readers reach each row
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
