|
@@ -0,0 +1,418 @@
|
|
|
|
|
+#include "smartbotic/database/grpc/query_service.hpp"
|
|
|
|
|
+
|
|
|
|
|
+#include <spdlog/spdlog.h>
|
|
|
|
|
+
|
|
|
|
|
+#include <algorithm>
|
|
|
|
|
+
|
|
|
|
|
+#include "smartbotic/database/grpc/proto_convert.hpp"
|
|
|
|
|
+
|
|
|
|
|
+namespace smartbotic::database::grpc {
|
|
|
|
|
+
|
|
|
|
|
+QueryServiceImpl::QueryServiceImpl(std::shared_ptr<storage::Database> db) : db_(std::move(db)) {}
|
|
|
|
|
+
|
|
|
|
|
+::grpc::Status QueryServiceImpl::Query(::grpc::ServerContext* /*context*/,
|
|
|
|
|
+ const ::smartbotic::database::QueryRequest* request,
|
|
|
|
|
+ ::smartbotic::database::QueryResponse* response) {
|
|
|
|
|
+ spdlog::debug("Query: collection={}, limit={}, offset={}", request->collection(), request->limit(),
|
|
|
|
|
+ request->offset());
|
|
|
|
|
+
|
|
|
|
|
+ auto* coll = db_->GetCollection(request->collection());
|
|
|
|
|
+ if (coll == nullptr) {
|
|
|
|
|
+ return ::grpc::Status(::grpc::StatusCode::NOT_FOUND,
|
|
|
|
|
+ "Collection not found: " + request->collection());
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Get all documents and filter
|
|
|
|
|
+ auto all_docs = coll->GetAll();
|
|
|
|
|
+ std::vector<storage::Document> filtered_docs;
|
|
|
|
|
+
|
|
|
|
|
+ // Apply filter if present
|
|
|
|
|
+ if (request->has_filter()) {
|
|
|
|
|
+ for (const auto& doc : all_docs) {
|
|
|
|
|
+ if (MatchesFilter(doc, request->filter())) {
|
|
|
|
|
+ filtered_docs.push_back(doc);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ } else {
|
|
|
|
|
+ filtered_docs = std::move(all_docs);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Store total before pagination
|
|
|
|
|
+ response->set_total_count(static_cast<int64_t>(filtered_docs.size()));
|
|
|
|
|
+
|
|
|
|
|
+ // Apply sorting
|
|
|
|
|
+ if (!request->order_by().empty()) {
|
|
|
|
|
+ std::sort(filtered_docs.begin(), filtered_docs.end(),
|
|
|
|
|
+ [this, &request](const storage::Document& a, const storage::Document& b) {
|
|
|
|
|
+ for (const auto& order : request->order_by()) {
|
|
|
|
|
+ auto val_a = GetFieldValue(a.GetData(), order.field());
|
|
|
|
|
+ auto val_b = GetFieldValue(b.GetData(), order.field());
|
|
|
|
|
+
|
|
|
|
|
+ if (val_a == val_b) {
|
|
|
|
|
+ continue;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ bool less = val_a < val_b;
|
|
|
|
|
+ if (order.direction() ==
|
|
|
|
|
+ ::smartbotic::database::SORT_DIRECTION_DESCENDING) {
|
|
|
|
|
+ less = !less;
|
|
|
|
|
+ }
|
|
|
|
|
+ return less;
|
|
|
|
|
+ }
|
|
|
|
|
+ return false;
|
|
|
|
|
+ });
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Apply pagination
|
|
|
|
|
+ int32_t offset = request->offset();
|
|
|
|
|
+ int32_t limit = request->limit() > 0 ? request->limit() : 100;
|
|
|
|
|
+
|
|
|
|
|
+ if (static_cast<size_t>(offset) >= filtered_docs.size()) {
|
|
|
|
|
+ // Offset beyond results, return empty
|
|
|
|
|
+ return ::grpc::Status::OK;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ size_t end_index = std::min(static_cast<size_t>(offset + limit), filtered_docs.size());
|
|
|
|
|
+
|
|
|
|
|
+ for (size_t i = static_cast<size_t>(offset); i < end_index; ++i) {
|
|
|
|
|
+ DocumentToProto(filtered_docs[i], request->collection(), response->add_documents());
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Set next page token if there are more results
|
|
|
|
|
+ if (end_index < filtered_docs.size()) {
|
|
|
|
|
+ response->set_next_page_token(std::to_string(end_index));
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ spdlog::debug("Query: returned {} of {} documents", response->documents_size(), filtered_docs.size());
|
|
|
|
|
+ return ::grpc::Status::OK;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+::grpc::Status QueryServiceImpl::QueryStream(
|
|
|
|
|
+ ::grpc::ServerContext* /*context*/, const ::smartbotic::database::QueryRequest* request,
|
|
|
|
|
+ ::grpc::ServerWriter<::smartbotic::database::Document>* writer) {
|
|
|
|
|
+ spdlog::debug("QueryStream: collection={}", request->collection());
|
|
|
|
|
+
|
|
|
|
|
+ auto* coll = db_->GetCollection(request->collection());
|
|
|
|
|
+ if (coll == nullptr) {
|
|
|
|
|
+ return ::grpc::Status(::grpc::StatusCode::NOT_FOUND,
|
|
|
|
|
+ "Collection not found: " + request->collection());
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ auto all_docs = coll->GetAll();
|
|
|
|
|
+
|
|
|
|
|
+ for (const auto& doc : all_docs) {
|
|
|
|
|
+ // Apply filter if present
|
|
|
|
|
+ if (request->has_filter() && !MatchesFilter(doc, request->filter())) {
|
|
|
|
|
+ continue;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ ::smartbotic::database::Document proto_doc;
|
|
|
|
|
+ DocumentToProto(doc, request->collection(), &proto_doc);
|
|
|
|
|
+
|
|
|
|
|
+ if (!writer->Write(proto_doc)) {
|
|
|
|
|
+ spdlog::warn("QueryStream: client disconnected");
|
|
|
|
|
+ break;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return ::grpc::Status::OK;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+::grpc::Status QueryServiceImpl::Aggregate(::grpc::ServerContext* /*context*/,
|
|
|
|
|
+ const ::smartbotic::database::AggregateRequest* request,
|
|
|
|
|
+ ::smartbotic::database::AggregateResponse* response) {
|
|
|
|
|
+ spdlog::debug("Aggregate: collection={}", request->collection());
|
|
|
|
|
+
|
|
|
|
|
+ auto* coll = db_->GetCollection(request->collection());
|
|
|
|
|
+ if (coll == nullptr) {
|
|
|
|
|
+ return ::grpc::Status(::grpc::StatusCode::NOT_FOUND,
|
|
|
|
|
+ "Collection not found: " + request->collection());
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Get all matching documents
|
|
|
|
|
+ auto all_docs = coll->GetAll();
|
|
|
|
|
+ std::vector<storage::Document> filtered_docs;
|
|
|
|
|
+
|
|
|
|
|
+ if (request->has_filter()) {
|
|
|
|
|
+ for (const auto& doc : all_docs) {
|
|
|
|
|
+ if (MatchesFilter(doc, request->filter())) {
|
|
|
|
|
+ filtered_docs.push_back(doc);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ } else {
|
|
|
|
|
+ filtered_docs = std::move(all_docs);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // For simplicity, compute aggregations without grouping for now
|
|
|
|
|
+ auto* result = response->add_results();
|
|
|
|
|
+
|
|
|
|
|
+ for (const auto& agg : request->aggregations()) {
|
|
|
|
|
+ ::smartbotic::database::Value agg_value;
|
|
|
|
|
+
|
|
|
|
|
+ switch (agg.type()) {
|
|
|
|
|
+ case ::smartbotic::database::AGGREGATION_TYPE_COUNT:
|
|
|
|
|
+ agg_value.set_int_value(static_cast<int64_t>(filtered_docs.size()));
|
|
|
|
|
+ break;
|
|
|
|
|
+
|
|
|
|
|
+ case ::smartbotic::database::AGGREGATION_TYPE_SUM: {
|
|
|
|
|
+ double sum = 0.0;
|
|
|
|
|
+ for (const auto& doc : filtered_docs) {
|
|
|
|
|
+ auto val = GetFieldValue(doc.GetData(), agg.field());
|
|
|
|
|
+ if (val.is_number()) {
|
|
|
|
|
+ sum += val.get<double>();
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ agg_value.set_double_value(sum);
|
|
|
|
|
+ break;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ case ::smartbotic::database::AGGREGATION_TYPE_AVG: {
|
|
|
|
|
+ double sum = 0.0;
|
|
|
|
|
+ size_t count = 0;
|
|
|
|
|
+ for (const auto& doc : filtered_docs) {
|
|
|
|
|
+ auto val = GetFieldValue(doc.GetData(), agg.field());
|
|
|
|
|
+ if (val.is_number()) {
|
|
|
|
|
+ sum += val.get<double>();
|
|
|
|
|
+ ++count;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ agg_value.set_double_value(count > 0 ? sum / static_cast<double>(count) : 0.0);
|
|
|
|
|
+ break;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ case ::smartbotic::database::AGGREGATION_TYPE_MIN: {
|
|
|
|
|
+ std::optional<double> min_val;
|
|
|
|
|
+ for (const auto& doc : filtered_docs) {
|
|
|
|
|
+ auto val = GetFieldValue(doc.GetData(), agg.field());
|
|
|
|
|
+ if (val.is_number()) {
|
|
|
|
|
+ double v = val.get<double>();
|
|
|
|
|
+ if (!min_val || v < *min_val) {
|
|
|
|
|
+ min_val = v;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ if (min_val) {
|
|
|
|
|
+ agg_value.set_double_value(*min_val);
|
|
|
|
|
+ } else {
|
|
|
|
|
+ agg_value.set_null_value(true);
|
|
|
|
|
+ }
|
|
|
|
|
+ break;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ case ::smartbotic::database::AGGREGATION_TYPE_MAX: {
|
|
|
|
|
+ std::optional<double> max_val;
|
|
|
|
|
+ for (const auto& doc : filtered_docs) {
|
|
|
|
|
+ auto val = GetFieldValue(doc.GetData(), agg.field());
|
|
|
|
|
+ if (val.is_number()) {
|
|
|
|
|
+ double v = val.get<double>();
|
|
|
|
|
+ if (!max_val || v > *max_val) {
|
|
|
|
|
+ max_val = v;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ if (max_val) {
|
|
|
|
|
+ agg_value.set_double_value(*max_val);
|
|
|
|
|
+ } else {
|
|
|
|
|
+ agg_value.set_null_value(true);
|
|
|
|
|
+ }
|
|
|
|
|
+ break;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ default:
|
|
|
|
|
+ agg_value.set_null_value(true);
|
|
|
|
|
+ break;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ (*result->mutable_aggregations())[agg.alias()] = agg_value;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return ::grpc::Status::OK;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+::grpc::Status QueryServiceImpl::Count(::grpc::ServerContext* /*context*/,
|
|
|
|
|
+ const ::smartbotic::database::CountRequest* request,
|
|
|
|
|
+ ::smartbotic::database::CountResponse* response) {
|
|
|
|
|
+ spdlog::debug("Count: collection={}", request->collection());
|
|
|
|
|
+
|
|
|
|
|
+ auto* coll = db_->GetCollection(request->collection());
|
|
|
|
|
+ if (coll == nullptr) {
|
|
|
|
|
+ return ::grpc::Status(::grpc::StatusCode::NOT_FOUND,
|
|
|
|
|
+ "Collection not found: " + request->collection());
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if (!request->has_filter()) {
|
|
|
|
|
+ // No filter, return total count
|
|
|
|
|
+ response->set_count(static_cast<int64_t>(coll->Count()));
|
|
|
|
|
+ return ::grpc::Status::OK;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Count matching documents
|
|
|
|
|
+ auto all_docs = coll->GetAll();
|
|
|
|
|
+ int64_t count = 0;
|
|
|
|
|
+
|
|
|
|
|
+ for (const auto& doc : all_docs) {
|
|
|
|
|
+ if (MatchesFilter(doc, request->filter())) {
|
|
|
|
|
+ ++count;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ response->set_count(count);
|
|
|
|
|
+ return ::grpc::Status::OK;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto QueryServiceImpl::MatchesFilter(const storage::Document& doc,
|
|
|
|
|
+ const ::smartbotic::database::Filter& filter) const -> bool {
|
|
|
|
|
+ switch (filter.filter_type_case()) {
|
|
|
|
|
+ case ::smartbotic::database::Filter::kField:
|
|
|
|
|
+ return MatchesFieldFilter(doc, filter.field());
|
|
|
|
|
+ case ::smartbotic::database::Filter::kComposite:
|
|
|
|
|
+ return MatchesCompositeFilter(doc, filter.composite());
|
|
|
|
|
+ case ::smartbotic::database::Filter::FILTER_TYPE_NOT_SET:
|
|
|
|
|
+ default:
|
|
|
|
|
+ return true; // No filter means match all
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto QueryServiceImpl::MatchesFieldFilter(const storage::Document& doc,
|
|
|
|
|
+ const ::smartbotic::database::FieldFilter& filter) const
|
|
|
|
|
+ -> bool {
|
|
|
|
|
+ auto doc_value = GetFieldValue(doc.GetData(), filter.field());
|
|
|
|
|
+ return CompareValues(doc_value, filter.value(), filter.operator_());
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto QueryServiceImpl::MatchesCompositeFilter(
|
|
|
|
|
+ const storage::Document& doc, const ::smartbotic::database::CompositeFilter& filter) const -> bool {
|
|
|
|
|
+ switch (filter.operator_()) {
|
|
|
|
|
+ case ::smartbotic::database::COMPOSITE_OPERATOR_AND:
|
|
|
|
|
+ for (const auto& sub_filter : filter.filters()) {
|
|
|
|
|
+ if (!MatchesFilter(doc, sub_filter)) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return true;
|
|
|
|
|
+
|
|
|
|
|
+ case ::smartbotic::database::COMPOSITE_OPERATOR_OR:
|
|
|
|
|
+ for (const auto& sub_filter : filter.filters()) {
|
|
|
|
|
+ if (MatchesFilter(doc, sub_filter)) {
|
|
|
|
|
+ return true;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return false;
|
|
|
|
|
+
|
|
|
|
|
+ default:
|
|
|
|
|
+ return true;
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto QueryServiceImpl::GetFieldValue(const nlohmann::json& data, const std::string& path) const
|
|
|
|
|
+ -> nlohmann::json {
|
|
|
|
|
+ // Handle dot notation paths like "user.name"
|
|
|
|
|
+ nlohmann::json current = data;
|
|
|
|
|
+ std::string::size_type start = 0;
|
|
|
|
|
+ std::string::size_type pos = 0;
|
|
|
|
|
+
|
|
|
|
|
+ while ((pos = path.find('.', start)) != std::string::npos) {
|
|
|
|
|
+ std::string key = path.substr(start, pos - start);
|
|
|
|
|
+ if (!current.is_object() || !current.contains(key)) {
|
|
|
|
|
+ return nullptr;
|
|
|
|
|
+ }
|
|
|
|
|
+ current = current[key];
|
|
|
|
|
+ start = pos + 1;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Handle the last segment
|
|
|
|
|
+ std::string key = path.substr(start);
|
|
|
|
|
+ if (!current.is_object() || !current.contains(key)) {
|
|
|
|
|
+ return nullptr;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return current[key];
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+auto QueryServiceImpl::CompareValues(const nlohmann::json& doc_value,
|
|
|
|
|
+ const ::smartbotic::database::Value& filter_value,
|
|
|
|
|
+ ::smartbotic::database::FilterOperator op) const -> bool {
|
|
|
|
|
+ auto filter_json = ProtoValueToJson(filter_value);
|
|
|
|
|
+
|
|
|
|
|
+ switch (op) {
|
|
|
|
|
+ case ::smartbotic::database::FILTER_OPERATOR_EQUAL:
|
|
|
|
|
+ return doc_value == filter_json;
|
|
|
|
|
+
|
|
|
|
|
+ case ::smartbotic::database::FILTER_OPERATOR_NOT_EQUAL:
|
|
|
|
|
+ return doc_value != filter_json;
|
|
|
|
|
+
|
|
|
|
|
+ case ::smartbotic::database::FILTER_OPERATOR_LESS_THAN:
|
|
|
|
|
+ return doc_value < filter_json;
|
|
|
|
|
+
|
|
|
|
|
+ case ::smartbotic::database::FILTER_OPERATOR_LESS_THAN_OR_EQUAL:
|
|
|
|
|
+ return doc_value <= filter_json;
|
|
|
|
|
+
|
|
|
|
|
+ case ::smartbotic::database::FILTER_OPERATOR_GREATER_THAN:
|
|
|
|
|
+ return doc_value > filter_json;
|
|
|
|
|
+
|
|
|
|
|
+ case ::smartbotic::database::FILTER_OPERATOR_GREATER_THAN_OR_EQUAL:
|
|
|
|
|
+ return doc_value >= filter_json;
|
|
|
|
|
+
|
|
|
|
|
+ case ::smartbotic::database::FILTER_OPERATOR_IN:
|
|
|
|
|
+ if (filter_json.is_array()) {
|
|
|
|
|
+ for (const auto& item : filter_json) {
|
|
|
|
|
+ if (doc_value == item) {
|
|
|
|
|
+ return true;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return false;
|
|
|
|
|
+
|
|
|
|
|
+ case ::smartbotic::database::FILTER_OPERATOR_NOT_IN:
|
|
|
|
|
+ if (filter_json.is_array()) {
|
|
|
|
|
+ for (const auto& item : filter_json) {
|
|
|
|
|
+ if (doc_value == item) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return true;
|
|
|
|
|
+
|
|
|
|
|
+ case ::smartbotic::database::FILTER_OPERATOR_CONTAINS:
|
|
|
|
|
+ if (doc_value.is_string() && filter_json.is_string()) {
|
|
|
|
|
+ return doc_value.get<std::string>().find(filter_json.get<std::string>()) !=
|
|
|
|
|
+ std::string::npos;
|
|
|
|
|
+ }
|
|
|
|
|
+ if (doc_value.is_array()) {
|
|
|
|
|
+ for (const auto& item : doc_value) {
|
|
|
|
|
+ if (item == filter_json) {
|
|
|
|
|
+ return true;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return false;
|
|
|
|
|
+
|
|
|
|
|
+ case ::smartbotic::database::FILTER_OPERATOR_STARTS_WITH:
|
|
|
|
|
+ if (doc_value.is_string() && filter_json.is_string()) {
|
|
|
|
|
+ return doc_value.get<std::string>().rfind(filter_json.get<std::string>(), 0) == 0;
|
|
|
|
|
+ }
|
|
|
|
|
+ return false;
|
|
|
|
|
+
|
|
|
|
|
+ case ::smartbotic::database::FILTER_OPERATOR_ENDS_WITH:
|
|
|
|
|
+ if (doc_value.is_string() && filter_json.is_string()) {
|
|
|
|
|
+ const auto& doc_str = doc_value.get<std::string>();
|
|
|
|
|
+ const auto& filter_str = filter_json.get<std::string>();
|
|
|
|
|
+ if (doc_str.size() >= filter_str.size()) {
|
|
|
|
|
+ return doc_str.compare(doc_str.size() - filter_str.size(), filter_str.size(),
|
|
|
|
|
+ filter_str) == 0;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return false;
|
|
|
|
|
+
|
|
|
|
|
+ case ::smartbotic::database::FILTER_OPERATOR_IS_NULL:
|
|
|
|
|
+ return doc_value.is_null();
|
|
|
|
|
+
|
|
|
|
|
+ case ::smartbotic::database::FILTER_OPERATOR_IS_NOT_NULL:
|
|
|
|
|
+ return !doc_value.is_null();
|
|
|
|
|
+
|
|
|
|
|
+ default:
|
|
|
|
|
+ return true;
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+} // namespace smartbotic::database::grpc
|