Merge willster.(none):/home/stewart/Documents/MySQL/5.1/ndb

into  willster.(none):/home/stewart/Documents/MySQL/5.1/bug21253

mysql-test/r/ndb_basic.result

@@ -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	1	Using where
+1	SIMPLE	t1	ref	a12345678901234567890123456789a1234567890	a12345678901234567890123456789a1234567890	5	const	#	Using where
 select * from t1 where a12345678901234567890123456789a1234567890=2;
 a1234567890123456789012345678901234567890	a12345678901234567890123456789a1234567890
 5	2

mysql-test/r/ndb_blob.result

@@ -40,7 +40,7 @@ insert into t1 values(2,@b2,222,@d2);
 commit;
 explain select * from t1 where a = 1;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t1	const	PRIMARY	PRIMARY	4	const	1	
+1	SIMPLE	t1	const	PRIMARY	PRIMARY	4	const	#	
 select a,length(b),substr(b,1+2*900,2),length(d),substr(d,1+3*900,3)
 from t1 where a=1;
 a	length(b)	substr(b,1+2*900,2)	length(d)	substr(d,1+3*900,3)
@@ -87,7 +87,7 @@ replace t1 set a=2,b=@b2,c=222,d=@d2;
 commit;
 explain select * from t1 where a = 1;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t1	const	PRIMARY	PRIMARY	4	const	1	
+1	SIMPLE	t1	const	PRIMARY	PRIMARY	4	const	#	
 select a,length(b),substr(b,1+2*900,2),length(d),substr(d,1+3*900,3)
 from t1 where a=1;
 a	length(b)	substr(b,1+2*900,2)	length(d)	substr(d,1+3*900,3)
@@ -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	1	
+1	SIMPLE	t1	ref	c	c	4	const	#	
 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)
@@ -177,7 +177,7 @@ insert into t1 values(9,'b9',999,'dd9');
 commit;
 explain select * from t1;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	9	
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	#	
 select * from t1 order by a;
 a	b	c	d
 1	b1	111	dd1
@@ -212,7 +212,7 @@ insert into t1 values(2,@b2,222,@d2);
 commit;
 explain select * from t1;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	2	
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	#	
 select a,length(b),substr(b,1+2*900,2),length(d),substr(d,1+3*900,3)
 from t1 order by a;
 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	9	Using where; Using filesort
+1	SIMPLE	t1	range	c	c	4	NULL	#	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	2	Using where; Using filesort
+1	SIMPLE	t1	range	c	c	4	NULL	#	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)

mysql-test/r/ndb_charset.result

@@ -186,9 +186,6 @@ p	a
 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	const	a	NULL	NULL	NULL	1	
 select * from t1 where a = 'aAa' order by p;
 p	a
 1	aAa
@@ -223,9 +220,6 @@ p	a
 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	const	a	NULL	NULL	NULL	1	
 select * from t1 where a = 'aAa' order by p;
 p	a
 1	aAa

mysql-test/r/ndb_condition_pushdown.result

@@ -513,7 +513,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	ALL	NULL	NULL	NULL	NULL	4	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	#	Using where with pushed condition; Using filesort
 select auto from t1 where 
 string = "aaaa" and 
 vstring = "aaaa" and 
@@ -570,7 +570,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	ALL	NULL	NULL	NULL	NULL	4	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	#	Using where with pushed condition; Using filesort
 select auto from t1 where 
 string != "aaaa" and 
 vstring != "aaaa" and 
@@ -629,7 +629,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	ALL	NULL	NULL	NULL	NULL	4	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	#	Using where with pushed condition; Using filesort
 select auto from t1 where 
 string > "aaaa" and 
 vstring > "aaaa" and 
@@ -688,7 +688,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	ALL	NULL	NULL	NULL	NULL	4	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	#	Using where with pushed condition; Using filesort
 select auto from t1 where 
 string >= "aaaa" and 
 vstring >= "aaaa" and 
@@ -748,7 +748,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	ALL	NULL	NULL	NULL	NULL	4	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	#	Using where with pushed condition; Using filesort
 select auto from t1 where 
 string < "dddd" and 
 vstring < "dddd" and 
@@ -807,7 +807,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	ALL	NULL	NULL	NULL	NULL	4	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	#	Using where with pushed condition; Using filesort
 select auto from t1 where 
 string <= "dddd" and 
 vstring <= "dddd" and 
@@ -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	1	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	ref	medium_index	medium_index	3	const	#	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	3	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	range	medium_index	medium_index	3	NULL	#	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	3	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	range	medium_index	medium_index	3	NULL	#	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	4	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	range	medium_index	medium_index	3	NULL	#	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	3	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	range	medium_index	medium_index	3	NULL	#	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	4	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	range	medium_index	medium_index	3	NULL	#	Using where with pushed condition; Using filesort
 select auto from t1 where 
 string <= "dddd" and 
 vstring <= "dddd" and 
@@ -1202,7 +1202,7 @@ bin like concat(0xBB, '%') and
 vbin like concat(0xBB, '%')
 order by auto;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	4	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	#	Using where with pushed condition; Using filesort
 select auto from t1 where 
 string like "b%" and
 vstring like "b%" and
@@ -1219,7 +1219,7 @@ bin not like concat(0xBB, '%') and
 vbin not like concat(0xBB, '%')
 order by auto;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	4	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	#	Using where with pushed condition; Using filesort
 select auto from t1 where 
 string not like "b%" and
 vstring not like "b%" 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	3	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	range	medium_index	medium_index	3	NULL	#	Using where with pushed condition; Using filesort
 select auto from t1 where
 (string between "aaaa" and "cccc") and 
 (vstring between "aaaa" and "cccc") and 
@@ -1307,7 +1307,7 @@ select auto from t1 where
 ('1901-01-01 01:01:01' between date_time and date_time) 
 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	1	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	range	medium_index	medium_index	3	NULL	#	Using where with pushed condition; Using filesort
 select auto from t1 where
 ("aaaa" between string and string) and 
 ("aaaa" between vstring and vstring) 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	1	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	range	medium_index	medium_index	3	NULL	#	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 
@@ -1409,7 +1409,7 @@ select auto from t1 where
 ('1901-01-01 01:01:01' not between date_time and date_time) 
 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	3	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	range	medium_index	medium_index	3	NULL	#	Using where with pushed condition; Using filesort
 select auto from t1 where
 ("aaaa" not between string and string) and 
 ("aaaa" not between vstring and vstring) 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	2	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	range	medium_index	medium_index	3	NULL	#	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	1	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	ref	medium_index	medium_index	3	const	#	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	6	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	range	medium_index	medium_index	3	NULL	#	Using where with pushed condition; Using filesort
 select auto from t1 where
 string not in("aaaa","cccc") and 
 vstring not in("aaaa","cccc") and 
@@ -1617,7 +1617,7 @@ select auto from t1 where
 '1901-01-01 01:01:01' not in(date_time) 
 order by auto;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	4	Using where with pushed condition; Using filesort
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	#	Using where with pushed condition; Using filesort
 select auto from t1 where
 "aaaa" not in(string) and 
 "aaaa" not in(vstring) and 
@@ -1704,7 +1704,7 @@ count(*)
 explain 
 select * from t2 where attr3 is null or attr1 > 2 and pk1= 3 order by pk1;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t2	ALL	PRIMARY	NULL	NULL	NULL	6	Using where with pushed condition; Using filesort
+1	SIMPLE	t2	ALL	PRIMARY	NULL	NULL	NULL	#	Using where with pushed condition; Using filesort
 select * from t2 where attr3 is null or attr1 > 2 and pk1= 3 order by pk1;
 pk1	attr1	attr2	attr3
 2	2	NULL	NULL
@@ -1712,7 +1712,7 @@ pk1	attr1	attr2	attr3
 explain
 select * from t2 where attr3 is not null and attr1 > 2 order by pk1;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t2	ALL	NULL	NULL	NULL	NULL	6	Using where with pushed condition; Using filesort
+1	SIMPLE	t2	ALL	NULL	NULL	NULL	NULL	#	Using where with pushed condition; Using filesort
 select * from t2 where attr3 is not null and attr1 > 2 order by pk1;
 pk1	attr1	attr2	attr3
 3	3	3	d
@@ -1721,7 +1721,7 @@ pk1	attr1	attr2	attr3
 explain
 select * from t3 where attr2 >  9223372036854775803 and attr3 != 3 order by pk1;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t3	ALL	NULL	NULL	NULL	NULL	6	Using where with pushed condition; Using filesort
+1	SIMPLE	t3	ALL	NULL	NULL	NULL	NULL	#	Using where with pushed condition; Using filesort
 select * from t3 where attr2 >  9223372036854775803 and attr3 != 3 order by pk1;
 pk1	attr1	attr2	attr3	attr4
 2	2	9223372036854775804	2	c
@@ -1730,15 +1730,15 @@ pk1	attr1	attr2	attr3	attr4
 explain
 select * from t2,t3 where t2.attr1 < 1 and t2.attr2 = t3.attr2 and t3.attr1 < 5 order by t2.pk1;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t2	ALL	NULL	NULL	NULL	NULL	6	Using where with pushed condition; Using temporary; Using filesort
-1	SIMPLE	t3	ALL	NULL	NULL	NULL	NULL	6	Using where with pushed condition
+1	SIMPLE	t2	ALL	NULL	NULL	NULL	NULL	#	Using where with pushed condition; Using temporary; Using filesort
+1	SIMPLE	t3	ALL	NULL	NULL	NULL	NULL	#	Using where with pushed condition
 select * from t2,t3 where t2.attr1 < 1 and t2.attr2 = t3.attr2 and t3.attr1 < 5 order by t2.pk1;
 pk1	attr1	attr2	attr3	pk1	attr1	attr2	attr3	attr4
 0	0	0	a	0	0	0	0	a
 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	5	Using where with pushed condition; Using filesort
+1	SIMPLE	t4	range	attr1	attr1	4	NULL	#	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,8 +1746,8 @@ 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	4	Using where with pushed condition; Using temporary; Using filesort
-1	SIMPLE	t3	ALL	NULL	NULL	NULL	NULL	6	Using where
+1	SIMPLE	t4	range	attr1	attr1	4	NULL	#	Using where with pushed condition; Using temporary; Using filesort
+1	SIMPLE	t3	ALL	NULL	NULL	NULL	NULL	#	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
 2	2	9223372036854775804	2	c	2	2	9223372036854775804	2	c
@@ -1756,16 +1756,16 @@ pk1	attr1	attr2	attr3	attr4	pk1	attr1	attr2	attr3	attr4
 explain
 select auto from t1 where string = "aaaa" collate latin1_general_ci order by auto;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	3	Using where; Using filesort
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	#	Using where; Using filesort
 explain
 select * from t2 where (attr1 < 2) = (attr2 < 2) order by pk1;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t2	ALL	NULL	NULL	NULL	NULL	6	Using where; Using filesort
+1	SIMPLE	t2	ALL	NULL	NULL	NULL	NULL	#	Using where; Using filesort
 explain
 select * from t3 left join t4 on t4.attr2 = t3.attr2 where t4.attr1 > 1 and t4.attr3 < 5 or t4.attr1 is null order by t4.pk1;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t3	ALL	NULL	NULL	NULL	NULL	6	Using temporary; Using filesort
-1	SIMPLE	t4	ALL	NULL	NULL	NULL	NULL	6	Using where
+1	SIMPLE	t3	ALL	NULL	NULL	NULL	NULL	#	Using temporary; Using filesort
+1	SIMPLE	t4	ALL	NULL	NULL	NULL	NULL	#	Using where
 create table t5 (a int primary key auto_increment, b tinytext not null) 
 engine = ndb;
 insert into t5 (b) values ('jonas'), ('jensing'), ('johan');
@@ -1777,7 +1777,7 @@ a	b
 set engine_condition_pushdown = on;
 explain select * from t5 where b like '%jo%';
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t5	ALL	NULL	NULL	NULL	NULL	3	Using where
+1	SIMPLE	t5	ALL	NULL	NULL	NULL	NULL	#	Using where
 select * from t5 where b like '%jo%' order by a;
 a	b
 1	jonas
@@ -1793,13 +1793,13 @@ auto
 set engine_condition_pushdown = on;
 explain select auto from t1 where date_time like '1902-02-02 %';
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	3	Using where
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	#	Using where
 select auto from t1 where date_time like '1902-02-02 %' order by auto;
 auto
 2
 explain select auto from t1 where date_time not like '1902-02-02 %';
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	3	Using where
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	#	Using where
 select auto from t1 where date_time not like '1902-02-02 %' order by auto;
 auto
 3

mysql-test/r/ndb_dd_sql_features.result

