Merge cockroachdb#106525 cockroachdb#107743 cockroachdb#108516 cockroachdb#108520 cockroachdb#108531

106525: sql: add trie tree based workload index recommendations r=qiyanghe1998 a=qiyanghe1998

#### sql: add trie tree based workload index recommendations

This commit adds the trie and the logic for getting the workload index
recommendations. In addition, it fills the gap between built-in functions and
backend implementation for workload index recommendations.  The whole process
consists of collecting candidates and finding representative indexes.

All the index recommendations in the table `system.statement_statistics`
(satisfying some time requirement) will be collected as the candidates and then
inserted to the trie.

The trie is designed for all the indexes of one table. The indexed columns will
be regarded as the key to insert into the tree in their original orders. The
storing part will be attached to the node after the insertion of indexed
columns.

The general idea of finding representative indexes is to use all the indexes
represented by the leaf nodes. One optimization is to use the remove the
storings that are covered by some leaf nodes. Next, we will push down all the
storings attached to the internal nodes to the shallowest leaf nodes (You can
find the reasons in RFC). Finally, all the indexes represented by the leaf
nodes will be returned.

As for the `DROP INDEX`, since we collect all the indexes represented by the
leaf nodes (a superset of dropped indexes), so we can directly drop all of
them.

Release note (sql change): new builtin functions `workload_index_recs()` and
`workload_index_recs(timestamptz)`, return workload level index recommendations
(columns of string, each string represent an index recommendation) from
statement level index recommendations (as candidates) in
`system.statement_statistics`. If the timestamptz is given, it will only
consider those candidates who is generated after that timestampsz.

Epic: None


107743: sql: Add new builtin generator function `crdb_internal.scan_storage_internal_keys()` r=RahulAggarwal1016 a=RahulAggarwal1016

This new builtin is used to gather specific pebble metrics for a node and store id (within an given keyspan). The builtin returns information about the different types of keys (including snapshot pinned keys) and bytes.

Informs: cockroachdb#94659
Release-note: None

108516: backupccl: fix error message for descriptor version mismatch r=adityamaru a=renatolabs

Expected and actual versions were swapped.

Epic: none

Release note: None

108520: backupccl: remove spurious print line in test r=msbutler a=msbutler

Print statements should not be in commited code

Epic: none

Release note: none

108531: roachtest: Ensure tpcc workloads runs for a bit r=miretskiy a=miretskiy

An issue in roachtest cockroachdb#108530 prevents clean test
termination when calling Wait() on a test monitor
that did not have at least 1 task started.

This cause `cdc/kafka-oauth` test to hang.
Add a '30s' duration to the tpcc task to go around this problem.

Fixes cockroachdb#108507

Release note: None

Co-authored-by: qiyanghe1998 <[email protected]>
Co-authored-by: craig[bot] <[email protected]>
Co-authored-by: Renato Costa <[email protected]>
Co-authored-by: Michael Butler <[email protected]>
Co-authored-by: Yevgeniy Miretskiy <[email protected]>

Author: 5 people

docs/generated/sql/functions.md

@@ -1386,11 +1386,7 @@ the locality flag on node startup. Returns an error if no region is set.</p>
 </span></td><td>Immutable</td></tr>
 <tr><td><a name="workload_index_recs"></a><code>workload_index_recs() &rarr; <a href="string.html">string</a></code></td><td><span class="funcdesc"><p>Returns set of index recommendations</p>
 </span></td><td>Immutable</td></tr>
-<tr><td><a name="workload_index_recs"></a><code>workload_index_recs(budget: <a href="string.html">string</a>) &rarr; <a href="string.html">string</a></code></td><td><span class="funcdesc"><p>Returns set of index recommendations</p>
-</span></td><td>Immutable</td></tr>
 <tr><td><a name="workload_index_recs"></a><code>workload_index_recs(timestamptz: <a href="timestamp.html">timestamptz</a>) &rarr; <a href="string.html">string</a></code></td><td><span class="funcdesc"><p>Returns set of index recommendations</p>
-</span></td><td>Immutable</td></tr>
-<tr><td><a name="workload_index_recs"></a><code>workload_index_recs(timestamptz: <a href="timestamp.html">timestamptz</a>, budget: <a href="string.html">string</a>) &rarr; <a href="string.html">string</a></code></td><td><span class="funcdesc"><p>Returns set of index recommendations</p>
 </span></td><td>Immutable</td></tr></tbody>
 </table>
 

pkg/BUILD.bazel

