v16-0005-use-LIKE-not-substring.patch

diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index d13cfc3adf..15205a8e08 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -896,10 +896,9 @@ static const SchemaQuery Query_for_trigger_of_table = {
  * name; make a SchemaQuery instead.
  *
  * In these queries, there must be a restriction clause of the form
- *		substring(OUTPUT,1,%d)='%s'
- * where "OUTPUT" is the same string that the query returns.  The %d and %s
- * will be replaced by the string length of the text and the text itself,
- * causing the results to be limited to those matching the already-typed text.
+ *		output LIKE '%s'
+ * where "output" is the same string that the query returns.  The %s
+ * will be replaced by a LIKE pattern to match the already-typed text.
  *
  * There can be a second '%s', which will be replaced by a suitably-escaped
  * version of the string provided in completion_info_object.  If there is a
@@ -912,56 +911,56 @@ static const SchemaQuery Query_for_trigger_of_table = {
 #define Query_for_list_of_template_databases \
 "SELECT d.datname "\
 "  FROM pg_catalog.pg_database d "\
-" WHERE substring(d.datname,1,%d)='%s' "\
+" WHERE d.datname LIKE '%s' "\
 "   AND (d.datistemplate OR pg_catalog.pg_has_role(d.datdba, 'USAGE'))"
 
 #define Query_for_list_of_databases \
 "SELECT datname FROM pg_catalog.pg_database "\
-" WHERE substring(datname,1,%d)='%s'"
+" WHERE datname LIKE '%s'"
 
 #define Query_for_list_of_tablespaces \
 "SELECT spcname FROM pg_catalog.pg_tablespace "\
-" WHERE substring(spcname,1,%d)='%s'"
+" WHERE spcname LIKE '%s'"
 
 #define Query_for_list_of_encodings \
 " SELECT DISTINCT pg_catalog.pg_encoding_to_char(conforencoding) "\
 "   FROM pg_catalog.pg_conversion "\
-"  WHERE substring(pg_catalog.pg_encoding_to_char(conforencoding),1,%d)=UPPER('%s')"
+"  WHERE pg_catalog.pg_encoding_to_char(conforencoding) LIKE UPPER('%s')"
 
 #define Query_for_list_of_languages \
 "SELECT lanname "\
 "  FROM pg_catalog.pg_language "\
 " WHERE lanname != 'internal' "\
-"   AND substring(lanname,1,%d)='%s'"
+"   AND lanname LIKE '%s'"
 
 #define Query_for_list_of_schemas \
 "SELECT nspname FROM pg_catalog.pg_namespace "\
-" WHERE substring(nspname,1,%d)='%s'"
+" WHERE nspname LIKE '%s'"
 
 #define Query_for_list_of_alter_system_set_vars \
 "SELECT name FROM "\
 " (SELECT pg_catalog.lower(name) AS name FROM pg_catalog.pg_settings "\
 "  WHERE context != 'internal' "\
 " ) ss "\
-" WHERE substring(name,1,%d)='%s'"
+" WHERE name LIKE '%s'"
 
 #define Query_for_list_of_set_vars \
 "SELECT name FROM "\
 " (SELECT pg_catalog.lower(name) AS name FROM pg_catalog.pg_settings "\
 "  WHERE context IN ('user', 'superuser') "\
 " ) ss "\
-" WHERE substring(name,1,%d)='%s'"
+" WHERE name LIKE '%s'"
 
 #define Query_for_list_of_show_vars \
 "SELECT name FROM "\
 " (SELECT pg_catalog.lower(name) AS name FROM pg_catalog.pg_settings "\
 " ) ss "\
-" WHERE substring(name,1,%d)='%s'"
+" WHERE name LIKE '%s'"
 
 #define Query_for_list_of_roles \
 " SELECT rolname "\
 "   FROM pg_catalog.pg_roles "\
-"  WHERE substring(rolname,1,%d)='%s'"
+"  WHERE rolname LIKE '%s'"
 
 /* add these to Query_for_list_of_roles in GRANT contexts */
 #define Keywords_for_list_of_grant_roles \
@@ -971,119 +970,119 @@ static const SchemaQuery Query_for_trigger_of_table = {
 "SELECT conname "\
 "  FROM pg_catalog.pg_constraint c "\
 " WHERE c.conrelid <> 0 "\
-"       and substring(conname,1,%d)='%s'"
+"       and conname LIKE '%s'"
 
 #define Query_for_list_of_ts_configurations \
 "SELECT cfgname FROM pg_catalog.pg_ts_config "\
-" WHERE substring(cfgname,1,%d)='%s'"
+" WHERE cfgname LIKE '%s'"
 
 #define Query_for_list_of_ts_dictionaries \
 "SELECT dictname FROM pg_catalog.pg_ts_dict "\
-" WHERE substring(dictname,1,%d)='%s'"
+" WHERE dictname LIKE '%s'"
 
 #define Query_for_list_of_ts_parsers \
 "SELECT prsname FROM pg_catalog.pg_ts_parser "\
-" WHERE substring(prsname,1,%d)='%s'"
+" WHERE prsname LIKE '%s'"
 
 #define Query_for_list_of_ts_templates \
 "SELECT tmplname FROM pg_catalog.pg_ts_template "\
-" WHERE substring(tmplname,1,%d)='%s'"
+" WHERE tmplname LIKE '%s'"
 
 #define Query_for_list_of_fdws \
 " SELECT fdwname "\
 "   FROM pg_catalog.pg_foreign_data_wrapper "\
-"  WHERE substring(fdwname,1,%d)='%s'"
+"  WHERE fdwname LIKE '%s'"
 
 #define Query_for_list_of_servers \
 " SELECT srvname "\
 "   FROM pg_catalog.pg_foreign_server "\
-"  WHERE substring(srvname,1,%d)='%s'"
+"  WHERE srvname LIKE '%s'"
 
 #define Query_for_list_of_user_mappings \
 " SELECT usename "\
 "   FROM pg_catalog.pg_user_mappings "\
-"  WHERE substring(usename,1,%d)='%s'"
+"  WHERE usename LIKE '%s'"
 
 #define Query_for_list_of_access_methods \
 " SELECT amname "\
 "   FROM pg_catalog.pg_am "\
-"  WHERE substring(amname,1,%d)='%s'"
+"  WHERE amname LIKE '%s'"
 
 #define Query_for_list_of_index_access_methods \
 " SELECT amname "\
 "   FROM pg_catalog.pg_am "\
-"  WHERE substring(amname,1,%d)='%s' AND "\
+"  WHERE amname LIKE '%s' AND "\
 "   amtype=" CppAsString2(AMTYPE_INDEX)
 
 #define Query_for_list_of_table_access_methods \
 " SELECT amname "\
 "   FROM pg_catalog.pg_am "\
-"  WHERE substring(amname,1,%d)='%s' AND "\
+"  WHERE amname LIKE '%s' AND "\
 "   amtype=" CppAsString2(AMTYPE_TABLE)
 
 #define Query_for_list_of_extensions \
 " SELECT extname "\
 "   FROM pg_catalog.pg_extension "\
-"  WHERE substring(extname,1,%d)='%s'"
+"  WHERE extname LIKE '%s'"
 
 #define Query_for_list_of_available_extensions \
 " SELECT name "\
 "   FROM pg_catalog.pg_available_extensions "\
-"  WHERE substring(name,1,%d)='%s' AND installed_version IS NULL"
+"  WHERE name LIKE '%s' AND installed_version IS NULL"
 
 /* the result of this query is not an identifier, so use VERBATIM */
 #define Query_for_list_of_available_extension_versions \
 " SELECT version "\
 "   FROM pg_catalog.pg_available_extension_versions "\
-"  WHERE substring(version,1,%d)='%s'"\
+"  WHERE version LIKE '%s'"\
 "    AND name='%s'"
 
 /* the result of this query is not an identifier, so use VERBATIM */
 #define Query_for_list_of_available_extension_versions_with_TO \
 " SELECT 'TO ' || version "\
 "   FROM pg_catalog.pg_available_extension_versions "\
-"  WHERE substring('TO ' || version,1,%d)='%s'"\
+"  WHERE ('TO ' || version) LIKE '%s'"\
 "    AND name='%s'"
 
 #define Query_for_list_of_prepared_statements \
 " SELECT name "\
 "   FROM pg_catalog.pg_prepared_statements "\
-"  WHERE substring(name,1,%d)='%s'"
+"  WHERE name LIKE '%s'"
 
 #define Query_for_list_of_event_triggers \
 " SELECT evtname "\
 "   FROM pg_catalog.pg_event_trigger "\
-"  WHERE substring(evtname,1,%d)='%s'"
+"  WHERE evtname LIKE '%s'"
 
 #define Query_for_list_of_tablesample_methods \
 " SELECT proname "\
 "   FROM pg_catalog.pg_proc "\
 "  WHERE prorettype = 'pg_catalog.tsm_handler'::pg_catalog.regtype AND "\
 "        proargtypes[0] = 'pg_catalog.internal'::pg_catalog.regtype AND "\
-"        substring(proname,1,%d)='%s'"
+"        proname LIKE '%s'"
 
 #define Query_for_list_of_policies \
 " SELECT polname "\
 "   FROM pg_catalog.pg_policy "\
-"  WHERE substring(polname,1,%d)='%s'"
+"  WHERE polname LIKE '%s'"
 
 #define Query_for_values_of_enum_GUC \
 " SELECT val FROM ( "\
 "   SELECT name, pg_catalog.unnest(enumvals) AS val "\
 "     FROM pg_catalog.pg_settings "\
 "    ) ss "\
-"  WHERE substring(val,1,%d)='%s'"\
+"  WHERE val LIKE '%s'"\
 "        and pg_catalog.lower(name)=pg_catalog.lower('%s')"
 
 #define Query_for_list_of_channels \
 " SELECT channel "\
 "   FROM pg_catalog.pg_listening_channels() AS channel "\
-"  WHERE substring(channel,1,%d)='%s'"
+"  WHERE channel LIKE '%s'"
 
 #define Query_for_list_of_cursors \
 " SELECT name "\
 "   FROM pg_catalog.pg_cursors "\
-"  WHERE substring(name,1,%d)='%s'"
+"  WHERE name LIKE '%s'"
 
 /*
  * These object types were introduced later than our support cutoff of
@@ -1095,7 +1094,7 @@ static const VersionedQuery Query_for_list_of_publications[] = {
 	{100000,
 		" SELECT pubname "
 		"   FROM pg_catalog.pg_publication "
-		"  WHERE substring(pubname,1,%d)='%s'"
+		"  WHERE pubname LIKE '%s'"
 	},
 	{0, NULL}
 };
@@ -1104,7 +1103,7 @@ static const VersionedQuery Query_for_list_of_subscriptions[] = {
 	{100000,
 		" SELECT s.subname "
 		"   FROM pg_catalog.pg_subscription s, pg_catalog.pg_database d "
-		"  WHERE substring(s.subname,1,%d)='%s' "
+		"  WHERE s.subname LIKE '%s' "
 		"    AND d.datname = pg_catalog.current_database() "
 		"    AND s.subdbid = d.oid"
 	},
@@ -1150,7 +1149,7 @@ static const pgsql_thing_t words_after_create[] = {
 	 * to be used only by pg_dump.
 	 */
 	{"CONFIGURATION", Query_for_list_of_ts_configurations, NULL, NULL, NULL, THING_NO_SHOW},
-	{"CONVERSION", "SELECT conname FROM pg_catalog.pg_conversion WHERE substring(conname,1,%d)='%s'"},
+	{"CONVERSION", "SELECT conname FROM pg_catalog.pg_conversion WHERE conname LIKE '%s'"},
 	{"DATABASE", Query_for_list_of_databases},
 	{"DEFAULT PRIVILEGES", NULL, NULL, NULL, NULL, THING_NO_CREATE | THING_NO_DROP},
 	{"DICTIONARY", Query_for_list_of_ts_dictionaries, NULL, NULL, NULL, THING_NO_SHOW},
@@ -1175,7 +1174,7 @@ static const pgsql_thing_t words_after_create[] = {
 	{"PUBLICATION", NULL, Query_for_list_of_publications},
 	{"ROLE", Query_for_list_of_roles},
 	{"ROUTINE", NULL, NULL, &Query_for_list_of_routines, NULL, THING_NO_CREATE},
-	{"RULE", "SELECT rulename FROM pg_catalog.pg_rules WHERE substring(rulename,1,%d)='%s'"},
+	{"RULE", "SELECT rulename FROM pg_catalog.pg_rules WHERE rulename LIKE '%s'"},
 	{"SCHEMA", Query_for_list_of_schemas},
 	{"SEQUENCE", NULL, NULL, &Query_for_list_of_sequences},
 	{"SERVER", Query_for_list_of_servers},
@@ -1191,7 +1190,7 @@ static const pgsql_thing_t words_after_create[] = {
 																			 * TABLE ... */
 	{"TEXT SEARCH", NULL, NULL, NULL},
 	{"TRANSFORM", NULL, NULL, NULL, NULL, THING_NO_ALTER},
-	{"TRIGGER", "SELECT tgname FROM pg_catalog.pg_trigger WHERE substring(tgname,1,%d)='%s' AND NOT tgisinternal"},
+	{"TRIGGER", "SELECT tgname FROM pg_catalog.pg_trigger WHERE tgname LIKE '%s' AND NOT tgisinternal"},
 	{"TYPE", NULL, NULL, &Query_for_list_of_datatypes},
 	{"UNIQUE", NULL, NULL, NULL, NULL, THING_NO_DROP | THING_NO_ALTER}, /* for CREATE UNIQUE
 																		 * INDEX ... */
@@ -1273,6 +1272,7 @@ static char *complete_from_files(const char *text, int state);
 
 static char *pg_strdup_keyword_case(const char *s, const char *ref);
 static char *escape_string(const char *text);
+static char *make_like_pattern(const char *word);
 static void parse_identifier(const char *ident,
 							 char **schemaname, char **objectname,
 							 bool *schemaquoted, bool *objectquoted);
@@ -1775,9 +1775,7 @@ psql_completion(const char *text, int start, int end)
 		COMPLETE_WITH("(", "ALL TABLES IN SCHEMA", "TABLE");
 	else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|DROP|SET", "ALL", "TABLES", "IN", "SCHEMA"))
 		COMPLETE_WITH_QUERY_PLUS(Query_for_list_of_schemas
-								 " AND nspname != 'pg_catalog' "
-								 " AND nspname not like 'pg\\_toast%%' "
-								 " AND nspname not like 'pg\\_temp%%' ",
+								 " AND nspname NOT LIKE E'pg\\\\_%'",
 								 "CURRENT_SCHEMA");
 	/* ALTER PUBLICATION <name> SET ( */
 	else if (HeadMatches("ALTER", "PUBLICATION", MatchAny) && TailMatches("SET", "("))
@@ -2911,9 +2909,7 @@ psql_completion(const char *text, int start, int end)
 	 */
 	else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL", "TABLES", "IN", "SCHEMA"))
 		COMPLETE_WITH_QUERY_PLUS(Query_for_list_of_schemas
-								 " AND nspname != 'pg_catalog' "
-								 " AND nspname not like 'pg\\_toast%%' "
-								 " AND nspname not like 'pg\\_temp%%' ",
+								 " AND nspname NOT LIKE E'pg\\\\_%'",
 								 "CURRENT_SCHEMA");
 	else if (Matches("CREATE", "PUBLICATION", MatchAny, "FOR", "ALL", "TABLES", "IN", "SCHEMA", MatchAny) && (!ends_with(prev_wd, ',')))
 		COMPLETE_WITH("WITH (");
@@ -4119,10 +4115,13 @@ psql_completion(const char *text, int start, int end)
 						  "US", "European", "NonEuropean",
 						  "DEFAULT");
 		else if (TailMatches("search_path", "TO|="))
+		{
+			/* Here, we want to allow pg_catalog, so use narrower exclusion */
 			COMPLETE_WITH_QUERY_PLUS(Query_for_list_of_schemas
-									 " AND nspname not like 'pg\\_toast%%' "
-									 " AND nspname not like 'pg\\_temp%%' ",
+									 " AND nspname NOT LIKE E'pg\\\\_toast%%'"
+									 " AND nspname NOT LIKE E'pg\\\\_temp%%'",
 									 "DEFAULT");
+		}
 		else
 		{
 			/* generic, type based, GUC support */
@@ -4689,10 +4688,9 @@ complete_from_versioned_schema_query(const char *text, int state)
  * The query can be one of two kinds:
  *
  * 1. A simple query, which must contain a restriction clause of the form
- *		substring(OUTPUT,1,%d)='%s'
- * where "OUTPUT" is the same string that the query returns.  The %d and %s
- * will be replaced by the string length of the text and the text itself,
- * causing the results to be limited to those matching the already-typed text.
+ *		output LIKE '%s'
+ * where "output" is the same string that the query returns.  The %s
+ * will be replaced by a LIKE pattern to match the already-typed text.
  * There can be a second '%s', which will be replaced by a suitably-escaped
  * version of the string provided in completion_info_object.  If there is a
  * third '%s', it will be replaced by a suitably-escaped version of the string
@@ -4747,9 +4745,8 @@ _complete_from_query(const char *simple_query,
 		PQExpBufferData query_buffer;
 		char	   *schemaname;
 		char	   *objectname;
-		int			object_length = 0;
+		char	   *e_object_like;
 		char	   *e_schemaname;
-		char	   *e_objectname;
 		char	   *e_info_object;
 		char	   *e_info_schema;
 
@@ -4777,22 +4774,16 @@ _complete_from_query(const char *simple_query,
 		non_empty_object = (*objectname != '\0');
 
 		/*
-		 * Count length as number of characters (not bytes), for passing to
-		 * substring
+		 * Convert objectname to a LIKE prefix pattern (e.g. 'foo%'), and set
+		 * up suitably-escaped copies of all the strings we need.
 		 */
-		for (const char *p = objectname;
-			 *p;
-			 p += PQmblenBounded(p, pset.encoding))
-			object_length++;
+		e_object_like = make_like_pattern(objectname);
 
-		/* Set up suitably-escaped copies of textual inputs */
 		if (schemaname)
 			e_schemaname = escape_string(schemaname);
 		else
 			e_schemaname = NULL;
 
-		e_objectname = escape_string(objectname);
-
 		if (completion_info_object)
 			e_info_object = escape_string(completion_info_object);
 		else
@@ -4831,9 +4822,9 @@ _complete_from_query(const char *simple_query,
 				if (schema_query->selcondition)
 					appendPQExpBuffer(&query_buffer, "%s AND ",
 									  schema_query->selcondition);
-				appendPQExpBuffer(&query_buffer, "substring(%s,1,%d)='%s'",
+				appendPQExpBuffer(&query_buffer, "(%s) LIKE '%s'",
 								  schema_query->result,
-								  object_length, e_objectname);
+								  e_object_like);
 				if (schema_query->viscondition)
 					appendPQExpBuffer(&query_buffer, " AND %s",
 									  schema_query->viscondition);
@@ -4878,8 +4869,8 @@ _complete_from_query(const char *simple_query,
 					appendPQExpBuffer(&query_buffer, "\nUNION ALL\n"
 									  "SELECT NULL::pg_catalog.text, n.nspname "
 									  "FROM pg_catalog.pg_namespace n "
-									  "WHERE substring(n.nspname,1,%d)='%s'",
-									  object_length, e_objectname);
+									  "WHERE n.nspname LIKE '%s'",
+									  e_object_like);
 
 					/*
 					 * Likewise, suppress system schemas unless the
@@ -4887,7 +4878,7 @@ _complete_from_query(const char *simple_query,
 					 */
 					if (strncmp(objectname, "pg_", 3) != 0)
 						appendPQExpBufferStr(&query_buffer,
-											 " AND n.nspname NOT LIKE 'pg\\_%'");
+											 " AND n.nspname NOT LIKE E'pg\\\\_%'");
 
 					/*
 					 * Since we're matching these schema names to the object
@@ -4915,9 +4906,9 @@ _complete_from_query(const char *simple_query,
 				if (schema_query->selcondition)
 					appendPQExpBuffer(&query_buffer, "%s AND ",
 									  schema_query->selcondition);
-				appendPQExpBuffer(&query_buffer, "substring(%s,1,%d)='%s' AND ",
+				appendPQExpBuffer(&query_buffer, "(%s) LIKE '%s' AND ",
 								  schema_query->result,
-								  object_length, e_objectname);
+								  e_object_like);
 				appendPQExpBuffer(&query_buffer, "n.nspname = '%s'",
 								  e_schemaname);
 				if (schema_query->infoname)
@@ -4942,7 +4933,7 @@ _complete_from_query(const char *simple_query,
 			Assert(simple_query);
 			/* simple_query is an sprintf-style format string */
 			appendPQExpBuffer(&query_buffer, simple_query,
-							  object_length, e_objectname,
+							  e_object_like,
 							  e_info_object, e_info_schema);
 		}
 
@@ -4955,9 +4946,9 @@ _complete_from_query(const char *simple_query,
 
 		/* Clean up */
 		termPQExpBuffer(&query_buffer);
+		free(e_object_like);
 		if (e_schemaname)
 			free(e_schemaname);
-		free(e_objectname);
 		if (e_info_object)
 			free(e_info_object);
 		if (e_info_schema)
@@ -5414,6 +5405,48 @@ escape_string(const char *text)
 }
 
 
+/*
+ * make_like_pattern - Convert argument to a LIKE prefix pattern.
+ *
+ * We escape _ and % in the given text by backslashing, append a % to
+ * represent "any subsequent characters", and then pass the string through
+ * escape_string() so it's ready to insert in a query.  The result needs
+ * to be freed.
+ */
+static char *
+make_like_pattern(const char *word)
+{
+	char	   *result;
+	char	   *buffer = pg_malloc(strlen(word) * 2 + 2);
+	char	   *bptr = buffer;
+
+	while (*word)
+	{
+		if (*word == '_' || *word == '%')
+			*bptr++ = '\\';
+		if (IS_HIGHBIT_SET(*word))
+		{
+			/*
+			 * Transfer multibyte characters without further processing, to
+			 * avoid getting confused in unsafe client encodings.
+			 */
+			int			chlen = PQmblenBounded(word, pset.encoding);
+
+			while (chlen-- > 0)
+				*bptr++ = *word++;
+		}
+		else
+			*bptr++ = *word++;
+	}
+	*bptr++ = '%';
+	*bptr = '\0';
+
+	result = escape_string(buffer);
+	free(buffer);
+	return result;
+}
+
+
 /*
  * parse_identifier - Parse a possibly-schema-qualified SQL identifier.
  *