This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 1284cd5b1f3 [improvement](parquet) Optimize V2 sparse runtime-filter
scans (#66056)
1284cd5b1f3 is described below
commit 1284cd5b1f31c8a29719775bb7159c04726a91ee
Author: Gabriel <[email protected]>
AuthorDate: Tue Jul 28 09:40:38 2026 +0800
[improvement](parquet) Optimize V2 sparse runtime-filter scans (#66056)
TPC-DS Iceberg profiles showed that sparse Parquet scans spent most of
their CPU time repeatedly decoding and skipping tiny selection ranges. In the
observed shape, each range contained only about one selected row. V2 also
missed several opportunities that V1 or the ordinary vectorized path already
used:
Condition-cache hits split adjacent true granules instead of preserving the
original contiguous range.
Dictionary decoders processed fragmented selections range by range and
materialized surviving dictionary values through repeated generic insertion.
RuntimeFilterExpr did not delegate raw fixed-value or dictionary-filter
capabilities to its implementation, so IN, MIN/MAX, and Bloom runtime filters
could not use the new scan fast paths.
Parallel Iceberg position-delete splits could race while creating the same
shared child profile and abort the BE.
What is changed?
Merge adjacent condition-cache intersections while preserving false-granule
gaps.
Batch fragmented dictionary selections and gather survivors directly into
supported fixed-width and string columns.
Add direct typed dictionary probing for runtime filters.
Delegate raw and dictionary evaluation through RuntimeFilterExpr.
Add Bloom runtime-filter support:
typed dictionary probing covers every Bloom-supported fixed scalar type:
BOOLEAN, integer and floating-point types, DATE/DATETIME variants, TIMESTAMPTZ,
DECIMAL variants, IPv4, and IPv6;
CHAR, VARCHAR, and STRING probe dictionary Field values with the same
string hash as the normal column path;
plain-page raw probing is enabled only when the Parquet physical
representation is identical to the Doris value representation: INT, BIGINT,
FLOAT, DOUBLE, DECIMAL32, and DECIMAL64.
Keep null-aware runtime filters on the normal expression path because raw
and dictionary streams omit NULL payloads; this preserves null-aware semantics.
Add atomic runtime-profile child lookup/creation for parallel
position-delete readers.
---
.../core/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 | 61 ++++--
be/src/format_v2/parquet/reader/column_reader.cpp | 2 +-
be/src/format_v2/parquet/reader/column_reader.h | 3 +-
.../parquet/reader/native/column_chunk_reader.cpp | 41 ++--
.../parquet/reader/native/column_chunk_reader.h | 4 +-
.../parquet/reader/native/column_reader.cpp | 19 +-
.../parquet/reader/native/column_reader.h | 16 +-
be/src/format_v2/parquet/reader/native/decoder.h | 61 ++++++
.../parquet/reader/native_column_reader.cpp | 52 ++---
.../parquet/reader/native_column_reader.h | 7 +-
be/src/format_v2/parquet/selection_vector.h | 22 ++-
be/test/exprs/bloom_filter_func_test.cpp | 129 ++++++++++++
be/test/exprs/expr_zonemap_filter_test.cpp | 95 +++++++++
be/test/format_v2/parquet/native_decoder_test.cpp | 169 ++++++++++++++++
.../parquet/parquet_reader_control_test.cpp | 7 +
be/test/format_v2/parquet/parquet_reader_test.cpp | 28 +++
be/test/format_v2/parquet/parquet_scan_test.cpp | 218 +++++++++++++++++++++
25 files changed, 1147 insertions(+), 83 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 ca75bcb8948..68df6074ddf 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 <cstring>
#include <limits>
+#include <type_traits>
+#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 <PrimitiveType TYPE>
-bool try_gather_vector(IColumn& destination, const IColumn& dictionary, const
uint32_t* indices,
- size_t num_values) {
- using ColumnType = ColumnVector<TYPE>;
+template <typename ColumnType>
+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<ValueType>);
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 <PrimitiveType TYPE>
+bool try_gather_vector(IColumn& destination, const IColumn& dictionary, const
uint32_t* indices,
+ size_t num_values) {
+ return try_gather_fixed_width<ColumnVector<TYPE>>(destination, dictionary,
indices, num_values);
+}
+
template <typename Offset>
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<ColumnDecimal32>(destination, dictionary,
indices, num_values) ||
+ try_gather_fixed_width<ColumnDecimal64>(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<UInt32>(destination, dictionary, indices,
num_values) ||
diff --git a/be/src/exprs/bloom_filter_func.h b/be/src/exprs/bloom_filter_func.h
index ae7057a6e90..bd54bd12f52 100644
--- a/be/src/exprs/bloom_filter_func.h
+++ b/be/src/exprs/bloom_filter_func.h
@@ -17,9 +17,12 @@
#pragma once
+#include <cstring>
+
#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 @@ public:
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 @@ public:
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<type>::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<type>::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<fixed_len_to_uint32_v2>(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<type>();
+ return _bloom_filter->test_element<fixed_len_to_uint32_v2>(
+ StringRef(value.data(), value.size()));
+ } else {
+ return
_bloom_filter->test_element<fixed_len_to_uint32_v2>(field.get<type>());
+ }
+ }
+
template <bool is_nullable>
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 18b59338b26..a7c3f6580ca 100644
--- a/be/src/exprs/hybrid_set.h
+++ b/be/src/exprs/hybrid_set.h
@@ -20,6 +20,8 @@
#include <gen_cpp/internal_service.pb.h>
#include <pdqsort.h>
+#include <cstring>
+
#include "common/object_pool.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
@@ -204,6 +206,13 @@ public:
// 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 @@ public:
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 e0f6e7acc1b..cca2f5d5fd2 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<int>& 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 7994d2a71ae..4879b649a2f 100644
--- a/be/src/exprs/runtime_filter_expr.h
+++ b/be/src/exprs/runtime_filter_expr.h
@@ -107,6 +107,13 @@ public:
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<int>& 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 552e3db1c19..6004a74181c 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<VSlotRef>(_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<T>; 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<VSlotRef>(_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<VSlotRef>(_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 410bb5c8d37..1697e524f11 100644
--- a/be/src/exprs/vbloom_predicate.h
+++ b/be/src/exprs/vbloom_predicate.h
@@ -49,6 +49,14 @@ public:
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 2fd1e9a35fe..7a74815a622 100644
--- a/be/src/exprs/vdirect_in_predicate.h
+++ b/be/src/exprs/vdirect_in_predicate.h
@@ -98,6 +98,54 @@ public:
std::dynamic_pointer_cast<VSlotRef>(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<VSlotRef>(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<VSlotRef>(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 @@ public:
}
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<TYPE>::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 a60d3efbeea..667cc4de41f 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"
@@ -796,11 +797,17 @@ int64_t count_range_rows(const std::vector<RowRange>&
ranges) {
}
void append_intersection(const RowRange& left, const RowRange& right,
- std::vector<RowRange>* result) {
+ std::vector<RowRange>& 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 +839,7 @@ std::vector<RowRange>
filter_ranges_by_condition_cache(const std::vector<RowRang
const int64_t granule_end = granule_start +
ConditionCacheContext::GRANULE_SIZE;
const RowRange file_granule_range {.start = granule_start -
row_group_first_row,
.length = granule_end -
granule_start};
- append_intersection(range, file_granule_range, &result);
+ append_intersection(range, file_granule_range, result);
}
}
return result;
@@ -1412,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<ColumnInt32>(dictionary,
values, value_width);
- case TYPE_BIGINT:
- return get_fixed_dictionary_raw_values<ColumnInt64>(dictionary,
values, value_width);
- case TYPE_FLOAT:
- return get_fixed_dictionary_raw_values<ColumnFloat32>(dictionary,
values, value_width);
- case TYPE_DOUBLE:
- return get_fixed_dictionary_raw_values<ColumnFloat64>(dictionary,
values, value_width);
+#define GET_TYPED_DICTIONARY_VALUES(TYPE)
\
+ case TYPE:
\
+ return get_fixed_dictionary_raw_values<typename
PrimitiveTypeTraits<TYPE>::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;
}
@@ -1539,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) {
@@ -1832,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
@@ -1839,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 151206e4295..9eb78718d57 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 82848e4ec58..9e33e1a17d9 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 @@ public:
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 b7bea8e7053..9625d82557d 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<IN_COLLECTION,
OFFSET_INDEX>::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) {
@@ -1752,7 +1754,7 @@ bool try_filter_and_project_dictionary_values(
const IColumn* typed_dictionary, IColumn* projected_values,
const std::vector<uint32_t>& 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;
}
@@ -1765,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) {
@@ -1776,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;
}
@@ -1788,12 +1793,12 @@ bool try_filter_and_project_fixed_width_dictionary(
const IColumn* typed_dictionary, IColumn* projected_values,
const std::vector<uint32_t>& 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<ColumnType>(
\
- 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<ColumnType>(
\
+ 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)
@@ -1828,15 +1833,18 @@ template <bool IN_COLLECTION, bool OFFSET_INDEX>
Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::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();
@@ -1909,15 +1917,16 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::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) {
@@ -1928,10 +1937,12 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::filter_dictionary_indices
if (keep && matched != nullptr) {
matched->push_back(cast_set<int32_t>(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 5b6a7ad8be8..e006f3c5949 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 @@ public:
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 bffad92baf2..b9c501df9d9 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 <bool IN_COLLECTION, bool OFFSET_INDEX>
Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::_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<IN_COLLECTION,
OFFSET_INDEX>::_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 <bool IN_COLLECTION, bool OFFSET_INDEX>
Status ScalarColumnReader<IN_COLLECTION, OFFSET_INDEX>::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<IN_COLLECTION,
OFFSET_INDEX>::read_dictionary_filter(
_current_row_index += skip_values;
const size_t values =
std::min(static_cast<size_t>(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 d96d11e4886..c6b844e47f7 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 @@ public:
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 @@ public:
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<MutableColumnPtr> materialize_dictionary_values(const ColumnInt32*
dict_column,
@@ -444,7 +447,8 @@ private:
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/decoder.h
b/be/src/format_v2/parquet/reader/native/decoder.h
index 379176d5e45..628a34ac976 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 @@ public:
std::vector<uint32_t>* 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 @@ public:
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 @@ public:
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<uint32_t>(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/src/format_v2/parquet/reader/native_column_reader.cpp
b/be/src/format_v2/parquet/reader/native_column_reader.cpp
index ab86e2aea8c..99c0154eea2 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<size_t>(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<int64_t>(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<size_t>(std::count(row_filter->begin(),
row_filter->end(), uint8_t {1}));
+ *survivor_count = cast_set<uint16_t>(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<int64_t>(survivor_count));
- update_reader_skip_rows(batch_rows -
cast_set<int64_t>(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<size_t>(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<int64_t>(survivor_count));
- update_reader_skip_rows(batch_rows - cast_set<int64_t>(survivor_count));
+ *survivor_count = cast_set<uint16_t>(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 97b92e522a7..d4f3df59c80 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 @@ public:
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 @@ private:
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 ab2bf93c785..033478875fa 100644
--- a/be/src/format_v2/parquet/selection_vector.h
+++ b/be/src/format_v2/parquet/selection_vector.h
@@ -67,6 +67,7 @@ public:
_owned.clear();
_data = data;
_size = count;
+ _identity = data == nullptr;
++_generation;
}
@@ -77,6 +78,7 @@ public:
for (size_t idx = 0; idx < count; ++idx) {
_data[idx] = static_cast<Index>(idx);
}
+ _identity = true;
++_generation;
}
@@ -84,6 +86,7 @@ public:
_owned.clear();
_data = nullptr;
_size = 0;
+ _identity = true;
++_generation;
}
@@ -91,7 +94,13 @@ public:
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 @@ public:
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 @@ private:
std::vector<Index> _owned;
Index* _data = nullptr;
size_t _size = 0;
+ bool _identity = true;
uint64_t _generation = 0;
mutable std::vector<uint8_t> _filter;
mutable uint64_t _filter_generation = std::numeric_limits<uint64_t>::max();
diff --git a/be/test/exprs/bloom_filter_func_test.cpp
b/be/test/exprs/bloom_filter_func_test.cpp
index 5e0c02479be..e8e5cdee5bd 100644
--- a/be/test/exprs/bloom_filter_func_test.cpp
+++ b/be/test/exprs/bloom_filter_func_test.cpp
@@ -20,6 +20,7 @@
#include <gtest/gtest-message.h>
#include <gtest/gtest-test-part.h>
+#include <array>
#include <cstdint>
#include <string>
@@ -227,6 +228,134 @@ TEST_F(BloomFilterFuncTest, InsertFixedLen) {
ASSERT_EQ(offsets[1], 2);
}
+TEST_F(BloomFilterFuncTest,
RawFixedCapabilitiesCoverEveryFixedRuntimeFilterType) {
+#define EXPECT_RAW_FIXED(TYPE)
\
+ do {
\
+ BloomFilterFunc<TYPE> filter(false);
\
+ EXPECT_TRUE(filter.supports_raw_fixed_values()) <<
type_to_string(TYPE); \
+ EXPECT_EQ(filter.raw_fixed_value_size(),
\
+ sizeof(typename PrimitiveTypeTraits<TYPE>::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<TYPE_STRING> 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<TYPE_INT> 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<DataTypeInt32>({2, 4});
+ filter.insert_fixed_len(build_column, 0);
+
+ const std::array<int32_t, 4> values {1, 2, 3, 4};
+ std::array<uint8_t, 4> raw_matches {1, 1, 1, 1};
+ ASSERT_TRUE(filter.find_batch_raw_fixed(reinterpret_cast<const
uint8_t*>(values.data()),
+ values.size(), sizeof(int32_t),
raw_matches.data())
+ .ok());
+
+ auto probe_column = ColumnHelper::create_column<DataTypeInt32>({1, 2, 3,
4});
+ std::array<uint8_t, 4> 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 = []<PrimitiveType TYPE>() {
+ BloomFilterFunc<TYPE> 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<TYPE>::CppType;
+ ValueType value {};
+ auto set = std::make_shared<HybridSet<TYPE>>(false);
+ set->insert(&value);
+ filter.insert_set(set);
+
+ uint8_t match = 1;
+ ASSERT_TRUE(filter.find_batch_raw_fixed(reinterpret_cast<const
uint8_t*>(&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<TYPE>(value))) <<
type_to_string(TYPE);
+ };
+
+#define EXPECT_RAW_FIXED_MATCH(TYPE) expect_match.template operator()<TYPE>()
+ 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 = []<PrimitiveType TYPE>() {
+ BloomFilterFunc<TYPE> 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<TYPE_STRING>(std::string("value"))))
+ << type_to_string(TYPE);
+ };
+
+ expect_match.template operator()<TYPE_CHAR>();
+ expect_match.template operator()<TYPE_VARCHAR>();
+ expect_match.template operator()<TYPE_STRING>();
+}
+
TEST_F(BloomFilterFuncTest, Merge) {
BloomFilterFunc<PrimitiveType::TYPE_INT> 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 6e7e42f7baa..6d29cbf4d08 100644
--- a/be/test/exprs/expr_zonemap_filter_test.cpp
+++ b/be/test/exprs/expr_zonemap_filter_test.cpp
@@ -19,6 +19,7 @@
#include <gtest/gtest.h>
+#include <array>
#include <map>
#include <memory>
#include <set>
@@ -26,6 +27,7 @@
#include <vector>
#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<HybridSetBase> 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<VDirectInPredicate>(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<int32_t, 4> values {1, 2, 3, 4};
+ std::array<uint8_t, 4> matches {1, 1, 1, 1};
+ ASSERT_TRUE(runtime_filter
+ ->execute_on_raw_fixed_values(
+ reinterpret_cast<const
uint8_t*>(values.data()), values.size(),
+ sizeof(int32_t), type, 0, matches.data())
+ .ok());
+ EXPECT_EQ((std::array<uint8_t, 4> {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<BloomFilterFuncBase> 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<int32_t, 3> values {missing, 2, 4};
+ std::array<uint8_t, 3> matches {1, 1, 1};
+ ASSERT_TRUE(runtime_filter
+ ->execute_on_raw_fixed_values(
+ reinterpret_cast<const
uint8_t*>(values.data()), values.size(),
+ sizeof(int32_t), type, 0, matches.data())
+ .ok());
+ EXPECT_EQ((std::array<uint8_t, 3> {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 b71e94222a3..0d037f61549 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 @@ public:
std::vector<uint8_t> 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<uint32_t> indices;
+};
+
TEST(ParquetV2NativeDecoderTest, FragmentedPlainSelectionUsesOneConsumerBatch)
{
std::array<int32_t, 32> input {};
std::iota(input.begin(), input.end(), 0);
@@ -288,6 +306,90 @@ TEST(ParquetV2NativeDecoderTest,
FragmentedPlainSelectionUsesOneConsumerBatch) {
EXPECT_EQ(consumer.values<int32_t>(), expected);
}
+TEST(ParquetV2NativeDecoderTest,
FragmentedDictionarySelectionUsesOneConsumerBatch) {
+ constexpr size_t VALUE_COUNT = 32;
+ std::array<int32_t, VALUE_COUNT> dictionary_values {};
+ std::iota(dictionary_values.begin(), dictionary_values.end(), 0);
+ auto dictionary = make_unique_buffer<uint8_t>(sizeof(dictionary_values));
+ memcpy(dictionary.get(), dictionary_values.data(),
sizeof(dictionary_values));
+
+ std::unique_ptr<Decoder> 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<uint32_t> encoder(&encoded_ids, 5);
+ for (uint32_t id = 0; id < VALUE_COUNT; ++id) {
+ encoder.Put(id);
+ }
+ encoder.Flush();
+ std::vector<uint8_t> 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<uint32_t> 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<uint32_t> 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<int32_t, DICTIONARY_SIZE> dictionary_values {};
+ std::iota(dictionary_values.begin(), dictionary_values.end(), 0);
+ auto dictionary = make_unique_buffer<uint8_t>(sizeof(dictionary_values));
+ memcpy(dictionary.get(), dictionary_values.data(),
sizeof(dictionary_values));
+
+ std::unique_ptr<Decoder> 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<uint32_t> encoder(&encoded_ids, 5);
+ for (uint32_t row = 0; row < VALUE_COUNT; ++row) {
+ encoder.Put(row % DICTIONARY_SIZE);
+ }
+ encoder.Flush();
+ std::vector<uint8_t> 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<uint32_t>(selection.selected_values, 0));
+}
+
class ScriptedDictionaryMaterializationSource final : public
ParquetDecodeSource {
public:
ScriptedDictionaryMaterializationSource(std::vector<uint32_t> 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<uint32_t, 8> indices {3, 1, 2, 0, 0, 2, 1, 3};
+ const auto verify = [&]<typename ColumnType>() {
+ 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<decltype(ValueType::value)>(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()<ColumnDecimal32>();
+ verify.template operator()<ColumnDecimal64>();
+}
+
+template <PrimitiveType Type>
+void expect_fixed_width_dictionary_gather() {
+ using ColumnType = ColumnVector<Type>;
+ 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<uint32_t, 8> 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<TYPE_INT>();
+ expect_fixed_width_dictionary_gather<TYPE_BIGINT>();
+ expect_fixed_width_dictionary_gather<TYPE_FLOAT>();
+ expect_fixed_width_dictionary_gather<TYPE_DOUBLE>();
+ expect_fixed_width_dictionary_gather<TYPE_DATE>();
+ expect_fixed_width_dictionary_gather<TYPE_DATETIME>();
+ expect_fixed_width_dictionary_gather<TYPE_DATEV2>();
+ expect_fixed_width_dictionary_gather<TYPE_DATETIMEV2>();
+ expect_fixed_width_dictionary_gather<TYPE_TIMESTAMPTZ>();
+ expect_fixed_width_dictionary_gather<TYPE_IPV4>();
+ expect_fixed_width_dictionary_gather<TYPE_TIMEV2>();
+ expect_fixed_width_dictionary_gather<TYPE_UINT32>();
+ expect_fixed_width_dictionary_gather<TYPE_UINT64>();
+}
+
TEST(ParquetV2NativeDecoderTest,
ComplexPageStatisticsPreservePerLeafCrossings) {
ColumnChunkReaderStatistics first_chunk;
first_chunk.page_read_counter = 1;
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 75e6545906f..9de951ec5e4 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<uint8_t>({0, 1, 1, 0}));
}
+TEST(SelectionVectorTest, IdentitySelectionDoesNotMaterializeFilter) {
+ SelectionVector selection(4);
+ const uint8_t* filter = reinterpret_cast<const uint8_t*>(1);
+ ASSERT_TRUE(selection.materialize_filter(4, 4, &filter).ok());
+ EXPECT_EQ(filter, nullptr);
+}
+
TEST(ParquetColumnReaderControlTest, BaseSelectUsesSkipReadRanges) {
CursorColumnReader reader;
SelectionVector selection(3);
diff --git a/be/test/format_v2/parquet/parquet_reader_test.cpp
b/be/test/format_v2/parquet/parquet_reader_test.cpp
index adb3b4b383c..af0897d6ef2 100644
--- a/be/test/format_v2/parquet/parquet_reader_test.cpp
+++ b/be/test/format_v2/parquet/parquet_reader_test.cpp
@@ -2010,6 +2010,34 @@ TEST_F(NewParquetReaderTest,
ConditionCacheHitSkipsFalseGranulesBeforeColumnRead
EXPECT_EQ(rows, 0);
}
+TEST_F(NewParquetReaderTest,
ConditionCacheHitKeepsBatchAcrossAdjacentTrueGranules) {
+ write_condition_cache_parquet_file(_file_path);
+ auto reader = create_reader();
+ reader->set_batch_size(ConditionCacheContext::GRANULE_SIZE * 2);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ ASSERT_EQ(schema.size(), 1);
+
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->non_predicate_columns = {field_projection(0)};
+ ASSERT_TRUE(reader->open(request).ok());
+
+ auto ctx = std::make_shared<ConditionCacheContext>();
+ ctx->is_hit = true;
+ ctx->filter_result = std::make_shared<std::vector<bool>>(std::vector<bool>
{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);
diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp
b/be/test/format_v2/parquet/parquet_scan_test.cpp
index 11cfdf99641..f3049a3d03a 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<Int32DirectGreaterExpr>(column_id, lower_bound));
}
+TExprNode make_runtime_in_node() {
+ TExprNode node;
+ node.__set_type(std::make_shared<DataTypeUInt8>()->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<int32_t>& values,
+ int filter_id) {
+ std::shared_ptr<HybridSetBase> 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<DataTypeInt32>()),
+ "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<int32_t>& values,
+ int filter_id) {
+ std::shared_ptr<BloomFilterFuncBase> 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<DataTypeInt32>()),
+ "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<std::string>& values,
+ int filter_id) {
+ std::shared_ptr<BloomFilterFuncBase>
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<DataTypeString>()),
+ "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<DataTypeUInt8>()->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<Int64DirectGreaterExpr>(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<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok());
+
ASSERT_TRUE(request_builder.add_non_predicate_column(format::LocalColumnId(1)).ok());
+ request->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<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok());
+
ASSERT_TRUE(request_builder.add_non_predicate_column(format::LocalColumnId(1)).ok());
+ request->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<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok());
+
ASSERT_TRUE(request_builder.add_non_predicate_column(format::LocalColumnId(1)).ok());
+ request->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<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ format::FileScanRequestBuilder request_builder(request.get());
+
ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok());
+
ASSERT_TRUE(request_builder.add_non_predicate_column(format::LocalColumnId(1)).ok());
+ request->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");
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
