diff --git a/mysql-test/r/ndb_index_ordered.result b/mysql-test/r/ndb_index_ordered.result
index 7558e5cdb11..9aa7559d660 100644
--- a/mysql-test/r/ndb_index_ordered.result
+++ b/mysql-test/r/ndb_index_ordered.result
@@ -432,16 +432,16 @@ index(ye),
 index(ti),
 index(ts)
 ) engine=ndb;
-insert into t1 (pk,dt,da,ye,ti) values
-(1, '1901-05-05 23:00:59', '1901-05-05', '1901', '23:00:59'),
-(2, '1912-09-05 13:00:59', '1912-09-05', '1912', '13:00:59'),
-(3, '1945-12-31 00:00:00', '1945-12-31', '1945', '00:00:00'),
-(4, '1955-12-31 00:00:00', '1955-12-31', '1955', '00:00:00'),
-(5, '1963-06-06 06:06:06', '1963-06-06', '1963', '06:06:06'),
-(6, '1993-06-06 06:06:06', '1993-06-06', '1993', '06:06:06'),
-(7, '2001-01-01 10:11:10', '2001-01-01', '2001', '10:11:10'),
-(8, '2001-01-01 10:11:11', '2001-01-01', '2001', '10:11:11'),
-(9, '2005-01-31 23:59:59', '2005-01-31', '2005', '23:59:59');
+insert into t1 (pk,dt,da,ye,ti,ts) values
+(1, '1901-05-05 23:00:59', '1901-05-05', '1901', '23:00:59', '2001-01-01 23:00:59'),
+(2, '1912-09-05 13:00:59', '1912-09-05', '1912', '13:00:59', '2001-01-01 13:00:59'),
+(3, '1945-12-31 00:00:00', '1945-12-31', '1945', '00:00:00', '2001-01-01 00:00:00'),
+(4, '1955-12-31 00:00:00', '1955-12-31', '1955', '00:00:00', '2001-01-01 00:00:00'),
+(5, '1963-06-06 06:06:06', '1963-06-06', '1963', '06:06:06', '2001-01-01 06:06:06'),
+(6, '1993-06-06 06:06:06', '1993-06-06', '1993', '06:06:06', '2001-01-01 06:06:06'),
+(7, '2001-01-01 10:11:10', '2001-01-01', '2001', '10:11:10', '2001-01-01 10:11:10'),
+(8, '2001-01-01 10:11:11', '2001-01-01', '2001', '10:11:11', '2001-01-01 10:11:11'),
+(9, '2005-01-31 23:59:59', '2005-01-31', '2005', '23:59:59', '2001-01-01 23:59:59');
 select count(*)-9 from t1 use index (dt) where dt >  '1900-01-01 00:00:00';
 count(*)-9
 0
@@ -529,6 +529,30 @@ count(*)-8
 select count(*)-9 from t1 use index (ti) where ti <= '23:59:59';
 count(*)-9
 0
+select count(*)-9 from t1 use index (ts) where ts >= '2001-01-01 00:00:00';
+count(*)-9
+0
+select count(*)-7 from t1 use index (ts) where ts >  '2001-01-01 00:00:00';
+count(*)-7
+0
+select count(*)-7 from t1 use index (ts) where ts >  '2001-01-01 05:05:05';
+count(*)-7
+0
+select count(*)-5 from t1 use index (ts) where ts >  '2001-01-01 06:06:06';
+count(*)-5
+0
+select count(*)-5 from t1 use index (ts) where ts <  '2001-01-01 10:11:11';
+count(*)-5
+0
+select count(*)-6 from t1 use index (ts) where ts <= '2001-01-01 10:11:11';
+count(*)-6
+0
+select count(*)-8 from t1 use index (ts) where ts <  '2001-01-01 23:59:59';
+count(*)-8
+0
+select count(*)-9 from t1 use index (ts) where ts <= '2001-01-01 23:59:59';
+count(*)-9
+0
 drop table t1;
 create table t1(a int primary key, b int not null, index(b));
 insert into t1 values (1,1), (2,2);
