Issue during Upsert

[heman026]

Question

I am getting the following error when using table.upsert() to update data.

Exception has occurred: RecursionError
maximum recursion depth exceeded
AttributeError: '_thread._local' object has no attribute 'f_unhandled_exc'

During handling of the above exception, another exception occurred:

File "C:\loader_pyiceberg.py", line 118, in upsert
table.upsert(pyarrow_table,join_cols=join_cols)
File "C:\loader_pyiceberg.py", line 222, in
upsert(iceberg_table,pyarrow,['COL1','COL2','COL3','COL4'])
RecursionError: maximum recursion depth exceeded

This works fine if the no: of rows in pyarrow table (source) is around 400. I am getting the error if the number of rows is around 500. Is there any limitations for this method, based on size of the data?

Could you help me resolve this issue.

Thanks

[kevinjqliu]

can you share more on what code you ran? specifically what is table, pyarrow_table, and join_cols

[heman026]

can you share more on what code you ran? specifically what is table, pyarrow_table, and join_cols

iceberg_table = 'test.table1'
table = catalog.load_table(iceberg_table)
data= table.scan(row_filter=E.EqualTo('id', 1)).to_arrow()

icerbeg table is fitlered and set to data pyarrow table). I am doing some computation on data and modify one of the column values. I update the columns values to the original iceberg table using upsert.

table.upsert(data,join_cols=['id','col1','col2,'col3'])

This is a sample code. The iceberg table which I am testing with has 144 million rows with 14 columns. And the filtered data has 500 rows with the same number of columns. I have 4 primary key columns which I have provided in join_cols argument of upsert method.

If the filtered data has around 400 rows, upsert method is working fine, if it has around 500 data, it is giving the above exception.

Let me know if you need more details

[heman026]

Full stack trace of the issue:

Exception has occurred: RecursionError

maximum recursion depth exceeded
AttributeError: '_thread._local' object has no attribute 'f_unhandled_exc'

During handling of the above exception, another exception occurred:

File "C:\Hemanath\pyiceberg\expressions\visitors.py", line 198, in _
left_result: T = visit(obj.left, visitor=visitor)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Hemanath\pyiceberg\expressions\visitors.py", line 198, in _
left_result: T = visit(obj.left, visitor=visitor)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Hemanath\pyiceberg\expressions\visitors.py", line 198, in _
left_result: T = visit(obj.left, visitor=visitor)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
[Previous line repeated 488 more times]
File "C:\Hemanath\pyiceberg\expressions\visitors.py", line 214, in bind
return visit(expression, BindVisitor(schema, case_sensitive))
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Hemanath\pyiceberg\io\pyarrow.py", line 1515, in init
self.bound_row_filter = bind(table_metadata.schema(), row_filter, case_sensitive=case_sensitive)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Hemanath\pyiceberg\table_init.py", line 1759, in to_arrow
self.table_metadata, self.io, self.projection(), self.row_filter, self.case_sensitive, self.limit
).to_table(self.plan_files())
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Hemanath\pyiceberg\table_init_.py", line 1187, in upsert
matched_iceberg_table = self.scan(row_filter=matched_predicate, case_sensitive=case_sensitive).to_arrow()
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Hemanath\loader_pyiceberg.py", line 138, in upsert1
table.upsert(arrow,join_cols=join_cols)
File "C:\Hemanath\loader_pyiceberg.py", line 293, in
upsert1(iceberg_table,arrow,['col1','col2','col3','col4'])
RecursionError: maximum recursion depth exceeded

[suarez-agu]

I also have the same issue trying to upsert 10k rows.

Starting from an empty Table btw and using two identifier_fields. Not using join_cols

[Fokko]

Thanks everyone, for raising this. The number of columns that you upsert on can cause problems since each column will duplicate the complexity of the predicate.

As a intermediate solution, you might want to try to bump the recursion depth:

import sys
sys.setrecursionlimit(5000)

[smaheshwar-pltr]

Have a test that I think repros this in (draft) #1783.

Also, (draft) #1783 naively removes recursion by implementing the call stack / frames explicitly, to fix this. Don't think we should actually do this though.

[mattmartin14]

I encountered this issue when writing the upsert method. If your join column is a single column, then the function scales fine and won't hit a recursion limit. I stressed tested a single column join on 1M rows and it handled it. Where the issue lies is on composite keys (2 or more columns to join on). I was not able to work on the visitor function that builds out the overwrite predicate but that is the function that needs work and needs to get away from using recursion to find all the recs to overwrite

[kevinjqliu]

Thanks everyone. i think this is a more generic issue with bind and the visitors which i opened #1785 to track.

I believe this issue is showing up in upsert in a few different ways.

1. the table itself has deeply nested schema, in which case the issue occurs for upsert and other functions like scan/delete
2. large number of join_cols which influences the expression variable passed to bind

(2) can show up in many ways, such as

• when we create the target table

  iceberg-python/pyiceberg/table/__init__.py

  Lines 1185 to 1187 in 738ef11

  	# get list of rows that exist so we don't have to load the entire target table
  	matched_predicate = upsert_util.create_match_filter(df, join_cols)
  	matched_iceberg_table = self.scan(row_filter=matched_predicate, case_sensitive=case_sensitive).to_arrow()

• when we build and use the overwrite filter

  iceberg-python/pyiceberg/table/__init__.py

  Lines 1201 to 1204 in 738ef11

  	# build the match predicate filter
  	overwrite_mask_predicate = upsert_util.create_match_filter(rows_to_update, join_cols)
  	tx.overwrite(rows_to_update, overwrite_filter=overwrite_mask_predicate)

• when we create the filter expression for insert

  iceberg-python/pyiceberg/table/__init__.py

  Lines 1207 to 1210 in 738ef11

  	expr_match = upsert_util.create_match_filter(matched_iceberg_table, join_cols)
  	expr_match_bound = bind(self.schema(), expr_match, case_sensitive=case_sensitive)
  	expr_match_arrow = expression_to_pyarrow(expr_match_bound)
  	rows_to_insert = df.filter(~expr_match_arrow)

[mattmartin14]

Hey @kevinjqliu ,
From my original testing, insert filters were not affected by this problem. It was only the overwrite filters that were an issue.

Has something changed?

[kevinjqliu]

@mattmartin14 its based on the scale of the input. For insert filters, it depends on the output of create_match_filter and the table schema.
Essentially anytime bind is called, we can run into the max recursion error.

[mattmartin14]

What if instead of using a filter for the insert rows, we instead do a pyarrow compute anti-left join to identify the rows needing to be inserted? That might help avoid the problem for the insert part.

I know @Fokko originally did not want to use joins but it might make sense here to solve that part of the problem