| 328 | // --------------------------------------------------------------------------- |
| 329 | |
| 330 | PJ::Expected<PJ::NodeId> DerivedEngine::addSisoTransform( |
| 331 | PJ::TopicId input_topic_id, std::string output_topic_name, PJ::DatasetId output_dataset_id, |
| 332 | std::unique_ptr<ISISOTransform> op, std::size_t input_column_index) { |
| 333 | // Atomic against worker ingest: the input getTopicStorage() read, the |
| 334 | // typeRegistry().registerOrGet() write, and createTopic() (rehash) run as one |
| 335 | // unit. Recursive mutex, so the nested createTopic re-acquires harmlessly. |
| 336 | auto lock = engine_.lockEngine(); |
| 337 | // 1. Check input topic exists |
| 338 | const TopicStorage* in_storage = engine_.getTopicStorage(input_topic_id); |
| 339 | if (!in_storage) { |
| 340 | return PJ::unexpected(fmt::format("add_siso_transform: input topic {} not found", input_topic_id)); |
| 341 | } |
| 342 | |
| 343 | // 2. Determine the single leaf column's StorageKind. |
| 344 | // Prefer TypeRegistry (via schema_id). Fall back to the first sealed chunk's |
| 345 | // column_descriptors when schema_id == 0 (e.g. topics created via |
| 346 | // register_scalar_series, which stores schema only in the writer's internal state). |
| 347 | PJ::SchemaId schema_id = in_storage->descriptor().schema_id; |
| 348 | |
| 349 | std::size_t num_cols = 0; |
| 350 | std::optional<PJ::PrimitiveType> leaf_primitive; |
| 351 | |
| 352 | if (schema_id != 0) { |
| 353 | const PJ::TypeTreeNode* root = engine_.typeRegistry().lookup(schema_id); |
| 354 | if (root) { |
| 355 | num_cols = PJ::countLeafFields(*root); |
| 356 | if (input_column_index < num_cols) { |
| 357 | std::size_t seen = 0; |
| 358 | leaf_primitive = nthLeafPrimitive(*root, input_column_index, seen); |
| 359 | } |
| 360 | } |
| 361 | } |
| 362 | |
| 363 | if (num_cols == 0) { |
| 364 | // Fall back 1: inline column layout stored in TopicStorage at registration time. |
| 365 | // This covers schema_id==0 topics (register_scalar_series) with no committed chunks yet. |
| 366 | const auto& stored = in_storage->columnDescriptors(); |
| 367 | if (!stored.empty()) { |
| 368 | num_cols = stored.size(); |
| 369 | if (input_column_index < num_cols) { |
| 370 | leaf_primitive = stored[input_column_index].logical_type; |
| 371 | } |
| 372 | } |
| 373 | } |
| 374 | |
| 375 | if (num_cols == 0) { |
| 376 | // Fall back 2: first committed chunk's columnDescriptors (legacy path). |
| 377 | const auto& chunks = in_storage->sealedChunks(); |
| 378 | if (!chunks.empty() && !chunks[0].columns.empty()) { |
| 379 | num_cols = chunks[0].columns.size(); |
| 380 | if (input_column_index < num_cols) { |
| 381 | leaf_primitive = chunks[0].columns[input_column_index].descriptor->logical_type; |
| 382 | } |
| 383 | } |
| 384 | } |
| 385 | |
| 386 | if (num_cols == 0) { |
| 387 | return PJ::unexpected( |