diff --git a/cpp/CMakeLists.txt b/cpp/CMakeLists.txt
index 605e78eb0c14..e43eb596fd17 100644
--- a/cpp/CMakeLists.txt
+++ b/cpp/CMakeLists.txt
@@ -796,6 +796,7 @@ add_library(
   src/io/parquet/experimental/hybrid_scan_preprocess.cu
   src/io/parquet/experimental/page_index_filter.cu
   src/io/parquet/experimental/page_index_filter_utils.cu
+  src/io/parquet/experimental/variant_encode.cu
   src/io/parquet/experimental/variant_extract.cu
   src/io/parquet/experimental/variant_path.cpp
   src/io/parquet/expression_transform_helpers.cpp
diff --git a/cpp/include/cudf/io/experimental/variant.hpp b/cpp/include/cudf/io/experimental/variant.hpp
index 7c53a89e4e7f..4bac662773f1 100644
--- a/cpp/include/cudf/io/experimental/variant.hpp
+++ b/cpp/include/cudf/io/experimental/variant.hpp
@@ -7,13 +7,16 @@
 
 #include <cudf/column/column.hpp>
 #include <cudf/column/column_view.hpp>
+#include <cudf/table/table_view.hpp>
 #include <cudf/types.hpp>
 #include <cudf/utilities/default_stream.hpp>
 #include <cudf/utilities/memory_resource.hpp>
+#include <cudf/utilities/span.hpp>
 
 #include <rmm/cuda_stream_view.hpp>
 
 #include <memory>
+#include <string>
 #include <string_view>
 
 /**
@@ -109,6 +112,34 @@ 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 Encode a table as a VARIANT object column, one row per VARIANT.
+ *
+ * Each table row is encoded as a VARIANT object with one field per column. The shared metadata
+ * blob stores column names as a sorted UTF-8 key dictionary (version 1, offset_size chosen to
+ * fit the total key-string length). Null column values are encoded as VARIANT null primitives.
+ *
+ * Supported column types: `INT8`, `INT16`, `INT32`, `INT64`, `FLOAT32`, `FLOAT64`, `STRING`, and
+ * columns whose type is `EMPTY` (treated as all-null). Other types throw. Tables must have fewer
+ * than 256 columns.
+ *
+ * @param input Table to encode (typically produced by `cudf::io::read_json`)
+ * @param column_names Column name for each column in `input`; must satisfy
+ *                     `column_names.size() == input.num_columns()`
+ * @param stream CUDA stream
+ * @param mr Device memory resource
+ * @return VARIANT column: `STRUCT<list<uint8>, list<uint8>>` (metadata child, value child)
+ *
+ * @throws std::invalid_argument if any column has an unsupported type, if the table has ≥ 256
+ *         columns, or if `column_names.size() != input.num_columns()`
+ * @throws std::overflow_error if the total encoded value bytes exceed 2 GiB
+ */
+[[nodiscard]] std::unique_ptr<column> encode_variant(
+  cudf::table_view const& input,
+  cudf::host_span<std::string const> column_names,
+  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
diff --git a/cpp/src/io/parquet/experimental/variant_encode.cu b/cpp/src/io/parquet/experimental/variant_encode.cu
new file mode 100644
index 000000000000..326d850177ed
--- /dev/null
+++ b/cpp/src/io/parquet/experimental/variant_encode.cu
@@ -0,0 +1,438 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include <cudf/column/column_device_view.cuh>
+#include <cudf/column/column_factories.hpp>
+#include <cudf/detail/null_mask.hpp>
+#include <cudf/detail/nvtx/ranges.hpp>
+#include <cudf/detail/utilities/grid_1d.cuh>
+#include <cudf/detail/utilities/vector_factories.hpp>
+#include <cudf/io/experimental/variant.hpp>
+#include <cudf/null_mask.hpp>
+#include <cudf/strings/detail/strings_children.cuh>
+#include <cudf/strings/string_view.cuh>
+#include <cudf/table/table_device_view.cuh>
+#include <cudf/table/table_view.hpp>
+#include <cudf/utilities/error.hpp>
+#include <cudf/utilities/span.hpp>
+
+#include <rmm/cuda_stream_view.hpp>
+#include <rmm/device_buffer.hpp>
+#include <rmm/device_uvector.hpp>
+#include <rmm/exec_policy.hpp>
+
+#include <cuda/std/cstring>
+#include <thrust/sequence.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <numeric>
+#include <string>
+#include <vector>
+
+namespace cudf {
+namespace io::parquet::experimental {
+namespace {
+
+// ──────────────────────────────────────────────────────────────────────────────
+// Encoding constants (from the VARIANT spec; mirrors variant_extract.cu)
+// ──────────────────────────────────────────────────────────────────────────────
+
+constexpr uint8_t k_null_value         = 0x00;
+constexpr uint8_t k_bool_true_header   = 0x04;  // primitive type 1 (BOOLEAN_TRUE)
+constexpr uint8_t k_bool_false_header  = 0x08;  // primitive type 2 (BOOLEAN_FALSE)
+constexpr uint8_t k_int8_header        = 0x0c;
+constexpr uint8_t k_int16_header       = 0x10;
+constexpr uint8_t k_int32_header       = 0x14;
+constexpr uint8_t k_int64_header       = 0x18;
+constexpr uint8_t k_float64_header     = 0x1c;  // primitive type 7
+constexpr uint8_t k_float32_header     = 0x38;  // primitive type 14
+constexpr uint8_t k_long_string_header = 0x40;  // primitive type 16
+
+// short_string: basic_type=1, length in bits 2..7 of the value_metadata byte
+constexpr uint8_t k_short_string_basic_type = 0x01;
+
+// Object with field_id_size=1, field_offset_size=4, num_elements_size=1:
+//   value_header = (is_large=0 << 4) | (field_id_size-1=0 << 2) | (field_offset_size-1=3) = 0x03
+//   value_metadata = (value_header << 2) | basic_type::object(2) = 0x0e
+constexpr uint8_t k_object_value_metadata = 0x0e;
+
+constexpr int block_size = 256;
+
+// ──────────────────────────────────────────────────────────────────────────────
+// Per-field device helpers
+// ──────────────────────────────────────────────────────────────────────────────
+
+__device__ int64_t field_encoded_size(column_device_view const& col, size_type row)
+{
+  if (col.type().id() == type_id::EMPTY || !col.is_valid(row)) { return 1; }
+  switch (col.type().id()) {
+    case type_id::BOOL8: return 1;
+    case type_id::INT8: return 2;
+    case type_id::INT16: return 3;
+    case type_id::INT32: return 5;
+    case type_id::INT64: return 9;
+    case type_id::FLOAT32: return 5;
+    case type_id::FLOAT64: return 9;
+    case type_id::STRING: {
+      auto const len = static_cast<int64_t>(col.element<cudf::string_view>(row).size_bytes());
+      return len < 64 ? 1 + len : 5 + len;
+    }
+    default: return 1;
+  }
+}
+
+// Write the encoded bytes for one field into `out`, returns bytes written.
+__device__ size_type write_field_value(uint8_t* out, column_device_view const& col, size_type row)
+{
+  if (col.type().id() == type_id::EMPTY || !col.is_valid(row)) {
+    out[0] = k_null_value;
+    return 1;
+  }
+  switch (col.type().id()) {
+    case type_id::BOOL8: {
+      out[0] = col.element<bool>(row) ? k_bool_true_header : k_bool_false_header;
+      return 1;
+    }
+    case type_id::INT8: {
+      auto const v = col.element<int8_t>(row);
+      out[0]       = k_int8_header;
+      cuda::std::memcpy(out + 1, &v, 1);
+      return 2;
+    }
+    case type_id::INT16: {
+      auto const v = col.element<int16_t>(row);
+      out[0]       = k_int16_header;
+      cuda::std::memcpy(out + 1, &v, 2);
+      return 3;
+    }
+    case type_id::INT32: {
+      auto const v = col.element<int32_t>(row);
+      out[0]       = k_int32_header;
+      cuda::std::memcpy(out + 1, &v, 4);
+      return 5;
+    }
+    case type_id::INT64: {
+      auto const v = col.element<int64_t>(row);
+      out[0]       = k_int64_header;
+      cuda::std::memcpy(out + 1, &v, 8);
+      return 9;
+    }
+    case type_id::FLOAT32: {
+      auto const v = col.element<float>(row);
+      out[0]       = k_float32_header;
+      cuda::std::memcpy(out + 1, &v, 4);
+      return 5;
+    }
+    case type_id::FLOAT64: {
+      auto const v = col.element<double>(row);
+      out[0]       = k_float64_header;
+      cuda::std::memcpy(out + 1, &v, 8);
+      return 9;
+    }
+    case type_id::STRING: {
+      auto const sv  = col.element<cudf::string_view>(row);
+      auto const len = static_cast<size_type>(sv.size_bytes());
+      if (len < 64) {
+        out[0] = static_cast<uint8_t>(k_short_string_basic_type | (len << 2));
+        cuda::std::memcpy(out + 1, sv.data(), len);
+        return 1 + len;
+      } else {
+        out[0]             = k_long_string_header;
+        uint32_t const u32 = static_cast<uint32_t>(len);
+        cuda::std::memcpy(out + 1, &u32, 4);
+        cuda::std::memcpy(out + 5, sv.data(), len);
+        return 5 + len;
+      }
+    }
+    default: out[0] = k_null_value; return 1;
+  }
+}
+
+// ──────────────────────────────────────────────────────────────────────────────
+// encode_variant kernels
+// ──────────────────────────────────────────────────────────────────────────────
+
+// Pass 1: compute per-row value blob sizes.
+CUDF_KERNEL __launch_bounds__(block_size) void object_value_sizes_kernel(
+  cudf::table_device_view tbl,
+  device_span<int32_t const> sort_order,
+  device_span<size_type> d_val_sizes)
+{
+  auto const num_rows = static_cast<size_type>(d_val_sizes.size());
+  auto const N        = tbl.num_columns();
+  auto const tid      = cudf::detail::grid_1d::global_thread_id<block_size>();
+  auto const stride   = cudf::detail::grid_1d::grid_stride<block_size>();
+
+  for (auto row = tid; row < num_rows; row += stride) {
+    // Header: value_metadata(1) + num_elements(1) + field_ids(N) + field_offsets((N+1)*4)
+    int64_t size = static_cast<int64_t>(2 + N + (N + 1) * 4);
+    for (int i = 0; i < N; i++) {
+      size += field_encoded_size(tbl.column(sort_order[i]), row);
+    }
+    // Saturate to INT32_MAX so the host-side total-bytes overflow check fires.
+    d_val_sizes[row] = size <= static_cast<int64_t>(cuda::std::numeric_limits<size_type>::max())
+                         ? static_cast<size_type>(size)
+                         : cuda::std::numeric_limits<size_type>::max();
+  }
+}
+
+// Pass 2: write metadata and value blobs.
+// Metadata is constant, stored at stride `meta_size` bytes per row.
+CUDF_KERNEL __launch_bounds__(block_size) void object_encode_write_kernel(
+  cudf::table_device_view tbl,
+  device_span<int32_t const> sort_order,
+  device_span<size_type const> d_val_offsets,
+  uint8_t* d_val_buf,
+  uint8_t* d_meta_buf,
+  uint8_t const* d_meta_template,
+  size_type meta_size)
+{
+  auto const num_rows = static_cast<size_type>(d_val_offsets.size()) - 1;
+  auto const N        = tbl.num_columns();
+  auto const tid      = cudf::detail::grid_1d::global_thread_id<block_size>();
+  auto const stride   = cudf::detail::grid_1d::grid_stride<block_size>();
+
+  for (auto row = tid; row < num_rows; row += stride) {
+    // Broadcast constant metadata template
+    cuda::std::memcpy(d_meta_buf + row * meta_size, d_meta_template, meta_size);
+
+    // Write the VARIANT object value
+    uint8_t* p = d_val_buf + d_val_offsets[row];
+
+    *p++ = k_object_value_metadata;
+    *p++ = static_cast<uint8_t>(N);
+
+    // field_ids: sorted column IDs are 0..N-1 (dictionary is in sorted order)
+    for (int i = 0; i < N; i++) {
+      *p++ = static_cast<uint8_t>(i);
+    }
+
+    // Write (N+1) field offsets, 4 bytes each (LE).
+    // field_encoded_size is called once per field here (and again below in write_field_value),
+    // which avoids the 1 KiB field_sizes[256] stack array that would spill to local memory.
+    uint32_t running = 0;
+    for (int i = 0; i <= N; i++) {
+      cuda::std::memcpy(p, &running, 4);
+      p += 4;
+      if (i < N) {
+        running += static_cast<uint32_t>(field_encoded_size(tbl.column(sort_order[i]), row));
+      }
+    }
+
+    // Write field values in sorted order
+    for (int i = 0; i < N; i++) {
+      p += write_field_value(p, tbl.column(sort_order[i]), row);
+    }
+  }
+}
+
+// ──────────────────────────────────────────────────────────────────────────────
+// Host: build metadata blob and column sort order
+// ──────────────────────────────────────────────────────────────────────────────
+
+// Returns (metadata_bytes, sort_order) where sort_order[i] is the original column index of
+// the i-th lexicographically sorted column. Column names appear in sorted order in the blob.
+std::pair<std::vector<uint8_t>, std::vector<int32_t>> build_metadata_blob(
+  cudf::host_span<std::string const> column_names)
+{
+  auto const N = static_cast<int>(column_names.size());
+
+  std::vector<int32_t> sort_order(N);
+  std::iota(sort_order.begin(), sort_order.end(), 0);
+  std::stable_sort(sort_order.begin(), sort_order.end(), [&](int a, int b) {
+    return column_names[a] < column_names[b];
+  });
+
+  // Choose offset_size based on total UTF-8 key length
+  std::size_t total_key_bytes = 0;
+  for (auto const& name : column_names) {
+    total_key_bytes += name.size();
+  }
+  int const offset_size = total_key_bytes <= 0xFFu ? 1 : total_key_bytes <= 0xFFFFu ? 2 : 4;
+
+  // header: bits[7:6] = offset_size-1, bits[3:0] = version=1
+  uint8_t const header = static_cast<uint8_t>(((offset_size - 1) << 6) | 0x01u);
+
+  std::vector<uint8_t> blob;
+  blob.push_back(header);
+
+  // num_keys (offset_size bytes LE)
+  for (int b = 0; b < offset_size; b++) {
+    blob.push_back(static_cast<uint8_t>((static_cast<uint32_t>(N) >> (8 * b)) & 0xFFu));
+  }
+
+  // offsets[0..N] (offset_size bytes each LE), relative to start of string_data
+  std::size_t running = 0;
+  auto push_offset    = [&](std::size_t v) {
+    for (int b = 0; b < offset_size; b++) {
+      blob.push_back(static_cast<uint8_t>((v >> (8 * b)) & 0xFFu));
+    }
+  };
+  push_offset(0);
+  for (int i = 0; i < N; i++) {
+    running += column_names[sort_order[i]].size();
+    push_offset(running);
+  }
+
+  // string_data: keys in sorted order
+  for (int i = 0; i < N; i++) {
+    auto const& name = column_names[sort_order[i]];
+    blob.insert(blob.end(), name.begin(), name.end());
+  }
+
+  return {std::move(blob), std::move(sort_order)};
+}
+
+// Build a list<uint8> column for constant-stride blobs (all rows the same M bytes).
+// Offsets are [0, M, 2M, ..., num_rows*M]; data is filled by caller via kernel.
+std::pair<std::unique_ptr<column>, std::unique_ptr<column>> make_constant_list_buffers(
+  size_type num_rows, size_type M, rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr)
+{
+  CUDF_EXPECTS(static_cast<int64_t>(num_rows) * M <= std::numeric_limits<size_type>::max(),
+               "metadata size exceeds 2 GiB limit",
+               std::overflow_error);
+  auto offsets = make_numeric_column(
+    data_type{type_id::INT32}, num_rows + 1, mask_state::UNALLOCATED, stream, mr);
+  thrust::sequence(rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()),
+                   offsets->mutable_view().begin<int32_t>(),
+                   offsets->mutable_view().end<int32_t>(),
+                   int32_t{0},
+                   static_cast<int32_t>(M));
+
+  auto data = make_numeric_column(
+    data_type{type_id::UINT8}, num_rows * M, mask_state::UNALLOCATED, stream, mr);
+  return {std::move(offsets), std::move(data)};
+}
+
+}  // namespace
+
+namespace detail {
+
+// ──────────────────────────────────────────────────────────────────────────────
+// encode_variant
+// ──────────────────────────────────────────────────────────────────────────────
+
+std::unique_ptr<column> encode_variant(cudf::table_view const& input,
+                                       cudf::host_span<std::string const> column_names,
+                                       rmm::cuda_stream_view stream,
+                                       rmm::device_async_resource_ref mr)
+{
+  auto const N        = input.num_columns();
+  auto const num_rows = input.num_rows();
+
+  CUDF_EXPECTS(static_cast<std::size_t>(N) == column_names.size(),
+               "encode_variant: column_names.size() must equal input.num_columns()",
+               std::invalid_argument);
+  CUDF_EXPECTS(
+    N < 256, "encode_variant: table must have fewer than 256 columns", std::invalid_argument);
+
+  for (int i = 0; i < N; i++) {
+    auto const id = input.column(i).type().id();
+    CUDF_EXPECTS(id == type_id::EMPTY || id == type_id::BOOL8 || id == type_id::INT8 ||
+                   id == type_id::INT16 || id == type_id::INT32 || id == type_id::INT64 ||
+                   id == type_id::FLOAT32 || id == type_id::FLOAT64 || id == type_id::STRING,
+                 "encode_variant: unsupported column type — supported: EMPTY, BOOL8, "
+                 "INT8/16/32/64, FLOAT32/64, STRING",
+                 std::invalid_argument);
+  }
+
+  // ── Metadata blob, sort order, and duplicate check (sort once) ──
+  auto [meta_bytes, sort_order] = build_metadata_blob(column_names);
+  for (int i = 1; i < N; i++) {
+    CUDF_EXPECTS(column_names[sort_order[i]] != column_names[sort_order[i - 1]],
+                 "encode_variant: duplicate column names are not allowed",
+                 std::invalid_argument);
+  }
+
+  auto make_empty_list = [&] {
+    return make_lists_column(
+      0, make_empty_column(type_id::INT32), make_empty_column(type_id::UINT8), 0, {});
+  };
+
+  if (num_rows == 0) {
+    std::vector<std::unique_ptr<column>> ch;
+    ch.push_back(make_empty_list());
+    ch.push_back(make_empty_list());
+    return make_structs_column(0, std::move(ch), 0, {}, stream, mr);
+  }
+
+  auto const M = static_cast<size_type>(meta_bytes.size());
+
+  rmm::device_uvector<uint8_t> d_meta_template = cudf::detail::make_device_uvector_async(
+    meta_bytes, stream, cudf::get_current_device_resource_ref());
+
+  rmm::device_uvector<int32_t> d_sort_order = cudf::detail::make_device_uvector_async(
+    sort_order, stream, cudf::get_current_device_resource_ref());
+
+  // ── Column device views on device (via table_device_view) ──
+  auto d_table = cudf::table_device_view::create(input, stream);
+
+  // ── Metadata buffers: constant M bytes per row ──
+  auto [meta_offsets_col, meta_data_col] = make_constant_list_buffers(num_rows, M, stream, mr);
+
+  // ── Value sizes (pass 1) ──
+  rmm::device_uvector<size_type> d_val_sizes(
+    num_rows, stream, cudf::get_current_device_resource_ref());
+  {
+    cudf::detail::grid_1d grid{num_rows, block_size};
+    object_value_sizes_kernel<<<grid.num_blocks, block_size, 0, stream.value()>>>(
+      *d_table, d_sort_order, d_val_sizes);
+    CUDF_CUDA_TRY(cudaGetLastError());
+  }
+
+  auto [val_offsets_col, total_val_bytes] =
+    cudf::detail::make_offsets_child_column(d_val_sizes.begin(), d_val_sizes.end(), stream, mr);
+  CUDF_EXPECTS(val_offsets_col->type().id() == type_id::INT32,
+               "VARIANT value bytes exceed 2 GiB limit",
+               std::overflow_error);
+
+  auto val_data_col = make_numeric_column(data_type{type_id::UINT8},
+                                          static_cast<size_type>(total_val_bytes),
+                                          mask_state::UNALLOCATED,
+                                          stream,
+                                          mr);
+
+  // ── Write pass (pass 2): metadata + values ──
+  {
+    device_span<size_type const> d_val_offsets{val_offsets_col->view().data<size_type>(),
+                                               static_cast<std::size_t>(num_rows + 1)};
+    cudf::detail::grid_1d grid{num_rows, block_size};
+    object_encode_write_kernel<<<grid.num_blocks, block_size, 0, stream.value()>>>(
+      *d_table,
+      d_sort_order,
+      d_val_offsets,
+      val_data_col->mutable_view().data<uint8_t>(),
+      meta_data_col->mutable_view().data<uint8_t>(),
+      d_meta_template.data(),
+      M);
+    CUDF_CUDA_TRY(cudaGetLastError());
+  }
+
+  // ── Assemble STRUCT<list<uint8>, list<uint8>> (output rows are never null) ──
+  auto meta_col =
+    make_lists_column(num_rows, std::move(meta_offsets_col), std::move(meta_data_col), 0, {});
+  auto val_col =
+    make_lists_column(num_rows, std::move(val_offsets_col), std::move(val_data_col), 0, {});
+
+  std::vector<std::unique_ptr<column>> children;
+  children.push_back(std::move(meta_col));
+  children.push_back(std::move(val_col));
+  return make_structs_column(num_rows, std::move(children), 0, {}, stream, mr);
+}
+
+}  // namespace detail
+
+std::unique_ptr<column> encode_variant(cudf::table_view const& input,
+                                       cudf::host_span<std::string const> column_names,
+                                       rmm::cuda_stream_view stream,
+                                       rmm::device_async_resource_ref mr)
+{
+  CUDF_FUNC_RANGE();
+  return detail::encode_variant(input, column_names, stream, mr);
+}
+
+}  // namespace io::parquet::experimental
+}  // namespace cudf
diff --git a/cpp/tests/CMakeLists.txt b/cpp/tests/CMakeLists.txt
index 06c0462ebed9..9633b7eda166 100644
--- a/cpp/tests/CMakeLists.txt
+++ b/cpp/tests/CMakeLists.txt
@@ -360,6 +360,7 @@ ConfigureTest(
   io/parquet_common.cpp
   io/parquet_test.cpp
 )
+ConfigureTest(VARIANT_ENCODE_TEST io/experimental/variant_encode_test.cpp)
 ConfigureTest(VARIANT_EXTRACT_TEST io/experimental/variant_extract_test.cpp)
 ConfigureTest(
   PARQUET_DELETION_VECTORS_TEST
diff --git a/cpp/tests/io/experimental/variant_encode_test.cpp b/cpp/tests/io/experimental/variant_encode_test.cpp
new file mode 100644
index 000000000000..96d173467a69
--- /dev/null
+++ b/cpp/tests/io/experimental/variant_encode_test.cpp
@@ -0,0 +1,555 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include <cudf_test/base_fixture.hpp>
+#include <cudf_test/column_utilities.hpp>
+#include <cudf_test/column_wrapper.hpp>
+#include <cudf_test/cudf_gtest.hpp>
+#include <cudf_test/table_utilities.hpp>
+
+#include <cudf/column/column_factories.hpp>
+#include <cudf/copying.hpp>
+#include <cudf/io/experimental/variant.hpp>
+#include <cudf/structs/structs_column_view.hpp>
+#include <cudf/table/table.hpp>
+
+#include <cstdint>
+#include <numeric>
+#include <string>
+#include <vector>
+
+// ──────────────────────────────────────────────────────────────────────────────
+// Test helpers
+// ──────────────────────────────────────────────────────────────────────────────
+
+namespace {
+
+// Byte-level encoding helpers (host-side, matching the GPU implementation)
+
+inline std::vector<uint8_t> enc_null() { return {0x00}; }
+
+inline std::vector<uint8_t> enc_bool_true() { return {0x04}; }
+
+inline std::vector<uint8_t> enc_bool_false() { return {0x08}; }
+
+inline std::vector<uint8_t> enc_int8(int8_t v) { return {0x0c, static_cast<uint8_t>(v)}; }
+
+inline std::vector<uint8_t> enc_int16(int16_t v)
+{
+  auto const u = static_cast<uint16_t>(v);
+  return {0x10, static_cast<uint8_t>(u & 0xffu), static_cast<uint8_t>((u >> 8) & 0xffu)};
+}
+
+inline std::vector<uint8_t> enc_int32(int32_t v)
+{
+  auto const u = static_cast<uint32_t>(v);
+  return {0x14,
+          static_cast<uint8_t>(u & 0xffu),
+          static_cast<uint8_t>((u >> 8) & 0xffu),
+          static_cast<uint8_t>((u >> 16) & 0xffu),
+          static_cast<uint8_t>((u >> 24) & 0xffu)};
+}
+
+inline std::vector<uint8_t> enc_int64(int64_t v)
+{
+  auto const u = static_cast<uint64_t>(v);
+  std::vector<uint8_t> out{0x18};
+  for (int b = 0; b < 8; b++) {
+    out.push_back(static_cast<uint8_t>((u >> (8 * b)) & 0xffu));
+  }
+  return out;
+}
+
+inline std::vector<uint8_t> enc_short_string(std::string_view s)
+{
+  std::vector<uint8_t> out{static_cast<uint8_t>(0x01 | (s.size() << 2))};
+  out.insert(out.end(), s.begin(), s.end());
+  return out;
+}
+
+// Build VARIANT metadata blob for a sorted list of key names (offset_size=1 assumed; keys must
+// have total byte length <= 255).
+inline std::vector<uint8_t> build_metadata(std::vector<std::string> const& sorted_keys)
+{
+  std::vector<uint8_t> out{0x01, static_cast<uint8_t>(sorted_keys.size())};
+  std::vector<uint8_t> offsets{0x00};
+  uint8_t running = 0;
+  for (auto const& k : sorted_keys) {
+    running = static_cast<uint8_t>(running + k.size());
+    offsets.push_back(running);
+  }
+  out.insert(out.end(), offsets.begin(), offsets.end());
+  for (auto const& k : sorted_keys) {
+    out.insert(out.end(), k.begin(), k.end());
+  }
+  return out;
+}
+
+// Build an object VARIANT value blob with field_id_size=1, field_offset_size=4.
+// `field_values` must be in the same order as sorted_keys (IDs 0..N-1).
+inline std::vector<uint8_t> build_object_value(
+  std::vector<std::vector<uint8_t>> const& field_values)
+{
+  auto const N = static_cast<int>(field_values.size());
+  std::vector<uint8_t> out;
+
+  // value_metadata = 0x0e: object, field_id_size=1, field_offset_size=4, num_elements_size=1
+  out.push_back(0x0e);
+  out.push_back(static_cast<uint8_t>(N));
+
+  // field_ids = 0, 1, ..., N-1
+  for (int i = 0; i < N; i++) {
+    out.push_back(static_cast<uint8_t>(i));
+  }
+
+  // field_offsets[0..N] (4 bytes each LE)
+  uint32_t running = 0;
+  for (int i = 0; i <= N; i++) {
+    for (int b = 0; b < 4; b++) {
+      out.push_back(static_cast<uint8_t>((running >> (8 * b)) & 0xffu));
+    }
+    if (i < N) { running += static_cast<uint32_t>(field_values[i].size()); }
+  }
+
+  // field values
+  for (auto const& fv : field_values) {
+    out.insert(out.end(), fv.begin(), fv.end());
+  }
+  return out;
+}
+
+// Build a multi-row VARIANT struct column from explicit per-row (meta, value) byte vectors.
+inline cudf::test::structs_column_wrapper make_variant_column(
+  std::vector<std::vector<uint8_t>> const& meta_rows,
+  std::vector<std::vector<uint8_t>> const& val_rows)
+{
+  auto build = [](std::vector<std::vector<uint8_t>> const& rows) {
+    auto n = static_cast<cudf::size_type>(rows.size());
+    std::vector<int32_t> offsets(n + 1, 0);
+    std::vector<uint8_t> flat;
+    for (int i = 0; i < n; i++) {
+      flat.insert(flat.end(), rows[i].begin(), rows[i].end());
+      offsets[i + 1] = static_cast<int32_t>(flat.size());
+    }
+    auto offs_col =
+      cudf::test::fixed_width_column_wrapper<int32_t>(offsets.begin(), offsets.end()).release();
+    auto data_col =
+      cudf::test::fixed_width_column_wrapper<uint8_t>(flat.begin(), flat.end()).release();
+    return cudf::make_lists_column(n, std::move(offs_col), std::move(data_col), 0, {});
+  };
+
+  std::vector<std::unique_ptr<cudf::column>> children;
+  children.push_back(build(meta_rows));
+  children.push_back(build(val_rows));
+  return cudf::test::structs_column_wrapper{std::move(children)};
+}
+
+}  // namespace
+
+// ──────────────────────────────────────────────────────────────────────────────
+// encode_variant tests
+// ──────────────────────────────────────────────────────────────────────────────
+
+struct EncodeVariantTest : public cudf::test::BaseFixture {};
+
+TEST_F(EncodeVariantTest, EmptyTable)
+{
+  cudf::test::fixed_width_column_wrapper<int32_t> col{};
+  cudf::table_view tbl{{col}};
+  std::vector<std::string> names{"x"};
+  auto got =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  EXPECT_EQ(got->type().id(), cudf::type_id::STRUCT);
+  EXPECT_EQ(got->size(), 0);
+}
+
+TEST_F(EncodeVariantTest, UnsupportedTypeThrows)
+{
+  cudf::test::fixed_width_column_wrapper<uint64_t> col{1, 2};
+  cudf::table_view tbl{{col}};
+  std::vector<std::string> names{"u"};
+  EXPECT_THROW(static_cast<void>(cudf::io::parquet::experimental::encode_variant(
+                 tbl, names, cudf::test::get_default_stream())),
+               std::invalid_argument);
+}
+
+TEST_F(EncodeVariantTest, ColumnNamesMismatchThrows)
+{
+  cudf::test::fixed_width_column_wrapper<int32_t> col{1, 2};
+  cudf::table_view tbl{{col}};
+  std::vector<std::string> names{"a", "b"};  // 2 names, 1 column
+  EXPECT_THROW(static_cast<void>(cudf::io::parquet::experimental::encode_variant(
+                 tbl, names, cudf::test::get_default_stream())),
+               std::invalid_argument);
+}
+
+TEST_F(EncodeVariantTest, SingleInt32Column)
+{
+  // Table: col "x" = [42]
+  cudf::test::fixed_width_column_wrapper<int32_t> col{42};
+  cudf::table_view tbl{{col}};
+  std::vector<std::string> names{"x"};
+
+  auto got =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  EXPECT_EQ(got->size(), 1);
+  EXPECT_EQ(got->null_count(), 0);
+
+  // metadata: {0x01, num_keys=1, offsets=[0,1], "x"} = {0x01, 0x01, 0x00, 0x01, 'x'}
+  auto const exp_meta = build_metadata({"x"});
+  // value: object with 1 field, field_id=0, field_value=enc_int32(42)
+  auto const exp_val = build_object_value({enc_int32(42)});
+
+  auto expected = make_variant_column({exp_meta}, {exp_val});
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*got, expected);
+}
+
+TEST_F(EncodeVariantTest, SingleStringColumn)
+{
+  cudf::test::strings_column_wrapper col{"hello"};
+  cudf::table_view tbl{{col}};
+  std::vector<std::string> names{"s"};
+
+  auto got =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  auto const exp_meta = build_metadata({"s"});
+  auto const exp_val  = build_object_value({enc_short_string("hello")});
+
+  auto expected = make_variant_column({exp_meta}, {exp_val});
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*got, expected);
+}
+
+TEST_F(EncodeVariantTest, NullValueEncodedAsVariantNull)
+{
+  // A null INT32 value should produce a null primitive (0x00) inside the object
+  cudf::test::fixed_width_column_wrapper<int32_t> col({0}, {false});
+  cudf::table_view tbl{{col}};
+  std::vector<std::string> names{"x"};
+
+  auto got =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  EXPECT_EQ(got->null_count(), 0);  // struct row is never null
+
+  auto const exp_meta = build_metadata({"x"});
+  auto const exp_val  = build_object_value({enc_null()});
+
+  auto expected = make_variant_column({exp_meta}, {exp_val});
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*got, expected);
+}
+
+TEST_F(EncodeVariantTest, MultipleColumns_SortedNames)
+{
+  // Columns "b", "a": the output dictionary must sort to ["a","b"]
+  cudf::test::fixed_width_column_wrapper<int32_t> col_b{10};
+  cudf::test::strings_column_wrapper col_a{"hi"};
+  cudf::table_view tbl{{col_b, col_a}};
+  std::vector<std::string> names{"b", "a"};
+
+  auto got =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  // Dictionary is sorted: ["a", "b"]. sort_order = [1 (col "a"), 0 (col "b")].
+  // Field 0 (id=0, name="a") → col_a value = enc_short_string("hi")
+  // Field 1 (id=1, name="b") → col_b value = enc_int32(10)
+  auto const exp_meta = build_metadata({"a", "b"});
+  auto const exp_val  = build_object_value({enc_short_string("hi"), enc_int32(10)});
+
+  auto expected = make_variant_column({exp_meta}, {exp_val});
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*got, expected);
+}
+
+TEST_F(EncodeVariantTest, AllIntegerTypes)
+{
+  cudf::test::fixed_width_column_wrapper<int8_t> c8{int8_t{7}};
+  cudf::test::fixed_width_column_wrapper<int16_t> c16{int16_t{300}};
+  cudf::test::fixed_width_column_wrapper<int32_t> c32{int32_t{70000}};
+  cudf::test::fixed_width_column_wrapper<int64_t> c64{int64_t{5000000000LL}};
+  cudf::table_view tbl{{c8, c16, c32, c64}};
+  std::vector<std::string> names{"i8", "i16", "i32", "i64"};
+
+  auto got =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  EXPECT_EQ(got->size(), 1);
+  EXPECT_EQ(got->null_count(), 0);
+
+  // sorted names: i16, i32, i64, i8
+  auto const exp_meta = build_metadata({"i16", "i32", "i64", "i8"});
+  auto const exp_val =
+    build_object_value({enc_int16(300), enc_int32(70000), enc_int64(5000000000LL), enc_int8(7)});
+
+  auto expected = make_variant_column({exp_meta}, {exp_val});
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*got, expected);
+}
+
+TEST_F(EncodeVariantTest, MultiRow)
+{
+  // 3 rows: col "a" (int32), col "b" (string)
+  cudf::test::fixed_width_column_wrapper<int32_t> col_a{1, 2, 3};
+  cudf::test::strings_column_wrapper col_b{"x", "yy", "zzz"};
+  cudf::table_view tbl{{col_a, col_b}};
+  std::vector<std::string> names{"a", "b"};
+
+  auto got =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  EXPECT_EQ(got->size(), 3);
+  EXPECT_EQ(got->null_count(), 0);
+
+  auto const meta     = build_metadata({"a", "b"});
+  auto const exp_val0 = build_object_value({enc_int32(1), enc_short_string("x")});
+  auto const exp_val1 = build_object_value({enc_int32(2), enc_short_string("yy")});
+  auto const exp_val2 = build_object_value({enc_int32(3), enc_short_string("zzz")});
+
+  auto expected = make_variant_column({meta, meta, meta}, {exp_val0, exp_val1, exp_val2});
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*got, expected);
+}
+
+TEST_F(EncodeVariantTest, MixedNullsInMultiColumn)
+{
+  // col "a": [1, null, 3]
+  // col "b": ["x", "y", null]
+  cudf::test::fixed_width_column_wrapper<int32_t> col_a({1, 0, 3}, {true, false, true});
+  cudf::test::strings_column_wrapper col_b({"x", "y", ""}, {true, true, false});
+  cudf::table_view tbl{{col_a, col_b}};
+  std::vector<std::string> names{"a", "b"};
+
+  auto got =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  EXPECT_EQ(got->null_count(), 0);  // struct rows never null in encode_variant
+
+  auto const meta     = build_metadata({"a", "b"});
+  auto const exp_val0 = build_object_value({enc_int32(1), enc_short_string("x")});
+  auto const exp_val1 = build_object_value({enc_null(), enc_short_string("y")});
+  auto const exp_val2 = build_object_value({enc_int32(3), enc_null()});
+
+  auto expected = make_variant_column({meta, meta, meta}, {exp_val0, exp_val1, exp_val2});
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*got, expected);
+}
+
+TEST_F(EncodeVariantTest, RoundtripInt32WithExtractVariant)
+{
+  // Encode, then extract "x" and decode as INT32
+  cudf::test::fixed_width_column_wrapper<int32_t> col{10, 20, 30};
+  cudf::table_view tbl{{col}};
+  std::vector<std::string> names{"x"};
+
+  auto variant =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  auto decoded = cudf::io::parquet::experimental::extract_variant_field(
+    *variant, "x", cudf::data_type{cudf::type_id::INT32}, cudf::test::get_default_stream());
+
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*decoded, col);
+}
+
+TEST_F(EncodeVariantTest, RoundtripStringWithExtractVariant)
+{
+  cudf::test::strings_column_wrapper col{"alpha", "beta", "gamma"};
+  cudf::table_view tbl{{col}};
+  std::vector<std::string> names{"s"};
+
+  auto variant =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  auto decoded = cudf::io::parquet::experimental::extract_variant_field(
+    *variant, "s", cudf::data_type{cudf::type_id::STRING}, cudf::test::get_default_stream());
+
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*decoded, col);
+}
+
+TEST_F(EncodeVariantTest, RoundtripNullsWithExtractVariant)
+{
+  cudf::test::fixed_width_column_wrapper<int32_t> col({5, 0, 15}, {true, false, true});
+  cudf::table_view tbl{{col}};
+  std::vector<std::string> names{"v"};
+
+  auto variant =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  auto decoded = cudf::io::parquet::experimental::extract_variant_field(
+    *variant, "v", cudf::data_type{cudf::type_id::INT32}, cudf::test::get_default_stream());
+
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*decoded, col);
+}
+
+TEST_F(EncodeVariantTest, RoundtripMultiColumnWithExtractVariant)
+{
+  cudf::test::fixed_width_column_wrapper<int32_t> col_a{100, 200, 300};
+  cudf::test::strings_column_wrapper col_b{"p", "q", "r"};
+  cudf::table_view tbl{{col_a, col_b}};
+  std::vector<std::string> names{"a", "b"};
+
+  auto variant =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  auto decoded_a = cudf::io::parquet::experimental::extract_variant_field(
+    *variant, "a", cudf::data_type{cudf::type_id::INT32}, cudf::test::get_default_stream());
+  auto decoded_b = cudf::io::parquet::experimental::extract_variant_field(
+    *variant, "b", cudf::data_type{cudf::type_id::STRING}, cudf::test::get_default_stream());
+
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*decoded_a, col_a);
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*decoded_b, col_b);
+}
+
+TEST_F(EncodeVariantTest, DuplicateColumnNamesThrows)
+{
+  cudf::test::fixed_width_column_wrapper<int32_t> col_a{1};
+  cudf::test::fixed_width_column_wrapper<int32_t> col_b{2};
+  cudf::table_view tbl{{col_a, col_b}};
+  std::vector<std::string> names{"a", "a"};
+  EXPECT_THROW(static_cast<void>(cudf::io::parquet::experimental::encode_variant(
+                 tbl, names, cudf::test::get_default_stream())),
+               std::invalid_argument);
+}
+
+TEST_F(EncodeVariantTest, DuplicateColumnNamesAfterSortThrows)
+{
+  // Names that sort into adjacent positions: ["b", "a", "a"] → sorted ["a", "a", "b"]
+  cudf::test::fixed_width_column_wrapper<int32_t> col_b{1};
+  cudf::test::fixed_width_column_wrapper<int32_t> col_a0{2};
+  cudf::test::fixed_width_column_wrapper<int32_t> col_a1{3};
+  cudf::table_view tbl{{col_b, col_a0, col_a1}};
+  std::vector<std::string> names{"b", "a", "a"};
+  EXPECT_THROW(static_cast<void>(cudf::io::parquet::experimental::encode_variant(
+                 tbl, names, cudf::test::get_default_stream())),
+               std::invalid_argument);
+}
+
+TEST_F(EncodeVariantTest, LargeStringFieldRoundtrip)
+{
+  // Strings longer than 64 bytes must use long_string encoding in the object field
+  std::string const long_s(200, 'k');
+  cudf::test::strings_column_wrapper col{long_s};
+  cudf::table_view tbl{{col}};
+  std::vector<std::string> names{"big"};
+
+  auto variant =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  auto decoded = cudf::io::parquet::experimental::extract_variant_field(
+    *variant, "big", cudf::data_type{cudf::type_id::STRING}, cudf::test::get_default_stream());
+
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*decoded, col);
+}
+
+TEST_F(EncodeVariantTest, MultiBlock)
+{
+  // 300 rows > block_size (256) to exercise a multi-block kernel launch
+  constexpr int N = 300;
+  std::vector<int32_t> vals(N);
+  std::iota(vals.begin(), vals.end(), 0);
+  cudf::test::fixed_width_column_wrapper<int32_t> col(vals.begin(), vals.end());
+  cudf::table_view tbl{{col}};
+  std::vector<std::string> names{"v"};
+
+  auto got =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  EXPECT_EQ(got->size(), N);
+  EXPECT_EQ(got->null_count(), 0);
+
+  auto decoded = cudf::io::parquet::experimental::extract_variant_field(
+    *got, "v", cudf::data_type{cudf::type_id::INT32}, cudf::test::get_default_stream());
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*decoded, col);
+}
+
+TEST_F(EncodeVariantTest, SlicedColumnInput)
+{
+  // Non-zero column offset: build a 5-row table and slice to rows [1, 4).
+  cudf::test::fixed_width_column_wrapper<int32_t> col_full{0, 10, 20, 30, 0};
+  auto const col_sliced = cudf::slice(col_full, {1, 4})[0];
+  cudf::table_view tbl{{col_sliced}};
+  std::vector<std::string> names{"x"};
+
+  auto got =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  EXPECT_EQ(got->size(), 3);
+
+  cudf::test::fixed_width_column_wrapper<int32_t> expected_col{10, 20, 30};
+  auto decoded = cudf::io::parquet::experimental::extract_variant_field(
+    *got, "x", cudf::data_type{cudf::type_id::INT32}, cudf::test::get_default_stream());
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*decoded, expected_col);
+}
+
+TEST_F(EncodeVariantTest, SingleBoolColumnBytes)
+{
+  // Verify boolean values produce the correct 1-byte header (no payload).
+  cudf::test::fixed_width_column_wrapper<bool> col{true, false};
+  cudf::table_view tbl{{col}};
+  std::vector<std::string> names{"b"};
+
+  auto got =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  EXPECT_EQ(got->size(), 2);
+  EXPECT_EQ(got->null_count(), 0);
+
+  auto const meta     = build_metadata({"b"});
+  auto const exp_val0 = build_object_value({enc_bool_true()});
+  auto const exp_val1 = build_object_value({enc_bool_false()});
+
+  auto expected = make_variant_column({meta, meta}, {exp_val0, exp_val1});
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*got, expected);
+}
+
+TEST_F(EncodeVariantTest, RoundtripBoolWithExtractVariant)
+{
+  cudf::test::fixed_width_column_wrapper<bool> col{true, false, true, false};
+  cudf::table_view tbl{{col}};
+  std::vector<std::string> names{"flag"};
+
+  auto variant =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  auto decoded = cudf::io::parquet::experimental::extract_variant_field(
+    *variant, "flag", cudf::data_type{cudf::type_id::BOOL8}, cudf::test::get_default_stream());
+
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*decoded, col);
+}
+
+TEST_F(EncodeVariantTest, RoundtripBoolNullsWithExtractVariant)
+{
+  // Null bool values should encode as VARIANT null and decode back as null BOOL8.
+  cudf::test::fixed_width_column_wrapper<bool> col({true, false, true}, {true, false, true});
+  cudf::table_view tbl{{col}};
+  std::vector<std::string> names{"flag"};
+
+  auto variant =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  EXPECT_EQ(variant->null_count(), 0);  // struct rows never null in encode_variant
+
+  auto decoded = cudf::io::parquet::experimental::extract_variant_field(
+    *variant, "flag", cudf::data_type{cudf::type_id::BOOL8}, cudf::test::get_default_stream());
+
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*decoded, col);
+}
+
+TEST_F(EncodeVariantTest, RoundtripBoolWithIntColumnMultiColumn)
+{
+  // Bool column alongside an int column — exercises multi-column sort and mixed types.
+  cudf::test::fixed_width_column_wrapper<bool> col_flag{true, false, true};
+  cudf::test::fixed_width_column_wrapper<int32_t> col_id{1, 2, 3};
+  cudf::table_view tbl{{col_flag, col_id}};
+  std::vector<std::string> names{"flag", "id"};
+
+  auto variant =
+    cudf::io::parquet::experimental::encode_variant(tbl, names, cudf::test::get_default_stream());
+
+  auto decoded_flag = cudf::io::parquet::experimental::extract_variant_field(
+    *variant, "flag", cudf::data_type{cudf::type_id::BOOL8}, cudf::test::get_default_stream());
+  auto decoded_id = cudf::io::parquet::experimental::extract_variant_field(
+    *variant, "id", cudf::data_type{cudf::type_id::INT32}, cudf::test::get_default_stream());
+
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*decoded_flag, col_flag);
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*decoded_id, col_id);
+}
diff --git a/cpp/tests/io/experimental/variant_extract_test.cpp b/cpp/tests/io/experimental/variant_extract_test.cpp
index 03c334f250e4..d2629eb030fe 100644
--- a/cpp/tests/io/experimental/variant_extract_test.cpp
+++ b/cpp/tests/io/experimental/variant_extract_test.cpp
@@ -445,6 +445,13 @@ inline std::vector<uint8_t> enc_int64(int64_t v)
   return out;
 }
 
