diff --git a/Cargo.lock b/Cargo.lock index aaa49b2af13..d30ce2bfc8c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9252,13 +9252,13 @@ dependencies = [ [[package]] name = "sqlparser" version = "0.34.0" -source = "git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=c3814f08afa19786b13d72b1731a1e8b3cac4ab9#c3814f08afa19786b13d72b1731a1e8b3cac4ab9" +source = "git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=296a4f6c73b129d6f565a42a2e5e53c6bc2b9da4#296a4f6c73b129d6f565a42a2e5e53c6bc2b9da4" dependencies = [ "lazy_static", "log", "regex", "sqlparser 0.35.0", - "sqlparser_derive 0.1.1 (git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=c3814f08afa19786b13d72b1731a1e8b3cac4ab9)", + "sqlparser_derive 0.1.1 (git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=296a4f6c73b129d6f565a42a2e5e53c6bc2b9da4)", ] [[package]] @@ -9285,7 +9285,7 @@ dependencies = [ [[package]] name = "sqlparser_derive" version = "0.1.1" -source = "git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=c3814f08afa19786b13d72b1731a1e8b3cac4ab9#c3814f08afa19786b13d72b1731a1e8b3cac4ab9" +source = "git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=296a4f6c73b129d6f565a42a2e5e53c6bc2b9da4#296a4f6c73b129d6f565a42a2e5e53c6bc2b9da4" dependencies = [ "proc-macro2", "quote", diff --git a/Cargo.toml b/Cargo.toml index 16054c93b18..6e17a0fcc0a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -90,7 +90,7 @@ regex = "1.8" serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" snafu = { version = "0.7", features = ["backtraces"] } -sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "c3814f08afa19786b13d72b1731a1e8b3cac4ab9", features = [ +sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "296a4f6c73b129d6f565a42a2e5e53c6bc2b9da4", features = [ "visitor", ] } tempfile = "3" diff --git a/src/query/Cargo.toml b/src/query/Cargo.toml index 6607c49eb1b..3e59d1b44c8 100644 --- a/src/query/Cargo.toml +++ b/src/query/Cargo.toml @@ -8,6 +8,7 @@ license.workspace = true ahash = { version = "0.8", features = ["compile-time-rng"] } arc-swap = "1.0" arrow-schema.workspace = true +arrow.workspace = true async-recursion = "1.0" async-stream.workspace = true async-trait = "0.1" diff --git a/src/query/src/range_select/plan.rs b/src/query/src/range_select/plan.rs index d3bda025d27..26918b4f81a 100644 --- a/src/query/src/range_select/plan.rs +++ b/src/query/src/range_select/plan.rs @@ -12,29 +12,48 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::collections::hash_map::Entry; +use std::collections::HashMap; use std::fmt::Display; use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; use std::time::Duration; -use arrow_schema::{Field, Schema, SchemaRef}; +use ahash::RandomState; +use arrow::compute; +use arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit}; use common_query::DfPhysicalPlan; +use common_recordbatch::DfSendableRecordBatchStream; use datafusion::common::{Result as DataFusionResult, Statistics}; use datafusion::error::Result as DfResult; use datafusion::execution::context::SessionState; +use datafusion::physical_plan::metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet}; +use datafusion::physical_plan::udaf::create_aggregate_expr as create_aggr_udf_expr; use datafusion::physical_plan::{ DisplayAs, DisplayFormatType, ExecutionPlan, RecordBatchStream, SendableRecordBatchStream, }; -use datafusion_common::{DFField, DFSchema, DFSchemaRef}; +use datafusion_common::utils::get_arrayref_at_indices; +use datafusion_common::{DFField, DFSchema, DFSchemaRef, DataFusionError, ScalarValue}; use datafusion_expr::utils::exprlist_to_fields; -use datafusion_expr::{Expr, ExprSchemable, LogicalPlan, UserDefinedLogicalNodeCore}; +use datafusion_expr::{Accumulator, Expr, ExprSchemable, LogicalPlan, UserDefinedLogicalNodeCore}; +use datafusion_physical_expr::expressions::create_aggregate_expr as create_aggr_expr; +use datafusion_physical_expr::hash_utils::create_hashes; +use datafusion_physical_expr::{create_physical_expr, AggregateExpr, PhysicalExpr}; +use datatypes::arrow::array::{ + Array, ArrayRef, TimestampMillisecondArray, TimestampMillisecondBuilder, UInt32Builder, +}; +use datatypes::arrow::datatypes::{ArrowPrimitiveType, TimestampMillisecondType}; use datatypes::arrow::record_batch::RecordBatch; -use futures::{Stream, StreamExt}; +use datatypes::arrow::row::{OwnedRow, RowConverter, SortField}; +use futures::{ready, Stream}; +use futures_util::StreamExt; use snafu::ResultExt; use crate::error::{DataFusionSnafu, Result}; +type Millisecond = ::Native; + #[derive(PartialEq, Eq, Hash, Clone, Debug)] pub struct RangeFn { pub expr: Expr, @@ -64,6 +83,7 @@ pub struct RangeSelect { pub time_index: String, pub by: Vec, pub schema: DFSchemaRef, + pub by_schema: DFSchemaRef, } impl RangeSelect { @@ -80,7 +100,10 @@ impl RangeSelect { Ok(DFField::new_unqualified( &expr.display_name()?, expr.get_type(input.schema())?, - expr.nullable(input.schema())?, + // TODO(Taylor-lagrange): We have not implemented fill currently, + // it is possible that some columns may not be able to aggregate data, + // so we temporarily set that all data is nullable + true, )) }) .collect::>>() @@ -92,17 +115,20 @@ impl RangeSelect { let time_index_name = ts_field.name().clone(); fields.push(ts_field); // add by - fields.extend( - exprlist_to_fields(by.iter().collect::>(), &input).context(DataFusionSnafu)?, - ); + let by_fields = + exprlist_to_fields(by.iter().collect::>(), &input).context(DataFusionSnafu)?; + fields.extend(by_fields.clone()); let schema = DFSchema::new_with_metadata(fields, input.schema().metadata().clone()) .context(DataFusionSnafu)?; + let by_schema = DFSchema::new_with_metadata(by_fields, input.schema().metadata().clone()) + .context(DataFusionSnafu)?; Ok(Self { input, range_expr, align, time_index: time_index_name, schema: Arc::new(schema), + by_schema: Arc::new(by_schema), by, }) } @@ -152,16 +178,30 @@ impl UserDefinedLogicalNodeCore for RangeSelect { time_index: self.time_index.clone(), schema: self.schema.clone(), by: self.by.clone(), + by_schema: self.by_schema.clone(), } } } impl RangeSelect { + fn create_physical_expr_list( + &self, + exprs: &[Expr], + df_schema: &Arc, + schema: &Schema, + session_state: &SessionState, + ) -> DfResult>> { + exprs + .iter() + .map(|by| create_physical_expr(by, df_schema, schema, session_state.execution_props())) + .collect::>>() + } + pub fn to_execution_plan( &self, - _logical_input: &LogicalPlan, + logical_input: &LogicalPlan, exec_input: Arc, - _session_state: &SessionState, + session_state: &SessionState, ) -> DfResult> { let fields: Vec<_> = self .schema @@ -169,22 +209,126 @@ impl RangeSelect { .iter() .map(|field| Field::new(field.name(), field.data_type().clone(), field.is_nullable())) .collect(); + let by_fields: Vec<_> = self + .by_schema + .fields() + .iter() + .map(|field| Field::new(field.name(), field.data_type().clone(), field.is_nullable())) + .collect(); + let input_dfschema = logical_input.schema(); + let input_schema = exec_input.schema(); + let range_exec: Vec = self + .range_expr + .iter() + .map(|range_fn| { + let (expr, args) = match &range_fn.expr { + Expr::AggregateFunction(aggr) => { + let args = self.create_physical_expr_list( + &aggr.args, + input_dfschema, + &input_schema, + session_state, + )?; + Ok(( + create_aggr_expr( + &aggr.fun, + false, + &args, + &[], + &input_schema, + range_fn.expr.display_name()?, + )?, + args, + )) + } + Expr::AggregateUDF(aggr_udf) => { + let args = self.create_physical_expr_list( + &aggr_udf.args, + input_dfschema, + &input_schema, + session_state, + )?; + Ok(( + create_aggr_udf_expr( + &aggr_udf.fun, + &args, + &input_schema, + range_fn.expr.display_name()?, + )?, + args, + )) + } + _ => Err(DataFusionError::Plan(format!( + "Unexpected Expr:{} in RangeSelect", + range_fn.expr.display_name()? + ))), + }?; + Ok(RangeFnExec { + expr, + args, + range: range_fn.range.as_millis() as Millisecond, + }) + }) + .collect::>>()?; Ok(Arc::new(RangeSelectExec { input: exec_input, + range_exec, + align: self.align.as_millis() as Millisecond, + by: self.create_physical_expr_list( + &self.by, + input_dfschema, + &input_schema, + session_state, + )?, + time_index: self.time_index.clone(), schema: Arc::new(Schema::new(fields)), + by_schema: Arc::new(Schema::new(by_fields)), + metric: ExecutionPlanMetricsSet::new(), })) } } +#[derive(Debug, Clone)] +struct RangeFnExec { + pub expr: Arc, + pub args: Vec>, + pub range: Millisecond, +} + #[derive(Debug)] pub struct RangeSelectExec { input: Arc, + range_exec: Vec, + align: Millisecond, + time_index: String, + by: Vec>, schema: SchemaRef, + by_schema: SchemaRef, + metric: ExecutionPlanMetricsSet, } impl DisplayAs for RangeSelectExec { - fn fmt_as(&self, _t: DisplayFormatType, f: &mut std::fmt::Formatter) -> std::fmt::Result { - write!(f, "RangeSelectExec: ") + fn fmt_as(&self, t: DisplayFormatType, f: &mut std::fmt::Formatter) -> std::fmt::Result { + match t { + DisplayFormatType::Default | DisplayFormatType::Verbose => { + write!(f, "RangeSelectExec: ")?; + let range_expr_strs: Vec = self + .range_exec + .iter() + .map(|e| format!("RangeFnExec{{ {}, range: {:?}}}", e.expr.name(), e.range)) + .collect(); + let by: Vec = self.by.iter().map(|e| e.to_string()).collect(); + write!( + f, + "range_expr=[{}], align={}, time_index={}, by=[{}]", + range_expr_strs.join(", "), + self.align, + self.time_index, + by.join(", ") + )?; + } + } + Ok(()) } } @@ -216,7 +360,13 @@ impl ExecutionPlan for RangeSelectExec { assert!(!children.is_empty()); Ok(Arc::new(Self { input: children[0].clone(), + range_exec: self.range_exec.clone(), + time_index: self.time_index.clone(), + by: self.by.clone(), + align: self.align, schema: self.schema.clone(), + by_schema: self.by_schema.clone(), + metric: self.metric.clone(), })) } @@ -224,22 +374,248 @@ impl ExecutionPlan for RangeSelectExec { &self, partition: usize, context: Arc, - ) -> datafusion_common::Result { + ) -> DfResult { + let baseline_metric = BaselineMetrics::new(&self.metric, partition); let input = self.input.execute(partition, context)?; + let schema = input.schema(); + let time_index = schema + .column_with_name(&self.time_index) + .ok_or(DataFusionError::Execution( + "time index column not found".into(), + ))? + .0; + let row_converter = RowConverter::new( + self.by_schema + .fields() + .iter() + .map(|f| SortField::new(f.data_type().clone())) + .collect(), + )?; Ok(Box::pin(RangeSelectStream { schema: self.schema.clone(), + range_exec: self.range_exec.clone(), input, + random_state: RandomState::new(), + time_index, + align: self.align, + by: self.by.clone(), + series_map: HashMap::new(), + exec_state: ExecutionState::ReadingInput, + output_num_rows: 0, + row_converter, + modify_map: HashMap::new(), + metric: baseline_metric, })) } + fn metrics(&self) -> Option { + Some(self.metric.clone_inner()) + } + fn statistics(&self) -> Statistics { self.input.statistics() } } -pub struct RangeSelectStream { +struct RangeSelectStream { + /// the schema of output column schema: SchemaRef, + range_exec: Vec, input: SendableRecordBatchStream, + /// Column index of TIME INDEX column's position in the input schema + time_index: usize, + /// the unit of `align` is millisecond + align: Millisecond, + by: Vec>, + exec_state: ExecutionState, + /// Converter for the by values + row_converter: RowConverter, + random_state: RandomState, + /// key: time series's hash value + /// value: time series's state on different align_ts + series_map: HashMap, + /// key: (hash of by rows, align_ts) + /// value: [row_ids] + /// It is used to record the data that needs to be aggregated in each time slot during the data update process + modify_map: HashMap<(u64, Millisecond), Vec>, + /// The number of rows of the final output + output_num_rows: usize, + metric: BaselineMetrics, +} + +struct SeriesState { + /// by values written by `RowWriter` + row: OwnedRow, + /// key: align_ts + /// value: a vector, each element is a range_fn follow the order of `range_exec` + align_ts_accumulator: HashMap>>, +} + +/// According to `align`, produces a calendar-based aligned time. +/// Combining the parameters related to the range query, +/// determine for each `Accumulator` `(hash, align_ts)` define, +/// which rows of data will be applied to it. +fn align_to_calendar( + range: Millisecond, + align: Millisecond, + ts_column: &TimestampMillisecondArray, + by_columns_hash: &[u64], + modify_map: &mut HashMap<(u64, Millisecond), Vec>, +) { + modify_map.clear(); + // make modify_map for range_fn[i] + for (row, hash) in by_columns_hash.iter().enumerate() { + let ts = ts_column.value(row); + let mut align_ts = ((ts + align - 1) / align) * align; + while align_ts - range < ts && ts <= align_ts { + modify_map + .entry((*hash, align_ts)) + .or_default() + .push(row as u32); + align_ts += align; + } + } +} + +impl RangeSelectStream { + fn evaluate_many( + &self, + batch: &RecordBatch, + exprs: &[Arc], + ) -> DfResult> { + exprs + .iter() + .map(|expr| { + let value = expr.evaluate(batch)?; + Ok(value.into_array(batch.num_rows())) + }) + .collect::>>() + } + + fn update_range_context(&mut self, batch: RecordBatch) -> DfResult<()> { + let _timer = self.metric.elapsed_compute().timer(); + let num_rows = batch.num_rows(); + let by_arrays = self.evaluate_many(&batch, &self.by)?; + let mut hashes = vec![0; num_rows]; + create_hashes(&by_arrays, &self.random_state, &mut hashes)?; + let by_rows = self.row_converter.convert_columns(&by_arrays)?; + let mut ts_column = batch.column(self.time_index).clone(); + if !matches!( + ts_column.data_type(), + DataType::Timestamp(TimeUnit::Millisecond, _) + ) { + ts_column = compute::cast( + ts_column.as_ref(), + &DataType::Timestamp(TimeUnit::Millisecond, None), + )?; + } + let ts_column_ref = ts_column + .as_any() + .downcast_ref::() + .ok_or(DataFusionError::Execution( + "Time index Column downcast to TimestampMillisecondArray failed".into(), + ))?; + for i in 0..self.range_exec.len() { + let args = self.evaluate_many(&batch, &self.range_exec[i].args)?; + // use self.modify_map record (hash, align_ts) => [row_nums] + align_to_calendar( + self.range_exec[i].range, + self.align, + ts_column_ref, + &hashes, + &mut self.modify_map, + ); + // build modify_rows/modify_index/offsets for batch update + let mut modify_rows = UInt32Builder::with_capacity(0); + // (hash, align_ts, row_num) + // row_num use to find a by value + // So we just need to record the row_num of a modify row randomly, because they all have the same by value + let mut modify_index = Vec::with_capacity(self.modify_map.len()); + let mut offsets = vec![0]; + let mut offset_so_far = 0; + for ((hash, ts), modify) in &self.modify_map { + modify_rows.append_slice(modify); + offset_so_far += modify.len(); + offsets.push(offset_so_far); + modify_index.push((*hash, *ts, modify[0])); + } + let modify_rows = modify_rows.finish(); + let args = get_arrayref_at_indices(&args, &modify_rows)?; + modify_index.iter().zip(offsets.windows(2)).try_for_each( + |((hash, ts, row), offset)| { + let (offset, length) = (offset[0], offset[1] - offset[0]); + let sliced_arrays: Vec = args + .iter() + .map(|array| array.slice(offset, length)) + .collect(); + let accumulators_map = + self.series_map.entry(*hash).or_insert_with(|| SeriesState { + row: by_rows.row(*row as usize).owned(), + align_ts_accumulator: HashMap::new(), + }); + match accumulators_map.align_ts_accumulator.entry(*ts) { + Entry::Occupied(mut e) => { + let accumulators = e.get_mut(); + accumulators[i].update_batch(&sliced_arrays) + } + Entry::Vacant(e) => { + self.output_num_rows += 1; + let mut accumulators = self + .range_exec + .iter() + .map(|range| range.expr.create_accumulator()) + .collect::>>()?; + let result = accumulators[i].update_batch(&sliced_arrays); + e.insert(accumulators); + result + } + } + }, + )?; + } + Ok(()) + } + + fn generate_output(&mut self) -> DfResult { + let _timer = self.metric.elapsed_compute().timer(); + if self.series_map.is_empty() { + return Ok(RecordBatch::new_empty(self.schema.clone())); + } + // 1 for time index column + let mut columns: Vec> = + Vec::with_capacity(1 + self.range_exec.len() + self.by.len()); + let mut ts_builder = TimestampMillisecondBuilder::with_capacity(self.output_num_rows); + let mut all_scalar = vec![vec![]; self.range_exec.len()]; + let mut by_rows = Vec::with_capacity(self.output_num_rows); + for SeriesState { + row, + align_ts_accumulator, + } in self.series_map.values() + { + for (ts, accumulators) in align_ts_accumulator { + for (i, accumulator) in accumulators.iter().enumerate() { + all_scalar[i].push(accumulator.evaluate()?); + } + by_rows.push(row.row()); + ts_builder.append_value(*ts); + } + } + for column_scalar in all_scalar { + columns.push(ScalarValue::iter_to_array(column_scalar)?); + } + let ts_column = ts_builder.finish(); + // output schema follow the order of range expr | time index | by columns + let ts_column = compute::cast(&ts_column, self.schema.field(columns.len()).data_type())?; + columns.push(ts_column); + columns.extend(self.row_converter.convert_rows(by_rows)?); + Ok(RecordBatch::try_new(self.schema(), columns)?) + } +} + +enum ExecutionState { + ReadingInput, + ProducingOutput, + Done, } impl RecordBatchStream for RangeSelectStream { @@ -252,12 +628,234 @@ impl Stream for RangeSelectStream { type Item = DataFusionResult; fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - match self.input.poll_next_unpin(cx) { - Poll::Ready(Some(Ok(_batch))) => { - Poll::Ready(Some(Ok(RecordBatch::new_empty(self.schema.clone())))) + loop { + match self.exec_state { + ExecutionState::ReadingInput => { + match ready!(self.input.poll_next_unpin(cx)) { + // new batch to aggregate + Some(Ok(batch)) => { + if let Err(e) = self.update_range_context(batch) { + return Poll::Ready(Some(Err(e))); + } + } + // inner had error, return to caller + Some(Err(e)) => return Poll::Ready(Some(Err(e))), + // inner is done, producing output + None => { + self.exec_state = ExecutionState::ProducingOutput; + } + } + } + ExecutionState::ProducingOutput => { + let result = self.generate_output(); + match result { + // made output + Ok(batch) => { + self.exec_state = ExecutionState::Done; + return Poll::Ready(Some(Ok(batch))); + } + // error making output + Err(error) => return Poll::Ready(Some(Err(error))), + } + } + ExecutionState::Done => return Poll::Ready(None), } - Poll::Ready(other) => Poll::Ready(other), - Poll::Pending => Poll::Pending, } } } + +#[cfg(test)] +mod test { + use arrow_schema::SortOptions; + use datafusion::arrow::datatypes::{ + ArrowPrimitiveType, DataType, Field, Schema, TimestampMillisecondType, + }; + use datafusion::physical_plan::memory::MemoryExec; + use datafusion::physical_plan::sorts::sort::SortExec; + use datafusion::prelude::SessionContext; + use datafusion_physical_expr::expressions::{self, Column}; + use datafusion_physical_expr::PhysicalSortExpr; + use datatypes::arrow::array::{Int64Array, TimestampMillisecondArray}; + use datatypes::arrow_array::StringArray; + + use super::*; + + const TIME_INDEX_COLUMN: &str = "timestamp"; + + fn prepare_test_data() -> MemoryExec { + let schema = Arc::new(Schema::new(vec![ + Field::new(TIME_INDEX_COLUMN, TimestampMillisecondType::DATA_TYPE, true), + Field::new("value", DataType::Int64, true), + Field::new("host", DataType::Utf8, true), + ])); + let timestamp_column = Arc::new(TimestampMillisecondArray::from(vec![ + // host 1 every 5s + 0, 5_000, 10_000, 15_000, 20_000, 25_000, 30_000, 35_000, 40_000, + // host 2 every 5s + 0, 5_000, 10_000, 15_000, 20_000, 25_000, 30_000, 35_000, 40_000, + ])) as _; + let values = vec![ + 0, 1, 2, 3, 4, 5, 6, 7, 8, // data for host 1 + 9, 10, 11, 12, 13, 14, 15, 16, 17, // data for host 2 + ]; + let mut host = vec!["host1"; 9]; + host.extend(vec!["host2"; 9]); + let value_column = Arc::new(Int64Array::from(values)) as _; + let host_column = Arc::new(StringArray::from(host)) as _; + let data = RecordBatch::try_new( + schema.clone(), + vec![timestamp_column, value_column, host_column], + ) + .unwrap(); + + MemoryExec::try_new(&[vec![data]], schema, None).unwrap() + } + + async fn do_range_select_test( + range1: Millisecond, + range2: Millisecond, + align: Millisecond, + expected: String, + ) { + let memory_exec = Arc::new(prepare_test_data()); + let range_select_exec = Arc::new(RangeSelectExec { + input: memory_exec, + range_exec: vec![ + RangeFnExec { + expr: Arc::new(expressions::Min::new( + Arc::new(Column::new("value", 1)), + "MIN(value)", + DataType::Int64, + )), + args: vec![Arc::new(Column::new("value", 1))], + range: range1, + }, + RangeFnExec { + expr: Arc::new(expressions::Max::new( + Arc::new(Column::new("value", 1)), + "MAX(value)", + DataType::Int64, + )), + args: vec![Arc::new(Column::new("value", 1))], + range: range2, + }, + ], + align, + by: vec![Arc::new(Column::new("host", 2))], + time_index: TIME_INDEX_COLUMN.to_string(), + schema: Arc::new(Schema::new(vec![ + Field::new("MIN(value)", DataType::Int64, true), + Field::new("MAX(value)", DataType::Int64, true), + Field::new(TIME_INDEX_COLUMN, TimestampMillisecondType::DATA_TYPE, true), + Field::new("host", DataType::Utf8, true), + ])), + by_schema: Arc::new(Schema::new(vec![Field::new("host", DataType::Utf8, true)])), + metric: ExecutionPlanMetricsSet::new(), + }); + let sort_exec = SortExec::new( + vec![ + PhysicalSortExpr { + expr: Arc::new(Column::new("host", 3)), + options: SortOptions { + descending: false, + nulls_first: true, + }, + }, + PhysicalSortExpr { + expr: Arc::new(Column::new(TIME_INDEX_COLUMN, 2)), + options: SortOptions { + descending: false, + nulls_first: true, + }, + }, + ], + range_select_exec, + ); + let session_context = SessionContext::default(); + let result = + datafusion::physical_plan::collect(Arc::new(sort_exec), session_context.task_ctx()) + .await + .unwrap(); + + let result_literal = datatypes::arrow::util::pretty::pretty_format_batches(&result) + .unwrap() + .to_string(); + + assert_eq!(result_literal, expected); + } + + #[tokio::test] + async fn range_10s_align_5s() { + let expected = String::from( + "+------------+------------+---------------------+-------+\ + \n| MIN(value) | MAX(value) | timestamp | host |\ + \n+------------+------------+---------------------+-------+\ + \n| 0 | 0 | 1970-01-01T00:00:00 | host1 |\ + \n| 0 | 1 | 1970-01-01T00:00:05 | host1 |\ + \n| 1 | 2 | 1970-01-01T00:00:10 | host1 |\ + \n| 2 | 3 | 1970-01-01T00:00:15 | host1 |\ + \n| 3 | 4 | 1970-01-01T00:00:20 | host1 |\ + \n| 4 | 5 | 1970-01-01T00:00:25 | host1 |\ + \n| 5 | 6 | 1970-01-01T00:00:30 | host1 |\ + \n| 6 | 7 | 1970-01-01T00:00:35 | host1 |\ + \n| 7 | 8 | 1970-01-01T00:00:40 | host1 |\ + \n| 8 | 8 | 1970-01-01T00:00:45 | host1 |\ + \n| 9 | 9 | 1970-01-01T00:00:00 | host2 |\ + \n| 9 | 10 | 1970-01-01T00:00:05 | host2 |\ + \n| 10 | 11 | 1970-01-01T00:00:10 | host2 |\ + \n| 11 | 12 | 1970-01-01T00:00:15 | host2 |\ + \n| 12 | 13 | 1970-01-01T00:00:20 | host2 |\ + \n| 13 | 14 | 1970-01-01T00:00:25 | host2 |\ + \n| 14 | 15 | 1970-01-01T00:00:30 | host2 |\ + \n| 15 | 16 | 1970-01-01T00:00:35 | host2 |\ + \n| 16 | 17 | 1970-01-01T00:00:40 | host2 |\ + \n| 17 | 17 | 1970-01-01T00:00:45 | host2 |\ + \n+------------+------------+---------------------+-------+", + ); + do_range_select_test(10_000, 10_000, 5_000, expected).await; + } + + #[tokio::test] + async fn range_10s_align_1000s() { + let expected = String::from( + "+------------+------------+---------------------+-------+\ + \n| MIN(value) | MAX(value) | timestamp | host |\ + \n+------------+------------+---------------------+-------+\ + \n| 0 | 0 | 1970-01-01T00:00:00 | host1 |\ + \n| 9 | 9 | 1970-01-01T00:00:00 | host2 |\ + \n+------------+------------+---------------------+-------+", + ); + do_range_select_test(10_000, 10_000, 1_000_000, expected).await; + } + + #[tokio::test] + async fn range_10s_5s_align_5s() { + let expected = String::from( + "+------------+------------+---------------------+-------+\ + \n| MIN(value) | MAX(value) | timestamp | host |\ + \n+------------+------------+---------------------+-------+\ + \n| 0 | 0 | 1970-01-01T00:00:00 | host1 |\ + \n| 0 | 1 | 1970-01-01T00:00:05 | host1 |\ + \n| 1 | 2 | 1970-01-01T00:00:10 | host1 |\ + \n| 2 | 3 | 1970-01-01T00:00:15 | host1 |\ + \n| 3 | 4 | 1970-01-01T00:00:20 | host1 |\ + \n| 4 | 5 | 1970-01-01T00:00:25 | host1 |\ + \n| 5 | 6 | 1970-01-01T00:00:30 | host1 |\ + \n| 6 | 7 | 1970-01-01T00:00:35 | host1 |\ + \n| 7 | 8 | 1970-01-01T00:00:40 | host1 |\ + \n| 8 | | 1970-01-01T00:00:45 | host1 |\ + \n| 9 | 9 | 1970-01-01T00:00:00 | host2 |\ + \n| 9 | 10 | 1970-01-01T00:00:05 | host2 |\ + \n| 10 | 11 | 1970-01-01T00:00:10 | host2 |\ + \n| 11 | 12 | 1970-01-01T00:00:15 | host2 |\ + \n| 12 | 13 | 1970-01-01T00:00:20 | host2 |\ + \n| 13 | 14 | 1970-01-01T00:00:25 | host2 |\ + \n| 14 | 15 | 1970-01-01T00:00:30 | host2 |\ + \n| 15 | 16 | 1970-01-01T00:00:35 | host2 |\ + \n| 16 | 17 | 1970-01-01T00:00:40 | host2 |\ + \n| 17 | | 1970-01-01T00:00:45 | host2 |\ + \n+------------+------------+---------------------+-------+", + ); + do_range_select_test(10_000, 5_000, 5_000, expected).await; + } +} diff --git a/src/query/src/range_select/plan_rewrite.rs b/src/query/src/range_select/plan_rewrite.rs index 40f3400946a..89e9a908ce5 100644 --- a/src/query/src/range_select/plan_rewrite.rs +++ b/src/query/src/range_select/plan_rewrite.rs @@ -71,7 +71,9 @@ impl<'a> RangeExprRewriter<'a> { fn parse_str_expr(args: &[Expr], i: usize) -> DFResult<&str> { match args.get(i) { Some(Expr::Literal(ScalarValue::Utf8(Some(str)))) => Ok(str.as_str()), - _ => Err(DataFusionError::Plan("Illegal str expr in range_fn".into())), + _ => Err(DataFusionError::Plan( + "Illegal argument in range select query".into(), + )), } } @@ -79,8 +81,18 @@ fn parse_expr_list(args: &[Expr], start: usize, len: usize) -> DFResult args[i].clone(), - _ => return Err(DataFusionError::Plan("Illegal expr in range_fn".into())), + Some( + Expr::Column(_) + | Expr::Literal(_) + | Expr::BinaryExpr(_) + | Expr::ScalarFunction(_) + | Expr::ScalarUDF(_), + ) => args[i].clone(), + _ => { + return Err(DataFusionError::Plan( + "Illegal expr argument in range select query".into(), + )) + } }); } Ok(outs) @@ -276,10 +288,7 @@ impl RangePlanRewriter { table: table_ref.to_string(), })?; // assert time_index's datatype is timestamp - if let ConcreteDataType::Timestamp(datatypes::types::TimestampType::Millisecond( - _, - )) = time_index_column.data_type - { + if let ConcreteDataType::Timestamp(_) = time_index_column.data_type { default_by = table .table_info() .meta diff --git a/tests/cases/standalone/common/select/range_select.result b/tests/cases/standalone/common/select/range_select.result new file mode 100644 index 00000000000..e39746bdb2e --- /dev/null +++ b/tests/cases/standalone/common/select/range_select.result @@ -0,0 +1,349 @@ +CREATE TABLE host ( + ts timestamp(3) time index, + host STRING PRIMARY KEY, + val DOUBLE, +); + +Affected Rows: 0 + +INSERT INTO TABLE host VALUES + (0, 'host1', 0.0), + (5000, 'host1', 1.0), + (10000, 'host1', 2.0), + (15000, 'host1', 3.0), + (20000, 'host1', 4.0), + (25000, 'host1', 5.0), + (30000, 'host1', 6.0), + (35000, 'host1', 7.0), + (40000, 'host1', 8.0), + (0, 'host2', 9.0), + (5000, 'host2', 10.0), + (10000, 'host2', 11.0), + (15000, 'host2', 12.0), + (20000, 'host2', 13.0), + (25000, 'host2', 14.0), + (30000, 'host2', 15.0), + (35000, 'host2', 16.0), + (40000, 'host2', 17.0); + +Affected Rows: 18 + +SELECT ts, host, min(val) RANGE '10s', max(val) RANGE '10s' FROM host ALIGN '5s' ORDER BY host, ts; + ++---------------+---------------+---------------------+-------+ +| MIN(host.val) | MAX(host.val) | ts | host | ++---------------+---------------+---------------------+-------+ +| 0.0 | 0.0 | 1970-01-01T00:00:00 | host1 | +| 0.0 | 1.0 | 1970-01-01T00:00:05 | host1 | +| 1.0 | 2.0 | 1970-01-01T00:00:10 | host1 | +| 2.0 | 3.0 | 1970-01-01T00:00:15 | host1 | +| 3.0 | 4.0 | 1970-01-01T00:00:20 | host1 | +| 4.0 | 5.0 | 1970-01-01T00:00:25 | host1 | +| 5.0 | 6.0 | 1970-01-01T00:00:30 | host1 | +| 6.0 | 7.0 | 1970-01-01T00:00:35 | host1 | +| 7.0 | 8.0 | 1970-01-01T00:00:40 | host1 | +| 8.0 | 8.0 | 1970-01-01T00:00:45 | host1 | +| 9.0 | 9.0 | 1970-01-01T00:00:00 | host2 | +| 9.0 | 10.0 | 1970-01-01T00:00:05 | host2 | +| 10.0 | 11.0 | 1970-01-01T00:00:10 | host2 | +| 11.0 | 12.0 | 1970-01-01T00:00:15 | host2 | +| 12.0 | 13.0 | 1970-01-01T00:00:20 | host2 | +| 13.0 | 14.0 | 1970-01-01T00:00:25 | host2 | +| 14.0 | 15.0 | 1970-01-01T00:00:30 | host2 | +| 15.0 | 16.0 | 1970-01-01T00:00:35 | host2 | +| 16.0 | 17.0 | 1970-01-01T00:00:40 | host2 | +| 17.0 | 17.0 | 1970-01-01T00:00:45 | host2 | ++---------------+---------------+---------------------+-------+ + +SELECT ts, host, min(val / 2.0)/2 RANGE '10s', max(val / 2.0)/2 RANGE '10s' FROM host ALIGN '5s' ORDER BY host, ts; + ++---------------------+-------+---------------------------------------+---------------------------------------+ +| ts | host | MIN(host.val / Float64(2)) / Int64(2) | MAX(host.val / Float64(2)) / Int64(2) | ++---------------------+-------+---------------------------------------+---------------------------------------+ +| 1970-01-01T00:00:00 | host1 | 0.0 | 0.0 | +| 1970-01-01T00:00:05 | host1 | 0.0 | 0.25 | +| 1970-01-01T00:00:10 | host1 | 0.25 | 0.5 | +| 1970-01-01T00:00:15 | host1 | 0.5 | 0.75 | +| 1970-01-01T00:00:20 | host1 | 0.75 | 1.0 | +| 1970-01-01T00:00:25 | host1 | 1.0 | 1.25 | +| 1970-01-01T00:00:30 | host1 | 1.25 | 1.5 | +| 1970-01-01T00:00:35 | host1 | 1.5 | 1.75 | +| 1970-01-01T00:00:40 | host1 | 1.75 | 2.0 | +| 1970-01-01T00:00:45 | host1 | 2.0 | 2.0 | +| 1970-01-01T00:00:00 | host2 | 2.25 | 2.25 | +| 1970-01-01T00:00:05 | host2 | 2.25 | 2.5 | +| 1970-01-01T00:00:10 | host2 | 2.5 | 2.75 | +| 1970-01-01T00:00:15 | host2 | 2.75 | 3.0 | +| 1970-01-01T00:00:20 | host2 | 3.0 | 3.25 | +| 1970-01-01T00:00:25 | host2 | 3.25 | 3.5 | +| 1970-01-01T00:00:30 | host2 | 3.5 | 3.75 | +| 1970-01-01T00:00:35 | host2 | 3.75 | 4.0 | +| 1970-01-01T00:00:40 | host2 | 4.0 | 4.25 | +| 1970-01-01T00:00:45 | host2 | 4.25 | 4.25 | ++---------------------+-------+---------------------------------------+---------------------------------------+ + +SELECT ts, host, covar(val, val) RANGE '10s' FROM host ALIGN '5s' ORDER BY host, ts; + ++-------------------------------+---------------------+-------+ +| COVARIANCE(host.val,host.val) | ts | host | ++-------------------------------+---------------------+-------+ +| | 1970-01-01T00:00:00 | host1 | +| 0.5 | 1970-01-01T00:00:05 | host1 | +| 0.5 | 1970-01-01T00:00:10 | host1 | +| 0.5 | 1970-01-01T00:00:15 | host1 | +| 0.5 | 1970-01-01T00:00:20 | host1 | +| 0.5 | 1970-01-01T00:00:25 | host1 | +| 0.5 | 1970-01-01T00:00:30 | host1 | +| 0.5 | 1970-01-01T00:00:35 | host1 | +| 0.5 | 1970-01-01T00:00:40 | host1 | +| | 1970-01-01T00:00:45 | host1 | +| | 1970-01-01T00:00:00 | host2 | +| 0.5 | 1970-01-01T00:00:05 | host2 | +| 0.5 | 1970-01-01T00:00:10 | host2 | +| 0.5 | 1970-01-01T00:00:15 | host2 | +| 0.5 | 1970-01-01T00:00:20 | host2 | +| 0.5 | 1970-01-01T00:00:25 | host2 | +| 0.5 | 1970-01-01T00:00:30 | host2 | +| 0.5 | 1970-01-01T00:00:35 | host2 | +| 0.5 | 1970-01-01T00:00:40 | host2 | +| | 1970-01-01T00:00:45 | host2 | ++-------------------------------+---------------------+-------+ + +SELECT ts, host, covar(sin(val), cos(val)) RANGE '10s' FROM host ALIGN '5s' ORDER BY host, ts; + ++-----------------------------------------+---------------------+-------+ +| COVARIANCE(sin(host.val),cos(host.val)) | ts | host | ++-----------------------------------------+---------------------+-------+ +| | 1970-01-01T00:00:00 | host1 | +| -0.1934111356975278 | 1970-01-01T00:00:05 | host1 | +| -0.032436271150495265 | 1970-01-01T00:00:10 | host1 | +| 0.2204076389548557 | 1970-01-01T00:00:15 | host1 | +| -0.15100761225328058 | 1970-01-01T00:00:20 | host1 | +| -0.09472495868737531 | 1970-01-01T00:00:25 | host1 | +| 0.2298465960529006 | 1970-01-01T00:00:30 | host1 | +| -0.09657490898971162 | 1970-01-01T00:00:35 | host1 | +| -0.1494679103211072 | 1970-01-01T00:00:40 | host1 | +| | 1970-01-01T00:00:45 | host1 | +| | 1970-01-01T00:00:00 | host2 | +| -0.03444910384248826 | 1970-01-01T00:00:05 | host2 | +| -0.19230433390872206 | 1970-01-01T00:00:10 | host2 | +| 0.19450278426332843 | 1970-01-01T00:00:15 | host2 | +| 0.03042089716713121 | 1970-01-01T00:00:20 | host2 | +| -0.21982190450538358 | 1970-01-01T00:00:25 | host2 | +| 0.1525354831602356 | 1970-01-01T00:00:30 | host2 | +| 0.09286758694873963 | 1970-01-01T00:00:35 | host2 | +| -0.22982858821320498 | 1970-01-01T00:00:40 | host2 | +| | 1970-01-01T00:00:45 | host2 | ++-----------------------------------------+---------------------+-------+ + +SELECT ts, host, covar((sin(val) + cos(val))/2.0 + 1.0, 2.0) RANGE '10s' FROM host ALIGN '5s' ORDER BY host, ts; + ++--------------------------------------------------------------------------------+---------------------+-------+ +| COVARIANCE(sin(host.val) + cos(host.val) / Float64(2) + Float64(1),Float64(2)) | ts | host | ++--------------------------------------------------------------------------------+---------------------+-------+ +| | 1970-01-01T00:00:00 | host1 | +| 0.0 | 1970-01-01T00:00:05 | host1 | +| 0.0 | 1970-01-01T00:00:10 | host1 | +| 0.0 | 1970-01-01T00:00:15 | host1 | +| 0.0 | 1970-01-01T00:00:20 | host1 | +| 0.0 | 1970-01-01T00:00:25 | host1 | +| 0.0 | 1970-01-01T00:00:30 | host1 | +| 0.0 | 1970-01-01T00:00:35 | host1 | +| 0.0 | 1970-01-01T00:00:40 | host1 | +| | 1970-01-01T00:00:45 | host1 | +| | 1970-01-01T00:00:00 | host2 | +| 0.0 | 1970-01-01T00:00:05 | host2 | +| 0.0 | 1970-01-01T00:00:10 | host2 | +| 0.0 | 1970-01-01T00:00:15 | host2 | +| 0.0 | 1970-01-01T00:00:20 | host2 | +| 0.0 | 1970-01-01T00:00:25 | host2 | +| 0.0 | 1970-01-01T00:00:30 | host2 | +| 0.0 | 1970-01-01T00:00:35 | host2 | +| 0.0 | 1970-01-01T00:00:40 | host2 | +| | 1970-01-01T00:00:45 | host2 | ++--------------------------------------------------------------------------------+---------------------+-------+ + +SELECT ts, host, min(val) RANGE '10s', max(val) RANGE '10s' FROM host ALIGN '1000s' ORDER BY host, ts; + ++---------------+---------------+---------------------+-------+ +| MIN(host.val) | MAX(host.val) | ts | host | ++---------------+---------------+---------------------+-------+ +| 0.0 | 0.0 | 1970-01-01T00:00:00 | host1 | +| 9.0 | 9.0 | 1970-01-01T00:00:00 | host2 | ++---------------+---------------+---------------------+-------+ + +SELECT ts, host, min(val) RANGE '10s', max(val) RANGE '5s' FROM host ALIGN '5s' ORDER BY host, ts; + ++---------------+---------------+---------------------+-------+ +| MIN(host.val) | MAX(host.val) | ts | host | ++---------------+---------------+---------------------+-------+ +| 0.0 | 0.0 | 1970-01-01T00:00:00 | host1 | +| 0.0 | 1.0 | 1970-01-01T00:00:05 | host1 | +| 1.0 | 2.0 | 1970-01-01T00:00:10 | host1 | +| 2.0 | 3.0 | 1970-01-01T00:00:15 | host1 | +| 3.0 | 4.0 | 1970-01-01T00:00:20 | host1 | +| 4.0 | 5.0 | 1970-01-01T00:00:25 | host1 | +| 5.0 | 6.0 | 1970-01-01T00:00:30 | host1 | +| 6.0 | 7.0 | 1970-01-01T00:00:35 | host1 | +| 7.0 | 8.0 | 1970-01-01T00:00:40 | host1 | +| 8.0 | | 1970-01-01T00:00:45 | host1 | +| 9.0 | 9.0 | 1970-01-01T00:00:00 | host2 | +| 9.0 | 10.0 | 1970-01-01T00:00:05 | host2 | +| 10.0 | 11.0 | 1970-01-01T00:00:10 | host2 | +| 11.0 | 12.0 | 1970-01-01T00:00:15 | host2 | +| 12.0 | 13.0 | 1970-01-01T00:00:20 | host2 | +| 13.0 | 14.0 | 1970-01-01T00:00:25 | host2 | +| 14.0 | 15.0 | 1970-01-01T00:00:30 | host2 | +| 15.0 | 16.0 | 1970-01-01T00:00:35 | host2 | +| 16.0 | 17.0 | 1970-01-01T00:00:40 | host2 | +| 17.0 | | 1970-01-01T00:00:45 | host2 | ++---------------+---------------+---------------------+-------+ + +SELECT ts, host, (min(val)+max(val))/4 RANGE '10s' FROM host ALIGN '5s' ORDER BY host, ts; + ++---------------------+-------+------------------------------------------+ +| ts | host | MIN(host.val) + MAX(host.val) / Int64(4) | ++---------------------+-------+------------------------------------------+ +| 1970-01-01T00:00:00 | host1 | 0.0 | +| 1970-01-01T00:00:05 | host1 | 0.25 | +| 1970-01-01T00:00:10 | host1 | 0.75 | +| 1970-01-01T00:00:15 | host1 | 1.25 | +| 1970-01-01T00:00:20 | host1 | 1.75 | +| 1970-01-01T00:00:25 | host1 | 2.25 | +| 1970-01-01T00:00:30 | host1 | 2.75 | +| 1970-01-01T00:00:35 | host1 | 3.25 | +| 1970-01-01T00:00:40 | host1 | 3.75 | +| 1970-01-01T00:00:45 | host1 | 4.0 | +| 1970-01-01T00:00:00 | host2 | 4.5 | +| 1970-01-01T00:00:05 | host2 | 4.75 | +| 1970-01-01T00:00:10 | host2 | 5.25 | +| 1970-01-01T00:00:15 | host2 | 5.75 | +| 1970-01-01T00:00:20 | host2 | 6.25 | +| 1970-01-01T00:00:25 | host2 | 6.75 | +| 1970-01-01T00:00:30 | host2 | 7.25 | +| 1970-01-01T00:00:35 | host2 | 7.75 | +| 1970-01-01T00:00:40 | host2 | 8.25 | +| 1970-01-01T00:00:45 | host2 | 8.5 | ++---------------------+-------+------------------------------------------+ + +SELECT ts, host, foo FROM (SELECT ts, host, (min(val)+max(val))/4 RANGE '10s' AS foo FROM host ALIGN '5s' ORDER BY host, ts) WHERE foo > 5 ORDER BY host, ts; + ++---------------------+-------+------+ +| ts | host | foo | ++---------------------+-------+------+ +| 1970-01-01T00:00:10 | host2 | 5.25 | +| 1970-01-01T00:00:15 | host2 | 5.75 | +| 1970-01-01T00:00:20 | host2 | 6.25 | +| 1970-01-01T00:00:25 | host2 | 6.75 | +| 1970-01-01T00:00:30 | host2 | 7.25 | +| 1970-01-01T00:00:35 | host2 | 7.75 | +| 1970-01-01T00:00:40 | host2 | 8.25 | +| 1970-01-01T00:00:45 | host2 | 8.5 | ++---------------------+-------+------+ + +SELECT ts, b, (min(c)+max(c))/4 RANGE '10s' FROM (SELECT ts, host AS b, val AS c FROM host WHERE val > 8.0) ALIGN '5s' BY (b) ORDER BY b, ts; + ++---------------------+-------+----------------------------+ +| ts | b | MIN(c) + MAX(c) / Int64(4) | ++---------------------+-------+----------------------------+ +| 1970-01-01T00:00:00 | host2 | 4.5 | +| 1970-01-01T00:00:05 | host2 | 4.75 | +| 1970-01-01T00:00:10 | host2 | 5.25 | +| 1970-01-01T00:00:15 | host2 | 5.75 | +| 1970-01-01T00:00:20 | host2 | 6.25 | +| 1970-01-01T00:00:25 | host2 | 6.75 | +| 1970-01-01T00:00:30 | host2 | 7.25 | +| 1970-01-01T00:00:35 | host2 | 7.75 | +| 1970-01-01T00:00:40 | host2 | 8.25 | +| 1970-01-01T00:00:45 | host2 | 8.5 | ++---------------------+-------+----------------------------+ + +-- Test Invalid cases +-- 1. error timestamp +SELECT min(val) RANGE 'not_time' FROM host ALIGN '5s'; + +Error: 2000(InvalidSyntax), sql parser error: not a valid duration string: not_time + +SELECT min(val) RANGE '5s' FROM host ALIGN 'not_time'; + +Error: 2000(InvalidSyntax), sql parser error: not a valid duration string: not_time + +-- 2.1 no range param +SELECT min(val) FROM host ALIGN '5s'; + +Error: 2000(InvalidSyntax), sql parser error: RANGE argument not found in min(val) + +SELECT min(val) RANGE '10s', max(val) FROM host ALIGN '5s'; + +Error: 2000(InvalidSyntax), sql parser error: RANGE argument not found in max(val) + +-- 2.2 no align param +SELECT min(val) RANGE '5s' FROM host; + +Error: 1003(Internal), Error during planning: Illegal argument in range select query + +DROP TABLE host; + +Affected Rows: 1 + +CREATE TABLE host_sec ( + ts timestamp(0) time index, + host STRING PRIMARY KEY, + val DOUBLE, +); + +Affected Rows: 0 + +INSERT INTO TABLE host_sec VALUES + (0, 'host1', 0.0), + (5, 'host1', 1.0), + (10, 'host1', 2.0), + (15, 'host1', 3.0), + (20, 'host1', 4.0), + (25, 'host1', 5.0), + (30, 'host1', 6.0), + (35, 'host1', 7.0), + (40, 'host1', 8.0), + (0, 'host2', 9.0), + (5, 'host2', 10.0), + (10, 'host2', 11.0), + (15, 'host2', 12.0), + (20, 'host2', 13.0), + (25, 'host2', 14.0), + (30, 'host2', 15.0), + (35, 'host2', 16.0), + (40, 'host2', 17.0); + +Affected Rows: 18 + +SELECT ts, host, min(val) RANGE '10s', max(val) RANGE '10s' FROM host_sec ALIGN '5s' ORDER BY host, ts; + ++-------------------+-------------------+---------------------+-------+ +| MIN(host_sec.val) | MAX(host_sec.val) | ts | host | ++-------------------+-------------------+---------------------+-------+ +| 0.0 | 0.0 | 1970-01-01T00:00:00 | host1 | +| 0.0 | 1.0 | 1970-01-01T00:00:05 | host1 | +| 1.0 | 2.0 | 1970-01-01T00:00:10 | host1 | +| 2.0 | 3.0 | 1970-01-01T00:00:15 | host1 | +| 3.0 | 4.0 | 1970-01-01T00:00:20 | host1 | +| 4.0 | 5.0 | 1970-01-01T00:00:25 | host1 | +| 5.0 | 6.0 | 1970-01-01T00:00:30 | host1 | +| 6.0 | 7.0 | 1970-01-01T00:00:35 | host1 | +| 7.0 | 8.0 | 1970-01-01T00:00:40 | host1 | +| 8.0 | 8.0 | 1970-01-01T00:00:45 | host1 | +| 9.0 | 9.0 | 1970-01-01T00:00:00 | host2 | +| 9.0 | 10.0 | 1970-01-01T00:00:05 | host2 | +| 10.0 | 11.0 | 1970-01-01T00:00:10 | host2 | +| 11.0 | 12.0 | 1970-01-01T00:00:15 | host2 | +| 12.0 | 13.0 | 1970-01-01T00:00:20 | host2 | +| 13.0 | 14.0 | 1970-01-01T00:00:25 | host2 | +| 14.0 | 15.0 | 1970-01-01T00:00:30 | host2 | +| 15.0 | 16.0 | 1970-01-01T00:00:35 | host2 | +| 16.0 | 17.0 | 1970-01-01T00:00:40 | host2 | +| 17.0 | 17.0 | 1970-01-01T00:00:45 | host2 | ++-------------------+-------------------+---------------------+-------+ + +DROP TABLE host_sec; + +Affected Rows: 1 + diff --git a/tests/cases/standalone/common/select/range_select.sql b/tests/cases/standalone/common/select/range_select.sql new file mode 100644 index 00000000000..ef472e92420 --- /dev/null +++ b/tests/cases/standalone/common/select/range_select.sql @@ -0,0 +1,95 @@ +CREATE TABLE host ( + ts timestamp(3) time index, + host STRING PRIMARY KEY, + val DOUBLE, +); + +INSERT INTO TABLE host VALUES + (0, 'host1', 0.0), + (5000, 'host1', 1.0), + (10000, 'host1', 2.0), + (15000, 'host1', 3.0), + (20000, 'host1', 4.0), + (25000, 'host1', 5.0), + (30000, 'host1', 6.0), + (35000, 'host1', 7.0), + (40000, 'host1', 8.0), + (0, 'host2', 9.0), + (5000, 'host2', 10.0), + (10000, 'host2', 11.0), + (15000, 'host2', 12.0), + (20000, 'host2', 13.0), + (25000, 'host2', 14.0), + (30000, 'host2', 15.0), + (35000, 'host2', 16.0), + (40000, 'host2', 17.0); + +SELECT ts, host, min(val) RANGE '10s', max(val) RANGE '10s' FROM host ALIGN '5s' ORDER BY host, ts; + +SELECT ts, host, min(val / 2.0)/2 RANGE '10s', max(val / 2.0)/2 RANGE '10s' FROM host ALIGN '5s' ORDER BY host, ts; + +SELECT ts, host, covar(val, val) RANGE '10s' FROM host ALIGN '5s' ORDER BY host, ts; + +SELECT ts, host, covar(sin(val), cos(val)) RANGE '10s' FROM host ALIGN '5s' ORDER BY host, ts; + +SELECT ts, host, covar((sin(val) + cos(val))/2.0 + 1.0, 2.0) RANGE '10s' FROM host ALIGN '5s' ORDER BY host, ts; + +SELECT ts, host, min(val) RANGE '10s', max(val) RANGE '10s' FROM host ALIGN '1000s' ORDER BY host, ts; + +SELECT ts, host, min(val) RANGE '10s', max(val) RANGE '5s' FROM host ALIGN '5s' ORDER BY host, ts; + +SELECT ts, host, (min(val)+max(val))/4 RANGE '10s' FROM host ALIGN '5s' ORDER BY host, ts; + +SELECT ts, host, foo FROM (SELECT ts, host, (min(val)+max(val))/4 RANGE '10s' AS foo FROM host ALIGN '5s' ORDER BY host, ts) WHERE foo > 5 ORDER BY host, ts; + +SELECT ts, b, (min(c)+max(c))/4 RANGE '10s' FROM (SELECT ts, host AS b, val AS c FROM host WHERE val > 8.0) ALIGN '5s' BY (b) ORDER BY b, ts; + +-- Test Invalid cases + +-- 1. error timestamp + +SELECT min(val) RANGE 'not_time' FROM host ALIGN '5s'; + +SELECT min(val) RANGE '5s' FROM host ALIGN 'not_time'; + +-- 2.1 no range param + +SELECT min(val) FROM host ALIGN '5s'; + +SELECT min(val) RANGE '10s', max(val) FROM host ALIGN '5s'; + +-- 2.2 no align param + +SELECT min(val) RANGE '5s' FROM host; + +DROP TABLE host; + +CREATE TABLE host_sec ( + ts timestamp(0) time index, + host STRING PRIMARY KEY, + val DOUBLE, +); + +INSERT INTO TABLE host_sec VALUES + (0, 'host1', 0.0), + (5, 'host1', 1.0), + (10, 'host1', 2.0), + (15, 'host1', 3.0), + (20, 'host1', 4.0), + (25, 'host1', 5.0), + (30, 'host1', 6.0), + (35, 'host1', 7.0), + (40, 'host1', 8.0), + (0, 'host2', 9.0), + (5, 'host2', 10.0), + (10, 'host2', 11.0), + (15, 'host2', 12.0), + (20, 'host2', 13.0), + (25, 'host2', 14.0), + (30, 'host2', 15.0), + (35, 'host2', 16.0), + (40, 'host2', 17.0); + +SELECT ts, host, min(val) RANGE '10s', max(val) RANGE '10s' FROM host_sec ALIGN '5s' ORDER BY host, ts; + +DROP TABLE host_sec;