fix(glue): add glue.schema-columns property to skip StorageDescriptor columns

Introduces a new catalog property `glue.schema-columns` (default: `true`)
that controls whether Iceberg schema columns are written to the Glue
StorageDescriptor on table create and update operations.

When set to `false`, the StorageDescriptor is populated with location only,
avoiding the Glue API payload size limit that is hit for tables with very
large schemas (e.g. ~3600 fields).

Fixes #701

Author: laskoviymishka

catalog/glue/glue.go

@@ -65,6 +65,13 @@ const (
 	SkipArchive        = "glue.skip-archive"
 	SkipArchiveDefault = true
 
+	// GlueSchemaColumns controls whether schema columns are written to the Glue
+	// StorageDescriptor on table create/update. Set to "false" to omit columns,
+	// which avoids exceeding Glue's API payload limit for tables with very large
+	// schemas. Note: disabling columns breaks Athena column discovery via Glue.
+	GlueSchemaColumns        = "glue.schema-columns"
+	GlueSchemaColumnsDefault = true
+
 	AccessKeyID     = "glue.access-key-id"
 	SecretAccessKey = "glue.secret-access-key"
 	SessionToken    = "glue.session-token"
@@ -258,7 +265,7 @@ func (c *Catalog) CreateTable(ctx context.Context, identifier table.Identifier,
 	_, err = c.glueSvc.CreateTable(ctx, &glue.CreateTableInput{
 		CatalogId:    c.catalogId,
 		DatabaseName: aws.String(database),
-		TableInput:   constructTableInput(tableName, staged.Table, nil),
+		TableInput:   constructTableInput(tableName, staged.Table, nil, c.props.GetBool(GlueSchemaColumns, GlueSchemaColumnsDefault)),
 	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to create table %s.%s: %w", database, tableName, err)
@@ -290,7 +297,7 @@ func (c *Catalog) RegisterTable(ctx context.Context, identifier table.Identifier
 	_, err = c.glueSvc.CreateTable(ctx, &glue.CreateTableInput{
 		CatalogId:    c.catalogId,
 		DatabaseName: aws.String(database),
-		TableInput:   constructTableInput(tableName, tbl, nil),
+		TableInput:   constructTableInput(tableName, tbl, nil, c.props.GetBool(GlueSchemaColumns, GlueSchemaColumnsDefault)),
 	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to register table %s.%s: %w", database, tableName, err)
@@ -338,7 +345,7 @@ func (c *Catalog) CommitTable(ctx context.Context, identifier table.Identifier,
 		_, err = c.glueSvc.UpdateTable(ctx, &glue.UpdateTableInput{
 			CatalogId:    c.catalogId,
 			DatabaseName: aws.String(database),
-			TableInput:   constructTableInput(tableName, staged.Table, currentGlueTable),
+			TableInput:   constructTableInput(tableName, staged.Table, currentGlueTable, c.props.GetBool(GlueSchemaColumns, GlueSchemaColumnsDefault)),
 			// use `VersionId` to implement optimistic locking
 			VersionId:   currentGlueTable.VersionId,
 			SkipArchive: aws.Bool(c.props.GetBool(SkipArchive, SkipArchiveDefault)),
@@ -350,7 +357,7 @@ func (c *Catalog) CommitTable(ctx context.Context, identifier table.Identifier,
 		_, err = c.glueSvc.CreateTable(ctx, &glue.CreateTableInput{
 			CatalogId:    c.catalogId,
 			DatabaseName: aws.String(database),
-			TableInput:   constructTableInput(tableName, staged.Table, nil),
+			TableInput:   constructTableInput(tableName, staged.Table, nil, c.props.GetBool(GlueSchemaColumns, GlueSchemaColumnsDefault)),
 		})
 		if err != nil {
 			return nil, "", err
@@ -761,15 +768,19 @@ func constructParameters(staged *table.Table, previousGlueTable *types.Table) ma
 	return parameters
 }
 
-func constructTableInput(tableName string, staged *table.Table, previousGlueTable *types.Table) *types.TableInput {
+func constructTableInput(tableName string, staged *table.Table, previousGlueTable *types.Table, includeColumns bool) *types.TableInput {
+	sd := &types.StorageDescriptor{
+		Location: aws.String(staged.Location()),
+	}
+	if includeColumns {
+		sd.Columns = schemasToGlueColumns(staged.Metadata())
+	}
+
 	tableInput := &types.TableInput{
-		Name:       aws.String(tableName),
-		TableType:  aws.String(glueTableType),
-		Parameters: constructParameters(staged, previousGlueTable),
-		StorageDescriptor: &types.StorageDescriptor{
-			Location: aws.String(staged.Location()),
-			Columns:  schemasToGlueColumns(staged.Metadata()),
-		},
+		Name:              aws.String(tableName),
+		TableType:         aws.String(glueTableType),
+		Parameters:        constructParameters(staged, previousGlueTable),
+		StorageDescriptor: sd,
 	}
 
 	if comment, ok := staged.Properties()[PropsKeyDescription]; ok {

catalog/glue/glue_test.go

@@ -1296,6 +1296,41 @@ func TestGlueCheckTableNotExists(t *testing.T) {
 	assert.False(exists)
 }
 
+func TestConstructTableInputSchemaColumns(t *testing.T) {
+	schema := iceberg.NewSchemaWithIdentifiers(1, []int{1},
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.Int64Type{}, Required: true},
+		iceberg.NestedField{ID: 2, Name: "data", Type: iceberg.StringType{}},
+	)
+
+	builder, err := table.NewMetadataBuilder(2)
+	require.NoError(t, err)
+	require.NoError(t, builder.SetLoc("s3://bucket/db/tbl"))
+	require.NoError(t, builder.AddSchema(schema))
+	require.NoError(t, builder.SetCurrentSchemaID(1))
+	require.NoError(t, builder.AddPartitionSpec(iceberg.UnpartitionedSpec, true))
+	require.NoError(t, builder.SetDefaultSpecID(iceberg.UnpartitionedSpec.ID()))
+	require.NoError(t, builder.AddSortOrder(&table.UnsortedSortOrder))
+	require.NoError(t, builder.SetDefaultSortOrderID(table.UnsortedSortOrderID))
+	meta, err := builder.Build()
+	require.NoError(t, err)
+
+	tbl := table.New([]string{"db", "tbl"}, meta, "s3://bucket/db/tbl/metadata/v1.json", nil, nil)
+
+	t.Run("columns included by default", func(t *testing.T) {
+		input := constructTableInput("tbl", tbl, nil, true)
+		require.NotNil(t, input.StorageDescriptor)
+		assert := require.New(t)
+		assert.Len(input.StorageDescriptor.Columns, 2)
+	})
+
+	t.Run("columns omitted when disabled", func(t *testing.T) {
+		input := constructTableInput("tbl", tbl, nil, false)
+		require.NotNil(t, input.StorageDescriptor)
+		assert := require.New(t)
+		assert.Empty(input.StorageDescriptor.Columns)
+	})
+}
+
 func cleanupTable(t *testing.T, ctlg catalog.Catalog, tbIdent table.Identifier, awsCfg aws.Config) {
 	t.Helper()