!216 support pg_partition_filepath function

Merge pull request !216 from xiliu/GPI_BUGFIX

src/common/backend/catalog/builtin_funcs.ini

@@ -6554,6 +6554,10 @@
         "pg_partition_filenode", 1, 
         AddBuiltinFunc(_0(3197), _1("pg_partition_filenode"), _2(1), _3(true), _4(false), _5(pg_partition_filenode), _6(26), _7(PG_CATALOG_NAMESPACE), _8(BOOTSTRAP_SUPERUSERID), _9(INTERNALlanguageId), _10(1), _11(0), _12(0), _13(0), _14('f'), _15(false), _16(false), _17('s'), _18(0), _19(1, 2205), _20(NULL), _21(NULL), _22(NULL), _23(NULL), _24("pg_partition_filenode"), _25(NULL), _26(NULL), _27(NULL), _28(0), _29(false), _30(NULL), _31(false))
     ),
+    AddFuncGroup(
+        "pg_partition_filepath", 1, 
+        AddBuiltinFunc(_0(3220), _1("pg_partition_filepath"), _2(1), _3(true), _4(false), _5(pg_partition_filepath), _6(25), _7(PG_CATALOG_NAMESPACE), _8(BOOTSTRAP_SUPERUSERID), _9(INTERNALlanguageId), _10(1), _11(0), _12(0), _13(0), _14('f'), _15(false), _16(false), _17('s'), _18(0), _19(1, 2205), _20(NULL), _21(NULL), _22(NULL), _23(NULL), _24("pg_partition_filepath"), _25(NULL), _26(NULL), _27(NULL), _28(0), _29(false), _30(NULL), _31(false))
+    ),
     AddFuncGroup(
         "pg_partition_indexes_size", 2, 
         AddBuiltinFunc(_0(3195), _1("pg_partition_indexes_size"), _2(2), _3(true), _4(false), _5(pg_partition_indexes_size_name), _6(20), _7(PG_CATALOG_NAMESPACE), _8(BOOTSTRAP_SUPERUSERID), _9(INTERNALlanguageId), _10(1), _11(0), _12(0), _13(0), _14('f'), _15(false), _16(false), _17('v'), _18(0), _19(2, 25, 25), _20(NULL), _21(NULL), _22(NULL), _23(NULL), _24("pg_partition_indexes_size_name"), _25(NULL), _26(NULL), _27(NULL), _28(0), _29(false), _30(NULL), _31(false)),

src/common/backend/utils/adt/dbsize.cpp

@@ -1764,6 +1764,62 @@ Datum pg_partition_filenode(PG_FUNCTION_ARGS)
     PG_RETURN_OID(result);
 }
 
+/*
+ * Get the pathname (relative to $PGDATA) of a partition
+ *
+ */
+Datum pg_partition_filepath(PG_FUNCTION_ARGS)
+{
+    Oid partRelId = PG_GETARG_OID(0);
+    HeapTuple tuple;
+    Form_pg_partition partRelForm;
+    RelFileNode rnode;
+    BackendId backend = InvalidBackendId;
+    char* path = NULL;
+
+    tuple = SearchSysCache1(PARTRELID, ObjectIdGetDatum(partRelId));
+    if (!HeapTupleIsValid(tuple))
+        PG_RETURN_NULL();
+    partRelForm = (Form_pg_partition)GETSTRUCT(tuple);
+
+    switch (partRelForm->parttype) {
+        case PARTTYPE_PARTITIONED_RELATION:
+        case PARTTYPE_VALUE_PARTITIONED_RELATION:
+            rnode.spcNode = ConvertToRelfilenodeTblspcOid(partRelForm->reltablespace);
+            if (rnode.spcNode == GLOBALTABLESPACE_OID) {
+                rnode.dbNode = InvalidOid;
+            } else {
+                rnode.dbNode = u_sess->proc_cxt.MyDatabaseId;
+            }
+            if (partRelForm->relfilenode) {
+                rnode.relNode = partRelForm->relfilenode;
+            } else {
+                /* Consult the relation mapper */
+                rnode.relNode = RelationMapOidToFilenode(partRelId, false);
+            }
+            rnode.bucketNode = InvalidBktId;
+            break;
+
+        default:
+            /* no storage, return NULL */
+            rnode.relNode = InvalidOid;
+            /* some compilers generate warnings without these next two lines */
+            rnode.dbNode = InvalidOid;
+            rnode.spcNode = InvalidOid;
+            rnode.bucketNode = InvalidBktId;
+            break;
+    }
+
+    if (!OidIsValid(rnode.relNode)) {
+        ReleaseSysCache(tuple);
+        PG_RETURN_NULL();
+    }
+
+    ReleaseSysCache(tuple);
+    path = relpathbackend(rnode, backend, MAIN_FORKNUM);
+    PG_RETURN_TEXT_P(cstring_to_text(path));
+}
+
 /*
  * Get the pathname (relative to $PGDATA) of a relation
  *

src/include/utils/builtins.h

@@ -551,6 +551,7 @@ extern Datum pg_partition_indexes_size(PG_FUNCTION_ARGS);
 extern Datum pg_indexes_size(PG_FUNCTION_ARGS);
 extern Datum pg_relation_filenode(PG_FUNCTION_ARGS);
 extern Datum pg_filenode_relation(PG_FUNCTION_ARGS);
+extern Datum pg_partition_filepath(PG_FUNCTION_ARGS);
 extern Datum pg_relation_filepath(PG_FUNCTION_ARGS);
 #ifdef PGXC
 extern Datum pg_relation_with_compression(PG_FUNCTION_ARGS);

src/test/regress/expected/hw_functions.out

@@ -426,4 +426,54 @@ select * from test_numeric order by id;
 -- drop table at last
 drop table test_null_repeat;
 drop table test_numeric;
+--test function pg_partition_filepath
+CREATE TABLE test_func_partition_filepath_table
+(
+    c1 int,
+    c2 int,
+    c3 int
+)
+partition by range (c1)
+(
+    partition p0_partition_filepath_table1 VALUES less than (10000),
+    partition p1_partition_filepath_table1 VALUES less than (20000),
+    partition p2_partition_filepath_table1 VALUES less than (30000),
+    partition p3_partition_filepath_table1 VALUES less than (maxvalue)
+);
+create or replace function func_get_partition_filepath(partname text) returns text as $$ 
+declare
+    partoid integer;
+    filepath text;
+begin
+    select oid from pg_partition where relname = partname into partoid;
+    select * from pg_partition_filepath(partoid) into filepath;
+    return filepath;
+end;
+$$ language plpgsql;
+select func_get_partition_filepath('p0_partition_filepath_table1');
+--? func_get_partition_filepath 
+--?.*
+--? base/.*/.*
+(1 row)
+
+select func_get_partition_filepath('p1_partition_filepath_table1');
+--? func_get_partition_filepath 
+--?.*
+--?.base/.*/.*
+(1 row)
+
+select func_get_partition_filepath('p2_partition_filepath_table1');
+--? func_get_partition_filepath 
+--?.*
+--? base/.*/.*
+(1 row)
+
+select func_get_partition_filepath('p3_partition_filepath_table1');
+--? func_get_partition_filepath 
+--?.*
+--? base/.*/.*
+(1 row)
+
+drop function func_get_partition_filepath;
+drop table test_func_partition_filepath_table;
 drop schema basefunc cascade;

