Fix direcotory table problems and add guc allow_dml_directory_table.

gpMgmt/bin/gpdirtableload

@@ -69,6 +69,9 @@ def parseargs():
                         help='Database to connect to')
     parser.add_argument('--mode', choices=['upload', 'download'], default="upload",
                         help='Upload or download file to/from directory table')
+    parser.add_argument('--match', choices=['full', 'regex'], default="full",
+                        help='Input file character match mode, use full match ' 
+                        'for file or regex match for directory')
     parser.add_argument('--dest-path', help='In upload mode, this means path relative to '
                                             'the table root directory, while in download '
                                             'mode, means directory to download')
@@ -165,6 +168,7 @@ class gpdirtableload:
         self.startTimestamp = time.time()
         self.pool = None
         self.upload = True
+        self.regexMatch = False
 
         # set default log level
         if self.options.verbose is not None:
@@ -176,6 +180,10 @@ class gpdirtableload:
         if self.options.mode is not None and self.options.mode == 'download':
             self.upload = False
 
+        # set download character match mode
+        if self.options.match is not None and self.options.match == 'regex':
+            self.regexMatch = True
+
         if self.options.dest_path is None:
             self.log(self.ERROR, '--dest-path must be set')        
 
@@ -355,17 +363,24 @@ class gpdirtableload:
         self.allFilesToDownload = []
         self.numFiles = 0
 
-        qry = "SELECT relative_path FROM %s " % self.options.table
+        qry = "SELECT relative_path FROM %s WHERE " % self.options.table
+
+        qry += " relative_path like \'%s" % self.options.input_file
+        if self.regexMatch:
+        	qry += '%'
+        qry += "\'"
 
         if self.options.tag:
-            qry += "WHERE tag = \'%s\'" % self.options.tag
+            qry += "AND tag = \'%s\'" % self.options.tag
 
         self.allFilesToDownload = [s[0] for s in
                                    self.db.query(qry).getresult()]
         self.numFiles = len(self.allFilesToDownload)
 
     def confirmWorkers(self):
-        if self.numFiles < self.options.tasks:
+        if self.numFiles == 0:
+        	self.numWorkers = 1
+        elif self.numFiles < self.options.tasks:
             self.numWorkers = self.numFiles
         else:
             self.numWorkers = self.options.tasks

src/backend/catalog/dependency.c

@@ -70,6 +70,7 @@
 #include "catalog/pg_type.h"
 #include "catalog/pg_type_encoding.h"
 #include "catalog/pg_user_mapping.h"
+#include "catalog/storage_directory_table.h"
 #include "commands/comment.h"
 #include "commands/dbcommands.h"
 #include "commands/defrem.h"
