diff --git a/flex/codegen/src/codegen_utils.h b/flex/codegen/src/codegen_utils.h
index 563cd6544521..a8ab552edf51 100644
--- a/flex/codegen/src/codegen_utils.h
+++ b/flex/codegen/src/codegen_utils.h
@@ -132,10 +132,23 @@ std::string generate_output_list(std::string input_name, int32_t input_size,
 // check type consistent
 bool data_type_consistent(const common::DataType& left,
                           const common::DataType& right) {
-  if (left == common::DataType::NONE || right == common::DataType::NONE) {
-    return true;
+  if (left.item_case() == common::DataType::ITEM_NOT_SET) {
+    return false;
+  }
+  if (left.item_case() != right.item_case()) {
+    return false;
+  }
+  if (left.item_case() == common::DataType::kPrimitiveType) {
+    return left.primitive_type() == right.primitive_type();
+  } else if (left.item_case() == common::DataType::kArray ||
+             left.item_case() == common::DataType::kMap) {
+    LOG(FATAL) << "Not support list or map type";
+  } else if (left.item_case() == common::DataType::kString) {
+    return true;  // string type is always consistent
+  } else {
+    LOG(FATAL) << "Unexpected data type";
+    return false;
   }
-  return left == right;
 }
 
 std::tuple<std::string, std::string> decode_param_from_decoder(
diff --git a/flex/codegen/src/graph_types.h b/flex/codegen/src/graph_types.h
index fa0316b399a4..c9e45c2e2df5 100644
--- a/flex/codegen/src/graph_types.h
+++ b/flex/codegen/src/graph_types.h
@@ -19,6 +19,7 @@ limitations under the License.
 #include <type_traits>
 
 #include "flex/codegen/src/string_utils.h"
+#include "flex/proto_generated_gie/basic_type.pb.h"
 #include "flex/proto_generated_gie/common.pb.h"
 #include "glog/logging.h"
 #include "google/protobuf/any.h"
@@ -63,62 +64,93 @@ inline bool operator==(const ParamConst& lhs, const ParamConst& rhs) {
 
 }  // namespace codegen
 
-static codegen::DataType common_data_type_pb_2_data_type(
-    const common::DataType& data_type) {
-  switch (data_type) {
-  case common::DataType::INT32:
+static codegen::DataType primitive_type_to_data_type(
+    const common::PrimitiveType& type) {
+  switch (type) {
+  case common::PrimitiveType::DT_SIGNED_INT32:
     return codegen::DataType::kInt32;
-  case common::DataType::INT64:
+  case common::PrimitiveType::DT_SIGNED_INT64:
     return codegen::DataType::kInt64;
-  case common::DataType::DOUBLE:
+  case common::PrimitiveType::DT_FLOAT:
+    return codegen::DataType::kFloat;
+  case common::PrimitiveType::DT_DOUBLE:
     return codegen::DataType::kDouble;
-  case common::DataType::STRING:
-    return codegen::DataType::kString;
-  case common::DataType::INT64_ARRAY:
-    return codegen::DataType::kInt64Array;
-  case common::DataType::INT32_ARRAY:
-    return codegen::DataType::kInt32Array;
-  case common::DataType::BOOLEAN:
+  case common::PrimitiveType::DT_BOOL:
     return codegen::DataType::kBoolean;
-  case common::DataType::DATE32:
-    return codegen::DataType::kDate;
-  case common::DataType::TIME32:
-    return codegen::DataType::kTime;
-  case common::DataType::TIMESTAMP:
-    return codegen::DataType::kTimeStamp;
+  default:
+    // LOG(FATAL) << "unknown primitive type";
+    throw std::runtime_error(
+        "unknown primitive type when converting primitive type to data type:" +
+        std::to_string(static_cast<int>(type)));
+  }
+}
+
+static codegen::DataType common_data_type_pb_2_data_type(
+    const common::DataType& data_type) {
+  switch (data_type.item_case()) {
+  case common::DataType::ItemCase::kPrimitiveType:
+    return primitive_type_to_data_type(data_type.primitive_type());
+  case common::DataType::ItemCase::kDecimal:
+    LOG(FATAL) << "Not support decimal type";
+  case common::DataType::ItemCase::kString:
+    return codegen::DataType::kString;
+  case common::DataType::ItemCase::kTemporal:
+    LOG(FATAL) << "Not support temporal type";
+  case common::DataType::ItemCase::kArray:
+  case common::DataType::ItemCase::kMap:
+    LOG(FATAL) << "Not support array or map type";
   default:
     // LOG(FATAL) << "unknown data type";
     throw std::runtime_error(
         "unknown data type when converting common_data_type to inner data "
         "type:" +
-        std::to_string(static_cast<int>(data_type)));
+        data_type.DebugString());
   }
 }
 
-static std::string single_common_data_type_pb_2_str(
-    const common::DataType& data_type) {
-  switch (data_type) {
-  case common::DataType::BOOLEAN:
-    return "bool";
-  case common::DataType::INT32:
+static std::string primitive_type_to_str(const common::PrimitiveType& type) {
+  switch (type) {
+  case common::PrimitiveType::DT_SIGNED_INT32:
     return "int32_t";
-  case common::DataType::INT64:
+  case common::PrimitiveType::DT_UNSIGNED_INT32:
+    return "uint32_t";
+  case common::PrimitiveType::DT_SIGNED_INT64:
     return "int64_t";
-  case common::DataType::DOUBLE:
+  case common::PrimitiveType::DT_UNSIGNED_INT64:
+    return "uint64_t";
+  case common::PrimitiveType::DT_FLOAT:
+    return "float";
+  case common::PrimitiveType::DT_DOUBLE:
     return "double";
-  case common::DataType::STRING:
+  case common::PrimitiveType::DT_BOOL:
+    return "bool";
+  default:
+    // LOG(FATAL) << "unknown primitive type";
+    throw std::runtime_error(
+        "unknown primitive type when converting primitive type to string:" +
+        std::to_string(static_cast<int>(type)));
+  }
+}
+
+static std::string single_common_data_type_pb_2_str(
+    const common::DataType& data_type) {
+  switch (data_type.item_case()) {
+  case common::DataType::ItemCase::kPrimitiveType:
+    return primitive_type_to_str(data_type.primitive_type());
+  case common::DataType::ItemCase::kDecimal:
+    LOG(FATAL) << "Not support decimal type";
+  case common::DataType::ItemCase::kString:
     return "std::string_view";
-  case common::DataType::INT64_ARRAY:
-    return "std::vector<int64_t>";
-  case common::DataType::INT32_ARRAY:
-    return "std::vector<int32_t>";
-  case common::DataType::DATE32:
-    return "Date";
+  case common::DataType::ItemCase::kTemporal:
+    LOG(FATAL) << "Not support temporal type";
+  case common::DataType::ItemCase::kArray:
+  case common::DataType::ItemCase::kMap:
+    LOG(FATAL) << "Not support array or map type";
     // TODO: support time32 and timestamp
   default:
     throw std::runtime_error(
         "unknown data type when convert common data type to string:" +
-        std::to_string(static_cast<int>(data_type)));
+        data_type.DebugString());
   }
 }
 
@@ -266,21 +298,22 @@ static std::string data_type_2_rust_string(const codegen::DataType& data_type) {
 }
 
 static common::DataType common_value_2_data_type(const common::Value& value) {
+  common::DataType ret;
   switch (value.item_case()) {
   case common::Value::kI32:
-    return common::DataType::INT32;
+    ret.set_primitive_type(common::PrimitiveType::DT_SIGNED_INT32);
   case common::Value::kI64:
-    return common::DataType::INT64;
+    ret.set_primitive_type(common::PrimitiveType::DT_SIGNED_INT64);
   case common::Value::kBoolean:
-    return common::DataType::BOOLEAN;
+    ret.set_primitive_type(common::PrimitiveType::DT_BOOL);
   case common::Value::kF64:
-    return common::DataType::DOUBLE;
+    ret.set_primitive_type(common::PrimitiveType::DT_DOUBLE);
   case common::Value::kStr:
-    return common::DataType::STRING;
+    ret.mutable_string()->mutable_long_text();
   default:
     LOG(FATAL) << "unknown value" << value.DebugString();
   }
-  return common::DataType::NONE;
+  return ret;
 }
 
 static void parse_param_const_from_pb(
diff --git a/flex/codegen/src/hqps/hqps_case_when_builder.h b/flex/codegen/src/hqps/hqps_case_when_builder.h
index f3ea57a16b21..9c867fee5cb9 100644
--- a/flex/codegen/src/hqps/hqps_case_when_builder.h
+++ b/flex/codegen/src/hqps/hqps_case_when_builder.h
@@ -118,9 +118,8 @@ class CaseWhenBuilder : public ExprBuilder {
 
     std::string str = formater.str();
 
-    return std::make_tuple(
-        class_name_, construct_params_, tag_selectors_, str,
-        std::vector{common::DataType::DataType_INT_MIN_SENTINEL_DO_NOT_USE_});
+    return std::make_tuple(class_name_, construct_params_, tag_selectors_, str,
+                           std::vector{common::DataType()});
   }
 
  protected:
diff --git a/flex/codegen/src/hqps/hqps_edge_expand_builder.h b/flex/codegen/src/hqps/hqps_edge_expand_builder.h
index 9a2b678d7730..0deb25f41f4a 100644
--- a/flex/codegen/src/hqps/hqps_edge_expand_builder.h
+++ b/flex/codegen/src/hqps/hqps_edge_expand_builder.h
@@ -224,7 +224,9 @@ static void BuildExprFromPredicate(BuildingContext& ctx,
                                    std::string& func_construct_params_str,
                                    std::string& property_selectors_str) {
   auto expr_builder = ExprBuilder(ctx);
-  expr_builder.set_return_type(common::DataType::BOOLEAN);
+  common::DataType type;
+  type.set_primitive_type(common::PrimitiveType::DT_BOOL);
+  expr_builder.set_return_type(type);
   expr_builder.AddAllExprOpr(expr.operators());
   std::string expr_code;
   std::vector<codegen::ParamConst> func_call_param_const;
diff --git a/flex/codegen/src/hqps/hqps_get_v_builder.h b/flex/codegen/src/hqps/hqps_get_v_builder.h
index e9c7b95231c6..202b2ffcdff2 100644
--- a/flex/codegen/src/hqps/hqps_get_v_builder.h
+++ b/flex/codegen/src/hqps/hqps_get_v_builder.h
@@ -135,7 +135,9 @@ class GetVOpBuilder {
 
       auto& expr_oprs = expr.operators();
       expr_builder.AddAllExprOpr(expr_oprs);
-      expr_builder.set_return_type(common::DataType::BOOLEAN);
+      common::DataType data_type;
+      data_type.set_primitive_type(common::PrimitiveType::DT_BOOL);
+      expr_builder.set_return_type(data_type);
       std::vector<common::DataType> unused_expr_ret_type;
       if (!expr_builder.empty()) {
         std::tie(expr_name_, expr_call_param_, tag_properties_, expr_code_,
diff --git a/flex/codegen/src/hqps/hqps_scan_builder.h b/flex/codegen/src/hqps/hqps_scan_builder.h
index cf210adf30f8..93df39b9ae57 100644
--- a/flex/codegen/src/hqps/hqps_scan_builder.h
+++ b/flex/codegen/src/hqps/hqps_scan_builder.h
@@ -149,7 +149,9 @@ class ScanOpBuilder {
 
       // TODO: make expr_builder a member of ScanOpBuilder
       // auto expr_builder = ExprBuilder(ctx_);
-      expr_builder_.set_return_type(common::DataType::BOOLEAN);
+      common::DataType type;
+      type.set_primitive_type(common::PrimitiveType::DT_BOOL);
+      expr_builder_.set_return_type(common::DataType(type));
       // Add extra (, ) to wrap the code, since we may append index_predicate
       // afterwards.
       common::ExprOpr left_brace, right_brace;
diff --git a/flex/codegen/src/hqps/hqps_select_builder.h b/flex/codegen/src/hqps/hqps_select_builder.h
index 4653f24e24a5..3bdd68867437 100644
--- a/flex/codegen/src/hqps/hqps_select_builder.h
+++ b/flex/codegen/src/hqps/hqps_select_builder.h
@@ -41,7 +41,9 @@ class SelectOpBuilder {
 
   SelectOpBuilder& expr(const common::Expression expr) {
     ExprBuilder expr_builder(ctx_);
-    expr_builder.set_return_type(common::DataType::BOOLEAN);
+    common::DataType data_type;
+    data_type.set_primitive_type(common::PrimitiveType::DT_BOOL);
+    expr_builder.set_return_type(data_type);
     expr_builder.AddAllExprOpr(expr.operators());
 
     std::string func_code;
diff --git a/flex/codegen/src/pegasus/pegasus_order_by_builder.h b/flex/codegen/src/pegasus/pegasus_order_by_builder.h
index d948c07fcd5c..e799da4c2b6e 100644
--- a/flex/codegen/src/pegasus/pegasus_order_by_builder.h
+++ b/flex/codegen/src/pegasus/pegasus_order_by_builder.h
@@ -72,20 +72,29 @@ class OrderByOpBuilder {
         ss << ".then(";
       }
       std::string cmp_type;
-      switch (data_type) {
-      case common::DataType::BOOLEAN:
-      case common::DataType::INT32:
-      case common::DataType::INT64:
-      case common::DataType::STRING: {
-        cmp_type = "cmp";
-        break;
+      switch (data_type.item_case()) {
+      case common::DataType::kPrimitiveType: {
+        switch (data_type.primitive_type()) {
+        case common::PrimitiveType::DT_BOOL:
+        case common::PrimitiveType::DT_SIGNED_INT32:
+        case common::PrimitiveType::DT_SIGNED_INT64:
+          cmp_type = "cmp";
+          break;
+        case common::PrimitiveType::DT_DOUBLE: {
+          cmp_type = "partial_cmp";
+          break;
+        }
+        default:
+          LOG(FATAL) << "Unsupported type "
+                     << static_cast<int32_t>(data_type.primitive_type());
+        }
       }
-      case common::DataType::DOUBLE: {
-        cmp_type = "partial_cmp";
+      case common::DataType::kString: {
+        cmp_type = "cmp";
         break;
       }
       default:
-        LOG(FATAL) << "Unsupported type " << data_type;
+        LOG(FATAL) << "Unsupported type " << data_type.DebugString();
       }
       std::string reverse_str;
       if (ordering_pair_[i].order() == algebra::OrderBy_OrderingPair_Order::
diff --git a/flex/codegen/src/pegasus/pegasus_project_builder.h b/flex/codegen/src/pegasus/pegasus_project_builder.h
index 7ff5d4883821..f0595aa15797 100644
--- a/flex/codegen/src/pegasus/pegasus_project_builder.h
+++ b/flex/codegen/src/pegasus/pegasus_project_builder.h
@@ -176,14 +176,24 @@ class ProjectOpBuilder {
           ctx_.SetOutput(i, data_types);
         } else if (column_meta.type().type_case() ==
                    common::IrDataType::kDataType) {
-          switch (column_meta.type().data_type()) {
-          case common::DataType::INT64: {
-            std::vector<codegen::DataType> data_types;
-            data_types.push_back(codegen::DataType::kInt64);
-            ctx_.SetOutput(i, data_types);
-            break;
+          switch (column_meta.type().data_type().item_case()) {
+          case common::DataType::kPrimitiveType: {
+            auto data_type = column_meta.type().data_type().primitive_type();
+            switch (data_type) {
+            case common::PrimitiveType::DT_SIGNED_INT64: {
+              std::vector<codegen::DataType> data_types;
+              data_types.push_back(codegen::DataType::kInt64);
+              ctx_.SetOutput(i, data_types);
+              break;
+            }
+            default: {
+              std::vector<codegen::DataType> data_types;
+              data_types.push_back(codegen::DataType::kString);
+              ctx_.SetOutput(i, data_types);
+            }
+            }
           }
-          case common::DataType::STRING: {
+          case common::DataType::kString: {
             std::vector<codegen::DataType> data_types;
             data_types.push_back(codegen::DataType::kString);
             ctx_.SetOutput(i, data_types);
diff --git a/flex/engines/graph_db/runtime/adhoc/expr_impl.cc b/flex/engines/graph_db/runtime/adhoc/expr_impl.cc
index 7df4cdbeed15..b965f6045fcf 100644
--- a/flex/engines/graph_db/runtime/adhoc/expr_impl.cc
+++ b/flex/engines/graph_db/runtime/adhoc/expr_impl.cc
@@ -16,6 +16,7 @@
 #include "flex/engines/graph_db/runtime/adhoc/expr_impl.h"
 #include <regex>
 #include <stack>
+#include "flex/proto_generated_gie/basic_type.pb.h"
 
 namespace gs {
 
@@ -441,21 +442,36 @@ static RTAny parse_param(const common::DynamicParam& param,
       common::IrDataType::TypeCase::kDataType) {
     common::DataType dt = param.data_type().data_type();
     const std::string& name = param.name();
-    if (dt == common::DataType::DATE32) {
-      int64_t val = std::stoll(input.at(name));
-      return RTAny::from_int64(val);
-    } else if (dt == common::DataType::STRING) {
+    if (dt.item_case() == common::DataType::ItemCase::kPrimitiveType) {
+      switch (dt.primitive_type()) {
+      case common::PrimitiveType::DT_SIGNED_INT32: {
+        int val = std::stoi(input.at(name));
+        return RTAny::from_int32(val);
+      }
+      case common::PrimitiveType::DT_SIGNED_INT64: {
+        int64_t val = std::stoll(input.at(name));
+        return RTAny::from_int64(val);
+      }
+      case common::PrimitiveType::DT_DOUBLE:
+        return RTAny::from_double(std::stod(input.at(name)));
+      case common::PrimitiveType::DT_BOOL:
+        return RTAny::from_bool(input.at(name) == "true");
+      default:
+        LOG(FATAL) << "not support type: " << dt.DebugString();
+      }
+    } else if (dt.item_case() == common::DataType::ItemCase::kTemporal) {
+      if (dt.temporal().item_case() == common::Temporal::kDate32) {
+        int64_t val = std::stoll(input.at(name));
+        return RTAny::from_int64(val);
+      } else {
+        LOG(FATAL) << "not support type: " << dt.temporal().DebugString();
+      }
+    } else if (dt.item_case() == common::DataType::ItemCase::kString) {
       const std::string& val = input.at(name);
       return RTAny::from_string(val);
-    } else if (dt == common::DataType::INT32) {
-      int val = std::stoi(input.at(name));
-      return RTAny::from_int32(val);
-    } else if (dt == common::DataType::INT64) {
-      int64_t val = std::stoll(input.at(name));
-      return RTAny::from_int64(val);
+    } else {
+      LOG(FATAL) << "not support type: " << dt.DebugString();
     }
-
-    LOG(FATAL) << "not support type: " << common::DataType_Name(dt);
   }
   LOG(FATAL) << "graph data type not expected....";
   return RTAny();
diff --git a/flex/engines/graph_db/runtime/adhoc/operators/scan.cc b/flex/engines/graph_db/runtime/adhoc/operators/scan.cc
index d9cff3bfa4de..2dd417155427 100644
--- a/flex/engines/graph_db/runtime/adhoc/operators/scan.cc
+++ b/flex/engines/graph_db/runtime/adhoc/operators/scan.cc
@@ -16,6 +16,7 @@
 #include "flex/engines/graph_db/runtime/common/operators/scan.h"
 #include "flex/engines/graph_db/runtime/adhoc/expr_impl.h"
 #include "flex/engines/graph_db/runtime/adhoc/operators/operators.h"
+#include "flex/proto_generated_gie/basic_type.pb.h"
 namespace gs {
 
 namespace runtime {
@@ -202,22 +203,28 @@ bool parse_idx_predicate(const algebra::IndexPredicate& predicate,
     const common::DynamicParam& p = triplet.param();
     if (p.data_type().type_case() == common::IrDataType::TypeCase::kDataType) {
       auto dt = p.data_type().data_type();
-      if (dt == common::DataType::INT64) {
-        std::string name = p.name();
-        std::string value = params.at(name);
-        int64_t v = std::stoll(value);
-        oids.emplace_back(v);
-      } else if (dt == common::DataType::STRING) {
+      if (dt.item_case() == common::DataType::ItemCase::kPrimitiveType) {
+        if (dt.primitive_type() == common::PrimitiveType::DT_SIGNED_INT64) {
+          std::string name = p.name();
+          std::string value = params.at(name);
+          int64_t v = std::stoll(value);
+          oids.emplace_back(v);
+        } else if (dt.primitive_type() ==
+                   common::PrimitiveType::DT_SIGNED_INT32) {
+          std::string name = p.name();
+          std::string value = params.at(name);
+          int32_t v = std::stoi(value);
+          oids.emplace_back(v);
+        } else {
+          LOG(FATAL) << "unsupported primary key type" << dt.DebugString();
+          return false;
+        }
+      } else if (dt.item_case() == common::DataType::ItemCase::kString) {
         std::string name = p.name();
         std::string value = params.at(name);
         oids.emplace_back(Any::From(value));
-      } else if (dt == common::DataType::INT32) {
-        std::string name = p.name();
-        std::string value = params.at(name);
-        int32_t v = std::stoi(value);
-        oids.emplace_back(v);
       } else {
-        LOG(FATAL) << "unsupported primary key type" << dt;
+        LOG(FATAL) << "unsupported primary key type" << dt.item_case();
         return false;
       }
     }
diff --git a/flex/engines/graph_db/runtime/adhoc/utils.cc b/flex/engines/graph_db/runtime/adhoc/utils.cc
index bd5135b0d2ca..473b5a0c9182 100644
--- a/flex/engines/graph_db/runtime/adhoc/utils.cc
+++ b/flex/engines/graph_db/runtime/adhoc/utils.cc
@@ -15,6 +15,7 @@
 #include "flex/engines/graph_db/runtime/adhoc/utils.h"
 #include "flex/engines/graph_db/runtime/common/columns/value_columns.h"
 #include "flex/engines/graph_db/runtime/common/columns/vertex_columns.h"
+#include "flex/proto_generated_gie/basic_type.pb.h"
 
 namespace gs {
 
@@ -144,68 +145,82 @@ std::shared_ptr<IContextColumnBuilder> create_column_builder(RTAnyType type) {
   return nullptr;
 }
 
-std::shared_ptr<IContextColumn> build_optional_column(
-    const common::IrDataType& data_type, const Expr& expr, size_t row_num) {
-  switch (data_type.type_case()) {
-  case common::IrDataType::kDataType: {
-    switch (data_type.data_type()) {
-    case common::DataType::INT64: {
-      OptionalValueColumnBuilder<int64_t> builder;
-      builder.reserve(row_num);
-      for (size_t i = 0; i < row_num; ++i) {
-        auto v = expr.eval_path(i, 0);
-        if (v.is_null()) {
-          builder.push_back_null();
-        } else {
-          builder.push_back_opt(v.as_int64(), true);
-        }
+std::shared_ptr<IContextColumn> build_optional_primitive_column(
+    const common::PrimitiveType& data_type, const Expr& expr, size_t row_num) {
+  switch (data_type) {
+  case common::PrimitiveType::DT_SIGNED_INT64: {
+    OptionalValueColumnBuilder<int64_t> builder;
+    builder.reserve(row_num);
+    for (size_t i = 0; i < row_num; ++i) {
+      auto v = expr.eval_path(i, 0);
+      if (v.is_null()) {
+        builder.push_back_null();
+      } else {
+        builder.push_back_opt(v.as_int64(), true);
       }
+    }
 
-      return builder.finish();
-    } break;
-    case common::DataType::INT32: {
-      OptionalValueColumnBuilder<int> builder;
-      builder.reserve(row_num);
-      for (size_t i = 0; i < row_num; ++i) {
-        auto v = expr.eval_path(i, 0);
-        if (v.is_null()) {
-          builder.push_back_null();
-        } else {
-          builder.push_back_opt(v.as_int32(), true);
-        }
+    return builder.finish();
+  }
+  case common::PrimitiveType::DT_SIGNED_INT32: {
+    OptionalValueColumnBuilder<int> builder;
+    builder.reserve(row_num);
+    for (size_t i = 0; i < row_num; ++i) {
+      auto v = expr.eval_path(i, 0);
+      if (v.is_null()) {
+        builder.push_back_null();
+      } else {
+        builder.push_back_opt(v.as_int32(), true);
       }
+    }
 
-      return builder.finish();
-    } break;
-    case common::DataType::DOUBLE: {
-      OptionalValueColumnBuilder<double> builder;
-      builder.reserve(row_num);
-      for (size_t i = 0; i < row_num; ++i) {
-        auto v = expr.eval_path(i, 0);
-        if (v.is_null()) {
-          builder.push_back_null();
-        } else {
-          builder.push_back_opt(v.as_double(), true);
-        }
+    return builder.finish();
+  }
+  case common::PrimitiveType::DT_DOUBLE: {
+    OptionalValueColumnBuilder<double> builder;
+    builder.reserve(row_num);
+    for (size_t i = 0; i < row_num; ++i) {
+      auto v = expr.eval_path(i, 0);
+      if (v.is_null()) {
+        builder.push_back_null();
+      } else {
+        builder.push_back_opt(v.as_double(), true);
       }
+    }
 
-      return builder.finish();
-    } break;
-    case common::DataType::BOOLEAN: {
-      OptionalValueColumnBuilder<bool> builder;
-      builder.reserve(row_num);
-      for (size_t i = 0; i < row_num; ++i) {
-        auto v = expr.eval_path(i, 0);
-        if (v.is_null()) {
-          builder.push_back_null();
-        } else {
-          builder.push_back_opt(v.as_bool(), true);
-        }
+    return builder.finish();
+  }
+  case common::PrimitiveType::DT_BOOL: {
+    OptionalValueColumnBuilder<bool> builder;
+    builder.reserve(row_num);
+    for (size_t i = 0; i < row_num; ++i) {
+      auto v = expr.eval_path(i, 0);
+      if (v.is_null()) {
+        builder.push_back_null();
+      } else {
+        builder.push_back_opt(v.as_bool(), true);
       }
+    }
 
-      return builder.finish();
-    } break;
-    case common::DataType::STRING: {
+    return builder.finish();
+  }
+  default:
+    LOG(FATAL) << "not support";
+    return nullptr;
+  }
+}
+
+std::shared_ptr<IContextColumn> build_optional_column(
+    const common::IrDataType& data_type, const Expr& expr, size_t row_num) {
+  switch (data_type.type_case()) {
+  case common::IrDataType::kDataType: {
+    switch (data_type.data_type().item_case()) {
+    case common::DataType::ItemCase::kPrimitiveType: {
+      return build_optional_primitive_column(
+          data_type.data_type().primitive_type(), expr, row_num);
+    }
+
+    case common::DataType::ItemCase::kString: {
       OptionalValueColumnBuilder<std::string_view> builder;
       builder.reserve(row_num);
       for (size_t i = 0; i < row_num; ++i) {
@@ -218,25 +233,30 @@ std::shared_ptr<IContextColumn> build_optional_column(
       }
 
       return builder.finish();
-    } break;
-    case common::DataType::TIMESTAMP: {
-      OptionalValueColumnBuilder<Date> builder;
-      builder.reserve(row_num);
-      for (size_t i = 0; i < row_num; ++i) {
-        auto v = expr.eval_path(i, 0);
-        if (v.is_null()) {
-          builder.push_back_null();
-        } else {
-          builder.push_back_opt(v.as_date32(), true);
+    }
+    case common::DataType::ItemCase::kTemporal: {
+      if (data_type.data_type().temporal().item_case() ==
+          common::Temporal::kDate32) {
+        OptionalValueColumnBuilder<Date> builder;
+        builder.reserve(row_num);
+        for (size_t i = 0; i < row_num; ++i) {
+          auto v = expr.eval_path(i, 0);
+          if (v.is_null()) {
+            builder.push_back_null();
+          } else {
+            builder.push_back_opt(v.as_date32(), true);
+          }
         }
-      }
 
-      return builder.finish();
-    } break;
+        return builder.finish();
+      } else {
+        LOG(FATAL) << "not support type: "
+                   << data_type.data_type().DebugString();
+      }
+    }
 
     default: {
-      LOG(FATAL) << "not support"
-                 << common::DataType_Name(data_type.data_type());
+      LOG(FATAL) << "not support" << data_type.data_type().DebugString();
       break;
     }
     }
@@ -252,6 +272,56 @@ std::shared_ptr<IContextColumn> build_optional_column(
   return nullptr;
 }
 
+std::shared_ptr<IContextColumn> build_primitive_column(
+    const common::PrimitiveType& data_type, const Expr& expr, size_t row_num) {
+  switch (data_type) {
+  case common::PrimitiveType::DT_SIGNED_INT64: {
+    ValueColumnBuilder<int64_t> builder;
+    builder.reserve(row_num);
+    for (size_t i = 0; i < row_num; ++i) {
+      auto v = expr.eval_path(i).as_int64();
+      builder.push_back_opt(v);
+    }
+
+    return builder.finish();
+  }
+  case common::PrimitiveType::DT_SIGNED_INT32: {
+    ValueColumnBuilder<int> builder;
+    builder.reserve(row_num);
+    for (size_t i = 0; i < row_num; ++i) {
+      auto v = expr.eval_path(i).as_int32();
+      builder.push_back_opt(v);
+    }
+
+    return builder.finish();
+  }
+  case common::PrimitiveType::DT_DOUBLE: {
+    ValueColumnBuilder<double> builder;
+    builder.reserve(row_num);
+    for (size_t i = 0; i < row_num; ++i) {
+      auto v = expr.eval_path(i).as_double();
+      builder.push_back_opt(v);
+    }
+
+    return builder.finish();
+  }
+
+  case common::PrimitiveType::DT_BOOL: {
+    ValueColumnBuilder<bool> builder;
+    builder.reserve(row_num);
+    for (size_t i = 0; i < row_num; ++i) {
+      auto v = expr.eval_path(i).as_bool();
+      builder.push_back_opt(v);
+    }
+
+    return builder.finish();
+  }
+  default:
+    LOG(FATAL) << "not support";
+    return nullptr;
+  }
+}
+
 std::shared_ptr<IContextColumn> build_column(
     const common::IrDataType& data_type, const Expr& expr, size_t row_num) {
   if (expr.is_optional()) {
@@ -259,28 +329,12 @@ std::shared_ptr<IContextColumn> build_column(
   }
   switch (data_type.type_case()) {
   case common::IrDataType::kDataType: {
-    switch (data_type.data_type()) {
-    case common::DataType::INT64: {
-      ValueColumnBuilder<int64_t> builder;
-      builder.reserve(row_num);
-      for (size_t i = 0; i < row_num; ++i) {
-        auto v = expr.eval_path(i).as_int64();
-        builder.push_back_opt(v);
-      }
-
-      return builder.finish();
-    } break;
-    case common::DataType::INT32: {
-      ValueColumnBuilder<int> builder;
-      builder.reserve(row_num);
-      for (size_t i = 0; i < row_num; ++i) {
-        auto v = expr.eval_path(i).as_int32();
-        builder.push_back_opt(v);
-      }
-
-      return builder.finish();
-    } break;
-    case common::DataType::STRING: {
+    switch (data_type.data_type().item_case()) {
+    case common::DataType::ItemCase::kPrimitiveType: {
+      return build_primitive_column(data_type.data_type().primitive_type(),
+                                    expr, row_num);
+    }
+    case common::DataType::ItemCase::kString: {
       ValueColumnBuilder<std::string_view> builder;
       builder.reserve(row_num);
       for (size_t i = 0; i < row_num; ++i) {
@@ -289,61 +343,49 @@ std::shared_ptr<IContextColumn> build_column(
       }
 
       return builder.finish();
-    } break;
-    case common::DataType::DATE32: {
-      ValueColumnBuilder<Date> builder;
-      builder.reserve(row_num);
-      for (size_t i = 0; i < row_num; ++i) {
-        auto v = expr.eval_path(i).as_date32();
-        builder.push_back_opt(v);
-      }
-
-      return builder.finish();
-    } break;
-    case common::DataType::STRING_ARRAY: {
-      ValueColumnBuilder<std::set<std::string>> builder;
-      builder.reserve(row_num);
-      for (size_t i = 0; i < row_num; ++i) {
-        const auto& v = expr.eval_path(i).as_string_set();
-        builder.push_back_opt(v);
-      }
+    }
+    case common::DataType::ItemCase::kTemporal: {
+      if (data_type.data_type().temporal().item_case() ==
+              common::Temporal::kDate32 ||
+          data_type.data_type().temporal().item_case() ==
+              common::Temporal::kTimestamp) {
+        ValueColumnBuilder<Date> builder;
+        builder.reserve(row_num);
+        for (size_t i = 0; i < row_num; ++i) {
+          auto v = expr.eval_path(i).as_date32();
+          builder.push_back_opt(v);
+        }
 
-      return builder.finish();
-    } break;
-    case common::DataType::TIMESTAMP: {
-      ValueColumnBuilder<Date> builder;
-      builder.reserve(row_num);
-      for (size_t i = 0; i < row_num; ++i) {
-        auto v = expr.eval_path(i).as_date32();
-        builder.push_back_opt(v);
+        return builder.finish();
+      } else {
+        LOG(FATAL) << "not support type: "
+                   << data_type.data_type().temporal().DebugString();
       }
+    }
 
-      return builder.finish();
-    } break;
-    case common::DataType::BOOLEAN: {
-      ValueColumnBuilder<bool> builder;
-      builder.reserve(row_num);
-      for (size_t i = 0; i < row_num; ++i) {
-        auto v = expr.eval_path(i).as_bool();
-        builder.push_back_opt(v);
-      }
-      return builder.finish();
-    } break;
-    case common::DataType::DOUBLE: {
-      ValueColumnBuilder<double> builder;
-      builder.reserve(row_num);
-      for (size_t i = 0; i < row_num; ++i) {
-        auto v = expr.eval_path(i).as_double();
-        builder.push_back_opt(v);
+    case common::DataType::ItemCase::kArray: {
+      if (data_type.data_type().array().component_type().item_case() ==
+          common::DataType::ItemCase::kString) {
+        ValueColumnBuilder<std::set<std::string>> builder;
+        builder.reserve(row_num);
+        for (size_t i = 0; i < row_num; ++i) {
+          const auto& v = expr.eval_path(i).as_string_set();
+          builder.push_back_opt(v);
+        }
+        return builder.finish();
+      } else {
+        LOG(FATAL) << "not support: " << data_type.DebugString();
       }
-      return builder.finish();
-    } break;
+    }
+    case common::DataType::ItemCase::kMap: {
+      LOG(FATAL) << "not support: " << data_type.DebugString();
+    }
     default: {
-      LOG(FATAL) << "not support: "
-                 << common::DataType_Name(data_type.data_type());
+      LOG(FATAL) << "not support" << data_type.data_type().DebugString();
+      break;
     }
     }
-  } break;
+
   case common::IrDataType::kGraphType: {
     const common::GraphDataType& graph_data_type = data_type.graph_type();
     common::GraphDataType_GraphElementOpt elem_opt =
@@ -383,17 +425,17 @@ std::shared_ptr<IContextColumn> build_column(
     } else {
       LOG(FATAL) << "unexpected type";
     }
-  } break;
+  }
   case common::IrDataType::TYPE_NOT_SET: {
     return build_column_beta(expr, row_num);
-  } break;
+  }
   default:
-    LOG(FATAL) << "unexpected type"
-               << common::DataType_Name(data_type.data_type());
+    LOG(FATAL) << "unexpected type" << data_type.data_type().DebugString();
     break;
   }
 
-  return nullptr;
+    return nullptr;
+  }
 }
 
 std::shared_ptr<IContextColumn> build_optional_column_beta(const Expr& expr,
@@ -563,7 +605,6 @@ std::shared_ptr<IContextColumn> build_column_beta(const Expr& expr,
 
   return nullptr;
 }
-
 }  // namespace runtime
 
 }  // namespace gs
diff --git a/flex/engines/graph_db/runtime/common/rt_any.cc b/flex/engines/graph_db/runtime/common/rt_any.cc
index e3cf4a1d55d8..c2f60da57346 100644
--- a/flex/engines/graph_db/runtime/common/rt_any.cc
+++ b/flex/engines/graph_db/runtime/common/rt_any.cc
@@ -14,6 +14,7 @@
  */
 
 #include "flex/engines/graph_db/runtime/common/rt_any.h"
+#include "flex/proto_generated_gie/basic_type.pb.h"
 
 namespace gs {
 
@@ -53,25 +54,37 @@ RTAnyType parse_from_ir_data_type(const ::common::IrDataType& dt) {
   switch (dt.type_case()) {
   case ::common::IrDataType::TypeCase::kDataType: {
     const ::common::DataType ddt = dt.data_type();
-    switch (ddt) {
-    case ::common::DataType::BOOLEAN:
-      return RTAnyType::kBoolValue;
-    case ::common::DataType::INT64:
-      return RTAnyType::kI64Value;
-    case ::common::DataType::STRING:
+    switch (ddt.item_case()) {
+    case ::common::DataType::kPrimitiveType: {
+      const ::common::PrimitiveType pt = ddt.primitive_type();
+      switch (pt) {
+      case ::common::PrimitiveType::DT_SIGNED_INT32:
+        return RTAnyType::kI32Value;
+      case ::common::PrimitiveType::DT_SIGNED_INT64:
+        return RTAnyType::kI64Value;
+      case ::common::PrimitiveType::DT_DOUBLE:
+        return RTAnyType::kF64Value;
+      case ::common::PrimitiveType::DT_BOOL:
+        return RTAnyType::kBoolValue;
+      default:
+        LOG(FATAL) << "unrecoginized primitive type - " << pt;
+        break;
+      }
+    }
+    case ::common::DataType::kString:
       return RTAnyType::kStringValue;
-    case ::common::DataType::INT32:
-      return RTAnyType::kI32Value;
-    case ::common::DataType::DATE32:
-      return RTAnyType::kDate32;
-    case ::common::DataType::STRING_ARRAY:
+    case ::common::DataType::kTemporal: {
+      if (ddt.temporal().item_case() == ::common::Temporal::kDate32) {
+        return RTAnyType::kDate32;
+      } else {
+        LOG(FATAL) << "unrecoginized temporal type - "
+                   << ddt.temporal().DebugString();
+      }
+    }
+    case ::common::DataType::kArray:
       return RTAnyType::kList;
-    case ::common::DataType::TIMESTAMP:
-      return RTAnyType::kDate32;
-    case ::common::DataType::DOUBLE:
-      return RTAnyType::kF64Value;
     default:
-      LOG(FATAL) << "unrecoginized data type - " << ddt;
+      LOG(FATAL) << "unrecoginized data type - " << ddt.DebugString();
       break;
     }
   } break;
diff --git a/flex/utils/CMakeLists.txt b/flex/utils/CMakeLists.txt
index c25376746fbc..a85c74491710 100644
--- a/flex/utils/CMakeLists.txt
+++ b/flex/utils/CMakeLists.txt
@@ -32,6 +32,7 @@ set(CODE_GEN_PROTOBUF_FILES
         ${GIE_COMPILER_PROTO_DIR}/schema.proto
         ${GIE_COMPILER_PROTO_DIR}/type.proto
         ${GIE_COMPILER_PROTO_DIR}/stored_procedure.proto
+        ${GIE_COMPILER_PROTO_DIR}/basic_type.proto
         ${GIE_COMPILER_PROTO_JOB_DIR}/job_service.proto
 )
 
diff --git a/interactive_engine/executor/ir/common/build.rs b/interactive_engine/executor/ir/common/build.rs
index 96cb3a96dea9..a0b395e72db8 100644
--- a/interactive_engine/executor/ir/common/build.rs
+++ b/interactive_engine/executor/ir/common/build.rs
@@ -32,6 +32,7 @@ fn codegen_inplace() -> Result<(), Box<dyn std::error::Error>> {
     println!("cargo:rerun-if-changed=../proto/physical.proto");
     println!("cargo:rerun-if-changed=../proto/type.proto");
     println!("cargo:rerun-if-changed=../proto/stored_procedure.proto");
+    println!("cargo:rerun-if-changed=../proto/basic_type.proto");
     let out_dir = PathBuf::from(GEN_DIR);
     if out_dir.exists() {
         let _ = std::fs::remove_dir_all(GEN_DIR);
@@ -50,6 +51,7 @@ fn codegen_inplace() -> Result<(), Box<dyn std::error::Error>> {
                 "../proto/physical.proto",
                 "../proto/type.proto",
                 "../proto/stored_procedure.proto",
+                "../proto/basic_type.proto",
             ],
             &["../proto"],
         )?;
@@ -67,6 +69,7 @@ fn codegen_inplace() -> Result<(), Box<dyn std::error::Error>> {
     println!("cargo:rerun-if-changed=../proto/physical.proto");
     println!("cargo:rerun-if-changed=../proto/type.proto");
     println!("cargo:rerun-if-changed=../proto/stored_procedure.proto");
+    println!("cargo:rerun-if-changed=../proto/basic_type.proto");
     prost_build::Config::new()
         .type_attribute(".", "#[derive(Serialize,Deserialize)]")
         .compile_protos(
@@ -79,6 +82,7 @@ fn codegen_inplace() -> Result<(), Box<dyn std::error::Error>> {
                 "../proto/physical.proto",
                 "../proto/type.proto",
                 "../proto/stored_procedure.proto",
+                "../proto/basic_type.proto",
             ],
             &["../proto"],
         )?;
diff --git a/interactive_engine/executor/ir/proto/basic_type.proto b/interactive_engine/executor/ir/proto/basic_type.proto
new file mode 100644
index 000000000000..8e1d72d45c24
--- /dev/null
+++ b/interactive_engine/executor/ir/proto/basic_type.proto
@@ -0,0 +1,120 @@
+syntax = "proto3";
+
+package common;
+option java_package = "com.alibaba.graphscope.proto.type";
+option java_outer_classname = "Common";
+
+enum PrimitiveType {
+    DT_ANY = 0;
+    DT_SIGNED_INT32 = 1;
+    DT_UNSIGNED_INT32 = 2;
+    DT_SIGNED_INT64 = 3;
+    DT_UNSIGNED_INT64 = 4;
+    DT_BOOL = 5;
+    DT_FLOAT = 6;
+    DT_DOUBLE = 7;
+}
+
+message Decimal { // precision=4 scale=2 : 23.12
+    uint32 precision = 1;
+    uint32 scale = 2;
+}
+
+message String {
+    // string with unlimited length
+    message LongText {
+    }
+
+    // string with fixed length
+    message Char {
+        uint32 fixed_length = 1;
+    }
+
+    // string with variable length, bounded by max_length
+    message VarChar {
+        uint32 max_length = 1;
+    }
+
+    oneof item {
+        LongText long_text = 1;
+        Char char = 2;
+        VarChar var_char = 3;
+    }
+}
+
+// temporal types
+
+message Temporal {
+    enum DateFormat {
+        DF_YYYY_MM_DD = 0; // ISO fomat: 2019-01-01
+    }
+
+    enum TimeFormat {
+        TF_HH_MM_SS_SSS = 0; // ISO format: 00:00:00.000
+    }
+
+    enum DateTimeFormat {
+        DTF_YYYY_MM_DD_HH_MM_SS_SSS = 0; // ISO format: 2019-01-01 00:00:00.000
+    }
+
+    enum TimeZoneFormat {
+        TZF_UTC = 0; // Z
+        TZF_OFFSET = 1; // +08:00 or -08:00
+    }
+
+    message Date {
+        DateFormat date_format = 1;
+    }
+
+    message Time {
+        TimeFormat time_format = 1;
+        TimeZoneFormat time_zone_format = 2;
+    }
+
+    message DateTime {
+        DateTimeFormat date_time_format = 1;
+        TimeZoneFormat time_zone_format = 2;
+    }
+
+    // int32 days since 1970-01-01
+    message Date32 {
+    }
+
+    // int32 milliseconds past midnight
+    message Time32 {
+    }
+
+    // int64 milliseconds since 1970-01-01 00:00:00.000000
+    message Timestamp {
+    }
+
+    oneof item {
+        Date date = 1;
+        Time time = 2;
+        DateTime date_time = 3;
+        Date32 date32 = 4;
+        Time32 time32 = 5;
+        Timestamp timestamp = 6;
+    }
+}
+
+message Array {
+    DataType component_type = 1;
+    uint32 max_length = 2;
+}
+
+message Map {
+    DataType key_type = 1;
+    DataType value_type = 2;
+}
+
+message DataType {
+    oneof item {
+        PrimitiveType primitive_type = 1;
+        Decimal decimal = 2;
+        String string = 3;
+        Temporal temporal = 4;
+        Array array = 5;
+        Map map = 6;
+    }
+}
diff --git a/interactive_engine/executor/ir/proto/common.proto b/interactive_engine/executor/ir/proto/common.proto
index 034c6581f74a..14a07f03da4f 100644
--- a/interactive_engine/executor/ir/proto/common.proto
+++ b/interactive_engine/executor/ir/proto/common.proto
@@ -89,21 +89,3 @@ message Value {
     Timestamp timestamp      = 16;
   }
 }
-
-enum DataType {
-  BOOLEAN = 0;
-  INT32 = 1;
-  INT64 = 2 ;
-  DOUBLE = 3;
-  STRING = 4;
-  BYTES = 5;
-  INT32_ARRAY = 6;
-  INT64_ARRAY = 7;
-  DOUBLE_ARRAY = 8;
-  STRING_ARRAY = 9;
-  PAIR_ARRAY = 10;
-  NONE = 11;
-  DATE32 = 12;
-  TIME32 = 13;
-  TIMESTAMP = 14;
-}
diff --git a/interactive_engine/executor/ir/proto/schema.proto b/interactive_engine/executor/ir/proto/schema.proto
index 18bdafe759ee..22d28ea664c8 100644
--- a/interactive_engine/executor/ir/proto/schema.proto
+++ b/interactive_engine/executor/ir/proto/schema.proto
@@ -21,6 +21,7 @@ option java_package = "com.alibaba.graphscope.gaia.proto";
 option java_outer_classname = "OuterSchema";
 
 import "common.proto";
+import "basic_type.proto";
 
 message LabelMeta {
   int32 id = 1;
@@ -29,9 +30,8 @@ message LabelMeta {
 
 message ColumnMeta {
   LabelMeta key = 1;
-  common.DataType data_type = 2;
   // Whether this column is a part of a primary key
-  bool is_primary_key = 3;
+  bool is_primary_key = 2;
 }
 
 message EntityMeta {
diff --git a/interactive_engine/executor/ir/proto/type.proto b/interactive_engine/executor/ir/proto/type.proto
index 3494e7eea13d..6f258ee9a8fa 100644
--- a/interactive_engine/executor/ir/proto/type.proto
+++ b/interactive_engine/executor/ir/proto/type.proto
@@ -20,9 +20,9 @@ option java_package = "com.alibaba.graphscope.gaia.proto";
 option java_outer_classname = "DataType";
 
 import "common.proto";
+import "basic_type.proto";
 import "google/protobuf/wrappers.proto";
 
-
 // GraphDataType, including structured data types of vertices or edges.
 message GraphDataType {
   // The Option of the graph element, i.e., vertex or edge