| 492 | namespace { |
| 493 | |
| 494 | Result<acero::ExecNode*> MakeWriteNode(acero::ExecPlan* plan, |
| 495 | std::vector<acero::ExecNode*> inputs, |
| 496 | const acero::ExecNodeOptions& options) { |
| 497 | if (inputs.size() != 1) { |
| 498 | return Status::Invalid("Write SinkNode requires exactly 1 input, got ", |
| 499 | inputs.size()); |
| 500 | } |
| 501 | |
| 502 | const WriteNodeOptions write_node_options = |
| 503 | checked_cast<const WriteNodeOptions&>(options); |
| 504 | std::shared_ptr<Schema> custom_schema = write_node_options.custom_schema; |
| 505 | const std::shared_ptr<const KeyValueMetadata>& custom_metadata = |
| 506 | write_node_options.custom_metadata; |
| 507 | const FileSystemDatasetWriteOptions& write_options = write_node_options.write_options; |
| 508 | |
| 509 | const std::shared_ptr<Schema>& input_schema = inputs[0]->output_schema(); |
| 510 | |
| 511 | if (custom_schema != nullptr) { |
| 512 | if (custom_metadata) { |
| 513 | return Status::TypeError( |
| 514 | "Do not provide both custom_metadata and custom_schema. If custom_schema is " |
| 515 | "used then custom_schema->metadata should be used instead of custom_metadata"); |
| 516 | } |
| 517 | |
| 518 | if (custom_schema->num_fields() != input_schema->num_fields()) { |
| 519 | return Status::TypeError( |
| 520 | "The provided custom_schema did not have the same number of fields as the " |
| 521 | "data. The custom schema can only be used to add metadata / nullability to " |
| 522 | "fields and cannot change the type or number of fields."); |
| 523 | } |
| 524 | for (int field_idx = 0; field_idx < input_schema->num_fields(); field_idx++) { |
| 525 | if (!input_schema->field(field_idx)->type()->Equals( |
| 526 | custom_schema->field(field_idx)->type())) { |
| 527 | return Status::TypeError("The provided custom_schema specified type ", |
| 528 | custom_schema->field(field_idx)->type()->ToString(), |
| 529 | " for field ", field_idx, "and the input data has type ", |
| 530 | input_schema->field(field_idx), |
| 531 | "The custom schema can only be used to add metadata / " |
| 532 | "nullability to fields and " |
| 533 | "cannot change the type or number of fields."); |
| 534 | } |
| 535 | } |
| 536 | } |
| 537 | |
| 538 | if (custom_metadata) { |
| 539 | custom_schema = input_schema->WithMetadata(custom_metadata); |
| 540 | } |
| 541 | |
| 542 | if (!write_options.partitioning) { |
| 543 | return Status::Invalid("Must provide partitioning"); |
| 544 | } |
| 545 | |
| 546 | std::shared_ptr<DatasetWritingSinkNodeConsumer> consumer = |
| 547 | std::make_shared<DatasetWritingSinkNodeConsumer>(custom_schema, write_options); |
| 548 | |
| 549 | ARROW_ASSIGN_OR_RAISE( |
| 550 | auto node, |
| 551 | // to preserve order explicitly, sequence the exec batches |
nothing calls this directly
no test coverage detected