diff --git a/src/common/recordbatch/src/filter.rs b/src/common/recordbatch/src/filter.rs index dcb00d7407a..1b398d6cc45 100644 --- a/src/common/recordbatch/src/filter.rs +++ b/src/common/recordbatch/src/filter.rs @@ -399,7 +399,6 @@ impl SimpleFilterEvaluator { // semantics: null results are filtered out). return Some(TimestampUnitCast::Pruned); }; - // Fast path: the literal is already in the target unit. if unit == target_unit { return Some(TimestampUnitCast::Filter(self.clone())); } @@ -464,15 +463,19 @@ fn timestamp_scalar_parts(scalar: &ScalarValue) -> Option<(Option, TimeUnit (timezone.is_none()).then_some((value, unit)) } -/// Builds a one-element scalar array holding `value` in `unit`. -fn timestamp_scalar(value: i64, unit: TimeUnit) -> Option> { - let scalar = match unit { +/// Builds a tz-naive timestamp literal for `value` in `unit`. +pub fn timestamp_scalar_value(value: i64, unit: TimeUnit) -> ScalarValue { + match unit { TimeUnit::Second => ScalarValue::TimestampSecond(Some(value), None), TimeUnit::Millisecond => ScalarValue::TimestampMillisecond(Some(value), None), TimeUnit::Microsecond => ScalarValue::TimestampMicrosecond(Some(value), None), TimeUnit::Nanosecond => ScalarValue::TimestampNanosecond(Some(value), None), - }; - scalar.to_scalar().ok() + } +} + +/// Builds a one-element scalar array holding `value` in `unit`. +fn timestamp_scalar(value: i64, unit: TimeUnit) -> Option> { + timestamp_scalar_value(value, unit).to_scalar().ok() } /// Evaluate the predicate on the input [RecordBatch], and return a new [RecordBatch]. diff --git a/src/common/time/src/timestamp.rs b/src/common/time/src/timestamp.rs index 4290e88cb06..c04cd84bf90 100644 --- a/src/common/time/src/timestamp.rs +++ b/src/common/time/src/timestamp.rs @@ -611,6 +611,17 @@ impl From<&ArrowTimeUnit> for TimeUnit { } } +impl From for ArrowTimeUnit { + fn from(unit: TimeUnit) -> Self { + match unit { + TimeUnit::Second => Self::Second, + TimeUnit::Millisecond => Self::Millisecond, + TimeUnit::Microsecond => Self::Microsecond, + TimeUnit::Nanosecond => Self::Nanosecond, + } + } +} + /// The exact division of a timestamp value into a different unit: /// `quotient * from_scale + remainder == value * to_scale` with /// `0 <= remainder < from_scale`. `quotient` is the value floored in diff --git a/src/operator/src/statement/ddl.rs b/src/operator/src/statement/ddl.rs index ba1332a6d85..ba25f58d7ae 100644 --- a/src/operator/src/statement/ddl.rs +++ b/src/operator/src/statement/ddl.rs @@ -58,10 +58,11 @@ use common_meta::rpc::ddl::{ SubmitDdlTaskResponse, TriggerReason, }; use common_query::{Output, OutputData}; +use common_recordbatch::filter::timestamp_scalar_value; use common_recordbatch::{RecordBatch, RecordBatches}; use common_sql::convert::sql_value_to_value; use common_telemetry::{debug, info, tracing, warn}; -use common_time::timestamp::{TimeUnit, div_mod_units}; +use common_time::timestamp::div_mod_units; use common_time::{Timestamp, Timezone}; use datafusion::datasource::provider_as_source; use datafusion_common::tree_node::TreeNodeVisitor; @@ -1943,17 +1944,6 @@ impl StatementExecutor { Ok(Output::new_with_affected_rows(0)) } - /// Builds a tz-naive timestamp literal in the given unit. - fn ts_scalar(value: i64, unit: TimeUnit) -> datafusion_common::ScalarValue { - use datafusion_common::ScalarValue; - match unit { - TimeUnit::Second => ScalarValue::TimestampSecond(Some(value), None), - TimeUnit::Millisecond => ScalarValue::TimestampMillisecond(Some(value), None), - TimeUnit::Microsecond => ScalarValue::TimestampMicrosecond(Some(value), None), - TimeUnit::Nanosecond => ScalarValue::TimestampNanosecond(Some(value), None), - } - } - /// Preflights a time index unit widening: rejects the alter if any existing /// timestamp would overflow the target unit's `i64` range (arrow casts such /// values to NULL, which a NOT NULL time index cannot hold). @@ -2008,47 +1998,60 @@ impl StatementExecutor { .map(|q| q.quotient + (q.remainder != 0) as i64) .unwrap_or(i64::MIN); - let filter = datafusion_expr::col(&ts_column.name) - .gt(datafusion_expr::lit(Self::ts_scalar(max_cur, current_unit))) - .or(datafusion_expr::col(&ts_column.name) - .lt(datafusion_expr::lit(Self::ts_scalar(min_cur, current_unit)))); + // One single-sided existence scan per bound, run separately: a single + // `ts > max OR ts < min` scan would widen to the full time range and + // read every SST's metadata, while each single-sided predicate prunes + // files (in-memory time ranges) and skips memtables (stats) without + // any IO when nothing overflows. + let ts_col = datafusion_expr::col(&ts_column.name); + let bounds = [ + ts_col + .clone() + .gt(datafusion_expr::lit(timestamp_scalar_value( + max_cur, + current_unit.into(), + ))), + ts_col.lt(datafusion_expr::lit(timestamp_scalar_value( + min_cur, + current_unit.into(), + ))), + ]; + for filter in bounds { + let provider = std::sync::Arc::new(DfTableProviderAdapter::new(table.clone())); + let plan = LogicalPlanBuilder::scan( + expr.table_name.as_str(), + provider_as_source(provider), + Some(vec![ts_index]), + ) + .and_then(|builder| builder.filter(filter)) + .and_then(|builder| builder.limit(0, Some(1))) + .and_then(LogicalPlanBuilder::build) + .map_err(|e| { + error::UnexpectedSnafu { + violated: format!("Failed to build time index overflow preflight plan: {e}"), + } + .build() + })?; + let output = self.exec_plan(plan, query_context.clone()).await?; - // An existence scan across all regions (file and row-group pruning makes - // it cheap when nothing overflows). - let provider = std::sync::Arc::new(DfTableProviderAdapter::new(table.clone())); - let plan = LogicalPlanBuilder::scan( - expr.table_name.as_str(), - provider_as_source(provider), - Some(vec![ts_index]), - ) - .and_then(|builder| builder.filter(filter)) - .and_then(|builder| builder.limit(0, Some(1))) - .and_then(LogicalPlanBuilder::build) - .map_err(|e| { - error::UnexpectedSnafu { - violated: format!("Failed to build time index overflow preflight plan: {e}"), - } - .build() - })?; - let output = self.exec_plan(plan, query_context.clone()).await?; - - let has_overflow = match output.data { - OutputData::Stream(mut stream) => futures::StreamExt::next(&mut stream) - .await - .is_some_and(|batch| batch.map(|batch| batch.num_rows() > 0).unwrap_or(true)), - _ => false, - }; - ensure!( - !has_overflow, - error::TimeIndexWideningOverflowSnafu { - table: format_full_table_name( - &expr.catalog_name, - &expr.schema_name, - &expr.table_name - ), - column: ts_column.name.clone(), - } - ); + let has_overflow = match output.data { + OutputData::Stream(mut stream) => futures::StreamExt::next(&mut stream) + .await + .is_some_and(|batch| batch.map(|batch| batch.num_rows() > 0).unwrap_or(true)), + _ => false, + }; + ensure!( + !has_overflow, + error::TimeIndexWideningOverflowSnafu { + table: format_full_table_name( + &expr.catalog_name, + &expr.schema_name, + &expr.table_name + ), + column: ts_column.name.clone(), + } + ); + } Ok(()) }