feat(restore_test): test roundtrip schema restore for Scylla 6.0

This commit implements TestRestoreSchemaRoundtripIntegration which tests scenario where:
- schema is backed up on src
- restored on dst
- backed up on dst
- dropped on dst
- restored on dst
It also checks if restore brings back non-default keyspace+table options.

pkg/service/restore/helper_integration_test.go

@@ -393,6 +393,10 @@ func grantRestoreTablesPermissions(t *testing.T, s gocqlx.Session, restoredTable
 	}
 }
 
+func grantRestoreSchemaPermissions(t *testing.T, s gocqlx.Session, user string) {
+	ExecStmt(t, s, "GRANT CREATE ON ALL KEYSPACES TO "+user)
+}
+
 func validateCompleteProgress(t *testing.T, pr Progress, tables []table) {
 	if pr.Size != pr.Restored || pr.Size != pr.Downloaded {
 		t.Fatal("Expected complete restore")

pkg/service/restore/restore_integration_test.go

@@ -7,13 +7,16 @@ package restore_test
 
 import (
 	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/pkg/errors"
 	. "github.com/scylladb/scylla-manager/v3/pkg/service/backup/backupspec"
 	. "github.com/scylladb/scylla-manager/v3/pkg/testutils"
 	. "github.com/scylladb/scylla-manager/v3/pkg/testutils/db"
 	. "github.com/scylladb/scylla-manager/v3/pkg/testutils/testconfig"
+	"github.com/scylladb/scylla-manager/v3/pkg/util/maputil"
+	"github.com/scylladb/scylla-manager/v3/pkg/util/query"
 )
 
 func TestRestoreTablesUserIntegration(t *testing.T) {
@@ -93,3 +96,106 @@ func TestRestoreTablesNoReplicationIntegration(t *testing.T) {
 
 	h.validateIdenticalTables(t, []table{{ks: ks, tab: tab}})
 }
+
+func TestRestoreSchemaRoundtripIntegration(t *testing.T) {
+	h := newTestHelper(t, ManagedSecondClusterHosts(), ManagedClusterHosts())
+	hRev := newTestHelper(t, ManagedClusterHosts(), ManagedSecondClusterHosts())
+
+	if !checkAnyConstraint(t, h.dstCluster.Client, ">= 5.5, < 2000", ">= 2024.2, > 1000") {
+		t.Skip("This test assumes that schema is backed up and restored via DESCRIBE SCHEMA WITH INTERNALS")
+	}
+
+	ks := randomizedName("roundtrip_")
+	tab := randomizedName("tab_")
+	Print("Prepare schema with non-default options")
+	ksOpt := "durable_writes = false"
+	tabOpt := "compaction = {'class': 'NullCompactionStrategy', 'enabled': 'false'}"
+	objWithOpt := map[string]string{
+		ks:  ksOpt,
+		tab: tabOpt,
+	}
+	ksStmt := "CREATE KEYSPACE %q WITH replication = {'class': 'NetworkTopologyStrategy', 'dc1': %d} AND %s"
+	ExecStmt(t, h.srcCluster.rootSession, fmt.Sprintf(ksStmt, ks, 2, ksOpt))
+	tabStmt := "CREATE TABLE %q.%q (id int PRIMARY KEY, data blob) WITH %s"
+	ExecStmt(t, h.srcCluster.rootSession, fmt.Sprintf(tabStmt, ks, tab, tabOpt))
+
+	Print("Save src describe schema output")
+	srcSchema, err := query.DescribeSchemaWithInternals(h.srcCluster.rootSession)
+	if err != nil {
+		t.Fatal(errors.Wrap(err, "describe src schema"))
+	}
+
+	Print("Run src backup")
+	loc := []Location{testLocation("schema-roundtrip", "")}
+	S3InitBucket(t, loc[0].Path)
+	tag := h.runBackup(t, map[string]any{
+		"location": loc,
+	})
+
+	Print("Run restore of src backup")
+	grantRestoreSchemaPermissions(t, h.dstCluster.rootSession, h.dstUser)
+	h.runRestore(t, map[string]any{
+		"location":       loc,
+		"snapshot_tag":   tag,
+		"restore_schema": true,
+	})
+
+	Print("Save dst describe schema output from src backup")
+	dstSchemaSrcBackup, err := query.DescribeSchemaWithInternals(h.srcCluster.rootSession)
+	if err != nil {
+		t.Fatal(errors.Wrap(err, "describe dst schema from src backup"))
+	}
+
+	Print("Run dst backup")
+	tag = hRev.runBackup(t, map[string]any{
+		"location": loc,
+	})
+
+	Print("Drop restored schema")
+	ExecStmt(t, h.dstCluster.rootSession, "DROP KEYSPACE "+ks)
+
+	Print("Run restore of dst backup")
+	h.runRestore(t, map[string]any{
+		"location":       loc,
+		"snapshot_tag":   tag,
+		"restore_schema": true,
+	})
+
+	Print("Save dst describe schema output from dst backup")
+	dstSchemaDstBackup, err := query.DescribeSchemaWithInternals(h.srcCluster.rootSession)
+	if err != nil {
+		t.Fatal(errors.Wrap(err, "describe dst schema from dst backup"))
+	}
+
+	Print("Validate that schema contains objects with options")
+	var (
+		m1 = map[query.DescribedSchemaRow]struct{}{}
+		m2 = map[query.DescribedSchemaRow]struct{}{}
+		m3 = map[query.DescribedSchemaRow]struct{}{}
+	)
+	for _, row := range srcSchema {
+		m1[row] = struct{}{}
+		if opt, ok := objWithOpt[row.Name]; ok {
+			if !strings.Contains(row.CQLStmt, opt) {
+				t.Fatalf("Object: %v, with cql: %v, does not contain option: %v", row.Name, row.CQLStmt, opt)
+			}
+			delete(objWithOpt, row.Name)
+		}
+	}
+	if len(objWithOpt) > 0 {
+		t.Fatalf("Src schema: %v, is missing created objects: %v", m1, objWithOpt)
+	}
+	for _, row := range dstSchemaSrcBackup {
+		m2[row] = struct{}{}
+	}
+	for _, row := range dstSchemaDstBackup {
+		m3[row] = struct{}{}
+	}
+	Print("Validate that all schemas are the same")
+	if !maputil.Equal(m1, m2) {
+		t.Fatalf("Src schema: %v, dst schema from src backup: %v, are not equal", m1, m2)
+	}
+	if !maputil.Equal(m1, m3) {
+		t.Fatalf("Src schema: %v, dst schema from dst backup: %v, are not equal", m1, m3)
+	}
+}