ndb: wl-1893: range scanning backwards, handler

mysql-test/ndb/ndb_range_bounds.pl

 #
 # test range scan bounds
-# output to mysql-test/t/ndb_range_bounds.test
-#
-# give option --all to generate all cases
+# give option --all to test all cases
 #
 
 use strict;
 use integer;
 use Getopt::Long;
+use DBI;
 
 my $opt_all = 0;
 my $opt_cnt = 5;
-GetOptions("all" => \$opt_all, "cnt=i" => \$opt_cnt)
-  or die "options are:  --all --cnt=N";
+my $opt_verbose = 0;
+GetOptions("all" => \$opt_all, "cnt=i" => \$opt_cnt, "verbose" => \$opt_verbose)
+  or die "options are:  --all --cnt=N --verbose";
 
-my $table = 't';
+my $mysql_top = $ENV{MYSQL_TOP};
+my $dsn = "dbi:mysql:database=test;host=localhost;mysql_read_default_file=$mysql_top/.target/var/my.cnf";
+my $opts = { RaiseError => 0, PrintError => 0, AutoCommit => 1, };
 
-print <<EOF;
---source include/have_ndb.inc
+my $dbh;
+my $sth;
+my $sql;
 
---disable_warnings
-drop table if exists $table;
---enable_warnings
+$dbh = DBI->connect($dsn, "root", undef, $opts) or die $DBI::errstr;
 
-# test range scan bounds
-# generated by mysql-test/ndb/ndb_range_bounds.pl
-# all selects must return 0
+my $table = 't';
 
-EOF
+$sql = "drop table if exists $table";
+$dbh->do($sql) or die $DBI::errstr;
 
