[BACKPORT 2.23.0][#23543] docdb: Get colocated schema version correct…

…ly in WriteQuery

Summary:
Original commit: b9faf18 / D37634
For each write op we apply, we check that the schema version of the table is less than or equal to the schema of the write op, but the code to get the schema version of a table was not taking into account whether the table was colocated. For colocated tables, we were just checking against the schema version of the parent table which was always 0 until DB cloning which increased it to 1 during ImportSnapshot (which caused writes with schema_version 0 to fail). This diff changes the WriteQuery code to get the appropriate schema version if the table is colocated.
Jira: DB-12461

Test Plan:
`./yb_build.sh release --cxx-test integration-tests_minicluster-snapshot-test --gtest_filter *CreateTableAfterClone*`
Jenkins: urgent

Reviewers: yyan, mhaddad

Reviewed By: yyan

Subscribers: ybase

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

src/yb/integration-tests/minicluster-snapshot-test.cc

@@ -952,11 +952,25 @@ TEST_F(
   ASSERT_EQ(rows_t2[0], kRowT2);
 }
 
-TEST_F(PgCloneColocationTest, YB_DISABLE_TEST_IN_SANITIZERS(CreateTableAfterClone)) {
+TEST_P(PgCloneTestWithColocatedDBParam, YB_DISABLE_TEST_IN_SANITIZERS(CreateTableAfterClone)) {
+  ASSERT_OK(source_conn_->ExecuteFormat("INSERT INTO t1 VALUES (1, 1)"));
+
+  auto clone_time = ASSERT_RESULT(GetCurrentTime()).ToInt64();
+  ASSERT_OK(source_conn_->Execute("CREATE TABLE t2 (k int, v1 int)"));
+
+  // Clone before t2 was created and test that we can recreate t2.
   ASSERT_OK(source_conn_->ExecuteFormat(
-      "CREATE DATABASE $0 TEMPLATE $1", kTargetNamespaceName1, kSourceNamespaceName));
+      "CREATE DATABASE $0 TEMPLATE $1 AS OF $2", kTargetNamespaceName1, kSourceNamespaceName,
+      clone_time));
   auto target_conn = ASSERT_RESULT(ConnectToDB(kTargetNamespaceName1));
   ASSERT_OK(target_conn.Execute("CREATE TABLE t2 (k int, v1 int)"));
+
+  // Should be able to create new tables and indexes and insert into all tables.
+  ASSERT_OK(target_conn.Execute("CREATE INDEX i1 on t1(value)"));
+  ASSERT_OK(target_conn.Execute("INSERT INTO t1 VALUES (2, 2)"));
+  ASSERT_OK(target_conn.Execute("INSERT INTO t2 VALUES (1, 1)"));
+  ASSERT_OK(target_conn.Execute("CREATE TABLE t3 (k int, v1 int)"));
+  ASSERT_OK(target_conn.Execute("INSERT INTO t3 VALUES (1, 1)"));
 }
 
 }  // namespace master

src/yb/master/catalog_manager_ext.cc

@@ -1998,6 +1998,8 @@ Status CatalogManager::RepartitionTable(const scoped_refptr<TableInfo> table,
         }
         tablet->mutable_metadata()->mutable_dirty()->pb.set_colocated(table->colocated());
         new_tablets.push_back(tablet);
+        LOG(INFO) << Format("Created tablet $0 to replace tablet $1 in repartitioning of table $2",
+                            tablet->id(), source_tablet_id, table->id());
       }
 
       // Add tablets to catalog manager tablet_map_. This should be safe to do after creating
@@ -2083,7 +2085,7 @@ Status CatalogManager::RepartitionTable(const scoped_refptr<TableInfo> table,
   // The create tablet requests should be handled by bg tasks which find the PREPARING tablets after
   // commit.
 
-  // Update the colocated tablet in the tablegroup manager.
+  // Update the tablegroup manager to point to the new colocated tablet instead of the old one.
   if (table->colocated()) {
     SharedLock l(mutex_);
     SCHECK(

src/yb/tablet/write_query.cc

@@ -317,7 +317,20 @@ void WriteQuery::Finished(WriteOperation* operation, const Status& status) {
 
   for (const auto& sv : operation->request()->write_batch().table_schema_version()) {
     if (!status.IsAborted()) {
-      CHECK_LE(metadata.schema_version(), sv.schema_version()) << ", status: " << status;
+      auto schema_version = 0;
+      if (sv.table_id().empty()) {
+        schema_version = metadata.schema_version();
+      } else {
+        auto uuid = Uuid::FromSlice(sv.table_id());
+        CHECK(uuid.ok());
+        auto schema_version_result = metadata.schema_version(*uuid);
+        if (!schema_version_result.ok()) {
+          Complete(schema_version_result.status());
+          return;
+        }
+        schema_version = *schema_version_result;
+      }
+      CHECK_LE(schema_version, sv.schema_version()) << ", status: " << status;
     }
   }
 

src/yb/util/mem_tracker.cc

@@ -473,7 +473,7 @@ bool MemTracker::UpdateConsumption(bool force) {
           &FLAGS_mem_tracker_update_consumption_interval_us));
       auto value = consumption_functor_();
       VLOG(1) << "Setting consumption of tracker " << id_ << " to " << value
-              << "from consumption functor";
+              << " from consumption functor";
       consumption_.set_value(value);
       if (metrics_) {
         metrics_->metric_->set_value(value);