branch-4.2 [improvement](meta path) Port nested column pruning / meta path series

be/src/core/column/column_nullable.h

@@ -190,6 +190,14 @@ class ColumnNullable final : public COWHelper<IColumn, ColumnNullable> {
         get_nested_column().insert_many_continuous_binary_data(data, offsets, num);
     }
 
+    void insert_offsets_from_lengths(const uint32_t* lengths, size_t num) override {
+        if (UNLIKELY(num == 0)) {
+            return;
+        }
+        push_false_to_nullmap(num);
+        get_nested_column().insert_offsets_from_lengths(lengths, num);
+    }
+
     // Default value in `ColumnNullable` is null
     void insert_default() override {
         get_nested_column().insert_default();

be/src/storage/segment/binary_dict_page.cpp

@@ -29,6 +29,7 @@
 #include "common/logging.h"
 #include "common/status.h"
 #include "core/column/column.h"
+#include "core/column/column_string.h"
 #include "storage/segment/binary_plain_page_v2.h"
 #include "storage/segment/bitshuffle_page.h"
 #include "storage/segment/encoding_info.h"
@@ -280,11 +281,28 @@ Status BinaryDictPageDecoder::next_batch(size_t* n, MutableColumnPtr& dst) {
                                                         _bit_shuffle_ptr->_cur_index));
     *n = max_fetch;
 
-    const auto* data_array = reinterpret_cast<const int32_t*>(_bit_shuffle_ptr->get_data(0));
-    size_t start_index = _bit_shuffle_ptr->_cur_index;
+    if (_options.only_read_offsets) {
+        // OFFSET_ONLY mode: resolve dict codes to get real string lengths
+        // without copying actual char data. This allows length() to work.
+        const auto* data_array = reinterpret_cast<const int32_t*>(_bit_shuffle_ptr->get_data(0));
+        size_t start_index = _bit_shuffle_ptr->_cur_index;
+        // Reuse _buffer (int32_t vector) to store uint32_t lengths.
+        // int32_t and uint32_t have the same size/alignment, and string
+        // lengths are always non-negative, so the bit patterns are identical.
+        _buffer.resize(max_fetch);
+        for (size_t i = 0; i < max_fetch; ++i) {
+            int32_t codeword = data_array[start_index + i];
+            _buffer[i] = static_cast<int32_t>(_dict_word_info[codeword].size);
+        }
+        dst->insert_offsets_from_lengths(reinterpret_cast<const uint32_t*>(_buffer.data()),
+                                         max_fetch);
+    } else {
+        const auto* data_array = reinterpret_cast<const int32_t*>(_bit_shuffle_ptr->get_data(0));
+        size_t start_index = _bit_shuffle_ptr->_cur_index;
 
-    dst->insert_many_dict_data(data_array, start_index, _dict_word_info, max_fetch,
-                               _num_dict_items);
+        dst->insert_many_dict_data(data_array, start_index, _dict_word_info, max_fetch,
+                                   _num_dict_items);
+    }
 
     _bit_shuffle_ptr->_cur_index += max_fetch;
 
@@ -305,8 +323,32 @@ Status BinaryDictPageDecoder::read_by_rowids(const rowid_t* rowids, ordinal_t pa
         return Status::OK();
     }
 
-    const auto* data_array = reinterpret_cast<const int32_t*>(_bit_shuffle_ptr->get_data(0));
     auto total = *n;
+
+    if (_options.only_read_offsets) {
+        // OFFSET_ONLY mode: resolve dict codes to get real string lengths
+        // without copying actual char data. This allows length() to work correctly.
+        const auto* data_array = reinterpret_cast<const int32_t*>(_bit_shuffle_ptr->get_data(0));
+        size_t read_count = 0;
+        _buffer.resize(total);
+        for (size_t i = 0; i < total; ++i) {
+            ordinal_t ord = rowids[i] - page_first_ordinal;
+            if (ord >= _bit_shuffle_ptr->_num_elements) [[unlikely]] {
+                break;
+            }
+            int32_t codeword = data_array[ord];
+            _buffer[read_count] = static_cast<int32_t>(_dict_word_info[codeword].size);
+            read_count++;
+        }
+        if (read_count > 0) {
+            dst->insert_offsets_from_lengths(reinterpret_cast<const uint32_t*>(_buffer.data()),
+                                             read_count);
+        }
+        *n = read_count;
+        return Status::OK();
+    }
+
+    const auto* data_array = reinterpret_cast<const int32_t*>(_bit_shuffle_ptr->get_data(0));
     size_t read_count = 0;
     _buffer.resize(total);
     for (size_t i = 0; i < total; ++i) {

be/src/storage/segment/binary_plain_page.h

@@ -30,7 +30,6 @@
 
 #include "common/logging.h"
 #include "core/column/column_complex.h"
-#include "core/column/column_nullable.h"
 #include "storage/olap_common.h"
 #include "storage/segment/options.h"
 #include "storage/segment/page_builder.h"
@@ -202,6 +201,21 @@ class BinaryPlainPageDecoder : public PageDecoder {
         }
         const size_t max_fetch = std::min(*n, static_cast<size_t>(_num_elems - _cur_idx));
 
+        if (_options.only_read_offsets) {
+            // OFFSET_ONLY mode: read string lengths from page offset trailer
+            // without copying actual char data. This allows length() to work.
+            _offsets.resize(max_fetch);
+            for (size_t i = 0; i < max_fetch; ++i) {
+                uint32_t str_start = offset(_cur_idx + i);
+                uint32_t str_end = offset(_cur_idx + i + 1);
+                _offsets[i] = str_end - str_start;
+            }
+            dst->insert_offsets_from_lengths(_offsets.data(), max_fetch);
+            _cur_idx += max_fetch;
+            *n = max_fetch;
+            return Status::OK();
+        }
+
         uint32_t last_offset = guarded_offset(_cur_idx);
         _offsets.resize(max_fetch + 1);
         _offsets[0] = last_offset;
@@ -237,6 +251,29 @@ class BinaryPlainPageDecoder : public PageDecoder {
         }
 
         auto total = *n;
+
+        if (_options.only_read_offsets) {
+            // OFFSET_ONLY mode: read string lengths from page offset trailer
+            // without copying actual char data. This allows length() to work.
+            size_t read_count = 0;
+            _offsets.resize(total);
+            for (size_t i = 0; i < total; ++i) {
+                ordinal_t ord = rowids[i] - page_first_ordinal;
+                if (UNLIKELY(ord >= _num_elems)) {
+                    break;
+                }
+                uint32_t str_start = offset(ord);
+                uint32_t str_end = offset(ord + 1);
+                _offsets[read_count] = str_end - str_start;
+                read_count++;
+            }
+            if (read_count > 0) {
+                dst->insert_offsets_from_lengths(_offsets.data(), read_count);
+            }
+            *n = read_count;
+            return Status::OK();
+        }
+
         size_t read_count = 0;
         _binary_data.resize(total);
         for (size_t i = 0; i < total; ++i) {