Merge pull request #21 from DropB1t/interval-join

[ADD] Interval Join Operator Implementation

Author: mencagli

.gitignore

@@ -1,2 +1,2 @@
 # .DS_Store are banished
-.DS_Store
+.DS_Store

API

@@ -158,6 +158,11 @@ The corresponding builder needs two parameters (for the lift and combine logics)
  * Combine
     __host__ __device__ void(const result_t &, const result_t &, result_t &);
 
+Interval_Join
+----------------
+std::optional<result_t> (const tuple_t &, const tuple_t &)
+std::optional<result_t> (const tuple_t &, const tuple_t &, RuntimeContext &)
+
 SINK
 ----
 void(std::optional<tuple_t> &);

docs/windflow-doxygen.conf

@@ -845,8 +845,9 @@ INPUT                  = ../wf/doxygen-mainpage.hpp \
                          ../wf/source_shipper.hpp \
                          ../wf/windflow.hpp \
                          ../wf/windflow_gpu.hpp \
-                         ../wf/kafka/windflow_kafka.hpp
-                         ../wf/rocksdb/windflow_rocksdb.hpp
+                         ../wf/kafka/windflow_kafka.hpp \
+                         ../wf/rocksdb/windflow_rocksdb.hpp \
+                         ../wf/interval_join.hpp
 
 # This tag can be used to specify the character encoding of the source files
 # that doxygen parses. Internally doxygen uses the UTF-8 encoding. Doxygen uses

tests/CMakeLists.txt

@@ -3,6 +3,7 @@ add_subdirectory(graph_tests)
 add_subdirectory(merge_tests)
 add_subdirectory(split_tests)
 add_subdirectory(win_tests)
+add_subdirectory(join_tests)
 
 # Add custom target for all the CPU tests
 add_custom_target(all_cpu)
@@ -11,11 +12,11 @@ if(LIBRDKAFKA_FOUND AND LibRDKafka_LIBRARIES)
     # Add the sub-folder with Kafka tests
     add_subdirectory(kafka_tests)
     # Add dependencies to the all_cpu target
-    add_dependencies(all_cpu graph_tests merge_tests split_tests win_tests kafka_tests)
+    add_dependencies(all_cpu graph_tests merge_tests split_tests win_tests join_tests kafka_tests)
 else()
     message(STATUS "librdkafka needs to be installed to generate kafka tests")
     # Add dependencies to the all_cpu target
-    add_dependencies(all_cpu graph_tests merge_tests split_tests win_tests)
+    add_dependencies(all_cpu graph_tests merge_tests split_tests win_tests join_tests)
 endif()
 
 if(ROCKSDB_FOUND AND ROCKSDB_LIBRARIES)

tests/join_tests/CMakeLists.txt

@@ -0,0 +1,33 @@
+# Compiler and flags
+set(CMAKE_CXX_STANDARD 17)
+set(CMAKE_CXX_STANDARD_REQUIRED ON)
+set(CMAKE_CXX_FLAGS_DEBUG "-g -O0")
+set(CMAKE_CXX_FLAGS_RELEASE "-O3 -g -finline-functions")
+
+# Macros to be provided to the compiler
+add_definitions(-DFF_BOUNDED_BUFFER)
+# -DWF_JOIN_MEASUREMENT to enable the measurement of how uniformerly the tuples are distributed among the joiners
+# -DWF_TRACING_ENABLED to enable the tracing with Dashboard
+
+# Header files of WindFlow and FastFlow
+include_directories(${PROJECT_SOURCE_DIR}/wf ${ff_root_dir})
+
+# Linking to pthread
+# cdt gvc cgraph to enable the tracing with Dashboard
+link_libraries(pthread)
+
+# Set output directory
+set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ../../bin/join_tests)
+
+# Cpp files to be compiled
+file(GLOB SOURCES "*.cpp")
+
+# Add a target for each cpp file and a unique target for all the tests in this folder
+add_custom_target(join_tests)
+
+foreach(testsourcefile ${SOURCES})
+    get_filename_component(barename ${testsourcefile} NAME)
+    string(REPLACE ".cpp" "" testname ${barename})
+    add_executable(${testname} ${testsourcefile})
+    add_dependencies(join_tests ${testname})
+endforeach(testsourcefile ${SOURCES})

tests/join_tests/join_common.hpp

