use std::fmt;
use std::sync::Arc;
use arrow::datatypes::SchemaRef;
use base64::Engine as _;
use datafusion::error::DataFusionError;
use datafusion::execution::TaskContext;
use datafusion::logical_expr::execution_props::ScalarSubqueryResults;
use datafusion::physical_expr::EquivalenceProperties;
use datafusion::physical_plan::coalesce_partitions::CoalescePartitionsExec;
use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
use datafusion::physical_plan::repartition::RepartitionExec;
use datafusion::physical_plan::scalar_subquery::{ScalarSubqueryExec, ScalarSubqueryLink};
use datafusion::physical_plan::sorts::sort::SortExec;
use datafusion::physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
use datafusion::physical_plan::{
DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties as _, Partitioning,
PlanProperties, SendableRecordBatchStream,
};
use datafusion::prelude::SessionContext;
use datafusion_proto::physical_plan::PhysicalExtensionCodec;
use futures::{StreamExt as _, TryStreamExt as _};
use crate::{SqlError, SqlResult};
pub const DFPLAN_BODY_PREFIX: &str = "dfplan:v1:";
pub const STAGE_TARGET_PARTITIONS_ENV: &str = "KRISHIV_STAGE_TARGET_PARTITIONS";
pub const STAGE_SPLIT_ENV: &str = "KRISHIV_STAGE_SPLIT";
pub const BROADCAST_JOIN_BYTES_ENV: &str = "KRISHIV_BROADCAST_JOIN_BYTES";
const DEFAULT_BROADCAST_JOIN_BYTES: usize = 32 * 1024 * 1024;
const DEFAULT_BROADCAST_JOIN_ROWS: usize = 1_000_000;
const TASKS_PER_SLOT: usize = 2;
const MIN_STAGE_PARTITIONS: usize = 2;
const MAX_STAGE_PARTITIONS: usize = 512;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ClusterCapacity {
pub total_slots: usize,
}
#[must_use]
pub fn resolve_stage_target_partitions(cluster: Option<ClusterCapacity>) -> usize {
derive_stage_target_partitions(
std::env::var(STAGE_TARGET_PARTITIONS_ENV)
.ok()
.and_then(|v| v.trim().parse::<usize>().ok()),
cluster,
std::thread::available_parallelism()
.map(std::num::NonZeroUsize::get)
.unwrap_or(1),
)
}
#[must_use]
pub fn derive_stage_target_partitions(
explicit: Option<usize>,
cluster: Option<ClusterCapacity>,
local_cores: usize,
) -> usize {
if let Some(explicit) = explicit.filter(|&n| n >= MIN_STAGE_PARTITIONS) {
return explicit;
}
cluster
.map_or(local_cores, |c| c.total_slots)
.saturating_mul(TASKS_PER_SLOT)
.clamp(MIN_STAGE_PARTITIONS, MAX_STAGE_PARTITIONS)
}
pub fn stage_split_enabled() -> bool {
!matches!(
std::env::var(STAGE_SPLIT_ENV)
.unwrap_or_default()
.trim()
.to_ascii_lowercase()
.as_str(),
"off" | "0" | "false" | "disabled"
)
}
pub fn planning_session_context(target_partitions: usize) -> SessionContext {
planning_session_context_with_join_threshold(target_partitions, None)
}
pub fn planning_session_context_with_join_threshold(
target_partitions: usize,
spill_join_build_bytes: Option<u64>,
) -> SessionContext {
planning_session_context_with_options(target_partitions, spill_join_build_bytes, None)
}
pub fn planning_session_context_with_options(
target_partitions: usize,
spill_join_build_bytes: Option<u64>,
broadcast_join_bytes: Option<usize>,
) -> SessionContext {
let tp = std::num::NonZeroUsize::new(target_partitions.max(1))
.unwrap_or(std::num::NonZeroUsize::MIN);
let mut config = crate::build_single_node_session_config(tp, None);
config
.options_mut()
.optimizer
.enable_round_robin_repartition = false;
let broadcast_bytes = broadcast_join_bytes.unwrap_or_else(|| {
std::env::var(BROADCAST_JOIN_BYTES_ENV)
.ok()
.and_then(|v| v.trim().parse::<usize>().ok())
.filter(|n| *n > 0)
.unwrap_or(DEFAULT_BROADCAST_JOIN_BYTES)
});
let opts = config.options_mut();
opts.optimizer.hash_join_single_partition_threshold = broadcast_bytes;
opts.optimizer.hash_join_single_partition_threshold_rows = if broadcast_bytes == 0 {
0
} else {
DEFAULT_BROADCAST_JOIN_ROWS
};
let state_builder = crate::with_krishiv_optimizer_rules_with_join_threshold(
datafusion::execution::session_state::SessionStateBuilder::new().with_default_features(),
spill_join_build_bytes,
)
.with_config(config);
let state_builder = match datafusion::execution::runtime_env::RuntimeEnvBuilder::new()
.with_object_store_registry(Arc::new(
crate::object_store_registry::LazyCloudObjectStoreRegistry::new(),
))
.build_arc()
{
Ok(runtime) => state_builder.with_runtime_env(runtime),
Err(error) => {
tracing::warn!(%error, "cloud object-store registry unavailable for staged planning");
state_builder
}
};
SessionContext::new_with_state(state_builder.build())
}
pub fn shuffle_stage_key(stage_index: usize, map_task_index: usize) -> String {
format!("s{stage_index}.m{map_task_index}")
}
pub fn encode_dfplan_bytes(
plan: Arc<dyn ExecutionPlan>,
codec: &dyn PhysicalExtensionCodec,
) -> SqlResult<Vec<u8>> {
datafusion_proto::bytes::physical_plan_to_bytes_with_extension_codec(plan, codec)
.map(|b| b.to_vec())
.map_err(|e| SqlError::DataFusion {
message: format!("physical plan proto encode: {e}"),
})
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DfplanMapRange {
pub upstream_stage_index: usize,
pub start: usize,
pub end: usize,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DfplanTaskSpec {
pub partitions: Vec<usize>,
pub map_range: Option<DfplanMapRange>,
}
impl DfplanTaskSpec {
pub fn single(partition: usize) -> Self {
Self {
partitions: vec![partition],
map_range: None,
}
}
fn render(&self) -> String {
let mut out = self
.partitions
.iter()
.map(usize::to_string)
.collect::<Vec<_>>()
.join(",");
if let Some(range) = &self.map_range {
out.push_str(&format!(
"/s{}m{}-{}",
range.upstream_stage_index, range.start, range.end
));
}
out
}
}
pub fn dfplan_task_body(plan_bytes_b64: &str, partition: usize) -> String {
format!("{DFPLAN_BODY_PREFIX}{partition}:{plan_bytes_b64}")
}
pub fn dfplan_task_body_for_spec(plan_bytes_b64: &str, spec: &DfplanTaskSpec) -> String {
format!("{DFPLAN_BODY_PREFIX}{}:{plan_bytes_b64}", spec.render())
}
pub fn dfplan_body_with_spec(body: &str, spec: &DfplanTaskSpec) -> SqlResult<String> {
let (_, b64) = split_dfplan_body(body)?;
let trimmed = body.trim_start();
let rest = strip_leading_python_udf_directives(trimmed);
let directives = trimmed.get(..trimmed.len() - rest.len()).unwrap_or("");
Ok(format!(
"{directives}{DFPLAN_BODY_PREFIX}{}:{b64}",
spec.render()
))
}
pub(crate) fn strip_leading_python_udf_directives(body: &str) -> &str {
const CLOSE: &str = " */";
let mut rest = body.trim_start();
while rest.starts_with("/* krishiv-register-python-udf:")
|| rest.starts_with("/* krishiv-register-python-udaf:")
{
let Some(end) = rest.find(CLOSE) else { break };
rest = rest[end + CLOSE.len()..].trim_start();
}
rest
}
fn split_dfplan_body(body: &str) -> SqlResult<(&str, &str)> {
let rest = strip_leading_python_udf_directives(body)
.strip_prefix(DFPLAN_BODY_PREFIX)
.ok_or_else(|| SqlError::DataFusion {
message: format!(
"task body is not a {DFPLAN_BODY_PREFIX} fragment: {}",
body.chars().take(48).collect::<String>()
),
})?;
rest.split_once(':').ok_or_else(|| SqlError::DataFusion {
message: String::from("dfplan body missing partition segment"),
})
}
fn parse_partition_segment(segment: &str) -> SqlResult<DfplanTaskSpec> {
let (list, range) = match segment.split_once('/') {
Some((list, range_str)) => {
let rest = range_str
.strip_prefix('s')
.ok_or_else(|| SqlError::DataFusion {
message: format!("dfplan map range missing 's' prefix: {range_str}"),
})?;
let (stage, span) = rest.split_once('m').ok_or_else(|| SqlError::DataFusion {
message: format!("dfplan map range missing 'm' separator: {range_str}"),
})?;
let (start, end) = span.split_once('-').ok_or_else(|| SqlError::DataFusion {
message: format!("dfplan map range missing '-' separator: {range_str}"),
})?;
let parse = |s: &str, what: &str| {
s.trim().parse::<usize>().map_err(|e| SqlError::DataFusion {
message: format!("dfplan map range {what}: {e}"),
})
};
let range = DfplanMapRange {
upstream_stage_index: parse(stage, "stage")?,
start: parse(start, "start")?,
end: parse(end, "end")?,
};
if range.start >= range.end {
return Err(SqlError::DataFusion {
message: format!("dfplan map range is empty: m{}-{}", range.start, range.end),
});
}
(list, Some(range))
}
None => (segment, None),
};
let partitions = list
.split(',')
.map(|p| {
p.trim().parse::<usize>().map_err(|e| SqlError::DataFusion {
message: format!("dfplan partition index: {e}"),
})
})
.collect::<SqlResult<Vec<_>>>()?;
if partitions.is_empty() {
return Err(SqlError::DataFusion {
message: String::from("dfplan body has no partitions"),
});
}
Ok(DfplanTaskSpec {
partitions,
map_range: range,
})
}
pub fn dfplan_body_partition_spec(body: &str) -> SqlResult<DfplanTaskSpec> {
let (segment, _) = split_dfplan_body(body)?;
parse_partition_segment(segment)
}
pub fn parse_dfplan_body(body: &str) -> SqlResult<(DfplanTaskSpec, Vec<u8>)> {
let (segment, b64) = split_dfplan_body(body)?;
let spec = parse_partition_segment(segment)?;
let bytes = base64::engine::general_purpose::STANDARD
.decode(b64.as_bytes())
.map_err(|e| SqlError::DataFusion {
message: format!("dfplan base64 decode: {e}"),
})?;
Ok((spec, bytes))
}
fn verify_dfplan_roundtrip(
bytes: &[u8],
codec: &dyn PhysicalExtensionCodec,
ctx: &Arc<TaskContext>,
expected_plan: Option<&Arc<dyn ExecutionPlan>>,
) -> SqlResult<()> {
let decoded =
datafusion_proto::bytes::physical_plan_from_bytes_with_extension_codec(bytes, ctx, codec)
.map_err(|e| SqlError::DataFusion {
message: format!("physical plan proto decode: {e}"),
})?;
if let Some(expected) = expected_plan
&& let Some(difference) = first_schema_difference(expected, &decoded, "root")
{
return Err(SqlError::DataFusion {
message: format!(
"decoded plan differs from the encoded plan; the fragment would produce \
columns the reader does not expect. {difference}"
),
});
}
Ok(())
}
fn first_schema_difference(
original: &Arc<dyn ExecutionPlan>,
decoded: &Arc<dyn ExecutionPlan>,
path: &str,
) -> Option<String> {
let original_children = original.children();
let decoded_children = decoded.children();
if original_children.len() != decoded_children.len() {
return Some(format!(
"at {path}: {} has {} children, decoded {} has {}",
original.name(),
original_children.len(),
decoded.name(),
decoded_children.len()
));
}
for (index, (a, b)) in original_children
.iter()
.zip(decoded_children.iter())
.enumerate()
{
let child_path = format!("{path}/{}[{index}]", a.name());
if let Some(difference) = first_schema_difference(a, b, &child_path) {
return Some(difference);
}
}
if original.schema() != decoded.schema() {
return Some(format!(
"at {path} ({} vs {}):\n encoded: {:?}\n decoded: {:?}",
original.name(),
decoded.name(),
original.schema(),
decoded.schema()
));
}
None
}
#[must_use]
pub fn fragment_decode_session_context() -> SessionContext {
crate::SqlEngine::new_with_engine_memory(crate::EngineMemory::Unbounded)
.session_context()
.clone()
}
pub fn decode_dfplan_task(
body: &str,
ctx: &TaskContext,
codec: &dyn PhysicalExtensionCodec,
) -> SqlResult<(DfplanTaskSpec, Arc<dyn ExecutionPlan>)> {
let (spec, bytes) = parse_dfplan_body(body)?;
let plan =
datafusion_proto::bytes::physical_plan_from_bytes_with_extension_codec(&bytes, ctx, codec)
.map_err(|e| SqlError::DataFusion {
message: format!("physical plan proto decode: {e}"),
})?;
let plan = pin_file_scans_to_partitions(plan)?;
Ok((spec, plan))
}
fn pin_file_scans_to_partitions(plan: Arc<dyn ExecutionPlan>) -> SqlResult<Arc<dyn ExecutionPlan>> {
use datafusion::datasource::source::DataSourceExec;
if let Some(source_exec) = plan.downcast_ref::<DataSourceExec>() {
if let Some(pinned) = source_exec.data_source().with_preserve_order(true) {
return Ok(Arc::new(DataSourceExec::new(pinned)));
}
return Ok(plan);
}
let children = plan.children();
if children.is_empty() {
return Ok(plan);
}
let mut new_children = Vec::with_capacity(children.len());
let mut changed = false;
for child in children {
let pinned = pin_file_scans_to_partitions(Arc::clone(child))?;
changed = changed || !Arc::ptr_eq(&pinned, child);
new_children.push(pinned);
}
if !changed {
return Ok(plan);
}
plan.with_new_children(new_children)
.map_err(|e| SqlError::DataFusion {
message: format!("scan pinning rewrite: {e}"),
})
}
pub fn is_dfplan_body(body: &str) -> bool {
strip_leading_python_udf_directives(body).starts_with(DFPLAN_BODY_PREFIX)
}
fn apply_local_spill_strategy(plan: Arc<dyn ExecutionPlan>) -> Arc<dyn ExecutionPlan> {
use datafusion::physical_optimizer::PhysicalOptimizerRule;
if !crate::grace_hash_join::enabled() {
return plan;
}
let rule = crate::spillable_join::SpillableJoinSelection::for_local_execution();
match rule.optimize(
Arc::clone(&plan),
&datafusion::common::config::ConfigOptions::default(),
) {
Ok(rewritten) => rewritten,
Err(error) => {
tracing::warn!(%error, "local spill strategy declined; running the decoded plan as-is");
plan
}
}
}
pub fn execute_dfplan_body(
body: &str,
session: &SessionContext,
reader: Option<Arc<dyn ShufflePartitionReader>>,
) -> SqlResult<(SchemaRef, crate::SqlStream)> {
let spec_peek = dfplan_body_partition_spec(body)?;
let reader = match (&spec_peek.map_range, reader) {
(Some(range), Some(inner)) => Some(Arc::new(MapRangeShuffleReader {
inner,
range: range.clone(),
}) as Arc<dyn ShufflePartitionReader>),
(_, reader) => reader,
};
let codec = match reader {
Some(reader) => KrishivPhysicalCodec::executor(reader),
None => KrishivPhysicalCodec::coordinator(),
};
let task_ctx = session.task_ctx();
let (spec, plan) = decode_dfplan_task(body, &task_ctx, &codec)?;
let plan = apply_local_spill_strategy(plan);
let partition_count = plan.output_partitioning().partition_count();
if let Some(&bad) = spec.partitions.iter().find(|&&p| p >= partition_count) {
return Err(SqlError::DataFusion {
message: format!(
"dfplan partition {bad} out of range: decoded plan has \
{partition_count} partitions"
),
});
}
let schema = plan.schema();
let mut streams = Vec::with_capacity(spec.partitions.len());
for &partition in &spec.partitions {
let stream = plan
.execute(partition, Arc::clone(&task_ctx))
.map_err(|e| SqlError::DataFusion {
message: format!("dfplan execute (partition {partition}): {e}"),
})?;
streams.push(stream.map_err(|e| SqlError::DataFusion {
message: e.to_string(),
}));
}
let chained = futures::stream::iter(streams).flatten();
Ok((schema, Box::pin(chained)))
}
#[derive(Debug)]
struct MapRangeShuffleReader {
inner: Arc<dyn ShufflePartitionReader>,
range: DfplanMapRange,
}
impl ShufflePartitionReader for MapRangeShuffleReader {
fn open_partition(
&self,
upstream_stage_index: usize,
map_task_index: usize,
partition: usize,
) -> futures::future::BoxFuture<'static, Result<ShuffleFragmentStream, String>> {
if upstream_stage_index == self.range.upstream_stage_index
&& !(self.range.start..self.range.end).contains(&map_task_index)
{
return Box::pin(async {
Ok(Box::pin(futures::stream::empty()) as ShuffleFragmentStream)
});
}
self.inner
.open_partition(upstream_stage_index, map_task_index, partition)
}
}
pub fn dfplan_body_is_split_safe(body: &str) -> bool {
let ctx = SessionContext::new();
let codec = KrishivPhysicalCodec::coordinator();
let Ok((_, plan)) = decode_dfplan_task(body, &ctx.task_ctx(), &codec) else {
return false;
};
plan_is_split_safe(&plan)
}
fn plan_is_split_safe(plan: &Arc<dyn ExecutionPlan>) -> bool {
use datafusion::physical_plan::filter::FilterExec;
use datafusion::physical_plan::joins::HashJoinExec;
use datafusion::physical_plan::projection::ProjectionExec;
let safe = if let Some(join) = plan.downcast_ref::<HashJoinExec>() {
*join.join_type() == datafusion::logical_expr::JoinType::Inner
} else {
plan.is::<ShuffleReadExec>()
|| plan.is::<ProjectionExec>()
|| plan.is::<FilterExec>()
|| plan.name() == "CoalesceBatchesExec"
};
safe && plan.children().iter().all(|c| plan_is_split_safe(c))
}
fn register_object_store_for_path(ctx: &SessionContext, path: &str) -> SqlResult<()> {
if !(path.starts_with("s3://") || path.starts_with("s3a://")) {
return Ok(());
}
let url = url::Url::parse(path).map_err(|e| SqlError::DataFusion {
message: format!("staged planning: invalid object-store url {path}: {e}"),
})?;
let bucket = url.host_str().unwrap_or_default();
let store_url =
url::Url::parse(&format!("s3://{bucket}")).map_err(|e| SqlError::DataFusion {
message: format!("staged planning: invalid bucket url for {path}: {e}"),
})?;
let store = crate::build_s3_object_store(bucket).map_err(|e| SqlError::DataFusion {
message: format!("staged planning: object store init for {path}: {e}"),
})?;
ctx.register_object_store(&store_url, store);
Ok(())
}
#[derive(Debug, Clone)]
pub struct ParquetTableSpec {
pub name: String,
pub path: String,
pub primary_key: Vec<String>,
}
impl ParquetTableSpec {
pub fn new(name: impl Into<String>, path: impl Into<String>) -> Self {
Self {
name: name.into(),
path: path.into(),
primary_key: Vec::new(),
}
}
#[must_use]
pub fn with_primary_key<I, S>(mut self, columns: I) -> Self
where
I: IntoIterator<Item = S>,
S: Into<String>,
{
self.primary_key = columns.into_iter().map(Into::into).collect();
self
}
}
fn directory_aware_url(path: &str) -> String {
if !path.contains("://") || path.ends_with('/') {
return path.to_owned();
}
let looks_like_a_file = path
.rsplit('/')
.next()
.is_some_and(|segment| segment.contains('.'));
if looks_like_a_file {
path.to_owned()
} else {
format!("{path}/")
}
}
pub async fn register_parquet_table(
ctx: &SessionContext,
spec: &ParquetTableSpec,
) -> SqlResult<()> {
use datafusion::common::{Constraint, Constraints};
use datafusion::datasource::TableProvider as _;
use datafusion::datasource::file_format::options::ReadOptions as _;
use datafusion::datasource::listing::{ListingTable, ListingTableConfig, ListingTableUrl};
let read_options = datafusion::prelude::ParquetReadOptions::default();
if spec.primary_key.is_empty() {
return ctx
.register_parquet(&spec.name, &directory_aware_url(&spec.path), read_options)
.await
.map_err(|e| SqlError::DataFusion {
message: format!("staged planning: register '{}': {e}", spec.name),
});
}
let url = ListingTableUrl::parse(directory_aware_url(&spec.path)).map_err(|e| {
SqlError::DataFusion {
message: format!("staged planning: table url for '{}': {e}", spec.name),
}
})?;
let options = read_options.to_listing_options(&ctx.copied_config(), ctx.copied_table_options());
let config = ListingTableConfig::new(url)
.with_listing_options(options)
.infer_schema(&ctx.state())
.await
.map_err(|e| SqlError::DataFusion {
message: format!("staged planning: infer schema for '{}': {e}", spec.name),
})?;
let table = ListingTable::try_new(config).map_err(|e| SqlError::DataFusion {
message: format!("staged planning: listing table '{}': {e}", spec.name),
})?;
let schema = table.schema();
let mut indices = Vec::with_capacity(spec.primary_key.len());
for column in &spec.primary_key {
let index = schema.index_of(column).map_err(|_| SqlError::DataFusion {
message: format!(
"declared primary key column '{column}' is not in table '{}' \
(columns: {})",
spec.name,
schema
.fields()
.iter()
.map(|field| field.name().clone())
.collect::<Vec<String>>()
.join(", ")
),
})?;
indices.push(index);
}
let table = table.with_constraints(Constraints::new_unverified(vec![Constraint::PrimaryKey(
indices,
)]));
ctx.register_table(spec.name.as_str(), Arc::new(table))
.map_err(|e| SqlError::DataFusion {
message: format!("staged planning: register '{}': {e}", spec.name),
})?;
tracing::debug!(
table = %spec.name,
primary_key = ?spec.primary_key,
"registered parquet table with a declared primary key"
);
Ok(())
}
pub async fn build_stages_for_parquet_query(
query: &str,
tables: &[(String, String)],
cluster: Option<ClusterCapacity>,
) -> SqlResult<Option<DistributedStagePlan>> {
let specs: Vec<ParquetTableSpec> = tables
.iter()
.map(|(name, path)| ParquetTableSpec::new(name, path))
.collect();
build_stages_for_parquet_tables(query, &specs, cluster).await
}
pub async fn build_stages_for_parquet_tables(
query: &str,
tables: &[ParquetTableSpec],
cluster: Option<ClusterCapacity>,
) -> SqlResult<Option<DistributedStagePlan>> {
let target_partitions = resolve_stage_target_partitions(cluster);
tracing::debug!(
target_partitions,
total_slots = cluster.map(|c| c.total_slots),
"planning distributed stages"
);
let ctx = planning_session_context(target_partitions);
for spec in tables {
register_object_store_for_path(&ctx, &spec.path)?;
register_parquet_table(&ctx, spec).await?;
}
let udf_directive_source = query;
let query = register_python_udf_signatures_and_strip(&ctx, query)?;
let df = ctx.sql(&query).await.map_err(|e| SqlError::DataFusion {
message: format!("staged planning: {e}"),
})?;
let df = inline_uncorrelated_scalar_subqueries(&ctx, df).await?;
let plan = df
.create_physical_plan()
.await
.map_err(|e| SqlError::DataFusion {
message: format!("staged physical planning: {e}"),
})?;
build_distributed_stages_with_udf_directives(plan, udf_directive_source)
}
const MAX_FOLDABLE_SUBQUERY_INPUT_BYTES: usize = 256 * 1024 * 1024;
async fn subquery_is_cheap_to_fold(df: &datafusion::dataframe::DataFrame) -> bool {
use datafusion::common::stats::Precision;
let Ok(plan) = df.clone().create_physical_plan().await else {
return false;
};
let Ok(stats) = plan.partition_statistics(None) else {
return false;
};
match stats.total_byte_size {
Precision::Exact(bytes) | Precision::Inexact(bytes) => {
bytes <= MAX_FOLDABLE_SUBQUERY_INPUT_BYTES
}
Precision::Absent => false,
}
}
async fn inline_uncorrelated_scalar_subqueries(
ctx: &SessionContext,
df: datafusion::dataframe::DataFrame,
) -> SqlResult<datafusion::dataframe::DataFrame> {
use datafusion::common::tree_node::{Transformed, TreeNode, TreeNodeRecursion};
use datafusion::logical_expr::{Expr, LogicalPlan};
let plan = df.logical_plan().clone();
let mut pending: Vec<(String, LogicalPlan)> = Vec::new();
let collect = plan.apply(|node| {
for expr in node.expressions() {
expr.apply(|e| {
if let Expr::ScalarSubquery(sub) = e
&& sub.outer_ref_columns.is_empty()
{
let key = sub.subquery.display_indent().to_string();
if !pending.iter().any(|(k, _)| *k == key) {
pending.push((key, sub.subquery.as_ref().clone()));
}
}
Ok(TreeNodeRecursion::Continue)
})?;
}
Ok(TreeNodeRecursion::Continue)
});
if collect.is_err() || pending.is_empty() {
return Ok(df);
}
let mut folded: Vec<(String, datafusion::scalar::ScalarValue)> = Vec::new();
for (key, sub_plan) in pending {
let sub_df = datafusion::dataframe::DataFrame::new(ctx.state(), sub_plan);
if !subquery_is_cheap_to_fold(&sub_df).await {
tracing::info!(
subquery = %key.lines().next().unwrap_or_default(),
"scalar subquery too large to fold on the coordinator; leaving it in the plan"
);
continue;
}
let Ok(batches) = sub_df.collect().await else {
continue;
};
let rows: usize = batches
.iter()
.map(arrow::array::RecordBatch::num_rows)
.sum();
if rows > 1 {
continue;
}
let Some(batch) = batches.iter().find(|b| b.num_rows() == 1) else {
let Some(first) = batches.first() else {
continue;
};
let Some(field) = first.schema().fields().first().cloned() else {
continue;
};
if let Ok(null) = datafusion::scalar::ScalarValue::try_from(field.data_type()) {
folded.push((key, null));
}
continue;
};
let Some(column) = batch.columns().first() else {
continue;
};
if let Ok(value) = datafusion::scalar::ScalarValue::try_from_array(column, 0) {
folded.push((key, value));
}
}
if folded.is_empty() {
return Ok(df);
}
let rewritten = plan.transform_up(|node| {
let exprs = node.expressions();
if !exprs.iter().any(Expr::contains_scalar_subquery) {
return Ok(Transformed::no(node));
}
let mut changed = false;
let mut new_exprs = Vec::with_capacity(exprs.len());
for expr in exprs {
let out = expr.transform_up(|e| {
if let Expr::ScalarSubquery(sub) = &e
&& sub.outer_ref_columns.is_empty()
{
let key = sub.subquery.display_indent().to_string();
if let Some((_, value)) = folded.iter().find(|(k, _)| *k == key) {
return Ok(Transformed::yes(datafusion::prelude::lit(value.clone())));
}
}
Ok(Transformed::no(e))
})?;
changed |= out.transformed;
new_exprs.push(out.data);
}
if !changed {
return Ok(Transformed::no(node));
}
let inputs = node.inputs().into_iter().cloned().collect::<Vec<_>>();
let rebuilt = node.with_new_exprs(new_exprs, inputs)?;
Ok(Transformed::yes(rebuilt))
});
match rewritten {
Ok(t) if t.transformed => {
tracing::info!(
folded = folded.len(),
"q22: folded uncorrelated scalar subqueries to constants so the \
query can be staged instead of running as a single task"
);
Ok(datafusion::dataframe::DataFrame::new(ctx.state(), t.data))
}
_ => Ok(df),
}
}
pub fn register_python_udf_signatures_and_strip(
ctx: &SessionContext,
query: &str,
) -> SqlResult<String> {
use datafusion::logical_expr::{ColumnarValue, Volatility, create_udf};
const PREFIX: &str = "/* krishiv-register-python-udf:";
if !query.contains(PREFIX) {
return Ok(query.to_string());
}
let mut out = String::with_capacity(query.len());
let mut rest = query;
while let Some(start) = rest.find(PREFIX) {
out.push_str(&rest[..start]);
let after = &rest[start + PREFIX.len()..];
let Some(end) = after.find(" */") else {
out.push_str(&rest[start..]);
return Ok(out);
};
let body = &after[..end];
rest = &after[end + " */".len()..];
let mut parts = body.splitn(4, ':');
let (name, in_types, out_type) = match (parts.next(), parts.next(), parts.next()) {
(Some(n), Some(i), Some(o)) => (n, i, o),
_ => continue,
};
let input_types: Vec<arrow::datatypes::DataType> = if in_types.is_empty() {
Vec::new()
} else {
in_types
.split(',')
.map(crate::python_udf_arrow_type)
.collect()
};
let return_type = crate::python_udf_arrow_type(out_type);
let name_owned = name.to_string();
let udf = create_udf(
name,
input_types,
return_type,
Volatility::Volatile,
Arc::new(move |_: &[ColumnarValue]| {
Err(DataFusionError::NotImplemented(format!(
"python UDF '{name_owned}' executes on the executor, not during \
coordinator planning"
)))
}),
);
ctx.register_udf(udf);
}
out.push_str(rest);
Ok(out)
}
pub trait ShufflePartitionReader: fmt::Debug + Send + Sync {
fn open_partition(
&self,
upstream_stage_index: usize,
map_task_index: usize,
partition: usize,
) -> futures::future::BoxFuture<'static, Result<ShuffleFragmentStream, String>>;
}
pub type ShuffleFragmentStream =
futures::stream::BoxStream<'static, Result<arrow::record_batch::RecordBatch, String>>;
pub const SHUFFLE_FETCH_BUFFER_ENV: &str = "KRISHIV_SHUFFLE_FETCH_BUFFER";
const DEFAULT_SHUFFLE_FETCH_BUFFER: usize = 1;
fn shuffle_fetch_buffer() -> usize {
std::env::var(SHUFFLE_FETCH_BUFFER_ENV)
.ok()
.and_then(|v| v.trim().parse::<usize>().ok())
.unwrap_or(DEFAULT_SHUFFLE_FETCH_BUFFER)
.max(1)
}
#[derive(Debug)]
pub struct ShuffleReadExec {
upstream_stage_index: usize,
num_map_tasks: usize,
schema: SchemaRef,
properties: Arc<PlanProperties>,
reader: Option<Arc<dyn ShufflePartitionReader>>,
upstream_rows: Option<usize>,
upstream_bytes: Option<usize>,
}
impl ShuffleReadExec {
pub fn new(
upstream_stage_index: usize,
num_map_tasks: usize,
partition_count: usize,
schema: SchemaRef,
reader: Option<Arc<dyn ShufflePartitionReader>>,
) -> Self {
let properties = Arc::new(PlanProperties::new(
EquivalenceProperties::new(Arc::clone(&schema)),
Partitioning::UnknownPartitioning(partition_count.max(1)),
EmissionType::Incremental,
Boundedness::Bounded,
));
Self {
upstream_stage_index,
num_map_tasks,
schema,
properties,
reader,
upstream_rows: None,
upstream_bytes: None,
}
}
#[must_use]
pub fn with_upstream_estimate(mut self, rows: Option<usize>, bytes: Option<usize>) -> Self {
self.upstream_rows = rows;
self.upstream_bytes = bytes;
self
}
pub fn upstream_estimate(&self) -> (Option<usize>, Option<usize>) {
(self.upstream_rows, self.upstream_bytes)
}
fn precision_value(p: &datafusion::common::stats::Precision<usize>) -> Option<usize> {
match p {
datafusion::common::stats::Precision::Exact(v)
| datafusion::common::stats::Precision::Inexact(v) => Some(*v),
datafusion::common::stats::Precision::Absent => None,
}
}
pub fn estimate_of(plan: &Arc<dyn ExecutionPlan>) -> (Option<usize>, Option<usize>) {
plan.partition_statistics(None).map_or((None, None), |s| {
(
Self::precision_value(&s.num_rows),
Self::precision_value(&s.total_byte_size),
)
})
}
pub fn upstream_stage_index(&self) -> usize {
self.upstream_stage_index
}
pub(crate) fn clone_with_upstream_stage_index(&self, upstream_stage_index: usize) -> Self {
Self {
upstream_stage_index,
num_map_tasks: self.num_map_tasks,
schema: Arc::clone(&self.schema),
properties: Arc::clone(&self.properties),
reader: self.reader.clone(),
upstream_rows: self.upstream_rows,
upstream_bytes: self.upstream_bytes,
}
}
pub fn num_map_tasks(&self) -> usize {
self.num_map_tasks
}
pub fn partition_count(&self) -> usize {
self.properties.partitioning.partition_count()
}
}
impl DisplayAs for ShuffleReadExec {
fn fmt_as(&self, _t: DisplayFormatType, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"ShuffleReadExec: upstream_stage={}, map_tasks={}, partitions={}",
self.upstream_stage_index,
self.num_map_tasks,
self.partition_count()
)
}
}
impl ExecutionPlan for ShuffleReadExec {
fn name(&self) -> &str {
"ShuffleReadExec"
}
fn properties(&self) -> &Arc<PlanProperties> {
&self.properties
}
fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
Vec::new()
}
fn with_new_children(
self: Arc<Self>,
_children: Vec<Arc<dyn ExecutionPlan>>,
) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
Ok(self)
}
fn partition_statistics(
&self,
partition: Option<usize>,
) -> datafusion::error::Result<Arc<datafusion::common::Statistics>> {
use datafusion::common::stats::Precision;
let partitions = self.properties.partitioning.partition_count().max(1);
let divisor = if partition.is_some() { partitions } else { 1 };
let scale = |v: Option<usize>| -> Precision<usize> {
v.map_or(Precision::Absent, |v| Precision::Inexact(v / divisor))
};
let mut stats = datafusion::common::Statistics::new_unknown(&self.schema);
stats.num_rows = scale(self.upstream_rows);
stats.total_byte_size = scale(self.upstream_bytes);
Ok(Arc::new(stats))
}
fn execute(
&self,
partition: usize,
_context: Arc<TaskContext>,
) -> datafusion::error::Result<SendableRecordBatchStream> {
let reader = self.reader.clone().ok_or_else(|| {
DataFusionError::Execution(String::from(
"ShuffleReadExec has no shuffle reader: this plan was decoded without an \
executor-side codec (coordinator-side plans are not executable)",
))
})?;
let stage = self.upstream_stage_index;
let schema = Arc::clone(&self.schema);
let expected = Arc::clone(&self.schema);
let stream = futures::stream::iter(0..self.num_map_tasks)
.map(move |map_task| {
let reader = Arc::clone(&reader);
async move {
reader
.open_partition(stage, map_task, partition)
.await
.map(|batches| (map_task, batches))
.map_err(|e| {
DataFusionError::Execution(format!(
"shuffle read (stage {stage}, map {map_task}, partition \
{partition}): {e}"
))
})
}
})
.buffered(shuffle_fetch_buffer())
.map_ok(move |(map_task, batches)| {
let expected = Arc::clone(&expected);
batches.map(move |batch| {
let batch = batch.map_err(|e| {
DataFusionError::Execution(format!(
"shuffle read (stage {stage}, map {map_task}, partition \
{partition}): {e}"
))
})?;
check_shuffle_batch_schema(&expected, batch, stage, map_task, partition)
})
})
.try_flatten();
Ok(Box::pin(RecordBatchStreamAdapter::new(schema, stream)))
}
}
fn check_shuffle_batch_schema(
expected: &SchemaRef,
batch: arrow::record_batch::RecordBatch,
stage: usize,
map_task: usize,
partition: usize,
) -> Result<arrow::record_batch::RecordBatch, DataFusionError> {
let actual = batch.schema();
if Arc::ptr_eq(expected, &actual) {
return Ok(batch);
}
let where_ = || format!("stage {stage}, map {map_task}, partition {partition}");
if expected.fields().len() != actual.fields().len() {
return Err(DataFusionError::Execution(format!(
"shuffle read ({}) produced {} columns but the plan declares {}; \
the map stage did not produce the schema the reduce side was planned \
against.\n declared: {:?}\n produced: {:?}",
where_(),
actual.fields().len(),
expected.fields().len(),
expected.fields(),
actual.fields(),
)));
}
for (index, (want, got)) in expected
.fields()
.iter()
.zip(actual.fields().iter())
.enumerate()
{
if want.data_type() != got.data_type() {
return Err(DataFusionError::Execution(format!(
"shuffle read ({}) column {index} ({}) is {:?} but the plan declares {:?} \
({}); the map stage did not produce the schema the reduce side was \
planned against",
where_(),
got.name(),
got.data_type(),
want.data_type(),
want.name(),
)));
}
}
Ok(batch)
}
#[derive(serde::Serialize, serde::Deserialize)]
struct ShuffleReadNodePayload {
v: u32,
stage: usize,
map_tasks: usize,
partitions: usize,
schema_ipc_b64: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
upstream_rows: Option<usize>,
#[serde(default, skip_serializing_if = "Option::is_none")]
upstream_bytes: Option<usize>,
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(tag = "node")]
enum KrishivNodePayload {
RuntimeFilterBuild {
key_index: usize,
filter_bytes: usize,
},
RuntimeFilterProbe {
key_index: usize,
},
}
fn schema_to_ipc_bytes(schema: &arrow::datatypes::Schema) -> Result<Vec<u8>, String> {
let mut buf = Vec::new();
let mut writer = arrow::ipc::writer::StreamWriter::try_new(&mut buf, schema)
.map_err(|e| format!("schema ipc writer: {e}"))?;
writer
.finish()
.map_err(|e| format!("schema ipc finish: {e}"))?;
Ok(buf)
}
fn schema_from_ipc_bytes(bytes: &[u8]) -> Result<SchemaRef, String> {
let reader = arrow::ipc::reader::StreamReader::try_new(std::io::Cursor::new(bytes), None)
.map_err(|e| format!("schema ipc reader: {e}"))?;
Ok(reader.schema())
}
#[derive(Debug, Default)]
pub struct KrishivPhysicalCodec {
reader: Option<Arc<dyn ShufflePartitionReader>>,
}
impl KrishivPhysicalCodec {
pub fn coordinator() -> Self {
Self { reader: None }
}
pub fn executor(reader: Arc<dyn ShufflePartitionReader>) -> Self {
Self {
reader: Some(reader),
}
}
}
impl PhysicalExtensionCodec for KrishivPhysicalCodec {
fn try_decode(
&self,
buf: &[u8],
inputs: &[Arc<dyn ExecutionPlan>],
_ctx: &TaskContext,
) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
use crate::runtime_filter_exec::{RuntimeFilterBuildExec, RuntimeFilterProbeExec};
if let Ok(payload) = serde_json::from_slice::<KrishivNodePayload>(buf) {
return match payload {
KrishivNodePayload::RuntimeFilterBuild {
key_index,
filter_bytes,
} => {
let [input] = inputs else {
return Err(DataFusionError::Internal(format!(
"RuntimeFilterBuildExec expects one input, got {}",
inputs.len()
)));
};
Ok(Arc::new(RuntimeFilterBuildExec::try_new(
Arc::clone(input),
key_index,
filter_bytes,
)?))
}
KrishivNodePayload::RuntimeFilterProbe { key_index } => {
let [data, filter] = inputs else {
return Err(DataFusionError::Internal(format!(
"RuntimeFilterProbeExec expects two inputs, got {}",
inputs.len()
)));
};
Ok(Arc::new(RuntimeFilterProbeExec::try_new(
Arc::clone(data),
Arc::clone(filter),
key_index,
)?))
}
};
}
let payload: ShuffleReadNodePayload = serde_json::from_slice(buf)
.map_err(|e| DataFusionError::Internal(format!("shuffle-read node decode: {e}")))?;
if payload.v != 1 {
return Err(DataFusionError::Internal(format!(
"unsupported shuffle-read node version {}",
payload.v
)));
}
let schema_bytes = base64::engine::general_purpose::STANDARD
.decode(payload.schema_ipc_b64.as_bytes())
.map_err(|e| DataFusionError::Internal(format!("shuffle-read schema b64: {e}")))?;
let schema = schema_from_ipc_bytes(&schema_bytes).map_err(DataFusionError::Internal)?;
Ok(Arc::new(
ShuffleReadExec::new(
payload.stage,
payload.map_tasks,
payload.partitions,
schema,
self.reader.clone(),
)
.with_upstream_estimate(payload.upstream_rows, payload.upstream_bytes),
))
}
fn try_encode(
&self,
node: Arc<dyn ExecutionPlan>,
buf: &mut Vec<u8>,
) -> datafusion::error::Result<()> {
use crate::runtime_filter_exec::{RuntimeFilterBuildExec, RuntimeFilterProbeExec};
let filter_payload = if let Some(build) = node.downcast_ref::<RuntimeFilterBuildExec>() {
Some(KrishivNodePayload::RuntimeFilterBuild {
key_index: build.key_index(),
filter_bytes: build.filter_bytes(),
})
} else {
node.downcast_ref::<RuntimeFilterProbeExec>().map(|probe| {
KrishivNodePayload::RuntimeFilterProbe {
key_index: probe.key_index(),
}
})
};
if let Some(payload) = filter_payload {
let json = serde_json::to_vec(&payload).map_err(|e| {
DataFusionError::Internal(format!("runtime filter node encode: {e}"))
})?;
buf.extend_from_slice(&json);
return Ok(());
}
let read = node.downcast_ref::<ShuffleReadExec>().ok_or_else(|| {
DataFusionError::NotImplemented(format!(
"KrishivPhysicalCodec cannot encode node {}",
node.name()
))
})?;
let schema_bytes = schema_to_ipc_bytes(&read.schema).map_err(DataFusionError::Internal)?;
let (upstream_rows, upstream_bytes) = read.upstream_estimate();
let payload = ShuffleReadNodePayload {
v: 1,
stage: read.upstream_stage_index,
map_tasks: read.num_map_tasks,
partitions: read.partition_count(),
schema_ipc_b64: base64::engine::general_purpose::STANDARD.encode(&schema_bytes),
upstream_rows,
upstream_bytes,
};
let json = serde_json::to_vec(&payload)
.map_err(|e| DataFusionError::Internal(format!("shuffle-read node encode: {e}")))?;
buf.extend_from_slice(&json);
Ok(())
}
}
#[derive(Debug, Clone)]
pub struct StageShuffleOutput {
pub key_columns: Vec<String>,
pub num_output_partitions: usize,
}
#[derive(Debug, Clone)]
pub struct DistributedStage {
pub task_bodies: Vec<String>,
pub shuffle: Option<StageShuffleOutput>,
pub upstream_stage_indexes: Vec<usize>,
}
impl DistributedStage {
pub fn task_count(&self) -> usize {
self.task_bodies.len()
}
}
#[derive(Debug, Clone)]
pub struct DistributedStagePlan {
pub stages: Vec<DistributedStage>,
}
struct StageDraft {
plan: Arc<dyn ExecutionPlan>,
shuffle: Option<StageShuffleOutput>,
subqueries: Option<StageSubqueryContext>,
}
struct StageSubqueryContext {
links: Vec<ScalarSubqueryLink>,
results: ScalarSubqueryResults,
}
struct Unsupported(String);
pub fn build_distributed_stages(
plan: Arc<dyn ExecutionPlan>,
) -> SqlResult<Option<DistributedStagePlan>> {
build_distributed_stages_with_udf_directives(plan, "")
}
pub fn build_distributed_stages_with_udf_directives(
plan: Arc<dyn ExecutionPlan>,
udf_directive_source: &str,
) -> SqlResult<Option<DistributedStagePlan>> {
let plan = reduce_by_broadcast_dimension(plan)?;
let plan = redistribute_unsplittable_broadcast_joins(plan)?;
let plan = {
use datafusion::physical_optimizer::PhysicalOptimizerRule as _;
crate::spillable_join::SpillableJoinSelection::from_capacity()
.without_broadcast_rescue()
.optimize(plan, &datafusion::common::config::ConfigOptions::default())
}
.map_err(|e| SqlError::DataFusion {
message: format!("spillable-join pass over the redistributed plan: {e}"),
})?;
let mut drafts: Vec<StageDraft> = Vec::new();
let mut root = match cut_exchanges(plan, &mut drafts) {
Ok(root) => root,
Err(Unsupported(reason)) => {
return Err(SqlError::DataFusion {
message: format!("stage split unsupported: {reason}"),
});
}
};
if drafts.is_empty() {
return Err(SqlError::DataFusion {
message: String::from("plan has no exchange to cut, so it cannot be split into stages"),
});
}
dedupe_identical_stages(&mut root, &mut drafts);
inject_runtime_filters(&root, &mut drafts);
drafts.push(StageDraft {
plan: root,
shuffle: None,
subqueries: None,
});
for draft in &drafts {
if let Some(reason) = find_unsupported_stage_node(&draft.plan) {
return Err(SqlError::DataFusion {
message: format!("stage subtree not partition-independent: {reason}"),
});
}
}
let codec = KrishivPhysicalCodec::coordinator();
let decode_session = fragment_decode_session_context();
if !udf_directive_source.is_empty() {
register_python_udf_signatures_and_strip(&decode_session, udf_directive_source)?;
}
let decode_ctx = decode_session.task_ctx();
let mut stages = Vec::with_capacity(drafts.len());
for draft in drafts {
let partition_count = draft.plan.output_partitioning().partition_count();
if partition_count == 0 {
return Err(SqlError::DataFusion {
message: String::from("stage subtree has zero output partitions"),
});
}
let upstream_stage_indexes = collect_upstream_stage_indexes(&draft.plan);
let attempts = match &draft.subqueries {
Some(context) => vec![
Arc::clone(&draft.plan),
wrap_in_scalar_subquery_exec(Arc::clone(&draft.plan), context),
],
None => vec![Arc::clone(&draft.plan)],
};
let mut shippable = None;
let mut last_error = None;
for stage_plan in attempts {
let bytes = match encode_dfplan_bytes(Arc::clone(&stage_plan), &codec) {
Ok(bytes) => bytes,
Err(error) => {
last_error = Some(error.to_string());
continue;
}
};
match verify_dfplan_roundtrip(&bytes, &codec, &decode_ctx, Some(&stage_plan)) {
Ok(()) => {
shippable = Some(bytes);
break;
}
Err(error) => last_error = Some(error.to_string()),
}
}
let Some(bytes) = shippable else {
tracing::warn!(
error = %last_error.unwrap_or_else(|| String::from("unknown")),
"stage plan cannot be encoded and decoded; running this query as a SINGLE TASK"
);
return Ok(None);
};
let b64 = base64::engine::general_purpose::STANDARD.encode(&bytes);
let task_bodies = (0..partition_count)
.map(|p| dfplan_task_body(&b64, p))
.collect();
stages.push(DistributedStage {
task_bodies,
shuffle: draft.shuffle,
upstream_stage_indexes,
});
}
Ok(Some(DistributedStagePlan { stages }))
}
fn exchange_reuse_key(
plan: &Arc<dyn ExecutionPlan>,
key_columns: &[String],
num_partitions: usize,
) -> String {
use datafusion::physical_plan::displayable;
format!(
"keys={key_columns:?}|parts={num_partitions}|schema={:?}|plan=\n{}",
plan.schema(),
displayable(plan.as_ref()).indent(true)
)
}
const MAX_GATHERED_SORT_FETCH: usize = 10_000;
fn contains_partitioned_join(plan: &Arc<dyn ExecutionPlan>) -> bool {
use datafusion::physical_plan::joins::{HashJoinExec, PartitionMode};
if let Some(join) = plan.downcast_ref::<HashJoinExec>()
&& *join.partition_mode() == PartitionMode::Partitioned
{
return true;
}
plan.children()
.iter()
.any(|child| contains_partitioned_join(child))
}
fn cut_exchanges(
plan: Arc<dyn ExecutionPlan>,
stages: &mut Vec<StageDraft>,
) -> Result<Arc<dyn ExecutionPlan>, Unsupported> {
if let Some(repartition) = plan.downcast_ref::<RepartitionExec>() {
let Partitioning::Hash(exprs, num_partitions) = repartition.partitioning() else {
return Err(Unsupported(format!(
"non-hash exchange in plan: {}",
repartition.partitioning()
)));
};
let key_columns = hash_expr_column_names(exprs).ok_or_else(|| {
Unsupported(String::from(
"hash exchange uses non-column expressions; cannot derive shuffle keys",
))
})?;
let input = cut_exchanges(Arc::clone(repartition.input()), stages)?;
let map_task_count = input.output_partitioning().partition_count();
if map_task_count == 0 {
return Err(Unsupported(String::from("hash exchange over empty input")));
}
let schema = input.schema();
let estimate = ShuffleReadExec::estimate_of(&input);
let reuse_key = exchange_reuse_key(&input, &key_columns, *num_partitions);
let stage_index = match stages.iter().position(|draft| {
draft.subqueries.is_none()
&& draft.shuffle.as_ref().is_some_and(|sh| {
sh.key_columns == key_columns && sh.num_output_partitions == *num_partitions
})
&& exchange_reuse_key(&draft.plan, &key_columns, *num_partitions) == reuse_key
}) {
Some(existing) => existing,
None => {
let index = stages.len();
stages.push(StageDraft {
plan: input,
shuffle: Some(StageShuffleOutput {
key_columns,
num_output_partitions: *num_partitions,
}),
subqueries: None,
});
index
}
};
return Ok(Arc::new(
ShuffleReadExec::new(stage_index, map_task_count, *num_partitions, schema, None)
.with_upstream_estimate(estimate.0, estimate.1),
));
}
if let Some(coalesce) = plan.downcast_ref::<CoalescePartitionsExec>() {
let input = cut_exchanges(Arc::clone(coalesce.input()), stages)?;
let map_task_count = input.output_partitioning().partition_count();
if map_task_count <= 1 {
return plan
.with_new_children(vec![input])
.map_err(|e| Unsupported(format!("gather rewrite: {e}")));
}
let schema = input.schema();
let estimate = ShuffleReadExec::estimate_of(&input);
let stage_index = stages.len();
stages.push(StageDraft {
plan: input,
shuffle: Some(StageShuffleOutput {
key_columns: Vec::new(),
num_output_partitions: 1,
}),
subqueries: None,
});
return Ok(Arc::new(
ShuffleReadExec::new(stage_index, map_task_count, 1, schema, None)
.with_upstream_estimate(estimate.0, estimate.1),
));
}
if let Some(merge) = plan.downcast_ref::<SortPreservingMergeExec>() {
let input = cut_exchanges(Arc::clone(merge.input()), stages)?;
let fetch = merge.fetch();
let worth_cutting = match fetch {
Some(n) => n <= MAX_GATHERED_SORT_FETCH,
None => contains_partitioned_join(&input),
};
if !worth_cutting {
return plan
.with_new_children(vec![input])
.map_err(|e| Unsupported(format!("sort-merge passthrough: {e}")));
}
let map_task_count = input.output_partitioning().partition_count();
if map_task_count <= 1 {
return plan
.with_new_children(vec![input])
.map_err(|e| Unsupported(format!("sort-merge gather rewrite: {e}")));
}
let schema = input.schema();
let estimate = ShuffleReadExec::estimate_of(&input);
let stage_index = stages.len();
stages.push(StageDraft {
plan: input,
shuffle: Some(StageShuffleOutput {
key_columns: Vec::new(),
num_output_partitions: 1,
}),
subqueries: None,
});
let read = Arc::new(
ShuffleReadExec::new(stage_index, map_task_count, 1, schema, None)
.with_upstream_estimate(estimate.0, estimate.1),
);
return Ok(Arc::new(
SortExec::new(merge.expr().clone(), read).with_fetch(fetch),
));
}
if let Some(subquery_exec) = plan.downcast_ref::<ScalarSubqueryExec>() {
let first_new_stage = stages.len();
let input = cut_exchanges(Arc::clone(subquery_exec.input()), stages)?;
let context = || StageSubqueryContext {
links: subquery_exec.subqueries().to_vec(),
results: subquery_exec.results().clone(),
};
if let Some(new_stages) = stages.get_mut(first_new_stage..) {
for draft in new_stages {
draft.subqueries.get_or_insert_with(context);
}
}
let mut children = Vec::with_capacity(subquery_exec.subqueries().len() + 1);
children.push(input);
children.extend(
subquery_exec
.subqueries()
.iter()
.map(|link| Arc::clone(&link.plan)),
);
return plan
.with_new_children(children)
.map_err(|e| Unsupported(format!("scalar-subquery rewrite: {e}")));
}
let children = plan.children();
if children.is_empty() {
return Ok(plan);
}
let mut new_children = Vec::with_capacity(children.len());
let mut changed = false;
for child in children {
let rewritten = cut_exchanges(Arc::clone(child), stages)?;
changed = changed || !Arc::ptr_eq(&rewritten, child);
new_children.push(rewritten);
}
if !changed {
return Ok(plan);
}
plan.with_new_children(new_children)
.map_err(|e| Unsupported(format!("plan rewrite: {e}")))
}
fn emits_unmatched_build_rows(join_type: datafusion::logical_expr::JoinType) -> bool {
use datafusion::logical_expr::JoinType;
matches!(
join_type,
JoinType::Left
| JoinType::LeftAnti
| JoinType::LeftSemi
| JoinType::LeftMark
| JoinType::Full
)
}
fn is_unsplittable_broadcast_join(join: &datafusion::physical_plan::joins::HashJoinExec) -> bool {
use datafusion::physical_plan::joins::PartitionMode;
*join.partition_mode() == PartitionMode::CollectLeft
&& emits_unmatched_build_rows(*join.join_type())
&& join.right().output_partitioning().partition_count() > 1
}
fn broadcast_build_estimate_is_empty(
join: &datafusion::physical_plan::joins::HashJoinExec,
) -> bool {
crate::join_estimates::BuildSideEstimate::of(join.left()).is_wholly_degenerate()
}
fn is_degenerate_broadcast_join(join: &datafusion::physical_plan::joins::HashJoinExec) -> bool {
use datafusion::physical_plan::joins::PartitionMode;
if *join.partition_mode() != PartitionMode::CollectLeft {
return false;
}
if join.null_aware {
return false;
}
if !broadcast_build_estimate_is_empty(join) {
return false;
}
let build_input_partitions = match join.left().downcast_ref::<CoalescePartitionsExec>() {
Some(coalesce) => coalesce.input().output_partitioning().partition_count(),
None => join.left().output_partitioning().partition_count(),
};
join.right().output_partitioning().partition_count() > 1 || build_input_partitions > 1
}
fn broadcast_byte_ceiling() -> usize {
std::env::var(BROADCAST_JOIN_BYTES_ENV)
.ok()
.and_then(|v| v.trim().parse::<usize>().ok())
.filter(|n| *n > 0)
.unwrap_or(DEFAULT_BROADCAST_JOIN_BYTES)
}
fn broadcast_build_is_too_wide(join: &datafusion::physical_plan::joins::HashJoinExec) -> bool {
use datafusion::physical_plan::joins::PartitionMode;
if *join.partition_mode() != PartitionMode::CollectLeft {
return false;
}
if join.null_aware {
return false;
}
let build = join.left();
let Some(implied) =
crate::join_estimates::BuildSideEstimate::of(build).bytes_implied_by_rows(&build.schema())
else {
return false;
};
implied > broadcast_byte_ceiling()
}
pub const DIMENSION_REDUCTION_ENV: &str = "KRISHIV_DIMENSION_REDUCTION";
fn dimension_reduction_enabled() -> bool {
matches!(
std::env::var(DIMENSION_REDUCTION_ENV)
.unwrap_or_default()
.trim()
.to_ascii_lowercase()
.as_str(),
"1" | "on" | "true" | "yes"
)
}
#[cfg(test)]
fn reduce_by_broadcast_dimension_for_test(
plan: Arc<dyn ExecutionPlan>,
) -> SqlResult<Arc<dyn ExecutionPlan>> {
reduce_by_broadcast_dimension_inner(plan)
}
fn reduce_by_broadcast_dimension(
plan: Arc<dyn ExecutionPlan>,
) -> SqlResult<Arc<dyn ExecutionPlan>> {
if !dimension_reduction_enabled() {
return Ok(plan);
}
reduce_by_broadcast_dimension_inner(plan)
}
fn reduce_by_broadcast_dimension_inner(
plan: Arc<dyn ExecutionPlan>,
) -> SqlResult<Arc<dyn ExecutionPlan>> {
use datafusion::logical_expr::JoinType;
use datafusion::physical_plan::joins::{HashJoinExec, PartitionMode};
let children = plan.children();
let plan = if children.is_empty() {
plan
} else {
let mut rebuilt = Vec::with_capacity(children.len());
let mut changed = false;
for child in children {
let new_child = reduce_by_broadcast_dimension_inner(Arc::clone(child))?;
changed = changed || !Arc::ptr_eq(&new_child, child);
rebuilt.push(new_child);
}
if changed {
plan.with_new_children(rebuilt)
.map_err(|e| SqlError::DataFusion {
message: format!("dimension-reduction rewrite: {e}"),
})?
} else {
plan
}
};
let Some(join) = plan.downcast_ref::<HashJoinExec>() else {
return Ok(plan);
};
if *join.join_type() != JoinType::Inner
|| *join.partition_mode() != PartitionMode::CollectLeft
|| join.null_aware
{
return Ok(plan);
}
if !is_broadcastable_dimension(join.left()) {
return Ok(plan);
}
let fact = join.right();
for (dim_key, fact_key) in join.on() {
let (Some(dim_col), Some(fact_col)) = (
(dim_key.as_ref() as &dyn std::any::Any)
.downcast_ref::<datafusion::physical_plan::expressions::Column>(),
(fact_key.as_ref() as &dyn std::any::Any)
.downcast_ref::<datafusion::physical_plan::expressions::Column>(),
) else {
continue;
};
if leaf_scans_naming(fact, fact_col.name()) != 1 {
continue;
}
let Some(reduced_fact) =
attach_reducer(fact, fact_col.name(), join.left(), dim_col.name())?
else {
continue;
};
let rebuilt = join
.builder()
.reset_state()
.with_new_children(vec![Arc::clone(join.left()), reduced_fact])
.and_then(|b| b.recompute_properties().build_exec())
.map_err(|e| SqlError::DataFusion {
message: format!("dimension-reduction rebuild: {e}"),
})?;
tracing::debug!(
key = fact_col.name(),
"reduced a fact stream by a broadcast dimension before its joins"
);
return Ok(rebuilt);
}
Ok(plan)
}
fn is_broadcastable_dimension(plan: &Arc<dyn ExecutionPlan>) -> bool {
use datafusion::common::stats::Precision;
let estimate = crate::join_estimates::BuildSideEstimate::of(plan);
if estimate.is_wholly_degenerate() {
return false;
}
let Ok(stats) = plan.partition_statistics(None) else {
return false;
};
let bytes = match stats.total_byte_size {
Precision::Exact(b) | Precision::Inexact(b) => Some(b),
Precision::Absent => estimate.bytes_implied_by_rows(&plan.schema()),
};
matches!(bytes, Some(b) if b > 0 && b <= broadcast_byte_ceiling())
}
fn leaf_scans_naming(plan: &Arc<dyn ExecutionPlan>, name: &str) -> usize {
if plan.children().is_empty() {
return usize::from(plan.schema().index_of(name).is_ok());
}
plan.children()
.iter()
.map(|child| leaf_scans_naming(child, name))
.sum()
}
fn attach_reducer(
plan: &Arc<dyn ExecutionPlan>,
name: &str,
dimension: &Arc<dyn ExecutionPlan>,
dimension_key: &str,
) -> SqlResult<Option<Arc<dyn ExecutionPlan>>> {
use datafusion::logical_expr::JoinType;
use datafusion::physical_expr::PhysicalExpr;
use datafusion::physical_plan::expressions::Column;
use datafusion::physical_plan::joins::{HashJoinExec, PartitionMode};
if let Some(existing) = plan.downcast_ref::<HashJoinExec>()
&& *existing.join_type() == JoinType::RightSemi
{
return Ok(None);
}
for (at, child) in plan.children().iter().enumerate() {
if child.schema().index_of(name).is_err() {
continue;
}
if let Some(new_child) = attach_reducer(child, name, dimension, dimension_key)? {
let mut children: Vec<Arc<dyn ExecutionPlan>> =
plan.children().into_iter().map(Arc::clone).collect();
let Some(slot) = children.get_mut(at) else {
return Ok(None);
};
*slot = new_child;
return Arc::clone(plan)
.with_new_children(children)
.map(Some)
.map_err(|e| SqlError::DataFusion {
message: format!("dimension-reduction splice: {e}"),
});
}
}
let Ok(fact_index) = plan.schema().index_of(name) else {
return Ok(None);
};
let Ok(dim_index) = dimension.schema().index_of(dimension_key) else {
return Ok(None);
};
let on = vec![(
Arc::new(Column::new(dimension_key, dim_index)) as Arc<dyn PhysicalExpr>,
Arc::new(Column::new(name, fact_index)) as Arc<dyn PhysicalExpr>,
)];
let reducer = HashJoinExec::try_new(
Arc::clone(dimension),
Arc::clone(plan),
on,
None,
&JoinType::RightSemi,
None,
PartitionMode::CollectLeft,
datafusion::common::NullEquality::NullEqualsNothing,
false,
)
.map_err(|e| SqlError::DataFusion {
message: format!("dimension reducer: {e}"),
})?;
Ok(Some(Arc::new(reducer)))
}
pub fn redistribute_unsplittable_broadcast_joins(
plan: Arc<dyn ExecutionPlan>,
) -> SqlResult<Arc<dyn ExecutionPlan>> {
use datafusion::physical_plan::joins::{HashJoinExec, PartitionMode};
let children = plan.children();
let plan = if children.is_empty() {
plan
} else {
let mut new_children = Vec::with_capacity(children.len());
let mut changed = false;
for child in children {
let rewritten = redistribute_unsplittable_broadcast_joins(Arc::clone(child))?;
changed = changed || !Arc::ptr_eq(&rewritten, child);
new_children.push(rewritten);
}
if changed {
plan.with_new_children(new_children)
.map_err(|e| SqlError::DataFusion {
message: format!("broadcast-join redistribution rewrite: {e}"),
})?
} else {
plan
}
};
let Some(join) = plan.downcast_ref::<HashJoinExec>() else {
return Ok(plan);
};
let unsplittable = is_unsplittable_broadcast_join(join);
if !unsplittable && !is_degenerate_broadcast_join(join) && !broadcast_build_is_too_wide(join) {
return Ok(plan);
}
let (left_keys, right_keys): (Vec<_>, Vec<_>) = join
.on()
.iter()
.map(|(l, r)| (Arc::clone(l), Arc::clone(r)))
.unzip();
let build_side = match join.left().downcast_ref::<CoalescePartitionsExec>() {
Some(coalesce) => Arc::clone(coalesce.input()),
None => Arc::clone(join.left()),
};
let partitions = join
.right()
.output_partitioning()
.partition_count()
.max(build_side.output_partitioning().partition_count());
let exchange = |input: Arc<dyn ExecutionPlan>,
keys: Vec<Arc<dyn datafusion::physical_expr::PhysicalExpr>>|
-> SqlResult<Arc<dyn ExecutionPlan>> {
RepartitionExec::try_new(input, Partitioning::Hash(keys, partitions))
.map(|r| Arc::new(r) as Arc<dyn ExecutionPlan>)
.map_err(|e| SqlError::DataFusion {
message: format!("broadcast-join redistribution exchange: {e}"),
})
};
let converted = join
.builder()
.with_new_children(vec![
exchange(build_side, left_keys)?,
exchange(Arc::clone(join.right()), right_keys)?,
])
.and_then(|b| {
b.with_partition_mode(PartitionMode::Partitioned)
.recompute_properties()
.reset_state()
.build_exec()
})
.map_err(|e| SqlError::DataFusion {
message: format!("broadcast-join redistribution rebuild: {e}"),
})?;
tracing::debug!(
join_type = ?join.join_type(),
partitions,
reason = if unsplittable { "unsplittable" } else { "oversized" },
"converted a broadcast join to a hash-partitioned join"
);
Ok(converted)
}
fn wrap_in_scalar_subquery_exec(
plan: Arc<dyn ExecutionPlan>,
context: &StageSubqueryContext,
) -> Arc<dyn ExecutionPlan> {
Arc::new(ScalarSubqueryExec::new(
plan,
context.links.clone(),
context.results.clone(),
))
}
fn hash_expr_column_names(
exprs: &[Arc<dyn datafusion::physical_expr::PhysicalExpr>],
) -> Option<Vec<String>> {
use datafusion::physical_expr::expressions::Column;
let mut names = Vec::with_capacity(exprs.len());
for expr in exprs {
let column = (expr.as_ref() as &dyn std::any::Any).downcast_ref::<Column>()?;
names.push(column.name().to_owned());
}
(!names.is_empty()).then_some(names)
}
fn find_unsupported_stage_node(plan: &Arc<dyn ExecutionPlan>) -> Option<String> {
if plan.is::<RepartitionExec>() {
return Some(String::from("RepartitionExec inside stage subtree"));
}
if let Some(join) = plan.downcast_ref::<datafusion::physical_plan::joins::HashJoinExec>()
&& is_unsplittable_broadcast_join(join)
{
return Some(format!(
"broadcast {:?} join inside a stage subtree: its unmatched build rows are \
emitted only after the last probe partition, which a task executing one \
partition can never observe",
join.join_type()
));
}
if let Some(subquery_exec) = plan.downcast_ref::<ScalarSubqueryExec>() {
return find_unsupported_stage_node(subquery_exec.input());
}
for child in plan.children() {
if let Some(reason) = find_unsupported_stage_node(child) {
return Some(reason);
}
}
None
}
const RUNTIME_FILTER_MIN_RATIO: usize = 8;
#[derive(Debug, Clone, Copy)]
struct RuntimeFilterCandidate {
build_stage: usize,
probe_stage: usize,
build_key_index: usize,
probe_key_index: usize,
filter_bytes: usize,
}
#[derive(Debug, Clone, Copy)]
struct JoinSideRead {
stage: usize,
rows: Option<usize>,
}
fn join_side_read(plan: &Arc<dyn ExecutionPlan>) -> Option<JoinSideRead> {
let mut current = Arc::clone(plan);
loop {
if let Some(read) = current.downcast_ref::<ShuffleReadExec>() {
return Some(JoinSideRead {
stage: read.upstream_stage_index(),
rows: read.upstream_estimate().0,
});
}
let next = {
let children = current.children();
let [child] = children.as_slice() else {
return None;
};
if current.schema().fields() != child.schema().fields() {
return None;
}
Arc::clone(child)
};
current = next;
}
}
#[derive(Debug, Default, Clone, Copy)]
struct RuntimeFilterRejects {
joins: usize,
not_inner: usize,
side_not_a_shuffle_read: usize,
same_stage: usize,
no_row_estimate: usize,
not_selective: usize,
filter_too_large: usize,
no_encodable_key: usize,
joins_of_unsupported_kind: usize,
}
struct JoinView<'a> {
join_type: datafusion::logical_expr::JoinType,
left: &'a Arc<dyn ExecutionPlan>,
right: &'a Arc<dyn ExecutionPlan>,
on: datafusion::physical_plan::joins::utils::JoinOnRef<'a>,
}
fn runtime_filter_candidate(
join: &JoinView<'_>,
rejects: &mut RuntimeFilterRejects,
) -> Option<RuntimeFilterCandidate> {
use datafusion::logical_expr::JoinType;
use datafusion::physical_expr::expressions::Column;
use krishiv_shuffle::{FilterKeyType, MAX_FILTER_BYTES, plan_filter_bytes};
rejects.joins += 1;
if join.join_type != JoinType::Inner {
rejects.not_inner += 1;
return None;
}
let (Some(build), Some(probe)) = (join_side_read(join.left), join_side_read(join.right)) else {
rejects.side_not_a_shuffle_read += 1;
return None;
};
if build.stage == probe.stage {
rejects.same_stage += 1;
return None;
}
let (Some(build_rows), Some(probe_rows)) = (build.rows, probe.rows) else {
rejects.no_row_estimate += 1;
return None;
};
if build_rows == 0 || probe_rows / RUNTIME_FILTER_MIN_RATIO < build_rows {
rejects.not_selective += 1;
return None;
}
let filter_bytes = plan_filter_bytes(build_rows as u64);
if filter_bytes >= MAX_FILTER_BYTES {
rejects.filter_too_large += 1;
return None;
}
let found = join.on.iter().find_map(|(left, right)| {
let build_column = (left.as_ref() as &dyn std::any::Any).downcast_ref::<Column>()?;
let probe_column = (right.as_ref() as &dyn std::any::Any).downcast_ref::<Column>()?;
let build_schema = join.left.schema();
let probe_schema = join.right.schema();
let build_type =
FilterKeyType::for_data_type(build_schema.field(build_column.index()).data_type())?;
let probe_type =
FilterKeyType::for_data_type(probe_schema.field(probe_column.index()).data_type())?;
(build_type == probe_type).then_some(RuntimeFilterCandidate {
build_stage: build.stage,
probe_stage: probe.stage,
build_key_index: build_column.index(),
probe_key_index: probe_column.index(),
filter_bytes,
})
});
if found.is_none() {
rejects.no_encodable_key += 1;
}
found
}
fn collect_runtime_filter_candidates(
plan: &Arc<dyn ExecutionPlan>,
out: &mut Vec<RuntimeFilterCandidate>,
rejects: &mut RuntimeFilterRejects,
) {
if let Some(join) = plan.downcast_ref::<datafusion::physical_plan::joins::HashJoinExec>() {
let view = JoinView {
join_type: *join.join_type(),
left: join.left(),
right: join.right(),
on: join.on(),
};
if let Some(candidate) = runtime_filter_candidate(&view, rejects) {
out.push(candidate);
}
} else if let Some(join) =
plan.downcast_ref::<datafusion::physical_plan::joins::SortMergeJoinExec>()
{
let view = JoinView {
join_type: join.join_type(),
left: join.left(),
right: join.right(),
on: join.on(),
};
if let Some(candidate) = runtime_filter_candidate(&view, rejects) {
out.push(candidate);
}
} else if plan
.downcast_ref::<datafusion::physical_plan::joins::NestedLoopJoinExec>()
.is_some()
|| plan
.downcast_ref::<crate::grace_hash_join::GraceHashJoinExec>()
.is_some()
{
rejects.joins_of_unsupported_kind += 1;
}
for child in plan.children() {
collect_runtime_filter_candidates(child, out, rejects);
}
}
fn stage_depends_on(drafts: &[StageDraft], from: usize, target: usize) -> bool {
let mut seen = vec![false; drafts.len()];
let mut stack = vec![from];
while let Some(index) = stack.pop() {
if index == target {
return true;
}
match seen.get_mut(index) {
Some(flag) if !*flag => *flag = true,
_ => continue,
}
if let Some(draft) = drafts.get(index) {
stack.extend(collect_upstream_stage_indexes(&draft.plan));
}
}
false
}
pub const STAGE_REUSE_ENV: &str = "KRISHIV_STAGE_REUSE";
pub fn stage_reuse_enabled() -> bool {
std::env::var(STAGE_REUSE_ENV)
.map(|v| {
let v = v.trim();
v == "1" || v.eq_ignore_ascii_case("true") || v.eq_ignore_ascii_case("on")
})
.unwrap_or(false)
}
const VOLATILE_MARKERS: &[&str] = &[
"random(",
"rand(",
"uuid(",
"now(",
"current_timestamp",
"current_date",
"current_time",
"nextval",
];
fn remap_shuffle_reads(
plan: &Arc<dyn ExecutionPlan>,
remap: &std::collections::HashMap<usize, usize>,
) -> Arc<dyn ExecutionPlan> {
if let Some(read) = plan.downcast_ref::<ShuffleReadExec>() {
let old = read.upstream_stage_index();
if let Some(&new) = remap.get(&old)
&& new != old
{
return Arc::new(read.clone_with_upstream_stage_index(new));
}
return Arc::clone(plan);
}
let children = plan.children();
if children.is_empty() {
return Arc::clone(plan);
}
let new_children: Vec<_> = children
.iter()
.map(|child| remap_shuffle_reads(child, remap))
.collect();
let changed = new_children
.iter()
.zip(children.iter())
.any(|(new, old)| !Arc::ptr_eq(new, old));
if !changed {
return Arc::clone(plan);
}
Arc::clone(plan)
.with_new_children(new_children)
.unwrap_or_else(|_| Arc::clone(plan))
}
fn dedupe_identical_stages(
root: &mut Arc<dyn ExecutionPlan>,
drafts: &mut Vec<StageDraft>,
) -> usize {
if !stage_reuse_enabled() {
return 0;
}
dedupe_identical_stages_unconditionally(root, drafts)
}
fn dedupe_identical_stages_unconditionally(
root: &mut Arc<dyn ExecutionPlan>,
drafts: &mut Vec<StageDraft>,
) -> usize {
use datafusion::physical_plan::displayable;
let codec = KrishivPhysicalCodec::coordinator();
let mut first_seen: std::collections::HashMap<Vec<u8>, usize> =
std::collections::HashMap::new();
let mut replaced_by: std::collections::HashMap<usize, usize> = std::collections::HashMap::new();
for (index, draft) in drafts.iter().enumerate() {
if draft.subqueries.is_some() {
continue;
}
if !collect_upstream_stage_indexes(&draft.plan).is_empty() {
continue;
}
let Some(shuffle) = &draft.shuffle else {
continue;
};
let text = displayable(draft.plan.as_ref()).indent(true).to_string();
let lowered = text.to_ascii_lowercase();
if VOLATILE_MARKERS.iter().any(|m| lowered.contains(m)) {
continue;
}
let Ok(encoded) = encode_dfplan_bytes(Arc::clone(&draft.plan), &codec) else {
continue;
};
let mut key = encoded;
key.extend_from_slice(
format!(
"|keys={:?}|parts={}|map_tasks={}|schema={:?}",
shuffle.key_columns,
shuffle.num_output_partitions,
draft.plan.output_partitioning().partition_count(),
draft.plan.schema()
)
.as_bytes(),
);
match first_seen.get(&key) {
Some(&canonical) => {
replaced_by.insert(index, canonical);
}
None => {
first_seen.insert(key, index);
}
}
}
if replaced_by.is_empty() {
return 0;
}
let mut remap: std::collections::HashMap<usize, usize> = std::collections::HashMap::new();
let mut survivors: Vec<StageDraft> = Vec::with_capacity(drafts.len() - replaced_by.len());
for (old_index, draft) in std::mem::take(drafts).into_iter().enumerate() {
if replaced_by.contains_key(&old_index) {
continue;
}
remap.insert(old_index, survivors.len());
survivors.push(draft);
}
for (duplicate, canonical) in &replaced_by {
if let Some(&new_canonical) = remap.get(canonical) {
remap.insert(*duplicate, new_canonical);
}
}
let removed = replaced_by.len();
for draft in &mut survivors {
draft.plan = remap_shuffle_reads(&draft.plan, &remap);
}
*root = remap_shuffle_reads(root, &remap);
*drafts = survivors;
tracing::info!(
removed_stages = removed,
"collapsed identical leaf stages (stage reuse)"
);
removed
}
fn inject_runtime_filters(root: &Arc<dyn ExecutionPlan>, drafts: &mut Vec<StageDraft>) -> usize {
if !crate::runtime_filter_exec::enabled() {
report_runtime_filter_candidates(root, drafts);
return 0;
}
inject_runtime_filters_unconditionally(root, drafts)
}
fn report_runtime_filter_candidates(root: &Arc<dyn ExecutionPlan>, drafts: &[StageDraft]) {
let mut candidates = Vec::new();
let mut rejects = RuntimeFilterRejects::default();
collect_runtime_filter_candidates(root, &mut candidates, &mut rejects);
for draft in drafts {
collect_runtime_filter_candidates(&draft.plan, &mut candidates, &mut rejects);
}
tracing::info!(
joins_inspected = rejects.joins,
not_inner = rejects.not_inner,
side_not_a_shuffle_read = rejects.side_not_a_shuffle_read,
same_stage = rejects.same_stage,
no_row_estimate = rejects.no_row_estimate,
not_selective = rejects.not_selective,
filter_too_large = rejects.filter_too_large,
no_encodable_key = rejects.no_encodable_key,
joins_of_unsupported_kind = rejects.joins_of_unsupported_kind,
candidates = candidates.len(),
injected = 0,
enabled = false,
"runtime-filter: pass complete"
);
}
fn inject_runtime_filters_unconditionally(
root: &Arc<dyn ExecutionPlan>,
drafts: &mut Vec<StageDraft>,
) -> usize {
use crate::runtime_filter_exec::{
RuntimeFilterBuildExec, RuntimeFilterProbeExec, filter_schema,
};
use datafusion::physical_expr::expressions::Column;
use datafusion::physical_plan::projection::ProjectionExec;
let mut candidates = Vec::new();
let mut rejects = RuntimeFilterRejects::default();
collect_runtime_filter_candidates(root, &mut candidates, &mut rejects);
for draft in drafts.iter() {
collect_runtime_filter_candidates(&draft.plan, &mut candidates, &mut rejects);
}
let mut touched: Vec<usize> = Vec::new();
let mut injected = 0usize;
let candidate_count = candidates.len();
let (mut already_touched, mut missing_stage, mut severed_subquery) = (0usize, 0usize, 0usize);
let (mut would_cycle, mut no_key_field, mut build_failed, mut probe_failed) =
(0usize, 0usize, 0usize, 0usize);
for candidate in candidates {
if touched.contains(&candidate.build_stage) || touched.contains(&candidate.probe_stage) {
already_touched += 1;
continue;
}
let (Some(build), Some(probe)) = (
drafts.get(candidate.build_stage),
drafts.get(candidate.probe_stage),
) else {
missing_stage += 1;
continue;
};
if build.subqueries.is_some() || probe.subqueries.is_some() {
severed_subquery += 1;
continue;
}
if stage_depends_on(drafts, candidate.build_stage, candidate.probe_stage) {
would_cycle += 1;
continue;
}
let source = Arc::clone(&build.plan);
let probe_plan = Arc::clone(&probe.plan);
let schema = source.schema();
let Some(field) = schema.fields().get(candidate.build_key_index) else {
no_key_field += 1;
continue;
};
let name = field.name().clone();
let projected = ProjectionExec::try_new(
vec![(
Arc::new(Column::new(&name, candidate.build_key_index)) as _,
name.clone(),
)],
source,
);
let filter_plan = projected.and_then(|projected| {
let coalesced = Arc::new(CoalescePartitionsExec::new(Arc::new(projected)));
RuntimeFilterBuildExec::try_new(coalesced, 0, candidate.filter_bytes)
});
let filter_plan = match filter_plan {
Ok(plan) => Arc::new(plan) as Arc<dyn ExecutionPlan>,
Err(error) => {
build_failed += 1;
tracing::debug!(%error, "declined to build a runtime filter stage");
continue;
}
};
let filter_stage = drafts.len();
let read = ShuffleReadExec::new(filter_stage, 1, 1, filter_schema(), None);
let rewritten =
RuntimeFilterProbeExec::try_new(probe_plan, Arc::new(read), candidate.probe_key_index);
let rewritten = match rewritten {
Ok(plan) => Arc::new(plan) as Arc<dyn ExecutionPlan>,
Err(error) => {
probe_failed += 1;
tracing::debug!(%error, "declined to apply a runtime filter to the probe stage");
continue;
}
};
let Some(probe_draft) = drafts.get_mut(candidate.probe_stage) else {
missing_stage += 1;
continue;
};
probe_draft.plan = rewritten;
drafts.push(StageDraft {
plan: filter_plan,
shuffle: Some(StageShuffleOutput {
key_columns: Vec::new(),
num_output_partitions: 1,
}),
subqueries: None,
});
touched.push(candidate.build_stage);
touched.push(candidate.probe_stage);
injected += 1;
tracing::info!(
build_stage = candidate.build_stage,
probe_stage = candidate.probe_stage,
filter_stage,
filter_bytes = candidate.filter_bytes,
"injected a cross-stage runtime filter"
);
}
tracing::info!(
joins_inspected = rejects.joins,
not_inner = rejects.not_inner,
side_not_a_shuffle_read = rejects.side_not_a_shuffle_read,
same_stage = rejects.same_stage,
no_row_estimate = rejects.no_row_estimate,
not_selective = rejects.not_selective,
filter_too_large = rejects.filter_too_large,
no_encodable_key = rejects.no_encodable_key,
joins_of_unsupported_kind = rejects.joins_of_unsupported_kind,
candidates = candidate_count,
injected,
already_touched,
missing_stage,
severed_subquery,
would_cycle,
no_key_field,
build_failed,
probe_failed,
"runtime-filter: pass complete"
);
injected
}
fn collect_upstream_stage_indexes(plan: &Arc<dyn ExecutionPlan>) -> Vec<usize> {
let mut indexes = Vec::new();
collect_upstream_inner(plan, &mut indexes);
indexes.sort_unstable();
indexes.dedup();
indexes
}
fn collect_upstream_inner(plan: &Arc<dyn ExecutionPlan>, out: &mut Vec<usize>) {
if let Some(read) = plan.downcast_ref::<ShuffleReadExec>() {
out.push(read.upstream_stage_index());
}
for child in plan.children() {
collect_upstream_inner(child, out);
}
}
#[cfg(test)]
mod tests {
fn plan_has_scalar_subquery(plan: &datafusion::logical_expr::LogicalPlan) -> bool {
use datafusion::common::tree_node::{TreeNode, TreeNodeRecursion};
use datafusion::logical_expr::Expr;
let mut found = false;
let _ = plan.apply(|node| {
if node
.expressions()
.iter()
.any(Expr::contains_scalar_subquery)
{
found = true;
return Ok(TreeNodeRecursion::Stop);
}
Ok(TreeNodeRecursion::Continue)
});
found
}
#[tokio::test]
async fn an_uncorrelated_scalar_subquery_is_folded_so_the_query_can_stage() {
let ctx = planning_session_context(4);
ctx.sql(
"CREATE TABLE acct(id BIGINT, bal DOUBLE) AS VALUES (1, 10.0), (2, 30.0), (3, 50.0)",
)
.await
.unwrap()
.collect()
.await
.unwrap();
let sql = "SELECT id FROM acct WHERE bal > (SELECT avg(bal) FROM acct)";
let before = ctx.sql(sql).await.unwrap();
assert!(
plan_has_scalar_subquery(before.logical_plan()),
"precondition: the planned query must actually carry a scalar \
subquery, or this test proves nothing"
);
let after = inline_uncorrelated_scalar_subqueries(&ctx, before)
.await
.unwrap();
assert!(
!plan_has_scalar_subquery(after.logical_plan()),
"the uncorrelated subquery must be folded to a constant"
);
let rows = after.collect().await.unwrap();
let total: usize = rows.iter().map(|b| b.num_rows()).sum();
assert_eq!(
total, 1,
"folding a constant must not change the result set"
);
}
#[tokio::test]
async fn a_correlated_scalar_subquery_is_left_alone() {
let ctx = planning_session_context(4);
ctx.sql("CREATE TABLE t(k BIGINT, v DOUBLE) AS VALUES (1, 10.0), (2, 30.0)")
.await
.unwrap()
.collect()
.await
.unwrap();
ctx.sql("CREATE TABLE u(k BIGINT, w DOUBLE) AS VALUES (1, 5.0), (2, 40.0)")
.await
.unwrap()
.collect()
.await
.unwrap();
let sql = "SELECT k FROM t WHERE v > (SELECT max(w) FROM u WHERE u.k = t.k)";
let Ok(before) = ctx.sql(sql).await else {
return;
};
let had_subquery = plan_has_scalar_subquery(before.logical_plan());
let after = inline_uncorrelated_scalar_subqueries(&ctx, before)
.await
.unwrap();
let still_has = plan_has_scalar_subquery(after.logical_plan());
assert_eq!(
had_subquery, still_has,
"a correlated subquery depends on the outer row and must never be \
folded to a constant"
);
}
use datafusion::prelude::SessionConfig;
#[test]
fn a_shuffle_read_reports_its_upstream_estimate_instead_of_unknown() {
use datafusion::common::stats::Precision;
let schema = Arc::new(arrow::datatypes::Schema::new(vec![
arrow::datatypes::Field::new("a", arrow::datatypes::DataType::Int64, false),
]));
let unknown = ShuffleReadExec::new(0, 4, 4, Arc::clone(&schema), None);
assert_eq!(
unknown.partition_statistics(None).unwrap().total_byte_size,
Precision::Absent,
"a read with no estimate must stay Absent — inventing a size is how \
a spill decision gets made on a guess"
);
let known = ShuffleReadExec::new(0, 4, 4, Arc::clone(&schema), None)
.with_upstream_estimate(Some(1_000), Some(800_000));
let whole = known.partition_statistics(None).unwrap();
assert_eq!(whole.num_rows, Precision::Inexact(1_000));
assert_eq!(
whole.total_byte_size,
Precision::Inexact(800_000),
"the whole-plan question gets the whole stage's size"
);
let one = known.partition_statistics(Some(0)).unwrap();
assert_eq!(
one.total_byte_size,
Precision::Inexact(200_000),
"a per-partition question gets the even-split share of 4 partitions"
);
}
#[test]
fn the_upstream_estimate_survives_encode_decode() {
let schema = Arc::new(arrow::datatypes::Schema::new(vec![
arrow::datatypes::Field::new("a", arrow::datatypes::DataType::Int64, false),
]));
let node: Arc<dyn ExecutionPlan> = Arc::new(
ShuffleReadExec::new(3, 2, 4, schema, None)
.with_upstream_estimate(Some(77), Some(4_096)),
);
let codec = KrishivPhysicalCodec::coordinator();
let mut buf = Vec::new();
codec.try_encode(Arc::clone(&node), &mut buf).unwrap();
let ctx = crate::SqlEngine::new_with_engine_memory(crate::EngineMemory::Unbounded);
let task_ctx = ctx.session_context().task_ctx();
let decoded = codec.try_decode(&buf, &[], &task_ctx).unwrap();
let mut round_tripped = Vec::new();
codec.try_encode(decoded, &mut round_tripped).unwrap();
assert_eq!(
String::from_utf8(round_tripped).unwrap(),
String::from_utf8(buf).unwrap(),
"the upstream estimate must survive encode -> decode -> encode"
);
}
use super::*;
use arrow::record_batch::RecordBatch;
use datafusion::physical_plan::displayable;
use datafusion_proto::physical_plan::DefaultPhysicalExtensionCodec;
use std::collections::HashMap;
use std::sync::Mutex;
#[test]
fn the_roundtrip_guard_rejects_bytes_that_do_not_decode() {
let codec = DefaultPhysicalExtensionCodec {};
let err = verify_dfplan_roundtrip(
b"not a physical plan proto",
&codec,
&fragment_decode_session_context().task_ctx(),
None,
)
.expect_err("undecodable bytes must be rejected");
assert!(format!("{err}").contains("decode"), "got: {err}");
}
#[test]
fn the_verify_context_resolves_object_stores_like_the_executor() {
use datafusion::execution::object_store::ObjectStoreUrl;
let url = ObjectStoreUrl::parse("s3://roundtrip-bucket").expect("url");
assert!(
SessionContext::new()
.runtime_env()
.object_store(url.clone())
.is_err(),
"precondition: a bare context must NOT resolve s3, or this test proves nothing"
);
planning_session_context(1)
.task_ctx()
.runtime_env()
.object_store(url)
.expect("the verify context must resolve s3 buckets like the executor runtime");
}
fn optimizer_rule_names(ctx: &SessionContext) -> (Vec<String>, Vec<String>) {
let state = ctx.state();
(
state
.optimizers()
.iter()
.map(|r| r.name().to_owned())
.collect(),
state
.physical_optimizers()
.iter()
.map(|r| r.name().to_owned())
.collect(),
)
}
#[test]
fn the_staging_context_carries_the_engines_optimizer_rules() {
let engine = crate::SqlEngine::new_with_engine_memory(crate::EngineMemory::Unbounded);
let (engine_logical, engine_physical) = optimizer_rule_names(engine.session_context());
let staging = planning_session_context(engine.target_parallelism().get());
let (staging_logical, staging_physical) = optimizer_rule_names(&staging);
assert_eq!(
engine_logical, staging_logical,
"the staged planner must run the engine's logical optimizer rules; \
missing here means SemiJoinReductionThroughAggregate / \
SemiJoinPushdownThroughInnerJoin never fire distributed (D4)"
);
assert_eq!(
engine_physical, staging_physical,
"the staged planner must run the engine's physical optimizer rules; \
missing here means SpillableJoinSelection (D3) and \
CooperativeAmplifiers (distributed cancel) never fire"
);
let engine_opts = engine.session_context().copied_config();
let staging_opts = staging.copied_config();
for option in [
"datafusion.optimizer.enable_dynamic_filter_pushdown",
"datafusion.optimizer.enable_join_dynamic_filter_pushdown",
"datafusion.optimizer.enable_topk_dynamic_filter_pushdown",
"datafusion.optimizer.enable_aggregate_dynamic_filter_pushdown",
] {
assert_eq!(
engine_opts
.options()
.entries()
.iter()
.find(|e| e.key == option)
.map(|e| e.value.clone()),
staging_opts
.options()
.entries()
.iter()
.find(|e| e.key == option)
.map(|e| e.value.clone()),
"{option} must match the engine's setting on the staged planner"
);
}
assert_eq!(
engine_opts.options().sql_parser.dialect,
staging_opts.options().sql_parser.dialect,
"the staged planner must parse in the engine's dialect"
);
assert_eq!(
engine_opts.options().execution.batch_size,
staging_opts.options().execution.batch_size,
"the staged planner must use the engine's batch size"
);
}
#[tokio::test]
async fn the_roundtrip_guard_accepts_a_fragment_using_an_engine_udf() {
let engine = crate::SqlEngine::new_with_engine_memory(crate::EngineMemory::Unbounded);
let ctx = engine.session_context();
ctx.sql("CREATE TABLE docs AS VALUES ('{\"a\":1}'), ('{\"a\":2}')")
.await
.unwrap()
.collect()
.await
.unwrap();
let plan = ctx
.sql("SELECT get_json_object(column1, '$.a') AS a FROM docs")
.await
.unwrap()
.create_physical_plan()
.await
.unwrap();
let codec = DefaultPhysicalExtensionCodec {};
let bytes = encode_dfplan_bytes(plan, &codec).expect("encode");
let bare = planning_session_context(1).task_ctx();
assert!(
datafusion_proto::bytes::physical_plan_from_bytes_with_extension_codec(
&bytes, &bare, &codec
)
.is_err(),
"precondition: a bare planning context must NOT resolve engine UDFs"
);
verify_dfplan_roundtrip(
&bytes,
&codec,
&fragment_decode_session_context().task_ctx(),
None,
)
.expect(
"the guard must decode on the engine the executor uses; failing here \
silently degrades the query to a single task",
);
}
#[tokio::test]
async fn the_roundtrip_guard_accepts_an_ordinary_plan() {
let ctx = SessionContext::new();
ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b')")
.await
.unwrap()
.collect()
.await
.unwrap();
let plan = ctx
.sql("SELECT column1 FROM t WHERE column1 > 1")
.await
.unwrap()
.create_physical_plan()
.await
.unwrap();
let codec = DefaultPhysicalExtensionCodec {};
let bytes = encode_dfplan_bytes(plan, &codec).expect("encode");
verify_dfplan_roundtrip(
&bytes,
&codec,
&fragment_decode_session_context().task_ctx(),
None,
)
.expect("ordinary plans must pass");
}
#[tokio::test]
async fn s3_paths_resolve_on_the_planning_context_and_explicit_registration_wins() {
use datafusion::execution::object_store::ObjectStoreUrl;
let ctx = planning_session_context(4);
let url = ObjectStoreUrl::parse("s3://tpch-bucket").expect("bucket url");
let lazily_built = ctx
.runtime_env()
.object_store(url.clone())
.expect("the planning context must resolve an s3 bucket on demand");
register_object_store_for_path(&ctx, "s3://tpch-bucket/tpch/sf100/lineitem/")
.expect("registering an s3 path must succeed");
let explicit = ctx
.runtime_env()
.object_store(url)
.expect("after registration the planning context must resolve the bucket");
assert!(
!Arc::ptr_eq(&lazily_built, &explicit),
"explicit registration must replace the lazily-constructed store, \
or configured endpoints and credentials would be ignored"
);
}
#[tokio::test]
async fn local_paths_are_left_alone_by_object_store_registration() {
let ctx = planning_session_context(4);
register_object_store_for_path(&ctx, "/home/krishiv-bench-data/tpch/sf1/lineitem.parquet")
.expect("a local path must be a no-op, not an error");
register_object_store_for_path(&ctx, "relative/dir")
.expect("a relative local path must be a no-op, not an error");
}
async fn write_test_parquet(dir: &std::path::Path) -> std::path::PathBuf {
use arrow::array::{Int64Array, StringArray};
use arrow::datatypes::{DataType, Field, Schema};
let schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("category", DataType::Utf8, false),
Field::new("amount", DataType::Int64, false),
]));
let table_dir = dir.join("t");
std::fs::create_dir_all(&table_dir).expect("table dir");
for file_index in 0..4i64 {
let ids: Vec<i64> = (0..250).map(|i| file_index * 250 + i).collect();
let batch = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(Int64Array::from(ids.clone())),
Arc::new(StringArray::from(
ids.iter()
.map(|i| match i % 3 {
0 => "red",
1 => "green",
_ => "blue",
})
.collect::<Vec<_>>(),
)),
Arc::new(Int64Array::from(
ids.iter().map(|i| i * 3).collect::<Vec<_>>(),
)),
],
)
.expect("test batch");
let path = table_dir.join(format!("part-{file_index}.parquet"));
let file = std::fs::File::create(&path).expect("create parquet");
let mut writer =
datafusion::parquet::arrow::ArrowWriter::try_new(file, schema.clone(), None)
.expect("writer init");
writer.write(&batch).expect("write batch");
writer.close().expect("close writer");
}
table_dir
}
#[tokio::test]
async fn a_declared_primary_key_shrinks_the_group_by() {
let tmp = tempfile::tempdir().expect("tempdir");
let table_dir = write_test_parquet(tmp.path()).await;
let path = table_dir.to_string_lossy().to_string();
let sql = "SELECT id, count(*) AS n FROM t GROUP BY id, category, amount";
let plan_text = |spec: ParquetTableSpec| async move {
let ctx = planning_session_context(4);
register_parquet_table(&ctx, &spec)
.await
.expect("register table");
let df = ctx.sql(sql).await.expect("plan sql");
let optimized = df.into_optimized_plan().expect("optimize");
format!("{}", optimized.display_indent())
};
let without = plan_text(ParquetTableSpec::new("t", &path)).await;
let with = plan_text(ParquetTableSpec::new("t", &path).with_primary_key(["id"])).await;
let group_line = |text: &str| {
text.lines()
.find(|line| line.contains("Aggregate:"))
.unwrap_or("<no Aggregate>")
.to_owned()
};
let (g_without, g_with) = (group_line(&without), group_line(&with));
assert_ne!(
g_without, g_with,
"declaring a primary key changed nothing about the aggregate; \
the constraint is not reaching DataFusion's functional-dependency \
machinery.\n without: {g_without}\n with: {g_with}"
);
assert!(
g_with.len() < g_without.len(),
"the declared key should SHRINK the grouping list, not grow it.\n\
without: {g_without}\n with: {g_with}"
);
}
#[tokio::test]
async fn an_unknown_primary_key_column_is_rejected() {
let tmp = tempfile::tempdir().expect("tempdir");
let table_dir = write_test_parquet(tmp.path()).await;
let ctx = planning_session_context(4);
let spec = ParquetTableSpec::new("t", table_dir.to_string_lossy().as_ref())
.with_primary_key(["nonexistent_column"]);
let error = register_parquet_table(&ctx, &spec)
.await
.expect_err("an unknown key column must be rejected");
let message = error.to_string();
assert!(
message.contains("nonexistent_column") && message.contains("not in table"),
"the error must name the offending column and the table: {message}"
);
}
#[test]
fn target_partitions_scale_with_the_cluster_not_a_constant() {
let two_slots = ClusterCapacity { total_slots: 2 };
let thirty_two = ClusterCapacity { total_slots: 32 };
let small = derive_stage_target_partitions(None, Some(two_slots), 8);
let large = derive_stage_target_partitions(None, Some(thirty_two), 8);
assert!(
large > small,
"a 16x larger cluster planned {large} vs {small} partitions"
);
assert_eq!(large, 32 * TASKS_PER_SLOT);
}
#[test]
fn multiple_waves_per_slot_leave_room_to_absorb_stragglers() {
let cluster = ClusterCapacity { total_slots: 8 };
assert!(
derive_stage_target_partitions(None, Some(cluster), 8) > cluster.total_slots,
"a stage should plan more tasks than slots, not exactly one wave"
);
}
#[test]
fn an_explicit_setting_overrides_the_derivation() {
let cluster = ClusterCapacity { total_slots: 64 };
assert_eq!(derive_stage_target_partitions(Some(6), Some(cluster), 8), 6);
assert!(derive_stage_target_partitions(Some(1), Some(cluster), 8) >= MIN_STAGE_PARTITIONS);
assert!(derive_stage_target_partitions(Some(0), Some(cluster), 8) >= MIN_STAGE_PARTITIONS);
}
#[test]
fn no_cluster_view_falls_back_to_the_local_machine() {
assert_eq!(
derive_stage_target_partitions(None, None, 6),
6 * TASKS_PER_SLOT
);
}
#[test]
fn partition_counts_stay_inside_the_shuffle_fragment_budget() {
let huge = ClusterCapacity {
total_slots: usize::MAX,
};
assert_eq!(
derive_stage_target_partitions(None, Some(huge), 8),
MAX_STAGE_PARTITIONS
);
let one = ClusterCapacity { total_slots: 1 };
assert!(derive_stage_target_partitions(None, Some(one), 1) >= MIN_STAGE_PARTITIONS);
}
#[tokio::test]
async fn ungrouped_aggregate_splits_into_map_and_reduce_stages() {
let tmp = tempfile::tempdir().expect("tempdir");
let path = write_test_parquet(tmp.path()).await;
let tables = vec![(
String::from("t"),
path.to_str().expect("utf8 path").to_owned(),
)];
let staged = build_stages_for_parquet_query(
"SELECT SUM(amount) AS total, COUNT(*) AS n FROM t WHERE id >= 100",
&tables,
Some(ClusterCapacity { total_slots: 4 }),
)
.await
.expect("planning must not error")
.expect("an ungrouped aggregate must be stage-split, not declined");
assert!(
staged.stages.len() >= 2,
"expected a map stage and a reduce stage, got {} stage(s) — \
the gather was not cut, so the whole scan runs in one task",
staged.stages.len()
);
let map = &staged.stages[0];
let shuffle = map
.shuffle
.as_ref()
.expect("the map stage must write a shuffle output");
assert_eq!(
shuffle.num_output_partitions, 1,
"a gather must produce exactly one reduce partition"
);
assert!(
shuffle.key_columns.is_empty(),
"a gather has no partitioning key; got {:?}",
shuffle.key_columns
);
}
#[tokio::test]
async fn grouped_aggregate_still_cuts_at_the_hash_exchange() {
let tmp = tempfile::tempdir().expect("tempdir");
let path = write_test_parquet(tmp.path()).await;
let tables = vec![(
String::from("t"),
path.to_str().expect("utf8 path").to_owned(),
)];
let staged = build_stages_for_parquet_query(
"SELECT category, SUM(amount) AS total FROM t GROUP BY category",
&tables,
Some(ClusterCapacity { total_slots: 4 }),
)
.await
.expect("planning must not error")
.expect("a grouped aggregate must be stage-split");
let map = &staged.stages[0];
let shuffle = map.shuffle.as_ref().expect("map stage writes a shuffle");
assert_eq!(
shuffle.key_columns,
vec![String::from("category")],
"a grouped aggregate must shuffle on its grouping key"
);
}
#[tokio::test]
async fn aggregate_plan_round_trips_through_proto() {
let tmp = tempfile::tempdir().expect("tempdir");
let path = write_test_parquet(tmp.path()).await;
let ctx = SessionContext::new();
ctx.register_parquet(
"t",
path.to_str().expect("utf8 path"),
datafusion::prelude::ParquetReadOptions::default(),
)
.await
.expect("register parquet");
let df = ctx
.sql("SELECT category, COUNT(*) AS n, SUM(amount) AS total FROM t WHERE id >= 100 GROUP BY category")
.await
.expect("sql");
let plan = df.create_physical_plan().await.expect("physical plan");
let original_display = displayable(plan.as_ref()).indent(true).to_string();
let codec = DefaultPhysicalExtensionCodec {};
let bytes = encode_dfplan_bytes(Arc::clone(&plan), &codec).expect("encode");
let b64 = base64::engine::general_purpose::STANDARD.encode(&bytes);
let body = dfplan_task_body(&b64, 0);
assert!(is_dfplan_body(&body));
let exec_ctx = SessionContext::new();
let (spec, decoded) =
decode_dfplan_task(&body, &exec_ctx.task_ctx(), &codec).expect("decode");
assert_eq!(spec, DfplanTaskSpec::single(0));
assert_eq!(
original_display,
displayable(decoded.as_ref()).indent(true).to_string(),
"decoded plan display must match original"
);
let task_ctx = exec_ctx.task_ctx();
let mut results = Vec::new();
for partition in 0..decoded.output_partitioning().partition_count() {
let stream = decoded
.execute(partition, Arc::clone(&task_ctx))
.expect("execute decoded partition");
let batches: Vec<_> = futures::TryStreamExt::try_collect(stream)
.await
.expect("collect decoded stream");
results.extend(batches);
}
let total_rows: usize = results.iter().map(|b| b.num_rows()).sum();
assert_eq!(total_rows, 3, "three category groups expected");
}
#[test]
fn non_dfplan_body_is_rejected() {
let err = parse_dfplan_body("sql: SELECT 1").unwrap_err();
assert!(err.to_string().contains("not a dfplan:v1: fragment"));
}
#[test]
fn a_shuffle_batch_that_contradicts_the_declared_schema_is_named_not_passed_on() {
use arrow::array::{Int64Array, StringViewArray};
use arrow::datatypes::{DataType, Field, Schema};
let declared: SchemaRef = Arc::new(Schema::new(vec![Field::new(
"revenue",
DataType::Decimal128(15, 2),
false,
)]));
let batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new(
"p_brand",
DataType::Utf8View,
false,
)])),
vec![Arc::new(StringViewArray::from(vec!["Brand#23"]))],
)
.expect("utf8view batch");
let error = check_shuffle_batch_schema(&declared, batch, 3, 7, 5)
.expect_err("a contradicting batch must not be passed on");
let text = error.to_string();
for expected in ["stage 3", "map 7", "partition 5", "revenue", "p_brand"] {
assert!(
text.contains(expected),
"error must name {expected}, got: {text}"
);
}
let two_col = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("a", DataType::Int64, false),
Field::new("b", DataType::Int64, false),
])),
vec![
Arc::new(Int64Array::from(vec![1])),
Arc::new(Int64Array::from(vec![2])),
],
)
.expect("two column batch");
let error = check_shuffle_batch_schema(&declared, two_col, 0, 0, 0)
.expect_err("column-count disagreement must not be passed on");
assert!(
error
.to_string()
.contains("2 columns but the plan declares 1"),
"got: {error}"
);
}
#[test]
fn matching_column_types_pass_even_when_metadata_and_nullability_differ() {
use arrow::array::Int64Array;
use arrow::datatypes::{DataType, Field, Schema};
let declared: SchemaRef = Arc::new(Schema::new(vec![
Field::new("n", DataType::Int64, false).with_metadata(
[(String::from("origin"), String::from("coordinator"))]
.into_iter()
.collect(),
),
]));
let batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new("n", DataType::Int64, true)])),
vec![Arc::new(Int64Array::from(vec![1, 2, 3]))],
)
.expect("batch");
check_shuffle_batch_schema(&declared, batch, 0, 0, 0)
.expect("metadata and nullability differences must not fail the query");
}
#[derive(Debug, Default)]
struct TestShuffleStore {
partitions: Mutex<HashMap<(usize, usize, usize), Vec<RecordBatch>>>,
}
impl TestShuffleStore {
fn write(&self, stage: usize, map_task: usize, partition: usize, batch: RecordBatch) {
self.partitions
.lock()
.expect("store lock")
.entry((stage, map_task, partition))
.or_default()
.push(batch);
}
}
impl ShufflePartitionReader for Arc<TestShuffleStore> {
fn open_partition(
&self,
upstream_stage_index: usize,
map_task_index: usize,
partition: usize,
) -> futures::future::BoxFuture<'static, Result<ShuffleFragmentStream, String>> {
let batches = self
.partitions
.lock()
.expect("store lock")
.get(&(upstream_stage_index, map_task_index, partition))
.cloned()
.unwrap_or_default();
Box::pin(async move {
Ok(Box::pin(futures::stream::iter(batches.into_iter().map(Ok)))
as ShuffleFragmentStream)
})
}
}
#[derive(Debug)]
struct ServeLimitedReader {
inner: Arc<TestShuffleStore>,
permits: Arc<tokio::sync::Semaphore>,
reverse_open_order: bool,
map_tasks: usize,
}
impl ShufflePartitionReader for ServeLimitedReader {
fn open_partition(
&self,
stage: usize,
map_task: usize,
partition: usize,
) -> futures::future::BoxFuture<'static, Result<ShuffleFragmentStream, String>> {
let batches = self
.inner
.partitions
.lock()
.expect("store lock")
.get(&(stage, map_task, partition))
.cloned()
.unwrap_or_default();
let permits = Arc::clone(&self.permits);
let delay = if self.reverse_open_order {
20 * (self.map_tasks.saturating_sub(map_task)) as u64
} else {
0
};
Box::pin(async move {
if delay > 0 {
tokio::time::sleep(std::time::Duration::from_millis(delay)).await;
}
let permit = permits
.acquire_owned()
.await
.map_err(|_| String::from("serve semaphore closed"))?;
let held = futures::stream::iter(batches.into_iter().map(Ok)).chain(
futures::stream::unfold(Some(permit), |permit| async move {
drop(permit?);
None
}),
);
Ok(Box::pin(held) as ShuffleFragmentStream)
})
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_reduce_read_completes_even_when_the_producer_serves_one_at_a_time() {
use arrow::array::Int64Array;
use arrow::datatypes::{DataType, Field, Schema};
let store = Arc::new(TestShuffleStore::default());
let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, false)]));
let map_tasks = 4usize;
for map_task in 0..map_tasks {
let batch = RecordBatch::try_new(
Arc::clone(&schema),
vec![Arc::new(Int64Array::from(vec![map_task as i64; 3]))],
)
.expect("batch");
store.write(0, map_task, 0, batch);
}
let reader: Arc<dyn ShufflePartitionReader> = Arc::new(ServeLimitedReader {
inner: Arc::clone(&store),
permits: Arc::new(tokio::sync::Semaphore::new(map_tasks - 1)),
reverse_open_order: true,
map_tasks,
});
let read = ShuffleReadExec::new(0, map_tasks, 1, Arc::clone(&schema), Some(reader));
let ctx = SessionContext::new();
let stream = read.execute(0, ctx.task_ctx()).expect("execute");
let batches = tokio::time::timeout(
std::time::Duration::from_secs(20),
futures::TryStreamExt::try_collect::<Vec<_>>(stream),
)
.await
.expect(
"the reduce read deadlocked: it is holding more producer response \
streams open than the producer will serve, and only downstream \
consumption releases them",
)
.expect("read");
let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
assert_eq!(rows, map_tasks * 3, "every fragment's rows must arrive");
}
fn partition_batch_by_key(
batch: &RecordBatch,
key_column: &str,
num_partitions: usize,
) -> Vec<RecordBatch> {
use std::hash::{Hash as _, Hasher as _};
let key_idx = batch.schema().index_of(key_column).expect("key column");
let column = batch.column(key_idx);
let mut selections: Vec<Vec<u32>> = vec![Vec::new(); num_partitions];
for row in 0..batch.num_rows() {
let value = arrow::util::display::array_value_to_string(column, row).expect("value");
let mut hasher = std::collections::hash_map::DefaultHasher::new();
value.hash(&mut hasher);
let bucket = (hasher.finish() as usize) % num_partitions;
selections[bucket].push(row as u32);
}
selections
.into_iter()
.map(|rows| {
let indices = arrow::array::UInt32Array::from(rows);
arrow::compute::take_record_batch(batch, &indices).expect("take")
})
.collect()
}
#[tokio::test]
async fn staged_group_by_matches_direct_execution() {
let tmp = tempfile::tempdir().expect("tempdir");
let path = write_test_parquet(tmp.path()).await;
let plan_ctx = planning_session_context(4);
plan_ctx
.register_parquet(
"t",
path.to_str().expect("utf8 path"),
datafusion::prelude::ParquetReadOptions::default(),
)
.await
.expect("register parquet");
let query = "SELECT category, COUNT(*) AS n, SUM(amount) AS total FROM t GROUP BY category ORDER BY category";
let df = plan_ctx.sql(query).await.expect("sql");
let plan = df.create_physical_plan().await.expect("physical plan");
let staged = build_distributed_stages(plan)
.expect("build stages")
.expect("plan must be splittable");
assert_eq!(staged.stages.len(), 2, "one map stage + one result stage");
let map_stage = &staged.stages[0];
let result_stage = &staged.stages[1];
let shuffle = map_stage.shuffle.as_ref().expect("map stage shuffles");
assert_eq!(shuffle.key_columns, vec!["category".to_owned()]);
assert!(
map_stage.task_count() > 1,
"multi-file scan must yield a multi-task map stage, got {}",
map_stage.task_count()
);
assert!(result_stage.shuffle.is_none());
assert_eq!(result_stage.upstream_stage_indexes, vec![0]);
let store = Arc::new(TestShuffleStore::default());
let exec_ctx = SessionContext::new();
let exec_codec = KrishivPhysicalCodec::executor(Arc::new(Arc::clone(&store)));
for (task_index, body) in map_stage.task_bodies.iter().enumerate() {
let (spec, plan) =
decode_dfplan_task(body, &exec_ctx.task_ctx(), &exec_codec).expect("decode map");
assert_eq!(spec, DfplanTaskSpec::single(task_index));
let stream = plan
.execute(task_index, exec_ctx.task_ctx())
.expect("execute map partition");
let batches: Vec<_> = futures::TryStreamExt::try_collect(stream)
.await
.expect("collect map output");
for batch in batches {
if batch.num_rows() == 0 {
continue;
}
for (bucket, part) in partition_batch_by_key(
&batch,
&shuffle.key_columns[0],
shuffle.num_output_partitions,
)
.into_iter()
.enumerate()
{
if part.num_rows() > 0 {
store.write(0, task_index, bucket, part);
}
}
}
}
let mut staged_results = Vec::new();
for (task_index, body) in result_stage.task_bodies.iter().enumerate() {
let (spec, plan) =
decode_dfplan_task(body, &exec_ctx.task_ctx(), &exec_codec).expect("decode result");
assert_eq!(spec, DfplanTaskSpec::single(task_index));
let stream = plan
.execute(task_index, exec_ctx.task_ctx())
.expect("execute result partition");
let batches: Vec<_> = futures::TryStreamExt::try_collect(stream)
.await
.expect("collect result output");
staged_results.extend(batches);
}
let direct = plan_ctx
.sql(query)
.await
.expect("direct sql")
.collect()
.await
.expect("direct collect");
let render = |batches: &[RecordBatch]| {
let mut rows: Vec<String> = batches
.iter()
.flat_map(|b| {
(0..b.num_rows()).map(move |r| {
(0..b.num_columns())
.map(|c| {
arrow::util::display::array_value_to_string(b.column(c), r)
.expect("cell")
})
.collect::<Vec<_>>()
.join("|")
})
})
.collect();
rows.sort();
rows
};
assert_eq!(
render(&staged_results),
render(&direct),
"staged execution must match direct execution"
);
}
#[tokio::test]
async fn scan_only_plan_declines_with_a_stated_reason() {
let tmp = tempfile::tempdir().expect("tempdir");
let path = write_test_parquet(tmp.path()).await;
let plan_ctx = planning_session_context(4);
plan_ctx
.register_parquet(
"t",
path.to_str().expect("utf8 path"),
datafusion::prelude::ParquetReadOptions::default(),
)
.await
.expect("register parquet");
let df = plan_ctx
.sql("SELECT id, amount FROM t WHERE id < 10")
.await
.expect("sql");
let plan = df.create_physical_plan().await.expect("physical plan");
let reason = build_distributed_stages(plan)
.expect_err("a scan-only plan has no exchange and must decline")
.to_string();
assert!(
reason.contains("no exchange"),
"the decline must name the missing plan property, got: {reason}"
);
}
#[tokio::test]
async fn staged_join_matches_direct_execution() {
let tmp = tempfile::tempdir().expect("tempdir");
let path = write_test_parquet(tmp.path()).await;
let mut config = SessionConfig::new().with_target_partitions(4);
config
.options_mut()
.optimizer
.enable_round_robin_repartition = false;
config
.options_mut()
.optimizer
.hash_join_single_partition_threshold = 0;
config
.options_mut()
.optimizer
.hash_join_single_partition_threshold_rows = 0;
let plan_ctx = SessionContext::new_with_config(config);
for name in ["a", "b"] {
plan_ctx
.register_parquet(
name,
path.to_str().expect("utf8 path"),
datafusion::prelude::ParquetReadOptions::default(),
)
.await
.expect("register parquet");
}
let query = "SELECT a.category, COUNT(*) AS n, SUM(b.amount) AS total \
FROM a JOIN b ON a.id = b.id GROUP BY a.category";
let df = plan_ctx.sql(query).await.expect("sql");
let plan = df.create_physical_plan().await.expect("physical plan");
let staged = build_distributed_stages(plan)
.expect("build stages")
.expect("partitioned join must split into stages");
assert!(
staged.stages.len() >= 3,
"expected two join-side map stages + result, got {}",
staged.stages.len()
);
let store = Arc::new(TestShuffleStore::default());
let exec_ctx = SessionContext::new();
let exec_codec = KrishivPhysicalCodec::executor(Arc::new(Arc::clone(&store)));
let mut staged_results = Vec::new();
for (stage_index, stage) in staged.stages.iter().enumerate() {
for (task_index, body) in stage.task_bodies.iter().enumerate() {
let (spec, plan) = decode_dfplan_task(body, &exec_ctx.task_ctx(), &exec_codec)
.expect("decode stage task");
assert_eq!(spec, DfplanTaskSpec::single(task_index));
let stream = plan
.execute(task_index, exec_ctx.task_ctx())
.expect("execute stage partition");
let batches: Vec<_> = futures::TryStreamExt::try_collect(stream)
.await
.expect("collect stage output");
match &stage.shuffle {
Some(shuffle) => {
for batch in batches {
if batch.num_rows() == 0 {
continue;
}
for (bucket, part) in partition_batch_by_key(
&batch,
&shuffle.key_columns[0],
shuffle.num_output_partitions,
)
.into_iter()
.enumerate()
{
if part.num_rows() > 0 {
store.write(stage_index, task_index, bucket, part);
}
}
}
}
None => staged_results.extend(batches),
}
}
}
let direct = plan_ctx
.sql(query)
.await
.expect("direct sql")
.collect()
.await
.expect("direct collect");
let render = |batches: &[RecordBatch]| {
let mut rows: Vec<String> = batches
.iter()
.flat_map(|b| {
(0..b.num_rows()).map(move |r| {
(0..b.num_columns())
.map(|c| {
arrow::util::display::array_value_to_string(b.column(c), r)
.expect("cell")
})
.collect::<Vec<_>>()
.join("|")
})
})
.collect();
rows.sort();
rows
};
assert_eq!(
render(&staged_results),
render(&direct),
"staged join must match direct execution"
);
}
#[test]
fn partition_spec_grammar_round_trips() {
let multi = DfplanTaskSpec {
partitions: vec![1, 4, 7],
map_range: None,
};
let body = dfplan_task_body_for_spec("QUJD", &multi);
assert_eq!(body, "dfplan:v1:1,4,7:QUJD");
assert_eq!(dfplan_body_partition_spec(&body).expect("parse"), multi);
let split = DfplanTaskSpec {
partitions: vec![5],
map_range: Some(DfplanMapRange {
upstream_stage_index: 0,
start: 2,
end: 4,
}),
};
let body = dfplan_task_body_for_spec("QUJD", &split);
assert_eq!(body, "dfplan:v1:5/s0m2-4:QUJD");
assert_eq!(dfplan_body_partition_spec(&body).expect("parse"), split);
assert_eq!(
dfplan_body_partition_spec("dfplan:v1:3:QUJD").expect("parse"),
DfplanTaskSpec::single(3)
);
}
#[test]
fn partition_spec_rewrite_preserves_payload() {
let original = dfplan_task_body("cGF5bG9hZA==", 2);
let rewritten = dfplan_body_with_spec(
&original,
&DfplanTaskSpec {
partitions: vec![0, 2],
map_range: None,
},
)
.expect("rewrite");
assert_eq!(rewritten, "dfplan:v1:0,2:cGF5bG9hZA==");
}
#[test]
fn partition_spec_rejects_malformed_segments() {
assert!(dfplan_body_partition_spec("dfplan:v1::QUJD").is_err());
assert!(dfplan_body_partition_spec("dfplan:v1:x:QUJD").is_err());
assert!(dfplan_body_partition_spec("dfplan:v1:1/s0m4-4:QUJD").is_err());
assert!(dfplan_body_partition_spec("dfplan:v1:1/m0-2:QUJD").is_err());
}
#[tokio::test]
async fn coalesced_result_stage_matches_direct_execution() {
let tmp = tempfile::tempdir().expect("tempdir");
let path = write_test_parquet(tmp.path()).await;
let plan_ctx = planning_session_context(4);
plan_ctx
.register_parquet(
"t",
path.to_str().expect("utf8 path"),
datafusion::prelude::ParquetReadOptions::default(),
)
.await
.expect("register parquet");
let query = "SELECT category, COUNT(*) AS n, SUM(amount) AS total FROM t GROUP BY category";
let df = plan_ctx.sql(query).await.expect("sql");
let plan = df.create_physical_plan().await.expect("physical plan");
let staged = build_distributed_stages(plan)
.expect("build stages")
.expect("splittable");
let map_stage = staged.stages.first().expect("map stage");
let result_stage = staged.stages.get(1).expect("result stage");
let shuffle = map_stage.shuffle.as_ref().expect("map shuffles");
let store = Arc::new(TestShuffleStore::default());
let exec_ctx = SessionContext::new();
for (task_index, body) in map_stage.task_bodies.iter().enumerate() {
let reader: Arc<dyn ShufflePartitionReader> = Arc::new(Arc::clone(&store));
let (_, mut stream) =
execute_dfplan_body(body, &exec_ctx, Some(reader)).expect("map exec");
while let Some(batch) = futures::StreamExt::next(&mut stream).await {
let batch = batch.expect("map batch");
if batch.num_rows() == 0 {
continue;
}
for (bucket, part) in partition_batch_by_key(
&batch,
&shuffle.key_columns[0],
shuffle.num_output_partitions,
)
.into_iter()
.enumerate()
{
if part.num_rows() > 0 {
store.write(0, task_index, bucket, part);
}
}
}
}
let all_partitions: Vec<usize> = (0..result_stage.task_count()).collect();
let coalesced_body = dfplan_body_with_spec(
result_stage.task_bodies.first().expect("result body"),
&DfplanTaskSpec {
partitions: all_partitions,
map_range: None,
},
)
.expect("coalesce rewrite");
let reader: Arc<dyn ShufflePartitionReader> = Arc::new(Arc::clone(&store));
let (_, stream) =
execute_dfplan_body(&coalesced_body, &exec_ctx, Some(reader)).expect("coalesced exec");
let coalesced: Vec<RecordBatch> = futures::TryStreamExt::try_collect(stream)
.await
.expect("coalesced results");
let mut baseline = Vec::new();
for body in &result_stage.task_bodies {
let reader: Arc<dyn ShufflePartitionReader> = Arc::new(Arc::clone(&store));
let (_, stream) =
execute_dfplan_body(body, &exec_ctx, Some(reader)).expect("baseline exec");
let batches: Vec<RecordBatch> = futures::TryStreamExt::try_collect(stream)
.await
.expect("baseline results");
baseline.extend(batches);
}
let render = |batches: &[RecordBatch]| {
let mut rows: Vec<String> = batches
.iter()
.flat_map(|b| {
(0..b.num_rows()).map(move |r| {
(0..b.num_columns())
.map(|c| {
arrow::util::display::array_value_to_string(b.column(c), r)
.expect("cell")
})
.collect::<Vec<_>>()
.join("|")
})
})
.collect();
rows.sort();
rows
};
assert_eq!(
render(&coalesced),
render(&baseline),
"coalesced task must produce the same union as per-partition tasks"
);
assert!(!coalesced.is_empty(), "group-by must produce rows");
}
#[tokio::test]
async fn skew_split_result_tasks_match_unsplit_execution() {
let tmp = tempfile::tempdir().expect("tempdir");
let path = write_test_parquet(tmp.path()).await;
let mut config = SessionConfig::new().with_target_partitions(4);
config
.options_mut()
.optimizer
.enable_round_robin_repartition = false;
config
.options_mut()
.optimizer
.hash_join_single_partition_threshold = 0;
config
.options_mut()
.optimizer
.hash_join_single_partition_threshold_rows = 0;
let plan_ctx = SessionContext::new_with_config(config);
for name in ["a", "b"] {
plan_ctx
.register_parquet(
name,
path.to_str().expect("utf8 path"),
datafusion::prelude::ParquetReadOptions::default(),
)
.await
.expect("register parquet");
}
let query = "SELECT a.id, a.category, b.amount FROM a JOIN b ON a.id = b.id";
let df = plan_ctx.sql(query).await.expect("sql");
let plan = df.create_physical_plan().await.expect("physical plan");
let staged = build_distributed_stages(plan)
.expect("build stages")
.expect("partitioned join must split");
let result_stage = staged.stages.last().expect("result stage");
assert!(result_stage.shuffle.is_none());
let result_body = result_stage.task_bodies.first().expect("result body");
assert!(
dfplan_body_is_split_safe(result_body),
"pure inner join result stage must be split-safe"
);
let store = Arc::new(TestShuffleStore::default());
let exec_ctx = SessionContext::new();
let mut probe_map_tasks = 0usize;
for (stage_index, stage) in staged.stages.iter().enumerate() {
let Some(shuffle) = &stage.shuffle else {
continue;
};
if stage_index == 0 {
probe_map_tasks = stage.task_count();
}
for (task_index, body) in stage.task_bodies.iter().enumerate() {
let reader: Arc<dyn ShufflePartitionReader> = Arc::new(Arc::clone(&store));
let (_, stream) =
execute_dfplan_body(body, &exec_ctx, Some(reader)).expect("map exec");
let batches: Vec<RecordBatch> = futures::TryStreamExt::try_collect(stream)
.await
.expect("map results");
for batch in batches {
if batch.num_rows() == 0 {
continue;
}
for (bucket, part) in partition_batch_by_key(
&batch,
&shuffle.key_columns[0],
shuffle.num_output_partitions,
)
.into_iter()
.enumerate()
{
if part.num_rows() > 0 {
store.write(stage_index, task_index, bucket, part);
}
}
}
}
}
assert!(
probe_map_tasks >= 2,
"need >=2 map tasks to split, got {probe_map_tasks}"
);
let collect_body = |body: String| {
let store = Arc::clone(&store);
let exec_ctx = exec_ctx.clone();
async move {
let reader: Arc<dyn ShufflePartitionReader> = Arc::new(store);
let (_, stream) =
execute_dfplan_body(&body, &exec_ctx, Some(reader)).expect("exec");
let batches: Vec<RecordBatch> = futures::TryStreamExt::try_collect(stream)
.await
.expect("results");
batches
}
};
let render = |batches: &[RecordBatch]| {
let mut rows: Vec<String> = batches
.iter()
.flat_map(|b| {
(0..b.num_rows()).map(move |r| {
(0..b.num_columns())
.map(|c| {
arrow::util::display::array_value_to_string(b.column(c), r)
.expect("cell")
})
.collect::<Vec<_>>()
.join("|")
})
})
.collect();
rows.sort();
rows
};
for (partition, body) in result_stage.task_bodies.iter().enumerate() {
let baseline = collect_body(body.clone()).await;
let mid = probe_map_tasks / 2;
let mut split_union = Vec::new();
for (start, end) in [(0, mid), (mid, probe_map_tasks)] {
let split_body = dfplan_body_with_spec(
body,
&DfplanTaskSpec {
partitions: vec![partition],
map_range: Some(DfplanMapRange {
upstream_stage_index: 0,
start,
end,
}),
},
)
.expect("split rewrite");
split_union.extend(collect_body(split_body).await);
}
assert_eq!(
render(&split_union),
render(&baseline),
"partition {partition}: split union must equal unsplit output"
);
}
let agg_ctx = planning_session_context(4);
agg_ctx
.register_parquet(
"t",
path.to_str().expect("utf8 path"),
datafusion::prelude::ParquetReadOptions::default(),
)
.await
.expect("register parquet");
let agg_plan = agg_ctx
.sql("SELECT category, COUNT(*) FROM t GROUP BY category")
.await
.expect("sql")
.create_physical_plan()
.await
.expect("plan");
let agg_staged = build_distributed_stages(agg_plan)
.expect("build stages")
.expect("splittable");
let agg_body = agg_staged
.stages
.last()
.expect("result stage")
.task_bodies
.first()
.expect("body");
assert!(
!dfplan_body_is_split_safe(agg_body),
"final aggregation must NOT be split-safe"
);
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod roundtrip_schema_guard_tests {
use super::*;
use arrow::datatypes::{DataType, Field, Schema};
fn alien_schema() -> Schema {
Schema::new(vec![Field::new(
"not_a_real_column",
DataType::Boolean,
true,
)])
}
#[tokio::test]
async fn the_guard_rejects_a_decode_whose_schema_differs() {
let ctx = fragment_decode_session_context();
ctx.sql("CREATE TABLE t(a INT) AS VALUES (1), (2)")
.await
.unwrap()
.collect()
.await
.unwrap();
let plan = ctx
.sql("SELECT a FROM t")
.await
.unwrap()
.create_physical_plan()
.await
.unwrap();
let codec = KrishivPhysicalCodec::coordinator();
let bytes = encode_dfplan_bytes(Arc::clone(&plan), &codec).unwrap();
let task_ctx = ctx.task_ctx();
verify_dfplan_roundtrip(&bytes, &codec, &task_ctx, Some(&plan))
.expect("a plan must round-trip against itself");
let alien: Arc<dyn ExecutionPlan> = Arc::new(
datafusion::physical_plan::empty::EmptyExec::new(Arc::new(alien_schema())),
);
let err = verify_dfplan_roundtrip(&bytes, &codec, &task_ctx, Some(&alien))
.expect_err("a schema disagreement must be refused");
let msg = err.to_string();
assert!(
msg.contains("decoded plan differs"),
"unexpected message: {msg}"
);
}
#[tokio::test]
async fn the_guard_compares_the_whole_tree_not_just_the_root() {
use arrow::datatypes::{DataType, Field, Schema};
use datafusion::physical_plan::empty::EmptyExec;
let same_root = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)]));
let original: Arc<dyn ExecutionPlan> =
Arc::new(datafusion::physical_plan::limit::GlobalLimitExec::new(
Arc::new(EmptyExec::new(Arc::clone(&same_root))),
0,
None,
));
let decoded: Arc<dyn ExecutionPlan> =
Arc::new(datafusion::physical_plan::limit::GlobalLimitExec::new(
Arc::new(EmptyExec::new(Arc::new(Schema::new(vec![Field::new(
"a",
DataType::Int64,
true,
)])))),
0,
None,
));
let difference = first_schema_difference(&original, &decoded, "root")
.expect("an interior disagreement must be reported");
assert!(
difference.contains("root"),
"the difference must name where it is: {difference}"
);
assert!(first_schema_difference(&original, &original, "root").is_none());
}
#[tokio::test]
async fn passing_no_expected_schema_keeps_the_old_decode_only_behaviour() {
let ctx = fragment_decode_session_context();
ctx.sql("CREATE TABLE t2(a INT) AS VALUES (1)")
.await
.unwrap()
.collect()
.await
.unwrap();
let plan = ctx
.sql("SELECT a FROM t2")
.await
.unwrap()
.create_physical_plan()
.await
.unwrap();
let codec = KrishivPhysicalCodec::coordinator();
let bytes = encode_dfplan_bytes(plan, &codec).unwrap();
verify_dfplan_roundtrip(&bytes, &codec, &ctx.task_ctx(), None)
.expect("decode-only checking must still pass");
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod staged_tpch_tests {
use super::*;
use arrow::record_batch::RecordBatch;
use datafusion::prelude::{ParquetReadOptions, SessionContext};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
const Q17: &str = "SELECT sum(l_extendedprice) / 7.0 AS avg_yearly FROM lineitem, part \
WHERE p_partkey = l_partkey AND p_brand = 'Brand#23' AND p_container = 'MED BOX' \
AND l_quantity < (SELECT 0.2 * avg(l_quantity) FROM lineitem \
WHERE l_partkey = p_partkey)";
const Q19: &str = "SELECT sum(l_extendedprice * (1 - l_discount)) AS revenue \
FROM lineitem, part \
WHERE (p_partkey = l_partkey AND p_brand = 'Brand#12' \
AND p_container IN ('SM CASE', 'SM BOX', 'SM PACK', 'SM PKG') \
AND l_quantity >= 1 AND l_quantity <= 11 AND p_size BETWEEN 1 AND 5 \
AND l_shipmode IN ('AIR', 'AIR REG') AND l_shipinstruct = 'DELIVER IN PERSON') \
OR (p_partkey = l_partkey AND p_brand = 'Brand#23' \
AND p_container IN ('MED BAG', 'MED BOX', 'MED PKG', 'MED PACK') \
AND l_quantity >= 10 AND l_quantity <= 20 AND p_size BETWEEN 1 AND 10 \
AND l_shipmode IN ('AIR', 'AIR REG') AND l_shipinstruct = 'DELIVER IN PERSON') \
OR (p_partkey = l_partkey AND p_brand = 'Brand#34' \
AND p_container IN ('LG CASE', 'LG BOX', 'LG PACK', 'LG PKG') \
AND l_quantity >= 20 AND l_quantity <= 30 AND p_size BETWEEN 1 AND 15 \
AND l_shipmode IN ('AIR', 'AIR REG') AND l_shipinstruct = 'DELIVER IN PERSON')";
#[derive(Debug, Default)]
struct StageStore {
partitions: Mutex<HashMap<(usize, usize, usize), Vec<RecordBatch>>>,
}
impl ShufflePartitionReader for Arc<StageStore> {
fn open_partition(
&self,
upstream_stage_index: usize,
map_task_index: usize,
partition: usize,
) -> futures::future::BoxFuture<'static, Result<ShuffleFragmentStream, String>> {
let batches = self
.partitions
.lock()
.expect("store lock")
.get(&(upstream_stage_index, map_task_index, partition))
.cloned()
.unwrap_or_default();
Box::pin(async move {
Ok(Box::pin(futures::stream::iter(batches.into_iter().map(Ok)))
as ShuffleFragmentStream)
})
}
}
fn write_parquet(path: &std::path::Path, batch: &RecordBatch) {
let file = std::fs::File::create(path).expect("create parquet");
let mut writer =
datafusion::parquet::arrow::ArrowWriter::try_new(file, batch.schema(), None)
.expect("writer init");
writer.write(batch).expect("write batch");
writer.close().expect("close writer");
}
fn write_tpch_fixture(dir: &std::path::Path) -> (std::path::PathBuf, std::path::PathBuf) {
use arrow::array::{Decimal128Array, Int32Array, Int64Array, StringArray};
use arrow::datatypes::{DataType, Field, Schema};
let lineitem_schema = Arc::new(Schema::new(vec![
Field::new("l_partkey", DataType::Int64, false),
Field::new("l_quantity", DataType::Decimal128(15, 2), false),
Field::new("l_extendedprice", DataType::Decimal128(15, 2), false),
Field::new("l_discount", DataType::Decimal128(15, 2), false),
Field::new("l_shipmode", DataType::Utf8, false),
Field::new("l_shipinstruct", DataType::Utf8, false),
]));
let part_schema = Arc::new(Schema::new(vec![
Field::new("p_partkey", DataType::Int64, false),
Field::new("p_brand", DataType::Utf8, false),
Field::new("p_container", DataType::Utf8, false),
Field::new("p_size", DataType::Int32, false),
]));
let money = |values: Vec<i128>| -> Arc<dyn arrow::array::Array> {
Arc::new(
Decimal128Array::from(values)
.with_precision_and_scale(15, 2)
.expect("decimal(15,2)"),
)
};
let lineitem_dir = dir.join("lineitem");
std::fs::create_dir_all(&lineitem_dir).expect("lineitem dir");
for file_index in 0..2i64 {
let keys: Vec<i64> = (0..200).map(|i| (file_index * 200 + i) % 60).collect();
let batch = RecordBatch::try_new(
Arc::clone(&lineitem_schema),
vec![
Arc::new(Int64Array::from(keys.clone())),
money(keys.iter().map(|k| i128::from(k % 30 + 1) * 100).collect()),
money(keys.iter().map(|k| i128::from(k + 1) * 1_000).collect()),
money(keys.iter().map(|k| i128::from(k % 10)).collect()),
Arc::new(StringArray::from(
keys.iter()
.map(|k| if k % 2 == 0 { "AIR" } else { "RAIL" })
.collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
keys.iter()
.map(|k| {
if k % 3 == 0 {
"DELIVER IN PERSON"
} else {
"TAKE BACK RETURN"
}
})
.collect::<Vec<_>>(),
)),
],
)
.expect("lineitem batch");
write_parquet(
&lineitem_dir.join(format!("l-{file_index}.parquet")),
&batch,
);
}
let part_dir = dir.join("part");
std::fs::create_dir_all(&part_dir).expect("part dir");
for file_index in 0..2i64 {
let keys: Vec<i64> = (0..30).map(|i| file_index * 30 + i).collect();
let batch = RecordBatch::try_new(
Arc::clone(&part_schema),
vec![
Arc::new(Int64Array::from(keys.clone())),
Arc::new(StringArray::from(
keys.iter()
.map(|k| match k % 3 {
0 => "Brand#12",
1 => "Brand#23",
_ => "Brand#34",
})
.collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
keys.iter()
.map(|k| match k % 4 {
0 => "SM BOX",
1 => "MED BOX",
2 => "LG BOX",
_ => "JUMBO BOX",
})
.collect::<Vec<_>>(),
)),
Arc::new(Int32Array::from(
keys.iter().map(|k| (k % 15 + 1) as i32).collect::<Vec<_>>(),
)),
],
)
.expect("part batch");
write_parquet(&part_dir.join(format!("p-{file_index}.parquet")), &batch);
}
(lineitem_dir, part_dir)
}
const Q22: &str = "SELECT cntrycode, count(*) AS numcust, sum(c_acctbal) AS totacctbal FROM ( \
SELECT substr(c_phone, 1, 2) AS cntrycode, c_acctbal FROM customer \
WHERE substr(c_phone, 1, 2) IN ('13','31','23','29','30','18','17') \
AND c_acctbal > (SELECT avg(c_acctbal) FROM customer \
WHERE c_acctbal > 0.00 \
AND substr(c_phone, 1, 2) IN ('13','31','23','29','30','18','17')) \
AND NOT EXISTS (SELECT * FROM orders WHERE o_custkey = c_custkey)) AS custsale \
GROUP BY cntrycode ORDER BY cntrycode";
fn write_q22_fixture(dir: &std::path::Path) -> (std::path::PathBuf, std::path::PathBuf) {
use arrow::array::{Decimal128Array, Int64Array, StringArray};
use arrow::datatypes::{DataType, Field, Schema};
let customer_schema = Arc::new(Schema::new(vec![
Field::new("c_custkey", DataType::Int64, false),
Field::new("c_phone", DataType::Utf8, false),
Field::new("c_acctbal", DataType::Decimal128(15, 2), false),
]));
let orders_schema = Arc::new(Schema::new(vec![
Field::new("o_orderkey", DataType::Int64, false),
Field::new("o_custkey", DataType::Int64, false),
]));
let codes = ["13", "31", "23", "29", "30", "18", "17", "44"];
let customer_dir = dir.join("customer");
std::fs::create_dir_all(&customer_dir).expect("customer dir");
for file_index in 0..2i64 {
let keys: Vec<i64> = (0..120).map(|i| file_index * 120 + i).collect();
let batch = RecordBatch::try_new(
Arc::clone(&customer_schema),
vec![
Arc::new(Int64Array::from(keys.clone())),
Arc::new(StringArray::from(
keys.iter()
.map(|k| format!("{}-555-0100", codes[(*k as usize) % codes.len()]))
.collect::<Vec<_>>(),
)),
Arc::new(
Decimal128Array::from(
keys.iter()
.map(|k| i128::from(k % 900) * 100)
.collect::<Vec<_>>(),
)
.with_precision_and_scale(15, 2)
.expect("decimal(15,2)"),
),
],
)
.expect("customer batch");
write_parquet(
&customer_dir.join(format!("c-{file_index}.parquet")),
&batch,
);
}
let orders_dir = dir.join("orders");
std::fs::create_dir_all(&orders_dir).expect("orders dir");
for file_index in 0..2i64 {
let keys: Vec<i64> = (0..80).map(|i| file_index * 80 + i).collect();
let batch = RecordBatch::try_new(
Arc::clone(&orders_schema),
vec![
Arc::new(Int64Array::from(keys.clone())),
Arc::new(Int64Array::from(
keys.iter().map(|k| k * 3 % 240).collect::<Vec<_>>(),
)),
],
)
.expect("orders batch");
write_parquet(&orders_dir.join(format!("o-{file_index}.parquet")), &batch);
}
(customer_dir, orders_dir)
}
async fn q22_context(dir: &std::path::Path) -> SessionContext {
q22_context_with_broadcast(dir, None).await
}
async fn q22_context_with_broadcast(
dir: &std::path::Path,
broadcast_bytes: Option<usize>,
) -> SessionContext {
let (customer, orders) = write_q22_fixture(dir);
let ctx = planning_session_context_with_options(4, None, broadcast_bytes);
for (name, path) in [("customer", customer), ("orders", orders)] {
ctx.register_parquet(
name,
path.to_str().expect("utf8 path"),
ParquetReadOptions::default(),
)
.await
.expect("register parquet");
}
ctx
}
#[tokio::test]
async fn a_severed_scalar_subquery_stage_does_not_decode_until_the_wrapper_is_restored() {
let tmp = tempfile::tempdir().expect("tempdir");
let ctx = q22_context(tmp.path()).await;
let plan = ctx
.sql(Q22)
.await
.expect("sql")
.create_physical_plan()
.await
.expect("physical plan");
let mut drafts: Vec<StageDraft> = Vec::new();
let root = cut_exchanges(Arc::clone(&plan), &mut drafts)
.unwrap_or_else(|Unsupported(reason)| panic!("q22 stage split: {reason}"));
drafts.push(StageDraft {
plan: root,
shuffle: None,
subqueries: None,
});
assert!(
drafts.iter().any(|d| d.subqueries.is_some()),
"q22 must cut at least one stage out from beneath the ScalarSubqueryExec, \
or there is nothing for the repair to act on"
);
let codec = KrishivPhysicalCodec::coordinator();
let decode_ctx = fragment_decode_session_context().task_ctx();
let mut saw_severed_stage = false;
for draft in &drafts {
let Some(context) = &draft.subqueries else {
continue;
};
let bytes =
encode_dfplan_bytes(Arc::clone(&draft.plan), &codec).expect("q22 stage encodes");
let Err(error) =
verify_dfplan_roundtrip(&bytes, &codec, &decode_ctx, Some(&draft.plan))
else {
continue;
};
assert!(
error
.to_string()
.contains("ScalarSubqueryExpr can only be deserialized"),
"expected the severed-wrapper decode failure, got: {error}"
);
saw_severed_stage = true;
let repaired = wrap_in_scalar_subquery_exec(Arc::clone(&draft.plan), context);
let bytes =
encode_dfplan_bytes(Arc::clone(&repaired), &codec).expect("repaired stage encodes");
verify_dfplan_roundtrip(&bytes, &codec, &decode_ctx, Some(&repaired))
.expect("restoring the wrapper must make the fragment decodable");
}
assert!(
saw_severed_stage,
"precondition: a q22 stage must actually fail to decode bare, or this \
test proves nothing about the repair"
);
}
#[tokio::test]
async fn q22_distributes_instead_of_running_as_a_single_task() {
let tmp = tempfile::tempdir().expect("tempdir");
let ctx = q22_context(tmp.path()).await;
let plan = ctx
.sql(Q22)
.await
.expect("sql")
.create_physical_plan()
.await
.expect("physical plan");
let staged = build_distributed_stages(plan)
.expect("build stages")
.expect("q22 must stage: a severed scalar-subquery wrapper is repaired, not declined");
assert!(
staged.stages.len() >= 2,
"expected a map stage and a result stage, got {}",
staged.stages.len()
);
assert!(
staged.stages.iter().any(|s| s.task_count() > 1),
"some stage must run more than one task, or 'distributed' means nothing: {:?}",
staged
.stages
.iter()
.map(DistributedStage::task_count)
.collect::<Vec<_>>()
);
}
#[tokio::test]
async fn an_unsplittable_broadcast_join_is_detected_and_converted() {
use datafusion::logical_expr::JoinType;
use datafusion::physical_plan::joins::{HashJoinExec, PartitionMode};
let tmp = tempfile::tempdir().expect("tempdir");
let ctx = q22_context(tmp.path()).await;
let scan = |sql: &'static str| {
let ctx = ctx.clone();
async move {
ctx.sql(sql)
.await
.expect("sql")
.create_physical_plan()
.await
.expect("physical plan")
}
};
let build = scan("SELECT c_custkey FROM customer").await;
let probe = scan("SELECT o_custkey FROM orders").await;
assert!(
probe.output_partitioning().partition_count() > 1,
"precondition: the probe side must have several partitions, or there \
is no rendezvous to miss"
);
let on = vec![(
datafusion::physical_plan::expressions::col("c_custkey", &build.schema())
.expect("build key"),
datafusion::physical_plan::expressions::col("o_custkey", &probe.schema())
.expect("probe key"),
)];
let unsafe_join: Arc<dyn ExecutionPlan> = Arc::new(
HashJoinExec::try_new(
Arc::new(CoalescePartitionsExec::new(build)),
probe,
on,
None,
&JoinType::LeftAnti,
None,
PartitionMode::CollectLeft,
datafusion::common::NullEquality::NullEqualsNothing,
false,
)
.expect("hand-built broadcast anti-join"),
);
let join_ref = unsafe_join
.downcast_ref::<HashJoinExec>()
.expect("hash join");
assert!(
is_unsplittable_broadcast_join(join_ref),
"a CollectLeft LeftAnti join over a multi-partition probe must be \
recognised as unsplittable"
);
assert!(
find_unsupported_stage_node(&unsafe_join).is_some(),
"and the stage guard must refuse it, so it can never ship unconverted"
);
let converted = redistribute_unsplittable_broadcast_joins(Arc::clone(&unsafe_join))
.expect("conversion must succeed");
let converted_join = converted
.downcast_ref::<HashJoinExec>()
.expect("still a hash join");
assert_eq!(
*converted_join.partition_mode(),
PartitionMode::Partitioned,
"conversion must switch to the mode whose probe counter is per-task"
);
assert!(
!is_unsplittable_broadcast_join(converted_join),
"the converted join must no longer be unsplittable"
);
assert_eq!(
*converted_join.join_type(),
JoinType::LeftAnti,
"conversion must not change the join's meaning"
);
assert_eq!(
converted.schema(),
unsafe_join.schema(),
"conversion must preserve the join's output schema"
);
}
async fn collect_left_over(
ctx: &SessionContext,
build_sql: &str,
build_key: &str,
null_aware: bool,
join_type: datafusion::logical_expr::JoinType,
) -> Arc<dyn ExecutionPlan> {
use datafusion::physical_plan::joins::{HashJoinExec, PartitionMode};
let plan = |sql: String| {
let ctx = ctx.clone();
async move {
ctx.sql(&sql)
.await
.expect("sql")
.create_physical_plan()
.await
.expect("physical plan")
}
};
let build = plan(build_sql.to_owned()).await;
let probe = plan(String::from("SELECT o_custkey FROM orders")).await;
let on = vec![(
datafusion::physical_plan::expressions::col(build_key, &build.schema())
.expect("build key"),
datafusion::physical_plan::expressions::col("o_custkey", &probe.schema())
.expect("probe key"),
)];
Arc::new(
HashJoinExec::try_new(
Arc::new(CoalescePartitionsExec::new(build)),
probe,
on,
None,
&join_type,
None,
PartitionMode::CollectLeft,
datafusion::common::NullEquality::NullEqualsNothing,
null_aware,
)
.expect("hand-built broadcast join"),
)
}
#[tokio::test]
async fn a_zero_estimate_is_not_proof_that_a_build_side_is_small() {
use datafusion::physical_plan::joins::{HashJoinExec, PartitionMode};
let tmp = tempfile::tempdir().expect("tempdir");
let ctx = q22_context(tmp.path()).await;
let join = collect_left_over(
&ctx,
"SELECT c_custkey FROM customer WHERE 1 = 0",
"c_custkey",
false,
datafusion::logical_expr::JoinType::Inner,
)
.await;
let join_ref = join.downcast_ref::<HashJoinExec>().expect("hash join");
assert!(
broadcast_build_estimate_is_empty(join_ref),
"a zero estimate is the estimator giving up, not a measurement"
);
assert!(
is_degenerate_broadcast_join(join_ref),
"so the join must be recognised as one that should not broadcast"
);
assert!(
!is_unsplittable_broadcast_join(join_ref),
"this is a throughput problem, not a correctness one"
);
assert!(
find_unsupported_stage_node(&join).is_none(),
"and the stage guard must not refuse it"
);
let converted =
redistribute_unsplittable_broadcast_joins(Arc::clone(&join)).expect("conversion");
let converted_join = converted
.downcast_ref::<HashJoinExec>()
.expect("still a hash join");
assert_eq!(
*converted_join.partition_mode(),
PartitionMode::Partitioned,
"the build side must be hash-partitioned instead of gathered"
);
assert_eq!(
converted.schema(),
join.schema(),
"conversion must preserve the join's output schema"
);
}
#[tokio::test]
async fn the_reducer_lands_on_the_scan_that_owns_the_key() {
use datafusion::physical_plan::joins::{HashJoinExec, PartitionMode};
let tmp = tempfile::tempdir().expect("tempdir");
let ctx = q22_context(tmp.path()).await;
let plan = |sql: String| {
let ctx = ctx.clone();
async move {
ctx.sql(&sql)
.await
.expect("sql")
.create_physical_plan()
.await
.expect("plan")
}
};
let customer = plan(String::from("SELECT c_custkey, c_phone FROM customer")).await;
let orders = plan(String::from("SELECT o_custkey FROM orders")).await;
let on_fact = vec![(
datafusion::physical_plan::expressions::col("c_custkey", &customer.schema())
.expect("c_custkey"),
datafusion::physical_plan::expressions::col("o_custkey", &orders.schema())
.expect("o_custkey"),
)];
let fact: Arc<dyn ExecutionPlan> = Arc::new(
HashJoinExec::try_new(
Arc::new(CoalescePartitionsExec::new(customer)),
orders,
on_fact,
None,
&datafusion::logical_expr::JoinType::Inner,
None,
PartitionMode::CollectLeft,
datafusion::common::NullEquality::NullEqualsNothing,
false,
)
.expect("fact join"),
);
let dimension = plan(String::from(
"SELECT c_phone FROM customer WHERE c_custkey = 1",
))
.await;
let on_top = vec![(
datafusion::physical_plan::expressions::col("c_phone", &dimension.schema())
.expect("dim key"),
datafusion::physical_plan::expressions::col("c_phone", &fact.schema())
.expect("fact key"),
)];
let top: Arc<dyn ExecutionPlan> = Arc::new(
HashJoinExec::try_new(
Arc::clone(&dimension),
Arc::clone(&fact),
on_top,
None,
&datafusion::logical_expr::JoinType::Inner,
None,
PartitionMode::CollectLeft,
datafusion::common::NullEquality::NullEqualsNothing,
false,
)
.expect("top join"),
);
let before = datafusion::physical_plan::displayable(top.as_ref())
.indent(true)
.to_string();
assert_eq!(
before.matches("RightSemi").count(),
0,
"precondition: nothing reduced yet"
);
let after_plan = reduce_by_broadcast_dimension_for_test(Arc::clone(&top))
.expect("rewrite must not fail");
let after = datafusion::physical_plan::displayable(after_plan.as_ref())
.indent(true)
.to_string();
assert_eq!(
after.matches("RightSemi").count(),
1,
"exactly one reducer, attached once:\n{after}"
);
let semi = after
.lines()
.position(|l| l.contains("RightSemi"))
.expect("reducer");
let fact_join = after
.lines()
.position(|l| l.contains("on=[(c_custkey"))
.expect("fact join");
assert!(
semi > fact_join,
"the reducer must be below the fact join, not above it:\n{after}"
);
assert_eq!(
after_plan.schema(),
top.schema(),
"the rewrite must preserve the plan's schema exactly"
);
}
#[tokio::test]
async fn a_rescue_never_narrows_to_the_probe_sides_partition_count() {
use datafusion::physical_plan::joins::{HashJoinExec, PartitionMode};
let tmp = tempfile::tempdir().expect("tempdir");
let ctx = q22_context(tmp.path()).await;
let scan = ctx
.sql("SELECT c_custkey FROM customer WHERE 1 = 0")
.await
.expect("sql")
.create_physical_plan()
.await
.expect("physical plan");
let build_key =
datafusion::physical_plan::expressions::col("c_custkey", &scan.schema()).expect("col");
let build: Arc<dyn ExecutionPlan> = Arc::new(
RepartitionExec::try_new(scan, Partitioning::Hash(vec![build_key], 4))
.expect("hash exchange"),
);
let probe = ctx
.sql("SELECT o_custkey FROM orders LIMIT 5")
.await
.expect("sql")
.create_physical_plan()
.await
.expect("physical plan");
let build_partitions = build.output_partitioning().partition_count();
assert_eq!(
probe.output_partitioning().partition_count(),
1,
"precondition: the probe must be single-partition or this tests nothing"
);
assert!(
build_partitions > 1,
"precondition: the build side must have parallelism to lose, got {build_partitions}"
);
let on = vec![(
datafusion::physical_plan::expressions::col("c_custkey", &build.schema())
.expect("build key"),
datafusion::physical_plan::expressions::col("o_custkey", &probe.schema())
.expect("probe key"),
)];
let join: Arc<dyn ExecutionPlan> = Arc::new(
HashJoinExec::try_new(
Arc::new(CoalescePartitionsExec::new(build)),
probe,
on,
None,
&datafusion::logical_expr::JoinType::Inner,
None,
PartitionMode::CollectLeft,
datafusion::common::NullEquality::NullEqualsNothing,
false,
)
.expect("hand-built broadcast join"),
);
assert!(
is_degenerate_broadcast_join(join.downcast_ref::<HashJoinExec>().expect("hash join")),
"precondition: the zero estimate must make this a rescue candidate"
);
let converted =
redistribute_unsplittable_broadcast_joins(Arc::clone(&join)).expect("conversion");
let converted_join = converted
.downcast_ref::<HashJoinExec>()
.expect("still a hash join");
assert_eq!(*converted_join.partition_mode(), PartitionMode::Partitioned);
assert_eq!(
converted_join.left().output_partitioning().partition_count(),
build_partitions,
"the rescue must keep the build side's parallelism, not collapse to the probe's 1"
);
assert_eq!(
converted_join
.right()
.output_partitioning()
.partition_count(),
build_partitions,
"and both sides must agree, as PartitionMode::Partitioned requires"
);
assert_eq!(
converted.schema(),
join.schema(),
"conversion must preserve the join's output schema"
);
}
#[tokio::test]
async fn a_provably_small_build_side_is_still_broadcast() {
use datafusion::physical_plan::joins::{HashJoinExec, PartitionMode};
let tmp = tempfile::tempdir().expect("tempdir");
let ctx = q22_context(tmp.path()).await;
let join = collect_left_over(
&ctx,
"SELECT c_custkey FROM customer",
"c_custkey",
false,
datafusion::logical_expr::JoinType::Inner,
)
.await;
let join_ref = join.downcast_ref::<HashJoinExec>().expect("hash join");
assert!(
!broadcast_build_estimate_is_empty(join_ref),
"a non-empty parquet scan must report a positive estimate"
);
assert!(
!is_degenerate_broadcast_join(join_ref),
"so it must keep its broadcast"
);
let after =
redistribute_unsplittable_broadcast_joins(Arc::clone(&join)).expect("conversion");
assert_eq!(
*after
.downcast_ref::<HashJoinExec>()
.expect("still a hash join")
.partition_mode(),
PartitionMode::CollectLeft,
"the rule must leave a legitimately small broadcast alone"
);
}
#[tokio::test]
async fn a_large_but_positive_estimate_keeps_its_broadcast() {
use datafusion::physical_plan::joins::{HashJoinExec, PartitionMode};
let tmp = tempfile::tempdir().expect("tempdir");
let ctx = q22_context(tmp.path()).await;
let join = collect_left_over(
&ctx,
"SELECT a.c_custkey FROM customer a CROSS JOIN customer b",
"c_custkey",
false,
datafusion::logical_expr::JoinType::Inner,
)
.await;
let join_ref = join.downcast_ref::<HashJoinExec>().expect("hash join");
let stats = join_ref
.left()
.partition_statistics(None)
.expect("statistics");
assert!(
matches!(
stats.num_rows,
datafusion::common::stats::Precision::Exact(n)
| datafusion::common::stats::Precision::Inexact(n) if n > 0
),
"precondition: the build side must estimate a positive row count, \
got {:?}",
stats.num_rows
);
assert!(
!broadcast_build_estimate_is_empty(join_ref),
"a positive estimate is a measurement, however large"
);
assert!(
!is_degenerate_broadcast_join(join_ref),
"and must not be converted — overriding DataFusion's ceiling cost \
q8 4x and q9 2.5x"
);
let after =
redistribute_unsplittable_broadcast_joins(Arc::clone(&join)).expect("conversion");
assert_eq!(
*after
.downcast_ref::<HashJoinExec>()
.expect("still a hash join")
.partition_mode(),
PartitionMode::CollectLeft,
"the plan must come back untouched"
);
}
#[tokio::test]
async fn a_null_aware_anti_join_is_never_converted() {
use datafusion::physical_plan::joins::HashJoinExec;
let tmp = tempfile::tempdir().expect("tempdir");
let ctx = q22_context(tmp.path()).await;
let join = collect_left_over(
&ctx,
"SELECT c_custkey FROM customer WHERE 1 = 0",
"c_custkey",
true,
datafusion::logical_expr::JoinType::LeftAnti,
)
.await;
let join_ref = join.downcast_ref::<HashJoinExec>().expect("hash join");
assert!(
broadcast_build_estimate_is_empty(join_ref),
"precondition: its build-side estimate is degenerate, so only the \
null-aware check can be what spares it"
);
assert!(
!is_degenerate_broadcast_join(join_ref),
"a null-aware anti join must never be converted for throughput"
);
}
#[tokio::test]
async fn staged_q22_matches_direct_execution() {
let tmp = tempfile::tempdir().expect("tempdir");
let ctx = q22_context(tmp.path()).await;
let expected = render(&direct(&ctx, Q22).await);
let actual = run_staged(&ctx, Q22)
.await
.unwrap_or_else(|e| panic!("q22: staged execution failed: {e}"));
assert_eq!(
render(&actual),
expected,
"q22: staged result differs from single-node execution"
);
assert!(!expected.is_empty(), "the q22 fixture must produce rows");
}
async fn tpch_context_with_broadcast(
dir: &std::path::Path,
join_threshold: Option<u64>,
broadcast_bytes: Option<usize>,
) -> SessionContext {
let (lineitem, part) = write_tpch_fixture(dir);
let ctx = planning_session_context_with_options(4, join_threshold, broadcast_bytes);
for (name, path) in [("lineitem", lineitem), ("part", part)] {
ctx.register_parquet(
name,
path.to_str().expect("utf8 path"),
ParquetReadOptions::default(),
)
.await
.expect("register parquet");
}
ctx
}
fn route(batch: &RecordBatch, key_column: &str, num_partitions: usize) -> Vec<RecordBatch> {
use std::hash::{Hash as _, Hasher as _};
let key_idx = batch.schema().index_of(key_column).expect("key column");
let column = batch.column(key_idx);
let mut selections: Vec<Vec<u32>> = vec![Vec::new(); num_partitions];
for row in 0..batch.num_rows() {
let value = arrow::util::display::array_value_to_string(column, row).expect("value");
let mut hasher = std::collections::hash_map::DefaultHasher::new();
value.hash(&mut hasher);
let bucket = (hasher.finish() as usize) % num_partitions;
selections[bucket].push(row as u32);
}
selections
.into_iter()
.map(|rows| {
let indices = arrow::array::UInt32Array::from(rows);
arrow::compute::take_record_batch(batch, &indices).expect("take")
})
.collect()
}
async fn run_staged(ctx: &SessionContext, sql: &str) -> Result<Vec<RecordBatch>, String> {
let df = ctx.sql(sql).await.map_err(|e| e.to_string())?;
let plan = df.create_physical_plan().await.map_err(|e| e.to_string())?;
let staged = build_distributed_stages(plan)
.map_err(|e| e.to_string())?
.ok_or_else(|| String::from("declined to stage"))?;
let store = Arc::new(StageStore::default());
let exec_ctx = fragment_decode_session_context();
let mut result = Vec::new();
for (stage_index, stage) in staged.stages.iter().enumerate() {
for (task_index, body) in stage.task_bodies.iter().enumerate() {
let reader: Arc<dyn ShufflePartitionReader> = Arc::new(Arc::clone(&store));
let (declared, mut stream) = execute_dfplan_body(body, &exec_ctx, Some(reader))
.map_err(|e| format!("stage {stage_index} task {task_index} start: {e}"))?;
while let Some(batch) = futures::StreamExt::next(&mut stream).await {
let batch =
batch.map_err(|e| format!("stage {stage_index} task {task_index}: {e}"))?;
if batch.schema() != declared {
return Err(format!(
"stage {stage_index} task {task_index} declares {declared:?} but \
produced {:?}; ShuffleReadExec labels the reduce side with the \
declared schema, so this disagreement becomes a reduce-side Arrow \
error on a real cluster",
batch.schema()
));
}
if batch.num_rows() == 0 {
continue;
}
match &stage.shuffle {
None => result.push(batch),
Some(shuffle) => match shuffle.key_columns.first() {
Some(key) => {
for (bucket, part) in
route(&batch, key, shuffle.num_output_partitions)
.into_iter()
.enumerate()
{
if part.num_rows() > 0 {
store
.partitions
.lock()
.expect("store lock")
.entry((stage_index, task_index, bucket))
.or_default()
.push(part);
}
}
}
None => store
.partitions
.lock()
.expect("store lock")
.entry((stage_index, task_index, 0))
.or_default()
.push(batch),
},
}
}
}
}
Ok(result)
}
fn render(batches: &[RecordBatch]) -> Vec<String> {
let mut rows: Vec<String> = batches
.iter()
.flat_map(|b| {
(0..b.num_rows()).map(move |r| {
(0..b.num_columns())
.map(|c| {
arrow::util::display::array_value_to_string(b.column(c), r)
.expect("cell")
})
.collect::<Vec<_>>()
.join("|")
})
})
.collect();
rows.sort();
rows
}
async fn direct(ctx: &SessionContext, sql: &str) -> Vec<RecordBatch> {
ctx.sql(sql)
.await
.expect("sql")
.collect()
.await
.expect("direct execution")
}
async fn staged_matches_direct(sql: &str, join_threshold: Option<u64>, label: &str) {
staged_matches_direct_with_broadcast(sql, join_threshold, None, label).await;
}
async fn staged_matches_direct_with_broadcast(
sql: &str,
join_threshold: Option<u64>,
broadcast_bytes: Option<usize>,
label: &str,
) {
let tmp = tempfile::tempdir().expect("tempdir");
let ctx = tpch_context_with_broadcast(tmp.path(), join_threshold, broadcast_bytes).await;
let expected = render(&direct(&ctx, sql).await);
let actual = run_staged(&ctx, sql)
.await
.unwrap_or_else(|e| panic!("{label}: staged execution failed: {e}"));
assert_eq!(
render(&actual),
expected,
"{label}: staged result differs from single-node execution"
);
}
#[tokio::test]
async fn staged_q17_matches_direct_execution() {
staged_matches_direct(Q17, None, "q17/unconverted").await;
}
#[tokio::test]
async fn staged_q17_matches_direct_execution_with_converted_joins() {
staged_matches_direct(Q17, Some(0), "q17/converted").await;
}
#[tokio::test]
async fn staged_q19_matches_direct_execution() {
staged_matches_direct(Q19, None, "q19/unconverted").await;
}
#[tokio::test]
async fn staged_q19_matches_direct_execution_with_converted_joins() {
staged_matches_direct(Q19, Some(0), "q19/converted").await;
}
#[tokio::test]
async fn staged_q17_matches_direct_execution_without_broadcast() {
staged_matches_direct_with_broadcast(Q17, None, Some(0), "q17/no-broadcast").await;
}
#[tokio::test]
async fn staged_q19_matches_direct_execution_without_broadcast() {
staged_matches_direct_with_broadcast(Q19, None, Some(0), "q19/no-broadcast").await;
}
#[tokio::test]
async fn staged_q17_matches_direct_execution_converted_and_without_broadcast() {
staged_matches_direct_with_broadcast(Q17, Some(0), Some(0), "q17/converted+no-broadcast")
.await;
}
#[tokio::test]
async fn staged_q19_matches_direct_execution_converted_and_without_broadcast() {
staged_matches_direct_with_broadcast(Q19, Some(0), Some(0), "q19/converted+no-broadcast")
.await;
}
#[tokio::test]
async fn staged_q22_matches_direct_execution_without_broadcast() {
let tmp = tempfile::tempdir().expect("tempdir");
let ctx = q22_context_with_broadcast(tmp.path(), Some(0)).await;
let expected = render(&direct(&ctx, Q22).await);
let actual = run_staged(&ctx, Q22)
.await
.unwrap_or_else(|e| panic!("q22/no-broadcast: staged execution failed: {e}"));
assert_eq!(
render(&actual),
expected,
"q22/no-broadcast: staged result differs from single-node execution"
);
assert!(!expected.is_empty(), "the q22 fixture must produce rows");
}
#[tokio::test]
async fn a_final_avg_over_a_decimal_survives_the_fragment_round_trip() {
for sql in [
"SELECT avg(l_quantity) FROM lineitem",
"SELECT l_partkey, avg(l_quantity) FROM lineitem GROUP BY l_partkey",
"SELECT sum(l_extendedprice) / 7.0 FROM lineitem",
] {
let tmp = tempfile::tempdir().expect("tempdir");
let ctx = tpch_context_with_broadcast(tmp.path(), None, None).await;
let expected = render(&direct(&ctx, sql).await);
let actual = run_staged(&ctx, sql)
.await
.unwrap_or_else(|e| panic!("{sql}: staged execution failed: {e}"));
assert_eq!(
render(&actual),
expected,
"{sql}: staged result differs from single-node execution"
);
}
}
#[tokio::test]
async fn without_broadcast_a_reduce_stage_reads_two_upstream_stages() {
let tmp = tempfile::tempdir().expect("tempdir");
let ctx = tpch_context_with_broadcast(tmp.path(), None, Some(0)).await;
let plan = ctx
.sql(Q19)
.await
.expect("sql")
.create_physical_plan()
.await
.expect("physical plan");
let staged = build_distributed_stages(plan)
.expect("staging must not error")
.expect("q19 must stage");
let widest = staged
.stages
.iter()
.map(|stage| stage.upstream_stage_indexes.len())
.max()
.unwrap_or(0);
assert!(
widest >= 2,
"expected a stage reading 2+ upstream stages, widest was {widest}; \
the no-broadcast tests are not exercising the cluster's join shape"
);
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod codec_completeness_tests {
#[test]
fn every_custom_execution_plan_is_encodable_or_declared_local() {
const ENCODABLE: &[&str] = &[
"RuntimeFilterBuildExec",
"RuntimeFilterProbeExec",
"ShuffleReadExec",
];
const EXECUTION_LOCAL: &[&str] = &["GraceHashJoinExec", "OnceStreamExec"];
let mut found = Vec::new();
let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
let mut stack = vec![dir];
while let Some(path) = stack.pop() {
for entry in std::fs::read_dir(&path).expect("read src") {
let entry = entry.expect("dir entry").path();
if entry.is_dir() {
stack.push(entry);
continue;
}
if entry.extension().is_none_or(|e| e != "rs") {
continue;
}
let text = std::fs::read_to_string(&entry).expect("read file");
for line in text.lines() {
if let Some(rest) = line.trim().strip_prefix("impl ExecutionPlan for ") {
let name = rest
.trim_end_matches(" {")
.split(['<', ' '])
.next()
.unwrap_or(rest)
.to_string();
found.push(name);
}
}
}
}
found.sort();
found.dedup();
assert!(
!found.is_empty(),
"the scan found no ExecutionPlan impls at all"
);
let undeclared: Vec<&String> = found
.iter()
.filter(|n| !ENCODABLE.contains(&n.as_str()) && !EXECUTION_LOCAL.contains(&n.as_str()))
.collect();
assert!(
undeclared.is_empty(),
"custom ExecutionPlan(s) {undeclared:?} are neither encodable nor declared \
execution-local. If such a node can reach a stage plan, the coordinator will \
silently run the query as a SINGLE TASK. Add a codec arm, or confine it to \
post-decode and list it in EXECUTION_LOCAL."
);
}
#[test]
fn rebuilding_a_body_keeps_the_python_udf_directive() {
use super::{
DfplanTaskSpec, dfplan_body_partition_spec, dfplan_body_with_spec, is_dfplan_body,
};
let directive = "/* krishiv-register-python-udf:addk:int64:int64:QUJD */";
let body = format!("{directive}\ndfplan:v1:0:QUJD");
let spec = DfplanTaskSpec {
partitions: vec![3, 4],
map_range: None,
};
let rebuilt = dfplan_body_with_spec(&body, &spec).expect("rebuild");
assert!(
rebuilt.starts_with(directive),
"the rebuilt body must still carry the UDF directive: {rebuilt}"
);
assert!(
is_dfplan_body(&rebuilt),
"and must still parse as a dfplan body: {rebuilt}"
);
assert_eq!(
dfplan_body_partition_spec(&rebuilt)
.expect("spec")
.partitions,
vec![3, 4],
"the new partition spec must be the one asked for"
);
}
mod stage_reuse {
use super::super::*;
fn scan_schema() -> SchemaRef {
Arc::new(arrow::datatypes::Schema::new(vec![
arrow::datatypes::Field::new("k", arrow::datatypes::DataType::Int64, false),
arrow::datatypes::Field::new("v", arrow::datatypes::DataType::Int64, false),
]))
}
fn leaf_draft(rows: usize, key: &str, parts: usize) -> StageDraft {
use datafusion::catalog::memory::MemorySourceConfig;
use datafusion::datasource::source::DataSourceExec;
let schema = scan_schema();
let batches: Vec<Vec<arrow::record_batch::RecordBatch>> = vec![vec![
arrow::record_batch::RecordBatch::try_new(
Arc::clone(&schema),
vec![
Arc::new(arrow::array::Int64Array::from(
(0..rows as i64).collect::<Vec<_>>(),
)),
Arc::new(arrow::array::Int64Array::from(
(0..rows as i64).collect::<Vec<_>>(),
)),
],
)
.expect("batch"),
]];
let source = MemorySourceConfig::try_new(&batches, Arc::clone(&schema), None)
.expect("memory source");
StageDraft {
plan: Arc::new(DataSourceExec::new(Arc::new(source))),
shuffle: Some(StageShuffleOutput {
key_columns: vec![String::from(key)],
num_output_partitions: parts,
}),
subqueries: None,
}
}
fn reader(stage: usize) -> Arc<dyn ExecutionPlan> {
Arc::new(ShuffleReadExec::new(stage, 4, 4, scan_schema(), None))
}
#[test]
fn identical_leaf_stages_collapse_and_readers_are_repointed() {
let mut drafts = vec![leaf_draft(4, "k", 4), leaf_draft(4, "k", 4)];
let mut root: Arc<dyn ExecutionPlan> =
datafusion::physical_plan::union::UnionExec::try_new(vec![reader(0), reader(1)])
.expect("union");
let removed = dedupe_identical_stages_unconditionally(&mut root, &mut drafts);
assert_eq!(
removed, 1,
"one of the two identical stages must be removed"
);
assert_eq!(drafts.len(), 1, "one stage must survive");
let upstreams = collect_upstream_stage_indexes(&root);
assert_eq!(
upstreams,
vec![0],
"both readers must point at the surviving stage, got {upstreams:?}"
);
}
#[test]
fn different_shuffle_contracts_do_not_collapse() {
let mut drafts = vec![leaf_draft(4, "k", 4), leaf_draft(4, "k", 8)];
let mut root: Arc<dyn ExecutionPlan> =
datafusion::physical_plan::union::UnionExec::try_new(vec![reader(0), reader(1)])
.expect("union");
assert_eq!(
dedupe_identical_stages_unconditionally(&mut root, &mut drafts),
0,
"a different output-partition count is a different stage"
);
assert_eq!(drafts.len(), 2);
}
#[test]
fn different_shuffle_keys_do_not_collapse() {
let mut drafts = vec![leaf_draft(4, "k", 4), leaf_draft(4, "v", 4)];
let mut root: Arc<dyn ExecutionPlan> =
datafusion::physical_plan::union::UnionExec::try_new(vec![reader(0), reader(1)])
.expect("union");
assert_eq!(
dedupe_identical_stages_unconditionally(&mut root, &mut drafts),
0,
"a different partitioning key is a different stage"
);
}
#[test]
fn different_content_does_not_collapse() {
let mut drafts = vec![leaf_draft(4, "k", 4), leaf_draft(9, "k", 4)];
let mut root: Arc<dyn ExecutionPlan> =
datafusion::physical_plan::union::UnionExec::try_new(vec![reader(0), reader(1)])
.expect("union");
assert_eq!(
dedupe_identical_stages_unconditionally(&mut root, &mut drafts),
0,
"stages producing different rows must stay separate"
);
assert_eq!(drafts.len(), 2);
}
#[test]
fn non_leaf_stages_are_left_alone() {
let mk = || StageDraft {
plan: reader(7),
shuffle: Some(StageShuffleOutput {
key_columns: vec![String::from("k")],
num_output_partitions: 4,
}),
subqueries: None,
};
let mut drafts = vec![mk(), mk()];
let mut root: Arc<dyn ExecutionPlan> =
datafusion::physical_plan::union::UnionExec::try_new(vec![reader(0), reader(1)])
.expect("union");
assert_eq!(
dedupe_identical_stages_unconditionally(&mut root, &mut drafts),
0,
"stages that read a shuffle are not eligible for leaf reuse"
);
}
#[test]
fn three_identical_stages_collapse_to_one() {
let mut drafts = vec![
leaf_draft(4, "k", 4),
leaf_draft(4, "k", 4),
leaf_draft(4, "k", 4),
];
let mut root: Arc<dyn ExecutionPlan> =
datafusion::physical_plan::union::UnionExec::try_new(vec![
reader(0),
reader(1),
reader(2),
])
.expect("union");
assert_eq!(
dedupe_identical_stages_unconditionally(&mut root, &mut drafts),
2
);
assert_eq!(drafts.len(), 1);
assert_eq!(collect_upstream_stage_indexes(&root), vec![0]);
}
#[test]
fn survivor_indexes_are_compacted_correctly() {
let mut drafts = vec![
leaf_draft(4, "k", 4),
leaf_draft(7, "k", 4),
leaf_draft(4, "k", 4),
leaf_draft(9, "k", 4),
];
let mut root: Arc<dyn ExecutionPlan> =
datafusion::physical_plan::union::UnionExec::try_new(vec![
reader(1),
reader(2),
reader(3),
])
.expect("union");
assert_eq!(
dedupe_identical_stages_unconditionally(&mut root, &mut drafts),
1
);
assert_eq!(drafts.len(), 3, "A, B, C survive");
assert_eq!(collect_upstream_stage_indexes(&root), vec![0, 1, 2]);
}
#[test]
fn volatile_markers_block_reuse() {
assert!(
VOLATILE_MARKERS.contains(&"random("),
"random() must block reuse"
);
assert!(VOLATILE_MARKERS.contains(&"now("), "now() must block reuse");
assert!(
VOLATILE_MARKERS.contains(&"uuid("),
"uuid() must block reuse"
);
}
#[test]
fn reuse_is_off_by_default() {
let mut drafts = vec![leaf_draft(4, "k", 4), leaf_draft(4, "k", 4)];
let mut root: Arc<dyn ExecutionPlan> =
datafusion::physical_plan::union::UnionExec::try_new(vec![reader(0), reader(1)])
.expect("union");
if std::env::var(STAGE_REUSE_ENV).is_err() {
assert_eq!(
dedupe_identical_stages(&mut root, &mut drafts),
0,
"stage reuse must be off unless {STAGE_REUSE_ENV} is set"
);
}
}
}
mod runtime_filters {
use super::super::*;
use datafusion::common::{JoinType, NullEquality};
use datafusion::physical_expr::expressions::Column;
use datafusion::physical_plan::joins::{HashJoinExec, PartitionMode};
#[test]
fn a_sort_merge_join_across_stages_is_a_candidate() {
use datafusion::physical_expr::{LexOrdering, PhysicalSortExpr};
use datafusion::physical_plan::joins::SortMergeJoinExec;
let build = read(0, Some(1_000), "k", arrow::datatypes::DataType::Int64);
let probe = read(1, Some(10_000_000), "k", arrow::datatypes::DataType::Int64);
let on: Vec<(
Arc<dyn datafusion::physical_expr::PhysicalExpr>,
Arc<dyn datafusion::physical_expr::PhysicalExpr>,
)> = vec![(Arc::new(Column::new("k", 0)), Arc::new(Column::new("k", 0)))];
let sort_options = LexOrdering::new(vec![PhysicalSortExpr::new_default(Arc::new(
Column::new("k", 0),
))])
.expect("ordering")
.iter()
.map(|e| e.options)
.collect();
let smj = SortMergeJoinExec::try_new(
build,
probe,
on,
None,
JoinType::Inner,
sort_options,
NullEquality::NullEqualsNothing,
)
.expect("sort-merge join over two shuffle reads");
let plan: Arc<dyn ExecutionPlan> = Arc::new(smj);
let mut candidates = Vec::new();
let mut rejects = RuntimeFilterRejects::default();
collect_runtime_filter_candidates(&plan, &mut candidates, &mut rejects);
assert_eq!(
rejects.joins, 1,
"the sort-merge join must now be inspected like any other equijoin"
);
assert_eq!(
rejects.joins_of_unsupported_kind, 0,
"and must no longer be written off as unreadable"
);
assert_eq!(
candidates.len(),
1,
"a selective cross-stage join is a candidate"
);
assert_eq!(candidates[0].build_stage, 0);
assert_eq!(candidates[0].probe_stage, 1);
}
#[test]
fn a_nested_loop_join_remains_unreadable() {
use datafusion::physical_plan::joins::NestedLoopJoinExec;
let left = read(0, Some(1_000), "k", arrow::datatypes::DataType::Int64);
let right = read(1, Some(10_000_000), "k", arrow::datatypes::DataType::Int64);
let nlj = NestedLoopJoinExec::try_new(left, right, None, &JoinType::Inner, None)
.expect("nested-loop join");
let plan: Arc<dyn ExecutionPlan> = Arc::new(nlj);
let mut candidates = Vec::new();
let mut rejects = RuntimeFilterRejects::default();
collect_runtime_filter_candidates(&plan, &mut candidates, &mut rejects);
assert_eq!(rejects.joins_of_unsupported_kind, 1);
assert_eq!(rejects.joins, 0);
assert!(candidates.is_empty());
}
fn schema(name: &str, key: arrow::datatypes::DataType) -> SchemaRef {
Arc::new(arrow::datatypes::Schema::new(vec![
arrow::datatypes::Field::new(name, key, false),
arrow::datatypes::Field::new("payload", arrow::datatypes::DataType::Utf8, false),
]))
}
fn draft(stage: usize, rows: Option<usize>, key: &str) -> StageDraft {
StageDraft {
plan: Arc::new(
ShuffleReadExec::new(
stage,
4,
4,
schema(key, arrow::datatypes::DataType::Int64),
None,
)
.with_upstream_estimate(rows, None),
),
shuffle: Some(StageShuffleOutput {
key_columns: vec![String::from(key)],
num_output_partitions: 4,
}),
subqueries: None,
}
}
fn read(
stage: usize,
rows: Option<usize>,
key: &str,
key_type: arrow::datatypes::DataType,
) -> Arc<dyn ExecutionPlan> {
Arc::new(
ShuffleReadExec::new(stage, 4, 4, schema(key, key_type), None)
.with_upstream_estimate(rows, None),
)
}
fn join_of(
build: Arc<dyn ExecutionPlan>,
probe: Arc<dyn ExecutionPlan>,
join_type: JoinType,
) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
Ok(Arc::new(HashJoinExec::try_new(
build,
probe,
vec![(
Arc::new(Column::new("bkey", 0)),
Arc::new(Column::new("pkey", 0)),
)],
None,
&join_type,
None,
PartitionMode::Partitioned,
NullEquality::NullEqualsNothing,
false,
)?))
}
fn q10_shaped() -> datafusion::error::Result<(Arc<dyn ExecutionPlan>, Vec<StageDraft>)> {
let root = join_of(
read(
0,
Some(1_000_000),
"bkey",
arrow::datatypes::DataType::Int64,
),
read(
1,
Some(100_000_000),
"pkey",
arrow::datatypes::DataType::Int64,
),
JoinType::Inner,
)?;
Ok((
root,
vec![
draft(0, Some(1_000_000), "bkey"),
draft(1, Some(100_000_000), "pkey"),
],
))
}
#[test]
fn a_filter_stage_is_injected_and_the_probe_stage_waits_on_it() {
let (root, mut drafts) = q10_shaped().expect("plan");
assert_eq!(
inject_runtime_filters_unconditionally(&root, &mut drafts),
1
);
assert_eq!(drafts.len(), 3, "one filter stage must have been appended");
let filter = &drafts[2];
let shuffle = filter.shuffle.as_ref().expect("filter stage shuffles");
assert!(
shuffle.key_columns.is_empty() && shuffle.num_output_partitions == 1,
"the filter must gather to ONE keyless partition; any other shape means \
every probe task fetches N partials instead of one filter"
);
assert_eq!(
filter.plan.output_partitioning().partition_count(),
1,
"the filter stage must be a single task, or the broadcast it feeds \
multiplies by the task count"
);
assert!(
collect_upstream_stage_indexes(&drafts[1].plan).contains(&2),
"the probe stage must declare the filter stage upstream, or the \
scheduler will run it before the filter exists"
);
}
#[test]
fn the_probe_stages_output_schema_is_untouched() {
let (root, mut drafts) = q10_shaped().expect("plan");
let before = drafts[1].plan.schema();
inject_runtime_filters_unconditionally(&root, &mut drafts);
assert_eq!(
before.fields(),
drafts[1].plan.schema().fields(),
"wrapping the probe stage must not change its columns"
);
}
#[test]
fn the_filter_stage_emits_one_binary_column() {
let (root, mut drafts) = q10_shaped().expect("plan");
inject_runtime_filters_unconditionally(&root, &mut drafts);
assert_eq!(
drafts[2].plan.schema().fields().len(),
1,
"a filter stage carries only the serialized bloom"
);
}
#[test]
fn a_non_inner_join_gets_no_filter() {
for join_type in [
JoinType::Full,
JoinType::Right,
JoinType::RightAnti,
JoinType::LeftAnti,
] {
let root = join_of(
read(
0,
Some(1_000_000),
"bkey",
arrow::datatypes::DataType::Int64,
),
read(
1,
Some(100_000_000),
"pkey",
arrow::datatypes::DataType::Int64,
),
join_type,
)
.expect("join");
let mut drafts = vec![
draft(0, Some(1_000_000), "bkey"),
draft(1, Some(100_000_000), "pkey"),
];
assert_eq!(
inject_runtime_filters_unconditionally(&root, &mut drafts),
0,
"{join_type:?} can preserve unmatched PROBE rows; dropping them is a \
wrong answer, not a slow one"
);
}
}
#[test]
fn a_join_inside_one_stage_gets_no_filter() {
let root = join_of(
read(
0,
Some(1_000_000),
"bkey",
arrow::datatypes::DataType::Int64,
),
read(
0,
Some(100_000_000),
"pkey",
arrow::datatypes::DataType::Int64,
),
JoinType::Inner,
)
.expect("join");
let mut drafts = vec![draft(0, Some(1_000_000), "bkey")];
assert_eq!(
inject_runtime_filters_unconditionally(&root, &mut drafts),
0,
"a same-stage join already gets DataFusion's own dynamic filter"
);
}
#[test]
fn an_absent_estimate_refuses_rather_than_guesses() {
for (build_rows, probe_rows) in [(None, Some(100_000_000)), (Some(1_000_000), None)] {
let root = join_of(
read(0, build_rows, "bkey", arrow::datatypes::DataType::Int64),
read(1, probe_rows, "pkey", arrow::datatypes::DataType::Int64),
JoinType::Inner,
)
.expect("join");
let mut drafts = vec![draft(0, build_rows, "bkey"), draft(1, probe_rows, "pkey")];
assert_eq!(
inject_runtime_filters_unconditionally(&root, &mut drafts),
0,
"Precision::Absent means 'no idea', never 'small' — guessing here is \
the SpillableJoinSelection lesson"
);
}
}
#[test]
fn a_probe_barely_bigger_than_the_build_is_not_worth_a_stage() {
let root = join_of(
read(
0,
Some(1_000_000),
"bkey",
arrow::datatypes::DataType::Int64,
),
read(
1,
Some(2_000_000),
"pkey",
arrow::datatypes::DataType::Int64,
),
JoinType::Inner,
)
.expect("join");
let mut drafts = vec![
draft(0, Some(1_000_000), "bkey"),
draft(1, Some(2_000_000), "pkey"),
];
assert_eq!(
inject_runtime_filters_unconditionally(&root, &mut drafts),
0,
"2x is not enough to repay an extra scan plus a broadcast"
);
}
#[test]
fn an_oversized_filter_is_refused() {
let huge = 5_000_000_000usize;
let root = join_of(
read(0, Some(huge), "bkey", arrow::datatypes::DataType::Int64),
read(1, Some(huge * 8), "pkey", arrow::datatypes::DataType::Int64),
JoinType::Inner,
)
.expect("join");
let mut drafts = vec![
draft(0, Some(huge), "bkey"),
draft(1, Some(huge * 8), "pkey"),
];
assert_eq!(
inject_runtime_filters_unconditionally(&root, &mut drafts),
0
);
}
#[test]
fn an_unsupported_key_type_is_refused_not_guessed() {
let root = join_of(
read(
0,
Some(1_000_000),
"bkey",
arrow::datatypes::DataType::Float64,
),
read(
1,
Some(100_000_000),
"pkey",
arrow::datatypes::DataType::Float64,
),
JoinType::Inner,
)
.expect("join");
let mut drafts = vec![
draft(0, Some(1_000_000), "bkey"),
draft(1, Some(100_000_000), "pkey"),
];
assert_eq!(
inject_runtime_filters_unconditionally(&root, &mut drafts),
0,
"-0.0 == 0.0 compares equal but encodes differently, so a float bloom \
would produce false negatives"
);
}
#[test]
fn a_filter_that_would_close_a_cycle_is_not_injected() {
let root = join_of(
read(
0,
Some(1_000_000),
"bkey",
arrow::datatypes::DataType::Int64,
),
read(
1,
Some(100_000_000),
"pkey",
arrow::datatypes::DataType::Int64,
),
JoinType::Inner,
)
.expect("join");
let mut drafts = vec![
draft(0, Some(1_000_000), "bkey"),
draft(1, Some(100_000_000), "pkey"),
];
drafts[0].plan = Arc::new(
ShuffleReadExec::new(
1,
4,
4,
schema("bkey", arrow::datatypes::DataType::Int64),
None,
)
.with_upstream_estimate(Some(1_000_000), None),
);
assert!(
stage_depends_on(&drafts, 0, 1),
"precondition: build reads probe"
);
assert_eq!(
inject_runtime_filters_unconditionally(&root, &mut drafts),
0
);
}
#[test]
fn the_dry_run_reports_without_rewriting_when_the_flag_is_off() {
let root: Arc<dyn ExecutionPlan> = Arc::new(ShuffleReadExec::new(
0,
1,
1,
schema("k", arrow::datatypes::DataType::Int64),
None,
));
let mut drafts = vec![draft(0, Some(1), "a"), draft(1, Some(1), "b")];
let before = drafts.len();
assert!(
!crate::runtime_filter_exec::enabled(),
"precondition: the feature ships dark"
);
assert_eq!(
inject_runtime_filters(&root, &mut drafts),
0,
"a disabled pass must inject nothing"
);
assert_eq!(drafts.len(), before, "a disabled pass must not add a stage");
}
#[test]
fn stage_dependency_reachability_is_transitive_and_terminates_on_cycles() {
let mut drafts = vec![
draft(0, Some(1), "a"),
draft(1, Some(1), "b"),
draft(2, Some(1), "c"),
];
drafts[1].plan = Arc::new(ShuffleReadExec::new(
0,
1,
1,
schema("b", arrow::datatypes::DataType::Int64),
None,
));
drafts[2].plan = Arc::new(ShuffleReadExec::new(
1,
1,
1,
schema("c", arrow::datatypes::DataType::Int64),
None,
));
assert!(
stage_depends_on(&drafts, 2, 0),
"reachability must be transitive"
);
assert!(!stage_depends_on(&drafts, 0, 2), "and directional");
}
#[test]
fn a_subquery_parameterised_stage_is_never_cloned_into_a_filter() {
let (root, mut drafts) = q10_shaped().expect("plan");
drafts[0].subqueries = Some(StageSubqueryContext {
links: Vec::new(),
results: Default::default(),
});
assert_eq!(
inject_runtime_filters_unconditionally(&root, &mut drafts),
0
);
}
#[test]
fn the_feature_is_off_unless_the_flag_says_otherwise() {
let (root, mut drafts) = q10_shaped().expect("plan");
assert_eq!(
inject_runtime_filters(&root, &mut drafts),
0,
"the flag-checking entry point must decline by default: this rule \
rewrites the stage DAG and ships dark until a clean 22-query sweep"
);
assert_eq!(drafts.len(), 2, "and it must not have touched the drafts");
}
#[test]
fn the_injected_stages_round_trip_through_the_codec() {
let (root, mut drafts) = q10_shaped().expect("plan");
assert_eq!(
inject_runtime_filters_unconditionally(&root, &mut drafts),
1
);
let codec = KrishivPhysicalCodec::coordinator();
let session = fragment_decode_session_context();
let ctx = session.task_ctx();
for (index, draft) in drafts.iter().enumerate() {
let bytes = encode_dfplan_bytes(Arc::clone(&draft.plan), &codec)
.unwrap_or_else(|e| panic!("stage {index} did not encode: {e}"));
verify_dfplan_roundtrip(&bytes, &codec, &ctx, Some(&draft.plan))
.unwrap_or_else(|e| panic!("stage {index} did not decode: {e}"));
}
}
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod registration_parity_tests {
use super::*;
async fn parquet_at(dir: &std::path::Path) -> String {
let path = dir.join("t.parquet");
let path = path.to_str().expect("temp path is utf-8").to_owned();
let ctx = SessionContext::new();
ctx.sql(&format!(
"COPY (SELECT * FROM (VALUES (1, 'a'), (2, 'b'), (3, 'c')) t(k, v)) \
TO '{path}' STORED AS PARQUET"
))
.await
.unwrap()
.collect()
.await
.unwrap();
path
}
async fn both_ways(
path: &str,
) -> (
Arc<dyn datafusion::datasource::TableProvider>,
Arc<dyn datafusion::datasource::TableProvider>,
) {
let ctx = planning_session_context(4);
register_parquet_table(&ctx, &ParquetTableSpec::new("plain", path))
.await
.unwrap();
register_parquet_table(
&ctx,
&ParquetTableSpec::new("keyed", path).with_primary_key(["k"]),
)
.await
.unwrap();
(
ctx.table_provider("plain").await.unwrap(),
ctx.table_provider("keyed").await.unwrap(),
)
}
fn listing_options(
provider: &Arc<dyn datafusion::datasource::TableProvider>,
) -> datafusion::datasource::listing::ListingOptions {
let any = provider.as_ref() as &dyn std::any::Any;
any.downcast_ref::<datafusion::datasource::listing::ListingTable>()
.expect("parquet registration produces a ListingTable")
.options()
.clone()
}
#[test]
fn an_extensionless_object_store_path_is_treated_as_a_directory() {
assert_eq!(
super::directory_aware_url("s3://b/sf100/lineitem"),
"s3://b/sf100/lineitem/"
);
assert_eq!(
super::directory_aware_url("s3://b/sf100/lineitem/"),
"s3://b/sf100/lineitem/"
);
assert_eq!(
super::directory_aware_url("s3://b/sf100/nation.parquet"),
"s3://b/sf100/nation.parquet"
);
assert_eq!(
super::directory_aware_url("/data/sf100/lineitem"),
"/data/sf100/lineitem"
);
assert_eq!(super::directory_aware_url("relative/dir"), "relative/dir");
}
#[tokio::test]
async fn declaring_a_primary_key_changes_only_the_constraints() {
let dir = tempfile::tempdir().unwrap();
let path = parquet_at(dir.path()).await;
let (plain, keyed) = both_ways(&path).await;
let (plain_options, keyed_options) = (listing_options(&plain), listing_options(&keyed));
assert_eq!(
format!("{plain_options:?}"),
format!("{keyed_options:?}"),
"a declared key must not change how the table is read"
);
assert!(
keyed_options.collect_stat,
"statistics collection must stay on: every size-based rule goes \
blind without it"
);
assert!(plain.constraints().is_none_or(|c| c.is_empty()));
assert!(
keyed.constraints().is_some_and(|c| !c.is_empty()),
"the declared key must reach the optimizer as a constraint"
);
}
#[tokio::test]
async fn a_keyed_table_still_reports_row_counts() {
use datafusion::common::stats::Precision;
let dir = tempfile::tempdir().unwrap();
let path = parquet_at(dir.path()).await;
let (_, keyed) = both_ways(&path).await;
let ctx = planning_session_context(4);
ctx.register_table("keyed", Arc::clone(&keyed)).unwrap();
let stats = ctx
.sql("SELECT k, v FROM keyed")
.await
.unwrap()
.create_physical_plan()
.await
.unwrap()
.partition_statistics(None)
.unwrap();
assert!(
matches!(stats.num_rows, Precision::Exact(3) | Precision::Inexact(3)),
"expected a row count for a keyed table, got {:?} — this is the \
shape that made every join unmeasurable at SF100",
stats.num_rows
);
}
}