Merge pull request IvorySQL#926 from yuanyl630/master

code refactor of oracle-compatibility feature rowid

Author: gaoxueyu

contrib/ivorysql_ora/preload_ora_misc.sql

@@ -6,7 +6,9 @@ insert into dual values('X');
 GRANT SELECT ON dual TO PUBLIC;
 
 --
--- ROWID type
+-- Oracle-compatible ROWID and UROWID types
+-- ROWID: Composite type containing row object ID and row number
+-- UROWID: Universal ROWID, compatible with ROWID
 --
 CREATE TYPE sys.rowid AS(rowoid OID, rowno bigint);
 CREATE TYPE sys.urowid AS(rowoid OID, rowno bigint);

contrib/tablefunc/expected/ivy_tablefunc.out

@@ -14,6 +14,7 @@ SELECT avg(normal_rand)::int, count(*) FROM normal_rand(-1, 250, 0.2);
 ERROR:  number of rows cannot be negative
 --
 -- crosstab()
+-- Use 'orarowid' instead of 'rowid' to avoid conflict with system column
 --
 CREATE TABLE ct(id int, rowclass text, orarowid text, attribute text, val text);
 \copy ct from 'data/ct.data'

contrib/tablefunc/sql/ivy_tablefunc.sql

@@ -10,6 +10,7 @@ SELECT avg(normal_rand)::int, count(*) FROM normal_rand(-1, 250, 0.2);
 
 --
 -- crosstab()
+-- Use 'orarowid' instead of 'rowid' to avoid conflict with system column
 --
 CREATE TABLE ct(id int, rowclass text, orarowid text, attribute text, val text);
 \copy ct from 'data/ct.data'

src/backend/access/transam/xlog.c

@@ -5372,8 +5372,6 @@ BootStrapXLOG(uint32 data_checksum_version)
 	/* save database compatible level value */
 	ControlFile->dbmode = bootstrap_database_mode;
 	ControlFile->casemode = identifier_case_switch;
-	
-	
 
 	/* some additional ControlFile fields are set in WriteControlFile() */
 	WriteControlFile();
@@ -5571,7 +5569,7 @@ CleanupAfterArchiveRecovery(TimeLineID EndOfLogTLI, XLogRecPtr EndOfLog,
 	}
 }
 
-/* IvorySQL: BEGIN - case sensitive indentify
+/* IvorySQL: BEGIN - case-sensitive indentifier support
  * Read database compatibility mode from pg_control file
  *
  */
@@ -5588,7 +5586,7 @@ int GetDatabaseStyleFromControl(char* path)
 	else
 		configdir = make_absolute_path(getenv("PGDATA"));
 
-	sprintf(pathname,"%s/%s",configdir,XLOG_CONTROL_FILE);
+	snprintf(pathname, sizeof(pathname), "%s/%s", configdir, XLOG_CONTROL_FILE);
 
 	fd = open(pathname, O_RDONLY | PG_BINARY, 0);
 
@@ -5630,7 +5628,7 @@ int GetCaseSwitchModeFromControl(char* path)
 	else
 		configdir = make_absolute_path(getenv("PGDATA"));
 
-	sprintf(pathname,"%s/%s",configdir,XLOG_CONTROL_FILE);
+	snprintf(pathname, sizeof(pathname), "%s/%s", configdir, XLOG_CONTROL_FILE);
 
 	fd = open(pathname, O_RDONLY | PG_BINARY, 0);
 
@@ -5665,7 +5663,7 @@ void SetCaseGucOption(char* path)
 
 	dbstyle = GetDatabaseStyleFromControl(path);
 
-	//the pg mode does not need to set
+	/* PG mode leaves identifier_case_switch unchanged. */
 	if (DB_ORACLE == dbstyle)
 	{
 		int casemode;
@@ -5691,6 +5689,7 @@ void SetCaseGucOption(char* path)
 					(errmsg("Incorrect case conversion mode value \"%d\"", casemode)));
 	}
 }
