Method group_by_partition_values

datafusion/datasource/src/file_groups.rs:492–543

ScalarValue has interior mutability but is intentionally used as hash key

(
        self,
        max_target_partitions: usize,
    )

    /// number of unique partition values is less than the target.
    #[allow(clippy::allow_attributes, clippy::mutable_key_type)] // ScalarValue has interior mutability but is intentionally used as hash key
    pub fn group_by_partition_values(
        self,
        max_target_partitions: usize,
    ) -> Vec<FileGroup> {
        if self.is_empty() || max_target_partitions == 0 {
            return vec![];
        }

        let mut partition_groups: HashMap<
            Vec<datafusion_common::ScalarValue>,
            Vec<PartitionedFile>,
        > = HashMap::new();

        for file in self.files {
            partition_groups
                .entry(file.partition_values.clone())
                .or_default()
                .push(file);
        }

        let num_unique_partitions = partition_groups.len();

        // Sort for deterministic bucket assignment across query executions.
        let mut sorted_partitions: Vec<_> = partition_groups.into_iter().collect();
        let sort_options =
            vec![
                SortOptions::default();
                sorted_partitions.first().map(|(k, _)| k.len()).unwrap_or(0)
            ];
        sorted_partitions.sort_by(|a, b| {
            compare_rows(&a.0, &b.0, &sort_options).unwrap_or(Ordering::Equal)
        });

        if num_unique_partitions <= max_target_partitions {
            sorted_partitions
                .into_iter()
                .map(|(_, files)| FileGroup::new(files))
                .collect()
        } else {
            // Merge into max_target_partitions buckets using round-robin.
            // This maintains grouping by partition value as we are merging groups which already
            // contain all values for a partition key.
            let mut target_groups = vec![vec![]; max_target_partitions];

            for (idx, (_, files)) in sorted_partitions.into_iter().enumerate() {
                let bucket = idx % max_target_partitions;
                target_groups[bucket].extend(files);
            }

            target_groups.into_iter().map(FileGroup::new).collect()
        }
    }
}

impl Index<usize> for FileGroup {