Skip to content

M
mariadb
Commit 415d163a authored by pekka@mysql.com's avatar pekka@mysql.com
Browse files
Options

Merge pnousiainen@bk-internal.mysql.com:/home/bk/mysql-4.1-ndb 

into mysql.com:/space/pekka/ndb/version/my41-tux
parents 0b9a99db 75586fe4
Hide whitespace changes
Inline Side-by-side
Showing with 307 additions and 380 deletions
ndb/include/util/NdbSqlUtil.hpp
View file @ 415d163a
......@@ -40,8 +40,9 @@ public:
* Compare kernel attribute values. Returns -1, 0, +1 for less,
* equal, greater, respectively. Parameters are pointers to values,
* full attribute size in words, and size of available data in words.
* There are 2 special return values to check first. All values fit
* into a signed char.
* If available size is less than full size, CmpUnknown may be
* 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);
......@@ -49,8 +50,7 @@ public:
CmpLess = -1,
CmpEqual = 0,
CmpGreater = 1,
CmpUnknown = 126, // insufficient partial data
CmpError = 127 // bad data format or unimplemented comparison
CmpUnknown = 2 // insufficient partial data
};
/**
......@@ -82,19 +82,13 @@ public:
Text // Text blob
};
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.
*/
static const Type& type(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);
static const Type& getType(Uint32 typeId);
private:
/**
......@@ -127,323 +121,4 @@ private:
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
ndb/src/common/util/NdbSqlUtil.cpp
View file @ 415d163a
......@@ -167,7 +167,7 @@ NdbSqlUtil::m_typeList[] = {
};
const NdbSqlUtil::Type&
NdbSqlUtil::type(Uint32 typeId)
NdbSqlUtil::getType(Uint32 typeId)
{
if (typeId < sizeof(m_typeList) / sizeof(m_typeList[0]) &&
m_typeList[typeId].m_typeId != Type::Undefined) {
......@@ -181,127 +181,352 @@ NdbSqlUtil::type(Uint32 typeId)
int
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
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
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
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
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
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
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
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
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
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
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
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
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
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
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
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
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
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
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
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
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
......
ndb/src/kernel/blocks/dbtux/DbtuxCmp.cpp
View file @ 415d163a
......@@ -35,7 +35,7 @@ Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, Cons
searchKey += start;
int ret = 0;
while (start < numAttrs) {
if (len2 < AttributeHeaderSize) {
if (len2 <= AttributeHeaderSize) {
jam();
ret = NdbSqlUtil::CmpUnknown;
break;
......@@ -46,7 +46,8 @@ Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, Cons
jam();
// current attribute
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
const unsigned size1 = AttributeDescriptor::getSizeInWords(descAttr.m_attrDesc);
ndbrequire(size1 != 0 && size1 == entryData.ah().getDataSize());
......@@ -55,7 +56,7 @@ Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, Cons
// compare
const Uint32* const p1 = *searchKey;
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) {
jam();
break;
......@@ -78,8 +79,6 @@ Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, Cons
entryData += AttributeHeaderSize + entryData.ah().getDataSize();
start++;
}
// XXX until data format errors are handled
ndbrequire(ret != NdbSqlUtil::CmpError);
return ret;
}
......@@ -103,13 +102,14 @@ Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, Tabl
jam();
// current attribute
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
const unsigned size1 = AttributeDescriptor::getSizeInWords(descAttr.m_attrDesc);
// compare
const Uint32* const p1 = *searchKey;
const Uint32* const p2 = *entryKey;
ret = NdbSqlUtil::cmp(typeId, p1, p2, size1, size1);
ret = (*type.m_cmp)(p1, p2, size1, size1);
if (ret != 0) {
jam();
break;
......@@ -132,8 +132,6 @@ Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, Tabl
entryKey += 1;
start++;
}
// XXX until data format errors are handled
ndbrequire(ret != NdbSqlUtil::CmpError);
return ret;
}
......@@ -172,7 +170,7 @@ Dbtux::cmpScanBound(const Frag& frag, unsigned dir, ConstData boundInfo, unsigne
*/
unsigned type = 4;
while (boundCount != 0) {
if (len2 < AttributeHeaderSize) {
if (len2 <= AttributeHeaderSize) {
jam();
return NdbSqlUtil::CmpUnknown;
}
......@@ -186,7 +184,8 @@ Dbtux::cmpScanBound(const Frag& frag, unsigned dir, ConstData boundInfo, unsigne
// current attribute
const unsigned index = boundInfo.ah().getAttributeId();
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);
// full data size
const unsigned size1 = boundInfo.ah().getDataSize();
......@@ -196,9 +195,7 @@ Dbtux::cmpScanBound(const Frag& frag, unsigned dir, ConstData boundInfo, unsigne
// compare
const Uint32* const p1 = &boundInfo[AttributeHeaderSize];
const Uint32* const p2 = &entryData[AttributeHeaderSize];
int ret = NdbSqlUtil::cmp(typeId, p1, p2, size1, size2);
// XXX until data format errors are handled
ndbrequire(ret != NdbSqlUtil::CmpError);
int ret = (*type.m_cmp)(p1, p2, size1, size2);
if (ret != 0) {
jam();
return ret;
......@@ -269,15 +266,14 @@ Dbtux::cmpScanBound(const Frag& frag, unsigned dir, ConstData boundInfo, unsigne
// current attribute
const unsigned index = boundInfo.ah().getAttributeId();
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
const unsigned size1 = AttributeDescriptor::getSizeInWords(descAttr.m_attrDesc);
// compare
const Uint32* const p1 = &boundInfo[AttributeHeaderSize];
const Uint32* const p2 = *entryKey;
int ret = NdbSqlUtil::cmp(typeId, p1, p2, size1, size1);
// XXX until data format errors are handled
ndbrequire(ret != NdbSqlUtil::CmpError);
int ret = (*type.m_cmp)(p1, p2, size1, size1);
if (ret != 0) {
jam();
return ret;
......
ndb/src/kernel/blocks/dbtux/DbtuxMeta.cpp
View file @ 415d163a
......@@ -184,7 +184,7 @@ Dbtux::execTUX_ADD_ATTRREQ(Signal* signal)
}
#endif
// 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 ||
type.m_cmp == 0) {
jam();
......
ndb/src/kernel/blocks/dbtux/Times.txt
View file @ 415d163a
......@@ -28,6 +28,8 @@ d
shows ms / 1000 rows for each and index time overhead
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
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
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
mc02/b 47 ms 77 ms 61 pct
mc02/c 9 ms 12 ms 44 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:
ndb/src/ndbapi/NdbScanOperation.cpp
View file @ 415d163a
......@@ -1224,10 +1224,10 @@ NdbIndexScanOperation::compare(Uint32 skip, Uint32 cols,
Uint32 type = NdbColumnImpl::getImpl(* r1->m_column).m_extType;
Uint32 size = (r1->theAttrSize * r1->theArraySize + 3) / 4;
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){
assert(r != NdbSqlUtil::CmpUnknown);
assert(r != NdbSqlUtil::CmpError);
return r;
}
}
......
ndb/test/ndbapi/testOIBasic.cpp
View file @ 415d163a
......@@ -34,6 +34,7 @@
struct Opt {
// common options
unsigned m_batch;
const char* m_bound;
const char* m_case;
bool m_core;
bool m_dups;
......@@ -43,6 +44,7 @@ struct Opt {
unsigned m_loop;
bool m_nologging;
bool m_msglock;
unsigned m_pctnull;
unsigned m_rows;
unsigned m_samples;
unsigned m_scanrd;
......@@ -54,6 +56,7 @@ struct Opt {
unsigned m_v;
Opt() :
m_batch(32),
m_bound("01234"),
m_case(0),
m_core(false),
m_dups(false),
......@@ -63,6 +66,7 @@ struct Opt {
m_loop(1),
m_nologging(false),
m_msglock(true),
m_pctnull(10),
m_rows(1000),
m_samples(0),
m_scanrd(240),
......@@ -87,6 +91,7 @@ printhelp()
ndbout
<< "usage: testOIbasic [options]" << 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
<< " -core core dump on error [" << d.m_core << "]" << 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
<< " -loop N loop count full suite 0=forever [" << d.m_loop << "]" << 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
<< " -samples N samples for some timings (0=all) [" << d.m_samples << "]" << 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; }
unsigned m_totrows;
// value calculation
unsigned m_pctnull;
unsigned m_range;
unsigned m_pctrange;
// do verify after read
......@@ -214,7 +219,6 @@ struct Par : public Opt {
m_set(0),
m_tmr(0),
m_totrows(m_threads * m_rows),
m_pctnull(10),
m_range(m_rows),
m_pctrange(0),
m_verify(false),
......@@ -1622,7 +1626,6 @@ Set::calc(Par par, unsigned i)
m_row[i] = new Row(tab);
Row& row = *m_row[i];
// value generation parameters
par.m_pctnull = 10;
par.m_pctrange = 40;
row.calc(par, i);
}
......@@ -1898,8 +1901,11 @@ BSet::calc(Par par)
BVal& bval = *new BVal(icol);
m_bval[m_bvals++] = &bval;
bval.m_null = false;
// equality bound only on i==0
unsigned sel = urandom(5 - i);
unsigned sel;
do {
// equality bound only on i==0
sel = urandom(5 - i);
} while (strchr(par.m_bound, '0' + sel) == 0);
if (sel < 2)
bval.m_type = 0 | (1 << i);
else if (sel < 4)
......@@ -3207,6 +3213,15 @@ NDB_COMMAND(testOIBasic, "testOIBasic", "testOIBasic", "testOIBasic", 65535)
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 (++argv, --argc > 0) {
g_opt.m_case = strdup(argv[0]);
......@@ -3257,6 +3272,12 @@ NDB_COMMAND(testOIBasic, "testOIBasic", "testOIBasic", "testOIBasic", 65535)
g_opt.m_nologging = true;
continue;
}
if (strcmp(arg, "-pctnull") == 0) {
if (++argv, --argc > 0) {
g_opt.m_pctnull = atoi(argv[0]);
continue;
}
}
if (strcmp(arg, "-rows") == 0) {
if (++argv, --argc > 0) {
g_opt.m_rows = atoi(argv[0]);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7