diff --git a/src/common/query/src/series_frame/tests/lifecycle.rs b/src/common/query/src/series_frame/tests/lifecycle.rs index 1c487fc73f2..0e467730403 100644 --- a/src/common/query/src/series_frame/tests/lifecycle.rs +++ b/src/common/query/src/series_frame/tests/lifecycle.rs @@ -180,9 +180,7 @@ async fn streaming_round_trip_preserves_sliced_multifield_payload_and_owners() { true, ) .unwrap(); - let input_times_ptr = match &chunk.time { - RawTimeAxis::Explicit(values) => values.as_ptr(), - }; + let input_times_ptr = chunk.time.as_explicit().unwrap().as_ptr(); let input_float_ptr = float_a.values().as_ptr(); let input_hist_indices_ptr = indices.values().as_ptr(); let input_mixed_float_ptr = mixed_float_owner.values().as_ptr(); @@ -255,7 +253,7 @@ async fn streaming_round_trip_preserves_sliced_multifield_payload_and_owners() { .to_data() .ptr_eq(&sliced_table.as_arrow().to_data()) ); - let RawTimeAxis::Explicit(first_times) = &output[0].time; + let first_times = output[0].time.as_explicit().unwrap(); assert_eq!(first_times.as_ptr(), input_times_ptr); drop(decoded); diff --git a/src/datatypes/src/series/raw.rs b/src/datatypes/src/series/raw.rs index 56baeca1d30..2bad48e7b3b 100644 --- a/src/datatypes/src/series/raw.rs +++ b/src/datatypes/src/series/raw.rs @@ -23,6 +23,7 @@ use crate::series::samples::{SeriesFields, SeriesSchema}; use crate::series::{MetricName, invalid_series}; /// Explicit source timestamps in milliseconds for a raw series chunk. +#[non_exhaustive] #[derive(Clone, Debug)] pub enum RawTimeAxis { /// Source timestamps in strictly increasing order. @@ -32,10 +33,29 @@ pub enum RawTimeAxis { impl RawTimeAxis { /// Validates and wraps strictly increasing timestamps; an empty axis is valid on its own. pub fn try_new(timestamps: ScalarBuffer) -> Result { - if timestamps.windows(2).any(|pair| pair[0] >= pair[1]) { - return invalid_series("raw timestamps must be strictly increasing"); + let axis = Self::Explicit(timestamps); + axis.try_len()?; + Ok(axis) + } + + /// Returns the explicit timestamp buffer, if this axis uses explicit timestamps. + /// + /// Future axis variants may not have an explicit buffer. + pub fn as_explicit(&self) -> Option<&ScalarBuffer> { + match self { + Self::Explicit(values) => Some(values), + } + } + + fn try_len(&self) -> Result { + match self { + Self::Explicit(values) => { + if values.windows(2).any(|pair| pair[0] >= pair[1]) { + return invalid_series("raw timestamps must be strictly increasing"); + } + Ok(values.len()) + } } - Ok(Self::Explicit(timestamps)) } /// Returns the number of timestamps. @@ -128,10 +148,7 @@ impl RawSeriesChunk { if time.is_empty() { return invalid_series("raw series chunks must not be empty"); } - let RawTimeAxis::Explicit(timestamps) = &time; - if timestamps.windows(2).any(|pair| pair[0] >= pair[1]) { - return invalid_series("raw series timestamps must be strictly increasing"); - } + time.try_len()?; Ok(Self { read_key, metric, diff --git a/src/datatypes/src/series/raw/tests.rs b/src/datatypes/src/series/raw/tests.rs index 3229f647fd9..ee982148d73 100644 --- a/src/datatypes/src/series/raw/tests.rs +++ b/src/datatypes/src/series/raw/tests.rs @@ -133,7 +133,7 @@ fn sliced_axes_and_chunks_keep_shared_buffers_and_values_alive() { let time_owner: ScalarBuffer = vec![0, 10, 20, 30, 40, 80].into(); let time_source = RawTimeAxis::try_new(time_owner.clone()).unwrap(); let time = time_source.slice(1, 4).unwrap(); - let RawTimeAxis::Explicit(time_buffer) = &time; + let time_buffer = time.as_explicit().unwrap(); assert_eq!(time_buffer.as_ptr(), time_owner.as_ptr().wrapping_add(1)); let nan_payload = f64::from_bits(0x7ff8_0000_0000_0042); diff --git a/src/store-api/src/region_engine.rs b/src/store-api/src/region_engine.rs index a30f0cbd701..315133cdcf8 100644 --- a/src/store-api/src/region_engine.rs +++ b/src/store-api/src/region_engine.rs @@ -550,17 +550,22 @@ pub trait RegionScanner: Debug + DisplayAs + Send { /// Scans a partition as raw series when this source supports that path. /// - /// `None` means raw capability is absent before execution and permits a legacy scan instead. - /// `Some` selects this entrypoint, including for an empty successful stream. Errors propagate; - /// callers must not replay a failed raw scan through the legacy path. Use only one entrypoint - /// for a given scanner partition per execution. + /// Implementations may return `None` when raw capability is absent before execution, permitting + /// a legacy scan instead. `Some` selects this entrypoint, including for an empty successful + /// stream. Errors propagate; callers must not replay a failed raw scan through the legacy path. + /// Use only one entrypoint for a given scanner partition per execution. + /// + /// # Panics + /// + /// The default implementation panics. Implementations must override this method to support + /// raw series scanning. fn scan_series_partition( &self, _ctx: &QueryScanContext, _metrics_set: &ExecutionPlanMetricsSet, _partition: usize, ) -> Result, BoxedError> { - Ok(None) + todo!() } /// Check if there is any predicate exclude region partition exprs that may be executed in this scanner. @@ -813,8 +818,16 @@ mod tests { use super::*; #[tokio::test] - async fn single_partition_scanner_preserves_legacy_fallback_after_raw_capability_miss() { - let schema = Arc::new(datatypes::schema::Schema::new(vec![ + async fn single_partition_scanner_raw_scan_panics_and_preserves_legacy_read() { + use std::panic::{AssertUnwindSafe, catch_unwind}; + + use api::v1::SemanticType; + use datatypes::arrow::array::Int32Array; + use datatypes::schema::Schema as ArrowSchema; + + use crate::metadata::{ColumnMetadata, RegionMetadataBuilder}; + + let schema = Arc::new(ArrowSchema::new(vec![ ColumnSchema::new("value", ConcreteDataType::int32_datatype(), false), ColumnSchema::new( "ts", @@ -828,20 +841,19 @@ mod tests { ]; let batch = RecordBatch::new(schema.clone(), columns).unwrap(); let batches = RecordBatches::try_new(schema, vec![batch]).unwrap(); - let mut metadata_builder = - crate::metadata::RegionMetadataBuilder::new(RegionId::new(1234, 1)); - metadata_builder.push_column_metadata(crate::metadata::ColumnMetadata { + let mut metadata_builder = RegionMetadataBuilder::new(RegionId::new(1234, 1)); + metadata_builder.push_column_metadata(ColumnMetadata { column_schema: ColumnSchema::new("value", ConcreteDataType::int32_datatype(), false), - semantic_type: api::v1::SemanticType::Field, + semantic_type: SemanticType::Field, column_id: 1, }); - metadata_builder.push_column_metadata(crate::metadata::ColumnMetadata { + metadata_builder.push_column_metadata(ColumnMetadata { column_schema: ColumnSchema::new( "ts", ConcreteDataType::timestamp_millisecond_datatype(), false, ), - semantic_type: api::v1::SemanticType::Timestamp, + semantic_type: SemanticType::Timestamp, column_id: 2, }); let metadata = Arc::new(metadata_builder.build().unwrap()); @@ -849,12 +861,11 @@ mod tests { let ctx = QueryScanContext::default(); let metrics = ExecutionPlanMetricsSet::new(); - assert!( - scanner - .scan_series_partition(&ctx, &metrics, 0) - .unwrap() - .is_none() - ); + // SinglePartitionScanner uses the default raw-scan implementation, which panics. + let raw_scan = catch_unwind(AssertUnwindSafe(|| { + scanner.scan_series_partition(&ctx, &metrics, 0) + })); + assert!(raw_scan.is_err()); let legacy = scanner.scan_partition(&ctx, &metrics, 0).unwrap(); let legacy_batches = legacy.try_collect::>().await.unwrap(); assert_eq!(legacy_batches.len(), 1); @@ -864,19 +875,11 @@ mod tests { .column(0) .as_ref() .as_any() - .downcast_ref::() + .downcast_ref::() .unwrap() .value(0), 42 ); - - let second = scanner - .scan_partition(&ctx, &metrics, 0) - .unwrap() - .try_collect::>() - .await - .unwrap(); - assert!(second.is_empty()); } #[test]