[🐞] Collection of a few bugs

[kevinjqliu]

Apache Iceberg version

None

Please describe the bug 🐞

While working on a Python REST server implementation (https://github.com/kevinjqliu/iceberg-rest-catalog), I noticed a few Pyiceberg bugs while integrating with PySpark. The REST server implementation is using SqlCatalog as the underlying catalog.

Here's the collection of bugs that I would like to upstream (main...kevinjqliu:iceberg-python:kevinjqliu/iceberg-rest-catalog)
To summarize:

• rest.py's drop_namespace function is missing handling the 409 status code. (Handle error when trying to drop a non-empty namespace on rest.py #868)
• sql.py's drop_namespace function does not return NoSuchNamespaceError when namespace is missing. (Raise error on nonexistent namespace on sql.py #865)
• sql.py's update_namespace_properties function only updates the first property and drops the rest. (fix: update all namespace properties in update_namespace_properties #873)
• TableMetadata is initialized with the default Pydantic object for schema, partition_spec, and sort_order, which does not play well with table updates.

I'll be happy to work on this but I think these are "good first issues". We can break these up into separate PRs with the relevant tests.

[uatach]

@kevinjqliu, I've just created a PR for the second item on the list.

[ranli]

@kevinjqliu created a PR to fix 3rd item on the list

[HonahX]

TableMetadata is initialized with the default Pydantic object for schema, partition_spec, and sort_order, which does not play well with table updates.

Hi @kevinjqliu. Could you please elaborate more on this? I am curious of the corresponding use-cases that are affected by default values of these fields.

I remembered I also encountered similar difficulties when working on #498 and I ended up with a workaround to let update_table_metadata handle some "initial" table updates specially.

[kevinjqliu]

@HonahX Ok so I've found a way to reproduce this. Requires a bit of a setup.
I found this while working on https://github.com/kevinjqliu/iceberg-rest-catalog and testing its integration with Spark.

Setup

Have pyiceberg's integration test docker running

make test-integration

Pull the kevinjqliu/iceberg-rest-catalog repo and run the fastapi server locally

poetry install
poetry run fastapi dev src/iceberg_rest/main.py

It requires an on-disk sqlite db located in /tmp/warehouse so you also need to create the dir

mkdir /tmp/warehouse

Reset the iceberg-python vendor folder to refer to main branch (tracking the official project's main branch), instead of the branch with the fixes.

cd vendor/iceberg-python
git checkout main
git pull 

Run Spark test

Have fastapi server running in one terminal.

poetry run fastapi dev src/iceberg_rest/main.py

In another terminal, run the spark test

poetry run python tests/pyspark_test.py

The fastapi server will error, sending HTTP 500 request.

Here's the stack trace

INFO:     127.0.0.1:55910 - "POST /v1/namespaces/default/tables/test_null_nan HTTP/1.1" 500 Internal Server Error
ERROR:    Exception in ASGI application
Traceback (most recent call last):
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/uvicorn/protocols/http/httptools_impl.py", line 399, in run_asgi
    result = await app(  # type: ignore[func-returns-value]
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/uvicorn/middleware/proxy_headers.py", line 70, in __call__
    return await self.app(scope, receive, send)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/fastapi/applications.py", line 1054, in __call__
    await super().__call__(scope, receive, send)
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/starlette/applications.py", line 123, in __call__
    await self.middleware_stack(scope, receive, send)
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/starlette/middleware/errors.py", line 186, in __call__
    raise exc
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/starlette/middleware/errors.py", line 164, in __call__
    await self.app(scope, receive, _send)
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 65, in __call__
    await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/starlette/_exception_handler.py", line 64, in wrapped_app
    raise exc
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
    await app(scope, receive, sender)
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/starlette/routing.py", line 756, in __call__
    await self.middleware_stack(scope, receive, send)
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/starlette/routing.py", line 776, in app
    await route.handle(scope, receive, send)
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/starlette/routing.py", line 297, in handle
    await self.app(scope, receive, send)
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/starlette/routing.py", line 77, in app
    await wrap_app_handling_exceptions(app, request)(scope, receive, send)
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/starlette/_exception_handler.py", line 64, in wrapped_app
    raise exc
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
    await app(scope, receive, sender)
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/starlette/routing.py", line 72, in app
    response = await func(request)
               ^^^^^^^^^^^^^^^^^^^
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/fastapi/routing.py", line 278, in app
    raw_response = await run_endpoint_function(
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/fastapi/routing.py", line 193, in run_endpoint_function
    return await run_in_threadpool(dependant.call, **values)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/starlette/concurrency.py", line 42, in run_in_threadpool
    return await anyio.to_thread.run_sync(func, *args)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/anyio/to_thread.py", line 56, in run_sync
    return await get_async_backend().run_sync_in_worker_thread(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 2177, in run_sync_in_worker_thread
    return await future
           ^^^^^^^^^^^^
  File "/Users/kevinliu/repos/iceberg-rest-catalog/.venv/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 859, in run
    result = context.run(func, *args)
             ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/kevinliu/repos/iceberg-rest-catalog/src/iceberg_rest/api/catalog_api.py", line 453, in update_table
    resp = catalog._commit_table(commit_table_request)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/kevinliu/repos/iceberg-rest-catalog/vendor/iceberg-python/pyiceberg/catalog/sql.py", line 423, in _commit_table
    updated_staged_table = self._update_and_stage_table(current_table, table_request)
                           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/kevinliu/repos/iceberg-rest-catalog/vendor/iceberg-python/pyiceberg/catalog/__init__.py", line 781, in _update_and_stage_table
    updated_metadata = update_table_metadata(
                       ^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/kevinliu/repos/iceberg-rest-catalog/vendor/iceberg-python/pyiceberg/table/__init__.py", line 1179, in update_table_metadata
    new_metadata = _apply_table_update(update, new_metadata, context)
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/kevinliu/.pyenv/versions/3.11.0/lib/python3.11/functools.py", line 909, in wrapper
    return dispatch(args[0].__class__)(*args, **kw)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/kevinliu/repos/iceberg-rest-catalog/vendor/iceberg-python/pyiceberg/table/__init__.py", line 1022, in _
    raise ValueError(f"Partition spec with id {spec.spec_id} already exists: {spec}")
ValueError: Partition spec with id 0 already exists: []

From here, you can add breakpoints to the codebase in vendor/iceberg-python and debug.

Here are the fixes that I added to make the spark test work
main...kevinjqliu:iceberg-python:kevinjqliu/iceberg-rest-catalog

[kevinjqliu]

I dont think the REST server API is at fault here since this request corresponds with the update_table function.
Which just serialize the CommitTableRequest object and pass it to _commit_table.

INFO:     127.0.0.1:55910 - "POST /v1/namespaces/default/tables/test_null_nan HTTP/1.1" 500 Internal Server Error

And since I assume Spark is sending the right request, it has to be an issue downstream of _commit_table

[sungwy]

TableMetadata is initialized with the default Pydantic object for schema, partition_spec, and sort_order, which does not play well with table updates.

Hi @kevinjqliu - I just added a test case similar to yours and verified that it works with the tabular rest catalog image in the test suite. Could we double check if this is an issue in the Rest catalog implementation?

Branch with test: https://github.com/syun64/iceberg-python/tree/rest-integration

The test is: pytest tests/integration/test_rest.py::test_write_round_trip

I would appreciate you double checking my work as well to make sure it correctly replicates your test!

[kevinjqliu]

Thanks for taking a look at this!

I reran the steps above; the issue is actually with the CREATE OR REPLACE TABLE statement
https://github.com/kevinjqliu/iceberg-rest-catalog/blob/main/tests/pyspark_test.py#L94-L108

I verified that all Pyspark statements work with the tabular REST catalog. https://github.com/kevinjqliu/iceberg-rest-catalog/blob/7c5548133ae266d4fac215b063911c35f08461d9/tests/pyspark_test.py#L75-L116

To reproduce, add this test to test_rest_manifest.py and running pytest tests/integration/test_rest_manifest.py::test_write_round_trip -v -m integration

from pyspark.sql import SparkSession

@pytest.mark.integration
def test_write_round_trip(catalog: Catalog, spark: SparkSession) -> None:
    from pyiceberg.exceptions import NoSuchTableError
    from pyiceberg.schema import Schema
    from pyiceberg.types import NestedField, UUIDType, FixedType
    schema = Schema(
        NestedField(field_id=1, name="uuid_col", field_type=UUIDType(), required=False),
        NestedField(field_id=2, name="fixed_col", field_type=FixedType(25), required=False),
    )

    identifier = "default.test_table_write_round_trip"
    try:
        catalog.drop_table(identifier)
    except NoSuchTableError:
        pass

    catalog.create_table(
        identifier=identifier, schema=schema
    )

    # write to table with spark sql
    spark.sql(
        f"""
        INSERT INTO integration.{identifier} VALUES
        ('102cb62f-e6f8-4eb0-9973-d9b012ff0967', CAST('1234567890123456789012345' AS BINARY)),
        ('ec33e4b2-a834-4cc3-8c4a-a1d3bfc2f226', CAST('1231231231231231231231231' AS BINARY)),
        ('639cccce-c9d2-494a-a78c-278ab234f024', CAST('12345678901234567ass12345' AS BINARY)),
        ('c1b0d8e0-0b0e-4b1e-9b0a-0e0b0d0c0a0b', CAST('asdasasdads12312312312111' AS BINARY)),
        ('923dae77-83d6-47cd-b4b0-d383e64ee57e', CAST('qweeqwwqq1231231231231111' AS BINARY));
        """
    )

    tbl = catalog.load_table(identifier)
    assert tbl.schema() == schema
    df = tbl.scan().to_arrow().to_pandas()
    assert len(df) == 5
    assert b"1234567890123456789012345" in df["fixed_col"].to_list()

    # create a table with spark sql
    spark.sql(
        f"""
        CREATE OR REPLACE TABLE integration.{identifier}
        USING iceberg
        AS SELECT
        1            AS idx,
        float('NaN') AS col_numeric
    UNION ALL SELECT
        2            AS idx,
        null         AS col_numeric
    UNION ALL SELECT
        3            AS idx,
        1            AS col_numeric
    """
    )

    # write to table with spark sql
    spark.sql(
        f"""
        INSERT INTO integration.{identifier} VALUES
        (4, 999);
        """
    )

Could we double check if this is an issue in the Rest catalog implementation?

The issue is with this specific Rest catalog implementation. However, the implementation uses the Pyiceberg library to process catalog requests.
CREATE OR REPLACE TABLE triggers an update_table request to the catalog. And the catalog is implemented by proxying the request directly to the underlying pyiceberg function https://github.com/kevinjqliu/iceberg-rest-catalog/blob/7c5548133ae266d4fac215b063911c35f08461d9/src/iceberg_rest/api/catalog_api.py#L453

I believe the issue is with the _commit_table function and how it deals with the incoming CommitTableRequest