From 240b863c691e8eb3879268e682711f22ce1a40ec Mon Sep 17 00:00:00 2001 From: Gabriel Date: Sun, 26 Jul 2026 13:24:07 +0800 Subject: [PATCH 1/3] [fix](parquet) Coalesce adjacent condition cache ranges ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: FileScannerV2 split every selected Parquet range at each 2048-row condition-cache granule. Unlike the V1 RowRanges implementation, it did not merge adjacent surviving granules, so an all-true cache hit capped physical batches at the cache granule size and inflated reader calls. Merge adjacent intersections while preserving gaps for false granules. ### Release note Condition-cache hits no longer split adjacent surviving Parquet ranges at cache granule boundaries. ### Check List (For Author) - Test: Unit Test - ./run-be-ut.sh --run --filter=*ConditionCache* -j 32 - Behavior changed: Yes. Adjacent true cache granules retain the original read-range batch boundary. - Does this need documentation: No. --- be/src/format_v2/parquet/parquet_scan.cpp | 12 ++++++-- .../format_v2/parquet/parquet_reader_test.cpp | 28 +++++++++++++++++++ 2 files changed, 37 insertions(+), 3 deletions(-) diff --git a/be/src/format_v2/parquet/parquet_scan.cpp b/be/src/format_v2/parquet/parquet_scan.cpp index a60d3efbeeab81..4dba81f20a801a 100644 --- a/be/src/format_v2/parquet/parquet_scan.cpp +++ b/be/src/format_v2/parquet/parquet_scan.cpp @@ -796,11 +796,17 @@ int64_t count_range_rows(const std::vector& ranges) { } void append_intersection(const RowRange& left, const RowRange& right, - std::vector* result) { + std::vector& result) { const int64_t start = std::max(left.start, right.start); const int64_t end = std::min(left.start + left.length, right.start + right.length); if (start < end) { - result->push_back(RowRange {.start = start, .length = end - start}); + // Cache granules are only filter coordinates. Merge adjacent survivors so cache hits + // preserve the original read-range batch boundaries, matching V1 RowRanges semantics. + if (!result.empty() && result.back().start + result.back().length == start) { + result.back().length = end - result.back().start; + return; + } + result.push_back(RowRange {.start = start, .length = end - start}); } } @@ -832,7 +838,7 @@ std::vector filter_ranges_by_condition_cache(const std::vectorset_batch_size(ConditionCacheContext::GRANULE_SIZE * 2); + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + ASSERT_TRUE(reader->init(&state).ok()); + + std::vector schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + ASSERT_EQ(schema.size(), 1); + + auto request = std::make_shared(); + request->non_predicate_columns = {field_projection(0)}; + ASSERT_TRUE(reader->open(request).ok()); + + auto ctx = std::make_shared(); + ctx->is_hit = true; + ctx->filter_result = std::make_shared>(std::vector {true, true}); + reader->set_condition_cache_context(ctx); + + Block block = build_file_block(schema); + size_t rows = 0; + bool eof = false; + ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok()); + EXPECT_FALSE(eof); + EXPECT_EQ(rows, ConditionCacheContext::GRANULE_SIZE * 2); +} + TEST_F(NewParquetReaderTest, ReadMultipleRowGroups) { write_parquet_file(_file_path, 2); auto parquet_file_reader = ::parquet::ParquetFileReader::OpenFile(_file_path, false); From 02ab60ae1379c922e557408939dc0eac59434380 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 27 Jul 2026 16:07:26 +0800 Subject: [PATCH 2/3] [improvement](parquet) Optimize sparse runtime filter scans --- .../data_type_serde/parquet_decode_source.cpp | 23 +- be/src/exprs/bloom_filter_func.h | 60 +++++ be/src/exprs/hybrid_set.h | 19 ++ be/src/exprs/runtime_filter_expr.cpp | 31 +++ be/src/exprs/runtime_filter_expr.h | 7 + be/src/exprs/vbloom_predicate.cpp | 69 +++++- be/src/exprs/vbloom_predicate.h | 8 + be/src/exprs/vdirect_in_predicate.h | 79 +++++++ be/src/format_v2/parquet/parquet_scan.cpp | 42 +++- .../reader/native/column_chunk_reader.cpp | 6 +- .../format_v2/parquet/reader/native/decoder.h | 61 +++++ be/test/exprs/bloom_filter_func_test.cpp | 129 +++++++++++ be/test/exprs/expr_zonemap_filter_test.cpp | 95 ++++++++ .../format_v2/parquet/native_decoder_test.cpp | 169 ++++++++++++++ .../format_v2/parquet/parquet_scan_test.cpp | 218 ++++++++++++++++++ 15 files changed, 997 insertions(+), 19 deletions(-) diff --git a/be/src/core/data_type_serde/parquet_decode_source.cpp b/be/src/core/data_type_serde/parquet_decode_source.cpp index ca75bcb8948516..68df6074ddfa7a 100644 --- a/be/src/core/data_type_serde/parquet_decode_source.cpp +++ b/be/src/core/data_type_serde/parquet_decode_source.cpp @@ -19,7 +19,9 @@ #include #include +#include +#include "core/column/column_decimal.h" #include "core/column/column_string.h" #include "core/column/column_vector.h" #include "util/simd/parquet_kernels.h" @@ -27,11 +29,11 @@ namespace doris { namespace { -template -bool try_gather_vector(IColumn& destination, const IColumn& dictionary, const uint32_t* indices, - size_t num_values) { - using ColumnType = ColumnVector; +template +bool try_gather_fixed_width(IColumn& destination, const IColumn& dictionary, + const uint32_t* indices, size_t num_values) { using ValueType = typename ColumnType::value_type; + static_assert(std::is_trivially_copyable_v); if constexpr (sizeof(ValueType) != 4 && sizeof(ValueType) != 8) { return false; } else { @@ -57,6 +59,12 @@ bool try_gather_vector(IColumn& destination, const IColumn& dictionary, const ui } } +template +bool try_gather_vector(IColumn& destination, const IColumn& dictionary, const uint32_t* indices, + size_t num_values) { + return try_gather_fixed_width>(destination, dictionary, indices, num_values); +} + template bool try_gather_strings(IColumn& destination, const IColumn& dictionary, const uint32_t* indices, size_t num_values) { @@ -108,11 +116,18 @@ bool try_simd_insert_parquet_dictionary_indices(IColumn& destination, const ICol TRY_PARQUET_GATHER(TYPE_DATETIME); TRY_PARQUET_GATHER(TYPE_DATEV2); TRY_PARQUET_GATHER(TYPE_DATETIMEV2); + TRY_PARQUET_GATHER(TYPE_TIMESTAMPTZ); TRY_PARQUET_GATHER(TYPE_IPV4); TRY_PARQUET_GATHER(TYPE_TIMEV2); TRY_PARQUET_GATHER(TYPE_UINT32); TRY_PARQUET_GATHER(TYPE_UINT64); #undef TRY_PARQUET_GATHER + // Keep every 4/8-byte POD column on this path: aliases such as TIMESTAMPTZ and decimals do not + // participate in the ordinary numeric dispatch and otherwise silently fall back to Field insertion. + if (try_gather_fixed_width(destination, dictionary, indices, num_values) || + try_gather_fixed_width(destination, dictionary, indices, num_values)) { + return true; + } // String survivors have variable widths, so pre-size both buffers and copy each selected // dictionary slice exactly once instead of routing every id through generic Field insertion. if (try_gather_strings(destination, dictionary, indices, num_values) || diff --git a/be/src/exprs/bloom_filter_func.h b/be/src/exprs/bloom_filter_func.h index ae7057a6e90cbd..bd54bd12f52785 100644 --- a/be/src/exprs/bloom_filter_func.h +++ b/be/src/exprs/bloom_filter_func.h @@ -17,9 +17,12 @@ #pragma once +#include + #include "common/exception.h" #include "common/status.h" #include "core/column/column_dictionary.h" +#include "core/field.h" #include "exec/runtime_filter/runtime_filter_definitions.h" #include "exprs/bloom_filter_func_impl.h" #include "exprs/filter_base.h" @@ -136,6 +139,13 @@ class BloomFilterFuncBase : public FilterBase { virtual void find_fixed_len(const ColumnPtr& column, uint8_t* results, const uint8_t* __restrict filter = nullptr) = 0; + virtual PrimitiveType primitive_type() const = 0; + virtual bool supports_raw_fixed_values() const = 0; + virtual size_t raw_fixed_value_size() const = 0; + virtual Status find_batch_raw_fixed(const uint8_t* values, size_t rows, size_t value_width, + uint8_t* matches) const = 0; + virtual bool test_field(const Field& field) const = 0; + virtual uint16_t find_fixed_len_olap_engine(const IColumn& column, const uint8_t* nullmap, uint16_t* offsets, int number, bool is_parse_column) = 0; @@ -181,6 +191,56 @@ class BloomFilterFunc final : public BloomFilterFuncBase { OpV2::find_batch(*_bloom_filter, column, results, filter); } + PrimitiveType primitive_type() const override { return type; } + + bool supports_raw_fixed_values() const override { return !is_string_type(type); } + + size_t raw_fixed_value_size() const override { + if constexpr (is_string_type(type)) { + return 0; + } else { + return sizeof(typename PrimitiveTypeTraits::CppType); + } + } + + Status find_batch_raw_fixed(const uint8_t* values, size_t rows, size_t value_width, + uint8_t* matches) const override { + if constexpr (is_string_type(type)) { + return Status::NotSupported("String Bloom filter cannot probe fixed-width values"); + } else { + using ValueType = typename PrimitiveTypeTraits::CppType; + if (_bloom_filter == nullptr) { + return Status::InternalError("Bloom filter is not initialized"); + } + if (value_width != sizeof(ValueType)) { + return Status::Corruption("Raw Bloom filter width {} does not match expected {}", + value_width, sizeof(ValueType)); + } + DORIS_CHECK(values != nullptr || rows == 0); + DORIS_CHECK(matches != nullptr || rows == 0); + for (size_t row = 0; row < rows; ++row) { + ValueType value; + std::memcpy(&value, values + row * sizeof(ValueType), sizeof(ValueType)); + matches[row] &= _bloom_filter->test_element(value) ? 1 : 0; + } + return Status::OK(); + } + } + + bool test_field(const Field& field) const override { + DORIS_CHECK(_bloom_filter != nullptr); + if (field.is_null()) { + return _bloom_filter->contain_null(); + } + if constexpr (is_string_type(type)) { + const auto& value = field.get(); + return _bloom_filter->test_element( + StringRef(value.data(), value.size())); + } else { + return _bloom_filter->test_element(field.get()); + } + } + template uint16_t find_dict_olap_engine(const ColumnDictI32* column, const uint8_t* nullmap, uint16_t* offsets, int number) { diff --git a/be/src/exprs/hybrid_set.h b/be/src/exprs/hybrid_set.h index 18b59338b26d04..a7c3f6580ca891 100644 --- a/be/src/exprs/hybrid_set.h +++ b/be/src/exprs/hybrid_set.h @@ -20,6 +20,8 @@ #include #include +#include + #include "common/object_pool.h" #include "core/column/column_nullable.h" #include "core/column/column_string.h" @@ -204,6 +206,13 @@ class HybridSetBase : public FilterBase { // use in vectorize execute engine virtual bool find(const void* data, size_t) const = 0; + virtual void find_batch_raw_fixed(const uint8_t* values, size_t rows, size_t value_width, + uint8_t* matches) const { + for (size_t row = 0; row < rows; ++row) { + matches[row] &= find(values + row * value_width) ? 1 : 0; + } + } + virtual void find_batch(const doris::IColumn& column, size_t rows, doris::ColumnUInt8::Container& results, const uint8_t* __restrict filter = nullptr) = 0; @@ -297,6 +306,16 @@ class HybridSet : public HybridSetBase { bool find(const void* data, size_t /*unused*/) const override { return find(data); } + void find_batch_raw_fixed(const uint8_t* values, size_t rows, size_t value_width, + uint8_t* matches) const override { + DORIS_CHECK_EQ(value_width, sizeof(ElementType)); + for (size_t row = 0; row < rows; ++row) { + ElementType value; + std::memcpy(&value, values + row * sizeof(ElementType), sizeof(ElementType)); + matches[row] &= _set.find(value) ? 1 : 0; + } + } + void find_batch(const doris::IColumn& column, size_t rows, doris::ColumnUInt8::Container& results, const uint8_t* __restrict filter = nullptr) override { diff --git a/be/src/exprs/runtime_filter_expr.cpp b/be/src/exprs/runtime_filter_expr.cpp index e0f6e7acc1b8d0..cca2f5d5fd2bf9 100644 --- a/be/src/exprs/runtime_filter_expr.cpp +++ b/be/src/exprs/runtime_filter_expr.cpp @@ -236,6 +236,37 @@ bool RuntimeFilterExpr::can_evaluate_zonemap_filter() const { return _impl->can_evaluate_zonemap_filter(); } +bool RuntimeFilterExpr::can_execute_on_raw_fixed_values(const DataTypePtr& data_type, + int column_id) const { + // Raw and dictionary streams omit NULL payloads and currently map NULL rows to false. A + // null-aware RF must therefore stay on execute_filter(), which restores its NULL semantics. + return !_null_aware && _impl->can_execute_on_raw_fixed_values(data_type, column_id); +} + +Status RuntimeFilterExpr::execute_on_raw_fixed_values(const uint8_t* values, size_t num_values, + size_t value_width, + const DataTypePtr& data_type, int column_id, + uint8_t* matches) const { + if (!can_execute_on_raw_fixed_values(data_type, column_id)) { + return Status::NotSupported("Runtime filter {} cannot evaluate raw fixed-width values", + _filter_id); + } + return _impl->execute_on_raw_fixed_values(values, num_values, value_width, data_type, column_id, + matches); +} + +ZoneMapFilterResult RuntimeFilterExpr::evaluate_dictionary_filter( + const DictionaryEvalContext& ctx) const { + if (!can_evaluate_dictionary_filter()) { + return ZoneMapFilterResult::kUnsupported; + } + return _impl->evaluate_dictionary_filter(ctx); +} + +bool RuntimeFilterExpr::can_evaluate_dictionary_filter() const { + return !_null_aware && _impl->can_evaluate_dictionary_filter(); +} + void RuntimeFilterExpr::collect_slot_column_ids(std::set& column_ids) const { _impl->collect_slot_column_ids(column_ids); } diff --git a/be/src/exprs/runtime_filter_expr.h b/be/src/exprs/runtime_filter_expr.h index 7994d2a71ae14f..4879b649a2f25d 100644 --- a/be/src/exprs/runtime_filter_expr.h +++ b/be/src/exprs/runtime_filter_expr.h @@ -107,6 +107,13 @@ class RuntimeFilterExpr final : public VExpr { ZoneMapFilterResult evaluate_zonemap_filter(const ZoneMapEvalContext& ctx) const override; bool can_evaluate_zonemap_filter() const override; + bool can_execute_on_raw_fixed_values(const DataTypePtr& data_type, + int column_id) const override; + Status execute_on_raw_fixed_values(const uint8_t* values, size_t num_values, size_t value_width, + const DataTypePtr& data_type, int column_id, + uint8_t* matches) const override; + ZoneMapFilterResult evaluate_dictionary_filter(const DictionaryEvalContext& ctx) const override; + bool can_evaluate_dictionary_filter() const override; void collect_slot_column_ids(std::set& column_ids) const override; int filter_id() const { return _filter_id; } diff --git a/be/src/exprs/vbloom_predicate.cpp b/be/src/exprs/vbloom_predicate.cpp index 552e3db1c198f6..6004a74181cc50 100644 --- a/be/src/exprs/vbloom_predicate.cpp +++ b/be/src/exprs/vbloom_predicate.cpp @@ -31,6 +31,8 @@ #include "core/data_type/data_type_nullable.h" #include "core/types.h" #include "exprs/bloom_filter_func.h" +#include "exprs/expr_zonemap_filter.h" +#include "exprs/vslot_ref.h" #include "runtime/runtime_state.h" namespace doris { @@ -106,6 +108,71 @@ Status VBloomPredicate::execute_runtime_filter(VExprContext* context, const Bloc ColumnPtr* arg_column) const { return _do_execute(context, block, filter, nullptr, count, result_column); } + +namespace { + +bool bloom_filter_type_matches(PrimitiveType filter_type, const DataTypePtr& data_type) { + if (data_type == nullptr) { + return false; + } + const auto value_type = remove_nullable(data_type)->get_primitive_type(); + return filter_type == value_type || (is_string_type(filter_type) && is_string_type(value_type)); +} + +} // namespace + +bool VBloomPredicate::can_execute_on_raw_fixed_values(const DataTypePtr& data_type, + int column_id) const { + if (_filter == nullptr || !_filter->supports_raw_fixed_values() || _children.size() != 1) { + return false; + } + const auto slot = std::dynamic_pointer_cast(_children[0]); + return slot != nullptr && slot->column_id() == column_id && + bloom_filter_type_matches(_filter->primitive_type(), slot->data_type()) && + bloom_filter_type_matches(_filter->primitive_type(), data_type); +} + +Status VBloomPredicate::execute_on_raw_fixed_values(const uint8_t* values, size_t num_values, + size_t value_width, + const DataTypePtr& data_type, int column_id, + uint8_t* matches) const { + if (!can_execute_on_raw_fixed_values(data_type, column_id)) { + return Status::NotSupported("Bloom predicate cannot evaluate raw fixed-width values"); + } + // Hash physical values inside BloomFilterFunc; reconstructing an untyped hash here could + // disagree with the build-side hash for dates, decimals, and other fixed-width wrappers. + return _filter->find_batch_raw_fixed(values, num_values, value_width, matches); +} + +ZoneMapFilterResult VBloomPredicate::evaluate_dictionary_filter( + const DictionaryEvalContext& ctx) const { + if (!can_evaluate_dictionary_filter()) { + return ZoneMapFilterResult::kUnsupported; + } + const auto slot = std::dynamic_pointer_cast(_children[0]); + DORIS_CHECK(slot != nullptr); + const auto* dictionary = ctx.slot(slot->column_id()); + if (dictionary == nullptr || + !bloom_filter_type_matches(_filter->primitive_type(), dictionary->data_type)) { + return ZoneMapFilterResult::kUnsupported; + } + for (const auto& value : dictionary->values) { + if (_filter->test_field(value)) { + return ZoneMapFilterResult::kMayMatch; + } + } + return ZoneMapFilterResult::kNoMatch; +} + +bool VBloomPredicate::can_evaluate_dictionary_filter() const { + if (_filter == nullptr || _children.size() != 1) { + return false; + } + const auto slot = std::dynamic_pointer_cast(_children[0]); + return slot != nullptr && + bloom_filter_type_matches(_filter->primitive_type(), slot->data_type()); +} + const std::string& VBloomPredicate::expr_name() const { return EXPR_NAME; } @@ -125,4 +192,4 @@ uint64_t VBloomPredicate::get_digest(uint64_t seed) const { return 0; } -} // namespace doris \ No newline at end of file +} // namespace doris diff --git a/be/src/exprs/vbloom_predicate.h b/be/src/exprs/vbloom_predicate.h index 410bb5c8d370b3..1697e524f11152 100644 --- a/be/src/exprs/vbloom_predicate.h +++ b/be/src/exprs/vbloom_predicate.h @@ -49,6 +49,14 @@ class VBloomPredicate final : public VExpr { const uint8_t* __restrict filter, size_t count, ColumnPtr& result_column, ColumnPtr* arg_column) const override; + bool can_execute_on_raw_fixed_values(const DataTypePtr& data_type, + int column_id) const override; + Status execute_on_raw_fixed_values(const uint8_t* values, size_t num_values, size_t value_width, + const DataTypePtr& data_type, int column_id, + uint8_t* matches) const override; + ZoneMapFilterResult evaluate_dictionary_filter(const DictionaryEvalContext& ctx) const override; + bool can_evaluate_dictionary_filter() const override; + Status prepare(RuntimeState* state, const RowDescriptor& desc, VExprContext* context) override; Status open(RuntimeState* state, VExprContext* context, FunctionContext::FunctionStateScope scope) override; diff --git a/be/src/exprs/vdirect_in_predicate.h b/be/src/exprs/vdirect_in_predicate.h index 2fd1e9a35febc7..7a74815a62271f 100644 --- a/be/src/exprs/vdirect_in_predicate.h +++ b/be/src/exprs/vdirect_in_predicate.h @@ -98,6 +98,54 @@ class VDirectInPredicate final : public VExpr { std::dynamic_pointer_cast(get_child(0)) != nullptr; } + ZoneMapFilterResult evaluate_dictionary_filter( + const DictionaryEvalContext& ctx) const override { + return expr_zonemap::eval_in_dictionary(ctx, get_child(0), false, _seg_filter_values); + } + + bool can_evaluate_dictionary_filter() const override { + return _zonemap_materialized && + std::dynamic_pointer_cast(get_child(0)) != nullptr; + } + + bool can_execute_on_raw_fixed_values(const DataTypePtr& data_type, + int column_id) const override { + if (!_hybrid_set_values_match_child_type || data_type == nullptr || _filter == nullptr || + get_num_children() != 1) { + return false; + } + const auto slot = std::dynamic_pointer_cast(get_child(0)); + if (slot == nullptr || slot->column_id() != column_id) { + return false; + } + const auto raw_type = remove_nullable(data_type); + if (!remove_nullable(slot->data_type())->equals(*raw_type)) { + return false; + } + return _raw_fixed_value_size(raw_type->get_primitive_type()) != 0; + } + + Status execute_on_raw_fixed_values(const uint8_t* values, size_t num_values, size_t value_width, + const DataTypePtr& data_type, int column_id, + uint8_t* matches) const override { + if (!can_execute_on_raw_fixed_values(data_type, column_id)) { + return Status::NotSupported( + "Direct IN predicate cannot evaluate raw fixed-width values"); + } + DORIS_CHECK(values != nullptr || num_values == 0); + DORIS_CHECK(matches != nullptr || num_values == 0); + const size_t expected_width = + _raw_fixed_value_size(remove_nullable(data_type)->get_primitive_type()); + if (value_width != expected_width) { + return Status::Corruption("Raw direct IN width {} does not match expected {}", + value_width, expected_width); + } + // Dispatch once per decoder batch so large runtime-filter sets retain the typed HybridSet + // loop instead of paying a virtual lookup for every physical value. + _filter->find_batch_raw_fixed(values, num_values, value_width, matches); + return Status::OK(); + } + Status clone_node(VExprSPtr* cloned_expr) const override { DORIS_CHECK(cloned_expr != nullptr); *cloned_expr = VDirectInPredicate::create_shared(clone_texpr_node(), _filter, @@ -155,6 +203,37 @@ class VDirectInPredicate final : public VExpr { } private: + static size_t _raw_fixed_value_size(PrimitiveType primitive_type) { + switch (primitive_type) { +#define RETURN_RAW_FIXED_SIZE(TYPE) \ + case TYPE: \ + return sizeof(typename PrimitiveTypeTraits::CppType) + RETURN_RAW_FIXED_SIZE(TYPE_BOOLEAN); + RETURN_RAW_FIXED_SIZE(TYPE_TINYINT); + RETURN_RAW_FIXED_SIZE(TYPE_SMALLINT); + RETURN_RAW_FIXED_SIZE(TYPE_INT); + RETURN_RAW_FIXED_SIZE(TYPE_BIGINT); + RETURN_RAW_FIXED_SIZE(TYPE_LARGEINT); + RETURN_RAW_FIXED_SIZE(TYPE_FLOAT); + RETURN_RAW_FIXED_SIZE(TYPE_DOUBLE); + RETURN_RAW_FIXED_SIZE(TYPE_DATE); + RETURN_RAW_FIXED_SIZE(TYPE_DATETIME); + RETURN_RAW_FIXED_SIZE(TYPE_DATEV2); + RETURN_RAW_FIXED_SIZE(TYPE_DATETIMEV2); + RETURN_RAW_FIXED_SIZE(TYPE_TIMESTAMPTZ); + RETURN_RAW_FIXED_SIZE(TYPE_DECIMAL32); + RETURN_RAW_FIXED_SIZE(TYPE_DECIMAL64); + RETURN_RAW_FIXED_SIZE(TYPE_DECIMALV2); + RETURN_RAW_FIXED_SIZE(TYPE_DECIMAL128I); + RETURN_RAW_FIXED_SIZE(TYPE_DECIMAL256); + RETURN_RAW_FIXED_SIZE(TYPE_IPV4); + RETURN_RAW_FIXED_SIZE(TYPE_IPV6); +#undef RETURN_RAW_FIXED_SIZE + default: + return 0; + } + } + Status _do_execute(VExprContext* context, const Block* block, const uint8_t* __restrict filter, const Selector* selector, size_t count, ColumnPtr& result_column, ColumnPtr* arg_column) const { diff --git a/be/src/format_v2/parquet/parquet_scan.cpp b/be/src/format_v2/parquet/parquet_scan.cpp index 4dba81f20a801a..e736a09f23c165 100644 --- a/be/src/format_v2/parquet/parquet_scan.cpp +++ b/be/src/format_v2/parquet/parquet_scan.cpp @@ -30,6 +30,7 @@ #include "common/status.h" #include "core/assert_cast.h" #include "core/block/block.h" +#include "core/column/column_decimal.h" #include "core/column/column_vector.h" #include "exprs/expr_zonemap_filter.h" #include "exprs/vcompound_pred.h" @@ -1418,17 +1419,34 @@ bool get_fixed_dictionary_raw_values(const IColumn& dictionary, const uint8_t** return true; } -bool get_numeric_dictionary_raw_values(PrimitiveType primitive_type, const IColumn& dictionary, - const uint8_t** values, size_t* value_width) { +bool get_typed_dictionary_raw_values(PrimitiveType primitive_type, const IColumn& dictionary, + const uint8_t** values, size_t* value_width) { switch (primitive_type) { - case TYPE_INT: - return get_fixed_dictionary_raw_values(dictionary, values, value_width); - case TYPE_BIGINT: - return get_fixed_dictionary_raw_values(dictionary, values, value_width); - case TYPE_FLOAT: - return get_fixed_dictionary_raw_values(dictionary, values, value_width); - case TYPE_DOUBLE: - return get_fixed_dictionary_raw_values(dictionary, values, value_width); +#define GET_TYPED_DICTIONARY_VALUES(TYPE) \ + case TYPE: \ + return get_fixed_dictionary_raw_values::ColumnType>( \ + dictionary, values, value_width) + GET_TYPED_DICTIONARY_VALUES(TYPE_BOOLEAN); + GET_TYPED_DICTIONARY_VALUES(TYPE_TINYINT); + GET_TYPED_DICTIONARY_VALUES(TYPE_SMALLINT); + GET_TYPED_DICTIONARY_VALUES(TYPE_INT); + GET_TYPED_DICTIONARY_VALUES(TYPE_BIGINT); + GET_TYPED_DICTIONARY_VALUES(TYPE_LARGEINT); + GET_TYPED_DICTIONARY_VALUES(TYPE_FLOAT); + GET_TYPED_DICTIONARY_VALUES(TYPE_DOUBLE); + GET_TYPED_DICTIONARY_VALUES(TYPE_DATE); + GET_TYPED_DICTIONARY_VALUES(TYPE_DATETIME); + GET_TYPED_DICTIONARY_VALUES(TYPE_DATEV2); + GET_TYPED_DICTIONARY_VALUES(TYPE_DATETIMEV2); + GET_TYPED_DICTIONARY_VALUES(TYPE_TIMESTAMPTZ); + GET_TYPED_DICTIONARY_VALUES(TYPE_DECIMAL32); + GET_TYPED_DICTIONARY_VALUES(TYPE_DECIMAL64); + GET_TYPED_DICTIONARY_VALUES(TYPE_DECIMALV2); + GET_TYPED_DICTIONARY_VALUES(TYPE_DECIMAL128I); + GET_TYPED_DICTIONARY_VALUES(TYPE_DECIMAL256); + GET_TYPED_DICTIONARY_VALUES(TYPE_IPV4); + GET_TYPED_DICTIONARY_VALUES(TYPE_IPV6); +#undef GET_TYPED_DICTIONARY_VALUES default: return false; } @@ -1545,8 +1563,8 @@ Status build_dictionary_entry_filter(size_t block_position, return conjunct->root()->can_execute_on_raw_fixed_values( column_schema.type, expression_column_id); }) && - get_numeric_dictionary_raw_values(typed_data_type->get_primitive_type(), dictionary, - &raw_values, &value_width)) { + get_typed_dictionary_raw_values(typed_data_type->get_primitive_type(), dictionary, + &raw_values, &value_width)) { // A dictionary is immutable for the row group, so compare its contiguous typed values once // and reuse the resulting id bitmap for every data page. for (const auto& conjunct : conjuncts) { 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 b7bea8e7053f3e..2af1fe237b0764 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 @@ -1644,8 +1644,10 @@ bool ColumnChunkReader::can_filter_fixed_width_valu } const auto primitive_type = remove_nullable(_field_schema->data_type)->get_primitive_type(); const bool has_identity_width = - (_metadata.type == tparquet::Type::INT32 && primitive_type == TYPE_INT) || - (_metadata.type == tparquet::Type::INT64 && primitive_type == TYPE_BIGINT) || + (_metadata.type == tparquet::Type::INT32 && + (primitive_type == TYPE_INT || primitive_type == TYPE_DECIMAL32)) || + (_metadata.type == tparquet::Type::INT64 && + (primitive_type == TYPE_BIGINT || primitive_type == TYPE_DECIMAL64)) || (_metadata.type == tparquet::Type::FLOAT && primitive_type == TYPE_FLOAT) || (_metadata.type == tparquet::Type::DOUBLE && primitive_type == TYPE_DOUBLE); if (!has_identity_width) { diff --git a/be/src/format_v2/parquet/reader/native/decoder.h b/be/src/format_v2/parquet/reader/native/decoder.h index 379176d5e459e0..628a34ac97675e 100644 --- a/be/src/format_v2/parquet/reader/native/decoder.h +++ b/be/src/format_v2/parquet/reader/native/decoder.h @@ -182,6 +182,12 @@ class BaseDictDecoder : public Decoder { std::vector* indices) override { DORIS_CHECK(indices != nullptr); const size_t num_dictionary_values = dictionary_size(); + if (_is_fragmented_selection(selection)) { + RETURN_IF_ERROR(_decode_fragmented_selection(selection, num_dictionary_values)); + indices->assign(_skip_indices.begin(), + _skip_indices.begin() + selection.selected_values); + return Status::OK(); + } indices->resize(selection.selected_values); size_t cursor = 0; size_t output = 0; @@ -223,6 +229,10 @@ class BaseDictDecoder : public Decoder { Status decode_selected_dictionary_values(const ParquetSelection& selection, ParquetDictionaryValueConsumer& consumer) override { const size_t num_dictionary_values = dictionary_size(); + if (_is_fragmented_selection(selection)) { + RETURN_IF_ERROR(_decode_fragmented_selection(selection, num_dictionary_values)); + return consumer.consume_indices(_skip_indices.data(), selection.selected_values); + } size_t cursor = 0; for (const auto& range : selection.ranges) { DORIS_CHECK(range.first >= cursor); @@ -246,6 +256,57 @@ class BaseDictDecoder : public Decoder { size_t active_scratch_bytes() const override { return _skip_indices.size() * sizeof(uint32_t); } protected: + static bool _is_fragmented_selection(const ParquetSelection& selection) { + constexpr size_t MIN_FRAGMENTED_RANGES = 8; + constexpr size_t MAX_AVERAGE_RANGE_VALUES = 4; + constexpr size_t MAX_DECODE_EXPANSION = 8; + return selection.ranges.size() >= MIN_FRAGMENTED_RANGES && selection.selected_values != 0 && + selection.total_values / selection.selected_values <= MAX_DECODE_EXPANSION && + selection.selected_values <= selection.ranges.size() * MAX_AVERAGE_RANGE_VALUES; + } + + Status _decode_fragmented_selection(const ParquetSelection& selection, + size_t num_dictionary_values) { + // Decode and validate the page batch once when predicate survivors alternate in tiny runs. + // Walking each range separately turns one RLE batch into millions of decoder calls for + // low-cardinality predicates such as TPC-DS quantity buckets. + _skip_indices.resize(selection.total_values); + const auto decoded = _index_batch_decoder->GetBatch( + _skip_indices.data(), cast_set(selection.total_values)); + if (UNLIKELY(decoded != selection.total_values)) { + return Status::IOError("Can't read enough Parquet dictionary indices"); + } + if (UNLIKELY(!dictionary_indices_in_bounds(_skip_indices.data(), selection.total_values, + num_dictionary_values))) { + for (size_t row = 0; row < selection.total_values; ++row) { + if (_skip_indices[row] < num_dictionary_values) { + continue; + } + return Status::Corruption( + "Parquet dictionary index {} at row {} exceeds dictionary size {}", + _skip_indices[row], row, num_dictionary_values); + } + } + size_t output = 0; + constexpr size_t MAX_INLINE_COPY_VALUES = 4; + for (const auto& range : selection.ranges) { + DORIS_CHECK(range.first + range.count <= selection.total_values); + // Alternating predicates mostly produce one-row spans; inline tiny forward copies so + // range compaction does not replace decoder calls with equally numerous libc calls. + if (range.count <= MAX_INLINE_COPY_VALUES) { + for (size_t row = 0; row < range.count; ++row) { + _skip_indices[output + row] = _skip_indices[range.first + row]; + } + } else { + memmove(_skip_indices.data() + output, _skip_indices.data() + range.first, + range.count * sizeof(uint32_t)); + } + output += range.count; + } + DORIS_CHECK_EQ(output, selection.selected_values); + return Status::OK(); + } + Status skip_values(size_t num_values) override { return _decode_and_validate_skipped(num_values, 0, dictionary_size()); } diff --git a/be/test/exprs/bloom_filter_func_test.cpp b/be/test/exprs/bloom_filter_func_test.cpp index 5e0c02479be719..e8e5cdee5bdc12 100644 --- a/be/test/exprs/bloom_filter_func_test.cpp +++ b/be/test/exprs/bloom_filter_func_test.cpp @@ -20,6 +20,7 @@ #include #include +#include #include #include @@ -227,6 +228,134 @@ TEST_F(BloomFilterFuncTest, InsertFixedLen) { ASSERT_EQ(offsets[1], 2); } +TEST_F(BloomFilterFuncTest, RawFixedCapabilitiesCoverEveryFixedRuntimeFilterType) { +#define EXPECT_RAW_FIXED(TYPE) \ + do { \ + BloomFilterFunc filter(false); \ + EXPECT_TRUE(filter.supports_raw_fixed_values()) << type_to_string(TYPE); \ + EXPECT_EQ(filter.raw_fixed_value_size(), \ + sizeof(typename PrimitiveTypeTraits::CppType)) \ + << type_to_string(TYPE); \ + } while (false) + EXPECT_RAW_FIXED(TYPE_BOOLEAN); + EXPECT_RAW_FIXED(TYPE_TINYINT); + EXPECT_RAW_FIXED(TYPE_SMALLINT); + EXPECT_RAW_FIXED(TYPE_INT); + EXPECT_RAW_FIXED(TYPE_BIGINT); + EXPECT_RAW_FIXED(TYPE_LARGEINT); + EXPECT_RAW_FIXED(TYPE_FLOAT); + EXPECT_RAW_FIXED(TYPE_DOUBLE); + EXPECT_RAW_FIXED(TYPE_DATE); + EXPECT_RAW_FIXED(TYPE_DATETIME); + EXPECT_RAW_FIXED(TYPE_DATEV2); + EXPECT_RAW_FIXED(TYPE_DATETIMEV2); + EXPECT_RAW_FIXED(TYPE_TIMESTAMPTZ); + EXPECT_RAW_FIXED(TYPE_DECIMAL32); + EXPECT_RAW_FIXED(TYPE_DECIMAL64); + EXPECT_RAW_FIXED(TYPE_DECIMALV2); + EXPECT_RAW_FIXED(TYPE_DECIMAL128I); + EXPECT_RAW_FIXED(TYPE_DECIMAL256); + EXPECT_RAW_FIXED(TYPE_IPV4); + EXPECT_RAW_FIXED(TYPE_IPV6); +#undef EXPECT_RAW_FIXED + + BloomFilterFunc string_filter(false); + EXPECT_FALSE(string_filter.supports_raw_fixed_values()); + EXPECT_EQ(string_filter.raw_fixed_value_size(), 0); +} + +TEST_F(BloomFilterFuncTest, RawFixedProbeUsesTheSameHashAsColumnProbe) { + BloomFilterFunc filter(false); + RuntimeFilterParams params { + 1, RuntimeFilterType::BLOOM_FILTER, TYPE_INT, false, 0, 0, 0, 256, 0, 0}; + filter.init_params(¶ms); + ASSERT_TRUE(filter.init_with_fixed_length(1024).ok()); + auto build_column = ColumnHelper::create_column({2, 4}); + filter.insert_fixed_len(build_column, 0); + + const std::array values {1, 2, 3, 4}; + std::array raw_matches {1, 1, 1, 1}; + ASSERT_TRUE(filter.find_batch_raw_fixed(reinterpret_cast(values.data()), + values.size(), sizeof(int32_t), raw_matches.data()) + .ok()); + + auto probe_column = ColumnHelper::create_column({1, 2, 3, 4}); + std::array column_matches {}; + filter.find_fixed_len(probe_column, column_matches.data()); + EXPECT_EQ(column_matches, raw_matches); +} + +TEST_F(BloomFilterFuncTest, RawFixedProbeMatchesBuildHashForEveryFixedRuntimeFilterType) { + const auto expect_match = []() { + BloomFilterFunc filter(false); + RuntimeFilterParams params; + params.filter_type = RuntimeFilterType::BLOOM_FILTER; + params.column_return_type = TYPE; + params.bloom_filter_size = 1024; + filter.init_params(¶ms); + ASSERT_TRUE(filter.init_with_fixed_length(1024).ok()) << type_to_string(TYPE); + + using ValueType = typename PrimitiveTypeTraits::CppType; + ValueType value {}; + auto set = std::make_shared>(false); + set->insert(&value); + filter.insert_set(set); + + uint8_t match = 1; + ASSERT_TRUE(filter.find_batch_raw_fixed(reinterpret_cast(&value), 1, + sizeof(ValueType), &match) + .ok()) + << type_to_string(TYPE); + EXPECT_EQ(match, 1) << type_to_string(TYPE); + EXPECT_TRUE(filter.test_field(Field::create_field(value))) << type_to_string(TYPE); + }; + +#define EXPECT_RAW_FIXED_MATCH(TYPE) expect_match.template operator()() + EXPECT_RAW_FIXED_MATCH(TYPE_BOOLEAN); + EXPECT_RAW_FIXED_MATCH(TYPE_TINYINT); + EXPECT_RAW_FIXED_MATCH(TYPE_SMALLINT); + EXPECT_RAW_FIXED_MATCH(TYPE_INT); + EXPECT_RAW_FIXED_MATCH(TYPE_BIGINT); + EXPECT_RAW_FIXED_MATCH(TYPE_LARGEINT); + EXPECT_RAW_FIXED_MATCH(TYPE_FLOAT); + EXPECT_RAW_FIXED_MATCH(TYPE_DOUBLE); + EXPECT_RAW_FIXED_MATCH(TYPE_DATE); + EXPECT_RAW_FIXED_MATCH(TYPE_DATETIME); + EXPECT_RAW_FIXED_MATCH(TYPE_DATEV2); + EXPECT_RAW_FIXED_MATCH(TYPE_DATETIMEV2); + EXPECT_RAW_FIXED_MATCH(TYPE_TIMESTAMPTZ); + EXPECT_RAW_FIXED_MATCH(TYPE_DECIMAL32); + EXPECT_RAW_FIXED_MATCH(TYPE_DECIMAL64); + EXPECT_RAW_FIXED_MATCH(TYPE_DECIMALV2); + EXPECT_RAW_FIXED_MATCH(TYPE_DECIMAL128I); + EXPECT_RAW_FIXED_MATCH(TYPE_DECIMAL256); + EXPECT_RAW_FIXED_MATCH(TYPE_IPV4); + EXPECT_RAW_FIXED_MATCH(TYPE_IPV6); +#undef EXPECT_RAW_FIXED_MATCH +} + +TEST_F(BloomFilterFuncTest, DictionaryFieldProbeSupportsEveryStringRuntimeFilterType) { + const auto expect_match = []() { + BloomFilterFunc filter(false); + RuntimeFilterParams params; + params.filter_type = RuntimeFilterType::BLOOM_FILTER; + params.column_return_type = TYPE; + params.bloom_filter_size = 1024; + filter.init_params(¶ms); + ASSERT_TRUE(filter.init_with_fixed_length(1024).ok()) << type_to_string(TYPE); + + auto column = ColumnString::create(); + column->insert_data("value", 5); + filter.insert_fixed_len(std::move(column), 0); + EXPECT_TRUE(filter.test_field(Field::create_field(std::string("value")))) + << type_to_string(TYPE); + }; + + expect_match.template operator()(); + expect_match.template operator()(); + expect_match.template operator()(); +} + TEST_F(BloomFilterFuncTest, Merge) { BloomFilterFunc bloom_filter_func(false); const size_t runtime_length = 1024; diff --git a/be/test/exprs/expr_zonemap_filter_test.cpp b/be/test/exprs/expr_zonemap_filter_test.cpp index 6e7e42f7baa2a4..6d29cbf4d085ca 100644 --- a/be/test/exprs/expr_zonemap_filter_test.cpp +++ b/be/test/exprs/expr_zonemap_filter_test.cpp @@ -19,6 +19,7 @@ #include +#include #include #include #include @@ -26,6 +27,7 @@ #include #include "common/object_pool.h" +#include "core/column/column_vector.h" #include "core/data_type/data_type_date_or_datetime_v2.h" #include "core/data_type/data_type_decimal.h" #include "core/data_type/data_type_nullable.h" @@ -34,11 +36,13 @@ #include "core/field.h" #include "core/string_ref.h" #include "core/value/vdatetime_value.h" +#include "exprs/bloom_filter_func.h" #include "exprs/create_predicate_function.h" #include "exprs/function/functions_comparison.h" #include "exprs/function/simple_function_factory.h" #include "exprs/hybrid_set.h" #include "exprs/runtime_filter_expr.h" +#include "exprs/vbloom_predicate.h" #include "exprs/vcompound_pred.h" #include "exprs/vexpr.h" #include "exprs/vexpr_context.h" @@ -908,6 +912,97 @@ TEST(ExprZonemapFilterTest, RuntimeFilterExprNullAwareZonemapKeepsZonesWithNull) runtime_filter->evaluate_zonemap_filter(only_null_ctx)); } +TEST(ExprZonemapFilterTest, RuntimeFilterExprDelegatesDirectInDictionaryAndRawEvaluation) { + auto type = int_type(); + auto slot = make_slot(0, type); + std::shared_ptr filter(create_set(PrimitiveType::TYPE_INT, false)); + int32_t two = 2; + int32_t four = 4; + filter->insert(&two); + filter->insert(&four); + + auto direct_in_expr = + std::make_shared(make_in_predicate_node(false, 1), filter, true); + direct_in_expr->add_child(slot); + ASSERT_TRUE(direct_in_expr->_materialize_for_zonemap_filter().ok()); + + auto runtime_filter = RuntimeFilterExpr::create_shared(make_in_predicate_node(false, 1), + direct_in_expr, 0.0, false, 7); + EXPECT_TRUE(runtime_filter->can_evaluate_dictionary_filter()); + EXPECT_EQ(ZoneMapFilterResult::kNoMatch, + runtime_filter->evaluate_dictionary_filter( + make_dictionary_context({int_field(1), int_field(3)}, type))); + EXPECT_EQ(ZoneMapFilterResult::kMayMatch, + runtime_filter->evaluate_dictionary_filter( + make_dictionary_context({int_field(3), int_field(4)}, type))); + + EXPECT_TRUE(runtime_filter->can_execute_on_raw_fixed_values(type, 0)); + const std::array values {1, 2, 3, 4}; + std::array matches {1, 1, 1, 1}; + ASSERT_TRUE(runtime_filter + ->execute_on_raw_fixed_values( + reinterpret_cast(values.data()), values.size(), + sizeof(int32_t), type, 0, matches.data()) + .ok()); + EXPECT_EQ((std::array {0, 1, 0, 1}), matches); + + auto null_aware_runtime_filter = RuntimeFilterExpr::create_shared( + make_in_predicate_node(false, 1), direct_in_expr, 0.0, true, 8); + EXPECT_FALSE(null_aware_runtime_filter->can_evaluate_dictionary_filter()); + EXPECT_FALSE(null_aware_runtime_filter->can_execute_on_raw_fixed_values(type, 0)); +} + +TEST(ExprZonemapFilterTest, RuntimeFilterExprDelegatesBloomDictionaryAndRawEvaluation) { + auto type = int_type(); + std::shared_ptr filter(create_bloom_filter(TYPE_INT, false)); + RuntimeFilterParams params; + params.filter_type = RuntimeFilterType::BLOOM_FILTER; + params.column_return_type = TYPE_INT; + params.bloom_filter_size = 1024; + filter->init_params(¶ms); + ASSERT_TRUE(filter->init_with_fixed_length(1024).ok()); + auto build_values = ColumnInt32::create(); + build_values->insert_value(2); + build_values->insert_value(4); + filter->insert_fixed_len(std::move(build_values), 0); + + auto node = make_in_predicate_node(false, 1); + node.__set_node_type(TExprNodeType::BLOOM_PRED); + node.__set_opcode(TExprOpcode::RT_FILTER); + auto bloom = VBloomPredicate::create_shared(node); + bloom->set_filter(filter); + bloom->add_child(make_slot(0, type)); + auto runtime_filter = RuntimeFilterExpr::create_shared(node, bloom, 0.0, false, 9); + + EXPECT_TRUE(runtime_filter->can_evaluate_dictionary_filter()); + EXPECT_EQ(ZoneMapFilterResult::kMayMatch, + runtime_filter->evaluate_dictionary_filter( + make_dictionary_context({int_field(2)}, type))); + + int32_t missing = 1; + while (missing < 10000 && filter->test_field(int_field(missing))) { + ++missing; + } + ASSERT_LT(missing, 10000); + EXPECT_EQ(ZoneMapFilterResult::kNoMatch, + runtime_filter->evaluate_dictionary_filter( + make_dictionary_context({int_field(missing)}, type))); + + EXPECT_TRUE(runtime_filter->can_execute_on_raw_fixed_values(type, 0)); + const std::array values {missing, 2, 4}; + std::array matches {1, 1, 1}; + ASSERT_TRUE(runtime_filter + ->execute_on_raw_fixed_values( + reinterpret_cast(values.data()), values.size(), + sizeof(int32_t), type, 0, matches.data()) + .ok()); + EXPECT_EQ((std::array {0, 1, 1}), matches); + + auto null_aware_runtime_filter = RuntimeFilterExpr::create_shared(node, bloom, 0.0, true, 10); + EXPECT_FALSE(null_aware_runtime_filter->can_evaluate_dictionary_filter()); + EXPECT_FALSE(null_aware_runtime_filter->can_execute_on_raw_fixed_values(type, 0)); +} + TEST(ExprZonemapFilterTest, CompoundPredicateEvaluatesChildrenForZonemap) { ZoneMapEvalContext ctx; diff --git a/be/test/format_v2/parquet/native_decoder_test.cpp b/be/test/format_v2/parquet/native_decoder_test.cpp index b71e94222a36c5..0d037f61549281 100644 --- a/be/test/format_v2/parquet/native_decoder_test.cpp +++ b/be/test/format_v2/parquet/native_decoder_test.cpp @@ -264,6 +264,24 @@ class CaptureFixedConsumer final : public ParquetFixedValueConsumer { std::vector bytes; }; +class CaptureDictionaryConsumer final : public ParquetDictionaryValueConsumer { +public: + Status consume_indices(const uint32_t* values, size_t num_values) override { + ++consume_calls; + indices.insert(indices.end(), values, values + num_values); + return Status::OK(); + } + + Status consume_repeated(uint32_t value, size_t num_values) override { + ++consume_calls; + indices.insert(indices.end(), num_values, value); + return Status::OK(); + } + + size_t consume_calls = 0; + std::vector indices; +}; + TEST(ParquetV2NativeDecoderTest, FragmentedPlainSelectionUsesOneConsumerBatch) { std::array input {}; std::iota(input.begin(), input.end(), 0); @@ -288,6 +306,90 @@ TEST(ParquetV2NativeDecoderTest, FragmentedPlainSelectionUsesOneConsumerBatch) { EXPECT_EQ(consumer.values(), expected); } +TEST(ParquetV2NativeDecoderTest, FragmentedDictionarySelectionUsesOneConsumerBatch) { + constexpr size_t VALUE_COUNT = 32; + std::array dictionary_values {}; + std::iota(dictionary_values.begin(), dictionary_values.end(), 0); + auto dictionary = make_unique_buffer(sizeof(dictionary_values)); + memcpy(dictionary.get(), dictionary_values.data(), sizeof(dictionary_values)); + + std::unique_ptr decoder; + ASSERT_TRUE( + Decoder::get_decoder(tparquet::Type::INT32, tparquet::Encoding::RLE_DICTIONARY, decoder) + .ok()); + decoder->set_type_length(sizeof(int32_t)); + ASSERT_TRUE(decoder->set_dict(dictionary, sizeof(dictionary_values), dictionary_values.size()) + .ok()); + + faststring encoded_ids; + RleEncoder encoder(&encoded_ids, 5); + for (uint32_t id = 0; id < VALUE_COUNT; ++id) { + encoder.Put(id); + } + encoder.Flush(); + std::vector payload(encoded_ids.size() + 1); + payload[0] = 5; + memcpy(payload.data() + 1, encoded_ids.data(), encoded_ids.size()); + Slice data(payload.data(), payload.size()); + ASSERT_TRUE(decoder->set_data(&data).ok()); + + ParquetSelection selection { + .total_values = VALUE_COUNT, .selected_values = VALUE_COUNT / 2, .ranges = {}}; + std::vector expected; + for (uint32_t id = 0; id < VALUE_COUNT; id += 2) { + selection.ranges.push_back({.first = id, .count = 1}); + expected.push_back(id); + } + CaptureDictionaryConsumer consumer; + ASSERT_TRUE(decoder->decode_selected_dictionary_values(selection, consumer).ok()); + + EXPECT_EQ(consumer.consume_calls, 1); + EXPECT_EQ(consumer.indices, expected); + + ASSERT_TRUE(decoder->set_data(&data).ok()); + std::vector selected_indices; + ASSERT_TRUE(decoder->decode_selected_dictionary_indices(selection, &selected_indices).ok()); + EXPECT_EQ(selected_indices, expected); +} + +TEST(ParquetV2NativeDecoderTest, HighlySparseDictionarySelectionAvoidsFullBatchDecode) { + constexpr size_t DICTIONARY_SIZE = 32; + constexpr size_t VALUE_COUNT = 256; + std::array dictionary_values {}; + std::iota(dictionary_values.begin(), dictionary_values.end(), 0); + auto dictionary = make_unique_buffer(sizeof(dictionary_values)); + memcpy(dictionary.get(), dictionary_values.data(), sizeof(dictionary_values)); + + std::unique_ptr decoder; + ASSERT_TRUE( + Decoder::get_decoder(tparquet::Type::INT32, tparquet::Encoding::RLE_DICTIONARY, decoder) + .ok()); + decoder->set_type_length(sizeof(int32_t)); + ASSERT_TRUE(decoder->set_dict(dictionary, sizeof(dictionary_values), DICTIONARY_SIZE).ok()); + + faststring encoded_ids; + RleEncoder encoder(&encoded_ids, 5); + for (uint32_t row = 0; row < VALUE_COUNT; ++row) { + encoder.Put(row % DICTIONARY_SIZE); + } + encoder.Flush(); + std::vector payload(encoded_ids.size() + 1); + payload[0] = 5; + memcpy(payload.data() + 1, encoded_ids.data(), encoded_ids.size()); + Slice data(payload.data(), payload.size()); + ASSERT_TRUE(decoder->set_data(&data).ok()); + + ParquetSelection selection {.total_values = VALUE_COUNT, .selected_values = 8, .ranges = {}}; + for (size_t row = 0; row < selection.selected_values; ++row) { + selection.ranges.push_back({.first = row * 32, .count = 1}); + } + CaptureDictionaryConsumer consumer; + ASSERT_TRUE(decoder->decode_selected_dictionary_values(selection, consumer).ok()); + + EXPECT_EQ(consumer.consume_calls, selection.ranges.size()); + EXPECT_EQ(consumer.indices, std::vector(selection.selected_values, 0)); +} + class ScriptedDictionaryMaterializationSource final : public ParquetDecodeSource { public: ScriptedDictionaryMaterializationSource(std::vector ids, bool prefer_indices) @@ -4606,6 +4708,73 @@ TEST(ParquetV2NativeDecoderTest, DictionaryStringGatherAppendsCompactSurvivors) EXPECT_EQ(destination.get_data_at(3).to_string_view(), "charlie"); } +TEST(ParquetV2NativeDecoderTest, DictionaryDecimalGatherAppendsCompactSurvivors) { + const std::array indices {3, 1, 2, 0, 0, 2, 1, 3}; + const auto verify = [&]() { + using ValueType = typename ColumnType::value_type; + auto dictionary = ColumnType::create(0, 2); + auto& dictionary_data = dictionary->get_data(); + dictionary_data.resize(4); + for (size_t row = 0; row < dictionary_data.size(); ++row) { + dictionary_data[row].value = static_cast(100 * (row + 1)); + } + auto destination = ColumnType::create(0, 2); + destination->get_data().push_back(ValueType {50}); + + ASSERT_TRUE(try_simd_insert_parquet_dictionary_indices(*destination, *dictionary, + indices.data(), indices.size())); + ASSERT_EQ(destination->size(), indices.size() + 1); + EXPECT_EQ(destination->get_data()[0].value, 50); + for (size_t row = 0; row < indices.size(); ++row) { + EXPECT_EQ(destination->get_data()[row + 1].value, dictionary_data[indices[row]].value); + } + }; + verify.template operator()(); + verify.template operator()(); +} + +template +void expect_fixed_width_dictionary_gather() { + using ColumnType = ColumnVector; + using ValueType = typename ColumnType::value_type; + static_assert(sizeof(ValueType) == 4 || sizeof(ValueType) == 8); + + auto dictionary = ColumnType::create(); + auto& dictionary_data = dictionary->get_data(); + dictionary_data.resize(4); + for (size_t row = 0; row < dictionary_data.size(); ++row) { + const uint64_t bits = 0x0102030405060708ULL + row; + memcpy(&dictionary_data[row], &bits, sizeof(ValueType)); + } + auto destination = ColumnType::create(); + destination->get_data().push_back(dictionary_data[0]); + const std::array indices {3, 1, 2, 0, 0, 2, 1, 3}; + + ASSERT_TRUE(try_simd_insert_parquet_dictionary_indices(*destination, *dictionary, + indices.data(), indices.size())); + ASSERT_EQ(destination->size(), indices.size() + 1); + for (size_t row = 0; row < indices.size(); ++row) { + EXPECT_EQ(0, memcmp(&destination->get_data()[row + 1], &dictionary_data[indices[row]], + sizeof(ValueType))); + } +} + +TEST(ParquetV2NativeDecoderTest, DictionaryGatherSupportsEverySimdFixedWidthColumn) { + expect_fixed_width_dictionary_gather(); + expect_fixed_width_dictionary_gather(); + expect_fixed_width_dictionary_gather(); + expect_fixed_width_dictionary_gather(); + expect_fixed_width_dictionary_gather(); + expect_fixed_width_dictionary_gather(); + expect_fixed_width_dictionary_gather(); + expect_fixed_width_dictionary_gather(); + expect_fixed_width_dictionary_gather(); + expect_fixed_width_dictionary_gather(); + expect_fixed_width_dictionary_gather(); + expect_fixed_width_dictionary_gather(); + expect_fixed_width_dictionary_gather(); +} + TEST(ParquetV2NativeDecoderTest, ComplexPageStatisticsPreservePerLeafCrossings) { ColumnChunkReaderStatistics first_chunk; first_chunk.page_read_counter = 1; diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp b/be/test/format_v2/parquet/parquet_scan_test.cpp index 11cfdf99641c10..f3049a3d03adf8 100644 --- a/be/test/format_v2/parquet/parquet_scan_test.cpp +++ b/be/test/format_v2/parquet/parquet_scan_test.cpp @@ -46,7 +46,12 @@ #include "core/data_type/data_type_number.h" #include "core/data_type/data_type_string.h" #include "core/field.h" +#include "exprs/bloom_filter_func.h" +#include "exprs/create_predicate_function.h" +#include "exprs/runtime_filter_expr.h" +#include "exprs/vbloom_predicate.h" #include "exprs/vcompound_pred.h" +#include "exprs/vdirect_in_predicate.h" #include "exprs/vectorized_fn_call.h" #include "exprs/vexpr.h" #include "exprs/vexpr_context.h" @@ -597,6 +602,104 @@ VExprContextSPtr create_int32_direct_greater_conjunct(int column_id, int32_t low std::make_shared(column_id, lower_bound)); } +TExprNode make_runtime_in_node() { + TExprNode node; + node.__set_type(std::make_shared()->to_thrift()); + node.__set_node_type(TExprNodeType::IN_PRED); + node.in_predicate.__set_is_not_in(false); + node.__set_opcode(TExprOpcode::FILTER_IN); + node.__set_is_nullable(false); + return node; +} + +VExprContextSPtr wrap_runtime_filter(VExprSPtr impl, const TExprNode& node, int filter_id) { + auto context = VExprContext::create_shared( + RuntimeFilterExpr::create_shared(node, std::move(impl), 0.0, false, filter_id)); + context->_prepared = true; + context->_opened = true; + return context; +} + +VExprContextSPtr create_int32_runtime_in_conjunct(int column_id, const std::vector& values, + int filter_id) { + std::shared_ptr filter(create_set(PrimitiveType::TYPE_INT, false)); + for (const auto value : values) { + filter->insert(&value); + } + auto node = make_runtime_in_node(); + auto impl = VDirectInPredicate::create_shared(node, std::move(filter), true); + impl->add_child(VSlotRef::create_shared(column_id, column_id, -1, + make_nullable(std::make_shared()), + "runtime_in_key")); + return wrap_runtime_filter(std::move(impl), node, filter_id); +} + +VExprContextSPtr create_int32_runtime_bloom_conjunct(int column_id, + const std::vector& values, + int filter_id) { + std::shared_ptr filter(create_bloom_filter(TYPE_INT, false)); + RuntimeFilterParams params; + params.filter_type = RuntimeFilterType::BLOOM_FILTER; + params.column_return_type = TYPE_INT; + params.bloom_filter_size = 1024; + filter->init_params(¶ms); + EXPECT_TRUE(filter->init_with_fixed_length(1024).ok()); + auto build_values = ColumnInt32::create(); + for (const auto value : values) { + build_values->insert_value(value); + } + filter->insert_fixed_len(std::move(build_values), 0); + + auto node = make_runtime_in_node(); + node.__set_node_type(TExprNodeType::BLOOM_PRED); + node.__set_opcode(TExprOpcode::RT_FILTER); + auto impl = VBloomPredicate::create_shared(node); + impl->set_filter(std::move(filter)); + impl->add_child(VSlotRef::create_shared(column_id, column_id, -1, + make_nullable(std::make_shared()), + "runtime_bloom_key")); + return wrap_runtime_filter(std::move(impl), node, filter_id); +} + +VExprContextSPtr create_string_runtime_bloom_conjunct(int column_id, + const std::vector& values, + int filter_id) { + std::shared_ptr filter(create_bloom_filter(TYPE_STRING, false)); + RuntimeFilterParams params; + params.filter_type = RuntimeFilterType::BLOOM_FILTER; + params.column_return_type = TYPE_STRING; + params.bloom_filter_size = 1024; + filter->init_params(¶ms); + EXPECT_TRUE(filter->init_with_fixed_length(1024).ok()); + auto build_values = ColumnString::create(); + for (const auto& value : values) { + build_values->insert_data(value.data(), value.size()); + } + filter->insert_fixed_len(std::move(build_values), 0); + + auto node = make_runtime_in_node(); + node.__set_node_type(TExprNodeType::BLOOM_PRED); + node.__set_opcode(TExprOpcode::RT_FILTER); + auto impl = VBloomPredicate::create_shared(node); + impl->set_filter(std::move(filter)); + impl->add_child(VSlotRef::create_shared(column_id, column_id, -1, + make_nullable(std::make_shared()), + "runtime_bloom_key")); + return wrap_runtime_filter(std::move(impl), node, filter_id); +} + +VExprContextSPtr create_int32_runtime_comparison_conjunct(int column_id, + const std::string& function_name, + TExprOpcode::type opcode, int32_t value, + int filter_id) { + auto impl_context = + create_int32_function_conjunct(column_id, function_name, opcode, value, true); + TExprNode node; + node.__set_type(std::make_shared()->to_thrift()); + node.__set_is_nullable(false); + return wrap_runtime_filter(impl_context->root(), node, filter_id); +} + VExprContextSPtr create_int64_direct_greater_conjunct(int column_id, int64_t lower_bound) { return VExprContext::create_shared( std::make_shared(column_id, lower_bound)); @@ -1884,6 +1987,64 @@ TEST_F(ParquetScanTest, PredicateOnlyPlainComparisonUsesPhysicalDirectPath) { EXPECT_EQ(counter_value(profile, "PredicateCompactionBytes"), 0); } +TEST_F(ParquetScanTest, PredicateOnlyPlainRuntimeFiltersUsePhysicalDirectPath) { + 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 schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + auto request = std::make_shared(); + 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->predicate_only_columns.push_back(format::LocalColumnId(0)); + request->conjuncts.push_back( + create_int32_runtime_comparison_conjunct(0, "gt", TExprOpcode::GT, 2, 7)); + request->conjuncts.push_back(create_int32_runtime_in_conjunct(0, {3, 5, 6}, 8)); + 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, 3); + EXPECT_EQ(int32_data_column(*block.get_by_position(1).column).get_data(), + (ColumnInt32::Container {30, 50, 60})); + EXPECT_EQ(counter_value(profile, "FixedWidthPredicateDirectBatches"), 1); + EXPECT_EQ(counter_value(profile, "FixedWidthPredicateDirectRows"), 6); +} + +TEST_F(ParquetScanTest, PredicateOnlyPlainBloomRuntimeFilterUsesPhysicalDirectPath) { + 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 schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + auto request = std::make_shared(); + 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->predicate_only_columns.push_back(format::LocalColumnId(0)); + request->conjuncts.push_back(create_int32_runtime_bloom_conjunct(0, {3, 5, 6}, 9)); + 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, 3); + EXPECT_EQ(int32_data_column(*block.get_by_position(1).column).get_data(), + (ColumnInt32::Container {30, 50, 60})); + EXPECT_EQ(counter_value(profile, "FixedWidthPredicateDirectBatches"), 1); + EXPECT_EQ(counter_value(profile, "FixedWidthPredicateDirectRows"), 6); +} + TEST_F(ParquetScanTest, PredicateOnlyDictionaryRangeSkipsTypedValueMaterialization) { write_dictionary_int_pair_parquet_file(_file_path); RuntimeProfile profile("profile"); @@ -1922,6 +2083,63 @@ TEST_F(ParquetScanTest, PredicateOnlyDictionaryRangeSkipsTypedValueMaterializati conjunct->close(); } +TEST_F(ParquetScanTest, PredicateOnlyDictionaryBloomRuntimeFilterUsesTypedValues) { + write_dictionary_int_pair_parquet_file(_file_path); + RuntimeProfile profile("profile"); + auto reader = create_reader(0, -1, &profile); + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + ASSERT_TRUE(reader->init(&state).ok()); + + std::vector schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + auto request = std::make_shared(); + 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->predicate_only_columns.push_back(format::LocalColumnId(0)); + request->conjuncts.push_back(create_int32_runtime_bloom_conjunct(0, {3, 5, 6}, 10)); + 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, 3); + EXPECT_EQ(int32_data_column(*block.get_by_position(1).column).get_data(), + (ColumnInt32::Container {30, 50, 60})); + EXPECT_EQ(counter_value(profile, "DictFilterColumns"), 1); + EXPECT_EQ(counter_value(profile, "DictFilterTypedCompareColumns"), 1); + EXPECT_EQ(counter_value(profile, "DictionaryPredicateDirectRows"), 6); +} + +TEST_F(ParquetScanTest, PredicateOnlyStringDictionaryBloomRuntimeFilterUsesDictionaryValues) { + write_dictionary_string_pair_parquet_file(_file_path); + RuntimeProfile profile("profile"); + auto reader = create_reader(0, -1, &profile); + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + ASSERT_TRUE(reader->init(&state).ok()); + + std::vector schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + auto request = std::make_shared(); + 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->predicate_only_columns.push_back(format::LocalColumnId(0)); + request->conjuncts.push_back(create_string_runtime_bloom_conjunct(0, {"bravo", "delta"}, 11)); + 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, 2); + EXPECT_EQ(int32_data_column(*block.get_by_position(1).column).get_data(), + (ColumnInt32::Container {20, 40})); + EXPECT_EQ(counter_value(profile, "DictFilterColumns"), 1); + EXPECT_EQ(counter_value(profile, "DictionaryPredicateDirectRows"), 4); +} + TEST_F(ParquetScanTest, ProjectedDictionaryRangeGathersOnlySurvivors) { write_dictionary_int_pair_parquet_file(_file_path); RuntimeProfile profile("profile"); From 8072fce6fdcbbabd3be52336548302061688d6f1 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 27 Jul 2026 19:03:58 +0800 Subject: [PATCH 3/3] [improvement](parquet) Reduce dictionary filter control overhead --- be/src/format_v2/parquet/parquet_scan.cpp | 7 ++- .../parquet/reader/column_reader.cpp | 2 +- .../format_v2/parquet/reader/column_reader.h | 3 +- .../reader/native/column_chunk_reader.cpp | 35 ++++++++----- .../reader/native/column_chunk_reader.h | 4 +- .../parquet/reader/native/column_reader.cpp | 19 ++++--- .../parquet/reader/native/column_reader.h | 16 +++--- .../parquet/reader/native_column_reader.cpp | 52 ++++++++++--------- .../parquet/reader/native_column_reader.h | 7 +-- be/src/format_v2/parquet/selection_vector.h | 22 +++++++- .../parquet/parquet_reader_control_test.cpp | 7 +++ 11 files changed, 113 insertions(+), 61 deletions(-) diff --git a/be/src/format_v2/parquet/parquet_scan.cpp b/be/src/format_v2/parquet/parquet_scan.cpp index e736a09f23c165..667cc4de41f9dd 100644 --- a/be/src/format_v2/parquet/parquet_scan.cpp +++ b/be/src/format_v2/parquet/parquet_scan.cpp @@ -1856,6 +1856,7 @@ Status ParquetScanScheduler::read_filter_columns(int64_t batch_rows, if (dictionary_filter_it != _current_dictionary_filters.end()) { const uint16_t selected_rows_before = *selected_rows; IColumn::Filter compact_filter; + uint16_t new_selected_rows = 0; bool used_filter = false; const bool predicate_only = request.is_predicate_only(local_id); // Dictionary ids are sufficient for predicate-only slots; skipping typed survivor @@ -1863,13 +1864,15 @@ Status ParquetScanScheduler::read_filter_columns(int64_t batch_rows, IColumn* projected_column = predicate_only ? nullptr : column.get(); RETURN_IF_ERROR(column_reader->select_with_dictionary_filter( *selection, *selected_rows, batch_rows, dictionary_filter_it->second, - projected_column, &compact_filter, &used_filter)); + projected_column, &compact_filter, &new_selected_rows, &used_filter)); if (used_filter) { DORIS_CHECK(compact_filter.size() == selected_rows_before); + DORIS_CHECK(new_selected_rows <= selected_rows_before); update_counter_if_not_null(_scan_profile.dictionary_predicate_direct_batches, 1); update_counter_if_not_null(_scan_profile.dictionary_predicate_direct_rows, selected_rows_before); - const uint16_t new_selected_rows = count_selected_rows(compact_filter); + // The decoder already observes every keep bit while producing compact_filter, so + // reuse its count instead of adding another full filter scan at this boundary. if (!predicate_only) { update_counter_if_not_null(_scan_profile.dictionary_predicate_projected_rows, new_selected_rows); diff --git a/be/src/format_v2/parquet/reader/column_reader.cpp b/be/src/format_v2/parquet/reader/column_reader.cpp index 151206e4295dce..9eb78718d57e3b 100644 --- a/be/src/format_v2/parquet/reader/column_reader.cpp +++ b/be/src/format_v2/parquet/reader/column_reader.cpp @@ -76,7 +76,7 @@ Status ParquetColumnReader::select(const SelectionVector& selection, uint16_t se Status ParquetColumnReader::select_with_dictionary_filter(const SelectionVector&, uint16_t, int64_t, const IColumn::Filter&, IColumn*, - IColumn::Filter*, bool*) { + IColumn::Filter*, uint16_t*, bool*) { return Status::NotSupported("Parquet dictionary filter is not implemented for column {}", name()); } diff --git a/be/src/format_v2/parquet/reader/column_reader.h b/be/src/format_v2/parquet/reader/column_reader.h index 82848e4ec58efb..9e33e1a17d968f 100644 --- a/be/src/format_v2/parquet/reader/column_reader.h +++ b/be/src/format_v2/parquet/reader/column_reader.h @@ -59,7 +59,8 @@ class ParquetColumnReader { uint16_t selected_rows, int64_t batch_rows, const IColumn::Filter& dictionary_filter, IColumn* projected_column, - IColumn::Filter* row_filter, bool* used_filter); + IColumn::Filter* row_filter, + uint16_t* survivor_count, 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 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 2af1fe237b0764..9625d82557d8c4 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 @@ -1754,7 +1754,7 @@ bool try_filter_and_project_dictionary_values( const IColumn* typed_dictionary, IColumn* projected_values, const std::vector& selected_dictionary_indices, const NullMap& nullable_selection_nulls, const IColumn::Filter& dictionary_filter, - IColumn::Filter* row_filter) { + IColumn::Filter* row_filter, size_t* survivor_count) { if (typed_dictionary == nullptr || projected_values == nullptr) { return false; } @@ -1767,8 +1767,9 @@ bool try_filter_and_project_dictionary_values( const auto& dictionary_data = dictionary->get_data(); auto& projected_data = projected->get_data(); projected_data.reserve(projected_data.size() + selected_dictionary_indices.size()); - row_filter->reserve(nullable_selection_nulls.size()); + row_filter->reserve(row_filter->size() + nullable_selection_nulls.size()); size_t physical_row = 0; + size_t survivors = 0; for (const uint8_t is_null : nullable_selection_nulls) { bool keep = false; if (is_null == 0) { @@ -1778,11 +1779,13 @@ bool try_filter_and_project_dictionary_values( keep = dictionary_filter[dictionary_id] != 0; if (keep) { projected_data.push_back(dictionary_data[dictionary_id]); + ++survivors; } } row_filter->push_back(keep ? 1 : 0); } DORIS_CHECK_EQ(physical_row, selected_dictionary_indices.size()); + *survivor_count = survivors; return true; } @@ -1790,12 +1793,12 @@ bool try_filter_and_project_fixed_width_dictionary( const IColumn* typed_dictionary, IColumn* projected_values, const std::vector& selected_dictionary_indices, const NullMap& nullable_selection_nulls, const IColumn::Filter& dictionary_filter, - IColumn::Filter* row_filter) { -#define TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnType) \ - if (try_filter_and_project_dictionary_values( \ - typed_dictionary, projected_values, selected_dictionary_indices, \ - nullable_selection_nulls, dictionary_filter, row_filter)) { \ - return true; \ + IColumn::Filter* row_filter, size_t* survivor_count) { +#define TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnType) \ + if (try_filter_and_project_dictionary_values( \ + typed_dictionary, projected_values, selected_dictionary_indices, \ + nullable_selection_nulls, dictionary_filter, row_filter, survivor_count)) { \ + return true; \ } TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnUInt8) TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnInt8) @@ -1830,15 +1833,18 @@ template Status ColumnChunkReader::filter_dictionary_indices( const IColumn::Filter& dictionary_filter, ColumnSelectVector& select_vector, const IColumn* typed_dictionary, IColumn* projected_values, - ColumnInt32* matched_dictionary_ids, IColumn::Filter* row_filter, bool* projected_directly, - bool* used_filter) { + ColumnInt32* matched_dictionary_ids, IColumn::Filter* row_filter, size_t* survivor_count, + bool* projected_directly, bool* used_filter) { DORIS_CHECK(row_filter != nullptr); + DORIS_CHECK(survivor_count != nullptr); DORIS_CHECK(projected_directly != nullptr); DORIS_CHECK(used_filter != nullptr); DORIS_CHECK((typed_dictionary == nullptr) == (projected_values == nullptr)); *projected_directly = false; *used_filter = false; - row_filter->clear(); + // The top-level reader clears once and owns the final compact filter. Every fully validated + // page fragment appends here so page/range boundaries never require intermediate copies. + *survivor_count = 0; if (_current_encoding != tparquet::Encoding::RLE_DICTIONARY || _page_decoder == nullptr || !_page_decoder->has_dictionary()) { return Status::OK(); @@ -1911,15 +1917,16 @@ Status ColumnChunkReader::filter_dictionary_indices const bool direct_fixed_width_projection = try_filter_and_project_fixed_width_dictionary( typed_dictionary, projected_values, _selected_dictionary_indices, - _nullable_selection_nulls, dictionary_filter, row_filter); + _nullable_selection_nulls, dictionary_filter, row_filter, survivor_count); auto* matched = matched_dictionary_ids == nullptr ? nullptr : &matched_dictionary_ids->get_data(); if (!direct_fixed_width_projection && matched != nullptr) { matched->reserve(matched->size() + selection.selected_values); } if (!direct_fixed_width_projection) { - row_filter->reserve(_nullable_selection_nulls.size()); + row_filter->reserve(row_filter->size() + _nullable_selection_nulls.size()); size_t physical_row = 0; + size_t survivors = 0; for (const uint8_t is_null : _nullable_selection_nulls) { bool keep = false; if (is_null == 0) { @@ -1930,10 +1937,12 @@ Status ColumnChunkReader::filter_dictionary_indices if (keep && matched != nullptr) { matched->push_back(cast_set(dictionary_id)); } + survivors += keep; } row_filter->push_back(keep ? 1 : 0); } DORIS_CHECK_EQ(physical_row, _selected_dictionary_indices.size()); + *survivor_count = survivors; } // Commit page progress only after every external dictionary id has been validated. _remaining_num_values -= select_vector.num_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 5b6a7ad8be888b..e006f3c59494b3 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 @@ -209,8 +209,8 @@ class ColumnChunkReader { ColumnSelectVector& select_vector, const IColumn* typed_dictionary, IColumn* projected_values, ColumnInt32* matched_dictionary_ids, - IColumn::Filter* row_filter, bool* projected_directly, - bool* used_filter); + IColumn::Filter* row_filter, size_t* survivor_count, + bool* projected_directly, bool* used_filter); // 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 bffad92baf23b7..b9c501df9d9eb1 100644 --- a/be/src/format_v2/parquet/reader/native/column_reader.cpp +++ b/be/src/format_v2/parquet/reader/native/column_reader.cpp @@ -1218,9 +1218,10 @@ template Status ScalarColumnReader::_read_dictionary_filter_values( size_t num_values, const IColumn::Filter& dictionary_filter, FilterMap& filter_map, const IColumn* typed_dictionary, IColumn* projected_values, - ColumnInt32* matched_dictionary_ids, IColumn::Filter* row_filter, + ColumnInt32* matched_dictionary_ids, IColumn::Filter* row_filter, size_t* survivor_count, bool* projected_directly) { DORIS_CHECK(row_filter != nullptr); + DORIS_CHECK(survivor_count != nullptr); DORIS_CHECK(projected_directly != nullptr); _null_run_lengths.clear(); if (_chunk_reader->max_def_level() > 0) { @@ -1263,7 +1264,7 @@ Status ScalarColumnReader::_read_dictionary_filter_ bool used_filter = false; RETURN_IF_ERROR(_chunk_reader->filter_dictionary_indices( dictionary_filter, _select_vector, typed_dictionary, projected_values, - matched_dictionary_ids, row_filter, projected_directly, &used_filter)); + matched_dictionary_ids, row_filter, survivor_count, projected_directly, &used_filter)); // Pure-dictionary chunks are prevalidated before definition levels are consumed. DORIS_CHECK(used_filter); return Status::OK(); @@ -1273,14 +1274,15 @@ template Status ScalarColumnReader::read_dictionary_filter( const IColumn::Filter& dictionary_filter, FilterMap& filter_map, size_t batch_size, const IColumn* typed_dictionary, IColumn* projected_values, - ColumnInt32* matched_dictionary_ids, IColumn::Filter* row_filter, size_t* read_rows, - bool* eof, bool* projected_directly, bool* used_filter) { + ColumnInt32* matched_dictionary_ids, IColumn::Filter* row_filter, size_t* survivor_count, + size_t* read_rows, bool* eof, bool* projected_directly, bool* used_filter) { DORIS_CHECK(row_filter != nullptr); + DORIS_CHECK(survivor_count != nullptr); DORIS_CHECK(read_rows != nullptr); DORIS_CHECK(eof != nullptr); DORIS_CHECK(projected_directly != nullptr); DORIS_CHECK(used_filter != nullptr); - row_filter->clear(); + *survivor_count = 0; *read_rows = 0; *projected_directly = false; *used_filter = false; @@ -1313,16 +1315,17 @@ Status ScalarColumnReader::read_dictionary_filter( _current_row_index += skip_values; const size_t values = std::min(static_cast(range.to() - range.from()), batch_size - has_read); - IColumn::Filter fragment_filter; + size_t fragment_survivors = 0; bool fragment_projected_directly = false; RETURN_IF_ERROR(_read_dictionary_filter_values( values, dictionary_filter, filter_map, typed_dictionary, projected_values, - matched_dictionary_ids, &fragment_filter, &fragment_projected_directly)); + matched_dictionary_ids, row_filter, &fragment_survivors, + &fragment_projected_directly)); if (has_read != 0) { DORIS_CHECK_EQ(*projected_directly, fragment_projected_directly); } *projected_directly = fragment_projected_directly; - row_filter->insert(row_filter->end(), fragment_filter.begin(), fragment_filter.end()); + *survivor_count += fragment_survivors; has_read += values; *read_rows += values; _current_row_index += 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 d96d11e4886344..c6b844e47f74d3 100644 --- a/be/src/format_v2/parquet/reader/native/column_reader.h +++ b/be/src/format_v2/parquet/reader/native/column_reader.h @@ -203,14 +203,16 @@ class ColumnReader { virtual Status read_dictionary_filter(const IColumn::Filter&, FilterMap&, size_t, const IColumn*, IColumn*, ColumnInt32*, - IColumn::Filter* row_filter, size_t* read_rows, bool* eof, - bool* projected_directly, bool* used_filter) { + IColumn::Filter* row_filter, size_t* survivor_count, + size_t* read_rows, bool* eof, bool* projected_directly, + bool* used_filter) { DORIS_CHECK(row_filter != nullptr); + DORIS_CHECK(survivor_count != nullptr); DORIS_CHECK(read_rows != nullptr); DORIS_CHECK(eof != nullptr); DORIS_CHECK(projected_directly != nullptr); DORIS_CHECK(used_filter != nullptr); - row_filter->clear(); + *survivor_count = 0; *read_rows = 0; *projected_directly = false; *used_filter = false; @@ -308,8 +310,9 @@ class ScalarColumnReader : public ColumnReader { Status read_dictionary_filter(const IColumn::Filter& dictionary_filter, FilterMap& filter_map, size_t batch_size, const IColumn* typed_dictionary, IColumn* projected_values, ColumnInt32* matched_dictionary_ids, - IColumn::Filter* row_filter, size_t* read_rows, bool* eof, - bool* projected_directly, bool* used_filter) override; + IColumn::Filter* row_filter, size_t* survivor_count, + size_t* read_rows, bool* eof, bool* projected_directly, + bool* used_filter) override; Status read_column_levels(FilterMap& filter_map, size_t batch_size, size_t* read_rows, bool* eof) override; Result materialize_dictionary_values(const ColumnInt32* dict_column, @@ -444,7 +447,8 @@ class ScalarColumnReader : public ColumnReader { FilterMap& filter_map, const IColumn* typed_dictionary, IColumn* projected_values, ColumnInt32* matched_dictionary_ids, - IColumn::Filter* row_filter, bool* projected_directly); + IColumn::Filter* row_filter, size_t* survivor_count, + bool* projected_directly); 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 ab86e2aea8c371..99c0154eea2e35 100644 --- a/be/src/format_v2/parquet/reader/native_column_reader.cpp +++ b/be/src/format_v2/parquet/reader/native_column_reader.cpp @@ -385,13 +385,15 @@ Status NativeColumnReader::read_with_dictionary_filter( int64_t rows, const uint8_t* filter_data, bool filter_all, const IColumn::Filter& dictionary_filter, const IColumn* typed_dictionary, IColumn* projected_values, ColumnInt32* matched_dictionary_ids, IColumn::Filter* row_filter, - int64_t* rows_read, bool* projected_directly, bool* used_filter) { + int64_t* survivor_count, int64_t* rows_read, bool* projected_directly, bool* used_filter) { DORIS_CHECK(rows >= 0); DORIS_CHECK(row_filter != nullptr); + DORIS_CHECK(survivor_count != nullptr); DORIS_CHECK(rows_read != nullptr); DORIS_CHECK(projected_directly != nullptr); DORIS_CHECK(used_filter != nullptr); row_filter->clear(); + *survivor_count = 0; *rows_read = 0; *projected_directly = false; *used_filter = false; @@ -406,13 +408,13 @@ Status NativeColumnReader::read_with_dictionary_filter( int64_t consecutive_empty_calls = 0; while (*rows_read < rows && !eof) { size_t loop_rows = 0; - IColumn::Filter loop_filter; + size_t loop_survivors = 0; bool loop_projected_directly = false; bool loop_used = false; RETURN_IF_ERROR(_native_reader->read_dictionary_filter( dictionary_filter, filter, static_cast(rows - *rows_read), typed_dictionary, - projected_values, matched_dictionary_ids, &loop_filter, &loop_rows, &eof, - &loop_projected_directly, &loop_used)); + projected_values, matched_dictionary_ids, row_filter, &loop_survivors, &loop_rows, + &eof, &loop_projected_directly, &loop_used)); if (!loop_used) { if (UNLIKELY(*rows_read != 0)) { return Status::Corruption( @@ -426,7 +428,7 @@ Status NativeColumnReader::read_with_dictionary_filter( DORIS_CHECK_EQ(*projected_directly, loop_projected_directly); } *projected_directly = loop_projected_directly; - row_filter->insert(row_filter->end(), loop_filter.begin(), loop_filter.end()); + *survivor_count += cast_set(loop_survivors); if (loop_rows == 0 && !eof) { if (++consecutive_empty_calls > _row_group_rows + 1) { return Status::Corruption( @@ -565,21 +567,22 @@ Status NativeColumnReader::select(const SelectionVector& selection, uint16_t sel return Status::OK(); } -Status NativeColumnReader::select_with_dictionary_filter(const SelectionVector& selection, - uint16_t selected_rows, int64_t batch_rows, - const IColumn::Filter& dictionary_filter, - IColumn* projected_column, - IColumn::Filter* row_filter, - bool* used_filter) { +Status NativeColumnReader::select_with_dictionary_filter( + const SelectionVector& selection, uint16_t selected_rows, int64_t batch_rows, + const IColumn::Filter& dictionary_filter, IColumn* projected_column, + IColumn::Filter* row_filter, uint16_t* survivor_count, bool* used_filter) { DORIS_CHECK(row_filter != nullptr); + DORIS_CHECK(survivor_count != nullptr); DORIS_CHECK(used_filter != nullptr); RETURN_IF_ERROR(validate_selected_span(batch_rows)); *used_filter = false; + *survivor_count = 0; row_filter->clear(); if (!_dictionary_filter_enabled) { return Status::OK(); } *used_filter = true; + row_filter->reserve(selected_rows); const uint8_t* filter_data = nullptr; RETURN_IF_ERROR(selection.materialize_filter(selected_rows, batch_rows, &filter_data)); @@ -601,29 +604,29 @@ Status NativeColumnReader::select_with_dictionary_filter(const SelectionVector& } } int64_t direct_rows_read = 0; + int64_t direct_survivor_count = 0; bool projected_directly = false; bool direct_filter_used = false; RETURN_IF_ERROR(read_with_dictionary_filter( batch_rows, filter_data, selected_rows == 0, dictionary_filter, typed_dictionary, - projected_values, direct_matched_ids, row_filter, &direct_rows_read, - &projected_directly, &direct_filter_used)); + projected_values, direct_matched_ids, row_filter, &direct_survivor_count, + &direct_rows_read, &projected_directly, &direct_filter_used)); if (direct_filter_used) { advance_selected_span(direct_rows_read); - const size_t survivor_count = - cast_set(std::count(row_filter->begin(), row_filter->end(), uint8_t {1})); + *survivor_count = cast_set(direct_survivor_count); if (projected_column != nullptr) { if (projected_directly) { DORIS_CHECK(direct_matched_ids->empty()); if (projected_nullable != nullptr) { auto& null_map = projected_nullable->get_null_map_data(); - null_map.resize_fill(null_map.size() + survivor_count, 0); + null_map.resize_fill(null_map.size() + *survivor_count, 0); } if (_profile.dictionary_predicate_fused_projected_rows != nullptr) { COUNTER_UPDATE(_profile.dictionary_predicate_fused_projected_rows, - survivor_count); + *survivor_count); } } else { - DORIS_CHECK_EQ(direct_matched_ids->size(), survivor_count); + DORIS_CHECK_EQ(direct_matched_ids->size(), *survivor_count); RETURN_IF_ERROR(_native_reader->append_dictionary_values(direct_matched_ids, _type, projected_column)); } @@ -631,8 +634,8 @@ Status NativeColumnReader::select_with_dictionary_filter(const SelectionVector& if (_profile.reader_select_rows != nullptr) { COUNTER_UPDATE(_profile.reader_select_rows, selected_rows); } - update_reader_read_rows(cast_set(survivor_count)); - update_reader_skip_rows(batch_rows - cast_set(survivor_count)); + update_reader_read_rows(*survivor_count); + update_reader_skip_rows(batch_rows - *survivor_count); return Status::OK(); } @@ -677,7 +680,7 @@ Status NativeColumnReader::select_with_dictionary_filter(const SelectionVector& } row_filter->reserve(selected_rows); const auto& id_data = ids->get_data(); - size_t survivor_count = 0; + size_t fallback_survivor_count = 0; for (size_t row = 0; row < selected_rows; ++row) { bool keep = false; if (null_map == nullptr || (*null_map)[row] == 0) { @@ -690,7 +693,7 @@ Status NativeColumnReader::select_with_dictionary_filter(const SelectionVector& } keep = dictionary_filter[static_cast(dictionary_id)] != 0; if (keep) { - ++survivor_count; + ++fallback_survivor_count; if (matched_ids != nullptr) { matched_ids->push_back(dictionary_id); } @@ -708,8 +711,9 @@ Status NativeColumnReader::select_with_dictionary_filter(const SelectionVector& if (_profile.reader_select_rows != nullptr) { COUNTER_UPDATE(_profile.reader_select_rows, selected_rows); } - update_reader_read_rows(cast_set(survivor_count)); - update_reader_skip_rows(batch_rows - cast_set(survivor_count)); + *survivor_count = cast_set(fallback_survivor_count); + update_reader_read_rows(*survivor_count); + update_reader_skip_rows(batch_rows - *survivor_count); return Status::OK(); } 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 97b92e522a78ac..d4f3df59c804d4 100644 --- a/be/src/format_v2/parquet/reader/native_column_reader.h +++ b/be/src/format_v2/parquet/reader/native_column_reader.h @@ -87,7 +87,7 @@ class NativeColumnReader final : public ParquetColumnReader { int64_t batch_rows, const IColumn::Filter& dictionary_filter, IColumn* projected_column, IColumn::Filter* row_filter, - bool* used_filter) override; + uint16_t* survivor_count, 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, @@ -120,8 +120,9 @@ class NativeColumnReader final : public ParquetColumnReader { const IColumn::Filter& dictionary_filter, const IColumn* typed_dictionary, IColumn* projected_values, ColumnInt32* matched_dictionary_ids, - IColumn::Filter* row_filter, int64_t* rows_read, - bool* projected_directly, bool* used_filter); + IColumn::Filter* row_filter, int64_t* survivor_count, + int64_t* rows_read, bool* projected_directly, + bool* used_filter); void release_batch_scratch_if_needed(); int64_t sync_native_profile(); void record_page_fragments(int64_t page_fragments); diff --git a/be/src/format_v2/parquet/selection_vector.h b/be/src/format_v2/parquet/selection_vector.h index ab2bf93c785edc..033478875fad95 100644 --- a/be/src/format_v2/parquet/selection_vector.h +++ b/be/src/format_v2/parquet/selection_vector.h @@ -67,6 +67,7 @@ class SelectionVector { _owned.clear(); _data = data; _size = count; + _identity = data == nullptr; ++_generation; } @@ -77,6 +78,7 @@ class SelectionVector { for (size_t idx = 0; idx < count; ++idx) { _data[idx] = static_cast(idx); } + _identity = true; ++_generation; } @@ -84,6 +86,7 @@ class SelectionVector { _owned.clear(); _data = nullptr; _size = 0; + _identity = true; ++_generation; } @@ -91,7 +94,13 @@ class SelectionVector { bool is_set() const { return _data != nullptr; } - Index* data() { return _data; } + Index* data() { + // A mutable pointer can change indices without set_index(), so identity can no longer be + // proven until resize() rebuilds it. This keeps the O(1) dense fast path conservative. + _identity = false; + ++_generation; + return _data; + } const Index* data() const { return _data; } @@ -104,11 +113,21 @@ class SelectionVector { void set_index(size_t idx, Index value) { _data[idx] = value; + if (value != idx) { + _identity = false; + } ++_generation; } Status materialize_filter(size_t count, int64_t batch_rows, const uint8_t** filter) const { DORIS_CHECK(filter != nullptr); + if (batch_rows >= 0 && std::cmp_equal(count, batch_rows) && _identity && + (_data == nullptr || count <= _size)) { + // A proven identity selection is equivalent to no FilterMap. Returning nullptr avoids + // constructing and rescanning one dense byte per source row. + *filter = nullptr; + return Status::OK(); + } RETURN_IF_ERROR(verify(count, batch_rows)); if (_filter_generation != _generation || _filter_count != count || _filter_batch_rows != batch_rows) { @@ -161,6 +180,7 @@ class SelectionVector { std::vector _owned; Index* _data = nullptr; size_t _size = 0; + bool _identity = true; uint64_t _generation = 0; mutable std::vector _filter; mutable uint64_t _filter_generation = std::numeric_limits::max(); diff --git a/be/test/format_v2/parquet/parquet_reader_control_test.cpp b/be/test/format_v2/parquet/parquet_reader_control_test.cpp index 75e6545906f263..9de951ec5e47dd 100644 --- a/be/test/format_v2/parquet/parquet_reader_control_test.cpp +++ b/be/test/format_v2/parquet/parquet_reader_control_test.cpp @@ -177,6 +177,13 @@ TEST(SelectionVectorTest, MaterializedFilterIsReusedUntilSelectionChanges) { std::vector({0, 1, 1, 0})); } +TEST(SelectionVectorTest, IdentitySelectionDoesNotMaterializeFilter) { + SelectionVector selection(4); + const uint8_t* filter = reinterpret_cast(1); + ASSERT_TRUE(selection.materialize_filter(4, 4, &filter).ok()); + EXPECT_EQ(filter, nullptr); +} + TEST(ParquetColumnReaderControlTest, BaseSelectUsesSkipReadRanges) { CursorColumnReader reader; SelectionVector selection(3);