From 6d43c886ee5d0681772ff57912fa267962a6ba80 Mon Sep 17 00:00:00 2001 From: Mryange Date: Mon, 17 Aug 2026 11:32:43 +0800 Subject: [PATCH] [opt](exec) Skip null maps for predicate batches without nulls ### What problem does this PR solve? Issue Number: N/A Problem Summary: Nullable predicate columns always process their null maps even when a scanned batch is known to contain no NULL values. Track the batch null state so predicate evaluation can use the nested column and selector copying can rebuild an all-zero null map only for the nullable output. ### Release note None ### Check List (For Author) - Test: Unit Test added but not run - Added BE unit tests for NO_NULLS, HAS_NULLS, and UNKNOWN states - clang-format-16 and git diff --check passed - Behavior changed: No - Does this need documentation: No --- be/src/storage/segment/segment_iterator.cpp | 58 +++++-- be/src/storage/segment/segment_iterator.h | 23 ++- ...ent_iterator_predicate_null_state_test.cpp | 158 ++++++++++++++++++ 3 files changed, 222 insertions(+), 17 deletions(-) create mode 100644 be/test/storage/segment/segment_iterator_predicate_null_state_test.cpp diff --git a/be/src/storage/segment/segment_iterator.cpp b/be/src/storage/segment/segment_iterator.cpp index e7bf020536021c..a91adacc43d75d 100644 --- a/be/src/storage/segment/segment_iterator.cpp +++ b/be/src/storage/segment/segment_iterator.cpp @@ -566,6 +566,8 @@ Status SegmentIterator::_lazy_init(Block* block) { _block_rowids.resize(_initial_block_row_max); } _current_return_columns.resize(_schema->columns().size()); + _predicate_column_null_states.assign(_schema->columns().size(), + PredicateColumnNullState::UNKNOWN); for (size_t i = 0; i < _schema->column_ids().size(); i++) { ColumnId cid = _schema->column_ids()[i]; @@ -2153,6 +2155,16 @@ bool SegmentIterator::_can_evaluated_by_vectorized(std::shared_ptris_nullable() && + _predicate_column_null_states[column_id] == PredicateColumnNullState::NO_NULLS) { + return &assert_cast(*column).get_nested_column(); + } + return column.get(); +} + // These placeholders are used only when the real column data is skipped after // index/count pushdown has already identified the matching rows. The value is // irrelevant, but nullable columns must stay non-NULL so COUNT(col) can count @@ -2344,6 +2356,7 @@ Status SegmentIterator::_read_columns_by_index(uint32_t nrows_read_limit, uint16 nrows_read > 0 ? _block_rowids[nrows_read - 1] : 0); for (auto cid : _predicate_column_ids) { auto& column = _current_return_columns[cid]; + _predicate_column_null_states[cid] = PredicateColumnNullState::UNKNOWN; VLOG_DEBUG << fmt::format("Reading column {}, col_name {}", cid, _schema->column(cid)->name()); if (!_virtual_column_exprs.contains(cid)) { @@ -2352,6 +2365,7 @@ Status SegmentIterator::_read_columns_by_index(uint32_t nrows_read_limit, uint16 continue; } if (_prune_column(cid, column, nrows_read)) { + _predicate_column_null_states[cid] = PredicateColumnNullState::NO_NULLS; VLOG_DEBUG << fmt::format("Column {} is pruned. No need to read data.", cid); continue; } @@ -2379,6 +2393,7 @@ Status SegmentIterator::_read_columns_by_index(uint32_t nrows_read_limit, uint16 if (is_continuous) { size_t rows_read = nrows_read; + bool batch_has_null = true; _opts.stats->predicate_column_read_seek_num += 1; if (_opts.runtime_state && _opts.runtime_state->enable_profile()) { SCOPED_RAW_TIMER(&_opts.stats->predicate_column_read_seek_ns); @@ -2386,7 +2401,10 @@ Status SegmentIterator::_read_columns_by_index(uint32_t nrows_read_limit, uint16 } else { RETURN_IF_ERROR(column_iter->seek_to_ordinal(_block_rowids[0])); } - RETURN_IF_ERROR(column_iter->next_batch(&rows_read, column)); + RETURN_IF_ERROR(column_iter->next_batch(&rows_read, column, &batch_has_null)); + _predicate_column_null_states[cid] = batch_has_null + ? PredicateColumnNullState::HAS_NULLS + : PredicateColumnNullState::NO_NULLS; if (rows_read != nrows_read) { return Status::Error("nrows({}) != rows_read({})", nrows_read, rows_read); @@ -2394,6 +2412,7 @@ Status SegmentIterator::_read_columns_by_index(uint32_t nrows_read_limit, uint16 } else { const uint32_t batch_size = _range_iter->get_batch_size(); uint32_t processed = 0; + auto column_null_state = PredicateColumnNullState::NO_NULLS; while (processed < nrows_read) { uint32_t current_batch_size = std::min(batch_size, nrows_read - processed); bool batch_continuous = (current_batch_size > 1) && @@ -2403,6 +2422,7 @@ Status SegmentIterator::_read_columns_by_index(uint32_t nrows_read_limit, uint16 if (batch_continuous) { size_t rows_read = current_batch_size; + bool batch_has_null = true; _opts.stats->predicate_column_read_seek_num += 1; if (_opts.runtime_state && _opts.runtime_state->enable_profile()) { SCOPED_RAW_TIMER(&_opts.stats->predicate_column_read_seek_ns); @@ -2410,7 +2430,10 @@ Status SegmentIterator::_read_columns_by_index(uint32_t nrows_read_limit, uint16 } else { RETURN_IF_ERROR(column_iter->seek_to_ordinal(_block_rowids[processed])); } - RETURN_IF_ERROR(column_iter->next_batch(&rows_read, column)); + RETURN_IF_ERROR(column_iter->next_batch(&rows_read, column, &batch_has_null)); + if (batch_has_null && column_null_state != PredicateColumnNullState::UNKNOWN) { + column_null_state = PredicateColumnNullState::HAS_NULLS; + } if (rows_read != current_batch_size) { return Status::Error( "batch nrows({}) != rows_read({})", current_batch_size, rows_read); @@ -2418,9 +2441,11 @@ Status SegmentIterator::_read_columns_by_index(uint32_t nrows_read_limit, uint16 } else { RETURN_IF_ERROR(column_iter->read_by_rowids(&_block_rowids[processed], current_batch_size, column)); + column_null_state = PredicateColumnNullState::UNKNOWN; } processed += current_batch_size; } + _predicate_column_null_states[cid] = column_null_state; } } @@ -2543,13 +2568,12 @@ uint16_t SegmentIterator::_evaluate_vectorization_predicate(uint16_t* sel_rowid_ if (pred->always_true()) { continue; } - auto column_id = pred->column_id(); - auto& column = _current_return_columns[column_id]; + const auto* predicate_column = _get_predicate_column(*pred); if (is_first) { - pred->evaluate_vec(*column, original_size, (bool*)_ret_flags.data()); + pred->evaluate_vec(*predicate_column, original_size, (bool*)_ret_flags.data()); is_first = false; } else { - pred->evaluate_and_vec(*column, original_size, (bool*)_ret_flags.data()); + pred->evaluate_and_vec(*predicate_column, original_size, (bool*)_ret_flags.data()); } } @@ -2598,9 +2622,8 @@ uint16_t SegmentIterator::_evaluate_short_circuit_predicate(uint16_t* vec_sel_ro uint16_t original_size = selected_size; for (auto predicate : _short_cir_eval_predicate) { - auto column_id = predicate->column_id(); - auto& short_cir_column = _current_return_columns[column_id]; - selected_size = predicate->evaluate(*short_cir_column, vec_sel_rowid_idx, selected_size); + selected_size = predicate->evaluate(*_get_predicate_column(*predicate), vec_sel_rowid_idx, + selected_size); } _opts.stats->short_circuit_cond_input_rows += original_size; @@ -2868,7 +2891,8 @@ Status SegmentIterator::_convert_to_expected_type(const std::vector& c Status SegmentIterator::copy_column_data_by_selector(IColumn* input_col_ptr, MutableColumnPtr& output_col, uint16_t* sel_rowid_idx, uint16_t select_size, - size_t batch_size) { + size_t batch_size, + PredicateColumnNullState input_null_state) { if (is_column_nullable(*output_col) != is_column_nullable(*input_col_ptr)) { LOG(WARNING) << "nullable mismatch for output_column: " << output_col->dump_structure() << " input_column: " << input_col_ptr->dump_structure() @@ -2876,6 +2900,20 @@ Status SegmentIterator::copy_column_data_by_selector(IColumn* input_col_ptr, return Status::RuntimeError("copy_column_data_by_selector nullable mismatch"); } output_col->reserve(select_size); + // Predicate evaluation and selector copying are independent stages. Once the selected rowids + // are known, an input batch proven to have no NULLs can be copied without filtering its + // all-zero null map. + if (input_col_ptr->is_nullable() && input_null_state == PredicateColumnNullState::NO_NULLS) { + const auto& input_nullable = assert_cast(*input_col_ptr); + auto& output_nullable = assert_cast(*output_col); + RETURN_IF_ERROR(input_nullable.get_nested_column().filter_by_selector( + sel_rowid_idx, select_size, output_nullable.get_nested_column_ptr().get())); + auto& output_null_map = output_nullable.get_null_map_data(); + DCHECK(output_null_map.empty()); + // Preserve the nullable output type while reconstructing the known all-zero null map. + output_null_map.resize_fill(select_size, 0); + return Status::OK(); + } return input_col_ptr->filter_by_selector(sel_rowid_idx, select_size, output_col.get()); } diff --git a/be/src/storage/segment/segment_iterator.h b/be/src/storage/segment/segment_iterator.h index b67361d53ffa7a..967060ff8324bd 100644 --- a/be/src/storage/segment/segment_iterator.h +++ b/be/src/storage/segment/segment_iterator.h @@ -149,6 +149,12 @@ class SegmentIterator : public RowwiseIterator { } private: + enum class PredicateColumnNullState : uint8_t { + UNKNOWN, + NO_NULLS, + HAS_NULLS, + }; + Status _next_batch_internal(Block* block); Status _check_output_block(Block* block); @@ -235,7 +241,8 @@ class SegmentIterator : public RowwiseIterator { Status copy_column_data_by_selector(IColumn* input_col_ptr, MutableColumnPtr& output_col, uint16_t* sel_rowid_idx, uint16_t select_size, - size_t batch_size); + size_t batch_size, + PredicateColumnNullState input_null_state); template [[nodiscard]] Status _output_column_by_sel_idx(Block* block, const Container& column_ids, @@ -257,24 +264,25 @@ class SegmentIterator : public RowwiseIterator { if (storage_type && !storage_type->equals(*block->get_by_position(block_cid).type)) { // Do additional cast MutableColumnPtr tmp = storage_type->create_column(); - RETURN_IF_ERROR(copy_column_data_by_selector(_current_return_columns[cid].get(), - tmp, sel_rowid_idx, select_size, - _opts.block_row_max)); + RETURN_IF_ERROR(copy_column_data_by_selector( + _current_return_columns[cid].get(), tmp, sel_rowid_idx, select_size, + _opts.block_row_max, _predicate_column_null_states[cid])); RETURN_IF_ERROR(variant_util::cast_column( {tmp->get_ptr(), storage_type, ""}, block->get_by_position(block_cid).type, &block->get_by_position(block_cid).column)); } else { MutableColumnPtr output_column = block->get_by_position(block_cid).column->assert_mutable(); - RETURN_IF_ERROR(copy_column_data_by_selector(_current_return_columns[cid].get(), - output_column, sel_rowid_idx, - select_size, _opts.block_row_max)); + RETURN_IF_ERROR(copy_column_data_by_selector( + _current_return_columns[cid].get(), output_column, sel_rowid_idx, + select_size, _opts.block_row_max, _predicate_column_null_states[cid])); } } return Status::OK(); } bool _can_evaluated_by_vectorized(std::shared_ptr predicate); + const IColumn* _get_predicate_column(const ColumnPredicate& predicate) const; [[nodiscard]] Status _extract_common_expr_columns(const VExprSPtr& expr); [[nodiscard]] Status _execute_common_expr(uint16_t* sel_rowid_idx, uint16_t& selected_size, @@ -385,6 +393,7 @@ class SegmentIterator : public RowwiseIterator { std::map _need_read_data_indices; std::vector _is_common_expr_column; MutableColumns _current_return_columns; + std::vector _predicate_column_null_states; std::vector> _pre_eval_block_predicate; std::vector> _short_cir_eval_predicate; std::vector _delete_range_column_ids; diff --git a/be/test/storage/segment/segment_iterator_predicate_null_state_test.cpp b/be/test/storage/segment/segment_iterator_predicate_null_state_test.cpp new file mode 100644 index 00000000000000..e391e63bf23ccd --- /dev/null +++ b/be/test/storage/segment/segment_iterator_predicate_null_state_test.cpp @@ -0,0 +1,158 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include + +#include +#include + +#include "core/assert_cast.h" +#include "core/column/column_nullable.h" +#include "core/column/column_vector.h" +#include "storage/olap_common.h" +#include "storage/predicate/null_predicate.h" +#include "storage/tablet/tablet_schema.h" + +#if defined(__clang__) +#pragma clang diagnostic push +#pragma clang diagnostic ignored "-Wkeyword-macro" +#endif +#include "storage/segment/segment_iterator.h" +#if defined(__clang__) +#pragma clang diagnostic pop +#endif + +namespace doris::segment_v2 { +namespace { + +MutableColumnPtr make_nullable_int_column(const std::vector& values, + const std::vector& null_map) { + auto nested = ColumnInt32::create(); + auto nulls = ColumnUInt8::create(); + for (auto value : values) { + nested->insert_value(value); + } + for (auto is_null : null_map) { + nulls->insert_value(is_null); + } + return ColumnNullable::create(std::move(nested), std::move(nulls)); +} + +TabletSchemaSPtr make_nullable_int_schema() { + TabletSchemaPB schema_pb; + schema_pb.set_keys_type(KeysType::DUP_KEYS); + auto* column = schema_pb.add_column(); + column->set_unique_id(0); + column->set_name("c0"); + column->set_type("INT"); + column->set_is_key(true); + column->set_is_nullable(true); + + auto tablet_schema = std::make_shared(); + tablet_schema->init_from_pb(schema_pb); + return tablet_schema; +} + +SchemaSPtr make_read_schema(const TabletSchemaSPtr& tablet_schema) { + return std::make_shared(tablet_schema->columns(), std::vector {0}); +} + +void expect_nullable_int_column(const MutableColumnPtr& column, + const std::vector& expected_values, + const std::vector& expected_null_map) { + const auto& nullable = assert_cast(*column); + const auto& nested = assert_cast(nullable.get_nested_column()); + ASSERT_EQ(expected_values.size(), nested.size()); + ASSERT_EQ(expected_null_map.size(), nullable.get_null_map_data().size()); + for (size_t i = 0; i < expected_values.size(); ++i) { + EXPECT_EQ(expected_values[i], nested.get_data()[i]); + EXPECT_EQ(expected_null_map[i], nullable.get_null_map_data()[i]); + } +} + +} // namespace + +class SegmentIteratorPredicateNullStateTest : public ::testing::Test { +protected: + void SetUp() override { + _tablet_schema = make_nullable_int_schema(); + _read_schema = make_read_schema(_tablet_schema); + } + + std::unique_ptr make_iter() { + return std::make_unique(nullptr, _read_schema); + } + + TabletSchemaSPtr _tablet_schema; + SchemaSPtr _read_schema; +}; + +TEST_F(SegmentIteratorPredicateNullStateTest, UsesNestedColumnOnlyWhenNoNullsAreKnown) { + auto iter = make_iter(); + auto input = make_nullable_int_column({10, 20, 30}, {0, 0, 0}); + const auto* input_ptr = input.get(); + const auto* nested_ptr = &assert_cast(*input).get_nested_column(); + iter->_current_return_columns.emplace_back(std::move(input)); + iter->_predicate_column_null_states.resize(1); + auto predicate = NullPredicate::create_shared(0, "c0", true, PrimitiveType::TYPE_INT); + + iter->_predicate_column_null_states[0] = SegmentIterator::PredicateColumnNullState::NO_NULLS; + EXPECT_EQ(nested_ptr, iter->_get_predicate_column(*predicate)); + + iter->_predicate_column_null_states[0] = SegmentIterator::PredicateColumnNullState::HAS_NULLS; + EXPECT_EQ(input_ptr, iter->_get_predicate_column(*predicate)); + + iter->_predicate_column_null_states[0] = SegmentIterator::PredicateColumnNullState::UNKNOWN; + EXPECT_EQ(input_ptr, iter->_get_predicate_column(*predicate)); +} + +TEST_F(SegmentIteratorPredicateNullStateTest, CopiesNullableColumnWithoutFilteringNullMap) { + auto iter = make_iter(); + auto input = make_nullable_int_column({10, 20, 30}, {0, 0, 0}); + MutableColumnPtr output = ColumnNullable::create(ColumnInt32::create(), ColumnUInt8::create()); + uint16_t selector[] = {2, 0}; + + auto status = + iter->copy_column_data_by_selector(input.get(), output, selector, 2, 3, + SegmentIterator::PredicateColumnNullState::NO_NULLS); + + ASSERT_TRUE(status.ok()) << status.to_string(); + ASSERT_TRUE(output->is_nullable()); + EXPECT_FALSE(assert_cast(*output).has_null()); + expect_nullable_int_column(output, {30, 10}, {0, 0}); +} + +TEST_F(SegmentIteratorPredicateNullStateTest, PreservesNullsForKnownAndUnknownStates) { + for (auto null_state : {SegmentIterator::PredicateColumnNullState::HAS_NULLS, + SegmentIterator::PredicateColumnNullState::UNKNOWN}) { + SCOPED_TRACE(static_cast(null_state)); + auto iter = make_iter(); + auto input = make_nullable_int_column({10, 20, 30}, {0, 1, 0}); + MutableColumnPtr output = + ColumnNullable::create(ColumnInt32::create(), ColumnUInt8::create()); + uint16_t selector[] = {1, 2}; + + auto status = + iter->copy_column_data_by_selector(input.get(), output, selector, 2, 3, null_state); + + ASSERT_TRUE(status.ok()) << status.to_string(); + EXPECT_TRUE(assert_cast(*output).has_null()); + expect_nullable_int_column(output, {20, 30}, {1, 0}); + } +} + +} // namespace doris::segment_v2