[BACKPORT 2024.1][#20616] YSQL: PG view for YCQL metrics

Summary:
Original commits:
- ce7d880 / D31705
- 0eab5c4 / D33565

**ycql_stat_statements**
Added a new YSQL view for YCQL statement metrics so that it can be joined with YCQL wait events in yb_active_universe_history table. The columns of this table are similar to what pg_stat_statements columns are.

| Column | Type | Description |
| queryid | int8 | Hash code to identify identical normalized queries. |
| query | text | Text of a representative statement |
| is_prepared | bool | Prepared statement or unprepared query |
| calls | int8 | Number of times the statement was executed |
| total_time | float8 | Total time spent executing the statement, in milliseconds |
| min_time | float8 | Minimum time spent executing the statement, in milliseconds |
| max_time | float8 | Maximum time spent executing the statement, in milliseconds |
| mean_time | float8 | Mean time spent executing the statement, in milliseconds |
| stddev_time | float8 | Population standard deviation of time spent executing the statement, in milliseconds |

This view is added in an extension - yb_ycql_utils, which is not enabled by default. The extension has to be enabled before the view can be used. If ycql is disabled, this view will not return any data.

```
yugabyte=# create extension yb_ycql_utils;
CREATE EXTENSION
yugabyte=# select * from ycql_stat_statements;
       queryid        |                     query                      | is_prepared | calls | total_time | min_time | max_time  | mean_time |    stddev_time
----------------------+------------------------------------------------+-------------+-------+------------+----------+-----------+-----------+--------------------
  5685694520060019787 | SELECT * FROM system.peers_v2                  | f           |     2 |   23.15232 | 6.606628 | 16.545692 |  11.57616 |           4.969532
 -6069774349418914791 | SELECT * FROM system.local WHERE key='local'   | f           |     2 |  11.241228 | 0.516517 | 10.724711 |  5.620614 |           5.104097
 -4656374775045675304 | SELECT * FROM system_schema.tables             | f           |     2 |   5.773887 | 1.937739 |  3.836148 | 2.8869435 |          0.9492045
   477300852678741015 | SELECT * FROM system.peers                     | f           |     2 |   1.867352 | 0.709899 |  1.157453 |  0.933676 |           0.223777
  6930116125454979846 | SELECT * FROM system_schema.views              | f           |     2 |   3.671581 | 1.320594 |  2.350987 | 1.8357905 |          0.5151965
 -1896671018756022147 | SELECT * FROM system_schema.functions          | f           |     2 |   3.156819 | 1.502305 |  1.654514 | 1.5784095 |          0.0761045
  1413414562899452953 | SELECT * FROM system_schema.indexes            | f           |     2 |    4.56578 | 1.694535 |  2.871245 |   2.28289 |           0.588355
 -3220527242581763013 | SELECT * FROM system_schema.keyspaces          | f           |     2 |   4.304142 | 1.504853 |  2.799289 |  2.152071 |           0.647218
 -2674195503457906853 | SELECT * FROM system_schema.aggregates         | f           |     2 |    2.64328 | 1.123457 |  1.519823 |   1.32164 |           0.198183
 -2256941656319582329 | SELECT * FROM system_schema.columns            | f           |     2 |  11.021614 | 3.881885 |  7.139729 |  5.510807 |           1.628922
  2381078306997755979 | SELECT * FROM system_schema.triggers           | f           |     2 |   2.722613 | 1.216926 |  1.505687 | 1.3613065 |          0.1443805
  6202644009413539627 | SELECT * FROM system_schema.types              | f           |     2 |   3.718101 | 1.490935 |  2.227166 | 1.8590505 |          0.3681155
 -7196980672916816745 | select * from system.local where key = 'local' | f           |     2 |   1.169078 | 0.524634 |  0.644444 |  0.584539 | 0.0599050000000001
(13 rows)

```
**Upgrade/Rollback safety:**
The proto file changes are for the communication between local tserver and PG. There is no communication between PG and remote tserver and hence the diff is upgrade safe.
Jira: DB-9617

Test Plan: ./yb_build.sh  release --java-test 'org.yb.pgsql.TestPgViewYCQLStats'

Reviewers: asaha, amitanand, jason

Reviewed By: jason

Subscribers: bogdan, ybase, yql, mihnea

Tags: #jenkins-ready

Differential Revision: https://phorge.dev.yugabyte.com/D33917

java/yb-pgsql/pom.xml

@@ -62,6 +62,10 @@
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>com.yugabyte</groupId>
+      <artifactId>cassandra-driver-core</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.apache.directory.server</groupId>
       <artifactId>apacheds-all</artifactId>

java/yb-pgsql/src/test/java/org/yb/pgsql/TestPgRegressContribYbYcqlUtils.java

@@ -0,0 +1,36 @@
+// Copyright (c) Yugabyte, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+// in compliance with the License.  You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software distributed under the License
+// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+// or implied.  See the License for the specific language governing permissions and limitations
+// under the License.
+//
+package org.yb.pgsql;
+
+import java.io.File;
+import java.util.Map;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.yb.client.TestUtils;
+import org.yb.YBTestRunner;
+
+@RunWith(value=YBTestRunner.class)
+public class TestPgRegressContribYbYcqlUtils extends BasePgSQLTest {
+
+  @Override
+  public int getTestMethodTimeoutSec() {
+    return 600;
+  }
+
+  @Test
+  public void schedule() throws Exception {
+    runPgRegressTest(new File(TestUtils.getBuildRootDir(), "postgres_build/contrib/yb_ycql_utils"),
+                     "yb_schedule");
+  }
+}

java/yb-pgsql/src/test/java/org/yb/pgsql/TestPgViewYCQLStats.java

@@ -0,0 +1,151 @@
+// Copyright (c) YugabyteDB, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+// in compliance with the License.  You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software distributed under the License
+// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+// or implied.  See the License for the specific language governing permissions and limitations
+// under the License.
+//
+
+package org.yb.pgsql;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import static org.yb.AssertionWrappers.*;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+
+import java.util.Collections;
+import java.util.Map;
+import com.datastax.driver.core.*;
+import com.datastax.driver.core.QueryOptions;
+import com.datastax.driver.core.SocketOptions;
+import com.datastax.driver.core.ConsistencyLevel;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.yb.util.YBTestRunnerNonTsanOnly;
+
+import com.yugabyte.util.PSQLException;
+
+@RunWith(value = YBTestRunnerNonTsanOnly.class)
+public class TestPgViewYCQLStats extends BasePgSQLTest {
+
+  private static final Logger LOG = LoggerFactory.getLogger(TestPgViewYCQLStats.class);
+  protected int cqlClientTimeoutMs = 120 * 1000;
+
+  /** Convenient default cluster for tests to use, cleaned after each test. */
+  protected Cluster cluster;
+
+  /** Convenient default session for tests to use, cleaned after each test. */
+  protected Session session;
+
+  @Override
+  protected void resetSettings() {
+    super.resetSettings();
+    startCqlProxy = true;
+  }
+
+  public Cluster.Builder getDefaultClusterBuilder() {
+    // Set default consistency level to strong consistency
+    QueryOptions queryOptions = new QueryOptions();
+    queryOptions.setConsistencyLevel(ConsistencyLevel.YB_STRONG);
+    // Set a long timeout for CQL queries since build servers might be really slow (especially Mac
+    // Mini).
+    SocketOptions socketOptions = new SocketOptions();
+    socketOptions.setReadTimeoutMillis(cqlClientTimeoutMs);
+    socketOptions.setConnectTimeoutMillis(cqlClientTimeoutMs);
+    return Cluster.builder()
+              .addContactPointsWithPorts(miniCluster.getCQLContactPoints())
+              .withQueryOptions(queryOptions)
+              .withSocketOptions(socketOptions);
+  }
+
+  /** Create a CQL client  */
+  public void setUpCqlClient() throws Exception {
+    LOG.info("setUpCqlClient is running");
+
+    if (miniCluster == null) {
+      final String errorMsg =
+          "Mini-cluster must already be running by the time setUpCqlClient is invoked";
+      LOG.error(errorMsg);
+      throw new RuntimeException(errorMsg);
+    }
+
+    try {
+      cluster = getDefaultClusterBuilder().build();
+      session = cluster.connect();
+      LOG.info("Connected to cluster: " + cluster.getMetadata().getClusterName());
+    } catch (Exception ex) {
+      LOG.error("Error while setting up a CQL client", ex);
+      throw ex;
+    }
+
+    final int numTServers = miniCluster.getTabletServers().size();
+    final int expectedNumPeers = Math.max(0, Math.min(numTServers - 1, 2));
+    LOG.info("Waiting until system.peers contains at least " + expectedNumPeers + " entries (" +
+        "number of tablet servers: " + numTServers + ")");
+    int attemptsMade = 0;
+    boolean waitSuccessful = false;
+
+    while (attemptsMade < 30) {
+      int numPeers = session.execute("SELECT peer FROM system.peers").all().size();
+      if (numPeers >= expectedNumPeers) {
+        waitSuccessful = true;
+        break;
+      }
+      LOG.info("system.peers still contains only " + numPeers + " entries, waiting");
+      attemptsMade++;
+      Thread.sleep(1000);
+    }
+    if (waitSuccessful) {
+      LOG.info("Succeeded waiting for " + expectedNumPeers + " peers to show up in system.peers");
+    } else {
+      LOG.warn("Timed out waiting for " + expectedNumPeers + " peers to show up in system.peers");
+    }
+  }
+
+  @Test
+  public void testYCQLStats() throws Exception {
+    setUpCqlClient();
+    session.execute("create keyspace k1").one();
+    session.execute("use k1").one();
+    session.execute("create table table1(col1 int, col2 int, primary key (col1))").one();
+    PreparedStatement ps = session.prepare("select col1 from table1");
+    for (int i = 0; i < 10; i++) {
+      session.execute(ps.bind()).iterator();
+    }
+    for (int i = 0; i < 10; i++) {
+      session.execute("select col2 from table1").one();
+    }
+    try (Statement statement = connection.createStatement()) {
+      statement.execute("create extension yb_ycql_utils");
+      int count_prepared = getSingleRow(statement, "SELECT COUNT(*) FROM ycql_stat_statements"  +
+          " WHERE is_prepared='t' and query like '%select col1%'").getLong(0).intValue();
+      int count_unprepared = getSingleRow(statement, "SELECT COUNT(*) FROM ycql_stat_statements"  +
+          " WHERE is_prepared='f' and query like '%select col2%'").getLong(0).intValue();
+
+      assertEquals(count_prepared, 1);
+      assertEquals(count_unprepared, 1);
+
+    }
+    session.execute("drop table table1").one();
+  }
+
+  public void cleanUpAfter() throws Exception {
+    if (session != null) {
+      session.close();
+    }
+    session = null;
+    if (cluster != null) {
+      cluster.close();
+    }
+    cluster = null;
+    super.cleanUpAfter();
+  }
+}

src/postgres/contrib/Makefile

@@ -53,7 +53,8 @@ SUBDIRS = \
 		unaccent	\
 		vacuumlo    \
 		yb_pg_metrics \
-		yb_test_extension
+		yb_test_extension \
+		yb_ycql_utils
 
 ifeq ($(with_openssl),yes)
 SUBDIRS += sslinfo

src/postgres/contrib/yb_ycql_utils/Makefile

@@ -0,0 +1,21 @@
+# contrib/yb_ycql_utils/Makefile
+
+PGFILEDESC = "yb_ycql_utils - An extension for fetching YCQL data in PG side."
+
+MODULE_big = yb_ycql_utils
+OBJS = yb_ycql_utils.o $(WIN32RES)
+
+EXTENSION = yb_ycql_utils
+DATA = yb_ycql_utils--1.0.sql
+SHLIB_LINK += -L$(YB_BUILD_ROOT)/lib -lyb_pggate -lyb_pggate_util
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = contrib/yb_ycql_utils
+top_builddir = ../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif

src/postgres/contrib/yb_ycql_utils/expected/yb_ycql_utils_basic.out

@@ -0,0 +1,10 @@
+create extension yb_ycql_utils;
+select ycql_stat_statements();
+ ycql_stat_statements 
+----------------------
+(0 rows)
+
+select * from ycql_stat_statements;
+ queryid | query | is_prepared | calls | total_time | min_time | max_time | mean_time | stddev_time 
+---------+-------+-------------+-------+------------+----------+----------+-----------+-------------
+(0 rows)

src/postgres/contrib/yb_ycql_utils/sql/yb_ycql_utils_basic.sql

@@ -0,0 +1,3 @@
+create extension yb_ycql_utils;
+select ycql_stat_statements();
+select * from ycql_stat_statements;

src/postgres/contrib/yb_ycql_utils/yb_cql_utils.conf

@@ -0,0 +1 @@
+shared_preload_libraries = 'yb_ycql_utils'

src/postgres/contrib/yb_ycql_utils/yb_schedule

@@ -0,0 +1,3 @@
+#contrib/yb_ycql_utils/yb_schedule
+
+test: yb_ycql_utils_basic

src/postgres/contrib/yb_ycql_utils/yb_ycql_utils--1.0.sql

@@ -0,0 +1,22 @@
+/* contrib/yb_ycql_utils/yb_ycql_utils--1.0.sql */
+
+CREATE FUNCTION ycql_stat_statements(
+    OUT queryid             int8,
+    OUT query               text,
+    OUT is_prepared         bool,
+    OUT calls               int8,
+    OUT total_time          float8,
+    OUT min_time            float8,
+    OUT max_time            float8,
+    OUT mean_time           float8,
+    OUT stddev_time         float8
+)
+RETURNS SETOF record
+AS 'MODULE_PATHNAME'
+LANGUAGE C STRICT VOLATILE PARALLEL SAFE;
+
+CREATE VIEW ycql_stat_statements AS
+    SELECT *
+    FROM ycql_stat_statements();
+
+GRANT SELECT ON ycql_stat_statements TO PUBLIC;

src/postgres/contrib/yb_ycql_utils/yb_ycql_utils.c

@@ -0,0 +1,84 @@
+#include "postgres.h"
+
+#include "funcapi.h"
+#include "postmaster/syslogger.h"
+#include "pg_yb_utils.h"
+#include "utils/builtins.h"
+#include "yb/yql/pggate/ybc_pggate.h"
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(ycql_stat_statements);
+
+Datum
+ycql_stat_statements(PG_FUNCTION_ARGS)
+{
+	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
+	TupleDesc	tupdesc;
+	Tuplestorestate *tupstore;
+	MemoryContext per_query_ctx;
+	MemoryContext oldcontext;
+	int	i;
+#define YCQL_STAT_STATEMENTS_COLS 9
+
+	/* check to see if caller supports us returning a tuplestore */
+	if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("set-valued function called in context that cannot accept a set")));
+
+	if (!(rsinfo->allowedModes & SFRM_Materialize))
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("materialize mode required, but it is not " \
+						"allowed in this context")));
+
+	/* Switch context to construct returned data structures */
+	per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
+	oldcontext = MemoryContextSwitchTo(per_query_ctx);
+
+	/* Build a tuple descriptor */
+	if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
+		ereport(ERROR,
+				(errmsg_internal("return type must be a row type")));
+
+	tupstore = tuplestore_begin_heap(true, false, work_mem);
+	rsinfo->returnMode = SFRM_Materialize;
+	rsinfo->setResult = tupstore;
+	rsinfo->setDesc = tupdesc;
+
+	MemoryContextSwitchTo(oldcontext);
+
+	YCQLStatementStats *stat_list = NULL;
+	size_t num_stats = 0;
+	HandleYBStatus(YBCYcqlStatementStats(&stat_list, &num_stats));
+
+	for (i = 0; i < num_stats; ++i)
+	{
+		YCQLStatementStats *stats = (YCQLStatementStats *) stat_list + i;
+		Datum values[YCQL_STAT_STATEMENTS_COLS];
+		bool nulls[YCQL_STAT_STATEMENTS_COLS];
+		memset(values, 0, sizeof(values));
+		memset(nulls, 0, sizeof(nulls));
+
+		values[0] = Int64GetDatum(stats->queryid);
+		values[1] = CStringGetTextDatum(stats->query);
+		values[2] = BoolGetDatum(stats->is_prepared);
+		values[3] = Int64GetDatum(stats->calls);
+		values[4] = Float8GetDatumFast(stats->total_time);
+		values[5] = Float8GetDatumFast(stats->min_time);
+		values[6] = Float8GetDatumFast(stats->max_time);
+		values[7] = Float8GetDatumFast(stats->mean_time);
+		values[8] = Float8GetDatumFast(stats->stddev_time);
+
+		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
+	}
+#undef YCQL_STAT_STATEMENTS_COLS
+
+	/* clean up and return the tuplestore */
+	tuplestore_donestoring(tupstore);
+	if (stat_list) {
+		pfree(stat_list);
+	}
+	return (Datum) 0;
+}

src/postgres/contrib/yb_ycql_utils/yb_ycql_utils.control

@@ -0,0 +1,5 @@
+# yb_ycql_utils extension
+comment = 'An extension for fetching YCQL data in PG side'
+default_version = '1.0'
+module_pathname = '$libdir/yb_ycql_utils'
+relocatable = true

src/yb/master/master_tserver.cc

@@ -29,6 +29,7 @@
 #include "yb/tablet/tablet_peer.h"
 
 #include "yb/tserver/tserver.pb.h"
+#include "yb/tserver/pg_client.pb.h"
 
 #include "yb/util/atomic.h"
 #include "yb/util/metric_entity.h"
@@ -191,5 +192,11 @@ void MasterTabletServer::ClearAllMetaCachesOnServer() {
   client()->ClearAllMetaCachesOnServer();
 }
 
+Status MasterTabletServer::YCQLStatementStats(const tserver::PgYCQLStatementStatsRequestPB& req,
+    tserver::PgYCQLStatementStatsResponsePB* resp) const {
+  LOG(FATAL) << "Unexpected call of YCQLStatementStats()";
+  return Status::OK();
+}
+
 } // namespace master
 } // namespace yb