+inline std::vector<uint8_t> enc_float32(float v)
+{
+  std::vector<uint8_t> out{make_variant_primitive(variant_primitive_type::FLOAT32)};
+  append_le(out, std::bit_cast<uint32_t>(v), 4);
+  return out;
+}
+
 inline std::vector<uint8_t> enc_float64(double v)
 {
   std::vector<uint8_t> out{make_variant_primitive(variant_primitive_type::FLOAT64)};
@@ -1155,10 +1162,11 @@ TEST_F(CastVariantTest, UnsupportedTypeThrows)
 TEST_F(CastVariantTest, CastSourceTargetMatrix)
 {
   // Exhaustively covers (source physical type) x (supported target) casts. The supported targets
-  // are INT8/16/32/64 and STRING. Expected behaviour:
-  //   - integer targets: only a source whose physical type has the *exact* same width decodes;
-  //   every
-  //     other source (including narrower/wider ints) yields null — cast_variant does not widen.
+  // are INT8/16/32/64, FLOAT32/64, and STRING. Expected behaviour:
+  //   - integer/float targets: only a source whose physical type has the *exact* same width
+  //   decodes;
+  //     every other source (including narrower/wider types) yields null — cast_variant does not
+  //     widen or convert between ints and floats.
   //   - STRING target: short_string and long_string sources decode; every other source yields null.
   auto const stream = cudf::test::get_default_stream();
 
@@ -1174,6 +1182,7 @@ TEST_F(CastVariantTest, CastSourceTargetMatrix)
     {"int16", enc_int16(1234)},
     {"int32", enc_int32(123456)},
     {"int64", enc_int64(1234567890123456789LL)},
+    {"float32", enc_float32(1.5f)},
     {"float64", enc_float64(2.5)},
     {"short_string", enc_short_string("hi")},
     {"long_string", enc_long_string(std::string(70, 'a'))},
@@ -1204,6 +1213,25 @@ TEST_F(CastVariantTest, CastSourceTargetMatrix)
   check_int_target.template operator()<int32_t>("int32", int32_t{123456});
   check_int_target.template operator()<int64_t>("int64", int64_t{1234567890123456789LL});
 
+  // Float targets: exact-width match only; no int<->float conversion.
+  auto check_float_target = [&]<typename T>(char const* match_label, T match_value) {
+    auto const target = cudf::data_type{cudf::type_to_id<T>()};
+    for (auto const& src : sources) {
+      SCOPED_TRACE(std::string{"float target "} + match_label + ", source " + src.label);
+      auto values = values_of(src.bytes);
+      auto got    = cudf::io::parquet::experimental::cast_variant(values, target, stream);
+      if (std::string_view{src.label} == match_label) {
+        cudf::test::fixed_width_column_wrapper<T> const expected{match_value};
+        CUDF_TEST_EXPECT_COLUMNS_EQUAL(*got, expected);
+      } else {
+        ASSERT_EQ(got->size(), 1);
+        EXPECT_EQ(got->null_count(), 1);
+      }
+    }
+  };
+  check_float_target.template operator()<float>("float32", float{1.5f});
+  check_float_target.template operator()<double>("float64", double{2.5});
+
   // STRING target: short_string and long_string decode; every other source is null.
   auto const string_type = cudf::data_type{cudf::type_id::STRING};
   for (auto const& src : sources) {
@@ -1392,3 +1420,105 @@ TEST_F(InvalidInputShapeTest, CastVariantRejectsMalformedInput)
                  std::invalid_argument);
   }
 }
