From d5c2aea296c11ed218ad22f906a17cd27c83265f Mon Sep 17 00:00:00 2001
From: Robin Tang <rtang.cs@gmail.com>
Date: Wed, 18 Dec 2024 10:59:30 -0800
Subject: [PATCH] Clean up.

---
 sources/mysql/streaming/ddl/table_adapter.go | 34 +++++++++++---------
 1 file changed, 19 insertions(+), 15 deletions(-)

diff --git a/sources/mysql/streaming/ddl/table_adapter.go b/sources/mysql/streaming/ddl/table_adapter.go
index 2a25a4a2..f0795379 100644
--- a/sources/mysql/streaming/ddl/table_adapter.go
+++ b/sources/mysql/streaming/ddl/table_adapter.go
@@ -68,20 +68,8 @@ func (t TableAdapter) buildFieldConverters() ([]transformer.FieldConverter, erro
 		return nil, nil
 	}
 
-	// Exclude columns (if any) from the table metadata
-	cols, err := column.FilterOutExcludedColumns(t.GetParsedColumns(), t.tableCfg.ExcludeColumns, t.PartitionKeys())
-	if err != nil {
-		return nil, err
-	}
-
-	// Include columns (if any) from the table metadata
-	cols, err = column.FilterForIncludedColumns(cols, t.tableCfg.IncludeColumns, t.PartitionKeys())
-	if err != nil {
-		return nil, err
-	}
-
-	fieldConverters := make([]transformer.FieldConverter, len(cols))
-	for i, col := range cols {
+	fieldConverters := make([]transformer.FieldConverter, len(t.parsedColumns))
+	for i, col := range t.parsedColumns {
 		converter, err := converters.ValueConverterForType(col.Type, col.Opts)
 		if err != nil {
 			return nil, fmt.Errorf("failed to build value converter for column %q: %w", col.Name, err)
@@ -107,7 +95,23 @@ func (t TableAdapter) buildParsedColumns() ([]schema.Column, error) {
 		})
 	}
 
-	return parsedColumns, nil
+	if t.tableCfg == nil {
+		return parsedColumns, nil
+	}
+
+	// Exclude columns (if any) from the table metadata
+	cols, err := column.FilterOutExcludedColumns(parsedColumns, t.tableCfg.ExcludeColumns, t.PartitionKeys())
+	if err != nil {
+		return nil, err
+	}
+
+	// Include columns (if any) from the table metadata
+	cols, err = column.FilterForIncludedColumns(cols, t.tableCfg.IncludeColumns, t.PartitionKeys())
+	if err != nil {
+		return nil, err
+	}
+
+	return cols, nil
 }
 
 func (t TableAdapter) TopicSuffix() string {