- Add the "skipcol" option to Pivot tables.

modified:
  storage/connect/ha_connect.cc
  storage/connect/tabpivot.cpp
  storage/connect/tabpivot.h

storage/connect/ha_connect.cc

@@ -3916,7 +3916,7 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
   char        v, spc= ',', qch= 0;
   const char *fncn= "?";
   const char *user, *fn, *db, *host, *pwd, *sep, *tbl, *src;
-  const char *col, *ocl, *rnk, *pic, *fcl;
+  const char *col, *ocl, *rnk, *pic, *fcl, *skc;
   char       *tab, *dsn, *shm; 
 #if defined(WIN32)
   char       *nsp= NULL, *cls= NULL;
@@ -3941,7 +3941,7 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
   if (!g)
     return HA_ERR_INTERNAL_ERROR;
 
-  user= host= pwd= tbl= src= col= ocl= pic= fcl= rnk= dsn= NULL;
+  user= host= pwd= tbl= src= col= ocl= pic= fcl= skc= rnk= dsn= NULL;
 
   // Get the useful create options
   ttp= GetTypeID(topt->type);
@@ -3967,6 +3967,7 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
     ocl= GetListOption(g, "occurcol", topt->oplist, NULL);
     pic= GetListOption(g, "pivotcol", topt->oplist, NULL);
     fcl= GetListOption(g, "fnccol", topt->oplist, NULL);
+    skc= GetListOption(g, "skipcol", topt->oplist, NULL);
     rnk= GetListOption(g, "rankcol", topt->oplist, NULL);
     pwd= GetListOption(g, "password", topt->oplist);
 #if defined(WIN32)
@@ -4229,7 +4230,7 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
 
         break;
       case TAB_PIVOT:
-        qrp= PivotColumns(g, tab, src, pic, fcl, host, db, user, pwd, port);
+        qrp= PivotColumns(g, tab, src, pic, fcl, skc, host, db, user, pwd, port);
         break;
       case TAB_OEM:
         qrp= OEMColumns(g, topt, tab, (char*)db, fnc == FNC_COL);

storage/connect/tabpivot.cpp

@@ -58,11 +58,11 @@ extern "C" int trace;
 /***********************************************************************/
 PQRYRES PivotColumns(PGLOBAL g, const char *tab,   const char *src, 
                                 const char *picol, const char *fncol,
-                                const char *host,  const char *db,
-                                const char *user,  const char *pwd,
-                                int port)
+                                const char *skcol, const char *host,  
+                                const char *db,    const char *user,
+                                const char *pwd,   int port)
   {
-  PIVAID pvd(tab, src, picol, fncol, host, db, user, pwd, port);
+  PIVAID pvd(tab, src, picol, fncol, skcol, host, db, user, pwd, port);
 
   return pvd.MakePivotColumns(g);
   } // end of PivotColumns
