fix(interactive): Align Data Type in GIE Physical Pb with Flex

.github/workflows/interactive.yml

@@ -212,12 +212,9 @@ jobs:
         GLOG_v=10 ./bin/bulk_loader -g ${SCHEMA_FILE} -l ${BULK_LOAD_FILE} -d ${TMP_INTERACTIVE_WORKSPACE}/data/modern_graph/indices/
         cd ${GITHUB_WORKSPACE}/flex/tests/hqps
         sed -i 's/interactive_workspace/temp_workspace/g' ./interactive_config_test.yaml
-        # set thread_num_per_worker to 4
-        sed -i 's/thread_num_per_worker: 1/thread_num_per_worker: 4/g' ./interactive_config_test.yaml
         bash hqps_sdk_test.sh ${TMP_INTERACTIVE_WORKSPACE} ./interactive_config_test.yaml java
         bash hqps_sdk_test.sh ${TMP_INTERACTIVE_WORKSPACE} ./interactive_config_test.yaml python
         sed -i 's/temp_workspace/interactive_workspace/g' ./interactive_config_test.yaml
-        sed -i 's/thread_num_per_worker: 4/thread_num_per_worker: 1/g' ./interactive_config_test.yaml
 
     - name: Robustness test 
       env:

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(

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,109 @@ 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:
+  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 temporal_type_to_data_type(
+    const common::Temporal& type) {
+  switch (type.item_case()) {
+  case common::Temporal::ItemCase::kDate:
     return codegen::DataType::kDate;
-  case common::DataType::TIME32:
+  case common::Temporal::ItemCase::kTime:
     return codegen::DataType::kTime;
-  case common::DataType::TIMESTAMP:
+  case common::Temporal::kTimestamp:
     return codegen::DataType::kTimeStamp;
+  default:
+    throw std::runtime_error(
+        "unknown temporal type when converting temporal type to data type:" +
+        std::to_string(static_cast<int>(type.item_case())));
+  }
+}
+
+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:
+    return temporal_type_to_data_type(data_type.temporal());
+  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 +314,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(

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:

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;

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_,

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;

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;

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::

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);