@@ -441,7 +441,7 @@ INSERT INTO test.t1 values(1,@vc1,@d1);
 INSERT INTO test.t1 values(2,@vc2,@d2);
 explain SELECT * from test.t1 WHERE a1 = 1;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t1	const	PRIMARY	PRIMARY	4	const	1	
+1	SIMPLE	t1	const	PRIMARY	PRIMARY	4	const	#	
 SELECT a1,length(a2),substr(a2,1+2*900,2),length(a3),substr(a3,1+3*900,3)
 FROM test.t1 WHERE a1=1 ORDER BY a1;
 a1	length(a2)	substr(a2,1+2*900,2)	length(a3)	substr(a3,1+3*900,3)

mysql-test/r/ndb_index_ordered.result

@@ -663,7 +663,7 @@ 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_enable	OFF
 ndb_index_stat_update_freq	20
 set ndb_index_stat_enable = off;
 show session variables like 'ndb_index_stat_%';
@@ -794,5 +794,5 @@ 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_enable	OFF
 ndb_index_stat_update_freq	20

mysql-test/r/ndb_subquery.result

@@ -8,32 +8,32 @@ insert into t1 values (1,1,1),(2,2,2),(3,3,3);
 insert into t2 values (1,1,1),(2,2,2),(3,3,3), (4,4,4), (5,5,5);
 explain select * from t2 where p NOT IN (select p from t1);
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	PRIMARY	t2	ALL	NULL	NULL	NULL	NULL	5	Using where
-2	DEPENDENT SUBQUERY	t1	unique_subquery	PRIMARY	PRIMARY	4	func	1	Using index
+1	PRIMARY	t2	ALL	NULL	NULL	NULL	NULL	#	Using where
+2	DEPENDENT SUBQUERY	t1	unique_subquery	PRIMARY	PRIMARY	4	func	#	Using index
 select * from t2 where p NOT IN (select p from t1) order by p;
 p	u	o
 4	4	4
 5	5	5
 explain select * from t2 where p NOT IN (select u from t1);
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	PRIMARY	t2	ALL	NULL	NULL	NULL	NULL	5	Using where
-2	DEPENDENT SUBQUERY	t1	unique_subquery	u	u	4	func	1	Using index
+1	PRIMARY	t2	ALL	NULL	NULL	NULL	NULL	#	Using where
+2	DEPENDENT SUBQUERY	t1	unique_subquery	u	u	4	func	#	Using index
 select * from t2 where p NOT IN (select u from t1) order by p;
 p	u	o
 4	4	4
 5	5	5
 explain select * from t2 where p NOT IN (select o from t1);
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	PRIMARY	t2	ALL	NULL	NULL	NULL	NULL	5	Using where
-2	DEPENDENT SUBQUERY	t1	index_subquery	o	o	4	func	1	Using index
+1	PRIMARY	t2	ALL	NULL	NULL	NULL	NULL	#	Using where
+2	DEPENDENT SUBQUERY	t1	index_subquery	o	o	4	func	#	Using index
 select * from t2 where p NOT IN (select o from t1) order by p;
 p	u	o
 4	4	4
 5	5	5
 explain select * from t2 where p NOT IN (select p+0 from t1);
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	PRIMARY	t2	ALL	NULL	NULL	NULL	NULL	5	Using where
-2	DEPENDENT SUBQUERY	t1	ALL	NULL	NULL	NULL	NULL	3	Using where
+1	PRIMARY	t2	ALL	NULL	NULL	NULL	NULL	#	Using where
+2	DEPENDENT SUBQUERY	t1	ALL	NULL	NULL	NULL	NULL	#	Using where
 select * from t2 where p NOT IN (select p+0 from t1) order by p;
 p	u	o
 4	4	4

mysql-test/t/ndb_basic.test

@@ -550,6 +550,7 @@ index(a12345678901234567890123456789a1234567890)
 ) engine=ndb;
 show tables;
 insert into t1 values (1,1),(2,1),(3,1),(4,1),(5,2),(6,1),(7,1);
+--replace_column 9 #
 explain select * from t1 where a12345678901234567890123456789a1234567890=2;
 select * from t1 where a12345678901234567890123456789a1234567890=2;
 drop table t1;

mysql-test/t/ndb_blob.test

@@ -65,6 +65,7 @@ select length(@x0),length(@b2),length(@d2) from dual;
 insert into t1 values(1,@b1,111,@d1);
 insert into t1 values(2,@b2,222,@d2);
 commit;
+--replace_column 9 #
 explain select * from t1 where a = 1;
 
 # pk read
@@ -108,6 +109,7 @@ select count(*) from t1;
 replace t1 set a=1,b=@b1,c=111,d=@d1;
 replace t1 set a=2,b=@b2,c=222,d=@d2;
 commit;
+--replace_column 9 #
 explain select * from t1 where a = 1;
 
 # pk read
@@ -150,6 +152,7 @@ select count(*) from t1;
 insert into t1 values(1,@b1,111,@d1);
 insert into t1 values(2,@b2,222,@d2);
 commit;
+--replace_column 9 #
 explain select * from t1 where c = 111;
 
 # hash key read
@@ -190,6 +193,7 @@ insert into t1 values(7,'b7',777,'dd7');
 insert into t1 values(8,'b8',888,'dd8');
 insert into t1 values(9,'b9',999,'dd9');
 commit;
+--replace_column 9 #
 explain select * from t1;
 
 # table scan read
@@ -210,6 +214,7 @@ select count(*) from t1;
 insert into t1 values(1,@b1,111,@d1);
 insert into t1 values(2,@b2,222,@d2);
 commit;
+--replace_column 9 #
 explain select * from t1;
 
 # table scan read
@@ -239,6 +244,7 @@ insert into t1 values(7,'b7',777,'dd7');
 insert into t1 values(8,'b8',888,'dd8');
 insert into t1 values(9,'b9',999,'dd9');
 commit;
+--replace_column 9 #
 explain select * from t1 where c >= 100 order by a;
 
 # range scan read
@@ -260,6 +266,7 @@ select count(*) from t1;
 insert into t1 values(1,@b1,111,@d1);
 insert into t1 values(2,@b2,222,@d2);
 commit;
+--replace_column 9 #
 explain select * from t1 where c >= 100 order by a;
 
 # range scan read

mysql-test/t/ndb_charset.test

@@ -159,8 +159,9 @@ 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;
+# plan too flaky
+#--replace_column 9 #
+#explain select * from t1 where a = 'zZz' order by p;
 # 2
 select * from t1 where a = 'aAa' order by p;
 # 2
@@ -187,8 +188,9 @@ 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;
+# plan too flaky
+#--replace_column 9 #
+#explain select * from t1 where a = 'zZz' order by p;
 # 6
 select * from t1 where a = 'aAa' order by p;
 # 6

mysql-test/t/ndb_condition_pushdown.test

@@ -456,6 +456,7 @@ select * from t3,t4 where t4.attr1 > 1 and t4.attr2 = t3.attr2 and t4.attr3 < 5
 set engine_condition_pushdown = on;
 
 # Test all types and compare operators
+--replace_column 9 #
 explain
 select auto from t1 where 
 string = "aaaa" and 
@@ -511,6 +512,7 @@ time_field = '01:01:01' and
 date_time = '1901-01-01 01:01:01' 
 order by auto;
 
+--replace_column 9 #
 explain
 select auto from t1 where 
 string != "aaaa" and 
@@ -566,6 +568,7 @@ time_field != '01:01:01' and
 date_time != '1901-01-01 01:01:01' 
 order by auto;
 
+--replace_column 9 #
 explain
 select auto from t1 where 
 string > "aaaa" and 
@@ -621,6 +624,7 @@ time_field > '01:01:01' and
 date_time > '1901-01-01 01:01:01'
 order by auto;
 
+--replace_column 9 #
 explain
 select auto from t1 where 
 string >= "aaaa" and 
@@ -676,6 +680,7 @@ time_field >= '01:01:01' and
 date_time >= '1901-01-01 01:01:01' 
 order by auto;
 
+--replace_column 9 #
 explain
 select auto from t1 where 
 string < "dddd" and 
@@ -731,6 +736,7 @@ time_field < '04:04:04' and
 date_time < '1904-04-04 04:04:04' 
 order by auto;
 
+--replace_column 9 #
 explain
 select auto from t1 where 
 string <= "dddd" and 
@@ -790,6 +796,7 @@ order by auto;
 create index medium_index on t1(medium);
 
 # Test all types and compare operators
+--replace_column 9 #
 explain
 select auto from t1 where 
 string = "aaaa" and 
@@ -845,6 +852,7 @@ time_field = '01:01:01' and
 date_time = '1901-01-01 01:01:01' 
 order by auto;
 
+--replace_column 9 #
 explain
 select auto from t1 where 
 string != "aaaa" and 
@@ -900,6 +908,7 @@ time_field != '01:01:01' and
 date_time != '1901-01-01 01:01:01' 
 order by auto;
 
+--replace_column 9 #
 explain
 select auto from t1 where 
 string > "aaaa" and 
@@ -955,6 +964,7 @@ time_field > '01:01:01' and
 date_time > '1901-01-01 01:01:01'
 order by auto;
 
+--replace_column 9 #
 explain
 select auto from t1 where 
 string >= "aaaa" and 
@@ -1010,6 +1020,7 @@ time_field >= '01:01:01' and
 date_time >= '1901-01-01 01:01:01' 
 order by auto;
 
+--replace_column 9 #
 explain
 select auto from t1 where 
 string < "dddd" and 
@@ -1065,6 +1076,7 @@ time_field < '04:04:04' and
 date_time < '1904-04-04 04:04:04' 
 order by auto;
 
+--replace_column 9 #
 explain
 select auto from t1 where 
 string <= "dddd" and 
@@ -1121,6 +1133,7 @@ date_time <= '1904-04-04 04:04:04'
 order by auto;
 
 # Test LIKE/NOT LIKE
+--replace_column 9 #
 explain
 select auto from t1 where 
 string like "b%" and
@@ -1136,6 +1149,7 @@ bin like concat(0xBB, '%') and
 vbin like concat(0xBB, '%')
 order by auto;
 
+--replace_column 9 #
 explain
 select auto from t1 where 
 string not like "b%" and
@@ -1152,6 +1166,7 @@ vbin not like concat(0xBB, '%')
 order by auto;
 
 # BETWEEN
+--replace_column 9 #
 explain 
 select auto from t1 where
 (string between "aaaa" and "cccc") and 
@@ -1201,6 +1216,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;
 
+--replace_column 9 #
 explain
 select auto from t1 where
 ("aaaa" between string and string) and 
@@ -1251,6 +1267,7 @@ select auto from t1 where
 order by auto;
 
 # NOT BETWEEN
+--replace_column 9 #
 explain 
 select auto from t1 where
 (string not between "aaaa" and "cccc") and 
@@ -1300,6 +1317,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;
 
+--replace_column 9 #
 explain
 select auto from t1 where
 ("aaaa" not between string and string) and 
@@ -1350,6 +1368,7 @@ select auto from t1 where
 order by auto;
 
 # IN
+--replace_column 9 #
 explain
 select auto from t1 where
 string in("aaaa","cccc") and 
@@ -1399,6 +1418,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;
 
+--replace_column 9 #
 explain
 select auto from t1 where
 "aaaa" in(string) and 
@@ -1449,6 +1469,7 @@ select auto from t1 where
 order by auto;
 
 # NOT IN
+--replace_column 9 #
 explain
 select auto from t1 where
 string not in("aaaa","cccc") and 
@@ -1498,6 +1519,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;
 
+--replace_column 9 #
 explain
 select auto from t1 where
 "aaaa" not in(string) and 
@@ -1607,35 +1629,44 @@ date_time = '1901-01-01 01:01:01';
 select count(*) from t1;
 
 # Various tests 
+--replace_column 9 #
 explain 
 select * from t2 where attr3 is null or attr1 > 2 and pk1= 3 order by pk1;
 select * from t2 where attr3 is null or attr1 > 2 and pk1= 3 order by pk1;
 
+--replace_column 9 #
 explain
 select * from t2 where attr3 is not null and attr1 > 2 order by pk1;
 select * from t2 where attr3 is not null and attr1 > 2 order by pk1;
 
+--replace_column 9 #
 explain
 select * from t3 where attr2 >  9223372036854775803 and attr3 != 3 order by pk1;
 select * from t3 where attr2 >  9223372036854775803 and attr3 != 3 order by pk1;
 
+--replace_column 9 #
 explain
 select * from t2,t3 where t2.attr1 < 1 and t2.attr2 = t3.attr2 and t3.attr1 < 5 order by t2.pk1;
 select * from t2,t3 where t2.attr1 < 1 and t2.attr2 = t3.attr2 and t3.attr1 < 5 order by t2.pk1;
 
