diff --git a/datafusion/core/src/physical_optimizer/enforce_distribution.rs b/datafusion/core/src/physical_optimizer/enforce_distribution.rs
index 70fb75a50f99d..e1f715b634a90 100644
--- a/datafusion/core/src/physical_optimizer/enforce_distribution.rs
+++ b/datafusion/core/src/physical_optimizer/enforce_distribution.rs
@@ -1679,6 +1679,8 @@ impl TreeNode for PlanWithKeyRequirements {
     }
 }
 
+/// Since almost all of these tests explicitly use `ParquetExec` they only run with the parquet  feature flag on
+#[cfg(feature = "parquet")]
 #[cfg(test)]
 mod tests {
     use std::ops::Deref;
@@ -1688,7 +1690,6 @@ mod tests {
     use crate::datasource::listing::PartitionedFile;
     use crate::datasource::object_store::ObjectStoreUrl;
     use crate::datasource::physical_plan::FileScanConfig;
-    #[cfg(feature = "parquet")]
     use crate::datasource::physical_plan::ParquetExec;
     use crate::physical_optimizer::enforce_sorting::EnforceSorting;
     use crate::physical_optimizer::output_requirements::OutputRequirements;
@@ -1828,12 +1829,10 @@ mod tests {
         ]))
     }
 
-    #[cfg(feature = "parquet")]
     fn parquet_exec() -> Arc<ParquetExec> {
         parquet_exec_with_sort(vec![])
     }
 
-    #[cfg(feature = "parquet")]
     fn parquet_exec_with_sort(
         output_ordering: Vec<Vec<PhysicalSortExpr>>,
     ) -> Arc<ParquetExec> {
@@ -1854,13 +1853,11 @@ mod tests {
         ))
     }
 
-    #[cfg(feature = "parquet")]
     fn parquet_exec_multiple() -> Arc<ParquetExec> {
         parquet_exec_multiple_sorted(vec![])
     }
 
     // Created a sorted parquet exec with multiple files
