DRAFT: experiment of #9637

datafusion/common/src/dfschema.rs

@@ -488,6 +488,19 @@ impl DFSchema {
         })
     }
 
+    pub fn equivalent_names_and_types_v2(&self, fields: &Vec<DFField>) -> bool {
+        if self.fields().len() != fields.len() {
+            return false;
+        }
+        let self_fields = self.fields().iter();
+        let other_fields = fields.iter();
+        self_fields.zip(other_fields).all(|(f1, f2)| {
+            f1.qualifier() == f2.qualifier()
+                && f1.name() == f2.name()
+                && Self::datatype_is_semantically_equal(f1.data_type(), f2.data_type())
+        })
+    }
+
     /// Checks if two [`DataType`]s are logically equal. This is a notably weaker constraint
     /// than datatype_is_semantically_equal in that a Dictionary<K,V> type is logically
     /// equal to a plain V type, but not semantically equal. Dictionary<K1, V1> is also

datafusion/common/src/lib.rs

@@ -35,6 +35,7 @@ pub mod file_options;
 pub mod format;
 pub mod hash_utils;
 pub mod instant;
+pub mod optimize_node;
 pub mod parsers;
 pub mod rounding;
 pub mod scalar;

datafusion/common/src/optimize_node.rs

@@ -0,0 +1,64 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::DataFusionError;
+
+#[derive(Debug, PartialEq, Eq)]
+pub enum OptimizedState {
+    Yes,
+    No,
+    Fail,
+}
+
+#[derive(Debug)]
+pub struct Optimized<T, E = DataFusionError> {
+    pub optimzied_data: Option<T>,
+    // Used to store the original data if optimized successfully
+    pub original_data: T,
+    pub optimized_state: OptimizedState,
+    // Used to store the error if optimized failed, so we can early return but preserve the original data
+    pub error: Option<E>,
+}
+
+impl<T, E> Optimized<T, E> {
+    pub fn yes(optimzied_data: T, original_data: T) -> Self {
+        Self {
+            optimzied_data: Some(optimzied_data),
+            original_data,
+            optimized_state: OptimizedState::Yes,
+            error: None,
+        }
+    }
+
+    pub fn no(original_data: T) -> Self {
+        Self {
+            optimzied_data: None,
+            original_data,
+            optimized_state: OptimizedState::No,
+            error: None,
+        }
+    }
+
+    pub fn fail(original_data: T, e: E) -> Self {
+        Self {
+            optimzied_data: None,
+            original_data,
+            optimized_state: OptimizedState::Fail,
+            error: Some(e),
+        }
+    }
+}

datafusion/common/src/tree_node.rs

@@ -469,6 +469,10 @@ impl<T> Transformed<T> {
         }
     }
 
+    pub fn is_transformed(&self) -> bool {
+        self.transformed
+    }
+
     /// Wrapper for transformed data with [`TreeNodeRecursion::Continue`] statement.
     pub fn yes(data: T) -> Self {
         Self::new(data, true, TreeNodeRecursion::Continue)

datafusion/core/src/execution/context/mod.rs

@@ -1875,16 +1875,29 @@ impl SessionState {
             stringified_plans
                 .push(analyzed_plan.to_stringified(PlanType::FinalAnalyzedLogicalPlan));
 
-            // optimize the child plan, capturing the output of each optimizer
-            let optimized_plan = self.optimizer.optimize(
-                &analyzed_plan,
-                self,
-                |optimized_plan, optimizer| {
-                    let optimizer_name = optimizer.name().to_string();
-                    let plan_type = PlanType::OptimizedLogicalPlan { optimizer_name };
-                    stringified_plans.push(optimized_plan.to_stringified(plan_type));
-                },
-            );
+            let optimized_plan = if self.options().optimizer.skip_failed_rules {
+                self.optimizer.optimize(
+                    &analyzed_plan,
+                    self,
+                    |optimized_plan, optimizer| {
+                        let optimizer_name = optimizer.name().to_string();
+                        let plan_type = PlanType::OptimizedLogicalPlan { optimizer_name };
+                        stringified_plans.push(optimized_plan.to_stringified(plan_type));
+                    },
+                )
+            } else {
+                // optimize the child plan, capturing the output of each optimizer
+                self.optimizer.optimize_owned(
+                    analyzed_plan,
+                    self,
+                    |optimized_plan, optimizer| {
+                        let optimizer_name = optimizer.name().to_string();
+                        let plan_type = PlanType::OptimizedLogicalPlan { optimizer_name };
+                        stringified_plans.push(optimized_plan.to_stringified(plan_type));
+                    },
+                )
+            };
+
             let (plan, logical_optimization_succeeded) = match optimized_plan {
                 Ok(plan) => (Arc::new(plan), true),
                 Err(DataFusionError::Context(optimizer_name, err)) => {
@@ -1904,10 +1917,18 @@ impl SessionState {
                 logical_optimization_succeeded,
             }))
         } else {
-            let analyzed_plan =
-                self.analyzer
-                    .execute_and_check(plan, self.options(), |_, _| {})?;
-            self.optimizer.optimize(&analyzed_plan, self, |_, _| {})
+            if self.options().optimizer.skip_failed_rules {
+                let analyzed_plan =
+                    self.analyzer
+                        .execute_and_check(plan, self.options(), |_, _| {})?;
+                self.optimizer.optimize(&analyzed_plan, self, |_, _| {})
+            } else {
+                let analyzed_plan =
+                    self.analyzer
+                        .execute_and_check(plan, self.options(), |_, _| {})?;
+                self.optimizer
+                    .optimize_owned(analyzed_plan, self, |_, _| {})
+            }
         }
     }