tuplestore.c 26.7 KB
Newer Older
1 2 3 4 5 6 7 8
/*-------------------------------------------------------------------------
 *
 * tuplestore.c
 *	  Generalized routines for temporary tuple storage.
 *
 * This module handles temporary storage of tuples for purposes such
 * as Materialize nodes, hashjoin batch files, etc.  It is essentially
 * a dumbed-down version of tuplesort.c; it does no sorting of tuples
9 10
 * but can only store and regurgitate a sequence of tuples.  However,
 * because no sort is required, it is allowed to start reading the sequence
B
Bruce Momjian 已提交
11
 * before it has all been written.	This is particularly useful for cursors,
12 13
 * because it allows random access within the already-scanned portion of
 * a query without having to process the underlying scan to completion.
14 15 16 17
 * A temporary file is used to handle the data if it exceeds the
 * space limit specified by the caller.
 *
 * The (approximate) amount of memory allowed to the tuplestore is specified
B
Bruce Momjian 已提交
18
 * in kilobytes by the caller.	We absorb tuples and simply store them in an
19
 * in-memory array as long as we haven't exceeded maxKBytes.  If we do exceed
20
 * maxKBytes, we dump all the tuples into a temp file and then read from that
21
 * when needed.
22
 *
23
 * When the caller requests backward-scan capability, we write the temp file
24
 * in a format that allows either forward or backward scan.  Otherwise, only
25 26 27 28
 * forward scan is allowed.  Rewind and markpos/restorepos are normally allowed
 * but can be turned off via tuplestore_set_eflags; turning off both backward
 * scan and rewind enables truncation of the tuplestore at the mark point
 * (if any) for minimal memory usage.
29 30 31
 *
 * Because we allow reading before writing is complete, there are two
 * interesting positions in the temp file: the current read position and
B
Bruce Momjian 已提交
32
 * the current write position.	At any given instant, the temp file's seek
33 34
 * position corresponds to one of these, and the other one is remembered in
 * the Tuplestore's state.
35 36
 *
 *
37
 * Portions Copyright (c) 1996-2007, PostgreSQL Global Development Group
38 39 40
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
41
 *	  $PostgreSQL: pgsql/src/backend/utils/sort/tuplestore.c,v 1.31 2007/05/21 17:57:34 tgl Exp $
42 43 44 45 46 47 48
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/heapam.h"
49
#include "executor/executor.h"
50
#include "storage/buffile.h"
51
#include "utils/memutils.h"
52 53
#include "utils/tuplestore.h"

54

55
/*
B
Bruce Momjian 已提交
56
 * Possible states of a Tuplestore object.	These denote the states that
57 58 59 60
 * persist between calls of Tuplestore routines.
 */
typedef enum
{
61 62 63
	TSS_INMEM,					/* Tuples still fit in memory */
	TSS_WRITEFILE,				/* Writing to temp file */
	TSS_READFILE				/* Reading from temp file */
64 65 66 67 68 69 70 71
} TupStoreStatus;

/*
 * Private state of a Tuplestore operation.
 */
struct Tuplestorestate
{
	TupStoreStatus status;		/* enumerated value as shown above */
72
	int			eflags;			/* capability flags */
73
	bool		interXact;		/* keep open through transactions? */
74 75 76 77
	long		availMem;		/* remaining memory available, in bytes */
	BufFile    *myfile;			/* underlying file, or NULL if none */

	/*
B
Bruce Momjian 已提交
78 79 80
	 * These function pointers decouple the routines that must know what kind
	 * of tuple we are handling from the routines that don't need to know it.
	 * They are set up by the tuplestore_begin_xxx routines.
81
	 *
B
Bruce Momjian 已提交
82 83 84
	 * (Although tuplestore.c currently only supports heap tuples, I've copied
	 * this part of tuplesort.c so that extension to other kinds of objects
	 * will be easy if it's ever needed.)
85
	 *
B
Bruce Momjian 已提交
86
	 * Function to copy a supplied input tuple into palloc'd space. (NB: we
B
Bruce Momjian 已提交
87 88 89
	 * assume that a single pfree() is enough to release the tuple later, so
	 * the representation must be "flat" in one palloc chunk.) state->availMem
	 * must be decreased by the amount of space used.
90 91 92 93
	 */
	void	   *(*copytup) (Tuplestorestate *state, void *tup);

