| 155 | } |
| 156 | |
| 157 | Expected<FieldId> DataEngine::createTopicField( |
| 158 | TopicId topic_id, std::string_view field_name, PrimitiveType type, std::optional<FieldId> requested_id) { |
| 159 | std::unique_lock<std::recursive_mutex> lock(impl_->mutex_); |
| 160 | auto topic_it = impl_->topics.find(topic_id); |
| 161 | if (topic_it == impl_->topics.end()) { |
| 162 | return PJ::unexpected(fmt::format("createTopicField: topic {} not found", topic_id)); |
| 163 | } |
| 164 | TopicStorage& storage = *topic_it.value(); |
| 165 | |
| 166 | // Idempotent re-mirror: if a column with this name already exists with the |
| 167 | // same type, return its id. With a non-matching type, fail loudly — the |
| 168 | // caller is racing two registrations with different shapes. |
| 169 | const auto& cols = storage.columnDescriptors(); |
| 170 | for (const auto& col : cols) { |
| 171 | if (col.field_path == field_name) { |
| 172 | if (col.logical_type != type) { |
| 173 | return PJ::unexpected( |
| 174 | fmt::format("createTopicField: field '{}' already exists with a different type", field_name)); |
| 175 | } |
| 176 | if (requested_id.has_value() && *requested_id != col.field_id) { |
| 177 | return PJ::unexpected( |
| 178 | fmt::format( |
| 179 | "createTopicField: field '{}' already exists with id {} but requested {}", field_name, col.field_id, |
| 180 | *requested_id)); |
| 181 | } |
| 182 | return col.field_id; |
| 183 | } |
| 184 | } |
| 185 | |
| 186 | // FieldIds are dense starting at 0. The next id is cols.size(). A forced |
| 187 | // requested_id of 0 is a legitimate value (first field of the topic), which |
| 188 | // is exactly why the sentinel for "auto" is std::nullopt, not 0. |
| 189 | const FieldId next_id = static_cast<FieldId>(cols.size()); |
| 190 | if (requested_id.has_value() && *requested_id != next_id) { |
| 191 | return PJ::unexpected( |
| 192 | fmt::format( |
| 193 | "createTopicField: requested_id {} would create a non-dense field layout " |
| 194 | "(topic {} currently has {} columns; next dense id is {})", |
| 195 | *requested_id, topic_id, cols.size(), next_id)); |
| 196 | } |
| 197 | |
| 198 | std::vector<ColumnDescriptor> new_cols = cols; |
| 199 | ColumnDescriptor desc; |
| 200 | desc.field_id = next_id; |
| 201 | desc.logical_type = type; |
| 202 | desc.field_path = std::string(field_name); |
| 203 | new_cols.push_back(std::move(desc)); |
| 204 | storage.setColumnDescriptors(std::move(new_cols)); |
| 205 | return next_id; |
| 206 | } |
| 207 | |
| 208 | TopicStorage* DataEngine::getTopicStorage(TopicId id) { |
| 209 | auto it = impl_->topics.find(id); |