large_object.h File Reference

#include "utils/snapshot.h"

Include dependency graph for large_object.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	LargeObjectDesc

Macros
#define	IFS_RDLOCK   (1 << 0) /* LO was opened for reading */
#define	IFS_WRLOCK   (1 << 1) /* LO was opened for writing */
#define	IFS_RD_PERM_OK   (1 << 2) /* read permission has been verified */
#define	IFS_WR_PERM_OK   (1 << 3) /* write permission has been verified */
#define	LOBLKSIZE   (BLCKSZ / 4)
#define	MAX_LARGE_OBJECT_SIZE   ((int64) INT_MAX * LOBLKSIZE)

Typedefs
typedef struct LargeObjectDesc	LargeObjectDesc

Functions
void	close_lo_relation (bool isCommit)
Oid	inv_create (Oid lobjId)
LargeObjectDesc *	inv_open (Oid lobjId, int flags, MemoryContext mcxt)
void	inv_close (LargeObjectDesc *obj_desc)
int	inv_drop (Oid lobjId)
int64	inv_seek (LargeObjectDesc *obj_desc, int64 offset, int whence)
int64	inv_tell (LargeObjectDesc *obj_desc)
int	inv_read (LargeObjectDesc *obj_desc, char *buf, int nbytes)
int	inv_write (LargeObjectDesc *obj_desc, const char *buf, int nbytes)
void	inv_truncate (LargeObjectDesc *obj_desc, int64 len)

Macro Definition Documentation

#define IFS_RD_PERM_OK   (1 << 2) /* read permission has been verified */

Definition at line 50 of file large_object.h.

Referenced by lo_read().

#define IFS_RDLOCK   (1 << 0) /* LO was opened for reading */

Definition at line 48 of file large_object.h.

Referenced by inv_open().

#define IFS_WR_PERM_OK   (1 << 3) /* write permission has been verified */

Definition at line 51 of file large_object.h.

Referenced by lo_truncate_internal(), and lo_write().

#define IFS_WRLOCK   (1 << 1) /* LO was opened for writing */

Definition at line 49 of file large_object.h.

Referenced by inv_open(), inv_truncate(), inv_write(), lo_truncate_internal(), and lo_write().

#define LOBLKSIZE   (BLCKSZ / 4)

Definition at line 72 of file large_object.h.

Referenced by getdatafield(), GuessControlValues(), inv_getsize(), inv_read(), inv_truncate(), inv_write(), ReadControlFile(), and WriteControlFile().

#define MAX_LARGE_OBJECT_SIZE   ((int64) INT_MAX * LOBLKSIZE)

Definition at line 78 of file large_object.h.

Referenced by inv_seek(), inv_truncate(), and inv_write().

Typedef Documentation

typedef struct LargeObjectDesc LargeObjectDesc

Function Documentation

void close_lo_relation

(

bool

isCommit

)

Definition at line 102 of file inv_api.c.

References CurrentResourceOwner, heap_close, index_close(), NoLock, NULL, PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, and TopTransactionResourceOwner.

Referenced by AtEOXact_LargeObject().

{
    if (lo_heap_r || lo_index_r)
    {
        /*
         * Only bother to close if committing; else abort cleanup will handle
         * it
         */
        if (isCommit)
        {
            ResourceOwner currentOwner;

            currentOwner = CurrentResourceOwner;
            PG_TRY();
            {
                CurrentResourceOwner = TopTransactionResourceOwner;

                if (lo_index_r)
                    index_close(lo_index_r, NoLock);
                if (lo_heap_r)
                    heap_close(lo_heap_r, NoLock);
            }
            PG_CATCH();
            {
                /* Ensure CurrentResourceOwner is restored on error */
                CurrentResourceOwner = currentOwner;
                PG_RE_THROW();
            }
            PG_END_TRY();
            CurrentResourceOwner = currentOwner;
        }
        lo_heap_r = NULL;
        lo_index_r = NULL;
    }
}

void inv_close

(

LargeObjectDesc *

obj_desc

)

Definition at line 320 of file inv_api.c.

References Assert, pfree(), PointerIsValid, LargeObjectDesc::snapshot, TopTransactionResourceOwner, and UnregisterSnapshotFromOwner().

