bk commit into 5.1 tree (pekka:1.2600)

[pekka]

Below is the list of changes that have just been committed into a local
5.1 repository of pekka.  When pekka does a push these changes
will be propagated to the main repository and, within 24 hours after the
push, to the public repository.
For information on how to access the public repository
see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html

ChangeSet@stripped, 2008-05-20 11:02:28+03:00, pekka@stripped +5 -0
  wl#2124 05_fix01
  cache quicksort blew up, use heapsort
  handler stat_size,stat_scale had no effect due to missing code

  sql/ha_ndbcluster.cc@stripped, 2008-05-20 11:02:18+03:00, pekka@stripped +11 -4
    wl#2124 05_fix01

  storage/ndb/src/kernel/blocks/dbtux/DbtuxStat.cpp@stripped, 2008-05-20 11:02:14+03:00,
pekka@stripped +1 -1
    wl#2124 05_fix01

  storage/ndb/src/ndbapi/NdbIndexStatImpl.cpp@stripped, 2008-05-20 11:02:12+03:00,
pekka@stripped +134 -44
    wl#2124 05_fix01

  storage/ndb/src/ndbapi/NdbIndexStatImpl.hpp@stripped, 2008-05-20 11:02:08+03:00,
pekka@stripped +10 -1
    wl#2124 05_fix01

  storage/ndb/tools/ndb_index_stat.cpp@stripped, 2008-05-20 11:02:15+03:00,
pekka@stripped +1 -1
    wl#2124 05_fix01

