BABEL-5975: Handle long identifiers for Babelfish

Signed-off-by: anju15bharti <abanju@amazon.com>

Author: anju15bharti

src/backend/access/nbtree/nbtinsert.c

@@ -16,6 +16,7 @@
 #include "postgres.h"
 
 #include "access/nbtree.h"
+#include "executor/executor.h"
 #include "access/nbtxlog.h"
 #include "access/transam.h"
 #include "access/xloginsert.h"
@@ -663,14 +664,22 @@ _bt_check_unique(Relation rel, BTInsertState insertstate, Relation heapRel,
 						key_desc = BuildIndexValueDescription(rel, values,
 															  isnull);
 
-						ereport(ERROR,
-								(errcode(ERRCODE_UNIQUE_VIOLATION),
-								 errmsg("duplicate key value violates unique constraint \"%s\"",
-										RelationGetRelationName(rel)),
-								 key_desc ? errdetail("Key %s already exists.",
-													  key_desc) : 0,
-								 errtableconstraint(heapRel,
-													RelationGetRelationName(rel))));
+						{
+							const char *conname = RelationGetRelationName(rel);
+
+							if (bbf_get_original_index_name_hook)
+								conname = bbf_get_original_index_name_hook(conname);
+							if (bbf_get_original_constraint_name_hook && strlen(conname) >= NAMEDATALEN - 1)
+								conname = bbf_get_original_constraint_name_hook(conname);
+							ereport(ERROR,
+									(errcode(ERRCODE_UNIQUE_VIOLATION),
+									 errmsg("duplicate key value violates unique constraint \"%s\"",
+											conname),
+									 key_desc ? errdetail("Key %s already exists.",
+														  key_desc) : 0,
+									 errtableconstraint(heapRel,
+														RelationGetRelationName(rel))));
+						}
 					}
 				}
 				else if (all_dead && (!inposting ||

src/backend/catalog/genbki.pl

@@ -68,6 +68,7 @@
     PROCNAMENSPSIGNATURE
     SYSNAMESPACENAME
     AUTHIDUSEREXTROLENAME
+    IDENTMAPPINGNAME
 );
 
 foreach my $header (@ARGV)

src/backend/commands/view.c

@@ -19,6 +19,8 @@
 #include "catalog/namespace.h"
 #include "commands/tablecmds.h"
 #include "commands/view.h"
+#include "parser/parser.h"
+#include "parser/scansup.h"
 #include "nodes/makefuncs.h"
 #include "nodes/nodeFuncs.h"
 #include "parser/analyze.h"
@@ -70,6 +72,12 @@ DefineVirtualRelation(RangeVar *relation, List *tlist, bool replace,
 											exprTypmod((Node *) tle->expr),
 											exprCollation((Node *) tle->expr));
 
+			/* For TSQL, MD5-truncate long column names for attname storage */
+			if (sql_dialect == SQL_DIALECT_TSQL && def->colname &&
+				strlen(def->colname) >= NAMEDATALEN)
+				truncate_identifier(def->colname,
+								strlen(def->colname), false);
+
 			if (inherit_view_constraints_from_table_hook)
 				(*inherit_view_constraints_from_table_hook) (def, tle->resorigtbl, tle->resorigcol);
 

src/backend/executor/execMain.c

@@ -65,6 +65,9 @@
 #include "utils/rls.h"
 #include "utils/snapmgr.h"
 
+bbf_get_original_constraint_name_hook_type bbf_get_original_constraint_name_hook = NULL;
+bbf_get_original_index_name_hook_type bbf_get_original_index_name_hook = NULL;
+
 
 /* Hooks for plugins to get control in ExecutorStart/Run/Finish/End */
 ExecutorStart_hook_type ExecutorStart_hook = NULL;
@@ -2096,12 +2099,17 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 													 tupdesc,
 													 modifiedCols,
 													 64);
+			{
+				const char *display_name = failed;
+				if (bbf_get_original_constraint_name_hook)
+					display_name = bbf_get_original_constraint_name_hook(failed);
 			ereport(ERROR,
 					(errcode(ERRCODE_CHECK_VIOLATION),
 					 errmsg("new row for relation \"%s\" violates check constraint \"%s\"",
-							RelationGetRelationName(orig_rel), failed),
+							RelationGetRelationName(orig_rel), display_name),
 					 val_desc ? errdetail("Failing row contains %s.", val_desc) : 0,
 					 errtableconstraint(orig_rel, failed)));
+			}
 		}
 	}
 }

src/backend/parser/parse_clause.c

