Repository: apache/iceberg-go
Pull request #448 (2025-05-30 22:26:13)

testing(catalog/glue): Verify glue catalog snapshot management is supported
### Description
* With commit table implemented in glue catalog https://github.com/apache/iceberg-go/pull/403, snapshot management is supported. 
* Add integration test to verify and showcase the usage
* Update README to reflect the status
* Reuse `cleanupTable` as a private util method 

### Testing
```
--- PASS: TestSnapshotManagementIntegration (7.10s)
PASS
```


Patch:
diff --git a/README.md b/README.md
index b1e2cc2..63d277c 100644
--- a/README.md
+++ b/README.md
@@ -66,8 +66,8 @@ $ cd iceberg-go/cmd/iceberg && go build .
 | List Tables                 |  X   |      |          |  X   |  X  |
 | Create Table                |  X   |      |          |  X   |  X  |
 | Register Table              |  X   |      |          |  X   |     |
-| Update Current Snapshot     |  X   |      |          |      |  X  |
-| Create New Snapshot         |  X   |      |          |      |  X  |
+| Update Current Snapshot     |  X   |      |          |  X   |  X  |
+| Create New Snapshot         |  X   |      |          |  X   |  X  |
 | Rename Table                |  X   |      |          |  X   |  X  |
 | Drop Table                  |  X   |      |          |  X   |  X  |
 | Alter Table                 |  X   |      |          |  X   |  X  |
diff --git a/dev/release/release.sh b/dev/release/release.sh
index 46c4755..bb81b87 100755
--- a/dev/release/release.sh
+++ b/dev/release/release.sh
@@ -40,16 +40,30 @@ fi
 tag="v${version}"
 rc_tag="${tag}-rc${rc}"
 echo "Tagging for release: ${tag}"
-git tag "${tag}" "${rc_tag}"
+git tag "${tag}" "${rc_tag}^{}" -m "Release ${tag}"
 git push origin "${tag}"
 
+release_id="apache-iceberg-go-${version}"
 dist_url="https://dist.apache.org/repos/dist/release/iceberg"
 dist_dev_url="https://dist.apache.org/repos/dist/dev/iceberg"
 
 svn \
