diff --git a/src/servers/src/batcher/logical_table/pending_worker.rs b/src/servers/src/batcher/logical_table/pending_worker.rs index 6f46c717588..ff363c7637c 100644 --- a/src/servers/src/batcher/logical_table/pending_worker.rs +++ b/src/servers/src/batcher/logical_table/pending_worker.rs @@ -288,6 +288,7 @@ mod tests { use common_meta::kv_backend::memory::MemoryKvBackend; use common_meta::node_manager::NodeManagerRef; use moka::future::CacheBuilder; + use operator::write_dispatcher::WriteDispatcher; use partition::cache::new_partition_info_cache; use partition::manager::PartitionRuleManager; use tokio::sync::{Semaphore, broadcast, mpsc, oneshot}; @@ -504,7 +505,7 @@ mod tests { let node_manager: NodeManagerRef = Arc::new(ConcurrentMockNodeManager { datanodes: Arc::new(HashMap::new()), }); - let write_dispatcher = Arc::new(operator::write_dispatcher::WriteDispatcher::new( + let write_dispatcher = Arc::new(WriteDispatcher::new( node_manager, partition_manager.clone(), )); diff --git a/src/servers/src/batcher/logical_table/region_write.rs b/src/servers/src/batcher/logical_table/region_write.rs index 34bc2b7bf46..434984c23d3 100644 --- a/src/servers/src/batcher/logical_table/region_write.rs +++ b/src/servers/src/batcher/logical_table/region_write.rs @@ -274,13 +274,16 @@ pub(in crate::batcher::logical_table) fn map_ipc_error(error: GrpcError) -> Erro #[cfg(test)] mod tests { - use api::v1::region::region_request; + use std::collections::HashMap; + use std::sync::Arc; + + use api::v1::PartitionExprVersion; + use api::v1::region::{bulk_insert_request, region_request}; use arrow::array::{Array, BinaryArray, BooleanArray, StringArray, TimestampMillisecondArray}; use arrow::datatypes::{DataType as ArrowDataType, Field, Schema as ArrowSchema}; use arrow::record_batch::RecordBatch; + use common_grpc::flight::FlightDecoder; use partition::partition::{PartitionRule, RegionMask}; - use std::collections::HashMap; - use std::sync::Arc; use store_api::storage::RegionId; use crate::batcher::logical_table::region_write::{ @@ -340,11 +343,10 @@ mod tests { request.partition_expr_version.unwrap().value, 7 + index as u64 ); - let Some(api::v1::region::bulk_insert_request::Body::ArrowIpc(ipc)) = request.body - else { + let Some(bulk_insert_request::Body::ArrowIpc(ipc)) = request.body else { panic!("expected IPC") }; - let batch = common_grpc::flight::FlightDecoder::try_from_schema_bytes(&ipc.schema) + let batch = FlightDecoder::try_from_schema_bytes(&ipc.schema) .unwrap() .try_decode_record_batch(&ipc.data_header, &ipc.payload) .unwrap(); @@ -436,7 +438,7 @@ mod tests { assert_eq!(RegionId::new(1024, 1).as_u64(), request.region_id); assert_eq!(skip_wal, request.skip_wal); assert_eq!( - Some(api::v1::PartitionExprVersion { value: 7 }), + Some(PartitionExprVersion { value: 7 }), request.partition_expr_version ); }