Use ColumnBase.create more in frame.py (rapidsai#21394)

Towards rapidsai#21229

Additionally preserves the original types more than just converting `from_pylibcudf`

Authors:
  - Matthew Roeschke (https://github.com/mroeschke)

Approvers:
  - GALI PREM SAGAR (https://github.com/galipremsagar)

URL: rapidsai#21394

Author: mroeschke

python/cudf/cudf/core/frame.py

@@ -18,9 +18,6 @@
 
 import cudf
 from cudf.api.extensions import no_default
-
-# TODO: The `numpy` import is needed for typing purposes during doc builds
-# only, need to figure out why the `np` alias is insufficient then remove.
 from cudf.api.types import is_dtype_equal, is_scalar, is_string_dtype
 from cudf.core._compat import PANDAS_LT_300
 from cudf.core._internals import copying, sorting
@@ -36,6 +33,7 @@
 from cudf.core.dtype.validators import is_dtype_obj_numeric
 from cudf.core.mixins import BinaryOperand, Scannable
 from cudf.utils.dtypes import (
+    dtype_from_pylibcudf_column,
     find_common_type,
     is_pandas_nullable_extension_dtype,
 )
@@ -1517,16 +1515,17 @@ def searchsorted(
             )
         ]
 
-        outcol = ColumnBase.from_pylibcudf(
-            sorting.search_sorted(
-                sources,
-                values,
-                side,
-                ascending=itertools.repeat(ascending, times=len(sources)),
-                na_position=itertools.repeat(na_position, times=len(sources)),
-            )
+        plc_outcol = sorting.search_sorted(
+            sources,
+            values,
+            side,
+            ascending=itertools.repeat(ascending, times=len(sources)),
+            na_position=itertools.repeat(na_position, times=len(sources)),
+        )
+        outcol = ColumnBase.create(
+            plc.unary.cast(plc_outcol, plc.DataType(plc.TypeId.INT64)),
+            np.dtype("int64"),
         )
-        outcol = outcol.astype(np.dtype("int64"))
 
         # Return result as cupy array if the values is non-scalar
         # If values is scalar, result is expected to be scalar.
@@ -1645,13 +1644,14 @@ def _get_sorted_inds(
             )
         else:
             ascending_iter = ascending
-        return ColumnBase.from_pylibcudf(
-            sorting.order_by(
-                to_sort,
-                ascending_iter,
-                itertools.repeat(na_position, times=len(to_sort)),
-                stable=True,
-            )
+        plc_result = sorting.order_by(
+            to_sort,
+            ascending_iter,
+            itertools.repeat(na_position, times=len(to_sort)),
+            stable=True,
+        )
+        return ColumnBase.create(
+            plc_result, dtype_from_pylibcudf_column(plc_result)
         )
 
     @_performance_tracking
@@ -1668,7 +1668,12 @@ def _split(self, splits: list[int]) -> list[Self]:
             return []
         return [
             self._from_columns_like_self(
-                [ColumnBase.from_pylibcudf(col) for col in split],
+                [
+                    ColumnBase.create(col, dtype)
+                    for col, (_, dtype) in zip(
+                        split, self._dtypes, strict=True
+                    )
+                ],
                 self._column_names,
             )
             for split in copying.columns_split(self._columns, splits)
@@ -1680,9 +1685,14 @@ def _encode(self) -> tuple[Self, ColumnBase]:
             plc.Table([col.plc_column for col in self._columns])
         )
         columns = [
-            ColumnBase.from_pylibcudf(col) for col in plc_table.columns()
+            ColumnBase.create(col, dtype)
+            for col, (_, dtype) in zip(
+                plc_table.columns(), self._dtypes, strict=True
+            )
         ]
-        indices = ColumnBase.from_pylibcudf(plc_column)
+        indices = ColumnBase.create(
+            plc_column, dtype_from_pylibcudf_column(plc_column)
+        )
         keys = self._from_columns_like_self(columns)
         return keys, indices
 
@@ -2142,8 +2152,12 @@ def _repeat(
             else:
                 repeats_plc = repeats
             return [
-                ColumnBase.from_pylibcudf(col)
-                for col in plc.filling.repeat(plc_table, repeats_plc).columns()
+                ColumnBase.create(col, reference_col.dtype)
+                for col, reference_col in zip(
+                    plc.filling.repeat(plc_table, repeats_plc).columns(),
+                    columns,
+                    strict=True,
+                )
             ]
 
     @_performance_tracking