Referenced by AtEOSubXact_LargeObject(), AtEOXact_LargeObject(), be_lo_close(), be_lo_export(), be_lo_from_bytea(), be_lo_put(), be_lo_unlink(), lo_get_fragment_internal(), and lo_import_internal().

{
    Assert(PointerIsValid(obj_desc));

    UnregisterSnapshotFromOwner(obj_desc->snapshot,
                                TopTransactionResourceOwner);

    pfree(obj_desc);
}

Oid inv_create

(

Oid

lobjId

)

Definition at line 224 of file inv_api.c.

References CommandCounterIncrement(), GetUserId(), InvokeObjectPostCreateHook, LargeObjectCreate(), LargeObjectRelationId, and recordDependencyOnOwner().

Referenced by be_lo_creat(), be_lo_create(), be_lo_from_bytea(), and lo_import_internal().

{
    Oid         lobjId_new;

    /*
     * Create a new largeobject with empty data pages
     */
    lobjId_new = LargeObjectCreate(lobjId);

    /*
     * dependency on the owner of largeobject
     *
     * The reason why we use LargeObjectRelationId instead of
     * LargeObjectMetadataRelationId here is to provide backward compatibility
     * to the applications which utilize a knowledge about internal layout of
     * system catalogs. OID of pg_largeobject_metadata and loid of
     * pg_largeobject are same value, so there are no actual differences here.
     */
    recordDependencyOnOwner(LargeObjectRelationId,
                            lobjId_new, GetUserId());

    /* Post creation hook for new large object */
    InvokeObjectPostCreateHook(LargeObjectRelationId, lobjId_new, 0);

    /*
     * Advance command counter to make new tuple visible to later operations.
     */
    CommandCounterIncrement();

    return lobjId_new;
}

int inv_drop

(

Oid

lobjId

)

Definition at line 336 of file inv_api.c.

References ObjectAddress::classId, CommandCounterIncrement(), DROP_CASCADE, LargeObjectRelationId, and performDeletion().

Referenced by be_lo_unlink().

{
    ObjectAddress object;

    /*
     * Delete any comments and dependencies on the large object
     */
    object.classId = LargeObjectRelationId;
    object.objectId = lobjId;
    object.objectSubId = 0;
    performDeletion(&object, DROP_CASCADE, 0);

    /*
     * Advance command counter so that tuple removal will be seen by later
     * large-object operations in this transaction.
     */
    CommandCounterIncrement();

    return 1;
}

LargeObjectDesc* inv_open	(	Oid	lobjId,
		int	flags,
		MemoryContext	mcxt
	)

Definition at line 265 of file inv_api.c.

References ereport, errcode(), errmsg(), ERROR, LargeObjectDesc::flags, GetActiveSnapshot(), GetCurrentSubTransactionId(), LargeObjectDesc::id, IFS_RDLOCK, IFS_WRLOCK, INV_READ, INV_WRITE, MemoryContextAlloc(), myLargeObjectExists(), NULL, LargeObjectDesc::offset, RegisterSnapshotOnOwner(), LargeObjectDesc::snapshot, LargeObjectDesc::subid, and TopTransactionResourceOwner.

Referenced by be_lo_export(), be_lo_from_bytea(), be_lo_open(), be_lo_put(), lo_get_fragment_internal(), and lo_import_internal().

{
    LargeObjectDesc *retval;
    Snapshot    snapshot = NULL;
    int         descflags = 0;

    if (flags & INV_WRITE)
    {
        snapshot = NULL;        /* instantaneous MVCC snapshot */
        descflags = IFS_WRLOCK | IFS_RDLOCK;
    }
    else if (flags & INV_READ)
    {
        snapshot = GetActiveSnapshot();
        descflags = IFS_RDLOCK;
    }
    else
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("invalid flags for opening a large object: %d",
                        flags)));

    /* Can't use LargeObjectExists here because we need to specify snapshot */
    if (!myLargeObjectExists(lobjId, snapshot))
        ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_OBJECT),
                 errmsg("large object %u does not exist", lobjId)));

    /*
     * We must register the snapshot in TopTransaction's resowner, because it
     * must stay alive until the LO is closed rather than until the current
     * portal shuts down. Do this after checking that the LO exists, to avoid
     * leaking the snapshot if an error is thrown.
     */
    if (snapshot)
        snapshot = RegisterSnapshotOnOwner(snapshot,
                                           TopTransactionResourceOwner);

    /* All set, create a descriptor */
    retval = (LargeObjectDesc *) MemoryContextAlloc(mcxt,
                                                    sizeof(LargeObjectDesc));
    retval->id = lobjId;
    retval->subid = GetCurrentSubTransactionId();
    retval->offset = 0;
    retval->snapshot = snapshot;
    retval->flags = descflags;

    return retval;
}

