Add pglogical.batch_inserts option to control multi insert

pglogical.c

@@ -85,6 +85,7 @@ static const struct config_enum_entry server_message_level_options[] = {
 bool	pglogical_synchronous_commit = false;
 char   *pglogical_temp_directory;
 bool	pglogical_use_spi = false;
+bool	pglogical_batch_inserts = true;
 
 void _PG_init(void);
 void pglogical_supervisor_main(Datum main_arg);
@@ -748,6 +749,19 @@ _PG_init(void)
 							 0,
 							 NULL, NULL, NULL);
 
+	DefineCustomBoolVariable("pglogical.batch_inserts",
+							 "Batch inserts if possible",
+							 NULL,
+							 &pglogical_batch_inserts,
+#ifdef XCP
+							 true,
+#else
+							 false,
+#endif
+							 PGC_POSTMASTER,
+							 0,
+							 NULL, NULL, NULL);
+
 	/*
 	 * We can't use the temp_tablespace safely for our dumps, because Pg's
 	 * crash recovery is very careful to delete only particularly formatted

pglogical.h

@@ -39,6 +39,7 @@
 extern bool pglogical_synchronous_commit;
 extern char *pglogical_temp_directory;
 extern bool pglogical_use_spi;
+extern bool pglogical_batch_inserts;
 extern char *pglogical_extra_connection_options;
 
 extern char *shorten_hash(const char *str, int maxlen);

pglogical_apply.c

@@ -399,9 +399,10 @@ handle_insert(StringInfo s)
 			return;
 		}
 	}
-	else if (RelationGetRelid(rel->rel) != QueueRelid &&
-		apply_api.can_multi_insert &&
-		apply_api.can_multi_insert(rel))
+	else if (pglogical_batch_inserts &&
+			 RelationGetRelid(rel->rel) != QueueRelid &&
+			 apply_api.can_multi_insert &&
+			 apply_api.can_multi_insert(rel))
 	{
 		if (rel != last_insert_rel)
 		{