- Set connection charset before calling mysql_real_connect for MYSQL

  tables. This should fix bug MDEV-7343.
modified:
  storage/connect/ha_connect.cc
  storage/connect/myconn.cpp
  storage/connect/myconn.h
  storage/connect/reldef.cpp
  storage/connect/reldef.h
  storage/connect/table.cpp
  storage/connect/tabmysql.cpp
  storage/connect/xtable.h

- Prevent double column evaluation when CONNECT does filtering
modified:
  storage/connect/connect.cc

- Export CreateFileMap and CloseMemMap (for OEM tables)
modified:
  storage/connect/maputil.h

- Add the compute function to be used on VALUE types.
  Preserve precision for DOUBLE values.
modified:
  storage/connect/value.cpp
  storage/connect/value.h

- Typo (in preparation to the future JSON table type)
modified:
  storage/connect/ha_connect.cc
  storage/connect/mycat.cc
  storage/connect/plgdbsem.h

14 files changed, 305 insertions, 26 deletions

diff --git a/storage/connect/connect.cc b/storage/connect/connect.cc
index 87c782ba953..a54d8ebcc44 100644
--- a/storage/connect/connect.cc
+++ b/storage/connect/connect.cc
@@ -52,7 +52,7 @@
 /*  Routines called internally by semantic routines.                   */
 /***********************************************************************/
 void  CntEndDB(PGLOBAL);
-RCODE EvalColumns(PGLOBAL g, PTDB tdbp, bool mrr= false);
+RCODE EvalColumns(PGLOBAL g, PTDB tdbp, bool reset, bool mrr= false);
 
 /***********************************************************************/
 /*  MySQL routines called externally by semantic routines.             */
@@ -388,7 +388,7 @@ bool CntRewindTable(PGLOBAL g, PTDB tdbp)
 /***********************************************************************/
 /*  Evaluate all columns after a record is read.                       */
 /***********************************************************************/
