remove some unwrap.

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<TableIdentifier>,
@@ -110,7 +111,6 @@ impl ExecutionPlan for ResolvedPartitionedScan {
         self
     }
 
-    // TODO: check if it is right.
     fn schema(&self) -> SchemaRef {
         self.remote_exec_plans
             .first()

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,

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 },
 }

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<Arc<dyn ExecutionPlan>> {
         // 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.