mrhhsg commented on code in PR #68032:
URL: https://github.com/apache/doris/pull/68032#discussion_r4240592979
##########
be/src/io/fs/s3_file_writer.h:
##########
@@ -21,6 +21,7 @@
#include <chrono>
#include <cstddef>
+#include <functional>
Review Comment:
Done in 12097fb2462: `s3_file_writer.{h,cpp}` and `file_writer.h` are
identical to master again. `SpillFileWriter` derives the statistics after a
successful close from `completed_parts()`, `upload_id()` and
`bytes_appended()`, and bounds uploads itself: a remote part is appended one
upload buffer at a time, waiting while the BE's S3 upload queue holds at least
`spill_s3_max_inflight_upload_bytes` of buffers.
##########
be/src/cloud/cloud_meta_mgr.cpp:
##########
@@ -1908,6 +1909,9 @@ Status
CloudMetaMgr::get_storage_vault_info(StorageVaultInfos* vault_infos, bool
}
*is_vault_mode = resp.enable_storage_vault();
+ if (default_vault_id != nullptr) {
+ *default_vault_id = resp.default_storage_vault_id();
Review Comment:
Done in 79622c9849f: the storage vault code (cloud_meta_mgr,
cloud_storage_engine) is back to master. Spill picks
`spill_s3_storage_vault_id` when configured, otherwise the latest object
storage of an instance without storage vaults, otherwise the S3 vault with the
smallest numeric ID as @HappenLee suggested, so adding a vault or changing the
default vault does not move spill. Every spill file keeps the file system it
was created on, so existing files are still read and deleted on their original
vault.
##########
be/src/common/config.cpp:
##########
@@ -1700,6 +1700,28 @@ DEFINE_String(spill_storage_limit, "20%");
// 20%
DEFINE_mInt32(spill_gc_interval_ms, "2000"); // 2s
DEFINE_mInt32(spill_gc_work_time_ms, "2000"); // 2s
DEFINE_mInt64(spill_file_part_size_bytes, "1073741824"); // 1GB
+DEFINE_String(spill_storage_type, "local");
+// s3: cloud mode only; spill is written to the S3 storage vault under
+// spill/{ip}_{port}/{query_id}/, where ip is the address of this BE and port
its
+// heartbeat_service_port.
+// The BE deletes the objects of a spill file when the query is done with it.
What a BE that
+// crashed left behind is not deleted by Doris: give the bucket a lifecycle
rule that expires
+// the keys under spill/ and aborts incomplete multipart uploads.
+DEFINE_Validator(spill_storage_type, [](const std::string& config) -> bool {
+ return config == "local" || config == "s3";
+});
+// Empty means the default vault of the instance, resolved when this BE spills
for the first
+// time; a later SET DEFAULT STORAGE VAULT applies to spill after the BE
restarts.
+DEFINE_String(spill_s3_storage_vault, "");
+DEFINE_mInt64(spill_s3_storage_limit_bytes, "0");
+DEFINE_Validator(spill_s3_storage_limit_bytes, [](int64_t config) -> bool {
return config >= 0; });
Review Comment:
Done in ce1832325c5: renamed to `spill_storage_limit_bytes`. It caps every
spill store separately: for a local path the limit is min(`spill_storage_limit`
percentage/size, `spill_storage_limit_bytes`), and it is the only limit of the
S3 store. 0 means no byte limit.
##########
be/src/common/config.cpp:
##########
@@ -1700,6 +1700,28 @@ DEFINE_String(spill_storage_limit, "20%");
// 20%
DEFINE_mInt32(spill_gc_interval_ms, "2000"); // 2s
DEFINE_mInt32(spill_gc_work_time_ms, "2000"); // 2s
DEFINE_mInt64(spill_file_part_size_bytes, "1073741824"); // 1GB
+DEFINE_String(spill_storage_type, "local");
+// s3: cloud mode only; spill is written to the S3 storage vault under
+// spill/{ip}_{port}/{query_id}/, where ip is the address of this BE and port
its
+// heartbeat_service_port.
+// The BE deletes the objects of a spill file when the query is done with it.
What a BE that
+// crashed left behind is not deleted by Doris: give the bucket a lifecycle
rule that expires
+// the keys under spill/ and aborts incomplete multipart uploads.
+DEFINE_Validator(spill_storage_type, [](const std::string& config) -> bool {
+ return config == "local" || config == "s3";
+});
+// Empty means the default vault of the instance, resolved when this BE spills
for the first
+// time; a later SET DEFAULT STORAGE VAULT applies to spill after the BE
restarts.
+DEFINE_String(spill_s3_storage_vault, "");
Review Comment:
Yes. Renamed to `spill_s3_storage_vault_id` in ce1832325c5; the value is a
vault ID, not a name.
##########
be/src/exec/spill/spill_file_manager.cpp:
##########
@@ -28,21 +29,58 @@
#include <utility>
#include <vector>
+#include "common/config.h"
#include "common/logging.h"
#include "common/metrics/doris_metrics.h"
+#include "exec/spill/remote_spill_data_dir.h"
#include "exec/spill/spill_file.h"
#include "io/fs/file_system.h"
#include "io/fs/local_file_system.h"
#include "runtime/query_context.h"
-#include "storage/olap_define.h"
#include "util/debug_points.h"
-#include "util/parse_util.h"
#include "util/pretty_printer.h"
#include "util/time.h"
#include "util/uid_util.h"
namespace doris {
+Status reconcile_remote_spill_bytes(SpillDataDir* data_dir, const
io::FileSystemSPtr& fs,
+ const std::string& dir, int64_t*
accounted_bytes) {
+ DORIS_CHECK(data_dir->is_remote());
+ std::vector<io::FileInfo> objects;
+ bool exists = false;
+ RETURN_IF_ERROR(fs->list(dir, true, &objects, &exists));
+ int64_t actual_bytes = 0;
+ for (const auto& object : objects) {
+ if (object.is_file) {
+ actual_bytes += object.file_size;
+ }
+ }
+ const int64_t delta = actual_bytes - *accounted_bytes;
+ if (delta > 0) {
+ data_dir->record_persisted_bytes(delta);
+ } else if (delta < 0) {
+ data_dir->release_persisted_bytes(-delta);
+ }
+ *accounted_bytes = actual_bytes;
+ return Status::OK();
+}
+
+// BE-wide object storage traffic of spill. The PerSecond windows give QPS and
bytes/s directly.
+bvar::Adder<int64_t> g_spill_remote_read_bytes("spill_remote_read_bytes");
+bvar::Adder<int64_t> g_spill_remote_write_bytes("spill_remote_write_bytes");
+bvar::Adder<int64_t> g_spill_remote_get_requests("spill_remote_get_requests");
+bvar::Adder<int64_t> g_spill_remote_put_requests("spill_remote_put_requests");
+bvar::PerSecond<bvar::Adder<int64_t>>
g_spill_remote_read_throughput("spill_remote_read_throughput",
Review Comment:
Done in cfd601265ab: the `bvar::PerSecond` wrappers are removed; only the
four adders remain.
--
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]
