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 2020/04/21 12:41:43 UTC
[GitHub] [arrow] fsaintjacques opened a new pull request #7000: ARROW-8065: [C++][Dataset] Refactor ScanOptions and Fragment relation
fsaintjacques opened a new pull request #7000:
URL: https://github.com/apache/arrow/pull/7000
This is the first part of a refactor to make Fragment accessible without a Scan operation instance. This is a breaking change. It introduces the concept of a physical schema and read schema, these concepts are analogous to Avro writer and reader schema.
- Move ScanOptions at Fragment::Scan instead of a property
- Refactor Dataset::GetFragments(ScanOptions) to Dataset::GetFragments(Expression)
- Add Fragment::ReadPhysicalSchema
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] fsaintjacques commented on a change in pull request #7000: ARROW-8065: [C++][Dataset] Refactor ScanOptions and Fragment relation
Posted by GitBox <gi...@apache.org>.
fsaintjacques commented on a change in pull request #7000:
URL: https://github.com/apache/arrow/pull/7000#discussion_r413045582
##########
File path: cpp/src/arrow/dataset/file_base.cc
##########
@@ -222,9 +214,8 @@ FragmentIterator FileSystemDataset::GetFragmentsImpl(
}
Result<std::shared_ptr<FileSystemDataset>> FileSystemDataset::Write(
- const WritePlan& plan, std::shared_ptr<ScanContext> scan_context) {
- std::vector<std::shared_ptr<ScanOptions>> options(plan.paths.size());
-
+ const WritePlan& plan, std::shared_ptr<ScanOptions> scan_options,
Review comment:
That's a task for [ARROW-8382](https://jira.apache.org/jira/projects/ARROW/issues/ARROW-8382). The writer shouldn't be aware of such issue, it takes a file source, a format, a RecordBatchReader (the dropping should happen internally here) and writes it.
Since the Writer API has no bindings and not in used, I added this to ARROW-8382 so we can prioritise ARROW-8062 (parquet dataset) and ARROW-8318 (fragment stored in dataset).
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] fsaintjacques commented on a change in pull request #7000: ARROW-8065: [C++][Dataset] Refactor ScanOptions and Fragment relation
Posted by GitBox <gi...@apache.org>.
fsaintjacques commented on a change in pull request #7000:
URL: https://github.com/apache/arrow/pull/7000#discussion_r413007995
##########
File path: cpp/src/arrow/dataset/dataset.h
##########
@@ -84,13 +82,12 @@ class ARROW_DS_EXPORT Fragment {
class ARROW_DS_EXPORT InMemoryFragment : public Fragment {
public:
InMemoryFragment(RecordBatchVector record_batches,
- std::shared_ptr<ScanOptions> scan_options);
+ std::shared_ptr<Expression> = NULLPTR);
Review comment:
Doesn't compile.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] fsaintjacques commented on a change in pull request #7000: ARROW-8065: [C++][Dataset] Refactor ScanOptions and Fragment relation
Posted by GitBox <gi...@apache.org>.
fsaintjacques commented on a change in pull request #7000:
URL: https://github.com/apache/arrow/pull/7000#discussion_r413045582
##########
File path: cpp/src/arrow/dataset/file_base.cc
##########
@@ -222,9 +214,8 @@ FragmentIterator FileSystemDataset::GetFragmentsImpl(
}
Result<std::shared_ptr<FileSystemDataset>> FileSystemDataset::Write(
- const WritePlan& plan, std::shared_ptr<ScanContext> scan_context) {
- std::vector<std::shared_ptr<ScanOptions>> options(plan.paths.size());
-
+ const WritePlan& plan, std::shared_ptr<ScanOptions> scan_options,
Review comment:
That's a task for [ARROW-8382](https://jira.apache.org/jira/projects/ARROW/issues/ARROW-8382). The writer shouldn't be aware of such issue, it takes a file source, a format, a RecordBatchReader (the dropping should happen internally here) and writes it.
Since the Writer API has no bindings and not in use yet, I added this to ARROW-8382 so we can prioritise ARROW-8062 (parquet dataset) and ARROW-8318 (fragment stored in dataset).
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] bkietz commented on a change in pull request #7000: ARROW-8065: [C++][Dataset] Refactor ScanOptions and Fragment relation
Posted by GitBox <gi...@apache.org>.
bkietz commented on a change in pull request #7000:
URL: https://github.com/apache/arrow/pull/7000#discussion_r412252930
##########
File path: cpp/src/arrow/dataset/dataset.h
##########
@@ -84,13 +82,12 @@ class ARROW_DS_EXPORT Fragment {
class ARROW_DS_EXPORT InMemoryFragment : public Fragment {
public:
InMemoryFragment(RecordBatchVector record_batches,
- std::shared_ptr<ScanOptions> scan_options);
+ std::shared_ptr<Expression> = NULLPTR);
Review comment:
`scalar` is declared such that you can use `scalar(true)` for default arguments
```suggestion
std::shared_ptr<Expression> = scalar(true));
```
##########
File path: cpp/src/arrow/dataset/dataset.h
##########
@@ -100,16 +97,20 @@ class ARROW_DS_EXPORT InMemoryFragment : public Fragment {
RecordBatchVector record_batches_;
};
-/// \brief A container of zero or more Fragments. A Dataset acts as a discovery mechanism
-/// of Fragments and partitions, e.g. files deeply nested in a directory.
+/// \brief A container of zero or more Fragments.
+///
+/// A Dataset acts as a union of Fragments, e.g. files deeply nested in a
+/// directory. A Dataset has a schema, also known as the "reader" schema.
Review comment:
Is this known as the "reader" schema outside Avro space? If not I think it's still acceptable to explain this by analogy to Avro, but we should be explicit whose jargon we're borrowing
##########
File path: cpp/src/arrow/dataset/dataset.h
##########
@@ -100,16 +97,20 @@ class ARROW_DS_EXPORT InMemoryFragment : public Fragment {
RecordBatchVector record_batches_;
};
-/// \brief A container of zero or more Fragments. A Dataset acts as a discovery mechanism
-/// of Fragments and partitions, e.g. files deeply nested in a directory.
+/// \brief A container of zero or more Fragments.
+///
+/// A Dataset acts as a union of Fragments, e.g. files deeply nested in a
+/// directory. A Dataset has a schema, also known as the "reader" schema.
+///
class ARROW_DS_EXPORT Dataset : public std::enable_shared_from_this<Dataset> {
public:
/// \brief Begin to build a new Scan operation against this Dataset
Result<std::shared_ptr<ScannerBuilder>> NewScan(std::shared_ptr<ScanContext> context);
Result<std::shared_ptr<ScannerBuilder>> NewScan();
- /// \brief GetFragments returns an iterator of Fragments given ScanOptions.
- FragmentIterator GetFragments(std::shared_ptr<ScanOptions> options);
+ /// \brief GetFragments returns an iterator of Fragments given a predicate.
+ FragmentIterator GetFragments(std::shared_ptr<Expression> predicate);
+ FragmentIterator GetFragments();
Review comment:
```suggestion
FragmentIterator GetFragments(std::shared_ptr<Expression> predicate = scalar(true));
```
##########
File path: cpp/src/arrow/dataset/dataset.cc
##########
@@ -30,34 +30,40 @@
namespace arrow {
namespace dataset {
-Fragment::Fragment(std::shared_ptr<ScanOptions> scan_options)
- : scan_options_(std::move(scan_options)), partition_expression_(scalar(true)) {}
+Fragment::Fragment(std::shared_ptr<Expression> partition_expression)
+ : partition_expression_(partition_expression ? partition_expression : scalar(true)) {}
-const std::shared_ptr<Schema>& Fragment::schema() const {
- return scan_options_->schema();
-}
+Result<std::shared_ptr<Schema>> InMemoryFragment::ReadPhysicalSchema() {
Review comment:
Probably follow up: `InMemoryFragment` should be constructed with an explicit schema with convenience constructor to extract batches[0]->schema that fails if batches are empty.
##########
File path: cpp/src/arrow/dataset/dataset.cc
##########
@@ -72,36 +78,15 @@ Result<std::shared_ptr<ScannerBuilder>> Dataset::NewScan() {
return NewScan(std::make_shared<ScanContext>());
}
-bool Dataset::AssumePartitionExpression(
- const std::shared_ptr<ScanOptions>& scan_options,
- std::shared_ptr<ScanOptions>* simplified_scan_options) const {
- if (partition_expression_ == nullptr) {
- if (simplified_scan_options != nullptr) {
- *simplified_scan_options = scan_options;
- }
- return true;
- }
+FragmentIterator Dataset::GetFragments() { return GetFragments(scalar(true)); }
- auto expr = scan_options->filter->Assume(*partition_expression_);
- if (expr->IsNull() || expr->Equals(false)) {
- // selector is not satisfiable; yield no fragments
- return false;
+FragmentIterator Dataset::GetFragments(std::shared_ptr<Expression> predicate) {
+ if (partition_expression_) {
Review comment:
Isn't this initialized to `scalar(true)`?
##########
File path: cpp/src/arrow/dataset/filter.cc
##########
@@ -655,31 +655,32 @@ std::string ScalarExpression::ToString() const {
return value_->ToString() + ":" + type_repr;
}
+using arrow::internal::JoinStrings;
Review comment:
:+1:
##########
File path: cpp/src/arrow/dataset/file_parquet_test.cc
##########
@@ -470,8 +466,8 @@ TEST_F(TestParquetFileFormat, PredicatePushdownRowGroupFragments) {
CountRowGroupsInFragment(fragment, internal::Iota(5, static_cast<int>(kNumRowGroups)),
"i64"_ >= int64_t(6));
- CountRowGroupsInFragment(fragment, {5, 6, 7}, "i64"_ >= int64_t(6),
- "i64"_ < int64_t(8));
+ CountRowGroupsInFragment(fragment, {5, 6, 7},
+ "i64"_ >= int64_t(6) && "i64"_ < int64_t(8));
Review comment:
consistency nit
```suggestion
"i64"_ >= int64_t(6) and "i64"_ < int64_t(8));
```
##########
File path: cpp/src/arrow/dataset/dataset.h
##########
@@ -30,12 +30,21 @@
namespace arrow {
namespace dataset {
-/// \brief A granular piece of a Dataset, such as an individual file, which can be
-/// read/scanned separately from other fragments.
+/// \brief A granular piece of a Dataset, such as an individual file.
///
-/// A Fragment yields a collection of RecordBatch, encapsulated in one or more ScanTasks.
+/// A Fragment can be read/scanned separately from other fragments. It yields a
+/// collection of RecordBatch, encapsulated in one or more ScanTasks.
+///
+/// A notable difference from Dataset is that Fragments have physical schemas
+/// which may differ from Fragments.
Review comment:
```suggestion
/// Note that Fragments have well defined physical schemas which are reconciled by
/// the Datasets which contain them; these physical schemas may differ from a parent
/// Dataset's schema and the physical schemas of sibling Fragments.
```
##########
File path: cpp/src/arrow/dataset/partition.cc
##########
@@ -444,10 +432,9 @@ struct DirectoryPartitioningFactory::MakeWritePlanImpl {
ARROW_ASSIGN_OR_RAISE(out.partitioning,
this_->Finish(std::move(partitioning_schema)));
- auto fragment_schema =
- source_fragments_.empty() ? schema({}) : source_fragments_.front()->schema();
+ // There's no guarantee that all Fragments have the same schema.
ARROW_ASSIGN_OR_RAISE(out.schema,
- UnifySchemas({out.partitioning->schema(), fragment_schema}));
+ UnifySchemas({out.partitioning->schema(), schema_}));
Review comment:
This doesn't drop fields implicit in `out.partitioning`, which means we'll be writing redundant information. This should be handled by removing `scan_options` from the parameters of `FileDataset::Write`, see following comment
##########
File path: python/pyarrow/_dataset.pyx
##########
@@ -104,27 +104,45 @@ cdef class Dataset:
self.dataset.ReplaceSchema(pyarrow_unwrap_schema(schema)))
return Dataset.wrap(move(copy))
- def get_fragments(self, columns=None, filter=None):
+ def get_fragments(self, Expression filter=None):
"""Returns an iterator over the fragments in this dataset.
Parameters
----------
- columns : list of str, default None
- List of columns to project.
filter : Expression, default None
- Scan will return only the rows matching the filter.
+ Return fragments matching the optional filter, either using the
+ partition_expression or internal information like Parquet's
+ statistics.
Review comment:
```suggestion
Return fragments matching the optional filter, using explicit
partition expressions and/or embedded information like Parquet's
statistics.
```
##########
File path: cpp/src/arrow/dataset/scanner_internal.h
##########
@@ -55,28 +56,55 @@ inline RecordBatchIterator ProjectRecordBatch(RecordBatchIterator it,
class FilterAndProjectScanTask : public ScanTask {
public:
- explicit FilterAndProjectScanTask(std::shared_ptr<ScanTask> task)
- : ScanTask(task->options(), task->context()), task_(std::move(task)) {}
+ explicit FilterAndProjectScanTask(std::shared_ptr<ScanTask> task,
+ std::shared_ptr<Expression> partition)
+ : ScanTask(task->options(), task->context()),
+ task_(std::move(task)),
+ partition_(std::move(partition)),
+ filter_(NULLPTR),
Review comment:
Slight tangent: why aren't `*_internal.h` excluded in lint_cpp_cli.py? It seems they should be
##########
File path: cpp/src/arrow/dataset/filter.h
##########
@@ -183,6 +183,11 @@ class ARROW_DS_EXPORT Expression {
return Assume(*given);
}
+ /// Indicates if the expression is satisfiable.
+ ///
+ /// This is a shortcut to check if the expression is neither null nor false.
+ bool IsSatisfiable() const { return !IsNull() && !Equals(false); }
Review comment:
:+1:
##########
File path: python/pyarrow/_dataset.pyx
##########
@@ -104,27 +104,45 @@ cdef class Dataset:
self.dataset.ReplaceSchema(pyarrow_unwrap_schema(schema)))
return Dataset.wrap(move(copy))
- def get_fragments(self, columns=None, filter=None):
+ def get_fragments(self, Expression filter=None):
"""Returns an iterator over the fragments in this dataset.
Parameters
----------
- columns : list of str, default None
- List of columns to project.
filter : Expression, default None
- Scan will return only the rows matching the filter.
+ Return fragments matching the optional filter, either using the
+ partition_expression or internal information like Parquet's
+ statistics.
Returns
-------
fragments : iterator of Fragment
"""
- return Scanner(self, columns=columns, filter=filter).get_fragments()
+ cdef:
+ CFragmentIterator iterator
+ shared_ptr[CFragment] fragment
- def scan(self, columns=None, filter=None, MemoryPool memory_pool=None):
+ if filter is None or filter.expr == nullptr:
+ iterator = self.dataset.GetFragments()
+ else:
+ iterator = self.dataset.GetFragments(filter.unwrap())
+
+ while True:
+ fragment = GetResultValue(iterator.Next())
+ if fragment.get() == nullptr:
+ raise StopIteration()
+ else:
+ yield Fragment.wrap(fragment)
+
+ def _scanner(self, **kwargs):
+ return Scanner.from_dataset(self, **kwargs)
+
+ def scan(self, columns=None, filter=None,
+ MemoryPool memory_pool=None, **kwargs):
"""Builds a scan operation against the dataset.
- It poduces a stream of ScanTasks which is meant to be a unit of work to
- be dispatched. The tasks are not executed automatically, the user is
+ It produces a stream of ScanTasks which is meant to be a unit of work
+ to be dispatched. The tasks are not executed automatically, the user is
responsible to execute and dispatch the individual tasks, so custom
local task scheduling can be implemented.
Review comment:
```suggestion
It produces a stream of ScanTasks. Each task is meant to be a unit of work
to be dispatched by the user; they are not executed automatically. This allows
customization of local task scheduling and execution.
```
##########
File path: python/pyarrow/_dataset.pyx
##########
@@ -104,27 +104,45 @@ cdef class Dataset:
self.dataset.ReplaceSchema(pyarrow_unwrap_schema(schema)))
return Dataset.wrap(move(copy))
- def get_fragments(self, columns=None, filter=None):
+ def get_fragments(self, Expression filter=None):
"""Returns an iterator over the fragments in this dataset.
Parameters
----------
- columns : list of str, default None
- List of columns to project.
filter : Expression, default None
- Scan will return only the rows matching the filter.
+ Return fragments matching the optional filter, either using the
+ partition_expression or internal information like Parquet's
+ statistics.
Returns
-------
fragments : iterator of Fragment
"""
- return Scanner(self, columns=columns, filter=filter).get_fragments()
+ cdef:
+ CFragmentIterator iterator
+ shared_ptr[CFragment] fragment
- def scan(self, columns=None, filter=None, MemoryPool memory_pool=None):
+ if filter is None or filter.expr == nullptr:
Review comment:
This coercion is common, should we have `_unwrap_expression_default_true()`?
##########
File path: cpp/src/arrow/dataset/file_base.cc
##########
@@ -222,9 +214,8 @@ FragmentIterator FileSystemDataset::GetFragmentsImpl(
}
Result<std::shared_ptr<FileSystemDataset>> FileSystemDataset::Write(
- const WritePlan& plan, std::shared_ptr<ScanContext> scan_context) {
- std::vector<std::shared_ptr<ScanOptions>> options(plan.paths.size());
-
+ const WritePlan& plan, std::shared_ptr<ScanOptions> scan_options,
Review comment:
```suggestion
const WritePlan& plan,
```
Instead, `scan_options` should be derived from `plan.{schema,partitioning}` to project out columns implicit in the partitioning
##########
File path: cpp/src/arrow/dataset/dataset.h
##########
@@ -30,12 +30,21 @@
namespace arrow {
namespace dataset {
-/// \brief A granular piece of a Dataset, such as an individual file, which can be
-/// read/scanned separately from other fragments.
+/// \brief A granular piece of a Dataset, such as an individual file.
///
-/// A Fragment yields a collection of RecordBatch, encapsulated in one or more ScanTasks.
+/// A Fragment can be read/scanned separately from other fragments. It yields a
+/// collection of RecordBatch, encapsulated in one or more ScanTasks.
Review comment:
```suggestion
/// A Fragment can be read/scanned separately from other fragments. It yields a
/// collection of RecordBatches when scanned, encapsulated in one or more ScanTasks.
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] jorisvandenbossche commented on a change in pull request #7000: ARROW-8065: [C++][Dataset] Refactor ScanOptions and Fragment relation
Posted by GitBox <gi...@apache.org>.
jorisvandenbossche commented on a change in pull request #7000:
URL: https://github.com/apache/arrow/pull/7000#discussion_r413622598
##########
File path: cpp/src/arrow/dataset/file_parquet.cc
##########
@@ -402,23 +401,21 @@ Result<ScanTaskIterator> ParquetFileFormat::ScanFile(
}
Result<std::shared_ptr<FileFragment>> ParquetFileFormat::MakeFragment(
- FileSource source, std::shared_ptr<ScanOptions> options,
- std::shared_ptr<Expression> partition_expression, std::vector<int> row_groups) {
+ FileSource source, std::shared_ptr<Expression> partition_expression,
+ std::vector<int> row_groups) {
return std::shared_ptr<FileFragment>(
- new ParquetFileFragment(std::move(source), shared_from_this(), std::move(options),
+ new ParquetFileFragment(std::move(source), shared_from_this(),
std::move(partition_expression), std::move(row_groups)));
}
Result<std::shared_ptr<FileFragment>> ParquetFileFormat::MakeFragment(
- FileSource source, std::shared_ptr<ScanOptions> options,
- std::shared_ptr<Expression> partition_expression) {
- return std::shared_ptr<FileFragment>(
- new ParquetFileFragment(std::move(source), shared_from_this(), std::move(options),
- std::move(partition_expression), {}));
+ FileSource source, std::shared_ptr<Expression> partition_expression) {
+ return std::shared_ptr<FileFragment>(new ParquetFileFragment(
+ std::move(source), shared_from_this(), std::move(partition_expression), {}));
}
Result<FragmentIterator> ParquetFileFormat::GetRowGroupFragments(
- const ParquetFileFragment& fragment, std::shared_ptr<Expression> extra_filter) {
+ const ParquetFileFragment& fragment, std::shared_ptr<Expression> filter) {
Review comment:
Small comment: in the header file it is still called "extra_filter"
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] fsaintjacques commented on issue #7000: ARROW-8065: [C++][Dataset] Refactor ScanOptions and Fragment relation
Posted by GitBox <gi...@apache.org>.
fsaintjacques commented on issue #7000:
URL: https://github.com/apache/arrow/pull/7000#issuecomment-618110604
Addressed most comments and updated followup ticket with what's missing. PTAL and merge quickly so we can unblock the blocked tickets :)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] fsaintjacques commented on a change in pull request #7000: ARROW-8065: [C++][Dataset] Refactor ScanOptions and Fragment relation
Posted by GitBox <gi...@apache.org>.
fsaintjacques commented on a change in pull request #7000:
URL: https://github.com/apache/arrow/pull/7000#discussion_r413074200
##########
File path: cpp/src/arrow/dataset/dataset.cc
##########
@@ -30,34 +30,40 @@
namespace arrow {
namespace dataset {
-Fragment::Fragment(std::shared_ptr<ScanOptions> scan_options)
- : scan_options_(std::move(scan_options)), partition_expression_(scalar(true)) {}
+Fragment::Fragment(std::shared_ptr<Expression> partition_expression)
+ : partition_expression_(partition_expression ? partition_expression : scalar(true)) {}
-const std::shared_ptr<Schema>& Fragment::schema() const {
- return scan_options_->schema();
-}
+Result<std::shared_ptr<Schema>> InMemoryFragment::ReadPhysicalSchema() {
Review comment:
The effort to open a JIRA is equal to implementing it. I added it.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] fsaintjacques commented on a change in pull request #7000: ARROW-8065: [C++][Dataset] Refactor ScanOptions and Fragment relation
Posted by GitBox <gi...@apache.org>.
fsaintjacques commented on a change in pull request #7000:
URL: https://github.com/apache/arrow/pull/7000#discussion_r413046083
##########
File path: cpp/src/arrow/dataset/dataset.h
##########
@@ -100,16 +97,20 @@ class ARROW_DS_EXPORT InMemoryFragment : public Fragment {
RecordBatchVector record_batches_;
};
-/// \brief A container of zero or more Fragments. A Dataset acts as a discovery mechanism
-/// of Fragments and partitions, e.g. files deeply nested in a directory.
+/// \brief A container of zero or more Fragments.
+///
+/// A Dataset acts as a union of Fragments, e.g. files deeply nested in a
+/// directory. A Dataset has a schema, also known as the "reader" schema.
+///
class ARROW_DS_EXPORT Dataset : public std::enable_shared_from_this<Dataset> {
public:
/// \brief Begin to build a new Scan operation against this Dataset
Result<std::shared_ptr<ScannerBuilder>> NewScan(std::shared_ptr<ScanContext> context);
Result<std::shared_ptr<ScannerBuilder>> NewScan();
- /// \brief GetFragments returns an iterator of Fragments given ScanOptions.
- FragmentIterator GetFragments(std::shared_ptr<ScanOptions> options);
+ /// \brief GetFragments returns an iterator of Fragments given a predicate.
+ FragmentIterator GetFragments(std::shared_ptr<Expression> predicate);
+ FragmentIterator GetFragments();
Review comment:
Ditto.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] fsaintjacques commented on a change in pull request #7000: ARROW-8065: [C++][Dataset] Refactor ScanOptions and Fragment relation
Posted by GitBox <gi...@apache.org>.
fsaintjacques commented on a change in pull request #7000:
URL: https://github.com/apache/arrow/pull/7000#discussion_r412407398
##########
File path: cpp/src/arrow/dataset/dataset.cc
##########
@@ -72,36 +78,15 @@ Result<std::shared_ptr<ScannerBuilder>> Dataset::NewScan() {
return NewScan(std::make_shared<ScanContext>());
}
-bool Dataset::AssumePartitionExpression(
- const std::shared_ptr<ScanOptions>& scan_options,
- std::shared_ptr<ScanOptions>* simplified_scan_options) const {
- if (partition_expression_ == nullptr) {
- if (simplified_scan_options != nullptr) {
- *simplified_scan_options = scan_options;
- }
- return true;
- }
+FragmentIterator Dataset::GetFragments() { return GetFragments(scalar(true)); }
- auto expr = scan_options->filter->Assume(*partition_expression_);
- if (expr->IsNull() || expr->Equals(false)) {
- // selector is not satisfiable; yield no fragments
- return false;
+FragmentIterator Dataset::GetFragments(std::shared_ptr<Expression> predicate) {
+ if (partition_expression_) {
Review comment:
A unit test triggered a segfault. I added this to move on to the next failure.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] jorisvandenbossche commented on a change in pull request #7000: ARROW-8065: [C++][Dataset] Refactor ScanOptions and Fragment relation
Posted by GitBox <gi...@apache.org>.
jorisvandenbossche commented on a change in pull request #7000:
URL: https://github.com/apache/arrow/pull/7000#discussion_r412150862
##########
File path: cpp/src/arrow/dataset/dataset.h
##########
@@ -30,12 +30,22 @@
namespace arrow {
namespace dataset {
-/// \brief A granular piece of a Dataset, such as an individual file, which can be
-/// read/scanned separately from other fragments.
+/// \brief A granular piece of a Dataset, such as an individual file.
///
-/// A Fragment yields a collection of RecordBatch, encapsulated in one or more ScanTasks.
+/// A Fragment can be read/scanned separately from other fragments. It yields a
+/// collection of RecordBatch, encapsulated in one or more ScanTasks.
+///
+/// A notable difference from Dataset is that Fragments have physical schemas
+/// which may differ from Fragments.
Review comment:
```suggestion
/// which may differ from other Fragments.
```
##########
File path: python/pyarrow/_dataset.pyx
##########
@@ -519,30 +500,69 @@ cdef class Fragment:
"""
return Expression.wrap(self.fragment.partition_expression())
- def to_table(self, use_threads=True, MemoryPool memory_pool=None):
- """Convert this Fragment into a Table.
+ def _scanner(self, **kwargs):
+ return Scanner.from_fragment(self, **kwargs)
- Use this convenience utility with care. This will serially materialize
- the Scan result in memory before creating the Table.
+ def scan(self, columns=None, filter=None, use_threads=True,
+ MemoryPool memory_pool=None, **kwargs):
+ """Builds a scan operation against the dataset.
+
Review comment:
When using Fragment.scan, it uses the Fragment's physical schema for the resulting table? (since the Fragment is not aware of the dataset "read" schema?)
If so, we should note that here in the docstring I think
##########
File path: cpp/src/arrow/dataset/file_parquet.cc
##########
@@ -433,26 +430,22 @@ Result<FragmentIterator> ParquetFileFormat::GetRowGroupFragments(
}
FragmentVector fragments(row_groups.size());
- auto new_options = std::make_shared<ScanOptions>(*fragment.scan_options());
- if (!extra_filter->Equals(true)) {
- new_options->filter = and_(std::move(extra_filter), std::move(new_options->filter));
- }
-
- RowGroupSkipper skipper(std::move(metadata), std::move(arrow_properties),
- new_options->filter, std::move(row_groups));
+ RowGroupSkipper skipper(std::move(metadata), std::move(arrow_properties), extra_filter,
Review comment:
We should probably rename "extra_filter" to just "filter" or "predicate" as how it is called in Dataset::GetFragments, since it is no longer "extra" ?
##########
File path: python/pyarrow/tests/test_dataset.py
##########
@@ -671,41 +669,29 @@ def test_fragments(tempdir):
f = fragments[0]
# file's schema does not include partition column
- phys_schema = f.schema.remove(f.schema.get_field_index('part'))
- assert f.format.inspect(f.path, f.filesystem) == phys_schema
+ assert f.physical_schema.names == ['f1', 'f2']
+ assert f.format.inspect(f.path, f.filesystem) == f.physical_schema
assert f.partition_expression.equals(ds.field('part') == 'a')
# scanning fragment includes partition columns
- result = f.to_table()
- assert f.schema == result.schema
+ result = f.to_table(schema=dataset.schema)
Review comment:
Can you also test without passing the dataset's schema, and assert that the column_names are [f1, f2] ?
##########
File path: python/pyarrow/tests/test_dataset.py
##########
@@ -671,41 +669,29 @@ def test_fragments(tempdir):
f = fragments[0]
# file's schema does not include partition column
- phys_schema = f.schema.remove(f.schema.get_field_index('part'))
- assert f.format.inspect(f.path, f.filesystem) == phys_schema
+ assert f.physical_schema.names == ['f1', 'f2']
+ assert f.format.inspect(f.path, f.filesystem) == f.physical_schema
assert f.partition_expression.equals(ds.field('part') == 'a')
# scanning fragment includes partition columns
- result = f.to_table()
- assert f.schema == result.schema
+ result = f.to_table(schema=dataset.schema)
assert result.column_names == ['f1', 'f2', 'part']
- assert len(result) == 4
assert result.equals(table.slice(0, 4))
-
- # scanning fragments follow column projection
- fragments = list(dataset.get_fragments(columns=['f1', 'part']))
Review comment:
Keep this but where the columns selection is passe to `to_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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] fsaintjacques commented on a change in pull request #7000: ARROW-8065: [C++][Dataset] Refactor ScanOptions and Fragment relation
Posted by GitBox <gi...@apache.org>.
fsaintjacques commented on a change in pull request #7000:
URL: https://github.com/apache/arrow/pull/7000#discussion_r412452967
##########
File path: python/pyarrow/tests/test_dataset.py
##########
@@ -671,41 +669,29 @@ def test_fragments(tempdir):
f = fragments[0]
# file's schema does not include partition column
- phys_schema = f.schema.remove(f.schema.get_field_index('part'))
- assert f.format.inspect(f.path, f.filesystem) == phys_schema
+ assert f.physical_schema.names == ['f1', 'f2']
+ assert f.format.inspect(f.path, f.filesystem) == f.physical_schema
assert f.partition_expression.equals(ds.field('part') == 'a')
# scanning fragment includes partition columns
- result = f.to_table()
- assert f.schema == result.schema
+ result = f.to_table(schema=dataset.schema)
assert result.column_names == ['f1', 'f2', 'part']
- assert len(result) == 4
assert result.equals(table.slice(0, 4))
-
- # scanning fragments follow column projection
- fragments = list(dataset.get_fragments(columns=['f1', 'part']))
Review comment:
Via kwargs to `self._scanner` then forwarding kwargs to `Scanner.from_fragment`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] fsaintjacques commented on a change in pull request #7000: ARROW-8065: [C++][Dataset] Refactor ScanOptions and Fragment relation
Posted by GitBox <gi...@apache.org>.
fsaintjacques commented on a change in pull request #7000:
URL: https://github.com/apache/arrow/pull/7000#discussion_r413164872
##########
File path: python/pyarrow/_dataset.pyx
##########
@@ -519,30 +500,69 @@ cdef class Fragment:
"""
return Expression.wrap(self.fragment.partition_expression())
- def to_table(self, use_threads=True, MemoryPool memory_pool=None):
- """Convert this Fragment into a Table.
+ def _scanner(self, **kwargs):
+ return Scanner.from_fragment(self, **kwargs)
- Use this convenience utility with care. This will serially materialize
- the Scan result in memory before creating the Table.
+ def scan(self, columns=None, filter=None, use_threads=True,
+ MemoryPool memory_pool=None, **kwargs):
+ """Builds a scan operation against the dataset.
+
Review comment:
Yes, but you can still pass schema (which is not documented. I'll add 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] jorisvandenbossche commented on a change in pull request #7000: ARROW-8065: [C++][Dataset] Refactor ScanOptions and Fragment relation
Posted by GitBox <gi...@apache.org>.
jorisvandenbossche commented on a change in pull request #7000:
URL: https://github.com/apache/arrow/pull/7000#discussion_r412191161
##########
File path: python/pyarrow/tests/test_dataset.py
##########
@@ -671,41 +669,29 @@ def test_fragments(tempdir):
f = fragments[0]
# file's schema does not include partition column
- phys_schema = f.schema.remove(f.schema.get_field_index('part'))
- assert f.format.inspect(f.path, f.filesystem) == phys_schema
+ assert f.physical_schema.names == ['f1', 'f2']
+ assert f.format.inspect(f.path, f.filesystem) == f.physical_schema
assert f.partition_expression.equals(ds.field('part') == 'a')
# scanning fragment includes partition columns
- result = f.to_table()
- assert f.schema == result.schema
+ result = f.to_table(schema=dataset.schema)
assert result.column_names == ['f1', 'f2', 'part']
- assert len(result) == 4
assert result.equals(table.slice(0, 4))
-
- # scanning fragments follow column projection
- fragments = list(dataset.get_fragments(columns=['f1', 'part']))
Review comment:
Keep this but where the columns selection is passed to `to_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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] kszucs commented on issue #7000: ARROW-8065: [C++][Dataset] Refactor ScanOptions and Fragment relation
Posted by GitBox <gi...@apache.org>.
kszucs commented on issue #7000:
URL: https://github.com/apache/arrow/pull/7000#issuecomment-617868408
@ursabot build
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] github-actions[bot] commented on issue #7000: ARROW-8065: [C++][Dataset] Refactor ScanOptions and Fragment relation
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on issue #7000:
URL: https://github.com/apache/arrow/pull/7000#issuecomment-617157131
https://issues.apache.org/jira/browse/ARROW-8065
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org