Skip to content

Commit 58aab9e

Browse files
danhhzdanhhz
authored
Merge pull request cockroachdb#14292 from danhhz/placeholder
sqlccl: make backup/restore work with sql placeholders
2 parents 69521f9 + 275560c commit 58aab9e

File tree

9 files changed

+4395
-4299
lines changed

9 files changed

+4395
-4299
lines changed

‎pkg/ccl/sqlccl/backup.go‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -293,6 +293,15 @@ func backupPlanHook(
293293
return nil, nil, err
294294
}
295295

296+
toFn, err := p.TypeAsString(&backup.To)
297+
if err != nil {
298+
return nil, nil, err
299+
}
300+
incrementalFromFn, err := p.TypeAsStringArray(&backup.IncrementalFrom)
301+
if err != nil {
302+
return nil, nil, err
303+
}
304+
296305
header := sql.ResultColumns{
297306
{Name: "to", Typ: parser.TypeString},
298307
{Name: "startTs", Typ: parser.TypeString},
@@ -307,7 +316,7 @@ func backupPlanHook(
307316
var startTime hlc.Timestamp
308317
if backup.IncrementalFrom != nil {
309318
var err error
310-
startTime, err = ValidatePreviousBackups(ctx, backup.IncrementalFrom)
319+
startTime, err = ValidatePreviousBackups(ctx, incrementalFromFn())
311320
if err != nil {
312321
return nil, err
313322
}
@@ -319,9 +328,11 @@ func backupPlanHook(
319328
return nil, err
320329
}
321330
}
331+
332+
to := toFn()
322333
desc, err := Backup(ctx,
323334
p,
324-
backup.To,
335+
to,
325336
backup.Targets,
326337
startTime, endTime,
327338
backup.Options,
@@ -330,7 +341,7 @@ func backupPlanHook(
330341
return nil, err
331342
}
332343
ret := []parser.Datums{{
333-
parser.NewDString(backup.To),
344+
parser.NewDString(to),
334345
parser.NewDString(startTime.String()),
335346
parser.NewDString(endTime.String()),
336347
parser.NewDInt(parser.DInt(desc.DataSize)),

‎pkg/ccl/sqlccl/backup_test.go‎

Lines changed: 52 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -163,7 +163,7 @@ func backupAndRestore(
163163

164164
var unused string
165165
var dataSize int64
166-
sqlDB.QueryRow(fmt.Sprintf(`BACKUP DATABASE bench TO '%s'`, dest)).Scan(
166+
sqlDB.QueryRow(`BACKUP DATABASE bench TO $1`, dest).Scan(
167167
&unused, &unused, &unused, &dataSize,
168168
)
169169
approxDataSize := int64(backupRestoreRowPayloadSize) * numAccounts
@@ -188,7 +188,7 @@ func backupAndRestore(
188188
// Force the ID of the restored bank table to be different.
189189
sqlDBRestore.Exec(`CREATE TABLE bench.empty (a INT PRIMARY KEY)`)
190190

191-
sqlDBRestore.Exec(fmt.Sprintf(`RESTORE bench.* FROM '%s'`, dest))
191+
sqlDBRestore.Exec(`RESTORE bench.* FROM $1`, dest)
192192

193193
var rowCount int64
194194
sqlDBRestore.QueryRow(`SELECT COUNT(*) FROM bench.bank`).Scan(&rowCount)
@@ -224,19 +224,19 @@ func TestBackupRestoreInterleaved(t *testing.T) {
224224
// The bank table has numAccounts accounts, put 2x that in i0, 3x in i0_0,
225225
// and 4x in i1.
226226
for i := 0; i < numAccounts; i++ {
227-
_ = sqlDB.Exec(fmt.Sprintf(`INSERT INTO i0 VALUES (%d, 1), (%d, 2)`, i, i))
228-
_ = sqlDB.Exec(fmt.Sprintf(`INSERT INTO i0_0 VALUES (%d, 1, 1), (%d, 2, 2), (%d, 3, 3)`, i, i, i))
229-
_ = sqlDB.Exec(fmt.Sprintf(`INSERT INTO i1 VALUES (%d, 1), (%d, 2), (%d, 3), (%d, 4)`, i, i, i, i))
227+
_ = sqlDB.Exec(`INSERT INTO i0 VALUES ($1, 1), ($1, 2)`, i)
228+
_ = sqlDB.Exec(`INSERT INTO i0_0 VALUES ($1, 1, 1), ($1, 2, 2), ($1, 3, 3)`, i)
229+
_ = sqlDB.Exec(`INSERT INTO i1 VALUES ($1, 1), ($1, 2), ($1, 3), ($1, 4)`, i)
230230
}
231-
_ = sqlDB.Exec(fmt.Sprintf(`BACKUP DATABASE bench TO '%s'`, dir))
231+
_ = sqlDB.Exec(`BACKUP DATABASE bench TO $1`, dir)
232232

233233
t.Run("all tables in interleave hierarchy", func(t *testing.T) {
234234
tcRestore := testcluster.StartTestCluster(t, singleNode, base.TestClusterArgs{})
235235
defer tcRestore.Stopper().Stop()
236236
sqlDBRestore := sqlutils.MakeSQLRunner(t, tcRestore.Conns[0])
237237
sqlDBRestore.Exec(bankCreateDatabase)
238238

239-
sqlDBRestore.Exec(fmt.Sprintf(`RESTORE bench.* FROM '%s'`, dir))
239+
sqlDBRestore.Exec(`RESTORE bench.* FROM $1`, dir)
240240

241241
var rowCount int64
242242
sqlDBRestore.QueryRow(`SELECT COUNT(*) FROM bench.bank`).Scan(&rowCount)
@@ -263,7 +263,7 @@ func TestBackupRestoreInterleaved(t *testing.T) {
263263
sqlDBRestore := sqlutils.MakeSQLRunner(t, tcRestore.Conns[0])
264264
sqlDBRestore.Exec(bankCreateDatabase)
265265

266-
_, err := sqlDBRestore.DB.Exec(fmt.Sprintf(`RESTORE TABLE bench.i0 FROM '%s'`, dir))
266+
_, err := sqlDBRestore.DB.Exec(`RESTORE TABLE bench.i0 FROM $1`, dir)
267267
if !testutils.IsError(err, "without interleave parent") {
268268
t.Fatalf("expected 'without interleave parent' error but got: %+v", err)
269269
}
@@ -275,7 +275,7 @@ func TestBackupRestoreInterleaved(t *testing.T) {
275275
sqlDBRestore := sqlutils.MakeSQLRunner(t, tcRestore.Conns[0])
276276
sqlDBRestore.Exec(bankCreateDatabase)
277277

278-
_, err := sqlDBRestore.DB.Exec(fmt.Sprintf(`RESTORE TABLE bench.bank FROM '%s'`, dir))
278+
_, err := sqlDBRestore.DB.Exec(`RESTORE TABLE bench.bank FROM $1`, dir)
279279
if !testutils.IsError(err, "without interleave child") {
280280
t.Fatalf("expected 'without interleave child' error but got: %+v", err)
281281
}
@@ -325,7 +325,7 @@ func TestBackupRestoreFKs(t *testing.T) {
325325
)`)
326326

327327
for i := 0; i < numAccounts; i++ {
328-
origDB.Exec(fmt.Sprintf(`INSERT INTO store.customers VALUES (%d, '%d')`, i, i))
328+
origDB.Exec(`INSERT INTO store.customers VALUES ($1, $1::string)`, i)
329329
}
330330
// Each even customerID gets 3 orders, with predictable order and receipt IDs.
331331
for cID := 0; cID < numAccounts; cID += 2 {
@@ -339,7 +339,7 @@ func TestBackupRestoreFKs(t *testing.T) {
339339
}
340340
}
341341
}
342-
_ = origDB.Exec(fmt.Sprintf(`BACKUP DATABASE store TO '%s'`, dir))
342+
_ = origDB.Exec(`BACKUP DATABASE store TO $1`, dir)
343343
}
344344

345345
origCustomers := origDB.QueryStr(`SHOW CONSTRAINTS FROM store.customers`)
@@ -351,7 +351,7 @@ func TestBackupRestoreFKs(t *testing.T) {
351351
defer tc.Stopper().Stop()
352352
db := sqlutils.MakeSQLRunner(t, tc.Conns[0])
353353
db.Exec(createStore)
354-
db.Exec(fmt.Sprintf(`RESTORE store.* FROM '%s'`, dir))
354+
db.Exec(`RESTORE store.* FROM $1`, dir)
355355
// Restore's Validate checks all the tables point to each other correctly.
356356

357357
db.CheckQueryResults(`SHOW CONSTRAINTS FROM store.customers`, origCustomers)
@@ -393,7 +393,7 @@ func TestBackupRestoreFKs(t *testing.T) {
393393
defer tc.Stopper().Stop()
394394
db := sqlutils.MakeSQLRunner(t, tc.Conns[0])
395395
db.Exec(createStore)
396-
db.Exec(fmt.Sprintf(`RESTORE store.customers, store.orders FROM '%s'`, dir))
396+
db.Exec(`RESTORE store.customers, store.orders FROM $1`, dir)
397397
// Restore's Validate checks all the tables point to each other correctly.
398398

399399
// FK validation on customers from orders is preserved.
@@ -415,15 +415,14 @@ func TestBackupRestoreFKs(t *testing.T) {
415415

416416
// FK validation of self-FK is preserved.
417417
if _, err := db.DB.Exec(
418-
fmt.Sprintf(`RESTORE store.orders FROM '%s'`, dir),
418+
`RESTORE store.orders FROM $1`, dir,
419419
); !testutils.IsError(
420420
err, "cannot restore table \"orders\" without referenced table 53 \\(or \"skip_missing_foreign_keys\" option\\)",
421421
) {
422422
t.Fatal(err)
423423
}
424424

425-
db.Exec(
426-
fmt.Sprintf(`RESTORE store.orders FROM '%s' WITH OPTIONS ('skip_missing_foreign_keys')`, dir))
425+
db.Exec(`RESTORE store.orders FROM $1 WITH OPTIONS ('skip_missing_foreign_keys')`, dir)
427426
// Restore's Validate checks all the tables point to each other correctly.
428427

429428
// FK validation is gone.
@@ -436,8 +435,7 @@ func TestBackupRestoreFKs(t *testing.T) {
436435
defer tc.Stopper().Stop()
437436
db := sqlutils.MakeSQLRunner(t, tc.Conns[0])
438437
db.Exec(createStore)
439-
db.Exec(fmt.Sprintf(
440-
`RESTORE store.receipts FROM '%s' WITH OPTIONS ('skip_missing_foreign_keys')`, dir))
438+
db.Exec(`RESTORE store.receipts FROM $1 WITH OPTIONS ('skip_missing_foreign_keys')`, dir)
441439
// Restore's Validate checks all the tables point to each other correctly.
442440

443441
// FK validation of orders and customer is gone.
@@ -456,8 +454,7 @@ func TestBackupRestoreFKs(t *testing.T) {
456454
defer tc.Stopper().Stop()
457455
db := sqlutils.MakeSQLRunner(t, tc.Conns[0])
458456
db.Exec(createStore)
459-
db.Exec(fmt.Sprintf(
460-
`RESTORE store.receipts, store.customers FROM '%s' WITH OPTIONS ('skip_missing_foreign_keys')`, dir))
457+
db.Exec(`RESTORE store.receipts, store.customers FROM $1 WITH OPTIONS ('skip_missing_foreign_keys')`, dir)
461458
// Restore's Validate checks all the tables point to each other correctly.
462459

463460
// FK validation of orders is gone.
@@ -678,12 +675,12 @@ func TestBackupRestoreWithConcurrentWrites(t *testing.T) {
678675
<-bgActivity
679676

680677
// Backup DB while concurrent writes continue.
681-
sqlDB.Exec(fmt.Sprintf(`BACKUP DATABASE bench TO '%s'`, baseDir))
678+
sqlDB.Exec(`BACKUP DATABASE bench TO $1`, baseDir)
682679

683680
// Drop the table and restore from backup and check our invariant.
684681
atomic.StoreInt32(&allowErrors, 1)
685682
sqlDB.Exec(`DROP TABLE bench.bank`)
686-
sqlDB.Exec(fmt.Sprintf(`RESTORE bench.* FROM '%s'`, baseDir))
683+
sqlDB.Exec(`RESTORE bench.* FROM $1`, baseDir)
687684
atomic.StoreInt32(&allowErrors, 0)
688685

689686
bad := sqlDB.QueryStr(`SELECT id, balance, payload FROM bench.bank WHERE id != balance`)
@@ -718,7 +715,7 @@ func TestBackupAsOfSystemTime(t *testing.T) {
718715

719716
sqlDB.Exec(`DROP TABLE bench.bank`)
720717

721-
sqlDB.Exec(fmt.Sprintf(`RESTORE bench.* FROM '%s'`, dir))
718+
sqlDB.Exec(`RESTORE bench.* FROM $1`, dir)
722719

723720
sqlDB.QueryRow(`SELECT COUNT(*) FROM bench.bank`).Scan(&rowCount)
724721
if rowCount != numAccounts {
@@ -736,7 +733,7 @@ func TestBackupRestoreChecksum(t *testing.T) {
736733
_, dir, _, sqlDB, cleanupFn := backupRestoreTestSetup(t, singleNode, numAccounts)
737734
defer cleanupFn()
738735

739-
sqlDB.Exec(fmt.Sprintf(`BACKUP DATABASE bench TO '%s'`, dir))
736+
sqlDB.Exec(`BACKUP DATABASE bench TO $1`, dir)
740737

741738
var backupDesc BackupDescriptor
742739
{
@@ -768,7 +765,7 @@ func TestBackupRestoreChecksum(t *testing.T) {
768765
}
769766

770767
sqlDB.Exec(`DROP TABLE bench.bank`)
771-
_, err = sqlDB.DB.Exec(fmt.Sprintf(`RESTORE bench.* FROM '%s'`, dir))
768+
_, err = sqlDB.DB.Exec(`RESTORE bench.* FROM $1`, dir)
772769
if !testutils.IsError(err, "checksum mismatch") {
773770
t.Fatalf("expected 'checksum mismatch' error got: %+v", err)
774771
}
@@ -788,40 +785,40 @@ func TestTimestampMismatch(t *testing.T) {
788785
incrementalT2FromT1 := filepath.Join(dir, "2")
789786
incrementalT3FromT1OneTable := filepath.Join(dir, "3")
790787

791-
sqlDB.Exec(fmt.Sprintf(`BACKUP DATABASE bench TO '%s'`,
792-
fullBackup))
793-
sqlDB.Exec(fmt.Sprintf(`BACKUP DATABASE bench TO '%s' INCREMENTAL FROM '%s'`,
794-
incrementalT1FromFull, fullBackup))
795-
sqlDB.Exec(fmt.Sprintf(`BACKUP TABLE bench.bank TO '%s' INCREMENTAL FROM '%s'`,
796-
incrementalT3FromT1OneTable, fullBackup))
797-
sqlDB.Exec(fmt.Sprintf(`BACKUP DATABASE bench TO '%s' INCREMENTAL FROM '%s', '%s'`,
798-
incrementalT2FromT1, fullBackup, incrementalT1FromFull))
788+
sqlDB.Exec(`BACKUP DATABASE bench TO $1`,
789+
fullBackup)
790+
sqlDB.Exec(`BACKUP DATABASE bench TO $1 INCREMENTAL FROM $2`,
791+
incrementalT1FromFull, fullBackup)
792+
sqlDB.Exec(`BACKUP TABLE bench.bank TO $1 INCREMENTAL FROM $2`,
793+
incrementalT3FromT1OneTable, fullBackup)
794+
sqlDB.Exec(`BACKUP DATABASE bench TO $1 INCREMENTAL FROM $2, $3`,
795+
incrementalT2FromT1, fullBackup, incrementalT1FromFull)
799796

800797
t.Run("Backup", func(t *testing.T) {
801798
// Missing the initial full backup.
802-
_, err := sqlDB.DB.Exec(fmt.Sprintf(`BACKUP DATABASE bench TO '%s' INCREMENTAL FROM '%s'`,
803-
dir, incrementalT1FromFull))
799+
_, err := sqlDB.DB.Exec(`BACKUP DATABASE bench TO $1 INCREMENTAL FROM $2`,
800+
dir, incrementalT1FromFull)
804801
if !testutils.IsError(err, "no backup covers time") {
805802
t.Errorf("expected 'no backup covers time' error got: %+v", err)
806803
}
807804

808805
// Missing an intermediate incremental backup.
809-
_, err = sqlDB.DB.Exec(fmt.Sprintf(`BACKUP DATABASE bench TO '%s' INCREMENTAL FROM '%s', '%s'`,
810-
dir, fullBackup, incrementalT2FromT1))
806+
_, err = sqlDB.DB.Exec(`BACKUP DATABASE bench TO $1 INCREMENTAL FROM $2, $3`,
807+
dir, fullBackup, incrementalT2FromT1)
811808
if !testutils.IsError(err, "no backup covers time") {
812809
t.Errorf("expected 'no backup covers time' error got: %+v", err)
813810
}
814811

815812
// Backups specified out of order.
816-
_, err = sqlDB.DB.Exec(fmt.Sprintf(`BACKUP DATABASE bench TO '%s' INCREMENTAL FROM '%s', '%s'`,
817-
dir, incrementalT1FromFull, fullBackup))
813+
_, err = sqlDB.DB.Exec(`BACKUP DATABASE bench TO $1 INCREMENTAL FROM $2, $3`,
814+
dir, incrementalT1FromFull, fullBackup)
818815
if !testutils.IsError(err, "out of order") {
819816
t.Errorf("expected 'out of order' error got: %+v", err)
820817
}
821818

822819
// Missing data for one table in the most recent backup.
823-
_, err = sqlDB.DB.Exec(fmt.Sprintf(`BACKUP DATABASE bench TO '%s' INCREMENTAL FROM '%s', '%s'`,
824-
dir, fullBackup, incrementalT3FromT1OneTable))
820+
_, err = sqlDB.DB.Exec(`BACKUP DATABASE bench TO $1 INCREMENTAL FROM $2, $3`,
821+
dir, fullBackup, incrementalT3FromT1OneTable)
825822
if !testutils.IsError(err, "no backup covers time") {
826823
t.Errorf("expected 'no backup covers time' error got: %+v", err)
827824
}
@@ -831,29 +828,29 @@ func TestTimestampMismatch(t *testing.T) {
831828
sqlDB.Exec(`DROP TABLE bench.t2`)
832829
t.Run("Restore", func(t *testing.T) {
833830
// Missing the initial full backup.
834-
_, err := sqlDB.DB.Exec(fmt.Sprintf(`RESTORE bench.* FROM '%s'`,
835-
incrementalT1FromFull))
831+
_, err := sqlDB.DB.Exec(`RESTORE bench.* FROM $1`,
832+
incrementalT1FromFull)
836833
if !testutils.IsError(err, "no backup covers time") {
837834
t.Errorf("expected 'no backup covers time' error got: %+v", err)
838835
}
839836

840837
// Missing an intermediate incremental backup.
841-
_, err = sqlDB.DB.Exec(fmt.Sprintf(`RESTORE bench.* FROM '%s', '%s'`,
842-
fullBackup, incrementalT2FromT1))
838+
_, err = sqlDB.DB.Exec(`RESTORE bench.* FROM $1, $2`,
839+
fullBackup, incrementalT2FromT1)
843840
if !testutils.IsError(err, "no backup covers time") {
844841
t.Errorf("expected 'no backup covers time' error got: %+v", err)
845842
}
846843

847844
// Backups specified out of order.
848-
_, err = sqlDB.DB.Exec(fmt.Sprintf(`RESTORE bench.* FROM '%s', '%s'`,
849-
incrementalT1FromFull, fullBackup))
845+
_, err = sqlDB.DB.Exec(`RESTORE bench.* FROM $1, $2`,
846+
incrementalT1FromFull, fullBackup)
850847
if !testutils.IsError(err, "out of order") {
851848
t.Errorf("expected 'out of order' error got: %+v", err)
852849
}
853850

854851
// Missing data for one table in the most recent backup.
855-
_, err = sqlDB.DB.Exec(fmt.Sprintf(`RESTORE bench.* FROM '%s', '%s'`,
856-
fullBackup, incrementalT3FromT1OneTable))
852+
_, err = sqlDB.DB.Exec(`RESTORE bench.* FROM $1, $2`,
853+
fullBackup, incrementalT3FromT1OneTable)
857854
if !testutils.IsError(err, "no backup covers time") {
858855
t.Errorf("expected 'no backup covers time' error got: %+v", err)
859856
}
@@ -924,7 +921,7 @@ func TestBackupLevelDB(t *testing.T) {
924921
_, dir, _, sqlDB, cleanupFn := backupRestoreTestSetup(t, singleNode, 0)
925922
defer cleanupFn()
926923

927-
_ = sqlDB.Exec(fmt.Sprintf(`BACKUP DATABASE bench TO '%s'`, dir))
924+
_ = sqlDB.Exec(`BACKUP DATABASE bench TO $1`, dir)
928925

929926
// Verify that the sstables are in LevelDB format by checking the trailer
930927
// magic.
@@ -967,15 +964,15 @@ func TestRestoredPrivileges(t *testing.T) {
967964

968965
withGrants := sqlDB.QueryStr(`SHOW GRANTS ON bench.bank`)
969966

970-
sqlDB.Exec(fmt.Sprintf(`BACKUP DATABASE bench TO '%s'`, dir))
967+
sqlDB.Exec(`BACKUP DATABASE bench TO $1`, dir)
971968
sqlDB.Exec(`DROP TABLE bench.bank`)
972969

973970
t.Run("into fresh db", func(t *testing.T) {
974971
tc := testcluster.StartTestCluster(t, singleNode, base.TestClusterArgs{})
975972
defer tc.Stopper().Stop()
976973
sqlDBRestore := sqlutils.MakeSQLRunner(t, tc.Conns[0])
977974
sqlDBRestore.Exec(`CREATE DATABASE bench`)
978-
sqlDBRestore.Exec(fmt.Sprintf(`RESTORE bench.bank FROM '%s'`, dir))
975+
sqlDBRestore.Exec(`RESTORE bench.bank FROM $1`, dir)
979976
sqlDBRestore.CheckQueryResults(`SHOW GRANTS ON bench.bank`, rootOnly)
980977
})
981978

@@ -986,7 +983,7 @@ func TestRestoredPrivileges(t *testing.T) {
986983
sqlDBRestore.Exec(`CREATE DATABASE bench`)
987984
sqlDBRestore.Exec(`CREATE USER someone`)
988985
sqlDBRestore.Exec(`GRANT SELECT, INSERT, UPDATE, DELETE ON DATABASE bench TO someone`)
989-
sqlDBRestore.Exec(fmt.Sprintf(`RESTORE bench.bank FROM '%s'`, dir))
986+
sqlDBRestore.Exec(`RESTORE bench.bank FROM $1`, dir)
990987
sqlDBRestore.CheckQueryResults(`SHOW GRANTS ON bench.bank`, withGrants)
991988
})
992989
}
@@ -1001,7 +998,7 @@ func TestRestoreInto(t *testing.T) {
1001998
_, dir, _, sqlDB, cleanupFn := backupRestoreTestSetup(t, singleNode, numAccounts)
1002999
defer cleanupFn()
10031000

1004-
sqlDB.Exec(fmt.Sprintf(`BACKUP DATABASE bench TO '%s'`, dir))
1001+
sqlDB.Exec(`BACKUP DATABASE bench TO $1`, dir)
10051002

10061003
restoreStmt := fmt.Sprintf(`RESTORE bench.bank FROM '%s' WITH OPTIONS ('into_db'='bench2')`, dir)
10071004

@@ -1058,7 +1055,7 @@ func TestBackupRestorePermissions(t *testing.T) {
10581055
// Root doesn't have CREATE on `system` DB, so that should fail. Still need
10591056
// a valid `dir` though, since descriptors are always loaded first.
10601057
if _, err := sqlDB.DB.Exec(
1061-
fmt.Sprintf(`RESTORE bench.bank FROM '%s' WITH OPTIONS ('into_db'='system')`, dir),
1058+
`RESTORE bench.bank FROM $1 WITH OPTIONS ('into_db'='system')`, dir,
10621059
); !testutils.IsError(err, "user root does not have CREATE privilege") {
10631060
t.Fatal(err)
10641061
}

‎pkg/ccl/sqlccl/restore.go‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -653,12 +653,17 @@ func restorePlanHook(
653653
return nil, nil, err
654654
}
655655

656+
fromFn, err := p.TypeAsStringArray(&restore.From)
657+
if err != nil {
658+
return nil, nil, err
659+
}
660+
656661
fn := func() ([]parser.Datums, error) {
657662
// TODO(dan): Move this span into sql.
658663
ctx, span := tracing.ChildSpan(baseCtx, stmt.StatementTag())
659664
defer tracing.FinishSpan(span)
660665

661-
err := Restore(ctx, p, restore.From, restore.Targets, restore.Options)
666+
err := Restore(ctx, p, fromFn(), restore.Targets, restore.Options)
662667
return nil, err
663668
}
664669
return fn, nil, nil

0 commit comments

Comments
 (0)