Conjunctive filtering

CMakeLists.txt

@@ -1,5 +1,5 @@
 cmake_minimum_required(VERSION 3.3)
-project(learning_arrow_cpp)
+project(arrow_simpledb)
 
 ADD_SUBDIRECTORY(libdb)
 ADD_SUBDIRECTORY(testdb)

libdb/CMakeLists.txt

@@ -15,7 +15,7 @@ set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
 
 set(SOURCE_FILES ChunkedColumnCursor.cpp ScanTableCursor.cpp
         ColumnCursorWrapper.cpp GenericColumnCursor.cpp TableCursor.cpp
-        FilterProjectTableCursor.cpp GreaterThanFilter.cpp GreaterThanFilter.h)
+        FilterProjectTableCursor.cpp filters/GreaterThanFilter.cpp filters/GreaterThanFilter.h filters/LessThanFilter.cpp filters/LessThanFilter.h filters/AndFilter.cpp filters/AndFilter.h)
 add_library(libdb STATIC ${SOURCE_FILES})
 
 target_link_libraries(libdb ${ARROW_LIBS} ${CMAKE_THREAD_LIBS_INIT})

libdb/FilterProjectTableCursor.h

@@ -6,7 +6,7 @@
 #include <arrow/table.h>
 #include "ColumnCursorWrapper.h"
 #include "TableCursor.h"
