Skip to content

[fix](parquet) Restrict V2 dictionary filtering to strings #66343

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

Closed
Gabriel39 wants to merge 2 commits into from
Closed

[fix](parquet) Restrict V2 dictionary filtering to strings #66343

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
a01b368
[fix](parquet) Restrict V2 dictionary filtering to strings
Gabriel39 Aug 1, 2026
b577cc4
[fix](parquet) Decode dictionary values for direct filters
Gabriel39 Aug 1, 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
10 changes: 5 additions & 5 deletions be/src/format_v2/parquet/parquet_scan.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -162,6 +162,11 @@ bool supports_row_level_dictionary_filter(const ParquetColumnSchema& column_sche
column_schema.max_repetition_level > 0) {
return false;
}
if (!is_string_type(remove_nullable(column_schema.type)->get_primitive_type())) {

@github-actions github-actions Bot Aug 1, 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.

This guard rejects every primitive handled by get_typed_dictionary_raw_values() before build_dictionary_entry_filter() runs. As a result, TYPED_FIXED_WIDTH, the fixed-width fused projection path, DictFilterTypedCompareColumns, and DictionaryPredicateFusedProjectedRows now have no reachable producer, while the design metrics table and INT64 benchmark matrix still advertise them. Please remove/update the dead machinery and counters, or retain a semantics-safe reachable producer, so Profiles and benchmark coverage describe executable behavior.

// Dictionary-id predicates are only equivalent to value predicates for strings here.
// Decode other dictionary types first to preserve the V1-compatible filtering domain.
return false;
}
bool is_supported_physical_type = false;
switch (column_metadata.type) {
case tparquet::Type::BYTE_ARRAY:
Expand All @@ -182,11 +187,6 @@ bool supports_row_level_dictionary_filter(const ParquetColumnSchema& column_sche
if (!is_supported_physical_type) {
return false;
}
if (remove_nullable(column_schema.type)->get_primitive_type() == TYPE_VARBINARY) {
// A table STRING predicate can be rewritten through a raw VARBINARY file slot. Evaluating
// it on dictionary Fields before the mapping expression is neither type-safe nor exact.
return false;
}
// The row filter consumes dictionary ids rather than decoded values, so a plain data page
// cannot resume this reader without changing its output domain. Keep mixed chunks on the
// normal decoded-value path to preserve one representation for the complete column chunk.
Expand Down
8 changes: 7 additions & 1 deletion be/src/format_v2/parquet/reader/native/column_chunk_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1782,7 +1782,8 @@ bool ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::can_filter_fixed_width_valu
}
const bool encoding_supports_raw_values =
supports_raw_fixed_filter_encoding(_current_encoding, _metadata.type) ||
supports_raw_binary_filter_encoding(_current_encoding, _metadata.type);
supports_raw_binary_filter_encoding(_current_encoding, _metadata.type) ||
supports_dictionary_fixed_filter_encoding(_current_encoding, _metadata.type);
const bool can_convert_logical_values = serde != nullptr && decode_context != nullptr &&
encoding_supports_raw_values &&
serde->supports_parquet_raw_predicate(*decode_context);
Expand Down Expand Up @@ -1837,6 +1838,11 @@ Status ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::filter_fixed_width_values
// Direct filtering can be the first value operation on a page, so its SerDe dispatch must not
// depend on materialize_values() having synchronized the page encoding first.
RETURN_IF_ERROR(translate_value_encoding(_current_encoding, &page_decode_context.encoding));
if (supports_dictionary_fixed_filter_encoding(_current_encoding, _metadata.type)) {
// The dictionary decoder expands IDs into physical values before SerDe conversion, making
// the predicate input identical to a decoded PLAIN page while retaining dictionary cursors.
page_decode_context.encoding = ParquetValueEncoding::PLAIN;
}
if (!can_filter_fixed_width_values(conjuncts, column_id, &serde, &page_decode_context)) {
return Status::OK();
}
Expand Down
13 changes: 13 additions & 0 deletions be/src/format_v2/parquet/reader/native/column_chunk_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -215,6 +215,19 @@ class ColumnChunkReader {
encoding == tparquet::Encoding::BYTE_STREAM_SPLIT);
}

