Core: REST Scan Planning Task Implementation

[singhpk234]

🥞 Stacked PR

Please use this link for viewing incremental changes.

Current Stack status:

• Add integration with iceberg-core
  * Add Rest Model and Parsers [Merged]

About the change

[1] Add routes for the scan Planning API
[2] Adds RestTable and RestTableScan which can call the scan endpoint
[3] ScanIterable which uses ParallelIterable to fetch scan tasks

Testing

Added unit testing for the E2E req response loop

[amogh-jahagirdar]

I'm still going through things but I do believe this works at least from a correctness perspective. I still would need to give some more thought as to cancellation, client/server backpressure and how this would fit in for engines which immediately start task consumption/execution during planning (like Trino)

[singhpk234]

Thank you for the review, presently i just made E2E machinery work, with later parser changes, i think in addition to the points you mentioned, i was thinking of more also from the POV :

1. can server force client to call scan plan api ?
2. can server tell the client what should be interval between the fetch scan tasks ? (i think you refer that as back-pressure in above comment of yours)
3. I don't know but i keep coming to this feature [CORE] Support file filtering based on schema #4842 , it may be orthogonal
   but since for schema evolution, we certainly don't wanna send back data file objects which certainly don't have the column

[singhpk234]

Update : Working on refactoring this a bit more.

client/server backpressure and how this would fit in for engines which immediately start task consumption/execution during planning (like Trino)

My understanding was ParallelIterable could be helpful here as its aware of the both consumer and the producer,
and handle backpressure via yields

iceberg/core/src/main/java/org/apache/iceberg/util/ParallelIterable.java

Lines 200 to 205 in be03c99

	// If the consumer is processing records more slowly than the producers, the producers will
	// eventually fill the queue and yield, returning continuations. Continuations and new tasks
	// are started by checkTasks(). The check here prevents us from restarting continuations or
	// starting new tasks before the queue is emptied. Restarting too early would lead to tasks
	// yielding very quickly (CPU waste on scheduling).
	if (!queue.isEmpty()) {

Agree need to think this more thoroughly also from the server POV,

from my cursory reading of Trino source code (I am fairly new to it) https://github.com/trinodb/trino/blob/master/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitManager.java#L149 split generation and consuming it, should mostly work, as this seems like this is built in engine itself. for engine which needs all the splits computed first we have no choice but to consume everything.

[nastra]

we should maybe try and avoid "busy waiting" here and instead use a higher level concurrency concept to achieve the same thing?

[nastra]

maybe this entire block should be using the functionality from Tasks where you can also specify an exponential backoff

[singhpk234]

Tasks retry behaviour is wired to exception handling, i can throw a inprogress exception to make it work with Task, though i think throwing an exception might not be the right thing to do.
failsafe has good retry policy i see in aws module we do use that please let me know if you are open to introducing this in core and then we can use lib to do exponential backoff.

[nastra]

I did another pass and left a few more comments, focusing mostly on RESTTable and RESTTableScan.

I believe we also need some tests with a REST server that doesn't support the new endpoints and where we verify that the correct errors are coming back depending on which endpoints the server supports. Basically you'd have a custom Adapter which would throw or not when a particular endpoint is called

Maybe we should extract all of the server-side scan planning tests into a separate class, because TestRESTCatalog is hitting almost 4k LOC.