| 1024 | } |
| 1025 | |
| 1026 | Result<std::unique_ptr<substrait::ReadRel>> ScanRelationConverter( |
| 1027 | const std::shared_ptr<Schema>& schema, const acero::Declaration& declaration, |
| 1028 | ExtensionSet* ext_set, const ConversionOptions& conversion_options) { |
| 1029 | auto read_rel = std::make_unique<substrait::ReadRel>(); |
| 1030 | const auto& scan_node_options = |
| 1031 | checked_cast<const dataset::ScanNodeOptions&>(*declaration.options); |
| 1032 | auto dataset = |
| 1033 | dynamic_cast<dataset::FileSystemDataset*>(scan_node_options.dataset.get()); |
| 1034 | if (dataset == nullptr) { |
| 1035 | return Status::Invalid( |
| 1036 | "Can only convert scan node with FileSystemDataset to a Substrait plan."); |
| 1037 | } |
| 1038 | |
| 1039 | // set schema |
| 1040 | ARROW_ASSIGN_OR_RAISE(auto named_struct, ToProto(*schema, ext_set, conversion_options)); |
| 1041 | read_rel->set_allocated_base_schema(named_struct.release()); |
| 1042 | |
| 1043 | // set local files |
| 1044 | auto read_rel_lfs = std::make_unique<substrait::ReadRel::LocalFiles>(); |
| 1045 | for (const auto& file : dataset->files()) { |
| 1046 | auto read_rel_lfs_ffs = |
| 1047 | std::make_unique<substrait::ReadRel::LocalFiles::FileOrFiles>(); |
| 1048 | ARROW_ASSIGN_OR_RAISE(auto uri_path, UriFromAbsolutePath(file)); |
| 1049 | read_rel_lfs_ffs->set_uri_path(std::move(uri_path)); |
| 1050 | // set file format |
| 1051 | auto format_type_name = dataset->format()->type_name(); |
| 1052 | if (format_type_name == "parquet") { |
| 1053 | read_rel_lfs_ffs->set_allocated_parquet( |
| 1054 | new substrait::ReadRel::LocalFiles::FileOrFiles::ParquetReadOptions()); |
| 1055 | } else if (format_type_name == "ipc") { |
| 1056 | read_rel_lfs_ffs->set_allocated_arrow( |
| 1057 | new substrait::ReadRel::LocalFiles::FileOrFiles::ArrowReadOptions()); |
| 1058 | } else if (format_type_name == "orc") { |
| 1059 | read_rel_lfs_ffs->set_allocated_orc( |
| 1060 | new substrait::ReadRel::LocalFiles::FileOrFiles::OrcReadOptions()); |
| 1061 | } else { |
| 1062 | return Status::NotImplemented("Unsupported file type: ", format_type_name); |
| 1063 | } |
| 1064 | read_rel_lfs->mutable_items()->AddAllocated(read_rel_lfs_ffs.release()); |
| 1065 | } |
| 1066 | read_rel->set_allocated_local_files(read_rel_lfs.release()); |
| 1067 | return read_rel; |
| 1068 | } |
| 1069 | |
| 1070 | Result<std::unique_ptr<substrait::FilterRel>> FilterRelationConverter( |
| 1071 | const std::shared_ptr<Schema>& schema, const acero::Declaration& declaration, |