MERGE works incorrectly with partitioned table if the data column order is not same as table column order

[ion-elgreco]

Environment

Delta-rs version: 0.12.0

Binding: Python

---

Bug

What happened: When you do a merge on a table that has partition columns in a different order than the source data it will cause the wrong values to be updated in the partition columns.

What you expected to happen:
Update the columns correctly irrespective of the order.

How to reproduce it:
Write table first.

import polars as pl
from polars.io.delta import _convert_pa_schema_to_delta
from deltalake import DeltaTable, write_deltalake

df = pl.DataFrame({
    "foo": ['foo_value'],
    "bar": ['bar_value'],
    'value': [10]
})

arrow_data = df.to_arrow()
arrow_data = arrow_data.cast(_convert_pa_schema_to_delta(arrow_data.schema))
write_deltalake("test_table", arrow_data,  partition_by=['bar','foo'])

Merge the result, the partition column order is bar foo. Data order is foo & `bar

df_merge = pl.DataFrame({
    "foo": ['foo_value'],
    "bar": ['bar_value'],
    'value': [20]
})

arrow_data_merge = df_merge.to_arrow()
arrow_data_merge = arrow_data_merge.cast(_convert_pa_schema_to_delta(arrow_data_merge.schema))

dt = DeltaTable('test_table')

dt.merge(arrow_data_merge, 
         predicate='s.value = t.value', 
         source_alias='s', target_alias='t').when_not_matched_insert_all().execute()

The result is incorrect as you can see:

dt = DeltaTable('test_table')

print(pl.from_arrow(dt.to_pyarrow_table()))

┌───────────┬───────────┬───────┐
│ foo       ┆ bar       ┆ value │
│ ---       ┆ ---       ┆ ---   │
│ str       ┆ str       ┆ i64   │
╞═══════════╪═══════════╪═══════╡
│ bar_value ┆ foo_value ┆ 10    │
│ foo_value ┆ bar_value ┆ 20    │
└───────────┴───────────┴───────┘

[MrPowers]

@Blajda - would you like me to assign this to you?

[Blajda]

@MrPowers Sure you can assign me. Would be nice to have permissions to self assign since most I'll handle issue related to merge.