diff --git a/df_engine_extensions/src/dist_sql_query/partitioned_table_scan.rs b/df_engine_extensions/src/dist_sql_query/partitioned_table_scan.rs index c078b6c980..f1c3c5887e 100644 --- a/df_engine_extensions/src/dist_sql_query/partitioned_table_scan.rs +++ b/df_engine_extensions/src/dist_sql_query/partitioned_table_scan.rs @@ -31,6 +31,7 @@ use table_engine::{ table::ReadRequest, }; +/// Unresolved partitioned table scan which can't be executed before resolving #[derive(Debug)] pub struct UnresolvedPartitionedScan { pub sub_tables: Vec, @@ -110,7 +111,6 @@ impl ExecutionPlan for ResolvedPartitionedScan { self } - // TODO: check if it is right. fn schema(&self) -> SchemaRef { self.remote_exec_plans .first() diff --git a/df_engine_extensions/src/dist_sql_query/sub_table_scan.rs b/df_engine_extensions/src/dist_sql_query/sub_table_scan.rs index cbb5b4681d..b2477b95e1 100644 --- a/df_engine_extensions/src/dist_sql_query/sub_table_scan.rs +++ b/df_engine_extensions/src/dist_sql_query/sub_table_scan.rs @@ -28,7 +28,7 @@ use datafusion::{ }; use table_engine::{provider::ScanTable, remote::model::TableIdentifier, table::ReadRequest}; -/// Unresolved sub table scan which can't be executed before rewriting +/// Unresolved sub table scan which can't be executed before resolving #[derive(Debug)] pub struct UnresolvedSubTableScan { pub table: TableIdentifier, diff --git a/partition_table_engine/src/error.rs b/partition_table_engine/src/error.rs index 46db30710b..47c96f286d 100644 --- a/partition_table_engine/src/error.rs +++ b/partition_table_engine/src/error.rs @@ -22,4 +22,7 @@ define_result!(Error); pub enum Error { #[snafu(display("Internal error, message:{}, err:{}", msg, source))] Internal { msg: String, source: GenericError }, + + #[snafu(display("Datafusion error, message:{}, err:{}", msg, source))] + Datafusion { msg: String, source: GenericError }, } diff --git a/partition_table_engine/src/provider.rs b/partition_table_engine/src/provider.rs index abb10ea125..f9d81c700b 100644 --- a/partition_table_engine/src/provider.rs +++ b/partition_table_engine/src/provider.rs @@ -45,7 +45,7 @@ use table_engine::{ }; use trace_metric::MetricsCollector; -const SCAN_TABLE_METRICS_COLLECTOR_NAME: &str = "scan_table"; +const SCAN_TABLE_METRICS_COLLECTOR_NAME: &str = "scan_partitioned_table"; /// `TableProviderAdapter` for `PartitionedTableImpl`. // TODO: use single `TableProviderAdapter` for normal table and partitioned table @@ -155,12 +155,16 @@ impl TableProviderAdapter { ) -> Result> { // Build partition rule. let df_partition_rule = - DfPartitionRuleAdapter::new(partition_info.clone(), &self.table.schema()).unwrap(); + DfPartitionRuleAdapter::new(partition_info.clone(), &self.table.schema()).map_err( + |e| DataFusionError::Internal(format!("failed to build partition rule, err:{e}")), + )?; // Evaluate expr and locate partition. let partitions = df_partition_rule .locate_partitions_for_read(request.predicate.exprs()) - .unwrap(); + .map_err(|e| { + DataFusionError::Internal(format!("failed to locate partition for read, err:{e}")) + })?; let sub_tables = self.get_sub_table_idents(partition_info, partitions); // Build plan.