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

ndb - wl#2624 re-commit due to bk problem

Browse Source
This commit is contained in:
pekka@mysql.com
2005-09-15 02:33:28 +02:00
parent 44a6b536d2
commit 81d8cc1e84
27 changed files with 1336 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
mysql-test/r/ndb_basic.result
View File

@ -568,7 +568,7 @@ t1
insert into t1 values (1,1),(2,1),(3,1),(4,1),(5,2),(6,1),(7,1);
explain select * from t1 where a12345678901234567890123456789a1234567890=2;
id select_type table type possible_keys key key_len ref rows Extra
1 SIMPLE t1 ref a12345678901234567890123456789a1234567890 a12345678901234567890123456789a1234567890 5 const 10 Using where
1 SIMPLE t1 ref a12345678901234567890123456789a1234567890 a12345678901234567890123456789a1234567890 5 const 1 Using where
select * from t1 where a12345678901234567890123456789a1234567890=2;
a1234567890123456789012345678901234567890 a12345678901234567890123456789a1234567890
5 2

6
mysql-test/r/ndb_blob.result
View File

@ -134,7 +134,7 @@ insert into t1 values(2,@b2,222,@d2);
commit;
explain select * from t1 where c = 111;
id select_type table type possible_keys key key_len ref rows Extra
1 SIMPLE t1 ref c c 4 const 10
1 SIMPLE t1 ref c c 4 const 1
select a,length(b),substr(b,1+2*900,2),length(d),substr(d,1+3*900,3)
from t1 where c=111;
a length(b) substr(b,1+2*900,2) length(d) substr(d,1+3*900,3)
@ -242,7 +242,7 @@ insert into t1 values(9,'b9',999,'dd9');
commit;
explain select * from t1 where c >= 100 order by a;
id select_type table type possible_keys key key_len ref rows Extra
1 SIMPLE t1 range c c 4 NULL 10 Using where; Using filesort
1 SIMPLE t1 range c c 4 NULL 9 Using where; Using filesort
select * from t1 where c >= 100 order by a;
a b c d
1 b1 111 dd1
@ -278,7 +278,7 @@ insert into t1 values(2,@b2,222,@d2);
commit;
explain select * from t1 where c >= 100 order by a;
id select_type table type possible_keys key key_len ref rows Extra
1 SIMPLE t1 range c c 4 NULL 10 Using where; Using filesort
1 SIMPLE t1 range c c 4 NULL 2 Using where; Using filesort
select a,length(b),substr(b,1+2*900,2),length(d),substr(d,1+3*900,3)
from t1 where c >= 100 order by a;
a length(b) substr(b,1+2*900,2) length(d) substr(d,1+3*900,3)

4
mysql-test/r/ndb_charset.result
View File

@ -188,7 +188,7 @@ p a
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
1 SIMPLE t1 const a NULL NULL NULL 1
select * from t1 where a = 'aAa' order by p;
p a
1 aAa
@ -225,7 +225,7 @@ p a
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
1 SIMPLE t1 const a NULL NULL NULL 1
select * from t1 where a = 'aAa' order by p;
p a
1 aAa

26
mysql-test/r/ndb_condition_pushdown.result
View File

@ -868,7 +868,7 @@ time_field = '01:01:01' and
date_time = '1901-01-01 01:01:01'
order by auto;
id select_type table type possible_keys key key_len ref rows Extra
1 SIMPLE t1 ref medium_index medium_index 3 const 10 Using where with pushed condition; Using filesort
1 SIMPLE t1 ref medium_index medium_index 3 const 1 Using where with pushed condition; Using filesort
select auto from t1 where
string = "aaaa" and
vstring = "aaaa" and
@ -925,7 +925,7 @@ time_field != '01:01:01' and
date_time != '1901-01-01 01:01:01'
order by auto;
id select_type table type possible_keys key key_len ref rows Extra
1 SIMPLE t1 range medium_index medium_index 3 NULL 20 Using where with pushed condition; Using filesort
1 SIMPLE t1 range medium_index medium_index 3 NULL 3 Using where with pushed condition; Using filesort
select auto from t1 where
string != "aaaa" and
vstring != "aaaa" and
@ -984,7 +984,7 @@ time_field > '01:01:01' and
date_time > '1901-01-01 01:01:01'
order by auto;
id select_type table type possible_keys key key_len ref rows Extra
1 SIMPLE t1 range medium_index medium_index 3 NULL 10 Using where with pushed condition; Using filesort
1 SIMPLE t1 range medium_index medium_index 3 NULL 3 Using where with pushed condition; Using filesort
select auto from t1 where
string > "aaaa" and
vstring > "aaaa" and
@ -1043,7 +1043,7 @@ time_field >= '01:01:01' and
date_time >= '1901-01-01 01:01:01'
order by auto;
id select_type table type possible_keys key key_len ref rows Extra
1 SIMPLE t1 range medium_index medium_index 3 NULL 10 Using where with pushed condition; Using filesort
1 SIMPLE t1 range medium_index medium_index 3 NULL 4 Using where with pushed condition; Using filesort
select auto from t1 where
string >= "aaaa" and
vstring >= "aaaa" and
@ -1103,7 +1103,7 @@ time_field < '04:04:04' and
date_time < '1904-04-04 04:04:04'
order by auto;
id select_type table type possible_keys key key_len ref rows Extra
1 SIMPLE t1 range medium_index medium_index 3 NULL 10 Using where with pushed condition; Using filesort
1 SIMPLE t1 range medium_index medium_index 3 NULL 3 Using where with pushed condition; Using filesort
select auto from t1 where
string < "dddd" and
vstring < "dddd" and
@ -1162,7 +1162,7 @@ time_field <= '04:04:04' and
date_time <= '1904-04-04 04:04:04'
order by auto;
id select_type table type possible_keys key key_len ref rows Extra
1 SIMPLE t1 range medium_index medium_index 3 NULL 10 Using where with pushed condition; Using filesort
1 SIMPLE t1 range medium_index medium_index 3 NULL 4 Using where with pushed condition; Using filesort
select auto from t1 where
string <= "dddd" and
vstring <= "dddd" and
@ -1255,7 +1255,7 @@ select auto from t1 where
(date_time between '1901-01-01 01:01:01' and '1903-03-03 03:03:03')
order by auto;
id select_type table type possible_keys key key_len ref rows Extra
1 SIMPLE t1 range medium_index medium_index 3 NULL 10 Using where with pushed condition; Using filesort
1 SIMPLE t1 range medium_index medium_index 3 NULL 3 Using where with pushed condition; Using filesort
select auto from t1 where
(string between "aaaa" and "cccc") and
(vstring between "aaaa" and "cccc") and
@ -1358,7 +1358,7 @@ select auto from t1 where
(date_time not between '1901-01-01 01:01:01' and '1903-03-03 03:03:03')
order by auto;
id select_type table type possible_keys key key_len ref rows Extra
1 SIMPLE t1 range medium_index medium_index 3 NULL 20 Using where with pushed condition; Using filesort
1 SIMPLE t1 range medium_index medium_index 3 NULL 1 Using where with pushed condition; Using filesort
select auto from t1 where
(string not between "aaaa" and "cccc") and
(vstring not between "aaaa" and "cccc") and
@ -1462,7 +1462,7 @@ time_field in('01:01:01','03:03:03') and
date_time in('1901-01-01 01:01:01','1903-03-03 03:03:03')
order by auto;
id select_type table type possible_keys key key_len ref rows Extra
1 SIMPLE t1 range medium_index medium_index 3 NULL 20 Using where with pushed condition; Using filesort
1 SIMPLE t1 range medium_index medium_index 3 NULL 2 Using where with pushed condition; Using filesort
select auto from t1 where
string in("aaaa","cccc") and
vstring in("aaaa","cccc") and
@ -1514,7 +1514,7 @@ select auto from t1 where
'1901-01-01 01:01:01' in(date_time)
order by auto;
id select_type table type possible_keys key key_len ref rows Extra
1 SIMPLE t1 ref medium_index medium_index 3 const 10 Using where with pushed condition; Using filesort
1 SIMPLE t1 ref medium_index medium_index 3 const 1 Using where with pushed condition; Using filesort
select auto from t1 where
"aaaa" in(string) and
"aaaa" in(vstring) and
@ -1565,7 +1565,7 @@ time_field not in('01:01:01','03:03:03') and
date_time not in('1901-01-01 01:01:01','1903-03-03 03:03:03')
order by auto;
id select_type table type possible_keys key key_len ref rows Extra
1 SIMPLE t1 range medium_index medium_index 3 NULL 30 Using where with pushed condition; Using filesort
1 SIMPLE t1 range medium_index medium_index 3 NULL 2 Using where with pushed condition; Using filesort
select auto from t1 where
string not in("aaaa","cccc") and
vstring not in("aaaa","cccc") and
@ -1738,7 +1738,7 @@ pk1 attr1 attr2 attr3 pk1 attr1 attr2 attr3 attr4
explain
select * from t4 where attr1 < 5 and attr2 > 9223372036854775803 and attr3 != 3 order by t4.pk1;
id select_type table type possible_keys key key_len ref rows Extra
1 SIMPLE t4 range attr1 attr1 4 NULL 10 Using where with pushed condition; Using filesort
1 SIMPLE t4 range attr1 attr1 4 NULL 5 Using where with pushed condition; Using filesort
select * from t4 where attr1 < 5 and attr2 > 9223372036854775803 and attr3 != 3 order by t4.pk1;
pk1 attr1 attr2 attr3 attr4
2 2 9223372036854775804 2 c
@ -1746,7 +1746,7 @@ pk1 attr1 attr2 attr3 attr4
explain
select * from t3,t4 where t4.attr1 > 1 and t4.attr2 = t3.attr2 and t4.attr3 < 5 order by t4.pk1;
id select_type table type possible_keys key key_len ref rows Extra
1 SIMPLE t4 range attr1 attr1 4 NULL 10 Using where with pushed condition; Using temporary; Using filesort
1 SIMPLE t4 range attr1 attr1 4 NULL 4 Using where with pushed condition; Using temporary; Using filesort
1 SIMPLE t3 ALL NULL NULL NULL NULL 6 Using where
select * from t3,t4 where t4.attr1 > 1 and t4.attr2 = t3.attr2 and t4.attr3 < 5 order by t4.pk1;
pk1 attr1 attr2 attr3 attr4 pk1 attr1 attr2 attr3 attr4

