sink,storage-consumer: fix iceberg storage replay

Signed-off-by: tenfyzhong <tenfy@tenfy.cn>

Author: tenfyzhong

cmd/storage-consumer/consumer.go

@@ -20,6 +20,7 @@ import (
 	"net/url"
 	"sort"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/pingcap/errors"
@@ -421,6 +422,60 @@ func (c *consumer) flushDMLEvents(ctx context.Context, tableID int64) error {
 	}
 }
 
+func (c *consumer) flushIcebergDMLEvents(ctx context.Context, tableID int64) error {
+	group := c.eventsGroup[tableID]
+	if group == nil {
+		return nil
+	}
+	events := group.GetAllEvents()
+	total := len(events)
+	if total == 0 {
+		return nil
+	}
+	var (
+		schema string
+		table  string
+	)
+	if events[0].TableInfo != nil {
+		schema = events[0].TableInfo.GetSchemaName()
+		table = events[0].TableInfo.GetTableName()
+	}
+
+	start := time.Now()
+	for i, e := range events {
+		done := make(chan struct{})
+		var once sync.Once
+		e.AddPostFlushFunc(func() {
+			once.Do(func() {
+				close(done)
+			})
+		})
+		c.sink.AddDMLEvent(e)
+
+		ticker := time.NewTicker(defaultLogInterval)
+		for {
+			select {
+			case <-ctx.Done():
+				ticker.Stop()
+				return context.Cause(ctx)
+			case <-done:
+				ticker.Stop()
+				goto NEXT
+			case <-ticker.C:
+				log.Warn("iceberg DML event cannot be flushed in time",
+					zap.Int("total", total), zap.Int("flushed", i),
+					zap.Uint64("commitTs", e.CommitTs), zap.String("schema", schema), zap.String("table", table))
+			}
+		}
+	NEXT:
+	}
+
+	log.Info("flush iceberg DML events done",
+		zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
+		zap.Int("total", total), zap.Duration("duration", time.Since(start)))
+	return nil
+}
+
 func (c *consumer) parseDMLFilePath(ctx context.Context, path string) error {
 	var dmlkey cloudstorage.DmlPathKey
 	dispatcherID, err := dmlkey.ParseIndexFilePath(

cmd/storage-consumer/iceberg.go

@@ -64,7 +64,7 @@ func buildIcebergDDLEvents(
 				fmt.Sprintf(
 					"CREATE TABLE IF NOT EXISTS %s (%s)",
 					commonType.QuoteSchema(curr.SchemaName, curr.TableName),
-					strings.Join(icebergColumnSQLDefs(curr.Columns), ","),
+					strings.Join(icebergCreateTableSQLDefs(curr.Columns), ","),
 				),
 				tableInfo,
 				model.ActionCreateTable,
@@ -243,6 +243,32 @@ func icebergColumnSQLDefs(columns []sinkiceberg.Column) []string {
 	return defs
 }
 
+func icebergCreateTableSQLDefs(columns []sinkiceberg.Column) []string {
+	defs := icebergColumnSQLDefs(columns)
+	if primaryKey := icebergPrimaryKeySQLDef(columns); primaryKey != "" {
+		defs = append(defs, primaryKey)
+	}
+	return defs
+}
+
+func icebergPrimaryKeySQLDef(columns []sinkiceberg.Column) string {
+	primaryKeyColumns := make([]string, 0, len(columns))
+	for _, col := range columns {
+		if !icebergColumnIsPrimaryKey(col) {
+			continue
+		}
+		primaryKeyColumns = append(primaryKeyColumns, commonType.QuoteName(col.Name))
+	}
+	if len(primaryKeyColumns) == 0 {
+		return ""
+	}
+	return fmt.Sprintf("PRIMARY KEY (%s)", strings.Join(primaryKeyColumns, ","))
+}
+
+func icebergColumnIsPrimaryKey(col sinkiceberg.Column) bool {
+	return col.OriginalTableCol != nil && strings.EqualFold(strings.TrimSpace(col.OriginalTableCol.IsPK), "true")
+}
+
 func icebergColumnSQLDef(col sinkiceberg.Column) string {
 	return fmt.Sprintf("%s %s %s",
 		commonType.QuoteName(col.Name),
@@ -259,6 +285,10 @@ func icebergColumnNullability(required bool) string {
 }
 
 func icebergColumnSQLType(col sinkiceberg.Column) string {
+	if sqlType, ok := originalIcebergColumnSQLType(col); ok {
+		return sqlType
+	}
+
 	switch strings.TrimSpace(col.Type) {
 	case "int":
 		return "INT"
@@ -283,6 +313,10 @@ func icebergColumnSQLType(col sinkiceberg.Column) string {
 }
 
 func icebergColumnToTableCol(col sinkiceberg.Column) (cloudstorage.TableCol, error) {
+	if original := cloneOriginalIcebergTableCol(col); original != nil {
+		return *original, nil
+	}
+
 	result := cloudstorage.TableCol{
 		Name: col.Name,
 	}
@@ -318,6 +352,65 @@ func icebergColumnToTableCol(col sinkiceberg.Column) (cloudstorage.TableCol, err
 	return result, nil
 }
 
+func originalIcebergColumnSQLType(col sinkiceberg.Column) (string, bool) {
+	if col.OriginalTableCol == nil {
+		return "", false
+	}
+
+	switch strings.ToUpper(strings.TrimSpace(col.OriginalTableCol.Tp)) {
+	case "CHAR", "VARCHAR", "VAR_STRING",
+		"BINARY", "VARBINARY",
+		"TINYTEXT", "TEXT", "MEDIUMTEXT", "LONGTEXT",
+		"TINYBLOB", "BLOB", "MEDIUMBLOB", "LONGBLOB",
+		"ENUM", "SET":
+		return formatTableColSQLType(*cloneOriginalIcebergTableCol(col)), true
+	default:
+		return "", false
+	}
+}
+
+func cloneOriginalIcebergTableCol(col sinkiceberg.Column) *cloudstorage.TableCol {
+	if col.OriginalTableCol == nil {
+		return nil
+	}
+
+	cloned := *col.OriginalTableCol
+	cloned.Name = col.Name
+	if col.Required {
+		cloned.Nullable = "false"
+	} else {
+		cloned.Nullable = ""
+	}
+	return &cloned
+}
+
+func formatTableColSQLType(col cloudstorage.TableCol) string {
+	tp := strings.ToUpper(strings.TrimSpace(col.Tp))
+	baseType := strings.TrimSuffix(tp, " UNSIGNED")
+
+	switch baseType {
+	case "CHAR", "VARCHAR", "VAR_STRING", "BINARY", "VARBINARY", "BIT":
+		if col.Precision != "" {
+			return fmt.Sprintf("%s(%s)", tp, col.Precision)
+		}
+	case "ENUM", "SET":
+		if len(col.Elems) > 0 {
+			return fmt.Sprintf("%s(%s)", tp, quoteSQLStringList(col.Elems))
+		}
+	}
+	return tp
+}
+
+func quoteSQLStringList(values []string) string {
+	quoted := make([]string, 0, len(values))
+	for _, value := range values {
+		escaped := strings.ReplaceAll(value, "\\", "\\\\")
+		escaped = strings.ReplaceAll(escaped, "'", "''")
+		quoted = append(quoted, fmt.Sprintf("'%s'", escaped))
+	}
+	return strings.Join(quoted, ",")
+}
+
 func parseIcebergDecimalType(raw string) (int, int, bool) {
 	s := strings.TrimSpace(raw)
 	if !strings.HasPrefix(s, "decimal(") || !strings.HasSuffix(s, ")") {
@@ -514,7 +607,7 @@ func (c *consumer) handleIceberg(ctx context.Context, round uint64) error {
 				}
 			}
 
-			if err := c.flushDMLEvents(ctx, tableID); err != nil {
+			if err := c.flushIcebergDMLEvents(ctx, tableID); err != nil {
 				return err
 			}
 			state.lastMetadataVersion = metadataVersion

cmd/storage-consumer/iceberg_integration_test.go

@@ -0,0 +1,101 @@
+// Copyright 2026 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"context"
+	"testing"
+
+	"github.com/pingcap/ticdc/pkg/common"
+	sinkiceberg "github.com/pingcap/ticdc/pkg/sink/iceberg"
+	"github.com/pingcap/ticdc/pkg/util"
+	"github.com/pingcap/tidb/pkg/meta/model"
+	"github.com/pingcap/tidb/pkg/parser/ast"
+	"github.com/pingcap/tidb/pkg/parser/charset"
+	"github.com/pingcap/tidb/pkg/parser/mysql"
+	"github.com/pingcap/tidb/pkg/types"
+	"github.com/stretchr/testify/require"
+)
+
+func TestBuildIcebergDDLEventsPreservesOriginalStringTypes(t *testing.T) {
+	ctx := context.Background()
+
+	tmpDir := t.TempDir()
+	extStorage, warehouseURL, err := util.GetTestExtStorage(ctx, tmpDir)
+	require.NoError(t, err)
+	defer extStorage.Close()
+
+	cfg := sinkiceberg.NewConfig()
+	cfg.WarehouseURI = warehouseURL.String()
+	cfg.WarehouseLocation = warehouseURL.String()
+	cfg.Namespace = "ns"
+
+	tableWriter := sinkiceberg.NewTableWriter(cfg, extStorage)
+	changefeedID := common.NewChangefeedID4Test("default", "cf")
+
+	textType := types.NewFieldType(mysql.TypeBlob)
+	textType.SetCharset(charset.CharsetUTF8MB4)
+
+	varcharType := types.NewFieldType(mysql.TypeVarchar)
+	varcharType.SetCharset(charset.CharsetUTF8MB4)
+	varcharType.SetFlen(32)
+
+	idType := types.NewFieldType(mysql.TypeLonglong)
+	idType.AddFlag(mysql.PriKeyFlag | mysql.NotNullFlag)
+
+	tableInfoV1 := common.WrapTableInfo("test", &model.TableInfo{
+		ID:         20,
+		Name:       ast.NewCIStr("t_iceberg"),
+		PKIsHandle: true,
+		Columns: []*model.ColumnInfo{
+			{ID: 1, Name: ast.NewCIStr("id"), FieldType: *idType},
+			{ID: 2, Name: ast.NewCIStr("v"), FieldType: *textType},
+			{ID: 3, Name: ast.NewCIStr("score"), FieldType: *types.NewFieldType(mysql.TypeLong)},
+		},
+	})
+	require.NoError(t, tableWriter.EnsureTable(ctx, changefeedID, tableInfoV1))
+
+	tableInfoV2 := common.WrapTableInfo("test", &model.TableInfo{
+		ID:         20,
+		Name:       ast.NewCIStr("t_iceberg"),
+		PKIsHandle: true,
+		Columns: []*model.ColumnInfo{
+			{ID: 1, Name: ast.NewCIStr("id"), FieldType: *idType},
+			{ID: 2, Name: ast.NewCIStr("v"), FieldType: *textType},
+			{ID: 3, Name: ast.NewCIStr("score"), FieldType: *types.NewFieldType(mysql.TypeLong)},
+			{ID: 4, Name: ast.NewCIStr("extra"), FieldType: *varcharType},
+		},
+	})
+	require.NoError(t, tableWriter.EnsureTable(ctx, changefeedID, tableInfoV2))
+
+	version1, err := sinkiceberg.LoadTableVersion(ctx, cfg, extStorage, "test", "t_iceberg", 1)
+	require.NoError(t, err)
+	version2, err := sinkiceberg.LoadTableVersion(ctx, cfg, extStorage, "test", "t_iceberg", 2)
+	require.NoError(t, err)
+
+	createEvents, err := buildIcebergDDLEvents(nil, version1)
+	require.NoError(t, err)
+	require.Len(t, createEvents, 2)
+	require.Contains(t, createEvents[1].Query, "`v` TEXT NULL")
+	require.NotContains(t, createEvents[1].Query, "`v` LONGBLOB NULL")
+	require.Contains(t, createEvents[1].Query, "PRIMARY KEY (`id`)")
+
+	alterEvents, err := buildIcebergDDLEvents(version1, version2)
+	require.NoError(t, err)
+	require.Len(t, alterEvents, 1)
+	require.Equal(t,
+		"ALTER TABLE `test`.`t_iceberg` ADD COLUMN `extra` VARCHAR(32) NULL",
+		alterEvents[0].Query,
+	)
+}