common/util/NdbPack.cpp

/*
   Copyright (c) 2011, Oracle and/or its affiliates. All rights reserved.

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License, version 2.0,
   as published by the Free Software Foundation.

   This program is also distributed with certain software (including
   but not limited to OpenSSL) that is licensed under separate terms,
   as designated in a particular file or component or in included license
   documentation.  The authors of MySQL hereby grant you an additional
   permission to link the program and your derivative works with the
   separately licensed software that they have included with MySQL.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License, version 2.0, for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
*/

#include <ndb_global.h>
#include <NdbPack.hpp>
#include <NdbOut.hpp>
#include <NdbEnv.h>

// NdbPack::Error

int
NdbPack::Error::get_error_code() const
{
  return m_error_code;
}

int
NdbPack::Error::get_error_line() const
{
  return m_error_line;
}

void
NdbPack::Error::set_error(int code, int line) const
{
  m_error_code = code;
  m_error_line = line;
#ifdef VM_TRACE
  const char* p = NdbEnv_GetEnv("NDB_PACK_ABORT_ON_ERROR", (char*)0, 0);
  if (p != 0 && strchr("1Y", p[0]) != 0)
    require(false);
#endif
}

void
NdbPack::Error::set_error(const Error& e2) const
{
  set_error(e2.m_error_code, e2.m_error_line);
}

// NdbPack::Endian

void
NdbPack::Endian::convert(void* ptr, Uint32 len)
{
  Uint8* p = (Uint8*)ptr;
  for (Uint32 i = 0; i < len / 2; i++)
  {
    Uint32 j = len - i - 1;
    Uint8 tmp = p[i];
    p[i] = p[j];
    p[j] = tmp;
  }
}

// NdbPack::Type

struct Ndb_pack_type_info {
  bool m_supported;
  Uint16 m_fixSize;     // if non-zero must have this exact size
  Uint16 m_arrayType;   // 0,1,2 length bytes
  bool m_charType;      // type with character set
  bool m_convert;       // convert endian (reverse byte order)
};

static const Ndb_pack_type_info
g_ndb_pack_type_info[] = {
  { 0, 0, 0, 0, 0 }, // NDB_TYPE_UNDEFINED
  { 1, 1, 0, 0, 1 }, // NDB_TYPE_TINYINT
  { 1, 1, 0, 0, 1 }, // NDB_TYPE_TINYUNSIGNED
  { 1, 2, 0, 0, 1 }, // NDB_TYPE_SMALLINT
  { 1, 2, 0, 0, 1 }, // NDB_TYPE_SMALLUNSIGNED
  { 1, 3, 0, 0, 1 }, // NDB_TYPE_MEDIUMINT
  { 1, 3, 0, 0, 1 }, // NDB_TYPE_MEDIUMUNSIGNED
  { 1, 4, 0, 0, 1 }, // NDB_TYPE_INT
  { 1, 4, 0, 0, 1 }, // NDB_TYPE_UNSIGNED
  { 1, 8, 0, 0, 1 }, // NDB_TYPE_BIGINT
  { 1, 8, 0, 0, 1 }, // NDB_TYPE_BIGUNSIGNED
  { 1, 4, 0, 0, 1 }, // NDB_TYPE_FLOAT
  { 1, 8, 0, 0, 1 }, // NDB_TYPE_DOUBLE
  { 1, 0, 0, 0, 0 }, // NDB_TYPE_OLDDECIMAL
  { 1, 0, 0, 1, 0 }, // NDB_TYPE_CHAR
  { 1, 0, 1, 1, 0 }, // NDB_TYPE_VARCHAR
  { 1, 0, 0, 0, 0 }, // NDB_TYPE_BINARY
  { 1, 0, 1, 0, 0 }, // NDB_TYPE_VARBINARY
  { 1, 8, 0, 0, 0 }, // NDB_TYPE_DATETIME
  { 1, 3, 0, 0, 0 }, // NDB_TYPE_DATE
  { 0, 0, 0, 0, 0 }, // NDB_TYPE_BLOB
  { 0, 0, 0, 1, 0 }, // NDB_TYPE_TEXT
  { 0, 0, 0, 0, 0 }, // NDB_TYPE_BIT
  { 1, 0, 2, 1, 0 }, // NDB_TYPE_LONGVARCHAR
  { 1, 0, 2, 0, 0 }, // NDB_TYPE_LONGVARBINARY
  { 1, 3, 0, 0, 0 }, // NDB_TYPE_TIME
  { 1, 1, 0, 0, 0 }, // NDB_TYPE_YEAR
  { 1, 4, 0, 0, 0 }, // NDB_TYPE_TIMESTAMP
  { 1, 0, 0, 0, 0 }, // NDB_TYPE_OLDDECIMALUNSIGNED
  { 1, 0, 0, 0, 0 }, // NDB_TYPE_DECIMAL
  { 1, 0, 0, 0, 0 }  // NDB_TYPE_DECIMALUNSIGNED
};

static const int g_ndb_pack_type_info_cnt =
  sizeof(g_ndb_pack_type_info) / sizeof(g_ndb_pack_type_info[0]);

int
NdbPack::Type::complete()
{
  if (m_typeId == 0)
  {
    set_error(TypeNotSet, __LINE__);
    return -1;
  }
  if (m_typeId >= g_ndb_pack_type_info_cnt)
  {
    set_error(TypeNotSet, __LINE__);
    return -1;
  }
  const Ndb_pack_type_info& info = g_ndb_pack_type_info[m_typeId];
  if (!info.m_supported)
  {
    set_error(TypeNotSupported, __LINE__);
    return -1;
  }
  if (m_byteSize == 0)
  {
    set_error(TypeSizeZero, __LINE__);
    return -1;
  }
  if (info.m_fixSize != 0 && m_byteSize != info.m_fixSize)
  {
    set_error(TypeFixSizeInvalid, __LINE__);
    return -1;
  }
  if (!(m_nullable <= 1))
  {
    set_error(TypeNullableNotBool, __LINE__);
    return -1;
  }
  if (info.m_charType && m_csNumber == 0)
  {
    set_error(CharsetNotSpecified, __LINE__);
    return -1;
  }
  if (info.m_charType && all_charsets[m_csNumber] == 0)
  {
    CHARSET_INFO* cs = get_charset(m_csNumber, MYF(0));
    if (cs == 0)
    {
      set_error(CharsetNotFound, __LINE__);
      return -1;
    }
    all_charsets[m_csNumber] = cs; // yes caller must do this
  }
  if (!info.m_charType && m_csNumber != 0)
  {
    set_error(CharsetNotAllowed, __LINE__);
    return -1;
  }
  m_arrayType = info.m_arrayType;
  return 0;
}

// NdbPack::Spec

int
NdbPack::Spec::add(Type type)
{
  Uint32 cnt = m_cnt;
  Uint32 nullable_cnt = m_nullableCnt;
  Uint32 varsize_cnt = m_varsizeCnt;
  Uint32 max_byte_size = m_maxByteSize;
  if (type.complete() == -1)
  {
    set_error(type);
    return -1;
  }
  type.m_nullbitPos = 0xFFFF;
  if (type.m_nullable)
  {
    type.m_nullbitPos = nullable_cnt;
    nullable_cnt++;
  }
  if (type.m_arrayType != 0)
  {
    varsize_cnt++;
  }
  max_byte_size += type.m_byteSize;
  if (cnt >= m_bufMaxCnt)
  {
    set_error(SpecBufOverflow, __LINE__);
    return -1;
  }
  m_buf[cnt] = type;
  cnt++;
  m_cnt = cnt;
  m_nullableCnt = nullable_cnt;
  m_varsizeCnt = varsize_cnt;
  m_maxByteSize = max_byte_size;
  return 0;
}

int
NdbPack::Spec::add(Type type, Uint32 cnt)
{
  for (Uint32 i = 0; i < cnt; i++)
  {
    if (add(type) == -1)
      return -1;
  }
  return 0;
}

