[improvement](segment) Skip nullable row ID gaps in sparse reads

be/src/storage/segment/column_reader.cpp

@@ -2758,16 +2758,28 @@ 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<uint32_t>(_page.offset_in_page + _page.first_ordinal);
+                    size_t gap =
+                            rowids[offset] > current_ordinal_in_page
+                                    ? std::min<size_t>(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)) {
                         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<uint32_t>(_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<uint32_t>(_page.offset_in_page + _page.first_ordinal);
+                size_t gap = rowids[offset] > current_ordinal_in_page
+                                     ? std::min<size_t>(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<uint32_t>(_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) {

be/src/util/rle_encoding.h

@@ -428,6 +428,41 @@ size_t RleDecoder<T>::Skip(size_t to_skip) {
     return set_count;
 }
 
+template <>
+inline size_t RleDecoder<bool>::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<int>(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 <typename T>

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 <gtest/gtest.h>
+
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <string_view>
+#include <vector>
+
+#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<ColumnReader> 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<int32_t> values(num_rows);
+    std::vector<uint8_t> null_map(num_rows, 0);
+    for (size_t i = 0; i < num_rows; ++i) {
+        values[i] = 10000 + static_cast<int32_t>(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<int32_t>(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<ColumnWriter> 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<ColumnReader> reader;
+    ASSERT_TRUE(ColumnReader::create(reader_options, meta, num_rows, file_reader, &reader).ok());
+
+    std::vector<rowid_t> sparse_rowids = {1, 6, 17, 24, 31, 48, 63, 64, 79, 96, 111, 127, 145, 159};
+    std::vector<rowid_t> 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<FileColumnIterator*>(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<const ColumnNullable&>(*result);
+        const auto& nested = assert_cast<const ColumnInt32&>(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<rowid_t>& 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<const ColumnNullable&>(*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

be/test/util/rle_encoding_test.cpp

@@ -419,6 +419,78 @@ TEST_F(TestRle, TestSkip) {
     encoder.Flush();
 }
 
+TEST_F(TestRle, TestBoolSkipPreservesLiteralOffsetAcrossRuns) {
+    std::vector<bool> 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<bool> encoder(&buffer, 1);
+    for (bool value : values) {
+        encoder.Put(value);
+    }
+    encoder.Flush();
+
+    const std::vector<size_t> skip_counts = {1, 3, 7, 9, 17, 23, 25, 39, 41, 63, 77};
+    for (size_t skip_count : skip_counts) {
+        RleDecoder<bool> 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<bool> 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<bool> encoder(&buffer, 1);
+    for (bool value : values) {
+        encoder.Put(value);
+    }
+    encoder.Flush();
+
+    RleDecoder<bool> 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 <typename T>
 void ValidatePutRunLength(const std::vector<std::pair<T, size_t>>& runs, int bit_width) {