+/* IvorySQL: END - case-sensitive identifier support */
 
 /*
  * Check to see if required parameters are set high enough on this server

src/backend/catalog/heap.c

@@ -228,7 +228,9 @@ static const FormData_pg_attribute a6 = {
 };
 
 /*
- * Compatible Oracle ROWID pseudo column.
+ * Oracle-compatible ROWID pseudo-column.
+ * This system column provides a unique identifier for each row,
+ * compatible with Oracle's ROWID functionality.
  */
 static const FormData_pg_attribute a7 = {
    .attname = {"rowid"},

src/backend/catalog/namespace.c

@@ -4420,7 +4420,8 @@ preprocessNamespacePath(const char *searchPath, Oid roleid,
 		char	   *curname = (char *) lfirst(l);
 		Oid			namespaceId;
 
-		if ((compatible_db == ORA_PARSER && enable_case_switch && identifier_case_switch != NORMAL) ?	/* IvorySQL: case sensitive indentify */
+		if ((compatible_db == ORA_PARSER && enable_case_switch && 
+		         identifier_case_switch != NORMAL) ?	/* IvorySQL: case-sensitive indentifier support */
 			pg_strcasecmp(curname, "$user") == 0 : strcmp(curname, "$user") == 0)
 		{
 			/* $user --- substitute namespace matching user name, if any */
@@ -5433,4 +5434,3 @@ get_nodefvalargposition(HeapTuple proctup, int pronargs)
 	}
 	return nodefvalargposition;
 }
-

src/backend/parser/parse_utilcmd.c

@@ -68,7 +68,7 @@
 #include "utils/ruleutils.h"
 #include "utils/syscache.h"
 #include "utils/typcache.h"
-#include <math.h> 
+#include <math.h>
 #include "utils/guc.h"
 #include "utils/ora_compatible.h"
 
@@ -280,10 +280,12 @@ transformCreateStmt(CreateStmt *stmt, const char *queryString)
 	if (cxt.inhRelations)
 	{
 		ListCell   *inher;
+
 		foreach(inher, cxt.inhRelations)
 		{
 			RangeVar   *inh = lfirst_node(RangeVar, inher);
 			Relation	prel;
+
 			prel = table_openrv(inh, AccessShareLock);
 			cxt.hasrowid = cxt.hasrowid || prel->rd_rel->relhasrowid;
 			table_close(prel, NoLock);
@@ -302,7 +304,8 @@ transformCreateStmt(CreateStmt *stmt, const char *queryString)
 		{
 			case T_ColumnDef:
 				{
-					ColumnDef *col = (ColumnDef *) element;
+					ColumnDef  *col = (ColumnDef *) element;
+
 					if (compatible_db == ORA_PARSER && strcmp(col->colname, "rowid") == 0)
 						elog(ERROR, "column name \"%s\" conflicts with a system column name", col->colname);
 					else
@@ -331,13 +334,14 @@ transformCreateStmt(CreateStmt *stmt, const char *queryString)
 	 */
 	foreach(elements, cxt.columns)
 	{
-		ColumnDef          *element = lfirst(elements);
-		if(element->identity == ATTRIBUTE_IDENTITY_DEFAULT_ON_NULL ||
-				element->identity == ATTRIBUTE_ORA_IDENTITY_ALWAYS ||
-					element->identity == ATTRIBUTE_ORA_IDENTITY_BY_DEFAULT)
+		ColumnDef  *element = lfirst(elements);
+
+		if (element->identity == ATTRIBUTE_IDENTITY_DEFAULT_ON_NULL ||
+			element->identity == ATTRIBUTE_ORA_IDENTITY_ALWAYS ||
+			element->identity == ATTRIBUTE_ORA_IDENTITY_BY_DEFAULT)
 			ora_identity_cnt++;
 	}
-	if(ora_identity_cnt > 1)
+	if (ora_identity_cnt > 1)
 		elog(ERROR, "table can have only one identity column");
 
 	if (like_found && cxt.hasrowid)
@@ -348,7 +352,7 @@ transformCreateStmt(CreateStmt *stmt, const char *queryString)
 
 	if (cxt.hasrowid)
 	{
-		Oid	   snamespaceid;
+		Oid			snamespaceid;
 		char	   *snamespace;
 		char	   *sname;
 
@@ -368,47 +372,48 @@ transformCreateStmt(CreateStmt *stmt, const char *queryString)
 								   false);
 
 		/*
-		 * Build a CREATE SEQUENCE command to create the sequence object,
-		 * and add it to the list of things to be done before this CREATE/ALTER TABLE
+		 * Build a CREATE SEQUENCE command to create the sequence object, and
+		 * add it to the list of things to be done before this CREATE/ALTER
+		 * TABLE
 		 */
 		seqstmt = makeNode(CreateSeqStmt);
 		seqstmt->with_rowid = true;
 		seqstmt->sequence = makeRangeVar(snamespace, sname, -1);
 		seqstmt->options = lcons(makeDefElem("as",
-					 	(Node *) makeTypeNameFromOid(INT8OID, -1),
-						 -1),
-					 seqstmt->options);
+											 (Node *) makeTypeNameFromOid(INT8OID, -1),
+											 -1),
+								 seqstmt->options);
 		if (rowid_seq_cache > 1)
 		{
 			seqstmt->options = lcons(makeDefElem("cache",
-							 (Node *) makeInteger(rowid_seq_cache),
-							 -1),
-						 seqstmt->options);
+												 (Node *) makeInteger(rowid_seq_cache),
+												 -1),
+									 seqstmt->options);
 		}
 		else
 			seqstmt->options = lcons(makeDefElem("nocache",
-								 NULL,
-							 -1),
-						seqstmt->options);
+												 NULL,
+												 -1),
+									 seqstmt->options);
 
 		if (cxt.rel)
 			seqstmt->ownerId = cxt.rel->rd_rel->relowner;
 
 		cxt.blist = lappend(cxt.blist, seqstmt);
 
 		/*
-		 * Build an ALTER SEQUENCE ... OWNED BY command to mark the sequence as owned
-		 * by this table.
+		 * Build an ALTER SEQUENCE ... OWNED BY command to mark the sequence
+		 * as owned by this table.
 		 */
 		altseqstmt = makeNode(AlterSeqStmt);
 		altseqstmt->sequence = makeRangeVar(snamespace, sname, -1);
 
 		relnamelist = list_make3(makeString(snamespace),
-					 makeString(cxt.relation->relname),
-					 makeString(sname));
+								 makeString(cxt.relation->relname),
+								 makeString(sname));
 
 		altseqstmt->options = list_make1(makeDefElem("owned_by",
-								 (Node *) relnamelist, -1));
+													 (Node *) relnamelist, -1));
 		cxt.alist = lappend(cxt.alist, altseqstmt);
 	}
 
