| 720 | } |
| 721 | |
| 722 | [[nodiscard]] bool appendBoundRecord( |
| 723 | TopicHandle topic, Timestamp timestamp, const PJ_bound_field_value_t* fields, std::size_t field_count) { |
| 724 | auto engine_locks = lockWriteEngines(engine, secondary_engine); |
| 725 | if (engine.getTopicStorage(topic.id) == nullptr) { |
| 726 | setError(fmt::format("topic {} not found", topic.id)); |
| 727 | return false; |
| 728 | } |
| 729 | |
| 730 | tsl::robin_set<FieldId> seen_ids; |
| 731 | struct ResolvedField { |
| 732 | PrimitiveType type; |
| 733 | const PJ_bound_field_value_t* raw; |
| 734 | }; |
| 735 | std::vector<ResolvedField> resolved; |
| 736 | resolved.reserve(field_count); |
| 737 | for (std::size_t i = 0; i < field_count; ++i) { |
| 738 | const auto& field = fields[i]; |
| 739 | if (field.field.topic.id != topic.id) { |
| 740 | setError("field handle does not belong to the target topic"); |
| 741 | return false; |
| 742 | } |
| 743 | if (!seen_ids.insert(field.field.id).second) { |
| 744 | setError(fmt::format("duplicate field id {}", field.field.id)); |
| 745 | return false; |
| 746 | } |
| 747 | PrimitiveType type{}; |
| 748 | if (!lookupFieldType(topic, field.field.id, &type)) { |
| 749 | return false; |
| 750 | } |
| 751 | if (!field.is_null && !validateScalar(field.value, type, "appendBoundRecord")) { |
| 752 | return false; |
| 753 | } |
| 754 | resolved.push_back({type, &field}); |
| 755 | } |
| 756 | |
| 757 | auto begin_status = writer.beginRow(topic.id, timestamp); |
| 758 | if (!begin_status.has_value()) { |
| 759 | setError(begin_status.error()); |
| 760 | return false; |
| 761 | } |
| 762 | for (const auto& field : resolved) { |
| 763 | if (field.raw->is_null) { |
| 764 | writer.setNull(topic.id, static_cast<std::size_t>(field.raw->field.id)); |
| 765 | } else { |
| 766 | setFieldValue(topic.id, static_cast<std::size_t>(field.raw->field.id), field.type, field.raw->value); |
| 767 | } |
| 768 | } |
| 769 | auto finish_status = writer.finishRow(topic.id); |
| 770 | if (!finish_status.has_value()) { |
| 771 | setError(finish_status.error()); |
| 772 | return false; |
| 773 | } |
| 774 | last_error.clear(); |
| 775 | return true; |
| 776 | } |
| 777 | |
| 778 | /// Ingest a whole Arrow C Data Interface stream into a topic. |
| 779 | /// |