	/*
B
Bruce Momjian 已提交
94 95 96 97 98
	 * Function to write a stored tuple onto tape.	The representation of the
	 * tuple on tape need not be the same as it is in memory; requirements on
	 * the tape representation are given below.  After writing the tuple,
	 * pfree() it, and increase state->availMem by the amount of memory space
	 * thereby released.
99 100 101 102
	 */
	void		(*writetup) (Tuplestorestate *state, void *tup);

	/*
B
Bruce Momjian 已提交
103 104 105 106
	 * Function to read a stored tuple from tape back into memory. 'len' is
	 * the already-read length of the stored tuple.  Create and return a
	 * palloc'd copy, and decrease state->availMem by the amount of memory
	 * space consumed.
107 108 109 110
	 */
	void	   *(*readtup) (Tuplestorestate *state, unsigned int len);

	/*
B
Bruce Momjian 已提交
111 112
	 * This array holds pointers to tuples in memory if we are in state INMEM.
	 * In states WRITEFILE and READFILE it's not used.
113 114 115 116 117 118
	 */
	void	  **memtuples;		/* array of pointers to palloc'd tuples */
	int			memtupcount;	/* number of tuples currently present */
	int			memtupsize;		/* allocated length of memtuples array */

	/*
119 120
	 * These variables are used to keep track of the current position.
	 *
121 122 123 124 125 126
	 * In state WRITEFILE, the current file seek position is the write point,
	 * and the read position is remembered in readpos_xxx; in state READFILE,
	 * the current file seek position is the read point, and the write
	 * position is remembered in writepos_xxx.	(The write position is the
	 * same as EOF, but since BufFileSeek doesn't currently implement
	 * SEEK_END, we have to remember it explicitly.)
127
	 *
128 129 130 131
	 * Special case: if we are in WRITEFILE state and eof_reached is true,
	 * then the read position is implicitly equal to the write position (and
	 * hence to the file seek position); this way we need not update the
	 * readpos_xxx variables on each write.
132
	 */
133 134 135 136 137
	bool		eof_reached;	/* read reached EOF (always valid) */
	int			current;		/* next array index (valid if INMEM) */
	int			readpos_file;	/* file# (valid if WRITEFILE and not eof) */
	long		readpos_offset; /* offset (valid if WRITEFILE and not eof) */
	int			writepos_file;	/* file# (valid if READFILE) */
B
Bruce Momjian 已提交
138
	long		writepos_offset;	/* offset (valid if READFILE) */
139 140

	/* markpos_xxx holds marked position for mark and restore */
B
Bruce Momjian 已提交
141
	int			markpos_current;	/* saved "current" */
142 143
	int			markpos_file;	/* saved "readpos_file" */
	long		markpos_offset; /* saved "readpos_offset" */
144 145 146
};

#define COPYTUP(state,tup)	((*(state)->copytup) (state, tup))
B
Bruce Momjian 已提交
147
#define WRITETUP(state,tup) ((*(state)->writetup) (state, tup))
148 149 150 151 152 153 154 155 156 157
#define READTUP(state,len)	((*(state)->readtup) (state, len))
#define LACKMEM(state)		((state)->availMem < 0)
#define USEMEM(state,amt)	((state)->availMem -= (amt))
#define FREEMEM(state,amt)	((state)->availMem += (amt))

