StarRocks: Backend and Compute Node

// starrocks_main.cpp (abbreviated)
bool as_cn = false;
if (argc > 1 && strcmp(argv[1], "--cn") == 0) {
    as_cn = true;
}

listen (TCP)
  -> accept connection
  -> protocol decode (Thrift frame | brpc + protobuf)
  -> service method stub
  -> [often] offer to query_rpc_pool / thrift worker threads
  -> deserialize payload (Thrift struct | protobuf + attachment)
  -> business handler (AgentServer / FragmentExecutor / DataStreamRecvr / ...)
  -> serialize Status / result
  -> write response

// Abbreviated; omitted branches marked "// ...".
// Names are the functions that actually run, in order.

// ---- 1. Process entry ----
// starrocks::main  (service/starrocks_main.cpp)
int main(int argc, char** argv) {
    bool as_cn = (argc > 1 && strcmp(argv[1], "--cn") == 0);
    // ... flags, store paths ...
    starrocks::start_be(paths, as_cn);
}

// ---- 2. Listen: Thrift BackendService on be_port ----
// starrocks::start_be
void start_be(const std::vector<StorePath>& paths, bool as_cn) {
    int thrift_port = config::be_port;
    auto thrift_server = BackendService::create(
            exec_env, orchestration_env.get(),
            process_metrics_registry->root_registry(), thrift_port);
    thrift_server->start();
    // ...
}

// starrocks::BackendService::create
std::unique_ptr<ThriftServer> BackendService::create(
        ExecEnv* exec_env, orchestration::OrchestrationEnv* orchestration_env,
        MetricRegistry* metrics, int port) {
    auto handler = std::make_shared<BackendService>(exec_env, orchestration_env);
    auto processor = std::make_shared<BackendServiceProcessor>(handler);
    return std::make_unique<ThriftServer>(
            "BackendService", processor, port, metrics, config::be_service_threads);
}

// starrocks::ThriftServer::start  — bind + worker pool + binary protocol
Status ThriftServer::start() {
    auto protocol_factory = std::make_shared<apache::thrift::protocol::TBinaryProtocolFactory>();
    protocol_factory->setStrict(config::thrift_rpc_strict_mode, true);
    auto thread_mgr = apache::thrift::concurrency::ThreadManager::newSimpleThreadManager(_num_worker_threads);
    thread_mgr->start();
    auto port = std::make_shared<apache::thrift::transport::TNonblockingServerSocket>(_port);
    _server = std::make_unique<apache::thrift::server::TNonblockingServer>(
            _processor, /*in/out transport*/ transport_factory, transport_factory,
            protocol_factory, protocol_factory, port, thread_mgr);
    _server->setServerEventHandler(event_processor);
    return event_processor->start_and_wait_for_server();
    // accept → TBinaryProtocol decode → BackendServiceProcessor
    //        → BackendService::submit_tasks / get_tablet_stat / ...
}

// ---- 3. Listen: Thrift HeartbeatService ----
// starrocks::create_heartbeat_server
StatusOr<std::unique_ptr<ThriftServer>> create_heartbeat_server(
        MetricRegistry* metrics, uint32_t server_port, uint32_t worker_thread_num) {
    auto handler = std::shared_ptr<HeartbeatServer>(new HeartbeatServer());
    handler->init_cluster_id_or_die();
    auto processor = std::shared_ptr<TProcessor>(new HeartbeatServiceProcessor(handler));
    return std::make_unique<ThriftServer>(
            "heartbeat", processor, server_port, metrics, worker_thread_num);
}
// start_be then calls heartbeat_server->start()  → ThriftServer::start
// accept → HeartbeatServiceProcessor → HeartbeatServer::heartbeat

// ---- 4. Listen: brpc PInternalService on brpc_port ----
// still inside starrocks::start_be
void start_be(...) {
    auto brpc_server = std::make_unique<brpc::Server>();
    BackendInternalServiceImpl<PInternalService> internal_service(
            exec_env, orchestration_env.get(), load_channel_mgr, batch_write_mgr);
    brpc_server->AddService(&internal_service, brpc::SERVER_DOESNT_OWN_SERVICE);
    // LakeServiceImpl also AddService on shared-data builds
    butil::EndPoint point;
    butil::str2endpoint(BackendOptions::get_service_bind_address(), config::brpc_port, &point);
    brpc_server->Start(point, &options);
    // brpc accept → protobuf method stub → PInternalServiceImplBase::*
}

// ---- 5. Receive control RPC: queue, then deserialize attachment ----
// starrocks::PInternalServiceImplBase<T>::exec_plan_fragment
void PInternalServiceImplBase<T>::exec_plan_fragment(
        google::protobuf::RpcController* cntl_base,
        const PExecPlanFragmentRequest* request,
        PExecPlanFragmentResult* response,
        google::protobuf::Closure* done) {
    auto task = [=]() { this->_exec_plan_fragment(cntl_base, request, response, done); };
    if (!_exec_env->execution_services().query_rpc_pool->try_offer(std::move(task))) {
        ClosureGuard closure_guard(done);
        Status::ServiceUnavailable("submit exec_plan_fragment task failed")
                .to_protobuf(response->mutable_status());
    }
}

// starrocks::PInternalServiceImplBase<T>::_exec_plan_fragment  (pool worker)
void PInternalServiceImplBase<T>::_exec_plan_fragment(
        google::protobuf::RpcController* cntl_base,
        const PExecPlanFragmentRequest* request,
        PExecPlanFragmentResult* response,
        google::protobuf::Closure* done) {
    ClosureGuard closure_guard(done);
    auto* cntl = static_cast<brpc::Controller*>(cntl_base);
    auto st = _exec_plan_fragment(cntl, request, response);
    st.to_protobuf(response->mutable_status());
}