+
+struct EncodeVariantTest : public cudf::test::BaseFixture {};
+
+TEST_F(EncodeVariantTest, Float32RoundTrip)
+{
+  auto const stream = cudf::test::get_default_stream();
+
+  cudf::test::fixed_width_column_wrapper<float> floats({1.5f, -2.25f, 3.14159f});
+  cudf::table_view tbl{{floats}};
+  std::vector<std::string> names{"f"};
+
+  auto encoded = cudf::io::parquet::experimental::encode_variant(tbl, names, stream);
+  auto got     = cudf::io::parquet::experimental::extract_variant_field(
+    *encoded, "f", cudf::data_type{cudf::type_id::FLOAT32}, stream);
+
+  cudf::test::fixed_width_column_wrapper<float> expected({1.5f, -2.25f, 3.14159f});
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*got, expected);
+}
+
+TEST_F(EncodeVariantTest, Float64RoundTrip)
+{
+  auto const stream = cudf::test::get_default_stream();
+
+  cudf::test::fixed_width_column_wrapper<double> doubles({1.5, -2.25, 3.141592653589793});
+  cudf::table_view tbl{{doubles}};
+  std::vector<std::string> names{"d"};
+
+  auto encoded = cudf::io::parquet::experimental::encode_variant(tbl, names, stream);
+  auto got     = cudf::io::parquet::experimental::extract_variant_field(
+    *encoded, "d", cudf::data_type{cudf::type_id::FLOAT64}, stream);
+
+  cudf::test::fixed_width_column_wrapper<double> expected({1.5, -2.25, 3.141592653589793});
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*got, expected);
+}
+
+TEST_F(EncodeVariantTest, FloatNullsRoundTrip)
+{
+  auto const stream = cudf::test::get_default_stream();
+
+  cudf::test::fixed_width_column_wrapper<float> floats({1.0f, 0.0f, -3.5f}, {true, false, true});
+  cudf::table_view tbl{{floats}};
+  std::vector<std::string> names{"f"};
+
+  auto encoded = cudf::io::parquet::experimental::encode_variant(tbl, names, stream);
+  auto got     = cudf::io::parquet::experimental::extract_variant_field(
+    *encoded, "f", cudf::data_type{cudf::type_id::FLOAT32}, stream);
+
+  cudf::test::fixed_width_column_wrapper<float> expected({1.0f, 0.0f, -3.5f}, {true, false, true});
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*got, expected);
+}
+
+TEST_F(EncodeVariantTest, MixedFloatAndIntColumns)
+{
+  auto const stream = cudf::test::get_default_stream();
+
+  cudf::test::fixed_width_column_wrapper<int32_t> ints({10, 20, 30});
+  cudf::test::fixed_width_column_wrapper<float> floats({1.5f, 2.5f, 3.5f});
+  cudf::test::fixed_width_column_wrapper<double> doubles({10.1, 20.2, 30.3});
+  cudf::table_view tbl{{ints, floats, doubles}};
+  std::vector<std::string> names{"i", "f", "d"};
+
+  auto encoded = cudf::io::parquet::experimental::encode_variant(tbl, names, stream);
+
+  auto got_i = cudf::io::parquet::experimental::extract_variant_field(
+    *encoded, "i", cudf::data_type{cudf::type_id::INT32}, stream);
+  auto got_f = cudf::io::parquet::experimental::extract_variant_field(
+    *encoded, "f", cudf::data_type{cudf::type_id::FLOAT32}, stream);
+  auto got_d = cudf::io::parquet::experimental::extract_variant_field(
+    *encoded, "d", cudf::data_type{cudf::type_id::FLOAT64}, stream);
+
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*got_i,
+                                 cudf::test::fixed_width_column_wrapper<int32_t>({10, 20, 30}));
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*got_f,
+                                 cudf::test::fixed_width_column_wrapper<float>({1.5f, 2.5f, 3.5f}));
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(
+    *got_d, cudf::test::fixed_width_column_wrapper<double>({10.1, 20.2, 30.3}));
+}
+
+TEST_F(EncodeVariantTest, ApacheFixtureValuesRoundTrip)
+{
+  // Round-trip the specific float32/float64 values from the Apache parquet-testing fixtures
+  // (1234567936.0f and 1234567890.1234) through encode_variant + extract_variant_field.
+  auto const stream = cudf::test::get_default_stream();
+
+  cudf::test::fixed_width_column_wrapper<float> f32_col({1234567936.0f});
+  cudf::test::fixed_width_column_wrapper<double> f64_col({1234567890.1234});
+  cudf::table_view tbl{{f32_col, f64_col}};
+  std::vector<std::string> names{"float_field", "double_field"};
+
+  auto encoded = cudf::io::parquet::experimental::encode_variant(tbl, names, stream);
+
+  // Extract and cast back — the decoded values must match the originals exactly.
+  auto got_f32 = cudf::io::parquet::experimental::extract_variant_field(
+    *encoded, "float_field", cudf::data_type{cudf::type_id::FLOAT32}, stream);
+  auto got_f64 = cudf::io::parquet::experimental::extract_variant_field(
+    *encoded, "double_field", cudf::data_type{cudf::type_id::FLOAT64}, stream);
+
+  cudf::test::fixed_width_column_wrapper<float> expected_f32({1234567936.0f});
+  cudf::test::fixed_width_column_wrapper<double> expected_f64({1234567890.1234});
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*got_f32, expected_f32);
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(*got_f64, expected_f64);
+}