fix: resolve conflicts and review conversations

docker/dev-builder/centos/Dockerfile

@@ -26,4 +26,5 @@ ARG RUST_TOOLCHAIN
 RUN rustup toolchain install ${RUST_TOOLCHAIN}
 
 # Install nextest.
-RUN cargo install cargo-nextest --locked
+RUN cargo install cargo-binstall --locked
+RUN cargo binstall cargo-nextest --no-confirm

docker/dev-builder/ubuntu/Dockerfile

@@ -43,4 +43,5 @@ ARG RUST_TOOLCHAIN
 RUN rustup toolchain install ${RUST_TOOLCHAIN}
 
 # Install nextest.
-RUN cargo install cargo-nextest --locked
+RUN cargo install cargo-binstall --locked
+RUN cargo binstall cargo-nextest --no-confirm

docker/dev-builder/ubuntu/Dockerfile-18.10

@@ -44,4 +44,5 @@ ARG RUST_TOOLCHAIN
 RUN rustup toolchain install ${RUST_TOOLCHAIN}
 
 # Install nextest.
-RUN cargo install cargo-nextest --locked
+RUN cargo install cargo-binstall --locked
+RUN cargo binstall cargo-nextest --no-confirm

src/catalog/src/kvbackend/manager.rs

@@ -127,13 +127,11 @@ impl CatalogManager for KvBackendCatalogManager {
             .try_collect::<BTreeSet<_>>()
             .await
             .map_err(BoxedError::new)
-            .context(ListSchemasSnafu { catalog })?
-            .into_iter()
-            .collect::<Vec<_>>();
+            .context(ListSchemasSnafu { catalog })?;
 
-        keys.extend_from_slice(&self.system_catalog.schema_names());
+        keys.extend(self.system_catalog.schema_names());
 
-        Ok(keys)
+        Ok(keys.into_iter().collect())
     }
 
     async fn table_names(&self, catalog: &str, schema: &str) -> CatalogResult<Vec<String>> {

src/catalog/src/memory/manager.rs

@@ -143,6 +143,12 @@ impl MemoryCatalogManager {
                 schema: DEFAULT_PRIVATE_SCHEMA_NAME.to_string(),
             })
             .unwrap();
+        manager
+            .register_schema_sync(RegisterSchemaRequest {
+                catalog: DEFAULT_CATALOG_NAME.to_string(),
+                schema: INFORMATION_SCHEMA_NAME.to_string(),
+            })
+            .unwrap();
 
         manager
     }
@@ -256,6 +262,7 @@ impl MemoryCatalogManager {
             Arc::downgrade(self) as Weak<dyn CatalogManager>,
         );
         let information_schema = information_schema_provider.tables().clone();
+
         let mut catalog = HashMap::new();
         catalog.insert(INFORMATION_SCHEMA_NAME.to_string(), information_schema);
         catalog

src/common/meta/src/instruction.rs

@@ -98,6 +98,8 @@ pub struct OpenRegion {
     pub region_options: HashMap<String, String>,
     #[serde(default)]
     pub region_wal_options: HashMap<String, String>,
+    #[serde(default)]
+    pub skip_wal_replay: bool,
 }
 
 impl OpenRegion {
@@ -106,12 +108,14 @@ impl OpenRegion {
         path: &str,
         region_options: HashMap<String, String>,
         region_wal_options: HashMap<String, String>,
+        skip_wal_replay: bool,
     ) -> Self {
         Self {
             region_ident,
             region_storage_path: path.to_string(),
             region_options,
             region_wal_options,
+            skip_wal_replay,
         }
     }
 }
@@ -227,12 +231,13 @@ mod tests {
             "test/foo",
             HashMap::new(),
             HashMap::new(),
+            false,
         ));
 
         let serialized = serde_json::to_string(&open_region).unwrap();
 
         assert_eq!(
-            r#"{"OpenRegion":{"region_ident":{"cluster_id":1,"datanode_id":2,"table_id":1024,"region_number":1,"engine":"mito2"},"region_storage_path":"test/foo","region_options":{},"region_wal_options":{}}}"#,
+            r#"{"OpenRegion":{"region_ident":{"cluster_id":1,"datanode_id":2,"table_id":1024,"region_number":1,"engine":"mito2"},"region_storage_path":"test/foo","region_options":{},"region_wal_options":{},"skip_wal_replay":false}}"#,
             serialized
         );
 