void
NdbPack::Spec::copy(const Spec& s2)
{
  assert(m_bufMaxCnt >= s2.m_cnt);
  reset();
  m_cnt = s2.m_cnt;
  m_nullableCnt = s2.m_nullableCnt;
  m_varsizeCnt = s2.m_varsizeCnt;
  m_maxByteSize = s2.m_maxByteSize;
  for (Uint32 i = 0; i < m_cnt; i++)
  {
    m_buf[i] = s2.m_buf[i];
  }
}

// NdbPack::Iter

int
NdbPack::Iter::desc(const Uint8* item)
{
  const Uint32 i = m_cnt; // item index
  assert(i < m_spec.m_cnt);
  const Type& type = m_spec.m_buf[i];
  const Uint32 lenBytes = type.m_arrayType;
  Uint32 bareLen = 0;
  switch (lenBytes) {
  case 0:
    bareLen = type.m_byteSize;
    break;
  case 1:
    bareLen = item[0];
    break;
  case 2:
    bareLen = item[0] + (item[1] << 8);
    break;
  default:
    assert(false);
    set_error(InternalError, __LINE__);
    return -1;
  }
  const Uint32 itemLen = lenBytes + bareLen;
  if (itemLen > type.m_byteSize)
  {
    set_error(DataValueOverflow, __LINE__);
    return -1;
  }
  m_itemPos += m_itemLen; // skip previous item
  m_cnt++;
  m_lenBytes = lenBytes;
  m_bareLen = bareLen;
  m_itemLen = itemLen;
  return 0;
}

int
NdbPack::Iter::desc_null()
{
  assert(m_cnt < m_spec.m_cnt);
  // caller checks if null allowed
  m_itemPos += m_itemLen; // skip previous item
  m_cnt++;
  m_nullCnt++;
  m_lenBytes = 0;
  m_bareLen = 0;
  m_itemLen = 0;
  return 0;
}

int
NdbPack::Iter::cmp(const Iter& r2, const Uint8* buf1, const Uint8* buf2) const
{
  const Iter& r1 = *this;
  assert(&r1.m_spec == &r2.m_spec);
  assert(r1.m_cnt == r2.m_cnt && r1.m_cnt > 0);
  const Uint32 i = r1.m_cnt - 1; // item index
  int res = 0;
  const Uint32 n1 = r1.m_itemLen;
  const Uint32 n2 = r2.m_itemLen;
  if (n1 != 0)
  {
    if (n2 != 0)
    {
      const Type& type = r1.m_spec.m_buf[i];
      const NdbSqlUtil::Type& sqlType = getSqlType(type.m_typeId);
      const Uint8* p1 = &buf1[r1.m_itemPos];
      const Uint8* p2 = &buf2[r2.m_itemPos];
      CHARSET_INFO* cs = all_charsets[type.m_csNumber];
      res = (*sqlType.m_cmp)(cs, p1, n1, p2, n2);
    }
    else
    {
      res = +1;
    }
  }
  else
  {
    if (n2 != 0)
      res = -1;
  }
  return res;
}

// NdbPack::DataC

int
NdbPack::DataC::desc(Iter& r) const
{
  const Uint32 i = r.m_cnt; // item index
  assert(i < m_cnt);
  const Type& type = m_spec.m_buf[i];
  if (type.m_nullable || m_allNullable)
  {
    Uint32 nullbitPos = 0;
    if (!m_allNullable)
      nullbitPos = type.m_nullbitPos;
    else
      nullbitPos = i;
    const Uint32 byte_pos = nullbitPos / 8;
    const Uint32 bit_pos = nullbitPos % 8;
    const Uint8 bit_mask = (1 << bit_pos);
    const Uint8& the_byte = m_buf[byte_pos];
    if ((the_byte & bit_mask) != 0)
    {
      if (r.desc_null() == -1)
      {
        set_error(r);
        return -1;
      }
      return 0;
    }
  }
  const Uint32 pos = r.m_itemPos + r.m_itemLen;
  const Uint8* item = &m_buf[pos];
  if (r.desc(item) == -1)
  {
    set_error(r);
    return -1;
  }
  return 0;
}

int
NdbPack::DataC::cmp(const DataC& d2, Uint32 cnt, Uint32& num_eq) const
{
  const DataC& d1 = *this;
  assert(cnt <= d1.m_cnt);
  assert(cnt <= d2.m_cnt);
  Iter r1(d1);
  Iter r2(d2);
  int res = 0;
  Uint32 i; // remember last
  for (i = 0; i < cnt; i++)
  {
    d1.desc(r1);
    d2.desc(r2);
    res = r1.cmp(r2, d1.m_buf, d2.m_buf);
    if (res != 0)
      break;
  }
  num_eq = i;
  return res;
}

// NdbPack::Data

int
NdbPack::Data::add(const void* data, Uint32* len_out)
{
  assert(data != 0);
  const Uint8* item = (const Uint8*)data;
  const Uint32 i = m_cnt; // item index
  if (i >= m_spec.m_cnt)
  {
    set_error(DataCntOverflow, __LINE__);
    return -1;
  }
  Iter& r = m_iter;
  assert(r.m_cnt == i);
  const Uint32 fullLen = m_varBytes + r.m_itemPos + r.m_itemLen;
  if (r.desc(item) == -1)
  {
    set_error(r);
    return -1;
  }
  if (fullLen + r.m_itemLen > m_bufMaxLen)
  {
    set_error(DataBufOverflow, __LINE__);
    return -1;
  }
  memcpy(&m_buf[fullLen], item, r.m_itemLen);
  *len_out = r.m_itemLen;
  m_cnt++;
  return 0;
}

int
NdbPack::Data::add(const void* data, Uint32 cnt, Uint32* len_out)
{
  const Uint8* data_ptr = (const Uint8*)data;
  Uint32 len_tot = 0;
  for (Uint32 i = 0; i < cnt; i++)
  {
    Uint32 len;
    if (add(data_ptr, &len) == -1)
      return -1;
    if (data != 0)
      data_ptr += len;
    len_tot += len;
  }
  *len_out = len_tot;
  return 0;
}

int
NdbPack::Data::add_null(Uint32* len_out)
{
  const Uint32 i = m_cnt; // item index
  if (i >= m_spec.m_cnt)
  {
    set_error(DataCntOverflow, __LINE__);
    return -1;
  }
  Iter& r = m_iter;
  assert(r.m_cnt == i);
  if (r.desc_null() == -1)
  {
    set_error(r);
    return -1;
  }
  Uint32 nullbitPos = 0;
  if (!m_allNullable)
  {
    const Type& type = m_spec.m_buf[i];
    if (!type.m_nullable)
    {
      set_error(DataNotNullable, __LINE__);
      return -1;
    }
    nullbitPos = type.m_nullbitPos;
  }
  else
  {
    nullbitPos = i;
  }
  const Uint32 byte_pos = nullbitPos / 8;
  const Uint32 bit_pos = nullbitPos % 8;
  const Uint8 bit_mask = (1 << bit_pos);
  Uint8& the_byte = m_buf[m_varBytes + byte_pos];
  assert((the_byte & bit_mask) == 0);
  the_byte |= bit_mask;
  *len_out = r.m_itemLen;
  m_cnt++;
  return 0;
}

int
NdbPack::Data::add_null(Uint32 cnt, Uint32* len_out)
{
  Uint32 len_tot = 0;
  for (Uint32 i = 0; i < cnt; i++)
  {
    Uint32 len;
    if (add_null(&len) == -1)
      return -1;
    len_tot += len;
  }
  *len_out = len_tot;
  return 0;
}

