diff --git a/be/src/cloud/cloud_rowset_builder.cpp b/be/src/cloud/cloud_rowset_builder.cpp index d9065c340ece9e..7d363aea2d1aff 100644 --- a/be/src/cloud/cloud_rowset_builder.cpp +++ b/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, diff --git a/be/src/cloud/cloud_rowset_builder.h b/be/src/cloud/cloud_rowset_builder.h index c65e192444865d..d7db14f43f2595 100644 --- a/be/src/cloud/cloud_rowset_builder.h +++ b/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(); diff --git a/be/src/cloud/cloud_rowset_writer.cpp b/be/src/cloud/cloud_rowset_writer.cpp index 9753ca3dd99cd9..c633bc25e30b13 100644 --- a/be/src/cloud/cloud_rowset_writer.cpp +++ b/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(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 diff --git a/be/src/cloud/cloud_rowset_writer.h b/be/src/cloud/cloud_rowset_writer.h index 4841fbf6cf7132..e351405f738d06 100644 --- a/be/src/cloud/cloud_rowset_writer.h +++ b/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; }; diff --git a/be/src/exec/pipeline/pipeline_fragment_context.cpp b/be/src/exec/pipeline/pipeline_fragment_context.cpp index 5fa5c4c2932294..cc1d359d02df24 100644 --- a/be/src/exec/pipeline/pipeline_fragment_context.cpp +++ b/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( pool, next_sink_operator_id(), child_node_id + 1, row_desc, output_exprs); } else { diff --git a/be/src/exec/sink/load_stream_stub.cpp b/be/src/exec/sink/load_stream_stub.cpp index e46eec24333868..967e35fed49181 100644 --- a/be/src/exec/sink/load_stream_stub.cpp +++ b/be/src/exec/sink/load_stream_stub.cpp @@ -170,7 +170,8 @@ Status LoadStreamStub::open(BrpcClientCache* client_cache, const NodeInfo& node_info, int64_t txn_id, const OlapTableSchemaParam& schema, const std::vector& 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 lock(_open_mutex); if (_is_init.load()) { return _status; @@ -194,6 +195,9 @@ Status LoadStreamStub::open(BrpcClientCache* 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,7 +595,8 @@ Status LoadStreamStubs::open(BrpcClientCache* client_cache const NodeInfo& node_info, int64_t txn_id, const OlapTableSchemaParam& schema, const std::vector& 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; @@ -599,10 +604,12 @@ Status LoadStreamStubs::open(BrpcClientCache* client_cache 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, diff --git a/be/src/exec/sink/load_stream_stub.h b/be/src/exec/sink/load_stream_stub.h index 8951c483220c01..4fb16b2db71708 100644 --- a/be/src/exec/sink/load_stream_stub.h +++ b/be/src/exec/sink/load_stream_stub.h @@ -144,7 +144,8 @@ class LoadStreamStub : public std::enable_shared_from_this { Status open(BrpcClientCache* client_cache, const NodeInfo& node_info, int64_t txn_id, const OlapTableSchemaParam& schema, const std::vector& 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* client_cache, const NodeInfo& node_info, int64_t txn_id, const OlapTableSchemaParam& schema, const std::vector& 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; } diff --git a/be/src/exec/sink/writer/vtablet_writer_v2.cpp b/be/src/exec/sink/writer/vtablet_writer_v2.cpp index 8a5fe58500a1d9..fa97d272af148c 100644 --- a/be/src/exec/sink/writer/vtablet_writer_v2.cpp +++ b/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 { diff --git a/be/src/exec/sink/writer/vtablet_writer_v2.h b/be/src/exec/sink/writer/vtablet_writer_v2.h index a3d86a613bbde1..7c6269e9e8726f 100644 --- a/be/src/exec/sink/writer/vtablet_writer_v2.h +++ b/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 _schema; diff --git a/be/src/load/channel/load_stream.cpp b/be/src/load/channel/load_stream.cpp index 64c8beb0a2a073..d0ac39c7782d69 100644 --- a/be/src/load/channel/load_stream.cpp +++ b/be/src/load/channel/load_stream.cpp @@ -26,6 +26,7 @@ #include #include "bvar/bvar.h" +#include "cloud/cloud_meta_mgr.h" #include "cloud/config.h" #include "common/signal_handler.h" #include "load/channel/load_channel.h" @@ -57,11 +58,16 @@ bvar::LatencyRecorder g_load_stream_flush_wait_ms("load_stream_flush_wait_ms"); bvar::Adder g_load_stream_flush_running_threads("load_stream_flush_wait_threads"); TabletStream::TabletStream(const PUniqueId& load_id, int64_t id, int64_t txn_id, - LoadStreamMgr* load_stream_mgr, RuntimeProfile* profile) + LoadStreamMgr* load_stream_mgr, RuntimeProfile* profile, + int64_t txn_expiration, std::string storage_vault_id, + bool write_file_cache) : _id(id), _next_segid(0), _load_id(load_id), _txn_id(txn_id), + _txn_expiration(txn_expiration), + _storage_vault_id(std::move(storage_vault_id)), + _write_file_cache(write_file_cache), _load_stream_mgr(load_stream_mgr) { load_stream_mgr->create_token(_flush_token); _profile = profile->create_child(fmt::format("TabletStream {}", id), true, true); @@ -77,16 +83,17 @@ inline std::ostream& operator<<(std::ostream& ostr, const TabletStream& tablet_s } Status TabletStream::init(std::shared_ptr schema, int64_t index_id, - int64_t partition_id) { + int64_t partition_id, bool is_empty) { WriteRequest req { .tablet_id = _id, .txn_id = _txn_id, + .txn_expiration = _txn_expiration, .index_id = index_id, .partition_id = partition_id, .load_id = _load_id, .table_schema_param = schema, - // TODO(plat1ko): write_file_cache - .storage_vault_id {}, + .write_file_cache = _write_file_cache, + .storage_vault_id = _storage_vault_id, }; _load_stream_writer = std::make_shared(&req, _profile); @@ -94,7 +101,7 @@ Status TabletStream::init(std::shared_ptr schema, int64_t _status.update(Status::Uninitialized("fault injection")); return _status.status(); }); - _status.update(_load_stream_writer->init()); + _status.update(_load_stream_writer->init(is_empty)); if (!_status.ok()) { LOG(INFO) << "failed to init rowset builder due to " << *this; } @@ -346,10 +353,15 @@ Status TabletStream::close() { IndexStream::IndexStream(const PUniqueId& load_id, int64_t id, int64_t txn_id, std::shared_ptr schema, - LoadStreamMgr* load_stream_mgr, RuntimeProfile* profile) + LoadStreamMgr* load_stream_mgr, RuntimeProfile* profile, + int64_t txn_expiration, std::string storage_vault_id, + bool write_file_cache) : _id(id), _load_id(load_id), _txn_id(txn_id), + _txn_expiration(txn_expiration), + _storage_vault_id(std::move(storage_vault_id)), + _write_file_cache(write_file_cache), _schema(schema), _load_stream_mgr(load_stream_mgr) { _profile = profile->create_child(fmt::format("IndexStream {}", id), true, true); @@ -375,7 +387,7 @@ Status IndexStream::append_data(const PStreamHeader& header, butil::IOBuf* data) std::lock_guard lock_guard(_lock); auto it = _tablet_streams_map.find(tablet_id); if (it == _tablet_streams_map.end()) { - _init_tablet_stream(tablet_stream, tablet_id, header.partition_id()); + _init_tablet_stream(tablet_stream, tablet_id, header.partition_id(), false); } else { tablet_stream = it->second; } @@ -385,11 +397,12 @@ Status IndexStream::append_data(const PStreamHeader& header, butil::IOBuf* data) } void IndexStream::_init_tablet_stream(TabletStreamSharedPtr& tablet_stream, int64_t tablet_id, - int64_t partition_id) { - tablet_stream = std::make_shared(_load_id, tablet_id, _txn_id, _load_stream_mgr, - _profile); + int64_t partition_id, bool is_empty) { + tablet_stream = + std::make_shared(_load_id, tablet_id, _txn_id, _load_stream_mgr, _profile, + _txn_expiration, _storage_vault_id, _write_file_cache); _tablet_streams_map[tablet_id] = tablet_stream; - auto st = tablet_stream->init(_schema, _id, partition_id); + auto st = tablet_stream->init(_schema, _id, partition_id, is_empty); if (!st.ok()) { LOG(WARNING) << "tablet stream init failed " << *tablet_stream; } @@ -414,7 +427,8 @@ void IndexStream::close(const std::vector& tablets_to_commit, TabletStreamSharedPtr tablet_stream; auto it = _tablet_streams_map.find(tablet.tablet_id()); if (it == _tablet_streams_map.end()) { - _init_tablet_stream(tablet_stream, tablet.tablet_id(), tablet.partition_id()); + // A tablet first seen at close received no files from any sender. + _init_tablet_stream(tablet_stream, tablet.tablet_id(), tablet.partition_id(), true); } else { tablet_stream = it->second; } @@ -430,8 +444,28 @@ void IndexStream::close(const std::vector& tablets_to_commit, tablet_stream->pre_close(); } + const bool is_cloud = config::is_cloud_mode(); + std::vector close_statuses; + if (is_cloud) { + close_statuses.resize(_tablet_streams_map.size()); + std::vector> tasks; + tasks.reserve(_tablet_streams_map.size()); + size_t i = 0; + for (auto& [_, tablet_stream] : _tablet_streams_map) { + tasks.emplace_back([tablet_stream, &close_statuses, i] { + close_statuses[i] = tablet_stream->close(); + // A tablet failure must not stop the remaining tablets from closing. + return Status::OK(); + }); + ++i; + } + auto st = cloud::bthread_fork_join(tasks, 10); + DORIS_CHECK(st.ok()) << st; + } + + size_t i = 0; for (auto& [_, tablet_stream] : _tablet_streams_map) { - auto st = tablet_stream->close(); + auto st = is_cloud ? std::move(close_statuses[i++]) : tablet_stream->close(); if (st.ok()) { success_tablet_ids->push_back(tablet_stream->id()); } else { @@ -489,7 +523,9 @@ Status LoadStream::init(const POpenLoadStreamRequest* request) { RETURN_IF_ERROR(_schema->init(request->schema())); for (auto& index : request->schema().indexes()) { _index_streams_map[index.id()] = std::make_shared( - _load_id, index.id(), _txn_id, _schema, _load_stream_mgr, _profile.get()); + _load_id, index.id(), _txn_id, _schema, _load_stream_mgr, _profile.get(), + request->txn_expiration(), request->storage_vault_id(), + request->write_file_cache()); } LOG(INFO) << "succeed to init load stream " << *this; return Status::OK(); diff --git a/be/src/load/channel/load_stream.h b/be/src/load/channel/load_stream.h index 4c8865e160306d..22f5845bd1a0fb 100644 --- a/be/src/load/channel/load_stream.h +++ b/be/src/load/channel/load_stream.h @@ -22,6 +22,7 @@ #include #include +#include #include #include @@ -45,10 +46,11 @@ using FailedTablets = std::vector>; class TabletStream { public: TabletStream(const PUniqueId& load_id, int64_t id, int64_t txn_id, - LoadStreamMgr* load_stream_mgr, RuntimeProfile* profile); + LoadStreamMgr* load_stream_mgr, RuntimeProfile* profile, int64_t txn_expiration, + std::string storage_vault_id, bool write_file_cache); Status init(std::shared_ptr schema, int64_t index_id, - int64_t partition_id); + int64_t partition_id, bool is_empty = false); Status append_data(const PStreamHeader& header, butil::IOBuf* data); Status add_segment(const PStreamHeader& header, butil::IOBuf* data); @@ -78,6 +80,9 @@ class TabletStream { AtomicStatus _status; PUniqueId _load_id; int64_t _txn_id; + int64_t _txn_expiration; + std::string _storage_vault_id; + bool _write_file_cache; RuntimeProfile* _profile = nullptr; RuntimeProfile::Counter* _append_data_timer = nullptr; RuntimeProfile::Counter* _add_segment_timer = nullptr; @@ -91,7 +96,8 @@ class IndexStream { public: IndexStream(const PUniqueId& load_id, int64_t id, int64_t txn_id, std::shared_ptr schema, LoadStreamMgr* load_stream_mgr, - RuntimeProfile* profile); + RuntimeProfile* profile, int64_t txn_expiration, std::string storage_vault_id, + bool write_file_cache); ~IndexStream(); Status append_data(const PStreamHeader& header, butil::IOBuf* data); @@ -103,7 +109,7 @@ class IndexStream { private: void _init_tablet_stream(TabletStreamSharedPtr& tablet_stream, int64_t tablet_id, - int64_t partition_id); + int64_t partition_id, bool is_empty); private: int64_t _id; @@ -111,6 +117,9 @@ class IndexStream { bthread::Mutex _lock; PUniqueId _load_id; int64_t _txn_id; + int64_t _txn_expiration; + std::string _storage_vault_id; + bool _write_file_cache; std::shared_ptr _schema; std::unordered_map _tablet_partitions; RuntimeProfile* _profile = nullptr; diff --git a/be/src/load/channel/load_stream_writer.cpp b/be/src/load/channel/load_stream_writer.cpp index df67245ef375d7..b507c019fb3b5d 100644 --- a/be/src/load/channel/load_stream_writer.cpp +++ b/be/src/load/channel/load_stream_writer.cpp @@ -29,6 +29,7 @@ #include #include "bvar/bvar.h" +#include "cloud/cloud_rowset_builder.h" #include "cloud/config.h" #include "common/compiler_util.h" // IWYU pragma: keep #include "common/config.h" @@ -75,9 +76,13 @@ bvar::Adder g_load_stream_file_writer_cnt("load_stream_file_writer_coun LoadStreamWriter::LoadStreamWriter(WriteRequest* context, RuntimeProfile* profile) : _req(*context), _rowset_writer(nullptr) { g_load_stream_writer_cnt << 1; - // TODO(plat1ko): CloudStorageEngine - _rowset_builder = std::make_unique( - ExecEnv::GetInstance()->storage_engine().to_local(), *context, profile); + auto& engine = ExecEnv::GetInstance()->storage_engine(); + if (config::is_cloud_mode()) { + _rowset_builder = + std::make_unique(engine.to_cloud(), *context, profile); + } else { + _rowset_builder = std::make_unique(engine.to_local(), *context, profile); + } _resource_ctx = thread_context()->resource_ctx(); // from load stream } @@ -87,9 +92,15 @@ LoadStreamWriter::~LoadStreamWriter() { g_load_stream_writer_cnt << -1; } -Status LoadStreamWriter::init() { +Status LoadStreamWriter::init(bool is_empty) { DBUG_EXECUTE_IF("LoadStreamWriter.init.failure", { return Status::InternalError("fault injection"); }); + if (config::is_cloud_mode()) { + // Set this before initialization so empty tablets also skip PREPARE_ROWSET. + static_cast(_rowset_builder.get()) + ->set_skip_writing_rowset_metadata(is_empty && + config::skip_writing_empty_rowset_metadata); + } RETURN_IF_ERROR(_rowset_builder->init()); _rowset_writer = _rowset_builder->rowset_writer(); _is_init = true; @@ -178,6 +189,17 @@ Status LoadStreamWriter::close_writer(uint32_t segid, FileType file_type) { return Status::Corruption("file {} closed with 0 bytes, file type is {}", file_writer->path().native(), file_type); } + if (config::is_cloud_mode()) { + // Streamed writers are owned here, outside the RowsetWriter's file collections. + // Use the logical path used at creation, not file_writer->path(), which may be an S3 URI. + auto file_path = _rowset_writer->context().segment_path(segid); + if (file_type == FileType::INVERTED_INDEX_FILE) { + auto prefix = InvertedIndexDescriptor::get_index_file_path_prefix(file_path); + file_path = InvertedIndexDescriptor::get_index_file_path_v2(std::string(prefix)); + } + return _rowset_writer->rowset_meta()->collect_packed_slice_location(*file_writer, + file_path); + } return Status::OK(); } @@ -252,7 +274,7 @@ Status LoadStreamWriter::_pre_close() { // in same partition has data loaded. // so we have to also init this LoadStreamWriter, so that it can create an empty rowset // for this tablet when being closed. - RETURN_IF_ERROR(init()); + RETURN_IF_ERROR(init(true)); } DCHECK(_is_init) @@ -306,6 +328,26 @@ Status LoadStreamWriter::_pre_close() { } RETURN_IF_ERROR(_rowset_builder->build_rowset()); + if (config::is_cloud_mode()) { + // Forwarded files are owned here, outside CloudRowsetWriter's collections. + auto meta = _rowset_writer->rowset_meta(); + DORIS_CHECK(meta->segments_file_size().empty()); + DORIS_CHECK(meta->inverted_index_file_info().empty()); + std::vector sizes; + sizes.reserve(_segment_file_writers.size()); + for (const auto& writer : _segment_file_writers) { + sizes.push_back(writer->bytes_appended()); + } + meta->add_segments_file_size(sizes); + std::vector indexes(_inverted_file_writers.size()); + std::vector index_ptrs; + index_ptrs.reserve(indexes.size()); + for (size_t pos = 0; pos < indexes.size(); ++pos) { + indexes[pos].set_index_size(_inverted_file_writers[pos]->bytes_appended()); + index_ptrs.push_back(&indexes[pos]); + } + meta->add_inverted_index_files_info(index_ptrs); + } RETURN_IF_ERROR(_rowset_builder->submit_calc_delete_bitmap_task()); _pre_closed = true; return Status::OK(); @@ -317,9 +359,7 @@ Status LoadStreamWriter::close() { RETURN_IF_ERROR(_pre_close()); } RETURN_IF_ERROR(_rowset_builder->wait_calc_delete_bitmap()); - // FIXME(plat1ko): No `commit_txn` operation in cloud mode, need better abstractions - RETURN_IF_ERROR(static_cast(_rowset_builder.get())->commit_txn()); - + RETURN_IF_ERROR(_rowset_builder->commit_txn()); return Status::OK(); } diff --git a/be/src/load/channel/load_stream_writer.h b/be/src/load/channel/load_stream_writer.h index e82497ec02bfa1..8c4f4d79109e88 100644 --- a/be/src/load/channel/load_stream_writer.h +++ b/be/src/load/channel/load_stream_writer.h @@ -51,7 +51,8 @@ class LoadStreamWriter { ~LoadStreamWriter(); - Status init(); + // is_empty is true for tablets first seen when closing the load. + Status init(bool is_empty = false); Status append_data(uint32_t segid, uint64_t offset, butil::IOBuf buf, FileType file_type = FileType::SEGMENT_FILE); diff --git a/be/src/storage/rowset/beta_rowset_writer.cpp b/be/src/storage/rowset/beta_rowset_writer.cpp index 265448a831472b..8c0101594c9132 100644 --- a/be/src/storage/rowset/beta_rowset_writer.cpp +++ b/be/src/storage/rowset/beta_rowset_writer.cpp @@ -426,13 +426,20 @@ Status BaseBetaRowsetWriter::_generate_delete_bitmap(int32_t segment_id) { _context.tablet->get_rowset_by_ids(_context.mow_context->rowset_ids.get()); } + RowsetSharedPtr streamed_rowset; + if (_seg_files.get(segment_id) == nullptr) { + // Streamed files are closed by LoadStreamWriter. Snapshot their packed mappings on + // its serial receive thread before later file closes mutate the shared rowset meta. + RETURN_IF_ERROR(_build_tmp(streamed_rowset, segment_id)); + } + // Submit the entire delete bitmap calculation process to thread pool for async execution // This avoids blocking memtable flush thread while waiting for file upload to complete // The process includes: file_writer->close(), _build_tmp, load_segments, and calc_delete_bitmap const auto submit_time_us = MonotonicMicros(); return _calc_delete_bitmap_token->submit_func([this, segment_id, specified_rowsets = std::move(specified_rowsets), - submit_time_us]() -> Status { + submit_time_us, streamed_rowset]() -> Status { const auto queue_time_us = MonotonicMicros() - submit_time_us; Status st = Status::OK(); // Step 1: Close file_writer (must be done before load_segments) @@ -457,10 +464,9 @@ Status BaseBetaRowsetWriter::_generate_delete_bitmap(int32_t segment_id) { OlapStopWatch watch; // Step 2: Build tmp rowset (needs file_writer to be closed) - RowsetSharedPtr rowset_ptr; - st = _build_tmp(rowset_ptr); - if (!st.ok()) { - return st; + RowsetSharedPtr rowset_ptr = streamed_rowset; + if (rowset_ptr == nullptr) { + RETURN_IF_ERROR(_build_tmp(rowset_ptr, segment_id)); } // Step 3: Load segments (needs file_writer to be closed and rowset to be built) @@ -1156,7 +1162,7 @@ Status BaseBetaRowsetWriter::_build_rowset_meta(RowsetMeta* rowset_meta, bool ch return Status::OK(); } -Status BaseBetaRowsetWriter::_build_tmp(RowsetSharedPtr& rowset_ptr) { +Status BaseBetaRowsetWriter::_build_tmp(RowsetSharedPtr& rowset_ptr, int32_t segment_id) { Status status; std::shared_ptr tmp_rs_meta = std::make_shared(); tmp_rs_meta->init(_rowset_meta.get()); @@ -1169,6 +1175,15 @@ Status BaseBetaRowsetWriter::_build_tmp(RowsetSharedPtr& rowset_ptr) { } tmp_rs_meta->set_segment_ids(completed_segment_ids); + if (_context.packed_file_active) { + // Bitmap calculation reads only this segment's primary-key index. Other flushes + // may still be creating/closing files, so do not traverse their writer collections. + if (auto* writer = _seg_files.get(segment_id); writer != nullptr) { + RETURN_IF_ERROR(tmp_rs_meta->collect_packed_slice_location( + *writer, _context.segment_path(segment_id))); + } + } + status = RowsetFactory::create_rowset(_context.tablet_schema, _context.tablet_path, tmp_rs_meta, &rowset_ptr); DBUG_EXECUTE_IF("BaseBetaRowsetWriter::_build_tmp.create_rowset_failed", diff --git a/be/src/storage/rowset/beta_rowset_writer.h b/be/src/storage/rowset/beta_rowset_writer.h index 49811ac7714512..918edb057f0bdc 100644 --- a/be/src/storage/rowset/beta_rowset_writer.h +++ b/be/src/storage/rowset/beta_rowset_writer.h @@ -224,7 +224,7 @@ class BaseBetaRowsetWriter : public RowsetWriter { virtual Status _check_segment_number_limit(size_t segnum); virtual int64_t _num_seg() const; // build a tmp rowset for load segment to calc delete_bitmap for this segment - Status _build_tmp(RowsetSharedPtr& rowset_ptr); + Status _build_tmp(RowsetSharedPtr& rowset_ptr, int32_t segment_id); uint64_t get_rowset_num_rows() { std::lock_guard l(_segid_statistics_map_mutex); diff --git a/be/src/storage/rowset/rowset_meta.cpp b/be/src/storage/rowset/rowset_meta.cpp index 05ca4be46b7b10..123813c30c11d3 100644 --- a/be/src/storage/rowset/rowset_meta.cpp +++ b/be/src/storage/rowset/rowset_meta.cpp @@ -37,6 +37,7 @@ #include "io/fs/local_file_system.h" #include "io/fs/packed_file_manager.h" #include "io/fs/packed_file_system.h" +#include "io/fs/packed_file_writer.h" #include "json2pb/json_to_pb.h" #include "json2pb/pb_to_json.h" #include "runtime/exec_env.h" @@ -56,6 +57,27 @@ RowsetMeta::~RowsetMeta() { } } +Status RowsetMeta::collect_packed_slice_location(const io::FileWriter& file_writer, + const std::string& file_path) { + if (file_writer.state() != io::FileWriter::State::CLOSED) { + return Status::OK(); + } + if (!file_writer.is_in_packed_file()) { + return Status::OK(); + } + + // Read the writer-owned location so collection cannot race with global index cleanup. + io::PackedSliceLocation location; + RETURN_IF_ERROR(static_cast(file_writer) + .get_packed_slice_location(&location)); + DORIS_CHECK(!location.packed_file_path.empty()); + add_packed_slice_location(file_path, location.packed_file_path, location.offset, location.size, + location.packed_file_size); + LOG(INFO) << "collect packed file index: " << file_path << " -> " << location.packed_file_path + << ", offset: " << location.offset << ", size: " << location.size; + return Status::OK(); +} + bool RowsetMeta::init(std::string_view pb_rowset_meta) { bool ret = _deserialize_from_pb(pb_rowset_meta); if (!ret) { diff --git a/be/src/storage/rowset/rowset_meta.h b/be/src/storage/rowset/rowset_meta.h index 9255cdb9a09d08..d140e4d891e2f6 100644 --- a/be/src/storage/rowset/rowset_meta.h +++ b/be/src/storage/rowset/rowset_meta.h @@ -535,6 +535,11 @@ class RowsetMeta : public MetadataAdder { RowsetMeta(const RowsetMeta&) = delete; RowsetMeta operator=(const RowsetMeta&) = delete; + // Collect only completed files; interim rowsets may still have open writers. + // file_path is the logical path used at creation, not the S3 writer's absolute path. + Status collect_packed_slice_location(const io::FileWriter& file_writer, + const std::string& file_path); + void add_packed_slice_location(const std::string& segment_path, const std::string& packed_file_path, int64_t offset, int64_t size, int64_t packed_file_size) { diff --git a/be/test/cloud/cloud_load_stream_test.cpp b/be/test/cloud/cloud_load_stream_test.cpp new file mode 100644 index 00000000000000..884249f6f7d6cc --- /dev/null +++ b/be/test/cloud/cloud_load_stream_test.cpp @@ -0,0 +1,291 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "cloud/cloud_committed_rs_mgr.h" +#include "cloud/cloud_rowset_builder.h" +#include "cloud/cloud_rowset_writer.h" +#include "cloud/cloud_storage_engine.h" +#include "cloud/cloud_tablet.h" +#include "cloud/cloud_txn_delete_bitmap_cache.h" +#include "cloud/config.h" +#include "load/channel/load_stream_mgr.h" +#include "load/channel/load_stream_writer.h" +#include "runtime/exec_env.h" +#include "storage/rowset/rowset.h" +#include "storage/rowset/rowset_factory.h" +#include "storage/rowset/rowset_meta.h" +#include "storage/tablet/tablet_meta.h" +#include "util/time.h" +#include "util/work_thread_pool.hpp" + +namespace doris { +namespace { + +class EmptyLoadRowsetBuilder : public CloudRowsetBuilder { +public: + using CloudRowsetBuilder::CloudRowsetBuilder; + + Status init() override { + skip_metadata_at_init = _skip_writing_rowset_metadata; + return Status::InternalError("stop before preparing rowset metadata"); + } + + Status commit_rowset(const std::string&, int64_t) override { + ++commit_calls; + return Status::InternalError("injected commit failure"); + } + + bool skip_metadata_at_init = false; + int commit_calls = 0; +}; + +class ClosingLoadRowsetBuilder : public BaseRowsetBuilder { +public: + ClosingLoadRowsetBuilder(CloudStorageEngine& engine, const WriteRequest& req, + std::atomic& prepared, std::function commit) + : BaseRowsetBuilder(req, nullptr), _prepared(prepared), _commit(std::move(commit)) { + _rowset_writer = std::make_shared(engine); + _rowset_writer->_rowset_meta = std::make_shared(); + } + + Status init() override { return Status::OK(); } + Status build_rowset() override { + ++_prepared; + return Status::OK(); + } + Status submit_calc_delete_bitmap_task() override { return Status::OK(); } + Status wait_calc_delete_bitmap() override { return Status::OK(); } + Status commit_txn() override { return _commit(); } + +private: + std::atomic& _prepared; + std::function _commit; +}; + +} // namespace + +class CloudLoadStreamTest : public testing::Test { +protected: + void SetUp() override { + _old_cloud_unique_id = config::cloud_unique_id; + _old_skip_empty = config::skip_writing_empty_rowset_metadata; + _old_make_visible = config::enable_cloud_make_rs_visible_on_be; + config::cloud_unique_id = "cloud_load_stream_test"; + config::enable_cloud_make_rs_visible_on_be = true; + _old_engine = std::move(ExecEnv::GetInstance()->_storage_engine); + auto engine = std::make_unique(EngineOptions {}); + _engine = engine.get(); + ExecEnv::GetInstance()->set_storage_engine(std::move(engine)); + _engine->_committed_rs_mgr = std::make_unique(); + _engine->_txn_delete_bitmap_cache = std::make_unique(1024); + ASSERT_TRUE(_engine->_txn_delete_bitmap_cache->init().ok()); + } + + void TearDown() override { + ExecEnv::GetInstance()->set_storage_engine(std::move(_old_engine)); + config::cloud_unique_id = _old_cloud_unique_id; + config::skip_writing_empty_rowset_metadata = _old_skip_empty; + config::enable_cloud_make_rs_visible_on_be = _old_make_visible; + } + + CloudTabletSPtr create_tablet(KeysType keys_type, bool mow) { + TabletMetaPB meta; + meta.set_tablet_id(10001); + meta.set_table_id(10002); + meta.set_enable_unique_key_merge_on_write(mow); + meta.mutable_schema()->set_keys_type(keys_type); + auto tablet_meta = std::make_shared(); + tablet_meta->init_from_pb(meta); + auto tablet = std::make_shared(*_engine, tablet_meta); + std::unique_lock lock(tablet->get_header_lock()); + tablet->reset_approximate_stats(0, 0, 0, 0); + return tablet; + } + + CloudStorageEngine* _engine = nullptr; + std::unique_ptr _old_engine; + std::string _old_cloud_unique_id; + bool _old_skip_empty = false; + bool _old_make_visible = false; +}; + +TEST_F(CloudLoadStreamTest, SetEmptyPolicyBeforePreparingMetadata) { + for (bool skip_empty : {false, true}) { + config::skip_writing_empty_rowset_metadata = skip_empty; + for (bool is_empty : {false, true}) { + WriteRequest req; + LoadStreamWriter writer(&req, nullptr); + auto builder = std::make_unique(*_engine, req, nullptr); + auto* observed_builder = builder.get(); + writer._rowset_builder = std::move(builder); + auto st = writer.init(is_empty); + ASSERT_FALSE(st.ok()); + EXPECT_NE(st.to_string().find("stop before preparing"), std::string::npos); + EXPECT_EQ(observed_builder->skip_metadata_at_init, skip_empty && is_empty); + } + } +} + +TEST_F(CloudLoadStreamTest, EmptyCommitSkipsRpcAndRegistersMarkers) { + for (auto keys_type : {DUP_KEYS, AGG_KEYS, UNIQUE_KEYS}) { + for (bool mow : {false, true}) { + if (mow && keys_type != UNIQUE_KEYS) { + continue; + } + WriteRequest req; + req.tablet_id = 10001; + req.txn_id = 100 + keys_type * 2 + mow; + req.txn_expiration = UnixSeconds() + 3600; + EmptyLoadRowsetBuilder builder(*_engine, req, nullptr); + auto tablet = create_tablet(keys_type, mow); + builder._tablet = tablet; + builder._tablet_schema = tablet->tablet_schema(); + auto meta = std::make_shared(); + meta->set_rowset_type(BETA_ROWSET); + meta->set_rowset_state(PREPARED); + meta->set_num_segments(0); + ASSERT_TRUE( + RowsetFactory::create_rowset(builder._tablet_schema, "", meta, &builder._rowset) + .ok()); + builder.set_skip_writing_rowset_metadata(true); + + auto st = builder.commit_txn(); + ASSERT_TRUE(st.ok()) << st.to_string(); + EXPECT_EQ(builder.commit_calls, 0); + EXPECT_TRUE(builder._is_committed); + EXPECT_EQ(tablet->fetch_add_approximate_num_rowsets(0), 1); + if (mow) { + EXPECT_TRUE(_engine->txn_delete_bitmap_cache().is_empty_rowset(req.txn_id, + req.tablet_id)); + } else { + auto marker = + _engine->committed_rs_mgr().get_committed_rowset(req.txn_id, req.tablet_id); + ASSERT_TRUE(marker.has_value()); + EXPECT_EQ(marker->first, nullptr); + EXPECT_GE(marker->second, req.txn_expiration); + } + } + } +} + +TEST_F(CloudLoadStreamTest, MetadataCommitFailureDoesNotRegisterMarker) { + WriteRequest req; + req.tablet_id = 10001; + req.txn_id = 200; + EmptyLoadRowsetBuilder builder(*_engine, req, nullptr); + auto tablet = create_tablet(DUP_KEYS, false); + builder._tablet = tablet; + builder.set_skip_writing_rowset_metadata(false); + + auto st = builder.commit_txn(); + ASSERT_FALSE(st.ok()); + EXPECT_NE(st.to_string().find("injected commit failure"), std::string::npos); + EXPECT_EQ(builder.commit_calls, 1); + EXPECT_FALSE(builder._is_committed); + EXPECT_EQ(tablet->fetch_add_approximate_num_rowsets(0), 0); + EXPECT_FALSE(_engine->committed_rs_mgr() + .get_committed_rowset(req.txn_id, req.tablet_id) + .has_value()); +} + +TEST_F(CloudLoadStreamTest, CloseTabletsConcurrentlyAndCollectAllResults) { + constexpr int tablet_count = 25; + FifoThreadPool pool(16, 64, "CloudLoadStreamCloseTest"); + LoadStreamMgr manager(1); + manager.set_heavy_work_pool(&pool); + RuntimeProfile profile("CloudLoadStreamCloseTest"); + PUniqueId load_id; + IndexStream index(load_id, 1, 1, nullptr, &manager, &profile, UnixSeconds() + 3600, "", false); + + std::atomic prepared {0}; + std::mutex mutex; + std::condition_variable cv; + int started = 0; + int active = 0; + int peak_active = 0; + bool timed_out = false; + for (int64_t tablet_id = 1; tablet_id <= tablet_count; ++tablet_id) { + auto tablet = std::make_shared(load_id, tablet_id, 1, &manager, &profile, + UnixSeconds() + 3600, "", false); + WriteRequest req; + req.tablet_id = tablet_id; + auto writer = std::make_shared(&req, nullptr); + writer->_rowset_builder = + std::make_unique(*_engine, req, prepared, [&, tablet_id] { + EXPECT_EQ(prepared.load(), tablet_count); + std::unique_lock lock(mutex); + ++started; + ++active; + peak_active = std::max(peak_active, active); + cv.notify_all(); + // Hold the first wave until ten tablets are closing concurrently. + // A serial implementation times out once instead of hanging the test. + if (!cv.wait_for(lock, std::chrono::seconds(5), + [&] { return started >= 10 || timed_out; })) { + timed_out = true; + cv.notify_all(); + } + --active; + return tablet_id % 2 == 0 + ? Status::OK() + : Status::InternalError("injected tablet close failure"); + }); + writer->_rowset_writer = writer->_rowset_builder->rowset_writer(); + writer->_is_init = true; + tablet->_load_stream_writer = std::move(writer); + index._tablet_streams_map.emplace(tablet_id, std::move(tablet)); + } + + std::vector success_tablets; + FailedTablets failed_tablets; + index.close({}, &success_tablets, &failed_tablets); + + EXPECT_FALSE(timed_out); + EXPECT_EQ(peak_active, 10); + EXPECT_EQ(started, tablet_count); + std::sort(success_tablets.begin(), success_tablets.end()); + std::vector expected_success; + for (int64_t id = 2; id <= tablet_count; id += 2) { + expected_success.push_back(id); + } + EXPECT_EQ(success_tablets, expected_success); + std::vector failed_ids; + for (const auto& [id, st] : failed_tablets) { + failed_ids.push_back(id); + EXPECT_NE(st.to_string().find("injected tablet close failure"), std::string::npos); + } + std::sort(failed_ids.begin(), failed_ids.end()); + std::vector expected_failed; + for (int64_t id = 1; id <= tablet_count; id += 2) { + expected_failed.push_back(id); + } + EXPECT_EQ(failed_ids, expected_failed); +} + +} // namespace doris diff --git a/be/test/storage/rowset/beta_rowset_test.cpp b/be/test/storage/rowset/beta_rowset_test.cpp index c7337ff4f9bfc6..7f6d39d0fa7497 100644 --- a/be/test/storage/rowset/beta_rowset_test.cpp +++ b/be/test/storage/rowset/beta_rowset_test.cpp @@ -60,6 +60,8 @@ #include "storage/segment/segment.h" #include "storage/storage_engine.h" #include "storage/storage_policy.h" +#include "storage/tablet/tablet.h" +#include "storage/tablet/tablet_meta.h" #include "storage/tablet/tablet_schema.h" #include "storage/utils.h" #include "util/s3_util.h" @@ -239,9 +241,44 @@ class BetaRowsetWriterForTest : public BetaRowsetWriter { public: explicit BetaRowsetWriterForTest(StorageEngine& engine) : BetaRowsetWriter(engine) {} - Status build_tmp(RowsetSharedPtr& rowset) { return _build_tmp(rowset); } + Status build_tmp(RowsetSharedPtr& rowset, int32_t segment_id) { + return _build_tmp(rowset, segment_id); + } }; +TEST_F(BetaRowsetTest, StreamedBitmapSnapshotFailurePreventsQueueing) { + class SnapshotFailureWriter : public BetaRowsetWriter { + public: + explicit SnapshotFailureWriter(StorageEngine& engine) : BetaRowsetWriter(engine) {} + + protected: + Status _build_rowset_meta(RowsetMeta*, bool, std::vector*) override { + return Status::InternalError("streamed metadata snapshot failed"); + } + }; + + EngineOptions options; + StorageEngine engine(options); + auto tablet_meta = std::make_shared(); + tablet_meta->set_enable_unique_key_merge_on_write(true); + SnapshotFailureWriter writer(engine); + auto tablet_schema = std::make_shared(); + create_tablet_schema(tablet_schema); + RowsetWriterContext writer_context; + create_rowset_writer_context(tablet_schema, &writer_context); + ASSERT_TRUE(writer.init(writer_context).ok()); + writer._context.tablet = std::make_shared(engine, tablet_meta, nullptr); + writer._context.mow_context = std::make_shared( + 0, 1, std::make_shared(), std::vector {}, + std::make_shared(0)); + // No owned file writer: LoadStreamWriter has already closed the streamed file. + // Snapshot errors must be returned before touching the asynchronous executor. + ASSERT_EQ(writer._calc_delete_bitmap_token, nullptr); + auto st = writer._generate_delete_bitmap(0); + EXPECT_FALSE(st.ok()); + EXPECT_NE(st.to_string().find("streamed metadata snapshot failed"), std::string::npos); +} + class S3ClientMock : public Aws::S3::S3Client { S3ClientMock() {} S3ClientMock(const Aws::Auth::AWSCredentials& credentials, @@ -533,7 +570,7 @@ TEST_F(BetaRowsetTest, TmpRowsetUsesCompletedSegmentIds) { ASSERT_TRUE(writer.add_segment(2, segment_statistics).ok()); RowsetSharedPtr tmp_rowset; - ASSERT_TRUE(writer.build_tmp(tmp_rowset).ok()); + ASSERT_TRUE(writer.build_tmp(tmp_rowset, 6).ok()); ASSERT_NE(tmp_rowset, nullptr); EXPECT_EQ(tmp_rowset->num_segments(), 2); EXPECT_EQ(tmp_rowset->rowset_meta()->position_of(2), 0); diff --git a/be/test/storage/rowset/rowset_meta_test.cpp b/be/test/storage/rowset/rowset_meta_test.cpp index 633c4f4999abed..b8a415e574a0bc 100644 --- a/be/test/storage/rowset/rowset_meta_test.cpp +++ b/be/test/storage/rowset/rowset_meta_test.cpp @@ -25,15 +25,19 @@ #include #include +#include #include #include +#include "common/exception.h" #include "common/status.h" #include "cpp/sync_point.h" #include "gtest/gtest_pred_impl.h" #include "io/fs/file_reader.h" #include "io/fs/file_system.h" +#include "io/fs/file_writer.h" #include "io/fs/local_file_system.h" +#include "io/fs/packed_file_writer.h" #include "storage/olap_common.h" #include "storage/olap_meta.h" #include "storage/tablet/tablet_schema.h" @@ -47,6 +51,98 @@ using std::string; namespace doris { +namespace { + +class PackedLocationWriter : public io::FileWriter { +public: + explicit PackedLocationWriter(std::string path) : _path(std::move(path)) {} + + Status close(bool non_block = false) override { + _state = non_block ? State::ASYNC_CLOSING : State::CLOSED; + return Status::OK(); + } + Status appendv(const Slice*, size_t) override { return Status::NotSupported("test writer"); } + const io::Path& path() const override { return _path; } + size_t bytes_appended() const override { return 10; } + State state() const override { return _state; } + +private: + io::Path _path; + State _state = State::OPENED; +}; + +} // namespace + +TEST(RowsetMetaPackedFileTest, CollectCompletedSegmentAndIndex) { + RowsetMeta meta; + RowsetId rowset_id; + rowset_id.init(1); + meta.set_rowset_id(rowset_id); + // Use destination paths with a nonzero segment ID, including the V2 index file. + for (const auto* suffix : {".dat", ".idx"}) { + std::string path = std::string("data/123/packed_location_test_7") + suffix; + io::PackedFileWriter writer( + std::make_unique("s3://bucket/prefix/" + path), path); + io::PackedSliceLocation location; + location.packed_file_path = "data/packed_file/test.bin"; + location.offset = 20; + location.size = 10; + location.packed_file_size = 100; + // The writer's handle remains sufficient after the manager's index is gone. + writer._packed_slice_handle = std::make_shared(location); + writer._packed_slice_handle->set_upload_result(io::PackedSliceUploadState::UPLOADED, + location.packed_file_size); + + ASSERT_TRUE(meta.collect_packed_slice_location(writer, path).ok()); + EXPECT_EQ(meta.get_rowset_pb().packed_slice_locations().count(path), 0); + writer._state = io::FileWriter::State::ASYNC_CLOSING; + ASSERT_TRUE(meta.collect_packed_slice_location(writer, path).ok()); + EXPECT_EQ(meta.get_rowset_pb().packed_slice_locations().count(path), 0); + writer._state = io::FileWriter::State::CLOSED; + ASSERT_TRUE(meta.collect_packed_slice_location(writer, path).ok()); + // Repeated collection must not duplicate the mapping. + ASSERT_TRUE(meta.collect_packed_slice_location(writer, path).ok()); + auto pb = meta.get_rowset_pb(); + const auto& stored = pb.packed_slice_locations().at(path); + EXPECT_EQ(stored.packed_file_path(), location.packed_file_path); + EXPECT_EQ(stored.offset(), location.offset); + EXPECT_EQ(stored.size(), location.size); + EXPECT_EQ(stored.packed_file_size(), location.packed_file_size); + } + RowsetMeta restored; + std::string serialized; + ASSERT_TRUE(meta.serialize(&serialized)); + ASSERT_TRUE(restored.init(serialized)); + EXPECT_EQ(restored.get_rowset_pb().packed_slice_locations_size(), 2); +} + +TEST(RowsetMetaPackedFileTest, DirectFileNeedsNoPackedLocation) { + RowsetMeta meta; + PackedLocationWriter writer("direct_file.dat"); + ASSERT_TRUE(writer.close().ok()); + EXPECT_TRUE(meta.collect_packed_slice_location(writer, "direct_file.dat").ok()); + EXPECT_EQ(meta.get_rowset_pb().packed_slice_locations_size(), 0); +} + +TEST(RowsetMetaPackedFileTest, MissingPackedLocationFails) { + RowsetMeta meta; + io::PackedFileWriter writer( + std::make_unique("missing_packed_location.dat"), + "missing_packed_location.dat"); + writer._state = io::FileWriter::State::CLOSED; +#ifndef NDEBUG + GTEST_FLAG_SET(death_test_style, "threadsafe"); + EXPECT_DEATH(static_cast( + meta.collect_packed_slice_location(writer, "missing_packed_location.dat")), + "location.packed_file_path.empty"); +#else + EXPECT_THROW(static_cast( + meta.collect_packed_slice_location(writer, "missing_packed_location.dat")), + Exception); +#endif + EXPECT_EQ(meta.get_rowset_pb().packed_slice_locations_size(), 0); +} + const std::string rowset_meta_path = "./be/test/storage/test_data/rowset.json"; class RowsetMetaTest : public testing::Test { diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java index 3c43e26f971048..99cd1439fb64e2 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java +++ b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java @@ -379,7 +379,7 @@ private void createLoadingTask(Database db, BrokerPendingTaskAttachment attachme } } } - if (isPartialUpdate() || hasInvertedIndexV1 || Config.isCloudMode()) { + if (isPartialUpdate() || hasInvertedIndexV1) { isEnableMemtableOnSinkNode = false; } diff --git a/gensrc/proto/internal_service.proto b/gensrc/proto/internal_service.proto index 55c64d302a03a5..9d9bd1a496a2f9 100644 --- a/gensrc/proto/internal_service.proto +++ b/gensrc/proto/internal_service.proto @@ -1020,6 +1020,9 @@ message POpenLoadStreamRequest { optional bool enable_profile = 6 [default = false]; optional int64 total_streams = 7; optional int64 idle_timeout_ms = 8; + optional int64 txn_expiration = 9; + optional string storage_vault_id = 10; + optional bool write_file_cache = 11; } message PTabletSchemaWithIndex { diff --git a/regression-test/data/cloud_p0/test_cloud_memtable_mow_forward.out b/regression-test/data/cloud_p0/test_cloud_memtable_mow_forward.out new file mode 100644 index 00000000000000..6f0e9ba4e9beb3 --- /dev/null +++ b/regression-test/data/cloud_p0/test_cloud_memtable_mow_forward.out @@ -0,0 +1,109 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !initial_false -- +0 119960 11997 +1 119970 11998 +2 119980 11999 +3 119990 12000 + +-- !index_false -- +1 119970 11998 +3 119990 12000 + +-- !layout_false -- +true false true true true true + +-- !deleted_false -- +0 119960 11997 +1 119970 11998 +3 119990 12000 + +-- !stream_status_false -- +Success + +-- !stream_false -- +0 150000 15000 +1 119970 11998 +2 140000 14000 +3 119990 12000 + +-- !group_status_false -- +Success true + +-- !group_false -- +0 150000 15000 +1 160000 16000 +2 140000 14000 +3 119990 12000 + +-- !compacted_false -- +0 150000 15000 +1 160000 16000 +2 140000 14000 +3 119990 12000 + +-- !initial_true -- +0 119960 11997 +1 119970 11998 +2 119980 11999 +3 119990 12000 + +-- !index_true -- +1 119970 11998 +3 119990 12000 + +-- !layout_true -- +true true true true true true + +-- !deleted_true -- +0 119960 11997 +1 119970 11998 +3 119990 12000 + +-- !stream_status_true -- +Success + +-- !stream_true -- +0 150000 15000 +1 119970 11998 +2 140000 14000 +3 119990 12000 + +-- !group_status_true -- +Success true + +-- !group_true -- +0 150000 15000 +1 160000 16000 +2 140000 14000 +3 119990 12000 + +-- !compacted_true -- +0 150000 15000 +1 160000 16000 +2 140000 14000 +3 119990 12000 + +-- !mow_fallback -- +1 20 + +-- !broker_rows -- +20 40 +21 42 +31 62 +41 82 +49 98 +50 100 +57 114 +58 116 +60 120 +62 124 +65 130 +66 132 +68 136 +8 16 +80 160 +81 162 +85 170 +90 180 +91 182 + diff --git a/regression-test/data/cloud_p0/test_cloud_memtable_on_sink.out b/regression-test/data/cloud_p0/test_cloud_memtable_on_sink.out new file mode 100644 index 00000000000000..6948612c886f8f --- /dev/null +++ b/regression-test/data/cloud_p0/test_cloud_memtable_on_sink.out @@ -0,0 +1,148 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !s3_rows_false -- +20 1133 2266 + +-- !s3_index_false -- +31 62 +50 100 +50 100 +57 114 + +-- !s3_packed_meta_false -- +6 20 2 1 1 0 0,1,2,3,4,5 + +-- !stream_rows -- +22 1334 2668 + +-- !agg_false -- +0 17994000 0 11996 3000 1 10 20 +1 17997000 1 11997 3000 1 11 21 +2 18000000 2 11998 3000 1 12 22 +3 18003000 3 11999 3000 1 13 23 + +-- !mor_false -- +0 0 +1 10 +2 20 +3 30 + +-- !seq_false -- +0 119960 11996 +1 119970 11997 +2 119980 11998 +3 119990 11999 + +-- !seq_index_false -- +1 119970 11997 +3 119990 11999 + +-- !seq_lower_false -- +0 119960 11996 +1 119970 11997 +2 119980 11998 +3 119990 11999 + +-- !mor_new_version_false -- +0 99 +1 10 +2 20 +3 30 + +-- !agg_new_version_false -- +0 17994010 -1 13000 3000 1 99 20 +1 17997000 1 11997 3000 1 11 21 +2 18000000 2 11998 3000 1 12 22 +3 18003000 3 11999 3000 1 13 23 + +-- !seq_empty_false -- +0 119960 11996 +1 119970 11997 +2 119980 11998 +3 119990 11999 + +-- !agg_compacted_false -- +0 17994010 -1 13000 3000 1 99 20 +1 17997000 1 11997 3000 1 11 21 +2 18000000 2 11998 3000 1 12 22 +3 18003000 3 11999 3000 1 13 23 + +-- !mor_compacted_false -- +0 99 +1 10 +2 20 +3 30 + +-- !seq_compacted_false -- +0 119960 11996 +1 119970 11997 +2 119980 11998 +3 119990 11999 + +-- !agg_broker_false -- +20 40 +21 42 +31 62 +41 82 +49 98 +50 200 +57 114 +58 116 +60 120 +62 124 +65 130 +66 132 +68 136 +8 16 +80 160 +81 162 +85 170 +90 180 +91 182 + +-- !mor_broker_false -- +20 40 +21 42 +31 62 +41 82 +49 98 +50 100 +57 114 +58 116 +60 120 +62 124 +65 130 +66 132 +68 136 +8 16 +80 160 +81 162 +85 170 +90 180 +91 182 + +-- !agg_stream -- +50 203 + +-- !mor_stream -- +0 119960 11996 +1 130010 13001 +2 119980 11998 +3 119990 11999 + +-- !mor_delete -- +0 119960 11996 +1 130010 13001 +3 119990 11999 + +-- !mor_reinsert -- +0 119960 11996 +1 130010 13001 +2 150000 15000 +3 119990 11999 + +-- !mor_final_compacted -- +0 119960 11996 +1 130010 13001 +2 150000 15000 +3 119990 11999 + diff --git a/regression-test/suites/cloud_p0/test_cloud_memtable_mow_forward.groovy b/regression-test/suites/cloud_p0/test_cloud_memtable_mow_forward.groovy new file mode 100644 index 00000000000000..1b84ca7214a838 --- /dev/null +++ b/regression-test/suites/cloud_p0/test_cloud_memtable_mow_forward.groovy @@ -0,0 +1,255 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import org.apache.doris.regression.action.ProfileAction +import org.apache.doris.regression.suite.ClusterOptions +import org.apache.doris.regression.suite.client.BackendClientImpl +import org.apache.doris.thrift.TGetRealtimeExecStatusRequest +import org.apache.doris.thrift.TNetworkAddress +import org.apache.doris.thrift.TStatusCode +import org.apache.doris.thrift.TUniqueId + +suite("test_cloud_memtable_mow_forward", "p0, docker") { + def options = new ClusterOptions() + options.cloudMode = true + options.setFeNum(1) + options.setBeNum(3) + options.enableDebugPoints() + options.beConfigs += ['small_file_threshold_bytes=1048576', + 'enable_merge_on_write_correctness_check=true'] + options.feConfigs += ['stream_load_default_memtable_on_sink_node=true'] + docker(options) { + def backends = sql_return_maparray("SHOW BACKENDS") + def awaitGroupProfile = { label, tabletId -> + def idParts = label.substring('group_commit_'.length()).split('_') + def queryId = new TUniqueId(Long.parseUnsignedLong(idParts[0], 16), + Long.parseUnsignedLong(idParts[1], 16)) + def request = new TGetRealtimeExecStatusRequest().setId(queryId).setReqType("profile") + awaitUntil(60) { + backends.any { be -> + def backend = new BackendClientImpl( + new TNetworkAddress(be.Host, be.BePort as int), be.HttpPort as int) + try { + backend.client.inputProtocol.transport.setTimeout(5000) + def response = backend.client.getRealtimeExecStatus(request) + if (response.status.statusCode == TStatusCode.NOT_FOUND) { + return false + } + if (response.status.statusCode != TStatusCode.OK) { + throw new IllegalStateException("Get BE profile failed: ${response.status}") + } + def nodes = response.reportExecStatusParams.queryProfile.fragmentIdToProfile + .values().flatten().collectMany { it.profile.nodes } + return nodes.any { it.name == "DeltaWriterV2 ${tabletId}" } + } finally { + backend.close() + } + } + } + } + def checkUniqueKeys = { + sql """ + SELECT assert_true(COUNT(*) = 0, 'cloud_mow_forward contains duplicate keys') FROM ( + SELECT k, COUNT(*) AS a FROM cloud_mow_forward GROUP BY k HAVING a > 1 + ) duplicates + """ + } + + sql "DROP TABLE IF EXISTS cloud_mow_forward_source" + sql """ + CREATE TABLE cloud_mow_forward_source (n BIGINT NOT NULL) + DUPLICATE KEY(n) DISTRIBUTED BY HASH(n) BUCKETS 12 + PROPERTIES ("replication_num"="1") + """ + sql "SET enable_memtable_on_sink_node=false" + sql "INSERT INTO cloud_mow_forward_source SELECT number FROM numbers('number'='12000')" + sql "SET parallel_pipeline_task_num=4" + sql "SET profile_level=2" + sql "SET enable_sql_cache=false" + sql "SET enable_file_cache=false" + [false, true].each { packed -> + setBeConfigTemporary(['enable_packed_file': packed.toString()]) { + sql "DROP TABLE IF EXISTS cloud_mow_forward" + sql """ + CREATE TABLE cloud_mow_forward ( + k BIGINT NOT NULL, v BIGINT, seq BIGINT NOT NULL, + INDEX idx_k(k) USING INVERTED + ) UNIQUE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ("replication_num"="1", "enable_unique_key_merge_on_write"="true", + "function_column.sequence_col"="seq", "inverted_index_storage_format"="V2", + "disable_auto_compaction"="true", "group_commit_interval_ms"="200") + """ + sql "SET enable_memtable_on_sink_node=false" + sql "INSERT INTO cloud_mow_forward VALUES (0,-1,-1),(1,-1,-1),(2,-1,-1),(3,-1,-1)" + sql "INSERT INTO cloud_mow_forward VALUES (0,0,0),(1,0,0),(2,0,0),(3,0,0)" + sql "SET enable_memtable_on_sink_node=true" + sql "SET enable_profile=true" + sql """ + /* cloud_mow_forward_${packed} */ + INSERT INTO cloud_mow_forward SELECT n%4,n*10,n+1 FROM cloud_mow_forward_source + """ + new ProfileAction(context).getProfileBySql("cloud_mow_forward_${packed}", ["DeltaWriterV2"]) + sql "SET enable_profile=false" + quickTest("initial_${packed}", "SELECT * FROM cloud_mow_forward", true) + checkUniqueKeys() + quickTest("index_${packed}", "SELECT * FROM cloud_mow_forward WHERE k IN (1,3)", true) + def tablet = sql_return_maparray("SHOW TABLETS FROM cloud_mow_forward")[0] + def partition = sql_return_maparray("SHOW PARTITIONS FROM cloud_mow_forward")[0] + def ms = cluster.getAllMetaservices()[0] + getSegmentFilesFromMs("${ms.host}:${ms.httpPort}", tablet.TabletId, partition.VisibleVersion) { + code, body -> + assertEquals(200, code) + def meta = parseJson(body) + quickTest("layout_${packed}", """ + SELECT ${meta.num_segments as int} > 1, + ${(meta.packed_slice_locations ?: [:]).size() > 0}, + ${meta.segments_file_size.size() == (meta.num_segments as int)}, + ${meta.segments_file_size.every { (it as long) > 0 }}, + ${meta.inverted_index_file_info.size() == (meta.num_segments as int)}, + ${meta.inverted_index_file_info.every { (it.index_size as long) > 0 }} + """, true) + } + sql "INSERT INTO cloud_mow_forward VALUES (0,-1,1)" + sql "INSERT INTO cloud_mow_forward (k,v,seq,__DORIS_DELETE_SIGN__) VALUES (2,130000,13000,1)" + sql "INSERT INTO cloud_mow_forward VALUES (2,-1,1)" + quickTest("deleted_${packed}", "SELECT * FROM cloud_mow_forward", true) + sql "INSERT INTO cloud_mow_forward VALUES (2,140000,14000)" + streamLoad { + table "cloud_mow_forward" + set "column_separator", "," + set "memtable_on_sink_node", "true" + set "group_commit", "off_mode" + inputText "0,150000,15000\n3,-1,1\n" + check { result, exception, startTime, endTime -> + if (exception != null) { throw exception } + def response = parseJson(result) + quickTest("stream_status_${packed}", "SELECT '${response.Status}'", true) + } + } + quickTest("stream_${packed}", "SELECT * FROM cloud_mow_forward", true) + def originalEnableProfile = sql("SHOW GLOBAL VARIABLES LIKE 'enable_profile'")[0][1] + def originalProfileLevel = sql("SHOW GLOBAL VARIABLES LIKE 'profile_level'")[0][1] + def profileBlock = "VTabletWriterV2.close.profile_ready" + def groupLabel = "" + try { + sql "SET GLOBAL enable_profile=true" + sql "SET GLOBAL profile_level=2" + GetDebugPoint().enableDebugPointForAllBEs(profileBlock, [timeout: "120"]) + streamLoad { + table "cloud_mow_forward" + set "column_separator", "," + set "group_commit", "async_mode" + set "memtable_on_sink_node", "false" + unset "label" + inputText "1,160000,16000\n" + check { result, exception, startTime, endTime -> + if (exception != null) { throw exception } + def response = parseJson(result) + groupLabel = response.Label + quickTest("group_status_${packed}", "SELECT '${response.Status}', '${response.GroupCommit}'", true) + } + } + awaitGroupProfile(groupLabel, tablet.TabletId) + } finally { + GetDebugPoint().disableDebugPointForAllBEs(profileBlock) + sql "SET GLOBAL enable_profile=${originalEnableProfile}" + sql "SET GLOBAL profile_level=${originalProfileLevel}" + } + awaitUntil(60) { + (sql "SELECT v FROM cloud_mow_forward WHERE k=1")[0][0] == 160000 + } + quickTest("group_${packed}", "SELECT * FROM cloud_mow_forward", true) + checkUniqueKeys() + trigger_and_wait_compaction("cloud_mow_forward", "full") + quickTest("compacted_${packed}", "SELECT * FROM cloud_mow_forward", true) + checkUniqueKeys() + } + } + // MOW without sink upload forwards files and calculates bitmaps on the target BE. + sql "DROP TABLE IF EXISTS cloud_memtable_mow_fallback" + sql """ + CREATE TABLE cloud_memtable_mow_fallback (k BIGINT NOT NULL, v BIGINT) + UNIQUE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ("replication_num"="1", "enable_unique_key_merge_on_write"="true") + """ + sql "INSERT INTO cloud_memtable_mow_fallback VALUES (1,10)" + sql "INSERT INTO cloud_memtable_mow_fallback VALUES (1,20)" + order_qt_mow_fallback "SELECT * FROM cloud_memtable_mow_fallback" + + sql "DROP TABLE IF EXISTS cloud_mow_forward_broker" + sql """ + CREATE TABLE cloud_mow_forward_broker (k BIGINT NOT NULL, v BIGINT) + UNIQUE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ("replication_num"="1", "enable_unique_key_merge_on_write"="true", + "disable_auto_compaction"="true") + """ + sql "INSERT INTO cloud_mow_forward_broker VALUES (50,-1)" + def brokerLabel = "mow_forward_" + UUID.randomUUID().toString().replace('-', '_') + sql "SET enable_profile=true" + try { + sql """ + LOAD LABEL ${brokerLabel} ( + DATA INFILE("s3://${getS3BucketName()}/regression/load/data/basic_data.csv") + INTO TABLE cloud_mow_forward_broker COLUMNS TERMINATED BY "|" FORMAT AS "CSV" + (k, c01, c02, c03, c04, c05, c06, c07, c08, c09, + c10, c11, c12, c13, c14, c15, c16, c17, c18) SET (v=k*2) + ) WITH S3 ( + "AWS_ACCESS_KEY"="${getS3AK()}", "AWS_SECRET_KEY"="${getS3SK()}", + "AWS_ENDPOINT"="${getS3Endpoint()}", "AWS_REGION"="${getS3Region()}", + "provider"="${getS3Provider()}" + ) PROPERTIES ("load_parallelism"="1") + """ + waitForBrokerLoadDone(brokerLabel) + def load = sql_return_maparray("SHOW LOAD WHERE LABEL = '${brokerLabel}'")[0] + new ProfileAction(context).getProfile(load.JobId.toString(), ["DeltaWriterV2"]) + } finally { + sql "SET enable_profile=false" + } + sql """ + SELECT assert_true(COUNT(*) = 0, 'cloud_mow_forward_broker contains duplicate keys') FROM ( + SELECT k, COUNT(*) AS a FROM cloud_mow_forward_broker GROUP BY k HAVING a > 1 + ) duplicates + """ + order_qt_broker_rows "SELECT k,v FROM cloud_mow_forward_broker" + + sql "DROP TABLE IF EXISTS cloud_memtable_partial_update" + sql """ + CREATE TABLE cloud_memtable_partial_update (k INT NOT NULL, v INT, untouched INT) + UNIQUE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ("replication_num"="1", "enable_unique_key_merge_on_write"="true") + """ + sql "INSERT INTO cloud_memtable_partial_update VALUES (1,10,100)" + streamLoad { + table "cloud_memtable_partial_update" + set "column_separator", "," + set "columns", "k,v" + set "partial_columns", "true" + set "memtable_on_sink_node", "true" + set "group_commit", "off_mode" + inputText "1,20\n" + check { result, exception, startTime, endTime -> + if (exception != null) { throw exception } + def response = parseJson(result) + sql "SELECT assert_true('${response.Status}' = 'Success', 'partial-update Stream Load failed')" + } + } + sql """ + SELECT assert_true(COUNT(*) = 1 AND SUM(v) = 20 AND SUM(untouched) = 100, 'partial update result mismatch') + FROM cloud_memtable_partial_update WHERE k = 1 + """ + } +} diff --git a/regression-test/suites/cloud_p0/test_cloud_memtable_on_sink.groovy b/regression-test/suites/cloud_p0/test_cloud_memtable_on_sink.groovy new file mode 100644 index 00000000000000..f34dcc078b3745 --- /dev/null +++ b/regression-test/suites/cloud_p0/test_cloud_memtable_on_sink.groovy @@ -0,0 +1,405 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import org.apache.doris.regression.action.ProfileAction +import org.apache.doris.regression.suite.ClusterOptions + +suite("test_cloud_memtable_on_sink", "p0, docker") { + def options = new ClusterOptions() + options.cloudMode = true + options.setFeNum(1) + options.setBeNum(3) + options.enableDebugPoints() + // Split the small S3 CSV across all three BEs with load_parallelism = 1. + options.feConfigs += ['min_bytes_per_broker_scanner = 100'] + // The segment-count checks require the configured four-row CSV batches. + options.beConfigs += ['enable_packed_file=true', 'small_file_threshold_bytes=1048576', + 'enable_adaptive_batch_size=false'] + + docker(options) { + def loadS3 = { table, label -> + sql """ + LOAD LABEL ${label} ( + DATA INFILE("s3://${getS3BucketName()}/regression/load/data/basic_data.csv") + INTO TABLE ${table} COLUMNS TERMINATED BY "|" FORMAT AS "CSV" + (k, c01, c02, c03, c04, c05, c06, c07, c08, c09, + c10, c11, c12, c13, c14, c15, c16, c17, c18) + SET (v = k * 2) + ) WITH S3 ( + "AWS_ACCESS_KEY"="${getS3AK()}", "AWS_SECRET_KEY"="${getS3SK()}", + "AWS_ENDPOINT"="${getS3Endpoint()}", "AWS_REGION"="${getS3Region()}", + "provider"="${getS3Provider()}" + ) PROPERTIES ("load_parallelism"="1") + """ + waitForBrokerLoadDone(label) + return sql_return_maparray("SHOW LOAD WHERE LABEL = '${label}'")[0] + } + + def testDuplicate = { + sql "SET enable_sql_cache = false" + sql "DROP TABLE IF EXISTS test_cloud_duplicate_memtable_on_sink_source" + sql "DROP TABLE IF EXISTS test_cloud_duplicate_memtable_on_sink" + + sql """ + CREATE TABLE test_cloud_duplicate_memtable_on_sink_source ( + k BIGINT NOT NULL, + v BIGINT NOT NULL + ) + DUPLICATE KEY(k) + DISTRIBUTED BY HASH(k) BUCKETS 3 + PROPERTIES ("replication_num" = "1") + """ + sql """ + CREATE TABLE test_cloud_duplicate_memtable_on_sink ( + k BIGINT NOT NULL, + v BIGINT NOT NULL + ) + DUPLICATE KEY(k) + DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ("replication_num" = "1") + """ + + sql "SET enable_memtable_on_sink_node = false" + sql """ + INSERT INTO test_cloud_duplicate_memtable_on_sink_source + SELECT number, number * 2 FROM numbers("number" = "100000") + """ + + try { + sql "SET enable_memtable_on_sink_node = true" + sql "SET profile_level = 2" + sql "SET enable_profile = true" + sql """ + /* cloud_duplicate_memtable_on_sink_profile_false */ + INSERT INTO test_cloud_duplicate_memtable_on_sink + SELECT k, v FROM test_cloud_duplicate_memtable_on_sink_source + """ + def required = ["DeltaWriterV2"] + def profileString = new ProfileAction(context).getProfileBySql( + "cloud_duplicate_memtable_on_sink_profile_false", required) + logger.info("memtable-on-sink profile:\n{}", profileString) + } finally { + sql "SET enable_profile = false" + sql "SET enable_memtable_on_sink_node = false" + } + + sql """ + SELECT assert_true( + COUNT(*) = 100000 + AND SUM(k) = 4999950000 + AND SUM(v) = 9999900000, + 'cloud duplicate memtable-on-sink result mismatch') + FROM test_cloud_duplicate_memtable_on_sink + """ + + sql "DROP TABLE IF EXISTS test_cloud_duplicate_memtable_on_sink_s3" + sql """ + CREATE TABLE test_cloud_duplicate_memtable_on_sink_s3 ( + k BIGINT NOT NULL, + v BIGINT NOT NULL, + INDEX idx_v (v) USING INVERTED + ) + DUPLICATE KEY(k) + DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "inverted_index_storage_format" = "V2", + "disable_auto_compaction" = "true" + ) + """ + + def label = "cloud_duplicate_memtable_on_sink_s3_" + UUID.randomUUID().toString().replace('-', '_') + try { + sql "SET enable_memtable_on_sink_node = true" + sql "SET enable_profile = true" + // The three scanners read 6, 6, and 8 rows. Flush after the first four rows on + // each BE, then flush the remaining rows at close to produce two segments each. + sql "SET broker_load_batch_size = 4" + GetDebugPoint().enableDebugPointForAllBEs("MemTable.need_flush", [execute: 1]) + def load = loadS3("test_cloud_duplicate_memtable_on_sink_s3", label) + assertEquals("FINISHED", load.State, "S3 load did not finish: ${load}") + + def profileString = new ProfileAction(context).getProfile( + load.JobId.toString(), ["DeltaWriterV2", "NumScanners"]) + logger.info("S3 memtable-on-sink profile:\n{}", profileString) + + // Only inspect per-BE pipelines, excluding the merged profile's duplicate counters. + def pipelines = profileString.split(/(?m)(?=^[ \t]*(?:Pipeline \d+|FragmentLevelProfile:)\(host=)/) + .findAll { it.trim().startsWith("Pipeline ") } + def backends = sql_return_maparray("SHOW BACKENDS") + backends.each { backend -> + def pipeline = pipelines.find { + it.readLines()[0].contains("hostname:${backend.Host},") && it.contains("DeltaWriterV2") + } + assertNotNull(pipeline, "Missing S3 sink writer on BE ${backend.Host}") + assertTrue(pipeline.contains("FILE_SCAN_OPERATOR"), "Missing S3 scanner on BE ${backend.Host}") + assertTrue((pipeline =~ /(?m)^\s*- NumScanners: 1\s*$/).find(), + "Expected one S3 scanner on BE ${backend.Host}") + assertTrue((pipeline =~ /(?m)^\s*- SegmentNum: 2\s*$/).find(), + "Expected two flushed segments on BE ${backend.Host}") + } + } finally { + GetDebugPoint().disableDebugPointForAllBEs("MemTable.need_flush") + sql "SET enable_profile = false" + sql "SET enable_memtable_on_sink_node = false" + } + + // Read both segment data and V2 indexes without the uploader's file cache. + sql "SET enable_file_cache = false" + quickTest("s3_rows_false", """ + SELECT COUNT(*), SUM(k), SUM(v) + FROM test_cloud_duplicate_memtable_on_sink_s3 + """, true) + quickTest("s3_index_false", """ + SELECT k, v FROM test_cloud_duplicate_memtable_on_sink_s3 WHERE v IN (62, 100, 114) + """, true) + def tablet = sql_return_maparray("SHOW TABLETS FROM test_cloud_duplicate_memtable_on_sink_s3")[0] + def partition = sql_return_maparray("SHOW PARTITIONS FROM test_cloud_duplicate_memtable_on_sink_s3")[0] + def ms = cluster.getAllMetaservices()[0] + getSegmentFilesFromMs("${ms.host}:${ms.httpPort}", tablet.TabletId, partition.VisibleVersion) { + responseCode, body -> + assertEquals(200, responseCode) + logger.info("S3 memtable-on-sink rowset meta: {}", body) + def rowsetMeta = parseJson(body) + def locations = rowsetMeta.packed_slice_locations + def segmentIds = rowsetMeta.segment_ids ?: (0..<(rowsetMeta.num_segments as int)).toList() + logger.info("S3 forwarded rowset layout: {}", [ + tablet_id: tablet.TabletId, version: partition.VisibleVersion, + rowset_id: rowsetMeta.rowset_id_v2, segment_ids: segmentIds, + num_segment_rows: rowsetMeta.num_segment_rows, + segments_file_size: rowsetMeta.segments_file_size, + packed_files: locations.keySet().sort() + ]) + // Only the destination rowset's first segment and its V2 index are packed. + // The other five segments retain independent files. + quickTest("s3_packed_meta_false", """ + SELECT ${rowsetMeta.num_segments as int}, ${rowsetMeta.num_rows as long}, + ${locations.size()}, + ${locations.keySet().count { it.endsWith('_0.dat') }}, + ${locations.keySet().count { it.endsWith('_0.idx') }}, + ${segmentIds.count { (it as long) >= 1000 }}, '${segmentIds.join(",")}' + """, true) + } + + // Stream Load also uses the file-forwarding path when memtable-on-sink is enabled. + // This hook is only reached by the memtable-on-sink receiving writer. + // A successful fallback would fail this check instead of silently passing. + def failClose = "LoadStreamWriter.close.cancelled" + try { + GetDebugPoint().enableDebugPointForAllBEs(failClose) + streamLoad { + table "test_cloud_duplicate_memtable_on_sink_s3" + set "column_separator", "," + set "memtable_on_sink_node", "true" + set "group_commit", "off_mode" + inputText "100,200\n101,202\n" + check { result, exception, startTime, endTime -> + if (exception != null) { throw exception } + def response = parseJson(result) + sql "SELECT assert_true('${response.Status}' = 'Fail', 'Stream Load should fail at receiver close')" + } + } + } finally { + GetDebugPoint().disableDebugPointForAllBEs(failClose) + } + sql """ + SELECT assert_true(COUNT(*) = 20 AND SUM(k) = 1133 AND SUM(v) = 2266, 'failed Stream Load changed visible rows') + FROM test_cloud_duplicate_memtable_on_sink_s3 + """ + + streamLoad { + table "test_cloud_duplicate_memtable_on_sink_s3" + set "column_separator", "," + set "memtable_on_sink_node", "true" + set "group_commit", "off_mode" + inputStream new ByteArrayInputStream("100,200\n101,202\n".getBytes()) + time 30000 + check { result, exception, startTime, endTime -> + if (exception != null) { + throw exception + } + assertEquals("Success", parseJson(result).Status) + } + } + order_qt_stream_rows """ + SELECT COUNT(*), SUM(k), SUM(v) FROM test_cloud_duplicate_memtable_on_sink_s3 + """ + } + + def testAggregateAndMor = { + sql "DROP TABLE IF EXISTS cloud_memtable_models_source" + sql """ + CREATE TABLE cloud_memtable_models_source (n BIGINT NOT NULL) + DUPLICATE KEY(n) DISTRIBUTED BY HASH(n) BUCKETS 12 + PROPERTIES ("replication_num"="1") + """ + sql "SET enable_memtable_on_sink_node=false" + sql "INSERT INTO cloud_memtable_models_source SELECT number FROM numbers('number'='12000')" + sql "SET enable_memtable_on_sink_node=true" + sql "SET parallel_pipeline_task_num=4" + sql "SET profile_level=2" + sql "SET enable_file_cache=false" + + def aggQuery = """ + SELECT k, s, lo, hi, bitmap_count(b), hll_cardinality(h), r, rn + FROM cloud_memtable_agg + """ + def morQuery = "SELECT k, v FROM cloud_memtable_mor" + def seqQuery = "SELECT k, v, seq FROM cloud_memtable_mor_seq" + def loadModelS3 = { table -> + def label = "agg_mor_" + UUID.randomUUID().toString().replace('-', '_') + sql "SET enable_profile=true" + def load = loadS3(table, label) + def required = ["DeltaWriterV2"] + new ProfileAction(context).getProfile(load.JobId.toString(), required) + sql "SET enable_profile=false" + } + + sql "DROP TABLE IF EXISTS cloud_memtable_agg" + sql "DROP TABLE IF EXISTS cloud_memtable_mor" + sql "DROP TABLE IF EXISTS cloud_memtable_mor_seq" + sql "DROP TABLE IF EXISTS cloud_memtable_agg_broker" + sql "DROP TABLE IF EXISTS cloud_memtable_mor_broker" + sql """ + CREATE TABLE cloud_memtable_agg ( + k BIGINT NOT NULL, s BIGINT SUM, lo BIGINT MIN, hi BIGINT MAX, + b BITMAP BITMAP_UNION, h HLL HLL_UNION, + r BIGINT REPLACE, rn BIGINT REPLACE_IF_NOT_NULL + ) AGGREGATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ("replication_num"="1", "disable_auto_compaction"="true") + """ + sql """ + CREATE TABLE cloud_memtable_mor (k BIGINT NOT NULL, v BIGINT) + UNIQUE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ("replication_num"="1", "enable_unique_key_merge_on_write"="false", + "disable_auto_compaction"="true") + """ + sql """ + CREATE TABLE cloud_memtable_mor_seq ( + k BIGINT NOT NULL, v BIGINT, seq BIGINT NOT NULL, + INDEX idx_k (k) USING INVERTED + ) UNIQUE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ("replication_num"="1", "enable_unique_key_merge_on_write"="false", + "function_column.sequence_col"="seq", "inverted_index_storage_format"="V2", + "disable_auto_compaction"="true") + """ + sql "SET enable_profile=true" + sql """ + /* cloud_memtable_agg_false */ + INSERT INTO cloud_memtable_agg + SELECT n % 4, n, n, n, to_bitmap(n), hll_hash(CAST(n % 4 AS STRING)), + n % 4 + 10, IF(n % 8 < 4, n % 4 + 20, NULL) + FROM cloud_memtable_models_source + """ + sql """ + /* cloud_memtable_mor_false */ + INSERT INTO cloud_memtable_mor SELECT n % 4, n % 4 * 10 FROM cloud_memtable_models_source + """ + sql """ + /* cloud_memtable_mor_seq_false */ + INSERT INTO cloud_memtable_mor_seq SELECT n % 4, n * 10, n FROM cloud_memtable_models_source + """ + ['agg', 'mor', 'mor_seq'].each { model -> + def required = ["DeltaWriterV2"] + new ProfileAction(context).getProfileBySql("cloud_memtable_${model}_false", required) + } + sql "SET enable_profile=false" + quickTest("agg_false", aggQuery, true) + quickTest("mor_false", morQuery, true) + quickTest("seq_false", seqQuery, true) + quickTest("seq_index_false", seqQuery + " WHERE k IN (1,3)", true) + + // A newer transaction with a lower Sequence must not replace the business-newer row. + sql "INSERT INTO cloud_memtable_mor_seq VALUES (0,-1,1)" + quickTest("seq_lower_false", seqQuery, true) + sql "INSERT INTO cloud_memtable_mor VALUES (0,99)" + quickTest("mor_new_version_false", morQuery, true) + sql """ + INSERT INTO cloud_memtable_agg + VALUES (0,10,-1,13000,bitmap_empty(),hll_empty(),99,NULL) + """ + quickTest("agg_new_version_false", aggQuery, true) + sql "INSERT INTO cloud_memtable_mor_seq SELECT n % 4,n*10,n FROM cloud_memtable_models_source WHERE n<0" + quickTest("seq_empty_false", seqQuery, true) + ['agg', 'mor', 'mor_seq'].each { model -> + trigger_and_wait_compaction("cloud_memtable_${model}", "full") + } + quickTest("agg_compacted_false", aggQuery, true) + quickTest("mor_compacted_false", morQuery, true) + quickTest("seq_compacted_false", seqQuery, true) + + sql """ + CREATE TABLE cloud_memtable_agg_broker (k BIGINT NOT NULL, v BIGINT SUM) + AGGREGATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ("replication_num"="1", "disable_auto_compaction"="true") + """ + sql """ + CREATE TABLE cloud_memtable_mor_broker (k BIGINT NOT NULL, v BIGINT) + UNIQUE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ("replication_num"="1", "enable_unique_key_merge_on_write"="false", + "disable_auto_compaction"="true") + """ + loadModelS3("cloud_memtable_agg_broker") + loadModelS3("cloud_memtable_mor_broker") + quickTest("agg_broker_false", "SELECT k,v FROM cloud_memtable_agg_broker", true) + quickTest("mor_broker_false", "SELECT k,v FROM cloud_memtable_mor_broker", true) + + streamLoad { + table "cloud_memtable_agg_broker" + set "column_separator", "," + set "memtable_on_sink_node", "true" + inputStream new ByteArrayInputStream("50,1\n50,2\n".getBytes()) + check { result, exception, startTime, endTime -> + if (exception != null) { throw exception } + assertEquals("Success", parseJson(result).Status) + } + } + order_qt_agg_stream "SELECT k,v FROM cloud_memtable_agg_broker WHERE k=50" + streamLoad { + table "cloud_memtable_mor_seq" + set "column_separator", "," + set "memtable_on_sink_node", "true" + inputStream new ByteArrayInputStream("0,1,1\n1,130010,13001\n".getBytes()) + check { result, exception, startTime, endTime -> + if (exception != null) { throw exception } + assertEquals("Success", parseJson(result).Status) + } + } + order_qt_mor_stream seqQuery + sql "INSERT INTO cloud_memtable_mor_seq (k,v,seq,__DORIS_DELETE_SIGN__) VALUES (2,140000,14000,1)" + order_qt_mor_delete seqQuery + sql "INSERT INTO cloud_memtable_mor_seq VALUES (2,150000,15000)" + order_qt_mor_reinsert seqQuery + trigger_and_wait_compaction("cloud_memtable_mor_seq", "full") + order_qt_mor_final_compacted seqQuery + } + + def originalBatchSize = sql("SHOW VARIABLES LIKE 'broker_load_batch_size'")[0][1] + try { + testDuplicate() + } finally { + sql "SET broker_load_batch_size=${originalBatchSize}" + sql "SET enable_profile=false" + GetDebugPoint().disableDebugPointForAllBEs("MemTable.need_flush") + } + try { + testAggregateAndMor() + } finally { + sql "SET enable_profile=false" + } + } +}