diff --git a/mysql-test/t/ndb_index_ordered.test b/mysql-test/t/ndb_index_ordered.test
index 1e70eac181e..bea21f65381 100644
--- a/mysql-test/t/ndb_index_ordered.test
+++ b/mysql-test/t/ndb_index_ordered.test
@@ -220,16 +220,16 @@ create table t1 (
   index(ts)
 ) engine=ndb;
 
-insert into t1 (pk,dt,da,ye,ti) values
-  (1, '1901-05-05 23:00:59', '1901-05-05', '1901', '23:00:59'),
-  (2, '1912-09-05 13:00:59', '1912-09-05', '1912', '13:00:59'),
-  (3, '1945-12-31 00:00:00', '1945-12-31', '1945', '00:00:00'),
-  (4, '1955-12-31 00:00:00', '1955-12-31', '1955', '00:00:00'),
-  (5, '1963-06-06 06:06:06', '1963-06-06', '1963', '06:06:06'),
-  (6, '1993-06-06 06:06:06', '1993-06-06', '1993', '06:06:06'),
-  (7, '2001-01-01 10:11:10', '2001-01-01', '2001', '10:11:10'),
-  (8, '2001-01-01 10:11:11', '2001-01-01', '2001', '10:11:11'),
-  (9, '2005-01-31 23:59:59', '2005-01-31', '2005', '23:59:59');
+insert into t1 (pk,dt,da,ye,ti,ts) values
+  (1, '1901-05-05 23:00:59', '1901-05-05', '1901', '23:00:59', '2001-01-01 23:00:59'),
+  (2, '1912-09-05 13:00:59', '1912-09-05', '1912', '13:00:59', '2001-01-01 13:00:59'),
+  (3, '1945-12-31 00:00:00', '1945-12-31', '1945', '00:00:00', '2001-01-01 00:00:00'),
+  (4, '1955-12-31 00:00:00', '1955-12-31', '1955', '00:00:00', '2001-01-01 00:00:00'),
+  (5, '1963-06-06 06:06:06', '1963-06-06', '1963', '06:06:06', '2001-01-01 06:06:06'),
+  (6, '1993-06-06 06:06:06', '1993-06-06', '1993', '06:06:06', '2001-01-01 06:06:06'),
+  (7, '2001-01-01 10:11:10', '2001-01-01', '2001', '10:11:10', '2001-01-01 10:11:10'),
+  (8, '2001-01-01 10:11:11', '2001-01-01', '2001', '10:11:11', '2001-01-01 10:11:11'),
+  (9, '2005-01-31 23:59:59', '2005-01-31', '2005', '23:59:59', '2001-01-01 23:59:59');
 
 # datetime
 select count(*)-9 from t1 use index (dt) where dt >  '1900-01-01 00:00:00';
@@ -268,6 +268,16 @@ select count(*)-6 from t1 use index (ti) where ti <= '10:11:11';
 select count(*)-8 from t1 use index (ti) where ti <  '23:59:59';
 select count(*)-9 from t1 use index (ti) where ti <= '23:59:59';
 
+# timestamp
+select count(*)-9 from t1 use index (ts) where ts >= '2001-01-01 00:00:00';
+select count(*)-7 from t1 use index (ts) where ts >  '2001-01-01 00:00:00';
+select count(*)-7 from t1 use index (ts) where ts >  '2001-01-01 05:05:05';
+select count(*)-5 from t1 use index (ts) where ts >  '2001-01-01 06:06:06';
+select count(*)-5 from t1 use index (ts) where ts <  '2001-01-01 10:11:11';
+select count(*)-6 from t1 use index (ts) where ts <= '2001-01-01 10:11:11';
+select count(*)-8 from t1 use index (ts) where ts <  '2001-01-01 23:59:59';
+select count(*)-9 from t1 use index (ts) where ts <= '2001-01-01 23:59:59';
+
 drop table t1;
 
 # bug#7798
