Use ColumnBase.create more in cudf classic APIs (#21306)

Towards #21229

Namely,

* `list.get`
* `str.join`
* `explode`

Note: I added `tests/extension/test_arrow.py::test_str_join_string_type` to conftest-patch but expect to remove it again in a following PR. We need to enhance our dtype conversion utilities to convert between nested dtype variants like `cudf.ListDtype(object) -> pd.ArrowDtype(pa.string())`

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

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

URL: #21306

Author: mroeschke

python/cudf/cudf/core/accessors/lists.py

@@ -115,13 +115,6 @@ def get(
                     out_of_bounds_mask,
                     pa_scalar_to_plc_scalar(pa.scalar(default)),
                 )
-
-        if self._column.element_type != out.dtype:
-            # libcudf doesn't maintain struct labels so we must transfer over
-            # manually from the input column if we lost some information
-            # somewhere. Not doing this unilaterally since the cost is
-            # non-zero..
-            out = out._with_type_metadata(self._column.element_type)
         return self._return_or_inplace(out)
 
     def contains(self, search_key: ScalarLike) -> Series | Index:

python/cudf/cudf/core/accessors/string.py

@@ -517,17 +517,21 @@ def join(
 
         if isinstance(self._column.dtype, ListDtype):
             list_column = self._column
+            result_dtype = cast("ListDtype", list_column.dtype).element_type
         else:
             # If self._column is not a ListColumn, we will have to
             # split each row by character and create a ListColumn out of it.
             list_column = self._column.fillna("").character_tokenize()
+            result_dtype = cast("ListDtype", list_column.dtype).element_type
             if len(list_column) == 0:
                 list_column = column_empty(  # type: ignore[assignment]
                     len(self._column), dtype=list_column.dtype
                 )
 
         if is_scalar(sep):
-            data = list_column.join_list_elements(sep, string_na_rep, "")  # type: ignore[attr-defined]
+            data = list_column.join_list_elements(  # type: ignore[attr-defined]
+                sep, string_na_rep, "", result_dtype
+            )
         elif can_convert_to_column(sep):
             sep_column = as_column(sep)
             if len(sep_column) != len(list_column):
@@ -544,16 +548,15 @@ def join(
                 sep_column,
                 sep_na_rep,
                 string_na_rep,
+                result_dtype,
             )
         else:
             raise TypeError(
                 f"sep should be an str, array-like or Series object, "
                 f"found {type(sep)}"
             )
 
-        return self._return_or_inplace(
-            data._with_type_metadata(self._column.dtype)
-        )
+        return self._return_or_inplace(data)
 
     def extract(
         self, pat: str, flags: int = 0, expand: bool = True
@@ -4682,8 +4685,9 @@ def character_tokenize(self) -> Series | Index:
         2    .
         dtype: object
         """
-        result_col = ColumnBase.from_pylibcudf(
-            self._column.character_tokenize().plc_column.children()[1]
+        result_col = ColumnBase.create(
+            self._column.character_tokenize().plc_column.children()[1],
+            self._column.dtype,
         )
         if isinstance(self._parent, cudf.Series):
             lengths = self.len().fillna(0)

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

@@ -321,20 +321,24 @@ def sort_lists(
 
     def extract_element_scalar(self, index: int) -> ColumnBase:
         with self.access(mode="read", scope="internal"):
-            return ColumnBase.from_pylibcudf(
-                plc.lists.extract_list_element(
-                    self.plc_column,
-                    index,
-                )
+            plc_column = plc.lists.extract_list_element(
+                self.plc_column,
+                index,
+            )
+            return ColumnBase.create(
+                plc_column,
+                self.dtype.element_type,  # type: ignore[union-attr]
             )
 
     def extract_element_column(self, index: ColumnBase) -> ColumnBase:
         with self.access(mode="read", scope="internal"):
-            return ColumnBase.from_pylibcudf(
-                plc.lists.extract_list_element(
-                    self.plc_column,
-                    index.plc_column,
-                )
+            plc_column = plc.lists.extract_list_element(
+                self.plc_column,
+                index.plc_column,
+            )
+            return ColumnBase.create(
+                plc_column,
+                self.dtype.element_type,  # type: ignore[union-attr]
             )
 
     def contains_scalar(self, search_key: pa.Scalar) -> ColumnBase:
@@ -404,6 +408,7 @@ def join_list_elements(
         separator: str | StringColumn,
         sep_na_rep: str,
         string_na_rep: str,
+        result_dtype: DtypeObj,
     ) -> StringColumn:
         with self.access(mode="read", scope="internal"):
             if isinstance(separator, str):
@@ -422,7 +427,7 @@ def join_list_elements(
             )
             return cast(
                 "cudf.core.column.string.StringColumn",
-                type(self).from_pylibcudf(plc_column),
+                ColumnBase.create(plc_column, result_dtype),
             )
 
     def minhash_ngrams(

python/cudf/cudf/core/indexed_frame.py

@@ -5488,6 +5488,17 @@ def _explode(self, explode_column: Any, ignore_index: bool):
         else:
             idx_cols = ()
 
+        explode_column_idx = column_index + len(idx_cols)
+        # We must copy inner datatype of the exploded list column to
+        # maintain struct dtype key names
+        exploded_type = cast(
+            "ListDtype", self._columns[column_index].dtype
+        ).element_type
+        result_types = (
+            exploded_type if i == explode_column_idx else col.dtype
+            for i, col in enumerate(itertools.chain(idx_cols, self._columns))
+        )
+
         with access_columns(
             *itertools.chain(idx_cols, self._columns),
             mode="read",
@@ -5500,33 +5511,14 @@ def _explode(self, explode_column: Any, ignore_index: bool):
                         for col in itertools.chain(idx_cols, self._columns)
                     ]
                 ),
-                column_index + len(idx_cols),
+                explode_column_idx,
             )
             exploded = [
-                ColumnBase.from_pylibcudf(col) for col in plc_table.columns()
-            ]
-        # We must copy inner datatype of the exploded list column to
-        # maintain struct dtype key names
-        element_type = cast(
-            ListDtype, self._columns[column_index].dtype
-        ).element_type
-
-        column_index += len(idx_cols)
-        exploded = [
-            ColumnBase.create(
-                new_column.plc_column,
-                element_type,
-            )
-            if i == column_index
-            else ColumnBase.create(new_column.plc_column, old_column.dtype)
-            for i, (new_column, old_column) in enumerate(
-                zip(
-                    exploded,
-                    itertools.chain(idx_cols, self._columns),
-                    strict=True,
+                ColumnBase.create(plc_column, dtype=dtype)
+                for plc_column, dtype in zip(
+                    plc_table.columns(), result_types, strict=True
                 )
-            )
-        ]
+            ]
 
         data = type(self._data)(
             dict(
@@ -7078,9 +7070,7 @@ def _append_new_row_inplace(col: ColumnBase, value: ScalarLike) -> None:
             to_type = col.dtype
     val_col = val_col.astype(to_type)
     old_col = col.astype(to_type)
-    res_col = ColumnBase.create(
-        concat_columns([old_col, val_col]).plc_column, to_type
-    )
+    res_col = concat_columns([old_col, val_col])
     if (
         cudf.get_option("mode.pandas_compatible")
         and res_col.dtype != col.dtype

python/cudf/cudf/pandas/scripts/conftest-patch.py

@@ -1606,6 +1606,7 @@ def set_copy_on_write_option():
     "tests/extension/test_arrow.py::test_round",
     "tests/extension/test_arrow.py::test_str_extract_non_symbolic",
     "tests/extension/test_arrow.py::test_str_find[ab-0-None-exp0-exp_typ0]",
+    "tests/extension/test_arrow.py::test_str_join_string_type",
     "tests/extension/test_arrow.py::test_str_len",
     "tests/extension/test_arrow.py::test_str_repeat_unsupported",
     "tests/extension/test_arrow.py::test_str_replace_unsupported[pat-b]",
@@ -6056,7 +6057,6 @@ def set_copy_on_write_option():
     "tests/series/accessors/test_dt_accessor.py::TestSeriesDatetimeValues::test_strftime",
     "tests/series/accessors/test_dt_accessor.py::TestSeriesDatetimeValues::test_strftime_dt64_days",
     "tests/series/accessors/test_dt_accessor.py::TestSeriesDatetimeValues::test_valid_dt_with_missing_values",
-    "tests/series/accessors/test_list_accessor.py::test_list_getitem_invalid_index[list_dtype0]",
     "tests/series/accessors/test_list_accessor.py::test_list_len",
     "tests/series/accessors/test_struct_accessor.py::test_struct_accessor_dtypes",
     "tests/series/accessors/test_struct_accessor.py::test_struct_accessor_field_expanded[0-int_col]",