QDB-16714 - Dask integration into Python API

quasardb/cluster.cpp

@@ -191,7 +191,10 @@ void register_cluster(py::module_ & m)
         .def("compact_progress", &qdb::cluster::compact_progress)                          //
         .def("compact_abort", &qdb::cluster::compact_abort)                                //
         .def("wait_for_compaction", &qdb::cluster::wait_for_compaction)                    //
-        .def("endpoints", &qdb::cluster::endpoints);                                       //
+        .def("endpoints", &qdb::cluster::endpoints)                                        //
+        .def("validate_query", &qdb::cluster::validate_query)                              //
+        .def("split_query_range", &qdb::cluster::split_query_range);                       //
+
 }
 
 }; // namespace qdb

quasardb/cluster.hpp

@@ -61,6 +61,7 @@
 #include <chrono>
 #include <iostream>
 
+
 namespace qdb
 {
 
@@ -496,6 +497,36 @@ class cluster
         return results;
     }
 
+    py::object validate_query(const std::string & query_string)
+    {
+        check_open();
+
+        std::string query = query_string;
+        const std::string limit_string = "LIMIT 1";
+        query += " " + limit_string;
+
+        // TODO:
+        // should return dict of column names and dtypes
+        // currently returns numpy masked arrays
+        return py::cast(qdb::numpy_query(_handle, query));
+    }
+
+    py::object split_query_range(std::chrono::system_clock::time_point start, std::chrono::system_clock::time_point end, std::chrono::milliseconds delta)
+    {
+        std::vector<std::pair<std::chrono::system_clock::time_point, std::chrono::system_clock::time_point>> ranges;
+
+        for (auto current_start = start; current_start < end; ) {
+            auto current_end = current_start + delta;
+            if (current_end > end) {
+                current_end = end;
+            }
+            ranges.emplace_back(current_start, current_end);
+            current_start = current_end;
+        }
+        return py::cast(ranges);
+    }
+
+
 private:
     std::string _uri;
     handle_ptr _handle;

tests/conftest.py

@@ -764,10 +764,15 @@ def deduplication_mode(request):
     return request.param
 
 
+@pytest.fixture(params=["S"], ids=["frequency=S"])
+def frequency(request):
+    yield request.param
+
+
 @pytest.fixture
-def gen_index(start_date, row_count):
+def gen_index(start_date, row_count, frequency):
     return pd.Index(
-        pd.date_range(start_date, periods=row_count, freq="S"), name="$timestamp"
+        pd.date_range(start_date, periods=row_count, freq=frequency), name="$timestamp"
     )