sharedtuplestore.h File Reference

#include "access/htup.h"
#include "storage/fd.h"
#include "storage/sharedfileset.h"

Include dependency graph for sharedtuplestore.h:

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

Go to the source code of this file.

Macros
#define	SHARED_TUPLESTORE_SINGLE_PASS   0x01

Typedefs
typedef struct SharedTuplestore	SharedTuplestore
typedef struct SharedTuplestoreAccessor	SharedTuplestoreAccessor

Functions
size_t	sts_estimate (int participants)
SharedTuplestoreAccessor *	sts_initialize (SharedTuplestore *sts, int participants, int my_participant_number, size_t meta_data_size, int flags, SharedFileSet *fileset, const char *name)
SharedTuplestoreAccessor *	sts_attach (SharedTuplestore *sts, int my_participant_number, SharedFileSet *fileset)
void	sts_end_write (SharedTuplestoreAccessor *accessor)
void	sts_reinitialize (SharedTuplestoreAccessor *accessor)
void	sts_begin_parallel_scan (SharedTuplestoreAccessor *accessor)
void	sts_end_parallel_scan (SharedTuplestoreAccessor *accessor)
void	sts_puttuple (SharedTuplestoreAccessor *accessor, void *meta_data, MinimalTuple tuple)
MinimalTuple	sts_parallel_scan_next (SharedTuplestoreAccessor *accessor, void *meta_data)

Macro Definition Documentation

SHARED_TUPLESTORE_SINGLE_PASS

#define SHARED_TUPLESTORE_SINGLE_PASS   0x01

Definition at line 30 of file sharedtuplestore.h.

Typedef Documentation

SharedTuplestore

typedef struct SharedTuplestore SharedTuplestore

Definition at line 21 of file sharedtuplestore.h.

SharedTuplestoreAccessor

typedef struct SharedTuplestoreAccessor SharedTuplestoreAccessor

Definition at line 24 of file sharedtuplestore.h.

Function Documentation

sts_attach()

SharedTuplestoreAccessor * sts_attach ( SharedTuplestore *  sts, int  my_participant_number, SharedFileSet *  fileset  )

extern

Definition at line 177 of file sharedtuplestore.c.

{
    SharedTuplestoreAccessor *accessor;
 
    Assert(my_participant_number < sts->nparticipants);
 
    accessor = palloc0_object(SharedTuplestoreAccessor);
    accessor->participant = my_participant_number;
    accessor->sts = sts;
    accessor->fileset = fileset;
    accessor->context = CurrentMemoryContext;
 
    return accessor;
}

References Assert, CurrentMemoryContext, fb(), and palloc0_object.

Referenced by ExecParallelHashEnsureBatchAccessors(), and ExecParallelHashRepartitionRest().

sts_begin_parallel_scan()

void sts_begin_parallel_scan ( SharedTuplestoreAccessor *  accessor )

extern

Definition at line 252 of file sharedtuplestore.c.

{
    int         i PG_USED_FOR_ASSERTS_ONLY;
 
    /* End any existing scan that was in progress. */
    sts_end_parallel_scan(accessor);
 
    /*
     * Any backend that might have written into this shared tuplestore must
     * have called sts_end_write(), so that all buffers are flushed and the
     * files have stopped growing.
     */
    for (i = 0; i < accessor->sts->nparticipants; ++i)
        Assert(!accessor->sts->participants[i].writing);
 
    /*
     * We will start out reading the file that THIS backend wrote.  There may
     * be some caching locality advantage to that.
     */
    accessor->read_participant = accessor->participant;
    accessor->read_file = NULL;
    accessor->read_next_page = 0;
}

References Assert, fb(), i, PG_USED_FOR_ASSERTS_ONLY, and sts_end_parallel_scan().

Referenced by ExecParallelHashJoinNewBatch(), and ExecParallelHashRepartitionRest().

sts_end_parallel_scan()

void sts_end_parallel_scan ( SharedTuplestoreAccessor *  accessor )

extern

Definition at line 280 of file sharedtuplestore.c.

{
    /*
     * Here we could delete all files if SHARED_TUPLESTORE_SINGLE_PASS, but
     * we'd probably need a reference count of current parallel scanners so we
     * could safely do it only when the reference count reaches zero.
     */
    if (accessor->read_file != NULL)
    {
        BufFileClose(accessor->read_file);
        accessor->read_file = NULL;
    }
}