int inv_read	(	LargeObjectDesc *	obj_desc,
		char *	buf,
		int	nbytes
	)

Definition at line 464 of file inv_api.c.

References Anum_pg_largeobject_loid, Anum_pg_largeobject_pageno, Assert, BTEqualStrategyNumber, BTGreaterEqualStrategyNumber, elog, ERROR, ForwardScanDirection, getdatafield(), GETSTRUCT, HeapTupleHasNulls, LargeObjectDesc::id, Int32GetDatum, LOBLKSIZE, MemSet, NULL, ObjectIdGetDatum, LargeObjectDesc::offset, open_lo_relation(), pfree(), PointerIsValid, ScanKeyInit(), LargeObjectDesc::snapshot, systable_beginscan_ordered(), systable_endscan_ordered(), systable_getnext_ordered(), and VARDATA.

Referenced by be_lo_export(), lo_get_fragment_internal(), and lo_read().

{
    int         nread = 0;
    int64       n;
    int64       off;
    int         len;
    int32       pageno = (int32) (obj_desc->offset / LOBLKSIZE);
    uint64      pageoff;
    ScanKeyData skey[2];
    SysScanDesc sd;
    HeapTuple   tuple;

    Assert(PointerIsValid(obj_desc));
    Assert(buf != NULL);

    if (nbytes <= 0)
        return 0;

    open_lo_relation();

    ScanKeyInit(&skey[0],
                Anum_pg_largeobject_loid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(obj_desc->id));

    ScanKeyInit(&skey[1],
                Anum_pg_largeobject_pageno,
                BTGreaterEqualStrategyNumber, F_INT4GE,
                Int32GetDatum(pageno));

    sd = systable_beginscan_ordered(lo_heap_r, lo_index_r,
                                    obj_desc->snapshot, 2, skey);

    while ((tuple = systable_getnext_ordered(sd, ForwardScanDirection)) != NULL)
    {
        Form_pg_largeobject data;
        bytea      *datafield;
        bool        pfreeit;

        if (HeapTupleHasNulls(tuple))   /* paranoia */
            elog(ERROR, "null field found in pg_largeobject");
        data = (Form_pg_largeobject) GETSTRUCT(tuple);

        /*
         * We expect the indexscan will deliver pages in order.  However,
         * there may be missing pages if the LO contains unwritten "holes". We
         * want missing sections to read out as zeroes.
         */
        pageoff = ((uint64) data->pageno) * LOBLKSIZE;
        if (pageoff > obj_desc->offset)
        {
            n = pageoff - obj_desc->offset;
            n = (n <= (nbytes - nread)) ? n : (nbytes - nread);
            MemSet(buf + nread, 0, n);
            nread += n;
            obj_desc->offset += n;
        }

        if (nread < nbytes)
        {
            Assert(obj_desc->offset >= pageoff);
            off = (int) (obj_desc->offset - pageoff);
            Assert(off >= 0 && off < LOBLKSIZE);

            getdatafield(data, &datafield, &len, &pfreeit);
            if (len > off)
            {
                n = len - off;
                n = (n <= (nbytes - nread)) ? n : (nbytes - nread);
                memcpy(buf + nread, VARDATA(datafield) + off, n);
                nread += n;
                obj_desc->offset += n;
            }
            if (pfreeit)
                pfree(datafield);
        }

        if (nread >= nbytes)
            break;
    }

    systable_endscan_ordered(sd);

    return nread;
}

int64 inv_seek	(	LargeObjectDesc *	obj_desc,
		int64	offset,
		int	whence
	)

