| 624 | } |
| 625 | |
| 626 | [[nodiscard]] bool appendRecord( |
| 627 | TopicHandle topic, Timestamp timestamp, const PJ_named_field_value_t* fields, std::size_t field_count) { |
| 628 | auto engine_locks = lockWriteEngines(engine, secondary_engine); |
| 629 | if (engine.getTopicStorage(topic.id) == nullptr) { |
| 630 | setError(fmt::format("topic {} not found", topic.id)); |
| 631 | return false; |
| 632 | } |
| 633 | |
| 634 | tsl::robin_set<std::string_view> seen_names; |
| 635 | struct ResolvedField { |
| 636 | FieldHandle handle; |
| 637 | PrimitiveType type; |
| 638 | const PJ_named_field_value_t* raw; |
| 639 | }; |
| 640 | std::vector<ResolvedField> resolved; |
| 641 | resolved.reserve(field_count); |
| 642 | // Backing store for names that needed "//" → "/" normalization; a deque keeps |
| 643 | // element addresses stable so the string_views below (and in seen_names) stay |
| 644 | // valid for the whole record. Untouched on the common (no-"//") path. |
| 645 | std::deque<std::string> normalized_names; |
| 646 | for (std::size_t i = 0; i < field_count; ++i) { |
| 647 | const auto& field = fields[i]; |
| 648 | auto name = toStringView(field.name); |
| 649 | if (needsFieldNormalization(name)) { |
| 650 | std::string norm = normalizeSlashes(name); |
| 651 | if (!norm.empty() && norm.front() == '/') { |
| 652 | norm.erase(0, 1); |
| 653 | } |
| 654 | normalized_names.push_back(std::move(norm)); |
| 655 | name = normalized_names.back(); |
| 656 | } |
| 657 | if (!seen_names.insert(name).second) { |
| 658 | setError(fmt::format("duplicate field name '{}'", name)); |
| 659 | return false; |
| 660 | } |
| 661 | if (field.is_null) { |
| 662 | // Null values: look up existing field by name (borrowed probe, no alloc). |
| 663 | auto it = field_cache.find(TopicFieldKeyView{.topic_id = topic.id, .field_name = name}); |
| 664 | if (it == field_cache.end()) { |
| 665 | // Field has never been seen. Check if this is a typed null (the ABI |
| 666 | // carries value.type even when is_null is true). A valid type lets |
| 667 | // us create the column now; an untyped null (kNull) is silently |
| 668 | // skipped — the column will be created when a non-null value arrives. |
| 669 | auto type_or = fromAbiType(field.value.type); |
| 670 | if (type_or.has_value()) { |
| 671 | FieldHandle handle{}; |
| 672 | if (!ensureField(topic, name, field.value.type, &handle)) { |
| 673 | return false; |
| 674 | } |
| 675 | resolved.push_back({handle, *type_or, &field}); |
| 676 | } |
| 677 | continue; |
| 678 | } |
| 679 | PrimitiveType existing{}; |
| 680 | if (!lookupFieldType(topic, it->second.id, &existing)) { |
| 681 | return false; |
| 682 | } |
| 683 | resolved.push_back({it->second, existing, &field}); |