Fix issues with read_intermediate_result signature

src/backend/distributed/Makefile

@@ -13,7 +13,7 @@ EXTVERSIONS = 5.0 5.0-1 5.0-2  \
 	6.2-1 6.2-2 6.2-3 6.2-4 \
 	7.0-1 7.0-2 7.0-3 7.0-4 7.0-5 7.0-6 7.0-7 7.0-8 7.0-9 7.0-10 7.0-11 7.0-12 7.0-13 7.0-14 7.0-15 \
 	7.1-1 7.1-2 7.1-3 7.1-4 \
-	7.2-1 7.2-2
+	7.2-1 7.2-2 7.2-3
 
 # All citus--*.sql files in the source directory
 DATA = $(patsubst $(citus_abs_srcdir)/%.sql,%.sql,$(wildcard $(citus_abs_srcdir)/$(EXTENSION)--*--*.sql))
@@ -183,6 +183,8 @@ $(EXTENSION)--7.2-1.sql: $(EXTENSION)--7.1-4.sql $(EXTENSION)--7.1-4--7.2-1.sql
 	cat $^ > $@
 $(EXTENSION)--7.2-2.sql: $(EXTENSION)--7.2-1.sql $(EXTENSION)--7.2-1--7.2-2.sql
 	cat $^ > $@
+$(EXTENSION)--7.2-3.sql: $(EXTENSION)--7.2-2.sql $(EXTENSION)--7.2-2--7.2-3.sql
+	cat $^ > $@
 
 NO_PGXS = 1
 

src/backend/distributed/citus--7.2-2--7.2-3.sql

@@ -0,0 +1,13 @@
+/* citus--7.2-2--7.2-3 */
+
+DROP FUNCTION pg_catalog.read_intermediate_result(text,citus.copy_format);
+DROP TYPE citus.copy_format;
+
+CREATE TYPE pg_catalog.citus_copy_format AS ENUM ('csv', 'binary', 'text');
+
+CREATE OR REPLACE FUNCTION pg_catalog.read_intermediate_result(result_id text, format pg_catalog.citus_copy_format default 'csv')
+    RETURNS SETOF record
+    LANGUAGE C STRICT VOLATILE PARALLEL SAFE
+    AS 'MODULE_PATHNAME', $$read_intermediate_result$$;
+COMMENT ON FUNCTION pg_catalog.read_intermediate_result(text,pg_catalog.citus_copy_format)
+    IS 'read a file and return it as a set of records';

src/backend/distributed/citus.control

@@ -1,6 +1,6 @@
 # Citus extension
 comment = 'Citus distributed database'
-default_version = '7.2-2'
+default_version = '7.2-3'
 module_pathname = '$libdir/citus'
 relocatable = false
 schema = pg_catalog

src/backend/distributed/utils/metadata_cache.c

@@ -117,6 +117,7 @@ typedef struct MetadataCacheData
 	Oid distTransactionRelationId;
 	Oid distTransactionGroupIndexId;
 	Oid distTransactionRecordIndexId;
+	Oid copyFormatTypeId;
 	Oid readIntermediateResultFuncId;
 	Oid extraDataContainerFuncId;
 	Oid workerHashFunctionId;
@@ -1835,22 +1836,17 @@ DistPlacementGroupidIndexId(void)
 }
 
 
