fix(query): preserve raw time axis lifecycle proof across contract update

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-10-15 21:19:20 +08:00
4 changed files with 58 additions and 40 deletions
@@ -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);
+24 -7
View File
@@ -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<i64>) -> Result<Self> {
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<i64>> {
match self {
Self::Explicit(values) => Some(values),
}
}
fn try_len(&self) -> Result<usize> {
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,
+1 -1
View File
@@ -133,7 +133,7 @@ fn sliced_axes_and_chunks_keep_shared_buffers_and_values_alive() {
let time_owner: ScalarBuffer<i64> = 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);
+31 -28
View File
@@ -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<Option<SendableRawSeriesStream>, 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::<Vec<_>>().await.unwrap();
assert_eq!(legacy_batches.len(), 1);
@@ -864,19 +875,11 @@ mod tests {
.column(0)
.as_ref()
.as_any()
.downcast_ref::<datatypes::arrow::array::Int32Array>()
.downcast_ref::<Int32Array>()
.unwrap()
.value(0),
42
);
let second = scanner
.scan_partition(&ctx, &metrics, 0)
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
assert!(second.is_empty());
}
#[test]