Support copy from directory table for entry distribution policy.

We support copy from entry distribution policy directory table, which
may be used in cloud.

Authored-by: Zhang Wenchao zwcpostgres@gmail.com

Author: zhangwenchao-123

src/backend/commands/copyfrom.c

@@ -36,6 +36,8 @@
 #include "cdb/cdbaocsam.h"
 #include "cdb/cdbappendonlyam.h"
 #include "cdb/cdbdisp.h"
+#include "cdb/cdbhash.h"
+#include "cdb/cdbutil.h"
 #include "cdb/cdbvars.h"
 #include "commands/copy.h"
 #include "commands/copyfrom_internal.h"
@@ -726,7 +728,7 @@ formDirTableSlot(CopyFromState cstate,
 	char	   *field[5];
 	FmgrInfo   *in_functions = cstate->in_functions;
 	Oid		   *typioparams = cstate->typioparams;
-	List	   *attnumlist = cstate->qd_attnumlist;
+	List	   *attnumlist = cstate->attnumlist;
 	pg_time_t	stampTime = (pg_time_t) time(NULL);
 	char		lastModified[128];
 
@@ -783,6 +785,7 @@ CopyFromDirectoryTable(CopyFromState cstate)
 	char		buffer[DIR_FILE_BUFF_SIZE];
 	int64		processed = 0;
 	int64		fileSize = 0;
+	UFile	   *file;
 	CdbCopy	   *cdbCopy = NULL;
 	char	   *dirTablePath;
 	char	   *orgiFileName;
@@ -795,6 +798,7 @@ CopyFromDirectoryTable(CopyFromState cstate)
 	DirectoryTable     *dirTable;
 	pg_cryptohash_ctx  *hashCtx;
 	uint8 md5Sum[MD5_DIGEST_LENGTH];
+	char		errorMessage[256];
 	GpDistributionData *distData = NULL; /* distribution data used to compute target seg */
 
 	/*
@@ -989,8 +993,6 @@ CopyFromDirectoryTable(CopyFromState cstate)
 		List	   *recheckIndexes = NIL;
 		CommandId	mycid = GetCurrentCommandId(true);
 		MemoryContext oldcontext = CurrentMemoryContext;
-		char		errorMessage[256];
-		UFile		*file;
 		bool		has_tuple = false;
 		bool		update_indexes;
 
@@ -1164,9 +1166,124 @@ CopyFromDirectoryTable(CopyFromState cstate)
 	}
 	else
 	{
-		ereport(ERROR,
-					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-					 errmsg("This copy from dispatch mode is not supported.")));
+		List	   *recheckIndexes = NIL;
+		CommandId	mycid = GetCurrentCommandId(true);
+		bool		update_indexes;
+
+		formDirTableSlot(cstate,
+						 dirTable->spcId,
+						 relaFileName,
+						 0,
+						 NULL,
+						 cstate->opts.tags,
+						 myslot->tts_values,
+						 myslot->tts_isnull);
+		ExecStoreVirtualTuple(myslot);
+
+		/* OK, store the tuple and create index entries for it */
+		table_tuple_insert(resultRelInfo->ri_RelationDesc,
+						   myslot, mycid, 0, NULL);
+
+		recheckIndexes = ExecInsertIndexTuples(resultRelInfo,
+											   myslot,
+											   estate,
+											   false,
+											   false,
+											   NULL,
+											   NIL);
+
+		/* AFTER ROW INSERT Triggers */
+		ExecARInsertTriggers(estate, resultRelInfo, myslot,
+							 recheckIndexes, cstate->transition_capture);
+
+		list_free(recheckIndexes);
+
+		CommandCounterIncrement();
+
+		if (UFileExists(dirTable->spcId, orgiFileName))
+		{
+			UFileUnlink(dirTable->spcId, orgiFileName);
+		}
+
+		if (orgiFileDir)
+			UFileEnsurePath(dirTable->spcId, orgiFileDir);
+
+		file = UFileOpen(dirTable->spcId,
+						 orgiFileName,
+						 O_CREAT | O_WRONLY,
+						 errorMessage,
+						 sizeof(errorMessage));
+
+		if (file == NULL)
+			ereport(ERROR,
+						(errcode(ERRCODE_IO_ERROR),
+						 errmsg("failed to open file \"%s\": %s", orgiFileName, errorMessage)));
+
+		/* Delete uploaded file when the transaction fails */
+		UFileAddPendingDelete(cstate->rel, dirTable->spcId, orgiFileName, false);
+
+		hashCtx = pg_cryptohash_create(PG_MD5);
+		if (hashCtx == NULL)
+			ereport(ERROR,
+						(errcode(ERRCODE_OUT_OF_MEMORY),
+						 errmsg("failed to create md5hash context: out of memory")));
+		pg_cryptohash_init(hashCtx);
+
+		for (;;)
+		{
+			CHECK_FOR_INTERRUPTS();
+
+			bytesRead = CopyReadBinaryData(cstate, buffer, DIR_FILE_BUFF_SIZE);
+
+			if (bytesRead > 0)
+			{
+				if (UFileWrite(file, buffer, bytesRead) == -1)
+					ereport(ERROR,
+								(errcode(ERRCODE_IO_ERROR),
+								 errmsg("failed to write file \"%s\": %s", orgiFileName, UFileGetLastError(file))));
+
+				fileSize += bytesRead;
+				pg_cryptohash_update(hashCtx, (const uint8 *) buffer, bytesRead);
+			}
+
+			if (bytesRead != DIR_FILE_BUFF_SIZE)
+			{
+				break;
+			}
+		}
+
+		if (UFileSync(file) != 0)
+			ereport(ERROR,
+						(errcode(ERRCODE_IO_ERROR),
+						 errmsg("unable to sync file \"%s\": %s", glob_copystmt->dirfilename, UFileGetLastError(file))));
+
+		UFileClose(file);
+
+		pg_cryptohash_final(hashCtx, md5Sum, sizeof(md5Sum));
+		pg_cryptohash_free(hashCtx);
+		bytesToHex(md5Sum, hexMd5Sum);
+
+		myslot->tts_values[1] = Int64GetDatum(fileSize);
+		myslot->tts_values[3] = CStringGetTextDatum((char *) hexMd5Sum);
+		myslot->tts_isnull[3] = false;
+
+		simple_table_tuple_update(resultRelInfo->ri_RelationDesc, &myslot->tts_tid, myslot,
+								  estate->es_snapshot, &update_indexes);
+
+		ExecClearTuple(myslot);
+
+		/*
+		 * We count only tuples not suppressed by a BEFORE INSERT trigger
+		 * or FDW; this is the same definition used by nodeModifyTable.c
+		 * for counting tuples inserted by an INSERT command.  Update
+		 * progress of the COPY command as well.
+		 *
+		 * MPP: incrementing this counter here only matters for utility
+		 * mode. in dispatch mode only the dispatcher COPY collects row
+		 * count, so this counter is meaningless.
+		 */
+		pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED,
+									 ++processed);
 	}
 
 	cstate->filename = NULL;
