[improvement](cloud) support memtable on sink node

be/src/cloud/cloud_rowset_builder.cpp

@@ -249,6 +249,17 @@ Status CloudRowsetBuilder::commit_rowset(const std::string& job_id, int64_t tabl
     return _engine.meta_mgr().commit_rowset(*rowset_meta(), job_id, table_id);
 }
 
+Status CloudRowsetBuilder::commit_txn() {
+    DCHECK(is_data_builder());
+    if (!_skip_writing_rowset_metadata) {
+        RETURN_IF_ERROR(commit_rowset("", _tablet->table_id()));
+    }
+    RETURN_IF_ERROR(set_txn_related_info());
+    update_tablet_stats();
+    _is_committed = true;
+    return Status::OK();
+}
+
 Status CloudRowsetBuilder::set_txn_related_info() {
     if (_tablet->enable_unique_key_merge_on_write() || _tablet->is_row_binlog_tablet()) {
         // For empty rowsets when skip_writing_empty_rowset_metadata=true,

be/src/cloud/cloud_rowset_builder.h

@@ -33,6 +33,8 @@ class CloudRowsetBuilder : public BaseRowsetBuilder {
 
     Status init() override;
 
+    Status commit_txn() override;
+
     virtual void update_tablet_stats();
 
     const RowsetMetaSharedPtr& rowset_meta();

be/src/cloud/cloud_rowset_writer.cpp

@@ -20,8 +20,7 @@
 #include "common/logging.h"
 #include "common/status.h"
 #include "io/cache/block_file_cache_factory.h"
-#include "io/fs/packed_file_manager.h"
-#include "io/fs/packed_file_writer.h"
+#include "io/fs/file_writer.h"
 #include "storage/rowset/rowset_factory.h"
 
 namespace doris {
@@ -103,8 +102,12 @@ Status CloudRowsetWriter::_build_rowset_meta(RowsetMeta* rowset_meta, bool check
     RETURN_IF_ERROR(BaseBetaRowsetWriter::_build_rowset_meta(rowset_meta, check_segment_num,
                                                              completed_segment_ids));
 
-    // Collect packed file segment index information for interim rowsets as well.
-    return _collect_all_packed_slice_locations(rowset_meta);
+    // Temporary bitmap rowsets collect their own segment after its file is closed.
+    // The complete collections are safe to traverse only after all flushes finish.
+    if (completed_segment_ids == nullptr) {
+        return _collect_all_packed_slice_locations(rowset_meta);
+    }
+    return Status::OK();
 }
 
 Status CloudRowsetWriter::build(RowsetSharedPtr& rowset) {
@@ -116,8 +119,6 @@ Status CloudRowsetWriter::build(RowsetSharedPtr& rowset) {
     // TODO(plat1ko): check_segment_footer
 
     RETURN_IF_ERROR(_build_rowset_meta(_rowset_meta.get()));
-    // At this point all writers have been closed, so collecting packed file indices is safe.
-    RETURN_IF_ERROR(_collect_all_packed_slice_locations(_rowset_meta.get()));
     // If the current load is a partial update, new segments may be appended to the tmp rowset after the tmp rowset
     // has been committed if conflicts occur due to concurrent partial updates. However, when the recycler do recycling,
     // it will generate the paths for the segments to be recycled on the object storage based on the number of segments
@@ -187,8 +188,7 @@ Status CloudRowsetWriter::_collect_all_packed_slice_locations(RowsetMeta* rowset
     const auto& file_writers = _seg_files.get_file_writers();
     for (const auto& [seg_id, writer_ptr] : file_writers) {
         auto segment_path = _context.segment_path(seg_id);
-        RETURN_IF_ERROR(
-                _collect_packed_slice_location(writer_ptr.get(), segment_path, rowset_meta));
+        RETURN_IF_ERROR(rowset_meta->collect_packed_slice_location(*writer_ptr, segment_path));
     }
 
     // Collect inverted index file packed indices
@@ -200,43 +200,12 @@ Status CloudRowsetWriter::_collect_all_packed_slice_locations(RowsetMeta* rowset
                     InvertedIndexDescriptor::get_index_file_path_prefix(segment_path);
             std::string index_path =
                     InvertedIndexDescriptor::get_index_file_path_v2(std::string(index_prefix_view));
-            RETURN_IF_ERROR(_collect_packed_slice_location(idx_writer_ptr->get_file_writer(),
-                                                           index_path, rowset_meta));
+            RETURN_IF_ERROR(rowset_meta->collect_packed_slice_location(
+                    *idx_writer_ptr->get_file_writer(), index_path));
         }
     }
 
     return Status::OK();
 }
 
-Status CloudRowsetWriter::_collect_packed_slice_location(io::FileWriter* file_writer,
-                                                         const std::string& file_path,
-                                                         RowsetMeta* rowset_meta) {
-    VLOG_NOTICE << "collect packed slice location for file: " << file_path;
-    // Check if file writer is closed
-    if (file_writer->state() != io::FileWriter::State::CLOSED) {
-        // Writer is still open; index will be collected after it is closed.
-        return Status::OK();
-    }
-
-    // Check if file is actually in packed file (not direct write for large files)
-    if (!file_writer->is_in_packed_file()) {
-        return Status::OK();
-    }
-
-    // Ask the writer, which holds a reference to its own slice location. Looking it up by
-    // path in PackedFileManager would race with the retention based cleanup of the index.
-    io::PackedSliceLocation index;
-    RETURN_IF_ERROR(
-            static_cast<io::PackedFileWriter*>(file_writer)->get_packed_slice_location(&index));
-    if (index.packed_file_path.empty()) {
-        return Status::OK(); // File not in packed file, skip
-    }
-
-    rowset_meta->add_packed_slice_location(file_path, index.packed_file_path, index.offset,
-                                           index.size, index.packed_file_size);
-    LOG(INFO) << "collect packed file index: " << file_path << " -> " << index.packed_file_path
-              << ", offset: " << index.offset << ", size: " << index.size;
-    return Status::OK();
-}
-
 } // namespace doris

be/src/cloud/cloud_rowset_writer.h

@@ -38,8 +38,6 @@ class CloudRowsetWriter : public BaseBetaRowsetWriter {
 
     Status _collect_all_packed_slice_locations(RowsetMeta* rowset_meta);
 
-    Status _collect_packed_slice_location(io::FileWriter* file_writer, const std::string& file_path,
-                                          RowsetMeta* rowset_meta);
     CloudStorageEngine& _engine;
 };
 

be/src/exec/pipeline/pipeline_fragment_context.cpp

@@ -1305,7 +1305,7 @@ Status PipelineFragmentContext::_create_data_sink(ObjectPool* pool, const TDataS
         int child_node_id = pipeline->operators().back()->node_id();
         if (state->query_options().enable_memtable_on_sink_node &&
             !_has_inverted_index_v1_or_partial_update(thrift_sink.olap_table_sink) &&
-            !_has_row_binlog(thrift_sink.olap_table_sink) && !config::is_cloud_mode()) {
+            !_has_row_binlog(thrift_sink.olap_table_sink)) {
             _sink = std::make_shared<OlapTableSinkV2OperatorX>(
                     pool, next_sink_operator_id(), child_node_id + 1, row_desc, output_exprs);
         } else {

be/src/exec/sink/load_stream_stub.cpp

@@ -170,7 +170,8 @@ Status LoadStreamStub::open(BrpcClientCache<PBackendService_Stub>* client_cache,
                             const NodeInfo& node_info, int64_t txn_id,
                             const OlapTableSchemaParam& schema,
                             const std::vector<PTabletID>& tablets_for_schema, int total_streams,
-                            int64_t idle_timeout_ms, bool enable_profile) {
+                            int64_t idle_timeout_ms, bool enable_profile, int64_t txn_expiration,
+                            const std::string& storage_vault_id, bool write_file_cache) {
     std::unique_lock<bthread::Mutex> lock(_open_mutex);
     if (_is_init.load()) {
         return _status;
@@ -194,6 +195,9 @@ Status LoadStreamStub::open(BrpcClientCache<PBackendService_Stub>* client_cache,
     request.set_src_id(_src_id);
     request.set_txn_id(txn_id);
     request.set_enable_profile(enable_profile);
+    request.set_txn_expiration(txn_expiration);
+    request.set_storage_vault_id(storage_vault_id);
+    request.set_write_file_cache(write_file_cache);
     if (_is_incremental) {
         request.set_total_streams(0);
     } else if (total_streams > 0) {
@@ -591,18 +595,21 @@ Status LoadStreamStubs::open(BrpcClientCache<PBackendService_Stub>* client_cache
                              const NodeInfo& node_info, int64_t txn_id,
                              const OlapTableSchemaParam& schema,
                              const std::vector<PTabletID>& tablets_for_schema, int total_streams,
-                             int64_t idle_timeout_ms, bool enable_profile) {
+                             int64_t idle_timeout_ms, bool enable_profile, int64_t txn_expiration,
+                             const std::string& storage_vault_id, bool write_file_cache) {
     bool get_schema = true;
     auto status = Status::OK();
     bool first_stream = true;
     for (auto& stream : _streams) {
         Status st;
         if (get_schema) {
             st = stream->open(client_cache, node_info, txn_id, schema, tablets_for_schema,
-                              total_streams, idle_timeout_ms, enable_profile);
+                              total_streams, idle_timeout_ms, enable_profile, txn_expiration,
+                              storage_vault_id, write_file_cache);
         } else {
             st = stream->open(client_cache, node_info, txn_id, schema, {}, total_streams,
-                              idle_timeout_ms, enable_profile);
+                              idle_timeout_ms, enable_profile, txn_expiration, storage_vault_id,
+                              write_file_cache);
         }
         // Simulate one stream open failure within LoadStreamStubs.
         // This causes the successfully opened streams to be cancelled,

be/src/exec/sink/load_stream_stub.h

@@ -144,7 +144,8 @@ class LoadStreamStub : public std::enable_shared_from_this<LoadStreamStub> {
     Status open(BrpcClientCache<PBackendService_Stub>* client_cache, const NodeInfo& node_info,
                 int64_t txn_id, const OlapTableSchemaParam& schema,
                 const std::vector<PTabletID>& tablets_for_schema, int total_streams,
-                int64_t idle_timeout_ms, bool enable_profile);
+                int64_t idle_timeout_ms, bool enable_profile, int64_t txn_expiration,
+                const std::string& storage_vault_id, bool write_file_cache);
 
 // for mock this class in UT
 #ifdef BE_TEST
@@ -331,7 +332,8 @@ class LoadStreamStubs {
     Status open(BrpcClientCache<PBackendService_Stub>* client_cache, const NodeInfo& node_info,
                 int64_t txn_id, const OlapTableSchemaParam& schema,
                 const std::vector<PTabletID>& tablets_for_schema, int total_streams,
-                int64_t idle_timeout_ms, bool enable_profile);
+                int64_t idle_timeout_ms, bool enable_profile, int64_t txn_expiration,
+                const std::string& storage_vault_id, bool write_file_cache);
 
     bool is_incremental() const { return _is_incremental; }
 

be/src/exec/sink/writer/vtablet_writer_v2.cpp

@@ -149,9 +149,18 @@ Status VTabletWriterV2::_init(RuntimeState* state, RuntimeProfile* profile) {
     _load_id.set_lo(table_sink.load_id.lo);
     signal::set_signal_task_id(_load_id);
     _txn_id = table_sink.txn_id;
+    if (config::is_cloud_mode()) {
+        if (!table_sink.__isset.txn_timeout_s || table_sink.txn_timeout_s <= 0) {
+            return Status::InternalError("The txn_timeout_s of TDataSink is invalid");
+        }
+        _txn_expiration = UnixSeconds() + table_sink.txn_timeout_s;
+    }
     _num_replicas = table_sink.num_replicas;
     _tuple_desc_id = table_sink.tuple_id;
     _write_file_cache = table_sink.write_file_cache;
+    if (table_sink.__isset.storage_vault_id) {
+        _storage_vault_id = table_sink.storage_vault_id;
+    }
     _schema.reset(new OlapTableSchemaParam());
     RETURN_IF_ERROR(_schema->init(table_sink.schema));
     _schema->set_timestamp_ms(state->timestamp_ms());
@@ -318,7 +327,8 @@ Status VTabletWriterV2::_open_streams_to_backend(int64_t dst_id, LoadStreamStubs
                     { tablets_for_schema.clear(); });
     auto st = streams.open(_state->exec_env()->brpc_streaming_client_cache(), *node_info, _txn_id,
                            *_schema, tablets_for_schema, _total_streams, idle_timeout_ms,
-                           _state->enable_profile());
+                           _state->enable_profile(), _txn_expiration, _storage_vault_id,
+                           _write_file_cache);
     if (!st.ok()) {
         LOG(WARNING) << "failed to open stream to backend " << dst_id
                      << ", load_id=" << print_id(_load_id) << ", err=" << st;
@@ -791,6 +801,9 @@ Status VTabletWriterV2::close(Status exec_status) {
             _row_distribution.output_profile_info(_operator_profile);
         }
 
+        // Keep the fragment alive so tests can fetch the completed writer profile over Thrift.
+        DBUG_EXECUTE_IF("VTabletWriterV2.close.profile_ready", DBUG_BLOCK);
+
         LOG(INFO) << "finished to close olap table sink. load_id=" << print_id(_load_id)
                   << ", txn_id=" << _txn_id;
     } else {

be/src/exec/sink/writer/vtablet_writer_v2.h

@@ -179,6 +179,7 @@ class VTabletWriterV2 final : public AsyncResultWriter {
     // unique load id
     PUniqueId _load_id;
     int64_t _txn_id = -1;
+    int64_t _txn_expiration = 0;
     int _num_replicas = -1;
     int _tuple_desc_id = -1;
 
@@ -197,6 +198,7 @@ class VTabletWriterV2 final : public AsyncResultWriter {
     int _num_local_sink = -1;
     bool _is_high_priority = false;
     bool _write_file_cache = false;
+    std::string _storage_vault_id;
 
     // TODO(zc): think about cache this data
     std::shared_ptr<OlapTableSchemaParam> _schema;