@@ -2115,7 +2115,7 @@ findTargetlistEntrySQL92(ParseState *pstate, Node *node, List **tlist,
 					((!tle_name_comparison_hook &&
 					  strcmp(tle->resname, name) == 0) ||
 					 (tle_name_comparison_hook &&
-					  (*tle_name_comparison_hook)(tle->resname, name))))
+					  (*tle_name_comparison_hook)(tle->resname, name, pstate->p_sourcetext, location))))
 				{
 					if (target_result != NULL)
 					{

src/backend/rewrite/rewriteDefine.c

@@ -28,6 +28,8 @@
 #include "parser/parse_utilcmd.h"
 #include "rewrite/rewriteDefine.h"
 #include "rewrite/rewriteManip.h"
+#include "parser/parser.h"
+#include "parser/scansup.h"
 #include "rewrite/rewriteSupport.h"
 #include "utils/acl.h"
 #include "utils/builtins.h"
@@ -562,7 +564,11 @@ checkRuleResultList(List *targetList, TupleDesc resultDesc, bool isSelect,
 					 errmsg("cannot create a RETURNING list for a relation containing dropped columns")));
 
 		/* Check name match if required; no need for two error texts here */
-		if (requireColumnNameMatch && strcmp(tle->resname, attname) != 0)
+		if (requireColumnNameMatch && strcmp(tle->resname, attname) != 0 &&
+			!(sql_dialect == SQL_DIALECT_TSQL &&
+			  strlen(tle->resname) >= NAMEDATALEN &&
+			  strlen(attname) < NAMEDATALEN &&
+			  pg_strcasecmp(downcase_truncate_identifier(tle->resname, strlen(tle->resname), false), attname) == 0))
 			ereport(ERROR,
 					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
 					 errmsg("SELECT rule's target entry %d has different column name from column \"%s\"",

src/bin/pg_dump/dump_babel_utils.c

@@ -1056,6 +1056,7 @@ setBabelfishDependenciesForLogicalDatabaseDump(Archive *fout)
 	 * sys.babelfish_namespace_ext
 	 * sys.babelfish_extended_properties
 	 * sys.babelfish_schema_permissions
+	 * sys.babelfish_identifier_mapping
 	 * sys.babelfish_partition_function
 	 * sys.babelfish_partition_scheme
 	 * sys.babelfish_partition_depend
@@ -1065,6 +1066,7 @@ setBabelfishDependenciesForLogicalDatabaseDump(Archive *fout)
 						 "FROM pg_class "
 						 "WHERE relname in ('babelfish_schema_permissions', "
 						 "'babelfish_namespace_ext', "
+						 "'babelfish_identifier_mapping', "
 						 "'babelfish_partition_function', "
 						 "'babelfish_partition_scheme', "
 						 "'babelfish_partition_depend', "
@@ -1137,6 +1139,12 @@ addFromClauseForLogicalDatabaseDump(PQExpBuffer buf, TableInfo *tbinfo)
 						  "ON a.nspname = b.nspname "
 						  "WHERE b.dbid = %d",
 						  fmtQualifiedDumpable(tbinfo), bbf_db_id);
+	else if (strcmp(tbinfo->dobj.name, "babelfish_identifier_mapping") == 0)
+		appendPQExpBuffer(buf, " FROM ONLY %s a "
+						  "INNER JOIN sys.babelfish_namespace_ext b "
+						  "ON a.nspname = b.nspname "
+						  "WHERE b.dbid = %d",
+						  fmtQualifiedDumpable(tbinfo), bbf_db_id);
 	else if(strcmp(tbinfo->dobj.name, "babelfish_authid_user_ext") == 0)
 	{
 		appendPQExpBuffer(buf, " FROM ONLY %s a "
@@ -1200,6 +1208,7 @@ addFromClauseForPhysicalDatabaseDump(PQExpBuffer buf, TableInfo *tbinfo)
 						fmtQualifiedDumpable(tbinfo), babel_init_user);
 	else if(strcmp(tbinfo->dobj.name, "babelfish_domain_mapping") == 0 ||
 			strcmp(tbinfo->dobj.name, "babelfish_function_ext") == 0 ||
+			strcmp(tbinfo->dobj.name, "babelfish_identifier_mapping") == 0 ||
 			strcmp(tbinfo->dobj.name, "babelfish_view_def") == 0 ||
 			strcmp(tbinfo->dobj.name, "babelfish_server_options") == 0 ||
 			strcmp(tbinfo->dobj.name, "babelfish_extended_properties") == 0 ||

src/include/executor/executor.h

@@ -101,6 +101,12 @@ extern PGDLLEXPORT TriggerRecuresiveCheck_hook_type TriggerRecuresiveCheck_hook;
 typedef bool (*bbfViewHasInsteadofTrigger_hook_type) (Relation view, CmdType event);
 extern PGDLLIMPORT bbfViewHasInsteadofTrigger_hook_type bbfViewHasInsteadofTrigger_hook;
 
+typedef const char *(*bbf_get_original_constraint_name_hook_type)(const char *conname);
+extern PGDLLIMPORT bbf_get_original_constraint_name_hook_type bbf_get_original_constraint_name_hook;
+
+typedef const char *(*bbf_get_original_index_name_hook_type)(const char *idxname);
+extern PGDLLIMPORT bbf_get_original_index_name_hook_type bbf_get_original_index_name_hook;
+
 typedef Datum (*adjust_numeric_result_hook_type) (Plan *plan, Node *expr, Datum result, bool result_isnull, Oid result_type, int32 result_typmod);
 extern PGDLLIMPORT adjust_numeric_result_hook_type adjust_numeric_result_hook;
 

src/include/parser/parse_clause.h

@@ -51,7 +51,7 @@ extern List *addTargetToSortList(ParseState *pstate, TargetEntry *tle,
 extern Index assignSortGroupRef(TargetEntry *tle, List *tlist);
 extern bool targetIsInSortList(TargetEntry *tle, Oid sortop, List *sortList);
 
-typedef bool (*tle_name_comparison_hook_type)(const char *tlename, const char *identifier);
+typedef bool (*tle_name_comparison_hook_type)(const char *tlename, const char *identifier, const char *sourcetext, int location);
 extern PGDLLEXPORT tle_name_comparison_hook_type tle_name_comparison_hook;
 
 typedef void (*sortby_nulls_hook_type)(SortGroupClause *sortcl, bool reverse);