Function nodeInputRegressed

pj_datastore/src/derived_engine.cpp:1089–1123

True when a not-yet-processed input chunk lands at or before data this node already consumed. Applying it incrementally would feed the transform out of ascending-timestamp order (SISO) or silently skip joined rows (MIMO), so the scheduler must reset + fully replay the node instead.

// ascending-timestamp order (SISO) or silently skip joined rows (MIMO), so
// the scheduler must reset + fully replay the node instead.
static bool nodeInputRegressed(DataEngine& engine, const DerivedNode& node) {
  if (!node.is_mimo) {
    const TopicStorage* in_storage = engine.getTopicStorage(node.siso_input_topic_id);
    if (in_storage == nullptr) {
      return false;
    }
    // Walk unprocessed chunks in commit order: each must start at or after
    // everything fed so far (including its predecessors in this same batch).
    PJ::Timestamp watermark = node.siso_last_ts;
    for (const TopicChunk& chunk : in_storage->sealedChunks()) {
      if (chunk.id <= node.last_processed_chunk_id || chunk.stats.row_count == 0) {
        continue;
      }
      if (chunk.stats.t_min < watermark) {
        return true;
      }
      watermark = std::max(watermark, chunk.stats.t_max);
    }
    return false;
  }
  for (PJ::TopicId in_tid : node.mimo_input_topic_ids) {
    const TopicStorage* in_storage = engine.getTopicStorage(in_tid);
    if (in_storage == nullptr) {
      continue;
    }
    for (const TopicChunk& chunk : in_storage->sealedChunks()) {
      // <= : the MIMO gather skips rows at the watermark timestamp, so a late
      // row exactly at it would otherwise be lost.
      if (chunk.id > node.mimo_last_chunk_id && chunk.stats.row_count > 0 && chunk.stats.t_min <= node.mimo_last_ts) {
        return true;
      }
    }
  }
  return false;
}

// ---------------------------------------------------------------------------
// scheduleAll / scheduleActive