Iceberg scan fails with a hard-wired 10s OpenDAL io_timeout that cannot be configured

[comphead]

Describe the bug

A native Iceberg scan fails on a hard-wired 10 second per-IO timeout that nothing in Comet can configure:

org.apache.comet.CometNativeException: Iceberg scan error: Unexpected => Failed to read a Parquet file,
  source: External: Unexpected => Failure in doing io operation,
  source: Unexpected (persistent) at read, context: { timeout: 10 } => io operation timeout reached

The 10 is OpenDAL's default io_timeout. iceberg-storage-opendal applies TimeoutLayer::new() to every FileIO operator with no way to override it:

https://github.com/apache/iceberg-rust/blob/665c64e48e8d33797ecb1a421f327edd9b024879/crates/storage/opendal/src/lib.rs#L394

It bounds one ReadStream::read() chunk, not the whole file, so this is not a large-file problem. The RetryLayer outside it re-sends the same request, so a range that cannot fit in 10s fails identically on all four attempts and surfaces as persistent.

Comet cannot work around it: create_operator and the S3 config parsers are pub(crate), and iceberg::io::Storage is opaque, so BlobHostPromotingS3StorageFactory receives an already-layered Arc<dyn Storage>. Layers stack rather than replace, so adding a longer one on the outside does not help.

Expected behavior

The per-IO timeout should be tunable rather than failing the job.

Additional context

Fix proposed upstream: apache/iceberg-rust#3263, which adds client.io-timeout-ms and hands it to TimeoutLayer::with_io_timeout. Unset keeps the 10s default. Tracked by apache/iceberg-rust#2977; #3179 covers only the write path.

Once it merges, the Comet side is small because "client." is already in STORAGE_PROPERTY_PREFIXES (native/core/src/execution/operators/iceberg_common.rs:39):

• Bump the pinned rev at native/Cargo.toml:67-68. No native change needed.
• Add a CometConf entry (.timeConf(TimeUnit.MILLISECONDS)) and inject it at CometScanRule.scala:590 and CometIcebergNativeWrite.scala:687. Both read SQLConf at planning time, so unlike the fs.comet.* Hadoop keys this one would honor a runtime spark.conf.set.

Worth ruling out first, unverified for this report: TimeoutLayer measures wall clock, and Comet shares one tokio runtime per executor sized to spark.executor.cores (jni_api.rs:288-296). CPU-bound shuffle work on the same workers can leave an IO future unpolled past 10s on a healthy connection. Raising COMET_WORKER_THREADS tests that without a code change.

[andygrove]

The worker-starvation explanation in the last paragraph holds up in a standalone test. Only Tokio workers drive the I/O and timer driver, so an I/O future polled from a Spark task thread's block_on gets no wake-ups while every worker is busy. With tokio 1.53 and one worker stuck in a 2 s poll, a 10 ms sleep in another thread's block_on took 1.9 s, and so did a socket read whose peer wrote after 300 ms. With two workers both busy it was the same.

Two things make that more likely than the executor-cores sizing suggests. On a standalone cluster without spark.executor.cores the runtime has a single worker (#6292). And several JVM calls block a worker from inside a poll (#6293), including the S3 credential provider, which is called on every request. Running the failing query with COMET_WORKER_THREADS set to the executor's core count would show whether this is what you're hitting.