Postgres-R: primary key patches - Mailing list pgsql-hackers

Hi,

as you might know, Postgres-R relies on primary keys to address tuples
of a table. It cannot replicate tables without a primary key.

Primary keys currently aren't really used within the executor, so I had
to extended and modify Postgres here and there, to get the required
information. To ease reviewing I have split out these modifications and
present them here as two separate little patches.

The first one, get_pkey_index_oid.diff, changes the function
relationHasPrimaryKey into GetPrimaryKeyIndexOid, which now returns an
index oid instead of just a boolean. It works pretty much the same,
except from returning an oid instead of just a boolean. (In the current
Postgres-R code, I've duplicated that code to
src/backend/replication/recovery.c)

And secondly, the add_pkey_info.diff patch adds a boolean field
ii_Primary to the IndexInfo struct and ri_PrimaryKey to the
ResultRelInfo struct, which is an index into the indexInfoArray.

I think these are relatively trivial modifications which could be
helpful for other purposes as well. So I suggest to apply them to
mainline whenever appropriate (read: choose the appropriate commit fest).

This also raises the more general question of how to start collaborating
on Postgres-R. I realize that it's a pretty huge project. However, I'm
unsure on how to ease reviewing for others, so if you have any ideas or
questions, please don't hesitate to ask.

Regards

Markus

============================================================
*** src/backend/commands/indexcmds.c    61a8b3774b682554e8670624583ab4cf4b9dbdb9
--- src/backend/commands/indexcmds.c    dc6fc2a3fbce90748bcf4cd7a60ea2ea887bc97f
*************** static Oid GetIndexOpClass(List *opclass
*** 64,70 ****
                    bool isconstraint);
  static Oid GetIndexOpClass(List *opclass, Oid attrType,
                  char *accessMethodName, Oid accessMethodId);
- static bool relationHasPrimaryKey(Relation rel);


  /*
--- 64,69 ----
*************** DefineIndex(RangeVar *heapRelation,
*** 324,330 ****
           * it's no problem either.
           */
          if (is_alter_table &&
!             relationHasPrimaryKey(rel))
          {
              ereport(ERROR,
                      (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
--- 323,329 ----
           * it's no problem either.
           */
          if (is_alter_table &&
!             (GetPrimaryKeyIndexOid(rel) != InvalidOid))
          {
              ereport(ERROR,
                      (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
*************** ChooseRelationName(const char *name1, co
*** 1216,1229 ****
  }

  /*
!  * relationHasPrimaryKey -
   *
!  *    See whether an existing relation has a primary key.
   */
! static bool
! relationHasPrimaryKey(Relation rel)
  {
!     bool        result = false;
      List       *indexoidlist;
      ListCell   *indexoidscan;

--- 1215,1229 ----
  }

  /*
!  * GetPrimaryKeyIndexOid
   *
!  * Returns the oid of the primary key index of the relation, if any,
!  * otherwise InvalidOid is returned.
   */
! Oid
! GetPrimaryKeyIndexOid(Relation rel)
  {
!     Oid            result = InvalidOid;
      List       *indexoidlist;
      ListCell   *indexoidscan;

*************** relationHasPrimaryKey(Relation rel)
*** 1244,1257 ****
                                      0, 0, 0);
          if (!HeapTupleIsValid(indexTuple))        /* should not happen */
              elog(ERROR, "cache lookup failed for index %u", indexoid);
!         result = ((Form_pg_index) GETSTRUCT(indexTuple))->indisprimary;
          ReleaseSysCache(indexTuple);
!         if (result)
              break;
      }

      list_free(indexoidlist);
-
      return result;
  }

--- 1244,1260 ----
                                      0, 0, 0);
          if (!HeapTupleIsValid(indexTuple))        /* should not happen */
              elog(ERROR, "cache lookup failed for index %u", indexoid);
!
!         if (((Form_pg_index) GETSTRUCT(indexTuple))->indisprimary)
!             result = indexoid;
!
          ReleaseSysCache(indexTuple);
!
!         if (result != InvalidOid)
              break;
      }

      list_free(indexoidlist);
      return result;
  }

============================================================
*** src/include/commands/defrem.h    e2384af33d917bff68234bbe407ea16e3ec43123
--- src/include/commands/defrem.h    58bb763402c9bef8ead035a3524505ad8fe58de5
***************
*** 15,22 ****
  #define DEFREM_H

  #include "nodes/parsenodes.h"

-
  /* commands/indexcmds.c */
  extern void DefineIndex(RangeVar *heapRelation,
              char *indexRelationName,
--- 15,22 ----
  #define DEFREM_H

  #include "nodes/parsenodes.h"
+ #include "utils/relcache.h"

  /* commands/indexcmds.c */
  extern void DefineIndex(RangeVar *heapRelation,
              char *indexRelationName,
*************** extern Oid    GetDefaultOpClass(Oid type_id
*** 43,48 ****
--- 43,49 ----
  extern char *ChooseRelationName(const char *name1, const char *name2,
                     const char *label, Oid namespace);
  extern Oid    GetDefaultOpClass(Oid type_id, Oid am_id);
+ extern Oid    GetPrimaryKeyIndexOid(Relation rel);

  /* commands/functioncmds.c */
  extern void CreateFunction(CreateFunctionStmt *stmt);
============================================================
*** src/backend/catalog/index.c    c360fcfd1002ffa557c1a376d3e74c9c2a0924db
--- src/backend/catalog/index.c    7201f06c5c1ad213a6acb6b694b666dd38358234
*************** BuildIndexInfo(Relation index)
*** 999,1004 ****
--- 999,1005 ----

      /* other info */
      ii->ii_Unique = indexStruct->indisunique;
+     ii->ii_Primary = indexStruct->indisprimary;
      ii->ii_ReadyForInserts = indexStruct->indisready;

      /* initialize index-build state to default */
============================================================
*** src/backend/executor/execUtils.c    54719433b61db70e6b433cd165c9c7a7b15e6531
--- src/backend/executor/execUtils.c    62756367e3fb34d96e0bb8c1bff4ee25f6402a4a
*************** ExecOpenIndices(ResultRelInfo *resultRel
*** 935,940 ****
--- 935,943 ----
          /* extract index key information from the index's pg_index info */
          ii = BuildIndexInfo(indexDesc);

+         if (ii->ii_Primary)
+             resultRelInfo->ri_PrimaryKey = i;
+
          relationDescs[i] = indexDesc;
          indexInfoArray[i] = ii;
          i++;
============================================================
*** src/include/nodes/execnodes.h    9c75d10763d7bedc0a4db62a319d31e1549ad542
--- src/include/nodes/execnodes.h    3bc5ba3ad42bfdb298878f69ea3538fa96ba0815
*************** typedef struct IndexInfo
*** 59,64 ****
--- 59,65 ----
      List       *ii_Predicate;    /* list of Expr */
      List       *ii_PredicateState;        /* list of ExprState */
      bool        ii_Unique;
+     bool        ii_Primary;
      bool        ii_ReadyForInserts;
      bool        ii_Concurrent;
      bool        ii_BrokenHotChain;
*************** typedef struct ResultRelInfo
*** 289,294 ****
--- 290,296 ----
      Index        ri_RangeTableIndex;
      Relation    ri_RelationDesc;
      int            ri_NumIndices;
+     int            ri_PrimaryKey;
      RelationPtr ri_IndexRelationDescs;
      IndexInfo **ri_IndexRelationInfo;
      TriggerDesc *ri_TrigDesc;