138
mysql-test/r/ndb_index_ordered.result
View File

@ -658,3 +658,141 @@ insert into t1 (a, c) values (1,'aaa'),(3,'bbb');
select count(*) from t1 where c<'bbb';
count(*)
1
drop table t1;
set autocommit=1;
show session variables like 'ndb_index_stat_%';
Variable_name Value
ndb_index_stat_cache_entries 32
ndb_index_stat_enable ON
ndb_index_stat_update_freq 20
set ndb_index_stat_enable = off;
show session variables like 'ndb_index_stat_%';
Variable_name Value
ndb_index_stat_cache_entries 32
ndb_index_stat_enable OFF
ndb_index_stat_update_freq 20
create table t1 (a int, b int, c varchar(10) not null,
primary key using hash (a), index(b,c)) engine=ndb;
insert into t1 values
(1,10,'aaa'),(2,10,'bbb'),(3,10,'ccc'),
(4,20,'aaa'),(5,20,'bbb'),(6,20,'ccc'),
(7,30,'aaa'),(8,30,'bbb'),(9,30,'ccc');
select count(*) from t1 where b < 10;
count(*)
0
select count(*) from t1 where b >= 10 and c >= 'bbb';
count(*)
6
select count(*) from t1 where b > 10;
count(*)
6
select count(*) from t1 where b <= 20 and c < 'ccc';
count(*)
4
select count(*) from t1 where b = 20 and c = 'ccc';
count(*)
1
select count(*) from t1 where b > 20;
count(*)
3
select count(*) from t1 where b = 30 and c > 'aaa';
count(*)
2
select count(*) from t1 where b <= 20;
count(*)
6
select count(*) from t1 where b >= 20 and c > 'aaa';
count(*)
4
drop table t1;
set ndb_index_stat_enable = on;
set ndb_index_stat_cache_entries = 0;
show session variables like 'ndb_index_stat_%';
Variable_name Value
ndb_index_stat_cache_entries 0
ndb_index_stat_enable ON
ndb_index_stat_update_freq 20
create table t1 (a int, b int, c varchar(10) not null,
primary key using hash (a), index(b,c)) engine=ndb;
insert into t1 values
(1,10,'aaa'),(2,10,'bbb'),(3,10,'ccc'),
(4,20,'aaa'),(5,20,'bbb'),(6,20,'ccc'),
(7,30,'aaa'),(8,30,'bbb'),(9,30,'ccc');
select count(*) from t1 where b < 10;
count(*)
0
select count(*) from t1 where b >= 10 and c >= 'bbb';
count(*)
6
select count(*) from t1 where b > 10;
count(*)
6
select count(*) from t1 where b <= 20 and c < 'ccc';
count(*)
4
select count(*) from t1 where b = 20 and c = 'ccc';
count(*)
1
select count(*) from t1 where b > 20;
count(*)
3
select count(*) from t1 where b = 30 and c > 'aaa';
count(*)
2
select count(*) from t1 where b <= 20;
count(*)
6
select count(*) from t1 where b >= 20 and c > 'aaa';
count(*)
4
drop table t1;
set ndb_index_stat_enable = on;
set ndb_index_stat_cache_entries = 4;
set ndb_index_stat_update_freq = 2;
show session variables like 'ndb_index_stat_%';
Variable_name Value
ndb_index_stat_cache_entries 4
ndb_index_stat_enable ON
ndb_index_stat_update_freq 2
create table t1 (a int, b int, c varchar(10) not null,
primary key using hash (a), index(b,c)) engine=ndb;
insert into t1 values
(1,10,'aaa'),(2,10,'bbb'),(3,10,'ccc'),
(4,20,'aaa'),(5,20,'bbb'),(6,20,'ccc'),
(7,30,'aaa'),(8,30,'bbb'),(9,30,'ccc');
select count(*) from t1 where b < 10;
count(*)
0
select count(*) from t1 where b >= 10 and c >= 'bbb';
count(*)
6
select count(*) from t1 where b > 10;
count(*)
6
select count(*) from t1 where b <= 20 and c < 'ccc';
count(*)
4
select count(*) from t1 where b = 20 and c = 'ccc';
count(*)
1
select count(*) from t1 where b > 20;
count(*)
3
select count(*) from t1 where b = 30 and c > 'aaa';
count(*)
2
select count(*) from t1 where b <= 20;
count(*)
6
select count(*) from t1 where b >= 20 and c > 'aaa';
count(*)
4
drop table t1;
set ndb_index_stat_enable = @@global.ndb_index_stat_enable;
set ndb_index_stat_cache_entries = @@global.ndb_index_stat_cache_entries;
set ndb_index_stat_update_freq = @@global.ndb_index_stat_update_freq;
show session variables like 'ndb_index_stat_%';
Variable_name Value
ndb_index_stat_cache_entries 32
ndb_index_stat_enable ON
ndb_index_stat_update_freq 20

74
mysql-test/t/ndb_index_ordered.test
View File

@ -354,5 +354,79 @@ create table t1 (a int, c varchar(10),
primary key using hash (a), index(c)) engine=ndb;
insert into t1 (a, c) values (1,'aaa'),(3,'bbb');
select count(*) from t1 where c<'bbb';
drop table t1;
# -- index statistics --
set autocommit=1;
show session variables like 'ndb_index_stat_%';
set ndb_index_stat_enable = off;
show session variables like 'ndb_index_stat_%';
create table t1 (a int, b int, c varchar(10) not null,
primary key using hash (a), index(b,c)) engine=ndb;
insert into t1 values
(1,10,'aaa'),(2,10,'bbb'),(3,10,'ccc'),
(4,20,'aaa'),(5,20,'bbb'),(6,20,'ccc'),
(7,30,'aaa'),(8,30,'bbb'),(9,30,'ccc');
select count(*) from t1 where b < 10;
select count(*) from t1 where b >= 10 and c >= 'bbb';
select count(*) from t1 where b > 10;
select count(*) from t1 where b <= 20 and c < 'ccc';
select count(*) from t1 where b = 20 and c = 'ccc';
select count(*) from t1 where b > 20;
select count(*) from t1 where b = 30 and c > 'aaa';
select count(*) from t1 where b <= 20;
select count(*) from t1 where b >= 20 and c > 'aaa';
drop table t1;
set ndb_index_stat_enable = on;
set ndb_index_stat_cache_entries = 0;
show session variables like 'ndb_index_stat_%';
create table t1 (a int, b int, c varchar(10) not null,
primary key using hash (a), index(b,c)) engine=ndb;
insert into t1 values
(1,10,'aaa'),(2,10,'bbb'),(3,10,'ccc'),
(4,20,'aaa'),(5,20,'bbb'),(6,20,'ccc'),
(7,30,'aaa'),(8,30,'bbb'),(9,30,'ccc');
select count(*) from t1 where b < 10;
select count(*) from t1 where b >= 10 and c >= 'bbb';
select count(*) from t1 where b > 10;
select count(*) from t1 where b <= 20 and c < 'ccc';
select count(*) from t1 where b = 20 and c = 'ccc';
select count(*) from t1 where b > 20;
select count(*) from t1 where b = 30 and c > 'aaa';
select count(*) from t1 where b <= 20;
select count(*) from t1 where b >= 20 and c > 'aaa';
drop table t1;
set ndb_index_stat_enable = on;
set ndb_index_stat_cache_entries = 4;
set ndb_index_stat_update_freq = 2;
show session variables like 'ndb_index_stat_%';
create table t1 (a int, b int, c varchar(10) not null,
primary key using hash (a), index(b,c)) engine=ndb;
insert into t1 values
(1,10,'aaa'),(2,10,'bbb'),(3,10,'ccc'),
(4,20,'aaa'),(5,20,'bbb'),(6,20,'ccc'),
(7,30,'aaa'),(8,30,'bbb'),(9,30,'ccc');
select count(*) from t1 where b < 10;
select count(*) from t1 where b >= 10 and c >= 'bbb';
select count(*) from t1 where b > 10;
select count(*) from t1 where b <= 20 and c < 'ccc';
select count(*) from t1 where b = 20 and c = 'ccc';
select count(*) from t1 where b > 20;
select count(*) from t1 where b = 30 and c > 'aaa';
select count(*) from t1 where b <= 20;
select count(*) from t1 where b >= 20 and c > 'aaa';
drop table t1;
set ndb_index_stat_enable = @@global.ndb_index_stat_enable;
set ndb_index_stat_cache_entries = @@global.ndb_index_stat_cache_entries;
set ndb_index_stat_update_freq = @@global.ndb_index_stat_update_freq;
show session variables like 'ndb_index_stat_%';
# End of 4.1 tests