+--replace_column 9 #
 explain
 select * from t4 where attr1 < 5 and attr2 > 9223372036854775803 and attr3 != 3 order by t4.pk1;
 select * from t4 where attr1 < 5 and attr2 > 9223372036854775803 and attr3 != 3 order by t4.pk1;
 
+--replace_column 9 #
 explain
 select * from t3,t4 where t4.attr1 > 1 and t4.attr2 = t3.attr2 and t4.attr3 < 5 order by t4.pk1;
 select * from t3,t4 where t4.attr1 > 1 and t4.attr2 = t3.attr2 and t4.attr3 < 5 order by t4.pk1;
 
 # Some tests that are currently not supported and should not push condition
+--replace_column 9 #
 explain
 select auto from t1 where string = "aaaa" collate latin1_general_ci order by auto;
+--replace_column 9 #
 explain
 select * from t2 where (attr1 < 2) = (attr2 < 2) order by pk1;
+--replace_column 9 #
 explain
 select * from t3 left join t4 on t4.attr2 = t3.attr2 where t4.attr1 > 1 and t4.attr3 < 5 or t4.attr1 is null order by t4.pk1;
 
@@ -1646,6 +1677,7 @@ insert into t5 (b) values ('jonas'), ('jensing'), ('johan');
 set engine_condition_pushdown = off;
 select * from t5 where b like '%jo%' order by a;
 set engine_condition_pushdown = on;
+--replace_column 9 #
 explain select * from t5 where b like '%jo%';
 select * from t5 where b like '%jo%' order by a;
 
@@ -1654,8 +1686,10 @@ set engine_condition_pushdown = off;
 select auto from t1 where date_time like '1902-02-02 %' order by auto;
 select auto from t1 where date_time not like '1902-02-02 %' order by auto;
 set engine_condition_pushdown = on;
+--replace_column 9 #
 explain select auto from t1 where date_time like '1902-02-02 %';
 select auto from t1 where date_time like '1902-02-02 %' order by auto;
+--replace_column 9 #
 explain select auto from t1 where date_time not like '1902-02-02 %';
 select auto from t1 where date_time not like '1902-02-02 %' order by auto;
 

mysql-test/t/ndb_dd_sql_features.test

@@ -480,6 +480,7 @@ select length(@x0),length(@b2),length(@d2) from dual;
  
  INSERT INTO test.t1 values(1,@vc1,@d1);
  INSERT INTO test.t1 values(2,@vc2,@d2);
+ --replace_column 9 #
  explain SELECT * from test.t1 WHERE a1 = 1;
 
  SELECT a1,length(a2),substr(a2,1+2*900,2),length(a3),substr(a3,1+3*900,3)

mysql-test/t/ndb_subquery.test

@@ -18,18 +18,22 @@ insert into t1 values (1,1,1),(2,2,2),(3,3,3);
 insert into t2 values (1,1,1),(2,2,2),(3,3,3), (4,4,4), (5,5,5);
 
 # Use pk
+--replace_column 9 #
 explain select * from t2 where p NOT IN (select p from t1);
 select * from t2 where p NOT IN (select p from t1) order by p;
 
 # Use unique index
+--replace_column 9 #
 explain select * from t2 where p NOT IN (select u from t1);
 select * from t2 where p NOT IN (select u from t1) order by p;
 
 # Use ordered index
+--replace_column 9 #
 explain select * from t2 where p NOT IN (select o from t1);
 select * from t2 where p NOT IN (select o from t1) order by p;
 
 # Use scan
+--replace_column 9 #
 explain select * from t2 where p NOT IN (select p+0 from t1);
 select * from t2 where p NOT IN (select p+0 from t1) order by p;
 

sql/ha_ndbcluster.cc