// starrocks::PInternalServiceImplBase<T>::_exec_plan_fragment  (serde)
Status PInternalServiceImplBase<T>::_exec_plan_fragment(
        brpc::Controller* cntl,
        const PExecPlanFragmentRequest* request,
        PExecPlanFragmentResult* response) {
    auto ser_request = cntl->request_attachment().to_string();
    TExecPlanFragmentParams t_request;
    const auto* buf = (const uint8_t*)ser_request.data();
    uint32_t len = ser_request.size();
    RETURN_IF_ERROR(deserialize_thrift_msg(buf, &len, request->attachment_protocol(), &t_request));
    if (!t_request.__isset.fragment) {
        return orchestration::FragmentExecutor::append_incremental_scan_ranges(
                _exec_env, t_request, &t_result);
    }
    return _exec_plan_fragment_by_pipeline(t_request, t_request);
}

// starrocks::deserialize_thrift_msg  (string protocol → typed overload)
template <class T>
Status deserialize_thrift_msg(const uint8_t* buf, uint32_t* len,
                              const std::string& protocol, T* deserialized_msg) {
    if (protocol == "json") {
        return deserialize_thrift_msg<T>(buf, len, TProtocolType::JSON, deserialized_msg);
    } else if (protocol == "compact") {
        return deserialize_thrift_msg<T>(buf, len, TProtocolType::COMPACT, deserialized_msg);
    } else {
        return deserialize_thrift_msg<T>(buf, len, TProtocolType::BINARY, deserialized_msg);
    }
}

template <class T>
Status deserialize_thrift_msg(const uint8_t* buf, uint32_t* len,
                              TProtocolType type, T* deserialized_msg) {
    auto tmem_transport = std::make_shared<apache::thrift::transport::TMemoryBuffer>(
            const_cast<uint8_t*>(buf), *len, TMemoryBuffer::MemoryPolicy::OBSERVE,
            create_thrift_configuration());
    auto tproto = create_deserialize_protocol(tmem_transport, type);
    deserialized_msg->read(tproto.get());   // TExecPlanFragmentParams::read
    return Status::OK();
}

// ---- 6. Receive data RPC: protobuf already decoded; slice chunk attachment ----
// starrocks::PInternalServiceImplBase<T>::transmit_chunk
void PInternalServiceImplBase<T>::transmit_chunk(
        google::protobuf::RpcController* cntl_base,
        const PTransmitChunkParams* request,
        PTransmitChunkResult* response,
        google::protobuf::Closure* done) {
    auto task = [=]() { this->_transmit_chunk(cntl_base, request, response, done); };
    _exec_env->execution_services().query_rpc_pool->try_offer(std::move(task));
}

// starrocks::PInternalServiceImplBase<T>::_transmit_chunk
void PInternalServiceImplBase<T>::_transmit_chunk(
        google::protobuf::RpcController* cntl_base,
        const PTransmitChunkParams* request,
        PTransmitChunkResult* response,
        google::protobuf::Closure* done) {
    auto* cntl = static_cast<brpc::Controller*>(cntl_base);
    auto* req = const_cast<PTransmitChunkParams*>(request);
    if (cntl->request_attachment().size() > 0) {
        butil::IOBuf& io_buf = cntl->request_attachment();
        for (size_t i = 0; i < req->chunks().size(); ++i) {
            auto* chunk = req->mutable_chunks(i);
            io_buf.cutn(chunk->mutable_data(), chunk->data_size());
        }
    }
    _exec_env->stream_mgr()->transmit_chunk(*request, &wrapped_done);
    // DataStreamMgr::transmit_chunk → DataStreamRecvr
}

CREATE TABLE sales (
  dt DATE,
  region VARCHAR(32),
  amount BIGINT
)
DUPLICATE KEY(dt, region)
PARTITION BY RANGE(dt) (
  PARTITION p202603 VALUES [("2026-03-01"), ("2026-04-01"))
)
DISTRIBUTED BY HASH(region) BUCKETS 2;

-- dt -> partition p202603; HASH(region) -> one of 2 tablets in that partition
-- load A -> new Rowset rs020a (version 3-3) on those replicas (full schema)
INSERT INTO sales VALUES
  ('2026-03-01', 'east', 100),
  ('2026-03-01', 'west', 200),
  ('2026-03-02', 'east', 150);

-- load B -> new Rowset rs031b (version 4-4); does not rewrite rs020a or split its columns
INSERT INTO sales VALUES
  ('2026-03-03', 'east', 80),
  ('2026-03-03', 'west', 90);

{storage_root}/
├── meta/                              # RocksDB (KVStore): tablet / rowset / txn meta
├── persistent/                        # DataDir helper for persistent index
├── trash/  snapshot/  clone/  ...
└── data/
    └── {shard}/                       # e.g. 1
        └── {tablet_id}/               # e.g. 100
            └── {schema_hash}/         # BaseTablet::schema_hash_path
                ├── rs020a_0.dat       # Segment 0 of rowset rs020a
                ├── rs020a_1.dat       # Segment 1 (same rowset, if multi-segment)
                ├── rs031b_0.dat       # Segment 0 of a later published rowset
                ├── rs020a_0.del       # PK delete bitmap (when present)
                ├── rs020a_0.upt
                ├── rs020a_0_{ver}_{i}.cols
                ├── rs020a_0_{index_id}.ivt/
                ├── rs020a_0_{index_id}.vi
                └── index.l0.*         # PersistentIndex (PRIMARY_KEYS only)

CREATE TABLE sales (
  dt DATE,
  region VARCHAR(32),
  amount BIGINT
)
DUPLICATE KEY(dt, region)
DISTRIBUTED BY HASH(region) BUCKETS 2;

rs020a_0.dat
├── column dt      → data pages …
├── column region  → data pages …
├── column amount  → data pages …
└── footer         → page pointers, short-key index, num_rows

// TabletMeta::_save_meta (abbreviated)
TabletMetaPB tablet_meta_pb;
to_meta_pb(&tablet_meta_pb, skip_tablet_schema);
return TabletMetaManager::save(data_dir, tablet_meta_pb);
// key: tabletmeta_{tablet_id}_{schema_hash}

// DataDir::load (abbreviated)
TabletMetaManager::walk_with_compact_on_timeout(_kv_store, [&](tablet_id, schema_hash, value) {
    _tablet_manager->load_tablet_from_meta(this, tablet_id, schema_hash, value, ...);
    return true;
}, ...);
RowsetMetaManager::traverse_rowset_metas(_kv_store, /* COMMITTED → TxnManager; VISIBLE → non-PK load_rowset */);