-RCODE EvalColumns(PGLOBAL g, PTDB tdbp, bool mrr)
+RCODE EvalColumns(PGLOBAL g, PTDB tdbp, bool reset, bool mrr)
   {
   RCODE rc= RC_OK;
   PCOL  colp;
@@ -413,7 +413,8 @@ RCODE EvalColumns(PGLOBAL g, PTDB tdbp, bool mrr)
 
   for (colp= tdbp->GetColumns(); rc == RC_OK && colp;
        colp= colp->GetNext()) {
-    colp->Reset();
+    if (reset)
+      colp->Reset();
 
     // Virtual columns are computed by MariaDB
     if (!colp->GetColUse(U_VIRTUAL) && (!mrr || colp->GetKcol()))
@@ -457,6 +458,10 @@ RCODE CntReadNext(PGLOBAL g, PTDB tdbp)
     goto err;
     } // endif rc
 
+  // Do it now to avoid double eval when filtering
+  for (PCOL colp= tdbp->GetColumns(); colp; colp= colp->GetNext())
+    colp->Reset();
+
   do {
     if ((rc= (RCODE)tdbp->ReadDB(g)) == RC_OK)
       if (!ApplyFilter(g, tdbp->GetFilter()))
@@ -466,7 +471,7 @@ RCODE CntReadNext(PGLOBAL g, PTDB tdbp)
 
  err:
   g->jump_level--;
-  return (rc != RC_OK) ? rc : EvalColumns(g, tdbp);
+  return (rc != RC_OK) ? rc : EvalColumns(g, tdbp, false);
   } // end of CntReadNext
 
 /***********************************************************************/
@@ -812,7 +817,7 @@ RCODE CntIndexRead(PGLOBAL g, PTDB ptdb, OPVAL op,
 
  rnd:
   if ((rc= (RCODE)ptdb->ReadDB(g)) == RC_OK)
-    rc= EvalColumns(g, ptdb, mrr);
+    rc= EvalColumns(g, ptdb, true, mrr);
 
   return rc;
   } // end of CntIndexRead
diff --git a/storage/connect/ha_connect.cc b/storage/connect/ha_connect.cc
index ae8f7f0efa8..9c847b1d250 100644
--- a/storage/connect/ha_connect.cc
+++ b/storage/connect/ha_connect.cc
@@ -1055,6 +1055,14 @@ char *ha_connect::GetStringOption(char *opname, char *sdef)
     opval= (char*)options->colist;
   else if (!stricmp(opname, "Data_charset"))
     opval= (char*)options->data_charset;
+  else if (!stricmp(opname, "Table_charset")) {
+    const CHARSET_INFO *chif= (tshp) ? tshp->table_charset 
+                                     : table->s->table_charset;
+
+    if (chif)
+      opval= (char*)chif->csname;
+
+  } // endif Table_charset
 
   if (!opval && options && options->oplist)
     opval= GetListOption(xp->g, opname, options->oplist);
@@ -3806,6 +3814,7 @@ bool ha_connect::check_privileges(THD *thd, PTOS options, char *dbn)
     case TAB_XML:
     case TAB_INI:
     case TAB_VEC:
+//  case TAB_JSON:
       if (options->filename && *options->filename) {
         char *s, path[FN_REFLEN], dbpath[FN_REFLEN];
 #if defined(WIN32)
diff --git a/storage/connect/maputil.h b/storage/connect/maputil.h
index b5e54affcea..e310488eb5d 100644
--- a/storage/connect/maputil.h
+++ b/storage/connect/maputil.h
@@ -11,8 +11,8 @@ typedef struct {
   DWORD lenH;
   } MEMMAP;
 
-HANDLE  CreateFileMap(PGLOBAL, LPCSTR, MEMMAP *, MODE, bool);
-bool    CloseMemMap(void *memory, size_t dwSize);
+DllExport HANDLE  CreateFileMap(PGLOBAL, LPCSTR, MEMMAP *, MODE, bool);
+DllExport bool    CloseMemMap(void *memory, size_t dwSize);
 
 #ifdef __cplusplus
 }
diff --git a/storage/connect/mycat.cc b/storage/connect/mycat.cc
index fc6c29092a1..57b2e5d1a9c 100644
--- a/storage/connect/mycat.cc
+++ b/storage/connect/mycat.cc
@@ -1,4 +1,4 @@
-/* Copyright (C) Olivier Bertrand 2004 - 2013
+/* Copyright (C) Olivier Bertrand 2004 - 2014
 
   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
@@ -18,7 +18,7 @@
 /* -------------                                                       */
 /*  Version 1.4                                                        */
 /*                                                                     */
-/*  Author: Olivier Bertrand                       2012 - 2013         */
+/*  Author: Olivier Bertrand                       2012 - 2014         */
 /*                                                                     */
 /* WHAT THIS PROGRAM DOES:                                             */
 /* -----------------------                                             */
@@ -89,6 +89,7 @@
 #include "tabpivot.h"
 #endif   // PIVOT_SUPPORT
 #include "tabvir.h"
+//#include "tabjson.h"
 #include "ha_connect.h"
 #include "mycat.h"
 
@@ -139,6 +140,7 @@ TABTYPE GetTypeID(const char *type)
                  : (!stricmp(type, "PIVOT")) ? TAB_PIVOT
 #endif
                  : (!stricmp(type, "VIR"))   ? TAB_VIR
+//               : (!stricmp(type, "JSON"))  ? TAB_JSON
                  : (!stricmp(type, "OEM"))   ? TAB_OEM : TAB_NIY;
   } // end of GetTypeID
 
@@ -159,6 +161,7 @@ bool IsFileType(TABTYPE type)
     case TAB_XML:
     case TAB_INI:
     case TAB_VEC:
+//  case TAB_JSON:
       isfile= true;
       break;
     default:
@@ -181,6 +184,7 @@ bool IsExactType(TABTYPE type)
     case TAB_BIN:
     case TAB_DBF:
 //  case TAB_XML:     depends on Multiple || Xpand || Coltype
+//  case TAB_JSON:    depends on Multiple || Xpand || Coltype
     case TAB_VEC:
     case TAB_VIR:
       exact= true;
