## Coverage Report

#### Created: 2025-11-22 06:00

next uncovered line (L), next uncovered region (R), next uncovered branch (B)

    /Users/andrewlamb/Software/datafusion/datafusion/physical-expr/src/partitioning.rs

    Line
    Count
    Source
    1
    // Licensed to the Apache Software Foundation (ASF) under one
    2
    // or more contributor license agreements.  See the NOTICE file
    3
    // distributed with this work for additional information
    4
    // regarding copyright ownership.  The ASF licenses this file
    5
    // to you under the Apache License, Version 2.0 (the
    6
    // "License"); you may not use this file except in compliance
    7
    // with the License.  You may obtain a copy of the License at
    8
    //
    9
    //   http://www.apache.org/licenses/LICENSE-2.0
    10
    //
    11
    // Unless required by applicable law or agreed to in writing,
    12
    // software distributed under the License is distributed on an
    13
    // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
    14
    // KIND, either express or implied.  See the License for the
    15
    // specific language governing permissions and limitations
    16
    // under the License.
    17
    18
    //! [`Partitioning`] and [`Distribution`] for `ExecutionPlans`
    19
    20
    use crate::{
    21
        equivalence::ProjectionMapping, expressions::UnKnownColumn, physical_exprs_equal,
    22
        EquivalenceProperties, PhysicalExpr,
    23
    };
    24
    use datafusion_physical_expr_common::physical_expr::format_physical_expr_list;
    25
    use std::fmt;
    26
    use std::fmt::Display;
    27
    use std::sync::Arc;
    28
    29
    /// Output partitioning supported by [`ExecutionPlan`]s.
    30
    ///
    31
    /// Calling [`ExecutionPlan::execute`] produce one or more independent streams of
    32
    /// [`RecordBatch`]es in parallel, referred to as partitions. The streams are Rust
    33
    /// `async` [`Stream`]s (a special kind of future). The number of output
    34
    /// partitions varies based on the input and the operation performed.
    35
    ///
    36
    /// For example, an `ExecutionPlan` that has output partitioning of 3 will
    37
    /// produce 3 distinct output streams as the result of calling
    38
    /// `ExecutionPlan::execute(0)`, `ExecutionPlan::execute(1)`, and
    39
    /// `ExecutionPlan::execute(2)`, as shown below:
    40
    ///
    41
    /// ```text
    42
    ///                                                   ...         ...        ...
    43
    ///               ...                                  ▲           ▲           ▲
    44
    ///                                                    │           │           │
    45
    ///                ▲                                   │           │           │
    46
    ///                │                                   │           │           │
    47
    ///                │                               ┌───┴────┐  ┌───┴────┐  ┌───┴────┐
    48
    ///     ┌────────────────────┐                     │ Stream │  │ Stream │  │ Stream │
    49
    ///     │   ExecutionPlan    │                     │  (0)   │  │  (1)   │  │  (2)   │
    50
    ///     └────────────────────┘                     └────────┘  └────────┘  └────────┘
    51
    ///                ▲                                   ▲           ▲           ▲
    52
    ///                │                                   │           │           │
    53
    ///     ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─                          │           │           │
    54
    ///             Input        │                         │           │           │
    55
    ///     └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─                          │           │           │
    56
    ///                ▲                               ┌ ─ ─ ─ ─   ┌ ─ ─ ─ ─   ┌ ─ ─ ─ ─
    57
    ///                │                                 Input  │    Input  │    Input  │
    58
    ///                │                               │ Stream    │ Stream    │ Stream
    59
    ///                                                   (0)   │     (1)   │     (2)   │
    60
    ///               ...                              └ ─ ▲ ─ ─   └ ─ ▲ ─ ─   └ ─ ▲ ─ ─
    61
    ///                                                    │           │           │
    62
    ///                                                    │           │           │
    63
    ///                                                    │           │           │
    64
    ///
    65
    /// ExecutionPlan with 1 input                      3 (async) streams, one for each
    66
    /// that has 3 partitions, which itself             output partition
    67
    /// has 3 output partitions
    68
    /// ```
    69
    ///
    70
    /// It is common (but not required) that an `ExecutionPlan` has the same number
    71
    /// of input partitions as output partitions. However, some plans have different
    72
    /// numbers such as the `RepartitionExec` that redistributes batches from some
    73
    /// number of inputs to some number of outputs
    74
    ///
    75
    /// ```text
    76
    ///               ...                                     ...         ...        ...
    77
    ///
    78
    ///                                                        ▲           ▲           ▲
    79
    ///                ▲                                       │           │           │
    80
    ///                │                                       │           │           │
    81
    ///       ┌────────┴───────────┐                           │           │           │
    82
    ///       │  RepartitionExec   │                      ┌────┴───┐  ┌────┴───┐  ┌────┴───┐
    83
    ///       └────────────────────┘                      │ Stream │  │ Stream │  │ Stream │
    84
    ///                ▲                                  │  (0)   │  │  (1)   │  │  (2)   │
    85
    ///                │                                  └────────┘  └────────┘  └────────┘
    86
    ///                │                                       ▲           ▲           ▲
    87
    ///                ...                                     │           │           │
    88
    ///                                                        └──────────┐│┌──────────┘
    89
    ///                                                                   │││
    90
    ///                                                                   │││
    91
    /// RepartitionExec with 1 input
    92
    /// partition and 3 output partitions                 3 (async) streams, that internally
    93
    ///                                                    pull from the same input stream
    94
    ///                                                                  ...
    95
    /// ```
    96
    ///
    97
    /// # Additional Examples
    98
    ///
    99
    /// A simple `FileScanExec` might produce one output stream (partition) for each
    100
    /// file (note the actual DataFusion file scanners can read individual files in
    101
    /// parallel, potentially producing multiple partitions per file)
    102
    ///
    103
    /// Plans such as `SortPreservingMerge` produce a single output stream
    104
    /// (1 output partition) by combining some number of input streams (input partitions)
    105
    ///
    106
    /// Plans such as `FilterExec` produce the same number of output streams
    107
    /// (partitions) as input streams (partitions).
    108
    ///
    109
    /// [`RecordBatch`]: arrow::record_batch::RecordBatch
    110
    /// [`ExecutionPlan::execute`]: https://docs.rs/datafusion/latest/datafusion/physical_plan/trait.ExecutionPlan.html#tymethod.execute
    111
    /// [`ExecutionPlan`]: https://docs.rs/datafusion/latest/datafusion/physical_plan/trait.ExecutionPlan.html
    112
    /// [`Stream`]: https://docs.rs/futures/latest/futures/stream/trait.Stream.html
    113
    #[derive(Debug, Clone)]
    114
    pub enum Partitioning {
    115
        /// Allocate batches using a round-robin algorithm and the specified number of partitions
    116
        RoundRobinBatch(usize),
    117
        /// Allocate rows based on a hash of one of more expressions and the specified number of
    118
        /// partitions
    119
        Hash(Vec<Arc<dyn PhysicalExpr>>, usize),
    120
        /// Unknown partitioning scheme with a known number of partitions
    121
        UnknownPartitioning(usize),
    122
    }
    123
    124
    impl Display for Partitioning {
    125
    0
        fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
    126
    0
            match self {
    127
    0
                Partitioning::RoundRobinBatch(size) => write!(f, "RoundRobinBatch({size})"),
    128
    0
                Partitioning::Hash(phy_exprs, size) => {
    129
    0
                    let phy_exprs_str = phy_exprs
    130
    0
                        .iter()
    131
    0
                        .map(|e| format!("{e}"))
    132
    0
                        .collect::<Vec<String>>()
    133
    0
                        .join(", ");
    134
    0
                    write!(f, "Hash([{phy_exprs_str}], {size})")
    135
                }
    136
    0
                Partitioning::UnknownPartitioning(size) => {
    137
    0
                    write!(f, "UnknownPartitioning({size})")
    138
                }
    139
            }
    140
    0
        }
    141
    }
    142
    impl Partitioning {
    143
        /// Returns the number of partitions in this partitioning scheme
    144
    0
        pub fn partition_count(&self) -> usize {
    145
            use Partitioning::*;
    146
    0
            match self {
    147
    0
                RoundRobinBatch(n) | Hash(_, n) | UnknownPartitioning(n) => *n,
    148
            }
    149
    0
        }
    150
    151
        /// Returns true when the guarantees made by this [`Partitioning`] are sufficient to
    152
        /// satisfy the partitioning scheme mandated by the `required` [`Distribution`].
    153
    0
        pub fn satisfy(
    154
    0
            &self,
    155
    0
            required: &Distribution,
    156
    0
            eq_properties: &EquivalenceProperties,
    157
    0
        ) -> bool {
    158
    0
            match required {
    159
    0
                Distribution::UnspecifiedDistribution => true,
    160
    0
                Distribution::SinglePartition if self.partition_count() == 1 => true,
    161
                // When partition count is 1, hash requirement is satisfied.
    162
    0
                Distribution::HashPartitioned(_) if self.partition_count() == 1 => true,
    163
    0
                Distribution::HashPartitioned(required_exprs) => {
    164
    0
                    match self {
    165
                        // Here we do not check the partition count for hash partitioning and assumes the partition count
    166
                        // and hash functions in the system are the same. In future if we plan to support storage partition-wise joins,
    167
                        // then we need to have the partition count and hash functions validation.
    168
    0
                        Partitioning::Hash(partition_exprs, _) => {
    169
    0
                            let fast_match =
    170
    0
                                physical_exprs_equal(required_exprs, partition_exprs);
    171
                            // If the required exprs do not match, need to leverage the eq_properties provided by the child
    172
                            // and normalize both exprs based on the equivalent groups.
    173
    0
                            if !fast_match {
    174
    0
                                let eq_groups = eq_properties.eq_group();
    175
    0
                                if !eq_groups.is_empty() {
    176
    0
                                    let normalized_required_exprs = required_exprs
    177
    0
                                        .iter()
    178
    0
                                        .map(|e| eq_groups.normalize_expr(Arc::clone(e)))
    179
    0
                                        .collect::<Vec<_>>();
    180
    0
                                    let normalized_partition_exprs = partition_exprs
    181
    0
                                        .iter()
    182
    0
                                        .map(|e| eq_groups.normalize_expr(Arc::clone(e)))
    183
    0
                                        .collect::<Vec<_>>();
    184
    0
                                    return physical_exprs_equal(
    185
    0
                                        &normalized_required_exprs,
    186
    0
                                        &normalized_partition_exprs,
    187
                                    );
    188
    0
                                }
    189
    0
                            }
    190
    0
                            fast_match
    191
                        }
    192
    0
                        _ => false,
    193
                    }
    194
                }
    195
    0
                _ => false,
    196
            }
    197
    0
        }
    198
    199
        /// Calculate the output partitioning after applying the given projection.
    200
    0
        pub fn project(
    201
    0
            &self,
    202
    0
            mapping: &ProjectionMapping,
    203
    0
            input_eq_properties: &EquivalenceProperties,
    204
    0
        ) -> Self {
    205
    0
            if let Partitioning::Hash(exprs, part) = self {
    206
    0
                let normalized_exprs = input_eq_properties
    207
    0
                    .project_expressions(exprs, mapping)
    208
    0
                    .zip(exprs)
    209
    0
                    .map(|(proj_expr, expr)| {
    210
    0
                        proj_expr.unwrap_or_else(|| {
    211
    0
                            Arc::new(UnKnownColumn::new(&expr.to_string()))
    212
    0
                        })
    213
    0
                    })
    214
    0
                    .collect();
    215
    0
                Partitioning::Hash(normalized_exprs, *part)
    216
            } else {
    217
    0
                self.clone()
    218
            }
    219
    0
        }
    220
    }
    221
    222
    impl PartialEq for Partitioning {
    223
    0
        fn eq(&self, other: &Partitioning) -> bool {
    224
    0
            match (self, other) {
    225
                (
    226
    0
                    Partitioning::RoundRobinBatch(count1),
    227
    0
                    Partitioning::RoundRobinBatch(count2),
    228
    0
                ) if count1 == count2 => true,
    229
    0
                (Partitioning::Hash(exprs1, count1), Partitioning::Hash(exprs2, count2))
    230
    0
                    if physical_exprs_equal(exprs1, exprs2) && (count1 == count2) =>
    231
                {
    232
    0
                    true
    233
                }
    234
    0
                _ => false,
    235
            }
    236
    0
        }
    237
    }
    238
    239
    /// How data is distributed amongst partitions. See [`Partitioning`] for more
    240
    /// details.
    241
    #[derive(Debug, Clone)]
    242
    pub enum Distribution {
    243
        /// Unspecified distribution
    244
        UnspecifiedDistribution,
    245
        /// A single partition is required
    246
        SinglePartition,
    247
        /// Requires children to be distributed in such a way that the same
    248
        /// values of the keys end up in the same partition
    249
        HashPartitioned(Vec<Arc<dyn PhysicalExpr>>),
    250
    }
    251
    252
    impl Distribution {
    253
        /// Creates a `Partitioning` that satisfies this `Distribution`
    254
    0
        pub fn create_partitioning(self, partition_count: usize) -> Partitioning {
    255
    0
            match self {
    256
                Distribution::UnspecifiedDistribution => {
    257
    0
                    Partitioning::UnknownPartitioning(partition_count)
    258
                }
    259
    0
                Distribution::SinglePartition => Partitioning::UnknownPartitioning(1),
    260
    0
                Distribution::HashPartitioned(expr) => {
    261
    0
                    Partitioning::Hash(expr, partition_count)
    262
                }
    263
            }
    264
    0
        }
    265
    }
    266
    267
    impl Display for Distribution {
    268
    0
        fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
    269
    0
            match self {
    270
    0
                Distribution::UnspecifiedDistribution => write!(f, "Unspecified"),
    271
    0
                Distribution::SinglePartition => write!(f, "SinglePartition"),
    272
    0
                Distribution::HashPartitioned(exprs) => {
    273
    0
                    write!(f, "HashPartitioned[{}])", format_physical_expr_list(exprs))
    274
                }
    275
            }
    276
    0
        }
    277
    }
    278
    279
    #[cfg(test)]
    280
    mod tests {
    281
    282
        use super::*;
    283
        use crate::expressions::Column;
    284
    285
        use arrow::datatypes::{DataType, Field, Schema};
    286
        use datafusion_common::Result;
    287
    288
        #[test]
    289
    0
        fn partitioning_satisfy_distribution() -> Result<()> {
    290
    0
            let schema = Arc::new(Schema::new(vec![
    291
    0
                Field::new("column_1", DataType::Int64, false),
    292
    0
                Field::new("column_2", DataType::Utf8, false),
    293
            ]));
    294
    295
    0
            let partition_exprs1: Vec<Arc<dyn PhysicalExpr>> = vec![
    296
    0
                Arc::new(Column::new_with_schema("column_1", &schema).unwrap()),
    297
    0
                Arc::new(Column::new_with_schema("column_2", &schema).unwrap()),
    298
            ];
    299
    300
    0
            let partition_exprs2: Vec<Arc<dyn PhysicalExpr>> = vec![
    301
    0
                Arc::new(Column::new_with_schema("column_2", &schema).unwrap()),
    302
    0
                Arc::new(Column::new_with_schema("column_1", &schema).unwrap()),
    303
            ];
    304
    305
    0
            let distribution_types = vec![
    306
    0
                Distribution::UnspecifiedDistribution,
    307
    0
                Distribution::SinglePartition,
    308
    0
                Distribution::HashPartitioned(partition_exprs1.clone()),
    309
            ];
    310
    311
    0
            let single_partition = Partitioning::UnknownPartitioning(1);
    312
    0
            let unspecified_partition = Partitioning::UnknownPartitioning(10);
    313
    0
            let round_robin_partition = Partitioning::RoundRobinBatch(10);
    314
    0
            let hash_partition1 = Partitioning::Hash(partition_exprs1, 10);
    315
    0
            let hash_partition2 = Partitioning::Hash(partition_exprs2, 10);
    316
    0
            let eq_properties = EquivalenceProperties::new(schema);
    317
    318
    0
            for distribution in distribution_types {
    319
    0
                let result = (
    320
    0
                    single_partition.satisfy(&distribution, &eq_properties),
    321
    0
                    unspecified_partition.satisfy(&distribution, &eq_properties),
    322
    0
                    round_robin_partition.satisfy(&distribution, &eq_properties),
    323
    0
                    hash_partition1.satisfy(&distribution, &eq_properties),
    324
    0
                    hash_partition2.satisfy(&distribution, &eq_properties),
    325
    0
                );
    326
    327
    0
                match distribution {
    328
                    Distribution::UnspecifiedDistribution => {
    329
    0
                        assert_eq!(result, (true, true, true, true, true))
    330
                    }
    331
                    Distribution::SinglePartition => {
    332
    0
                        assert_eq!(result, (true, false, false, false, false))
    333
                    }
    334
                    Distribution::HashPartitioned(_) => {
    335
    0
                        assert_eq!(result, (true, false, false, true, false))
    336
                    }
    337
                }
    338
            }
    339
    340
    0
            Ok(())
    341
    0
        }
    342
    }