enhance: eliminate redundant timestamp copy in sealed segments (#47689)

issue: #47872

Replace `ConcurrentVector<Timestamp>` with a lightweight `TimestampData`
class that pins column chunks directly (zero-copy) for StorageV2, and
falls back to owned `std::vector` for StorageV1.

---------

Signed-off-by: Shawn Wang <shawn.wang@zilliz.com>

internal/core/src/segcore/ChunkedSegmentSealedImpl.cpp

@@ -489,28 +489,9 @@ ChunkedSegmentSealedImpl::load_column_group_data_internal(
                 for (auto& [_, info] : load_info.field_infos) {
                     num_rows = info.row_count;
                 }
-                auto all_ts_chunks =
-                    timestamp_proxy_column->GetAllChunks(nullptr);
-                std::vector<Timestamp> timestamps(num_rows);
-                int64_t offset = 0;
-                for (int i = 0; i < all_ts_chunks.size(); i++) {
-                    auto chunk_data = all_ts_chunks[i].get();
-                    auto fixed_chunk =
-                        static_cast<FixedWidthChunk*>(chunk_data);
-                    auto span = fixed_chunk->Span();
-                    for (size_t j = 0; j < span.row_count(); j++) {
-                        auto ts = *(int64_t*)((char*)span.data() +
-                                              j * span.element_sizeof());
-                        timestamps[offset++] = ts;
-                    }
-                }
-                init_timestamp_index(timestamps, num_rows);
+                init_timestamp_index_from_column(timestamp_proxy_column,
+                                                 num_rows);
                 system_ready_count_++;
-                AssertInfo(offset == num_rows,
-                           "[StorageV2] timestamp total row count {} not equal "
-                           "to expected {}",
-                           offset,
-                           num_rows);
             }
         }
 
@@ -646,7 +627,7 @@ ChunkedSegmentSealedImpl::load_system_field_internal(
             offset += chunk_ptr->Span().row_count();
         }
 
