refactor: narrow Err(_) catch-all in with_generated_columns to schema resolution errors only

Address review feedback:
- Replace broad Err(_) catch-all with a guard that only matches column
  resolution errors (No field named / Schema error), letting parse and
  type errors propagate normally.
- Bind the error and include it in the debug log message.
- Add e2e test through WriteBuilder & SchemaMode::Merge that writes
  with and without the referenced column, then asserts actual values.

Author: veeceey

crates/core/src/operations/write/generated_columns.rs

@@ -23,6 +23,18 @@ pub fn gc_is_enabled(snapshot: &EagerSnapshot) -> bool {
         .is_feature_enabled(&TableFeature::GeneratedColumns)
 }
 
+/// Returns `true` when the error indicates that a column referenced in the
+/// generation expression could not be found in the current plan schema.
+/// This happens during `SchemaMode::Merge` when the input batch omits
+/// nullable columns that schema evolution will add later.  All other errors
+/// (e.g. SQL syntax errors, type mismatches) should still be surfaced.
+fn is_column_resolution_error(err: &crate::DeltaTableError) -> bool {
+    let msg = err.to_string();
+    // DataFusion emits "No field named ..." for unresolved column references
+    // and "Schema error: ..." for broader schema resolution failures.
+    msg.contains("No field named") || msg.contains("Schema error")
+}
+
 pub fn with_generated_columns(
     session: &dyn Session,
     plan: LogicalPlan,
@@ -68,11 +80,11 @@ pub fn with_generated_columns(
                 }
                 e
             }
-            Err(_) => {
+            Err(ref err) if is_column_resolution_error(err) => {
                 debug!(
-                    "Could not resolve generation expression for column {}, \
+                    "Could not resolve generation expression for column {} ({}), \
                      inserting NULL placeholder (will be resolved after schema evolution).",
-                    name
+                    name, err
                 );
                 // Use the target data type from the table schema if available,
                 // otherwise fall back to a bare NULL.
@@ -82,6 +94,7 @@ pub fn with_generated_columns(
                     lit(ScalarValue::Null).alias(name)
                 }
             }
+            Err(err) => return Err(err.into()),
         };
         projection.push(expr);
     }

crates/core/tests/integration_datafusion.rs

@@ -1669,6 +1669,105 @@ async fn test_schema_merge_append_missing_non_nullable_column_with_generated_col
     );
 }
 
+/// E2E test for #4169: a generated column whose expression references a nullable
+/// column that is *not* present in the input batch should still succeed when
+/// appending with `SchemaMode::Merge`.  The generated column value should be
+/// NULL for the rows where the referenced column is absent, and correctly
+/// computed when the referenced column is present.
+#[tokio::test]
+async fn test_schema_merge_generated_column_referencing_missing_column_e2e() {
+    let ctx = SessionContext::new();
+    let tmp_dir = tempfile::tempdir().unwrap();
+    let table_uri = tmp_dir.path().to_str().unwrap();
+
+    // Table schema: id (int, required), tag (string, nullable),
+    //               tag_upper (string, nullable, generated = upper(tag))
+    let table_schema = StructType::try_new(vec![
+        StructField::new(
+            "id".to_string(),
+            DataType::Primitive(PrimitiveType::Integer),
+            false,
+        ),
+        StructField::new(
+            "tag".to_string(),
+            DataType::Primitive(PrimitiveType::String),
+            true,
+        ),
+        StructField::new(
+            "tag_upper".to_string(),
+            DataType::Primitive(PrimitiveType::String),
+            true,
+        )
+        .with_metadata(vec![(
+            ColumnMetadataKey::GenerationExpression.as_ref().to_string(),
+            MetadataValue::String("upper(tag)".to_string()),
+        )]),
+    ])
+    .unwrap();
+
+    let table = create_table_with_schema(table_uri, &table_schema).await;
+
+    // First write: input has both id and tag — generated column should be computed.
+    let batch_with_tag = RecordBatch::try_from_iter_with_nullable(vec![
+        (
+            "id",
+            Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef,
+            false,
+        ),
+        (
+            "tag",
+            Arc::new(StringArray::from(vec![Some("hello"), Some("world")])) as ArrayRef,
+            true,
+        ),
+    ])
+    .unwrap();
+
+    let table = table
+        .write(vec![batch_with_tag])
+        .with_schema_mode(SchemaMode::Merge)
+        .await
+        .unwrap();
+
+    // Second write: input omits `tag` entirely — generated column should become NULL.
+    let batch_without_tag = RecordBatch::try_from_iter_with_nullable(vec![(
+        "id",
+        Arc::new(Int32Array::from(vec![3, 4])) as ArrayRef,
+        false,
+    )])
+    .unwrap();
+
+    let table = table
+        .write(vec![batch_without_tag])
+        .with_schema_mode(SchemaMode::Merge)
+        .await
+        .unwrap();
+
+    // Read back and verify values.
+    let batches = ctx
+        .read_table(table.table_provider().await.unwrap())
+        .unwrap()
+        .select_exprs(&["id", "tag", "tag_upper"])
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+
+    assert_batches_sorted_eq!(
+        #[rustfmt::skip]
+        &[
+            "+----+-------+-----------+",
+            "| id | tag   | tag_upper |",
+            "+----+-------+-----------+",
+            "| 1  | hello | HELLO     |",
+            "| 2  | world | WORLD     |",
+            "| 3  |       |           |",
+            "| 4  |       |           |",
+            "+----+-------+-----------+",
+        ],
+        &batches
+    );
+}
+
 mod date_partitions {
     use super::*;