Definition at line 412 of file inv_api.c.

References Assert, ereport, errcode(), errmsg(), errmsg_internal(), ERROR, INT64_FORMAT, inv_getsize(), MAX_LARGE_OBJECT_SIZE, LargeObjectDesc::offset, and PointerIsValid.

Referenced by be_lo_lseek(), be_lo_lseek64(), be_lo_put(), and lo_get_fragment_internal().

{
    int64       newoffset;

    Assert(PointerIsValid(obj_desc));

    /*
     * Note: overflow in the additions is possible, but since we will reject
     * negative results, we don't need any extra test for that.
     */
    switch (whence)
    {
        case SEEK_SET:
            newoffset = offset;
            break;
        case SEEK_CUR:
            newoffset = obj_desc->offset + offset;
            break;
        case SEEK_END:
            newoffset = inv_getsize(obj_desc) + offset;
            break;
        default:
            ereport(ERROR,
                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                     errmsg("invalid whence setting: %d", whence)));
            newoffset = 0;      /* keep compiler quiet */
            break;
    }

    /*
     * use errmsg_internal here because we don't want to expose INT64_FORMAT
     * in translatable strings; doing better is not worth the trouble
     */
    if (newoffset < 0 || newoffset > MAX_LARGE_OBJECT_SIZE)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg_internal("invalid large object seek target: " INT64_FORMAT,
                                 newoffset)));

    obj_desc->offset = newoffset;
    return newoffset;
}

int64 inv_tell

(

LargeObjectDesc *

obj_desc

)

Definition at line 456 of file inv_api.c.

References Assert, LargeObjectDesc::offset, and PointerIsValid.

Referenced by be_lo_tell(), and be_lo_tell64().

{
    Assert(PointerIsValid(obj_desc));

    return obj_desc->offset;
}

void inv_truncate	(	LargeObjectDesc *	obj_desc,
		int64	len
	)

Definition at line 744 of file inv_api.c.

References Anum_pg_largeobject_data, Anum_pg_largeobject_loid, Anum_pg_largeobject_pageno, Assert, BTEqualStrategyNumber, BTGreaterEqualStrategyNumber, CatalogCloseIndexes(), CatalogOpenIndexes(), CatalogTupleDelete(), CatalogTupleInsertWithInfo(), CatalogTupleUpdateWithInfo(), CommandCounterIncrement(), elog, ereport, errcode(), errmsg_internal(), ERROR, LargeObjectDesc::flags, ForwardScanDirection, getdatafield(), GETSTRUCT, heap_form_tuple(), heap_freetuple(), heap_modify_tuple(), HeapTupleHasNulls, LargeObjectDesc::id, IFS_WRLOCK, Int32GetDatum, INT64_FORMAT, LOBLKSIZE, MAX_LARGE_OBJECT_SIZE, MemSet, Natts_pg_largeobject, NULL, ObjectIdGetDatum, open_lo_relation(), pfree(), PointerGetDatum, PointerIsValid, RelationData::rd_att, RelationGetDescr, ScanKeyInit(), SET_VARSIZE, LargeObjectDesc::snapshot, systable_beginscan_ordered(), systable_endscan_ordered(), systable_getnext_ordered(), HeapTupleData::t_self, values, VARDATA, and VARHDRSZ.

Referenced by lo_truncate_internal().

