Fix for backward cursors with ORDER BY.

29 years ago · 78351f422b
parent 9acf938c10
commit 78351f422b
2 changed files with 113 additions and 25 deletions
--- a/src/backend/utils/sort/psort.c
+++ b/src/backend/utils/sort/psort.c
@ -7,7 +7,7 @@
 *
 *
 * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/utils/sort/Attic/psort.c,v 1.25 1997/09/26 20:05:47 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/utils/sort/Attic/psort.c,v 1.26 1997/10/15 06:36:08 vadim Exp $
 *
 * NOTES
 *		Sorts the first relation into the second relation.
@ -80,7 +80,12 @@ static int _psort_cmp (HeapTuple *ltup, HeapTuple *rtup);

 #define TEMPDIR "./"

-static long shortzero = 0;		/* used to delimit runs */
+/* 
+ * tlenzero used to delimit runs; both vars below must have
+ * the same size as HeapTuple->t_len
+ */
+static unsigned int tlenzero = 0;
+static unsigned int tlendummy;

 static TupleDesc	PsortTupDesc;
 static ScanKey		PsortKeys;		/* used by _psort_cmp */
@ -150,6 +155,7 @@ psort_begin(Sort * node, int nkeys, ScanKey key)
 	PS(node)->tupcount = 0;

 	PS(node)->using_tape_files = false;
+	PS(node)->all_fetched = false;
 	PS(node)->psort_grab_file = NULL;
 	PS(node)->memtuples = NULL;

@ -219,21 +225,24 @@ inittapes(Sort * node)
 *		GETTUP			- reads the tuple
 *
 *		Note:
- *				LEN field must be a short; FP is a stream
+ *				LEN field must be as HeapTuple->t_len; FP is a stream
 */


 #define PUTTUP(NODE, TUP, FP) do {\
 	((Psortstate *)NODE->psortstate)->BytesWritten += (TUP)->t_len; \
-	fwrite((char *)TUP, (TUP)->t_len, 1, FP);} while (0)
-#define ENDRUN(FP)		fwrite((char *)&shortzero, sizeof (shortzero), 1, FP)
-#define GETLEN(LEN, FP) fread((char *)&(LEN), sizeof (shortzero), 1, FP)
+	fwrite((char *)TUP, (TUP)->t_len, 1, FP); \
+	fwrite((char *)&((TUP)->t_len), sizeof (tlendummy), 1, FP); \
+	} while (0)
+#define ENDRUN(FP)		fwrite((char *)&tlenzero, sizeof (tlenzero), 1, FP)
+#define GETLEN(LEN, FP) fread((char *)&(LEN), sizeof (tlenzero), 1, FP)
 #define ALLOCTUP(LEN)	((HeapTuple)palloc((unsigned)LEN))
 #define GETTUP(NODE, TUP, LEN, FP) do {\
 	IncrProcessed(); \
-	((Psortstate *)NODE->psortstate)->BytesRead += (LEN) - sizeof (shortzero); \
-	fread((char *)(TUP) + sizeof (shortzero), (LEN) - sizeof (shortzero), 1, FP);} \
-								while (0)
+	((Psortstate *)NODE->psortstate)->BytesRead += (LEN) - sizeof (tlenzero); \
+	fread((char *)(TUP) + sizeof (tlenzero), (LEN) - sizeof (tlenzero), 1, FP); \
+	fread((char *)&tlendummy, sizeof (tlendummy), 1, FP); \
+	} while (0)
 #define SETTUPLEN(TUP, LEN)		(TUP)->t_len = LEN

 /*
@ -629,11 +638,11 @@ merge(Sort * node, struct tape * dest)
 	register struct tape *lasttp;		/* (TAPE[P]) */
 	register struct tape *tp;
 	struct leftist *tuples;
-	FILE	   *destfile;
-	int			times;			/* runs left to merge */
-	int			outdummy;		/* complete dummy runs */
-	short		fromtape;
-	long		tuplen;
+	FILE		   *destfile;
+	int				times;			/* runs left to merge */
+	int				outdummy;		/* complete dummy runs */
+	short			fromtape;
+	unsigned int	tuplen;

 	Assert(node != (Sort *) NULL);
 	Assert(PS(node) != (Psortstate *) NULL);
