Path finding operator

.gitignore

@@ -6,4 +6,5 @@ duckdb_unittest_tempdir/
 testext
 test/python/__pycache__/
 .Rhistory
+.vscode
 .venv

src/core/CMakeLists.txt

@@ -1,8 +1,11 @@
 add_subdirectory(functions)
 add_subdirectory(operator)
+add_subdirectory(option)
+add_subdirectory(optimizer)
 add_subdirectory(parser)
 add_subdirectory(utils)
 
+
 set(EXTENSION_SOURCES
         ${CMAKE_CURRENT_SOURCE_DIR}/module.cpp
         ${EXTENSION_SOURCES}

src/core/functions/function_data/CMakeLists.txt

@@ -1,10 +1,13 @@
 set(EXTENSION_SOURCES
         ${EXTENSION_SOURCES}
         ${CMAKE_CURRENT_SOURCE_DIR}/cheapest_path_length_function_data.cpp
+        ${CMAKE_CURRENT_SOURCE_DIR}/csr_function_data.cpp
+        ${CMAKE_CURRENT_SOURCE_DIR}/csr_operator_function_data.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/iterative_length_function_data.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/local_clustering_coefficient_function_data.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/pagerank_function_data.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/weakly_connected_component_function_data.cpp
+        ${CMAKE_CURRENT_SOURCE_DIR}/shortest_path_operator_function_data.cpp
 
         PARENT_SCOPE
 )

src/core/functions/function_data/cheapest_path_length_function_data.cpp

@@ -1,6 +1,5 @@
 #include "duckpgq/core/functions/function_data/cheapest_path_length_function_data.hpp"
 #include "duckpgq/core/utils/duckpgq_utils.hpp"
-#include "duckdb/execution/expression_executor.hpp"
 
 namespace duckpgq {
 
@@ -10,14 +9,8 @@ unique_ptr<FunctionData> CheapestPathLengthFunctionData::CheapestPathLengthBind(
     ClientContext &context, ScalarFunction &bound_function,
     vector<unique_ptr<Expression>> &arguments) {
 
-  if (!arguments[0]->IsFoldable()) {
-    throw InvalidInputException("Id must be constant.");
-  }
-
+  auto csr_id = GetCSRId(arguments[0], context);
   auto duckpgq_state = GetDuckPGQState(context);
-
-  int32_t csr_id = ExpressionExecutor::EvaluateScalar(context, *arguments[0])
-                       .GetValue<int32_t>();
   CSR *csr = duckpgq_state->GetCSR(csr_id);
 
   if (!(csr->initialized_v && csr->initialized_e && csr->initialized_w)) {

src/core/functions/function_data/csr_function_data.cpp

@@ -0,0 +1,56 @@
+#include "duckpgq/core/functions/function_data/csr_function_data.hpp"
+#include <duckpgq/core/utils/compressed_sparse_row.hpp>
+
+namespace duckpgq {
+
+namespace core {
+
+CSRFunctionData::CSRFunctionData(ClientContext &context, int32_t id,
+                                 LogicalType weight_type)
+    : context(context), id(id), weight_type(std::move(weight_type)) {}
+
+unique_ptr<FunctionData> CSRFunctionData::Copy() const {
+  return make_uniq<CSRFunctionData>(context, id, weight_type);
+}
+
+bool CSRFunctionData::Equals(const FunctionData &other_p) const {
+  auto &other = (const CSRFunctionData &)other_p;
+  return id == other.id && weight_type == other.weight_type;
+}
+
+
+unique_ptr<FunctionData>
+CSRFunctionData::CSRVertexBind(ClientContext &context,
+                               ScalarFunction &bound_function,
+                               vector<unique_ptr<Expression>> &arguments) {
+  auto csr_id = GetCSRId(arguments[0], context);
+  if (arguments.size() == 4) {
+    auto logical_type = LogicalType::SQLNULL;
+    return make_uniq<CSRFunctionData>(context, csr_id, logical_type);
+  }
+  return make_uniq<CSRFunctionData>(context, csr_id, arguments[3]->return_type);
+}
+
+unique_ptr<FunctionData>
+CSRFunctionData::CSREdgeBind(ClientContext &context,
+                             ScalarFunction &bound_function,
+                             vector<unique_ptr<Expression>> &arguments) {
+  auto csr_id = GetCSRId(arguments[0], context);
+  if (arguments.size() == 8) {
+    return make_uniq<CSRFunctionData>(context, csr_id,
+                                      arguments[7]->return_type);
+  }
+  auto logical_type = LogicalType::SQLNULL;
+  return make_uniq<CSRFunctionData>(context, csr_id, logical_type);
+}
+
+unique_ptr<FunctionData>
+CSRFunctionData::CSRBind(ClientContext &context, ScalarFunction &bound_function,
+                         vector<unique_ptr<Expression>> &arguments) {
+  auto csr_id = GetCSRId(arguments[0], context);
+  return make_uniq<CSRFunctionData>(context, csr_id, LogicalType::BOOLEAN);
+}
+
+} // namespace core
+
+} // namespace duckpgq

src/core/functions/function_data/csr_operator_function_data.cpp

@@ -0,0 +1,28 @@
+#include "duckpgq/core/functions/function_data/csr_operator_function_data.hpp"
+
+namespace duckpgq {
+
+namespace core {
+
+CSROperatorFunctionData::CSROperatorFunctionData(ClientContext &context)
+    : context(context) {}
+
+unique_ptr<FunctionData> CSROperatorFunctionData::Copy() const {
+    return make_uniq<CSROperatorFunctionData>(context);
+}
+
+unique_ptr<FunctionData>
+CSROperatorFunctionData::CSRBind(ClientContext &context,
+                               ScalarFunction &bound_function,
+                               vector<unique_ptr<Expression>> &arguments) {
+    return make_uniq<CSROperatorFunctionData>(context);
+}
+
+bool CSROperatorFunctionData::Equals(const FunctionData &other_p) const {
+    // TODO implement me
+    // auto &other = (const CSROperatorFunctionData &)other_p;
+    return true;
+}
+} // namespace core
+
+} // namespace duckpgq

src/core/functions/function_data/iterative_length_function_data.cpp

@@ -1,13 +1,14 @@
 #include "duckpgq/core/functions/function_data/iterative_length_function_data.hpp"
-#include "duckdb/execution/expression_executor.hpp"
 #include "duckpgq/common.hpp"
 
+#include <duckpgq/core/utils/compressed_sparse_row.hpp>
+
 namespace duckpgq {
 
 namespace core {
 
 unique_ptr<FunctionData> IterativeLengthFunctionData::Copy() const {
-  return make_uniq<IterativeLengthFunctionData>(context, csr_id);
+  return make_uniq<IterativeLengthFunctionData>(context, table_to_scan, csr_id);
 }
 
 bool IterativeLengthFunctionData::Equals(const FunctionData &other_p) const {
@@ -19,14 +20,13 @@ bool IterativeLengthFunctionData::Equals(const FunctionData &other_p) const {
 unique_ptr<FunctionData> IterativeLengthFunctionData::IterativeLengthBind(
     ClientContext &context, ScalarFunction &bound_function,
     vector<unique_ptr<Expression>> &arguments) {
-  if (!arguments[0]->IsFoldable()) {
-    throw InvalidInputException("Id must be constant.");
+  if (arguments.size() == 3) {
+    string table_to_scan = ExpressionExecutor::EvaluateScalar(context, *arguments[2]).GetValue<string>();
+    return make_uniq<IterativeLengthFunctionData>(context, table_to_scan, 0);
   }
+ auto csr_id = GetCSRId(arguments[0], context);
 
-  int32_t csr_id = ExpressionExecutor::EvaluateScalar(context, *arguments[0])
-                       .GetValue<int32_t>();
-
-  return make_uniq<IterativeLengthFunctionData>(context, csr_id);
+  return make_uniq<IterativeLengthFunctionData>(context, "", csr_id);
 }
 
 

src/core/functions/function_data/local_clustering_coefficient_function_data.cpp

@@ -1,5 +1,5 @@
 #include "duckpgq/core/functions/function_data/local_clustering_coefficient_function_data.hpp"
-#include "duckdb/execution/expression_executor.hpp"
+#include <duckpgq/core/utils/compressed_sparse_row.hpp>
 
 namespace duckpgq {
 
@@ -15,12 +15,7 @@ unique_ptr<FunctionData>
 LocalClusteringCoefficientFunctionData::LocalClusteringCoefficientBind(
     ClientContext &context, ScalarFunction &bound_function,
     vector<unique_ptr<Expression>> &arguments) {
-  if (!arguments[0]->IsFoldable()) {
-    throw InvalidInputException("Id must be constant.");
-  }
-
-  int32_t csr_id = ExpressionExecutor::EvaluateScalar(context, *arguments[0])
-                       .GetValue<int32_t>();
+  auto csr_id = GetCSRId(arguments[0], context);
 
   return make_uniq<LocalClusteringCoefficientFunctionData>(context, csr_id);
 }

src/core/functions/function_data/shortest_path_operator_function_data.cpp

@@ -0,0 +1,26 @@
+#include "duckpgq/core/functions/function_data/shortest_path_operator_function_data.hpp"
+#include <duckpgq/core/utils/compressed_sparse_row.hpp>
+
+namespace duckpgq {
+
+namespace core {
+
+unique_ptr<FunctionData>
+ShortestPathOperatorData::ShortestPathOperatorBind(
+    ClientContext &context, ScalarFunction &bound_function,
+    vector<unique_ptr<Expression>> &arguments) {
+  string table_to_scan = ExpressionExecutor::EvaluateScalar(context, *arguments[2]).GetValue<string>();
+  return make_uniq<ShortestPathOperatorData>(context, table_to_scan);
+}
+unique_ptr<FunctionData> ShortestPathOperatorData::Copy() const {
+  return make_uniq<ShortestPathOperatorData>(context, table_to_scan);
+}
+
+bool ShortestPathOperatorData::Equals(const FunctionData &other_p) const {
+  auto &other = (const ShortestPathOperatorData &)other_p;
+  return other.table_to_scan == table_to_scan;
+}
+
+} // namespace core
+
+} // namespace duckpgq

src/core/functions/scalar/CMakeLists.txt

@@ -4,12 +4,15 @@ set(EXTENSION_SOURCES
         ${CMAKE_CURRENT_SOURCE_DIR}/csr_creation.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/csr_deletion.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/csr_get_w_type.cpp
+        ${CMAKE_CURRENT_SOURCE_DIR}/csr_operator_creation.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/iterativelength.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/iterativelength2.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/iterativelength_bidirectional.cpp
+        ${CMAKE_CURRENT_SOURCE_DIR}/iterativelength_operator.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/pagerank.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/reachability.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/shortest_path.cpp
+        ${CMAKE_CURRENT_SOURCE_DIR}/shortest_path_operator.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/csr_creation.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/local_clustering_coefficient.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/weakly_connected_component.cpp

src/core/functions/scalar/csr_creation.cpp

@@ -4,10 +4,9 @@
 #include "duckpgq/common.hpp"
 #include "duckpgq/core/utils/compressed_sparse_row.hpp"
 #include <cmath>
+#include <duckpgq/core/functions/function_data/csr_function_data.hpp>
 #include <duckpgq/core/functions/scalar.hpp>
 #include <duckpgq/core/utils/duckpgq_utils.hpp>
-#include <duckpgq_extension.hpp>
-#include <mutex>
 
 namespace duckpgq {
 
@@ -49,6 +48,9 @@ static void CsrInitializeEdge(DuckPGQState &context, int32_t id, int64_t v_size,
   const lock_guard<mutex> csr_init_lock(context.csr_lock);
 
   auto csr_entry = context.csr_list.find(id);
+  if (csr_entry == context.csr_list.end()) {
+    throw InvalidInputException("CSR has not been initialized properly");
+  }
   if (csr_entry->second->initialized_e) {
     return;
   }
@@ -70,7 +72,9 @@ static void CsrInitializeWeight(DuckPGQState &context, int32_t id,
                                 int64_t e_size, PhysicalType weight_type) {
   const lock_guard<mutex> csr_init_lock(context.csr_lock);
   auto csr_entry = context.csr_list.find(id);
-
+  if (csr_entry == context.csr_list.end()) {
+    throw InvalidInputException("CSR has not been initialized properly");
+  }
   if (csr_entry->second->initialized_w) {
     return;
   }
@@ -143,6 +147,9 @@ static void CreateCsrEdgeFunction(DataChunk &args, ExpressionState &state,
   }
 
   auto csr_entry = duckpgq_state->csr_list.find(info.id);
+  if (csr_entry == duckpgq_state->csr_list.end()) {
+    throw InvalidInputException("CSR has not been initialized properly");
+  }
   if (!csr_entry->second->initialized_e) {
     CsrInitializeEdge(*duckpgq_state, info.id, vertex_size, edge_size);
   }
@@ -211,6 +218,13 @@ ScalarFunctionSet GetCSREdgeFunction() {
    * 7. <optional> edge weight (INT OR DOUBLE)
    */
 
+
+  set.AddFunction(ScalarFunction({LogicalType::BIGINT, LogicalType::BIGINT,
+                                  LogicalType::BIGINT, LogicalType::BIGINT,
+                                  LogicalType::BIGINT, LogicalType::BIGINT},
+                                 LogicalType::INTEGER, CreateCsrEdgeFunction,
+                                 CSRFunctionData::CSREdgeBind));
+
   //! No edge weight
   set.AddFunction(ScalarFunction({LogicalType::INTEGER, LogicalType::BIGINT,
                                   LogicalType::BIGINT, LogicalType::BIGINT, LogicalType::BIGINT,

src/core/functions/scalar/csr_deletion.cpp

@@ -1,9 +1,9 @@
 #include "duckdb/main/client_data.hpp"
 #include "duckdb/planner/expression/bound_function_expression.hpp"
 #include "duckpgq/common.hpp"
+#include <duckpgq/core/functions/function_data/csr_function_data.hpp>
 #include <duckpgq/core/functions/scalar.hpp>
 #include <duckpgq/core/utils/duckpgq_utils.hpp>
-#include <duckpgq_extension.hpp>
 
 namespace duckpgq {
 

src/core/functions/scalar/csr_get_w_type.cpp

@@ -2,6 +2,7 @@
 #include "duckpgq/common.hpp"
 #include <duckpgq_extension.hpp>
 
+#include <duckpgq/core/functions/function_data/csr_function_data.hpp>
 #include <duckpgq/core/functions/scalar.hpp>
 #include <duckpgq/core/utils/duckpgq_utils.hpp>
 

src/core/functions/scalar/csr_operator_creation.cpp

@@ -0,0 +1,33 @@
+#include "duckdb/function/scalar_function.hpp"
+#include "duckpgq/common.hpp"
+#include <duckpgq/core/functions/function_data/csr_operator_function_data.hpp>
+#include <duckpgq/core/functions/scalar.hpp>
+
+namespace duckpgq {
+
+namespace core {
+
+static void CreateCSROperatorFunction(DataChunk &args, ExpressionState &state,
+                                      Vector &result) {
+  throw NotImplementedException(
+      "CSR operator creation function not implemented, should have gone to "
+      "operator instead.");
+}
+
+//------------------------------------------------------------------------------
+// Register functions
+//------------------------------------------------------------------------------
+void CoreScalarFunctions::RegisterCSROperatorCreationScalarFunctions(
+    DatabaseInstance &db) {
+  ExtensionUtil::RegisterFunction(
+      db, ScalarFunction("csr_operator",
+                         {LogicalType::BIGINT, LogicalType::BIGINT,
+                          LogicalType::BIGINT, LogicalType::BIGINT,
+                          LogicalType::BIGINT, LogicalType::BIGINT},
+                         LogicalType::INTEGER, CreateCSROperatorFunction,
+                         CSROperatorFunctionData::CSRBind));
+}
+
+} // namespace core
+
+} // namespace duckpgq

src/core/functions/scalar/iterativelength.cpp

@@ -154,21 +154,31 @@ static void IterativeLengthFunction(DataChunk &args, ExpressionState &state,
   duckpgq_state->csr_to_delete.insert(info.csr_id);
 }
 
+ScalarFunctionSet GetIterativeLengthFunction() {
+  ScalarFunctionSet set("iterativelength");
+
+  set.AddFunction(ScalarFunction(
+                            {LogicalType::INTEGER, LogicalType::BIGINT,
+                             LogicalType::BIGINT, LogicalType::BIGINT},
+                            LogicalType::BIGINT, IterativeLengthFunction,
+                            IterativeLengthFunctionData::IterativeLengthBind));
+
+  set.AddFunction(
+  ScalarFunction(
+    {LogicalType::BIGINT, LogicalType::BIGINT, LogicalType::VARCHAR},
+           LogicalType::BIGINT, IterativeLengthFunction,
+           IterativeLengthFunctionData::IterativeLengthBind));
+  return set;
+}
+
 //------------------------------------------------------------------------------
 // Register functions
 //------------------------------------------------------------------------------
 void CoreScalarFunctions::RegisterIterativeLengthScalarFunction(
     DatabaseInstance &db) {
-  ExtensionUtil::RegisterFunction(
-      db,
-      ScalarFunction("iterativelength",
-                            {LogicalType::INTEGER, LogicalType::BIGINT,
-                             LogicalType::BIGINT, LogicalType::BIGINT},
-                            LogicalType::BIGINT, IterativeLengthFunction,
-                            IterativeLengthFunctionData::IterativeLengthBind));
+  ExtensionUtil::RegisterFunction(db, GetIterativeLengthFunction());
 }
 
 } // namespace core
 
 } // namespace duckpgq
-