-sub cut ($$@) {
-  my($op, $key, @v) = @_;
+sub cut ($$$) {
+  my($op, $key, $val) = @_;
   $op = '==' if $op eq '=';
-  my(@w);
-  eval "\@w = grep(\$_ $op $key, \@v)";
+  my(@w) = @$val;
+  eval "\@w = grep(\$_ $op $key, \@w)";
   $@ and die $@;
-  return @w;
+  return [ @w ];
 }
 
-sub mkdummy (\@) {
+sub mkdummy ($) {
   my ($val) = @_;
   return {
     'dummy' => 1,
     'exp' => '9 = 9',
-    'cnt' => scalar @$val,
+    'res' => $val,
   };
 }
 
-sub mkone ($$$\@) {
+sub mkone ($$$$) {
   my($col, $op, $key, $val) = @_;
-  my $cnt = scalar cut($op, $key, @$val);
+  my $res = cut($op, $key, $val);
   return {
     'exp' => "$col $op $key",
-    'cnt' => $cnt,
+    'res' => $res,
   };
 }
 
-sub mktwo ($$$$$\@) {
+sub mktwo ($$$$$$) {
   my($col, $op1, $key1, $op2, $key2, $val) = @_;
-  my $cnt = scalar cut($op2, $key2, cut($op1, $key1, @$val));
+  my $res = cut($op2, $key2, cut($op1, $key1, $val));
   return {
     'exp' => "$col $op1 $key1 and $col $op2 $key2",
-    'cnt' => $cnt,
+    'res' => $res,
   };
 }
 
-sub mkall ($$$\@) {
+sub mkall ($$$$) {
   my($col, $key1, $key2, $val) = @_;
   my @a = ();
-  my $p = mkdummy(@$val);
+  my $p = mkdummy($val);
   push(@a, $p) if $opt_all;
   my @ops = qw(< <= = >= >);
   for my $op (@ops) {
-    my $p = mkone($col, $op, $key1, @$val);
-    push(@a, $p) if $opt_all || $p->{cnt} != 0;
+    my $p = mkone($col, $op, $key1, $val);
+    push(@a, $p) if $opt_all || @{$p->{res}} != 0;
   }
   my @ops1 = $opt_all ? @ops : qw(= >= >);
   my @ops2 = $opt_all ? @ops : qw(<= <);
   for my $op1 (@ops1) {
     for my $op2 (@ops2) {
-      my $p = mktwo($col, $op1, $key1, $op2, $key2, @$val);
-      push(@a, $p) if $opt_all || $p->{cnt} != 0;
+      my $p = mktwo($col, $op1, $key1, $op2, $key2, $val);
+      push(@a, $p) if $opt_all || @{$p->{res}} != 0;
     }
   }
+  warn scalar(@a)." cases\n" if $opt_verbose;
   return \@a;
 }
 
+my $casecnt = 0;
+
+sub verify ($$$) {
+  my($sql, $ord, $res) = @_;
+  warn "$sql\n" if $opt_verbose;
+  $sth = $dbh->prepare($sql) or die "prepare: $sql: $DBI::errstr";
+  $sth->execute() or die "execute: $sql: $DBI::errstr";
+  #
+  # BUG: execute can return success on error so check again
+  #
+  $sth->err and die "execute: $sql: $DBI::errstr";
+  my @out = ();
+  for my $b (@{$res->[0]}) {
+    for my $c (@{$res->[1]}) {
+      for my $d (@{$res->[2]}) {
+	push(@out, [$b, $c, $d]);
+      }
+    }
+  }
+  if ($ord) {
+    @out = sort {
+      $ord * ($a->[0] - $b->[0]) ||
+      $ord * ($a->[1] - $b->[1]) ||
+      $ord * ($a->[2] - $b->[2]) ||
+      0
+    } @out;
+  }
+  my $cnt = scalar @out;
+  my $n = 0;
+  while (1) {
+    my $row = $sth->fetchrow_arrayref;
+    $row || last;
+    @$row == 3 or die "bad row: $sql:  @$row";
+    for my $v (@$row) {
+      $v =~ s/^\s+|\s+$//g;
+      $v =~ /^\d+$/ or die "bad value: $sql:  $v";
+    }
+    if ($ord) {
+      my $out = $out[$n];
+      $row->[0] == $out->[0] &&
+      $row->[1] == $out->[1] &&
+      $row->[2] == $out->[2] or
+        die "$sql: row $n: got row @$row != @$out";
+    }
+    $n++;
+  }
+  $sth->err and die "fetch: $sql: $DBI::errstr";
+  $n == $cnt or die "verify: $sql: got row count $n != $cnt";
+  $casecnt++;
+}
+
 for my $nn ("bcd", "") {
   my %nn;
   for my $x (qw(b c d)) {
     $nn{$x} = $nn =~ /$x/ ? "not null" : "null";
   }
-  print <<EOF;
+  warn "create table\n";
+  $sql = <<EOF;
 create table $table (
   a int primary key,
   b int $nn{b},
   c int $nn{c},
   d int $nn{d},
   index (b, c, d)
-) engine=ndb;
+) engine=ndb
 EOF
+  $dbh->do($sql) or die $DBI::errstr;
+  warn "insert\n";
+  $sql = "insert into $table values(?, ?, ?, ?)";
+  $sth = $dbh->prepare($sql) or die $DBI::errstr;
   my @val = (0..($opt_cnt-1));
   my $v0 = 0;
   for my $v1 (@val) {
     for my $v2 (@val) {
       for my $v3 (@val) {
-	print "insert into $table values($v0, $v1, $v2, $v3);\n";
+	$sth->bind_param(1, $v0) or die $DBI::errstr;
+	$sth->bind_param(2, $v1) or die $DBI::errstr;
+	$sth->bind_param(3, $v2) or die $DBI::errstr;
+	$sth->bind_param(4, $v3) or die $DBI::errstr;
+	$sth->execute or die $DBI::errstr;
 	$v0++;
       }
     }
   }
+  warn "generate cases\n";
   my $key1 = 1;
   my $key2 = 3;
-  my $a1 = mkall('b', $key1, $key2, @val);
-  my $a2 = mkall('c', $key1, $key2, @val);
-  my $a3 = mkall('d', $key1, $key2, @val);
+  my $a1 = mkall('b', $key1, $key2, \@val);
+  my $a2 = mkall('c', $key1, $key2, \@val);
+  my $a3 = mkall('d', $key1, $key2, \@val);
+  warn "select\n";
+  for my $ord (0, +1, -1) {
+    my $orderby =
+      $ord == 0 ? "" :
+      $ord == +1 ? " order by b, c, d" :
+      $ord == -1 ? " order by b desc, c desc, d desc" : die "not here";
     for my $p1 (@$a1) {
-    my $cnt1 = $p1->{cnt} * @val * @val;
-    print "select count(*) - $cnt1 from $table";
-    print " where $p1->{exp};\n";
+      my $res = [ $p1->{res}, \@val, \@val ];
+      $sql = "select b, c, d from $table" .
+	     " where $p1->{exp}" .
+	     $orderby;
+      verify($sql, $ord, $res);
       for my $p2 (@$a2) {
-      my $cnt2 = $p1->{cnt} * $p2->{cnt} * @val;
-      print "select count(*) - $cnt2 from $table";
-      print " where $p1->{exp} and $p2->{exp};\n";
+	my $res = [ $p1->{res}, $p2->{res}, \@val ];
+	$sql = "select b, c, d from $table" .
+	       " where $p1->{exp} and $p2->{exp}" .
+	       $orderby;
+	verify($sql, $ord, $res);
 	for my $p3 (@$a3) {
-	my $cnt3 = $p1->{cnt} * $p2->{cnt} * $p3->{cnt};
-	print "select count(*) - $cnt3 from $table";
-	print " where $p1->{exp} and $p2->{exp} and $p3->{exp};\n";
+	  my $res = [ $p1->{res}, $p2->{res}, $p3->{res} ];
+	  $sql = "select b, c, d from $table" .
+		 " where $p1->{exp} and $p2->{exp} and $p3->{exp}" .
+		 $orderby;
+	  verify($sql, $ord, $res);
 	}
       }
     }
-  print <<EOF;
-drop table $table;
-EOF
+  }
+  warn "drop table\n";
+  $sql = "drop table $table";
+  $dbh->do($sql) or die $DBI::errstr;
 }
 
+warn "verified $casecnt cases\n";
+warn "done\n";
+
 # vim: set sw=2:

mysql-test/r/ndb_index_ordered.result

@@ -275,6 +275,114 @@ a	b	c
 1	1	1
 4	4	NULL
 drop table t1;
+create table t1 (
+a int unsigned primary key,
+b int unsigned,
+c char(10),
+key bc (b, c)
+) engine=ndb;
+insert into t1 values(1,1,'a'),(2,2,'b'),(3,3,'c'),(4,4,'d'),(5,5,'e');
+insert into t1 select a*7,10*b,'f' from t1;
+insert into t1 select a*13,10*b,'g' from t1;
+insert into t1 select a*17,10*b,'h' from t1;
+insert into t1 select a*19,10*b,'i' from t1;
+insert into t1 select a*23,10*b,'j' from t1;
+insert into t1 select a*29,10*b,'k' from t1;
+select b, c from t1 where b <= 10 and c <'f' order by b, c;
+b	c
+1	a
+2	b
+3	c
+4	d
+5	e
+select b, c from t1 where b <= 10 and c <'f' order by b desc, c desc;
+b	c
+5	e
+4	d
+3	c
+2	b
+1	a
+select b, c from t1 where b=4000 and c<'k' order by b, c;
+b	c
+4000	h
+4000	i
+4000	i
+4000	i
+4000	j
+4000	j
+4000	j
+4000	j
+4000	j
+4000	j
+select b, c from t1 where b=4000 and c<'k' order by b desc, c desc;
+b	c
+4000	j
+4000	j
+4000	j
+4000	j
+4000	j
+4000	j
+4000	i
+4000	i
+4000	i
+4000	h
+select b, c from t1 where 1000<=b and b<=100000 and c<'j' order by b, c;
+b	c
+1000	h
+1000	i
+1000	i
+1000	i
+2000	h
+2000	i
+2000	i
+2000	i
+3000	h
+3000	i
+3000	i
+3000	i
+4000	h
+4000	i
+4000	i
+4000	i
+5000	h
+5000	i
+5000	i
+5000	i
+10000	i
+20000	i
+30000	i
+40000	i
+50000	i
+select b, c from t1 where 1000<=b and b<=100000 and c<'j' order by b desc, c desc;
+b	c
+50000	i
+40000	i
+30000	i
+20000	i
+10000	i
+5000	i
+5000	i
+5000	i
+5000	h
+4000	i
+4000	i
+4000	i
+4000	h
+3000	i
+3000	i
+3000	i
+3000	h
+2000	i
+2000	i
+2000	i
+2000	h
+1000	i
+1000	i
+1000	i
+1000	h
+select min(b), max(b) from t1;
+min(b)	max(b)
+1	5000000
 CREATE TABLE test1 (
 SubscrID int(11) NOT NULL auto_increment,
 UsrID int(11) NOT NULL default '0',

mysql-test/t/ndb_index_ordered.test

@@ -147,6 +147,35 @@ select * from t1 use index (bc) where b < 4 order by a;
 select * from t1 use index (bc) where b IS NOT NULL order by a;
 drop table t1;
 
+#
+# Order by again, including descending.
+#
+
+create table t1 (
+  a int unsigned primary key,
+  b int unsigned,
+  c char(10),
+  key bc (b, c)
+) engine=ndb;
+
+insert into t1 values(1,1,'a'),(2,2,'b'),(3,3,'c'),(4,4,'d'),(5,5,'e');
+insert into t1 select a*7,10*b,'f' from t1;
+insert into t1 select a*13,10*b,'g' from t1;
+insert into t1 select a*17,10*b,'h' from t1;
+insert into t1 select a*19,10*b,'i' from t1;
+insert into t1 select a*23,10*b,'j' from t1;
+insert into t1 select a*29,10*b,'k' from t1;
+#
+select b, c from t1 where b <= 10 and c <'f' order by b, c;
+select b, c from t1 where b <= 10 and c <'f' order by b desc, c desc;
+#
+select b, c from t1 where b=4000 and c<'k' order by b, c;
+select b, c from t1 where b=4000 and c<'k' order by b desc, c desc;
+select b, c from t1 where 1000<=b and b<=100000 and c<'j' order by b, c;
+select b, c from t1 where 1000<=b and b<=100000 and c<'j' order by b desc, c desc;
+#
+select min(b), max(b) from t1;
+
 #
 # Bug #6435
 CREATE TABLE test1 (

sql/ha_ndbcluster.h

@@ -96,6 +96,7 @@ class ha_ndbcluster: public handler
   int index_prev(byte *buf);
   int index_first(byte *buf);
   int index_last(byte *buf);
+  int index_read_last(byte * buf, const byte * key, uint key_len);
   int rnd_init(bool scan);
   int rnd_end();
   int rnd_next(byte *buf);
@@ -176,7 +177,7 @@ class ha_ndbcluster: public handler
 			byte *buf);
   int ordered_index_scan(const key_range *start_key,
 			 const key_range *end_key,
-			 bool sorted, byte* buf);
+			 bool sorted, bool descending, byte* buf);
   int full_table_scan(byte * buf);
   int fetch_next(NdbScanOperation* op);
   int next_result(byte *buf);