// Abbreviated; omitted branches marked "// ...".

// ---- 1. open ----
StatusOr<std::unique_ptr<DeltaWriter>> DeltaWriter::open(const DeltaWriterOptions& opt,
                                                         MemTracker* mem_tracker) {
    std::unique_ptr<DeltaWriter> writer(new DeltaWriter(opt, mem_tracker, StorageEngine::instance()));
    SCOPED_THREAD_LOCAL_MEM_SETTER(mem_tracker, false);
    RETURN_IF_ERROR(writer->_init());
    return std::move(writer);
}

Status DeltaWriter::_init() {
    // ...
    TabletManager* tablet_mgr = _storage_engine->tablet_manager();
    _tablet = tablet_mgr->get_tablet(_opt.tablet_id, false);
    // ... null / PK mem / version-count checks ...

    std::lock_guard push_lock(_tablet->get_push_lock());
    auto st = _storage_engine->txn_manager()->prepare_txn(_opt.partition_id, _tablet, _opt.txn_id, _opt.load_id);
    if (!st.ok()) {
        _set_state(kAborted, st);
        return st;
    }

    RowsetWriterContext writer_context;
    // ... schema / partial-update fields ...
    writer_context.rowset_id = _storage_engine->next_rowset_id();
    writer_context.tablet_id = _opt.tablet_id;
    writer_context.txn_id = _opt.txn_id;
    writer_context.rowset_path_prefix = _tablet->schema_hash_path();
    writer_context.rowset_state = PREPARED;
    // ...

    st = RowsetFactory::create_rowset_writer(writer_context, &_rowset_writer);
    if (!st.ok()) {
        auto msg = strings::Substitute("Fail to create rowset writer. tablet_id: $0, error: $1", _opt.tablet_id,
                                       st.to_string());
        st = Status::InternalError(msg);
        _set_state(kAborted, st);
        return st;
    }
    _mem_table_sink = std::make_unique<MemTableRowsetWriterSink>(_rowset_writer.get());
    _flush_token = _storage_engine->memtable_flush_executor()->create_flush_token();
    // ... Primary replicate_token / Secondary segment_flush_token ...
    _set_state(kWriting, Status::OK());
    return Status::OK();
}

// ---- 2. write (repeated) ----
Status DeltaWriter::write(const Chunk& chunk, const uint32_t* indexes, uint32_t from, uint32_t size) {
    // ...
    if (_mem_table == nullptr) {
        RETURN_IF_ERROR(_reset_mem_table());
    }
    // ... state / Secondary / PK partial-update checks ...
    Status st;
    ASSIGN_OR_RETURN(auto full, _mem_table->insert(chunk, indexes, from, size));
    if (_mem_tracker->limit_exceeded()) {
        st = _flush_memtable();
        RETURN_IF_ERROR(_reset_mem_table());
    } else if (_mem_tracker->parent() && _mem_tracker->parent()->limit_exceeded()) {
        st = _flush_memtable();
        RETURN_IF_ERROR(_reset_mem_table());
    } else if (full) {
        st = flush_memtable_async();
        RETURN_IF_ERROR(_reset_mem_table());
    }
    if (!st.ok()) {
        _set_state(kAborted, st);
    }
    return st;
}

// MemTable::insert appends rows into _chunk; returns suggest_flush (true when is_full()).

// ---- 3. flush ----
Status DeltaWriter::flush_memtable_async(bool eos) {
    DeferOp defer([this] { _mem_table.reset(); });
    if (_mem_table != nullptr) {
        RETURN_IF_ERROR(_mem_table->finalize());   // sort / agg → _result_chunk
    }
    // ... auto-increment fill when needed ...
    // Peer (and Primary without replicate_token): submit when result chunk non-null
    if (_mem_table != nullptr && _mem_table->get_result_chunk() != nullptr) {
        return _flush_token->submit(std::move(_mem_table), eos, [this](SegmentPBPtr seg, bool, int64_t) {
            if (seg) {
                _tablet->add_in_writing_data_size(_opt.txn_id, seg->data_size());
            }
            // ... immutable tablet size check ...
        });
    }
    // ... Primary+replicate_token / Secondary paths ...
    return Status::OK();
}

Status DeltaWriter::_flush_memtable() {
    RETURN_IF_ERROR(flush_memtable_async());
    Status st = _flush_token->wait();
    // ... stats ...
    return st;
}

Status FlushToken::submit(std::unique_ptr<MemTable> mem_table, bool eos,
                          std::function<void(SegmentPBPtr, bool, int64_t)> cb) {
    RETURN_IF_ERROR(status());
    auto task = std::make_shared<MemtableFlushTask>(this, std::move(mem_table), eos, std::move(cb), _slot_idx);
    _slot_idx++;
    return _flush_token->submit(std::move(task));
}

// MemtableFlushTask::run → FlushToken::_flush_memtable → MemTable::flush
Status MemTable::flush(SegmentPB* seg_info, bool eos, int64_t* flush_data_size, int64_t slot_idx) {
    if (UNLIKELY(_result_chunk == nullptr)) {
        return Status::OK();
    }
    // ...
    if (_deletes) {
        RETURN_IF_ERROR(_sink->flush_chunk_with_deletes(*_result_chunk, *_deletes, seg_info, eos, flush_data_size,
                                                        slot_idx));
    } else if (_has_op_slot && _sink->keep_op_column()) {
        RETURN_IF_ERROR(_sink->flush_chunk_with_op(*_result_chunk, seg_info, eos, flush_data_size, slot_idx));
    } else {
        RETURN_IF_ERROR(_sink->flush_chunk(*_result_chunk, seg_info, eos, flush_data_size, slot_idx));
    }
    return Status::OK();
}

Status MemTableRowsetWriterSink::flush_chunk(const Chunk& chunk, SegmentPB* seg_info, bool eos,
                                             int64_t* flush_data_size, int64_t slot_idx) {
    return _rowset_writer->flush_chunk(chunk, seg_info);
}

Status HorizontalRowsetWriter::flush_chunk(const Chunk& chunk, SegmentPB* seg_info) {
    // ... _flush_chunk_state UPSERT / MIXED ...
    return _flush_chunk(chunk, seg_info);
}

