Add logical type inspection for raw Parquet VARIANT values

cpp/include/cudf/io/experimental/variant.hpp

@@ -7,6 +7,7 @@
 
 #include <cudf/column/column.hpp>
 #include <cudf/column/column_view.hpp>
+#include <cudf/io/experimental/variant_spec.hpp>
 #include <cudf/types.hpp>
 #include <cudf/utilities/default_stream.hpp>
 #include <cudf/utilities/memory_resource.hpp>
@@ -109,6 +110,26 @@ namespace io::parquet::experimental {
   rmm::cuda_stream_view stream      = cudf::get_default_stream(),
   rmm::device_async_resource_ref mr = cudf::get_current_device_resource_ref());
 
+/**
+ * @brief Return the logical type of each VARIANT value blob in a `list<uint8>` column.
+ *
+ * Classifies only the value_metadata header byte; does not validate the remaining payload.
+ * A recognized header returns its logical type even when the payload is truncated. A null output
+ * row is produced when the input row is null, the blob is empty, or the header carries an
+ * unrecognized type. An encoded Variant null (NULLVAL) produces a valid `NULL_VALUE` row.
+ *
+ * @param values `list<uint8>` column of VARIANT-encoded value bytes
+ * @param stream CUDA stream
+ * @param mr Device memory resource
+ * @return `UINT8` column of `variant_logical_type` values cast to `uint8_t`
+ *
+ * @throws std::invalid_argument if `values` is not a `list<uint8>` column
+ */
+[[nodiscard]] std::unique_ptr<column> get_variant_type_id(
+  column_view const& values,
+  rmm::cuda_stream_view stream      = cudf::get_default_stream(),
+  rmm::device_async_resource_ref mr = cudf::get_current_device_resource_ref());
+
 /** @} */
 }  // namespace io::parquet::experimental
 }  // namespace CUDF_EXPORT cudf

cpp/include/cudf/io/experimental/variant_spec.hpp

@@ -46,4 +46,25 @@ enum class variant_primitive_type : uint8_t {
   UUID                 = 20,
 };
 
+/**
+ * @brief Logical type of a VARIANT value as returned by get_variant_type_id.
+ */
+enum class variant_logical_type : uint8_t {
+  OBJECT        = 0,
+  ARRAY         = 1,
+  NULL_VALUE    = 2,
+  BOOLEAN       = 3,
+  LONG_VALUE    = 4,
+  STRING        = 5,
+  DOUBLE_VALUE  = 6,
+  DECIMAL       = 7,
+  DATE          = 8,
+  TIMESTAMP     = 9,
+  TIMESTAMP_NTZ = 10,
+  FLOAT_VALUE   = 11,
+  BINARY        = 12,
+  UUID          = 13,
+  TIME_NTZ      = 14,
+};
+
 }  // namespace cudf::io::parquet::experimental

cpp/src/io/parquet/experimental/variant_extract.cu

@@ -789,6 +789,47 @@ struct cast_variant_fn {
   }
 };
 
