Make ColumnBase.deserialize construct via pylibcudf (#20142)

Broken off from #20087

Authors:
  - Matthew Roeschke (https://github.com/mroeschke)
  - Mads R. B. Kristensen (https://github.com/madsbk)

Approvers:
  - Vyas Ramasubramani (https://github.com/vyasr)

URL: #20142

Author: mroeschke

python/cudf/cudf/core/column/column.py

@@ -44,6 +44,7 @@
     as_buffer,
     cuda_array_interface_wrapper,
 )
+from cudf.core.buffer.spillable_buffer import SpillableBuffer
 from cudf.core.copy_types import GatherMap
 from cudf.core.dtypes import (
     CategoricalDtype,
@@ -121,6 +122,41 @@ def _can_values_be_equal(left: DtypeObj, right: DtypeObj) -> bool:
     return False
 
 
+class spillable_gpumemoryview(plc.gpumemoryview):
+    """
+    HACK: Prevent automatic unspilling of `SpillableBuffer` objects
+    when constructing `plc.Column`.
+
+    The `plc.Column()` constructor expects a `gpumemoryview` object,
+    but wrapping a `SpillableBuffer` directly in a `gpumemoryview`
+    forces the buffer to unspill (materialize) its device data prematurely.
+
+    To avoid this, we wrap spillable buffers in this subclass that implements
+    only the `.obj` attribute; the only attribute actually accessed by
+    `.from_pylibcudf()`. All other attributes intentionally raise errors to
+    prevent accidental usage paths that would cause unspilling.
+    """
+
+    def __init__(self, buf: SpillableBuffer) -> None:
+        self._buf = buf
+
+    @property
+    def obj(self) -> SpillableBuffer:
+        return self._buf
+
+    @property
+    def cai(self) -> None:  # type: ignore[override]
+        assert False
+
+    @property
+    def ptr(self) -> None:  # type: ignore[override]
+        assert False
+
+    @property
+    def nbytes(self) -> None:  # type: ignore[override]
+        assert False
+
+
 class ColumnBase(Serializable, BinaryOperand, Reducible):
     """
     A ColumnBase stores columnar data in device memory.
@@ -411,9 +447,13 @@ def set_mask(self, value) -> Self:
         new_plc_column = self.to_pylibcudf(
             mode="read", use_base=False
         ).with_mask(new_mask, new_null_count)
-        return self.from_pylibcudf(  # type: ignore[return-value]
-            new_plc_column,
-        )._with_type_metadata(self.dtype)
+        return (
+            type(self)
+            .from_pylibcudf(  # type: ignore[return-value]
+                new_plc_column,
+            )
+            ._with_type_metadata(self.dtype)
+        )
 
     @property
     def null_count(self) -> int:
@@ -1948,6 +1988,8 @@ def serialize(self) -> tuple[dict, list]:
             )
             header["subheaders"] = list(child_headers)
             frames.extend(chain(*child_frames))
+        if isinstance(self.dtype, CategoricalDtype):
+            header["codes_dtype"] = self.codes.dtype.str  # type: ignore[attr-defined]
         header["size"] = self.size
         header["frame_count"] = len(frames)
         return header, frames
@@ -1984,13 +2026,45 @@ def unpack(header, frames) -> tuple[Any, list]:
                 child, frames = unpack(h, frames)
                 children.append(child)
         assert len(frames) == 0, "Deserialization did not consume all frames"
-        return build_column(
-            data=data,
-            dtype=dtype,
-            mask=mask,
-            size=header.get("size", None),
-            children=tuple(children),
+        if "codes_dtype" in header:
+            codes_dtype = np.dtype(header["codes_dtype"])
+        else:
+            codes_dtype = None
+        if mask is None:
+            null_count = 0
+        else:
+            null_count = plc.null_mask.null_count(
+                plc.gpumemoryview(mask), 0, header["size"]
+            )
+        if isinstance(dtype, IntervalDtype):
+            # TODO: Handle in dtype_to_pylibcudf_type?
+            plc_type = plc.DataType(plc.TypeId.STRUCT)
+        else:
+            plc_type = dtype_to_pylibcudf_type(
+                codes_dtype if codes_dtype is not None else dtype
+            )
+        if isinstance(dtype, CategoricalDtype):
+            data = children.pop(0)
+
+        if isinstance(data, SpillableBuffer):
+            data = spillable_gpumemoryview(data)
+        elif data is not None:
+            data = plc.gpumemoryview(data)
+        if isinstance(mask, SpillableBuffer):
+            mask = spillable_gpumemoryview(mask)
+        elif mask is not None:
+            mask = plc.gpumemoryview(mask)
+
+        plc_column = plc.Column(
+            plc_type,
+            header["size"],
+            data,
+            mask,
+            null_count,
+            0,
+            [child.to_pylibcudf(mode="read") for child in children],
         )
+        return cls.from_pylibcudf(plc_column)._with_type_metadata(dtype)
 
     def unary_operator(self, unaryop: str):
         raise TypeError(