- Add support to NULL values. This concern the MYSQL

  and ODBC table types. Not supported yet for indexes.

modified:
  storage/connect/colblk.cpp
  storage/connect/colblk.h
  storage/connect/connect.cc
  storage/connect/ha_connect.cc
  storage/connect/tabmysql.cpp
  storage/connect/tabodbc.cpp
  storage/connect/valblk.cpp
  storage/connect/valblk.h
  storage/connect/value.cpp
  storage/connect/value.h
  storage/connect/xindex.cpp

storage/connect/colblk.cpp

 /************* Colblk C++ Functions Source Code File (.CPP) ************/
-/*  Name: COLBLK.CPP  Version 1.9                                      */
+/*  Name: COLBLK.CPP  Version 2.0                                      */
 /*                                                                     */
-/*  (C) Copyright to the author Olivier BERTRAND          1998-2012    */
+/*  (C) Copyright to the author Olivier BERTRAND          1998-2013    */
 /*                                                                     */
 /*  This file contains the COLBLK class functions.                     */
 /***********************************************************************/
@@ -49,6 +49,7 @@ COLBLK::COLBLK(PCOLDEF cdp, PTDB tdbp, int i)
   } // endif cdp
 
   To_Tdb = tdbp;
+  Nullable = false;
   Status = BUF_NO;
 //Value = NULL;                  done in XOBJECT constructor
   To_Kcol = NULL;
@@ -190,6 +191,7 @@ bool COLBLK::InitValue(PGLOBAL g)
     return true;
 
   Status = BUF_READY;