@@ -971,11 +976,14 @@ transformColumnDefinition(CreateStmtContext *cxt, ColumnDef *column)
 					ctype = typenameType(cxt->pstate, column->typeName, &typmod);
 					typeOid = ((Form_pg_type) GETSTRUCT(ctype))->oid;
 
-					/* Convert compatible identity smallint/int type column to bigint type */
+					/*
+					 * Convert compatible identity smallint/int type column to
+					 * bigint type
+					 */
 					if ((constraint->generated_when == ATTRIBUTE_IDENTITY_DEFAULT_ON_NULL
-							|| constraint->generated_when == ATTRIBUTE_ORA_IDENTITY_ALWAYS
-								|| constraint->generated_when == ATTRIBUTE_ORA_IDENTITY_BY_DEFAULT)
-									&& (typeOid == INT4OID || typeOid == INT2OID))
+						 || constraint->generated_when == ATTRIBUTE_ORA_IDENTITY_ALWAYS
+						 || constraint->generated_when == ATTRIBUTE_ORA_IDENTITY_BY_DEFAULT)
+						&& (typeOid == INT4OID || typeOid == INT2OID))
 					{
 						column->typeName = makeTypeName("int8");
 						typeOid = INT8OID;
@@ -1296,7 +1304,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 	if (relation->rd_rel->relkind == RELKIND_COMPOSITE_TYPE)
 	{
 		aclresult = object_aclcheck(TypeRelationId, relation->rd_rel->reltype, GetUserId(),
-									 ACL_USAGE);
+									ACL_USAGE);
 		if (aclresult != ACLCHECK_OK)
 			aclcheck_error(aclresult, OBJECT_TYPE,
 						   RelationGetRelationName(relation));
@@ -2690,7 +2698,7 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
 				 * mentioned above.
 				 */
 				Datum		attoptions =
-				get_attoptions(RelationGetRelid(index_rel), i + 1);
+					get_attoptions(RelationGetRelid(index_rel), i + 1);
 
 				defopclass = GetDefaultOpClass(attform->atttypid,
 											   index_rel->rd_rel->relam);
@@ -3765,12 +3773,12 @@ transformAlterTableStmt(Oid relid, AlterTableStmt *stmt,
 
 			case AT_AddRowidsRecurse:
 				{
-					Oid 			snamespaceid;
-					char	 		*snamespace;
-					char	 		*sname;
-					List 	 		*relnamelist;
-					CreateSeqStmt 	*seqstmt;
-					AlterSeqStmt 	*altseqstmt;
+					Oid			snamespaceid;
+					char	   *snamespace;
+					char	   *sname;
+					List	   *relnamelist;
+					CreateSeqStmt *seqstmt;
+					AlterSeqStmt *altseqstmt;
 
 					if (cmd->is_rowid)
 					{
@@ -3790,45 +3798,46 @@ transformAlterTableStmt(Oid relid, AlterTableStmt *stmt,
 												   false);
 
 						/*
-						 * Build a CREATE SEQUENCE command to create the sequence object,
-						 * and add it to the list of things to be done before this CREATE/ALTER TABLE
+						 * Build a CREATE SEQUENCE command to create the
+						 * sequence object, and add it to the list of things
+						 * to be done before this CREATE/ALTER TABLE
 						 */
 						seqstmt = makeNode(CreateSeqStmt);
 						seqstmt->with_rowid = true;
 						seqstmt->sequence = makeRangeVar(snamespace, sname, -1);
 						seqstmt->options = lcons(makeDefElem("as",
-									 (Node *) makeTypeNameFromOid(INT8OID, -1),
-										 -1), seqstmt->options);
+															 (Node *) makeTypeNameFromOid(INT8OID, -1),
+															 -1), seqstmt->options);
 						if (rowid_seq_cache > 1)
 						{
 							seqstmt->options = lcons(makeDefElem("cache",
-									 (Node *) makeInteger(rowid_seq_cache),
-										 -1), seqstmt->options);
+																 (Node *) makeInteger(rowid_seq_cache),
+																 -1), seqstmt->options);
 						}
 						else
 							seqstmt->options = lcons(makeDefElem("nocache",
-										 NULL,
-										 -1),
-										seqstmt->options);
+																 NULL,
+																 -1),
+													 seqstmt->options);
 
 						if (cxt.rel)
 							seqstmt->ownerId = cxt.rel->rd_rel->relowner;
 
 						cxt.blist = lappend(cxt.blist, seqstmt);
 
 						/*
-						 * Build an ALTER SEQUENCE ... OWNED BY command to mark the sequence as owned
-						 * by this table.
+						 * Build an ALTER SEQUENCE ... OWNED BY command to
+						 * mark the sequence as owned by this table.
 						 */
 						altseqstmt = makeNode(AlterSeqStmt);
 						altseqstmt->sequence = makeRangeVar(snamespace, sname, -1);
 
 						relnamelist = list_make3(makeString(snamespace),
-									 makeString(cxt.relation->relname),
-									 makeString(sname));
+												 makeString(cxt.relation->relname),
+												 makeString(sname));
 
 						altseqstmt->options = list_make1(makeDefElem("owned_by",
-										 (Node *) relnamelist, -1));
+																	 (Node *) relnamelist, -1));
 						cxt.alist = lappend(cxt.alist, altseqstmt);
 
 						newcmds = lappend(newcmds, cmd);