@@ -486,6 +486,7 @@ ALL_TESTS = [
     "//pkg/sql/opt/testutils/opttester:opttester_test",
     "//pkg/sql/opt/testutils/testcat:testcat_test",
     "//pkg/sql/opt/testutils:testutils_test",
+    "//pkg/sql/opt/workloadindexrec:workloadindexrec_test",
     "//pkg/sql/opt/xform:xform_test",
     "//pkg/sql/opt:opt_test",
     "//pkg/sql/parser:parser_disallowed_imports_test",
@@ -1892,6 +1893,8 @@ GO_TARGETS = [
     "//pkg/sql/opt/testutils/testexpr:testexpr",
     "//pkg/sql/opt/testutils:testutils",
     "//pkg/sql/opt/testutils:testutils_test",
+    "//pkg/sql/opt/workloadindexrec:workloadindexrec",
+    "//pkg/sql/opt/workloadindexrec:workloadindexrec_test",
     "//pkg/sql/opt/xform:xform",
     "//pkg/sql/opt/xform:xform_test",
     "//pkg/sql/opt:opt",

pkg/ccl/backupccl/datadriven_test.go

@@ -848,7 +848,6 @@ func runTestDataDriven(t *testing.T, testFilePathFromWorkspace string) {
 			err = ds.getSQLDB(t, cluster, user).QueryRow(filePathQuery).Scan(&filePath)
 			require.NoError(t, err)
 			fullPath := filepath.Join(ds.getIODir(t, cluster), parsedURI.Path, filePath)
-			print(fullPath)
 			data, err := os.ReadFile(fullPath)
 			require.NoError(t, err)
 			data[20] ^= 1

pkg/ccl/backupccl/restore_job.go

@@ -2375,7 +2375,7 @@ func prefetchDescriptors(
 		if got[i].GetVersion() != expVersion[id] {
 			return nstree.Catalog{}, errors.Errorf(
 				"version mismatch for descriptor %d, expected version %d, got %v",
-				got[i].GetID(), got[i].GetVersion(), expVersion[id],
+				got[i].GetID(), expVersion[id], got[i].GetVersion(),
 			)
 		}
 		all.UpsertDescriptor(got[i])

pkg/cmd/roachtest/tests/cdc.go

@@ -1327,7 +1327,10 @@ func registerCDC(r registry.Registry) {
 			ct := newCDCTester(ctx, t, c)
 			defer ct.Close()
 
-			ct.runTPCCWorkload(tpccArgs{warehouses: 1})
+			// Run tpcc workload for tiny bit.  Roachtest monitor does not
+			// like when there are no tasks that were started with the monitor
+			// (This can be removed once #108530 resolved).
+			ct.runTPCCWorkload(tpccArgs{warehouses: 1, duration: "30s"})
 
 			kafkaNode := ct.kafkaSinkNode()
 			kafka := kafkaManager{

pkg/kv/kvserver/api.proto

@@ -89,6 +89,18 @@ message GetTableMetricsResponse{
   repeated storage.enginepb.SSTableMetricsInfo table_metrics = 1 [(gogoproto.nullable) = false, (gogoproto.customname) = "TableMetrics"];
 }
 
+// ScanStorageInternalKeysRequest retrieves metrics about keys within a range belonging
+// to a particular node and store.
+message ScanStorageInternalKeysRequest {
+  StoreRequestHeader header = 1 [(gogoproto.nullable) = false, (gogoproto.embed) = true];
+  roachpb.Span span = 2 [(gogoproto.nullable) = false];
+  int64 MegabytesPerSecond = 3;
+}
+
+message ScanStorageInternalKeysResponse {
+  repeated storage.enginepb.StorageInternalKeysMetrics advanced_metrics = 1 [(gogoproto.nullable) = false, (gogoproto.customname) = "AdvancedPebbleMetrics"];
+}
+
 message CompactEngineSpanResponse {
 }
 

pkg/kv/kvserver/storage_engine_client.go

@@ -76,6 +76,33 @@ func (c *StorageEngineClient) GetTableMetrics(
 	return resp.TableMetrics, nil
 }
 
+// ScanStorageInternalKeys is a tree.ScanStorageInternalKeys
+func (c *StorageEngineClient) ScanStorageInternalKeys(
+	ctx context.Context, nodeID, storeID int32, startKey, endKey []byte, megabytesPerSecond int64,
+) ([]enginepb.StorageInternalKeysMetrics, error) {
+	conn, err := c.nd.Dial(ctx, roachpb.NodeID(nodeID), rpc.DefaultClass)
+	if err != nil {
+		return []enginepb.StorageInternalKeysMetrics{}, errors.Wrapf(err, "could not dial node ID %d", nodeID)
+	}
+
+	client := NewPerStoreClient(conn)
+	req := &ScanStorageInternalKeysRequest{
+		StoreRequestHeader: StoreRequestHeader{
+			NodeID:  roachpb.NodeID(nodeID),
+			StoreID: roachpb.StoreID(storeID),
+		},
+		Span:               roachpb.Span{Key: roachpb.Key(startKey), EndKey: roachpb.Key(endKey)},
+		MegabytesPerSecond: megabytesPerSecond,
+	}
+
+	resp, err := client.ScanStorageInternalKeys(ctx, req)
+
+	if err != nil {
+		return []enginepb.StorageInternalKeysMetrics{}, err
+	}
+	return resp.AdvancedPebbleMetrics, nil
+}
+
 // SetCompactionConcurrency is a tree.CompactionConcurrencyFunc.
 func (c *StorageEngineClient) SetCompactionConcurrency(
 	ctx context.Context, nodeID, storeID int32, compactionConcurrency uint64,

pkg/kv/kvserver/storage_services.proto

@@ -44,5 +44,6 @@ service PerReplica {
 service PerStore {
     rpc CompactEngineSpan(cockroach.kv.kvserver.CompactEngineSpanRequest) returns (cockroach.kv.kvserver.CompactEngineSpanResponse) {}
     rpc GetTableMetrics(cockroach.kv.kvserver.GetTableMetricsRequest) returns (cockroach.kv.kvserver.GetTableMetricsResponse) {}
+    rpc ScanStorageInternalKeys(cockroach.kv.kvserver.ScanStorageInternalKeysRequest) returns (cockroach.kv.kvserver.ScanStorageInternalKeysResponse) {}
     rpc SetCompactionConcurrency(cockroach.kv.kvserver.CompactionConcurrencyRequest) returns (cockroach.kv.kvserver.CompactionConcurrencyResponse) {}
 }

pkg/kv/kvserver/stores_server.go

@@ -173,6 +173,24 @@ func (is Server) GetTableMetrics(
 	return resp, err
 }
 
+func (is Server) ScanStorageInternalKeys(
+	ctx context.Context, req *ScanStorageInternalKeysRequest,
+) (*ScanStorageInternalKeysResponse, error) {
+	resp := &ScanStorageInternalKeysResponse{}
+	err := is.execStoreCommand(ctx, req.StoreRequestHeader,
+		func(ctx context.Context, s *Store) error {
+			metrics, err := s.TODOEngine().ScanStorageInternalKeys(req.Span.Key, req.Span.EndKey, req.MegabytesPerSecond)
+
+			if err != nil {
+				return err
+			}
+
+			resp.AdvancedPebbleMetrics = metrics
+			return nil
+		})
+	return resp, err
+}
+
 // SetCompactionConcurrency implements PerStoreServer. It changes the compaction
 // concurrency of a store. While SetCompactionConcurrency is safe for concurrent
 // use, it adds uncertainty about the compaction concurrency actually set on

pkg/server/server_sql.go

@@ -961,16 +961,17 @@ func newSQLServer(ctx context.Context, cfg sqlServerArgs) (*SQLServer, error) {
 		ClientCertExpirationCache: security.NewClientCertExpirationCache(
 			ctx, cfg.Settings, cfg.stopper, &timeutil.DefaultTimeSource{}, rootSQLMemoryMonitor,
 		),
-		RootMemoryMonitor:         rootSQLMemoryMonitor,
-		TestingKnobs:              sqlExecutorTestingKnobs,
-		CompactEngineSpanFunc:     storageEngineClient.CompactEngineSpan,
-		CompactionConcurrencyFunc: storageEngineClient.SetCompactionConcurrency,
-		GetTableMetricsFunc:       storageEngineClient.GetTableMetrics,
-		TraceCollector:            traceCollector,
-		TenantUsageServer:         cfg.tenantUsageServer,
-		KVStoresIterator:          cfg.kvStoresIterator,
-		InspectzServer:            cfg.inspectzServer,
-		RangeDescIteratorFactory:  cfg.rangeDescIteratorFactory,
+		RootMemoryMonitor:           rootSQLMemoryMonitor,
+		TestingKnobs:                sqlExecutorTestingKnobs,
+		CompactEngineSpanFunc:       storageEngineClient.CompactEngineSpan,
+		CompactionConcurrencyFunc:   storageEngineClient.SetCompactionConcurrency,
+		GetTableMetricsFunc:         storageEngineClient.GetTableMetrics,
+		ScanStorageInternalKeysFunc: storageEngineClient.ScanStorageInternalKeys,
+		TraceCollector:              traceCollector,
+		TenantUsageServer:           cfg.tenantUsageServer,
+		KVStoresIterator:            cfg.kvStoresIterator,
+		InspectzServer:              cfg.inspectzServer,
+		RangeDescIteratorFactory:    cfg.rangeDescIteratorFactory,
 		SyntheticPrivilegeCache: syntheticprivilegecache.New(
 			cfg.Settings, cfg.stopper, cfg.db,
 			serverCacheMemoryMonitor.MakeBoundAccount(),