From 4e624bbea339fbc489c166dd4e4f330a0964ad26 Mon Sep 17 00:00:00 2001 From: meiyi Date: Tue, 15 Sep 2026 17:17:39 +0800 Subject: [PATCH 1/3] [refactor](cloud) Share packed slice metadata collection ### What problem does this PR solve? Problem Summary: Packed slice collection depends only on a closed file writer, its logical path, and the destination rowset metadata, but was a private method of CloudRowsetWriter. Move it to RowsetMeta::collect_packed_slice_location and reuse it for existing Cloud segment and index collection without changing collection timing. Preserve skipping unfinished and non-packed files. Require a nonempty packed object path after a successful packed-location lookup, replacing the previous silent skip with an invariant assertion. Move the associated RowsetMetaPackedFileTest coverage with this refactor: completed data/index files, logical paths distinct from writer URIs, repeated collection, serialization, non-packed files, and missing mappings. Memtable forwarding call sites and MOW collection timing changes remain in subsequent commits. During rebase, retain the upstream writer-owned packed slice lookup so global index cleanup cannot invalidate metadata collection. Adapt the unit coverage to real PackedFileWriter handles, including incomplete writers, direct files, serialization, repeated collection, and missing-handle invariant failures. ### Release note None ### Check List (For Author) - Test: Unit tests adapted but not run, per request. No build or service execution. Conflict-resolution files passed clang-format 16, format checks, build hygiene, and Git whitespace checks. - Behavior changed: Yes; an empty packed object path after successful lookup is an invariant failure instead of a silent skip - Does this need documentation: No --- be/src/cloud/cloud_rowset_writer.cpp | 41 +-------- be/src/cloud/cloud_rowset_writer.h | 2 - be/src/storage/rowset/rowset_meta.cpp | 22 +++++ be/src/storage/rowset/rowset_meta.h | 5 ++ be/test/storage/rowset/rowset_meta_test.cpp | 96 +++++++++++++++++++++ 5 files changed, 127 insertions(+), 39 deletions(-) diff --git a/be/src/cloud/cloud_rowset_writer.cpp b/be/src/cloud/cloud_rowset_writer.cpp index 9753ca3dd99cd9..330527329d5bd6 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 { @@ -187,8 +186,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 +198,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/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/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 { From fa0de8781640995f7ef10697c521195be627ca28 Mon Sep 17 00:00:00 2001 From: meiyi Date: Tue, 15 Sep 2026 17:39:18 +0800 Subject: [PATCH 2/3] [fix](cloud) Avoid concurrent and duplicate packed metadata collection ### What problem does this PR solve? Problem Summary: Cloud MOW loads build temporary rowsets in asynchronous delete bitmap tasks while other flushes can still add or close file writers. Building those temporary rowsets previously traversed all segment and index writer collections without their locks to collect packed mappings. Collect only the current segment's mapping after its writer closes, and reserve full collection for final rowset construction after flushes finish. Pass the segment ID through _build_tmp and update the existing unit-test wrapper and call accordingly. CloudRowsetWriter::build also collected the same mappings a second time directly after _build_rowset_meta had already collected them. Remove that redundant call. These issues affect existing Cloud load paths independently of memtable forwarding. Forwarded-file snapshots and direct-upload logic remain in the following feature. ### Release note Avoid concurrent traversal of file writer collections during Cloud MOW packed file loads and eliminate duplicate packed mapping collection at finalization. ### Check List (For Author) - Test: Static call-site review and git diff --check; existing TmpRowsetUsesCompletedSegmentIds test adapted to the signature change. No builds or runtime tests at user request; concurrent execution was not reproduced during this split. - Behavior changed: Yes; temporary rowsets collect only the current segment's packed mapping and final rowsets collect all mappings once - Does this need documentation: No --- be/src/cloud/cloud_rowset_writer.cpp | 10 ++++++---- be/src/storage/rowset/beta_rowset_writer.cpp | 13 +++++++++++-- be/src/storage/rowset/beta_rowset_writer.h | 2 +- be/test/storage/rowset/beta_rowset_test.cpp | 6 ++++-- 4 files changed, 22 insertions(+), 9 deletions(-) diff --git a/be/src/cloud/cloud_rowset_writer.cpp b/be/src/cloud/cloud_rowset_writer.cpp index 330527329d5bd6..c633bc25e30b13 100644 --- a/be/src/cloud/cloud_rowset_writer.cpp +++ b/be/src/cloud/cloud_rowset_writer.cpp @@ -102,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) { @@ -115,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 diff --git a/be/src/storage/rowset/beta_rowset_writer.cpp b/be/src/storage/rowset/beta_rowset_writer.cpp index 265448a831472b..72b3b5dcca0095 100644 --- a/be/src/storage/rowset/beta_rowset_writer.cpp +++ b/be/src/storage/rowset/beta_rowset_writer.cpp @@ -458,7 +458,7 @@ 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); + st = _build_tmp(rowset_ptr, segment_id); if (!st.ok()) { return st; } @@ -1156,7 +1156,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 +1169,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/test/storage/rowset/beta_rowset_test.cpp b/be/test/storage/rowset/beta_rowset_test.cpp index c7337ff4f9bfc6..c6c1d3e4343722 100644 --- a/be/test/storage/rowset/beta_rowset_test.cpp +++ b/be/test/storage/rowset/beta_rowset_test.cpp @@ -239,7 +239,9 @@ 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); + } }; class S3ClientMock : public Aws::S3::S3Client { @@ -533,7 +535,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); From cc5cea49d1b652227b3af22c88e3bb7343ce0df8 Mon Sep 17 00:00:00 2001 From: meiyi Date: Mon, 21 Sep 2026 15:44:12 +0800 Subject: [PATCH 3/3] [feature](cloud) Support memtable sink loads through file forwarding ### What problem does this PR solve? Problem Summary: Cloud loads previously excluded the memtable-on-sink path. Allow eligible full-row DUPLICATE, AGGREGATE, UNIQUE MOR, and UNIQUE MOW loads to build memtables and segments on sink BEs, then forward segment and index files to the target BE for object storage upload and rowset commit. Retain the restrictions for partial updates, V1 indexes, and row binlog. Pass transaction expiration, storage vault, and cache policy through stream-open requests. Select CloudRowsetBuilder in LoadStreamWriter, collect forwarded packed-file mappings and file sizes, and snapshot temporary rowset metadata before asynchronous MOW bitmap calculation on the target BE. Honor skip_writing_empty_rowset_metadata for untouched tablets by setting the policy before initialization and skipping prepare/commit RPCs while preserving empty markers and statistics. Close cloud tablet writers with concurrency 10 after pre-close, collecting all tablet results even when some fail. Preserve local-mode close behavior and existing locking. Add coverage for DUP/AGG/MOR/MOW, packed files, Stream Load, Broker Load, async Group Commit internal writer profiles, duplicate-key checks, cold reads, failed load invisibility and retry, and partial-update loading results. Add BE coverage for bitmap snapshot failures, empty-rowset lifecycle, and bounded concurrent close with mixed results. Align regression output blocks with the file-forwarding suite. ### Release note Cloud loads can enable memtable-on-sink for eligible full-row DUPLICATE, AGGREGATE, and UNIQUE MOR/MOW tables. Sink BEs construct files and target BEs upload them and calculate MOW bitmaps. Empty tablets honor the existing metadata-skip configuration, and cloud tablet closes run with bounded concurrency. ### Check List (For Author) - Test: Static call-site review, clang-format 16 checks, build hygiene checks, and source git diff --check passed. BE unit tests and regression cases added or updated; no compilation, source generation, or runtime tests performed during these changes, as requested. Unused regression output blocks removed without changing retained results; existing output-file trailing separators preserved. - Behavior changed: Yes; eligible Cloud loads use file forwarding, preserve empty metadata skipping, and close tablet writers concurrently. - Does this need documentation: Yes; design documents excluded from this commit at user request. --- be/src/cloud/cloud_rowset_builder.cpp | 11 + be/src/cloud/cloud_rowset_builder.h | 2 + .../pipeline/pipeline_fragment_context.cpp | 2 +- be/src/exec/sink/load_stream_stub.cpp | 15 +- be/src/exec/sink/load_stream_stub.h | 6 +- be/src/exec/sink/writer/vtablet_writer_v2.cpp | 15 +- be/src/exec/sink/writer/vtablet_writer_v2.h | 2 + be/src/load/channel/load_stream.cpp | 64 ++- be/src/load/channel/load_stream.h | 17 +- be/src/load/channel/load_stream_writer.cpp | 56 ++- be/src/load/channel/load_stream_writer.h | 3 +- be/src/storage/rowset/beta_rowset_writer.cpp | 16 +- be/test/cloud/cloud_load_stream_test.cpp | 291 +++++++++++++ be/test/storage/rowset/beta_rowset_test.cpp | 35 ++ .../doris/load/loadv2/BrokerLoadJob.java | 2 +- gensrc/proto/internal_service.proto | 3 + .../test_cloud_memtable_mow_forward.out | 109 +++++ .../cloud_p0/test_cloud_memtable_on_sink.out | 148 +++++++ .../test_cloud_memtable_mow_forward.groovy | 255 +++++++++++ .../test_cloud_memtable_on_sink.groovy | 405 ++++++++++++++++++ 20 files changed, 1416 insertions(+), 41 deletions(-) create mode 100644 be/test/cloud/cloud_load_stream_test.cpp create mode 100644 regression-test/data/cloud_p0/test_cloud_memtable_mow_forward.out create mode 100644 regression-test/data/cloud_p0/test_cloud_memtable_on_sink.out create mode 100644 regression-test/suites/cloud_p0/test_cloud_memtable_mow_forward.groovy create mode 100644 regression-test/suites/cloud_p0/test_cloud_memtable_on_sink.groovy 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/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 72b3b5dcca0095..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, segment_id); - 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) 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 c6c1d3e4343722..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" @@ -244,6 +246,39 @@ class BetaRowsetWriterForTest : public BetaRowsetWriter { } }; +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, 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" + } + } +}