1use std::collections::{HashMap, VecDeque};
20use std::sync::Arc;
21
22use common_telemetry::{info, warn};
23use store_api::logstore::LogStore;
24use store_api::storage::RegionId;
25
26use crate::cache::file_cache::{FileType, IndexKey};
27use crate::cache::CacheManagerRef;
28use crate::error::{RegionBusySnafu, RegionNotFoundSnafu, Result};
29use crate::manifest::action::{
30 RegionChange, RegionEdit, RegionMetaAction, RegionMetaActionList, RegionTruncate,
31};
32use crate::metrics::WRITE_CACHE_INFLIGHT_DOWNLOAD;
33use crate::region::version::VersionBuilder;
34use crate::region::{MitoRegionRef, RegionLeaderState, RegionRoleState};
35use crate::request::{
36 BackgroundNotify, OptionOutputTx, RegionChangeResult, RegionEditRequest, RegionEditResult,
37 RegionSyncRequest, TruncateResult, WorkerRequest,
38};
39use crate::sst::location;
40use crate::worker::{RegionWorkerLoop, WorkerListener};
41
42pub(crate) type RegionEditQueues = HashMap<RegionId, RegionEditQueue>;
43
44pub(crate) struct RegionEditQueue {
49 region_id: RegionId,
50 requests: VecDeque<RegionEditRequest>,
51}
52
53impl RegionEditQueue {
54 const QUEUE_MAX_LEN: usize = 128;
55
56 fn new(region_id: RegionId) -> Self {
57 Self {
58 region_id,
59 requests: VecDeque::new(),
60 }
61 }
62
63 fn enqueue(&mut self, request: RegionEditRequest) {
64 if self.requests.len() > Self::QUEUE_MAX_LEN {
65 let _ = request.tx.send(
66 RegionBusySnafu {
67 region_id: self.region_id,
68 }
69 .fail(),
70 );
71 return;
72 };
73 self.requests.push_back(request);
74 }
75
76 fn dequeue(&mut self) -> Option<RegionEditRequest> {
77 self.requests.pop_front()
78 }
79}
80
81impl<S: LogStore> RegionWorkerLoop<S> {
82 pub(crate) async fn handle_manifest_region_change_result(
84 &mut self,
85 change_result: RegionChangeResult,
86 ) {
87 let region = match self.regions.get_region(change_result.region_id) {
88 Some(region) => region,
89 None => {
90 self.reject_region_stalled_requests(&change_result.region_id);
91 change_result.sender.send(
92 RegionNotFoundSnafu {
93 region_id: change_result.region_id,
94 }
95 .fail(),
96 );
97 return;
98 }
99 };
100
101 if change_result.result.is_ok() {
102 region
104 .version_control
105 .alter_schema(change_result.new_meta, ®ion.memtable_builder);
106
107 let version = region.version();
108 info!(
109 "Region {} is altered, metadata is {:?}, options: {:?}",
110 region.region_id, version.metadata, version.options,
111 );
112 }
113
114 region.switch_state_to_writable(RegionLeaderState::Altering);
116 change_result.sender.send(change_result.result.map(|_| 0));
118
119 self.handle_region_stalled_requests(&change_result.region_id)
121 .await;
122 }
123
124 pub(crate) async fn handle_region_sync(&mut self, request: RegionSyncRequest) {
129 let region_id = request.region_id;
130 let sender = request.sender;
131 let region = match self.regions.follower_region(region_id) {
132 Ok(region) => region,
133 Err(e) => {
134 let _ = sender.send(Err(e));
135 return;
136 }
137 };
138
139 let original_manifest_version = region.manifest_ctx.manifest_version().await;
140 let manifest = match region
141 .manifest_ctx
142 .install_manifest_to(request.manifest_version)
143 .await
144 {
145 Ok(manifest) => manifest,
146 Err(e) => {
147 let _ = sender.send(Err(e));
148 return;
149 }
150 };
151 let version = region.version();
152 if !version.memtables.is_empty() {
153 warn!(
154 "Region {} memtables is not empty, which should not happen, manifest version: {}",
155 region.region_id, manifest.manifest_version
156 );
157 }
158 let region_options = version.options.clone();
159 let new_mutable = Arc::new(
160 region
161 .version()
162 .memtables
163 .mutable
164 .new_with_part_duration(version.compaction_time_window),
165 );
166 let metadata = manifest.metadata.clone();
167 let version = VersionBuilder::new(metadata, new_mutable)
168 .add_files(region.file_purger.clone(), manifest.files.values().cloned())
169 .flushed_entry_id(manifest.flushed_entry_id)
170 .flushed_sequence(manifest.flushed_sequence)
171 .truncated_entry_id(manifest.truncated_entry_id)
172 .compaction_time_window(manifest.compaction_time_window)
173 .options(region_options)
174 .build();
175 region.version_control.overwrite_current(Arc::new(version));
176
177 let updated = manifest.manifest_version > original_manifest_version;
178 let _ = sender.send(Ok((manifest.manifest_version, updated)));
179 }
180}
181
182impl<S> RegionWorkerLoop<S> {
183 pub(crate) async fn handle_region_edit(&mut self, request: RegionEditRequest) {
185 let region_id = request.region_id;
186 let Some(region) = self.regions.get_region(region_id) else {
187 let _ = request.tx.send(RegionNotFoundSnafu { region_id }.fail());
188 return;
189 };
190
191 if !region.is_writable() {
192 if region.state() == RegionRoleState::Leader(RegionLeaderState::Editing) {
193 self.region_edit_queues
194 .entry(region_id)
195 .or_insert_with(|| RegionEditQueue::new(region_id))
196 .enqueue(request);
197 } else {
198 let _ = request.tx.send(RegionBusySnafu { region_id }.fail());
199 }
200 return;
201 }
202
203 let RegionEditRequest {
204 region_id: _,
205 edit,
206 tx: sender,
207 } = request;
208
209 if let Err(e) = region.set_editing() {
211 let _ = sender.send(Err(e));
212 return;
213 }
214
215 let request_sender = self.sender.clone();
216 let cache_manager = self.cache_manager.clone();
217 let listener = self.listener.clone();
218 common_runtime::spawn_global(async move {
221 let result = edit_region(®ion, edit.clone(), cache_manager, listener).await;
222 let notify = WorkerRequest::Background {
223 region_id,
224 notify: BackgroundNotify::RegionEdit(RegionEditResult {
225 region_id,
226 sender,
227 edit,
228 result,
229 }),
230 };
231 if let Err(res) = request_sender.send(notify).await {
233 warn!(
234 "Failed to send region edit result back to the worker, region_id: {}, res: {:?}",
235 region_id, res
236 );
237 }
238 });
239 }
240
241 pub(crate) async fn handle_region_edit_result(&mut self, edit_result: RegionEditResult) {
243 let region = match self.regions.get_region(edit_result.region_id) {
244 Some(region) => region,
245 None => {
246 let _ = edit_result.sender.send(
247 RegionNotFoundSnafu {
248 region_id: edit_result.region_id,
249 }
250 .fail(),
251 );
252 return;
253 }
254 };
255
256 let need_compaction =
257 edit_result.result.is_ok() && !edit_result.edit.files_to_add.is_empty();
258
259 if edit_result.result.is_ok() {
260 region
262 .version_control
263 .apply_edit(edit_result.edit, &[], region.file_purger.clone());
264 }
265
266 region.switch_state_to_writable(RegionLeaderState::Editing);
268
269 let _ = edit_result.sender.send(edit_result.result);
270
271 if let Some(edit_queue) = self.region_edit_queues.get_mut(&edit_result.region_id) {
272 if let Some(request) = edit_queue.dequeue() {
273 self.handle_region_edit(request).await;
274 }
275 }
276
277 if need_compaction {
278 self.schedule_compaction(®ion).await;
279 }
280 }
281
282 pub(crate) fn handle_manifest_truncate_action(
284 &self,
285 region: MitoRegionRef,
286 truncate: RegionTruncate,
287 sender: OptionOutputTx,
288 ) {
289 if let Err(e) = region.set_truncating() {
292 sender.send(Err(e));
293 return;
294 }
295 let request_sender = self.sender.clone();
298 let manifest_ctx = region.manifest_ctx.clone();
299
300 common_runtime::spawn_global(async move {
302 let action_list =
304 RegionMetaActionList::with_action(RegionMetaAction::Truncate(truncate.clone()));
305
306 let result = manifest_ctx
307 .update_manifest(RegionLeaderState::Truncating, action_list)
308 .await
309 .map(|_| ());
310
311 let truncate_result = TruncateResult {
313 region_id: truncate.region_id,
314 sender,
315 result,
316 truncated_entry_id: truncate.truncated_entry_id,
317 truncated_sequence: truncate.truncated_sequence,
318 };
319 let _ = request_sender
320 .send(WorkerRequest::Background {
321 region_id: truncate.region_id,
322 notify: BackgroundNotify::Truncate(truncate_result),
323 })
324 .await
325 .inspect_err(|_| warn!("failed to send truncate result"));
326 });
327 }
328
329 pub(crate) fn handle_manifest_region_change(
331 &self,
332 region: MitoRegionRef,
333 change: RegionChange,
334 sender: OptionOutputTx,
335 ) {
336 if let Err(e) = region.set_altering() {
338 sender.send(Err(e));
339 return;
340 }
341 let listener = self.listener.clone();
342 let request_sender = self.sender.clone();
343 common_runtime::spawn_global(async move {
345 let new_meta = change.metadata.clone();
346 let action_list = RegionMetaActionList::with_action(RegionMetaAction::Change(change));
347
348 let result = region
349 .manifest_ctx
350 .update_manifest(RegionLeaderState::Altering, action_list)
351 .await
352 .map(|_| ());
353 let notify = WorkerRequest::Background {
354 region_id: region.region_id,
355 notify: BackgroundNotify::RegionChange(RegionChangeResult {
356 region_id: region.region_id,
357 sender,
358 result,
359 new_meta,
360 }),
361 };
362 listener
363 .on_notify_region_change_result_begin(region.region_id)
364 .await;
365
366 if let Err(res) = request_sender.send(notify).await {
367 warn!(
368 "Failed to send region change result back to the worker, region_id: {}, res: {:?}",
369 region.region_id, res
370 );
371 }
372 });
373 }
374}
375
376async fn edit_region(
378 region: &MitoRegionRef,
379 edit: RegionEdit,
380 cache_manager: CacheManagerRef,
381 listener: WorkerListener,
382) -> Result<()> {
383 let region_id = region.region_id;
384 if let Some(write_cache) = cache_manager.write_cache() {
385 for file_meta in &edit.files_to_add {
386 let write_cache = write_cache.clone();
387 let layer = region.access_layer.clone();
388 let listener = listener.clone();
389
390 let index_key = IndexKey::new(region_id, file_meta.file_id, FileType::Parquet);
391 let remote_path = location::sst_file_path(layer.region_dir(), file_meta.file_id);
392
393 let is_index_exist = file_meta.exists_index();
394 let index_file_size = file_meta.index_file_size();
395
396 let index_file_index_key =
397 IndexKey::new(region_id, file_meta.file_id, FileType::Puffin);
398 let index_remote_path =
399 location::index_file_path(layer.region_dir(), file_meta.file_id);
400
401 let file_size = file_meta.file_size;
402 common_runtime::spawn_global(async move {
403 WRITE_CACHE_INFLIGHT_DOWNLOAD.add(1);
404
405 if write_cache
406 .download(index_key, &remote_path, layer.object_store(), file_size)
407 .await
408 .is_ok()
409 {
410 let _ = write_cache
413 .file_cache()
414 .get_parquet_meta_data(index_key)
415 .await;
416
417 listener.on_file_cache_filled(index_key.file_id);
418 }
419 if is_index_exist {
420 if let Err(err) = write_cache
422 .download(
423 index_file_index_key,
424 &index_remote_path,
425 layer.object_store(),
426 index_file_size,
427 )
428 .await
429 {
430 common_telemetry::error!(
431 err; "Failed to download puffin file, region_id: {}, index_file_index_key: {:?}, index_remote_path: {}", region_id, index_file_index_key, index_remote_path
432 );
433 }
434 }
435
436 WRITE_CACHE_INFLIGHT_DOWNLOAD.sub(1);
437 });
438 }
439 }
440
441 info!("Applying {edit:?} to region {}", region_id);
442
443 let action_list = RegionMetaActionList::with_action(RegionMetaAction::Edit(edit));
444 region
445 .manifest_ctx
446 .update_manifest(RegionLeaderState::Editing, action_list)
447 .await
448 .map(|_| ())
449}