refactor: move DataSource to datafusion-datasource

datafusion/datasource/src/memory.rs

@@ -708,7 +708,7 @@ impl MemorySourceConfig {
 }
 
 #[cfg(test)]
-mod memory_exec_tests {
+mod memory_source_tests {
     use std::sync::Arc;
 
     use crate::memory::MemorySourceConfig;

datafusion/physical-plan/src/aggregates/mod.rs

@@ -1350,7 +1350,7 @@ mod tests {
     use crate::metrics::MetricValue;
     use crate::test::assert_is_pending;
     use crate::test::exec::{assert_strong_count_converges_to_zero, BlockingExec};
-    use crate::test::MockMemorySourceConfig;
+    use crate::test::TestMemoryExec;
     use crate::RecordBatchStream;
 
     use arrow::array::{
@@ -2207,7 +2207,7 @@ mod tests {
             vec![test_last_value_agg_expr(&schema, sort_options)?]
         };
 
-        let memory_exec = MockMemorySourceConfig::try_new_exec(
+        let memory_exec = TestMemoryExec::try_new_exec(
             &[
                 vec![partition1],
                 vec![partition2],
@@ -2442,11 +2442,8 @@ mod tests {
             })
             .collect();
 
-        let input = MockMemorySourceConfig::try_new_exec(
-            &[input_batches],
-            Arc::clone(&schema),
-            None,
-        )?;
+        let input =
+            TestMemoryExec::try_new_exec(&[input_batches], Arc::clone(&schema), None)?;
 
         let aggregate_exec = Arc::new(AggregateExec::try_new(
             AggregateMode::Single,
@@ -2557,7 +2554,7 @@ mod tests {
         .build()
         .map(Arc::new)?];
 
-        let input = MockMemorySourceConfig::try_new_exec(
+        let input = TestMemoryExec::try_new_exec(
             &[vec![batch.clone()]],
             Arc::<Schema>::clone(&batch.schema()),
             None,
@@ -2626,11 +2623,8 @@ mod tests {
             .unwrap(),
         ];
 
-        let input = MockMemorySourceConfig::try_new_exec(
-            &[input_data],
-            Arc::clone(&schema),
-            None,
-        )?;
+        let input =
+            TestMemoryExec::try_new_exec(&[input_data], Arc::clone(&schema), None)?;
         let aggregate_exec = Arc::new(AggregateExec::try_new(
             AggregateMode::Partial,
             group_by,
@@ -2716,11 +2710,8 @@ mod tests {
             .unwrap(),
         ];
 
-        let input = MockMemorySourceConfig::try_new_exec(
-            &[input_data],
-            Arc::clone(&schema),
-            None,
-        )?;
+        let input =
+            TestMemoryExec::try_new_exec(&[input_data], Arc::clone(&schema), None)?;
         let aggregate_exec = Arc::new(AggregateExec::try_new(
             AggregateMode::Partial,
             group_by,
@@ -2835,7 +2826,7 @@ mod tests {
             create_record_batch(&schema, (vec![2, 3, 4, 4], vec![1.0, 2.0, 3.0, 4.0]))?,
         ];
         let plan: Arc<dyn ExecutionPlan> =
-            MockMemorySourceConfig::try_new_exec(&[batches], Arc::clone(&schema), None)?;
+            TestMemoryExec::try_new_exec(&[batches], Arc::clone(&schema), None)?;
 
         let grouping_set = PhysicalGroupBy::new(
             vec![(col("a", &schema)?, "a".to_string())],

datafusion/physical-plan/src/joins/hash_join.rs

@@ -1638,7 +1638,7 @@ impl EmbeddedProjection for HashJoinExec {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::test::MockMemorySourceConfig;
+    use crate::test::TestMemoryExec;
     use crate::{
         common, expressions::Column, repartition::RepartitionExec, test::build_table_i32,
         test::exec::MockExec,
@@ -1680,7 +1680,7 @@ mod tests {
     ) -> Arc<dyn ExecutionPlan> {
         let batch = build_table_i32(a, b, c);
         let schema = batch.schema();
-        MockMemorySourceConfig::try_new_exec(&[vec![batch]], schema, None).unwrap()
+        TestMemoryExec::try_new_exec(&[vec![batch]], schema, None).unwrap()
     }
 
     fn join(
@@ -2082,12 +2082,9 @@ mod tests {
         let batch2 =
             build_table_i32(("a1", &vec![2]), ("b2", &vec![2]), ("c1", &vec![9]));
         let schema = batch1.schema();
-        let left = MockMemorySourceConfig::try_new_exec(
-            &[vec![batch1], vec![batch2]],
-            schema,
-            None,
-        )
-        .unwrap();
+        let left =
+            TestMemoryExec::try_new_exec(&[vec![batch1], vec![batch2]], schema, None)
+                .unwrap();
 
         let right = build_table(
             ("a1", &vec![1, 2, 3]),
@@ -2157,12 +2154,9 @@ mod tests {
         );
         let schema = batch1.schema();
 
-        let left = MockMemorySourceConfig::try_new_exec(
-            &[vec![batch1], vec![batch2]],
-            schema,
-            None,
-        )
-        .unwrap();
+        let left =
+            TestMemoryExec::try_new_exec(&[vec![batch1], vec![batch2]], schema, None)
+                .unwrap();
         let right = build_table(
             ("a2", &vec![20, 30, 10]),
             ("b2", &vec![5, 6, 4]),
@@ -2214,12 +2208,9 @@ mod tests {
         let batch2 =
             build_table_i32(("a2", &vec![30]), ("b1", &vec![5]), ("c2", &vec![90]));
         let schema = batch1.schema();
-        let right = MockMemorySourceConfig::try_new_exec(
-            &[vec![batch1], vec![batch2]],
-            schema,
-            None,
-        )
-        .unwrap();
+        let right =
+            TestMemoryExec::try_new_exec(&[vec![batch1], vec![batch2]], schema, None)
+                .unwrap();
 
         let on = vec![(
             Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
@@ -2297,8 +2288,7 @@ mod tests {
     ) -> Arc<dyn ExecutionPlan> {
         let batch = build_table_i32(a, b, c);
         let schema = batch.schema();
-        MockMemorySourceConfig::try_new_exec(&[vec![batch.clone(), batch]], schema, None)
-            .unwrap()
+        TestMemoryExec::try_new_exec(&[vec![batch.clone(), batch]], schema, None).unwrap()
     }
 
     #[apply(batch_sizes)]
@@ -2403,8 +2393,7 @@ mod tests {
             Arc::new(Column::new_with_schema("b1", &right.schema()).unwrap()) as _,
         )];
         let schema = right.schema();
-        let right =
-            MockMemorySourceConfig::try_new_exec(&[vec![right]], schema, None).unwrap();
+        let right = TestMemoryExec::try_new_exec(&[vec![right]], schema, None).unwrap();
         let join = join(left, right, on, &JoinType::Left, false).unwrap();
 
         let columns = columns(&join.schema());
@@ -2441,8 +2430,7 @@ mod tests {
             Arc::new(Column::new_with_schema("b2", &right.schema()).unwrap()) as _,
         )];
         let schema = right.schema();
-        let right =
-            MockMemorySourceConfig::try_new_exec(&[vec![right]], schema, None).unwrap();
+        let right = TestMemoryExec::try_new_exec(&[vec![right]], schema, None).unwrap();
         let join = join(left, right, on, &JoinType::Full, false).unwrap();
 
         let columns = columns(&join.schema());
@@ -3746,17 +3734,13 @@ mod tests {
         let dates: ArrayRef = Arc::new(Date32Array::from(vec![19107, 19108, 19109]));
         let n: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3]));
         let batch = RecordBatch::try_new(Arc::clone(&schema), vec![dates, n])?;
-        let left = MockMemorySourceConfig::try_new_exec(
-            &[vec![batch]],
-            Arc::clone(&schema),
-            None,
-        )
-        .unwrap();
+        let left =
+            TestMemoryExec::try_new_exec(&[vec![batch]], Arc::clone(&schema), None)
+                .unwrap();
         let dates: ArrayRef = Arc::new(Date32Array::from(vec![19108, 19108, 19109]));
         let n: ArrayRef = Arc::new(Int32Array::from(vec![4, 5, 6]));
         let batch = RecordBatch::try_new(Arc::clone(&schema), vec![dates, n])?;
-        let right =
-            MockMemorySourceConfig::try_new_exec(&[vec![batch]], schema, None).unwrap();
+        let right = TestMemoryExec::try_new_exec(&[vec![batch]], schema, None).unwrap();
         let on = vec![(
             Arc::new(Column::new_with_schema("date", &left.schema()).unwrap()) as _,
             Arc::new(Column::new_with_schema("date", &right.schema()).unwrap()) as _,
@@ -4046,7 +4030,7 @@ mod tests {
             ("b1", &vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 0]),
             ("c1", &vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 0]),
         );
-        let left = MockMemorySourceConfig::try_new_exec(
+        let left = TestMemoryExec::try_new_exec(
             &[vec![left_batch.clone()], vec![left_batch.clone()]],
             left_batch.schema(),
             None,
@@ -4057,7 +4041,7 @@ mod tests {
             ("b2", &vec![12, 13]),
             ("c2", &vec![14, 15]),
         );
-        let right = MockMemorySourceConfig::try_new_exec(
+        let right = TestMemoryExec::try_new_exec(
             &[vec![right_batch.clone()], vec![right_batch.clone()]],
             right_batch.schema(),
             None,
@@ -4142,7 +4126,7 @@ mod tests {
         )
         .unwrap();
         let schema_ref = batch.schema();
-        MockMemorySourceConfig::try_new_exec(&[vec![batch]], schema_ref, None).unwrap()
+        TestMemoryExec::try_new_exec(&[vec![batch]], schema_ref, None).unwrap()
     }
 
     #[tokio::test]

datafusion/physical-plan/src/joins/nested_loop_join.rs

@@ -1030,8 +1030,7 @@ impl EmbeddedProjection for NestedLoopJoinExec {
 #[cfg(test)]
 pub(crate) mod tests {
     use super::*;
-    use crate::test::MockMemorySourceConfig;
-    // use crate::test::MockMemorySourceConfig;
+    use crate::test::TestMemoryExec;
     use crate::{
         common, expressions::Column, repartition::RepartitionExec, test::build_table_i32,
     };
@@ -1072,8 +1071,7 @@ pub(crate) mod tests {
         };
 
         let mut source =
-            MockMemorySourceConfig::try_new(&[batches], Arc::clone(&schema), None)
-                .unwrap();
+            TestMemoryExec::try_new(&[batches], Arc::clone(&schema), None).unwrap();
         if !sorted_column_names.is_empty() {
             let mut sort_info = LexOrdering::default();
             for name in sorted_column_names {
@@ -1090,7 +1088,7 @@ pub(crate) mod tests {
             source = source.try_with_sort_information(vec![sort_info]).unwrap();
         }
 
-        Arc::new(MockMemorySourceConfig::update_cache(Arc::new(source)))
+        Arc::new(TestMemoryExec::update_cache(Arc::new(source)))
     }
 
     fn build_left_table() -> Arc<dyn ExecutionPlan> {

datafusion/physical-plan/src/joins/sort_merge_join.rs

@@ -2547,7 +2547,7 @@ mod tests {
     use crate::joins::sort_merge_join::{get_corrected_filter_mask, JoinedRecordBatches};
     use crate::joins::utils::{ColumnIndex, JoinFilter, JoinOn};
     use crate::joins::SortMergeJoinExec;
-    use crate::test::MockMemorySourceConfig;
+    use crate::test::TestMemoryExec;
     use crate::test::{build_table_i32, build_table_i32_two_cols};
     use crate::{common, ExecutionPlan};
 
@@ -2558,12 +2558,12 @@ mod tests {
     ) -> Arc<dyn ExecutionPlan> {
         let batch = build_table_i32(a, b, c);
         let schema = batch.schema();
-        MockMemorySourceConfig::try_new_exec(&[vec![batch]], schema, None).unwrap()
+        TestMemoryExec::try_new_exec(&[vec![batch]], schema, None).unwrap()
     }
 
     fn build_table_from_batches(batches: Vec<RecordBatch>) -> Arc<dyn ExecutionPlan> {
         let schema = batches.first().unwrap().schema();
-        MockMemorySourceConfig::try_new_exec(&[batches], schema, None).unwrap()
+        TestMemoryExec::try_new_exec(&[batches], schema, None).unwrap()
     }
 
     fn build_date_table(
@@ -2588,7 +2588,7 @@ mod tests {
         .unwrap();
 
         let schema = batch.schema();
-        MockMemorySourceConfig::try_new_exec(&[vec![batch]], schema, None).unwrap()
+        TestMemoryExec::try_new_exec(&[vec![batch]], schema, None).unwrap()
     }
 
     fn build_date64_table(
@@ -2613,7 +2613,7 @@ mod tests {
         .unwrap();
 
         let schema = batch.schema();
-        MockMemorySourceConfig::try_new_exec(&[vec![batch]], schema, None).unwrap()
+        TestMemoryExec::try_new_exec(&[vec![batch]], schema, None).unwrap()
     }
 
     /// returns a table with 3 columns of i32 in memory
@@ -2636,7 +2636,7 @@ mod tests {
             ],
         )
         .unwrap();
-        MockMemorySourceConfig::try_new_exec(&[vec![batch]], schema, None).unwrap()
+        TestMemoryExec::try_new_exec(&[vec![batch]], schema, None).unwrap()
     }
 
     pub fn build_table_two_cols(
@@ -2645,7 +2645,7 @@ mod tests {
     ) -> Arc<dyn ExecutionPlan> {
         let batch = build_table_i32_two_cols(a, b);
         let schema = batch.schema();
-        MockMemorySourceConfig::try_new_exec(&[vec![batch]], schema, None).unwrap()
+        TestMemoryExec::try_new_exec(&[vec![batch]], schema, None).unwrap()
     }
 
     fn join(

datafusion/physical-plan/src/joins/test_utils.rs

@@ -24,8 +24,7 @@ use crate::joins::{
     HashJoinExec, PartitionMode, StreamJoinPartitionMode, SymmetricHashJoinExec,
 };
 use crate::repartition::RepartitionExec;
-use crate::test::MockMemorySourceConfig;
-// use crate::test::MockMemorySourceConfig;
+use crate::test::TestMemoryExec;
 use crate::{common, ExecutionPlan, ExecutionPlanProperties, Partitioning};
 
 use arrow::array::{
@@ -530,14 +529,14 @@ pub fn create_memory_table(
     right_sorted: Vec<LexOrdering>,
 ) -> Result<(Arc<dyn ExecutionPlan>, Arc<dyn ExecutionPlan>)> {
     let left_schema = left_partition[0].schema();
-    let left = MockMemorySourceConfig::try_new(&[left_partition], left_schema, None)?
+    let left = TestMemoryExec::try_new(&[left_partition], left_schema, None)?
         .try_with_sort_information(left_sorted)?;
     let right_schema = right_partition[0].schema();
-    let right = MockMemorySourceConfig::try_new(&[right_partition], right_schema, None)?
+    let right = TestMemoryExec::try_new(&[right_partition], right_schema, None)?
         .try_with_sort_information(right_sorted)?;
     Ok((
-        Arc::new(MockMemorySourceConfig::update_cache(Arc::new(left))),
-        Arc::new(MockMemorySourceConfig::update_cache(Arc::new(right))),
+        Arc::new(TestMemoryExec::update_cache(Arc::new(left))),
+        Arc::new(TestMemoryExec::update_cache(Arc::new(right))),
     ))
 }