Skip to content

fix(schema): Add duplicate field ID checking mechanism #658

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dttung2905 wants to merge 8 commits into
base: main
Choose a base branch
from
+1,169 −437
Open

fix(schema): Add duplicate field ID checking mechanism #658

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
c34a336
Add duplicate field ID checking mechanism
dttung2905 Dec 15, 2025
fc8bb0e
Small refactoring for the init() func
dttung2905 Dec 25, 2025
6327eb5
Fix CI
dttung2905 Jan 7, 2026
3740668
Fix errors
dttung2905 Jan 25, 2026
0358326
Fix errors
dttung2905 Jan 25, 2026
907fb9b
Fix errors
dttung2905 Jan 25, 2026
da73141
Fix errors
dttung2905 Jan 25, 2026
43e5602
Fix errors
dttung2905 Jan 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 11 additions & 6 deletions catalog/glue/glue_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -172,7 +172,7 @@ var testNonIcebergGlueTable = types.Table{
},
}

var testSchema = iceberg.NewSchemaWithIdentifiers(0, []int{},
var testSchema = iceberg.MustNewSchemaWithIdentifiers(0, []int{},
iceberg.NestedField{ID: 1, Name: "foo", Type: iceberg.PrimitiveTypes.String},
iceberg.NestedField{ID: 2, Name: "bar", Type: iceberg.PrimitiveTypes.Int32, Required: true},
iceberg.NestedField{ID: 3, Name: "baz", Type: iceberg.PrimitiveTypes.Bool})
Expand Down Expand Up @@ -1005,10 +1005,11 @@ func TestGlueCreateTableInvalidMetadataRollback(t *testing.T) {
func TestGlueCreateTableRollbackOnInvalidMetadata(t *testing.T) {
assert := require.New(t)
mockGlueSvc := &mockGlueClient{}
schema := iceberg.NewSchemaWithIdentifiers(1, []int{1},
schema, err := iceberg.NewSchemaWithIdentifiers(1, []int{1},
iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.Int64Type{}, Required: true},
iceberg.NestedField{ID: 2, Name: "name", Type: iceberg.StringType{}, Required: true},
)
assert.NoError(err)
mockGlueSvc.On("CreateTable", mock.Anything, mock.Anything, mock.Anything).Return(&glue.CreateTableOutput{}, nil)
mockGlueSvc.On("GetTable", mock.Anything, &glue.GetTableInput{
DatabaseName: aws.String("test_database"),
Expand All @@ -1026,7 +1027,7 @@ func TestGlueCreateTableRollbackOnInvalidMetadata(t *testing.T) {
glueSvc: mockGlueSvc,
awsCfg: &aws.Config{},
}
_, err := glueCatalog.CreateTable(context.TODO(),
_, err = glueCatalog.CreateTable(context.TODO(),
TableIdentifier("test_database", "test_rollback_table"),
schema,
catalog.WithLocation("s3://non-existent-test-bucket"))
Expand Down Expand Up @@ -1094,10 +1095,11 @@ func TestAlterTableIntegration(t *testing.T) {
metadataLocation := os.Getenv("TEST_TABLE_LOCATION")
tbName := fmt.Sprintf("table_%d", time.Now().UnixNano())
tbIdent := TableIdentifier(dbName, tbName)
schema := iceberg.NewSchemaWithIdentifiers(0, []int{},
schema, err := iceberg.NewSchemaWithIdentifiers(0, []int{},
iceberg.NestedField{ID: 1, Name: "foo", Type: iceberg.PrimitiveTypes.String},
iceberg.NestedField{ID: 2, Name: "bar", Type: iceberg.PrimitiveTypes.Int32},
iceberg.NestedField{ID: 3, Name: "baz", Type: iceberg.PrimitiveTypes.Bool})
assert.NoError(err)

awsCfg, err := config.LoadDefaultConfig(context.TODO(), config.WithClientLogMode(aws.LogRequest|aws.LogResponse))
assert.NoError(err)
Expand Down Expand Up @@ -1169,7 +1171,9 @@ func TestAlterTableIntegration(t *testing.T) {
}
newFields := append(currentSchema.Fields(), addField) // add column 'new_col'
newFields = append(newFields[:1], newFields[2:]...) // drop column 'bar'
updateColumns := table.NewAddSchemaUpdate(iceberg.NewSchemaWithIdentifiers(newSchemaId, currentSchema.IdentifierFieldIDs, newFields...))
newSchema, err := iceberg.NewSchemaWithIdentifiers(newSchemaId, currentSchema.IdentifierFieldIDs, newFields...)
require.NoError(t, err)
updateColumns := table.NewAddSchemaUpdate(newSchema)
setSchema := table.NewSetCurrentSchemaUpdate(newSchemaId)

_, _, err = ctlg.CommitTable(
Expand Down Expand Up @@ -1342,9 +1346,10 @@ func TestCommitTableOptimisticLockingIntegration(t *testing.T) {
tbName := fmt.Sprintf("optimistic_lock_test_%d", time.Now().UnixNano())
tbIdent := TableIdentifier(dbName, tbName)

schema := iceberg.NewSchemaWithIdentifiers(0, []int{},
schema, err := iceberg.NewSchemaWithIdentifiers(0, []int{},
iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64},
iceberg.NestedField{ID: 2, Name: "data", Type: iceberg.PrimitiveTypes.String})
assert.NoError(err)

awsCfg, err := config.LoadDefaultConfig(context.TODO(), config.WithClientLogMode(aws.LogRequest|aws.LogResponse))
assert.NoError(err)
Expand Down
77 changes: 40 additions & 37 deletions catalog/glue/schema_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import (
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/service/glue/types"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)

func TestIcebergTypeToGlueType(t *testing.T) {
Expand Down Expand Up @@ -349,7 +350,8 @@ func TestSchemaToGlueColumns(t *testing.T) {
Doc: "User tags",
},
}
schema := iceberg.NewSchema(1, fields...)
schema, err := iceberg.NewSchema(1, fields...)
require.NoError(t, err)
columns := schemaToGlueColumns(schema, true)
assert.Equal(t, 4, len(columns))
assert.Equal(t, "id", aws.ToString(columns[0].Name))
Expand All @@ -365,42 +367,43 @@ func TestSchemaToGlueColumns(t *testing.T) {
}

func TestSchemasToGlueColumns(t *testing.T) {
schemas := []*iceberg.Schema{
iceberg.NewSchema(0,
iceberg.NestedField{
ID: 1,
Name: "id",
Type: iceberg.Int64Type{},
Required: true,
},
iceberg.NestedField{
ID: 2,
Name: "name",
Type: iceberg.StringType{},
Required: true,
},
iceberg.NestedField{
ID: 3,
Name: "address",
Type: iceberg.StringType{},
Required: false,
},
),
iceberg.NewSchema(1,
iceberg.NestedField{
ID: 1,
Name: "id",
Type: iceberg.Int64Type{},
Required: true,
},
iceberg.NestedField{
ID: 2,
Name: "name",
Type: iceberg.StringType{},
Required: true,
},
),
}
schema0, err := iceberg.NewSchema(0,
iceberg.NestedField{
ID: 1,
Name: "id",
Type: iceberg.Int64Type{},
Required: true,
},
iceberg.NestedField{
ID: 2,
Name: "name",
Type: iceberg.StringType{},
Required: true,
},
iceberg.NestedField{
ID: 3,
Name: "address",
Type: iceberg.StringType{},
Required: false,
},
)
require.NoError(t, err)
schema1, err := iceberg.NewSchema(1,
iceberg.NestedField{
ID: 1,
Name: "id",
Type: iceberg.Int64Type{},
Required: true,
},
iceberg.NestedField{
ID: 2,
Name: "name",
Type: iceberg.StringType{},
Required: true,
},
)
require.NoError(t, err)
schemas := []*iceberg.Schema{schema0, schema1}

expectedColumns := []types.Column{
{
Expand Down
6 changes: 5 additions & 1 deletion catalog/internal/utils.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,7 +201,11 @@ func UpdateAndStageTable(ctx context.Context, current *table.Table, ident table.
metadataLoc = current.MetadataLocation()
} else {
var err error
baseMeta, err = table.NewMetadata(iceberg.NewSchema(0), nil, table.UnsortedSortOrder, "", nil)
emptySchema, err := iceberg.NewSchema(0)
if err != nil {
return nil, err
}
baseMeta, err = table.NewMetadata(emptySchema, nil, table.UnsortedSortOrder, "", nil)
if err != nil {
return nil, err
}
Expand Down
2 changes: 1 addition & 1 deletion catalog/rest/rest_integration_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ func (s *RestIntegrationSuite) SetupTest() {
s.cat = s.loadCatalog(s.ctx)
}

var tableSchemaNested = iceberg.NewSchemaWithIdentifiers(1,
var tableSchemaNested = iceberg.MustNewSchemaWithIdentifiers(1,
[]int{1},
iceberg.NestedField{
ID: 1, Name: "foo", Type: iceberg.PrimitiveTypes.String, Required: true,
Expand Down
11 changes: 7 additions & 4 deletions catalog/rest/rest_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1150,7 +1150,7 @@ var (
}
}`, exampleTableMetadataNoSnapshotV1)

tableSchemaSimple = iceberg.NewSchemaWithIdentifiers(1, []int{2},
tableSchemaSimple = iceberg.MustNewSchemaWithIdentifiers(1, []int{2},
iceberg.NestedField{ID: 1, Name: "foo", Type: iceberg.StringType{}, Required: false},
iceberg.NestedField{ID: 2, Name: "bar", Type: iceberg.PrimitiveTypes.Int32, Required: true},
iceberg.NestedField{ID: 3, Name: "baz", Type: iceberg.PrimitiveTypes.Bool, Required: false},
Expand Down Expand Up @@ -2329,12 +2329,13 @@ func (r *RestCatalogSuite) TestCreateView200() {
ns := "ns"
viewName := "view"
identifier := table.Identifier{ns, viewName}
schema := iceberg.NewSchemaWithIdentifiers(0, []int{1}, iceberg.NestedField{
schema, err := iceberg.NewSchemaWithIdentifiers(0, []int{1}, iceberg.NestedField{
ID: 1,
Name: "id",
Type: iceberg.PrimitiveTypes.Int32,
Required: true,
})
r.Require().NoError(err)
config := iceberg.Properties{
"comment": "Example view created via REST catalog",
"owner": "admin",
Expand Down Expand Up @@ -2384,12 +2385,13 @@ func (r *RestCatalogSuite) TestCreateView409() {
ns := "ns"
viewName := "view"
identifier := table.Identifier{ns, viewName}
schema := iceberg.NewSchema(1, iceberg.NestedField{
schema, err := iceberg.NewSchema(1, iceberg.NestedField{
ID: 1,
Name: "id",
Type: iceberg.PrimitiveTypes.Int32,
Required: true,
})
r.Require().NoError(err)
sql := "SELECT * FROM table"
reprs := []view.Representation{view.NewRepresentation(sql, "default")}
version, err := view.NewVersion(1, 1, reprs, table.Identifier{ns})
Expand Down Expand Up @@ -2418,12 +2420,13 @@ func (r *RestCatalogSuite) TestCreateView404() {
ns := "ns"
viewName := "view"
identifier := table.Identifier{ns, viewName}
schema := iceberg.NewSchema(1, iceberg.NestedField{
schema, err := iceberg.NewSchema(1, iceberg.NestedField{
ID: 1,
Name: "id",
Type: iceberg.PrimitiveTypes.Int32,
Required: true,
})
r.Require().NoError(err)
sql := "SELECT * FROM table"
reprs := []view.Representation{
view.NewRepresentation(sql, "spark"),
Expand Down
4 changes: 2 additions & 2 deletions catalog/sql/sql_integration_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ const (
location = "s3://warehouse/iceberg"
)

var tableSchemaSimple = iceberg.NewSchemaWithIdentifiers(1, []int{2},
var tableSchemaSimple = iceberg.MustNewSchemaWithIdentifiers(1, []int{2},
iceberg.NestedField{ID: 1, Name: "foo", Type: iceberg.StringType{}, Required: false},
iceberg.NestedField{ID: 2, Name: "bar", Type: iceberg.PrimitiveTypes.Int32, Required: true},
iceberg.NestedField{ID: 3, Name: "baz", Type: iceberg.PrimitiveTypes.Bool, Required: false},
Expand Down Expand Up @@ -92,7 +92,7 @@ func (s *SQLIntegrationSuite) TearDownTest() {
}
}

var tableSchemaNestedTest = iceberg.NewSchemaWithIdentifiers(1,
var tableSchemaNestedTest = iceberg.MustNewSchemaWithIdentifiers(1,
[]int{1},
iceberg.NestedField{
ID: 1, Name: "foo", Type: iceberg.PrimitiveTypes.String, Required: true,
Expand Down
22 changes: 14 additions & 8 deletions catalog/sql/sql_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ import (
"github.com/uptrace/bun/driver/sqliteshim"
)

var tableSchemaNested = iceberg.NewSchemaWithIdentifiers(1,
var tableSchemaNested = iceberg.MustNewSchemaWithIdentifiers(1,
[]int{1},
iceberg.NestedField{
ID: 1, Name: "foo", Type: iceberg.PrimitiveTypes.String, Required: false,
Expand Down Expand Up @@ -1058,9 +1058,10 @@ func (s *SqliteCatalogTestSuite) TestCreateView() {
s.Require().NoError(db.CreateNamespace(context.Background(), []string{nsName}, nil))

viewSQL := "SELECT * FROM test_table"
schema := iceberg.NewSchema(1, iceberg.NestedField{
schema, err := iceberg.NewSchema(1, iceberg.NestedField{
ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true,
})
s.Require().NoError(err)
s.Require().NoError(db.CreateView(context.Background(), []string{nsName, viewName}, schema, viewSQL, nil))

exists, err := db.CheckViewExists(context.Background(), []string{nsName, viewName})
Expand All @@ -1077,9 +1078,10 @@ func (s *SqliteCatalogTestSuite) TestDropView() {
s.Require().NoError(db.CreateNamespace(context.Background(), []string{nsName}, nil))

viewSQL := "SELECT * FROM test_table"
schema := iceberg.NewSchema(1, iceberg.NestedField{
schema, err := iceberg.NewSchema(1, iceberg.NestedField{
ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true,
})
s.Require().NoError(err)
s.Require().NoError(db.CreateView(context.Background(), []string{nsName, viewName}, schema, viewSQL, nil))

exists, err := db.CheckViewExists(context.Background(), []string{nsName, viewName})
Expand All @@ -1106,17 +1108,18 @@ func (s *SqliteCatalogTestSuite) TestDropViewWithInvalidMetadataLocation() {
s.Require().NoError(db.CreateNamespace(context.Background(), []string{nsName}, nil))

viewSQL := "SELECT * FROM test_table"
schema := iceberg.NewSchema(1, iceberg.NestedField{
schema, err := iceberg.NewSchema(1, iceberg.NestedField{
ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true,
})
s.Require().NoError(err)
s.Require().NoError(db.CreateView(context.Background(), []string{nsName, viewName}, schema, viewSQL, nil))

// Manually update the metadata location to a URL with an unsupported scheme
// This will cause io.LoadFS to fail with "IO for file '...' not implemented"
sqldb := s.getDB()
defer sqldb.Close()

_, err := sqldb.Exec(
_, err = sqldb.Exec(
"UPDATE iceberg_tables SET metadata_location = ? WHERE table_namespace = ? AND table_name = ?",
"unsupported-scheme://bucket/metadata.json",
nsName,
Expand All @@ -1142,9 +1145,10 @@ func (s *SqliteCatalogTestSuite) TestCheckViewExists() {
s.False(exists)

viewSQL := "SELECT * FROM test_table"
schema := iceberg.NewSchema(1, iceberg.NestedField{
schema, err := iceberg.NewSchema(1, iceberg.NestedField{
ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true,
})
s.Require().NoError(err)
s.Require().NoError(db.CreateView(context.Background(), []string{nsName, viewName}, schema, viewSQL, nil))

exists, err = db.CheckViewExists(context.Background(), []string{nsName, viewName})
Expand All @@ -1162,9 +1166,10 @@ func (s *SqliteCatalogTestSuite) TestListViews() {
viewNames := []string{tableName(), tableName(), tableName()}
for _, viewName := range viewNames {
viewSQL := "SELECT * FROM test_table"
schema := iceberg.NewSchema(1, iceberg.NestedField{
schema, err := iceberg.NewSchema(1, iceberg.NestedField{
ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true,
})
s.Require().NoError(err)
s.Require().NoError(db.CreateView(context.Background(), []string{nsName, viewName}, schema, viewSQL, nil))
}

Expand Down Expand Up @@ -1199,9 +1204,10 @@ func (s *SqliteCatalogTestSuite) TestLoadView() {
s.Require().NoError(db.CreateNamespace(context.Background(), []string{nsName}, nil))

viewSQL := "SELECT * FROM test_table"
schema := iceberg.NewSchema(1, iceberg.NestedField{
schema, err := iceberg.NewSchema(1, iceberg.NestedField{
ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true,
})
s.Require().NoError(err)
props := iceberg.Properties{
"comment": "Test view",
"owner": "test-user",
Expand Down
Loading
Loading