References BufFileClose(), and fb().

Referenced by ExecHashTableDetach(), ExecHashTableDetachBatch(), ExecParallelHashCloseBatchAccessors(), ExecParallelHashJoinNewBatch(), ExecParallelHashRepartitionRest(), ExecParallelPrepHashTableForUnmatched(), and sts_begin_parallel_scan().

sts_end_write()

void sts_end_write ( SharedTuplestoreAccessor *  accessor )

extern

Definition at line 212 of file sharedtuplestore.c.

{
    if (accessor->write_file != NULL)
    {
        sts_flush_chunk(accessor);
        BufFileClose(accessor->write_file);
        pfree(accessor->write_chunk);
        accessor->write_chunk = NULL;
        accessor->write_file = NULL;
        accessor->sts->participants[accessor->participant].writing = false;
    }
}

References BufFileClose(), fb(), pfree(), and sts_flush_chunk().

Referenced by ExecHashTableDetach(), ExecParallelHashCloseBatchAccessors(), ExecParallelHashJoinPartitionOuter(), and MultiExecParallelHash().

sts_estimate()

size_t sts_estimate ( int  participants )

extern

Definition at line 103 of file sharedtuplestore.c.

{
    return offsetof(SharedTuplestore, participants) +
        sizeof(SharedTuplestoreParticipant) * participants;
}

References fb().

sts_initialize()

SharedTuplestoreAccessor * sts_initialize ( SharedTuplestore *  sts, int  participants, int  my_participant_number, size_t  meta_data_size, int  flags, SharedFileSet *  fileset, const char *  name  )

extern

Definition at line 125 of file sharedtuplestore.c.

{
    SharedTuplestoreAccessor *accessor;
    int         i;
 
    Assert(my_participant_number < participants);
 
    sts->nparticipants = participants;
    sts->meta_data_size = meta_data_size;
    sts->flags = flags;
 
    if (strlen(name) > sizeof(sts->name) - 1)
        elog(ERROR, "SharedTuplestore name too long");
    strcpy(sts->name, name);
 
    /*
     * Limit meta-data so it + tuple size always fits into a single chunk.
     * sts_puttuple() and sts_read_tuple() could be made to support scenarios
     * where that's not the case, but it's not currently required. If so,
     * meta-data size probably should be made variable, too.
     */
    if (meta_data_size + sizeof(uint32) >= STS_CHUNK_DATA_SIZE)
        elog(ERROR, "meta-data too long");
 
    for (i = 0; i < participants; ++i)
    {
        LWLockInitialize(&sts->participants[i].lock,
                         LWTRANCHE_SHARED_TUPLESTORE);
        sts->participants[i].read_page = 0;
        sts->participants[i].npages = 0;
        sts->participants[i].writing = false;
    }
 
    accessor = palloc0_object(SharedTuplestoreAccessor);
    accessor->participant = my_participant_number;
    accessor->sts = sts;
    accessor->fileset = fileset;
    accessor->context = CurrentMemoryContext;
 
    return accessor;
}

References Assert, CurrentMemoryContext, elog, ERROR, fb(), SharedTuplestore::flags, i, SharedTuplestoreParticipant::lock, LWLockInitialize(), SharedTuplestore::meta_data_size, name, SharedTuplestore::name, SharedTuplestoreParticipant::npages, SharedTuplestore::nparticipants, palloc0_object, SharedTuplestore::participants, SharedTuplestoreParticipant::read_page, STS_CHUNK_DATA_SIZE, and SharedTuplestoreParticipant::writing.

Referenced by ExecParallelHashJoinSetUpBatches().

sts_parallel_scan_next()

MinimalTuple sts_parallel_scan_next ( SharedTuplestoreAccessor *  accessor, void *  meta_data  )

extern

Definition at line 494 of file sharedtuplestore.c.