/*--------------------
 *
 * NOTES about on-tape representation of tuples:
 *
 * We require the first "unsigned int" of a stored tuple to be the total size
158 159
 * on-tape of the tuple, including itself (so it is never zero).
 * The remainder of the stored tuple
160 161 162
 * may or may not match the in-memory representation of the tuple ---
 * any conversion needed is the job of the writetup and readtup routines.
 *
163 164
 * If state->eflags & EXEC_FLAG_BACKWARD, then the stored representation of
 * the tuple must be followed by another "unsigned int" that is a copy of the
165 166
 * length --- so the total tape space used is actually sizeof(unsigned int)
 * more than the stored length value.  This allows read-backwards.	When
167
 * EXEC_FLAG_BACKWARD is not set, the write/read routines may omit the extra
168 169 170 171 172 173 174 175
 * length word.
 *
 * writetup is expected to write both length words as well as the tuple
 * data.  When readtup is called, the tape is positioned just after the
 * front length word; readtup must read the tuple data and advance past
 * the back length word (if present).
 *
 * The write/read routines can make use of the tuple description data
B
Bruce Momjian 已提交
176
 * stored in the Tuplestorestate record, if needed. They are also expected
177 178
 * to adjust state->availMem by the amount of memory space (not tape space!)
 * released or consumed.  There is no error return from either writetup
179
 * or readtup; they should ereport() on failure.
180 181 182 183
 *
 *
 * NOTES about memory consumption calculations:
 *
184 185 186
 * We count space allocated for tuples against the maxKBytes limit,
 * plus the space used by the variable-size array memtuples.
 * Fixed-size space (primarily the BufFile I/O buffer) is not counted.
187
 *
188 189 190 191 192
 * Note that we count actual space used (as shown by GetMemoryChunkSpace)
 * rather than the originally-requested size.  This is important since
 * palloc can add substantial overhead.  It's not a complete answer since
 * we won't count any wasted space in palloc allocation blocks, but it's
 * a lot better than what we were doing before 7.3.
193 194 195 196 197
 *
 *--------------------
 */


198
static Tuplestorestate *tuplestore_begin_common(int eflags,
B
Bruce Momjian 已提交
199 200
						bool interXact,
						int maxKBytes);
201
static void tuplestore_puttuple_common(Tuplestorestate *state, void *tuple);
202
static void dumptuples(Tuplestorestate *state);
203
static void tuplestore_trim(Tuplestorestate *state, int ntuples);
204 205 206 207 208 209 210 211 212 213 214 215
static unsigned int getlen(Tuplestorestate *state, bool eofOK);
static void *copytup_heap(Tuplestorestate *state, void *tup);
static void writetup_heap(Tuplestorestate *state, void *tup);
static void *readtup_heap(Tuplestorestate *state, unsigned int len);


/*
 *		tuplestore_begin_xxx
 *
 * Initialize for a tuple store operation.
 */
static Tuplestorestate *
216
tuplestore_begin_common(int eflags, bool interXact, int maxKBytes)
217 218 219
{
	Tuplestorestate *state;

220
	state = (Tuplestorestate *) palloc0(sizeof(Tuplestorestate));
221

222
	state->status = TSS_INMEM;
223
	state->eflags = eflags;
224
	state->interXact = interXact;
225 226 227 228
	state->availMem = maxKBytes * 1024L;
	state->myfile = NULL;

	state->memtupcount = 0;
B
Bruce Momjian 已提交
229
	state->memtupsize = 1024;	/* initial guess */
230 231
	state->memtuples = (void **) palloc(state->memtupsize * sizeof(void *));

232 233
	USEMEM(state, GetMemoryChunkSpace(state->memtuples));

234 235 236
	state->eof_reached = false;
	state->current = 0;

237 238 239
	return state;
}

240 241 242 243 244 245 246 247 248 249
/*
 * tuplestore_begin_heap
 *
 * Create a new tuplestore; other types of tuple stores (other than
 * "heap" tuple stores, for heap tuples) are possible, but not presently
 * implemented.
 *
 * randomAccess: if true, both forward and backward accesses to the
 * tuple store are allowed.
 *
250
 * interXact: if true, the files used for on-disk storage persist beyond the
B
Bruce Momjian 已提交
251
 * end of the current transaction.	NOTE: It's the caller's responsibility to
252 253 254
 * create such a tuplestore in a memory context that will also survive
 * transaction boundaries, and to ensure the tuplestore is closed when it's
 * no longer wanted.
255 256
 *
 * maxKBytes: how much data to store in memory (any data beyond this
257
 * amount is paged to disk).  When in doubt, use work_mem.
258
 */
259
Tuplestorestate *
260
tuplestore_begin_heap(bool randomAccess, bool interXact, int maxKBytes)
261
{
262 263 264 265 266 267 268 269 270 271 272 273
	Tuplestorestate *state;
	int		eflags;

	/*
	 * This interpretation of the meaning of randomAccess is compatible
	 * with the pre-8.3 behavior of tuplestores.
	 */
	eflags = randomAccess ?
		(EXEC_FLAG_BACKWARD | EXEC_FLAG_REWIND | EXEC_FLAG_MARK) :
		(EXEC_FLAG_REWIND | EXEC_FLAG_MARK);

	state = tuplestore_begin_common(eflags, interXact, maxKBytes);
274 275 276 277 278 279 280 281

	state->copytup = copytup_heap;
	state->writetup = writetup_heap;
	state->readtup = readtup_heap;

	return state;
}

