Data structures for approximate queries.

CMakeLists.txt

@@ -30,7 +30,9 @@ set(CMAKE_CXX_FLAGS_RELWITHDEBINFO "${CMAKE_CXX_FLAGS_RELWITHDEBINFO} -Wall -ped
 set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fno-strict-aliasing")
 set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -ldl")
 
-set(CMAKE_BUILD_TYPE RelWithDebInfo)
+if (NOT CMAKE_BUILD_TYPE)
+  set(CMAKE_BUILD_TYPE "RelWithDebInfo" CACHE STRING "" FORCE)
+endif()
 
 if ("${CMAKE_SOURCE_DIR}" STREQUAL "${CMAKE_BINARY_DIR}")
   message(FATAL_ERROR "In-source builds are not allowed.")
@@ -69,23 +71,23 @@ add_subdirectory(libconfluo)
 
 if (BUILD_RPC)
   # RPC Framework
-  add_subdirectory(librpc)
+  #add_subdirectory(librpc)
 
   # Python Client
-  if (WITH_PY_CLIENT)
-    add_subdirectory(pyclient)
-  endif()
-
-  # Java Client
-  if (WITH_JAVA_CLIENT)
-    add_subdirectory(javaclient)
-  endif()
+  #  if (WITH_PY_CLIENT)
+  #    add_subdirectory(pyclient)
+  #  endif()
+  #  
+  #  # Java Client
+  #  if (WITH_JAVA_CLIENT)
+  #    add_subdirectory(javaclient)
+  #  endif()
 endif()
 
 # Confluo examples
 if (BUILD_EXAMPLES)
-  add_subdirectory(examples/libtimeseries)
-  add_subdirectory(examples/libstreaming)
+    #  add_subdirectory(examples/libtimeseries)
+    #  add_subdirectory(examples/libstreaming)
 endif()
 
 if (BUILD_DOC)

build.sh

@@ -3,7 +3,7 @@ set -e
 
 mkdir -p build
 cd build
-cmake ..
+cmake "$@" .. 
 
 START=$(date +%s)
 make

libconfluo/CMakeLists.txt

@@ -1,4 +1,4 @@
-set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-unused-parameter")
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-unused-parameter -I/Users/Ujval/dev/research/univmon_extension/simulator/V0.3/")
 if (NOT APPLE AND UNIX)
   set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -Wl,--no-as-needed -ldl")
 endif ()
@@ -95,6 +95,14 @@ add_library(confluo STATIC
         confluo/container/monolog/monolog_linear.h
         confluo/container/monolog/monolog_linear_bucket.h
         confluo/container/radix_tree.h
+        confluo/container/sketch
+        confluo/container/sketch/confluo_universal_sketch.h 
+        confluo/container/sketch/count_sketch.h
+        confluo/container/sketch/hash_manager.h
+        confluo/container/sketch/priority_queue.h
+        confluo/container/sketch/sketch_utils.h
+        confluo/container/sketch/substream_summary.h
+        confluo/container/sketch/universal_sketch.h
         confluo/schema/field.h
         confluo/schema/record.h
         confluo/schema/record_batch.h
@@ -119,8 +127,9 @@ add_library(confluo STATIC
         confluo/conf
         confluo/conf/configuration_params.h
         confluo/conf/defaults.h
-        confluo/filter.h
         confluo/alert.h
+        confluo/filter.h
+        confluo/univ_sketch_log.h
         src/confluo_store.cc
         src/atomic_multilog.cc
         src/filter.cc
@@ -140,6 +149,7 @@ add_library(confluo STATIC
         src/container/cursor/alert_cursor.cc
         src/container/cursor/offset_cursors.cc
         src/container/cursor/record_cursors.cc
+        src/container/sketch/hash_manager.cc
         src/parser/aggregate_parser.cc
         src/parser/expression_compiler.cc
         src/parser/expression_parser.cc
@@ -185,10 +195,17 @@ add_library(confluo STATIC
         src/types/raw_data.cc
         src/types/numeric.cc
         src/types/type_properties.cc
-        src/types/type_manager.cc)
+        src/types/type_manager.cc )
+
 target_link_libraries(confluo confluoutils ${CMAKE_THREAD_LIBS_INIT} ${lz4_STATIC_LIB})
 add_dependencies(confluo lz4)
 
+add_executable(usexperiment
+    test/sketch_experiment.cc
+    )
+find_package(Boost 1.45.0 COMPONENTS program_options)
+target_link_libraries(usexperiment confluo ${Boost_LIBRARIES})
+
 if (BUILD_TESTS)
   # Build test
   add_executable(ctest
@@ -222,6 +239,8 @@ if (BUILD_TESTS)
           test/container/cursor/batched_cursor_test.h
           test/container/bitmap/bitmap_test.h
           test/container/bitmap/bitmap_array_test.h
+          test/container/sketch/count_sketch_test.h
+          test/container/sketch/universal_sketch_test.h
           test/container/bitmap/delta_encoded_array_test.h
           test/container/stream_test.h
           test/container/string_map_test.h
@@ -238,8 +257,8 @@ if (BUILD_TESTS)
           test/compression/lz4_encode_test.h
           test/compression/delta_encode_test.h
           test/aggregated_reflog_test.h
-          test/confluo_store_test.h)
-  target_link_libraries(ctest confluo gtest gtest_main)
+          test/confluo_store_test.h test/container/sketch/priority_queue_test.h)
+  target_link_libraries(ctest confluo gtest gtest_main) #alan_univmon)
   add_dependencies(ctest googletest)
 
   file(GLOB_RECURSE test_sources test/*.cc)

libconfluo/confluo/atomic_multilog.h

@@ -22,6 +22,7 @@
 #include "container/cursor/record_cursors.h"
 #include "container/cursor/alert_cursor.h"
 #include "container/monolog/monolog.h"
+#include "container/sketch/confluo_universal_sketch.h"
 #include "container/radix_tree.h"
 #include "container/string_map.h"
 #include "filter.h"
@@ -42,9 +43,11 @@
 #include "string_utils.h"
 #include "threads/periodic_task.h"
 #include "threads/task_pool.h"
+#include "univ_sketch_log.h"
 
 using namespace ::confluo::archival;
 using namespace ::confluo::monolog;
+using namespace ::confluo::sketch;
 using namespace ::confluo::index;
 using namespace ::confluo::monitor;
 using namespace ::confluo::parser;
@@ -78,6 +81,11 @@ class atomic_multilog {
    */
   typedef metadata_writer metadata_writer_type;
 
+  /**
+   * Identifier for universal sketch
+   */
+  typedef size_t univ_sketch_id_t;
+
   /**
    * Identifier for filter
    */
@@ -174,6 +182,19 @@ class atomic_multilog {
    */
   bool is_indexed(const std::string &field_name);
 
+  /**
+   * Adds a universal sketch ton monitor a field.
+   * @param name The name of the sketch
+   * @param field_name The name of the field to monitor.
+   * @param epsilon margin of error
+   */
+  void add_universal_sketch(const std::string &name, const std::string &field_name, double epsilon = 0.01);
+
+  void remove_universal_sketch(const std::string &name);
+
+  template<typename g_ret_t>
+  g_ret_t evaluate_frequency_fn(const std::string &name, std::function<g_ret_t(int64_t)> &g_fn);
+
   /**
    * Adds filter to the atomic multilog
    * @param name The name of the filter
@@ -434,6 +455,13 @@ class atomic_multilog {
    */
   void remove_index_task(const std::string &field_name, optional<management_exception> &ex);
 
+  void add_universal_sketch_task(const std::string &name,
+                                 const std::string &field_name,
+                                 double epsilon,
+                                 optional<management_exception> &ex);
+
+  void remove_universal_sketch_task(const std::string &name, optional<management_exception> &ex);
+
   /**
    * Adds a filter to be executed on the data
    *
@@ -534,9 +562,13 @@ class atomic_multilog {
   filter_log filters_;
   /** The list of indexes */
   index_log indexes_;
+  /** The list of universal sketches */
+  univ_sketch_log univ_sketches_;
   /** The list of alerts */
   alert_index alerts_;
 
+  /** A map from id to universal_sketch */
+  string_map<univ_sketch_id_t> univ_sketch_map_;
   /** A map from id to filter */
   string_map<filter_id_t> filter_map_;
   /** A map from id to aggregate */

libconfluo/confluo/container/monolog/monolog_exp2.h

@@ -421,8 +421,23 @@ class monolog_exp2 : public monolog_exp2_base<T, NBUCKETS> {
   /**
    * Constructs a default monolog iterator
    */
-  monolog_exp2()
-      : tail_(0) {
+  monolog_exp2(size_t count = 0)
+      : tail_(count) {
+  }
+
+  monolog_exp2(const monolog_exp2& other) {
+    atomic::store(&tail_, other.size());
+    for (size_t i = 0; i < this->size(); i++) {
+      this->set(i, other.at(i));
+    }
+  }
+
+  monolog_exp2& operator=(const monolog_exp2& other) {
+    atomic::store(&tail_, other.size());
+    for (size_t i = 0; i < this->size(); i++) {
+      this->set(i, other.at(i));
+    }
+    return *this;
   }
 
   /**

libconfluo/confluo/container/monolog/monolog_exp2_linear.h

@@ -432,8 +432,23 @@ class monolog_exp2_linear : public monolog_exp2_linear_base<T, NCONTAINERS, BUCK
   /**
    * Constructs a relaxed monolog implementation
    */
-  monolog_exp2_linear()
-      : tail_(0) {
+  monolog_exp2_linear(size_t count = 0)
+      : tail_(count) {
+  }
+
+  monolog_exp2_linear(const monolog_exp2_linear& other) {
+    atomic::store(&tail_, other.size());
+    for (size_t i = 0; i < this->size(); i++) {
+      this->set(i, other.at(i));
+    }
+  }
+
+  monolog_exp2_linear& operator=(const monolog_exp2_linear& other) {
+    atomic::store(&tail_, other.size());
+    for (size_t i = 0; i < this->size(); i++) {
+      this->set(i, other.at(i));
+    }
+    return *this;
   }
 
   /**