Conversation
… traffic in SHOW DATA
Cloud-mode BEs can now spill to the instance's S3 storage vault instead of
local disks, selected by be.conf (`spill_storage_type = local | s3`). The
existing spill operators and part-file format are unchanged; a remote
`SpillDataDir` binds the vault file system lazily (no meta-service sync at
startup) and `SpillFileWriter`/`SpillFileReader` reuse `S3FileWriter` and
`S3FileReader`.
BE:
- Remote spill store keyed by `spill/{cloud_unique_id}/{boot_id}/...`; other
boot generations are cleaned by the GC thread once the store is ready.
- Atomic capacity reservation (`spill_s3_storage_limit_bytes`) and a
submit-time upload budget (`spill_s3_max_inflight_upload_bytes`) that
backpressures writers; `FileWriterOptions` gains an upload gate and a
done callback for that, honoured by `S3FileWriter`.
- Failed parts are drained before the budget is reconciled; multipart
uploads are aborted on failure.
- bvars/metrics for remote spill bytes, requests and throughput; profile
counters for remote spill.
- Since-boot upload totals are reported to meta-service about once a
minute and once at shutdown after all tasks are done (bounded retries).
Meta-service:
- `report_spill_stats` / `get_spill_stats`: one record per BE
(`stats/{instance}/spill/{cloud_unique_id}`); a new boot folds the
previous process into `prior_boots_*`, so the record count is bounded by
the number of BEs. Records are removed with the instance.
- Recycler task `recycle_expired_spill_objects` deletes spill objects older
than `spill_objects_expire_time_second` (S3/MOCK accessors).
FE:
- `SHOW DATA PROPERTIES("entire_warehouse"="true")` gains a
`RemoteSpillWriteSize` column (total row) as the billing input.
Tests: `SpillFileS3Test` (in-memory object store, 16 cases incl. budget,
cancellation and upload-failure paths), `MetaServiceTest.SpillStatsTest`,
`KeysTest.StatsSpillKeyTest`, recycler tests, and a column check in
`test_show_data_warehouse.groovy`.
Claude-Session: https://claude.ai/code/session_01Jrdwwwh8bZSnVwoCwykHse
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
|
/review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Static verdict: REQUEST_CHANGES. The S3-backed spill path and warehouse-level SHOW DATA plumbing are present, but 16 blocking correctness and operability issues remain. The highest risks are cross-BE deletion caused by a shared identity, non-durable and non-idempotent lifetime statistics, cleanup paths that can delete live data or leak billable storage, request counters that do not measure physical provider traffic, and unbounded recovery/metadata scans.
Critical checkpoints (Part 1.3):
- Goal and proof: The goal is remote S3 spill with capacity control, request/byte accounting, persisted warehouse totals, cleanup, and SHOW DATA exposure. Happy-path BE/cloud/FE/regression tests were added, but the goal is not correctly achieved because identity, durability, cleanup, retry, and accounting failures remain.
- Scope and focus: The 47-file change is cohesive around that feature, and the cross-layer breadth is necessary, but the accepted defects mean it is not yet review-ready.
- Concurrency: I traced the spill-GC thread, writer/upload pools, non-blocking close, upload-budget mutex/condition variable, atomic counters, heartbeat mutation, and FDB transactions. Reachable upload callbacks are balanced and I found no separate deadlock, but release builds mask an underflow invariant (MMF-10), and transaction serialization does not fence stale generations (MMF-9).
- Lifecycle and SIOF: No cross-TU static-initialization dependency was found. Process boot, query/file teardown, startup cleanup, graceful/abrupt shutdown, vault binding, and multipart ownership were traced; MMF-4, MMF-6, MMF-7, MMF-8, MMF-11, MMF-12, and MMF-13 show lifecycle failures.
- Configuration: New mutable storage, upload-budget, and recycler controls are observed online, but negative storage limits disable the cap (MMF-3), non-positive TTLs can select live objects (MMF-7), and runtime accepts/persists an upload budget that startup rejects (MMF-15).
- Compatibility: The protobuf additions are additive, SHOW DATA row width is consistent, non-cloud mode avoids the RPC, and the repository's supported latest-MS/older-FE direction remains compatible. The FE-first/older-MS hypothesis was dismissed under that explicit contract.
- Parallel paths: Local and S3 writer paths, PutObject/UploadPart, data/footer reads, synchronous/non-blocking close, and hooks with no stats sink were checked. Local behavior remains balanced, but both read and write request accounting miss physical retries/failures (MMF-5, MMF-16).
- Special conditions: Remote-only selection, default-vault lookup, positive-limit gates, age cutoffs, and release underflow handling were checked. The non-positive and defensive-continue conditions are not safe (MMF-3, MMF-7, MMF-10, MMF-15).
- Test coverage: Ordinary round trips, rotation, cleanup retries, capacity concurrency, budget blocking/cancellation, several upload failures, meta-service aggregation, and SHOW DATA shape are covered. Missing negative coverage includes two live BEs sharing an identity, reordered/stale reports, crash before report, live identity migration, real expiry/liveness, abort failure/restart residue, paginated cleanup, churn cardinality, invalid dynamic updates, and SDK/GET retries.
- Test results: The changed expected SHOW DATA output is internally consistent with the five-column metadata. Per the review-only contract, I did not run builds or tests; author/CI results were not independently verified here.
- Observability: Profiles, workload counters, BE metrics, logs, and meta-service bvars were added, but the report RPC omits its FDB read metrics (MMF-1), physical S3 attempts are undercounted (MMF-5/MMF-16), and the persisted total itself is not reliable across crashes or identity transitions.
- Transactions and persistence: The FDB read-modify-write is atomic per attempt, but unequal boot IDs are not an ordering fence (MMF-9), process counters lose unreported crash suffixes (MMF-6), and retired identity records are never compacted (MMF-14). No FE EditLog change is involved.
- Data writes and crash behavior: S3 multipart creation/upload/abort, visible-object deletion, capacity reservation, and stats writes were traced. Crashes can lose totals and multipart ownership, deletion failure releases live capacity, and cleanup can delete another live BE or a long-running query's data (MMF-4/MMF-6/MMF-7/MMF-8/MMF-11).
- FE/BE variable passing: No new scattered FE-to-BE session variable path was introduced; the new MS/FE protobuf and five-column mapping are consistent. The existing heartbeat cloud identity is nevertheless shared and mutable in ways incompatible with its new use as storage/stats identity (MMF-4/MMF-13).
- Performance: Startup recursively materializes the entire spill namespace outside its work budget (MMF-12), and SHOW DATA range-reads every historical backend identity without pagination or compaction (MMF-14).
- Other issues: Three capped normal/risk convergence rounds found no additional distinct issue after the 16 inline findings. Two hypotheses were dismissed with code evidence: dynamic default-vault rebinding is not an explicit contract, and FE-first upgrade is outside the documented MS-first rollout direction.
Focus response: no additional user-provided review focus was supplied; full-scope review was performed.
| const ReportSpillStatsRequest* request, | ||
| ReportSpillStatsResponse* response, | ||
| ::google::protobuf::Closure* done) { | ||
| RPC_PREPROCESS(report_spill_stats, put); |
There was a problem hiding this comment.
report_spill_stats reads the existing record on every periodic report, but this declaration publishes only the put side of the transaction. RPC_PREPROCESS exports detailed KV counters only for the operation names passed here, and the PR adds no report-spill get bvars, so this RPC's FDB read QPS/bytes are invisible under use_detailed_metrics. Please add the get counter/byte bvars and declare this as RPC_PREPROCESS(report_spill_stats, get, put).
| } | ||
| } | ||
|
|
||
| public Cloud.GetSpillStatsResponse getSpillStats(Cloud.GetSpillStatsRequest request) |
There was a problem hiding this comment.
This new proxy method bypasses MetaServiceClientWrapper.executeRequest, unlike comparable calls such as getTabletStats. One UNAVAILABLE/UNKNOWN/retryable timeout or MS_TOO_BUSY response therefore immediately fails SHOW DATA PROPERTIES("entire_warehouse"="true"), and a failed client is not reconnected. Please implement this through executeWithMetrics("getSpillStats", client -> client.getSpillStats(request), Cloud.GetSpillStatsResponse::getStatus) so it retains the standard retry and reconnection semantics.
| return config == "local" || config == "s3"; | ||
| }); | ||
| DEFINE_String(spill_s3_storage_vault, ""); | ||
| DEFINE_mInt64(spill_s3_storage_limit_bytes, "0"); |
There was a problem hiding this comment.
The documented unlimited sentinel is 0, but this mutable config accepts negative values too. SpillDataDir::_reach_limit_unlocked checks only > 0, so setting -1 at startup or dynamically silently disables the billable object-storage cap and exposes a negative limit metric. Please add a validator requiring spill_s3_storage_limit_bytes >= 0 and cover rejection of a negative update.
| void SpillDataDir::init_remote_fs(io::FileSystemSPtr fs, const std::string& cloud_unique_id) { | ||
| DCHECK(_is_remote); | ||
| _fs = std::move(fs); | ||
| _remote_be_root = fmt::format("{}/{}", SPILL_DIR_PREFIX, cloud_unique_id); |
There was a problem hiding this comment.
cloud_unique_id is not unique per BE: CloudSystemInfoService.addBackends generates one value before looping over all requested hosts, ResourceManager supports multiple nodes with that value, and FE sends it to each BE. Two live BEs can therefore both use spill/C/{boot_id}; when either starts cleanup it classifies the other's active boot as old and deletes its spill objects. Their reports also collide in one (instance_id, C) record and are repeatedly folded as restarts. Please include a persisted per-node identity such as heartbeat backend_id in both the object root and stats key, and test two simultaneously live BEs, including existing duplicate cloud IDs.
| COUNTER_UPDATE(_read_file_size, bytes_read); | ||
| ExecEnv::GetInstance()->spill_file_mgr()->update_spill_read_bytes(bytes_read); | ||
| if (_is_remote) { | ||
| // One read_at() is exactly one GET request on object storage. |
There was a problem hiding this comment.
One logical read_at() is not necessarily one object-store GET. S3FileReader::read_at_impl may issue multiple get_object attempts after throttling, and it also issues requests that ultimately fail; this callback runs only once after a successful logical read. Query, workload, and global spill GET counters therefore under-report actual request traffic. Please account attempts at the S3 request boundary (or propagate the per-call attempt delta, including failures) and test retry-then-success plus exhausted failure.
| watch.start(); | ||
| std::vector<io::FileInfo> files; | ||
| bool exists = false; | ||
| RETURN_IF_ERROR(fs->list(be_root, true, &files, &exists)); |
There was a problem hiding this comment.
This eager recursive list materializes every object below the BE root although cleanup needs only distinct first-level boot IDs. S3FileSystem::list_impl first drains the paginated iterator into a complete ObjectMeta vector and then builds a complete FileInfo vector, including current-boot parts. After failed cleanup/restarts this can be arbitrarily large; it ignores the GC time budget and blocks the sole spill-GC thread, which stop() joins. Please use streaming/delimiter-based generation discovery and bound deletion across GC rounds, with a multi-page residue test.
| ReportSpillStatsResponse resp; | ||
| req.set_cloud_unique_id(config::cloud_unique_id); | ||
| auto* stats = req.mutable_stats(); | ||
| stats->set_cloud_unique_id(config::cloud_unique_id); |
There was a problem hiding this comment.
The reporter identity is reread from mutable config::cloud_unique_id on every call, but boot ID and counters are cumulative for the unchanged process. Dropping and re-adding the same live endpoint gives it a new FE-generated ID: after reporting 100 under C1, its next cumulative report of 120 creates C2 while C1 remains, so get_spill_stats returns 220. Please keep an immutable stats identity for the process generation while routing with the current outer ID, or make generations independently idempotent in meta-service; add a same-boot C1-to-C2 test.
| msg = fmt::format("failed to create txn, err={}", err); | ||
| return; | ||
| } | ||
| std::string key0 = stats_spill_key_prefix(instance_id); |
There was a problem hiding this comment.
This range is not bounded by the number of live BEs. Every add/replacement generates a new identity and persistent stats key, ordinary node/cluster removal never removes or folds it, and only whole-instance recycling clears the range. A long-lived elastic instance therefore scans all identities ever created in one transaction and returns every PB, eventually hitting read/response limits although FE needs only the scalar lifetime total. Please compact retired-node totals into a bounded instance aggregate with fenced finalization and add churn coverage for bounded key/read cardinality.
| static_cast<void>(dir->update_capacity()); | ||
| } | ||
| // Both configs are mutable; observe changes without a restart. | ||
| int64_t budget_limit = config::spill_s3_max_inflight_upload_bytes; |
There was a problem hiding this comment.
Startup treats limit < 2 * s3_write_buffer_size as fatal, but this runtime path installs exactly that invalid value after only a warning; both operands are mutable and the update can be persisted. acquire() then admits one oversized buffer whenever inflight reaches zero, so the live setting can exceed its documented bound, and the persisted configuration prevents the next S3-spill restart. Please enforce the same cross-config invariant for dynamic updates, including buffer-size changes, or make startup accept the documented degraded mode; test live-update/restart parity.
| if (_remote_write_stats == nullptr) { | ||
| return; | ||
| } | ||
| ((*_remote_write_stats).*counter)++; |
There was a problem hiding this comment.
This increments once per aggregate SDK call, not per object-store request attempt. The AWS client uses S3CustomRetryStrategy(max_s3_client_retry), so a throttled PutObject/UploadPart that succeeds on its third attempt sends three billable requests but contributes one here; an exhausted operation is likewise under-counted. _finish_part() persists this value as the warehouse PUT-request total. Please attribute initial and retry attempts at the SDK request boundary, including terminal failures, and cover retry success and exhaustion for both operations.
…w of the S3 spill path
- Object root and spill stats record are keyed by the FE-assigned backend_id
instead of cloud_unique_id, which every BE added by one ADD BACKEND
statement shares. SpillStatsPB gains backend_id; a live BE whose id
changes keeps the bound id and logs a warning.
- Startup cleanup discovers old boot generations through a per-boot marker
object (spill/{backend_id}/boots/{boot_id}, refreshed daily) and deletes
one generation per GC round instead of listing the whole BE prefix.
- Meta-service: report_spill_stats declares its get side for the KV bvars,
rejects reports of an older boot_id (a late duplicate or a clock that went
backwards) and never rolls back the totals of the current boot.
- FE getSpillStats goes through executeWithMetrics for retries/reconnect.
- Validators for spill_s3_storage_limit_bytes (>= 0) and
spill_s3_max_inflight_upload_bytes (> 0); a budget below two upload
buffers is a warning (degraded mode) at startup as at runtime; the recycler
skips a non-positive spill_objects_expire_time_second.
- The upload budget over-release invariant is checked in release builds.
- Reporting documented as best-effort; multipart residue guidance added to
the config comments.
Claude-Session: https://claude.ai/code/session_01Jrdwwwh8bZSnVwoCwykHse
|
Thanks for the review. Pushed a follow-up commit addressing it; per finding: Fixed
Kept as designed (happy to discuss)
|
Pure refactor, no behaviour change. The spill store was one class with an
`_is_remote` flag and a branch in every method. It is now:
- spill_data_dir.{h,cpp}: `SpillDataDir`, the abstract base owning what
both kinds share (path, spill root, byte accounting against the limit,
metrics), and `LocalSpillDataDir` (disk probing, spill_gc directory,
usage-based disk selection).
- remote_spill_data_dir.{h,cpp}: `RemoteSpillDataDir` (storage vault
binding, backend_id/boot_id, data and boot-marker key layout). Cloud
headers are only included by its .cpp.
`SpillFileManager` keeps typed views of the stores (`_local_stores` and
at most one `_remote_store`) instead of testing `is_remote()` in every
loop; writers and readers still use the virtual `is_remote()` only to pick
local or remote counters.
Claude-Session: https://claude.ai/code/session_01Jrdwwwh8bZSnVwoCwykHse
…elled queries, drop spill of deleted instances - The upload budget is charged with the allocated capacity of every submitted buffer (s3_write_buffer_size), not with its payload: a partially filled last buffer keeps its full allocation, so with parts smaller than the buffer the budget did not bound memory. The done callback reports the same capacity. A part size below the buffer size is warned at startup. - SpillRemoteUploadBudget::acquire checks cancellation before admitting on the fast path and after every wake-up, so a cancelled query never starts a new upload. - recycle_deleted_instance_data deletes the spill/ prefix on the snapshot-enabled path too, where only referenced rowsets are recycled selectively; spill objects are not referenced by anything. - Tests: PartialBufferIsChargedAtCapacity, UploadBudgetRejectsCancelledQuery Immediately, spill objects in recycle_deleted_instance_with_orphan_tmp_rowset. Claude-Session: https://claude.ai/code/session_01Jrdwwwh8bZSnVwoCwykHse
…ate with every buffer
- Spill objects live under spill/{instance_id}/{backend_id}/... . A storage
vault can be shared by several instances (snapshot clones, rollback heirs)
and backend ids are allocated per FE cluster, so the recycler of one
instance must never touch another instance's spill. RemoteSpillDataDir
learns the instance id through CloudMetaMgr::get_instance_id() when it
binds; both recycler paths (TTL sweep and deleted-instance cleanup) act on
spill/{instance_id}/ only, and the deleted-instance cleanup is restricted
to S3/MOCK accessors like the TTL sweep.
- S3FileWriter::close() builds the buffer of an empty object itself so that
it passes the upload gate like every other buffer; the done callback is
now paired with the gate on every path.
- TQueryStatistics gains spill_write_bytes_to_remote_storage /
spill_read_bytes_from_remote_storage, filled by the BE.
- Meta-service clears server-owned SpillStatsPB fields of a report before
storing it; comments on the budget unit, cancellation and stale reports
updated.
- Tests: GateAndDoneCallbackReportTheSameCapacity (partial PutObject,
multipart with partial last buffer, empty object),
CancelledQueryCloseIssuesNoUpload, other-instance objects in the recycler
tests and in the startup-cleanup test.
Claude-Session: https://claude.ai/code/session_01Jrdwwwh8bZSnVwoCwykHse
2 participants
What problem does this PR solve?
Issue Number: None
Problem Summary:
Cloud-mode BEs could only spill to local disks. This PR adds spill to the instance's S3 storage vault, selected by
be.conf(spill_storage_type = local | s3, mutually exclusive), while keeping the existing spill operators and part-file format. The S3 traffic generated by spill is a billing input, so it is also reported to meta-service and shown inSHOW DATA.BE
RemoteSpillDataDirbound to the storage vault (spill_s3_storage_vault, default: the instance's default vault); objects live underspill/{instance_id}/{backend_id}/data/{boot_id}/{query_id}/...plus one empty marker per boot underspill/{instance_id}/{backend_id}/boots/{boot_id}. The instance id comes first because a vault can be shared by several instances (snapshot clones, rollback heirs) and backend ids are allocated per FE cluster;backend_idis the FE-assigned id (cloud_unique_idis shared by every BE added in oneADD BACKENDstatement, so it cannot identify a BE). The store becomes ready lazily (no meta-service sync during startup); once ready, the GC thread lists only theboots/directory and deletes the data of older boot generations, one generation per GC round.SpillFileWriter/SpillFileReaderreuseS3FileWriter/S3FileReader. Atomic capacity reservation (spill_s3_storage_limit_bytes) and a submit-time upload budget (spill_s3_max_inflight_upload_bytes, charged with the allocated capacity of every in-flight upload buffer) that blocks writers when too many upload buffers are in flight and refuses cancelled queries;FileWriterOptionsgains an upload gate + done callback for this, honoured byS3FileWriter. The spill store is split intoSpillDataDir(base),LocalSpillDataDirandRemoteSpillDataDir.spill_remote_{read,write}_bytes,spill_remote_{get,put}_requestsand per-second throughput/QPS; profile countersSpillRemote*.TQueryStatisticsgainsspill_write_bytes_to_remote_storage/spill_read_bytes_from_remote_storage(BE side filled; FE consumers such as the audit log still show the local fields only).report_spill_stats) about once a minute and once more at shutdown after all tasks are done, with bounded retries (best-effort: a crash loses at most one reporting interval).Meta-service
report_spill_stats/get_spill_statswith keystats/{instance_id}/spill/{backend_id}: one record per BE; a report of a newer boot folds the previous process' totals intoprior_boots_*(so the number of records is bounded by the number of BEs), a report of the same boot replaces without rolling back, a report of an older boot is rejected. The range is removed when the instance is recycled.recycle_expired_spill_objectsremoves spill objects older thanspill_objects_expire_time_second(default 7 days) as a safety net for crashed BEs (S3/MOCK accessors); both the TTL sweep and the deleted-instance cleanup (also on the snapshot-enabled path) touch only this instance'sspill/{instance_id}/prefix, so a vault shared with a live instance is safe. Incomplete multipart uploads of a crashed BE are invisible to prefix listings; anAbortIncompleteMultipartUploadbucket lifecycle rule is the intended safety net for those.FE
SHOW DATA PROPERTIES("entire_warehouse"="true")gains aRemoteSpillWriteSizecolumn; the value is on thetotalrow (spill is not attributable to a database). A meta-service failure is reported instead of showing 0.Design notes and the local review records are kept outside the repository.
Release note
Cloud mode: spill can be written to the S3 storage vault (
spill_storage_type = s3in be.conf);SHOW DATA PROPERTIES("entire_warehouse"="true")shows the bytes uploaded by spill in the newRemoteSpillWriteSizecolumn.Check List (For Author)
Test
Behavior changed:
spill_storage_type,spill_s3_storage_vault,spill_s3_storage_limit_bytes,spill_s3_max_inflight_upload_bytes), new meta-service configspill_objects_expire_time_second, new column inSHOW DATA ... entire_warehouse.Does this need documentation?
Check List (For Reviewer who merge this PR)
https://claude.ai/code/session_01Jrdwwwh8bZSnVwoCwykHse