Fix direcotory table problems and add guc allow_dml_directory_table.

In this commit, fix gpdirtable can not specify file name. We can specify
file name using two match mode, full match and regex match. Full match
using for download one file while regex match using for downloading files
in directory. What's more, we add new guc allow_dml_directory_table, which
allow us to do dml on directory table's schema table. Lastly, we support a
new grammar for drop directory table. We don't remove directory table's
directory by default, while we can remove it by using 'with content' keywords
in drop directory table grammar, for example:

DROP DIRECTORY TABLE dir_table WITH CONTENT;

Authored-by: Zhang Wenchao [email protected]

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",