Ingest a whole Arrow C Data Interface stream into a topic. Ownership contract: callers pass a producer-owned @p stream. The caller decides whether to release after this call — this method does NOT call stream->release. That lets the outermost ABI trampoline enforce the "success releases, failure retains" rule uniformly.
| 782 | /// call stream->release. That lets the outermost ABI trampoline enforce |
| 783 | /// the "success releases, failure retains" rule uniformly. |
| 784 | [[nodiscard]] bool appendArrowStream( |
| 785 | TopicHandle topic, struct ArrowArrayStream* stream, PJ_string_view_t timestamp_column) { |
| 786 | auto engine_locks = lockWriteEngines(engine, secondary_engine); |
| 787 | if (stream == nullptr) { |
| 788 | setError("append_arrow_stream: null stream"); |
| 789 | return false; |
| 790 | } |
| 791 | if (engine.getTopicStorage(topic.id) == nullptr) { |
| 792 | setError(fmt::format("topic {} not found", topic.id)); |
| 793 | return false; |
| 794 | } |
| 795 | |
| 796 | auto schema_or = arrow_import::schemaFromArrowStream(stream); |
| 797 | if (!schema_or.has_value()) { |
| 798 | setError(schema_or.error()); |
| 799 | return false; |
| 800 | } |
| 801 | |
| 802 | const std::string_view timestamp_name = toStringView(timestamp_column); |
| 803 | int ts_arrow_col = -1; |
| 804 | std::vector<arrow_import::ArrowColumnMapping> mappings; |
| 805 | for (const auto& mapping : schema_or->second) { |
| 806 | if (!timestamp_name.empty() && mapping.field_name == timestamp_name) { |
| 807 | ts_arrow_col = mapping.arrow_column_index; |
| 808 | continue; |
| 809 | } |
| 810 | |
| 811 | FieldHandle field{}; |
| 812 | if (!ensureField(topic, mapping.field_name, static_cast<PJ_primitive_type_t>(mapping.pj_type), &field)) { |
| 813 | return false; |
| 814 | } |
| 815 | auto adjusted = mapping; |
| 816 | adjusted.pj_column_index = static_cast<std::size_t>(field.id); |
| 817 | mappings.push_back(std::move(adjusted)); |
| 818 | } |
| 819 | |
| 820 | if (!timestamp_name.empty() && ts_arrow_col < 0) { |
| 821 | setError(fmt::format("timestamp column '{}' not found in stream schema", timestamp_name)); |
| 822 | return false; |
| 823 | } |
| 824 | |
| 825 | auto status = arrow_import::importArrowStream(writer, topic.id, stream, mappings, ts_arrow_col); |
| 826 | if (!status.has_value()) { |
| 827 | setError(status.error()); |
| 828 | return false; |
| 829 | } |
| 830 | last_error.clear(); |
| 831 | return true; |
| 832 | } |
| 833 | |
| 834 | void flushPending() { |
| 835 | auto lock = engine.lockEngine(); |
no test coverage detected