Patched DF 49.0.1 (take 1)

Cargo.toml

@@ -99,7 +99,6 @@ arrow-flight = { version = "55.2.0", features = [
 ] }
 arrow-ipc = { version = "55.2.0", default-features = false, features = [
     "lz4",
-    "zstd",
 ] }
 arrow-ord = { version = "55.2.0", default-features = false }
 arrow-schema = { version = "55.2.0", default-features = false }

datafusion-examples/examples/advanced_udwf.rs

@@ -199,6 +199,7 @@ impl WindowUDFImpl for SimplifySmoothItUdf {
                     order_by: window_function.params.order_by,
                     window_frame: window_function.params.window_frame,
                     null_treatment: window_function.params.null_treatment,
+                    distinct: window_function.params.distinct,
                 },
             }))
         };

datafusion/common/src/config.rs

@@ -389,7 +389,7 @@ config_namespace! {
         ///
         /// 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
+        pub skip_physical_aggregate_schema_check: bool, default = true
 
         /// Sets the compression codec used when spilling data to disk.
         ///

datafusion/core/Cargo.toml

@@ -47,6 +47,7 @@ compression = [
     "bzip2",
     "flate2",
     "zstd",
+    "arrow-ipc/zstd",
     "datafusion-datasource/compression",
 ]
 crypto_expressions = ["datafusion-functions/crypto_expressions"]

datafusion/core/src/physical_planner.rs

@@ -712,6 +712,9 @@ impl DefaultPhysicalPlanner {
                             differences.push(format!("field nullability at index {} [{}]: (physical) {} vs (logical) {}", i, physical_field.name(), physical_field.is_nullable(), logical_field.is_nullable()));
                         }
                     }
+
+                    log::warn!("Physical input schema should be the same as the one converted from logical input schema, but did not match for logical plan:\n{}", input.display_indent());
+
                     return internal_err!("Physical input schema should be the same as the one converted from logical input schema. Differences: {}", differences
                         .iter()
                         .map(|s| format!("\n\t- {s}"))
@@ -1646,6 +1649,7 @@ pub fn create_window_expr_with_name(
                         order_by,
                         window_frame,
                         null_treatment,
+                        distinct,
                     },
             } = window_fun.as_ref();
             let physical_args =
@@ -1674,6 +1678,7 @@ pub fn create_window_expr_with_name(
                 window_frame,
                 physical_schema,
                 ignore_nulls,
+                *distinct,
             )
         }
         other => plan_err!("Invalid window expression '{other:?}'"),

datafusion/core/tests/fuzz_cases/window_fuzz.rs

@@ -288,6 +288,7 @@ async fn bounded_window_causal_non_causal() -> Result<()> {
                     Arc::new(window_frame),
                     &extended_schema,
                     false,
+                    false,
                 )?;
                 let running_window_exec = Arc::new(BoundedWindowAggExec::try_new(
                     vec![window_expr],
@@ -660,6 +661,7 @@ async fn run_window_test(
             Arc::new(window_frame.clone()),
             &extended_schema,
             false,
+            false,
         )?],
         exec1,
         false,
@@ -678,6 +680,7 @@ async fn run_window_test(
             Arc::new(window_frame.clone()),
             &extended_schema,
             false,
+            false,
         )?],
         exec2,
         search_mode.clone(),