int
NdbPack::Data::add_poai(const Uint32* poai, Uint32* len_out)
{
  const AttributeHeader ah = *(const AttributeHeader*)&poai[0];
  if (!ah.isNULL())
  {
    if (add(&poai[1], len_out) == -1)
      return -1;
  }
  else
  {
    if (add_null(len_out) == -1)
      return -1;
  }
  if (ah.getByteSize() != *len_out)
  {
    set_error(InvalidAttrInfo, __LINE__);
    return -1;
  }
  return 0;
}

int
NdbPack::Data::add_poai(const Uint32* poai, Uint32 cnt, Uint32* len_out)
{
  Uint32 len_tot = 0;
  for (Uint32 i = 0; i < cnt; i++)
  {
    Uint32 len;
    if (add_poai(poai, &len) == -1)
      return -1;
    len_tot += len;
    poai += 1 + (len + 3) / 4;
  }
  *len_out = len_tot;
  return 0;
}

int
NdbPack::Data::finalize_impl()
{
  const Uint32 dataLen = m_iter.m_itemPos + m_iter.m_itemLen;
  switch (m_varBytes) {
  // case 0: inlined
  case 1:
    if (dataLen <= 0xFF)
    {
      m_buf[0] = dataLen;
      return 0;
    }
    break;
  case 2:
    if (dataLen <= 0xFFFF)
    {
      m_buf[0] = (dataLen & 0xFF);
      m_buf[1] = (dataLen >> 8);
      return 0;
    }
    break;
  default:
    break;
  }
  set_error(InternalError, __LINE__);
  return -1;
}

int
NdbPack::Data::desc_all(Uint32 cnt, Endian::Value from_endian)
{
  if (from_endian == NdbPack::Endian::Native)
    from_endian = NdbPack::Endian::get_endian();
  m_endian = from_endian;
  assert(m_cnt == 0); // reset() would destroy nullmask
  for (Uint32 i = 0; i < cnt; i++)
  {
    m_cnt++;
    if (desc(m_iter) == -1)
      return -1;
  }
  if (finalize() == -1)
    return -1;
  return 0;
}

int
NdbPack::Data::copy(const DataC& d2)
{
  reset();
  Iter r2(d2);
  const Uint32 cnt2 = d2.m_cnt;
  for (Uint32 i = 0; i < cnt2; i++)
  {
    if (d2.desc(r2) == -1)
      return -1;
    Uint32 len_out = ~(Uint32)0;
    if (r2.m_itemLen != 0)
    {
      if (add(&d2.m_buf[r2.m_itemPos], &len_out) == -1)
          return -1;
      assert(len_out == r2.m_itemLen);
    }
    else
    {
      if (add_null(&len_out) == -1)
        return -1;
      assert(len_out ==0);
    }
  }
  if (finalize() == -1)
    return -1;
  return 0;
}

int
NdbPack::Data::convert_impl(Endian::Value to_endian)
{
  const Spec& spec = m_spec;
  Iter r(*this);
  for (Uint32 i = 0; i < m_cnt; i++)
  {
    if (DataC::desc(r) == -1)
    {
      set_error(r);
      return -1;
    }
    const Type& type = spec.m_buf[i];
    const Uint32 typeId = type.m_typeId;
    const Ndb_pack_type_info& info = g_ndb_pack_type_info[typeId];
    if (info.m_convert)
    {
      Uint8* ptr = &m_buf[m_varBytes + r.m_itemPos];
      Uint32 len = r.m_itemLen;
      Endian::convert(ptr, len);
    }
  }
  return 0;
}

// NdbPack::BoundC

int
NdbPack::BoundC::finalize(int side)
{
  if (m_data.m_cnt == 0 && side != 0)
  {
    set_error(BoundEmptySide, __LINE__);
    return -1;
  }
  if (m_data.m_cnt != 0 && side != -1 && side != +1)
  {
    set_error(BoundNonemptySide, __LINE__);
    return -1;
  }
  m_side = side;
  return 0;
}

int
NdbPack::BoundC::cmp(const BoundC& b2, Uint32 cnt, Uint32& num_eq) const
{
  const BoundC& b1 = *this;
  const DataC& d1 = b1.m_data;
  const DataC& d2 = b2.m_data;
  int res = d1.cmp(d2, cnt, num_eq);
  if (res == 0)
  {
    if (cnt < d1.m_cnt && cnt < d2.m_cnt)
      ;
    else if (d1.m_cnt < d2.m_cnt)
      res = (+1) * b1.m_side;
    else if (d1.m_cnt > d2.m_cnt)
      res = (-1) * b2.m_side;
    else if (b1.m_side < b2.m_side)
      res = -1;
    else if (b1.m_side > b2.m_side)
      res = +1;
  }
  return res;
}

// NdbPack::Bound

// print

NdbPack::Print::Print(char* buf, Uint32 bufsz) :
  m_buf(buf), m_bufsz(bufsz), m_sz(0) {}

void
NdbPack::Print::print(const char* fmt, ...)
{
  va_list ap;
  va_start(ap, fmt);
  if (m_bufsz > m_sz)
  {
    BaseString::vsnprintf(&m_buf[m_sz], m_bufsz - m_sz, fmt, ap);
    m_sz += (Uint32)strlen(&m_buf[m_sz]);
  }
  va_end(ap);
}

// print Type

NdbOut&
operator<<(NdbOut& out, const NdbPack::Type& a)
{
  a.print(out);
  return out;
}

void
NdbPack::Type::print(NdbOut& out) const
{
  char buf[200];
  out << print(buf, sizeof(buf));
}

const char*
NdbPack::Type::print(char* buf, Uint32 bufsz) const
{
  Print p(buf, bufsz);
  p.print("typeId:%u", m_typeId);
  p.print(" byteSize:%u", m_byteSize);
  p.print(" nullable:%u", m_nullable);
  p.print(" csNumber:%u", m_csNumber);
  return buf;
}

// print Spec

NdbOut&
operator<<(NdbOut& out, const NdbPack::Spec& a)
{
  a.print(out);
  return out;
}

void
NdbPack::Spec::print(NdbOut& out) const
{
  char buf[8000];
  out << print(buf, sizeof(buf));
}

const char*
NdbPack::Spec::print(char* buf, Uint32 bufsz) const
{
  Print p(buf, bufsz);
  p.print("cnt:%u", m_cnt);
  p.print(" nullableCnt:%u", m_nullableCnt);
  p.print(" varsizeCnt:%u", m_varsizeCnt);
  p.print(" nullmaskLen:%u", get_nullmask_len(false));
  p.print(" maxByteSize:%u", m_maxByteSize);
  for (Uint32 i = 0; i < m_cnt; i++)
  {
    const Type& type = m_buf[i];
    p.print(" [%u", i);
    p.print(" typeId:%u", type.m_typeId);
    p.print(" nullable:%u", type.m_nullable);
    p.print(" byteSize:%u", type.m_byteSize);
    p.print(" csNumber:%u", type.m_csNumber);
    p.print("]");
  }
  return buf;
}

// print DataC

bool g_ndb_pack_print_hex_always = true;

NdbOut&
operator<<(NdbOut& out, const NdbPack::DataC& a)
{
  a.print(out);
  return out;
}

void
NdbPack::DataC::print(NdbOut& out) const
{
  char buf[8000];
  out << print(buf, sizeof(buf));
}

