| 56 | } |
| 57 | |
| 58 | static Record JoinConsume(OpBase *opBase) { |
| 59 | OpJoin *op = (OpJoin *)opBase; |
| 60 | Record r = NULL; |
| 61 | bool new_stream = false; |
| 62 | |
| 63 | while(!r) { |
| 64 | // Try pulling from current stream. |
| 65 | r = OpBase_Consume(op->stream); |
| 66 | |
| 67 | if(!r) { |
| 68 | // Stream depleted |
| 69 | // Propagate reset to release RediSearch index lock if any exists |
| 70 | OpBase_PropagateReset(op->stream); |
| 71 | // See if there's a new stream to pull from. |
| 72 | op->streamIdx++; |
| 73 | if(op->streamIdx >= op->op.childCount) break; |
| 74 | |
| 75 | op->stream = op->op.children[op->streamIdx]; |
| 76 | // Switched streams, need to update the ResultSet column mapping |
| 77 | new_stream = true; |
| 78 | continue; |
| 79 | } |
| 80 | |
| 81 | if(op->update_column_map && new_stream) { |
| 82 | // We have a new record mapping, update the ResultSet column map to match it. |
| 83 | ResultSet_MapProjection(QueryCtx_GetResultSet(), r->mapping); |
| 84 | } |
| 85 | } |
| 86 | |
| 87 | return r; |
| 88 | } |
| 89 | |
| 90 | static inline OpBase *JoinClone(const ExecutionPlan *plan, const OpBase *opBase) { |
| 91 | ASSERT(opBase->type == OPType_JOIN); |
nothing calls this directly
no test coverage detected