Implemented hash join

Author: Griezn

CMakeLists.txt

@@ -40,6 +40,8 @@ add_library(SR STATIC
         include/buffer.h
         src/queue.c
         include/queue.h
+        include/hash_table.h
+        src/hash_table.c
 )
 
 enable_testing()

include/hash_table.h

@@ -0,0 +1,35 @@
+//
+// Created by Seppe Degryse on 15/04/2025.
+//
+
+#ifndef HASH_TABLE_H
+#define HASH_TABLE_H
+
+#include <stdio.h>
+#include <stdlib.h>
+
+#include <data.h>
+
+typedef struct Bucket {
+    triple_t **tuples;
+    size_t count;
+    size_t capacity;
+} bucket_t;
+
+typedef struct HashTable {
+    bucket_t *table;
+    size_t size;
+} hash_table_t;
+
+
+uint32_t hash(uint32_t key, size_t table_size);
+
+hash_table_t create_table(size_t num_elements);
+
+void free_table(const hash_table_t *ht);
+
+bool insert(hash_table_t *ht, uint32_t offset, triple_t *el);
+
+bucket_t *contains(const hash_table_t *ht, uint32_t offset_in1, uint32_t offset_in2, triple_t *el);
+
+#endif //HASH_TABLE_H

include/memory.h

@@ -11,10 +11,14 @@ void* tracked_malloc(size_t size);
 
 void* tracked_realloc(void* ptr, size_t new_size);
 
+void tracked_free(void* ptr, size_t size);
+
 size_t get_alloc_count();
 
 size_t get_total_allocated();
 
+size_t get_peak_allocated();
+
 void reset_memory_counter();
 
 #endif //MEMORY_H

include/operator.h

