Add Server-Side Scan Planning Support

[geruh]

Feature Request / Improvement

Now that Java has added server-side scan planning support PR #14480, I believe Python is a great place to integrate this functionality! We have all the building blocks and they are almost brought to completion. I'm creating this issue to track all of the tasks we need to drive it through

Context

We have some open PRs with some needed model changes, but we have pivoted to using our existing models and ensuring they're properly serializable with pydantic.

For example, initially we can work on:

1. Expression Serialization to ensure BooleanExpression and subclasses serialize correctly for REST API. Related to @Fokko's work on Remove Generic from expressions #2750 and @rambleraptor Server-side planning models #2435.

2. DataFile Serialization to ensure we can properly deserialize the data/deletefiles from the server response. Open API uses kebab-case (file-format, file-path) but our models expect snake_case (file_format, file_path).

REST API Endpoints to Implement

Based on the Iceberg REST spec:

1. POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan - Submit scan for planning
2. GET /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id} - Fetch planning result
3. DELETE /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id} - Cancel planning
4. POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/tasks - Fetch scan tasks for a plan task

Tasks

Initially we can start with core sync planning and once that's in place we can add the async support as it looks like it exists in https://github.com/apache/iceberg/blob/main/core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java

Core Sync Planning

• Build on @Fokko's expression work to ensure Expression classes serialize properly with Pydantic.
  • JSON to expression #2783
• Construct only plan Request/Response models for synchronous planning
• Add support for REST scan to fetch scan tasks replicating DataScan behavior
• Parse server response to FileScanTask objects (handle Data/DeleteFile construction)
• Add plan_table_scan() methods
• Add documentation

Full Scan planning support (Follow-up)

Complete the full scan planning API with async operations and pagination.

• Add the rest of the models for async planning
• Add support for endpoint 2 & 3 to RESTScan
• Add endpoint 4 support to RestCatalog
• Complete documentation with all scenarios

[kevinjqliu]

Thanks @geruh, super excited to see this.

I was able to prototype a client against the iceberg-rest-fixture docker image using the openapi spec and LLM. The api response look very similar to pyiceberg's FileScanTask (likely by design!)

Working on the "core sync planning" first makes sense to me; as i dont think the REST of the apis are implemented yet in the iceberg-rest-fixture.

From an user-facing api standpoint, today we use table.scan().to_arrow(). Perhaps we can expose this remote scan planning functionality like table.scan(remote=true).to_arrow() or table.remote_scan().to_arrow()
Just my initial thoughts, would love to hear what others think regarding how to expose this functionality

[geruh]

Hey @kevinjqliu, thanks! I built out a POC I'll open as a draft here shortly.

Working on the "core sync planning" first makes sense to me; as i dont think the REST of the apis are implemented yet in the iceberg-rest-fixture.

Yeah the logic is there but nothing to enable the async logic.

From an user-facing api standpoint, today we use table.scan().to_arrow(). Perhaps we can expose this remote scan planning functionality like table.scan(remote=true).to_arrow() or table.remote_scan().to_arrow()

I've been giving this some thought, and I found a PR on the Java side where they're using a catalog level property to enable REST scan planning: https://github.com/apache/iceberg/pull/13400/files#diff-86450612dbe323d6d06cbc3846aa1913f042eaedadc0ca027c36bfbe08d3a46cR263. I think it would make sense to align with this approach.

Especially if a catalog server supports scan planning, it can provide an config override to force scan planning. wdyt?