const char*
NdbPack::DataC::print(char* buf, Uint32 bufsz, bool convert_flag) const
{
  Print p(buf, bufsz);
  const Spec& spec = m_spec;
  const Uint32 nullmask_len = spec.get_nullmask_len(m_allNullable);
  if (nullmask_len != 0)
  {
    p.print("nullmask:");
    for (Uint32 i = 0; i < nullmask_len; i++)
    {
      int x = m_buf[i];
      p.print("%02x", x);
    }
  }
  Iter r(*this);
  for (Uint32 i = 0; i < m_cnt; i++)
  {
    desc(r);
    const Uint8* value = &m_buf[r.m_itemPos];
    p.print(" [%u", i);
    p.print(" pos:%u", r.m_itemPos);
    p.print(" len:%u", r.m_itemLen);
    if (r.m_itemLen > 0)
    {
      p.print(" value:");
      // some specific types for debugging
      const Type& type = spec.m_buf[i];
      bool ok = true;
      switch (type.m_typeId) {
      case NDB_TYPE_TINYINT:
        {
          Int8 x;
          memcpy(&x, value, 1);
          if (convert_flag)
            Endian::convert(&x, 1);
          p.print("%d", (int)x);
        }
        break;
      case NDB_TYPE_TINYUNSIGNED:
        {
          Uint8 x;
          memcpy(&x, value, 1);
          if (convert_flag)
            Endian::convert(&x, 1);
          p.print("%u", (uint)x);
        }
        break;
      case NDB_TYPE_SMALLINT:
        {
          Int16 x;
          memcpy(&x, value, 2);
          if (convert_flag)
            Endian::convert(&x, 2);
          p.print("%d", (int)x);
        }
        break;
      case NDB_TYPE_SMALLUNSIGNED:
        {
          Uint16 x;
          memcpy(&x, value, 2);
          if (convert_flag)
            Endian::convert(&x, 2);
          p.print("%u", (uint)x);
        }
        break;
      case NDB_TYPE_INT:
        {
          Int32 x;
          memcpy(&x, value, 4);
          if (convert_flag)
            Endian::convert(&x, 4);
          p.print("%d", (int)x);
        }
        break;
      case NDB_TYPE_UNSIGNED:
        {
          Uint32 x;
          memcpy(&x, value, 4);
          if (convert_flag)
            Endian::convert(&x, 4);
          p.print("%u", (uint)x);
        }
        break;
      case NDB_TYPE_FLOAT:
        {
          float x;
          memcpy(&x, value, 4);
          if (convert_flag)
            Endian::convert(&x, 4);
          p.print("%g", (double)x);
        }
        break;
      case NDB_TYPE_DOUBLE:
        {
          double x;
          memcpy(&x, value, 8);
          if (convert_flag)
            Endian::convert(&x, 8);
          p.print("%g", x);
        }
        break;
      case NDB_TYPE_CHAR:
      case NDB_TYPE_VARCHAR:
      case NDB_TYPE_LONGVARCHAR:
        {
          const Uint32 off = type.m_arrayType;
          for (Uint32 j = 0; j < r.m_bareLen; j++)
          {
            Uint8 x = value[off + j];
            p.print("%c", (int)x);
          }
        }
        break;
      default:
        ok = false;
        break;
      }
      if (!ok || g_ndb_pack_print_hex_always)
      {
        p.print("<");
        for (Uint32 j = 0; j < r.m_itemLen; j++)
        {
          int x = value[j];
          p.print("%02x", x);
        }
        p.print(">");
      }
    }
    p.print("]");
  }
  return buf;
}

// print Data

NdbOut&
operator<<(NdbOut& out, const NdbPack::Data& a)
{
  a.print(out);
  return out;
}

void
NdbPack::Data::print(NdbOut& out) const
{
  char buf[8000];
  out << print(buf, sizeof(buf));
}

const char*
NdbPack::Data::print(char* buf, Uint32 bufsz) const
{
  Print p(buf, bufsz);
  char* ptr = buf;
  if (m_varBytes != 0)
  {
    p.print("varBytes:");
    for (Uint32 i = 0; i < m_varBytes; i++)
    {
      int r = m_buf[i];
      p.print("%02x", r);
    }
    p.print(" ");
  }
  p.print("dataLen:%u", m_iter.m_itemPos + m_iter.m_itemLen);
  p.print(" ");
  const bool convert_flag =
    m_endian != Endian::Native &&
    m_endian != Endian::get_endian();
  DataC::print(&buf[p.m_sz], bufsz - p.m_sz, convert_flag);
  return buf;
}

// print BoundC

NdbOut&
operator<<(NdbOut& out, const NdbPack::BoundC& a)
{
  a.print(out);
  return out;
}

void
NdbPack::BoundC::print(NdbOut& out) const
{
  char buf[8000];
  out << print(buf, sizeof(buf));
}

const char*
NdbPack::BoundC::print(char* buf, Uint32 bufsz) const
{
  Print p(buf, bufsz);
  p.print("side:%s ", m_side < 0 ? "-" : m_side > 0 ? "+" : "0");
  m_data.print(&buf[p.m_sz], bufsz - p.m_sz);
  return buf;
}

// print Bound

NdbOut&
operator<<(NdbOut& out, const NdbPack::Bound& a)
{
  a.print(out);
  return out;
}

void
NdbPack::Bound::print(NdbOut& out) const
{
  char buf[8000];
  out << print(buf, sizeof(buf));
}

const char*
NdbPack::Bound::print(char* buf, Uint32 bufsz) const
{
  BoundC::print(buf, bufsz);
  return buf;
}

// validate

int
NdbPack::Type::validate() const
{
  Type type2 = *this;
  if (type2.complete() == -1)
  {
    set_error(type2);
    return -1;
  }
  if (memcmp(this, &type2, sizeof(Type)) != 0)
  {
    set_error(ValidationError, __LINE__);
    return -1;
  }
  return 0;
}

int
NdbPack::Spec::validate() const
{
  Uint32 nullableCnt = 0;
  Uint32 varsizeCnt = 0;
  for (Uint32 i = 0; i < m_cnt; i++)
  {
    const Type& type = m_buf[i];
    if (type.validate() == -1)
    {
      set_error(type);
      return -1;
    }
    if (type.m_nullable)
      nullableCnt++;
    if (type.m_arrayType != 0)
      varsizeCnt++;
  }
  if (m_nullableCnt != nullableCnt)
  {
    set_error(ValidationError, __LINE__);
    return -1;
  }
  if (m_varsizeCnt != varsizeCnt)
  {
    set_error(ValidationError, __LINE__);
    return -1;
  }
  return 0;
}

int
NdbPack::Data::validate() const
{
  if (DataC::validate() == -1)
    return -1;
  const Iter& r = m_iter;
  if (r.m_cnt != m_cnt)
  {
    set_error(ValidationError, __LINE__);
    return -1;
  }
  Iter r2(*this);
  for (Uint32 i = 0; i < m_cnt; i++)
  {
    if (desc(r2) == -1)
      return -1;
  }
  if (r.m_itemPos != r2.m_itemPos)
  {
    set_error(ValidationError, __LINE__);
    return -1;
  }
  if (r.m_cnt != r2.m_cnt)
  {
    set_error(ValidationError, __LINE__);
    return -1;
  }
  if (r.m_nullCnt != r2.m_nullCnt)
  {
    set_error(ValidationError, __LINE__);
    return -1;
  }
  if (r.m_itemLen != r2.m_itemLen)
  {
    set_error(ValidationError, __LINE__);
    return -1;
  }
  return 0;
}

int
NdbPack::BoundC::validate() const
{
  if (m_data.validate() == -1)
  {
    set_error(m_data);
    return -1;
  }
  if (m_data.m_cnt == 0 && m_side != 0)
  {
    set_error(ValidationError, __LINE__);
    return -1;
  }
  if (m_data.m_cnt != 0 && m_side != -1 && m_side != +1)
  {
    set_error(ValidationError, __LINE__);
    return -1;
  }
  return 0;
}

int
NdbPack::Bound::validate() const
{
  if (BoundC::validate() == -1)
    return -1;
  if (m_data.validate() == -1)
  {
    set_error(m_data);
    return -1;
  }
  return 0;
}

#ifdef TEST_NDB_PACK
#include <util/NdbTap.hpp>

#define chk1(x) do { if (x) break; ndbout << "line " << __LINE__ << ": " << #x << endl; require(false); } while (0)

