GH-34824: [C++][Python] Construct month-day-nanosecond intervals from parts

cpp/src/arrow/CMakeLists.txt

@@ -865,6 +865,7 @@ if(ARROW_COMPUTE)
        compute/kernels/scalar_string_ascii.cc
        compute/kernels/scalar_string_utf8.cc
        compute/kernels/scalar_temporal_binary.cc
+       compute/kernels/scalar_temporal_ternary.cc
        compute/kernels/scalar_temporal_unary.cc
        compute/kernels/scalar_validity.cc
        compute/kernels/util_internal.cc

cpp/src/arrow/compute/api_scalar.cc

@@ -953,6 +953,12 @@ SCALAR_EAGER_BINARY(MillisecondsBetween, "milliseconds_between")
 SCALAR_EAGER_BINARY(MicrosecondsBetween, "microseconds_between")
 SCALAR_EAGER_BINARY(NanosecondsBetween, "nanoseconds_between")
 
+Result<Datum> MonthDayNanoIntervalFromParts(const Datum& months, const Datum& days,
+                                            const Datum& nanoseconds, ExecContext* ctx) {
+  return CallFunction("month_day_nano_interval_from_parts", {months, days, nanoseconds},
+                      ctx);
+}
+
 // ----------------------------------------------------------------------
 // Structural transforms
 Result<Datum> MapLookup(const Datum& arg, MapLookupOptions options, ExecContext* ctx) {

cpp/src/arrow/compute/api_scalar.h

@@ -1732,6 +1732,24 @@ ARROW_EXPORT Result<Datum> WeeksBetween(const Datum& left, const Datum& right,
 ARROW_EXPORT Result<Datum> MonthDayNanoBetween(const Datum& left, const Datum& right,
                                                ExecContext* ctx = NULLPTR);
 
+/// \brief Construct month-day-nanosecond intervals from integer components
+///
+/// Months and days are safely cast to int32, and nanoseconds to int64.
+/// Scalar components are broadcast. If any component is null, the output is null.
+/// Component values are preserved without normalization.
+///
+/// \param[in] months integer array or scalar of months
+/// \param[in] days integer array or scalar of days
+/// \param[in] nanoseconds integer array or scalar of nanoseconds
+/// \param[in] ctx the function execution context, optional
+/// \return the resulting datum
+///
+/// \since 27.0.0
+ARROW_EXPORT Result<Datum> MonthDayNanoIntervalFromParts(const Datum& months,
+                                                         const Datum& days,
+                                                         const Datum& nanoseconds,
+                                                         ExecContext* ctx = NULLPTR);
+
 /// \brief DayTime Between finds the number of days and milliseconds between two values
 ///
 /// \param[in] left input treated as the start time

cpp/src/arrow/compute/initialize.cc

@@ -39,6 +39,7 @@ Status RegisterComputeKernels() {
   internal::RegisterScalarStringAscii(registry);
   internal::RegisterScalarStringUtf8(registry);
   internal::RegisterScalarTemporalBinary(registry);
+  internal::RegisterScalarTemporalTernary(registry);
   internal::RegisterScalarTemporalUnary(registry);
   internal::RegisterScalarValidity(registry);
 

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

@@ -19,6 +19,8 @@
 
 #include "benchmark/benchmark.h"
 
+#include "arrow/array.h"
+#include "arrow/array/builder_time.h"
 #include "arrow/compute/api_scalar.h"
 #include "arrow/testing/gtest_util.h"
 #include "arrow/testing/random.h"
@@ -166,6 +168,75 @@ static void BenchmarkAssumeTimezone(benchmark::State& state) {
   state.SetItemsProcessed(state.iterations() * array_size);
 }
 
+void IntervalFromPartsSetArgs(benchmark::internal::Benchmark* bench) {
+  for (const auto size : {1024, 65536, 1048576}) {
+    for (const auto inverse_null_proportion : {0, 100, 2}) {
+      bench->Args({size, inverse_null_proportion});
+    }
+  }
+}
+
+struct IntervalPartsInputs {
+  Datum months;
+  Datum days;
+  std::shared_ptr<Int64Array> nanoseconds;
+};
+
+template <bool ScalarParts>
+static IntervalPartsInputs MakeIntervalPartsInputs(const RegressionArgs& args) {
+  auto rand = random::RandomArrayGenerator(kSeed);
+  IntervalPartsInputs inputs;
+  if constexpr (ScalarParts) {
+    inputs.months = Datum(int32_t{2});
+    inputs.days = Datum(int32_t{-3});
+  } else {
+    inputs.months = rand.Numeric<Int32Type>(args.size, -12, 12, args.null_proportion);
+    inputs.days = rand.Numeric<Int32Type>(args.size, -31, 31, args.null_proportion);
+  }
+  inputs.nanoseconds = std::static_pointer_cast<Int64Array>(
+      rand.Numeric<Int64Type>(args.size, kInt64Min, kInt64Max, args.null_proportion));
+  return inputs;
+}
+
+template <bool ScalarParts>
+static void BenchmarkMonthDayNanoIntervalFromParts(benchmark::State& state) {
+  RegressionArgs args(state, /*size_is_bytes=*/false);
+  const IntervalPartsInputs inputs = MakeIntervalPartsInputs<ScalarParts>(args);
+  ExecContext* ctx = default_exec_context();
+
+  for (auto _ : state) {
+    ABORT_NOT_OK(MonthDayNanoIntervalFromParts(inputs.months, inputs.days,
+                                               inputs.nanoseconds, ctx));
+  }
+}
+
+template <bool ScalarParts>
+static void BenchmarkMonthDayNanoIntervalBuilder(benchmark::State& state) {
+  RegressionArgs args(state, /*size_is_bytes=*/false);
+  const IntervalPartsInputs inputs = MakeIntervalPartsInputs<ScalarParts>(args);
+  const auto months = ScalarParts ? nullptr : inputs.months.array_as<Int32Array>();
+  const auto days = ScalarParts ? nullptr : inputs.days.array_as<Int32Array>();
+
+  for (auto _ : state) {
+    MonthDayNanoIntervalBuilder builder;
+    ABORT_NOT_OK(builder.Reserve(args.size));
+    for (int64_t i = 0; i < args.size; ++i) {
+      bool valid = inputs.nanoseconds->IsValid(i);
+      if constexpr (!ScalarParts) {
+        valid = valid && months->IsValid(i) && days->IsValid(i);
+      }
+      if (valid) {
+        builder.UnsafeAppend({ScalarParts ? int32_t{2} : months->Value(i),
+                              ScalarParts ? int32_t{-3} : days->Value(i),
+                              inputs.nanoseconds->Value(i)});
+      } else {
+        builder.UnsafeAppendNull();
+      }
+    }
+    ABORT_NOT_OK(builder.Finish());
+  }
+}
+
 auto zoned = timestamp(TimeUnit::NANO, "Pacific/Marquesas");
 auto non_zoned = timestamp(TimeUnit::NANO);
 auto time32_type = time32(TimeUnit::MILLI);
@@ -247,6 +318,17 @@ BENCHMARK_TEMPLATE(BenchmarkStrptime, non_zoned)->Apply(SetArgs);
 BENCHMARK_TEMPLATE(BenchmarkStrptime, zoned)->Apply(SetArgs);
 BENCHMARK(BenchmarkAssumeTimezone)->Apply(SetArgs);
 
+// Null proportions apply independently to each array input. Scalar months and
+// days are valid constants, and nanoseconds remains an array in both cases.
+BENCHMARK_TEMPLATE(BenchmarkMonthDayNanoIntervalFromParts, false)
+    ->Apply(IntervalFromPartsSetArgs);
+BENCHMARK_TEMPLATE(BenchmarkMonthDayNanoIntervalFromParts, true)
+    ->Apply(IntervalFromPartsSetArgs);
+BENCHMARK_TEMPLATE(BenchmarkMonthDayNanoIntervalBuilder, false)
+    ->Apply(IntervalFromPartsSetArgs);
+BENCHMARK_TEMPLATE(BenchmarkMonthDayNanoIntervalBuilder, true)
+    ->Apply(IntervalFromPartsSetArgs);
+
 // binary temporal benchmarks
 DECLARE_TEMPORAL_BINARY_BENCHMARKS_DATES_AND_TIMESTAMPS(YearsBetween);
 DECLARE_TEMPORAL_BINARY_BENCHMARKS_DATES_AND_TIMESTAMPS(QuartersBetween);

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

@@ -0,0 +1,107 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/compute/kernels/common_internal.h"
+#include "arrow/compute/registry_internal.h"
+#include "arrow/util/bit_run_reader.h"
+#include "arrow/util/logging_internal.h"
+
+namespace arrow {
+namespace compute {
+namespace internal {
+namespace {
+
+struct MonthDayNanoIntervalFromPartsFunction : ScalarFunction {
+  using ScalarFunction::ScalarFunction;
+
+  Result<const Kernel*> DispatchBest(std::vector<TypeHolder>* types) const override {
+    RETURN_NOT_OK(CheckArity(types->size()));
+    EnsureDictionaryDecoded(types);
+    for (const auto& type : *types) {
+      if (!is_integer(type.id()) && type.id() != Type::NA) {
+        return arrow::compute::detail::NoMatchingKernel(this, *types);
+      }
+    }
+    // The executor safely casts integer components to their physical widths.
+    (*types)[0] = int32();
+    (*types)[1] = int32();
+    (*types)[2] = int64();
+    return ScalarFunction::DispatchExact(*types);
+  }
+};
+
+template <typename Type>
+struct IntervalPartValues {
+  using CType = typename Type::c_type;
+
+  explicit IntervalPartValues(const ExecValue& value) {
+    if (value.is_array() && value.array.buffers[1].data != nullptr) {
+      values = value.array.GetValues<CType>(1);
+    } else if (value.is_scalar() && value.scalar->is_valid) {
+      scalar = UnboxScalar<Type>::Unbox(*value.scalar);
+    }
+  }
+
+  CType operator[](int64_t i) const { return values ? values[i] : scalar; }
+
+  const CType* values = nullptr;
+  CType scalar = 0;
+};
+
+Status ExecMonthDayNanoIntervalFromParts(KernelContext*, const ExecSpan& batch,
+                                         ExecResult* out) {
+  if (batch.length == 0) return Status::OK();
+
+  const IntervalPartValues<Int32Type> months(batch[0]);
+  const IntervalPartValues<Int32Type> days(batch[1]);
+  const IntervalPartValues<Int64Type> nanoseconds(batch[2]);
+  auto* output = out->array_span_mutable();
+  auto* values = output->GetValues<MonthDayNanoIntervalType::MonthDayNanos>(1);
+  // The executor has already intersected the three input validity bitmaps.
+  // Visit only valid rows, including when an input is a null scalar.
+  arrow::internal::VisitSetBitRunsVoid(
+      output->buffers[0].data, output->offset, batch.length,
+      [&](int64_t start, int64_t length) {
+        for (int64_t i = start; i < start + length; ++i) {
+          values[i] = {months[i], days[i], nanoseconds[i]};
+        }
+      });
+  return Status::OK();
+}
+
+const FunctionDoc month_day_nano_interval_from_parts_doc{
+    "Construct month-day-nanosecond intervals from separate components",
+    ("Months, days, and nanoseconds must be integer arrays or scalars.\n"
+     "Months and days are safely cast to int32, and nanoseconds to int64.\n"
+     "Scalar inputs are broadcast. If any component is null, the interval is null.\n"
+     "Components are preserved without normalization."),
+    {"months", "days", "nanoseconds"}};
+
+}  // namespace
+
+void RegisterScalarTemporalTernary(FunctionRegistry* registry) {
+  auto function = std::make_shared<MonthDayNanoIntervalFromPartsFunction>(
+      "month_day_nano_interval_from_parts", Arity::Ternary(),
+      month_day_nano_interval_from_parts_doc);
+  DCHECK_OK(function->AddKernel({int32(), int32(), int64()}, month_day_nano_interval(),
+                                ExecMonthDayNanoIntervalFromParts));
+  DCHECK_OK(registry->AddFunction(std::move(function)));
+}
+
+}  // namespace internal
+}  // namespace compute
+}  // namespace arrow