@@ -214,7 +218,7 @@ bool IsTypeNullable(TABTYPE type)
   } // end of IsTypeNullable
 
 /***********************************************************************/
-/*  Return true for indexable table by XINDEX.                         */
+/*  Return true for fixed record length tables.                        */
 /***********************************************************************/
 bool IsTypeFixed(TABTYPE type)
   {
@@ -538,6 +542,7 @@ PRELDEF MYCAT::MakeTableDesc(PGLOBAL g, LPCSTR name, LPCSTR am)
     case TAB_PIVOT: tdp= new(g) PIVOTDEF; break;
 #endif   // PIVOT_SUPPORT
     case TAB_VIR: tdp= new(g) VIRDEF;   break;
+//  case TAB_JSON: tdp= new(g) JSONDEF; break;
     default:
       sprintf(g->Message, MSG(BAD_TABLE_TYPE), am, name);
     } // endswitch
diff --git a/storage/connect/myconn.cpp b/storage/connect/myconn.cpp
index 92c2faea676..2f3d75b52fa 100644
--- a/storage/connect/myconn.cpp
+++ b/storage/connect/myconn.cpp
@@ -430,10 +430,11 @@ int MYSQLC::GetResultSize(PGLOBAL g, PSZ sql)
 /***********************************************************************/
 int MYSQLC::Open(PGLOBAL g, const char *host, const char *db,
                             const char *user, const char *pwd,
-                            int pt)
+                            int pt, const char *csname)
   {
   const char *pipe = NULL;
-  uint cto = 6000, nrt = 12000;
+  uint        cto = 6000, nrt = 12000;
+  my_bool     my_true= 1;
 
   m_DB = mysql_init(NULL);
 
@@ -470,6 +471,18 @@ int MYSQLC::Open(PGLOBAL g, const char *host, const char *db,
     } // endif pwd
 #endif // 0
 
+/***********************************************************************/
+/*	BUG# 17044 Federated Storage Engine is not UTF8 clean              */
+/*	Add set names to whatever charset the table is at open of table    */
+/*  this sets the csname like 'set names utf8'.                        */
+/***********************************************************************/
+  if (csname)
+    mysql_options(m_DB, MYSQL_SET_CHARSET_NAME, csname);
+
+  // Don't know what this one do but FEDERATED does it
+  mysql_options(m_DB, MYSQL_OPT_USE_THREAD_SPECIFIC_MEMORY,
+                  (char*)&my_true);
+
   if (!mysql_real_connect(m_DB, host, user, pwd, db, pt, pipe, CLIENT_MULTI_RESULTS)) {
 #if defined(_DEBUG)
     sprintf(g->Message, "mysql_real_connect failed: (%d) %s",
diff --git a/storage/connect/myconn.h b/storage/connect/myconn.h
index 65e6531aee4..79b8a43fe5a 100644
--- a/storage/connect/myconn.h
+++ b/storage/connect/myconn.h
@@ -67,7 +67,7 @@ class DllItem MYSQLC {
   int     GetTableSize(PGLOBAL g, PSZ query);
   int     Open(PGLOBAL g, const char *host, const char *db,
                           const char *user= "root", const char *pwd= "*",
-                          int pt= 0);
+                          int pt= 0, const char *csname = NULL);
   int     KillQuery(ulong id);
   int     ExecSQL(PGLOBAL g, const char *query, int *w = NULL);
   int     ExecSQLcmd(PGLOBAL g, const char *query, int *w);
@@ -98,5 +98,6 @@ class DllItem MYSQLC {
   int         m_Fields;     // The number of result fields
   int         m_Afrw;       // The number of affected rows
   bool        m_Use;        // Use or store result set
+  const char *csname;       // Table charset name
   }; // end of class MYSQLC
 
diff --git a/storage/connect/plgdbsem.h b/storage/connect/plgdbsem.h
index bbbbc1486b6..b4af13c57cc 100644
--- a/storage/connect/plgdbsem.h
+++ b/storage/connect/plgdbsem.h
@@ -74,9 +74,11 @@ enum TABTYPE {TAB_UNDEF =  0,   /* Table of undefined type             */
               TAB_PLG   = 20,   /* PLG NIY                             */
               TAB_PIVOT = 21,   /* PIVOT table                         */
               TAB_VIR   = 22,   /* Virtual tables                      */
-              TAB_JCT   = 23,   /* Junction tables NIY                 */
-              TAB_DMY   = 24,   /* DMY Dummy tables NIY                */
-              TAB_NIY   = 25};  /* Table not implemented yet           */
+              TAB_JSON  = 23,   /* JSON tables                         */
+              TAB_JSN   = 24,   /* Semi-json tables                    */
+              TAB_JCT   = 25,   /* Junction tables NIY                 */
+              TAB_DMY   = 26,   /* DMY Dummy tables NIY                */
+              TAB_NIY   = 27};  /* Table not implemented yet           */
 
 enum AMT {TYPE_AM_ERROR =   0,        /* Type not defined              */
           TYPE_AM_ROWID =   1,        /* ROWID type (special column)   */
diff --git a/storage/connect/reldef.cpp b/storage/connect/reldef.cpp
index e469ae40f1f..51d777a7d17 100644
--- a/storage/connect/reldef.cpp
+++ b/storage/connect/reldef.cpp
@@ -202,6 +202,8 @@ TABDEF::TABDEF(void)
   Degree = 0;
   Pseudo = 0;
   Read_Only = false;
+  m_data_charset = NULL;
+  csname = NULL;
   } // end of TABDEF constructor
 
 /***********************************************************************/
@@ -224,6 +226,7 @@ bool TABDEF::Define(PGLOBAL g, PCATLG cat, LPCSTR name, LPCSTR am)
   m_data_charset= data_charset_name ?
                   get_charset_by_csname(data_charset_name, MY_CS_PRIMARY, 0):
                   NULL;
+  csname = GetStringCatInfo(g, "Table_charset", NULL);
 
   // Get The column definitions
   if ((poff = GetColCatInfo(g)) < 0)
diff --git a/storage/connect/reldef.h b/storage/connect/reldef.h
index a1dfe87dca8..6160ea71680 100644
--- a/storage/connect/reldef.h
+++ b/storage/connect/reldef.h
@@ -68,6 +68,7 @@ class DllExport TABDEF : public RELDEF {   /* Logical table descriptor */
   friend class CATALOG;
   friend class PLUGCAT;
   friend class MYCAT;
+  friend class TDBASE;
  public:
   // Constructor
   TABDEF(void);                  // Constructor
@@ -110,6 +111,7 @@ class DllExport TABDEF : public RELDEF {   /* Logical table descriptor */
   int     Pseudo;               /* Bit: 1 ROWID Ok, 2 FILEID Ok        */
   bool    Read_Only;            /* true for read only tables           */
   const CHARSET_INFO *m_data_charset;
+  const char *csname;           /* Table charset name                  */
   }; // end of TABDEF
 
 /***********************************************************************/
diff --git a/storage/connect/table.cpp b/storage/connect/table.cpp
index b093e2102c2..cbd5910d8c8 100644
--- a/storage/connect/table.cpp
+++ b/storage/connect/table.cpp
@@ -146,6 +146,7 @@ TDBASE::TDBASE(PTABDEF tdp) : TDB(tdp)
   Knum = 0;
   Read_Only = (tdp) ? tdp->IsReadOnly() : false;
   m_data_charset=  (tdp) ? tdp->data_charset() : NULL;
+  csname = (tdp) ? tdp->csname : NULL;
   } // end of TDBASE constructor
 
 TDBASE::TDBASE(PTDBASE tdbp) : TDB(tdbp)
@@ -161,6 +162,7 @@ TDBASE::TDBASE(PTDBASE tdbp) : TDB(tdbp)
   Knum = tdbp->Knum;
   Read_Only = tdbp->Read_Only;
   m_data_charset= tdbp->m_data_charset;
+  csname = tdbp->csname;
   } // end of TDBASE copy constructor
 
 /***********************************************************************/
diff --git a/storage/connect/tabmysql.cpp b/storage/connect/tabmysql.cpp
index 3ec9a1feaee..54627ba43fd 100644
--- a/storage/connect/tabmysql.cpp
+++ b/storage/connect/tabmysql.cpp
@@ -777,7 +777,7 @@ int TDBMYSQL::Cardinality(PGLOBAL g)
     char   query[96];
     MYSQLC myc;
 
-    if (myc.Open(g, Host, Database, User, Pwd, Port))
+    if (myc.Open(g, Host, Database, User, Pwd, Port, csname))
       return -1;
 
     strcpy(query, "SELECT COUNT(*) FROM ");
@@ -871,7 +871,7 @@ bool TDBMYSQL::OpenDB(PGLOBAL g)
   /*  servers allowing concurency in getting results ???               */
   /*********************************************************************/
   if (!Myc.Connected()) {
-    if (Myc.Open(g, Host, Database, User, Pwd, Port))
+    if (Myc.Open(g, Host, Database, User, Pwd, Port, csname))
       return true;
 
     } // endif Connected
diff --git a/storage/connect/value.cpp b/storage/connect/value.cpp
index 7227e637a14..3ed58d3e257 100644
--- a/storage/connect/value.cpp
+++ b/storage/connect/value.cpp
@@ -330,7 +330,7 @@ int ConvertType(int target, int type, CONV kind, bool match)
 /***********************************************************************/
 /*  AllocateConstant: allocates a constant Value.                      */
 /***********************************************************************/
-PVAL AllocateValue(PGLOBAL g, void *value, short type)
+PVAL AllocateValue(PGLOBAL g, void *value, short type, short prec)
   {
   PVAL valp;
 
@@ -351,7 +351,7 @@ PVAL AllocateValue(PGLOBAL g, void *value, short type)
       valp = new(g) TYPVAL<longlong>(*(longlong*)value, TYPE_BIGINT);
       break;
     case TYPE_DOUBLE:
-      valp = new(g) TYPVAL<double>(*(double *)value, TYPE_DOUBLE, 2);
+      valp = new(g) TYPVAL<double>(*(double *)value, TYPE_DOUBLE, prec);
       break;
     case TYPE_TINY:
       valp = new(g) TYPVAL<char>(*(char *)value, TYPE_TINY);
@@ -475,7 +475,7 @@ PVAL AllocateValue(PGLOBAL g, PVAL valp, int newtype, int uns)
       break;
     case TYPE_DOUBLE:
       valp = new(g) TYPVAL<double>(valp->GetFloatValue(), TYPE_DOUBLE,
-                                   valp->GetValPrec());
+                     (uns) ? uns : valp->GetValPrec());
       break;
     case TYPE_TINY:
       if (un)
@@ -542,6 +542,15 @@ BYTE VALUE::TestValue(PVAL vp)
   return (n > 0) ? 0x04 : (n < 0) ? 0x02 : 0x01;
   } // end of TestValue
 
