Upsert fails after update_schema().union_by_name() due to schema mismatch #3105

Activity

1. laserninja commented on Mar 25, 2026

   

   laserninja

   on Mar 25, 2026

   More actions

   Let me know if you need help fixing it.
2. AndrejIring commented on Apr 30, 2026

   

   AndrejIring

   on Apr 30, 2026

   More actions

   Same issue with add_column, the issue stems from the get_rows_to_update in upsert_utils where the data is cast to the underlying table schema. Which fails since the schemas are different (pyarrow cast is not adding/deleting/reordering the columns).

   This makes upsert unusable for the tables with updated schemas.

   from pyiceberg.catalog import load_catalog
   import polars as pl
   from pyiceberg.types import StringType
   catalog = load_catalog("default", **{"type": "in-memory"})
   
   df = pl.DataFrame(
       [
           {"id": 1, "name": "Alice", "age": 30, "city": "São Paulo"},
           {"id": 2, "name": "Bob", "age": 25, "city": "Rio de Janeiro"},
           {"id": 3, "name": "Carol", "age": 35, "city": "Belo Horizonte"},
           {"id": 4, "name": "David", "age": 28, "city": "Curitiba"},
       ]
   )
   
   arrow = df.to_arrow()
   
   catalog.create_namespace_if_not_exists("default")
   catalog.create_table_if_not_exists("default.my_table", arrow.schema)
   table = catalog.load_table("default.my_table")
   
   table.append(arrow)
   
   # Add a new column
   arrow = df.with_columns(ping=pl.lit("pong")).to_arrow()
   
   # Update schema to include the new column
   with table.update_schema() as update_schema:
       update_schema.add_column("ping", StringType())
       table = table.refresh()
   print(table.schema())
   try:   
       # This fails with ValueError
       table.upsert(arrow, ["id"])
   finally:
       catalog.drop_table("default.my_table")

   Hopefully I will create today a PR for resolving this basically these lines

   iceberg-python/pyiceberg/table/upsert_util.py

   Lines 91 to 95 in b67b724

   	source_index = (
   	source_table.cast(target_table.schema)
   	.select(join_cols_set)
   	.append_column(SOURCE_INDEX_COLUMN_NAME, pa.array(range(len(source_table))))
   	)

   should be changed to something like this:

       join_cols_schema = pa.schema([target_table.schema.field(name) for name in join_cols_set])
       source_index = (
           source_table.select(join_cols_set).cast(join_cols_schema)
           .append_column(SOURCE_INDEX_COLUMN_NAME, pa.array(range(len(source_table))))
       )

   Currently working workaround is to use table.overwrite after the schema evolution, to update the underlying tables.

   Example of the workaround:

   from pyiceberg.catalog import load_catalog
   import polars as pl
   from pyiceberg.types import StringType
   catalog = load_catalog("default", **{"type": "in-memory"})
   
   df = pl.DataFrame(
       [
           {"id": 1, "name": "Alice", "age": 30, "city": "São Paulo"},
           {"id": 2, "name": "Bob", "age": 25, "city": "Rio de Janeiro"},
           {"id": 3, "name": "Carol", "age": 35, "city": "Belo Horizonte"},
           {"id": 4, "name": "David", "age": 28, "city": "Curitiba"},
       ]
   )
   
   arrow = df.to_arrow()
   
   catalog.create_namespace_if_not_exists("default")
   catalog.create_table_if_not_exists("default.my_table", arrow.schema)
   table = catalog.load_table("default.my_table")
   
   table.append(arrow)
   
   # Add a new column
   # Update schema to include the new column
   with table.update_schema() as update_schema:
       update_schema.add_column("ping", StringType())
       table = table.refresh()
   arrow = df.with_columns(ping=pl.lit(None, dtype=pl.String)).to_arrow()
   table.overwrite(arrow)
   print(table.scan().to_polars())
   try:   
       arrow = df.with_columns(ping=pl.lit("pong")).to_arrow()
       # No longer fails
       table.upsert(arrow, ["id"])
   finally:
       catalog.drop_table("default.my_table")
   print(table.scan().to_polars())
3. clifflau1120 commented on May 22, 2026

   

   clifflau1120

   on May 22, 2026

   Last edited by clifflau1120

   More actions

   This is a real issue. This implies full rewrite whenever we add a column during schema evolution. This goes against the whole idea of Apache Iceberg.
4. blarghmatey

   added a commit that references this issue on Sep 11, 2026

   fix(dlt): commit a snapshot under the new schema before Iceberg upser…

   ...

   217409b