From 4458a3de09e770e7f42e3c97c99b7a4ae9372e4b Mon Sep 17 00:00:00 2001
From: "Maarten A. Breddels" <maartenbreddels@gmail.com>
Date: Tue, 22 Dec 2020 13:21:43 +0100
Subject: [PATCH 1/5] ARROW-10959: [C++] Add scalar string join kernel

---
 cpp/src/arrow/array/builder_binary.h          |  19 +++
 cpp/src/arrow/compute/function.cc             |   5 +-
 .../arrow/compute/kernels/scalar_string.cc    | 140 ++++++++++++++++++
 .../compute/kernels/scalar_string_test.cc     |  29 ++++
 cpp/src/arrow/compute/kernels/test_util.cc    |  78 +++++++---
 cpp/src/arrow/compute/kernels/test_util.h     |   5 +
 docs/source/cpp/compute.rst                   |  18 ++-
 python/pyarrow/tests/test_compute.py          |  11 ++
 8 files changed, 281 insertions(+), 24 deletions(-)

diff --git a/cpp/src/arrow/array/builder_binary.h b/cpp/src/arrow/array/builder_binary.h
index bc49c7d6787..16ca1694b9f 100644
--- a/cpp/src/arrow/array/builder_binary.h
+++ b/cpp/src/arrow/array/builder_binary.h
@@ -77,6 +77,25 @@ class BaseBinaryBuilder : public ArrayBuilder {
     return Append(value.data(), static_cast<offset_type>(value.size()));
   }
 
+  /// AppendCurrent does not add a new offset
+  Status AppendCurrent(const uint8_t* value, offset_type length) {
+    // Safety check for UBSAN.
+    if (ARROW_PREDICT_TRUE(length > 0)) {
+      ARROW_RETURN_NOT_OK(ValidateOverflow(length));
+      ARROW_RETURN_NOT_OK(value_data_builder_.Append(value, length));
+    }
+
+    return Status::OK();
+  }
+
+  Status AppendCurrent(const char* value, offset_type length) {
+    return AppendCurrent(reinterpret_cast<const uint8_t*>(value), length);
+  }
+
+  Status AppendCurrent(util::string_view value) {
+    return AppendCurrent(value.data(), static_cast<offset_type>(value.size()));
+  }
+
   Status AppendNulls(int64_t length) final {
     const int64_t num_bytes = value_data_builder_.length();
     ARROW_RETURN_NOT_OK(Reserve(length));
diff --git a/cpp/src/arrow/compute/function.cc b/cpp/src/arrow/compute/function.cc
index f74bb245d77..0f94baaedfc 100644
--- a/cpp/src/arrow/compute/function.cc
+++ b/cpp/src/arrow/compute/function.cc
@@ -210,8 +210,9 @@ Status Function::Validate() const {
     if (arity_.is_varargs && arg_count == arity_.num_args + 1) {
       return Status::OK();
     }
-    return Status::Invalid("In function '", name_,
-                           "': ", "number of argument names != function arity");
+    return Status::Invalid(
+        "In function '", name_,
+        "': ", "number of argument names for function documentation != function arity");
   }
   return Status::OK();
 }
diff --git a/cpp/src/arrow/compute/kernels/scalar_string.cc b/cpp/src/arrow/compute/kernels/scalar_string.cc
index 1d87bd86c67..96852c91f66 100644
--- a/cpp/src/arrow/compute/kernels/scalar_string.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_string.cc
@@ -2637,6 +2637,145 @@ void AddUtf8Length(FunctionRegistry* registry) {
   DCHECK_OK(registry->AddFunction(std::move(func)));
 }
 
+// binary join
+
+template <typename Type>
+struct BinaryJoin {
+  using ArrayType = typename TypeTraits<Type>::ArrayType;
+  using ListArrayType = ListArray;
+  using offset_type = typename Type::offset_type;
+  using BuilderType = typename TypeTraits<Type>::BuilderType;
+
+  static Status Exec(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
+    if (batch[0].kind() == Datum::SCALAR) {
+      const ListScalar& list = checked_cast<const ListScalar&>(*batch[0].scalar());
+      if (!list.is_valid) {
+        return Status::OK();
+      }
+      if (batch[1].kind() == Datum::SCALAR) {
+        const BaseBinaryScalar& separator_scalar =
+            checked_cast<const BaseBinaryScalar&>(*batch[1].scalar());
+        if (!separator_scalar.is_valid) {
+          return Status::OK();
+        }
+        util::string_view separator(*separator_scalar.value);
+
+        TypedBufferBuilder<uint8_t> builder(ctx->memory_pool());
+        auto Append = [&](util::string_view value) {
+          return builder.Append(reinterpret_cast<const uint8_t*>(value.data()),
+                                static_cast<offset_type>(value.size()));
+        };
+
+        const ArrayType* strings = static_cast<const ArrayType*>(list.value.get());
+        if (strings->null_count() > 0) {
+          // since the input list is not null, the out datum needs to be assigned to
+          *out = MakeNullScalar(list.value->type());
+          return Status::OK();
+        }
+        if (strings->length() > 0) {
+          RETURN_NOT_OK(Append(strings->GetView(0)));
+          for (int64_t j = 1; j < strings->length(); j++) {
+            RETURN_NOT_OK(Append(separator));
+            RETURN_NOT_OK(Append(strings->GetView(j)));
+          }
+        }
+        std::shared_ptr<Buffer> string_buffer;
+        RETURN_NOT_OK(builder.Finish(&string_buffer));
+        ARROW_ASSIGN_OR_RAISE(auto scalar_right_type,
+                              MakeScalar<std::shared_ptr<Buffer>>(
+                                  list.value->type(), std::move(string_buffer)));
+        *out = scalar_right_type;
+      }
+      // XXX do we want to support scalar[list[str]] with array[str] ?
+    } else {
+      const ListArrayType list(batch[0].array());
+      ArrayData* output = out->mutable_array();
+
+      BuilderType builder(ctx->memory_pool());
+      RETURN_NOT_OK(builder.Reserve(list.length()));
+      if (batch[1].kind() == Datum::ARRAY) {
+        ArrayType separator_array(batch[1].array());
+        for (int64_t i = 0; i < list.length(); ++i) {
+          const std::shared_ptr<Array> slice = list.value_slice(i);
+          const ArrayType* strings = static_cast<const ArrayType*>(slice.get());
+          if ((strings->null_count() > 0) || (list.IsNull(i)) ||
+              separator_array.IsNull(i)) {
+            RETURN_NOT_OK(builder.AppendNull());
+          } else {
+            const auto separator = separator_array.GetView(i);
+            if (strings->length() > 0) {
+              RETURN_NOT_OK(builder.Append(strings->GetView(0)));
+              for (int64_t j = 1; j < strings->length(); j++) {
+                RETURN_NOT_OK(builder.AppendCurrent(separator));
+                RETURN_NOT_OK(builder.AppendCurrent(strings->GetView(j)));
+              }
+            } else {
+              RETURN_NOT_OK(builder.AppendEmptyValue());
+            }
+          }
+        }
+      } else if (batch[1].kind() == Datum::SCALAR) {
+        const auto& separator_scalar =
+            checked_cast<const BaseBinaryScalar&>(*batch[1].scalar());
+        if (!separator_scalar.is_valid) {
+          ARROW_ASSIGN_OR_RAISE(
+              auto nulls,
+              MakeArrayOfNull(list.value_type(), list.length(), ctx->memory_pool()));
+          *output = *nulls->data();
+          output->type = list.value_type();
+          return Status::OK();
+        }
+        util::string_view separator(*separator_scalar.value);
+
+        for (int64_t i = 0; i < list.length(); ++i) {
+          const std::shared_ptr<Array> slice = list.value_slice(i);
+          const ArrayType* strings = static_cast<const ArrayType*>(slice.get());
+          if ((strings->null_count() > 0) || (list.IsNull(i))) {
+            RETURN_NOT_OK(builder.AppendNull());
+          } else {
+            if (strings->length() > 0) {
+              RETURN_NOT_OK(builder.Append(strings->GetView(0)));
+              for (int64_t j = 1; j < strings->length(); j++) {
+                RETURN_NOT_OK(builder.AppendCurrent(separator));
+                RETURN_NOT_OK(builder.AppendCurrent(strings->GetView(j)));
+              }
+            } else {
+              RETURN_NOT_OK(builder.AppendEmptyValue());
+            }
+          }
+        }
+      }
+      std::shared_ptr<Array> string_array;
+      RETURN_NOT_OK(builder.Finish(&string_array));
+      *output = *string_array->data();
+      // correct the output type based on the input
+      output->type = list.value_type();
+    }
+    return Status::OK();
+  }
+};
+
+const FunctionDoc binary_join_doc(
+    "Join a list of strings together with a `separator` to form a single string",
+    ("Insert `separator` between each list element, and concatenate them."),
+    {"list", "separator"});
+
+void AddJoin(FunctionRegistry* registry) {
+  auto func =
+      std::make_shared<ScalarFunction>("binary_join", Arity::Binary(), &binary_join_doc);
+  for (const std::shared_ptr<DataType>& ty : BaseBinaryTypes()) {
+    auto exec = GenerateTypeAgnosticVarBinaryBase<BinaryJoin>(*ty);
+    // TODO add large_list inputs
+    DCHECK_OK(
+        func->AddKernel({InputType::Array(list(ty)), InputType::Scalar(ty)}, ty, exec));
+    DCHECK_OK(
+        func->AddKernel({InputType::Array(list(ty)), InputType::Array(ty)}, ty, exec));
+    DCHECK_OK(
+        func->AddKernel({InputType::Scalar(list(ty)), InputType::Scalar(ty)}, ty, exec));
+  }
+  DCHECK_OK(registry->AddFunction(std::move(func)));
+}
+
 template <template <typename> class ExecFunctor>
 void MakeUnaryStringBatchKernel(
     std::string name, FunctionRegistry* registry, const FunctionDoc* doc,
@@ -2942,6 +3081,7 @@ void RegisterScalarStringAscii(FunctionRegistry* registry) {
   AddSlice(registry);
   AddSplit(registry);
   AddStrptime(registry);
+  AddJoin(registry);
 }
 
 }  // namespace internal