Status HorizontalRowsetWriter::_flush_chunk(const Chunk& chunk, SegmentPB* seg_info) {
    auto segment_writer = _create_segment_writer();
    if (!segment_writer.ok()) {
        return segment_writer.status();
    }
    RETURN_IF_ERROR((*segment_writer)->append_chunk(chunk));
    // ... update _num_rows_written / seg_info row counts ...
    return _flush_segment_writer(&segment_writer.value(), seg_info);
}

Status HorizontalRowsetWriter::_flush_segment_writer(std::unique_ptr<SegmentWriter>* segment_writer,
                                                     SegmentPB* seg_info) {
    uint64_t segment_size = 0;
    uint64_t index_size = 0;
    uint64_t footer_position = 0;
    RETURN_IF_ERROR((*segment_writer)->finalize(&segment_size, &index_size, &footer_position));
    // ... sizes / partial footers / seg_info (path, index sidecars) ...
    (*segment_writer).reset();
    return Status::OK();
}

// ---- 4. close / commit ----
Status DeltaWriter::close() {
    auto state = get_state();
    switch (state) {
    // ... kUninitialized / kAborted / kCommitted / kClosed ...
    case kWriting: {
        Status st = flush_memtable_async(true);
        _set_state(st.ok() ? kClosed : kAborted, st);
        return st;
    }
    }
    return Status::OK();
}

Status DeltaWriter::commit() {
    auto state = get_state();
    switch (state) {
    // ... kUninitialized / kAborted / kWriting → error; kCommitted → OK ...
    case kClosed:
        break;
    }
    std::call_once(_commit_once, [this] { _commit_result = _do_commit_body(); });
    return _commit_result;
}

Status DeltaWriter::_do_commit_body() {
    if (auto st = _flush_token->wait(); UNLIKELY(!st.ok())) {
        _set_state(kAborted, st);
        return st;
    }
    if (auto res = _rowset_writer->build(); res.ok()) {
        _cur_rowset = std::move(res).value();
    } else {
        _set_state(kAborted, res.status());
        return res.status();
    }
    if (_replicate_token != nullptr) {
        if (auto st = _replicate_token->wait(); UNLIKELY(!st.ok())) {
            _set_state(kAborted, st);
            return st;
        }
    }
    Status res;
    FAIL_POINT_TRIGGER_ASSIGN_STATUS_OR_DEFAULT(
            load_commit_txn, res, COMMIT_TXN_FP_ACTION(_opt.txn_id, _opt.tablet_id),
            _storage_engine->txn_manager()->commit_txn(_opt.partition_id, _tablet, _opt.txn_id, _opt.load_id,
                                                       _cur_rowset, false, _is_shadow));
    if (!res.ok()) {
        _storage_engine->update_manager()->on_rowset_cancel(_tablet.get(), _cur_rowset.get());
    }
    if (!res.ok() && !res.is_already_exist()) {
        _set_state(kAborted, res);
        return res;
    }
    {
        std::lock_guard l(_state_lock);
        if (_state == kClosed) {
            _state = kCommitted;
        } else {
            return Status::InternalError(
                    fmt::format("Delta writer has been aborted. tablet_id: {}, state: {}", _opt.tablet_id,
                                state_name(_state)));
        }
    }
    return Status::OK();
}

// ---- 5. publish (FE PUBLISH_VERSION → agent) ----
Status TxnManager::publish_txn(TPartitionId partition_id, const TabletSharedPtr& tablet, TTransactionId transaction_id,
                               int64_t version, const RowsetSharedPtr& rowset, uint32_t wait_time,
                               bool is_double_write) {
    if (tablet->updates() != nullptr) {
        auto st = tablet->rowset_commit(version, rowset, wait_time, false, is_double_write);
        if (!st.ok()) {
            return st;
        }
    } else {
        auto st = tablet->add_inc_rowset(rowset, version);
        tablet->remove_in_writing_data_size(transaction_id);
        if (!st.ok() && !st.is_already_exist()) {
            return st;
        }
    }
    tablet->erase_committed_rowset(rowset);
    // ... erase txn_tablet_map entry ...
    return Status::OK();
}

Hostile (many concurrent micro-publishes on one tablet)
  load₁ → rs_v101   load₂ → rs_v102   …   load_N → rs_v1000+
  scan must union many tiny .dat files; compaction rewrites them in bulk
  → version_count hits tablet_max_versions → new DeltaWriter open fails

Preferred (batched / fewer publishes)
  one load → one larger rowset (or few segments from MemTable flushes)
  compaction merges occasionally; scan stays short

// DeltaWriter::_init (abbreviated)
if (_tablet->version_count() > config::tablet_max_versions) {
    // optionally kick event-based compaction
    return Status::ServiceUnavailable(
        /* too many versions; reduce load concurrency or increase batch size */);
}

// CompactionManager::update_tablet_async / update_tablet (abbreviated)
void CompactionManager::update_tablet_async(const TabletSharedPtr& tablet) {
    std::lock_guard lock(_dispatch_mutex);
    // coalesce by tablet_id into _dispatch_map; dispatch worker calls update_tablet
    _dispatch_map.emplace(/* or refresh */ tablet->tablet_id(), std::make_pair(tablet, 0));
}

void CompactionManager::update_tablet(const TabletSharedPtr& tablet) {
    if (tablet->need_compaction()) {
        CompactionCandidate candidate;
        candidate.tablet = tablet;
        candidate.score = tablet->compaction_score();
        candidate.type = tablet->compaction_type();
        update_candidates({candidate});
    }
}

// SizeTieredCompactionPolicy::create_compaction (abbreviated)
Version output_version{first_input->start_version(), last_input->end_version()};
for (const auto& rowset : _rowsets) {
    rowset->set_is_compacting(true);
}
CompactionTaskFactory factory(output_version, tablet, std::move(_rowsets), _score, _compaction_type);
return factory.create_compaction_task();  // Horizontal or Vertical via CompactionUtils

