Repository navigation
feat(storage): Implement ObjectStoreStorage::S3 (Supersedes #2257) - #3165
Sruhvx-jpg wants to merge 32 commits into
Conversation
…oncurrent deletes - Hoist `object_store` 0.13 to workspace dependencies to align with DataFusion. - Support `s3n://` scheme alongside `s3://` and `s3a://` in `parse_s3_url`. - Optimize `delete_stream` with `try_for_each_concurrent` instead of sequential loop. - Add unit tests for `s3n://` URL parsing and FileIO/Storage serialization roundtrips. - Wire crate workspace lints and publish flag.
|
Apologies for any notification noise from the extra PR earlier. Everything has been cleanly unified into this PR :) |
a476be2 to
37180ce
Compare
|
Hey everyone, got all the CI checks passing and green now! Since this is a pretty big PR, just wanted to say that if you guys like the work, I'd really love to stick around and continue making it better—handling any feedback, tuning performance, and helping add other backends like GCS or Azure down the line. Whenever you get some time to check it out, let me know what you think! :) |
|
cc @CTTY @kevinjqliu — CI is completely green on this. Since this directly revives and finishes #2257, whenever you have a moment to take a look, I'd really appreciate your review on the S3 backend implementation! |
laskoviymishka
left a comment
laskoviymishka
left a comment
There was a problem hiding this comment.
Really glad to see object_store wired into the Storage framework — the factory, per-bucket cache, and path plumbing are all in good shape, and this is the right base for the follow-up backends. I'd hold it before merge though, since the whole object_store stack is going to build on top of this crate and a few of these are hard to walk back once it's published.
The one that worries me most is build_s3_store silently dropping most of S3Config. When an operator sets s3.sse.type=kms or custom, TryFrom populates the SSE fields but nothing forwards them to the builder, so we'd write data unencrypted even though encryption was explicitly required — a silent security regression versus opendal/Java. AmazonS3Builder has with_sse_kms_encryption/with_ssec_encryption for this, and for the fields object_store genuinely can't express (assume-role, disable-ec2-metadata) I'd return an error rather than drop them silently.
Things I'd like to settle in this PR before the follow-ups build on it:
- Forward the SSE config, and error on the config fields object_store can't express instead of dropping them
- Rework
parse_s3_urlto use the parsedUrlfields instead of slicing the raw string (uppercase scheme + percent-encoded bucket both break today) - Make the
store_cache/configvariant fields private before the first publish - Add a
Dropthat aborts the multipart upload so a dropped writer doesn't orphan parts - Land at least a thin integration test against localstack/MinIO — nothing currently exercises a real read/write
None of it is structural — the design is right. Once those are addressed I'm happy to take another pass and approve.
|
@laskoviymishka thanks for follow up, perhaps it's my fault I didnt properly review the CITY's code. Now that u have pointed out these issues I suspect there must be more of em, So I would like to take about 2 days minimum to get everything resolved-review-amended and also look for unknown hiccups. This means more research Currently its night here so I will get to reading ur followup thoroughly tommarrow 😊 Again, thanks for the detailed follow up |
|
@Sruhvx-jpg no rush, keep your time here. |
|
@laskoviymishka addressed all the feedback and added several hardening improvements:
All checks, clippy lints, and CI tests are green. Ready for another pass whenever you have time! |
37180ce to
6457dbd
Compare
|
Also on the struct wrappers ( If u like such style, we can maybe expand its usage as I have never seen a rust code with wrapper structs, which is understandable as is an repetitive job - but with advent of ai its much easier |
…blic-api baseline
97bcddf to
ed8a391
Compare
|
@laskoviymishka Hey! All CI's Green - "Make no mistake" worked this time 😄 |
Inherited from main (tracked in apache#3222); same bump as apache#3165. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
laskoviymishka
left a comment
laskoviymishka
left a comment
There was a problem hiding this comment.
Almost there, thanks for turning this around fast.
The one thing that blocks merge is the SSE-KMS default-key path. Forwarding the key landed, but when s3.sse.type=kms is set with no explicit key we pass "" into with_sse_kms_encryption, which emits an empty key-id header on every PUT and S3 rejects it with InvalidArgument. The store still builds, so writes fail silently — and "use the bucket's default CMK" is the most common KMS setup. Java and opendal send aws:kms with no key-id header there; I'd match that.
Everything else I asked for last round is in: SSE config is forwarded, parse_s3_url is rebuilt on Url fields (uppercase schemes work now), store_cache/config are private, the Drop abort is present, and the MinIO tests exercise real reads and writes.
Two of those aren't fully closed yet, though — both smaller than the blocker:
- the percent-encoded bucket half of the parse fix is still open (
s3://my%2Dbucket/...gives a bucket no S3 store has; the current test pins that behaviour rather than fixing it) - the
Dropis in, butclose()consumes the writer beforefinish(), so on a failed finish there's nothing left to abort — the exact case it's for — and the multipartwriter()path still has no test
Fix the KMS path and I'm happy to take another pass. I'd like the other two closed here too since the follow-up backends build on this, but they're mechanical from where you are now.
…efix, and multipart tests
|
@laskoviymishka Addressed all feedback from the latest review:
All checks, clippy lints, and integration tests are green. Ready for another pass! |
laskoviymishka
left a comment
laskoviymishka
left a comment
There was a problem hiding this comment.
The panic is gone — delete_batch_size() clamps in the accessor now, so a 0 coming in through Deserialize can't reach chunks(0), and test_storage_deserialization_clamps_delete_batch_size pins it. The failed-write() path takes self.upload out too, so Drop can't double-abort, and the three consts are in public-api.txt. The percent-encoded key is handled right as well — raw extraction keeps %2F literal and lets ObjectStorePath::from re-encode, so the wire key matches Java. Thanks for turning all that around.
Two things before I'm comfortable merging. The public-api.txt guard still isn't real — the consts landed, but publish = false keeps the crate out of check-public-api, so nothing diffs the snapshot and we're back to the decorative state from round 4. That's the half I was actually after when I folded it into the gate last round; since this is the base crate the others snapshot against, I'd like the guard running in CI before we merge (inline).
The other is new from reading the throttle again: the task-panic arm is the one spot the abort doesn't reach. When a spawned put_part task panics rather than returning Err, join_next() hands back Err(join_err) and we return without aborting, so the multipart upload leaks until a lifecycle rule expires it — the Ok(Err(e)) arm right beside it already does the right thing (inline).
Those two and I'm ready to sign off — the rest is in good shape.
| edition = { workspace = true } | ||
| license = { workspace = true } | ||
| name = "iceberg-storage-object_store" | ||
| publish = false |
There was a problem hiding this comment.
The three consts are in the snapshot now, thanks. The guard still isn't actually running, though — publish = false keeps this crate out of check-public-api (the Makefile selects publish == null), so nothing diffs the file and we're back to the decorative state from round 4. Since this is the base crate the other backends snapshot against, I'd make it real before merge: flip to publish = true with a pre-release version suffix if we're not ready for crates.io, or add this crate to the public-api CI job explicitly.
| Ok(Err(e)) => { | ||
| return Err(Self::abort_and_wrap(upload, e, "part upload failure").await); | ||
| } | ||
| Err(join_err) => { |
There was a problem hiding this comment.
The task-panic arm is the one spot the abort doesn't reach. When a spawned put_part task panics instead of returning Err, join_next() hands back Err(join_err) and we return without calling upload.abort(), so the multipart upload is left open until a lifecycle rule expires it — the Ok(Err(e)) arm right above already does the right thing via abort_and_wrap. I'd mirror it here, and in the matching arm in close() (line 553):
Err(join_err) => {
if let Err(abort_err) = upload.abort().await {
tracing::warn!(error = %abort_err, "Failed to abort multipart upload after task panic");
}
return Err(Error::new(ErrorKind::Unexpected, "Part upload task panicked").with_source(join_err));
}| let target_partition = storage | ||
| .get_store_and_path("s3://my-bucket/dt=a%2Fb/file.parquet") | ||
| .unwrap(); | ||
| assert_eq!(target_partition.path.as_ref(), "dt=a%252Fb/file.parquet"); |
There was a problem hiding this comment.
The key handling reads correctly now — raw extraction keeps %2F literal and ObjectStorePath::from re-encodes the %, so S3 decodes it back to dt=a%2Fb on the wire and we match Java/opendal. This assertion only checks the intermediate object_store repr, though, not what S3 ends up storing. Not blocking, but an integration test in file_io_s3_test.rs that writes and reads back a dt=a%2Fb partition path would lock the wire behavior down.
104c53b to
9f565af
Compare
|
@laskoviymishka Addressed the feedback! Flipping What should I do? |
laskoviymishka
left a comment
laskoviymishka
left a comment
There was a problem hiding this comment.
The two from last round both landed — thanks. publish = true is set now and the crate actually flows into check-public-api, so the snapshot is a real guard instead of the decorative file from round 4. And the task-panic arm aborts: join_next() returning Err(join_err) now calls abort() in both drain_capacity and close(), with test_writer_task_panic_in_write_drain_aborts pinning it.
By what I said last round that'd be a sign-off, but I have to walk one thing back. I credited the percent-encoded key as handled right — raw extraction keeping %2F literal and letting ObjectStorePath::from re-encode to match Java. Looking again, that's not what happens: % is in Path::from's INVALID set, so dt=a%2Fb becomes dt=a%252Fb, and object_store's S3 client encodes once more on the wire. We store and look up dt=a%252Fb where Java and PyIceberg use dt=a%2Fb — so any table written elsewhere with a %XX in a partition value or filename 404s through this backend, and files we write land at a key the manifest doesn't point to. The unit test pins the wrong value and the integration test only round-trips through the same client, so the double-encode cancels and hides it. ObjectStorePath::parse treats the input as already-encoded and fixes it (inline).
The SSE thread I thought I'd closed in round 5 is also still open. The KMS matcher accepts any error stringifying to contain "kms", and the request URL carries the test key (..._sse_kms_default), so a connection refusal or 403 passes without the header ever being sent; and the AES256 test still has the Ok(_) branch a plain unencrypted PUT satisfies. That's the round-1 "can't catch a dropped header" state again — both inline.
Two public-surface things worth settling before this publishes, since it's the base crate the others snapshot against: the two storage enums are exhaustive, so adding GCS/Azure later is a breaking change — #[non_exhaustive] now; and the three delete-batch consts are pub from a backend crate when S3_DELETE_BATCH_SIZE is really a property key that belongs in iceberg::io. Both inline.
The lifecycle work is solid otherwise — the throttle, the clamp, the abort paths are all in good shape. Fix the key encoding and the two SSE tests and I'm back to signing off; the rest are smaller.
| .value() | ||
| .clone(); | ||
|
|
||
| let object_path = ObjectStorePath::from(parsed.relative.as_str()); |
There was a problem hiding this comment.
Build the path with ObjectStorePath::parse(parsed.relative.as_str()) instead of Path::from — from percent-encodes each segment and % is in its INVALID set, so dt=a%2Fb becomes dt=a%252Fb. object_store's S3 client then encodes once more on the wire, so we store and look up dt=a%252Fb where Java and PyIceberg use dt=a%2Fb. Any table written elsewhere with a %XX in a partition value or filename 404s through this backend, and files we write here land at a key the manifest doesn't record — position deletes and delete_prefix miss too.
The unit test at line 682 pins the wrong value (dt=a%252Fb/file.parquet), and the integration test only round-trips through the same client, so the double-encode cancels and the bug stays hidden — worth asserting the raw key via a second client or a bucket listing. parse treats the input as already-encoded, and it also errors on the empty segment in s3://b//x instead of silently collapsing it to x the way from does today.
There was a problem hiding this comment.
leaving this un-resolved till u approve of the updated test
| } | ||
|
|
||
| async fn close(&mut self) -> Result<FileMetadata> { | ||
| let mut upload = self.upload.take().ok_or_else(|| { |
There was a problem hiding this comment.
close() moves upload out of self.upload into a local, so if the close() future is dropped mid-await (a select!/timeout upstream) the local drops with no abort and self.upload is already None — Drop sees nothing and the upload leaks. write() doesn't have this hole because it borrows. I'd keep it in self.upload and borrow with as_mut() until complete() succeeds, or at least document that close() isn't cancel-safe.
There was a problem hiding this comment.
again, keeping these un-resolved for a while
7b6e624 to
e17addd
Compare
|
Thanks for the detailed review @Laskov! All addressed in the latest push:
|
laskoviymishka
left a comment
laskoviymishka
left a comment
There was a problem hiding this comment.
Everything from last round landed — ObjectStorePath::parse replacing Path::from so %2F stays literal, #[non_exhaustive] on both storage enums, S3_DELETE_BATCH_SIZE/S3_MULTIPART_PART_SIZE_BYTES moved into iceberg::io where they belong, and the SSE tests now expect_err. On the key-encoding one I want to be upfront: I credited it fixed in round 6 and then had to walk that back in round 7, so this time I checked the integration test actually lists the key back through a separate client and compares the literal — it does, and it would catch an asymmetric double-encode. The residual is that it's still object_store reading object_store; a real Java/PyIceberg read of the same key is the only true parity check, worth a follow-up.
I'm holding again on two new things, both in the writer/key area this PR keeps circling. First, the multipart abort still races its siblings: on a part failure, a complete() failure, or a task panic we await upload.abort() while up to 7 other put_part tasks are still live in self.tasks, so AbortMultipartUpload can land before the in-flight UploadParts finish and we orphan parts that bill until a lifecycle rule clears them — the same leak-until-expiry we held on for the panic arm in round 6. I'd abort_all() and drain before upload.abort() on all three terminal arms. Second, the ObjectStorePath::parse call that fixed the encoding now rejects // keys, so s3://b/wh//db/tbl/f.parquet returns DataInvalid and tables other engines wrote become unreadable (and delete_stream aborts mid-stream after earlier chunks are already gone). There's a real tension between those two — the fix for one created the other — so I'd settle them together.
Two things worth closing before this publishes to crates.io rather than after: the unsupported-config rejection (role_arn, disable_ec2_metadata, disable_config_load) fires lazily per-bucket on first I/O instead of at Factory::build, so a misconfig surfaces mid-commit — and from_env() always loads env, so disable_config_load=true can't actually be honored today; I'd validate in build and document the env behavior. The other I want to confirm, not assert: with enable_virtual_host_style defaulting true and a custom s3.endpoint set (the MinIO/Ceph/R2/REST-vended case), does object_store prepend the bucket as a subdomain the way Java/opendal do, or use the endpoint as-is and drop the bucket? Every integration test sets path-style=true, so the default path is untested.
Fix the abort drain and the // gap and I think we're basically there — this is close now.
| .value() | ||
| .clone(); | ||
|
|
||
| let object_path = ObjectStorePath::parse(parsed.relative.as_str()).map_err(|e| { |
There was a problem hiding this comment.
This is the same ObjectStorePath::parse that fixed the double-encode, which is good — but it also rejects empty path segments, so s3://b/wh//db/tbl/f.parquet comes back DataInvalid. Java's S3URI and opendal both accept those keys, and they turn up whenever a trailing-slash warehouse path gets joined with /, so tables other engines wrote become unreadable through this backend — and in delete_stream one such path aborts the whole stream after earlier chunks are already deleted.
I'd at least document this as a known parity gap with an error that names it ("object_store cannot represent empty path segments"), and make delete_stream collect-and-continue rather than bail midway. Avoiding Path::parse for the empty-segment case (e.g. Path::from_iter on pre-split parts) would close it properly.
|
|
||
| let part_data = std::mem::take(buffer).freeze(); | ||
| let part_fut = upload.put_part(PutPayload::from_bytes(part_data)); | ||
| tasks.spawn(part_fut); |
There was a problem hiding this comment.
tasks.spawn panics when there's no Tokio runtime, and this is library code reachable from write()/close() under any executor (same for the close() spawn). Drop already guards exactly this with Handle::try_current(), so the two paths disagree. Returning an error — or routing through the same try_current check — keeps it consistent and off the panic path.
| while let Some(res) = self.tasks.join_next().await { | ||
| match res { | ||
| Ok(Err(e)) => { | ||
| let err = Self::abort_and_wrap(upload, e, "part upload failure").await; |
There was a problem hiding this comment.
Before upload.abort() fires in abort_and_wrap, I'd tasks.abort_all() and drain — right now we return on the first failing part while up to 7 other put_part tasks are still live in self.tasks, since they're only cancelled when the JoinSet drops after we've already returned. So AbortMultipartUpload races the in-flight UploadParts, which can land after the abort and leave orphaned parts that bill until a lifecycle rule clears them — the same leak-until-expiry we held on for the panic arm in round 6.
Same drain needs to go on all three terminal arms (this one, the Err(join_err) panic arm just below, and the complete()-failure arm). The abort-exactly-once mock only counts calls, so it's blind to this — a test where a sibling blocks on a Notify and one part fails would pin the ordering.
| } | ||
|
|
||
| /// Writes 12 MiB (3 × 4 MiB chunks) to exercise real S3 multipart uploads | ||
| /// past the 10 MiB `WriteMultipart` buffer threshold, then reads back and |
There was a problem hiding this comment.
This comment still cites the 10 MiB WriteMultipart threshold, but the code uses part_size_bytes and the default is 32 MiB, so writing 12 MiB produces exactly one part at close() — this never exercises a real multi-part upload. Setting s3.multipart.part-size-bytes=5MiB and writing several parts would actually run part numbering/ETag assembly across ≥2 parts, the 5 MiB-min rule, and the 8-part throttle against MinIO. Worth fixing the stale comment in the same pass.
* feat(storage): Add HDFS support via opendal services-hdfs-native Adds an opendal-hdfs-native cargo feature plus OpenDalStorageFactory::Hdfs and OpenDalStorage::Hdfs variants in iceberg-storage-opendal, using OpenDAL's services-hdfs-native (pure-Rust HDFS RPC, no JNI/libhdfs). The NameNode for a path resolves as: the hdfs.name-node property when set (comma-separated endpoints enable HA failover), otherwise the path authority. hadoop.-prefixed properties are forwarded to the HDFS client configuration, overriding values loaded from $HADOOP_CONF_DIR. Operators are cached per effective NameNode since each holds live RPC connections. Revives and updates PR #2441 (by @jordepic) against the current storage layer and opendal 0.58, where name_node became mandatory and the comma list is the HA mechanism. Closes #2440 Co-authored-by: Jordan Epstein <jordepic@users.noreply.github.com> Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * refactor: address review — Properties derive for HdfsConfig, drop HDFS CI special-casing - HdfsConfig is now pub(crate) and parsed via #[derive(Properties)] (key/prefix attributes) instead of hand-written TryFrom + TypedBuilder. - HDFS integration tests are no longer #[ignore]d and the dedicated CI step is gone; they run under the default nextest invocation since make docker-up already starts the fixture and the Tests job is Linux-only. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix: reject non-hierarchical hdfs URLs; key delete_stream batches by effective NameNode - Url::parse accepts non-hierarchical forms like `hdfs:x`; the byte-7 slice then panicked. Require the literal `hdfs://` prefix instead. - batch_key_for_path grouped by URL host only, so NameNodes differing by port shared one deleter; key by the effective NameNode (configured hdfs.name-node, else authority incl. port), matching the operator cache. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * refactor(storage): rename hdfs module to hdfs_native, drop unused HdfsConfig Per review: the opendal module, its functions and the storage/factory variants are now hdfs_native-prefixed (leaving room for a libhdfs-backed variant, see #1130), and the never-consumed HdfsConfig struct is removed from the core crate — config/hdfs.rs keeps only the property constants that iceberg-storage-opendal uses. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * ci: retrigger after setup-python manifest timeout on macos runner Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(storage): address review — cache newtype, build outside lock, single NameNode rule, env-gated HDFS tests - HdfsNativeOperatorCache newtype keeps the cache representation out of the public API. - Operators are built outside the cache lock (the build reads Hadoop XML synchronously); a racing first caller's duplicate is dropped unopened. - hdfs_native_effective_name_node is the single source of the configured-else-authority rule for both create_operator and the delete_stream batch key; unresolvable paths key on themselves. - hdfs.name-node is trimmed and an empty value is treated as unset, so it no longer shadows the path-authority fallback. - HDFS integration tests self-skip unless ICEBERG_TEST_HDFS_ENDPOINT is set (as the HF tests do) and the compose services sit behind the hdfs profile; the Linux CI job opts in via env. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * chore(deps): bump rustls to 0.23.45 (RUSTSEC-2026-0285) Inherited from main (tracked in #3222); same bump as #3165. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * test(storage): pin IPv6 authority handling in HDFS path parsing Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(storage): resolve the HDFS NameNode in relativize_path too An authority-less path now errors in relativize_path exactly as in create_operator unless hdfs.name-node is set, so the two cannot disagree. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * docs(storage): state the cost of the synchronous HDFS operator build Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * docs(storage): describe the HDFS operator cache lifecycle Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(storage): normalize each entry of the hdfs.name-node list A space after a comma was passed through to opendal unchanged, which silently broke failover to that NameNode. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * feat(storage): accept PyIceberg's hdfs.host/hdfs.port for authority-less HDFS paths They map to Hadoop's fs.defaultFS and apply only when the path has no authority, as in PyIceberg; hdfs.name-node stays the explicit override. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(storage): require a port in every hdfs.name-node entry An entry without a port can only fail at the first I/O as 'invalid socket address'; reject it at build time with the property and entry named. Bare host:port stays accepted, as in Hadoop and opendal. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * docs(storage): state the runtime binding of cached HDFS operators Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * test(storage): cover two-NameNode delete_stream batches and first-insert-wins caching Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * test(utils): name the HDFS port constant like its siblings Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(storage): rebuild cached HDFS operators once their runtime is gone Each cache entry parks a sentinel task on the runtime that built the operator and keeps a Weak to a token the task owns. When that runtime shuts down the token dies, the entry reads as stale and the operator is rebuilt on the current runtime instead of hdfs-native panicking on the dead one. Sentinels are aborted when entries drop. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(storage): resolve HDFS NameNodes like Hadoop does An authority with a port is used as is; hdfs.name-node serves logical nameservice authorities and authority-less paths, which otherwise fall back to fs.defaultFS. Portless strings from any source are logical names that only hdfs.name-node resolves, reported at resolution instead of at the first I/O. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * refactor(storage): keep the HDFS operator cache out of the public API OpenDalStorage::HdfsNative now wraps one opaque HdfsNativeStorage with crate-private fields, built only by the factories. The serialized form keeps its nesting; the io-timeout propagation test now covers the HDFS factory too. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * perf(storage): build HDFS operators on a blocking thread The build reads the Hadoop XML config synchronously; create_operator is now async and offloads it with spawn_blocking, outside the cache lock as before. The builder itself stays sync and runtime-free. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * ci: run the HDFS integration tests in their own job The default suite no longer pulls the Hadoop image or waits on the HDFS health checks; a dedicated required job starts only the two HDFS containers and runs the HDFS test binary. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(storage): stop dropping HDFS properties silently PyIceberg's hdfs.user and hdfs.kerberos_ticket log a warning instead of vanishing (opendal's config has no user field), hdfs.port is validated whether or not a host is set and warns when it has no host to apply to, and a bare hadoop. key is rejected. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * test(storage): derive the second NameNode spelling from the endpoint host Resolving the host to its IPv4 address gives any hostname endpoint a second batch key; an IP-literal endpoint skips the test explicitly instead of passing as a no-op. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * chore: record tracing as a dependency of iceberg-storage-opendal in Cargo.lock Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(storage): report a missing tokio runtime instead of panicking in HDFS create_operator The runtime handle is resolved up front and passed into the cache, so the sentinel is no longer optional and the unreachable no-runtime branch is gone. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(storage): normalize every HDFS NameNode spelling to hdfs://host:port One parser classifies configured entries, path authorities and fs.defaultFS: other schemes, logical names, port 0, userinfo, paths and unbracketed IPv6 are rejected, and bare and prefixed spellings share one cache key. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * feat(storage): declare HDFS nameservices with hdfs.name-node.<nameservice> The key expands to Hadoop's dfs.ha.namenodes and dfs.namenode.rpc-address keys in the forwarded options and the resolver reads them back, so nameservices declared through hadoop.* work the same way. Plain hdfs.name-node stays the documented single-cluster default; an undeclared name without it is a pointed error. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(storage): require a declaration for every logical HDFS nameservice Plain hdfs.name-node serves authority-less paths only; an undeclared logical authority is an error naming the key to set instead of a guess that could reach the wrong cluster. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(storage): reject userinfo and port 0 in HDFS path authorities Userinfo was silently dropped and port 0 was misread as a logical nameservice; both are DataInvalid now, and the userinfo message names the host only so a password never reaches a log. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(storage): validate the NameNode composed from hdfs.host and hdfs.port Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * test(storage): use an authority-less path in the cross-runtime HDFS test A logical authority now requires a declaration, so the test must take the plain-key route. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * test(storage): tighten the HDFS tests after review Drop three redundant unit tests, assert the warnings and the password redaction directly, make the serde round trip populate the cache first, check which error each negative case produces, cover the plain key and hdfs.host serving authority-less paths against the fixture, and name the declared-nameservice test for what it does. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix(storage): mask userinfo in HDFS errors that echo a path or NameNode A path that failed to parse, or had another scheme, was echoed whole, so `hdfs://user:pw@nn:bad/x` put the password in the error. The NameNode config errors (`hdfs.name-node`, `hdfs.name-node.<nameservice>`, `hdfs.host`/`hdfs.port`, `fs.defaultFS`) echoed their values the same way. Every `Invalid hdfs path` error now goes through one helper, and the config errors use the same masking: everything between the scheme and the last `@` becomes `***`, so a password holding a raw `/`, `?`, `#` or `://` is covered too. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(storage): recover a poisoned HDFS operator cache lock A panic elsewhere while the cache lock was held turned every later HDFS call into a permanent `Unexpected` error. Writes only ever replace whole entries, so the map stays consistent: both lock sites now recover the guard with `PoisonError::into_inner`, and `get`/`insert` are infallible. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(storage): flush the fake NameNode before sampling dials The runtime test sampled its dial counter before the listener thread had necessarily counted every dial from the first runtime. It now drops that runtime and waits for the listener to close a marker connection; accepts are served in order, so every earlier dial is counted first. A failed accept no longer stops the listener. The runtime-shutdown unit test also asserts that the rebuilt operator is a new instance, not the stale one with a fresh sentinel. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * docs(storage): state where portless HDFS authorities differ from Hadoop Hadoop resolves a portless authority from hdfs-site.xml, else dials port 8020. Here it must be declared with `hdfs.name-node.<nameservice>`, because opendal builds the client as `hdfs://nameservice` and never hands hdfs-native the authority. The `HDFS_NAME_NODE` and resolver docs no longer claim Hadoop parity, the error says to add a port or declare the nameservice, and the README notes the limitation in its own note. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(storage): rename the HDFS config module to hdfs_native Matches its siblings, which are named after their storage feature (azdls, gcs, hf, oss, s3). The module is private and re-exported, so the public iceberg::io::HDFS_* paths are unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(storage): move the HDFS tests out of lib.rs into hdfs_native The serde round-trip and relativize_path tests now live next to the code they cover, with the module's test_hdfs_native_ prefix. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(storage): read HDFS properties through #[derive(Properties)] hdfs.name-node, the hdfs.name-node.<nameservice> and hadoop. prefixes, hdfs.host and hdfs.port are declared in one struct parsed by the Properties macro instead of being looked up by hand. Host and port stay trimmed strings so a blank value still counts as unset, as in PyIceberg. Validation, warnings and error messages are unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * ci: run the HDFS tests in the default suite like the other fixtures The HDFS fixture joins the shared bridge network with published ports instead of using host networking behind a compose profile, so make docker-up starts it on every platform: the DataNode advertises 127.0.0.1 and the tests set dfs.client.use.datanode.hostname. The tests no longer skip without ICEBERG_TEST_HDFS_ENDPOINT, and the dedicated Tests (hdfs) job is gone. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(storage): build HDFS clients in a crate-owned runtime An hdfs-native client spawns its NameNode connections on the runtime current when it is built and panics once that runtime is dropped. The cache used to detect that with a sentinel task per entry, which needed Handle::try_current, and the build ran under spawn_blocking, which made create_operator async for every backend. Clients are now built inside a crate-owned runtime started on first use, so a cached operator works from any caller's runtime: the sentinel, Handle::try_current and spawn_blocking are gone, and create_operator is sync again as on main. The Hadoop XML read (~0.1 ms) runs inline. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(storage): type HDFS path authorities and validate declarations at parse hdfs_native_parse_path returns an HdfsNativeAuthority (NameNode, Nameservice, Default), so the resolver matches on it instead of re-parsing the authority to learn whether it has a port. Every HA nameservice declaration, from hdfs.name-node.<nameservice> or passed through as hadoop.dfs.ha.namenodes.*, and the fs.defaultFS that serves authority-less paths are now validated when the config is parsed, so a broken one fails the first call rather than only the paths that use it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(storage): drop index slicing and combinator chains in the HDFS backend The nameservice lookup is a for loop with let-else, and the userinfo masking and the relative-path extraction use split_once/rsplit_once instead of slicing at computed indices. A test pins that doubled slashes after the authority are still trimmed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(storage): attach the hdfs.port parse error as the source The parse error was interpolated into the message; it is now attached as the error's source, which iceberg prints after the message. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(storage): share one HDFS client across host name spellings host_str keeps the host's case for hdfs:// URLs, so hdfs://NN1:8020 and hdfs://nn1:8020 cached two clients for one NameNode. Endpoint hosts are now lowercased by one helper used by the config normalizer and the path parser, as Hadoop's FileSystem cache key does; nameservice names stay case-sensitive like the Hadoop keys that declare them. The cache's doc states that it is unbounded by design. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * docs(storage): note the upstream gap behind the HDFS client runtime opendal's hdfs-native service does not expose hdfs-native's ClientBuilder::with_io_runtime (Kimahriman/hdfs-native#255); a TODO at the crate-owned runtime records that a runtime could be passed in once it does. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(storage): make the HDFS HA test fail over for real The declared-nameservice test listed the single-node fixture twice, which drove opendal's list path but never needed a failover. Its first NameNode is now a local port nothing listens on, so every call has to fail over through opendal's synthetic nameservice; an opendal change that kept only the first entry of the list would fail it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore: retrigger CI Docker Hub rate-limited the previous run's image pulls, so the default test job never started its fixtures. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore: retrigger CI The previous run hit Docker Hub's unauthenticated pull rate limit again before any test ran. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * docs(storage): document how to enable and configure the HDFS backend The iceberg-storage-opendal README gains an HDFS section: enabling the opendal-hdfs-native feature, the factory, the hdfs.* and hadoop.* properties, the Hadoop config, user and Kerberos it reads, and the portless-authority limitation. The experimental-backends note now points to it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * docs(storage): drop the HDFS README section and link follow-up issues Configuration docs for every backend are tracked in #3510; the README keeps the one-line note on the portless-authority limitation. The TODO on the crate-owned HDFS runtime now points at #3509. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Jordan Epstein <jordepic@users.noreply.github.com> Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
…, and configure 5MiB test parts
|
b5ac18a to
d197881
Compare
|
@laskoviymishka hey, haven't heard from ya? Apologies for publish = true part |
d197881 to
08a6e4f
Compare
|
@Sruhvx-jpg i'll take a look next week, can you ensure CI is green here? |
|
@laskoviymishka ok, but the Public API is going to red, right? |
…ustive attributes
|
@laskoviymishka Updated the public-api snapshot with |
|
Related: #3605 makes the S3 retry policy configurable, and it would be good for this backend to use the same settings. Why: S3, S3 Tables included, answers What #3605 does: it adds Iceberg Java's existing S3FileIO keys to
The opendal S3 operator applies them to its Suggestion: if #3605 merges first, this PR could rebase on it and read the same keys through
With none set, |
|
@emanuilov Ya it would be great if ur pr lands before mine, than i just have to rebase. Also, why didnt u add a jitter ; just curious. Thanks for the coordination 😄 hey @laskoviymishka what do you think? |
|
@Sruhvx-jpg |
Which issue does this PR close?
What changes are included in this PR?
Implements
ObjectStoreStorage::S3backed by Apache Arrow'sobject_storecrate (0.13), integrated into Iceberg'sStorageandStorageFactoryframework:ErrorKind::FeatureUnsupported) on propertiesobject_storecannot express (role_arn,disable_ec2_metadata,disable_config_load).Urlparser supportings3://,s3a://, ands3n://schemes, case-insensitivity, percent-decoded bucket names, and preserved consecutive path slashes.Box<dyn MultipartUpload>lifecycle with explicit part-failure and complete-failure aborts.Dropsafety net aborting orphaned in-progress multipart uploads.delete_streamanddelete_prefixvias3.delete.batch-size(default: 250, clamped to 1..1000) leveraging S3's bulkDeleteObjects.S3Storage.ErrorKind::Unexpectedmatching OpenDAL and Iceberg conventions.Are these changes tested?
MockMultipartUploadspies for drop/failure aborts and in-flight concurrency caps.crates/storage/object_store/tests/file_io_s3_test.rs) verifying real reads, writes, multipart uploads across thresholds (>12 MiB), range reads, SSE rejection on KES-less setups, and bulk prefix deletions.AI Assistance Disclosure:
AI tooling was used to assist in scaffolding and expanding unit/integration test matrices, error-handling boilerplate, and exploring edge cases, with all logic, types, and implementations manually reviewed, verified, and tested against local S3/MinIO.
EDIT / Iteration Log:
Urlscheme/percent-decoding parsing, privateS3Storageencapsulation,Dropabort safety net, and MinIO integration test harness.aws:kmswith no key ID header), percent-decoded bucket cache keys, S3 bulkdelete_prefixviadelete_stream, SSE-C MD5 documentation, and multi-part integration tests.WriteMultipartwith directBox<dyn MultipartUpload>management for complete/part failure aborts, spy-basedMockMultipartUploadunit tests, OpenDAL-alignedErrorKind::Unexpectedmappings, and preserved double-slash path segments.s3.delete.batch-size(default: 250, clamped to 1..1000), privatedelete_batch_sizevisibility, and refactoredabort_and_wraperror helper.