Skip to content

perf: cache Iceberg FileIO per executor instead of building one per task - #6106

Open
mixermt wants to merge 3 commits into
apache:mainfrom
mixermt:perf/cache-iceberg-file-io
Open

mixermt wants to merge 3 commits into
apache:mainfrom
mixermt:perf/cache-iceberg-file-io

Conversation

@mixermt

@mixermt mixermt commented Sep 22, 2026

Copy link
Copy Markdown

Which issue does this PR close?

Closes #6105.

Rationale for this change

Every native Iceberg scan or write task builds its own FileIO. On a production workload that was 40,317 FileIO constructions against one endpoint for a single query, each one rebuilding the storage factory, its parsed configuration and, for S3 with a custom access provider, the JNI bridge. Spark's Hadoop layer shares one client per executor JVM. Details in #6105.

What changes are included in this PR?

load_file_io in iceberg_common.rs now serves clones of a per-executor LRU cache and builds only on a miss; the previous body becomes build_file_io. Clones share iceberg-rust's Arc<OnceLock<Arc<dyn Storage>>>, so all tasks on an executor share one Storage.

The key is everything that shapes the client: access mode, catalog name, the full reference path, and the whole catalog property bag. The path is included because the S3 access bridge is constructed with the bucket and url.path() of the reference location and the JVM provider is called with exactly that pair, so a bridge built for one table must not serve another. memory:/// is never cached; the write path assembles manifest bytes there per task. The cache holds 64 entries and evicts the least recently used one, dropping it after the lock is released since the last clone of a FileIO releases JNI global refs. release_runtime drains it with the Tokio runtime. A read whose configured S3 access provider failed to initialise falls back to opendal's default chain, as before; that build is returned but not cached, so the next task retries the provider instead of inheriting the fallback.

What is actually shared, per backend

Sharing a FileIO shares what its Storage holds. At the pinned iceberg-rust rev that differs by backend:

Backend Storage holds Shared across tasks by this PR
s3, s3a, aliases, gs, oss config and the access loader; create_operator builds a new opendal Operator per file open factory and parsed config, and for a custom S3 provider the JNI bridge and its ensureInitialized. Not the client or signer: opendal builds a new Signer per operator, and HTTP pooling is already process-wide
hdfs (#5898) config plus an operator cache keyed by NameNode the hdfs-native client and its NameNode session, which is the case that motivated this change
memory the operator itself excluded from the cache on purpose
file nothing harmless

So for S3-family backends this PR is necessary but not sufficient for client reuse: an operator cache inside OpenDalStorage::S3, mirroring what the HDFS backend already does, would make the shared Storage also share the signer and its access cache. That is an iceberg-rust change, tracked in #6109. Without an executor-level FileIO cache such an operator cache would die with every task, which is why this lands first.

How are these changes tested?

  • Unit tests in iceberg_common.rs: the key separates access mode, path, catalog and properties and is None for memory:///; cached_file_io builds once for repeated identical loads and every time without a key; load_file_io("memory:///") leaves the cache untouched; a degraded build is returned but never cached; eviction removes the least recently used entry and a re-insert replaces without evicting.
  • cargo clippy --all-targets -p datafusion-comet -- -D warnings is clean.
  • JVM, with the rebuilt library: CometIcebergNativeSuite, CometIcebergWriteActionSuite and the MinIO-backed IcebergReadFromS3Suite, which runs REST catalog vending with wrong access immediately before correct access against the same bucket and catalog.
  • Not measured here: the end-to-end gain on the production workload.

AI Disclosure

Drafted, implemented and tested with AI assistance (Claude Code); reviewed before submission.

🤖 Generated with Claude Code

Every native Iceberg scan or write task built its own FileIO, and since
iceberg-rust creates the storage client lazily per instance, every task also
opened and tore down its own client: an S3 client, or with hdfs-native a
NameNode session. Spark's Hadoop layer shares one client per executor JVM.

load_file_io now serves clones from a per-executor cache and only builds on a
miss. The key is everything that shapes the client: access mode, catalog
name, the full reference path, and the whole catalog property bag. The path
is in the key because the S3 access bridge is constructed with the bucket and
path of the reference location, so a bridge built for one table must not
serve another. memory:/// is never cached, the map clears past 64 entries,
and release_runtime clears it with the runtime.

Closes apache#6105

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Review of apache#6106 found that the bound cleared the whole cache, including the
entry a running query was using, and dropped every FileIO while holding the
lock, where the last clone of an S3 access bridge releases JNI global refs.
The cache is now a small LRU that evicts one entry per insert, and evicted or
replaced FileIOs are dropped after the lock is released.

The builder is injected into cached_file_io so tests can count builds: a
repeated identical load builds once, a load without a key builds every time,
memory:/// never enters the cache, and eviction removes the least recently
used entry.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
@mixermt

mixermt commented Sep 22, 2026

Copy link
Copy Markdown
Author

Review of the first revision found three problems, all addressed in the latest push:

  1. The description overclaimed for S3. At the pinned iceberg-rust rev OpenDalStorage::S3 holds only config and the access loader; create_operator builds a new opendal Operator, and with it a new Signer, on every file open, while HTTP pooling is already process-wide. A shared FileIO therefore reuses the factory, the parsed config and the JNI bridge, not the S3 client. Client reuse holds for the HDFS backend in feat(iceberg): support HDFS storage via iceberg-rust hdfs-native backend [NOT FOR MERGE: pending apache/iceberg-rust#3111] #5898, whose Storage caches operators per NameNode. The description now states this per backend, and an operator cache for the S3 family is noted as an iceberg-rust follow-up in Native Iceberg scan and write build a new FileIO, and a new storage client, for every task #6105.
  2. The bound cleared the whole map, including the entry a running query was using, and dropped every FileIO while holding the lock. Replaced with LRU eviction of one entry, dropped after the lock is released.
  3. The test never exercised a cache hit. Added tests that a repeated identical load builds once, that a load without a key builds every time, that memory:/// never enters the cache, and that eviction removes the least recently used entry.

…lise

A read whose configured S3 access provider fails to initialise falls back
to opendal's default chain. Caching that FileIO made the fallback sticky
for every later task on the executor. storage_factory_for and
build_file_io now report whether the build is cacheable, and a degraded
build is returned without being inserted so the next task retries.

Also splits the S3 operator cache follow-up into apache#6109 and updates two
comments that still described per-task lifetimes.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
@mixermt

mixermt commented Sep 22, 2026

Copy link
Copy Markdown
Author

Second review round, three findings, all addressed in the latest push:

  1. A read-side provider init failure became sticky. build_s3_credential_loader falls back to opendal's default chain on a read, and that FileIO was cached like a success, so later tasks never retried ensureInitialized. storage_factory_for and build_file_io now report whether the build is cacheable; a fallback is returned but not inserted, so the next task retries. Covered by cached_file_io_does_not_cache_a_degraded_build.
  2. The S3 operator cache follow-up lived only in Native Iceberg scan and write build a new FileIO, and a new storage client, for every task #6105, which this PR closes. Split into Iceberg S3-family storage rebuilds the opendal Operator and signer on every file open #6109 and linked from the description.
  3. Two comments still described per-task lifetimes. The FILE_IO_CACHE rustdoc now says what is shared at this pin, and the missing-expiry latch comment in credential_bridge.rs reflects that bridges live as long as their cache entry.

@andygrove andygrove left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the careful per-backend breakdown in the description. It makes the scope of this change much easier to reason about.

HDFS support (#5898) isn't on main yet, so today the gain comes only from the S3 factory and bridge construction. CometS3CredentialDispatcher.ensureInitialized already caches the provider JVM-side via KEY_TO_HANDLE.computeIfAbsent, so the per-task bridge cost is two new_string calls, two global refs and a map lookup. The provider isn't re-initialised. Do you have measurements showing the per-task FileIO build cost matters for S3 on its own? If not, would it make sense to land this together with #5898 or #6109, where the cache is clearly needed?

const FILE_IO_CACHE_CAPACITY: usize = 64;

/// Least recently used entries are evicted first.
struct FileIoCache {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A cached CometS3CredentialBridge now keeps its dispatcher handle for the executor's lifetime, where before each task got a fresh one from ensureInitialized. If CometS3CredentialDispatcher.closeAll() runs, every cached bridge will fail on getCredentialsForPath with a missing-handle error and never recover. Could we either clear FILE_IO_CACHE from closeAll, or document that closeAll only runs at JVM shutdown? CometS3CredentialDispatcherTest does call it directly.

}
}

#[derive(Clone, Debug, PartialEq, Eq, Hash)]

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The key carries the whole catalog property bag, which can include vended s3.access-key-id, s3.secret-access-key and s3.session-token. Could we drop Debug here, or implement a redacting one, so a future {:?} can't leak credentials into logs?


/// Shared per executor so tasks reuse one FileIO: its factory, parsed config and access bridge,
/// plus the storage client where the backend caches operators.
static FILE_IO_CACHE: LazyLock<Mutex<FileIoCache>> =

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

With #5898 this makes the hdfs-native client and its NameNode session live for the executor's lifetime. Have you checked how that client behaves when a Kerberos ticket or delegation token expires on a long-lived connection?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

area:Iceberg area:scan Parquet scan / data reading enhancement New feature or request performance

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Native Iceberg scan and write build a new FileIO, and a new storage client, for every task

2 participants