- Extending connect_assisted_discovery column automatic definition

  to OCCUR and PIVOT table types.

modified:
  storage/connect/ha_connect.cc
  storage/connect/myconn.cpp
  storage/connect/myconn.h
  storage/connect/plgdbsem.h
  storage/connect/plgdbutl.cpp
  storage/connect/taboccur.cpp
  storage/connect/taboccur.h
  storage/connect/tabpivot.cpp
  storage/connect/tabpivot.h

- Fix wrong definition of GetVlen for TYPE template

modified:
  storage/connect/valblk.h

10 files changed, 494 insertions, 47 deletions

diff --git a/storage/connect/ha_connect.cc b/storage/connect/ha_connect.cc
index ba96f99e28a..db789dda5bd 100644
--- a/storage/connect/ha_connect.cc
+++ b/storage/connect/ha_connect.cc
@@ -27,11 +27,12 @@
   ha_connect will let you create/open/delete tables, the created table can be
   done specifying an already existing file, the drop table command will just
   suppress the table definition but not the eventual data file.
-  Indexes are not yet supported but data can be inserted, updated or deleted.
+  Indexes are not supported for all table types but data can be inserted, 
+  updated or deleted.
 
   You can enable the CONNECT storage engine in your build by doing the
   following during your build process:<br> ./configure
-  --with-connect-storage-engine (not implemented yet)
+  --with-connect-storage-engine
 
   You can install the CONNECT handler as all other storage handlers.
 
@@ -166,6 +167,16 @@ extern "C" {
        int  trace= 0;              // The general trace value
 } // extern "C"
 
+bool OcrColumns(PGLOBAL g, PQRYRES qrp, const char *col, 
+                       const char *ocr, const char *rank);
+bool OcrSrcCols(PGLOBAL g, PQRYRES qrp, const char *col, 
+                       const char *ocr, const char *rank);
+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);
+
 /****************************************************************************/
 /*  Initialize the ha_connect static members.                               */
 /****************************************************************************/
@@ -3333,7 +3344,7 @@ static char *encode(PGLOBAL g, char *cnm)
 */
 
 static bool add_field(String *sql, const char *field_name, const char *type,
-                      int len, int dec, uint tm, const char *rem)
+                      int len, int dec, uint tm, const char *rem, int flag)
 {
   bool error= false;
 
@@ -3341,15 +3352,18 @@ static bool add_field(String *sql, const char *field_name, const char *type,
   error|= sql->append(field_name);
   error|= sql->append("` ");
   error|= sql->append(type);
+
   if (len) {
     error|= sql->append('(');
     error|= sql->append_ulonglong(len);
+
     if (dec || !strcmp(type, "DOUBLE")) {
       error|= sql->append(',');
       error|= sql->append_ulonglong(dec);
-      }
+      } // endif dec
+
     error|= sql->append(')');
-    }
+    } // endif len
   
   if (tm)
     error|= sql->append(STRING_WITH_LEN(" NOT NULL"), system_charset_info);
@@ -3358,10 +3372,14 @@ static bool add_field(String *sql, const char *field_name, const char *type,
     error|= sql->append(" COMMENT '");
     error|= sql->append_for_single_quote(rem, strlen(rem));
     error|= sql->append("'");
-    }
+    } // endif rem
 
-  sql->append(',');
+  if (flag) {
+    error|= sql->append(" FLAG=");
+    error|= sql->append_ulonglong(flag);
+    } // endif flag
 
+  sql->append(',');
   return error;
 } // end of add_field
 
@@ -3381,6 +3399,7 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
   char        spc= ',', qch= 0;
   const char *fncn= "?";
   const char *user, *fn, *db, *host, *pwd, *prt, *sep, *tbl, *src;
+  const char *col, *ocl, *rnk, *pic, *fcl;
   char       *tab, *dsn; 
 #if defined(WIN32)
   char       *nsp= NULL, *cls= NULL;
@@ -3402,7 +3421,7 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
 
   sql.copy(STRING_WITH_LEN("CREATE TABLE whatever ("), system_charset_info);
 
