Method resolveInputField

pj_runtime/src/DataProcessorService.cpp:321–386

}

std::optional<DataProcessorService::ResolvedInput> DataProcessorService::resolveInputField(
    const std::string& name) const {
  // Exact whole-topic name → the topic's first leaf column.
  if (const auto topic = resolveInputTopic(name)) {
    return ResolvedInput{topic->first, topic->second, 0};
  }
  // Otherwise treat the name as "<topic>/<field-path>": find the LONGEST live topic
  // whose name prefixes `name` (topic names can themselves contain '/'), then map the
  // trailing field path to its flattened-leaf column index.
  const TopicStorage* best_storage = nullptr;
  TopicId best_tid = 0;
  DatasetId best_ds = 0;
  std::string best_field;
  std::size_t best_topic_len = 0;
  for (const DatasetId ds : engine_.listDatasets()) {
    for (const TopicId tid : engine_.listTopics(ds)) {
      const TopicStorage* storage = engine_.getTopicStorage(tid);
      if (storage == nullptr) {
        continue;
      }
      const std::string& tn = storage->descriptor().name;
      if (name.size() > tn.size() + 1 && name.compare(0, tn.size(), tn) == 0 && name[tn.size()] == '/' &&
          tn.size() > best_topic_len) {
        best_topic_len = tn.size();
        best_storage = storage;
        best_tid = tid;
        best_ds = ds;
        best_field = name.substr(tn.size() + 1);
      }
    }
  }
  if (best_storage == nullptr) {
    return std::nullopt;
  }
  // Map the field path to its leaf column index. The dropped/display name separates
  // nested fields with '/', but both the type tree (flattenFieldPaths) and the stored
  // column descriptors use '.'-separated paths (e.g. "orientation.w"); normalize the
  // queried field to '.' before matching so a nested input resolves, not just a flat
  // one. The returned index is the engine's leaf-column order — exactly what
  // add_mimo_transform reads per input.
  std::string field_dotted = best_field;
  std::replace(field_dotted.begin(), field_dotted.end(), '/', '.');
  const auto matches = [&](const std::string& path) { return path == best_field || path == field_dotted; };

  // Schema-bearing topics (e.g. ROS/CDR): resolve through the TypeRegistry tree.
  const PJ::SchemaId schema_id = best_storage->descriptor().schema_id;
  if (schema_id != 0) {
    if (const PJ::TypeTreeNode* root = engine_.typeRegistry().lookup(schema_id)) {
      const std::vector<std::string> paths = PJ::flattenFieldPaths(*root);
      for (std::size_t i = 0; i < paths.size(); ++i) {
        if (matches(paths[i])) {
          return ResolvedInput{best_tid, best_ds, i};
        }
      }
    }
  }
  // Schema-less topics (schema_id == 0, e.g. JSON scalar-only ingest): the leaf names
  // live in the stored column descriptors, whose order is the engine column index.