From 5b32913a519585871a9bebf50d821551674e67ca Mon Sep 17 00:00:00 2001 From: yangtao555 Date: Thu, 10 Sep 2026 17:55:18 +0800 Subject: [PATCH] [improvement](segment) Skip nullable row ID gaps in sparse reads --- be/src/storage/segment/column_reader.cpp | 34 +++- be/src/util/rle_encoding.h | 35 ++++ .../segment/nullable_column_reader_test.cpp | 166 ++++++++++++++++++ be/test/util/rle_encoding_test.cpp | 72 ++++++++ 4 files changed, 303 insertions(+), 4 deletions(-) create mode 100644 be/test/storage/segment/nullable_column_reader_test.cpp diff --git a/be/src/storage/segment/column_reader.cpp b/be/src/storage/segment/column_reader.cpp index 125209a19cfb33..4af5002618c29a 100644 --- a/be/src/storage/segment/column_reader.cpp +++ b/be/src/storage/segment/column_reader.cpp @@ -2758,6 +2758,19 @@ Status FileColumnIterator::read_by_rowids(const rowid_t* rowids, const size_t co if (_page.has_null) { size_t already_read = 0; while ((nrows_to_read - already_read) > 0) { + size_t offset = total_read_count + already_read; + rowid_t current_ordinal_in_page = + cast_set(_page.offset_in_page + _page.first_ordinal); + size_t gap = + rowids[offset] > current_ordinal_in_page + ? std::min(rowids[offset] - current_ordinal_in_page, + _page.remaining()) + : 0; + if (gap > 0) { + _page.null_decoder.Skip(gap); + _page.offset_in_page += gap; + } + bool is_null = false; size_t this_run = std::min(nrows_to_read - already_read, _page.remaining()); if (UNLIKELY(this_run == 0)) { @@ -2765,9 +2778,8 @@ Status FileColumnIterator::read_by_rowids(const rowid_t* rowids, const size_t co } this_run = _page.null_decoder.GetNextRun(&is_null, this_run); - size_t offset = total_read_count + already_read; size_t this_read_count = 0; - rowid_t current_ordinal_in_page = + current_ordinal_in_page = cast_set(_page.offset_in_page + _page.first_ordinal); for (size_t i = 0; i < this_run; ++i) { if (rowids[offset + i] - current_ordinal_in_page >= this_run) { @@ -2824,15 +2836,29 @@ Status FileColumnIterator::read_by_rowids(const rowid_t* rowids, const size_t co if (_page.has_null) { size_t already_read = 0; while ((nrows_to_read - already_read) > 0) { + size_t offset = total_read_count + already_read; + rowid_t current_ordinal_in_page = + cast_set(_page.offset_in_page + _page.first_ordinal); + size_t gap = rowids[offset] > current_ordinal_in_page + ? std::min(rowids[offset] - current_ordinal_in_page, + _page.remaining()) + : 0; + if (gap > 0) { + auto origin_index = _page.data_decoder->current_index(); + size_t null_count = _page.null_decoder.Skip(gap); + RETURN_IF_ERROR(_page.data_decoder->seek_to_position_in_page(origin_index + + gap - null_count)); + _page.offset_in_page += gap; + } + bool is_null = false; size_t this_run = std::min(nrows_to_read - already_read, _page.remaining()); if (UNLIKELY(this_run == 0)) { break; } this_run = _page.null_decoder.GetNextRun(&is_null, this_run); - size_t offset = total_read_count + already_read; size_t this_read_count = 0; - rowid_t current_ordinal_in_page = + current_ordinal_in_page = cast_set(_page.offset_in_page + _page.first_ordinal); for (size_t i = 0; i < this_run; ++i) { if (rowids[offset + i] - current_ordinal_in_page >= this_run) { diff --git a/be/src/util/rle_encoding.h b/be/src/util/rle_encoding.h index f96d00e7941127..e1bb59aaef6263 100644 --- a/be/src/util/rle_encoding.h +++ b/be/src/util/rle_encoding.h @@ -428,6 +428,41 @@ size_t RleDecoder::Skip(size_t to_skip) { return set_count; } +template <> +inline size_t RleDecoder::Skip(size_t to_skip) { + DCHECK(bit_reader_.is_initialized()); + DCHECK_EQ(bit_width_, 1); + + size_t set_count = 0; + while (to_skip > 0) { + bool result = ReadHeader(); + DCHECK(result); + + if (repeat_count_ > 0) [[likely]] { + size_t nskip = (repeat_count_ < to_skip) ? repeat_count_ : to_skip; + repeat_count_ -= nskip; + to_skip -= nskip; + if (current_value_ != 0) { + set_count += nskip; + } + } else { + DCHECK(literal_count_ > 0); + size_t nskip = (literal_count_ < to_skip) ? literal_count_ : to_skip; + literal_count_ -= nskip; + to_skip -= nskip; + while (nskip > 0) { + uint64_t values = 0; + size_t bit_count = std::min(nskip, sizeof(values) * 8); + bool result1 = bit_reader_.GetValue(cast_set(bit_count), &values); + DCHECK(result1); + set_count += BitUtil::popcount(values); + nskip -= bit_count; + } + } + } + return set_count; +} + // This function buffers input values 8 at a time. After seeing all 8 values, // it decides whether they should be encoded as a literal or repeated run. template diff --git a/be/test/storage/segment/nullable_column_reader_test.cpp b/be/test/storage/segment/nullable_column_reader_test.cpp new file mode 100644 index 00000000000000..a11061a5a8de48 --- /dev/null +++ b/be/test/storage/segment/nullable_column_reader_test.cpp @@ -0,0 +1,166 @@ +// 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 +#include +#include + +#include "core/column/column_nullable.h" +#include "core/column/column_vector.h" +#include "io/fs/file_writer.h" +#include "io/fs/local_file_system.h" +#include "storage/olap_common.h" +#include "storage/segment/column_reader.h" +#include "storage/segment/column_writer.h" +#include "storage/tablet/tablet_schema.h" + +namespace doris::segment_v2 { + +namespace { +class NullMapOnlyTestFileColumnIterator final : public FileColumnIterator { +public: + explicit NullMapOnlyTestFileColumnIterator(std::shared_ptr reader) + : FileColumnIterator(std::move(reader)) {} + + void force_null_map_only() { _meta_read_mode = MetaReadMode::NULL_MAP_ONLY; } +}; +} // namespace + +class NullableColumnReaderTest : public testing::Test { +protected: + void SetUp() override { + ASSERT_TRUE(io::global_local_filesystem()->delete_directory(TEST_DIR).ok()); + ASSERT_TRUE(io::global_local_filesystem()->create_directory(TEST_DIR).ok()); + } + + void TearDown() override { + EXPECT_TRUE(io::global_local_filesystem()->delete_directory(TEST_DIR).ok()); + } + + static constexpr std::string_view TEST_DIR = "./ut_dir/nullable_column_reader_test"; +}; + +TEST_F(NullableColumnReaderTest, ReadByRowidsKeepsNullAndDataOrdinalsAligned) { + constexpr size_t num_rows = 160; + std::vector values(num_rows); + std::vector null_map(num_rows, 0); + for (size_t i = 0; i < num_rows; ++i) { + values[i] = 10000 + static_cast(i) * 37; + const bool is_null = (i < 23 && i % 3 == 0) || (i >= 23 && i < 55) || + (i >= 79 && i < 111 && (i / 2) % 2 == 0) || (i >= 111 && i < 140) || + (i >= 140 && i % 2 == 0); + null_map[i] = is_null; + } + + auto fs = io::global_local_filesystem(); + const std::string file_path = std::string(TEST_DIR) + "/nullable_int"; + io::FileWriterPtr file_writer; + ASSERT_TRUE(fs->create_file(file_path, &file_writer).ok()); + + ColumnMetaPB meta; + meta.set_column_id(0); + meta.set_unique_id(0); + meta.set_type(static_cast(FieldType::OLAP_FIELD_TYPE_INT)); + meta.set_length(sizeof(int32_t)); + meta.set_encoding(EncodingTypePB::BIT_SHUFFLE); + meta.set_compression(CompressionTypePB::LZ4F); + meta.set_is_nullable(true); + + ColumnWriterOptions writer_options; + writer_options.meta = &meta; + TabletColumn tablet_column(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, + FieldType::OLAP_FIELD_TYPE_INT, true); + std::unique_ptr writer; + ASSERT_TRUE( + ColumnWriter::create(writer_options, &tablet_column, file_writer.get(), &writer).ok()); + ASSERT_TRUE(writer->init().ok()); + ASSERT_TRUE(writer->append(null_map.data(), values.data(), num_rows).ok()); + ASSERT_TRUE(writer->finish().ok()); + ASSERT_TRUE(writer->write_data().ok()); + ASSERT_TRUE(writer->write_ordinal_index().ok()); + ASSERT_TRUE(file_writer->close().ok()); + + io::FileReaderSPtr file_reader; + ASSERT_TRUE(fs->open_file(file_path, &file_reader).ok()); + ColumnReaderOptions reader_options; + std::shared_ptr reader; + ASSERT_TRUE(ColumnReader::create(reader_options, meta, num_rows, file_reader, &reader).ok()); + + std::vector sparse_rowids = {1, 6, 17, 24, 31, 48, 63, 64, 79, 96, 111, 127, 145, 159}; + std::vector dense_rowids; + for (rowid_t rowid = 18; rowid <= 120; ++rowid) { + dense_rowids.push_back(rowid); + } + + for (const auto& rowids : {sparse_rowids, dense_rowids}) { + ColumnIteratorUPtr iterator; + ASSERT_TRUE(reader->new_iterator(&iterator, &tablet_column).ok()); + ASSERT_NE(dynamic_cast(iterator.get()), nullptr); + + ColumnIteratorOptions iterator_options; + OlapReaderStatistics stats; + iterator_options.file_reader = file_reader.get(); + iterator_options.stats = &stats; + ASSERT_TRUE(iterator->init(iterator_options).ok()); + + MutableColumnPtr result = + ColumnNullable::create(ColumnInt32::create(), ColumnUInt8::create()); + ASSERT_TRUE(iterator->read_by_rowids(rowids.data(), rowids.size(), result).ok()); + ASSERT_EQ(result->size(), rowids.size()); + + const auto& nullable = assert_cast(*result); + const auto& nested = assert_cast(nullable.get_nested_column()); + for (size_t i = 0; i < rowids.size(); ++i) { + const rowid_t rowid = rowids[i]; + EXPECT_EQ(nullable.is_null_at(i), null_map[rowid] != 0) << "rowid=" << rowid; + if (!nullable.is_null_at(i)) { + EXPECT_EQ(nested.get_data()[i], values[rowid]) << "rowid=" << rowid; + } + } + } + + NullMapOnlyTestFileColumnIterator null_map_iterator(reader); + ColumnIteratorOptions iterator_options; + OlapReaderStatistics stats; + iterator_options.file_reader = file_reader.get(); + iterator_options.stats = &stats; + ASSERT_TRUE(null_map_iterator.init(iterator_options).ok()); + null_map_iterator.force_null_map_only(); + + auto read_and_check_null_map = [&](const std::vector& rowids) { + MutableColumnPtr result = + ColumnNullable::create(ColumnInt32::create(), ColumnUInt8::create()); + ASSERT_TRUE(null_map_iterator.read_by_rowids(rowids.data(), rowids.size(), result).ok()); + ASSERT_EQ(result->size(), rowids.size()); + + const auto& nullable = assert_cast(*result); + for (size_t i = 0; i < rowids.size(); ++i) { + EXPECT_EQ(nullable.is_null_at(i), null_map[rowids[i]] != 0) << "rowid=" << rowids[i]; + } + }; + + read_and_check_null_map({1, 96, 145}); + ASSERT_TRUE(null_map_iterator.get_current_page()->contains(159)); + read_and_check_null_map({159}); + read_and_check_null_map({6, 24}); +} + +} // namespace doris::segment_v2 diff --git a/be/test/util/rle_encoding_test.cpp b/be/test/util/rle_encoding_test.cpp index 5af34168869f6d..321ea85546f9f1 100644 --- a/be/test/util/rle_encoding_test.cpp +++ b/be/test/util/rle_encoding_test.cpp @@ -419,6 +419,78 @@ TEST_F(TestRle, TestSkip) { encoder.Flush(); } +TEST_F(TestRle, TestBoolSkipPreservesLiteralOffsetAcrossRuns) { + std::vector values; + for (int i = 0; i < 24; ++i) { + values.push_back(i % 3 == 1); + } + values.insert(values.end(), 16, true); + for (int i = 0; i < 24; ++i) { + values.push_back((i / 2) % 2 == 0); + } + values.insert(values.end(), 12, false); + for (int i = 0; i < 17; ++i) { + values.push_back(i % 2 == 0); + } + + faststring buffer; + RleEncoder encoder(&buffer, 1); + for (bool value : values) { + encoder.Put(value); + } + encoder.Flush(); + + const std::vector skip_counts = {1, 3, 7, 9, 17, 23, 25, 39, 41, 63, 77}; + for (size_t skip_count : skip_counts) { + RleDecoder decoder(buffer.data(), buffer.size(), 1); + EXPECT_EQ(std::count(values.begin(), values.begin() + skip_count, true), + decoder.Skip(skip_count)) + << "skip_count=" << skip_count; + + for (size_t i = skip_count; i < values.size(); ++i) { + bool decoded = false; + ASSERT_TRUE(decoder.Get(&decoded)) << "skip_count=" << skip_count << ", index=" << i; + EXPECT_EQ(values[i], decoded) << "skip_count=" << skip_count << ", index=" << i; + } + } +} + +TEST_F(TestRle, TestBoolSkipReadsMultipleChunksFromUnalignedLiteral) { + constexpr size_t literal_count = 192; + std::vector values; + values.reserve(literal_count + 16); + for (size_t i = 0; i < literal_count; ++i) { + values.push_back(i % 2 == 0); + } + values.insert(values.end(), 16, true); + + faststring buffer; + RleEncoder encoder(&buffer, 1); + for (bool value : values) { + encoder.Put(value); + } + encoder.Flush(); + + RleDecoder decoder(buffer.data(), buffer.size(), 1); + constexpr size_t prefix_count = 3; + for (size_t i = 0; i < prefix_count; ++i) { + bool decoded = false; + ASSERT_TRUE(decoder.Get(&decoded)); + EXPECT_EQ(values[i], decoded); + } + + constexpr size_t skip_count = 129; + EXPECT_EQ(std::count(values.begin() + prefix_count, values.begin() + prefix_count + skip_count, + true), + decoder.Skip(skip_count)); + + for (size_t i = prefix_count + skip_count; i < values.size(); ++i) { + bool decoded = false; + ASSERT_TRUE(decoder.Get(&decoded)) << "index=" << i; + EXPECT_EQ(values[i], decoded) << "index=" << i; + } +} + // Helper to compare Put with run_length vs multiple Put(value) calls template void ValidatePutRunLength(const std::vector>& runs, int bit_width) {