-  user= host= pwd= prt= tbl= src= dsn= NULL;
+  user= host= pwd= prt= tbl= src= col= ocl= pic= fcl= rnk= dsn= NULL;
 
   // Get the useful create options
   ttp= GetTypeID(topt->type);
@@ -3417,12 +3436,18 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
   qch= topt->qchar ? *topt->qchar : topt->quoted >= 0 ? '"' : 0;
   hdr= (int)topt->header;
   tbl= topt->tablist;
+  col= topt->colist;
 
   if (topt->oplist) {
     host= GetListOption(g,"host", topt->oplist, "localhost");
     user= GetListOption(g,"user", topt->oplist, "root");
     // Default value db can come from the DBNAME=xxx option.
     db= GetListOption(g,"database", topt->oplist, db);
+    col= GetListOption(g,"colist", topt->oplist, col);
+    ocl= GetListOption(g,"occurcol", topt->oplist, NULL);
+    pic= GetListOption(g,"pivotcol", topt->oplist, NULL);
+    fcl= GetListOption(g,"fnccol", topt->oplist, NULL);
+    rnk= GetListOption(g,"rankcol", topt->oplist, NULL);
     pwd= GetListOption(g,"password", topt->oplist);
     prt= GetListOption(g,"port", topt->oplist);
     port= (prt) ? atoi(prt) : 0;
@@ -3537,9 +3562,12 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
       ok= true;
       break;
 #endif   // WIN32
+    case TAB_PIVOT:
+      supfnc = FNC_NO;
     case TAB_PRX:
     case TAB_TBL:
     case TAB_XCL:
+    case TAB_OCCUR:
       ok= true;
       break;
     default:
@@ -3563,7 +3591,7 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
 
   if (ok) {
     char *cnm, *rem;
-    int   i, len, dec, typ;
+    int   i, len, dec, typ, flg;
     const char *type;
     PDBUSER dup= PlgGetUser(g);
     PCATLG  cat= (dup) ? dup->Catalog : NULL;
@@ -3573,9 +3601,16 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
     else
       return HA_ERR_INTERNAL_ERROR;           // Should never happen
 
-    if (src)
+    if (src && ttp != TAB_PIVOT) {
       qrp= SrcColumns(g, host, db, user, pwd, src, port);
-    else switch (ttp) {
+
+      if (ttp == TAB_OCCUR)
+        if (OcrSrcCols(g, qrp, col, ocl, rnk)) {
+          my_message(ER_UNKNOWN_ERROR, g->Message, MYF(0));
+          return HA_ERR_INTERNAL_ERROR;
+          } // endif OcrSrcCols
+
+    } else switch (ttp) {
       case TAB_DBF:
         qrp= DBFColumns(g, fn, fnc == FNC_COL);
         break;
@@ -3618,12 +3653,22 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
       case TAB_PRX:
       case TAB_TBL:
       case TAB_XCL:
+      case TAB_OCCUR:
         bif= fnc == FNC_COL;
         qrp= TabColumns(g, thd, db, tab, bif);
 
         if (!qrp && bif && fnc != FNC_COL)         // tab is a view
           qrp= MyColumns(g, host, db, user, pwd, tab, NULL, port, false);
 
+        if (ttp == TAB_OCCUR && fnc != FNC_COL)
+          if (OcrColumns(g, qrp, col, ocl, rnk)) {
+            my_message(ER_UNKNOWN_ERROR, g->Message, MYF(0));
+            return HA_ERR_INTERNAL_ERROR;
+            } // endif OcrColumns
+
+        break;
+      case TAB_PIVOT:
+        qrp= PivotColumns(g, tab, src, pic, fcl, host, db, user, pwd, port);
         break;
       default:
         strcpy(g->Message, "System error during assisted discovery");
@@ -3635,16 +3680,17 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
       return HA_ERR_INTERNAL_ERROR;
       } // endif qrp
 
-    if (fnc != FNC_NO || src) {
-      // Catalog table
+    if (fnc != FNC_NO || src || ttp == TAB_PIVOT) {
+      // Catalog like table
       for (crp=qrp->Colresp; !b && crp; crp= crp->Next) {
         cnm= encode(g, crp->Name);
         type= PLGtoMYSQLtype(crp->Type, dbf);
         len= crp->Length;
         dec= crp->Prec;
+        flg= crp->Flag;
      
         // Now add the field
-        if (add_field(&sql, cnm, type, len, dec, NOT_NULL_FLAG, 0))
+        if (add_field(&sql, cnm, type, len, dec, NOT_NULL_FLAG, 0, flg))
           b= HA_ERR_OUT_OF_MEM;
         } // endfor crp
 
@@ -3714,7 +3760,7 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
           len= 0;
 
         // Now add the field
-        if (add_field(&sql, cnm, type, len, dec, tm, rem))
+        if (add_field(&sql, cnm, type, len, dec, tm, rem, 0))
           b= HA_ERR_OUT_OF_MEM;
 
         } // endfor i
diff --git a/storage/connect/myconn.cpp b/storage/connect/myconn.cpp
index 220be7ca0be..9ebf77ff35a 100644
--- a/storage/connect/myconn.cpp
+++ b/storage/connect/myconn.cpp
@@ -64,6 +64,12 @@ static char *server_groups[] = {
 extern "C" int   trace;
 extern MYSQL_PLUGIN_IMPORT uint mysqld_port;
 
+// Returns the current used port
+uint GetDefaultPort(void)
+{
+  return mysqld_port;
+} // end of GetDefaultPort
+
 /************************************************************************/
 /*  MyColumns: constructs the result blocks containing all columns      */
 /*  of a MySQL table or view.                                           */
@@ -673,6 +679,7 @@ PQRYRES MYSQLC::GetResult(PGLOBAL g, bool pdb)
     *pcrp = (PCOLRES)PlugSubAlloc(g, NULL, sizeof(COLRES));
     crp = *pcrp;
     pcrp = &crp->Next;
+    memset(crp, 0, sizeof(COLRES));
     crp->Ncol = ++qrp->Nbcol;
 
     crp->Name = (char*)PlugSubAlloc(g, NULL, fld->name_length + 1);
@@ -686,7 +693,7 @@ PQRYRES MYSQLC::GetResult(PGLOBAL g, bool pdb)
       // For direct MySQL connection, display the MySQL date string
       crp->Type = TYPE_STRING;
 
-    crp->Prec = fld->decimals;
+    crp->Prec = (crp->Type == TYPE_FLOAT) ? fld->decimals : 0;
     crp->Length = fld->max_length;
     crp->Clen = GetTypeSize(crp->Type, crp->Length);
 
diff --git a/storage/connect/myconn.h b/storage/connect/myconn.h
index 8148630b812..f8c8c3dcbae 100644
--- a/storage/connect/myconn.h
+++ b/storage/connect/myconn.h
@@ -44,6 +44,8 @@ PQRYRES SrcColumns(PGLOBAL g, const char *host, const char *db,
                    const char *user, const char *pwd,
                    const char *srcdef, int port);
 
+uint GetDefaultPort(void);
+
 /* -------------------------- MYCONN class --------------------------- */
 
 /***********************************************************************/
diff --git a/storage/connect/plgdbsem.h b/storage/connect/plgdbsem.h
index dfa46a650a6..919f3452a4d 100644
--- a/storage/connect/plgdbsem.h
+++ b/storage/connect/plgdbsem.h
@@ -518,6 +518,7 @@ typedef  struct _colres {
   int     Clen;                    /* Data individual internal size    */
   int     Length;                  /* Data individual print length     */
   int     Prec;                    /* Precision                        */
+  int     Flag;                    /* Flag option value                */
   XFLD    Fld;                     /* Type of field info               */
   } COLRES;
 
diff --git a/storage/connect/plgdbutl.cpp b/storage/connect/plgdbutl.cpp
index 598075ac52a..73b468c9209 100644
--- a/storage/connect/plgdbutl.cpp
+++ b/storage/connect/plgdbutl.cpp
@@ -298,6 +298,7 @@ PQRYRES PlgAllocResult(PGLOBAL g, int ncol, int maxres, int ids,
     *pcrp = (PCOLRES)PlugSubAlloc(g, NULL, sizeof(COLRES));
     crp = *pcrp;
     pcrp = &crp->Next;
+    memset(crp, 0, sizeof(COLRES));
     crp->Colp = NULL;
     crp->Ncol = ++qrp->Nbcol;
     crp->Type = buftyp[i];
diff --git a/storage/connect/taboccur.cpp b/storage/connect/taboccur.cpp
index 6c33aefbb68..065da269caa 100644
--- a/storage/connect/taboccur.cpp
+++ b/storage/connect/taboccur.cpp
@@ -1,5 +1,5 @@
 /************ TabOccur CPP Declares Source Code File (.CPP) ************/
-/*  Name: TABOCCUR.CPP   Version 1.0                                   */
+/*  Name: TABOCCUR.CPP   Version 1.1                                   */
 /*                                                                     */
 /*  (C) Copyright to the author Olivier BERTRAND          2013         */
 /*                                                                     */
@@ -53,6 +53,210 @@
 
 extern "C" int trace;
 
+/***********************************************************************/
+/*  Prepare and count columns in the column list.                      */
+/***********************************************************************/
+int PrepareColist(char *colist)
+	{
+	char *p, *pn;
+	int   n = 0;
+
+	// Count the number of columns and change separator into null char
+	for (pn = colist; ; pn += (strlen(pn) + 1))
+    // Separator can be ; if colist was specified in the option_list
+		if ((p = strchr(pn, ',')) || (p = strchr(pn, ';'))) {
+			*p++ = '\0';
+			n++;
+		} else {
+			if (*pn)
+				n++;
+
+			break;
+		} // endif p
+
+	return n;
+	} // end of PrepareColist
+
+/************************************************************************/
+/*  OcrColumns: constructs the result blocks containing all the columns */
+/*  of the object table that will be retrieved by GetData commands.     */
+/************************************************************************/
+bool OcrColumns(PGLOBAL g, PQRYRES qrp, const char *col, 
+                       const char *ocr, const char *rank)
+  {
+  char   *pn, *colist;
+  int     i, k, m, n = 0, c = 0, j = qrp->Nblin;
+  bool    rk, b = false;
+  PCOLRES crp;
+
+  if (!col || !*col) {
+    strcpy(g->Message, "Missing colist");
+    return true;
+    } // endif col
+
+  // Prepare the column list
+  colist = (char*)PlugSubAlloc(g, NULL, strlen(col) + 1);
+  strcpy(colist, col);
+  m = PrepareColist(colist);
+
+  if ((rk = (rank && *rank))) {
+    if (m == 1) {
+      strcpy(g->Message, "Cannot handle one column colist and rank");
+      return true;
+      } // endif m
+
+    for (k = 0, pn = colist; k < m; k++, pn += (strlen(pn) + 1))
+      n = max(n, (signed)strlen(pn));
+
+    } // endif k
+             
+  // Default occur column name is the 1st colist column name
+  if (!ocr || !*ocr)
+    ocr = colist;
+
+  /**********************************************************************/
+  /*  Replace the columns of the colist by the rank and occur columns.  */
+  /**********************************************************************/
+  for (i = 0; i < qrp->Nblin; i++) {
+    for (k = 0, pn = colist; k < m; k++, pn += (strlen(pn) + 1))
+      if (!stricmp(pn, qrp->Colresp->Kdata->GetCharValue(i)))
+        break;
+
+    if (k < m) {
+      // This column belongs to colist
+      if (rk) {
+        // Place the rank column here
+        for (crp = qrp->Colresp; crp; crp = crp->Next)
+          switch (crp->Fld) {
+            case FLD_NAME:  crp->Kdata->SetValue((char*)rank, i); break;
+            case FLD_TYPE:  crp->Kdata->SetValue(TYPE_STRING, i); break;
+            case FLD_PREC:  crp->Kdata->SetValue(n, i);           break;
+            case FLD_SCALE: crp->Kdata->SetValue(0, i);           break;
+            case FLD_NULL:  crp->Kdata->SetValue(0, i);           break;
+            case FLD_REM:   crp->Kdata->Reset(i);                 break;
+            default: ;   // Ignored by CONNECT
+            } // endswich Fld
+    
+        rk = false;
+      } else if (!b) {
+        // First remaining listed column, will be the occur column
+        for (crp = qrp->Colresp; crp; crp = crp->Next)
+          switch (crp->Fld) {
+            case FLD_NAME: crp->Kdata->SetValue((char*)ocr, i); break;
+            case FLD_REM:  crp->Kdata->Reset(i);                break;
+            default: ;   // Nothing to do
+            } // endswich Fld
+    
+        b = true;
+      } else if (j == qrp->Nblin)
+        j = i;     // Column to remove
+
+      c++;
+    } else if (j < i) {
+      // Move this column in empty spot
+      for (crp = qrp->Colresp; crp; crp = crp->Next)
+        crp->Kdata->Move(i, j);
+
+      j++;
+    } // endif k
+
+    } // endfor i
+
+  // Check whether all columns of the list where found
+  if (c < m) {
+    strcpy(g->Message, "Some colist columns are not in the source table");
+    return true;
+    } // endif crp
+
+  /**********************************************************************/
+  /*  Set the number of columns of the table.                           */
+  /**********************************************************************/
+  qrp->Nblin = j;
+  return false;
+  } // end of OcrColumns
+
+/************************************************************************/
+/*  OcrSrcCols: constructs the result blocks containing all the columns */
+/*  of the object table that will be retrieved by GetData commands.     */
+/************************************************************************/
+bool OcrSrcCols(PGLOBAL g, PQRYRES qrp, const char *col, 
+                       const char *ocr, const char *rank)
+  {
+  char   *pn, *colist;
+  int     i, k, m, n = 0, c = 0;
+  bool    rk, b = false;
+  PCOLRES crp, rcrp, *pcrp;
+
+  if (!col || !*col) {
+    strcpy(g->Message, "Missing colist");
+    return true;
+    } // endif col
+
+  // Prepare the column list
+  colist = (char*)PlugSubAlloc(g, NULL, strlen(col) + 1);
+  strcpy(colist, col);
+  m = PrepareColist(colist);
+
+  if ((rk = (rank && *rank)))
+    for (k = 0, pn = colist; k < m; k++, pn += (strlen(pn) + 1))
+      n = max(n, (signed)strlen(pn));
+             
+  // Default occur column name is the 1st colist column name
+  if (!ocr || !*ocr)
+    ocr = colist;
+
+  /**********************************************************************/
+  /*  Replace the columns of the colist by the rank and occur columns.  */
+  /**********************************************************************/
+  for (i = 0, pcrp = &qrp->Colresp; crp = *pcrp; ) {
+    for (k = 0, pn = colist; k < m; k++, pn += (strlen(pn) + 1))
+      if (!stricmp(pn, crp->Name))
+        break;
+
+    if (k < m) {
+      // This column belongs to colist
+      c++;
+
+      if (!b) {
+        if (rk) {
+          // Add the rank column here
+          rcrp = (PCOLRES)PlugSubAlloc(g, NULL, sizeof(COLRES));
+          memset(rcrp, 0, sizeof(COLRES));
+          rcrp->Next = crp;
+          rcrp->Name = (char*)rank;
+          rcrp->Type = TYPE_STRING;
+          rcrp->Length = n;
+          rcrp->Ncol = ++i;
+          *pcrp = rcrp;
+        } // endif rk
+
+        // First remaining listed column, will be the occur column
+        crp->Name = (char*)ocr;
+        b = true;
+      } else {
+        *pcrp = crp->Next;     // Remove this column
+        continue;
+      } // endif b
+
+      } // endif k
+
+    crp->Ncol = ++i;
+    pcrp = &crp->Next;
+    } // endfor pcrp
+
+  // Check whether all columns of the list where found
+  if (c < m) {
+    strcpy(g->Message, "Some colist columns are not in the source table");
+    return true;
+    } // endif crp
+
+  /**********************************************************************/
+  /*  Set the number of columns of the table.                           */
+  /**********************************************************************/
+  qrp->Nblin = i;
+  return false;
+  } // end of OcrSrcCols
+
 /* -------------- Implementation of the OCCUR classes	---------------- */
 
 /***********************************************************************/
@@ -60,9 +264,9 @@ extern "C" int trace;
 /***********************************************************************/
 bool OCCURDEF::DefineAM(PGLOBAL g, LPCSTR am, int poff)
   {
-  Xcol = Cat->GetStringCatInfo(g, "OccurCol", "");
   Rcol = Cat->GetStringCatInfo(g, "RankCol", "");
   Colist = Cat->GetStringCatInfo(g, "Colist", "");
+  Xcol = Cat->GetStringCatInfo(g, "OccurCol", Colist);
   return PRXDEF::DefineAM(g, am, poff);
   } // end of DefineAM
 
@@ -92,37 +296,13 @@ TDBOCCUR::TDBOCCUR(POCCURDEF tdp) : TDBPRX(tdp)
 	Rcolumn = tdp->Rcol;							 // Rank column name     
 	Xcolp = NULL;											 // To the OCCURCOL column
 	Col = NULL;                        // To source column blocks array
-	Mult = PrepareColist();      			 // Multiplication factor
+	Mult = PrepareColist(Colist);      // Multiplication factor
 	N = 0;									           // The current table index
 	M = 0;                             // The occurence rank
 	RowFlag = 0;    				           // 0: Ok, 1: Same, 2: Skip
   } // end of TDBOCCUR constructor
 
 /***********************************************************************/
-/*  Prepare and count columns in the column list.                      */
-/***********************************************************************/
-int TDBOCCUR::PrepareColist(void)
-	{
-	char *p, *pn;
-	int   n = 0;
-
-	// Count the number of columns and change separator into null char
-	for (pn = Colist; ; pn += (strlen(pn) + 1))
-    // Separator can be ; if colist was specified in the option_list
-		if ((p = strchr(pn, ',')) || (p = strchr(pn, ';'))) {
-			*p++ = '\0';
-			n++;
-		} else {
-			if (*pn)
-				n++;
-
-			break;
-		} // endif p
-
-	return n;
-	} // end of PrepareColist
-
-/***********************************************************************/
 /*  Allocate OCCUR/SRC column description block.                       */
 /***********************************************************************/
 PCOL TDBOCCUR::MakeCol(PGLOBAL g, PCOLDEF cdp, PCOL cprec, int n)
diff --git a/storage/connect/taboccur.h b/storage/connect/taboccur.h
index b7d51e05b7d..10f94329703 100644
--- a/storage/connect/taboccur.h
+++ b/storage/connect/taboccur.h
@@ -58,7 +58,6 @@ class TDBOCCUR : public TDBPRX {
   // Methods
 	virtual void ResetDB(void) {N = 0; Tdbp->ResetDB();}
 	virtual int  RowNumber(PGLOBAL g, bool b = FALSE);
-					int  PrepareColist(void);
 					bool MakeColumnList(PGLOBAL g);
           bool ViewColumnList(PGLOBAL g);
 
diff --git a/storage/connect/tabpivot.cpp b/storage/connect/tabpivot.cpp
index 916a3c2584e..dbe702c7462 100644
--- a/storage/connect/tabpivot.cpp
+++ b/storage/connect/tabpivot.cpp
@@ -1,7 +1,7 @@
 /************ TabPivot C++ Program Source Code File (.CPP) *************/
 /* PROGRAM NAME: TABPIVOT                                              */
 /* -------------                                                       */
-/*  Version 1.5                                                        */
+/*  Version 1.6                                                        */
 /*                                                                     */
 /* COPYRIGHT:                                                          */
 /* ----------                                                          */
@@ -53,6 +53,182 @@
 
 extern "C" int trace;
 
+/***********************************************************************/
+/*  Make the Pivot table column list.                                  */
+/***********************************************************************/
+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)
+  {
+  PIVAID pvd(tab, src, picol, fncol, host, db, user, pwd, port);
+

+  return pvd.MakePivotColumns(g);
+  } // end of PivotColumns
+
+/* --------------- Implementation of the PIVAID classe --------------- */
+
+/***********************************************************************/
+/*  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)
+  {
+  Host = (char*)host;
+  User = (char*)user;
+  Pwd = (char*)pwd;
+  Qryp = NULL;
+  Database = (char*)db;
+  Tabname = (char*)tab;
+  Tabsrc = (char*)src;
+  Picol = (char*)picol;
+  Fncol = (char*)fncol;
+  Rblkp = NULL;
+  Port = (port) ? port : GetDefaultPort();
+  } // end of PIVAID constructor
+
+/***********************************************************************/
+/*  Make the Pivot table column list.                                  */
+/***********************************************************************/
+PQRYRES PIVAID::MakePivotColumns(PGLOBAL g)
+  {
+  char    *query, *colname, buf[32];
+  int      ndif, nblin, w = 0;

+  PVAL     valp;

+  PCOLRES *pcrp, crp, fncrp = NULL;
+
+  if (!Tabsrc && Tabname) {
+    // Locate the  query
+    query = (char*)PlugSubAlloc(g, NULL, strlen(Tabname) + 16);
+    sprintf(query, "SELECT * FROM %s", Tabname);
+  } else if (!Tabsrc) {
+    strcpy(g->Message, MSG(SRC_TABLE_UNDEF));
+    return NULL;
+  } else
+    query = Tabsrc;
+
+  // Open a MySQL connection for this table

+  if (Myc.Open(g, Host, Database, User, Pwd, Port))

+    return NULL;

+

+  // Send the source command to MySQL

+  if (Myc.ExecSQL(g, query, &w) == RC_FX) {

+    Myc.Close();

+    return NULL;

+    } // endif Exec

+

+  // We must have a storage query to get pivot column values

+  Qryp = Myc.GetResult(g);

+  Myc.Close();

+

+  if (!Fncol) {
+    for (crp = Qryp->Colresp; crp; crp = crp->Next)
+      if (!Picol || stricmp(Picol, crp->Name))
+        Fncol = crp->Name;
+  
+    if (!Fncol) {
+      strcpy(g->Message, MSG(NO_DEF_FNCCOL));
+      return NULL;
+      } // endif Fncol
+  
+    } // endif Fncol
+  
+  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))
+        Picol = crp->Name;
+  
+    if (!Picol) {
+      strcpy(g->Message, MSG(NO_DEF_PIVOTCOL));
+      return NULL;
+      } // endif Picol
+  
+    } // endif picol
+  
+  // Prepare the column list
+  for (pcrp = &Qryp->Colresp; crp = *pcrp; )
+    if (!stricmp(Picol, crp->Name)) {
+      Rblkp = crp->Kdata;
+      *pcrp = crp->Next;
+    } else if (!stricmp(Fncol, crp->Name)) {
+      fncrp = crp;
+      *pcrp = crp->Next;
+    } else
+      pcrp = &crp->Next;
+
+  if (!Rblkp) {
+    strcpy(g->Message, MSG(NO_DEF_PIVOTCOL));
+    return NULL;
+  } else if (!fncrp) {
+    strcpy(g->Message, MSG(NO_DEF_FNCCOL));
+    return NULL;
+  } // endif
+
+  // Before calling sort, initialize all

+  nblin = Qryp->Nblin;
+
+  Index.Size = nblin * sizeof(int);

+  Index.Sub = TRUE;                  // Should be small enough

+

+  if (!PlgDBalloc(g, NULL, Index))

+    return NULL;

+

+  Offset.Size = (nblin + 1) * sizeof(int);

+  Offset.Sub = TRUE;                 // Should be small enough

+

+  if (!PlgDBalloc(g, NULL, Offset))

+    return NULL;

+

+  ndif = Qsort(g, nblin);

+

+  if (ndif < 0)           // error

+    return NULL;

+

+  // Allocate the Value used to retieve column names

+  if (!(valp = AllocateValue(g, Rblkp->GetType(),
+                                Rblkp->GetVlen(),

+                                Rblkp->GetPrec())))

+    return NULL;

+

+  // Now make the functional columns

+  for (int i = 0; i < ndif; i++) {
+    if (i) {
+      crp = (PCOLRES)PlugSubAlloc(g, NULL, sizeof(COLRES));
+      memcpy(crp, fncrp, sizeof(COLRES));
+    } else
+      crp = fncrp;
+
+    // Get the value that will be the generated column name

+    valp->SetValue_pvblk(Rblkp, Pex[Pof[i]]);

+    colname = valp->GetCharString(buf);
+    crp->Name = (char*)PlugSubAlloc(g, NULL, strlen(colname) + 1);
+    strcpy(crp->Name, colname);
+    crp->Flag = 1;
+
+    // Add this column
+    *pcrp = crp;
+    crp->Next = NULL;
+    pcrp = &crp->Next;
+    } // endfor i
+
+  // We added ndif columns and removed 2 (picol and fncol)
+  Qryp->Nbcol += (ndif - 2);
+  return Qryp;
+  } // end of MakePivotColumns
+
+/***********************************************************************/

+/*  PIVAID: Compare routine for sorting pivot column values.           */

+/***********************************************************************/

+int PIVAID::Qcompare(int *i1, int *i2)

+  {

+  // TODO: the actual comparison between pivot column result values.

+  return Rblkp->CompVal(*i1, *i2);

+  } // end of Qcompare

+

 /* --------------- Implementation of the PIVOT classes --------------- */
 
 /***********************************************************************/
diff --git a/storage/connect/tabpivot.h b/storage/connect/tabpivot.h
index 1617801298b..4a766700162 100644
--- a/storage/connect/tabpivot.h
+++ b/storage/connect/tabpivot.h
@@ -1,5 +1,5 @@
 /************** TabPivot H Declares Source Code File (.H) **************/
-/*  Name: TABPIVOT.H    Version 1.4                                    */
+/*  Name: TABPIVOT.H    Version 1.5                                    */
 /*                                                                     */
 /*  (C) Copyright to the author Olivier BERTRAND          2005-2013    */
 /*                                                                     */
@@ -10,6 +10,40 @@ typedef class TDBPIVOT *PTDBPIVOT;
 typedef class FNCCOL   *PFNCCOL;
 typedef class SRCCOL   *PSRCCOL;
 
+/***********************************************************************/

+/*  This class is used to generate PIVOT table column definitions.     */

+/***********************************************************************/

+class PIVAID : public CSORT {

+  friend class FNCCOL;

+  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);

+

+  // Methods

+  PQRYRES MakePivotColumns(PGLOBAL g);

+

+  // The sorting function

+  virtual int  Qcompare(int *, int *);

+

+ protected:

+  // Members

+  MYSQLC  Myc;                      // MySQL connection class

+  char   *Host;                     // Host machine to use

+  char   *User;                     // User logon info

+  char   *Pwd;                      // Password logon info

+  char   *Database;                 // Database to be used by server

+  PQRYRES Qryp;                     // Points to Query result block

+  char   *Tabname;                  // Name of source table

+  char   *Tabsrc;                   // SQL of source table

+  char   *Picol;                    // Pivot column name

+  char   *Fncol;                    // Function column name

+  PVBLK   Rblkp;                    // The value block of the pivot column

+  int     Port;                     // MySQL port number

+  }; // end of class PIVAID

+

 /* -------------------------- PIVOT classes -------------------------- */
 
 /***********************************************************************/
diff --git a/storage/connect/valblk.h b/storage/connect/valblk.h
index a7b1b5046d7..d9286b72f9f 100644
--- a/storage/connect/valblk.h
+++ b/storage/connect/valblk.h
@@ -36,6 +36,7 @@ class VALBLK : public BLOCK {
           void  *GetValPointer(void) {return Blkp;}
           void   SetValPointer(void *mp) {Blkp = mp;}
           int    GetType(void) {return Type;}
+          int    GetPrec(void) {return Prec;}
           void   SetCheck(bool b) {Check = b;}
           void   MoveNull(int i, int j)
                   {if (To_Nulls) To_Nulls[j] = To_Nulls[j];}
@@ -110,7 +111,7 @@ class TYPBLK : public VALBLK {
 
   // Implementation
   virtual void   Init(PGLOBAL g, bool check);
-  virtual int    GetVlen(void) {return sizeof(int);}
+  virtual int    GetVlen(void) {return sizeof(TYPE);}
 //virtual PSZ    GetCharValue(int n);
   virtual short  GetShortValue(int n) {return (short)Typp[n];}
   virtual int    GetIntValue(int n) {return (int)Typp[n];}