cherry-pick inlist fix (#17254)

Author: haohuaijin

datafusion/proto/src/physical_plan/mod.rs

@@ -611,29 +611,6 @@ impl protobuf::PhysicalPlanNode {
     ) -> Result<Arc<dyn ExecutionPlan>> {
         let input: Arc<dyn ExecutionPlan> =
             into_physical_plan(&filter.input, registry, runtime, extension_codec)?;
-        let projection = if !filter.projection.is_empty() {
-            Some(
-                filter
-                    .projection
-                    .iter()
-                    .map(|i| *i as usize)
-                    .collect::<Vec<_>>(),
-            )
-        } else {
-            None
-        };
-
-        // Use the projected schema if projection is present, otherwise use the full schema
-        let predicate_schema = if let Some(ref proj_indices) = projection {
-            // Create projected schema for parsing the predicate
-            let projected_fields: Vec<_> = proj_indices
-                .iter()
-                .map(|&i| input.schema().field(i).clone())
-                .collect();
-            Arc::new(Schema::new(projected_fields))
-        } else {
-            input.schema()
-        };
 
         let predicate = filter
             .expr
@@ -642,7 +619,7 @@ impl protobuf::PhysicalPlanNode {
                 parse_physical_expr(
                     expr,
                     registry,
-                    predicate_schema.as_ref(),
+                    input.schema().as_ref(),
                     extension_codec,
                 )
             })
@@ -653,6 +630,17 @@ impl protobuf::PhysicalPlanNode {
                 )
             })?;
         let filter_selectivity = filter.default_filter_selectivity.try_into();
+        let projection = if !filter.projection.is_empty() {
+            Some(
+                filter
+                    .projection
+                    .iter()
+                    .map(|i| *i as usize)
+                    .collect::<Vec<_>>(),
+            )
+        } else {
+            None
+        };
         let filter =
             FilterExec::try_new(predicate, input)?.with_projection(projection)?;
         match filter_selectivity {

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -1839,3 +1839,36 @@ async fn test_round_trip_tpch_queries() -> Result<()> {
 
     Ok(())
 }
+
+#[tokio::test]
+async fn test_tpch_part_in_list_query_with_real_parquet_data() -> Result<()> {
+    use datafusion_common::test_util::datafusion_test_data;
+
+    let ctx = SessionContext::new();
+
+    // Register the TPC-H part table using the local test data
+    let test_data = datafusion_test_data();
+    let table_sql = format!(
+    "CREATE EXTERNAL TABLE part STORED AS PARQUET LOCATION '{test_data}/tpch_part_small.parquet'"
+);
+    ctx.sql(&table_sql).await.map_err(|e| {
+        DataFusionError::External(format!("Failed to create part table: {e}").into())
+    })?;
+
+    // Test the exact problematic query
+    let sql =
+        "SELECT p_size FROM part WHERE p_size IN (14, 6, 5, 31) and p_partkey > 1000";
+
+    let logical_plan = ctx.sql(sql).await?.into_unoptimized_plan();
+    let optimized_plan = ctx.state().optimize(&logical_plan)?;
+    let physical_plan = ctx.state().create_physical_plan(&optimized_plan).await?;
+
+    // Serialize the physical plan - bug may happen here already but not necessarily manifests
+    let codec = DefaultPhysicalExtensionCodec {};
+    let proto = PhysicalPlanNode::try_from_physical_plan(physical_plan.clone(), &codec)?;
+
+    // This will fail with the bug, but should succeed when fixed
+    let _deserialized_plan =
+        proto.try_into_physical_plan(&ctx, ctx.runtime_env().as_ref(), &codec)?;
+    Ok(())
+}