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

@@ -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<api::v1::ColumnSchema>, tag_values: &[&[&str]], start: i64| {
+        let write = |schema: Vec<ColumnSchema>, 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

src/mito2/src/engine/scan_test.rs

@@ -4763,12 +4763,18 @@ fn scan_field_values(batches: &[common_recordbatch::RecordBatch]) -> Vec<f64> {
 /// 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<f64>, f1: Option<f64>, gate: f64, ts: i64| api::v1::Row {
+    let row = |tag: &str, f0: Option<f64>, f1: Option<f64>, 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::<Vec<_>>()
         );
-        let mut closed_keys = std::collections::HashSet::<Vec<u8>>::new();
+        let mut closed_keys = HashSet::<Vec<u8>>::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::<Vec<_>>();
@@ -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)

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<Option<store_api::region_engine::SendableRawSeriesStream>, BoxedError> {
+        label_source: &RawLabelSource,
+    ) -> Result<Option<SendableRawSeriesStream>, 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);
                 }

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::<StructArray>()
                 .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<SeriesSchema>,
     label_source: RawLabelSource,
-    inner: futures::stream::BoxStream<'static, Result<RawSeriesChunk, BoxedError>>,
+    inner: BoxStream<'static, Result<RawSeriesChunk, BoxedError>>,
 }
 
 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;

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<i64>]) -> 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<NativeHistogram>) {
     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<NativeHistogram>) {
     )
 }
 
-fn input(items: Vec<crate::error::Result<ScanBatch>>) -> ScanBatchStream {
+fn input(items: Vec<Result<ScanBatch>>) -> ScanBatchStream {
     Box::pin(stream::iter(items))
 }
 
-fn metadata() -> Arc<store_api::metadata::RegionMetadata> {
-    metadata_with_field_type(datatypes::prelude::ConcreteDataType::int64_datatype())
+fn metadata() -> Arc<RegionMetadata> {
+    metadata_with_field_type(ConcreteDataType::int64_datatype())
 }
 
-fn metadata_with_field_type(
-    field_type: datatypes::prelude::ConcreteDataType,
-) -> Arc<store_api::metadata::RegionMetadata> {
+fn metadata_with_field_type(field_type: ConcreteDataType) -> Arc<RegionMetadata> {
     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<SeriesSchema>,
     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<store_api::metadata::RegionMetadata>) -> RawLabelSource {
+fn primary_key_source(metadata: Arc<RegionMetadata>) -> RawLabelSource {
     RawLabelSource::PrimaryKey {
         metric_name: Arc::from("metric"),
         physical_metadata: metadata,
@@ -218,7 +215,7 @@ fn convert(inner: ScanBatchStream, schema: Arc<SeriesSchema>) -> SendableRawSeri
         inner,
         schema,
         primary_key_source(metadata()),
-        datatypes::prelude::ConcreteDataType::int64_datatype(),
+        ConcreteDataType::int64_datatype(),
     )
 }
 
@@ -227,8 +224,8 @@ struct DropSpy {
     drops: Arc<AtomicUsize>,
 }
 
-impl futures::Stream for DropSpy {
-    type Item = crate::error::Result<ScanBatch>;
+impl Stream for DropSpy {
+    type Item = Result<ScanBatch>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         self.inner.as_mut().poll_next(cx)
@@ -244,8 +241,8 @@ impl Drop for DropSpy {
 fn values(chunk: &RawSeriesChunk) -> Vec<Option<f64>> {
     (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::<crate::error::Error>()
+    let Some(Error::InvalidRecordBatch { reason, .. }) = error.as_any().downcast_ref::<Error>()
     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(),