ARROW-15779: [Python] Create python bindings for Substrait consumer

.github/workflows/python.yml

@@ -130,6 +130,7 @@ jobs:
       ARROW_PLASMA: ON
       ARROW_PYTHON: ON
       ARROW_S3: ON
+      ARROW_SUBSTRAIT: ON
       ARROW_WITH_ZLIB: ON
       ARROW_WITH_LZ4: ON
       ARROW_WITH_BZ2: ON

ci/scripts/python_build.sh

@@ -64,6 +64,7 @@ export PYARROW_WITH_PLASMA=${ARROW_PLASMA:-OFF}
 export PYARROW_WITH_PARQUET=${ARROW_PARQUET:-OFF}
 export PYARROW_WITH_PARQUET_ENCRYPTION=${PARQUET_REQUIRE_ENCRYPTION:-ON}
 export PYARROW_WITH_S3=${ARROW_S3:-OFF}
+export PYARROW_WITH_SUBSTRAIT=${ARROW_SUBSTRAIT:-OFF}
 
 export PYARROW_PARALLEL=${n_jobs}
 

cpp/src/arrow/engine/CMakeLists.txt

@@ -26,7 +26,8 @@ set(ARROW_SUBSTRAIT_SRCS
     substrait/serde.cc
     substrait/plan_internal.cc
     substrait/relation_internal.cc
-    substrait/type_internal.cc)
+    substrait/type_internal.cc
+    substrait/util.cc)
 
 add_arrow_lib(arrow_substrait
               CMAKE_PACKAGE_NAME

cpp/src/arrow/engine/substrait/serde_test.cc

@@ -16,6 +16,7 @@
 // under the License.
 
 #include "arrow/engine/substrait/serde.h"
+#include "arrow/engine/substrait/util.h"
 
 #include <google/protobuf/descriptor.h>
 #include <google/protobuf/util/json_util.h>
@@ -752,5 +753,66 @@ TEST(Substrait, ExtensionSetFromPlanMissingFunc) {
           &ext_set));
 }
 
+Result<std::string> GetSubstraitJSON() {
+  ARROW_ASSIGN_OR_RAISE(std::string dir_string,
+                        arrow::internal::GetEnvVar("PARQUET_TEST_DATA"));
+  auto file_name =
+      arrow::internal::PlatformFilename::FromString(dir_string)->Join("binary.parquet");
+  auto file_path = file_name->ToString();
+  std::string substrait_json = R"({
+    "relations": [
+      {"rel": {
+        "read": {
+          "base_schema": {
+            "struct": {
+              "types": [ 
+                         {"binary": {}}
+                       ]
+            },
+            "names": [
+                      "foo"
+                      ]
+          },
+          "local_files": {
+            "items": [
+              {
+                "uri_file": "file://FILENAME_PLACEHOLDER",
+                "format": "FILE_FORMAT_PARQUET"
+              }
+            ]
+          }
+        }
+      }}
+    ]
+  })";
+  std::string filename_placeholder = "FILENAME_PLACEHOLDER";
+  substrait_json.replace(substrait_json.find(filename_placeholder),
+                         filename_placeholder.size(), file_path);
+  return substrait_json;
+}
+
+TEST(Substrait, GetRecordBatchReader) {
+#ifdef _WIN32
+  GTEST_SKIP() << "ARROW-16392: Substrait File URI not supported for Windows";
+#else
+  ASSERT_OK_AND_ASSIGN(std::string substrait_json, GetSubstraitJSON());
+  ASSERT_OK_AND_ASSIGN(auto buf, substrait::SerializeJsonPlan(substrait_json));
+  ASSERT_OK_AND_ASSIGN(auto reader, substrait::ExecuteSerializedPlan(*buf));
+  ASSERT_OK_AND_ASSIGN(auto table, Table::FromRecordBatchReader(reader.get()));
+  // Note: assuming the binary.parquet file contains fixed amount of records
+  // in case of a test failure, re-evalaute the content in the file
+  EXPECT_EQ(table->num_rows(), 12);
+#endif
+}
+
+TEST(Substrait, InvalidPlan) {
+  std::string substrait_json = R"({
+    "relations": [
+    ]
+  })";
+  ASSERT_OK_AND_ASSIGN(auto buf, substrait::SerializeJsonPlan(substrait_json));
+  ASSERT_RAISES(Invalid, substrait::ExecuteSerializedPlan(*buf));
+}
+
 }  // namespace engine
 }  // namespace arrow