// HorizontalCompactionTask::run_impl + CompactionTask::_commit_compaction (abbreviated)
Status HorizontalCompactionTask::run_impl() {
    RETURN_IF_ERROR(_shortcut_compact(&statistics));
    RETURN_IF_ERROR(_horizontal_compact_data(&statistics));  // TabletReader → RowsetWriter → build
    RETURN_IF_ERROR(_validate_compaction(statistics));
    RETURN_IF_ERROR(_commit_compaction());
    return Status::OK();
}

Status CompactionTask::_commit_compaction() {
    std::unique_lock wrlock(_tablet->get_header_lock());
    // ensure each input version still present
    _tablet->modify_rowsets_without_lock({_output_rowset}, _input_rowsets, &to_replace);
    _tablet->save_meta(/* skip_schema_in_rowset_meta */);
    Rowset::close_rowsets(_input_rowsets);
    // unused inputs → GC
    return Status::OK();
}

// Abbreviated; omitted branches marked "// ...".

// ---- 1–3. OlapChunkSource::_init_olap_reader ----
Status OlapChunkSource::_init_olap_reader(RuntimeState* runtime_state) {
    RETURN_IF_ERROR(_get_tablet(_scan_range));
    // ... schema / dict / scanner columns ...
    RETURN_IF_ERROR(_init_reader_params(_scan_ctx->key_ranges()));

    std::vector<RowsetSharedPtr> rowsets;
    for (auto& rowset : _morsel->rowsets()) {
        rowsets.emplace_back(std::dynamic_pointer_cast<Rowset>(rowset));
    }
    _reader = std::make_shared<TabletReader>(
            _tablet, Version(_morsel->from_version(), _version),
            std::move(child_schema), std::move(rowsets), &_tablet_schema);
    // ... optional projection iterator → _prj_iter ...

    RETURN_IF_ERROR(_reader->prepare());
    RETURN_IF_ERROR(_reader->open(_params));
    return Status::OK();
}

// ---- 4. TabletReader::prepare / open ----
Status TabletReader::prepare() {
    if (_rowsets.empty() && !_use_gtid) {
        std::shared_lock l(_tablet->get_header_lock());
        RETURN_IF_ERROR(_tablet->capture_consistent_rowsets(_version, &_rowsets));
    }
    Rowset::acquire_readers(_rowsets);
    for (const auto& rowset : _rowsets) {
        RETURN_IF_ERROR(rowset->load());
    }
    return Status::OK();
}

Status TabletReader::open(const TabletReaderParams& read_params) {
    if (read_params.use_pk_index) {
        _reader_params = &read_params;
        return Status::OK();  // PK path deferred
    }
    return _init_collector(read_params);
}

Status TabletReader::_init_collector(const TabletReaderParams& params) {
    std::vector<ChunkIteratorPtr> seg_iters;
    RETURN_IF_ERROR(get_segment_iterators(params, &seg_iters));
    // … union / heap-merge / aggregate collectors by KeysType …
    // get_segment_iterators → Rowset::get_segment_iterators → Segment::new_iterator
    return Status::OK();
}

// SegmentIterator prune order (details in §3.6)
Status SegmentIterator::_init_scan_range_and_context() {
    RETURN_IF_ERROR(_get_row_ranges_by_rowid_range());
    RETURN_IF_ERROR(_get_row_ranges_by_keys());
    RETURN_IF_ERROR(_apply_bitmap_index());
    RETURN_IF_ERROR(_get_row_ranges_by_zone_map());
    RETURN_IF_ERROR(_get_row_ranges_by_bloom_filter());
    RETURN_IF_ERROR(_apply_inverted_index());
    RETURN_IF_ERROR(_get_row_ranges_by_vector_index());
    return Status::OK();
}

// ---- 5. pull chunks ----
Status OlapChunkSource::_read_chunk_from_storage(RuntimeState* state, Chunk* chunk) {
    Status status = _prj_iter->get_next(chunk);
    // … residual conjuncts / expr filter above storage …
    return status;
}

CREATE TABLE sales (
  dt DATE,
  region VARCHAR(32),
  amount BIGINT
)
DUPLICATE KEY(dt, region)
DISTRIBUTED BY HASH(region) BUCKETS 2;

-- prefix on dt uses the short-key index
SELECT region, SUM(amount)
FROM sales
WHERE dt = '2026-03-01'
GROUP BY region;

// SegmentIterator::_get_row_ranges_by_keys (abbreviated)
SparseRange<> scan_range_by_keys;
if (!_opts.short_key_ranges.empty()) {
    ASSIGN_OR_RETURN(scan_range_by_keys, _get_row_ranges_by_short_key_ranges());
} else {
    ASSIGN_OR_RETURN(scan_range_by_keys, _get_row_ranges_by_key_ranges());
}
_scan_range &= scan_range_by_keys;

CREATE TABLE sales (
  dt DATE,
  region VARCHAR(32),
  amount BIGINT
)
DUPLICATE KEY(dt, region)
DISTRIBUTED BY HASH(region) BUCKETS 2;

-- pages whose max(amount) < 1000 are skipped
SELECT *
FROM sales
WHERE amount >= 1000;

// Segment::_new_iterator — segment-level prune before opening the iterator
RETURN_IF_ERROR(_prune_by_segment_zone_map(read_options));
// if SegmentZoneMapPruner says impossible → Status::EndOfFile (skip .dat)

// SegmentIterator::_get_row_ranges_by_zone_map — page-level
RETURN_IF(!config::enable_index_page_level_zonemap_filter, Status::OK());
ASSIGN_OR_RETURN(auto hit_row_ranges,
                 _opts.pred_tree_for_zone_map.visit(ZoneMapFilterEvaluator{...}));
_scan_range = _scan_range.intersection(zm_range);

CREATE TABLE lineorder (
  lo_orderkey INT,
  lo_shipmode VARCHAR(16),
  lo_revenue INT
)
DUPLICATE KEY(lo_orderkey)
DISTRIBUTED BY HASH(lo_orderkey) BUCKETS 4;

CREATE INDEX idx_shipmode ON lineorder (lo_shipmode) USING BITMAP;

SELECT SUM(lo_revenue)
FROM lineorder
WHERE lo_shipmode = 'MAIL';

