Implement BIT_(AND|OR|XOR) functions as window functions.

sql/item_sum.cc

@@ -2101,6 +2101,8 @@ longlong Item_sum_bit::val_int()
 void Item_sum_bit::clear()
 {
   bits= reset_bits;
+  if (as_window_function)
+    clear_as_window();
 }
 
 Item *Item_sum_or::copy_or_same(THD* thd)
@@ -2108,15 +2110,79 @@ Item *Item_sum_or::copy_or_same(THD* thd)
   return new (thd->mem_root) Item_sum_or(thd, this);
 }
 
+bool Item_sum_bit::clear_as_window()
+{
+  memset(bit_counters, 0, sizeof(bit_counters));
+  num_values_added= 0;
+  set_bits_from_counters();
+  return 0;
+}
+
+bool Item_sum_bit::remove_as_window(ulonglong value)
+{
+  DBUG_ASSERT(as_window_function);
+  for (int i= 0; i < NUM_BIT_COUNTERS; i++)
+  {
+    if (!bit_counters[i])
+    {
+      // Don't attempt to remove values that were never added.
+      DBUG_ASSERT((value & (1 << i)) == 0);
+      continue;
+    }
+    bit_counters[i]-= (value & (1 << i)) ? 1 : 0;
+  }
+  DBUG_ASSERT(num_values_added > 0);
+  // Prevent overflow;
+  num_values_added = std::min(num_values_added, num_values_added - 1);
+  set_bits_from_counters();
+  return 0;
+}
+
+bool Item_sum_bit::add_as_window(ulonglong value)
+{
+  DBUG_ASSERT(as_window_function);
+  for (int i= 0; i < NUM_BIT_COUNTERS; i++)
+  {
+    bit_counters[i]+= (value & (1 << i)) ? 1 : 0;
+  }
+  // Prevent overflow;
+  num_values_added = std::max(num_values_added, num_values_added + 1);
+  set_bits_from_counters();
+  return 0;
+}
+
+void Item_sum_or::set_bits_from_counters()
+{
+  ulonglong value= 0;
+  for (int i= 0; i < NUM_BIT_COUNTERS; i++)
+  {
+    value|= bit_counters[i] > 0 ? (1 << i) : 0;
+  }
+  bits= value | reset_bits;
+}
 
 bool Item_sum_or::add()
 {
   ulonglong value= (ulonglong) args[0]->val_int();
   if (!args[0]->null_value)
+  {
+    if (as_window_function)
+      return add_as_window(value);
     bits|=value;
+  }
   return 0;
 }
 