diff --git a/ndb/include/kernel/signaldata/DictTabInfo.hpp b/ndb/include/kernel/signaldata/DictTabInfo.hpp
index 6ae8a8a2985..61c043f2998 100644
--- a/ndb/include/kernel/signaldata/DictTabInfo.hpp
+++ b/ndb/include/kernel/signaldata/DictTabInfo.hpp
@@ -276,7 +276,9 @@ public:
     ExtBit = NdbSqlUtil::Type::Bit,
     ExtLongvarchar = NdbSqlUtil::Type::Longvarchar,
     ExtLongvarbinary = NdbSqlUtil::Type::Longvarbinary,
-    ExtTime = NdbSqlUtil::Type::Time
+    ExtTime = NdbSqlUtil::Type::Time,
+    ExtYear = NdbSqlUtil::Type::Year,
+    ExtTimestamp = NdbSqlUtil::Type::Timestamp
   };
 
   // Attribute data interpretation
@@ -389,6 +391,14 @@ public:
         AttributeSize = DictTabInfo::an8Bit;
         AttributeArraySize = 3 * AttributeExtLength;
         break;
+      case DictTabInfo::ExtYear:
+        AttributeSize = DictTabInfo::an8Bit;
+        AttributeArraySize = 1 * AttributeExtLength;
+        break;
+      case DictTabInfo::ExtTimestamp:
+        AttributeSize = DictTabInfo::an8Bit;
+        AttributeArraySize = 4 * AttributeExtLength;
+        break;
       default:
         return false;
       };
diff --git a/ndb/include/ndbapi/NdbDictionary.hpp b/ndb/include/ndbapi/NdbDictionary.hpp
index 80c0831f675..91cfc0cb22a 100644
--- a/ndb/include/ndbapi/NdbDictionary.hpp
+++ b/ndb/include/ndbapi/NdbDictionary.hpp
@@ -196,7 +196,9 @@ public:
       Bit = NDB_TYPE_BIT,          ///< Bit, length specifies no of bits
       Longvarchar = NDB_TYPE_LONG_VARCHAR,  ///< Length bytes: 2, little-endian
       Longvarbinary = NDB_TYPE_LONG_VARBINARY, ///< Length bytes: 2, little-endian
