database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-07-30 16:24:05 +03:00
Code Activity

tux optim 14 - use pointers to cmp functions

Browse Source 
ndb/test/ndbapi/testOIBasic.cpp:
  null related options, not related to this changeset
This commit is contained in:
unknown
2004-08-27 13:57:51 +02:00
parent e387719d46
commit 47eb0b464c
7 changed files with 307 additions and 380 deletions
Download Patch File Download Diff File Expand all files Collapse all files

337
ndb/include/util/NdbSqlUtil.hpp
View File

@ -40,8 +40,9 @@ public:
* Compare kernel attribute values. Returns -1, 0, +1 for less, * Compare kernel attribute values. Returns -1, 0, +1 for less,
* equal, greater, respectively. Parameters are pointers to values, * equal, greater, respectively. Parameters are pointers to values,
* full attribute size in words, and size of available data in words. * full attribute size in words, and size of available data in words.
* There are 2 special return values to check first. All values fit * If available size is less than full size, CmpUnknown may be
* into a signed char. * returned. If a value cannot be parsed, it compares like NULL i.e.
* less than any valid value.
*/ */
typedef int Cmp(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size); typedef int Cmp(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size);
@ -49,8 +50,7 @@ public:
CmpLess = -1, CmpLess = -1,
CmpEqual = 0, CmpEqual = 0,
CmpGreater = 1, CmpGreater = 1,
CmpUnknown = 126, // insufficient partial data CmpUnknown = 2 // insufficient partial data
CmpError = 127 // bad data format or unimplemented comparison
}; };
/** /**
@ -82,19 +82,13 @@ public:
Text // Text blob Text // Text blob
}; };
Enum m_typeId; Enum m_typeId;
Cmp* m_cmp; // set to NULL if cmp not implemented Cmp* m_cmp; // comparison method
}; };
/** /**
* Get type by id. Can return the Undefined type. * Get type by id. Can return the Undefined type.
*/ */
static const Type& type(Uint32 typeId); static const Type& getType(Uint32 typeId);
/**
* Inline comparison method. Most or all real methods Type::m_cmp are
* implemented via this (trusting dead code elimination).
*/
static int cmp(Uint32 typeId, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size);
private: private:
/** /**
@ -127,323 +121,4 @@ private:
static Cmp cmpText; static Cmp cmpText;
}; };
inline int
NdbSqlUtil::cmp(Uint32 typeId, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{
// XXX require size >= 1
if (size > full)
return CmpError;
switch ((Type::Enum)typeId) {
case Type::Undefined:
break;
case Type::Tinyint:
{
if (size >= 1) {
union { Uint32 p[1]; Int8 v; } u1, u2;
u1.p[0] = p1[0];
u2.p[0] = p2[0];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
}
return CmpUnknown;
}
case Type::Tinyunsigned:
{
if (size >= 1) {
union { Uint32 p[1]; Uint8 v; } u1, u2;
u1.p[0] = p1[0];
u2.p[0] = p2[0];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
}
return CmpUnknown;
}
case Type::Smallint:
{
if (size >= 1) {
union { Uint32 p[1]; Int16 v; } u1, u2;
u1.p[0] = p1[0];
u2.p[0] = p2[0];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
}
return CmpUnknown;
}
case Type::Smallunsigned:
{
if (size >= 1) {
union { Uint32 p[1]; Uint16 v; } u1, u2;
u1.p[0] = p1[0];
u2.p[0] = p2[0];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
}
return CmpUnknown;
}
case Type::Mediumint:
{
if (size >= 1) {
union { const Uint32* p; const unsigned char* v; } u1, u2;
u1.p = p1;
u2.p = p2;
Int32 v1 = sint3korr(u1.v);
Int32 v2 = sint3korr(u2.v);
if (v1 < v2)
return -1;
if (v1 > v2)
return +1;
return 0;
}
return CmpUnknown;
}
case Type::Mediumunsigned:
{
if (size >= 1) {
union { const Uint32* p; const unsigned char* v; } u1, u2;
u1.p = p1;
u2.p = p2;
Uint32 v1 = uint3korr(u1.v);
Uint32 v2 = uint3korr(u2.v);
if (v1 < v2)
return -1;
if (v1 > v2)
return +1;
return 0;
}
return CmpUnknown;
}
case Type::Int:
{
if (size >= 1) {
union { Uint32 p[1]; Int32 v; } u1, u2;
u1.p[0] = p1[0];
u2.p[0] = p2[0];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
}
return CmpUnknown;
}
case Type::Unsigned:
{
if (size >= 1) {
union { Uint32 p[1]; Uint32 v; } u1, u2;
u1.v = p1[0];
u2.v = p2[0];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
}
return CmpUnknown;
}
case Type::Bigint:
{
if (size >= 2) {
union { Uint32 p[2]; Int64 v; } u1, u2;
u1.p[0] = p1[0];
u1.p[1] = p1[1];
u2.p[0] = p2[0];
u2.p[1] = p2[1];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
}
return CmpUnknown;
}
case Type::Bigunsigned:
{
if (size >= 2) {
union { Uint32 p[2]; Uint64 v; } u1, u2;
u1.p[0] = p1[0];
u1.p[1] = p1[1];
u2.p[0] = p2[0];
u2.p[1] = p2[1];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
}
return CmpUnknown;
}
case Type::Float:
{
if (size >= 1) {
union { Uint32 p[1]; float v; } u1, u2;
u1.p[0] = p1[0];
u2.p[0] = p2[0];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
}
return CmpUnknown;
}
case Type::Double:
{
if (size >= 2) {
union { Uint32 p[2]; double v; } u1, u2;
u1.p[0] = p1[0];
u1.p[1] = p1[1];
u2.p[0] = p2[0];
u2.p[1] = p2[1];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
}
return CmpUnknown;
}
case Type::Decimal:
// XXX not used by MySQL or NDB
break;
case Type::Char:
{
/*
* Char is blank-padded to length and null-padded to word size.
* There is no terminator so we must compare the full values.
*/
union { const Uint32* p; const char* v; } u1, u2;
u1.p = p1;
u2.p = p2;
int k = memcmp(u1.v, u2.v, size << 2);
return k < 0 ? -1 : k > 0 ? +1 : full == size ? 0 : CmpUnknown;
}
case Type::Varchar:
{
/*
* Varchar is not allowed to contain a null byte and the stored
* value is null-padded. Therefore comparison does not need to
* use the length.
*/
if (size >= 1) {
union { const Uint32* p; const char* v; } u1, u2;
u1.p = p1;
u2.p = p2;
// length in first 2 bytes
int k = strncmp(u1.v + 2, u2.v + 2, (size << 2) - 2);
return k < 0 ? -1 : k > 0 ? +1 : full == size ? 0 : CmpUnknown;
}
return CmpUnknown;
}
case Type::Binary:
{
// compare byte wise
union { const Uint32* p; const char* v; } u1, u2;
u1.p = p1;
u2.p = p2;
int k = memcmp(u1.v, u2.v, size << 2);
return k < 0 ? -1 : k > 0 ? +1 : full == size ? 0 : CmpUnknown;
}
case Type::Varbinary:
{
// assume correctly padded and compare byte wise
if (size >= 1) {
union { const Uint32* p; const char* v; } u1, u2;
u1.p = p1;
u2.p = p2;
// length in first 2 bytes
int k = memcmp(u1.v + 2, u2.v + 2, (size << 2) - 2);
return k < 0 ? -1 : k > 0 ? +1 : full == size ? 0 : CmpUnknown;
}
return CmpUnknown;
}
case Type::Datetime:
{
/*
* Datetime is CC YY MM DD hh mm ss \0
*/
if (size >= 1) {
union { const Uint32* p; const char* v; } u1, u2;
u1.p = p1;
u2.p = p2;
// skip format check
int k = memcmp(u1.v, u2.v, 4);
if (k != 0)
return k < 0 ? -1 : +1;
if (size >= 2) {
k = memcmp(u1.v + 4, u2.v + 4, 4);
return k < 0 ? -1 : k > 0 ? +1 : 0;
}
}
return CmpUnknown;
}
case Type::Timespec:
{
/*
* Timespec is CC YY MM DD hh mm ss \0 NN NN NN NN
*/
if (size >= 1) {
union { const Uint32* p; const char* v; } u1, u2;
u1.p = p1;
u2.p = p2;
// skip format check
int k = memcmp(u1.v, u2.v, 4);
if (k != 0)
return k < 0 ? -1 : +1;
if (size >= 2) {
k = memcmp(u1.v + 4, u2.v + 4, 4);
if (k != 0)
return k < 0 ? -1 : +1;
Uint32 n1 = *(const Uint32*)(u1.v + 8);
Uint32 n2 = *(const Uint32*)(u2.v + 8);
if (n1 < n2)
return -1;
if (n2 > n1)
return +1;
return 0;
}
}
return CmpUnknown;
}
case Type::Blob:
{
// skip blob head, the rest is binary
const unsigned skip = NDB_BLOB_HEAD_SIZE;
if (size >= skip + 1) {
union { const Uint32* p; const char* v; } u1, u2;
u1.p = p1 + skip;
u2.p = p2 + skip;
int k = memcmp(u1.v, u2.v, (size - 1) << 2);
return k < 0 ? -1 : k > 0 ? +1 : full == size ? 0 : CmpUnknown;
}
return CmpUnknown;
}
case Type::Text:
{
// skip blob head, the rest is char
const unsigned skip = NDB_BLOB_HEAD_SIZE;
if (size >= skip + 1) {
union { const Uint32* p; const char* v; } u1, u2;
u1.p = p1 + skip;
u2.p = p2 + skip;
int k = memcmp(u1.v, u2.v, (size - 1) << 2);
return k < 0 ? -1 : k > 0 ? +1 : full == size ? 0 : CmpUnknown;
}
return CmpUnknown;
}
}
return CmpError;
}
#endif #endif