{
    SharedTuplestoreParticipant *p;
    BlockNumber read_page;
    bool        eof;
 
    for (;;)
    {
        /* Can we read more tuples from the current chunk? */
        if (accessor->read_ntuples < accessor->read_ntuples_available)
            return sts_read_tuple(accessor, meta_data);
 
        /* Find the location of a new chunk to read. */
        p = &accessor->sts->participants[accessor->read_participant];
 
        LWLockAcquire(&p->lock, LW_EXCLUSIVE);
        /* We can skip directly past overflow pages we know about. */
        if (p->read_page < accessor->read_next_page)
            p->read_page = accessor->read_next_page;
        eof = p->read_page >= p->npages;
        if (!eof)
        {
            /* Claim the next chunk. */
            read_page = p->read_page;
            /* Advance the read head for the next reader. */
            p->read_page += STS_CHUNK_PAGES;
            accessor->read_next_page = p->read_page;
        }
        LWLockRelease(&p->lock);
 
        if (!eof)
        {
            SharedTuplestoreChunk chunk_header;
 
            /* Make sure we have the file open. */
            if (accessor->read_file == NULL)
            {
                char        name[MAXPGPATH];
                MemoryContext oldcxt;
 
                sts_filename(name, accessor, accessor->read_participant);
 
                oldcxt = MemoryContextSwitchTo(accessor->context);
                accessor->read_file =
                    BufFileOpenFileSet(&accessor->fileset->fs, name, O_RDONLY,
                                       false);
                MemoryContextSwitchTo(oldcxt);
            }
 
            /* Seek and load the chunk header. */
            if (BufFileSeekBlock(accessor->read_file, read_page) != 0)
                ereport(ERROR,
                        (errcode_for_file_access(),
                         errmsg("could not seek to block %u in shared tuplestore temporary file",
                                read_page)));
            BufFileReadExact(accessor->read_file, &chunk_header, STS_CHUNK_HEADER_SIZE);
 
            /*
             * If this is an overflow chunk, we skip it and any following
             * overflow chunks all at once.
             */
            if (chunk_header.overflow > 0)
            {
                accessor->read_next_page = read_page +
                    chunk_header.overflow * STS_CHUNK_PAGES;
                continue;
            }
 
            accessor->read_ntuples = 0;
            accessor->read_ntuples_available = chunk_header.ntuples;
            accessor->read_bytes = STS_CHUNK_HEADER_SIZE;
 
            /* Go around again, so we can get a tuple from this chunk. */
        }
        else
        {
            if (accessor->read_file != NULL)
            {
                BufFileClose(accessor->read_file);
                accessor->read_file = NULL;
            }
 
            /*
             * Try the next participant's file.  If we've gone full circle,
             * we're done.
             */
            accessor->read_participant = (accessor->read_participant + 1) %
                accessor->sts->nparticipants;
            if (accessor->read_participant == accessor->participant)
                break;
            accessor->read_next_page = 0;
 
            /* Go around again, so we can get a chunk from this file. */
        }
    }
 
    return NULL;
}

References BufFileClose(), BufFileOpenFileSet(), BufFileReadExact(), BufFileSeekBlock(), ereport, errcode_for_file_access(), errmsg(), ERROR, fb(), SharedTuplestoreParticipant::lock, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MAXPGPATH, MemoryContextSwitchTo(), name, SharedTuplestoreParticipant::npages, SharedTuplestoreParticipant::read_page, STS_CHUNK_HEADER_SIZE, STS_CHUNK_PAGES, sts_filename(), and sts_read_tuple().

Referenced by ExecParallelHashJoinNewBatch(), ExecParallelHashJoinOuterGetTuple(), and ExecParallelHashRepartitionRest().

sts_puttuple()

void sts_puttuple ( SharedTuplestoreAccessor *  accessor, void *  meta_data, MinimalTuple  tuple  )

extern

Definition at line 299 of file sharedtuplestore.c.

