Replace nth() calls in inner loops with traversal of the list via

lnext, to eliminate O(N^2) behavior with lots of indexquals.

diff --git a/src/backend/executor/nodeIndexscan.c b/src/backend/executor/nodeIndexscan.c
index b9a54cb938acc73c73cbe8c19dd48723f9b13de1..e5f7642c85cc6e2de127c0f9673c07557614d9d7 100644 (file)
--- a/src/backend/executor/nodeIndexscan.c
+++ b/src/backend/executor/nodeIndexscan.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *       $Header: /cvsroot/pgsql/src/backend/executor/nodeIndexscan.c,v 1.45 2000/01/26 05:56:23 momjian Exp $
+ *       $Header: /cvsroot/pgsql/src/backend/executor/nodeIndexscan.c,v 1.46 2000/02/05 23:19:44 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -124,7 +124,7 @@ IndexNext(IndexScan *node)
        if (estate->es_evTuple != NULL &&
                estate->es_evTuple[node->scan.scanrelid - 1] != NULL)
        {
-               int                     iptr;
+               List       *qual;
 
                ExecClearTuple(slot);
                if (estate->es_evTupleNull[node->scan.scanrelid - 1])
@@ -134,20 +134,23 @@ IndexNext(IndexScan *node)
                slot->val = estate->es_evTuple[node->scan.scanrelid - 1];
                slot->ttc_shouldFree = false;
 
-               for (iptr = 0; iptr < numIndices; iptr++)
+               scanstate->cstate.cs_ExprContext->ecxt_scantuple = slot;
+
+               /* Does the tuple meet any of the OR'd indxqual conditions? */
+               foreach(qual, node->indxqualorig)
                {
-                       scanstate->cstate.cs_ExprContext->ecxt_scantuple = slot;
-                       if (ExecQual(nth(iptr, node->indxqualorig),
+                       if (ExecQual((List *) lfirst(qual),
                                                 scanstate->cstate.cs_ExprContext,
                                                 false))
                                break;
                }
-               if (iptr == numIndices) /* would not be returned by indices */
+               if (qual == NIL)                /* would not be returned by indices */
                        slot->val = NULL;
 
                /* Flag for the next call that no more tuples */
                estate->es_evTupleNull[node->scan.scanrelid - 1] = true;
-               return (slot);
+
+               return slot;
        }
 
        tuple = &(indexstate->iss_htup);
@@ -189,14 +192,14 @@ IndexNext(IndexScan *node)
                        {
                                bool            prev_matches = false;
                                int                     prev_index;
+                               List       *qual;
 
-                               /* ----------------
+                               /*
                                 *      store the scanned tuple in the scan tuple slot of
                                 *      the scan state.  Eventually we will only do this and not
                                 *      return a tuple.  Note: we pass 'false' because tuples
                                 *      returned by amgetnext are pointers onto disk pages and
                                 *      must not be pfree()'d.
-                                * ----------------
                                 */
                                ExecStoreTuple(tuple,   /* tuple to store */
                                                           slot,        /* slot to store in */
@@ -211,28 +214,29 @@ IndexNext(IndexScan *node)
                                ReleaseBuffer(buffer);
 
                                /*
-                                * We must check to see if the current tuple would have
-                                * been matched by an earlier index, so we don't double
-                                * report it. We do this by passing the tuple through
-                                * ExecQual and look for failure with all previous
-                                * qualifications.
+                                * We must check to see if the current tuple was already
+                                * matched by an earlier index, so we don't double-report it.
+                                * We do this by passing the tuple through ExecQual and
+                                * checking for failure with all previous qualifications.
                                 */
+                               scanstate->cstate.cs_ExprContext->ecxt_scantuple = slot;
+                               qual = node->indxqualorig;
                                for (prev_index = 0; prev_index < indexstate->iss_IndexPtr;
                                         prev_index++)
                                {
-                                       scanstate->cstate.cs_ExprContext->ecxt_scantuple = slot;
-                                       if (ExecQual(nth(prev_index, node->indxqualorig),
+                                       if (ExecQual((List *) lfirst(qual),
                                                                 scanstate->cstate.cs_ExprContext,
                                                                 false))
                                        {
                                                prev_matches = true;
                                                break;
                                        }
+                                       qual = lnext(qual);
                                }
                                if (!prev_matches)
-                                       return slot;
-                               else
-                                       ExecClearTuple(slot);
+                                       return slot;            /* OK to return tuple */
+                               /* Duplicate tuple, so drop it and loop back for another */
+                               ExecClearTuple(slot);
                        }
                }
                if (indexNumber < numIndices)
@@ -329,7 +333,6 @@ ExecIndexReScan(IndexScan *node, ExprContext *exprCtxt, Plan *parent)
        scanDescs = indexstate->iss_ScanDescs;
        scanKeys = indexstate->iss_ScanKeys;
        runtimeKeyInfo = (Pointer *) indexstate->iss_RuntimeKeyInfo;
-       indxqual = node->indxqual;
        numScanKeys = indexstate->iss_NumScanKeys;
        indexstate->iss_IndexPtr = -1;
        if (ScanDirectionIsBackward(node->indxorderdir))
@@ -352,9 +355,11 @@ ExecIndexReScan(IndexScan *node, ExprContext *exprCtxt, Plan *parent)
        /*
         * get the index qualifications and recalculate the appropriate values
         */
+       indxqual = node->indxqual;
        for (i = 0; i < numIndices; i++)
        {
-               qual = nth(i, indxqual);
+               qual = lfirst(indxqual);
+               indxqual = lnext(indxqual);
                n_keys = numScanKeys[i];
                scan_keys = (ScanKey) scanKeys[i];
 
@@ -672,7 +677,6 @@ ExecInitIndexScan(IndexScan *node, EState *estate, Plan *parent)
         * ----------------
         */
        indxid = node->indxid;
-       indxqual = node->indxqual;
        numIndices = length(indxid);
        indexPtr = -1;
 
@@ -701,6 +705,7 @@ ExecInitIndexScan(IndexScan *node, EState *estate, Plan *parent)
         *      build the index scan keys from the index qualification
         * ----------------
         */
+       indxqual = node->indxqual;
        for (i = 0; i < numIndices; i++)
        {
                int                     j;
@@ -709,7 +714,8 @@ ExecInitIndexScan(IndexScan *node, EState *estate, Plan *parent)
                ScanKey         scan_keys;
                int                *run_keys;
 
-               qual = nth(i, indxqual);
+               qual = lfirst(indxqual);
+               indxqual = lnext(indxqual);
                n_keys = length(qual);
                scan_keys = (n_keys <= 0) ? NULL :
                        (ScanKey) palloc(n_keys * sizeof(ScanKeyData));
@@ -743,8 +749,8 @@ ExecInitIndexScan(IndexScan *node, EState *estate, Plan *parent)
                        clause = nth(j, qual);
 
                        op = (Oper *) clause->oper;
-                       if (!IsA(op, Oper))
-                               elog(ERROR, "ExecInitIndexScan: op not an Oper!");
+                       if (!IsA(clause, Expr) || !IsA(op, Oper))
+                               elog(ERROR, "ExecInitIndexScan: indxqual not an opclause!");
 
                        opid = op->opid;
 
@@ -1066,9 +1072,7 @@ ExecInitIndexScan(IndexScan *node, EState *estate, Plan *parent)
         */
        for (i = 0; i < numIndices; i++)
        {
-               Oid                     indexOid;
-
-               indexOid = (Oid) nthi(i, indxid);
+               Oid                     indexOid = (Oid) nthi(i, indxid);
 
                if (indexOid != 0)
                {