282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305
/*
 * tuplestore_set_eflags
 *
 * Set capability flags at a finer grain than is allowed by
 * tuplestore_begin_xxx.  This must be called before inserting any data
 * into the tuplestore.
 *
 * eflags is a bitmask following the meanings used for executor node
 * startup flags (see executor.h).  tuplestore pays attention to these bits:
 *		EXEC_FLAG_REWIND		need rewind to start
 *		EXEC_FLAG_BACKWARD		need backward fetch
 *		EXEC_FLAG_MARK			need mark/restore
 * If tuplestore_set_eflags is not called, REWIND and MARK are allowed,
 * and BACKWARD is set per "randomAccess" in the tuplestore_begin_xxx call.
 */
void
tuplestore_set_eflags(Tuplestorestate *state, int eflags)
{
	Assert(state->status == TSS_INMEM);
	Assert(state->memtupcount == 0);

	state->eflags = eflags;
}

306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326
/*
 * tuplestore_end
 *
 *	Release resources and clean up.
 */
void
tuplestore_end(Tuplestorestate *state)
{
	int			i;

	if (state->myfile)
		BufFileClose(state->myfile);
	if (state->memtuples)
	{
		for (i = 0; i < state->memtupcount; i++)
			pfree(state->memtuples[i]);
		pfree(state->memtuples);
	}
}

/*
327 328 329 330 331 332 333 334 335 336 337 338
 * tuplestore_ateof
 *
 * Returns the current eof_reached state.
 */
bool
tuplestore_ateof(Tuplestorestate *state)
{
	return state->eof_reached;
}

/*
 * Accept one tuple and append it to the tuplestore.
339 340
 *
 * Note that the input tuple is always copied; the caller need not save it.
341 342 343 344
 *
 * If the read status is currently "AT EOF" then it remains so (the read
 * pointer advances along with the write pointer); otherwise the read
 * pointer is unchanged.  This is for the convenience of nodeMaterial.c.
345 346 347
 *
 * tuplestore_puttupleslot() is a convenience routine to collect data from
 * a TupleTableSlot without an extra copy operation.
348 349
 */
void
350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371
tuplestore_puttupleslot(Tuplestorestate *state,
						TupleTableSlot *slot)
{
	MinimalTuple tuple;

	/*
	 * Form a MinimalTuple in working memory
	 */
	tuple = ExecCopySlotMinimalTuple(slot);
	USEMEM(state, GetMemoryChunkSpace(tuple));

	tuplestore_puttuple_common(state, (void *) tuple);
}

/*
 * "Standard" case to copy from a HeapTuple.  This is actually now somewhat
 * deprecated, but not worth getting rid of in view of the number of callers.
 * (Consider adding something that takes a tupdesc+values/nulls arrays so
 * that we can use heap_form_minimal_tuple() and avoid a copy step.)
 */
void
tuplestore_puttuple(Tuplestorestate *state, HeapTuple tuple)
372 373
{
	/*
B
Bruce Momjian 已提交
374
	 * Copy the tuple.	(Must do this even in WRITEFILE case.)
375 376 377
	 */
	tuple = COPYTUP(state, tuple);

378 379 380 381 382 383
	tuplestore_puttuple_common(state, (void *) tuple);
}

