Support create directory table with location.

When we create directory table with location, the file directory path will concat
with location at beginning.

Authored-by: Zhang Wenchao [email protected]

src/backend/commands/dirtablecmds.c

@@ -135,7 +135,24 @@ CreateDirectoryTable(CreateDirectoryTableStmt *stmt, Oid relId)
 	relFileNode.spcNode = spcId;
 	relFileNode.dbNode = MyDatabaseId;
 	relFileNode.relNode = pg_class_tuple->relfilenode;
-	dirTablePath = UFileFormatPathName(&relFileNode);
+	if (stmt->location)
+	{
+		if (spcId == InvalidOid ||
+			spcId == DEFAULTTABLESPACE_OID)
+			dirTablePath = psprintf("base/%s", stmt->location);
+		else if (spcId == GLOBALTABLESPACE_OID)
+			dirTablePath = psprintf("global/%s", stmt->location);
+		else
+			dirTablePath = psprintf("pg_tblspc/%s", stmt->location);
+	}
+	else
+		dirTablePath = UFileFormatPathName(&relFileNode);
+
+	if (UFileExists(spcId, dirTablePath) && GetTablespaceFileHandler(spcId) == &localFileAm)
+		ereport(ERROR,
+				(errcode(ERRCODE_DUPLICATE_OBJECT),
+				 errmsg("directory table path \"%s\" already exists",
+				 dirTablePath)));
 
 	ReleaseSysCache(class_tuple);
 

src/backend/nodes/copyfuncs.c

@@ -6215,6 +6215,7 @@ _copyCreateDirectoryTableStmt(const CreateDirectoryTableStmt *from)
 	CopyCreateStmtFields((const CreateStmt *) from, (CreateStmt *) newnode);
 
 	COPY_STRING_FIELD(tablespacename);
+	COPY_STRING_FIELD(location);
 
 	return newnode;
 }

src/backend/nodes/equalfuncs.c

@@ -3474,6 +3474,7 @@ _equalCreateDirectoryTableStmt(const CreateDirectoryTableStmt *a, const CreateDi
 		return false;
 
 	COMPARE_STRING_FIELD(tablespacename);
+	COMPARE_STRING_FIELD(location);
 
 	return true;
 }

src/backend/nodes/outfuncs.c

@@ -4035,6 +4035,7 @@ _outCreateDirectoryTableStmt(StringInfo str, const CreateDirectoryTableStmt *nod
 
 	_outCreateStmtInfo(str, (const CreateStmt *) node);
 	WRITE_STRING_FIELD(tablespacename);
+	WRITE_STRING_FIELD(location);
 }
 
 static void

src/backend/nodes/readfast.c

@@ -1758,6 +1758,7 @@ _readCreateDirectoryTableStmt(void)
 	_readCreateStmt_common(&local_node->base);
 
 	READ_STRING_FIELD(tablespacename);
+	READ_STRING_FIELD(location);
 
 	READ_DONE();
 }

src/backend/parser/gram.y

@@ -415,6 +415,7 @@ static void check_expressions_in_partition_key(PartitionSpec *spec, core_yyscan_
 				access_method_clause attr_name
 				table_access_method_clause name cursor_name file_name
 				opt_index_name cluster_index_specification opt_file_name
+%type <str>		OptWithLocation
 
 %type <list>	func_name handler_name qual_Op qual_all_Op subquery_Op
 				opt_class opt_inline_handler opt_validator validator_clause
@@ -6006,6 +6007,11 @@ OptWith:
 			| /*EMPTY*/					{ $$ = NIL; }
 		;
 
+OptWithLocation:
+			WITH LOCATION Sconst		{ $$ = $3; }
+			| /*EMPTY*/					{ $$ = NULL; }
+		;
+
 OnCommitOption:  ON COMMIT DROP				{ $$ = ONCOMMIT_DROP; }
 			| ON COMMIT DELETE_P ROWS		{ $$ = ONCOMMIT_DELETE_ROWS; }
 			| ON COMMIT PRESERVE ROWS		{ $$ = ONCOMMIT_PRESERVE_ROWS; }
@@ -8523,7 +8529,7 @@ AlterStorageUserMappingStmt:
 
 CreateDirectoryTableStmt:
             CREATE DIRECTORY TABLE qualified_name
-            table_access_method_clause OptTableSpace OptTagOptList
+            table_access_method_clause OptTableSpace OptWithLocation OptTagOptList
                 {
                     CreateDirectoryTableStmt *n = makeNode(CreateDirectoryTableStmt);
                     $4->relpersistence = RELPERSISTENCE_PERMANENT;
@@ -8538,13 +8544,14 @@ CreateDirectoryTableStmt:
                     n->base.if_not_exists = false;
                     n->base.distributedBy = GetDirectoryTableDistributedBy();
                     n->base.relKind = RELKIND_DIRECTORY_TABLE;
-                    n->base.tags = $7;
                     n->tablespacename = $6;
+                    n->location = $7;
+                    n->base.tags = $8;
 
                     $$ = (Node *) n;
                 }
             | CREATE DIRECTORY TABLE IF_P NOT EXISTS qualified_name
-            table_access_method_clause OptTableSpace OptTagOptList
+            table_access_method_clause OptTableSpace OptWithLocation OptTagOptList
                 {
                     CreateDirectoryTableStmt *n = makeNode(CreateDirectoryTableStmt);
                     $7->relpersistence = RELPERSISTENCE_PERMANENT;
@@ -8559,8 +8566,9 @@ CreateDirectoryTableStmt:
                     n->base.if_not_exists = true;
                     n->base.distributedBy = GetDirectoryTableDistributedBy();
                     n->base.relKind = RELKIND_DIRECTORY_TABLE;
-                    n->base.tags = $10;
                     n->tablespacename = $9;
+                    n->location = $10;
+                    n->base.tags = $11;
 
                     $$ = (Node *) n;
                 }

src/include/nodes/parsenodes.h

@@ -3412,6 +3412,7 @@ typedef struct CreateDirectoryTableStmt
 {
 	CreateStmt	base;
 	char	   *tablespacename;
+	char       *location;
 } CreateDirectoryTableStmt;
 
 typedef struct AlterDirectoryTableStmt

src/test/regress/input/directory_table.source

@@ -648,6 +648,12 @@ SELECT relative_path, tag FROM dir_table4 ORDER BY 1;
 SAVEPOINT s2;
 ROLLBACK;
 
+-- test create directory table with location
+CREATE DIRECTORY TABLE dir_table_with_location TABLESPACE directory_tblspc WITH LOCATION '@testtablespace@/test_dirtable';
+CREATE DIRECTORY TABLE dir_table_with_location2 WITH LOCATION '@testtablespace@/test_dirtable';	-- error
+SELECT * FROM pg_directory_table ORDER BY 1;
+DROP DIRECTORY TABLE dir_table_with_location WITH CONTENT;
+
 -- clean up
 DROP DIRECTORY TABLE IF EXISTS dir_table1;
 DROP DIRECTORY TABLE IF EXISTS dir_table2;