// SegmentIterator::_apply_bitmap_index (abbreviated)
RETURN_IF(!config::enable_index_bitmap_filter, Status::OK());
RETURN_IF_ERROR(_bitmap_index_evaluator.init(/* new_bitmap_index_iterator per cid */));
RETURN_IF(!_bitmap_index_evaluator.has_bitmap_index(), Status::OK());
RETURN_IF_ERROR(_bitmap_index_evaluator.evaluate(_scan_range, _opts.pred_tree));
// evaluate intersects matching roaring rowids into _scan_range

CREATE TABLE events (
  event_id BIGINT,
  user_id BIGINT,
  ts DATETIME
)
DUPLICATE KEY(event_id)
DISTRIBUTED BY HASH(event_id) BUCKETS 4
PROPERTIES ("bloom_filter_columns" = "user_id");

SELECT *
FROM events
WHERE user_id = 9876543210;

// SegmentIterator::_get_row_ranges_by_bloom_filter (abbreviated)
RETURN_IF(!config::enable_index_bloom_filter, Status::OK());
const bool support = _opts.pred_tree.visit(BloomFilterSupportChecker{...});
RETURN_IF(!support, Status::OK());
RETURN_IF_ERROR(
    _opts.pred_tree.visit(BloomFilterEvaluator{...}, _scan_range));
// absent → drop rowids from _scan_range; maybe → keep

CREATE TABLE docs (
  id BIGINT,
  body VARCHAR(1024),
  INDEX gin_body (body) USING GIN("parser" = "english")
)
DUPLICATE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 2;

SELECT id, body
FROM docs
WHERE body MATCH 'starrocks';

// SegmentIterator::_apply_inverted_index (abbreviated)
RETURN_IF(!_opts.enable_gin_filter, Status::OK());
roaring::Roaring row_bitmap = range2roaring(_scan_range);
// pred->seek_inverted_index(..., &row_bitmap) for MATCH / expr preds
_scan_range = roaring2range(row_bitmap);
erase_column_pred_from_pred_tree(_opts.pred_tree, erased_preds);

CREATE TABLE items (
  id BIGINT,
  embedding ARRAY<FLOAT> NOT NULL,
  INDEX idx_emb (embedding) USING VECTOR (
    "index_type" = "hnsw",
    "dim" = "3",
    "metric_type" = "l2_distance",
    "M" = "16",
    "efconstruction" = "40"
  )
)
DUPLICATE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 2;

SELECT id
FROM items
ORDER BY approx_l2_distance(embedding, [0.1, 0.2, 0.3])
LIMIT 10;

// SegmentIterator::_get_row_ranges_by_vector_index (abbreviated)
if (!_vector_index_ctx || !_vector_index_ctx->use_vector_index) {
    return Status::OK();  // non-ANN query
}
// optional: narrow by residual predicates, then ANN search in .vi
// intersect hits into _scan_range; fill id2distance_map for the read path

CREATE TABLE users (
  id BIGINT,
  name VARCHAR(64),
  balance BIGINT
)
PRIMARY KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 2;

-- publish updates PrimaryIndex; point get may use it
SELECT *
FROM users
WHERE id = 42;

// publish / apply — PrimaryIndex::upsert(rssid, rowid_start, pks, &deletes)
// point lookup API — PrimaryIndex::get(pks, &rowids)

// TabletReader::open — optional PK-index scan path
if (read_params.use_pk_index) {
    _reader_params = &read_params;
    return Status::OK();  // collector deferred; uses PK index
}
// otherwise normal segment iterators + _apply_del_vector

// Abbreviated; omitted branches marked "// ...".

// ---- 1. RPC entry: queue on query_rpc_pool ----
void PInternalServiceImplBase::exec_plan_fragment(
        google::protobuf::RpcController* cntl_base,
        const PExecPlanFragmentRequest* request,
        PExecPlanFragmentResult* response,
        google::protobuf::Closure* done) {
    auto task = [=]() { this->_exec_plan_fragment(cntl_base, request, response, done); };
    if (!_exec_env->execution_services().query_rpc_pool->try_offer(std::move(task))) {
        ClosureGuard closure_guard(done);
        Status::ServiceUnavailable("submit exec_plan_fragment task failed")
                .to_protobuf(response->mutable_status());
    }
}

// ---- 2. Pool worker: ClosureGuard + Status to protobuf ----
void PInternalServiceImplBase::_exec_plan_fragment(
        google::protobuf::RpcController* cntl_base,
        const PExecPlanFragmentRequest* request,
        PExecPlanFragmentResult* response,
        google::protobuf::Closure* done) {
    ClosureGuard closure_guard(done);
    auto* cntl = static_cast<brpc::Controller*>(cntl_base);
    if (process_exit_in_progress()) {
        cntl->SetFailed(brpc::EINTERNAL, "BE is shutting down");
        return;
    }
    auto st = _exec_plan_fragment(cntl, request, response);
    st.to_protobuf(response->mutable_status());
}

// ---- 3. Deserialize Thrift; incremental vs full deploy ----
Status PInternalServiceImplBase::_exec_plan_fragment(
        brpc::Controller* cntl,
        const PExecPlanFragmentRequest* request,
        PExecPlanFragmentResult* response) {
    auto ser_request = cntl->request_attachment().to_string();
    TExecPlanFragmentParams t_request;
    {
        const auto* buf = (const uint8_t*)ser_request.data();
        uint32_t len = ser_request.size();
        RETURN_IF_ERROR(deserialize_thrift_msg(buf, &len, request->attachment_protocol(), &t_request));
    }
    // No fragment body → append scan ranges to a live instance (step 12)
    if (!t_request.__isset.fragment) {
        TExecPlanFragmentResult t_result;
        Status code = orchestration::FragmentExecutor::append_incremental_scan_ranges(
                _exec_env, t_request, &t_result);
        copy_result_from_thrift_to_protobuf(t_result, response);
        return code;
    }

    bool is_pipeline = t_request.__isset.is_pipeline && t_request.is_pipeline;
    if (is_pipeline) {
        return _exec_plan_fragment_by_pipeline(t_request, t_request);
    }
    // SchemaTableSink may still use non-pipeline; otherwise rejected since 3.2
    return Status::InvalidArgument(
            "non-pipeline engine is no longer supported since 3.2, ...");
}

