Support create directory table with location.

src/backend/commands/dirtablecmds.c

@@ -122,22 +122,45 @@ CreateDirectoryTable(CreateDirectoryTableStmt *stmt, Oid relId)
 	bool 		nulls[Natts_pg_directory_table];
 	HeapTuple	tuple;
 	char 		*dirTablePath;
-	Form_pg_class pg_class_tuple;
-	HeapTuple	class_tuple;
 	Oid 		spcId = chooseTableSpace(stmt);
-	RelFileNode relFileNode = {0};
 
-	class_tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relId));
-	if (!HeapTupleIsValid(class_tuple))
-		elog(ERROR, "cache lookup failed for relation %u", relId);
-	pg_class_tuple = (Form_pg_class) GETSTRUCT(class_tuple);
+	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
+	{
+		Form_pg_class pg_class_tuple = NULL;
+		HeapTuple	class_tuple = NULL;
+		RelFileNode relFileNode = {0};
+
+		class_tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relId));
+		if (!HeapTupleIsValid(class_tuple))
+			elog(ERROR, "cache lookup failed for relation %u", relId);
+		pg_class_tuple = (Form_pg_class) GETSTRUCT(class_tuple);
 
-	relFileNode.spcNode = spcId;
-	relFileNode.dbNode = MyDatabaseId;
-	relFileNode.relNode = pg_class_tuple->relfilenode;
-	dirTablePath = UFileFormatPathName(&relFileNode);
+		relFileNode.spcNode = spcId;
+		relFileNode.dbNode = MyDatabaseId;
+		relFileNode.relNode = pg_class_tuple->relfilenode;
 
-	ReleaseSysCache(class_tuple);
+		dirTablePath = UFileFormatPathName(&relFileNode);
+		ReleaseSysCache(class_tuple);
+	}
+
+	/*
+	 * We will check whether the directory path has existed only when use the local file am.
+	 */
+	if (UFileExists(spcId, dirTablePath) && GetTablespaceFileHandler(spcId) == &localFileAm)
+		ereport(ERROR,
+				(errcode(ERRCODE_DUPLICATE_OBJECT),
+				 errmsg("directory table path \"%s\" already exists",
+				 dirTablePath)));
 
 	/*
 	 * Acquire DirectoryTableLock to ensure that no DROP DIRECTORY TABLE

src/backend/nodes/copyfuncs.c

@@ -6290,6 +6290,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

@@ -4094,6 +4094,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

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

src/backend/parser/gram.y

@@ -416,6 +416,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
@@ -6013,6 +6014,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; }
@@ -8530,7 +8536,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;
@@ -8545,13 +8551,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;
@@ -8566,8 +8573,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;   /* dtlocation for pg_directory_table */
 } CreateDirectoryTableStmt;
 
 typedef struct AlterDirectoryTableStmt

src/test/regress/input/directory_table.source

@@ -648,6 +648,14 @@ 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 '/test_dirtable';
+CREATE DIRECTORY TABLE dir_table_with_location2 TABLESPACE directory_tblspc WITH LOCATION '/test_dirtable'; -- error
+CREATE DIRECTORY TABLE dir_table_with_location3 WITH LOCATION '/test_dirtable';
+SELECT count(*) FROM pg_directory_table;
+DROP DIRECTORY TABLE dir_table_with_location WITH CONTENT;
+DROP DIRECTORY TABLE dir_table_with_location3 WITH CONTENT;
+
 -- clean up
 DROP DIRECTORY TABLE IF EXISTS dir_table1;
 DROP DIRECTORY TABLE IF EXISTS dir_table2;

src/test/regress/output/directory_table.source

@@ -2063,6 +2063,19 @@ 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 '/test_dirtable';
+CREATE DIRECTORY TABLE dir_table_with_location2 TABLESPACE directory_tblspc WITH LOCATION '/test_dirtable'; -- error
+ERROR:  directory table path "pg_tblspc//test_dirtable" already exists
+CREATE DIRECTORY TABLE dir_table_with_location3 WITH LOCATION '/test_dirtable';
+SELECT count(*) FROM pg_directory_table;
+ count 
+-------
+     8
+(1 row)
+
+DROP DIRECTORY TABLE dir_table_with_location WITH CONTENT;
+DROP DIRECTORY TABLE dir_table_with_location3 WITH CONTENT;
 -- clean up
 DROP DIRECTORY TABLE IF EXISTS dir_table1;
 DROP DIRECTORY TABLE IF EXISTS dir_table2;

src/test/regress/output/directory_table_optimizer.source

@@ -2063,6 +2063,19 @@ 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 '/test_dirtable';
+CREATE DIRECTORY TABLE dir_table_with_location2 TABLESPACE directory_tblspc WITH LOCATION '/test_dirtable'; -- error
+ERROR:  directory table path "pg_tblspc//test_dirtable" already exists
+CREATE DIRECTORY TABLE dir_table_with_location3 WITH LOCATION '/test_dirtable';
+SELECT count(*) FROM pg_directory_table;
+ count 
+-------
+     8
+(1 row)
+
+DROP DIRECTORY TABLE dir_table_with_location WITH CONTENT;
+DROP DIRECTORY TABLE dir_table_with_location3 WITH CONTENT;
 -- clean up
 DROP DIRECTORY TABLE IF EXISTS dir_table1;
 DROP DIRECTORY TABLE IF EXISTS dir_table2;