269
ndb/src/common/util/NdbSqlUtil.cpp
View File

@ -167,7 +167,7 @@ NdbSqlUtil::m_typeList[] = {
}; };
const NdbSqlUtil::Type& const NdbSqlUtil::Type&
NdbSqlUtil::type(Uint32 typeId) NdbSqlUtil::getType(Uint32 typeId)
{ {
if (typeId < sizeof(m_typeList) / sizeof(m_typeList[0]) && if (typeId < sizeof(m_typeList) / sizeof(m_typeList[0]) &&
m_typeList[typeId].m_typeId != Type::Undefined) { m_typeList[typeId].m_typeId != Type::Undefined) {
@ -181,127 +181,352 @@ NdbSqlUtil::type(Uint32 typeId)
int int
NdbSqlUtil::cmpTinyint(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpTinyint(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Tinyint, p1, p2, full, size); assert(full >= size && size > 0);
union { Uint32 p[1]; Int8 v; } u1, u2;
u1.p[0] = p1[0];
u2.p[0] = p2[0];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
} }
int int
NdbSqlUtil::cmpTinyunsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpTinyunsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Tinyunsigned, p1, p2, full, size); assert(full >= size && size > 0);
union { Uint32 p[1]; Uint8 v; } u1, u2;
u1.p[0] = p1[0];
u2.p[0] = p2[0];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
} }
int int
NdbSqlUtil::cmpSmallint(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpSmallint(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Smallint, p1, p2, full, size); assert(full >= size && size > 0);
union { Uint32 p[1]; Int16 v; } u1, u2;
u1.p[0] = p1[0];
u2.p[0] = p2[0];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
} }
int int
NdbSqlUtil::cmpSmallunsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpSmallunsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Smallunsigned, p1, p2, full, size); assert(full >= size && size > 0);
union { Uint32 p[1]; Uint16 v; } u1, u2;
u1.p[0] = p1[0];
u2.p[0] = p2[0];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
} }
int int
NdbSqlUtil::cmpMediumint(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpMediumint(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Mediumint, p1, p2, full, size); assert(full >= size && size > 0);
union { const Uint32* p; const unsigned char* v; } u1, u2;
u1.p = p1;
u2.p = p2;
Int32 v1 = sint3korr(u1.v);
Int32 v2 = sint3korr(u2.v);
if (v1 < v2)
return -1;
if (v1 > v2)
return +1;
return 0;
} }
int int
NdbSqlUtil::cmpMediumunsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpMediumunsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Mediumunsigned, p1, p2, full, size); assert(full >= size && size > 0);
union { const Uint32* p; const unsigned char* v; } u1, u2;
u1.p = p1;
u2.p = p2;
Uint32 v1 = uint3korr(u1.v);
Uint32 v2 = uint3korr(u2.v);
if (v1 < v2)
return -1;
if (v1 > v2)
return +1;
return 0;
} }
int int
NdbSqlUtil::cmpInt(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpInt(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Int, p1, p2, full, size); assert(full >= size && size > 0);
union { Uint32 p[1]; Int32 v; } u1, u2;
u1.p[0] = p1[0];
u2.p[0] = p2[0];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
} }
int int
NdbSqlUtil::cmpUnsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpUnsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Unsigned, p1, p2, full, size); assert(full >= size && size > 0);
union { Uint32 p[1]; Uint32 v; } u1, u2;
u1.v = p1[0];
u2.v = p2[0];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
} }
int int
NdbSqlUtil::cmpBigint(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpBigint(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Bigint, p1, p2, full, size); assert(full >= size && size > 0);
if (size >= 2) {
union { Uint32 p[2]; Int64 v; } u1, u2;
u1.p[0] = p1[0];
u1.p[1] = p1[1];
u2.p[0] = p2[0];
u2.p[1] = p2[1];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
}
return CmpUnknown;
} }
int int
NdbSqlUtil::cmpBigunsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpBigunsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Bigunsigned, p1, p2, full, size); assert(full >= size && size > 0);
if (size >= 2) {
union { Uint32 p[2]; Uint64 v; } u1, u2;
u1.p[0] = p1[0];
u1.p[1] = p1[1];
u2.p[0] = p2[0];
u2.p[1] = p2[1];
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
}
return CmpUnknown;
} }
int int
NdbSqlUtil::cmpFloat(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpFloat(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Float, p1, p2, full, size); assert(full >= size && size > 0);
union { Uint32 p[1]; float v; } u1, u2;
u1.p[0] = p1[0];
u2.p[0] = p2[0];
// no format check
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
} }
int int
NdbSqlUtil::cmpDouble(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpDouble(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Double, p1, p2, full, size); assert(full >= size && size > 0);
if (size >= 2) {
union { Uint32 p[2]; double v; } u1, u2;
u1.p[0] = p1[0];
u1.p[1] = p1[1];
u2.p[0] = p2[0];
u2.p[1] = p2[1];
// no format check
if (u1.v < u2.v)
return -1;
if (u1.v > u2.v)
return +1;
return 0;
}
return CmpUnknown;
} }
int int
NdbSqlUtil::cmpDecimal(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpDecimal(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Decimal, p1, p2, full, size); assert(full >= size && size > 0);
// not used by MySQL or NDB
assert(false);
return 0;
} }
int int
NdbSqlUtil::cmpChar(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpChar(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Char, p1, p2, full, size); assert(full >= size && size > 0);
/*
* Char is blank-padded to length and null-padded to word size. There
* is no terminator so we compare the full values.
*/
union { const Uint32* p; const char* v; } u1, u2;
u1.p = p1;
u2.p = p2;
int k = memcmp(u1.v, u2.v, size << 2);
return k < 0 ? -1 : k > 0 ? +1 : full == size ? 0 : CmpUnknown;
} }
int int
NdbSqlUtil::cmpVarchar(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpVarchar(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Varchar, p1, p2, full, size); assert(full >= size && size > 0);
/*
* Varchar is not allowed to contain a null byte and the value is
* null-padded. Therefore comparison does not need to use the length.
*/
union { const Uint32* p; const char* v; } u1, u2;
u1.p = p1;
u2.p = p2;
// skip length in first 2 bytes
int k = strncmp(u1.v + 2, u2.v + 2, (size << 2) - 2);
return k < 0 ? -1 : k > 0 ? +1 : full == size ? 0 : CmpUnknown;
} }
int int
NdbSqlUtil::cmpBinary(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpBinary(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Binary, p1, p2, full, size); assert(full >= size && size > 0);
/*
* Binary data of full length. Compare bytewise.
*/
union { const Uint32* p; const unsigned char* v; } u1, u2;
u1.p = p1;
u2.p = p2;
int k = memcmp(u1.v, u2.v, size << 2);
return k < 0 ? -1 : k > 0 ? +1 : full == size ? 0 : CmpUnknown;
} }
int int
NdbSqlUtil::cmpVarbinary(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpVarbinary(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Varbinary, p1, p2, full, size); assert(full >= size && size > 0);
/*
* Binary data of variable length padded with nulls. The comparison
* does not need to use the length.
*/
union { const Uint32* p; const unsigned char* v; } u1, u2;
u1.p = p1;
u2.p = p2;
// skip length in first 2 bytes
int k = memcmp(u1.v + 2, u2.v + 2, (size << 2) - 2);
return k < 0 ? -1 : k > 0 ? +1 : full == size ? 0 : CmpUnknown;
} }
int int
NdbSqlUtil::cmpDatetime(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpDatetime(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Datetime, p1, p2, full, size); assert(full >= size && size > 0);
/*
* Datetime is CC YY MM DD hh mm ss \0
*/
union { const Uint32* p; const unsigned char* v; } u1, u2;
u1.p = p1;
u2.p = p2;
// no format check
int k = memcmp(u1.v, u2.v, 4);
if (k != 0)
return k < 0 ? -1 : +1;
if (size >= 2) {
k = memcmp(u1.v + 4, u2.v + 4, 4);
return k < 0 ? -1 : k > 0 ? +1 : 0;
}
return CmpUnknown;
} }
int int
NdbSqlUtil::cmpTimespec(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpTimespec(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Timespec, p1, p2, full, size); assert(full >= size && size > 0);
/*
* Timespec is CC YY MM DD hh mm ss \0 NN NN NN NN
*/
union { const Uint32* p; const unsigned char* v; } u1, u2;
u1.p = p1;
u2.p = p2;
// no format check
int k = memcmp(u1.v, u2.v, 4);
if (k != 0)
return k < 0 ? -1 : +1;
if (size >= 2) {
k = memcmp(u1.v + 4, u2.v + 4, 4);
if (k != 0)
return k < 0 ? -1 : +1;
if (size >= 3) {
Uint32 n1 = *(const Uint32*)(u1.v + 8);
Uint32 n2 = *(const Uint32*)(u2.v + 8);
if (n1 < n2)
return -1;
if (n2 > n1)
return +1;
return 0;
}
}
return CmpUnknown;
} }
int int
NdbSqlUtil::cmpBlob(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpBlob(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Blob, p1, p2, full, size); assert(full >= size && size > 0);
/*
* Blob comparison is on the inline bytes. Except for larger header
* the format is like Varbinary.
*/
const unsigned head = NDB_BLOB_HEAD_SIZE;
// skip blob head
if (size >= head + 1) {
union { const Uint32* p; const unsigned char* v; } u1, u2;
u1.p = p1 + head;
u2.p = p2 + head;
int k = memcmp(u1.v, u2.v, (size - head) << 2);
return k < 0 ? -1 : k > 0 ? +1 : full == size ? 0 : CmpUnknown;
}
return CmpUnknown;
} }
int int
NdbSqlUtil::cmpText(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) NdbSqlUtil::cmpText(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{ {
return cmp(Type::Text, p1, p2, full, size); assert(full >= size && size > 0);
/*
* Text comparison is on the inline bytes. Except for larger header
* the format is like Varchar.
*/
const unsigned head = NDB_BLOB_HEAD_SIZE;
// skip blob head
if (size >= head + 1) {
union { const Uint32* p; const char* v; } u1, u2;
u1.p = p1 + head;
u2.p = p2 + head;
int k = memcmp(u1.v, u2.v, (size - head) << 2);
return k < 0 ? -1 : k > 0 ? +1 : full == size ? 0 : CmpUnknown;
}
return CmpUnknown;
} }
#ifdef NDB_SQL_UTIL_TEST #ifdef NDB_SQL_UTIL_TEST

32
ndb/src/kernel/blocks/dbtux/DbtuxCmp.cpp
View File

@ -35,7 +35,7 @@ Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, Cons
searchKey += start; searchKey += start;
int ret = 0; int ret = 0;
while (start < numAttrs) { while (start < numAttrs) {
if (len2 < AttributeHeaderSize) { if (len2 <= AttributeHeaderSize) {
jam(); jam();
ret = NdbSqlUtil::CmpUnknown; ret = NdbSqlUtil::CmpUnknown;
break; break;
@ -46,7 +46,8 @@ Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, Cons
jam(); jam();
// current attribute // current attribute
const DescAttr& descAttr = descEnt.m_descAttr[start]; const DescAttr& descAttr = descEnt.m_descAttr[start];
const unsigned typeId = descAttr.m_typeId; const NdbSqlUtil::Type& type = NdbSqlUtil::getType(descAttr.m_typeId);
ndbassert(type.m_typeId != NdbSqlUtil::Type::Undefined);
// full data size // full data size
const unsigned size1 = AttributeDescriptor::getSizeInWords(descAttr.m_attrDesc); const unsigned size1 = AttributeDescriptor::getSizeInWords(descAttr.m_attrDesc);
ndbrequire(size1 != 0 && size1 == entryData.ah().getDataSize()); ndbrequire(size1 != 0 && size1 == entryData.ah().getDataSize());
@ -55,7 +56,7 @@ Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, Cons
// compare // compare
const Uint32* const p1 = *searchKey; const Uint32* const p1 = *searchKey;
const Uint32* const p2 = &entryData[AttributeHeaderSize]; const Uint32* const p2 = &entryData[AttributeHeaderSize];
ret = NdbSqlUtil::cmp(typeId, p1, p2, size1, size2); ret = (*type.m_cmp)(p1, p2, size1, size2);
if (ret != 0) { if (ret != 0) {
jam(); jam();
break; break;
@ -78,8 +79,6 @@ Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, Cons
entryData += AttributeHeaderSize + entryData.ah().getDataSize(); entryData += AttributeHeaderSize + entryData.ah().getDataSize();
start++; start++;
} }
// XXX until data format errors are handled
ndbrequire(ret != NdbSqlUtil::CmpError);
return ret; return ret;
} }
@ -103,13 +102,14 @@ Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, Tabl
jam(); jam();
// current attribute // current attribute
const DescAttr& descAttr = descEnt.m_descAttr[start]; const DescAttr& descAttr = descEnt.m_descAttr[start];
const unsigned typeId = descAttr.m_typeId; const NdbSqlUtil::Type& type = NdbSqlUtil::getType(descAttr.m_typeId);
ndbassert(type.m_typeId != NdbSqlUtil::Type::Undefined);
// full data size // full data size
const unsigned size1 = AttributeDescriptor::getSizeInWords(descAttr.m_attrDesc); const unsigned size1 = AttributeDescriptor::getSizeInWords(descAttr.m_attrDesc);
// compare // compare
const Uint32* const p1 = *searchKey; const Uint32* const p1 = *searchKey;
const Uint32* const p2 = *entryKey; const Uint32* const p2 = *entryKey;
ret = NdbSqlUtil::cmp(typeId, p1, p2, size1, size1); ret = (*type.m_cmp)(p1, p2, size1, size1);
if (ret != 0) { if (ret != 0) {
jam(); jam();
break; break;
@ -132,8 +132,6 @@ Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, Tabl
entryKey += 1; entryKey += 1;
start++; start++;
} }
// XXX until data format errors are handled
ndbrequire(ret != NdbSqlUtil::CmpError);
return ret; return ret;
} }
@ -172,7 +170,7 @@ Dbtux::cmpScanBound(const Frag& frag, unsigned dir, ConstData boundInfo, unsigne
*/ */
unsigned type = 4; unsigned type = 4;
while (boundCount != 0) { while (boundCount != 0) {
if (len2 < AttributeHeaderSize) { if (len2 <= AttributeHeaderSize) {
jam(); jam();
return NdbSqlUtil::CmpUnknown; return NdbSqlUtil::CmpUnknown;
} }
@ -186,7 +184,8 @@ Dbtux::cmpScanBound(const Frag& frag, unsigned dir, ConstData boundInfo, unsigne
// current attribute // current attribute
const unsigned index = boundInfo.ah().getAttributeId(); const unsigned index = boundInfo.ah().getAttributeId();
const DescAttr& descAttr = descEnt.m_descAttr[index]; const DescAttr& descAttr = descEnt.m_descAttr[index];
const unsigned typeId = descAttr.m_typeId; const NdbSqlUtil::Type& type = NdbSqlUtil::getType(descAttr.m_typeId);
ndbassert(type.m_typeId != NdbSqlUtil::Type::Undefined);
ndbrequire(entryData.ah().getAttributeId() == descAttr.m_primaryAttrId); ndbrequire(entryData.ah().getAttributeId() == descAttr.m_primaryAttrId);
// full data size // full data size
const unsigned size1 = boundInfo.ah().getDataSize(); const unsigned size1 = boundInfo.ah().getDataSize();
@ -196,9 +195,7 @@ Dbtux::cmpScanBound(const Frag& frag, unsigned dir, ConstData boundInfo, unsigne
// compare // compare
const Uint32* const p1 = &boundInfo[AttributeHeaderSize]; const Uint32* const p1 = &boundInfo[AttributeHeaderSize];
const Uint32* const p2 = &entryData[AttributeHeaderSize]; const Uint32* const p2 = &entryData[AttributeHeaderSize];
int ret = NdbSqlUtil::cmp(typeId, p1, p2, size1, size2); int ret = (*type.m_cmp)(p1, p2, size1, size2);
// XXX until data format errors are handled
ndbrequire(ret != NdbSqlUtil::CmpError);
if (ret != 0) { if (ret != 0) {
jam(); jam();
return ret; return ret;
@ -269,15 +266,14 @@ Dbtux::cmpScanBound(const Frag& frag, unsigned dir, ConstData boundInfo, unsigne
// current attribute // current attribute
const unsigned index = boundInfo.ah().getAttributeId(); const unsigned index = boundInfo.ah().getAttributeId();
const DescAttr& descAttr = descEnt.m_descAttr[index]; const DescAttr& descAttr = descEnt.m_descAttr[index];
const unsigned typeId = descAttr.m_typeId; const NdbSqlUtil::Type& type = NdbSqlUtil::getType(descAttr.m_typeId);
ndbassert(type.m_typeId != NdbSqlUtil::Type::Undefined);
// full data size // full data size
const unsigned size1 = AttributeDescriptor::getSizeInWords(descAttr.m_attrDesc); const unsigned size1 = AttributeDescriptor::getSizeInWords(descAttr.m_attrDesc);
// compare // compare
const Uint32* const p1 = &boundInfo[AttributeHeaderSize]; const Uint32* const p1 = &boundInfo[AttributeHeaderSize];
const Uint32* const p2 = *entryKey; const Uint32* const p2 = *entryKey;
int ret = NdbSqlUtil::cmp(typeId, p1, p2, size1, size1); int ret = (*type.m_cmp)(p1, p2, size1, size1);
// XXX until data format errors are handled
ndbrequire(ret != NdbSqlUtil::CmpError);
if (ret != 0) { if (ret != 0) {
jam(); jam();
return ret; return ret;

2
ndb/src/kernel/blocks/dbtux/DbtuxMeta.cpp
View File

@ -184,7 +184,7 @@ Dbtux::execTUX_ADD_ATTRREQ(Signal* signal)
} }
#endif #endif
// check if type is valid and has a comparison method // check if type is valid and has a comparison method
const NdbSqlUtil::Type& type = NdbSqlUtil::type(descAttr.m_typeId); const NdbSqlUtil::Type& type = NdbSqlUtil::getType(descAttr.m_typeId);
if (type.m_typeId == NdbSqlUtil::Type::Undefined || if (type.m_typeId == NdbSqlUtil::Type::Undefined ||
type.m_cmp == 0) { type.m_cmp == 0) {
jam(); jam();

12
ndb/src/kernel/blocks/dbtux/Times.txt
View File

@ -28,6 +28,8 @@ d
shows ms / 1000 rows for each and index time overhead shows ms / 1000 rows for each and index time overhead
samples 10% of all PKs (100,000 pk reads, 100,000 scans) samples 10% of all PKs (100,000 pk reads, 100,000 scans)
the "pct" values are from more accurate total times (not shown)
040616 mc02/a 40 ms 87 ms 114 pct 040616 mc02/a 40 ms 87 ms 114 pct
mc02/b 51 ms 128 ms 148 pct mc02/b 51 ms 128 ms 148 pct
@ -75,11 +77,19 @@ optim 13 mc02/a 40 ms 57 ms 42 pct
mc02/d 170 ms 256 ms 50 pct mc02/d 170 ms 256 ms 50 pct
after wl-1884 store all-NULL keys (the tests have pctnull=10 per column) after wl-1884 store all-NULL keys (the tests have pctnull=10 per column)
[ what happened to PK read performance? ]
optim 13 mc02/a 39 ms 59 ms 50 pct optim 13 mc02/a 39 ms 59 ms 50 pct
mc02/b 47 ms 77 ms 61 pct mc02/b 47 ms 77 ms 61 pct
mc02/c 9 ms 12 ms 44 pct mc02/c 9 ms 12 ms 44 pct
mc02/d 246 ms 289 ms 17 pct mc02/d 246 ms 289 ms 17 pct
[ case d: what happened to PK read performance? ]
optim 14 mc02/a 41 ms 60 ms 44 pct
mc02/b 46 ms 81 ms 73 pct
mc02/c 9 ms 13 ms 37 pct
mc02/d 242 ms 285 ms 17 pct
[ case b: do long keys suffer from many subroutine calls? ]
vim: set et: vim: set et:

4
ndb/src/ndbapi/NdbScanOperation.cpp
View File

@ -1227,10 +1227,10 @@ NdbIndexScanOperation::compare(Uint32 skip, Uint32 cols,
Uint32 type = NdbColumnImpl::getImpl(* r1->m_column).m_extType; Uint32 type = NdbColumnImpl::getImpl(* r1->m_column).m_extType;
Uint32 size = (r1->theAttrSize * r1->theArraySize + 3) / 4; Uint32 size = (r1->theAttrSize * r1->theArraySize + 3) / 4;
if(!r1_null){ if(!r1_null){
char r = NdbSqlUtil::cmp(type, d1, d2, size, size); const NdbSqlUtil::Type& t = NdbSqlUtil::getType(type);
int r = (*t.m_cmp)(d1, d2, size, size);
if(r){ if(r){
assert(r != NdbSqlUtil::CmpUnknown); assert(r != NdbSqlUtil::CmpUnknown);
assert(r != NdbSqlUtil::CmpError);
return r; return r;
} }
} }

31
ndb/test/ndbapi/testOIBasic.cpp
View File

@ -34,6 +34,7 @@
struct Opt { struct Opt {
// common options // common options
unsigned m_batch; unsigned m_batch;
const char* m_bound;
const char* m_case; const char* m_case;
bool m_core; bool m_core;
bool m_dups; bool m_dups;
@ -43,6 +44,7 @@ struct Opt {
unsigned m_loop; unsigned m_loop;
bool m_nologging; bool m_nologging;
bool m_msglock; bool m_msglock;
unsigned m_pctnull;
unsigned m_rows; unsigned m_rows;
unsigned m_samples; unsigned m_samples;
unsigned m_scanrd; unsigned m_scanrd;
@ -54,6 +56,7 @@ struct Opt {
unsigned m_v; unsigned m_v;
Opt() : Opt() :
m_batch(32), m_batch(32),
m_bound("01234"),
m_case(0), m_case(0),
m_core(false), m_core(false),
m_dups(false), m_dups(false),
@ -63,6 +66,7 @@ struct Opt {
m_loop(1), m_loop(1),
m_nologging(false), m_nologging(false),
m_msglock(true), m_msglock(true),
m_pctnull(10),
m_rows(1000), m_rows(1000),
m_samples(0), m_samples(0),
m_scanrd(240), m_scanrd(240),
@ -87,6 +91,7 @@ printhelp()
ndbout ndbout
<< "usage: testOIbasic [options]" << endl << "usage: testOIbasic [options]" << endl
<< " -batch N pk operations in batch [" << d.m_batch << "]" << endl << " -batch N pk operations in batch [" << d.m_batch << "]" << endl
<< " -bound xyz use only these bound types 0-4 [" << d.m_bound << "]" << endl
<< " -case abc only given test cases (letters a-z)" << endl << " -case abc only given test cases (letters a-z)" << endl
<< " -core core dump on error [" << d.m_core << "]" << endl << " -core core dump on error [" << d.m_core << "]" << endl
<< " -dups allow duplicate tuples from index scan [" << d.m_dups << "]" << endl << " -dups allow duplicate tuples from index scan [" << d.m_dups << "]" << endl
@ -94,6 +99,7 @@ printhelp()
<< " -index xyz only given index numbers (digits 1-9)" << endl << " -index xyz only given index numbers (digits 1-9)" << endl
<< " -loop N loop count full suite 0=forever [" << d.m_loop << "]" << endl << " -loop N loop count full suite 0=forever [" << d.m_loop << "]" << endl
<< " -nologging create tables in no-logging mode" << endl << " -nologging create tables in no-logging mode" << endl
<< " -pctnull N pct NULL values in nullable column [" << d.m_pctnull << "]" << endl
<< " -rows N rows per thread [" << d.m_rows << "]" << endl << " -rows N rows per thread [" << d.m_rows << "]" << endl
<< " -samples N samples for some timings (0=all) [" << d.m_samples << "]" << endl << " -samples N samples for some timings (0=all) [" << d.m_samples << "]" << endl
<< " -scanrd N scan read parallelism [" << d.m_scanrd << "]" << endl << " -scanrd N scan read parallelism [" << d.m_scanrd << "]" << endl
@ -198,7 +204,6 @@ struct Par : public Opt {
Tmr& tmr() const { assert(m_tmr != 0); return *m_tmr; } Tmr& tmr() const { assert(m_tmr != 0); return *m_tmr; }
unsigned m_totrows; unsigned m_totrows;
// value calculation // value calculation
unsigned m_pctnull;
unsigned m_range; unsigned m_range;
unsigned m_pctrange; unsigned m_pctrange;
// do verify after read // do verify after read
@ -214,7 +219,6 @@ struct Par : public Opt {
m_set(0), m_set(0),
m_tmr(0), m_tmr(0),
m_totrows(m_threads * m_rows), m_totrows(m_threads * m_rows),
m_pctnull(10),
m_range(m_rows), m_range(m_rows),
m_pctrange(0), m_pctrange(0),
m_verify(false), m_verify(false),
@ -1622,7 +1626,6 @@ Set::calc(Par par, unsigned i)
m_row[i] = new Row(tab); m_row[i] = new Row(tab);
Row& row = *m_row[i]; Row& row = *m_row[i];
// value generation parameters // value generation parameters
par.m_pctnull = 10;
par.m_pctrange = 40; par.m_pctrange = 40;
row.calc(par, i); row.calc(par, i);
} }
@ -1898,8 +1901,11 @@ BSet::calc(Par par)
BVal& bval = *new BVal(icol); BVal& bval = *new BVal(icol);
m_bval[m_bvals++] = &bval; m_bval[m_bvals++] = &bval;
bval.m_null = false; bval.m_null = false;
// equality bound only on i==0 unsigned sel;
unsigned sel = urandom(5 - i); do {
// equality bound only on i==0
sel = urandom(5 - i);
} while (strchr(par.m_bound, '0' + sel) == 0);
if (sel < 2) if (sel < 2)
bval.m_type = 0 | (1 << i); bval.m_type = 0 | (1 << i);
else if (sel < 4) else if (sel < 4)
@ -3207,6 +3213,15 @@ NDB_COMMAND(testOIBasic, "testOIBasic", "testOIBasic", "testOIBasic", 65535)
continue; continue;
} }
} }
if (strcmp(arg, "-bound") == 0) {
if (++argv, --argc > 0) {
const char* p = argv[0];
if (strlen(p) != 0 && strlen(p) == strspn(p, "01234")) {
g_opt.m_bound = strdup(p);
continue;
}
}
}
if (strcmp(arg, "-case") == 0) { if (strcmp(arg, "-case") == 0) {
if (++argv, --argc > 0) { if (++argv, --argc > 0) {
g_opt.m_case = strdup(argv[0]); g_opt.m_case = strdup(argv[0]);
@ -3257,6 +3272,12 @@ NDB_COMMAND(testOIBasic, "testOIBasic", "testOIBasic", "testOIBasic", 65535)
g_opt.m_nologging = true; g_opt.m_nologging = true;
continue; continue;
} }
if (strcmp(arg, "-pctnull") == 0) {
if (++argv, --argc > 0) {
g_opt.m_pctnull = atoi(argv[0]);
continue;
}
}
if (strcmp(arg, "-rows") == 0) { if (strcmp(arg, "-rows") == 0) {
if (++argv, --argc > 0) { if (++argv, --argc > 0) {
g_opt.m_rows = atoi(argv[0]); g_opt.m_rows = atoi(argv[0]);