Restrict live migration to single table-list and added guardrails for…

… that around table-list flags (#2354)

yb-voyager/cmd/common.go

@@ -1040,6 +1040,9 @@ func storeTableListInMSR(tableList []sqlname.NameTuple) error {
 	}))
 	err := metaDB.UpdateMigrationStatusRecord(func(record *metadb.MigrationStatusRecord) {
 		record.TableListExportedFromSource = minQuotedTableList
+		record.SourceExportedTableListWithLeafPartitions = lo.Map(tableList, func(t sqlname.NameTuple, _ int) string {
+			return t.ForOutput()
+		})
 	})
 	if err != nil {
 		return fmt.Errorf("update migration status record: %v", err)

yb-voyager/src/metadb/migrationStatus.go

@@ -11,18 +11,21 @@ import (
 )
 
 type MigrationStatusRecord struct {
-	MigrationUUID               string            `json:"MigrationUUID"`
-	VoyagerVersion              string            `json:"VoyagerVersion"`
-	ExportType                  string            `json:"ExportType"`
-	ArchivingEnabled            bool              `json:"ArchivingEnabled"`
-	FallForwardEnabled          bool              `json:"FallForwardEnabled"`
-	FallbackEnabled             bool              `json:"FallbackEnabled"`
-	UseYBgRPCConnector          bool              `json:"UseYBgRPCConnector"`
-	TargetDBConf                *tgtdb.TargetConf `json:"TargetDBConf"`
-	SourceReplicaDBConf         *tgtdb.TargetConf `json:"SourceReplicaDBConf"`
-	SourceDBAsTargetConf        *tgtdb.TargetConf `json:"SourceDBAsTargetConf"`
-	TableListExportedFromSource []string          `json:"TableListExportedFromSource"`
-	SourceDBConf                *srcdb.Source     `json:"SourceDBConf"`
+	MigrationUUID                             string            `json:"MigrationUUID"`
+	VoyagerVersion                            string            `json:"VoyagerVersion"`
+	ExportType                                string            `json:"ExportType"`
+	ArchivingEnabled                          bool              `json:"ArchivingEnabled"`
+	FallForwardEnabled                        bool              `json:"FallForwardEnabled"`
+	FallbackEnabled                           bool              `json:"FallbackEnabled"`
+	UseYBgRPCConnector                        bool              `json:"UseYBgRPCConnector"`
+	TargetDBConf                              *tgtdb.TargetConf `json:"TargetDBConf"`
+	SourceReplicaDBConf                       *tgtdb.TargetConf `json:"SourceReplicaDBConf"`
+	SourceDBAsTargetConf                      *tgtdb.TargetConf `json:"SourceDBAsTargetConf"`
+	TableListExportedFromSource               []string          `json:"TableListExportedFromSource"`
+	SourceExportedTableListWithLeafPartitions []string          `json:"SourceExportedTableListWithLeafPartitions"` // will be same as `TableListExportedFromSource` for Oracle and MySQL but will have leaf partitions in case of PG
+	TargetExportedTableListWithLeafPartitions []string          `json:"TargetExportedTableListWithLeafPartitions"` // will be the table list for export data from target with leaf partitions
+
+	SourceDBConf *srcdb.Source `json:"SourceDBConf"`
 
 	CutoverToTargetRequested                        bool `json:"CutoverToTargetRequested"`
 	CutoverProcessedBySourceExporter                bool `json:"CutoverProcessedBySourceExporter"`

yb-voyager/src/namereg/namereg.go

@@ -3,6 +3,7 @@ package namereg
 import (
 	"errors"
 	"fmt"
+	"slices"
 	"strings"
 
 	"github.com/samber/lo"
@@ -68,6 +69,9 @@ type NameRegistry struct {
 	// Source replica has same table name list as original source.
 
 	params NameRegistryParams
+
+	SourceDBSequenceNames map[string][]string
+	YBSequenceNames       map[string][]string
 }
 
 func InitNameRegistry(params NameRegistryParams) error {
@@ -148,6 +152,7 @@ func (reg *NameRegistry) registerSourceNames() (bool, error) {
 	reg.SourceDBType = reg.params.SourceDBType
 	reg.initSourceDBSchemaNames()
 	m := make(map[string][]string)
+	m1 := make(map[string][]string)
 	for _, schemaName := range reg.SourceDBSchemaNames {
 		tableNames, err := reg.params.SDB.GetAllTableNamesRaw(schemaName)
 		if err != nil {
@@ -159,11 +164,49 @@ func (reg *NameRegistry) registerSourceNames() (bool, error) {
 			return false, fmt.Errorf("get all sequence names: %w", err)
 		}
 		m[schemaName] = append(m[schemaName], seqNames...)
+		m1[schemaName] = append(m1[schemaName], seqNames...)
 	}
 	reg.SourceDBTableNames = m
+	reg.SourceDBSequenceNames = m1
 	return true, nil
 }
 
+// this function returns the tables names in the namereg which is required by the export data
+// table-list code path for getting a full list of tables from first run i.e. registered table list
+// returning the objectNames from here as we need to get nameTuple based on type of tables either leaf partition or normal..
+func (reg *NameRegistry) GetRegisteredTableList() ([]*sqlname.ObjectName, error) {
+	var res []*sqlname.ObjectName
+	var m map[string][]string            // Complete list of tables and sequences
+	var sequencesMap map[string][]string // only sequence list
+	var dbType string
+	var defaultSchemaName string
+	switch reg.params.Role {
+	case SOURCE_DB_EXPORTER_ROLE, SOURCE_DB_IMPORTER_ROLE, SOURCE_REPLICA_DB_IMPORTER_ROLE:
+		m = reg.SourceDBTableNames
+		sequencesMap = reg.SourceDBSequenceNames
+		dbType = reg.SourceDBType
+		defaultSchemaName = reg.DefaultSourceDBSchemaName
+		if reg.params.Role == SOURCE_REPLICA_DB_IMPORTER_ROLE {
+			defaultSchemaName = reg.DefaultSourceReplicaDBSchemaName
+		}
+	case TARGET_DB_EXPORTER_FB_ROLE, TARGET_DB_EXPORTER_FF_ROLE, TARGET_DB_IMPORTER_ROLE:
+		m = reg.YBTableNames
+		sequencesMap = reg.YBSequenceNames
+		dbType = constants.YUGABYTEDB
+		defaultSchemaName = reg.DefaultYBSchemaName
+	}
+	for s, tables := range m {
+		for _, t := range tables {
+			if slices.Contains(sequencesMap[s], t) {
+				//If its a sequence continue and not append in the registerd list
+				continue
+			}
+			res = append(res, sqlname.NewObjectName(dbType, defaultSchemaName, s, t))
+		}
+	}
+	return res, nil
+}
+
 func (reg *NameRegistry) initSourceDBSchemaNames() {
 	// source.Schema contains only one schema name for MySQL and Oracle; whereas
 	// it contains a pipe separated list for postgres.
@@ -191,6 +234,7 @@ func (reg *NameRegistry) registerYBNames() (bool, error) {
 	yb := reg.params.YBDB
 
 	m := make(map[string][]string)
+	m1 := make(map[string][]string)
 	reg.DefaultYBSchemaName = reg.params.TargetDBSchema
 	if reg.SourceDBTableNames != nil && reg.SourceDBType == constants.POSTGRESQL {
 		reg.DefaultYBSchemaName = reg.DefaultSourceDBSchemaName
@@ -212,8 +256,10 @@ func (reg *NameRegistry) registerYBNames() (bool, error) {
 			return false, fmt.Errorf("get all sequence names: %w", err)
 		}
 		m[schemaName] = append(m[schemaName], seqNames...)
+		m1[schemaName] = append(m1[schemaName], seqNames...)
 	}
 	reg.YBTableNames = m
+	reg.YBSequenceNames = m1
 	return true, nil
 }
 

yb-voyager/src/namereg/namereg_test.go

@@ -607,6 +607,8 @@ func TestNameRegistryStructs(t *testing.T) {
 				YBTableNames                     map[string][]string
 				DefaultSourceReplicaDBSchemaName string
 				params                           NameRegistryParams
+				SourceDBSequenceNames           map[string][]string
+				YBSequenceNames                 map[string][]string
 			}{},
 		},
 	}
@@ -692,7 +694,9 @@ func TestNameRegistryJson(t *testing.T) {
 		`      "lower_caps"`,
 		"    ]",
 		"  },",
-		`  "DefaultSourceReplicaDBSchemaName": "SAKILA_FF"`,
+		`  "DefaultSourceReplicaDBSchemaName": "SAKILA_FF",`,
+		`  "SourceDBSequenceNames": null,`,
+        `  "YBSequenceNames": null`,
 		"}",
 	}, "\n")
 

yb-voyager/src/utils/logging.go

@@ -26,7 +26,7 @@ import (
 
 var ErrExitErr error
 
-func ErrExit(formatString string, args ...interface{}) {
+var ErrExit = func(formatString string, args ...interface{}) {
 	ErrExitErr = fmt.Errorf(formatString, args...)
 	formatString = strings.Replace(formatString, "%w", "%s", -1)
 	fmt.Fprintf(os.Stderr, formatString+"\n", args...)

yb-voyager/src/utils/sqlname/nametuple.go

@@ -22,6 +22,7 @@ import (
 	"strings"
 
 	"github.com/samber/lo"
+
 	"github.com/yugabyte/yb-voyager/yb-voyager/src/constants"
 )
 
@@ -60,6 +61,14 @@ func NewObjectName(dbType, defaultSchemaName, schemaName, tableName string) *Obj
 	return result
 }
 
+func NewObjectNameWithQualifiedName(dbType, defaultSchemaName, objName string) *ObjectName {
+	parts := strings.Split(objName, ".")
+	if len(parts) != 2 {
+		panic(fmt.Sprintf("invalid qualified name: %s", objName))
+	}
+	return NewObjectName(dbType, defaultSchemaName, parts[0], unquote(parts[1], dbType))
+}
+
 func (nv *ObjectName) String() string {
 	return nv.MinQualified.MinQuoted
 }