Patches for Vadim's multikey indexing...

3 files changed, 190 insertions, 144 deletions

diff --git a/src/backend/access/nbtree/nbtree.c b/src/backend/access/nbtree/nbtree.c
index 8317789a94..0fe6787c01 100644
--- a/src/backend/access/nbtree/nbtree.c
+++ b/src/backend/access/nbtree/nbtree.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *    $Header: /cvsroot/pgsql/src/backend/access/nbtree/nbtree.c,v 1.15 1997/02/22 10:04:14 vadim Exp $
+ *    $Header: /cvsroot/pgsql/src/backend/access/nbtree/nbtree.c,v 1.16 1997/03/18 18:38:35 scrappy Exp $
  *
  * NOTES
  *    This file contains only the public interface routines.
@@ -423,6 +423,7 @@ btrescan(IndexScanDesc scan, bool fromEnd, ScanKey scankey)
     
     /* reset the scan key */
     so->numberOfKeys = scan->numberOfKeys;
+    so->numberOfFirstKeys = 0;
     so->qual_ok = 1;			/* may be changed by _bt_orderkeys */
     if (scan->numberOfKeys > 0) {
 	memmove(scan->keyData,
@@ -433,7 +434,9 @@ btrescan(IndexScanDesc scan, bool fromEnd, ScanKey scankey)
 		so->numberOfKeys * sizeof(ScanKeyData));
 	/* order the keys in the qualification */
 	if (so->numberOfKeys > 1)
-		_bt_orderkeys(scan->relation, &so->numberOfKeys, so->keyData, &so->qual_ok);
+	    _bt_orderkeys(scan->relation, so);
+	else
+	    so->numberOfFirstKeys = 1;
     }
     
     /* finally, be sure that the scan exploits the tree order */
diff --git a/src/backend/access/nbtree/nbtsearch.c b/src/backend/access/nbtree/nbtsearch.c
index 94521ff1c0..2e802ee852 100644
--- a/src/backend/access/nbtree/nbtsearch.c
+++ b/src/backend/access/nbtree/nbtsearch.c
@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *    $Header: /cvsroot/pgsql/src/backend/access/nbtree/nbtsearch.c,v 1.14 1997/02/18 17:13:48 momjian Exp $
+ *    $Header: /cvsroot/pgsql/src/backend/access/nbtree/nbtsearch.c,v 1.15 1997/03/18 18:38:41 scrappy Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -562,7 +562,6 @@ _bt_next(IndexScanDesc scan, ScanDirection dir)
     Page page;
     OffsetNumber offnum;
     RetrieveIndexResult res;
-    BlockNumber blkno;
     ItemPointer current;
     BTItem btitem;
     IndexTuple itup;
@@ -584,31 +583,35 @@ _bt_next(IndexScanDesc scan, ScanDirection dir)
     
     /* we still have the buffer pinned and locked */
     buf = so->btso_curbuf;
-    blkno = BufferGetBlockNumber(buf);
     
