From 72fedd949429c8bbe7e7bce8461e4582f27c683e Mon Sep 17 00:00:00 2001 From: unknown Date: Fri, 17 Sep 2004 14:09:21 +0200 Subject: [PATCH] ndb charsets (wl-1732) final part: use strxfrm + strcoll ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp: oops ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp: jamEntry ndb/src/kernel/blocks/dblqh/DblqhMain.cpp: jamEntry mysql-test/r/ndb_index.result: ndb charsets: minimal fix to make test pass mysql-test/t/ndb_index.test: ndb charsets: minimal fix to make test pass --- mysql-test/r/ndb_charset.result | 191 ++++++++++++++++++ mysql-test/r/ndb_index.result | 2 +- mysql-test/t/ndb_charset.test | 159 +++++++++++++++ mysql-test/t/ndb_index.test | 2 +- ndb/include/util/NdbSqlUtil.hpp | 6 +- ndb/src/common/util/NdbSqlUtil.cpp | 88 ++++---- ndb/src/kernel/blocks/dblqh/DblqhMain.cpp | 1 + ndb/src/kernel/blocks/dbtup/Dbtup.hpp | 18 +- .../kernel/blocks/dbtup/DbtupExecQuery.cpp | 21 +- ndb/src/kernel/blocks/dbtup/DbtupIndex.cpp | 4 +- ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp | 6 +- ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp | 83 +++++++- ndb/src/kernel/blocks/dbtup/DbtupTrigger.cpp | 9 +- ndb/src/kernel/blocks/dbtux/DbtuxCmp.cpp | 5 +- ndb/src/kernel/blocks/dbtux/DbtuxGen.cpp | 4 +- ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp | 1 + ndb/src/kernel/blocks/dbtux/Times.txt | 18 +- ndb/src/ndbapi/NdbIndexOperation.cpp | 32 ++- ndb/src/ndbapi/NdbOperationDefine.cpp | 11 + ndb/src/ndbapi/NdbOperationSearch.cpp | 31 ++- ndb/src/ndbapi/NdbScanOperation.cpp | 35 +++- ndb/src/ndbapi/ndberror.c | 2 +- ndb/test/ndbapi/testOIBasic.cpp | 20 ++ 23 files changed, 672 insertions(+), 77 deletions(-) create mode 100644 mysql-test/r/ndb_charset.result create mode 100644 mysql-test/t/ndb_charset.test diff --git a/mysql-test/r/ndb_charset.result b/mysql-test/r/ndb_charset.result new file mode 100644 index 00000000000..93429a1fcb0 --- /dev/null +++ b/mysql-test/r/ndb_charset.result @@ -0,0 +1,191 @@ +drop table if exists t1; +create table t1 ( +a char(3) character set latin1 collate latin1_bin primary key +) engine=ndb; +insert into t1 values('aAa'); +insert into t1 values('aaa'); +insert into t1 values('AAA'); +select * from t1 order by a; +a +AAA +aAa +aaa +select * from t1 where a = 'aAa'; +a +aAa +select * from t1 where a = 'aaa'; +a +aaa +select * from t1 where a = 'AaA'; +a +select * from t1 where a = 'AAA'; +a +AAA +drop table t1; +create table t1 ( +a char(3) character set latin1 collate latin1_swedish_ci primary key +) engine=ndb; +insert into t1 values('aAa'); +insert into t1 values('aaa'); +ERROR 23000: Duplicate entry 'aaa' for key 1 +insert into t1 values('AAA'); +ERROR 23000: Duplicate entry 'AAA' for key 1 +select * from t1 order by a; +a +aAa +select * from t1 where a = 'aAa'; +a +aAa +select * from t1 where a = 'aaa'; +a +aAa +select * from t1 where a = 'AaA'; +a +aAa +select * from t1 where a = 'AAA'; +a +aAa +drop table t1; +create table t1 ( +p int primary key, +a char(3) character set latin1 collate latin1_bin not null, +unique key(a) +) engine=ndb; +insert into t1 values(1, 'aAa'); +insert into t1 values(2, 'aaa'); +insert into t1 values(3, 'AAA'); +select * from t1 order by p; +p a +1 aAa +2 aaa +3 AAA +select * from t1 where a = 'aAa'; +p a +1 aAa +select * from t1 where a = 'aaa'; +p a +2 aaa +select * from t1 where a = 'AaA'; +p a +select * from t1 where a = 'AAA'; +p a +3 AAA +drop table t1; +create table t1 ( +p int primary key, +a char(3) character set latin1 collate latin1_swedish_ci not null, +unique key(a) +) engine=ndb; +insert into t1 values(1, 'aAa'); +insert into t1 values(2, 'aaa'); +ERROR 23000: Can't write, because of unique constraint, to table 't1' +insert into t1 values(3, 'AAA'); +ERROR 23000: Can't write, because of unique constraint, to table 't1' +select * from t1 order by p; +p a +1 aAa +select * from t1 where a = 'aAa'; +p a +1 aAa +select * from t1 where a = 'aaa'; +p a +1 aAa +select * from t1 where a = 'AaA'; +p a +1 aAa +select * from t1 where a = 'AAA'; +p a +1 aAa +drop table t1; +create table t1 ( +p int primary key, +a char(3) character set latin1 collate latin1_bin not null, +index(a) +) engine=ndb; +insert into t1 values(1, 'aAa'); +insert into t1 values(2, 'aaa'); +insert into t1 values(3, 'AAA'); +insert into t1 values(4, 'aAa'); +insert into t1 values(5, 'aaa'); +insert into t1 values(6, 'AAA'); +select * from t1 order by p; +p a +1 aAa +2 aaa +3 AAA +4 aAa +5 aaa +6 AAA +explain select * from t1 where a = 'zZz' order by p; +id select_type table type possible_keys key key_len ref rows Extra +1 SIMPLE t1 ref a a 3 const 10 Using where; Using filesort +select * from t1 where a = 'aAa' order by p; +p a +1 aAa +4 aAa +select * from t1 where a = 'aaa' order by p; +p a +2 aaa +5 aaa +select * from t1 where a = 'AaA' order by p; +p a +select * from t1 where a = 'AAA' order by p; +p a +3 AAA +6 AAA +drop table t1; +create table t1 ( +p int primary key, +a char(3) character set latin1 collate latin1_swedish_ci not null, +index(a) +) engine=ndb; +insert into t1 values(1, 'aAa'); +insert into t1 values(2, 'aaa'); +insert into t1 values(3, 'AAA'); +insert into t1 values(4, 'aAa'); +insert into t1 values(5, 'aaa'); +insert into t1 values(6, 'AAA'); +select * from t1 order by p; +p a +1 aAa +2 aaa +3 AAA +4 aAa +5 aaa +6 AAA +explain select * from t1 where a = 'zZz' order by p; +id select_type table type possible_keys key key_len ref rows Extra +1 SIMPLE t1 ref a a 3 const 10 Using where; Using filesort +select * from t1 where a = 'aAa' order by p; +p a +1 aAa +2 aaa +3 AAA +4 aAa +5 aaa +6 AAA +select * from t1 where a = 'aaa' order by p; +p a +1 aAa +2 aaa +3 AAA +4 aAa +5 aaa +6 AAA +select * from t1 where a = 'AaA' order by p; +p a +1 aAa +2 aaa +3 AAA +4 aAa +5 aaa +6 AAA +select * from t1 where a = 'AAA' order by p; +p a +1 aAa +2 aaa +3 AAA +4 aAa +5 aaa +6 AAA +drop table t1; diff --git a/mysql-test/r/ndb_index.result b/mysql-test/r/ndb_index.result index dd92c237ace..5702552b0b5 100644 --- a/mysql-test/r/ndb_index.result +++ b/mysql-test/r/ndb_index.result @@ -4,7 +4,7 @@ PORT varchar(16) NOT NULL, ACCESSNODE varchar(16) NOT NULL, POP varchar(48) NOT NULL, ACCESSTYPE int unsigned NOT NULL, -CUSTOMER_ID varchar(20) NOT NULL, +CUSTOMER_ID varchar(20) collate latin1_bin NOT NULL, PROVIDER varchar(16), TEXPIRE int unsigned, NUM_IP int unsigned, diff --git a/mysql-test/t/ndb_charset.test b/mysql-test/t/ndb_charset.test new file mode 100644 index 00000000000..b9f28ed0faf --- /dev/null +++ b/mysql-test/t/ndb_charset.test @@ -0,0 +1,159 @@ +--source include/have_ndb.inc + +--disable_warnings +drop table if exists t1; +--enable_warnings + +# +# Minimal NDB charset test. +# + +# pk - binary + +create table t1 ( + a char(3) character set latin1 collate latin1_bin primary key +) engine=ndb; +# ok +insert into t1 values('aAa'); +insert into t1 values('aaa'); +insert into t1 values('AAA'); +# 3 +select * from t1 order by a; +# 1 +select * from t1 where a = 'aAa'; +# 1 +select * from t1 where a = 'aaa'; +# 0 +select * from t1 where a = 'AaA'; +# 1 +select * from t1 where a = 'AAA'; +drop table t1; + +# pk - case insensitive + +create table t1 ( + a char(3) character set latin1 collate latin1_swedish_ci primary key +) engine=ndb; +# ok +insert into t1 values('aAa'); +# fail +--error 1062 +insert into t1 values('aaa'); +--error 1062 +insert into t1 values('AAA'); +# 1 +select * from t1 order by a; +# 1 +select * from t1 where a = 'aAa'; +# 1 +select * from t1 where a = 'aaa'; +# 1 +select * from t1 where a = 'AaA'; +# 1 +select * from t1 where a = 'AAA'; +drop table t1; + +# unique hash index - binary + +create table t1 ( + p int primary key, + a char(3) character set latin1 collate latin1_bin not null, + unique key(a) +) engine=ndb; +# ok +insert into t1 values(1, 'aAa'); +insert into t1 values(2, 'aaa'); +insert into t1 values(3, 'AAA'); +# 3 +select * from t1 order by p; +# 1 +select * from t1 where a = 'aAa'; +# 1 +select * from t1 where a = 'aaa'; +# 0 +select * from t1 where a = 'AaA'; +# 1 +select * from t1 where a = 'AAA'; +drop table t1; + +# unique hash index - case insensitive + +create table t1 ( + p int primary key, + a char(3) character set latin1 collate latin1_swedish_ci not null, + unique key(a) +) engine=ndb; +# ok +insert into t1 values(1, 'aAa'); +# fail +--error 1169 +insert into t1 values(2, 'aaa'); +--error 1169 +insert into t1 values(3, 'AAA'); +# 1 +select * from t1 order by p; +# 1 +select * from t1 where a = 'aAa'; +# 1 +select * from t1 where a = 'aaa'; +# 1 +select * from t1 where a = 'AaA'; +# 1 +select * from t1 where a = 'AAA'; +drop table t1; + +# ordered index - binary + +create table t1 ( + p int primary key, + a char(3) character set latin1 collate latin1_bin not null, + index(a) +) engine=ndb; +# ok +insert into t1 values(1, 'aAa'); +insert into t1 values(2, 'aaa'); +insert into t1 values(3, 'AAA'); +insert into t1 values(4, 'aAa'); +insert into t1 values(5, 'aaa'); +insert into t1 values(6, 'AAA'); +# 6 +select * from t1 order by p; +# plan +explain select * from t1 where a = 'zZz' order by p; +# 2 +select * from t1 where a = 'aAa' order by p; +# 2 +select * from t1 where a = 'aaa' order by p; +# 0 +select * from t1 where a = 'AaA' order by p; +# 2 +select * from t1 where a = 'AAA' order by p; +drop table t1; + +# ordered index - case insensitive + +create table t1 ( + p int primary key, + a char(3) character set latin1 collate latin1_swedish_ci not null, + index(a) +) engine=ndb; +# ok +insert into t1 values(1, 'aAa'); +insert into t1 values(2, 'aaa'); +insert into t1 values(3, 'AAA'); +insert into t1 values(4, 'aAa'); +insert into t1 values(5, 'aaa'); +insert into t1 values(6, 'AAA'); +# 6 +select * from t1 order by p; +# plan +explain select * from t1 where a = 'zZz' order by p; +# 6 +select * from t1 where a = 'aAa' order by p; +# 6 +select * from t1 where a = 'aaa' order by p; +# 6 +select * from t1 where a = 'AaA' order by p; +# 6 +select * from t1 where a = 'AAA' order by p; +drop table t1; diff --git a/mysql-test/t/ndb_index.test b/mysql-test/t/ndb_index.test index d3977dc3ea4..e65b24a9b20 100644 --- a/mysql-test/t/ndb_index.test +++ b/mysql-test/t/ndb_index.test @@ -9,7 +9,7 @@ CREATE TABLE t1 ( ACCESSNODE varchar(16) NOT NULL, POP varchar(48) NOT NULL, ACCESSTYPE int unsigned NOT NULL, - CUSTOMER_ID varchar(20) NOT NULL, + CUSTOMER_ID varchar(20) collate latin1_bin NOT NULL, PROVIDER varchar(16), TEXPIRE int unsigned, NUM_IP int unsigned, diff --git a/ndb/include/util/NdbSqlUtil.hpp b/ndb/include/util/NdbSqlUtil.hpp index 00216057d58..3062d1e4e1b 100644 --- a/ndb/include/util/NdbSqlUtil.hpp +++ b/ndb/include/util/NdbSqlUtil.hpp @@ -40,11 +40,14 @@ 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 is also pointer to type specific extra info. Char types + * receive CHARSET_INFO in it. + * * 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); + typedef int Cmp(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size); enum CmpResult { CmpLess = -1, @@ -55,6 +58,7 @@ public: /** * Kernel data types. Must match m_typeList in NdbSqlUtil.cpp. + * Now also must match types in NdbDictionary. */ struct Type { enum Enum { diff --git a/ndb/src/common/util/NdbSqlUtil.cpp b/ndb/src/common/util/NdbSqlUtil.cpp index ffcf29a7242..6e4e5919e43 100644 --- a/ndb/src/common/util/NdbSqlUtil.cpp +++ b/ndb/src/common/util/NdbSqlUtil.cpp @@ -198,7 +198,7 @@ NdbSqlUtil::getTypeBinary(Uint32 typeId) // compare int -NdbSqlUtil::cmpTinyint(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpTinyint(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); union { Uint32 p[1]; Int8 v; } u1, u2; @@ -212,7 +212,7 @@ NdbSqlUtil::cmpTinyint(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 s } int -NdbSqlUtil::cmpTinyunsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpTinyunsigned(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); union { Uint32 p[1]; Uint8 v; } u1, u2; @@ -226,7 +226,7 @@ NdbSqlUtil::cmpTinyunsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uin } int -NdbSqlUtil::cmpSmallint(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpSmallint(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); union { Uint32 p[1]; Int16 v; } u1, u2; @@ -240,7 +240,7 @@ NdbSqlUtil::cmpSmallint(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 } int -NdbSqlUtil::cmpSmallunsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpSmallunsigned(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); union { Uint32 p[1]; Uint16 v; } u1, u2; @@ -254,7 +254,7 @@ NdbSqlUtil::cmpSmallunsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Ui } int -NdbSqlUtil::cmpMediumint(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpMediumint(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); union { const Uint32* p; const unsigned char* v; } u1, u2; @@ -270,7 +270,7 @@ NdbSqlUtil::cmpMediumint(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 } int -NdbSqlUtil::cmpMediumunsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpMediumunsigned(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); union { const Uint32* p; const unsigned char* v; } u1, u2; @@ -286,7 +286,7 @@ NdbSqlUtil::cmpMediumunsigned(const Uint32* p1, const Uint32* p2, Uint32 full, U } int -NdbSqlUtil::cmpInt(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpInt(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); union { Uint32 p[1]; Int32 v; } u1, u2; @@ -300,7 +300,7 @@ NdbSqlUtil::cmpInt(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) } int -NdbSqlUtil::cmpUnsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpUnsigned(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); union { Uint32 p[1]; Uint32 v; } u1, u2; @@ -314,7 +314,7 @@ NdbSqlUtil::cmpUnsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 } int -NdbSqlUtil::cmpBigint(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpBigint(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); if (size >= 2) { @@ -333,7 +333,7 @@ NdbSqlUtil::cmpBigint(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 si } int -NdbSqlUtil::cmpBigunsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpBigunsigned(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); if (size >= 2) { @@ -352,7 +352,7 @@ NdbSqlUtil::cmpBigunsigned(const Uint32* p1, const Uint32* p2, Uint32 full, Uint } int -NdbSqlUtil::cmpFloat(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpFloat(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); union { Uint32 p[1]; float v; } u1, u2; @@ -367,7 +367,7 @@ NdbSqlUtil::cmpFloat(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 siz } int -NdbSqlUtil::cmpDouble(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpDouble(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); if (size >= 2) { @@ -387,7 +387,7 @@ NdbSqlUtil::cmpDouble(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 si } int -NdbSqlUtil::cmpDecimal(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpDecimal(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); // not used by MySQL or NDB @@ -396,27 +396,34 @@ NdbSqlUtil::cmpDecimal(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 s } int -NdbSqlUtil::cmpChar(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpChar(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { - assert(full >= size && size > 0); + // collation does not work on prefix for some charsets + 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. + * Char is blank-padded to length and null-padded to word size. */ - union { const Uint32* p; const char* v; } u1, u2; + union { const Uint32* p; const uchar* 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; + // not const in MySQL + CHARSET_INFO* cs = (CHARSET_INFO*)(info); + // length in bytes including null padding to Uint32 + uint l1 = (full << 2); + int k = (*cs->coll->strnncollsp)(cs, u1.v, l1, u2.v, l1); + return k < 0 ? -1 : k > 0 ? +1 : 0; } int -NdbSqlUtil::cmpVarchar(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpVarchar(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 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. + * + * Not used before MySQL 5.0. Format is likely to change. Handle + * only binary collation for now. */ union { const Uint32* p; const char* v; } u1, u2; u1.p = p1; @@ -427,7 +434,7 @@ NdbSqlUtil::cmpVarchar(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 s } int -NdbSqlUtil::cmpBinary(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpBinary(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); /* @@ -441,12 +448,14 @@ NdbSqlUtil::cmpBinary(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 si } int -NdbSqlUtil::cmpVarbinary(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpVarbinary(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); /* * Binary data of variable length padded with nulls. The comparison * does not need to use the length. + * + * Not used before MySQL 5.0. Format is likely to change. */ union { const Uint32* p; const unsigned char* v; } u1, u2; u1.p = p1; @@ -457,11 +466,13 @@ NdbSqlUtil::cmpVarbinary(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 } int -NdbSqlUtil::cmpDatetime(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpDatetime(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); /* * Datetime is CC YY MM DD hh mm ss \0 + * + * Not used via MySQL. */ union { const Uint32* p; const unsigned char* v; } u1, u2; u1.p = p1; @@ -478,11 +489,13 @@ NdbSqlUtil::cmpDatetime(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 } int -NdbSqlUtil::cmpTimespec(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpTimespec(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); /* * Timespec is CC YY MM DD hh mm ss \0 NN NN NN NN + * + * Not used via MySQL. */ union { const Uint32* p; const unsigned char* v; } u1, u2; u1.p = p1; @@ -509,12 +522,11 @@ NdbSqlUtil::cmpTimespec(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 } int -NdbSqlUtil::cmpBlob(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpBlob(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { assert(full >= size && size > 0); /* - * Blob comparison is on the inline bytes. Except for larger header - * the format is like Varbinary. + * Blob comparison is on the inline bytes (null padded). */ const unsigned head = NDB_BLOB_HEAD_SIZE; // skip blob head @@ -529,21 +541,26 @@ NdbSqlUtil::cmpBlob(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size } int -NdbSqlUtil::cmpText(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) +NdbSqlUtil::cmpText(const void* info, const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size) { - assert(full >= size && size > 0); + // collation does not work on prefix for some charsets + assert(full == size && size > 0); /* - * Text comparison is on the inline bytes. Except for larger header - * the format is like Varchar. + * Text comparison is on the inline bytes (blank padded). Currently + * not supported for multi-byte charsets. */ const unsigned head = NDB_BLOB_HEAD_SIZE; // skip blob head if (size >= head + 1) { - union { const Uint32* p; const char* v; } u1, u2; + union { const Uint32* p; const uchar* 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; + // not const in MySQL + CHARSET_INFO* cs = (CHARSET_INFO*)(info); + // length in bytes including null padding to Uint32 + uint l1 = (full << 2); + int k = (*cs->coll->strnncollsp)(cs, u1.v, l1, u2.v, l1); + return k < 0 ? -1 : k > 0 ? +1 : 0; } return CmpUnknown; } @@ -652,6 +669,7 @@ const Testcase testcase[] = { int main(int argc, char** argv) { + ndb_init(); // for charsets unsigned count = argc > 1 ? atoi(argv[1]) : 1000000; ndbout_c("count = %u", count); assert(count != 0); diff --git a/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp b/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp index 467df4aca3d..8342870d69c 100644 --- a/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp +++ b/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp @@ -7700,6 +7700,7 @@ void Dblqh::accScanConfScanLab(Signal* signal) ndbrequire(sz == boundAiLength); EXECUTE_DIRECT(DBTUX, GSN_TUX_BOUND_INFO, signal, TuxBoundInfo::SignalLength + boundAiLength); + jamEntry(); if (req->errorCode != 0) { jam(); /* diff --git a/ndb/src/kernel/blocks/dbtup/Dbtup.hpp b/ndb/src/kernel/blocks/dbtup/Dbtup.hpp index 1444066a76c..ce81c1c9bc8 100644 --- a/ndb/src/kernel/blocks/dbtup/Dbtup.hpp +++ b/ndb/src/kernel/blocks/dbtup/Dbtup.hpp @@ -1374,7 +1374,8 @@ private: const Uint32* inBuffer, Uint32 inBufLen, Uint32* outBuffer, - Uint32 TmaxRead); + Uint32 TmaxRead, + bool xfrmFlag); //------------------------------------------------------------------ //------------------------------------------------------------------ @@ -1620,6 +1621,20 @@ private: Uint32 attrDescriptor, Uint32 attrDes2); +// ***************************************************************** +// Read char routines optionally (tXfrmFlag) apply strxfrm +// ***************************************************************** + + bool readCharNotNULL(Uint32* outBuffer, + AttributeHeader* ahOut, + Uint32 attrDescriptor, + Uint32 attrDes2); + + bool readCharNULLable(Uint32* outBuffer, + AttributeHeader* ahOut, + Uint32 attrDescriptor, + Uint32 attrDes2); + //------------------------------------------------------------------ //------------------------------------------------------------------ bool nullFlagCheck(Uint32 attrDes2); @@ -2225,6 +2240,7 @@ private: Uint32 tMaxRead; Uint32 tOutBufIndex; Uint32* tTupleHeader; + bool tXfrmFlag; // updateAttributes module Uint32 tInBufIndex; diff --git a/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp b/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp index 0a47778f7c1..dfd1e37d4f5 100644 --- a/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp +++ b/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp @@ -903,7 +903,8 @@ int Dbtup::handleReadReq(Signal* signal, &cinBuffer[0], regOperPtr->attrinbufLen, dst, - dstLen); + dstLen, + false); if (TnoOfDataRead != (Uint32)-1) { /* ------------------------------------------------------------------------- */ // We have read all data into coutBuffer. Now send it to the API. @@ -1274,7 +1275,8 @@ int Dbtup::interpreterStartLab(Signal* signal, &cinBuffer[5], RinitReadLen, &dst[0], - dstLen); + dstLen, + false); if (TnoDataRW != (Uint32)-1) { RattroutCounter = TnoDataRW; RinstructionCounter += RinitReadLen; @@ -1347,7 +1349,8 @@ int Dbtup::interpreterStartLab(Signal* signal, &cinBuffer[RinstructionCounter], RfinalRLen, &dst[RattroutCounter], - (dstLen - RattroutCounter)); + (dstLen - RattroutCounter), + false); if (TnoDataRW != (Uint32)-1) { RattroutCounter += TnoDataRW; } else { @@ -1487,7 +1490,8 @@ int Dbtup::interpreterNextLab(Signal* signal, &theAttrinfo, (Uint32)1, &TregMemBuffer[theRegister], - (Uint32)3); + (Uint32)3, + false); if (TnoDataRW == 2) { /* ------------------------------------------------------------- */ // Two words read means that we get the instruction plus one 32 @@ -1833,7 +1837,8 @@ int Dbtup::interpreterNextLab(Signal* signal, Int32 TnoDataR = readAttributes(pagePtr, TupHeadOffset, &attrId, 1, - tmpArea, tmpAreaSz); + tmpArea, tmpAreaSz, + false); if (TnoDataR == -1) { jam(); @@ -1929,7 +1934,8 @@ int Dbtup::interpreterNextLab(Signal* signal, Int32 TnoDataR = readAttributes(pagePtr, TupHeadOffset, &attrId, 1, - tmpArea, tmpAreaSz); + tmpArea, tmpAreaSz, + false); if (TnoDataR == -1) { jam(); @@ -1957,7 +1963,8 @@ int Dbtup::interpreterNextLab(Signal* signal, Int32 TnoDataR = readAttributes(pagePtr, TupHeadOffset, &attrId, 1, - tmpArea, tmpAreaSz); + tmpArea, tmpAreaSz, + false); if (TnoDataR == -1) { jam(); diff --git a/ndb/src/kernel/blocks/dbtup/DbtupIndex.cpp b/ndb/src/kernel/blocks/dbtup/DbtupIndex.cpp index d864bac8b59..2dd707ebafc 100644 --- a/ndb/src/kernel/blocks/dbtup/DbtupIndex.cpp +++ b/ndb/src/kernel/blocks/dbtup/DbtupIndex.cpp @@ -160,7 +160,7 @@ Dbtup::tuxReadAttrs(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tu operPtr.i = RNIL; operPtr.p = NULL; // do it - int ret = readAttributes(pagePtr.p, pageOffset, attrIds, numAttrs, dataOut, ZNIL); + int ret = readAttributes(pagePtr.p, pageOffset, attrIds, numAttrs, dataOut, ZNIL, true); // restore globals tabptr = tabptr_old; fragptr = fragptr_old; @@ -200,7 +200,7 @@ Dbtup::tuxReadPk(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32* data operPtr.i = RNIL; operPtr.p = NULL; // do it - int ret = readAttributes(pagePtr.p, pageOffset, attrIds, numAttrs, dataOut, ZNIL); + int ret = readAttributes(pagePtr.p, pageOffset, attrIds, numAttrs, dataOut, ZNIL, true); // restore globals tabptr = tabptr_old; fragptr = fragptr_old; diff --git a/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp b/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp index dc04650cd1b..efea312b865 100644 --- a/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp +++ b/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp @@ -332,11 +332,11 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal) } if (i == fragOperPtr.p->charsetIndex) { ljam(); - ndbrequire(i < regTabPtr.p->noOfCharsets); - regTabPtr.p->charsetArray[i] = cs; - AttributeOffset::setCharsetPos(attrDes2, i); fragOperPtr.p->charsetIndex++; } + ndbrequire(i < regTabPtr.p->noOfCharsets); + regTabPtr.p->charsetArray[i] = cs; + AttributeOffset::setCharsetPos(attrDes2, i); } setTabDescrWord(firstTabDesIndex + 1, attrDes2); diff --git a/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp b/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp index 49ca52b2b9c..a4e7cb47249 100644 --- a/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp +++ b/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp @@ -35,6 +35,7 @@ Dbtup::setUpQueryRoutines(Tablerec* const regTabPtr) for (Uint32 i = 0; i < regTabPtr->noOfAttr; i++) { Uint32 attrDescriptorStart = startDescriptor + (i << ZAD_LOG_SIZE); Uint32 attrDescriptor = tableDescriptor[attrDescriptorStart].tabDescr; + Uint32 attrOffset = tableDescriptor[attrDescriptorStart + 1].tabDescr; if (!AttributeDescriptor::getDynamic(attrDescriptor)) { if ((AttributeDescriptor::getArrayType(attrDescriptor) == ZNON_ARRAY) || (AttributeDescriptor::getArrayType(attrDescriptor) == ZFIXED_ARRAY)) { @@ -54,6 +55,11 @@ Dbtup::setUpQueryRoutines(Tablerec* const regTabPtr) } else { ndbrequire(false); }//if + // replace read function of char attribute + if (AttributeOffset::getCharsetFlag(attrOffset)) { + ljam(); + regTabPtr->readFunctionArray[i] = &Dbtup::readCharNotNULL; + } } else { if (AttributeDescriptor::getSizeInWords(attrDescriptor) == 1) { ljam(); @@ -72,6 +78,11 @@ Dbtup::setUpQueryRoutines(Tablerec* const regTabPtr) regTabPtr->readFunctionArray[i] = &Dbtup::readFixedSizeTHZeroWordNULLable; regTabPtr->updateFunctionArray[i] = &Dbtup::updateFixedSizeTHManyWordNULLable; }//if + // replace read function of char attribute + if (AttributeOffset::getCharsetFlag(attrOffset)) { + ljam(); + regTabPtr->readFunctionArray[i] = &Dbtup::readCharNULLable; + } }//if } else if (AttributeDescriptor::getArrayType(attrDescriptor) == ZVAR_ARRAY) { if (!AttributeDescriptor::getNullable(attrDescriptor)) { @@ -149,7 +160,8 @@ int Dbtup::readAttributes(Page* const pagePtr, const Uint32* inBuffer, Uint32 inBufLen, Uint32* outBuffer, - Uint32 maxRead) + Uint32 maxRead, + bool xfrmFlag) { Tablerec* const regTabPtr = tabptr.p; Uint32 numAttributes = regTabPtr->noOfAttr; @@ -162,6 +174,7 @@ int Dbtup::readAttributes(Page* const pagePtr, tCheckOffset = regTabPtr->tupheadsize; tMaxRead = maxRead; tTupleHeader = &pagePtr->pageWord[tupHeadOffset]; + tXfrmFlag = xfrmFlag; ndbrequire(tupHeadOffset + tCheckOffset <= ZWORDS_ON_PAGE); while (inBufIndex < inBufLen) { @@ -542,6 +555,74 @@ Dbtup::readDynSmallVarSize(Uint32* outBuffer, return false; }//Dbtup::readDynSmallVarSize() + +bool +Dbtup::readCharNotNULL(Uint32* outBuffer, + AttributeHeader* ahOut, + Uint32 attrDescriptor, + Uint32 attrDes2) +{ + Uint32 indexBuf = tOutBufIndex; + Uint32 readOffset = AttributeOffset::getOffset(attrDes2); + Uint32 attrNoOfWords = AttributeDescriptor::getSizeInWords(attrDescriptor); + Uint32 newIndexBuf = indexBuf + attrNoOfWords; + Uint32 maxRead = tMaxRead; + + ndbrequire((readOffset + attrNoOfWords - 1) < tCheckOffset); + if (newIndexBuf <= maxRead) { + ljam(); + ahOut->setDataSize(attrNoOfWords); + if (! tXfrmFlag) { + MEMCOPY_NO_WORDS(&outBuffer[indexBuf], + &tTupleHeader[readOffset], + attrNoOfWords); + } else { + ljam(); + Tablerec* regTabPtr = tabptr.p; + Uint32 i = AttributeOffset::getCharsetPos(attrDes2); + ndbrequire(i < tabptr.p->noOfCharsets); + // not const in MySQL + CHARSET_INFO* cs = tabptr.p->charsetArray[i]; + // XXX should strip Uint32 null padding + const unsigned nBytes = attrNoOfWords << 2; + unsigned n = + (*cs->coll->strnxfrm)(cs, + (uchar*)&outBuffer[indexBuf], + nBytes, + (const uchar*)&tTupleHeader[readOffset], + nBytes); + // pad with ascii spaces + while (n < nBytes) + ((uchar*)&outBuffer[indexBuf])[n++] = 0x20; + } + tOutBufIndex = newIndexBuf; + return true; + } else { + ljam(); + terrorCode = ZTRY_TO_READ_TOO_MUCH_ERROR; + return false; + } +} + +bool +Dbtup::readCharNULLable(Uint32* outBuffer, + AttributeHeader* ahOut, + Uint32 attrDescriptor, + Uint32 attrDes2) +{ + if (!nullFlagCheck(attrDes2)) { + ljam(); + return readCharNotNULL(outBuffer, + ahOut, + attrDescriptor, + attrDes2); + } else { + ljam(); + ahOut->setNULL(); + return true; + } +} + /* ---------------------------------------------------------------------- */ /* THIS ROUTINE IS USED TO UPDATE A NUMBER OF ATTRIBUTES. IT IS */ /* USED BY THE INSERT ROUTINE, THE UPDATE ROUTINE AND IT CAN BE */ diff --git a/ndb/src/kernel/blocks/dbtup/DbtupTrigger.cpp b/ndb/src/kernel/blocks/dbtup/DbtupTrigger.cpp index a93ff4566e7..c0b49364ee6 100644 --- a/ndb/src/kernel/blocks/dbtup/DbtupTrigger.cpp +++ b/ndb/src/kernel/blocks/dbtup/DbtupTrigger.cpp @@ -751,7 +751,8 @@ bool Dbtup::readTriggerInfo(TupTriggerData* const trigPtr, &tableDescriptor[regTabPtr->readKeyArray].tabDescr, regTabPtr->noOfKeyAttr, keyBuffer, - ZATTR_BUFFER_SIZE); + ZATTR_BUFFER_SIZE, + true); ndbrequire(noPrimKey != (Uint32)-1); Uint32 numAttrsToRead; @@ -792,7 +793,8 @@ bool Dbtup::readTriggerInfo(TupTriggerData* const trigPtr, &readBuffer[0], numAttrsToRead, mainBuffer, - ZATTR_BUFFER_SIZE); + ZATTR_BUFFER_SIZE, + true); ndbrequire(noMainWords != (Uint32)-1); } else { ljam(); @@ -816,7 +818,8 @@ bool Dbtup::readTriggerInfo(TupTriggerData* const trigPtr, &readBuffer[0], numAttrsToRead, copyBuffer, - ZATTR_BUFFER_SIZE); + ZATTR_BUFFER_SIZE, + true); ndbrequire(noCopyWords != (Uint32)-1); if ((noMainWords == noCopyWords) && diff --git a/ndb/src/kernel/blocks/dbtux/DbtuxCmp.cpp b/ndb/src/kernel/blocks/dbtux/DbtuxCmp.cpp index 4ce413af138..549720cc17c 100644 --- a/ndb/src/kernel/blocks/dbtux/DbtuxCmp.cpp +++ b/ndb/src/kernel/blocks/dbtux/DbtuxCmp.cpp @@ -58,7 +58,7 @@ Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, ConstData searchKey, Cons NdbSqlUtil::Cmp* const cmp = c_sqlCmp[start]; const Uint32* const p1 = &searchKey[AttributeHeaderSize]; const Uint32* const p2 = &entryData[AttributeHeaderSize]; - ret = (*cmp)(p1, p2, size1, size2); + ret = (*cmp)(0, p1, p2, size1, size2); if (ret != 0) { jam(); break; @@ -132,6 +132,7 @@ Dbtux::cmpScanBound(const Frag& frag, unsigned dir, ConstData boundInfo, unsigne jam(); // current attribute const unsigned index = boundInfo.ah().getAttributeId(); + ndbrequire(index < frag.m_numAttrs); const DescAttr& descAttr = descEnt.m_descAttr[index]; ndbrequire(entryData.ah().getAttributeId() == descAttr.m_primaryAttrId); // full data size @@ -143,7 +144,7 @@ Dbtux::cmpScanBound(const Frag& frag, unsigned dir, ConstData boundInfo, unsigne NdbSqlUtil::Cmp* const cmp = c_sqlCmp[index]; const Uint32* const p1 = &boundInfo[AttributeHeaderSize]; const Uint32* const p2 = &entryData[AttributeHeaderSize]; - int ret = (*cmp)(p1, p2, size1, size2); + int ret = (*cmp)(0, p1, p2, size1, size2); if (ret != 0) { jam(); return ret; diff --git a/ndb/src/kernel/blocks/dbtux/DbtuxGen.cpp b/ndb/src/kernel/blocks/dbtux/DbtuxGen.cpp index 958ba4b0839..39cd8e25184 100644 --- a/ndb/src/kernel/blocks/dbtux/DbtuxGen.cpp +++ b/ndb/src/kernel/blocks/dbtux/DbtuxGen.cpp @@ -201,8 +201,8 @@ Dbtux::execREAD_CONFIG_REQ(Signal* signal) // allocate buffers c_keyAttrs = (Uint32*)allocRecord("c_keyAttrs", sizeof(Uint32), MaxIndexAttributes); c_sqlCmp = (NdbSqlUtil::Cmp**)allocRecord("c_sqlCmp", sizeof(NdbSqlUtil::Cmp*), MaxIndexAttributes); - c_searchKey = (Uint32*)allocRecord("c_searchKey", sizeof(Uint32*), MaxIndexAttributes); - c_entryKey = (Uint32*)allocRecord("c_entryKey", sizeof(Uint32*), MaxIndexAttributes); + c_searchKey = (Uint32*)allocRecord("c_searchKey", sizeof(Uint32), MaxAttrDataSize); + c_entryKey = (Uint32*)allocRecord("c_entryKey", sizeof(Uint32), MaxAttrDataSize); c_dataBuffer = (Uint32*)allocRecord("c_dataBuffer", sizeof(Uint64), (MaxAttrDataSize + 1) >> 1); // ack ReadConfigConf * conf = (ReadConfigConf*)signal->getDataPtrSend(); diff --git a/ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp b/ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp index 9d7d4b06bf7..5b161d3c4ce 100644 --- a/ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp +++ b/ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp @@ -112,6 +112,7 @@ Dbtux::execACC_SCANREQ(Signal* signal) void Dbtux::execTUX_BOUND_INFO(Signal* signal) { + jamEntry(); struct BoundInfo { unsigned offset; unsigned size; diff --git a/ndb/src/kernel/blocks/dbtux/Times.txt b/ndb/src/kernel/blocks/dbtux/Times.txt index 84819ddcf97..03473353a52 100644 --- a/ndb/src/kernel/blocks/dbtux/Times.txt +++ b/ndb/src/kernel/blocks/dbtux/Times.txt @@ -83,7 +83,7 @@ optim 13 mc02/a 39 ms 59 ms 50 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? ] +[ case d: bug in testOIBasic killed PK read performance ] optim 14 mc02/a 41 ms 60 ms 44 pct mc02/b 46 ms 81 ms 73 pct @@ -91,5 +91,21 @@ optim 14 mc02/a 41 ms 60 ms 44 pct mc02/d 242 ms 285 ms 17 pct [ case b: do long keys suffer from many subroutine calls? ] +[ case d: bug in testOIBasic killed PK read performance ] + +none mc02/a 35 ms 60 ms 71 pct + mc02/b 42 ms 75 ms 76 pct + mc02/c 5 ms 12 ms 106 pct + mc02/d 165 ms 238 ms 44 pct + +[ johan re-installed mc02 as fedora gcc-3.3.2 ] +[ case c: table scan has improved... ] + +charsets mc02/a 35 ms 60 ms 71 pct + mc02/b 42 ms 84 ms 97 pct + mc02/c 5 ms 12 ms 109 pct + mc02/d 190 ms 236 ms 23 pct + +[ case b: TUX can no longer use pointers to TUP data ] vim: set et: diff --git a/ndb/src/ndbapi/NdbIndexOperation.cpp b/ndb/src/ndbapi/NdbIndexOperation.cpp index 0742f8d911c..bf4b07842f6 100644 --- a/ndb/src/ndbapi/NdbIndexOperation.cpp +++ b/ndb/src/ndbapi/NdbIndexOperation.cpp @@ -164,6 +164,7 @@ int NdbIndexOperation::equal_impl(const NdbColumnImpl* tAttrInfo, Uint32 tData; Uint32 tKeyInfoPosition; const char* aValue = aValuePassed; + Uint32 xfrmData[1024]; Uint32 tempData[1024]; if ((theStatus == OperationDefined) && @@ -224,6 +225,21 @@ int NdbIndexOperation::equal_impl(const NdbColumnImpl* tAttrInfo, m_theIndexDefined[i][2] = true; Uint32 sizeInBytes = tAttrInfo->m_attrSize * tAttrInfo->m_arraySize; + const char* aValueToWrite = aValue; + + CHARSET_INFO* cs = tAttrInfo->m_cs; + if (cs != 0) { + // current limitation: strxfrm does not increase length + assert(cs->strxfrm_multiply == 1); + unsigned n = + (*cs->coll->strnxfrm)(cs, + (uchar*)xfrmData, sizeof(xfrmData), + (const uchar*)aValue, sizeInBytes); + while (n < sizeInBytes) + ((uchar*)xfrmData)[n++] = 0x20; + aValue = (char*)xfrmData; + } + Uint32 bitsInLastWord = 8 * (sizeInBytes & 3) ; Uint32 totalSizeInWords = (sizeInBytes + 3)/4;// Inc. bits in last word Uint32 sizeInWords = sizeInBytes / 4; // Exc. bits in last word @@ -314,13 +330,20 @@ int NdbIndexOperation::equal_impl(const NdbColumnImpl* tAttrInfo, if ((tOpType == InsertRequest) || (tOpType == WriteRequest)) { if (!tAttrInfo->m_indexOnly){ + // invalid data can crash kernel + if (cs != NULL && + (*cs->cset->well_formed_len)(cs, + aValueToWrite, + aValueToWrite + sizeInBytes, + sizeInBytes) != sizeInBytes) + goto equal_error4; Uint32 ahValue; Uint32 sz = totalSizeInWords; AttributeHeader::init(&ahValue, tAttrId, sz); insertATTRINFO( ahValue ); - insertATTRINFOloop((Uint32*)aValue, sizeInWords); + insertATTRINFOloop((Uint32*)aValueToWrite, sizeInWords); if (bitsInLastWord != 0) { - tData = *(Uint32*)(aValue + (sizeInWords << 2)); + tData = *(Uint32*)(aValueToWrite + (sizeInWords << 2)); tData = convertEndian(tData); tData = tData & ((1 << bitsInLastWord) - 1); tData = convertEndian(tData); @@ -411,7 +434,10 @@ int NdbIndexOperation::equal_impl(const NdbColumnImpl* tAttrInfo, equal_error3: setErrorCodeAbort(4209); - + return -1; + + equal_error4: + setErrorCodeAbort(744); return -1; } diff --git a/ndb/src/ndbapi/NdbOperationDefine.cpp b/ndb/src/ndbapi/NdbOperationDefine.cpp index 6d995e06582..ad838ddd601 100644 --- a/ndb/src/ndbapi/NdbOperationDefine.cpp +++ b/ndb/src/ndbapi/NdbOperationDefine.cpp @@ -492,6 +492,17 @@ NdbOperation::setValue( const NdbColumnImpl* tAttrInfo, // Insert Attribute Id into ATTRINFO part. const Uint32 sizeInBytes = tAttrInfo->m_attrSize * tAttrInfo->m_arraySize; + + CHARSET_INFO* cs = tAttrInfo->m_cs; + // invalid data can crash kernel + if (cs != NULL && + (*cs->cset->well_formed_len)(cs, + aValue, + aValue + sizeInBytes, + sizeInBytes) != sizeInBytes) { + setErrorCodeAbort(744); + return -1; + } #if 0 tAttrSize = tAttrInfo->theAttrSize; tArraySize = tAttrInfo->theArraySize; diff --git a/ndb/src/ndbapi/NdbOperationSearch.cpp b/ndb/src/ndbapi/NdbOperationSearch.cpp index 19cb133dbf7..e5166fc4a82 100644 --- a/ndb/src/ndbapi/NdbOperationSearch.cpp +++ b/ndb/src/ndbapi/NdbOperationSearch.cpp @@ -60,6 +60,7 @@ NdbOperation::equal_impl(const NdbColumnImpl* tAttrInfo, Uint32 tData; Uint32 tKeyInfoPosition; const char* aValue = aValuePassed; + Uint32 xfrmData[1024]; Uint32 tempData[1024]; if ((theStatus == OperationDefined) && @@ -117,6 +118,21 @@ NdbOperation::equal_impl(const NdbColumnImpl* tAttrInfo, theTupleKeyDefined[i][2] = true; Uint32 sizeInBytes = tAttrInfo->m_attrSize * tAttrInfo->m_arraySize; + const char* aValueToWrite = aValue; + + CHARSET_INFO* cs = tAttrInfo->m_cs; + if (cs != 0) { + // current limitation: strxfrm does not increase length + assert(cs->strxfrm_multiply == 1); + unsigned n = + (*cs->coll->strnxfrm)(cs, + (uchar*)xfrmData, sizeof(xfrmData), + (const uchar*)aValue, sizeInBytes); + while (n < sizeInBytes) + ((uchar*)xfrmData)[n++] = 0x20; + aValue = (char*)xfrmData; + } + Uint32 bitsInLastWord = 8 * (sizeInBytes & 3) ; Uint32 totalSizeInWords = (sizeInBytes + 3)/4; // Inc. bits in last word Uint32 sizeInWords = sizeInBytes / 4; // Exc. bits in last word @@ -206,13 +222,20 @@ NdbOperation::equal_impl(const NdbColumnImpl* tAttrInfo, if ((tOpType == InsertRequest) || (tOpType == WriteRequest)) { if (!tAttrInfo->m_indexOnly){ + // invalid data can crash kernel + if (cs != NULL && + (*cs->cset->well_formed_len)(cs, + aValueToWrite, + aValueToWrite + sizeInBytes, + sizeInBytes) != sizeInBytes) + goto equal_error4; Uint32 ahValue; const Uint32 sz = totalSizeInWords; AttributeHeader::init(&ahValue, tAttrId, sz); insertATTRINFO( ahValue ); - insertATTRINFOloop((Uint32*)aValue, sizeInWords); + insertATTRINFOloop((Uint32*)aValueToWrite, sizeInWords); if (bitsInLastWord != 0) { - tData = *(Uint32*)(aValue + (sizeInWords << 2)); + tData = *(Uint32*)(aValueToWrite + (sizeInWords << 2)); tData = convertEndian(tData); tData = tData & ((1 << bitsInLastWord) - 1); tData = convertEndian(tData); @@ -311,6 +334,10 @@ NdbOperation::equal_impl(const NdbColumnImpl* tAttrInfo, equal_error3: setErrorCodeAbort(4209); return -1; + + equal_error4: + setErrorCodeAbort(744); + return -1; } /****************************************************************************** diff --git a/ndb/src/ndbapi/NdbScanOperation.cpp b/ndb/src/ndbapi/NdbScanOperation.cpp index 86c174c4545..ac5f4268386 100644 --- a/ndb/src/ndbapi/NdbScanOperation.cpp +++ b/ndb/src/ndbapi/NdbScanOperation.cpp @@ -1096,30 +1096,43 @@ NdbIndexScanOperation::setBound(const NdbColumnImpl* tAttrInfo, theStatus == SetBound && (0 <= type && type <= 4) && len <= 8000) { - // bound type - + // insert bound type insertATTRINFO(type); - // attribute header Uint32 sizeInBytes = tAttrInfo->m_attrSize * tAttrInfo->m_arraySize; + // normalize char bound + CHARSET_INFO* cs = tAttrInfo->m_cs; + Uint32 xfrmData[2000]; + if (cs != NULL && aValue != NULL) { + // current limitation: strxfrm does not increase length + assert(cs->strxfrm_multiply == 1); + unsigned n = + (*cs->coll->strnxfrm)(cs, + (uchar*)xfrmData, sizeof(xfrmData), + (const uchar*)aValue, sizeInBytes); + while (n < sizeInBytes) + ((uchar*)xfrmData)[n++] = 0x20; + aValue = (char*)xfrmData; + } if (len != sizeInBytes && (len != 0)) { setErrorCodeAbort(4209); return -1; } + // insert attribute header len = aValue != NULL ? sizeInBytes : 0; Uint32 tIndexAttrId = tAttrInfo->m_attrId; Uint32 sizeInWords = (len + 3) / 4; AttributeHeader ah(tIndexAttrId, sizeInWords); insertATTRINFO(ah.m_value); if (len != 0) { - // attribute data + // insert attribute data if ((UintPtr(aValue) & 0x3) == 0 && (len & 0x3) == 0) insertATTRINFOloop((const Uint32*)aValue, sizeInWords); else { - Uint32 temp[2000]; - memcpy(temp, aValue, len); + Uint32 tempData[2000]; + memcpy(tempData, aValue, len); while ((len & 0x3) != 0) - ((char*)temp)[len++] = 0; - insertATTRINFOloop(temp, sizeInWords); + ((char*)tempData)[len++] = 0; + insertATTRINFOloop(tempData, sizeInWords); } } @@ -1206,11 +1219,11 @@ NdbIndexScanOperation::compare(Uint32 skip, Uint32 cols, if((r1_null ^ (unsigned)r2->isNULL())){ return (r1_null ? -1 : 1); } - Uint32 type = NdbColumnImpl::getImpl(* r1->m_column).m_extType; + const NdbColumnImpl & col = NdbColumnImpl::getImpl(* r1->m_column); Uint32 size = (r1->theAttrSize * r1->theArraySize + 3) / 4; if(!r1_null){ - const NdbSqlUtil::Type& t = NdbSqlUtil::getType(type); - int r = (*t.m_cmp)(d1, d2, size, size); + const NdbSqlUtil::Type& sqlType = NdbSqlUtil::getType(col.m_extType); + int r = (*sqlType.m_cmp)(col.m_cs, d1, d2, size, size); if(r){ assert(r != NdbSqlUtil::CmpUnknown); return r; diff --git a/ndb/src/ndbapi/ndberror.c b/ndb/src/ndbapi/ndberror.c index 2ebcf4be444..037c441cc38 100644 --- a/ndb/src/ndbapi/ndberror.c +++ b/ndb/src/ndbapi/ndberror.c @@ -282,7 +282,7 @@ ErrorBundle ErrorCodes[] = { { 741, SE, "Unsupported alter table" }, { 742, SE, "Unsupported attribute type in index" }, { 743, SE, "Unsupported character set in table or index" }, - { 744, SE, "Character conversion error" }, + { 744, SE, "Character string is invalid for given character set" }, { 241, SE, "Invalid schema object version" }, { 283, SE, "Table is being dropped" }, { 284, SE, "Table not defined in transaction coordinator" }, diff --git a/ndb/test/ndbapi/testOIBasic.cpp b/ndb/test/ndbapi/testOIBasic.cpp index ac28b96af80..f9eb3514926 100644 --- a/ndb/test/ndbapi/testOIBasic.cpp +++ b/ndb/test/ndbapi/testOIBasic.cpp @@ -28,6 +28,7 @@ #include #include #include +#include // options @@ -37,6 +38,8 @@ struct Opt { const char* m_bound; const char* m_case; bool m_core; + const char* m_csname; + CHARSET_INFO* m_cs; bool m_dups; NdbDictionary::Object::FragmentType m_fragtype; unsigned m_idxloop; @@ -59,6 +62,8 @@ struct Opt { m_bound("01234"), m_case(0), m_core(false), + m_csname("latin1_bin"), + m_cs(0), m_dups(false), m_fragtype(NdbDictionary::Object::FragUndefined), m_idxloop(4), @@ -94,6 +99,7 @@ printhelp() << " -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 + << " -csname S charset (collation) of non-pk char column [" << d.m_csname << "]" << endl << " -dups allow duplicate tuples from index scan [" << d.m_dups << "]" << endl << " -fragtype T fragment type single/small/medium/large" << endl << " -index xyz only given index numbers (digits 1-9)" << endl @@ -983,6 +989,10 @@ createtable(Par par) c.setLength(col.m_length); c.setPrimaryKey(col.m_pk); c.setNullable(col.m_nullable); + if (c.getCharset()) { // test if char type + if (! col.m_pk) + c.setCharset(par.m_cs); + } t.addColumn(c); } con.m_dic = con.m_ndb->getDictionary(); @@ -3149,6 +3159,10 @@ runtest(Par par) LL1("start"); if (par.m_seed != 0) srandom(par.m_seed); + assert(par.m_csname != 0); + CHARSET_INFO* cs; + CHK((cs = get_charset_by_name(par.m_csname, MYF(0))) != 0 || (cs = get_charset_by_csname(par.m_csname, MY_CS_PRIMARY, MYF(0))) != 0); + par.m_cs = cs; Con con; CHK(con.connect() == 0); par.m_con = &con; @@ -3232,6 +3246,12 @@ NDB_COMMAND(testOIBasic, "testOIBasic", "testOIBasic", "testOIBasic", 65535) g_opt.m_core = true; continue; } + if (strcmp(arg, "-csname") == 0) { + if (++argv, --argc > 0) { + g_opt.m_csname = strdup(argv[0]); + continue; + } + } if (strcmp(arg, "-dups") == 0) { g_opt.m_dups = true; continue;