[improvement](cloud) support memtable on sink node: support sink upload file to object storage - #68366
Open
mymeiyi wants to merge 6 commits into
Open
[improvement](cloud) support memtable on sink node: support sink upload file to object storage#68366mymeiyi wants to merge 6 commits into
mymeiyi wants to merge 6 commits into
Conversation
### 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
### 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
mymeiyi
requested review from
924060929,
airborne12,
csun5285,
eldenmoon,
englefly,
gavinchou,
liaoxin01,
luwei16,
morrySnow,
starocean999 and
yiguolei
as code owners
September 22, 2026 04:03
Contributor
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
Contributor
There was a problem hiding this comment.
Copilot review overview
🟡 Changes recommended
Unresolved protocol compatibility, rowset, cache-policy, and configuration issues block approval.
Get a fresh assessment by requesting another Copilot review.
Review effort: Lite
Findings: 3
Open (4)
What changed in this PR
Adds cloud memtable-on-sink uploads to object storage, including MOW handling, cache warmup, protocol extensions, and load-path integration.
Changes:
- Adds sink-upload configuration and stream protocols.
- Implements distributed rowset assembly, caching, and MOW support.
- Adds regression and unit-test coverage.
| File | Reviewed scope | Final review note |
|---|---|---|
regression-test/suites/cloud_p0/test_cloud_memtable_sink_upload_unshared.groovy |
Tests unshared sink uploads. | — |
regression-test/suites/cloud_p0/test_cloud_memtable_sink_upload_mow.groovy |
Tests MOW sink uploads. | — |
regression-test/suites/cloud_p0/test_cloud_memtable_sink_upload_cache.groovy |
Tests sink-upload cache behavior. | — |
regression-test/suites/cloud_p0/test_cloud_memtable_on_sink.groovy |
Tests cloud memtables on sink nodes. | — |
regression-test/suites/cloud_p0/test_cloud_memtable_mow_forward.groovy |
Tests MOW forwarding fallback. | — |
regression-test/suites/cloud_p0/test_cloud_memtable_group_commit.groovy |
Tests Group Commit integration. | — |
regression-test/data/cloud_p0/test_cloud_memtable_sink_upload_unshared.out |
Expected unshared-upload results. | — |
regression-test/data/cloud_p0/test_cloud_memtable_sink_upload_mow.out |
Expected MOW results. | — |
regression-test/data/cloud_p0/test_cloud_memtable_sink_upload_cache.out |
Expected cache results. | — |
regression-test/data/cloud_p0/test_cloud_memtable_on_sink.out |
Expected sink-memtable results. | — |
regression-test/data/cloud_p0/test_cloud_memtable_mow_forward.out |
Expected forwarding results. | — |
regression-test/data/cloud_p0/test_cloud_memtable_group_commit.out |
Expected Group Commit results. | — |
gensrc/thrift/PaloInternalService.thrift |
Adds the sink-upload query option. | — |
gensrc/thrift/FrontendService.thrift |
Adds the Stream Load sink-upload option. | — |
gensrc/proto/internal_service.proto |
Adds sink-upload context and partial-rowset messages. | — |
fe/fe-core/src/test/java/org/apache/doris/planner/StreamLoadPlannerTest.java |
Tests Stream Load option propagation. | — |
fe/fe-core/src/test/java/org/apache/doris/load/loadv2/BrokerLoadJobTest.java |
Tests Broker Load propagation. | — |
fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java |
Applies HTTP Stream Load configuration. | Moderate (1 vote): The config overrides an HTTP false header, preventing the documented per-request override. |
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java |
Defines and serializes the session option. | — |
fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadTask.java |
Stores Stream Load sink-upload state. | Moderate (3 votes): Group Commit internal loads bypass the disabling cloud configuration. |
fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadPlanner.java |
Sends the option to BE planners. | — |
fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadTaskInfo.java |
Adds the load-task option contract. | — |
fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadLoadingTask.java |
Propagates Broker Load configuration. | — |
fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BulkLoadJob.java |
Captures session configuration. | — |
fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java |
Configures Broker Load tasks. | — |
fe/fe-common/src/main/java/org/apache/doris/common/Config.java |
Adds cloud sink-upload configuration. | — |
be/test/storage/rowset/rowset_meta_test.cpp |
Tests packed-location collection. | — |
be/test/storage/rowset/beta_rowset_test.cpp |
Tests streamed rowset metadata and bitmap behavior. | — |
be/test/storage/compaction/compaction_file_cache_test.cpp |
Tests file-cache disabling. | — |
be/test/storage/cloud_file_cache_write_index_only_test.cpp |
Tests remote index caching. | — |
be/test/runtime/load_stream_test.cpp |
Tests stream-upload metadata and MOW assembly. | — |
be/test/load/channel/load_stream_stub_map_test.cpp |
Tests concurrent segment accumulation. | — |
be/test/cloud/cloud_warm_up_manager_test.cpp |
Tests local cache warmup. | — |
be/src/storage/segment/segment_index_file_cache_loader.cpp |
Skips index preload when caching is disabled. | — |
be/src/storage/rowset/vertical_beta_rowset_writer.cpp |
Preserves partial-output segment IDs. | — |
be/src/storage/rowset/rowset_writer_context.h |
Adds cache and partial-writer context flags. | Moderate (1 vote): The packed-file cache flag ignores write_file_cache=false. |
be/src/storage/rowset/rowset_meta.h |
Adds packed-location and bounds APIs. | — |
be/src/storage/rowset/rowset_meta.cpp |
Implements packed-location collection. | Critical (2 votes): Empty buffered inverted-index files can trigger an assertion with no slice location. |
be/src/storage/rowset/beta_rowset_writer.h |
Shares segment-overlap detection. | — |
be/src/storage/rowset/beta_rowset_writer.cpp |
Supports partial rowset metadata and bitmap snapshots. | — |
be/src/storage/rowset/beta_rowset_writer_v2.h |
Removes an obsolete segment accessor. | — |
be/src/storage/olap_common.h |
Stores immutable MOW snapshot state. | — |
be/src/service/http/http_common.h |
Defines the sink-upload HTTP header. | — |
be/src/service/http/action/stream_load.cpp |
Parses the sink-upload header. | — |
be/src/load/delta_writer/delta_writer_v2.h |
Adds sink-upload writer state. | — |
be/src/load/delta_writer/delta_writer_v2.cpp |
Implements sink upload and MOW result submission. | Critical (2 votes): The new handshake can target older BEs without capability negotiation, causing timeouts. |
be/src/load/delta_writer/delta_writer_context.h |
Adds the sink-upload request flag. | — |
be/src/load/channel/load_stream.h |
Extends stream context and control APIs. | — |
be/src/load/channel/load_stream.cpp |
Handles context and partial-rowset messages. | — |
be/src/load/channel/load_stream_writer.h |
Tracks distributed partial rowsets. | — |
be/src/load/channel/load_stream_writer.cpp |
Registers writers and assembles target rowsets. | — |
be/src/exec/sink/writer/vtablet_writer_v2.h |
Stores transaction and storage metadata. | — |
be/src/exec/sink/writer/vtablet_writer_v2.cpp |
Passes sink-upload settings to writers. | Critical (1 vote): The new protocol is enabled without mixed-version negotiation, so forwarding is incompatible during rolling upgrades. |
be/src/exec/sink/load_stream_stub.h |
Adds sink-upload RPC helpers. | — |
be/src/exec/sink/load_stream_stub.cpp |
Sends context and partial-rowset messages. | — |
be/src/exec/sink/load_stream_map_pool.h |
Exposes segment-count accumulation. | — |
be/src/exec/sink/load_stream_map_pool.cpp |
Accumulates counts from unshared writers. | — |
be/src/exec/pipeline/pipeline_fragment_context.cpp |
Enables cloud memtables on sink nodes. | — |
be/src/cloud/cloud_warm_up_manager.h |
Adds local warmup support. | — |
be/src/cloud/cloud_warm_up_manager.cpp |
Implements local cache warmup. | — |
be/src/cloud/cloud_rowset_writer.h |
Adds assembled-rowset construction. | — |
be/src/cloud/cloud_rowset_writer.cpp |
Builds rowsets from assembled metadata. | — |
be/src/cloud/cloud_rowset_builder.h |
Declares MOW and metadata assembly APIs. | — |
be/src/cloud/cloud_rowset_builder.cpp |
Implements validation and rowset assembly. | — |
💡 Configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Comment on lines
+576
to
+577
| .cloud_sink_upload = config::is_cloud_mode() && | ||
| _state->query_options().enable_cloud_memtable_sink_upload, |
Comment on lines
+137
to
+138
| if (_req.cloud_sink_upload) { | ||
| RETURN_IF_ERROR(_init_sink_upload_writer(context)); |
| io::PackedSliceLocation location; | ||
| RETURN_IF_ERROR(static_cast<const io::PackedFileWriter&>(file_writer) | ||
| .get_packed_slice_location(&location)); | ||
| DORIS_CHECK(!location.packed_file_path.empty()); |
| private boolean enableProfile = false; | ||
|
|
||
| private boolean memtableOnSinkNode = false; | ||
| private boolean cloudMemtableSinkUpload = true; |
### 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.
### What problem does this PR solve? Problem Summary: Distributed compaction and memtable sink uploads need common rowset writer controls independently of distributed compaction orchestration. Extract these existing controls into a standalone prerequisite. Add is_partial_output_writer to retain assigned physical segment IDs for partial compaction outputs, skip writer-local segment compaction, and keep complete per-segment key bounds without aggregation or truncation. Let RowsetMeta callers explicitly disable key-bound truncation. Add disable_file_cache to override index-only and adaptive cache admission in file writer options, with the existing cache-policy unit test. This commit only extracts existing changes; the final branch source tree is unchanged. The sink upload feature subsequently enables explicit segment IDs for partial load writers as well. ### Release note None ### Check List (For Author) - Test: No compilation or runtime tests run, as requested. Clang-format 16, build hygiene, and git diff whitespace checks passed. Static application to cloud-load-2 and its previously conflicting sink-upload writer changes succeeded in an isolated index. Retained CompactionFileCacheTest.DisableFileCacheOverridesAllWritePolicies. - Behavior changed: No; extracts existing internal controls without enabling them for default writers. - Does this need documentation: No; internal prerequisite extraction.
…adata ### What problem does this PR solve? Problem Summary: Temporary rowsets collect completed segment IDs from a statistics snapshot, but metadata construction used the consecutive segment count from _num_seg(). Out-of-order completion can make these counts differ. When segment IDs are recorded, such as for partial output writers, the size check can fail before _build_tmp sets the completed segment IDs and corrects the count. Use the completed-ID snapshot size for temporary rowsets, while retaining the existing count for final rowset construction. Extend TmpRowsetUsesCompletedSegmentIds with is_partial_output_writer=true to cover the mismatch using completed segments 2 and 6. Keep the streamed bitmap metadata snapshot change in the following Cloud forwarding feature. ### Release note Fix a segment-count check failure when constructing temporary rowset metadata with explicit segment IDs after out-of-order segment completion. ### Check List (For Author) - Test: Existing TmpRowsetUsesCompletedSegmentIds test extension moved with the fix. No compilation or tests run, per user request. Static review, repository clang-format v16 checks of the extracted changes, and git diff --check passed. Final code tree verified identical before and after splitting. - Behavior changed: Yes; temporary rowset metadata uses the completed segment count instead of the consecutive segment count. - Does this need documentation: No
…rage ### What problem does this PR solve? Problem Summary: Cloud memtable-on-sink file forwarding routes segment and index bytes through the target BE. Allow sink BEs to upload these files directly to object storage while retaining file forwarding when disabled. Register sink writers with the target BE, allocate stable segment ID ranges, and return the storage, encryption, and cache context via GET_WRITE_CONTEXT. Writers report completed metadata through ADD_PARTIAL_ROWSET. Validate and assemble their metadata, preserving sparse segment IDs, packed mappings, key bounds, and merged VARIANT schemas. Distribute a target-owned snapshot for MOW bitmap calculation, then validate and merge sink results before commit-phase bitmap work. Preserve target-local caching and asynchronously warm uploaded rowsets on the target BE. Propagate enable_cloud_memtable_sink_upload through SQL and Broker Load. Use cloud_stream_load_default_memtable_sink_upload for HTTP Stream, Routine Load, Group Commit internal loads, and ordinary Stream Load requests without a cloud_memtable_sink_upload header. An explicit header overrides that configuration. Both the session option and FE configuration default to true; the upload option applies only when memtable-on-sink is selected. Leave the optional Thrift request field without a default so generated C++ requests preserve absence and allow FE to resolve the configured default. Extend unit and regression coverage for configuration inheritance, header overrides, metadata assembly, MOW, cache controls, independent writers, Group Commit, packed files, VARIANT, and DUP/AGG/MOR loads. Restore the source commit's generated outputs for the reintroduced upload test branches. Check C++ request presence for omitted and explicit upload options, and fail the MOW forwarding regression on Stream Load errors even during output generation. ### Release note Cloud memtable-on-sink loads can upload segment and index files directly from sink BEs and asynchronously warm the target BE cache. SQL uses session variable enable_cloud_memtable_sink_upload. Stream Load defaults are controlled by cloud_stream_load_default_memtable_sink_upload and can be overridden per ordinary Stream Load request using cloud_memtable_sink_upload. These upload options default to true; the Stream Load memtable-on-sink default remains off. ### Check List (For Author) - Test: No local compilation or runtime tests run, as requested. Repository clang-format 16, build hygiene, and source git diff --check passed for the changes. Inspected the user's remote test_cloud_memtable_on_sink run: the suite completed successfully and generated output matched HEAD. The new request-presence unit test has not been run. This squash preserves the complete code tree unchanged. - Behavior changed: Yes; eligible cloud loads can upload sink-built files directly, distribute MOW bitmap work, and warm the target cache. Ordinary Stream Load without an upload header inherits the FE configuration. - Does this need documentation: Yes; cloud_memtable.md and cloud_memtable_on_sink.md are maintained on the source development branch and excluded from this cherry-pick as requested.
mymeiyi
force-pushed
the
cloud-load-2
branch
from
September 22, 2026 09:04
921babd to
8bbd7a4
Compare
Contributor
Author
|
/review |
Contributor
|
Codex automated review failed and did not complete. Error: No eligible Codex review credentials: all are usage-limited or have a reused refresh token. Earliest usage-limit retry is 2026-09-23T07:04:00Z. Credentials with reused refresh tokens must be replaced. Please inspect the workflow logs and rerun the review after the underlying issue is resolved. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
based on #68365