-      Time = NDB_TYPE_TIME         ///< Time without date
+      Time = NDB_TYPE_TIME,        ///< Time without date
+      Year = NDB_TYPE_YEAR,   ///< Year 1901-2155 (1 byte)
+      Timestamp = NDB_TYPE_TIMESTAMP  ///< Unix time
     };
 
     /** 
diff --git a/ndb/include/util/NdbSqlUtil.hpp b/ndb/include/util/NdbSqlUtil.hpp
index 75e2a819174..a8a2979558a 100644
--- a/ndb/include/util/NdbSqlUtil.hpp
+++ b/ndb/include/util/NdbSqlUtil.hpp
@@ -92,7 +92,9 @@ public:
       Bit = NDB_TYPE_BIT,
       Longvarchar = NDB_TYPE_LONG_VARCHAR,
       Longvarbinary = NDB_TYPE_LONG_VARBINARY,
-      Time = NDB_TYPE_TIME
+      Time = NDB_TYPE_TIME,
+      Year = NDB_TYPE_YEAR,
+      Timestamp = NDB_TYPE_TIMESTAMP
     };
     Enum m_typeId;      // redundant
     Cmp* m_cmp;         // comparison method
@@ -161,6 +163,8 @@ private:
   static Cmp cmpLongvarchar;
   static Cmp cmpLongvarbinary;
   static Cmp cmpTime;
+  static Cmp cmpYear;
+  static Cmp cmpTimestamp;
 };
 
 #endif
diff --git a/ndb/src/common/util/NdbSqlUtil.cpp b/ndb/src/common/util/NdbSqlUtil.cpp
index 82f99d0e7fd..d5f06f119ee 100644
--- a/ndb/src/common/util/NdbSqlUtil.cpp
+++ b/ndb/src/common/util/NdbSqlUtil.cpp
@@ -179,6 +179,14 @@ NdbSqlUtil::m_typeList[] = {
   {
     Type::Time,
     cmpTime
+  },
+  {
+    Type::Year,
+    cmpYear
+  },
+  {
+    Type::Timestamp,
+    cmpTimestamp
   }
 };
 
@@ -670,6 +678,40 @@ NdbSqlUtil::cmpLongvarbinary(const void* info, const void* p1, unsigned n1, cons
   return CmpUnknown;
 }
 
+int
+NdbSqlUtil::cmpYear(const void* info, const void* p1, unsigned n1, const void* p2, unsigned n2, bool full)
+{
+  if (n2 >= sizeof(Uint8)) {
+    Uint8 v1, v2;
+    memcpy(&v1, p1, sizeof(Uint8));
+    memcpy(&v2, p2, sizeof(Uint8));
+    if (v1 < v2)
+      return -1;
+    if (v1 > v2)
+      return +1;
+    return 0;
+  }
+  assert(! full);
+  return CmpUnknown;
+}
+
+int
+NdbSqlUtil::cmpTimestamp(const void* info, const void* p1, unsigned n1, const void* p2, unsigned n2, bool full)
+{
+  if (n2 >= sizeof(Uint32)) {
+    Uint32 v1, v2;
+    memcpy(&v1, p1, sizeof(Uint32));
+    memcpy(&v2, p2, sizeof(Uint32));
+    if (v1 < v2)
+      return -1;
+    if (v1 > v2)
+      return +1;
+    return 0;
+  }
+  assert(! full);
+  return CmpUnknown;
+}
+
 // check charset
 
 bool
diff --git a/ndb/src/ndbapi/NdbDictionary.cpp b/ndb/src/ndbapi/NdbDictionary.cpp
index 93a82e4bc83..346f5de9eb5 100644
--- a/ndb/src/ndbapi/NdbDictionary.cpp
+++ b/ndb/src/ndbapi/NdbDictionary.cpp
@@ -980,6 +980,12 @@ operator<<(NdbOut& out, const NdbDictionary::Column& col)
   case NdbDictionary::Column::Time:
     out << "Time";
     break;
+  case NdbDictionary::Column::Year:
+    out << "Year";
+    break;
+  case NdbDictionary::Column::Timestamp:
+    out << "Timestamp";
+    break;
   case NdbDictionary::Column::Undefined:
     out << "Undefined";
     break;
diff --git a/ndb/src/ndbapi/NdbDictionaryImpl.cpp b/ndb/src/ndbapi/NdbDictionaryImpl.cpp
index 498c5716c73..ca118cd0994 100644
--- a/ndb/src/ndbapi/NdbDictionaryImpl.cpp
+++ b/ndb/src/ndbapi/NdbDictionaryImpl.cpp
@@ -144,6 +144,8 @@ NdbColumnImpl::init(Type t)
     m_cs = default_cs;
     break;
   case Time:
+  case Year:
+  case Timestamp:
     m_precision = 0;
     m_scale = 0;
     m_length = 1;
diff --git a/ndb/src/ndbapi/NdbRecAttr.cpp b/ndb/src/ndbapi/NdbRecAttr.cpp
index 30ba5c112fa..993fc8b6327 100644
--- a/ndb/src/ndbapi/NdbRecAttr.cpp
+++ b/ndb/src/ndbapi/NdbRecAttr.cpp
@@ -178,10 +178,11 @@ NdbOut& operator<<(NdbOut& out, const NdbRecAttr &r)
     return out;
   }
 
-  if (r.arraySize() > 1)
+  uint length = r.getColumn()->getLength();
+  if (length > 1)
     out << "[";
 
-  for (Uint32 j = 0; j < r.arraySize(); j++) 
+  for (Uint32 j = 0; j < length; j++) 
   {
     if (j > 0)
       out << " ";
@@ -221,13 +222,13 @@ NdbOut& operator<<(NdbOut& out, const NdbRecAttr &r)
 	break;
       case NdbDictionary::Column::Char:
 	ndbrecattr_print_string(out,"Char",r.aRef(),r.arraySize());
-	j = r.arraySize();
+	j = length;
 	break;
       case NdbDictionary::Column::Varchar:
         {
           unsigned len = *(const unsigned char*)r.aRef();
           ndbrecattr_print_string(out,"Varchar", r.aRef()+1,len);
-          j = r.arraySize();
+          j = length;
         }
 	break;
       case NdbDictionary::Column::Float:
@@ -236,6 +237,74 @@ NdbOut& operator<<(NdbOut& out, const NdbRecAttr &r)
       case NdbDictionary::Column::Double:
 	out << r.double_value();
 	break;
+      // for dates cut-and-paste from field.cc
+      case NdbDictionary::Column::Datetime:
+        {
+          ulonglong tmp=r.u_64_value();
+          long part1,part2,part3;
+          part1=(long) (tmp/LL(1000000));
+          part2=(long) (tmp - (ulonglong) part1*LL(1000000));
+          char buf[40];
+          char* pos=(char*) buf+19;
+          *pos--=0;
+          *pos--= (char) ('0'+(char) (part2%10)); part2/=10; 
+          *pos--= (char) ('0'+(char) (part2%10)); part3= (int) (part2 / 10);
+          *pos--= ':';
+          *pos--= (char) ('0'+(char) (part3%10)); part3/=10;
+          *pos--= (char) ('0'+(char) (part3%10)); part3/=10;
+          *pos--= ':';
+          *pos--= (char) ('0'+(char) (part3%10)); part3/=10;
+          *pos--= (char) ('0'+(char) part3);
+          *pos--= '/';
+          *pos--= (char) ('0'+(char) (part1%10)); part1/=10;
+          *pos--= (char) ('0'+(char) (part1%10)); part1/=10;
+          *pos--= '-';
+          *pos--= (char) ('0'+(char) (part1%10)); part1/=10;
+          *pos--= (char) ('0'+(char) (part1%10)); part3= (int) (part1/10);
+          *pos--= '-';
+          *pos--= (char) ('0'+(char) (part3%10)); part3/=10;
+          *pos--= (char) ('0'+(char) (part3%10)); part3/=10;
+          *pos--= (char) ('0'+(char) (part3%10)); part3/=10;
+          *pos=(char) ('0'+(char) part3);
+          out << buf;
+        }
+	break;
+      case NdbDictionary::Column::Date:
+        {
+          uint tmp=uint3korr(r.aRef());
+          int year=(int) ((uint32) tmp/10000L % 10000);
+          int month=(int) ((uint32) tmp/100 % 100);
+          int day=(int) ((uint32) tmp % 100);
+          char buf[40];
+          sprintf(buf, "%04d-%02d-%02d", year, month, day);
+          out << buf;
+        }
+	break;
+      case NdbDictionary::Column::Time:
+        {
+          long tmp=(long) sint3korr(r.aRef());
+          int hour=(uint) (tmp/10000);
+          int minute=(uint) (tmp/100 % 100);
+          int second=(uint) (tmp % 100);
+          char buf[40];
+          sprintf(buf, "%02d:%02d:%02d", hour, minute, second);
+          out << buf;
+        }
+	break;
+      case NdbDictionary::Column::Year:
+        {
+          uint year = 1900 + r.u_char_value();
+          char buf[40];
+          sprintf(buf, "%04d", year);
+          out << buf;
+        }
+	break;
+      case NdbDictionary::Column::Timestamp:
+        {
+          time_t time = r.u_32_value();
+          out << (uint)time;
+        }
+	break;
       case NdbDictionary::Column::Blob:
         {
           const NdbBlob::Head* h = (const NdbBlob::Head*)r.aRef();
@@ -244,7 +313,7 @@ NdbOut& operator<<(NdbOut& out, const NdbRecAttr &r)
           unsigned n = r.arraySize() - sizeof(*h);
           for (unsigned k = 0; k < n && k < h->length; k++)
             out.print("%02X", (int)p[k]);
-          j = r.arraySize();
+          j = length;
         }
         break;
       case NdbDictionary::Column::Text:
@@ -255,26 +324,26 @@ NdbOut& operator<<(NdbOut& out, const NdbRecAttr &r)
           unsigned n = r.arraySize() - sizeof(*h);
           for (unsigned k = 0; k < n && k < h->length; k++)
             out.print("%c", (int)p[k]);
-          j = r.arraySize();
+          j = length;
         }
         break;
       case NdbDictionary::Column::Longvarchar:
         {
           unsigned len = uint2korr(r.aRef());
           ndbrecattr_print_string(out,"Longvarchar", r.aRef()+2,len);
-          j = r.arraySize();
+          j = length;
         }
         break;
       default: /* no print functions for the rest, just print type */
 	out << (int) r.getType();
