crdb_internal.fingerprint(span: bytes[], start_time: timestamptz, all_revisions: bool) → intThis function is used only by CockroachDB’s developers for testing purposes.
crdb_internal.fingerprint(span: bytes[], stripped: bool) → intThis function is used only by CockroachDB’s developers for testing purposes.
+crdb_internal.force_assertion_error(msg: string) → intThis function is used only by CockroachDB’s developers for testing purposes.
crdb_internal.force_error(errorCode: string, msg: string) → intThis function is used only by CockroachDB’s developers for testing purposes.
diff --git a/pkg/ccl/backupccl/backup_test.go b/pkg/ccl/backupccl/backup_test.go index 4354b7d7db06..52300810646b 100644 --- a/pkg/ccl/backupccl/backup_test.go +++ b/pkg/ccl/backupccl/backup_test.go @@ -1080,6 +1080,8 @@ SELECT payload FROM "".crdb_internal.system_jobs ORDER BY created DESC LIMIT 10 sqlDB.Exec(t, incBackupQuery, queryArgs...) } + bankTableID := sqlutils.QueryTableID(t, conn, "data", "public", "bank") + backupTableFingerprint := sqlutils.FingerprintTable(t, sqlDB, bankTableID) sqlDB.Exec(t, `DROP DATABASE data CASCADE`) @@ -1100,7 +1102,7 @@ SELECT payload FROM "".crdb_internal.system_jobs ORDER BY created DESC LIMIT 10 restoreQuery = fmt.Sprintf("%s WITH kms = %s", restoreQuery, kmsURIFmtString) } queryArgs := append(restoreURIArgs, kmsURIArgs...) - verifyRestoreData(t, sqlDB, storageSQLDB, restoreQuery, queryArgs, numAccounts) + verifyRestoreData(t, sqlDB, storageSQLDB, restoreQuery, queryArgs, numAccounts, backupTableFingerprint) } func verifyRestoreData( @@ -1110,6 +1112,7 @@ func verifyRestoreData( restoreQuery string, restoreURIArgs []interface{}, numAccounts int, + bankStrippedFingerprint int, ) { var unused string var restored struct { @@ -1160,6 +1163,8 @@ func verifyRestoreData( t.Fatal("unexpected span start at primary index") } } + restorebankID := sqlutils.QueryTableID(t, sqlDB.DB, "data", "public", "bank") + require.Equal(t, bankStrippedFingerprint, sqlutils.FingerprintTable(t, sqlDB, restorebankID)) } func TestBackupRestoreSystemTables(t *testing.T) { diff --git a/pkg/ccl/backupccl/backuprand/BUILD.bazel b/pkg/ccl/backupccl/backuprand/BUILD.bazel index 5c782131328f..7a1597d20f90 100644 --- a/pkg/ccl/backupccl/backuprand/BUILD.bazel +++ b/pkg/ccl/backupccl/backuprand/BUILD.bazel @@ -30,6 +30,7 @@ go_test( "//pkg/util/log", "//pkg/util/randutil", "@com_github_stretchr_testify//assert", + "@com_github_stretchr_testify//require", ], ) diff --git a/pkg/ccl/backupccl/backuprand/backup_rand_test.go b/pkg/ccl/backupccl/backuprand/backup_rand_test.go index 40f684ea111b..9da5923a656d 100644 --- a/pkg/ccl/backupccl/backuprand/backup_rand_test.go +++ b/pkg/ccl/backupccl/backuprand/backup_rand_test.go @@ -28,6 +28,7 @@ import ( "github.com/cockroachdb/cockroach/pkg/util/log" "github.com/cockroachdb/cockroach/pkg/util/randutil" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) // TestBackupRestoreRandomDataRoundtrips conducts backup/restore roundtrips on @@ -97,12 +98,13 @@ database_name = 'rand' AND schema_name = 'public'`) } expectedCreateTableStmt := make(map[string]string) - expectedData := make(map[string][][]string) + expectedData := make(map[string]int) for _, tableName := range tableNames { expectedCreateTableStmt[tableName] = sqlDB.QueryStr(t, fmt.Sprintf(`SELECT create_statement FROM [SHOW CREATE TABLE %s]`, tree.NameString(tableName)))[0][0] if runSchemaOnlyExtension == "" { - expectedData[tableName] = sqlDB.QueryStr(t, fmt.Sprintf(`SELECT * FROM %s`, tree.NameString(tableName))) + tableID := sqlutils.QueryTableID(t, sqlDB.DB, "rand", "public", tableName) + expectedData[tableName] = sqlutils.FingerprintTable(t, sqlDB, tableID) } } @@ -135,7 +137,8 @@ database_name = 'rand' AND schema_name = 'public'`) assert.Equal(t, expectedCreateTableStmt[tableName], createStmt, "SHOW CREATE %s not equal after RESTORE", tableName) if runSchemaOnlyExtension == "" { - sqlDB.CheckQueryResults(t, fmt.Sprintf(`SELECT * FROM %s`, restoreTable), expectedData[tableName]) + tableID := sqlutils.QueryTableID(t, sqlDB.DB, "restoredb", "public", tableName) + require.Equal(t, expectedData[tableName], sqlutils.FingerprintTable(t, sqlDB, tableID)) } else { sqlDB.CheckQueryResults(t, fmt.Sprintf(`SELECT count(*) FROM %s`, restoreTable), [][]string{{"0"}}) diff --git a/pkg/kv/kvpb/api.proto b/pkg/kv/kvpb/api.proto index 303b60458029..625c3fff8cee 100644 --- a/pkg/kv/kvpb/api.proto +++ b/pkg/kv/kvpb/api.proto @@ -1696,7 +1696,13 @@ message ExportRequest { reserved 2, 5, 7, 8, 11; - // Next ID: 15 + FingerprintOptions fingerprint_options = 15 [(gogoproto.nullable) = false]; + + // Next ID: 16 +} + +message FingerprintOptions { + bool strip_index_prefix_and_timestamp = 1; } // BulkOpSummary summarizes the data processed by an operation, counting the diff --git a/pkg/kv/kvserver/batcheval/cmd_export.go b/pkg/kv/kvserver/batcheval/cmd_export.go index 25713c2f7076..a435e4e3d941 100644 --- a/pkg/kv/kvserver/batcheval/cmd_export.go +++ b/pkg/kv/kvserver/batcheval/cmd_export.go @@ -194,8 +194,23 @@ func evalExport( // values before fingerprinting so that the fingerprint is tenant // agnostic. opts.FingerprintOptions = storage.MVCCExportFingerprintOptions{ - StripTenantPrefix: true, - StripValueChecksum: true, + StripTenantPrefix: true, + StripValueChecksum: true, + StripIndexPrefixAndTimestamp: args.FingerprintOptions.StripIndexPrefixAndTimestamp, + } + if opts.FingerprintOptions.StripIndexPrefixAndTimestamp && args.MVCCFilter == kvpb.MVCCFilter_All { + // If a key's value were updated from a to b, the xor hash without + // timestamps of those two mvcc values would look the same if the key + // were updated from b to a. In other words, the order of key value + // updates without timestamps does not affect the xor hash; but this + // order clearly presents different mvcc history, therefore, do not + // strip timestamps if fingerprinting all mvcc history. + return result.Result{}, errors.New("cannot fingerprint without mvcc timestamps and with mvcc history") + } + if opts.FingerprintOptions.StripIndexPrefixAndTimestamp && !args.StartTime.IsEmpty() { + // Supplying a startKey only complicates results (e.g. it surfaces + // tombstones), given that only the latest keys are surfaced. + return result.Result{}, errors.New("cannot fingerprint without mvcc timestamps and with a start time") } var hasRangeKeys bool summary, resumeInfo, fingerprint, hasRangeKeys, err = storage.MVCCExportFingerprint(ctx, diff --git a/pkg/sql/sem/builtins/BUILD.bazel b/pkg/sql/sem/builtins/BUILD.bazel index 638635796fd1..7ddf32853374 100644 --- a/pkg/sql/sem/builtins/BUILD.bazel +++ b/pkg/sql/sem/builtins/BUILD.bazel @@ -9,6 +9,7 @@ go_library( "all_builtins.go", "builtins.go", "compression.go", + "fingerprint_builtins.go", "fixed_oids.go", "generator_builtins.go", "generator_probe_ranges.go", diff --git a/pkg/sql/sem/builtins/builtins.go b/pkg/sql/sem/builtins/builtins.go index f7308fec7ef4..0442e386b476 100644 --- a/pkg/sql/sem/builtins/builtins.go +++ b/pkg/sql/sem/builtins/builtins.go @@ -42,7 +42,6 @@ import ( "github.com/cockroachdb/cockroach/pkg/keys" "github.com/cockroachdb/cockroach/pkg/kv" "github.com/cockroachdb/cockroach/pkg/kv/kvpb" - "github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/lock" "github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverbase" "github.com/cockroachdb/cockroach/pkg/roachpb" "github.com/cockroachdb/cockroach/pkg/security/password" @@ -77,10 +76,7 @@ import ( "github.com/cockroachdb/cockroach/pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil" "github.com/cockroachdb/cockroach/pkg/sql/sqltelemetry" "github.com/cockroachdb/cockroach/pkg/sql/types" - "github.com/cockroachdb/cockroach/pkg/storage" "github.com/cockroachdb/cockroach/pkg/util" - "github.com/cockroachdb/cockroach/pkg/util/admission/admissionpb" - "github.com/cockroachdb/cockroach/pkg/util/contextutil" "github.com/cockroachdb/cockroach/pkg/util/duration" "github.com/cockroachdb/cockroach/pkg/util/encoding" "github.com/cockroachdb/cockroach/pkg/util/errorutil/unimplemented" @@ -7524,158 +7520,61 @@ expires until the statement bundle is collected`, Category: builtinconstants.CategorySystemInfo, }, tree.Overload{ + // If the second arg is set to true, this overload allows the caller to + // execute a "stripped" fingerprint on the latest keys in a span. This + // stripped fingerprint strips each key's timestamp and index prefix + // before hashing, enabling a user to assert that two different tables + // have the same latest keys, for example. Because the index prefix is + // stripped, this option should only get used in the table key space. + // + // If the stripped param is set to false, this overload is equivalent to + // 'crdb_internal.fingerprint(span,NULL,LATEST)' + Types: tree.ParamTypes{ {Name: "span", Typ: types.BytesArray}, - {Name: "start_time", Typ: types.TimestampTZ}, - {Name: "all_revisions", Typ: types.Bool}, + {Name: "stripped", Typ: types.Bool}, // NB: The function can be called with an AOST clause that will be used // as the `end_time` when issuing the ExportRequests for the purposes of // fingerprinting. }, ReturnType: tree.FixedReturnType(types.Int), Fn: func(ctx context.Context, evalCtx *eval.Context, args tree.Datums) (tree.Datum, error) { - ctx, sp := tracing.ChildSpan(ctx, "crdb_internal.fingerprint") - defer sp.Finish() - - if !evalCtx.Settings.Version.IsActive(ctx, clusterversion.V23_1) { - return nil, errors.Errorf("cannot use crdb_internal.fingerprint until the cluster version is at least %s", - clusterversion.V23_1.String()) + if len(args) != 2 { + return nil, errors.New("argument list must have two elements") } - - isAdmin, err := evalCtx.SessionAccessor.HasAdminRole(ctx) + span, err := parseSpan(args[0]) if err != nil { return nil, err } - if !isAdmin { - return nil, errors.New("crdb_internal.fingerprint() requires admin privilege") + skipTimestamp := bool(tree.MustBeDBool(args[1])) + return fingerprint(ctx, evalCtx, span, hlc.Timestamp{} /* allRevisions */, false, + skipTimestamp) + }, + Info: "This function is used only by CockroachDB's developers for testing purposes.", + Volatility: volatility.Stable, + }, + tree.Overload{ + Types: tree.ParamTypes{ + {Name: "span", Typ: types.BytesArray}, + {Name: "start_time", Typ: types.TimestampTZ}, + {Name: "all_revisions", Typ: types.Bool}, + // NB: The function can be called with an AOST clause that will be used + // as the `end_time` when issuing the ExportRequests for the purposes of + // fingerprinting. + }, + ReturnType: tree.FixedReturnType(types.Int), + Fn: func(ctx context.Context, evalCtx *eval.Context, args tree.Datums) (tree.Datum, error) { + if len(args) != 3 { + return nil, errors.New("argument list must have three elements") } - - arr := tree.MustBeDArray(args[0]) - if arr.Len() != 2 { - return nil, errors.New("expected an array of two elements") + span, err := parseSpan(args[0]) + if err != nil { + return nil, err } - startKey := []byte(tree.MustBeDBytes(arr.Array[0])) - endKey := []byte(tree.MustBeDBytes(arr.Array[1])) - startTime := args[1].(*tree.DTimestampTZ).Time + startTime := tree.MustBeDTimestampTZ(args[1]).Time startTimestamp := hlc.Timestamp{WallTime: startTime.UnixNano()} - allRevisions := *args[2].(*tree.DBool) - filter := kvpb.MVCCFilter_Latest - if allRevisions { - filter = kvpb.MVCCFilter_All - } - - header := kvpb.Header{ - Timestamp: evalCtx.Txn.ReadTimestamp(), - // We set WaitPolicy to Error, so that the export will return an error - // to us instead of a blocking wait if it hits any other txns. - // - // TODO(adityamaru): We might need to handle WriteIntentErrors - // specially in the future so as to allow the fingerprint to complete - // in the face of intents. - WaitPolicy: lock.WaitPolicy_Error, - // TODO(ssd): Setting this disables async sending in - // DistSender so it likely substantially impacts - // performance. - ReturnElasticCPUResumeSpans: true, - } - admissionHeader := kvpb.AdmissionHeader{ - Priority: int32(admissionpb.BulkNormalPri), - CreateTime: timeutil.Now().UnixNano(), - Source: kvpb.AdmissionHeader_FROM_SQL, - NoMemoryReservedAtSource: true, - } - - todo := make(chan kvpb.RequestHeader, 1) - todo <- kvpb.RequestHeader{Key: startKey, EndKey: endKey} - - ctxDone := ctx.Done() - var fingerprint uint64 - // TODO(adityamaru): Memory monitor this slice of buffered SSTs that - // contain range keys across ExportRequests. - ssts := make([][]byte, 0) - for { - select { - case <-ctxDone: - return nil, ctx.Err() - case reqHeader := <-todo: - req := &kvpb.ExportRequest{ - RequestHeader: reqHeader, - StartTime: startTimestamp, - MVCCFilter: filter, - ExportFingerprint: true, - } - var rawResp kvpb.Response - var pErr *kvpb.Error - exportRequestErr := contextutil.RunWithTimeout(ctx, - fmt.Sprintf("ExportRequest fingerprint for span %s", roachpb.Span{Key: startKey, EndKey: endKey}), - 5*time.Minute, func(ctx context.Context) error { - rawResp, pErr = kv.SendWrappedWithAdmission(ctx, - evalCtx.Txn.DB().NonTransactionalSender(), header, admissionHeader, req) - if pErr != nil { - return pErr.GoError() - } - return nil - }) - if exportRequestErr != nil { - return nil, exportRequestErr - } - - resp := rawResp.(*kvpb.ExportResponse) - for _, file := range resp.Files { - fingerprint = fingerprint ^ file.Fingerprint - - // Aggregate all the range keys that need fingerprinting once all - // ExportRequests have been completed. - if len(file.SST) != 0 { - ssts = append(ssts, file.SST) - } - } - if resp.ResumeSpan != nil { - if !resp.ResumeSpan.Valid() { - return nil, errors.Errorf("invalid resume span: %s", resp.ResumeSpan) - } - todo <- kvpb.RequestHeaderFromSpan(*resp.ResumeSpan) - } - default: - // No ExportRequests left to send. We've aggregated range keys - // across all ExportRequests and can now fingerprint them. - // - // NB: We aggregate rangekeys across ExportRequests and then - // fingerprint them on the client, instead of fingerprinting them as - // part of the ExportRequest command evaluation, because range keys - // do not have a stable, discrete identity. Their fragmentation can - // be influenced by rangekeys outside the time interval that we are - // fingerprinting, or by range splits. So, we need to "defragment" - // all the rangekey stacks we observe such that the fragmentation is - // deterministic on only the data we want to fingerprint in our key - // and time interval. - // - // Egs: - // - // t2 [-----)[----) - // - // t1 [----)[-----) - // a b c d - // - // Assume we have two rangekeys [a, c)@t1 and [b, d)@t2. They will - // fragment as shown in the diagram above. If we wish to fingerprint - // key [a-d) in time interval (t1, t2] the fragmented rangekey - // [a, c)@t1 is outside our time interval and should not influence our - // fingerprint. The iterator in `fingerprintRangekeys` will - // "defragment" the rangekey stacks [b-c)@t2 and [c-d)@t2 and - // fingerprint them as a single rangekey with bounds [b-d)@t2. - rangekeyFingerprint, err := storage.FingerprintRangekeys(ctx, evalCtx.Settings, - storage.MVCCExportFingerprintOptions{ - StripTenantPrefix: true, - StripValueChecksum: true, - }, ssts) - if err != nil { - return nil, err - } - fingerprint = fingerprint ^ rangekeyFingerprint - return tree.NewDInt(tree.DInt(fingerprint)), nil - } - } + allRevisions := bool(tree.MustBeDBool(args[2])) + return fingerprint(ctx, evalCtx, span, startTimestamp, allRevisions /* stripped */, false) }, Info: "This function is used only by CockroachDB's developers for testing purposes.", Volatility: volatility.Stable, @@ -10595,3 +10494,13 @@ func prettyStatement(p tree.PrettyCfg, stmt string) (string, error) { } return formattedStmt.String(), nil } + +func parseSpan(arg tree.Datum) (roachpb.Span, error) { + arr := tree.MustBeDArray(arg) + if arr.Len() != 2 { + return roachpb.Span{}, errors.New("expected an array of two elements") + } + startKey := []byte(tree.MustBeDBytes(arr.Array[0])) + endKey := []byte(tree.MustBeDBytes(arr.Array[1])) + return roachpb.Span{Key: startKey, EndKey: endKey}, nil +} diff --git a/pkg/sql/sem/builtins/fingerprint_builtin_test.go b/pkg/sql/sem/builtins/fingerprint_builtin_test.go index 7fe2451a81c8..8928d103d35f 100644 --- a/pkg/sql/sem/builtins/fingerprint_builtin_test.go +++ b/pkg/sql/sem/builtins/fingerprint_builtin_test.go @@ -273,3 +273,24 @@ FROM db.QueryRow(t, fingerprintQuery).Scan(&fingerprint4) require.NotEqual(t, fingerprint1, fingerprint4) } + +func TestFingerprintStripped(t *testing.T) { + defer leaktest.AfterTest(t)() + defer log.Scope(t).Close(t) + + ctx := context.Background() + s, sqlDB, _ := serverutils.StartServer(t, base.TestServerArgs{}) + defer s.Stopper().Stop(ctx) + db := sqlutils.MakeSQLRunner(sqlDB) + db.Exec(t, "CREATE DATABASE IF NOT EXISTS test") + db.Exec(t, "CREATE TABLE test.test (k PRIMARY KEY) AS SELECT generate_series(1, 10)") + + // Create the same sql rows in a different table, committed at a different timestamp. + db.Exec(t, "CREATE TABLE test.test2 (k PRIMARY KEY) AS SELECT generate_series(1, 10)") + + strippedFingerprint := func(tableName string) int { + tableID := sqlutils.QueryTableID(t, sqlDB, "test", "public", tableName) + return sqlutils.FingerprintTable(t, db, tableID) + } + require.Equal(t, strippedFingerprint("test"), strippedFingerprint("test2")) +} diff --git a/pkg/sql/sem/builtins/fingerprint_builtins.go b/pkg/sql/sem/builtins/fingerprint_builtins.go new file mode 100644 index 000000000000..4f9d0c4579e1 --- /dev/null +++ b/pkg/sql/sem/builtins/fingerprint_builtins.go @@ -0,0 +1,174 @@ +// Copyright 2023 The Cockroach Authors. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.txt. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0, included in the file +// licenses/APL.txt. + +package builtins + +import ( + "context" + "fmt" + "time" + + "github.com/cockroachdb/cockroach/pkg/clusterversion" + "github.com/cockroachdb/cockroach/pkg/kv" + "github.com/cockroachdb/cockroach/pkg/kv/kvpb" + "github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/lock" + "github.com/cockroachdb/cockroach/pkg/roachpb" + "github.com/cockroachdb/cockroach/pkg/sql/sem/eval" + "github.com/cockroachdb/cockroach/pkg/sql/sem/tree" + "github.com/cockroachdb/cockroach/pkg/storage" + "github.com/cockroachdb/cockroach/pkg/util/admission/admissionpb" + "github.com/cockroachdb/cockroach/pkg/util/contextutil" + "github.com/cockroachdb/cockroach/pkg/util/hlc" + "github.com/cockroachdb/cockroach/pkg/util/timeutil" + "github.com/cockroachdb/cockroach/pkg/util/tracing" + "github.com/cockroachdb/errors" +) + +func fingerprint( + ctx context.Context, + evalCtx *eval.Context, + span roachpb.Span, + startTime hlc.Timestamp, + allRevisions, stripped bool, +) (tree.Datum, error) { + ctx, sp := tracing.ChildSpan(ctx, "crdb_internal.fingerprint") + defer sp.Finish() + + if !evalCtx.Settings.Version.IsActive(ctx, clusterversion.V23_1) { + return nil, errors.Errorf("cannot use crdb_internal.fingerprint until the cluster version is at least %s", + clusterversion.V23_1.String()) + } + + isAdmin, err := evalCtx.SessionAccessor.HasAdminRole(ctx) + if err != nil { + return nil, err + } + if !isAdmin { + return nil, errors.New("crdb_internal.fingerprint() requires admin privilege") + } + + filter := kvpb.MVCCFilter_Latest + if allRevisions { + filter = kvpb.MVCCFilter_All + } + header := kvpb.Header{ + Timestamp: evalCtx.Txn.ReadTimestamp(), + // We set WaitPolicy to Error, so that the export will return an error + // to us instead of a blocking wait if it hits any other txns. + // + // TODO(adityamaru): We might need to handle WriteIntentErrors + // specially in the future so as to allow the fingerprint to complete + // in the face of intents. + WaitPolicy: lock.WaitPolicy_Error, + // TODO(ssd): Setting this disables async sending in + // DistSender so it likely substantially impacts + // performance. + ReturnElasticCPUResumeSpans: true, + } + admissionHeader := kvpb.AdmissionHeader{ + Priority: int32(admissionpb.BulkNormalPri), + CreateTime: timeutil.Now().UnixNano(), + Source: kvpb.AdmissionHeader_FROM_SQL, + NoMemoryReservedAtSource: true, + } + + todo := make(chan kvpb.RequestHeader, 1) + todo <- kvpb.RequestHeader{Key: span.Key, EndKey: span.EndKey} + ctxDone := ctx.Done() + var fingerprint uint64 + // TODO(adityamaru): Memory monitor this slice of buffered SSTs that + // contain range keys across ExportRequests. + ssts := make([][]byte, 0) + for { + select { + case <-ctxDone: + return nil, ctx.Err() + case reqHeader := <-todo: + req := &kvpb.ExportRequest{ + RequestHeader: reqHeader, + StartTime: startTime, + MVCCFilter: filter, + ExportFingerprint: true, + FingerprintOptions: kvpb.FingerprintOptions{StripIndexPrefixAndTimestamp: stripped}} + var rawResp kvpb.Response + var pErr *kvpb.Error + exportRequestErr := contextutil.RunWithTimeout(ctx, + fmt.Sprintf("ExportRequest fingerprint for span %s", roachpb.Span{Key: span.Key, + EndKey: span.EndKey}), + 5*time.Minute, func(ctx context.Context) error { + rawResp, pErr = kv.SendWrappedWithAdmission(ctx, + evalCtx.Txn.DB().NonTransactionalSender(), header, admissionHeader, req) + if pErr != nil { + return pErr.GoError() + } + return nil + }) + if exportRequestErr != nil { + return nil, exportRequestErr + } + + resp := rawResp.(*kvpb.ExportResponse) + for _, file := range resp.Files { + fingerprint = fingerprint ^ file.Fingerprint + + // Aggregate all the range keys that need fingerprinting once all + // ExportRequests have been completed. + if len(file.SST) != 0 { + ssts = append(ssts, file.SST) + } + } + if resp.ResumeSpan != nil { + if !resp.ResumeSpan.Valid() { + return nil, errors.Errorf("invalid resume span: %s", resp.ResumeSpan) + } + todo <- kvpb.RequestHeaderFromSpan(*resp.ResumeSpan) + } + default: + // No ExportRequests left to send. We've aggregated range keys + // across all ExportRequests and can now fingerprint them. + // + // NB: We aggregate rangekeys across ExportRequests and then + // fingerprint them on the client, instead of fingerprinting them as + // part of the ExportRequest command evaluation, because range keys + // do not have a stable, discrete identity. Their fragmentation can + // be influenced by rangekeys outside the time interval that we are + // fingerprinting, or by range splits. So, we need to "defragment" + // all the rangekey stacks we observe such that the fragmentation is + // deterministic on only the data we want to fingerprint in our key + // and time interval. + // + // Egs: + // + // t2 [-----)[----) + // + // t1 [----)[-----) + // a b c d + // + // Assume we have two rangekeys [a, c)@t1 and [b, d)@t2. They will + // fragment as shown in the diagram above. If we wish to fingerprint + // key [a-d) in time interval (t1, t2] the fragmented rangekey + // [a, c)@t1 is outside our time interval and should not influence our + // fingerprint. The iterator in `fingerprintRangekeys` will + // "defragment" the rangekey stacks [b-c)@t2 and [c-d)@t2 and + // fingerprint them as a single rangekey with bounds [b-d)@t2. + rangekeyFingerprint, err := storage.FingerprintRangekeys(ctx, evalCtx.Settings, + storage.MVCCExportFingerprintOptions{ + StripTenantPrefix: true, + StripValueChecksum: true, + StripIndexPrefixAndTimestamp: stripped, + }, ssts) + if err != nil { + return nil, err + } + fingerprint = fingerprint ^ rangekeyFingerprint + return tree.NewDInt(tree.DInt(fingerprint)), nil + } + } +} diff --git a/pkg/sql/sem/builtins/fixed_oids.go b/pkg/sql/sem/builtins/fixed_oids.go index 680d96e4a513..f6578eedeec2 100644 --- a/pkg/sql/sem/builtins/fixed_oids.go +++ b/pkg/sql/sem/builtins/fixed_oids.go @@ -2377,6 +2377,7 @@ var builtinOidsArray = []string{ 2403: `ts_rank(vector: tsvector, query: tsquery, normalization: int) -> float4`, 2404: `ts_rank(vector: tsvector, query: tsquery) -> float4`, 2405: `ts_rank(weights: float[], vector: tsvector, query: tsquery) -> float4`, + 2406: `crdb_internal.fingerprint(span: bytes[], stripped: bool) -> int`, } var builtinOidsBySignature map[string]oid.Oid diff --git a/pkg/storage/fingerprint_writer.go b/pkg/storage/fingerprint_writer.go index dc79afa324d1..fc24a35d00aa 100644 --- a/pkg/storage/fingerprint_writer.go +++ b/pkg/storage/fingerprint_writer.go @@ -20,6 +20,7 @@ import ( "github.com/cockroachdb/cockroach/pkg/keys" "github.com/cockroachdb/cockroach/pkg/roachpb" "github.com/cockroachdb/cockroach/pkg/settings/cluster" + "github.com/cockroachdb/cockroach/pkg/util/hlc" "github.com/cockroachdb/cockroach/pkg/util/tracing" "github.com/cockroachdb/errors" ) @@ -114,13 +115,11 @@ func (f *fingerprintWriter) PutRawMVCCRangeKey(key MVCCRangeKey, bytes []byte) e // PutRawMVCC implements the Writer interface. func (f *fingerprintWriter) PutRawMVCC(key MVCCKey, value []byte) error { defer f.hasher.Reset() - // Hash the key/timestamp and value of the RawMVCC. if err := f.hashKey(key.Key); err != nil { return err } - f.timestampBuf = EncodeMVCCTimestampToBuf(f.timestampBuf, key.Timestamp) - if err := f.hash(f.timestampBuf); err != nil { + if err := f.hashTimestamp(key.Timestamp); err != nil { return err } if err := f.hashValue(value); err != nil { @@ -147,12 +146,26 @@ func (f *fingerprintWriter) PutUnversioned(key roachpb.Key, value []byte) error } func (f *fingerprintWriter) hashKey(key []byte) error { + if f.options.StripIndexPrefixAndTimestamp { + return f.hash(f.stripIndexPrefix(key)) + } if f.options.StripTenantPrefix { return f.hash(f.stripTenantPrefix(key)) } return f.hash(key) } +func (f *fingerprintWriter) hashTimestamp(timestamp hlc.Timestamp) error { + if f.options.StripIndexPrefixAndTimestamp { + return nil + } + f.timestampBuf = EncodeMVCCTimestampToBuf(f.timestampBuf, timestamp) + if err := f.hash(f.timestampBuf); err != nil { + return err + } + return nil +} + func (f *fingerprintWriter) hashValue(value []byte) error { if f.options.StripValueChecksum { return f.hash(f.stripValueChecksum(value)) @@ -177,7 +190,15 @@ func (f *fingerprintWriter) stripValueChecksum(value []byte) []byte { } func (f *fingerprintWriter) stripTenantPrefix(key []byte) []byte { - remainder, _, err := keys.DecodeTenantPrefixE(key) + remainder, err := keys.StripTenantPrefix(key) + if err != nil { + return key + } + return remainder +} + +func (f *fingerprintWriter) stripIndexPrefix(key []byte) []byte { + remainder, err := keys.StripIndexPrefix(key) if err != nil { return key } @@ -247,8 +268,7 @@ func FingerprintRangekeys( return 0, err } for _, v := range stack.Versions { - fw.timestampBuf = EncodeMVCCTimestampToBuf(fw.timestampBuf, v.Timestamp) - if err := fw.hash(fw.timestampBuf); err != nil { + if err := fw.hashTimestamp(v.Timestamp); err != nil { return 0, err } mvccValue, ok, err := tryDecodeSimpleMVCCValue(v.Value) diff --git a/pkg/storage/mvcc.go b/pkg/storage/mvcc.go index b7035833f9cb..bf324a0f482c 100644 --- a/pkg/storage/mvcc.go +++ b/pkg/storage/mvcc.go @@ -6798,6 +6798,10 @@ type MVCCExportFingerprintOptions struct { // If StripValueChecksum is true, checksums are removed from // the value before hashing. StripValueChecksum bool + // If StripIndexPrefixAndTimestamp is true, the key's timestamp and index + // prefix are not hashed. Because the index prefix is stripped, this option + // should only get used in the table key space. + StripIndexPrefixAndTimestamp bool } // MVCCIsSpanEmptyOptions configures the MVCCIsSpanEmpty function. diff --git a/pkg/storage/mvcc_history_test.go b/pkg/storage/mvcc_history_test.go index f36cd8abd3d7..f77c62fd045a 100644 --- a/pkg/storage/mvcc_history_test.go +++ b/pkg/storage/mvcc_history_test.go @@ -1381,8 +1381,9 @@ func cmdExport(e *evalCtx) error { ExportAllRevisions: e.hasArg("allRevisions"), StopMidKey: e.hasArg("stopMidKey"), FingerprintOptions: storage.MVCCExportFingerprintOptions{ - StripTenantPrefix: e.hasArg("stripTenantPrefix"), - StripValueChecksum: e.hasArg("stripValueChecksum"), + StripTenantPrefix: e.hasArg("stripTenantPrefix"), + StripValueChecksum: e.hasArg("stripValueChecksum"), + StripIndexPrefixAndTimestamp: e.hasArg("stripped"), }, } if e.hasArg("maxIntents") { diff --git a/pkg/storage/testdata/mvcc_histories/export_fingerprint_no_timestamps b/pkg/storage/testdata/mvcc_histories/export_fingerprint_no_timestamps new file mode 100644 index 000000000000..821b22190130 --- /dev/null +++ b/pkg/storage/testdata/mvcc_histories/export_fingerprint_no_timestamps @@ -0,0 +1,77 @@ +# Test MVCC Fingerprinting when configured to ignore timestamps +# +# This test ensures that when the stripTimestamp flag is applied to a fingerpint export request, the +# timestamps in the MVCC history of a key span are ignored. To test, a simple history is +# constructed, fingerprinted, and wiped. Then the same history is reconstructed with the timestamps +# shifted up. + +# Sets up the following dataset, where x is MVCC point tombstone, o-o is MVCC +# range tombstone, [] is intent. We include some local timestamps, which should +# not be export fingerprinted. +# +# 6 +# 5 x o---o +# 4 a2 b1 o------o +# ----------------------------------------- +# 3 o---o +# 2 a2 x +# 1 b1 o-------o +# a b c d e + + +run ok +put k=a ts=2 v=a2 +put k=a ts=4 v=a2 +put k=b ts=1 v=b1 +del k=b ts=2 +put k=b ts=4 v=b1 +del k=b ts=5 +del_range_ts k=c end=e ts=1 +del_range_ts k=c end=d ts=3 +del_range_ts k=c end=e ts=4 +del_range_ts k=c end=d ts=5 +---- +del: "b": found key true +del: "b": found key true +>> at end: +rangekey: {c-d}/[5.000000000,0=/