Fix create and drop table

Author: Peng Gu

libspectrum/include/spectrum.h

@@ -45,6 +45,9 @@ extern void spectrum_row_extract_fields(TABLE *table, ::spectrum::Row *spectrum_
 extern void spectrum_row_extract_fields(TABLE *table, uchar* record, ::spectrum::Row *spectrum_row);
 extern void spectrum_thread_fill(THD *thd, spectrum::Thread *spectrum_thread);
 
+template<typename T>
+int spectrum_compute_update_metadata(THD *thd, const T *object);
+
 extern int spectrum_compute_create_table(THD *thd, TABLE *table);
 extern int spectrum_compute_delete_table(THD *thd, const dd::Table *table_def, const char* table_path);
 extern int spectrum_compute_lock_table(THD *thd, TABLE *table);

libspectrum/src/protobuf/spectrum.proto

@@ -6,6 +6,7 @@ service StorageNode {
   rpc BeginAttachableTransaction(BeginAttachableTransactionRequest) returns (BeginAttachableTransactionResponse) {}
   rpc EndAttachableTransaction(EndAttachableTransactionRequest) returns (EndAttachableTransactionResponse) {}
 
+  rpc UpdateMetadata(UpdateMetadataRequest) returns (UpdateMetadataResponse) {}
   rpc AcquireMetadataLock(AcquireMetadataLockRequest) returns (AcquireMetadataLockResponse) {}
   rpc UpgradeMetadataLock(UpgradeMetadataLockRequest) returns (UpgradeMetadataLockResponse) {}
   rpc ReleaseMetadataLock(ReleaseMetadataLockRequest) returns (ReleaseMetadataLockResponse) {}
@@ -59,6 +60,16 @@ message Field {
   bool is_null = 3;
 }
 
+message UpdateMetadataRequest {
+  Thread thread = 1;
+  string table = 2;
+  uint64 object_id = 3;
+  string object_name = 4;
+}
+
+message UpdateMetadataResponse {
+}
+
 message AcquireMetadataLockRequest {
   Thread thread = 1;
   int32 namespace = 2;

libspectrum/src/spectrum_common.cc

@@ -127,7 +127,7 @@ void spectrum_print_row(char* method, TABLE* table, uchar* record) {
     row.pop_back();
     row.pop_back();
   }
-  sql_print_information("%s[%s:%d]: %s", method, table->s->table_name.str, hander_id, row.c_str());
+  sql_print_information("%s[%s:%s:%d]: %s", method, table->s->db.str, table->s->table_name.str, hander_id, row.c_str());
 
   repoint_field_to_record(table, record, table->record[0]);
   table->read_set = temp_read_set;

libspectrum/src/spectrum_compute.cc

@@ -1,6 +1,6 @@
-drop database if exists spectrum_test_45;
-create database spectrum_test_45;
-use spectrum_test_45;
+drop database if exists spectrum_test_66;
+create database spectrum_test_66;
+use spectrum_test_66;
 create table employees (id INT PRIMARY KEY, first_name VARCHAR(50) NOT NULL, last_name VARCHAR(50) NOT NULL, hire_date DATE, salary DECIMAL(10, 2));
 insert into employees (id, first_name, last_name, hire_date, salary) VALUES (2, 'Peng', 'Gu', '2024-12-01', 60000.00);
 insert into employees (id, first_name, last_name, hire_date, salary) VALUES (1, 'John', 'Doe', '2024-12-01', 50000.00);
@@ -12,7 +12,7 @@ update employees set salary=555 where id=1;
 select * from employees where id=1;
 id	first_name	last_name	hire_date	salary
 1	John	Doe	2024-12-01	555.00
-use spectrum_test_45;
+use spectrum_test_66;
 select * from employees;
 id	first_name	last_name	hire_date	salary
 1	John	Doe	2024-12-01	555.00

mysql-test/r/spectrum.result

@@ -1,7 +1,7 @@
 connect(compute0,localhost,root,,,3306,/tmp/mysql-spectrum-compute-0.sock);
 connect(compute1,localhost,root,,,3306,/tmp/mysql-spectrum-compute-1.sock);
 
---let $db=spectrum_test_45
+--let $db=spectrum_test_66
 
 connection compute0;
 

mysql-test/t/spectrum.test

@@ -101,6 +101,12 @@ class StorageNodeImpl final : public spectrum::StorageNode::Service {
       thd->lex->sql_command = (enum_sql_command)spectrum_thread.sql_command();
       thd->tx_isolation = (enum_tx_isolation)spectrum_thread.tx_isolation();
       thd->variables.option_bits = spectrum_thread.system_variables().option_bits();
+      
+      // This is needed because register_uncommitted_object() and register_dropped_object() require a non-default
+      // auto releaser, even though it's not actually required because uncommitted/dropped objects will be
+      // released in remove_uncommitted_objects() when transaction commits.
+      new dd::cache::Dictionary_client::Auto_releaser(thd->dd_client());
+      
       return (thd);
     }
 
@@ -190,21 +196,13 @@ class StorageNodeImpl final : public spectrum::StorageNode::Service {
       const char* table_name = request->table().c_str();
       uint64 handler_id = request->handler();
 
-      sql_print_information("CreateTable[%s:%d]", table_name, handler_id);
+      sql_print_information("CreateTable[%s:%s:%d]", db_name, table_name, handler_id);
 
       thd = handler_create_thd(request->thread());
 
-      // Invalidate shared dd cache for db
-      dd::cache::Dictionary_client::Auto_releaser releaser(thd->dd_client());
-      thd->dd_client()->acquire(db_name, &schema_def);
-      if (schema_def == nullptr) {
-        sql_print_error("CreateTable[%s:%s:%d]: can not find schema definition", db_name, table_name, handler_id);
-        goto end;
-      }
-      thd->dd_client()->invalidate(schema_def);
-
       // Retrive table definition
-      thd->dd_client()->reload_uncommitted(db_name, table_name, (const dd::Abstract_table **)&table_def);
+      dd::cache::Dictionary_client::Auto_releaser releaser(thd->dd_client());
+      thd->dd_client()->acquire(db_name, table_name, (const dd::Abstract_table **)&table_def);
       if (table_def == nullptr) {
         sql_print_error("CreateTable[%s:%s:%d]: can not find table definition", db_name, table_name, handler_id);
         goto end;
@@ -219,9 +217,6 @@ class StorageNodeImpl final : public spectrum::StorageNode::Service {
       }
 
     end:
-      thd->lex->sql_command = SQLCOM_CREATE_TABLE;
-      handlerton *db_type = get_viable_handlerton_for_create(thd, table_name, create_info);
-      thd->m_transactional_ddl.init(db_name, table_name, db_type);
       return grpc::Status::OK; 
     }
 
@@ -283,7 +278,7 @@ class StorageNodeImpl final : public spectrum::StorageNode::Service {
       thr_lock_type lock_type = (thr_lock_type)request->lock_type();
       thr_locked_row_action lock_action = (thr_locked_row_action)request->lock_action();
 
-      sql_print_information("LockTable[%s:%d]: lock_type=%d", table_name, handler_id, lock_type);
+      sql_print_information("LockTable[%s:%s:%d]: lock_type=%d", db_name, table_name, handler_id, lock_type);
 
       thd = handler_create_thd(request->thread());
       table = handler_find_or_open_table(thd, db_name, table_name, handler_id, lock_type, lock_action);
@@ -304,7 +299,7 @@ class StorageNodeImpl final : public spectrum::StorageNode::Service {
       thr_lock_type lock_type = (thr_lock_type)request->lock_type();
       thr_locked_row_action lock_action = (thr_locked_row_action)request->lock_action();
 
-      sql_print_information("UnLockTable[%s:%d]", table_name, handler_id);
+      sql_print_information("UnLockTable[%s:%s:%d]", db_name, table_name, handler_id);
 
       thd = handler_create_thd(request->thread());
       table = handler_find_or_open_table(thd, db_name, table_name, handler_id, lock_type, lock_action);
@@ -321,7 +316,7 @@ class StorageNodeImpl final : public spectrum::StorageNode::Service {
       const char* table_name = request->table().c_str();
       uint64 handler_id = request->handler();
 
-      sql_print_information("CloseTable[%s:%d]", table_name, handler_id);
+      sql_print_information("CloseTable[%s:%s:%d]", db_name, table_name, handler_id);
 
       thd = handler_create_thd(request->thread());
       handler_find_and_close_table(thd, db_name, table_name, handler_id);
@@ -335,7 +330,7 @@ class StorageNodeImpl final : public spectrum::StorageNode::Service {
       thr_lock_type lock_type = (thr_lock_type)request->lock_type();
       thr_locked_row_action lock_action = (thr_locked_row_action)request->lock_action();
 
-      sql_print_information("InitIndex[%s:%d]: index=%d", request->table().c_str(), request->handler(), request->index());
+      sql_print_information("InitIndex[%s:%s:%d]: index=%d", request->database().c_str(), request->table().c_str(), request->handler(), request->index());
 
       thd = handler_create_thd(request->thread());
       table = handler_find_or_open_table(thd, request->database().c_str(), request->table().c_str(), request->handler(), lock_type, lock_action);
@@ -351,7 +346,7 @@ class StorageNodeImpl final : public spectrum::StorageNode::Service {
       thr_lock_type lock_type = (thr_lock_type)request->lock_type();
       thr_locked_row_action lock_action = (thr_locked_row_action)request->lock_action();
 
-      sql_print_information("InitRnd[%s:%d]: scan=%d", request->table().c_str(), request->handler(), request->scan());
+      sql_print_information("InitRnd[%s:%s:%d]: scan=%d", request->database().c_str(), request->table().c_str(), request->handler(), request->scan());
 
       thd = handler_create_thd(request->thread());
       table = handler_find_or_open_table(thd, request->database().c_str(), request->table().c_str(), request->handler(), lock_type, lock_action);
@@ -367,7 +362,7 @@ class StorageNodeImpl final : public spectrum::StorageNode::Service {
       thr_lock_type lock_type = (thr_lock_type)request->lock_type();
       thr_locked_row_action lock_action = (thr_locked_row_action)request->lock_action();
 
-      sql_print_information("EndIndex[%s:%d]", request->table().c_str(), request->handler());
+      sql_print_information("EndIndex[%s:%s:%d]", request->database().c_str(), request->table().c_str(), request->handler());
 
       thd = handler_create_thd(request->thread());
       table = handler_find_or_open_table(thd, request->database().c_str(), request->table().c_str(), request->handler(), lock_type, lock_action);
@@ -394,7 +389,7 @@ class StorageNodeImpl final : public spectrum::StorageNode::Service {
         key = (const uchar *)request->key().data();
       }
 
-      sql_print_information("ReadRow[%s:%d]", request->table().c_str(), request->handler());
+      sql_print_information("ReadRow[%s:%s:%d]", request->database().c_str(), request->table().c_str(), request->handler());
 
       thd = handler_create_thd(request->thread());
       table = handler_find_or_open_table(thd, request->database().c_str(), request->table().c_str(), request->handler(), lock_type, lock_action);
@@ -415,7 +410,7 @@ class StorageNodeImpl final : public spectrum::StorageNode::Service {
       thr_locked_row_action lock_action = (thr_locked_row_action)request->lock_action();
       int error;
 
-      sql_print_information("ReadNextRow[%s:%d]", request->table().c_str(), request->handler());
+      sql_print_information("ReadNextRow[%s:%s:%d]", request->database().c_str(), request->table().c_str(), request->handler());
 
       thd = handler_create_thd(request->thread());
       table = handler_find_or_open_table(thd, request->database().c_str(), request->table().c_str(), request->handler(), lock_type, lock_action);
@@ -444,7 +439,7 @@ class StorageNodeImpl final : public spectrum::StorageNode::Service {
       thr_locked_row_action lock_action = (thr_locked_row_action)request->lock_action();
       int error;
 
-      sql_print_information("ReadPrevRow[%s:%d]", request->table().c_str(), request->handler());
+      sql_print_information("ReadPrevRow[%s:%s:%d]", request->database().c_str(), request->table().c_str(), request->handler());
 
       thd = handler_create_thd(request->thread());
       table = handler_find_or_open_table(thd, request->database().c_str(), request->table().c_str(), request->handler(), lock_type, lock_action);
@@ -584,6 +579,27 @@ class StorageNodeImpl final : public spectrum::StorageNode::Service {
       return grpc::Status::OK; 
     }
 
+    ::grpc::Status UpdateMetadata(::grpc::ServerContext* context, const ::spectrum::UpdateMetadataRequest* request, ::spectrum::UpdateMetadataResponse* response) {
+      THD *thd;
+      const std::string& table = request->table();
+      const dd::Object_id object_id = request->object_id();
+      const std::string& object_name = request->object_name();
+
+      sql_print_information("UpdateMetadata: table=%s, object_id=%d, object_name=%s",
+            table.c_str(), object_id, object_name.c_str()); 
+
+      thd = handler_create_thd(request->thread());
+
+      if (!strcmp(table.c_str(), "schemata")) {
+        const dd::Schema *object;
+        thd->dd_client()->reload_uncommitted(object_id, &object);
+      } else if (!strcmp(table.c_str(), "tables")) {
+        const dd::Abstract_table *object;
+        thd->dd_client()->reload_uncommitted(object_id, &object);
+      }
+      return grpc::Status::OK; 
+    }
+
     ::grpc::Status AcquireMetadataLock(::grpc::ServerContext* context, const ::spectrum::AcquireMetadataLockRequest* request, ::spectrum::AcquireMetadataLockResponse* response) {
       THD *thd;
       MDL_key::enum_mdl_namespace namespace_ = static_cast<MDL_key::enum_mdl_namespace>(request->namespace_());
@@ -706,7 +722,7 @@ class StorageNodeImpl final : public spectrum::StorageNode::Service {
       mysql_unlock_some_tables(thd, &table, 1);
 
       if (err) {
-        sql_print_error("ReplicateRow[%s:%d]: error=%d", request->table().c_str(), request->handler(), err);
+        sql_print_error("ReplicateRow[%s:%s:%d]: error=%d", request->database().c_str(), request->table().c_str(), request->handler(), err);
       }
       return grpc::Status::OK; 
     }

plugin/spectrum_storage/spectrum_storage.cc

@@ -686,22 +686,20 @@ class Dictionary_client {
   template <typename T>
   [[nodiscard]] bool acquire(const String_type &schema_name,
                              const String_type &object_name, const T **object);
-
+  
   /**
-    Reload an uncommitted object from storage and put it to uncommitted cache
+    Reload an uncommitted object from storage and put it to uncommitted local cache
 
     @tparam       T             Dictionary object type.
-    @param        schema_name   Name of the schema containing the object.
-    @param        object_name   Name of the object.
+    @param        id            id of the object.
     @param [out]  object        Dictionary object, if present; otherwise NULL.
 
     @retval       false   No error.
     @retval       true    Error
   */
-
   template <typename T>
-  [[nodiscard]] bool reload_uncommitted(const String_type &schema_name,
-                                        const String_type &object_name, const T **object);
+  [[nodiscard]] bool reload_uncommitted(const Object_id id,
+                                        const T **object);
 
   /**
     Retrieve an object by its schema- and object name.

sql/dd/cache/dictionary_client.h

@@ -1397,54 +1397,41 @@ bool Dictionary_client::acquire(const String_type &schema_name,
 }
 
 template <typename T>
-bool Dictionary_client::reload_uncommitted(const String_type &schema_name,
-                                           const String_type &object_name,
+bool Dictionary_client::reload_uncommitted(Object_id id,
                                            const T **object) {
-  // We must make sure the schema is released and unlocked in the right order.
-  Schema_MDL_locker mdl_locker(m_thd);
-  Auto_releaser releaser(this);
-
-  assert(object);
-  *object = nullptr;
-
-  // Get the schema object by name.
-  const Schema *schema = nullptr;
-  bool error = mdl_locker.ensure_locked(schema_name.c_str()) ||
-               acquire(schema_name, &schema);
-
-  // If there was an error, or if we found no valid schema, return here.
-  if (error) {
-    assert(m_thd->is_error() || m_thd->killed);
-    return true;
-  }
-
-  // A non existing schema is not reported as an error.
-  if (!schema) return false;
-
-  // Create the name key for the object.
-  typename T::Name_key key;
-  T::update_name_key(&key, schema->id(), object_name);
-
-  // Cache dictionary objects with UTC time
-  Timestamp_timezone_guard ts(m_thd);
+  bool error;
+  typename T::Id_key key(id);
 
   // Read the uncached dictionary object using ISO_READ_UNCOMMITTED
   // isolation level.
   const typename T::Cache_partition *stored_object = nullptr;
   error = Shared_dictionary_cache::instance()->get_uncached(
       m_thd, key, ISO_READ_UNCOMMITTED, &stored_object);
-  if (!error) {
-    // Here, stored_object is a newly created instance, so we do not need to
-    // clone() it, but we must delete it if dynamic cast fails.
-    *object = const_cast<T *>(dynamic_cast<const T *>(stored_object));
-    if (stored_object && !*object) delete stored_object;
-    if (*object) {
-      register_uncommitted_object(*object);
-    }
-  } else
+  if (error) {
     assert(m_thd->is_error() || m_thd->killed);
+    return true;
+  }
+  *object = const_cast<T *>(dynamic_cast<const T *>(stored_object));
+  if (stored_object && !*object) delete stored_object;
+  if (*object) {
+    register_uncommitted_object(*object);
+    return false;
+  }
 
-  return error;
+  // The object is dropped, invalidate the object from shared cache.
+  Cache_element<T> *element = nullptr;
+  error = Shared_dictionary_cache::instance()->get(m_thd, key, &element);
+  if (error) {
+    assert(m_thd->is_error() || m_thd->killed);
+    return true;
+  }
+  if (element) {
+    T *dropped_object = element->object()->clone_dropped_object_placeholder();
+    Shared_dictionary_cache::instance()->drop(element);
+    register_dropped_object(dropped_object);
+    *object = nullptr;
+  }
+  return false;
 }
 
 template <typename T>
@@ -3088,12 +3075,11 @@ template bool Dictionary_client::acquire_uncached(Object_id, Abstract_table **);
 template bool Dictionary_client::acquire(const String_type &,
                                          const String_type &,
                                          const Abstract_table **);
-template bool Dictionary_client::reload_uncommitted(const String_type &,
-                                                    const String_type &,
-                                                    const Abstract_table **);
 template bool Dictionary_client::acquire_for_modification(const String_type &,
                                                           const String_type &,
                                                           Abstract_table **);
+template bool Dictionary_client::reload_uncommitted(const Object_id,
+                                                    const Abstract_table **);
 template void Dictionary_client::remove_uncommitted_objects<Abstract_table>(
     bool);
 template bool Dictionary_client::drop(const Abstract_table *);
@@ -3145,6 +3131,8 @@ template bool Dictionary_client::drop(const Schema *);
 template bool Dictionary_client::store(Schema *);
 template bool Dictionary_client::update(Schema *);
 template void Dictionary_client::dump<Schema>() const;
+template bool Dictionary_client::reload_uncommitted(Object_id,
+                                                    const Schema **);
 
 template bool Dictionary_client::acquire(Object_id,
                                          const Spatial_reference_system **);