#define chk2(x, e) do { if (x) break; ndbout << "line " << __LINE__ << ": " << #x << endl; ndbout << "NdbPack code: " << (e).get_error_code() << " line: " << (e).get_error_line() << endl; require(false); } while (0)

#define ll0(x) do { if (verbose < 0) break; ndbout << "0- " << x << endl; } while (0)
#define ll1(x) do { if (verbose < 1) break; ndbout << "1- " << x << endl; } while (0)
#define ll2(x) do { if (verbose < 2) break; ndbout << "2- " << x << endl; } while (0)
#define ll3(x) do { if (verbose < 3) break; ndbout << "3- " << x << endl; } while (0)

#define xmin(a, b) ((a) < (b) ? (a) : (b))

#include <ndb_rand.h>

static uint // random 0..n-1
getrandom(uint n)
{
  if (n != 0) {
    uint k = ndb_rand();
    return k % n;
  }
  return 0;
}

static uint // random 0..n-1 biased exponentially to smaller
getrandom(uint n, uint bias)
{
  assert(bias != 0);
  uint k = getrandom(n);
  bias--;
  while (bias != 0) {
    k = getrandom(k + 1);
    bias--;
  }
  return k;
}

static bool
getrandompct(uint pct)
{
  return getrandom(100) < pct;
}

// change in TAPTEST
static int seed = -1; // random
static int loops = 0;
static int spec_cnt = -1; // random
static int fix_type = 0; // all types
static int no_nullable = 0;
static int data_cnt = -1; // Max
static int bound_cnt = -1; // Max
static int verbose = 0;

struct Tspec {
  enum { Max = 100 };
  enum { MaxBuf = Max * 4000 };
  NdbPack::Spec m_spec;
  NdbPack::Type m_type[Max];
  Tspec() {
    m_spec.set_buf(m_type, Max);
  }
  void create();
};

static NdbOut&
operator<<(NdbOut& out, const Tspec& tspec)
{
  out << tspec.m_spec;
  return out;
}

void
Tspec::create()
{
  m_spec.reset();
  int cnt = spec_cnt == -1 ? 1 + getrandom(Tspec::Max, 3) : spec_cnt;
  int i = 0;
  while (i < cnt) {
    int typeId = fix_type;
    if (typeId == 0)
      typeId = getrandom(g_ndb_pack_type_info_cnt);
    const Ndb_pack_type_info& info = g_ndb_pack_type_info[typeId];
    switch (typeId) {
    case NDB_TYPE_INT:
    case NDB_TYPE_UNSIGNED:
    case NDB_TYPE_CHAR:
    case NDB_TYPE_VARCHAR:
    case NDB_TYPE_LONGVARCHAR:
      break;
    default:
      continue;
    }
    require(info.m_supported);
    int byteSize = 0;
    if (info.m_fixSize != 0)
      byteSize = info.m_fixSize;
    else if (info.m_arrayType == 0)
      byteSize = 1 + getrandom(128, 1);  // char(1-128)
    else if (info.m_arrayType == 1)
      byteSize = 1 + getrandom(256, 2);  // varchar(0-255)
    else if (info.m_arrayType == 2)
      byteSize = 2 + getrandom(1024, 3); // longvarchar(0-1023)
    else
      require(false);
    bool nullable = no_nullable ? false : getrandompct(50);
    int csNumber = 0;
    if (info.m_charType) {
      csNumber = 8; // should include ascii
    }
    NdbPack::Type type(typeId, byteSize, nullable, csNumber);
    chk2(m_spec.add(type) == 0, m_spec);
    i++;
  }
  chk2(m_spec.validate() == 0, m_spec);
}

struct Tdata {
  const Tspec& m_tspec;
  NdbPack::Data m_data;
  const bool m_isBound;
  int m_cnt;
  Uint8* m_xbuf;        // unpacked
  int m_xsize;
  int m_xoff[Tspec::Max];
  int m_xlen[Tspec::Max];
  bool m_xnull[Tspec::Max];
  int m_xnulls;
  Uint32* m_poaiBuf;    // plain old attr info
  int m_poaiSize;
  Uint8* m_packBuf;     // packed
  int m_packLen;
  Tdata(Tspec& tspec, bool isBound, uint varBytes) :
    m_tspec(tspec),
    m_data(tspec.m_spec, isBound, varBytes),
    m_isBound(isBound)
  {
    m_cnt = tspec.m_spec.get_cnt();
    m_xbuf = 0;
    m_poaiBuf = 0;
    m_packBuf = 0;
  }
  ~Tdata() {
    delete [] m_xbuf;
    delete [] m_poaiBuf;
    delete [] m_packBuf;
  }
  void create();
  void add();
  void finalize();
  // compare using unpacked data
  int xcmp(const Tdata& tdata2, int* num_eq) const;
};

static NdbOut&
operator<<(NdbOut& out, const Tdata& tdata)
{
  out << tdata.m_data;
  return out;
}

void
Tdata::create()
{
  union {
    Uint8 xbuf[Tspec::MaxBuf];
    Uint64 xbuf_align;
  };
  memset(xbuf, 0x3f, sizeof(xbuf));
  m_xsize = 0;
  m_xnulls = 0;
  Uint32 poaiBuf[Tspec::MaxBuf / 4];
  memset(poaiBuf, 0x5f, sizeof(poaiBuf));
  m_poaiSize = 0;
  m_packLen = m_data.get_var_bytes();
  m_packLen += (m_tspec.m_spec.get_nullable_cnt(m_isBound) + 7) / 8;
  int i = 0, j;
  while (i < m_cnt) {
    const NdbPack::Type& type = m_tspec.m_spec.get_type(i);
    const int typeId = type.get_type_id();
    const Ndb_pack_type_info& info = g_ndb_pack_type_info[typeId];
    m_xnull[i] = type.get_nullable() && getrandompct(25);
    m_xnull[i] = false;
    if (type.get_nullable() || m_isBound)
      m_xnull[i] = getrandompct(20);
    int pad = 0; // null-char pad not counted in xlen
    if (!m_xnull[i]) {
      m_xoff[i] = m_xsize;
      Uint8* xptr = &xbuf[m_xsize];
      switch (typeId) {
      case NDB_TYPE_INT:
        {
          Int32 x = getrandom(10);
          if (getrandompct(50))
            x = (-1) * x;
          memcpy(xptr, &x, 4);
          m_xlen[i] = info.m_fixSize;
        }
        break;
      case NDB_TYPE_UNSIGNED:
        {
          Uint32 x = getrandom(10);
          memcpy(xptr, &x, 4);
          m_xlen[i] = info.m_fixSize;
        }
        break;
      case NDB_TYPE_CHAR:
        {
          require(type.get_byte_size() >= 1);
          int max_len = type.get_byte_size();
          int len = getrandom(max_len + 1, 1);
          for (j = 0; j < len; j++)
          {
            xptr[j] = 'a' + getrandom(3);
          }
          for (j = len; j < max_len; j++)
          {
            xptr[j] = 0x20;
          }
          m_xlen[i] = max_len;
          xptr[max_len] = 0;
          pad = 1;
        }
        break;
      case NDB_TYPE_VARCHAR:
        {
          require(type.get_byte_size() >= 1);
          int max_len = type.get_byte_size() - 1;
          int len = getrandom(max_len, 2);
          require(len < 256);
          xptr[0] = len;
          for (j = 0; j < len; j++)
          {
            xptr[1 + j] = 'a' + getrandom(3);
          }
          m_xlen[i] = 1 + len;
          xptr[1 + len] = 0;
          pad = 1;
        }
        break;
      case NDB_TYPE_LONGVARCHAR:
        {
          require(type.get_byte_size() >= 2);
          int max_len = type.get_byte_size() - 2;
          int len = getrandom(max_len, 3);
          require(len < 256 * 256);
          xptr[0] = (len & 0xFF);
          xptr[1] = (len >> 8);
          for (j = 0; j < len; j++)
          {
            xptr[2 + j] = 'a' + getrandom(3);
          }
          m_xlen[i] = 2 + len;
          xptr[2 + len] = 0;
          pad = 1;
        }
        break;
      default:
        require(false);
        break;
      }
      m_xsize += m_xlen[i] + pad;
      while (m_xsize % 8 != 0)
        m_xsize++;
      m_packLen += m_xlen[i];
    } else {
      m_xoff[i] = -1;
      m_xlen[i] = 0;
      m_xnulls++;
    }
    require(m_xnull[i] == (m_xoff[i] == -1));
    require(m_xnull[i] == (m_xlen[i] == 0));
    AttributeHeader* ah = (AttributeHeader*)&poaiBuf[m_poaiSize];
    ah->setAttributeId(i); // not used
    ah->setByteSize(m_xlen[i]);
    m_poaiSize++;
    if (!m_xnull[i]) {
      memcpy(&poaiBuf[m_poaiSize], &xbuf[m_xoff[i]], m_xlen[i]);
      m_poaiSize += (m_xlen[i] + 3) / 4;
    }
    i++;
  }
  require(m_xsize % 8 == 0);
  m_xbuf = (Uint8*) new Uint64 [m_xsize / 8];
  memcpy(m_xbuf, xbuf, m_xsize);
  m_poaiBuf = (Uint32*) new Uint32 [m_poaiSize];
  memcpy(m_poaiBuf, poaiBuf, m_poaiSize << 2);
}

