feat: add support for catalogs with glue implementation to start

.gitignore

@@ -44,3 +44,5 @@ lib/
 *.ipr
 *.iws
 *.iml
+
+.envrc*

catalog/catalog.go

@@ -0,0 +1,51 @@
+// 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 catalog
+
+import (
+	"context"
+	"errors"
+
+	"github.com/apache/iceberg-go/table"
+)
+
+type CatalogType string
+
+const (
+	REST     CatalogType = "rest"
+	Hive     CatalogType = "hive"
+	Glue     CatalogType = "glue"
+	DynamoDB CatalogType = "dynamodb"
+	SQL      CatalogType = "sql"
+)
+
+var (
+	// ErrNoSuchTable is returned when a table does not exist in the catalog.
+	ErrNoSuchTable = errors.New("table does not exist")
+)
+
+// Catalog for iceberg table operations like create, drop, load, list and others.
+type Catalog interface {
+	// ListTables returns a list of table identifiers in the catalog, with the returned
+	// identifiers containing the information required to load the table via that catalog.
+	ListTables(ctx context.Context, namespace table.Identifier) ([]table.Identifier, error)
+	// LoadTable loads a table from the catalog and returns a Table with the metadata.
+	LoadTable(ctx context.Context, identifier table.Identifier) (*table.Table, error)
+	// CatalogType returns the type of the catalog.
+	CatalogType() CatalogType
+}

catalog/glue.go

@@ -0,0 +1,162 @@
+// 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 catalog
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/apache/iceberg-go/io"
+	"github.com/apache/iceberg-go/table"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/glue"
+	"github.com/aws/aws-sdk-go-v2/service/glue/types"
+)
+
+var (
+	_ Catalog = (*GlueCatalog)(nil)
+)
+
+type GlueAPI interface {
+	GetTable(ctx context.Context, params *glue.GetTableInput, optFns ...func(*glue.Options)) (*glue.GetTableOutput, error)
+	GetTables(ctx context.Context, params *glue.GetTablesInput, optFns ...func(*glue.Options)) (*glue.GetTablesOutput, error)
+}
+
+type GlueCatalog struct {
+	glueSvc GlueAPI
+}
+
+func NewGlueCatalog(awscfg aws.Config) *GlueCatalog {
+	return &GlueCatalog{
+		glueSvc: glue.NewFromConfig(awscfg),
+	}
+}
+
+// ListTables returns a list of iceberg tables in the given Glue database.
+//
+// The namespace should just contain the Glue database name.
+func (c *GlueCatalog) ListTables(ctx context.Context, namespace table.Identifier) ([]table.Identifier, error) {
+	database, err := identifierToGlueDatabase(namespace)
+	if err != nil {
+		return nil, err
+	}
+
+	params := &glue.GetTablesInput{DatabaseName: aws.String(database)}
+
+	tblsRes, err := c.glueSvc.GetTables(ctx, params)
+	if err != nil {
+		return nil, fmt.Errorf("failed to list tables in namespace %s: %w", database, err)
+	}
+
+	var icebergTables []table.Identifier
+
+	for _, tbl := range tblsRes.TableList {
+		// skip non iceberg tables
+		// TODO: consider what this would look like for non ICEBERG tables as you can convert them to ICEBERG tables via the Glue catalog API.
+		if tbl.Parameters["table_type"] != "ICEBERG" {
+			continue
+		}
+
+		icebergTables = append(icebergTables,
+			GlueTableIdentifier(database, aws.ToString(tbl.Name)),
+		)
+	}
+
+	return icebergTables, nil
+}
+
+// LoadTable loads a table from the catalog table details.
+//
+// The identifier should contain the Glue database name, then glue table name.
+func (c *GlueCatalog) LoadTable(ctx context.Context, identifier table.Identifier) (*table.Table, error) {
+	database, tableName, err := identifierToGlueTable(identifier)
+	if err != nil {
+		return nil, err
+	}
+
+	location, err := c.getTable(ctx, database, tableName)
+	if err != nil {
+		return nil, err
+	}
+
+	// TODO: consider providing a way to directly access the S3 iofs to enable testing of the catalog.
+	iofs, err := io.LoadFS(map[string]string{}, location)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load table %s.%s: %w", database, tableName, err)
+	}
+
+	icebergTable, err := table.NewFromLocation([]string{tableName}, location, iofs)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create table from location %s.%s: %w", database, tableName, err)
+	}
+
+	return icebergTable, nil
+}
+
+func (c *GlueCatalog) CatalogType() CatalogType {
+	return Glue
+}
+
+// GetTable loads a table from the Glue Catalog using the given database and table name.
+func (c *GlueCatalog) getTable(ctx context.Context, database, tableName string) (string, error) {
+	tblRes, err := c.glueSvc.GetTable(ctx,
+		&glue.GetTableInput{
+			DatabaseName: aws.String(database),
+			Name:         aws.String(tableName),
+		},
+	)
+	if err != nil {
+		if errors.Is(err, &types.EntityNotFoundException{}) {
+			return "", ErrNoSuchTable
+		}
+		return "", fmt.Errorf("failed to get table %s.%s: %w", database, tableName, err)
+	}
+
+	if tblRes.Table.Parameters["table_type"] != "ICEBERG" {
+		return "", errors.New("table is not an iceberg table")
+	}
+
+	return tblRes.Table.Parameters["metadata_location"], nil
+}
+
+func identifierToGlueTable(identifier table.Identifier) (string, string, error) {
+	if len(identifier) != 2 {
+		return "", "", fmt.Errorf("invalid identifier, missing database name: %v", identifier)
+	}
+
+	return identifier[0], identifier[1], nil
+}
+
+func identifierToGlueDatabase(identifier table.Identifier) (string, error) {
+	if len(identifier) != 1 {
+		return "", fmt.Errorf("invalid identifier, missing database name: %v", identifier)
+	}
+
+	return identifier[0], nil
+}
+
+// GlueTableIdentifier returns a glue table identifier for an iceberg table in the format [database, table].
+func GlueTableIdentifier(database string, table string) table.Identifier {
+	return []string{database, table}
+}
+
+// GlueDatabaseIdentifier returns a database identifier for a Glue database in the format [database].
+func GlueDatabaseIdentifier(database string) table.Identifier {
+	return []string{database}
+}

