Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ use std::any::Any;
use std::fmt::Debug;
use std::fmt::Formatter;
use std::sync::Arc;
use std::time::Instant;

use databend_common_exception::Result;
use databend_common_expression::DataBlock;
Expand Down Expand Up @@ -44,6 +45,7 @@ pub struct TransformHashJoin {
projection: ColumnSet,
rf_desc: Arc<RuntimeFiltersDesc>,
runtime_filter_builder: Option<RuntimeFilterLocalBuilder>,
instant: Instant,
}

impl TransformHashJoin {
Expand Down Expand Up @@ -78,6 +80,7 @@ impl TransformHashJoin {
finished: false,
build_data: None,
}),
instant: Instant::now(),
})))
}
}
Expand Down Expand Up @@ -196,6 +199,7 @@ impl Processor for TransformHashJoin {
}

async fn async_process(&mut self) -> Result<()> {
let elapsed = self.instant.elapsed();
let wait_res = self.stage_sync_barrier.wait().await;

self.stage = match &mut self.stage {
Expand All @@ -205,26 +209,77 @@ impl Processor for TransformHashJoin {
self.join.add_runtime_filter_packet(packet);
}

let rf_build_elapsed = self.instant.elapsed() - elapsed;
let _wait_res = self.stage_sync_barrier.wait().await;
let before_wait = self.instant.elapsed();

if wait_res.is_leader() {
let packet = self.join.build_runtime_filter()?;
self.rf_desc.globalization(packet).await?;
}

let _wait_res = self.stage_sync_barrier.wait().await;
let wait_rf_elapsed = self.instant.elapsed() - before_wait;

log::info!(
"HashJoin build stage, sync work elapsed: {:?}, build rf elapsed: {:?}, wait other node rf elapsed: {:?}",
elapsed,
rf_build_elapsed,
wait_rf_elapsed
);

self.instant = Instant::now();
Stage::BuildFinal(BuildFinalState::new())
}
Stage::BuildFinal(_) => Stage::Probe(ProbeState::new()),
Stage::Probe(_) => Stage::ProbeFinal(ProbeFinalState::new()),
Stage::BuildFinal(_) => {
let wait_elapsed = self.instant.elapsed() - elapsed;
log::info!(
"HashJoin build final stage, sync work elapsed: {:?}, wait elapsed: {:?}",
elapsed,
wait_elapsed
);

self.instant = Instant::now();
Stage::Probe(ProbeState::new())
}
Stage::Probe(_) => {
let wait_elapsed = self.instant.elapsed() - elapsed;
log::info!(
"HashJoin probe stage, sync work elapsed: {:?}, wait elapsed: {:?}",
elapsed,
wait_elapsed
);

self.instant = Instant::now();
Stage::ProbeFinal(ProbeFinalState::new())
}
Stage::ProbeFinal(state) => match state.finished {
true => Stage::Finished,
false => Stage::ProbeFinal(ProbeFinalState {
initialize: true,
finished: state.finished,
stream: state.stream.take(),
}),
true => {
let wait_elapsed = self.instant.elapsed() - elapsed;
log::info!(
"HashJoin probe final stage, sync work elapsed: {:?}, wait elapsed: {:?}",
elapsed,
wait_elapsed
);

self.instant = Instant::now();
Stage::Finished
}
false => {
let wait_elapsed = self.instant.elapsed() - elapsed;
log::info!(
"HashJoin probe final stage, sync work elapsed: {:?}, wait elapsed: {:?}",
elapsed,
wait_elapsed
);

self.instant = Instant::now();
Stage::ProbeFinal(ProbeFinalState {
initialize: true,
finished: state.finished,
stream: state.stream.take(),
})
}
},
Stage::Finished => Stage::Finished,
};
Expand Down
Loading