Speed up findObjectByCatalogId() to get rid of the other salient

bottleneck in the new pg_dump code.

diff --git a/src/bin/pg_dump/common.c b/src/bin/pg_dump/common.c
index 81d9cc284c9812c26619724391cd09582c30b95d..45bc90fcccb5784cf7ad26e8825df2a20a16255f 100644 (file)
--- a/src/bin/pg_dump/common.c
+++ b/src/bin/pg_dump/common.c
@@ -11,7 +11,7 @@
  *
  *
  * IDENTIFICATION
- *   $PostgreSQL: pgsql/src/bin/pg_dump/common.c,v 1.78 2003/12/06 03:00:11 tgl Exp $
+ *   $PostgreSQL: pgsql/src/bin/pg_dump/common.c,v 1.79 2003/12/07 03:14:01 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -37,6 +37,13 @@ static DumpableObject **dumpIdMap = NULL;
 static int allocedDumpIds = 0;
 static DumpId lastDumpId = 0;
 
+/*
+ * Variables for mapping CatalogId to DumpableObject
+ */
+static bool catalogIdMapValid = false;
+static DumpableObject **catalogIdMap = NULL;
+static int numCatalogIds = 0;
+
 /*
  * These variables are static to avoid the notational cruft of having to pass
  * them into findTableByOid() and friends.
@@ -51,12 +58,13 @@ static int      numFuncs;
 static int     numOperators;
 
 
-static void findParentsByOid(TableInfo *self,
-                            InhInfo *inhinfo, int numInherits);
 static void flagInhTables(TableInfo *tbinfo, int numTables,
              InhInfo *inhinfo, int numInherits);
 static void flagInhAttrs(TableInfo *tbinfo, int numTables,
             InhInfo *inhinfo, int numInherits);
+static int DOCatalogIdCompare(const void *p1, const void *p2);
+static void findParentsByOid(TableInfo *self,
+                            InhInfo *inhinfo, int numInherits);
 static int strInArray(const char *pattern, char **arr, int arr_size);
 
 
@@ -413,6 +421,9 @@ AssignDumpId(DumpableObject *dobj)
        allocedDumpIds = newAlloc;
    }
    dumpIdMap[dobj->dumpId] = dobj;
+
+   /* mark catalogIdMap invalid, but don't rebuild it yet */
+   catalogIdMapValid = false;
 }
 
 /*
@@ -454,25 +465,74 @@ findObjectByDumpId(DumpId dumpId)
  *
  * Returns NULL for unknown ID
  *
- * NOTE:  should hash this, but just do linear search for now
+ * We use binary search in a sorted list that is built on first call.
+ * If AssignDumpId() and findObjectByCatalogId() calls were intermixed,
+ * the code would work, but possibly be very slow.  In the current usage
+ * pattern that does not happen, indeed we only need to build the list once.
  */
 DumpableObject *
 findObjectByCatalogId(CatalogId catalogId)
 {
-   DumpId      i;
+   DumpableObject **low;
+   DumpableObject **high;
 
-   for (i = 1; i < allocedDumpIds; i++)
+   if (!catalogIdMapValid)
    {
-       DumpableObject *dobj = dumpIdMap[i];
+       if (catalogIdMap)
+           free(catalogIdMap);
+       getDumpableObjects(&catalogIdMap, &numCatalogIds);
+       if (numCatalogIds > 1)
+           qsort((void *) catalogIdMap, numCatalogIds,
+                 sizeof(DumpableObject *), DOCatalogIdCompare);
+       catalogIdMapValid = true;
+   }
 
-       if (dobj &&
-           dobj->catId.tableoid == catalogId.tableoid &&
-           dobj->catId.oid == catalogId.oid)
-           return dobj;
+   /*
+    * We could use bsearch() here, but the notational cruft of calling
+    * bsearch is nearly as bad as doing it ourselves; and the generalized
+    * bsearch function is noticeably slower as well.
+    */
+   if (numCatalogIds <= 0)
+       return NULL;
+   low = catalogIdMap;
+   high = catalogIdMap + (numCatalogIds - 1);
+   while (low <= high)
+   {
+       DumpableObject **middle;
+       int         difference;
+
+       middle = low + (high - low) / 2;
+       /* comparison must match DOCatalogIdCompare, below */
+       difference = oidcmp((*middle)->catId.oid, catalogId.oid);
+       if (difference == 0)
+           difference = oidcmp((*middle)->catId.tableoid, catalogId.tableoid);
+       if (difference == 0)
+           return *middle;
+       else if (difference < 0)
+           low = middle + 1;
+       else
+           high = middle - 1;
    }
    return NULL;
 }
 
+static int
+DOCatalogIdCompare(const void *p1, const void *p2)
+{
+   DumpableObject *obj1 = *(DumpableObject **) p1;
+   DumpableObject *obj2 = *(DumpableObject **) p2;
+   int         cmpval;
+
+   /*
+    * Compare OID first since it's usually unique, whereas there will
+    * only be a few distinct values of tableoid.
+    */
+   cmpval = oidcmp(obj1->catId.oid, obj2->catId.oid);
+   if (cmpval == 0)
+       cmpval = oidcmp(obj1->catId.tableoid, obj2->catId.tableoid);
+   return cmpval;
+}
+
 /*
  * Build an array of pointers to all known dumpable objects
  *