-#include "Filter.h"
+#include "filters/Filter.h"
 
 class FilterProjectTableCursor : public TableCursor {
 public:

libdb/filters/AndFilter.cpp

@@ -0,0 +1,27 @@
+
+#include "arrow/api.h"
+
+#include "AndFilter.h"
+
+AndFilter::AndFilter(std::string column_name, std::shared_ptr<Filter> left, std::shared_ptr<Filter> right)
+        : _column_name(column_name), _left(left), _right(right)
+{
+
+}
+
+void
+AndFilter::initialize(TableCursor& table_cursor) {
+    _left->initialize(table_cursor);
+    _right->initialize(table_cursor);
+}
+
+bool
+AndFilter::evaluate()
+{
+    // short circuit explicitly
+    if (_left->evaluate()) {
+        return _right->evaluate();
+    } else {
+        return false;
+    }
+}

libdb/filters/AndFilter.h

@@ -0,0 +1,30 @@
+
+
+#ifndef ANDFILTER_H
+#define ANDFILTER_H
+
+#include <string>
+#include "filters/Filter.h"
+#include "TableCursor.h"
+
+class AndFilter : public Filter {
+public:
+
+    AndFilter(std::string column_name, std::shared_ptr<Filter> left, std::shared_ptr<Filter> right);
+
+    void initialize(TableCursor& table_cursor) override;
+
+    bool evaluate() override;
+
+private:
+
+    std::string _column_name;
+
+    std::shared_ptr<Filter> _left;
+
+    std::shared_ptr<Filter> _right;
+
+};
+
+
+#endif // ANDFILTER_H

libdb/GreaterThanFilter.cpp → libdb/filters/GreaterThanFilter.cpp

@@ -2,7 +2,6 @@
 #include "arrow/api.h"
 
 #include "GreaterThanFilter.h"
-#include "TableCursor.h"
 
 GreaterThanFilter::GreaterThanFilter(std::string column_name, double value)
         : _column_name(column_name), _value(value)

libdb/GreaterThanFilter.h → libdb/filters/GreaterThanFilter.h

@@ -4,7 +4,7 @@
 #define GREATERTHANFILTER_H
 
 #include <string>
-#include "Filter.h"
+#include "filters/Filter.h"
 #include "TableCursor.h"
 #include "ColumnCursorWrapper.h"
 

libdb/filters/LessThanFilter.cpp

@@ -0,0 +1,23 @@
+
+
+#include "LessThanFilter.h"
+
+#include "arrow/api.h"
+
+LessThanFilter::LessThanFilter(std::string column_name, double value)
+        : _column_name(column_name), _value(value)
+{
+
+}
+
+void
+LessThanFilter::initialize(TableCursor& table_cursor) {
+    _cursor = std::dynamic_pointer_cast<ColumnCursorWrapper<arrow::DoubleArray>>(
+            table_cursor.getColumn(_column_name));
+}
+
+bool
+LessThanFilter::evaluate()
+{
+    return _cursor->get() < _value;
+}

libdb/filters/LessThanFilter.h

@@ -0,0 +1,30 @@
+
+
+#ifndef LESSTHANFILTER_H
+#define LESSTHANFILTER_H
+
+#include <string>
+#include "filters/Filter.h"
+#include "TableCursor.h"
+#include "ColumnCursorWrapper.h"
+
+class LessThanFilter  : public Filter {
+public:
+
+    LessThanFilter(std::string column_name, double value);
+
+    void initialize(TableCursor &table_cursor) override;
+
+    bool evaluate() override;
+
+private:
+
+    std::string _column_name;
+
+    double _value;
+
+    std::shared_ptr <ColumnCursorWrapper<arrow::DoubleArray>> _cursor;
+};
+
+
+#endif // LESSTHANFILTER_H

testdb/TableTest.cpp

@@ -1,5 +1,7 @@
 
-#include <GreaterThanFilter.h>
+#include <filters/GreaterThanFilter.h>
+#include <filters/AndFilter.h>
+#include <filters/LessThanFilter.h>
 #include "arrow/api.h"
 
 #include "TableTest.h"
@@ -92,7 +94,7 @@ TEST_F(TableTest, ComplexColumns) {
     EXPECT_EQ(200, count);
 }
 
-TEST_F(TableTest, SingleFilter) {
+TEST_F(TableTest, SimpleFilter) {
     std::shared_ptr<Table> table;
     EXPECT_EQ(Status::OK().code(), Tables::createSmallChunkedColumns(table).code());
     EXPECT_EQ(4, table->num_rows());
@@ -112,4 +114,32 @@ TEST_F(TableTest, SingleFilter) {
     EXPECT_EQ(32, id_cursor->get());
     EXPECT_EQ(42.9, cost_cursor->get());
     EXPECT_FALSE(fptc.hasMore());
+}
+
+TEST_F(TableTest, ConjunctiveFilter) {
+    std::shared_ptr<Table> table;
+    EXPECT_EQ(Status::OK().code(), Tables::createSmallChunkedColumns(table).code());
+    EXPECT_EQ(4, table->num_rows());
+    EXPECT_EQ(2, table->num_columns());
+    ScanTableCursor tc(table);
+
+    std::shared_ptr<Filter> leftFilter = std::make_shared<GreaterThanFilter>("id", 11);
+    std::shared_ptr<Filter> rightFilter = std::make_shared<LessThanFilter>("cost", 42);
+    std::shared_ptr<Filter> andFilter =
+            std::make_shared<AndFilter>("id", leftFilter, rightFilter);
+
+    FilterProjectTableCursor fptc(tc, andFilter);
+
+    uint64_t count = 0;
+    auto id_cursor = std::dynamic_pointer_cast<ColumnCursorWrapper<arrow::Int64Array>>(
+            fptc.getColumn(std::string("id")));
+    auto cost_cursor = std::dynamic_pointer_cast<ColumnCursorWrapper<arrow::DoubleArray>>(
+            fptc.getColumn(std::string("cost")));
+    EXPECT_TRUE(fptc.hasMore());
+    EXPECT_EQ(12, id_cursor->get());
+    EXPECT_EQ(22.9, cost_cursor->get());
+    EXPECT_TRUE(fptc.hasMore());
+    EXPECT_EQ(31, id_cursor->get());
+    EXPECT_EQ(41.9, cost_cursor->get());
+    EXPECT_FALSE(fptc.hasMore());
 }