+/***********************************************************************/
+/*  Compute a function on a string.                                    */
+/***********************************************************************/
+bool VALUE::Compute(PGLOBAL g, PVAL *vp, int np, OPVAL op)
+  {
+  strcpy(g->Message, "Compute not implemented for this value type");
+  return true;
+  } // end of Compute
+
 /* -------------------------- Class TYPVAL ---------------------------- */
 
 /***********************************************************************/
@@ -930,6 +939,188 @@ int TYPVAL<TYPE>::CompareValue(PVAL vp)
   return (Tval > n) ? 1 : (Tval < n) ? (-1) : 0;
   } // end of CompareValue
 
+#if 0
+/***********************************************************************/
+/*  Return max type value if b is true, else min type value.           */
+/***********************************************************************/
+template <>
+short TYPVAL<short>::MinMaxVal(bool b)
+  {return (b) ? MAXINT16 : MININT16;}
+
+template <>
+USHORT TYPVAL<USHORT>::MinMaxVal(bool b)
+  {return (b) ? MAXUINT16 : 0;}
+
+template <>
+int TYPVAL<int>::MinMaxVal(bool b)
+  {return (b) ? MAXINT32 : MININT32;}
+
+template <>
+UINT TYPVAL<UINT>::MinMaxVal(bool b)
+  {return (b) ? MAXUINT32 : 0;}
+
+template <>
+longlong TYPVAL<longlong>::MinMaxVal(bool b)
+  {return (b) ? MAXINT64 : MININT64;}
+
+template <>
+ulonglong TYPVAL<ulonglong>::MinMaxVal(bool b)
+  {return (b) ? MAXUINT64 : 0;}
+
+template <>
+double TYPVAL<double>::MinMaxVal(bool b)
+  {assert(false); return 0.0;}
+
+template <>
+char TYPVAL<char>::MinMaxVal(bool b)
+  {return (b) ? MAXINT8 : MININT8;}
+
+template <>
+UCHAR TYPVAL<UCHAR>::MinMaxVal(bool b)
+  {return (b) ? MAXUINT8 : 0;}
+
+/***********************************************************************/
+/*  SafeAdd: adds a value and test whether overflow/underflow occured. */
+/***********************************************************************/
+template <class TYPE>
+TYPE TYPVAL<TYPE>::SafeAdd(TYPE n1, TYPE n2)
+  {
+  PGLOBAL& g = Global;
+  TYPE     n = n1 + n2;
+
+  if ((n2 > 0) && (n < n1)) {
+    // Overflow
+    strcpy(g->Message, MSG(FIX_OVFLW_ADD));
+    longjmp(g->jumper[g->jump_level], 138);
+  } else if ((n2 < 0) && (n > n1)) {
+    // Underflow
+    strcpy(g->Message, MSG(FIX_UNFLW_ADD));
+    longjmp(g->jumper[g->jump_level], 138);
+  } // endif's n2
+
+  return n;
+  } // end of SafeAdd
+
+template <>
+inline double TYPVAL<double>::SafeAdd(double n1, double n2)
+  {
+  assert(false); return 0;
+  } // end of SafeAdd
+
+/***********************************************************************/
+/*  SafeMult: multiply values and test whether overflow occured.       */
+/***********************************************************************/
+template <class TYPE>
+TYPE TYPVAL<TYPE>::SafeMult(TYPE n1, TYPE n2)
+  {
+  PGLOBAL& g = Global;
+  double   n = (double)n1 * (double)n2;
+
+  if (n > MinMaxVal(true)) {
+    // Overflow
+    strcpy(g->Message, MSG(FIX_OVFLW_TIMES));
+    longjmp(g->jumper[g->jump_level], 138);
+  } else if (n < MinMaxVal(false)) {
+    // Underflow
+    strcpy(g->Message, MSG(FIX_UNFLW_TIMES));
+    longjmp(g->jumper[g->jump_level], 138);
+  } // endif's n2
+
+  return (TYPE)n;
+  } // end of SafeMult
+
+template <>
+inline double TYPVAL<double>::SafeMult(double n1, double n2)
+  {
+  assert(false); return 0;
+  } // end of SafeMult
+#endif // 0
+
+/***********************************************************************/
+/*  Compute defined functions for the type.                            */
+/***********************************************************************/
+template <class TYPE>
+bool TYPVAL<TYPE>::Compute(PGLOBAL g, PVAL *vp, int np, OPVAL op)
+  {
+  bool rc = false;
+  TYPE val[2];
+
+  assert(np == 2);
+
+  for (int i = 0; i < np; i++)
+    val[i] = GetTypedValue(vp[i]);
+
+  switch (op) {
+    case OP_ADD:
+//    Tval = SafeAdd(val[0], val[1]);
+      Tval = val[0] + val[1];
+      break;
+    case OP_MULT:
+//    Tval = SafeMult(val[0], val[1]);
+      Tval = val[0] * val[1];
+      break;
+    default:
+      rc = Compall(g, vp, np, op);
+      break;
+    } // endswitch op
+
+  return rc;
+  } // end of Compute
+
+#if 0
+template <>
+bool TYPVAL<double>::Compute(PGLOBAL g, PVAL *vp, int np, OPVAL op)
+  {
+  bool   rc = false;
+  double val[2];
+
+  assert(np == 2);
+
+  for (int i = 0; i < np; i++)
+    val[i] = vp[i]->GetFloatValue();
+
+  switch (op) {
+    case OP_ADD:
+      Tval = val[0] + val[1];
+      break;
+    case OP_MULT:
+      Tval = val[0] * val[1];
+      break;
+    default:
+      rc = Compall(g, vp, np, op);
+    } // endswitch op
+
+  return rc;
+  } // end of Compute
+#endif // 0
+
+/***********************************************************************/
+/*  Compute a function for all types.                                  */
+/***********************************************************************/
+template <class TYPE>
+bool TYPVAL<TYPE>::Compall(PGLOBAL g, PVAL *vp, int np, OPVAL op)
+  {
+  TYPE val[2];
+
+  for (int i = 0; i < np; i++)
+    val[i] = GetTypedValue(vp[i]);
+
+  switch (op) {
+    case OP_MIN:
+      Tval = MY_MIN(val[0], val[1]);
+      break;
+    case OP_MAX:
+      Tval = MY_MAX(val[0], val[1]);
+      break;
+    default:
+//    sprintf(g->Message, MSG(BAD_EXP_OPER), op);
+      strcpy(g->Message, "Function not supported");
+      return true;
+    } // endswitch op
+
+  return false;
+  } // end of Compall
+
 /***********************************************************************/
 /*  FormatValue: This function set vp (a STRING value) to the string   */
 /*  constructed from its own value formated using the fmt format.      */