@@ -10491,7 +10491,7 @@ static int ndbcluster_fill_files_table(handlerton *hton,
 
     while ((id= g_ndb_cluster_connection->get_next_node(iter)))
     {
-      uint c= 0;
+      init_fill_schema_files_row(table);
       NdbDictionary::Datafile df= dict->getDatafile(id, elt.name);
       ndberr= dict->getNdbError();
       if(ndberr.classification != NdbError::NoError)
@@ -10509,76 +10509,48 @@ static int ndbcluster_fill_files_table(handlerton *hton,
         ERR_RETURN(ndberr);
       }
 
-      table->field[c++]->set_null(); // FILE_ID
-      table->field[c]->set_notnull();
-      table->field[c++]->store(elt.name, strlen(elt.name),
+      table->field[IS_FILES_FILE_NAME]->set_notnull();
+      table->field[IS_FILES_FILE_NAME]->store(elt.name, strlen(elt.name),
                                               system_charset_info);
-      table->field[c]->set_notnull();
-      table->field[c++]->store("DATAFILE",8,system_charset_info);
-      table->field[c]->set_notnull();
-      table->field[c++]->store(df.getTablespace(), strlen(df.getTablespace()),
+      table->field[IS_FILES_FILE_TYPE]->set_notnull();
+      table->field[IS_FILES_FILE_TYPE]->store("DATAFILE",8,
                                               system_charset_info);
-      table->field[c++]->set_null(); // TABLE_CATALOG
-      table->field[c++]->set_null(); // TABLE_SCHEMA
-      table->field[c++]->set_null(); // TABLE_NAME
-
-      // LOGFILE_GROUP_NAME
-      table->field[c]->set_notnull();
-      table->field[c++]->store(ts.getDefaultLogfileGroup(),
+      table->field[IS_FILES_TABLESPACE_NAME]->set_notnull();
+      table->field[IS_FILES_TABLESPACE_NAME]->store(df.getTablespace(),
+                                                    strlen(df.getTablespace()),
+                                                    system_charset_info);
+      table->field[IS_FILES_LOGFILE_GROUP_NAME]->set_notnull();
+      table->field[IS_FILES_LOGFILE_GROUP_NAME]->
+        store(ts.getDefaultLogfileGroup(),
               strlen(ts.getDefaultLogfileGroup()),
               system_charset_info);
-      table->field[c++]->set_null(); // LOGFILE_GROUP_NUMBER
-      table->field[c]->set_notnull();
-      table->field[c++]->store(ndbcluster_hton_name,
+      table->field[IS_FILES_ENGINE]->set_notnull();
+      table->field[IS_FILES_ENGINE]->store(ndbcluster_hton_name,
                                            ndbcluster_hton_name_length,
-                               system_charset_info); // ENGINE
-
-      table->field[c++]->set_null(); // FULLTEXT_KEYS
-      table->field[c++]->set_null(); // DELETED_ROWS
-      table->field[c++]->set_null(); // UPDATE_COUNT
-      table->field[c]->set_notnull();
-      table->field[c++]->store(df.getFree() / ts.getExtentSize()); // FREE_EXTENTS
-      table->field[c]->set_notnull();
-      table->field[c++]->store(df.getSize() / ts.getExtentSize()); // TOTAL_EXTENTS
-      table->field[c]->set_notnull();
-      table->field[c++]->store(ts.getExtentSize()); // EXTENT_SIZE
-
-      table->field[c]->set_notnull();
-      table->field[c++]->store(df.getSize()); // INITIAL_SIZE
-      table->field[c]->set_notnull();
-      table->field[c++]->store(df.getSize()); // MAXIMUM_SIZE
-      table->field[c++]->set_null(); // AUTOEXTEND_SIZE
-
-      table->field[c++]->set_null(); // CREATION_TIME
-      table->field[c++]->set_null(); // LAST_UPDATE_TIME
-      table->field[c++]->set_null(); // LAST_ACCESS_TIME
-      table->field[c++]->set_null(); // RECOVER_TIME
-      table->field[c++]->set_null(); // TRANSACTION_COUNTER
-
-      table->field[c]->set_notnull();
-      table->field[c++]->store(df.getObjectVersion()); // VERSION
-
-      table->field[c]->set_notnull();
-      table->field[c++]->store("FIXED", 5, system_charset_info); // ROW_FORMAT
-
-      table->field[c++]->set_null(); // TABLE_ROWS
-      table->field[c++]->set_null(); // AVG_ROW_LENGTH
-      table->field[c++]->set_null(); // DATA_LENGTH
-      table->field[c++]->set_null(); // MAX_DATA_LENGTH
-      table->field[c++]->set_null(); // INDEX_LENGTH
-      table->field[c++]->set_null(); // DATA_FREE
-      table->field[c++]->set_null(); // CREATE_TIME
-      table->field[c++]->set_null(); // UPDATE_TIME
-      table->field[c++]->set_null(); // CHECK_TIME
-      table->field[c++]->set_null(); // CHECKSUM
-
-      table->field[c]->set_notnull();
-      table->field[c++]->store("NORMAL", 6, system_charset_info);
+                                           system_charset_info);
+
+      table->field[IS_FILES_FREE_EXTENTS]->set_notnull();
+      table->field[IS_FILES_FREE_EXTENTS]->store(df.getFree()
+                                                 / ts.getExtentSize());
+      table->field[IS_FILES_TOTAL_EXTENTS]->set_notnull();
+      table->field[IS_FILES_TOTAL_EXTENTS]->store(df.getSize()
+                                                  / ts.getExtentSize());
+      table->field[IS_FILES_EXTENT_SIZE]->set_notnull();
+      table->field[IS_FILES_EXTENT_SIZE]->store(ts.getExtentSize());
+      table->field[IS_FILES_INITIAL_SIZE]->set_notnull();
+      table->field[IS_FILES_INITIAL_SIZE]->store(df.getSize());
+      table->field[IS_FILES_MAXIMUM_SIZE]->set_notnull();
+      table->field[IS_FILES_MAXIMUM_SIZE]->store(df.getSize());
+      table->field[IS_FILES_VERSION]->set_notnull();
+      table->field[IS_FILES_VERSION]->store(df.getObjectVersion());
+
+      table->field[IS_FILES_ROW_FORMAT]->set_notnull();
+      table->field[IS_FILES_ROW_FORMAT]->store("FIXED", 5, system_charset_info);
 
       char extra[30];
       int len= my_snprintf(extra, sizeof(extra), "CLUSTER_NODE=%u", id);
-      table->field[c]->store(extra, len, system_charset_info);
-      table->field[c]->set_notnull();
+      table->field[IS_FILES_EXTRA]->set_notnull();
+      table->field[IS_FILES_EXTRA]->store(extra, len, system_charset_info);
       schema_table_store_record(thd, table);
     }
   }
@@ -10617,76 +10589,43 @@ static int ndbcluster_fill_files_table(handlerton *hton,
         ERR_RETURN(ndberr);
       }
 
-      int c= 0;
-      table->field[c++]->set_null(); // FILE_ID
-      table->field[c]->set_notnull();
-      table->field[c++]->store(elt.name, strlen(elt.name),
+      init_fill_schema_files_row(table);
+      table->field[IS_FILES_FILE_NAME]->set_notnull();
+      table->field[IS_FILES_FILE_NAME]->store(elt.name, strlen(elt.name),
+                                              system_charset_info);
+      table->field[IS_FILES_FILE_TYPE]->set_notnull();
+      table->field[IS_FILES_FILE_TYPE]->store("UNDO LOG", 8,
                                               system_charset_info);
-      table->field[c]->set_notnull();
-      table->field[c++]->store("UNDO LOG", 8, system_charset_info);
-      table->field[c++]->set_null(); // TABLESPACE NAME
-      table->field[c++]->set_null(); // TABLE_CATALOG
-      table->field[c++]->set_null(); // TABLE_SCHEMA
-      table->field[c++]->set_null(); // TABLE_NAME
-
-      // LOGFILE_GROUP_NAME
       NdbDictionary::ObjectId objid;
       uf.getLogfileGroupId(&objid);
-      table->field[c]->set_notnull();
-      table->field[c++]->store(uf.getLogfileGroup(),
+      table->field[IS_FILES_LOGFILE_GROUP_NAME]->set_notnull();
+      table->field[IS_FILES_LOGFILE_GROUP_NAME]->store(uf.getLogfileGroup(),
                                                   strlen(uf.getLogfileGroup()),
                                                        system_charset_info);
-      table->field[c]->set_notnull();
-      table->field[c++]->store(objid.getObjectId()); // LOGFILE_GROUP_NUMBER
-      table->field[c]->set_notnull();
-      table->field[c++]->store(ndbcluster_hton_name,
+      table->field[IS_FILES_LOGFILE_GROUP_NUMBER]->set_notnull();
+      table->field[IS_FILES_LOGFILE_GROUP_NUMBER]->store(objid.getObjectId());
+      table->field[IS_FILES_ENGINE]->set_notnull();
+      table->field[IS_FILES_ENGINE]->store(ndbcluster_hton_name,
                                            ndbcluster_hton_name_length,
-                               system_charset_info); // ENGINE
-
-      table->field[c++]->set_null(); // FULLTEXT_KEYS
-      table->field[c++]->set_null(); // DELETED_ROWS
-      table->field[c++]->set_null(); // UPDATE_COUNT
-      table->field[c++]->set_null(); // FREE_EXTENTS
-      table->field[c]->set_notnull();
-      table->field[c++]->store(uf.getSize()/4); // TOTAL_EXTENTS
-      table->field[c]->set_notnull();
-      table->field[c++]->store(4); // EXTENT_SIZE
-
-      table->field[c]->set_notnull();
-      table->field[c++]->store(uf.getSize()); // INITIAL_SIZE
-      table->field[c]->set_notnull();
-      table->field[c++]->store(uf.getSize()); // MAXIMUM_SIZE
-      table->field[c++]->set_null(); // AUTOEXTEND_SIZE
-
-      table->field[c++]->set_null(); // CREATION_TIME
-      table->field[c++]->set_null(); // LAST_UPDATE_TIME
-      table->field[c++]->set_null(); // LAST_ACCESS_TIME
-      table->field[c++]->set_null(); // RECOVER_TIME
-      table->field[c++]->set_null(); // TRANSACTION_COUNTER
-
-      table->field[c]->set_notnull();
-      table->field[c++]->store(uf.getObjectVersion()); // VERSION
-
-      table->field[c++]->set_null(); // ROW FORMAT
-
-      table->field[c++]->set_null(); // TABLE_ROWS
-      table->field[c++]->set_null(); // AVG_ROW_LENGTH
-      table->field[c++]->set_null(); // DATA_LENGTH
-      table->field[c++]->set_null(); // MAX_DATA_LENGTH
-      table->field[c++]->set_null(); // INDEX_LENGTH
-      table->field[c++]->set_null(); // DATA_FREE
-      table->field[c++]->set_null(); // CREATE_TIME
-      table->field[c++]->set_null(); // UPDATE_TIME
-      table->field[c++]->set_null(); // CHECK_TIME
-      table->field[c++]->set_null(); // CHECKSUM
-
-      table->field[c]->set_notnull();
-      table->field[c++]->store("NORMAL", 6, system_charset_info);
+                                           system_charset_info);
+
+      table->field[IS_FILES_TOTAL_EXTENTS]->set_notnull();
+      table->field[IS_FILES_TOTAL_EXTENTS]->store(uf.getSize()/4);
+      table->field[IS_FILES_EXTENT_SIZE]->set_notnull();
+      table->field[IS_FILES_EXTENT_SIZE]->store(4);
+
+      table->field[IS_FILES_INITIAL_SIZE]->set_notnull();
+      table->field[IS_FILES_INITIAL_SIZE]->store(uf.getSize());
+      table->field[IS_FILES_MAXIMUM_SIZE]->set_notnull();
+      table->field[IS_FILES_MAXIMUM_SIZE]->store(uf.getSize());
+
+      table->field[IS_FILES_VERSION]->set_notnull();
+      table->field[IS_FILES_VERSION]->store(uf.getObjectVersion());
 
       char extra[100];
       int len= my_snprintf(extra,sizeof(extra),"CLUSTER_NODE=%u;UNDO_BUFFER_SIZE=%lu",id,lfg.getUndoBufferSize());
-      table->field[c]->set_notnull();
-      table->field[c]->store(extra, len, system_charset_info);
+      table->field[IS_FILES_EXTRA]->set_notnull();
+      table->field[IS_FILES_EXTRA]->store(extra, len, system_charset_info);
       schema_table_store_record(thd, table);
     }
   }
@@ -10712,69 +10651,36 @@ static int ndbcluster_fill_files_table(handlerton *hton,
       ERR_RETURN(ndberr);
     }
 
-    int c= 0;
-    table->field[c++]->set_null(); // FILE_ID
-    table->field[c++]->set_null(); // name
-    table->field[c]->set_notnull();
-    table->field[c++]->store("UNDO LOG", 8, system_charset_info);
-    table->field[c++]->set_null(); // TABLESPACE NAME
-    table->field[c++]->set_null(); // TABLE_CATALOG
-    table->field[c++]->set_null(); // TABLE_SCHEMA
-    table->field[c++]->set_null(); // TABLE_NAME
+    init_fill_schema_files_row(table);
+    table->field[IS_FILES_FILE_TYPE]->set_notnull();
+    table->field[IS_FILES_FILE_TYPE]->store("UNDO LOG", 8,
+                                            system_charset_info);
 
-    // LOGFILE_GROUP_NAME
-    table->field[c]->set_notnull();
-    table->field[c++]->store(elt.name, strlen(elt.name),
+    table->field[IS_FILES_LOGFILE_GROUP_NAME]->set_notnull();
+    table->field[IS_FILES_LOGFILE_GROUP_NAME]->store(elt.name,
+                                                     strlen(elt.name),
                                                      system_charset_info);
-    table->field[c]->set_notnull();
-    table->field[c++]->store(lfg.getObjectId()); // LOGFILE_GROUP_NUMBER
-    table->field[c]->set_notnull();
-    table->field[c++]->store(ndbcluster_hton_name,
+    table->field[IS_FILES_LOGFILE_GROUP_NUMBER]->set_notnull();
+    table->field[IS_FILES_LOGFILE_GROUP_NUMBER]->store(lfg.getObjectId());
+    table->field[IS_FILES_ENGINE]->set_notnull();
+    table->field[IS_FILES_ENGINE]->store(ndbcluster_hton_name,
                                          ndbcluster_hton_name_length,
-                             system_charset_info); // ENGINE
-
-    table->field[c++]->set_null(); // FULLTEXT_KEYS
-    table->field[c++]->set_null(); // DELETED_ROWS
-    table->field[c++]->set_null(); // UPDATE_COUNT
-    table->field[c]->set_notnull();
-    table->field[c++]->store(lfg.getUndoFreeWords()); // FREE_EXTENTS
-    table->field[c++]->set_null(); //store(uf.getSize()/4); // TOTAL_EXTENTS
-    table->field[c]->set_notnull();
-    table->field[c++]->store(4); // EXTENT_SIZE
-
-    table->field[c++]->set_null();//store(uf.getSize()); // INITIAL_SIZE
-    table->field[c++]->set_null(); //store(uf.getSize()); // MAXIMUM_SIZE
-    table->field[c++]->set_null(); // AUTOEXTEND_SIZE
-
-    table->field[c++]->set_null(); // CREATION_TIME
-    table->field[c++]->set_null(); // LAST_UPDATE_TIME
-    table->field[c++]->set_null(); // LAST_ACCESS_TIME
-    table->field[c++]->set_null(); // RECOVER_TIME
-    table->field[c++]->set_null(); // TRANSACTION_COUNTER
-
-    table->field[c]->set_notnull();
-    table->field[c++]->store(lfg.getObjectVersion()); // VERSION
-
-    table->field[c++]->set_null(); // ROW FORMAT
-
-    table->field[c++]->set_null(); // TABLE_ROWS
-    table->field[c++]->set_null(); // AVG_ROW_LENGTH
-    table->field[c++]->set_null(); // DATA_LENGTH
-    table->field[c++]->set_null(); // MAX_DATA_LENGTH
-    table->field[c++]->set_null(); // INDEX_LENGTH
-    table->field[c++]->set_null(); // DATA_FREE
-    table->field[c++]->set_null(); // CREATE_TIME
-    table->field[c++]->set_null(); // UPDATE_TIME
-    table->field[c++]->set_null(); // CHECK_TIME
-    table->field[c++]->set_null(); // CHECKSUM
-
-    table->field[c]->set_notnull();
-    table->field[c++]->store("NORMAL", 6, system_charset_info);
+                                         system_charset_info);
+
+    table->field[IS_FILES_FREE_EXTENTS]->set_notnull();
+    table->field[IS_FILES_FREE_EXTENTS]->store(lfg.getUndoFreeWords());
+    table->field[IS_FILES_EXTENT_SIZE]->set_notnull();
+    table->field[IS_FILES_EXTENT_SIZE]->store(4);
+
+    table->field[IS_FILES_VERSION]->set_notnull();
+    table->field[IS_FILES_VERSION]->store(lfg.getObjectVersion());
 
     char extra[100];
-    int len= my_snprintf(extra,sizeof(extra),"UNDO_BUFFER_SIZE=%lu",id,lfg.getUndoBufferSize());
-    table->field[c]->set_notnull();
-    table->field[c]->store(extra, len, system_charset_info);
+    int len= my_snprintf(extra,sizeof(extra),
+                         "UNDO_BUFFER_SIZE=%lu",
+                         lfg.getUndoBufferSize());
+    table->field[IS_FILES_EXTRA]->set_notnull();
+    table->field[IS_FILES_EXTRA]->store(extra, len, system_charset_info);
     schema_table_store_record(thd, table);
   }
   DBUG_RETURN(0);

sql/mysql_priv.h

@@ -2007,6 +2007,45 @@ inline void kill_delayed_threads(void) {}
 #endif
 
 /* Used by handlers to store things in schema tables */
+#define IS_FILES_FILE_ID              0
+#define IS_FILES_FILE_NAME            1
+#define IS_FILES_FILE_TYPE            2
+#define IS_FILES_TABLESPACE_NAME      3
+#define IS_FILES_TABLE_CATALOG        4
+#define IS_FILES_TABLE_SCHEMA         5
+#define IS_FILES_TABLE_NAME           6
+#define IS_FILES_LOGFILE_GROUP_NAME   7
+#define IS_FILES_LOGFILE_GROUP_NUMBER 8
+#define IS_FILES_ENGINE               9
+#define IS_FILES_FULLTEXT_KEYS       10
+#define IS_FILES_DELETED_ROWS        11
+#define IS_FILES_UPDATE_COUNT        12
+#define IS_FILES_FREE_EXTENTS        13
+#define IS_FILES_TOTAL_EXTENTS       14
+#define IS_FILES_EXTENT_SIZE         15
+#define IS_FILES_INITIAL_SIZE        16
+#define IS_FILES_MAXIMUM_SIZE        17
+#define IS_FILES_AUTOEXTEND_SIZE     18
+#define IS_FILES_CREATION_TIME       19
+#define IS_FILES_LAST_UPDATE_TIME    20
+#define IS_FILES_LAST_ACCESS_TIME    21
+#define IS_FILES_RECOVER_TIME        22
+#define IS_FILES_TRANSACTION_COUNTER 23
+#define IS_FILES_VERSION             24
+#define IS_FILES_ROW_FORMAT          25
+#define IS_FILES_TABLE_ROWS          26
+#define IS_FILES_AVG_ROW_LENGTH      27
+#define IS_FILES_DATA_LENGTH         28
+#define IS_FILES_MAX_DATA_LENGTH     29
+#define IS_FILES_INDEX_LENGTH        30
+#define IS_FILES_DATA_FREE           31
+#define IS_FILES_CREATE_TIME         32
+#define IS_FILES_UPDATE_TIME         33
+#define IS_FILES_CHECK_TIME          34
+#define IS_FILES_CHECKSUM            35
+#define IS_FILES_STATUS              36
+#define IS_FILES_EXTRA               37
+void init_fill_schema_files_row(TABLE* table);
 bool schema_table_store_record(THD *thd, TABLE *table);
 
 #endif /* MYSQL_SERVER */

sql/mysqld.cc

@@ -4704,7 +4704,6 @@ enum options_mysqld
   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_NDB_EXTRA_LOGGING,
   OPT_NDB_REPORT_THRESH_BINLOG_EPOCH_SLIP,
   OPT_NDB_REPORT_THRESH_BINLOG_MEM_USAGE,
@@ -5414,19 +5413,7 @@ Disable with --skip-ndbcluster (will save memory).",
    "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},
+   0, GET_BOOL, OPT_ARG, 0, 0, 1, 0, 0, 0},
 #endif
   {"ndb-use-copying-alter-table",
    OPT_NDB_USE_COPYING_ALTER_TABLE,
@@ -7055,7 +7042,7 @@ static void mysql_init_variables(void)
 #endif
 #ifdef WITH_NDBCLUSTER_STORAGE_ENGINE
   have_ndbcluster=SHOW_OPTION_DISABLED;
-  global_system_variables.ndb_index_stat_enable=TRUE;
+  global_system_variables.ndb_index_stat_enable=FALSE;
   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;

sql/sql_show.cc

@@ -5642,6 +5642,16 @@ ST_FIELD_INFO files_fields_info[]=
   {0, 0, MYSQL_TYPE_STRING, 0, 0, 0}
 };
 
+void init_fill_schema_files_row(TABLE* table)
+{
+  int i;
+  for(i=0; files_fields_info[i].field_name!=NULL; i++)
+    table->field[i]->set_null();
+
+  table->field[IS_FILES_STATUS]->set_notnull();
+  table->field[IS_FILES_STATUS]->store("NORMAL", 6, system_charset_info);
+}
+
 ST_FIELD_INFO referential_constraints_fields_info[]=
 {
   {"CONSTRAINT_CATALOG", FN_REFLEN, MYSQL_TYPE_STRING, 0, 1, 0},

storage/ndb/src/kernel/blocks/ERROR_codes.txt

@@ -8,7 +8,7 @@ Next DBDICT 6007
 Next DBDIH 7178
 Next DBTC 8038
 Next CMVMI 9000
-Next BACKUP 10022
+Next BACKUP 10036
 Next DBUTIL 11002
 Next DBTUX 12008
 Next SUMA 13001
@@ -411,6 +411,11 @@ Backup Stuff:
 10028: Abort backup by error at reception of BACKUP_FRAGMENT_CONF at master (code 305)
 10029: Abort backup by error at reception of FSAPPENDCONF in slave (FileOrScanError = 5)
 10030: Simulate buffer full from trigger execution => abort backup
+10031: Error 331 for dictCommitTableMutex_locked
+10032: backup checkscan
+10033: backup checkscan
+10034: define backup reply error
+10035: Fail to allocate buffers
 
 11001: Send UTIL_SEQUENCE_REF (in master)
 

storage/ndb/src/kernel/blocks/backup/Backup.cpp

@@ -2004,7 +2004,15 @@ Backup::sendDropTrig(Signal* signal, BackupRecordPtr ptr)
 
   if (ptr.p->slaveData.dropTrig.tableId == RNIL) {
     jam();
+    if(ptr.p->tables.count())
       ptr.p->tables.first(tabPtr);
+    else
+    {
+      // Early abort, go to close files
+      jam();
+      closeFiles(signal, ptr);
+      return;
+    }
   } else {
     jam();
     ndbrequire(findTable(ptr, tabPtr, ptr.p->slaveData.dropTrig.tableId));
@@ -2105,8 +2113,11 @@ Backup::execDROP_TRIG_REF(Signal* signal)
   BackupRecordPtr ptr;
   c_backupPool.getPtr(ptr, ptrI);
 
+  if(ref->getConf()->getTriggerId() != -1)
+  {
     ndbout << "ERROR DROPPING TRIGGER: " << ref->getConf()->getTriggerId();
     ndbout << " Err: " << (Uint32)ref->getErrorCode() << endl << endl;
+  }
 
   dropTrigReply(signal, ptr);
 }
@@ -2539,7 +2550,8 @@ Backup::execDEFINE_BACKUP_REQ(Signal* signal)
     files[i].p->m_flags = 0;
     files[i].p->errorCode = 0;
 
-    if(files[i].p->pages.seize(noOfPages[i]) == false) {
+    if(ERROR_INSERTED(10035) || files[i].p->pages.seize(noOfPages[i]) == false)
+    {
       jam();
       DEBUG_OUT("Failed to seize " << noOfPages[i] << " pages");
       defineBackupRef(signal, ptr, DefineBackupRef::FailedToAllocateBuffers);
@@ -4451,14 +4463,24 @@ Backup::closeFilesDone(Signal* signal, BackupRecordPtr ptr)
   }
   
   jam();
-  BackupFilePtr filePtr;
-  ptr.p->files.getPtr(filePtr, ptr.p->logFilePtr);
 
   StopBackupConf* conf = (StopBackupConf*)signal->getDataPtrSend();
   conf->backupId = ptr.p->backupId;
   conf->backupPtr = ptr.i;
-  conf->noOfLogBytes = filePtr.p->operation.noOfBytes;
-  conf->noOfLogRecords = filePtr.p->operation.noOfRecords;
+
+  BackupFilePtr filePtr;
+  if(ptr.p->logFilePtr != RNIL)
+  {
+    ptr.p->files.getPtr(filePtr, ptr.p->logFilePtr);
+    conf->noOfLogBytes= filePtr.p->operation.noOfBytes;
+    conf->noOfLogRecords= filePtr.p->operation.noOfRecords;
+  }
+  else
+  {
+    conf->noOfLogBytes= 0;
+    conf->noOfLogRecords= 0;
+  }
+
   sendSignal(ptr.p->masterRef, GSN_STOP_BACKUP_CONF, signal,
 	     StopBackupConf::SignalLength, JBB);
   

storage/ndb/src/kernel/blocks/backup/Backup.hpp

@@ -427,6 +427,7 @@ public:
       : slaveState(b, validSlaveTransitions, validSlaveTransitionsCount,1)
       , tables(tp), triggers(trp), files(bp)
       , masterData(b), backup(b)
+      , ctlFilePtr(RNIL), logFilePtr(RNIL), dataFilePtr(RNIL)
       {
       }
     

storage/ndb/test/ndbapi/Makefile.am

@@ -36,7 +36,8 @@ testPartitioning \
 testBitfield \
 DbCreate DbAsyncGenerator \
 testSRBank \
-test_event_merge
+test_event_merge \
+testIndexStat
 
 EXTRA_PROGRAMS = \
  test_event \
@@ -87,6 +88,7 @@ DbAsyncGenerator_SOURCES = bench/mainAsyncGenerator.cpp bench/asyncGenerator.cpp
 testSRBank_SOURCES = testSRBank.cpp
 test_event_merge_SOURCES = test_event_merge.cpp
 test_event_multi_table_SOURCES = test_event_multi_table.cpp
+testIndexStat_SOURCES = testIndexStat.cpp
 
 INCLUDES_LOC = -I$(top_srcdir)/storage/ndb/include/kernel
 
@@ -168,3 +170,4 @@ testScan.dsp: Makefile \
 	@$(top_srcdir)/storage/ndb/config/win-sources $@ $(testScan_SOURCES)
 	@$(top_srcdir)/storage/ndb/config/win-libraries $@ LINK $(LDADD)
 
+

storage/ndb/test/ndbapi/testIndexStat.cpp

+/* Copyright (C) 2005 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 <ndb_opts.h>
+#include <NdbApi.hpp>
+#include <NdbIndexStat.hpp>
+#include <NdbTest.hpp>
+#include <my_sys.h>
+#include <ndb_version.h>
+#include <math.h>
+
+/*
+ * Sample results:
+ *
+ * 0. err pct: count: 1000 min: -99.99 max: 99.92 avg: 6.88 stddev: 27.61
+ *
+ * 0. baseline with same options as handler
+ */
+
+#undef min
+#undef max
+#define min(a, b) ((a) <= (b) ? (a) : (b))
+#define max(a, b) ((a) >= (b) ? (a) : (b))
+
+inline NdbOut&
+NdbOut::operator<<(double x)
+{
+  char buf[100];
+  sprintf(buf, "%.2f", x);
+  *this << buf;
+  return *this;
+}
+
+struct Opts {
+  int loglevel;
+  uint seed;
+  uint loop;
+  uint rows;
+  uint ops;
+  uint nullkeys;
+  uint dupkeys;
+  uint scanpct;
+  uint eqscans;
+  uint dupscans;
+  my_bool keeptable;
+  my_bool loaddata;
+  my_bool nochecks;
+  my_bool abort;
+  // internal
+  uint tryhard;
+  Opts() :
+    loglevel(0),
+    seed(-1),
+    loop(1),
+    rows(100000),
+    ops(1000),
+    nullkeys(10),
+    dupkeys(1000),
+    scanpct(5),
+    eqscans(50),
+    dupscans(10),
+    keeptable(false),
+    loaddata(true),
+    nochecks(false),
+    abort(false),
+    // internal
+    tryhard(20)
+  {}
+};
+
+static Opts g_opts;
+const char* g_progname = "testIndexStat";
+static uint g_loop = 0;
+
+static const char* g_tabname = "ts0";
+static const char* g_indname = "ts0x1";
+static const char g_numattrs = 3;
+static const uint g_charlen = 10;
+static const char* g_csname = "latin1_swedish_ci";
+static CHARSET_INFO* g_cs;
+
+// value and bound ranges
+static uint g_val_b_max = 10;
+static uint g_bnd_b_max = 20;
+static const char* g_val_c_char = "bcd";
+static const char* g_bnd_c_char = "abcde";
+static uint g_val_d_max = 100;
+static uint g_bnd_d_max = 200;
+
+static Ndb_cluster_connection* g_ncc = 0;
+static Ndb* g_ndb = 0;
+static NdbDictionary::Dictionary* g_dic = 0;
+static const NdbDictionary::Table* g_tab = 0;
+static const NdbDictionary::Index* g_ind = 0;
+
+static NdbIndexStat* g_stat = 0;
+
+static NdbTransaction* g_con = 0;
+static NdbOperation* g_op = 0;
+static NdbScanOperation* g_scan_op = 0;
+static NdbIndexScanOperation* g_rangescan_op = 0;
+
+static uint
+urandom()
+{
+  uint r = (uint)random();
+  return r;
+}
+
+static uint
+urandom(uint m)
+{
+  if (m == 0)
+    return 0;
+  uint r = urandom();
+  r = r % m;
+  return r;
+}
+
+static int& g_loglevel = g_opts.loglevel; // default log level
+
+#define chkdb(x) \
+  do { if (likely(x)) break; ndbout << "line " << __LINE__ << " FAIL " << #x << endl; errdb(); if (g_opts.abort) abort(); return -1; } while (0)
+
+#define chkrc(x) \
+  do { if (likely(x)) break; ndbout << "line " << __LINE__ << " FAIL " << #x << endl; if (g_opts.abort) abort(); return -1; } while (0)
+
+#define reqrc(x) \
+  do { if (likely(x)) break; ndbout << "line " << __LINE__ << " ASSERT " << #x << endl; abort(); } while (0)
+
+#define llx(n, x) \
+  do { if (likely(g_loglevel < n)) break; ndbout << x << endl; } while (0)
+
+#define ll0(x) llx(0, x)
+#define ll1(x) llx(1, x)
+#define ll2(x) llx(2, x)
+#define ll3(x) llx(3, x)
+
+static void
+errdb()
+{
+  uint any = 0;
+  // g_ncc return no error...
+  if (g_ndb != 0) {
+    const NdbError& e = g_ndb->getNdbError();
+    if (e.code != 0)
+      ll0(++any << " ndb: error " << e);
+  }
+  if (g_dic != 0) {
+    const NdbError& e = g_dic->getNdbError();
+    if (e.code != 0)
+      ll0(++any << " dic: error " << e);
+  }
+  if (g_con != 0) {
+    const NdbError& e = g_con->getNdbError();
+    if (e.code != 0)
+      ll0(++any << " con: error " << e);
+  }
+  if (g_op != 0) {
+    const NdbError& e = g_op->getNdbError();
+    if (e.code != 0)
+      ll0(++any << " op: error " << e);
+  }
+  if (g_scan_op != 0) {
+    const NdbError& e = g_scan_op->getNdbError();
+    if (e.code != 0)
+      ll0(++any << " scan_op: error " << e);
+  }
+  if (g_rangescan_op != 0) {
+    const NdbError& e = g_rangescan_op->getNdbError();
+    if (e.code != 0)
+      ll0(++any << " rangescan_op: error " << e);
+  }
+  if (g_stat != 0) {
+    const NdbError& e = g_stat->getNdbError();
+    if (e.code != 0)
+      ll0(++any << " stat: error " << e);
+  }
+  if (! any)
+    ll0("unknown db error");
+}
+
+// create table ts0 (
+//   a int unsigned, b smallint not null, c varchar(10), d int unsigned,
+//   primary key using hash (a), index (b, c, d) )
+
+static int
+createtable()
+{
+  NdbDictionary::Table tab(g_tabname);
+  tab.setLogging(false);
+  {
+    NdbDictionary::Column col("a");
+    col.setType(NdbDictionary::Column::Unsigned);
+    col.setPrimaryKey(true);
+    tab.addColumn(col);
+  }
+  {
+    NdbDictionary::Column col("b");
+    col.setType(NdbDictionary::Column::Smallint);
+    col.setNullable(false);
+    tab.addColumn(col);
+  }
+  {
+    NdbDictionary::Column col("c");
+    col.setType(NdbDictionary::Column::Varchar);
+    col.setLength(g_charlen);
+    col.setCharset(g_cs);
+    col.setNullable(true);
+    tab.addColumn(col);
+  }
+  {
+    NdbDictionary::Column col("d");
+    col.setType(NdbDictionary::Column::Unsigned);
+    col.setNullable(true);
+    tab.addColumn(col);
+  }
+  NdbDictionary::Index ind(g_indname);
+  ind.setTable(g_tabname);
+  ind.setType(NdbDictionary::Index::OrderedIndex);
+  ind.setLogging(false);
+  ind.addColumnName("b");
+  ind.addColumnName("c");
+  ind.addColumnName("d");
+  g_dic = g_ndb->getDictionary();
+  if (! g_opts.keeptable) {
+    if (g_dic->getTable(g_tabname) != 0)
+      chkdb(g_dic->dropTable(g_tabname) == 0);
+    chkdb(g_dic->createTable(tab) == 0);
+    chkdb(g_dic->createIndex(ind) == 0);
+  } else {
+    if (g_dic->getTable(g_tabname) == 0) {
+      chkdb(g_dic->createTable(tab) == 0);
+      chkdb(g_dic->createIndex(ind) == 0);
+    } else
+      g_opts.loaddata = false;
+  }
+  chkdb((g_tab = g_dic->getTable(g_tabname)) != 0);
+  chkdb((g_ind = g_dic->getIndex(g_indname, g_tabname)) != 0);
+  g_dic = 0;
+  return 0;
+}
+
+static int
+droptable()
+{
+  g_dic = g_ndb->getDictionary();
+  if (! g_opts.keeptable)
+    chkdb(g_dic->dropTable(g_tabname) == 0);
+  g_dic = 0;
+  return 0;
+}
+
+struct Val {
+  Int16 b;
+  bool c_null;
+  uchar c[1 + g_charlen];
+  bool d_null;
+  Uint32 d;
+  // partial values for use in Bnd
+  uint numattrs;
+  void make(uint n = g_numattrs, bool is_val = true);
+  int cmp(const Val& val, uint n = g_numattrs) const;
+};
+
+static NdbOut&
+operator<<(NdbOut& out, const Val& val)
+{
+  out << "[";
+  if (val.numattrs >= 1) {
+    out << val.b;
+  }
+  if (val.numattrs >= 2) {
+    out << " ";
+    if (val.c_null)
+      out << "NULL";
+    else {
+      char buf[1 + g_charlen];
+      sprintf(buf, "%.*s", val.c[0], &val.c[1]);
+      out << "'" << buf << "'";
+    }
+  }
+  if (val.numattrs >= 3) {
+    out << " ";
+    if (val.d_null)
+      out <<" NULL";
+    else
+      out << val.d;
+  }
+  out << "]";
+  return out;
+}
+
+void
+Val::make(uint n, bool is_val)
+{
+  if (n >= 1) {
+    uint b_max = is_val ? g_val_b_max : g_bnd_b_max;
+    b = (int)urandom(2 * b_max) - (int)b_max;
+  }
+  if (n >= 2) {
+    if (urandom(100) < g_opts.nullkeys)
+      c_null = 1;
+    else {
+      const char* c_char = is_val ? g_val_c_char : g_bnd_c_char;
+      // prefer shorter
+      uint len = urandom(urandom(g_charlen + 2));
+      c[0] = len;
+      uint j;
+      for (j = 0; j < len; j++) {
+        uint k = urandom(strlen(c_char));
+        c[1 + j] = c_char[k];
+      }
+      c_null = 0;
+    }
+  }
+  if (n >= 3) {
+    if (urandom(100) < g_opts.nullkeys)
+      d_null = 1;
+    else {
+      uint d_max = is_val ? g_val_d_max : g_bnd_d_max;
+      d = urandom(d_max);
+      d_null = 0;
+    }
+  }
+  numattrs = n;
+}
+
+int
+Val::cmp(const Val& val, uint n) const
+{
+  int k = 0;
+  if (k == 0 && n >= 1) {
+    if (b < val.b)
+      k = -1;
+    else if (b > val.b)
+      k = +1;
+  }
+  if (k == 0 && n >= 2) {
+    if (! c_null && ! val.c_null) {
+      const uchar* s1 = &c[1];
+      const uchar* s2 = &val.c[1];
+      const uint l1 = (uint)c[0];
+      const uint l2 = (uint)val.c[0];
+      assert(l1 <= g_charlen && l2 <= g_charlen);
+      k = g_cs->coll->strnncollsp(g_cs, s1, l1, s2, l2, 0);
+    } else if (! c_null) {
+      k = +1;
+    } else if (! val.c_null) {
+      k = -1;
+    }
+  }
+  if (k == 0 && n >= 3) {
+    if (! d_null && ! val.d_null) {
+      if (d < val.d)
+        k = -1;
+      else if (d > val.d)
+        k = +1;
+    } else if (! d_null) {
+      k = +1;
+    } else if (! val.d_null) {
+      k = -1;
+    }
+  }
+  return k;
+}
+
+struct Key {
+  Val val;
+  union {
+    bool flag;
+    uint count;
+    uint rpk;
+  };
+};
+
+static NdbOut&
+operator<<(NdbOut& out, const Key& key)
+{
+  out << key.val << " info:" << key.count;
+  return out;
+}
+
+static Key* g_keys = 0;
+static Key* g_sortkeys = 0;
+static uint g_sortcount = 0;
+static Key* g_minkey = 0;
+static Key* g_maxkey = 0;
+
+static void
+freekeys()
+{
+  if (g_keys != 0)
+    my_free((char*)g_keys, MYF(0));
+  if (g_sortkeys != 0)
+    my_free((char*)g_sortkeys, MYF(0));
+  g_keys = 0;
+  g_sortkeys = 0;
+}
+
+static int
+allockeys()
+{
+  freekeys();
+  size_t sz = sizeof(Key) * g_opts.rows;
+  g_keys = (Key*)my_malloc(sz, MYF(0));
+  g_sortkeys = (Key*)my_malloc(sz, MYF(0));
+  chkrc(g_keys != 0 && g_sortkeys != 0);
+  memset(g_keys, 0x1f, sz);
+  memset(g_sortkeys, 0x1f, sz);
+  return 0;
+}
+
+static void
+makekeys()
+{
+  uint i;
+  for (i = 0; i < g_opts.rows; i++) {
+    Key& key = g_keys[i];
+    key.val.make();
+    key.flag = false; // mark for dup generation done
+  }
+  for (i = 0; i < g_opts.rows; i++) {
+    Key& key = g_keys[i];
+    if (key.flag)
+      continue;
+    key.flag = true;
+    uint fudge = 9;
+    uint n = (urandom(fudge * (g_opts.dupkeys - 100)) + 99) / 100;
+    uint k;
+    for (k = 1; k < n; k++) {
+      uint j = urandom(g_opts.rows);
+      do {
+        Key& dst = g_keys[j];
+        if (! dst.flag) {
+          dst.val = key.val;
+          dst.flag = true;
+          break;
+        }
+      } while (urandom(g_opts.tryhard) != 0);
+    }
+  }
+}
+
+static int
+insertdata()
+{
+  const uint batch = 512;
+  chkdb((g_con = g_ndb->startTransaction()) != 0);
+  uint i = 0;
+  while (i < g_opts.rows) {
+    chkdb((g_op = g_con->getNdbOperation(g_tab)) != 0);
+    chkdb(g_op->insertTuple() == 0);
+    Uint32 a = i;
+    const Val& val = g_keys[i].val;
+    const char* a_addr = (const char*)&a;
+    const char* b_addr = (const char*)&val.b;
+    const char* c_addr = ! val.c_null ? (const char*)val.c : 0;
+    const char* d_addr = ! val.d_null ? (const char*)&val.d : 0;
+    Uint32 no = 0;
+    chkdb(g_op->equal(no++, a_addr) == 0);
+    chkdb(g_op->setValue(no++, b_addr) == 0);
+    chkdb(g_op->setValue(no++, c_addr) == 0);
+    chkdb(g_op->setValue(no++, d_addr) == 0);
+    if (i++ % batch == 0) {
+      chkdb(g_con->execute(NdbTransaction::Commit) == 0);
+      g_ndb->closeTransaction(g_con);
+      g_con = 0;
+      g_op = 0;
+      chkdb((g_con = g_ndb->startTransaction()) != 0);
+    }
+  }
+  chkdb(g_con->execute(NdbTransaction::Commit) == 0);
+  g_ndb->closeTransaction(g_con);
+  g_con = 0;
+  g_op = 0;
+  ll0(g_tabname << ": inserted " << g_opts.rows << " rows");
+  return 0;
+}
+
+static int
+countrows()
+{
+  Uint64 rows = 0;
+  Uint64 r;
+  char* r_addr = (char*)&r;
+  chkdb((g_con = g_ndb->startTransaction()) != 0);
+  chkdb((g_scan_op = g_con->getNdbScanOperation(g_tab)) != 0);
+  chkdb(g_scan_op->readTuples() == 0);
+  chkdb(g_scan_op->interpret_exit_last_row() == 0);
+  chkdb(g_scan_op->getValue(NdbDictionary::Column::ROW_COUNT, r_addr) != 0);
+  chkdb(g_con->execute(NdbTransaction::NoCommit) == 0);
+  while (1) {
+    int ret;
+    r = ~(Uint64)0;
+    chkdb((ret = g_scan_op->nextResult()) == 0 || ret == 1);
+    if (ret == 1)
+      break;
+    rows += r;
+  }
+  g_ndb->closeTransaction(g_con);
+  g_con = 0;
+  g_scan_op = 0;
+  g_opts.rows = rows;
+  return 0;
+}
+
+static int
+scandata()
+{
+  chkdb((g_con = g_ndb->startTransaction()) != 0);
+  chkdb((g_scan_op = g_con->getNdbScanOperation(g_tab)) != 0);
+  chkdb(g_scan_op->readTuples() == 0);
+  Uint32 a;
+  Val val;
+  char* a_addr = (char*)&a;
+  char* b_addr = (char*)&val.b;
+  char* c_addr = (char*)val.c;
+  char* d_addr = (char*)&val.d;
+  Uint32 no = 0;
+  NdbRecAttr* b_ra;
+  NdbRecAttr* c_ra;
+  NdbRecAttr* d_ra;
+  chkdb(g_scan_op->getValue(no++, a_addr) != 0);
+  chkdb((b_ra = g_scan_op->getValue(no++, b_addr)) != 0);
+  chkdb((c_ra = g_scan_op->getValue(no++, c_addr)) != 0);
+  chkdb((d_ra = g_scan_op->getValue(no++, d_addr)) != 0);
+  chkdb(g_con->execute(NdbTransaction::NoCommit) == 0);
+  uint count = 0;
+  uint i;
+  for (i = 0; i < g_opts.rows; i++)
+    g_keys[i].count = 0;
+  while (1) {
+    int ret;
+    a = ~(Uint32)0;
+    chkdb((ret = g_scan_op->nextResult()) == 0 || ret == 1);
+    if (ret == 1)
+      break;
+    assert(b_ra->isNULL() == 0 && c_ra->isNULL() != -1 && d_ra->isNULL() != -1);
+    val.c_null = c_ra->isNULL();
+    val.d_null = d_ra->isNULL();
+    i = (uint)a;
+    chkrc(i < g_opts.rows);
+    Key& key = g_keys[i];
+    if (g_opts.loaddata)
+      chkrc(key.val.cmp(val) == 0);
+    else
+      key.val = val;
+    key.count++;
+    count++;
+  }
+  g_ndb->closeTransaction(g_con);
+  g_con = 0;
+  g_scan_op = 0;
+  for (i = 0; i < g_opts.rows; i++)
+    chkrc(g_keys[i].count == 1);
+  assert(count == g_opts.rows);
+  int level = g_opts.loaddata ? 1 : 0;
+  llx(level, g_tabname << ": scanned " << g_opts.rows << " rows");
+  return 0;
+}
+
+static int
+loaddata()
+{
+  if (g_opts.loaddata) {
+    chkrc(allockeys() == 0);
+    makekeys();
+    chkrc(insertdata() == 0);
+  } else {
+    chkrc(countrows() == 0);
+    chkrc(g_opts.rows != 0);
+    ll0(g_tabname << ": using old table of " << g_opts.rows << " rows");
+    chkrc(allockeys() == 0);
+  }
+  chkrc(scandata() == 0);
+  uint i;
+  for (i = 0; i < g_opts.rows; i++)
+    ll3(i << ": " << g_keys[i]);
+  return 0;
+}
+
+// true = match, index = match or next higher
+static bool
+sortval(const Val& val, int& index)
+{
+  if (unlikely(g_sortcount == 0)) {
+    index = 0;
+    return false;
+  }
+  int lo = -1;
+  int hi = (int)g_sortcount;
+  int ret;
+  int j;
+  do {
+    j = (hi + lo) / 2;
+    ret = val.cmp(g_sortkeys[j].val);
+    if (ret < 0)
+      hi = j;
+    else if (ret > 0)
+      lo = j;
+    else
+      break;
+  } while (hi - lo > 1);
+  if (ret == 0) {
+    index = j;
+    return true;
+  }
+  index = hi;
+  return false;
+}
+
+static void
+sortkeys()
+{
+  // insert sort with binary search
+  g_sortcount = 0;
+  uint i;
+  for (i = 0; i < g_opts.rows; i++) {
+    const Val& val = g_keys[i].val;
+    int index;
+    bool match = sortval(val, index);
+    Key& dst = g_sortkeys[index];
+    if (match) {
+      dst.rpk++;
+    } else {
+      uint bytes = ((int)g_sortcount - index) * sizeof(Key);
+      memmove(&dst + 1, &dst, bytes);
+      dst.val = val;
+      dst.rpk = 1;
+      g_sortcount++;
+    }
+  }
+  g_minkey = &g_sortkeys[0];
+  g_maxkey = &g_sortkeys[g_sortcount - 1];
+  ll1("counted " << g_sortcount << " distinct keys");
+}
+
+struct Bnd {
+  Val val;
+  /*
+   * A bound is a partial key value (0 to g_numattrs attributes).
+   * It is not equal to any key value.  Instead, it has a "side".
+   *
+   * side = 0 if the bound is empty
+   * side = -1 if the bound is "just before" its value
+   * side = +1 if the bound is "just after" its value
+   *
+   * This is another way of looking at strictness of non-empty
+   * start and end keys in a range.
+   *
+   * start key is strict if side = +1
+   * end key is strict if side = -1
+   *
+   * NDB API specifies strictness in the bound type of the last
+   * index attribute which is part of the start/end key.
+   *
+   * LE (0) - strict: n - side: -1
+   * LT (1) - strict: y - side: +1
+   * GE (2) - strict: n - side: +1
+   * GT (3) - strict: y - side: -1
+   *
+   * A non-empty bound divides keys into 2 disjoint subsets:
+   * keys before (cmp() == -1) and keys after (cmp() == +1).
+   */
+  int side;
+  Bnd& make(uint minattrs);
+  Bnd& make(uint minattrs, const Val& theval);
+  int cmp(const Val& val) const;
+  int type(uint lohi, uint colno) const; // for setBound
+};
+
+static NdbOut&
+operator<<(NdbOut& out, const Bnd& bnd)
+{
+  out << bnd.val;
+  out << " side: " << bnd.side;
+  return out;
+}
+
+Bnd&
+Bnd::make(uint minattrs)
+{
+  uint numattrs = minattrs + urandom(g_numattrs - minattrs);
+  val.make(numattrs, false);
+  side = val.numattrs == 0 ? 0 : urandom(2) == 0 ? -1 : +1;
+  return *this;
+}
+
+Bnd&
+Bnd::make(uint minattrs, const Val& theval)
+{
+  uint numattrs = minattrs + urandom(g_numattrs - minattrs);
+  val = theval;
+  val.numattrs = numattrs;
+  side = val.numattrs == 0 ? 0 : urandom(2) == 0 ? -1 : +1;
+  return *this;
+}
+
+int
+Bnd::cmp(const Val& theval) const
+{
+  int place; // debug
+  int ret;
+  do {
+    assert(theval.numattrs == g_numattrs);
+    int k = theval.cmp(val, val.numattrs);
+    if (k != 0) {
+      place = 1;
+      ret = k;
+      break;
+    }
+    if (side != 0) {
+      place = 2;
+      ret = -side;
+      break;
+    }
+    place = 3;
+    ret = 0;
+    assert(val.numattrs == 0);
+  } while (0);
+  ll3("cmp: val: " << theval << " bnd: " << *this <<
+      " return: " << ret << " at " << place);
+  return ret;
+}
+
+int
+Bnd::type(uint lohi, uint colno) const
+{
+  int t;
+  assert(lohi <= 1 && colno < val.numattrs && (side == -1 || side == +1));
+  if (lohi == 0) {
+    if (colno + 1 < val.numattrs)
+      t = 0; // LE
+    else if (side == -1)
+      t = 0; // LE
+    else
+      t = 1; // LT
+  } else {
+    if (colno + 1 < val.numattrs)
+      t = 2; // GE
+    else if (side == +1)
+      t = 2; // GE
+    else
+      t = 3; // GT
+  }
+  return t;
+}
+
+struct Range {
+  Bnd bnd[2];
+  uint minattrs() const;
+  uint maxattrs() const;
+  int cmp(const Val& val) const; // -1,0,+1 = key is before,in,after range
+  uint rowcount() const;
+  bool iseq() const;
+  // stats
+  bool flag;
+  uint statrows;
+  uint scanrows;
+  double errpct;
+};
+
+static NdbOut&
+operator<<(NdbOut& out, const Range& range)
+{
+  out << "bnd0: " << range.bnd[0] << " bnd1: " << range.bnd[1];
+  return out;
+}
+
+uint
+Range::minattrs() const
+{
+  return min(bnd[0].val.numattrs, bnd[1].val.numattrs);
+}
+
+uint
+Range::maxattrs() const
+{
+  return max(bnd[0].val.numattrs, bnd[1].val.numattrs);
+}
+
+int
+Range::cmp(const Val& theval) const
+{
+  int place; // debug
+  int ret;
+  do  {
+    int k;
+    k = bnd[0].cmp(theval);
+    if (k < 0) {
+      place = 1;
+      ret = -1;
+      break;
+    }
+    k = bnd[1].cmp(theval);
+    if (k > 0) {
+      place = 2;
+      ret = +1;
+      break;
+    }
+    place = 3;
+    ret = 0;
+  } while (0);
+  ll3("cmp: val: " << theval << " range: " << *this <<
+      " return: " << ret << " at " << place);
+  return ret;
+}
+
+uint
+Range::rowcount() const
+{
+  ll2("rowcount: " << *this);
+  int i;
+  // binary search for first and last in range
+  int lim[2];
+  for (i = 0; i <= 1; i++) {
+    ll3("search i=" << i);
+    int lo = -1;
+    int hi = (int)g_sortcount;
+    int ret;
+    int j;
+    do {
+      j = (hi + lo) / 2;
+      ret = cmp(g_sortkeys[j].val);
+      if (i == 0) {
+        if (ret < 0)
+          lo = j;
+        else
+          hi = j;
+      } else {
+        if (ret > 0)
+          hi = j;
+        else
+          lo = j;
+      }
+    } while (hi - lo > 1);
+    if (ret == 0)
+      lim[i] = j;
+    else if (i == 0)
+      lim[i] = hi;
+    else
+      lim[i] = lo;
+  }
+  // the range
+  const int lo = max(lim[0], 0);
+  const int hi = min(lim[1], (int)g_sortcount - 1);
+  if (! g_opts.nochecks) {
+    int curr = -1;
+    for (i = 0; i < (int)g_sortcount; i++) {
+      int k = cmp(g_sortkeys[i].val);
+      if (k < 0)
+        assert(i < lo);
+      else if (k == 0)
+        assert(lo <= i && i <= hi);
+      else
+        assert(i > hi);
+      assert(curr <= k);
+      if (curr < k)
+        curr = k;
+    }
+  }
+  // sum them up
+  uint count = 0;
+  for (i = lo; i <= hi; i++)
+    count += g_sortkeys[i].count;
+  ll2("count: " << count << " index lim: " << lim[0] << " " << lim[1]);
+  return count;
+}
+
+bool
+Range::iseq() const
+{
+  return
+    minattrs() == maxattrs() &&
+    bnd[0].val.cmp(bnd[1].val, minattrs()) == 0 &&
+    bnd[0].side < bnd[1].side;
+}
+
+static Range* g_ranges = 0;
+
+static void
+freeranges()
+{
+  if (g_ranges != 0)
+    my_free((char*)g_ranges, MYF(0));
+  g_ranges = 0;
+}
+
+static int
+allocranges()
+{
+  freeranges();
+  size_t sz = sizeof(Range) * g_opts.ops;
+  g_ranges = (Range*)my_malloc(sz, MYF(0));
+  chkrc(g_ranges != 0);
+  memset(g_ranges, 0x1f, sz);
+  return 0;
+}
+
+static void
+makeranges()
+{
+  uint i;
+  for (i = 0; i < g_opts.ops; i++) {
+    Range& range = g_ranges[i];
+    range.flag = false; // mark for dup generation done
+    bool fulleq = (urandom(100) < g_opts.eqscans);
+    bool eq = fulleq || (urandom(100) < g_opts.eqscans);
+    bool matcheq = eq && (urandom(10) != 0);
+    if (! eq) {
+      // random but prefer non-empty and no more than scanpct
+      do {
+        range.bnd[0].make(0);
+        range.bnd[1].make(0);
+        uint count = range.rowcount();
+        if (count != 0 && 100 * count <= g_opts.scanpct * g_opts.rows)
+          break;
+      } while (urandom(g_opts.tryhard) != 0);
+    } else {
+      uint minattrs = fulleq ? g_numattrs : 1;
+      if (! matcheq) {
+        range.bnd[0].make(minattrs);
+      } else {
+        uint m = urandom(g_sortcount);
+        const Val& val = g_sortkeys[m].val;
+        range.bnd[0].make(minattrs, val);
+      }
+      range.bnd[1] = range.bnd[0];
+      range.bnd[0].side = -1;
+      range.bnd[1].side = +1;
+      // fix types
+      range.bnd[0];
+      range.bnd[1];
+      assert(range.iseq());
+    }
+  }
+  for (i = 0; i < g_opts.ops; i++) {
+    Range& range = g_ranges[i];
+    if (range.flag)
+      continue;
+    range.flag = true;
+    if (urandom(100) < g_opts.dupscans) {
+      uint j = urandom(g_opts.ops);
+      do {
+        Range& dst = g_ranges[j];
+        if (! dst.flag) {
+          dst.bnd[0] = range.bnd[0];
+          dst.bnd[1] = range.bnd[1];
+          dst.flag = true;
+          break;
+        }
+      } while (urandom(g_opts.tryhard) != 0);
+    }
+  }
+}
+
+static int
+setbounds(const Range& range)
+{
+  // currently must do each attr in order
+  ll2("setbounds: " << range);
+  uint i;
+  const Bnd (&bnd)[2] = range.bnd;
+  for (i = 0; i < g_numattrs; i++) {
+    const Uint32 no = i; // index attribute number
+    uint j;
+    int type[2] = { -1, -1 };
+    for (j = 0; j <= 1; j++) {
+      if (no < bnd[j].val.numattrs)
+        type[j] = bnd[j].type(j, no);
+    }
+    for (j = 0; j <= 1; j++) {
+      int t = type[j];
+      if (t == -1)
+        continue;
+      if (no + 1 < bnd[j].val.numattrs)
+        t &= ~(uint)1; // strict bit is set on last bound only
+      const Val& val = bnd[j].val;
+      const void* addr = 0;
+      if (no == 0)
+        addr = (const void*)&val.b;
+      else if (no == 1)
+        addr = ! val.c_null ? (const void*)val.c : 0;
+      else if (no == 2)
+        addr = ! val.d_null ? (const void*)&val.d : 0;
+      else
+        assert(false);
+      ll2("setBound attr:" << no << " type:" << t << " val: " << val);
+      chkdb(g_rangescan_op->setBound(no, t, addr) == 0);
+    }
+  }
+  return 0;
+}
+
+static int
+allocstat()
+{
+  g_stat = new NdbIndexStat(g_ind);
+  chkdb(g_stat->alloc_cache(32) == 0);
+  return 0;
+}
+
+static int
+runstat(Range& range, int flags)
+{
+  ll2("runstat: " << range << " flags=" << flags);
+  chkdb((g_con = g_ndb->startTransaction()) != 0);
+  chkdb((g_rangescan_op = g_con->getNdbIndexScanOperation(g_ind, g_tab)) != 0);
+  chkdb(g_rangescan_op->readTuples(NdbOperation::LM_CommittedRead) == 0);
+  chkrc(setbounds(range) == 0);
+  Uint64 count = ~(Uint64)0;
+  chkdb(g_stat->records_in_range(g_ind, g_rangescan_op, g_opts.rows, &count, flags) == 0);
+  g_ndb->closeTransaction(g_con);
+  g_con = 0;
+  g_rangescan_op = 0;
+  range.statrows = (uint)count;
+  chkrc((Uint64)range.statrows == count);
+  ll2("stat: " << range.statrows);
+  return 0;
+}
+
+static int
+runscan(Range& range)
+{
+  ll2("runscan: " << range);
+  chkdb((g_con = g_ndb->startTransaction()) != 0);
+  chkdb((g_rangescan_op = g_con->getNdbIndexScanOperation(g_ind, g_tab)) != 0);
+  chkdb(g_rangescan_op->readTuples() == 0);
+  chkrc(setbounds(range) == 0);
+  Uint32 a;
+  char* a_addr = (char*)&a;
+  Uint32 no = 0;
+  chkdb(g_rangescan_op->getValue(no++, a_addr) != 0);
+  chkdb(g_con->execute(NdbTransaction::NoCommit) == 0);
+  uint count = 0;
+  uint i;
+  for (i = 0; i < g_opts.rows; i++)
+    g_keys[i].count = 0;
+  while (1) {
+    int ret;
+    a = ~(Uint32)0;
+    chkdb((ret = g_rangescan_op->nextResult()) == 0 || ret == 1);
+    if (ret == 1)
+      break;
+    i = (uint)a;
+    chkrc(i < g_opts.rows);
+    Key& key = g_keys[i];
+    ll3("scan: " << key);
+    int k = range.cmp(key.val);
+    chkrc(k == 0);
+    chkrc(key.count == 0);
+    key.count++;
+    count++;
+  }
+  g_ndb->closeTransaction(g_con);
+  g_con = 0;
+  g_rangescan_op = 0;
+  if (! g_opts.nochecks) {
+    for (i = 0; i < g_opts.rows; i++) {
+      const Key& key = g_keys[i];
+      int k = range.cmp(key.val);
+      assert((k != 0 && key.count == 0) || (k == 0 && key.count == 1));
+    }
+    assert(range.rowcount() == count);
+  }
+  range.scanrows = count;
+  ll2("scan: " << range.scanrows);
+  return 0;
+}
+
+static int
+runscans()
+{
+  uint i;
+  for (i = 0; i < g_opts.ops; i++) {
+    Range& range = g_ranges[i];
+    ll1("range " << i << ": " << range);
+    // simulate old handler code
+    int flags = 0;
+    if (i < 32 || i % 20 == 0)
+      flags |= NdbIndexStat::RR_UseDb;
+    chkrc(runstat(range, flags) == 0);
+    chkrc(runscan(range) == 0);
+    // if stat is 0 then it is exact scan count
+    chkrc(range.statrows != 0 || range.scanrows == 0);
+    // measure error as fraction of total rows
+    double x = (double)range.statrows;
+    double y = (double)range.scanrows;
+    double z = (double)g_opts.rows;
+    double err = (x - y) / z;
+    // report in pct
+    range.errpct = 100.0 * err;
+    ll1("range " << i << ":" <<
+        " stat: " << range.statrows << " scan: " << range.scanrows <<
+        " errpct: " << range.errpct);
+  }
+  return 0;
+}
+
+struct Stat {
+  const char* name;
+  uint count;
+  double sum;
+  double minval;
+  double maxval;
+  double avg;
+  double varsum;
+  double var;
+  double stddev;
+  void init();
+  void add(const Stat& stat);
+};
+
+void 
+Stat::init()
+{
+  name = "stat";
+  count = 0;
+  sum = minval = maxval = avg = varsum = var = stddev = 0.0;
+}
+
+void
+Stat::add(const Stat& stat)
+{
+  if (count == 0) {
+    *this = stat;
+    return;
+  }
+  Stat tmp = *this;
+  tmp.count = count + stat.count;
+  tmp.sum = sum + stat.sum;
+  tmp.minval = minval <= stat.minval ? minval : stat.minval;
+  tmp.maxval = maxval >= stat.maxval ? maxval : stat.maxval;
+  tmp.avg = tmp.sum / double(tmp.count);
+  tmp.varsum = varsum + stat.varsum;
+  tmp.var = tmp.varsum / double(tmp.count);
+  tmp.stddev = sqrt(tmp.var);
+  *this = tmp;
+}
+
+static NdbOut&
+operator<<(NdbOut& out, const Stat& stat)
+{
+  out << stat.name << ": " << "count: " << stat.count
+      << " min: " << stat.minval << " max: " << stat.maxval
+      << " avg: " << stat.avg << " stddev: " << stat.stddev;
+  return out;
+}
+
+template <class T, class V>
+static void
+computestat(Stat& stat)
+{
+  stat.init();
+  stat.name = V::name();
+  const T* array = V::array();
+  stat.count = V::count();
+  assert(stat.count != 0);
+  uint i;
+  for (i = 0; i < stat.count; i++) {
+    const T& item = array[i];
+    double data = V::data(item);
+    stat.sum += data;
+    if (i == 0)
+      stat.minval = stat.maxval = data;
+    else {
+      if (stat.minval > data)
+        stat.minval = data;
+      if (stat.maxval < data)
+        stat.maxval = data;
+    }
+  }
+  stat.avg = stat.sum / double(stat.count);
+  stat.varsum = 0.0;
+  for (i = 0; i < stat.count; i++) {
+    const T& item = array[i];
+    double data = V::data(item);
+    double x = data - stat.avg;
+    stat.varsum += x * x;
+  }
+  stat.var = stat.varsum / double(stat.count);
+  stat.stddev = sqrt(stat.var);
+}
+
+struct V_rpk {
+  static const char* name() { return "rec per key"; }
+  static const Key* array() { return g_sortkeys; }
+  static uint count() { return g_sortcount; }
+  static double data(const Key& key) { return (double)key.rpk; }
+};
+
+struct V_rir {
+  static const char* name() { return "rir err pct"; }
+  static const Range* array() { return g_ranges; }
+  static uint count() { return g_opts.ops; }
+  static double data(const Range& range) { return (double)range.errpct; }
+};
+
+template static void computestat<Key, V_rpk>(Stat& stat);
+template static void computestat<Range, V_rir>(Stat& stat);
+
+static Stat g_stat_rpk; // summaries over loops
+static Stat g_stat_rir;
+
+static void
+loopstats()
+{
+  Stat stat_rpk; // records per key
+  Stat stat_rir; // record in range
+  if (g_loop == 0) {
+    g_stat_rpk.init();
+    g_stat_rir.init();
+  }
+  computestat<Key, V_rpk>(stat_rpk);
+  computestat<Range, V_rir>(stat_rir);
+  if (g_opts.loop != 1) {
+    ll0("=== loop " << g_loop << " summary ===");
+    ll0(stat_rpk);
+    ll0(stat_rir);
+  }
+  // accumulate
+  g_stat_rpk.add(stat_rpk);
+  g_stat_rir.add(stat_rir);
+}
+
+static void
+finalstats()
+{
+  ll0("=== summary ===");
+  ll0(g_stat_rpk);
+  ll0(g_stat_rir);
+}
+
+static void
+setseed(int n)
+{
+  uint seed;
+  if (n == -1) {
+    if (g_opts.seed == 0)
+      return;
+    if (g_opts.seed != -1)
+      seed = (uint)g_opts.seed;
+    else
+      seed = 1 + (ushort)getpid();
+  } else {
+    if (g_opts.seed != 0)
+      return;
+    seed = n;
+  }
+  ll0("seed=" << seed);
+  srandom(seed);
+}
+
+static int
+runtest()
+{
+  setseed(-1);
+  g_cs = get_charset_by_name(g_csname, MYF(0));
+  if (g_cs == 0)
+    g_cs = get_charset_by_csname(g_csname, MY_CS_PRIMARY, MYF(0));
+  chkrc(g_cs != 0);
+  for (g_loop = 0; g_opts.loop == 0 || g_loop < g_opts.loop; g_loop++) {
+    ll0("=== loop " << g_loop << " ===");
+    setseed(g_loop);
+    chkrc(createtable() == 0);
+    chkrc(loaddata() == 0);
+    sortkeys();
+    chkrc(allocranges() == 0);
+    makeranges();
+    chkrc(allocstat() == 0);
+    chkrc(runscans() == 0);
+    chkrc(droptable() == 0);
+    loopstats();
+  }
+  finalstats();
+  return 0;
+}
+
+NDB_STD_OPTS_VARS;
+
+static struct my_option
+my_long_options[] =
+{
+  NDB_STD_OPTS("testIndexStat"),
+  { "loglevel", 1001, "Logging level in this program 0-3 (default 0)",
+    (gptr*)&g_opts.loglevel, (gptr*)&g_opts.loglevel, 0,
+    GET_INT, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },
+  { "seed", 1002, "Random seed (0=loop number, default -1=random)",
+    (gptr*)&g_opts.seed, (gptr*)&g_opts.seed, 0,
+    GET_INT, REQUIRED_ARG, -1, 0, 0, 0, 0, 0 },
+  { "loop", 1003, "Number of test loops (default 1, 0=forever)",
+    (gptr*)&g_opts.loop, (gptr*)&g_opts.loop, 0,
+    GET_INT, REQUIRED_ARG, 1, 0, 0, 0, 0, 0 },
+  { "rows", 1004, "Number of rows (default 100000)",
+    (gptr*)&g_opts.rows, (gptr*)&g_opts.rows, 0,
+    GET_UINT, REQUIRED_ARG, 100000, 0, 0, 0, 0, 0 },
+  { "ops", 1005, "Number of index scans per loop (default 1000)",
+    (gptr*)&g_opts.ops, (gptr*)&g_opts.ops, 0,
+    GET_UINT, REQUIRED_ARG, 1000, 0, 0, 0, 0, 0 },
+  { "dupkeys", 1006, "Pct records per key (min 100, default 1000)",
+    (gptr*)&g_opts.dupkeys, (gptr*)&g_opts.dupkeys, 0,
+    GET_UINT, REQUIRED_ARG, 1000, 0, 0, 0, 0, 0 },
+  { "scanpct", 1007, "Preferred max pct of total rows per scan (default 5)",
+    (gptr*)&g_opts.scanpct, (gptr*)&g_opts.scanpct, 0,
+    GET_UINT, REQUIRED_ARG, 5, 0, 0, 0, 0, 0 },
+  { "nullkeys", 1008, "Pct nulls in each key attribute (default 10)",
+    (gptr*)&g_opts.nullkeys, (gptr*)&g_opts.nullkeys, 0,
+    GET_UINT, REQUIRED_ARG, 10, 0, 0, 0, 0, 0 },
+  { "eqscans", 1009, "Pct scans for partial/full equality (default 50)",
+    (gptr*)&g_opts.eqscans, (gptr*)&g_opts.eqscans, 0,
+    GET_UINT, REQUIRED_ARG, 50, 0, 0, 0, 0, 0 },
+  { "dupscans", 1010, "Pct scans using same bounds (default 10)",
+    (gptr*)&g_opts.dupscans, (gptr*)&g_opts.dupscans, 0,
+    GET_UINT, REQUIRED_ARG, 10, 0, 0, 0, 0, 0 },
+  { "keeptable", 1011, "Use existing table and data if any and do not drop",
+    (gptr*)&g_opts.keeptable, (gptr*)&g_opts.keeptable, 0,
+    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
+  { "no-extra-checks", 1012, "Omit expensive consistency checks",
+    (gptr*)&g_opts.nochecks, (gptr*)&g_opts.nochecks, 0,
+    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
+  { "abort-on-error", 1013, "Dump core on any error",
+    (gptr*)&g_opts.abort, (gptr*)&g_opts.abort, 0,
+    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
+  { 0, 0, 0,
+    0, 0, 0,
+    GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0 }
+};
+
+static void
+usage()
+{
+  ndbout
+    << g_progname
+    << ": measure records_in_range error as percentage of total rows" << endl;
+  my_print_help(my_long_options);
+}
+
+static int
+checkoptions()
+{
+  chkrc(g_opts.rows != 0);
+  chkrc(g_opts.nullkeys <= 100);
+  chkrc(g_opts.dupkeys >= 100);
+  chkrc(g_opts.scanpct <= 100);
+  chkrc(g_opts.eqscans <= 100);
+  chkrc(g_opts.dupscans <= 100);
+  return 0;
+}
+
+static int
+doconnect()
+{
+  g_ncc = new Ndb_cluster_connection();
+  chkdb(g_ncc->connect(30) == 0);
+  g_ndb = new Ndb(g_ncc, "TEST_DB");
+  chkdb(g_ndb->init() == 0 && g_ndb->waitUntilReady(30) == 0);
+  return 0;
+}
+
+static void
+freeall()
+{
+  delete g_stat;
+  freekeys();
+  freeranges();
+  delete g_ndb;
+  delete g_ncc;
+}
+
+int
+main(int argc, char** argv)
+{
+  ndb_init();
+  const char* g_progname =
+    strchr(argv[0], '/') ? strrchr(argv[0], '/') + 1 : argv[0];
+  uint i;
+  ndbout << g_progname;
+  for (i = 1; i < argc; i++)
+    ndbout << " " << argv[i];
+  ndbout << endl;
+  int ret;
+  ret = handle_options(&argc, &argv, my_long_options, ndb_std_get_one_option);
+  if (ret != 0 || argc != 0)
+    return NDBT_ProgramExit(NDBT_WRONGARGS);
+  if (checkoptions() == 0 && doconnect() == 0 && runtest() == 0) {
+    freeall();
+    return NDBT_ProgramExit(NDBT_OK);
+  }
+  freeall();
+  return NDBT_ProgramExit(NDBT_FAILED);
+}

storage/ndb/test/src/NdbBackup.cpp

@@ -350,7 +350,8 @@ int
 FailS_codes[] = {
   10025,
   10027,
-  10033
+  10033,
+  10035
 };
 
 int
@@ -362,7 +363,8 @@ FailM_codes[] = {
   10027,
   10028,
   10031,
-  10033
+  10033,
+  10035
 };
 
 int