ARROW-12614: [C++][Compute] Remove support for Tables in ExecuteScalarExpression

cpp/src/arrow/compute/exec/expression.cc

@@ -526,23 +526,6 @@ Result<Datum> ExecuteScalarExpression(const Expression& expr, const Datum& input
         "ExecuteScalarExpression cannot Execute non-scalar expression ", expr.ToString());
   }
 
-  if (input.kind() == Datum::TABLE) {
-    TableBatchReader reader(*input.table());
-    std::shared_ptr<RecordBatch> batch;
-
-    while (true) {
-      RETURN_NOT_OK(reader.ReadNext(&batch));
-      if (batch != nullptr) {
-        break;
-      }
-      ARROW_ASSIGN_OR_RAISE(Datum res, ExecuteScalarExpression(expr, batch));
-      if (res.is_scalar()) {
-        ARROW_ASSIGN_OR_RAISE(res, MakeArrayFromScalar(*res.scalar(), batch->num_rows(),
-                                                       exec_context->memory_pool()));
-      }
-    }
-  }
-
   if (auto lit = expr.literal()) return *lit;
 
   if (auto ref = expr.field_ref()) {

cpp/src/arrow/compute/exec/expression_test.cc

@@ -172,6 +172,7 @@ TEST(Expression, ToString) {
   EXPECT_EQ(literal("a").ToString(), "\"a\"");
   EXPECT_EQ(literal("a\nb").ToString(), "\"a\\nb\"");
   EXPECT_EQ(literal(std::make_shared<BooleanScalar>()).ToString(), "null");
+  EXPECT_EQ(literal(std::make_shared<Int64Scalar>()).ToString(), "null");
   EXPECT_EQ(literal(std::make_shared<BinaryScalar>(Buffer::FromString("az"))).ToString(),
             "\"617A\"");
 
@@ -540,7 +541,7 @@ void ExpectExecute(Expression expr, Datum in, Datum* actual_out = NULLPTR) {
   if (in.is_value()) {
     ASSERT_OK_AND_ASSIGN(expr, expr.Bind(in.descr()));
   } else {
-    ASSERT_OK_AND_ASSIGN(expr, expr.Bind(*in.record_batch()->schema()));
+    ASSERT_OK_AND_ASSIGN(expr, expr.Bind(*in.schema()));
   }
 
   ASSERT_OK_AND_ASSIGN(Datum actual, ExecuteScalarExpression(expr, in));

cpp/src/arrow/dataset/dataset_internal.h

@@ -54,11 +54,6 @@ inline Result<FragmentIterator> GetFragmentsFromDatasets(const DatasetVector& da
   return MakeFlattenIterator(std::move(fragments_it));
 }
 
-inline RecordBatchIterator IteratorFromReader(
-    const std::shared_ptr<RecordBatchReader>& reader) {
-  return MakeFunctionIterator([reader] { return reader->Next(); });
-}
-
 inline std::shared_ptr<Schema> SchemaFromColumnNames(
     const std::shared_ptr<Schema>& input, const std::vector<std::string>& column_names) {
   std::vector<std::shared_ptr<Field>> columns;

cpp/src/arrow/dataset/test_util.h

@@ -1161,7 +1161,7 @@ class WriteFileSystemDatasetMixin : public MakeFileSystemDatasetMixin {
       std::shared_ptr<Array> actual_struct;
 
       for (auto maybe_batch :
-           IteratorFromReader(std::make_shared<TableBatchReader>(*actual_table))) {
+           MakeIteratorFromReader(std::make_shared<TableBatchReader>(*actual_table))) {
         ASSERT_OK_AND_ASSIGN(auto batch, maybe_batch);
         ASSERT_OK_AND_ASSIGN(actual_struct, batch->ToStructArray());
       }

cpp/src/arrow/record_batch.h

@@ -199,6 +199,8 @@ class ARROW_EXPORT RecordBatch {
 /// \brief Abstract interface for reading stream of record batches
 class ARROW_EXPORT RecordBatchReader {
  public:
+  using ValueType = std::shared_ptr<RecordBatch>;
+
   virtual ~RecordBatchReader() = default;
 
   /// \return the shared schema of the record batches in the stream

cpp/src/arrow/util/iterator.h

@@ -560,4 +560,10 @@ Iterator<T> MakeFlattenIterator(Iterator<Iterator<T>> it) {
   return Iterator<T>(FlattenIterator<T>(std::move(it)));
 }
 
+template <typename Reader>
+Iterator<typename Reader::ValueType> MakeIteratorFromReader(
+    const std::shared_ptr<Reader>& reader) {
+  return MakeFunctionIterator([reader] { return reader->Next(); });
+}
+
 }  // namespace arrow