@ -767,16 +776,20 @@ dumptuples(FILE * file, Sort * node)
 HeapTuple
 psort_grabtuple(Sort * node, bool * should_free)
 {
-	register HeapTuple tup;
-	long		tuplen;
+	register HeapTuple	tup;

 	Assert(node != (Sort *) NULL);
 	Assert(PS(node) != (Psortstate *) NULL);

 	if (PS(node)->using_tape_files == true)
 	{
-		if (!feof(PS(node)->psort_grab_file))
+		unsigned int	tuplen;
+		
+		*should_free = true;
+		if (ScanDirectionIsForward (node->plan.state->es_direction))
 		{
+			if (PS(node)->all_fetched)
+				return NULL;
 			if (GETLEN(tuplen, PS(node)->psort_grab_file) && tuplen != 0)
 			{
 				tup = (HeapTuple) palloc((unsigned) tuplen);
@ -784,25 +797,99 @@ psort_grabtuple(Sort * node, bool * should_free)
 				GETTUP(node, tup, tuplen, PS(node)->psort_grab_file);

 				/* Update current merged sort file position */
-				PS(node)->psort_current += tuplen;
-				*should_free = true;
+				PS(node)->psort_current += tuplen + sizeof (tlendummy);
 				return tup;
 			}
 			else
+			{
+				PS(node)->all_fetched = true;
 				return NULL;
+			}
 		}
-		else
+		/* Backward */
+		if (PS(node)->psort_current <= sizeof (tlendummy))
 			return NULL;
+		/* 
+		 * if all tuples are fetched already then we return last tuple, 
+		 * else - tuple before last returned.
+		 */
+		if (PS(node)->all_fetched)
+		{
+			/* psort_current is pointing to the zero tuplen at the end of file */
+			fseek(PS(node)->psort_grab_file, 
+					PS(node)->psort_current - sizeof (tlendummy), SEEK_SET);
+			GETLEN(tuplen, PS(node)->psort_grab_file);
+			if (PS(node)->psort_current < tuplen)
+				elog (FATAL, "psort_grabtuple: too big last tuple len in backward scan");
+			PS(node)->all_fetched = false;
+		}
+		else
+		{
+			/* move to position of end tlen of prev tuple */
+			PS(node)->psort_current -= sizeof (tlendummy);
+			fseek(PS(node)->psort_grab_file, PS(node)->psort_current, SEEK_SET);
+			GETLEN(tuplen, PS(node)->psort_grab_file);	/* get tlen of prev tuple */
+			if (tuplen == 0)
+				elog (FATAL, "psort_grabtuple: tuplen is 0 in backward scan");
+			if (PS(node)->psort_current <= tuplen + sizeof (tlendummy))
+			{	/* prev tuple should be first one */
+				if (PS(node)->psort_current != tuplen)
+					elog (FATAL, "psort_grabtuple: first tuple expected in backward scan");
+				PS(node)->psort_current = 0;
+				fseek(PS(node)->psort_grab_file, PS(node)->psort_current, SEEK_SET);
+				return NULL;
+			}
+			/* 
+			 * Get position of prev tuple. This tuple becomes current tuple
+			 * now and we have to return previous one.
+			 */
+			PS(node)->psort_current -= tuplen;
+			/* move to position of end tlen of prev tuple */
+			fseek(PS(node)->psort_grab_file, 
+					PS(node)->psort_current - sizeof (tlendummy), SEEK_SET);
+			GETLEN(tuplen, PS(node)->psort_grab_file);
+			if (PS(node)->psort_current < tuplen + sizeof (tlendummy))
+				elog (FATAL, "psort_grabtuple: too big tuple len in backward scan");
+		}
+		/* 
+		 * move to prev (or last) tuple start position + sizeof(t_len) 
+		 */
+		fseek(PS(node)->psort_grab_file,
+				PS(node)->psort_current - tuplen, SEEK_SET);
+		tup = (HeapTuple) palloc((unsigned) tuplen);
+		SETTUPLEN(tup, tuplen);
+		GETTUP(node, tup, tuplen, PS(node)->psort_grab_file);
+		return tup;		/* file position is equal to psort_current */
 	}
 	else
 	{
-		if (PS(node)->psort_current < PS(node)->tupcount)
+		*should_free = false;
+		if (ScanDirectionIsForward (node->plan.state->es_direction))
 		{
-			*should_free = false;
-			return (PS(node)->memtuples[PS(node)->psort_current++]);
+			if (PS(node)->psort_current < PS(node)->tupcount)
+				return (PS(node)->memtuples[PS(node)->psort_current++]);
+			else
+			{
+				PS(node)->all_fetched = true;
+				return NULL;
+			}
 		}
-		else
+		/* Backward */
+		if (PS(node)->psort_current <= 0)
 			return NULL;
+		/* 
+		 * if all tuples are fetched already then we return last tuple, 
+		 * else - tuple before last returned.
+		 */
+		if (PS(node)->all_fetched)		
+			PS(node)->all_fetched = false;
+		else
+		{
+			PS(node)->psort_current--;			/* last returned tuple */
+			if (PS(node)->psort_current <= 0)
+				return NULL;
+		}
+		return (PS(node)->memtuples[PS(node)->psort_current - 1]);
 	}
 }

--- a/src/include/utils/psort.h
+++ b/src/include/utils/psort.h
@ -6,7 +6,7 @@
 *
 * Copyright (c) 1994, Regents of the University of California
 *
- * $Id: psort.h,v 1.13 1997/09/18 14:42:35 vadim Exp $
+ * $Id: psort.h,v 1.14 1997/10/15 06:36:36 vadim Exp $
 *
 *-------------------------------------------------------------------------
 */
@ -62,6 +62,7 @@ typedef struct Psortstate
 	long		psort_current;	/* could be file offset, or array index */
 	long		psort_saved;	/* could be file offset, or array index */
 	bool		using_tape_files;
+	bool		all_fetched;	/* this is for cursors */

 	HeapTuple  *memtuples;
 } Psortstate;