-    #[cfg(feature = "parquet")]
     fn parquet_exec_multiple_sorted(
         output_ordering: Vec<Vec<PhysicalSortExpr>>,
     ) -> Arc<ParquetExec> {
@@ -2215,7 +2212,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn multi_hash_joins() -> Result<()> {
         let left = parquet_exec();
         let alias_pairs: Vec<(String, String)> = vec![
@@ -2378,7 +2374,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn multi_joins_after_alias() -> Result<()> {
         let left = parquet_exec();
         let right = parquet_exec();
@@ -2458,7 +2453,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn multi_joins_after_multi_alias() -> Result<()> {
         let left = parquet_exec();
         let right = parquet_exec();
@@ -2514,7 +2508,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn join_after_agg_alias() -> Result<()> {
         // group by (a as a1)
         let left = aggregate_exec_with_alias(
@@ -2554,7 +2547,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn hash_join_key_ordering() -> Result<()> {
         // group by (a as a1, b as b1)
         let left = aggregate_exec_with_alias(
@@ -2607,7 +2599,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn multi_hash_join_key_ordering() -> Result<()> {
         let left = parquet_exec();
         let alias_pairs: Vec<(String, String)> = vec![
@@ -2724,7 +2715,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn reorder_join_keys_to_left_input() -> Result<()> {
         let left = parquet_exec();
         let alias_pairs: Vec<(String, String)> = vec![
@@ -2855,7 +2845,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn reorder_join_keys_to_right_input() -> Result<()> {
         let left = parquet_exec();
         let alias_pairs: Vec<(String, String)> = vec![
@@ -2981,7 +2970,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn multi_smj_joins() -> Result<()> {
         let left = parquet_exec();
         let alias_pairs: Vec<(String, String)> = vec![
@@ -3255,7 +3243,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn smj_join_key_ordering() -> Result<()> {
         // group by (a as a1, b as b1)
         let left = aggregate_exec_with_alias(
@@ -3351,7 +3338,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn merge_does_not_need_sort() -> Result<()> {
         // see https://github.com/apache/arrow-datafusion/issues/4331
         let schema = schema();
@@ -3392,7 +3378,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn union_to_interleave() -> Result<()> {
         // group by (a as a1)
         let left = aggregate_exec_with_alias(
@@ -3434,7 +3419,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn added_repartition_to_single_partition() -> Result<()> {
         let alias = vec![("a".to_string(), "a".to_string())];
         let plan = aggregate_exec_with_alias(parquet_exec(), alias);
@@ -3453,7 +3437,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn repartition_deepest_node() -> Result<()> {
         let alias = vec![("a".to_string(), "a".to_string())];
         let plan = aggregate_exec_with_alias(filter_exec(parquet_exec()), alias);
@@ -3473,7 +3456,7 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
+
     fn repartition_unsorted_limit() -> Result<()> {
         let plan = limit_exec(filter_exec(parquet_exec()));
 
@@ -3493,7 +3476,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn repartition_sorted_limit() -> Result<()> {
         let schema = schema();
         let sort_key = vec![PhysicalSortExpr {
@@ -3516,7 +3498,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn repartition_sorted_limit_with_filter() -> Result<()> {
         let schema = schema();
         let sort_key = vec![PhysicalSortExpr {
@@ -3542,7 +3523,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn repartition_ignores_limit() -> Result<()> {
         let alias = vec![("a".to_string(), "a".to_string())];
         let plan = aggregate_exec_with_alias(
@@ -3573,7 +3553,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn repartition_ignores_union() -> Result<()> {
         let plan = union_exec(vec![parquet_exec(); 5]);
 
@@ -3593,7 +3572,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn repartition_through_sort_preserving_merge() -> Result<()> {
         // sort preserving merge with non-sorted input
         let schema = schema();
@@ -3616,7 +3594,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn repartition_ignores_sort_preserving_merge() -> Result<()> {
         // sort preserving merge already sorted input,
         let schema = schema();
@@ -3648,7 +3625,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn repartition_ignores_sort_preserving_merge_with_union() -> Result<()> {
         // 2 sorted parquet files unioned (partitions are concatenated, sort is preserved)
         let schema = schema();
@@ -3681,7 +3657,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn repartition_does_not_destroy_sort() -> Result<()> {
         //  SortRequired
         //    Parquet(sorted)
@@ -3707,7 +3682,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn repartition_does_not_destroy_sort_more_complex() -> Result<()> {
         // model a more complicated scenario where one child of a union can be repartitioned for performance
         // but the other can not be
@@ -3746,7 +3720,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn repartition_transitively_with_projection() -> Result<()> {
         let schema = schema();
         let proj_exprs = vec![(
@@ -3789,7 +3762,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn repartition_ignores_transitively_with_projection() -> Result<()> {
         let schema = schema();
         let sort_key = vec![PhysicalSortExpr {
@@ -3820,7 +3792,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn repartition_transitively_past_sort_with_projection() -> Result<()> {
         let schema = schema();
         let sort_key = vec![PhysicalSortExpr {
@@ -3850,7 +3821,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn repartition_transitively_past_sort_with_filter() -> Result<()> {
         let schema = schema();
         let sort_key = vec![PhysicalSortExpr {
@@ -3925,7 +3895,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn parallelization_single_partition() -> Result<()> {
         let alias = vec![("a".to_string(), "a".to_string())];
         let plan_parquet = aggregate_exec_with_alias(parquet_exec(), alias.clone());
@@ -4014,7 +3983,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn parallelization_two_partitions() -> Result<()> {
         let alias = vec![("a".to_string(), "a".to_string())];
         let plan_parquet =
@@ -4042,7 +4010,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn parallelization_two_partitions_into_four() -> Result<()> {
         let alias = vec![("a".to_string(), "a".to_string())];
         let plan_parquet =
@@ -4070,7 +4037,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn parallelization_sorted_limit() -> Result<()> {
         let schema = schema();
         let sort_key = vec![PhysicalSortExpr {
@@ -4103,7 +4069,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn parallelization_limit_with_filter() -> Result<()> {
         let schema = schema();
         let sort_key = vec![PhysicalSortExpr {
@@ -4149,7 +4114,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn parallelization_ignores_limit() -> Result<()> {
         let alias = vec![("a".to_string(), "a".to_string())];
         let plan_parquet = aggregate_exec_with_alias(
@@ -4200,7 +4164,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn parallelization_union_inputs() -> Result<()> {
         let plan_parquet = union_exec(vec![parquet_exec(); 5]);
         let plan_csv = union_exec(vec![csv_exec(); 5]);
@@ -4230,7 +4193,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn parallelization_prior_to_sort_preserving_merge() -> Result<()> {
         let schema = schema();
         let sort_key = vec![PhysicalSortExpr {
@@ -4261,7 +4223,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn parallelization_sort_preserving_merge_with_union() -> Result<()> {
         let schema = schema();
         let sort_key = vec![PhysicalSortExpr {
@@ -4296,7 +4257,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn parallelization_does_not_benefit() -> Result<()> {
         let schema = schema();
         let sort_key = vec![PhysicalSortExpr {
@@ -4325,7 +4285,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn parallelization_ignores_transitively_with_projection_parquet() -> Result<()> {
         // sorted input
         let schema = schema();
@@ -4406,7 +4365,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn remove_redundant_roundrobins() -> Result<()> {
         let input = parquet_exec();
         let repartition = repartition_exec(repartition_exec(input));
@@ -4457,7 +4415,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn do_not_preserve_ordering_through_repartition() -> Result<()> {
         let schema = schema();
         let sort_key = vec![PhysicalSortExpr {
@@ -4490,7 +4447,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn do_not_preserve_ordering_through_repartition2() -> Result<()> {
         let schema = schema();
         let sort_key = vec![PhysicalSortExpr {
@@ -4529,7 +4485,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn do_not_preserve_ordering_through_repartition3() -> Result<()> {
         let schema = schema();
         let sort_key = vec![PhysicalSortExpr {
@@ -4552,7 +4507,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn do_not_put_sort_when_input_is_invalid() -> Result<()> {
         let schema = schema();
         let sort_key = vec![PhysicalSortExpr {
@@ -4591,7 +4545,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn put_sort_when_input_is_valid() -> Result<()> {
         let schema = schema();
         let sort_key = vec![PhysicalSortExpr {
@@ -4634,7 +4587,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn do_not_add_unnecessary_hash() -> Result<()> {
         let schema = schema();
         let sort_key = vec![PhysicalSortExpr {
@@ -4690,7 +4642,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn optimize_away_unnecessary_repartition() -> Result<()> {
         let physical_plan = coalesce_partitions_exec(repartition_exec(parquet_exec()));
         let expected = &[
@@ -4710,7 +4661,6 @@ mod tests {
     }
 
     #[test]
-    #[cfg(feature = "parquet")]
     fn optimize_away_unnecessary_repartition2() -> Result<()> {
         let physical_plan = filter_exec(repartition_exec(coalesce_partitions_exec(
             filter_exec(repartition_exec(parquet_exec())),