src/backend/replication/logical/reorderbuffer.c
static ReorderBufferChange *
ReorderBufferIterTXNNext(ReorderBuffer *rb, ReorderBufferIterTXNState *state)
{
	ReorderBufferChange *change;
	ReorderBufferIterTXNEntry *entry;
	int32		off;

	/* nothing there anymore */
	if (state->heap->bh_size == 0)
		return NULL;

	off = DatumGetInt32(binaryheap_first(state->heap));
	entry = &state->entries[off];

	/* free memory we might have "leaked" in the previous *Next call */
	if (!dlist_is_empty(&state->old_change))
	{
		change = dlist_container(ReorderBufferChange, node,
								 dlist_pop_head_node(&state->old_change));
		ReorderBufferReturnChange(rb, change);
		Assert(dlist_is_empty(&state->old_change));
	}

	change = entry->change;

	/*
	 * update heap with information about which transaction has the next
	 * relevant change in LSN order
	 */

	/* there are in-memory changes */
	if (dlist_has_next(&entry->txn->changes, &entry->change->node))
	{
		dlist_node *next = dlist_next_node(&entry->txn->changes, &change->node);
		ReorderBufferChange *next_change =
		dlist_container(ReorderBufferChange, node, next);

		/* txn stays the same */
		state->entries[off].lsn = next_change->lsn;
		state->entries[off].change = next_change;

		binaryheap_replace_first(state->heap, Int32GetDatum(off));
		return change;
	}

	/* try to load changes from disk */
	if (entry->txn->nentries != entry->txn->nentries_mem)
	{
		/*
		 * Ugly: restoring changes will reuse *Change records, thus delete the
		 * current one from the per-tx list and only free in the next call.
		 */
		dlist_delete(&change->node);
		dlist_push_tail(&state->old_change, &change->node);

		if (ReorderBufferRestoreChanges(rb, entry->txn, &entry->fd,
										&state->entries[off].segno))
		{
			/* successfully restored changes from disk */
			ReorderBufferChange *next_change =
			dlist_head_element(ReorderBufferChange, node,
							   &entry->txn->changes);

			elog(DEBUG2, "restored %u/%u changes from disk",
				 (uint32) entry->txn->nentries_mem,
				 (uint32) entry->txn->nentries);

			Assert(entry->txn->nentries_mem);
			/* txn stays the same */
			state->entries[off].lsn = next_change->lsn;
			state->entries[off].change = next_change;
			binaryheap_replace_first(state->heap, Int32GetDatum(off));

			return change;
		}
	}

	/* ok, no changes there anymore, remove */
	binaryheap_remove_first(state->heap);

	return change;
}
/*
 * Return the next change when iterating over a transaction and its
 * subtransactions.
 *
 * Returns NULL when no further changes exist.
 */
Return the next change when iterating over a transaction and its
subtransactions.

