Fix bugs and add test cases

This commit resolved the following issues: - view usage in pivot stmt produce empty result - NULL should be the aggregate result when no row is selected(except COUNT) This commit also added the following test cases: - join usage in pivot stmt - view usage in pivot stmt - aggregate string values with NULL output result for pivot stmt Task: BABEL-4558 Signed-off-by: Yanjie Xu <[email protected]>
babelfish-for-postgresql · Nov 21, 2023 · 4c26d9b · 4c26d9b
1 parent 455f57a
commit 4c26d9b
Show file tree

Hide file tree

Showing 9 changed files with 414 additions and 40 deletions.
diff --git a/contrib/babelfishpg_tsql/runtime/functions.c b/contrib/babelfishpg_tsql/runtime/functions.c
@@ -165,13 +165,14 @@ void	   *get_servername_internal(void);
 void	   *get_servicename_internal(void);
 void	   *get_language(void);
 void	   *get_host_id(void);
-int 		SPI_execute_raw_parsetree(RawStmt *parsetree, const char * sourcetext, bool read_only, long tcount);
-static HTAB *load_categories_hash(RawStmt *cats_sql, const char * sourcetext, MemoryContext per_query_ctx);
-static Tuplestorestate *get_bbf_pivot_tuplestore(RawStmt *sql,
-										const char * sourcetext,
-										HTAB *bbf_pivot_hash,
-										TupleDesc tupdesc,
-										bool randomAccess);
+int 		SPI_execute_raw_parsetree(RawStmt *parsetree, const char *sourcetext, bool read_only, long tcount);
+static HTAB *load_categories_hash(RawStmt *cats_sql, const char *sourcetext, MemoryContext per_query_ctx);
+static Tuplestorestate *get_bbf_pivot_tuplestore(RawStmt 	*sql,
+												const char 	*sourcetext,
+												String 		*funcName,
+												HTAB 		*bbf_pivot_hash,
+												TupleDesc 	tupdesc,
+												bool 		randomAccess);
 extern bool canCommitTransaction(void);
 extern bool is_ms_shipped(char *object_name, int type, Oid schema_id);
 
@@ -3814,6 +3815,7 @@ bbf_pivot(PG_FUNCTION_ARGS)
 	int				nestlevel;
 	List			*per_pivot_list;
 	char			*query_string;
+	String			*funcName;
 
 	/* check to see if caller supports us returning a tuplestore */
 	if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
@@ -3846,9 +3848,11 @@ bbf_pivot(PG_FUNCTION_ARGS)
 		bbf_pivot_src_sql = list_nth_node(RawStmt, per_pivot_list, 0);
 		bbf_pivot_cat_sql = list_nth_node(RawStmt, per_pivot_list, 1);
 		query_string = list_nth(per_pivot_list, 2);
+		funcName = list_nth(per_pivot_list, 3);
 	}
 	PG_CATCH();
 	{
+		MemoryContextSwitchTo(oldcontext);
 		PG_RE_THROW();
 	}
 	PG_END_TRY();
@@ -3881,6 +3885,7 @@ bbf_pivot(PG_FUNCTION_ARGS)
 	/* now go build it */
 	rsinfo->setResult = get_bbf_pivot_tuplestore(bbf_pivot_src_sql,
 												query_string,
+												funcName,
 												bbf_pivot_hash,
 												tupdesc,
 												rsinfo->allowedModes & SFRM_Materialize_Random);
@@ -3959,22 +3964,24 @@ load_categories_hash(RawStmt *cats_sql, const char * sourcetext, MemoryContext p
 		{
 			bbf_pivot_cat_desc *catdesc;
 			char	   *catname;
+			char	   *catname_lower;
 			HeapTuple	spi_tuple;
 
 			/* get the next sql result tuple */
 			spi_tuple = spi_tuptable->vals[i];
 
 			/* get the category from the current sql result tuple */
 			catname = SPI_getvalue(spi_tuple, spi_tupdesc, 1);
-			if (catname == NULL)
+			catname_lower = downcase_identifier(catname, strlen(catname), false, false);
+			if (catname_lower == NULL)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("provided \"categories\" SQL must " \
 								"not return NULL values")));
 
 			SPIcontext = MemoryContextSwitchTo(per_query_ctx);
 			catdesc = (bbf_pivot_cat_desc *) palloc(sizeof(bbf_pivot_cat_desc));
