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 97f7f957a81 [improvement](parquet) Optimize typed dictionary range
filtering (#66036)
97f7f957a81 is described below
commit 97f7f957a81142781cc916f8484275bef3c172e1
Author: Gabriel <[email protected]>
AuthorDate: Sun Jul 26 10:58:58 2026 +0800
[improvement](parquet) Optimize typed dictionary range filtering (#66036)
### What problem does this PR solve?
File Scanner V2 could only use row-level Parquet dictionary filtering
efficiently for a narrow set of predicates and projected INT columns.
Range predicates and other typed dictionaries could still pay per-entry
generic expression evaluation, generic SerDe insertion, or row-sized
predicate-column materialization.
### What is changed?
- Enable exact typed dictionary evaluation for `=`, `!=`, `<`, `<=`,
`>`, and `>=`, including symmetric literal-on-left comparisons.
- Accept both `RLE_DICTIONARY` and legacy `PLAIN_DICTIONARY` data-page
encodings for supported primitive columns.
- Build numeric dictionary bitmaps over contiguous typed INT, BIGINT,
FLOAT, and DOUBLE dictionary values, evaluating each conjunct once per
dictionary generation.
- Build string equality and range bitmaps by comparing dictionary slices
directly, without per-entry `Field` construction or generic dictionary
expression dispatch.
- Decode selected dictionary IDs at the native page-reader layer and
apply the per-entry bitmap without materializing a complete predicate
value column.
- Write all supported fixed-width survivors directly from the filter
loop into the target column.
- Gather string survivors with pre-sized character/offset buffers and
one copy per selected dictionary value.
- Add direct-path profile counters and focused INT32/BIGINT/BYTE_ARRAY
dictionary microbench scenarios.
### Verification
- ASAN focused tests: 14/14 passed, covering typed numeric/string range
filters, literal-on-left normalization, fixed-width fused gather,
compact string gather, and benchmark scenario registration.
- Related ASAN suite: 487 applicable tests passed across Parquet, File
Scanner V2, SerDe, column mapping, JSON, WAL, and remote reader
coverage. Three unrelated Flight tests could not bind their fixed
localhost endpoint because it was already occupied by a pre-existing
service.
- `git diff --check` passed.
### Microbenchmark: upstream master vs this PR
Compared upstream master `7809a73814e` directly with this PR at
`943c0b94ffc`. Both binaries use the same Release compiler options,
benchmark harness, fixtures, and fixed CPU. Each binary received three
warmups; measurements used interleaved master/PR ordering and 20 samples
per scenario. Values below are median CPU time. `raw_rows`,
`selected_rows`, and fixture sizes were identical for every pair.
| Dictionary type | Selectivity | Projection | Master | This PR | Change
|
|---|---:|---|---:|---:|---:|
| INT32 | 10% | predicate only | 1,332,627 ns | 329,241 ns | -75.3% |
| INT32 | 50% | predicate only | 1,344,746 ns | 338,433 ns | -74.8% |
| INT32 | 10% | predicate projected | 1,686,162 ns | 1,000,973 ns |
-40.6% |
| INT32 | 50% | predicate projected | 1,713,684 ns | 1,041,358 ns |
-39.2% |
| BIGINT | 10% | predicate only | 1,053,504 ns | 329,214 ns | -68.8% |
| BIGINT | 50% | predicate only | 1,053,577 ns | 340,627 ns | -67.7% |
| BIGINT | 10% | predicate projected | 1,367,541 ns | 996,056 ns |
-27.2% |
| BIGINT | 50% | predicate projected | 1,438,891 ns | 1,053,813 ns |
-26.8% |
| BYTE_ARRAY | 10% | predicate only | 2,007,277 ns | 386,762 ns | -80.7%
|
| BYTE_ARRAY | 50% | predicate only | 2,017,380 ns | 398,810 ns | -80.2%
|
| BYTE_ARRAY | 10% | predicate projected | 2,437,263 ns | 1,194,903 ns |
-51.0% |
| BYTE_ARRAY | 50% | predicate projected | 2,541,460 ns | 1,323,767 ns |
-47.9% |
All twelve direct master-to-PR scenarios improved. Predicate-only scans
avoid row-sized materialization and generic per-entry dispatch;
projected scans additionally avoid generic survivor insertion.
---
be/benchmark/parquet/benchmark_parquet_reader.hpp | 118 ++++++++-
be/benchmark/parquet/parquet_benchmark_scenarios.h | 49 +++-
.../core/data_type_serde/parquet_decode_source.cpp | 47 ++++
be/src/exprs/function/functions_comparison.h | 41 ++-
be/src/format_v2/parquet/parquet_profile.cpp | 18 ++
be/src/format_v2/parquet/parquet_profile.h | 18 +-
be/src/format_v2/parquet/parquet_scan.cpp | 223 ++++++++++++++--
be/src/format_v2/parquet/reader/column_reader.cpp | 2 +-
be/src/format_v2/parquet/reader/column_reader.h | 2 +-
.../parquet/reader/native/column_chunk_reader.cpp | 195 ++++++++++++++
.../parquet/reader/native/column_chunk_reader.h | 17 +-
.../parquet/reader/native/column_reader.cpp | 225 +++++++++++++++-
.../parquet/reader/native/column_reader.h | 38 +++
be/src/format_v2/parquet/reader/native/decoder.cpp | 3 +
.../parquet/reader/native_column_reader.cpp | 169 ++++++++++--
.../parquet/reader/native_column_reader.h | 8 +-
be/test/exprs/expr_zonemap_filter_test.cpp | 33 ++-
be/test/format_v2/parquet/native_decoder_test.cpp | 99 ++++---
.../parquet/parquet_benchmark_scenarios_test.cpp | 22 +-
be/test/format_v2/parquet/parquet_scan_test.cpp | 289 ++++++++++++++++++++-
docs/file-scanner-v2-parquet-scan-design.md | 25 +-
21 files changed, 1512 insertions(+), 129 deletions(-)
diff --git a/be/benchmark/parquet/benchmark_parquet_reader.hpp
b/be/benchmark/parquet/benchmark_parquet_reader.hpp
index 63094738040..b456244ba0e 100644
--- a/be/benchmark/parquet/benchmark_parquet_reader.hpp
+++ b/be/benchmark/parquet/benchmark_parquet_reader.hpp
@@ -94,6 +94,64 @@ inline std::shared_ptr<arrow::Array> build_int32_array(int
null_percent, Pattern
return builder.Finish().ValueOrDie();
}
+inline std::shared_ptr<arrow::Array> build_int64_array(int null_percent,
Pattern pattern) {
+ arrow::Int64Builder builder;
+ PARQUET_THROW_NOT_OK(builder.Reserve(READER_ROWS));
+ for (size_t row = 0; row < READER_ROWS; ++row) {
+ if (is_null_row(row, null_percent, pattern)) {
+ PARQUET_THROW_NOT_OK(builder.AppendNull());
+ } else {
+ PARQUET_THROW_NOT_OK(builder.Append(static_cast<int64_t>(row %
100)));
+ }
+ }
+ return builder.Finish().ValueOrDie();
+}
+
+inline std::string padded_decimal(size_t value) {
+ std::string result = std::to_string(value);
+ result.insert(0, 3 - result.size(), '0');
+ return result;
+}
+
+inline std::shared_ptr<arrow::Array> build_string_array(int null_percent,
Pattern pattern) {
+ arrow::StringBuilder builder;
+ PARQUET_THROW_NOT_OK(builder.Reserve(READER_ROWS));
+ for (size_t row = 0; row < READER_ROWS; ++row) {
+ if (is_null_row(row, null_percent, pattern)) {
+ PARQUET_THROW_NOT_OK(builder.AppendNull());
+ } else {
+ PARQUET_THROW_NOT_OK(builder.Append(padded_decimal(row % 100)));
+ }
+ }
+ return builder.Finish().ValueOrDie();
+}
+
+inline std::shared_ptr<arrow::Array> build_value_array(const ReaderScenario&
scenario) {
+ switch (scenario.value_type) {
+ case ValueType::INT32:
+ return build_int32_array(scenario.null_percent, scenario.null_pattern);
+ case ValueType::INT64:
+ return build_int64_array(scenario.null_percent, scenario.null_pattern);
+ case ValueType::BYTE_ARRAY:
+ return build_string_array(scenario.null_percent,
scenario.null_pattern);
+ default:
+ throw std::logic_error("unsupported Parquet reader benchmark value
type");
+ }
+}
+
+inline std::shared_ptr<arrow::DataType> arrow_value_type(ValueType value_type)
{
+ switch (value_type) {
+ case ValueType::INT32:
+ return arrow::int32();
+ case ValueType::INT64:
+ return arrow::int64();
+ case ValueType::BYTE_ARRAY:
+ return arrow::utf8();
+ default:
+ throw std::logic_error("unsupported Parquet reader benchmark value
type");
+ }
+}
+
inline ::parquet::Encoding::type file_encoding(Encoding encoding) {
switch (encoding) {
case Encoding::PLAIN:
@@ -113,9 +171,10 @@ inline ::parquet::Encoding::type file_encoding(Encoding
encoding) {
}
inline std::string fixture_name(const ReaderScenario& scenario) {
- return "v2_" + to_string(scenario.encoding) + "_null" +
std::to_string(scenario.null_percent) +
- "_" + to_string(scenario.null_pattern) + "_w" +
std::to_string(scenario.schema_width) +
- "_p" + std::to_string(scenario.predicate_position) + ".parquet";
+ return "v2_" + to_string(scenario.encoding) + "_" +
to_string(scenario.value_type) + "_null" +
+ std::to_string(scenario.null_percent) + "_" +
to_string(scenario.null_pattern) + "_w" +
+ std::to_string(scenario.schema_width) + "_p" +
+ std::to_string(scenario.predicate_position) + ".parquet";
}
inline void verify_fixture_encoding(const std::filesystem::path& path,
@@ -154,13 +213,14 @@ inline std::filesystem::path ensure_fixture(const
ReaderScenario& scenario) {
std::filesystem::create_directories(directory);
const auto temporary_path = path.string() + ".tmp";
std::filesystem::remove(temporary_path);
- const auto values = build_int32_array(scenario.null_percent,
scenario.null_pattern);
+ const auto values = build_value_array(scenario);
std::vector<std::shared_ptr<arrow::Field>> fields;
std::vector<std::shared_ptr<arrow::ChunkedArray>> columns;
fields.reserve(scenario.schema_width);
columns.reserve(scenario.schema_width);
for (int column = 0; column < scenario.schema_width; ++column) {
- fields.push_back(arrow::field("c" + std::to_string(column),
arrow::int32(), true));
+ fields.push_back(arrow::field("c" + std::to_string(column),
+ arrow_value_type(scenario.value_type),
true));
columns.push_back(std::make_shared<arrow::ChunkedArray>(values));
}
const auto table = arrow::Table::Make(arrow::schema(std::move(fields)),
std::move(columns));
@@ -305,6 +365,24 @@ inline VExprSPtr make_int32_comparison(const std::string&
function_name, TExprOp
return comparison;
}
+inline VExprSPtr make_reader_literal(const ReaderScenario& scenario, const
DataTypePtr& type) {
+ switch (scenario.value_type) {
+ case ValueType::INT32:
+ return VLiteral::create_shared(remove_nullable(type),
+
Field::create_field<TYPE_INT>(scenario.selectivity_percent));
+ case ValueType::INT64:
+ return VLiteral::create_shared(
+ remove_nullable(type),
+
Field::create_field<TYPE_BIGINT>(scenario.selectivity_percent));
+ case ValueType::BYTE_ARRAY:
+ return VLiteral::create_shared(
+ remove_nullable(type),
+
Field::create_field<TYPE_STRING>(padded_decimal(scenario.selectivity_percent)));
+ default:
+ throw std::logic_error("unsupported Parquet reader benchmark predicate
type");
+ }
+}
+
inline VExprContextSPtr make_complex_residual_predicate(int
selectivity_percent, int first_position,
int
later_left_position,
int
later_right_position,
@@ -384,8 +462,16 @@ inline std::unique_ptr<ReaderSession> open_reader(const
std::filesystem::path& p
request_builder.add_non_predicate_column(format::LocalColumnId(payload)));
}
const auto predicate_position =
session->request->local_positions.at(predicate_id).value();
- session->request->conjuncts.push_back(
- make_predicate(static_cast<int>(predicate_position),
scenario.selectivity_percent));
+ auto context = VExprContext::create_shared(make_int32_comparison(
+ "lt", TExprOpcode::LT,
+ VSlotRef::create_shared(static_cast<int>(predicate_position),
+ static_cast<int>(predicate_position),
-1,
+
session->schema[scenario.predicate_position].type, "c0"),
+ make_reader_literal(scenario,
session->schema[scenario.predicate_position].type)));
+ throw_if_error(context->prepare(&session->runtime_state,
RowDescriptor()));
+ throw_if_error(context->open(&session->runtime_state));
+ session->request->conjuncts.push_back(context);
+ session->opened_conjuncts.push_back(std::move(context));
} else if (scenario.operation == ReaderOperation::COMPLEX_RESIDUAL_SCAN) {
DORIS_CHECK(scenario.schema_width >= 5);
std::array<int, 3> predicate_columns {0, 2, 3};
@@ -472,6 +558,19 @@ inline int projected_columns(const ReaderScenario&
scenario) {
return std::min(2, scenario.schema_width);
}
+inline size_t value_width(const ReaderScenario& scenario) {
+ switch (scenario.value_type) {
+ case ValueType::INT32:
+ return sizeof(int32_t);
+ case ValueType::INT64:
+ return sizeof(int64_t);
+ case ValueType::BYTE_ARRAY:
+ return 3;
+ default:
+ return sizeof(int32_t);
+ }
+}
+
inline void run_reader(benchmark::State& state, ReaderScenario scenario) {
std::filesystem::path fixture;
try {
@@ -494,8 +593,9 @@ inline void run_reader(benchmark::State& state,
ReaderScenario scenario) {
const auto raw_rows = raw_rows_per_iteration(scenario);
state.SetItemsProcessed(static_cast<int64_t>(state.iterations() *
selected_rows));
- state.SetBytesProcessed(static_cast<int64_t>(
- state.iterations() * raw_rows * projected_columns(scenario) *
sizeof(int32_t)));
+ state.SetBytesProcessed(
+ static_cast<int64_t>(state.iterations() * raw_rows *
projected_columns(scenario) *
+ value_width(scenario)));
state.counters["raw_rows"] = static_cast<double>(raw_rows);
state.counters["selected_rows"] = static_cast<double>(selected_rows);
state.counters["fixture_bytes"] =
static_cast<double>(std::filesystem::file_size(fixture));
diff --git a/be/benchmark/parquet/parquet_benchmark_scenarios.h
b/be/benchmark/parquet/parquet_benchmark_scenarios.h
index e6ef9f367be..900cb29a583 100644
--- a/be/benchmark/parquet/parquet_benchmark_scenarios.h
+++ b/be/benchmark/parquet/parquet_benchmark_scenarios.h
@@ -67,6 +67,7 @@ struct ReaderScenario {
Projection projection;
int schema_width;
int predicate_position;
+ ValueType value_type = ValueType::INT32;
};
struct KernelScenario {
@@ -145,12 +146,14 @@ inline std::vector<KernelScenario> kernel_scenarios() {
inline std::vector<ReaderScenario> reader_scenarios() {
std::vector<ReaderScenario> scenarios;
- std::set<std::tuple<ReaderOperation, Encoding, int, Pattern, int,
Projection, int, int>> seen;
+ std::set<std::tuple<ReaderOperation, Encoding, int, Pattern, int,
Projection, int, int,
+ ValueType>>
+ seen;
const auto add = [&](ReaderScenario scenario) {
- const auto key = std::make_tuple(scenario.operation, scenario.encoding,
- scenario.null_percent,
scenario.null_pattern,
- scenario.selectivity_percent,
scenario.projection,
- scenario.schema_width,
scenario.predicate_position);
+ const auto key = std::make_tuple(
+ scenario.operation, scenario.encoding, scenario.null_percent,
scenario.null_pattern,
+ scenario.selectivity_percent, scenario.projection,
scenario.schema_width,
+ scenario.predicate_position, scenario.value_type);
if (seen.insert(key).second) {
scenarios.push_back(scenario);
}
@@ -193,6 +196,31 @@ inline std::vector<ReaderScenario> reader_scenarios() {
}
}
}
+ for (const int selectivity : {1, 10, 50, 90}) {
+ for (const auto projection :
+ {Projection::PREDICATE_ONLY, Projection::PREDICATE_PROJECTED}) {
+ auto scenario = baseline;
+ scenario.operation = ReaderOperation::PREDICATE_SCAN;
+ scenario.encoding = Encoding::DICTIONARY;
+ scenario.selectivity_percent = selectivity;
+ scenario.projection = projection;
+ add(scenario);
+ }
+ }
+ for (const auto value_type : {ValueType::INT64, ValueType::BYTE_ARRAY}) {
+ for (const int selectivity : {10, 50}) {
+ for (const auto projection :
+ {Projection::PREDICATE_ONLY,
Projection::PREDICATE_PROJECTED}) {
+ auto scenario = baseline;
+ scenario.operation = ReaderOperation::PREDICATE_SCAN;
+ scenario.encoding = Encoding::DICTIONARY;
+ scenario.selectivity_percent = selectivity;
+ scenario.projection = projection;
+ scenario.value_type = value_type;
+ add(scenario);
+ }
+ }
+ }
for (const int width : {4, 32, 128, 512}) {
for (const int predicate_position : {0, width - 1}) {
auto scenario = baseline;
@@ -326,11 +354,12 @@ inline std::string to_string(ReaderOperation value) {
}
inline std::string reader_scenario_name(const ReaderScenario& scenario) {
- return to_string(scenario.operation) + "/" + to_string(scenario.encoding)
+ "/null_" +
- std::to_string(scenario.null_percent) + "/" +
to_string(scenario.null_pattern) +
- "/sel_" + std::to_string(scenario.selectivity_percent) + "/" +
- to_string(scenario.projection) + "/width_" +
std::to_string(scenario.schema_width) +
- "/predicate_" + std::to_string(scenario.predicate_position);
+ return to_string(scenario.operation) + "/" + to_string(scenario.encoding)
+ "/" +
+ to_string(scenario.value_type) + "/null_" +
std::to_string(scenario.null_percent) + "/" +
+ to_string(scenario.null_pattern) + "/sel_" +
+ std::to_string(scenario.selectivity_percent) + "/" +
to_string(scenario.projection) +
+ "/width_" + std::to_string(scenario.schema_width) + "/predicate_" +
+ std::to_string(scenario.predicate_position);
}
inline std::string to_string(Kernel value) {
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 8a60288811c..ca75bcb8948 100644
--- a/be/src/core/data_type_serde/parquet_decode_source.cpp
+++ b/be/src/core/data_type_serde/parquet_decode_source.cpp
@@ -17,6 +17,10 @@
#include "core/data_type_serde/parquet_decode_source.h"
+#include <cstring>
+#include <limits>
+
+#include "core/column/column_string.h"
#include "core/column/column_vector.h"
#include "util/simd/parquet_kernels.h"
@@ -53,6 +57,43 @@ bool try_gather_vector(IColumn& destination, const IColumn&
dictionary, const ui
}
}
+template <typename Offset>
+bool try_gather_strings(IColumn& destination, const IColumn& dictionary, const
uint32_t* indices,
+ size_t num_values) {
+ auto* destination_string = dynamic_cast<ColumnStr<Offset>*>(&destination);
+ const auto* dictionary_string = dynamic_cast<const
ColumnStr<Offset>*>(&dictionary);
+ if (destination_string == nullptr || dictionary_string == nullptr) {
+ return false;
+ }
+
+ size_t bytes = 0;
+ for (size_t row = 0; row < num_values; ++row) {
+ const size_t value_size =
dictionary_string->get_data_at(indices[row]).size;
+ if (value_size > std::numeric_limits<size_t>::max() - bytes) {
+ return false;
+ }
+ bytes += value_size;
+ }
+ auto& chars = destination_string->get_chars();
+ auto& offsets = destination_string->get_offsets();
+ if (bytes > std::numeric_limits<Offset>::max() - chars.size()) {
+ return false;
+ }
+ const size_t old_chars_size = chars.size();
+ chars.resize(old_chars_size + bytes);
+ offsets.reserve(offsets.size() + num_values);
+ size_t output_offset = old_chars_size;
+ for (size_t row = 0; row < num_values; ++row) {
+ const StringRef value = dictionary_string->get_data_at(indices[row]);
+ if (value.size != 0) {
+ memcpy(chars.data() + output_offset, value.data, value.size);
+ }
+ output_offset += value.size;
+ offsets.push_back(static_cast<Offset>(output_offset));
+ }
+ return true;
+}
+
} // namespace
bool try_simd_insert_parquet_dictionary_indices(IColumn& destination, const
IColumn& dictionary,
@@ -72,6 +113,12 @@ bool try_simd_insert_parquet_dictionary_indices(IColumn&
destination, const ICol
TRY_PARQUET_GATHER(TYPE_UINT32);
TRY_PARQUET_GATHER(TYPE_UINT64);
#undef TRY_PARQUET_GATHER
+ // 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) ||
+ try_gather_strings<UInt64>(destination, dictionary, indices,
num_values)) {
+ return true;
+ }
return false;
}
diff --git a/be/src/exprs/function/functions_comparison.h
b/be/src/exprs/function/functions_comparison.h
index 9434f78aece..a0789c259c7 100644
--- a/be/src/exprs/function/functions_comparison.h
+++ b/be/src/exprs/function/functions_comparison.h
@@ -370,12 +370,47 @@ inline bool can_evaluate_equality(const VExprSPtrs&
arguments, Op op) {
return op == Op::EQ && can_evaluate(arguments);
}
+inline bool dictionary_value_matches(const Field& value, const Field& literal,
Op op) {
+ switch (op) {
+ case Op::EQ:
+ return value == literal;
+ case Op::NE:
+ return value != literal;
+ case Op::LT:
+ return value < literal;
+ case Op::LE:
+ return value <= literal;
+ case Op::GT:
+ return value > literal;
+ case Op::GE:
+ return value >= literal;
+ }
+ __builtin_unreachable();
+}
+
inline ZoneMapFilterResult evaluate_dictionary(const DictionaryEvalContext&
ctx,
const VExprSPtrs& arguments, Op
op) {
- DORIS_CHECK(op == Op::EQ);
auto slot_literal = expr_zonemap::extract_slot_and_literal(arguments);
DORIS_CHECK(slot_literal.has_value());
- return expr_zonemap::eval_eq_dictionary(ctx, *slot_literal);
+ const auto* dictionary = ctx.slot(slot_literal->slot_index);
+ if (dictionary == nullptr || dictionary->data_type == nullptr) {
+ return ZoneMapFilterResult::kUnsupported;
+ }
+ DORIS_CHECK(
+ expr_zonemap::data_types_compatible(dictionary->data_type,
slot_literal->slot_type));
+ if (slot_literal->literal.is_null()) {
+ return ZoneMapFilterResult::kUnsupported;
+ }
+ const auto effective_op = slot_literal->literal_on_left ? symmetric_op(op)
: op;
+ // Compare typed Fields so dictionary filtering preserves the regular
expression semantics for
+ // strings, dates, decimals, floating-point edge cases, and every other
supported logical type.
+ return std::ranges::any_of(dictionary->values,
+ [&](const Field& value) {
+ return dictionary_value_matches(value,
slot_literal->literal,
+
effective_op);
+ })
+ ? ZoneMapFilterResult::kMayMatch
+ : ZoneMapFilterResult::kNoMatch;
}
inline ZoneMapFilterResult evaluate_bloom_filter(const BloomFilterEvalContext&
ctx,
@@ -611,7 +646,7 @@ public:
bool can_evaluate_dictionary_filter(const VExprSPtrs& arguments) const
override {
auto op = comparison_zonemap_detail::op_from_name(name);
- return op.has_value() &&
comparison_zonemap_detail::can_evaluate_equality(arguments, *op);
+ return op.has_value() &&
comparison_zonemap_detail::can_evaluate(arguments);
}
ZoneMapFilterResult evaluate_bloom_filter(const BloomFilterEvalContext&
ctx,
diff --git a/be/src/format_v2/parquet/parquet_profile.cpp
b/be/src/format_v2/parquet/parquet_profile.cpp
index 9332dac0182..6bf354f63bb 100644
--- a/be/src/format_v2/parquet/parquet_profile.cpp
+++ b/be/src/format_v2/parquet/parquet_profile.cpp
@@ -181,6 +181,14 @@ void ParquetProfile::init(RuntimeProfile* profile) {
profile, "FixedWidthPredicateDirectBatches", TUnit::UNIT,
parquet_profile, 1);
fixed_width_predicate_direct_rows = ADD_CHILD_COUNTER_WITH_LEVEL(
profile, "FixedWidthPredicateDirectRows", TUnit::UNIT,
parquet_profile, 1);
+ dictionary_predicate_direct_batches = ADD_CHILD_COUNTER_WITH_LEVEL(
+ profile, "DictionaryPredicateDirectBatches", TUnit::UNIT,
parquet_profile, 1);
+ dictionary_predicate_direct_rows = ADD_CHILD_COUNTER_WITH_LEVEL(
+ profile, "DictionaryPredicateDirectRows", TUnit::UNIT,
parquet_profile, 1);
+ dictionary_predicate_projected_rows = ADD_CHILD_COUNTER_WITH_LEVEL(
+ profile, "DictionaryPredicateProjectedRows", TUnit::UNIT,
parquet_profile, 1);
+ dictionary_predicate_fused_projected_rows = ADD_CHILD_COUNTER_WITH_LEVEL(
+ profile, "DictionaryPredicateFusedProjectedRows", TUnit::UNIT,
parquet_profile, 1);
dict_filter_rewrite_time =
ADD_CHILD_TIMER_WITH_LEVEL(profile, "DictFilterRewriteTime",
parquet_profile, 1);
dict_filter_expr_rewrite_time =
@@ -193,6 +201,10 @@ void ParquetProfile::init(RuntimeProfile* profile) {
profile, "DictFilterCandidateColumns", TUnit::UNIT,
parquet_profile, 1);
dict_filter_columns = ADD_CHILD_COUNTER_WITH_LEVEL(profile,
"DictFilterColumns", TUnit::UNIT,
parquet_profile, 1);
+ dict_filter_typed_compare_columns = ADD_CHILD_COUNTER_WITH_LEVEL(
+ profile, "DictFilterTypedCompareColumns", TUnit::UNIT,
parquet_profile, 1);
+ dict_filter_string_compare_columns = ADD_CHILD_COUNTER_WITH_LEVEL(
+ profile, "DictFilterStringCompareColumns", TUnit::UNIT,
parquet_profile, 1);
dict_filter_unsupported_columns = ADD_CHILD_COUNTER_WITH_LEVEL(
profile, "DictFilterUnsupportedColumns", TUnit::UNIT,
parquet_profile, 1);
dict_filter_read_failures = ADD_CHILD_COUNTER_WITH_LEVEL(profile,
"DictFilterReadFailures",
@@ -276,6 +288,7 @@ ParquetColumnReaderProfile
ParquetProfile::column_reader_profile() const {
.hybrid_selection_batches = hybrid_selection_batches,
.hybrid_selection_ranges = hybrid_selection_ranges,
.hybrid_selection_null_fallback_batches =
hybrid_selection_null_fallback_batches,
+ .dictionary_predicate_fused_projected_rows =
dictionary_predicate_fused_projected_rows,
.decompress_time = decompress_time,
.decompress_count = decompress_cnt,
.decode_header_time = decode_header_time,
@@ -321,12 +334,17 @@ ParquetScanProfile ParquetProfile::scan_profile() const {
.predicate_alignment_columns = predicate_alignment_columns,
.fixed_width_predicate_direct_batches =
fixed_width_predicate_direct_batches,
.fixed_width_predicate_direct_rows =
fixed_width_predicate_direct_rows,
+ .dictionary_predicate_direct_batches =
dictionary_predicate_direct_batches,
+ .dictionary_predicate_direct_rows =
dictionary_predicate_direct_rows,
+ .dictionary_predicate_projected_rows =
dictionary_predicate_projected_rows,
.dict_filter_rewrite_time = dict_filter_rewrite_time,
.dict_filter_expr_rewrite_time = dict_filter_expr_rewrite_time,
.dict_filter_read_dict_time = dict_filter_read_dict_time,
.dict_filter_build_time = dict_filter_build_time,
.dict_filter_candidate_columns = dict_filter_candidate_columns,
.dict_filter_columns = dict_filter_columns,
+ .dict_filter_typed_compare_columns =
dict_filter_typed_compare_columns,
+ .dict_filter_string_compare_columns =
dict_filter_string_compare_columns,
.dict_filter_unsupported_columns = dict_filter_unsupported_columns,
.dict_filter_read_failures = dict_filter_read_failures,
.rows_filtered_by_dict_filter = rows_filtered_by_dict_filter,
diff --git a/be/src/format_v2/parquet/parquet_profile.h
b/be/src/format_v2/parquet/parquet_profile.h
index 438d1a9a4b2..80198bc7a14 100644
--- a/be/src/format_v2/parquet/parquet_profile.h
+++ b/be/src/format_v2/parquet/parquet_profile.h
@@ -41,6 +41,7 @@ struct ParquetColumnReaderProfile {
RuntimeProfile::Counter* hybrid_selection_batches = nullptr;
RuntimeProfile::Counter* hybrid_selection_ranges = nullptr;
RuntimeProfile::Counter* hybrid_selection_null_fallback_batches = nullptr;
+ RuntimeProfile::Counter* dictionary_predicate_fused_projected_rows =
nullptr;
// Native page/encoding reader internals. These counters keep page IO,
decompression, levels,
// value decode and conversion attributable to separate stages.
RuntimeProfile::Counter* decompress_time = nullptr;
@@ -91,14 +92,21 @@ struct ParquetScanProfile {
RuntimeProfile::Counter* predicate_alignment_columns = nullptr;
RuntimeProfile::Counter* fixed_width_predicate_direct_batches = nullptr;
RuntimeProfile::Counter* fixed_width_predicate_direct_rows = nullptr;
+ RuntimeProfile::Counter* dictionary_predicate_direct_batches = nullptr;
+ RuntimeProfile::Counter* dictionary_predicate_direct_rows = nullptr;
+ RuntimeProfile::Counter* dictionary_predicate_projected_rows = nullptr;
RuntimeProfile::Counter* dict_filter_rewrite_time = nullptr; // dictionary
rewrite time (ns)
RuntimeProfile::Counter* dict_filter_expr_rewrite_time =
nullptr; // expression/residual rewrite time (ns)
RuntimeProfile::Counter* dict_filter_read_dict_time = nullptr; //
dictionary page read time (ns)
RuntimeProfile::Counter* dict_filter_build_time =
nullptr; // dictionary entry bitmap build time (ns)
- RuntimeProfile::Counter* dict_filter_candidate_columns = nullptr; //
candidate columns
- RuntimeProfile::Counter* dict_filter_columns = nullptr; //
optimized columns
+ RuntimeProfile::Counter* dict_filter_candidate_columns = nullptr; //
candidate columns
+ RuntimeProfile::Counter* dict_filter_columns = nullptr; //
optimized columns
+ RuntimeProfile::Counter* dict_filter_typed_compare_columns =
+ nullptr; // fixed-width typed comparison columns
+ RuntimeProfile::Counter* dict_filter_string_compare_columns =
+ nullptr; // string typed comparison columns
RuntimeProfile::Counter* dict_filter_unsupported_columns = nullptr; //
unsupported columns
RuntimeProfile::Counter* dict_filter_read_failures = nullptr; //
dictionary read failures
RuntimeProfile::Counter* rows_filtered_by_dict_filter = nullptr; //
rows filtered by dict
@@ -154,6 +162,7 @@ struct ParquetProfile {
RuntimeProfile::Counter* hybrid_selection_batches = nullptr;
RuntimeProfile::Counter* hybrid_selection_ranges = nullptr;
RuntimeProfile::Counter* hybrid_selection_null_fallback_batches = nullptr;
+ RuntimeProfile::Counter* dictionary_predicate_fused_projected_rows =
nullptr;
RuntimeProfile::Counter* native_read_calls = nullptr;
RuntimeProfile::Counter* native_page_fragments = nullptr;
RuntimeProfile::Counter* page_crossing_batches = nullptr;
@@ -207,12 +216,17 @@ struct ParquetProfile {
RuntimeProfile::Counter* predicate_alignment_columns = nullptr;
RuntimeProfile::Counter* fixed_width_predicate_direct_batches = nullptr;
RuntimeProfile::Counter* fixed_width_predicate_direct_rows = nullptr;
+ RuntimeProfile::Counter* dictionary_predicate_direct_batches = nullptr;
+ RuntimeProfile::Counter* dictionary_predicate_direct_rows = nullptr;
+ RuntimeProfile::Counter* dictionary_predicate_projected_rows = nullptr;
RuntimeProfile::Counter* dict_filter_rewrite_time = nullptr;
RuntimeProfile::Counter* dict_filter_expr_rewrite_time = nullptr;
RuntimeProfile::Counter* dict_filter_read_dict_time = nullptr;
RuntimeProfile::Counter* dict_filter_build_time = nullptr;
RuntimeProfile::Counter* dict_filter_candidate_columns = nullptr;
RuntimeProfile::Counter* dict_filter_columns = nullptr;
+ RuntimeProfile::Counter* dict_filter_typed_compare_columns = nullptr;
+ RuntimeProfile::Counter* dict_filter_string_compare_columns = nullptr;
RuntimeProfile::Counter* dict_filter_unsupported_columns = nullptr;
RuntimeProfile::Counter* dict_filter_read_failures = nullptr;
RuntimeProfile::Counter* rows_filtered_by_dict_filter = nullptr;
diff --git a/be/src/format_v2/parquet/parquet_scan.cpp
b/be/src/format_v2/parquet/parquet_scan.cpp
index 3730f303254..a60d3efbeea 100644
--- a/be/src/format_v2/parquet/parquet_scan.cpp
+++ b/be/src/format_v2/parquet/parquet_scan.cpp
@@ -31,7 +31,9 @@
#include "core/assert_cast.h"
#include "core/block/block.h"
#include "core/column/column_vector.h"
+#include "exprs/expr_zonemap_filter.h"
#include "exprs/vcompound_pred.h"
+#include "exprs/vectorized_fn_call.h"
#include "exprs/vexpr_context.h"
#include "format_v2/parquet/parquet_column_schema.h"
#include "format_v2/parquet/parquet_file_context.h"
@@ -1392,14 +1394,178 @@ uint16_t count_selected_rows(const IColumn::Filter&
filter) {
return selected_rows;
}
-IColumn::Filter build_dictionary_entry_filter(size_t block_position,
- const ParquetColumnSchema&
column_schema,
- const VExprContextSPtrs&
conjuncts,
- const IColumn& dictionary) {
- IColumn::Filter dictionary_filter(dictionary.size(), 1);
+enum class DictionaryEntryFilterKernel {
+ GENERIC,
+ TYPED_FIXED_WIDTH,
+ TYPED_STRING,
+};
+
+template <typename ColumnType>
+bool get_fixed_dictionary_raw_values(const IColumn& dictionary, const
uint8_t** values,
+ size_t* value_width) {
+ const auto* typed_dictionary =
check_and_get_column<ColumnType>(dictionary);
+ if (typed_dictionary == nullptr) {
+ return false;
+ }
+ *values = reinterpret_cast<const
uint8_t*>(typed_dictionary->get_data().data());
+ *value_width = sizeof(typename ColumnType::value_type);
+ return true;
+}
+
+bool get_numeric_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);
+ default:
+ return false;
+ }
+}
+
+enum class StringDictionaryCompareOp {
+ EQ,
+ NE,
+ LT,
+ LE,
+ GT,
+ GE,
+};
+
+std::optional<StringDictionaryCompareOp>
string_dictionary_compare_op(std::string_view name,
+ bool
reverse) {
+ StringDictionaryCompareOp op;
+ if (name == "eq") {
+ op = StringDictionaryCompareOp::EQ;
+ } else if (name == "ne") {
+ op = StringDictionaryCompareOp::NE;
+ } else if (name == "lt") {
+ op = StringDictionaryCompareOp::LT;
+ } else if (name == "le") {
+ op = StringDictionaryCompareOp::LE;
+ } else if (name == "gt") {
+ op = StringDictionaryCompareOp::GT;
+ } else if (name == "ge") {
+ op = StringDictionaryCompareOp::GE;
+ } else {
+ return std::nullopt;
+ }
+ if (!reverse || op == StringDictionaryCompareOp::EQ || op ==
StringDictionaryCompareOp::NE) {
+ return op;
+ }
+ switch (op) {
+ case StringDictionaryCompareOp::LT:
+ return StringDictionaryCompareOp::GT;
+ case StringDictionaryCompareOp::LE:
+ return StringDictionaryCompareOp::GE;
+ case StringDictionaryCompareOp::GT:
+ return StringDictionaryCompareOp::LT;
+ case StringDictionaryCompareOp::GE:
+ return StringDictionaryCompareOp::LE;
+ default:
+ __builtin_unreachable();
+ }
+}
+
+bool string_compare_matches(int comparison, StringDictionaryCompareOp op) {
+ switch (op) {
+ case StringDictionaryCompareOp::EQ:
+ return comparison == 0;
+ case StringDictionaryCompareOp::NE:
+ return comparison != 0;
+ case StringDictionaryCompareOp::LT:
+ return comparison < 0;
+ case StringDictionaryCompareOp::LE:
+ return comparison <= 0;
+ case StringDictionaryCompareOp::GT:
+ return comparison > 0;
+ case StringDictionaryCompareOp::GE:
+ return comparison >= 0;
+ }
+ __builtin_unreachable();
+}
+
+bool try_apply_string_dictionary_conjunct(size_t block_position, const
DataTypePtr& column_type,
+ const VExprSPtr& root, const
IColumn& dictionary,
+ IColumn::Filter* dictionary_filter) {
+ const auto fn = std::dynamic_pointer_cast<VectorizedFnCall>(root);
+ if (fn == nullptr || (!dictionary.is_column_string() &&
!dictionary.is_column_string64())) {
+ return false;
+ }
+ const auto slot_literal =
expr_zonemap::extract_slot_and_literal(fn->children());
+ if (!slot_literal.has_value() || slot_literal->slot_index !=
block_position ||
+ slot_literal->literal.get_type() != TYPE_STRING ||
+
!remove_nullable(slot_literal->slot_type)->equals(*remove_nullable(column_type))
||
+
!remove_nullable(slot_literal->literal_type)->equals(*remove_nullable(column_type)))
{
+ return false;
+ }
+ const auto op =
+ string_dictionary_compare_op(fn->function_name(),
slot_literal->literal_on_left);
+ if (!op.has_value()) {
+ return false;
+ }
+ const auto& literal = slot_literal->literal.get<TYPE_STRING>();
+ const StringRef literal_ref(literal.data(), literal.size());
+ for (size_t dictionary_id = 0; dictionary_id < dictionary.size();
++dictionary_id) {
+ const int comparison =
dictionary.get_data_at(dictionary_id).compare(literal_ref);
+ (*dictionary_filter)[dictionary_id] &=
string_compare_matches(comparison, *op) ? 1 : 0;
+ }
+ return true;
+}
+
+Status build_dictionary_entry_filter(size_t block_position,
+ const ParquetColumnSchema& column_schema,
+ const VExprContextSPtrs& conjuncts, const
IColumn& dictionary,
+ IColumn::Filter* dictionary_filter,
+ DictionaryEntryFilterKernel* kernel) {
+ DORIS_CHECK(dictionary_filter != nullptr);
+ DORIS_CHECK(kernel != nullptr);
+ dictionary_filter->clear();
+ dictionary_filter->resize_fill(dictionary.size(), 1);
+ *kernel = DictionaryEntryFilterKernel::GENERIC;
+ // Block positions are expression slot IDs here; validate the narrowing
once so every
+ // dictionary evaluation path uses the same representable ID.
+ const int expression_column_id = cast_set<int>(block_position);
+ const auto typed_data_type = remove_nullable(column_schema.type);
+ const uint8_t* raw_values = nullptr;
+ size_t value_width = 0;
+ if (std::ranges::all_of(conjuncts,
+ [&](const auto& conjunct) {
+ 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)) {
+ // 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) {
+ RETURN_IF_ERROR(conjunct->root()->execute_on_raw_fixed_values(
+ raw_values, dictionary.size(), value_width,
column_schema.type,
+ expression_column_id, dictionary_filter->data()));
+ }
+ *kernel = DictionaryEntryFilterKernel::TYPED_FIXED_WIDTH;
+ return Status::OK();
+ }
+
+ if (std::ranges::all_of(conjuncts, [&](const auto& conjunct) {
+ return try_apply_string_dictionary_conjunct(block_position,
column_schema.type,
+ conjunct->root(),
dictionary,
+ dictionary_filter);
+ })) {
+ *kernel = DictionaryEntryFilterKernel::TYPED_STRING;
+ return Status::OK();
+ }
+
+ dictionary_filter->clear();
+ dictionary_filter->resize_fill(dictionary.size(), 1);
DictionaryEvalContext ctx;
auto& slot = ctx.slots
- .emplace(static_cast<int>(block_position),
+ .emplace(expression_column_id,
DictionaryEvalContext::SlotDictionary {
.data_type = column_schema.type,
.values = {}})
.first->second;
@@ -1409,12 +1575,13 @@ IColumn::Filter build_dictionary_entry_filter(size_t
block_position,
dictionary.get(dictionary_id, value);
slot.values.clear();
slot.values.push_back(std::move(value));
- dictionary_filter[dictionary_id] =
VExprContext::evaluate_dictionary_filter(
- conjuncts, ctx) ==
ZoneMapFilterResult::kNoMatch
- ? 0
- : 1;
+ (*dictionary_filter)[dictionary_id] =
+ VExprContext::evaluate_dictionary_filter(conjuncts, ctx) ==
+ ZoneMapFilterResult::kNoMatch
+ ? 0
+ : 1;
}
- return dictionary_filter;
+ return Status::OK();
}
} // namespace
@@ -1499,8 +1666,15 @@ Status
ParquetScanScheduler::prepare_current_dictionary_filters(
OwnedExpressionConjuncts residual_conjuncts;
{
SCOPED_TIMER(_scan_profile.dict_filter_build_time);
- dictionary_filter = build_dictionary_entry_filter(
- block_position, *column_schema, conjunct_it->second,
*dictionary_values);
+ DictionaryEntryFilterKernel filter_kernel =
DictionaryEntryFilterKernel::GENERIC;
+ RETURN_IF_ERROR(build_dictionary_entry_filter(block_position,
*column_schema,
+ conjunct_it->second,
*dictionary_values,
+ &dictionary_filter,
&filter_kernel));
+ if (filter_kernel ==
DictionaryEntryFilterKernel::TYPED_FIXED_WIDTH) {
+
update_counter_if_not_null(_scan_profile.dict_filter_typed_compare_columns, 1);
+ } else if (filter_kernel ==
DictionaryEntryFilterKernel::TYPED_STRING) {
+
update_counter_if_not_null(_scan_profile.dict_filter_string_compare_columns, 1);
+ }
residual_conjuncts =
build_dictionary_residual_conjuncts(conjunct_it->second);
}
@@ -1659,12 +1833,23 @@ Status
ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
const uint16_t selected_rows_before = *selected_rows;
IColumn::Filter compact_filter;
bool used_filter = false;
+ const bool predicate_only = request.is_predicate_only(local_id);
+ // Dictionary ids are sufficient for predicate-only slots;
skipping typed survivor
+ // gathers preserves the block row shape without materializing an
unobservable payload.
+ 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, column,
- &compact_filter, &used_filter));
+ *selection, *selected_rows, batch_rows,
dictionary_filter_it->second,
+ projected_column, &compact_filter, &used_filter));
if (used_filter) {
DORIS_CHECK(compact_filter.size() == 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);
+ if (!predicate_only) {
+
update_counter_if_not_null(_scan_profile.dictionary_predicate_projected_rows,
+ new_selected_rows);
+ }
const auto filtered_rows =
static_cast<int64_t>(selected_rows_before) -
static_cast<int64_t>(new_selected_rows);
if (conjunct_filtered_rows != nullptr) {
@@ -1679,7 +1864,13 @@ Status ParquetScanScheduler::read_filter_columns(int64_t
batch_rows,
*selected_rows =
apply_compact_filter_to_selection(compact_filter, selection,
selected_rows_before);
}
- file_block->replace_by_position(block_position,
std::move(column));
+ if (predicate_only) {
+ auto placeholder = column->clone_empty();
+ placeholder->insert_many_defaults(*selected_rows);
+ file_block->replace_by_position(block_position,
std::move(placeholder));
+ } else {
+ file_block->replace_by_position(block_position,
std::move(column));
+ }
read_column_positions.push_back(cast_set<uint32_t>(block_position));
remember_column_selection(cast_set<uint32_t>(block_position));
*used_dictionary_filter = true;
diff --git a/be/src/format_v2/parquet/reader/column_reader.cpp
b/be/src/format_v2/parquet/reader/column_reader.cpp
index db5def75b8c..151206e4295 100644
--- a/be/src/format_v2/parquet/reader/column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/column_reader.cpp
@@ -75,7 +75,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&, MutableColumnPtr&,
+ const
IColumn::Filter&, IColumn*,
IColumn::Filter*,
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 6914a5be7d1..82848e4ec58 100644
--- a/be/src/format_v2/parquet/reader/column_reader.h
+++ b/be/src/format_v2/parquet/reader/column_reader.h
@@ -58,7 +58,7 @@ public:
virtual Status select_with_dictionary_filter(const SelectionVector&
selection,
uint16_t selected_rows,
int64_t batch_rows,
const IColumn::Filter&
dictionary_filter,
- MutableColumnPtr& column,
+ IColumn* projected_column,
IColumn::Filter* row_filter,
bool* used_filter);
// Consume batch_rows and evaluate eligible fixed-width values without
first constructing a
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 4a97daaf21e..0d59b7e6bc2 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
@@ -1712,6 +1712,201 @@ Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::filter_fixed_width_values
return Status::OK();
}
+namespace {
+
+template <typename ColumnType>
+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) {
+ if (typed_dictionary == nullptr || projected_values == nullptr) {
+ return false;
+ }
+ const auto* dictionary =
check_and_get_column<ColumnType>(*typed_dictionary);
+ auto* projected = check_and_get_column<ColumnType>(*projected_values);
+ if (dictionary == nullptr || projected == nullptr) {
+ return false;
+ }
+
+ 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());
+ size_t physical_row = 0;
+ for (const uint8_t is_null : nullable_selection_nulls) {
+ bool keep = false;
+ if (is_null == 0) {
+ const uint32_t dictionary_id =
selected_dictionary_indices[physical_row++];
+ // The decoder validates the complete id batch first, preserving
atomic output while
+ // allowing this hot gather loop to use unchecked dictionary
lookups.
+ keep = dictionary_filter[dictionary_id] != 0;
+ if (keep) {
+ projected_data.push_back(dictionary_data[dictionary_id]);
+ }
+ }
+ row_filter->push_back(keep ? 1 : 0);
+ }
+ DORIS_CHECK_EQ(physical_row, selected_dictionary_indices.size());
+ return true;
+}
+
+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;
\
+ }
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnUInt8)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnInt8)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnInt16)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnInt32)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnInt64)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnInt128)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnFloat32)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnFloat64)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnDate)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnDateTime)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnDateV2)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnDateTimeV2)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnTimeV2)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnTimeStampTz)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnIPv4)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnIPv6)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnOffset32)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnOffset64)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnDecimal32)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnDecimal64)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnDecimal128V2)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnDecimal128V3)
+ TRY_FIXED_WIDTH_DICTIONARY_COLUMN(ColumnDecimal256)
+#undef TRY_FIXED_WIDTH_DICTIONARY_COLUMN
+ return false;
+}
+
+} // namespace
+
+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) {
+ DORIS_CHECK(row_filter != 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();
+ if (_current_encoding != tparquet::Encoding::RLE_DICTIONARY ||
_page_decoder == nullptr ||
+ !_page_decoder->has_dictionary()) {
+ return Status::OK();
+ }
+ if (UNLIKELY(_remaining_num_values < select_vector.num_values())) {
+ return Status::IOError("Decode too many values in current page");
+ }
+ if (UNLIKELY(dictionary_filter.size() !=
_page_decoder->dictionary_size())) {
+ return Status::Corruption("Parquet predicate dictionary has {}
entries, expected {}",
+ dictionary_filter.size(),
_page_decoder->dictionary_size());
+ }
+
+ ParquetSelection selection;
+ _nullable_selection_nulls.clear();
+ _nullable_selection_nulls.reserve(select_vector.num_values() -
select_vector.num_filtered());
+ size_t physical_cursor = 0;
+ auto build_selection = [&]<bool HAS_FILTER>() {
+ ColumnSelectVector::DataReadType read_type;
+ while (const size_t run_length =
select_vector.get_next_run<HAS_FILTER>(&read_type)) {
+ switch (read_type) {
+ case ColumnSelectVector::CONTENT:
+ if (!selection.ranges.empty() &&
+ selection.ranges.back().first +
selection.ranges.back().count ==
+ physical_cursor) {
+ selection.ranges.back().count += run_length;
+ } else {
+ selection.ranges.push_back({.first = physical_cursor,
.count = run_length});
+ }
+ selection.selected_values += run_length;
+
_nullable_selection_nulls.resize_fill(_nullable_selection_nulls.size() +
run_length,
+ 0);
+ physical_cursor += run_length;
+ break;
+ case ColumnSelectVector::NULL_DATA:
+
_nullable_selection_nulls.resize_fill(_nullable_selection_nulls.size() +
run_length,
+ 1);
+ break;
+ case ColumnSelectVector::FILTERED_CONTENT:
+ physical_cursor += run_length;
+ break;
+ case ColumnSelectVector::FILTERED_NULL:
+ break;
+ }
+ }
+ };
+ if (select_vector.has_filter()) {
+ build_selection.template operator()<true>();
+ } else {
+ build_selection.template operator()<false>();
+ }
+ selection.total_values = physical_cursor;
+ DORIS_CHECK_EQ(selection.total_values, select_vector.num_values() -
select_vector.num_nulls());
+ DORIS_CHECK_EQ(_nullable_selection_nulls.size(),
+ select_vector.num_values() - select_vector.num_filtered());
+ if (UNLIKELY(_empty_value_section && selection.total_values != 0)) {
+ return Status::Corruption(
+ "Parquet definition levels require {} values from an empty
value section",
+ selection.total_values);
+ }
+
+ _selected_dictionary_indices.clear();
+ if (selection.selected_values == 0) {
+ RETURN_IF_ERROR(_page_decoder->skip_values(selection.total_values));
+ } else {
+ SCOPED_RAW_TIMER(&_chunk_statistics.decode_value_time);
+ RETURN_IF_ERROR(_page_decoder->decode_selected_dictionary_indices(
+ selection, &_selected_dictionary_indices));
+ }
+ DORIS_CHECK_EQ(_selected_dictionary_indices.size(),
selection.selected_values);
+
+ 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);
+ 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());
+ size_t physical_row = 0;
+ for (const uint8_t is_null : _nullable_selection_nulls) {
+ bool keep = false;
+ if (is_null == 0) {
+ const uint32_t dictionary_id =
_selected_dictionary_indices[physical_row++];
+ // The complete id batch was validated before this loop, so
unchecked bitmap access
+ // cannot leak partial output for a corrupt page.
+ keep = dictionary_filter[dictionary_id] != 0;
+ if (keep && matched != nullptr) {
+ matched->push_back(cast_set<int32_t>(dictionary_id));
+ }
+ }
+ row_filter->push_back(keep ? 1 : 0);
+ }
+ DORIS_CHECK_EQ(physical_row, _selected_dictionary_indices.size());
+ }
+ // Commit page progress only after every external dictionary id has been
validated.
+ _remaining_num_values -= select_vector.num_values();
+ *projected_directly = direct_fixed_width_projection;
+ *used_filter = true;
+ return Status::OK();
+}
+
template <bool IN_COLLECTION, bool OFFSET_INDEX>
Status ColumnChunkReader<IN_COLLECTION,
OFFSET_INDEX>::seek_to_nested_row(size_t left_row) {
if constexpr (OFFSET_INDEX) {
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 102366f2494..4a6dbecd4c5 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
@@ -30,6 +30,7 @@
#include "common/status.h"
#include "core/column/column_string.h"
+#include "core/column/column_vector.h"
#include "core/data_type/data_type.h"
#include "core/data_type_serde/parquet_decode_source.h"
#include "exprs/vexpr_fwd.h"
@@ -204,6 +205,13 @@ public:
IColumn::Filter* row_filter, bool*
used_filter);
bool can_filter_fixed_width_values(const VExprSPtrs& conjuncts, int
column_id) const;
+ Status 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);
+
// Get the repetition level decoder of current page.
LevelDecoder& rep_level_decoder() { return _rep_level_decoder; }
// Get the definition level decoder of current page.
@@ -224,6 +232,9 @@ public:
// Level decoders may batch-convert unsigned RLE values into Doris'
signed level_t.
_rep_level_decoder.release_scratch(max_retained_bytes);
_def_level_decoder.release_scratch(max_retained_bytes);
+ if (_selected_dictionary_indices.capacity() * sizeof(uint32_t) >
max_retained_bytes) {
+ std::vector<uint32_t>().swap(_selected_dictionary_indices);
+ }
if (_decompress_buf_size > max_retained_bytes) {
if (_page_uses_decompress_buf) {
// Keep the request until the page boundary because decoders
still point into this
@@ -246,7 +257,7 @@ public:
for (const auto& [encoding, decoder] : _decoders) {
bytes += decoder->retained_scratch_bytes();
}
- return bytes;
+ return bytes + _selected_dictionary_indices.capacity() *
sizeof(uint32_t);
}
size_t active_decoder_scratch_bytes() const {
@@ -255,7 +266,8 @@ public:
return _active_decompress_bytes +
(_page_decoder == nullptr ? 0 :
_page_decoder->active_scratch_bytes()) +
_rep_level_decoder.active_scratch_bytes() +
- _def_level_decoder.active_scratch_bytes();
+ _def_level_decoder.active_scratch_bytes() +
+ _selected_dictionary_indices.size() * sizeof(uint32_t);
}
tparquet::Encoding::type current_encoding() const { return
_current_encoding; }
@@ -408,6 +420,7 @@ private:
// Plain or Dictionary encoding. If the dictionary grows too big, the
encoding will fall back to the plain encoding
std::unordered_map<int, std::unique_ptr<Decoder>> _decoders;
NullMap _nullable_selection_nulls;
+ std::vector<uint32_t> _selected_dictionary_indices;
ColumnChunkReaderStatistics _chunk_statistics;
};
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 592e048846e..3f06d92dc9a 100644
--- a/be/src/format_v2/parquet/reader/native/column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/native/column_reader.cpp
@@ -38,6 +38,7 @@
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_struct.h"
#include "core/data_type/define_primitive_type.h"
+#include "core/data_type_serde/parquet_decode_source.h"
#include "exprs/vexpr.h"
#include "format_v2/parquet/native_schema_desc.h"
#include "format_v2/parquet/reader/native/column_chunk_reader.h"
@@ -75,6 +76,36 @@ size_t retained_set_bytes(const std::unordered_set<size_t>&
values) {
return values.bucket_count() * sizeof(void*) + values.size() *
sizeof(size_t);
}
+bool append_compact_int_dictionary(const IColumn& dictionary, const uint32_t*
indices, size_t count,
+ IColumn* destination) {
+ auto* destination_int = check_and_get_column<ColumnInt32>(*destination);
+ const auto* dictionary_int = check_and_get_column<ColumnInt32>(dictionary);
+ if (destination_int == nullptr || dictionary_int == nullptr) {
+ return false;
+ }
+ if (count == 0) {
+ return true;
+ }
+ auto& output = destination_int->get_data();
+ const auto& values = dictionary_int->get_data();
+ const size_t old_size = output.size();
+ output.resize(old_size + count);
+ int32_t* dst = output.data() + old_size;
+ size_t row = 0;
+ // A tiny Parquet INT dictionary normally remains in L1. This V1-style
scalar gather avoids
+ // AVX gather setup and keeps the survivor-id loop compact enough for the
compiler to unroll.
+ for (; row + 4 <= count; row += 4) {
+ dst[row] = values[indices[row]];
+ dst[row + 1] = values[indices[row + 1]];
+ dst[row + 2] = values[indices[row + 2]];
+ dst[row + 3] = values[indices[row + 3]];
+ }
+ for (; row < count; ++row) {
+ dst[row] = values[indices[row]];
+ }
+ return true;
+}
+
Status validate_decimal_physical_type(const NativeFieldSchema& field, int
precision, int scale) {
if (precision <= 0 || scale < 0 || scale > precision) {
return Status::Corruption("Parquet decimal field {} has invalid
precision {} and scale {}",
@@ -1183,6 +1214,135 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::read_fixed_width_filter(
return Status::OK();
}
+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,
+ bool* projected_directly) {
+ DORIS_CHECK(row_filter != nullptr);
+ DORIS_CHECK(projected_directly != nullptr);
+ _null_run_lengths.clear();
+ if (_chunk_reader->max_def_level() > 0) {
+ LevelDecoder& def_decoder = _chunk_reader->def_level_decoder();
+ size_t has_read = 0;
+ bool prev_is_null = true;
+ while (has_read < num_values) {
+ level_t def_level = -1;
+ const size_t loop_read = def_decoder.get_next_run(&def_level,
num_values - has_read);
+ if (loop_read == 0) {
+ return Status::Corruption(
+ "Parquet definition level stream ended while filtering
dictionary ids");
+ }
+ const bool is_null = def_level < _field_schema->definition_level;
+ if (!(prev_is_null ^ is_null)) {
+ _null_run_lengths.emplace_back(0);
+ }
+ size_t remaining = loop_read;
+ while (remaining > USHRT_MAX) {
+ _null_run_lengths.emplace_back(USHRT_MAX);
+ _null_run_lengths.emplace_back(0);
+ remaining -= USHRT_MAX;
+ }
+ _null_run_lengths.emplace_back(cast_set<uint16_t>(remaining));
+ prev_is_null = is_null;
+ has_read += loop_read;
+ }
+ } else {
+ size_t remaining = num_values;
+ while (remaining > USHRT_MAX) {
+ _null_run_lengths.emplace_back(USHRT_MAX);
+ _null_run_lengths.emplace_back(0);
+ remaining -= USHRT_MAX;
+ }
+ _null_run_lengths.emplace_back(cast_set<uint16_t>(remaining));
+ }
+ RETURN_IF_ERROR(_select_vector.init(_null_run_lengths, num_values,
nullptr, &filter_map,
+ _filter_map_index));
+ _filter_map_index += num_values;
+ 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));
+ // Pure-dictionary chunks are prevalidated before definition levels are
consumed.
+ DORIS_CHECK(used_filter);
+ return Status::OK();
+}
+
+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) {
+ DORIS_CHECK(row_filter != nullptr);
+ DORIS_CHECK(read_rows != nullptr);
+ DORIS_CHECK(eof != nullptr);
+ DORIS_CHECK(projected_directly != nullptr);
+ DORIS_CHECK(used_filter != nullptr);
+ row_filter->clear();
+ *read_rows = 0;
+ *projected_directly = false;
+ *used_filter = false;
+ if (_in_nested || dictionary_filter.empty()) {
+ return Status::OK();
+ }
+
+ int64_t right_row = 0;
+ if constexpr (OFFSET_INDEX == false) {
+ RETURN_IF_ERROR(_chunk_reader->parse_page_header());
+ right_row = _chunk_reader->page_end_row();
+ } else {
+ right_row = _chunk_reader->page_end_row();
+ }
+ RowRanges read_ranges;
+ _generate_read_ranges(RowRange {_current_row_index, right_row},
&read_ranges);
+ if (read_ranges.count() == 0) {
+ _current_row_index = right_row;
+ } else {
+ RETURN_IF_ERROR(_chunk_reader->parse_page_header());
+ RETURN_IF_ERROR(_chunk_reader->load_page_data_idempotent());
+ if (_chunk_reader->current_encoding() !=
tparquet::Encoding::RLE_DICTIONARY) {
+ return Status::OK();
+ }
+ size_t has_read = 0;
+ for (size_t idx = 0; idx < read_ranges.range_size(); ++idx) {
+ const auto range = read_ranges.get_range(idx);
+ const size_t skip_values = range.from() - _current_row_index;
+ RETURN_IF_ERROR(_skip_values(skip_values));
+ _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;
+ 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));
+ 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());
+ has_read += values;
+ *read_rows += values;
+ _current_row_index += values;
+ if (has_read == batch_size) {
+ break;
+ }
+ }
+ }
+
+ if (right_row == _current_row_index) {
+ if (!_chunk_reader->has_next_page()) {
+ *eof = true;
+ } else {
+ RETURN_IF_ERROR(_chunk_reader->next_page());
+ }
+ }
+ *used_filter = true;
+ return Status::OK();
+}
+
template <bool IN_COLLECTION, bool OFFSET_INDEX>
Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::read_column_levels(FilterMap& filter_map,
size_t batch_size,
@@ -1259,10 +1419,8 @@ Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::read_column_levels(Filte
}
template <bool IN_COLLECTION, bool OFFSET_INDEX>
-Result<MutableColumnPtr>
-ScalarColumnReader<IN_COLLECTION, OFFSET_INDEX>::materialize_dictionary_values(
- const ColumnInt32* dict_column, const DataTypePtr& target_type) {
- DORIS_CHECK(dict_column != nullptr);
+Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::_ensure_typed_dictionary(
+ const DataTypePtr& target_type) {
DORIS_CHECK(target_type != nullptr);
Decoder* dictionary_decoder = _chunk_reader->dictionary_decoder();
DORIS_CHECK(dictionary_decoder != nullptr);
@@ -1284,7 +1442,7 @@ ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::materialize_dictionary_values(
auto status = dictionary_serde->read_parquet_dictionary(
*_materialization_state.typed_dictionary, *dictionary_decoder,
dictionary_context);
if (!status.ok()) {
- return ResultError(std::move(status));
+ return status;
}
DORIS_CHECK_EQ(_materialization_state.typed_dictionary->size(),
dictionary_decoder->dictionary_size());
@@ -1293,8 +1451,24 @@ ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::materialize_dictionary_values(
++_dictionary_materialization_count;
#endif
}
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::prepare_typed_dictionary(
+ const DataTypePtr& target_type, const IColumn** dictionary) {
+ DORIS_CHECK(dictionary != nullptr);
+ RETURN_IF_ERROR(_ensure_typed_dictionary(target_type));
+ *dictionary = _materialization_state.typed_dictionary.get();
+ return Status::OK();
+}
- auto result = _materialization_state.typed_dictionary->clone_empty();
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Status ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::append_dictionary_values(
+ const ColumnInt32* dict_column, const DataTypePtr& target_type,
IColumn* destination) {
+ DORIS_CHECK(dict_column != nullptr);
+ DORIS_CHECK(destination != nullptr);
+ RETURN_IF_ERROR(_ensure_typed_dictionary(target_type));
const auto& source_indices = dict_column->get_data();
auto& indices = _materialization_state.dictionary_indices;
indices.resize(source_indices.size());
@@ -1302,14 +1476,45 @@ ScalarColumnReader<IN_COLLECTION,
OFFSET_INDEX>::materialize_dictionary_values(
if (UNLIKELY(source_indices[row] < 0 ||
static_cast<size_t>(source_indices[row]) >=
_materialization_state.typed_dictionary->size()))
{
- return ResultError(Status::Corruption(
+ return Status::Corruption(
"Parquet dictionary index {} at row {} exceeds dictionary
size {}",
- source_indices[row], row,
_materialization_state.typed_dictionary->size()));
+ source_indices[row], row,
_materialization_state.typed_dictionary->size());
}
indices[row] = static_cast<uint32_t>(source_indices[row]);
}
- result->insert_indices_from(*_materialization_state.typed_dictionary,
indices.data(),
- indices.data() + indices.size());
+
+ IColumn* values_destination = destination;
+ ColumnNullable* nullable_destination =
check_and_get_column<ColumnNullable>(*destination);
+ if (nullable_destination != nullptr) {
+ values_destination = &nullable_destination->get_nested_column();
+ }
+ // The compact id array contains survivors only. INT uses the V1-style
L1-friendly gather;
+ // other fixed-width types use SIMD and variable-width types retain
generic typed insertion.
+ if
(!append_compact_int_dictionary(*_materialization_state.typed_dictionary,
indices.data(),
+ indices.size(), values_destination) &&
+ !try_simd_insert_parquet_dictionary_indices(*values_destination,
+
*_materialization_state.typed_dictionary,
+ indices.data(),
indices.size())) {
+
values_destination->insert_indices_from(*_materialization_state.typed_dictionary,
+ indices.data(), indices.data()
+ indices.size());
+ }
+ if (nullable_destination != nullptr) {
+ auto& null_map = nullable_destination->get_null_map_data();
+ null_map.resize_fill(null_map.size() + indices.size(), 0);
+ }
+ return Status::OK();
+}
+
+template <bool IN_COLLECTION, bool OFFSET_INDEX>
+Result<MutableColumnPtr>
+ScalarColumnReader<IN_COLLECTION, OFFSET_INDEX>::materialize_dictionary_values(
+ const ColumnInt32* dict_column, const DataTypePtr& target_type) {
+ DORIS_CHECK(target_type != nullptr);
+ auto result = remove_nullable(target_type)->create_column();
+ auto status = append_dictionary_values(dict_column, target_type,
result.get());
+ if (!status.ok()) {
+ return ResultError(std::move(status));
+ }
return result;
}
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 6ea2787f9f0..d96d11e4886 100644
--- a/be/src/format_v2/parquet/reader/native/column_reader.h
+++ b/be/src/format_v2/parquet/reader/native/column_reader.h
@@ -201,6 +201,22 @@ public:
return Status::OK();
}
+ 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) {
+ DORIS_CHECK(row_filter != nullptr);
+ DORIS_CHECK(read_rows != nullptr);
+ DORIS_CHECK(eof != nullptr);
+ DORIS_CHECK(projected_directly != nullptr);
+ DORIS_CHECK(used_filter != nullptr);
+ row_filter->clear();
+ *read_rows = 0;
+ *projected_directly = false;
+ *used_filter = false;
+ return Status::OK();
+ }
+
// Consume a nested batch while retaining only definition/repetition
levels. This is used when
// schema evolution makes every projected STRUCT child synthetic: the
parent still needs one
// physical leaf's shape, but decoding that leaf's strings or other
payload would be wasted.
@@ -212,6 +228,12 @@ public:
throw Exception(
Status::FatalError("Method materialize_dictionary_values is
not supported"));
}
+ virtual Status append_dictionary_values(const ColumnInt32*, const
DataTypePtr&, IColumn*) {
+ return Status::NotSupported("Appending parquet dictionary values is
not supported");
+ }
+ virtual Status prepare_typed_dictionary(const DataTypePtr&, const
IColumn**) {
+ return Status::NotSupported("Typed parquet dictionary is not
supported");
+ }
virtual Result<MutableColumnPtr> dictionary_values(const DataTypePtr&
target_type) {
return ResultError(Status::NotSupported("Parquet dictionary values are
not supported"));
}
@@ -283,10 +305,19 @@ public:
FilterMap& filter_map, size_t batch_size,
IColumn* projected_column, IColumn::Filter*
row_filter,
size_t* read_rows, bool* eof, bool*
used_filter) override;
+ Status read_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;
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,
const DataTypePtr&
target_type) override;
+ Status append_dictionary_values(const ColumnInt32* dict_column, const
DataTypePtr& target_type,
+ IColumn* destination) override;
+ Status prepare_typed_dictionary(const DataTypePtr& target_type,
+ const IColumn** dictionary) override;
Result<MutableColumnPtr> dictionary_values(const DataTypePtr& target_type)
override;
const std::vector<level_t>& get_rep_level() const override { return
_rep_levels; }
const std::vector<level_t>& get_def_level() const override { return
_def_levels; }
@@ -408,9 +439,16 @@ private:
Status _read_fixed_width_filter_values(size_t num_values, const
VExprSPtrs& conjuncts,
int column_id, FilterMap&
filter_map,
IColumn* projected_column,
IColumn::Filter* row_filter);
+ Status _read_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, 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);
+ Status _ensure_typed_dictionary(const DataTypePtr& target_type);
Status _try_load_dict_page(bool* loaded, bool* has_dict);
};
diff --git a/be/src/format_v2/parquet/reader/native/decoder.cpp
b/be/src/format_v2/parquet/reader/native/decoder.cpp
index 5f6a978317b..c068cafaf0d 100644
--- a/be/src/format_v2/parquet/reader/native/decoder.cpp
+++ b/be/src/format_v2/parquet/reader/native/decoder.cpp
@@ -122,7 +122,10 @@ Status Decoder::get_decoder(tparquet::Type::type type,
tparquet::Encoding::type
switch (encoding) {
case tparquet::Encoding::PLAIN:
return create_plain_decoder(type, decoder);
+ case tparquet::Encoding::PLAIN_DICTIONARY:
case tparquet::Encoding::RLE_DICTIONARY:
+ // PLAIN_DICTIONARY is the legacy page enum for the same
RLE/bit-packed id stream; accepting
+ // it here keeps every dictionary decode entry point consistent with
ColumnChunkReader.
return create_dictionary_decoder(type, decoder);
case tparquet::Encoding::RLE:
if (type != tparquet::Type::BOOLEAN) {
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 233b37c096e..ab86e2aea8c 100644
--- a/be/src/format_v2/parquet/reader/native_column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/native_column_reader.cpp
@@ -143,20 +143,6 @@ void collect_projected_ids(const ParquetColumnSchema&
schema,
}
}
-Status append_non_null_dictionary_values(MutableColumnPtr& target,
MutableColumnPtr values) {
- DORIS_CHECK(target);
- DORIS_CHECK(values);
- const size_t value_count = values->size();
- if (auto* nullable = check_and_get_column<ColumnNullable>(*target);
nullable != nullptr) {
- nullable->get_nested_column().insert_range_from(*values, 0,
value_count);
- auto& null_map = nullable->get_null_map_data();
- null_map.resize_fill(null_map.size() + value_count, 0);
- return Status::OK();
- }
- target->insert_range_from(*values, 0, value_count);
- return Status::OK();
-}
-
} // namespace
NativeColumnReader::NativeColumnReader(const ParquetColumnSchema& schema,
@@ -395,6 +381,73 @@ Status
NativeColumnReader::read_with_fixed_width_filter(int64_t rows, const uint
return Status::OK();
}
+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) {
+ DORIS_CHECK(rows >= 0);
+ DORIS_CHECK(row_filter != nullptr);
+ DORIS_CHECK(rows_read != nullptr);
+ DORIS_CHECK(projected_directly != nullptr);
+ DORIS_CHECK(used_filter != nullptr);
+ row_filter->clear();
+ *rows_read = 0;
+ *projected_directly = false;
+ *used_filter = false;
+ if (rows == 0) {
+ return Status::OK();
+ }
+
+ native::FilterMap filter;
+ RETURN_IF_ERROR(filter.init(filter_data, static_cast<size_t>(rows),
filter_all));
+ _native_reader->reset_filter_map_index();
+ bool eof = false;
+ int64_t consecutive_empty_calls = 0;
+ while (*rows_read < rows && !eof) {
+ size_t loop_rows = 0;
+ IColumn::Filter loop_filter;
+ 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));
+ if (!loop_used) {
+ if (UNLIKELY(*rows_read != 0)) {
+ return Status::Corruption(
+ "Parquet dictionary predicate encoding changed after
{} rows for column {}",
+ *rows_read, _name);
+ }
+ row_filter->clear();
+ return Status::OK();
+ }
+ if (*rows_read != 0) {
+ 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());
+ if (loop_rows == 0 && !eof) {
+ if (++consecutive_empty_calls > _row_group_rows + 1) {
+ return Status::Corruption(
+ "Native parquet dictionary predicate made no progress
for column {}",
+ _name);
+ }
+ continue;
+ }
+ consecutive_empty_calls = 0;
+ *rows_read += static_cast<int64_t>(loop_rows);
+ }
+ if (*rows_read != rows) {
+ return Status::Corruption(
+ "Native parquet dictionary predicate returned {} rows,
expected {} for {}",
+ *rows_read, rows, _name);
+ }
+ *used_filter = true;
+ release_batch_scratch_if_needed();
+ return Status::OK();
+}
+
void NativeColumnReader::release_batch_scratch_if_needed() {
// PLAIN predicate batches bypass materialization but share the same
persistent decoder tree,
// so both read paths must advance the retained-capacity aging clock.
@@ -515,7 +568,7 @@ Status NativeColumnReader::select(const SelectionVector&
selection, uint16_t sel
Status NativeColumnReader::select_with_dictionary_filter(const
SelectionVector& selection,
uint16_t
selected_rows, int64_t batch_rows,
const
IColumn::Filter& dictionary_filter,
- MutableColumnPtr&
column,
+ IColumn*
projected_column,
IColumn::Filter*
row_filter,
bool* used_filter) {
DORIS_CHECK(row_filter != nullptr);
@@ -530,6 +583,59 @@ Status
NativeColumnReader::select_with_dictionary_filter(const SelectionVector&
const uint8_t* filter_data = nullptr;
RETURN_IF_ERROR(selection.materialize_filter(selected_rows, batch_rows,
&filter_data));
+ ColumnInt32* direct_matched_ids = nullptr;
+ const IColumn* typed_dictionary = nullptr;
+ IColumn* projected_values = projected_column;
+ ColumnNullable* projected_nullable = nullptr;
+ if (projected_column != nullptr) {
+ if (!_matched_dictionary_ids) {
+ _matched_dictionary_ids = ColumnInt32::create();
+ }
+ _matched_dictionary_ids->clear();
+ direct_matched_ids =
check_and_get_column<ColumnInt32>(*_matched_dictionary_ids);
+ DORIS_CHECK(direct_matched_ids != nullptr);
+ RETURN_IF_ERROR(_native_reader->prepare_typed_dictionary(_type,
&typed_dictionary));
+ projected_nullable =
check_and_get_column<ColumnNullable>(*projected_column);
+ if (projected_nullable != nullptr) {
+ projected_values = &projected_nullable->get_nested_column();
+ }
+ }
+ int64_t direct_rows_read = 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));
+ 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}));
+ 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);
+ }
+ if (_profile.dictionary_predicate_fused_projected_rows !=
nullptr) {
+
COUNTER_UPDATE(_profile.dictionary_predicate_fused_projected_rows,
+ survivor_count);
+ }
+ } else {
+ DORIS_CHECK_EQ(direct_matched_ids->size(), survivor_count);
+
RETURN_IF_ERROR(_native_reader->append_dictionary_values(direct_matched_ids,
_type,
+
projected_column));
+ }
+ }
+ 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));
+ return Status::OK();
+ }
+
const bool nullable = _type->is_nullable();
DataTypePtr id_type = std::make_shared<DataTypeInt32>();
if (nullable) {
@@ -560,13 +666,18 @@ Status
NativeColumnReader::select_with_dictionary_filter(const SelectionVector&
}
DORIS_CHECK(ids != nullptr);
- if (!_matched_dictionary_ids) {
- _matched_dictionary_ids = ColumnInt32::create();
+ ColumnInt32::Container* matched_ids = nullptr;
+ if (projected_column != nullptr) {
+ if (!_matched_dictionary_ids) {
+ _matched_dictionary_ids = ColumnInt32::create();
+ }
+ _matched_dictionary_ids->clear();
+ matched_ids =
&assert_cast<ColumnInt32&>(*_matched_dictionary_ids).get_data();
+ matched_ids->reserve(selected_rows);
}
- _matched_dictionary_ids->clear();
- auto& matched_ids =
assert_cast<ColumnInt32&>(*_matched_dictionary_ids).get_data();
row_filter->reserve(selected_rows);
const auto& id_data = ids->get_data();
+ size_t survivor_count = 0;
for (size_t row = 0; row < selected_rows; ++row) {
bool keep = false;
if (null_map == nullptr || (*null_map)[row] == 0) {
@@ -579,22 +690,26 @@ Status
NativeColumnReader::select_with_dictionary_filter(const SelectionVector&
}
keep = dictionary_filter[static_cast<size_t>(dictionary_id)] != 0;
if (keep) {
- matched_ids.push_back(dictionary_id);
+ ++survivor_count;
+ if (matched_ids != nullptr) {
+ matched_ids->push_back(dictionary_id);
+ }
}
}
row_filter->push_back(keep ? 1 : 0);
}
- const auto* matched_id_column =
check_and_get_column<ColumnInt32>(*_matched_dictionary_ids);
- DORIS_CHECK(matched_id_column != nullptr);
- auto matched_values =
-
DORIS_TRY(_native_reader->materialize_dictionary_values(matched_id_column,
_type));
- RETURN_IF_ERROR(append_non_null_dictionary_values(column,
std::move(matched_values)));
+ if (projected_column != nullptr) {
+ const auto* matched_id_column =
check_and_get_column<ColumnInt32>(*_matched_dictionary_ids);
+ DORIS_CHECK(matched_id_column != nullptr);
+
RETURN_IF_ERROR(_native_reader->append_dictionary_values(matched_id_column,
_type,
+
projected_column));
+ }
if (_profile.reader_select_rows != nullptr) {
COUNTER_UPDATE(_profile.reader_select_rows, selected_rows);
}
- update_reader_read_rows(cast_set<int64_t>(matched_ids.size()));
- update_reader_skip_rows(batch_rows -
cast_set<int64_t>(matched_ids.size()));
+ update_reader_read_rows(cast_set<int64_t>(survivor_count));
+ update_reader_skip_rows(batch_rows - cast_set<int64_t>(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 1d92735bfa1..97b92e522a7 100644
--- a/be/src/format_v2/parquet/reader/native_column_reader.h
+++ b/be/src/format_v2/parquet/reader/native_column_reader.h
@@ -86,7 +86,7 @@ public:
Status select_with_dictionary_filter(const SelectionVector& selection,
uint16_t selected_rows,
int64_t batch_rows,
const IColumn::Filter&
dictionary_filter,
- MutableColumnPtr& column,
IColumn::Filter* row_filter,
+ IColumn* projected_column,
IColumn::Filter* row_filter,
bool* used_filter) override;
Status select_with_fixed_width_filter(const SelectionVector& selection,
uint16_t selected_rows,
int64_t batch_rows, const
VExprSPtrs& conjuncts,
@@ -116,6 +116,12 @@ private:
const VExprSPtrs& conjuncts, int
column_id,
IColumn* projected_column,
IColumn::Filter* row_filter,
int64_t* rows_read, bool* used_filter);
+ Status 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);
void release_batch_scratch_if_needed();
int64_t sync_native_profile();
void record_page_fragments(int64_t page_fragments);
diff --git a/be/test/exprs/expr_zonemap_filter_test.cpp
b/be/test/exprs/expr_zonemap_filter_test.cpp
index 08c4d2af754..6e7e42f7baa 100644
--- a/be/test/exprs/expr_zonemap_filter_test.cpp
+++ b/be/test/exprs/expr_zonemap_filter_test.cpp
@@ -368,7 +368,7 @@ TEST(ExprZonemapFilterTest,
ComparisonZonemapHandlesNullAndUnsupportedInputs) {
EXPECT_EQ(1, pass_all_ctx.stats.unusable_zonemap_eval_count);
}
-TEST(ExprZonemapFilterTest, ComparisonDictionaryAndBloomUseEqualityLiterals) {
+TEST(ExprZonemapFilterTest,
ComparisonDictionarySupportsTypedRangesWhileBloomUsesEquality) {
auto type = int_type();
auto slot = make_slot(0, type);
FunctionComparison<EqualsOp, NameEquals> equals;
@@ -389,8 +389,37 @@ TEST(ExprZonemapFilterTest,
ComparisonDictionaryAndBloomUseEqualityLiterals) {
equals.evaluate_bloom_filter(bloom_ctx, {slot,
make_int_literal(3)}));
FunctionComparison<NotEqualsOp, NameNotEquals> not_equals;
- EXPECT_FALSE(not_equals.can_evaluate_dictionary_filter({slot,
make_int_literal(3)}));
+ FunctionComparison<LessOp, NameLess> less;
+ FunctionComparison<LessOrEqualsOp, NameLessOrEquals> less_equal;
+ FunctionComparison<GreaterOp, NameGreater> greater;
+ FunctionComparison<GreaterOrEqualsOp, NameGreaterOrEquals> greater_equal;
+ EXPECT_TRUE(not_equals.can_evaluate_dictionary_filter({slot,
make_int_literal(3)}));
+ EXPECT_TRUE(less.can_evaluate_dictionary_filter({slot,
make_int_literal(2)}));
+ EXPECT_TRUE(less_equal.can_evaluate_dictionary_filter({slot,
make_int_literal(1)}));
+ EXPECT_TRUE(greater.can_evaluate_dictionary_filter({slot,
make_int_literal(2)}));
+ EXPECT_TRUE(greater_equal.can_evaluate_dictionary_filter({slot,
make_int_literal(3)}));
+ EXPECT_EQ(ZoneMapFilterResult::kMayMatch,
+ less.evaluate_dictionary_filter(dictionary_ctx, {slot,
make_int_literal(2)}));
+ EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
+ less.evaluate_dictionary_filter(dictionary_ctx, {slot,
make_int_literal(1)}));
+ EXPECT_EQ(ZoneMapFilterResult::kMayMatch,
+ less.evaluate_dictionary_filter(dictionary_ctx,
{make_int_literal(2), slot}));
+ EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
+ less.evaluate_dictionary_filter(dictionary_ctx,
{make_int_literal(4), slot}));
EXPECT_FALSE(not_equals.can_evaluate_bloom_filter({slot,
make_int_literal(3)}));
+
+ auto string_type = std::make_shared<DataTypeString>();
+ auto string_slot = make_slot(0, string_type);
+ auto string_dictionary =
make_dictionary_context({Field::create_field<TYPE_STRING>("alpha"),
+
Field::create_field<TYPE_STRING>("charlie")},
+ string_type);
+ EXPECT_TRUE(less.can_evaluate_dictionary_filter({string_slot,
make_string_literal("bravo")}));
+ EXPECT_EQ(ZoneMapFilterResult::kMayMatch,
+ less.evaluate_dictionary_filter(string_dictionary,
+ {string_slot,
make_string_literal("bravo")}));
+ EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
+ greater.evaluate_dictionary_filter(string_dictionary,
+ {string_slot,
make_string_literal("delta")}));
}
TEST(ExprZonemapFilterTest,
DefaultFunctionForwardsDictionaryAndBloomEvaluation) {
diff --git a/be/test/format_v2/parquet/native_decoder_test.cpp
b/be/test/format_v2/parquet/native_decoder_test.cpp
index 9192a055f01..3dfad7f4abd 100644
--- a/be/test/format_v2/parquet/native_decoder_test.cpp
+++ b/be/test/format_v2/parquet/native_decoder_test.cpp
@@ -38,6 +38,7 @@
#include "core/data_type/data_type_string.h"
#include "core/data_type/data_type_struct.h"
#include "core/data_type/data_type_timestamptz.h"
+#include "core/data_type_serde/parquet_decode_source.h"
#include "core/data_type_serde/parquet_timestamp.h"
#include "exprs/vectorized_fn_call.h"
#include "exprs/vliteral.h"
@@ -1358,44 +1359,57 @@ TEST(ParquetV2NativeDecoderTest,
DictionaryProbeMaterializesTypedValuesOnlyOnce)
EXPECT_EQ(reader.dictionary_materialization_count_for_test(), 1);
EXPECT_EQ(assert_cast<const ColumnInt32&>(**matched_values).get_data(),
(ColumnInt32::Container {20, 10}));
+
+ auto nullable_output =
make_nullable(std::make_shared<DataTypeInt32>())->create_column();
+ ASSERT_TRUE(reader.append_dictionary_values(&assert_cast<const
ColumnInt32&>(*ids),
+ field.data_type,
nullable_output.get())
+ .ok());
+ const auto& nullable = assert_cast<const
ColumnNullable&>(*nullable_output);
+ EXPECT_EQ(assert_cast<const
ColumnInt32&>(nullable.get_nested_column()).get_data(),
+ (ColumnInt32::Container {20, 10}));
+ EXPECT_EQ(nullable.get_null_map_data(), (NullMap {0, 0}));
+ EXPECT_EQ(reader.dictionary_materialization_count_for_test(), 1);
}
TEST(ParquetV2NativeDecoderTest,
DictionaryRepeatedRunsGatherDirectlyIntoDestination) {
- const std::array<int32_t, 2> dictionary_values {10, 20};
- 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())
+ for (const auto encoding :
+ {tparquet::Encoding::RLE_DICTIONARY,
tparquet::Encoding::PLAIN_DICTIONARY}) {
+ const std::array<int32_t, 2> dictionary_values {10, 20};
+ 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, encoding,
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, 1);
- for (size_t row = 0; row < 64; ++row) {
- encoder.Put(1);
- }
- encoder.Flush();
- std::vector<uint8_t> payload(encoded_ids.size() + 1);
- payload[0] = 1;
- memcpy(payload.data() + 1, encoded_ids.data(), encoded_ids.size());
- Slice id_slice(payload.data(), payload.size());
- ASSERT_TRUE(decoder->set_data(&id_slice).ok());
-
- DataTypeInt32 type;
- auto output = type.create_column();
- ParquetMaterializationState state;
- ParquetDecodeContext context {.physical_type = ParquetPhysicalType::INT32,
- .encoding =
ParquetValueEncoding::DICTIONARY};
- ASSERT_TRUE(
- type.get_serde()->read_column_from_parquet(*output, *decoder,
context, 64, state).ok());
- EXPECT_EQ(state.dictionary_materialization_strategy,
- ParquetDictionaryMaterializationStrategy::DIRECT);
- ASSERT_EQ(output->size(), 64);
- for (size_t row = 0; row < output->size(); ++row) {
- EXPECT_EQ(assert_cast<const ColumnInt32&>(*output).get_element(row),
20);
+ faststring encoded_ids;
+ RleEncoder<uint32_t> encoder(&encoded_ids, 1);
+ for (size_t row = 0; row < 64; ++row) {
+ encoder.Put(1);
+ }
+ encoder.Flush();
+ std::vector<uint8_t> payload(encoded_ids.size() + 1);
+ payload[0] = 1;
+ memcpy(payload.data() + 1, encoded_ids.data(), encoded_ids.size());
+ Slice id_slice(payload.data(), payload.size());
+ ASSERT_TRUE(decoder->set_data(&id_slice).ok());
+
+ DataTypeInt32 type;
+ auto output = type.create_column();
+ ParquetMaterializationState state;
+ ParquetDecodeContext context {.physical_type =
ParquetPhysicalType::INT32,
+ .encoding =
ParquetValueEncoding::DICTIONARY};
+ ASSERT_TRUE(type.get_serde()
+ ->read_column_from_parquet(*output, *decoder,
context, 64, state)
+ .ok());
+ EXPECT_EQ(state.dictionary_materialization_strategy,
+ ParquetDictionaryMaterializationStrategy::DIRECT);
+ ASSERT_EQ(output->size(), 64);
+ for (size_t row = 0; row < output->size(); ++row) {
+ EXPECT_EQ(assert_cast<const
ColumnInt32&>(*output).get_element(row), 20);
+ }
}
}
@@ -4073,6 +4087,25 @@ TEST(ParquetV2NativeDecoderTest,
FixedLengthStringsAppendAsOneContiguousSpan) {
EXPECT_EQ(column.get_data_at(2).to_string_view(), "ccc");
}
+TEST(ParquetV2NativeDecoderTest,
DictionaryStringGatherAppendsCompactSurvivors) {
+ ColumnString dictionary;
+ dictionary.insert_data("alpha", 5);
+ dictionary.insert_data("bravo", 5);
+ dictionary.insert_data("charlie", 7);
+ dictionary.insert_data("delta", 5);
+ ColumnString destination;
+ destination.insert_data("prefix", 6);
+ const std::array<uint32_t, 3> indices {3, 1, 2};
+
+ ASSERT_TRUE(try_simd_insert_parquet_dictionary_indices(destination,
dictionary, indices.data(),
+ indices.size()));
+ ASSERT_EQ(destination.size(), 4);
+ EXPECT_EQ(destination.get_data_at(0).to_string_view(), "prefix");
+ EXPECT_EQ(destination.get_data_at(1).to_string_view(), "delta");
+ EXPECT_EQ(destination.get_data_at(2).to_string_view(), "bravo");
+ EXPECT_EQ(destination.get_data_at(3).to_string_view(), "charlie");
+}
+
TEST(ParquetV2NativeDecoderTest,
ComplexPageStatisticsPreservePerLeafCrossings) {
ColumnChunkReaderStatistics first_chunk;
first_chunk.page_read_counter = 1;
diff --git a/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
b/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
index 7cc0838e125..89e1f56ac3e 100644
--- a/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
+++ b/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp
@@ -112,7 +112,7 @@ TEST(ParquetBenchmarkScenariosTest,
KernelMatrixCoversEverySimdStageAndBoundaryS
TEST(ParquetBenchmarkScenariosTest,
ReaderMatrixCoversNullableSparseAndProjectionAxes) {
const auto scenarios = reader_scenarios();
// Keep the exact count aligned with the upstream complex-residual
scenario retained by rebase.
- EXPECT_EQ(scenarios.size(), size_t {152});
+ EXPECT_EQ(scenarios.size(), size_t {167});
for (const int null_percent : {0, 1, 10, 50, 90}) {
for (const auto pattern : {Pattern::CLUSTERED, Pattern::ALTERNATING}) {
for (const int selectivity : {0, 1, 10, 50, 90, 100}) {
@@ -162,7 +162,7 @@ TEST(ParquetBenchmarkScenariosTest,
ReaderMatrixCoversOperationsEncodingsAndSche
TEST(ParquetBenchmarkScenariosTest,
ReaderMatrixHasExactUniqueRegistrationNames) {
const auto scenarios = reader_scenarios();
- EXPECT_EQ(scenarios.size(), size_t {152});
+ EXPECT_EQ(scenarios.size(), size_t {167});
std::set<std::string> names;
for (const auto& scenario : scenarios) {
@@ -199,6 +199,24 @@ TEST(ParquetBenchmarkScenariosTest,
ReaderMatrixCoversFixedWidthRawFilterAxes) {
}
}
+TEST(ParquetBenchmarkScenariosTest,
ReaderMatrixCoversTypedDictionaryFilterAxes) {
+ const auto scenarios = reader_scenarios();
+ for (const auto value_type : {ValueType::INT64, ValueType::BYTE_ARRAY}) {
+ for (const int selectivity : {10, 50}) {
+ for (const auto projection :
+ {Projection::PREDICATE_ONLY,
Projection::PREDICATE_PROJECTED}) {
+ EXPECT_TRUE(std::ranges::any_of(scenarios, [&](const
ReaderScenario& scenario) {
+ return scenario.operation ==
ReaderOperation::PREDICATE_SCAN &&
+ scenario.encoding == Encoding::DICTIONARY &&
+ scenario.value_type == value_type &&
+ scenario.selectivity_percent == selectivity &&
+ scenario.projection == projection;
+ })) << "missing typed dictionary filter axis";
+ }
+ }
+ }
+}
+
TEST(ParquetBenchmarkScenariosTest,
SelectionPlanDistinguishesClusteredAndSparseRuns) {
const auto clustered = make_selection_plan(1000, 10, Pattern::CLUSTERED);
EXPECT_EQ(clustered.total_rows, 1000);
diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp
b/be/test/format_v2/parquet/parquet_scan_test.cpp
index a4e5abb91a4..11cfdf99641 100644
--- a/be/test/format_v2/parquet/parquet_scan_test.cpp
+++ b/be/test/format_v2/parquet/parquet_scan_test.cpp
@@ -393,7 +393,8 @@ VExprContextSPtr create_int32_zonemap_conjunct(int
column_id, Int32ZoneMapExpr::
}
VExprContextSPtr create_int32_function_conjunct(int column_id, const
std::string& function_name,
- TExprOpcode::type opcode,
int32_t value) {
+ TExprOpcode::type opcode,
int32_t value,
+ bool mark_prepared = true) {
const auto int_type = std::make_shared<DataTypeInt32>();
const auto nullable_int_type = make_nullable(int_type);
const auto result_type = make_nullable(std::make_shared<DataTypeUInt8>());
@@ -419,11 +420,75 @@ VExprContextSPtr create_int32_function_conjunct(int
column_id, const std::string
auto context = VExprContext::create_shared(std::move(root));
// Direct evaluation does not execute the expression, but a fallback must
still fail this test
// instead of silently using an unprepared test-only context.
- context->_prepared = true;
- context->_opened = true;
+ if (mark_prepared) {
+ context->_prepared = true;
+ context->_opened = true;
+ }
return context;
}
+VExprContextSPtr create_int64_function_conjunct(int column_id, const
std::string& function_name,
+ TExprOpcode::type opcode,
int64_t value) {
+ const auto bigint_type = std::make_shared<DataTypeInt64>();
+ const auto nullable_bigint_type = make_nullable(bigint_type);
+ const auto result_type = make_nullable(std::make_shared<DataTypeUInt8>());
+ TFunctionName fn_name;
+ fn_name.__set_function_name(function_name);
+ TFunction fn;
+ fn.__set_name(fn_name);
+ fn.__set_binary_type(TFunctionBinaryType::BUILTIN);
+ fn.__set_arg_types({nullable_bigint_type->to_thrift(),
bigint_type->to_thrift()});
+ fn.__set_ret_type(result_type->to_thrift());
+ fn.__set_has_var_args(false);
+ TExprNode node;
+ node.__set_node_type(TExprNodeType::BINARY_PRED);
+ node.__set_opcode(opcode);
+ node.__set_type(result_type->to_thrift());
+ node.__set_fn(fn);
+ node.__set_num_children(2);
+ node.__set_is_nullable(true);
+ auto root = VectorizedFnCall::create_shared(node);
+ root->add_child(
+ VSlotRef::create_shared(column_id, column_id, -1,
nullable_bigint_type, "dict_bigint"));
+ root->add_child(VLiteral::create_shared(bigint_type,
Field::create_field<TYPE_BIGINT>(value)));
+ return VExprContext::create_shared(std::move(root));
+}
+
+VExprContextSPtr create_string_function_conjunct(int column_id, const
std::string& function_name,
+ TExprOpcode::type opcode,
const std::string& value,
+ bool literal_on_left = false)
{
+ const DataTypePtr string_type = std::make_shared<DataTypeString>();
+ const auto nullable_string_type = make_nullable(string_type);
+ const auto result_type = make_nullable(std::make_shared<DataTypeUInt8>());
+ TFunctionName fn_name;
+ fn_name.__set_function_name(function_name);
+ TFunction fn;
+ fn.__set_name(fn_name);
+ fn.__set_binary_type(TFunctionBinaryType::BUILTIN);
+ fn.__set_arg_types({nullable_string_type->to_thrift(),
string_type->to_thrift()});
+ fn.__set_ret_type(result_type->to_thrift());
+ fn.__set_has_var_args(false);
+ TExprNode node;
+ node.__set_node_type(TExprNodeType::BINARY_PRED);
+ node.__set_opcode(opcode);
+ node.__set_type(result_type->to_thrift());
+ node.__set_fn(fn);
+ node.__set_num_children(2);
+ node.__set_is_nullable(true);
+ auto root = VectorizedFnCall::create_shared(node);
+ auto slot =
+ VSlotRef::create_shared(column_id, column_id, -1,
nullable_string_type, "dict_text");
+ auto literal = VLiteral::create_shared(string_type,
Field::create_field<TYPE_STRING>(value));
+ if (literal_on_left) {
+ root->add_child(std::move(literal));
+ root->add_child(std::move(slot));
+ } else {
+ root->add_child(std::move(slot));
+ root->add_child(std::move(literal));
+ }
+ return VExprContext::create_shared(std::move(root));
+}
+
VExprContextSPtr create_int32_mod_greater_than_conjunct(int column_id) {
const auto int_type = std::make_shared<DataTypeInt32>();
const auto nullable_int_type = make_nullable(int_type);
@@ -561,6 +626,14 @@ std::shared_ptr<arrow::Array> build_int32_array(const
std::vector<int32_t>& valu
return finish_array(&builder);
}
+std::shared_ptr<arrow::Array> build_int64_array(const std::vector<int64_t>&
values) {
+ arrow::Int64Builder builder;
+ for (const auto value : values) {
+ EXPECT_TRUE(builder.Append(value).ok());
+ }
+ return finish_array(&builder);
+}
+
std::shared_ptr<arrow::Array> build_uint32_array(const std::vector<uint32_t>&
values) {
arrow::UInt32Builder builder;
for (const auto value : values) {
@@ -692,6 +765,37 @@ void write_int_pair_parquet_file(const std::string&
file_path, int64_t row_group
write_table(file_path, table, row_group_size, false, false,
enable_statistics, encoding);
}
+void write_dictionary_int_pair_parquet_file(const std::string& file_path) {
+ auto schema = arrow::schema({
+ arrow::field("id", arrow::int32(), false),
+ arrow::field("score", arrow::int32(), false),
+ });
+ auto table = arrow::Table::Make(schema, {build_int32_array({1, 2, 3, 4, 5,
6}),
+ build_int32_array({10, 20, 30,
40, 50, 60})});
+ write_table(file_path, table, 6, true, false, false);
+}
+
+void write_dictionary_bigint_pair_parquet_file(const std::string& file_path) {
+ auto schema = arrow::schema({
+ arrow::field("id", arrow::int64(), false),
+ arrow::field("score", arrow::int32(), false),
+ });
+ auto table = arrow::Table::Make(schema, {build_int64_array({1, 2, 3, 4, 5,
6}),
+ build_int32_array({10, 20, 30,
40, 50, 60})});
+ write_table(file_path, table, 6, true, false, false);
+}
+
+void write_dictionary_string_pair_parquet_file(const std::string& file_path) {
+ auto schema = arrow::schema({
+ arrow::field("text", arrow::utf8(), false),
+ arrow::field("score", arrow::int32(), false),
+ });
+ auto table =
+ arrow::Table::Make(schema, {build_string_array({"alpha", "bravo",
"charlie", "delta"}),
+ build_int32_array({10, 20, 30, 40})});
+ write_table(file_path, table, 4, true, false, false);
+}
+
void write_int_triple_parquet_file(const std::string& file_path) {
auto schema = arrow::schema({
arrow::field("left", arrow::int32(), false),
@@ -1780,6 +1884,185 @@ TEST_F(ParquetScanTest,
PredicateOnlyPlainComparisonUsesPhysicalDirectPath) {
EXPECT_EQ(counter_value(profile, "PredicateCompactionBytes"), 0);
}
+TEST_F(ParquetScanTest,
PredicateOnlyDictionaryRangeSkipsTypedValueMaterialization) {
+ 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));
+ auto conjunct = create_int32_function_conjunct(0, "gt", TExprOpcode::GT,
2, false);
+ ASSERT_TRUE(conjunct->prepare(&state, RowDescriptor()).ok());
+ ASSERT_TRUE(conjunct->open(&state).ok());
+ request->conjuncts.push_back(conjunct);
+ ASSERT_TRUE(reader->open(request).ok());
+
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ ASSERT_EQ(rows, 4);
+ EXPECT_EQ(int32_data_column(*block.get_by_position(1).column).get_data(),
+ (ColumnInt32::Container {30, 40, 50, 60}));
+ EXPECT_EQ(block.get_by_position(0).column->size(), rows);
+ EXPECT_EQ(counter_value(profile, "DictFilterCandidateColumns"), 1);
+ EXPECT_EQ(counter_value(profile, "DictFilterColumns"), 1);
+ EXPECT_EQ(counter_value(profile, "RowsFilteredByDictFilter"), 2);
+ EXPECT_EQ(counter_value(profile, "DictionaryPredicateDirectBatches"), 1);
+ EXPECT_EQ(counter_value(profile, "DictionaryPredicateDirectRows"), 6);
+ EXPECT_EQ(counter_value(profile, "DictionaryPredicateProjectedRows"), 0);
+ EXPECT_EQ(counter_value(profile, "PredicateCompactionCount"), 0);
+ conjunct->close();
+}
+
+TEST_F(ParquetScanTest, ProjectedDictionaryRangeGathersOnlySurvivors) {
+ 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());
+ auto conjunct = create_int32_function_conjunct(0, "gt", TExprOpcode::GT,
2, false);
+ ASSERT_TRUE(conjunct->prepare(&state, RowDescriptor()).ok());
+ ASSERT_TRUE(conjunct->open(&state).ok());
+ request->conjuncts.push_back(conjunct);
+ ASSERT_TRUE(reader->open(request).ok());
+
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ ASSERT_EQ(rows, 4);
+ EXPECT_EQ(int32_data_column(*block.get_by_position(0).column).get_data(),
+ (ColumnInt32::Container {3, 4, 5, 6}));
+ EXPECT_EQ(int32_data_column(*block.get_by_position(1).column).get_data(),
+ (ColumnInt32::Container {30, 40, 50, 60}));
+ EXPECT_EQ(counter_value(profile, "DictFilterColumns"), 1);
+ EXPECT_EQ(counter_value(profile, "DictionaryPredicateDirectBatches"), 1);
+ EXPECT_EQ(counter_value(profile, "DictionaryPredicateDirectRows"), 6);
+ EXPECT_EQ(counter_value(profile, "DictionaryPredicateProjectedRows"), 4);
+ EXPECT_EQ(counter_value(profile, "PredicateCompactionCount"), 0);
+ conjunct->close();
+}
+
+TEST_F(ParquetScanTest, ProjectedBigIntDictionaryRangeUsesFusedGather) {
+ write_dictionary_bigint_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());
+ auto conjunct = create_int64_function_conjunct(0, "gt", TExprOpcode::GT,
2);
+ ASSERT_TRUE(conjunct->prepare(&state, RowDescriptor()).ok());
+ ASSERT_TRUE(conjunct->open(&state).ok());
+ request->conjuncts.push_back(conjunct);
+ ASSERT_TRUE(reader->open(request).ok());
+
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ ASSERT_EQ(rows, 4);
+ EXPECT_EQ(int64_data_column(*block.get_by_position(0).column).get_data(),
+ (ColumnInt64::Container {3, 4, 5, 6}));
+ EXPECT_EQ(int32_data_column(*block.get_by_position(1).column).get_data(),
+ (ColumnInt32::Container {30, 40, 50, 60}));
+ auto* fused_rows =
profile.get_counter("DictionaryPredicateFusedProjectedRows");
+ ASSERT_NE(fused_rows, nullptr);
+ EXPECT_EQ(fused_rows->value(), 4);
+ auto* typed_filter_columns =
profile.get_counter("DictFilterTypedCompareColumns");
+ ASSERT_NE(typed_filter_columns, nullptr);
+ EXPECT_EQ(typed_filter_columns->value(), 1);
+ conjunct->close();
+}
+
+TEST_F(ParquetScanTest, ProjectedStringDictionaryRangeGathersOnlySurvivors) {
+ 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());
+ auto conjunct = create_string_function_conjunct(0, "gt", TExprOpcode::GT,
"bravo");
+ ASSERT_TRUE(conjunct->prepare(&state, RowDescriptor()).ok());
+ ASSERT_TRUE(conjunct->open(&state).ok());
+ request->conjuncts.push_back(conjunct);
+ 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);
+ const auto& text = string_data_column(*block.get_by_position(0).column);
+ EXPECT_EQ(text.get_data_at(0).to_string(), "charlie");
+ EXPECT_EQ(text.get_data_at(1).to_string(), "delta");
+ EXPECT_EQ(int32_data_column(*block.get_by_position(1).column).get_data(),
+ (ColumnInt32::Container {30, 40}));
+ EXPECT_EQ(counter_value(profile, "DictionaryPredicateProjectedRows"), 2);
+ EXPECT_EQ(counter_value(profile, "PredicateCompactionCount"), 0);
+ auto* string_filter_columns =
profile.get_counter("DictFilterStringCompareColumns");
+ ASSERT_NE(string_filter_columns, nullptr);
+ EXPECT_EQ(string_filter_columns->value(), 1);
+ conjunct->close();
+}
+
+TEST_F(ParquetScanTest, StringDictionaryRangeNormalizesLiteralOnLeft) {
+ 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());
+ auto conjunct = create_string_function_conjunct(0, "lt", TExprOpcode::LT,
"bravo", true);
+ ASSERT_TRUE(conjunct->prepare(&state, RowDescriptor()).ok());
+ ASSERT_TRUE(conjunct->open(&state).ok());
+ request->conjuncts.push_back(conjunct);
+ 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);
+ const auto& text = string_data_column(*block.get_by_position(0).column);
+ EXPECT_EQ(text.get_data_at(0).to_string_view(), "charlie");
+ EXPECT_EQ(text.get_data_at(1).to_string_view(), "delta");
+ EXPECT_EQ(counter_value(profile, "DictFilterStringCompareColumns"), 1);
+ conjunct->close();
+}
+
TEST_F(ParquetScanTest,
ProjectedPlainComparisonUsesPhysicalFilterAndProjectPath) {
write_int_pair_parquet_file(_file_path, 6, false);
RuntimeProfile profile("profile");
diff --git a/docs/file-scanner-v2-parquet-scan-design.md
b/docs/file-scanner-v2-parquet-scan-design.md
index 70c041a89cc..b81a779ac5e 100644
--- a/docs/file-scanner-v2-parquet-scan-design.md
+++ b/docs/file-scanner-v2-parquet-scan-design.md
@@ -333,11 +333,19 @@ flowchart LR
B -- "Yes" --> C[Read Dictionary Page]
C --> D[Evaluate Predicate on Dictionary Values<br>Build Dictionary-ID
Bitmap]
D --> E[Decode Data-page Dictionary IDs<br>Update SelectionVector Directly]
- E --> G[Materialize Only Survivors]
+ E --> G{Predicate Column Projected?}
+ G -- "No" --> H[Keep Row Shape Without Typed Values]
+ G -- "Yes" --> I[Gather Survivors Directly Into Target Column]
```
-- Applies to non-repeated primitive, string-like BYTE_ARRAY /
FIXED_LEN_BYTE_ARRAY columns whose
- complete Column Chunk uses dictionary data encoding.
+- Applies to compatible equality and range predicates on non-repeated
primitive columns whose
+ complete Column Chunk uses `RLE_DICTIONARY` or legacy `PLAIN_DICTIONARY`
data encoding.
+- Predicate-only batches decode selected IDs straight from the page decoder
and never materialize
+ a row-sized predicate value column. The typed dictionary is cached once per
generation. Simple
+ numeric comparisons build its ID bitmap over contiguous typed values, while
string equality and
+ range comparisons operate directly on dictionary slices. When projection is
required, all
+ fixed-width survivors are written by the filtering loop itself; strings
pre-size their character
+ and offset buffers and copy each compact survivor once.
- Safe AND subexpressions may remove components exactly covered by dictionary
evaluation. OR or
non-equivalent expressions are not rewritten aggressively.
- Stateful, potentially throwing, or whole-batch-sensitive expressions disable
staged
@@ -593,7 +601,9 @@ a dictionary encoding. The predicate is evaluated against
the current dictionary
entry bitmap; decoded IDs are checked against both dictionary length and
bitmap length. A mixed
dictionary/plain transition falls back before consuming data. Once selected
dictionary reading has
advanced a page cursor, loss of dictionary output is corruption rather than a
retry through another
-path with shifted state.
+path with shifted state. Predicate-only slots stop after decoder-level ID
filtering and retain only
+the output row shape. Projected slots write fixed-width survivors in the
filtering loop or gather
+compact string survivors directly into pre-sized target buffers.
Missing optional indexes or unsupported predicate/type combinations retain
rows. Malformed
offsets, inconsistent page counts, out-of-range dictionary IDs,
overlapping/unsorted invalid ranges,
@@ -608,11 +618,11 @@ storage indexes for external Parquet files.
| Capability | Granularity | Suitable predicates | Result property | Main
limitations |
| --- | --- | --- | --- | --- |
| Footer Statistics / ZoneMap | Row Group | Ranges, comparisons, IS NULL/IS
NOT NULL, and expressions safely convertible to ZoneMap | Can prove the entire
group cannot match | Requires valid min/max/null_count and safe type conversion
|
-| Dictionary Pruning | Row Group | Single-column predicates exactly evaluable
over the dictionary domain | Can prove the entire group cannot match |
Low-cardinality string-like primitive with complete dictionary encoding |
+| Dictionary Pruning | Row Group | Single-column predicates exactly evaluable
over the dictionary domain | Can prove the entire group cannot match |
Compatible primitive with complete dictionary encoding |
| Parquet Bloom Filter | Row Group / Column Chunk | Equality and IN
membership-negation predicates | Negative result can prune; positive result
requires verification | Controlled by configuration; file must contain Bloom
data; false positives are possible |
| ColumnIndex | Page | Predicates evaluable from min/max/null | Produces
candidate pages and row ranges | Requires an index and decodable compatible
types |
| OffsetIndex | Page → Row Range | Does not evaluate predicates directly |
Maps page results to row numbers and physical skip plans | Normally used with
ColumnIndex |
-| Dictionary-ID Filter | Row / Batch | Safe single-column string-like
predicates | Exact filtering of actual rows | Complete dictionary encoding and
non-repeated primitive only |
+| Dictionary-ID Filter | Row / Batch | Safe typed equality and range
predicates | Exact filtering of actual rows without a full predicate value
column | Complete dictionary encoding and non-repeated primitive only |
| Condition Cache Bitmap | File-global granule | Stable cacheable conditions |
Reuses previous filtering to reduce row ranges | Not a native Parquet index;
uncovered ranges remain candidates |
### Index-selection overview
@@ -866,7 +876,8 @@ flowchart TD
| --- | --- |
| Row Group pruning | How many total Row Groups were pruned by
Statistics/Dictionary/Bloom, and how much time did each stage take? |
| Page index pruning | How many indexes were checked, pages/rows were pruned,
ranges selected, and pages skipped? |
-| Dictionary row filter | How often were predicates rewritten, dictionaries
read, bitmaps built, and attempts successful or rejected? |
+| Dictionary row filter | How often were predicates rewritten, dictionaries
read, and bitmaps built? `DictFilterTypedCompareColumns` and
`DictFilterStringCompareColumns` distinguish typed kernels from the generic
expression fallback. |
+| Dictionary direct predicate | How many batches and input rows were filtered
through dictionary IDs, and how many survivor values were projected? Inspect
`DictionaryPredicateDirectBatches/Rows`, `DictionaryPredicateProjectedRows`,
and `DictionaryPredicateFusedProjectedRows`. |
| Predicate / raw rows | How many rows were read and rejected, and was lazy
materialization worthwhile? |
| Predicate compaction | Did selection-first evaluation avoid repeated
movement? Inspect `PredicateCompactionTime/Bytes/Count`; single-column rounds
retain row mappings and compact at multi-column/delete/output boundaries. |
| PLAIN direct predicate | How many eligible predicate-only physical batches
and input rows bypassed Doris-column materialization? Inspect
`PlainPredicateDirectBatches/Rows`. |
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
