#include "postgres.h"
#include "access/htup.h"
#include "access/htup_details.h"
#include "storage/buffile.h"
#include "storage/lwlock.h"
#include "storage/sharedfileset.h"
#include "utils/sharedtuplestore.h"

Include dependency graph for sharedtuplestore.c:

Data Structures
struct	SharedTuplestoreChunk

struct	SharedTuplestoreParticipant

struct	SharedTuplestore

struct	SharedTuplestoreAccessor

Macros
#define	STS_CHUNK_PAGES 4

#define	STS_CHUNK_HEADER_SIZE offsetof(SharedTuplestoreChunk, data)

#define	STS_CHUNK_DATA_SIZE (STS_CHUNK_PAGES * BLCKSZ - STS_CHUNK_HEADER_SIZE)

Typedefs
typedef struct SharedTuplestoreChunk	SharedTuplestoreChunk

typedef struct SharedTuplestoreParticipant	SharedTuplestoreParticipant

Functions
static void	sts_filename (char name, SharedTuplestoreAccessor accessor, int participant)

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)

static void	sts_flush_chunk (SharedTuplestoreAccessor *accessor)

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)

static MinimalTuple	sts_read_tuple (SharedTuplestoreAccessor accessor, void meta_data)

MinimalTuple	sts_parallel_scan_next (SharedTuplestoreAccessor accessor, void meta_data)

Macro Definition Documentation

◆ STS_CHUNK_DATA_SIZE

#define STS_CHUNK_DATA_SIZE (STS_CHUNK_PAGES * BLCKSZ - STS_CHUNK_HEADER_SIZE)

Definition at line 39 of file sharedtuplestore.c.

◆ STS_CHUNK_HEADER_SIZE

#define STS_CHUNK_HEADER_SIZE offsetof(SharedTuplestoreChunk, data)

Definition at line 38 of file sharedtuplestore.c.

◆ STS_CHUNK_PAGES

#define STS_CHUNK_PAGES 4

Definition at line 37 of file sharedtuplestore.c.

Typedef Documentation

◆ SharedTuplestoreChunk

typedef struct SharedTuplestoreChunk SharedTuplestoreChunk

◆ SharedTuplestoreParticipant

typedef struct SharedTuplestoreParticipant SharedTuplestoreParticipant

Function Documentation

◆ sts_attach()

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

Definition at line 177 of file sharedtuplestore.c.

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

References Assert(), SharedTuplestoreAccessor::context, CurrentMemoryContext, SharedTuplestoreAccessor::fileset, palloc0(), SharedTuplestoreAccessor::participant, and SharedTuplestoreAccessor::sts.

Referenced by ExecParallelHashEnsureBatchAccessors(), and ExecParallelHashRepartitionRest().

◆ sts_begin_parallel_scan()

void sts_begin_parallel_scan ( SharedTuplestoreAccessor * accessor )

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(), i, SharedTuplestore::nparticipants, SharedTuplestoreAccessor::participant, SharedTuplestore::participants, PG_USED_FOR_ASSERTS_ONLY, SharedTuplestoreAccessor::read_file, SharedTuplestoreAccessor::read_next_page, SharedTuplestoreAccessor::read_participant, SharedTuplestoreAccessor::sts, sts_end_parallel_scan(), and SharedTuplestoreParticipant::writing.

Referenced by ExecParallelHashJoinNewBatch(), and ExecParallelHashRepartitionRest().

◆ sts_end_parallel_scan()

void sts_end_parallel_scan ( SharedTuplestoreAccessor * accessor )

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 SharedTuplestoreAccessor::read_file.

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

◆ sts_end_write()

void sts_end_write ( SharedTuplestoreAccessor * accessor )

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(), SharedTuplestoreAccessor::participant, SharedTuplestore::participants, pfree(), SharedTuplestoreAccessor::sts, sts_flush_chunk(), SharedTuplestoreAccessor::write_chunk, SharedTuplestoreAccessor::write_file, and SharedTuplestoreParticipant::writing.

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

◆ sts_estimate()

size_t sts_estimate ( int participants )

Definition at line 103 of file sharedtuplestore.c.

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

◆ sts_filename()

static void sts_filename	(	char *	name,
		SharedTuplestoreAccessor *	accessor,
		int	participant
	)

static

Definition at line 597 of file sharedtuplestore.c.

{
    snprintf(name, MAXPGPATH, "%s.p%d", accessor->sts->name, participant);
}

References MAXPGPATH, name, SharedTuplestore::name, snprintf, and SharedTuplestoreAccessor::sts.

Referenced by sts_parallel_scan_next(), and sts_puttuple().

◆ sts_flush_chunk()

static void sts_flush_chunk ( SharedTuplestoreAccessor * accessor )

static

Definition at line 195 of file sharedtuplestore.c.

{
    size_t      size;
 
    size = STS_CHUNK_PAGES * BLCKSZ;
    BufFileWrite(accessor->write_file, accessor->write_chunk, size);
    memset(accessor->write_chunk, 0, size);
    accessor->write_pointer = &accessor->write_chunk->data[0];
    accessor->sts->participants[accessor->participant].npages +=
        STS_CHUNK_PAGES;
}

References BufFileWrite(), SharedTuplestoreChunk::data, SharedTuplestoreParticipant::npages, SharedTuplestoreAccessor::participant, SharedTuplestore::participants, SharedTuplestoreAccessor::sts, STS_CHUNK_PAGES, SharedTuplestoreAccessor::write_chunk, SharedTuplestoreAccessor::write_file, and SharedTuplestoreAccessor::write_pointer.

