| 1011 | namespace { |
| 1012 | |
| 1013 | Result<acero::ExecNode*> MakeScanNode(acero::ExecPlan* plan, |
| 1014 | std::vector<acero::ExecNode*> inputs, |
| 1015 | const acero::ExecNodeOptions& options) { |
| 1016 | const auto& scan_node_options = checked_cast<const ScanNodeOptions&>(options); |
| 1017 | auto scan_options = scan_node_options.scan_options; |
| 1018 | auto dataset = scan_node_options.dataset; |
| 1019 | bool require_sequenced_output = scan_node_options.require_sequenced_output; |
| 1020 | bool implicit_ordering = scan_node_options.implicit_ordering; |
| 1021 | |
| 1022 | RETURN_NOT_OK(NormalizeScanOptions(scan_options, dataset->schema())); |
| 1023 | |
| 1024 | // using a generator for speculative forward compatibility with async fragment discovery |
| 1025 | ARROW_ASSIGN_OR_RAISE(auto fragments_it, dataset->GetFragments(scan_options->filter)); |
| 1026 | ARROW_ASSIGN_OR_RAISE(auto fragments_vec, fragments_it.ToVector()); |
| 1027 | auto fragment_gen = MakeVectorGenerator(std::move(fragments_vec)); |
| 1028 | |
| 1029 | ARROW_ASSIGN_OR_RAISE(auto batch_gen_gen, |
| 1030 | FragmentsToBatches(std::move(fragment_gen), scan_options)); |
| 1031 | |
| 1032 | AsyncGenerator<EnumeratedRecordBatch> merged_batch_gen; |
| 1033 | if (require_sequenced_output) { |
| 1034 | if (scan_options->fragment_readahead > 1) { |
| 1035 | ARROW_ASSIGN_OR_RAISE(merged_batch_gen, MakeSequencedMergedGenerator( |
| 1036 | std::move(batch_gen_gen), |
| 1037 | scan_options->fragment_readahead)); |
| 1038 | } else { |
| 1039 | merged_batch_gen = MakeConcatenatedGenerator(std::move(batch_gen_gen)); |
| 1040 | } |
| 1041 | } else { |
| 1042 | merged_batch_gen = |
| 1043 | MakeMergedGenerator(std::move(batch_gen_gen), scan_options->fragment_readahead); |
| 1044 | } |
| 1045 | |
| 1046 | AsyncGenerator<EnumeratedRecordBatch> batch_gen; |
| 1047 | if (scan_options->fragment_readahead > 1) { |
| 1048 | batch_gen = MakeReadaheadGenerator(std::move(merged_batch_gen), |
| 1049 | scan_options->fragment_readahead); |
| 1050 | } else { |
| 1051 | batch_gen = std::move(merged_batch_gen); |
| 1052 | } |
| 1053 | |
| 1054 | auto gen = MakeMappedGenerator( |
| 1055 | std::move(batch_gen), |
| 1056 | [scan_options](const EnumeratedRecordBatch& partial) |
| 1057 | -> Result<std::optional<compute::ExecBatch>> { |
| 1058 | // TODO(ARROW-13263) fragments may be able to attach more guarantees to batches |
| 1059 | // than this, for example parquet's row group stats. Failing to do this leaves |
| 1060 | // perf on the table because row group stats could be used to skip kernel execs in |
| 1061 | // FilterNode. |
| 1062 | // |
| 1063 | // Additionally, if a fragment failed to perform projection pushdown there may be |
| 1064 | // unnecessarily materialized columns in batch. We could drop them now instead of |
| 1065 | // letting them coast through the rest of the plan. |
| 1066 | auto guarantee = partial.fragment.value->partition_expression(); |
| 1067 | |
| 1068 | ARROW_ASSIGN_OR_RAISE( |
| 1069 | std::optional<compute::ExecBatch> batch, |
| 1070 | compute::MakeExecBatch(*scan_options->dataset_schema, |