feat(mito): Allow to retry create request and alter request (#2447)

* feat: RegionMetadataBuilder allow adding/dropping columns multiple times

* test: test add if not exists/drop if exists

* feat: change validator and add need_alter

* test: fix tests and test need_alter

* test: test alter retry

* feat: open before create

* style: fix clippy

src/mito2/src/engine/alter_test.rs

@@ -106,7 +106,6 @@ async fn test_alter_region() {
         assert_eq!(1, version_data.last_entry_id);
         assert_eq!(3, version_data.committed_sequence);
         assert_eq!(1, version_data.version.flushed_entry_id);
-        assert_eq!(1, version_data.version.flushed_entry_id);
         assert_eq!(3, version_data.version.flushed_sequence);
     };
     check_region(&engine);
@@ -245,3 +244,53 @@ async fn test_put_after_alter() {
     let batches = RecordBatches::try_collect(stream).await.unwrap();
     assert_eq!(expected, batches.pretty_print().unwrap());
 }
+
+#[tokio::test]
+async fn test_alter_region_retry() {
+    common_telemetry::init_default_ut_logging();
+
+    let mut env = TestEnv::new();
+    let engine = env.create_engine(MitoConfig::default()).await;
+
+    let region_id = RegionId::new(1, 1);
+    let request = CreateRequestBuilder::new().build();
+
+    let column_schemas = rows_schema(&request);
+    engine
+        .handle_request(region_id, RegionRequest::Create(request))
+        .await
+        .unwrap();
+
+    let rows = Rows {
+        schema: column_schemas,
+        rows: build_rows_for_key("a", 0, 2, 0),
+    };
+    put_rows(&engine, region_id, rows).await;
+
+    let request = add_tag1();
+    engine
+        .handle_request(region_id, RegionRequest::Alter(request))
+        .await
+        .unwrap();
+    // Retries request.
+    let request = add_tag1();
+    engine
+        .handle_request(region_id, RegionRequest::Alter(request))
+        .await
+        .unwrap();
+
+    let expected = "\
++-------+-------+---------+---------------------+
+| tag_1 | tag_0 | field_0 | ts                  |
++-------+-------+---------+---------------------+
+|       | a     | 0.0     | 1970-01-01T00:00:00 |
+|       | a     | 1.0     | 1970-01-01T00:00:01 |
++-------+-------+---------+---------------------+";
+    scan_check_after_alter(&engine, region_id, expected).await;
+    let region = engine.get_region(region_id).unwrap();
+    let version_data = region.version_control.current();
+    assert_eq!(1, version_data.last_entry_id);
+    assert_eq!(2, version_data.committed_sequence);
+    assert_eq!(1, version_data.version.flushed_entry_id);
+    assert_eq!(2, version_data.version.flushed_sequence);
+}

src/mito2/src/engine/create_test.rs

@@ -14,8 +14,6 @@
 
 use std::time::Duration;
 
-use common_error::ext::ErrorExt;
-use common_error::status_code::StatusCode;
 use store_api::region_engine::RegionEngine;
 use store_api::region_request::RegionRequest;
 use store_api::storage::RegionId;
@@ -39,12 +37,12 @@ async fn test_engine_create_new_region() {
 }
 
 #[tokio::test]
-async fn test_engine_create_region_if_not_exists() {
-    let mut env = TestEnv::with_prefix("create-not-exists");
+async fn test_engine_create_existing_region() {
+    let mut env = TestEnv::with_prefix("create-existing");
     let engine = env.create_engine(MitoConfig::default()).await;
 
     let region_id = RegionId::new(1, 1);
-    let builder = CreateRequestBuilder::new().create_if_not_exists(true);
+    let builder = CreateRequestBuilder::new();
     engine
         .handle_request(region_id, RegionRequest::Create(builder.build()))
         .await
@@ -58,8 +56,8 @@ async fn test_engine_create_region_if_not_exists() {
 }
 
 #[tokio::test]
-async fn test_engine_create_existing_region() {
-    let mut env = TestEnv::with_prefix("create-existing");
+async fn test_engine_create_with_different_id() {
+    let mut env = TestEnv::new();
     let engine = env.create_engine(MitoConfig::default()).await;
 
     let region_id = RegionId::new(1, 1);
@@ -69,15 +67,51 @@ async fn test_engine_create_existing_region() {
         .await
         .unwrap();
 
-    // Create the same region again.
-    let err = engine
+    // Creates with different id.
+    engine
+        .handle_request(RegionId::new(2, 1), RegionRequest::Create(builder.build()))
+        .await
+        .unwrap_err();
+}
+
+#[tokio::test]
+async fn test_engine_create_with_different_schema() {
+    let mut env = TestEnv::new();
+    let engine = env.create_engine(MitoConfig::default()).await;
+
+    let region_id = RegionId::new(1, 1);
+    let builder = CreateRequestBuilder::new();
+    engine
+        .handle_request(region_id, RegionRequest::Create(builder.build()))
+        .await
+        .unwrap();
+
+    // Creates with different schema.
+    let builder = builder.tag_num(2);
+    engine
+        .handle_request(region_id, RegionRequest::Create(builder.build()))
+        .await
+        .unwrap_err();
+}
+
+#[tokio::test]
+async fn test_engine_create_with_different_primary_key() {
+    let mut env = TestEnv::new();
+    let engine = env.create_engine(MitoConfig::default()).await;
+
+    let region_id = RegionId::new(1, 1);
+    let builder = CreateRequestBuilder::new().tag_num(2);
+    engine
+        .handle_request(region_id, RegionRequest::Create(builder.build()))
+        .await
+        .unwrap();
+
+    // Creates with different schema.
+    let builder = builder.primary_key(vec![1]);
+    engine
         .handle_request(region_id, RegionRequest::Create(builder.build()))
         .await
         .unwrap_err();
-    assert!(
-        matches!(err.status_code(), StatusCode::RegionAlreadyExists),
-        "unexpected err: {err}"
-    );
 }
 
 #[tokio::test]

src/mito2/src/engine/flush_test.rs

@@ -250,7 +250,6 @@ async fn test_flush_reopen_region() {
         assert_eq!(1, version_data.last_entry_id);
         assert_eq!(3, version_data.committed_sequence);
         assert_eq!(1, version_data.version.flushed_entry_id);
-        assert_eq!(1, version_data.version.flushed_entry_id);
         assert_eq!(3, version_data.version.flushed_sequence);
     };
     check_region();

src/mito2/src/error.rs

@@ -459,6 +459,17 @@ pub enum Error {
         source: serde_json::Error,
         location: Location,
     },
+
+    #[snafu(display(
+        "Empty region directory, region_id: {}, region_dir: {}",
+        region_id,
+        region_dir,
+    ))]
+    EmptyRegionDir {
+        region_id: RegionId,
+        region_dir: String,
+        location: Location,
+    },
 }
 
 pub type Result<T, E = Error> = std::result::Result<T, E>;
@@ -529,6 +540,7 @@ impl ErrorExt for Error {
             InvalidRegionRequest { source, .. } => source.status_code(),
             RegionReadonly { .. } => StatusCode::RegionReadonly,
             JsonOptions { .. } => StatusCode::InvalidArguments,
+            EmptyRegionDir { .. } => StatusCode::RegionNotFound,
         }
     }