Repository: apache/iceberg-go
Pull request #352 (2025-03-20 16:57:41)

fix(catalog): add rest integration write test
Includes a few fixes that were found when writing the test


Patch:
diff --git a/catalog/rest/options.go b/catalog/rest/options.go
index 5aae122..ae1782c 100644
--- a/catalog/rest/options.go
+++ b/catalog/rest/options.go
@@ -21,6 +21,7 @@ import (
 	"crypto/tls"
 	"net/url"
 
+	"github.com/apache/iceberg-go"
 	"github.com/aws/aws-sdk-go-v2/aws"
 )
 
@@ -100,6 +101,12 @@ func WithScope(scope string) Option {
 	}
 }
 
+func WithAdditionalProps(props iceberg.Properties) Option {
+	return func(o *options) {
+		o.additionalProps = props
+	}
+}
+
 type options struct {
 	awsConfig         aws.Config
 	tlsConfig         *tls.Config
@@ -113,4 +120,6 @@ type options struct {
 	prefix            string
 	authUri           *url.URL
 	scope             string
+
+	additionalProps iceberg.Properties
 }
diff --git a/catalog/rest/rest.go b/catalog/rest/rest.go
index cbeed0f..5eb870f 100644
--- a/catalog/rest/rest.go
+++ b/catalog/rest/rest.go
@@ -377,7 +377,9 @@ func handleNon200(rsp *http.Response, override map[int]error) error {
 }
 
 func fromProps(props iceberg.Properties) *options {
-	o := &options{}
+	o := &options{
+		additionalProps: iceberg.Properties{},
+	}
 	for k, v := range props {
 		switch k {
 		case keyOauthToken:
@@ -411,6 +413,11 @@ func fromProps(props iceberg.Properties) *options {
 			} else {
 				o.tlsConfig.InsecureSkipVerify = verify
 			}
+		case "uri", "type":
+		default:
+			if v != "" {
+				o.additionalProps[k] = v
+			}
 		}
 	}
 
@@ -418,7 +425,12 @@ func fromProps(props iceberg.Properties) *options {
 }
 
 func toProps(o *options) iceberg.Properties {
-	props := iceberg.Properties{}
+	var props iceberg.Properties
+	if o.additionalProps != nil {
+		props = o.additionalProps
+	} else {
+		props = iceberg.Properties{}
+	}
 
 	setIf := func(key, v string) {
 		if v != "" {
diff --git a/catalog/sql/sql.go b/catalog/sql/sql.go
index b939e9f..34593c0 100644
--- a/catalog/sql/sql.go
+++ b/catalog/sql/sql.go
@@ -24,7 +24,7 @@ import (
 	"fmt"
 	"iter"
 	"maps"
-	"path"
+	"net/url"
 	"slices"
 	"strings"
 	"sync"
@@ -265,11 +265,11 @@ func (c *Catalog) getDefaultWarehouseLocation(ctx context.Context, nsname, table
 	}
 
 	if dblocation := dbprops.Get("location", ""); dblocation != "" {
-		return path.Join(dblocation, tableName), nil
+		return url.JoinPath(dblocation, tableName)
 	}
 
 	if warehousepath := c.props.Get("warehouse", ""); warehousepath != "" {
-		return warehousepath + "/" + path.Join(nsname+".db", tableName), nil
+		return url.JoinPath(warehousepath, nsname+".db", tableName)
 	}
 
 	return "", errors.New("no default path set, please specify a location when creating a table")
diff --git a/table/metadata.go b/table/metadata.go
index 08d5512..44986d5 100644
--- a/table/metadata.go
+++ b/table/metadata.go
@@ -828,14 +828,14 @@ type commonMetadata struct {
 	Specs              []iceberg.PartitionSpec `json:"partition-specs"`
 	DefaultSpecID      int                     `json:"default-spec-id"`
 	LastPartitionID    *int                    `json:"last-partition-id,omitempty"`
-	Props              iceberg.Properties      `json:"properties"`
+	Props              iceberg.Properties      `json:"properties,omitempty"`
 	SnapshotList       []Snapshot              `json:"snapshots,omitempty"`
 	CurrentSnapshotID  *int64                  `json:"current-snapshot-id,omitempty"`
-	SnapshotLog        []SnapshotLogEntry      `json:"snapshot-log"`
-	MetadataLog        []MetadataLogEntry      `json:"metadata-log"`
+	SnapshotLog        []SnapshotLogEntry      `json:"snapshot-log,omitempty"`
+	MetadataLog        []MetadataLogEntry      `json:"metadata-log,omitempty"`
 	SortOrderList      []SortOrder             `json:"sort-orders"`
 	DefaultSortOrderID int                     `json:"default-sort-order-id"`
-	SnapshotRefs       map[string]SnapshotRef  `json:"refs"`
+	SnapshotRefs       map[string]SnapshotRef  `json:"refs,omitempty"`
 }
 
 func (c *commonMetadata) Ref() SnapshotRef                     { return c.SnapshotRefs[MainBranch] }
@@ -1094,8 +1094,8 @@ func (c *commonMetadata) NameMapping() iceberg.NameMapping {
 func (c *commonMetadata) Version() int { return c.FormatVersion }
 
 type metadataV1 struct {
-	Schema    *iceberg.Schema          `json:"schema"`
-	Partition []iceberg.PartitionField `json:"partition-spec"`
+	Schema    *iceberg.Schema          `json:"schema,omitempty"`
+	Partition []iceberg.PartitionField `json:"partition-spec,omitempty"`
 
 	commonMetadata
 }
diff --git a/table/table.go b/table/table.go
index 6578ead..07b8a38 100644
--- a/table/table.go
+++ b/table/table.go
@@ -308,11 +308,11 @@ func (t Table) Scan(opts ...ScanOption) *Scan {
 	return s
 }
 
-func New(ident Identifier, meta Metadata, location string, fs io.IO, cat CatalogIO) *Table {
+func New(ident Identifier, meta Metadata, metadataLocation string, fs io.IO, cat CatalogIO) *Table {
 	return &Table{
 		identifier:       ident,
 		metadata:         meta,
-		metadataLocation: location,
+		metadataLocation: metadataLocation,
 		fs:               fs,
 		cat:              cat,
 	}


Test patch:
diff --git a/catalog/rest/rest_integration_test.go b/catalog/rest/rest_integration_test.go
index b17725a..b68a32f 100644
--- a/catalog/rest/rest_integration_test.go
+++ b/catalog/rest/rest_integration_test.go
@@ -21,12 +21,17 @@ package rest_test
 
 import (
 	"context"
+	"net/url"
 	"testing"
 
+	"github.com/apache/arrow-go/v18/arrow/array"
+	"github.com/apache/arrow-go/v18/arrow/memory"
+	"github.com/apache/arrow-go/v18/parquet/pqarrow"
 	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/catalog"
 	"github.com/apache/iceberg-go/catalog/rest"
 	"github.com/apache/iceberg-go/io"
+	"github.com/apache/iceberg-go/table"
 	"github.com/stretchr/testify/suite"
 )
 
@@ -43,6 +48,7 @@ func (s *RestIntegrationSuite) loadCatalog(ctx context.Context) *rest.Catalog {
 	cat, err := catalog.Load(ctx, "local", iceberg.Properties{
 		"type":               "rest",
 		"uri":                "http://localhost:8181",
+		io.S3Region:          "us-east-1",
 		io.S3AccessKeyID:     "admin",
 		io.S3SecretAccessKey: "password",
 	})
@@ -60,7 +66,7 @@ var (
 	tableSchemaNested = iceberg.NewSchemaWithIdentifiers(1,
 		[]int{1},
 		iceberg.NestedField{
-			ID: 1, Name: "foo", Type: iceberg.PrimitiveTypes.String, Required: false},
+			ID: 1, Name: "foo", Type: iceberg.PrimitiveTypes.String, Required: true},
 		iceberg.NestedField{
 			ID: 2, Name: "bar", Type: iceberg.PrimitiveTypes.Int32, Required: true},
 		iceberg.NestedField{
@@ -175,6 +181,72 @@ func (s *RestIntegrationSuite) TestCreateTable() {
 	s.Require().NoError(s.cat.DropTable(s.ctx, catalog.ToIdentifier(TestNamespaceIdent, "test-table")))
 }
 
+func (s *RestIntegrationSuite) TestWriteCommitTable() {
+	s.ensureNamespace()
+
+	const location = "s3://warehouse/iceberg"
+
+	tbl, err := s.cat.CreateTable(s.ctx,
+		catalog.ToIdentifier(TestNamespaceIdent, "test-table-2"),
+		tableSchemaNested, catalog.WithLocation(location))
+	s.Require().NoError(err)
+	s.Require().NotNil(tbl)
+
+	defer func() {
+		s.Require().NoError(s.cat.DropTable(s.ctx, catalog.ToIdentifier(TestNamespaceIdent, "test-table-2")))
+	}()
+
+	s.Equal(location, tbl.Location())
+
+	arrSchema, err := table.SchemaToArrowSchema(tableSchemaNested, nil, false, false)
+	s.Require().NoError(err)
+
+	table, err := array.TableFromJSON(memory.DefaultAllocator, arrSchema,
+		[]string{`[
+		{
+			"foo": "foo_string",
+			"bar": 123,
+			"baz": true,
+			"qux": ["a", "b", "c"],
+			"quux": [{"key": "gopher", "value": [
+				{"key": "golang", "value": "1337"}]}],
+			"location": [{"latitude": 37.7749, "longitude": -122.4194}],
+			"person": {"name": "gopher", "age": 10}
+		}
+	]`})
+	s.Require().NoError(err)
+	defer table.Release()
+
+	pqfile, err := url.JoinPath(location, "data", "test_commit_table_data", "test.parquet")
+	s.Require().NoError(err)
+
+	fw, err := tbl.FS().(io.WriteFileIO).Create(pqfile)
+	s.Require().NoError(err)
+	s.Require().NoError(pqarrow.WriteTable(table, fw, table.NumRows(),
+		nil, pqarrow.DefaultWriterProps()))
+	defer tbl.FS().Remove(pqfile)
+
+	txn := tbl.NewTransaction()
+	s.Require().NoError(txn.AddFiles([]string{pqfile}, nil, false))
+	updated, err := txn.Commit(s.ctx)
+	s.Require().NoError(err)
+
+	mf := []iceberg.ManifestFile{}
+	for m, err := range updated.AllManifests() {
+		s.Require().NoError(err)
+		s.Require().NotNil(m)
+		mf = append(mf, m)
+	}
+
+	s.Len(mf, 1)
+	s.EqualValues(1, mf[0].AddedDataFiles())
+	entries, err := mf[0].FetchEntries(updated.FS(), false)
+	s.Require().NoError(err)
+
+	s.Len(entries, 1)
+	s.Equal(pqfile, entries[0].DataFile().FilePath())
+}
+
 func TestRestIntegration(t *testing.T) {
 	suite.Run(t, new(RestIntegrationSuite))
 }
diff --git a/io/s3_integration_test.go b/io/s3_integration_test.go
index 95434a4..12e9a0d 100644
--- a/io/s3_integration_test.go
+++ b/io/s3_integration_test.go
@@ -59,3 +59,32 @@ func TestMinioWarehouse(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, tbl)
 }
+
+func TestMinioWarehouseNoLocation(t *testing.T) {
+	cat, err := catalog.Load(context.Background(), "default", iceberg.Properties{
+		"uri":                ":memory:",
+		sqlcat.DriverKey:     sqliteshim.ShimName,
+		sqlcat.DialectKey:    string(sqlcat.SQLite),
+		"type":               "sql",
+		"warehouse":          "s3a://warehouse/iceberg/",
+		io.S3Region:          "local",
+		io.S3AccessKeyID:     "admin",
+		io.S3SecretAccessKey: "password",
+		// endpoint is passed via AWS_S3_ENDPOINT env var
+	})
+	require.NoError(t, err)
+
+	require.NotNil(t, cat)
+
+	c := cat.(*sqlcat.Catalog)
+	ctx := context.Background()
+	require.NoError(t, c.CreateNamespace(ctx, catalog.ToIdentifier("iceberg-test-2"), nil))
+
+	tbl, err := c.CreateTable(ctx,
+		catalog.ToIdentifier("iceberg-test-2", "test-table-2"),
+		iceberg.NewSchema(0, iceberg.NestedField{
+			Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true, ID: 1,
+		}))
+	require.NoError(t, err)
+	require.NotNil(t, tbl)
+}
diff --git a/table/metadata_internal_test.go b/table/metadata_internal_test.go
index e28dfd4..9e33678 100644
--- a/table/metadata_internal_test.go
+++ b/table/metadata_internal_test.go
@@ -206,8 +206,54 @@ func TestSerializeMetadataV1(t *testing.T) {
 	data, err := json.Marshal(&meta)
 	require.NoError(t, err)
 
-	assert.JSONEq(t, `{"location": "s3://bucket/test/location", "table-uuid": "d20125c8-7284-442c-9aea-15fee620737c", "last-updated-ms": 1602638573874, "last-column-id": 3, "schemas": [{"type": "struct", "fields": [{"id": 1, "name": "x", "type": "long", "required": true}, {"id": 2, "name": "y", "type": "long", "required": true, "doc": "comment"}, {"id": 3, "name": "z", "type": "long", "required": true}], "schema-id": 0, "identifier-field-ids": []}], "current-schema-id": 0, "partition-specs": [{"spec-id": 0, "fields": [{"source-id": 1, "field-id": 1000, "transform": "identity", "name": "x"}]}], "default-spec-id": 0, "last-partition-id": 1000, "properties": {}, "snapshots": [{"snapshot-id": 1925, "sequence-number": 0, "timestamp-ms": 1602638573822}], "snapshot-log": [], "metadata-log": [], "sort-orders": [{"order-id": 0, "fields": []}], "default-sort-order-id": 0, "refs": {}, "format-version": 1, "schema": {"type": "struct", "fields": [{"id": 1, "name": "x", "type": "long", "required": true}, {"id": 2, "name": "y", "type": "long", "required": true, "doc": "comment"}, {"id": 3, "name": "z", "type": "long", "required": true}], "schema-id": 0, "identifier-field-ids": []}, "partition-spec": [{"name": "x", "transform": "identity", "source-id": 1, "field-id": 1000}]}`,
-		string(data))
+	assert.JSONEq(t, `{
+		"location": "s3://bucket/test/location", 
+		"table-uuid": "d20125c8-7284-442c-9aea-15fee620737c", 
+		"last-updated-ms": 1602638573874, 
+		"last-column-id": 3, 
+		"schemas": [
+			{
+				"type": "struct", 
+				"fields": [
+					{"id": 1, "name": "x", "type": "long", "required": true}, 
+					{"id": 2, "name": "y", "type": "long", "required": true, "doc": "comment"}, 
+					{"id": 3, "name": "z", "type": "long", "required": true}
+				], 
+				"schema-id": 0, 
+				"identifier-field-ids": []
+			}
+		], 
+		"current-schema-id": 0, 
+		"partition-specs": [
+			{
+				"spec-id": 0, 
+				"fields": [
+					{"source-id": 1, "field-id": 1000, "transform": "identity", "name": "x"}
+				]
+			}
+		], 
+		"default-spec-id": 0, 
+		"last-partition-id": 1000, 
+		"snapshots": [
+			{"snapshot-id": 1925, "sequence-number": 0, "timestamp-ms": 1602638573822}
+		], 
+		"sort-orders": [{"order-id": 0, "fields": []}], 
+		"default-sort-order-id": 0,
+		"format-version": 1, 
+		"schema": {
+			"type": "struct", 
+			"fields": [
+				{"id": 1, "name": "x", "type": "long", "required": true}, 
+				{"id": 2, "name": "y", "type": "long", "required": true, "doc": "comment"}, 
+				{"id": 3, "name": "z", "type": "long", "required": true}
+			], 
+			"schema-id": 0, 
+			"identifier-field-ids": []
+		}, 
+		"partition-spec": [
+			{"name": "x", "transform": "identity", "source-id": 1, "field-id": 1000}
+		]
+	}`, string(data))
 }
 
 func TestSerializeMetadataV2(t *testing.T) {
@@ -615,3 +661,91 @@ func TestNewMetadataV2Format(t *testing.T) {
 
 	assert.Truef(t, expected.Equals(actual), "expected: %s\ngot: %s", expected, actual)
 }
+
+func TestMetadataV1Serialize(t *testing.T) {
+	sc := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "int", Type: iceberg.PrimitiveTypes.Int32})
+	toserialize := &metadataV1{
+		commonMetadata: commonMetadata{
+			FormatVersion:      1,
+			UUID:               uuid.MustParse("dd93fa46-a1a7-43bb-8748-6cc7eff107a3"),
+			Loc:                "s3a://warehouse/iceberg/iceberg-test-2.db/test-table-2",
+			LastUpdatedMS:      1742412491193,
+			LastColumnId:       1,
+			SchemaList:         []*iceberg.Schema{sc},
+			CurrentSchemaID:    0,
+			Specs:              []iceberg.PartitionSpec{*iceberg.UnpartitionedSpec},
+			DefaultSpecID:      0,
+			SortOrderList:      []SortOrder{UnsortedSortOrder},
+			DefaultSortOrderID: 0,
+		},
+	}
+
+	data, err := json.Marshal(toserialize)
+	require.NoError(t, err)
+	assert.JSONEq(t, `{		
+		"format-version":1,
+		"table-uuid":"dd93fa46-a1a7-43bb-8748-6cc7eff107a3",
+		"location":"s3a://warehouse/iceberg/iceberg-test-2.db/test-table-2",
+		"last-updated-ms":1742412491193,
+		"last-column-id":1,
+		"schemas": [
+			{
+				"type":"struct",
+				"fields":[{"type":"int","id":1,"name":"int","required":false}],
+				"schema-id":0,
+				"identifier-field-ids":[]
+			}
+		],
+		"current-schema-id":0,
+		"partition-specs":[{"spec-id":0,"fields":[]}],
+		"default-spec-id":0,
+		"sort-orders":[{"order-id":0,"fields":[]}],
+		"default-sort-order-id":0
+	}`, string(data))
+}
+
+func TestMetadataV2Serialize(t *testing.T) {
+	sc := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "int", Type: iceberg.PrimitiveTypes.Int32})
+	toserialize := &metadataV2{
+		LastSeqNum: 1,
+		commonMetadata: commonMetadata{
+			FormatVersion:      1,
+			UUID:               uuid.MustParse("dd93fa46-a1a7-43bb-8748-6cc7eff107a3"),
+			Loc:                "s3a://warehouse/iceberg/iceberg-test-2.db/test-table-2",
+			LastUpdatedMS:      1742412491193,
+			LastColumnId:       1,
+			SchemaList:         []*iceberg.Schema{sc},
+			CurrentSchemaID:    0,
+			Specs:              []iceberg.PartitionSpec{*iceberg.UnpartitionedSpec},
+			DefaultSpecID:      0,
+			SortOrderList:      []SortOrder{UnsortedSortOrder},
+			DefaultSortOrderID: 0,
+		},
+	}
+
+	data, err := json.Marshal(toserialize)
+	require.NoError(t, err)
+	assert.JSONEq(t, `{
+		"last-sequence-number": 1,
+		"format-version":1,
+		"table-uuid":"dd93fa46-a1a7-43bb-8748-6cc7eff107a3",
+		"location":"s3a://warehouse/iceberg/iceberg-test-2.db/test-table-2",
+		"last-updated-ms":1742412491193,
+		"last-column-id":1,
+		"schemas": [
+			{
+				"type":"struct",
+				"fields":[{"type":"int","id":1,"name":"int","required":false}],
+				"schema-id":0,
+				"identifier-field-ids":[]
+			}
+		],
+		"current-schema-id":0,
+		"partition-specs":[{"spec-id":0,"fields":[]}],
+		"default-spec-id":0,
+		"sort-orders":[{"order-id":0,"fields":[]}],
+		"default-sort-order-id":0
+	}`, string(data))
+}
diff --git a/table/table_test.go b/table/table_test.go
index c061459..825fb7d 100644
--- a/table/table_test.go
+++ b/table/table_test.go
@@ -280,10 +280,10 @@ func (t *TableWritingTestSuite) writeParquet(fio iceio.WriteFileIO, filePath str
 
 func (t *TableWritingTestSuite) createTable(identifier table.Identifier, formatVersion int, spec iceberg.PartitionSpec, sc *iceberg.Schema) *table.Table {
 	meta, err := table.NewMetadata(sc, &spec, table.UnsortedSortOrder,
-		t.getMetadataLoc(), iceberg.Properties{"format-version": strconv.Itoa(formatVersion)})
+		t.location, iceberg.Properties{"format-version": strconv.Itoa(formatVersion)})
 	t.Require().NoError(err)
 
-	return table.New(identifier, meta, t.location, iceio.LocalFS{}, nil)
+	return table.New(identifier, meta, t.getMetadataLoc(), iceio.LocalFS{}, nil)
 }
 
 func (t *TableWritingTestSuite) TestAddFilesUnpartitioned() {