(iceberg-go) branch main updated: refactor(catalog): unify metadata IO operations by passing Table instances directly (#1167)

[ze...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git


The following commit(s) were added to refs/heads/main by this push:
     new 914dd635 refactor(catalog): unify metadata IO operations by passing Table instances directly (#1167)
914dd635 is described below

commit 914dd635527f595207b0c9081f2d1a6b55932633
Author: Blue Li <10...@users.noreply.github.com>
AuthorDate: Tue Jun 9 03:58:37 2026 +0800

    refactor(catalog): unify metadata IO operations by passing Table instances directly (#1167)
    
    **What changes were proposed in this pull request?**
    This PR fixes an authentication issue when updating tables.
    
    Currently, `internal.UpdateAndStageTable` and `internal.WriteMetadata`
    only use table properties. This causes IO failures for remote storage
    (like S3) because catalog-level credentials (e.g., `s3.access-key-id`)
    are lost.
    
    **Fix:**
    - Updated `UpdateAndStageTable` to accept `catprops` and merge it with
    table properties before initializing the filesystem.
    - Applied the same property merge to `internal.WriteMetadata`.
    - Note: This aligns the update logic with how `CreateStagedTable`
    already works. `nsprops` is intentionally omitted here since table
    locations are already fixed during updates.
    
    **How was this patch tested?**
    - `make test` passes successfully.
    - *No specific unit test was added for this fix, as our mock filesystems
    do not validate remote credentials, making it hard to test credential
    merging effectively. Relying on alignment with `CreateStagedTable`
    instead.*
---
 catalog/glue/glue.go      | 16 +++-------------
 catalog/hive/hive.go      | 16 +++-------------
 catalog/internal/utils.go | 21 +++++++++++++++------
 catalog/sql/sql.go        | 16 +++-------------
 4 files changed, 24 insertions(+), 45 deletions(-)

diff --git a/catalog/glue/glue.go b/catalog/glue/glue.go
index f623730f..489764c7 100644
--- a/catalog/glue/glue.go
+++ b/catalog/glue/glue.go
@@ -255,17 +255,7 @@ func (c *Catalog) CreateTable(ctx context.Context, identifier table.Identifier,
 		return nil, err
 	}
 
-	afs, err := staged.FS(ctx)
-	if err != nil {
-		return nil, err
-	}
-	wfs, ok := afs.(io.WriteFileIO)
-	if !ok {
-		return nil, errors.New("loaded filesystem IO does not support writing")
-	}
-
-	compression := staged.Table.Properties().Get(table.MetadataCompressionKey, table.MetadataCompressionDefault)
-	if err := internal.WriteTableMetadata(staged.Metadata(), wfs, staged.MetadataLocation(), compression); err != nil {
+	if err := internal.WriteMetadata(ctx, staged.Table); err != nil {
 		return nil, err
 	}
 
@@ -334,14 +324,14 @@ func (c *Catalog) CommitTable(ctx context.Context, identifier table.Identifier,
 	}
 
 	// Create a staging table with the updates applied
-	staged, err := internal.UpdateAndStageTable(ctx, current, identifier, requirements, updates, c)
+	staged, err := internal.UpdateAndStageTable(ctx, c.props, current, identifier, requirements, updates, c)
 	if err != nil {
 		return nil, "", err
 	}
 	if current != nil && staged.Metadata().Equals(current.Metadata()) {
 		return current.Metadata(), current.MetadataLocation(), nil
 	}
-	if err := internal.WriteMetadata(ctx, staged.Metadata(), staged.MetadataLocation(), staged.Properties()); err != nil {
+	if err := internal.WriteMetadata(ctx, staged.Table); err != nil {
 		return nil, "", err
 	}
 
diff --git a/catalog/hive/hive.go b/catalog/hive/hive.go
index e88d30a2..3021d1ed 100644
--- a/catalog/hive/hive.go
+++ b/catalog/hive/hive.go
@@ -178,17 +178,7 @@ func (c *Catalog) CreateTable(ctx context.Context, identifier table.Identifier,
 		return nil, err
 	}
 
-	afs, err := staged.FS(ctx)
-	if err != nil {
-		return nil, err
-	}
-	wfs, ok := afs.(io.WriteFileIO)
-	if !ok {
-		return nil, errors.New("loaded filesystem IO does not support writing")
-	}
-
-	compression := staged.Table.Properties().Get(table.MetadataCompressionKey, table.MetadataCompressionDefault)
-	if err := internal.WriteTableMetadata(staged.Metadata(), wfs, staged.MetadataLocation(), compression); err != nil {
+	if err := internal.WriteMetadata(ctx, staged.Table); err != nil {
 		return nil, err
 	}
 
@@ -483,7 +473,7 @@ func (c *Catalog) CommitTable(ctx context.Context, identifier table.Identifier,
 		}
 	}
 
-	staged, err := internal.UpdateAndStageTable(ctx, current, identifier, requirements, updates, c)
+	staged, err := internal.UpdateAndStageTable(ctx, c.opts.props, current, identifier, requirements, updates, c)
 	if err != nil {
 		return nil, "", err
 	}
@@ -492,7 +482,7 @@ func (c *Catalog) CommitTable(ctx context.Context, identifier table.Identifier,
 		return current.Metadata(), current.MetadataLocation(), nil
 	}
 
-	if err := internal.WriteMetadata(ctx, staged.Metadata(), staged.MetadataLocation(), staged.Properties()); err != nil {
+	if err := internal.WriteMetadata(ctx, staged.Table); err != nil {
 		return nil, "", err
 	}
 
diff --git a/catalog/internal/utils.go b/catalog/internal/utils.go
index 87383bf1..7f513a98 100644
--- a/catalog/internal/utils.go
+++ b/catalog/internal/utils.go
@@ -74,8 +74,8 @@ func WriteTableMetadata(metadata table.Metadata, fs icebergio.WriteFileIO, loc s
 	return
 }
 
-func WriteMetadata(ctx context.Context, metadata table.Metadata, loc string, props iceberg.Properties) error {
-	fs, err := icebergio.LoadFS(ctx, props, loc)
+func WriteMetadata(ctx context.Context, tbl *table.Table) error {
+	fs, err := tbl.FS(ctx)
 	if err != nil {
 		return err
 	}
@@ -85,9 +85,9 @@ func WriteMetadata(ctx context.Context, metadata table.Metadata, loc string, pro
 		return errors.New("filesystem IO does not support writing")
 	}
 
-	compression := props.Get(table.MetadataCompressionKey, table.MetadataCompressionDefault)
+	compression := tbl.Properties().Get(table.MetadataCompressionKey, table.MetadataCompressionDefault)
 
-	return WriteTableMetadata(metadata, wfs, loc, compression)
+	return WriteTableMetadata(tbl.Metadata(), wfs, tbl.MetadataLocation(), compression)
 }
 
 func UpdateTableMetadata(base table.Metadata, updates []table.Update, metadataLoc string) (table.Metadata, error) {
@@ -125,6 +125,9 @@ func CreateStagedTable(ctx context.Context, catprops iceberg.Properties, nsprops
 	}
 
 	ioProps := maps.Clone(catprops)
+	if ioProps == nil {
+		ioProps = iceberg.Properties{}
+	}
 	maps.Copy(ioProps, cfg.Properties)
 
 	return table.StagedTable{
@@ -192,7 +195,7 @@ func ParseMetadataVersion(location string) int {
 	return v
 }
 
-func UpdateAndStageTable(ctx context.Context, current *table.Table, ident table.Identifier, reqs []table.Requirement, updates []table.Update, cat table.CatalogIO) (*table.StagedTable, error) {
+func UpdateAndStageTable(ctx context.Context, catprops iceberg.Properties, current *table.Table, ident table.Identifier, reqs []table.Requirement, updates []table.Update, cat table.CatalogIO) (*table.StagedTable, error) {
 	var (
 		baseMeta    table.Metadata
 		metadataLoc string
@@ -231,12 +234,18 @@ func UpdateAndStageTable(ctx context.Context, current *table.Table, ident table.
 		return nil, err
 	}
 
+	ioProps := maps.Clone(catprops)
+	if ioProps == nil {
+		ioProps = iceberg.Properties{}
+	}
+	maps.Copy(ioProps, updated.Properties())
+
 	return &table.StagedTable{
 		Table: table.New(
 			ident,
 			updated,
 			newLocation,
-			icebergio.LoadFSFunc(updated.Properties(), newLocation),
+			icebergio.LoadFSFunc(ioProps, newLocation),
 			cat,
 		),
 	}, nil
diff --git a/catalog/sql/sql.go b/catalog/sql/sql.go
index 1c2d6b29..d4cc0ba4 100644
--- a/catalog/sql/sql.go
+++ b/catalog/sql/sql.go
@@ -293,17 +293,7 @@ func (c *Catalog) CreateTable(ctx context.Context, ident table.Identifier, sc *i
 		return nil, fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace, ns)
 	}
 
-	afs, err := staged.FS(ctx)
-	if err != nil {
-		return nil, err
-	}
-	wfs, ok := afs.(io.WriteFileIO)
-	if !ok {
-		return nil, errors.New("loaded filesystem IO does not support writing")
-	}
-
-	compression := staged.Table.Properties().Get(table.MetadataCompressionKey, table.MetadataCompressionDefault)
-	if err := internal.WriteTableMetadata(staged.Metadata(), wfs, staged.MetadataLocation(), compression); err != nil {
+	if err := internal.WriteMetadata(ctx, staged.Table); err != nil {
 		return nil, err
 	}
 
@@ -337,7 +327,7 @@ func (c *Catalog) CommitTable(ctx context.Context, ident table.Identifier, reqs
 		return nil, "", err
 	}
 
-	staged, err := internal.UpdateAndStageTable(ctx, current, ident, reqs, updates, c)
+	staged, err := internal.UpdateAndStageTable(ctx, c.props, current, ident, reqs, updates, c)
 	if err != nil {
 		return nil, "", err
 	}
@@ -347,7 +337,7 @@ func (c *Catalog) CommitTable(ctx context.Context, ident table.Identifier, reqs
 		return current.Metadata(), current.MetadataLocation(), nil
 	}
 
-	if err := internal.WriteMetadata(ctx, staged.Metadata(), staged.MetadataLocation(), staged.Properties()); err != nil {
+	if err := internal.WriteMetadata(ctx, staged.Table); err != nil {
 		return nil, "", err
 	}