func()

in catalog/glue/glue.go [418:482]
52 lines of code
9 McCabe index (conditional complexity)

func (c *Catalog) RenameTable(ctx context.Context, from, to table.Identifier) (*table.Table, error) {
	fromDatabase, fromTable, err := identifierToGlueTable(from)
	if err != nil {
		return nil, err
	}

	toDatabase, toTable, err := identifierToGlueTable(to)
	if err != nil {
		return nil, err
	}

	if fromDatabase != toDatabase {
		return nil, fmt.Errorf("cannot rename table across namespaces: %s -> %s", fromDatabase, toDatabase)
	}

	// Fetch the existing Glue table to copy the metadata into the new table.
	fromGlueTable, err := c.getTable(ctx, fromDatabase, fromTable)
	if err != nil {
		return nil, fmt.Errorf("failed to fetch the table %s.%s: %w", fromDatabase, fromTable, err)
	}

	// Create the new table.
	_, err = c.glueSvc.CreateTable(ctx, &glue.CreateTableInput{
		CatalogId:    c.catalogId,
		DatabaseName: aws.String(toDatabase),
		TableInput: &types.TableInput{
			Name:              aws.String(toTable),
			Owner:             fromGlueTable.Owner,
			Description:       fromGlueTable.Description,
			Parameters:        fromGlueTable.Parameters,
			StorageDescriptor: fromGlueTable.StorageDescriptor,
		},
	})
	if err != nil {
		return nil, fmt.Errorf("failed to create the table %s.%s: %w", fromDatabase, fromTable, err)
	}

	// Drop the old table.
	_, err = c.glueSvc.DeleteTable(ctx, &glue.DeleteTableInput{
		CatalogId:    c.catalogId,
		DatabaseName: aws.String(fromDatabase),
		Name:         aws.String(fromTable),
	})
	if err != nil {
		// Best-effort rollback the table creation.
		_, rollbackErr := c.glueSvc.DeleteTable(ctx, &glue.DeleteTableInput{
			CatalogId:    c.catalogId,
			DatabaseName: aws.String(toDatabase),
			Name:         aws.String(toTable),
		})
		if rollbackErr != nil {
			fmt.Printf("failed to rollback the new table %s.%s: %v", toDatabase, toTable, rollbackErr)
		}

		return nil, fmt.Errorf("failed to rename the table %s.%s: %w", fromDatabase, fromTable, err)
	}

	// Load the new table to return.
	renamedTable, err := c.LoadTable(ctx, TableIdentifier(toDatabase, toTable), nil)
	if err != nil {
		return nil, fmt.Errorf("failed to load renamed table %s.%s: %w", toDatabase, toTable, err)
	}

	return renamedTable, nil
}