release/src/router/mysql/storage/ndb/test/ndbapi/testIndexStat.cpp

   1 /* Copyright (c) 2005-2007 MySQL AB
   2
   3    This program is free software; you can redistribute it and/or modify
   4    it under the terms of the GNU General Public License as published by
   5    the Free Software Foundation; version 2 of the License.
   6
   7    This program is distributed in the hope that it will be useful,
   8    but WITHOUT ANY WARRANTY; without even the implied warranty of
   9    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  10    GNU General Public License for more details.
  11
  12    You should have received a copy of the GNU General Public License
  13    along with this program; if not, write to the Free Software
  14    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA */
  15
  16 #include <ndb_global.h>
  17 #include <ndb_opts.h>
  18 #include <NdbApi.hpp>
  19 #include <NdbIndexStat.hpp>
  20 #include <NdbTest.hpp>
  21 #include <my_sys.h>
  22 #include <ndb_version.h>
  23 #include <math.h>
  24
  25 /*
  26  * Sample results:
  27  *
  28  * 0. err pct: count: 1000 min: -99.99 max: 99.92 avg: 6.88 stddev: 27.61
  29  *
  30  * 0. baseline with same options as handler
  31  */
  32
  33 #undef min
  34 #undef max
  35 #define min(a, b) ((a) <= (b) ? (a) : (b))
  36 #define max(a, b) ((a) >= (b) ? (a) : (b))
  37
  38 inline NdbOut&
  39 NdbOut::operator<<(double x)
  40 {
  41   char buf[100];
  42   sprintf(buf, "%.2f", x);
  43   *this << buf;
  44   return *this;
  45 }
  46
  47 struct Opts {
  48   int loglevel;
  49   uint seed;
  50   uint loop;
  51   uint rows;
  52   uint ops;
  53   uint nullkeys;
  54   uint dupkeys;
  55   uint scanpct;
  56   uint eqscans;
  57   uint dupscans;
  58   my_bool keeptable;
  59   my_bool loaddata;
  60   my_bool nochecks;
  61   my_bool abort;
  62   // internal
  63   uint tryhard;
  64   Opts() :
  65     loglevel(0),
  66     seed(-1),
  67     loop(1),
  68     rows(100000),
  69     ops(1000),
  70     nullkeys(10),
  71     dupkeys(1000),
  72     scanpct(5),
  73     eqscans(50),
  74     dupscans(10),
  75     keeptable(false),
  76     loaddata(true),
  77     nochecks(false),
  78     abort(false),
  79     // internal
  80     tryhard(20)
  81   {}
  82 };
  83
  84 static Opts g_opts;
  85 const char* g_progname = "testIndexStat";
  86 static uint g_loop = 0;
  87
  88 static const char* g_tabname = "ts0";
  89 static const char* g_indname = "ts0x1";
  90 static const char g_numattrs = 3;
  91 static const uint g_charlen = 10;
  92 static const char* g_csname = "latin1_swedish_ci";
  93 static CHARSET_INFO* g_cs;
  94
  95 // value and bound ranges
  96 static uint g_val_b_max = 10;
  97 static uint g_bnd_b_max = 20;
  98 static const char* g_val_c_char = "bcd";
  99 static const char* g_bnd_c_char = "abcde";
 100 static uint g_val_d_max = 100;
 101 static uint g_bnd_d_max = 200;
 102
 103 static Ndb_cluster_connection* g_ncc = 0;
 104 static Ndb* g_ndb = 0;
 105 static NdbDictionary::Dictionary* g_dic = 0;
 106 static const NdbDictionary::Table* g_tab = 0;
 107 static const NdbDictionary::Index* g_ind = 0;
 108
 109 static NdbIndexStat* g_stat = 0;
 110
 111 static NdbTransaction* g_con = 0;
 112 static NdbOperation* g_op = 0;
 113 static NdbScanOperation* g_scan_op = 0;
 114 static NdbIndexScanOperation* g_rangescan_op = 0;
 115
 116 static uint
 117 urandom()
 118 {
 119   uint r = (uint)random();
 120   return r;
 121 }
 122
 123 static uint
 124 urandom(uint m)
 125 {
 126   if (m == 0)
 127     return 0;
 128   uint r = urandom();
 129   r = r % m;
 130   return r;
 131 }
 132
 133 static int& g_loglevel = g_opts.loglevel; // default log level
 134
 135 #define chkdb(x) \
 136   do { if (likely(x)) break; ndbout << "line " << __LINE__ << " FAIL " << #x << endl; errdb(); if (g_opts.abort) abort(); return -1; } while (0)
 137
 138 #define chkrc(x) \
 139   do { if (likely(x)) break; ndbout << "line " << __LINE__ << " FAIL " << #x << endl; if (g_opts.abort) abort(); return -1; } while (0)
 140
 141 #define reqrc(x) \
 142   do { if (likely(x)) break; ndbout << "line " << __LINE__ << " ASSERT " << #x << endl; abort(); } while (0)
 143
 144 #define llx(n, x) \
 145   do { if (likely(g_loglevel < n)) break; ndbout << x << endl; } while (0)
 146
 147 #define ll0(x) llx(0, x)
 148 #define ll1(x) llx(1, x)
 149 #define ll2(x) llx(2, x)
 150 #define ll3(x) llx(3, x)
 151
 152 static void
 153 errdb()
 154 {
 155   uint any = 0;
 156   // g_ncc return no error...
 157   if (g_ndb != 0) {
 158     const NdbError& e = g_ndb->getNdbError();
 159     if (e.code != 0)
 160       ll0(++any << " ndb: error " << e);
 161   }
 162   if (g_dic != 0) {
 163     const NdbError& e = g_dic->getNdbError();
 164     if (e.code != 0)
 165       ll0(++any << " dic: error " << e);
 166   }
 167   if (g_con != 0) {
 168     const NdbError& e = g_con->getNdbError();
 169     if (e.code != 0)
 170       ll0(++any << " con: error " << e);
 171   }
 172   if (g_op != 0) {
 173     const NdbError& e = g_op->getNdbError();
 174     if (e.code != 0)
 175       ll0(++any << " op: error " << e);
 176   }
 177   if (g_scan_op != 0) {
 178     const NdbError& e = g_scan_op->getNdbError();
 179     if (e.code != 0)
 180       ll0(++any << " scan_op: error " << e);
 181   }
 182   if (g_rangescan_op != 0) {
 183     const NdbError& e = g_rangescan_op->getNdbError();
 184     if (e.code != 0)
 185       ll0(++any << " rangescan_op: error " << e);
 186   }
 187   if (g_stat != 0) {
 188     const NdbError& e = g_stat->getNdbError();
 189     if (e.code != 0)
 190       ll0(++any << " stat: error " << e);
 191   }
 192   if (! any)
 193     ll0("unknown db error");
 194 }
 195
 196 // create table ts0 (
 197 //   a int unsigned, b smallint not null, c varchar(10), d int unsigned,
 198 //   primary key using hash (a), index (b, c, d) )
 199
 200 static int
 201 createtable()
 202 {
 203   NdbDictionary::Table tab(g_tabname);
 204   tab.setLogging(false);
 205   {
 206     NdbDictionary::Column col("a");
 207     col.setType(NdbDictionary::Column::Unsigned);
 208     col.setPrimaryKey(true);
 209     tab.addColumn(col);
 210   }
 211   {
 212     NdbDictionary::Column col("b");
 213     col.setType(NdbDictionary::Column::Smallint);
 214     col.setNullable(false);
 215     tab.addColumn(col);
 216   }
 217   {
 218     NdbDictionary::Column col("c");
 219     col.setType(NdbDictionary::Column::Varchar);
 220     col.setLength(g_charlen);
 221     col.setCharset(g_cs);
 222     col.setNullable(true);
 223     tab.addColumn(col);
 224   }
 225   {
 226     NdbDictionary::Column col("d");
 227     col.setType(NdbDictionary::Column::Unsigned);
 228     col.setNullable(true);
 229     tab.addColumn(col);
 230   }
 231   NdbDictionary::Index ind(g_indname);
 232   ind.setTable(g_tabname);
 233   ind.setType(NdbDictionary::Index::OrderedIndex);
 234   ind.setLogging(false);
 235   ind.addColumnName("b");
 236   ind.addColumnName("c");
 237   ind.addColumnName("d");
 238   g_dic = g_ndb->getDictionary();
 239   if (! g_opts.keeptable) {
 240     if (g_dic->getTable(g_tabname) != 0)
 241       chkdb(g_dic->dropTable(g_tabname) == 0);
 242     chkdb(g_dic->createTable(tab) == 0);
 243     chkdb(g_dic->createIndex(ind) == 0);
 244   } else {
 245     if (g_dic->getTable(g_tabname) == 0) {
 246       chkdb(g_dic->createTable(tab) == 0);
 247       chkdb(g_dic->createIndex(ind) == 0);
 248     } else
 249       g_opts.loaddata = false;
 250   }
 251   chkdb((g_tab = g_dic->getTable(g_tabname)) != 0);
 252   chkdb((g_ind = g_dic->getIndex(g_indname, g_tabname)) != 0);
 253   g_dic = 0;
 254   return 0;
 255 }
 256
 257 static int
 258 droptable()
 259 {
 260   g_dic = g_ndb->getDictionary();
 261   if (! g_opts.keeptable)
 262     chkdb(g_dic->dropTable(g_tabname) == 0);
 263   g_dic = 0;
 264   return 0;
 265 }
 266
 267 struct Val {
 268   Int16 b;
 269   bool c_null;
 270   uchar c[1 + g_charlen];
 271   bool d_null;
 272   Uint32 d;
 273   // partial values for use in Bnd
 274   uint numattrs;
 275   void make(uint n = g_numattrs, bool is_val = true);
 276   int cmp(const Val& val, uint n = g_numattrs) const;
 277 };
 278
 279 static NdbOut&
 280 operator<<(NdbOut& out, const Val& val)
 281 {
 282   out << "[";
 283   if (val.numattrs >= 1) {
 284     out << val.b;
 285   }
 286   if (val.numattrs >= 2) {
 287     out << " ";
 288     if (val.c_null)
 289       out << "NULL";
 290     else {
 291       char buf[1 + g_charlen];
 292       sprintf(buf, "%.*s", val.c[0], &val.c[1]);
 293       out << "'" << buf << "'";
 294     }
 295   }
 296   if (val.numattrs >= 3) {
 297     out << " ";
 298     if (val.d_null)
 299       out <<" NULL";
 300     else
 301       out << val.d;
 302   }
 303   out << "]";
 304   return out;
 305 }
 306
 307 void
 308 Val::make(uint n, bool is_val)
 309 {
 310   if (n >= 1) {
 311     uint b_max = is_val ? g_val_b_max : g_bnd_b_max;
 312     b = (int)urandom(2 * b_max) - (int)b_max;
 313   }
 314   if (n >= 2) {
 315     if (urandom(100) < g_opts.nullkeys)
 316       c_null = 1;
 317     else {
 318       const char* c_char = is_val ? g_val_c_char : g_bnd_c_char;
 319       // prefer shorter
 320       uint len = urandom(urandom(g_charlen + 2));
 321       c[0] = len;
 322       uint j;
 323       for (j = 0; j < len; j++) {
 324         uint k = urandom(strlen(c_char));
 325         c[1 + j] = c_char[k];
 326       }
 327       c_null = 0;
 328     }
 329   }
 330   if (n >= 3) {
 331     if (urandom(100) < g_opts.nullkeys)
 332       d_null = 1;
 333     else {
 334       uint d_max = is_val ? g_val_d_max : g_bnd_d_max;
 335       d = urandom(d_max);
 336       d_null = 0;
 337     }
 338   }
 339   numattrs = n;
 340 }
 341
 342 int
 343 Val::cmp(const Val& val, uint n) const
 344 {
 345   int k = 0;
 346   if (k == 0 && n >= 1) {
 347     if (b < val.b)
 348       k = -1;
 349     else if (b > val.b)
 350       k = +1;
 351   }
 352   if (k == 0 && n >= 2) {
 353     if (! c_null && ! val.c_null) {
 354       const uchar* s1 = &c[1];
 355       const uchar* s2 = &val.c[1];
 356       const uint l1 = (uint)c[0];
 357       const uint l2 = (uint)val.c[0];
 358       assert(l1 <= g_charlen && l2 <= g_charlen);
 359       k = g_cs->coll->strnncollsp(g_cs, s1, l1, s2, l2, 0);
 360     } else if (! c_null) {
 361       k = +1;
 362     } else if (! val.c_null) {
 363       k = -1;
 364     }
 365   }
 366   if (k == 0 && n >= 3) {
 367     if (! d_null && ! val.d_null) {
 368       if (d < val.d)
 369         k = -1;
 370       else if (d > val.d)
 371         k = +1;
 372     } else if (! d_null) {
 373       k = +1;
 374     } else if (! val.d_null) {
 375       k = -1;
 376     }
 377   }
 378   return k;
 379 }
 380
 381 struct Key {
 382   Val val;
 383   union {
 384     bool flag;
 385     uint count;
 386     uint rpk;
 387   };
 388 };
 389
 390 static NdbOut&
 391 operator<<(NdbOut& out, const Key& key)
 392 {
 393   out << key.val << " info:" << key.count;
 394   return out;
 395 }
 396
 397 static Key* g_keys = 0;
 398 static Key* g_sortkeys = 0;
 399 static uint g_sortcount = 0;
 400 static Key* g_minkey = 0;
 401 static Key* g_maxkey = 0;
 402
 403 static void
 404 freekeys()
 405 {
 406   if (g_keys != 0)
 407     my_free((char*)g_keys, MYF(0));
 408   if (g_sortkeys != 0)
 409     my_free((char*)g_sortkeys, MYF(0));
 410   g_keys = 0;
 411   g_sortkeys = 0;
 412 }
 413
 414 static int
 415 allockeys()
 416 {
 417   freekeys();
 418   size_t sz = sizeof(Key) * g_opts.rows;
 419   g_keys = (Key*)my_malloc(sz, MYF(0));
 420   g_sortkeys = (Key*)my_malloc(sz, MYF(0));
 421   chkrc(g_keys != 0 && g_sortkeys != 0);
 422   memset(g_keys, 0x1f, sz);
 423   memset(g_sortkeys, 0x1f, sz);
 424   return 0;
 425 }
 426
 427 static void
 428 makekeys()
 429 {
 430   uint i;
 431   for (i = 0; i < g_opts.rows; i++) {
 432     Key& key = g_keys[i];
 433     key.val.make();
 434     key.flag = false; // mark for dup generation done
 435   }
 436   for (i = 0; i < g_opts.rows; i++) {
 437     Key& key = g_keys[i];
 438     if (key.flag)
 439       continue;
 440     key.flag = true;
 441     uint fudge = 9;
 442     uint n = (urandom(fudge * (g_opts.dupkeys - 100)) + 99) / 100;
 443     uint k;
 444     for (k = 1; k < n; k++) {
 445       uint j = urandom(g_opts.rows);
 446       do {
 447         Key& dst = g_keys[j];
 448         if (! dst.flag) {
 449           dst.val = key.val;
 450           dst.flag = true;
 451           break;
 452         }
 453       } while (urandom(g_opts.tryhard) != 0);
 454     }
 455   }
 456 }
 457
 458 static int
 459 insertdata()
 460 {
 461   const uint batch = 512;
 462   chkdb((g_con = g_ndb->startTransaction()) != 0);
 463   uint i = 0;
 464   while (i < g_opts.rows) {
 465     chkdb((g_op = g_con->getNdbOperation(g_tab)) != 0);
 466     chkdb(g_op->insertTuple() == 0);
 467     Uint32 a = i;
 468     const Val& val = g_keys[i].val;
 469     const char* a_addr = (const char*)&a;
 470     const char* b_addr = (const char*)&val.b;
 471     const char* c_addr = ! val.c_null ? (const char*)val.c : 0;
 472     const char* d_addr = ! val.d_null ? (const char*)&val.d : 0;
 473     Uint32 no = 0;
 474     chkdb(g_op->equal(no++, a_addr) == 0);
 475     chkdb(g_op->setValue(no++, b_addr) == 0);
 476     chkdb(g_op->setValue(no++, c_addr) == 0);
 477     chkdb(g_op->setValue(no++, d_addr) == 0);
 478     if (i++ % batch == 0) {
 479       chkdb(g_con->execute(NdbTransaction::Commit) == 0);
 480       g_ndb->closeTransaction(g_con);
 481       g_con = 0;
 482       g_op = 0;
 483       chkdb((g_con = g_ndb->startTransaction()) != 0);
 484     }
 485   }
 486   chkdb(g_con->execute(NdbTransaction::Commit) == 0);
 487   g_ndb->closeTransaction(g_con);
 488   g_con = 0;
 489   g_op = 0;
 490   ll0(g_tabname << ": inserted " << g_opts.rows << " rows");
 491   return 0;
 492 }
 493
 494 static int
 495 countrows()
 496 {
 497   Uint64 rows = 0;
 498   Uint64 r;
 499   char* r_addr = (char*)&r;
 500   chkdb((g_con = g_ndb->startTransaction()) != 0);
 501   chkdb((g_scan_op = g_con->getNdbScanOperation(g_tab)) != 0);
 502   chkdb(g_scan_op->readTuples() == 0);
 503   chkdb(g_scan_op->interpret_exit_last_row() == 0);
 504   chkdb(g_scan_op->getValue(NdbDictionary::Column::ROW_COUNT, r_addr) != 0);
 505   chkdb(g_con->execute(NdbTransaction::NoCommit) == 0);
 506   while (1) {
 507     int ret;
 508     r = ~(Uint64)0;
 509     chkdb((ret = g_scan_op->nextResult()) == 0 || ret == 1);
 510     if (ret == 1)
 511       break;
 512     rows += r;
 513   }
 514   g_ndb->closeTransaction(g_con);
 515   g_con = 0;
 516   g_scan_op = 0;
 517   g_opts.rows = rows;
 518   return 0;
 519 }
 520
 521 static int
 522 scandata()
 523 {
 524   chkdb((g_con = g_ndb->startTransaction()) != 0);
 525   chkdb((g_scan_op = g_con->getNdbScanOperation(g_tab)) != 0);
 526   chkdb(g_scan_op->readTuples() == 0);
 527   Uint32 a;
 528   Val val;
 529   char* a_addr = (char*)&a;
 530   char* b_addr = (char*)&val.b;
 531   char* c_addr = (char*)val.c;
 532   char* d_addr = (char*)&val.d;
 533   Uint32 no = 0;
 534   NdbRecAttr* b_ra;
 535   NdbRecAttr* c_ra;
 536   NdbRecAttr* d_ra;
 537   chkdb(g_scan_op->getValue(no++, a_addr) != 0);
 538   chkdb((b_ra = g_scan_op->getValue(no++, b_addr)) != 0);
 539   chkdb((c_ra = g_scan_op->getValue(no++, c_addr)) != 0);
 540   chkdb((d_ra = g_scan_op->getValue(no++, d_addr)) != 0);
 541   chkdb(g_con->execute(NdbTransaction::NoCommit) == 0);
 542   uint count = 0;
 543   uint i;
 544   for (i = 0; i < g_opts.rows; i++)
 545     g_keys[i].count = 0;
 546   while (1) {
 547     int ret;
 548     a = ~(Uint32)0;
 549     chkdb((ret = g_scan_op->nextResult()) == 0 || ret == 1);
 550     if (ret == 1)
 551       break;
 552     assert(b_ra->isNULL() == 0 && c_ra->isNULL() != -1 && d_ra->isNULL() != -1);
 553     val.c_null = c_ra->isNULL();
 554     val.d_null = d_ra->isNULL();
 555     i = (uint)a;
 556     chkrc(i < g_opts.rows);
 557     Key& key = g_keys[i];
 558     if (g_opts.loaddata)
 559       chkrc(key.val.cmp(val) == 0);
 560     else
 561       key.val = val;
 562     key.count++;
 563     count++;
 564   }
 565   g_ndb->closeTransaction(g_con);
 566   g_con = 0;
 567   g_scan_op = 0;
 568   for (i = 0; i < g_opts.rows; i++)
 569     chkrc(g_keys[i].count == 1);
 570   assert(count == g_opts.rows);
 571   int level = g_opts.loaddata ? 1 : 0;
 572   llx(level, g_tabname << ": scanned " << g_opts.rows << " rows");
 573   return 0;
 574 }
 575
 576 static int
 577 loaddata()
 578 {
 579   if (g_opts.loaddata) {
 580     chkrc(allockeys() == 0);
 581     makekeys();
 582     chkrc(insertdata() == 0);
 583   } else {
 584     chkrc(countrows() == 0);
 585     chkrc(g_opts.rows != 0);
 586     ll0(g_tabname << ": using old table of " << g_opts.rows << " rows");
 587     chkrc(allockeys() == 0);
 588   }
 589   chkrc(scandata() == 0);
 590   uint i;
 591   for (i = 0; i < g_opts.rows; i++)
 592     ll3(i << ": " << g_keys[i]);
 593   return 0;
 594 }
 595
 596 // true = match, index = match or next higher
 597 static bool
 598 sortval(const Val& val, int& index)
 599 {
 600   if (unlikely(g_sortcount == 0)) {
 601     index = 0;
 602     return false;
 603   }
 604   int lo = -1;
 605   int hi = (int)g_sortcount;
 606   int ret;
 607   int j;
 608   do {
 609     j = (hi + lo) / 2;
 610     ret = val.cmp(g_sortkeys[j].val);
 611     if (ret < 0)
 612       hi = j;
 613     else if (ret > 0)
 614       lo = j;
 615     else
 616       break;
 617   } while (hi - lo > 1);
 618   if (ret == 0) {
 619     index = j;
 620     return true;
 621   }
 622   index = hi;
 623   return false;
 624 }
 625
 626 static void
 627 sortkeys()
 628 {
 629   // insert sort with binary search
 630   g_sortcount = 0;
 631   uint i;
 632   for (i = 0; i < g_opts.rows; i++) {
 633     const Val& val = g_keys[i].val;
 634     int index;
 635     bool match = sortval(val, index);
 636     Key& dst = g_sortkeys[index];
 637     if (match) {
 638       dst.rpk++;
 639     } else {
 640       uint bytes = ((int)g_sortcount - index) * sizeof(Key);
 641       memmove(&dst + 1, &dst, bytes);
 642       dst.val = val;
 643       dst.rpk = 1;
 644       g_sortcount++;
 645     }
 646   }
 647   g_minkey = &g_sortkeys[0];
 648   g_maxkey = &g_sortkeys[g_sortcount - 1];
 649   ll1("counted " << g_sortcount << " distinct keys");
 650 }
 651
 652 struct Bnd {
 653   Val val;
 654   /*
 655    * A bound is a partial key value (0 to g_numattrs attributes).
 656    * It is not equal to any key value.  Instead, it has a "side".
 657    *
 658    * side = 0 if the bound is empty
 659    * side = -1 if the bound is "just before" its value
 660    * side = +1 if the bound is "just after" its value
 661    *
 662    * This is another way of looking at strictness of non-empty
 663    * start and end keys in a range.
 664    *
 665    * start key is strict if side = +1
 666    * end key is strict if side = -1
 667    *
 668    * NDB API specifies strictness in the bound type of the last
 669    * index attribute which is part of the start/end key.
 670    *
 671    * LE (0) - strict: n - side: -1
 672    * LT (1) - strict: y - side: +1
 673    * GE (2) - strict: n - side: +1
 674    * GT (3) - strict: y - side: -1
 675    *
 676    * A non-empty bound divides keys into 2 disjoint subsets:
 677    * keys before (cmp() == -1) and keys after (cmp() == +1).
 678    */
 679   int side;
 680   Bnd& make(uint minattrs);
 681   Bnd& make(uint minattrs, const Val& theval);
 682   int cmp(const Val& val) const;
 683   int type(uint lohi, uint colno) const; // for setBound
 684 };
 685
 686 static NdbOut&
 687 operator<<(NdbOut& out, const Bnd& bnd)
 688 {
 689   out << bnd.val;
 690   out << " side: " << bnd.side;
 691   return out;
 692 }
 693
 694 Bnd&
 695 Bnd::make(uint minattrs)
 696 {
 697   uint numattrs = minattrs + urandom(g_numattrs - minattrs);
 698   val.make(numattrs, false);
 699   side = val.numattrs == 0 ? 0 : urandom(2) == 0 ? -1 : +1;
 700   return *this;
 701 }
 702
 703 Bnd&
 704 Bnd::make(uint minattrs, const Val& theval)
 705 {
 706   uint numattrs = minattrs + urandom(g_numattrs - minattrs);
 707   val = theval;
 708   val.numattrs = numattrs;
 709   side = val.numattrs == 0 ? 0 : urandom(2) == 0 ? -1 : +1;
 710   return *this;
 711 }
 712
 713 int
 714 Bnd::cmp(const Val& theval) const
 715 {
 716   int place; // debug
 717   int ret;
 718   do {
 719     assert(theval.numattrs == g_numattrs);
 720     int k = theval.cmp(val, val.numattrs);
 721     if (k != 0) {
 722       place = 1;
 723       ret = k;
 724       break;
 725     }
 726     if (side != 0) {
 727       place = 2;
 728       ret = -side;
 729       break;
 730     }
 731     place = 3;
 732     ret = 0;
 733     assert(val.numattrs == 0);
 734   } while (0);
 735   ll3("cmp: val: " << theval << " bnd: " << *this <<
 736       " return: " << ret << " at " << place);
 737   return ret;
 738 }
 739
 740 int
 741 Bnd::type(uint lohi, uint colno) const
 742 {
 743   int t;
 744   assert(lohi <= 1 && colno < val.numattrs && (side == -1 || side == +1));
 745   if (lohi == 0) {
 746     if (colno + 1 < val.numattrs)
 747       t = 0; // LE
 748     else if (side == -1)
 749       t = 0; // LE
 750     else
 751       t = 1; // LT
 752   } else {
 753     if (colno + 1 < val.numattrs)
 754       t = 2; // GE
 755     else if (side == +1)
 756       t = 2; // GE
 757     else
 758       t = 3; // GT
 759   }
 760   return t;
 761 }
 762
 763 struct Range {
 764   Bnd bnd[2];
 765   uint minattrs() const;
 766   uint maxattrs() const;
 767   int cmp(const Val& val) const; // -1,0,+1 = key is before,in,after range
 768   uint rowcount() const;
 769   bool iseq() const;
 770   // stats
 771   bool flag;
 772   uint statrows;
 773   uint scanrows;
 774   double errpct;
 775 };
 776
 777 static NdbOut&
 778 operator<<(NdbOut& out, const Range& range)
 779 {
 780   out << "bnd0: " << range.bnd[0] << " bnd1: " << range.bnd[1];
 781   return out;
 782 }
 783
 784 uint
 785 Range::minattrs() const
 786 {
 787   return min(bnd[0].val.numattrs, bnd[1].val.numattrs);
 788 }
 789
 790 uint
 791 Range::maxattrs() const
 792 {
 793   return max(bnd[0].val.numattrs, bnd[1].val.numattrs);
 794 }
 795
 796 int
 797 Range::cmp(const Val& theval) const
 798 {
 799   int place; // debug
 800   int ret;
 801   do  {
 802     int k;
 803     k = bnd[0].cmp(theval);
 804     if (k < 0) {
 805       place = 1;
 806       ret = -1;
 807       break;
 808     }
 809     k = bnd[1].cmp(theval);
 810     if (k > 0) {
 811       place = 2;
 812       ret = +1;
 813       break;
 814     }
 815     place = 3;
 816     ret = 0;
 817   } while (0);
 818   ll3("cmp: val: " << theval << " range: " << *this <<
 819       " return: " << ret << " at " << place);
 820   return ret;
 821 }
 822
 823 uint
 824 Range::rowcount() const
 825 {
 826   ll2("rowcount: " << *this);
 827   int i;
 828   // binary search for first and last in range
 829   int lim[2];
 830   for (i = 0; i <= 1; i++) {
 831     ll3("search i=" << i);
 832     int lo = -1;
 833     int hi = (int)g_sortcount;
 834     int ret;
 835     int j;
 836     do {
 837       j = (hi + lo) / 2;
 838       ret = cmp(g_sortkeys[j].val);
 839       if (i == 0) {
 840         if (ret < 0)
 841           lo = j;
 842         else
 843           hi = j;
 844       } else {
 845         if (ret > 0)
 846           hi = j;
 847         else
 848           lo = j;
 849       }
 850     } while (hi - lo > 1);
 851     if (ret == 0)
 852       lim[i] = j;
 853     else if (i == 0)
 854       lim[i] = hi;
 855     else
 856       lim[i] = lo;
 857   }
 858   // the range
 859   const int lo = max(lim[0], 0);
 860   const int hi = min(lim[1], (int)g_sortcount - 1);
 861   if (! g_opts.nochecks) {
 862     int curr = -1;
 863     for (i = 0; i < (int)g_sortcount; i++) {
 864       int k = cmp(g_sortkeys[i].val);
 865       if (k < 0)
 866         assert(i < lo);
 867       else if (k == 0)
 868         assert(lo <= i && i <= hi);
 869       else
 870         assert(i > hi);
 871       assert(curr <= k);
 872       if (curr < k)
 873         curr = k;
 874     }
 875   }
 876   // sum them up
 877   uint count = 0;
 878   for (i = lo; i <= hi; i++)
 879     count += g_sortkeys[i].count;
 880   ll2("count: " << count << " index lim: " << lim[0] << " " << lim[1]);
 881   return count;
 882 }
 883
 884 bool
 885 Range::iseq() const
 886 {
 887   return
 888     minattrs() == maxattrs() &&
 889     bnd[0].val.cmp(bnd[1].val, minattrs()) == 0 &&
 890     bnd[0].side < bnd[1].side;
 891 }
 892
 893 static Range* g_ranges = 0;
 894
 895 static void
 896 freeranges()
 897 {
 898   if (g_ranges != 0)
 899     my_free((char*)g_ranges, MYF(0));
 900   g_ranges = 0;
 901 }
 902
 903 static int
 904 allocranges()
 905 {
 906   freeranges();
 907   size_t sz = sizeof(Range) * g_opts.ops;
 908   g_ranges = (Range*)my_malloc(sz, MYF(0));
 909   chkrc(g_ranges != 0);
 910   memset(g_ranges, 0x1f, sz);
 911   return 0;
 912 }
 913
 914 static void
 915 makeranges()
 916 {
 917   uint i;
 918   for (i = 0; i < g_opts.ops; i++) {
 919     Range& range = g_ranges[i];
 920     range.flag = false; // mark for dup generation done
 921     bool fulleq = (urandom(100) < g_opts.eqscans);
 922     bool eq = fulleq || (urandom(100) < g_opts.eqscans);
 923     bool matcheq = eq && (urandom(10) != 0);
 924     if (! eq) {
 925       // random but prefer non-empty and no more than scanpct
 926       do {
 927         range.bnd[0].make(0);
 928         range.bnd[1].make(0);
 929         uint count = range.rowcount();
 930         if (count != 0 && 100 * count <= g_opts.scanpct * g_opts.rows)
 931           break;
 932       } while (urandom(g_opts.tryhard) != 0);
 933     } else {
 934       uint minattrs = fulleq ? g_numattrs : 1;
 935       if (! matcheq) {
 936         range.bnd[0].make(minattrs);
 937       } else {
 938         uint m = urandom(g_sortcount);
 939         const Val& val = g_sortkeys[m].val;
 940         range.bnd[0].make(minattrs, val);
 941       }
 942       range.bnd[1] = range.bnd[0];
 943       range.bnd[0].side = -1;
 944       range.bnd[1].side = +1;
 945       // fix types
 946       range.bnd[0];
 947       range.bnd[1];
 948       assert(range.iseq());
 949     }
 950   }
 951   for (i = 0; i < g_opts.ops; i++) {
 952     Range& range = g_ranges[i];
 953     if (range.flag)
 954       continue;
 955     range.flag = true;
 956     if (urandom(100) < g_opts.dupscans) {
 957       uint j = urandom(g_opts.ops);
 958       do {
 959         Range& dst = g_ranges[j];
 960         if (! dst.flag) {
 961           dst.bnd[0] = range.bnd[0];
 962           dst.bnd[1] = range.bnd[1];
 963           dst.flag = true;
 964           break;
 965         }
 966       } while (urandom(g_opts.tryhard) != 0);
 967     }
 968   }
 969 }
 970
 971 static int
 972 setbounds(const Range& range)
 973 {
 974   // currently must do each attr in order
 975   ll2("setbounds: " << range);
 976   uint i;
 977   const Bnd (&bnd)[2] = range.bnd;
 978   for (i = 0; i < g_numattrs; i++) {
 979     const Uint32 no = i; // index attribute number
 980     uint j;
 981     int type[2] = { -1, -1 };
 982     for (j = 0; j <= 1; j++) {
 983       if (no < bnd[j].val.numattrs)
 984         type[j] = bnd[j].type(j, no);
 985     }
 986     for (j = 0; j <= 1; j++) {
 987       int t = type[j];
 988       if (t == -1)
 989         continue;
 990       if (no + 1 < bnd[j].val.numattrs)
 991         t &= ~(uint)1; // strict bit is set on last bound only
 992       const Val& val = bnd[j].val;
 993       const void* addr = 0;
 994       if (no == 0)
 995         addr = (const void*)&val.b;
 996       else if (no == 1)
 997         addr = ! val.c_null ? (const void*)val.c : 0;
 998       else if (no == 2)
 999         addr = ! val.d_null ? (const void*)&val.d : 0;
1000       else
1001         assert(false);
1002       ll2("setBound attr:" << no << " type:" << t << " val: " << val);
1003       chkdb(g_rangescan_op->setBound(no, t, addr) == 0);
1004     }
1005   }
1006   return 0;
1007 }
1008
1009 static int
1010 allocstat()
1011 {
1012   g_stat = new NdbIndexStat(g_ind);
1013   chkdb(g_stat->alloc_cache(32) == 0);
1014   return 0;
1015 }
1016
1017 static int
1018 runstat(Range& range, int flags)
1019 {
1020   ll2("runstat: " << range << " flags=" << flags);
1021   chkdb((g_con = g_ndb->startTransaction()) != 0);
1022   chkdb((g_rangescan_op = g_con->getNdbIndexScanOperation(g_ind, g_tab)) != 0);
1023   chkdb(g_rangescan_op->readTuples(NdbOperation::LM_CommittedRead) == 0);
1024   chkrc(setbounds(range) == 0);
1025   Uint64 count = ~(Uint64)0;
1026   chkdb(g_stat->records_in_range(g_ind, g_rangescan_op, g_opts.rows, &count, flags) == 0);
1027   g_ndb->closeTransaction(g_con);
1028   g_con = 0;
1029   g_rangescan_op = 0;
1030   range.statrows = (uint)count;
1031   chkrc((Uint64)range.statrows == count);
1032   ll2("stat: " << range.statrows);
1033   return 0;
1034 }
1035
1036 static int
1037 runscan(Range& range)
1038 {
1039   ll2("runscan: " << range);
1040   chkdb((g_con = g_ndb->startTransaction()) != 0);
1041   chkdb((g_rangescan_op = g_con->getNdbIndexScanOperation(g_ind, g_tab)) != 0);
1042   chkdb(g_rangescan_op->readTuples() == 0);
1043   chkrc(setbounds(range) == 0);
1044   Uint32 a;
1045   char* a_addr = (char*)&a;
1046   Uint32 no = 0;
1047   chkdb(g_rangescan_op->getValue(no++, a_addr) != 0);
1048   chkdb(g_con->execute(NdbTransaction::NoCommit) == 0);
1049   uint count = 0;
1050   uint i;
1051   for (i = 0; i < g_opts.rows; i++)
1052     g_keys[i].count = 0;
1053   while (1) {
1054     int ret;
1055     a = ~(Uint32)0;
1056     chkdb((ret = g_rangescan_op->nextResult()) == 0 || ret == 1);
1057     if (ret == 1)
1058       break;
1059     i = (uint)a;
1060     chkrc(i < g_opts.rows);
1061     Key& key = g_keys[i];
1062     ll3("scan: " << key);
1063     int k = range.cmp(key.val);
1064     chkrc(k == 0);
1065     chkrc(key.count == 0);
1066     key.count++;
1067     count++;
1068   }
1069   g_ndb->closeTransaction(g_con);
1070   g_con = 0;
1071   g_rangescan_op = 0;
1072   if (! g_opts.nochecks) {
1073     for (i = 0; i < g_opts.rows; i++) {
1074       const Key& key = g_keys[i];
1075       int k = range.cmp(key.val);
1076       assert((k != 0 && key.count == 0) || (k == 0 && key.count == 1));
1077     }
1078     assert(range.rowcount() == count);
1079   }
1080   range.scanrows = count;
1081   ll2("scan: " << range.scanrows);
1082   return 0;
1083 }
1084
1085 static int
1086 runscans()
1087 {
1088   uint i;
1089   for (i = 0; i < g_opts.ops; i++) {
1090     Range& range = g_ranges[i];
1091     ll1("range " << i << ": " << range);
1092     // simulate old handler code
1093     int flags = 0;
1094     if (i < 32 || i % 20 == 0)
1095       flags |= NdbIndexStat::RR_UseDb;
1096     chkrc(runstat(range, flags) == 0);
1097     chkrc(runscan(range) == 0);
1098     // if stat is 0 then it is exact scan count
1099     chkrc(range.statrows != 0 || range.scanrows == 0);
1100     // measure error as fraction of total rows
1101     double x = (double)range.statrows;
1102     double y = (double)range.scanrows;
1103     double z = (double)g_opts.rows;
1104     double err = (x - y) / z;
1105     // report in pct
1106     range.errpct = 100.0 * err;
1107     ll1("range " << i << ":" <<
1108         " stat: " << range.statrows << " scan: " << range.scanrows <<
1109         " errpct: " << range.errpct);
1110   }
1111   return 0;
1112 }
1113
1114 struct Stat {
1115   const char* name;
1116   uint count;
1117   double sum;
1118   double minval;
1119   double maxval;
1120   double avg;
1121   double varsum;
1122   double var;
1123   double stddev;
1124   void init();
1125   void add(const Stat& stat);
1126 };
1127
1128 void
1129 Stat::init()
1130 {
1131   name = "stat";
1132   count = 0;
1133   sum = minval = maxval = avg = varsum = var = stddev = 0.0;
1134 }
1135
1136 void
1137 Stat::add(const Stat& stat)
1138 {
1139   if (count == 0) {
1140     *this = stat;
1141     return;
1142   }
1143   Stat tmp = *this;
1144   tmp.count = count + stat.count;
1145   tmp.sum = sum + stat.sum;
1146   tmp.minval = minval <= stat.minval ? minval : stat.minval;
1147   tmp.maxval = maxval >= stat.maxval ? maxval : stat.maxval;
1148   tmp.avg = tmp.sum / double(tmp.count);
1149   tmp.varsum = varsum + stat.varsum;
1150   tmp.var = tmp.varsum / double(tmp.count);
1151   tmp.stddev = sqrt(tmp.var);
1152   *this = tmp;
1153 }
1154
1155 static NdbOut&
1156 operator<<(NdbOut& out, const Stat& stat)
1157 {
1158   out << stat.name << ": " << "count: " << stat.count
1159       << " min: " << stat.minval << " max: " << stat.maxval
1160       << " avg: " << stat.avg << " stddev: " << stat.stddev;
1161   return out;
1162 }
1163
1164 template <class T, class V>
1165 static void
1166 computestat(Stat& stat)
1167 {
1168   stat.init();
1169   stat.name = V::name();
1170   const T* array = V::array();
1171   stat.count = V::count();
1172   assert(stat.count != 0);
1173   uint i;
1174   for (i = 0; i < stat.count; i++) {
1175     const T& item = array[i];
1176     double data = V::data(item);
1177     stat.sum += data;
1178     if (i == 0)
1179       stat.minval = stat.maxval = data;
1180     else {
1181       if (stat.minval > data)
1182         stat.minval = data;
1183       if (stat.maxval < data)
1184         stat.maxval = data;
1185     }
1186   }
1187   stat.avg = stat.sum / double(stat.count);
1188   stat.varsum = 0.0;
1189   for (i = 0; i < stat.count; i++) {
1190     const T& item = array[i];
1191     double data = V::data(item);
1192     double x = data - stat.avg;
1193     stat.varsum += x * x;
1194   }
1195   stat.var = stat.varsum / double(stat.count);
1196   stat.stddev = sqrt(stat.var);
1197 }
1198
1199 struct V_rpk {
1200   static const char* name() { return "rec per key"; }
1201   static const Key* array() { return g_sortkeys; }
1202   static uint count() { return g_sortcount; }
1203   static double data(const Key& key) { return (double)key.rpk; }
1204 };
1205
1206 struct V_rir {
1207   static const char* name() { return "rir err pct"; }
1208   static const Range* array() { return g_ranges; }
1209   static uint count() { return g_opts.ops; }
1210   static double data(const Range& range) { return (double)range.errpct; }
1211 };
1212
1213 template void computestat<Key, V_rpk>(Stat& stat);
1214 template void computestat<Range, V_rir>(Stat& stat);
1215
1216 static Stat g_stat_rpk; // summaries over loops
1217 static Stat g_stat_rir;
1218
1219 static void
1220 loopstats()
1221 {
1222   Stat stat_rpk; // records per key
1223   Stat stat_rir; // record in range
1224   if (g_loop == 0) {
1225     g_stat_rpk.init();
1226     g_stat_rir.init();
1227   }
1228   computestat<Key, V_rpk>(stat_rpk);
1229   computestat<Range, V_rir>(stat_rir);
1230   if (g_opts.loop != 1) {
1231     ll0("=== loop " << g_loop << " summary ===");
1232     ll0(stat_rpk);
1233     ll0(stat_rir);
1234   }
1235   // accumulate
1236   g_stat_rpk.add(stat_rpk);
1237   g_stat_rir.add(stat_rir);
1238 }
1239
1240 static void
1241 finalstats()
1242 {
1243   ll0("=== summary ===");
1244   ll0(g_stat_rpk);
1245   ll0(g_stat_rir);
1246 }
1247
1248 static void
1249 setseed(int n)
1250 {
1251   uint seed;
1252   if (n == -1) {
1253     if (g_opts.seed == 0)
1254       return;
1255     if (g_opts.seed != -1)
1256       seed = (uint)g_opts.seed;
1257     else
1258       seed = 1 + (ushort)getpid();
1259   } else {
1260     if (g_opts.seed != 0)
1261       return;
1262     seed = n;
1263   }
1264   ll0("seed=" << seed);
1265   srandom(seed);
1266 }
1267
1268 static int
1269 runtest()
1270 {
1271   setseed(-1);
1272   g_cs = get_charset_by_name(g_csname, MYF(0));
1273   if (g_cs == 0)
1274     g_cs = get_charset_by_csname(g_csname, MY_CS_PRIMARY, MYF(0));
1275   chkrc(g_cs != 0);
1276   for (g_loop = 0; g_opts.loop == 0 || g_loop < g_opts.loop; g_loop++) {
1277     ll0("=== loop " << g_loop << " ===");
1278     setseed(g_loop);
1279     chkrc(createtable() == 0);
1280     chkrc(loaddata() == 0);
1281     sortkeys();
1282     chkrc(allocranges() == 0);
1283     makeranges();
1284     chkrc(allocstat() == 0);
1285     chkrc(runscans() == 0);
1286     chkrc(droptable() == 0);
1287     loopstats();
1288   }
1289   finalstats();
1290   return 0;
1291 }
1292
1293 NDB_STD_OPTS_VARS;
1294
1295 static struct my_option
1296 my_long_options[] =
1297 {
1298   NDB_STD_OPTS("testIndexStat"),
1299   { "loglevel", 1001, "Logging level in this program 0-3 (default 0)",
1300     (uchar **)&g_opts.loglevel, (uchar **)&g_opts.loglevel, 0,
1301     GET_INT, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },
1302   { "seed", 1002, "Random seed (0=loop number, default -1=random)",
1303     (uchar **)&g_opts.seed, (uchar **)&g_opts.seed, 0,
1304     GET_INT, REQUIRED_ARG, -1, 0, 0, 0, 0, 0 },
1305   { "loop", 1003, "Number of test loops (default 1, 0=forever)",
1306     (uchar **)&g_opts.loop, (uchar **)&g_opts.loop, 0,
1307     GET_INT, REQUIRED_ARG, 1, 0, 0, 0, 0, 0 },
1308   { "rows", 1004, "Number of rows (default 100000)",
1309     (uchar **)&g_opts.rows, (uchar **)&g_opts.rows, 0,
1310     GET_UINT, REQUIRED_ARG, 100000, 0, 0, 0, 0, 0 },
1311   { "ops", 1005, "Number of index scans per loop (default 1000)",
1312     (uchar **)&g_opts.ops, (uchar **)&g_opts.ops, 0,
1313     GET_UINT, REQUIRED_ARG, 1000, 0, 0, 0, 0, 0 },
1314   { "dupkeys", 1006, "Pct records per key (min 100, default 1000)",
1315     (uchar **)&g_opts.dupkeys, (uchar **)&g_opts.dupkeys, 0,
1316     GET_UINT, REQUIRED_ARG, 1000, 0, 0, 0, 0, 0 },
1317   { "scanpct", 1007, "Preferred max pct of total rows per scan (default 5)",
1318     (uchar **)&g_opts.scanpct, (uchar **)&g_opts.scanpct, 0,
1319     GET_UINT, REQUIRED_ARG, 5, 0, 0, 0, 0, 0 },
1320   { "nullkeys", 1008, "Pct nulls in each key attribute (default 10)",
1321     (uchar **)&g_opts.nullkeys, (uchar **)&g_opts.nullkeys, 0,
1322     GET_UINT, REQUIRED_ARG, 10, 0, 0, 0, 0, 0 },
1323   { "eqscans", 1009, "Pct scans for partial/full equality (default 50)",
1324     (uchar **)&g_opts.eqscans, (uchar **)&g_opts.eqscans, 0,
1325     GET_UINT, REQUIRED_ARG, 50, 0, 0, 0, 0, 0 },
1326   { "dupscans", 1010, "Pct scans using same bounds (default 10)",
1327     (uchar **)&g_opts.dupscans, (uchar **)&g_opts.dupscans, 0,
1328     GET_UINT, REQUIRED_ARG, 10, 0, 0, 0, 0, 0 },
1329   { "keeptable", 1011, "Use existing table and data if any and do not drop",
1330     (uchar **)&g_opts.keeptable, (uchar **)&g_opts.keeptable, 0,
1331     GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
1332   { "no-extra-checks", 1012, "Omit expensive consistency checks",
1333     (uchar **)&g_opts.nochecks, (uchar **)&g_opts.nochecks, 0,
1334     GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
1335   { "abort-on-error", 1013, "Dump core on any error",
1336     (uchar **)&g_opts.abort, (uchar **)&g_opts.abort, 0,
1337     GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
1338   { 0, 0, 0,
1339     0, 0, 0,
1340     GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0 }
1341 };
1342
1343 static void
1344 usage()
1345 {
1346   ndbout
1347     << g_progname
1348     << ": measure records_in_range error as percentage of total rows" << endl;
1349   my_print_help(my_long_options);
1350 }
1351
1352 static int
1353 checkoptions()
1354 {
1355   chkrc(g_opts.rows != 0);
1356   chkrc(g_opts.nullkeys <= 100);
1357   chkrc(g_opts.dupkeys >= 100);
1358   chkrc(g_opts.scanpct <= 100);
1359   chkrc(g_opts.eqscans <= 100);
1360   chkrc(g_opts.dupscans <= 100);
1361   return 0;
1362 }
1363
1364 static int
1365 doconnect()
1366 {
1367   g_ncc = new Ndb_cluster_connection();
1368   chkdb(g_ncc->connect(30) == 0);
1369   g_ndb = new Ndb(g_ncc, "TEST_DB");
1370   chkdb(g_ndb->init() == 0 && g_ndb->waitUntilReady(30) == 0);
1371   return 0;
1372 }
1373
1374 static void
1375 freeall()
1376 {
1377   delete g_stat;
1378   freekeys();
1379   freeranges();
1380   delete g_ndb;
1381   delete g_ncc;
1382 }
1383
1384 int
1385 main(int argc, char** argv)
1386 {
1387   ndb_init();
1388   const char* g_progname =
1389     strchr(argv[0], '/') ? strrchr(argv[0], '/') + 1 : argv[0];
1390   uint i;
1391   ndbout << g_progname;
1392   for (i = 1; i < argc; i++)
1393     ndbout << " " << argv[i];
1394   ndbout << endl;
1395   int ret;
1396   ret = handle_options(&argc, &argv, my_long_options, ndb_std_get_one_option);
1397   if (ret != 0 || argc != 0)
1398     return NDBT_ProgramExit(NDBT_WRONGARGS);
1399   if (checkoptions() == 0 && doconnect() == 0 && runtest() == 0) {
1400     freeall();
1401     return NDBT_ProgramExit(NDBT_OK);
1402   }
1403   freeall();
1404   return NDBT_ProgramExit(NDBT_FAILED);
1405 }