Function runMimoIncremental

pj_datastore/src/derived_engine.cpp:921–1079

// ---------------------------------------------------------------------------

static PJ::Status runMimoIncremental(DerivedEngineImpl& /*impl*/, DataEngine& engine, DerivedNode& node) {
  const std::size_t num_inputs = node.mimo_input_topic_ids.size();
  if (num_inputs == 0) {
    return PJ::okStatus();
  }

  // 1. Collect (timestamp, chunk*, row_index) for each input topic,
  //    only for rows strictly newer than the watermark.
  struct SampleLoc {
    PJ::Timestamp ts;
    const TopicChunk* chunk;
    uint32_t row;
  };
  std::vector<std::vector<SampleLoc>> per_topic(num_inputs);

  PJ::ChunkId max_chunk_seen = node.mimo_last_chunk_id;
  for (std::size_t i = 0; i < num_inputs; ++i) {
    const TopicStorage* storage = engine.getTopicStorage(node.mimo_input_topic_ids[i]);
    if (!storage) {
      return PJ::unexpected(
          fmt::format("run_mimo_incremental: input topic {} not found", node.mimo_input_topic_ids[i]));
    }
    for (const TopicChunk& chunk : storage->sealedChunks()) {
      max_chunk_seen = std::max(max_chunk_seen, chunk.id);
      if (chunk.stats.t_max <= node.mimo_last_ts) {
        continue;  // entire chunk already processed
      }
      for (uint32_t r = 0; r < chunk.stats.row_count; ++r) {
        PJ::Timestamp ts = chunk.timestamps[r];
        if (ts <= node.mimo_last_ts) {
          continue;
        }
        per_topic[i].push_back({ts, &chunk, r});
      }
    }
  }
  // Every committed chunk has now been considered — regression detection in
  // the scheduler compares against this watermark. Updated even when the run
  // produces no joins, so a fruitless chunk is not re-flagged forever.
  node.mimo_last_chunk_id = max_chunk_seen;
  for (std::size_t i = 0; i < num_inputs; ++i) {
    // If any topic has no new data, no new join is possible.
    if (per_topic[i].empty()) {
      return PJ::okStatus();
    }
  }
  // Chunks are gathered in commit order, which under out-of-order ingest is
  // not time order; joined_ts is derived from topic 0, so sort it (stable:
  // duplicate timestamps keep commit order for last-write-wins lookups).
  std::stable_sort(
      per_topic[0].begin(), per_topic[0].end(), [](const SampleLoc& a, const SampleLoc& b) { return a.ts < b.ts; });

  // 2. N-way timestamp intersection: find timestamps present in ALL input topics.
  //    Start from topic 0's sorted timestamps, remove any not in subsequent topics.
  std::vector<PJ::Timestamp> joined_ts;
  joined_ts.reserve(per_topic[0].size());
  for (const auto& s : per_topic[0]) {
    joined_ts.push_back(s.ts);