src/test/regress/expected/opr_sanity.out

@@ -1915,6 +1915,7 @@ WHERE d.classoid IS NULL AND p1.oid <= 9999 order by 1;
  3217 | cupointer_bigint
  3218 | pg_stat_bad_block
  3219 | pg_stat_bad_block_clear
+ 3220 | pg_partition_filepath
  3221 | pg_stat_get_tuples_changed
  3222 | pg_total_autovac_tuples
  3223 | pg_autovac_status
@@ -2638,7 +2639,7 @@ WHERE d.classoid IS NULL AND p1.oid <= 9999 order by 1;
  9016 | pg_advisory_lock
  9017 | pgxc_unlock_for_sp_database
  9999 | pg_test_err_contain_err
-(2276 rows)
+(2277 rows)
 
 -- **************** pg_cast ****************
 -- Catch bogus values in pg_cast columns (other than cases detected by

src/test/regress/expected/single_node_opr_sanity.out

@@ -1954,6 +1954,7 @@ WHERE d.classoid IS NULL AND p1.oid <= 9999 order by 1;
  3217 | cupointer_bigint
  3218 | pg_stat_bad_block
  3219 | pg_stat_bad_block_clear
+ 3220 | pg_partition_filepath
  3221 | pg_stat_get_tuples_changed
  3222 | pg_total_autovac_tuples
  3223 | pg_autovac_status
@@ -2677,7 +2678,7 @@ WHERE d.classoid IS NULL AND p1.oid <= 9999 order by 1;
  9016 | pg_advisory_lock
  9017 | pgxc_unlock_for_sp_database
  9999 | pg_test_err_contain_err
-(2276 rows)
+(2277 rows)
 
 -- Check prokind
 select count(*) from pg_proc where prokind = 'a';
@@ -2695,7 +2696,7 @@ select count(*) from pg_proc where prokind = 'w';
 select count(*) from pg_proc where prokind = 'f';
  count 
 -------
-  3150
+  3151
 (1 row)
 
 select count(*) from pg_proc where prokind = 'p';

src/test/regress/sql/hw_functions.sql

@@ -197,4 +197,38 @@ select * from test_numeric order by id;
 -- drop table at last
 drop table test_null_repeat;
 drop table test_numeric;
+
+
+--test function pg_partition_filepath
+CREATE TABLE test_func_partition_filepath_table
+(
+    c1 int,
+    c2 int,
+    c3 int
+)
+partition by range (c1)
+(
+    partition p0_partition_filepath_table1 VALUES less than (10000),
+    partition p1_partition_filepath_table1 VALUES less than (20000),
+    partition p2_partition_filepath_table1 VALUES less than (30000),
+    partition p3_partition_filepath_table1 VALUES less than (maxvalue)
+);
+
+create or replace function func_get_partition_filepath(partname text) returns text as $$ 
+declare
+    partoid integer;
+    filepath text;
+begin
+    select oid from pg_partition where relname = partname into partoid;
+    select * from pg_partition_filepath(partoid) into filepath;
+    return filepath;
+end;
+$$ language plpgsql;
+select func_get_partition_filepath('p0_partition_filepath_table1');
+select func_get_partition_filepath('p1_partition_filepath_table1');
+select func_get_partition_filepath('p2_partition_filepath_table1');
+select func_get_partition_filepath('p3_partition_filepath_table1');
+drop function func_get_partition_filepath;
+drop table test_func_partition_filepath_table;
+
 drop schema basefunc cascade;