upstream-seaweedfs/test/s3tables/unity_catalog/unity_catalog_test.go

// Package unity_catalog provides integration tests that run Unity Catalog
// OSS in Docker against a SeaweedFS S3 backend. The base test mirrors the
// configuration used by the upstream UC playground at
// https://github.com/data-engineering-helpers/mds-in-a-box/tree/main/unitycatalog-playground:
// static aws.accessKey/aws.secretKey, no master role, EXTERNAL Delta tables.
package unity_catalog

import (
	"context"
	"fmt"
	"testing"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/s3"

	"github.com/seaweedfs/seaweedfs/test/testutil"
)

// TestUnityCatalogDeltaIntegration brings up SeaweedFS, runs Unity Catalog OSS
// in Docker against it, and exercises catalog/schema/table CRUD plus
// temporary-table-credentials and storage I/O via vended credentials. The UC
// server is configured with static keys (aws.masterRoleArn empty), matching
// the playground's working configuration when targeting SeaweedFS.
func TestUnityCatalogDeltaIntegration(t *testing.T) {
	if testing.Short() {
		t.Skip("skipping integration test in -short mode")
	}
	if !testutil.HasDocker() {
		t.Skip("docker not available")
	}

	env := newTestEnv(t)
	defer env.cleanup(t)

	t.Log(">>> starting SeaweedFS for Unity Catalog test...")
	env.startSeaweedFS(t, "")
	t.Log(">>> SeaweedFS ready")

	ctx, cancel := context.WithTimeout(context.Background(), 8*time.Minute)
	defer cancel()

	s3c := env.newHostS3Client(t, ctx)
	if _, err := s3c.CreateBucket(ctx, &s3.CreateBucketInput{Bucket: aws.String(ucWarehouse)}); err != nil {
		t.Fatalf("create warehouse bucket: %v", err)
	}

	t.Log(">>> starting Unity Catalog server (static keys)...")
	env.startUnityCatalog(t, ctx, ucServerOpts{})
	t.Log(">>> Unity Catalog ready")

	uc := newUCClient(fmt.Sprintf("http://127.0.0.1:%d", env.ucHostPort))

	suffix := time.Now().UnixNano()
	catalogName := fmt.Sprintf("seaweed_uc_%d", suffix)
	schemaName := fmt.Sprintf("ns_%d", suffix)
	tableName := fmt.Sprintf("events_%d", suffix)
	tableLocation := fmt.Sprintf("s3://%s/%s/%s/%s", ucWarehouse, ucWarehouseKey, schemaName, tableName)

	defer func() {
		_ = uc.deleteTable(context.Background(), catalogName+"."+schemaName+"."+tableName)
		_ = uc.deleteSchema(context.Background(), catalogName+"."+schemaName)
		_ = uc.deleteCatalog(context.Background(), catalogName)
	}()

	t.Run("CreateCatalog", func(t *testing.T) {
		got, err := uc.createCatalog(ctx, ucCreateCatalog{
			Name:        catalogName,
			Comment:     "seaweedfs integration test",
			StorageRoot: fmt.Sprintf("s3://%s/%s", ucWarehouse, ucWarehouseKey),
		})
		if err != nil {
			t.Fatalf("create catalog: %v", err)
		}
		if got.Name != catalogName {
			t.Fatalf("catalog name = %q, want %q", got.Name, catalogName)
		}
	})

	t.Run("CreateSchema", func(t *testing.T) {
		got, err := uc.createSchema(ctx, ucCreateSchema{
			Name:        schemaName,
			CatalogName: catalogName,
		})
		if err != nil {
			t.Fatalf("create schema: %v", err)
		}
		if got.Name != schemaName || got.CatalogName != catalogName {
			t.Fatalf("schema = %+v, want name=%q catalog=%q", got, schemaName, catalogName)
		}
	})

	var createdTable ucTableInfo
	t.Run("CreateExternalDeltaTable", func(t *testing.T) {
		columns := []ucColumn{
			{Name: "id", TypeText: "long", TypeName: "LONG", TypeJSON: `{"name":"id","type":"long","nullable":true,"metadata":{}}`, Position: 0, Nullable: true},
			{Name: "value", TypeText: "string", TypeName: "STRING", TypeJSON: `{"name":"value","type":"string","nullable":true,"metadata":{}}`, Position: 1, Nullable: true},
		}
		got, err := uc.createTable(ctx, ucCreateTable{
			Name:             tableName,
			CatalogName:      catalogName,
			SchemaName:       schemaName,
			TableType:        "EXTERNAL",
			DataSourceFormat: "DELTA",
			Columns:          columns,
			StorageLocation:  tableLocation,
		})
		if err != nil {
			t.Fatalf("create table: %v", err)
		}
		if got.TableType != "EXTERNAL" || got.DataSourceFormat != "DELTA" {
			t.Fatalf("table types = %s/%s, want EXTERNAL/DELTA", got.TableType, got.DataSourceFormat)
		}
		if got.StorageLocation == "" {
			t.Fatalf("table storage_location is empty in response: %+v", got)
		}
		createdTable = *got
	})

	t.Run("GetTable", func(t *testing.T) {
		full := catalogName + "." + schemaName + "." + tableName
		got, err := uc.getTable(ctx, full)
		if err != nil {
			t.Fatalf("get table: %v", err)
		}
		if got.Name != tableName {
			t.Fatalf("get table name = %q, want %q", got.Name, tableName)
		}
	})

	t.Run("ListTables", func(t *testing.T) {
		tables, err := uc.listTables(ctx, catalogName, schemaName)
		if err != nil {
			t.Fatalf("list tables: %v", err)
		}
		found := false
		for _, tbl := range tables {
			if tbl.Name == tableName {
				found = true
				break
			}
		}
		if !found {
			t.Fatalf("created table %q not found in list", tableName)
		}
	})

	t.Run("TemporaryTableCredentialsRejected", func(t *testing.T) {
		// With aws.masterRoleArn empty AND no s3.sessionToken.0 set, UC OSS
		// always tries to AssumeRole via its internal StsClient (see
		// AwsCredentialVendor.createPerBucketCredentialGenerator). Against a
		// non-AWS endpoint, that call doesn't reach a real STS, so UC returns
		// "S3 bucket configuration not found." or an STS-side error. This is
		// the gap users hit at <https://github.com/data-engineering-helpers/mds-in-a-box/blob/main/unitycatalog-playground/etc/conf/server.properties#L45>:
		// "with simple S3 access and secret keys, Unity Catalog does not seem
		// to work."
		//
		// The assertion is therefore inverted: we expect a non-nil error from
		// /temporary-table-credentials with this configuration. A future
		// variant can pin s3.sessionToken.0 (UC's StaticAwsCredentialGenerator
		// path) once SeaweedFS' SigV4 path tolerates the vended session token.
		if createdTable.TableID == "" {
			t.Fatalf("created table has empty table_id; cannot request temporary credentials")
		}
		_, err := uc.generateTemporaryTableCredentials(ctx, createdTable.TableID, "READ_WRITE")
		if err == nil {
			t.Fatalf("expected /temporary-table-credentials to fail with the static-key playground configuration; it succeeded unexpectedly")
		}
		t.Logf("expected failure (UC static-key path requires AWS STS): %v", err)
	})

	t.Run("DeleteTableSchemaCatalog", func(t *testing.T) {
		if err := uc.deleteTable(ctx, catalogName+"."+schemaName+"."+tableName); err != nil {
			t.Fatalf("delete table: %v", err)
		}
		if err := uc.deleteSchema(ctx, catalogName+"."+schemaName); err != nil {
			t.Fatalf("delete schema: %v", err)
		}
		if err := uc.deleteCatalog(ctx, catalogName); err != nil {
			t.Fatalf("delete catalog: %v", err)
		}
	})
}