Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 48 additions & 10 deletions be/src/storage/segment/segment_iterator.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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];
Expand Down Expand Up @@ -2153,6 +2155,16 @@ bool SegmentIterator::_can_evaluated_by_vectorized(std::shared_ptr<ColumnPredica
}
}

const IColumn* SegmentIterator::_get_predicate_column(const ColumnPredicate& predicate) const {
const auto column_id = predicate.column_id();
const auto& column = _current_return_columns[column_id];
if (column->is_nullable() &&
_predicate_column_null_states[column_id] == PredicateColumnNullState::NO_NULLS) {
return &assert_cast<const ColumnNullable&>(*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
Expand Down Expand Up @@ -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)) {
Expand All @@ -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;
}
Expand Down Expand Up @@ -2379,21 +2393,26 @@ 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);
RETURN_IF_ERROR(column_iter->seek_to_ordinal(_block_rowids[0]));
} 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<ErrorCode::INTERNAL_ERROR>("nrows({}) != rows_read({})",
nrows_read, rows_read);
}
} 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) &&
Expand All @@ -2403,24 +2422,30 @@ 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);
RETURN_IF_ERROR(column_iter->seek_to_ordinal(_block_rowids[processed]));
} 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<ErrorCode::INTERNAL_ERROR>(
"batch nrows({}) != rows_read({})", current_batch_size, rows_read);
}
} 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;
}
}

Expand Down Expand Up @@ -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());
}
}

Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -2868,14 +2891,29 @@ Status SegmentIterator::_convert_to_expected_type(const std::vector<ColumnId>& 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()
<< " select_size: " << select_size;
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<const ColumnNullable&>(*input_col_ptr);
auto& output_nullable = assert_cast<ColumnNullable&>(*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());
}

Expand Down
23 changes: 16 additions & 7 deletions be/src/storage/segment/segment_iterator.h
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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 <class Container>
[[nodiscard]] Status _output_column_by_sel_idx(Block* block, const Container& column_ids,
Expand All @@ -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<ColumnPredicate> 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,
Expand Down Expand Up @@ -385,6 +393,7 @@ class SegmentIterator : public RowwiseIterator {
std::map<uint32_t, bool> _need_read_data_indices;
std::vector<bool> _is_common_expr_column;
MutableColumns _current_return_columns;
std::vector<PredicateColumnNullState> _predicate_column_null_states;
std::vector<std::shared_ptr<ColumnPredicate>> _pre_eval_block_predicate;
std::vector<std::shared_ptr<ColumnPredicate>> _short_cir_eval_predicate;
std::vector<uint32_t> _delete_range_column_ids;
Expand Down
Original file line number Diff line number Diff line change
@@ -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 <gtest/gtest.h>

#include <memory>
#include <vector>

#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<int32_t>& values,
const std::vector<uint8_t>& 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<TabletSchema>();
tablet_schema->init_from_pb(schema_pb);
return tablet_schema;
}

SchemaSPtr make_read_schema(const TabletSchemaSPtr& tablet_schema) {
return std::make_shared<Schema>(tablet_schema->columns(), std::vector<ColumnId> {0});
}

void expect_nullable_int_column(const MutableColumnPtr& column,
const std::vector<int32_t>& expected_values,
const std::vector<uint8_t>& expected_null_map) {
const auto& nullable = assert_cast<const ColumnNullable&>(*column);
const auto& nested = assert_cast<const ColumnInt32&>(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<SegmentIterator> make_iter() {
return std::make_unique<SegmentIterator>(nullptr, _read_schema);
}

TabletSchemaSPtr _tablet_schema;
SchemaSPtr _read_schema;
};

TEST_F(SegmentIteratorPredicateNullStateTest, UsesNestedColumnOnlyWhenNoNullsAreKnown) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please exercise the production state producer in this test file. All three tests assign _predicate_column_null_states directly, so a regression in _read_columns_by_index—for example, a NULL-containing later chunk incorrectly remaining NO_NULLS, or every real no-NULL batch remaining conservative—would still pass even though that state controls predicate evaluation and null-map elision. Add a SegmentIterator test backed by real nullable segment data that covers no-NULL and NULL-containing batches plus continuous and rowid-fallback/discontinuous reads, and asserts returned rows/null maps for representative vectorized and short-circuit predicates.

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<const ColumnNullable&>(*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<const ColumnNullable&>(*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<int>(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<const ColumnNullable&>(*output).has_null());
expect_nullable_int_column(output, {20, 30}, {1, 0});
}
}

} // namespace doris::segment_v2
Loading