@@ -1410,6 +1601,45 @@ int TYPVAL<PSZ>::CompareValue(PVAL vp)
   } // end of CompareValue
 
 /***********************************************************************/
+/*  Compute a function on a string.                                    */
+/***********************************************************************/
+bool TYPVAL<PSZ>::Compute(PGLOBAL g, PVAL *vp, int np, OPVAL op)
+  {
+  char *p[2], val[2][32];
+  int   i;
+
+  for (i = 0; i < np; i++)
+    p[i] = vp[i]->GetCharString(val[i]);
+
+  switch (op) {
+    case OP_CNC:
+      assert(np == 1 || np == 2);
+
+      if (np == 2)
+        strncpy(Strp, p[0], Len);
+
+      if ((i = Len - (signed)strlen(Strp)) > 0)
+        strncat(Strp, p[np - 1], i);
+
+      break;
+    case OP_MIN:
+      assert(np == 2);
+      strcpy(Strp, (strcmp(p[0], p[1]) < 0) ? p[0] : p[1]);
+      break;
+    case OP_MAX:
+      assert(np == 2);
+      strcpy(Strp, (strcmp(p[0], p[1]) > 0) ? p[0] : p[1]);
+      break;
+    default:
+//    sprintf(g->Message, MSG(BAD_EXP_OPER), op);
+      strcpy(g->Message, "Function not supported");
+      return true;
+    } // endswitch op
+
+  return false;
+  } // end of Compute
+
+/***********************************************************************/
 /*  FormatValue: This function set vp (a STRING value) to the string   */
 /*  constructed from its own value formated using the fmt format.      */
 /*  This function assumes that the format matches the value type.      */