@@ -1273,6 +1390,8 @@ BeginCopyFromDirectoryTable(ParseState *pstate,
 		cstate->dispatch_mode = COPY_DISPATCH;
 	else if (Gp_role == GP_ROLE_EXECUTE)
 		cstate->dispatch_mode = COPY_EXECUTOR;
+	else
+		cstate->dispatch_mode = COPY_DIRECT;
 
 	cstate->cur_relname = RelationGetRelationName(cstate->rel);
 	cstate->cur_lineno = 0;

src/backend/commands/tablecmds.c

@@ -645,7 +645,6 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 	if (relkind == RELKIND_DIRECTORY_TABLE)
 	{
 		schema = GetDirectoryTableSchema();
-		stmt->distributedBy = GetDirectoryTableDistributedBy();
 	}
 	else
 	{

src/backend/parser/gram.y

@@ -8414,7 +8414,7 @@ AlterStorageUserMappingStmt:
 
 CreateDirectoryTableStmt:
             CREATE DIRECTORY TABLE qualified_name
-            table_access_method_clause OptTableSpace OptDistributedBy OptTagOptList
+            table_access_method_clause OptTableSpace OptTagOptList
                 {
                     CreateDirectoryTableStmt *n = makeNode(CreateDirectoryTableStmt);
                     $4->relpersistence = RELPERSISTENCE_PERMANENT;
@@ -8427,20 +8427,15 @@ CreateDirectoryTableStmt:
                     /* TODO: support tablespace for data table ? */
                     n->base.tablespacename = NULL;
                     n->base.if_not_exists = false;
-                    n->base.distributedBy = (DistributedBy *) $7;
-                    if (n->base.distributedBy != NULL)
-                        ereport(ERROR,
-                                    (errcode(ERRCODE_SYNTAX_ERROR),
-                                     errmsg("Create directory table is not allowed to set distributed by."),
-                                     parser_errposition(@7)));
+                    n->base.distributedBy = GetDirectoryTableDistributedBy();
                     n->base.relKind = RELKIND_DIRECTORY_TABLE;
-                    n->base.tags = $8;
+                    n->base.tags = $7;
                     n->tablespacename = $6;
 
                     $$ = (Node *) n;
                 }
             | CREATE DIRECTORY TABLE IF_P NOT EXISTS qualified_name
-            table_access_method_clause OptTableSpace OptDistributedBy OptTagOptList
+            table_access_method_clause OptTableSpace OptTagOptList
                 {
                     CreateDirectoryTableStmt *n = makeNode(CreateDirectoryTableStmt);
                     $7->relpersistence = RELPERSISTENCE_PERMANENT;
@@ -8453,14 +8448,9 @@ CreateDirectoryTableStmt:
                     /* TODO: support tablespace for data table? */
                     n->base.tablespacename = NULL;
                     n->base.if_not_exists = true;
-                    n->base.distributedBy = (DistributedBy *) $10;
-                    if (n->base.distributedBy != NULL)
-                        ereport(ERROR,
-                                    (errcode(ERRCODE_SYNTAX_ERROR),
-                                     errmsg("Create directory table is not allowed to set distributed by."),
-                                     parser_errposition(@10)));
+                    n->base.distributedBy = GetDirectoryTableDistributedBy();
                     n->base.relKind = RELKIND_DIRECTORY_TABLE;
-                    n->base.tags = $11;
+                    n->base.tags = $10;
                     n->tablespacename = $9;
 
                     $$ = (Node *) n;

src/test/regress/output/directory_table.source

@@ -511,15 +511,15 @@ SELECT relname, relisshared, relpersistence, relkind FROM pg_class WHERE relname
 CREATE DIRECTORY TABLE dir_table1;
 CREATE DIRECTORY TABLE dir_table2 TABLESPACE directory_tblspc;
 CREATE DIRECTORY TABLE dir_table3 TABLESPACE directory_tblspc DISTRIBUTED BY(relative_path);    -- fail
-ERROR:  Create directory table is not allowed to set distributed by.
+ERROR:  syntax error at or near "DISTRIBUTED"
 LINE 1: ...TORY TABLE dir_table3 TABLESPACE directory_tblspc DISTRIBUTE...
                                                              ^
 CREATE DIRECTORY TABLE dir_table3 TABLESPACE directory_tblspc DISTRIBUTED RANDOMLY; -- fail
-ERROR:  Create directory table is not allowed to set distributed by.
+ERROR:  syntax error at or near "DISTRIBUTED"
 LINE 1: ...TORY TABLE dir_table3 TABLESPACE directory_tblspc DISTRIBUTE...
                                                              ^
 CREATE DIRECTORY TABLE dir_table3 TABLESPACE directory_tblspc DISTRIBUTED REPLICATED;   -- fail
-ERROR:  Create directory table is not allowed to set distributed by.
+ERROR:  syntax error at or near "DISTRIBUTED"
 LINE 1: ...TORY TABLE dir_table3 TABLESPACE directory_tblspc DISTRIBUTE...
                                                              ^
 CREATE DIRECTORY TABLE dir_table3 TABLESPACE directory_tblspc;

src/test/regress/output/directory_table_optimizer.source

@@ -511,15 +511,15 @@ SELECT relname, relisshared, relpersistence, relkind FROM pg_class WHERE relname
 CREATE DIRECTORY TABLE dir_table1;
 CREATE DIRECTORY TABLE dir_table2 TABLESPACE directory_tblspc;
 CREATE DIRECTORY TABLE dir_table3 TABLESPACE directory_tblspc DISTRIBUTED BY(relative_path);    -- fail
-ERROR:  Create directory table is not allowed to set distributed by.
+ERROR:  syntax error at or near "DISTRIBUTED"
 LINE 1: ...TORY TABLE dir_table3 TABLESPACE directory_tblspc DISTRIBUTE...
                                                              ^
 CREATE DIRECTORY TABLE dir_table3 TABLESPACE directory_tblspc DISTRIBUTED RANDOMLY; -- fail
-ERROR:  Create directory table is not allowed to set distributed by.
+ERROR:  syntax error at or near "DISTRIBUTED"
 LINE 1: ...TORY TABLE dir_table3 TABLESPACE directory_tblspc DISTRIBUTE...
                                                              ^
 CREATE DIRECTORY TABLE dir_table3 TABLESPACE directory_tblspc DISTRIBUTED REPLICATED;   -- fail
-ERROR:  Create directory table is not allowed to set distributed by.
+ERROR:  syntax error at or near "DISTRIBUTED"
 LINE 1: ...TORY TABLE dir_table3 TABLESPACE directory_tblspc DISTRIBUTE...
                                                              ^
 CREATE DIRECTORY TABLE dir_table3 TABLESPACE directory_tblspc;