diff --git a/Cargo.lock b/Cargo.lock
index 56dbcb397d..d105168a1b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -5936,7 +5936,7 @@ dependencies = [
 [[package]]
 name = "greptime-proto"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=00e1d312ecc4b11ffb3c48161d61b03ed0e848a5#00e1d312ecc4b11ffb3c48161d61b03ed0e848a5"
+source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=5e358318890606a2961e12dcdf9674f6763cec3a#5e358318890606a2961e12dcdf9674f6763cec3a"
 dependencies = [
  "prost 0.14.1",
  "prost-types 0.14.1",
diff --git a/Cargo.toml b/Cargo.toml
index 8ad88ae37c..15652e474e 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -158,7 +158,7 @@ fs2 = "0.4"
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "00e1d312ecc4b11ffb3c48161d61b03ed0e848a5" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "5e358318890606a2961e12dcdf9674f6763cec3a" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"
diff --git a/src/common/meta/src/ddl/create_table/template.rs b/src/common/meta/src/ddl/create_table/template.rs
index b649900074..4097c947cd 100644
--- a/src/common/meta/src/ddl/create_table/template.rs
+++ b/src/common/meta/src/ddl/create_table/template.rs
@@ -21,6 +21,7 @@ use api::v1::{ColumnDef, CreateTableExpr, SemanticType};
 use common_telemetry::warn;
 use snafu::{OptionExt, ResultExt};
 use store_api::metric_engine_consts::LOGICAL_TABLE_METADATA_KEY;
+use store_api::region_request::RegionRequirements;
 use store_api::storage::{RegionId, RegionNumber};
 use table::metadata::{TableId, TableInfo};
 
@@ -70,6 +71,7 @@ pub fn build_template_from_raw_table_info(table_info: &TableInfo) -> Result<Crea
         path: String::new(),
         options,
         partition: None,
+        requirements: None,
     };
 
     Ok(template)
@@ -126,6 +128,7 @@ pub fn build_template_from_raw_table_info_for_physical_table(
         path: String::new(),
         options,
         partition: None,
+        requirements: None,
     };
 
     Ok(template)
@@ -188,6 +191,7 @@ pub(crate) fn build_template(create_table_expr: &CreateTableExpr) -> Result<Crea
         path: String::new(),
         options: create_table_expr.table_options.clone(),
         partition: None,
+        requirements: None,
     };
 
     Ok(template)