-    /* step one tuple in the appropriate direction */
-    if (!_bt_step(scan, &buf, dir))
-	return ((RetrieveIndexResult) NULL);
+    do
+    {
+    	/* step one tuple in the appropriate direction */
+    	if (!_bt_step(scan, &buf, dir))
+	    return ((RetrieveIndexResult) NULL);
     
-    /* by here, current is the tuple we want to return */
-    offnum = ItemPointerGetOffsetNumber(current);
-    page = BufferGetPage(buf);
-    btitem = (BTItem) PageGetItem(page, PageGetItemId(page, offnum));
-    itup = &btitem->bti_itup;
+    	/* by here, current is the tuple we want to return */
+    	offnum = ItemPointerGetOffsetNumber(current);
+    	page = BufferGetPage(buf);
+    	btitem = (BTItem) PageGetItem(page, PageGetItemId(page, offnum));
+    	itup = &btitem->bti_itup;
     
-    if (_bt_checkqual(scan, itup)) {
-	res = FormRetrieveIndexResult(current, &(itup->t_tid));
+    	if (_bt_checkqual(scan, itup)) 
+    	{
+	    res = FormRetrieveIndexResult(current, &(itup->t_tid));
 	
-	/* remember which buffer we have pinned and locked */
-	so->btso_curbuf = buf;
-    } else {
-	ItemPointerSetInvalid(current);
-	so->btso_curbuf = InvalidBuffer;
-	_bt_relbuf(rel, buf, BT_READ);
-	res = (RetrieveIndexResult) NULL;
-    }
+	    /* remember which buffer we have pinned and locked */
+	    so->btso_curbuf = buf;
+	    return (res);
+	}
+
+    } while ( _bt_checkforkeys (scan, itup, so->numberOfFirstKeys) );
+
+    ItemPointerSetInvalid(current);
+    so->btso_curbuf = InvalidBuffer;
+    _bt_relbuf(rel, buf, BT_READ);
     
-    return (res);
+    return ((RetrieveIndexResult) NULL);
 }
 
 /*
@@ -660,13 +663,6 @@ _bt_first(IndexScanDesc scan, ScanDirection dir)
      *  ordered to take advantage of index ordering) to position ourselves
      *  at the right place in the scan.
      */
-    
-    /*
-     *  XXX -- The attribute number stored in the scan key is the attno
-     *	       in the heap relation.  We need to transmogrify this into
-     *         the index relation attno here.  For the moment, we have
-     *	       hardwired attno == 1.
-     */
     proc = index_getprocid(rel, 1, BTORDER_PROC);
     ScanKeyEntryInitialize(&skdata, so->keyData[0].sk_flags, 1, proc,
 			   so->keyData[0].sk_argument);
@@ -802,12 +798,20 @@ _bt_first(IndexScanDesc scan, ScanDirection dir)
     btitem = (BTItem) PageGetItem(page, PageGetItemId(page, offnum));
     itup = &btitem->bti_itup;
     
