GH-36845: [C++][Python] Allow type promotion on pa.concat_tables

cpp/src/arrow/dataset/discovery.cc

@@ -47,7 +47,7 @@ Result<std::shared_ptr<Schema>> DatasetFactory::Inspect(InspectOptions options)
     return arrow::schema({});
   }
 
-  return UnifySchemas(schemas);
+  return UnifySchemas(schemas, options.field_merge_options);
 }
 
 Result<std::shared_ptr<Dataset>> DatasetFactory::Finish() {

cpp/src/arrow/dataset/discovery.h

@@ -58,6 +58,10 @@ struct InspectOptions {
   /// `kInspectAllFragments`. A value of `0` disables inspection of fragments
   /// altogether so only the partitioning schema will be inspected.
   int fragments = 1;
+
+  /// Control how to unify types. By default, types are merged strictly (the
+  /// type must match exactly, except nulls can be merged with other types).
+  Field::MergeOptions field_merge_options = Field::MergeOptions::Defaults();
 };
 
 struct FinishOptions {

cpp/src/arrow/dataset/discovery_test.cc

@@ -120,6 +120,12 @@ TEST_F(MockDatasetFactoryTest, UnifySchemas) {
   ASSERT_RAISES(Invalid, factory_->Inspect());
   // Return the individual schema for closer inspection should not fail.
   AssertInspectSchemas({schema({i32, f64}), schema({f64, i32_fake})});
+
+  MakeFactory({schema({field("num", int32())}), schema({field("num", float64())})});
+  ASSERT_RAISES(Invalid, factory_->Inspect());
+  InspectOptions permissive_options;
+  permissive_options.field_merge_options = Field::MergeOptions::Permissive();
+  AssertInspect(schema({field("num", float64())}), permissive_options);
 }
 
 class FileSystemDatasetFactoryTest : public DatasetFactoryTest {
@@ -474,6 +480,12 @@ TEST(UnionDatasetFactoryTest, ConflictingSchemas) {
   auto i32_schema = schema({i32});
   ASSERT_OK_AND_ASSIGN(auto dataset, factory->Finish(i32_schema));
   EXPECT_EQ(*dataset->schema(), *i32_schema);
+
+  // The user decided to allow merging the types.
+  FinishOptions options;
+  options.inspect_options.field_merge_options = Field::MergeOptions::Permissive();
+  ASSERT_OK_AND_ASSIGN(dataset, factory->Finish(options));
+  EXPECT_EQ(*dataset->schema(), *schema({f64, i32}));
 }
 
 }  // namespace dataset

cpp/src/arrow/table.cc

@@ -38,9 +38,15 @@
 #include "arrow/type_fwd.h"
 #include "arrow/type_traits.h"
 #include "arrow/util/checked_cast.h"
+// Get ARROW_COMPUTE definition
+#include "arrow/util/config.h"
 #include "arrow/util/logging.h"
 #include "arrow/util/vector.h"
 
+#ifdef ARROW_COMPUTE
+#include "arrow/compute/cast.h"
+#endif
+
 namespace arrow {
 
 using internal::checked_cast;
@@ -449,7 +455,8 @@ Result<std::shared_ptr<Table>> ConcatenateTables(
 
 Result<std::shared_ptr<Table>> PromoteTableToSchema(const std::shared_ptr<Table>& table,
                                                     const std::shared_ptr<Schema>& schema,
-                                                    MemoryPool* pool) {
+                                                    MemoryPool* pool,
+                                                    const compute::CastOptions options) {
   const std::shared_ptr<Schema> current_schema = table->schema();
   if (current_schema->Equals(*schema, /*check_metadata=*/false)) {
     return table->ReplaceSchemaMetadata(schema->metadata());
@@ -502,9 +509,23 @@ Result<std::shared_ptr<Table>> PromoteTableToSchema(const std::shared_ptr<Table>
       continue;
     }
 
+#ifdef ARROW_COMPUTE
+    if (!compute::CanCast(*current_field->type(), *field->type())) {
+      return Status::Invalid("Unable to promote field ", field->name(),
+                             ": incompatible types: ", field->type()->ToString(), " vs ",
+                             current_field->type()->ToString());
+    }
+    compute::ExecContext ctx(pool);
+    ARROW_ASSIGN_OR_RAISE(auto casted, compute::Cast(table->column(field_index),
+                                                     field->type(), options, &ctx));
+    columns.push_back(casted.chunked_array());
+#else
     return Status::Invalid("Unable to promote field ", field->name(),
                            ": incompatible types: ", field->type()->ToString(), " vs ",
-                           current_field->type()->ToString());
+                           current_field->type()->ToString(),
+                           " (Arrow must be built with ARROW_COMPUTE "
+                           "in order to cast incompatible types)");
+#endif
   }
 
   auto unseen_field_iter = std::find(fields_seen.begin(), fields_seen.end(), false);

cpp/src/arrow/table.h

@@ -23,6 +23,7 @@
 #include <vector>
 
 #include "arrow/chunked_array.h"  // IWYU pragma: keep
+#include "arrow/compute/cast.h"
 #include "arrow/record_batch.h"
 #include "arrow/status.h"
 #include "arrow/type.h"
@@ -315,22 +316,27 @@ Result<std::shared_ptr<Table>> ConcatenateTables(
 
 /// \brief Promotes a table to conform to the given schema.
 ///
-/// If a field in the schema does not have a corresponding column in the
-/// table, a column of nulls will be added to the resulting table.
-/// If the corresponding column is of type Null, it will be promoted to
-/// the type specified by schema, with null values filled.
+/// If a field in the schema does not have a corresponding column in
+/// the table, a column of nulls will be added to the resulting table.
+/// If the corresponding column is of type Null, it will be promoted
+/// to the type specified by schema, with null values filled. If Arrow
+/// was built with ARROW_COMPUTE, then the column will be casted to
+/// the type specified by the schema.
+///
 /// Returns an error:
 /// - if the corresponding column's type is not compatible with the
 ///   schema.
 /// - if there is a column in the table that does not exist in the schema.
+/// - if the cast fails or casting would be required but is not available.
 ///
 /// \param[in] table the input Table
 /// \param[in] schema the target schema to promote to
 /// \param[in] pool The memory pool to be used if null-filled arrays need to
+/// \param[in] options The cast options to allow promotion of types
 /// be created.
 ARROW_EXPORT
 Result<std::shared_ptr<Table>> PromoteTableToSchema(
     const std::shared_ptr<Table>& table, const std::shared_ptr<Schema>& schema,
-    MemoryPool* pool = default_memory_pool());
-
+    MemoryPool* pool = default_memory_pool(),
+    compute::CastOptions options = compute::CastOptions::Safe());
 }  // namespace arrow

cpp/src/arrow/table_test.cc

@@ -420,8 +420,9 @@ TEST_F(TestPromoteTableToSchema, IncompatibleTypes) {
   // Invalid promotion: int32 to null.
   ASSERT_RAISES(Invalid, PromoteTableToSchema(table, schema({field("field", null())})));
 
-  // Invalid promotion: int32 to uint32.
-  ASSERT_RAISES(Invalid, PromoteTableToSchema(table, schema({field("field", uint32())})));
+  // Invalid promotion: int32 to list.
+  ASSERT_RAISES(Invalid,
+                PromoteTableToSchema(table, schema({field("field", list(int32()))})));
 }
 
 TEST_F(TestPromoteTableToSchema, IncompatibleNullity) {
@@ -520,6 +521,36 @@ TEST_F(ConcatenateTablesWithPromotionTest, Simple) {
   AssertTablesEqualUnorderedFields(*expected, *result);
 }
 
+TEST_F(ConcatenateTablesWithPromotionTest, Unify) {
+  auto t1 = TableFromJSON(schema({field("f0", int32())}), {"[[0], [1]]"});
+  auto t2 = TableFromJSON(schema({field("f0", int64())}), {"[[2], [3]]"});
+  auto t3 = TableFromJSON(schema({field("f0", null())}), {"[[null], [null]]"});
+
+  auto expected_int64 =
+      TableFromJSON(schema({field("f0", int64())}), {"[[0], [1], [2], [3]]"});
+  auto expected_null =
+      TableFromJSON(schema({field("f0", int32())}), {"[[0], [1], [null], [null]]"});
+
+  ConcatenateTablesOptions options;
+  EXPECT_RAISES_WITH_MESSAGE_THAT(Invalid,
+                                  ::testing::HasSubstr("Schema at index 1 was different"),
+                                  ConcatenateTables({t1, t2}, options));
+  EXPECT_RAISES_WITH_MESSAGE_THAT(Invalid,
+                                  ::testing::HasSubstr("Schema at index 1 was different"),
+                                  ConcatenateTables({t1, t3}, options));
+
+  options.unify_schemas = true;
+  EXPECT_RAISES_WITH_MESSAGE_THAT(Invalid,
+                                  ::testing::HasSubstr("Field f0 has incompatible types"),
+                                  ConcatenateTables({t1, t2}, options));
+  ASSERT_OK_AND_ASSIGN(auto actual, ConcatenateTables({t1, t3}, options));
+  AssertTablesEqual(*expected_null, *actual, /*same_chunk_layout=*/false);
+
+  options.field_merge_options.promote_numeric_width = true;
+  ASSERT_OK_AND_ASSIGN(actual, ConcatenateTables({t1, t2}, options));
+  AssertTablesEqual(*expected_int64, *actual, /*same_chunk_layout=*/false);
+}
+
 TEST_F(TestTable, Slice) {
   const int64_t length = 10;