void
Tdata::add()
{
  m_packBuf = new Uint8 [m_packLen];
  m_data.set_buf(m_packBuf, m_packLen);
  int i, j;
  j = 0;
  while (j <= 1) {
    if (j == 1)
      m_data.reset();
    i = 0;
    while (i < m_cnt) {
      Uint32 xlen = ~(Uint32)0;
      if (!m_xnull[i]) {
        int xoff = m_xoff[i];
        const Uint8* xptr = &m_xbuf[xoff];
        chk2(m_data.add(xptr, &xlen) == 0, m_data);
        chk1((int)xlen == m_xlen[i]);
      } else {
        chk2(m_data.add_null(&xlen) == 0, m_data);
        chk1(xlen == 0);
      }
      i++;
    }
    chk2(m_data.validate() == 0, m_data);
    chk1((int)m_data.get_null_cnt() == m_xnulls);
    j++;
  }
}

void
Tdata::finalize()
{
  chk2(m_data.finalize() == 0, m_data);
  ll3("create: " << m_data);
  chk1((int)m_data.get_full_len() == m_packLen);
  {
    const Uint8* p = (const Uint8*)m_data.get_full_buf();
    chk1(p[0] + (p[1] << 8) == m_packLen - 2);
  }
}

int
Tdata::xcmp(const Tdata& tdata2, int* num_eq) const
{
  const Tdata& tdata1 = *this;
  require(&tdata1.m_tspec == &tdata2.m_tspec);
  const Tspec& tspec = tdata1.m_tspec;
  int res = 0;
  int cnt = xmin(tdata1.m_cnt, tdata2.m_cnt);
  int i;
  for (i = 0; i < cnt; i++) {
    if (!tdata1.m_xnull[i]) {
      if (!tdata2.m_xnull[i]) {
        // the pointers are Uint64-aligned
        const Uint8* xptr1 = &tdata1.m_xbuf[tdata1.m_xoff[i]];
        const Uint8* xptr2 = &tdata2.m_xbuf[tdata2.m_xoff[i]];
        const int xlen1 = tdata1.m_xlen[i];
        const int xlen2 = tdata2.m_xlen[i];
        const NdbPack::Type& type = tspec.m_spec.get_type(i);
        const int typeId = type.get_type_id();
        const int csNumber = type.get_cs_number();
        CHARSET_INFO* cs = all_charsets[csNumber];
        switch (typeId) {
        case NDB_TYPE_INT:
          {
            require(cs == 0);
            Int32 x1 = *(const Int32*)xptr1;
            Int32 x2 = *(const Int32*)xptr2;
            if (x1 < x2)
              res = -1;
            else if (x1 > x2)
              res = +1;
            ll3("cmp res:" << res <<" x1:" << x1 << " x2:" << x2);
          }
          break;
        case NDB_TYPE_UNSIGNED:
          {
            require(cs == 0);
            Uint32 x1 = *(const Uint32*)xptr1;
            Uint32 x2 = *(const Uint32*)xptr2;
            if (x1 < x2)
              res = -1;
            else if (x1 > x2)
              res = +1;
            ll3("cmp res:" << res <<" x1:" << x1 << " x2:" << x2);
          }
          break;
        case NDB_TYPE_CHAR:
          {
            require(cs != 0 && cs->coll != 0);
            const uint n1 = xlen1;
            const uint n2 = xlen2;
            const uchar* t1 = &xptr1[0];
            const uchar* t2 = &xptr2[0];
            const char* s1 = (const char*)t1;
            const char* s2 = (const char*)t2;
            chk1(n1 == strlen(s1));
            chk1(n2 == strlen(s2));
            res = (*cs->coll->strnncollsp)(cs, t1, n1, t2, n2, false);
            ll3("cmp res:" << res <<" s1:" << s1 << " s2:" << s2);
          }
          break;
        case NDB_TYPE_VARCHAR:
          {
            require(cs != 0 && cs->coll != 0);
            const uint n1 = xptr1[0];
            const uint n2 = xptr2[0];
            const uchar* t1 = &xptr1[1];
            const uchar* t2 = &xptr2[1];
            const char* s1 = (const char*)t1;
            const char* s2 = (const char*)t2;
            chk1(n1 == strlen(s1));
            chk1(n2 == strlen(s2));
            res = (*cs->coll->strnncollsp)(cs, t1, n1, t2, n2, false);
            ll3("cmp res:" << res <<" s1:" << s1 << " s2:" << s2);
          }
          break;
        case NDB_TYPE_LONGVARCHAR:
          {
            require(cs != 0 && cs->coll != 0);
            const uint n1 = xptr1[0] | (xptr1[1] << 8);
            const uint n2 = xptr2[0] | (xptr2[1] << 8);
            const uchar* t1 = &xptr1[2];
            const uchar* t2 = &xptr2[2];
            const char* s1 = (const char*)t1;
            const char* s2 = (const char*)t2;
            chk1(n1 == strlen(s1));
            chk1(n2 == strlen(s2));
            res = (*cs->coll->strnncollsp)(cs, t1, n1, t2, n2, false);
            ll3("cmp res:" << res <<" s1:" << s1 << " s2:" << s2);
          }
          break;
        default:
          require(false);
          break;
        }
      } else
        res = +1;
    } else if (!tdata2.m_xnull[i])
      res = -1;
    if (res != 0)
      break;
  }
  *num_eq = i;
  ll3("xcmp res:" << res << " num_eq:" << *num_eq);
  return res;
}

struct Tbound {
  Tdata& m_tdata;
  NdbPack::Bound m_bound;
  Tbound(Tdata& tdata) :
    m_tdata(tdata),
    m_bound(tdata.m_data)
  {
    m_tdata.m_cnt = 1 + getrandom(m_tdata.m_cnt);
  }
  void create();
  void add();
  void finalize();
  int xcmp(const Tdata& tdata2, int* num_eq) const;
  int xcmp(const Tbound& tbound2, int* num_eq) const;
};

static NdbOut&
operator<<(NdbOut& out, const Tbound& tbound)
{
  out << tbound.m_bound;
  return out;
}

void
Tbound::create()
{
  m_tdata.create();
}

void
Tbound::add()
{
  m_tdata.add();
}

void
Tbound::finalize()
{
  int side = getrandompct(50) ? -1 : +1;
  chk2(m_bound.finalize(side) == 0, m_bound);
  chk2(m_bound.validate() == 0, m_bound);
  chk1((int)m_tdata.m_data.get_full_len() == m_tdata.m_packLen);
}

