Feature/iceberg gcs reader - #1392
Conversation
…direct mode - direct catalog: resolve a bare-integer `version-hint.text` (e.g. `42`) to `v42.metadata.json` — the Iceberg Hadoop file-based catalog convention — in addition to the existing filename / full-path forms. - S3 reader: `parse_s3_uri` accepts `gs://` (alongside `s3://`/`s3a://`) so Iceberg metadata/manifests for GCS-backed tables resolve against the configured GCS S3-interoperability endpoint. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Reading an Iceberg table on GCS via the AWS S3 SDK pointed at the GCS S3-interop endpoint fails on HTTP range GETs over HTTP/2 (smithy-rs rejects the partial-content body); the Parquet reader is range-based, so data reads fail although metadata reads work. - io::gcs: add `GcsXmlStorage`, a native GCS object reader (read / read_range / file_size) over the Cloud Storage XML API using reqwest (HTTP/1.1), authenticated with a GCP OAuth2 bearer token from the `FLUREE_GCS_OAUTH_TOKEN` environment variable. - io::gcs: add an `IcebergBackend` enum dispatching the storage trait to the S3 SDK or the native GCS reader; the generic Parquet/scan/direct readers work over either backend unchanged. - r2rml: select the GCS reader when the configured endpoint is storage.googleapis.com (both Direct and Rest construction sites). - parquet: enable the `snap` (Snappy) codec (zstd was already enabled). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- graph-sources/iceberg.md: add a "Google Cloud Storage (GCS)" section (native reader + rationale, FLUREE_GCS_OAUTH_TOKEN auth, gs:// / integer version-hint / Snappy handled automatically) and note the bare-integer version-hint form. - cli/iceberg.md: document storage.googleapis.com for --s3-endpoint and add a GCS direct-mode example. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
ff13f04 to
a71bfa5
Compare
aaj3f
left a comment
There was a problem hiding this comment.
Hi @christophediprima 👋 thanks for this PR, it certainly identifies a gap we'd like to resolve and we're excited by the contribution!
The PR makes me realize that we likely owe you and other contributors a top-level contributions doc, that foregrounds some design principles we're generally trying to maintain. Which is to say, I think your PR here is generally correct, and I think the team is likely to accept it with a few changes, but there are some other source code conventions that, if adopted by your solution, may also simplify the solution coverage slightly.
That is, I'd be happy to approach this PR in one of two ways: (1) a slight pivot in the solution design that might simplify the pattern a bit and align it closer to other repo-wide conventions, or (2) simply some requested changes to your current solution design as-is.
I'm happy to accept either set of changes from you though, so I'll defer to you as a contributor.
Design Pivot
If you look at S3IcebergStorage and fluree-db-storage-aws, our typical pattern for something like "an object store that speaks S3" is to maintain the same code paths but vary on configuration like endpoint_url or transport. I think GCS interop falls into that case as well. You've correctly identified the bug with h2 range-GET, but we may be able to solve for that as simply a transport configuration override while keeping the AWS SDK paths generally intact.
If you're up for it, I'd recommend investigating a solution where we pin the existing aws-sdk-s3 client to HTTP/1.1 with a custom .http_client(...) call (aws-smithy-http-client is already a dependency for this). That would keep one storage path and would inherit the SDK's correct S3 signing, key encoding, credential refresh, retries, and our disk cache, without having to independently solve for a GCS-custom path capable of doing all of those. We'd likely still need your direct.rs integer hint fix as well as the parquet snap feature, but it would shrink the PR diff to just those 3 updates.
There's still a possibility the GCS story fails for a separate reason and, at the end of the day, perhaps a parallel reader path would in fact be necessary, but this would be the outcome we'd at least reach for first, were we to solve for this PR's objective as an internal team
Requested Edits As-Is
As mentioned above, very happy to take your design solution as-is. Internally, if the team feels strongly, we can ourselves independently investigate the tighter design but welcome your fix regardless. You'll see file-specific suggestions/requests below, if you'd like to leave the PR solution design as-is
| let mut settings = SigningSettings::default(); | ||
| // S3-style services expect the payload hash header on signed requests. | ||
| settings.payload_checksum_kind = PayloadChecksumKind::XAmzSha256; |
There was a problem hiding this comment.
I noted your PR e2e tests pass, but when I tested against partitioned tables, I failed to read every partitioned table's data files. I've traced most of that back to here.
SigningSettings::default() is PercentEncodingMode::Double + UriPathNormalizationMode::Enabled (cf. aws-sigv4-1.4.2/src/http_request/settings.rs:135,140).
The code overrides only the payload checksum, so Double stays in effect. In Double mode the signer runs percent_encode_path() over the path, whose char set encodes = as %3D. But the wire URL (object_url, then http::Request::builder().uri(url) at gcs.rs:225) carries the raw =.
S3-compatible services (GCS interop included) canonicalize the path single / as-received, so:
- client canonical path:
…/event_date%3D2024-01-01/… - server canonical path:
…/event_date=2024-01-01/…→ signatures differ → 403
Iceberg/Hive partition directories are field=value, so this inconsistency means every partitioned data file read fails. The AWS SDK itself avoids this by setting Single & Disabled for all S3 calls.
| /// Build the path-style XML-API URL for an object. | ||
| fn object_url(&self, path: &str) -> Result<String> { | ||
| let (bucket, key) = Self::split_uri(path)?; | ||
| Ok(format!("{}/{}/{}", self.endpoint, bucket, key)) | ||
| } |
There was a problem hiding this comment.
Same root cause as the above comment. Unless object_url generation mirrors the SDK's "encode the key, then sign Single" strategy, then this opens up failure conditions if partitions are on values with non-ASCII chars or on values with spaces, e.g. city=São Paulo
| pub mod gcs; | ||
| pub mod parquet; | ||
| #[cfg(feature = "aws")] | ||
| pub mod send_parquet; | ||
| pub mod storage; | ||
|
|
||
| pub use batch::{BatchSchema, Column, ColumnBatch, FieldInfo, FieldType}; | ||
| pub use gcs::is_gcs_endpoint; |
There was a problem hiding this comment.
These should get gated on aws feature, e.g.
| pub mod gcs; | |
| pub mod parquet; | |
| #[cfg(feature = "aws")] | |
| pub mod send_parquet; | |
| pub mod storage; | |
| pub use batch::{BatchSchema, Column, ColumnBatch, FieldInfo, FieldType}; | |
| pub use gcs::is_gcs_endpoint; | |
| #[cfg(feature = "aws")] | |
| pub mod gcs; | |
| pub mod parquet; | |
| #[cfg(feature = "aws")] | |
| pub mod send_parquet; | |
| pub mod storage; | |
| pub use batch::{BatchSchema, Column, ColumnBatch, FieldInfo, FieldType}; | |
| #[cfg(feature = "aws")] | |
| pub use gcs::is_gcs_endpoint; |
There was a problem hiding this comment.
I'm being a bit redundant as I've mentioned this at the top of the review, but as I now see the value of this file (and to be clear, it does solve for the h2 range-read bug), I do think that, if the solution could be handled by forcing HTTP/1.1 on the AWS SDK client e.g. with aws-smith-http-client and a custom .http_client(...), then it would dodge the bug and avoid the 447-line parallel reader established by this file
| /// Build a SigV4-signed `reqwest::Request` for `method url` with an optional | ||
| /// `Range` header. | ||
| /// | ||
| /// The signature is computed over the canonical request (host + `x-amz-date` + | ||
| /// `x-amz-content-sha256` are auto-included by the signer); the `Range` header | ||
| /// is sent unsigned, which S3-style services accept. | ||
| fn signed_request( | ||
| &self, | ||
| method: &str, | ||
| url: &str, | ||
| range: Option<&str>, | ||
| ) -> Result<reqwest::Request> { |
There was a problem hiding this comment.
In the AWS SDK, credentials refresh via the provider, but signed_request clones the stored Credentials permanently. For either REST-vended or STS temp credentials, we'd likely see even a single scan that outlives the TTL and starts returning a 403 w/ no path for re-fetch or retry. I see the expires_at field but it seems inert / not read by the signer, etc
| /// Supports formats: | ||
| /// - `s3://bucket/key/path` | ||
| /// - `s3a://bucket/key/path` (Hadoop style) | ||
| /// - `gs://bucket/key/path` (Google Cloud Storage, read via the GCS | ||
| /// S3-interoperability endpoint). Iceberg metadata/manifests for GCS-backed | ||
| /// tables reference `gs://` paths; when the storage endpoint is | ||
| /// set to `storage.googleapis.com`, `gs://bucket/key` and `s3://bucket/key` | ||
| /// address the same object, so the scheme is accepted and resolved against | ||
| /// the configured endpoint. | ||
| pub fn parse_s3_uri(path: &str) -> Result<(&str, &str)> { |
There was a problem hiding this comment.
In fluree-db-iceberg/src/io/gcs.rs:160-173, you also introduce split_uri, used by GcsXmlStorage. Because S3IcebergStorage is never built for a GCS endpoint, the additions to split_uri to accommodate GCS endpoints are effectively inert. Some of your changes to parse_s3_uri are nice though, e.g. the empty-bucket guard (which isn't in-place for the duplicate split_uri.
If we keep gcs.rs, we should likely either delete the gs:// branch from parse_s3_uri or make parse_s3_uri pub and have GcsXmlStorage::split_uri delegate to it
| let resolved_region = sdk_config | ||
| .region() | ||
| .map(|r| r.as_ref().to_string()) | ||
| .unwrap_or_else(|| "auto".to_string()); |
There was a problem hiding this comment.
Not as confident about this, but I don't believe all GCS S3-compatible backends support auto, given that region is a part of credential scope. If true, then this opens up 403s for possibly opaque reasons. Should possibly be elevated to an error or at least a warning if region is unset
…ative reader Pivot per PR fluree#1392 review: drop the bespoke native GCS reader (gcs.rs) and read GCS-backed Iceberg tables through the existing aws-sdk-s3 client with the transport pinned to HTTP/1.1. This inherits the SDK's SigV4 signing (fixing the `%3D` partition-path signature mismatch and non-ASCII/space partition values), credential refresh, retries, and disk cache. - storage.rs: build the S3 client with a hyper-rustls connector that offers only http/1.1 in ALPN (via aws-smithy-http-client's hyper-014 builder), sidestepping the smithy-rs HTTP/2 range-read bug against the GCS S3-interop endpoint; set response checksum validation to WhenRequired (an object-level checksum cannot validate a partial range body); walk the error source() chain for diagnostics. - r2rml.rs: route GCS-backed tables through S3IcebergStorage; merge_vended_with_io folds graph-source connection params into vended credentials as a fallback. - Remove gcs.rs, GcsXmlStorage, IcebergBackend, is_gcs_endpoint; drop the now-unused aws-sigv4 and http deps; add aws-smithy-http-client + hyper-rustls. - Retain the direct-mode integer version-hint / gs:// resolution and Snappy codec. - Add an ignored, env-gated live-GCS integration test (it_gcs_sdk_reads). - Docs: rewrite the GCS section for the SDK / HTTP-1.1 approach. Verified end-to-end against a real GCS-backed table (JSON-LD + SPARQL both return rows) and via it_gcs_sdk_reads (range GET succeeds over HTTP/1.1; a missing `field=value/` key returns NoSuchKey, not SignatureDoesNotMatch). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Reconcile the GCS HTTP/1.1 pivot with upstream's vended-credential work (fluree#1396): adopt upstream's S3IcebergStorage::from_vended_credentials(creds, region, endpoint, path_style) + resolve_io precedence (incl. the deliberate path-style OR) and drop the local merge_vended_with_io helper it supersedes; keep the HTTP/1.1-pinned http_client, WhenRequired response-checksum validation, and error_chain diagnostics on both constructors. Also picks up upstream's r2rml media-type/table-count and CLI graph-source query routing. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
Thanks @aaj3f — I took the design-pivot path. Mechanics: This resolves your review points directly:
One extra fix the SDK path surfaced: aws-sdk-s3's default response-checksum validation Retained from the original PR: the Hadoop integer Verified end-to-end against a real GCS-backed Iceberg table (JSON-LD + SPARQL both return rows). Merged latest
|
b10d15c to
0f6fea0
Compare
Iceberg data files produced by the Apache Iceberg Kafka Connect connector use GZIP-compressed Parquet; without the parquet crate's flate2 feature every such data-file read fails with "Disabled feature at compile time: flate2". Enable it alongside zstd + snap so GZIP-compressed Iceberg data files decode. Verified end-to-end reading GCS-backed Iceberg tables over the S3-interop endpoint. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
0f6fea0 to
ccb58f0
Compare
|
While validating this reader against real GCS-backed Iceberg tables, every populated table failed at Cause: the data files were written by the Apache Iceberg Kafka Connect connector, which Fix (one line): add This is orthogonal to the HTTP/1.1 pivot but part of "actually read real GCS Iceberg tables," so it |
…eader # Conflicts: # fluree-db-api/src/graph_source/r2rml.rs
Addressed by the HTTP/1.1-pin design pivot: native GCS reader removed, GCS now read via the existing S3 SDK path (HTTP/1.1 pinned). Validated live against Snowflake-managed Iceberg over AWS S3.
bplatz
left a comment
There was a problem hiding this comment.
Design pivot addresses the review — native GCS reader dropped in favor of the existing S3 SDK path pinned to HTTP/1.1 (single storage path, SDK signing/creds/retries retained). Merged latest main and resolved the r2rml.rs conflict; validated end-to-end with a live read against Snowflake-managed Iceberg over AWS S3 (vended credentials, range reads clean — no 403/checksum/HTTP2 issues). Residual: region-unset warning is a minor follow-up.
Summary
Adds support for reading Iceberg graph sources stored in Google Cloud Storage. Previously,
registering such a source worked but querying it failed at the data-read stage.
Three commits:
version-hint+gs://paths in direct mode.Motivation
Reading an Iceberg table stored on GCS (written by Spark, Flink, or any Iceberg engine) hit
several gaps — all generic to Iceberg-on-GCS, none vendor-specific:
version-hint.textinteger form. The Iceberg Hadoop file-based catalog writes a bareinteger version
Nwhose metadata file isvN.metadata.json. Direct mode only treated thehint as a literal filename, so it resolved the wrong path and the load failed.
gs://paths. Iceberg metadata/manifests for GCS-backed tables referencegs://paths;the S3 reader only accepted
s3:///s3a://.Rust SDK pointed at the GCS S3-interop endpoint fails on range GETs over HTTP/2: smithy-rs
rejects the partial response (
"streaming error") while full GETs succeed. The Parquet readeris range-based, so data reads fail although metadata/manifest reads work. (Range reads are an
HTTP/1.1 feature and work fine once the client is pinned to 1.1 — the native reader fetches
only the footer and the column chunks a query needs, never the whole object.)
parquetcratewas built without the
snapfeature (Disabled feature at compile time: snap).Changes
fluree-db-iceberg— direct catalog (catalog/direct.rs): resolve a bare-integerversion-hint.texttov{N}.metadata.json(the Iceberg Hadoop file-based catalog convention),in addition to the existing filename / full-path forms. + unit test.
fluree-db-iceberg— S3 reader (io/storage.rs):parse_s3_uriacceptsgs://(alongsides3:///s3a://), resolved against the configured GCS S3-interop endpoint. + test.fluree-db-iceberg— native GCS reader (io/gcs.rs, new):GcsXmlStoragereads GCS objects(
read/read_range/file_size) over the Cloud Storage XML API usingreqwestpinned toHTTP/1.1 (
.http1_only(), with connect/request timeouts), so the h2 range issue cannot occur.Requests are signed with AWS SigV4 using GCS HMAC interop keys — the same credentials the
S3-interop path uses, resolved from the standard AWS credential chain (or catalog-vended
credentials in REST mode). A range GET is validated to return
206 Partial Content(a200full-object response is rejected rather than silently fed to the range-based Parquet reader).
Adds an
IcebergBackendenum that dispatches the storage trait to the AWS SDK or the native GCSreader, so the generic Parquet/scan/direct readers work over either backend unchanged. + unit tests.
fluree-db-api(graph_source/r2rml.rs): select the native GCS reader when the configuredendpoint is
storage.googleapis.com(both Direct and Rest construction sites); otherwise theAWS S3 SDK is used exactly as before. In REST mode, catalog-vended credentials for a GCS-backed
table are honored.
Cargo.toml): addaws-sigv4(signs the GCS reader's requests) and enable theparquetsnap(Snappy) codec (zstd was already on).docs/graph-sources/iceberg.md(nativereader + rationale, SigV4/HMAC-key auth, and the conventions handled automatically:
gs://paths, integerversion-hint, Snappy) + the bare-integerversion-hintform;storage.googleapis.comdocumented for--s3-endpointindocs/cli/iceberg.mdwith an example.Selection is endpoint-gated: nothing changes for S3 / non-GCS endpoints.
Auth
The GCS reader signs each request with AWS SigV4 using GCS HMAC interoperability keys —
the same
AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEYthe S3-interop path already uses, resolvedfrom the standard AWS credential chain (and from catalog-vended credentials in REST mode). No new
credential type, no new env var, and HMAC keys do not expire (no token-refresh handling).
s3_regionis the SigV4 signing region (the bucket location).
Testing
cargo test -p fluree-db-iceberg --features aws): URI parsing (gs:///s3://),path-style URL building, GCS-endpoint detection, integer
version-hintresolution, and that theGCS reader's request signing produces a valid
AWS4-HMAC-SHA256Authorization header (withx-amz-date/x-amz-content-sha256, and theRangeheader carried through).cargo fmt/cargo clippy --all-targets -- -D warnings: clean on the touched crates.directmode withs3_endpoint: https://storage.googleapis.com: the native reader signs and readsversion-hint.text, the metadata JSON, the Avro manifest list + manifest, and the Snappy Parquetdata file — the Iceberg scan completes and returns the table's rows (
Iceberg scan complete total_rows=Nin the logs). The signing approach is the same SigV4-over-the-XML-API the AWS SDKalready uses for GCS metadata reads. (Surfacing those rows through
POST /query/<alias>additionally requires the graph-source alias query-routing fix from v4.1.0: querying a registered Iceberg/R2RML graph source by alias fails with
Nameservice error: Serialization error: missing field \f:ledger\`—NameService::lookupdeserializes graph-source records as ledgerNsFileV2` #1369/fix: query a graph source by alias instead of loading it as a ledger #1375 — now mergedinto
main, so this branch already sits on top of it — which is orthogonal to this reader change.)Known limitations / follow-ups
upstream report; this PR routes around it for GCS rather than fixing the SDK.
s3_endpointcontainsstorage.googleapis.com); thegs://scheme in metadata paths is accepted but does not by itself select the GCS reader.
Breaking changes
None. Additive; the native GCS reader is selected only when the endpoint is
storage.googleapis.com, and the AWS S3 path is unchanged for all other endpoints.