static void
tuplestore_puttuple_common(Tuplestorestate *state, void *tuple)
{
384 385
	switch (state->status)
	{
386
		case TSS_INMEM:
B
Bruce Momjian 已提交
387

388 389 390 391 392 393 394
			/*
			 * Grow the array as needed.  Note that we try to grow the array
			 * when there is still one free slot remaining --- if we fail,
			 * there'll still be room to store the incoming tuple, and then
			 * we'll switch to tape-based operation.
			 */
			if (state->memtupcount >= state->memtupsize - 1)
395
			{
396 397 398 399 400 401 402 403 404 405 406 407 408 409
				/*
				 * See grow_memtuples() in tuplesort.c for the rationale
				 * behind these two tests.
				 */
				if (state->availMem > (long) (state->memtupsize * sizeof(void *)) &&
					(Size) (state->memtupsize * 2) < MaxAllocSize / sizeof(void *))
				{
					FREEMEM(state, GetMemoryChunkSpace(state->memtuples));
					state->memtupsize *= 2;
					state->memtuples = (void **)
						repalloc(state->memtuples,
								 state->memtupsize * sizeof(void *));
					USEMEM(state, GetMemoryChunkSpace(state->memtuples));
				}
410
			}
411 412

			/* Stash the tuple in the in-memory array */
413 414
			state->memtuples[state->memtupcount++] = tuple;

415 416 417 418
			/* If eof_reached, keep read position in sync */
			if (state->eof_reached)
				state->current = state->memtupcount;

419
			/*
420
			 * Done if we still fit in available memory and have array slots.
421
			 */
422
			if (state->memtupcount < state->memtupsize && !LACKMEM(state))
423 424 425 426 427
				return;

			/*
			 * Nope; time to switch to tape-based operation.
			 */
428
			state->myfile = BufFileCreateTemp(state->interXact);
429 430 431 432 433 434
			state->status = TSS_WRITEFILE;
			dumptuples(state);
			break;
		case TSS_WRITEFILE:
			WRITETUP(state, tuple);
			break;
435
		case TSS_READFILE:
B
Bruce Momjian 已提交
436

437
			/*
438
			 * Switch from reading to writing.
439
			 */
440 441 442 443 444 445
			if (!state->eof_reached)
				BufFileTell(state->myfile,
							&state->readpos_file, &state->readpos_offset);
			if (BufFileSeek(state->myfile,
							state->writepos_file, state->writepos_offset,
							SEEK_SET) != 0)
446
				elog(ERROR, "seek to EOF failed");
447 448
			state->status = TSS_WRITEFILE;
			WRITETUP(state, tuple);
449 450
			break;
		default:
451
			elog(ERROR, "invalid tuplestore state");
452 453 454 455 456 457 458 459
			break;
	}
}

/*
 * Fetch the next tuple in either forward or back direction.
 * Returns NULL if no more tuples.	If should_free is set, the
 * caller must pfree the returned tuple when done with it.
460 461 462
 *
 * Backward scan is only allowed if randomAccess was set true or
 * EXEC_FLAG_BACKWARD was specified to tuplestore_set_eflags().
463
 */
464
static void *
465
tuplestore_gettuple(Tuplestorestate *state, bool forward,
B
Bruce Momjian 已提交
466
					bool *should_free)