int
Tbound::xcmp(const Tdata& tdata2, int* num_eq) const
{
  const Tbound& tbound1 = *this;
  const Tdata& tdata1 = tbound1.m_tdata;
  require(tdata1.m_cnt <= tdata2.m_cnt);
  *num_eq = -1;
  int res = tdata1.xcmp(tdata2, num_eq);
  if (res == 0) {
    chk1(*num_eq == tdata1.m_cnt);
    res = m_bound.get_side();
  }
  return res;
}

int
Tbound::xcmp(const Tbound& tbound2, int* num_eq) const
{
  const Tbound& tbound1 = *this;
  const Tdata& tdata1 = tbound1.m_tdata;
  const Tdata& tdata2 = tbound2.m_tdata;
  *num_eq = -1;
  int res = tdata1.xcmp(tdata2, num_eq);
  chk1(0 <= *num_eq && *num_eq <= xmin(tdata1.m_cnt, tdata2.m_cnt));
  if (res == 0) {
    chk1(*num_eq == xmin(tdata1.m_cnt, tdata2.m_cnt));
    if (tdata1.m_cnt < tdata2.m_cnt)
      res = (+1) * tbound1.m_bound.get_side();
    else if (tdata1.m_cnt > tdata2.m_cnt)
      res = (-1) * tbound2.m_bound.get_side();
    else if (tbound1.m_bound.get_side() < tbound2.m_bound.get_side())
      res = -1;
    else if (tbound1.m_bound.get_side() > tbound2.m_bound.get_side())
      res = +1;
  }
  return res;
}

struct Tdatalist {
  enum { Max = 1000 };
  Tdata* m_tdata[Max];
  int m_cnt;
  Tdatalist(Tspec& tspec) {
    m_cnt = data_cnt == -1 ? Max : data_cnt;
    int i;
    for (i = 0; i < m_cnt; i++) {
      m_tdata[i] = new Tdata(tspec, false, 2);
    }
  }
  ~Tdatalist() {
    int i;
    for (i = 0; i < m_cnt; i++) {
      delete m_tdata[i];
    }
  }
  void create();
  void sort();
};

static NdbOut&
operator<<(NdbOut& out, const Tdatalist& tdatalist)
{
  int i;
  for (i = 0; i < tdatalist.m_cnt; i++) {
    out << "data " << i << ": " << *tdatalist.m_tdata[i];
    if (i + 1 < tdatalist.m_cnt)
      out << endl;
  }
  return out;
}

void
Tdatalist::create()
{
  int i;
  for (i = 0; i < m_cnt; i++) {
    Tdata& tdata = *m_tdata[i];
    tdata.create();
    tdata.add();
    tdata.finalize();
  }
}

static int
data_cmp(const void* a1, const void* a2)
{
  const Tdata& tdata1 = **(const Tdata**)a1;
  const Tdata& tdata2 = **(const Tdata**)a2;
  require(tdata1.m_cnt == tdata2.m_cnt);
  const Uint32 cnt = tdata1.m_cnt;
  Uint32 num_eq = ~(Uint32)0;
  int res = tdata1.m_data.cmp(tdata2.m_data, cnt, num_eq);
  require(num_eq <= (Uint32)tdata1.m_cnt);
  require(num_eq <= (Uint32)tdata2.m_cnt);
  return res;
}

void
Tdatalist::sort()
{
  ll1("data sort: in");
  ll3(endl << *this);
  qsort(m_tdata, m_cnt, sizeof(Tdata*), data_cmp);
  ll1("data sort: out");
  ll3(endl << *this);
  int i;
  for (i = 0; i + 1 < m_cnt; i++) {
    const Tdata& tdata1 = *m_tdata[i];
    const Tdata& tdata2 = *m_tdata[i + 1];
    require(tdata1.m_cnt == tdata2.m_cnt);
    const Uint32 cnt = tdata1.m_cnt;
    Uint32 num_eq1 = ~(Uint32)0;
    int res = tdata1.m_data.cmp(tdata2.m_data, cnt, num_eq1);
    chk1(res <= 0);
    // also via unpacked data
    int num_eq2 = -1;
    int res2 = tdata1.xcmp(tdata2, &num_eq2);
    if (res < 0)
      chk1(res2 < 0);
    else if (res == 0)
      chk1(res2 == 0);
    else
      chk1(res2 > 0);
    chk1(num_eq1 == (Uint32)num_eq2);
  }
}

struct Tboundlist {
  enum { Max = 1000 };
  Tbound* m_tbound[Max];
  int m_cnt;
  Tboundlist(Tspec& tspec) {
    m_cnt = bound_cnt == -1 ? Max : bound_cnt;
    int i;
    for (i = 0; i < m_cnt; i++) {
      Tdata* tdata = new Tdata(tspec, true, 0);
      m_tbound[i] = new Tbound(*tdata);
    }
  }
  ~Tboundlist() {
    int i;
    for (i = 0; i < m_cnt; i++) {
      Tdata* tdata = &m_tbound[i]->m_tdata;
      delete m_tbound[i];
      delete tdata;
    }
  }
  void create();
  void sort();
};

static NdbOut&
operator<<(NdbOut& out, const Tboundlist& tboundlist)
{
  int i;
  for (i = 0; i < tboundlist.m_cnt; i++) {
    out << "bound " << i << ": " << *tboundlist.m_tbound[i];
    if (i + 1 < tboundlist.m_cnt)
      out << endl;
  }
  return out;
}

void
Tboundlist::create()
{
  int i;
  for (i = 0; i < m_cnt; i++) {
    Tbound& tbound = *m_tbound[i];
    tbound.create();
    tbound.add();
    tbound.finalize();
  }
}

static int
bound_cmp(const void* a1, const void* a2)
{
  const Tbound& tbound1 = **(const Tbound**)a1;
  const Tbound& tbound2 = **(const Tbound**)a2;
  const Uint32 cnt = xmin(tbound1.m_tdata.m_cnt, tbound2.m_tdata.m_cnt);
  Uint32 num_eq = ~(Uint32)0;
  int res = tbound1.m_bound.cmp(tbound2.m_bound, cnt, num_eq);
  require(num_eq <= cnt);
  require(num_eq <= cnt);
  return res;
}

void
Tboundlist::sort()
{
  ll1("bound sort: in");
  ll3(endl << *this);
  qsort(m_tbound, m_cnt, sizeof(Tbound*), bound_cmp);
  ll1("bound sort: out");
  ll3(endl << *this);
  int i;
  for (i = 0; i + 1 < m_cnt; i++) {
    const Tbound& tbound1 = *m_tbound[i];
    const Tbound& tbound2 = *m_tbound[i + 1];
    const Uint32 cnt = xmin(tbound1.m_tdata.m_cnt, tbound2.m_tdata.m_cnt);
    Uint32 num_eq1 = ~(Uint32)0;
    int res = tbound1.m_bound.cmp(tbound2.m_bound, cnt, num_eq1);
    chk1(res <= 0);
    // also via unpacked data
    int num_eq2 = -1;
    int res2 = tbound1.xcmp(tbound2, &num_eq2);
    if (res < 0)
      chk1(res2 < 0);
    else if (res == 0)
      chk1(res2 == 0);
    else
      chk1(res2 > 0);
    chk1(num_eq1 == (Uint32)num_eq2);
  }
}