// ---- 4. Pipeline path: prepare then execute ----
Status PInternalServiceImplBase::_exec_plan_fragment_by_pipeline(
        const TExecPlanFragmentParams& t_common_param,
        const TExecPlanFragmentParams& t_unique_request) {
    orchestration::FragmentExecutor fragment_executor(_batch_write_mgr);
    auto status = fragment_executor.prepare(_exec_env, t_common_param, t_unique_request);
    if (status.ok()) {
        return fragment_executor.execute(_exec_env);
    }
    return status.is_duplicate_rpc_invocation() ? Status::OK() : status;
}

// ---- 5. prepare: ordered stages ----
Status FragmentExecutor::prepare(ExecEnv* exec_env,
                                 const TExecPlanFragmentParams& common_request,
                                 const TExecPlanFragmentParams& unique_request) {
    UnifiedExecPlanFragmentParams request(common_request, unique_request);
    // DeferOp: profile on success / _fail_cleanup on failure
    RETURN_IF_ERROR(
            RuntimeEnv::GetInstance()->query_pool_mem_tracker()->check_mem_limit(
                    "Start execute plan fragment."));
    RETURN_IF_ERROR(_prepare_query_ctx(exec_env, request));
    RETURN_IF_ERROR(_prepare_fragment_ctx(request));
    RETURN_IF_ERROR(_prepare_workgroup(request));
    RETURN_IF_ERROR(_prepare_runtime_state(exec_env, request));
    {
        auto mem_tracker = _fragment_ctx->runtime_state()->instance_mem_tracker();
        SCOPED_THREAD_LOCAL_MEM_TRACKER_SETTER(mem_tracker);
        RETURN_IF_ERROR(_prepare_global_dict(request));
        RETURN_IF_ERROR(_prepare_exec_plan(exec_env, request));
    }
    {
        auto mem_tracker = _fragment_ctx->runtime_state()->instance_mem_tracker();
        SCOPED_THREAD_LOCAL_MEM_TRACKER_SETTER(mem_tracker);
        RETURN_IF_ERROR(_prepare_pipeline_driver(exec_env, request));
        RETURN_IF_ERROR(_prepare_stream_load_pipe(exec_env, request));  // no-op unless stream-load sink
    }
    RETURN_IF_ERROR(_query_ctx->fragment_mgr()->register_ctx(
            request.fragment_instance_id(), _fragment_ctx));
    _query_ctx->mark_prepared();
    return Status::OK();
}

// ---- 5a. QueryContext ----
Status FragmentExecutor::_prepare_query_ctx(ExecEnv* exec_env,
                                            const UnifiedExecPlanFragmentParams& request) {
    const auto& query_id = request.common().params.query_id;
    const auto& fragment_instance_id = request.fragment_instance_id();
    _query_ctx_mgr = exec_env->query_context_mgr();

    if (auto existing = _query_ctx_mgr->get(query_id)) {
        if (existing->fragment_mgr()->get(fragment_instance_id)) {
            return Status::DuplicateRpcInvocation("Duplicate invocations of exec_plan_fragment");
        }
    }
    ASSIGN_OR_RETURN(_query_ctx, _query_ctx_mgr->get_or_register(query_id, /*should_exist*/));
    _query_ctx_hold = _query_ctx->get_shared_ptr();
    if (request.common().params.__isset.instances_number) {
        _query_ctx->set_total_fragments(request.common().params.instances_number);
    }
    // delivery / query expire, profile flags, ...
    return Status::OK();
}

// ---- 5b. FragmentContext shell ----
Status FragmentExecutor::_prepare_fragment_ctx(const UnifiedExecPlanFragmentParams& request) {
    _fragment_ctx = std::make_shared<FragmentContext>();
    _fragment_ctx->set_query_id(request.common().params.query_id);
    _fragment_ctx->set_fragment_instance_id(request.fragment_instance_id());
    _fragment_ctx->set_fe_addr(request.common().coord);
    // adaptive DOP / pred_tree_params when present
    return Status::OK();
}

// ---- 5c. WorkGroup ----
Status FragmentExecutor::_prepare_workgroup(const UnifiedExecPlanFragmentParams& request) {
    WorkGroupPtr wg;
    if (!request.common().__isset.workgroup ||
        request.common().workgroup.id == WorkGroup::DEFAULT_WG_ID) {
        wg = ExecEnv::GetInstance()->workgroup_manager()->get_default_workgroup();
    } else {
        wg = std::make_shared<WorkGroup>(request.common().workgroup);
        wg = ExecEnv::GetInstance()->workgroup_manager()->add_workgroup(wg);
    }
    RETURN_IF_ERROR(_query_ctx->init_query_once(wg.get(), /*enable_group_level_query_queue*/));
    _fragment_ctx->set_workgroup(wg);
    _wg = wg;
    return Status::OK();
}

// ---- 5d. RuntimeState + DescriptorTbl + mem ----
Status FragmentExecutor::_prepare_runtime_state(ExecEnv* exec_env,
                                                const UnifiedExecPlanFragmentParams& request) {
    _fragment_ctx->set_runtime_state(std::make_unique<RuntimeState>(
            query_id, fragment_instance_id, query_options, query_globals,
            &exec_env->query_execution_services(), exec_env));
    auto* runtime_state = _fragment_ctx->runtime_state();
    runtime_state->init_fragment_mem_pool();
    runtime_state->set_enable_pipeline_engine(true);
    _fragment_ctx->attach_to_runtime_state(runtime_state);
    _query_ctx->attach_to_runtime_state(runtime_state);
    RuntimeStateHelper::init_runtime_filter_port(runtime_state);

    _query_ctx->init_mem_tracker(/* query / big_query / spill limits ... */);
    runtime_state->init_mem_trackers(_query_ctx->mem_tracker());
    // open RuntimeFilterWorker when coordinator params present
    _fragment_ctx->prepare_pass_through_chunk_buffer();
    // DescriptorTbl: reuse cached on query_ctx or create into pool
    runtime_state->set_desc_tbl(desc_tbl);
    if (query_options.__isset.enable_spill && query_options.enable_spill) {
        RETURN_IF_ERROR(_query_ctx->init_spill_manager(query_options));
    }
    return Status::OK();
}

