Add option to skip physical aggregate check

datafusion/common/src/config.rs

@@ -268,6 +268,17 @@ config_namespace! {
         /// Defaults to the number of CPU cores on the system
         pub planning_concurrency: usize, default = num_cpus::get()
 
+        /// When set to true, skips verifying that the schema produced by
+        /// planning `LogicalPlan::Aggregate` exactly matches the schema of the
+        /// input plan.
+        ///
+        /// When set to false, if the schema does not match exactly
+        /// (including nullability and metadata), a planning error will be raised.
+        ///
+        /// This is used to workaround bugs in the planner that are now caught by
+        /// the new schema verification step.
+        pub skip_physical_aggregate_schema_check: bool, default = false
+
         /// Specifies the reserved memory for each spillable sort operation to
         /// facilitate an in-memory merge.
         ///

datafusion/core/src/physical_planner.rs

@@ -649,14 +649,16 @@ impl DefaultPhysicalPlanner {
                 aggr_expr,
                 ..
             }) => {
+                let options = session_state.config().options();
                 // Initially need to perform the aggregate and then merge the partitions
                 let input_exec = children.one()?;
                 let physical_input_schema = input_exec.schema();
                 let logical_input_schema = input.as_ref().schema();
-                let physical_input_schema_from_logical: Arc<Schema> =
-                    logical_input_schema.as_ref().clone().into();
+                let physical_input_schema_from_logical = logical_input_schema.inner();
 
-                if physical_input_schema != physical_input_schema_from_logical {
+                if &physical_input_schema != physical_input_schema_from_logical
+                    && !options.execution.skip_physical_aggregate_schema_check
+                {
                     return internal_err!("Physical input schema should be the same as the one converted from logical input schema.");
                 }