#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:

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 1 of file sharedtuplestore.h.

◆ SharedTuplestoreAccessor

typedef struct SharedTuplestoreAccessor SharedTuplestoreAccessor

Definition at line 1 of file sharedtuplestore.h.

Function Documentation

◆ sts_attach()

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

Definition at line 178 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 253 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 281 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 213 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 104 of file sharedtuplestore.c.

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

◆ 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 126 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(), LWTRANCHE_SHARED_TUPLESTORE, 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 495 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 300 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_reinitialize()

void sts_reinitialize ( SharedTuplestoreAccessor * accessor )

Definition at line 234 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.

Macros

Typedefs

Functions