// ---- 5e. Global dicts ----
Status FragmentExecutor::_prepare_global_dict(const UnifiedExecPlanFragmentParams& request) {
    auto* fragment_dict_state = _fragment_ctx->runtime_state()->fragment_dict_state();
    const auto& fragment = request.common().fragment;
    if (fragment.__isset.query_global_dicts) {
        RETURN_IF_ERROR(fragment_dict_state->init_query_global_dict(
                runtime_state, fragment.query_global_dicts));
    }
    if (fragment.__isset.load_global_dicts) {
        RETURN_IF_ERROR(fragment_dict_state->init_load_global_dict(
                runtime_state, fragment.load_global_dicts));
    }
    return Status::OK();
}

// ---- 5f. ExecNode tree + morsel factories ----
Status FragmentExecutor::_prepare_exec_plan(ExecEnv* exec_env,
                                            const UnifiedExecPlanFragmentParams& request) {
    auto* runtime_state = _fragment_ctx->runtime_state();
    const auto pipeline_dop = _calc_dop(exec_env, request);

    _fragment_ctx->move_tplan(*const_cast<TPlan*>(&request.common().fragment.plan));
    RETURN_IF_ERROR(ExecFactory::create_tree(
            runtime_state, runtime_state->obj_pool(), _fragment_ctx->tplan(),
            runtime_state->desc_tbl(), &_fragment_ctx->plan()));
    ExecNode* plan = _fragment_ctx->plan();
    plan->push_down_tuple_slot_mappings(runtime_state, empty_mappings);
    // set ExchangeNode senders / LookUpNode fetchers from params

    for (auto* scan_node : scan_nodes) {
        // scan_ranges (+ optional per_driver_seq) from FE
        RETURN_IF_ERROR(add_scan_ranges_partition_values(runtime_state, scan_ranges));
        ASSIGN_OR_RETURN(auto morsel_queue_factory,
                         scan_node->convert_scan_range_to_morsel_queue_factory(/*...*/));
        morsel_queue_factory->set_has_more_scan_ranges(has_more_morsel);
        morsel_queue_factories.emplace(scan_node->id(), std::move(morsel_queue_factory));
    }
    return Status::OK();
}

// ---- 5g. Pipelines + drivers ----
Status FragmentExecutor::_prepare_pipeline_driver(ExecEnv* exec_env,
                                                  const UnifiedExecPlanFragmentParams& request) {
    const auto degree_of_parallelism = _calc_dop(exec_env, request);
    size_t sink_dop = _calc_sink_dop(ExecEnv::GetInstance(), request);
    ExecNode* plan = _fragment_ctx->plan();

    PipelineBuilderContext context(_fragment_ctx.get(), degree_of_parallelism, sink_dop);
    context.init_colocate_groups(std::move(_colocate_exec_groups));
    PipelineBuilder builder(context);
    ASSIGN_OR_RETURN(auto exec_ops, plan->decompose_to_pipeline(&context));
    exec_ops = maybe_interpolate_grouped_exchange(&context, plan->id(), exec_ops);

    std::unique_ptr<DataSink> datasink;
    if (request.isset_output_sink()) {
        RETURN_IF_ERROR(DataSink::create_data_sink(/*...*/, &datasink));
        RETURN_IF_ERROR(datasink->decompose_data_sink_to_pipeline(
                &context, runtime_state, std::move(exec_ops), request, tsink, output_exprs));
    }
    _fragment_ctx->set_data_sink(std::move(datasink));
    auto [exec_groups, pipelines] = builder.build();
    _fragment_ctx->set_pipelines(std::move(exec_groups), std::move(pipelines));

    RETURN_IF_ERROR(_fragment_ctx->prepare_all_pipelines());
    // bind morsel_queue_factory onto scan sources; instantiate drivers x DOP
    ASSIGN_OR_RETURN(auto driver_token,
                     exec_env->compute_env()->driver_limiter()->try_acquire(
                             _fragment_ctx->total_dop()));
    _fragment_ctx->set_driver_token(std::move(driver_token));
    return Status::OK();
}

// ---- 6. execute: prepare drivers, submit to workgroup ----
Status FragmentExecutor::execute(ExecEnv* exec_env) {
    bool prepare_success = false;
    DeferOp defer([this, &prepare_success]() {
        if (!prepare_success) {
            _fail_cleanup(true);
        }
    });
    _fragment_ctx->acquire_runtime_filters();
    RETURN_IF_ERROR(_fragment_ctx->prepare_active_drivers());
    prepare_success = true;

    auto* executor = _wg->executors()->driver_executor();
    RETURN_IF_ERROR(_fragment_ctx->submit_active_drivers(executor));
    _fragment_ctx->runtime_state()->set_fragment_prepared(true);
    return Status::OK();  // RPC returns; drivers run asynchronously
}

Status FragmentContext::prepare_active_drivers() {
    for (auto& group : _execution_groups) {
        RETURN_IF_ERROR(group->prepare_drivers(_runtime_state.get()));
    }
    // sequential or parallel prepare via pipeline_prepare_pool
    for (auto& group : _execution_groups) {
        RETURN_IF_ERROR(group->prepare_active_drivers_sequentially(_runtime_state.get()));
        // or prepare_active_drivers_parallel(...)
    }
    return Status::OK();
}

Status FragmentContext::submit_active_drivers(DriverExecutor* executor) {
    for (auto& group : _execution_groups) {
        group->attach_driver_executor(executor);
        group->submit_active_drivers();
    }
    return Status::OK();
}

// ---- 7. Incremental scan ranges (no fragment body) ----
Status FragmentExecutor::append_incremental_scan_ranges(ExecEnv* exec_env,
                                                        const TExecPlanFragmentParams& request,
                                                        TExecPlanFragmentResult* response) {
    // lookup QueryContext + FragmentContext by query_id / fragment_instance_id
    // for each scan node: build ScanMorsels → morsel_queue_factory->append_morsels
    // set_has_more_scan_ranges; optionally close scan nodes that hit limit
    // notify_source_observers so blocked scan drivers wake up
    return Status::OK();
}