Repository: Altinity/clickhouse-backup (go)

Issue:
Implementation --restore-table-mapping
It looks like the tool doesn't support restoring a table backup to a table with a different name. In clickhouse, we can do this by using the `[AS [db.]table_name_in_backup]` clause in the `RESTORE TABLE`. For the restore related commands, we can have an option similar to `--restore-database-mapping` which can be used to accomplish this.
Implementation --restore-table-mapping
It looks like the tool doesn't support restoring a table backup to a table with a different name. In clickhouse, we can do this by using the `[AS [db.]table_name_in_backup]` clause in the `RESTORE TABLE`. For the restore related commands, we can have an option similar to `--restore-database-mapping` which can be used to accomplish this.


Discussion:
you are welcome to make pull request

sure
you are welcome to make pull request

sure

Patch:
diff --git a/Manual.md b/Manual.md
--- a/Manual.md
+++ b/Manual.md
@@ -147,13 +147,14 @@ NAME:
    clickhouse-backup restore - Create schema and restore data from backup
 
 USAGE:
-   clickhouse-backup restore  [-t, --tables=<db>.<table>] [-m, --restore-database-mapping=<originDB>:<targetDB>[,<...>]] [--partitions=<partitions_names>] [-s, --schema] [-d, --data] [--rm, --drop] [-i, --ignore-dependencies] [--rbac] [--configs] <backup_name>
+   clickhouse-backup restore  [-t, --tables=<db>.<table>] [-m, --restore-database-mapping=<originDB>:<targetDB>[,<...>]] [--tm, --restore-table-mapping=<originTable>:<targetTable>[,<...>]] [--partitions=<partitions_names>] [-s, --schema] [-d, --data] [--rm, --drop] [-i, --ignore-dependencies] [--rbac] [--configs] <backup_name>
 
 OPTIONS:
    --config value, -c value                    Config 'FILE' name. (default: "/etc/clickhouse-backup/config.yml") [$CLICKHOUSE_BACKUP_CONFIG]
    --environment-override value, --env value   override any environment variable via CLI parameter
    --table value, --tables value, -t value     Restore only database and objects which matched with table name patterns, separated by comma, allow ? and * as wildcard
    --restore-database-mapping value, -m value  Define the rule to restore data. For the database not defined in this struct, the program will not deal with it.
+   --restore-table-mapping value, --tm value   Define the rule to restore data. For the table not defined in this struct, the program will not deal with it.
    --partitions partition_id                   Restore backup only for selected partition names, separated by comma
 If PARTITION BY clause returns numeric not hashed values for partition_id field in system.parts table, then use --partitions=partition_id1,partition_id2 format
 If PARTITION BY clause returns hashed string values, then use --partitions=('non_numeric_field_value_for_part1'),('non_numeric_field_value_for_part2') format
@@ -177,13 +178,14 @@ NAME:
    clickhouse-backup restore_remote - Download and restore
 
 USAGE:
-   clickhouse-backup restore_remote [--schema] [--data] [-t, --tables=<db>.<table>] [-m, --restore-database-mapping=<originDB>:<targetDB>[,<...>]] [--partitions=<partitions_names>] [--rm, --drop] [-i, --ignore-dependencies] [--rbac] [--configs] [--skip-rbac] [--skip-configs] [--resumable] <backup_name>
+   clickhouse-backup restore_remote [--schema] [--data] [-t, --tables=<db>.<table>] [-m, --restore-database-mapping=<originDB>:<targetDB>[,<...>]] [--tm, --restore-table-mapping=<originTable>:<targetTable>[,<...>]] [--partitions=<partitions_names>] [--rm, --drop] [-i, --ignore-dependencies] [--rbac] [--configs] [--skip-rbac] [--skip-configs] [--resumable] <backup_name>
 
 OPTIONS:
    --config value, -c value                    Config 'FILE' name. (default: "/etc/clickhouse-backup/config.yml") [$CLICKHOUSE_BACKUP_CONFIG]
    --environment-override value, --env value   override any environment variable via CLI parameter
    --table value, --tables value, -t value     Download and restore objects which matched with table name patterns, separated by comma, allow ? and * as wildcard
    --restore-database-mapping value, -m value  Define the rule to restore data. For the database not defined in this struct, the program will not deal with it.
+   --restore-table-mapping value, --tm value    Define the rule to restore data. For the table not defined in this struct, the program will not deal with it.
    --partitions partition_id                   Download and restore backup only for selected partition names, separated by comma
 If PARTITION BY clause returns numeric not hashed values for partition_id field in system.parts table, then use --partitions=partition_id1,partition_id2 format
 If PARTITION BY clause returns hashed string values, then use --partitions=('non_numeric_field_value_for_part1'),('non_numeric_field_value_for_part2') format
diff --git a/ReadMe.md b/ReadMe.md
--- a/ReadMe.md
+++ b/ReadMe.md
@@ -119,6 +119,10 @@ general:
   # RESTORE_DATABASE_MAPPING, restore rules from backup databases to target databases, which is useful when changing destination database, all atomic tables will be created with new UUIDs.
   # The format for this env variable is "src_db1:target_db1,src_db2:target_db2". For YAML please continue using map syntax
   restore_database_mapping: {}
+
+  # RESTORE_TABLE_MAPPING, restore rules from backup tables to target tables, which is useful when changing destination tables.
+  # The format for this env variable is "src_table1:target_table1,src_table2:target_table2". For YAML please continue using map syntax
+  restore_table_mapping: {}
   retries_on_failure: 3          # RETRIES_ON_FAILURE, how many times to retry after a failure during upload or download
   retries_pause: 30s             # RETRIES_PAUSE, duration time to pause after each download or upload failure
 
