feat: glue table creation with some docs on testing

catalog/catalog.go

@@ -20,10 +20,13 @@ package catalog
 import (
 	"context"
 	"crypto/tls"
+	"encoding/json"
 	"errors"
+	"fmt"
 	"net/url"
 
 	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/io"
 	"github.com/apache/iceberg-go/table"
 	"github.com/aws/aws-sdk-go-v2/aws"
 )
@@ -52,6 +55,14 @@ func WithAwsConfig(cfg aws.Config) Option[GlueCatalog] {
 	}
 }
 
+// WithDefaultLocation sets the default location for the catalog, this is used
+// when a location is not provided in the create table operation.
+func WithDefaultLocation(location string) Option[GlueCatalog] {
+	return func(o *options) {
+		o.defaultLocation = location
+	}
+}
+
 func WithCredential(cred string) Option[RestCatalog] {
 	return func(o *options) {
 		o.credential = cred
@@ -117,7 +128,8 @@ func WithPrefix(prefix string) Option[RestCatalog] {
 type Option[T GlueCatalog | RestCatalog] func(*options)
 
 type options struct {
-	awsConfig aws.Config
+	awsConfig       aws.Config
+	defaultLocation string
 
 	tlsConfig         *tls.Config
 	credential        string
@@ -185,3 +197,17 @@ func TableNameFromIdent(ident table.Identifier) string {
 func NamespaceFromIdent(ident table.Identifier) table.Identifier {
 	return ident[:len(ident)-1]
 }
+
+func writeTableMetaData(iofs io.IO, metadataPath string, metadata table.Metadata) error {
+	data, err := json.Marshal(metadata)
+	if err != nil {
+		return fmt.Errorf("failed to marshal table metadata: %w", err)
+	}
+
+	err = iofs.WriteFile(metadataPath, data, 0644)
+	if err != nil {
+		return fmt.Errorf("failed to write metadata file: %w", err)
+	}
+
+	return nil
+}

catalog/catalog_test.go

@@ -0,0 +1,18 @@
+// 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

catalog/glue.go

@@ -21,6 +21,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/url"
+	"strings"
 
 	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/io"
@@ -39,6 +41,7 @@ var (
 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)
+	CreateTable(ctx context.Context, params *glue.CreateTableInput, optFns ...func(*glue.Options)) (*glue.CreateTableOutput, error)
 }
 
 type GlueCatalog struct {
@@ -121,6 +124,71 @@ func (c *GlueCatalog) LoadTable(ctx context.Context, identifier table.Identifier
 	return icebergTable, nil
 }
 
+// CreateTable creates a new table in the catalog.
+//
+// The identifier should contain the Glue database name, then glue table name.
+// The location should be the S3 prefix for the table, which will have the database name, table name, and metadata file appended to it.
+func (c *GlueCatalog) CreateTable(ctx context.Context, identifier table.Identifier, schema *iceberg.Schema, partitionSpec iceberg.PartitionSpec, sortOrder table.SortOrder, location string, props map[string]string) (*table.Table, error) {
+	database, tableName, err := identifierToGlueTable(identifier)
+	if err != nil {
+		return nil, err
+	}
+
+	// s3://bucket/prefix/database.db/tablename
+	locationURL, err := url.Parse(location)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse location URL %s: %w", location, err)
+	}
+
+	// 00000-UUID.metadata.json
+	newManifest, err := table.GenerateMetadataFileName(0)
+	if err != nil {
+		return nil, fmt.Errorf("failed to generate metadata file name: %w", err)
+	}
+
+	// s3://bucket/prefix/database.db/tablename/manifest/00000-UUID.metadata.json
+	metadataURL := locationURL.JoinPath("metadata", newManifest)
+
+	// prefix/database.db/tablename/manifest/00000-UUID.metadata.json
+	metadataLocation := strings.TrimPrefix(metadataURL.Path, "/")
+
+	tbl, err := table.NewTableBuilder(identifier, schema, location, metadataLocation).
+		WithPartitionSpec(partitionSpec).
+		WithSortOrder(sortOrder).
+		WithProperties(props).
+		Build()
+	if err != nil {
+		return nil, err
+	}
+
+	err = writeTableMetaData(tbl.FS(), tbl.MetadataLocation(), tbl.Metadata())
+	if err != nil {
+		return nil, err
+	}
+
+	// TODO: need to convert the schema to a glue schema and provide that to create table.
+	params := &glue.CreateTableInput{
+		DatabaseName: aws.String(database),
+
+		TableInput: &types.TableInput{
+			Name:      aws.String(tableName),
+			TableType: aws.String("EXTERNAL_TABLE"),
+			Parameters: map[string]string{
+				"table_type":        glueTableTypeIceberg,
+				"metadata_location": metadataURL.String(),
+			},
+			StorageDescriptor: &types.StorageDescriptor{Location: aws.String(locationURL.String())},
+		},
+	}
+
+	_, err = c.glueSvc.CreateTable(ctx, params)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create table %s.%s: %w", database, tableName, err)
+	}
+
+	return tbl, nil
+}
+
 func (c *GlueCatalog) CatalogType() CatalogType {
 	return Glue
 }

catalog/glue_test.go

@@ -22,6 +22,8 @@ import (
 	"os"
 	"testing"
 
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/table"
 	"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"
@@ -44,6 +46,11 @@ func (m *mockGlueClient) GetTables(ctx context.Context, params *glue.GetTablesIn
 	return args.Get(0).(*glue.GetTablesOutput), args.Error(1)
 }
 
+func (m *mockGlueClient) CreateTable(ctx context.Context, params *glue.CreateTableInput, optFns ...func(*glue.Options)) (*glue.CreateTableOutput, error) {
+	args := m.Called(ctx, params, optFns)
+	return args.Get(0).(*glue.CreateTableOutput), args.Error(1)
+}
+
 func TestGlueGetTable(t *testing.T) {
 	assert := require.New(t)
 
@@ -131,9 +138,6 @@ func TestGlueLoadTableIntegration(t *testing.T) {
 	if os.Getenv("TEST_TABLE_NAME") == "" {
 		t.Skip()
 	}
-	if os.Getenv("TEST_TABLE_LOCATION") == "" {
-		t.Skip()
-	}
 
 	assert := require.New(t)
 
@@ -146,3 +150,44 @@ func TestGlueLoadTableIntegration(t *testing.T) {
 	assert.NoError(err)
 	assert.Equal([]string{os.Getenv("TEST_TABLE_NAME")}, table.Identifier())
 }
+
+func TestGlueCreateTableIntegration(t *testing.T) {
+	if os.Getenv("TEST_DATABASE_NAME") == "" {
+		t.Skip()
+	}
+	if os.Getenv("TEST_CREATE_TABLE_NAME") == "" {
+		t.Skip()
+	}
+	if os.Getenv("TEST_CREATE_TABLE_LOCATION") == "" {
+		t.Skip()
+	}
+
+	assert := require.New(t)
+
+	location := os.Getenv("TEST_CREATE_TABLE_LOCATION")
+
+	schema := iceberg.NewSchemaWithIdentifiers(1, []int{},
+		iceberg.NestedField{
+			ID: 1, Name: "vendor_id", Type: iceberg.PrimitiveTypes.String},
+		iceberg.NestedField{
+			ID: 2, Name: "name", Type: iceberg.PrimitiveTypes.String},
+		iceberg.NestedField{
+			ID: 3, Name: "datetime", Type: iceberg.PrimitiveTypes.TimestampTz})
+	partSpec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+		SourceID: 3, FieldID: 1000, Name: "datetime", Transform: iceberg.DayTransform{}})
+
+	props := map[string]string{
+		"write.target-file-size-bytes": "536870912",
+		"write.format.default":         "parquet",
+	}
+
+	awscfg, err := config.LoadDefaultConfig(context.TODO(), config.WithClientLogMode(aws.LogRequest|aws.LogResponse))
+	assert.NoError(err)
+
+	catalog := NewGlueCatalog(WithAwsConfig(awscfg))
+
+	table, err := catalog.CreateTable(context.TODO(),
+		[]string{os.Getenv("TEST_DATABASE_NAME"), os.Getenv("TEST_CREATE_TABLE_NAME")}, schema, partSpec, table.UnsortedSortOrder, location, props)
+	assert.NoError(err)
+	assert.Equal([]string{os.Getenv("TEST_DATABASE_NAME"), os.Getenv("TEST_CREATE_TABLE_NAME")}, table.Identifier())
+}

cfn/AWS_TESTING.md

@@ -0,0 +1,74 @@
+<!--
+ - 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.
+ -->
+
+# AWS integration testing
+
+To validate the glue catalog you will need to create some test resources. 
+
+# Prerequisites
+
+1. An AWS account.
+2. [AWS CLI](https://aws.amazon.com/cli/) is installed.
+2. Exported environment variables for `AWS_DEFAULT_REGION`, `AWS_REGION` and `AWS_PROFILE`, I use [direnv](https://direnv.net/) to maintain these variables in a `.envrc` file. 
+3. Your have logged into an AWS account via the AWS CLI.
+
+The way to deploy this template is using the included cloudformation template is as follows:
+
+```
+aws cloudformation deploy --stack-name test-iceberg-glue-catalog --template-file docs/cfn/glue-catalog.yaml
+```
+
+Once deployed you can retrieve the outputs of the stack.
+
+```
+aws cloudformation describe-stacks --stack-name test-iceberg-glue-catalog --query 'Stacks[0].Outputs'
+```
+
+This should output JSON as follows:
+
+```
+[
+    {
+        "OutputKey": "IcebergBucket",
+        "OutputValue": "test-iceberg-glue-catalog-icebergbucket-abc123abc123"
+    },
+    {
+        "OutputKey": "GlueDatabase",
+        "OutputValue": "iceberg_test"
+    }
+]
+```
+
+Export the required environment variables.
+
+```
+# the glue database from the outputs of the stack
+export TEST_DATABASE_NAME=iceberg_test
+
+# the s3 bucket name from the outputs of the stack
+export TEST_CREATE_TABLE_LOCATION=s3://test-iceberg-glue-catalog-icebergbucket-abc123abc123/testing
+
+# the name of the table you will create in the glue catalog
+export TEST_CREATE_TABLE_NAME=records
+```
+
+Run the creation integration test to validate the catalog creation, and provide a table which can be used to validate other integration tests.
+
+```
+go test -v -run TestGlueCreateTableIntegration ./catalog
+```
+

cfn/glue-catalog.yaml

@@ -0,0 +1,70 @@
+# 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.
+AWSTemplateFormatVersion: "2010-09-09"
+Description: "apache-iceberg: Glue database and S3 bucket for integration testing"
+
+Parameters:
+  Stage:
+    Type: String
+    Description: The stage where the stack is running in, e.g., dev, test, prod.
+    Default: test
+
+Outputs:
+  GlueDatabase:
+    Value: !Ref GlueDatabase
+  IcebergBucket:
+    Value: !Ref IcebergBucket
+
+Resources:
+  GlueDatabase:
+    Type: AWS::Glue::Database
+    Properties:
+      CatalogId: !Ref AWS::AccountId
+      DatabaseInput:
+        Name: !Sub iceberg_${Stage}
+        Description: iceberg database
+
+  IcebergBucket:
+    Type: AWS::S3::Bucket
+    Properties:
+      BucketEncryption:
+        ServerSideEncryptionConfiguration:
+          - ServerSideEncryptionByDefault:
+              SSEAlgorithm: AES256
+      PublicAccessBlockConfiguration:
+        BlockPublicAcls: true
+        BlockPublicPolicy: true
+        IgnorePublicAcls: true
+        RestrictPublicBuckets: true
+
+  IcebergBucketPolicy:
+    Type: AWS::S3::BucketPolicy
+    Properties:
+      Bucket: !Ref IcebergBucket
+      PolicyDocument:
+        Statement:
+          - Sid: AllowSSLRequestsOnly
+            Effect: Deny
+            Principal: "*"
+            Action:
+              - s3:*
+            Resource:
+              - Fn::Sub: arn:aws:s3:::${IcebergBucket}/*
+              - Fn::Sub: arn:aws:s3:::${IcebergBucket}
+            Condition:
+              Bool:
+                aws:SecureTransport: false

internal/mock_fs.go

@@ -39,6 +39,10 @@ func (m *MockFS) Remove(name string) error {
 	return m.Called(name).Error(0)
 }
 
+func (m *MockFS) WriteFile(name string, data []byte, perm fs.FileMode) error {
+	return m.Called(name, data, perm).Error(0)
+}
+
 type MockFSReadFile struct {
 	MockFS
 }