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.

[github-actions]

This issue has been automatically marked as stale because it has been open for 180 days with no activity. It will be closed in next 14 days if no further activity occurs. To permanently prevent this issue from being considered stale, add the label 'not-stale', but commenting on the issue is preferred when possible.