laskoviymishka commented on code in PR #2001:
URL: https://github.com/apache/iceberg-go/pull/2001#discussion_r4007811010
##########
catalog/glue/glue.go:
##########
@@ -277,18 +282,39 @@ var _ catalog.Closer = (*Catalog)(nil)
// This function will create the metadata file in S3 using the catalog and
table properties,
// to determine the bucket and key for the metadata location.
func (c *Catalog) CreateTable(ctx context.Context, identifier
table.Identifier, schema *iceberg.Schema, opts ...catalog.CreateTableOpt)
(*table.Table, error) {
+ // A missing namespace is reported before touching Glue, matching the
contract
+ // callers rely on (an identifier without a database is not a missing
table).
+ if len(identifier) < 2 {
+ return nil, fmt.Errorf("%w: missing namespace or invalid
identifier %v", catalog.ErrNoSuchNamespace, identifier)
+ }
+
+ database, tableName, err := identifierToGlueTable(identifier)
+ if err != nil {
+ return nil, err
+ }
+
// The reporter is resolved once at construction (see NewCatalog), so a
bad
// metrics-reporter-impl already failed there — no per-op guard is
needed
// before mutating the catalog, and the trailing LoadTable reuses the
cached
// reporter.
staged, err := internal.CreateStagedTable(ctx, c.props,
c.LoadNamespaceProperties, identifier, schema, opts...)
if err != nil {
- return nil, err
- }
+ // S3 Tables federated databases assign storage themselves, so
client-side
+ // location resolution fails with ErrNoDefaultLocation. Only
then probe for
+ // federation and retry via the S3 Tables path, keeping the
extra
+ // GetDatabase off every other create.
+ if !errors.Is(err, internal.ErrNoDefaultLocation) {
Review Comment:
The federation probe only fires on `ErrNoDefaultLocation`. When a caller
passes `WithLocation` into an S3 Tables database, `CreateStagedTable` succeeds,
we take the generic branch, and we write a plain `EXTERNAL_TABLE` pointing at
the caller's path. The table never lands in S3 Tables managed storage and
carries no `FederatedTable` marker, so pyiceberg and Java clients would happily
treat it as a normal table while the S3 Tables APIs can't see it. pyiceberg
treats this as an invariant: `_create_table_s3tables` rejects a non-nil
location up front.
What worries me most is that
`TestGlueCreateTableExplicitLocationSkipsFederationProbe` blesses exactly this
as intentional, so we've encoded the divergence in a test.
I'd add a pre-check at the top of `CreateTable`: if a location was supplied
and the database is S3 Tables, return an error the way pyiceberg does, and flip
that test to assert the rejection. It costs one `GetDatabase` on the
explicit-location path, which is worth it to not silently break the
managed-storage contract. wdyt?
##########
catalog/glue/glue.go:
##########
@@ -307,6 +333,113 @@ func (c *Catalog) CreateTable(ctx context.Context,
identifier table.Identifier,
return c.LoadTable(ctx, identifier)
}
+// isS3TablesDatabase reports whether the Glue database is federated to the
+// Amazon S3 Tables service, which owns table storage and location assignment.
+func (c *Catalog) isS3TablesDatabase(ctx context.Context, database string)
(bool, error) {
+ db, err := c.getDatabase(ctx, database)
+ if err != nil {
+ // A missing database is not fatal here; let the generic create
path
+ // surface it, so this probe never changes the error a caller
already saw.
+ if errors.Is(err, catalog.ErrNoSuchNamespace) {
+ return false, nil
+ }
+
+ return false, err
+ }
+
+ return db.FederatedDatabase != nil &&
+
strings.EqualFold(aws.ToString(db.FederatedDatabase.ConnectionType),
s3TablesConnectionType), nil
+}
+
+// createS3TablesTable creates a table in an S3 Tables federated database. The
+// service assigns storage, so a minimal entry is created first to allocate the
+// location, then updated with the written metadata pointer; on any later
+// failure the minimal entry is removed so no half-created table is left
behind.
+func (c *Catalog) createS3TablesTable(ctx context.Context, database, tableName
string, identifier table.Identifier, schema *iceberg.Schema, opts
...catalog.CreateTableOpt) (*table.Table, error) {
+ _, err := c.glueSvc.CreateTable(ctx, &glue.CreateTableInput{
+ CatalogId: c.catalogId,
+ DatabaseName: aws.String(database),
+ TableInput: &types.TableInput{
+ Name: aws.String(tableName),
+ Parameters: map[string]string{glueParamFormat:
glueTypeIceberg},
+ },
+ })
+ if err != nil {
+ return nil, fmt.Errorf("failed to allocate S3 Tables storage
for %s.%s: %w", database, tableName, err)
+ }
+
+ if err := c.commitS3TablesTable(ctx, database, tableName, identifier,
schema, opts...); err != nil {
+ if _, delErr := c.glueSvc.DeleteTable(ctx,
&glue.DeleteTableInput{
+ CatalogId: c.catalogId,
+ DatabaseName: aws.String(database),
+ Name: aws.String(tableName),
+ }); delErr != nil {
+ return nil, fmt.Errorf("%w (failed to clean up
allocated table %s.%s: %w)", err, database, tableName, delErr)
Review Comment:
If both `commitS3TablesTable` and this `DeleteTable` fail, we return the
combined error and leave a minimal entry behind whose parameters are just
`{format: "ICEBERG"}` with no iceberg table_type key. `getRawTable` then
rejects it (`isIceberg` is false without the table_type param, and the service
`TableType` isn't `EXTERNAL_TABLE`), so `DropTable` can't remove it and the
operator has no catalog API to repair the state.
The PR's stated invariant is "no half-created table left behind", and this
is the one case that breaks it and can't be cleaned up afterward. I'd either
let `getRawTable`/`DropTable` tolerate a minimal S3 Tables entry
(`FederatedTable` set and `format=ICEBERG`) so `DropTable` works as cleanup, or
document the out-of-band cleanup for the double-failure case. wdyt?
##########
catalog/glue/glue.go:
##########
@@ -307,6 +333,113 @@ func (c *Catalog) CreateTable(ctx context.Context,
identifier table.Identifier,
return c.LoadTable(ctx, identifier)
}
+// isS3TablesDatabase reports whether the Glue database is federated to the
+// Amazon S3 Tables service, which owns table storage and location assignment.
+func (c *Catalog) isS3TablesDatabase(ctx context.Context, database string)
(bool, error) {
+ db, err := c.getDatabase(ctx, database)
+ if err != nil {
+ // A missing database is not fatal here; let the generic create
path
+ // surface it, so this probe never changes the error a caller
already saw.
+ if errors.Is(err, catalog.ErrNoSuchNamespace) {
+ return false, nil
+ }
+
+ return false, err
+ }
+
+ return db.FederatedDatabase != nil &&
+
strings.EqualFold(aws.ToString(db.FederatedDatabase.ConnectionType),
s3TablesConnectionType), nil
+}
+
+// createS3TablesTable creates a table in an S3 Tables federated database. The
+// service assigns storage, so a minimal entry is created first to allocate the
+// location, then updated with the written metadata pointer; on any later
+// failure the minimal entry is removed so no half-created table is left
behind.
+func (c *Catalog) createS3TablesTable(ctx context.Context, database, tableName
string, identifier table.Identifier, schema *iceberg.Schema, opts
...catalog.CreateTableOpt) (*table.Table, error) {
+ _, err := c.glueSvc.CreateTable(ctx, &glue.CreateTableInput{
+ CatalogId: c.catalogId,
+ DatabaseName: aws.String(database),
+ TableInput: &types.TableInput{
+ Name: aws.String(tableName),
+ Parameters: map[string]string{glueParamFormat:
glueTypeIceberg},
+ },
+ })
+ if err != nil {
+ return nil, fmt.Errorf("failed to allocate S3 Tables storage
for %s.%s: %w", database, tableName, err)
+ }
+
+ if err := c.commitS3TablesTable(ctx, database, tableName, identifier,
schema, opts...); err != nil {
+ if _, delErr := c.glueSvc.DeleteTable(ctx,
&glue.DeleteTableInput{
Review Comment:
When `UpdateTable` fails after `WriteMetadata` already succeeded, this
rollback deletes the Glue entry but leaves the metadata file in the S3 Tables
managed location. For a regular bucket that's the same orphan we already accept
on the non-federated path, so I'm not treating it as a hard blocker. What makes
me pause for S3 Tables is that the caller has no direct access to the managed
bucket to clean it up, and it's not documented whether deleting the Glue entry
makes S3 Tables reclaim the objects.
I'd at least document that an `UpdateTable` failure leaves metadata behind
and that reclamation is up to S3 Tables, and add a test that exercises the
`UpdateTable`-failure path. If `fs.Remove` on the metadata file before the
`DeleteTable` is cheap here, even better. Thoughts?
##########
catalog/glue/glue_test.go:
##########
@@ -2455,3 +2458,498 @@ func TestTableOperationsRejectEmptyIdentifiers(t
*testing.T) {
require.ErrorIs(t, err, catalog.ErrNoSuchTable)
}
}
+
+func s3TablesTestSchema() *iceberg.Schema {
+ return iceberg.NewSchemaWithIdentifiers(1, []int{1},
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.Int64Type{}, Required: true},
+ iceberg.NestedField{ID: 2, Name: "name", Type:
iceberg.StringType{}, Required: false},
+ )
+}
+
+func federatedDatabaseOutput(connectionType string) *glue.GetDatabaseOutput {
+ db := &types.Database{Name: aws.String("test_database")}
+ if connectionType != "" {
+ db.FederatedDatabase = &types.FederatedDatabase{ConnectionType:
aws.String(connectionType)}
+ }
+
+ return &glue.GetDatabaseOutput{Database: db}
+}
+
+func TestGlueIsS3TablesDatabase(t *testing.T) {
+ tests := []struct {
+ name string
+ connectionType string
+ getErr error
+ want bool
+ wantErr bool
+ }{
+ {name: "federated to s3 tables", connectionType:
"aws:s3tables", want: true},
+ {name: "federated case insensitive", connectionType:
"AWS:S3Tables", want: true},
+ {name: "federated to another source", connectionType:
"aws:redshift", want: false},
+ {name: "not federated", connectionType: "", want: false},
+ {name: "missing database is not federated", getErr:
&types.EntityNotFoundException{}, want: false},
+ {name: "get database error", getErr: errors.New("boom"),
wantErr: true},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ mockGlueSvc := &mockGlueClient{}
+ if tt.getErr != nil {
+ mockGlueSvc.On("GetDatabase", mock.Anything,
&glue.GetDatabaseInput{
+ Name: aws.String("test_database"),
+ },
mock.Anything).Return((*glue.GetDatabaseOutput)(nil), tt.getErr).Once()
+ } else {
+ mockGlueSvc.On("GetDatabase", mock.Anything,
&glue.GetDatabaseInput{
+ Name: aws.String("test_database"),
+ },
mock.Anything).Return(federatedDatabaseOutput(tt.connectionType), nil).Once()
+ }
+
+ cat := &Catalog{glueSvc: mockGlueSvc, awsCfg:
&aws.Config{}}
+ got, err :=
cat.isS3TablesDatabase(context.Background(), "test_database")
+ if tt.wantErr {
+ require.Error(t, err)
+ } else {
+ require.NoError(t, err)
+ require.Equal(t, tt.want, got)
+ }
+ mockGlueSvc.AssertExpectations(t)
+ })
+ }
+}
+
+// TestGlueCreateTableS3TablesFederated exercises the full two-phase create:
+// detect federation, allocate storage with a minimal entry, read the assigned
+// location, write metadata to it, and repoint the Glue entry.
+func TestGlueCreateTableS3TablesFederated(t *testing.T) {
+ ctx := context.Background()
+ managedLocation := "file://" + t.TempDir()
+ schema := s3TablesTestSchema()
+
+ mockGlueSvc := &mockGlueClient{}
+ mockGlueSvc.On("GetDatabase", mock.Anything, &glue.GetDatabaseInput{
+ Name: aws.String("test_database"),
+ }, mock.Anything).Return(federatedDatabaseOutput("aws:s3tables"),
nil).Times(2)
+
+ mockGlueSvc.On("CreateTable", mock.Anything, mock.MatchedBy(func(in
*glue.CreateTableInput) bool {
+ return aws.ToString(in.TableInput.Name) == "test_table" &&
+ in.TableInput.Parameters[glueParamFormat] ==
glueTypeIceberg &&
+ in.TableInput.StorageDescriptor == nil
+ }), mock.Anything).Return(&glue.CreateTableOutput{}, nil).Once()
+
+ allocated := &types.Table{
+ Name: aws.String("test_table"),
+ DatabaseName: aws.String("test_database"),
+ VersionId: aws.String("1"),
+ Parameters: map[string]string{glueParamFormat:
glueTypeIceberg},
+ StorageDescriptor: &types.StorageDescriptor{Location:
aws.String(managedLocation)},
+ }
+ mockGlueSvc.On("GetTable", mock.Anything, &glue.GetTableInput{
+ DatabaseName: aws.String("test_database"),
+ Name: aws.String("test_table"),
+ }, mock.Anything).Return(&glue.GetTableOutput{Table: allocated},
nil).Once()
+
+ // LoadTable at the end reads this entry; UpdateTable's Run below fills
in the
+ // iceberg parameters (including the metadata pointer) before it is
read.
+ loaded := &types.Table{
+ Name: aws.String("test_table"),
+ DatabaseName: aws.String("test_database"),
+ // S3 Tables reports its own service TableType plus a
FederatedTable marker,
+ // exercising the relaxed getRawTable gate on the create
success path.
+ TableType: aws.String("customer"),
+ FederatedTable: &types.FederatedTable{ConnectionType:
aws.String(s3TablesConnectionType)},
+ Parameters: map[string]string{},
+ StorageDescriptor: &types.StorageDescriptor{Location:
aws.String(managedLocation)},
+ }
+ var capturedMetadataLocation string
+ mockGlueSvc.On("UpdateTable", mock.Anything, mock.MatchedBy(func(in
*glue.UpdateTableInput) bool {
+ return in.TableInput != nil && aws.ToString(in.VersionId) ==
"1" &&
+ in.TableInput.Parameters[tableParamTableType] ==
glueTypeIceberg
+ }), mock.Anything).Run(func(args mock.Arguments) {
+ in := args.Get(1).(*glue.UpdateTableInput)
+ capturedMetadataLocation =
in.TableInput.Parameters[tableParamMetadataLocation]
+ loaded.Parameters[tableParamTableType] = glueTypeIceberg
+ loaded.Parameters[tableParamMetadataLocation] =
capturedMetadataLocation
+ }).Return(&glue.UpdateTableOutput{}, nil).Once()
+
+ mockGlueSvc.On("GetTable", mock.Anything, &glue.GetTableInput{
+ DatabaseName: aws.String("test_database"),
+ Name: aws.String("test_table"),
+ }, mock.Anything).Return(&glue.GetTableOutput{Table: loaded}, nil)
+
+ cat := &Catalog{glueSvc: mockGlueSvc, awsCfg: &aws.Config{}}
+ tbl, err := cat.CreateTable(ctx, TableIdentifier("test_database",
"test_table"), schema)
+ require.NoError(t, err)
+ require.Equal(t, TableIdentifier("test_database", "test_table"),
tbl.Identifier())
+ require.Equal(t, schema.Fields(), tbl.Schema().Fields())
+ require.Contains(t, tbl.MetadataLocation(), managedLocation)
+ require.Equal(t, capturedMetadataLocation, tbl.MetadataLocation())
+ require.FileExists(t, strings.TrimPrefix(capturedMetadataLocation,
"file://"))
+ mockGlueSvc.AssertNotCalled(t, "DeleteTable", mock.Anything,
mock.Anything, mock.Anything)
+ mockGlueSvc.AssertExpectations(t)
+}
+
+// TestGlueCreateTableS3TablesCleanupOnFailure verifies the allocated entry is
+// deleted when the second phase fails, leaving no half-created table behind.
+func TestGlueCreateTableS3TablesCleanupOnFailure(t *testing.T) {
+ ctx := context.Background()
+ schema := s3TablesTestSchema()
+
+ mockGlueSvc := &mockGlueClient{}
+ mockGlueSvc.On("GetDatabase", mock.Anything, &glue.GetDatabaseInput{
+ Name: aws.String("test_database"),
+ }, mock.Anything).Return(federatedDatabaseOutput("aws:s3tables"),
nil).Times(2)
+ mockGlueSvc.On("CreateTable", mock.Anything, mock.Anything,
mock.Anything).
+ Return(&glue.CreateTableOutput{}, nil).Once()
+ mockGlueSvc.On("GetTable", mock.Anything, &glue.GetTableInput{
+ DatabaseName: aws.String("test_database"),
+ Name: aws.String("test_table"),
+ }, mock.Anything).Return(&glue.GetTableOutput{Table: &types.Table{
+ Name: aws.String("test_table"),
+ DatabaseName: aws.String("test_database"),
+ StorageDescriptor: &types.StorageDescriptor{Location:
aws.String("")},
+ }}, nil).Once()
+ mockGlueSvc.On("DeleteTable", mock.Anything, &glue.DeleteTableInput{
+ DatabaseName: aws.String("test_database"),
+ Name: aws.String("test_table"),
+ }, mock.Anything).Return(&glue.DeleteTableOutput{}, nil).Once()
+
+ cat := &Catalog{glueSvc: mockGlueSvc, awsCfg: &aws.Config{}}
+ _, err := cat.CreateTable(ctx, TableIdentifier("test_database",
"test_table"), schema)
+ require.ErrorContains(t, err, "did not assign a storage location")
+ mockGlueSvc.AssertExpectations(t)
+}
+
+// TestGlueCreateTableS3TablesCleanupErrorWrapped surfaces both the original
+// failure and the cleanup failure when deleting the allocated entry also
fails.
+func TestGlueCreateTableS3TablesCleanupErrorWrapped(t *testing.T) {
+ ctx := context.Background()
+ schema := s3TablesTestSchema()
+
+ mockGlueSvc := &mockGlueClient{}
+ mockGlueSvc.On("GetDatabase", mock.Anything, &glue.GetDatabaseInput{
+ Name: aws.String("test_database"),
+ }, mock.Anything).Return(federatedDatabaseOutput("aws:s3tables"),
nil).Times(2)
+ mockGlueSvc.On("CreateTable", mock.Anything, mock.Anything,
mock.Anything).
+ Return(&glue.CreateTableOutput{}, nil).Once()
+ mockGlueSvc.On("GetTable", mock.Anything, mock.Anything, mock.Anything).
+ Return((*glue.GetTableOutput)(nil), errors.New("get
boom")).Once()
+ mockGlueSvc.On("DeleteTable", mock.Anything, mock.Anything,
mock.Anything).
+ Return((*glue.DeleteTableOutput)(nil), errors.New("delete
boom")).Once()
+
+ cat := &Catalog{glueSvc: mockGlueSvc, awsCfg: &aws.Config{}}
+ _, err := cat.CreateTable(ctx, TableIdentifier("test_database",
"test_table"), schema)
+ require.ErrorContains(t, err, "get boom")
+ require.ErrorContains(t, err, "failed to clean up allocated table")
+ require.ErrorContains(t, err, "delete boom")
+ mockGlueSvc.AssertExpectations(t)
+}
+
+// TestGlueCreateTableS3TablesAllocateError returns early without cleanup when
+// the initial allocation call itself fails.
+func TestGlueCreateTableS3TablesAllocateError(t *testing.T) {
+ ctx := context.Background()
+ schema := s3TablesTestSchema()
+
+ mockGlueSvc := &mockGlueClient{}
+ mockGlueSvc.On("GetDatabase", mock.Anything, &glue.GetDatabaseInput{
+ Name: aws.String("test_database"),
+ }, mock.Anything).Return(federatedDatabaseOutput("aws:s3tables"),
nil).Times(2)
+ mockGlueSvc.On("CreateTable", mock.Anything, mock.Anything,
mock.Anything).
+ Return((*glue.CreateTableOutput)(nil), errors.New("allocate
boom")).Once()
+
+ cat := &Catalog{glueSvc: mockGlueSvc, awsCfg: &aws.Config{}}
+ _, err := cat.CreateTable(ctx, TableIdentifier("test_database",
"test_table"), schema)
+ require.ErrorContains(t, err, "failed to allocate S3 Tables storage")
+ mockGlueSvc.AssertNotCalled(t, "GetTable", mock.Anything,
mock.Anything, mock.Anything)
+ mockGlueSvc.AssertNotCalled(t, "DeleteTable", mock.Anything,
mock.Anything, mock.Anything)
+ mockGlueSvc.AssertExpectations(t)
+}
+
+// TestGlueCreateTableExplicitLocationSkipsFederationProbe verifies a create
with
+// an explicit location goes straight through the generic path without probing
for
+// federation, so such creates never newly require glue:GetDatabase.
+func TestGlueCreateTableExplicitLocationSkipsFederationProbe(t *testing.T) {
+ ctx := context.Background()
+ location := "file://" + t.TempDir()
+ schema := s3TablesTestSchema()
+
+ mockGlueSvc := &mockGlueClient{}
+ mockGlueSvc.On("CreateTable", mock.Anything, mock.Anything,
mock.Anything).
+ Return(&glue.CreateTableOutput{}, nil).Once()
+ // The trailing reload is not the point here; fail it fast to avoid a
full
+ // metadata round-trip. What matters is that GetDatabase is never
called.
+ mockGlueSvc.On("GetTable", mock.Anything, mock.Anything, mock.Anything).
+ Return((*glue.GetTableOutput)(nil), errors.New("load
boom")).Once()
+
+ cat := &Catalog{glueSvc: mockGlueSvc, awsCfg: &aws.Config{}}
+ _, err := cat.CreateTable(ctx, TableIdentifier("test_database",
"test_table"), schema,
+ catalog.WithLocation(location))
+ require.ErrorContains(t, err, "load boom")
+ mockGlueSvc.AssertNotCalled(t, "GetDatabase", mock.Anything,
mock.Anything, mock.Anything)
+ mockGlueSvc.AssertExpectations(t)
+}
+
+// TestGlueCreateTableS3TablesMissingVersionId fails and rolls back when the
+// allocated entry has no Glue version id to commit against.
+func TestGlueCreateTableS3TablesMissingVersionId(t *testing.T) {
+ ctx := context.Background()
+ schema := s3TablesTestSchema()
+
+ mockGlueSvc := &mockGlueClient{}
+ mockGlueSvc.On("GetDatabase", mock.Anything, &glue.GetDatabaseInput{
+ Name: aws.String("test_database"),
+ }, mock.Anything).Return(federatedDatabaseOutput("aws:s3tables"),
nil).Times(2)
+ mockGlueSvc.On("CreateTable", mock.Anything, mock.Anything,
mock.Anything).
+ Return(&glue.CreateTableOutput{}, nil).Once()
+ mockGlueSvc.On("GetTable", mock.Anything, &glue.GetTableInput{
+ DatabaseName: aws.String("test_database"),
+ Name: aws.String("test_table"),
+ }, mock.Anything).Return(&glue.GetTableOutput{Table: &types.Table{
+ Name: aws.String("test_table"),
+ DatabaseName: aws.String("test_database"),
+ StorageDescriptor: &types.StorageDescriptor{Location:
aws.String("file:///tmp/whatever")},
+ }}, nil).Once()
+ mockGlueSvc.On("DeleteTable", mock.Anything, &glue.DeleteTableInput{
+ DatabaseName: aws.String("test_database"),
+ Name: aws.String("test_table"),
+ }, mock.Anything).Return(&glue.DeleteTableOutput{}, nil).Once()
+
+ cat := &Catalog{glueSvc: mockGlueSvc, awsCfg: &aws.Config{}}
+ _, err := cat.CreateTable(ctx, TableIdentifier("test_database",
"test_table"), schema)
+ require.ErrorContains(t, err, "Glue table version id is missing")
+ mockGlueSvc.AssertNotCalled(t, "UpdateTable", mock.Anything,
mock.Anything, mock.Anything)
+ mockGlueSvc.AssertExpectations(t)
+}
+
+// TestGlueCreateTableNonFederatedFallsThrough confirms a non-federated
database
+// with no location still hits the generic path (and its "no default path"
error).
+func TestGlueCreateTableNonFederatedFallsThrough(t *testing.T) {
+ ctx := context.Background()
+ schema := s3TablesTestSchema()
+
+ mockGlueSvc := &mockGlueClient{}
+ mockGlueSvc.On("GetDatabase", mock.Anything, &glue.GetDatabaseInput{
+ Name: aws.String("test_database"),
+ }, mock.Anything).Return(federatedDatabaseOutput(""), nil)
+
+ cat := &Catalog{glueSvc: mockGlueSvc, awsCfg: &aws.Config{}}
+ _, err := cat.CreateTable(ctx, TableIdentifier("test_database",
"test_table"), schema)
+ require.ErrorContains(t, err, "no default path set")
+ mockGlueSvc.AssertNotCalled(t, "CreateTable", mock.Anything,
mock.Anything, mock.Anything)
+}
+
+// TestGlueCreateTableS3TablesFederatedIntegration creates a real table in an
S3
+// Tables federated catalog through the native Glue path. Gated by env vars:
+//
+// TEST_S3TABLES_CATALOG_ID = <account-id>:s3tablescatalog/<table-bucket>
+// TEST_S3TABLES_DATABASE = a namespace you can create tables in
+func TestGlueCreateTableS3TablesFederatedIntegration(t *testing.T) {
+ catalogID := os.Getenv("TEST_S3TABLES_CATALOG_ID")
+ dbName := os.Getenv("TEST_S3TABLES_DATABASE")
+ if catalogID == "" || dbName == "" {
+ t.Skip()
+ }
+ assert := require.New(t)
+ ctx := context.Background()
+ awsCfg, err := config.LoadDefaultConfig(ctx)
+ assert.NoError(err)
+ ctlg, err := NewCatalog(WithAwsConfig(awsCfg),
WithAwsProperties(AwsProperties{CatalogIdKey: catalogID}))
+ assert.NoError(err)
+
+ tableName := fmt.Sprintf("it_%d", time.Now().UnixNano())
+ ident := TableIdentifier(dbName, tableName)
+ schema := s3TablesTestSchema()
+
+ tbl, err := ctlg.CreateTable(ctx, ident, schema)
+ assert.NoError(err)
+ defer func() { assert.NoError(ctlg.DropTable(ctx, ident)) }()
+
+ assert.Equal(ident, tbl.Identifier())
+ assert.Equal(schema.Fields(), tbl.Schema().Fields())
+ assert.Contains(tbl.MetadataLocation(), "--table-s3", "metadata must
land in the S3 Tables managed location")
+
+ reloaded, err := ctlg.LoadTable(ctx, ident)
+ assert.NoError(err)
+ assert.Equal(schema.Fields(), reloaded.Schema().Fields())
+ assert.Equal(tbl.MetadataLocation(), reloaded.MetadataLocation())
+}
+
+// TestGlueGetRawTableTableType covers the relaxed TableType gate: standard
+// EXTERNAL_TABLE tables and S3 Tables federated iceberg tables (whose Glue
+// TableType is service-specific) are both accepted, while non-iceberg entries
+// with an unexpected TableType are still rejected.
+func TestGlueGetRawTableTableType(t *testing.T) {
+ tests := []struct {
+ name string
+ tableType string
+ params map[string]string
+ federated bool
+ wantErr bool
+ }{
+ {
+ name: "standard external table",
+ tableType: glueTableType,
+ params: map[string]string{tableParamTableType:
glueTypeIceberg},
+ },
+ {
+ name: "s3 tables federated iceberg",
+ tableType: "customer",
+ params: map[string]string{tableParamTableType:
glueTypeIceberg},
+ federated: true,
+ },
+ {
+ name: "s3 tables federated renaming",
+ tableType: "customer",
+ params: map[string]string{tableParamTableType:
glueTypeIcebergRenaming},
+ federated: true,
+ },
+ {
+ name: "non-federated iceberg param unexpected type
rejected",
+ tableType: "customer",
+ params: map[string]string{tableParamTableType:
glueTypeIceberg},
+ wantErr: true,
+ },
+ {
+ name: "non-iceberg unexpected type rejected",
+ tableType: "VIRTUAL_VIEW",
+ params: map[string]string{},
+ wantErr: true,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ mockGlueSvc := &mockGlueClient{}
+ glueTable := &types.Table{
+ Name: aws.String("test_table"),
+ DatabaseName: aws.String("test_database"),
+ TableType: aws.String(tt.tableType),
+ Parameters: tt.params,
+ }
+ if tt.federated {
+ glueTable.FederatedTable =
&types.FederatedTable{ConnectionType: aws.String(s3TablesConnectionType)}
+ }
+ mockGlueSvc.On("GetTable", mock.Anything,
&glue.GetTableInput{
+ DatabaseName: aws.String("test_database"),
+ Name: aws.String("test_table"),
+ }, mock.Anything).Return(&glue.GetTableOutput{Table:
glueTable}, nil).Once()
+
+ cat := &Catalog{glueSvc: mockGlueSvc, awsCfg:
&aws.Config{}}
+ tbl, err := cat.getRawTable(context.Background(),
"test_database", "test_table")
+ if tt.wantErr {
+ require.ErrorContains(t, err, "is not an
EXTERNAL_TABLE")
+ } else {
+ require.NoError(t, err)
+ require.Equal(t, tt.tableType,
aws.ToString(tbl.TableType))
+ }
+ mockGlueSvc.AssertExpectations(t)
+ })
+ }
+}
+
+// TestGlueCreateTableS3TablesRollbackOnMetadataWriteFailure covers the
mid-flight
+// case where storage is allocated but writing metadata to it fails: the
+// allocated entry must be rolled back and UpdateTable never issued.
+func TestGlueCreateTableS3TablesRollbackOnMetadataWriteFailure(t *testing.T) {
+ ctx := context.Background()
+ schema := s3TablesTestSchema()
+
+ mockGlueSvc := &mockGlueClient{}
+ mockGlueSvc.On("GetDatabase", mock.Anything, &glue.GetDatabaseInput{
+ Name: aws.String("test_database"),
+ }, mock.Anything).Return(federatedDatabaseOutput("aws:s3tables"),
nil).Times(2)
+ mockGlueSvc.On("CreateTable", mock.Anything, mock.Anything,
mock.Anything).
+ Return(&glue.CreateTableOutput{}, nil).Once()
+ mockGlueSvc.On("GetTable", mock.Anything, &glue.GetTableInput{
+ DatabaseName: aws.String("test_database"),
+ Name: aws.String("test_table"),
+ }, mock.Anything).Return(&glue.GetTableOutput{Table: &types.Table{
+ Name: aws.String("test_table"),
+ DatabaseName: aws.String("test_database"),
+ VersionId: aws.String("1"),
+ StorageDescriptor: &types.StorageDescriptor{Location:
aws.String("s3://nonexistent-test-bucket")},
+ }}, nil).Once()
+ mockGlueSvc.On("DeleteTable", mock.Anything, &glue.DeleteTableInput{
+ DatabaseName: aws.String("test_database"),
+ Name: aws.String("test_table"),
+ }, mock.Anything).Return(&glue.DeleteTableOutput{}, nil).Once()
+
+ cat := &Catalog{glueSvc: mockGlueSvc, awsCfg: &aws.Config{}}
+ _, err := cat.CreateTable(ctx, TableIdentifier("test_database",
"test_table"), schema)
+ require.Error(t, err)
Review Comment:
`require.Error` here is the weak spot: it passes for any failure, so it
doesn't prove we failed on the metadata write rather than somewhere else. The
`UpdateTable` `AssertNotCalled` below is good and pins the ordering.
I'd tighten this to `require.ErrorContains` on the write failure so the test
actually proves the mid-flight rollback case it's named for.
--
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]