diff --git a/cpp/src/arrow/compute/kernels/scalar_string_test.cc b/cpp/src/arrow/compute/kernels/scalar_string_test.cc
index fe069810dbd..16c231e5ce1 100644
--- a/cpp/src/arrow/compute/kernels/scalar_string_test.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_string_test.cc
@@ -42,6 +42,7 @@ template <typename TestType>
 class BaseTestStringKernels : public ::testing::Test {
  protected:
   using OffsetType = typename TypeTraits<TestType>::OffsetType;
+  using ScalarType = typename TypeTraits<TestType>::ScalarType;
 
   void CheckUnary(std::string func_name, std::string json_input,
                   std::shared_ptr<DataType> out_ty, std::string json_expected,
@@ -59,6 +60,11 @@ class BaseTestStringKernels : public ::testing::Test {
 
   std::shared_ptr<DataType> type() { return TypeTraits<TestType>::type_singleton(); }
 
+  template <typename CType>
+  std::shared_ptr<ScalarType> scalar(CType value) {
+    return std::make_shared<ScalarType>(value);
+  }
+
   std::shared_ptr<DataType> offset_type() {
     return TypeTraits<OffsetType>::type_singleton();
   }
@@ -775,6 +781,29 @@ TYPED_TEST(TestStringKernels, StrptimeDoesNotProvideDefaultOptions) {
   ASSERT_RAISES(Invalid, CallFunction("strptime", {input}));
 }
 
+TYPED_TEST(TestStringKernels, BinaryJoin) {
+  auto separator = this->scalar("--");
+  auto list_input = ArrayFromJSON(
+      list(this->type()),
+      R"([["a", "bb", "ccc"], [], null, ["dd"], ["eee", null], ["ff", ""]])");
+  auto expected =
+      ArrayFromJSON(this->type(), R"(["a--bb--ccc", "", null, "dd", null, "ff--"])");
+  CheckScalarBinary("binary_join", list_input, separator, expected);
+
+  auto separator_null = MakeNullScalar(this->type());
+  expected = ArrayFromJSON(this->type(), R"([null, null, null, null, null, null])");
+  CheckScalarBinary("binary_join", list_input, separator_null, expected);
+
+  auto separators =
+      ArrayFromJSON(this->type(), R"(["1", "2", "3", "4", "5", "6", null])");
+  list_input = ArrayFromJSON(
+      list(this->type()),
+      R"([["a", "bb", "ccc"], [], null, ["dd"], ["eee", null], ["ff", ""], ["hh", "ii"]])");
+  expected =
+      ArrayFromJSON(this->type(), R"(["a1bb1ccc", "", null, "dd", null, "ff6", null])");
+  CheckScalarBinary("binary_join", list_input, separators, expected);
+}
+
 #ifdef ARROW_WITH_UTF8PROC
 
 TYPED_TEST(TestStringKernels, TrimWhitespaceUTF8) {
diff --git a/cpp/src/arrow/compute/kernels/test_util.cc b/cpp/src/arrow/compute/kernels/test_util.cc
index c74ef3b76dd..2e6bb558674 100644
--- a/cpp/src/arrow/compute/kernels/test_util.cc
+++ b/cpp/src/arrow/compute/kernels/test_util.cc
@@ -35,8 +35,10 @@ namespace compute {
 
 namespace {
 
+using DatumVector = std::vector<Datum>;
+
 template <typename T>
-std::vector<Datum> GetDatums(const std::vector<T>& inputs) {
+DatumVector GetDatums(const std::vector<T>& inputs) {
   std::vector<Datum> datums;
   for (const auto& input : inputs) {
     datums.emplace_back(input);
@@ -44,28 +46,36 @@ std::vector<Datum> GetDatums(const std::vector<T>& inputs) {
   return datums;
 }
 
-void CheckScalarNonRecursive(const std::string& func_name, const ArrayVector& inputs,
+void CheckScalarNonRecursive(const std::string& func_name, const DatumVector& inputs,
                              const std::shared_ptr<Array>& expected,
                              const FunctionOptions* options) {
-  ASSERT_OK_AND_ASSIGN(Datum out, CallFunction(func_name, GetDatums(inputs), options));
+  ASSERT_OK_AND_ASSIGN(Datum out, CallFunction(func_name, inputs, options));
   std::shared_ptr<Array> actual = std::move(out).make_array();
   ASSERT_OK(actual->ValidateFull());
   AssertArraysEqual(*expected, *actual, /*verbose=*/true);
 }
 
 template <typename... SliceArgs>
-ArrayVector SliceAll(const ArrayVector& inputs, SliceArgs... slice_args) {
-  ArrayVector sliced;
+DatumVector SliceArrays(const DatumVector& inputs, SliceArgs... slice_args) {
+  DatumVector sliced;
   for (const auto& input : inputs) {
-    sliced.push_back(input->Slice(slice_args...));
+    if (input.is_array()) {
+      sliced.push_back(*input.make_array()->Slice(slice_args...));
+    } else {
+      sliced.push_back(input);
+    }
   }
   return sliced;
 }
 
-ScalarVector GetScalars(const ArrayVector& inputs, int64_t index) {
+ScalarVector GetScalars(const DatumVector& inputs, int64_t index) {
   ScalarVector scalars;
   for (const auto& input : inputs) {
-    scalars.push_back(*input->GetScalar(index));
+    if (input.is_array()) {
+      scalars.push_back(*input.make_array()->GetScalar(index));
+    } else {
+      scalars.push_back(input.scalar());
+    }
   }
   return scalars;
 }
@@ -93,44 +103,63 @@ void CheckScalar(std::string func_name, const ScalarVector& inputs,
   }
 }
 
-void CheckScalar(std::string func_name, const ArrayVector& inputs,
+void CheckScalar(std::string func_name, const DatumVector& inputs,
                  std::shared_ptr<Array> expected, const FunctionOptions* options) {
   CheckScalarNonRecursive(func_name, inputs, expected, options);
 
+  // check for at least 1 array, and make sure the others are of equal length
+  std::shared_ptr<Array> array;
+  for (const auto& input : inputs) {
+    if (input.is_array()) {
+      if (!array) {
+        array = input.make_array();
+      } else {
+        ASSERT_EQ(input.array()->length, array->length());
+      }
+    }
+  }
+
   // Check all the input scalars, if scalars are implemented
-  if (std::none_of(inputs.begin(), inputs.end(), [](const std::shared_ptr<Array>& array) {
-        return array->type_id() == Type::EXTENSION;
+  if (std::none_of(inputs.begin(), inputs.end(), [](const Datum& datum) {
+        return datum.type()->id() == Type::EXTENSION;
       })) {
-    for (int64_t i = 0; i < inputs[0]->length(); ++i) {
+    // Check all the input scalars
+    for (int64_t i = 0; i < array->length(); ++i) {
       CheckScalar(func_name, GetScalars(inputs, i), *expected->GetScalar(i), options);
     }
   }
 
   // Since it's a scalar function, calling it on sliced inputs should
   // result in the sliced expected output.
-  const auto slice_length = inputs[0]->length() / 3;
+  const auto slice_length = array->length() / 3;
   if (slice_length > 0) {
-    CheckScalarNonRecursive(func_name, SliceAll(inputs, 0, slice_length),
+    CheckScalarNonRecursive(func_name, SliceArrays(inputs, 0, slice_length),
                             expected->Slice(0, slice_length), options);
 
-    CheckScalarNonRecursive(func_name, SliceAll(inputs, slice_length, slice_length),
+    CheckScalarNonRecursive(func_name, SliceArrays(inputs, slice_length, slice_length),
                             expected->Slice(slice_length, slice_length), options);
 
-    CheckScalarNonRecursive(func_name, SliceAll(inputs, 2 * slice_length),
+    CheckScalarNonRecursive(func_name, SliceArrays(inputs, 2 * slice_length),
                             expected->Slice(2 * slice_length), options);
   }
 
   // Should also work with an empty slice
-  CheckScalarNonRecursive(func_name, SliceAll(inputs, 0, 0), expected->Slice(0, 0),
+  CheckScalarNonRecursive(func_name, SliceArrays(inputs, 0, 0), expected->Slice(0, 0),
                           options);
 
   // Ditto with ChunkedArray inputs
   if (slice_length > 0) {
-    std::vector<std::shared_ptr<ChunkedArray>> chunked_inputs;
+    DatumVector chunked_inputs;
     chunked_inputs.reserve(inputs.size());
     for (const auto& input : inputs) {
-      chunked_inputs.push_back(std::make_shared<ChunkedArray>(
-          ArrayVector{input->Slice(0, slice_length), input->Slice(slice_length)}));
+      if (input.is_array()) {
+        auto ar = input.make_array();
+        auto ar_chunked = std::make_shared<ChunkedArray>(
+            ArrayVector{ar->Slice(0, slice_length), ar->Slice(slice_length)});
+        chunked_inputs.push_back(ar_chunked);
+      } else {
+        chunked_inputs.push_back(input.scalar());
+      }
     }
     ArrayVector expected_chunks{expected->Slice(0, slice_length),
                                 expected->Slice(slice_length)};
@@ -144,7 +173,8 @@ void CheckScalar(std::string func_name, const ArrayVector& inputs,
 
 void CheckScalarUnary(std::string func_name, std::shared_ptr<Array> input,
                       std::shared_ptr<Array> expected, const FunctionOptions* options) {
-  CheckScalar(std::move(func_name), {input}, expected, options);
+  ArrayVector input_vector = {input};
+  CheckScalar(std::move(func_name), GetDatums(input_vector), expected, options);
 }
 
 void CheckScalarUnary(std::string func_name, std::shared_ptr<DataType> in_ty,
@@ -179,6 +209,12 @@ void CheckScalarBinary(std::string func_name, std::shared_ptr<Array> left_input,
   CheckScalar(std::move(func_name), {left_input, right_input}, expected, options);
 }
 
+void CheckScalarBinary(std::string func_name, std::shared_ptr<Array> left_input,
+                       std::shared_ptr<Scalar> right_input,
+                       std::shared_ptr<Array> expected, const FunctionOptions* options) {
+  CheckScalar(std::move(func_name), {left_input, right_input}, expected, options);
+}
+
 void CheckDispatchBest(std::string func_name, std::vector<ValueDescr> original_values,
                        std::vector<ValueDescr> expected_equivalent_values) {
   ASSERT_OK_AND_ASSIGN(auto function, GetFunctionRegistry()->GetFunction(func_name));
diff --git a/cpp/src/arrow/compute/kernels/test_util.h b/cpp/src/arrow/compute/kernels/test_util.h
index cadcc4fe35c..927e0eb699f 100644
--- a/cpp/src/arrow/compute/kernels/test_util.h
+++ b/cpp/src/arrow/compute/kernels/test_util.h
@@ -120,6 +120,11 @@ void CheckScalarBinary(std::string func_name, std::shared_ptr<Array> left_input,
                        std::shared_ptr<Array> expected,
                        const FunctionOptions* options = nullptr);
 
+void CheckScalarBinary(std::string func_name, std::shared_ptr<Array> left_input,
+                       std::shared_ptr<Scalar> right_input,
+                       std::shared_ptr<Array> expected,
+                       const FunctionOptions* options = nullptr);
+
 void CheckVectorUnary(std::string func_name, Datum input, std::shared_ptr<Array> expected,
                       const FunctionOptions* options = nullptr);
 
diff --git a/docs/source/cpp/compute.rst b/docs/source/cpp/compute.rst
index 3f30bbcaa06..94bb0720acd 100644
--- a/docs/source/cpp/compute.rst
+++ b/docs/source/cpp/compute.rst
@@ -653,10 +653,26 @@ String component extraction
   ``(?P<letter>[ab])(?P<digit>\\d)``.
 
 
+String joining
+~~~~~~~~~~~~~~
+
+This function does the inverse of string splitting.
+
++-----------------+-----------+----------------------+----------------+-------------------+---------+
+| Function name   | Arity     | Input type 1         | Input type 2   | Output type       | Notes   |
++=================+===========+======================+================+===================+=========+
+| binary_join     | Binary    | List of string-like  | String-like    | String-like       | \(1)    |
++-----------------+-----------+----------------------+----------------+-------------------+---------+
+
+* \(1) The first input must be an array, while the second can be a scalar or array.
+  Each list of values in the first input is joined using each second input
+  as separator.
+
+
 Slicing
 ~~~~~~~
 
-These function transform each sequence of the array to a subsequence, according
+This function transforms each sequence of the array to a subsequence, according
 to start and stop indices, and a non-zero step (defaulting to 1).  Slicing
 semantics follow Python slicing semantics: the start index is inclusive,
 the stop index exclusive; if the step is negative, the sequence is followed
diff --git a/python/pyarrow/tests/test_compute.py b/python/pyarrow/tests/test_compute.py
index b3f87127397..a78be20f9cf 100644
--- a/python/pyarrow/tests/test_compute.py
+++ b/python/pyarrow/tests/test_compute.py
@@ -699,6 +699,17 @@ def test_extract_regex():
         'letter': 'b', 'digit': '2'}]
 
 
+def test_binary_join():
+    ar_list = pa.array([['foo', 'bar'], None, []])
+    expected = pa.array(['foo-bar', None, ''])
+    assert pc.binary_join(ar_list, '-').equals(expected)
+
+    separator_array = pa.array(['1', '2'], type=pa.binary())
+    expected = pa.array(['a1b', 'c2d'], type=pa.binary())
+    ar_list = pa.array([['a', 'b'], ['c', 'd']], type=pa.list_(pa.binary()))
+    assert pc.binary_join(ar_list, separator_array).equals(expected)
+
+
 @pytest.mark.parametrize(('ty', 'values'), all_array_types)
 def test_take(ty, values):
     arr = pa.array(values, type=ty)

From 1ab419056d3ff2c24899c9cdf21784be2900e276 Mon Sep 17 00:00:00 2001
From: Antoine Pitrou <antoine@python.org>
Date: Thu, 3 Jun 2021 12:52:47 +0200
Subject: [PATCH 2/5] Updates and perf improvements

---
 cpp/src/arrow/array/builder_binary.h          |  32 ++-
 .../arrow/compute/kernels/scalar_string.cc    | 272 +++++++++++-------
 .../kernels/scalar_string_benchmark.cc        |  44 +++
 docs/source/cpp/compute.rst                   |   3 +-
 4 files changed, 240 insertions(+), 111 deletions(-)

diff --git a/cpp/src/arrow/array/builder_binary.h b/cpp/src/arrow/array/builder_binary.h
index 16ca1694b9f..197e74999d6 100644
--- a/cpp/src/arrow/array/builder_binary.h
+++ b/cpp/src/arrow/array/builder_binary.h
@@ -77,23 +77,21 @@ class BaseBinaryBuilder : public ArrayBuilder {
     return Append(value.data(), static_cast<offset_type>(value.size()));
   }
 
-  /// AppendCurrent does not add a new offset
-  Status AppendCurrent(const uint8_t* value, offset_type length) {
+  /// Append to the last appended value
+  ///
+  /// Unlike Append, this does not create a new offset.
+  Status AppendToCurrent(const uint8_t* value, offset_type length) {
     // Safety check for UBSAN.
     if (ARROW_PREDICT_TRUE(length > 0)) {
       ARROW_RETURN_NOT_OK(ValidateOverflow(length));
       ARROW_RETURN_NOT_OK(value_data_builder_.Append(value, length));
     }
-
     return Status::OK();
   }
 
-  Status AppendCurrent(const char* value, offset_type length) {
-    return AppendCurrent(reinterpret_cast<const uint8_t*>(value), length);
-  }
-
-  Status AppendCurrent(util::string_view value) {
-    return AppendCurrent(value.data(), static_cast<offset_type>(value.size()));
+  Status AppendToCurrent(util::string_view value) {
+    return AppendToCurrent(reinterpret_cast<const uint8_t*>(value.data()),
+                           static_cast<offset_type>(value.size()));
   }
 
   Status AppendNulls(int64_t length) final {
@@ -152,12 +150,28 @@ class BaseBinaryBuilder : public ArrayBuilder {
     UnsafeAppend(value.data(), static_cast<offset_type>(value.size()));
   }
 
+  /// Like AppendToCurrent, but do not check capacity
+  void UnsafeAppendToCurrent(const uint8_t* value, offset_type length) {
+    value_data_builder_.UnsafeAppend(value, length);
+  }
+
+  void UnsafeAppendToCurrent(util::string_view value) {
+    UnsafeAppendToCurrent(reinterpret_cast<const uint8_t*>(value.data()),
+                          static_cast<offset_type>(value.size()));
+  }
+
   void UnsafeAppendNull() {
     const int64_t num_bytes = value_data_builder_.length();
     offsets_builder_.UnsafeAppend(static_cast<offset_type>(num_bytes));
     UnsafeAppendToBitmap(false);
   }
 
+  void UnsafeAppendEmptyValue() {
+    const int64_t num_bytes = value_data_builder_.length();
+    offsets_builder_.UnsafeAppend(static_cast<offset_type>(num_bytes));
+    UnsafeAppendToBitmap(true);
+  }
+
   /// \brief Append a sequence of strings in one shot.
   ///
   /// \param[in] values a vector of strings
diff --git a/cpp/src/arrow/compute/kernels/scalar_string.cc b/cpp/src/arrow/compute/kernels/scalar_string.cc
index 96852c91f66..f0e2a886b77 100644
--- a/cpp/src/arrow/compute/kernels/scalar_string.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_string.cc
@@ -2637,8 +2637,6 @@ void AddUtf8Length(FunctionRegistry* registry) {
   DCHECK_OK(registry->AddFunction(std::move(func)));
 }
 
-// binary join
-
 template <typename Type>
 struct BinaryJoin {
   using ArrayType = typename TypeTraits<Type>::ArrayType;
@@ -2648,119 +2646,191 @@ struct BinaryJoin {
 
   static Status Exec(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
     if (batch[0].kind() == Datum::SCALAR) {
-      const ListScalar& list = checked_cast<const ListScalar&>(*batch[0].scalar());
-      if (!list.is_valid) {
-        return Status::OK();
-      }
       if (batch[1].kind() == Datum::SCALAR) {
-        const BaseBinaryScalar& separator_scalar =
-            checked_cast<const BaseBinaryScalar&>(*batch[1].scalar());
-        if (!separator_scalar.is_valid) {
-          return Status::OK();
-        }
-        util::string_view separator(*separator_scalar.value);
-
-        TypedBufferBuilder<uint8_t> builder(ctx->memory_pool());
-        auto Append = [&](util::string_view value) {
-          return builder.Append(reinterpret_cast<const uint8_t*>(value.data()),
-                                static_cast<offset_type>(value.size()));
-        };
-
-        const ArrayType* strings = static_cast<const ArrayType*>(list.value.get());
-        if (strings->null_count() > 0) {
-          // since the input list is not null, the out datum needs to be assigned to
-          *out = MakeNullScalar(list.value->type());
-          return Status::OK();
-        }
-        if (strings->length() > 0) {
-          RETURN_NOT_OK(Append(strings->GetView(0)));
-          for (int64_t j = 1; j < strings->length(); j++) {
-            RETURN_NOT_OK(Append(separator));
-            RETURN_NOT_OK(Append(strings->GetView(j)));
-          }
-        }
-        std::shared_ptr<Buffer> string_buffer;
-        RETURN_NOT_OK(builder.Finish(&string_buffer));
-        ARROW_ASSIGN_OR_RAISE(auto scalar_right_type,
-                              MakeScalar<std::shared_ptr<Buffer>>(
-                                  list.value->type(), std::move(string_buffer)));
-        *out = scalar_right_type;
+        return ExecScalarScalar(ctx, *batch[0].scalar(), *batch[1].scalar(), out);
       }
       // XXX do we want to support scalar[list[str]] with array[str] ?
     } else {
-      const ListArrayType list(batch[0].array());
-      ArrayData* output = out->mutable_array();
+      DCHECK_EQ(batch[0].kind(), Datum::ARRAY);
+      if (batch[1].kind() == Datum::SCALAR) {
+        return ExecArrayScalar(ctx, batch[0].array(), *batch[1].scalar(), out);
+      }
+      DCHECK_EQ(batch[1].kind(), Datum::ARRAY);
+      return ExecArrayArray(ctx, batch[0].array(), batch[1].array(), out);
+    }
+    return Status::OK();
+  }
 
-      BuilderType builder(ctx->memory_pool());
-      RETURN_NOT_OK(builder.Reserve(list.length()));
-      if (batch[1].kind() == Datum::ARRAY) {
-        ArrayType separator_array(batch[1].array());
-        for (int64_t i = 0; i < list.length(); ++i) {
-          const std::shared_ptr<Array> slice = list.value_slice(i);
-          const ArrayType* strings = static_cast<const ArrayType*>(slice.get());
-          if ((strings->null_count() > 0) || (list.IsNull(i)) ||
-              separator_array.IsNull(i)) {
-            RETURN_NOT_OK(builder.AppendNull());
-          } else {
-            const auto separator = separator_array.GetView(i);
-            if (strings->length() > 0) {
-              RETURN_NOT_OK(builder.Append(strings->GetView(0)));
-              for (int64_t j = 1; j < strings->length(); j++) {
-                RETURN_NOT_OK(builder.AppendCurrent(separator));
-                RETURN_NOT_OK(builder.AppendCurrent(strings->GetView(j)));
-              }
-            } else {
-              RETURN_NOT_OK(builder.AppendEmptyValue());
-            }
-          }
-        }
-      } else if (batch[1].kind() == Datum::SCALAR) {
-        const auto& separator_scalar =
-            checked_cast<const BaseBinaryScalar&>(*batch[1].scalar());
-        if (!separator_scalar.is_valid) {
-          ARROW_ASSIGN_OR_RAISE(
-              auto nulls,
-              MakeArrayOfNull(list.value_type(), list.length(), ctx->memory_pool()));
-          *output = *nulls->data();
-          output->type = list.value_type();
-          return Status::OK();
-        }
-        util::string_view separator(*separator_scalar.value);
-
-        for (int64_t i = 0; i < list.length(); ++i) {
-          const std::shared_ptr<Array> slice = list.value_slice(i);
-          const ArrayType* strings = static_cast<const ArrayType*>(slice.get());
-          if ((strings->null_count() > 0) || (list.IsNull(i))) {
-            RETURN_NOT_OK(builder.AppendNull());
-          } else {
-            if (strings->length() > 0) {
-              RETURN_NOT_OK(builder.Append(strings->GetView(0)));
-              for (int64_t j = 1; j < strings->length(); j++) {
-                RETURN_NOT_OK(builder.AppendCurrent(separator));
-                RETURN_NOT_OK(builder.AppendCurrent(strings->GetView(j)));
-              }
-            } else {
-              RETURN_NOT_OK(builder.AppendEmptyValue());
-            }
-          }
-        }
+  // Scalar, scalar -> scalar
+  static Status ExecScalarScalar(KernelContext* ctx, const Scalar& left,
+                                 const Scalar& right, Datum* out) {
+    const auto& list = checked_cast<const ListScalar&>(left);
+    const auto& separator_scalar = checked_cast<const BaseBinaryScalar&>(right);
+    if (!list.is_valid || !separator_scalar.is_valid) {
+      return Status::OK();
+    }
+    util::string_view separator(*separator_scalar.value);
+
+    TypedBufferBuilder<uint8_t> builder(ctx->memory_pool());
+    auto Append = [&](util::string_view value) {
+      return builder.Append(reinterpret_cast<const uint8_t*>(value.data()),
+                            static_cast<offset_type>(value.size()));
+    };
+
+    const auto& strings = checked_cast<const ArrayType&>(*list.value);
+    if (strings.null_count() > 0) {
+      // Since the input list is not null, the out datum needs to be assigned to
+      *out = MakeNullScalar(list.value->type());
+      return Status::OK();
+    }
+    if (strings.length() > 0) {
+      auto data_length =
+          strings.total_values_length() + (strings.length() - 1) * separator.length();
+      RETURN_NOT_OK(builder.Reserve(data_length));
+      RETURN_NOT_OK(Append(strings.GetView(0)));
+      for (int64_t j = 1; j < strings.length(); j++) {
+        RETURN_NOT_OK(Append(separator));
+        RETURN_NOT_OK(Append(strings.GetView(j)));
       }
-      std::shared_ptr<Array> string_array;
-      RETURN_NOT_OK(builder.Finish(&string_array));
-      *output = *string_array->data();
-      // correct the output type based on the input
-      output->type = list.value_type();
     }
+    std::shared_ptr<Buffer> string_buffer;
+    RETURN_NOT_OK(builder.Finish(&string_buffer));
+    ARROW_ASSIGN_OR_RAISE(auto joined, MakeScalar<std::shared_ptr<Buffer>>(
+                                           list.value->type(), std::move(string_buffer)));
+    *out = std::move(joined);
+    return Status::OK();
+  }
+
+  // Array, scalar -> array
+  static Status ExecArrayScalar(KernelContext* ctx,
+                                const std::shared_ptr<ArrayData>& left,
+                                const Scalar& right, Datum* out) {
+    const ListArrayType list(left);
+    const auto& separator_scalar = checked_cast<const BaseBinaryScalar&>(right);
+
+    if (!separator_scalar.is_valid) {
+      ARROW_ASSIGN_OR_RAISE(auto nulls, MakeArrayOfNull(list.value_type(), list.length(),
+                                                        ctx->memory_pool()));
+      *out = *nulls->data();
+      return Status::OK();
+    }
+
+    util::string_view separator(*separator_scalar.value);
+    const auto& strings = checked_cast<const ArrayType&>(*list.values());
+    const auto list_offsets = list.raw_value_offsets();
+
+    BuilderType builder(ctx->memory_pool());
+    RETURN_NOT_OK(builder.Reserve(list.length()));
+
+    // Presize data to avoid multiple reallocations when joining strings
+    int64_t total_data_length = strings.total_values_length();
+    for (int64_t i = 0; i < list.length(); ++i) {
+      const auto j_start = list_offsets[i], j_end = list_offsets[i + 1];
+      bool has_null_string = false;
+      for (int64_t j = j_start; !has_null_string && j < j_end; ++j) {
+        has_null_string = strings.IsNull(j);
+      }
+      if (!has_null_string && j_end > j_start) {
+        total_data_length += (j_end - j_start - 1) * separator.length();
+      }
+    }
+    RETURN_NOT_OK(builder.ReserveData(total_data_length));
+
+    struct SeparatorLookup {
+      const util::string_view separator;
+
+      bool IsNull(int64_t i) { return false; }
+      util::string_view GetView(int64_t i) { return separator; }
+    };
+    return JoinStrings(list, strings, SeparatorLookup{separator}, &builder, out);
+  }
+
+  // Array, array -> array
+  static Status ExecArrayArray(KernelContext* ctx, const std::shared_ptr<ArrayData>& left,
+                               const std::shared_ptr<ArrayData>& right, Datum* out) {
+    const ListArrayType list(left);
+    const auto& strings = checked_cast<const ArrayType&>(*list.values());
+    const auto list_offsets = list.raw_value_offsets();
+    const auto string_offsets = strings.raw_value_offsets();
+    const ArrayType separators(right);
+
+    BuilderType builder(ctx->memory_pool());
+    RETURN_NOT_OK(builder.Reserve(list.length()));
+
+    // Presize data to avoid multiple reallocations when joining strings
+    int64_t total_data_length = 0;
+    for (int64_t i = 0; i < list.length(); ++i) {
+      if (separators.IsNull(i)) {
+        continue;
+      }
+      const auto j_start = list_offsets[i], j_end = list_offsets[i + 1];
+      bool has_null_string = false;
+      for (int64_t j = j_start; !has_null_string && j < j_end; ++j) {
+        has_null_string = strings.IsNull(j);
+      }
+      if (!has_null_string && j_end > j_start) {
+        total_data_length += string_offsets[j_end] - string_offsets[j_start];
+        total_data_length += (j_end - j_start - 1) * separators.value_length(i);
+      }
+    }
+    RETURN_NOT_OK(builder.ReserveData(total_data_length));
+
+    struct SeparatorLookup {
+      const ArrayType& separators;
+
+      bool IsNull(int64_t i) { return separators.IsNull(i); }
+      util::string_view GetView(int64_t i) { return separators.GetView(i); }
+    };
+    return JoinStrings(list, strings, SeparatorLookup{separators}, &builder, out);
+  }
+
+  template <typename SeparatorLookup>
+  static Status JoinStrings(const ListArrayType& list, const ArrayType& strings,
+                            SeparatorLookup&& separators, BuilderType* builder,
+                            Datum* out) {
+    const auto list_offsets = list.raw_value_offsets();
+
+    for (int64_t i = 0; i < list.length(); ++i) {
+      if (list.IsNull(i) || separators.IsNull(i)) {
+        builder->UnsafeAppendNull();
+        continue;
+      }
+      const auto j_start = list_offsets[i], j_end = list_offsets[i + 1];
+      if (j_start == j_end) {
+        builder->UnsafeAppendEmptyValue();
+        continue;
+      }
+      bool has_null_string = false;
+      for (int64_t j = j_start; !has_null_string && j < j_end; ++j) {
+        has_null_string = strings.IsNull(j);
+      }
+      if (has_null_string) {
+        builder->UnsafeAppendNull();
+        continue;
+      }
+      builder->UnsafeAppend(strings.GetView(j_start));
+      for (int64_t j = j_start + 1; j < j_end; ++j) {
+        builder->UnsafeAppendToCurrent(separators.GetView(i));
+        builder->UnsafeAppendToCurrent(strings.GetView(j));
+      }
+    }
+
+    std::shared_ptr<Array> string_array;
+    RETURN_NOT_OK(builder->Finish(&string_array));
+    *out = *string_array->data();
+    // Correct the output type based on the input
+    out->mutable_array()->type = list.value_type();
     return Status::OK();
   }
 };
 
 const FunctionDoc binary_join_doc(
     "Join a list of strings together with a `separator` to form a single string",
-    ("Insert `separator` between each list element, and concatenate them."),
+    ("Insert `separator` between `list` elements, and concatenate them.\n"
+     "Any null input and any null `list` element emits a null output.\n"),
     {"list", "separator"});
 
-void AddJoin(FunctionRegistry* registry) {
+void AddBinaryJoin(FunctionRegistry* registry) {
   auto func =
       std::make_shared<ScalarFunction>("binary_join", Arity::Binary(), &binary_join_doc);
   for (const std::shared_ptr<DataType>& ty : BaseBinaryTypes()) {
@@ -3081,7 +3151,7 @@ void RegisterScalarStringAscii(FunctionRegistry* registry) {
   AddSlice(registry);
   AddSplit(registry);
   AddStrptime(registry);
-  AddJoin(registry);
+  AddBinaryJoin(registry);
 }
 
 }  // namespace internal
diff --git a/cpp/src/arrow/compute/kernels/scalar_string_benchmark.cc b/cpp/src/arrow/compute/kernels/scalar_string_benchmark.cc
index 32ac5a7df3b..8528c0d9e5d 100644
--- a/cpp/src/arrow/compute/kernels/scalar_string_benchmark.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_string_benchmark.cc
@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <functional>
+
 #include "benchmark/benchmark.h"
 
 #include "arrow/compute/api_scalar.h"
@@ -22,8 +24,12 @@
 #include "arrow/testing/gtest_util.h"
 #include "arrow/testing/random.h"
 #include "arrow/util/benchmark_util.h"
+#include "arrow/util/checked_cast.h"
 
 namespace arrow {
+
+using internal::checked_cast;
+
 namespace compute {
 
 constexpr auto kSeed = 0x94378165;
@@ -104,6 +110,41 @@ static void TrimManyUtf8(benchmark::State& state) {
 }
 #endif
 
+using SeparatorFactory = std::function<Datum(int64_t n, double null_probability)>;
+
+static void BinaryJoin(benchmark::State& state, SeparatorFactory make_separator) {
+  const int64_t n_strings = 10000;
+  const int64_t n_lists = 1000;
+  const double null_probability = 0.02;
+
+  random::RandomArrayGenerator rng(kSeed);
+
+  auto strings =
+      rng.String(n_strings, /*min_length=*/5, /*max_length=*/20, null_probability);
+  auto lists = rng.List(*strings, n_lists, null_probability, /*force_empty_nulls=*/true);
+  auto separator = make_separator(n_lists, null_probability);
+
+  for (auto _ : state) {
+    ABORT_NOT_OK(CallFunction("binary_join", {lists, separator}));
+  }
+  state.SetBytesProcessed(
+      state.iterations() *
+      checked_cast<const StringArray&>(*strings).total_values_length());
+}
+
+static void BinaryJoinArrayScalar(benchmark::State& state) {
+  BinaryJoin(state, [](int64_t n, double null_probability) -> Datum {
+    return ScalarFromJSON(utf8(), R"("--")");
+  });
+}
+
+static void BinaryJoinArrayArray(benchmark::State& state) {
+  BinaryJoin(state, [](int64_t n, double null_probability) -> Datum {
+    random::RandomArrayGenerator rng(kSeed + 1);
+    return rng.String(n, /*min_length=*/0, /*max_length=*/4, null_probability);
+  });
+}
+
 BENCHMARK(AsciiLower);
 BENCHMARK(AsciiUpper);
 BENCHMARK(IsAlphaNumericAscii);
@@ -119,5 +160,8 @@ BENCHMARK(TrimSingleUtf8);
 BENCHMARK(TrimManyUtf8);
 #endif
 
+BENCHMARK(BinaryJoinArrayScalar);
+BENCHMARK(BinaryJoinArrayArray);
+
 }  // namespace compute
 }  // namespace arrow
diff --git a/docs/source/cpp/compute.rst b/docs/source/cpp/compute.rst
index 94bb0720acd..1508ee5d2b5 100644
--- a/docs/source/cpp/compute.rst
+++ b/docs/source/cpp/compute.rst
@@ -666,7 +666,8 @@ This function does the inverse of string splitting.
 
 * \(1) The first input must be an array, while the second can be a scalar or array.
   Each list of values in the first input is joined using each second input
-  as separator.
+  as separator.  If there is null element in an input list, the corresponding
+  output will be null.
 
 
 Slicing

From b52f8cdf0bd51f9afe5fb57e777cbac0ce2d1cc1 Mon Sep 17 00:00:00 2001
From: Antoine Pitrou <antoine@python.org>
Date: Thu, 3 Jun 2021 13:45:35 +0200
Subject: [PATCH 3/5] Add LargeList support, test AppendToCurrent

---
 cpp/src/arrow/array/array_binary_test.cc      | 70 +++++++++++++++++++
 .../arrow/compute/kernels/codegen_internal.h  |  6 +-
 .../arrow/compute/kernels/scalar_string.cc    | 36 ++++++----
 .../compute/kernels/scalar_string_test.cc     | 25 ++++---
 4 files changed, 110 insertions(+), 27 deletions(-)

diff --git a/cpp/src/arrow/array/array_binary_test.cc b/cpp/src/arrow/array/array_binary_test.cc
index 5c247a6dc66..add6b8e04b4 100644
--- a/cpp/src/arrow/array/array_binary_test.cc
+++ b/cpp/src/arrow/array/array_binary_test.cc
@@ -473,6 +473,70 @@ class TestStringBuilder : public TestBuilder {
     CheckStringArray(*result_, strings, is_valid, reps);
   }
 
+  void TestAppendToCurrent() {
+    std::vector<std::string> strings = {"", "bbbb", "aaaaa", "", "ccc"};
+    std::vector<uint8_t> is_valid = {1, 1, 1, 0, 1};
+
+    int N = static_cast<int>(strings.size());
+    int reps = 10;
+
+    for (int j = 0; j < reps; ++j) {
+      for (int i = 0; i < N; ++i) {
+        if (!is_valid[i]) {
+          ASSERT_OK(builder_->AppendNull());
+        } else if (strings[i].length() > 3) {
+          ASSERT_OK(builder_->Append(strings[i].substr(0, 3)));
+          ASSERT_OK(builder_->AppendToCurrent(strings[i].substr(3)));
+        } else {
+          ASSERT_OK(builder_->Append(strings[i]));
+        }
+      }
+    }
+    Done();
+
+    ASSERT_EQ(reps * N, result_->length());
+    ASSERT_EQ(reps, result_->null_count());
+    ASSERT_EQ(reps * 12, result_->value_data()->size());
+
+    CheckStringArray(*result_, strings, is_valid, reps);
+  }
+
+  void TestAppendToCurrentUnsafe() {
+    std::vector<std::string> strings = {"", "bbbb", "aaaaa", "", "ccc"};
+    std::vector<uint8_t> is_valid = {1, 1, 1, 0, 1};
+
+    int N = static_cast<int>(strings.size());
+    int reps = 13;
+    int64_t total_length = 0;
+    for (const auto& s : strings) {
+      total_length += static_cast<int64_t>(s.size());
+    }
+
+    ASSERT_OK(builder_->Reserve(N * reps));
+    ASSERT_OK(builder_->ReserveData(total_length * reps));
+
+    for (int j = 0; j < reps; ++j) {
+      for (int i = 0; i < N; ++i) {
+        if (!is_valid[i]) {
+          builder_->UnsafeAppendNull();
+        } else if (strings[i].length() > 3) {
+          builder_->UnsafeAppend(strings[i].substr(0, 3));
+          builder_->UnsafeAppendToCurrent(strings[i].substr(3));
+        } else {
+          builder_->UnsafeAppend(strings[i]);
+        }
+      }
+    }
+    ASSERT_EQ(builder_->value_data_length(), total_length * reps);
+    Done();
+
+    ASSERT_EQ(reps * N, result_->length());
+    ASSERT_EQ(reps, result_->null_count());
+    ASSERT_EQ(reps * 12, result_->value_data()->size());
+
+    CheckStringArray(*result_, strings, is_valid, reps);
+  }
+
   void TestVectorAppend() {
     std::vector<std::string> strings = {"", "bb", "a", "", "ccc"};
     std::vector<uint8_t> valid_bytes = {1, 1, 1, 0, 1};
@@ -608,6 +672,12 @@ TYPED_TEST(TestStringBuilder, TestScalarAppend) { this->TestScalarAppend(); }
 
 TYPED_TEST(TestStringBuilder, TestScalarAppendUnsafe) { this->TestScalarAppendUnsafe(); }
 
+TYPED_TEST(TestStringBuilder, TestAppendToCurrent) { this->TestAppendToCurrent(); }
+
+TYPED_TEST(TestStringBuilder, TestAppendToCurrentUnsafe) {
+  this->TestAppendToCurrentUnsafe();
+}
+
 TYPED_TEST(TestStringBuilder, TestVectorAppend) { this->TestVectorAppend(); }
 
 TYPED_TEST(TestStringBuilder, TestAppendCStringsWithValidBytes) {
diff --git a/cpp/src/arrow/compute/kernels/codegen_internal.h b/cpp/src/arrow/compute/kernels/codegen_internal.h
index 6d5c837f514..913c4dacf56 100644
--- a/cpp/src/arrow/compute/kernels/codegen_internal.h
+++ b/cpp/src/arrow/compute/kernels/codegen_internal.h
@@ -1193,15 +1193,15 @@ ArrayKernelExec GenerateTypeAgnosticPrimitive(detail::GetTypeId get_id) {
 }
 
 // similar to GenerateTypeAgnosticPrimitive, but for variable types
-template <template <typename...> class Generator>
+template <template <typename...> class Generator, typename... Args>
 ArrayKernelExec GenerateTypeAgnosticVarBinaryBase(detail::GetTypeId get_id) {
   switch (get_id.id) {
     case Type::BINARY:
     case Type::STRING:
-      return Generator<BinaryType>::Exec;
+      return Generator<BinaryType, Args...>::Exec;
     case Type::LARGE_BINARY:
     case Type::LARGE_STRING:
-      return Generator<LargeBinaryType>::Exec;
+      return Generator<LargeBinaryType, Args...>::Exec;
     default:
       DCHECK(false);
       return ExecFail;
diff --git a/cpp/src/arrow/compute/kernels/scalar_string.cc b/cpp/src/arrow/compute/kernels/scalar_string.cc
index f0e2a886b77..8450e729818 100644
--- a/cpp/src/arrow/compute/kernels/scalar_string.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_string.cc
@@ -2637,12 +2637,12 @@ void AddUtf8Length(FunctionRegistry* registry) {
   DCHECK_OK(registry->AddFunction(std::move(func)));
 }
 
-template <typename Type>
+template <typename BinaryType, typename ListType>
 struct BinaryJoin {
-  using ArrayType = typename TypeTraits<Type>::ArrayType;
-  using ListArrayType = ListArray;
-  using offset_type = typename Type::offset_type;
-  using BuilderType = typename TypeTraits<Type>::BuilderType;
+  using ArrayType = typename TypeTraits<BinaryType>::ArrayType;
+  using ListArrayType = typename TypeTraits<ListType>::ArrayType;
+  using ListScalarType = typename TypeTraits<ListType>::ScalarType;
+  using BuilderType = typename TypeTraits<BinaryType>::BuilderType;
 
   static Status Exec(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
     if (batch[0].kind() == Datum::SCALAR) {
@@ -2664,7 +2664,7 @@ struct BinaryJoin {
   // Scalar, scalar -> scalar
   static Status ExecScalarScalar(KernelContext* ctx, const Scalar& left,
                                  const Scalar& right, Datum* out) {
-    const auto& list = checked_cast<const ListScalar&>(left);
+    const auto& list = checked_cast<const ListScalarType&>(left);
     const auto& separator_scalar = checked_cast<const BaseBinaryScalar&>(right);
     if (!list.is_valid || !separator_scalar.is_valid) {
       return Status::OK();
@@ -2674,7 +2674,7 @@ struct BinaryJoin {
     TypedBufferBuilder<uint8_t> builder(ctx->memory_pool());
     auto Append = [&](util::string_view value) {
       return builder.Append(reinterpret_cast<const uint8_t*>(value.data()),
-                            static_cast<offset_type>(value.size()));
+                            static_cast<int64_t>(value.size()));
     };
 
     const auto& strings = checked_cast<const ArrayType&>(*list.value);
@@ -2830,19 +2830,25 @@ const FunctionDoc binary_join_doc(
      "Any null input and any null `list` element emits a null output.\n"),
     {"list", "separator"});
 
-void AddBinaryJoin(FunctionRegistry* registry) {
-  auto func =
-      std::make_shared<ScalarFunction>("binary_join", Arity::Binary(), &binary_join_doc);
+template <typename ListType>
+void AddBinaryJoinForListType(ScalarFunction* func) {
   for (const std::shared_ptr<DataType>& ty : BaseBinaryTypes()) {
-    auto exec = GenerateTypeAgnosticVarBinaryBase<BinaryJoin>(*ty);
-    // TODO add large_list inputs
+    auto exec = GenerateTypeAgnosticVarBinaryBase<BinaryJoin, ListType>(*ty);
+    auto list_ty = std::make_shared<ListType>(ty);
     DCHECK_OK(
-        func->AddKernel({InputType::Array(list(ty)), InputType::Scalar(ty)}, ty, exec));
+        func->AddKernel({InputType::Array(list_ty), InputType::Scalar(ty)}, ty, exec));
     DCHECK_OK(
-        func->AddKernel({InputType::Array(list(ty)), InputType::Array(ty)}, ty, exec));
+        func->AddKernel({InputType::Array(list_ty), InputType::Array(ty)}, ty, exec));
     DCHECK_OK(
-        func->AddKernel({InputType::Scalar(list(ty)), InputType::Scalar(ty)}, ty, exec));
+        func->AddKernel({InputType::Scalar(list_ty), InputType::Scalar(ty)}, ty, exec));
   }
+}
+
+void AddBinaryJoin(FunctionRegistry* registry) {
+  auto func =
+      std::make_shared<ScalarFunction>("binary_join", Arity::Binary(), &binary_join_doc);
+  AddBinaryJoinForListType<ListType>(func.get());
+  AddBinaryJoinForListType<LargeListType>(func.get());
   DCHECK_OK(registry->AddFunction(std::move(func)));
 }
 
diff --git a/cpp/src/arrow/compute/kernels/scalar_string_test.cc b/cpp/src/arrow/compute/kernels/scalar_string_test.cc
index 16c231e5ce1..34750bd02e4 100644
--- a/cpp/src/arrow/compute/kernels/scalar_string_test.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_string_test.cc
@@ -783,25 +783,32 @@ TYPED_TEST(TestStringKernels, StrptimeDoesNotProvideDefaultOptions) {
 
 TYPED_TEST(TestStringKernels, BinaryJoin) {
   auto separator = this->scalar("--");
-  auto list_input = ArrayFromJSON(
-      list(this->type()),
-      R"([["a", "bb", "ccc"], [], null, ["dd"], ["eee", null], ["ff", ""]])");
+  std::string list_json =
+      R"([["a", "bb", "ccc"], [], null, ["dd"], ["eee", null], ["ff", ""]])";
   auto expected =
       ArrayFromJSON(this->type(), R"(["a--bb--ccc", "", null, "dd", null, "ff--"])");
-  CheckScalarBinary("binary_join", list_input, separator, expected);
+  CheckScalarBinary("binary_join", ArrayFromJSON(list(this->type()), list_json),
+                    separator, expected);
+  CheckScalarBinary("binary_join", ArrayFromJSON(large_list(this->type()), list_json),
+                    separator, expected);
 
   auto separator_null = MakeNullScalar(this->type());
   expected = ArrayFromJSON(this->type(), R"([null, null, null, null, null, null])");
-  CheckScalarBinary("binary_join", list_input, separator_null, expected);
+  CheckScalarBinary("binary_join", ArrayFromJSON(list(this->type()), list_json),
+                    separator_null, expected);
+  CheckScalarBinary("binary_join", ArrayFromJSON(large_list(this->type()), list_json),
+                    separator_null, expected);
 
   auto separators =
       ArrayFromJSON(this->type(), R"(["1", "2", "3", "4", "5", "6", null])");
-  list_input = ArrayFromJSON(
-      list(this->type()),
-      R"([["a", "bb", "ccc"], [], null, ["dd"], ["eee", null], ["ff", ""], ["hh", "ii"]])");
+  list_json =
+      R"([["a", "bb", "ccc"], [], null, ["dd"], ["eee", null], ["ff", ""], ["hh", "ii"]])";
   expected =
       ArrayFromJSON(this->type(), R"(["a1bb1ccc", "", null, "dd", null, "ff6", null])");
-  CheckScalarBinary("binary_join", list_input, separators, expected);
+  CheckScalarBinary("binary_join", ArrayFromJSON(list(this->type()), list_json),
+                    separators, expected);
+  CheckScalarBinary("binary_join", ArrayFromJSON(large_list(this->type()), list_json),
+                    separators, expected);
 }
 
 #ifdef ARROW_WITH_UTF8PROC

From 0f8034695eed0e067e8088ee69c9ee2765a6d22e Mon Sep 17 00:00:00 2001
From: Antoine Pitrou <antoine@python.org>
Date: Thu, 3 Jun 2021 19:58:11 +0200
Subject: [PATCH 4/5] Address review comments

---
 .../arrow/compute/kernels/scalar_string.cc    | 216 +++++++++++-------
 .../compute/kernels/scalar_string_test.cc     |  17 ++
 cpp/src/arrow/compute/kernels/test_util.cc    |   6 +
 cpp/src/arrow/compute/kernels/test_util.h     |   5 +
 docs/source/cpp/compute.rst                   |   2 +-
 5 files changed, 168 insertions(+), 78 deletions(-)

diff --git a/cpp/src/arrow/compute/kernels/scalar_string.cc b/cpp/src/arrow/compute/kernels/scalar_string.cc
index 8450e729818..73a1d47bb47 100644
--- a/cpp/src/arrow/compute/kernels/scalar_string.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_string.cc
@@ -2642,6 +2642,7 @@ struct BinaryJoin {
   using ArrayType = typename TypeTraits<BinaryType>::ArrayType;
   using ListArrayType = typename TypeTraits<ListType>::ArrayType;
   using ListScalarType = typename TypeTraits<ListType>::ScalarType;
+  using ListOffsetType = typename ListArrayType::offset_type;
   using BuilderType = typename TypeTraits<BinaryType>::BuilderType;
 
   static Status Exec(KernelContext* ctx, const ExecBatch& batch, Datum* out) {
@@ -2649,18 +2650,52 @@ struct BinaryJoin {
       if (batch[1].kind() == Datum::SCALAR) {
         return ExecScalarScalar(ctx, *batch[0].scalar(), *batch[1].scalar(), out);
       }
-      // XXX do we want to support scalar[list[str]] with array[str] ?
-    } else {
-      DCHECK_EQ(batch[0].kind(), Datum::ARRAY);
-      if (batch[1].kind() == Datum::SCALAR) {
-        return ExecArrayScalar(ctx, batch[0].array(), *batch[1].scalar(), out);
-      }
       DCHECK_EQ(batch[1].kind(), Datum::ARRAY);
-      return ExecArrayArray(ctx, batch[0].array(), batch[1].array(), out);
+      return ExecScalarArray(ctx, *batch[0].scalar(), batch[1].array(), out);
     }
-    return Status::OK();
+    DCHECK_EQ(batch[0].kind(), Datum::ARRAY);
+    if (batch[1].kind() == Datum::SCALAR) {
+      return ExecArrayScalar(ctx, batch[0].array(), *batch[1].scalar(), out);
+    }
+    DCHECK_EQ(batch[1].kind(), Datum::ARRAY);
+    return ExecArrayArray(ctx, batch[0].array(), batch[1].array(), out);
   }
 
+  struct ListScalarOffsetLookup {
+    const ArrayType& values;
+
+    int64_t GetStart(int64_t i) { return 0; }
+    int64_t GetStop(int64_t i) { return values.length(); }
+    bool IsNull(int64_t i) { return false; }
+  };
+
+  struct ListArrayOffsetLookup {
+    explicit ListArrayOffsetLookup(const ListArrayType& lists)
+        : lists_(lists), offsets_(lists.raw_value_offsets()) {}
+
+    int64_t GetStart(int64_t i) { return offsets_[i]; }
+    int64_t GetStop(int64_t i) { return offsets_[i + 1]; }
+    bool IsNull(int64_t i) { return lists_.IsNull(i); }
+
+   private:
+    const ListArrayType& lists_;
+    const ListOffsetType* offsets_;
+  };
+
+  struct SeparatorScalarLookup {
+    const util::string_view separator;
+
+    bool IsNull(int64_t i) { return false; }
+    util::string_view GetView(int64_t i) { return separator; }
+  };
+
+  struct SeparatorArrayLookup {
+    const ArrayType& separators;
+
+    bool IsNull(int64_t i) { return separators.IsNull(i); }
+    util::string_view GetView(int64_t i) { return separators.GetView(i); }
+  };
+
   // Scalar, scalar -> scalar
   static Status ExecScalarScalar(KernelContext* ctx, const Scalar& left,
                                  const Scalar& right, Datum* out) {
@@ -2671,18 +2706,17 @@ struct BinaryJoin {
     }
     util::string_view separator(*separator_scalar.value);
 
+    const auto& strings = checked_cast<const ArrayType&>(*list.value);
+    if (strings.null_count() > 0) {
+      out->scalar()->is_valid = false;
+      return Status::OK();
+    }
+
     TypedBufferBuilder<uint8_t> builder(ctx->memory_pool());
     auto Append = [&](util::string_view value) {
       return builder.Append(reinterpret_cast<const uint8_t*>(value.data()),
                             static_cast<int64_t>(value.size()));
     };
-
-    const auto& strings = checked_cast<const ArrayType&>(*list.value);
-    if (strings.null_count() > 0) {
-      // Since the input list is not null, the out datum needs to be assigned to
-      *out = MakeNullScalar(list.value->type());
-      return Status::OK();
-    }
     if (strings.length() > 0) {
       auto data_length =
           strings.total_values_length() + (strings.length() - 1) * separator.length();
@@ -2693,84 +2727,110 @@ struct BinaryJoin {
         RETURN_NOT_OK(Append(strings.GetView(j)));
       }
     }
-    std::shared_ptr<Buffer> string_buffer;
-    RETURN_NOT_OK(builder.Finish(&string_buffer));
-    ARROW_ASSIGN_OR_RAISE(auto joined, MakeScalar<std::shared_ptr<Buffer>>(
-                                           list.value->type(), std::move(string_buffer)));
-    *out = std::move(joined);
-    return Status::OK();
+    auto out_scalar = checked_cast<BaseBinaryScalar*>(out->scalar().get());
+    return builder.Finish(&out_scalar->value);
+  }
+
+  // Scalar, array -> array
+  static Status ExecScalarArray(KernelContext* ctx, const Scalar& left,
+                                const std::shared_ptr<ArrayData>& right, Datum* out) {
+    const auto& list_scalar = checked_cast<const BaseListScalar&>(left);
+    if (!list_scalar.is_valid) {
+      ARROW_ASSIGN_OR_RAISE(
+          auto nulls, MakeArrayOfNull(right->type, right->length, ctx->memory_pool()));
+      *out = *nulls->data();
+      return Status::OK();
+    }
+    const auto& strings = checked_cast<const ArrayType&>(*list_scalar.value);
+    if (strings.null_count() != 0) {
+      ARROW_ASSIGN_OR_RAISE(
+          auto nulls, MakeArrayOfNull(right->type, right->length, ctx->memory_pool()));
+      *out = *nulls->data();
+      return Status::OK();
+    }
+    const ArrayType separators(right);
+
+    BuilderType builder(ctx->memory_pool());
+    RETURN_NOT_OK(builder.Reserve(separators.length()));
+
+    // Presize data to avoid multiple reallocations when joining strings
+    int64_t total_data_length = 0;
+    const int64_t list_length = strings.length();
+    if (list_length) {
+      const int64_t string_length = strings.total_values_length();
+      total_data_length +=
+          string_length * (separators.length() - separators.null_count());
+      for (int64_t i = 0; i < separators.length(); ++i) {
+        if (separators.IsNull(i)) {
+          continue;
+        }
+        total_data_length += (list_length - 1) * separators.value_length(i);
+      }
+    }
+    RETURN_NOT_OK(builder.ReserveData(total_data_length));
+
+    return JoinStrings(separators.length(), strings, ListScalarOffsetLookup{strings},
+                       SeparatorArrayLookup{separators}, &builder, out);
   }
 
   // Array, scalar -> array
   static Status ExecArrayScalar(KernelContext* ctx,
                                 const std::shared_ptr<ArrayData>& left,
                                 const Scalar& right, Datum* out) {
-    const ListArrayType list(left);
+    const ListArrayType lists(left);
     const auto& separator_scalar = checked_cast<const BaseBinaryScalar&>(right);
 
     if (!separator_scalar.is_valid) {
-      ARROW_ASSIGN_OR_RAISE(auto nulls, MakeArrayOfNull(list.value_type(), list.length(),
-                                                        ctx->memory_pool()));
+      ARROW_ASSIGN_OR_RAISE(
+          auto nulls,
+          MakeArrayOfNull(lists.value_type(), lists.length(), ctx->memory_pool()));
       *out = *nulls->data();
       return Status::OK();
     }
 
     util::string_view separator(*separator_scalar.value);
-    const auto& strings = checked_cast<const ArrayType&>(*list.values());
-    const auto list_offsets = list.raw_value_offsets();
+    const auto& strings = checked_cast<const ArrayType&>(*lists.values());
+    const auto list_offsets = lists.raw_value_offsets();
 
     BuilderType builder(ctx->memory_pool());
-    RETURN_NOT_OK(builder.Reserve(list.length()));
+    RETURN_NOT_OK(builder.Reserve(lists.length()));
 
     // Presize data to avoid multiple reallocations when joining strings
     int64_t total_data_length = strings.total_values_length();
-    for (int64_t i = 0; i < list.length(); ++i) {
-      const auto j_start = list_offsets[i], j_end = list_offsets[i + 1];
-      bool has_null_string = false;
-      for (int64_t j = j_start; !has_null_string && j < j_end; ++j) {
-        has_null_string = strings.IsNull(j);
-      }
-      if (!has_null_string && j_end > j_start) {
-        total_data_length += (j_end - j_start - 1) * separator.length();
+    for (int64_t i = 0; i < lists.length(); ++i) {
+      const auto start = list_offsets[i], end = list_offsets[i + 1];
+      if (end > start && !ValuesContainNull(strings, start, end)) {
+        total_data_length += (end - start - 1) * separator.length();
       }
     }
     RETURN_NOT_OK(builder.ReserveData(total_data_length));
 
-    struct SeparatorLookup {
-      const util::string_view separator;
-
-      bool IsNull(int64_t i) { return false; }
-      util::string_view GetView(int64_t i) { return separator; }
-    };
-    return JoinStrings(list, strings, SeparatorLookup{separator}, &builder, out);
+    return JoinStrings(lists.length(), strings, ListArrayOffsetLookup{lists},
+                       SeparatorScalarLookup{separator}, &builder, out);
   }
 
   // Array, array -> array
   static Status ExecArrayArray(KernelContext* ctx, const std::shared_ptr<ArrayData>& left,
                                const std::shared_ptr<ArrayData>& right, Datum* out) {
-    const ListArrayType list(left);
-    const auto& strings = checked_cast<const ArrayType&>(*list.values());
-    const auto list_offsets = list.raw_value_offsets();
+    const ListArrayType lists(left);
+    const auto& strings = checked_cast<const ArrayType&>(*lists.values());
+    const auto list_offsets = lists.raw_value_offsets();
     const auto string_offsets = strings.raw_value_offsets();
     const ArrayType separators(right);
 
     BuilderType builder(ctx->memory_pool());
-    RETURN_NOT_OK(builder.Reserve(list.length()));
+    RETURN_NOT_OK(builder.Reserve(lists.length()));
 
     // Presize data to avoid multiple reallocations when joining strings
     int64_t total_data_length = 0;
-    for (int64_t i = 0; i < list.length(); ++i) {
+    for (int64_t i = 0; i < lists.length(); ++i) {
       if (separators.IsNull(i)) {
         continue;
       }
-      const auto j_start = list_offsets[i], j_end = list_offsets[i + 1];
-      bool has_null_string = false;
-      for (int64_t j = j_start; !has_null_string && j < j_end; ++j) {
-        has_null_string = strings.IsNull(j);
-      }
-      if (!has_null_string && j_end > j_start) {
-        total_data_length += string_offsets[j_end] - string_offsets[j_start];
-        total_data_length += (j_end - j_start - 1) * separators.value_length(i);
+      const auto start = list_offsets[i], end = list_offsets[i + 1];
+      if (end > start && !ValuesContainNull(strings, start, end)) {
+        total_data_length += string_offsets[end] - string_offsets[start];
+        total_data_length += (end - start - 1) * separators.value_length(i);
       }
     }
     RETURN_NOT_OK(builder.ReserveData(total_data_length));
@@ -2781,30 +2841,25 @@ struct BinaryJoin {
       bool IsNull(int64_t i) { return separators.IsNull(i); }
       util::string_view GetView(int64_t i) { return separators.GetView(i); }
     };
-    return JoinStrings(list, strings, SeparatorLookup{separators}, &builder, out);
+    return JoinStrings(lists.length(), strings, ListArrayOffsetLookup{lists},
+                       SeparatorArrayLookup{separators}, &builder, out);
   }
 
-  template <typename SeparatorLookup>
-  static Status JoinStrings(const ListArrayType& list, const ArrayType& strings,
-                            SeparatorLookup&& separators, BuilderType* builder,
-                            Datum* out) {
-    const auto list_offsets = list.raw_value_offsets();
-
-    for (int64_t i = 0; i < list.length(); ++i) {
-      if (list.IsNull(i) || separators.IsNull(i)) {
+  template <typename ListOffsetLookup, typename SeparatorLookup>
+  static Status JoinStrings(int64_t length, const ArrayType& strings,
+                            ListOffsetLookup&& list_offsets, SeparatorLookup&& separators,
+                            BuilderType* builder, Datum* out) {
+    for (int64_t i = 0; i < length; ++i) {
+      if (list_offsets.IsNull(i) || separators.IsNull(i)) {
         builder->UnsafeAppendNull();
         continue;
       }
-      const auto j_start = list_offsets[i], j_end = list_offsets[i + 1];
+      const auto j_start = list_offsets.GetStart(i), j_end = list_offsets.GetStop(i);
       if (j_start == j_end) {
         builder->UnsafeAppendEmptyValue();
         continue;
       }
-      bool has_null_string = false;
-      for (int64_t j = j_start; !has_null_string && j < j_end; ++j) {
-        has_null_string = strings.IsNull(j);
-      }
-      if (has_null_string) {
+      if (ValuesContainNull(strings, j_start, j_end)) {
         builder->UnsafeAppendNull();
         continue;
       }
@@ -2819,9 +2874,21 @@ struct BinaryJoin {
     RETURN_NOT_OK(builder->Finish(&string_array));
     *out = *string_array->data();
     // Correct the output type based on the input
-    out->mutable_array()->type = list.value_type();
+    out->mutable_array()->type = strings.type();
     return Status::OK();
   }
+
+  static bool ValuesContainNull(const ArrayType& values, int64_t start, int64_t end) {
+    if (values.null_count() == 0) {
+      return false;
+    }
+    for (int64_t i = start; i < end; ++i) {
+      if (values.IsNull(i)) {
+        return true;
+      }
+    }
+    return false;
+  }
 };
 
 const FunctionDoc binary_join_doc(
@@ -2835,12 +2902,7 @@ void AddBinaryJoinForListType(ScalarFunction* func) {
   for (const std::shared_ptr<DataType>& ty : BaseBinaryTypes()) {
     auto exec = GenerateTypeAgnosticVarBinaryBase<BinaryJoin, ListType>(*ty);
     auto list_ty = std::make_shared<ListType>(ty);
-    DCHECK_OK(
-        func->AddKernel({InputType::Array(list_ty), InputType::Scalar(ty)}, ty, exec));
-    DCHECK_OK(
-        func->AddKernel({InputType::Array(list_ty), InputType::Array(ty)}, ty, exec));
-    DCHECK_OK(
-        func->AddKernel({InputType::Scalar(list_ty), InputType::Scalar(ty)}, ty, exec));
+    DCHECK_OK(func->AddKernel({InputType(list_ty), InputType(ty)}, ty, exec));
   }
 }
 
diff --git a/cpp/src/arrow/compute/kernels/scalar_string_test.cc b/cpp/src/arrow/compute/kernels/scalar_string_test.cc
index 34750bd02e4..a8dd9e0c9fc 100644
--- a/cpp/src/arrow/compute/kernels/scalar_string_test.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_string_test.cc
@@ -782,6 +782,7 @@ TYPED_TEST(TestStringKernels, StrptimeDoesNotProvideDefaultOptions) {
 }
 
 TYPED_TEST(TestStringKernels, BinaryJoin) {
+  // Scalar separator
   auto separator = this->scalar("--");
   std::string list_json =
       R"([["a", "bb", "ccc"], [], null, ["dd"], ["eee", null], ["ff", ""]])";
@@ -799,6 +800,7 @@ TYPED_TEST(TestStringKernels, BinaryJoin) {
   CheckScalarBinary("binary_join", ArrayFromJSON(large_list(this->type()), list_json),
                     separator_null, expected);
 
+  // Array list, Array separator
   auto separators =
       ArrayFromJSON(this->type(), R"(["1", "2", "3", "4", "5", "6", null])");
   list_json =
@@ -809,6 +811,21 @@ TYPED_TEST(TestStringKernels, BinaryJoin) {
                     separators, expected);
   CheckScalarBinary("binary_join", ArrayFromJSON(large_list(this->type()), list_json),
                     separators, expected);
+
+  // Scalar list, Array separator
+  separators = ArrayFromJSON(this->type(), R"(["1", "", null])");
+  list_json = R"(["a", "bb", "ccc"])";
+  expected = ArrayFromJSON(this->type(), R"(["a1bb1ccc", "abbccc", null])");
+  CheckScalarBinary("binary_join", ScalarFromJSON(list(this->type()), list_json),
+                    separators, expected);
+  CheckScalarBinary("binary_join", ScalarFromJSON(large_list(this->type()), list_json),
+                    separators, expected);
+  list_json = R"(["a", "bb", null])";
+  expected = ArrayFromJSON(this->type(), R"([null, null, null])");
+  CheckScalarBinary("binary_join", ScalarFromJSON(list(this->type()), list_json),
+                    separators, expected);
+  CheckScalarBinary("binary_join", ScalarFromJSON(large_list(this->type()), list_json),
+                    separators, expected);
 }
 
 #ifdef ARROW_WITH_UTF8PROC
diff --git a/cpp/src/arrow/compute/kernels/test_util.cc b/cpp/src/arrow/compute/kernels/test_util.cc
index 2e6bb558674..583bb9880d9 100644
--- a/cpp/src/arrow/compute/kernels/test_util.cc
+++ b/cpp/src/arrow/compute/kernels/test_util.cc
@@ -215,6 +215,12 @@ void CheckScalarBinary(std::string func_name, std::shared_ptr<Array> left_input,
   CheckScalar(std::move(func_name), {left_input, right_input}, expected, options);
 }
 
+void CheckScalarBinary(std::string func_name, std::shared_ptr<Scalar> left_input,
+                       std::shared_ptr<Array> right_input,
+                       std::shared_ptr<Array> expected, const FunctionOptions* options) {
+  CheckScalar(std::move(func_name), {left_input, right_input}, expected, options);
+}
+
 void CheckDispatchBest(std::string func_name, std::vector<ValueDescr> original_values,
                        std::vector<ValueDescr> expected_equivalent_values) {
   ASSERT_OK_AND_ASSIGN(auto function, GetFunctionRegistry()->GetFunction(func_name));
diff --git a/cpp/src/arrow/compute/kernels/test_util.h b/cpp/src/arrow/compute/kernels/test_util.h
index 927e0eb699f..161106d67f8 100644
--- a/cpp/src/arrow/compute/kernels/test_util.h
+++ b/cpp/src/arrow/compute/kernels/test_util.h
@@ -125,6 +125,11 @@ void CheckScalarBinary(std::string func_name, std::shared_ptr<Array> left_input,
                        std::shared_ptr<Array> expected,
                        const FunctionOptions* options = nullptr);
 
+void CheckScalarBinary(std::string func_name, std::shared_ptr<Scalar> left_input,
+                       std::shared_ptr<Array> right_input,
+                       std::shared_ptr<Array> expected,
+                       const FunctionOptions* options = nullptr);
+
 void CheckVectorUnary(std::string func_name, Datum input, std::shared_ptr<Array> expected,
                       const FunctionOptions* options = nullptr);
 
diff --git a/docs/source/cpp/compute.rst b/docs/source/cpp/compute.rst
index 1508ee5d2b5..02c8fb3eac7 100644
--- a/docs/source/cpp/compute.rst
+++ b/docs/source/cpp/compute.rst
@@ -666,7 +666,7 @@ This function does the inverse of string splitting.
 
 * \(1) The first input must be an array, while the second can be a scalar or array.
   Each list of values in the first input is joined using each second input
-  as separator.  If there is null element in an input list, the corresponding
+  as separator.  If any input list is null or contains a null, the corresponding
   output will be null.
 
 

From 27cc07444bf09f77f98e742215e27c06cb506d97 Mon Sep 17 00:00:00 2001
From: Antoine Pitrou <antoine@python.org>
Date: Mon, 7 Jun 2021 16:29:08 +0200
Subject: [PATCH 5/5] s/AppendToCurrent/ExtendCurrent/

---
 cpp/src/arrow/array/array_binary_test.cc      | 14 ++++++-------
 cpp/src/arrow/array/builder_binary.h          | 20 +++++++++----------
 .../arrow/compute/kernels/scalar_string.cc    |  4 ++--
 cpp/src/arrow/compute/kernels/test_util.cc    |  2 --
 cpp/src/arrow/compute/kernels/test_util.h     |  4 +++-
 5 files changed, 22 insertions(+), 22 deletions(-)

diff --git a/cpp/src/arrow/array/array_binary_test.cc b/cpp/src/arrow/array/array_binary_test.cc
index add6b8e04b4..e593cf7e6c4 100644
--- a/cpp/src/arrow/array/array_binary_test.cc
+++ b/cpp/src/arrow/array/array_binary_test.cc
@@ -473,7 +473,7 @@ class TestStringBuilder : public TestBuilder {
     CheckStringArray(*result_, strings, is_valid, reps);
   }
 
-  void TestAppendToCurrent() {
+  void TestExtendCurrent() {
     std::vector<std::string> strings = {"", "bbbb", "aaaaa", "", "ccc"};
     std::vector<uint8_t> is_valid = {1, 1, 1, 0, 1};
 
@@ -486,7 +486,7 @@ class TestStringBuilder : public TestBuilder {
           ASSERT_OK(builder_->AppendNull());
         } else if (strings[i].length() > 3) {
           ASSERT_OK(builder_->Append(strings[i].substr(0, 3)));
-          ASSERT_OK(builder_->AppendToCurrent(strings[i].substr(3)));
+          ASSERT_OK(builder_->ExtendCurrent(strings[i].substr(3)));
         } else {
           ASSERT_OK(builder_->Append(strings[i]));
         }
@@ -501,7 +501,7 @@ class TestStringBuilder : public TestBuilder {
     CheckStringArray(*result_, strings, is_valid, reps);
   }
 
-  void TestAppendToCurrentUnsafe() {
+  void TestExtendCurrentUnsafe() {
     std::vector<std::string> strings = {"", "bbbb", "aaaaa", "", "ccc"};
     std::vector<uint8_t> is_valid = {1, 1, 1, 0, 1};
 
@@ -521,7 +521,7 @@ class TestStringBuilder : public TestBuilder {
           builder_->UnsafeAppendNull();
         } else if (strings[i].length() > 3) {
           builder_->UnsafeAppend(strings[i].substr(0, 3));
-          builder_->UnsafeAppendToCurrent(strings[i].substr(3));
+          builder_->UnsafeExtendCurrent(strings[i].substr(3));
         } else {
           builder_->UnsafeAppend(strings[i]);
         }
@@ -672,10 +672,10 @@ TYPED_TEST(TestStringBuilder, TestScalarAppend) { this->TestScalarAppend(); }
 
 TYPED_TEST(TestStringBuilder, TestScalarAppendUnsafe) { this->TestScalarAppendUnsafe(); }
 
-TYPED_TEST(TestStringBuilder, TestAppendToCurrent) { this->TestAppendToCurrent(); }
+TYPED_TEST(TestStringBuilder, TestExtendCurrent) { this->TestExtendCurrent(); }
 
-TYPED_TEST(TestStringBuilder, TestAppendToCurrentUnsafe) {
-  this->TestAppendToCurrentUnsafe();
+TYPED_TEST(TestStringBuilder, TestExtendCurrentUnsafe) {
+  this->TestExtendCurrentUnsafe();
 }
 
 TYPED_TEST(TestStringBuilder, TestVectorAppend) { this->TestVectorAppend(); }
diff --git a/cpp/src/arrow/array/builder_binary.h b/cpp/src/arrow/array/builder_binary.h
index 197e74999d6..a60031258ad 100644
--- a/cpp/src/arrow/array/builder_binary.h
+++ b/cpp/src/arrow/array/builder_binary.h
@@ -77,10 +77,10 @@ class BaseBinaryBuilder : public ArrayBuilder {
     return Append(value.data(), static_cast<offset_type>(value.size()));
   }
 
-  /// Append to the last appended value
+  /// Extend the last appended value by appending more data at the end
   ///
   /// Unlike Append, this does not create a new offset.
-  Status AppendToCurrent(const uint8_t* value, offset_type length) {
+  Status ExtendCurrent(const uint8_t* value, offset_type length) {
     // Safety check for UBSAN.
     if (ARROW_PREDICT_TRUE(length > 0)) {
       ARROW_RETURN_NOT_OK(ValidateOverflow(length));
@@ -89,9 +89,9 @@ class BaseBinaryBuilder : public ArrayBuilder {
     return Status::OK();
   }
 
-  Status AppendToCurrent(util::string_view value) {
-    return AppendToCurrent(reinterpret_cast<const uint8_t*>(value.data()),
-                           static_cast<offset_type>(value.size()));
+  Status ExtendCurrent(util::string_view value) {
+    return ExtendCurrent(reinterpret_cast<const uint8_t*>(value.data()),
+                         static_cast<offset_type>(value.size()));
   }
 
   Status AppendNulls(int64_t length) final {
@@ -150,14 +150,14 @@ class BaseBinaryBuilder : public ArrayBuilder {
     UnsafeAppend(value.data(), static_cast<offset_type>(value.size()));
   }
 
-  /// Like AppendToCurrent, but do not check capacity
-  void UnsafeAppendToCurrent(const uint8_t* value, offset_type length) {
+  /// Like ExtendCurrent, but do not check capacity
+  void UnsafeExtendCurrent(const uint8_t* value, offset_type length) {
     value_data_builder_.UnsafeAppend(value, length);
   }
 
-  void UnsafeAppendToCurrent(util::string_view value) {
-    UnsafeAppendToCurrent(reinterpret_cast<const uint8_t*>(value.data()),
-                          static_cast<offset_type>(value.size()));
+  void UnsafeExtendCurrent(util::string_view value) {
+    UnsafeExtendCurrent(reinterpret_cast<const uint8_t*>(value.data()),
+                        static_cast<offset_type>(value.size()));
   }
 
   void UnsafeAppendNull() {
diff --git a/cpp/src/arrow/compute/kernels/scalar_string.cc b/cpp/src/arrow/compute/kernels/scalar_string.cc
index 73a1d47bb47..eef7c173425 100644
--- a/cpp/src/arrow/compute/kernels/scalar_string.cc
+++ b/cpp/src/arrow/compute/kernels/scalar_string.cc
@@ -2865,8 +2865,8 @@ struct BinaryJoin {
       }
       builder->UnsafeAppend(strings.GetView(j_start));
       for (int64_t j = j_start + 1; j < j_end; ++j) {
-        builder->UnsafeAppendToCurrent(separators.GetView(i));
-        builder->UnsafeAppendToCurrent(strings.GetView(j));
+        builder->UnsafeExtendCurrent(separators.GetView(i));
+        builder->UnsafeExtendCurrent(strings.GetView(j));
       }
     }
 
diff --git a/cpp/src/arrow/compute/kernels/test_util.cc b/cpp/src/arrow/compute/kernels/test_util.cc
index 583bb9880d9..18257973150 100644
--- a/cpp/src/arrow/compute/kernels/test_util.cc
+++ b/cpp/src/arrow/compute/kernels/test_util.cc
@@ -35,8 +35,6 @@ namespace compute {
 
 namespace {
 
-using DatumVector = std::vector<Datum>;
-
 template <typename T>
 DatumVector GetDatums(const std::vector<T>& inputs) {
   std::vector<Datum> datums;
diff --git a/cpp/src/arrow/compute/kernels/test_util.h b/cpp/src/arrow/compute/kernels/test_util.h
index 161106d67f8..85ed04c183a 100644
--- a/cpp/src/arrow/compute/kernels/test_util.h
+++ b/cpp/src/arrow/compute/kernels/test_util.h
@@ -43,6 +43,8 @@ using internal::checked_cast;
 
 namespace compute {
 
+using DatumVector = std::vector<Datum>;
+
 template <typename Type, typename T>
 std::shared_ptr<Array> _MakeArray(const std::shared_ptr<DataType>& type,
                                   const std::vector<T>& values,
@@ -93,7 +95,7 @@ void CheckScalar(std::string func_name, const ScalarVector& inputs,
                  std::shared_ptr<Scalar> expected,
                  const FunctionOptions* options = nullptr);
 
-void CheckScalar(std::string func_name, const ArrayVector& inputs,
+void CheckScalar(std::string func_name, const DatumVector& inputs,
                  std::shared_ptr<Array> expected,
                  const FunctionOptions* options = nullptr);