467 468 469 470
{
	unsigned int tuplen;
	void	   *tup;

471
	Assert(forward || (state->eflags & EXEC_FLAG_BACKWARD));
472

473 474
	switch (state->status)
	{
475
		case TSS_INMEM:
476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504
			*should_free = false;
			if (forward)
			{
				if (state->current < state->memtupcount)
					return state->memtuples[state->current++];
				state->eof_reached = true;
				return NULL;
			}
			else
			{
				if (state->current <= 0)
					return NULL;

				/*
				 * if all tuples are fetched already then we return last
				 * tuple, else - tuple before last returned.
				 */
				if (state->eof_reached)
					state->eof_reached = false;
				else
				{
					state->current--;	/* last returned tuple */
					if (state->current <= 0)
						return NULL;
				}
				return state->memtuples[state->current - 1];
			}
			break;

505 506 507 508
		case TSS_WRITEFILE:
			/* Skip state change if we'll just return NULL */
			if (state->eof_reached && forward)
				return NULL;
B
Bruce Momjian 已提交
509

510 511 512 513 514 515 516
			/*
			 * Switch from writing to reading.
			 */
			BufFileTell(state->myfile,
						&state->writepos_file, &state->writepos_offset);
			if (!state->eof_reached)
				if (BufFileSeek(state->myfile,
B
Bruce Momjian 已提交
517
								state->readpos_file, state->readpos_offset,
518
								SEEK_SET) != 0)
519
					elog(ERROR, "seek failed");
520 521 522
			state->status = TSS_READFILE;
			/* FALL THRU into READFILE case */

523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541
		case TSS_READFILE:
			*should_free = true;
			if (forward)
			{
				if ((tuplen = getlen(state, true)) != 0)
				{
					tup = READTUP(state, tuplen);
					return tup;
				}
				else
				{
					state->eof_reached = true;
					return NULL;
				}
			}

			/*
			 * Backward.
			 *
542 543
			 * if all tuples are fetched already then we return last tuple,
			 * else - tuple before last returned.
544
			 *
545 546
			 * Back up to fetch previously-returned tuple's ending length
			 * word. If seek fails, assume we are at start of file.
547
			 */
548 549 550 551 552
			if (BufFileSeek(state->myfile, 0, -(long) sizeof(unsigned int),
							SEEK_CUR) != 0)
				return NULL;
			tuplen = getlen(state, false);

553 554 555
			if (state->eof_reached)
			{
				state->eof_reached = false;
556
				/* We will return the tuple returned before returning NULL */
557 558 559 560 561 562 563
			}
			else
			{
				/*
				 * Back up to get ending length word of tuple before it.
				 */
				if (BufFileSeek(state->myfile, 0,
B
Bruce Momjian 已提交
564
								-(long) (tuplen + 2 * sizeof(unsigned int)),
565 566 567
								SEEK_CUR) != 0)
				{
					/*
B
Bruce Momjian 已提交
568 569 570 571
					 * If that fails, presumably the prev tuple is the first
					 * in the file.  Back up so that it becomes next to read
					 * in forward direction (not obviously right, but that is
					 * what in-memory case does).
572 573
					 */
					if (BufFileSeek(state->myfile, 0,
B
Bruce Momjian 已提交
574
									-(long) (tuplen + sizeof(unsigned int)),
575
									SEEK_CUR) != 0)
576
						elog(ERROR, "bogus tuple length in backward scan");
577 578
					return NULL;
				}
579
				tuplen = getlen(state, false);
580 581 582
			}

			/*
B
Bruce Momjian 已提交
583 584 585
			 * Now we have the length of the prior tuple, back up and read it.
			 * Note: READTUP expects we are positioned after the initial
			 * length word of the tuple, so back up to that point.
586 587
			 */
			if (BufFileSeek(state->myfile, 0,
B
Bruce Momjian 已提交
588
							-(long) tuplen,
589
							SEEK_CUR) != 0)
590
				elog(ERROR, "bogus tuple length in backward scan");
591 592 593 594
			tup = READTUP(state, tuplen);
			return tup;

		default:
595
			elog(ERROR, "invalid tuplestore state");
596 597 598 599
			return NULL;		/* keep compiler quiet */
	}
}

600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652
/*
 * tuplestore_gettupleslot - exported function to fetch a MinimalTuple
 *
 * If successful, put tuple in slot and return TRUE; else, clear the slot
 * and return FALSE.
 */
bool
tuplestore_gettupleslot(Tuplestorestate *state, bool forward,
						TupleTableSlot *slot)
{
	MinimalTuple tuple;
	bool		should_free;

	tuple = (MinimalTuple) tuplestore_gettuple(state, forward, &should_free);

	if (tuple)
	{
		ExecStoreMinimalTuple(tuple, slot, should_free);
		return true;
	}
	else
	{
		ExecClearTuple(slot);
		return false;
	}
}

/*
 * tuplestore_advance - exported function to adjust position without fetching
 *
 * We could optimize this case to avoid palloc/pfree overhead, but for the
 * moment it doesn't seem worthwhile.
 */
bool
tuplestore_advance(Tuplestorestate *state, bool forward)
{
	void	   *tuple;
	bool		should_free;

	tuple = tuplestore_gettuple(state, forward, &should_free);

	if (tuple)
	{
		if (should_free)
			pfree(tuple);
		return true;
	}
	else
	{
		return false;
	}
}

653 654
/*
 * dumptuples - remove tuples from memory and write to tape
655 656 657 658
 *
 * As a side effect, we must set readpos and markpos to the value
 * corresponding to "current"; otherwise, a dump would lose the current read
 * position.
659 660 661 662 663 664
 */