static bool supports_dictionary_fixed_filter_encoding(tparquet::Encoding::type encoding,
tparquet::Type::type physical_type) {
const bool is_dictionary = encoding == tparquet::Encoding::RLE_DICTIONARY ||
encoding == tparquet::Encoding::PLAIN_DICTIONARY;
if (!is_dictionary) {
return false;
}
return physical_type == tparquet::Type::INT32 || physical_type == tparquet::Type::INT64 ||
physical_type == tparquet::Type::INT96 || physical_type == tparquet::Type::FLOAT ||
physical_type == tparquet::Type::DOUBLE ||
physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY;
}

// 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.
Expand Down
3 changes: 3 additions & 0 deletions be/src/format_v2/parquet/reader/native/column_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1271,6 +1271,9 @@ Status ScalarColumnReader<IN_COLLECTION, OFFSET_INDEX>::read_fixed_width_filter(
ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::
supports_raw_binary_filter_encoding(
encoding, _chunk_meta.meta_data.type) ||
ColumnChunkReader<IN_COLLECTION, OFFSET_INDEX>::

@github-actions github-actions Bot Aug 1, 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.

[P1] This chunk-level whitelist is broader than the page-level predicate capability. For a logical string backed by FIXED_LEN_BYTE_ARRAY, a comparison can run only through the raw-binary consumer: PLAIN/BYTE_STREAM_SPLIT pages are accepted, but RLE/PLAIN_DICTIONARY pages are not, and String SerDe has no converted-fixed predicate path. Because this new clause makes metadata containing both encodings pass the up-front check, a valid PLAIN-first/dictionary-later chunk consumes the first page and then returns used_filter=false; NativeColumnReader converts that late fallback into corruption. Please reject dictionary encodings up front for raw-binary-only conjuncts (or add a dictionary binary consumer), with a regression covering that page order.

supports_dictionary_fixed_filter_encoding(
encoding, _chunk_meta.meta_data.type) ||
encoding == tparquet::Encoding::RLE ||
encoding == tparquet::Encoding::BIT_PACKED;
});
Expand Down
72 changes: 72 additions & 0 deletions be/src/format_v2/parquet/reader/native/fix_length_dict_decoder.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,77 @@ class FixLengthDictDecoder final : public BaseDictDecoder {
static_cast<size_t>(_type_length));
}

Status decode_selected_fixed_values(const ParquetSelection& selection,
ParquetFixedValueConsumer& consumer) override {
DORIS_CHECK_GT(_type_length, 0);
// Raw predicates on non-string dictionaries must observe decoded physical values, not
// dictionary IDs, so expand each validated index before invoking the predicate consumer.
class ExpandedValueConsumer final : public ParquetDictionaryValueConsumer {
public:
ExpandedValueConsumer(const uint8_t* dictionary, size_t dictionary_size,
size_t value_width, ParquetFixedValueConsumer& consumer,
std::vector<uint8_t>& scratch)
: _dictionary(dictionary),
_dictionary_size(dictionary_size),
_value_width(value_width),
_consumer(consumer),
_scratch(scratch) {}

Status consume_indices(const uint32_t* indices, size_t num_values) override {
DORIS_CHECK(indices != nullptr || num_values == 0);
if (UNLIKELY(num_values > std::numeric_limits<size_t>::max() / _value_width)) {
return Status::IOError("Parquet dictionary expansion size overflows");
}
_scratch.resize(num_values * _value_width);

@github-actions github-actions Bot Aug 1, 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.

[P1] This expansion is bounded by values, not bytes. FIXED_LEN_BYTE_ARRAY accepts any positive type_length, and DECIMAL validation only checks that the declared precision fits that width. A required DECIMAL(1,0) with a 1 MiB width, a one-entry dictionary, and a 1,024-row repeated-ID page is only slightly over 1 MiB on disk, but consume_repeated() resizes this scratch to exactly 1 GiB before Decimal SerDe rejects widths above Int256 (or marks them NULL in permissive mode). The fragmented/indices path can amplify further. Please validate the logical/physical width before selecting this path and byte-bound or chunk the expansion, with an oversized FLBA dictionary regression for strict and permissive scans.

for (size_t row = 0; row < num_values; ++row) {
DORIS_CHECK_LT(indices[row], _dictionary_size);
memcpy(_scratch.data() + row * _value_width,
_dictionary + static_cast<size_t>(indices[row]) * _value_width,
_value_width);
}
return _consumer.consume(_scratch.data(), num_values, _value_width);
}

Status consume_repeated(uint32_t index, size_t num_values) override {
DORIS_CHECK_LT(index, _dictionary_size);
constexpr size_t MAX_BATCH_VALUES = 1024;
const uint8_t* value = _dictionary + static_cast<size_t>(index) * _value_width;
while (num_values > 0) {
const size_t batch = std::min(num_values, MAX_BATCH_VALUES);
_scratch.resize(batch * _value_width);
for (size_t row = 0; row < batch; ++row) {
memcpy(_scratch.data() + row * _value_width, value, _value_width);
}
RETURN_IF_ERROR(_consumer.consume(_scratch.data(), batch, _value_width));
num_values -= batch;
}
return Status::OK();
}

private:
const uint8_t* const _dictionary;
const size_t _dictionary_size;
const size_t _value_width;
ParquetFixedValueConsumer& _consumer;
std::vector<uint8_t>& _scratch;
} expanded_consumer(_dict.get(), _num_dictionary_values, static_cast<size_t>(_type_length),
consumer, _expanded_values);
return decode_selected_dictionary_values(selection, expanded_consumer);

@github-actions github-actions Bot Aug 1, 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.

[P2] This streams directly into FixedWidthPredicateConsumer, which appends matching values to the caller-owned projected column for each validated range/run. A later selected ID, filtered-tail ID, or consumer conversion can still fail after those early appends, and neither this method nor filter_fixed_width_values() restores the column. The ordinary dictionary gather path explicitly snapshots and resize(old_size) on the same late-corruption case (DictionaryDirectGatherRollsBackLateCorruptRun). Please add the same all-or-nothing rollback around projected direct filtering and cover a valid selected head followed by a corrupt filtered tail or later run.

}

