From 4ad0ce71fe5f1caf4c34947c4a0a7fa862d2cb0b Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Thu, 15 Oct 2026 19:35:48 +0800 Subject: [PATCH 1/2] fix(store-api): mark default raw series scan as todo Leave the raw entrypoint explicitly unimplemented until a scanner overrides it. Document the default panic and keep the real scanner regression proving that its legacy input remains readable after the placeholder is caught. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/store-api/src/region_engine.rs | 27 +++++++++++++++------------ 1 file changed, 15 insertions(+), 12 deletions(-) diff --git a/src/store-api/src/region_engine.rs b/src/store-api/src/region_engine.rs index 82183f77ec3..746eacddc51 100644 --- a/src/store-api/src/region_engine.rs +++ b/src/store-api/src/region_engine.rs @@ -548,17 +548,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. @@ -811,7 +816,7 @@ mod tests { use super::*; #[tokio::test] - async fn single_partition_scanner_preserves_legacy_fallback_after_raw_capability_miss() { + async fn single_partition_scanner_raw_scan_panics_and_preserves_legacy_read() { let schema = Arc::new(datatypes::schema::Schema::new(vec![ ColumnSchema::new("value", ConcreteDataType::int32_datatype(), false), ColumnSchema::new( @@ -847,12 +852,10 @@ mod tests { let ctx = QueryScanContext::default(); let metrics = ExecutionPlanMetricsSet::new(); - assert!( - scanner - .scan_series_partition(&ctx, &metrics, 0) - .unwrap() - .is_none() - ); + let raw_scan = std::panic::catch_unwind(std::panic::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); From 5fa4fed582381a7c7e0494b32eec49b61c9ad2ab Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Thu, 15 Oct 2026 20:53:35 +0800 Subject: [PATCH 2/2] fix(datatypes): share raw axis validation and expose explicit buffer Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/datatypes/src/series/raw.rs | 31 ++++++++++++++++++------ src/datatypes/src/series/raw/tests.rs | 2 +- src/store-api/src/region_engine.rs | 34 +++++++++++++-------------- 3 files changed, 42 insertions(+), 25 deletions(-) diff --git a/src/datatypes/src/series/raw.rs b/src/datatypes/src/series/raw.rs index 1ff8cc873e0..6c5b04ceac7 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. @@ -125,10 +145,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 746eacddc51..48017c94f0f 100644 --- a/src/store-api/src/region_engine.rs +++ b/src/store-api/src/region_engine.rs @@ -817,7 +817,15 @@ mod tests { #[tokio::test] async fn single_partition_scanner_raw_scan_panics_and_preserves_legacy_read() { - let schema = Arc::new(datatypes::schema::Schema::new(vec![ + 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", @@ -831,20 +839,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()); @@ -852,7 +859,8 @@ mod tests { let ctx = QueryScanContext::default(); let metrics = ExecutionPlanMetricsSet::new(); - let raw_scan = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + // 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()); @@ -865,19 +873,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]