-			catdesc->catname = pstrdup(catname);
+			catdesc->catname = pstrdup(catname_lower);
 			catdesc->attidx = i;
 			/* Add the proc description block to the hashtable */
 			bbf_pivot_HashTableInsert(bbf_pivot_hash, catdesc);
@@ -3996,11 +4003,12 @@ load_categories_hash(RawStmt *cats_sql, const char * sourcetext, MemoryContext p
  * create and populate the bbf_pivot tuplestore
  */
 static Tuplestorestate *
-get_bbf_pivot_tuplestore(RawStmt *sql,
-						const char * sourcetext,
-						HTAB *bbf_pivot_hash,
-						TupleDesc tupdesc,
-						bool randomAccess)
+get_bbf_pivot_tuplestore(RawStmt 	*sql,
+						const char 	*sourcetext,
+						String		*funcName,
+						HTAB 		*bbf_pivot_hash,
+						TupleDesc 	tupdesc,
+						bool 		randomAccess)
 {
 	Tuplestorestate *tupstore;
 	int			num_categories = hash_get_num_entries(bbf_pivot_hash);
@@ -4035,6 +4043,8 @@ get_bbf_pivot_tuplestore(RawStmt *sql,
 		int			j;
 		int			non_pivot_columns;
 		int			result_ncols;
+		/* only COUNT will output 0 when no row is selected */
+		bool		output_zero = !strcmp(funcName->sval, "count");
 
 		if (num_categories == 0)
 		{
@@ -4080,6 +4090,7 @@ get_bbf_pivot_tuplestore(RawStmt *sql,
 			HeapTuple	spi_tuple;
 			bbf_pivot_cat_desc *catdesc;
 			char	   *catname;
+			char	   *catname_lower;
 			bool 	   	is_new_row = false;
 
 			/* get the next sql result tuple */
@@ -4118,11 +4129,16 @@ get_bbf_pivot_tuplestore(RawStmt *sql,
 					*/
 					if (!firstpass)
 					{
-						for (j = 0; j < result_ncols; j++)
+						/* only COUNT will output 0 when no row is selected */
+						if (output_zero)
 						{
-							if (values[j] == NULL)
-								values[j] = pstrdup("0");
+							for (j = 0; j < result_ncols; j++)
+							{
+								if (values[j] == NULL)
+									values[j] = pstrdup("0");
+							}
 						}
+
 						/* rowid changed, flush the previous output row */
 						tuple = BuildTupleFromCStrings(attinmeta, values);
 
@@ -4140,12 +4156,18 @@ get_bbf_pivot_tuplestore(RawStmt *sql,
 				}
 			}
 
-			/* look up the category and fill in the appropriate column */
+			/*
+			 * look up the category and fill in the appropriate column
+			 * Column names get from SPI result can be in mixed case but we only use
+			 * lowered cases column names for new pivot table, and so we need to lower 
+			 * the column names obtained from SPI results to get the tuple index from 
+			 * the hash table
+			 */
 			catname = SPI_getvalue(spi_tuple, spi_tupdesc, ncols - 1);
-
-			if (catname != NULL)
+			catname_lower = downcase_identifier(catname, strlen(catname), false, false);
+			if (catname_lower != NULL)
 			{
-				bbf_pivot_HashTableLookup(bbf_pivot_hash, catname, catdesc);
+				bbf_pivot_HashTableLookup(bbf_pivot_hash, catname_lower, catdesc);
 
 				if (catdesc)
 					values[catdesc->attidx + non_pivot_columns] =
@@ -4163,10 +4185,13 @@ get_bbf_pivot_tuplestore(RawStmt *sql,
 		}
 
 		/* flush the last output row */
-		for (i = 0; i < result_ncols; i++)
+		if (output_zero)
 		{
-			if (values[i] == NULL)
-				values[i] = pstrdup("0");
+			for (i = 0; i < result_ncols; i++)
+			{
+				if (values[i] == NULL)
+					values[i] = pstrdup("0");
+			}
 		}
 		tuple = BuildTupleFromCStrings(attinmeta, values);
 		tuplestore_puttuple(tupstore, tuple);

diff --git a/contrib/babelfishpg_tsql/src/backend_parser/gram-tsql-rule.y b/contrib/babelfishpg_tsql/src/backend_parser/gram-tsql-rule.y
@@ -1598,7 +1598,6 @@ tsql_pivot_expr: TSQL_PIVOT '(' func_application FOR columnref IN_P in_expr ')'
 					ResTarget 		*a_star_restarget;
 					RangeSubselect 	*range_sub_select;
 					Alias 			*temptable_alias;
-					SortBy 			*s;
 					List    *ret;
 					List    *value_col_strlist = NULL;
 					List	*subsel_valuelists = NULL;
@@ -1658,16 +1657,8 @@ tsql_pivot_expr: TSQL_PIVOT '(' func_application FOR columnref IN_P in_expr ')'
 					range_sub_select->subquery = (Node *) valuelists_sql;
 					range_sub_select->alias = temptable_alias;
 
-					s = makeNode(SortBy);
-					s->node = makeIntConst(1, -1);
-					s->sortby_dir = 0;
-					s->sortby_nulls = 0;     
-					s->useOp = NIL;
-					s->location = -1;
-
 					category_sql->targetList = list_make1(a_star_restarget);
 					category_sql->fromClause = list_make1(range_sub_select);
-					category_sql->sortClause = list_make1(s);
 
 					ret = list_make4($5, restarget_aggfunc, category_sql, value_col_strlist);
 					$$ = (Node*) ret; 

diff --git a/contrib/babelfishpg_tsql/src/hooks.c b/contrib/babelfishpg_tsql/src/hooks.c
@@ -4304,6 +4304,7 @@ transform_pivot_clause(ParseState *pstate, SelectStmt *stmt)
 	List		*src_sql_fromClause_copy;
 	char		*pivot_colstr;
 	char		*value_colstr;
+	String		*funcName;
 	ColumnRef	*value_col;
 	TargetEntry	*aggfunc_te;
 	RangeFunction	*pivot_from_function;
@@ -4336,6 +4337,7 @@ transform_pivot_clause(ParseState *pstate, SelectStmt *stmt)
 	/* Get pivot column str & value column str from parser result */
 	pivot_colstr = ((String *) llast(((ColumnRef *)stmt->pivotCol)->fields))->sval;
 	value_col = list_nth_node(ColumnRef, ((FuncCall *)((ResTarget *)stmt->aggFunc)->val)->args, 0);
+	funcName = list_nth_node(String, ((FuncCall *)((ResTarget *)stmt->aggFunc)->val)->funcname, 0);
 	value_colstr = list_nth_node(String, value_col->fields, 0)->sval;
 
 	/* Get the targetList of the src table */
@@ -4461,7 +4463,7 @@ transform_pivot_clause(ParseState *pstate, SelectStmt *stmt)
 	c_sql->stmt_len = 0;
 	sourcetext = pstrdup(pstate->p_sourcetext);
 
-	tsql_outmost_estat->pivot_parsetree_list = lappend(tsql_outmost_estat->pivot_parsetree_list, list_make3(copyObject(s_sql), copyObject(c_sql), sourcetext));
+	tsql_outmost_estat->pivot_parsetree_list = lappend(tsql_outmost_estat->pivot_parsetree_list, list_make4(copyObject(s_sql), copyObject(c_sql), sourcetext, copyObject(funcName)));
 	tsql_outmost_estat->pivot_number++;	
 	MemoryContextSwitchTo(oldContext);
 }

diff --git a/test/JDBC/expected/pivot-vu-cleanup.out b/test/JDBC/expected/pivot-vu-cleanup.out
@@ -1,4 +1,16 @@
-use pivot_test
+use pivot_test;
+GO
+
+drop table OSTable;
+GO
+
+drop table STable;
+GO
+
+drop table seating_tbl;
+GO
+
+drop view StoreReceipt_view;
 GO
 
 drop table pivot_insert_into;
@@ -16,8 +28,9 @@ GO
 drop table StoreReceipt;
 GO
 
-use master
+use master;
 GO
 
-drop database pivot_test
+drop database pivot_test;
 GO
+
diff --git a/test/JDBC/expected/pivot-vu-prepare.out b/test/JDBC/expected/pivot-vu-prepare.out
@@ -855,3 +855,142 @@ RETURN
         FOR StoreID in ([2], [3], [4], [5], [6])
     ) AS pvt2
 GO
+
+CREATE VIEW StoreReceipt_view
+AS
+SELECT * FROM StoreReceipt;
+GO
+
+-- BABEL-4558 
+CREATE TABLE OSTable(
+    [Oid] [int] NOT NULL,
+    [Sid] [int] NOT NULL
+)
+GO
+
+
+CREATE TABLE STable(
+    [Id] [int] IDENTITY(1,1) NOT NULL,
+    [Scode] [varchar](10) NOT NULL,
+    [Type] [smallint] NOT NULL
+)
+GO
+
+insert into OSTable (Oid, Sid) values (1, 2);
+GO
+~~ROW COUNT: 1~~
+
+insert into OSTable (Oid, Sid) values (2, 8);
+GO
+~~ROW COUNT: 1~~
+
+insert into OSTable (Oid, Sid) values (3, 5);
+GO
+~~ROW COUNT: 1~~
+
+insert into OSTable (Oid, Sid) values (4, 11);
+GO
+~~ROW COUNT: 1~~
+
+insert into OSTable (Oid, Sid) values (5, 12);
+GO
+~~ROW COUNT: 1~~
+
+insert into OSTable (Oid, Sid) values (6, 8);
+GO
+~~ROW COUNT: 1~~
+
+insert into OSTable (Oid, Sid) values (7, 5);
+GO
+~~ROW COUNT: 1~~
+
+insert into OSTable (Oid, Sid) values (8, 2);
+GO
+~~ROW COUNT: 1~~
+
+insert into OSTable (Oid, Sid) values (9, 15);
+GO
+~~ROW COUNT: 1~~
+
+insert into OSTable (Oid, Sid) values (10, 1);
+GO
+~~ROW COUNT: 1~~
+
+
+insert into STable (Scode, Type) values ('vestibulum', 11);
+GO
+~~ROW COUNT: 1~~
+
+insert into STable (Scode, Type) values ('eget', 15);
+GO
+~~ROW COUNT: 1~~
+
+insert into STable (Scode, Type) values ('pharetra', 13);
+GO
+~~ROW COUNT: 1~~
+
+insert into STable (Scode, Type) values ('nam', 15);
+GO
+~~ROW COUNT: 1~~
+
+insert into STable (Scode, Type) values ('fermentum', 13);
+GO
+~~ROW COUNT: 1~~
+
+insert into STable (Scode, Type) values ('hac', 12);
+GO
+~~ROW COUNT: 1~~
+
+insert into STable (Scode, Type) values ('molestie', 10);
+GO
+~~ROW COUNT: 1~~
+
+insert into STable (Scode, Type) values ('justo', 11);
+GO
+~~ROW COUNT: 1~~
+
+insert into STable (Scode, Type) values ('lobortis', 7);
+GO
+~~ROW COUNT: 1~~
+
+insert into STable (Scode, Type) values ('at', 3);
+GO
+~~ROW COUNT: 1~~
+
+insert into STable (Scode, Type) values ('augue', 9);
+GO
+~~ROW COUNT: 1~~
+
+insert into STable (Scode, Type) values ('luctus', 2);
+GO
+~~ROW COUNT: 1~~
+
+insert into STable (Scode, Type) values ('nisi', 9);
+GO
+~~ROW COUNT: 1~~
+
+insert into STable (Scode, Type) values ('sociis', 1);
+GO
+~~ROW COUNT: 1~~
+
+insert into STable (Scode, Type) values ('ultrices', 14);
+GO
+~~ROW COUNT: 1~~
+
+
+-- table for aggregate with string value
+CREATE TABLE seating_tbl (
+    seatings VARCHAR(20) NOT NULL,
+    left_right VARCHAR(20) NOT NULL
+);
+go
+
+INSERT INTO seating_tbl (seatings, left_right)
+VALUES ('SEAT1', 'LEFT'),
+       ('SEAT1', 'RIGHT'),
+       ('SEAT2', 'LEFT'),
+       ('SEAT3', 'LEFT'),
+       ('SEAT3', 'RIGHT');
+go
+~~ROW COUNT: 5~~
+