Use multi-inserts for pg_ts_config_map

Two locations working on pg_ts_config_map are switched from
CatalogTupleInsert() to a multi-insert approach with tuple slots:
- ALTER TEXT SEARCH CONFIGURATION ADD/ALTER MAPPING when inserting new
entries.  The number of entries to insert is known in advance, so is the
number of slots needed.  Note that CatalogTupleInsertWithInfo() is now
used for the entry updates.
- CREATE TEXT SEARCH CONFIGURATION, where up to ~20-ish records could be
inserted at once.  The number of slots is not known in advance, hence
a slot initialization is delayed until a tuple is stored in it.

Like all the changes of this kind (1ff4161, 63110c6 or e3931d01), an
insert batch is capped at 64kB.

Author: Michael Paquier, Ranier Vilela
Reviewed-by: Kyotaro Horiguchi
Discussion: https://postgr.es/m/Y3M5bovrkTQbAO4W@paquier.xyz

diff --git a/src/backend/commands/tsearchcmds.c b/src/backend/commands/tsearchcmds.c
index 9304c53d4baedbefaa03ed859c41b9e5bb54fcee..bf0bec2f7052ea58ed0e3b6e8c322ccc2b94331d 100644 (file)
--- a/src/backend/commands/tsearchcmds.c
+++ b/src/backend/commands/tsearchcmds.c
@@ -1004,8 +1004,24 @@ DefineTSConfiguration(List *names, List *parameters, ObjectAddress *copied)
                ScanKeyData skey;
                SysScanDesc scan;
                HeapTuple       maptup;
+               TupleDesc       mapDesc;
+               TupleTableSlot **slot;
+               CatalogIndexState indstate;
+               int                     max_slots,
+                                       slot_init_count,
+                                       slot_stored_count;
 
                mapRel = table_open(TSConfigMapRelationId, RowExclusiveLock);
+               mapDesc = RelationGetDescr(mapRel);
+
+               indstate = CatalogOpenIndexes(mapRel);
+
+               /*
+                * Allocate the slots to use, but delay costly initialization until we
+                * know that they will be used.
+                */
+               max_slots = MAX_CATALOG_MULTI_INSERT_BYTES / sizeof(FormData_pg_ts_config_map);
+               slot = palloc(sizeof(TupleTableSlot *) * max_slots);
 
                ScanKeyInit(&skey,
                                        Anum_pg_ts_config_map_mapcfg,
@@ -1015,29 +1031,54 @@ DefineTSConfiguration(List *names, List *parameters, ObjectAddress *copied)
                scan = systable_beginscan(mapRel, TSConfigMapIndexId, true,
                                                                  NULL, 1, &skey);
 
+               /* number of slots currently storing tuples */
+               slot_stored_count = 0;
+               /* number of slots currently initialized */
+               slot_init_count = 0;
+
                while (HeapTupleIsValid((maptup = systable_getnext(scan))))
                {
                        Form_pg_ts_config_map cfgmap = (Form_pg_ts_config_map) GETSTRUCT(maptup);
-                       HeapTuple       newmaptup;
-                       Datum           mapvalues[Natts_pg_ts_config_map];
-                       bool            mapnulls[Natts_pg_ts_config_map];
 
-                       memset(mapvalues, 0, sizeof(mapvalues));
-                       memset(mapnulls, false, sizeof(mapnulls));
+                       if (slot_init_count < max_slots)
+                       {
+                               slot[slot_stored_count] = MakeSingleTupleTableSlot(mapDesc,
+                                                                                                                                  &TTSOpsHeapTuple);
+                               slot_init_count++;
+                       }
+
+                       ExecClearTuple(slot[slot_stored_count]);
 
-                       mapvalues[Anum_pg_ts_config_map_mapcfg - 1] = cfgOid;
-                       mapvalues[Anum_pg_ts_config_map_maptokentype - 1] = cfgmap->maptokentype;
-                       mapvalues[Anum_pg_ts_config_map_mapseqno - 1] = cfgmap->mapseqno;
-                       mapvalues[Anum_pg_ts_config_map_mapdict - 1] = cfgmap->mapdict;
+                       memset(slot[slot_stored_count]->tts_isnull, false,
+                                  slot[slot_stored_count]->tts_tupleDescriptor->natts * sizeof(bool));
 
-                       newmaptup = heap_form_tuple(mapRel->rd_att, mapvalues, mapnulls);
+                       slot[slot_stored_count]->tts_values[Anum_pg_ts_config_map_mapcfg - 1] = cfgOid;
+                       slot[slot_stored_count]->tts_values[Anum_pg_ts_config_map_maptokentype - 1] = cfgmap->maptokentype;
+                       slot[slot_stored_count]->tts_values[Anum_pg_ts_config_map_mapseqno - 1] = cfgmap->mapseqno;
+                       slot[slot_stored_count]->tts_values[Anum_pg_ts_config_map_mapdict - 1] = cfgmap->mapdict;
 
-                       CatalogTupleInsert(mapRel, newmaptup);
+                       ExecStoreVirtualTuple(slot[slot_stored_count]);
+                       slot_stored_count++;
 
-                       heap_freetuple(newmaptup);
+                       /* If slots are full, insert a batch of tuples */
+                       if (slot_stored_count == max_slots)
+                       {
+                               CatalogTuplesMultiInsertWithInfo(mapRel, slot, slot_stored_count,
+                                                                                                indstate);
+                               slot_stored_count = 0;
+                       }
                }
 
+               /* Insert any tuples left in the buffer */
+               if (slot_stored_count > 0)
+                       CatalogTuplesMultiInsertWithInfo(mapRel, slot, slot_stored_count,
+                                                                                        indstate);
+
+               for (int i = 0; i < slot_init_count; i++)
+                       ExecDropSingleTupleTableSlot(slot[i]);
+
                systable_endscan(scan);
+               CatalogCloseIndexes(indstate);
        }
 
        address = makeConfigurationDependencies(tup, false, mapRel);
