Improve RECORD support

src/backend/distributed/planner/distributed_planner.c

@@ -84,7 +84,6 @@ static PlannedStmt * FinalizeNonRouterPlan(PlannedStmt *localPlan,
 										   DistributedPlan *distributedPlan,
 										   CustomScan *customScan);
 static PlannedStmt * FinalizeRouterPlan(PlannedStmt *localPlan, CustomScan *customScan);
-static int32 BlessRecordExpression(Expr *expr);
 static void CheckNodeIsDumpable(Node *node);
 static Node * CheckNodeCopyAndSerialization(Node *node);
 static void AdjustReadIntermediateResultCost(RangeTblEntry *rangeTableEntry,
@@ -648,7 +647,7 @@ CreateDistributedPlan(uint64 planId, Query *originalQuery, Query *query, ParamLi
 		/*
 		 * For select queries we, if router executor is enabled, first try to
 		 * plan the query as a router query. If not supported, otherwise try
-		 * the full blown plan/optimize/physical planing process needed to
+		 * the full blown plan/optimize/physical planning process needed to
 		 * produce distributed query plans.
 		 */
 
@@ -1151,7 +1150,7 @@ FinalizeRouterPlan(PlannedStmt *localPlan, CustomScan *customScan)
  * "bless" the tuple descriptor, which adds a transient type to the type cache
  * and assigns it a type mod value, which is the key in the type cache.
  */
-static int32
+int32
 BlessRecordExpression(Expr *expr)
 {
 	int32 typeMod = -1;

src/backend/distributed/planner/multi_logical_optimizer.c

@@ -1382,7 +1382,7 @@ MasterExtendedOpNode(MultiExtendedOp *originalOpNode,
 	foreach(targetEntryCell, targetEntryList)
 	{
 		TargetEntry *originalTargetEntry = (TargetEntry *) lfirst(targetEntryCell);
-		TargetEntry *newTargetEntry = copyObject(originalTargetEntry);
+		TargetEntry *newTargetEntry = flatCopyTargetEntry(originalTargetEntry);
 		Expr *originalExpression = originalTargetEntry->expr;
 		Expr *newExpression = NULL;
 
@@ -1412,6 +1412,11 @@ MasterExtendedOpNode(MultiExtendedOp *originalOpNode,
 			column->varoattno = walkerContext->columnId;
 			walkerContext->columnId++;
 
+			if (column->vartype == RECORDOID)
+			{
+				column->vartypmod = BlessRecordExpression(originalTargetEntry->expr);
+			}
+
 			newExpression = (Expr *) column;
 		}
 

src/backend/distributed/planner/multi_logical_planner.c

@@ -777,7 +777,7 @@ MultiNodeTree(Query *queryTree)
 
 /*
  * ContainsReadIntermediateResultFunction determines whether an expresion tree contains
- * a call to the read_intermediate_results function.
+ * a call to the read_intermediate_result function.
  */
 bool
 ContainsReadIntermediateResultFunction(Node *node)

src/backend/distributed/planner/multi_master_planner.c

@@ -13,6 +13,7 @@
 
 #include "postgres.h"
 
+#include "catalog/pg_type.h"
 #include "commands/extension.h"
 #include "distributed/citus_ruleutils.h"
 #include "distributed/function_utils.h"
@@ -83,7 +84,7 @@ MasterNodeSelectPlan(DistributedPlan *distributedPlan, CustomScan *remoteScan)
 
 /*
  * MasterTargetList uses the given worker target list's expressions, and creates
- * a target target list for the master node. This master target list keeps the
+ * a target list for the master node. This master target list keeps the
  * temporary table's columns on the master node.
  */
 static List *
@@ -105,13 +106,16 @@ MasterTargetList(List *workerTargetList)
 			continue;
 		}
 
-		masterTargetEntry = copyObject(workerTargetEntry);
-
 		masterColumn = makeVarFromTargetEntry(tableId, workerTargetEntry);
 		masterColumn->varattno = columnId;
 		masterColumn->varoattno = columnId;
 		columnId++;
 
+		if (masterColumn->vartype == RECORDOID)
+		{
+			masterColumn->vartypmod = BlessRecordExpression(workerTargetEntry->expr);
+		}
+
 		/*
 		 * The master target entry has two pieces to it. The first piece is the
 		 * target entry's expression, which we set to the newly created column.
@@ -119,6 +123,7 @@ MasterTargetList(List *workerTargetList)
 		 * from the worker target entry. Note that any changes to worker target
 		 * entry's sort and group clauses will *break* us here.
 		 */
+		masterTargetEntry = flatCopyTargetEntry(workerTargetEntry);
 		masterTargetEntry->expr = (Expr *) masterColumn;
 		masterTargetList = lappend(masterTargetList, masterTargetEntry);
 	}

src/backend/distributed/planner/recursive_planning.c

@@ -1542,7 +1542,7 @@ ShouldTransformRTE(RangeTblEntry *rangeTableEntry)
 	/*
 	 * We should wrap only function rtes that are not LATERAL and
 	 * without WITH ORDINALITY clause
-	 * */
+	 */
 	if (rangeTableEntry->rtekind != RTE_FUNCTION ||
 		rangeTableEntry->lateral ||
 		rangeTableEntry->funcordinality)

src/include/distributed/distributed_planner.h

@@ -122,5 +122,6 @@ extern Node * ResolveExternalParams(Node *inputNode, ParamListInfo boundParams);
 extern bool IsMultiTaskPlan(struct DistributedPlan *distributedPlan);
 extern RangeTblEntry * RemoteScanRangeTableEntry(List *columnNameList);
 extern int GetRTEIdentity(RangeTblEntry *rte);
+extern int32 BlessRecordExpression(Expr *expr);
 
 #endif /* DISTRIBUTED_PLANNER_H */

src/test/regress/expected/multi_select_distinct.out

@@ -77,6 +77,17 @@ SELECT DISTINCT l_orderkey, strpos(l_shipmode, 'I')
          33 |      2
 (5 rows)
 
+-- row types are supported
+SELECT DISTINCT (l_orderkey, l_partkey) AS pair FROM lineitem_hash_part ORDER BY 1 LIMIT 5;
+   pair    
+-----------
+ (1,2132)
+ (1,15635)
+ (1,24027)
+ (1,63700)
+ (1,67310)
+(5 rows)
+
 -- distinct on partition column
 -- verify counts match with respect to count(distinct)
 CREATE TEMP TABLE temp_orderkeys AS SELECT DISTINCT l_orderkey FROM lineitem_hash_part;

src/test/regress/expected/row_types.out

@@ -35,17 +35,52 @@ SELECT create_distributed_function('record_returner(int)');
 (1 row)
 
 INSERT INTO test VALUES (1,2), (1,3), (2,2), (2,3);
--- multi-shard queries do not support row types
+-- multi-shard queries support row types
 SELECT (x,y) FROM test ORDER BY x, y;
-ERROR:  input of anonymous composite types is not implemented
+  row  
+-------
+ (1,2)
+ (1,3)
+ (2,2)
+ (2,3)
+(4 rows)
+
 SELECT (x,y) FROM test GROUP BY x, y ORDER BY x, y;
-ERROR:  input of anonymous composite types is not implemented
+  row  
+-------
+ (1,2)
+ (1,3)
+ (2,2)
+ (2,3)
+(4 rows)
+
 select distinct (x,y) AS foo, x, y FROM test ORDER BY x, y;
-ERROR:  input of anonymous composite types is not implemented
+  foo  | x | y 
+-------+---+---
+ (1,2) | 1 | 2
+ (1,3) | 1 | 3
+ (2,2) | 2 | 2
+ (2,3) | 2 | 3
+(4 rows)
+
 SELECT table_returner(x) FROM test ORDER BY x, y;
-ERROR:  input of anonymous composite types is not implemented
+ table_returner 
+----------------
+ (1,1)
+ (1,1)
+ (2,2)
+ (2,2)
+(4 rows)
+
 SELECT record_returner(x) FROM test ORDER BY x, y;
-ERROR:  input of anonymous composite types is not implemented
+ record_returner 
+-----------------
+ (2,returned)
+ (2,returned)
+ (3,returned)
+ (3,returned)
+(4 rows)
+
 -- router queries support row types
 SELECT (x,y) FROM test WHERE x = 1 ORDER BY x, y;
   row  

src/test/regress/sql/multi_select_distinct.sql

@@ -30,6 +30,9 @@ SELECT DISTINCT l_orderkey, strpos(l_shipmode, 'I')
 	ORDER BY 2, 1
 	LIMIT 5;
 
+-- row types are supported
+SELECT DISTINCT (l_orderkey, l_partkey) AS pair FROM lineitem_hash_part ORDER BY 1 LIMIT 5;
+
 -- distinct on partition column
 -- verify counts match with respect to count(distinct)
 CREATE TEMP TABLE temp_orderkeys AS SELECT DISTINCT l_orderkey FROM lineitem_hash_part;

src/test/regress/sql/row_types.sql

@@ -26,7 +26,7 @@ SELECT create_distributed_function('record_returner(int)');
 
 INSERT INTO test VALUES (1,2), (1,3), (2,2), (2,3);
 
--- multi-shard queries do not support row types
+-- multi-shard queries support row types
 SELECT (x,y) FROM test ORDER BY x, y;
 SELECT (x,y) FROM test GROUP BY x, y ORDER BY x, y;
 select distinct (x,y) AS foo, x, y FROM test ORDER BY x, y;