src/backend/replication/logical/reorderbuffer.c
static ReorderBufferIterTXNState *
ReorderBufferIterTXNInit(ReorderBuffer *rb, ReorderBufferTXN *txn)
{
	Size		nr_txns = 0;
	ReorderBufferIterTXNState *state;
	dlist_iter	cur_txn_i;
	int32		off;

	/*
	 * Calculate the size of our heap: one element for every transaction that
	 * contains changes.  (Besides the transactions already in the reorder
	 * buffer, we count the one we were directly passed.)
	 */
	if (txn->nentries > 0)
		nr_txns++;

	dlist_foreach(cur_txn_i, &txn->subtxns)
	{
		ReorderBufferTXN *cur_txn;

		cur_txn = dlist_container(ReorderBufferTXN, node, cur_txn_i.cur);

		if (cur_txn->nentries > 0)
			nr_txns++;
	}

	/*
	 * TODO: Consider adding fastpath for the rather common nr_txns=1 case, no
	 * need to allocate/build a heap then.
	 */

	/* allocate iteration state */
	state = (ReorderBufferIterTXNState *)
		MemoryContextAllocZero(rb->context,
							   sizeof(ReorderBufferIterTXNState) +
							   sizeof(ReorderBufferIterTXNEntry) * nr_txns);

	state->nr_txns = nr_txns;
	dlist_init(&state->old_change);

	for (off = 0; off < state->nr_txns; off++)
	{
		state->entries[off].fd = -1;
		state->entries[off].segno = 0;
	}

	/* allocate heap */
	state->heap = binaryheap_allocate(state->nr_txns,
									  ReorderBufferIterCompare,
									  state);

	/*
	 * Now insert items into the binary heap, in an unordered fashion.  (We
	 * will run a heap assembly step at the end; this is more efficient.)
	 */

	off = 0;

	/* add toplevel transaction if it contains changes */
	if (txn->nentries > 0)
	{
		ReorderBufferChange *cur_change;

		if (txn->nentries != txn->nentries_mem)
		{
			/* serialize remaining changes */
			ReorderBufferSerializeTXN(rb, txn);
			ReorderBufferRestoreChanges(rb, txn, &state->entries[off].fd,
										&state->entries[off].segno);
		}

		cur_change = dlist_head_element(ReorderBufferChange, node,
										&txn->changes);

		state->entries[off].lsn = cur_change->lsn;
		state->entries[off].change = cur_change;
		state->entries[off].txn = txn;

		binaryheap_add_unordered(state->heap, Int32GetDatum(off++));
	}

	/* add subtransactions if they contain changes */
	dlist_foreach(cur_txn_i, &txn->subtxns)
	{
		ReorderBufferTXN *cur_txn;

		cur_txn = dlist_container(ReorderBufferTXN, node, cur_txn_i.cur);

		if (cur_txn->nentries > 0)
		{
			ReorderBufferChange *cur_change;

			if (cur_txn->nentries != cur_txn->nentries_mem)
			{
				/* serialize remaining changes */
				ReorderBufferSerializeTXN(rb, cur_txn);
				ReorderBufferRestoreChanges(rb, cur_txn,
											&state->entries[off].fd,
											&state->entries[off].segno);
			}
			cur_change = dlist_head_element(ReorderBufferChange, node,
											&cur_txn->changes);

			state->entries[off].lsn = cur_change->lsn;
			state->entries[off].change = cur_change;
			state->entries[off].txn = cur_txn;

			binaryheap_add_unordered(state->heap, Int32GetDatum(off++));
		}
	}

	/* assemble a valid binary heap */
	binaryheap_build(state->heap);

	return state;
}
/*
 * Allocate & initialize an iterator which iterates in lsn order over a
 * transaction and all its subtransactions.
 */
