Add index scan

This commit fixes performance degradation during insertion scenarios by replacing SeqScan with IndexScan.

Co-authored-by: Daria Barsukova <d.barsukova@g.nsu.ru>
Co-authored-by: Alexandra Bondar <s6311704@gmail.com>

Author: 3 people

src/backend/catalog/ag_label.c

@@ -21,6 +21,7 @@
 
 #include "access/genam.h"
 #include "catalog/indexing.h"
+#include "catalog/namespace.h"
 #include "executor/executor.h"
 #include "nodes/makefuncs.h"
 #include "utils/builtins.h"
@@ -297,46 +298,104 @@ List *get_all_edge_labels_per_graph(EState *estate, Oid graph_oid)
     HeapTuple tuple;
     TupleTableSlot *slot;
     ResultRelInfo *resultRelInfo;
+    Oid index_oid;
 
-    /* setup scan keys to get all edges for the given graph oid */
-    ScanKeyInit(&scan_keys[1], Anum_ag_label_graph, BTEqualStrategyNumber,
-                F_OIDEQ, ObjectIdGetDatum(graph_oid));
-    ScanKeyInit(&scan_keys[0], Anum_ag_label_kind, BTEqualStrategyNumber,
-                F_CHAREQ, CharGetDatum(LABEL_TYPE_EDGE));
+    index_oid = get_relname_relid("ag_label_graph_oid_index", 
+                                  get_namespace_oid("ag_catalog", false));
 
     /* setup the table to be scanned */
     ag_label = table_open(ag_label_relation_id(), RowExclusiveLock);
-    scan_desc = table_beginscan(ag_label, estate->es_snapshot, 2, scan_keys);
 
     resultRelInfo = create_entity_result_rel_info(estate, "ag_catalog",
                                                   "ag_label");
 
-    slot = ExecInitExtraTupleSlot(
-        estate, RelationGetDescr(resultRelInfo->ri_RelationDesc),
-        &TTSOpsHeapTuple);
-
-    /* scan through the results and get all the label names. */
-    while(true)
+    if (OidIsValid(index_oid))
+    {
+        Relation index_rel;
+        IndexScanDesc index_scan_desc;
+
+        slot = ExecInitExtraTupleSlot(
+            estate, RelationGetDescr(resultRelInfo->ri_RelationDesc),
+            &TTSOpsBufferHeapTuple);
+
+        index_rel = index_open(index_oid, RowExclusiveLock);
+
+        ScanKeyInit(&scan_keys[0], Anum_ag_label_name, BTEqualStrategyNumber,
+                    F_OIDEQ, ObjectIdGetDatum(graph_oid));
+
+        index_scan_desc = index_beginscan(ag_label, index_rel, estate->es_snapshot, NULL, 1, 0);
+        index_rescan(index_scan_desc, scan_keys, 1, NULL, 0);
+
+        while (index_getnext_slot(index_scan_desc, ForwardScanDirection, slot))
+        {
+            Name label;
+            Name lval;
+            bool isNull;
+            Datum datum;
+            char kind;
+
+            /*There isn't field kind in index. So we should check it by hands*/
+            datum = slot_getattr(slot, Anum_ag_label_kind, &isNull);
+            if (isNull)
+                continue;
+
+            kind = DatumGetChar(datum);
+            
+            if (kind != LABEL_TYPE_EDGE)
+                continue;
+
+            datum = slot_getattr(slot, Anum_ag_label_name, &isNull);
+            if (!isNull)
+            {
+                label = DatumGetName(datum);
+                lval = (Name) palloc(NAMEDATALEN);
+                namestrcpy(lval, NameStr(*label));
+                labels = lappend(labels, lval);
+            }
+        }
+
+        index_endscan(index_scan_desc);
+        index_close(index_rel, RowExclusiveLock);
+    } else
     {
-        Name label;
-        bool isNull;
-        Datum datum;
+        slot = ExecInitExtraTupleSlot(
+            estate, RelationGetDescr(resultRelInfo->ri_RelationDesc),
+            &TTSOpsHeapTuple);
 
-        tuple = heap_getnext(scan_desc, ForwardScanDirection);
+        // setup scan keys to get all edges for the given graph oid
+        ScanKeyInit(&scan_keys[1], Anum_ag_label_graph, BTEqualStrategyNumber,
+                  F_OIDEQ, ObjectIdGetDatum(graph_oid));
+        ScanKeyInit(&scan_keys[0], Anum_ag_label_kind, BTEqualStrategyNumber,
+                  F_CHAREQ, CharGetDatum(LABEL_TYPE_EDGE));
 
-        /* no more labels to process */
-        if (!HeapTupleIsValid(tuple))
-            break;
+        scan_desc = table_beginscan(ag_label, estate->es_snapshot, 2, scan_keys);
 
-        ExecStoreHeapTuple(tuple, slot, false);
+        // scan through the results and get all the label names.
+        while(true)
+        {
+            Name label;
+            Name lval;
+            bool isNull;
+            Datum datum;
 
-        datum = slot_getattr(slot, Anum_ag_label_name, &isNull);
-        label = DatumGetName(datum);
+            tuple = heap_getnext(scan_desc, ForwardScanDirection);
 
-        labels = lappend(labels, label);
-    }
+            // no more labels to process
+            if (!HeapTupleIsValid(tuple))
+                break;
+
+            ExecStoreHeapTuple(tuple, slot, false);
 
-    table_endscan(scan_desc);
+            datum = slot_getattr(slot, Anum_ag_label_name, &isNull);
+            label = DatumGetName(datum);
+
+            lval = (Name) palloc(NAMEDATALEN);
+            namestrcpy(lval, NameStr(*label));
+            labels = lappend(labels, lval);
+        }
+
+        table_endscan(scan_desc);
+    }
 
     destroy_entity_result_rel_info(resultRelInfo);
     table_close(resultRelInfo->ri_RelationDesc, RowExclusiveLock);