You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/05 11:43:14 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-6649-4b679c68c54ae1171fdb651adbd0b7c43932cde0 created (now 965c8bbe28)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6649-4b679c68c54ae1171fdb651adbd0b7c43932cde0
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at 965c8bbe28 fix: make the native Iceberg per-I/O timeout configurable (#6649)
This branch includes the following new commits:
new 965c8bbe28 fix: make the native Iceberg per-I/O timeout configurable (#6649)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: fix: make the native Iceberg per-I/O timeout configurable (#6649)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6649-4b679c68c54ae1171fdb651adbd0b7c43932cde0
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 965c8bbe289ff850b614c8c3833ed7802134115b
Author: Oleks V <co...@users.noreply.github.com>
AuthorDate: Mon Oct 5 11:43:05 2026 +0000
fix: make the native Iceberg per-I/O timeout configurable (#6649)
Bump iceberg-rust to af1da4c (apache/iceberg-rust#3263), which adds the
`opendal.io-timeout-ms` FileIO property, and set it from the new
`spark.comet.iceberg.ioTimeout` config (default 30s) for native Iceberg
scans and writes. Forward `opendal.*` keys to the native FileIO.
The bump also needs two API updates: `FileWrite::close` now returns
`FileMetadata`, and `UnboundPartitionField` is built with its builder.
Closes #6124.
---
.../s3-credential-provider-design.md | 2 +-
native/Cargo.lock | 24 +++--------------
native/Cargo.toml | 4 +--
.../core/src/execution/operators/iceberg_common.rs | 16 +++++++++--
.../execution/operators/iceberg_location_scoped.rs | 16 ++++++-----
native/core/src/execution/planner.rs | 31 +++++++++-------------
.../main/scala/org/apache/comet/CometConf.scala | 9 +++++++
.../serde/operator/CometIcebergNativeScan.scala | 6 ++++-
.../serde/operator/CometIcebergNativeWrite.scala | 3 ++-
9 files changed, 59 insertions(+), 52 deletions(-)
diff --git a/docs/source/contributor-guide/s3-credential-provider-design.md b/docs/source/contributor-guide/s3-credential-provider-design.md
index d2f00c0c41..d4fa3e1baa 100644
--- a/docs/source/contributor-guide/s3-credential-provider-design.md
+++ b/docs/source/contributor-guide/s3-credential-provider-design.md
@@ -142,7 +142,7 @@ The 5-minute fallback on the Iceberg path bounds how long reqsign reuses a crede
## Property-bag handling on the Iceberg path
-The full unfiltered FileIO property bag crosses JNI as `catalog_properties`. The storage-prefix filter (`s3.`/`gcs.`/`adls.`/`client.`) is applied native-side in `iceberg_common.rs::build_file_io` immediately before `FileIOBuilder.with_prop`. This means the bridge sees `credentials.uri`, OAuth tokens, and any vendor-custom keys with no parallel field on the operator and no driver-side broadcast. Vendors set their own keys on the catalog config and read them back inside `initialize(Map)`.
+The full unfiltered FileIO property bag crosses JNI as `catalog_properties`. The storage-prefix filter (`s3.`/`gcs.`/`adls.`/`client.`/`opendal.`) is applied native-side in `iceberg_common.rs::build_file_io` immediately before `FileIOBuilder.with_prop`. This means the bridge sees `credentials.uri`, OAuth tokens, and any vendor-custom keys with no parallel field on the operator and no driver-side broadcast. Vendors set their own keys on the catalog config and read them back inside `initia [...]
`IcebergScanExec` derives a redacting `Debug`, and the `FileIO` cache key's `Debug` omits the property bag, so plan dumps and tracing do not leak it.
diff --git a/native/Cargo.lock b/native/Cargo.lock
index 0e157bdda8..fe72674cdb 100644
--- a/native/Cargo.lock
+++ b/native/Cargo.lock
@@ -2896,12 +2896,6 @@ dependencies = [
"syn 3.0.5",
]
-[[package]]
-name = "dissimilar"
-version = "1.0.11"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "aeda16ab4059c5fd2a83f2b9c9e9c981327b18aa8e3b313f7e6563799d4f093e"
-
[[package]]
name = "dlv-list"
version = "0.5.2"
@@ -3002,16 +2996,6 @@ dependencies = [
"pin-project-lite",
]
-[[package]]
-name = "expect-test"
-version = "1.5.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "63af43ff4431e848fb47472a920f14fa71c24de13255a5692e93d4e90302acb0"
-dependencies = [
- "dissimilar",
- "once_cell",
-]
-
[[package]]
name = "fastnum"
version = "0.7.5"
@@ -3578,7 +3562,7 @@ dependencies = [
[[package]]
name = "iceberg"
version = "0.10.1"
-source = "git+https://github.com/apache/iceberg-rust?rev=bb1e4a4861f02377489eff818b75138f414c4cb0#bb1e4a4861f02377489eff818b75138f414c4cb0"
+source = "git+https://github.com/apache/iceberg-rust?rev=af1da4c5bc86178c38c1c6db0578bdd9fb7021e3#af1da4c5bc86178c38c1c6db0578bdd9fb7021e3"
dependencies = [
"aes-gcm",
"anyhow",
@@ -3601,7 +3585,6 @@ dependencies = [
"chrono",
"crc32fast",
"derive_builder",
- "expect-test",
"fastnum",
"flate2",
"fnv",
@@ -3637,7 +3620,7 @@ dependencies = [
[[package]]
name = "iceberg-property-macro"
version = "0.10.1"
-source = "git+https://github.com/apache/iceberg-rust?rev=bb1e4a4861f02377489eff818b75138f414c4cb0#bb1e4a4861f02377489eff818b75138f414c4cb0"
+source = "git+https://github.com/apache/iceberg-rust?rev=af1da4c5bc86178c38c1c6db0578bdd9fb7021e3#af1da4c5bc86178c38c1c6db0578bdd9fb7021e3"
dependencies = [
"proc-macro2",
"quote",
@@ -3647,7 +3630,7 @@ dependencies = [
[[package]]
name = "iceberg-storage-opendal"
version = "0.10.1"
-source = "git+https://github.com/apache/iceberg-rust?rev=bb1e4a4861f02377489eff818b75138f414c4cb0#bb1e4a4861f02377489eff818b75138f414c4cb0"
+source = "git+https://github.com/apache/iceberg-rust?rev=af1da4c5bc86178c38c1c6db0578bdd9fb7021e3#af1da4c5bc86178c38c1c6db0578bdd9fb7021e3"
dependencies = [
"anyhow",
"async-trait",
@@ -3655,6 +3638,7 @@ dependencies = [
"cfg-if",
"futures",
"iceberg",
+ "iceberg-property-macro",
"opendal",
"reqsign-aws-v4",
"reqsign-core",
diff --git a/native/Cargo.toml b/native/Cargo.toml
index 3b14429765..fa6ea26a21 100644
--- a/native/Cargo.toml
+++ b/native/Cargo.toml
@@ -80,8 +80,8 @@ aws-sdk-sts = { version = "1.114.0", default-features = false }
# Pinned to a revision rather than a release because no published iceberg-rust
# version is on our arrow: 0.10.1 requires arrow/parquet ^58. Move to a plain
# version requirement once a release ships on arrow 59. Bump policy: #5645.
-iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "bb1e4a4861f02377489eff818b75138f414c4cb0" }
-iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "bb1e4a4861f02377489eff818b75138f414c4cb0", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] }
+iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "af1da4c5bc86178c38c1c6db0578bdd9fb7021e3" }
+iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "af1da4c5bc86178c38c1c6db0578bdd9fb7021e3", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] }
reqsign-core = "3"
[profile.release]
diff --git a/native/core/src/execution/operators/iceberg_common.rs b/native/core/src/execution/operators/iceberg_common.rs
index 94df0bd70e..f4ea3f7983 100644
--- a/native/core/src/execution/operators/iceberg_common.rs
+++ b/native/core/src/execution/operators/iceberg_common.rs
@@ -44,8 +44,9 @@ const ICEBERG_PROVIDER_CLASS_PROPERTY: &str = "s3.comet.credential.provider.clas
/// Key prefixes forwarded to iceberg-rust's `FileIO`. The full unfiltered catalog bag (catalog
/// URI, OAuth tokens, credentials.uri, tenant-id, etc.) is kept upstream so
-/// `CometS3CredentialBridge` can read whatever the vendor needs.
-const STORAGE_PROPERTY_PREFIXES: &[&str] = &["s3.", "gcs.", "adls.", "client."];
+/// `CometS3CredentialBridge` can read whatever the vendor needs. `opendal.` carries
+/// iceberg-storage-opendal's own settings, such as `opendal.io-timeout-ms`.
+const STORAGE_PROPERTY_PREFIXES: &[&str] = &["s3.", "gcs.", "adls.", "client.", "opendal."];
/// Pick an OpenDAL storage backend from a URI's scheme. `file` (or no scheme) falls through to
/// the local file system. `memory` is used by the write path to assemble manifest bytes that
@@ -775,6 +776,17 @@ mod tests {
.all(|k| scheme_of(&k.reference_path) != "memory"));
}
+ /// The JVM sets this key from `spark.comet.iceberg.ioTimeout`.
+ #[test]
+ fn io_timeout_reaches_the_file_io() {
+ let key = iceberg_storage_opendal::OPENDAL_IO_TIMEOUT_MS;
+ assert_eq!(key, "opendal.io-timeout-ms");
+ let props = HashMap::from([(key.to_string(), "30000".to_string())]);
+ let (file_io, _) =
+ build_file_io(&props, "file:///tmp/warehouse", "", AccessMode::Read).unwrap();
+ assert_eq!(file_io.config().get(key).map(String::as_str), Some("30000"));
+ }
+
#[test]
fn cache_evicts_the_least_recently_used_entry() {
let mut cache = FileIoCache::new(2);
diff --git a/native/core/src/execution/operators/iceberg_location_scoped.rs b/native/core/src/execution/operators/iceberg_location_scoped.rs
index eb7fd93799..d1ad474485 100644
--- a/native/core/src/execution/operators/iceberg_location_scoped.rs
+++ b/native/core/src/execution/operators/iceberg_location_scoped.rs
@@ -654,12 +654,12 @@ impl LocationScopedFileWrite {
/// `result` of a write call made at the snapshot of `generation`, after fetching `bucket`'s
/// locations again if it failed in a way that may mean they changed. A free function so the
/// writer, which is not `Sync`, is not borrowed across the refresh.
-async fn refresh_on_failure(
+async fn refresh_on_failure<T>(
inner: &Inner,
bucket: &BucketLocations,
generation: u64,
- result: Result<()>,
-) -> Result<()> {
+ result: Result<T>,
+) -> Result<T> {
match result {
Err(e) if may_mean_stale_locations(&e) => {
match inner.shared.refresh(bucket, generation).await {
@@ -679,7 +679,7 @@ impl FileWrite for LocationScopedFileWrite {
refresh_on_failure(&self.inner, &self.route.bucket, generation, result).await
}
- async fn close(&mut self) -> Result<()> {
+ async fn close(&mut self) -> Result<FileMetadata> {
let generation = self.generation();
let result = self.writer.close().await;
refresh_on_failure(&self.inner, &self.route.bucket, generation, result).await
@@ -797,8 +797,10 @@ mod tests {
Ok(())
}
- async fn close(&mut self) -> Result<()> {
- self.0.check("close", &self.1)
+ async fn close(&mut self) -> Result<FileMetadata> {
+ self.0
+ .check("close", &self.1)
+ .map(|_| FileMetadata { size: 0 })
}
}
@@ -1512,7 +1514,7 @@ mod tests {
let output = storage.new_output("s3://b/t/new/f.parquet").unwrap();
let mut writer = output.writer().await.unwrap();
writer.write(Bytes::from("data")).await.unwrap();
- let err = writer.close().await.unwrap_err();
+ let err = writer.close().await.err().expect("close should fail");
assert!(may_mean_stale_locations(&err), "{err}");
assert_eq!(fixture.fetches(), 1);
let mut writer = output.writer().await.unwrap();
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index aec252e2e8..c8dc8fa46a 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -7298,15 +7298,20 @@ mod tests {
.expect("schema");
let schema_arc = Arc::new(iceberg_schema);
+ let identity_field = |source_id: i32, field_id: i32, name: &str| {
+ iceberg::spec::UnboundPartitionField::builder()
+ .source_ids(vec![source_id])
+ .field_id(field_id)
+ .name(name)
+ .transform(iceberg::spec::Transform::Identity)
+ .build()
+ .expect("unbound field")
+ };
+
// Spec 0 (older): field 1000 named "region_old".
let spec0 = PartitionSpec::builder(Arc::clone(&schema_arc))
.with_spec_id(0)
- .add_unbound_field(iceberg::spec::UnboundPartitionField {
- source_id: 2,
- field_id: Some(1000),
- name: "region_old".to_string(),
- transform: iceberg::spec::Transform::Identity,
- })
+ .add_unbound_field(identity_field(2, 1000, "region_old"))
.expect("add field")
.build()
.expect("build spec0");
@@ -7314,19 +7319,9 @@ mod tests {
// Spec 1 (newer): same field id 1000 renamed to "region_new", plus a new field 2000.
let spec1 = PartitionSpec::builder(Arc::clone(&schema_arc))
.with_spec_id(1)
- .add_unbound_field(iceberg::spec::UnboundPartitionField {
- source_id: 2,
- field_id: Some(1000),
- name: "region_new".to_string(),
- transform: iceberg::spec::Transform::Identity,
- })
+ .add_unbound_field(identity_field(2, 1000, "region_new"))
.expect("add field")
- .add_unbound_field(iceberg::spec::UnboundPartitionField {
- source_id: 3,
- field_id: Some(2000),
- name: "category".to_string(),
- transform: iceberg::spec::Transform::Identity,
- })
+ .add_unbound_field(identity_field(3, 2000, "category"))
.expect("add field")
.build()
.expect("build spec1");
diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala
index 5c588017ed..5465de1f64 100644
--- a/spark/src/main/scala/org/apache/comet/CometConf.scala
+++ b/spark/src/main/scala/org/apache/comet/CometConf.scala
@@ -152,6 +152,15 @@ object CometConf extends ShimCometConf {
.checkValue(v => v > 0, "Data file concurrency limit must be positive")
.createWithDefault(1)
+ val COMET_ICEBERG_IO_TIMEOUT: ConfigEntry[Long] =
+ conf("spark.comet.iceberg.ioTimeout")
+ .category(CATEGORY_SCAN)
+ .doc("Timeout for each storage I/O call, such as one read or write, that Comet's native " +
+ "Iceberg scan and write make. Passed to iceberg-rust as `opendal.io-timeout-ms`.")
+ .timeConf(TimeUnit.MILLISECONDS)
+ .checkValue(_ > 0, "The Iceberg I/O timeout must be positive")
+ .createWithDefault(TimeUnit.SECONDS.toMillis(30))
+
val COMET_CSV_V2_NATIVE_ENABLED: ConfigEntry[Boolean] =
conf("spark.comet.scan.csv.v2.enabled")
.category(CATEGORY_TESTING)
diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
index 689ed68389..d786583377 100644
--- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
@@ -645,6 +645,10 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
global.result() ++ promoted.result()
}
+ /** iceberg-rust's per-I/O-call timeout, set from `spark.comet.iceberg.ioTimeout`. */
+ def ioTimeoutProperty(): (String, String) =
+ "opendal.io-timeout-ms" -> CometConf.COMET_ICEBERG_IO_TIMEOUT.get().toString
+
/**
* Converts an Iceberg residual Expression into an IcebergPredicate for the native scan.
*
@@ -1020,7 +1024,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
commonBuilder.setDataFileConcurrencyLimit(
CometConf.COMET_ICEBERG_DATA_FILE_CONCURRENCY_LIMIT.get())
metadata.catalogName.foreach(commonBuilder.setCatalogName)
- metadata.catalogProperties.foreach { case (key, value) =>
+ (metadata.catalogProperties + ioTimeoutProperty()).foreach { case (key, value) =>
commonBuilder.putCatalogProperties(key, value)
}
diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala
index 2c5d00f2b5..5d5510fb7a 100644
--- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala
@@ -766,7 +766,8 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
val hadoopDerivedProperties = CometIcebergNativeScan.hadoopToIcebergS3Properties(
NativeConfig.extractObjectStoreOptions(writeHadoopConf, dataUri),
dataBucket)
- val catalogProperties = hadoopDerivedProperties ++ fileIOProperties
+ val catalogProperties =
+ hadoopDerivedProperties ++ fileIOProperties + CometIcebergNativeScan.ioTimeoutProperty()
val common = IcebergWriteProtoTranslation.buildCommon(
catalogProperties = catalogProperties,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org