{
    int32       pageno = (int32) (len / LOBLKSIZE);
    int32       off;
    ScanKeyData skey[2];
    SysScanDesc sd;
    HeapTuple   oldtuple;
    Form_pg_largeobject olddata;
    union
    {
        bytea       hdr;
        /* this is to make the union big enough for a LO data chunk: */
        char        data[LOBLKSIZE + VARHDRSZ];
        /* ensure union is aligned well enough: */
        int32       align_it;
    }           workbuf;
    char       *workb = VARDATA(&workbuf.hdr);
    HeapTuple   newtup;
    Datum       values[Natts_pg_largeobject];
    bool        nulls[Natts_pg_largeobject];
    bool        replace[Natts_pg_largeobject];
    CatalogIndexState indstate;

    Assert(PointerIsValid(obj_desc));

    /* enforce writability because snapshot is probably wrong otherwise */
    Assert(obj_desc->flags & IFS_WRLOCK);

    /*
     * use errmsg_internal here because we don't want to expose INT64_FORMAT
     * in translatable strings; doing better is not worth the trouble
     */
    if (len < 0 || len > MAX_LARGE_OBJECT_SIZE)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg_internal("invalid large object truncation target: " INT64_FORMAT,
                                 len)));

    open_lo_relation();

    indstate = CatalogOpenIndexes(lo_heap_r);

    /*
     * Set up to find all pages with desired loid and pageno >= target
     */
    ScanKeyInit(&skey[0],
                Anum_pg_largeobject_loid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(obj_desc->id));

    ScanKeyInit(&skey[1],
                Anum_pg_largeobject_pageno,
                BTGreaterEqualStrategyNumber, F_INT4GE,
                Int32GetDatum(pageno));

    sd = systable_beginscan_ordered(lo_heap_r, lo_index_r,
                                    obj_desc->snapshot, 2, skey);

    /*
     * If possible, get the page the truncation point is in. The truncation
     * point may be beyond the end of the LO or in a hole.
     */
    olddata = NULL;
    if ((oldtuple = systable_getnext_ordered(sd, ForwardScanDirection)) != NULL)
    {
        if (HeapTupleHasNulls(oldtuple))    /* paranoia */
            elog(ERROR, "null field found in pg_largeobject");
        olddata = (Form_pg_largeobject) GETSTRUCT(oldtuple);
        Assert(olddata->pageno >= pageno);
    }

    /*
     * If we found the page of the truncation point we need to truncate the
     * data in it.  Otherwise if we're in a hole, we need to create a page to
     * mark the end of data.
     */
    if (olddata != NULL && olddata->pageno == pageno)
    {
        /* First, load old data into workbuf */
        bytea      *datafield;
        int         pagelen;
        bool        pfreeit;

        getdatafield(olddata, &datafield, &pagelen, &pfreeit);
        memcpy(workb, VARDATA(datafield), pagelen);
        if (pfreeit)
            pfree(datafield);

        /*
         * Fill any hole
         */
        off = len % LOBLKSIZE;
        if (off > pagelen)
            MemSet(workb + pagelen, 0, off - pagelen);

        /* compute length of new page */
        SET_VARSIZE(&workbuf.hdr, off + VARHDRSZ);

        /*
         * Form and insert updated tuple
         */
        memset(values, 0, sizeof(values));
        memset(nulls, false, sizeof(nulls));
        memset(replace, false, sizeof(replace));
        values[Anum_pg_largeobject_data - 1] = PointerGetDatum(&workbuf);
        replace[Anum_pg_largeobject_data - 1] = true;
        newtup = heap_modify_tuple(oldtuple, RelationGetDescr(lo_heap_r),
                                   values, nulls, replace);
        CatalogTupleUpdateWithInfo(lo_heap_r, &newtup->t_self, newtup,
                                   indstate);
        heap_freetuple(newtup);
    }
    else
    {
        /*
         * If the first page we found was after the truncation point, we're in
         * a hole that we'll fill, but we need to delete the later page
         * because the loop below won't visit it again.
         */
        if (olddata != NULL)
        {
            Assert(olddata->pageno > pageno);
            CatalogTupleDelete(lo_heap_r, &oldtuple->t_self);
        }

        /*
         * Write a brand new page.
         *
         * Fill the hole up to the truncation point
         */
        off = len % LOBLKSIZE;
        if (off > 0)
            MemSet(workb, 0, off);

        /* compute length of new page */
        SET_VARSIZE(&workbuf.hdr, off + VARHDRSZ);

        /*
         * Form and insert new tuple
         */
        memset(values, 0, sizeof(values));
        memset(nulls, false, sizeof(nulls));
        values[Anum_pg_largeobject_loid - 1] = ObjectIdGetDatum(obj_desc->id);
        values[Anum_pg_largeobject_pageno - 1] = Int32GetDatum(pageno);
        values[Anum_pg_largeobject_data - 1] = PointerGetDatum(&workbuf);
        newtup = heap_form_tuple(lo_heap_r->rd_att, values, nulls);
        CatalogTupleInsertWithInfo(lo_heap_r, newtup, indstate);
        heap_freetuple(newtup);
    }

    /*
     * Delete any pages after the truncation point.  If the initial search
     * didn't find a page, then of course there's nothing more to do.
     */
    if (olddata != NULL)
    {
        while ((oldtuple = systable_getnext_ordered(sd, ForwardScanDirection)) != NULL)
        {
            CatalogTupleDelete(lo_heap_r, &oldtuple->t_self);
        }
    }

    systable_endscan_ordered(sd);

    CatalogCloseIndexes(indstate);

    /*
     * Advance command counter so that tuple updates will be seen by later
     * large-object operations in this transaction.
     */
    CommandCounterIncrement();
}