@@ -72,10 +72,10 @@ PQRYRES PivotColumns(PGLOBAL g, const char *tab,   const char *src,
 /***********************************************************************/
 /*  PIVAID constructor.                                                */
 /***********************************************************************/
-PIVAID::PIVAID(const char *tab,   const char *src,  const char *picol,
-               const char *fncol, const char *host, const char *db,
-               const char *user,  const char *pwd,  int port)
-      : CSORT(false)
+PIVAID::PIVAID(const char *tab,   const char *src,   const char *picol,
+               const char *fncol, const char *skcol, const char *host,
+               const char *db,    const char *user,  const char *pwd,
+               int port) : CSORT(false)
   {
   Host = (char*)host;
   User = (char*)user;
@@ -86,16 +86,30 @@ PIVAID::PIVAID(const char *tab,   const char *src,  const char *picol,
   Tabsrc = (char*)src;
   Picol = (char*)picol;
   Fncol = (char*)fncol;
+  Skcol = (char*)skcol;
   Rblkp = NULL;
   Port = (port) ? port : GetDefaultPort();
   } // end of PIVAID constructor
 
+/***********************************************************************/
+/*  Skip columns that are in the skipped column list.                  */
+/***********************************************************************/
+bool PIVAID::SkipColumn(PCOLRES crp, char *skc)
+  {
+  if (skc)
+    for (char *p = skc; *p; p += (strlen(p) + 1))
+      if (!stricmp(crp->Name, p))
+        return true;
+
+  return false;
+  } // end of SkipColumn
+
 /***********************************************************************/
 /*  Make the Pivot table column list.                                  */
 /***********************************************************************/
 PQRYRES PIVAID::MakePivotColumns(PGLOBAL g)
   {
-  char    *query, *colname, buf[64];
+  char    *p, *query, *colname, *skc, buf[64];
   int      rc, ndif, nblin, w = 0;
   bool     b = false;
   PVAL     valp;   
@@ -112,6 +126,21 @@ PQRYRES PIVAID::MakePivotColumns(PGLOBAL g)
     goto err;
     } // endif rc
 
+  // Are there columns to skip?
+  if (Skcol) {
+    uint n = strlen(Skcol);
+
+    skc = (char*)PlugSubAlloc(g, NULL, n + 2);
+    strcpy(skc, Skcol);
+    skc[n + 1] = 0;
+
+    // Replace ; by nulls in skc
+    for (p = strchr(skc, ';'); p; p = strchr(p, ';'))
+      *p++ = 0;
+
+  } else
+    skc = NULL;
+
   if (!Tabsrc && Tabname) {
     // Locate the  query
     query = (char*)PlugSubAlloc(g, NULL, strlen(Tabname) + 26);
@@ -138,7 +167,7 @@ PQRYRES PIVAID::MakePivotColumns(PGLOBAL g)
 
   if (!Fncol) {
     for (crp = Qryp->Colresp; crp; crp = crp->Next)
-      if (!Picol || stricmp(Picol, crp->Name))
+      if ((!Picol || stricmp(Picol, crp->Name)) && !SkipColumn(crp, skc))
         Fncol = crp->Name;
   
     if (!Fncol) {
@@ -151,7 +180,7 @@ PQRYRES PIVAID::MakePivotColumns(PGLOBAL g)
   if (!Picol) {
     // Find default Picol as the last one not equal to Fncol
     for (crp = Qryp->Colresp; crp; crp = crp->Next)
-      if (stricmp(Fncol, crp->Name))
+      if (stricmp(Fncol, crp->Name) && !SkipColumn(crp, skc))
         Picol = crp->Name;
   
     if (!Picol) {
@@ -163,7 +192,10 @@ PQRYRES PIVAID::MakePivotColumns(PGLOBAL g)
   
   // Prepare the column list
   for (pcrp = &Qryp->Colresp; crp = *pcrp; )
-    if (!stricmp(Picol, crp->Name)) {
+    if (SkipColumn(crp, skc)) {
+      // Ignore this column
+      *pcrp = crp->Next;
+    } else if (!stricmp(Picol, crp->Name)) {
       if (crp->Nulls) {
         sprintf(g->Message, "Pivot column %s cannot be nullable", Picol);
         goto err;

storage/connect/tabpivot.h

@@ -18,12 +18,13 @@ class PIVAID : public CSORT {
   friend class SRCCOL;
  public:
   // Constructor
-  PIVAID(const char *tab,   const char *src,  const char *picol,
-         const char *fncol, const char *host, const char *db,
-         const char *user,  const char *pwd,  int port);
+  PIVAID(const char *tab,   const char *src,   const char *picol,
+         const char *fncol, const char *skcol, const char *host, 
+         const char *db,    const char *user,  const char *pwd,  int port);
 
   // Methods
   PQRYRES MakePivotColumns(PGLOBAL g);
+  bool    SkipColumn(PCOLRES crp, char *skc);
 
   // The sorting function
   virtual int  Qcompare(int *, int *);
@@ -40,6 +41,7 @@ class PIVAID : public CSORT {
   char   *Tabsrc;                   // SQL of source table
   char   *Picol;                    // Pivot column name
   char   *Fncol;                    // Function column name
+  char   *Skcol;                    // Skipped columns
   PVBLK   Rblkp;                    // The value block of the pivot column
   int     Port;                     // MySQL port number
   }; // end of class PIVAID
@@ -191,6 +193,6 @@ class SRCCOL : public PRXCOL {
 
 PQRYRES PivotColumns(PGLOBAL g, const char *tab,   const char *src,
                                 const char *picol, const char *fncol,
-                                const char *host,  const char *db,   
-                                const char *user,  const char *pwd,  
-                                int port);
+                                const char *skcol, const char *host,
+                                const char *db,    const char *user,
+                                const char *pwd,   int port);