static void
testdesc(const Tdata& tdata)
{
  ll3("testdesc: " << tdata);
  const Tspec& tspec = tdata.m_tspec;
  const NdbPack::Data& data = tdata.m_data;
  const Uint8* buf_old = (const Uint8*)data.get_full_buf();
  const Uint32 varBytes = data.get_var_bytes();
  const Uint32 nullMaskLen = tspec.m_spec.get_nullmask_len(false);
  const Uint32 dataLen = data.get_data_len();
  const Uint32 fullLen = data.get_full_len();
  const Uint32 cnt = data.get_cnt();
  chk1(fullLen == varBytes + dataLen);
  NdbPack::Data data_new(tspec.m_spec, false, varBytes);
  Uint8 buf_new[Tspec::MaxBuf];
  data_new.set_buf(buf_new, sizeof(buf_new));
  memcpy(buf_new, buf_old, fullLen);
  chk2(data_new.desc_all(cnt, NdbPack::Endian::Native) == 0, data_new);
  chk1(memcmp(buf_new, data.get_full_buf(), data.get_full_len()) == 0);
  chk1(data_new.get_data_len() == data.get_data_len());
  chk1(data_new.get_cnt() == data.get_cnt());
  chk1(data_new.get_null_cnt() == data.get_null_cnt());
}

static void
testcopy(const Tdata& tdata)
{
  ll3("testcopy: " << tdata);
  const Tspec& tspec = tdata.m_tspec;
  const NdbPack::Data& data = tdata.m_data;
  uint n = getrandom(tdata.m_cnt + 1);
  do {
    ll3("testcopy: cnt:" << tdata.m_cnt << " n:" << n);
    NdbPack::DataC data_old(tspec.m_spec, false);
    data_old.set_buf(data.get_data_buf(), data.get_data_len(), n);
    chk1(data_old.get_cnt() == n);
    NdbPack::Data data_new(tspec.m_spec, false, 0);
    Uint8 buf_new[Tspec::MaxBuf];
    data_new.set_buf(buf_new, sizeof(buf_new));
    chk2(data_new.copy(data_old) == 0, data_new);
    chk1(data_new.get_cnt() == n);
    Uint32 num_eq1 = ~(Uint32)0;
    chk1(data_new.cmp(data_old, n, num_eq1) == 0);
    chk1(num_eq1 == n);
    Uint32 num_eq2 = ~(Uint32)0;
    chk1(data_old.cmp(data_new, n, num_eq2) == 0);
    chk1(num_eq2 == n);
    n = getrandom(n);
  } while (n != 0);
}

static void
testpoai(const Tdata& tdata)
{
  ll3("testpoai: " << tdata);
  const Tspec& tspec = tdata.m_tspec;
  const NdbPack::Data& data = tdata.m_data;
  NdbPack::Data data_new(tspec.m_spec, false, data.get_var_bytes());
  Uint8 buf_new[Tspec::MaxBuf];
  data_new.set_buf(buf_new, sizeof(buf_new));
  Uint32 poaiLen = ~(Uint32)0;
  chk2(data_new.add_poai(tdata.m_poaiBuf, tdata.m_cnt, &poaiLen) == 0, data);
  chk2(data_new.finalize() == 0, data_new);
  chk2(data_new.validate() == 0, data_new);
  chk1(tspec.m_spec.get_nullmask_len(false) + poaiLen == data.get_data_len());
  chk1(data_new.get_full_len() == data.get_full_len());
  chk1(memcmp(data_new.get_full_buf(), data.get_full_buf(), data.get_full_len()) == 0);
  chk1(data_new.get_null_cnt() == data.get_null_cnt());
}

static void
testconvert(const Tdata& tdata)
{
  ll3("testconvert: " << tdata);
  const Tspec& tspec = tdata.m_tspec;
  const NdbPack::Data& data = tdata.m_data;
  NdbPack::Data data_new(tspec.m_spec, false, 2);
  Uint8 buf_new[Tspec::MaxBuf];
  data_new.set_buf(buf_new, sizeof(buf_new));
  chk2(data_new.copy(data) == 0, data_new);
  require(tdata.m_cnt == (int)data.get_cnt());
  require(data.get_cnt() == data_new.get_cnt());
  const Uint32 cnt = tdata.m_cnt;
  Uint32 num_eq;
  int i;
  for (i = 0; i < 10; i++) {
    int k = getrandom(3); // assumes Endian::Value 0,1,2
    NdbPack::Endian::Value v = (NdbPack::Endian::Value)k;
    chk2(data_new.convert(v) == 0, data_new);
    if (v == NdbPack::Endian::Native ||
        v == NdbPack::Endian::get_endian()) {
      num_eq = ~(Uint32)0;
      chk1(data.cmp(data_new, cnt, num_eq) == 0);
      require(num_eq == cnt);
    }
  }
}

static void
testdata(const Tdatalist& tdatalist)
{
  int i;
  for (i = 0; i < tdatalist.m_cnt; i++) {
    const Tdata& tdata = *tdatalist.m_tdata[i];
    testdesc(tdata);
    testcopy(tdata);
    testpoai(tdata);
    testconvert(tdata);
  }
}

static void
testcmp(const Tbound& tbound, const Tdatalist& tdatalist, int* kb)
{
  ll3("testcmp: " << tbound);
  int oldres = 0;
  int n1 = 0;
  int n2 = 0;
  int i;
  for (i = 0; i < tdatalist.m_cnt; i++) {
    const Tdata& tdata = *tdatalist.m_tdata[i];
    require(tbound.m_tdata.m_cnt == (int)tbound.m_bound.get_data().get_cnt());
    const Uint32 cnt = tbound.m_tdata.m_cnt;
    Uint32 num_eq1 = ~(Uint32)0;
    // reverse result for key vs bound
    int res = (-1) * tbound.m_bound.cmp(tdata.m_data, cnt, num_eq1);
    chk1(res != 0);
    res = (res < 0 ? (n1++, -1) : (n2++, +1));
    if (i > 0) {
      // at some point flips from -1 to +1
      chk1(oldres <= res);
    }
    oldres = res;
    // also via unpacked data
    int num_eq2 = -1;
    int res2 = (-1) * tbound.xcmp(tdata, &num_eq2);
    if (res < 0)
      chk1(res2 < 0);
    else
      chk1(res2 > 0);
    chk1(num_eq1 == (Uint32)num_eq2);
  }
  require(n1 + n2 == tdatalist.m_cnt);
  ll2("keys before:" << n1 << " after:" << n2);
  *kb = n1;
}

static void
testcmp(const Tboundlist& tboundlist, const Tdatalist& tdatalist)
{
  int i;
  int oldkb = 0;
  for (i = 0; i < tboundlist.m_cnt; i++) {
    const Tbound& tbound = *tboundlist.m_tbound[i];
    int kb = 0;
    testcmp(tbound, tdatalist, &kb);
    if (i > 0) {
      chk1(oldkb <= kb);
    }
    oldkb = kb;
  }
}

static void
testrun()
{
  Tspec tspec;
  tspec.create();
  ll1("spec: " << tspec);
  Tdatalist tdatalist(tspec);
  tdatalist.create();
  tdatalist.sort();
  testdata(tdatalist);
  if (bound_cnt != 0) {
    Tboundlist tboundlist(tspec);
    tboundlist.create();
    tboundlist.sort();
    testcmp(tboundlist, tdatalist);
  }
}

extern void NdbOut_Init();

static int
testmain()
{
  my_init();
  NdbOut_Init();
  signal(SIGABRT, SIG_DFL);
  { const char* p = NdbEnv_GetEnv("TEST_NDB_PACK_VERBOSE", (char*)0, 0);
    if (p != 0)
      verbose = atoi(p);
  }
  if (seed == 0)
    ll0("random seed: loop number");
  else {
    if (seed < 0)
      seed = getpid();
    ll0("random seed: " << seed);
    ndb_srand(seed);
  }
  loops = 100;
  int i;
  for (i = 0; loops == 0 || i < loops; i++) {
    ll0("loop:" << i << "/" << loops);
    if (seed == 0)
      ndb_srand(i);
    testrun();
  }
  // do not print "ok" in TAPTEST
  ndbout << "passed" << endl;
  return 0;
}

TAPTEST(NdbPack)
{
  int ret = testmain();
  return (ret == 0);
}

#endif