@@ -395,6 +395,7 @@ if(ARROW_COMPUTE)
        compute/exec/options.cc
        compute/exec/order_by_impl.cc
        compute/exec/partition_util.cc
+       compute/exec/options.cc
        compute/exec/project_node.cc
        compute/exec/sink_node.cc
        compute/exec/source_node.cc
 
@@ -16,6 +16,10 @@
 // under the License.
 
 #include "arrow/compute/exec/options.h"
+#include "arrow/compute/exec/exec_plan.h"
+#include "arrow/io/util_internal.h"
+#include "arrow/table.h"
+#include "arrow/util/async_generator.h"
 #include "arrow/util/logging.h"
 
 namespace arrow {
@@ -44,5 +48,18 @@ std::string ToString(JoinType t) {
   std::abort();
 }
 
+Result<std::shared_ptr<SourceNodeOptions>> SourceNodeOptions::FromTable(
+    const Table& table, arrow::internal::Executor* exc) {
+  std::shared_ptr<RecordBatchReader> reader = std::make_shared<TableBatchReader>(table);
+
+  if (exc == nullptr) return Status::TypeError("No executor provided.");
+
+  // Map the RecordBatchReader to a SourceNode
+  ARROW_ASSIGN_OR_RAISE(auto batch_gen, MakeReaderGenerator(std::move(reader), exc));
+
+  return std::shared_ptr<SourceNodeOptions>(
+      new SourceNodeOptions(table.schema(), batch_gen));
+}
+
 }  // namespace compute
 }  // namespace arrow
@@ -26,16 +26,19 @@
 #include "arrow/compute/api_vector.h"
 #include "arrow/compute/exec.h"
 #include "arrow/compute/exec/expression.h"
+#include "arrow/result.h"
+#include "arrow/util/async_generator.h"
 #include "arrow/util/async_util.h"
 #include "arrow/util/optional.h"
 #include "arrow/util/visibility.h"
 
 namespace arrow {
 namespace compute {
 
+using AsyncExecBatchGenerator = AsyncGenerator<util::optional<ExecBatch>>;
+
 /// \addtogroup execnode-options
 /// @{
-
 class ARROW_EXPORT ExecNodeOptions {
  public:
   virtual ~ExecNodeOptions() = default;
@@ -51,6 +54,9 @@ class ARROW_EXPORT SourceNodeOptions : public ExecNodeOptions {
                     std::function<Future<util::optional<ExecBatch>>()> generator)
       : output_schema(std::move(output_schema)), generator(std::move(generator)) {}
 
+  static Result<std::shared_ptr<SourceNodeOptions>> FromTable(const Table& table,
+                                                              arrow::internal::Executor*);
+
   std::shared_ptr<Schema> output_schema;
   std::function<Future<util::optional<ExecBatch>>()> generator;
 };
 
@@ -399,6 +399,7 @@ set(CYTHON_EXTENSIONS
     lib
     _compute
     _csv
+    _exec_plan
     _feather
     _fs
     _hdfsio
 
@@ -49,3 +49,6 @@ cdef class Expression(_Weakrefable):
 
     @staticmethod
     cdef Expression _expr_or_scalar(object expr)
+
+
+cdef CExpression _true
@@ -2218,8 +2218,11 @@ cdef class Expression(_Weakrefable):
         return Expression._call("is_in", [self], options)
 
     @staticmethod
-    def _field(str name not None):
-        return Expression.wrap(CMakeFieldExpression(tobytes(name)))
+    def _field(name_or_idx not None):
+        if isinstance(name_or_idx, str):
+            return Expression.wrap(CMakeFieldExpression(tobytes(name_or_idx)))
+        else:
+            return Expression.wrap(CMakeFieldExpressionByIndex(name_or_idx))
 
     @staticmethod
     def _scalar(value):