Stephen,
The attached patch fixes the matter.
It fixes up attribute number of child relation when it is extracted.
(*) Injected elog()s are removed.
postgres=# select * from t1;
NOTICE: markRTEForSelectPriv: ACL_SELECT on t1.a
NOTICE: markRTEForSelectPriv: ACL_SELECT on t1.c
NOTICE: ExecCheckRTEPerms: ACL_SELECT on t1 perms = 0000 inh = 1
NOTICE: ExecCheckRTEPerms: selectedCols: t1.a
NOTICE: ExecCheckRTEPerms: selectedCols: t1.c
NOTICE: ExecCheckRTEPerms: ACL_SELECT on t1 perms = 0002 inh = 0
NOTICE: ExecCheckRTEPerms: selectedCols: t1.a
NOTICE: ExecCheckRTEPerms: selectedCols: t1.c
NOTICE: ExecCheckRTEPerms: ACL_SELECT on t2 perms = 0002 inh = 0
NOTICE: ExecCheckRTEPerms: selectedCols: t2.a
NOTICE: ExecCheckRTEPerms: selectedCols: t2.c
a | c
---+---
(0 rows)
postgres=# select t1 from t1;
NOTICE: markRTEForSelectPriv: ACL_SELECT on t1.t1
NOTICE: ExecCheckRTEPerms: ACL_SELECT on t1 perms = 0000 inh = 1
NOTICE: ExecCheckRTEPerms: selectedCols: t1.t1
NOTICE: ExecCheckRTEPerms: ACL_SELECT on t1 perms = 0002 inh = 0
NOTICE: ExecCheckRTEPerms: selectedCols: t1.t1
NOTICE: ExecCheckRTEPerms: ACL_SELECT on t2 perms = 0002 inh = 0
NOTICE: ExecCheckRTEPerms: selectedCols: t2.a
NOTICE: ExecCheckRTEPerms: selectedCols: t2.c
t1
----
(0 rows)
KaiGai Kohei wrote:
> KaiGai Kohei wrote:
>> I've observed the behavior of column-level privileges and
>> required permissions with a few elog()s injected.
>>
>> I noticed rte->selectedCols is incorrect when we make a query
>> on inherited tables.
>>
>> See below:
>> -------------------------------------------------
>> postgres=# CREATE TABLE t1 (a int, b int, c int);
>> CREATE TABLE
>> postgres=# ALTER TABLE t1 DROP COLUMN b;
>> ALTER TABLE
>> postgres=# CREATE TABLE t2 (d int) inherits (t1);
>> CREATE TABLE
>> postgres=# SELECT * FROM t1;
>> NOTICE: markRTEForSelectPriv: ACL_SELECT on t1.a
>> NOTICE: markRTEForSelectPriv: ACL_SELECT on t1.c
>> NOTICE: ExecCheckRTEPerms: ACL_SELECT on t1 perms = 0000 inh = 1
>> NOTICE: ExecCheckRTEPerms: selectedCols: t1.a
>> NOTICE: ExecCheckRTEPerms: selectedCols: t1.c
>> NOTICE: ExecCheckRTEPerms: ACL_SELECT on t1 perms = 0002 inh = 0
>> NOTICE: ExecCheckRTEPerms: selectedCols: t1.a
>> NOTICE: ExecCheckRTEPerms: selectedCols: t1.c
>> NOTICE: ExecCheckRTEPerms: ACL_SELECT on t2 perms = 0002 inh = 0
>> NOTICE: ExecCheckRTEPerms: selectedCols: t2.a
>> NOTICE: ExecCheckRTEPerms: selectedCols: t2.d <--- (*)
>> a | c
>> ---+---
>> (0 rows)
>> -------------------------------------------------
>>
>> I injected elog() at the head of ExecCheckRTEPerms() to print requiredPerms
>> and all the columns on selectedCols/modifiedCols.
>>
>> It seems to me the current implementation assumes the parant table and
>> child table have same set of attribute name/number pair, but incorrect.
>> It is necessary to lookup attribute names of "t2" when we extract
>> inherited tables.
>
> In addition, the whole-row-reference can be problematic.
>
> When we run "SELECT t1 FROM t1", the column level privilege tries to check
> all the columns within t1 and t2. But, I think it should not check on "t2.d"
> in this case, because the column is not a target of this query.
>
> In the whole-row-reference case, attno==0 on the parent table should be
> extracted into correct set of columns on the children.
>
> Thanks,
--
OSS Platform Development Division, NEC
KaiGai Kohei <kaigai@ak.jp.nec.com>
*** base/src/backend/optimizer/prep/prepunion.c 2009-02-26 11:04:20.000000000 +0900
--- sepgsql/src/backend/optimizer/prep/prepunion.c 2009-03-05 16:24:32.000000000 +0900
***************
*** 30,35 ****
--- 30,36 ----
#include "access/heapam.h"
+ #include "access/sysattr.h"
#include "catalog/namespace.h"
#include "catalog/pg_type.h"
#include "miscadmin.h"
***************
*** 49,54 ****
--- 50,56 ----
#include "utils/lsyscache.h"
#include "utils/rel.h"
#include "utils/selfuncs.h"
+ #include "utils/syscache.h"
static Plan *recurse_set_operations(Node *setOp, PlannerInfo *root,
***************
*** 1150,1155 ****
--- 1152,1253 ----
}
/*
+ * fixup_column_privileges
+ * Inherited tables have same columns as its parents have,
+ * but these columns may have different attribute numbers,
+ * so we need to lookup attribute numbers of child relation
+ * by its name.
+ */
+ static Bitmapset *
+ fixup_column_privileges(Oid parent, Oid child, Bitmapset *oldmap)
+ {
+ Bitmapset *newmap = NULL;
+ char *attname;
+ int attno, attno_fixup;
+
+ if (!oldmap || parent == child)
+ return oldmap; /* no need to fixup */
+
+ while ((attno = bms_first_member(oldmap)) >= 0)
+ {
+ /* remove the column number offset */
+ attno += FirstLowInvalidHeapAttributeNumber;
+
+ /*
+ * The whole-row-reference case need a special care
+ * because child relation has more columns than the
+ * parent, so it need to extract inherited columns
+ * only.
+ */
+ if (attno == InvalidAttrNumber)
+ {
+ HeapTuple reltup;
+ HeapTuple atttup;
+ Form_pg_class classForm;
+ Form_pg_attribute attForm;
+ int nattrs;
+
+ reltup = SearchSysCache(RELOID,
+ ObjectIdGetDatum(parent),
+ 0, 0, 0);
+ if (!HeapTupleIsValid(reltup))
+ elog(ERROR, "cache lookup failed for relation %u", parent);
+ classForm = (Form_pg_class) GETSTRUCT(reltup);
+
+ nattrs = classForm->relnatts;
+
+ ReleaseSysCache(reltup);
+
+ for (attno = 1; attno <= nattrs; attno++)
+ {
+ atttup = SearchSysCache(ATTNUM,
+ ObjectIdGetDatum(parent),
+ Int16GetDatum(attno),
+ 0, 0);
+ if (!HeapTupleIsValid(atttup))
+ continue;
+
+ attForm = (Form_pg_attribute) GETSTRUCT(atttup);
+
+ if (attForm->attisdropped)
+ {
+ ReleaseSysCache(atttup);
+ continue;
+ }
+
+ attno_fixup = get_attnum(child, NameStr(attForm->attname));
+ if (attno_fixup == InvalidAttrNumber)
+ elog(ERROR, "cache lookup failed for attribute %s of relation %u",
+ NameStr(attForm->attname), child);
+
+ /* add the column number offset */
+ attno_fixup -= FirstLowInvalidHeapAttributeNumber;
+ newmap = bms_add_member(newmap, attno_fixup);
+
+ ReleaseSysCache(atttup);
+ }
+ continue;
+ }
+
+ attname = get_attname(parent, attno);
+ if (!attname)
+ elog(ERROR, "cache lookup failed for attribute %d of relation %u",
+ attno, parent);
+
+ attno_fixup = get_attnum(child, attname);
+ if (attno_fixup == InvalidAttrNumber)
+ elog(ERROR, "cache lookup failed for attribute %s of relation %u",
+ attname, child);
+
+ /* add the column number offset */
+ attno_fixup -= FirstLowInvalidHeapAttributeNumber;
+ newmap = bms_add_member(newmap, attno_fixup);
+ }
+
+ return newmap;
+ }
+
+ /*
* expand_inherited_rtentry
* Check whether a rangetable entry represents an inheritance set.
* If so, add entries for all the child tables to the query's
***************
*** 1277,1282 ****
--- 1375,1384 ----
* and set inh = false.
*/
childrte = copyObject(rte);
+ childrte->selectedCols
+ = fixup_column_privileges(rte->relid, childOID, childrte->selectedCols);
+ childrte->modifiedCols
+ = fixup_column_privileges(rte->relid, childOID, childrte->modifiedCols);
childrte->relid = childOID;
childrte->inh = false;
parse->rtable = lappend(parse->rtable, childrte);