[BACKPORT 2.16][#16135] YCQL: Use Metadata Cache in IsYBTableAltered

Summary:
Before the fix `CQLStatement::IsYBTableAltered` always uses RPC to the Master
to get up-to-date Table Schema Version.
Now it's controlled by the new G-flag: `cql_use_metadata_cache_for_schema_version_check`.
If the G-flag is `true` the internal TS Table Metadata Cache is used to get the Table
Schema Version instead of the (potentially slow) RPC.
The new G-flag `cql_use_metadata_cache_for_schema_version_check` is `false` by default.

Note that `ParseTree::IsYBTableAltered` is used in 2 code points:
1. In `Executor::ProcessStatementStatus`
2. In `CQLStatement::IsYBTableAltered`
`CQLStatement::IsYBTableAltered` is called from the `PREPARE` request handling - in
`CQLServiceImpl::AllocatePreparedStatement`.
The new flag affects only `CQLStatement::IsYBTableAltered`.
`Executor::ProcessStatementStatus` remains unchanged.

Original diff: D22686 / 9bd9bb8
GH link: 9bd9bb8

Test Plan:
New tests:
ybd --java-test org.yb.cql.TestPrepareExecute#testAlterAdd_UseMetadataCache
ybd --java-test org.yb.cql.TestPrepareExecute#testAlterAdd_MetadataInExecResp_UseMetadataCache

Other tests for PREPARE/EXECUTE:
ybd --java-test org.yb.cql.TestPrepareExecute#testAlterAdd
ybd --java-test org.yb.cql.TestPrepareExecute#testAlterAdd_MetadataInExecResp
ybd --java-test org.yb.cql.TestPrepareExecute#testAlterDropAdd
ybd --java-test org.yb.cql.TestPrepareExecute#testAlterDropAddSameSizeType
ybd --java-test org.yb.cql.TestPrepareExecute#testMultiThreadedAlterAdd --tp 1
ybd --java-test org.yb.cql.TestPrepareExecute#testMultiThreadedAlterDropAdd
ybd --java-test org.yb.cql.TestPrepareExecute#testRecreateTable

ybd --cxx-test cql-test --gtest_filter CqlTest.AlteredPrepare
ybd --cxx-test cql-test --gtest_filter CqlTest.AlteredPrepare_MetadataInExecResp
ybd --cxx-test cql-test --gtest_filter CqlTest.AlteredPrepareWithPaging
ybd --cxx-test cql-test --gtest_filter CqlTest.AlteredPrepareWithPaging_NoSchemaCheck
ybd --cxx-test cql-test --gtest_filter CqlTest.AlteredPrepareWithPaging_MetadataInExecResp
ybd --cxx-test cql-test --gtest_filter CqlTest.PrepareWithDropTableWithPaging
ybd --cxx-test cql-test --gtest_filter CqlTest.PrepareWithDropTableWithPaging_NoSchemaCheck

Reviewers: mihnea, stiwary, neil

Reviewed By: neil

Subscribers: yql, kgupta

Differential Revision: https://phabricator.dev.yugabyte.com/D23065

java/yb-cql/src/test/java/org/yb/cql/BaseCQLTest.java

@@ -269,14 +269,16 @@ public void tearDownAfter() throws Exception {
     afterBaseCQLTestTearDown();
   }
 
-  protected void restartClusterWithFlag(String flag, String value) throws Exception {
+  protected void restartClusterWithTSFlags(Map<String, String> tserverFlags) throws Exception {
     destroyMiniCluster();
-    createMiniCluster(
-        Collections.emptyMap(),
-        Collections.singletonMap(flag, value));
+    createMiniCluster(Collections.emptyMap(), tserverFlags);
     setUpCqlClient();
   }
 
+  protected void restartClusterWithFlag(String flag, String value) throws Exception {
+    restartClusterWithTSFlags(Collections.singletonMap(flag, value));
+  }
+
   protected void afterBaseCQLTestTearDown() throws Exception {
   }
 

java/yb-cql/src/test/java/org/yb/cql/TestPrepareExecute.java

@@ -98,8 +98,10 @@ public void testBasicPrepareExecute() throws Exception {
   }
 
   protected enum MetadataInExecResp { ON, OFF; }
+  protected enum UseMetadataCache { ON, OFF; }
 
-  protected void doTestAlterAdd(MetadataInExecResp flag) throws Exception {
+  protected void doTestAlterAdd(MetadataInExecResp inResp,
+                                UseMetadataCache useCache) throws Exception {
     // Setup table.
     setupTable("test_prepare", 0 /* num_rows */);
     session.execute(
@@ -124,14 +126,23 @@ protected void doTestAlterAdd(MetadataInExecResp flag) throws Exception {
 
     // The driver uses the incoming schema info from the CQL response with the new column.
     row = session.execute(prepared.bind()).one();
-    if (flag == MetadataInExecResp.ON) {
+    if (inResp == MetadataInExecResp.ON) {
       assertEquals(7, row.getColumnDefinitions().size());
       assertEquals("Row[1, a, 2, b, 3, c, 9]", row.toString());
     } else {
       assertEquals(6, row.getColumnDefinitions().size());
       assertEquals("Row[1, a, 2, b, 3, c]", row.toString());
     }
 
+    if (useCache == UseMetadataCache.ON) {
+      // Run EXECUTE on all TSes to reset internal Table Metadata cache.
+      final int numTServers = miniCluster.getTabletServers().size();
+      // First TS was used above.
+      for (int i = 0; i < numTServers - 1; ++i) {
+        row = session.execute(prepared.bind()).one();
+      }
+    }
+
     // Run a new "application" = new driver instance = new cluster object & connection.
     try (Session s3 = connectWithTestDefaults().getSession()) {
       s3.execute("USE " + DEFAULT_TEST_KEYSPACE);
@@ -146,14 +157,40 @@ protected void doTestAlterAdd(MetadataInExecResp flag) throws Exception {
   @Test
   public void testAlterAdd() throws Exception {
     // By default: cql_always_return_metadata_in_execute_response=false.
-    doTestAlterAdd(MetadataInExecResp.OFF);
+    //             cql_use_metadata_cache_for_schema_version_check=false.
+    doTestAlterAdd(MetadataInExecResp.OFF, UseMetadataCache.OFF);
   }
 
   @Test
   public void testAlterAdd_MetadataInExecResp() throws Exception {
     try {
+      // By default: cql_use_metadata_cache_for_schema_version_check=false.
       restartClusterWithFlag("cql_always_return_metadata_in_execute_response", "true");
-      doTestAlterAdd(MetadataInExecResp.ON);
+      doTestAlterAdd(MetadataInExecResp.ON, UseMetadataCache.OFF);
+    } finally {
+      destroyMiniCluster(); // Destroy the recreated cluster when done.
+    }
+  }
+
+  @Test
+  public void testAlterAdd_UseMetadataCache() throws Exception {
+    try {
+      // By default: cql_always_return_metadata_in_execute_response=false.
+      restartClusterWithFlag("cql_use_metadata_cache_for_schema_version_check", "true");
+      doTestAlterAdd(MetadataInExecResp.OFF, UseMetadataCache.ON);
+    } finally {
+      destroyMiniCluster(); // Destroy the recreated cluster when done.
+    }
+  }
+
+  @Test
+  public void testAlterAdd_MetadataInExecResp_UseMetadataCache() throws Exception {
+    try {
+      Map<String, String> tserverFlags = new HashMap<>();
+      tserverFlags.put("cql_always_return_metadata_in_execute_response", "true");
+      tserverFlags.put("cql_use_metadata_cache_for_schema_version_check", "true");
+      restartClusterWithTSFlags(tserverFlags);
+      doTestAlterAdd(MetadataInExecResp.ON, UseMetadataCache.ON);
     } finally {
       destroyMiniCluster(); // Destroy the recreated cluster when done.
     }

src/yb/yql/cql/cqlserver/cql_statement.cc

@@ -17,6 +17,15 @@
 
 #include <openssl/md5.h>
 
+#include "yb/util/flag_tags.h"
+
+DEFINE_RUNTIME_bool(cql_use_metadata_cache_for_schema_version_check, false,
+                    "Use the internal Table Metadata Cache in TS to check the Table "
+                    "Schema Version when processing the YCQL PREPARE query."
+                    "Use the flag with caution: with the flag enabled PREPARE works "
+                    "faster, but it may return a stale table schema.");
+TAG_FLAG(cql_use_metadata_cache_for_schema_version_check, advanced);
+
 using std::string;
 
 namespace yb {
@@ -31,6 +40,12 @@ CQLStatement::CQLStatement(
 CQLStatement::~CQLStatement() {
 }
 
+Result<bool> CQLStatement::IsYBTableAltered(ql::QLEnv* ql_env) const {
+  const ql::ParseTree& parser_tree = VERIFY_RESULT(GetParseTree());
+  const bool use_cache = FLAGS_cql_use_metadata_cache_for_schema_version_check;
+  return parser_tree.IsYBTableAltered(ql_env, use_cache);
+}
+
 ql::CQLMessage::QueryId CQLStatement::GetQueryId(const string& keyspace, const string& query) {
   unsigned char md5[MD5_DIGEST_LENGTH];
   MD5_CTX md5ctx;

src/yb/yql/cql/cqlserver/cql_statement.h

@@ -58,10 +58,7 @@ class CQLStatement : public ql::Statement {
   }
 
   // Check if the used schema version is up to date with the Master.
-  Result<bool> IsYBTableAltered(ql::QLEnv* ql_env) const {
-    const ql::ParseTree& parser_tree = VERIFY_RESULT(GetParseTree());
-    return parser_tree.IsYBTableAltered(ql_env);
-  }
+  Result<bool> IsYBTableAltered(ql::QLEnv* ql_env) const;
 
   // Return the query id of a statement.
   static ql::CQLMessage::QueryId GetQueryId(const std::string& keyspace, const std::string& query);

src/yb/yql/cql/ql/exec/executor.cc

@@ -2495,7 +2495,8 @@ Status Executor::ProcessStatementStatus(const ParseTree& parse_tree, const Statu
         errcode == ErrorCode::TYPE_NOT_FOUND) {
       if (errcode == ErrorCode::INVALID_ARGUMENTS) {
         // Check the table schema is up-to-date.
-        const Result<bool> is_altered_res = parse_tree.IsYBTableAltered(ql_env_);
+        const Result<bool> is_altered_res =
+            parse_tree.IsYBTableAltered(ql_env_, false /* use_cache */);
         // The table is not available if (!is_altered_res.ok()).
         // Usually it happens if the table was deleted.
         if (is_altered_res.ok() && !(*is_altered_res)) {

src/yb/yql/cql/ql/ptree/parse_tree.cc

@@ -30,6 +30,8 @@
 
 #include "yb/client/table.h"
 
+#include "yb/util/logging.h"
+
 namespace yb {
 namespace ql {
 
@@ -125,21 +127,23 @@ Result<SchemaVersion> ParseTree::GetYBTableSchemaVersion() const {
   return table->schema().version();
 }
 
-Result<bool> ParseTree::IsYBTableAltered(QLEnv *ql_env) const {
+Result<bool> ParseTree::IsYBTableAltered(QLEnv *ql_env, bool use_cache) const {
   // We check the Main Table schema version even for "Index Only scan" & "Index + Table scan". This
   // is safe to do since if Index schema was changed - the Main Table schema version is also
   // incremented.
   const shared_ptr<const client::YBTable> table = GetYBTableFromTreeNode(root_.get());
   SCHECK(table, IllegalState, "Table missing");
   const SchemaVersion current_schema_ver = table->schema().version();
-  const SchemaVersion updated_schema_ver = VERIFY_RESULT(
-      DCHECK_NOTNULL(ql_env)->GetUpToDateTableSchemaVersion(table->name()));
+  DCHECK_ONLY_NOTNULL(ql_env);
+  const SchemaVersion updated_schema_ver =
+      VERIFY_RESULT(use_cache ? ql_env->GetCachedTableSchemaVersion(table->id())
+                              : ql_env->GetUpToDateTableSchemaVersion(table->id()));
 
   if (updated_schema_ver == current_schema_ver) {
     return false;
   } else {
     // Clean-up the internal cache for the stale table.
-    ql_env->RemoveCachedTableDesc(table->name());
+    ql_env->RemoveCachedTableDesc(table->id());
     return true;
   }
 }

src/yb/yql/cql/ql/ptree/parse_tree.h

@@ -104,7 +104,7 @@ class ParseTree {
   Result<SchemaVersion> GetYBTableSchemaVersion() const;
 
   // Check if the used schema version is not in sync with the Master.
-  Result<bool> IsYBTableAltered(QLEnv *ql_env) const;
+  Result<bool> IsYBTableAltered(QLEnv *ql_env, bool use_cache) const;
 
   // Add table to the set of tables used during semantic analysis.
   void AddAnalyzedTable(const client::YBTableName& table_name);

src/yb/yql/cql/ql/util/ql_env.cc

@@ -148,15 +148,17 @@ shared_ptr<YBTable> QLEnv::GetTableDesc(const TableId& table_id, bool* cache_use
   return yb_table;
 }
 
-Result<SchemaVersion> QLEnv::GetUpToDateTableSchemaVersion(const YBTableName& table_name) {
-  shared_ptr<YBTable> yb_table;
-  RETURN_NOT_OK(client_->OpenTable(table_name, &yb_table));
+Result<SchemaVersion> QLEnv::GetCachedTableSchemaVersion(const TableId& table_id) {
+  bool cache_used = false;
+  const shared_ptr<YBTable> yb_table = GetTableDesc(table_id, &cache_used);
+  SCHECK_FORMAT(yb_table, NotFound, "Cannot get table $0 from cache", table_id);
+  return yb_table->schema().version();
+}
 
-  if (yb_table) {
-    return yb_table->schema().version();
-  } else {
-    return STATUS_SUBSTITUTE(NotFound, "Cannot get table $0", table_name.ToString());
-  }
+Result<SchemaVersion> QLEnv::GetUpToDateTableSchemaVersion(const TableId& table_id) {
+  // Force update the metadata cache.
+  RemoveCachedTableDesc(table_id);
+  return GetCachedTableSchemaVersion(table_id);
 }
 
 shared_ptr<QLType> QLEnv::GetUDType(const std::string& keyspace_name,

src/yb/yql/cql/ql/util/ql_env.h

@@ -67,8 +67,8 @@ class QLEnv {
   virtual Status DeleteIndexTable(const client::YBTableName& name,
                                           client::YBTableName* indexed_table_name);
 
-  virtual Result<SchemaVersion> GetUpToDateTableSchemaVersion(
-      const client::YBTableName& table_name);
+  virtual Result<SchemaVersion> GetCachedTableSchemaVersion(const TableId& table_id);
+  virtual Result<SchemaVersion> GetUpToDateTableSchemaVersion(const TableId& table_id);
 
   virtual std::shared_ptr<client::YBTable> GetTableDesc(const client::YBTableName& table_name,
                                                         bool* cache_used);