Returns NULL when no further changes exist.
Return the next change when iterating over a transaction and its subtransactions . Returns NULL when no further changes exist .
static ReorderBufferChange *
ReorderBufferIterTXNNext(ReorderBuffer *rb, ReorderBufferIterTXNState *state)
{
	ReorderBufferChange *change;
	ReorderBufferIterTXNEntry *entry;
	int32		off;
	if (state->heap->bh_size == 0)
		return NULL;
	off = DatumGetInt32(binaryheap_first(state->heap));
	entry = &state->entries[off];
	if (!dlist_is_empty(&state->old_change))
	{
		change = dlist_container(ReorderBufferChange, node,
								 dlist_pop_head_node(&state->old_change));
		ReorderBufferReturnChange(rb, change);
		Assert(dlist_is_empty(&state->old_change));
	}
	change = entry->change;
	if (dlist_has_next(&entry->txn->changes, &entry->change->node))
	{
		dlist_node *next = dlist_next_node(&entry->txn->changes, &change->node);
		ReorderBufferChange *next_change =
		dlist_container(ReorderBufferChange, node, next);
		state->entries[off].lsn = next_change->lsn;
		state->entries[off].change = next_change;
		binaryheap_replace_first(state->heap, Int32GetDatum(off));
		return change;
	}
	if (entry->txn->nentries != entry->txn->nentries_mem)
	{
		dlist_delete(&change->node);
		dlist_push_tail(&state->old_change, &change->node);
		if (ReorderBufferRestoreChanges(rb, entry->txn, &entry->fd,
										&state->entries[off].segno))
		{
			ReorderBufferChange *next_change =
			dlist_head_element(ReorderBufferChange, node,
							   &entry->txn->changes);
			elog(DEBUG2, "restored %u/%u changes from disk",
				 (uint32) entry->txn->nentries_mem,
				 (uint32) entry->txn->nentries);
			Assert(entry->txn->nentries_mem);
			state->entries[off].lsn = next_change->lsn;
			state->entries[off].change = next_change;
			binaryheap_replace_first(state->heap, Int32GetDatum(off));
			return change;
		}
	}
	binaryheap_remove_first(state->heap);
	return change;
}
static ReorderBufferChange * ReorderBufferIterTXNNext ( ReorderBuffer * rb , ReorderBufferIterTXNState * state ) { ReorderBufferChange * change ; ReorderBufferIterTXNEntry * entry ; int32 off ; if ( state -> heap -> bh_size == 0 ) return NULL ; off = DatumGetInt32 ( binaryheap_first ( state -> heap ) ) ; entry = & state -> entries [ off ] ; if ( ! dlist_is_empty ( & state -> old_change ) ) { change = dlist_container ( ReorderBufferChange , node , dlist_pop_head_node ( & state -> old_change ) ) ; ReorderBufferReturnChange ( rb , change ) ; Assert ( dlist_is_empty ( & state -> old_change ) ) ; } change = entry -> change ; if ( dlist_has_next ( & entry -> txn -> changes , & entry -> change -> node ) ) { dlist_node * next = dlist_next_node ( & entry -> txn -> changes , & change -> node ) ; ReorderBufferChange * next_change = dlist_container ( ReorderBufferChange , node , next ) ; state -> entries [ off ] . lsn = next_change -> lsn ; state -> entries [ off ] . change = next_change ; binaryheap_replace_first ( state -> heap , Int32GetDatum ( off ) ) ; return change ; } if ( entry -> txn -> nentries != entry -> txn -> nentries_mem ) { dlist_delete ( & change -> node ) ; dlist_push_tail ( & state -> old_change , & change -> node ) ; if ( ReorderBufferRestoreChanges ( rb , entry -> txn , & entry -> fd , & state -> entries [ off ] . segno ) ) { ReorderBufferChange * next_change = dlist_head_element ( ReorderBufferChange , node , & entry -> txn -> changes ) ; elog ( DEBUG2 , " " , ( uint32 ) entry -> txn -> nentries_mem , ( uint32 ) entry -> txn -> nentries ) ; Assert ( entry -> txn -> nentries_mem ) ; state -> entries [ off ] . lsn = next_change -> lsn ; state -> entries [ off ] . change = next_change ; binaryheap_replace_first ( state -> heap , Int32GetDatum ( off ) ) ; return change ; } } binaryheap_remove_first ( state -> heap ) ; return change ; }
Return the next change when iterating over a transaction and its
 subtransactions.
Return the next change when iterating over a transaction and its subtransactions .
/* nothing there anymore */ /* free memory we might have "leaked" in the previous *Next call */ /*
	 * update heap with information about which transaction has the next
	 * relevant change in LSN order
	 */ /* there are in-memory changes */ /* txn stays the same */ /* try to load changes from disk */ /*
		 * Ugly: restoring changes will reuse *Change records, thus delete the
		 * current one from the per-tx list and only free in the next call.
		 */ /* successfully restored changes from disk */ /* txn stays the same */ /* ok, no changes there anymore, remove */