mrhhsg commented on code in PR #68032:
URL: https://github.com/apache/doris/pull/68032#discussion_r4240593455
##########
be/src/runtime/exec_env_init.cpp:
##########
@@ -232,8 +234,15 @@ Status ExecEnv::_init(const std::vector<StorePath>&
store_paths,
return Status::OK();
}
std::unordered_map<std::string, std::unique_ptr<SpillDataDir>>
spill_store_map;
+ if (config::spill_storage_type == "s3") {
Review Comment:
Done in cfd601265ab: `SpillFileManager` takes the spill store paths and
builds its stores in `init()`, which also validates `spill_storage_type=s3`
(cloud mode only). `exec_env_init.cpp` and `doris_main.cpp` no longer contain
spill-specific setup beyond allowing empty local spill paths in s3 mode.
##########
fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditLoader.java:
##########
@@ -178,6 +178,8 @@ private void fillLogBuffer(AuditEvent event, StringBuilder
logBuffer) {
logBuffer.append(event.shuffleSendBytes).append(AUDIT_TABLE_COL_SEPARATOR);
logBuffer.append(event.spillWriteBytesToLocalStorage).append(AUDIT_TABLE_COL_SEPARATOR);
logBuffer.append(event.spillReadBytesFromLocalStorage).append(AUDIT_TABLE_COL_SEPARATOR);
+
logBuffer.append(event.spillWriteBytesToRemoteStorage).append(AUDIT_TABLE_COL_SEPARATOR);
Review Comment:
Update: the retry design described above was removed in eb98d8947ea. It made
a single filtered row stall audit loading and could produce over-long labels.
The audit loader again discards a failed batch as on master; only the send-time
projection remains, which drops the two new columns while the audit table still
has the old schema, so a follower-first upgrade does not lose rows.
##########
fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditLoader.java:
##########
@@ -269,32 +273,31 @@ public synchronized void loadIfNecessary(boolean force) {
if (auditLogBuffer.length() != 0 && (force || auditLogBuffer.length()
>= GlobalVariable.auditPluginMaxBatchBytes
|| currentTime - lastLoadTimeAuditLog >=
GlobalVariable.auditPluginMaxBatchInternalSec * 1000)) {
- // begin to load
+ // Keep the batch until a successful load. In a follower-first
upgrade the table can
+ // still have its old schema, and a rejected request must not
erase audit rows.
try {
- String token = "";
- try {
- // Acquire token from master
- token =
Env.getCurrentEnv().getTokenManager().acquireToken();
- } catch (Exception e) {
- LOG.warn("Failed to get auth token: {}", e);
- discardLogNum += auditLogNum;
- return;
+ if (batchLabel == null) {
+ batchLabel = streamLoader.genLabel();
}
- AuditStreamLoader.LoadResponse response =
streamLoader.loadBatch(auditLogBuffer, token);
- if (LOG.isDebugEnabled()) {
- LOG.debug("audit loader response: {}", response);
- }
- } catch (Exception e) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("encounter exception when putting current audit
batch, discard current batch", e);
+ String token =
Env.getCurrentEnv().getTokenManager().acquireToken();
+ AuditStreamLoader.LoadResponse response =
streamLoader.loadBatch(auditLogBuffer, token, batchLabel);
+ if (!response.succeeded(auditLogNum)) {
+ failedLoad = true;
+ if (response.rejectedOrIncomplete(auditLogNum)) {
Review Comment:
No longer applicable: eb98d8947ea removed the retry and label reuse; audit
loads behave as on master (one label per attempt, failed batches discarded).
##########
be/src/cloud/cloud_storage_engine.cpp:
##########
@@ -452,12 +452,18 @@ Status
CloudStorageEngine::start_bg_threads(std::shared_ptr<WorkloadGroup> wg_sp
void CloudStorageEngine::sync_storage_vault() {
cloud::StorageVaultInfos vault_infos;
bool enable_storage_vault = false;
+ std::string default_vault_id;
- auto st = _meta_mgr->get_storage_vault_info(&vault_infos,
&enable_storage_vault);
+ auto st = _meta_mgr->get_storage_vault_info(&vault_infos,
&enable_storage_vault,
+ &default_vault_id);
if (!st.ok()) {
LOG(WARNING) << "failed to get storage vault info. err=" << st;
return;
}
+ {
+ std::lock_guard lock(_latest_fs_mtx);
+ _default_vault_id = std::move(default_vault_id);
Review Comment:
No longer applicable: 79622c9849f reverted the default-vault tracking in
cloud_storage_engine; spill no longer follows the default vault.
##########
be/src/exec/spill/spill_file_manager.cpp:
##########
@@ -153,14 +199,16 @@ void SpillFileManager::stop() {
// The GC thread may observe the stop latch before processing a recently
queued failed deletion.
// Retry the pending directories after the thread exits; later failures
get one final retry in
// the destructor.
- _retry_pending_query_spill_directories();
+ _retry_pending_spill_directories();
}
Status SpillFileManager::init() {
LOG(INFO) << "init spill stream manager";
RETURN_IF_ERROR(_init_spill_store_map());
+ _remote_upload_budget =
+
std::make_shared<SpillRemoteUploadBudget>(config::spill_s3_max_inflight_upload_bytes);
- for (const auto& [path, store] : _spill_store_map) {
+ for (auto* store : _local_stores) {
Review Comment:
Not addressed in this PR; kept as a known limitation in the PR description.
##########
be/src/exec/spill/spill_file_manager.cpp:
##########
@@ -431,10 +622,10 @@ void SpillFileManager::gc(int32_t max_work_time_ms) {
LOG(INFO) << msg;
}
}};
- _retry_pending_query_spill_directories();
- for (const auto& [path, store_dir] : _spill_store_map) {
+ _retry_pending_spill_directories();
Review Comment:
Not addressed in this PR; kept as a known limitation in the PR description.
--
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]