-  mv "${dist_dev_url}/apache-iceberg-go-${version}-rc${rc}/" \
-  "${dist_url}/apache-iceberg-go-${version}/" \
-  -m "Apache Iceberg-go ${version}"
+  mv "${dist_dev_url}/${release_id}-rc${rc}/" \
+  "${dist_url}/${release_id}" \
+  -m "Apache Iceberg Go ${version}"
+
+svn co "${dist_url}/${release_id}"
+pushd "${release_id}"
+gh release create "${tag}" \
+  --title "Apache Iceberg Go ${version}" \
+  --generate-notes \
+  --verify-tag \
+  ${release_id}.tar.gz \
+  ${release_id}.tar.gz.asc \
+  ${release_id}.tar.gz.sha512
+popd
+
+rm -rf "${release_id}"
 
 echo "Keep only the latest versions"
 old_releases=$(
diff --git a/dev/release/release_rc.sh b/dev/release/release_rc.sh
index 38b9e40..8a27ad6 100755
--- a/dev/release/release_rc.sh
+++ b/dev/release/release_rc.sh
@@ -98,15 +98,25 @@ if [ "${RELEASE_SIGN}" -gt 0 ]; then
     --repo "${repository}" \
     --skip-existing
 
-  echo "Signing tar.gz and creating checksums"
+  echo "Signing tar.gz"
   cd "${id}"
   gpg --armor --output "${tar_gz}.asc" --detach-sig "${tar_gz}"
-  sha512sum "${tar_gz}" > "${tar_gz}.sha512"
+  echo "Add signature to GitHub release"
+  gh release upload "${rc_tag}" \
+    --clobber \
+    --repo "${repository}" \
+    "${tar_gz}".asc
   cd ..
 fi
 
 if [ "${RELEASE_UPLOAD}" -gt 0 ]; then
   echo "Uploading to ASF dist/dev..."
+  # rename files to remove -rc${rc} suffix before uploading
+  pushd "${id}"
+  for fname in "./*"; do
+    mv "${fname}" "${fname//-rc${rc}/}"
+  done
+  popd
   svn import "${id}" "https://dist.apache.org/repos/dist/dev/iceberg/${id}" -m "Apache Iceberg Go ${version} RC${rc}"
 fi
 
diff --git a/table/evaluators.go b/table/evaluators.go
index acee1c3..c7d012f 100644
--- a/table/evaluators.go
+++ b/table/evaluators.go
@@ -751,11 +751,18 @@ func (m *inclusiveMetricsEval) Eval(file iceberg.DataFile) (bool, error) {
 		return rowsCannotMatch, nil
 	}
 
-	m.valueCounts, m.nullCounts = file.ValueCounts(), file.NullValueCounts()
-	m.nanCounts = file.NaNValueCounts()
-	m.lowerBounds, m.upperBounds = file.LowerBoundValues(), file.UpperBoundValues()
+	// avoid race condition while maintaining existing state
+	ev := inclusiveMetricsEval{
+		st:                m.st,
+		includeEmptyFiles: m.includeEmptyFiles,
+		expr:              m.expr,
+	}
 
-	return iceberg.VisitExpr(m.expr, m)
+	ev.valueCounts, ev.nullCounts = file.ValueCounts(), file.NullValueCounts()
+	ev.nanCounts = file.NaNValueCounts()
+	ev.lowerBounds, ev.upperBounds = file.LowerBoundValues(), file.UpperBoundValues()
+
+	return iceberg.VisitExpr(m.expr, &ev)
 }
 
 func (m *inclusiveMetricsEval) mayContainNull(fieldID int) bool {
diff --git a/transforms.go b/transforms.go
index fb37e6c..4cd02f5 100644
--- a/transforms.go
+++ b/transforms.go
@@ -720,7 +720,7 @@ func (t DayTransform) MarshalText() ([]byte, error) {
 
 func (DayTransform) String() string { return "day" }
 
-func (DayTransform) ResultType(Type) Type { return PrimitiveTypes.Date }
+func (DayTransform) ResultType(Type) Type { return PrimitiveTypes.Int32 }
 func (DayTransform) PreservesOrder() bool { return true }
 
 func (DayTransform) Equals(other Transform) bool {


Test patch:
diff --git a/catalog/glue/glue_test.go b/catalog/glue/glue_test.go
index 9cf7efd..2a685e1 100644
--- a/catalog/glue/glue_test.go
+++ b/catalog/glue/glue_test.go
@@ -864,31 +864,7 @@ func TestGlueCreateTableSuccessIntegration(t *testing.T) {
 		catalog.WithLocation(metadataLocation),
 	}
 	newTable, err := ctlg.CreateTable(context.TODO(), TableIdentifier(dbName, newTableName), sourceTable.Schema(), createOpts...)
-	defer func() {
-		cleanupErr := ctlg.DropTable(context.TODO(), TableIdentifier(dbName, newTableName))
-		if cleanupErr != nil {
-			t.Logf("Warning: Failed to clean up table %s.%s: %v", dbName, newTableName, cleanupErr)
-		}
-		// Clean up the newly created metadata file in S3
-		if newTable != nil {
-			s3Client := s3.NewFromConfig(awsCfg)
-			metadataLoc := newTable.MetadataLocation()
-			if strings.HasPrefix(metadataLoc, "s3://") {
-				parts := strings.SplitN(strings.TrimPrefix(metadataLoc, "s3://"), "/", 2)
-				if len(parts) == 2 {
-					bucket := parts[0]
-					key := parts[1]
-					_, deleteErr := s3Client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{
-						Bucket: aws.String(bucket),
-						Key:    aws.String(key),
-					})
-					if deleteErr != nil {
-						t.Logf("Warning: Failed to delete metadata file at %s: %v", metadataLoc, deleteErr)
-					}
-				}
-			}
-		}
-	}()
+	defer cleanupTable(t, ctlg, TableIdentifier(dbName, newTableName), awsCfg)
 	assert.NoError(err)
 	assert.Equal([]string{dbName, newTableName}, newTable.Identifier())
 
@@ -1067,31 +1043,7 @@ func TestAlterTableIntegration(t *testing.T) {
 	assert.True(schema.Equals(testTable.Schema()))
 
 	// Clean up table and table location after tests
-	defer func() {
-		cleanupErr := ctlg.DropTable(context.TODO(), TableIdentifier(dbName, tbName))
-		if cleanupErr != nil {
-			t.Logf("Warning: Failed to clean up table %s.%s: %v", dbName, tbName, cleanupErr)
-		}
-		// Clean up the newly created metadata file in S3
-		if testTable != nil {
-			s3Client := s3.NewFromConfig(awsCfg)
-			metadataLoc := testTable.MetadataLocation()
-			if strings.HasPrefix(metadataLoc, "s3://") {
-				parts := strings.SplitN(strings.TrimPrefix(metadataLoc, "s3://"), "/", 2)
-				if len(parts) == 2 {
-					bucket := parts[0]
-					key := parts[1]
-					_, deleteErr := s3Client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{
-						Bucket: aws.String(bucket),
-						Key:    aws.String(key),
-					})
-					if deleteErr != nil {
-						t.Logf("Warning: Failed to delete metadata file at %s: %v", metadataLoc, deleteErr)
-					}
-				}
-			}
-		}
-	}()
+	defer cleanupTable(t, ctlg, tbIdent, awsCfg)
 
 	// Test set table properties
 	updateProps := table.NewSetPropertiesUpdate(map[string]string{
@@ -1159,6 +1111,83 @@ func TestAlterTableIntegration(t *testing.T) {
 	assert.Equal(newFields, testTable.Schema().Fields())
 }
 
+func TestSnapshotManagementIntegration(t *testing.T) {
+	if os.Getenv("TEST_DATABASE_NAME") == "" || os.Getenv("TEST_TABLE_LOCATION") == "" {
+		t.Skip()
+	}
+
+	assert := require.New(t)
+	dbName := os.Getenv("TEST_DATABASE_NAME")
+	tbLocation := os.Getenv("TEST_TABLE_LOCATION")
+	tbName := fmt.Sprintf("table_%d", time.Now().UnixNano())
+	tbIdent := TableIdentifier(dbName, tbName)
+
+	awsCfg, err := config.LoadDefaultConfig(context.TODO(), config.WithClientLogMode(aws.LogRequest|aws.LogResponse))
+	assert.NoError(err)
+	ctlg := NewCatalog(WithAwsConfig(awsCfg))
+
+	// clean up table after test
+	defer cleanupTable(t, ctlg, tbIdent, awsCfg)
+
+	createOpts := []catalog.CreateTableOpt{
+		catalog.WithLocation(tbLocation),
+	}
+	_, err = ctlg.CreateTable(context.TODO(), tbIdent, testSchema, createOpts...)
+	assert.NoError(err)
+
+	testTable, err := ctlg.LoadTable(context.TODO(), tbIdent, nil)
+	assert.NoError(err)
+
+	// Test add new snapshot
+	manifest, schemaid := "s3:/a/b/c.avro", 3
+	newSnap := table.Snapshot{
+		SnapshotID:     25,
+		SequenceNumber: 200,
+		TimestampMs:    1602638573590,
+		ManifestList:   manifest,
+		SchemaID:       &schemaid,
+		Summary: &table.Summary{
+			Operation: table.OpAppend,
+		},
+	}
+
+	_, _, err = ctlg.CommitTable(context.TODO(), testTable, nil, []table.Update{
+		table.NewAddSnapshotUpdate(&newSnap),
+	})
+	assert.NoError(err)
+
+	testTable, err = ctlg.LoadTable(context.TODO(), tbIdent, nil)
+	assert.NoError(err)
+
+	actualSnap := testTable.SnapshotByID(25)
+	assert.Equal(newSnap.SnapshotID, actualSnap.SnapshotID)
+	assert.Equal(newSnap.ParentSnapshotID, actualSnap.ParentSnapshotID)
+	assert.Equal(newSnap.SequenceNumber, actualSnap.SequenceNumber)
+	assert.Equal(newSnap.ManifestList, actualSnap.ManifestList)
+	assert.Equal(newSnap.TimestampMs, actualSnap.TimestampMs)
+	assert.Equal(*newSnap.SchemaID, *actualSnap.SchemaID)
+	assert.Equal(newSnap.Summary.Operation, actualSnap.Summary.Operation)
+
+	// Test update current snapshot
+	_, _, err = ctlg.CommitTable(context.TODO(), testTable, nil, []table.Update{
+		table.NewSetSnapshotRefUpdate(table.MainBranch, 25, table.BranchRef,
+			-1, -1, -1),
+	})
+	assert.NoError(err)
+
+	testTable, err = ctlg.LoadTable(context.TODO(), tbIdent, nil)
+	assert.NoError(err)
+
+	currSnap := testTable.CurrentSnapshot()
+	assert.Equal(newSnap.SnapshotID, currSnap.SnapshotID)
+	assert.Equal(newSnap.ParentSnapshotID, actualSnap.ParentSnapshotID)
+	assert.Equal(newSnap.SequenceNumber, currSnap.SequenceNumber)
+	assert.Equal(newSnap.ManifestList, currSnap.ManifestList)
+	assert.Equal(newSnap.TimestampMs, currSnap.TimestampMs)
+	assert.Equal(*newSnap.SchemaID, *currSnap.SchemaID)
+	assert.Equal(newSnap.Summary.Operation, currSnap.Summary.Operation)
+}
+
 func TestGlueCheckTableExists(t *testing.T) {
 	assert := require.New(t)
 	mockGlueSvc := &mockGlueClient{}
@@ -1192,3 +1221,36 @@ func TestGlueCheckTableNotExists(t *testing.T) {
 	assert.Nil(err)
 	assert.False(exists)
 }
+
+func cleanupTable(t *testing.T, ctlg catalog.Catalog, tbIdent table.Identifier, awsCfg aws.Config) {
+	t.Helper()
+
+	testTable, err := ctlg.LoadTable(context.TODO(), tbIdent, nil)
+	if err != nil {
+		t.Logf("Warning: Failed to load table %s: %v", tbIdent, err)
+	}
+
+	cleanupErr := ctlg.DropTable(context.TODO(), tbIdent)
+	if cleanupErr != nil {
+		t.Logf("Warning: Failed to clean up table %s: %v", tbIdent, cleanupErr)
+	}
+
+	if testTable != nil {
+		s3Client := s3.NewFromConfig(awsCfg)
+		metadataLoc := testTable.MetadataLocation()
+		if strings.HasPrefix(metadataLoc, "s3://") {
+			parts := strings.SplitN(strings.TrimPrefix(metadataLoc, "s3://"), "/", 2)
+			if len(parts) == 2 {
+				bucket := parts[0]
+				key := parts[1]
+				_, deleteErr := s3Client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{
+					Bucket: aws.String(bucket),
+					Key:    aws.String(key),
+				})
+				if deleteErr != nil {
+					t.Logf("Warning: Failed to delete metadata file at %s: %v", metadataLoc, deleteErr)
+				}
+			}
+		}
+	}
+}
diff --git a/transforms_test.go b/transforms_test.go
index 3e7a6d5..4abebeb 100644
--- a/transforms_test.go
+++ b/transforms_test.go
@@ -18,8 +18,11 @@
 package iceberg_test
 
 import (
+	"bytes"
+	"reflect"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/apache/arrow-go/v18/arrow/decimal"
 	"github.com/apache/iceberg-go"
@@ -139,3 +142,98 @@ func TestToHumanString(t *testing.T) {
 		})
 	}
 }
+
+func TestManifestPartitionVals(t *testing.T) {
+	// Sanity checks that the source and result types of the transform are
+	// compatible with their use to generate partition data in manifests.
+	ts := time.Date(1971, 2, 10, 10, 20, 30, 4_000_000, time.UTC)
+	tests := []struct {
+		transform    iceberg.Transform
+		input        iceberg.Literal
+		expectResult iceberg.Literal
+	}{
+		{
+			transform:    iceberg.HourTransform{},
+			input:        iceberg.TimestampLiteral(ts.UnixMicro()),
+			expectResult: iceberg.Int32Literal((365+40)*24 + 10),
+		},
+		{
+			transform:    iceberg.DayTransform{},
+			input:        iceberg.TimestampLiteral(ts.UnixMicro()),
+			expectResult: iceberg.Int32Literal(365 + 40),
+		},
+		{
+			transform:    iceberg.MonthTransform{},
+			input:        iceberg.TimestampLiteral(ts.UnixMicro()),
+			expectResult: iceberg.Int32Literal(13),
+		},
+		{
+			transform:    iceberg.YearTransform{},
+			input:        iceberg.TimestampLiteral(ts.UnixMicro()),
+			expectResult: iceberg.Int32Literal(1),
+		},
+		{
+			transform:    iceberg.TruncateTransform{Width: 100},
+			input:        iceberg.Int64Literal(123456789),
+			expectResult: iceberg.Int64Literal(123456700),
+		},
+		{
+			transform:    iceberg.IdentityTransform{},
+			input:        iceberg.StringLiteral("foobar"),
+			expectResult: iceberg.StringLiteral("foobar"),
+		},
+		{
+			transform:    iceberg.BucketTransform{NumBuckets: 128},
+			input:        iceberg.StringLiteral("foobar"),
+			expectResult: iceberg.Int32Literal(61),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(reflect.TypeOf(tt.transform).String(), func(t *testing.T) {
+			result := tt.transform.Apply(iceberg.Optional[iceberg.Literal]{Val: tt.input, Valid: true})
+			require.True(t, result.Valid)
+			assert.Equal(t, tt.expectResult, result.Val)
+
+			schema := iceberg.NewSchema(0, iceberg.NestedField{
+				Name: "abc",
+				ID:   1,
+				Type: tt.input.Type(),
+			})
+			partitionSpec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+				Name:      "transformed_abc",
+				SourceID:  1,
+				FieldID:   1000,
+				Transform: tt.transform,
+			})
+			dataFile, err := iceberg.NewDataFileBuilder(
+				partitionSpec, iceberg.EntryContentData,
+				"1234.parquet", iceberg.ParquetFile,
+				map[int]any{1000: result.Val.Any()},
+				100, 100_000,
+			)
+			require.NoError(t, err)
+			snapshotID := int64(123)
+			seqNum := int64(456)
+			manifestEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapshotID, &seqNum, &seqNum, dataFile.Build())
+			var buf bytes.Buffer
+			manifestFile, err := iceberg.WriteManifest(
+				"abc.avro", &buf,
+				2, partitionSpec, schema, 123,
+				[]iceberg.ManifestEntry{manifestEntry},
+			)
+			require.NoError(t, err)
+
+			// Now make sure we can deserialize and re-serialize the manifest, too.
+			entries, err := iceberg.ReadManifest(manifestFile, &buf, false)
+			require.NoError(t, err)
+
+			buf.Reset()
+			_, err = iceberg.WriteManifest(
+				"abc.avro", &buf,
+				2, partitionSpec, schema, 123,
+				entries,
+			)
+			require.NoError(t, err)
+		})
+	}
+}