@@ -11,6 +11,7 @@
 
 enum OPERATORS {
     JOIN,
+    HASH_JOIN,
     CARTESIAN,
     FILTER,
     WINDOW,
@@ -22,14 +23,21 @@ typedef bool (*join_check_t)(triple_t in1, triple_t in2);
 typedef bool (*filter_check_t)(triple_t in);
 
 typedef struct JoinParams {
-    uint8_t      size;
+    uint8_t size;
     join_check_t *checks;
 } join_params_t;
 
 typedef struct CartJoinParams {
-    double  probability;
+    double probability;
 } cart_join_params_t;
 
+typedef struct HashJoinParams {
+    uint32_t    predicate_in1;
+    uint8_t     offset_in1;
+    uint32_t    predicate_in2;
+    uint8_t     offset_in2;
+} hash_join_params_t;
+
 typedef struct FilterParams {
     uint8_t size;
     filter_check_t *checks;
@@ -48,18 +56,19 @@ typedef struct WindowParams {
 } window_params_t;
 
 typedef union Parameters {
-    struct JoinParams       join;
-    struct CartJoinParams   cart_join;
-    struct FilterParams     filter;
-    struct SelectParams     select;
-    struct WindowParams     window;
+    struct JoinParams join;
+    struct HashJoinParams hash_join;
+    struct CartJoinParams cart_join;
+    struct FilterParams filter;
+    struct SelectParams select;
+    struct WindowParams window;
 } parameter_t;
 
 typedef struct Operator {
-    enum    OPERATORS   type;
-    struct  Operator    *left;
-    struct  Operator    *right;
-    union   Parameters  params;
+    enum OPERATORS type;
+    struct Operator *left;
+    struct Operator *right;
+    union Parameters params;
 } operator_t;
 
 #endif //OPERATOR_H

include/query.h

@@ -12,6 +12,8 @@
 #include <pthread.h>
 #include <stdatomic.h>
 
+#include "hash_table.h"
+
 
 typedef struct ExecutionStep {
     const operator_t *operator_;
@@ -63,6 +65,9 @@ void flatten_query(const operator_t* operator_, spsc_queue_t *results, uint8_t i
 void join_triple_copy(const data_t *src1, uint32_t index1,
                     const data_t *src2, uint32_t index2, data_t *dest);
 
+void join_bucket_copy(const data_t *src1, bucket_t *bucket,
+                    const data_t *src2, uint32_t index2, data_t *dest);
+
 bool join_check(const data_t *src1, uint32_t index1,
                     const data_t *src2, uint32_t index2, join_params_t check);
 

include/queue.h

@@ -27,4 +27,6 @@ bool spsc_is_empty(spsc_queue_t *q);
 
 void empty_queue(spsc_queue_t *q);
 
+void empty_queue_ndata(spsc_queue_t *q);
+
 #endif //SPSC_QUEUE_LIBRARY_H

src/data.c

@@ -7,12 +7,12 @@
 
 #include <assert.h>
 #include <stdlib.h>
-#include <stdbool.h>
 
+#include "hash_table.h"
 #include "memory.h"
 
-#define malloc(size) tracked_malloc(size)
-#define realloc(ptr, size) tracked_realloc(ptr, size)
+#define malloc(size) malloc(size)
+#define realloc(ptr, size) realloc(ptr, size)
 
 void join_triple_copy(const data_t *src1, const uint32_t index1,
                         const data_t *src2, const uint32_t index2, data_t *dest)
@@ -32,6 +32,27 @@ void join_triple_copy(const data_t *src1, const uint32_t index1,
 }
 
 
+void join_bucket_copy(const data_t *src1, bucket_t *bucket,
+                    const data_t *src2, uint32_t index2, data_t *dest)
+{
+    uint32_t index = dest->size * dest->width;
+    for (size_t j = 0; j < bucket->count; ++j) {
+        for (uint32_t i = 0; i < src1->width; ++i) {
+            //PUSH_TO_BUFFER(dest, index++, el1[i]);
+            PUSH_TO_BUFFER(dest, index++, bucket->tuples[j][i]);
+            //dest->data[index++] = src1->data[index1 + i];
+        }
+
+        for (uint32_t i = 0; i < src2->width; ++i) {
+            PUSH_TO_BUFFER(dest, index++, src2->data[index2 + i]);
+            //dest->data[index++] = src2->data[index2 + i];
+        }
+
+        dest->size++;
+    }
+}
+
+
 bool join_check(const data_t *src1, const uint32_t index1,
                     const data_t *src2, const uint32_t index2, const join_params_t check)
 {

src/hash_table.c

@@ -0,0 +1,104 @@
+//
+// Created by Seppe Degryse on 15/04/2025.
+//
+#include "hash_table.h"
+
+#include <assert.h>
+
+
+uint32_t hash(const uint32_t key, const size_t table_size)
+{
+    return key % table_size;
+}
+
+
+hash_table_t create_table(const size_t num_elements)
+{
+    const size_t size = num_elements * 2;  // double for lower load factor
+    hash_table_t ht;
+    ht.size = size;
+
+    // Allocate array of buckets
+    ht.table = (bucket_t *) calloc(size, sizeof(bucket_t));
+    if (ht.table == NULL) {
+        fprintf(stderr, "Failed to allocate hash table\n");
+        exit(EXIT_FAILURE);
+    }
+
+    return ht;
+}
+
+
+void free_table(const hash_table_t *ht)
+{
+    for (size_t i = 0; i < ht->size; i++) {
+        free(ht->table[i].tuples);
+    }
+    free(ht->table);
+}
+
+
+bool bucket_insert(bucket_t *bucket, triple_t *el) {
+    if (bucket->count == bucket->capacity) {
+        size_t new_capacity = bucket->capacity == 0 ? 4 : bucket->capacity * 2;
+        triple_t **new_tuples = realloc(bucket->tuples, new_capacity * sizeof(triple_t *));
+        if (new_tuples == NULL) return false;
+
+        bucket->tuples = new_tuples;
+        bucket->capacity = new_capacity;
+    }
+
+    bucket->tuples[bucket->count++] = el;
+    return true;
+}
+
+
+bool triple_offset_equal(const uint32_t offset, triple_t *el, uint32_t key)
+{
+    return *((uint32_t *) el + offset) == key;
+}
+
+
+bool insert(hash_table_t *ht, const uint32_t offset, triple_t *el) {
+    uint32_t key = *((uint32_t *)el + offset);
+    size_t idx = hash(key, ht->size);
+
+    for (size_t i = 0; i < ht->size; i++) {
+        size_t probe_idx = (idx + i) % ht->size;
+        bucket_t *bucket = &ht->table[probe_idx];
+
+        // Empty bucket: initialize it
+        if (bucket->tuples == NULL) {
+            bucket->tuples = malloc(4 * sizeof(triple_t *));
+            if (bucket->tuples == NULL) return false;
+            bucket->capacity = 4;
+            bucket->count = 0;
+        }
+
+        // Check if the key in this bucket matches or it's the first insert
+        if (bucket->count == 0 || triple_offset_equal(offset, bucket->tuples[0], key)) {
+            return bucket_insert(bucket, el);
+        }
+    }
+
+    return false; // Table full
+}
+
+
+bucket_t *contains(const hash_table_t *ht, const uint32_t offset_in1, const uint32_t offset_in2, triple_t *el) {
+    uint32_t key = *((uint32_t *)el + offset_in2);
+    size_t idx = hash(key, ht->size);
+
+    for (size_t i = 0; i < ht->size; i++) {
+        size_t probe_idx = (idx + i) % ht->size;
+        bucket_t *bucket = &ht->table[probe_idx];
+
+        if (bucket->tuples == NULL)
+            return NULL;
+
+        if (bucket->count > 0 && triple_offset_equal(offset_in1, bucket->tuples[0], key))
+            return bucket;
+    }
+
+    return NULL;
+}

src/memory.c

@@ -6,33 +6,55 @@
 
 static size_t total_allocated = 0;
 static size_t allocation_count = 0;
+static size_t current = 0;
+static size_t peak_allocated = 0;
 
 
-void* tracked_malloc(size_t size) {
-    void* ptr = malloc(size);
+void *tracked_malloc(size_t size)
+{
+    void *ptr = malloc(size);
     if (ptr) {
         total_allocated += size;
+        current += size;
+        if (current > peak_allocated) peak_allocated = current;
         allocation_count++;
     }
     return ptr;
 }
 
-void* tracked_realloc(void* ptr, size_t new_size) {
+
+void *tracked_realloc(void *ptr, size_t new_size)
+{
     total_allocated -= (new_size / 2);
+    current -= (new_size/2);
 
-    void* new_ptr = realloc(ptr, new_size);
+    void *new_ptr = realloc(ptr, new_size);
     if (new_ptr) {
         total_allocated += new_size;
+        current += new_size;
+        if (current > peak_allocated) peak_allocated = current;
         allocation_count++;
     }
     return new_ptr;
 }
 
-size_t get_alloc_count() {return allocation_count;}
-size_t get_total_allocated() {return total_allocated;}
+
+void tracked_free(void *ptr, const size_t size)
+{
+    free(ptr);
+    current -= size;
+}
+
+
+size_t get_alloc_count() { return allocation_count; }
+size_t get_total_allocated() { return total_allocated; }
+size_t get_peak_allocated() { return peak_allocated; }
+
 
 void reset_memory_counter()
 {
     total_allocated = 0;
     allocation_count = 0;
-}
+    peak_allocated = 0;
+    current = 0;
+}