-    if (_bt_checkqual(scan, itup)) {
+    if ( _bt_checkqual(scan, itup) )
+    {
 	res = FormRetrieveIndexResult(current, &(itup->t_tid));
 	
 	/* remember which buffer we have pinned */
 	so->btso_curbuf = buf;
-    } else {
+    }
+    else if ( _bt_checkforkeys (scan, itup, so->numberOfFirstKeys) )
+    {
+	so->btso_curbuf = buf;
+	return (_bt_next (scan, dir));
+    }
+    else
+    {
 	ItemPointerSetInvalid(current);
 	so->btso_curbuf = InvalidBuffer;
 	_bt_relbuf(rel, buf, BT_READ);
@@ -1224,7 +1228,14 @@ _bt_endpoint(IndexScanDesc scan, ScanDirection dir)
 	
 	/* remember which buffer we have pinned */
 	so->btso_curbuf = buf;
-    } else {
+    }
+    else if ( _bt_checkforkeys (scan, itup, so->numberOfFirstKeys) )
+    {
+	so->btso_curbuf = buf;
+	return (_bt_next (scan, dir));
+    }
+    else
+    {
 	_bt_relbuf(rel, buf, BT_READ);
 	res = (RetrieveIndexResult) NULL;
     }
diff --git a/src/backend/access/nbtree/nbtutils.c b/src/backend/access/nbtree/nbtutils.c
index 703acd62fa..6d0a40ef13 100644
--- a/src/backend/access/nbtree/nbtutils.c
+++ b/src/backend/access/nbtree/nbtutils.c
@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *    $Header: /cvsroot/pgsql/src/backend/access/nbtree/nbtutils.c,v 1.7 1996/11/05 10:35:38 scrappy Exp $
+ *    $Header: /cvsroot/pgsql/src/backend/access/nbtree/nbtutils.c,v 1.8 1997/03/18 18:38:46 scrappy Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -80,7 +80,7 @@ _bt_freestack(BTStack stack)
  *	more than one qual clauses using this index.
  */
 void
-_bt_orderkeys(Relation relation, uint16 *numberOfKeys, ScanKey key, uint16 *qual_ok)
+_bt_orderkeys(Relation relation, BTScanOpaque so)
 {
     ScanKey xform;
     ScanKeyData *cur;
@@ -89,42 +89,137 @@ _bt_orderkeys(Relation relation, uint16 *numberOfKeys, ScanKey key, uint16 *qual
     long test;
     int i, j;
     int init[BTMaxStrategyNumber+1];
+    ScanKey key;
+    uint16 numberOfKeys, new_numberOfKeys = 0;
+    AttrNumber attno = 1;
     
-    /* haven't looked at any strategies yet */
-    for (i = 0; i <= BTMaxStrategyNumber; i++)
-	init[i] = 0;
+    numberOfKeys = so->numberOfKeys;
+    key = so->keyData;
+    
+    if ( numberOfKeys <= 1 )
+    	return;
     
     /* get space for the modified array of keys */
     nbytes = BTMaxStrategyNumber * sizeof(ScanKeyData);
     xform = (ScanKey) palloc(nbytes);
-    memset(xform, 0, nbytes); 
     
-    
-    /* get the strategy map for this index/attribute pair */
-    /*
-     *  XXX
-     *  When we support multiple keys in a single index, this is what
-     *  we'll want to do.  At present, the planner is hosed, so we
-     *  hard-wire the attribute number below.  Postgres only does single-
-     *  key indices...
-     * map = IndexStrategyGetStrategyMap(RelationGetIndexStrategy(relation),
-     * 				    BTMaxStrategyNumber,
-     * 				    key->data[0].attributeNumber);
-     */
+    cur = &key[0];
+    if ( cur->sk_attno != 1 )
+	elog (WARN, "_bt_orderkeys: key(s) for attribute 1 missed");
+
+    memset(xform, 0, nbytes); 
     map = IndexStrategyGetStrategyMap(RelationGetIndexStrategy(relation),
 				      BTMaxStrategyNumber,
-				      1 /* XXX */ );
+				      attno);
+    for (j = 0; j <= BTMaxStrategyNumber; j++)
+	init[j] = 0;
     
     /* check each key passed in */
-    for (i = *numberOfKeys; --i >= 0; ) {
-	cur = &key[i];
-	for (j = BTMaxStrategyNumber; --j >= 0; ) {
+    for (i = 0; ; )
+    {
+	if ( i < numberOfKeys )
+	    cur = &key[i];
+	if ( i == numberOfKeys || cur->sk_attno != attno )
+	{
+	    if ( cur->sk_attno != attno + 1 && i < numberOfKeys )
+	    {
+	    	elog (WARN, "_bt_orderkeys: key(s) for attribute %d missed", attno + 1);
+	    }
+	    /* 
+	     * If = has been specified, no other key will be used.
+	     * In case of key < 2 && key == 1 and so on 
+     	     * we have to set qual_ok to 0
+     	     */
+	    if (init[BTEqualStrategyNumber - 1])
+	    {
+		ScanKeyData *eq, *chk;
+
+		eq = &xform[BTEqualStrategyNumber - 1];
+		for (j = BTMaxStrategyNumber; --j >= 0; )
+		{
+		    if ( j == (BTEqualStrategyNumber - 1) || init[j] == 0 )
+			continue;
+		    chk = &xform[j];
+		    test = (long) fmgr(chk->sk_procedure, eq->sk_argument, chk->sk_argument);
+		    if (!test)
+			so->qual_ok = 0;
+		}
+		init[BTLessStrategyNumber - 1] = 0;
+		init[BTLessEqualStrategyNumber - 1] = 0;
+		init[BTGreaterEqualStrategyNumber - 1] = 0;
+		init[BTGreaterStrategyNumber - 1] = 0;
+	    }
+    
+	    /* only one of <, <= */
+	    if (init[BTLessStrategyNumber - 1]
+			&& init[BTLessEqualStrategyNumber - 1])
+	    {
+		ScanKeyData *lt, *le;
+	
+		lt = &xform[BTLessStrategyNumber - 1];
+		le = &xform[BTLessEqualStrategyNumber - 1];
+		/*
+		 *  DO NOT use the cached function stuff here -- this is key
+		 *  ordering, happens only when the user expresses a hokey
+		 *  qualification, and gets executed only once, anyway.  The
+		 *  transform maps are hard-coded, and can't be initialized
+		 *  in the correct way.
+		 */
+		test = (long) fmgr(le->sk_procedure, lt->sk_argument, le->sk_argument);
+		if (test)
+	    	    init[BTLessEqualStrategyNumber - 1] = 0;
+		else
+		    init[BTLessStrategyNumber - 1] = 0;
+    	    }
+    
+	    /* only one of >, >= */
+	    if (init[BTGreaterStrategyNumber - 1]
+			&& init[BTGreaterEqualStrategyNumber - 1])
+	    {
+	    	ScanKeyData *gt, *ge;
+	
+		gt = &xform[BTGreaterStrategyNumber - 1];
+		ge = &xform[BTGreaterEqualStrategyNumber - 1];
+	
+		/* see note above on function cache */
+		test = (long) fmgr(ge->sk_procedure, gt->sk_argument, ge->sk_argument);
+		if (test)
+	    	    init[BTGreaterEqualStrategyNumber - 1] = 0;
+		else
+	    	    init[BTGreaterStrategyNumber - 1] = 0;
+    	    }
+    
+    	    /* okay, reorder and count */
+    	    for (j = BTMaxStrategyNumber; --j >= 0; )
+	    	if (init[j])
+		    key[new_numberOfKeys++] = xform[j];
+    
+    	    if ( attno == 1 )
+    		so->numberOfFirstKeys = new_numberOfKeys;
+    	    
+    	    if ( i == numberOfKeys )
+    	    	break;
+
+	    /* initialization for new attno */    	    
+    	    attno = cur->sk_attno;
+    	    memset(xform, 0, nbytes); 
+	    map = IndexStrategyGetStrategyMap(RelationGetIndexStrategy(relation),
+				      BTMaxStrategyNumber,
+				      attno);
+	    /* haven't looked at any strategies yet */
+	    for (j = 0; j <= BTMaxStrategyNumber; j++)
+		init[j] = 0;
+	}
+
+	for (j = BTMaxStrategyNumber; --j >= 0; )
+	{
 	    if (cur->sk_procedure == map->entry[j].sk_procedure)
-		break;
+	    	break;
 	}
 	
 	/* have we seen one of these before? */
-	if (init[j]) {
+	if (init[j])
+	{
 	    /* yup, use the appropriate value */
 	    test =
 		(long) FMGR_PTR2(cur->sk_func, cur->sk_procedure,
@@ -132,97 +227,18 @@ _bt_orderkeys(Relation relation, uint16 *numberOfKeys, ScanKey key, uint16 *qual
 	    if (test)
 		xform[j].sk_argument = cur->sk_argument;
 	    else if ( j == (BTEqualStrategyNumber - 1) )
-	    	*qual_ok = 0;		/* key == a && key == b, but a != b */
-	} else {
+	    	so->qual_ok = 0;	/* key == a && key == b, but a != b */
+	} else
+	{
 	    /* nope, use this value */
 	    memmove(&xform[j], cur, sizeof(*cur));
-	   
 	    init[j] = 1;
 	}
-    }
-    
-    /* if = has been specified, no other key will be used */
-    /*
-     * XXX
-     * But in case of key < 2 && key == 1 and so on 
-     * we have to set qual_ok to 0
-     */
-    if (init[BTEqualStrategyNumber - 1]) {
-
-	ScanKeyData *eq, *chk;
-
-	eq = &xform[BTEqualStrategyNumber - 1];
-
-	for (j = BTMaxStrategyNumber; --j >= 0; )
-	{
-		if ( j == (BTEqualStrategyNumber - 1) || init[j] == 0 )
-			continue;
-
-		chk = &xform[j];
-
-		test = (long) fmgr(chk->sk_procedure, eq->sk_argument, chk->sk_argument);
 	
-		if (!test)
-			*qual_ok = 0;
-	}
-
-	init[BTLessStrategyNumber - 1] = 0;
-	init[BTLessEqualStrategyNumber - 1] = 0;
-	init[BTGreaterEqualStrategyNumber - 1] = 0;
-	init[BTGreaterStrategyNumber - 1] = 0;
+	i++;
     }
     
-    /* only one of <, <= */
-    if (init[BTLessStrategyNumber - 1]
-	&& init[BTLessEqualStrategyNumber - 1]) {
-	
-	ScanKeyData *lt, *le;
-	
-	lt = &xform[BTLessStrategyNumber - 1];
-	le = &xform[BTLessEqualStrategyNumber - 1];
-	
-	/*
-	 *  DO NOT use the cached function stuff here -- this is key
-	 *  ordering, happens only when the user expresses a hokey
-	 *  qualification, and gets executed only once, anyway.  The
-	 *  transform maps are hard-coded, and can't be initialized
-	 *  in the correct way.
-	 */
-	
-	test = (long) fmgr(le->sk_procedure, lt->sk_argument, le->sk_argument);
-	
-	if (test)
-	    init[BTLessEqualStrategyNumber - 1] = 0;
-	else
-	    init[BTLessStrategyNumber - 1] = 0;
-    }
-    
-    /* only one of >, >= */
-    if (init[BTGreaterStrategyNumber - 1]
-	&& init[BTGreaterEqualStrategyNumber - 1]) {
-	
-	ScanKeyData *gt, *ge;
-	
-	gt = &xform[BTGreaterStrategyNumber - 1];
-	ge = &xform[BTGreaterEqualStrategyNumber - 1];
-	
-	/* see note above on function cache */
-	test = (long) fmgr(ge->sk_procedure, gt->sk_argument, ge->sk_argument);
-	
-	if (test)
-	    init[BTGreaterEqualStrategyNumber - 1] = 0;
-	else
-	    init[BTGreaterStrategyNumber - 1] = 0;
-    }
-    
-    /* okay, reorder and count */
-    j = 0;
-    
-    for (i = BTMaxStrategyNumber; --i >= 0; )
-	if (init[i])
-	    key[j++] = xform[i];
-    
-    *numberOfKeys = j;
+    so->numberOfKeys = new_numberOfKeys;
     
     pfree(xform);
 }
@@ -230,9 +246,25 @@ _bt_orderkeys(Relation relation, uint16 *numberOfKeys, ScanKey key, uint16 *qual
 bool
 _bt_checkqual(IndexScanDesc scan, IndexTuple itup)
 {
-    if (scan->numberOfKeys > 0)
+    BTScanOpaque so;
+    
+    so = (BTScanOpaque) scan->opaque;
+    if (so->numberOfKeys > 0)
+	return (index_keytest(itup, RelationGetTupleDescriptor(scan->relation),
+			      so->numberOfKeys, so->keyData));
+    else
+	return (true);
+}
+
+bool
+_bt_checkforkeys(IndexScanDesc scan, IndexTuple itup, Size keysz)
+{
+    BTScanOpaque so;
+    
+    so = (BTScanOpaque) scan->opaque;
+    if ( keysz > 0 && so->numberOfKeys >= keysz )
 	return (index_keytest(itup, RelationGetTupleDescriptor(scan->relation),
-			      scan->numberOfKeys, scan->keyData));
+			      keysz, so->keyData));
     else
 	return (true);
 }