Skip to content

Commit 3ca5d31

Browse files
SlachSlach
authored
Merge pull request #1303 from Altinity/improve_restore_table_mapping
Improve restore table mapping
2 parents 722de68 + a1bfd31 commit 3ca5d31

File tree

8 files changed

+493
-51
lines changed

8 files changed

+493
-51
lines changed

‎.github/workflows/build.yaml‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,7 @@ jobs:
108108
- '24.8'
109109
- '25.3'
110110
- '25.8'
111+
- '25.11'
111112
steps:
112113
- name: Checkout project
113114
uses: actions/checkout@v4
@@ -219,7 +220,7 @@ jobs:
219220
- '24.8'
220221
- '25.3'
221222
- '25.8'
222-
- '25.10'
223+
- '25.11'
223224
steps:
224225
- name: Checkout project
225226
uses: actions/checkout@v4

‎pkg/backup/backuper.go‎

Lines changed: 28 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -254,10 +254,7 @@ func (b *Backuper) getEmbeddedBackupLocation(ctx context.Context, backupName str
254254
return fmt.Sprintf("Disk('%s','%s')", b.cfg.ClickHouse.EmbeddedBackupDisk, backupName), nil
255255
}
256256
if b.cfg.General.RemoteStorage == "s3" {
257-
s3Endpoint, err := b.ch.ApplyMacros(ctx, b.buildEmbeddedLocationS3())
258-
if err != nil {
259-
return "", err
260-
}
257+
s3Endpoint := b.buildEmbeddedLocationS3(ctx)
261258
if b.cfg.S3.AccessKey != "" {
262259
return fmt.Sprintf("S3('%s/%s/','%s','%s')", s3Endpoint, backupName, b.cfg.S3.AccessKey, b.cfg.S3.SecretKey), nil
263260
}
@@ -267,10 +264,7 @@ func (b *Backuper) getEmbeddedBackupLocation(ctx context.Context, backupName str
267264
return "", errors.WithStack(errors.New("provide s3->access_key and s3->secret_key in config to allow embedded backup without `clickhouse->embedded_backup_disk`"))
268265
}
269266
if b.cfg.General.RemoteStorage == "gcs" {
270-
gcsEndpoint, err := b.ch.ApplyMacros(ctx, b.buildEmbeddedLocationGCS())
271-
if err != nil {
272-
return "", err
273-
}
267+
gcsEndpoint := b.buildEmbeddedLocationGCS(ctx)
274268
if b.cfg.GCS.EmbeddedAccessKey != "" {
275269
return fmt.Sprintf("S3('%s/%s/','%s','%s')", gcsEndpoint, backupName, b.cfg.GCS.EmbeddedAccessKey, b.cfg.GCS.EmbeddedSecretKey), nil
276270
}
@@ -280,45 +274,50 @@ func (b *Backuper) getEmbeddedBackupLocation(ctx context.Context, backupName str
280274
return "", fmt.Errorf("provide gcs->embedded_access_key and gcs->embedded_secret_key in config to allow embedded backup without `clickhouse->embedded_backup_disk`")
281275
}
282276
if b.cfg.General.RemoteStorage == "azblob" {
283-
azblobEndpoint, err := b.ch.ApplyMacros(ctx, b.buildEmbeddedLocationAZBLOB())
277+
azblobEndpoint := b.buildEmbeddedLocationAZBLOB()
278+
azblobPath, err := b.ch.ApplyMacros(ctx, b.cfg.AzureBlob.ObjectDiskPath)
284279
if err != nil {
285280
return "", err
286281
}
287282
if b.cfg.AzureBlob.Container != "" {
288-
return fmt.Sprintf("AzureBlobStorage('%s','%s','%s/%s/')", azblobEndpoint, b.cfg.AzureBlob.Container, b.cfg.AzureBlob.ObjectDiskPath, backupName), nil
283+
return fmt.Sprintf("AzureBlobStorage('%s','%s','%s/%s/')", azblobEndpoint, b.cfg.AzureBlob.Container, azblobPath, backupName), nil
289284
}
290285
return "", fmt.Errorf("provide azblob->container and azblob->account_name, azblob->account_key in config to allow embedded backup without `clickhouse->embedded_backup_disk`")
291286
}
292287
return "", fmt.Errorf("empty clickhouse->embedded_backup_disk and invalid general->remote_storage: %s", b.cfg.General.RemoteStorage)
293288
}
294289

295-
296-
func (b *Backuper) buildEmbeddedLocationS3() string {
290+
func (b *Backuper) buildEmbeddedLocationS3(ctx context.Context) string {
297291
s3backupURL := url.URL{}
298292
s3backupURL.Scheme = "https"
293+
s3Path, err := b.ch.ApplyMacros(ctx, b.cfg.S3.ObjectDiskPath)
294+
if err != nil {
295+
log.Error().Stack().Err(err).Send()
296+
return ""
297+
}
299298
if strings.HasPrefix(b.cfg.S3.Endpoint, "http") {
300299
newUrl, _ := s3backupURL.Parse(b.cfg.S3.Endpoint)
301300
s3backupURL = *newUrl
302-
s3backupURL.Path = path.Join(b.cfg.S3.Bucket, b.cfg.S3.ObjectDiskPath)
301+
s3backupURL.Path = path.Join(b.cfg.S3.Bucket, s3Path)
303302
} else {
304303
s3backupURL.Host = b.cfg.S3.Endpoint
305-
s3backupURL.Path = path.Join(b.cfg.S3.Bucket, b.cfg.S3.ObjectDiskPath)
304+
s3backupURL.Path = path.Join(b.cfg.S3.Bucket, s3Path)
306305
}
307306
if b.cfg.S3.DisableSSL {
308307
s3backupURL.Scheme = "http"
309308
}
310309
if s3backupURL.Host == "" && b.cfg.S3.Region != "" && b.cfg.S3.ForcePathStyle {
311310
s3backupURL.Host = "s3." + b.cfg.S3.Region + ".amazonaws.com"
312-
s3backupURL.Path = path.Join(b.cfg.S3.Bucket, b.cfg.S3.ObjectDiskPath)
311+
s3backupURL.Path = path.Join(b.cfg.S3.Bucket, s3Path)
313312
}
314313
if s3backupURL.Host == "" && b.cfg.S3.Bucket != "" && !b.cfg.S3.ForcePathStyle {
315314
s3backupURL.Host = b.cfg.S3.Bucket + "." + "s3." + b.cfg.S3.Region + ".amazonaws.com"
316-
s3backupURL.Path = b.cfg.S3.ObjectDiskPath
315+
s3backupURL.Path = s3Path
317316
}
318317
return s3backupURL.String()
319318
}
320319

321-
func (b *Backuper) buildEmbeddedLocationGCS() string {
320+
func (b *Backuper) buildEmbeddedLocationGCS(ctx context.Context) string {
322321
gcsBackupURL := url.URL{}
323322
gcsBackupURL.Scheme = "https"
324323
if b.cfg.GCS.ForceHttp {
@@ -328,14 +327,24 @@ func (b *Backuper) buildEmbeddedLocationGCS() string {
328327
if !strings.HasPrefix(b.cfg.GCS.Endpoint, "http") {
329328
gcsBackupURL.Host = b.cfg.GCS.Endpoint
330329
} else {
331-
newUrl, _ := gcsBackupURL.Parse(b.cfg.GCS.Endpoint)
330+
newUrl, err := gcsBackupURL.Parse(b.cfg.GCS.Endpoint)
331+
if err != nil {
332+
log.Error().Err(err).Stack().Send()
333+
return ""
334+
}
332335
gcsBackupURL = *newUrl
333336
}
334337
}
335338
if gcsBackupURL.Host == "" {
336339
gcsBackupURL.Host = "storage.googleapis.com"
337340
}
338-
gcsBackupURL.Path = path.Join(b.cfg.GCS.Bucket, b.cfg.GCS.ObjectDiskPath)
341+
gcsPath, err := b.ch.ApplyMacros(ctx, b.cfg.GCS.ObjectDiskPath)
342+
if err != nil {
343+
log.Error().Err(err).Stack().Send()
344+
return ""
345+
}
346+
347+
gcsBackupURL.Path = path.Join(b.cfg.GCS.Bucket, gcsPath)
339348
return gcsBackupURL.String()
340349
}
341350

‎pkg/backup/restore.go‎

Lines changed: 107 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1898,9 +1898,32 @@ func (b *Backuper) restoreDataRegular(ctx context.Context, backupName string, ba
18981898
}
18991899
// https://github.com/Altinity/clickhouse-backup/issues/937
19001900
if len(b.cfg.General.RestoreTableMapping) > 0 {
1901-
if targetTable, isMapped := b.cfg.General.RestoreTableMapping[table.Table]; isMapped {
1902-
dstTableName = targetTable
1903-
tablesForRestore[i].Table = targetTable
1901+
// Check full qualified name first (db.table), then table name only
1902+
fullName := table.Database + "." + table.Table
1903+
if targetValue, isMapped := b.cfg.General.RestoreTableMapping[fullName]; isMapped {
1904+
// Target may contain database (e.g., target_db.new_table)
1905+
if strings.Contains(targetValue, ".") {
1906+
parts := strings.SplitN(targetValue, ".", 2)
1907+
dstDatabase = parts[0]
1908+
dstTableName = parts[1]
1909+
tablesForRestore[i].Database = parts[0]
1910+
tablesForRestore[i].Table = parts[1]
1911+
} else {
1912+
dstTableName = targetValue
1913+
tablesForRestore[i].Table = targetValue
1914+
}
1915+
} else if targetTable, isMapped := b.cfg.General.RestoreTableMapping[table.Table]; isMapped {
1916+
// Handle target with database prefix
1917+
if strings.Contains(targetTable, ".") {
1918+
parts := strings.SplitN(targetTable, ".", 2)
1919+
dstDatabase = parts[0]
1920+
dstTableName = parts[1]
1921+
tablesForRestore[i].Database = parts[0]
1922+
tablesForRestore[i].Table = parts[1]
1923+
} else {
1924+
dstTableName = targetTable
1925+
tablesForRestore[i].Table = targetTable
1926+
}
19041927
}
19051928
}
19061929
logger := log.With().Str("table", fmt.Sprintf("%s.%s", dstDatabase, dstTableName)).Logger()
@@ -2255,8 +2278,24 @@ func (b *Backuper) checkMissingTables(tablesForRestore ListOfTables, chTables []
22552278
}
22562279
}
22572280
if len(b.cfg.General.RestoreTableMapping) > 0 {
2258-
if targetTable, isMapped := b.cfg.General.RestoreTableMapping[table.Table]; isMapped {
2259-
dstTable = targetTable
2281+
// Check full qualified name first (db.table), then table name only
2282+
fullName := table.Database + "." + table.Table
2283+
if targetValue, isMapped := b.cfg.General.RestoreTableMapping[fullName]; isMapped {
2284+
if strings.Contains(targetValue, ".") {
2285+
parts := strings.SplitN(targetValue, ".", 2)
2286+
dstDatabase = parts[0]
2287+
dstTable = parts[1]
2288+
} else {
2289+
dstTable = targetValue
2290+
}
2291+
} else if targetTable, isMapped := b.cfg.General.RestoreTableMapping[table.Table]; isMapped {
2292+
if strings.Contains(targetTable, ".") {
2293+
parts := strings.SplitN(targetTable, ".", 2)
2294+
dstDatabase = parts[0]
2295+
dstTable = parts[1]
2296+
} else {
2297+
dstTable = targetTable
2298+
}
22602299
}
22612300
}
22622301
found := false
@@ -2267,7 +2306,7 @@ func (b *Backuper) checkMissingTables(tablesForRestore ListOfTables, chTables []
22672306
}
22682307
}
22692308
if !found {
2270-
missingTables = append(missingTables, fmt.Sprintf("'%s.%s'", dstDatabase, table.Table))
2309+
missingTables = append(missingTables, fmt.Sprintf("'%s.%s'", dstDatabase, dstTable))
22712310
}
22722311
}
22732312
return missingTables
@@ -2290,25 +2329,80 @@ func (b *Backuper) changeTablePatternFromRestoreMapping(tablePattern, objType st
22902329
case "database":
22912330
mapping = b.cfg.General.RestoreDatabaseMapping
22922331
case "table":
2293-
mapping = b.cfg.General.RestoreDatabaseMapping
2332+
mapping = b.cfg.General.RestoreTableMapping
22942333
default:
2295-
return ""
2334+
return tablePattern
22962335
}
2336+
isDatabase := objType == "database"
22972337
for sourceObj, targetObj := range mapping {
22982338
if tablePattern != "" {
2299-
sourceObjRE := regexp.MustCompile(fmt.Sprintf("(^%s.*)|(,%s.*)", sourceObj, sourceObj))
2339+
var sourceObjRE *regexp.Regexp
2340+
if isDatabase {
2341+
sourceObjRE = regexp.MustCompile(fmt.Sprintf("(^%s\\.[^,]*)|(,%s\\.[^,]*)", sourceObj, sourceObj))
2342+
} else {
2343+
// Check if sourceObj is a full qualified name (db.table)
2344+
if strings.Contains(sourceObj, ".") {
2345+
// Full qualified mapping: source_db.table -> target_db.new_table
2346+
escapedSource := regexp.QuoteMeta(sourceObj)
2347+
sourceObjRE = regexp.MustCompile(fmt.Sprintf("(^%s)|(,%s)", escapedSource, escapedSource))
2348+
} else {
2349+
sourceObjRE = regexp.MustCompile(fmt.Sprintf("(^([^\\.]+)\\.%s)|(,([^\\.]+)\\.%s)", sourceObj, sourceObj))
2350+
}
2351+
}
2352+
23002353
if sourceObjRE.MatchString(tablePattern) {
23012354
matches := sourceObjRE.FindAllStringSubmatch(tablePattern, -1)
2302-
substitution := targetObj + ".*"
2303-
if strings.HasPrefix(matches[0][1], ",") {
2355+
var substitution string
2356+
if isDatabase {
2357+
substitution = targetObj + ".*"
2358+
} else {
2359+
// Check if sourceObj is full qualified
2360+
if strings.Contains(sourceObj, ".") {
2361+
// Use targetObj as-is (may contain database)
2362+
substitution = targetObj
2363+
} else {
2364+
// matches[0][2] has database name when first alternative matches (^...)
2365+
// matches[0][4] has database name when second alternative matches (,...)
2366+
dbName := matches[0][2]
2367+
if dbName == "" && len(matches[0]) > 4 {
2368+
dbName = matches[0][4]
2369+
}
2370+
// Check if targetObj contains database
2371+
if strings.Contains(targetObj, ".") {
2372+
substitution = targetObj
2373+
} else {
2374+
substitution = dbName + "." + targetObj
2375+
}
2376+
}
2377+
}
2378+
if strings.HasPrefix(matches[0][0], ",") {
23042379
substitution = "," + substitution
23052380
}
2381+
23062382
tablePattern = sourceObjRE.ReplaceAllString(tablePattern, substitution)
23072383
} else {
2308-
tablePattern += "," + targetObj + ".*"
2384+
if isDatabase {
2385+
tablePattern += "," + targetObj + ".*"
2386+
} else {
2387+
// Check if targetObj contains database
2388+
if strings.Contains(targetObj, ".") {
2389+
tablePattern += "," + targetObj
2390+
} else {
2391+
tablePattern += ",*." + targetObj
2392+
}
2393+
}
23092394
}
23102395
} else {
2311-
tablePattern += targetObj + ".*"
2396+
if isDatabase {
2397+
tablePattern += targetObj + ".*"
2398+
} else {
2399+
// Check if targetObj contains database
2400+
if strings.Contains(targetObj, ".") {
2401+
tablePattern += targetObj
2402+
} else {
2403+
tablePattern += "*." + targetObj
2404+
}
2405+
}
23122406
}
23132407
}
23142408
return tablePattern

0 commit comments

Comments
 (0)