@@ -476,6 +480,7 @@ Create schema and restore data from backup: `curl -s localhost:7171/backup/resto
 - Optional query argument `rbac` works the same as the `--rbac` CLI argument (restore RBAC).
 - Optional query argument `configs` works the same as the `--configs` CLI argument (restore configs).
 - Optional query argument `restore_database_mapping` works the same as the `--restore-database-mapping` CLI argument.
+- Optional query argument `restore_table_mapping` works the same as the `--restore-table-mapping` CLI argument.
 - Optional query argument `callback` allow pass callback URL which will call with POST with `application/json` with payload `{"status":"error|success","error":"not empty when error happens"}`.
 
 ### POST /backup/delete
@@ -705,13 +710,14 @@ NAME:
    clickhouse-backup restore - Create schema and restore data from backup
 
 USAGE:
-   clickhouse-backup restore  [-t, --tables=<db>.<table>] [-m, --restore-database-mapping=<originDB>:<targetDB>[,<...>]] [--partitions=<partitions_names>] [-s, --schema] [-d, --data] [--rm, --drop] [-i, --ignore-dependencies] [--rbac] [--configs] <backup_name>
+   clickhouse-backup restore  [-t, --tables=<db>.<table>] [-m, --restore-database-mapping=<originDB>:<targetDB>[,<...>]] [--tm, --restore-table-mapping=<originTable>:<targetTable>[,<...>]] [--partitions=<partitions_names>] [-s, --schema] [-d, --data] [--rm, --drop] [-i, --ignore-dependencies] [--rbac] [--configs] <backup_name>
 
 OPTIONS:
    --config value, -c value                    Config 'FILE' name. (default: "/etc/clickhouse-backup/config.yml") [$CLICKHOUSE_BACKUP_CONFIG]
    --environment-override value, --env value   override any environment variable via CLI parameter
    --table value, --tables value, -t value     Restore only database and objects which matched with table name patterns, separated by comma, allow ? and * as wildcard
    --restore-database-mapping value, -m value  Define the rule to restore data. For the database not defined in this struct, the program will not deal with it.
+   --restore-table-mapping value, --tm value   Define the rule to restore data. For the table not defined in this struct, the program will not deal with it.
    --partitions partition_id                   Restore backup only for selected partition names, separated by comma
 If PARTITION BY clause returns numeric not hashed values for partition_id field in system.parts table, then use --partitions=partition_id1,partition_id2 format
 If PARTITION BY clause returns hashed string values, then use --partitions=('non_numeric_field_value_for_part1'),('non_numeric_field_value_for_part2') format
@@ -735,13 +741,14 @@ NAME:
    clickhouse-backup restore_remote - Download and restore
 
 USAGE:
-   clickhouse-backup restore_remote [--schema] [--data] [-t, --tables=<db>.<table>] [-m, --restore-database-mapping=<originDB>:<targetDB>[,<...>]] [--partitions=<partitions_names>] [--rm, --drop] [-i, --ignore-dependencies] [--rbac] [--configs] [--skip-rbac] [--skip-configs] [--resumable] <backup_name>
+   clickhouse-backup restore_remote [--schema] [--data] [-t, --tables=<db>.<table>] [-m, --restore-database-mapping=<originDB>:<targetDB>[,<...>]] [--tm, --restore-table-mapping=<originTable>:<targetTable>[,<...>]] [--partitions=<partitions_names>] [--rm, --drop] [-i, --ignore-dependencies] [--rbac] [--configs] [--skip-rbac] [--skip-configs] [--resumable] <backup_name>
 
 OPTIONS:
    --config value, -c value                    Config 'FILE' name. (default: "/etc/clickhouse-backup/config.yml") [$CLICKHOUSE_BACKUP_CONFIG]
    --environment-override value, --env value   override any environment variable via CLI parameter
    --table value, --tables value, -t value     Download and restore objects which matched with table name patterns, separated by comma, allow ? and * as wildcard
    --restore-database-mapping value, -m value  Define the rule to restore data. For the database not defined in this struct, the program will not deal with it.
+   --restore-table-mapping value, --tm value   Define the rule to restore data. For the table not defined in this struct, the program will not deal with it.
    --partitions partition_id                   Download and restore backup only for selected partition names, separated by comma
 If PARTITION BY clause returns numeric not hashed values for partition_id field in system.parts table, then use --partitions=partition_id1,partition_id2 format
 If PARTITION BY clause returns hashed string values, then use --partitions=('non_numeric_field_value_for_part1'),('non_numeric_field_value_for_part2') format
diff --git a/cmd/clickhouse-backup/main.go b/cmd/clickhouse-backup/main.go
--- a/cmd/clickhouse-backup/main.go
+++ b/cmd/clickhouse-backup/main.go
@@ -6,15 +6,14 @@ import (
 	"os"
 	"strings"
 
-	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/logcli"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/status"
+	"github.com/apex/log"
+	"github.com/urfave/cli"
 
 	"github.com/Altinity/clickhouse-backup/v2/pkg/backup"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/logcli"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/server"
-
-	"github.com/apex/log"
-	"github.com/urfave/cli"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/status"
 )
 
 var (
@@ -337,10 +336,10 @@ func main() {
 		{
 			Name:      "restore",
 			Usage:     "Create schema and restore data from backup",
-			UsageText: "clickhouse-backup restore  [-t, --tables=<db>.<table>] [-m, --restore-database-mapping=<originDB>:<targetDB>[,<...>]] [--partitions=<partitions_names>] [-s, --schema] [-d, --data] [--rm, --drop] [-i, --ignore-dependencies] [--rbac] [--configs] <backup_name>",
+			UsageText: "clickhouse-backup restore  [-t, --tables=<db>.<table>] [-m, --restore-database-mapping=<originDB>:<targetDB>[,<...>]] [--tm, --restore-table-mapping=<originTable>:<targetTable>[,<...>]] [--partitions=<partitions_names>] [-s, --schema] [-d, --data] [--rm, --drop] [-i, --ignore-dependencies] [--rbac] [--configs] <backup_name>",
 			Action: func(c *cli.Context) error {
 				b := backup.NewBackuper(config.GetConfigFromCli(c))
-				return b.Restore(c.Args().First(), c.String("t"), c.StringSlice("restore-database-mapping"), c.StringSlice("partitions"), c.Bool("schema"), c.Bool("data"), c.Bool("drop"), c.Bool("ignore-dependencies"), c.Bool("rbac"), c.Bool("rbac-only"), c.Bool("configs"), c.Bool("configs-only"), version, c.Int("command-id"))
+				return b.Restore(c.Args().First(), c.String("t"), c.StringSlice("restore-database-mapping"), c.StringSlice("restore-table-mapping"), c.StringSlice("partitions"), c.Bool("schema"), c.Bool("data"), c.Bool("drop"), c.Bool("ignore-dependencies"), c.Bool("rbac"), c.Bool("rbac-only"), c.Bool("configs"), c.Bool("configs-only"), version, c.Int("command-id"))
 			},
 			Flags: append(cliapp.Flags,
 				cli.StringFlag{
@@ -353,6 +352,11 @@ func main() {
 					Usage:  "Define the rule to restore data. For the database not defined in this struct, the program will not deal with it.",
 					Hidden: false,
 				},
+				cli.StringSliceFlag{
+					Name:   "restore-table-mapping, tm",
+					Usage:  "Define the rule to restore data. For the table not defined in this struct, the program will not deal with it.",
+					Hidden: false,
+				},
 				cli.StringSliceFlag{
 					Name:   "partitions",
 					Hidden: false,
@@ -409,10 +413,10 @@ func main() {
 		{
 			Name:      "restore_remote",
 			Usage:     "Download and restore",
-			UsageText: "clickhouse-backup restore_remote [--schema] [--data] [-t, --tables=<db>.<table>] [-m, --restore-database-mapping=<originDB>:<targetDB>[,<...>]] [--partitions=<partitions_names>] [--rm, --drop] [-i, --ignore-dependencies] [--rbac] [--configs] [--skip-rbac] [--skip-configs] [--resumable] <backup_name>",
+			UsageText: "clickhouse-backup restore_remote [--schema] [--data] [-t, --tables=<db>.<table>] [-m, --restore-database-mapping=<originDB>:<targetDB>[,<...>]] [--tm, --restore-table-mapping=<originTable>:<targetTable>[,<...>]] [--partitions=<partitions_names>] [--rm, --drop] [-i, --ignore-dependencies] [--rbac] [--configs] [--skip-rbac] [--skip-configs] [--resumable] <backup_name>",
 			Action: func(c *cli.Context) error {
 				b := backup.NewBackuper(config.GetConfigFromCli(c))
-				return b.RestoreFromRemote(c.Args().First(), c.String("t"), c.StringSlice("restore-database-mapping"), c.StringSlice("partitions"), c.Bool("s"), c.Bool("d"), c.Bool("rm"), c.Bool("i"), c.Bool("rbac"), c.Bool("rbac-only"), c.Bool("configs"), c.Bool("configs-only"), c.Bool("resume"), version, c.Int("command-id"))
+				return b.RestoreFromRemote(c.Args().First(), c.String("t"), c.StringSlice("restore-database-mapping"), c.StringSlice("restore-table-mapping"), c.StringSlice("partitions"), c.Bool("s"), c.Bool("d"), c.Bool("rm"), c.Bool("i"), c.Bool("rbac"), c.Bool("rbac-only"), c.Bool("configs"), c.Bool("configs-only"), c.Bool("resume"), version, c.Int("command-id"))
 			},
 			Flags: append(cliapp.Flags,
 				cli.StringFlag{
@@ -425,6 +429,11 @@ func main() {
 					Usage:  "Define the rule to restore data. For the database not defined in this struct, the program will not deal with it.",
 					Hidden: false,
 				},
+				cli.StringSliceFlag{
+					Name:   "restore-table-mapping, tm",
+					Usage:  "Define the rule to restore data. For the database not defined in this struct, the program will not deal with it.",
+					Hidden: false,
+				},
 				cli.StringSliceFlag{
 					Name:   "partitions",
 					Hidden: false,
diff --git a/pkg/backup/create.go b/pkg/backup/create.go
--- a/pkg/backup/create.go
+++ b/pkg/backup/create.go
@@ -5,9 +5,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/storage"
-	"golang.org/x/sync/errgroup"
 	"os"
 	"path"
 	"path/filepath"
@@ -17,19 +14,22 @@ import (
 	"sync/atomic"
 	"time"
 
+	apexLog "github.com/apex/log"
+	"github.com/google/uuid"
+	recursiveCopy "github.com/otiai10/copy"
+	"golang.org/x/sync/errgroup"
+
 	"github.com/Altinity/clickhouse-backup/v2/pkg/clickhouse"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/common"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/filesystemhelper"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/keeper"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/metadata"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/partition"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/status"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/storage"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/storage/object_disk"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/utils"
-
-	apexLog "github.com/apex/log"
-	"github.com/google/uuid"
-	recursiveCopy "github.com/otiai10/copy"
 )
 
 const (
@@ -255,7 +255,7 @@ func (b *Backuper) createBackupLocal(ctx context.Context, backupName, diffFromRe
 	var backupDataSize, backupObjectDiskSize, backupMetadataSize uint64
 	var metaMutex sync.Mutex
 	createBackupWorkingGroup, createCtx := errgroup.WithContext(ctx)
-	createBackupWorkingGroup.SetLimit(max(b.cfg.ClickHouse.MaxConnections,1))
+	createBackupWorkingGroup.SetLimit(max(b.cfg.ClickHouse.MaxConnections, 1))
 
 	var tableMetas []metadata.TableTitle
 	for tableIdx, tableItem := range tables {
diff --git a/pkg/backup/restore.go b/pkg/backup/restore.go
--- a/pkg/backup/restore.go
+++ b/pkg/backup/restore.go
@@ -5,12 +5,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/keeper"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/status"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/storage"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/storage/object_disk"
-	"golang.org/x/sync/errgroup"
 	"io"
 	"io/fs"
 	"net/url"
@@ -23,23 +17,30 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/Altinity/clickhouse-backup/v2/pkg/common"
-
+	apexLog "github.com/apex/log"
 	"github.com/mattn/go-shellwords"
+	recursiveCopy "github.com/otiai10/copy"
+	"github.com/yargevad/filepathx"
+	"golang.org/x/sync/errgroup"
+	"golang.org/x/text/cases"
+	"golang.org/x/text/language"
 
 	"github.com/Altinity/clickhouse-backup/v2/pkg/clickhouse"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/common"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/filesystemhelper"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/keeper"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/metadata"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/status"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/storage"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/storage/object_disk"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/utils"
-	apexLog "github.com/apex/log"
-	recursiveCopy "github.com/otiai10/copy"
-	"github.com/yargevad/filepathx"
 )
 
 var CreateDatabaseRE = regexp.MustCompile(`(?m)^CREATE DATABASE (\s*)(\S+)(\s*)`)
 
 // Restore - restore tables matched by tablePattern from backupName
-func (b *Backuper) Restore(backupName, tablePattern string, databaseMapping, partitions []string, schemaOnly, dataOnly, dropExists, ignoreDependencies, restoreRBAC, rbacOnly, restoreConfigs, configsOnly bool, backupVersion string, commandId int) error {
+func (b *Backuper) Restore(backupName, tablePattern string, databaseMapping, tableMapping, partitions []string, schemaOnly, dataOnly, dropExists, ignoreDependencies, restoreRBAC, rbacOnly, restoreConfigs, configsOnly bool, backupVersion string, commandId int) error {
 	ctx, cancel, err := status.Current.GetContextWithCancel(commandId)
 	if err != nil {
 		return err
@@ -48,7 +49,10 @@ func (b *Backuper) Restore(backupName, tablePattern string, databaseMapping, par
 	defer cancel()
 	startRestore := time.Now()
 	backupName = utils.CleanBackupNameRE.ReplaceAllString(backupName, "")
-	if err := b.prepareRestoreDatabaseMapping(databaseMapping); err != nil {
+	if err := b.prepareRestoreMapping(databaseMapping, "database"); err != nil {
+		return err
+	}
+	if err := b.prepareRestoreMapping(tableMapping, "table"); err != nil {
 		return err
 	}
 
@@ -247,13 +251,23 @@ func (b *Backuper) getTablesForRestoreLocal(ctx context.Context, backupName stri
 	if err != nil {
 		return nil, nil, err
 	}
-	// if restore-database-mapping specified, create database in mapping rules instead of in backup files.
+	// if restore-database-mapping is specified, create database in mapping rules instead of in backup files.
 	if len(b.cfg.General.RestoreDatabaseMapping) > 0 {
 		err = changeTableQueryToAdjustDatabaseMapping(&tablesForRestore, b.cfg.General.RestoreDatabaseMapping)
 		if err != nil {
 			return nil, nil, err
 		}
 	}
+
+	// if restore-table-mapping is specified, create table in mapping rules instead of in backup files.
+	// https://github.com/Altinity/clickhouse-backup/issues/937
+	if len(b.cfg.General.RestoreTableMapping) > 0 {
+		err = changeTableQueryToAdjustTableMapping(&tablesForRestore, b.cfg.General.RestoreTableMapping)
+		if err != nil {
+			return nil, nil, err
+		}
+	}
+
 	if len(tablesForRestore) == 0 {
 		return nil, nil, fmt.Errorf("not found schemas by %s in %s, also check skip_tables and skip_table_engines setting", tablePattern, backupName)
 	}
@@ -356,15 +370,23 @@ func (b *Backuper) restoreEmptyDatabase(ctx context.Context, targetDB, tablePatt
 	return nil
 }
 
-func (b *Backuper) prepareRestoreDatabaseMapping(databaseMapping []string) error {
-	for i := 0; i < len(databaseMapping); i++ {
-		splitByCommas := strings.Split(databaseMapping[i], ",")
+func (b *Backuper) prepareRestoreMapping(objectMapping []string, objectType string) error {
+	if objectType != "database" && objectType != "table" {
+		return fmt.Errorf("objectType must be one of `database` or `table`")
+	}
+	for i := 0; i < len(objectMapping); i++ {
+		splitByCommas := strings.Split(objectMapping[i], ",")
 		for _, m := range splitByCommas {
 			splitByColon := strings.Split(m, ":")
 			if len(splitByColon) != 2 {
-				return fmt.Errorf("restore-database-mapping %s should only have srcDatabase:destinationDatabase format for each map rule", m)
+				objectTypeTitleCase := cases.Title(language.Und).String(objectType)
+				return fmt.Errorf("restore-%s-mapping %s should only have src%s:destination%s format for each map rule", objectType, m, objectTypeTitleCase, objectTypeTitleCase)
+			}
+			if objectType == "database" {
+				b.cfg.General.RestoreDatabaseMapping[splitByColon[0]] = splitByColon[1]
+			} else {
+				b.cfg.General.RestoreTableMapping[splitByColon[0]] = splitByColon[1]
 			}
-			b.cfg.General.RestoreDatabaseMapping[splitByColon[0]] = splitByColon[1]
 		}
 	}
 	return nil
@@ -1183,8 +1205,13 @@ func (b *Backuper) restoreDataEmbedded(ctx context.Context, backupName string, d
 
 func (b *Backuper) restoreDataRegular(ctx context.Context, backupName string, backupMetadata metadata.BackupMetadata, tablePattern string, tablesForRestore ListOfTables, diskMap, diskTypes map[string]string, disks []clickhouse.Disk, log *apexLog.Entry) error {
 	if len(b.cfg.General.RestoreDatabaseMapping) > 0 {
-		tablePattern = b.changeTablePatternFromRestoreDatabaseMapping(tablePattern)
+		tablePattern = b.changeTablePatternFromRestoreMapping(tablePattern, "database")
 	}
+	// https://github.com/Altinity/clickhouse-backup/issues/937
+	if len(b.cfg.General.RestoreTableMapping) > 0 {
+		tablePattern = b.changeTablePatternFromRestoreMapping(tablePattern, "table")
+	}
+
 	if err := b.applyMacrosToObjectDiskPath(ctx); err != nil {
 		return err
 	}
@@ -1200,23 +1227,32 @@ func (b *Backuper) restoreDataRegular(ctx context.Context, backupName string, ba
 		return fmt.Errorf("%s is not created. Restore schema first or create missing tables manually", strings.Join(missingTables, ", "))
 	}
 	restoreBackupWorkingGroup, restoreCtx := errgroup.WithContext(ctx)
-	restoreBackupWorkingGroup.SetLimit(max(b.cfg.ClickHouse.MaxConnections,1))
+	restoreBackupWorkingGroup.SetLimit(max(b.cfg.ClickHouse.MaxConnections, 1))
 
 	for i := range tablesForRestore {
 		tableRestoreStartTime := time.Now()
 		table := tablesForRestore[i]
-		// need mapped database path and original table.Database for HardlinkBackupPartsToStorage
+		// need mapped database path and original table.Database for HardlinkBackupPartsToStorage.
 		dstDatabase := table.Database
+		// The same goes for the table
+		dstTableName := table.Table
 		if len(b.cfg.General.RestoreDatabaseMapping) > 0 {
 			if targetDB, isMapped := b.cfg.General.RestoreDatabaseMapping[table.Database]; isMapped {
 				dstDatabase = targetDB
 				tablesForRestore[i].Database = targetDB
 			}
 		}
-		log := log.WithField("table", fmt.Sprintf("%s.%s", dstDatabase, table.Table))
+		// https://github.com/Altinity/clickhouse-backup/issues/937
+		if len(b.cfg.General.RestoreTableMapping) > 0 {
+			if targetTable, isMapped := b.cfg.General.RestoreTableMapping[table.Table]; isMapped {
+				dstTableName = targetTable
+				tablesForRestore[i].Table = targetTable
+			}
+		}
+		log := log.WithField("table", fmt.Sprintf("%s.%s", dstDatabase, dstTableName))
 		dstTable, ok := dstTablesMap[metadata.TableTitle{
 			Database: dstDatabase,
-			Table:    table.Table}]
+			Table:    dstTableName}]
 		if !ok {
 			return fmt.Errorf("can't find '%s.%s' in current system.tables", dstDatabase, table.Table)
 		}
@@ -1458,14 +1494,20 @@ func (b *Backuper) checkMissingTables(tablesForRestore ListOfTables, chTables []
 	var missingTables []string
 	for _, table := range tablesForRestore {
 		dstDatabase := table.Database
+		dstTable := table.Table
 		if len(b.cfg.General.RestoreDatabaseMapping) > 0 {
 			if targetDB, isMapped := b.cfg.General.RestoreDatabaseMapping[table.Database]; isMapped {
 				dstDatabase = targetDB
 			}
 		}
+		if len(b.cfg.General.RestoreTableMapping) > 0 {
+			if targetTable, isMapped := b.cfg.General.RestoreTableMapping[table.Table]; isMapped {
+				dstTable = targetTable
+			}
+		}
 		found := false
 		for _, chTable := range chTables {
-			if (dstDatabase == chTable.Database) && (table.Table == chTable.Name) {
+			if (dstDatabase == chTable.Database) && (dstTable == chTable.Name) {
 				found = true
 				break
 			}
@@ -1488,22 +1530,31 @@ func (b *Backuper) prepareDstTablesMap(chTables []clickhouse.Table) map[metadata
 	return dstTablesMap
 }
 
-func (b *Backuper) changeTablePatternFromRestoreDatabaseMapping(tablePattern string) string {
-	for sourceDb, targetDb := range b.cfg.General.RestoreDatabaseMapping {
+func (b *Backuper) changeTablePatternFromRestoreMapping(tablePattern, objType string) string {
+	var mapping map[string]string
+	switch objType {
+	case "database":
+		mapping = b.cfg.General.RestoreDatabaseMapping
+	case "table":
+		mapping = b.cfg.General.RestoreDatabaseMapping
+	default:
+		return ""
+	}
+	for sourceObj, targetObj := range mapping {
 		if tablePattern != "" {
-			sourceDbRE := regexp.MustCompile(fmt.Sprintf("(^%s.*)|(,%s.*)", sourceDb, sourceDb))
-			if sourceDbRE.MatchString(tablePattern) {
-				matches := sourceDbRE.FindAllStringSubmatch(tablePattern, -1)
-				substitution := targetDb + ".*"
+			sourceObjRE := regexp.MustCompile(fmt.Sprintf("(^%s.*)|(,%s.*)", sourceObj, sourceObj))
+			if sourceObjRE.MatchString(tablePattern) {
+				matches := sourceObjRE.FindAllStringSubmatch(tablePattern, -1)
+				substitution := targetObj + ".*"
 				if strings.HasPrefix(matches[0][1], ",") {
 					substitution = "," + substitution
 				}
-				tablePattern = sourceDbRE.ReplaceAllString(tablePattern, substitution)
+				tablePattern = sourceObjRE.ReplaceAllString(tablePattern, substitution)
 			} else {
-				tablePattern += "," + targetDb + ".*"
+				tablePattern += "," + targetObj + ".*"
 			}
 		} else {
-			tablePattern += targetDb + ".*"
+			tablePattern += targetObj + ".*"
 		}
 	}
 	return tablePattern
diff --git a/pkg/backup/restore_remote.go b/pkg/backup/restore_remote.go
--- a/pkg/backup/restore_remote.go
+++ b/pkg/backup/restore_remote.go
@@ -2,12 +2,12 @@ package backup
 
 import "errors"
 
-func (b *Backuper) RestoreFromRemote(backupName, tablePattern string, databaseMapping, partitions []string, schemaOnly, dataOnly, dropExists, ignoreDependencies, restoreRBAC, rbacOnly, restoreConfigs, configsOnly, resume bool, version string, commandId int) error {
+func (b *Backuper) RestoreFromRemote(backupName, tablePattern string, databaseMapping, tableMapping, partitions []string, schemaOnly, dataOnly, dropExists, ignoreDependencies, restoreRBAC, rbacOnly, restoreConfigs, configsOnly, resume bool, version string, commandId int) error {
 	if err := b.Download(backupName, tablePattern, partitions, schemaOnly, resume, version, commandId); err != nil {
 		// https://github.com/Altinity/clickhouse-backup/issues/625
 		if !errors.Is(err, ErrBackupIsAlreadyExists) {
 			return err
 		}
 	}
-	return b.Restore(backupName, tablePattern, databaseMapping, partitions, schemaOnly, dataOnly, dropExists, ignoreDependencies, restoreRBAC, rbacOnly, restoreConfigs, configsOnly, version, commandId)
+	return b.Restore(backupName, tablePattern, databaseMapping, tableMapping, partitions, schemaOnly, dataOnly, dropExists, ignoreDependencies, restoreRBAC, rbacOnly, restoreConfigs, configsOnly, version, commandId)
 }
diff --git a/pkg/backup/table_pattern.go b/pkg/backup/table_pattern.go
--- a/pkg/backup/table_pattern.go
+++ b/pkg/backup/table_pattern.go
@@ -4,10 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/partition"
-	apexLog "github.com/apex/log"
-	"github.com/google/uuid"
 	"io"
 	"net/url"
 	"os"
@@ -17,10 +13,14 @@ import (
 	"sort"
 	"strings"
 
+	apexLog "github.com/apex/log"
+	"github.com/google/uuid"
+
 	"github.com/Altinity/clickhouse-backup/v2/pkg/common"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/filesystemhelper"
-
 	"github.com/Altinity/clickhouse-backup/v2/pkg/metadata"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/partition"
 )
 
 type ListOfTables []metadata.TableMetadata
@@ -295,13 +295,13 @@ func (b *Backuper) enrichTablePatternsByInnerDependencies(metadataPath string, t
 	return tablePatterns, nil
 }
 
-var queryRE = regexp.MustCompile(`(?m)^(CREATE|ATTACH) (TABLE|VIEW|LIVE VIEW|MATERIALIZED VIEW|DICTIONARY|FUNCTION) (\x60?)([^\s\x60.]*)(\x60?)\.\x60?([^\s\x60.]*)\x60?( UUID '[^']+')?(?:( TO )(\x60?)([^\s\x60.]*)(\x60?)(\.))?(?:(.+FROM )(\x60?)([^\s\x60.]*)(\x60?)(\.))?`)
+var queryRE = regexp.MustCompile(`(?m)^(CREATE|ATTACH) (TABLE|VIEW|LIVE VIEW|MATERIALIZED VIEW|DICTIONARY|FUNCTION) (\x60?)([^\s\x60.]*)(\x60?)\.\x60?([^\s\x60.]*)\x60?( UUID '[^']+')?(?:( TO )(\x60?)([^\s\x60.]*)(\x60?)(\.)(\x60?)([^\s\x60.]*)(\x60?))?(?:(.+FROM )(\x60?)([^\s\x60.]*)(\x60?)(\.)(\x60?)([^\s\x60.]*)(\x60?))?`)
 var createOrAttachRE = regexp.MustCompile(`(?m)^(CREATE|ATTACH)`)
 var uuidRE = regexp.MustCompile(`UUID '([a-f\d\-]+)'`)
 
 var usualIdentifier = regexp.MustCompile(`^[a-zA-Z0-9_]+$`)
 var replicatedRE = regexp.MustCompile(`(Replicated[a-zA-Z]*MergeTree)\('([^']+)'([^)]+)\)`)