catalog/glue_test.go

@@ -0,0 +1,141 @@
+// 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 catalog
+
+import (
+	"context"
+	"os"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/service/glue"
+	"github.com/aws/aws-sdk-go-v2/service/glue/types"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+type mockGlueClient struct {
+	mock.Mock
+}
+
+func (m *mockGlueClient) GetTable(ctx context.Context, params *glue.GetTableInput, optFns ...func(*glue.Options)) (*glue.GetTableOutput, error) {
+	args := m.Called(ctx, params, optFns)
+	return args.Get(0).(*glue.GetTableOutput), args.Error(1)
+}
+
+func (m *mockGlueClient) GetTables(ctx context.Context, params *glue.GetTablesInput, optFns ...func(*glue.Options)) (*glue.GetTablesOutput, error) {
+	args := m.Called(ctx, params, optFns)
+	return args.Get(0).(*glue.GetTablesOutput), args.Error(1)
+}
+
+func TestGlueGetTable(t *testing.T) {
+	assert := require.New(t)
+
+	mockGlueSvc := &mockGlueClient{}
+
+	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{
+			Parameters: map[string]string{
+				"table_type":        "ICEBERG",
+				"metadata_location": "s3://test-bucket/test_table/metadata/abc123-123.metadata.json",
+			},
+		},
+	}, nil)
+
+	glueCatalog := &GlueCatalog{
+		glueSvc: mockGlueSvc,
+	}
+
+	location, err := glueCatalog.getTable(context.TODO(), "test_database", "test_table")
+	assert.NoError(err)
+	assert.Equal("s3://test-bucket/test_table/metadata/abc123-123.metadata.json", location)
+}
+
+func TestGlueListTables(t *testing.T) {
+	assert := require.New(t)
+
+	mockGlueSvc := &mockGlueClient{}
+
+	mockGlueSvc.On("GetTables", mock.Anything, &glue.GetTablesInput{
+		DatabaseName: aws.String("test_database"),
+	}, mock.Anything).Return(&glue.GetTablesOutput{
+		TableList: []types.Table{
+			{
+				Name: aws.String("test_table"),
+				Parameters: map[string]string{
+					"table_type":        "ICEBERG",
+					"metadata_location": "s3://test-bucket/test_table/metadata/abc123-123.metadata.json",
+				},
+			},
+		},
+	}, nil)
+
+	glueCatalog := &GlueCatalog{
+		glueSvc: mockGlueSvc,
+	}
+
+	tables, err := glueCatalog.ListTables(context.TODO(), GlueDatabaseIdentifier("test_database"))
+	assert.NoError(err)
+	assert.Equal([]string{"test_database", "test_table"}, tables[0])
+}
+
+func TestGlueListTableIntegration(t *testing.T) {
+	if os.Getenv("TEST_DATABASE_NAME") == "" {
+		t.Skip()
+	}
+	if os.Getenv("TEST_TABLE_NAME") == "" {
+		t.Skip()
+	}
+	assert := require.New(t)
+
+	awscfg, err := config.LoadDefaultConfig(context.TODO(), config.WithClientLogMode(aws.LogRequest|aws.LogResponse))
+	assert.NoError(err)
+
+	catalog := NewGlueCatalog(awscfg)
+
+	tables, err := catalog.ListTables(context.TODO(), GlueDatabaseIdentifier(os.Getenv("TEST_DATABASE_NAME")))
+	assert.NoError(err)
+	assert.Equal([]string{os.Getenv("TEST_DATABASE_NAME"), os.Getenv("TEST_TABLE_NAME")}, tables[1])
+}
+
+func TestGlueLoadTableIntegration(t *testing.T) {
+	if os.Getenv("TEST_DATABASE_NAME") == "" {
+		t.Skip()
+	}
+	if os.Getenv("TEST_TABLE_NAME") == "" {
+		t.Skip()
+	}
+	if os.Getenv("TEST_TABLE_LOCATION") == "" {
+		t.Skip()
+	}
+
+	assert := require.New(t)
+
+	awscfg, err := config.LoadDefaultConfig(context.TODO(), config.WithClientLogMode(aws.LogRequest|aws.LogResponse))
+	assert.NoError(err)
+
+	catalog := NewGlueCatalog(awscfg)
+
+	table, err := catalog.LoadTable(context.TODO(), []string{os.Getenv("TEST_DATABASE_NAME"), os.Getenv("TEST_TABLE_NAME")})
+	assert.NoError(err)
+	assert.Equal([]string{os.Getenv("TEST_TABLE_NAME")}, table.Identifier())
+}

