laskoviymishka commented on code in PR #2001:
URL: https://github.com/apache/iceberg-go/pull/2001#discussion_r4020373123
##########
catalog/glue/glue.go:
##########
@@ -307,6 +351,125 @@ 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 {
+ // Best-effort: a missing database or a caller lacking
glue:GetDatabase
+ // is treated as "not federated" so the generic create path can
proceed.
+ var apiErr smithy.APIError
+ if errors.Is(err, catalog.ErrNoSuchNamespace) ||
+ (errors.As(err, &apiErr) && apiErr.ErrorCode() ==
"AccessDeniedException") {
+ return false, nil
+ }
+
+ return false, err
+ }
+
+ return db.FederatedDatabase != nil &&
Review Comment:
`getDatabase`'s result isn't nil-guarded before we read
`db.FederatedDatabase`, so if it ever hands back a success with a nil
`Database` (a mock returning `&glue.GetDatabaseOutput{}` does exactly that)
this line panics. Since `isS3TablesDatabase` now runs on every
explicit-location create too, I'd add `if db == nil { return false, nil }`
before the dereference.
##########
catalog/glue/glue.go:
##########
@@ -307,6 +351,125 @@ 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 {
+ // Best-effort: a missing database or a caller lacking
glue:GetDatabase
+ // is treated as "not federated" so the generic create path can
proceed.
+ var apiErr smithy.APIError
+ if errors.Is(err, catalog.ErrNoSuchNamespace) ||
+ (errors.As(err, &apiErr) && apiErr.ErrorCode() ==
"AccessDeniedException") {
+ 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.
+// On a commit failure the minimal Glue entry is rolled back and the written
+// metadata object is best-effort deleted; if that delete fails (the managed
+// location may be unreachable) reclaiming it is left to S3 Tables. A failed
+// rollback leaves a minimal entry to clear out of band, matching pyiceberg.
+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)
+ }
+
+ return nil, err
+ }
+
+ // The table is committed; load it outside the rollback scope so a
transient
+ // read failure does not delete an already-created table.
+ return c.LoadTable(ctx, identifier)
+}
+
+// commitS3TablesTable reads the service-assigned location, writes the Iceberg
+// metadata to it, and points the Glue entry at that metadata. It does not
reload
+// the table: the caller does that only after a successful commit, so a read
+// failure never triggers a rollback of an already-created table.
+func (c *Catalog) commitS3TablesTable(ctx context.Context, database, tableName
string, identifier table.Identifier, schema *iceberg.Schema, opts
...catalog.CreateTableOpt) error {
+ allocated, err := c.glueSvc.GetTable(ctx, &glue.GetTableInput{
+ CatalogId: c.catalogId,
+ DatabaseName: aws.String(database),
+ Name: aws.String(tableName),
+ })
+ if err != nil {
+ return fmt.Errorf("failed to load allocated S3 Tables table
%s.%s: %w", database, tableName, err)
+ }
+ if allocated == nil || allocated.Table == nil ||
allocated.Table.StorageDescriptor == nil {
+ return fmt.Errorf("S3 Tables did not return a storage
descriptor for %s.%s", database, tableName)
+ }
+ managedLocation :=
aws.ToString(allocated.Table.StorageDescriptor.Location)
+ if managedLocation == "" {
+ return fmt.Errorf("S3 Tables did not assign a storage location
for %s.%s", database, tableName)
+ }
+ if allocated.Table.VersionId == nil {
+ return fmt.Errorf("cannot commit table %s.%s: because Glue
table version id is missing", database, tableName)
+ }
+
+ // Copy rather than append onto the caller's opts, whose backing array
may
+ // have spare capacity we would otherwise clobber.
+ stagedOpts := make([]catalog.CreateTableOpt, len(opts), len(opts)+1)
+ copy(stagedOpts, opts)
+ stagedOpts = append(stagedOpts, catalog.WithLocation(managedLocation))
+
+ staged, err := internal.CreateStagedTable(ctx, c.props,
c.LoadNamespaceProperties, identifier, schema, stagedOpts...)
+ if err != nil {
+ return err
+ }
+
+ if err := internal.WriteMetadata(ctx, staged.Table); err != nil {
+ return err
+ }
+
+ // constructTableInput sends TableType=EXTERNAL_TABLE; S3 Tables keeps
its own
+ // service type (e.g. "customer") on read, which getRawTable accepts.
+ _, err = c.glueSvc.UpdateTable(ctx, &glue.UpdateTableInput{
+ CatalogId: c.catalogId,
+ DatabaseName: aws.String(database),
+ TableInput: constructTableInput(tableName, staged.Table,
allocated.Table),
Review Comment:
the comment above says S3 Tables keeps its own service type on read, but
`constructTableInput` always sets `TableType=EXTERNAL_TABLE` and that's what we
send on this repoint. Do we actually know S3 Tables accepts `EXTERNAL_TABLE` on
write? If it validates and rejects the type, every S3 Tables create fails here
and rolls back, so the feature is silently broken for all callers.
Nothing in the unit tests catches this: the `UpdateTable` `MatchedBy` only
checks `VersionId` and the `table_type` param, not `TableInput.TableType`, so
the assumption rides entirely on the gated integration test.
I'd pass the allocated entry's own `TableType` through here instead of
forcing `EXTERNAL_TABLE`, or at minimum assert the `TableType` we send in a
unit test and cite the AWS contract in the comment. wdyt?
##########
catalog/glue/glue.go:
##########
@@ -307,6 +351,125 @@ 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 {
+ // Best-effort: a missing database or a caller lacking
glue:GetDatabase
+ // is treated as "not federated" so the generic create path can
proceed.
+ var apiErr smithy.APIError
+ if errors.Is(err, catalog.ErrNoSuchNamespace) ||
+ (errors.As(err, &apiErr) && apiErr.ErrorCode() ==
"AccessDeniedException") {
+ 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.
+// On a commit failure the minimal Glue entry is rolled back and the written
+// metadata object is best-effort deleted; if that delete fails (the managed
+// location may be unreachable) reclaiming it is left to S3 Tables. A failed
+// rollback leaves a minimal entry to clear out of band, matching pyiceberg.
Review Comment:
this is the orphan case I flagged last round, and I don't think it's
resolved. The comment now says the stranded entry is cleared out of band
"matching pyiceberg", but that equivalence doesn't hold: pyiceberg's
`drop_table` has no `table_type` guard and calls `delete_table` directly, so it
can remove the minimal entry. Ours can't.
The minimal entry only carries `Parameters{format: ICEBERG}` with no
`table_type`. So when `DropTable` runs `getRawTable`, `isFederatedIceberg` is
`federated && isIceberg`, and `isIceberg` comes from `tableParamTableType`,
which is empty here. The guard fires, `DropTable` returns "is not an
EXTERNAL_TABLE", and after a double failure the entry is invisible to
`ListTables` and undroppable through the public API. The only recovery is a raw
Glue call.
I'd either teach `getRawTable`/`DropTable` to accept a federated S3 Tables
entry whose `format` param is `ICEBERG` even when `table_type` is absent, or,
if we're consciously deferring cleanup, drop the "matching pyiceberg" line and
say plainly that the orphan needs an out-of-band delete. Either is fine, it
just shouldn't claim a parity it doesn't have. wdyt?
##########
catalog/glue/glue.go:
##########
@@ -307,6 +351,125 @@ 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 {
+ // Best-effort: a missing database or a caller lacking
glue:GetDatabase
+ // is treated as "not federated" so the generic create path can
proceed.
+ var apiErr smithy.APIError
+ if errors.Is(err, catalog.ErrNoSuchNamespace) ||
+ (errors.As(err, &apiErr) && apiErr.ErrorCode() ==
"AccessDeniedException") {
+ 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.
+// On a commit failure the minimal Glue entry is rolled back and the written
+// metadata object is best-effort deleted; if that delete fails (the managed
+// location may be unreachable) reclaiming it is left to S3 Tables. A failed
+// rollback leaves a minimal entry to clear out of band, matching pyiceberg.
+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:
the rollback `DeleteTable` reuses `ctx`, the same context that just failed
the commit. If that failure was a cancellation, `ctx.Done()` is already closed
and `DeleteTable` returns immediately with `context.Canceled`, so the minimal
entry leaks, and S3 Tables has per-account table limits.
`rollbackRenameDestination` in this same file already handles the identical
shape with `context.WithoutCancel(ctx)` plus `renameCleanupTimeout`; I'd follow
that here.
While we're in here, the combined error uses the parenthetical `%w (...%w)`
form, but everywhere else in this file the two-error case is `errors.Join`. I'd
switch to `errors.Join(err, fmt.Errorf(...))` for consistency.
--
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]
