pgrepack.c File Reference

#include "postgres.h"
#include "access/detoast.h"
#include "commands/repack.h"
#include "commands/repack_internal.h"
#include "replication/snapbuild.h"
#include "utils/memutils.h"

Include dependency graph for pgrepack.c:

Go to the source code of this file.

Functions
	PG_MODULE_MAGIC_EXT (.name="pgrepack",.version=PG_VERSION)
static void	repack_startup (LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool is_init)
static void	repack_shutdown (LogicalDecodingContext *ctx)
static void	repack_begin_txn (LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
static void	repack_commit_txn (LogicalDecodingContext *ctx, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
static void	repack_process_change (LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Relation relation, ReorderBufferChange *change)
static void	repack_store_change (LogicalDecodingContext *ctx, Relation relation, ConcurrentChangeKind kind, HeapTuple tuple)
void	_PG_output_plugin_init (OutputPluginCallbacks *cb)

Function Documentation

_PG_output_plugin_init()

void _PG_output_plugin_init

(

OutputPluginCallbacks *

cb

)

Definition at line 39 of file pgrepack.c.

{
    cb->startup_cb = repack_startup;
    cb->begin_cb = repack_begin_txn;
    cb->change_cb = repack_process_change;
    cb->commit_cb = repack_commit_txn;
    cb->shutdown_cb = repack_shutdown;
}

References OutputPluginCallbacks::begin_cb, OutputPluginCallbacks::change_cb, OutputPluginCallbacks::commit_cb, repack_begin_txn(), repack_commit_txn(), repack_process_change(), repack_shutdown(), repack_startup(), OutputPluginCallbacks::shutdown_cb, and OutputPluginCallbacks::startup_cb.

PG_MODULE_MAGIC_EXT()

PG_MODULE_MAGIC_EXT	(	.	name = "pgrepack",
		.	version = PG_VERSION
	)

repack_begin_txn()

static void repack_begin_txn ( LogicalDecodingContext *  ctx, ReorderBufferTXN *  txn  )

static

Definition at line 99 of file pgrepack.c.

{
}

Referenced by _PG_output_plugin_init().

repack_commit_txn()

static void repack_commit_txn ( LogicalDecodingContext *  ctx, ReorderBufferTXN *  txn, XLogRecPtr  commit_lsn  )

static

Definition at line 105 of file pgrepack.c.

{
}

Referenced by _PG_output_plugin_init().

repack_process_change()

static void repack_process_change ( LogicalDecodingContext *  ctx, ReorderBufferTXN *  txn, Relation  relation, ReorderBufferChange *  change  )

static

Definition at line 114 of file pgrepack.c.

{
    RepackDecodingState *private PG_USED_FOR_ASSERTS_ONLY =
        (RepackDecodingState *) ctx->output_writer_private;
 
    /* Changes of other relation should not have been decoded. */
    Assert(RelationGetRelid(relation) == private->relid);
 
    /* Decode entry depending on its type */
    switch (change->action)
    {
        case REORDER_BUFFER_CHANGE_INSERT:
            {
                HeapTuple   newtuple;
 
                newtuple = change->data.tp.newtuple;
 
                /*
                 * Identity checks in the main function should have made this
                 * impossible.
                 */
                if (newtuple == NULL)
                    elog(ERROR, "incomplete insert info");
 
                repack_store_change(ctx, relation, CHANGE_INSERT, newtuple);
            }
            break;
        case REORDER_BUFFER_CHANGE_UPDATE:
            {
                HeapTuple   oldtuple,
                            newtuple;
 
                oldtuple = change->data.tp.oldtuple;
                newtuple = change->data.tp.newtuple;
 
                if (newtuple == NULL)
                    elog(ERROR, "incomplete update info");
 
                if (oldtuple != NULL)
                    repack_store_change(ctx, relation, CHANGE_UPDATE_OLD, oldtuple);
 
                repack_store_change(ctx, relation, CHANGE_UPDATE_NEW, newtuple);
            }
            break;
        case REORDER_BUFFER_CHANGE_DELETE:
            {
                HeapTuple   oldtuple;
 
                oldtuple = change->data.tp.oldtuple;
 
                if (oldtuple == NULL)
                    elog(ERROR, "incomplete delete info");
 
                repack_store_change(ctx, relation, CHANGE_DELETE, oldtuple);
            }
            break;
        default:
 
            /*
             * Should not come here. This includes TRUNCATE of the table being
             * processed. heap_decode() cannot check the file locator easily,
             * but we assume that TRUNCATE uses AccessExclusiveLock on the
             * table so it should not occur during REPACK (CONCURRENTLY).
             */
            Assert(false);
            break;
    }
}

References ReorderBufferChange::action, Assert, CHANGE_DELETE, CHANGE_INSERT, CHANGE_UPDATE_NEW, CHANGE_UPDATE_OLD, ReorderBufferChange::data, elog, ERROR, fb(), ReorderBufferChange::newtuple, ReorderBufferChange::oldtuple, LogicalDecodingContext::output_writer_private, PG_USED_FOR_ASSERTS_ONLY, RelationGetRelid, REORDER_BUFFER_CHANGE_DELETE, REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_UPDATE, repack_store_change(), and ReorderBufferChange::tp.

Referenced by _PG_output_plugin_init().

repack_shutdown()

static void repack_shutdown ( LogicalDecodingContext *  ctx )

static

Definition at line 85 of file pgrepack.c.

{
}

Referenced by _PG_output_plugin_init().

repack_startup()

static void repack_startup ( LogicalDecodingContext *  ctx, OutputPluginOptions *  opt, bool  is_init  )

static

Definition at line 51 of file pgrepack.c.

{
    RepackDecodingState *dstate;
 
    if (!AmRepackWorker())
        ereport(ERROR,
                errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                errmsg("unsupported use of logical decoding plugin \"%s\"",
                       "pgrepack"),
                errdetail("This plugin can only be used by %s.",
                          "REPACK (CONCURRENTLY)"));
 
    /* Initial setup of our private state */
    Assert(CurrentMemoryContext == ctx->context);
    dstate = palloc0_object(RepackDecodingState);
    dstate->change_cxt = AllocSetContextCreate(ctx->context,
                                               "REPACK - change",
                                               ALLOCSET_DEFAULT_SIZES);
    /* repack_setup_logical_decoding fills in the rest */
    ctx->output_writer_private = dstate;
 
    /* Probably unnecessary, as we don't use the SQL interface ... */
    opt->output_type = OUTPUT_PLUGIN_BINARY_OUTPUT;
 
    if (ctx->output_plugin_options != NIL)
    {
        ereport(ERROR,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                errmsg("this plugin does not expect any options"));
    }
}

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, AmRepackWorker(), Assert, LogicalDecodingContext::context, CurrentMemoryContext, ereport, errcode(), errdetail(), errmsg, ERROR, fb(), NIL, OUTPUT_PLUGIN_BINARY_OUTPUT, LogicalDecodingContext::output_plugin_options, OutputPluginOptions::output_type, LogicalDecodingContext::output_writer_private, and palloc0_object.

Referenced by _PG_output_plugin_init().

repack_store_change()

static void repack_store_change ( LogicalDecodingContext *  ctx, Relation  relation, ConcurrentChangeKind  kind, HeapTuple  tuple  )

static

Definition at line 194 of file pgrepack.c.

{
    RepackDecodingState *dstate;
    MemoryContext oldcxt;
    BufFile    *file;
    List       *attrs_ext = NIL;
    int         natt_ext;
 
    dstate = (RepackDecodingState *) ctx->output_writer_private;
    file = dstate->file;
 
    /* Store the change kind. */
    BufFileWrite(file, &kind, 1);
 
    /* Use a frequently-reset context to avoid dealing with leaks manually */
    oldcxt = MemoryContextSwitchTo(dstate->change_cxt);
 
    /*
     * If the tuple contains "external indirect" attributes, we need to write
     * the contents to the file because we have no control over that memory.
     */
    if (HeapTupleHasExternal(tuple))
    {
        TupleDesc   desc = RelationGetDescr(relation);
        TupleTableSlot *slot;
 
        /* Initialize the slot, if not done already */
        if (dstate->slot == NULL)
        {
            ResourceOwner saveResourceOwner;
 
            MemoryContextSwitchTo(dstate->worker_cxt);
            saveResourceOwner = CurrentResourceOwner;
            CurrentResourceOwner = dstate->worker_resowner;
            dstate->slot = MakeSingleTupleTableSlot(desc, &TTSOpsHeapTuple);
            MemoryContextSwitchTo(dstate->change_cxt);
            CurrentResourceOwner = saveResourceOwner;
        }
 
        slot = dstate->slot;
        ExecStoreHeapTuple(tuple, slot, false);
 
        /*
         * Loop over all attributes, and find out which ones we need to spill
         * separately, to wit: each one that's a non-null varlena and stored
         * out of line.
         */
        for (int i = 0; i < desc->natts; i++)
        {
            CompactAttribute *attr = TupleDescCompactAttr(desc, i);
            varlena    *varlen;
 
            if (attr->attisdropped || attr->attlen != -1 ||
                slot_attisnull(slot, i + 1))
                continue;
 
            slot_getsomeattrs(slot, i + 1);
 
            /*
             * This is a non-null varlena datum, but we only care if it's
             * out-of-line
             */
            varlen = (varlena *) DatumGetPointer(slot->tts_values[i]);
            if (!VARATT_IS_EXTERNAL(varlen))
                continue;
 
            /*
             * We spill any indirect-external attributes separately from the
             * heap tuple.  Anything else is written as is.
             */
            if (VARATT_IS_EXTERNAL_INDIRECT(varlen))
                attrs_ext = lappend(attrs_ext, varlen);
            else
            {
                /*
                 * Logical decoding should not produce "external expanded"
                 * attributes (those actually should never appear on disk), so
                 * only TOASTed attribute can be seen here.
                 *
                 * We get here if the table has external values but only
                 * in-line values are being updated now.
                 */
                Assert(VARATT_IS_EXTERNAL_ONDISK(varlen));
            }
        }
 
        ExecClearTuple(slot);
    }
 
    /*
     * First, write the original heap tuple, prefixed by its length.  Note
     * that the external-toast tag for each toasted attribute will be present
     * in what we write, so that we know where to restore each one later.
     */
    BufFileWrite(file, &tuple->t_len, sizeof(tuple->t_len));
    BufFileWrite(file, tuple->t_data, tuple->t_len);
 
    /* Then, write the number of external attributes we found. */
    natt_ext = list_length(attrs_ext);
    BufFileWrite(file, &natt_ext, sizeof(natt_ext));
 
    /* Finally, the attributes themselves, if any */
    foreach_ptr(varlena, attr_val, attrs_ext)
    {
        attr_val = detoast_external_attr(attr_val);
        BufFileWrite(file, attr_val, VARSIZE_ANY(attr_val));
        /* These attributes could be large, so free them right away */
        pfree(attr_val);
    }
 
    /* Cleanup. */
    MemoryContextSwitchTo(oldcxt);
    MemoryContextReset(dstate->change_cxt);
}

References Assert, CompactAttribute::attisdropped, CompactAttribute::attlen, BufFileWrite(), CurrentResourceOwner, DatumGetPointer(), detoast_external_attr(), ExecClearTuple(), ExecStoreHeapTuple(), fb(), foreach_ptr, HeapTupleHasExternal(), i, lappend(), list_length(), MakeSingleTupleTableSlot(), MemoryContextReset(), MemoryContextSwitchTo(), TupleDescData::natts, NIL, LogicalDecodingContext::output_writer_private, pfree(), RelationGetDescr, slot_attisnull(), slot_getsomeattrs(), HeapTupleData::t_data, HeapTupleData::t_len, TupleTableSlot::tts_values, TTSOpsHeapTuple, TupleDescCompactAttr(), VARATT_IS_EXTERNAL(), VARATT_IS_EXTERNAL_INDIRECT(), VARATT_IS_EXTERNAL_ONDISK(), and VARSIZE_ANY().

Referenced by repack_process_change().