GH-50994: [C++][Compute] Implement casting from ListView to List with zero-copy fast-path

cpp/src/arrow/compute/kernels/scalar_cast_nested.cc

@@ -17,6 +17,7 @@
 
 // Implementation of casting to (or between) list types
 
+#include <iostream>
 #include <limits>
 #include <map>
 #include <utility>
@@ -26,6 +27,7 @@
 #include "arrow/array/builder_primitive.h"
 #include "arrow/compute/api_scalar.h"
 #include "arrow/compute/cast.h"
+#include "arrow/compute/exec.h"
 #include "arrow/compute/kernels/common_internal.h"
 #include "arrow/compute/kernels/scalar_cast_internal.h"
 #include "arrow/util/bitmap_ops.h"
@@ -141,6 +143,136 @@ void AddListCast(CastFunction* func) {
   DCHECK_OK(func->AddKernel(SrcType::type_id, std::move(kernel)));
 }
 
+template <typename SrcType, typename DestType>
+struct CastListView {
+  using src_offset_type = typename SrcType::offset_type;
+  using dest_offset_type = typename DestType::offset_type;
+
+  static constexpr bool is_downcast = sizeof(src_offset_type) > sizeof(dest_offset_type);
+
+  static bool IsContiguous(const ArraySpan& in_array) {
+    if (in_array.length == 0) return true;
+    const auto* offsets = in_array.GetValues<src_offset_type>(1);
+    const auto* sizes = in_array.GetValues<src_offset_type>(2);
+    for (int64_t i = 0; i < in_array.length - 1; ++i) {
+      if (offsets[i] + sizes[i] != offsets[i + 1]) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  static Status Exec(KernelContext* ctx, const ExecSpan& batch, ExecResult* out) {
+    const CastOptions& options = CastState::Get(ctx);
+    auto child_type = checked_cast<const DestType&>(*out->type()).value_type();
+    const ArraySpan& in_array = batch[0].array;
+    ArrayData* out_array = out->array_data().get();
+
+    ARROW_ASSIGN_OR_RAISE(out_array->buffers[0],
+                          GetOrCopyNullBitmapBuffer(in_array, ctx->memory_pool()));
+
+    std::shared_ptr<ArrayData> values = in_array.child_data[0].ToArrayData();
+
+    const auto* offsets = in_array.GetValues<src_offset_type>(1);
+    const auto* sizes = in_array.GetValues<src_offset_type>(2);
+
+    if (IsContiguous(in_array)) {
+      // Zero-copy fast-path: shift offsets and slice child values
+      ARROW_ASSIGN_OR_RAISE(
+          out_array->buffers[1],
+          ctx->Allocate(sizeof(dest_offset_type) * (in_array.length + 1)));
+      auto* dest_offsets = out_array->GetMutableValues<dest_offset_type>(1);
+
+      src_offset_type start_offset = in_array.length > 0 ? offsets[0] : 0;
+      for (int64_t i = 0; i < in_array.length; ++i) {
+        dest_offsets[i] = static_cast<dest_offset_type>(offsets[i] - start_offset);
+      }
+      if (in_array.length > 0) {
+        dest_offsets[in_array.length] = static_cast<dest_offset_type>(
+            offsets[in_array.length - 1] + sizes[in_array.length - 1] - start_offset);
+      } else {
+        dest_offsets[0] = 0;
+      }
+
+      if (is_downcast && in_array.length > 0) {
+        if (dest_offsets[in_array.length] >
+            std::numeric_limits<dest_offset_type>::max()) {
+          return Status::Invalid("ListView too large to convert to List");
+        }
+      }
+
+      if (in_array.length > 0) {
+        values = values->Slice(start_offset, dest_offsets[in_array.length]);
+      } else {
+        values = values->Slice(0, 0);
+      }
+    } else {
+      // Non-contiguous path: compute new offsets, build take indices, call Take
+      ARROW_ASSIGN_OR_RAISE(
+          out_array->buffers[1],
+          ctx->Allocate(sizeof(dest_offset_type) * (in_array.length + 1)));
+      auto* dest_offsets = out_array->GetMutableValues<dest_offset_type>(1);
+
+      dest_offset_type current_offset = 0;
+      dest_offsets[0] = 0;
+      for (int64_t i = 0; i < in_array.length; ++i) {
+        if (in_array.IsNull(i)) {
+          dest_offsets[i + 1] = current_offset;
+        } else {
+          current_offset += static_cast<dest_offset_type>(sizes[i]);
+          dest_offsets[i + 1] = current_offset;
+        }
+      }
+
+      if (is_downcast) {
+        if (current_offset > std::numeric_limits<dest_offset_type>::max()) {
+          return Status::Invalid("ListView too large to convert to List");
+        }
+      }
+
+      Int64Builder builder(ctx->memory_pool());
+      RETURN_NOT_OK(builder.Reserve(current_offset));
+      for (int64_t i = 0; i < in_array.length; ++i) {
+        if (!in_array.IsNull(i)) {
+          src_offset_type start = offsets[i];
+          src_offset_type size = sizes[i];
+          for (src_offset_type j = 0; j < size; ++j) {
+            builder.UnsafeAppend(start + j);
+          }
+        }
+      }
+
+      ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Array> take_indices, builder.Finish());
+
+      // Call take function
+      ExecContext* exec_ctx = ctx->exec_context();
+      ARROW_ASSIGN_OR_RAISE(
+          Datum taken_values,
+          CallFunction("take", {MakeArray(values), take_indices}, exec_ctx));
+      DCHECK(taken_values.is_array());
+      values = taken_values.array();
+    }
+
+    // Cast values
+    ARROW_ASSIGN_OR_RAISE(Datum cast_values,
+                          Cast(values, child_type, options, ctx->exec_context()));
+    DCHECK(cast_values.is_array());
+    out_array->child_data.push_back(cast_values.array());
+
+    return Status::OK();
+  }
+};
+
+template <typename SrcType, typename DestType>
+void AddListViewCast(CastFunction* func) {
+  ScalarKernel kernel;
+  kernel.exec = CastListView<SrcType, DestType>::Exec;
+  kernel.signature =
+      KernelSignature::Make({InputType(SrcType::type_id)}, kOutputTargetType);
+  kernel.null_handling = NullHandling::COMPUTED_NO_PREALLOCATE;
+  DCHECK_OK(func->AddKernel(SrcType::type_id, std::move(kernel)));
+}
+
 template <typename DestType>
 struct CastFixedToVarList {
   using dest_offset_type = typename DestType::offset_type;
@@ -487,18 +619,18 @@ std::vector<std::shared_ptr<CastFunction>> GetNestedCasts() {
   auto cast_list = std::make_shared<CastFunction>("cast_list", Type::LIST);
   AddCommonCasts(Type::LIST, kOutputTargetType, cast_list.get());
   AddListCast<ListType, ListType>(cast_list.get());
-  AddListCast<ListViewType, ListType>(cast_list.get());
+  AddListViewCast<ListViewType, ListType>(cast_list.get());
   AddListCast<LargeListType, ListType>(cast_list.get());
-  AddListCast<LargeListViewType, ListType>(cast_list.get());
+  AddListViewCast<LargeListViewType, ListType>(cast_list.get());
   AddTypeToTypeCast<CastFixedToVarList<ListType>, FixedSizeListType>(cast_list.get());
 
   auto cast_large_list =
       std::make_shared<CastFunction>("cast_large_list", Type::LARGE_LIST);
   AddCommonCasts(Type::LARGE_LIST, kOutputTargetType, cast_large_list.get());
   AddListCast<ListType, LargeListType>(cast_large_list.get());
-  AddListCast<ListViewType, LargeListType>(cast_large_list.get());
+  AddListViewCast<ListViewType, LargeListType>(cast_large_list.get());
   AddListCast<LargeListType, LargeListType>(cast_large_list.get());
-  AddListCast<LargeListViewType, LargeListType>(cast_large_list.get());
+  AddListViewCast<LargeListViewType, LargeListType>(cast_large_list.get());
   AddTypeToTypeCast<CastFixedToVarList<LargeListType>, FixedSizeListType>(
       cast_large_list.get());
 

cpp/src/arrow/compute/kernels/scalar_cast_test.cc

@@ -3646,6 +3646,72 @@ TEST(Cast, ListToListOptionsPassthru) {
   }
 }
 
+TEST(Cast, ListViewToList) {
+  // 1. Contiguous ListView
+  auto contiguous_src = ArrayFromJSON(list_view(int16()), "[[10, 20], [30], [40, 50]]");
+  auto contiguous_expected = ArrayFromJSON(list(int16()), "[[10, 20], [30], [40, 50]]");
+  CheckCast(contiguous_src, contiguous_expected);
+
+  // Assert zero-copy for contiguous values
+  ASSERT_OK_AND_ASSIGN(auto cast_result, Cast(contiguous_src, list(int16())));
+  auto src_lv = std::dynamic_pointer_cast<ListViewArray>(contiguous_src);
+  auto res_list = std::dynamic_pointer_cast<ListArray>(cast_result.make_array());
+  ASSERT_EQ(res_list->values()->data()->buffers[1]->address(),
+            src_lv->values()->data()->buffers[1]->address());
+
+  // 2. Gapped/Non-contiguous ListView
+  auto values = ArrayFromJSON(int16(), "[10, 20, 999, 30, 40, 50]");
+  auto offsets = ArrayFromJSON(int32(), "[0, 3]");
+  auto sizes = ArrayFromJSON(int32(), "[2, 3]");
+  ASSERT_OK_AND_ASSIGN(auto gapped_src,
+                       ListViewArray::FromArrays(*offsets, *sizes, *values));
+  auto gapped_expected = ArrayFromJSON(list(int16()), "[[10, 20], [30, 40, 50]]");
+  CheckCast(gapped_src, gapped_expected);
+
+  // 3. Overlapping ListView
+  auto overlapping_offsets = ArrayFromJSON(int32(), "[0, 1]");
+  auto overlapping_sizes = ArrayFromJSON(int32(), "[2, 2]");
+  ASSERT_OK_AND_ASSIGN(
+      auto overlapping_src,
+      ListViewArray::FromArrays(*overlapping_offsets, *overlapping_sizes, *values));
+  auto overlapping_expected = ArrayFromJSON(list(int16()), "[[10, 20], [20, 999]]");
+  CheckCast(overlapping_src, overlapping_expected);
+
+  // 4. Large ListView to List and vice versa
+  auto large_contiguous_src =
+      ArrayFromJSON(large_list_view(int16()), "[[10, 20], [30], [40, 50]]");
+  auto large_contiguous_expected =
+      ArrayFromJSON(large_list(int16()), "[[10, 20], [30], [40, 50]]");
+  CheckCast(large_contiguous_src, large_contiguous_expected);
+  CheckCast(contiguous_src, large_contiguous_expected);
+  CheckCast(large_contiguous_src, contiguous_expected);
+
+  // 5. Null Propagation
+  auto nulls_src = ArrayFromJSON(list_view(int16()), "[[10, null], null, [40, 50]]");
+  auto nulls_expected = ArrayFromJSON(list(int16()), "[[10, null], null, [40, 50]]");
+  CheckCast(nulls_src, nulls_expected);
+
+  // 6. Generic and Nested Type casting
+  auto string_src =
+      ArrayFromJSON(list_view(utf8()), "[[\"a\", \"b\"], [\"c\"], [\"d\", \"e\"]]");
+  auto string_expected =
+      ArrayFromJSON(list(utf8()), "[[\"a\", \"b\"], [\"c\"], [\"d\", \"e\"]]");
+  CheckCast(string_src, string_expected);
+
+  auto type_change_src = ArrayFromJSON(list_view(int16()), "[[10, 20], [30], [40, 50]]");
+  auto type_change_expected = ArrayFromJSON(list(int32()), "[[10, 20], [30], [40, 50]]");
+  CheckCast(type_change_src, type_change_expected);
+
+  // 7. Non-Contiguous Slice Boundary Verification
+  auto sliced_gapped_src = gapped_src->Slice(1, 1);
+  auto sliced_gapped_expected = ArrayFromJSON(list(int16()), "[[30, 40, 50]]");
+  CheckCast(sliced_gapped_src, sliced_gapped_expected);
+
+  auto sliced_overlapping_src = overlapping_src->Slice(1, 1);
+  auto sliced_overlapping_expected = ArrayFromJSON(list(int16()), "[[20, 999]]");
+  CheckCast(sliced_overlapping_src, sliced_overlapping_expected);
+}
+
 static void CheckFSLToFSL(const std::vector<std::shared_ptr<DataType>>& value_types,
                           const std::string& json_data,
                           const std::string& tweaked_val_bit_string,