@@ -0,0 +1,276 @@
+/**************************************************************************************
+ *  Copyright (c) 2023- Gabriele Mencagli and Yuriy Rymarchuk
+ *  
+ *  This file is part of WindFlow.
+ *  
+ *  WindFlow is free software dual licensed under the GNU LGPL or MIT License.
+ *  You can redistribute it and/or modify it under the terms of the
+ *    * GNU Lesser General Public License as published by
+ *      the Free Software Foundation, either version 3 of the License, or
+ *      (at your option) any later version
+ *    OR
+ *    * MIT License: https://github.com/ParaGroup/WindFlow/blob/master/LICENSE.MIT
+ *  
+ *  WindFlow is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU Lesser General Public License for more details.
+ *  You should have received a copy of the GNU Lesser General Public License and
+ *  the MIT License along with WindFlow. If not, see <http://www.gnu.org/licenses/>
+ *  and <http://opensource.org/licenses/MIT/>.
+ **************************************************************************************
+ */
+
+/*  
+ *  Data types and operator functors used by the join tests.
+ */ 
+
+// includes
+#include<cmath>
+#include<string>
+
+using namespace std;
+using namespace wf;
+
+// Global variable for the result
+atomic<long> global_sum;
+
+// Struct of the input tuple
+struct tuple_t
+{
+    size_t key;
+    int64_t value;
+};
+
+#if 1
+template<>
+struct std::hash<tuple_t>
+{
+    size_t operator()(const tuple_t &t) const
+    {
+        size_t h1 = std::hash<int64_t>()(t.value);
+        size_t h2 = std::hash<size_t>()(t.key);
+        return h1 ^ h2;
+    }
+};
+#endif
+
+struct res_t
+{
+    size_t key;
+    int64_t value;
+    size_t from;
+};
+
+// Source functor for generating positive numbers
+class Source_Positive_Functor
+{
+private:
+    size_t len; // stream length per key
+    size_t keys; // number of keys
+    uint64_t next_ts; // next timestamp
+    bool generateWS; // true if watermarks must be generated
+
+public:
+    // Constructor
+    Source_Positive_Functor(size_t _len,
+                            size_t _keys,
+                            bool _generateWS):
+                            len(_len),
+                            keys(_keys),
+                            next_ts(0),
+                            generateWS(_generateWS) {}
+
+    // operator()
+    void operator()(Source_Shipper<tuple_t> &shipper)
+    {
+        static thread_local std::mt19937 generator;
+        generator.seed(1234);
+        std::uniform_int_distribution<int> distribution(0, 250);
+        for (size_t i=1; i<=len; i++) { // generation loop
+            for (size_t k=1; k<=keys; k++) {
+                tuple_t t;
+                t.key = k;
+                t.value = i;
+                shipper.pushWithTimestamp(std::move(t), next_ts);
+                if (generateWS) {
+                    shipper.setNextWatermark(next_ts);
+                }
+                auto offset = (distribution(generator)+1);
+                next_ts += offset*1000; // in ms
+            }
+        }
+    }
+};
+
+// Source functor for generating negative numbers
+class Source_Negative_Functor
+{
+private:
+    size_t len; // stream length per key
+    size_t keys; // number of keys
+    vector<int> values; // list of values
+    uint64_t next_ts; // next timestamp
+    bool generateWS; // true if watermarks must be generated
+
+public:
+    // Constructor
+    Source_Negative_Functor(size_t _len,
+                            size_t _keys,
+                            bool _generateWS):
+                            len(_len),
+                            keys(_keys),
+                            values(_keys, 0),
+                            next_ts(0),
+                            generateWS(_generateWS) {}
+
+    // operator()
+    void operator()(Source_Shipper<tuple_t> &shipper)
+    {
+        static thread_local std::mt19937 generator;
+        generator.seed(4321);
+        std::uniform_int_distribution<int> distribution(0, 250);
+        for (size_t i=1; i<=len; i++) { // generation loop
+            for (size_t k=1; k<=keys; k++) {
+                values[k-1]--;
+                tuple_t t;
+                t.key = k;
+                t.value = values[k-1];
+                shipper.pushWithTimestamp(std::move(t), next_ts);
+                if (generateWS) {
+                    shipper.setNextWatermark(next_ts);
+                }
+                auto offset = (distribution(generator)+1);
+                next_ts += offset*1000; // in ms
+            }
+        }
+    }
+};
+
+// Map functor
+class Map_Functor
+{
+public:
+    // operator()
+    void operator()(tuple_t &t)
+    {
+        t.value = t.value + 2;
+    }
+};
+
+// Join functor
+class Join_Functor
+{
+public:
+    // operator()
+    optional<tuple_t> operator()(const tuple_t &a, const tuple_t &b, RuntimeContext &rc)
+    {
+        tuple_t out;
+        out.value = a.value * b.value;
+        out.key = a.key;
+        return out;
+    }
+};
+
+// Distinct Join functor
+class Distinct_Join_Functor
+{
+public:
+    // operator()
+    optional<tuple_t> operator()(const tuple_t &a, const tuple_t &b)
+    {
+        if (a.value != b.value) {
+            tuple_t out;
+            out.value = a.value * b.value;
+            out.key = a.key;
+            return out;
+        }
+        return {};
+    }
+};
+
+// Filter functor with keyby distribution
+class Filter_Functor_KB
+{
+private:
+    int mod;
+
+public:
+    // constructor
+    Filter_Functor_KB(int _mod): mod(_mod) {}
+
+    // operator()
+    bool operator()(tuple_t &t, RuntimeContext &rc)
+    {
+        assert(t.key % rc.getParallelism() == rc.getReplicaIndex());
+        if (t.value % mod == 0) {
+            return true;
+        }
+        else {
+            return false;
+        }
+    }
+};
+
+// Filter functor
+class Filter_Functor
+{
+private:
+    int mod;
+
+public:
+    // constructor
+    Filter_Functor(int _mod): mod(_mod) {}
+
+    // operator()
+    bool operator()(tuple_t &t)
+    {
+        if (t.value % mod == 0) {
+            return true;
+        }
+        else {
+            return false;
+        }
+    }
+};
+
+// FlatMap functor 
+class FlatMap_Functor
+{
+public:
+    // operator()
+    void operator()(const tuple_t &t, Shipper<tuple_t> &shipper)
+    {
+        for (size_t i=0; i<2; i++) {
+            shipper.push(t);
+        }
+    }
+};
+
+// Sink functor
+class Sink_Functor
+{
+private:
+    size_t received; // counter of received results
+    long totalsum;
+
+public:
+    // Constructor
+    Sink_Functor():
+                 received(0),
+                 totalsum(0) {}
+
+    // operator()
+    void operator()(optional<tuple_t> &out, RuntimeContext &rc)
+    {
+        if (out) {
+            received++;
+            totalsum += (*out).value;
+            size_t key = (*out).key;
+            int64_t value = (*out).value;
+        }
+        else {
+            global_sum.fetch_add(totalsum);
+        }
+    }
+};