static void
dumptuples(Tuplestorestate *state)
{
	int			i;

B
Bruce Momjian 已提交
665
	for (i = 0;; i++)
666 667 668 669 670 671 672 673 674
	{
		if (i == state->current)
			BufFileTell(state->myfile,
						&state->readpos_file, &state->readpos_offset);
		if (i == state->markpos_current)
			BufFileTell(state->myfile,
						&state->markpos_file, &state->markpos_offset);
		if (i >= state->memtupcount)
			break;
675
		WRITETUP(state, state->memtuples[i]);
676
	}
677 678 679 680 681 682 683 684 685
	state->memtupcount = 0;
}

/*
 * tuplestore_rescan		- rewind and replay the scan
 */
void
tuplestore_rescan(Tuplestorestate *state)
{
686 687
	Assert(state->eflags & EXEC_FLAG_REWIND);

688 689
	switch (state->status)
	{
690 691
		case TSS_INMEM:
			state->eof_reached = false;
692
			state->current = 0;
693 694
			break;
		case TSS_WRITEFILE:
695
			state->eof_reached = false;
696 697
			state->readpos_file = 0;
			state->readpos_offset = 0L;
698 699
			break;
		case TSS_READFILE:
700
			state->eof_reached = false;
701
			if (BufFileSeek(state->myfile, 0, 0L, SEEK_SET) != 0)
702
				elog(ERROR, "seek to start failed");
703 704
			break;
		default:
705
			elog(ERROR, "invalid tuplestore state");
706 707 708 709 710 711 712 713 714 715
			break;
	}
}

/*
 * tuplestore_markpos	- saves current position in the tuple sequence
 */
void
tuplestore_markpos(Tuplestorestate *state)
{
716 717
	Assert(state->eflags & EXEC_FLAG_MARK);

718 719
	switch (state->status)
	{
720 721
		case TSS_INMEM:
			state->markpos_current = state->current;
722 723 724 725 726 727 728 729 730 731 732 733 734 735
			/*
			 * We can truncate the tuplestore if neither backward scan nor
			 * rewind capability are required by the caller.  There will
			 * never be a need to back up past the mark point.
			 *
			 * Note: you might think we could remove all the tuples before
			 * "current", since that one is the next to be returned.  However,
			 * since tuplestore_gettuple returns a direct pointer to our
			 * internal copy of the tuple, it's likely that the caller has
			 * still got the tuple just before "current" referenced in a slot.
			 * Don't free it yet.
			 */
			if (!(state->eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_REWIND)))
				tuplestore_trim(state, 1);
736 737 738 739 740 741 742 743 744 745 746 747 748 749
			break;
		case TSS_WRITEFILE:
			if (state->eof_reached)
			{
				/* Need to record the implicit read position */
				BufFileTell(state->myfile,
							&state->markpos_file,
							&state->markpos_offset);
			}
			else
			{
				state->markpos_file = state->readpos_file;
				state->markpos_offset = state->readpos_offset;
			}
750 751 752 753 754 755 756
			break;
		case TSS_READFILE:
			BufFileTell(state->myfile,
						&state->markpos_file,
						&state->markpos_offset);
			break;
		default:
757
			elog(ERROR, "invalid tuplestore state");
758 759 760 761 762 763 764 765 766 767 768
			break;
	}
}

/*
 * tuplestore_restorepos - restores current position in tuple sequence to
 *						  last saved position
 */
void
tuplestore_restorepos(Tuplestorestate *state)
{
769 770
	Assert(state->eflags & EXEC_FLAG_MARK);

771 772
	switch (state->status)
	{
773 774 775 776 777 778 779 780
		case TSS_INMEM:
			state->eof_reached = false;
			state->current = state->markpos_current;
			break;
		case TSS_WRITEFILE:
			state->eof_reached = false;
			state->readpos_file = state->markpos_file;
			state->readpos_offset = state->markpos_offset;
781 782
			break;
		case TSS_READFILE:
783
			state->eof_reached = false;
784 785 786 787 788 789 790
			if (BufFileSeek(state->myfile,
							state->markpos_file,
							state->markpos_offset,
							SEEK_SET) != 0)
				elog(ERROR, "tuplestore_restorepos failed");
			break;
		default:
791
			elog(ERROR, "invalid tuplestore state");
792 793 794 795
			break;
	}
}

796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844
/*
 * tuplestore_trim	- remove all but ntuples tuples before current
 */
