Fix datasource-*

Author: DoumanAsh

datafusion/core/tests/dataframe/mod.rs

@@ -6897,7 +6897,7 @@ async fn write_physical_plan_results_with_prefix() -> Result<()> {
         ctx,
     } = FixtureDataGen::prepare_execution_plan_writes(config).await?;
 
-    let partitioned_file = format!("{out_dir}/c2=123/prefix-*");
+    let partitioned_file = format!("{out_dir}/prefix-*");
 
     let df = ctx
         .read_parquet(&partitioned_file, Default::default())

datafusion/datasource-csv/src/source.rs

@@ -455,15 +455,13 @@ pub async fn plan_to_csv(
     let store = task_ctx.runtime_env().object_store(&object_store_url)?;
     let exec_options = &task_ctx.session_config().options().execution;
     let writer_buffer_size = exec_options.objectstore_writer_buffer_size;
+    let file_name_prefix = exec_options.partitioned_file_prefix_name.as_str();
+
     let mut join_set = JoinSet::new();
     for i in 0..plan.output_partitioning().partition_count() {
         let storeref = Arc::clone(&store);
         let plan: Arc<dyn ExecutionPlan> = Arc::clone(&plan);
-        let filename = format!(
-            "{}/{}part-{i}.csv",
-            exec_options.partitioned_file_prefix_name,
-            parsed.prefix()
-        );
+        let filename = format!("{}/{file_name_prefix}part-{i}.csv", parsed.prefix(),);
         let file = object_store::path::Path::parse(filename)?;
 
         let mut stream = plan.execute(i, Arc::clone(&task_ctx))?;

datafusion/datasource-json/src/source.rs

@@ -277,15 +277,13 @@ pub async fn plan_to_json(
     let store = task_ctx.runtime_env().object_store(&object_store_url)?;
     let exec_options = &task_ctx.session_config().options().execution;
     let writer_buffer_size = exec_options.objectstore_writer_buffer_size;
+    let file_name_prefix = exec_options.partitioned_file_prefix_name.as_str();
+
     let mut join_set = JoinSet::new();
     for i in 0..plan.output_partitioning().partition_count() {
         let storeref = Arc::clone(&store);
         let plan: Arc<dyn ExecutionPlan> = Arc::clone(&plan);
-        let filename = format!(
-            "{}/{}part-{i}.json",
-            exec_options.partitioned_file_prefix_name,
-            parsed.prefix()
-        );
+        let filename = format!("{}/{file_name_prefix}part-{i}.json", parsed.prefix());
         let file = object_store::path::Path::parse(filename)?;
 
         let mut stream = plan.execute(i, Arc::clone(&task_ctx))?;

datafusion/datasource-parquet/src/writer.rs

@@ -40,13 +40,11 @@ pub async fn plan_to_parquet(
     let store = task_ctx.runtime_env().object_store(&object_store_url)?;
     let mut join_set = JoinSet::new();
     let exec_options = &task_ctx.session_config().options().execution;
+    let file_name_prefix = exec_options.partitioned_file_prefix_name.as_str();
+
     for i in 0..plan.output_partitioning().partition_count() {
         let plan: Arc<dyn ExecutionPlan> = Arc::clone(&plan);
-        let filename = format!(
-            "{}/{}part-{i}.parquet",
-            exec_options.partitioned_file_prefix_name,
-            parsed.prefix()
-        );
+        let filename = format!("{}/{file_name_prefix}part-{i}.parquet", parsed.prefix());
         let file = Path::parse(filename)?;
         let propclone = writer_properties.clone();