mixermt opened a new issue, #6105:
URL: https://github.com/apache/datafusion-comet/issues/6105
### What is the problem the feature request solves?
The native Iceberg operators build a fresh `FileIO`, and with it a fresh
storage client, for every Spark task.
`IcebergScanExec::execute_with_tasks` calls `load_file_io` on each partition
(`native/core/src/execution/operators/iceberg_scan.rs:181`), and the native
write path does the same per task (`iceberg_write.rs:484`). In iceberg-rust, a
`FileIO` owns its storage client lazily:
```rust
pub struct FileIO {
config: StorageConfig,
factory: Arc<dyn StorageFactory>,
storage: Arc<OnceLock<Arc<dyn Storage>>>, // created on first use, per
instance
}
```
Because each task constructs its own instance, each task also constructs its
own client: a new S3 client, or with the `hdfs-native` backend a new NameNode
RPC session and handshake. The instance is dropped when the task finishes, so
nothing is carried over to the next task on the same executor.
Spark's own Hadoop layer does the opposite. `FileSystem.get` returns a
cached instance per `(scheme, authority, user)` for the lifetime of the JVM, so
all tasks on an executor share one client and its connection state.
On a production workload the native Iceberg scan ran 40,317 tasks across
three stages, each performing a full client set-up and tear-down against the
same storage endpoint. The CPU cost of constructing the object is negligible;
the costs that matter are:
- **Connect latency per task.** Every task pays service discovery,
authentication and session establishment before its first byte is read. For
short tasks this is a visible share of task duration.
- **Churn on the storage service.** Tens of thousands of short-lived
sessions against the same endpoint, where Spark's reader opens a handful. On
HDFS this is NameNode connection churn.
- **No reuse of warm state.** Connection pools, TLS sessions and any
client-side caching are discarded at task boundaries and rebuilt from scratch.
This is independent of #6091, though both were found on the same workload.
### Describe the potential solution
Cache `FileIO` instances per executor in `iceberg_common.rs` and hand tasks
clones, which share the same underlying `Arc<OnceLock<Arc<dyn Storage>>>` and
therefore the same client.
The correctness of this rests entirely on the cache key, because a `FileIO`
carries the access configuration it was built with. Key on every input that
shapes the client:
- **Access mode.** Read and write can be granted different access by the S3
bridge.
- **Catalog name.** The dispatch key used when vending access.
- **`scheme://authority`.** The storage endpoint. Two tables under the same
authority legitimately share a client; two authorities must not.
- **The full catalog property bag.** A REST catalog vends short-lived access
that rotates, and endpoints can be reconfigured. Including the whole bag means
a rotated entry produces a new cache entry rather than reusing a client built
with the previous one.
Two additional constraints:
- **Never cache `memory:///`.** The write path assembles manifest bytes in
that in-process namespace (`iceberg_write.rs:1156`); sharing it would let
concurrent tasks observe each other's state.
- **Bound the cache.** With per-query rotation the key space is unbounded,
so cap the map and clear it when the cap is exceeded. Tasks holding clones are
unaffected, since each clone keeps its own reference.
Clear the cache from `release_runtime` so the clients are released together
with the Tokio runtime rather than outliving it.
### Additional context
An implementation following the above is ready and will be submitted as a PR
shortly. It is covered by unit tests for the key and the caching behaviour, and
verified against `CometIcebergNativeSuite` (107 tests),
`CometIcebergWriteActionSuite` (69 tests) and the MinIO-backed
`IcebergReadFromS3Suite` (9 tests). The S3 suite is the meaningful one for key
correctness: it exercises REST catalog vending, including a case that rejects a
wrong grant immediately before a case that reads successfully with a correct
one against the same bucket and catalog.
What is not measured here is the end-to-end gain on the production workload,
which requires a run with the patched build.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