-	j = r.arraySize();
+	j = length;
 	if (j > 1)
 	  out << " " << j << " times";
 	break;
       }
   }
 
-  if (r.arraySize() > 1)
+  if (length > 1)
   {
     out << "]";
   }
diff --git a/ndb/test/include/NdbSchemaOp.hpp b/ndb/test/include/NdbSchemaOp.hpp
index e2fb4015b88..77e704c0e5c 100644
--- a/ndb/test/include/NdbSchemaOp.hpp
+++ b/ndb/test/include/NdbSchemaOp.hpp
@@ -575,10 +575,6 @@ convertColumnTypeToAttrType(NdbDictionary::Column::Type _type)
   case NdbDictionary::Column::Binary:
   case NdbDictionary::Column::Varbinary:
     return String;
-  case NdbDictionary::Column::Datetime:
-  case NdbDictionary::Column::Date:
-  case NdbDictionary::Column::Time:
-  case NdbDictionary::Column::Undefined:
   default:
     return NoAttrTypeDef;
   }
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index cfd1698ae2d..f2b159353e3 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -3472,14 +3472,14 @@ static int create_ndb_column(NDBCOL &col,
     col.setLength(1);
     break;
   // Date types
-  case MYSQL_TYPE_TIMESTAMP:
-    col.setType(NDBCOL::Unsigned);
-    col.setLength(1);
-    break;
   case MYSQL_TYPE_DATETIME:    
     col.setType(NDBCOL::Datetime);
     col.setLength(1);
     break;
+  case MYSQL_TYPE_DATE: // ?
+    col.setType(NDBCOL::Char);
+    col.setLength(field->pack_length());
+    break;
   case MYSQL_TYPE_NEWDATE:
     col.setType(NDBCOL::Date);
     col.setLength(1);
@@ -3488,10 +3488,13 @@ static int create_ndb_column(NDBCOL &col,
     col.setType(NDBCOL::Time);
     col.setLength(1);
     break;
-  case MYSQL_TYPE_DATE: // ?
-  case MYSQL_TYPE_YEAR:        
-    col.setType(NDBCOL::Char);
-    col.setLength(field->pack_length());
+  case MYSQL_TYPE_YEAR:
+    col.setType(NDBCOL::Year);
+    col.setLength(1);
+    break;
+  case MYSQL_TYPE_TIMESTAMP:
+    col.setType(NDBCOL::Timestamp);
+    col.setLength(1);
     break;
   // Char types
   case MYSQL_TYPE_STRING: