zeroshade commented on code in PR #2535: URL: https://github.com/apache/iceberg-go/pull/2535#discussion_r4542900879
########## catalog/dynamodb/tables.go: ########## @@ -0,0 +1,378 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package dynamodb + +import ( + "context" + "errors" + "fmt" + "iter" + "log" + "maps" + "slices" + "strings" + "time" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/catalog" + "github.com/apache/iceberg-go/catalog/internal" + "github.com/apache/iceberg-go/table" + "github.com/apache/iceberg-go/table/maintenance" + "github.com/apache/iceberg-go/utils" + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/dynamodb" + "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" +) + +func (c *Catalog) loadTableEntry(ctx context.Context, ident table.Identifier) (item, error) { + if err := validateIdentifier(ident); err != nil { + return nil, err + } + entry, err := c.get(ctx, tableKey(ident)) + if err != nil { + return nil, err + } + if len(entry) == 0 || !strings.EqualFold(attrString(entry["p.table_type"]), "ICEBERG") { + return nil, fmt.Errorf("%w: %s", catalog.ErrNoSuchTable, ident) + } + if attrString(entry[metadataLocationProp]) == "" || attrString(entry[versionCol]) == "" { + return nil, errors.New("DynamoDB table metadata location or version is missing") + } + + return entry, nil +} + +func (c *Catalog) loadTable(ctx context.Context, ident table.Identifier, entry item) (*table.Table, error) { + location := attrString(entry[metadataLocationProp]) + reporter, err := c.reporter.Get(c.props) + if err != nil { + return nil, err + } + + return table.NewFromLocation(ctx, ident, location, c.fileSystem(c.props, location), c, + table.WithMetricsReporter(reporter), table.WithScanPlanningIOProperties(c.props)) +} + +func (c *Catalog) LoadTable(ctx context.Context, ident table.Identifier) (*table.Table, error) { + entry, err := c.loadTableEntry(ctx, ident) + if err != nil { + return nil, err + } + + return c.loadTable(ctx, ident, entry) +} + +func (c *Catalog) CheckTableExists(ctx context.Context, ident table.Identifier) (bool, error) { + _, err := c.loadTableEntry(ctx, ident) + if errors.Is(err, catalog.ErrNoSuchTable) { + return false, nil + } + + return err == nil, err +} + +func (c *Catalog) namespaceLocationProperties(ctx context.Context, ns table.Identifier) (iceberg.Properties, error) { + props, err := c.LoadNamespaceProperties(ctx, ns) + if err != nil { + return nil, err + } + // Java DynamoDbCatalog uses default_location, rather than the location + // namespace property used by the generic catalog staging helper. + props["location"] = props[DefaultLocationKey] + + return props, nil +} + +func (c *Catalog) CreateTable(ctx context.Context, ident table.Identifier, schema *iceberg.Schema, opts ...catalog.CreateTableOpt) (*table.Table, error) { + if err := validateIdentifier(ident); err != nil { + return nil, err + } + if _, err := c.loadNamespace(ctx, catalog.NamespaceFromIdent(ident)); err != nil { + return nil, err + } + staged, err := internal.CreateStagedTable(ctx, c.props, c.namespaceLocationProperties, ident, schema, opts...) + if err != nil { + return nil, err + } + if err := internal.WriteMetadata(utils.WithAWSConfig(ctx, &c.awsConfig), staged.Table); err != nil { + return nil, err + } + if err := c.createTableEntry(ctx, ident, staged.MetadataLocation()); err != nil { + return nil, err + } + + return c.LoadTable(ctx, ident) +} + +func newTableEntry(ident table.Identifier, location string) item { + entry := tableKey(ident) + entry["p.table_type"] = stringAttr("ICEBERG") + entry[metadataLocationProp] = stringAttr(location) + setEntryMetadata(entry, true) + + return entry +} + +func (c *Catalog) createTableEntry(ctx context.Context, ident table.Identifier, location string) error { + entry := newTableEntry(ident, location) + ns := catalog.NamespaceFromIdent(ident) + for range 6 { + namespace, err := c.loadNamespace(ctx, ns) + if err != nil { + return err + } + err = c.transaction(ctx, []types.TransactWriteItem{ + {Put: &types.Put{ + TableName: aws.String(c.tableName), Item: entry, + ConditionExpression: aws.String("attribute_not_exists(#id)"), ExpressionAttributeNames: map[string]string{"#id": identifierCol}, + }}, + c.touchNamespace(ns, namespace), + }) + if err == nil { + return nil + } + checkCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + current, checkErr := c.get(checkCtx, tableKey(ident)) + cancel() + if checkErr == nil && attrString(current[versionCol]) == attrString(entry[versionCol]) { + return nil + } + if !conditionFailed(err) && !transactionConflict(err) { + return err + } + if len(current) != 0 { + return fmt.Errorf("%w: %s", catalog.ErrTableAlreadyExists, ident) + } + } + + return fmt.Errorf("%w: namespace changed while creating table", table.ErrCommitFailed) +} + +func (c *Catalog) RegisterTable(ctx context.Context, ident table.Identifier, location string) (*table.Table, error) { + if err := validateIdentifier(ident); err != nil { + return nil, err + } + if _, err := c.loadNamespace(ctx, catalog.NamespaceFromIdent(ident)); err != nil { + return nil, err + } + if _, err := table.NewFromLocation(ctx, ident, location, c.fileSystem(c.props, location), c); err != nil { + return nil, err + } + if err := c.createTableEntry(ctx, ident, location); err != nil { + return nil, err + } + + return c.LoadTable(ctx, ident) +} + +func (c *Catalog) ListTables(ctx context.Context, ns table.Identifier) iter.Seq2[table.Identifier, error] { + return func(yield func(table.Identifier, error) bool) { + if _, err := c.loadNamespace(ctx, ns); err != nil { + yield(nil, err) + + return + } + paginator := dynamodb.NewQueryPaginator(c.client, &dynamodb.QueryInput{ + TableName: aws.String(c.tableName), IndexName: aws.String(namespaceIndex), + KeyConditionExpression: aws.String("#ns = :ns"), ExpressionAttributeNames: map[string]string{"#ns": namespaceCol}, + ExpressionAttributeValues: item{":ns": stringAttr(strings.Join(ns, "."))}, + }) + for paginator.HasMorePages() { + page, err := paginator.NextPage(ctx) + if err != nil { + yield(nil, err) + + return + } + for _, entry := range page.Items { + identifier := attrString(entry[identifierCol]) + if identifier == namespaceIdentifier { + continue + } + ident := strings.Split(identifier, ".") + if err := validateIdentifier(ident); err != nil { + yield(nil, err) + + return + } + if !yield(ident, nil) { + return + } + } + } + } +} + +func (c *Catalog) CommitTable(ctx context.Context, ident table.Identifier, requirements []table.Requirement, updates []table.Update) (table.Metadata, string, error) { + entry, err := c.loadTableEntry(ctx, ident) + if err != nil && !errors.Is(err, catalog.ErrNoSuchTable) { + return nil, "", err + } + var current *table.Table + if len(entry) != 0 { + current, err = c.loadTable(ctx, ident, entry) + if err != nil { + return nil, "", err + } + } + staged, err := internal.UpdateAndStageTable(ctx, c.props, current, ident, 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(utils.WithAWSConfig(ctx, &c.awsConfig), staged.Table); err != nil { + return nil, "", err + } + if current == nil { + if err := c.createTableEntry(ctx, ident, staged.MetadataLocation()); err != nil { + return nil, "", err + } + } else { + updated := maps.Clone(entry) + updated["p.previous_metadata_location"] = entry[metadataLocationProp] + updated[metadataLocationProp] = stringAttr(staged.MetadataLocation()) + setEntryMetadata(updated, false) + err := c.transaction(ctx, []types.TransactWriteItem{{Put: &types.Put{ + TableName: aws.String(c.tableName), Item: updated, ConditionExpression: aws.String("#v = :base"), + ExpressionAttributeNames: map[string]string{"#v": versionCol}, ExpressionAttributeValues: item{":base": entry[versionCol]}, + }}}) + if err != nil && !c.commitSucceeded(ctx, ident, staged.MetadataLocation()) { + if conditionFailed(err) || transactionConflict(err) { + err = fmt.Errorf("%w: table metadata changed: %w", table.ErrCommitFailed, err) + } + + // Retain the file on errors: a lost response can follow a successful Review Comment: This keeps the new metadata file on every error, including a confirmed lost race. Java's [`DynamoDbTableOperations.doCommit`](https://github.com/apache/iceberg/blob/014b3ce5ffbf7c1c51dabb144f4d94e45addcd88/aws/src/main/java/org/apache/iceberg/aws/dynamodb/DynamoDbTableOperations.java#L104-L163) deletes it when the status is `FAILURE`, which includes a `ConditionalCheckFailedException` that `checkCommitStatus` doesn't find committed. The SQL catalog does the same with `removeUncommittedMetadata`. Since `ClientRequestToken` makes SDK retries idempotent, a condition failure combined with `commitSucceeded` returning false means the file was never published, so nothing can refer to it. With the default `commit.retry.num-retries=4`, every commit that loses a race can leave up to five metadata files behind otherwise. Please remove the file in the `conditionFailed(err) || transactionConflict(err)` branch, and keep it only when the outcome is unknown (any other error, or when `commitSucceeded` couldn't load the table). ########## catalog/dynamodb/namespaces.go: ########## @@ -0,0 +1,251 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package dynamodb + +import ( + "context" + "errors" + "fmt" + "maps" + "slices" + "strconv" + "strings" + "time" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/catalog" + "github.com/apache/iceberg-go/catalog/internal" + "github.com/apache/iceberg-go/table" + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/dynamodb" + "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" + "github.com/google/uuid" +) + +func (c *Catalog) loadNamespace(ctx context.Context, ns table.Identifier) (item, error) { + if err := validateNamespace(ns, false); err != nil { + return nil, err + } + entry, err := c.get(ctx, namespaceKey(ns)) + if err != nil { + return nil, err + } + if len(entry) == 0 { + return nil, fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace, ns) + } + if attrString(entry[versionCol]) == "" { + return nil, errors.New("DynamoDB namespace version is missing") + } + + return entry, nil +} + +func (c *Catalog) CheckNamespaceExists(ctx context.Context, ns table.Identifier) (bool, error) { + _, err := c.loadNamespace(ctx, ns) + if errors.Is(err, catalog.ErrNoSuchNamespace) { + return false, nil + } + + return err == nil, err +} + +func (c *Catalog) LoadNamespaceProperties(ctx context.Context, ns table.Identifier) (iceberg.Properties, error) { + entry, err := c.loadNamespace(ctx, ns) + if err != nil { + return nil, err + } + + return properties(entry), nil +} + +func (c *Catalog) CreateNamespace(ctx context.Context, ns table.Identifier, props iceberg.Properties) error { + if err := validateNamespace(ns, false); err != nil { + return err + } + entry := namespaceKey(ns) + for key, value := range props { + entry["p."+key] = stringAttr(value) + } + setEntryMetadata(entry, true) + for range 6 { + writes := []types.TransactWriteItem{{Put: &types.Put{ + TableName: aws.String(c.tableName), Item: entry, ConditionExpression: aws.String("attribute_not_exists(#id)"), + ExpressionAttributeNames: map[string]string{"#id": identifierCol}, + }}} + if len(ns) > 1 { + parent := ns[:len(ns)-1] + current, err := c.loadNamespace(ctx, parent) + if err != nil { + return err + } + writes = append(writes, c.touchNamespace(parent, current)) + } + err := c.transaction(ctx, writes) + if err == nil { + return nil + } + checkCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + current, checkErr := c.get(checkCtx, namespaceKey(ns)) + cancel() + if checkErr == nil && attrString(current[versionCol]) == attrString(entry[versionCol]) { + return nil + } + if !conditionFailed(err) && !transactionConflict(err) { + return err + } + if len(current) != 0 { + return fmt.Errorf("%w: %s", catalog.ErrNamespaceAlreadyExists, ns) + } + } + + return fmt.Errorf("%w: namespace parent changed during creation", table.ErrCommitFailed) +} + +func (c *Catalog) ListNamespaces(ctx context.Context, parent table.Identifier) ([]table.Identifier, error) { + if err := validateNamespace(parent, true); err != nil { + return nil, err + } + if len(parent) != 0 { + if _, err := c.loadNamespace(ctx, parent); err != nil { + return nil, err + } + } + prefix := strings.Join(parent, ".") + if prefix != "" { + prefix += "." + } + input := &dynamodb.QueryInput{ + TableName: aws.String(c.tableName), ConsistentRead: aws.Bool(true), + KeyConditionExpression: aws.String("#id = :id"), + ExpressionAttributeNames: map[string]string{"#id": identifierCol}, + ExpressionAttributeValues: item{":id": stringAttr(namespaceIdentifier)}, + } + if prefix != "" { + input.KeyConditionExpression = aws.String("#id = :id AND begins_with(#ns, :prefix)") + input.ExpressionAttributeNames["#ns"] = namespaceCol + input.ExpressionAttributeValues[":prefix"] = stringAttr(prefix) + } + namespaces := []table.Identifier{} + paginator := dynamodb.NewQueryPaginator(c.client, input) + for paginator.HasMorePages() { + page, err := paginator.NextPage(ctx) + if err != nil { + return nil, err + } + for _, entry := range page.Items { + ns := strings.Split(attrString(entry[namespaceCol]), ".") + if len(ns) == len(parent)+1 && slices.Equal(ns[:len(parent)], parent) { + namespaces = append(namespaces, ns) + } + } + } + + return namespaces, nil +} + +func (c *Catalog) DropNamespace(ctx context.Context, ns table.Identifier) error { + current, err := c.loadNamespace(ctx, ns) + if err != nil { + return err + } + // Check every descendant, including namespaces written by Java without an + // intermediate parent entry. + children, err := c.client.Query(ctx, &dynamodb.QueryInput{ + TableName: aws.String(c.tableName), ConsistentRead: aws.Bool(true), Limit: aws.Int32(1), + KeyConditionExpression: aws.String("#id = :id AND begins_with(#ns, :prefix)"), + ExpressionAttributeNames: map[string]string{"#id": identifierCol, "#ns": namespaceCol}, + ExpressionAttributeValues: item{":id": stringAttr(namespaceIdentifier), ":prefix": stringAttr(strings.Join(ns, ".") + ".")}, + }) + if err != nil { + return err + } + if len(children.Items) != 0 { + return catalog.ErrNamespaceNotEmpty + } + + // GSI reads are eventually consistent. Check emptiness through a consistent + // base-table scan, then guard the delete with the namespace version. Go + // create/register/rename operations update that version in their transaction. + scan := dynamodb.NewScanPaginator(c.client, &dynamodb.ScanInput{ Review Comment: This consistent-read `Scan` reads every item in the catalog table on each successful `DropNamespace`. The filter is applied after the read, so it costs read capacity in proportion to the whole catalog, not to this namespace. Base-table keys don't allow a consistent query by namespace, so I see why a scan is needed for a strict check. Could you first query the `namespace-identifier` GSI with `Limit: 1`, so a namespace that obviously still has tables is rejected without scanning? Please also note the scan cost in the README so operators of large catalogs aren't surprised. ########## catalog/dynamodb/namespaces.go: ########## @@ -0,0 +1,251 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package dynamodb + +import ( + "context" + "errors" + "fmt" + "maps" + "slices" + "strconv" + "strings" + "time" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/catalog" + "github.com/apache/iceberg-go/catalog/internal" + "github.com/apache/iceberg-go/table" + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/dynamodb" + "github.com/aws/aws-sdk-go-v2/service/dynamodb/types" + "github.com/google/uuid" +) + +func (c *Catalog) loadNamespace(ctx context.Context, ns table.Identifier) (item, error) { + if err := validateNamespace(ns, false); err != nil { + return nil, err + } + entry, err := c.get(ctx, namespaceKey(ns)) + if err != nil { + return nil, err + } + if len(entry) == 0 { + return nil, fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace, ns) + } + if attrString(entry[versionCol]) == "" { + return nil, errors.New("DynamoDB namespace version is missing") + } + + return entry, nil +} + +func (c *Catalog) CheckNamespaceExists(ctx context.Context, ns table.Identifier) (bool, error) { + _, err := c.loadNamespace(ctx, ns) + if errors.Is(err, catalog.ErrNoSuchNamespace) { + return false, nil + } + + return err == nil, err +} + +func (c *Catalog) LoadNamespaceProperties(ctx context.Context, ns table.Identifier) (iceberg.Properties, error) { + entry, err := c.loadNamespace(ctx, ns) + if err != nil { + return nil, err + } + + return properties(entry), nil +} + +func (c *Catalog) CreateNamespace(ctx context.Context, ns table.Identifier, props iceberg.Properties) error { + if err := validateNamespace(ns, false); err != nil { + return err + } + entry := namespaceKey(ns) + for key, value := range props { + entry["p."+key] = stringAttr(value) + } + setEntryMetadata(entry, true) + for range 6 { + writes := []types.TransactWriteItem{{Put: &types.Put{ + TableName: aws.String(c.tableName), Item: entry, ConditionExpression: aws.String("attribute_not_exists(#id)"), + ExpressionAttributeNames: map[string]string{"#id": identifierCol}, + }}} + if len(ns) > 1 { + parent := ns[:len(ns)-1] + current, err := c.loadNamespace(ctx, parent) + if err != nil { + return err + } + writes = append(writes, c.touchNamespace(parent, current)) + } + err := c.transaction(ctx, writes) + if err == nil { + return nil + } + checkCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + current, checkErr := c.get(checkCtx, namespaceKey(ns)) + cancel() + if checkErr == nil && attrString(current[versionCol]) == attrString(entry[versionCol]) { + return nil + } + if !conditionFailed(err) && !transactionConflict(err) { + return err + } + if len(current) != 0 { + return fmt.Errorf("%w: %s", catalog.ErrNamespaceAlreadyExists, ns) + } + } + + return fmt.Errorf("%w: namespace parent changed during creation", table.ErrCommitFailed) +} + +func (c *Catalog) ListNamespaces(ctx context.Context, parent table.Identifier) ([]table.Identifier, error) { + if err := validateNamespace(parent, true); err != nil { + return nil, err + } + if len(parent) != 0 { + if _, err := c.loadNamespace(ctx, parent); err != nil { + return nil, err + } + } + prefix := strings.Join(parent, ".") + if prefix != "" { + prefix += "." + } + input := &dynamodb.QueryInput{ + TableName: aws.String(c.tableName), ConsistentRead: aws.Bool(true), + KeyConditionExpression: aws.String("#id = :id"), + ExpressionAttributeNames: map[string]string{"#id": identifierCol}, + ExpressionAttributeValues: item{":id": stringAttr(namespaceIdentifier)}, + } + if prefix != "" { + input.KeyConditionExpression = aws.String("#id = :id AND begins_with(#ns, :prefix)") + input.ExpressionAttributeNames["#ns"] = namespaceCol + input.ExpressionAttributeValues[":prefix"] = stringAttr(prefix) + } + namespaces := []table.Identifier{} + paginator := dynamodb.NewQueryPaginator(c.client, input) + for paginator.HasMorePages() { + page, err := paginator.NextPage(ctx) + if err != nil { + return nil, err + } + for _, entry := range page.Items { + ns := strings.Split(attrString(entry[namespaceCol]), ".") + if len(ns) == len(parent)+1 && slices.Equal(ns[:len(parent)], parent) { + namespaces = append(namespaces, ns) + } + } + } + + return namespaces, nil +} + +func (c *Catalog) DropNamespace(ctx context.Context, ns table.Identifier) error { + current, err := c.loadNamespace(ctx, ns) + if err != nil { + return err + } + // Check every descendant, including namespaces written by Java without an + // intermediate parent entry. + children, err := c.client.Query(ctx, &dynamodb.QueryInput{ + TableName: aws.String(c.tableName), ConsistentRead: aws.Bool(true), Limit: aws.Int32(1), + KeyConditionExpression: aws.String("#id = :id AND begins_with(#ns, :prefix)"), + ExpressionAttributeNames: map[string]string{"#id": identifierCol, "#ns": namespaceCol}, + ExpressionAttributeValues: item{":id": stringAttr(namespaceIdentifier), ":prefix": stringAttr(strings.Join(ns, ".") + ".")}, + }) + if err != nil { + return err + } + if len(children.Items) != 0 { + return catalog.ErrNamespaceNotEmpty + } + + // GSI reads are eventually consistent. Check emptiness through a consistent + // base-table scan, then guard the delete with the namespace version. Go + // create/register/rename operations update that version in their transaction. + scan := dynamodb.NewScanPaginator(c.client, &dynamodb.ScanInput{ + TableName: aws.String(c.tableName), ConsistentRead: aws.Bool(true), + FilterExpression: aws.String("#ns = :ns AND #id <> :marker"), + ExpressionAttributeNames: map[string]string{"#ns": namespaceCol, "#id": identifierCol}, + ExpressionAttributeValues: item{":ns": stringAttr(strings.Join(ns, ".")), ":marker": stringAttr(namespaceIdentifier)}, + }) + for scan.HasMorePages() { + page, err := scan.NextPage(ctx) + if err != nil { + return err + } + if len(page.Items) != 0 { + return catalog.ErrNamespaceNotEmpty + } + } + err = c.transaction(ctx, []types.TransactWriteItem{{Delete: &types.Delete{ + TableName: aws.String(c.tableName), Key: namespaceKey(ns), ConditionExpression: aws.String("#v = :base"), + ExpressionAttributeNames: map[string]string{"#v": versionCol}, ExpressionAttributeValues: item{":base": current[versionCol]}, + }}}) + if conditionFailed(err) || transactionConflict(err) { + return fmt.Errorf("%w: namespace changed during deletion", table.ErrCommitFailed) + } + + return err +} + +func (c *Catalog) UpdateNamespaceProperties(ctx context.Context, ns table.Identifier, removals []string, updates iceberg.Properties) (catalog.PropertiesUpdateSummary, error) { + current, err := c.loadNamespace(ctx, ns) + if err != nil { + return catalog.PropertiesUpdateSummary{}, err + } + updated, summary, err := internal.GetUpdatedPropsAndUpdateSummary(properties(current), removals, updates) + if err != nil { + return catalog.PropertiesUpdateSummary{}, err + } + entry := maps.Clone(current) + for key := range entry { + if strings.HasPrefix(key, "p.") { + delete(entry, key) + } + } + for key, value := range updated { + entry["p."+key] = stringAttr(value) + } + setEntryMetadata(entry, false) + err = c.transaction(ctx, []types.TransactWriteItem{{Put: &types.Put{ + TableName: aws.String(c.tableName), Item: entry, ConditionExpression: aws.String("#v = :base"), + ExpressionAttributeNames: map[string]string{"#v": versionCol}, ExpressionAttributeValues: item{":base": current[versionCol]}, + }}}) + if err != nil { + if conditionFailed(err) || transactionConflict(err) { + err = fmt.Errorf("%w: namespace properties changed", table.ErrCommitFailed) + } + + return catalog.PropertiesUpdateSummary{}, err + } + + return summary, nil +} + +func (c *Catalog) touchNamespace(ns table.Identifier, current item) types.TransactWriteItem { Review Comment: Every create, register and rename now also updates the namespace item's `v` in the same transaction. That makes these operations contend on one item: - Concurrent `CreateTable` calls in the same namespace conflict with each other, retry, and give up with `ErrCommitFailed` after 6 attempts. - A concurrent `UpdateNamespaceProperties` fails. Java's catalog doesn't do this, and Java writers skip it, so it only protects `DropNamespace` against Go writers. Is that worth the contention for workloads that create many tables in one namespace, such as ingestion that creates tables per tenant or per day? If you want to keep it, please describe the tradeoff in `catalog/dynamodb/README.md`. Alternatively, drop the touch and rely on the consistent emptiness check, accepting Java's semantics for a create that races with a drop. -- 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]
