Address reviews

datafusion-examples/examples/custom_datasource.rs

@@ -208,6 +208,7 @@ impl CustomExec {
         }
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(schema: SchemaRef) -> PlanPropertiesCache {
         let eq_properties = EquivalenceProperties::new(schema);
         PlanPropertiesCache::new(

datafusion/core/src/datasource/physical_plan/arrow_file.rs

@@ -84,6 +84,7 @@ impl ArrowExec {
         Partitioning::UnknownPartitioning(file_scan_config.file_groups.len())
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         schema: SchemaRef,
         projected_output_ordering: &[LexOrdering],

datafusion/core/src/datasource/physical_plan/avro.rs

@@ -69,6 +69,7 @@ impl AvroExec {
         &self.base_config
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         schema: SchemaRef,
         orderings: &[LexOrdering],

datafusion/core/src/datasource/physical_plan/csv.rs

@@ -121,6 +121,7 @@ impl CsvExec {
         Partitioning::UnknownPartitioning(file_scan_config.file_groups.len())
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         schema: SchemaRef,
         orderings: &[LexOrdering],

datafusion/core/src/datasource/physical_plan/json.rs

@@ -89,6 +89,7 @@ impl NdJsonExec {
         Partitioning::UnknownPartitioning(file_scan_config.file_groups.len())
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         schema: SchemaRef,
         orderings: &[LexOrdering],

datafusion/core/src/datasource/physical_plan/parquet/mod.rs

@@ -266,6 +266,7 @@ impl ParquetExec {
         Partitioning::UnknownPartitioning(file_config.file_groups.len())
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         schema: SchemaRef,
         orderings: &[LexOrdering],

datafusion/core/src/physical_optimizer/enforce_distribution.rs

@@ -1360,6 +1360,7 @@ pub(crate) mod tests {
             }
         }
 
+        /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
         fn create_cache(input: &Arc<dyn ExecutionPlan>) -> PlanPropertiesCache {
             PlanPropertiesCache::new(
                 input.equivalence_properties().clone(), // Equivalence Properties

datafusion/core/src/physical_optimizer/output_requirements.rs

@@ -112,6 +112,7 @@ impl OutputRequirementExec {
         self.input.clone()
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(input: &Arc<dyn ExecutionPlan>) -> PlanPropertiesCache {
         PlanPropertiesCache::new(
             input.equivalence_properties().clone(), // Equivalence Properties

datafusion/core/src/physical_planner.rs

@@ -2584,6 +2584,7 @@ mod tests {
             Self { cache }
         }
 
+        /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
         fn create_cache(schema: SchemaRef) -> PlanPropertiesCache {
             let eq_properties = EquivalenceProperties::new(schema);
             PlanPropertiesCache::new(

datafusion/core/src/test/mod.rs

@@ -384,6 +384,7 @@ impl StatisticsExec {
         }
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(schema: SchemaRef) -> PlanPropertiesCache {
         let eq_properties = EquivalenceProperties::new(schema);
         PlanPropertiesCache::new(

datafusion/core/src/test_util/mod.rs

@@ -46,6 +46,7 @@ use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
 use arrow::record_batch::RecordBatch;
 use datafusion_common::TableReference;
 use datafusion_expr::{CreateExternalTable, Expr, TableType};
+use datafusion_physical_expr::EquivalenceProperties;
 
 use async_trait::async_trait;
 use futures::Stream;
@@ -55,7 +56,6 @@ use tempfile::TempDir;
 #[cfg(feature = "parquet")]
 pub use datafusion_common::test_util::parquet_test_data;
 pub use datafusion_common::test_util::{arrow_test_data, get_data_dir};
-use datafusion_physical_expr::EquivalenceProperties;
 
 /// Scan an empty data source, mainly used in tests
 pub fn scan_empty(
@@ -246,6 +246,7 @@ impl UnboundedExec {
         }
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         schema: SchemaRef,
         batch_produce: Option<usize>,

datafusion/core/tests/custom_sources.rs

@@ -86,6 +86,7 @@ impl CustomExecutionPlan {
         Self { projection, cache }
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(schema: SchemaRef) -> PlanPropertiesCache {
         let eq_properties = EquivalenceProperties::new(schema);
         PlanPropertiesCache::new(

datafusion/core/tests/custom_sources_cases/provider_filter_pushdown.rs

@@ -67,6 +67,7 @@ impl CustomPlan {
         Self { batches, cache }
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(schema: SchemaRef) -> PlanPropertiesCache {
         let eq_properties = EquivalenceProperties::new(schema);
         PlanPropertiesCache::new(

datafusion/core/tests/custom_sources_cases/statistics.rs

@@ -61,6 +61,7 @@ impl StatisticsValidation {
         }
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(schema: SchemaRef) -> PlanPropertiesCache {
         let eq_properties = EquivalenceProperties::new(schema);
 

datafusion/core/tests/user_defined/user_defined_plan.rs

@@ -81,6 +81,7 @@ use datafusion::{
         UserDefinedLogicalNodeCore,
     },
     optimizer::{optimize_children, OptimizerConfig, OptimizerRule},
+    physical_expr::EquivalenceProperties,
     physical_plan::{
         DisplayAs, DisplayFormatType, Distribution, ExecutionMode, ExecutionPlan,
         Partitioning, PlanPropertiesCache, RecordBatchStream, SendableRecordBatchStream,
@@ -91,7 +92,6 @@ use datafusion::{
 };
 
 use async_trait::async_trait;
-use datafusion_physical_expr::EquivalenceProperties;
 use futures::{Stream, StreamExt};
 
 /// Execute the specified sql and return the resulting record batches
@@ -421,6 +421,7 @@ impl TopKExec {
         Self { input, k, cache }
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(schema: SchemaRef) -> PlanPropertiesCache {
         let eq_properties = EquivalenceProperties::new(schema);
 

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

@@ -506,6 +506,7 @@ impl AggregateExec {
         true
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         input: &Arc<dyn ExecutionPlan>,
         schema: SchemaRef,
@@ -1629,6 +1630,7 @@ mod tests {
             Self { yield_first, cache }
         }
 
+        /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
         fn create_cache(schema: SchemaRef) -> PlanPropertiesCache {
             let eq_properties = EquivalenceProperties::new(schema);
             PlanPropertiesCache::new(

datafusion/physical-plan/src/analyze.rs

@@ -81,6 +81,7 @@ impl AnalyzeExec {
         &self.input
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         input: &Arc<dyn ExecutionPlan>,
         schema: SchemaRef,

datafusion/physical-plan/src/coalesce_batches.rs

@@ -73,6 +73,7 @@ impl CoalesceBatchesExec {
         self.target_batch_size
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(input: &Arc<dyn ExecutionPlan>) -> PlanPropertiesCache {
         // The coalesce batches operator does not make any changes to the
         // partitioning of its input.

datafusion/physical-plan/src/coalesce_partitions.rs

@@ -57,6 +57,7 @@ impl CoalescePartitionsExec {
         &self.input
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(input: &Arc<dyn ExecutionPlan>) -> PlanPropertiesCache {
         // Coalescing partitions loses existing orderings:
         let mut eq_properties = input.equivalence_properties().clone();

datafusion/physical-plan/src/empty.rs

@@ -72,6 +72,7 @@ impl EmptyExec {
         Partitioning::UnknownPartitioning(n_partitions)
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(schema: SchemaRef, n_partitions: usize) -> PlanPropertiesCache {
         let eq_properties = EquivalenceProperties::new(schema);
         let output_partitioning = Self::output_partitioning_helper(n_partitions);

datafusion/physical-plan/src/explain.rs

@@ -72,6 +72,7 @@ impl ExplainExec {
         self.verbose
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(schema: SchemaRef) -> PlanPropertiesCache {
         let eq_properties = EquivalenceProperties::new(schema);
         PlanPropertiesCache::new(

datafusion/physical-plan/src/filter.rs

@@ -115,6 +115,7 @@ impl FilterExec {
         self.default_selectivity
     }
 
+    /// Calculates `Statistics` for `FilterExec`, by applying selectivity (either default, or estimated) to input statistics.
     fn statistics_helper(
         input: &Arc<dyn ExecutionPlan>,
         predicate: &Arc<dyn PhysicalExpr>,
@@ -157,6 +158,7 @@ impl FilterExec {
         })
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         input: &Arc<dyn ExecutionPlan>,
         predicate: &Arc<dyn PhysicalExpr>,

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

@@ -98,6 +98,7 @@ impl CrossJoinExec {
         &self.right
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         left: &Arc<dyn ExecutionPlan>,
         right: &Arc<dyn ExecutionPlan>,

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

@@ -37,7 +37,7 @@ use crate::{
         check_join_is_valid, estimate_join_statistics, get_final_indices_from_bit_map,
         need_produce_result_in_final, partitioned_join_output_partitioning,
         BuildProbeJoinMetrics, ColumnIndex, JoinFilter, JoinHashMap, JoinHashMapOffset,
-        JoinHashMapType, JoinOn, StatefulStreamResult,
+        JoinHashMapType, JoinOn, JoinOnRef, StatefulStreamResult,
     },
     metrics::{ExecutionPlanMetricsSet, MetricsSet},
     DisplayAs, DisplayFormatType, Distribution, ExecutionMode, ExecutionPlan,
@@ -65,7 +65,6 @@ use datafusion_execution::TaskContext;
 use datafusion_physical_expr::equivalence::join_equivalence_properties;
 use datafusion_physical_expr::PhysicalExprRef;
 
-use crate::joins::utils::JoinOnRef;
 use ahash::RandomState;
 use futures::{ready, Stream, StreamExt, TryStreamExt};
 
@@ -406,6 +405,7 @@ impl HashJoinExec {
         JoinSide::Right
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         left: &Arc<dyn ExecutionPlan>,
         right: &Arc<dyn ExecutionPlan>,

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

@@ -144,6 +144,7 @@ impl NestedLoopJoinExec {
         &self.join_type
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         left: &Arc<dyn ExecutionPlan>,
         right: &Arc<dyn ExecutionPlan>,

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

@@ -200,6 +200,7 @@ impl SortMergeJoinExec {
         self.left.as_ref()
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         left: &Arc<dyn ExecutionPlan>,
         right: &Arc<dyn ExecutionPlan>,

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

@@ -252,6 +252,7 @@ impl SymmetricHashJoinExec {
         })
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         left: &Arc<dyn ExecutionPlan>,
         right: &Arc<dyn ExecutionPlan>,

datafusion/physical-plan/src/limit.rs

@@ -82,6 +82,7 @@ impl GlobalLimitExec {
         self.fetch
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(input: &Arc<dyn ExecutionPlan>) -> PlanPropertiesCache {
         PlanPropertiesCache::new(
             input.equivalence_properties().clone(), // Equivalence Properties
@@ -292,6 +293,7 @@ impl LocalLimitExec {
         self.fetch
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(input: &Arc<dyn ExecutionPlan>) -> PlanPropertiesCache {
         PlanPropertiesCache::new(
             input.equivalence_properties().clone(), // Equivalence Properties

datafusion/physical-plan/src/memory.rs

@@ -204,6 +204,7 @@ impl MemoryExec {
         self.schema.clone()
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         schema: SchemaRef,
         orderings: &[LexOrdering],

datafusion/physical-plan/src/placeholder_row.rs

@@ -94,6 +94,7 @@ impl PlaceholderRowExec {
         Partitioning::UnknownPartitioning(n_partitions)
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(schema: SchemaRef, n_partitions: usize) -> PlanPropertiesCache {
         let eq_properties = EquivalenceProperties::new(schema);
         // Get output partitioning:

datafusion/physical-plan/src/projection.rs

@@ -113,6 +113,7 @@ impl ProjectionExec {
         &self.input
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         input: &Arc<dyn ExecutionPlan>,
         projection_mapping: &ProjectionMapping,

datafusion/physical-plan/src/recursive_query.rs

@@ -94,6 +94,7 @@ impl RecursiveQueryExec {
         })
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(schema: SchemaRef) -> PlanPropertiesCache {
         let eq_properties = EquivalenceProperties::new(schema);
 

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

@@ -616,12 +616,8 @@ impl RepartitionExec {
         input: &Arc<dyn ExecutionPlan>,
         preserve_order: bool,
     ) -> Vec<bool> {
-        if preserve_order {
-            vec![true]
-        } else {
-            // We preserve ordering when input partitioning is 1
-            vec![input.output_partitioning().partition_count() <= 1]
-        }
+        // We preserve ordering when repartition is order preserving variant or input partitioning is 1
+        vec![preserve_order || input.output_partitioning().partition_count() <= 1]
     }
 
     fn eq_properties_helper(
@@ -637,6 +633,7 @@ impl RepartitionExec {
         eq_properties
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         input: &Arc<dyn ExecutionPlan>,
         partitioning: Partitioning,

datafusion/physical-plan/src/sorts/partial_sort.rs

@@ -180,6 +180,7 @@ impl PartialSortExec {
         }
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         input: &Arc<dyn ExecutionPlan>,
         sort_exprs: LexOrdering,

datafusion/physical-plan/src/sorts/sort.rs

@@ -786,6 +786,7 @@ impl SortExec {
         }
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         input: &Arc<dyn ExecutionPlan>,
         sort_exprs: LexOrdering,

datafusion/physical-plan/src/sorts/sort_preserving_merge.rs

@@ -110,6 +110,7 @@ impl SortPreservingMergeExec {
         self.fetch
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(input: &Arc<dyn ExecutionPlan>) -> PlanPropertiesCache {
         PlanPropertiesCache::new(
             input.equivalence_properties().clone(), // Equivalence Properties

datafusion/physical-plan/src/streaming.rs

@@ -127,6 +127,7 @@ impl StreamingTableExec {
         self.infinite
     }
 
+    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
     fn create_cache(
         schema: SchemaRef,
         orderings: &[LexOrdering],