int inv_write	(	LargeObjectDesc *	obj_desc,
		const char *	buf,
		int	nbytes
	)

Definition at line 551 of file inv_api.c.

References Anum_pg_largeobject_data, Anum_pg_largeobject_loid, Anum_pg_largeobject_pageno, Assert, BTEqualStrategyNumber, BTGreaterEqualStrategyNumber, CatalogCloseIndexes(), CatalogOpenIndexes(), CatalogTupleInsertWithInfo(), CatalogTupleUpdateWithInfo(), CommandCounterIncrement(), elog, ereport, errcode(), errmsg(), ERROR, LargeObjectDesc::flags, ForwardScanDirection, getdatafield(), GETSTRUCT, heap_form_tuple(), heap_freetuple(), heap_modify_tuple(), HeapTupleHasNulls, LargeObjectDesc::id, IFS_WRLOCK, Int32GetDatum, LOBLKSIZE, MAX_LARGE_OBJECT_SIZE, MemSet, Natts_pg_largeobject, NULL, ObjectIdGetDatum, LargeObjectDesc::offset, open_lo_relation(), pfree(), PointerGetDatum, PointerIsValid, RelationData::rd_att, RelationGetDescr, ScanKeyInit(), SET_VARSIZE, LargeObjectDesc::snapshot, systable_beginscan_ordered(), systable_endscan_ordered(), systable_getnext_ordered(), HeapTupleData::t_self, values, VARDATA, and VARHDRSZ.

Referenced by be_lo_from_bytea(), be_lo_put(), lo_import_internal(), and lo_write().

