docs: explain nodeid use in check task

This commit is contained in:
discord9
2025-04-21 14:35:05 +08:00
parent 82cee11eea
commit d2c4767d41
2 changed files with 6 additions and 0 deletions

View File

@@ -158,6 +158,7 @@ pub struct FlowWorkerManager {
flow_err_collectors: RwLock<BTreeMap<FlowId, ErrCollector>>,
src_send_buf_lens: RwLock<BTreeMap<TableId, watch::Receiver<usize>>>,
tick_manager: FlowTickManager,
/// This node id is only available in distributed mode, on standalone mode this is guaranteed to be `None`
pub node_id: Option<u32>,
/// Lock for flushing, will be `read` by `handle_inserts` and `write` by `flush_flow`
///

View File

@@ -126,8 +126,11 @@ impl FlowDualEngine {
allow_create: bool,
allow_drop: bool,
) -> Result<(), Error> {
// use nodeid to determine if this is standalone/distributed mode, and retrieve all flows in this node(in distributed mode)/or all flows(in standalone mode)
let nodeid = self.streaming_engine.node_id;
let should_exists: Vec<_> = if let Some(nodeid) = nodeid {
// nodeid is available, so we only need to check flows on this node
// which also means we are in distributed mode
let to_be_recover = self
.flow_metadata_manager
.flownode_flow_manager()
@@ -139,6 +142,8 @@ impl FlowDualEngine {
})?;
to_be_recover.into_iter().map(|(id, _)| id).collect()
} else {
// nodeid is not available, so we need to check all flows
// which also means we are in standalone mode
let all_catalogs = self
.catalog_manager
.catalog_names()