Eliminate unnecessary refcount traffic from shared_ptr parameters passed by value that don't need to be

cpp/src/arrow/acero/asof_join_benchmark.cc

@@ -54,7 +54,7 @@ static void TableJoinOverhead(benchmark::State& state,
                               TableGenerationProperties right_table_properties,
                               int batch_size, int num_right_tables,
                               std::string factory_name,
-                              std::shared_ptr<ExecNodeOptions> options) {
+                              const std::shared_ptr<ExecNodeOptions>& options) {
   left_table_properties.column_prefix = "lt";
   left_table_properties.seed = 0;
   ASSERT_OK_AND_ASSIGN(TableStats left_table_stats, MakeTable(left_table_properties));

cpp/src/arrow/acero/asof_join_node.cc

@@ -238,7 +238,7 @@ struct MemoStore {
     Entry() = default;
 
     Entry(OnType time, std::shared_ptr<arrow::RecordBatch> batch, row_index_t row)
-        : time(time), batch(batch), row(row) {}
+        : time(time), batch(std::move(batch)), row(row) {}
 
     void swap(Entry& other) {
       std::swap(time, other.time);

cpp/src/arrow/acero/asof_join_node_test.cc

@@ -1398,7 +1398,7 @@ struct BackpressureCountingNode : public MapNode {
   BackpressureCountingNode(ExecPlan* plan, std::vector<ExecNode*> inputs,
                            std::shared_ptr<Schema> output_schema,
                            const BackpressureCountingNodeOptions& options)
-      : MapNode(plan, inputs, output_schema), counters(options.counters) {}
+      : MapNode(plan, inputs, std::move(output_schema)), counters(options.counters) {}
 
   static Result<ExecNode*> Make(ExecPlan* plan, std::vector<ExecNode*> inputs,
                                 const ExecNodeOptions& options) {

cpp/src/arrow/acero/hash_aggregate_test.cc

@@ -227,7 +227,7 @@ Result<Datum> NaiveGroupBy(std::vector<Datum> arguments, std::vector<Datum> keys
 }
 
 Result<Datum> MakeGroupByOutput(const std::vector<ExecBatch>& output_batches,
-                                const std::shared_ptr<Schema> output_schema,
+                                const std::shared_ptr<Schema>& output_schema,
                                 size_t num_aggregates, size_t num_keys, bool naive) {
   ArrayVector out_arrays(num_aggregates + num_keys);
   for (size_t i = 0; i < out_arrays.size(); ++i) {

cpp/src/arrow/acero/source_node.cc

@@ -301,7 +301,7 @@ struct SourceNode : ExecNode, public TracedNode {
 };
 
 struct TableSourceNode : public SourceNode {
-  TableSourceNode(ExecPlan* plan, std::shared_ptr<Table> table, int64_t batch_size)
+  TableSourceNode(ExecPlan* plan, const std::shared_ptr<Table>& table, int64_t batch_size)
       : SourceNode(plan, table->schema(), TableGenerator(*table, batch_size),
                    Ordering::Implicit()) {}
 
@@ -319,7 +319,7 @@ struct TableSourceNode : public SourceNode {
 
   const char* kind_name() const override { return "TableSourceNode"; }
 
-  static arrow::Status ValidateTableSourceNodeInput(const std::shared_ptr<Table> table,
+  static arrow::Status ValidateTableSourceNodeInput(const std::shared_ptr<Table>& table,
                                                     const int64_t batch_size) {
     if (table == nullptr) {
       return Status::Invalid("TableSourceNode requires table which is not null");
@@ -372,7 +372,7 @@ template <typename This, typename Options>
 struct SchemaSourceNode : public SourceNode {
   SchemaSourceNode(ExecPlan* plan, std::shared_ptr<Schema> schema,
                    arrow::AsyncGenerator<std::optional<ExecBatch>> generator)
-      : SourceNode(plan, schema, generator, Ordering::Implicit()) {}
+      : SourceNode(plan, std::move(schema), generator, Ordering::Implicit()) {}
 
   static Result<ExecNode*> Make(ExecPlan* plan, std::vector<ExecNode*> inputs,
                                 const ExecNodeOptions& options) {
@@ -408,7 +408,7 @@ struct SchemaSourceNode : public SourceNode {
 struct RecordBatchReaderSourceNode : public SourceNode {
   RecordBatchReaderSourceNode(ExecPlan* plan, std::shared_ptr<Schema> schema,
                               arrow::AsyncGenerator<std::optional<ExecBatch>> generator)
-      : SourceNode(plan, schema, generator, Ordering::Implicit()) {}
+      : SourceNode(plan, std::move(schema), generator, Ordering::Implicit()) {}
 
   static Result<ExecNode*> Make(ExecPlan* plan, std::vector<ExecNode*> inputs,
                                 const ExecNodeOptions& options) {

cpp/src/arrow/acero/source_node_test.cc

@@ -37,7 +37,7 @@ struct PauseThenStopNode : public MapNode {
   PauseThenStopNode(ExecPlan* plan, std::vector<ExecNode*> inputs,
                     std::shared_ptr<Schema> output_schema,
                     const PauseThenStopNodeOptions& options)
-      : MapNode(plan, inputs, output_schema), num_pass(options.num_pass) {}
+      : MapNode(plan, inputs, std::move(output_schema)), num_pass(options.num_pass) {}
 
   static Result<ExecNode*> Make(ExecPlan* plan, std::vector<ExecNode*> inputs,
                                 const ExecNodeOptions& options) {

cpp/src/arrow/acero/test_nodes.cc

@@ -275,7 +275,7 @@ struct GatedNode : public ExecNode, public TracedNode {
 
   GatedNode(ExecPlan* plan, std::vector<ExecNode*> inputs,
             std::shared_ptr<Schema> output_schema, const GatedNodeOptions& options)
-      : ExecNode(plan, inputs, {"input"}, output_schema),
+      : ExecNode(plan, inputs, {"input"}, std::move(output_schema)),
         TracedNode(this),
         gate_(options.gate) {}
 

cpp/src/arrow/acero/unmaterialized_table_internal.h

@@ -242,7 +242,7 @@ class UnmaterializedSliceBuilder {
       UnmaterializedCompositeTable<MAX_COMPOSITE_TABLES>* table_)
       : table(table_) {}
 
-  void AddEntry(std::shared_ptr<RecordBatch> rb, uint64_t start, uint64_t end) {
+  void AddEntry(const std::shared_ptr<RecordBatch>& rb, uint64_t start, uint64_t end) {
     if (rb) {
       table->AddRecordBatchRef(rb);
     }

cpp/src/arrow/adapters/orc/adapter.cc

@@ -137,7 +137,7 @@ class OrcStripeReader : public RecordBatchReader {
   OrcStripeReader(std::unique_ptr<liborc::RowReader> row_reader,
                   std::shared_ptr<Schema> schema, int64_t batch_size, MemoryPool* pool)
       : row_reader_(std::move(row_reader)),
-        schema_(schema),
+        schema_(std::move(schema)),
         pool_(pool),
         batch_size_{batch_size} {}
 

cpp/src/arrow/adapters/orc/adapter_test.cc

@@ -103,7 +103,7 @@ std::shared_ptr<Buffer> GenerateFixedDifferenceBuffer(int32_t fixed_length,
 }
 
 std::shared_ptr<Array> CastFixedSizeBinaryArrayToBinaryArray(
-    std::shared_ptr<Array> array) {
+    const std::shared_ptr<Array>& array) {
   auto fixed_size_binary_array = checked_pointer_cast<FixedSizeBinaryArray>(array);
   std::shared_ptr<Buffer> value_offsets = GenerateFixedDifferenceBuffer(
       fixed_size_binary_array->byte_width(), array->length() + 1);
@@ -114,7 +114,7 @@ std::shared_ptr<Array> CastFixedSizeBinaryArrayToBinaryArray(
 
 template <typename TargetArrayType>
 std::shared_ptr<Array> CastInt64ArrayToTemporalArray(
-    const std::shared_ptr<DataType>& type, std::shared_ptr<Array> array) {
+    const std::shared_ptr<DataType>& type, const std::shared_ptr<Array>& array) {
   std::shared_ptr<ArrayData> new_array_data =
       ArrayData::Make(type, array->length(), array->data()->buffers);
   return std::make_shared<TargetArrayType>(new_array_data);
@@ -128,7 +128,7 @@ Result<std::shared_ptr<Array>> GenerateRandomDate64Array(int64_t size,
 }
 
 Result<std::shared_ptr<Array>> GenerateRandomTimestampArray(
-    int64_t size, std::shared_ptr<TimestampType> type, double null_probability) {
+    int64_t size, const std::shared_ptr<TimestampType>& type, double null_probability) {
   random::RandomArrayGenerator rand(kRandomSeed);
   switch (type->unit()) {
     case TimeUnit::type::SECOND: {
@@ -1049,7 +1049,7 @@ namespace {
 // read them back and compare equality in the unit test). Because the orc reader
 // fills unselected values to nulls when reading from the file. So flattening
 // the SparseUnionArray before writing makes it easy for the array equality check.
-std::shared_ptr<Array> FlattenSparseUnionArray(std::shared_ptr<Array> array) {
+std::shared_ptr<Array> FlattenSparseUnionArray(const std::shared_ptr<Array>& array) {
   auto union_array = checked_pointer_cast<SparseUnionArray>(array);
   ArrayVector children;
   for (int i = 0; i < array->num_fields(); ++i) {
@@ -1061,7 +1061,7 @@ std::shared_ptr<Array> FlattenSparseUnionArray(std::shared_ptr<Array> array) {
                                             union_array->type_codes(), array->offset());
 }
 
-void TestUnionConversion(std::shared_ptr<Array> array) {
+void TestUnionConversion(const std::shared_ptr<Array>& array) {
   auto length = array->length();
   auto orc_type = liborc::Type::buildTypeFromString("uniontype<string,int>");
 

cpp/src/arrow/array/array_dict_test.cc

@@ -873,7 +873,7 @@ TEST(TestFixedSizeBinaryDictionaryBuilder, AppendArrayInvalidType) {
 #endif
 
 template <typename DecimalValue>
-void TestDecimalDictionaryBuilderBasic(std::shared_ptr<DataType> decimal_type) {
+void TestDecimalDictionaryBuilderBasic(const std::shared_ptr<DataType>& decimal_type) {
   // Build the dictionary Array
   DictionaryBuilder<FixedSizeBinaryType> builder(decimal_type);
 
@@ -903,7 +903,8 @@ TEST(TestDecimal256DictionaryBuilder, Basic) {
 }
 
 void TestDecimalDictionaryBuilderDoubleTableSize(
-    std::shared_ptr<DataType> decimal_type, FixedSizeBinaryBuilder& decimal_builder) {
+    const std::shared_ptr<DataType>& decimal_type,
+    FixedSizeBinaryBuilder& decimal_builder) {
   // Build the dictionary Array
   DictionaryBuilder<FixedSizeBinaryType> dict_builder(decimal_type);
 

cpp/src/arrow/array/array_nested.cc

@@ -130,15 +130,15 @@ Result<std::shared_ptr<typename TypeTraits<TYPE>::ArrayType>> ListArrayFromArray
   if (offsets.null_count() > 0) {
     ARROW_ASSIGN_OR_RAISE(auto buffers,
                           CleanListOffsets<TYPE>(null_bitmap, offsets, pool));
-    auto data = ArrayData::Make(type, offsets.length() - 1, std::move(buffers),
+    auto data = ArrayData::Make(std::move(type), offsets.length() - 1, std::move(buffers),
                                 {values.data()}, offsets.null_count(), /*offset=*/0);
     return std::make_shared<ArrayType>(std::move(data));
   }
 
   using OffsetArrayType = typename TypeTraits<OffsetArrowType>::ArrayType;
   const auto& typed_offsets = checked_cast<const OffsetArrayType&>(offsets);
   auto buffers = BufferVector({std::move(null_bitmap), typed_offsets.values()});
-  auto data = ArrayData::Make(type, offsets.length() - 1, std::move(buffers),
+  auto data = ArrayData::Make(std::move(type), offsets.length() - 1, std::move(buffers),
                               {values.data()}, null_count, offsets.offset());
   return std::make_shared<ArrayType>(std::move(data));
 }
@@ -209,8 +209,8 @@ Result<std::shared_ptr<typename TypeTraits<TYPE>::ArrayType>> ListViewArrayFromA
       typed_offsets.values(),
       typed_sizes.values(),
   });
-  auto data = ArrayData::Make(type, sizes.length(), std::move(buffers), {values.data()},
-                              null_count, array_offset);
+  auto data = ArrayData::Make(std::move(type), sizes.length(), std::move(buffers),
+                              {values.data()}, null_count, array_offset);
   return std::make_shared<ArrayType>(std::move(data));
 }
 

cpp/src/arrow/array/array_run_end_test.cc

@@ -68,7 +68,7 @@ class TestRunEndEncodedArray
   }
 
   std::shared_ptr<RunEndEncodedArray> RunEndEncodedArrayFromJSON(
-      int64_t logical_length, std::shared_ptr<DataType> value_type,
+      int64_t logical_length, const std::shared_ptr<DataType>& value_type,
       std::string_view run_ends_json, std::string_view values_json,
       int64_t logical_offset = 0) {
     auto run_ends = ArrayFromJSON(run_end_type, run_ends_json);

cpp/src/arrow/array/builder_run_end.cc

@@ -167,7 +167,8 @@ RunEndEncodedBuilder::ValueRunBuilder::ValueRunBuilder(
 RunEndEncodedBuilder::RunEndEncodedBuilder(
     MemoryPool* pool, const std::shared_ptr<ArrayBuilder>& run_end_builder,
     const std::shared_ptr<ArrayBuilder>& value_builder, std::shared_ptr<DataType> type)
-    : ArrayBuilder(pool), type_(internal::checked_pointer_cast<RunEndEncodedType>(type)) {
+    : ArrayBuilder(pool),
+      type_(internal::checked_pointer_cast<RunEndEncodedType>(std::move(type))) {
   auto value_run_builder =
       std::make_shared<ValueRunBuilder>(pool, value_builder, type_->value_type(), *this);
   value_run_builder_ = value_run_builder.get();

cpp/src/arrow/array/diff_test.cc

@@ -115,7 +115,7 @@ class DiffTest : public ::testing::Test {
                       /*verbose=*/true);
   }
 
-  void BaseAndTargetFromRandomFilter(std::shared_ptr<Array> values,
+  void BaseAndTargetFromRandomFilter(const std::shared_ptr<Array>& values,
                                      double filter_probability) {
     std::shared_ptr<Array> base_filter, target_filter;
     do {

cpp/src/arrow/buffer_test.cc

@@ -87,7 +87,8 @@ class MyDevice : public Device {
 
 class MyMemoryManager : public MemoryManager {
  public:
-  explicit MyMemoryManager(std::shared_ptr<Device> device) : MemoryManager(device) {}
+  explicit MyMemoryManager(const std::shared_ptr<Device>& device)
+      : MemoryManager(device) {}
 
   bool allow_copy() const {
     return checked_cast<const MyDevice&>(*device()).allow_copy();
@@ -135,7 +136,7 @@ class MyBuffer : public Buffer {
   MyBuffer(std::shared_ptr<MemoryManager> mm, const std::shared_ptr<Buffer>& parent)
       : Buffer(parent->data(), parent->size()) {
     parent_ = parent;
-    SetMemoryManager(mm);
+    SetMemoryManager(std::move(mm));
   }
 };
 

cpp/src/arrow/c/bridge.cc

@@ -1509,7 +1509,8 @@ class ImportedBuffer : public Buffer {
   ImportedBuffer(const uint8_t* data, int64_t size, std::shared_ptr<MemoryManager> mm,
                  DeviceAllocationType device_type,
                  std::shared_ptr<ImportedArrayData> import)
-      : Buffer(data, size, mm, nullptr, device_type), import_(std::move(import)) {}
+      : Buffer(data, size, std::move(mm), nullptr, device_type),
+        import_(std::move(import)) {}
 
   ~ImportedBuffer() override = default;
 
@@ -2288,22 +2289,22 @@ class ArrayStreamReader {
 
   Result<std::shared_ptr<RecordBatch>> ImportRecordBatchInternal(
       struct ArrowArray* array, std::shared_ptr<Schema> schema) {
-    return ImportRecordBatch(array, schema);
+    return ImportRecordBatch(array, std::move(schema));
   }
 
   Result<std::shared_ptr<RecordBatch>> ImportRecordBatchInternal(
       struct ArrowDeviceArray* array, std::shared_ptr<Schema> schema) {
-    return ImportDeviceRecordBatch(array, schema, mapper_);
+    return ImportDeviceRecordBatch(array, std::move(schema), mapper_);
   }
 
   Result<std::shared_ptr<Array>> ImportArrayInternal(
       struct ArrowArray* array, std::shared_ptr<arrow::DataType> type) {
-    return ImportArray(array, type);
+    return ImportArray(array, std::move(type));
   }
 
   Result<std::shared_ptr<Array>> ImportArrayInternal(
       struct ArrowDeviceArray* array, std::shared_ptr<arrow::DataType> type) {
-    return ImportDeviceArray(array, type, mapper_);
+    return ImportDeviceArray(array, std::move(type), mapper_);
   }
 
   Result<std::shared_ptr<Schema>> ReadSchema() {

cpp/src/arrow/c/bridge_test.cc

@@ -681,7 +681,7 @@ class TestArrayExport : public ::testing::Test {
   void SetUp() override { pool_ = default_memory_pool(); }
 
   static std::function<Result<std::shared_ptr<Array>>()> JSONArrayFactory(
-      std::shared_ptr<DataType> type, const char* json) {
+      const std::shared_ptr<DataType>& type, const char* json) {
     return [=]() { return ArrayFromJSON(type, json); };
   }
 
@@ -1458,14 +1458,14 @@ class TestDeviceArrayExport : public ::testing::Test {
   }
 
   static std::function<Result<std::shared_ptr<Array>>()> JSONArrayFactory(
-      const std::shared_ptr<MemoryManager>& mm, std::shared_ptr<DataType> type,
+      const std::shared_ptr<MemoryManager>& mm, const std::shared_ptr<DataType>& type,
       const char* json) {
     return [=]() { return ToDevice(mm, *ArrayFromJSON(type, json)->data()); };
   }
 
 #ifdef ARROW_COMPUTE
   static std::function<Result<std::shared_ptr<Array>>()> JSONREEArrayFactory(
-      const std::shared_ptr<MemoryManager>& mm, std::shared_ptr<DataType> type,
+      const std::shared_ptr<MemoryManager>& mm, const std::shared_ptr<DataType>& type,
       const char* json) {
     return [=]() -> Result<std::shared_ptr<Array>> {
       ARROW_ASSIGN_OR_RAISE(auto result, REEFromJSON(type, json));
@@ -3912,7 +3912,8 @@ class TestArrayRoundtrip : public ::testing::Test {
 
   void SetUp() override { pool_ = default_memory_pool(); }
 
-  static ArrayFactory JSONArrayFactory(std::shared_ptr<DataType> type, const char* json) {
+  static ArrayFactory JSONArrayFactory(const std::shared_ptr<DataType>& type,
+                                       const char* json) {
     return [=]() { return ArrayFromJSON(type, json); };
   }
 
@@ -4015,11 +4016,11 @@ class TestArrayRoundtrip : public ::testing::Test {
     ASSERT_EQ(pool_->bytes_allocated(), orig_bytes);
   }
 
-  void TestWithJSON(std::shared_ptr<DataType> type, const char* json) {
+  void TestWithJSON(const std::shared_ptr<DataType>& type, const char* json) {
     TestWithArrayFactory(JSONArrayFactory(type, json));
   }
 
-  void TestWithJSONSliced(std::shared_ptr<DataType> type, const char* json) {
+  void TestWithJSONSliced(const std::shared_ptr<DataType>& type, const char* json) {
     TestWithArrayFactory(SlicedArrayFactory(JSONArrayFactory(type, json)));
   }
 
@@ -4352,7 +4353,8 @@ class TestDeviceArrayRoundtrip : public ::testing::Test {
   }
 
   static ArrayFactory JSONArrayFactory(const std::shared_ptr<MemoryManager>& mm,
-                                       std::shared_ptr<DataType> type, const char* json) {
+                                       const std::shared_ptr<DataType>& type,
+                                       const char* json) {
     return [=]() { return ToDevice(mm, *ArrayFromJSON(type, json)->data()); };
   }
 
@@ -4463,12 +4465,12 @@ class TestDeviceArrayRoundtrip : public ::testing::Test {
   }
 
   void TestWithJSON(const std::shared_ptr<MemoryManager>& mm,
-                    std::shared_ptr<DataType> type, const char* json) {
+                    const std::shared_ptr<DataType>& type, const char* json) {
     TestWithArrayFactory(JSONArrayFactory(mm, type, json));
   }
 
   void TestWithJSONSliced(const std::shared_ptr<MemoryManager>& mm,
-                          std::shared_ptr<DataType> type, const char* json) {
+                          const std::shared_ptr<DataType>& type, const char* json) {
     TestWithArrayFactory(SlicedArrayFactory(JSONArrayFactory(mm, type, json)));
   }
 
@@ -4519,7 +4521,8 @@ class BaseArrayStreamTest : public ::testing::Test {
 
   void TearDown() override { ASSERT_EQ(pool_->bytes_allocated(), orig_allocated_); }
 
-  RecordBatchVector MakeBatches(std::shared_ptr<Schema> schema, ArrayVector arrays) {
+  RecordBatchVector MakeBatches(const std::shared_ptr<Schema>& schema,
+                                ArrayVector arrays) {
     DCHECK_EQ(schema->num_fields(), 1);
     RecordBatchVector batches;
     for (const auto& array : arrays) {

cpp/src/arrow/compute/function_benchmark.cc

@@ -37,7 +37,7 @@ namespace compute {
 constexpr int32_t kSeed = 0xfede4a7e;
 constexpr int64_t kScalarCount = 1 << 10;
 
-inline ScalarVector ToScalars(std::shared_ptr<Array> arr) {
+inline ScalarVector ToScalars(const std::shared_ptr<Array>& arr) {
   ScalarVector scalars{static_cast<size_t>(arr->length())};
   int64_t i = 0;
   for (auto& scalar : scalars) {