{
    int         nwritten = 0;
    int         n;
    int         off;
    int         len;
    int32       pageno = (int32) (obj_desc->offset / LOBLKSIZE);
    ScanKeyData skey[2];
    SysScanDesc sd;
    HeapTuple   oldtuple;
    Form_pg_largeobject olddata;
    bool        neednextpage;
    bytea      *datafield;
    bool        pfreeit;
    union
    {
        bytea       hdr;
        /* this is to make the union big enough for a LO data chunk: */
        char        data[LOBLKSIZE + VARHDRSZ];
        /* ensure union is aligned well enough: */
        int32       align_it;
    }           workbuf;
    char       *workb = VARDATA(&workbuf.hdr);
    HeapTuple   newtup;
    Datum       values[Natts_pg_largeobject];
    bool        nulls[Natts_pg_largeobject];
    bool        replace[Natts_pg_largeobject];
    CatalogIndexState indstate;

    Assert(PointerIsValid(obj_desc));
    Assert(buf != NULL);

    /* enforce writability because snapshot is probably wrong otherwise */
    Assert(obj_desc->flags & IFS_WRLOCK);

    if (nbytes <= 0)
        return 0;

    /* this addition can't overflow because nbytes is only int32 */
    if ((nbytes + obj_desc->offset) > MAX_LARGE_OBJECT_SIZE)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("invalid large object write request size: %d",
                        nbytes)));

    open_lo_relation();

    indstate = CatalogOpenIndexes(lo_heap_r);

    ScanKeyInit(&skey[0],
                Anum_pg_largeobject_loid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(obj_desc->id));

    ScanKeyInit(&skey[1],
                Anum_pg_largeobject_pageno,
                BTGreaterEqualStrategyNumber, F_INT4GE,
                Int32GetDatum(pageno));

    sd = systable_beginscan_ordered(lo_heap_r, lo_index_r,
                                    obj_desc->snapshot, 2, skey);

    oldtuple = NULL;
    olddata = NULL;
    neednextpage = true;

    while (nwritten < nbytes)
    {
        /*
         * If possible, get next pre-existing page of the LO.  We expect the
         * indexscan will deliver these in order --- but there may be holes.
         */
        if (neednextpage)
        {
            if ((oldtuple = systable_getnext_ordered(sd, ForwardScanDirection)) != NULL)
            {
                if (HeapTupleHasNulls(oldtuple))    /* paranoia */
                    elog(ERROR, "null field found in pg_largeobject");
                olddata = (Form_pg_largeobject) GETSTRUCT(oldtuple);
                Assert(olddata->pageno >= pageno);
            }
            neednextpage = false;
        }

        /*
         * If we have a pre-existing page, see if it is the page we want to
         * write, or a later one.
         */
        if (olddata != NULL && olddata->pageno == pageno)
        {
            /*
             * Update an existing page with fresh data.
             *
             * First, load old data into workbuf
             */
            getdatafield(olddata, &datafield, &len, &pfreeit);
            memcpy(workb, VARDATA(datafield), len);
            if (pfreeit)
                pfree(datafield);

            /*
             * Fill any hole
             */
            off = (int) (obj_desc->offset % LOBLKSIZE);
            if (off > len)
                MemSet(workb + len, 0, off - len);

            /*
             * Insert appropriate portion of new data
             */
            n = LOBLKSIZE - off;
            n = (n <= (nbytes - nwritten)) ? n : (nbytes - nwritten);
            memcpy(workb + off, buf + nwritten, n);
            nwritten += n;
            obj_desc->offset += n;
            off += n;
            /* compute valid length of new page */
            len = (len >= off) ? len : off;
            SET_VARSIZE(&workbuf.hdr, len + VARHDRSZ);

            /*
             * Form and insert updated tuple
             */
            memset(values, 0, sizeof(values));
            memset(nulls, false, sizeof(nulls));
            memset(replace, false, sizeof(replace));
            values[Anum_pg_largeobject_data - 1] = PointerGetDatum(&workbuf);
            replace[Anum_pg_largeobject_data - 1] = true;
            newtup = heap_modify_tuple(oldtuple, RelationGetDescr(lo_heap_r),
                                       values, nulls, replace);
            CatalogTupleUpdateWithInfo(lo_heap_r, &newtup->t_self, newtup,
                                       indstate);
            heap_freetuple(newtup);

            /*
             * We're done with this old page.
             */
            oldtuple = NULL;
            olddata = NULL;
            neednextpage = true;
        }
        else
        {
            /*
             * Write a brand new page.
             *
             * First, fill any hole
             */
            off = (int) (obj_desc->offset % LOBLKSIZE);
            if (off > 0)
                MemSet(workb, 0, off);

            /*
             * Insert appropriate portion of new data
             */
            n = LOBLKSIZE - off;
            n = (n <= (nbytes - nwritten)) ? n : (nbytes - nwritten);
            memcpy(workb + off, buf + nwritten, n);
            nwritten += n;
            obj_desc->offset += n;
            /* compute valid length of new page */
            len = off + n;
            SET_VARSIZE(&workbuf.hdr, len + VARHDRSZ);

            /*
             * Form and insert updated tuple
             */
            memset(values, 0, sizeof(values));
            memset(nulls, false, sizeof(nulls));
            values[Anum_pg_largeobject_loid - 1] = ObjectIdGetDatum(obj_desc->id);
            values[Anum_pg_largeobject_pageno - 1] = Int32GetDatum(pageno);
            values[Anum_pg_largeobject_data - 1] = PointerGetDatum(&workbuf);
            newtup = heap_form_tuple(lo_heap_r->rd_att, values, nulls);
            CatalogTupleInsertWithInfo(lo_heap_r, newtup, indstate);
            heap_freetuple(newtup);
        }
        pageno++;
    }

    systable_endscan_ordered(sd);

    CatalogCloseIndexes(indstate);

    /*
     * Advance command counter so that my tuple updates will be seen by later
     * large-object operations in this transaction.
     */
    CommandCounterIncrement();

    return nwritten;
}