crdb_internal.encode_key(table_id: int, index_id: int, row_tuple: anyelement) → bytesGenerate the key for a row on a particular table and index.
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.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/kv/kvserver/batcheval/cmd_export.go b/pkg/kv/kvserver/batcheval/cmd_export.go index 36e450df72c4..ef6826d5a0b0 100644 --- a/pkg/kv/kvserver/batcheval/cmd_export.go +++ b/pkg/kv/kvserver/batcheval/cmd_export.go @@ -181,19 +181,34 @@ func evalExport( var curSizeOfExportedSSTs int64 for start := args.Key; start != nil; { destFile := &storage.MemFile{} - summary, resume, err := storage.MVCCExportToSST(ctx, cArgs.EvalCtx.ClusterSettings(), reader, - storage.MVCCExportOptions{ - StartKey: storage.MVCCKey{Key: start, Timestamp: resumeKeyTS}, - EndKey: args.EndKey, - StartTS: args.StartTime, - EndTS: h.Timestamp, - ExportAllRevisions: exportAllRevisions, - TargetSize: targetSize, - MaxSize: maxSize, - MaxIntents: maxIntents, - StopMidKey: args.SplitMidKey, - ResourceLimiter: storage.NewResourceLimiter(storage.ResourceLimiterOptions{MaxRunTime: maxRunTime}, timeutil.DefaultTimeSource{}), - }, destFile) + opts := storage.MVCCExportOptions{ + StartKey: storage.MVCCKey{Key: start, Timestamp: resumeKeyTS}, + EndKey: args.EndKey, + StartTS: args.StartTime, + EndTS: h.Timestamp, + ExportAllRevisions: exportAllRevisions, + TargetSize: targetSize, + MaxSize: maxSize, + MaxIntents: maxIntents, + StopMidKey: args.SplitMidKey, + ResourceLimiter: storage.NewResourceLimiter(storage.ResourceLimiterOptions{MaxRunTime: maxRunTime}, timeutil.DefaultTimeSource{}), + } + var summary roachpb.BulkOpSummary + var resume storage.MVCCKey + var fingerprint uint64 + var err error + if args.ExportFingerprint { + // Default to stripping the tenant prefix from keys, and checksum from + // values before fingerprinting so that the fingerprint is tenant + // agnostic. + opts.FingerprintOptions = storage.MVCCExportFingerprintOptions{ + StripTenantPrefix: true, + StripValueChecksum: true, + } + summary, resume, fingerprint, err = storage.MVCCExportFingerprint(ctx, cArgs.EvalCtx.ClusterSettings(), reader, opts, destFile) + } else { + summary, resume, err = storage.MVCCExportToSST(ctx, cArgs.EvalCtx.ClusterSettings(), reader, opts, destFile) + } if err != nil { if errors.HasType(err, (*storage.ExceedMaxSizeError)(nil)) { err = errors.WithHintf(err, @@ -217,10 +232,11 @@ func evalExport( span.EndKey = args.EndKey } exported := roachpb.ExportResponse_File{ - Span: span, - EndKeyTS: resume.Timestamp, - Exported: summary, - SST: data, + Span: span, + EndKeyTS: resume.Timestamp, + Exported: summary, + SST: data, + Fingerprint: fingerprint, } reply.Files = append(reply.Files, exported) start = resume.Key diff --git a/pkg/roachpb/api.proto b/pkg/roachpb/api.proto index c9c0754504fd..9d0d8635e5ea 100644 --- a/pkg/roachpb/api.proto +++ b/pkg/roachpb/api.proto @@ -1546,7 +1546,25 @@ message ExportRequest { // then there is no limit. int64 target_file_size = 10; + // ExportFingerprint when set to true will result in ExportRequest command + // evaluation generating an fnv64 hash for every key/timestamp and value, for + // point keys encountered in the key/time interval. Each KV hash will be + // combined via a XOR into a running aggregate that is returned as part of the + // ExportResponse. + // + // Range keys are not fingerprinted but instead written to a pebble SST that + // is returned to the caller. This is because range keys do not have a stable, + // discrete identity. A range key can span multiple ranges, or it may be + // fragmented by range keys outside of the time bounds which are not relevant + // to the fingerprint. So, we could need multiple export requests to piece + // together the entire rangekey before fingerprinting it. It is up to the + // caller to define a deterministic fingerprinting scheme across all returned + // range keys. + bool export_fingerprint = 14; + reserved 2, 5, 7, 8, 11; + + // Next ID: 15 } // BulkOpSummary summarizes the data processed by an operation, counting the @@ -1586,14 +1604,24 @@ message ExportResponse { (gogoproto.customname) = "EndKeyTS" ]; string path = 2; - reserved 3; - reserved 4; - reserved 5; BulkOpSummary exported = 6 [(gogoproto.nullable) = false]; bytes sst = 7 [(gogoproto.customname) = "SST"]; string locality_kv = 8 [(gogoproto.customname) = "LocalityKV"]; + + // Fingerprint is the XOR aggregate of the fnv64 hash of every point + // key/timestamp and corresponding value that has been exported as part of + // the ExportRequest. This field is only set when the request is sent with + // `ExportFingerprint` set to true. + // + // Range keys are not fingerprinted but instead written to the sst above + // that is returned to the caller. This is because range keys do not have a + // stable, discrete identity and so it is up to the caller to define a + // deterministic fingerprinting scheme across all returned range keys. + uint64 fingerprint = 10; + + reserved 3, 4, 5; } ResponseHeader header = 1 [(gogoproto.nullable) = false, (gogoproto.embed) = true]; diff --git a/pkg/sql/sem/builtins/BUILD.bazel b/pkg/sql/sem/builtins/BUILD.bazel index 30c4f627bfd2..10d34ed06af1 100644 --- a/pkg/sql/sem/builtins/BUILD.bazel +++ b/pkg/sql/sem/builtins/BUILD.bazel @@ -46,6 +46,7 @@ go_library( "//pkg/keys", "//pkg/kv", "//pkg/kv/kvclient", + "//pkg/kv/kvserver/concurrency/lock", "//pkg/kv/kvserver/kvserverbase", "//pkg/repstream/streampb", "//pkg/roachpb", @@ -88,7 +89,9 @@ go_library( "//pkg/sql/storageparam", "//pkg/sql/storageparam/indexstorageparam", "//pkg/sql/types", + "//pkg/storage", "//pkg/util", + "//pkg/util/admission/admissionpb", "//pkg/util/arith", "//pkg/util/bitarray", "//pkg/util/contextutil", @@ -142,6 +145,7 @@ go_test( "all_builtins_test.go", "builtins_test.go", "datums_to_bytes_builtin_test.go", + "fingerprint_builtin_test.go", "generator_builtins_test.go", "geo_builtins_test.go", "help_test.go", @@ -158,6 +162,8 @@ go_test( "//pkg/base", "//pkg/keys", "//pkg/kv", + "//pkg/kv/kvserver", + "//pkg/roachpb", "//pkg/security/securityassets", "//pkg/security/securitytest", "//pkg/server", @@ -173,16 +179,21 @@ go_test( "//pkg/sql/sem/tree/treewindow", "//pkg/sql/sem/volatility", "//pkg/sql/types", + "//pkg/storage", "//pkg/testutils", "//pkg/testutils/serverutils", "//pkg/testutils/skip", "//pkg/testutils/sqlutils", + "//pkg/testutils/storageutils", "//pkg/testutils/testcluster", "//pkg/util", "//pkg/util/duration", + "//pkg/util/hlc", "//pkg/util/leaktest", + "//pkg/util/log", "//pkg/util/mon", "//pkg/util/randutil", + "//pkg/util/syncutil", "//pkg/util/timeutil", "@com_github_lib_pq//:pq", "@com_github_stretchr_testify//assert", diff --git a/pkg/sql/sem/builtins/builtins.go b/pkg/sql/sem/builtins/builtins.go index 2eba5e4f292f..2abd87efd145 100644 --- a/pkg/sql/sem/builtins/builtins.go +++ b/pkg/sql/sem/builtins/builtins.go @@ -40,6 +40,7 @@ import ( "github.com/cockroachdb/cockroach/pkg/config/zonepb" "github.com/cockroachdb/cockroach/pkg/keys" "github.com/cockroachdb/cockroach/pkg/kv" + "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" @@ -72,11 +73,15 @@ 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" "github.com/cockroachdb/cockroach/pkg/util/fuzzystrmatch" + "github.com/cockroachdb/cockroach/pkg/util/hlc" "github.com/cockroachdb/cockroach/pkg/util/humanizeutil" "github.com/cockroachdb/cockroach/pkg/util/ipaddr" "github.com/cockroachdb/cockroach/pkg/util/json" @@ -7327,6 +7332,167 @@ expires until the statement bundle is collected`, Volatility: volatility.Volatile, }, ), + "crdb_internal.fingerprint": makeBuiltin( + tree.FunctionProperties{ + Category: builtinconstants.CategorySystemInfo, + }, + tree.Overload{ + Types: tree.ArgTypes{ + {"span", types.BytesArray}, + {"start_time", types.TimestampTZ}, + {"all_revisions", 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()) + } + + isAdmin, err := evalCtx.SessionAccessor.HasAdminRole(ctx) + if err != nil { + return nil, err + } + if !isAdmin { + return nil, errors.New("crdb_internal.fingerprint() requires admin privilege") + } + arr := tree.MustBeDArray(args[0]) + if arr.Len() != 2 { + return nil, errors.New("expected an array of two elements") + } + startKey := []byte(tree.MustBeDBytes(arr.Array[0])) + endKey := []byte(tree.MustBeDBytes(arr.Array[1])) + endTime := hlc.Timestamp{WallTime: timeutil.Now().UnixNano()} + if evalCtx.AsOfSystemTime != nil { + endTime = evalCtx.AsOfSystemTime.Timestamp + } + header := roachpb.Header{ + Timestamp: endTime, + // 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, + } + startTime := args[1].(*tree.DTimestampTZ).Time + startTimestamp := hlc.Timestamp{WallTime: startTime.UnixNano()} + allRevisions := *args[2].(*tree.DBool) + filter := roachpb.MVCCFilter_Latest + if allRevisions { + filter = roachpb.MVCCFilter_All + } + req := &roachpb.ExportRequest{ + RequestHeader: roachpb.RequestHeader{Key: startKey, EndKey: endKey}, + StartTime: startTimestamp, + MVCCFilter: filter, + ExportFingerprint: true, + } + admissionHeader := roachpb.AdmissionHeader{ + Priority: int32(admissionpb.BulkNormalPri), + CreateTime: timeutil.Now().UnixNano(), + Source: roachpb.AdmissionHeader_FROM_SQL, + NoMemoryReservedAtSource: true, + } + todo := make(chan *roachpb.ExportRequest, 1) + todo <- req + 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 req := <-todo: + var rawResp roachpb.Response + var pErr *roachpb.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.(*roachpb.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) + } + + resumeReq := req + resumeReq.RequestHeader = roachpb.RequestHeaderFromSpan(*resp.ResumeSpan) + todo <- resumeReq + } + 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 + } + } + }, + Info: "This function is used only by CockroachDB's developers for testing purposes.", + Volatility: volatility.Immutable, + }, + ), } var lengthImpls = func(incBitOverload bool) builtinDefinition { diff --git a/pkg/sql/sem/builtins/fingerprint_builtin_test.go b/pkg/sql/sem/builtins/fingerprint_builtin_test.go new file mode 100644 index 000000000000..a15b49c9c001 --- /dev/null +++ b/pkg/sql/sem/builtins/fingerprint_builtin_test.go @@ -0,0 +1,218 @@ +// Copyright 2022 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_test + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/cockroachdb/cockroach/pkg/base" + "github.com/cockroachdb/cockroach/pkg/kv/kvserver" + "github.com/cockroachdb/cockroach/pkg/roachpb" + "github.com/cockroachdb/cockroach/pkg/server" + "github.com/cockroachdb/cockroach/pkg/sql/sem/eval" + "github.com/cockroachdb/cockroach/pkg/storage" + "github.com/cockroachdb/cockroach/pkg/testutils/serverutils" + "github.com/cockroachdb/cockroach/pkg/testutils/storageutils" + "github.com/cockroachdb/cockroach/pkg/util/hlc" + "github.com/cockroachdb/cockroach/pkg/util/leaktest" + "github.com/cockroachdb/cockroach/pkg/util/log" + "github.com/cockroachdb/cockroach/pkg/util/syncutil" + "github.com/cockroachdb/cockroach/pkg/util/timeutil" + "github.com/stretchr/testify/require" +) + +// TestFingerprint tests the `crdb_internal.fingerprint` builtin in the presence +// of rangekeys. +func TestFingerprint(t *testing.T) { + defer leaktest.AfterTest(t)() + defer log.Scope(t).Close(t) + + ctx := context.Background() + var mu syncutil.Mutex + var numExportResponses int + var numSSTsInExportResponses int + serv, sqlDB, db := serverutils.StartServer(t, base.TestServerArgs{ + Knobs: base.TestingKnobs{ + Store: &kvserver.StoreTestingKnobs{ + TestingResponseFilter: func(ctx context.Context, ba *roachpb.BatchRequest, br *roachpb.BatchResponse) *roachpb.Error { + mu.Lock() + defer mu.Unlock() + for i, ru := range br.Responses { + if _, ok := ba.Requests[i].GetInner().(*roachpb.ExportRequest); ok { + exportResponse := ru.GetInner().(*roachpb.ExportResponse) + numExportResponses++ + numSSTsInExportResponses += len(exportResponse.Files) + } + } + return nil + }, + }, + }, + }) + + resetVars := func() { + mu.Lock() + defer mu.Unlock() + numExportResponses = 0 + numSSTsInExportResponses = 0 + } + + returnPointAndRangeKeys := func(eng storage.Engine) ([]storage.MVCCKeyValue, []storage.MVCCRangeKey) { + var rangeKeys []storage.MVCCRangeKey + var pointKeys []storage.MVCCKeyValue + for _, kvI := range storageutils.ScanKeySpan(t, eng, roachpb.Key("a"), roachpb.Key("z")) { + switch kv := kvI.(type) { + case storage.MVCCRangeKeyValue: + rangeKeys = append(rangeKeys, kv.RangeKey) + + case storage.MVCCKeyValue: + pointKeys = append(pointKeys, kv) + + default: + t.Fatalf("unknown type %t", kvI) + } + } + return pointKeys, rangeKeys + } + + fingerprint := func(t *testing.T, startKey, endKey string, startTime, endTime hlc.Timestamp, allRevisions bool) int64 { + decimal := eval.TimestampToDecimal(endTime) + var fingerprint int64 + query := fmt.Sprintf(`SELECT * FROM crdb_internal.fingerprint(ARRAY[$1::BYTES, $2::BYTES], $3, $4) AS OF SYSTEM TIME %s`, decimal.String()) + require.NoError(t, sqlDB.QueryRow(query, roachpb.Key(startKey), roachpb.Key(endKey), startTime.GoTime(), allRevisions).Scan(&fingerprint)) + return fingerprint + } + + // Disable index recommendation so that >5 invocations of + // `crdb_internal.fingerprint` does not result in an additional call for + // generating index recommendations. + _, err := sqlDB.Exec(`SET CLUSTER SETTING sql.metrics.statement_details.index_recommendation_collection.enabled = false`) + require.NoError(t, err) + + t.Run("fingerprint-empty-store", func(t *testing.T) { + fingerprint := fingerprint(t, "a", "z", hlc.Timestamp{WallTime: 0}, + hlc.Timestamp{WallTime: timeutil.Now().UnixNano()}, true /* allRevisions */) + require.Zero(t, fingerprint) + }) + + s := serv.(*server.TestServer) + defer s.Stopper().Stop(ctx) + + store, err := s.Stores().GetStore(s.GetFirstStoreID()) + require.NoError(t, err) + eng := store.Engine() + + // Insert some point keys. + txn := db.NewTxn(ctx, "test-point-keys") + pointKeysTS := hlc.Timestamp{WallTime: timeutil.Now().Round(time.Microsecond).UnixNano()} + require.NoError(t, txn.SetFixedTimestamp(ctx, pointKeysTS)) + require.NoError(t, txn.Put(ctx, "a", "value")) + require.NoError(t, txn.Put(ctx, "b", "value")) + require.NoError(t, txn.Put(ctx, "c", "value")) + require.NoError(t, txn.Put(ctx, "d", "value")) + require.NoError(t, txn.Commit(ctx)) + + // Run a scan to force intent resolution. + _, err = db.Scan(ctx, "a", "z", 0) + require.NoError(t, err) + + pointKeys, rangeKeys := returnPointAndRangeKeys(eng) + require.Len(t, pointKeys, 4) + require.Len(t, rangeKeys, 0) + + // The store will have: + // + // ts2 [----------- rt -------------) + // + // ts1 value value value value + // a b c d + // + // Fingerprint the point keys. + fingerprintPointKeys := fingerprint(t, "a", "z", + pointKeysTS.Add(int64(-time.Microsecond), 0), pointKeysTS, true /* allRevisions */) + + // The store will have: + // + // ts2 [------ rt --------) + // + // ts1 value value value value + // a b c d + require.NoError(t, db.DelRangeUsingTombstone(ctx, "a", "c")) + pointKeys, rangeKeys = returnPointAndRangeKeys(eng) + require.Len(t, pointKeys, 4) + require.Len(t, rangeKeys, 1) + rangeKey1Timestamp := rangeKeys[0].Timestamp + // Note, the timestamp comparison is a noop here but we need the timestamp for + // future AOST fingerprint queries. + require.Equal(t, []storage.MVCCRangeKey{ + storageutils.RangeKeyWithTS("a", "c", rangeKey1Timestamp), + }, rangeKeys) + + // Fingerprint the point and range keys. + fingerprintPointAndRangeKeys := fingerprint(t, "a", "z", + pointKeysTS.Add(int64(-time.Microsecond), 0), rangeKey1Timestamp, true /* allRevisions */) + require.NotEqual(t, int64(0), fingerprintPointAndRangeKeys) + require.NotEqual(t, fingerprintPointKeys, fingerprintPointAndRangeKeys) + + // Fingerprint only the range key. + fingerprintRangekeys := fingerprint(t, "a", "z", + rangeKey1Timestamp.Add(int64(-time.Microsecond), 0), rangeKey1Timestamp, true /* allRevisions */) + require.NotEqual(t, int64(0), fingerprintRangekeys) + + require.Equal(t, fingerprintPointAndRangeKeys, fingerprintPointKeys^fingerprintRangekeys) + + // The store now has: + // + // ts3 [------)[-------) + // + // ts2 [----------)[------) + // + // ts1 value value value value + // a b c d + require.NoError(t, db.DelRangeUsingTombstone(ctx, "b", "d")) + pointKeys, rangeKeys = returnPointAndRangeKeys(eng) + require.Len(t, pointKeys, 4) + require.Len(t, rangeKeys, 4) + rangeKey2Timestamp := rangeKeys[1].Timestamp + require.Equal(t, []storage.MVCCRangeKey{ + storageutils.RangeKeyWithTS("a", "b", rangeKey1Timestamp), + storageutils.RangeKeyWithTS("b", "c", rangeKey2Timestamp), + storageutils.RangeKeyWithTS("b", "c", rangeKey1Timestamp), + storageutils.RangeKeyWithTS("c", "d", rangeKey2Timestamp), + }, rangeKeys) + + // Even with the fragmentation of the first range key, our fingerprint for the + // point keys and first range key should be the same as before. + fingerprintFragmentedPointAndRangeKeys := fingerprint(t, "a", "z", + pointKeysTS.Add(int64(-time.Microsecond), 0), rangeKey1Timestamp, true /* allRevisions */) + require.Equal(t, fingerprintPointAndRangeKeys, fingerprintFragmentedPointAndRangeKeys) + + // Insert a split point so that we're returned 2 SSTs with rangekeys instead + // of one. This should not affect the fingerprint. + resetVars() + fingerprintPreSplit := fingerprint(t, "a", "z", pointKeysTS.Add(int64(-time.Microsecond), 0), + hlc.Timestamp{WallTime: timeutil.Now().Round(time.Microsecond).UnixNano()}, true /* allRevisions */) + require.Equal(t, 1, numSSTsInExportResponses) + require.Equal(t, 1, numExportResponses) + + require.NoError(t, db.AdminSplit(ctx, "c", hlc.MaxTimestamp, roachpb.AdminSplitRequest_INGESTION)) + + resetVars() + fingerprintPostSplit := fingerprint(t, "a", "z", pointKeysTS.Add(int64(-time.Microsecond), 0), + hlc.Timestamp{WallTime: timeutil.Now().Round(time.Microsecond).UnixNano()}, true /* allRevisions */) + require.Equal(t, 2, numSSTsInExportResponses) + require.Equal(t, 2, numExportResponses) + + require.Equal(t, fingerprintPreSplit, fingerprintPostSplit) +} diff --git a/pkg/sql/sem/builtins/fixed_oids.go b/pkg/sql/sem/builtins/fixed_oids.go index 2e7a8a7de046..dcfd41753c14 100644 --- a/pkg/sql/sem/builtins/fixed_oids.go +++ b/pkg/sql/sem/builtins/fixed_oids.go @@ -399,6 +399,7 @@ var builtinOidsBySignature = map[string]oid.Oid{ `crdb_internal.destroy_tenant(id: int) -> int`: 1304, `crdb_internal.destroy_tenant(id: int, synchronous: bool) -> int`: 1305, `crdb_internal.encode_key(table_id: int, index_id: int, row_tuple: anyelement) -> bytes`: 1307, + `crdb_internal.fingerprint(span: bytes[], start_time: timestamptz, all_revisions: bool) -> int`: 2046, `crdb_internal.filter_multiregion_fields_from_zone_config_sql(val: string) -> string`: 1366, `crdb_internal.force_assertion_error(msg: string) -> int`: 1311, `crdb_internal.force_delete_table_data(id: int) -> bool`: 1369, diff --git a/pkg/storage/fingerprint_writer.go b/pkg/storage/fingerprint_writer.go index bf1199e1dcb1..c3be092a2fb8 100644 --- a/pkg/storage/fingerprint_writer.go +++ b/pkg/storage/fingerprint_writer.go @@ -13,11 +13,13 @@ package storage import ( "context" "hash" + "hash/fnv" "io" "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/tracing" "github.com/cockroachdb/errors" ) @@ -180,3 +182,109 @@ func (f *fingerprintWriter) stripTenantPrefix(key []byte) []byte { } return remainder } + +// FingerprintRangekeys iterates over the provided SSTs, that are expected to +// contain only rangekeys, and maintains a XOR aggregate of each rangekey's +// fingerprint. +func FingerprintRangekeys( + ctx context.Context, cs *cluster.Settings, opts MVCCExportFingerprintOptions, ssts [][]byte, +) (uint64, error) { + ctx, sp := tracing.ChildSpan(ctx, "storage.FingerprintRangekeys") + defer sp.Finish() + _ = ctx // ctx is currently unused, but this new ctx should be used below in the future. + + if len(ssts) == 0 { + return 0, nil + } + + // Assert that the SSTs do not contain any point keys. + // + // NB: Combined point/range key iteration is usually a fair bit more expensive + // than iterating over them separately. + pointKeyIterOpts := IterOptions{ + KeyTypes: IterKeyTypePointsOnly, + UpperBound: keys.MaxKey, + } + pointKeyIter, err := NewMultiMemSSTIterator(ssts, false /* verify */, pointKeyIterOpts) + if err != nil { + return 0, err + } + defer pointKeyIter.Close() + for pointKeyIter.SeekGE(NilKey); ; pointKeyIter.Next() { + if valid, err := pointKeyIter.Valid(); !valid || err != nil { + if err != nil { + return 0, err + } + break + } + hasPoint, _ := pointKeyIter.HasPointAndRange() + if hasPoint { + return 0, errors.AssertionFailedf("unexpected point key; ssts should only contain range keys") + } + } + + rangeKeyIterOpts := IterOptions{ + KeyTypes: IterKeyTypeRangesOnly, + LowerBound: keys.MinKey, + UpperBound: keys.MaxKey, + } + var fingerprint uint64 + iter, err := NewMultiMemSSTIterator(ssts, true /* verify */, rangeKeyIterOpts) + if err != nil { + return fingerprint, err + } + defer iter.Close() + + destFile := &MemFile{} + fw := makeFingerprintWriter(ctx, fnv.New64(), cs, destFile, opts) + defer fw.Close() + fingerprintRangeKey := func(stack MVCCRangeKeyStack) (uint64, error) { + defer fw.hasher.Reset() + if err := fw.hashKey(stack.Bounds.Key); err != nil { + return 0, err + } + if err := fw.hashKey(stack.Bounds.EndKey); err != nil { + return 0, err + } + for _, v := range stack.Versions { + fw.timestampBuf = EncodeMVCCTimestampToBuf(fw.timestampBuf, v.Timestamp) + if err := fw.hash(fw.timestampBuf); err != nil { + return 0, err + } + mvccValue, ok, err := tryDecodeSimpleMVCCValue(v.Value) + if !ok && err == nil { + mvccValue, err = decodeExtendedMVCCValue(v.Value) + } + if err != nil { + return 0, errors.Wrapf(err, "decoding mvcc value %s", v.Value) + } + if err := fw.hashValue(mvccValue.Value.RawBytes); err != nil { + return 0, err + } + } + return fw.hasher.Sum64(), nil + } + + for iter.SeekGE(MVCCKey{Key: keys.MinKey}); ; iter.Next() { + if ok, err := iter.Valid(); err != nil { + return fingerprint, err + } else if !ok { + break + } + hasPoint, _ := iter.HasPointAndRange() + if hasPoint { + return fingerprint, errors.AssertionFailedf("unexpected point key; ssts should only contain range keys") + } + rangekeyFingerprint, err := fingerprintRangeKey(iter.RangeKeys()) + if err != nil { + return fingerprint, err + } + fw.xorAgg.add(rangekeyFingerprint) + } + + if len(destFile.Data()) != 0 { + return 0, errors.AssertionFailedf("unexpected data found in destFile") + } + + return fw.Finish() +} diff --git a/pkg/storage/mvcc.go b/pkg/storage/mvcc.go index 1f6a7cd7ae85..93a00fd5f6aa 100644 --- a/pkg/storage/mvcc.go +++ b/pkg/storage/mvcc.go @@ -5833,7 +5833,7 @@ func MVCCIsSpanEmpty( func MVCCExportFingerprint( ctx context.Context, cs *cluster.Settings, reader Reader, opts MVCCExportOptions, dest io.Writer, ) (roachpb.BulkOpSummary, MVCCKey, uint64, error) { - ctx, span := tracing.ChildSpan(ctx, "storage.MVCCExportToSST") + ctx, span := tracing.ChildSpan(ctx, "storage.MVCCExportFingerprint") defer span.Finish() hasher := fnv.New64() diff --git a/pkg/storage/mvcc_history_test.go b/pkg/storage/mvcc_history_test.go index a7163903fa56..61f1b2714d08 100644 --- a/pkg/storage/mvcc_history_test.go +++ b/pkg/storage/mvcc_history_test.go @@ -1374,7 +1374,18 @@ func cmdExport(e *evalCtx) error { e.results.buf.Printf("\n") if shouldFingerprint { + // Fingerprint the rangekeys returned as a pebble SST. + rangekeyFingerprint, err := storage.FingerprintRangekeys(e.ctx, e.st, opts.FingerprintOptions, + [][]byte{sstFile.Bytes()}) + if err != nil { + return err + } + fingerprint = fingerprint ^ rangekeyFingerprint e.results.buf.Printf("fingerprint: %d\n", fingerprint) + + // Return early, we don't need to print the point and rangekeys if we are + // fingerprinting. + return nil } iter, err := storage.NewMemSSTIterator(sstFile.Bytes(), false /* verify */, storage.IterOptions{ diff --git a/pkg/storage/testdata/mvcc_histories/export_fingerprint b/pkg/storage/testdata/mvcc_histories/export_fingerprint index 057ef6744f28..e645a26e2bf1 100644 --- a/pkg/storage/testdata/mvcc_histories/export_fingerprint +++ b/pkg/storage/testdata/mvcc_histories/export_fingerprint @@ -98,15 +98,7 @@ run ok export fingerprint k=a end=z ts=6 allRevisions ---- export: data_size:165 fingerprint=true -fingerprint: 17693463359975730253 -export: {a-b}/[1.000000000,0=/