+/**
+ * @brief Classifies only the first (value_metadata) byte of enc; does not validate the remaining
+ * payload. A recognized header returns its logical type even when the payload is truncated. Returns
+ * nullopt for an empty blob or an unrecognized primitive type ID.
+ */
+__device__ cuda::std::optional<variant_logical_type> logical_type_of(device_span<uint8_t const> enc)
+{
+  if (enc.empty()) { return cuda::std::nullopt; }
+  auto const value_metadata = enc[0];
+  auto const btype          = decode_basic_type(value_metadata);
+
+  if (btype == basic_type::SHORT_STRING) { return variant_logical_type::STRING; }
+  if (btype == basic_type::OBJECT) { return variant_logical_type::OBJECT; }
+  if (btype == basic_type::ARRAY) { return variant_logical_type::ARRAY; }
+
+  switch (static_cast<primitive_type>(variant_value_header(value_metadata))) {
+    case primitive_type::NULLVAL: return variant_logical_type::NULL_VALUE;
+    case primitive_type::BOOLEAN_TRUE:
+    case primitive_type::BOOLEAN_FALSE: return variant_logical_type::BOOLEAN;
+    case primitive_type::INT8:
+    case primitive_type::INT16:
+    case primitive_type::INT32:
+    case primitive_type::INT64: return variant_logical_type::LONG_VALUE;
+    case primitive_type::FLOAT64: return variant_logical_type::DOUBLE_VALUE;
+    case primitive_type::DECIMAL4:
+    case primitive_type::DECIMAL8:
+    case primitive_type::DECIMAL16: return variant_logical_type::DECIMAL;
+    case primitive_type::DATE: return variant_logical_type::DATE;
+    case primitive_type::TIMESTAMP_MICROS:
+    case primitive_type::TIMESTAMP_NANOS: return variant_logical_type::TIMESTAMP;
+    case primitive_type::TIMESTAMP_NTZ_MICROS:
+    case primitive_type::TIMESTAMP_NTZ_NANOS: return variant_logical_type::TIMESTAMP_NTZ;
+    case primitive_type::FLOAT32: return variant_logical_type::FLOAT_VALUE;
+    case primitive_type::BINARY: return variant_logical_type::BINARY;
+    case primitive_type::LONG_STRING: return variant_logical_type::STRING;
+    case primitive_type::TIME_NTZ_MICROS: return variant_logical_type::TIME_NTZ;
+    case primitive_type::UUID: return variant_logical_type::UUID;
+    default: return cuda::std::nullopt;
+  }
+}
+
 std::unique_ptr<column> build_path_column(cudf::host_span<std::string const> steps,
                                           rmm::cuda_stream_view stream,
                                           rmm::device_async_resource_ref mr)
@@ -958,6 +999,45 @@ std::unique_ptr<column> cast_variant(column_view const& values,
                                                mr});
 }
 
+std::unique_ptr<column> get_variant_type_id(column_view const& values,
+                                            rmm::cuda_stream_view stream,
+                                            rmm::device_async_resource_ref mr)
+{
+  validate_variant_child(values);
+  size_type const num_rows = values.size();
+  if (num_rows == 0) { return make_empty_column(data_type{type_id::UINT8}); }
+
+  auto val_device_view = column_device_view::create(values, stream);
+  cudf::lists_column_device_view val_lists_device_view(*val_device_view);
+
+  auto null_mask    = values.nullable()
+                        ? cudf::detail::copy_bitmask(values, stream, mr)
+                        : cudf::create_null_mask(num_rows, mask_state::ALL_VALID, stream, mr);
+  auto* d_null_mask = static_cast<bitmask_type*>(null_mask.data());
+
+  rmm::device_buffer data{static_cast<std::size_t>(num_rows) * sizeof(uint8_t), stream, mr};
+
+  thrust::transform(
+    rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()),
+    cuda::counting_iterator<size_type>(0),
+    cuda::counting_iterator<size_type>(num_rows),
+    static_cast<uint8_t*>(data.data()),
+    [values = val_lists_device_view, d_null_mask] __device__(size_type row) -> uint8_t {
+      if (!cudf::bit_is_set(d_null_mask, row)) { return 0; }
+      auto const ltype = logical_type_of(list_row_span(values, row));
+      if (ltype.has_value()) { return static_cast<uint8_t>(ltype.value()); }
+      cudf::clear_bit(d_null_mask, row);
+      return 0;
+    });
+
+  auto const null_count = num_rows - cudf::detail::count_set_bits(d_null_mask, 0, num_rows, stream);
+  return std::make_unique<column>(data_type{type_id::UINT8},
+                                  num_rows,
+                                  std::move(data),
+                                  null_count > 0 ? std::move(null_mask) : rmm::device_buffer{},
+                                  null_count);
+}
+
 }  // namespace detail
 
 std::unique_ptr<column> get_variant_field(column_view const& variant_column,
@@ -978,6 +1058,14 @@ std::unique_ptr<column> cast_variant(column_view const& values,
   return detail::cast_variant(values, desired_type, stream, mr);
 }
 
+std::unique_ptr<column> get_variant_type_id(column_view const& values,
+                                            rmm::cuda_stream_view stream,
+                                            rmm::device_async_resource_ref mr)
+{
+  CUDF_FUNC_RANGE();
+  return detail::get_variant_type_id(values, stream, mr);
+}
+
 std::unique_ptr<column> extract_variant_field(column_view const& variant_column,
                                               std::string_view path,
                                               data_type desired_type,