[NoahKusaba]

Which issue does this PR close?

• Closes Partitioned INSERT computes partition values from the wrong columns when the input plan ends in a projection #35.

Rationale for this change

PartitionExpr read its source columns by position in the input batch, but reported no children. When ProjectionPushdown merged the partition projection into the projection below it, it had no column references to rewrite, so the positions kept pointing at the old layout:

INSERT INTO t SELECT b AS id, a AS val FROM source  -- t is partitioned by identity(id)

partitioned the rows by a instead of b. Files landed in the wrong partitions, and partition pruning then skipped matching rows.

What changes are included in this PR?

• Sources are children. PartitionExpr has one child per partition field: the top-level column holding that field's source. The optimizer rewrites them like any other column reference, and EXPLAIN shows what each field reads:

  iceberg_partition_values(id: identity(b@1))
  

• Transforms are evaluated in place. The expression applies the transforms itself instead of going through iceberg's positional PartitionValueCalculator. A nested source is read by descending from its top-level column, nulling values under a null parent struct as iceberg's RecordBatchProjector did.

• The public API can't build a broken expression. PartitionExpr::try_new(spec, schema, sources) now takes the sources, so a codec passes the decoded children instead of assuming a table-shaped input. Only project_with_partition, which checks that its input matches the table schema, derives them from the table (crate-private try_new_over_table). with_new_children rejects the wrong number of children.

• Spec and schema errors from try_new are iceberg DataInvalid errors. A missing source field was previously Unexpected.

Are these changes tested?

Yes.

• test_insert_into_partitioned_through_reordering_projection (integration) runs the INSERT above. It snapshots the optimized plan, checks the partitions recorded in the manifests and reads back with a filter on the partition column. On main the manifests record partitions {1, 2} instead of {100, 200}.
• Unit tests cover remapping the children, rebuilding from the retained spec, schema and children, equality depending on the children, the child-count check, each try_new error by kind and message, and a nested source under a null parent struct (unchanged behavior, previously untested).

cargo fmt --all -- --check, cargo clippy --workspace --locked --all-targets -- -D warnings and cargo test --workspace --locked all pass locally.

Are there any user-facing changes?

• Breaking: PartitionExpr::try_new takes a third argument, sources: Vec<Arc<dyn PhysicalExpr>>. A codec that rebuilds the expression should pass the decoded children.
• EXPLAIN output for partitioned inserts uses the format above.
• The try_new error kinds above.

#19 relaxes the input schema check in project_with_partition. A plan-time check of the source types is left for a follow-up, so that it can use the same rule.

AI Disclosure

• Used Claude Code to find the cause, write the fix and tests and draft this description. I reviewed the change and reproduced the failure on main.