{
    size_t      size;
 
    /* Do we have our own file yet? */
    if (accessor->write_file == NULL)
    {
        SharedTuplestoreParticipant *participant;
        char        name[MAXPGPATH];
        MemoryContext oldcxt;
 
        /* Create one.  Only this backend will write into it. */
        sts_filename(name, accessor, accessor->participant);
 
        oldcxt = MemoryContextSwitchTo(accessor->context);
        accessor->write_file =
            BufFileCreateFileSet(&accessor->fileset->fs, name);
        MemoryContextSwitchTo(oldcxt);
 
        /* Set up the shared state for this backend's file. */
        participant = &accessor->sts->participants[accessor->participant];
        participant->writing = true;    /* for assertions only */
    }
 
    /* Do we have space? */
    size = accessor->sts->meta_data_size + tuple->t_len;
    if (accessor->write_pointer + size > accessor->write_end)
    {
        if (accessor->write_chunk == NULL)
        {
            /* First time through.  Allocate chunk. */
            accessor->write_chunk = (SharedTuplestoreChunk *)
                MemoryContextAllocZero(accessor->context,
                                       STS_CHUNK_PAGES * BLCKSZ);
            accessor->write_chunk->ntuples = 0;
            accessor->write_pointer = &accessor->write_chunk->data[0];
            accessor->write_end = (char *)
                accessor->write_chunk + STS_CHUNK_PAGES * BLCKSZ;
        }
        else
        {
            /* See if flushing helps. */
            sts_flush_chunk(accessor);
        }
 
        /* It may still not be enough in the case of a gigantic tuple. */
        if (accessor->write_pointer + size > accessor->write_end)
        {
            size_t      written;
 
            /*
             * We'll write the beginning of the oversized tuple, and then
             * write the rest in some number of 'overflow' chunks.
             *
             * sts_initialize() verifies that the size of the tuple +
             * meta-data always fits into a chunk. Because the chunk has been
             * flushed above, we can be sure to have all of a chunk's usable
             * space available.
             */
            Assert(accessor->write_pointer + accessor->sts->meta_data_size +
                   sizeof(uint32) < accessor->write_end);
 
            /* Write the meta-data as one chunk. */
            if (accessor->sts->meta_data_size > 0)
                memcpy(accessor->write_pointer, meta_data,
                       accessor->sts->meta_data_size);
 
            /*
             * Write as much of the tuple as we can fit. This includes the
             * tuple's size at the start.
             */
            written = accessor->write_end - accessor->write_pointer -
                accessor->sts->meta_data_size;
            memcpy(accessor->write_pointer + accessor->sts->meta_data_size,
                   tuple, written);
            ++accessor->write_chunk->ntuples;
            size -= accessor->sts->meta_data_size;
            size -= written;
            /* Now write as many overflow chunks as we need for the rest. */
            while (size > 0)
            {
                size_t      written_this_chunk;
 
                sts_flush_chunk(accessor);
 
                /*
                 * How many overflow chunks to go?  This will allow readers to
                 * skip all of them at once instead of reading each one.
                 */
                accessor->write_chunk->overflow = (size + STS_CHUNK_DATA_SIZE - 1) /
                    STS_CHUNK_DATA_SIZE;
                written_this_chunk =
                    Min(accessor->write_end - accessor->write_pointer, size);
                memcpy(accessor->write_pointer, (char *) tuple + written,
                       written_this_chunk);
                accessor->write_pointer += written_this_chunk;
                size -= written_this_chunk;
                written += written_this_chunk;
            }
            return;
        }
    }
 
    /* Copy meta-data and tuple into buffer. */
    if (accessor->sts->meta_data_size > 0)
        memcpy(accessor->write_pointer, meta_data,
               accessor->sts->meta_data_size);
    memcpy(accessor->write_pointer + accessor->sts->meta_data_size, tuple,
           tuple->t_len);
    accessor->write_pointer += size;
    ++accessor->write_chunk->ntuples;
}

References Assert, BufFileCreateFileSet(), fb(), MAXPGPATH, MemoryContextAllocZero(), MemoryContextSwitchTo(), Min, name, STS_CHUNK_DATA_SIZE, STS_CHUNK_PAGES, sts_filename(), sts_flush_chunk(), MinimalTupleData::t_len, and SharedTuplestoreParticipant::writing.

Referenced by ExecParallelHashJoinPartitionOuter(), ExecParallelHashRepartitionFirst(), ExecParallelHashRepartitionRest(), and ExecParallelHashTableInsert().

sts_reinitialize()

void sts_reinitialize ( SharedTuplestoreAccessor *  accessor )

extern

Definition at line 233 of file sharedtuplestore.c.

{
    int         i;
 
    /*
     * Reset the shared read head for all participants' files.  Also set the
     * initial chunk size to the minimum (any increases from that size will be
     * recorded in chunk_expansion_log).
     */
    for (i = 0; i < accessor->sts->nparticipants; ++i)
    {
        accessor->sts->participants[i].read_page = 0;
    }
}

References fb(), and i.