This is an automated email from the ASF dual-hosted git repository.
westonpace pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/master by this push:
new 3c2479f58f ARROW-17736: [C++] Added a fallback name resolution
mechanism to the Substrait producer. (#14143)
3c2479f58f is described below
commit 3c2479f58f71d0417cadd9d8e51112e2220ac8ee
Author: Weston Pace <[email protected]>
AuthorDate: Mon Sep 26 10:41:43 2022 -1000
ARROW-17736: [C++] Added a fallback name resolution mechanism to the
Substrait producer. (#14143)
This will only be used if the URI part of the function reference is empty
or /. This is for compatibility with Isthmus which still hasn't decided what
to use for the URI portion. Longer term we may remove this. Or we may leave
it in as a utility for developers of new functions that don't want to have to
worry about the URI initially.
Authored-by: Weston Pace <[email protected]>
Signed-off-by: Weston Pace <[email protected]>
---
.../arrow/engine/substrait/expression_internal.cc | 31 +++-
cpp/src/arrow/engine/substrait/extension_set.cc | 40 ++++-
cpp/src/arrow/engine/substrait/extension_set.h | 16 ++
cpp/src/arrow/engine/substrait/plan_internal.cc | 3 +-
cpp/src/arrow/engine/substrait/plan_internal.h | 2 +-
.../arrow/engine/substrait/relation_internal.cc | 13 +-
cpp/src/arrow/engine/substrait/serde.cc | 3 +-
cpp/src/arrow/engine/substrait/serde_test.cc | 185 ++++++++++++++++-----
8 files changed, 239 insertions(+), 54 deletions(-)
diff --git a/cpp/src/arrow/engine/substrait/expression_internal.cc
b/cpp/src/arrow/engine/substrait/expression_internal.cc
index 0f1c41f14c..9445be6f03 100644
--- a/cpp/src/arrow/engine/substrait/expression_internal.cc
+++ b/cpp/src/arrow/engine/substrait/expression_internal.cc
@@ -37,6 +37,21 @@ using internal::checked_cast;
namespace engine {
+namespace {
+
+Id NormalizeFunctionName(Id id) {
+ // Substrait plans encode the types into the function name so it might look
like
+ // add:opt_i32_i32. We don't care about the :opt_i32_i32 so we just trim it
+ std::string_view func_name = id.name;
+ std::size_t colon_index = func_name.find_first_of(':');
+ if (colon_index != std::string_view::npos) {
+ func_name = func_name.substr(0, colon_index);
+ }
+ return {id.uri, func_name};
+}
+
+} // namespace
+
Status DecodeArg(const substrait::FunctionArgument& arg, uint32_t idx,
SubstraitCall* call, const ExtensionSet& ext_set,
const ConversionOptions& conversion_options) {
@@ -112,6 +127,7 @@ Result<SubstraitCall> FromProto(const
substrait::AggregateFunction& func, bool i
ARROW_ASSIGN_OR_RAISE(auto output_type_and_nullable,
FromProto(func.output_type(), ext_set,
conversion_options));
ARROW_ASSIGN_OR_RAISE(Id id,
ext_set.DecodeFunction(func.function_reference()));
+ id = NormalizeFunctionName(id);
SubstraitCall call(id, output_type_and_nullable.first,
output_type_and_nullable.second,
is_hash);
for (int i = 0; i < func.arguments_size(); i++) {
@@ -248,8 +264,19 @@ Result<compute::Expression> FromProto(const
substrait::Expression& expr,
ARROW_ASSIGN_OR_RAISE(Id function_id,
ext_set.DecodeFunction(scalar_fn.function_reference()));
- ARROW_ASSIGN_OR_RAISE(ExtensionIdRegistry::SubstraitCallToArrow
function_converter,
-
ext_set.registry()->GetSubstraitCallToArrow(function_id));
+ function_id = NormalizeFunctionName(function_id);
+ ExtensionIdRegistry::SubstraitCallToArrow function_converter;
+
+ if (function_id.uri.empty() || function_id.uri[0] == '/') {
+ // Currently the Substrait project has not aligned on a standard URI
and often
+ // seems to use /. In that case we fall back to name-only matching.
+ ARROW_ASSIGN_OR_RAISE(
+ function_converter,
+
ext_set.registry()->GetSubstraitCallToArrowFallback(function_id.name));
+ } else {
+ ARROW_ASSIGN_OR_RAISE(function_converter,
+
ext_set.registry()->GetSubstraitCallToArrow(function_id));
+ }
ARROW_ASSIGN_OR_RAISE(
SubstraitCall substrait_call,
DecodeScalarFunction(function_id, scalar_fn, ext_set,
conversion_options));
diff --git a/cpp/src/arrow/engine/substrait/extension_set.cc
b/cpp/src/arrow/engine/substrait/extension_set.cc
index af3d146704..63ba598c82 100644
--- a/cpp/src/arrow/engine/substrait/extension_set.cc
+++ b/cpp/src/arrow/engine/substrait/extension_set.cc
@@ -30,9 +30,6 @@ namespace arrow {
namespace engine {
namespace {
-// TODO(ARROW-16988): replace this with EXACT_ROUNDTRIP mode
-constexpr bool kExactRoundTrip = true;
-
struct TypePtrHashEq {
template <typename Ptr>
size_t operator()(const Ptr& type) const {
@@ -210,7 +207,8 @@ Status ExtensionSet::AddUri(Id id) {
Result<ExtensionSet> ExtensionSet::Make(
std::unordered_map<uint32_t, std::string_view> uris,
std::unordered_map<uint32_t, Id> type_ids,
- std::unordered_map<uint32_t, Id> function_ids, const ExtensionIdRegistry*
registry) {
+ std::unordered_map<uint32_t, Id> function_ids,
+ const ConversionOptions& conversion_options, const ExtensionIdRegistry*
registry) {
ExtensionSet set(default_extension_id_registry());
set.registry_ = registry;
@@ -219,7 +217,7 @@ Result<ExtensionSet> ExtensionSet::Make(
if (maybe_uri_internal) {
set.uris_[uri.first] = *maybe_uri_internal;
} else {
- if (kExactRoundTrip) {
+ if (conversion_options.strictness ==
ConversionStrictness::EXACT_ROUNDTRIP) {
return Status::Invalid(
"Plan contained a URI that the extension registry is unaware of: ",
uri.second);
@@ -249,7 +247,7 @@ Result<ExtensionSet> ExtensionSet::Make(
if (maybe_id_internal) {
set.functions_[function_id.first] = *maybe_id_internal;
} else {
- if (kExactRoundTrip) {
+ if (conversion_options.strictness ==
ConversionStrictness::EXACT_ROUNDTRIP) {
return Status::Invalid(
"Plan contained a function id that the extension registry is
unaware of: ",
function_id.second.uri, "#", function_id.second.name);
@@ -532,6 +530,21 @@ struct ExtensionIdRegistryImpl : ExtensionIdRegistry {
return maybe_converter->second;
}
+ Result<SubstraitCallToArrow> GetSubstraitCallToArrowFallback(
+ std::string_view function_name) const override {
+ for (const auto& converter_item : substrait_to_arrow_) {
+ if (converter_item.first.name == function_name) {
+ return converter_item.second;
+ }
+ }
+ if (parent_) {
+ return parent_->GetSubstraitCallToArrowFallback(function_name);
+ }
+ return Status::NotImplemented(
+ "No conversion function exists to convert the Substrait function ",
function_name,
+ " to an Arrow call expression");
+ }
+
Result<SubstraitAggregateToArrow> GetSubstraitAggregateToArrow(
Id substrait_function_id) const override {
auto maybe_converter = substrait_to_arrow_agg_.find(substrait_function_id);
@@ -547,6 +560,21 @@ struct ExtensionIdRegistryImpl : ExtensionIdRegistry {
return maybe_converter->second;
}
+ Result<SubstraitAggregateToArrow> GetSubstraitAggregateToArrowFallback(
+ std::string_view function_name) const override {
+ for (const auto& converter_item : substrait_to_arrow_agg_) {
+ if (converter_item.first.name == function_name) {
+ return converter_item.second;
+ }
+ }
+ if (parent_) {
+ return parent_->GetSubstraitAggregateToArrowFallback(function_name);
+ }
+ return Status::NotImplemented(
+ "No conversion function exists to convert the Substrait aggregate
function ",
+ function_name, " to an Arrow call expression");
+ }
+
Result<ArrowToSubstraitCall> GetArrowToSubstraitCall(
const std::string& arrow_function_name) const override {
auto maybe_converter = arrow_to_substrait_.find(arrow_function_name);
diff --git a/cpp/src/arrow/engine/substrait/extension_set.h
b/cpp/src/arrow/engine/substrait/extension_set.h
index 4df8952ff9..e2e09d0d92 100644
--- a/cpp/src/arrow/engine/substrait/extension_set.h
+++ b/cpp/src/arrow/engine/substrait/extension_set.h
@@ -32,6 +32,7 @@
#include "arrow/compute/api_aggregate.h"
#include "arrow/compute/exec/expression.h"
+#include "arrow/engine/substrait/options.h"
#include "arrow/engine/substrait/visibility.h"
#include "arrow/result.h"
#include "arrow/type_fwd.h"
@@ -253,6 +254,20 @@ class ARROW_ENGINE_EXPORT ExtensionIdRegistry {
/// \return A converter function or an invalid status if no converter is
registered
virtual Result<SubstraitCallToArrow> GetSubstraitCallToArrow(
Id substrait_function_id) const = 0;
+
+ /// \brief Similar to \see GetSubstraitCallToArrow but only uses the name
+ ///
+ /// There may be multiple functions with the same name and this will return
+ /// the first. This is slower than GetSubstraitCallToArrow and should only
+ /// be used when the plan does not include a URI (or the URI is "/")
+ virtual Result<SubstraitCallToArrow> GetSubstraitCallToArrowFallback(
+ std::string_view function_name) const = 0;
+
+ /// \brief Similar to \see GetSubstraitAggregateToArrow but only uses the
name
+ ///
+ /// \see GetSubstraitCallToArrowFallback for details on the fallback behavior
+ virtual Result<SubstraitAggregateToArrow>
GetSubstraitAggregateToArrowFallback(
+ std::string_view function_name) const = 0;
};
constexpr std::string_view kArrowExtTypesUri =
@@ -339,6 +354,7 @@ class ARROW_ENGINE_EXPORT ExtensionSet {
std::unordered_map<uint32_t, std::string_view> uris,
std::unordered_map<uint32_t, Id> type_ids,
std::unordered_map<uint32_t, Id> function_ids,
+ const ConversionOptions& conversion_options,
const ExtensionIdRegistry* = default_extension_id_registry());
const std::unordered_map<uint32_t, std::string_view>& uris() const { return
uris_; }
diff --git a/cpp/src/arrow/engine/substrait/plan_internal.cc
b/cpp/src/arrow/engine/substrait/plan_internal.cc
index 81dc6c2103..18915868ee 100644
--- a/cpp/src/arrow/engine/substrait/plan_internal.cc
+++ b/cpp/src/arrow/engine/substrait/plan_internal.cc
@@ -88,6 +88,7 @@ Status AddExtensionSetToPlan(const ExtensionSet& ext_set,
substrait::Plan* plan)
}
Result<ExtensionSet> GetExtensionSetFromPlan(const substrait::Plan& plan,
+ const ConversionOptions&
conversion_options,
const ExtensionIdRegistry*
registry) {
if (registry == NULLPTR) {
registry = default_extension_id_registry();
@@ -128,7 +129,7 @@ Result<ExtensionSet> GetExtensionSetFromPlan(const
substrait::Plan& plan,
}
return ExtensionSet::Make(std::move(uris), std::move(type_ids),
std::move(function_ids),
- registry);
+ conversion_options, registry);
}
Result<std::unique_ptr<substrait::Plan>> PlanToProto(
diff --git a/cpp/src/arrow/engine/substrait/plan_internal.h
b/cpp/src/arrow/engine/substrait/plan_internal.h
index e1ced549ce..7d6dd37528 100644
--- a/cpp/src/arrow/engine/substrait/plan_internal.h
+++ b/cpp/src/arrow/engine/substrait/plan_internal.h
@@ -50,7 +50,7 @@ Status AddExtensionSetToPlan(const ExtensionSet& ext_set,
substrait::Plan* plan)
/// correspond to Substrait's URI/name pairs
ARROW_ENGINE_EXPORT
Result<ExtensionSet> GetExtensionSetFromPlan(
- const substrait::Plan& plan,
+ const substrait::Plan& plan, const ConversionOptions& conversion_options,
const ExtensionIdRegistry* registry = default_extension_id_registry());
/// \brief Serialize a declaration into a substrait::Plan.
diff --git a/cpp/src/arrow/engine/substrait/relation_internal.cc
b/cpp/src/arrow/engine/substrait/relation_internal.cc
index 06c4adaba3..2afe177eb1 100644
--- a/cpp/src/arrow/engine/substrait/relation_internal.cc
+++ b/cpp/src/arrow/engine/substrait/relation_internal.cc
@@ -505,9 +505,16 @@ Result<DeclarationInfo> FromProto(const substrait::Rel&
rel, const ExtensionSet&
ARROW_ASSIGN_OR_RAISE(
SubstraitCall aggregate_call,
FromProto(agg_func, !keys.empty(), ext_set, conversion_options));
- ARROW_ASSIGN_OR_RAISE(
- ExtensionIdRegistry::SubstraitAggregateToArrow converter,
-
ext_set.registry()->GetSubstraitAggregateToArrow(aggregate_call.id()));
+ ExtensionIdRegistry::SubstraitAggregateToArrow converter;
+ if (aggregate_call.id().uri.empty() || aggregate_call.id().uri[0] ==
'/') {
+ ARROW_ASSIGN_OR_RAISE(
+ converter,
ext_set.registry()->GetSubstraitAggregateToArrowFallback(
+ aggregate_call.id().name));
+ } else {
+ ARROW_ASSIGN_OR_RAISE(
+ converter,
+
ext_set.registry()->GetSubstraitAggregateToArrow(aggregate_call.id()));
+ }
ARROW_ASSIGN_OR_RAISE(compute::Aggregate arrow_agg,
converter(aggregate_call));
// find aggregate field ids from schema
diff --git a/cpp/src/arrow/engine/substrait/serde.cc
b/cpp/src/arrow/engine/substrait/serde.cc
index c7792c7c76..b2416aa7ba 100644
--- a/cpp/src/arrow/engine/substrait/serde.cc
+++ b/cpp/src/arrow/engine/substrait/serde.cc
@@ -145,7 +145,8 @@ Result<std::vector<compute::Declaration>> DeserializePlans(
const ConversionOptions& conversion_options) {
ARROW_ASSIGN_OR_RAISE(auto plan, ParseFromBuffer<substrait::Plan>(buf));
- ARROW_ASSIGN_OR_RAISE(auto ext_set, GetExtensionSetFromPlan(plan, registry));
+ ARROW_ASSIGN_OR_RAISE(auto ext_set,
+ GetExtensionSetFromPlan(plan, conversion_options,
registry));
std::vector<compute::Declaration> sink_decls;
for (const substrait::PlanRel& plan_rel : plan.relations()) {
diff --git a/cpp/src/arrow/engine/substrait/serde_test.cc
b/cpp/src/arrow/engine/substrait/serde_test.cc
index de40a53cbc..564d68f360 100644
--- a/cpp/src/arrow/engine/substrait/serde_test.cc
+++ b/cpp/src/arrow/engine/substrait/serde_test.cc
@@ -95,7 +95,10 @@ Result<std::shared_ptr<Table>> GetTableFromPlan(
RETURN_NOT_OK(plan->Validate());
RETURN_NOT_OK(plan->StartProducing());
- return arrow::Table::FromRecordBatchReader(sink_reader.get());
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Table> table,
+
arrow::Table::FromRecordBatchReader(sink_reader.get()));
+ RETURN_NOT_OK(plan->finished().status());
+ return table;
}
class NullSinkNodeConsumer : public compute::SinkNodeConsumer {
@@ -194,7 +197,7 @@ void CheckRoundTripResult(const std::shared_ptr<Schema>
output_schema,
ASSERT_OK_AND_ASSIGN(output_table,
output_table->SelectColumns(include_columns));
}
ASSERT_OK_AND_ASSIGN(output_table, output_table->CombineChunks());
- EXPECT_TRUE(expected_table->Equals(*output_table));
+ AssertTablesEqual(*expected_table, *output_table);
}
TEST(Substrait, SupportedTypes) {
@@ -843,8 +846,13 @@ TEST(Substrait, ExtensionSetFromPlanMissingFunc) {
{std::shared_ptr<ExtensionIdRegistry>(), MakeExtensionIdRegistry()}) {
ExtensionIdRegistry* ext_id_reg = sp_ext_id_reg.get();
ExtensionSet ext_set(ext_id_reg);
- ASSERT_RAISES(Invalid, DeserializePlans(
- *buf, [] { return kNullConsumer; }, ext_id_reg,
&ext_set));
+ // Since the function is not referenced this plan is ok unless we are
asking for
+ // strict conversion.
+ ConversionOptions options;
+ options.strictness = ConversionStrictness::EXACT_ROUNDTRIP;
+ ASSERT_RAISES(Invalid,
+ DeserializePlans(
+ *buf, [] { return kNullConsumer; }, ext_id_reg,
&ext_set, options));
}
}
@@ -921,16 +929,19 @@ TEST(Substrait, ExtensionSetFromPlanRegisterFunc) {
ExtensionIdRegistry* ext_id_reg = sp_ext_id_reg.get();
// invalid before registration
ExtensionSet ext_set_invalid(ext_id_reg);
- ASSERT_RAISES(Invalid,
- DeserializePlans(
- *buf, [] { return kNullConsumer; }, ext_id_reg,
&ext_set_invalid));
+ ConversionOptions conversion_options;
+ conversion_options.strictness = ConversionStrictness::EXACT_ROUNDTRIP;
+ ASSERT_RAISES(Invalid, DeserializePlans(
+ *buf, [] { return kNullConsumer; }, ext_id_reg,
+ &ext_set_invalid, conversion_options));
ASSERT_OK(ext_id_reg->AddSubstraitCallToArrow(
{default_extension_types_uri(), "new_func"}, "multiply"));
// valid after registration
ExtensionSet ext_set_valid(ext_id_reg);
- ASSERT_OK_AND_ASSIGN(auto sink_decls, DeserializePlans(
- *buf, [] { return kNullConsumer; },
- ext_id_reg, &ext_set_valid));
+ ASSERT_OK_AND_ASSIGN(auto sink_decls,
+ DeserializePlans(
+ *buf, [] { return kNullConsumer; }, ext_id_reg,
&ext_set_valid,
+ conversion_options));
EXPECT_OK_AND_ASSIGN(Id decoded_add_func_id,
ext_set_valid.DecodeFunction(42));
EXPECT_EQ(decoded_add_func_id.uri, kArrowExtTypesUri);
EXPECT_EQ(decoded_add_func_id.name, "new_func");
@@ -2121,6 +2132,15 @@ TEST(Substrait, BasicPlanRoundTrippingEndToEnd) {
EXPECT_TRUE(expected_table->Equals(*rnd_trp_table));
}
+/// \brief Create a NamedTableProvider that provides `table` regardless of the
name
+NamedTableProvider AlwaysProvideSameTable(std::shared_ptr<Table> table) {
+ return [table = std::move(table)](const std::vector<std::string>&) {
+ std::shared_ptr<compute::ExecNodeOptions> options =
+ std::make_shared<compute::TableSourceNodeOptions>(table);
+ return compute::Declaration("table_source", {}, options, "mock_source");
+ };
+}
+
TEST(Substrait, ProjectRel) {
#ifdef _WIN32
GTEST_SKIP() << "ARROW-16392: Substrait File URI not supported for Windows";
@@ -2227,11 +2247,7 @@ TEST(Substrait, ProjectRel) {
[2, 2, 60, true]
])"});
- NamedTableProvider table_provider = [input_table](const
std::vector<std::string>&) {
- std::shared_ptr<compute::ExecNodeOptions> options =
- std::make_shared<compute::TableSourceNodeOptions>(input_table);
- return compute::Declaration("table_source", {}, options, "mock_source");
- };
+ NamedTableProvider table_provider =
AlwaysProvideSameTable(std::move(input_table));
ConversionOptions conversion_options;
conversion_options.named_table_provider = std::move(table_provider);
@@ -2350,11 +2366,7 @@ TEST(Substrait, ProjectRelOnFunctionWithEmit) {
[5, 50, true],
[2, 60, true]
])"});
- NamedTableProvider table_provider = [input_table](const
std::vector<std::string>&) {
- std::shared_ptr<compute::ExecNodeOptions> options =
- std::make_shared<compute::TableSourceNodeOptions>(input_table);
- return compute::Declaration("table_source", {}, options, "mock_source");
- };
+ NamedTableProvider table_provider =
AlwaysProvideSameTable(std::move(input_table));
ConversionOptions conversion_options;
conversion_options.named_table_provider = std::move(table_provider);
@@ -2413,11 +2425,7 @@ TEST(Substrait, ReadRelWithEmit) {
[4, 20]
])"});
- NamedTableProvider table_provider = [input_table](const
std::vector<std::string>&) {
- std::shared_ptr<compute::ExecNodeOptions> options =
- std::make_shared<compute::TableSourceNodeOptions>(input_table);
- return compute::Declaration("table_source", {}, options, "mock_source");
- };
+ NamedTableProvider table_provider =
AlwaysProvideSameTable(std::move(input_table));
ConversionOptions conversion_options;
conversion_options.named_table_provider = std::move(table_provider);
@@ -2535,11 +2543,7 @@ TEST(Substrait, FilterRelWithEmit) {
[6, 2],
[7, 1]
])"});
- NamedTableProvider table_provider = [input_table](const
std::vector<std::string>&) {
- std::shared_ptr<compute::ExecNodeOptions> options =
- std::make_shared<compute::TableSourceNodeOptions>(input_table);
- return compute::Declaration("table_source", {}, options, "mock_source");
- };
+ NamedTableProvider table_provider =
AlwaysProvideSameTable(std::move(input_table));
ConversionOptions conversion_options;
conversion_options.named_table_provider = std::move(table_provider);
@@ -2950,11 +2954,7 @@ TEST(Substrait, AggregateRel) {
[60, 40]
])"});
- NamedTableProvider table_provider = [input_table](const
std::vector<std::string>&) {
- std::shared_ptr<compute::ExecNodeOptions> options =
- std::make_shared<compute::TableSourceNodeOptions>(input_table);
- return compute::Declaration("table_source", {}, options, "mock_source");
- };
+ NamedTableProvider table_provider =
AlwaysProvideSameTable(std::move(input_table));
ConversionOptions conversion_options;
conversion_options.named_table_provider = std::move(table_provider);
@@ -3069,11 +3069,7 @@ TEST(Substrait, AggregateRelEmit) {
[60]
])"});
- NamedTableProvider table_provider = [input_table](const
std::vector<std::string>&) {
- std::shared_ptr<compute::ExecNodeOptions> options =
- std::make_shared<compute::TableSourceNodeOptions>(input_table);
- return compute::Declaration("table_source", {}, options, "mock_source");
- };
+ NamedTableProvider table_provider =
AlwaysProvideSameTable(std::move(input_table));
ConversionOptions conversion_options;
conversion_options.named_table_provider = std::move(table_provider);
@@ -3082,5 +3078,114 @@ TEST(Substrait, AggregateRelEmit) {
buf, {}, conversion_options);
}
+TEST(Substrait, IsthmusPlan) {
+ // This is a plan generated from Isthmus
+ // isthmus -c "CREATE TABLE T1(foo int)" "SELECT foo + 1 FROM T1"
+ //
+ // The plan had to be modified slightly to introduce the missing enum
+ // argument that isthmus did not put there.
+ std::string substrait_json = R"({
+ "extensionUris": [{
+ "extensionUriAnchor": 1,
+ "uri": "/functions_arithmetic.yaml"
+ }],
+ "extensions": [{
+ "extensionFunction": {
+ "extensionUriReference": 1,
+ "functionAnchor": 0,
+ "name": "add:opt_i32_i32"
+ }
+ }],
+ "relations": [{
+ "root": {
+ "input": {
+ "project": {
+ "common": {
+ "emit": {
+ "outputMapping": [1]
+ }
+ },
+ "input": {
+ "read": {
+ "common": {
+ "direct": {
+ }
+ },
+ "baseSchema": {
+ "names": ["FOO"],
+ "struct": {
+ "types": [{
+ "i32": {
+ "typeVariationReference": 0,
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ }],
+ "typeVariationReference": 0,
+ "nullability": "NULLABILITY_REQUIRED"
+ }
+ },
+ "namedTable": {
+ "names": ["T1"]
+ }
+ }
+ },
+ "expressions": [{
+ "scalarFunction": {
+ "functionReference": 0,
+ "args": [],
+ "outputType": {
+ "i32": {
+ "typeVariationReference": 0,
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ },
+ "arguments": [{
+ "enum": {
+ "unspecified": {}
+ }
+ }, {
+ "value": {
+ "selection": {
+ "directReference": {
+ "structField": {
+ "field": 0
+ }
+ },
+ "rootReference": {
+ }
+ }
+ }
+ }, {
+ "value": {
+ "literal": {
+ "i32": 1,
+ "nullable": false,
+ "typeVariationReference": 0
+ }
+ }
+ }]
+ }
+ }]
+ }
+ },
+ "names": ["EXPR$0"]
+ }
+ }],
+ "expectedTypeUrls": []
+ })";
+
+ auto test_schema = schema({field("foo", int32())});
+ auto input_table = TableFromJSON(test_schema, {"[[1], [2], [5]]"});
+ NamedTableProvider table_provider =
AlwaysProvideSameTable(std::move(input_table));
+ ConversionOptions conversion_options;
+ conversion_options.named_table_provider = std::move(table_provider);
+
+ ASSERT_OK_AND_ASSIGN(auto buf, internal::SubstraitFromJSON("Plan",
substrait_json));
+
+ auto expected_table = TableFromJSON(test_schema, {"[[2], [3], [6]]"});
+ CheckRoundTripResult(std::move(test_schema), std::move(expected_table),
+ *compute::default_exec_context(), buf, {},
conversion_options);
+}
+
} // namespace engine
} // namespace arrow