@@ -198,6 +202,7 @@ pub struct CreateRequestBuilder {
     template: CreateRequest,
     /// Optional. Only for metric engine.
     physical_table_id: Option<TableId>,
+    requirements: RegionRequirements,
 }
 
 impl CreateRequestBuilder {
@@ -205,6 +210,7 @@ impl CreateRequestBuilder {
         Self {
             template,
             physical_table_id,
+            requirements: RegionRequirements::empty(),
         }
     }
 
@@ -212,6 +218,11 @@ impl CreateRequestBuilder {
         &self.template
     }
 
+    pub fn with_requirements(mut self, requirements: RegionRequirements) -> Self {
+        self.requirements = requirements;
+        self
+    }
+
     pub fn build_one(
         &self,
         region_id: RegionId,
@@ -223,6 +234,7 @@ impl CreateRequestBuilder {
 
         request.region_id = region_id.as_u64();
         request.path = storage_path;
+        request.requirements = Some(self.requirements.into());
         // Stores the encoded wal options into the request options.
         prepare_wal_options(&mut request.options, region_id, region_wal_options);
         request.partition = Some(prepare_partition_expr(region_id, partition_exprs));
@@ -278,6 +290,7 @@ mod tests {
             path: String::new(),
             options: Default::default(),
             partition: None,
+            requirements: None,
         };
         let builder = CreateRequestBuilder::new(template, None);
 
@@ -293,6 +306,38 @@ mod tests {
             &partition_exprs,
         );
         assert_eq!(r0.partition.as_ref().unwrap().expression, expr_a);
+        assert_eq!(
+            r0.requirements.map(RegionRequirements::from),
+            Some(RegionRequirements::empty())
+        );
+    }
+
+    #[test]
+    fn test_build_one_sets_explicit_requirements() {
+        let template = CreateRequest {
+            region_id: 0,
+            engine: "mito".to_string(),
+            column_defs: vec![],
+            primary_key: vec![],
+            path: String::new(),
+            options: Default::default(),
+            partition: None,
+            requirements: None,
+        };
+        let builder = CreateRequestBuilder::new(template, None)
+            .with_requirements(RegionRequirements::object_storage());
+
+        let request = builder.build_one(
+            RegionId::new(42, 0),
+            "/p".to_string(),
+            &Default::default(),
+            &Default::default(),
+        );
+
+        assert_eq!(
+            request.requirements.map(RegionRequirements::from),
+            Some(RegionRequirements::object_storage())
+        );
     }
 
     #[test]
diff --git a/src/file-engine/src/engine.rs b/src/file-engine/src/engine.rs
index 2ddbb6c414..07d201b29a 100644
--- a/src/file-engine/src/engine.rs
+++ b/src/file-engine/src/engine.rs
@@ -186,7 +186,7 @@ struct EngineInner {
 
 type EngineInnerRef = Arc<EngineInner>;
 
-fn ensure_open_requirements(
+fn ensure_region_requirements(
     requirements: RegionRequirements,
     object_store: &ObjectStore,
 ) -> EngineResult<()> {
@@ -276,6 +276,8 @@ impl EngineInner {
             return Ok(0);
         }
 
+        ensure_region_requirements(request.requirements, &self.object_store)?;
+
         let res = FileRegion::create(region_id, request, &self.object_store).await;
         let region = res.inspect_err(|err| {
             error!(
@@ -307,7 +309,7 @@ impl EngineInner {
             return Ok(0);
         }
 
-        ensure_open_requirements(request.requirements, &self.object_store)?;
+        ensure_region_requirements(request.requirements, &self.object_store)?;
 
         let res = FileRegion::open(region_id, request, &self.object_store).await;
         let region = res.inspect_err(|err| {
@@ -402,13 +404,13 @@ mod tests {
     }
 
     #[test]
-    fn test_empty_open_requirements_are_supported() {
-        ensure_open_requirements(RegionRequirements::empty(), &build_fs_object_store()).unwrap();
+    fn test_empty_region_requirements_are_supported() {
+        ensure_region_requirements(RegionRequirements::empty(), &build_fs_object_store()).unwrap();
     }
 
     #[test]
-    fn test_object_storage_open_requirement_rejects_fs_object_store() {
-        let err = ensure_open_requirements(
+    fn test_object_storage_region_requirement_rejects_fs_object_store() {
+        let err = ensure_region_requirements(
             RegionRequirements::object_storage(),
             &build_fs_object_store(),
         )
@@ -418,8 +420,8 @@ mod tests {
     }
 
     #[test]
-    fn test_object_storage_open_requirement_accepts_s3_object_store() {
-        ensure_open_requirements(
+    fn test_object_storage_region_requirement_accepts_s3_object_store() {
+        ensure_region_requirements(
             RegionRequirements::object_storage(),
             &build_s3_object_store(),
         )
diff --git a/src/file-engine/src/region.rs b/src/file-engine/src/region.rs
index aceec21aa5..c132f4c6b8 100644
--- a/src/file-engine/src/region.rs
+++ b/src/file-engine/src/region.rs
@@ -125,6 +125,7 @@ mod tests {
             table_dir: "create_region_dir/".to_string(),
             path_type: PathType::Bare,
             partition_expr_json: Some("".to_string()),
+            requirements: Default::default(),
         };
         let region_id = RegionId::new(1, 0);
 
@@ -167,6 +168,7 @@ mod tests {
             table_dir: region_dir.clone(),
             path_type: PathType::Bare,
             partition_expr_json: Some("".to_string()),
+            requirements: Default::default(),
         };
         let region_id = RegionId::new(1, 0);
 
@@ -210,6 +212,7 @@ mod tests {
             table_dir: region_dir.clone(),
             path_type: PathType::Bare,
             partition_expr_json: Some("".to_string()),
+            requirements: Default::default(),
         };
         let region_id = RegionId::new(1, 0);
 
diff --git a/src/meta-srv/src/procedure/repartition/allocate_region.rs b/src/meta-srv/src/procedure/repartition/allocate_region.rs
index c49866ac8a..4634a74342 100644
--- a/src/meta-srv/src/procedure/repartition/allocate_region.rs
+++ b/src/meta-srv/src/procedure/repartition/allocate_region.rs
@@ -27,6 +27,7 @@ use common_procedure::{Context as ProcedureContext, Status};
 use common_telemetry::{debug, info};
 use serde::{Deserialize, Deserializer, Serialize};
 use snafu::{OptionExt, ResultExt};
+use store_api::region_request::RegionRequirements;
 use store_api::storage::{RegionId, RegionNumber, TableId};
 use table::metadata::TableInfo;
 use table::table_reference::TableReference;
@@ -382,7 +383,8 @@ impl AllocateRegion {
             table_id,
             request
         );
-        let builder = CreateRequestBuilder::new(request, None);
+        let builder = CreateRequestBuilder::new(request, None)
+            .with_requirements(RegionRequirements::object_storage());
         let region_count = region_routes.len();
         let wal_region_count = wal_options.len();
         info!(
@@ -404,6 +406,8 @@ mod tests {
     use std::collections::HashMap;
     use std::sync::Arc;
 
+    use api::v1::region::region_request::Body;
+    use common_meta::ddl::test_util::datanode_handler::DatanodeWatcher;
     use common_meta::key::TableMetadataManagerRef;
     use common_meta::key::datanode_table::DatanodeTableKey;
     use common_meta::peer::Peer;
@@ -412,7 +416,7 @@ mod tests {
     use common_procedure::{ContextProvider, ProcedureId, ProcedureState};
     use common_procedure_test::MockContextProvider;
     use store_api::storage::RegionId;
-    use tokio::sync::watch;
+    use tokio::sync::{mpsc, watch};
     use uuid::Uuid;
 
     use super::*;
@@ -741,7 +745,8 @@ mod tests {
         )
         .await;
 
-        let node_manager = Arc::new(MockDatanodeManager::new(()));
+        let (sender, mut receiver) = mpsc::channel(1);
+        let node_manager = Arc::new(MockDatanodeManager::new(DatanodeWatcher::new(sender)));
         let mut ctx = new_parent_context(&env, node_manager, table_id);
         ctx.persistent_ctx.plans = vec![RepartitionPlanEntry {
             group_id: Uuid::new_v4(),
@@ -770,6 +775,12 @@ mod tests {
 
         state.next(&mut ctx, &procedure_ctx).await.unwrap();
 
+        let (_, request) = receiver.recv().await.unwrap();
+        let Some(Body::Create(create)) = request.body else {
+            unreachable!()
+        };
+        assert!(create.requirements.unwrap().object_storage);
+
         let region_ids = current_parent_region_routes(&ctx)
             .await
             .into_iter()
diff --git a/src/meta-srv/src/procedure/tests.rs b/src/meta-srv/src/procedure/tests.rs
index 93879d3c08..4f70dcca95 100644
--- a/src/meta-srv/src/procedure/tests.rs
+++ b/src/meta-srv/src/procedure/tests.rs
@@ -160,6 +160,7 @@ fn test_region_request_builder() {
         path: String::new(),
         options: HashMap::new(),
         partition: None,
+        requirements: None,
     };
     assert_eq!(template.template(), &expected);
 }
diff --git a/src/metric-engine/src/engine/create.rs b/src/metric-engine/src/engine/create.rs
index 34aeb5577c..c0b5da4145 100644
--- a/src/metric-engine/src/engine/create.rs
+++ b/src/metric-engine/src/engine/create.rs
@@ -497,6 +497,7 @@ impl MetricEngineInner {
             table_dir: request.table_dir.clone(),
             path_type: PathType::Metadata,
             partition_expr_json: Some("".to_string()),
+            requirements: request.requirements,
         }
     }
 
@@ -654,7 +655,7 @@ mod test {
     use common_meta::ddl::utils::{parse_column_metadatas, parse_manifest_infos_from_extensions};
     use common_query::prelude::{greptime_timestamp, greptime_value};
     use store_api::metric_engine_consts::{METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY};
-    use store_api::region_request::BatchRegionDdlRequest;
+    use store_api::region_request::{BatchRegionDdlRequest, RegionRequirements};
 
     use super::*;
     use crate::config::EngineConfig;
@@ -699,6 +700,7 @@ mod test {
             primary_key: vec![],
             options: HashMap::new(),
             partition_expr_json: Some("".to_string()),
+            requirements: RegionRequirements::object_storage(),
         };
         let result = MetricEngineInner::verify_region_create_request(&request);
         assert!(result.is_err());
@@ -748,6 +750,7 @@ mod test {
                 .into_iter()
                 .collect(),
             partition_expr_json: Some("".to_string()),
+            requirements: Default::default(),
         };
         MetricEngineInner::verify_region_create_request(&request).unwrap();
 
@@ -790,6 +793,7 @@ mod test {
                 .into_iter()
                 .collect(),
             partition_expr_json: Some("".to_string()),
+            requirements: Default::default(),
         };
         MetricEngineInner::verify_region_create_request(&request).unwrap();
     }
@@ -823,6 +827,7 @@ mod test {
             primary_key: vec![],
             options: HashMap::new(),
             partition_expr_json: Some("".to_string()),
+            requirements: Default::default(),
         };
         MetricEngineInner::verify_region_create_request(&request).unwrap_err();
 
@@ -876,6 +881,7 @@ mod test {
             table_dir: "/test_dir".to_string(),
             path_type: PathType::Bare,
             partition_expr_json: Some("".to_string()),
+            requirements: RegionRequirements::object_storage(),
         };
 
         // set up
@@ -893,6 +899,10 @@ mod test {
             vec![ReservedColumnId::table_id(), ReservedColumnId::tsid(), 1]
         );
         assert!(data_region_request.options.contains_key("ttl"));
+        assert_eq!(
+            data_region_request.requirements,
+            RegionRequirements::object_storage()
+        );
 
         // check create metadata region request
         let metadata_region_request = engine_inner.create_request_for_metadata_region(&request);
@@ -903,6 +913,10 @@ mod test {
             "forever"
         );
         assert!(!metadata_region_request.options.contains_key("skip_wal"));
+        assert_eq!(
+            metadata_region_request.requirements,
+            RegionRequirements::object_storage()
+        );
     }
 
     #[tokio::test]
@@ -951,6 +965,7 @@ mod test {
             table_dir: "/test_dir".to_string(),
             path_type: PathType::Bare,
             partition_expr_json: Some("".to_string()),
+            requirements: Default::default(),
         };
 
         let env = TestEnv::new().await;
diff --git a/src/metric-engine/src/engine/create/extract_new_columns.rs b/src/metric-engine/src/engine/create/extract_new_columns.rs
index 82cc53ce3d..9859caa06b 100644
--- a/src/metric-engine/src/engine/create/extract_new_columns.rs
+++ b/src/metric-engine/src/engine/create/extract_new_columns.rs
@@ -97,6 +97,7 @@ mod tests {
                     table_dir: "test".to_string(),
                     path_type: PathType::Bare,
                     partition_expr_json: Some("".to_string()),
+                    requirements: Default::default(),
                 },
             ),
             (
@@ -118,6 +119,7 @@ mod tests {
                     table_dir: "test".to_string(),
                     path_type: PathType::Bare,
                     partition_expr_json: Some("".to_string()),
+                    requirements: Default::default(),
                 },
             ),
         ];
@@ -171,6 +173,7 @@ mod tests {
                 table_dir: "test".to_string(),
                 path_type: PathType::Bare,
                 partition_expr_json: Some("".to_string()),
+                requirements: Default::default(),
             },
         )];
 
diff --git a/src/metric-engine/src/test_util.rs b/src/metric-engine/src/test_util.rs
index 8d4a822b6b..71e5495eca 100644
--- a/src/metric-engine/src/test_util.rs
+++ b/src/metric-engine/src/test_util.rs
@@ -189,6 +189,7 @@ impl TestEnv {
             table_dir: table_dir.to_string(),
             path_type: PathType::Bare, // Use Bare path type for engine regions
             partition_expr_json: Some("".to_string()),
+            requirements: Default::default(),
         };
 
         // create physical region
@@ -374,6 +375,7 @@ pub fn create_logical_region_request(
         table_dir: table_dir.to_string(),
         path_type: PathType::Bare, // Use Bare path type for engine regions
         partition_expr_json: Some("".to_string()),
+        requirements: Default::default(),
     }
 }
 
diff --git a/src/mito2/src/engine/basic_test.rs b/src/mito2/src/engine/basic_test.rs
index 8af1773a64..805f43d3ba 100644
--- a/src/mito2/src/engine/basic_test.rs
+++ b/src/mito2/src/engine/basic_test.rs
@@ -34,6 +34,7 @@ use rstest_reuse::{self, apply};
 use store_api::metadata::ColumnMetadata;
 use store_api::region_request::{
     PathType, RegionCreateRequest, RegionFlushRequest, RegionOpenRequest, RegionPutRequest,
+    RegionRequirements,
 };
 use store_api::storage::RegionId;
 
@@ -83,6 +84,23 @@ async fn test_engine_new_stop_with_format(flat_format: bool) {
     );
 }
 
+#[tokio::test]
+async fn test_create_region_with_empty_requirements() {
+    let mut env = TestEnv::with_prefix("create-empty-requirements").await;
+    let engine = env.create_engine(MitoConfig::default()).await;
+
+    let region_id = RegionId::new(1, 1);
+    let mut request = CreateRequestBuilder::new().build();
+    request.requirements = RegionRequirements::empty();
+
+    engine
+        .handle_request(region_id, RegionRequest::Create(request))
+        .await
+        .unwrap();
+
+    assert!(engine.is_region_exists(region_id));
+}
+
 #[tokio::test]
 async fn test_write_to_region() {
     test_write_to_region_with_format(false).await;
@@ -807,6 +825,7 @@ async fn test_cache_null_primary_key_with_format(flat_format: bool) {
         table_dir: "test".to_string(),
         path_type: PathType::Bare,
         partition_expr_json: Some("".to_string()),
+        requirements: Default::default(),
     };
 
     let column_schemas = rows_schema(&request);
diff --git a/src/mito2/src/error.rs b/src/mito2/src/error.rs
index b4d8342d20..2cbe54c6f4 100644
--- a/src/mito2/src/error.rs
+++ b/src/mito2/src/error.rs
@@ -931,12 +931,12 @@ pub enum Error {
     },
 
     #[snafu(display(
-        "Region {} does not satisfy open requirement '{}': {}",
+        "Region {} does not satisfy requirement '{}': {}",
         region_id,
         requirement,
         reason
     ))]
-    OpenRegionRequirement {
+    RegionRequirement {
         region_id: RegionId,
         requirement: &'static str,
         reason: &'static str,
@@ -1404,7 +1404,7 @@ impl ErrorExt for Error {
             PrimaryKeyLengthMismatch { .. } => StatusCode::InvalidArguments,
             InvalidSender { .. } => StatusCode::InvalidArguments,
             InvalidSchedulerState { .. } => StatusCode::InvalidArguments,
-            OpenRegionRequirement { .. } => StatusCode::InvalidArguments,
+            RegionRequirement { .. } => StatusCode::InvalidArguments,
             DeleteSsts { .. } | DeleteIndex { .. } | DeleteIndexes { .. } => {
                 StatusCode::StorageUnavailable
             }
diff --git a/src/mito2/src/region/opener.rs b/src/mito2/src/region/opener.rs
index 84364d6baf..c5193d1306 100644
--- a/src/mito2/src/region/opener.rs
+++ b/src/mito2/src/region/opener.rs
@@ -217,8 +217,11 @@ impl RegionOpener {
         Ok(self)
     }
 
-    /// Ensures the current region open request satisfies its requirements.
-    pub(crate) fn ensure_open_requirements(&self, requirements: RegionRequirements) -> Result<()> {
+    /// Ensures the current region request satisfies its requirements.
+    pub(crate) fn ensure_region_requirements(
+        &self,
+        requirements: RegionRequirements,
+    ) -> Result<()> {
         if !requirements.object_storage {
             return Ok(());
         }
@@ -230,7 +233,7 @@ impl RegionOpener {
 
         ensure!(
             supports_open_region_object_storage_requirement(&object_store),
-            error::OpenRegionRequirementSnafu {
+            error::RegionRequirementSnafu {
                 region_id: self.region_id,
                 requirement: "object storage",
                 reason: "region data must be accessible from another datanode",
diff --git a/src/mito2/src/test_util.rs b/src/mito2/src/test_util.rs
index 9d7600ba57..240bac0f4f 100644
--- a/src/mito2/src/test_util.rs
+++ b/src/mito2/src/test_util.rs
@@ -907,6 +907,7 @@ impl CreateRequestBuilder {
             table_dir: self.table_dir.clone(),
             path_type: PathType::Bare,
             partition_expr_json: self.partition_expr_json.clone(),
+            requirements: Default::default(),
         }
     }
 
@@ -973,6 +974,7 @@ impl CreateRequestBuilder {
             table_dir: self.table_dir.clone(),
             path_type: PathType::Bare,
             partition_expr_json: self.partition_expr_json.clone(),
+            requirements: Default::default(),
         }
     }
 }
diff --git a/src/mito2/src/worker/handle_create.rs b/src/mito2/src/worker/handle_create.rs
index c1ab18d59d..6e43e74a0e 100644
--- a/src/mito2/src/worker/handle_create.rs
+++ b/src/mito2/src/worker/handle_create.rs
@@ -54,7 +54,8 @@ impl<S: LogStore> RegionWorkerLoop<S> {
         }
 
         // Create a MitoRegion from the RegionMetadata.
-        let region = RegionOpener::new(
+        let requirements = request.requirements;
+        let opener = RegionOpener::new(
             region_id,
             &request.table_dir,
             request.path_type,
@@ -70,12 +71,15 @@ impl<S: LogStore> RegionWorkerLoop<S> {
         .metadata_builder(builder)
         .parse_options(request.options)?
         .cache(Some(self.cache_manager.clone()))
-        .hook(self.plugins.get())
-        .create_or_open(&self.config, &self.wal)
-        .await?;
+        .hook(self.plugins.get());
+
+        opener.ensure_region_requirements(requirements)?;
+
+        let region = opener.create_or_open(&self.config, &self.wal).await?;
 
         info!(
-            "A new region created, worker: {}, region: {:?}",
+            "A new region created with requirement {:?}, worker: {}, region: {:?}",
+            requirements,
             self.id,
             region.metadata()
         );
diff --git a/src/mito2/src/worker/handle_open.rs b/src/mito2/src/worker/handle_open.rs
index 15e800c270..3d3520b0cc 100644
--- a/src/mito2/src/worker/handle_open.rs
+++ b/src/mito2/src/worker/handle_open.rs
@@ -119,7 +119,7 @@ impl<S: LogStore> RegionWorkerLoop<S> {
             }
         };
 
-        if let Err(err) = opener.ensure_open_requirements(requirements) {
+        if let Err(err) = opener.ensure_region_requirements(requirements) {
             sender.send(Err(err));
             return;
         }
@@ -141,8 +141,9 @@ impl<S: LogStore> RegionWorkerLoop<S> {
             match opener.open(&config, &wal).await {
                 Ok(region) => {
                     info!(
-                        "Region {} is opened, worker: {}, elapsed: {:?}",
+                        "Region {} is opened with requirements {:?}, worker: {}, elapsed: {:?}",
                         region_id,
+                        requirements,
                         worker_id,
                         now.elapsed()
                     );
diff --git a/src/store-api/src/region_request.rs b/src/store-api/src/region_request.rs
index 8ab2e3abd2..31cf6e1b61 100644
--- a/src/store-api/src/region_request.rs
+++ b/src/store-api/src/region_request.rs
@@ -261,6 +261,10 @@ fn parse_region_create(create: CreateRequest) -> Result<(RegionId, RegionCreateR
             table_dir,
             path_type: PathType::Bare,
             partition_expr_json,
+            requirements: create
+                .requirements
+                .map(RegionRequirements::from)
+                .unwrap_or_default(),
         },
     ))
 }
@@ -500,6 +504,8 @@ pub struct RegionCreateRequest {
     /// Partition expression JSON from table metadata. Set to empty string for a region without partition.
     /// `Option` to keep compatibility with old clients.
     pub partition_expr_json: Option<String>,
+    /// Requirements for creating the region.
+    pub requirements: RegionRequirements,
 }
 
 impl RegionCreateRequest {
@@ -591,6 +597,22 @@ impl RegionRequirements {
     }
 }
 
+impl From<api::v1::region::RegionRequirements> for RegionRequirements {
+    fn from(value: api::v1::region::RegionRequirements) -> Self {
+        Self {
+            object_storage: value.object_storage,
+        }
+    }
+}
+
+impl From<RegionRequirements> for api::v1::region::RegionRequirements {
+    fn from(value: RegionRequirements) -> Self {
+        Self {
+            object_storage: value.object_storage,
+        }
+    }
+}
+
 /// Open region request.
 #[derive(Debug, Clone)]
 pub struct RegionOpenRequest {
@@ -2065,11 +2087,58 @@ mod tests {
             table_dir: "path".to_string(),
             path_type: PathType::Bare,
             partition_expr_json: Some("".to_string()),
+            requirements: Default::default(),
         };
 
         assert!(create.validate().is_err());
     }
 
+    #[test]
+    fn test_parse_create_region_requirements_defaults_to_empty() {
+        let create = CreateRequest {
+            region_id: RegionId::new(42, 0).as_u64(),
+            engine: "mito".to_string(),
+            column_defs: vec![],
+            primary_key: vec![],
+            path: "test".to_string(),
+            options: HashMap::new(),
+            partition: None,
+            requirements: None,
+        };
+
+        let requests =
+            RegionRequest::try_from_request_body(region_request::Body::Create(create)).unwrap();
+        let RegionRequest::Create(request) = &requests[0].1 else {
+            unreachable!()
+        };
+
+        assert_eq!(request.requirements, RegionRequirements::empty());
+    }
+
+    #[test]
+    fn test_parse_create_region_requirements_from_proto() {
+        let create = CreateRequest {
+            region_id: RegionId::new(42, 0).as_u64(),
+            engine: "mito".to_string(),
+            column_defs: vec![],
+            primary_key: vec![],
+            path: "test".to_string(),
+            options: HashMap::new(),
+            partition: None,
+            requirements: Some(api::v1::region::RegionRequirements {
+                object_storage: true,
+            }),
+        };
+
+        let requests =
+            RegionRequest::try_from_request_body(region_request::Body::Create(create)).unwrap();
+        let RegionRequest::Create(request) = &requests[0].1 else {
+            unreachable!()
+        };
+
+        assert_eq!(request.requirements, RegionRequirements::object_storage());
+    }
+
     #[test]
     fn test_validate_modify_column_fulltext_options() {
         let kind = AlterKind::SetIndexes {