diff --git a/storage/connect/value.h b/storage/connect/value.h
index 3ce7027aeeb..c2cc24894ae 100644
--- a/storage/connect/value.h
+++ b/storage/connect/value.h
@@ -29,12 +29,9 @@ enum CONV {CNV_ANY     =   0,         /* Convert to any type           */
 class CONSTANT;                       // For friend setting
 typedef struct _datpar *PDTP;         // For DTVAL
 
-
 /***********************************************************************/
 /*  Utilities used to test types and to allocated values.              */
 /***********************************************************************/
-PVAL  AllocateValue(PGLOBAL, void *, short);
-
 // Exported functions
 DllExport PSZ   GetTypeName(int);
 DllExport int   GetTypeSize(int, int);
@@ -47,6 +44,7 @@ DllExport int   GetFormatType(char);
 DllExport bool  IsTypeChar(int type);
 DllExport bool  IsTypeNum(int type);
 DllExport int   ConvertType(int, int, CONV, bool match = false);
+DllExport PVAL  AllocateValue(PGLOBAL, void *, short, short = 2);
 DllExport PVAL  AllocateValue(PGLOBAL, PVAL, int = TYPE_VOID, int = 0);
 DllExport PVAL  AllocateValue(PGLOBAL, int, int len = 0, int prec = 0,
                               bool uns = false, PSZ fmt = NULL);
@@ -114,6 +112,7 @@ class DllExport VALUE : public BLOCK {
   virtual char  *ShowValue(char *buf, int len = 0) = 0;
   virtual char  *GetCharString(char *p) = 0;
   virtual bool   IsEqual(PVAL vp, bool chktype) = 0;
+  virtual bool   Compute(PGLOBAL g, PVAL *vp, int np, OPVAL op);
   virtual bool   FormatValue(PVAL vp, char *fmt) = 0;
 
  protected:
@@ -149,9 +148,9 @@ class DllExport TYPVAL : public VALUE {
   virtual bool   IsZero(void) {return Tval == 0;}
   virtual void   Reset(void) {Tval = 0;}
   virtual int    GetValLen(void);
-  virtual int    GetValPrec() {return 0;}
+  virtual int    GetValPrec() {return Prec;}
   virtual int    GetSize(void) {return sizeof(TYPE);}
-  virtual PSZ    GetCharValue(void) {return VALUE::GetCharValue();}
+//virtual PSZ    GetCharValue(void) {return VALUE::GetCharValue();}
   virtual char   GetTinyValue(void) {return (char)Tval;}
   virtual uchar  GetUTinyValue(void) {return (uchar)Tval;}
   virtual short  GetShortValue(void) {return (short)Tval;}
@@ -184,12 +183,18 @@ class DllExport TYPVAL : public VALUE {
   virtual char  *ShowValue(char *buf, int);
   virtual char  *GetCharString(char *p);
   virtual bool   IsEqual(PVAL vp, bool chktype);
+  virtual bool   Compute(PGLOBAL g, PVAL *vp, int np, OPVAL op);
   virtual bool   SetConstFormat(PGLOBAL, FORMAT&);
   virtual bool   FormatValue(PVAL vp, char *fmt);
   virtual void   Print(PGLOBAL g, FILE *, uint);
   virtual void   Print(PGLOBAL g, char *, uint);
 
  protected:
+//static  TYPE   MinMaxVal(bool b);
+//        TYPE   SafeAdd(TYPE n1, TYPE n2);
+//        TYPE   SafeMult(TYPE n1, TYPE n2);
+          bool   Compall(PGLOBAL g, PVAL *vp, int np, OPVAL op);
+
   // Default constructor not to be used
   TYPVAL(void) : VALUE(TYPE_ERROR) {}
 
@@ -253,6 +258,7 @@ class DllExport TYPVAL<PSZ>: public VALUE {
   virtual char  *ShowValue(char *buf, int);
   virtual char  *GetCharString(char *p);
   virtual bool   IsEqual(PVAL vp, bool chktype);
+  virtual bool   Compute(PGLOBAL g, PVAL *vp, int np, OPVAL op);
   virtual bool   FormatValue(PVAL vp, char *fmt);
   virtual bool   SetConstFormat(PGLOBAL, FORMAT&);
 
diff --git a/storage/connect/xtable.h b/storage/connect/xtable.h
index 49fbbb0de26..501a5e87cfa 100644
--- a/storage/connect/xtable.h
+++ b/storage/connect/xtable.h
@@ -209,6 +209,7 @@ class DllExport TDBASE : public TDB {
   int      Knum;              // Size of key arrays
   bool     Read_Only;         // True for read only tables
   const CHARSET_INFO *m_data_charset;
+  const char *csname;         // Table charset name
   }; // end of class TDBASE
 
 /***********************************************************************/