125
sql/ha_ndbcluster.cc
View File

@ -31,6 +31,7 @@
#include "ha_ndbcluster.h"
#include <ndbapi/NdbApi.hpp>
#include <ndbapi/NdbScanFilter.hpp>
#include <ndbapi/NdbIndexStat.hpp>
// options from from mysqld.cc
extern my_bool opt_ndb_optimized_node_selection;
@ -85,6 +86,14 @@ static handlerton ndbcluster_hton = {
DBUG_RETURN(ndb_to_mysql_error(&tmp)); \
}
#define ERR_BREAK(err, code) \
{ \
const NdbError& tmp= err; \
ERR_PRINT(tmp); \
code= ndb_to_mysql_error(&tmp); \
break; \
}
// Typedefs for long names
typedef NdbDictionary::Column NDBCOL;
typedef NdbDictionary::Table NDBTAB;
@ -1064,6 +1073,26 @@ int ha_ndbcluster::build_index_list(Ndb *ndb, TABLE *tab, enum ILBP phase)
const NDBINDEX *index= dict->getIndex(index_name, m_tabname);
if (!index) DBUG_RETURN(1);
m_index[i].index= (void *) index;
// ordered index - add stats
NDB_INDEX_DATA& d=m_index[i];
delete d.index_stat;
d.index_stat=NULL;
THD *thd=current_thd;
if (thd->variables.ndb_index_stat_enable)
{
d.index_stat=new NdbIndexStat(index);
d.index_stat_cache_entries=thd->variables.ndb_index_stat_cache_entries;
d.index_stat_update_freq=thd->variables.ndb_index_stat_update_freq;
d.index_stat_query_count=0;
d.index_stat->alloc_cache(d.index_stat_cache_entries);
DBUG_PRINT("info", ("index %s stat=on cache_entries=%u update_freq=%u",
index->getName(),
d.index_stat_cache_entries,
d.index_stat_update_freq));
} else
{
DBUG_PRINT("info", ("index %s stat=off", index->getName()));
}
}
if (idx_type == UNIQUE_ORDERED_INDEX || idx_type == UNIQUE_INDEX)
{
@ -1135,6 +1164,8 @@ void ha_ndbcluster::release_metadata()
my_free((char *)m_index[i].unique_index_attrid_map, MYF(0));
m_index[i].unique_index_attrid_map= NULL;
}
delete m_index[i].index_stat;
m_index[i].index_stat=NULL;
}
DBUG_VOID_RETURN;
@ -1648,10 +1679,12 @@ inline int ha_ndbcluster::next_result(byte *buf)
*/
int ha_ndbcluster::set_bounds(NdbIndexScanOperation *op,
uint inx,
bool rir,
const key_range *keys[2],
uint range_no)
{
const KEY *const key_info= table->key_info + active_index;
const KEY *const key_info= table->key_info + inx;
const uint key_parts= key_info->key_parts;
uint key_tot_len[2];
uint tot_len;
@ -1716,7 +1749,10 @@ int ha_ndbcluster::set_bounds(NdbIndexScanOperation *op,
switch (p.key->flag)
{
case HA_READ_KEY_EXACT:
if (! rir)
p.bound_type= NdbIndexScanOperation::BoundEQ;
else // differs for records_in_range
p.bound_type= NdbIndexScanOperation::BoundLE;
break;
// ascending
case HA_READ_KEY_OR_NEXT:
@ -1874,7 +1910,7 @@ int ha_ndbcluster::ordered_index_scan(const key_range *start_key,
{
const key_range *keys[2]= { start_key, end_key };
res= set_bounds(op, keys);
res= set_bounds(op, active_index, false, keys);
if (res)
DBUG_RETURN(res);
}
@ -4162,6 +4198,10 @@ ha_ndbcluster::ha_ndbcluster(TABLE *table_arg):
m_index[i].unique_index= NULL;
m_index[i].index= NULL;
m_index[i].unique_index_attrid_map= NULL;
m_index[i].index_stat=NULL;
m_index[i].index_stat_cache_entries=0;
m_index[i].index_stat_update_freq=0;
m_index[i].index_stat_query_count=0;
}
DBUG_VOID_RETURN;
@ -4914,6 +4954,84 @@ ha_ndbcluster::records_in_range(uint inx, key_range *min_key,
(max_key && max_key->length == key_length)))
DBUG_RETURN(1);
if ((idx_type == PRIMARY_KEY_ORDERED_INDEX ||
idx_type == UNIQUE_ORDERED_INDEX ||
idx_type == ORDERED_INDEX) &&
m_index[inx].index_stat != NULL)
{
NDB_INDEX_DATA& d=m_index[inx];
NDBINDEX* index=(NDBINDEX*)d.index;
Ndb* ndb=get_ndb();
NdbTransaction* trans=NULL;
NdbIndexScanOperation* op=NULL;
int res=0;
Uint64 rows;
do
{
// We must provide approx table rows
Uint64 table_rows=0;
Ndb_local_table_statistics *info=
(Ndb_local_table_statistics *)m_table_info;
if (info->records != ~(ha_rows)0 && info->records != 0)
{
table_rows = info->records;
DBUG_PRINT("info", ("use info->records: %llu", table_rows));
}
else
{
Ndb_statistics stat;
if ((res=ndb_get_table_statistics(ndb, m_tabname, &stat)) != 0)
break;
table_rows=stat.row_count;
DBUG_PRINT("info", ("use db row_count: %llu", table_rows));
if (table_rows == 0) {
// Problem if autocommit=0
#ifdef ndb_get_table_statistics_uses_active_trans
rows=0;
break;
#endif
}
}
// Define scan op for the range
if ((trans=m_active_trans) == NULL)
{
DBUG_PRINT("info", ("no active trans"));
if (! (trans=ndb->startTransaction()))
ERR_BREAK(ndb->getNdbError(), res);
}
if (! (op=trans->getNdbIndexScanOperation(index, (NDBTAB*)m_table)))
ERR_BREAK(trans->getNdbError(), res);
if ((op->readTuples(NdbOperation::LM_CommittedRead)) == -1)
ERR_BREAK(op->getNdbError(), res);
const key_range *keys[2]={ min_key, max_key };
if ((res=set_bounds(op, inx, true, keys)) != 0)
break;
// Decide if db should be contacted
int flags=0;
if (d.index_stat_query_count < d.index_stat_cache_entries ||
(d.index_stat_update_freq != 0 &&
d.index_stat_query_count % d.index_stat_update_freq == 0))
{
DBUG_PRINT("info", ("force stat from db"));
flags|=NdbIndexStat::RR_UseDb;
}
if (d.index_stat->records_in_range(index, op, table_rows, &rows, flags) == -1)
ERR_BREAK(d.index_stat->getNdbError(), res);
d.index_stat_query_count++;
} while (0);
if (trans != m_active_trans && rows == 0)
rows = 1;
if (trans != m_active_trans && trans != NULL)
ndb->closeTransaction(trans);
if (res != 0)
DBUG_RETURN(HA_POS_ERROR);
DBUG_RETURN(rows);
}
DBUG_RETURN(10); /* Good guess when you don't know anything */
}
@ -5612,7 +5730,8 @@ ha_ndbcluster::read_multi_range_first(KEY_MULTI_RANGE **found_range_p,
const key_range *keys[2]= { &multi_range_curr->start_key,
&multi_range_curr->end_key };
if ((res= set_bounds(scanOp, keys, multi_range_curr-ranges)))
if ((res= set_bounds(scanOp, active_index, false, keys,
multi_range_curr-ranges)))
DBUG_RETURN(res);
break;
}

10
sql/ha_ndbcluster.h
View File

@ -35,6 +35,7 @@ class NdbScanOperation;
class NdbScanFilter;
class NdbIndexScanOperation;
class NdbBlob;
class NdbIndexStat;
// connectstring to cluster if given by mysqld
extern const char *ndbcluster_connectstring;
@ -54,6 +55,12 @@ typedef struct ndb_index_data {
void *index;
void *unique_index;
unsigned char *unique_index_attrid_map;
// In this version stats are not shared between threads
NdbIndexStat* index_stat;
uint index_stat_cache_entries;
// Simple counter mechanism to decide when to connect to db
uint index_stat_update_freq;
uint index_stat_query_count;
} NDB_INDEX_DATA;
typedef struct st_ndbcluster_share {
@ -642,7 +649,8 @@ private:
int get_ndb_blobs_value(NdbBlob *last_ndb_blob);
int set_primary_key(NdbOperation *op, const byte *key);
int set_primary_key_from_record(NdbOperation *op, const byte *record);
int set_bounds(NdbIndexScanOperation*, const key_range *keys[2], uint= 0);
int set_bounds(NdbIndexScanOperation*, uint inx, bool rir,
const key_range *keys[2], uint= 0);
int key_cmp(uint keynr, const byte * old_row, const byte * new_row);
int set_index_key(NdbOperation *, const KEY *key_info, const byte *key_ptr);
void print_results();

25
sql/mysqld.cc
View File

@ -4335,6 +4335,8 @@ enum options_mysqld
OPT_NDB_SHM, OPT_NDB_OPTIMIZED_NODE_SELECTION, OPT_NDB_CACHE_CHECK_TIME,
OPT_NDB_MGMD, OPT_NDB_NODEID,
OPT_NDB_DISTRIBUTION,
OPT_NDB_INDEX_STAT_ENABLE,
OPT_NDB_INDEX_STAT_CACHE_ENTRIES, OPT_NDB_INDEX_STAT_UPDATE_FREQ,
OPT_SKIP_SAFEMALLOC,
OPT_TEMP_POOL, OPT_TX_ISOLATION, OPT_COMPLETION_TYPE,
OPT_SKIP_STACK_TRACE, OPT_SKIP_SYMLINKS,
@ -4949,6 +4951,23 @@ Disable with --skip-ndbcluster (will save memory).",
"A dedicated thread is created to, at the given millisecons interval, invalidate the query cache if another MySQL server in the cluster has changed the data in the database.",
(gptr*) &opt_ndb_cache_check_time, (gptr*) &opt_ndb_cache_check_time, 0, GET_ULONG, REQUIRED_ARG,
0, 0, LONG_TIMEOUT, 0, 1, 0},
{"ndb-index-stat-enable", OPT_NDB_INDEX_STAT_ENABLE,
"Use ndb index statistics in query optimization.",
(gptr*) &global_system_variables.ndb_index_stat_enable,
(gptr*) &max_system_variables.ndb_index_stat_enable,
0, GET_BOOL, OPT_ARG, 1, 0, 1, 0, 0, 0},
{"ndb-index-stat-cache-entries", OPT_NDB_INDEX_STAT_CACHE_ENTRIES,
"Number of start/end keys to store in statistics memory cache."
" Zero means no cache and forces query of db nodes always.",
(gptr*) &global_system_variables.ndb_index_stat_cache_entries,
(gptr*) &max_system_variables.ndb_index_stat_cache_entries,
0, GET_ULONG, OPT_ARG, 32, 0, ~0L, 0, 0, 0},
{"ndb-index-stat-update-freq", OPT_NDB_INDEX_STAT_UPDATE_FREQ,
"How often, in the long run, to query db nodes instead of statistics cache."
" For example 20 means every 20th time.",
(gptr*) &global_system_variables.ndb_index_stat_update_freq,
(gptr*) &max_system_variables.ndb_index_stat_update_freq,
0, GET_ULONG, OPT_ARG, 20, 0, ~0L, 0, 0, 0},
#endif
{"new", 'n', "Use very new possible 'unsafe' functions.",
(gptr*) &global_system_variables.new_mode,
@ -6211,6 +6230,12 @@ static void mysql_init_variables(void)
#endif
#ifdef HAVE_NDBCLUSTER_DB
have_ndbcluster=SHOW_OPTION_DISABLED;
global_system_variables.ndb_index_stat_enable=TRUE;
max_system_variables.ndb_index_stat_enable=TRUE;
global_system_variables.ndb_index_stat_cache_entries=32;
max_system_variables.ndb_index_stat_cache_entries=~0L;
global_system_variables.ndb_index_stat_update_freq=20;
max_system_variables.ndb_index_stat_update_freq=~0L;
#else
have_ndbcluster=SHOW_OPTION_NO;
#endif

17
sql/set_var.cc
View File

@ -433,6 +433,15 @@ sys_var_thd_bool
sys_ndb_use_transactions("ndb_use_transactions", &SV::ndb_use_transactions);
sys_var_long_ptr
sys_ndb_cache_check_time("ndb_cache_check_time", &ndb_cache_check_time);
sys_var_thd_bool
sys_ndb_index_stat_enable("ndb_index_stat_enable",
&SV::ndb_index_stat_enable);
sys_var_thd_ulong
sys_ndb_index_stat_cache_entries("ndb_index_stat_cache_entries",
&SV::ndb_index_stat_cache_entries);
sys_var_thd_ulong
sys_ndb_index_stat_update_freq("ndb_index_stat_update_freq",
&SV::ndb_index_stat_update_freq);
#endif
/* Time/date/datetime formats */
@ -725,6 +734,9 @@ sys_var *sys_variables[]=
&sys_ndb_force_send,
&sys_ndb_use_exact_count,
&sys_ndb_use_transactions,
&sys_ndb_index_stat_enable,
&sys_ndb_index_stat_cache_entries,
&sys_ndb_index_stat_update_freq,
#endif
&sys_unique_checks,
&sys_updatable_views_with_limit,
@ -906,10 +918,13 @@ struct show_var_st init_vars[]= {
#ifdef HAVE_NDBCLUSTER_DB
{sys_ndb_autoincrement_prefetch_sz.name,
(char*) &sys_ndb_autoincrement_prefetch_sz, SHOW_SYS},
{sys_ndb_cache_check_time.name,(char*) &sys_ndb_cache_check_time, SHOW_SYS},
{sys_ndb_force_send.name, (char*) &sys_ndb_force_send, SHOW_SYS},
{sys_ndb_index_stat_cache_entries.name, (char*) &sys_ndb_index_stat_cache_entries, SHOW_SYS},
{sys_ndb_index_stat_enable.name, (char*) &sys_ndb_index_stat_enable, SHOW_SYS},
{sys_ndb_index_stat_update_freq.name, (char*) &sys_ndb_index_stat_update_freq, SHOW_SYS},
{sys_ndb_use_exact_count.name,(char*) &sys_ndb_use_exact_count, SHOW_SYS},
{sys_ndb_use_transactions.name,(char*) &sys_ndb_use_transactions, SHOW_SYS},
{sys_ndb_cache_check_time.name,(char*) &sys_ndb_cache_check_time, SHOW_SYS},
#endif
{sys_net_buffer_length.name,(char*) &sys_net_buffer_length, SHOW_SYS},
{sys_net_read_timeout.name, (char*) &sys_net_read_timeout, SHOW_SYS},

3
sql/sql_class.h
View File

@ -565,6 +565,9 @@ struct system_variables
my_bool ndb_force_send;
my_bool ndb_use_exact_count;
my_bool ndb_use_transactions;
my_bool ndb_index_stat_enable;
ulong ndb_index_stat_cache_entries;
ulong ndb_index_stat_update_freq;
#endif /* HAVE_NDBCLUSTER_DB */
my_bool old_alter_table;
my_bool old_passwords;

2
storage/ndb/include/kernel/AttributeHeader.hpp
View File

@ -41,6 +41,8 @@ public:
STATIC_CONST( ROW_SIZE = 0xFFFA );
STATIC_CONST( FRAGMENT_MEMORY= 0xFFF9 );
STATIC_CONST( RECORDS_IN_RANGE = 0xFFF8 );
/** Initialize AttributeHeader at location aHeaderPtr */
static AttributeHeader& init(void* aHeaderPtr, Uint32 anAttributeId,
Uint32 aDataSize);

2
storage/ndb/include/ndbapi/NdbDictionary.hpp
View File

@ -456,6 +456,7 @@ public:
static const Column * COMMIT_COUNT;
static const Column * ROW_SIZE;
static const Column * RANGE_NO;
static const Column * RECORDS_IN_RANGE;
int getSizeInBytes() const;
#endif
@ -929,6 +930,7 @@ public:
private:
#ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
friend class NdbIndexImpl;
friend class NdbIndexStat;
#endif
class NdbIndexImpl & m_impl;
Index(NdbIndexImpl&);

3
storage/ndb/include/ndbapi/NdbIndexScanOperation.hpp
View File

@ -30,6 +30,7 @@ class NdbIndexScanOperation : public NdbScanOperation {
friend class NdbResultSet;
friend class NdbOperation;
friend class NdbScanOperation;
friend class NdbIndexStat;
#endif
public:
@ -149,12 +150,14 @@ public:
* Is current scan sorted descending
*/
bool getDescending() const { return m_descending; }
private:
NdbIndexScanOperation(Ndb* aNdb);
virtual ~NdbIndexScanOperation();
int setBound(const NdbColumnImpl*, int type, const void* aValue, Uint32 len);
int insertBOUNDS(Uint32 * data, Uint32 sz);
Uint32 getKeyFromSCANTABREQ(Uint32* data, Uint32 size);
virtual int equal_impl(const NdbColumnImpl*, const char*, Uint32);
virtual NdbRecAttr* getValue_impl(const NdbColumnImpl*, char*);

141
storage/ndb/include/ndbapi/NdbIndexStat.hpp Normal file
View File

@ -0,0 +1,141 @@
/* Copyright (C) 2003 MySQL AB
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#ifndef NdbIndexStat_H
#define NdbIndexStat_H
#include <ndb_global.h>
#include <NdbDictionary.hpp>
#include <NdbError.hpp>
class NdbIndexImpl;
class NdbIndexScanOperation;
/*
* Statistics for an ordered index.
*/
class NdbIndexStat {
public:
NdbIndexStat(const NdbDictionary::Index* index);
~NdbIndexStat();
/*
* Allocate memory for cache. Argument is minimum number of stat
* entries and applies to lower and upper bounds separately. More
* entries may fit (keys have variable size). If not used, db is
* contacted always.
*/
int alloc_cache(Uint32 entries);
/*
* Flags for records_in_range.
*/
enum {
RR_UseDb = 1, // contact db
RR_NoUpdate = 2 // but do not update cache
};
/*
* Estimate how many index records need to be scanned. The scan
* operation must be prepared with lock mode LM_CommittedRead and must
* have the desired bounds set. The routine may use local cache or
* may contact db by executing the operation.
*
* If returned count is zero then db was contacted and the count is
* exact. Otherwise the count is approximate. If cache is used then
* caller must provide estimated number of table rows. It will be
* multiplied by a percentage obtained from the cache (result zero is
* returned as 1).
*/
int records_in_range(NdbDictionary::Index* index,
NdbIndexScanOperation* op,
Uint64 table_rows,
Uint64* count,
int flags);
/*
* Get latest error.
*/
const NdbError& getNdbError() const;
private:
/*
* There are 2 areas: start keys and end keys. An area has pointers
* at beginning and entries at end. Pointers are sorted by key.
*
* A pointer contains entry offset and also entry timestamp. An entry
* contains the key and percentage of rows _not_ satisfying the bound
* i.e. less than start key or greater than end key.
*
* A key is an array of index key bounds. Each has type (0-4) in
* first word followed by data with AttributeHeader.
*
* Stat update comes as pair of start and end key and associated
* percentages. Stat query takes best match of start and end key from
* each area separately. Rows in range percentage is then computed by
* excluding the two i.e. as 100 - (start key pct + end key pct).
*
* TODO use more compact key format
*/
friend struct Area;
struct Pointer {
Uint16 m_pos;
Uint16 m_seq;
};
struct Entry {
float m_pct;
Uint32 m_keylen;
};
STATIC_CONST( EntrySize = sizeof(Entry) >> 2 );
STATIC_CONST( PointerSize = sizeof(Pointer) >> 2 );
struct Area {
Uint32* m_data;
Uint32 m_offset;
Uint32 m_free;
Uint16 m_entries;
Uint8 m_idir;
Uint8 pad1;
Pointer& get_pointer(unsigned i) const {
return *(Pointer*)&m_data[i];
}
Entry& get_entry(unsigned i) const {
return *(Entry*)&m_data[get_pointer(i).m_pos];
}
Uint32 get_pos(const Entry& e) const {
return (const Uint32*)&e - m_data;
}
unsigned get_firstpos() const {
return PointerSize * m_entries + m_free;
}
};
const NdbIndexImpl& m_index;
Uint32 m_areasize;
Uint16 m_seq;
Area m_area[2];
Uint32* m_cache;
NdbError m_error;
#ifdef VM_TRACE
void stat_verify();
#endif
int stat_cmpkey(const Area& a, const Uint32* key1, Uint32 keylen1,
const Uint32* key2, Uint32 keylen2);
int stat_search(const Area& a, const Uint32* key, Uint32 keylen,
Uint32* idx, bool* match);
int stat_oldest(const Area& a);
int stat_delete(Area& a, Uint32 k);
int stat_update(const Uint32* key1, Uint32 keylen1,
const Uint32* key2, Uint32 keylen2, const float pct[2]);
int stat_select(const Uint32* key1, Uint32 keylen1,
const Uint32* key2, Uint32 keylen2, float pct[2]);
void set_error(int code);
};
#endif

13
storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
View File

@ -2622,7 +2622,10 @@ Dblqh::execREAD_PSEUDO_REQ(Signal* signal){
regTcPtr.i = signal->theData[0];
ptrCheckGuard(regTcPtr, ctcConnectrecFileSize, tcConnectionrec);
if(signal->theData[1] != AttributeHeader::RANGE_NO)
if (signal->theData[1] == AttributeHeader::RANGE_NO) {
signal->theData[0] = regTcPtr.p->m_scan_curr_range_no;
}
else if (signal->theData[1] != AttributeHeader::RECORDS_IN_RANGE)
{
jam();
FragrecordPtr regFragptr;
@ -2634,7 +2637,13 @@ Dblqh::execREAD_PSEUDO_REQ(Signal* signal){
}
else
{
signal->theData[0] = regTcPtr.p->m_scan_curr_range_no;
jam();
// scanptr gets reset somewhere within the timeslice
ScanRecordPtr tmp;
tmp.i = regTcPtr.p->tcScanRec;
c_scanRecordPool.getPtr(tmp);
signal->theData[0] = tmp.p->scanAccPtr;
EXECUTE_DIRECT(DBTUX, GSN_READ_PSEUDO_REQ, signal, 2);
}
}

12
storage/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp
View File

@ -1028,6 +1028,18 @@ Dbtup::read_pseudo(Uint32 attrId, Uint32* outBuffer){
EXECUTE_DIRECT(DBLQH, GSN_READ_PSEUDO_REQ, signal, 2);
outBuffer[0] = signal->theData[0];
return 1;
case AttributeHeader::RECORDS_IN_RANGE:
signal->theData[0] = operPtr.p->userpointer;
signal->theData[1] = attrId;
EXECUTE_DIRECT(DBLQH, GSN_READ_PSEUDO_REQ, signal, 2);
outBuffer[0] = signal->theData[0];
outBuffer[1] = signal->theData[1];
outBuffer[2] = signal->theData[2];
outBuffer[3] = signal->theData[3];
return 4;
default:
return 0;
}

17
storage/ndb/src/kernel/blocks/dbtux/Dbtux.hpp
View File

@ -82,10 +82,14 @@
#define jam() jamLine(80000 + __LINE__)
#define jamEntry() jamEntryLine(80000 + __LINE__)
#endif
#ifdef DBTUX_DEBUG_CPP
#ifdef DBTUX_STAT_CPP
#define jam() jamLine(90000 + __LINE__)
#define jamEntry() jamEntryLine(90000 + __LINE__)
#endif
#ifdef DBTUX_DEBUG_CPP
#define jam() jamLine(100000 + __LINE__)
#define jamEntry() jamEntryLine(100000 + __LINE__)
#endif
#ifndef jam
#define jam() jamLine(__LINE__)
#define jamEntry() jamEntryLine(__LINE__)
@ -116,6 +120,7 @@ private:
STATIC_CONST( MaxPrefSize = MAX_TTREE_PREF_SIZE );
STATIC_CONST( ScanBoundSegmentSize = 7 );
STATIC_CONST( MaxAccLockOps = MAX_PARALLEL_OP_PER_SCAN );
STATIC_CONST( MaxTreeDepth = 32 ); // strict
BLOCK_DEFINES(Dbtux);
// forward declarations
@ -269,6 +274,7 @@ private:
Uint8 m_prefSize; // words in min prefix
Uint8 m_minOccup; // min entries in internal node
Uint8 m_maxOccup; // max entries in node
Uint32 m_entryCount; // stat: current entries
TupLoc m_root; // root node
TreeHead();
// methods
@ -659,6 +665,14 @@ private:
int cmpSearchKey(const Frag& frag, unsigned& start, ConstData searchKey, ConstData entryData, unsigned maxlen = MaxAttrDataSize);
int cmpScanBound(const Frag& frag, unsigned dir, ConstData boundInfo, unsigned boundCount, ConstData entryData, unsigned maxlen = MaxAttrDataSize);
/*
* DbtuxStat.cpp
*/
void execREAD_PSEUDO_REQ(Signal* signal);
void statRecordsInRange(ScanOpPtr scanPtr, Uint32* out);
Uint32 getEntriesBeforeOrAfter(Frag& frag, TreePos pos, unsigned idir);
unsigned getPathToNode(NodeHandle node, Uint16* path);
/*
* DbtuxDebug.cpp
*/
@ -952,6 +966,7 @@ Dbtux::TreeHead::TreeHead() :
m_prefSize(0),
m_minOccup(0),
m_maxOccup(0),
m_entryCount(0),
m_root()
{
}

4
storage/ndb/src/kernel/blocks/dbtux/DbtuxGen.cpp
View File

@ -65,6 +65,10 @@ Dbtux::Dbtux(const Configuration& conf) :
addRecSignal(GSN_ACCKEYCONF, &Dbtux::execACCKEYCONF);
addRecSignal(GSN_ACCKEYREF, &Dbtux::execACCKEYREF);
addRecSignal(GSN_ACC_ABORTCONF, &Dbtux::execACC_ABORTCONF);
/*
* DbtuxStat.cpp
*/
addRecSignal(GSN_READ_PSEUDO_REQ, &Dbtux::execREAD_PSEUDO_REQ);
/*
* DbtuxDebug.cpp
*/

160
storage/ndb/src/kernel/blocks/dbtux/DbtuxStat.cpp Normal file
View File

@ -0,0 +1,160 @@
/* Copyright (C) 2003 MySQL AB
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#define DBTUX_STAT_CPP
#include "Dbtux.hpp"
void
Dbtux::execREAD_PSEUDO_REQ(Signal* signal)
{
jamEntry();
ScanOpPtr scanPtr;
scanPtr.i = signal->theData[0];
c_scanOpPool.getPtr(scanPtr);
if (signal->theData[1] == AttributeHeader::RECORDS_IN_RANGE) {
jam();
statRecordsInRange(scanPtr, &signal->theData[0]);
} else {
ndbassert(false);
}
}
/*
* Estimate entries in range. Scan is at first entry. Search for last
* entry i.e. start of descending scan. Use the 2 positions to estimate
* entries before and after the range. Finally get entries in range by
* subtracting from total. Errors come from imperfectly balanced tree
* and from uncommitted entries which differ only in tuple version.
*
* Returns 4 Uint32 values: 0) total entries 1) in range 2) before range
* 3) after range. 1-3) are estimates and need not add up to 0).
*/
void
Dbtux::statRecordsInRange(ScanOpPtr scanPtr, Uint32* out)
{
ScanOp& scan = *scanPtr.p;
Frag& frag = *c_fragPool.getPtr(scan.m_fragPtrI);
TreeHead& tree = frag.m_tree;
// get first and last position
TreePos pos1 = scan.m_scanPos;
TreePos pos2;
{ // as in scanFirst()
TreeHead& tree = frag.m_tree;
setKeyAttrs(frag);
const unsigned idir = 1;
const ScanBound& bound = *scan.m_bound[idir];
ScanBoundIterator iter;
bound.first(iter);
for (unsigned j = 0; j < bound.getSize(); j++) {
jam();
c_dataBuffer[j] = *iter.data;
bound.next(iter);
}
searchToScan(frag, c_dataBuffer, scan.m_boundCnt[idir], true, pos2);
// committed read (same timeslice) and range not empty
ndbrequire(pos2.m_loc != NullTupLoc);
}
out[0] = frag.m_tree.m_entryCount;
out[2] = getEntriesBeforeOrAfter(frag, pos1, 0);
out[3] = getEntriesBeforeOrAfter(frag, pos2, 1);
if (pos1.m_loc == pos2.m_loc) {
ndbrequire(pos2.m_pos >= pos1.m_pos);
out[1] = pos2.m_pos - pos1.m_pos + 1;
} else {
Uint32 rem = out[2] + out[3];
if (out[0] > rem) {
out[1] = out[0] - rem;
} else {
// random guess one node apart
out[1] = tree.m_maxOccup;
}
}
}
/*
* Estimate number of entries strictly before or after given position.
* Each branch to right direction wins parent node and the subtree on
* the other side. Subtree entries is estimated from depth and total
* entries by assuming that the tree is perfectly balanced.
*/
Uint32
Dbtux::getEntriesBeforeOrAfter(Frag& frag, TreePos pos, unsigned idir)
{
NodeHandle node(frag);
selectNode(node, pos.m_loc);
Uint16 path[MaxTreeDepth + 1];
unsigned depth = getPathToNode(node, path);
ndbrequire(depth != 0 && depth <= MaxTreeDepth);
TreeHead& tree = frag.m_tree;
Uint32 cnt = 0;
Uint32 tot = tree.m_entryCount;
unsigned i = 0;
// contribution from levels above
while (i + 1 < depth) {
unsigned occup2 = (path[i] >> 8);
unsigned side = (path[i + 1] & 0xFF);
// subtree of this node has about half the entries
tot = tot >= occup2 ? (tot - occup2) / 2 : 0;
// branch to other side wins parent and a subtree
if (side != idir) {
cnt += occup2;
cnt += tot;
}
i++;
}
// contribution from this node
unsigned occup = (path[i] >> 8);
ndbrequire(pos.m_pos < occup);
if (idir == 0) {
if (pos.m_pos != 0)
cnt += pos.m_pos - 1;
} else {
cnt += occup - (pos.m_pos + 1);
}
// contribution from levels below
tot = tot >= occup ? (tot - occup) / 2 : 0;
cnt += tot;
return cnt;
}
/*
* Construct path to given node. Returns depth. Root node has path
* 2 and depth 1. In general the path is 2{0,1}* where 0,1 is the side
* (left,right branch). In addition the occupancy of each node is
* returned in the upper 8 bits.
*/
unsigned
Dbtux::getPathToNode(NodeHandle node, Uint16* path)
{
TupLoc loc = node.m_loc;
unsigned i = MaxTreeDepth;
while (loc != NullTupLoc) {
jam();
selectNode(node, loc);
path[i] = node.getSide() | (node.getOccup() << 8);
loc = node.getLink(2);
ndbrequire(i != 0);
i--;
}
unsigned depth = MaxTreeDepth - i;
unsigned j = 0;
while (j < depth) {
path[j] = path[i + 1 + j];
j++;
}
path[j] = 0xFFFF; // catch bug
return depth;
}

21
storage/ndb/src/kernel/blocks/dbtux/DbtuxTree.cpp
View File

@ -26,6 +26,7 @@ Dbtux::treeAdd(Frag& frag, TreePos treePos, TreeEnt ent)
{
TreeHead& tree = frag.m_tree;
NodeHandle node(frag);
do {
if (treePos.m_loc != NullTupLoc) {
// non-empty tree
jam();
@ -35,16 +36,19 @@ Dbtux::treeAdd(Frag& frag, TreePos treePos, TreeEnt ent)
// node has room
jam();
nodePushUp(node, pos, ent, RNIL);
return;
break;
}
treeAddFull(frag, node, pos, ent);
return;
break;
}
jam();
insertNode(node);
nodePushUp(node, 0, ent, RNIL);
node.setSide(2);
tree.m_root = node.m_loc;
break;
} while (0);
tree.m_entryCount++;
}
/*
@ -178,31 +182,36 @@ Dbtux::treeRemove(Frag& frag, TreePos treePos)
NodeHandle node(frag);
selectNode(node, treePos.m_loc);
TreeEnt ent;
do {
if (node.getOccup() > tree.m_minOccup) {
// no underflow in any node type
jam();
nodePopDown(node, pos, ent, 0);
return;
break;
}
if (node.getChilds() == 2) {
// underflow in interior node
jam();
treeRemoveInner(frag, node, pos);
return;
break;
}
// remove entry in semi/leaf
nodePopDown(node, pos, ent, 0);
if (node.getLink(0) != NullTupLoc) {
jam();
treeRemoveSemi(frag, node, 0);
return;
break;
}
if (node.getLink(1) != NullTupLoc) {
jam();
treeRemoveSemi(frag, node, 1);
return;
break;
}
treeRemoveLeaf(frag, node);
break;
} while (0);
ndbrequire(tree.m_entryCount != 0);
tree.m_entryCount--;
}
/*

1
storage/ndb/src/kernel/blocks/dbtux/Makefile.am
View File

@ -9,6 +9,7 @@ libdbtux_a_SOURCES = \
DbtuxScan.cpp \
DbtuxSearch.cpp \
DbtuxCmp.cpp \
DbtuxStat.cpp \
DbtuxDebug.cpp
INCLUDES_LOC = -I$(top_srcdir)/storage/ndb/src/kernel/blocks/dbtup

1
storage/ndb/src/ndbapi/Makefile.am
View File

@ -35,6 +35,7 @@ libndbapi_la_SOURCES = \
DictCache.cpp \
ndb_cluster_connection.cpp \
NdbBlob.cpp \
NdbIndexStat.cpp \
SignalSender.cpp
INCLUDES_LOC = -I$(top_srcdir)/storage/ndb/src/mgmapi

1
storage/ndb/src/ndbapi/NdbDictionary.cpp
View File

@ -1070,3 +1070,4 @@ const NdbDictionary::Column * NdbDictionary::Column::ROW_COUNT = 0;
const NdbDictionary::Column * NdbDictionary::Column::COMMIT_COUNT = 0;
const NdbDictionary::Column * NdbDictionary::Column::ROW_SIZE = 0;
const NdbDictionary::Column * NdbDictionary::Column::RANGE_NO = 0;
const NdbDictionary::Column * NdbDictionary::Column::RECORDS_IN_RANGE = 0;

11
storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp
View File

@ -266,6 +266,11 @@ NdbColumnImpl::create_pseudo(const char * name){
col->m_impl.m_attrId = AttributeHeader::RANGE_NO;
col->m_impl.m_attrSize = 4;
col->m_impl.m_arraySize = 1;
} else if(!strcmp(name, "NDB$RECORDS_IN_RANGE")){
col->setType(NdbDictionary::Column::Unsigned);
col->m_impl.m_attrId = AttributeHeader::RECORDS_IN_RANGE;
col->m_impl.m_attrSize = 4;
col->m_impl.m_arraySize = 4;
} else {
abort();
}
@ -739,12 +744,14 @@ NdbDictionaryImpl::~NdbDictionaryImpl()
delete NdbDictionary::Column::COMMIT_COUNT;
delete NdbDictionary::Column::ROW_SIZE;
delete NdbDictionary::Column::RANGE_NO;
delete NdbDictionary::Column::RECORDS_IN_RANGE;
NdbDictionary::Column::FRAGMENT= 0;
NdbDictionary::Column::FRAGMENT_MEMORY= 0;
NdbDictionary::Column::ROW_COUNT= 0;
NdbDictionary::Column::COMMIT_COUNT= 0;
NdbDictionary::Column::ROW_SIZE= 0;
NdbDictionary::Column::RANGE_NO= 0;
NdbDictionary::Column::RECORDS_IN_RANGE= 0;
}
m_globalHash->unlock();
} else {
@ -817,6 +824,8 @@ NdbDictionaryImpl::setTransporter(class Ndb* ndb,
NdbColumnImpl::create_pseudo("NDB$ROW_SIZE");
NdbDictionary::Column::RANGE_NO=
NdbColumnImpl::create_pseudo("NDB$RANGE_NO");
NdbDictionary::Column::RECORDS_IN_RANGE=
NdbColumnImpl::create_pseudo("NDB$RECORDS_IN_RANGE");
}
m_globalHash->unlock();
return true;
@ -2175,6 +2184,7 @@ NdbDictInterface::create_index_obj_from_table(NdbIndexImpl** dst,
}
* dst = idx;
return 0;
}
@ -3209,4 +3219,3 @@ template class Vector<Uint32>;
template class Vector<Vector<Uint32> >;
template class Vector<NdbTableImpl*>;
template class Vector<NdbColumnImpl*>;

490
storage/ndb/src/ndbapi/NdbIndexStat.cpp Normal file
View File

@ -0,0 +1,490 @@
/* Copyright (C) 2003 MySQL AB
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#include <ndb_global.h>
#include <AttributeHeader.hpp>
#include <NdbSqlUtil.hpp>
#include <NdbIndexStat.hpp>
#include <NdbTransaction.hpp>
#include <NdbIndexScanOperation.hpp>
#include "NdbDictionaryImpl.hpp"
#include <my_sys.h>
NdbIndexStat::NdbIndexStat(const NdbDictionary::Index* index) :
m_index(index->m_impl),
m_cache(NULL)
{
}
NdbIndexStat::~NdbIndexStat()
{
delete [] m_cache;
m_cache = NULL;
}
int
NdbIndexStat::alloc_cache(Uint32 entries)
{
delete [] m_cache;
m_cache = NULL;
if (entries == 0) {
return 0;
}
Uint32 i;
Uint32 keysize = 0;
for (i = 0; i < m_index.m_columns.size(); i++) {
NdbColumnImpl* c = m_index.m_columns[i];
keysize += 2; // counting extra headers
keysize += (c->m_attrSize * c->m_arraySize + 3 ) / 4;
}
Uint32 areasize = entries * (PointerSize + EntrySize + keysize);
if (areasize > (1 << 16))
areasize = (1 << 16);
Uint32 cachesize = 2 * areasize;
m_cache = new Uint32 [cachesize];
if (m_cache == NULL) {
set_error(4000);
return -1;
}
m_areasize = areasize;
m_seq = 0;
Uint32 idir;
for (idir = 0; idir <= 1; idir++) {
Area& a = m_area[idir];
a.m_data = &m_cache[idir * areasize];
a.m_offset = a.m_data - &m_cache[0];
a.m_free = areasize;
a.m_entries = 0;
a.m_idir = idir;
a.pad1 = 0;
}
#ifdef VM_TRACE
memset(&m_cache[0], 0x3f, cachesize << 2);
#endif
return 0;
}
#ifndef VM_TRACE
#define stat_verify()
#else
void
NdbIndexStat::stat_verify()
{
Uint32 idir;
for (idir = 0; idir <= 1; idir++) {
Uint32 i;
const Area& a = m_area[idir];
assert(a.m_offset == idir * m_areasize);
assert(a.m_data == &m_cache[a.m_offset]);
Uint32 pointerwords = PointerSize * a.m_entries;
Uint32 entrywords = 0;
for (i = 0; i < a.m_entries; i++) {
const Pointer& p = a.get_pointer(i);
const Entry& e = a.get_entry(i);
assert(a.get_pos(e) == p.m_pos);
entrywords += EntrySize + e.m_keylen;
}
assert(a.m_free <= m_areasize);
assert(pointerwords + a.m_free + entrywords == m_areasize);
Uint32 off = pointerwords + a.m_free;
for (i = 0; i < a.m_entries; i++) {
assert(off < m_areasize);
const Entry& e = *(const Entry*)&a.m_data[off];
off += EntrySize + e.m_keylen;
}
assert(off == m_areasize);
for (i = 0; i < a.m_entries; i++) {
const Entry& e = a.get_entry(i);
const Uint32* entrykey = (const Uint32*)&e + EntrySize;
Uint32 n = 0;
while (n + 2 <= e.m_keylen) {
Uint32 t = entrykey[n++];
assert(t == 2 * idir || t == 2 * idir + 1 || t == 4);
AttributeHeader ah = *(const AttributeHeader*)&entrykey[n++];
n += ah.getDataSize();
}
assert(n == e.m_keylen);
}
for (i = 0; i + 1 < a.m_entries; i++) {
const Entry& e1 = a.get_entry(i);
const Entry& e2 = a.get_entry(i + 1);
const Uint32* entrykey1 = (const Uint32*)&e1 + EntrySize;
const Uint32* entrykey2 = (const Uint32*)&e2 + EntrySize;
int ret = stat_cmpkey(a, entrykey1, e1.m_keylen, entrykey2, e2.m_keylen);
assert(ret == -1);
}
}
}
#endif
// compare keys
int
NdbIndexStat::stat_cmpkey(const Area& a, const Uint32* key1, Uint32 keylen1, const Uint32* key2, Uint32 keylen2)
{
const Uint32 idir = a.m_idir;
const int jdir = 1 - 2 * int(idir);
Uint32 i1 = 0, i2 = 0;
Uint32 t1 = 4, t2 = 4; //BoundEQ
int ret = 0;
Uint32 k = 0;
while (k < m_index.m_columns.size()) {
NdbColumnImpl* c = m_index.m_columns[k];
Uint32 n = c->m_attrSize * c->m_arraySize;
// absence of keypart is treated specially
bool havekp1 = (i1 + 2 <= keylen1);
bool havekp2 = (i2 + 2 <= keylen2);
AttributeHeader ah1;
AttributeHeader ah2;
if (havekp1) {
t1 = key1[i1++];
assert(t1 == 2 * idir || t1 == 2 * idir + 1 || t1 == 4);
ah1 = *(const AttributeHeader*)&key1[i1++];
}
if (havekp2) {
t2 = key2[i2++];
assert(t2 == 2 * idir || t2 == 2 * idir + 1 || t2 == 4);
ah2 = *(const AttributeHeader*)&key2[i2++];
}
if (havekp1) {
if (havekp2) {
if (! ah1.isNULL()) {
if (! ah2.isNULL()) {
const NdbSqlUtil::Type& sqlType = NdbSqlUtil::getType(c->m_type);
ret = (*sqlType.m_cmp)(c->m_cs, &key1[i1], n, &key2[i2], n, true);
if (ret != 0)
break;
} else {
ret = +1;
break;
}
} else if (! ah2.isNULL()) {
ret = -1;
break;
}
} else {
ret = +jdir;
break;
}
} else {
if (havekp2) {
ret = -jdir;
break;
} else {
// no more keyparts on either side
break;
}
}
i1 += ah1.getDataSize();
i2 += ah2.getDataSize();
k++;
}
if (ret == 0) {
// strict bound is greater as start key and less as end key
int s1 = t1 & 1;
int s2 = t2 & 1;
ret = (s1 - s2) * jdir;
}
return ret;
}
// find first key >= given key
int
NdbIndexStat::stat_search(const Area& a, const Uint32* key, Uint32 keylen, Uint32* idx, bool* match)
{
// points at minus/plus infinity
int lo = -1;
int hi = a.m_entries;
// loop invariant: key(lo) < key < key(hi)
while (hi - lo > 1) {
// observe lo < j < hi
int j = (hi + lo) / 2;
Entry& e = a.get_entry(j);
const Uint32* key2 = (Uint32*)&e + EntrySize;
Uint32 keylen2 = e.m_keylen;
int ret = stat_cmpkey(a, key, keylen, key2, keylen2);
// observe the loop invariant if ret != 0
if (ret < 0)
hi = j;
else if (ret > 0)
lo = j;
else {
*idx = j;
*match = true;
return 0;
}
}
// hi - lo == 1 and key(lo) < key < key(hi)
*idx = hi;
*match = false;
return 0;
}
// find oldest entry
int
NdbIndexStat::stat_oldest(const Area& a)
{
Uint32 i, k, m;
bool found = false;
for (i = 0; i < a.m_entries; i++) {
Pointer& p = a.get_pointer(i);
Entry& e = a.get_entry(i);
Uint32 m2 = m_seq >= p.m_seq ? m_seq - p.m_seq : p.m_seq - m_seq;
if (! found || m < m2) {
m = m2;
k = i;
found = true;
}
}
assert(found);
return k;
}
// delete entry
int
NdbIndexStat::stat_delete(Area& a, Uint32 k)
{
Uint32 i;
NdbIndexStat::Entry& e = a.get_entry(k);
Uint32 entrylen = EntrySize + e.m_keylen;
Uint32 pos = a.get_pos(e);
// adjust pointers to entries after
for (i = 0; i < a.m_entries; i++) {
Pointer& p = a.get_pointer(i);
if (p.m_pos < pos) {
p.m_pos += entrylen;
}
}
// compact entry area
unsigned firstpos = a.get_firstpos();
for (i = pos; i > firstpos; i--) {
a.m_data[i + entrylen - 1] = a.m_data[i - 1];
}
// compact pointer area
for (i = k; i + 1 < a.m_entries; i++) {
NdbIndexStat::Pointer& p = a.get_pointer(i);
NdbIndexStat::Pointer& q = a.get_pointer(i + 1);
p = q;
}
a.m_free += PointerSize + entrylen;
a.m_entries--;
stat_verify();
return 0;
}
// update or insert stat values
int
NdbIndexStat::stat_update(const Uint32* key1, Uint32 keylen1, const Uint32* key2, Uint32 keylen2, const float pct[2])
{
const Uint32* const key[2] = { key1, key2 };
const Uint32 keylen[2] = { keylen1, keylen2 };
Uint32 idir;
for (idir = 0; idir <= 1; idir++) {
Area& a = m_area[idir];
Uint32 k;
bool match;
stat_search(a, key[idir], keylen[idir], &k, &match);
Uint16 seq = m_seq++;
if (match) {
// update old entry
NdbIndexStat::Pointer& p = a.get_pointer(k);
NdbIndexStat::Entry& e = a.get_entry(k);
e.m_pct = pct[idir];
p.m_seq = seq;
} else {
Uint32 entrylen = NdbIndexStat::EntrySize + keylen[idir];
Uint32 need = NdbIndexStat::PointerSize + entrylen;
while (need > a.m_free) {
Uint32 j = stat_oldest(a);
if (j < k)
k--;
stat_delete(a, j);
}
// insert pointer
Uint32 i;
for (i = a.m_entries; i > k; i--) {
NdbIndexStat::Pointer& p1 = a.get_pointer(i);
NdbIndexStat::Pointer& p2 = a.get_pointer(i - 1);
p1 = p2;
}
NdbIndexStat::Pointer& p = a.get_pointer(k);
// insert entry
Uint32 firstpos = a.get_firstpos();
p.m_pos = firstpos - entrylen;
NdbIndexStat::Entry& e = a.get_entry(k);
e.m_pct = pct[idir];
e.m_keylen = keylen[idir];
Uint32* entrykey = (Uint32*)&e + EntrySize;
for (i = 0; i < keylen[idir]; i++) {
entrykey[i] = key[idir][i];
}
p.m_seq = seq;
// total
a.m_free -= PointerSize + entrylen;
a.m_entries++;
}
}
stat_verify();
return 0;
}
int
NdbIndexStat::stat_select(const Uint32* key1, Uint32 keylen1, const Uint32* key2, Uint32 keylen2, float pct[2])
{
const Uint32* const key[2] = { key1, key2 };
const Uint32 keylen[2] = { keylen1, keylen2 };
Uint32 idir;
for (idir = 0; idir <= 1; idir++) {
Area& a = m_area[idir];
Uint32 k;
bool match;
stat_search(a, key[idir], keylen[idir], &k, &match);
if (match) {
NdbIndexStat::Entry& e = a.get_entry(k);
pct[idir] = e.m_pct;
} else if (k == 0) {
NdbIndexStat::Entry& e = a.get_entry(k);
if (idir == 0)
pct[idir] = e.m_pct / 2;
else
pct[idir] = e.m_pct + (1 - e.m_pct) / 2;
} else if (k == a.m_entries) {
NdbIndexStat::Entry& e = a.get_entry(k - 1);
if (idir == 0)
pct[idir] = e.m_pct + (1 - e.m_pct) / 2;
else
pct[idir] = e.m_pct / 2;
} else {
NdbIndexStat::Entry& e1 = a.get_entry(k - 1);
NdbIndexStat::Entry& e2 = a.get_entry(k);
pct[idir] = (e1.m_pct + e2.m_pct) / 2;
}
}
return 0;
}
int
NdbIndexStat::records_in_range(NdbDictionary::Index* index, NdbIndexScanOperation* op, Uint64 table_rows, Uint64* count, int flags)
{
DBUG_ENTER("NdbIndexStat::records_in_range");
Uint64 rows;
Uint32 key1[1000], keylen1;
Uint32 key2[1000], keylen2;
if (m_cache == NULL)
flags |= RR_UseDb | RR_NoUpdate;
else if (m_area[0].m_entries == 0 || m_area[1].m_entries == 0)
flags |= RR_UseDb;
if ((flags & (RR_UseDb | RR_NoUpdate)) != RR_UseDb | RR_NoUpdate) {
// get start and end key - assume bound is ordered, wellformed
Uint32 bound[1000];
Uint32 boundlen = op->getKeyFromSCANTABREQ(bound, 1000);
keylen1 = keylen2 = 0;
Uint32 n = 0;
while (n < boundlen) {
Uint32 t = bound[n];
AttributeHeader ah(bound[n + 1]);
Uint32 sz = 2 + ah.getDataSize();
t &= 0xFFFF; // may contain length
assert(t <= 4);
bound[n] = t;
if (t == 0 || t == 1 || t == 4) {
memcpy(&key1[keylen1], &bound[n], sz << 2);
keylen1 += sz;
}
if (t == 2 || t == 3 || t == 4) {
memcpy(&key2[keylen2], &bound[n], sz << 2);
keylen2 += sz;
}
n += sz;
}
}
if (flags & RR_UseDb) {
Uint32 out[4] = { 0, 0, 0, 0 }; // rows, in, before, after
float tot[4] = { 0, 0, 0, 0 }; // totals of above
int cnt, ret;
bool forceSend = true;
NdbTransaction* trans = op->m_transConnection;
if (op->interpret_exit_last_row() == -1 ||
op->getValue(NdbDictionary::Column::RECORDS_IN_RANGE, (char*)out) == 0) {
DBUG_PRINT("error", ("op:%d", op->getNdbError().code));
DBUG_RETURN(-1);
}
if (trans->execute(NdbTransaction::NoCommit,
NdbTransaction::AbortOnError, forceSend) == -1) {
DBUG_PRINT("error", ("trans:%d op:%d", trans->getNdbError().code,
op->getNdbError().code));
DBUG_RETURN(-1);
}
cnt = 0;
while ((ret = op->nextResult(true, forceSend)) == 0) {
DBUG_PRINT("info", ("frag rows=%u in=%u before=%u after=%u [error=%d]",
out[0], out[1], out[2], out[3],
(int)(out[1] + out[2] + out[3]) - (int)out[0]));
unsigned i;
for (i = 0; i < 4; i++)
tot[i] += (float)out[i];
cnt++;
}
if (ret == -1) {
DBUG_PRINT("error", ("trans:%d op:%d", trans->getNdbError().code,
op->getNdbError().code));
DBUG_RETURN(-1);
}
op->close(forceSend);
rows = (Uint64)tot[1];
if (cnt != 0 && ! (flags & RR_NoUpdate)) {
float pct[2];
pct[0] = 100 * tot[2] / tot[0];
pct[1] = 100 * tot[3] / tot[0];
DBUG_PRINT("info", ("update stat pct"
" before=%.2f after=%.2f",
pct[0], pct[1]));
stat_update(key1, keylen1, key2, keylen2, pct);
}
} else {
float pct[2];
stat_select(key1, keylen1, key2, keylen2, pct);
float diff = 100.0 - (pct[0] + pct[1]);
float trows = (float)table_rows;
DBUG_PRINT("info", ("select stat pct"
" before=%.2f after=%.2f in=%.2f table_rows=%.2f",
pct[0], pct[1], diff, trows));
rows = 0;
if (diff >= 0)
rows = (Uint64)(diff * trows / 100);
if (rows == 0)
rows = 1;
}
*count = rows;
DBUG_PRINT("value", ("rows=%llu flags=%o", rows, flags));
DBUG_RETURN(0);
}
void
NdbIndexStat::set_error(int code)
{
m_error.code = code;
}
const NdbError&
NdbIndexStat::getNdbError() const
{
return m_error;
}

25
storage/ndb/src/ndbapi/NdbScanOperation.cpp
View File

@ -1198,6 +1198,31 @@ error:
return -1;
}
Uint32
NdbIndexScanOperation::getKeyFromSCANTABREQ(Uint32* data, Uint32 size)
{
DBUG_ENTER("NdbIndexScanOperation::getKeyFromSCANTABREQ");
assert(size >= theTotalNrOfKeyWordInSignal);
size = theTotalNrOfKeyWordInSignal;
NdbApiSignal* tSignal = theSCAN_TABREQ->next();
Uint32 pos = 0;
while (pos < size) {
assert(tSignal != NULL);
Uint32* tData = tSignal->getDataPtrSend();
Uint32 rem = size - pos;
if (rem > KeyInfo::DataLength)
rem = KeyInfo::DataLength;
Uint32 i = 0;
while (i < rem) {
data[pos + i] = tData[KeyInfo::HeaderLength + i];
i++;
}
pos += rem;
}
DBUG_DUMP("key", (char*)data, size << 2);
DBUG_RETURN(size);
}
int
NdbIndexScanOperation::readTuples(LockMode lm,
Uint32 scan_flags,