from_protobuf kernel (part1): scalar field decoding

src/main/cpp/src/protobuf/protobuf_builders.cu

@@ -123,4 +123,133 @@ std::unique_ptr<cudf::column> make_empty_list_column(std::unique_ptr<cudf::colum
     0, std::move(offsets_col), std::move(element_col), 0, rmm::device_buffer{});
 }
 
+// ============================================================================
+// Enum-as-string column builders
+// ============================================================================
+
+struct enum_string_lookup_tables {
+  rmm::device_uvector<int32_t> d_valid_enums;
+  rmm::device_uvector<int32_t> d_name_offsets;
+  rmm::device_uvector<uint8_t> d_name_chars;
+};
+
+enum_string_lookup_tables make_enum_string_lookup_tables(
+  cudf::detail::host_vector<int32_t> const& valid_enums,
+  std::vector<cudf::detail::host_vector<uint8_t>> const& enum_name_bytes,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr)
+{
+  auto d_valid_enums = cudf::detail::make_device_uvector_async(
+    valid_enums, stream, cudf::get_current_device_resource_ref());
+
+  auto h_name_offsets =
+    cudf::detail::make_pinned_vector_async<int32_t>(valid_enums.size() + 1, stream);
+  std::fill(h_name_offsets.begin(), h_name_offsets.end(), 0);
+  int64_t total_name_chars = 0;
+  for (size_t k = 0; k < enum_name_bytes.size(); ++k) {
+    total_name_chars += static_cast<int64_t>(enum_name_bytes[k].size());
+    CUDF_EXPECTS(total_name_chars <= std::numeric_limits<int32_t>::max(),
+                 "Enum name data exceeds 2 GB limit");
+    h_name_offsets[k + 1] = static_cast<int32_t>(total_name_chars);
+  }
+
+  auto h_name_chars = cudf::detail::make_pinned_vector_async<uint8_t>(total_name_chars, stream);
+  int32_t cursor    = 0;
+  for (auto const& name : enum_name_bytes) {
+    if (!name.empty()) {
+      std::copy(name.data(), name.data() + name.size(), h_name_chars.data() + cursor);
+      cursor += static_cast<int32_t>(name.size());
+    }
+  }
+
+  auto d_name_offsets = cudf::detail::make_device_uvector_async(
+    h_name_offsets, stream, cudf::get_current_device_resource_ref());
+
+  auto d_name_chars = [&]() {
+    if (total_name_chars > 0) {
+      return cudf::detail::make_device_uvector_async(
+        h_name_chars, stream, cudf::get_current_device_resource_ref());
+    }
+    return rmm::device_uvector<uint8_t>(0, stream, cudf::get_current_device_resource_ref());
+  }();
+
+  return {std::move(d_valid_enums), std::move(d_name_offsets), std::move(d_name_chars)};
+}
+
+std::unique_ptr<cudf::column> build_enum_string_values_column(
+  rmm::device_uvector<int32_t>& enum_values,
+  rmm::device_uvector<bool>& valid,
+  enum_string_lookup_tables const& lookup,
+  int num_rows,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr)
+{
+  rmm::device_uvector<int32_t> lengths(num_rows, stream, cudf::get_current_device_resource_ref());
+  launch_compute_enum_string_lengths(enum_values.data(),
+                                     valid.data(),
+                                     lookup.d_valid_enums.data(),
+                                     lookup.d_name_offsets.data(),
+                                     static_cast<int>(lookup.d_valid_enums.size()),
+                                     lengths.data(),
+                                     num_rows,
+                                     stream);
+
+  auto [offsets_col, total_chars] =
+    cudf::strings::detail::make_offsets_child_column(lengths.begin(), lengths.end(), stream, mr);
+
+  rmm::device_uvector<char> chars(total_chars, stream, mr);
+  if (total_chars > 0) {
+    launch_copy_enum_string_chars(enum_values.data(),
+                                  valid.data(),
+                                  lookup.d_valid_enums.data(),
+                                  lookup.d_name_offsets.data(),
+                                  lookup.d_name_chars.data(),
+                                  static_cast<int>(lookup.d_valid_enums.size()),
+                                  offsets_col->view().data<int32_t>(),
+                                  chars.data(),
+                                  num_rows,
+                                  stream);
+  }
+
+  auto [mask, null_count] = make_null_mask_from_valid(valid, stream, mr);
+  return cudf::make_strings_column(
+    num_rows, std::move(offsets_col), chars.release(), null_count, std::move(mask));
+}
+
+std::unique_ptr<cudf::column> build_enum_string_column(
+  rmm::device_uvector<int32_t>& enum_values,
+  rmm::device_uvector<bool>& valid,
+  cudf::detail::host_vector<int32_t> const& valid_enums,
+  std::vector<cudf::detail::host_vector<uint8_t>> const& enum_name_bytes,
+  rmm::device_uvector<bool>& d_row_force_null,
+  int num_rows,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr,
+  int32_t const* top_row_indices,
+  bool propagate_invalid_rows)
+{
+  auto lookup = make_enum_string_lookup_tables(valid_enums, enum_name_bytes, stream, mr);
+  rmm::device_uvector<bool> d_item_has_invalid_enum(
+    num_rows, stream, cudf::get_current_device_resource_ref());
+  thrust::fill(rmm::exec_policy_nosync(stream),
+               d_item_has_invalid_enum.begin(),
+               d_item_has_invalid_enum.end(),
+               false);
+
+  launch_validate_enum_values(enum_values.data(),
+                              valid.data(),
+                              d_item_has_invalid_enum.data(),
+                              lookup.d_valid_enums.data(),
+                              static_cast<int>(valid_enums.size()),
+                              num_rows,
+                              stream);
+  propagate_invalid_enum_flags_to_rows(d_item_has_invalid_enum,
+                                       d_row_force_null,
+                                       num_rows,
+                                       top_row_indices,
+                                       propagate_invalid_rows,
+                                       stream);
+  return build_enum_string_values_column(enum_values, valid, lookup, num_rows, stream, mr);
+}
+
 }  // namespace spark_rapids_jni::protobuf::detail

src/main/cpp/src/protobuf/protobuf_host_helpers.hpp

@@ -98,6 +98,20 @@ std::unique_ptr<cudf::column> make_empty_list_column(std::unique_ptr<cudf::colum
                                                      rmm::cuda_stream_view stream,
                                                      rmm::device_async_resource_ref mr);
 
+/**
+ * Extract output type from either nested_field_descriptor (.output_type is cudf::type_id)
+ * or device_nested_field_descriptor (.output_type_id is int).
+ */
+template <typename FieldT>
+inline cudf::type_id get_output_type_id(FieldT const& field)
+{
+  if constexpr (std::is_same_v<FieldT, device_nested_field_descriptor>) {
+    return static_cast<cudf::type_id>(field.output_type_id);
+  } else {
+    return field.output_type;
+  }
+}
+
 template <typename SchemaT>
 std::unique_ptr<cudf::column> make_empty_struct_column_with_schema(
   SchemaT const& schema,
@@ -110,7 +124,7 @@ std::unique_ptr<cudf::column> make_empty_struct_column_with_schema(
 
   std::vector<std::unique_ptr<cudf::column>> children;
   for (int child_idx : child_indices) {
-    auto child_type = cudf::data_type{schema[child_idx].output_type};
+    auto child_type = cudf::data_type{get_output_type_id(schema[child_idx])};
 
     std::unique_ptr<cudf::column> child_col;
     if (child_type.id() == cudf::type_id::STRUCT) {