You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by GitBox <gi...@apache.org> on 2022/09/01 16:04:22 UTC
[GitHub] [arrow] vibhatha opened a new pull request, #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
vibhatha opened a new pull request, #14024:
URL: https://github.com/apache/arrow/pull/14024
This PR includes a basic prototype WIP version to use NamedTable feature in Substrait. The idea is to provide the flexibility to write Python tests with in-memory PyArrow tables.
The test case here only covers the basic functionality. There is another level of refactor to integrate this with the existing API. Drafted for the purpose of testing with third-party applications.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r966666200
##########
python/pyarrow/_substrait.pyx:
##########
@@ -17,22 +17,99 @@
# cython: language_level = 3
from cython.operator cimport dereference as deref
+from libcpp.vector cimport vector as std_vector
from pyarrow import Buffer
-from pyarrow.lib import frombytes
+from pyarrow.lib import frombytes, tobytes
from pyarrow.lib cimport *
from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef CDeclaration _create_named_table_provider(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] c_in_table
+ shared_ptr[CTableSourceNodeOptions] c_tablesourceopts
+ shared_ptr[CExecNodeOptions] c_input_node_opts
+ vector[CDeclaration.Input] no_c_inputs
+
+ py_names = []
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+
+ py_table = named_args["provider"](py_names)
+ c_in_table = pyarrow_unwrap_table(py_table)
+ c_tablesourceopts = make_shared[CTableSourceNodeOptions](
+ c_in_table, 1 << 20)
Review Comment:
@westonpace I left this number by looking into an existing binding. What is the best to do here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r967204565
##########
cpp/src/arrow/compute/exec/exec_plan.cc:
##########
@@ -643,6 +643,10 @@ Declaration Declaration::Sequence(std::vector<Declaration> decls) {
return out;
}
+bool Declaration::IsValid(ExecFactoryRegistry* registry) const {
+ return registry->GetFactory(this->factory_name).ok() && this->options != nullptr;
Review Comment:
Understood. I agree, without the plan it is not possible to do the validation properly. Should we create a JIRA to address this concretely in future?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on PR #14024:
URL: https://github.com/apache/arrow/pull/14024#issuecomment-1236474588
@westonpace any more thoughts on this one? Should we add more tests?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
westonpace commented on PR #14024:
URL: https://github.com/apache/arrow/pull/14024#issuecomment-1236472120
> @westonpace one doubt is handling multiple names in the provider. Assume we get multiple tables, how should we generate the output?
Each call to the function should return a single table.
If multiple names are provided then they form a single combined name. I'm not really sure why it is structured that way but I think the idea is a that tablenames are often namespaces, for example `["sales", "orders"]` is a single table `sales.orders` and not two different tables.
A provider function is free to know about as many tables as it wants but it can only return one table each call.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r961349699
##########
python/pyarrow/_substrait.pyx:
##########
@@ -102,3 +103,40 @@ def get_supported_functions():
for c_id in c_ids:
functions_list.append(frombytes(c_id))
return functions_list
+
+
+cdef shared_ptr[CTable] _process_named_table(dict named_args, const std_vector[c_string]& names):
+ cdef c_string c_name
+ py_names = []
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+ return pyarrow_unwrap_table(named_args["provider"](py_names))
+
+
+def run_query_with_provider(plan, table_provider):
Review Comment:
Of course, this was just a prototype code to show case for test purpose in a short time. I already refactored the code. I think it covers what you are asking for.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r961196591
##########
cpp/src/arrow/engine/substrait/util.cc:
##########
@@ -126,6 +131,36 @@ Result<std::shared_ptr<RecordBatchReader>> ExecuteSerializedPlan(
return sink_reader;
}
+Result<std::shared_ptr<RecordBatchReader>> ExecuteSerializedPlan(
+ const Buffer& substrait_buffer, PythonTableProvider& table_provider,
+ const ExtensionIdRegistry* registry, compute::FunctionRegistry* func_registry) {
+ // TODO(ARROW-15732)
+ // retrieve input table from table provider
+
+ NamedTableProvider named_table_provider =
+ [table_provider](
+ const std::vector<std::string>& names) -> Result<compute::Declaration> {
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Table> input_table, table_provider(names));
+ std::shared_ptr<compute::ExecNodeOptions> options =
+ std::make_shared<compute::TableSourceNodeOptions>(input_table);
+ return compute::Declaration("table_source", {}, options,
+ "substrait_table_provider_source");
+ };
+
+ ConversionOptions conversion_options;
+ conversion_options.named_table_provider = std::move(named_table_provider);
+
+ compute::ExecContext exec_context(arrow::default_memory_pool(),
+ ::arrow::internal::GetCpuThreadPool(), func_registry);
+ ARROW_ASSIGN_OR_RAISE(auto plan, compute::ExecPlan::Make(&exec_context));
+ SubstraitExecutor executor(std::move(plan), exec_context, conversion_options);
+ RETURN_NOT_OK(executor.Init(substrait_buffer, registry));
+ ARROW_ASSIGN_OR_RAISE(auto sink_reader, executor.Execute());
+ // check closing here, not in destructor, to expose error to caller
+ RETURN_NOT_OK(executor.Close());
+ return sink_reader;
Review Comment:
Rather than copy / paste can we call the other method here?
##########
python/pyarrow/_substrait.pyx:
##########
@@ -102,3 +103,40 @@ def get_supported_functions():
for c_id in c_ids:
functions_list.append(frombytes(c_id))
return functions_list
+
+
+cdef shared_ptr[CTable] _process_named_table(dict named_args, const std_vector[c_string]& names):
+ cdef c_string c_name
+ py_names = []
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+ return pyarrow_unwrap_table(named_args["provider"](py_names))
+
+
+def run_query_with_provider(plan, table_provider):
Review Comment:
Can we make `table_provider` an optional argument for `run_query` rather than adding a new method? I don't know if we can do it in cython (I think we can for a `def` method) or not but ideally the python API wouldn't need two methods.
##########
cpp/src/arrow/engine/substrait/util.h:
##########
@@ -20,18 +20,29 @@
#include <memory>
#include "arrow/compute/registry.h"
#include "arrow/engine/substrait/api.h"
+#include "arrow/engine/substrait/options.h"
#include "arrow/util/iterator.h"
#include "arrow/util/optional.h"
namespace arrow {
namespace engine {
+using PythonTableProvider =
+ std::function<Result<std::shared_ptr<Table>>(const std::vector<std::string>&)>;
+
/// \brief Retrieve a RecordBatchReader from a Substrait plan.
ARROW_ENGINE_EXPORT Result<std::shared_ptr<RecordBatchReader>> ExecuteSerializedPlan(
const Buffer& substrait_buffer, const ExtensionIdRegistry* registry = NULLPTR,
compute::FunctionRegistry* func_registry = NULLPTR);
+/// \brief Retrieve a RecordBatchReader from a Substrait plan.
+/// Allows to use NamedT
Review Comment:
Finish this comment?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r969893499
##########
python/pyarrow/includes/libarrow_substrait.pxd:
##########
@@ -22,10 +22,24 @@ from libcpp.vector cimport vector as std_vector
from pyarrow.includes.common cimport *
from pyarrow.includes.libarrow cimport *
+# ctypedef CResult[shared_ptr[CTable]] named_table_provider(const std_vector[c_string]&)
Review Comment:
```suggestion
```
##########
cpp/src/arrow/compute/exec/exec_plan.cc:
##########
@@ -643,6 +643,10 @@ Declaration Declaration::Sequence(std::vector<Declaration> decls) {
return out;
}
+bool Declaration::IsValid(ExecFactoryRegistry* registry) const {
+ return registry->GetFactory(this->factory_name).ok() && this->options != nullptr;
Review Comment:
```suggestion
return !this->factory_name.empty() && this->options != nullptr;
```
I think the earlier advice I provided was slightly off. I think, all we need to do here is distinguish between "default-constructed invalid declarations" and "properly constructed declarations". Actual validation (e.g. hash-map lookup in the factory) doesn't really belong in this method as it should be really fast.
##########
python/pyarrow/_substrait.pyx:
##########
@@ -17,22 +17,98 @@
# cython: language_level = 3
from cython.operator cimport dereference as deref
+from libcpp.vector cimport vector as std_vector
from pyarrow import Buffer
-from pyarrow.lib import frombytes
+from pyarrow.lib import frombytes, tobytes
from pyarrow.lib cimport *
from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef CDeclaration _create_named_table_provider(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] c_in_table
+ shared_ptr[CTableSourceNodeOptions] c_tablesourceopts
+ shared_ptr[CExecNodeOptions] c_input_node_opts
+ vector[CDeclaration.Input] no_c_inputs
+
+ py_names = []
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+
+ py_table = named_args["provider"](py_names)
+ c_in_table = pyarrow_unwrap_table(py_table)
+ c_tablesourceopts = make_shared[CTableSourceNodeOptions](c_in_table)
+ c_input_node_opts = static_pointer_cast[CExecNodeOptions, CTableSourceNodeOptions](
+ c_tablesourceopts)
+ return CDeclaration(tobytes("table_source"),
+ no_c_inputs, c_input_node_opts)
+
+
+def run_query(plan, table_provider=None):
"""
Execute a Substrait plan and read the results as a RecordBatchReader.
Parameters
----------
plan : Buffer
The serialized Substrait plan to execute.
+ table_provider : object (optional)
+ A function to resolve any NamedTable relation to a table.
+ The function will receive a single argument which will be a list
+ of strings representing the table name and should return a pyarrow.Table.
+
+ Returns
+ -------
+ RecordBatchReader
+ A reader containing the result of the executed query
+
+ Examples
+ --------
+ >>> import pyarrow as pa
+ >>> from pyarrow.lib import tobytes
+ >>> import pyarrow.substrait as substrait
+ >>> test_table_1 = pa.Table.from_pydict({"x": [1, 2, 3]})
+ >>> test_table_2 = pa.Table.from_pydict({"x": [4, 5, 6]})
+ >>> def table_provider(names):
+ ... if not names:
+ ... raise Exception("No names provided")
+ ... elif names[0] == "t1":
+ ... return test_table_1
+ ... elif names[1] == "t2":
+ ... return test_table_2
+ ... else:
+ ... raise Exception("Unrecognized table name")
+ ...
+ >>> substrait_query = '''
+ ... {
+ ... "relations": [
+ ... {"rel": {
+ ... "read": {
+ ... "base_schema": {
+ ... "struct": {
+ ... "types": [
+ ... {"i64": {}}
+ ... ]
+ ... },
+ ... "names": [
+ ... "x"
+ ... ]
+ ... },
+ ... "namedTable": {
+ ... "names": ["t1"]
+ ... }
+ ... }
+ ... }}
+ ... ]
+ ... }
+ ... '''
+ >>> buf = pa._substrait._parse_json_plan(tobytes(substrait_query))
+ >>> pa.substrait.run_query_with_provider(buf, table_provider)
+ <pyarrow.lib.RecordBatchReader object at 0x100dcaf10>
Review Comment:
Do these examples get verified at build time? If so I don't think this line `0x100dcaf10` will be very reliable. If not, I suppose it is ok, but it is odd to see a random pointer value in docs. We could maybe just omit this line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] ursabot commented on pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
ursabot commented on PR #14024:
URL: https://github.com/apache/arrow/pull/14024#issuecomment-1248253540
Benchmark runs are scheduled for baseline = eb00620bdeb3bebd483b0205092d617b7b3552d4 and contender = 72b539f54233f6610b01ec7381755a84c652d151. 72b539f54233f6610b01ec7381755a84c652d151 is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Finished :arrow_down:0.0% :arrow_up:0.0%] [ec2-t3-xlarge-us-east-2](https://conbench.ursa.dev/compare/runs/ded2587653514a0b842a077dcfa0f230...bd3e4814f05642a9a447cc47803e9d5e/)
[Failed :arrow_down:0.71% :arrow_up:0.0%] [test-mac-arm](https://conbench.ursa.dev/compare/runs/05b9b4d5de9e4108b41a576933ccdd9e...2001963899e1420e8b5e48f33b99723b/)
[Failed :arrow_down:0.0% :arrow_up:0.0%] [ursa-i9-9960x](https://conbench.ursa.dev/compare/runs/bbf46415990642bdad7210d97a63af1e...74f81ab98a7c47be9f23ab221a52983a/)
[Finished :arrow_down:0.39% :arrow_up:0.04%] [ursa-thinkcentre-m75q](https://conbench.ursa.dev/compare/runs/255d961631c24594a92ac35e5f1a6546...5bbf70e1f7bc4cc39d90730cf18f26c7/)
Buildkite builds:
[Finished] [`72b539f5` ec2-t3-xlarge-us-east-2](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ec2-t3-xlarge-us-east-2/builds/1482)
[Failed] [`72b539f5` test-mac-arm](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-test-mac-arm/builds/1498)
[Failed] [`72b539f5` ursa-i9-9960x](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ursa-i9-9960x/builds/1482)
[Finished] [`72b539f5` ursa-thinkcentre-m75q](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ursa-thinkcentre-m75q/builds/1495)
[Finished] [`eb00620b` ec2-t3-xlarge-us-east-2](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ec2-t3-xlarge-us-east-2/builds/1481)
[Finished] [`eb00620b` test-mac-arm](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-test-mac-arm/builds/1497)
[Failed] [`eb00620b` ursa-i9-9960x](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ursa-i9-9960x/builds/1481)
[Finished] [`eb00620b` ursa-thinkcentre-m75q](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ursa-thinkcentre-m75q/builds/1494)
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r961978236
##########
python/pyarrow/tests/test_substrait.py:
##########
@@ -165,3 +165,172 @@ def test_get_supported_functions():
'functions_arithmetic.yaml', 'add')
assert has_function(supported_functions,
'functions_arithmetic.yaml', 'sum')
+
+
+def test_named_table():
+ test_table_1 = pa.Table.from_pydict({"x": [1, 2, 3]})
+ test_table_2 = pa.Table.from_pydict({"x": [4, 5, 6]})
+
+ def table_provider(names):
+ if not names:
+ raise Exception("No names provided")
+ elif names[0] == "t1":
+ return test_table_1
+ elif names[1] == "t2":
+ return test_table_2
+ else:
+ raise Exception("Unrecognized table name")
+
+ substrait_query = """
+ {
+ "relations": [
+ {"rel": {
+ "read": {
+ "base_schema": {
+ "struct": {
+ "types": [
+ {"i64": {}}
+ ]
+ },
+ "names": [
+ "x"
+ ]
+ },
+ "namedTable": {
+ "names": ["TABLE_NAME_PLACEHOLDER"]
+ }
+ }
+ }}
+ ]
+ }
+ """
+ table_name = "t1"
+ query = tobytes(substrait_query.replace(
+ "TABLE_NAME_PLACEHOLDER", table_name))
+ buf = pa._substrait._parse_json_plan(tobytes(query))
+ reader = pa.substrait.run_query(buf, table_provider)
+ res_tb = reader.read_all()
+ assert res_tb == test_table_1
+
+
+def test_named_table_invalid_table_name():
+ test_table_1 = pa.Table.from_pydict({"x": [1, 2, 3]})
+
+ def table_provider(names):
+ if not names:
+ raise Exception("No names provided")
+ elif names[0] == "t1":
+ return test_table_1
+ else:
+ raise Exception("Unrecognized table name")
+
+ substrait_query = """
+ {
+ "relations": [
+ {"rel": {
+ "read": {
+ "base_schema": {
+ "struct": {
+ "types": [
+ {"i64": {}}
+ ]
+ },
+ "names": [
+ "x"
+ ]
+ },
+ "namedTable": {
+ "names": ["TABLE_NAME_PLACEHOLDER"]
Review Comment:
No need to use a placeholder.
##########
cpp/src/arrow/engine/substrait/util.h:
##########
@@ -20,16 +20,22 @@
#include <memory>
#include "arrow/compute/registry.h"
#include "arrow/engine/substrait/api.h"
+#include "arrow/engine/substrait/options.h"
#include "arrow/util/iterator.h"
#include "arrow/util/optional.h"
namespace arrow {
namespace engine {
+using PythonTableProvider =
+ std::function<Result<std::shared_ptr<Table>>(const std::vector<std::string>&)>;
+
/// \brief Retrieve a RecordBatchReader from a Substrait plan.
+/// Allows to use NamedTableProvider in Substrait plan execution.
Review Comment:
```suggestion
```
If we are going to document one param we should document all the params
##########
python/pyarrow/_substrait.pyx:
##########
@@ -25,14 +26,84 @@ from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef shared_ptr[CTable] _process_named_table(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] empty_table
+ py_names = []
+
+ # provider function could raise an exception
+ try:
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+ return pyarrow_unwrap_table(named_args["provider"](py_names))
+ except Exception:
+ return empty_table
+
+
+def run_query(plan, table_provider=None):
"""
Execute a Substrait plan and read the results as a RecordBatchReader.
Parameters
----------
plan : Buffer
The serialized Substrait plan to execute.
+ table_provider : object (optional)
+ The function determining input table based on names
+
+ Returns
+ -------
+ RecordBatchReader
+ A reader containing the result of the executed query
+
+ Examples
+ --------
+ >>> import pyarrow as pa
+ >>> from pyarrow.lib import tobytes
+ >>> import pyarrow.substrait as substrait
+ >>> test_table_1 = pa.Table.from_pydict({"x": [1, 2, 3]})
+ >>> test_table_2 = pa.Table.from_pydict({"x": [4, 5, 6]})
+ >>> def table_provider(names):
+ ... if not names:
+ ... raise Exception("No names provided")
+ ... elif names[0] == "t1":
+ ... return test_table_1
+ ... elif names[1] == "t2":
+ ... return test_table_2
+ ... else:
+ ... raise Exception("Unrecognized table name")
+ ...
+ >>> substrait_query = '''
+ ... {
+ ... "relations": [
+ ... {"rel": {
+ ... "read": {
+ ... "base_schema": {
+ ... "struct": {
+ ... "types": [
+ ... {"i64": {}}
+ ... ]
+ ... },
+ ... "names": [
+ ... "x"
+ ... ]
+ ... },
+ ... "namedTable": {
+ ... "names": ["TABLE_NAME_PLACEHOLDER"]
+ ... }
+ ... }
+ ... }}
+ ... ]
+ ... }
+ ... '''
+ >>> table_name = "t1"
+ >>> query = tobytes(substrait_query.replace(
+ ... "TABLE_NAME_PLACEHOLDER", table_name))
Review Comment:
```suggestion
>>> substrait_query = '''
... {
... "relations": [
... {"rel": {
... "read": {
... "base_schema": {
... "struct": {
... "types": [
... {"i64": {}}
... ]
... },
... "names": [
... "x"
... ]
... },
... "namedTable": {
... "names": ["t1"]
... }
... }
... }}
... ]
... }
... '''
```
There is no need to use the placeholder pattern here.
##########
python/pyarrow/_substrait.pyx:
##########
@@ -25,14 +26,84 @@ from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef shared_ptr[CTable] _process_named_table(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] empty_table
+ py_names = []
+
+ # provider function could raise an exception
+ try:
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+ return pyarrow_unwrap_table(named_args["provider"](py_names))
+ except Exception:
+ return empty_table
+
+
+def run_query(plan, table_provider=None):
"""
Execute a Substrait plan and read the results as a RecordBatchReader.
Parameters
----------
plan : Buffer
The serialized Substrait plan to execute.
+ table_provider : object (optional)
+ The function determining input table based on names
Review Comment:
```suggestion
A function to resolve any NamedTable relation to a table. The function
will receive a single argument which will be a list of strings representing the table
name and should return a pyarrow.Table.
```
Presumably users of this API will be familiar with Substrait so we can reference `NamedTable` directly.
##########
cpp/src/arrow/engine/substrait/util.cc:
##########
@@ -107,19 +111,39 @@ class SubstraitExecutor {
bool plan_started_;
compute::ExecContext exec_context_;
std::shared_ptr<SubstraitSinkConsumer> sink_consumer_;
+ const ConversionOptions& conversion_options_;
};
} // namespace
Result<std::shared_ptr<RecordBatchReader>> ExecuteSerializedPlan(
- const Buffer& substrait_buffer, const ExtensionIdRegistry* extid_registry,
- compute::FunctionRegistry* func_registry) {
+ const Buffer& substrait_buffer, const PythonTableProvider& table_provider,
+ const ExtensionIdRegistry* registry, compute::FunctionRegistry* func_registry) {
// TODO(ARROW-15732)
+ // retrieve input table from table provider
+ ConversionOptions conversion_options;
+ if (table_provider) {
+ NamedTableProvider named_table_provider =
+ [table_provider](
+ const std::vector<std::string>& names) -> Result<compute::Declaration> {
+ if (names.empty()) {
+ return Status::Invalid("names for NamedTable not provided");
+ }
Review Comment:
It wasn't clear to me if a name was required or not so I added https://github.com/substrait-io/substrait/pull/308
If that merges then we are fine returning an invalid status here since this would be an invalid plan. However, we can probably reject it in `relation_internal.cc` instead so we don't have to repeat this logic for each binding.
##########
python/pyarrow/_substrait.pyx:
##########
@@ -25,14 +26,84 @@ from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef shared_ptr[CTable] _process_named_table(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] empty_table
+ py_names = []
+
+ # provider function could raise an exception
+ try:
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+ return pyarrow_unwrap_table(named_args["provider"](py_names))
+ except Exception:
+ return empty_table
Review Comment:
I don't know that an empty table is the correct thing to return here. Maybe we can return `CResult[shared_ptr[CTable]]` so that we can return an invalid status? Ideally we would capture the exception message to put into the status too. I'm not sure if we have an example of that.
##########
python/pyarrow/tests/test_substrait.py:
##########
@@ -165,3 +165,172 @@ def test_get_supported_functions():
'functions_arithmetic.yaml', 'add')
assert has_function(supported_functions,
'functions_arithmetic.yaml', 'sum')
+
+
+def test_named_table():
+ test_table_1 = pa.Table.from_pydict({"x": [1, 2, 3]})
+ test_table_2 = pa.Table.from_pydict({"x": [4, 5, 6]})
+
+ def table_provider(names):
+ if not names:
+ raise Exception("No names provided")
+ elif names[0] == "t1":
+ return test_table_1
+ elif names[1] == "t2":
+ return test_table_2
+ else:
+ raise Exception("Unrecognized table name")
+
+ substrait_query = """
+ {
+ "relations": [
+ {"rel": {
+ "read": {
+ "base_schema": {
+ "struct": {
+ "types": [
+ {"i64": {}}
+ ]
+ },
+ "names": [
+ "x"
+ ]
+ },
+ "namedTable": {
+ "names": ["TABLE_NAME_PLACEHOLDER"]
+ }
+ }
+ }}
+ ]
+ }
+ """
+ table_name = "t1"
+ query = tobytes(substrait_query.replace(
+ "TABLE_NAME_PLACEHOLDER", table_name))
+ buf = pa._substrait._parse_json_plan(tobytes(query))
+ reader = pa.substrait.run_query(buf, table_provider)
+ res_tb = reader.read_all()
+ assert res_tb == test_table_1
+
+
+def test_named_table_invalid_table_name():
+ test_table_1 = pa.Table.from_pydict({"x": [1, 2, 3]})
+
+ def table_provider(names):
+ if not names:
+ raise Exception("No names provided")
+ elif names[0] == "t1":
+ return test_table_1
+ else:
+ raise Exception("Unrecognized table name")
+
+ substrait_query = """
+ {
+ "relations": [
+ {"rel": {
+ "read": {
+ "base_schema": {
+ "struct": {
+ "types": [
+ {"i64": {}}
+ ]
+ },
+ "names": [
+ "x"
+ ]
+ },
+ "namedTable": {
+ "names": ["TABLE_NAME_PLACEHOLDER"]
+ }
+ }
+ }}
+ ]
+ }
+ """
+ table_name = "t3"
+ query = tobytes(substrait_query.replace(
+ "TABLE_NAME_PLACEHOLDER", table_name))
+ buf = pa._substrait._parse_json_plan(tobytes(query))
+
+ exec_message = "TableSourceNode requires table which is not null"
+ with pytest.raises(ArrowInvalid, match=exec_message):
+ substrait.run_query(buf, table_provider)
+
+
+def test_named_table_empty_names():
+ test_table_1 = pa.Table.from_pydict({"x": [1, 2, 3]})
+
+ def table_provider(names):
+ if not names:
+ raise Exception("No names provided")
+ elif names[0] == "t1":
+ return test_table_1
+ else:
+ raise Exception("Unrecognized table name")
+
+ substrait_query = """
+ {
+ "relations": [
+ {"rel": {
+ "read": {
+ "base_schema": {
+ "struct": {
+ "types": [
+ {"i64": {}}
+ ]
+ },
+ "names": [
+ "x"
+ ]
+ },
+ "namedTable": {
+ "names": []
+ }
+ }
+ }}
+ ]
+ }
+ """
+ query = tobytes(substrait_query)
+ buf = pa._substrait._parse_json_plan(tobytes(query))
+ exec_message = "names for NamedTable not provided"
+ with pytest.raises(ArrowInvalid, match=exec_message):
+ substrait.run_query(buf, table_provider)
+
+
+def test_named_table_no_named_table():
Review Comment:
I'm not sure how much value this test is providing. At the very least, it doesn't seem to be related to named tables at all and is more just testing the C++ consumer's ability to reject invalid plans.
##########
python/pyarrow/tests/test_substrait.py:
##########
@@ -165,3 +165,172 @@ def test_get_supported_functions():
'functions_arithmetic.yaml', 'add')
assert has_function(supported_functions,
'functions_arithmetic.yaml', 'sum')
+
+
+def test_named_table():
+ test_table_1 = pa.Table.from_pydict({"x": [1, 2, 3]})
+ test_table_2 = pa.Table.from_pydict({"x": [4, 5, 6]})
+
+ def table_provider(names):
+ if not names:
+ raise Exception("No names provided")
+ elif names[0] == "t1":
+ return test_table_1
+ elif names[1] == "t2":
+ return test_table_2
+ else:
+ raise Exception("Unrecognized table name")
+
+ substrait_query = """
+ {
+ "relations": [
+ {"rel": {
+ "read": {
+ "base_schema": {
+ "struct": {
+ "types": [
+ {"i64": {}}
+ ]
+ },
+ "names": [
+ "x"
+ ]
+ },
+ "namedTable": {
+ "names": ["TABLE_NAME_PLACEHOLDER"]
Review Comment:
No need to use a placeholder.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r962411864
##########
python/pyarrow/_substrait.pyx:
##########
@@ -25,14 +26,84 @@ from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef shared_ptr[CTable] _process_named_table(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] empty_table
+ py_names = []
+
+ # provider function could raise an exception
+ try:
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+ return pyarrow_unwrap_table(named_args["provider"](py_names))
+ except Exception:
+ return empty_table
Review Comment:
I am not quite sure if there is a `SafeCallIntoPython`, aren't we just wraping `cython` function into `std::function`?
If we don't handle it here, the `pytests` gives the aforementioned warning. What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r964736395
##########
cpp/src/arrow/engine/substrait/util.cc:
##########
@@ -107,19 +111,36 @@ class SubstraitExecutor {
bool plan_started_;
compute::ExecContext exec_context_;
std::shared_ptr<SubstraitSinkConsumer> sink_consumer_;
+ const ConversionOptions& conversion_options_;
};
} // namespace
Result<std::shared_ptr<RecordBatchReader>> ExecuteSerializedPlan(
- const Buffer& substrait_buffer, const ExtensionIdRegistry* extid_registry,
- compute::FunctionRegistry* func_registry) {
+ const Buffer& substrait_buffer, const PythonTableProvider& table_provider,
+ const ExtensionIdRegistry* registry, compute::FunctionRegistry* func_registry) {
Review Comment:
One thing, if we do this, I think we need to expose `Declaration` to Cython. Because we have to define the `NamedTableProvider` and bind it to the options. Is that okay?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r967106677
##########
python/pyarrow/_substrait.pyx:
##########
@@ -17,22 +17,99 @@
# cython: language_level = 3
from cython.operator cimport dereference as deref
+from libcpp.vector cimport vector as std_vector
from pyarrow import Buffer
-from pyarrow.lib import frombytes
+from pyarrow.lib import frombytes, tobytes
from pyarrow.lib cimport *
from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef CDeclaration _create_named_table_provider(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] c_in_table
+ shared_ptr[CTableSourceNodeOptions] c_tablesourceopts
+ shared_ptr[CExecNodeOptions] c_input_node_opts
+ vector[CDeclaration.Input] no_c_inputs
+
+ py_names = []
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+
+ py_table = named_args["provider"](py_names)
+ c_in_table = pyarrow_unwrap_table(py_table)
+ c_tablesourceopts = make_shared[CTableSourceNodeOptions](
+ c_in_table, 1 << 20)
Review Comment:
There is already a default for the second argument. Can you just do `make_shared[CTableSourceNodeOptions](c_in_table)`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on PR #14024:
URL: https://github.com/apache/arrow/pull/14024#issuecomment-1284745953
> This is a nice improvement!
>
> I am wondering one thing: why does the function take a list of names, instead of a single name?
>
> I understand from [#14024 (comment)](https://github.com/apache/arrow/pull/14024#issuecomment-1236472120) that this is because the name can _potentially_ be a list of names indicating some nested schema.
>
> But for the simple case of single names, having a function that just gets a string and returns a table would be simpler to implement. And for example that would allow to actually implement this easily with a mapping (eg `table_provide={"table1": table1, "table2: table2}.get`)
>
> Would it be an option to call the function with a single string, in case of length 1 list?
Good point. Probably we can create a Cython level wrapper for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r962061860
##########
python/pyarrow/_substrait.pyx:
##########
@@ -25,14 +26,84 @@ from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef shared_ptr[CTable] _process_named_table(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] empty_table
+ py_names = []
+
+ # provider function could raise an exception
+ try:
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+ return pyarrow_unwrap_table(named_args["provider"](py_names))
+ except Exception:
+ return empty_table
Review Comment:
How about doing a `raise` instead.
```bash
pyarrow/tests/test_substrait.py::test_named_table_invalid_table_name
/Users/vibhatha/venv/pyarrow_dev/lib/python3.10/site-packages/_pytest/unraisableexception.py:78: PytestUnraisableExceptionWarning: Exception ignored in: 'pyarrow._substrait._process_named_table'
Traceback (most recent call last):
File "pyarrow/_substrait.pyx", line 39, in pyarrow._substrait._process_named_table
return pyarrow_unwrap_table(named_args["provider"](py_names))
File "/Users/vibhatha/github/fork/arrow/python/pyarrow/tests/test_substrait.py", line 223, in table_provider
raise Exception("Unrecognized table name")
Exception: Unrecognized table name
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/Users/vibhatha/github/fork/arrow/python/pyarrow/tests/test_substrait.py", line 252, in test_named_table_invalid_table_name
substrait.run_query(buf, table_provider)
ValueError: Error occurred in extracting NamedTable : Unrecognized table name
warnings.warn(pytest.PytestUnraisableExceptionWarning(msg))
```
I want to fix this warning. Not sure if we can handle it here properly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r970186454
##########
python/pyarrow/_substrait.pyx:
##########
@@ -17,22 +17,98 @@
# cython: language_level = 3
from cython.operator cimport dereference as deref
+from libcpp.vector cimport vector as std_vector
from pyarrow import Buffer
-from pyarrow.lib import frombytes
+from pyarrow.lib import frombytes, tobytes
from pyarrow.lib cimport *
from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef CDeclaration _create_named_table_provider(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] c_in_table
+ shared_ptr[CTableSourceNodeOptions] c_tablesourceopts
+ shared_ptr[CExecNodeOptions] c_input_node_opts
+ vector[CDeclaration.Input] no_c_inputs
+
+ py_names = []
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+
+ py_table = named_args["provider"](py_names)
+ c_in_table = pyarrow_unwrap_table(py_table)
+ c_tablesourceopts = make_shared[CTableSourceNodeOptions](c_in_table)
+ c_input_node_opts = static_pointer_cast[CExecNodeOptions, CTableSourceNodeOptions](
+ c_tablesourceopts)
+ return CDeclaration(tobytes("table_source"),
+ no_c_inputs, c_input_node_opts)
+
+
+def run_query(plan, table_provider=None):
"""
Execute a Substrait plan and read the results as a RecordBatchReader.
Parameters
----------
plan : Buffer
The serialized Substrait plan to execute.
+ table_provider : object (optional)
+ A function to resolve any NamedTable relation to a table.
+ The function will receive a single argument which will be a list
+ of strings representing the table name and should return a pyarrow.Table.
+
+ Returns
+ -------
+ RecordBatchReader
+ A reader containing the result of the executed query
+
+ Examples
+ --------
+ >>> import pyarrow as pa
+ >>> from pyarrow.lib import tobytes
+ >>> import pyarrow.substrait as substrait
+ >>> test_table_1 = pa.Table.from_pydict({"x": [1, 2, 3]})
+ >>> test_table_2 = pa.Table.from_pydict({"x": [4, 5, 6]})
+ >>> def table_provider(names):
+ ... if not names:
+ ... raise Exception("No names provided")
+ ... elif names[0] == "t1":
+ ... return test_table_1
+ ... elif names[1] == "t2":
+ ... return test_table_2
+ ... else:
+ ... raise Exception("Unrecognized table name")
+ ...
+ >>> substrait_query = '''
+ ... {
+ ... "relations": [
+ ... {"rel": {
+ ... "read": {
+ ... "base_schema": {
+ ... "struct": {
+ ... "types": [
+ ... {"i64": {}}
+ ... ]
+ ... },
+ ... "names": [
+ ... "x"
+ ... ]
+ ... },
+ ... "namedTable": {
+ ... "names": ["t1"]
+ ... }
+ ... }
+ ... }}
+ ... ]
+ ... }
+ ... '''
+ >>> buf = pa._substrait._parse_json_plan(tobytes(substrait_query))
+ >>> pa.substrait.run_query_with_provider(buf, table_provider)
+ <pyarrow.lib.RecordBatchReader object at 0x100dcaf10>
Review Comment:
I see... let me come up with a better solution.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r967249411
##########
cpp/src/arrow/compute/exec/exec_plan.cc:
##########
@@ -643,6 +643,10 @@ Declaration Declaration::Sequence(std::vector<Declaration> decls) {
return out;
}
+bool Declaration::IsValid(ExecFactoryRegistry* registry) const {
+ return registry->GetFactory(this->factory_name).ok() && this->options != nullptr;
Review Comment:
You can but I don't know how high priority it would be. I think I'd want to see more users of Acero directly (and not just via Substrait or through bindings) before we worry too much about adding in utility validation at the declaration layer.
##########
python/pyarrow/_substrait.pyx:
##########
@@ -17,22 +17,99 @@
# cython: language_level = 3
from cython.operator cimport dereference as deref
+from libcpp.vector cimport vector as std_vector
from pyarrow import Buffer
-from pyarrow.lib import frombytes
+from pyarrow.lib import frombytes, tobytes
from pyarrow.lib cimport *
from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef CDeclaration _create_named_table_provider(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] c_in_table
+ shared_ptr[CTableSourceNodeOptions] c_tablesourceopts
+ shared_ptr[CExecNodeOptions] c_input_node_opts
+ vector[CDeclaration.Input] no_c_inputs
+
+ py_names = []
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+
+ py_table = named_args["provider"](py_names)
+ c_in_table = pyarrow_unwrap_table(py_table)
+ c_tablesourceopts = make_shared[CTableSourceNodeOptions](
+ c_in_table, 1 << 20)
Review Comment:
Sure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] github-actions[bot] commented on pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #14024:
URL: https://github.com/apache/arrow/pull/14024#issuecomment-1234517265
https://issues.apache.org/jira/browse/ARROW-17521
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r962420622
##########
python/pyarrow/_substrait.pyx:
##########
@@ -25,14 +26,84 @@ from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef shared_ptr[CTable] _process_named_table(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] empty_table
+ py_names = []
+
+ # provider function could raise an exception
+ try:
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+ return pyarrow_unwrap_table(named_args["provider"](py_names))
+ except Exception:
+ return empty_table
Review Comment:
BindFunction used SafeCallIntoPython. I think what you suggested was good.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r966665171
##########
cpp/src/arrow/compute/exec/exec_plan.cc:
##########
@@ -643,6 +643,10 @@ Declaration Declaration::Sequence(std::vector<Declaration> decls) {
return out;
}
+bool Declaration::IsValid(ExecFactoryRegistry* registry) const {
+ return registry->GetFactory(this->factory_name).ok() && this->options != nullptr;
Review Comment:
@westonpace should we do a full option verification here? I mean validate for all the supported options?
I am not quite sure if this is an adequate verification. Since it is not sure if doing that kind of a validation is a kind of redundant thing, I left it for discussion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r967209190
##########
python/pyarrow/_substrait.pyx:
##########
@@ -17,22 +17,99 @@
# cython: language_level = 3
from cython.operator cimport dereference as deref
+from libcpp.vector cimport vector as std_vector
from pyarrow import Buffer
-from pyarrow.lib import frombytes
+from pyarrow.lib import frombytes, tobytes
from pyarrow.lib cimport *
from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef CDeclaration _create_named_table_provider(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] c_in_table
+ shared_ptr[CTableSourceNodeOptions] c_tablesourceopts
+ shared_ptr[CExecNodeOptions] c_input_node_opts
+ vector[CDeclaration.Input] no_c_inputs
+
+ py_names = []
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+
+ py_table = named_args["provider"](py_names)
+ c_in_table = pyarrow_unwrap_table(py_table)
+ c_tablesourceopts = make_shared[CTableSourceNodeOptions](
+ c_in_table, 1 << 20)
Review Comment:
Shall we fix it here too: https://github.com/apache/arrow/blob/99b57e84277f24e8ec1ddadbb11ef8b4f43c8c89/python/pyarrow/_exec_plan.pyx#L95?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r964374357
##########
python/pyarrow/_substrait.pyx:
##########
@@ -25,14 +26,84 @@ from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef shared_ptr[CTable] _process_named_table(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] empty_table
+ py_names = []
+
+ # provider function could raise an exception
+ try:
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+ return pyarrow_unwrap_table(named_args["provider"](py_names))
+ except Exception:
+ return empty_table
Review Comment:
Only because of the warning. 🤔
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on PR #14024:
URL: https://github.com/apache/arrow/pull/14024#issuecomment-1246230329
cc @westonpace the failing CI is fixed here: https://github.com/apache/arrow/pull/14113
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r962062665
##########
python/pyarrow/_substrait.pyx:
##########
@@ -25,14 +26,84 @@ from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef shared_ptr[CTable] _process_named_table(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] empty_table
+ py_names = []
+
+ # provider function could raise an exception
+ try:
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+ return pyarrow_unwrap_table(named_args["provider"](py_names))
+ except Exception:
+ return empty_table
Review Comment:
`raise` looks good to me. I had forgotten that we were inside SafeCallIntoPython which handles that for us. Although in that case do we need to even catch the exception?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r967209618
##########
python/pyarrow/_substrait.pyx:
##########
@@ -17,22 +17,99 @@
# cython: language_level = 3
from cython.operator cimport dereference as deref
+from libcpp.vector cimport vector as std_vector
from pyarrow import Buffer
-from pyarrow.lib import frombytes
+from pyarrow.lib import frombytes, tobytes
from pyarrow.lib cimport *
from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef CDeclaration _create_named_table_provider(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] c_in_table
+ shared_ptr[CTableSourceNodeOptions] c_tablesourceopts
+ shared_ptr[CExecNodeOptions] c_input_node_opts
+ vector[CDeclaration.Input] no_c_inputs
+
+ py_names = []
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+
+ py_table = named_args["provider"](py_names)
+ c_in_table = pyarrow_unwrap_table(py_table)
+ c_tablesourceopts = make_shared[CTableSourceNodeOptions](
+ c_in_table, 1 << 20)
Review Comment:
Same PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r961349012
##########
cpp/src/arrow/engine/substrait/util.cc:
##########
@@ -126,6 +131,36 @@ Result<std::shared_ptr<RecordBatchReader>> ExecuteSerializedPlan(
return sink_reader;
}
+Result<std::shared_ptr<RecordBatchReader>> ExecuteSerializedPlan(
+ const Buffer& substrait_buffer, PythonTableProvider& table_provider,
+ const ExtensionIdRegistry* registry, compute::FunctionRegistry* func_registry) {
+ // TODO(ARROW-15732)
+ // retrieve input table from table provider
+
+ NamedTableProvider named_table_provider =
+ [table_provider](
+ const std::vector<std::string>& names) -> Result<compute::Declaration> {
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Table> input_table, table_provider(names));
+ std::shared_ptr<compute::ExecNodeOptions> options =
+ std::make_shared<compute::TableSourceNodeOptions>(input_table);
+ return compute::Declaration("table_source", {}, options,
+ "substrait_table_provider_source");
+ };
+
+ ConversionOptions conversion_options;
+ conversion_options.named_table_provider = std::move(named_table_provider);
+
+ compute::ExecContext exec_context(arrow::default_memory_pool(),
+ ::arrow::internal::GetCpuThreadPool(), func_registry);
+ ARROW_ASSIGN_OR_RAISE(auto plan, compute::ExecPlan::Make(&exec_context));
+ SubstraitExecutor executor(std::move(plan), exec_context, conversion_options);
+ RETURN_NOT_OK(executor.Init(substrait_buffer, registry));
+ ARROW_ASSIGN_OR_RAISE(auto sink_reader, executor.Execute());
+ // check closing here, not in destructor, to expose error to caller
+ RETURN_NOT_OK(executor.Close());
+ return sink_reader;
Review Comment:
Yes of course, this was merely a WIP proto code I just wrote for the sake of getting an insight from @richtia. I will update the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r962420903
##########
python/pyarrow/_substrait.pyx:
##########
@@ -25,14 +26,84 @@ from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef shared_ptr[CTable] _process_named_table(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] empty_table
+ py_names = []
+
+ # provider function could raise an exception
+ try:
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+ return pyarrow_unwrap_table(named_args["provider"](py_names))
+ except Exception:
+ return empty_table
Review Comment:
So should we leave as it is?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on PR #14024:
URL: https://github.com/apache/arrow/pull/14024#issuecomment-1234488508
cc @westonpace @richita
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r964759418
##########
cpp/src/arrow/engine/substrait/util.cc:
##########
@@ -107,19 +111,36 @@ class SubstraitExecutor {
bool plan_started_;
compute::ExecContext exec_context_;
std::shared_ptr<SubstraitSinkConsumer> sink_consumer_;
+ const ConversionOptions& conversion_options_;
};
} // namespace
Result<std::shared_ptr<RecordBatchReader>> ExecuteSerializedPlan(
- const Buffer& substrait_buffer, const ExtensionIdRegistry* extid_registry,
- compute::FunctionRegistry* func_registry) {
+ const Buffer& substrait_buffer, const PythonTableProvider& table_provider,
+ const ExtensionIdRegistry* registry, compute::FunctionRegistry* func_registry) {
Review Comment:
Please ignore my previous comment. It is already exposed in `_exec_plan.pyx`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r963795665
##########
python/pyarrow/_substrait.pyx:
##########
@@ -25,14 +26,84 @@ from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef shared_ptr[CTable] _process_named_table(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] empty_table
+ py_names = []
+
+ # provider function could raise an exception
+ try:
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+ return pyarrow_unwrap_table(named_args["provider"](py_names))
+ except Exception:
+ return empty_table
Review Comment:
Yes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r962049345
##########
cpp/src/arrow/engine/substrait/util.cc:
##########
@@ -107,19 +111,39 @@ class SubstraitExecutor {
bool plan_started_;
compute::ExecContext exec_context_;
std::shared_ptr<SubstraitSinkConsumer> sink_consumer_;
+ const ConversionOptions& conversion_options_;
};
} // namespace
Result<std::shared_ptr<RecordBatchReader>> ExecuteSerializedPlan(
- const Buffer& substrait_buffer, const ExtensionIdRegistry* extid_registry,
- compute::FunctionRegistry* func_registry) {
+ const Buffer& substrait_buffer, const PythonTableProvider& table_provider,
+ const ExtensionIdRegistry* registry, compute::FunctionRegistry* func_registry) {
// TODO(ARROW-15732)
+ // retrieve input table from table provider
+ ConversionOptions conversion_options;
+ if (table_provider) {
+ NamedTableProvider named_table_provider =
+ [table_provider](
+ const std::vector<std::string>& names) -> Result<compute::Declaration> {
+ if (names.empty()) {
+ return Status::Invalid("names for NamedTable not provided");
+ }
Review Comment:
Yeah that make sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on PR #14024:
URL: https://github.com/apache/arrow/pull/14024#issuecomment-1235139399
@westonpace one doubt is handling multiple names in the provider. Assume we get multiple tables, how should we generate the output?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace merged pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
westonpace merged PR #14024:
URL: https://github.com/apache/arrow/pull/14024
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r963800613
##########
cpp/src/arrow/engine/substrait/util.cc:
##########
@@ -107,19 +111,36 @@ class SubstraitExecutor {
bool plan_started_;
compute::ExecContext exec_context_;
std::shared_ptr<SubstraitSinkConsumer> sink_consumer_;
+ const ConversionOptions& conversion_options_;
};
} // namespace
Result<std::shared_ptr<RecordBatchReader>> ExecuteSerializedPlan(
- const Buffer& substrait_buffer, const ExtensionIdRegistry* extid_registry,
- compute::FunctionRegistry* func_registry) {
+ const Buffer& substrait_buffer, const PythonTableProvider& table_provider,
+ const ExtensionIdRegistry* registry, compute::FunctionRegistry* func_registry) {
Review Comment:
```suggestion
const Buffer& substrait_buffer,
const ExtensionIdRegistry* registry, compute::FunctionRegistry* func_registry, const ConversionOptions& conversion_options = {}) {
```
I suspect it will only be a matter of time before we want to specify other conversion options. Also, by defaulting to `{}` we can avoid changing existing tests that are unrelated to the table provider.
##########
python/pyarrow/_substrait.pyx:
##########
@@ -25,14 +26,84 @@ from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef shared_ptr[CTable] _process_named_table(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] empty_table
+ py_names = []
+
+ # provider function could raise an exception
+ try:
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+ return pyarrow_unwrap_table(named_args["provider"](py_names))
+ except Exception:
+ return empty_table
Review Comment:
Actually, no, not quite. I'm still not sure why we are catching the exception and rethrowing. Wouldn't it be better to just not catch it at all?
##########
cpp/src/arrow/engine/substrait/serde_test.cc:
##########
@@ -1000,15 +1000,17 @@ TEST(Substrait, GetRecordBatchReader) {
GTEST_SKIP() << "ARROW-16392: Substrait File URI not supported for Windows";
#else
ASSERT_OK_AND_ASSIGN(std::string substrait_json, GetSubstraitJSON());
- test_with_registries([&substrait_json](ExtensionIdRegistry* ext_id_reg,
- compute::FunctionRegistry* func_registry) {
- ASSERT_OK_AND_ASSIGN(auto buf, SerializeJsonPlan(substrait_json));
- ASSERT_OK_AND_ASSIGN(auto reader, 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);
- });
+ PythonTableProvider table_provider;
+ test_with_registries(
+ [&substrait_json, table_provider](ExtensionIdRegistry* ext_id_reg,
+ compute::FunctionRegistry* func_registry) {
+ ASSERT_OK_AND_ASSIGN(auto buf, SerializeJsonPlan(substrait_json));
+ ASSERT_OK_AND_ASSIGN(auto reader, ExecuteSerializedPlan(*buf, table_provider));
+ 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);
+ });
Review Comment:
```suggestion
test_with_registries(
[&substrait_json](ExtensionIdRegistry* ext_id_reg,
compute::FunctionRegistry* func_registry) {
ASSERT_OK_AND_ASSIGN(auto buf, SerializeJsonPlan(substrait_json));
ASSERT_OK_AND_ASSIGN(auto reader, 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);
});
```
Although...I don't think this is the best change. We should make `table_provider` an optional argument to `ExecuteSerializedPlan`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r963799075
##########
cpp/src/arrow/engine/substrait/serde_test.cc:
##########
@@ -1000,15 +1000,17 @@ TEST(Substrait, GetRecordBatchReader) {
GTEST_SKIP() << "ARROW-16392: Substrait File URI not supported for Windows";
#else
ASSERT_OK_AND_ASSIGN(std::string substrait_json, GetSubstraitJSON());
- test_with_registries([&substrait_json](ExtensionIdRegistry* ext_id_reg,
- compute::FunctionRegistry* func_registry) {
- ASSERT_OK_AND_ASSIGN(auto buf, SerializeJsonPlan(substrait_json));
- ASSERT_OK_AND_ASSIGN(auto reader, 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);
- });
+ PythonTableProvider table_provider;
+ test_with_registries(
+ [&substrait_json, table_provider](ExtensionIdRegistry* ext_id_reg,
+ compute::FunctionRegistry* func_registry) {
+ ASSERT_OK_AND_ASSIGN(auto buf, SerializeJsonPlan(substrait_json));
+ ASSERT_OK_AND_ASSIGN(auto reader, ExecuteSerializedPlan(*buf, table_provider));
+ 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);
+ });
Review Comment:
```suggestion
test_with_registries(
[&substrait_json](ExtensionIdRegistry* ext_id_reg,
compute::FunctionRegistry* func_registry) {
ASSERT_OK_AND_ASSIGN(auto buf, SerializeJsonPlan(substrait_json));
ASSERT_OK_AND_ASSIGN(auto reader, 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);
});
```
Although...I don't think this is the best suggestion. We should make `table_provider` an optional argument to `ExecuteSerializedPlan`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r967110525
##########
cpp/src/arrow/compute/exec/exec_plan.cc:
##########
@@ -643,6 +643,10 @@ Declaration Declaration::Sequence(std::vector<Declaration> decls) {
return out;
}
+bool Declaration::IsValid(ExecFactoryRegistry* registry) const {
+ return registry->GetFactory(this->factory_name).ok() && this->options != nullptr;
Review Comment:
That's beyond the scope of this PR. What you have is probably all that makes sense for the short term. Initially I would say, yes, it makes sense to do more extensive validation. However, in practice, I'm not sure it's possible to do without creating a plan and nodes from the declaration or requiring a second validation function when registering factories.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] vibhatha commented on a diff in pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
vibhatha commented on code in PR #14024:
URL: https://github.com/apache/arrow/pull/14024#discussion_r967205466
##########
python/pyarrow/_substrait.pyx:
##########
@@ -17,22 +17,99 @@
# cython: language_level = 3
from cython.operator cimport dereference as deref
+from libcpp.vector cimport vector as std_vector
from pyarrow import Buffer
-from pyarrow.lib import frombytes
+from pyarrow.lib import frombytes, tobytes
from pyarrow.lib cimport *
from pyarrow.includes.libarrow cimport *
from pyarrow.includes.libarrow_substrait cimport *
-def run_query(plan):
+cdef CDeclaration _create_named_table_provider(dict named_args, const std_vector[c_string]& names):
+ cdef:
+ c_string c_name
+ shared_ptr[CTable] c_in_table
+ shared_ptr[CTableSourceNodeOptions] c_tablesourceopts
+ shared_ptr[CExecNodeOptions] c_input_node_opts
+ vector[CDeclaration.Input] no_c_inputs
+
+ py_names = []
+ for i in range(names.size()):
+ c_name = names[i]
+ py_names.append(frombytes(c_name))
+
+ py_table = named_args["provider"](py_names)
+ c_in_table = pyarrow_unwrap_table(py_table)
+ c_tablesourceopts = make_shared[CTableSourceNodeOptions](
+ c_in_table, 1 << 20)
Review Comment:
Oh yes, I missed that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] jorisvandenbossche commented on pull request #14024: ARROW-17521: [Python] Add python bindings for NamedTableProvider for Substrait consumer
Posted by GitBox <gi...@apache.org>.
jorisvandenbossche commented on PR #14024:
URL: https://github.com/apache/arrow/pull/14024#issuecomment-1283743330
This is a nice improvement!
I am wondering one thing: why does the function take a list of names, instead of a single name?
I understand from https://github.com/apache/arrow/pull/14024#issuecomment-1236472120 that this is because the name can _potentially_ be a list of names indicating some nested schema.
But for the simple case of single names, having a function that just gets a string and returns a table would be simpler to implement. And for example that would allow to actually implement this easily with a mapping (eg `table_provide={"table1": table1, "table2: table2}.get`)
Would it be an option to call the function with a single string, in case of length 1 list?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org