Referenced by sts_end_write(), and sts_puttuple().

◆ 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
	)

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(sizeof(SharedTuplestoreAccessor));
    accessor->participant = my_participant_number;
    accessor->sts = sts;
    accessor->fileset = fileset;
    accessor->context = CurrentMemoryContext;
 
    return accessor;
}

References Assert(), SharedTuplestoreAccessor::context, CurrentMemoryContext, elog, ERROR, SharedTuplestoreAccessor::fileset, SharedTuplestore::flags, i, SharedTuplestoreParticipant::lock, LWLockInitialize(), SharedTuplestore::meta_data_size, name, SharedTuplestore::name, SharedTuplestoreParticipant::npages, SharedTuplestore::nparticipants, palloc0(), SharedTuplestoreAccessor::participant, SharedTuplestore::participants, SharedTuplestoreParticipant::read_page, SharedTuplestoreAccessor::sts, STS_CHUNK_DATA_SIZE, and SharedTuplestoreParticipant::writing.

Referenced by ExecParallelHashJoinSetUpBatches().

◆ sts_parallel_scan_next()

MinimalTuple sts_parallel_scan_next	(	SharedTuplestoreAccessor *	accessor,
		void *	meta_data
	)

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;
}

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

◆ sts_puttuple()

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

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;
}

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

◆ sts_read_tuple()

static MinimalTuple sts_read_tuple	(	SharedTuplestoreAccessor *	accessor,
		void *	meta_data
	)

static

Definition at line 414 of file sharedtuplestore.c.

{
    MinimalTuple tuple;
    uint32      size;
    size_t      remaining_size;
    size_t      this_chunk_size;
    char       *destination;
 
    /*
     * We'll keep track of bytes read from this chunk so that we can detect an
     * overflowing tuple and switch to reading overflow pages.
     */
    if (accessor->sts->meta_data_size > 0)
    {
        BufFileReadExact(accessor->read_file, meta_data, accessor->sts->meta_data_size);
        accessor->read_bytes += accessor->sts->meta_data_size;
    }
    BufFileReadExact(accessor->read_file, &size, sizeof(size));
    accessor->read_bytes += sizeof(size);
    if (size > accessor->read_buffer_size)
    {
        size_t      new_read_buffer_size;
 
        if (accessor->read_buffer != NULL)
            pfree(accessor->read_buffer);
        new_read_buffer_size = Max(size, accessor->read_buffer_size * 2);
        accessor->read_buffer =
            MemoryContextAlloc(accessor->context, new_read_buffer_size);
        accessor->read_buffer_size = new_read_buffer_size;
    }
    remaining_size = size - sizeof(uint32);
    this_chunk_size = Min(remaining_size,
                          BLCKSZ * STS_CHUNK_PAGES - accessor->read_bytes);
    destination = accessor->read_buffer + sizeof(uint32);
    BufFileReadExact(accessor->read_file, destination, this_chunk_size);
    accessor->read_bytes += this_chunk_size;
    remaining_size -= this_chunk_size;
    destination += this_chunk_size;
    ++accessor->read_ntuples;
 
    /* Check if we need to read any overflow chunks. */
    while (remaining_size > 0)
    {
        /* We are now positioned at the start of an overflow chunk. */
        SharedTuplestoreChunk chunk_header;
 
        BufFileReadExact(accessor->read_file, &chunk_header, STS_CHUNK_HEADER_SIZE);
        accessor->read_bytes = STS_CHUNK_HEADER_SIZE;
        if (chunk_header.overflow == 0)
            ereport(ERROR,
                    (errcode_for_file_access(),
                     errmsg("unexpected chunk in shared tuplestore temporary file"),
                     errdetail_internal("Expected overflow chunk.")));
        accessor->read_next_page += STS_CHUNK_PAGES;
        this_chunk_size = Min(remaining_size,
                              BLCKSZ * STS_CHUNK_PAGES -
                              STS_CHUNK_HEADER_SIZE);
        BufFileReadExact(accessor->read_file, destination, this_chunk_size);
        accessor->read_bytes += this_chunk_size;
        remaining_size -= this_chunk_size;
        destination += this_chunk_size;
 
        /*
         * These will be used to count regular tuples following the oversized
         * tuple that spilled into this overflow chunk.
         */
        accessor->read_ntuples = 0;
        accessor->read_ntuples_available = chunk_header.ntuples;
    }
 
    tuple = (MinimalTuple) accessor->read_buffer;
    tuple->t_len = size;
 
    return tuple;
}

Referenced by sts_parallel_scan_next().

◆ sts_reinitialize()

void sts_reinitialize ( SharedTuplestoreAccessor * accessor )

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 i, SharedTuplestore::nparticipants, SharedTuplestore::participants, SharedTuplestoreParticipant::read_page, and SharedTuplestoreAccessor::sts.

Data Structures

Macros

Typedefs

Functions

Macro Definition Documentation

◆ STS_CHUNK_DATA_SIZE

◆ STS_CHUNK_HEADER_SIZE

◆ STS_CHUNK_PAGES

Typedef Documentation

◆ SharedTuplestoreChunk

◆ SharedTuplestoreParticipant

Function Documentation

◆ sts_attach()

◆ sts_begin_parallel_scan()

◆ sts_end_parallel_scan()

◆ sts_end_write()

◆ sts_estimate()

◆ sts_filename()

◆ sts_flush_chunk()

◆ sts_initialize()

◆ sts_parallel_scan_next()

◆ sts_puttuple()

◆ sts_read_tuple()

◆ sts_reinitialize()