-        init_timestamp_index(timestamps, num_rows);
+        init_timestamp_index_owned(std::move(timestamps), num_rows);
         ++system_ready_count_;
     } else {
         AssertInfo(system_field_type == SystemFieldType::RowId,
@@ -1504,6 +1485,9 @@ ChunkedSegmentSealedImpl::ChunkedSegmentSealedImpl(
       ngram_fields_(std::unordered_set<FieldId>(schema->size())),
       scalar_indexings_(std::unordered_map<FieldId, index::CacheIndexBasePtr>(
           schema->size())),
+      mmap_descriptor_(storage::MmapManager::GetInstance()
+                           .GetMmapChunkManager()
+                           ->Register()),
       insert_record_(*schema, MAX_ROW_COUNT),
       segment_load_info_(milvus::proto::segcore::SegmentLoadInfo(), schema),
       schema_(schema),
@@ -1523,8 +1507,6 @@ ChunkedSegmentSealedImpl::ChunkedSegmentSealedImpl(
                   callback);
           },
           segment_id) {
-    auto mcm = storage::MmapManager::GetInstance().GetMmapChunkManager();
-    mmap_descriptor_ = mcm->Register();
 }
 
 ChunkedSegmentSealedImpl::~ChunkedSegmentSealedImpl() {
@@ -1553,17 +1535,21 @@ ChunkedSegmentSealedImpl::bulk_subscript(milvus::OpContext* op_ctx,
                "System field isn't ready when do bulk_insert, segID:{}",
                id_);
     switch (system_type) {
-        case SystemFieldType::Timestamp:
-            AssertInfo(insert_record_.timestamps_.num_chunk() == 1,
-                       "num chunk of timestamp not equal to 1 for "
-                       "sealed segment");
-            bulk_subscript_impl<Timestamp>(
-                op_ctx,
-                this->insert_record_.timestamps_.get_chunk_data(0),
-                seg_offsets,
-                count,
-                static_cast<Timestamp*>(output));
+        case SystemFieldType::Timestamp: {
+            auto* dst = static_cast<Timestamp*>(output);
+            auto& ts = insert_record_.timestamps_;
+            if (ts.num_chunks() == 1) {
+                auto* src = ts.chunk_data(0);
+                for (int64_t i = 0; i < count; ++i) {
+                    dst[i] = src[seg_offsets[i]];
+                }
+            } else {
+                for (int64_t i = 0; i < count; ++i) {
+                    dst[i] = ts[seg_offsets[i]];
+                }
+            }
             break;
+        }
         case SystemFieldType::RowId:
             ThrowInfo(ErrorCode::Unsupported, "RowId retrieve not supported");
             break;
@@ -2559,51 +2545,81 @@ ChunkedSegmentSealedImpl::get_active_count(Timestamp ts) const {
     return this->get_row_count();
 }
 
+// Helper: apply a per-element timestamp scan over a range [beg, end) on
+// segmented TimestampData, calling `pred(global_offset, ts_value)` for each.
+template <typename Pred>
+static void
+scan_timestamp_range(const TimestampData& ts,
+                     int64_t beg,
+                     int64_t end,
+                     Pred pred) {
+    for (int64_t c = 0; c < ts.num_chunks(); c++) {
+        auto chunk_start = ts.chunk_start_offset(c);
+        auto chunk_end = chunk_start + ts.chunk_row_count(c);
+        auto overlap_beg = std::max(beg, chunk_start);
+        auto overlap_end = std::min(end, chunk_end);
+        if (overlap_beg >= overlap_end) {
+            continue;
+        }
+        auto* data = ts.chunk_data(c);
+        auto local = overlap_beg - chunk_start;
+        for (int64_t i = overlap_beg; i < overlap_end; ++i, ++local) {
+            pred(i, data[local]);
+        }
+    }
+}
+
 void
 ChunkedSegmentSealedImpl::mask_with_timestamps(BitsetTypeView& bitset_chunk,
                                                Timestamp timestamp,
                                                Timestamp collection_ttl) const {
-    // TODO change the
-    AssertInfo(insert_record_.timestamps_.num_chunk() == 1,
-               "num chunk not equal to 1 for sealed segment");
-    auto timestamps_data =
-        (const milvus::Timestamp*)insert_record_.timestamps_.get_chunk_data(0);
-    auto timestamps_data_size = insert_record_.timestamps_.get_chunk_size(0);
+    auto& ts = insert_record_.timestamps_;
+    auto total_size = static_cast<int64_t>(ts.size());
+
     if (collection_ttl > 0) {
         auto range =
             insert_record_.timestamp_index_.get_active_range(collection_ttl);
-        if (range.first == range.second &&
-            range.first == timestamps_data_size) {
+        if (range.first == range.second && range.first == total_size) {
             bitset_chunk.set();
             return;
         } else {
-            auto ttl_mask = TimestampIndex::GenerateTTLBitset(
-                timestamps_data, timestamps_data_size, collection_ttl, range);
+            // TTL bitset: [0, beg) = true, [beg, end) = check, [end, size) = false
+            BitsetType ttl_mask;
+            ttl_mask.reserve(total_size);
+            ttl_mask.resize(range.first, true);
+            ttl_mask.resize(total_size, false);
+            scan_timestamp_range(
+                ts, range.first, range.second, [&](int64_t i, Timestamp val) {
+                    ttl_mask[i] = val <= collection_ttl;
+                });
             bitset_chunk |= ttl_mask;
         }
     }
 
-    AssertInfo(timestamps_data_size == get_row_count(),
+    AssertInfo(total_size == get_row_count(),
                fmt::format("Timestamp size not equal to row count: {}, {}",
-                           timestamps_data_size,
+                           total_size,
                            get_row_count()));
     auto range = insert_record_.timestamp_index_.get_active_range(timestamp);
 
-    // range == (size_, size_) and size_ is this->timestamps_.size().
-    // it means these data are all useful, we don't need to update bitset_chunk.
-    // It can be thought of as an OR operation with another bitmask that is all 0s, but it is not necessary to do so.
-    if (range.first == range.second && range.first == timestamps_data_size) {
-        // just skip
+    // range == (size_, size_): all data is useful, no filtering needed.
+    if (range.first == range.second && range.first == total_size) {
         return;
     }
-    // range == (0, 0). it means these data can not be used, directly set bitset_chunk to all 1s.
-    // It can be thought of as an OR operation with another bitmask that is all 1s.
+    // range == (0, 0): all data is too new, mask everything out.
     if (range.first == range.second && range.first == 0) {
         bitset_chunk.set();
         return;
     }
-    auto mask = TimestampIndex::GenerateBitset(
-        timestamp, range, timestamps_data, timestamps_data_size);
+    // [0, beg) = false, [beg, end) = check, [end, size) = true
+    BitsetType mask;
+    mask.reserve(total_size);
+    mask.resize(range.first, false);
+    mask.resize(total_size, true);
+    scan_timestamp_range(
+        ts, range.first, range.second, [&](int64_t i, Timestamp val) {
+            mask[i] = val > timestamp;
+        });
     bitset_chunk |= mask;
 }
 
@@ -2905,21 +2921,70 @@ ChunkedSegmentSealedImpl::load_field_data_common(
     }
 }
 
-void
-ChunkedSegmentSealedImpl::init_timestamp_index(
-    const std::vector<Timestamp>& timestamps, size_t num_rows) {
+static TimestampIndex
+build_timestamp_index(const Timestamp* data, size_t num_rows) {
     TimestampIndex index;
     auto min_slice_length = num_rows < 4096 ? 1 : 4096;
-    auto meta =
-        GenerateFakeSlices(timestamps.data(), num_rows, min_slice_length);
+    auto meta = GenerateFakeSlices(data, num_rows, min_slice_length);
     index.set_length_meta(std::move(meta));
-    // todo ::opt to avoid copy timestamps from field data
-    index.build_with(timestamps.data(), num_rows);
+    index.build_with(data, num_rows);
+    return index;
+}
 
-    // use special index
+void
+ChunkedSegmentSealedImpl::init_timestamp_index_from_column(
+    std::shared_ptr<ChunkedColumnInterface> column, size_t num_rows) {
+    auto all_chunks = column->GetAllChunks(nullptr);
+
+    // Build timestamp index — needs contiguous data
+    TimestampIndex index;
+    if (all_chunks.size() == 1) {
+        // Single chunk: build index directly from chunk data
+        auto* fixed_chunk = static_cast<FixedWidthChunk*>(all_chunks[0].get());
+        auto span = fixed_chunk->Span();
+        auto* ts_ptr = static_cast<const Timestamp*>(span.data());
+        AssertInfo(static_cast<size_t>(span.row_count()) == num_rows,
+                   "timestamp chunk row count {} != expected {}",
+                   span.row_count(),
+                   num_rows);
+        index = build_timestamp_index(ts_ptr, num_rows);
+    } else {
+        // Multi-chunk: temp contiguous copy for index building only
+        std::vector<Timestamp> temp(num_rows);
+        size_t offset = 0;
+        for (auto& pin : all_chunks) {
+            auto* fixed_chunk = static_cast<FixedWidthChunk*>(pin.get());
+            auto span = fixed_chunk->Span();
+            auto n = std::min(static_cast<size_t>(span.row_count()),
+                              num_rows - offset);
+            std::copy_n(static_cast<const Timestamp*>(span.data()),
+                        n,
+                        temp.data() + offset);
+            offset += n;
+        }
+        AssertInfo(offset == num_rows,
+                   "timestamp total row count {} != expected {}",
+                   offset,
+                   num_rows);
+        index = build_timestamp_index(temp.data(), num_rows);
+        // temp is freed here — runtime access uses pinned chunks directly
+    }
+
+    // Always pin mode: zero-copy segmented access
     std::unique_lock lck(mutex_);
     AssertInfo(insert_record_.timestamps_.empty(), "already exists");
-    insert_record_.init_timestamps(timestamps, index);
+    insert_record_.init_timestamps_from_column(
+        std::move(column), std::move(all_chunks), std::move(index));
+}
+
+void
+ChunkedSegmentSealedImpl::init_timestamp_index_owned(
+    std::vector<Timestamp> timestamps, size_t num_rows) {
+    auto index = build_timestamp_index(timestamps.data(), num_rows);
+    std::unique_lock lck(mutex_);
+    AssertInfo(insert_record_.timestamps_.empty(), "already exists");
+    insert_record_.init_timestamps_from_owned(std::move(timestamps),
+                                              std::move(index));
     stats_.mem_size += sizeof(Timestamp) * num_rows;
 }
 
@@ -3365,29 +3430,9 @@ ChunkedSegmentSealedImpl::LoadColumnGroup(
             auto timestamp_proxy_column = get_column(TimestampFieldID);
             AssertInfo(timestamp_proxy_column != nullptr,
                        "timestamp proxy column is nullptr");
-            // TODO check timestamp_index ready instead of check system_ready_count_
             int64_t num_rows = segment_load_info_.GetNumOfRows();
-            auto all_ts_chunks = timestamp_proxy_column->GetAllChunks(nullptr);
-            std::vector<Timestamp> timestamps(num_rows);
-            int64_t offset = 0;
-            for (auto& all_ts_chunk : all_ts_chunks) {
-                auto chunk_data = all_ts_chunk.get();
-                auto fixed_chunk = dynamic_cast<FixedWidthChunk*>(chunk_data);
-                auto span = fixed_chunk->Span();
-
-                for (size_t j = 0; j < span.row_count(); j++) {
-                    auto ts = *(int64_t*)((char*)span.data() +
-                                          j * span.element_sizeof());
-                    timestamps[offset++] = ts;
-                }
-            }
-            init_timestamp_index(timestamps, num_rows);
+            init_timestamp_index_from_column(timestamp_proxy_column, num_rows);
             system_ready_count_++;
-            AssertInfo(offset == num_rows,
-                       "[StorageV2] timestamp total row count {} not equal "
-                       "to expected {}",
-                       offset,
-                       num_rows);
         }
     }
 }

internal/core/src/segcore/ChunkedSegmentSealedImpl.h

@@ -442,7 +442,10 @@ class ChunkedSegmentSealedImpl : public SegmentSealed {
 
     const ConcurrentVector<Timestamp>&
     get_timestamps() const override {
-        return insert_record_.timestamps_;
+        // Sealed segments no longer store timestamps in ConcurrentVector.
+        // Only growing segments use this method.
+        ThrowInfo(NotImplemented,
+                  "sealed segment does not support get_timestamps()");
     }
 
     // Load Geometry cache for a field
@@ -457,6 +460,17 @@ class ChunkedSegmentSealedImpl : public SegmentSealed {
         FieldDataInfo& data,
         milvus::proto::common::LoadPriority load_priority);
 
+    // Initialize timestamp index from a column (zero-copy pin mode for single
+    // chunk, owned copy for multi-chunk)
+    void
+    init_timestamp_index_from_column(
+        std::shared_ptr<ChunkedColumnInterface> column, size_t num_rows);
+
+    // Initialize timestamp index with owned data (StorageV1 path)
+    void
+    init_timestamp_index_owned(std::vector<Timestamp> timestamps,
+                               size_t num_rows);
+
     template <typename PK>
     void
     search_sorted_pk_range_impl(
@@ -1018,10 +1032,6 @@ class ChunkedSegmentSealedImpl : public SegmentSealed {
     void
     FillDefaultValueFields(const std::vector<FieldId>& field_ids);
 
-    void
-    init_timestamp_index(const std::vector<Timestamp>& timestamps,
-                         size_t num_rows);
-
     void
     LoadFieldData(const LoadFieldDataInfo& load_info,
                   milvus::OpContext* op_ctx,

internal/core/src/segcore/InsertRecord.h

@@ -21,6 +21,7 @@
 #include <utility>
 #include <vector>
 
+#include "TimestampData.h"
 #include "TimestampIndex.h"
 #include "common/EasyAssert.h"
 #include "common/Schema.h"
@@ -426,12 +427,10 @@ class OffsetOrderedArray : public OffsetMap {
 
 class InsertRecordSealed {
  public:
-    InsertRecordSealed(const Schema& schema,
-                       const int64_t size_per_chunk,
-                       const storage::MmapChunkDescriptorPtr
-                       /* mmap_descriptor */
-                       = nullptr)
-        : timestamps_(size_per_chunk) {
+    InsertRecordSealed(
+        const Schema& schema,
+        const int64_t size_per_chunk,
+        const storage::MmapChunkDescriptorPtr mmap_descriptor = nullptr) {
         std::optional<FieldId> pk_field_id = schema.get_primary_field_id();
         // for sealed segment, only pk field is added.
         for (auto& field : schema) {
@@ -596,22 +595,39 @@ class InsertRecordSealed {
         estimated_memory_size_ += pk2offset_->memory_size();
     }
 
+    // Pin mode: zero-copy from column chunks (StorageV2, single or multi-chunk)
     void
-    init_timestamps(const std::vector<Timestamp>& timestamps,
-                    const TimestampIndex& timestamp_index) {
+    init_timestamps_from_column(
+        std::shared_ptr<ChunkedColumnInterface> column,
+        std::vector<cachinglayer::PinWrapper<Chunk*>> pins,
+        TimestampIndex timestamp_index) {
         std::lock_guard lck(shared_mutex_);
-        timestamps_.set_data_raw(0, timestamps.data(), timestamps.size());
+        timestamps_.InitFromPinnedChunks(std::move(column), std::move(pins));
         timestamp_index_ = std::move(timestamp_index);
-        AssertInfo(timestamps_.num_chunk() == 1,
-                   "num chunk not equal to 1 for sealed segment");
-        size_t memory_size = timestamps.size() * sizeof(Timestamp) +
-                             timestamp_index_.memory_size();
+        // Pin mode: timestamp data is managed by the column group,
+        // only charge the index metadata memory.
+        size_t ts_index_size = timestamp_index_.memory_size();
         cachinglayer::Manager::GetInstance().ChargeLoadedResource(
-            {static_cast<int64_t>(memory_size), 0});
-        estimated_memory_size_ += memory_size;
+            {static_cast<int64_t>(ts_index_size), 0});
+        estimated_memory_size_ += ts_index_size;
     }
 
-    const ConcurrentVector<Timestamp>&
+    // Own mode: takes ownership of timestamp data (StorageV1 / multi-chunk)
+    void
+    init_timestamps_from_owned(std::vector<Timestamp> data,
+                               TimestampIndex timestamp_index) {
+        std::lock_guard lck(shared_mutex_);
+        size_t count = data.size();
+        timestamps_.InitFromOwnedData(std::move(data));
+        timestamp_index_ = std::move(timestamp_index);
+        size_t ts_data_size = count * sizeof(Timestamp);
+        size_t ts_index_size = timestamp_index_.memory_size();
+        cachinglayer::Manager::GetInstance().ChargeLoadedResource(
+            {static_cast<int64_t>(ts_data_size + ts_index_size), 0});
+        estimated_memory_size_ += ts_data_size + ts_index_size;
+    }
+
+    const TimestampData&
     timestamps() const {
         return timestamps_;
     }
@@ -633,7 +649,7 @@ class InsertRecordSealed {
     }
 
  public:
-    ConcurrentVector<Timestamp> timestamps_;
+    TimestampData timestamps_;
     std::atomic<int64_t> reserved = 0;
     // used for timestamps index of sealed segment
     TimestampIndex timestamp_index_;

internal/core/src/segcore/SegmentInterface.cpp

@@ -462,49 +462,6 @@ SegmentInternalInterface::set_field_avg_size(FieldId field_id,
     }
 }
 
-void
-SegmentInternalInterface::timestamp_filter(BitsetType& bitset,
-                                           Timestamp timestamp) const {
-    auto& timestamps = get_timestamps();
-    auto cnt = bitset.size();
-    if (timestamps[cnt - 1] <= timestamp) {
-        // no need to filter out anything.
-        return;
-    }
-
-    auto pilot = upper_bound(timestamps, 0, cnt, timestamp);
-    // offset bigger than pilot should be filtered out.
-    auto offset = pilot;
-    while (offset < cnt) {
-        bitset[offset] = false;
-
-        const auto next_offset = bitset.find_next(offset);
-        if (!next_offset.has_value()) {
-            return;
-        }
-        offset = next_offset.value();
-    }
-}
-
-void
-SegmentInternalInterface::timestamp_filter(BitsetType& bitset,
-                                           const std::vector<int64_t>& offsets,
-                                           Timestamp timestamp) const {
-    auto& timestamps = get_timestamps();
-    auto cnt = bitset.size();
-    if (timestamps[cnt - 1] <= timestamp) {
-        // no need to filter out anything.
-        return;
-    }
-
-    // point query, faster than binary search.
-    for (auto& offset : offsets) {
-        if (timestamps[offset] > timestamp) {
-            bitset.set(offset, true);
-        }
-    }
-}
-
 const SkipIndex&
 SegmentInternalInterface::GetSkipIndex() const {
     return skip_index_;

internal/core/src/segcore/SegmentInterface.h

@@ -533,32 +533,6 @@ class SegmentInternalInterface : public SegmentInterface {
     virtual void
     search_ids(BitsetType& bitset, const IdArray& id_array) const = 0;
 
-    /**
-     * Apply timestamp filtering on bitset, the query can't see an entity whose
-     * timestamp is bigger than the timestamp of query.
-     *
-     * @param bitset The final bitset after scalar filtering and delta filtering,
-     *  `false` means that the entity will be filtered out.
-     * @param timestamp The timestamp of query.
-     */
-    void
-    timestamp_filter(BitsetType& bitset, Timestamp timestamp) const;
-
-    /**
-     * Apply timestamp filtering on bitset, the query can't see an entity whose
-     * timestamp is bigger than the timestamp of query. The passed offsets are
-     * all candidate entities.
-     *
-     * @param bitset The final bitset after scalar filtering and delta filtering,
-     *  `true` means that the entity will be filtered out.
-     * @param offsets The segment offsets of all candidates.
-     * @param timestamp The timestamp of query.
-     */
-    void
-    timestamp_filter(BitsetType& bitset,
-                     const std::vector<int64_t>& offsets,
-                     Timestamp timestamp) const;
-
     /**
      * Sort all candidates in ascending order, and then return the limit smallest.
      * Bitset is used to check if the candidate will be filtered out. `false_filtered_out`

internal/core/src/segcore/TimestampData.h

@@ -0,0 +1,171 @@
+// Copyright (C) 2019-2020 Zilliz. All rights reserved.
+//
+// Licensed 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
+
+#pragma once
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <vector>
+
+#include "cachinglayer/CacheSlot.h"
+#include "common/Chunk.h"
+#include "common/Types.h"
+#include "mmap/ChunkedColumnInterface.h"
+#include "segcore/storagev1translator/ChunkTranslator.h"
+
+namespace milvus::segcore {
+
+// TimestampData provides read-only access to timestamp data for sealed segments.
+// Supports segmented (multi-chunk) access with two storage modes:
+// - Pin mode: zero-copy from ChunkedColumnInterface chunks (StorageV2).
+//   Holds column reference and chunk pins to prevent eviction.
+// - Own mode: owns a std::vector<Timestamp> (StorageV1).
+//
+// Uses prefix-sum array (num_rows_until_chunk_) + virtual-chunk index
+// (vcid_to_cid_arr_) for O(1) offset-to-chunk lookup, consistent with
+// the pattern in ChunkTranslator / ChunkedColumn.
+class TimestampData {
+ public:
+    TimestampData() = default;
+
+    TimestampData(const TimestampData&) = delete;
+    TimestampData&
+    operator=(const TimestampData&) = delete;
+    TimestampData(TimestampData&&) = default;
+    TimestampData&
+    operator=(TimestampData&&) = default;
+
+    // Pin mode: zero-copy from column chunks (single or multi-chunk).
+    // Extracts data pointers from each pinned FixedWidthChunk.
+    void
+    InitFromPinnedChunks(std::shared_ptr<ChunkedColumnInterface> column,
+                         std::vector<cachinglayer::PinWrapper<Chunk*>> pins) {
+        column_ = std::move(column);
+        pins_ = std::move(pins);
+        chunk_data_.reserve(pins_.size());
+        num_rows_until_chunk_.reserve(pins_.size() + 1);
+        num_rows_until_chunk_.push_back(0);
+        for (auto& pin : pins_) {
+            auto* fixed_chunk = static_cast<FixedWidthChunk*>(pin.get());
+            auto span = fixed_chunk->Span();
+            chunk_data_.push_back(static_cast<const Timestamp*>(span.data()));
+            num_rows_until_chunk_.push_back(
+                num_rows_until_chunk_.back() +
+                static_cast<int64_t>(span.row_count()));
+        }
+        total_size_ = num_rows_until_chunk_.back();
+        build_virtual_chunk_index();
+    }
+
+    // Own mode: takes ownership of contiguous timestamp data.
+    void
+    InitFromOwnedData(std::vector<Timestamp> data) {
+        total_size_ = static_cast<int64_t>(data.size());
+        owned_ = std::move(data);
+        chunk_data_.push_back(owned_.data());
+        num_rows_until_chunk_.push_back(0);
+        num_rows_until_chunk_.push_back(total_size_);
+        // single chunk, no need for virtual chunk index
+        virt_chunk_order_ = 0;
+        vcid_to_cid_arr_.assign(1, 0);
+    }
+
+    Timestamp
+    operator[](int64_t i) const {
+        // Fast path: single chunk
+        if (chunk_data_.size() == 1) {
+            return chunk_data_[0][i];
+        }
+        // O(1) lookup via virtual chunk index
+        auto cid = vcid_to_cid_arr_[i >> virt_chunk_order_];
+        while (cid < static_cast<int64_t>(chunk_data_.size()) - 1 &&
+               i >= num_rows_until_chunk_[cid + 1]) {
+            ++cid;
+        }
+        return chunk_data_[cid][i - num_rows_until_chunk_[cid]];
+    }
+
+    int64_t
+    size() const {
+        return total_size_;
+    }
+
+    bool
+    empty() const {
+        return total_size_ == 0;
+    }
+
+    // Chunk-level access for callers that need raw data pointers
+    int64_t
+    num_chunks() const {
+        return static_cast<int64_t>(chunk_data_.size());
+    }
+
+    const Timestamp*
+    chunk_data(int64_t chunk_id) const {
+        return chunk_data_[chunk_id];
+    }
+
+    int64_t
+    chunk_row_count(int64_t chunk_id) const {
+        return num_rows_until_chunk_[chunk_id + 1] -
+               num_rows_until_chunk_[chunk_id];
+    }
+
+    int64_t
+    chunk_start_offset(int64_t chunk_id) const {
+        return num_rows_until_chunk_[chunk_id];
+    }
+
+    void
+    clear() {
+        chunk_data_.clear();
+        num_rows_until_chunk_.clear();
+        vcid_to_cid_arr_.clear();
+        virt_chunk_order_ = 0;
+        total_size_ = 0;
+        column_.reset();
+        pins_.clear();
+        owned_.clear();
+        owned_.shrink_to_fit();
+    }
+
+ private:
+    void
+    build_virtual_chunk_index() {
+        storagev1translator::virtual_chunk_config(
+            total_size_,
+            static_cast<int64_t>(chunk_data_.size()),
+            num_rows_until_chunk_,
+            virt_chunk_order_,
+            vcid_to_cid_arr_);
+    }
+
+    // Per-chunk raw data pointers (parallel with pins_)
+    std::vector<const Timestamp*> chunk_data_;
+    // Prefix-sum: num_rows_until_chunk_[i] = total rows before chunk i.
+    // Size = num_chunks + 1. num_rows_until_chunk_[num_chunks] = total_size_.
+    std::vector<int64_t> num_rows_until_chunk_;
+    // Virtual chunk index for O(1) offset->chunk_id lookup
+    int64_t virt_chunk_order_ = 0;
+    std::vector<int64_t> vcid_to_cid_arr_;
+    int64_t total_size_ = 0;
+
+    // Pin mode: hold column reference and pins to prevent eviction
+    std::shared_ptr<ChunkedColumnInterface> column_;
+    std::vector<cachinglayer::PinWrapper<Chunk*>> pins_;
+    // Own mode: self-owned data
+    std::vector<Timestamp> owned_;
+};
+
+}  // namespace milvus::segcore

internal/core/src/segcore/TimestampDataTest.cpp

@@ -0,0 +1,330 @@
+// Copyright (C) 2019-2020 Zilliz. All rights reserved.
+//
+// Licensed 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 <cstring>
+#include <numeric>
+#include <vector>
+
+#include "cachinglayer/CacheSlot.h"
+#include "common/Chunk.h"
+#include "common/Types.h"
+#include "segcore/TimestampData.h"
+
+using namespace milvus;
+using namespace milvus::segcore;
+
+namespace {
+
+// Helper: allocate a FixedWidthChunk holding `count` Timestamp values.
+// Returns the raw buffer (caller owns) and the chunk pointer.
+struct ChunkAlloc {
+    char* buf;
+    FixedWidthChunk* chunk;
+};
+
+ChunkAlloc
+make_ts_chunk(const Timestamp* src, int64_t count) {
+    auto byte_size = static_cast<uint64_t>(count * sizeof(Timestamp));
+    char* buf = new char[byte_size];
+    std::memcpy(buf, src, byte_size);
+    auto* chunk = new FixedWidthChunk(static_cast<int32_t>(count),
+                                      /*dim=*/1,
+                                      buf,
+                                      byte_size,
+                                      /*element_size=*/sizeof(Timestamp),
+                                      /*nullable=*/false,
+                                      /*chunk_mmap_guard=*/nullptr);
+    return {buf, chunk};
+}
+
+}  // namespace
+
+// =====================================================================
+// OwnedData tests
+// =====================================================================
+
+TEST(TimestampData, OwnedEmpty) {
+    TimestampData td;
+    ASSERT_TRUE(td.empty());
+    ASSERT_EQ(td.size(), 0);
+    ASSERT_EQ(td.num_chunks(), 0);
+}
+
+TEST(TimestampData, OwnedSingle) {
+    std::vector<Timestamp> data = {10, 20, 30, 40, 50};
+    TimestampData td;
+    td.InitFromOwnedData(std::move(data));
+
+    ASSERT_FALSE(td.empty());
+    ASSERT_EQ(td.size(), 5);
+    ASSERT_EQ(td.num_chunks(), 1);
+    ASSERT_EQ(td.chunk_row_count(0), 5);
+    ASSERT_EQ(td.chunk_start_offset(0), 0);
+
+    EXPECT_EQ(td[0], 10);
+    EXPECT_EQ(td[1], 20);
+    EXPECT_EQ(td[2], 30);
+    EXPECT_EQ(td[3], 40);
+    EXPECT_EQ(td[4], 50);
+
+    // chunk_data pointer should be valid
+    const Timestamp* ptr = td.chunk_data(0);
+    ASSERT_NE(ptr, nullptr);
+    EXPECT_EQ(ptr[0], 10);
+    EXPECT_EQ(ptr[4], 50);
+}
+
+TEST(TimestampData, OwnedLarge) {
+    const int64_t N = 100000;
+    std::vector<Timestamp> data(N);
+    std::iota(data.begin(), data.end(), 1);  // 1, 2, ..., N
+
+    TimestampData td;
+    td.InitFromOwnedData(std::move(data));
+
+    ASSERT_EQ(td.size(), N);
+    ASSERT_EQ(td.num_chunks(), 1);
+
+    EXPECT_EQ(td[0], 1);
+    EXPECT_EQ(td[N / 2], N / 2 + 1);
+    EXPECT_EQ(td[N - 1], N);
+}
+
+TEST(TimestampData, OwnedClear) {
+    std::vector<Timestamp> data = {100, 200, 300};
+    TimestampData td;
+    td.InitFromOwnedData(std::move(data));
+
+    ASSERT_EQ(td.size(), 3);
+    td.clear();
+    ASSERT_TRUE(td.empty());
+    ASSERT_EQ(td.size(), 0);
+    ASSERT_EQ(td.num_chunks(), 0);
+}
+
+TEST(TimestampData, MoveConstruct) {
+    std::vector<Timestamp> data = {10, 20, 30};
+    TimestampData td;
+    td.InitFromOwnedData(std::move(data));
+
+    TimestampData td2 = std::move(td);
+    ASSERT_EQ(td2.size(), 3);
+    EXPECT_EQ(td2[0], 10);
+    EXPECT_EQ(td2[2], 30);
+}
+
+// =====================================================================
+// PinnedChunks tests (single chunk)
+// =====================================================================
+
+TEST(TimestampData, PinnedSingleChunk) {
+    std::vector<Timestamp> raw = {5, 15, 25, 35, 45};
+    auto alloc = make_ts_chunk(raw.data(), raw.size());
+
+    // Wrap in PinWrapper (no real RAII needed for test)
+    std::vector<cachinglayer::PinWrapper<Chunk*>> pins;
+    pins.emplace_back(static_cast<Chunk*>(alloc.chunk));
+
+    TimestampData td;
+    td.InitFromPinnedChunks(/*column=*/nullptr, std::move(pins));
+
+    ASSERT_EQ(td.size(), 5);
+    ASSERT_EQ(td.num_chunks(), 1);
+    ASSERT_EQ(td.chunk_row_count(0), 5);
+    ASSERT_EQ(td.chunk_start_offset(0), 0);
+
+    EXPECT_EQ(td[0], 5);
+    EXPECT_EQ(td[2], 25);
+    EXPECT_EQ(td[4], 45);
+
+    td.clear();
+    delete alloc.chunk;
+    delete[] alloc.buf;
+}
+
+// =====================================================================
+// PinnedChunks tests (multi chunk)
+// =====================================================================
+
+TEST(TimestampData, PinnedMultiChunk) {
+    // 3 chunks: [10,20,30], [40,50], [60,70,80,90]
+    std::vector<Timestamp> c1 = {10, 20, 30};
+    std::vector<Timestamp> c2 = {40, 50};
+    std::vector<Timestamp> c3 = {60, 70, 80, 90};
+
+    auto a1 = make_ts_chunk(c1.data(), c1.size());
+    auto a2 = make_ts_chunk(c2.data(), c2.size());
+    auto a3 = make_ts_chunk(c3.data(), c3.size());
+
+    std::vector<cachinglayer::PinWrapper<Chunk*>> pins;
+    pins.emplace_back(static_cast<Chunk*>(a1.chunk));
+    pins.emplace_back(static_cast<Chunk*>(a2.chunk));
+    pins.emplace_back(static_cast<Chunk*>(a3.chunk));
+
+    TimestampData td;
+    td.InitFromPinnedChunks(nullptr, std::move(pins));
+
+    ASSERT_EQ(td.size(), 9);
+    ASSERT_EQ(td.num_chunks(), 3);
+
+    // chunk metadata
+    EXPECT_EQ(td.chunk_start_offset(0), 0);
+    EXPECT_EQ(td.chunk_row_count(0), 3);
+    EXPECT_EQ(td.chunk_start_offset(1), 3);
+    EXPECT_EQ(td.chunk_row_count(1), 2);
+    EXPECT_EQ(td.chunk_start_offset(2), 5);
+    EXPECT_EQ(td.chunk_row_count(2), 4);
+
+    // sequential access across all chunks
+    std::vector<Timestamp> expected = {10, 20, 30, 40, 50, 60, 70, 80, 90};
+    for (int64_t i = 0; i < 9; ++i) {
+        EXPECT_EQ(td[i], expected[i]) << "mismatch at index " << i;
+    }
+
+    // chunk_data pointers
+    EXPECT_EQ(td.chunk_data(0)[0], 10);
+    EXPECT_EQ(td.chunk_data(1)[0], 40);
+    EXPECT_EQ(td.chunk_data(2)[3], 90);
+
+    td.clear();
+    delete a1.chunk;
+    delete[] a1.buf;
+    delete a2.chunk;
+    delete[] a2.buf;
+    delete a3.chunk;
+    delete[] a3.buf;
+}
+
+TEST(TimestampData, PinnedManySmallChunks) {
+    // Stress the vcid_to_cid_arr with many small chunks
+    const int64_t num_chunks = 50;
+    const int64_t rows_per_chunk = 100;
+    const int64_t total = num_chunks * rows_per_chunk;
+
+    std::vector<std::vector<Timestamp>> raw_chunks(num_chunks);
+    std::vector<ChunkAlloc> allocs;
+    std::vector<cachinglayer::PinWrapper<Chunk*>> pins;
+
+    Timestamp val = 0;
+    for (int64_t c = 0; c < num_chunks; ++c) {
+        raw_chunks[c].resize(rows_per_chunk);
+        for (int64_t r = 0; r < rows_per_chunk; ++r) {
+            raw_chunks[c][r] = val++;
+        }
+        auto alloc = make_ts_chunk(raw_chunks[c].data(), rows_per_chunk);
+        allocs.push_back(alloc);
+        pins.emplace_back(static_cast<Chunk*>(alloc.chunk));
+    }
+
+    TimestampData td;
+    td.InitFromPinnedChunks(nullptr, std::move(pins));
+
+    ASSERT_EQ(td.size(), total);
+    ASSERT_EQ(td.num_chunks(), num_chunks);
+
+    // Verify every element via operator[]
+    for (int64_t i = 0; i < total; ++i) {
+        ASSERT_EQ(td[i], static_cast<Timestamp>(i)) << "mismatch at " << i;
+    }
+
+    // Verify chunk metadata
+    for (int64_t c = 0; c < num_chunks; ++c) {
+        EXPECT_EQ(td.chunk_start_offset(c), c * rows_per_chunk);
+        EXPECT_EQ(td.chunk_row_count(c), rows_per_chunk);
+    }
+
+    td.clear();
+    for (auto& a : allocs) {
+        delete a.chunk;
+        delete[] a.buf;
+    }
+}
+
+TEST(TimestampData, PinnedUnevenChunks) {
+    // Uneven chunk sizes: 1, 7, 3, 13, 2
+    std::vector<int64_t> sizes = {1, 7, 3, 13, 2};
+    int64_t total = 0;
+    for (auto s : sizes) total += s;
+
+    std::vector<std::vector<Timestamp>> raw_chunks;
+    std::vector<ChunkAlloc> allocs;
+    std::vector<cachinglayer::PinWrapper<Chunk*>> pins;
+
+    Timestamp val = 100;
+    for (auto sz : sizes) {
+        std::vector<Timestamp> chunk_data(sz);
+        for (int64_t r = 0; r < sz; ++r) {
+            chunk_data[r] = val++;
+        }
+        auto alloc = make_ts_chunk(chunk_data.data(), sz);
+        raw_chunks.push_back(std::move(chunk_data));
+        allocs.push_back(alloc);
+        pins.emplace_back(static_cast<Chunk*>(alloc.chunk));
+    }
+
+    TimestampData td;
+    td.InitFromPinnedChunks(nullptr, std::move(pins));
+
+    ASSERT_EQ(td.size(), total);
+    ASSERT_EQ(td.num_chunks(), static_cast<int64_t>(sizes.size()));
+
+    // Verify all elements
+    val = 100;
+    for (int64_t i = 0; i < total; ++i) {
+        EXPECT_EQ(td[i], val++) << "mismatch at " << i;
+    }
+
+    // Verify boundary elements at chunk edges
+    int64_t offset = 0;
+    for (size_t c = 0; c < sizes.size(); ++c) {
+        EXPECT_EQ(td.chunk_start_offset(c), offset);
+        EXPECT_EQ(td.chunk_row_count(c), sizes[c]);
+        offset += sizes[c];
+    }
+
+    td.clear();
+    for (auto& a : allocs) {
+        delete a.chunk;
+        delete[] a.buf;
+    }
+}
+
+// =====================================================================
+// Edge cases
+// =====================================================================
+
+TEST(TimestampData, PinnedSingleElement) {
+    std::vector<Timestamp> raw = {42};
+    auto alloc = make_ts_chunk(raw.data(), 1);
+
+    std::vector<cachinglayer::PinWrapper<Chunk*>> pins;
+    pins.emplace_back(static_cast<Chunk*>(alloc.chunk));
+
+    TimestampData td;
+    td.InitFromPinnedChunks(nullptr, std::move(pins));
+
+    ASSERT_EQ(td.size(), 1);
+    EXPECT_EQ(td[0], 42);
+
+    td.clear();
+    delete alloc.chunk;
+    delete[] alloc.buf;
+}
+
+TEST(TimestampData, OwnedSingleElement) {
+    TimestampData td;
+    td.InitFromOwnedData({999});
+
+    ASSERT_EQ(td.size(), 1);
+    EXPECT_EQ(td[0], 999);
+}

internal/core/src/segcore/TimestampIndex.cpp

@@ -77,42 +77,6 @@ TimestampIndex::get_active_range(Timestamp query_timestamp) const {
     return {start_locs_[block_id], start_locs_[block_id + 1]};
 }
 
-BitsetType
-TimestampIndex::GenerateBitset(Timestamp query_timestamp,
-                               std::pair<int64_t, int64_t> active_range,
-                               const Timestamp* timestamps,
-                               int64_t size) {
-    auto [beg, end] = active_range;
-    Assert(beg < end);
-    BitsetType bitset;
-    bitset.reserve(size);
-    bitset.resize(beg, false);
-    bitset.resize(size, true);
-    for (int64_t i = beg; i < end; ++i) {
-        bitset[i] = timestamps[i] > query_timestamp;
-    }
-    return bitset;
-}
-
-BitsetType
-TimestampIndex::GenerateTTLBitset(const Timestamp* timestamps,
-                                  int64_t size,
-                                  Timestamp expire_ts,
-                                  std::pair<int64_t, int64_t> active_range) {
-    auto beg = active_range.first;
-    auto end = active_range.second;
-
-    BitsetType bitset;
-    bitset.reserve(size);
-    bitset.resize(beg, true);
-    bitset.resize(size, false);
-
-    for (int64_t i = beg; i < end; ++i) {
-        bitset[i] = timestamps[i] <= expire_ts;
-    }
-    return bitset;
-}
-
 std::vector<int64_t>
 GenerateFakeSlices(const Timestamp* timestamps,
                    int64_t size,

internal/core/src/segcore/TimestampIndex.h

@@ -25,7 +25,6 @@ class TimestampIndex {
 
     void
     build_with(const Timestamp* timestamps, int64_t size);
-    // output bitset
 
     // Return range [beg, end) that is undecided
     // [0, beg) shall be all OK
@@ -33,18 +32,6 @@ class TimestampIndex {
     std::pair<int64_t, int64_t>
     get_active_range(Timestamp query_timestamp) const;
 
-    static BitsetType
-    GenerateBitset(Timestamp query_timestamp,
-                   std::pair<int64_t, int64_t> active_range,
-                   const Timestamp* timestamps,
-                   int64_t size);
-
-    static BitsetType
-    GenerateTTLBitset(const Timestamp* timestamps,
-                      int64_t size,
-                      Timestamp expire_ts,
-                      std::pair<int64_t, int64_t> active_range);
-
     size_t
     memory_size() const {
         return sizeof(*this) + lengths_.size() * sizeof(int64_t) +