static void
tuplestore_trim(Tuplestorestate *state, int ntuples)
{
	int			nremove;
	int			i;

	/*
	 * We don't bother trimming temp files since it usually would mean more
	 * work than just letting them sit in kernel buffers until they age out.
	 */
	if (state->status != TSS_INMEM)
		return;

	nremove = state->current - ntuples;
	if (nremove <= 0)
		return;					/* nothing to do */
	Assert(nremove <= state->memtupcount);

	/* Release no-longer-needed tuples */
	for (i = 0; i < nremove; i++)
	{
		FREEMEM(state, GetMemoryChunkSpace(state->memtuples[i]));
		pfree(state->memtuples[i]);
	}

	/*
	 * Slide the array down and readjust pointers.  This may look pretty
	 * stupid, but we expect that there will usually not be very many
	 * tuple-pointers to move, so this isn't that expensive; and it keeps
	 * a lot of other logic simple.
	 *
	 * In fact, in the current usage for merge joins, it's demonstrable that
	 * there will always be exactly one non-removed tuple; so optimize that
	 * case.
	 */
	if (nremove + 1 == state->memtupcount)
		state->memtuples[0] = state->memtuples[nremove];
	else
		memmove(state->memtuples, state->memtuples + nremove,
				(state->memtupcount - nremove) * sizeof(void *));

	state->memtupcount -= nremove;
	state->current -= nremove;
	state->markpos_current -= nremove;
}

845 846 847 848 849 850 851 852 853

/*
 * Tape interface routines
 */

static unsigned int
getlen(Tuplestorestate *state, bool eofOK)
{
	unsigned int len;
854
	size_t		nbytes;
855

856 857 858 859
	nbytes = BufFileRead(state->myfile, (void *) &len, sizeof(len));
	if (nbytes == sizeof(len))
		return len;
	if (nbytes != 0)
860
		elog(ERROR, "unexpected end of tape");
861
	if (!eofOK)
862
		elog(ERROR, "unexpected end of data");
863
	return 0;
864 865 866 867 868
}


/*
 * Routines specialized for HeapTuple case
869 870 871 872 873 874
 *
 * The stored form is actually a MinimalTuple, but for largely historical
 * reasons we allow COPYTUP to work from a HeapTuple.
 *
 * Since MinimalTuple already has length in its first word, we don't need
 * to write that separately.
875 876 877 878 879
 */

static void *
copytup_heap(Tuplestorestate *state, void *tup)
{
880
	MinimalTuple tuple;
881

882
	tuple = minimal_tuple_from_heap_tuple((HeapTuple) tup);
883 884
	USEMEM(state, GetMemoryChunkSpace(tuple));
	return (void *) tuple;
885 886 887 888 889
}

static void
writetup_heap(Tuplestorestate *state, void *tup)
{
890 891
	MinimalTuple tuple = (MinimalTuple) tup;
	unsigned int tuplen = tuple->t_len;
892

893
	if (BufFileWrite(state->myfile, (void *) tuple, tuplen) != (size_t) tuplen)
894
		elog(ERROR, "write failed");
895
	if (state->eflags & EXEC_FLAG_BACKWARD)	/* need trailing length word? */
896 897
		if (BufFileWrite(state->myfile, (void *) &tuplen,
						 sizeof(tuplen)) != sizeof(tuplen))
898
			elog(ERROR, "write failed");
899

900
	FREEMEM(state, GetMemoryChunkSpace(tuple));
901
	heap_free_minimal_tuple(tuple);
902 903 904 905 906
}

static void *
readtup_heap(Tuplestorestate *state, unsigned int len)
{
907 908
	MinimalTuple tuple = (MinimalTuple) palloc(len);
	unsigned int tuplen;
909

910
	USEMEM(state, GetMemoryChunkSpace(tuple));
911
	/* read in the tuple proper */
912 913 914
	tuple->t_len = len;
	if (BufFileRead(state->myfile, (void *) ((char *) tuple + sizeof(int)),
					len - sizeof(int)) != (size_t) (len - sizeof(int)))
915
		elog(ERROR, "unexpected end of data");
916
	if (state->eflags & EXEC_FLAG_BACKWARD)	/* need trailing length word? */
917 918
		if (BufFileRead(state->myfile, (void *) &tuplen,
						sizeof(tuplen)) != sizeof(tuplen))
919
			elog(ERROR, "unexpected end of data");
920 921
	return (void *) tuple;
}