@@ -4670,7 +4679,7 @@ transformPartitionRangeBounds(ParseState *pstate, List *blist,
 		 * as ColumnRefs.
 		 */
 		if (IsA(expr, ColumnRef) ||
-				IsA(expr, ColumnRefOrFuncCall))
+			IsA(expr, ColumnRefOrFuncCall))
 		{
 			ColumnRef  *cref = NULL;
 			char	   *cname = NULL;

src/backend/parser/scansup.c

@@ -56,11 +56,11 @@ downcase_identifier(const char *ident, int len, bool warn, bool truncate)
 	/*
 	 * SQL99 specifies Unicode-aware case normalization, which we don't yet
 	 * have the infrastructure for.  Instead we use tolower() to provide a
-	 * locale-aware translation.  However, there are some locales where this
-	 * is not right either (eg, Turkish may do strange things with 'i' and
-	 * 'I').  Our current compromise is to use tolower() for characters with
-	 * the high bit set, as long as they aren't part of a multi-byte
-	 * character, and use an ASCII-only downcasing for 7-bit characters.
+	 * locale-aware translation.  However, in some locales (for example, 
+	 * Turkish with 'i' and 'I') this still is not correct.  Our compromise is
+	 * to use tolower() for characters with the high bit set, as long as they
+	 * aren't part of a multi-byte character, and use an ASCII-only approach
+	 * for 7-bit characters.
 	 */
 	for (i = 0; i < len; i++)
 	{

src/backend/tcop/backend_startup.c

@@ -797,7 +797,7 @@ ProcessStartupPacket(Port *port, bool ssl_done, bool gss_done)
 
 			if (strcmp(nameptr, "database") == 0)
 			{
-				/* Oracle compatibility tranfor upper to lower */
+				/* Oracle compatibility: transform uppercase identifiers to lowercase. */
 				char *database_name = pstrdup(valptr);
 
 				if (ORA_PARSER == compatible_db && database_name != NULL)
@@ -850,7 +850,7 @@ ProcessStartupPacket(Port *port, bool ssl_done, bool gss_done)
 			}
 			else if (strcmp(nameptr, "user") == 0)
 			{
-				/* Oracle compatibility tranfor upper to lower */
+				/* Oracle compatibility: transform uppercase identifiers to lowercase. */
 				char *user_name = pstrdup(valptr);
 
 				if (ORA_PARSER == compatible_db && user_name != NULL)