-/* return oid of the read_intermediate_result(text,citus.copy_format) function */
+/* return oid of the read_intermediate_result(text,citus_copy_format) function */
 Oid
 CitusReadIntermediateResultFuncId(void)
 {
 	if (MetadataCache.readIntermediateResultFuncId == InvalidOid)
 	{
-		bool missingOK = false;
-
-		List *copyFormatTypeNameList = list_make2(makeString("citus"),
-												  makeString("copy_format"));
-		TypeName *copyFormatTypeName = makeTypeNameFromNameList(copyFormatTypeNameList);
-		Oid copyFormatTypeOid = LookupTypeNameOid(NULL, copyFormatTypeName, missingOK);
-
 		List *functionNameList = list_make2(makeString("pg_catalog"),
 											makeString("read_intermediate_result"));
+		Oid copyFormatTypeOid = CitusCopyFormatTypeId();
 		Oid paramOids[2] = { TEXTOID, copyFormatTypeOid };
+		bool missingOK = false;
 
 		MetadataCache.readIntermediateResultFuncId =
 			LookupFuncName(functionNameList, 2, paramOids, missingOK);
@@ -1860,6 +1856,22 @@ CitusReadIntermediateResultFuncId(void)
 }
 
 
+/* return oid of the citus.copy_format enum type */
+Oid
+CitusCopyFormatTypeId(void)
+{
+	if (MetadataCache.copyFormatTypeId == InvalidOid)
+	{
+		char *typeName = "citus_copy_format";
+		MetadataCache.copyFormatTypeId = GetSysCacheOid2(TYPENAMENSP,
+														 PointerGetDatum(typeName),
+														 PG_CATALOG_NAMESPACE);
+	}
+
+	return MetadataCache.copyFormatTypeId;
+}
+
+
 /* return oid of the citus_extradata_container(internal) function */
 Oid
 CitusExtraDataContainerFuncId(void)

src/include/distributed/metadata_cache.h

@@ -123,6 +123,9 @@ extern Oid DistTransactionGroupIndexId(void);
 extern Oid DistTransactionRecordIndexId(void);
 extern Oid DistPlacementGroupidIndexId(void);
 
+/* type oids */
+extern Oid CitusCopyFormatTypeId(void);
+
 /* function oids */
 extern Oid CitusReadIntermediateResultFuncId(void);
 extern Oid CitusExtraDataContainerFuncId(void);

src/test/regress/expected/multi_extension.out

@@ -131,6 +131,7 @@ ALTER EXTENSION citus UPDATE TO '7.1-3';
 ALTER EXTENSION citus UPDATE TO '7.1-4';
 ALTER EXTENSION citus UPDATE TO '7.2-1';
 ALTER EXTENSION citus UPDATE TO '7.2-2';
+ALTER EXTENSION citus UPDATE TO '7.2-3';
 -- show running version
 SHOW citus.version;
  citus.version 

src/test/regress/expected/multi_multiuser.out

@@ -170,6 +170,35 @@ COPY "postgresql.conf" TO STDOUT WITH (format transmit);
 ERROR:  operation is not allowed
 HINT:  Run the command with a superuser.
 SET citus.task_executor_type TO 'real-time';
+-- should be able to use intermediate results as any user
+BEGIN;
+SELECT create_intermediate_result('topten', 'SELECT s FROM generate_series(1,10) s');
+ create_intermediate_result 
+----------------------------
+                         10
+(1 row)
+
+SELECT * FROM read_intermediate_result('topten', 'binary'::citus_copy_format) AS res (s int) ORDER BY s;
+ s  
+----
+  1
+  2
+  3
+  4
+  5
+  6
+  7
+  8
+  9
+ 10
+(10 rows)
+
+END;
+-- as long as we don't read from a table
+BEGIN;
+SELECT create_intermediate_result('topten', 'SELECT count(*) FROM test');
+ERROR:  permission denied for relation test
+ABORT;
 RESET ROLE;
 DROP TABLE test;
 DROP USER full_access;

src/test/regress/sql/multi_extension.sql

@@ -131,6 +131,7 @@ ALTER EXTENSION citus UPDATE TO '7.1-3';
 ALTER EXTENSION citus UPDATE TO '7.1-4';
 ALTER EXTENSION citus UPDATE TO '7.2-1';
 ALTER EXTENSION citus UPDATE TO '7.2-2';
+ALTER EXTENSION citus UPDATE TO '7.2-3';
 
 -- show running version
 SHOW citus.version;

src/test/regress/sql/multi_multiuser.sql

@@ -116,6 +116,17 @@ COPY "postgresql.conf" TO STDOUT WITH (format transmit);
 
 SET citus.task_executor_type TO 'real-time';
 
+-- should be able to use intermediate results as any user
+BEGIN;
+SELECT create_intermediate_result('topten', 'SELECT s FROM generate_series(1,10) s');
+SELECT * FROM read_intermediate_result('topten', 'binary'::citus_copy_format) AS res (s int) ORDER BY s;
+END;
+
+-- as long as we don't read from a table
+BEGIN;
+SELECT create_intermediate_result('topten', 'SELECT count(*) FROM test');
+ABORT;
+
 RESET ROLE;
 
 DROP TABLE test;