ARROW-6341: [Python] Implement low-level bindings for Dataset

[kszucs]

No description provided.

[emkornfield]

@kszucs is this still WIP (I assume so based on CI builds?)

[kszucs]

@emkornfield yes, I'll continue to work on it after the 0.15.1 release. Theoretically the Dataset API should be ready for having bindings.

[jorisvandenbossche]

Something I ran into yesterday: trying to access the partition_scheme attribute of the discovery segfaults:

from pyarrow.dataset import FileSystemDataSourceDiscovery, ParquetFileFormat
from pyarrow.fs import Selector, LocalFileSystem

fs = LocalFileSystem()
selector = Selector('test_dataset/', recursive=True)
parquet_format = ParquetFileFormat()
discovery = FileSystemDataSourceDiscovery(fs, selector, parquet_format)
discovery.partition_scheme

where "test_dataset" is a simple directory with a single small parquet file in it.
(the actual tests you wrote do pass though)

[kszucs]

@jorisvandenbossche I've fixed that.

[fsaintjacques]

You need to add an implicit cast like in R for this to be bearable. Otherwise you get annoying errors:

In [41]: cond = ds.ComparisonExpression(ds.CompareOperator.Greater, ds.FieldExpression("total_amount"), ds.ScalarExpression(1000.0))                           

In [42]: scanner_builder.filter(cond)                                          
---------------------------------------------------------------------------
ArrowTypeError                            Traceback (most recent call last)
<ipython-input-42-bb6fba558cf8> in <module>
----> 1 scanner_builder.filter(cond)

~/src/db/arrow/python/pyarrow/_dataset.pyx in pyarrow._dataset.ScannerBuilder.filter()
    951         self : ScannerBuilder
    952         """
--> 953         check_status(self.builder.Filter(filter_expression.unwrap()))
    954         return self
    955 

~/src/db/arrow/python/pyarrow/error.pxi in pyarrow.lib.check_status()
     86             raise ArrowNotImplementedError(message)
     87         elif status.IsTypeError():
---> 88             raise ArrowTypeError(message)
     89         elif status.IsCapacityError():
     90             raise ArrowCapacityError(message)

ArrowTypeError: cannot compare expressions of differing type, float vs double

[kszucs]

Well, I didn't want to use any implicit behaviour in the first iteration. Perhaps this should be done by the C++ filter method?

[kszucs]

We should do it as a follow-up PR to handle it nicely.

[jorisvandenbossche]

It's not fully clear to me what this last sentence exactly means.
Does it mean this scheme will be used to filter/project the output of the data sources, or that the other projections/filters specified when scanning should result in something that matches this schema?

[kszucs]

Well, it was not clear to me either, it is copied from the C++ apidocs. In case of projections this schema seems to be omitted? cc @fsaintjacques

[kszucs]

Hmm, github doesn't allow to request @jorisvandenbossche's review, so here it is :)

[kszucs]

We should have a jira about introducing InMemoryDataSource, if I recall correctly we already have one?

[kszucs]

I'm not planning to implement any new features here, although we should discuss the possible follow-up PRs. A couple candidates:

• We need to refactor the scalar handling on the python side, I have a WIP patch for it.
• We need to define a more pythonic API for the dataset bindings, because the current one is pretty low-level.
• We need to improve the tests.
• We need to improve the apidocs.
• We should have hypothesis strategies and tests for the datasets.
• We should exercise the unit tests from the current parquet dataset implementation on the new one.
• We should develop a shim over the datasets api to make the transition from the previous parquet dataset implementation more fluid.

[jorisvandenbossche]

Quickly tested again some of the code snippets I wrote last week, and that's all still working nicely.
One thing I noticed is that due to the removal of DataFragment and FileSource, you no longer can check the file paths in a Dataset/FileSystemDataSource. Maybe something to consider later if / how we want to expose something like that.

There are still several classes and methods/properties that need docstrings, but it would be fine for me to do that as a follow-up if that makes we can merge this faster.

[jorisvandenbossche]

I'm not planning to implement any new features here, although we should discuss the possible follow-up PRs. A couple candidates:

That sounds good to me.

We need to define a more pythonic API for the dataset bindings, because the current one is pretty low-level.

Yes, something like the open_dataset from my notebook (https://nbviewer.jupyter.org/gist/jorisvandenbossche/73f7c8d0921a79b461c0a4928fbdc7fa) can probably be part of this (which was modelled after the existing R methods).

We should develop a shim over the datasets api to make the transition from the previous parquet dataset implementation more fluid.

And we also still need to discuss to what extent we want to keep the existing parquet dataset implementation, or deprecate it, or try to implement it partly using the new machinery (eg if we want to support dask's usage, we need to keep it)

[bkietz]

This is looking good, thanks for doing this!

I think most of your follow up candidates can wait for a different PR, but I think the unit tests need some work before this can be merged.

[pitrou]

Just some comments while I skimmed over this.

[pitrou]

It's not obvious to me why partition_expression should be equal to the source_partition constructor argument. Can you explain?

[bkietz]

That argument is used to set the partition_expression property, so they should be equal

[pitrou]

Need to add docstrings for all the public classes here.

[pitrou]

Should also add docstrings for public methods.

[wesm]

In my quick read didn't see anything too unreasonable. I left a handful of comments

[wesm]

Do you want to check that the one you pass in is passed on correctly to the ScanContext? I think you can use logging_memory_pool to check

[wesm]

You might consider doing dispatch with a dict instead

[bkietz]

Is it necessary to expose partition scheme as a class at all? I think it would suffice to have factories like make_hive_partition_scheme()

[jorisvandenbossche]

Is it necessary to expose partition scheme as a class at all? I think it would suffice to have factories like make_hive_partition_scheme()

What would a make_hive_partition_scheme() then return if the partition scheme object itself is not exposed?

[kszucs]

ds.partition_scheme(pa.Schema schema, string flavor='hive') -> ds.PartitionScheme

Although I prefer to have not opaque return types, so I can distinguish between:

ds.partition_scheme(schema)
ds.partition_scheme(schema, flavor='hive')

If we return the same class then I don't have the ability to inspect that they are differ.

[bkietz]

LGTM.

Follow up JIRA for winnowing the public classes: https://issues.apache.org/jira/browse/ARROW-7391