fix: Prevent data duplication for replicated tables by enforcing CopyOnMaster strategy

robertmu · robertmu · commit bf4e5a8f8f4a · 2025-08-19T10:04:14.000+08:00
- Always use CopyOnMaster strategy for replicated tables regardless of on-segment-threshold - Add IsReplicated field detection via gp_distribution_policy.policytype = 'r' - Ensure replicated table identification propagates through entire processing pipeline - Add version-specific SQL queries for GPDB 6+ compatibility - Include comprehensive tests for replicated table migration scenarios Fixes #21: Data duplicated when on-segment-threshold is negative and replicated tables incorrectly use CopyOnSegment strategy instead of CopyOnMaster.
diff --git a/copy/copy_command.go b/copy/copy_command.go
@@ -607,13 +607,23 @@ func createTestCopyStrategy(strategy string, workerId int, srcSegs []utils.Segme
 // - Number of segments in source and destination clusters (srcSegs, destSegs)
 // - Database versions of source and destination (srcConn.Version, destConn.Version)
 // It returns an instance of a struct that implements the CopyCommand interface.
-func CreateCopyStrategy(numTuples int64, workerId int, srcSegs []utils.SegmentHostInfo, destSegs []utils.SegmentIpInfo, srcConn, destConn *dbconn.DBConn) CopyCommand {
+func CreateCopyStrategy(isReplicated bool,
+	numTuples int64,
+	workerId int,
+	srcSegs []utils.SegmentHostInfo,
+	destSegs []utils.SegmentIpInfo,
+	srcConn, destConn *dbconn.DBConn) CopyCommand {
+	compArg := "--compress-type snappy"
+	if isReplicated {
+		gplog.Debug("Using CopyOnMaster strategy for replicated table")
+		return &CopyOnMaster{CopyBase: CopyBase{WorkerId: workerId, SrcSegmentsHostInfo: srcSegs, DestSegmentsIpInfo: destSegs, CompArg: compArg}}
+	}
+
 	if strategy := os.Getenv("TEST_COPY_STRATEGY"); strategy != "" {
 		gplog.Debug("Using test copy strategy: %s", strategy)
 		return createTestCopyStrategy(strategy, workerId, srcSegs, destSegs)
 	}
 
-	compArg := "--compress-type snappy"
 	if numTuples <= int64(utils.MustGetFlagInt(option.ON_SEGMENT_THRESHOLD)) {
 		if !utils.MustGetFlagBool(option.COMPRESSION) {
 			compArg = "--no-compression"
diff --git a/copy/copy_manager.go b/copy/copy_manager.go
@@ -182,7 +182,8 @@ func (tc *TableCopier) cleanupAfterCopy(isSkipped bool, inTxn bool, err error) {
 }
 
 func (tc *TableCopier) copyData() error {
-	command := CreateCopyStrategy(tc.srcTable.RelTuples,
+	command := CreateCopyStrategy(tc.srcTable.IsReplicated,
+		tc.srcTable.RelTuples,
 		tc.workerID,
 		tc.manager.srcSegmentsHostInfo,
 		tc.manager.destSegmentsIpInfo,
diff --git a/copy/copy_metadata.go b/copy/copy_metadata.go
@@ -116,9 +116,10 @@ func (m *MetadataManager) fillTablePairChan(srcTables, destTables []option.Table
 	for i, t := range srcTables {
 		tablec <- option.TablePair{
 			SrcTable: option.Table{
-				Schema:    t.Schema,
-				Name:      t.Name,
-				RelTuples: t.RelTuples,
+				Schema:       t.Schema,
+				Name:         t.Name,
+				RelTuples:    t.RelTuples,
+				IsReplicated: t.IsReplicated,
 			},
 			DestTable: option.Table{
 				Schema: destTables[i].Schema,
diff --git a/copy/copy_operation.go b/copy/copy_operation.go
@@ -149,7 +149,6 @@ func (op *CopyOperation) handleFailure(donec chan struct{}, toErr error, fromErr
 	gplog.Debug("[Worker %v] Failed to copy %v.%v to %v.%v : %v",
 		op.connNum, op.srcTable.Schema, op.srcTable.Name,
 		op.destTable.Schema, op.destTable.Name, toErr)
-	time.Sleep(2 * time.Second)
 	if fromErr != nil {
 		<-donec
 		return fromErr
diff --git a/copy/copy_query.go b/copy/copy_query.go
@@ -62,14 +62,18 @@ func (qm *QueryManager) GetAllDatabases(conn *dbconn.DBConn) ([]string, error) {
 
 // GetUserTables retrieves user tables from the database
 // example output:
-// map[public.t1:{"Partition":0,"RelTuples":1000} public.t2:{"Partition":0,"RelTuples":2000}]
+// map[public.t1:{"Partition":0,"RelTuples":1000,"IsReplicated":false} public.t2:{"Partition":0,"RelTuples":2000,"IsReplicated":true}]
 func (tm *QueryManager) GetUserTables(conn *dbconn.DBConn) (map[string]option.TableStatistics, error) {
 	var query string
 
 	if (conn.Version.IsGPDB() && conn.Version.AtLeast("7")) || conn.Version.IsCBDBFamily() {
 		query = `
 		SELECT
-			quote_ident(n.nspname) AS schema, quote_ident(c.relname) as name, 0 as partition, cast(c.reltuples as bigint) AS relTuples
+			quote_ident(n.nspname) AS schema,
+			quote_ident(c.relname) as name,
+			0 as partition,
+			cast(c.reltuples as bigint) AS relTuples,
+			CASE WHEN p.policytype = 'r' THEN true ELSE false END as isReplicated
 		FROM
 			pg_class c
 			JOIN pg_namespace n ON (c.relnamespace=n.oid)
@@ -81,11 +85,15 @@ func (tm *QueryManager) GetUserTables(conn *dbconn.DBConn) (map[string]option.Ta
 			AND c.relkind <> 'm'
 		ORDER BY c.relpages DESC
 		`
-	} else {
+	} else if conn.Version.IsGPDB() && conn.Version.AtLeast("6") {
 		query = `
 		SELECT t.* FROM (
 		SELECT
-			quote_ident(n.nspname) AS schema, quote_ident(c.relname) as name, 0 as partition, cast(c.reltuples as bigint) AS relTuples
+			quote_ident(n.nspname) AS schema,
+			quote_ident(c.relname) as name,
+			0 as partition,
+			cast(c.reltuples as bigint) AS relTuples,
+			CASE WHEN p.policytype = 'r' THEN true ELSE false END as isReplicated
 		FROM
 			pg_class c
 			JOIN pg_namespace n ON (c.relnamespace=n.oid)
@@ -99,7 +107,48 @@ func (tm *QueryManager) GetUserTables(conn *dbconn.DBConn) (map[string]option.Ta
 		ORDER BY c.relpages DESC ) t
 		UNION ALL
 		SELECT
-			quote_ident(n.nspname) AS schema, quote_ident(cparent.relname) AS name, 0 as partition, cast(cparent.reltuples as bigint) AS relTuples
+			quote_ident(n.nspname) AS schema,
+			quote_ident(cparent.relname) AS name,
+			0 as partition,
+			cast(cparent.reltuples as bigint) AS relTuples,
+			false as isReplicated
+		FROM pg_partition p
+			JOIN pg_partition_rule r ON p.oid = r.paroid
+			JOIN pg_class cparent ON cparent.oid = r.parchildrelid
+			JOIN pg_namespace n ON cparent.relnamespace=n.oid
+			JOIN (SELECT parrelid AS relid, max(parlevel) AS pl
+				FROM pg_partition GROUP BY parrelid) AS levels ON p.parrelid = levels.relid
+		WHERE
+			r.parchildrelid != 0
+			AND p.parlevel = levels.pl
+			AND n.nspname NOT IN ('gpexpand', 'pg_bitmapindex', 'information_schema', 'gp_toolkit')
+			AND n.nspname NOT LIKE 'pg_temp_%' AND cparent.relstorage NOT IN ('v', 'x', 'f')`
+	} else {
+		query = `
+		SELECT t.* FROM (
+		SELECT
+			quote_ident(n.nspname) AS schema,
+			quote_ident(c.relname) as name,
+			0 as partition,
+			cast(c.reltuples as bigint) AS relTuples,
+			false as isReplicated
+		FROM
+			pg_class c
+			JOIN pg_namespace n ON (c.relnamespace=n.oid)
+		WHERE
+			c.oid NOT IN ( SELECT parchildrelid as oid FROM pg_partition_rule )
+			AND c.oid NOT IN ( SELECT parrelid as oid FROM pg_partition )
+			AND n.nspname NOT IN ('gpexpand', 'pg_bitmapindex', 'information_schema', 'gp_toolkit')
+			AND n.nspname NOT LIKE 'pg_temp_%' AND c.relstorage NOT IN ('v', 'x', 'f')
+			AND c.relkind <> 'm'
+		ORDER BY c.relpages DESC ) t
+		UNION ALL
+		SELECT
+			quote_ident(n.nspname) AS schema,
+			quote_ident(cparent.relname) AS name,
+			0 as partition,
+			cast(cparent.reltuples as bigint) AS relTuples,
+			false as isReplicated
 		FROM pg_partition p
 			JOIN pg_partition_rule r ON p.oid = r.paroid
 			JOIN pg_class cparent ON cparent.oid = r.parchildrelid
@@ -123,7 +172,7 @@ func (tm *QueryManager) GetUserTables(conn *dbconn.DBConn) (map[string]option.Ta
 	results := make(map[string]option.TableStatistics)
 	for _, t := range tables {
 		k := t.Schema + "." + t.Name
-		results[k] = option.TableStatistics{Partition: t.Partition, RelTuples: t.RelTuples}
+		results[k] = option.TableStatistics{Partition: t.Partition, RelTuples: t.RelTuples, IsReplicated: t.IsReplicated}
 	}
 
 	return results, nil
diff --git a/copy/copy_query_wrapper.go b/copy/copy_query_wrapper.go
@@ -68,10 +68,11 @@ func (qw *QueryWrapper) excludeTables(includeTables, excludeTables map[string]op
 		if !exists {
 			sl := strings.Split(k, ".")
 			results = append(results, option.Table{
-				Schema:    sl[0],
-				Name:      sl[1],
-				Partition: v.Partition,
-				RelTuples: v.RelTuples,
+				Schema:       sl[0],
+				Name:         sl[1],
+				Partition:    v.Partition,
+				RelTuples:    v.RelTuples,
+				IsReplicated: v.IsReplicated,
 			})
 		}
 	}
@@ -116,10 +117,11 @@ func (qw *QueryWrapper) redirectIncludeTables(tables []option.Table) []option.Ta
 
 	for _, v := range tables {
 		results = append(results, option.Table{
-			Schema:    ds,
-			Name:      v.Name,
-			Partition: v.Partition,
-			RelTuples: v.RelTuples,
+			Schema:       ds,
+			Name:         v.Name,
+			Partition:    v.Partition,
+			RelTuples:    v.RelTuples,
+			IsReplicated: v.IsReplicated,
 		})
 	}
 
@@ -273,7 +275,11 @@ func (qw *QueryWrapper) FormUserTableMap(srcTables, destTables []option.Table) m
 	result := make(map[string]string)
 
 	for i, t := range srcTables {
-		result[destTables[i].Schema+"."+destTables[i].Name] = t.Schema + "." + t.Name + "." + strconv.FormatInt(t.RelTuples, 10)
+		isReplicatedStr := "false"
+		if t.IsReplicated {
+			isReplicatedStr = "true"
+		}
+		result[destTables[i].Schema+"."+destTables[i].Name] = t.Schema + "." + t.Name + "." + strconv.FormatInt(t.RelTuples, 10) + "." + isReplicatedStr
 	}
 
 	return result
@@ -391,7 +397,11 @@ func (qw *QueryWrapper) expandPartTables(conn *dbconn.DBConn, userTables map[str
 
 		stat, exists := userTables[fqn]
 		if exists {
-			expandMap[fqn] = option.TableStatistics{Partition: 0, RelTuples: stat.RelTuples}
+			expandMap[fqn] = option.TableStatistics{
+				Partition:    0,
+				RelTuples:    stat.RelTuples,
+				IsReplicated: stat.IsReplicated,
+			}
 		} else {
 			pendingCheckRels[fqn] = option.TableStatistics{Partition: 0, RelTuples: t.RelTuples}
 		}
@@ -471,16 +481,18 @@ func (qw *QueryWrapper) excludeTablePair(srcTables, destTables, exclTables []opt
 		sld := strings.Split(v, ".")
 
 		excludedSrcTabs = append(excludedSrcTabs, option.Table{
-			Schema:    sls[0],
-			Name:      sls[1],
-			Partition: u.Partition,
-			RelTuples: u.RelTuples,
+			Schema:       sls[0],
+			Name:         sls[1],
+			Partition:    u.Partition,
+			RelTuples:    u.RelTuples,
+			IsReplicated: u.IsReplicated,
 		})
 		excludedDstTabs = append(excludedDstTabs, option.Table{
-			Schema:    sld[0],
-			Name:      sld[1],
-			Partition: u.Partition,
-			RelTuples: u.RelTuples,
+			Schema:       sld[0],
+			Name:         sld[1],
+			Partition:    u.Partition,
+			RelTuples:    u.RelTuples,
+			IsReplicated: u.IsReplicated,
 		})
 
 		gplog.Debug("mapping table from \"%v\" to \"%v\"", k, v)
diff --git a/end_to_end/table_test.go b/end_to_end/table_test.go
diff --git a/meta/builtin/parallel_depend.go b/meta/builtin/parallel_depend.go
diff --git a/option/option.go b/option/option.go

Original file line number	Diff line number	Diff line change
`@@ -182,7 +182,8 @@ func (tc *TableCopier) cleanupAfterCopy(isSkipped bool, inTxn bool, err error) {`
`182`	`182`	`}`
`183`	`183`
`184`	`184`	`func (tc *TableCopier) copyData() error {`
`185`		`- command := CreateCopyStrategy(tc.srcTable.RelTuples,`
	`185`	`+ command := CreateCopyStrategy(tc.srcTable.IsReplicated,`
	`186`	`+ tc.srcTable.RelTuples,`
`186`	`187`	`tc.workerID,`
`187`	`188`	`tc.manager.srcSegmentsHostInfo,`
`188`	`189`	`tc.manager.destSegmentsIpInfo,`