+void Item_sum_xor::set_bits_from_counters()
+{
+  ulonglong value= 0;
+  for (int i= 0; i < NUM_BIT_COUNTERS; i++)
+  {
+    value|= (bit_counters[i] % 2) ? (1 << i) : 0;
+  }
+  bits= value ^ reset_bits;
+}
+
 Item *Item_sum_xor::copy_or_same(THD* thd)
 {
   return new (thd->mem_root) Item_sum_xor(thd, this);
@@ -2127,10 +2193,31 @@ bool Item_sum_xor::add()
 {
   ulonglong value= (ulonglong) args[0]->val_int();
   if (!args[0]->null_value)
+  {
+    if (as_window_function)
+      return add_as_window(value);
     bits^=value;
+  }
   return 0;
 }
 
+void Item_sum_and::set_bits_from_counters()
+{
+  ulonglong value= 0;
+  if (!num_values_added)
+  {
+    bits= reset_bits;
+    return;
+  }
+
+  for (int i= 0; i < NUM_BIT_COUNTERS; i++)
+  {
+    // We've only added values of 1 for this bit.
+    if (bit_counters[i] == num_values_added)
+      value|= (1 << i);
+  }
+  bits= value & reset_bits;
+}
 Item *Item_sum_and::copy_or_same(THD* thd)
 {
   return new (thd->mem_root) Item_sum_and(thd, this);
@@ -2141,7 +2228,11 @@ bool Item_sum_and::add()
 {
   ulonglong value= (ulonglong) args[0]->val_int();
   if (!args[0]->null_value)
+  {
+    if (as_window_function)
+      return add_as_window(value);
     bits&=value;
+  }
   return 0;
 }
 
@@ -2329,6 +2420,10 @@ void Item_sum_bit::reset_field()
 
 void Item_sum_bit::update_field()
 {
+  // We never call update_field when computing the function as a window
+  // function. Setting bits to a random value invalidates the bits counters and
+  // the result of the bit function becomes erroneous.
+  DBUG_ASSERT(!as_window_function);
   uchar *res=result_field->ptr;
   bits= uint8korr(res);
   add();

sql/item_sum.h

@@ -1031,14 +1031,18 @@ class Item_sum_max :public Item_sum_hybrid
 
 class Item_sum_bit :public Item_sum_int
 {
-protected:
-  ulonglong reset_bits,bits;
-
 public:
   Item_sum_bit(THD *thd, Item *item_par, ulonglong reset_arg):
-    Item_sum_int(thd, item_par), reset_bits(reset_arg), bits(reset_arg) {}
+    Item_sum_int(thd, item_par), reset_bits(reset_arg), bits(reset_arg),
+    as_window_function(FALSE), num_values_added(0) {}
   Item_sum_bit(THD *thd, Item_sum_bit *item):
-    Item_sum_int(thd, item), reset_bits(item->reset_bits), bits(item->bits) {}
+    Item_sum_int(thd, item), reset_bits(item->reset_bits), bits(item->bits),
+    as_window_function(item->as_window_function),
+    num_values_added(item->num_values_added)
+  {
+    if (as_window_function)
+      memcpy(bit_counters, item->bit_counters, sizeof(bit_counters));
+  }
   enum Sumfunctype sum_func () const {return SUM_BIT_FUNC;}
   void clear();
   longlong val_int();
@@ -1049,8 +1053,42 @@ class Item_sum_bit :public Item_sum_int
   void cleanup()
   {
     bits= reset_bits;
+    if (as_window_function)
+      clear_as_window();
     Item_sum_int::cleanup();
   }
+  void setup_window_func(THD *thd __attribute__((unused)),
+                         Window_spec *window_spec __attribute__((unused)))
+  {
+    as_window_function= TRUE;
+    clear_as_window();
+  }
+  void remove()
+  {
+    if (as_window_function)
+    {
+      remove_as_window(args[0]->val_int());
+      return;
+    }
+    // Unless we're counting bits, we can not remove anything.
+    DBUG_ASSERT(0);
+  }
+
+protected:
+  static const int NUM_BIT_COUNTERS= 64;
+  ulonglong reset_bits,bits;
+  /*
+    Marks whether the function is to be computed as a window function.
+  */
+  bool as_window_function;
+  // When used as an aggregate window function, we need to store
+  // this additional information.
+  ulonglong num_values_added;
+  ulonglong bit_counters[NUM_BIT_COUNTERS];
+  bool add_as_window(ulonglong value);
+  bool remove_as_window(ulonglong value);
+  bool clear_as_window();
+  virtual void set_bits_from_counters()= 0;
 };
 
 
@@ -1062,28 +1100,37 @@ class Item_sum_or :public Item_sum_bit
   bool add();
   const char *func_name() const { return "bit_or("; }
   Item *copy_or_same(THD* thd);
+
+private:
+  void set_bits_from_counters();
 };
 
 
 class Item_sum_and :public Item_sum_bit
 {
-  public:
+public:
   Item_sum_and(THD *thd, Item *item_par):
     Item_sum_bit(thd, item_par, ULONGLONG_MAX) {}
   Item_sum_and(THD *thd, Item_sum_and *item) :Item_sum_bit(thd, item) {}
   bool add();
   const char *func_name() const { return "bit_and("; }
   Item *copy_or_same(THD* thd);
+
+private:
+  void set_bits_from_counters();
 };
 
 class Item_sum_xor :public Item_sum_bit
 {
-  public:
+public:
   Item_sum_xor(THD *thd, Item *item_par): Item_sum_bit(thd, item_par, 0) {}
   Item_sum_xor(THD *thd, Item_sum_xor *item) :Item_sum_bit(thd, item) {}
   bool add();
   const char *func_name() const { return "bit_xor("; }
   Item *copy_or_same(THD* thd);
+
+private:
+  void set_bits_from_counters();
 };
 
 

sql/sql_window.cc

@@ -1226,6 +1226,7 @@ bool JOIN::process_window_functions(List<Item> *curr_fields_list)
               break;
             }
           case Item_sum::COUNT_FUNC:
+          case Item_sum::SUM_BIT_FUNC:
           {
             /*
               Frame-aware window function computation. It does one pass, but