fix rust ci cases

Signed-off-by: zenghua <[email protected]>

rust/lakesoul-datafusion/src/catalog/mod.rs

@@ -29,7 +29,7 @@ pub(crate) async fn create_table(client: MetaDataClientRef, table_name: &str, co
         TableInfo {
             table_id: format!("table_{}", uuid::Uuid::new_v4()),
             table_name: table_name.to_string(), 
-            table_path: [env::temp_dir().to_str().unwrap(), table_name].iter().collect::<PathBuf>().to_str().unwrap().to_string(),
+            table_path: format!("{}default/{}", env::temp_dir().to_str().unwrap(), table_name),
             table_schema: serde_json::to_string::<ArrowJavaSchema>(&config.schema().into()).unwrap(),
             table_namespace: "default".to_string(),
             properties: "{}".to_string(),

rust/lakesoul-datafusion/src/datasource/file_format/metadata_format.rs

@@ -156,7 +156,7 @@ impl LakeSoulParquetSink {
     fn table_info(&self) -> Arc<TableInfo> {
         self.table_info.clone()
     }
-    
+
 }
 
 #[async_trait]

rust/lakesoul-datafusion/src/lakesoul_table/helpers.rs

@@ -29,9 +29,9 @@ pub(crate) fn create_io_config_builder_from_table_info(table_info: Arc<TableInfo
 }
 
 
-pub(crate) fn create_sort_exprs(pks: &[String]) -> Vec<Expr> {
-    pks
+pub(crate) fn create_sort_exprs(colunms: &[String]) -> Vec<Expr> {
+    colunms
         .iter()
-        .map(|pk| col(pk).sort(true, true))
+        .map(|column| col(column).sort(true, true))
         .collect()
 }

rust/lakesoul-datafusion/src/planner/physical_planner.rs

@@ -53,13 +53,13 @@ impl PhysicalPlanner for LakeSoulPhysicalPlanner {
             }) => {
                 let name = table_name.table();
                 // let schema = session_state.schema_for_ref(table_name)?;
-                let table = LakeSoulTable::for_name(name).await.unwrap();
-                match table.as_sink_provider(session_state).await {
+                let lakesoul_table = LakeSoulTable::for_name(name).await.unwrap();
+                match lakesoul_table.as_sink_provider(session_state).await {
                     Ok(provider) => {
                         let builder = LogicalPlanBuilder::from(input.deref().clone());
 
-                        let builder = if table.primary_keys().is_empty() {
-                            if !table
+                        let builder = if lakesoul_table.primary_keys().is_empty() {
+                            if !lakesoul_table
                                 .schema()
                                 .logically_equivalent_names_and_types(&Schema::from(input.schema().as_ref()))
                             {
@@ -71,7 +71,7 @@ impl PhysicalPlanner for LakeSoulPhysicalPlanner {
 
                             builder
                         } else {
-                            let sort_exprs = create_sort_exprs(table.primary_keys());
+                            let sort_exprs = create_sort_exprs(lakesoul_table.primary_keys());
                             builder.sort(sort_exprs)?
                         };
 

rust/lakesoul-datafusion/src/test/upsert_tests.rs

@@ -1220,37 +1220,6 @@ mod upsert_with_metadata_tests {
         lakesoul_table.execute_upsert(record_batch).await
     }
 
-    async fn check_upsert_print(table_name: &str, selected_cols: Vec<&str>, filters: Option<String>, client: MetaDataClientRef, expected: &[&str]) -> Result<()> {
-        let lakesoul_table = LakeSoulTable::for_name(table_name).await?;
-        let builder = create_io_config_builder(client, None, false).await?;
-        let sess_ctx = create_session_context(&mut builder.clone().build())?;
-
-
-        let dataframe = lakesoul_table.to_dataframe(&sess_ctx).await?;
-        let schema = SchemaRef::new(dataframe.schema().into());
-
-        let dataframe = if let Some(f) = filters {
-            dataframe.filter(Parser::parse(f.clone(), schema))?
-        } else {
-            dataframe
-        };
-
-        let dataframe = if selected_cols.is_empty() {
-            dataframe
-        } else {
-            dataframe.select_columns(&selected_cols)?
-        };
-
-        let result = dataframe
-            // .explain(true, false)?
-            .collect()
-            .await?;
-
-        // print_batches(&result);
-        assert_batches_eq!(expected, &result);
-        Ok(())
-    }
-
     async fn check_upsert_debug(batch: RecordBatch, table_name: &str, selected_cols: Vec<&str>, filters: Option<String>, client: MetaDataClientRef, expected: &[&str]) -> Result<()> {
         let lakesoul_table = LakeSoulTable::for_name(table_name).await?;
         lakesoul_table.execute_upsert(batch).await?;

rust/lakesoul-io/src/datasource/listing.rs

@@ -87,7 +87,7 @@ impl LakeSoulListingTable {
                     .with_table_partition_cols(table_partition_cols)
                     .with_insert_mode(datafusion::datasource::listing::ListingTableInsertMode::AppendNewFiles);
 
-                let prefix = ListingTableUrl::parse(lakesoul_io_config.prefix.clone())?;
+                let prefix = ListingTableUrl::parse_create_local_if_not_exists(lakesoul_io_config.prefix.clone(), true)?;
 
                 ListingTableConfig::new(prefix)
                     .with_listing_options(listing_options)