+  Value->SetNullable(Nullable);
 
 #ifdef DEBTRACE
  htrc(" colp=%p type=%d value=%p coluse=%.4X status=%.4X\n",

storage/connect/colblk.h

 /*************** Colblk H Declares Source Code File (.H) ***************/
-/*  Name: COLBLK.H    Version 1.5                                      */
+/*  Name: COLBLK.H    Version 1.6                                      */
 /*                                                                     */
-/*  (C) Copyright to the author Olivier BERTRAND          2005-2012    */
+/*  (C) Copyright to the author Olivier BERTRAND          2005-2013    */
 /*                                                                     */
 /*  This file contains the COLBLK and derived classes declares.        */
 /***********************************************************************/
@@ -53,6 +53,8 @@ class DllExport COLBLK : public XOBJECT {
           PSZ     GetDomain(void) {return (Cdp) ? Cdp->Decode : NULL;}
           PSZ     GetDesc(void) {return (Cdp) ? Cdp->Desc : NULL;}
           PSZ     GetFmt(void) {return (Cdp) ? Cdp->Fmt : NULL;}
+          bool    IsNullable(void) {return Nullable;}
+          void    SetNullable(bool b) {Nullable = b;}
                   
   // Methods      
   virtual void    Reset(void);
@@ -81,6 +83,7 @@ class DllExport COLBLK : public XOBJECT {
   PCOLDEF Cdp;                 // To column definition block
   PTDB    To_Tdb;              // Points to Table Descriptor Block
   PXCOL   To_Kcol;             // Points to Xindex matching column
+  bool    Nullable;            // True if nullable
   int     Index;               // Column number in table
   int     Opt;                 // Cluster/sort information
   int     Buf_Type;            // Data type

storage/connect/connect.cc

@@ -273,8 +273,10 @@ bool CntOpenTable(PGLOBAL g, PTDB tdbp, MODE mode, char *c1, char *c2,
       cp= new(g) COLUMN(p + 1);
       cp->SetTo_Table(tdbp->GetTable());
       colp= ((PTDBASE)tdbp)->InsertSpcBlk(g, cp);
-    } else
-      colp= tdbp->ColDB(g, p, 0);
+    } else {
+      colp= tdbp->ColDB(g, p + 1, 0);
+      colp->SetNullable(*p == '1');
+    } // endif p
 
     if (!colp) {
       sprintf(g->Message, "Column %s not found in %s", p, tdbp->GetName());
@@ -338,7 +340,8 @@ bool CntOpenTable(PGLOBAL g, PTDB tdbp, MODE mode, char *c1, char *c2,
       utp->ColDB(g, NULL, 0);
     else for (p= c2; *p; p+= n) {
       // Allocate only used column blocks
-      utp->ColDB(g, p, 0);
+      colp= utp->ColDB(g, p + 1, 0);
+      colp->SetNullable(*p == '1');
       n= strlen(p) + 1;
       } // endfor p
 

storage/connect/ha_connect.cc

@@ -1211,7 +1211,7 @@ int ha_connect::GetColNameLen(Field *fp)
   if (fop && fop->special)
     n= strlen(fop->special) + 1;
   else
-    n= strlen(fp->field_name);
+    n= strlen(fp->field_name) + 1;
 
   return n;
 } // end of GetColNameLen
@@ -1238,7 +1238,7 @@ void ha_connect::AddColName(char *cp, Field *fp)
     // The prefix * mark the column as "special"
     strcat(strcpy(cp, "*"), strupr(fop->special));
   else
-    strcpy(cp, (char*)fp->field_name);
+    strcat(strcpy(cp, fp->maybe_null() ? "1" : "0"), (char*)fp->field_name);
 
 } // end of AddColName
 
@@ -1456,49 +1456,54 @@ int ha_connect::MakeRecord(char *buf)
       value= colp->GetValue();
 
       // All this could be better optimized
-      switch (value->GetType()) {
-        case TYPE_DATE:
-          if (!sdval)
-            sdval= AllocateValue(xp->g, TYPE_STRING, 20);
-
-          switch (fp->type()) {
-            case MYSQL_TYPE_DATE:
-              fmt= "%Y-%m-%d";
-              break;
-            case MYSQL_TYPE_TIME:
-              fmt= "%H:%M:%S";
-              break;
-            default:
-              fmt= "%Y-%m-%d %H:%M:%S";
-            } // endswitch type
-
-          // Get date in the format required by MySQL fields
-          value->FormatValue(sdval, fmt);
-          p= sdval->GetCharValue();
-          break;
-        case TYPE_FLOAT:
-          p= NULL;
-          break;
-        case TYPE_STRING:
-          // Passthru
-        default:
-          p= value->GetCharString(val);
-        } // endswitch Type
-
-      if (p) {
-        if (fp->store(p, strlen(p), charset, CHECK_FIELD_WARN)) {
-          // Avoid "error" on null fields
-          if (value->GetIntValue())
+      if (!value->IsNull()) {
+        switch (value->GetType()) {
+          case TYPE_DATE:
+            if (!sdval)
+              sdval= AllocateValue(xp->g, TYPE_STRING, 20);
+      
+            switch (fp->type()) {
+              case MYSQL_TYPE_DATE:
+                fmt= "%Y-%m-%d";
+                break;
+              case MYSQL_TYPE_TIME:
+                fmt= "%H:%M:%S";
+                break;
+              default:
+                fmt= "%Y-%m-%d %H:%M:%S";
+              } // endswitch type
+      
+            // Get date in the format required by MySQL fields
+            value->FormatValue(sdval, fmt);
+            p= sdval->GetCharValue();
+            break;
+          case TYPE_FLOAT:
+            p= NULL;
+            break;
+          case TYPE_STRING:
+            // Passthru
+          default:
+            p= value->GetCharString(val);
+          } // endswitch Type
+
+        if (p) {
+          if (fp->store(p, strlen(p), charset, CHECK_FIELD_WARN)) {
+            // Avoid "error" on null fields
+            if (value->GetIntValue())
+              rc= HA_ERR_WRONG_IN_RECORD;
+    
+            DBUG_PRINT("MakeRecord", (p));
+            } // endif store
+    
+        } else
+          if (fp->store(value->GetFloatValue())) {
             rc= HA_ERR_WRONG_IN_RECORD;
+            DBUG_PRINT("MakeRecord", (value->GetCharString(val)));
+            } // endif store
 
-          DBUG_PRINT("MakeRecord", (p));
-          } // endif store
-
+        fp->set_notnull();
       } else
-        if (fp->store(value->GetFloatValue())) {
-          rc= HA_ERR_WRONG_IN_RECORD;
-          DBUG_PRINT("MakeRecord", (value->GetCharString(val)));
-          } // endif store
+        fp->set_null();
 
       } // endif bitmap
 
@@ -1552,7 +1557,12 @@ int ha_connect::ScanRecord(PGLOBAL g, uchar *buf)
 
       // This is a used field, fill the value from the row buffer
       // All this could be better optimized
-      switch (value->GetType()) {
+      if (fp->is_null()) {
+        if (colp->IsNullable())
+          value->SetNull(true);
+
+        value->Reset();
+      } else switch (value->GetType()) {
         case TYPE_FLOAT:
           value->SetValue(fp->val_real());
           break;
@@ -3306,7 +3316,7 @@ bool ha_connect::add_fields(THD *thd, void *alt_info,
   alter_info->create_list.push_back(new_field);
 //lex->last_field=new_field;
   DBUG_RETURN(0);
-}
+} // end of add_fields
 
 /**
   @brief

storage/connect/tabmysql.cpp

@@ -109,14 +109,15 @@ MYSQLDEF::MYSQLDEF(void)
 /* An Example:                                                         */
 /*                                                                     */
 /* CREATE TABLE t1 (id int(32))                                        */
-/*   ENGINE="FEDERATEDX"                                               */
+/*   ENGINE="CONNECT" TABLE_TYPE="MYSQL"                               */
 /*   CONNECTION="mysql://joe:pwd@192.168.1.111:9308/dbname/tabname";   */
 /*                                                                     */
 /* CREATE TABLE t2 (                                                   */
 /*   id int(4) NOT NULL auto_increment,                                */
 /*   name varchar(32) NOT NULL,                                        */
 /*   PRIMARY KEY(id)                                                   */
-/*   ) ENGINE="FEDERATEDX" CONNECTION="my_conn";    (NIY)              */
+/*   ) ENGINE="CONNECT" TABLE_TYPE="MYSQL"                             */
+/*   CONNECTION="my_conn";    (NIY)                                    */
 /*                                                                     */
 /*  'password' and 'port' are both optional.                           */
 /*                                                                     */
@@ -1040,8 +1041,12 @@ void MYSQLCOL::ReadColumn(PGLOBAL g)
 
   if ((buf = ((PTDBMY)To_Tdb)->Myc.GetCharField(Rank)))
     Value->SetValue_char(buf, Long);
-  else
+  else {
+    if (Nullable)
+      Value->SetNull(true);
+
     Value->Reset();              // Null value
+    } // endelse
 
   } // end of ReadColumn
 

storage/connect/tabodbc.cpp

@@ -796,9 +796,13 @@ void ODBCCOL::ReadColumn(PGLOBAL g)
 
   if (StrLen[n] == SQL_NULL_DATA) {
     // Null value
+    if (Nullable)
+      Value->SetNull(true);
+
     Value->Reset();
     return;
-    } // endif StrLen
+  } else
+    Value->SetNull(false);
 
   if (Bufp && tdbp->Rows)
     if (Buf_Type == TYPE_DATE)

storage/connect/valblk.h

@@ -27,8 +27,7 @@ class VALBLK : public BLOCK {
 //friend void SemColData(PGLOBAL g, PSEM semp);
  public:
   // Constructors
-  VALBLK(void *mp, int type, int nval)
-                {Blkp = mp; Type = type; Nval = nval; Check = true;}
+  VALBLK(void *mp, int type, int nval);
 
   // Implementation
           int    GetNval(void) {return Nval;}
@@ -37,6 +36,12 @@ class VALBLK : public BLOCK {
           void   SetValPointer(void *mp) {Blkp = mp;}
           int    GetType(void) {return Type;}
           void   SetCheck(bool b) {Check = b;}
+          void   MoveNull(int i, int j)
+                  {if (To_Nulls) To_Nulls[j] = To_Nulls[j];}
+  virtual void   SetNull(int n, bool b)
+                  {if (To_Nulls) {To_Nulls[n] = (b) ? '*' : 0;}}
+  virtual bool   IsNull(int n) {return To_Nulls && To_Nulls[n];}
+  virtual void   SetNullable(bool b);
   virtual void   Init(PGLOBAL g, bool check) = 0;
   virtual int    GetVlen(void) = 0;
   virtual PSZ    GetCharValue(int n);
@@ -56,11 +61,13 @@ class VALBLK : public BLOCK {
   virtual void   SetValue(longlong lval, int n) {assert(false);}
   virtual void   SetValue(PSZ sp, int n) {assert(false);}
   virtual void   SetValue(PVAL valp, int n) = 0;
+  virtual void   SetValue(PVBLK pv, int n1, int n2) = 0;
+#if 0
   virtual void   SetMin(PVAL valp, int n) = 0;
   virtual void   SetMax(PVAL valp, int n) = 0;
-  virtual void   SetValue(PVBLK pv, int n1, int n2) = 0;
   virtual void   SetValues(PVBLK pv, int i, int n) = 0;
   virtual void   AddMinus1(PVBLK pv, int n1, int n2) {assert(false);}
+#endif // 0
   virtual void   Move(int i, int j) = 0;
   virtual int    CompVal(PVAL vp, int n) = 0;
   virtual int    CompVal(int i1, int i2) = 0;
@@ -80,10 +87,12 @@ class VALBLK : public BLOCK {
 
   // Members
   PGLOBAL Global;           // Used for messages and allocation
+  char   *To_Nulls;         // Null values array
   void   *Blkp;             // To value block
   int     Type;             // Type of individual values
   int     Nval;             // Max number of values in block
   bool    Check;            // If true SetValue types must match
+  bool    Nullable;         // True if values can be null
   }; // end of class VALBLK
 
 /***********************************************************************/
@@ -109,10 +118,12 @@ class CHRBLK : public VALBLK {
   // Methods
   virtual void   SetValue(PSZ sp, int n);
   virtual void   SetValue(PVAL valp, int n);
+  virtual void   SetValue(PVBLK pv, int n1, int n2);
+#if 0
   virtual void   SetMin(PVAL valp, int n);
   virtual void   SetMax(PVAL valp, int n);
-  virtual void   SetValue(PVBLK pv, int n1, int n2);
   virtual void   SetValues(PVBLK pv, int k, int n);
+#endif // 0
   virtual void   Move(int i, int j);
   virtual int    CompVal(PVAL vp, int n);
   virtual int    CompVal(int i1, int i2);
@@ -126,7 +137,7 @@ class CHRBLK : public VALBLK {
   char* const &Chrp;             // Pointer to char buffer
   PSZ   Valp;                    // Used to make a zero ended value
   bool  Blanks;                  // True for right filling with blanks
-  bool  Ci;                       // True if case insensitive
+  bool  Ci;                      // True if case insensitive
   int   Long;                    // Length of each string
   }; // end of class CHRBLK
 
@@ -141,6 +152,9 @@ class STRBLK : public VALBLK {
   STRBLK(PGLOBAL g, void *mp, int size);
 
   // Implementation
+  virtual void   SetNull(int n, bool b) {if (b) {Strp[n] = NULL;}}
+  virtual bool   IsNull(int n) {return Strp[n] == NULL;}
+  virtual void   SetNullable(bool b) {}    // Always nullable
   virtual void   Init(PGLOBAL g, bool check);
   virtual int    GetVlen(void) {return sizeof(PSZ);}
   virtual PSZ    GetCharValue(int n) {return Strp[n];}
@@ -153,10 +167,12 @@ class STRBLK : public VALBLK {
   // Methods
   virtual void   SetValue(PSZ sp, int n);
   virtual void   SetValue(PVAL valp, int n);
+  virtual void   SetValue(PVBLK pv, int n1, int n2);
+#if 0
   virtual void   SetMin(PVAL valp, int n);
   virtual void   SetMax(PVAL valp, int n);
-  virtual void   SetValue(PVBLK pv, int n1, int n2);
   virtual void   SetValues(PVBLK pv, int k, int n);
+#endif // 0
   virtual void   Move(int i, int j);
   virtual int    CompVal(PVAL vp, int n);
   virtual int    CompVal(int i1, int i2);
@@ -190,15 +206,20 @@ class SHRBLK : public VALBLK {
 
   // Methods
   virtual void   SetValue(PSZ sp, int n);
-  virtual void   SetValue(short sval, int n) {Shrp[n] = sval;}
-  virtual void   SetValue(int lval, int n) {Shrp[n] = (short)lval;}
-  virtual void   SetValue(longlong lval, int n) {Shrp[n] = (short)lval;}
+  virtual void   SetValue(short sval, int n)
+                  {Shrp[n] = sval; SetNull(n, false);}
+  virtual void   SetValue(int lval, int n)
+                  {Shrp[n] = (short)lval; SetNull(n, false);}
+  virtual void   SetValue(longlong lval, int n)
+                  {Shrp[n] = (short)lval; SetNull(n, false);}
   virtual void   SetValue(PVAL valp, int n);
+  virtual void   SetValue(PVBLK pv, int n1, int n2);
+#if 0
   virtual void   SetMin(PVAL valp, int n);
   virtual void   SetMax(PVAL valp, int n);
-  virtual void   SetValue(PVBLK pv, int n1, int n2);
   virtual void   SetValues(PVBLK pv, int k, int n);
   virtual void   AddMinus1(PVBLK pv, int n1, int n2);
+#endif // 0
   virtual void   Move(int i, int j);
   virtual int    CompVal(PVAL vp, int n);
   virtual int    CompVal(int i1, int i2);
@@ -232,15 +253,20 @@ class LNGBLK : public VALBLK {
 
   // Methods
   virtual void   SetValue(PSZ sp, int n);
-  virtual void   SetValue(short sval, int n) {Lngp[n] = (int)sval;}
-  virtual void   SetValue(int lval, int n) {Lngp[n] = lval;}
-  virtual void   SetValue(longlong lval, int n) {Lngp[n] = (int)lval;}
+  virtual void   SetValue(short sval, int n)
+                  {Lngp[n] = (int)sval; SetNull(n, false);}
+  virtual void   SetValue(int lval, int n)
+                  {Lngp[n] = lval; SetNull(n, false);}
+  virtual void   SetValue(longlong lval, int n)
+                  {Lngp[n] = (int)lval; SetNull(n, false);}
   virtual void   SetValue(PVAL valp, int n);
+  virtual void   SetValue(PVBLK pv, int n1, int n2);
+#if 0
   virtual void   SetMin(PVAL valp, int n);
   virtual void   SetMax(PVAL valp, int n);
-  virtual void   SetValue(PVBLK pv, int n1, int n2);
   virtual void   SetValues(PVBLK pv, int k, int n);
   virtual void   AddMinus1(PVBLK pv, int n1, int n2);
+#endif // 0
   virtual void   Move(int i, int j);
   virtual int    CompVal(PVAL vp, int n);
   virtual int    CompVal(int i1, int i2);
@@ -293,15 +319,20 @@ class BIGBLK : public VALBLK {
 
   // Methods
   virtual void   SetValue(PSZ sp, int n);
-  virtual void   SetValue(short sval, int n) {Lngp[n] = (longlong)sval;}
-  virtual void   SetValue(int lval, int n) {Lngp[n] = (longlong)lval;}
-  virtual void   SetValue(longlong lval, int n) {Lngp[n] = lval;}
+  virtual void   SetValue(short sval, int n)
+                  {Lngp[n] = (longlong)sval; SetNull(n, false);}
+  virtual void   SetValue(int lval, int n)
+                  {Lngp[n] = (longlong)lval; SetNull(n, false);}
+  virtual void   SetValue(longlong lval, int n)
+                  {Lngp[n] = lval; SetNull(n, false);}
   virtual void   SetValue(PVAL valp, int n);
+  virtual void   SetValue(PVBLK pv, int n1, int n2);
+#if 0
   virtual void   SetMin(PVAL valp, int n);
   virtual void   SetMax(PVAL valp, int n);
-  virtual void   SetValue(PVBLK pv, int n1, int n2);
   virtual void   SetValues(PVBLK pv, int k, int n);
   virtual void   AddMinus1(PVBLK pv, int n1, int n2);
+#endif // 0
   virtual void   Move(int i, int j);
   virtual int    CompVal(PVAL vp, int n);
   virtual int    CompVal(int i1, int i2);
@@ -337,10 +368,12 @@ class DBLBLK : public VALBLK {
   // Methods
   virtual void   SetValue(PSZ sp, int n);
   virtual void   SetValue(PVAL valp, int n);
+  virtual void   SetValue(PVBLK pv, int n1, int n2);
+#if 0
   virtual void   SetMin(PVAL valp, int n);
   virtual void   SetMax(PVAL valp, int n);
-  virtual void   SetValue(PVBLK pv, int n1, int n2);
   virtual void   SetValues(PVBLK pv, int k, int n);
+#endif // 0
   virtual void   Move(int i, int j);
   virtual int    CompVal(PVAL vp, int n);
   virtual int    CompVal(int i1, int i2);

storage/connect/xindex.cpp

@@ -279,7 +279,7 @@ int XINDEX::Qcompare(int *i1, int *i2)
 bool XINDEX::Make(PGLOBAL g, PIXDEF sxp)
   {
   /*********************************************************************/
-  /*  Table can be accessed through an index.                         */
+  /*  Table can be accessed through an index.                          */
   /*********************************************************************/
   int     k, rc = RC_OK;
   int   *bof, i, j, n, ndf, nkey;
@@ -2861,6 +2861,12 @@ bool KXYCOL::Init(PGLOBAL g, PCOL colp, int n, bool sm, int kln)
   {
   int len = colp->GetLength(), prec = colp->GetPrecision();
 
+  // Currently no indexing on NULL columns
+  if (colp->IsNullable()) {
+    sprintf(g->Message, "Cannot index nullable column %s", colp->GetName());
+    return true;
+    } // endif nullable
+
   if (kln && len > kln && colp->GetResultType() == TYPE_STRING) {
     len = kln;
     Prefix = true;