Allocate & initialize an iterator which iterates in lsn order over a
transaction and all its subtransactions.
Allocate & initialize an iterator which iterates in lsn order over a transaction and all its subtransactions .
static ReorderBufferIterTXNState *
ReorderBufferIterTXNInit(ReorderBuffer *rb, ReorderBufferTXN *txn)
{
	Size		nr_txns = 0;
	ReorderBufferIterTXNState *state;
	dlist_iter	cur_txn_i;
	int32		off;
	if (txn->nentries > 0)
		nr_txns++;
	dlist_foreach(cur_txn_i, &txn->subtxns)
	{
		ReorderBufferTXN *cur_txn;
		cur_txn = dlist_container(ReorderBufferTXN, node, cur_txn_i.cur);
		if (cur_txn->nentries > 0)
			nr_txns++;
	}
	state = (ReorderBufferIterTXNState *)
		MemoryContextAllocZero(rb->context,
							   sizeof(ReorderBufferIterTXNState) +
							   sizeof(ReorderBufferIterTXNEntry) * nr_txns);
	state->nr_txns = nr_txns;
	dlist_init(&state->old_change);
	for (off = 0; off < state->nr_txns; off++)
	{
		state->entries[off].fd = -1;
		state->entries[off].segno = 0;
	}
	state->heap = binaryheap_allocate(state->nr_txns,
									  ReorderBufferIterCompare,
									  state);
	off = 0;
	if (txn->nentries > 0)
	{
		ReorderBufferChange *cur_change;
		if (txn->nentries != txn->nentries_mem)
		{
			ReorderBufferSerializeTXN(rb, txn);
			ReorderBufferRestoreChanges(rb, txn, &state->entries[off].fd,
										&state->entries[off].segno);
		}
		cur_change = dlist_head_element(ReorderBufferChange, node,
										&txn->changes);
		state->entries[off].lsn = cur_change->lsn;
		state->entries[off].change = cur_change;
		state->entries[off].txn = txn;
		binaryheap_add_unordered(state->heap, Int32GetDatum(off++));
	}
	dlist_foreach(cur_txn_i, &txn->subtxns)
	{
		ReorderBufferTXN *cur_txn;
		cur_txn = dlist_container(ReorderBufferTXN, node, cur_txn_i.cur);
		if (cur_txn->nentries > 0)
		{
			ReorderBufferChange *cur_change;
			if (cur_txn->nentries != cur_txn->nentries_mem)
			{
				ReorderBufferSerializeTXN(rb, cur_txn);
				ReorderBufferRestoreChanges(rb, cur_txn,
											&state->entries[off].fd,
											&state->entries[off].segno);
			}
			cur_change = dlist_head_element(ReorderBufferChange, node,
											&cur_txn->changes);
			state->entries[off].lsn = cur_change->lsn;
			state->entries[off].change = cur_change;
			state->entries[off].txn = cur_txn;
			binaryheap_add_unordered(state->heap, Int32GetDatum(off++));
		}
	}
	binaryheap_build(state->heap);
	return state;
}
static ReorderBufferIterTXNState * ReorderBufferIterTXNInit ( ReorderBuffer * rb , ReorderBufferTXN * txn ) { Size nr_txns = 0 ; ReorderBufferIterTXNState * state ; dlist_iter cur_txn_i ; int32 off ; if ( txn -> nentries > 0 ) nr_txns ++ ; dlist_foreach ( cur_txn_i , & txn -> subtxns ) { ReorderBufferTXN * cur_txn ; cur_txn = dlist_container ( ReorderBufferTXN , node , cur_txn_i . cur ) ; if ( cur_txn -> nentries > 0 ) nr_txns ++ ; } state = ( ReorderBufferIterTXNState * ) MemoryContextAllocZero ( rb -> context , sizeof ( ReorderBufferIterTXNState ) + sizeof ( ReorderBufferIterTXNEntry ) * nr_txns ) ; state -> nr_txns = nr_txns ; dlist_init ( & state -> old_change ) ; for ( off = 0 ; off < state -> nr_txns ; off ++ ) { state -> entries [ off ] . fd = -1 ; state -> entries [ off ] . segno = 0 ; } state -> heap = binaryheap_allocate ( state -> nr_txns , ReorderBufferIterCompare , state ) ; off = 0 ; if ( txn -> nentries > 0 ) { ReorderBufferChange * cur_change ; if ( txn -> nentries != txn -> nentries_mem ) { ReorderBufferSerializeTXN ( rb , txn ) ; ReorderBufferRestoreChanges ( rb , txn , & state -> entries [ off ] . fd , & state -> entries [ off ] . segno ) ; } cur_change = dlist_head_element ( ReorderBufferChange , node , & txn -> changes ) ; state -> entries [ off ] . lsn = cur_change -> lsn ; state -> entries [ off ] . change = cur_change ; state -> entries [ off ] . txn = txn ; binaryheap_add_unordered ( state -> heap , Int32GetDatum ( off ++ ) ) ; } dlist_foreach ( cur_txn_i , & txn -> subtxns ) { ReorderBufferTXN * cur_txn ; cur_txn = dlist_container ( ReorderBufferTXN , node , cur_txn_i . cur ) ; if ( cur_txn -> nentries > 0 ) { ReorderBufferChange * cur_change ; if ( cur_txn -> nentries != cur_txn -> nentries_mem ) { ReorderBufferSerializeTXN ( rb , cur_txn ) ; ReorderBufferRestoreChanges ( rb , cur_txn , & state -> entries [ off ] . fd , & state -> entries [ off ] . segno ) ; } cur_change = dlist_head_element ( ReorderBufferChange , node , & cur_txn -> changes ) ; state -> entries [ off ] . lsn = cur_change -> lsn ; state -> entries [ off ] . change = cur_change ; state -> entries [ off ] . txn = cur_txn ; binaryheap_add_unordered ( state -> heap , Int32GetDatum ( off ++ ) ) ; } } binaryheap_build ( state -> heap ) ; return state ; }
Allocate & initialize an iterator which iterates in lsn order over a
 transaction and all its subtransactions.
Allocate & initialize an iterator which iterates in lsn order over a transaction and all its subtransactions .
/*
	 * Calculate the size of our heap: one element for every transaction that
	 * contains changes.  (Besides the transactions already in the reorder
	 * buffer, we count the one we were directly passed.)
	 */ /*
	 * TODO: Consider adding fastpath for the rather common nr_txns=1 case, no
	 * need to allocate/build a heap then.
	 */ /* allocate iteration state */ /* allocate heap */ /*
	 * Now insert items into the binary heap, in an unordered fashion.  (We
	 * will run a heap assembly step at the end; this is more efficient.)
	 */ /* add toplevel transaction if it contains changes */ /* serialize remaining changes */ /* add subtransactions if they contain changes */ /* serialize remaining changes */ /* assemble a valid binary heap */