@@ -289,6 +294,7 @@ mod tests {
             region_storage_path,
             region_options,
             region_wal_options: HashMap::new(),
+            skip_wal_replay: false,
         };
         assert_eq!(expected, deserialized);
     }

src/common/meta/src/key.rs

@@ -64,7 +64,7 @@ use std::sync::Arc;
 
 use bytes::Bytes;
 use common_catalog::consts::{
-    DEFAULT_CATALOG_NAME, DEFAULT_PRIVATE_SCHEMA_NAME, DEFAULT_SCHEMA_NAME,
+    DEFAULT_CATALOG_NAME, DEFAULT_PRIVATE_SCHEMA_NAME, DEFAULT_SCHEMA_NAME, INFORMATION_SCHEMA_NAME,
 };
 use common_telemetry::warn;
 use datanode_table::{DatanodeTableKey, DatanodeTableManager, DatanodeTableValue};
@@ -297,17 +297,20 @@ impl TableMetadataManager {
 
     pub async fn init(&self) -> Result<()> {
         let catalog_name = CatalogNameKey::new(DEFAULT_CATALOG_NAME);
-        let public_schema_name = SchemaNameKey::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME);
-        let private_schema_name =
-            SchemaNameKey::new(DEFAULT_CATALOG_NAME, DEFAULT_PRIVATE_SCHEMA_NAME);
 
         self.catalog_manager().create(catalog_name, true).await?;
-        self.schema_manager()
-            .create(public_schema_name, None, true)
-            .await?;
-        self.schema_manager()
-            .create(private_schema_name, None, true)
-            .await?;
+
+        let internal_schemas = [
+            DEFAULT_SCHEMA_NAME,
+            INFORMATION_SCHEMA_NAME,
+            DEFAULT_PRIVATE_SCHEMA_NAME,
+        ];
+
+        for schema_name in internal_schemas {
+            let schema_key = SchemaNameKey::new(DEFAULT_CATALOG_NAME, schema_name);
+
+            self.schema_manager().create(schema_key, None, true).await?;
+        }
 
         Ok(())
     }

src/datanode/src/heartbeat/handler.rs

