You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/07/14 15:58:09 UTC
(iceberg-go) branch main updated: fix(hive): lock table drops and renames (#1442)
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 a84cb97a fix(hive): lock table drops and renames (#1442)
a84cb97a is described below
commit a84cb97a1eab20815c340d9307760d93afaee85e
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Tue Jul 14 17:58:03 2026 +0200
fix(hive): lock table drops and renames (#1442)
---
catalog/hive/hive.go | 45 +++++++++++++++++++++++++++++++++++++
catalog/hive/hive_test.go | 57 +++++++++++++++++++++++++++++++++++++++++++++++
catalog/hive/lock.go | 45 ++++++++++++++++++++++++++++++++-----
catalog/hive/lock_test.go | 25 +++++++++++++++++++++
4 files changed, 167 insertions(+), 5 deletions(-)
diff --git a/catalog/hive/hive.go b/catalog/hive/hive.go
index 715bdad6..7bdf6b2f 100644
--- a/catalog/hive/hive.go
+++ b/catalog/hive/hive.go
@@ -373,6 +373,16 @@ func (c *Catalog) DropTable(ctx context.Context, identifier table.Identifier) er
return err
}
+ lock, err := acquireLock(ctx, c.client, database, tableName, c.opts)
+ if err != nil {
+ return fmt.Errorf("%w: failed to acquire lock for %s.%s: %w", table.ErrCommitFailed, database, tableName, err)
+ }
+ defer func() {
+ _ = lock.Release(ctx)
+ }()
+
+ // Re-read after acquiring the lock so drop cannot act on table state that a
+ // concurrent commit changed while lock acquisition was waiting.
if _, err := c.getIcebergTable(ctx, database, tableName); err != nil {
return err
}
@@ -423,10 +433,45 @@ func (c *Catalog) RenameTable(ctx context.Context, from, to table.Identifier) (*
return nil, fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace, toDB)
}
+ source, err := c.getIcebergTable(ctx, fromDB, fromTable)
+ if err != nil {
+ return nil, err
+ }
+ sourceMetadataLocation, err := getMetadataLocation(source)
+ if err != nil {
+ return nil, err
+ }
+
+ lockIdentifiers := []tableLockIdentifier{
+ {database: fromDB, table: fromTable},
+ {database: toDB, table: toTable},
+ }
+ lock, err := acquireLocks(ctx, c.client, lockIdentifiers, c.opts)
+ if err != nil {
+ return nil, fmt.Errorf("%w: failed to acquire locks for rename %s.%s to %s.%s: %w",
+ table.ErrCommitFailed, fromDB, fromTable, toDB, toTable, err)
+ }
+ defer func() {
+ _ = lock.Release(ctx)
+ }()
+
hiveTbl, err := c.getIcebergTable(ctx, fromDB, fromTable)
+ if err != nil {
+ if errors.Is(err, catalog.ErrNoSuchTable) {
+ return nil, fmt.Errorf("%w: source table %s.%s changed during rename: %w",
+ table.ErrCommitFailed, fromDB, fromTable, err)
+ }
+
+ return nil, err
+ }
+ lockedMetadataLocation, err := getMetadataLocation(hiveTbl)
if err != nil {
return nil, err
}
+ if lockedMetadataLocation != sourceMetadataLocation {
+ return nil, fmt.Errorf("%w: source table %s.%s metadata location changed from %s to %s",
+ table.ErrCommitFailed, fromDB, fromTable, sourceMetadataLocation, lockedMetadataLocation)
+ }
hiveTbl.TableName = toTable
hiveTbl.DbName = toDB
diff --git a/catalog/hive/hive_test.go b/catalog/hive/hive_test.go
index 2756d40e..3eed2e8e 100644
--- a/catalog/hive/hive_test.go
+++ b/catalog/hive/hive_test.go
@@ -20,6 +20,7 @@ package hive
import (
"context"
"errors"
+ "maps"
"net/url"
"os"
"path/filepath"
@@ -140,6 +141,12 @@ func (m *mockHiveClient) Close() error {
return args.Error(0)
}
+func expectImmediateTableLock(mockClient *mockHiveClient, lockID int64) {
+ mockClient.On("Lock", mock.Anything, mock.AnythingOfType("*hive_metastore.LockRequest")).
+ Return(&hive_metastore.LockResponse{Lockid: lockID, State: hive_metastore.LockState_ACQUIRED}, nil).Once()
+ mockClient.On("Unlock", mock.Anything, lockID).Return(nil).Once()
+}
+
// Test data
var testIcebergHiveTable1 = &hive_metastore.Table{
@@ -515,6 +522,7 @@ func TestHiveDropTable(t *testing.T) {
assert := require.New(t)
mockClient := &mockHiveClient{}
+ expectImmediateTableLock(mockClient, 1)
mockClient.On("GetTable", mock.Anything, "test_database", "test_table").
Return(testIcebergHiveTable1, nil).Once()
@@ -574,6 +582,50 @@ func TestHiveCommitTableValidatesRequirementsForMissingTable(t *testing.T) {
mockClient.AssertExpectations(t)
}
+func TestHiveDropTableLockFailureIsRetryable(t *testing.T) {
+ mockClient := &mockHiveClient{}
+ mockClient.On("Lock", mock.Anything, mock.AnythingOfType("*hive_metastore.LockRequest")).
+ Return(nil, errors.New("lock conflict")).Once()
+ hiveCatalog := NewCatalogWithClient(mockClient, iceberg.Properties{})
+
+ err := hiveCatalog.DropTable(context.Background(), TableIdentifier("test_database", "test_table"))
+ require.ErrorIs(t, err, table.ErrCommitFailed)
+ mockClient.AssertNotCalled(t, "GetTable", mock.Anything, mock.Anything, mock.Anything)
+ mockClient.AssertExpectations(t)
+}
+
+func TestHiveRenameTableRejectsChangedMetadataAfterLock(t *testing.T) {
+ mockClient := &mockHiveClient{}
+ before := *testIcebergHiveTable1
+ before.Parameters = maps.Clone(testIcebergHiveTable1.Parameters)
+ after := before
+ after.Parameters = maps.Clone(before.Parameters)
+ after.Parameters[MetadataLocationKey] = "s3://warehouse/db/table/metadata/v2.metadata.json"
+
+ mockClient.On("GetDatabase", mock.Anything, "target_database").
+ Return(&hive_metastore.Database{Name: "target_database"}, nil).Once()
+ mockClient.On("GetTable", mock.Anything, "test_database", "test_table").
+ Return(&before, nil).Once()
+ mockClient.On("Lock", mock.Anything, mock.MatchedBy(func(request *hive_metastore.LockRequest) bool {
+ return len(request.Component) == 2 &&
+ request.Component[0].Dbname == "target_database" && *request.Component[0].Tablename == "renamed" &&
+ request.Component[1].Dbname == "test_database" && *request.Component[1].Tablename == "test_table"
+ })).Return(&hive_metastore.LockResponse{Lockid: 2, State: hive_metastore.LockState_ACQUIRED}, nil).Once()
+ mockClient.On("GetTable", mock.Anything, "test_database", "test_table").
+ Return(&after, nil).Once()
+ mockClient.On("Unlock", mock.Anything, int64(2)).Return(nil).Once()
+ hiveCatalog := NewCatalogWithClient(mockClient, iceberg.Properties{})
+
+ _, err := hiveCatalog.RenameTable(
+ context.Background(),
+ TableIdentifier("test_database", "test_table"),
+ TableIdentifier("target_database", "renamed"),
+ )
+ require.ErrorIs(t, err, table.ErrCommitFailed)
+ mockClient.AssertNotCalled(t, "AlterTable", mock.Anything, mock.Anything, mock.Anything, mock.Anything)
+ mockClient.AssertExpectations(t)
+}
+
func TestHivePurgeTable(t *testing.T) {
assert := require.New(t)
ctx := context.Background()
@@ -582,6 +634,7 @@ func TestHivePurgeTable(t *testing.T) {
hiveTable := hivePurgeTable(metadataLocation, tableLocation)
mockClient := &mockHiveClient{}
+ expectImmediateTableLock(mockClient, 1)
mockClient.On("GetTable", mock.Anything, "test_database", "test_table").
Return(hiveTable, nil).Twice()
mockClient.On("DropTable", mock.Anything, "test_database", "test_table", false).
@@ -626,6 +679,7 @@ func TestHivePurgeTableSwallowsPurgeFilesError(t *testing.T) {
hiveTable := hivePurgeTable(metadataLocation, tableLocation)
mockClient := &mockHiveClient{}
+ expectImmediateTableLock(mockClient, 1)
mockClient.On("GetTable", mock.Anything, "test_database", "test_table").
Return(hiveTable, nil).Twice()
mockClient.On("DropTable", mock.Anything, "test_database", "test_table", false).
@@ -655,6 +709,7 @@ func TestHivePurgeTableWithGCDisabled(t *testing.T) {
hiveTable := hivePurgeTable(metadataLocation, tableLocation)
mockClient := &mockHiveClient{}
+ expectImmediateTableLock(mockClient, 1)
mockClient.On("GetTable", mock.Anything, "test_database", "test_table").
Return(hiveTable, nil).Twice()
mockClient.On("DropTable", mock.Anything, "test_database", "test_table", false).
@@ -741,6 +796,7 @@ func TestHiveDropTableNotExists(t *testing.T) {
assert := require.New(t)
mockClient := &mockHiveClient{}
+ expectImmediateTableLock(mockClient, 1)
mockClient.On("GetTable", mock.Anything, "test_database", "nonexistent").
Return(nil, errNoSuchObject).Once()
@@ -758,6 +814,7 @@ func TestHiveDropTableNonIceberg(t *testing.T) {
assert := require.New(t)
mockClient := &mockHiveClient{}
+ expectImmediateTableLock(mockClient, 1)
mockClient.On("GetTable", mock.Anything, "test_database", "other_table").
Return(testNonIcebergHiveTable, nil).Once()
diff --git a/catalog/hive/lock.go b/catalog/hive/lock.go
index 51f56505..d9fdc72c 100644
--- a/catalog/hive/lock.go
+++ b/catalog/hive/lock.go
@@ -22,6 +22,8 @@ import (
"errors"
"fmt"
"math"
+ "slices"
+ "strings"
"time"
"github.com/beltran/gohive/hive_metastore"
@@ -36,13 +38,37 @@ type HiveLock struct {
}
func acquireLock(ctx context.Context, client HiveClient, database, tableName string, opts *HiveOptions) (*HiveLock, error) {
- lockReq := &hive_metastore.LockRequest{
- Component: []*hive_metastore.LockComponent{{
+ return acquireLocks(ctx, client, []tableLockIdentifier{{database: database, table: tableName}}, opts)
+}
+
+type tableLockIdentifier struct {
+ database string
+ table string
+}
+
+func acquireLocks(ctx context.Context, client HiveClient, identifiers []tableLockIdentifier, opts *HiveOptions) (*HiveLock, error) {
+ identifiers = slices.Clone(identifiers)
+ slices.SortFunc(identifiers, func(a, b tableLockIdentifier) int {
+ if cmp := strings.Compare(a.database, b.database); cmp != 0 {
+ return cmp
+ }
+
+ return strings.Compare(a.table, b.table)
+ })
+ identifiers = slices.Compact(identifiers)
+
+ components := make([]*hive_metastore.LockComponent, len(identifiers))
+ for i, ident := range identifiers {
+ tableName := ident.table
+ components[i] = &hive_metastore.LockComponent{
Type: hive_metastore.LockType_EXCLUSIVE,
Level: hive_metastore.LockLevel_TABLE,
- Dbname: database,
+ Dbname: ident.database,
Tablename: &tableName,
- }},
+ }
+ }
+ lockReq := &hive_metastore.LockRequest{
+ Component: components,
}
lockResp, err := client.Lock(ctx, lockReq)
@@ -98,7 +124,16 @@ func acquireLock(ctx context.Context, client HiveClient, database, tableName str
_ = client.Unlock(ctx, lockResp.Lockid)
- return nil, fmt.Errorf("%w: exhausted %d retries for table %s.%s", ErrLockAcquisitionFailed, opts.LockRetries, database, tableName)
+ return nil, fmt.Errorf("%w: exhausted %d retries for tables %s", ErrLockAcquisitionFailed, opts.LockRetries, formatLockIdentifiers(identifiers))
+}
+
+func formatLockIdentifiers(identifiers []tableLockIdentifier) string {
+ names := make([]string, len(identifiers))
+ for i, ident := range identifiers {
+ names[i] = ident.database + "." + ident.table
+ }
+
+ return strings.Join(names, ", ")
}
func calculateBackoff(attempt int, minWait, maxWait time.Duration) time.Duration {
diff --git a/catalog/hive/lock_test.go b/catalog/hive/lock_test.go
index f61a8079..45a7a0c8 100644
--- a/catalog/hive/lock_test.go
+++ b/catalog/hive/lock_test.go
@@ -49,6 +49,31 @@ func TestAcquireLockImmediateSuccess(t *testing.T) {
mockClient.AssertExpectations(t)
}
+func TestAcquireLocksSortsComponents(t *testing.T) {
+ mockClient := new(mockHiveClient)
+ ctx := context.Background()
+ var request *hive_metastore.LockRequest
+ mockClient.On("Lock", ctx, mock.AnythingOfType("*hive_metastore.LockRequest")).
+ Run(func(args mock.Arguments) {
+ request = args.Get(1).(*hive_metastore.LockRequest)
+ }).
+ Return(&hive_metastore.LockResponse{Lockid: 124, State: hive_metastore.LockState_ACQUIRED}, nil).Once()
+ mockClient.On("Unlock", ctx, int64(124)).Return(nil).Once()
+
+ lock, err := acquireLocks(ctx, mockClient, []tableLockIdentifier{
+ {database: "z_db", table: "source"},
+ {database: "a_db", table: "destination"},
+ }, NewHiveOptions())
+ require.NoError(t, err)
+ require.Len(t, request.Component, 2)
+ assert.Equal(t, "a_db", request.Component[0].Dbname)
+ assert.Equal(t, "destination", *request.Component[0].Tablename)
+ assert.Equal(t, "z_db", request.Component[1].Dbname)
+ assert.Equal(t, "source", *request.Component[1].Tablename)
+ require.NoError(t, lock.Release(ctx))
+ mockClient.AssertExpectations(t)
+}
+
func TestAcquireLockWithRetry(t *testing.T) {
mockClient := new(mockHiveClient)
ctx := context.Background()