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