laskoviymishka commented on code in PR #1280:
URL: https://github.com/apache/iceberg-go/pull/1280#discussion_r3467400129
##########
catalog/hadoop/hadoop.go:
##########
@@ -135,35 +135,45 @@ func NewCatalog(name, warehouse string, props
iceberg.Properties) (*Catalog, err
if !isLocal && !allowUnsafeCommits {
return nil, fmt.Errorf("hadoop catalog: when using warehouse
scheme %q, `allow-unsafe-commits` must be set to true", u.Scheme)
- }
+ } else if isLocal {
+ if u.Opaque != "" {
+ warehouse = u.Opaque
+ } else {
+ warehouse = u.Path
+ }
- if u.Opaque != "" {
- warehouse = u.Opaque
- } else {
- warehouse = u.Path
- }
+ if warehouse == "" || warehouse == "/" {
+ return nil, errors.New("hadoop catalog: local
filesystem requires a non-root warehouse path")
+ }
- if warehouse == "" || warehouse == "/" {
- return nil, errors.New("hadoop catalog requires a non-root
warehouse path")
- }
+ warehouse = strings.TrimRight(warehouse, "/")
- warehouse = strings.TrimRight(warehouse, "/")
+ // Normalize to absolute path so the synthetic "location"
property
+ // always produces a valid file:// URI.
+ absWarehouse, err := filepath.Abs(warehouse)
+ if err != nil {
+ return nil, fmt.Errorf("hadoop catalog: failed to
resolve absolute warehouse path: %w", err)
+ }
- // Normalize to absolute path so the synthetic "location" property
- // always produces a valid file:// URI.
- absWarehouse, err := filepath.Abs(warehouse)
+ warehouse = absWarehouse
+ }
+
+ filesystem, err := io.LoadFS(context.Background(), nil, warehouse)
Review Comment:
We're passing `nil` here, but the catalog's `props` are in scope and the
cloud factories (`createS3Bucket`, `createGCSBucket`, `createAzureBucket`) read
region, access key, secret, endpoint, etc. straight off the props map. With
`nil` every lookup returns zero-value, so the FS falls back entirely to ambient
SDK credentials and silently ignores anything the user set via
`s3.access-key-id` / `s3.region` / `s3.endpoint`.
It's also inconsistent with the table path — `io.LoadFSFunc(c.props,
metaPath)` a few lines down passes the real props, so the catalog FS and table
FS end up configured differently.
`iceberg.Properties` is `map[string]string`, so this is just
`io.LoadFS(context.Background(), props, warehouse)`. wdyt?
##########
catalog/hadoop/hadoop.go:
##########
@@ -135,35 +135,45 @@ func NewCatalog(name, warehouse string, props
iceberg.Properties) (*Catalog, err
if !isLocal && !allowUnsafeCommits {
return nil, fmt.Errorf("hadoop catalog: when using warehouse
scheme %q, `allow-unsafe-commits` must be set to true", u.Scheme)
- }
+ } else if isLocal {
Review Comment:
There's a subtle issue with the third branch that this control flow hides.
When `isLocal` is false and `allowUnsafeCommits` is true, we fall through with
`warehouse` left as the raw URI, e.g. `s3://bucket/wh`.
Every path-building method — `namespaceToPath`, `tableToPath`,
`metadataDir`, `metadataFilePath`, `versionHintPath`, `defaultTableLocation` —
then calls `filepath.Join(c.warehouse, ...)`, and `filepath.Join` runs
`filepath.Clean`, which collapses `s3://bucket` to `s3:/bucket`. So the
`location` we write into table metadata comes out as `s3:/bucket/wh/...` — a
single slash after the scheme — which Java, PyIceberg, and iceberg-rust all
reject when they parse it as a URI.
I'd make the join scheme-aware: `filepath.Join` for local, and `path.Join`
(or `strings.TrimRight(base, "/") + "/" + ...`) preserving the
`scheme://authority` prefix for remote. Probably worth a single `joinPath`
helper that branches on `isLocal`.
Also, while we're here — the `else if` after an unconditional `return` isn't
idiomatic; dropping the `else` and using a plain `if isLocal {` would make the
non-local fallthrough visible rather than implied.
##########
catalog/hadoop/hadoop.go:
##########
@@ -135,35 +135,45 @@ func NewCatalog(name, warehouse string, props
iceberg.Properties) (*Catalog, err
if !isLocal && !allowUnsafeCommits {
return nil, fmt.Errorf("hadoop catalog: when using warehouse
scheme %q, `allow-unsafe-commits` must be set to true", u.Scheme)
- }
+ } else if isLocal {
+ if u.Opaque != "" {
+ warehouse = u.Opaque
+ } else {
+ warehouse = u.Path
+ }
- if u.Opaque != "" {
- warehouse = u.Opaque
- } else {
- warehouse = u.Path
- }
+ if warehouse == "" || warehouse == "/" {
+ return nil, errors.New("hadoop catalog: local
filesystem requires a non-root warehouse path")
+ }
- if warehouse == "" || warehouse == "/" {
- return nil, errors.New("hadoop catalog requires a non-root
warehouse path")
- }
+ warehouse = strings.TrimRight(warehouse, "/")
- warehouse = strings.TrimRight(warehouse, "/")
+ // Normalize to absolute path so the synthetic "location"
property
+ // always produces a valid file:// URI.
+ absWarehouse, err := filepath.Abs(warehouse)
+ if err != nil {
+ return nil, fmt.Errorf("hadoop catalog: failed to
resolve absolute warehouse path: %w", err)
+ }
- // Normalize to absolute path so the synthetic "location" property
- // always produces a valid file:// URI.
- absWarehouse, err := filepath.Abs(warehouse)
+ warehouse = absWarehouse
+ }
+
+ filesystem, err := io.LoadFS(context.Background(), nil, warehouse)
Review Comment:
`context.Background()` is the only option here since `NewCatalog` has no
`ctx` param, which is fine and out of scope to change. But the S3 factory does
use the context for `config.LoadDefaultConfig` and the bucket handle, so it'd
be good to leave a `// TODO: propagate caller context once NewCatalog accepts
one` here as a marker for the follow-up.
##########
catalog/hadoop/hadoop.go:
##########
@@ -135,35 +135,45 @@ func NewCatalog(name, warehouse string, props
iceberg.Properties) (*Catalog, err
if !isLocal && !allowUnsafeCommits {
return nil, fmt.Errorf("hadoop catalog: when using warehouse
scheme %q, `allow-unsafe-commits` must be set to true", u.Scheme)
- }
+ } else if isLocal {
+ if u.Opaque != "" {
+ warehouse = u.Opaque
+ } else {
+ warehouse = u.Path
+ }
- if u.Opaque != "" {
- warehouse = u.Opaque
- } else {
- warehouse = u.Path
- }
+ if warehouse == "" || warehouse == "/" {
+ return nil, errors.New("hadoop catalog: local
filesystem requires a non-root warehouse path")
+ }
- if warehouse == "" || warehouse == "/" {
- return nil, errors.New("hadoop catalog requires a non-root
warehouse path")
- }
+ warehouse = strings.TrimRight(warehouse, "/")
- warehouse = strings.TrimRight(warehouse, "/")
+ // Normalize to absolute path so the synthetic "location"
property
+ // always produces a valid file:// URI.
+ absWarehouse, err := filepath.Abs(warehouse)
+ if err != nil {
+ return nil, fmt.Errorf("hadoop catalog: failed to
resolve absolute warehouse path: %w", err)
+ }
- // Normalize to absolute path so the synthetic "location" property
- // always produces a valid file:// URI.
- absWarehouse, err := filepath.Abs(warehouse)
+ warehouse = absWarehouse
+ }
+
+ filesystem, err := io.LoadFS(context.Background(), nil, warehouse)
if err != nil {
- return nil, fmt.Errorf("hadoop catalog: failed to resolve
absolute warehouse path: %w", err)
+ return nil, fmt.Errorf("hadoop catalog: failed to load
filesystem: %w", err)
}
- warehouse = absWarehouse
+ hadoopFs, ok := filesystem.(HadoopCatalogFS)
Review Comment:
Minor, but the message is a little wordy given the `hadoop catalog:` prefix
— naming the interface directly is more actionable: `"hadoop catalog: %T does
not implement HadoopCatalogFS"`.
##########
catalog/hadoop/hadoop.go:
##########
@@ -135,35 +135,45 @@ func NewCatalog(name, warehouse string, props
iceberg.Properties) (*Catalog, err
if !isLocal && !allowUnsafeCommits {
return nil, fmt.Errorf("hadoop catalog: when using warehouse
scheme %q, `allow-unsafe-commits` must be set to true", u.Scheme)
- }
+ } else if isLocal {
+ if u.Opaque != "" {
+ warehouse = u.Opaque
+ } else {
+ warehouse = u.Path
+ }
- if u.Opaque != "" {
- warehouse = u.Opaque
- } else {
- warehouse = u.Path
- }
+ if warehouse == "" || warehouse == "/" {
+ return nil, errors.New("hadoop catalog: local
filesystem requires a non-root warehouse path")
+ }
- if warehouse == "" || warehouse == "/" {
- return nil, errors.New("hadoop catalog requires a non-root
warehouse path")
- }
+ warehouse = strings.TrimRight(warehouse, "/")
- warehouse = strings.TrimRight(warehouse, "/")
+ // Normalize to absolute path so the synthetic "location"
property
+ // always produces a valid file:// URI.
+ absWarehouse, err := filepath.Abs(warehouse)
+ if err != nil {
+ return nil, fmt.Errorf("hadoop catalog: failed to
resolve absolute warehouse path: %w", err)
+ }
- // Normalize to absolute path so the synthetic "location" property
- // always produces a valid file:// URI.
- absWarehouse, err := filepath.Abs(warehouse)
+ warehouse = absWarehouse
+ }
+
+ filesystem, err := io.LoadFS(context.Background(), nil, warehouse)
if err != nil {
- return nil, fmt.Errorf("hadoop catalog: failed to resolve
absolute warehouse path: %w", err)
+ return nil, fmt.Errorf("hadoop catalog: failed to load
filesystem: %w", err)
}
- warehouse = absWarehouse
+ hadoopFs, ok := filesystem.(HadoopCatalogFS)
Review Comment:
I think this assertion can never succeed for a remote store today, which
means the PR's headline capability isn't actually reachable.
`HadoopCatalogFS` requires `Stat`, `Rename`, `RemoveAll`, `MkdirAll` on top
of the read/write/list set. `blobFileIO` — what all the S3/GCS/Azure backends
return — only implements `Open`, `Remove`, `Create`, `WriteFile`, `WalkDir`,
`ReadFile`, `DeleteFiles`. So for any non-local scheme `LoadFS` succeeds, `ok`
is false, and `NewCatalog` errors out even with `allow-unsafe-commits=true`. A
user who correctly imports the gocloud backend sees "does not implement
necessary functions" and reasonably reads it as a library bug.
The hard part is `Rename` — it's fundamentally non-atomic on object stores,
which is exactly why Java warns against S3-backed `HadoopCatalog`. So I'd
either extend `blobFileIO` with the four methods (with `Rename` as copy+delete)
before merging, or scope this PR as a preparatory step and say so in the
description. Leaving the assertion with nothing that can satisfy it is the dead
end I'd want to avoid. Thoughts?
##########
catalog/hadoop/hadoop.go:
##########
@@ -135,35 +135,45 @@ func NewCatalog(name, warehouse string, props
iceberg.Properties) (*Catalog, err
if !isLocal && !allowUnsafeCommits {
return nil, fmt.Errorf("hadoop catalog: when using warehouse
scheme %q, `allow-unsafe-commits` must be set to true", u.Scheme)
- }
+ } else if isLocal {
+ if u.Opaque != "" {
+ warehouse = u.Opaque
+ } else {
+ warehouse = u.Path
+ }
- if u.Opaque != "" {
- warehouse = u.Opaque
- } else {
- warehouse = u.Path
- }
+ if warehouse == "" || warehouse == "/" {
+ return nil, errors.New("hadoop catalog: local
filesystem requires a non-root warehouse path")
+ }
- if warehouse == "" || warehouse == "/" {
- return nil, errors.New("hadoop catalog requires a non-root
warehouse path")
- }
+ warehouse = strings.TrimRight(warehouse, "/")
- warehouse = strings.TrimRight(warehouse, "/")
+ // Normalize to absolute path so the synthetic "location"
property
+ // always produces a valid file:// URI.
+ absWarehouse, err := filepath.Abs(warehouse)
+ if err != nil {
+ return nil, fmt.Errorf("hadoop catalog: failed to
resolve absolute warehouse path: %w", err)
+ }
- // Normalize to absolute path so the synthetic "location" property
- // always produces a valid file:// URI.
- absWarehouse, err := filepath.Abs(warehouse)
+ warehouse = absWarehouse
+ }
+
+ filesystem, err := io.LoadFS(context.Background(), nil, warehouse)
if err != nil {
- return nil, fmt.Errorf("hadoop catalog: failed to resolve
absolute warehouse path: %w", err)
+ return nil, fmt.Errorf("hadoop catalog: failed to load
filesystem: %w", err)
}
- warehouse = absWarehouse
+ hadoopFs, ok := filesystem.(HadoopCatalogFS)
Review Comment:
Related, but worth its own note since it lives outside the diff:
`LoadNamespaceProperties` builds the location property as `(&url.URL{Scheme:
"file", Path: path}).String()` with a hardcoded `file://` scheme. Once a
non-local warehouse is supported, an S3 namespace returns
`file://s3:/bucket/wh/ns` — a `file://` URL wrapping the corrupted S3 path. I'd
thread the actual warehouse scheme through there (or reuse the
`namespaceToPath` result) as part of the same scheme-awareness fix.
##########
catalog/hadoop/hadoop.go:
##########
@@ -135,35 +135,45 @@ func NewCatalog(name, warehouse string, props
iceberg.Properties) (*Catalog, err
if !isLocal && !allowUnsafeCommits {
return nil, fmt.Errorf("hadoop catalog: when using warehouse
scheme %q, `allow-unsafe-commits` must be set to true", u.Scheme)
- }
+ } else if isLocal {
+ if u.Opaque != "" {
+ warehouse = u.Opaque
+ } else {
+ warehouse = u.Path
+ }
- if u.Opaque != "" {
- warehouse = u.Opaque
- } else {
- warehouse = u.Path
- }
+ if warehouse == "" || warehouse == "/" {
+ return nil, errors.New("hadoop catalog: local
filesystem requires a non-root warehouse path")
+ }
- if warehouse == "" || warehouse == "/" {
- return nil, errors.New("hadoop catalog requires a non-root
warehouse path")
- }
+ warehouse = strings.TrimRight(warehouse, "/")
- warehouse = strings.TrimRight(warehouse, "/")
+ // Normalize to absolute path so the synthetic "location"
property
+ // always produces a valid file:// URI.
+ absWarehouse, err := filepath.Abs(warehouse)
+ if err != nil {
+ return nil, fmt.Errorf("hadoop catalog: failed to
resolve absolute warehouse path: %w", err)
+ }
- // Normalize to absolute path so the synthetic "location" property
- // always produces a valid file:// URI.
- absWarehouse, err := filepath.Abs(warehouse)
+ warehouse = absWarehouse
+ }
+
+ filesystem, err := io.LoadFS(context.Background(), nil, warehouse)
if err != nil {
- return nil, fmt.Errorf("hadoop catalog: failed to resolve
absolute warehouse path: %w", err)
+ return nil, fmt.Errorf("hadoop catalog: failed to load
filesystem: %w", err)
}
- warehouse = absWarehouse
+ hadoopFs, ok := filesystem.(HadoopCatalogFS)
+ if !ok {
+ return nil, fmt.Errorf("hadoop catalog: filesystem %T does not
implement necessary functions to be used as a Hadoop catalog", filesystem)
+ }
return &Catalog{
name: name,
warehouse: warehouse,
// for the time being, we default to localfs since there is not
yet
Review Comment:
This comment now contradicts the code right below it — we're no longer
defaulting to localfs, the FS is loaded dynamically from the registry by
scheme. I'd drop it or replace with something like "filesystem is resolved
dynamically from the IO registry based on the warehouse scheme."
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
