Repository: apache/iceberg-go
Pull request #400 (2025-04-22 06:33:56)

fix(catalog/glue): Fix glue integration test failures
### Description
* Fix glue integration test failures. Mostly caused by asserting table identifiers, which require both database name and table name.

### Testing
* Ran the integ tests locally with AWS environment variables set


Patch:
diff --git a/manifest.go b/manifest.go
index 6b0f4a9..e09e892 100644
--- a/manifest.go
+++ b/manifest.go
@@ -450,6 +450,10 @@ func fetchManifestEntries(m ManifestFile, fs iceio.IO, discardDeleted bool) ([]M
 	}
 	defer f.Close()
 
+	return readManifestEntries(m, f, discardDeleted)
+}
+
+func readManifestEntries(m ManifestFile, f io.Reader, discardDeleted bool) ([]ManifestEntry, error) {
 	dec, err := ocf.NewDecoder(f, ocf.WithDecoderSchemaCache(&avro.SchemaCache{}))
 	if err != nil {
 		return nil, err


Test patch:
diff --git a/catalog/glue/glue_test.go b/catalog/glue/glue_test.go
index 8f6e556..7de4012 100644
--- a/catalog/glue/glue_test.go
+++ b/catalog/glue/glue_test.go
@@ -781,17 +781,17 @@ func TestGlueListTablesIntegration(t *testing.T) {
 	ctlg := NewCatalog(WithAwsConfig(awsCfg))
 
 	iter := ctlg.ListTables(context.TODO(), DatabaseIdentifier(os.Getenv("TEST_DATABASE_NAME")))
-	var lastErr error
-	tbls := make([]table.Identifier, 0)
+
+	found := false
 	for tbl, err := range iter {
-		tbls = append(tbls, tbl)
-		if err != nil {
-			lastErr = err
+		assert.NoError(err)
+		if tbl[1] == os.Getenv("TEST_TABLE_NAME") {
+			found = true
+
+			break
 		}
 	}
-
-	assert.NoError(lastErr)
-	assert.Equal([]string{os.Getenv("TEST_DATABASE_NAME"), os.Getenv("TEST_TABLE_NAME")}, tbls[1])
+	assert.True(found, "expect test table name exists to be part of the list table results")
 }
 
 func TestGlueLoadTableIntegration(t *testing.T) {
@@ -814,7 +814,7 @@ func TestGlueLoadTableIntegration(t *testing.T) {
 
 	tbl, err := ctlg.LoadTable(context.TODO(), []string{os.Getenv("TEST_DATABASE_NAME"), os.Getenv("TEST_TABLE_NAME")}, nil)
 	assert.NoError(err)
-	assert.Equal([]string{os.Getenv("TEST_TABLE_NAME")}, tbl.Identifier())
+	assert.Equal([]string{os.Getenv("TEST_DATABASE_NAME"), os.Getenv("TEST_TABLE_NAME")}, tbl.Identifier())
 }
 
 func TestGlueListNamespacesIntegration(t *testing.T) {
@@ -852,7 +852,7 @@ func TestGlueCreateTableSuccessIntegration(t *testing.T) {
 	ctlg := NewCatalog(WithAwsConfig(awsCfg))
 	sourceTable, err := ctlg.LoadTable(context.TODO(), []string{dbName, sourceTableName}, nil)
 	assert.NoError(err)
-	assert.Equal([]string{sourceTableName}, sourceTable.Identifier())
+	assert.Equal([]string{dbName, sourceTableName}, sourceTable.Identifier())
 	newTableName := fmt.Sprintf("%d_%s", time.Now().UnixNano(), sourceTableName)
 	createOpts := []catalog.CreateTableOpt{
 		catalog.WithLocation(metadataLocation),
@@ -883,13 +883,12 @@ func TestGlueCreateTableSuccessIntegration(t *testing.T) {
 			}
 		}
 	}()
-	assert.True(false)
 	assert.NoError(err)
-	assert.Equal([]string{newTableName}, newTable.Identifier())
+	assert.Equal([]string{dbName, newTableName}, newTable.Identifier())
 
 	tableNewLoaded, err := ctlg.LoadTable(context.TODO(), []string{dbName, newTableName}, nil)
 	assert.NoError(err)
-	assert.Equal([]string{newTableName}, tableNewLoaded.Identifier())
+	assert.Equal([]string{dbName, newTableName}, tableNewLoaded.Identifier())
 	assert.Equal(sourceTable.Schema().Fields(), tableNewLoaded.Schema().Fields())
 	assert.Contains(tableNewLoaded.MetadataLocation(), metadataLocation)
 
@@ -928,7 +927,7 @@ func TestGlueCreateTableInvalidMetadataRollback(t *testing.T) {
 	assert.Error(err, "expected error when creating table with invalid metadata location")
 	_, err = ctlg.LoadTable(context.TODO(), []string{dbName, newTableName}, nil)
 	assert.Error(err, "expected table to not exist after failed creation")
-	assert.True(strings.Contains(err.Error(), "EntityNotFoundException: Entity Not Found"), "expected EntityNotFoundException error")
+	assert.True(strings.Contains(err.Error(), "table does not exist"), "expected EntityNotFoundException error")
 	// Verify that the table was not left in the catalog
 	tablesIter := ctlg.ListTables(context.TODO(), DatabaseIdentifier(dbName)) // TODO: Implement CheckTableExists
 	found := false
diff --git a/manifest_test.go b/manifest_test.go
index d065ab7..aeda897 100644
--- a/manifest_test.go
+++ b/manifest_test.go
@@ -791,6 +791,126 @@ func (m *ManifestTestSuite) TestReadManifestListIncompleteSchema() {
 	m.ErrorContains(err, "unknown type: field_summary")
 }
 
+func (m *ManifestTestSuite) TestReadManifestIncompleteSchema() {
+	// This prevents a regression that could be caused by using a schema cache
+	// across multiple read/write operations of an avro file. While it may sound
+	// like a reasonable idea (caches speed things up, right?), it isn't that
+	// sort of cache: it's really a resolver to allow files with incomplete
+	// schemas, which we don't want.
+
+	// If a schema cache *were* in use, this would populate it with a definition for
+	// the missing record type in the incomplete schema. So we'll first "warm up"
+	// any cache. (Note: if working correctly, this will have no such side effect.)
+	var buf bytes.Buffer
+	partitionSpec := NewPartitionSpecID(1)
+	snapshotID := int64(12345678)
+	seqNum := int64(9876)
+	dataFileBuilder, err := NewDataFileBuilder(
+		partitionSpec,
+		EntryContentData,
+		"s3://bucket/namespace/table/data/abcd-0123.parquet",
+		ParquetFile,
+		map[string]any{},
+		100,
+		100*1000*1000,
+	)
+	m.NoError(err)
+	file, err := WriteManifest(
+		"s3://bucket/namespace/table/metadata/abcd-0123.avro", &buf, 2,
+		partitionSpec,
+		NewSchema(123,
+			NestedField{ID: 1, Name: "id", Type: Int64Type{}},
+			NestedField{ID: 2, Name: "name", Type: StringType{}},
+		),
+		snapshotID,
+		[]ManifestEntry{NewManifestEntry(
+			EntryStatusADDED,
+			&snapshotID,
+			&seqNum, &seqNum,
+			dataFileBuilder.Build(),
+		)},
+	)
+	m.NoError(err)
+
+	entries, err := readManifestEntries(file, &buf, false)
+	m.NoError(err)
+	m.Len(entries, 1)
+
+	// This schema is that of a v2 manifest file, except that it refers to
+	// a type named "r2" for the "data_file" field, instead of actually
+	// including the definition of the "data_file" record type.
+	// This omission should result in an error. But if a schema cache were
+	// in use, this could get resolved based on a type of the same name read
+	// from a file that defined it.
+	incompleteSchema := `
+	{
+		"name": "manifest_entry",
+		"type": "record",
+		"fields": [
+			{
+				"name": "status",
+				"type": "int",
+				"field-id": 0
+			},
+			{
+				"name": "snapshot_id",
+				"type": [
+					"null",
+					"long"
+				],
+				"field-id": 1
+			},
+			{
+				"name": "sequence_number",
+				"type": [
+					"null",
+					"long"
+				],
+				"field-id": 3
+			},
+			{
+				"name": "file_sequence_number",
+				"type": [
+					"null",
+					"long"
+				],
+				"field-id": 4
+			},
+			{
+				"name": "data_file",
+				"type": "r2",
+				"field-id": 2
+			}
+		]
+	}`
+
+	// We'll generate a file that is missing part of its schema
+	cache := &avro.SchemaCache{}
+	partitionSchema, err := avro.NewRecordSchema("r102", "", nil) // empty struct
+	m.NoError(err)
+	sch, err := internal.NewManifestEntrySchema(partitionSchema, 2)
+	m.NoError(err)
+	enc, err := ocf.NewEncoderWithSchema(sch, &buf,
+		ocf.WithEncoderSchemaCache(cache),
+		ocf.WithSchemaMarshaler(func(schema avro.Schema) ([]byte, error) {
+			return []byte(incompleteSchema), nil
+		}),
+		ocf.WithMetadata(map[string][]byte{
+			"format-version": {'2'},
+			// TODO: spec says other things are required, like schema and partition-spec info,
+			//       but this package currently only looks at this one value when reading...
+		}),
+	)
+	m.NoError(err)
+	for _, entry := range entries {
+		m.NoError(enc.Encode(entry))
+	}
+
+	// This should fail because the file's schema is incomplete.
+	_, err = readManifestEntries(file, &buf, false)
+	m.ErrorContains(err, "unknown type: r2")
+}
+
 func (m *ManifestTestSuite) TestManifestEntriesV2() {
 	var mockfs internal.MockFS
 	manifest := manifestFile{