master(dm): support extra SHOW FULL TABLES columns from PolarDB-X (pingcap#12481)

close pingcap#12569

Author: GMHDBJD

dm/checker/check_test.go

@@ -863,7 +863,10 @@ func initMockDB(t *testing.T) sqlmock.Sqlmock {
 	mock, err := conn.MockDefaultDBProvider()
 	require.NoError(t, err)
 	mock.ExpectQuery("SHOW DATABASES").WillReturnRows(sqlmock.NewRows([]string{"DATABASE"}).AddRow(schema))
-	mock.ExpectQuery("SHOW FULL TABLES").WillReturnRows(sqlmock.NewRows([]string{"Tables_in_" + schema, "Table_type"}).AddRow(tb1, "BASE TABLE").AddRow(tb2, "BASE TABLE"))
+	mock.ExpectQuery("SHOW FULL TABLES").WillReturnRows(
+		sqlmock.NewRows([]string{"Tables_in_" + schema, "Table_type", "Auto_partition", "Table_group"}).
+			AddRow(tb1, "BASE TABLE", "NO", "single_tg").
+			AddRow(tb2, "BASE TABLE", "NO", "single_tg"))
 	return mock
 }
 

dm/master/openapi_view.go

@@ -391,7 +391,7 @@ func (s *Server) DMAPIGetSourceTableList(c *gin.Context, sourceName string, sche
 		return
 	}
 	defer baseDB.Close()
-	tableList, err := dbutil.GetTables(c.Request.Context(), baseDB.DB, schemaName)
+	tableList, err := conn.GetTables(c.Request.Context(), baseDB.DB, schemaName)
 	if err != nil {
 		_ = c.Error(err)
 		return

dm/master/openapi_view_test.go

@@ -706,7 +706,8 @@ func (s *OpenAPIViewSuite) TestSourceAPI() {
 	s.NoError(err)
 	tableName := "CHARACTER_SETS"
 	mockDB.ExpectQuery("SHOW FULL TABLES IN `information_schema` WHERE Table_Type != 'VIEW';").WillReturnRows(
-		sqlmock.NewRows([]string{"Tables_in_information_schema", "Table_type"}).AddRow(tableName, "BASE TABLE"))
+		sqlmock.NewRows([]string{"Tables_in_information_schema", "Table_type", "Auto_partition", "Table_group"}).
+			AddRow(tableName, "BASE TABLE", "NO", "single_tg"))
 	tableURL := fmt.Sprintf("%s/%s/schemas/%s", baseURL, source1.SourceName, schemaName)
 	result = testutil.NewRequest().Get(tableURL).GoWithHTTPHandler(s.T(), s1.openapiHandles)
 	s.Equal(http.StatusOK, result.Code())

dm/pkg/checker/onlineddl.go

@@ -17,8 +17,8 @@ import (
 	"context"
 	"database/sql"
 
-	"github.com/pingcap/tidb/pkg/util/dbutil"
 	"github.com/pingcap/tidb/pkg/util/filter"
+	"github.com/pingcap/tiflow/dm/pkg/conn"
 	onlineddl "github.com/pingcap/tiflow/dm/syncer/online-ddl-tools"
 )
 
@@ -47,7 +47,7 @@ func (c *OnlineDDLChecker) Check(ctx context.Context) *Result {
 	}
 
 	for schema := range c.checkSchemas {
-		tableList, err := dbutil.GetTables(ctx, c.db, schema)
+		tableList, err := conn.GetTables(ctx, c.db, schema)
 		if err != nil {
 			markCheckError(r, err)
 			return r

dm/pkg/conn/db.go

@@ -483,7 +483,7 @@ func FetchAllDoTables(ctx context.Context, db *BaseDB, bw *filter.Filter) (map[s
 	schemaToTables := make(map[string][]string)
 	for _, ftSchema := range ftSchemas {
 		schema := ftSchema.Schema
-		tables, err := dbutil.GetTables(ctx, db.DB, schema)
+		tables, err := GetTables(ctx, db.DB, schema)
 		if err != nil {
 			return nil, terror.DBErrorAdapt(err, db.Scope, terror.ErrDBDriverError)
 		}

dm/pkg/conn/db_test.go

@@ -80,6 +80,55 @@ func TestGetRandomServerID(t *testing.T) {
 	require.NotEqual(t, 101, serverID)
 }
 
+func TestGetTables(t *testing.T) {
+	t.Parallel()
+
+	db, mock, err := sqlmock.New()
+	require.NoError(t, err)
+
+	schema := "test_db"
+	tables := []string{"tbl1", "tbl2"}
+
+	rows := sqlmock.NewRows([]string{fmt.Sprintf("Tables_in_%s", schema), "Table_type"})
+	addRowsForTables(rows, tables)
+	mock.ExpectQuery(fmt.Sprintf("SHOW FULL TABLES IN `%s` WHERE Table_Type != 'VIEW'", schema)).WillReturnRows(rows)
+
+	got, err := GetTables(context.Background(), db, schema)
+	require.NoError(t, err)
+	require.Equal(t, tables, got)
+	require.NoError(t, mock.ExpectationsWereMet())
+
+	rows = sqlmock.NewRows([]string{fmt.Sprintf("Tables_in_%s", schema), "Table_type", "Auto_partition", "Table_group"})
+	addRowsForPolarDBXTables(rows, tables)
+	mock.ExpectQuery(fmt.Sprintf("SHOW FULL TABLES IN `%s` WHERE Table_Type != 'VIEW'", schema)).WillReturnRows(rows)
+
+	got, err = GetTables(context.Background(), db, schema)
+	require.NoError(t, err)
+	require.Equal(t, tables, got)
+	require.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestGetTablesErrors(t *testing.T) {
+	t.Parallel()
+
+	db, mock, err := sqlmock.New()
+	require.NoError(t, err)
+
+	schema := "test_db"
+	query := fmt.Sprintf("SHOW FULL TABLES IN `%s` WHERE Table_Type != 'VIEW'", schema)
+
+	mock.ExpectQuery(query).WillReturnError(errors.New("query failed"))
+	_, err = GetTables(context.Background(), db, schema)
+	require.ErrorContains(t, err, "query failed")
+	require.NoError(t, mock.ExpectationsWereMet())
+
+	rows := sqlmock.NewRows([]string{fmt.Sprintf("Tables_in_%s", schema)}).AddRow("tbl1")
+	mock.ExpectQuery(query).WillReturnRows(rows)
+	_, err = GetTables(context.Background(), db, schema)
+	require.ErrorContains(t, err, "unexpected SHOW FULL TABLES result column count 1")
+	require.NoError(t, mock.ExpectationsWereMet())
+}
+
 func TestGetMariaDBGtidDomainID(t *testing.T) {
 	t.Parallel()
 
@@ -571,6 +620,18 @@ func TestFetchAllDoTables(t *testing.T) {
 	require.Len(t, got, 1)
 	require.Equal(t, []string{"tbl1", "tbl2"}, got[doSchema])
 	require.NoError(t, mock.ExpectationsWereMet())
+
+	rows = sqlmock.NewRows([]string{"Database"})
+	addRowsForSchemas(rows, schemas)
+	mock.ExpectQuery(`SHOW DATABASES`).WillReturnRows(rows)
+	rows = sqlmock.NewRows([]string{fmt.Sprintf("Tables_in_%s", doSchema), "Table_type", "Auto_partition", "Table_group"})
+	addRowsForPolarDBXTables(rows, tables)
+	mock.ExpectQuery(fmt.Sprintf("SHOW FULL TABLES IN `%s` WHERE Table_Type != 'VIEW'", doSchema)).WillReturnRows(rows)
+	got, err = FetchAllDoTables(context.Background(), NewBaseDBForTest(db), ba)
+	require.NoError(t, err)
+	require.Len(t, got, 1)
+	require.Equal(t, []string{"tbl1", "tbl2"}, got[doSchema])
+	require.NoError(t, mock.ExpectationsWereMet())
 }
 
 func TestFetchTargetDoTables(t *testing.T) {
@@ -647,6 +708,12 @@ func addRowsForTables(rows *sqlmock.Rows, tables []string) {
 	}
 }
 
+func addRowsForPolarDBXTables(rows *sqlmock.Rows, tables []string) {
+	for _, table := range tables {
+		rows.AddRow(table, "BASE TABLE", "NO", "single_tg")
+	}
+}
+
 func TestGetGTIDMode(t *testing.T) {
 	t.Parallel()
 

dm/pkg/conn/table.go

@@ -0,0 +1,74 @@
+// 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 conn
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"strings"
+
+	"github.com/pingcap/errors"
+	"github.com/pingcap/tidb/pkg/util/dbutil"
+)
+
+// GetTables returns names of all non-view tables in the specified schema.
+// It is compatible with upstreams like PolarDB-X, whose SHOW FULL TABLES
+// may return extra columns after the standard table name and table type.
+func GetTables(ctx context.Context, db dbutil.QueryExecutor, schemaName string) ([]string, error) {
+	query := fmt.Sprintf("SHOW FULL TABLES IN `%s` WHERE Table_Type != 'VIEW';", escapeName(schemaName))
+	return queryTables(ctx, db, query)
+}
+
+func queryTables(ctx context.Context, db dbutil.QueryExecutor, query string) ([]string, error) {
+	rows, err := db.QueryContext(ctx, query)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	defer rows.Close()
+
+	columns, err := rows.Columns()
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	if len(columns) < 2 {
+		return nil, errors.Errorf("unexpected SHOW FULL TABLES result column count %d, expected at least 2", len(columns))
+	}
+
+	tables := make([]string, 0, 8)
+	for rows.Next() {
+		var table, tableType sql.NullString
+		values := make([]any, len(columns))
+		values[0] = &table
+		values[1] = &tableType
+		for i := 2; i < len(values); i++ {
+			values[i] = new(sql.RawBytes)
+		}
+
+		if err := rows.Scan(values...); err != nil {
+			return nil, errors.Trace(err)
+		}
+		if !table.Valid || !tableType.Valid {
+			continue
+		}
+
+		tables = append(tables, table.String)
+	}
+
+	return tables, errors.Trace(rows.Err())
+}
+
+func escapeName(name string) string {
+	return strings.ReplaceAll(name, "`", "``")
+}

sync_diff_inspector/source/mysql_shard.go

@@ -27,6 +27,7 @@ import (
 	"github.com/pingcap/tidb/pkg/util/dbutil"
 	"github.com/pingcap/tidb/pkg/util/filter"
 	tableFilter "github.com/pingcap/tidb/pkg/util/table-filter"
+	"github.com/pingcap/tiflow/dm/pkg/conn"
 	"github.com/pingcap/tiflow/sync_diff_inspector/config"
 	"github.com/pingcap/tiflow/sync_diff_inspector/source/common"
 	"github.com/pingcap/tiflow/sync_diff_inspector/splitter"
@@ -351,7 +352,7 @@ func NewMySQLSources(ctx context.Context, tableDiffs []*common.TableDiff, ds []*
 			if filter.IsSystemSchema(strings.ToLower(schema)) {
 				continue
 			}
-			allTables, err := dbutil.GetTables(ctx, sourceDB.Conn, schema)
+			allTables, err := conn.GetTables(ctx, sourceDB.Conn, schema)
 			if err != nil {
 				return nil, errors.Annotatef(err, "get tables from %d source %s", i, schema)
 			}

sync_diff_inspector/source/source.go

@@ -28,6 +28,7 @@ import (
 	"github.com/pingcap/tidb/pkg/util/filter"
 	tableFilter "github.com/pingcap/tidb/pkg/util/table-filter"
 	router "github.com/pingcap/tidb/pkg/util/table-router"
+	"github.com/pingcap/tiflow/dm/pkg/conn"
 	"github.com/pingcap/tiflow/sync_diff_inspector/config"
 	"github.com/pingcap/tiflow/sync_diff_inspector/source/common"
 	"github.com/pingcap/tiflow/sync_diff_inspector/splitter"
@@ -327,7 +328,7 @@ func initTables(ctx context.Context, cfg *config.Config) (cfgTables []*config.Ta
 		if filter.IsSystemSchema(strings.ToLower(schema)) {
 			continue
 		}
-		allTables, err := dbutil.GetTables(ctx, downStreamConn, schema)
+		allTables, err := conn.GetTables(ctx, downStreamConn, schema)
 		if err != nil {
 			return nil, errors.Annotatef(err, "get tables from target source %s", schema)
 		}