From d47d28da2bc1cfbbb7e0edfb98c3acfcb33b214a Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Thu, 15 Oct 2026 20:31:21 +0800 Subject: [PATCH] style(mito): import raw series types by name Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/metric-engine/src/engine/read.rs | 9 +-- src/mito2/src/engine/scan_test.rs | 44 +++++++----- src/mito2/src/read/series_scan.rs | 38 +++++----- src/mito2/src/read/series_scan/raw.rs | 30 ++++---- src/mito2/src/read/series_scan/raw/tests.rs | 80 ++++++++++----------- 5 files changed, 104 insertions(+), 97 deletions(-) diff --git a/src/metric-engine/src/engine/read.rs b/src/metric-engine/src/engine/read.rs index 83d8ce899c2..1e9095ead09 100644 --- a/src/metric-engine/src/engine/read.rs +++ b/src/metric-engine/src/engine/read.rs @@ -306,10 +306,11 @@ impl MetricEngineInner { #[cfg(test)] mod test { + use std::collections::HashSet; use std::fmt; use api::v1::value::ValueData; - use api::v1::{Row, Rows, Value}; + use api::v1::{ColumnSchema, Row, Rows, Value}; use common_error::ext::ErrorExt; use common_error::status_code::StatusCode; use datafusion::physical_plan::DisplayFormatType; @@ -587,7 +588,7 @@ mod test { .await .unwrap(); - let write = |schema: Vec, tag_values: &[&[&str]], start: i64| { + let write = |schema: Vec, tag_values: &[&[&str]], start: i64| { let rows = tag_values .iter() .enumerate() @@ -783,8 +784,8 @@ mod test { let scan_context = QueryScanContext::default(); let scan_metrics = ExecutionPlanMetricsSet::default(); let mut samples = Vec::new(); - let mut keys = std::collections::HashSet::new(); - let mut decoded_jobs = std::collections::HashSet::new(); + let mut keys = HashSet::new(); + let mut decoded_jobs = HashSet::new(); let mut last_markers = 0; for partition in 0..partitions { let stream = scanner diff --git a/src/mito2/src/engine/scan_test.rs b/src/mito2/src/engine/scan_test.rs index 8bf838a92d2..9a399ba78b2 100644 --- a/src/mito2/src/engine/scan_test.rs +++ b/src/mito2/src/engine/scan_test.rs @@ -4763,12 +4763,18 @@ fn scan_field_values(batches: &[common_recordbatch::RecordBatch]) -> Vec { /// Raw Mito output contains candidates after projection, filtering, and MVCC read processing. #[tokio::test] async fn test_mito_handle_query_raw_series_candidate_projection_and_mvcc() { + use std::collections::HashSet; + + use api::v1::{Row, Value}; + use common_recordbatch::filter; use datatypes::arrow::array::Array; use datatypes::series::raw::{RawSeriesOrdering, RawTimeAxis}; use datatypes::series::samples::SampleRef; use datatypes::value::Value as DatumValue; use mito_codec::row_converter::build_primary_key_codec; use store_api::region_engine::{BoundTag, QueryScanContext, RawLabelSource}; + use store_api::region_request::RegionDeleteRequest; + use table::predicate::Predicate; let mut env = TestEnv::with_prefix("test_mito_handle_query_raw_series_candidate_projection_and_mvcc") @@ -4781,21 +4787,21 @@ async fn test_mito_handle_query_raw_series_candidate_projection_and_mvcc() { .handle_request(region_id, RegionRequest::Create(request)) .await .unwrap(); - let row = |tag: &str, f0: Option, f1: Option, gate: f64, ts: i64| api::v1::Row { + let row = |tag: &str, f0: Option, f1: Option, gate: f64, ts: i64| Row { values: vec![ - api::v1::Value { + Value { value_data: Some(ValueData::StringValue(tag.to_string())), }, - api::v1::Value { + Value { value_data: f0.map(ValueData::F64Value), }, - api::v1::Value { + Value { value_data: f1.map(ValueData::F64Value), }, - api::v1::Value { + Value { value_data: Some(ValueData::F64Value(gate)), }, - api::v1::Value { + Value { value_data: Some(ValueData::TimestampMillisecondValue(ts)), }, ], @@ -4836,17 +4842,17 @@ async fn test_mito_handle_query_raw_series_candidate_projection_and_mvcc() { engine .handle_request( region_id, - RegionRequest::Delete(store_api::region_request::RegionDeleteRequest { + RegionRequest::Delete(RegionDeleteRequest { rows: Rows { schema: test_util::delete_rows_schema( &CreateRequestBuilder::new().field_num(3).build(), ), - rows: vec![api::v1::Row { + rows: vec![Row { values: vec![ - api::v1::Value { + Value { value_data: Some(ValueData::StringValue("A".to_string())), }, - api::v1::Value { + Value { value_data: Some(ValueData::TimestampMillisecondValue(2000)), }, ], @@ -4912,7 +4918,7 @@ async fn test_mito_handle_query_raw_series_candidate_projection_and_mvcc() { .map(|f| f.name.as_ref()) .collect::>() ); - let mut closed_keys = std::collections::HashSet::>::new(); + let mut closed_keys = HashSet::>::new(); while let Some(chunk) = stream.try_next().await.unwrap() { assert!(chunk.metric.is_none()); assert!( @@ -5008,7 +5014,7 @@ async fn test_mito_handle_query_raw_series_candidate_projection_and_mvcc() { } let candidate_count: usize = batches.iter().map(|batch| batch.num_rows()).sum(); assert_eq!(4, candidate_count); - let predicate = table::predicate::Predicate::to_physical_expr( + let predicate = Predicate::to_physical_expr( &col("field_2").gt(lit(0.0_f64)), legacy.schema().arrow_schema(), ) @@ -5017,7 +5023,7 @@ async fn test_mito_handle_query_raw_series_candidate_projection_and_mvcc() { .into_iter() .map(|batch| { let batch = batch.into_df_record_batch(); - let filtered = common_recordbatch::filter::batch_filter(&batch, &predicate).unwrap(); + let filtered = filter::batch_filter(&batch, &predicate).unwrap(); filtered.project(&[0, 2, 3, 4]).unwrap() }) .collect::>(); @@ -5100,7 +5106,7 @@ async fn test_mito_handle_query_raw_series_candidate_projection_and_mvcc() { async fn test_mito_raw_series_int64_predicate_uses_original_precision() { use datatypes::series::raw::RawTimeAxis; use datatypes::series::samples::SampleRef; - use store_api::region_engine::{QueryScanContext, RawLabelSource}; + use store_api::region_engine::{BoundTag, QueryScanContext, RawLabelSource}; let mut env = TestEnv::with_prefix("test_mito_raw_series_int64_predicate_uses_original_precision").await; @@ -5157,7 +5163,7 @@ async fn test_mito_raw_series_int64_predicate_uses_original_precision() { .column_metadatas .iter() .filter(|c| c.semantic_type == SemanticType::Tag) - .map(|c| store_api::region_engine::BoundTag { + .map(|c| BoundTag { column_id: c.column_id, name: Arc::from(c.column_schema.name.as_str()), }) @@ -5214,6 +5220,8 @@ async fn test_mito_raw_series_int64_predicate_uses_original_precision() { /// Invalid predicate bindings decline before consumption; explicit labels decline and legacy still reads. #[tokio::test] async fn test_mito_raw_series_invalid_binding_before_legacy_read() { + use datatypes::series::raw::RawTimeAxis; + use datatypes::series::samples::SampleRef; use store_api::region_engine::{BoundTag, QueryScanContext, RawLabelSource}; let mut env = TestEnv::with_prefix("test_mito_raw_series_invalid_binding_before_legacy_read").await; @@ -5307,10 +5315,8 @@ async fn test_mito_raw_series_invalid_binding_before_legacy_read() { let tuples = chunks .iter() .map(|chunk| { - let datatypes::series::raw::RawTimeAxis::Explicit(times) = &chunk.time; - let datatypes::series::samples::SampleRef::Float(value) = - chunk.fields.get(0).unwrap().sample(0) - else { + let RawTimeAxis::Explicit(times) = &chunk.time; + let SampleRef::Float(value) = chunk.fields.get(0).unwrap().sample(0) else { panic!("expected raw float sample"); }; (times[0], value) diff --git a/src/mito2/src/read/series_scan.rs b/src/mito2/src/read/series_scan.rs index 31698d6e652..85d95cf172b 100644 --- a/src/mito2/src/read/series_scan.rs +++ b/src/mito2/src/read/series_scan.rs @@ -16,6 +16,7 @@ mod raw; +use std::collections::HashSet; use std::fmt; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; @@ -30,22 +31,26 @@ use common_telemetry::warn; use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet; use datafusion::physical_plan::{DisplayAs, DisplayFormatType}; use datatypes::arrow::array::BinaryArray; +use datatypes::arrow::datatypes::DataType as ArrowDataType; use datatypes::arrow::record_batch::RecordBatch; use datatypes::schema::SchemaRef; +use datatypes::types::TimestampType; use futures::{StreamExt, TryStreamExt}; use smallvec::SmallVec; use snafu::{OptionExt, ResultExt, ensure}; use store_api::metadata::RegionMetadataRef; use store_api::region_engine::{ - PartitionRange, PrepareRequest, QueryScanContext, RegionScanner, ScannerProperties, + PartitionRange, PrepareRequest, QueryScanContext, RawLabelSource, RegionScanner, + ScannerProperties, SendableRawSeriesStream, }; +use table::predicate::Predicate; use tokio::sync::Semaphore; use tokio::sync::mpsc::error::{SendTimeoutError, TrySendError}; use tokio::sync::mpsc::{self, Receiver, Sender}; use crate::error::{ - Error, InvalidSenderSnafu, JoinSnafu, PartitionOutOfRangeSnafu, Result, ScanMultiTimesSnafu, - ScanSeriesSnafu, TooManyFilesToReadSnafu, + Error, InvalidRecordBatchSnafu, InvalidSenderSnafu, JoinSnafu, PartitionOutOfRangeSnafu, + Result, ScanMultiTimesSnafu, ScanSeriesSnafu, TooManyFilesToReadSnafu, }; use crate::read::ScannerMetrics; use crate::read::pruner::{PartitionPruner, Pruner, PrunerOptions}; @@ -571,13 +576,12 @@ impl RegionScanner for SeriesScan { ctx: &QueryScanContext, metrics_set: &ExecutionPlanMetricsSet, partition: usize, - label_source: &store_api::region_engine::RawLabelSource, - ) -> Result, BoxedError> { + label_source: &RawLabelSource, + ) -> Result, BoxedError> { use api::v1::SemanticType; use common_query::native_histogram::is_native_histogram_value_type; use datatypes::series::samples::{FieldSpec, SeriesSchema}; use store_api::metric_engine_consts::is_metric_engine_internal_column; - use store_api::region_engine::RawLabelSource; use store_api::storage::consts::{ReservedColumnId, is_internal_column}; let input = &self.stream_ctx.input; @@ -590,15 +594,15 @@ impl RegionScanner for SeriesScan { } => { if !Arc::ptr_eq(metadata, physical_metadata) { return Err(BoxedError::new( - crate::error::InvalidRecordBatchSnafu { + InvalidRecordBatchSnafu { reason: "raw label source metadata does not match scanner snapshot" .to_string(), } .build(), )); } - let mut ids = std::collections::HashSet::new(); - let mut names = std::collections::HashSet::new(); + let mut ids = HashSet::new(); + let mut names = HashSet::new(); for tag in logical_tags.iter() { let column = metadata.column_by_id(tag.column_id).filter(|column| { column.semantic_type == SemanticType::Tag @@ -613,7 +617,7 @@ impl RegionScanner for SeriesScan { }); if !valid || !ids.insert(tag.column_id) || !names.insert(tag.name.as_ref()) { return Err(BoxedError::new( - crate::error::InvalidRecordBatchSnafu { + InvalidRecordBatchSnafu { reason: "invalid raw label binding".to_string(), } .build(), @@ -641,7 +645,7 @@ impl RegionScanner for SeriesScan { .find(|(_, (id, _))| *id == column.column_id) else { return Err(BoxedError::new( - crate::error::InvalidRecordBatchSnafu { + InvalidRecordBatchSnafu { reason: format!( "selected field {} is missing from mapper batch schema", output.name @@ -658,10 +662,7 @@ impl RegionScanner for SeriesScan { name: Arc::from(output.name.as_str()), }); } - let time_is_millis = matches!( - metadata.time_index_type(), - datatypes::types::TimestampType::Millisecond(_) - ); + let time_is_millis = matches!(metadata.time_index_type(), TimestampType::Millisecond(_)); if fields.is_empty() || input.append_mode || input.series_row_selector.is_some() @@ -692,15 +693,12 @@ impl RegionScanner for SeriesScan { return Ok(None); } // Evaluate residuals with original types before raw conversion to Float64. - let Ok(physical) = - table::predicate::Predicate::to_physical_expr(expr, &typed_schema) - else { + let Ok(physical) = Predicate::to_physical_expr(expr, &typed_schema) else { return Ok(None); }; if !matches!( physical.data_type(&typed_schema), - Ok(datatypes::arrow::datatypes::DataType::Boolean - | datatypes::arrow::datatypes::DataType::Null) + Ok(ArrowDataType::Boolean | ArrowDataType::Null) ) { return Ok(None); } diff --git a/src/mito2/src/read/series_scan/raw.rs b/src/mito2/src/read/series_scan/raw.rs index 8ab154d48f9..ffc28564303 100644 --- a/src/mito2/src/read/series_scan/raw.rs +++ b/src/mito2/src/read/series_scan/raw.rs @@ -21,31 +21,37 @@ use std::task::{Context, Poll}; use std::time::Instant; use common_error::ext::BoxedError; +use common_query::error::Error as QueryError; use common_query::native_histogram::{ histogram_samples_from_arrow, is_native_histogram_value_type, }; +use common_recordbatch::filter; use datafusion::physical_plan::PhysicalExpr; use datatypes::arrow::array::{ Array, BinaryArray, DictionaryArray, Float64Array, StructArray, TimestampMillisecondArray, }; use datatypes::arrow::buffer::Buffer; use datatypes::arrow::compute::cast; -use datatypes::arrow::datatypes::UInt32Type; +use datatypes::arrow::datatypes::{DataType, UInt32Type}; use datatypes::arrow::record_batch::RecordBatch; use datatypes::prelude::ConcreteDataType; use datatypes::series::raw::{RawSeriesChunk, RawSeriesOrdering, RawTimeAxis}; use datatypes::series::samples::{Samples, SeriesFields, SeriesSchema}; +use futures::stream::BoxStream; use futures::{Stream, StreamExt}; +use snafu::Location; use snafu::{IntoError, ResultExt}; use store_api::region_engine::{RawLabelSource, RawSeriesStream, SendableRawSeriesStream}; -use crate::error::{ComputeArrowSnafu, DataTypeMismatchSnafu, InvalidRecordBatchSnafu}; +use crate::error::{ + ComputeArrowSnafu, DataTypeMismatchSnafu, Error, ExternalSnafu, InvalidRecordBatchSnafu, +}; use crate::read::scan_util::PartitionMetrics; use crate::read::series_scan::SeriesBatch; use crate::read::stream::{ScanBatch, ScanBatchStream}; use crate::sst::parquet::flat_format::{primary_key_column_index, time_index_column_index}; -fn boxed(error: crate::error::Error) -> BoxedError { +fn boxed(error: Error) -> BoxedError { BoxedError::new(error) } @@ -113,19 +119,19 @@ fn chunk( .downcast_ref::() .ok_or_else(|| invalid("invalid native histogram array"))?; samples.push(histogram_samples_from_arrow(hist).map_err(|error| { - let source = common_query::error::Error::GeneralDataFusion { + let source = QueryError::GeneralDataFusion { error, - location: snafu::Location::new(file!(), line!(), column!()), + location: Location::new(file!(), line!(), column!()), }; BoxedError::new( - crate::error::ExternalSnafu { + ExternalSnafu { context: "native histogram conversion".to_string(), } .into_error(BoxedError::new(source)), ) })?); } else { - let values = cast(&array, &datatypes::arrow::datatypes::DataType::Float64) + let values = cast(&array, &DataType::Float64) .context(ComputeArrowSnafu) .map_err(boxed)?; let values = values @@ -153,7 +159,7 @@ fn chunk( struct RawStream { schema: Arc, label_source: RawLabelSource, - inner: futures::stream::BoxStream<'static, Result>, + inner: BoxStream<'static, Result>, } impl Stream for RawStream { @@ -196,14 +202,14 @@ pub(crate) fn make_stream( for record in records { let mut record = record; for predicate in &predicates { - record = match common_recordbatch::filter::batch_filter(&record, predicate) { + record = match filter::batch_filter(&record, predicate) { Ok(record) => record, Err(error) => { - let source = common_query::error::Error::GeneralDataFusion { + let source = QueryError::GeneralDataFusion { error, - location: snafu::Location::new(file!(), line!(), column!()), + location: Location::new(file!(), line!(), column!()), }; - yield Err(BoxedError::new(crate::error::ExternalSnafu { + yield Err(BoxedError::new(ExternalSnafu { context: "raw predicate evaluation".to_string(), }.into_error(BoxedError::new(source)))); return; diff --git a/src/mito2/src/read/series_scan/raw/tests.rs b/src/mito2/src/read/series_scan/raw/tests.rs index 33ee4611182..e70bc43f405 100644 --- a/src/mito2/src/read/series_scan/raw/tests.rs +++ b/src/mito2/src/read/series_scan/raw/tests.rs @@ -18,9 +18,10 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::task::{Context, Poll}; use std::time::Instant; +use api::v1::SemanticType; use common_error::ext::ErrorExt; use common_query::native_histogram::{ - CounterResetHint, NativeHistogram, Span, build_histogram_array, + CounterResetHint, NativeHistogram, Span, build_histogram_array, native_histogram_value_type, }; use datafusion::physical_plan::PhysicalExpr; use datafusion::physical_plan::expressions::{BinaryExpr, Column, lit as physical_lit}; @@ -31,25 +32,27 @@ use datatypes::arrow::array::{ ArrayRef, BinaryArray, Int64Array, StructArray, TimestampMillisecondArray, UInt8Array, UInt64Array, }; -use datatypes::arrow::datatypes::{DataType, Field, Schema}; +use datatypes::arrow::datatypes::{DataType, Field, Schema, TimeUnit}; use datatypes::arrow::record_batch::RecordBatch; use datatypes::data_type::DataType as _; +use datatypes::prelude::ConcreteDataType; +use datatypes::schema::ColumnSchema; use datatypes::series::raw::RawSeriesChunk; -use datatypes::series::samples::SampleRef; -use datatypes::series::samples::{FieldSpec, SeriesSchema}; -use futures::{StreamExt, stream}; -use store_api::metadata::{ColumnMetadata, RegionMetadataBuilder}; +use datatypes::series::samples::{FieldSpec, SampleRef, SeriesSchema}; +use futures::{Stream, StreamExt, stream}; +use store_api::metadata::{ColumnMetadata, RegionMetadata, RegionMetadataBuilder}; use store_api::region_engine::{BoundTag, RawLabelSource, SendableRawSeriesStream}; use store_api::storage::RegionId; use super::make_stream; +use crate::error::{Error, InvalidRecordBatchSnafu, Result}; use crate::read::scan_util::PartitionMetrics; use crate::read::series_scan::{FlatSeriesBatch, SeriesBatch}; use crate::read::stream::{ScanBatch, ScanBatchStream}; fn metrics() -> PartitionMetrics { PartitionMetrics::new( - store_api::storage::RegionId::new(1, 0), + RegionId::new(1, 0), 0, "raw-test", Instant::now(), @@ -82,7 +85,7 @@ fn batch(keys: &[&[u8]], times: &[i64], values: &[Option]) -> RecordBatch { Field::new("value", DataType::Int64, true), Field::new( "ts", - DataType::Timestamp(datatypes::arrow::datatypes::TimeUnit::Millisecond, None), + DataType::Timestamp(TimeUnit::Millisecond, None), false, ), Field::new("__primary_key", DataType::Binary, false), @@ -129,12 +132,12 @@ fn histogram_input() -> (RecordBatch, Vec) { let fields = vec![ Field::new( "histogram", - common_query::native_histogram::native_histogram_value_type().as_arrow_type(), + native_histogram_value_type().as_arrow_type(), true, ), Field::new( "ts", - DataType::Timestamp(datatypes::arrow::datatypes::TimeUnit::Millisecond, None), + DataType::Timestamp(TimeUnit::Millisecond, None), false, ), Field::new("__primary_key", DataType::Binary, false), @@ -147,39 +150,33 @@ fn histogram_input() -> (RecordBatch, Vec) { ) } -fn input(items: Vec>) -> ScanBatchStream { +fn input(items: Vec>) -> ScanBatchStream { Box::pin(stream::iter(items)) } -fn metadata() -> Arc { - metadata_with_field_type(datatypes::prelude::ConcreteDataType::int64_datatype()) +fn metadata() -> Arc { + metadata_with_field_type(ConcreteDataType::int64_datatype()) } -fn metadata_with_field_type( - field_type: datatypes::prelude::ConcreteDataType, -) -> Arc { +fn metadata_with_field_type(field_type: ConcreteDataType) -> Arc { let mut builder = RegionMetadataBuilder::new(RegionId::new(1, 0)); builder.push_column_metadata(ColumnMetadata { - column_schema: datatypes::schema::ColumnSchema::new( - "tag", - datatypes::prelude::ConcreteDataType::string_datatype(), - true, - ), - semantic_type: api::v1::SemanticType::Tag, + column_schema: ColumnSchema::new("tag", ConcreteDataType::string_datatype(), true), + semantic_type: SemanticType::Tag, column_id: 0, }); builder.push_column_metadata(ColumnMetadata { - column_schema: datatypes::schema::ColumnSchema::new( + column_schema: ColumnSchema::new( "ts", - datatypes::prelude::ConcreteDataType::timestamp_millisecond_datatype(), + ConcreteDataType::timestamp_millisecond_datatype(), false, ), - semantic_type: api::v1::SemanticType::Timestamp, + semantic_type: SemanticType::Timestamp, column_id: 1, }); builder.push_column_metadata(ColumnMetadata { - column_schema: datatypes::schema::ColumnSchema::new("value", field_type, true), - semantic_type: api::v1::SemanticType::Field, + column_schema: ColumnSchema::new("value", field_type, true), + semantic_type: SemanticType::Field, column_id: 2, }); builder.primary_key(vec![0]); @@ -190,7 +187,7 @@ fn convert_with_source( inner: ScanBatchStream, schema: Arc, source: RawLabelSource, - field_type: datatypes::prelude::ConcreteDataType, + field_type: ConcreteDataType, ) -> SendableRawSeriesStream { make_stream( inner, @@ -202,7 +199,7 @@ fn convert_with_source( ) } -fn primary_key_source(metadata: Arc) -> RawLabelSource { +fn primary_key_source(metadata: Arc) -> RawLabelSource { RawLabelSource::PrimaryKey { metric_name: Arc::from("metric"), physical_metadata: metadata, @@ -218,7 +215,7 @@ fn convert(inner: ScanBatchStream, schema: Arc) -> SendableRawSeri inner, schema, primary_key_source(metadata()), - datatypes::prelude::ConcreteDataType::int64_datatype(), + ConcreteDataType::int64_datatype(), ) } @@ -227,8 +224,8 @@ struct DropSpy { drops: Arc, } -impl futures::Stream for DropSpy { - type Item = crate::error::Result; +impl Stream for DropSpy { + type Item = Result; fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { self.inner.as_mut().poll_next(cx) @@ -244,8 +241,8 @@ impl Drop for DropSpy { fn values(chunk: &RawSeriesChunk) -> Vec> { (0..chunk.time.len()) .map(|i| match chunk.fields.get(0).unwrap().sample(i) { - datatypes::series::samples::SampleRef::Missing => None, - datatypes::series::samples::SampleRef::Float(v) => Some(v), + SampleRef::Missing => None, + SampleRef::Float(v) => Some(v), _ => unreachable!(), }) .collect() @@ -344,7 +341,7 @@ async fn typed_predicate_filters_before_numeric_samples_are_cast() { )))]), schema(), primary_key_source(metadata()), - vec![(0, datatypes::prelude::ConcreteDataType::int64_datatype())], + vec![(0, ConcreteDataType::int64_datatype())], metrics(), vec![predicate], ); @@ -373,7 +370,7 @@ async fn predicate_error_is_terminal() { )))]), schema(), primary_key_source(metadata()), - vec![(0, datatypes::prelude::ConcreteDataType::int64_datatype())], + vec![(0, ConcreteDataType::int64_datatype())], metrics(), vec![predicate], ); @@ -383,7 +380,7 @@ async fn predicate_error_is_terminal() { #[tokio::test] async fn upstream_error_is_terminal_after_yielded_prefix_without_flushing_pending() { - let upstream_error = crate::error::InvalidRecordBatchSnafu { + let upstream_error = InvalidRecordBatchSnafu { reason: "upstream injected failure".to_string(), } .build(); @@ -409,8 +406,7 @@ async fn upstream_error_is_terminal_after_yielded_prefix_without_flushing_pendin assert!(second.metric.is_none()); assert!(!second.is_last); let error = output.next().await.unwrap().unwrap_err(); - let Some(crate::error::Error::InvalidRecordBatch { reason, .. }) = - error.as_any().downcast_ref::() + let Some(Error::InvalidRecordBatch { reason, .. }) = error.as_any().downcast_ref::() else { panic!("unexpected upstream error: {error}"); }; @@ -446,7 +442,7 @@ async fn raw_stream_owns_schema_and_bound_source_metadata() { input(vec![]), schema.clone(), primary_key_source(metadata.clone()), - datatypes::prelude::ConcreteDataType::int64_datatype(), + ConcreteDataType::int64_datatype(), ); drop(schema); drop(metadata); @@ -499,9 +495,9 @@ async fn native_histogram_adapter_preserves_complete_payload() { input(vec![Ok(ScanBatch::RecordBatch(record))]), series_schema, primary_key_source(metadata_with_field_type( - common_query::native_histogram::native_histogram_value_type().clone(), + native_histogram_value_type().clone(), )), - common_query::native_histogram::native_histogram_value_type().clone(), + native_histogram_value_type().clone(), ); assert!(matches!( raw.label_source(),