yiguolei commented on code in PR #69635:
URL: https://github.com/apache/doris/pull/69635#discussion_r4533831418
##########
be/src/exec/spill/spill_file_manager.h:
##########
@@ -174,44 +117,101 @@ class SpillFileManager {
// Delete SpillFile data synchronously.
void delete_spill_file(SpillFileSPtr spill_file);
- // Recursively delete a per-query spill directory during query teardown.
Failed deletions are
- // retained by the manager and retried by its GC and shutdown paths.
+ // Recursively delete a per-query spill directory of a local store during
query teardown.
+ // Failed deletions are retained by the manager and retried by its GC and
shutdown paths.
+ // A no-op for the remote store: its spill is deleted per spill file only.
void delete_query_spill_directory(const std::string& query_id,
SpillDataDir* data_dir);
+ // Take over a spill file directory whose deletion failed. Its objects are
still stored, so
+ // `charged_bytes` stay charged to `data_dir` until a retry deletes them;
otherwise a
+ // deletion outage would free capacity that is still in use.
+ void retry_spill_directory_deletion(SpillDataDir* data_dir,
io::FileSystemSPtr fs,
+ std::string dir, int64_t charged_bytes,
+ int64_t persisted_bytes, std::string
query_id,
+ std::vector<std::string>
file_log_contexts = {});
+
void gc(int32_t max_work_time_ms);
void update_spill_write_bytes(int64_t bytes) {
_spill_write_bytes_counter->increment(bytes); }
void update_spill_read_bytes(int64_t bytes) {
_spill_read_bytes_counter->increment(bytes); }
+ /// Bytes successfully uploaded by spill to object storage.
+ void update_spill_remote_write(int64_t bytes);
+
+ /// Bytes and GET requests issued by spill against object storage.
+ void update_spill_remote_read(int64_t bytes, int64_t get_requests);
+
+ /// Number of spill directories whose deletion failed and is being retried.
+ size_t pending_delete_dir_count();
+
+ /// Bytes of completed spill objects this process currently holds in
object storage;
+ /// uncommitted capacity reservations are excluded. Sent to the master FE
with every disk
+ /// report (cloud mode) for SHOW DATA.
+ int64_t remote_spill_data_bytes();
+
private:
friend class ExternalSpillSession;
- struct PendingQuerySpillDirectory {
+ struct PendingSpillDirectory {
int failed_count {0};
- std::string query_dir;
+ // A query directory or the directory of one spill file.
+ std::string dir;
+ std::string query_id;
+ SpillDataDir* data_dir {nullptr};
+ // Pinned for remote spill: a later change of the spill vault must not
redirect GC.
+ io::FileSystemSPtr fs;
+ // Bytes of the objects under `dir` still charged to `data_dir`;
released once deleted.
+ int64_t charged_bytes {0};
+ int64_t persisted_bytes {0};
+ // Per-file identities/sizes survive directory merging, until
successful deletion.
+ std::vector<std::string> file_log_contexts;
+ };
+
+ struct SpillGcStats {
+ bool has_work = false;
+ size_t backlog_dirs = 0;
+ size_t deleted_dirs = 0;
+ size_t deleted_files = 0;
+ size_t failed_deletes = 0;
};
void _init_metrics();
Status _init_spill_store_map();
void _spill_gc_thread_callback();
- Status _try_delete_query_spill_directory(const PendingQuerySpillDirectory&
pending_directory);
- void _retry_pending_query_spill_directories();
+ Status _try_delete_spill_directory(const PendingSpillDirectory&
pending_directory);
+ /// Retry the queued failed deletions, at least one, until
`max_work_time_ns` has passed on
+ /// `watch`; a remote one can take longer, since it waits for the object
store. Remote ones
+ /// are only retried when `retry_remote` is set. The rest stay queued for
the next round.
+ void _retry_pending_spill_directories(const MonotonicStopWatch& watch,
int64_t max_work_time_ns,
+ bool retry_remote);
+ /// Retry every queued failed deletion of a local store, for shutdown: it
does not wait for
+ /// the object store.
+ void _retry_pending_local_spill_directories();
+ void _gc_spill_store(LocalSpillDataDir* store_dir, const
MonotonicStopWatch& watch,
+ int64_t max_work_time_ns, SpillGcStats* stats);
+ /// Queue a failed deletion, merged with a pending ancestor or absorbing
pending descendants
+ /// of the same store so that an outage keeps about one entry per query.
+ void _add_pending_directory(PendingSpillDirectory pending_directory);
Status _initialize_external_spill_session(ExternalSpillSession*
spill_session);
void _release_external_spill_session(ExternalSpillSession* spill_session);
std::vector<SpillDataDir*> _get_stores_for_spill(TStorageMedium::type
storage_medium);
- SpillDataDir* _get_store_for_spill();
+ std::vector<SpillDataDir*>
_get_local_stores_for_spill(TStorageMedium::type storage_medium);
+ SpillDataDir* _get_local_store_for_external_spill();
std::unordered_map<std::string, std::unique_ptr<SpillDataDir>>
_spill_store_map;
+ // Views of _spill_store_map by kind. With spill_storage_type=s3 a BE has
the remote store
+ // for query spill and keeps its local stores for external writers.
+ std::vector<LocalSpillDataDir*> _local_stores;
+ RemoteSpillDataDir* _remote_store = nullptr;
CountDownLatch _stop_background_threads_latch;
std::shared_ptr<Thread> _spill_gc_thread;
- // Query cleanup uses the regular pending-deletion path. External leases
only defer deletion
- // while an SDK task can still access the same query directory; filesystem
I/O never holds this
- // mutex.
- std::mutex _pending_query_spill_directories_mutex;
- std::vector<PendingQuerySpillDirectory> _pending_query_spill_directories;
+ // External spill leases defer query-directory deletion until native
callbacks finish.
+ // Filesystem I/O never holds this mutex.
+ std::mutex _pending_spill_directories_mutex;
+ std::vector<PendingSpillDirectory> _pending_spill_directories;
Review Comment:
We need a deep discussion about PendingSpillDirectory
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
