(dataset, use_threads=True, implicit_ordering=False)
| 57 | |
| 58 | |
| 59 | def _dataset_to_decl(dataset, use_threads=True, implicit_ordering=False): |
| 60 | decl = Declaration("scan", ScanNodeOptions( |
| 61 | dataset, use_threads=use_threads, |
| 62 | implicit_ordering=implicit_ordering)) |
| 63 | |
| 64 | # Get rid of special dataset columns |
| 65 | # "__fragment_index", "__batch_index", "__last_in_fragment", "__filename" |
| 66 | projections = [field(f) for f in dataset.schema.names] |
| 67 | decl = Declaration.from_sequence( |
| 68 | [decl, Declaration("project", ProjectNodeOptions(projections))] |
| 69 | ) |
| 70 | |
| 71 | filter_expr = dataset._scan_options.get("filter") |
| 72 | if filter_expr is not None: |
| 73 | # Filters applied in CScanNodeOptions are "best effort" for the scan node itself |
| 74 | # so we always need to inject an additional Filter node to apply them for real. |
| 75 | decl = Declaration.from_sequence( |
| 76 | [decl, Declaration("filter", FilterNodeOptions(filter_expr))] |
| 77 | ) |
| 78 | |
| 79 | return decl |
| 80 | |
| 81 | |
| 82 | def _perform_join(join_type, left_operand, left_keys, |
no test coverage detected