@@ -1225,6 +1266,7 @@ MakeConfigurationMapping(AlterTSConfigurationStmt *stmt,
        Oid                *dictIds;
        int                     ndict;
        ListCell   *c;
+       CatalogIndexState indstate;
 
        tsform = (Form_pg_ts_config) GETSTRUCT(tup);
        cfgId = tsform->oid;
@@ -1275,6 +1317,8 @@ MakeConfigurationMapping(AlterTSConfigurationStmt *stmt,
                i++;
        }
 
+       indstate = CatalogOpenIndexes(relMap);
+
        if (stmt->replace)
        {
                /*
@@ -1334,7 +1378,7 @@ MakeConfigurationMapping(AlterTSConfigurationStmt *stmt,
                                newtup = heap_modify_tuple(maptup,
                                                                                   RelationGetDescr(relMap),
                                                                                   repl_val, repl_null, repl_repl);
-                               CatalogTupleUpdate(relMap, &newtup->t_self, newtup);
+                               CatalogTupleUpdateWithInfo(relMap, &newtup->t_self, newtup, indstate);
                        }
                }
 
@@ -1342,6 +1386,18 @@ MakeConfigurationMapping(AlterTSConfigurationStmt *stmt,
        }
        else
        {
+               TupleTableSlot **slot;
+               int                     slotCount = 0;
+               int                     nslots;
+
+               /* Allocate the slots to use and initialize them */
+               nslots = Min(ntoken * ndict,
+                                        MAX_CATALOG_MULTI_INSERT_BYTES / sizeof(FormData_pg_ts_config_map));
+               slot = palloc(sizeof(TupleTableSlot *) * nslots);
+               for (i = 0; i < nslots; i++)
+                       slot[i] = MakeSingleTupleTableSlot(RelationGetDescr(relMap),
+                                                                                          &TTSOpsHeapTuple);
+
                /*
                 * Insertion of new entries
                 */
@@ -1349,23 +1405,41 @@ MakeConfigurationMapping(AlterTSConfigurationStmt *stmt,
                {
                        for (j = 0; j < ndict; j++)
                        {
-                               Datum           values[Natts_pg_ts_config_map];
-                               bool            nulls[Natts_pg_ts_config_map];
+                               ExecClearTuple(slot[slotCount]);
+
+                               memset(slot[slotCount]->tts_isnull, false,
+                                          slot[slotCount]->tts_tupleDescriptor->natts * sizeof(bool));
 
-                               memset(nulls, false, sizeof(nulls));
-                               values[Anum_pg_ts_config_map_mapcfg - 1] = ObjectIdGetDatum(cfgId);
-                               values[Anum_pg_ts_config_map_maptokentype - 1] = Int32GetDatum(tokens[i]);
-                               values[Anum_pg_ts_config_map_mapseqno - 1] = Int32GetDatum(j + 1);
-                               values[Anum_pg_ts_config_map_mapdict - 1] = ObjectIdGetDatum(dictIds[j]);
+                               slot[slotCount]->tts_values[Anum_pg_ts_config_map_mapcfg - 1] = ObjectIdGetDatum(cfgId);
+                               slot[slotCount]->tts_values[Anum_pg_ts_config_map_maptokentype - 1] = Int32GetDatum(tokens[i]);
+                               slot[slotCount]->tts_values[Anum_pg_ts_config_map_mapseqno - 1] = Int32GetDatum(j + 1);
+                               slot[slotCount]->tts_values[Anum_pg_ts_config_map_mapdict - 1] = ObjectIdGetDatum(dictIds[j]);
 
-                               tup = heap_form_tuple(relMap->rd_att, values, nulls);
-                               CatalogTupleInsert(relMap, tup);
+                               ExecStoreVirtualTuple(slot[slotCount]);
+                               slotCount++;
 
-                               heap_freetuple(tup);
+                               /* If slots are full, insert a batch of tuples */
+                               if (slotCount == nslots)
+                               {
+                                       CatalogTuplesMultiInsertWithInfo(relMap, slot, slotCount,
+                                                                                                        indstate);
+                                       slotCount = 0;
+                               }
                        }
                }
+
+               /* Insert any tuples left in the buffer */
+               if (slotCount > 0)
+                       CatalogTuplesMultiInsertWithInfo(relMap, slot, slotCount,
+                                                                                        indstate);
+
+               for (i = 0; i < nslots; i++)
+                       ExecDropSingleTupleTableSlot(slot[i]);
        }
 
+       /* clean up */
+       CatalogCloseIndexes(indstate);
+
        EventTriggerCollectAlterTSConfig(stmt, cfgId, dictIds, ndict);
 }