Upsert with None values fails on "Invalid literal value: None"

[mdwint]

Apache Iceberg version

main (development)

Please describe the bug 🐞

Upserting a table fails when the input dataframe contains None in a join column.

I've reproduced this error by editing test_upsert_with_nulls from #1861, adding this at the end:

# upsert table with null value
data_with_null = pa.Table.from_pylist(
    [
        {"foo": None, "bar": 1, "baz": False},
    ],
    schema=schema,
)
upd = table.upsert(data_with_null, join_cols=["foo"])

The foo column contains None, causing TypeError: Invalid literal value: None.

tests/table/test_upsert.py:720: in test_upsert_with_nulls
    upd = table.upsert(data_with_null, join_cols=["foo"])
pyiceberg/table/__init__.py:1343: in upsert
    return tx.upsert(
pyiceberg/table/__init__.py:798: in upsert
    matched_predicate = upsert_util.create_match_filter(df, join_cols)
pyiceberg/table/upsert_util.py:37: in create_match_filter
    return In(join_cols[0], unique_keys[0].to_pylist())
pyiceberg/expressions/__init__.py:682: in __new__
    literals_set: Set[Literal[L]] = _to_literal_set(literals)
pyiceberg/expressions/__init__.py:52: in _to_literal_set
    return {_to_literal(v) for v in values}
pyiceberg/expressions/__init__.py:52: in <setcomp>
    return {_to_literal(v) for v in values}
pyiceberg/expressions/__init__.py:59: in _to_literal
    return literal(value)
pyiceberg/expressions/literals.py:159: in literal
    raise TypeError(f"Invalid literal value: {repr(value)}")
E   TypeError: Invalid literal value: None

Willingness to contribute

• I can contribute a fix for this bug independently
• I would be willing to contribute a fix for this bug with guidance from the Iceberg community
• I cannot contribute a fix for this bug at this time

[kevinjqliu]

thanks for the report!

i can reproduce it locally. the issue seems to be related to the foo column as the join_cols.
We hit this branch

iceberg-python/pyiceberg/table/upsert_util.py

Lines 33 to 37 in 52d810e

	def create_match_filter(df: pyarrow_table, join_cols: list[str]) -> BooleanExpression:
	unique_keys = df.select(join_cols).group_by(join_cols).aggregate([])
	if len(join_cols) == 1:
	return In(join_cols[0], unique_keys[0].to_pylist())

Seems like the In expression cannot handle None/Null values

iceberg-python/pyiceberg/expressions/__init__.py

Lines 678 to 682 in 52d810e

	class In(SetPredicate[L]):
	def __new__( # type: ignore # pylint: disable=W0221
	cls, term: Union[str, UnboundTerm[Any]], literals: Union[Iterable[L], Iterable[Literal[L]]]
	) -> BooleanExpression:
	literals_set: Set[Literal[L]] = _to_literal_set(literals)

[mdwint]

I will try implementing a fix tomorrow, but I may come back with some questions since I don't know the codebase yet.

I'm guessing IsNull would be the correct filter predicate. Does that sound right?

[mdwint]

I've made some progress fixing the filter expressions in main...mdwint:iceberg-python:fix/upsert-with-nulls

The tests now confirm that the right expressions are produced. For example, in the last test case I can print this:

Or(
  left=And(
    left=IsNull(term=Reference(name='foo')),
    right=EqualTo(term=Reference(name='bar'), literal=LongLiteral(1))
  ),
  right=And(
    left=EqualTo(term=Reference(name='foo'), literal=literal('lemon')),
    right=IsNull(term=Reference(name='bar'))
  )
)

However, that last test case fails to update the row {"foo": None, "bar": 1, ...} with join_cols=["foo", "bar"]:

tests/table/test_upsert.py:741: in test_upsert_with_nulls
    assert upd.rows_updated == 1
E   assert 0 == 1
E    +  where 0 = UpsertResult(rows_updated=0, rows_inserted=1).rows_updated

So I think the filter expressions are now correct, but the upsert logic still doesn't handle updates with null values correctly. Wild guess: maybe it has to do with the way null values are compared (null doesn't equal itself?). Any ideas?

[mdwint]

So I think the filter expressions are now correct, but the upsert logic still doesn't handle updates with null values correctly. Wild guess: maybe it has to do with the way null values are compared (null doesn't equal itself?).

I've isolated the problem to step 3 in get_rows_to_update:

iceberg-python/pyiceberg/table/upsert_util.py

Lines 123 to 124 in 075a966

	# Step 3: Perform an inner join to find which rows from source exist in target
	matching_indices = source_index.join(target_index, keys=list(join_cols_set), join_type="inner")

This join ignores None values. It returns no matches, but I want it to match bar: [[1]] foo: [[null]]:

(Pdb) p source_index
pyarrow.Table
bar: int32
foo: large_string
__source_index: int64
----
bar: [[1,null]]
foo: [[null,"lemon"]]
__source_index: [[0,1]]

(Pdb) p target_index
pyarrow.Table
bar: int32
foo: large_string
__target_index: int64
----
bar: [[1]]
foo: [[null]]
__target_index: [[0]]

(Pdb) p matching_indices
pyarrow.Table
bar: int32
foo: large_string
__source_index: int64
__target_index: int64
----
bar: []
foo: []
__source_index: []
__target_index: []

[mdwint]

Computing the inner join in Python solved the issue. I've now opened #2429 for review.