@@ -55,7 +55,8 @@ impl RegionHeartbeatResponseHandler {
                 region_storage_path,
                 region_options,
                 region_wal_options,
-            }) => Ok(Box::new(|region_server| {
+                skip_wal_replay,
+            }) => Ok(Box::new(move |region_server| {
                 Box::pin(async move {
                     let region_id = Self::region_ident_to_region_id(&region_ident);
                     // TODO(niebayes): extends region options with region_wal_options.
@@ -64,7 +65,7 @@ impl RegionHeartbeatResponseHandler {
                         engine: region_ident.engine,
                         region_dir: region_dir(&region_storage_path, region_id),
                         options: region_options,
-                        skip_wal_replay: false,
+                        skip_wal_replay,
                     });
                     let result = region_server.handle_request(region_id, request).await;
 
@@ -244,6 +245,7 @@ mod tests {
             path,
             HashMap::new(),
             HashMap::new(),
+            false,
         ))
     }
 

src/log-store/src/kafka/log_store.rs

@@ -53,7 +53,7 @@ impl LogStore for KafkaLogStore {
     type Entry = EntryImpl;
     type Namespace = NamespaceImpl;
 
-    /// Creates an entry.
+    /// Creates an entry of the associated Entry type.
     fn entry<D: AsRef<[u8]>>(
         &self,
         data: D,
@@ -74,19 +74,20 @@ impl LogStore for KafkaLogStore {
             .produce(&self.client_manager)
             .await
             .map(TryInto::try_into)??;
-        Ok(AppendResponse { entry_id })
+        Ok(AppendResponse {
+            last_entry_id: entry_id,
+        })
     }
 
-    /// Appends a batch of entries to the log store. The response contains a map where the key
-    /// is a region id while the value is the id of the entry, the first entry of the entries belong to the region,
-    /// written into the log store.
+    /// Appends a batch of entries and returns a response containing a map where the key is a region id
+    /// while the value is the id of the last successfully written entry of the region.
     async fn append_batch(&self, entries: Vec<Self::Entry>) -> Result<AppendBatchResponse> {
         if entries.is_empty() {
             return Ok(AppendBatchResponse::default());
         }
 
         // Groups entries by region id and pushes them to an associated record producer.
-        let mut producers: HashMap<_, RecordProducer> = HashMap::with_capacity(entries.len());
+        let mut producers = HashMap::with_capacity(entries.len());
         for entry in entries {
             producers
                 .entry(entry.ns.region_id)
@@ -108,7 +109,7 @@ impl LogStore for KafkaLogStore {
             .map(TryInto::try_into)
             .collect::<Result<Vec<_>>>()?;
         Ok(AppendBatchResponse {
-            entry_ids: region_ids.into_iter().zip(entry_ids).collect(),
+            last_entry_ids: region_ids.into_iter().zip(entry_ids).collect(),
         })
     }
 
@@ -144,7 +145,7 @@ impl LogStore for KafkaLogStore {
         Ok(Box::pin(stream))
     }
 
-    /// Creates a namespace.
+    /// Creates a namespace of the associated Namespace type.
     fn namespace(&self, ns_id: NamespaceId, wal_options: &WalOptions) -> Self::Namespace {
         // Safety: upon start, the datanode checks the consistency of the wal providers in the wal config of the
         // datanode and that of the metasrv. Therefore, the wal options passed into the kafka log store
@@ -158,21 +159,24 @@ impl LogStore for KafkaLogStore {
         }
     }
 
+    /// Creates a new `Namespace` from the given ref.
     async fn create_namespace(&self, _ns: &Self::Namespace) -> Result<()> {
         Ok(())
     }
 
+    /// Deletes an existing `Namespace` specified by the given ref.
     async fn delete_namespace(&self, _ns: &Self::Namespace) -> Result<()> {
         Ok(())
     }
 
+    /// Lists all existing namespaces.
     async fn list_namespaces(&self) -> Result<Vec<Self::Namespace>> {
         Ok(vec![])
     }
 
-    /// Marks all entry ids `<=id` of given `namespace` as obsolete so that logstore can safely delete
-    /// the log files if all entries inside are obsolete. This method may not delete log
-    /// files immediately.
+    /// Marks all entries with ids `<=entry_id` of the given `namespace` as obsolete,
+    /// so that the log store can safely delete those entries. This method does not guarantee
+    /// that the obsolete entries are deleted immediately.
     async fn obsolete(&self, _ns: Self::Namespace, _entry_id: EntryId) -> Result<()> {
         Ok(())
     }

src/log-store/src/kafka/record_utils.rs

@@ -30,6 +30,8 @@ type ConsumeResult = std::result::Result<(RecordAndOffset, i64), rskafka::client
 /// Record metadata which will be serialized/deserialized to/from the `key` of a Record.
 #[derive(Debug, Serialize, Deserialize, PartialEq)]
 struct RecordMeta {
+    /// Meta version. Used for backward compatibility.
+    version: u32,
     /// The namespace of the entries wrapped in the record.
     ns: NamespaceImpl,
     /// Ids of the entries built into the record.
@@ -41,6 +43,7 @@ struct RecordMeta {
 impl RecordMeta {
     fn new(ns: NamespaceImpl, entries: &[EntryImpl]) -> Self {
         Self {
+            version: 0,
             ns,
             entry_ids: entries.iter().map(|entry| entry.id).collect(),
             entry_offsets: entries

src/log-store/src/noop.rs

@@ -63,7 +63,7 @@ impl LogStore for NoopLogStore {
     }
 
     async fn append(&self, mut _e: Self::Entry) -> Result<AppendResponse> {
-        Ok(AppendResponse { entry_id: 0 })
+        Ok(AppendResponse::default())
     }
 
     async fn append_batch(&self, _e: Vec<Self::Entry>) -> Result<AppendBatchResponse> {