void release_scratch(size_t max_retained_bytes) override {
BaseDictDecoder::release_scratch(max_retained_bytes);
release_vector_if_oversized(&_expanded_values, max_retained_bytes);
}

size_t retained_scratch_bytes() const override {
return BaseDictDecoder::retained_scratch_bytes() + _expanded_values.capacity();
}

size_t active_scratch_bytes() const override {
return BaseDictDecoder::active_scratch_bytes() + _expanded_values.size();
}

Status set_dict(DorisUniqueBufferPtr<uint8_t>& dict, int32_t length,
size_t num_values) override {
if (UNLIKELY(_type_length <= 0 || length < 0 ||
Expand All @@ -59,6 +130,7 @@ class FixLengthDictDecoder final : public BaseDictDecoder {

private:
size_t _num_dictionary_values = 0;
std::vector<uint8_t> _expanded_values;
};

} // namespace doris::format::parquet::native
52 changes: 45 additions & 7 deletions be/test/format_v2/parquet/parquet_reader_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,7 @@
#include "storage/segment/condition_cache.h"
#include "storage/utils.h"
#include "util/defer_op.h"
#include "util/unaligned.h"

namespace doris {
namespace {
Expand Down Expand Up @@ -175,6 +176,23 @@ class Int32DictionaryEqualsExpr final : public VExpr {

bool can_evaluate_dictionary_filter() const override { return true; }

bool can_execute_on_raw_fixed_values(const DataTypePtr& data_type,
int column_id) const override {
return column_id == _column_id &&
remove_nullable(data_type)->get_primitive_type() == TYPE_INT;
}

Status execute_on_raw_fixed_values(const uint8_t* values, size_t num_values, size_t value_width,
const DataTypePtr&, int, uint8_t* matches) const override {
DORIS_CHECK_EQ(value_width, sizeof(int32_t));
// Non-string dictionaries bypass dictionary-id filtering, so this test predicate must
// preserve equality semantics when the reader evaluates decoded physical values directly.
for (size_t row = 0; row < num_values; ++row) {
matches[row] &= unaligned_load<int32_t>(values + row * sizeof(int32_t)) == _value;
}
return Status::OK();
}

ZoneMapFilterResult evaluate_dictionary_filter(
const DictionaryEvalContext& ctx) const override {
const auto* dictionary = ctx.slot(_column_id);
Expand Down Expand Up @@ -215,6 +233,15 @@ class DictionaryAcceptAllExpr final : public VExpr {

bool can_evaluate_dictionary_filter() const override { return true; }

bool can_execute_on_raw_fixed_values(const DataTypePtr&, int column_id) const override {
return column_id == _column_id;
}

Status execute_on_raw_fixed_values(const uint8_t*, size_t, size_t, const DataTypePtr&, int,
uint8_t*) const override {
return Status::OK();
}

ZoneMapFilterResult evaluate_dictionary_filter(
const DictionaryEvalContext& ctx) const override {
return ctx.slot(_column_id) == nullptr ? ZoneMapFilterResult::kUnsupported
Expand Down Expand Up @@ -2851,7 +2878,7 @@ TEST_F(NewParquetReaderTest, DictionaryPredicateFiltersRowsInsideRowGroup) {
EXPECT_GE(profile.get_counter("ReaderSelectRows")->value(), 8);
}

TEST_F(NewParquetReaderTest, FixedWidthDictionaryPredicateFiltersRowsByDictionaryId) {
TEST_F(NewParquetReaderTest, FixedWidthDictionaryPredicateUsesRawDirectFilter) {
write_fixed_width_dictionary_filter_parquet_file(_file_path);

RuntimeProfile profile("new_parquet_reader_fixed_width_dictionary_filter_profile");
Expand Down Expand Up @@ -2889,14 +2916,19 @@ TEST_F(NewParquetReaderTest, FixedWidthDictionaryPredicateFiltersRowsByDictionar

EXPECT_EQ(ids, std::vector<int32_t>({2, 4, 6}));
EXPECT_EQ(values, std::vector<int32_t>({20, 20, 20}));
EXPECT_EQ(profile.get_counter("RowsFilteredByDictFilter")->value(), 3);
EXPECT_EQ(profile.get_counter("RowsFilteredByConjunct")->value(), 3);
EXPECT_EQ(profile.get_counter("RowsFilteredByDictFilter")->value(), 0);
EXPECT_EQ(profile.get_counter("DictFilterCandidateColumns")->value(), 1);
EXPECT_EQ(profile.get_counter("DictFilterColumns")->value(), 1);
EXPECT_EQ(profile.get_counter("DictFilterUnsupportedColumns")->value(), 0);
EXPECT_EQ(profile.get_counter("DictFilterColumns")->value(), 0);
EXPECT_EQ(profile.get_counter("DictFilterUnsupportedColumns")->value(), 1);
EXPECT_EQ(profile.get_counter("DictFilterReadFailures")->value(), 0);
EXPECT_EQ(profile.get_counter("RawValuePredicateDirectBatches")->value(), 1);
EXPECT_EQ(profile.get_counter("RawValuePredicateDirectRows")->value(), 6);
EXPECT_EQ(profile.get_counter("FixedWidthPredicateDirectBatches")->value(), 1);
EXPECT_EQ(profile.get_counter("FixedWidthPredicateDirectRows")->value(), 6);
}

TEST_F(NewParquetReaderTest, AllFixedWidthDictionaryTypesDecodeThroughDictionaryIds) {
TEST_F(NewParquetReaderTest, OnlyStringFixedWidthDictionaryTypeUsesDictionaryIds) {
write_all_fixed_width_dictionary_filter_parquet_file(_file_path);

auto parquet_file_reader = ::parquet::ParquetFileReader::OpenFile(_file_path, false);
Expand Down Expand Up @@ -2941,9 +2973,15 @@ TEST_F(NewParquetReaderTest, AllFixedWidthDictionaryTypesDecodeThroughDictionary
EXPECT_EQ(total_rows, 6);
EXPECT_EQ(profile.get_counter("RowsFilteredByDictFilter")->value(), 0);
EXPECT_EQ(profile.get_counter("DictFilterCandidateColumns")->value(), 5);
EXPECT_EQ(profile.get_counter("DictFilterColumns")->value(), 5);
EXPECT_EQ(profile.get_counter("DictFilterUnsupportedColumns")->value(), 0);
EXPECT_EQ(profile.get_counter("DictFilterColumns")->value(), 1);
EXPECT_EQ(profile.get_counter("DictFilterUnsupportedColumns")->value(), 4);
EXPECT_EQ(profile.get_counter("DictFilterReadFailures")->value(), 0);
// INT64, FLOAT, and DOUBLE use decoded raw values directly. Legacy INT96 keeps its specialized
// conversion path, while FIXED_LEN_BYTE_ARRAY is the sole string dictionary-ID candidate.
EXPECT_EQ(profile.get_counter("RawValuePredicateDirectBatches")->value(), 3);
EXPECT_EQ(profile.get_counter("RawValuePredicateDirectRows")->value(), 18);
EXPECT_EQ(profile.get_counter("FixedWidthPredicateDirectBatches")->value(), 3);
EXPECT_EQ(profile.get_counter("FixedWidthPredicateDirectRows")->value(), 18);
}

TEST_F(NewParquetReaderTest, DictionaryPredicateReaderIsSharedOutsideMergeRangeReader) {
Expand Down
Loading
Loading