@@ -1489,12 +1490,22 @@ doDeletion(const ObjectAddress *object, int flags)
 				}
 				else
 				{
+					bool 		drop_with_content = (flags & PERFORM_DELETION_WITH_CONTENT) != 0;
+
 					if (object->objectSubId != 0)
 						RemoveAttributeById(object->objectId,
 											object->objectSubId);
 					else
 					{
 						heap_drop_with_catalog(object->objectId);
+
+						/*
+						 * Remove driectory table file conteng
+						 */
+						if (drop_with_content)
+						{
+							DirectoryTableDropStorage(object->objectId);
+						}
 
 						/*
 						 * Delete tag description.

src/backend/catalog/heap.c

@@ -2520,7 +2520,6 @@ heap_drop_with_catalog(Oid relid)
 	if (rel->rd_rel->relkind == RELKIND_DIRECTORY_TABLE)
 	{
 		RemoveDirectoryTableEntry(relid);
-		DirectoryTableDropStorage(rel);
 	}
 
 	/*

src/backend/catalog/storage_directory_table.c

@@ -28,6 +28,8 @@
 #include "utils/rel.h"
 #include "cdb/cdbvars.h"
 
+bool allow_dml_directory_table = false;
+
 /*
  * TODO: support ufile pending delete xlog
  *
@@ -87,8 +89,9 @@ struct PendingRelDeleteAction ufile_pending_rel_deletes_action = {
 };
 
 void
-DirectoryTableDropStorage(Relation rel)
+DirectoryTableDropStorage(Oid relid)
 {
+	Relation rel;
 	char	   *filePath;
 	DirectoryTable *dirTable;
 	TableScanDesc scandesc;
@@ -99,6 +102,7 @@ DirectoryTableDropStorage(Relation rel)
 	Oid			tablespaceoid;
 	char	   *tablespace_name;
 
+	rel = relation_open(relid, AccessExclusiveLock); 
 	dirTable = GetDirectoryTable(RelationGetRelid(rel));
 
 	/*
@@ -133,6 +137,8 @@ DirectoryTableDropStorage(Relation rel)
 	UFileAddPendingDelete(rel, dirTable->spcId, filePath, true);
 
 	pfree(filePath);
+
+	relation_close(rel, NoLock);
 }
 
 void

src/backend/commands/tablecmds.c

@@ -1765,6 +1765,13 @@ RemoveRelations(DropStmt *drop)
 
 	foreach(cell, drop->objects)
 	{
+		if (relkind == RELKIND_DIRECTORY_TABLE)
+		{
+			DropDirectoryTableStmt *dirtable_drop =  (DropDirectoryTableStmt *) drop;
+			if (dirtable_drop->with_content)
+				flags |= PERFORM_DELETION_WITH_CONTENT;
+		}
+
 		RangeVar   *rel = makeRangeVarFromNameList((List *) lfirst(cell));
 		Oid			relOid;
 		ObjectAddress obj;

src/backend/executor/execMain.c

@@ -49,6 +49,7 @@
 #include "access/xact.h"
 #include "catalog/namespace.h"
 #include "catalog/pg_publication.h"
+#include "catalog/storage_directory_table.h"
 #include "commands/matview.h"
 #include "commands/tablespace.h"
 #include "commands/trigger.h"
@@ -2259,10 +2260,11 @@ CheckValidResultRel(ResultRelInfo *resultRelInfo, CmdType operation, ModifyTable
 			{
 				case CMD_INSERT:
 				case CMD_DELETE:
-					ereport(ERROR,
-								(errcode(ERRCODE_WRONG_OBJECT_TYPE),
-								 errmsg("cannot change directory table \"%s\"",
-										RelationGetRelationName(resultRel))));
+					if (!allow_dml_directory_table)
+						ereport(ERROR,
+									(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+								 	 errmsg("cannot change directory table \"%s\"",
+											RelationGetRelationName(resultRel))));
 					break;
 				case CMD_UPDATE:
 					if (mtstate)
@@ -2276,7 +2278,7 @@ CheckValidResultRel(ResultRelInfo *resultRelInfo, CmdType operation, ModifyTable
 						{
 							AttrNumber targetattnum = lfirst_int(lc);
 
-							if (targetattnum != DIRECTORY_TABLE_TAG_COLUMN_ATTNUM)
+							if (targetattnum != DIRECTORY_TABLE_TAG_COLUMN_ATTNUM && !allow_dml_directory_table)
 								ereport(ERROR,
 											(errcode(ERRCODE_WRONG_OBJECT_TYPE),
 											 errmsg("Only allow to update directory \"tag\" column.")));

src/backend/nodes/copyfuncs.c

@@ -4222,6 +4222,22 @@ _copyDropStmt(const DropStmt *from)
 	return newnode;
 }
 
+/*
+ * CopyDropStmtFields
+ *
+ *		This function copies the fields of the DropStmt node.  It is used by
+ *		copy functions for classes which inherit from DropStmt.
+ */
+static void
+CopyDropStmtFields(const DropStmt *from, DropStmt *newnode)
+{
+	COPY_NODE_FIELD(objects);
+	COPY_SCALAR_FIELD(removeType);
+	COPY_SCALAR_FIELD(behavior);
+	COPY_SCALAR_FIELD(missing_ok);
+	COPY_SCALAR_FIELD(concurrent);
+}
+
 static TruncateStmt *
 _copyTruncateStmt(const TruncateStmt *from)
 {
@@ -6190,6 +6206,18 @@ _copyAlterDirectoryTableStmt(const AlterDirectoryTableStmt *from)
 	return newnode;
 }
 
+static DropDirectoryTableStmt *
+_copyDropDirectoryTableStmt(const DropDirectoryTableStmt *from)
+{
+	DropDirectoryTableStmt *newnode = makeNode(DropDirectoryTableStmt);
+
+	CopyDropStmtFields((const DropStmt *) from, (DropStmt *) newnode);
+
+	COPY_SCALAR_FIELD(with_content);
+
+	return newnode;
+}
+
 static EphemeralNamedRelationInfo*
 _copyEphemeralNamedRelationInfo(const EphemeralNamedRelationInfo *from)
 {
@@ -7351,6 +7379,10 @@ copyObjectImpl(const void *from)
 			retval = _copyAlterDirectoryTableStmt(from);
 			break;
 
+		case T_DropDirectoryTableStmt:
+			retval = _copyDropDirectoryTableStmt(from);
+			break;
+
 		case T_EphemeralNamedRelationInfo:
 			retval = _copyEphemeralNamedRelationInfo(from);
 			break;

src/backend/nodes/equalfuncs.c

@@ -3484,6 +3484,17 @@ _equalAlterDirectoryTableStmt(const AlterDirectoryTableStmt *a, const AlterDirec
 	return true;
 }
 
+static bool
+_equalDropDirectoryTableStmt(const DropDirectoryTableStmt *a, const DropDirectoryTableStmt *b)
+{
+	if (!_equalDropStmt(&a->base, &b->base))
+		return false;
+
+	COMPARE_SCALAR_FIELD(with_content);
+
+	return true;
+}
+
 /*
  * Stuff from pg_list.h
  */
@@ -4440,6 +4451,10 @@ equal(const void *a, const void *b)
 			retval = _equalAlterDirectoryTableStmt(a, b);
 			break;
 
+		case T_DropDirectoryTableStmt:
+			retval = _equalDropDirectoryTableStmt(a, b);
+			break;
+
 		default:
 			elog(ERROR, "unrecognized node type: %d",
 				 (int) nodeTag(a));

src/backend/nodes/outfast.c

@@ -1898,6 +1898,9 @@ _outNode(StringInfo str, void *obj)
 			case T_AlterDirectoryTableStmt:
 				_outAlterDirectoryTableStmt(str, obj);
 				break;
+			case T_DropDirectoryTableStmt:
+				_outDropDirectoryTableStmt(str, obj);
+				break;
 			case T_EphemeralNamedRelationInfo:
 				_outEphemeralNamedRelationInfo(str, obj);
 				break;

src/backend/nodes/outfuncs.c

@@ -4045,6 +4045,25 @@ _outAlterDirectoryTableStmt(StringInfo str, const AlterDirectoryTableStmt *node)
 	WRITE_BOOL_FIELD(unsettag);
 }
 
+static void
+_outDropStmtInfo(StringInfo str, const DropStmt *node)
+{
+	WRITE_NODE_FIELD(objects);
+	WRITE_ENUM_FIELD(removeType, ObjectType);
+	WRITE_ENUM_FIELD(behavior, DropBehavior);
+	WRITE_BOOL_FIELD(missing_ok);
+	WRITE_BOOL_FIELD(concurrent);
+}
+
+static void
+_outDropDirectoryTableStmt(StringInfo str, const DropDirectoryTableStmt *node)
+{
+	WRITE_NODE_TYPE("DROPDIRECTORYTABLESTMT");
+
+	_outDropStmtInfo(str, (const DropStmt *) node);
+	WRITE_BOOL_FIELD(with_content);
+}
+
 #include "outfuncs_common.c"
 #ifndef COMPILING_BINARY_FUNCS
 /*
@@ -5202,6 +5221,9 @@ outNode(StringInfo str, const void *obj)
 			case T_AlterDirectoryTableStmt:
 				_outAlterDirectoryTableStmt(str, obj);
 				break;
+			case T_DropDirectoryTableStmt:
+				_outDropDirectoryTableStmt(str, obj);
+				break;
 			default:
 
 				/*

src/backend/nodes/readfast.c

@@ -1774,6 +1774,33 @@ _readAlterDirectoryTableStmt(void)
 	READ_DONE();
 }
 
+static void
+_readDropStmt_common(DropStmt *local_node)
+{
+	READ_NODE_FIELD(objects);
+	READ_ENUM_FIELD(removeType,ObjectType);
+	READ_ENUM_FIELD(behavior,DropBehavior);
+	READ_BOOL_FIELD(missing_ok);
+	READ_BOOL_FIELD(concurrent);
+
+	/* Force 'missing_ok' in QEs */
+#ifdef COMPILING_BINARY_FUNCS
+	local_node->missing_ok=true;
+#endif /* COMPILING_BINARY_FUNCS */
+}
+
+static DropDirectoryTableStmt *
+_readDropDirectoryTableStmt(void)
+{
+	READ_LOCALS(DropDirectoryTableStmt);
+
+	_readDropStmt_common(&local_node->base);
+
+	READ_BOOL_FIELD(with_content);
+
+	READ_DONE();
+}
+
 static EphemeralNamedRelationInfo *
 _readEphemeralNamedRelationInfo(void)
 {
@@ -2858,6 +2885,9 @@ readNodeBinary(void)
 			case T_AlterDirectoryTableStmt:
 				return_value = _readAlterDirectoryTableStmt();
 				break;
+		case T_DropDirectoryTableStmt:
+				return_value = _readDropDirectoryTableStmt();
+				break;
 			default:
 				return_value = NULL; /* keep the compiler silent */
 				elog(ERROR, "could not deserialize unrecognized node type: %d",