| 383 | } // namespace |
| 384 | |
| 385 | Result<DeclarationInfo> FromProto(const substrait::Rel& rel, const ExtensionSet& ext_set, |
| 386 | const ConversionOptions& conversion_options) { |
| 387 | static bool dataset_init = false; |
| 388 | if (!dataset_init) { |
| 389 | dataset_init = true; |
| 390 | dataset::internal::Initialize(); |
| 391 | } |
| 392 | |
| 393 | switch (rel.rel_type_case()) { |
| 394 | case substrait::Rel::RelTypeCase::kRead: { |
| 395 | const auto& read = rel.read(); |
| 396 | RETURN_NOT_OK(CheckRelCommon(read, conversion_options)); |
| 397 | |
| 398 | // Get the base schema for the read relation |
| 399 | ARROW_ASSIGN_OR_RAISE(auto base_schema, |
| 400 | FromProto(read.base_schema(), ext_set, conversion_options)); |
| 401 | |
| 402 | auto scan_options = std::make_shared<dataset::ScanOptions>(); |
| 403 | scan_options->use_threads = true; |
| 404 | scan_options->add_augmented_fields = false; |
| 405 | |
| 406 | if (read.has_filter()) { |
| 407 | ARROW_ASSIGN_OR_RAISE(scan_options->filter, |
| 408 | FromProto(read.filter(), ext_set, conversion_options)); |
| 409 | } |
| 410 | |
| 411 | if (read.has_projection()) { |
| 412 | return Status::NotImplemented("substrait::ReadRel::projection"); |
| 413 | } |
| 414 | |
| 415 | if (read.has_named_table()) { |
| 416 | if (!conversion_options.named_table_provider) { |
| 417 | return Status::Invalid( |
| 418 | "plan contained a named table but a NamedTableProvider has not been " |
| 419 | "configured"); |
| 420 | } |
| 421 | |
| 422 | if (read.named_table().names().empty()) { |
| 423 | return Status::Invalid("names for NamedTable not provided"); |
| 424 | } |
| 425 | |
| 426 | const NamedTableProvider& named_table_provider = |
| 427 | conversion_options.named_table_provider; |
| 428 | const substrait::ReadRel::NamedTable& named_table = read.named_table(); |
| 429 | std::vector<std::string> table_names(named_table.names().begin(), |
| 430 | named_table.names().end()); |
| 431 | ARROW_ASSIGN_OR_RAISE(acero::Declaration source_decl, |
| 432 | named_table_provider(table_names, *base_schema)); |
| 433 | |
| 434 | if (!source_decl.IsValid()) { |
| 435 | return Status::Invalid("Invalid NamedTable Source"); |
| 436 | } |
| 437 | |
| 438 | return ProcessEmit(read, DeclarationInfo{std::move(source_decl), base_schema}, |
| 439 | base_schema); |
| 440 | } |
| 441 | |
| 442 | if (!read.has_local_files()) { |
nothing calls this directly
no test coverage detected