diff -Nrup a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
--- a/sql/ha_ndbcluster.cc	2008-05-15 21:43:19 +03:00
+++ b/sql/ha_ndbcluster.cc	2008-05-20 11:02:18 +03:00
@@ -11083,7 +11083,7 @@ struct Ndb_index_stat_opt {
     evict_batch(32),
     evict_delay(60),
     cache_limit(8*1024*1024),
-    analyze_loop(100),
+    analyze_loop(300),
     fudge_pct(10),
     stat_size(ndb_index_stat_uint_null),
     stat_scale(ndb_index_stat_uint_null)
@@ -11627,9 +11627,12 @@ ndb_index_stat_share(NDB_SHARE *share,
 
   if (st != 0)
   {
-    /* Move to top of current list */
-    st->access_time= now;
-    ndb_index_stat_list_move(st, st->list_type, true);
+    if (add)
+    {
+      /* Move to top of current list */
+      st->access_time= now;
+      ndb_index_stat_list_move(st, st->list_type, true);
+    }
   }
   else if (add && (st= ndb_index_stat_alloc()) != 0)
   {
@@ -11912,10 +11915,14 @@ ndb_index_stat_proc_load(Ndb_index_stat_
     return;
   }
 
+  is->param_get(param);
+
   if (opt.stat_size != ndb_index_stat_uint_null)
     param.stat_size= opt.stat_size;
   if (opt.stat_scale != ndb_index_stat_uint_null)
     param.stat_scale= opt.stat_scale;
+
+  is->param_set(param);
 
   if (is->do_scan_load() == -1)
   {
diff -Nrup a/storage/ndb/src/kernel/blocks/dbtux/DbtuxStat.cpp
b/storage/ndb/src/kernel/blocks/dbtux/DbtuxStat.cpp
--- a/storage/ndb/src/kernel/blocks/dbtux/DbtuxStat.cpp	2008-05-15 21:40:50 +03:00
+++ b/storage/ndb/src/kernel/blocks/dbtux/DbtuxStat.cpp	2008-05-20 11:02:14 +03:00
@@ -142,7 +142,7 @@ Dbtux::statScanInit(ScanOpPtr scanPtr, c
     }
 
     const double max_allowed = size * factor;
-    const double value_bytes = (1 + num_key_attrs) * sizeof(float);
+    const double value_bytes = (1 + num_key_attrs) * 2 * sizeof(float);
     const double avg_sample_bytes = avg_key_bytes + value_bytes;
     const double sample_count = max_allowed / avg_sample_bytes;
     const double frequency = entries / sample_count;
diff -Nrup a/storage/ndb/src/ndbapi/NdbIndexStatImpl.cpp
b/storage/ndb/src/ndbapi/NdbIndexStatImpl.cpp
--- a/storage/ndb/src/ndbapi/NdbIndexStatImpl.cpp	2008-05-15 21:42:00 +03:00
+++ b/storage/ndb/src/ndbapi/NdbIndexStatImpl.cpp	2008-05-20 11:02:12 +03:00
@@ -2639,7 +2639,7 @@ NdbIndexStatImpl::save_close()
   if (m_cacheBuild != 0) {
     Cache& c = *m_cacheBuild;
     if (!c.valid) {
-      delete m_cacheBuild;
+      cache_free(m_cacheBuild);
       m_cacheBuild = 0;
     }
   }
@@ -2721,7 +2721,9 @@ inline uint
 NdbIndexStatImpl::Cache::get_addr(uint pos) const
 {
   assert(pos < sampleCount);
-  const Uint8* src = &addrArray[pos * addrLen];
+  const uint offset = pos * addrLen;
+  assert(offset + addrLen <= addrBytes);
+  const Uint8* src = &addrArray[offset];
   uint addr = 0;
   switch (addrLen) {
   case 4:
@@ -2743,7 +2745,9 @@ inline void
 NdbIndexStatImpl::Cache::set_addr(uint pos, uint addr)
 {
   assert(pos < sampleCount);
-  Uint8* dst = &addrArray[pos * addrLen];
+  const uint offset = pos * addrLen;
+  assert(offset + addrLen <= addrBytes);
+  Uint8* dst = &addrArray[offset];
   switch (addrLen) {
   case 4:
     dst[3] = (addr >> 24) & 0xFF;
@@ -2822,6 +2826,8 @@ NdbIndexStatImpl::Cache::swap_entry(uint
 
     set_addr(pos2, hold_addr);
     memcpy(get_valueptr(pos2), hold_value, valueLen);
+
+    sort_swap_cnt++;
   }
 }
 
@@ -3000,53 +3006,137 @@ NdbIndexStatImpl::cache_cmpaddr(const Ca
 
   int ret = keyData1.data_cmp(keyData2);
   assert(addr1 == addr2 || ret == -1 || ret == +1);
+
+  c.sort_cmp_cnt++;
   return ret;
 }
 
 int
+NdbIndexStatImpl::cache_cmppos(const Cache& c, uint pos1, uint pos2) const
+{
+  uint addr1 = c.get_addr(pos1);
+  uint addr2 = c.get_addr(pos2);
+  return cache_cmpaddr(c, addr1, addr2);
+}
+
+/*
+ * Sort addr and value arrays via key values.
+ *
+ * The samples were inserted in key order and were read back
+ * via index scan (not ACC scan) so they may be nearly ordered
+ * at first.  This is quicksort worst case so we do not use it.
+ */
+int
 NdbIndexStatImpl::cache_sort(Cache& c)
 {
-  if (c.sampleCount != 0) {
-    if (cache_sort(c, 0, c.sampleCount - 1) == -1)
-      return -1;
-  }
+  if (c.sampleCount > 1)
+    cache_hsort(c);
+
+  DBUG_PRINT("index_stat", ("sort: entries=%d cmp=%d swap=%d",
+                             c.sampleCount, c.sort_cmp_cnt, c.sort_swap_cnt));
   return 0;
 }
 
-// sort addr and value arrays via key values
-int
-NdbIndexStatImpl::cache_sort(Cache& c, int lo, int hi)
+// insertion sort - expensive
+void
+NdbIndexStatImpl::cache_isort(Cache& c)
+{
+  int n = c.sampleCount;
+  int i;
+  for (i = 1; i < n; i++) {
+    int j;
+    for (j = i - 1; j >= 0; j--) {
+      int ret = cache_cmppos(c, j, j + 1);
+      if (ret < 0)
+        break;
+      c.swap_entry(j, j + 1);
+    }
+  }
+}
+
+// heapsort
+void
+NdbIndexStatImpl::cache_hsort(Cache& c)
 {
-  assert(lo < (int)c.sampleCount && hi < (int)c.sampleCount && lo
<= hi);
+  int count = c.sampleCount;
+  int i;
 
-  // based on a mawk example, does not use pivot for temp storage
+  // highest entry which can have children
+  i = count / 2;
 
-  const uint pivot_addr = c.get_addr(lo);
-  int i = lo;
-  int j = hi;
-  while (i <= j) {
-    while (cache_cmpaddr(c, c.get_addr(i), pivot_addr) < 0)
-      i++;
-    while (cache_cmpaddr(c, c.get_addr(j), pivot_addr) > 0)
-      j--;
-    if (i <= j) {
-      c.swap_entry(i, j);
-      i++;
-      j--;
-    }
-  }
-  if (j > lo)
-    cache_sort(c, lo, j);
-  if (i < hi)
-    cache_sort(c, i, hi);
-  return 0;
+  // make into heap (binary tree where child < parent)
+  while (i >= 0) {
+    cache_hsort_sift(c, i, count);
+    i--;
+  }
+#ifdef VM_TRACE
+  cache_hsort_verify(c, count);
+#endif
+
+  // sort
+  i = count - 1;
+  while (i > 0) {
+    // move current max to proper position
+    c.swap_entry(0, i);
+
+    // restore heap property for the rest
+    cache_hsort_sift(c, 0, i);
+#ifdef VM_TRACE
+    cache_hsort_verify(c, i);
+#endif
+    i--;
+  }
+}
+ 
+void
+NdbIndexStatImpl::cache_hsort_sift(Cache& c, int i, int count)
+{
+  int parent = i;
+
+  while (1) {
+    // left child if any
+    int child = parent * 2 + 1;
+    if (! (child < count))
+      break;
+
+    // replace by right child if bigger
+    if (child + 1 < count && cache_cmppos(c, child, child + 1) == -1)
+      child = child + 1;
+
+    // done if both children are less than parent
+    if (cache_cmppos(c, child, parent) == -1)
+      break;
+
+    c.swap_entry(parent, child);
+    parent = child;
+  }
+}
+
+// verify heap property
+void
+NdbIndexStatImpl::cache_hsort_verify(Cache& c, int count)
+{
+  int i;
+  for (i = 0; i < count; i++) {
+    int parent = i;
+    int child1 = 2 * i + 1;
+    int child2 = 2 * i + 2;
+    if (child1 < count) {
+      assert(cache_cmppos(c, child1, parent) == -1);
+      c.sort_cmp_cnt--;
+    }
+    if (child2 < count) {
+      assert(cache_cmppos(c, child2, parent) == -1);
+      c.sort_cmp_cnt--;
+    }
+  }
 }
 
 int
 NdbIndexStatImpl::cache_verify(const Cache& c)
 {
   DBUG_ENTER("NdbIndexStatImpl::cache_verify");
-  DBUG_PRINT("info", ("sampleCount: %u", c.sampleCount));
+  DBUG_PRINT("index_stat", ("sampleCount: %u", c.sampleCount));
 
   uint pos1;
   for (pos1 = 0; pos1 < c.sampleCount; pos1++) {
@@ -3067,12 +3157,12 @@ NdbIndexStatImpl::cache_verify(const Cac
       if (ret != -1) {
 #ifndef DBUG_OFF
         char buf[8000];
-        DBUG_PRINT("info", ("pos1=%u %s",
-                            pos1, keyData1.print(buf, sizeof(buf))));
-        DBUG_PRINT("info", ("pos2=%u %s",
-                            pos2, keyData2.print(buf, sizeof(buf))));
+        DBUG_PRINT("index_stat", ("pos1=%u %s",
+                                  pos1, keyData1.print(buf, sizeof(buf))));
+        DBUG_PRINT("index_stat", ("pos2=%u %s",
+                                  pos2, keyData2.print(buf, sizeof(buf))));
 #endif
-        DBUG_PRINT("info", ("disorder at pos=%u-%u", pos1, pos2));
+        DBUG_PRINT("index_stat", ("disorder at pos=%u-%u", pos1, pos2));
         setError(InvalidCache, __LINE__);
         DBUG_RETURN(-1);
       }
@@ -3102,9 +3192,9 @@ NdbIndexStatImpl::cache_print(const Cach
       c.get_value(pos, &x, i, 1);
       sprintf(buf2 + strlen(buf2), "%s%.2f", i == 0 ? "" : " ", x);
     }
-    DBUG_PRINT("info", ("pos: %u addr: %u", pos, addr));
-    DBUG_PRINT("info", ("key: %s", buf1));
-    DBUG_PRINT("info", ("value: %s", buf2));
+    DBUG_PRINT("index_stat", ("pos: %u addr: %u", pos, addr));
+    DBUG_PRINT("index_stat", ("key: %s", buf1));
+    DBUG_PRINT("index_stat", ("value: %s", buf2));
   }
   // total value
   {
@@ -3116,7 +3206,7 @@ NdbIndexStatImpl::cache_print(const Cach
       memcpy(&x, c.valueTotal + i * Cache::ValueSize, Cache::ValueSize);
       sprintf(buf2 + strlen(buf2), "%s%.2f", i == 0 ? "" : " ", x);
     }
-    DBUG_PRINT("info", ("total value: %s", buf2));
+    DBUG_PRINT("index_stat", ("total value: %s", buf2));
   }
 }
 #endif
@@ -3699,9 +3789,9 @@ NdbIndexStatImpl::setError(int code, int
   m_errorLine = line;
 
 #ifdef VM_TRACE
-  if (NdbEnv_GetEnv("NDB_INDEX_STAT_ABORT_ON_ERROR", (char*)0, 0)) {
-    abort();
-  }
+  const char* p = NdbEnv_GetEnv("NDB_INDEX_STAT_ABORT_ON_ERROR", (char*)0, 0);
+  if (p != 0 && strchr("1Y", p[0]) != 0)
+    assert(false);
 #endif
 }
 
diff -Nrup a/storage/ndb/src/ndbapi/NdbIndexStatImpl.hpp
b/storage/ndb/src/ndbapi/NdbIndexStatImpl.hpp
--- a/storage/ndb/src/ndbapi/NdbIndexStatImpl.hpp	2008-05-15 21:41:58 +03:00
+++ b/storage/ndb/src/ndbapi/NdbIndexStatImpl.hpp	2008-05-20 11:02:08 +03:00
@@ -363,6 +363,9 @@ private:
     Uint8* valueTotal;
     Cache* nextClean;
     enum { ValueSize = sizeof(float) };
+    // sorting statistics
+    mutable int sort_cmp_cnt;
+    mutable int sort_swap_cnt;
     Cache() {
       valid = false;
       histograms = false;
@@ -384,6 +387,8 @@ private:
       valueArray = 0;
       valueTotal = 0;
       nextClean = 0;
+      sort_cmp_cnt = 0;
+      sort_swap_cnt = 0;
     };
     // pos is index < sampleCount, addr is offset in keyArray
     uint get_addr(uint pos) const;
@@ -403,8 +408,12 @@ private:
   int cache_insert(Cache& c);
   int cache_commit(Cache& c);
   int cache_cmpaddr(const Cache& c, uint addr1, uint addr2) const;
+  int cache_cmppos(const Cache& c, uint pos1, uint pos2) const;
   int cache_sort(Cache& c);
-  int cache_sort(Cache& c, int lo, int hi);
+  void cache_isort(Cache& c);
+  void cache_hsort(Cache& c);
+  void cache_hsort_sift(Cache& c, int i, int count);
+  void cache_hsort_verify(Cache& c, int count);
   int cache_verify(const Cache& c);
 #ifndef DBUG_OFF
   void cache_print(const Cache& c);
diff -Nrup a/storage/ndb/tools/ndb_index_stat.cpp b/storage/ndb/tools/ndb_index_stat.cpp
--- a/storage/ndb/tools/ndb_index_stat.cpp	2008-05-15 21:42:02 +03:00
+++ b/storage/ndb/tools/ndb_index_stat.cpp	2008-05-20 11:02:15 +03:00
@@ -180,7 +180,7 @@ int main(int argc, char** argv)
 
 #define ERR_IS(is) \
   ERR(is.getNdbError()); \
-  g_err << "impl line" << is.getNdbErrorLine() << endl
+  g_err << "impl line " << is.getNdbErrorLine() << endl
 
   int l;
   for (l = 0; l < _loop || _loop == 0; l++) {