Support CREATE/DROP FULLTEXT INDEX for fulltext search (#1986)

Currently babelfish doesn't support fulltext index creation and drop for
fulltext search. This commit implements the logic of create and drop
fulltext index and also handles the unsupported features of the
CREATE/DROP FULLTEXT INDEX statements.

Task: BABEL-4383
Signed-off-by: Roshan Kanwar <[email protected]>

contrib/babelfishpg_tsql/sql/fts_contains_pgconfig.sql

@@ -10,10 +10,11 @@ DECLARE
   joined_text text;
   word text;
 BEGIN
-  -- Prefix term (Examples: '"word1*"', '"word1 word2*"') if 
-  -- (1) search term is surrounded by double quotes (Counter example: 'word1*', as it doesn't have double quotes)
-  -- (2) last word in the search term ends with a star (Counter example: '"word1* word2"', as last word doesn't end with star)
-  -- (3) last word is NOT a single star (Counter example: '"*"', '"word1 word2 *"', as last word is a single star)
+  /* 'Prefix term (Examples: '"word1*"', '"word1 word2*"') if 
+   * (1) search term is surrounded by double quotes (Counter example: 'word1*', as it doesn't have double quotes)
+   * (2) last word in the search term ends with a star (Counter example: '"word1* word2"', as last word doesn't end with star)
+   * (3) last word is NOT a single star (Counter example: '"*"', '"word1 word2 *"', as last word is a single star)
+  */
   IF (phrase COLLATE C) SIMILAR TO ('[ ]*"%\*"[ ]*' COLLATE C) AND (NOT (phrase COLLATE C) SIMILAR TO ('[ ]*"% \*"[ ]*' COLLATE C)) AND (NOT (phrase COLLATE C) SIMILAR TO ('[ ]*"\*"[ ]*' COLLATE C)) THEN
     RETURN 'simple'::regconfig;
   END IF;

contrib/babelfishpg_tsql/src/backend_parser/gram-tsql-epilogue.y.c

@@ -1869,7 +1869,7 @@ tsql_pivot_select_transformation(List *target_list, List *from_clause, List *piv
  * for DESC index, default should be NULLS LAST.
  */
 static void 
-tsql_index_nulls_order(List *indexParams)
+tsql_index_nulls_order(List *indexParams, const char *accessMethod)
 {
 	ListCell *lc;
 
@@ -1887,6 +1887,10 @@ tsql_index_nulls_order(List *indexParams)
 		if (indexElem->nulls_ordering != SORTBY_NULLS_DEFAULT)
 			continue;
 
+		/* GIN indexes don't support NULLS FIRST/LAST options */
+		if (strcmp(accessMethod, "gin") == 0)
+			return;
+
 		switch (indexElem->ordering)
 		{
 			case SORTBY_ASC:

contrib/babelfishpg_tsql/src/backend_parser/gram-tsql-prologue.y.h

@@ -88,4 +88,4 @@ static Node *tsql_update_output_into_cte_transformation(WithClause *opt_with_cla
 														List *from_clause, Node *where_or_current_clause, core_yyscan_t yyscanner);
 static List *get_transformed_output_list(List *tsql_output_clause);
 static bool returning_list_has_column_name(List *existing_colnames, char *current_colname);
-static void tsql_index_nulls_order(List *indexParams);
+static void tsql_index_nulls_order(List *indexParams, const char *accessMethod);

contrib/babelfishpg_tsql/src/backend_parser/gram-tsql-rule.y

@@ -3191,7 +3191,7 @@ tsql_IndexStmt:
 					n->transformed = false;
 					n->if_not_exists = false;
 
-					tsql_index_nulls_order(n->indexParams);
+					tsql_index_nulls_order(n->indexParams, n->accessMethod);
 					$$ = (Node *)n;
 				}
 		;

contrib/babelfishpg_tsql/src/codegen.c

@@ -302,6 +302,7 @@ stmt_default_act(Walker_context *ctx, PLtsql_stmt *stmt)
 		case PLTSQL_STMT_GRANTDB:
 		case PLTSQL_STMT_CHANGE_DBOWNER:
 		case PLTSQL_STMT_GRANTSCHEMA:
+		case PLTSQL_STMT_FULLTEXTINDEX:
 		case PLTSQL_STMT_INSERT_BULK:
 		case PLTSQL_STMT_DBCC:
 		case PLTSQL_STMT_SET_EXPLAIN_MODE:

contrib/babelfishpg_tsql/src/fts.c

@@ -3,6 +3,7 @@
 #include "utils/builtins.h"
 #include "utils/memutils.h"
 #include "fts_data.h"
+#include "guc.h"
 
 PG_FUNCTION_INFO_V1(babelfish_fts_rewrite);
 
@@ -13,6 +14,13 @@ babelfish_fts_rewrite(PG_FUNCTION_ARGS)
     char* input_str = text_to_cstring(input_text);
     char* translated_query;
     text* result_text = NULL; // Initialize result_text to NULL
+
+    if (!pltsql_allow_fulltext_parser)
+    {
+        ereport(ERROR,
+                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+                errmsg("Full Text Search is not yet supported.")));
+    }
 
     PG_TRY();
     {

contrib/babelfishpg_tsql/src/fts_parser.y

@@ -74,13 +74,13 @@ simple_term:
 
 prefix_term:
     PREFIX_TERM_TOKEN {
-        fts_yyerror(NULL, "Prefix term not supported");
+        fts_yyerror(NULL, "Prefix term is not currently supported in Babelfish");
     }
     ;
 
 generation_term:
     FORMSOF_TOKEN O_PAREN_TOKEN generation_type COMMA_TOKEN simple_term_list C_PAREN_TOKEN {
-        fts_yyerror(NULL, "Generation term not supported");
+        fts_yyerror(NULL, "Generation term is not currently supported in Babelfish");
     }
     ;
 

contrib/babelfishpg_tsql/src/fts_scan.l

@@ -21,9 +21,9 @@ static YY_BUFFER_STATE scanbufhandle;
 
 %%
 
-"AND"           { yylval = "AND"; return AND_TOKEN; }
-"AND NOT"       { yylval = "AND NOT"; return AND_NOT_TOKEN; }
-"OR"            { yylval = "OR"; return OR_TOKEN; }
+"AND"           { yylval = "AND"; yyerror(NULL, "Boolean operator is not currently supported in Babelfish"); return AND_TOKEN; }
+"AND NOT"       { yylval = "AND NOT"; yyerror(NULL, "Boolean operator is not currently supported in Babelfish"); return AND_NOT_TOKEN; }
+"OR"            { yylval = "OR"; yyerror(NULL, "Boolean operator is not currently supported in Babelfish"); return OR_TOKEN; }
 "INFLECTIONAL"  { yylval = "INFLECTIONAL"; return INFLECTIONAL_TOKEN; }
 "THESAURUS"     { yylval = "THESAURUS"; return THESAURUS_TOKEN; }
 "FORMSOF"       { yylval = "FORMSOF"; return FORMSOF_TOKEN; }

contrib/babelfishpg_tsql/src/guc.c

@@ -46,6 +46,7 @@ char	   *pltsql_psql_logical_babelfish_db_name = NULL;
 int			pltsql_lock_timeout = -1;
 bool		pltsql_enable_linked_servers = true;
 bool		pltsql_allow_windows_login = true;
+bool		pltsql_allow_fulltext_parser = false;
 
 bool		pltsql_xact_abort = false;
 bool		pltsql_implicit_transactions = false;
@@ -677,6 +678,16 @@ define_custom_variables(void)
 							 GUC_NO_SHOW_ALL | GUC_NOT_IN_SAMPLE | GUC_DISALLOW_IN_AUTO_FILE,
 							 NULL, NULL, NULL);
 
+	/* GUC for enabling or disabling full text search features */
+	DefineCustomBoolVariable("babelfishpg_tsql.allow_fulltext_parser",
+							 gettext_noop("GUC for enabling or disabling full text search features"),
+							 NULL,
+							 &pltsql_allow_fulltext_parser,
+							 false,
+							 PGC_SUSET,
+							 GUC_NO_SHOW_ALL | GUC_NOT_IN_SAMPLE | GUC_SUPERUSER_ONLY,
+							 NULL, NULL, NULL);
+
 	/* ISO standard settings */
 	DefineCustomBoolVariable("babelfishpg_tsql.ansi_defaults",
 							 gettext_noop("Controls a group of settings that collectively specify some "
@@ -1339,7 +1350,7 @@ define_escape_hatch_variables(void)
 
 	/* fulltext */
 	DefineCustomEnumVariable("babelfishpg_tsql.escape_hatch_fulltext",
-							 gettext_noop("escape hatch for fulltext"),
+							 gettext_noop("escape hatch for fulltext search"),
 							 NULL,
 							 &escape_hatch_fulltext,
 							 EH_STRICT,

contrib/babelfishpg_tsql/src/guc.h

@@ -19,6 +19,7 @@ extern bool pltsql_fmtonly;
 extern bool pltsql_enable_create_alter_view_from_pg;
 extern bool pltsql_enable_linked_servers;
 extern bool pltsql_allow_windows_login;
+extern bool pltsql_allow_fulltext_parser;
 extern char *pltsql_psql_logical_babelfish_db_name;
 extern int  pltsql_isolation_level_repeatable_read;
 extern int  pltsql_isolation_level_serializable;

contrib/babelfishpg_tsql/src/iterative_exec.c

@@ -818,6 +818,15 @@ dispatch_stmt(PLtsql_execstate *estate, PLtsql_stmt *stmt)
 			}
 			exec_stmt_grantschema(estate, (PLtsql_stmt_grantschema *) stmt);
 			break;
+		case PLTSQL_STMT_FULLTEXTINDEX:
+			if (pltsql_explain_only)
+			{
+				ereport(ERROR,
+						(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+						 errmsg("Showing Estimated Execution Plan for FULLTEXT INDEX statment is not yet supported")));
+			}
+			exec_stmt_fulltextindex(estate, (PLtsql_stmt_fulltextindex *) stmt);
+			break;
 		case PLTSQL_STMT_INSERT_BULK:
 			if (pltsql_explain_only)
 			{

contrib/babelfishpg_tsql/src/pl_exec-2.c

@@ -52,6 +52,7 @@ static int	exec_stmt_usedb(PLtsql_execstate *estate, PLtsql_stmt_usedb *stmt);
 static int	exec_stmt_usedb_explain(PLtsql_execstate *estate, PLtsql_stmt_usedb *stmt, bool shouldRestoreDb);
 static int	exec_stmt_grantdb(PLtsql_execstate *estate, PLtsql_stmt_grantdb *stmt);
 static int	exec_stmt_grantschema(PLtsql_execstate *estate, PLtsql_stmt_grantschema *stmt);
+static int	exec_stmt_fulltextindex(PLtsql_execstate *estate, PLtsql_stmt_fulltextindex *stmt);
 static int	exec_stmt_insert_execute_select(PLtsql_execstate *estate, PLtsql_expr *expr);
 static int	exec_stmt_insert_bulk(PLtsql_execstate *estate, PLtsql_stmt_insert_bulk *expr);
 static int	exec_stmt_dbcc(PLtsql_execstate *estate, PLtsql_stmt_dbcc *stmt);
@@ -87,6 +88,8 @@ extern void enable_sp_cursor_find_param_hook(void);
 extern void disable_sp_cursor_find_param_hook(void);
 extern void add_sp_cursor_param(char *name);
 extern void reset_sp_cursor_params();
+extern char *construct_unique_index_name(char *index_name, char *relation_name);
+extern const char *gen_schema_name_for_fulltext_index(const char *schema_name);
 
 extern void pltsql_commit_not_required_impl_txn(PLtsql_execstate *estate);
 
@@ -3822,3 +3825,101 @@ exec_stmt_change_dbowner(PLtsql_execstate *estate, PLtsql_stmt_change_dbowner *s
 	update_db_owner(stmt->new_owner_name, stmt->db_name);	
 	return PLTSQL_RC_OK;
 }
+
+static int
+exec_stmt_fulltextindex(PLtsql_execstate *estate, PLtsql_stmt_fulltextindex *stmt)
+{
+	char		*table_name;
+	char		*ft_index_name;
+	char		*query_str;
+	char		*old_ft_index_name;	// existing fulltext index name
+	char		*uniq_index_name;
+	const char	*schema_name;
+	Oid			schemaOid;
+	Oid			relid;
+	List		*column_name;
+	char	    *dbname = get_cur_db_name();
+	char		*login = GetUserNameFromId(GetSessionUserId(), false);
+	Oid			datdba;
+	bool		login_is_db_owner;
+	bool		is_create;
+
+	Assert(stmt->schema_name != NULL);
+
+	/*
+	 * If the login is not the db owner or the login is not the member of
+	 * sysadmin or login is not the schema owner, then it doesn't have the permission to CREATE/DROP FULLTEXT INDEX.
+	 */
+	login_is_db_owner = 0 == strncmp(login, get_owner_of_db(dbname), NAMEDATALEN);
+	datdba = get_role_oid("sysadmin", false);
+	schema_name = gen_schema_name_for_fulltext_index((char *)stmt->schema_name);
+	schemaOid = LookupExplicitNamespace(schema_name, true);						
+	table_name = stmt->table_name;
+	is_create = stmt->is_create;
+
+	// Check if schema exists
+	if (!OidIsValid(schemaOid))
+		ereport(ERROR,
+			(errcode(ERRCODE_UNDEFINED_SCHEMA),
+				errmsg("schema \"%s\" does not exist",
+					stmt->schema_name)));
+
+	// Check if the user has necessary permissions for CREATE/DROP FULLTEXT INDEX
+	if (!is_member_of_role(GetSessionUserId(), datdba) && !login_is_db_owner && !pg_namespace_ownercheck(schemaOid, GetUserId()))
+	{
+		const char *error_msg = is_create ? "A default full-text catalog does not exist in the database or user does not have permission to perform this action" : "Cannot drop the full-text index, because it does not exist or you do not have permission";
+    	ereport(ERROR, 
+			(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), 
+				errmsg("%s", error_msg)));	
+	}
+
+	relid = get_relname_relid((const char *) table_name, schemaOid);
+
+	// Check if table exists
+	if (!OidIsValid(relid))
+		ereport(ERROR,
+			(errcode(ERRCODE_UNDEFINED_TABLE),
+				errmsg("relation \"%s\" does not exist",
+					table_name)));
+
+	// Get the existing fulltext index name
+	old_ft_index_name = get_fulltext_index_name(relid, table_name);
+
+	if (is_create)
+	{
+		uniq_index_name = construct_unique_index_name((char *) stmt->index_name, table_name);
+		if(is_unique_index(relid, (const char *) uniq_index_name))
+		{
+			column_name = stmt->column_name;
+			ft_index_name = construct_unique_index_name("ft_index", table_name);
+			if (old_ft_index_name)
+				ereport(ERROR,
+					(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+						errmsg("A full-text index for table or indexed view \"%s\" has already been created.",
+							table_name)));
+			else
+				query_str = gen_createfulltextindex_cmds(table_name, schema_name, column_name, ft_index_name);
+		}
+		else
+			ereport(ERROR,
+				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+					errmsg("'\"%s\"' is not a valid index to enforce a full-text search key. A full-text search key must be a unique, non-nullable, single-column index which is not offline, is not defined on a non-deterministic or imprecise nonpersisted computed column, does not have a filter, and has maximum size of 900 bytes. Choose another index for the full-text key.",
+						stmt->index_name)));
+	}
+	else
+	{
+		if (old_ft_index_name)					
+			query_str = gen_dropfulltextindex_cmds(old_ft_index_name, schema_name);
+		else
+			ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+					errmsg("Table or indexed view \"%s\" does not have a full-text index or user does not have permission to perform this action.",
+						table_name))); 
+	}
+
+	/* The above query will be
+	 * executed using ProcessUtility()
+	 */
+	exec_utility_cmd_helper(query_str);
+	return PLTSQL_RC_OK;
+}

contrib/babelfishpg_tsql/src/pl_funcs-2.c

@@ -485,6 +485,7 @@ free_stmt2(PLtsql_stmt *stmt)
 		case PLTSQL_STMT_GRANTDB:
 		case PLTSQL_STMT_CHANGE_DBOWNER:
 		case PLTSQL_STMT_GRANTSCHEMA:
+		case PLTSQL_STMT_FULLTEXTINDEX:
 		case PLTSQL_STMT_SET_EXPLAIN_MODE:
 			{
 				/* Nothing to free */

contrib/babelfishpg_tsql/src/pltsql.h

@@ -194,7 +194,8 @@ typedef enum PLtsql_stmt_type
 	PLTSQL_STMT_GRANTDB,
 	PLTSQL_STMT_CHANGE_DBOWNER,
 	PLTSQL_STMT_DBCC,
-	PLTSQL_STMT_GRANTSCHEMA
+	PLTSQL_STMT_GRANTSCHEMA,
+	PLTSQL_STMT_FULLTEXTINDEX,
 } PLtsql_stmt_type;
 
 /*
@@ -1065,6 +1066,20 @@ typedef struct PLtsql_stmt_grantschema
 	char		*schema_name;	/* schema name */
 } PLtsql_stmt_grantschema;
 
+/*
+ *	Fulltext Index stmt
+ */
+typedef struct PLtsql_stmt_fulltextindex
+{
+	PLtsql_stmt_type cmd_type;
+	int			lineno;
+	char        *table_name;   /* table name */
+	List		*column_name;  /* column name */
+	char		*index_name;   /* index name */
+	char		*schema_name;  /* schema name */
+	bool		is_create;     /* flag for create index */		
+} PLtsql_stmt_fulltextindex;
+
 /*
  * ASSERT statement
  */
@@ -2034,6 +2049,12 @@ extern int	pltsql_yyparse(void);
 
 /* functions in pltsql_utils.c */
 extern List *gen_grantschema_subcmds(const char *schema, const char *db_user, bool is_grant, bool with_grant_option, const char *privilege);
+extern char *gen_createfulltextindex_cmds(const char *table_name, const char *schema_name, const List *column_name, const char *index_name);
+extern char *gen_dropfulltextindex_cmds(const char *index_name, const char *schema_name);
+extern char *get_fulltext_index_name(Oid relid, const char *table_name);
+extern const char *gen_schema_name_for_fulltext_index(const char *schema_name);
+extern bool check_fulltext_exist(const char *schema_name, const char *table_name);
+extern bool is_unique_index(Oid relid, const char *index_name);
 extern int	TsqlUTF8LengthInUTF16(const void *vin, int len);
 extern void TsqlCheckUTF16Length_bpchar(const char *s, int32 len, int32 maxlen, int charlen, bool isExplicit);
 extern void TsqlCheckUTF16Length_varchar(const char *s, int32 len, int32 maxlen, bool isExplicit);