check system.clusters during restore engine=Distributed and is not exists and if not macros use CLICKHOUSE_RESTORE_DISTRIBUTED_CLUSTER config parameter as cluster value, fix #1252

Signed-off-by: Slach <[email protected]>

Author: Slach

ChangeLog.md

@@ -1,10 +1,14 @@
+# v2.6.38
+BUG FIXES
+- check `system.clusters` during restore `engine=Distributed` and is not exists and if not macros use `CLICKHOUSE_RESTORE_DISTRIBUTED_CLUSTER` config parameter as cluster value, fix [1252](https://github.com/Altinity/clickhouse-backup/issues/1252)
+
 # v2.6.37
 BUG FIXES
 - fix slowdown and corner cases, for incremental backup creation, affected 2.6.30+, fix https://github.com/Altinity/clickhouse-backup/issues/1249
 
 # v2.6.36
 IMPROVEMENTS
-- add `GCS_SA_EMAIL` option to support service account authorization, fix [1246](https://github.com/Altinity/clickhouse-backup/pull/1246) thanks @kamushadenes
+- add `GCS_SA_EMAIL` option to support service account authorization, fix [1173](https://github.com/Altinity/clickhouse-backup/issues/1173) thanks @kamushadenes
 - improve `--hardlink-exists-files` behavior, will look to exists local backups for the same part to avoid download unnecessary, fix [1244](https://github.com/Altinity/clickhouse-backup/issues/1244)
 - add `S3_CHUNK_SIZE` option to allow more flexible workload for S3-compatible remote storage, fix [1248](https://github.com/Altinity/clickhouse-backup/pull/1248) thanks @KimDoKy
 

ReadMe.md

@@ -215,6 +215,7 @@ clickhouse:
   embedded_backup_disk: ""  # CLICKHOUSE_EMBEDDED_BACKUP_DISK - disk from system.disks which will use when `use_embedded_backup_restore: true` 
   backup_mutations: true # CLICKHOUSE_BACKUP_MUTATIONS, allow backup mutations from system.mutations WHERE is_done=0 and apply it during restore
   restore_as_attach: false # CLICKHOUSE_RESTORE_AS_ATTACH, allow restore tables which have inconsistent data parts structure and mutations in progress
+  restore_distributed_cluster: "" # CLICKHOUSE_RESTORE_DISTRIBUTED_CLUSTER, cluster name (can use macros) which will use during restore `engine=Distributed` tables, when cluster defined in backup table definition not exists in `system.clusters`
   check_parts_columns: true # CLICKHOUSE_CHECK_PARTS_COLUMNS, check data types from system.parts_columns during create backup to guarantee mutation is complete
   max_connections: 0 # CLICKHOUSE_MAX_CONNECTIONS, how many parallel connections could be opened during operations
 azblob:

pkg/clickhouse/clickhouse.go

@@ -972,6 +972,7 @@ var attachViewRe = regexp.MustCompile(`(?im)^(ATTACH[\s\w]+VIEW[^(]+)(\s+AS\s+.+
 var createObjRe = regexp.MustCompile(`(?is)^(CREATE [^(]+)(\(.+)`)
 var onClusterRe = regexp.MustCompile(`(?im)\s+ON\s+CLUSTER\s+`)
 var distributedRE = regexp.MustCompile(`(Distributed)\(([^,]+),([^)]+)\)`)
+var macroRE = regexp.MustCompile(`(?i){[a-z0-9-_]+}`)
 
 // CreateTable - create ClickHouse table
 func (ch *ClickHouse) CreateTable(table Table, query string, dropTable, ignoreDependencies bool, onCluster string, version int, defaultDataPath string, asAttach bool, databaseEngine string) error {
@@ -1019,11 +1020,22 @@ func (ch *ClickHouse) CreateTable(table Table, query string, dropTable, ignoreDe
 	}
 
 	// https://github.com/Altinity/clickhouse-backup/issues/574, replace ENGINE=Distributed to new cluster name
-	if onCluster != "" && distributedRE.MatchString(query) {
+	// https://github.com/Altinity/clickhouse-backup/issues/1252, check exists
+	if distributedRE.MatchString(query) {
 		matches := distributedRE.FindAllStringSubmatch(query, -1)
-		newCluster := onCluster
 		oldCluster := strings.Trim(matches[0][2], "'\" ")
-		if newCluster != oldCluster && !strings.Contains(oldCluster, "{cluster}") {
+		newCluster := onCluster
+		var existCluster string
+		if newCluster == "" && !macroRE.MatchString(oldCluster) {
+			if err = ch.SelectSingleRowNoCtx(&existCluster, "SELECT cluster FROM system.clusters WHERE cluster=?", oldCluster); err != nil {
+				return fmt.Errorf("check system.clusters for %s return error: %v", oldCluster, err)
+			}
+			if existCluster == "" {
+				newCluster = strings.Trim(ch.Config.RestoreDistributedCluster, "'\" ")
+			}
+		}
+		if newCluster != "" && existCluster == "" && newCluster != oldCluster && !macroRE.MatchString(oldCluster) {
+			newCluster = "'" + strings.Trim(ch.Config.RestoreDistributedCluster, "'\" ") + "'"
 			log.Warn().Msgf("will replace cluster ENGINE=Distributed %s -> %s", matches[0][2], newCluster)
 			query = distributedRE.ReplaceAllString(query, fmt.Sprintf("${1}(%s,${3})", newCluster))
 		}

pkg/config/config.go

@@ -233,6 +233,7 @@ type ClickHouseConfig struct {
 	EmbeddedBackupDisk               string            `yaml:"embedded_backup_disk" envconfig:"CLICKHOUSE_EMBEDDED_BACKUP_DISK"`
 	BackupMutations                  bool              `yaml:"backup_mutations" envconfig:"CLICKHOUSE_BACKUP_MUTATIONS"`
 	RestoreAsAttach                  bool              `yaml:"restore_as_attach" envconfig:"CLICKHOUSE_RESTORE_AS_ATTACH"`
+	RestoreDistributedCluster        string            `yaml:"restore_distributed_cluster" envconfig:"CLICKHOUSE_RESTORE_DISTRIBUTED_CLUSTER"`
 	CheckPartsColumns                bool              `yaml:"check_parts_columns" envconfig:"CLICKHOUSE_CHECK_PARTS_COLUMNS"`
 	Secure                           bool              `yaml:"secure" envconfig:"CLICKHOUSE_SECURE"`
 	SkipVerify                       bool              `yaml:"skip_verify" envconfig:"CLICKHOUSE_SKIP_VERIFY"`

test/integration/integration_test.go

@@ -2866,6 +2866,86 @@ func TestRestoreAsAttach(t *testing.T) {
 	env.Cleanup(t, r)
 }
 
+func TestRestoreDistributedCluster(t *testing.T) {
+	env, r := NewTestEnvironment(t)
+	env.connectWithWait(t, r, 0*time.Second, 1*time.Second, 1*time.Minute)
+	xml := `
+<yandex>
+    <remote_servers>
+        <new_cluster>
+            <shard>
+                <internal_replication>true</internal_replication>
+                <replica>
+                    <host>127.0.0.1</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+        </new_cluster>
+    </remote_servers>
+</yandex>
+`
+
+	env.DockerExecNoError(r, "clickhouse", "bash", "-c", fmt.Sprintf("echo -n '%s' > /etc/clickhouse-server/config.d/new-cluster.xml", xml))
+	var clusterExists string
+	for i := 0; i < 10 && clusterExists == ""; i++ {
+		r.NoError(env.ch.SelectSingleRowNoCtx(&clusterExists, "SELECT cluster FROM system.clusters WHERE cluster='new_cluster'"))
+		if clusterExists == "new_cluster" {
+			break
+		}
+		time.Sleep(1 * time.Second)
+	}
+	r.Equal("new_cluster", clusterExists)
+
+	// Create test database and table
+	dbName := "test_restore_distributed_cluster"
+	tableName := "test_table"
+	env.queryWithNoError(r, "CREATE DATABASE IF NOT EXISTS "+dbName)
+	env.queryWithNoError(r, "CREATE TABLE "+dbName+"."+tableName+" (id UInt64, value String) ENGINE=MergeTree() ORDER BY id")
+	env.queryWithNoError(r, "CREATE TABLE "+dbName+"."+tableName+"_dist (id UInt64, value String) ENGINE=Distributed('new_cluster',"+dbName+","+tableName+")")
+	env.queryWithNoError(r, "INSERT INTO "+dbName+"."+tableName+"_dist SELECT number, toString(number) FROM numbers(100)")
+
+	// Create backup
+	backupName := "test_restore_distributed_cluster"
+	env.DockerExecNoError(r, "clickhouse-backup", "clickhouse-backup", "create", "-c", "/etc/clickhouse-backup/config-s3.yml", "--tables="+dbName+".*", backupName)
+
+	// Get row count before dropping
+	var rowCount uint64
+	r.NoError(env.ch.SelectSingleRowNoCtx(&rowCount, "SELECT count() FROM "+dbName+"."+tableName+"_dist"))
+	r.Equal(uint64(100), rowCount)
+
+	// Drop table and database
+	r.NoError(env.dropDatabase(dbName, false))
+
+	// remove cluster and wait configuration reload
+	env.DockerExecNoError(r, "clickhouse", "bash", "-c", "rm -rfv /etc/clickhouse-server/config.d/new-cluster.xml")
+	newClusterExists := uint64(1)
+	for i := 0; i < 10 && newClusterExists == 1; i++ {
+		r.NoError(env.ch.SelectSingleRowNoCtx(&newClusterExists, "SELECT count() FROM system.clusters WHERE cluster='new_cluster'"))
+		if newClusterExists == 0 {
+			break
+		}
+		time.Sleep(1 * time.Second)
+	}
+	r.Equal(uint64(0), newClusterExists)
+
+	// Restore using CLICKHOUSE_RESTORE_DISTRIBUTED_CLUSTER
+	env.DockerExecNoError(r, "clickhouse-backup", "bash", "-c", "RESTORE_SCHEMA_ON_CLUSTER='' CLICKHOUSE_RESTORE_DISTRIBUTED_CLUSTER={cluster} clickhouse-backup -c /etc/clickhouse-backup/config-s3.yml restore "+backupName)
+
+	// Verify data was restored correctly
+	r.NoError(env.ch.SelectSingleRowNoCtx(&rowCount, "SELECT count() FROM "+dbName+"."+tableName+"_dist"))
+	r.Equal(uint64(100), rowCount)
+	var tableDDL string
+	r.NoError(env.ch.SelectSingleRowNoCtx(&tableDDL, "SHOW CREATE TABLE "+dbName+"."+tableName+"_dist"))
+	r.NotContains(tableDDL, "new_cluster")
+	r.Contains(tableDDL, "Distributed('{cluster}'")
+
+	// Clean up
+	r.NoError(env.dropDatabase(dbName, false))
+	env.DockerExecNoError(r, "clickhouse-backup", "clickhouse-backup", "-c", "/etc/clickhouse-backup/config-s3.yml", "delete", "local", backupName)
+
+	env.Cleanup(t, r)
+}
+
 func TestReplicatedCopyToDetached(t *testing.T) {
 	env, r := NewTestEnvironment(t)
 	env.connectWithWait(t, r, 0*time.Second, 1*time.Second, 1*time.Minute)