-var distributedRE = regexp.MustCompile(`(Distributed)\(([^,]+),([^,]+),([^)]+)\)`)
+var distributedRE = regexp.MustCompile(`(Distributed)\(([^,]+),([^,]+),([^,]+)([^)]+)\)`)
 
 func changeTableQueryToAdjustDatabaseMapping(originTables *ListOfTables, dbMapRule map[string]string) error {
 	for i := 0; i < len(*originTables); i++ {
@@ -329,9 +329,9 @@ func changeTableQueryToAdjustDatabaseMapping(originTables *ListOfTables, dbMapRu
 					createTargetDb = "`" + createTargetDb + "`"
 				}
 				toClauseTargetDb := setMatchedDb(matches[0][10])
-				fromClauseTargetDb := setMatchedDb(matches[0][15])
+				fromClauseTargetDb := setMatchedDb(matches[0][18])
 				// matching CREATE|ATTACH ... TO .. SELECT ... FROM ... command
-				substitution = fmt.Sprintf("${1} ${2} ${3}%v${5}.${6}${7}${8}${9}%v${11}${12}${13}${14}%v${16}${17}", createTargetDb, toClauseTargetDb, fromClauseTargetDb)
+				substitution = fmt.Sprintf("${1} ${2} ${3}%v${5}.${6}${7}${8}${9}%v${11}${12}${13}${14}${15}${16}${17}%v${19}${20}${21}${22}${23}", createTargetDb, toClauseTargetDb, fromClauseTargetDb)
 			} else {
 				if originTable.Query == "" {
 					continue
@@ -360,7 +360,7 @@ func changeTableQueryToAdjustDatabaseMapping(originTables *ListOfTables, dbMapRu
 				underlyingDB := matches[0][3]
 				underlyingDBClean := strings.NewReplacer(" ", "", "'", "").Replace(underlyingDB)
 				if underlyingTargetDB, isUnderlyingMapped := dbMapRule[underlyingDBClean]; isUnderlyingMapped {
-					substitution = fmt.Sprintf("${1}(${2},%s,${4})", strings.Replace(underlyingDB, underlyingDBClean, underlyingTargetDB, 1))
+					substitution = fmt.Sprintf("${1}(${2},%s,${4}${5})", strings.Replace(underlyingDB, underlyingDBClean, underlyingTargetDB, 1))
 					originTable.Query = distributedRE.ReplaceAllString(originTable.Query, substitution)
 				}
 			}
@@ -371,6 +371,74 @@ func changeTableQueryToAdjustDatabaseMapping(originTables *ListOfTables, dbMapRu
 	return nil
 }
 
+func changeTableQueryToAdjustTableMapping(originTables *ListOfTables, tableMapRule map[string]string) error {
+	for i := 0; i < len(*originTables); i++ {
+		originTable := (*originTables)[i]
+		if targetTable, isMapped := tableMapRule[originTable.Table]; isMapped {
+			// substitute table in the table create query
+			var substitution string
+
+			if createOrAttachRE.MatchString(originTable.Query) {
+				matches := queryRE.FindAllStringSubmatch(originTable.Query, -1)
+				if matches[0][6] != originTable.Table {
+					return fmt.Errorf("invalid SQL: %s for restore-table-mapping[%s]=%s", originTable.Query, originTable.Table, targetTable)
+				}
+				setMatchedDb := func(clauseTargetTable string) string {
+					if clauseMappedTable, isClauseMapped := tableMapRule[clauseTargetTable]; isClauseMapped {
+						clauseTargetTable = clauseMappedTable
+						if !usualIdentifier.MatchString(clauseTargetTable) {
+							clauseTargetTable = "`" + clauseTargetTable + "`"
+						}
+					}
+					return clauseTargetTable
+				}
+				createTargetTable := targetTable
+				if !usualIdentifier.MatchString(createTargetTable) {
+					createTargetTable = "`" + createTargetTable + "`"
+				}
+				toClauseTargetTable := setMatchedDb(matches[0][14])
+				fromClauseTargetTable := setMatchedDb(matches[0][22])
+				// matching CREATE|ATTACH ... TO .. SELECT ... FROM ... command
+				substitution = fmt.Sprintf("${1} ${2} ${3}${4}${5}.%v${7}${8}${9}${10}${11}${12}${13}%v${15}${16}${17}${18}${19}${20}${21}%v${23}", createTargetTable, toClauseTargetTable, fromClauseTargetTable)
+			} else {
+				if originTable.Query == "" {
+					continue
+				}
+				return fmt.Errorf("error when try to replace table `%s` to `%s` in query: %s", originTable.Table, targetTable, originTable.Query)
+			}
+			originTable.Query = queryRE.ReplaceAllString(originTable.Query, substitution)
+			if uuidRE.MatchString(originTable.Query) {
+				newUUID, _ := uuid.NewUUID()
+				substitution = fmt.Sprintf("UUID '%s'", newUUID.String())
+				originTable.Query = uuidRE.ReplaceAllString(originTable.Query, substitution)
+			}
+			// https://github.com/Altinity/clickhouse-backup/issues/547
+			if replicatedRE.MatchString(originTable.Query) {
+				matches := replicatedRE.FindAllStringSubmatch(originTable.Query, -1)
+				originPath := matches[0][2]
+				tableReplicatedPattern := "/" + originTable.Table
+				if strings.Contains(originPath, tableReplicatedPattern) {
+					substitution = fmt.Sprintf("${1}('%s'${3})", strings.Replace(originPath, tableReplicatedPattern, "/"+targetTable, 1))
+					originTable.Query = replicatedRE.ReplaceAllString(originTable.Query, substitution)
+				}
+			}
+			// https://github.com/Altinity/clickhouse-backup/issues/547
+			if distributedRE.MatchString(originTable.Query) {
+				matches := distributedRE.FindAllStringSubmatch(originTable.Query, -1)
+				underlyingTable := matches[0][4]
+				underlyingTableClean := strings.NewReplacer(" ", "", "'", "").Replace(underlyingTable)
+				if underlyingTargetTable, isUnderlyingMapped := tableMapRule[underlyingTableClean]; isUnderlyingMapped {
+					substitution = fmt.Sprintf("${1}(${2},${3},%s${5})", strings.Replace(underlyingTable, underlyingTableClean, underlyingTargetTable, 1))
+					originTable.Query = distributedRE.ReplaceAllString(originTable.Query, substitution)
+				}
+			}
+			originTable.Table = targetTable
+			(*originTables)[i] = originTable
+		}
+	}
+	return nil
+}
+
 func filterPartsAndFilesByPartitionsFilter(tableMetadata metadata.TableMetadata, partitionsFilter common.EmptyMap) {
 	if len(partitionsFilter) > 0 {
 		for disk, parts := range tableMetadata.Parts {
diff --git a/pkg/clickhouse/clickhouse.go b/pkg/clickhouse/clickhouse.go
--- a/pkg/clickhouse/clickhouse.go
+++ b/pkg/clickhouse/clickhouse.go
@@ -15,14 +15,15 @@ import (
 	"strings"
 	"time"
 
-	"github.com/Altinity/clickhouse-backup/v2/pkg/common"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/metadata"
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
 	"github.com/antchfx/xmlquery"
 	apexLog "github.com/apex/log"
 	"github.com/ricochet2200/go-disk-usage/du"
+
+	"github.com/Altinity/clickhouse-backup/v2/pkg/common"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/metadata"
 )
 
 // ClickHouse - provide
@@ -751,6 +752,9 @@ func (ch *ClickHouse) AttachDataParts(table metadata.TableMetadata, dstTable Tab
 	if dstTable.Database != "" && dstTable.Database != table.Database {
 		table.Database = dstTable.Database
 	}
+	if dstTable.Name != "" && dstTable.Name != table.Table {
+		table.Table = dstTable.Name
+	}
 	canContinue, err := ch.CheckReplicationInProgress(table)
 	if err != nil {
 		return err
@@ -784,6 +788,9 @@ func (ch *ClickHouse) AttachTable(ctx context.Context, table metadata.TableMetad
 	if dstTable.Database != "" && dstTable.Database != table.Database {
 		table.Database = dstTable.Database
 	}
+	if dstTable.Name != "" && dstTable.Name != table.Table {
+		table.Table = dstTable.Name
+	}
 	canContinue, err := ch.CheckReplicationInProgress(table)
 	if err != nil {
 		return err
diff --git a/pkg/config/config.go b/pkg/config/config.go
--- a/pkg/config/config.go
+++ b/pkg/config/config.go
@@ -10,9 +10,8 @@ import (
 	"strings"
 	"time"
 
-	s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
-
 	"github.com/apex/log"
+	s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
 	"github.com/kelseyhightower/envconfig"
 	"github.com/urfave/cli"
 	"gopkg.in/yaml.v3"
@@ -54,6 +53,7 @@ type GeneralConfig struct {
 	UploadByPart                        bool              `yaml:"upload_by_part" envconfig:"UPLOAD_BY_PART"`
 	DownloadByPart                      bool              `yaml:"download_by_part" envconfig:"DOWNLOAD_BY_PART"`
 	RestoreDatabaseMapping              map[string]string `yaml:"restore_database_mapping" envconfig:"RESTORE_DATABASE_MAPPING"`
+	RestoreTableMapping                 map[string]string `yaml:"restore_table_mapping" envconfig:"RESTORE_TABLE_MAPPING"`
 	RetriesOnFailure                    int               `yaml:"retries_on_failure" envconfig:"RETRIES_ON_FAILURE"`
 	RetriesPause                        string            `yaml:"retries_pause" envconfig:"RETRIES_PAUSE"`
 	WatchInterval                       string            `yaml:"watch_interval" envconfig:"WATCH_INTERVAL"`
@@ -529,6 +529,7 @@ func DefaultConfig() *Config {
 			FullDuration:                        24 * time.Hour,
 			WatchBackupNameTemplate:             "shard{shard}-{type}-{time:20060102150405}",
 			RestoreDatabaseMapping:              make(map[string]string, 0),
+			RestoreTableMapping:                 make(map[string]string, 0),
 			IONicePriority:                      "idle",
 			CPUNicePriority:                     15,
 			RBACBackupAlways:                    true,
diff --git a/pkg/server/server.go b/pkg/server/server.go
--- a/pkg/server/server.go
+++ b/pkg/server/server.go
@@ -22,6 +22,12 @@ import (
 	"syscall"
 	"time"
 
+	apexLog "github.com/apex/log"
+	"github.com/google/shlex"
+	"github.com/gorilla/mux"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/urfave/cli"
+
 	"github.com/Altinity/clickhouse-backup/v2/pkg/backup"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/clickhouse"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/common"
@@ -30,12 +36,6 @@ import (
 	"github.com/Altinity/clickhouse-backup/v2/pkg/server/metrics"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/status"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/utils"
-
-	apexLog "github.com/apex/log"
-	"github.com/google/shlex"
-	"github.com/gorilla/mux"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
-	"github.com/urfave/cli"
 )
 
 type APIServer struct {
@@ -1200,6 +1200,7 @@ func (api *APIServer) httpUploadHandler(w http.ResponseWriter, r *http.Request)
 }
 
 var databaseMappingRE = regexp.MustCompile(`[\w+]:[\w+]`)
+var tableMappingRE = regexp.MustCompile(`[\w+]:[\w+]`)
 
 // httpRestoreHandler - restore a backup from local storage
 func (api *APIServer) httpRestoreHandler(w http.ResponseWriter, r *http.Request) {
@@ -1215,6 +1216,7 @@ func (api *APIServer) httpRestoreHandler(w http.ResponseWriter, r *http.Request)
 	vars := mux.Vars(r)
 	tablePattern := ""
 	databaseMappingToRestore := make([]string, 0)
+	tableMappingToRestore := make([]string, 0)
 	partitionsToBackup := make([]string, 0)
 	schemaOnly := false
 	dataOnly := false
@@ -1244,6 +1246,24 @@ func (api *APIServer) httpRestoreHandler(w http.ResponseWriter, r *http.Request)
 
 		fullCommand = fmt.Sprintf("%s --restore-database-mapping=\"%s\"", fullCommand, strings.Join(databaseMappingToRestore, ","))
 	}
+
+	// https://github.com/Altinity/clickhouse-backup/issues/937
+	if tableMappingQuery, exist := query["restore_table_mapping"]; exist {
+		for _, tableMapping := range tableMappingQuery {
+			mappingItems := strings.Split(tableMapping, ",")
+			for _, m := range mappingItems {
+				if strings.Count(m, ":") != 1 || !tableMappingRE.MatchString(m) {
+					api.writeError(w, http.StatusInternalServerError, "restore", fmt.Errorf("invalid values in restore_table_mapping %s", m))
+					return
+
+				}
+			}
+			tableMappingToRestore = append(tableMappingToRestore, mappingItems...)
+		}
+
+		fullCommand = fmt.Sprintf("%s --restore-table-mapping=\"%s\"", fullCommand, strings.Join(tableMappingToRestore, ","))
+	}
+
 	if partitions, exist := query["partitions"]; exist {
 		partitionsToBackup = append(partitionsToBackup, partitions...)
 		fullCommand = fmt.Sprintf("%s --partitions=\"%s\"", fullCommand, strings.Join(partitions, "\" --partitions=\""))
@@ -1291,7 +1311,7 @@ func (api *APIServer) httpRestoreHandler(w http.ResponseWriter, r *http.Request)
 	go func() {
 		err, _ := api.metrics.ExecuteWithMetrics("restore", 0, func() error {
 			b := backup.NewBackuper(api.config)
-			return b.Restore(name, tablePattern, databaseMappingToRestore, partitionsToBackup, schemaOnly, dataOnly, dropExists, ignoreDependencies, restoreRBAC, false, restoreConfigs, false, api.cliApp.Version, commandId)
+			return b.Restore(name, tablePattern, databaseMappingToRestore, tableMappingToRestore, partitionsToBackup, schemaOnly, dataOnly, dropExists, ignoreDependencies, restoreRBAC, false, restoreConfigs, false, api.cliApp.Version, commandId)
 		})
 		status.Current.Stop(commandId, err)
 		if err != nil {


Test patch:
diff --git a/test/integration/integration_test.go b/test/integration/integration_test.go
--- a/test/integration/integration_test.go
+++ b/test/integration/integration_test.go
@@ -7,12 +7,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/logcli"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/partition"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/status"
-	"github.com/Altinity/clickhouse-backup/v2/pkg/utils"
-	"github.com/google/uuid"
 	"math/rand"
 	"os"
 	"os/exec"
@@ -24,13 +18,19 @@ import (
 	"testing"
 	"time"
 
-	"github.com/Altinity/clickhouse-backup/v2/pkg/clickhouse"
-	"github.com/apex/log"
-	"golang.org/x/mod/semver"
-
 	_ "github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/apex/log"
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/mod/semver"
+
+	"github.com/Altinity/clickhouse-backup/v2/pkg/clickhouse"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/logcli"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/partition"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/status"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/utils"
 )
 
 const dbNameAtomic = "_test#$.ДБ_atomic_"
@@ -896,10 +896,10 @@ func testAPIBackupClean(r *require.Assertions, ch *TestClickHouse) {
 	r.NotContains(out, "another operation is currently running")
 	r.NotContains(out, "\"status\":\"error\"")
 
-	runClickHouseClientInsertSystemBackupActions(r, ch, []string{"clean","clean_remote_broken"}, false)
+	runClickHouseClientInsertSystemBackupActions(r, ch, []string{"clean", "clean_remote_broken"}, false)
 }
 
-	func testAPIMetrics(r *require.Assertions, ch *TestClickHouse) {
+func testAPIMetrics(r *require.Assertions, ch *TestClickHouse) {
 	log.Info("Check /metrics clickhouse_backup_last_backup_size_remote")
 	var lastRemoteSize int64
 	r.NoError(ch.chbackend.SelectSingleRowNoCtx(&lastRemoteSize, "SELECT size FROM system.backup_list WHERE name='z_backup_5' AND location='remote'"))
@@ -2184,7 +2184,7 @@ func TestIntegrationEmbedded(t *testing.T) {
 	}
 }
 
-func TestRestoreDatabaseMapping(t *testing.T) {
+func TestRestoreMapping(t *testing.T) {
 	//t.Parallel()
 	r := require.New(t)
 	ch := &TestClickHouse{}
@@ -2220,7 +2220,7 @@ func TestRestoreDatabaseMapping(t *testing.T) {
 	r.NoError(dockerExec("clickhouse-backup", "clickhouse-backup", "-c", "/etc/clickhouse-backup/config-database-mapping.yml", "create", testBackupName))
 
 	log.Info("Restore schema")
-	r.NoError(dockerExec("clickhouse-backup", "clickhouse-backup", "-c", "/etc/clickhouse-backup/config-database-mapping.yml", "restore", "--schema", "--rm", "--restore-database-mapping", "database1:database-2", "--tables", "database1.*", testBackupName))
+	r.NoError(dockerExec("clickhouse-backup", "clickhouse-backup", "-c", "/etc/clickhouse-backup/config-database-mapping.yml", "restore", "--schema", "--rm", "--restore-database-mapping", "database1:database-2", "--restore-table-mapping", "t1:t3,t2:t4,d1:d2,mv1:mv2,v1:v2", "--tables", "database1.*", testBackupName))
 
 	log.Info("Check result database1")
 	ch.queryWithNoError(r, "INSERT INTO database1.t1 SELECT '2023-01-01 00:00:00', number FROM numbers(10)")
@@ -2233,13 +2233,13 @@ func TestRestoreDatabaseMapping(t *testing.T) {
 	r.NoError(ch.dropDatabase("database1"))
 
 	log.Info("Restore data")
-	r.NoError(dockerExec("clickhouse-backup", "clickhouse-backup", "-c", "/etc/clickhouse-backup/config-database-mapping.yml", "restore", "--data", "--restore-database-mapping", "database1:database-2", "--tables", "database1.*", testBackupName))
+	r.NoError(dockerExec("clickhouse-backup", "clickhouse-backup", "-c", "/etc/clickhouse-backup/config-database-mapping.yml", "restore", "--data", "--restore-database-mapping", "database1:database-2", "--restore-table-mapping", "t1:t3,t2:t4,d1:d2,mv1:mv2,v1:v2", "--tables", "database1.*", testBackupName))
 
 	log.Info("Check result database-2")
-	checkRecordset(1, 10, "SELECT count() FROM `database-2`.t1")
-	checkRecordset(1, 10, "SELECT count() FROM `database-2`.d1")
-	checkRecordset(1, 10, "SELECT count() FROM `database-2`.mv1")
-	checkRecordset(1, 10, "SELECT count() FROM `database-2`.v1")
+	checkRecordset(1, 10, "SELECT count() FROM `database-2`.t3")
+	checkRecordset(1, 10, "SELECT count() FROM `database-2`.d2")
+	checkRecordset(1, 10, "SELECT count() FROM `database-2`.mv2")
+	checkRecordset(1, 10, "SELECT count() FROM `database-2`.v2")
 
 	log.Info("Check database1 not exists")
 	checkRecordset(1, 0, "SELECT count() FROM system.databases WHERE name='database1'")
@@ -2793,7 +2793,7 @@ func generateTestDataWithDifferentStoragePolicy(remoteStorageType string, offset
 			}
 		}
 		//s3 disks support after 21.8
-		if compareVersion(os.Getenv("CLICKHOUSE_VERSION"), "21.8") >= 0 && strings.Contains(remoteStorageType,"S3") {
+		if compareVersion(os.Getenv("CLICKHOUSE_VERSION"), "21.8") >= 0 && strings.Contains(remoteStorageType, "S3") {
 			testDataWithStoragePolicy.Name = "test_s3"
 			testDataWithStoragePolicy.Schema = "(id UInt64) Engine=ReplicatedMergeTree('/clickhouse/tables/{cluster}/{shard}/{database}/{table}','{replica}') ORDER BY id PARTITION BY id SETTINGS storage_policy = 's3_only'"
 			addTestDataIfNotExists()
@@ -2805,7 +2805,7 @@ func generateTestDataWithDifferentStoragePolicy(remoteStorageType string, offset
 			addTestDataIfNotExists()
 		}
 		//encrypted s3 disks support after 21.12
-		if compareVersion(os.Getenv("CLICKHOUSE_VERSION"), "21.12") >= 0 && strings.Contains(remoteStorageType,"S3") {
+		if compareVersion(os.Getenv("CLICKHOUSE_VERSION"), "21.12") >= 0 && strings.Contains(remoteStorageType, "S3") {
 			testDataWithStoragePolicy.Name = "test_s3_encrypted"
 			testDataWithStoragePolicy.Schema = "(id UInt64) Engine=MergeTree ORDER BY id PARTITION BY id SETTINGS storage_policy = 's3_only_encrypted'"
 			// @todo wait when fix https://github.com/ClickHouse/ClickHouse/issues/58247
@@ -2821,7 +2821,7 @@ func generateTestDataWithDifferentStoragePolicy(remoteStorageType string, offset
 			addTestDataIfNotExists()
 		}
 		//check azure_blob_storage only in 23.3+ (added in 22.1)
-		if compareVersion(os.Getenv("CLICKHOUSE_VERSION"), "23.3") >= 0 && strings.Contains(remoteStorageType,"AZBLOB") {
+		if compareVersion(os.Getenv("CLICKHOUSE_VERSION"), "23.3") >= 0 && strings.Contains(remoteStorageType, "AZBLOB") {
 			testDataWithStoragePolicy.Name = "test_azure"
 			testDataWithStoragePolicy.Schema = "(id UInt64) Engine=ReplicatedMergeTree('/clickhouse/tables/{cluster}/{shard}/{database}/{table}','{replica}') ORDER BY id PARTITION BY id SETTINGS storage_policy = 'azure_only'"
 			addTestDataIfNotExists()
diff --git a/test/testflows/clickhouse_backup/tests/snapshots/cli.py.cli.snapshot b/test/testflows/clickhouse_backup/tests/snapshots/cli.py.cli.snapshot
--- a/test/testflows/clickhouse_backup/tests/snapshots/cli.py.cli.snapshot
+++ b/test/testflows/clickhouse_backup/tests/snapshots/cli.py.cli.snapshot
@@ -1,4 +1,4 @@
-default_config = r"""'[\'general:\', \'    remote_storage: none\', \'    backups_to_keep_local: 0\', \'    backups_to_keep_remote: 0\', \'    log_level: info\', \'    allow_empty_backups: false\', \'    use_resumable_state: true\', \'    restore_schema_on_cluster: ""\', \'    upload_by_part: true\', \'    download_by_part: true\', \'    restore_database_mapping: {}\', \'    retries_on_failure: 3\', \'    retries_pause: 30s\', \'    watch_interval: 1h\', \'    full_interval: 24h\', \'    watch_backup_name_template: shard{shard}-{type}-{time:20060102150405}\', \'    sharded_operation_mode: ""\', \'    cpu_nice_priority: 15\', \'    io_nice_priority: idle\', \'    rbac_backup_always: true\', \'    rbac_conflict_resolution: recreate\', \'    retriesduration: 100ms\', \'    watchduration: 1h0m0s\', \'    fullduration: 24h0m0s\', \'clickhouse:\', \'    username: default\', \'    password: ""\', \'    host: localhost\', \'    port: 9000\', \'    disk_mapping: {}\', \'    skip_tables:\', \'        - system.*\', \'        - INFORMATION_SCHEMA.*\', \'        - information_schema.*\', \'        - _temporary_and_external_tables.*\', \'    skip_table_engines: []\', \'    timeout: 30m\', \'    freeze_by_part: false\', \'    freeze_by_part_where: ""\', \'    use_embedded_backup_restore: false\', \'    embedded_backup_disk: ""\', \'    backup_mutations: true\', \'    restore_as_attach: false\', \'    check_parts_columns: true\', \'    secure: false\', \'    skip_verify: false\', \'    sync_replicated_tables: false\', \'    log_sql_queries: true\', \'    config_dir: /etc/clickhouse-server/\', \'    restart_command: exec:systemctl restart clickhouse-server\', \'    ignore_not_exists_error_during_freeze: true\', \'    check_replicas_before_attach: true\', \'    tls_key: ""\', \'    tls_cert: ""\', \'    tls_ca: ""\', \'    debug: false\', \'s3:\', \'    access_key: ""\', \'    secret_key: ""\', \'    bucket: ""\', \'    endpoint: ""\', \'    region: us-east-1\', \'    acl: private\', \'    assume_role_arn: ""\', \'    force_path_style: false\', \'    path: ""\', \'    object_disk_path: ""\', \'    disable_ssl: false\', \'    compression_level: 1\', \'    compression_format: tar\', \'    sse: ""\', \'    sse_kms_key_id: ""\', \'    sse_customer_algorithm: ""\', \'    sse_customer_key: ""\', \'    sse_customer_key_md5: ""\', \'    sse_kms_encryption_context: ""\', \'    disable_cert_verification: false\', \'    use_custom_storage_class: false\', \'    storage_class: STANDARD\', \'    custom_storage_class_map: {}\', \'    part_size: 0\', \'    allow_multipart_download: false\', \'    object_labels: {}\', \'    request_payer: ""\', \'    check_sum_algorithm: ""\', \'    debug: false\', \'gcs:\', \'    credentials_file: ""\', \'    credentials_json: ""\', \'    credentials_json_encoded: ""\', \'    embedded_access_key: ""\', \'    embedded_secret_key: ""\', \'    skip_credentials: false\', \'    bucket: ""\', \'    path: ""\', \'    object_disk_path: ""\', \'    compression_level: 1\', \'    compression_format: tar\', \'    debug: false\', \'    force_http: false\', \'    endpoint: ""\', \'    storage_class: STANDARD\', \'    object_labels: {}\', \'    custom_storage_class_map: {}\', \'    chunk_size: 0\', \'cos:\', \'    url: ""\', \'    timeout: 2m\', \'    secret_id: ""\', \'    secret_key: ""\', \'    path: ""\', \'    compression_format: tar\', \'    compression_level: 1\', \'    debug: false\', \'api:\', \'    listen: localhost:7171\', \'    enable_metrics: true\', \'    enable_pprof: false\', \'    username: ""\', \'    password: ""\', \'    secure: false\', \'    certificate_file: ""\', \'    private_key_file: ""\', \'    ca_cert_file: ""\', \'    ca_key_file: ""\', \'    create_integration_tables: false\', \'    integration_tables_host: ""\', \'    allow_parallel: false\', \'    complete_resumable_after_restart: true\', \'    watch_is_main_process: false\', \'ftp:\', \'    address: ""\', \'    timeout: 2m\', \'    username: ""\', \'    password: ""\', \'    tls: false\', \'    skip_tls_verify: false\', \'    path: ""\', \'    object_disk_path: ""\', \'    compression_format: tar\', \'    compression_level: 1\', \'    debug: false\', \'sftp:\', \'    address: ""\', \'    port: 22\', \'    username: ""\', \'    password: ""\', \'    key: ""\', \'    path: ""\', \'    object_disk_path: ""\', \'    compression_format: tar\', \'    compression_level: 1\', \'    debug: false\', \'azblob:\', \'    endpoint_schema: https\', \'    endpoint_suffix: core.windows.net\', \'    account_name: ""\', \'    account_key: ""\', \'    sas: ""\', \'    use_managed_identity: false\', \'    container: ""\', \'    path: ""\', \'    object_disk_path: ""\', \'    compression_level: 1\', \'    compression_format: tar\', \'    sse_key: ""\', \'    buffer_size: 0\', \'    buffer_count: 3\', \'    timeout: 4h\', \'    debug: false\', \'custom:\', \'    upload_command: ""\', \'    download_command: ""\', \'    list_command: ""\', \'    delete_command: ""\', \'    command_timeout: 4h\', \'    commandtimeoutduration: 4h0m0s\']'"""
+default_config = r"""'[\'general:\', \'    remote_storage: none\', \'    backups_to_keep_local: 0\', \'    backups_to_keep_remote: 0\', \'    log_level: info\', \'    allow_empty_backups: false\', \'    use_resumable_state: true\', \'    restore_schema_on_cluster: ""\', \'    upload_by_part: true\', \'    download_by_part: true\', \'    restore_database_mapping: {}\', \'    restore_table_mapping: {}\', \'    retries_on_failure: 3\', \'    retries_pause: 30s\', \'    watch_interval: 1h\', \'    full_interval: 24h\', \'    watch_backup_name_template: shard{shard}-{type}-{time:20060102150405}\', \'    sharded_operation_mode: ""\', \'    cpu_nice_priority: 15\', \'    io_nice_priority: idle\', \'    rbac_backup_always: true\', \'    rbac_conflict_resolution: recreate\', \'    retriesduration: 100ms\', \'    watchduration: 1h0m0s\', \'    fullduration: 24h0m0s\', \'clickhouse:\', \'    username: default\', \'    password: ""\', \'    host: localhost\', \'    port: 9000\', \'    disk_mapping: {}\', \'    skip_tables:\', \'        - system.*\', \'        - INFORMATION_SCHEMA.*\', \'        - information_schema.*\', \'        - _temporary_and_external_tables.*\', \'    skip_table_engines: []\', \'    timeout: 30m\', \'    freeze_by_part: false\', \'    freeze_by_part_where: ""\', \'    use_embedded_backup_restore: false\', \'    embedded_backup_disk: ""\', \'    backup_mutations: true\', \'    restore_as_attach: false\', \'    check_parts_columns: true\', \'    secure: false\', \'    skip_verify: false\', \'    sync_replicated_tables: false\', \'    log_sql_queries: true\', \'    config_dir: /etc/clickhouse-server/\', \'    restart_command: exec:systemctl restart clickhouse-server\', \'    ignore_not_exists_error_during_freeze: true\', \'    check_replicas_before_attach: true\', \'    tls_key: ""\', \'    tls_cert: ""\', \'    tls_ca: ""\', \'    debug: false\', \'s3:\', \'    access_key: ""\', \'    secret_key: ""\', \'    bucket: ""\', \'    endpoint: ""\', \'    region: us-east-1\', \'    acl: private\', \'    assume_role_arn: ""\', \'    force_path_style: false\', \'    path: ""\', \'    object_disk_path: ""\', \'    disable_ssl: false\', \'    compression_level: 1\', \'    compression_format: tar\', \'    sse: ""\', \'    sse_kms_key_id: ""\', \'    sse_customer_algorithm: ""\', \'    sse_customer_key: ""\', \'    sse_customer_key_md5: ""\', \'    sse_kms_encryption_context: ""\', \'    disable_cert_verification: false\', \'    use_custom_storage_class: false\', \'    storage_class: STANDARD\', \'    custom_storage_class_map: {}\', \'    part_size: 0\', \'    allow_multipart_download: false\', \'    object_labels: {}\', \'    request_payer: ""\', \'    check_sum_algorithm: ""\', \'    debug: false\', \'gcs:\', \'    credentials_file: ""\', \'    credentials_json: ""\', \'    credentials_json_encoded: ""\', \'    embedded_access_key: ""\', \'    embedded_secret_key: ""\', \'    skip_credentials: false\', \'    bucket: ""\', \'    path: ""\', \'    object_disk_path: ""\', \'    compression_level: 1\', \'    compression_format: tar\', \'    debug: false\', \'    force_http: false\', \'    endpoint: ""\', \'    storage_class: STANDARD\', \'    object_labels: {}\', \'    custom_storage_class_map: {}\', \'    chunk_size: 0\', \'cos:\', \'    url: ""\', \'    timeout: 2m\', \'    secret_id: ""\', \'    secret_key: ""\', \'    path: ""\', \'    compression_format: tar\', \'    compression_level: 1\', \'    debug: false\', \'api:\', \'    listen: localhost:7171\', \'    enable_metrics: true\', \'    enable_pprof: false\', \'    username: ""\', \'    password: ""\', \'    secure: false\', \'    certificate_file: ""\', \'    private_key_file: ""\', \'    ca_cert_file: ""\', \'    ca_key_file: ""\', \'    create_integration_tables: false\', \'    integration_tables_host: ""\', \'    allow_parallel: false\', \'    complete_resumable_after_restart: true\', \'    watch_is_main_process: false\', \'ftp:\', \'    address: ""\', \'    timeout: 2m\', \'    username: ""\', \'    password: ""\', \'    tls: false\', \'    skip_tls_verify: false\', \'    path: ""\', \'    object_disk_path: ""\', \'    compression_format: tar\', \'    compression_level: 1\', \'    debug: false\', \'sftp:\', \'    address: ""\', \'    port: 22\', \'    username: ""\', \'    password: ""\', \'    key: ""\', \'    path: ""\', \'    object_disk_path: ""\', \'    compression_format: tar\', \'    compression_level: 1\', \'    debug: false\', \'azblob:\', \'    endpoint_schema: https\', \'    endpoint_suffix: core.windows.net\', \'    account_name: ""\', \'    account_key: ""\', \'    sas: ""\', \'    use_managed_identity: false\', \'    container: ""\', \'    path: ""\', \'    object_disk_path: ""\', \'    compression_level: 1\', \'    compression_format: tar\', \'    sse_key: ""\', \'    buffer_size: 0\', \'    buffer_count: 3\', \'    timeout: 4h\', \'    debug: false\', \'custom:\', \'    upload_command: ""\', \'    download_command: ""\', \'    list_command: ""\', \'    delete_command: ""\', \'    command_timeout: 4h\', \'    commandtimeoutduration: 4h0m0s\']'"""
 
 help_flag = r"""'NAME:\n   clickhouse-backup - Tool for easy backup of ClickHouse with cloud supportUSAGE:\n   clickhouse-backup <command> [-t, --tables=<db>.<table>] <backup_name>DESCRIPTION:\n   Run as \'root\' or \'clickhouse\' userCOMMANDS:\n   tables               List of tables, exclude skip_tables\n   create               Create new backup\n   create_remote        Create and upload new backup\n   upload               Upload backup to remote storage\n   list                 List of backups\n   download             Download backup from remote storage\n   restore              Create schema and restore data from backup\n   restore_remote       Download and restore\n   delete               Delete specific backup\n   default-config       Print default config\n   print-config         Print current config merged with environment variables\n   clean                Remove data in \'shadow\' folder from all \'path\' folders available from \'system.disks\'\n   clean_remote_broken  Remove all broken remote backups\n   watch                Run infinite loop which create full + incremental backup sequence to allow efficient backup sequences\n   server               Run API server\n   help, h              Shows a list of commands or help for one commandGLOBAL OPTIONS:\n   --config value, -c value                   Config \'FILE\' name. (default: "/etc/clickhouse-backup/config.yml") [$CLICKHOUSE_BACKUP_CONFIG]\n   --environment-override value, --env value  override any environment variable via CLI parameter\n   --help, -h                                 show help\n   --version, -v                              print the version'"""