Fallback for upsert when arrow cannot compare source rows with target rows

[koenvo]

Rationale for this change

Upsert operations in PyIceberg rely on Arrow joins between source and target rows. However, Arrow Acero cannot compare certain complex types — like struct, list, and map — unless they’re part of the join key. When such types exist in non-join columns, the upsert fails with an error like:

ArrowInvalid: Data type struct<...> is not supported in join non-key field venue_geo

This PR introduces a fallback mechanism: if Arrow fails to join due to unsupported types, we fall back to comparing only the key columns. Non-key complex fields are ignored in the join condition, but still retained in the final upserted data.

---

Before

# Fails if venue_geo is a non-key struct field
txn.upsert(df, join_cols=["match_id"])

❌ ArrowInvalid: Data type struct<...> is not supported in join non-key field venue_geo

---

After

# Falls back to key-based join and proceeds
txn.upsert(df, join_cols=["match_id"])

✅ Successfully inserts or updates the record, skipping complex field comparison during join

---

✅ Are these changes tested?

Yes:

• A test was added to reproduce the failure scenario with complex non-key fields.
• The new behavior is verified by asserting that the upsert completes successfully using the fallback logic.

---

ℹ️ Note
This change does not affect users who do not include complex types in their schemas. For those who do, it improves resilience while preserving data correctness.

---

Are there any user-facing changes?

Yes — upserts involving complex non-key columns (like struct, list, or map) no longer fail. They now succeed by skipping unsupported comparisons during the join phase.

[Fokko]

Thanks for working on this @koenvo It looks like a lot of folks are waiting for this.

Could you run a poormans benchmark, similar to what I did here: #1685 (comment) Just to see how the two methods compare in terms of performance?

[Fokko]

We try to avoid assert outside of the tests. Could you raise a ValueError instead?

[koenvo]

Poor man's benchmark

This compares the performance of the original vs fallback upsert logic.
The "With skips" case simulates situations where non-matching rows can be skipped during comparison.

Condition	Original (s)	Fallback (s)	Diff (ms)	Diff (%)
Without skips	0.727	0.724	-2.73	-0.38%
With skips	0.681	0.732	+51.24	+7.53%

No significant performance regression observed. Fallback behaves as expected.

[Fokko]

First of all, sorry for the late reply, I was busy with the Iceberg summit :)

@koenvo The reason I was asking for a benchmark is to see if we can replace the existing logic with your logic that also works with {map,list,struct} types. I'm a simple man and I like simple things.

[Fokko]

I think we can optimize this allocation by avoiding creating a Python array, but that can be done in a separate PR.

[koenvo]

Fixed :-)