go.mod

@@ -23,6 +23,7 @@ require (
 	github.com/aws/aws-sdk-go-v2 v1.24.1
 	github.com/aws/aws-sdk-go-v2/config v1.26.3
 	github.com/aws/aws-sdk-go-v2/credentials v1.16.14
+	github.com/aws/aws-sdk-go-v2/service/glue v1.73.1
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.48.0
 	github.com/google/uuid v1.3.1
 	github.com/hamba/avro/v2 v2.16.0

go.sum

@@ -16,6 +16,8 @@ github.com/aws/aws-sdk-go-v2/internal/ini v1.7.2 h1:GrSw8s0Gs/5zZ0SX+gX4zQjRnRsM
 github.com/aws/aws-sdk-go-v2/internal/ini v1.7.2/go.mod h1:6fQQgfuGmw8Al/3M2IgIllycxV7ZW7WCdVSqfBeUiCY=
 github.com/aws/aws-sdk-go-v2/internal/v4a v1.2.10 h1:5oE2WzJE56/mVveuDZPJESKlg/00AaS2pY2QZcnxg4M=
 github.com/aws/aws-sdk-go-v2/internal/v4a v1.2.10/go.mod h1:FHbKWQtRBYUz4vO5WBWjzMD2by126ny5y/1EoaWoLfI=
+github.com/aws/aws-sdk-go-v2/service/glue v1.73.1 h1:z/NBYW8RygzWrDgNWib10fuLUBl0SLj0KruGoEHxnKQ=
+github.com/aws/aws-sdk-go-v2/service/glue v1.73.1/go.mod h1:F3B9DC5FsIHAxUtHZdY5KUeqN+tHoGlRPzSSYdXjC38=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.10.4 h1:/b31bi3YVNlkzkBrm9LfpaKoaYZUxIAj4sHfOTmLfqw=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.10.4/go.mod h1:2aGXHFmbInwgP9ZfpmdIfOELL79zhdNYNmReK8qDfdQ=
 github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.2.10 h1:L0ai8WICYHozIKK+OtPzVJBugL7culcuM4E4JOpIEm8=