| 54 | namespace { |
| 55 | |
| 56 | Status HandleUnalignedBuffers(ExecBatch* batch, UnalignedBufferHandling handling) { |
| 57 | if (handling == UnalignedBufferHandling::kIgnore) { |
| 58 | return Status::OK(); |
| 59 | } |
| 60 | for (auto& value : batch->values) { |
| 61 | if (value.is_array()) { |
| 62 | switch (handling) { |
| 63 | case UnalignedBufferHandling::kIgnore: |
| 64 | // Should be impossible to get here |
| 65 | return Status::OK(); |
| 66 | case UnalignedBufferHandling::kError: |
| 67 | if (!arrow::util::CheckAlignment(*value.array(), |
| 68 | arrow::util::kValueAlignment)) { |
| 69 | return Status::Invalid( |
| 70 | "An input buffer was poorly aligned and UnalignedBufferHandling is set " |
| 71 | "to kError"); |
| 72 | } |
| 73 | break; |
| 74 | case UnalignedBufferHandling::kWarn: |
| 75 | if (!arrow::util::CheckAlignment(*value.array(), |
| 76 | arrow::util::kValueAlignment)) { |
| 77 | ARROW_LOG(WARNING) |
| 78 | << "An input buffer was poorly aligned. This could lead to crashes or " |
| 79 | "poor performance on some hardware. Please ensure that all Acero " |
| 80 | "sources generate aligned buffers, or change the unaligned buffer " |
| 81 | "handling configuration to silence this warning."; |
| 82 | } |
| 83 | break; |
| 84 | case UnalignedBufferHandling::kReallocate: { |
| 85 | ARROW_ASSIGN_OR_RAISE(value, arrow::util::EnsureAlignment( |
| 86 | value.array(), arrow::util::kValueAlignment, |
| 87 | default_memory_pool())); |
| 88 | break; |
| 89 | } |
| 90 | } |
| 91 | } |
| 92 | } |
| 93 | return Status::OK(); |
| 94 | } |
| 95 | |
| 96 | struct SourceNode : ExecNode, public TracedNode { |
| 97 | SourceNode(ExecPlan* plan, std::shared_ptr<Schema> output_schema, |
no test coverage detected