dm: add openapi physical import config (#10194)

dm/config/task.go

@@ -345,9 +345,6 @@ func (m *LoaderConfig) adjust() error {
 		m.PoolSize = defaultPoolSize
 	}
 
-	if m.OnDuplicateLogical == "" && m.OnDuplicate != "" {
-		m.OnDuplicateLogical = m.OnDuplicate
-	}
 	if m.OnDuplicateLogical == "" {
 		m.OnDuplicateLogical = OnDuplicateReplace
 	}

dm/config/task_converters.go

@@ -199,6 +199,15 @@ func OpenAPITaskToSubTaskConfigs(task *openapi.Task, toDBCfg *dbconfig.DBConfig,
 		subTaskCfg.MydumperConfig = DefaultMydumperConfig()
 		subTaskCfg.LoaderConfig = DefaultLoaderConfig()
 		if fullCfg := task.SourceConfig.FullMigrateConf; fullCfg != nil {
+			if fullCfg.Analyze != nil {
+				subTaskCfg.LoaderConfig.Analyze = PhysicalPostOpLevel(*fullCfg.Analyze)
+			}
+			if fullCfg.Checksum != nil {
+				subTaskCfg.LoaderConfig.ChecksumPhysical = PhysicalPostOpLevel(*fullCfg.Checksum)
+			}
+			if fullCfg.CompressKvPairs != nil {
+				subTaskCfg.CompressKVPairs = *fullCfg.CompressKvPairs
+			}
 			if fullCfg.Consistency != nil {
 				subTaskCfg.MydumperConfig.ExtraArgs = fmt.Sprintf("--consistency %s", *fullCfg.Consistency)
 			}
@@ -211,7 +220,29 @@ func OpenAPITaskToSubTaskConfigs(task *openapi.Task, toDBCfg *dbconfig.DBConfig,
 			if fullCfg.DataDir != nil {
 				subTaskCfg.LoaderConfig.Dir = *fullCfg.DataDir
 			}
-			subTaskCfg.LoaderConfig.OnDuplicateLogical = LogicalDuplicateResolveType(task.OnDuplicate)
+			if fullCfg.DiskQuota != nil {
+				if err := subTaskCfg.LoaderConfig.DiskQuotaPhysical.UnmarshalText([]byte(*fullCfg.DiskQuota)); err != nil {
+					return nil, err
+				}
+			}
+			if fullCfg.ImportMode != nil {
+				subTaskCfg.LoaderConfig.ImportMode = LoadMode(*fullCfg.ImportMode)
+			}
+			if fullCfg.OnDuplicateLogical != nil {
+				subTaskCfg.LoaderConfig.OnDuplicateLogical = LogicalDuplicateResolveType(*fullCfg.OnDuplicateLogical)
+			}
+			if fullCfg.OnDuplicatePhysical != nil {
+				subTaskCfg.LoaderConfig.OnDuplicatePhysical = PhysicalDuplicateResolveType(*fullCfg.OnDuplicatePhysical)
+			}
+			if fullCfg.PdAddr != nil {
+				subTaskCfg.LoaderConfig.PDAddr = *fullCfg.PdAddr
+			}
+			if fullCfg.RangeConcurrency != nil {
+				subTaskCfg.LoaderConfig.RangeConcurrency = *fullCfg.RangeConcurrency
+			}
+			if fullCfg.SortingDir != nil {
+				subTaskCfg.LoaderConfig.SortingDirPhysical = *fullCfg.SortingDir
+			}
 		}
 		// set incremental config
 		subTaskCfg.SyncerConfig = DefaultSyncerConfig()

dm/openapi/fixtures/task.go

@@ -26,7 +26,7 @@ var (
 		"enhance_online_schema_change": true,
 		"meta_schema": "dm_meta",
 		"name": "test",
-		"on_duplicate": "error",
+		"on_duplicate": "replace",
 		"source_config": {
 		  "full_migrate_conf": {
 			"data_dir": "./exported_data",
@@ -63,7 +63,7 @@ var (
 		"enhance_online_schema_change": true,
 		"meta_schema": "dm_meta",
 		"name": "a5fb4a7540d343fa853c55ade2d08e6d03681d9e05d6240c0",
-		"on_duplicate": "error",
+		"on_duplicate": "replace",
 		"source_config": {
 		  "full_migrate_conf": {
 			"data_dir": "./exported_data",
@@ -103,7 +103,7 @@ var (
 		"enhance_online_schema_change": true,
 		"meta_schema": "dm_meta",
 		"name": "test",
-		"on_duplicate": "error",
+		"on_duplicate": "replace",
 		"shard_mode": "optimistic",
 		"strict_optimistic_shard_mode": true,
 		"source_config": {

dm/openapi/spec/dm.yaml

@@ -1685,6 +1685,62 @@ components:
           type: string
           example: "auto"
           description: "to control the way in which data is exported for consistency assurance"
+        import_mode:
+          type: string
+          example: "logical"
+          description: "to control import mode of full import"
+          enum:
+            - "logical"
+            - "physical"
+        sorting_dir:
+          type: string
+          example: "./sort_dir"
+          description: "sorting dir name for physical import"
+        disk_quota:
+          type: string
+          example: "80G"
+          description: "disk quota for physical import"
+        checksum:
+          type: string
+          example: "optional"
+          description: "to control checksum of physical import"
+          enum:
+            - "required"
+            - "optional"
+            - "off"
+        analyze:
+          type: string
+          example: "optional"
+          description: "to control checksum of physical import"
+          enum:
+            - "required"
+            - "optional"
+            - "off"
+        range_concurrency:
+          type: integer
+          description: "to control range concurrency of physical import"
+          default: 0
+        compress-kv-pairs:
+          type: string
+          description: "to control compress kv pairs of physical import"
+        pd_addr:
+          type: string
+          description: "address of pd"
+        on_duplicate_logical:
+          type: string
+          example: "replace"
+          description: "to control the duplication resolution when meet duplicate rows for logical import"
+          enum:
+            - "replace"
+            - "error"
+            - "ignore"
+        on_duplicate_physical:
+          type: string
+          example: "none"
+          description: "to control the duplication resolution when meet duplicate rows for physical import"
+          enum:
+            - "none"
+            - "manual"
     TaskIncrMigrateConf:
       description: configuration of incremental tasks
       type: object

dm/tests/lightning_mode/conf/dm-master.toml

@@ -4,3 +4,4 @@ advertise-addr = "127.0.0.1:8261"
 
 rpc-timeout = "30s"
 auto-compaction-retention = "3s"
+openapi = true