save

Author: tobixdev

Cargo.lock

@@ -4,7 +4,7 @@ use anyhow::{bail, Context};
 use clap::Parser;
 use graphfusion::io::{RdfFormat, RdfParser, RdfSerializer};
 use graphfusion::model::{GraphName, NamedNode};
-use graphfusion::store::Store;
+use graphfusion_engine::store::Store;
 use graphfusion_web::ServerConfig;
 use std::ffi::OsStr;
 use std::fs::File;

cli/src/main.rs

@@ -0,0 +1,132 @@
+use crate::planner::GraphFusionPlanner;
+use crate::sparql::error::QueryEvaluationError;
+use crate::sparql::{evaluate_query, Query, QueryExplanation, QueryOptions, QueryResults};
+use crate::DFResult;
+use arrow_rdf::TABLE_QUADS;
+use datafusion::dataframe::DataFrame;
+use datafusion::datasource::{DefaultTableSource, TableProvider};
+use datafusion::execution::{SendableRecordBatchStream, SessionStateBuilder};
+use datafusion::functions_aggregate::first_last::FirstValue;
+use datafusion::logical_expr::{AggregateUDF, LogicalPlan, LogicalPlanBuilder};
+use datafusion::prelude::SessionContext;
+use graphfusion_logical::paths::PathToJoinsRule;
+use graphfusion_logical::patterns::{
+    compute_filters_for_pattern, PatternNode, PatternToProjectionRule,
+};
+use model::{BlankNode, GraphNameRef, NamedNodeRef, QuadRef, SubjectRef, TermRef};
+use spargebra::term::TermPattern;
+use std::sync::Arc;
+
+pub struct GraphFusionInstance {
+    /// The DataFusion [SessionContext].
+    ctx: SessionContext,
+    /// The storage that backs this instance.
+    storage: Arc<dyn TableProvider>,
+}
+
+impl GraphFusionInstance {
+    /// Creates a new [GraphFusionInstance] with the default configuration and the given `storage`.
+    pub fn with_storage(storage: Arc<dyn TableProvider>) -> DFResult<Self> {
+        let state = SessionStateBuilder::new()
+            .with_query_planner(Arc::new(GraphFusionPlanner))
+            .with_aggregate_functions(vec![AggregateUDF::from(FirstValue::new()).into()])
+            .with_optimizer_rule(Arc::new(PathToJoinsRule::new(Arc::clone(&storage))))
+            .with_optimizer_rule(Arc::new(PatternToProjectionRule))
+            .build();
+
+        let session_context = SessionContext::from(state);
+        session_context.register_table("quads", Arc::clone(&storage))?;
+
+        Ok(Self {
+            ctx: session_context,
+            storage,
+        })
+    }
+
+    /// Checks whether `quad` is contained in the instance.
+    async fn contains(&self, quad: &QuadRef<'_>) -> DFResult<bool> {
+        let pattern_plan = self
+            .match_pattern(
+                Some(quad.graph_name),
+                Some(quad.subject),
+                Some(quad.predicate),
+                Some(quad.object),
+            )
+            .await?;
+
+        let count = DataFrame::new(self.ctx.state(), pattern_plan)
+            .count()
+            .await?;
+
+        Ok(count > 0)
+    }
+
+    /// Returns the number of quads in the instance.
+    async fn len(&self) -> DFResult<usize> {
+        self.ctx.table(TABLE_QUADS).await?.count().await
+    }
+
+    /// Returns a stream of all quads that match the given pattern.
+    async fn quads_for_pattern(
+        &self,
+        graph_name: Option<GraphNameRef<'_>>,
+        subject: Option<SubjectRef<'_>>,
+        predicate: Option<NamedNodeRef<'_>>,
+        object: Option<TermRef<'_>>,
+    ) -> DFResult<SendableRecordBatchStream> {
+        let plan = self
+            .match_pattern(graph_name, subject, predicate, object)
+            .await?;
+        let result = DataFrame::new(self.ctx.state(), plan)
+            .execute_stream()
+            .await?;
+        Ok(result)
+    }
+
+    /// Evaluates a SPARQL [Query] over the instance.
+    async fn execute_query(
+        &self,
+        query: &Query,
+        options: QueryOptions,
+    ) -> Result<(QueryResults, Option<QueryExplanation>), QueryEvaluationError> {
+        evaluate_query(&self.ctx, query, options).await
+    }
+
+    /// Creates a [LogicalPlan] for computing all quads that match the given pattern.
+    async fn match_pattern(
+        &self,
+        graph_name: Option<GraphNameRef<'_>>,
+        subject: Option<SubjectRef<'_>>,
+        predicate: Option<NamedNodeRef<'_>>,
+        object: Option<TermRef<'_>>,
+    ) -> DFResult<LogicalPlan> {
+        let plan = LogicalPlanBuilder::scan(
+            TABLE_QUADS,
+            Arc::new(DefaultTableSource::new(Arc::clone(&self.storage))),
+            None,
+        )?;
+
+        let graph_name = graph_name
+            .map(|g| TermPattern::from(g.to_owned()))
+            .unwrap_or(TermPattern::BlankNode(BlankNode::default()));
+        let subject = subject
+            .map(|s| TermPattern::from(s.into_owned()))
+            .unwrap_or(TermPattern::BlankNode(BlankNode::default()));
+        let predicate = predicate
+            .map(|p| TermPattern::from(p.into_owned()))
+            .unwrap_or(TermPattern::BlankNode(BlankNode::default()));
+        let object = object
+            .map(|o| TermPattern::from(o.into_owned()))
+            .unwrap_or(TermPattern::BlankNode(BlankNode::default()));
+        let pattern_node = PatternNode::try_new(
+            plan.clone().build()?,
+            vec![graph_name, subject, predicate, object],
+        )?;
+        let filter = compute_filters_for_pattern(&pattern_node);
+
+        let plan = match filter {
+            None => plan.build()?,
+            Some(filter) => plan.filter(filter)?.build()?,
+        };
+    }
+}

lib/graphfusion-engine/src/engine.rs

@@ -3,8 +3,11 @@ extern crate core;
 pub mod error;
 pub mod results;
 pub mod sparql;
-mod triple_store;
+mod quad_storage;
+mod planner;
+mod engine;
 
-pub use triple_store::QuadStorage;
+pub use quad_storage::QuadStorage;
+pub use engine::GraphFusionInstance;
 
 type DFResult<T> = datafusion::error::Result<T>;

lib/graphfusion-engine/src/lib.rs

@@ -1,10 +1,10 @@
-use crate::paths::KleenePlusPathPlanner;
 use async_trait::async_trait;
 use datafusion::execution::context::QueryPlanner;
 use datafusion::execution::SessionState;
 use datafusion::logical_expr::LogicalPlan;
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion::physical_planner::{DefaultPhysicalPlanner, PhysicalPlanner};
+use graphfusion_physical::paths::KleenePlusPathPlanner;
 use std::sync::Arc;
 
 #[derive(Debug, Default)]
@@ -18,7 +18,7 @@ impl QueryPlanner for GraphFusionPlanner {
         session_state: &SessionState,
     ) -> datafusion::common::Result<Arc<dyn ExecutionPlan>> {
         let physical_planner = DefaultPhysicalPlanner::with_extension_planners(vec![Arc::new(
-            KleenePlusPathPlanner::new(),
+            KleenePlusPathPlanner,
         )]);
         physical_planner
             .create_physical_plan(logical_plan, session_state)

lib/graphfusion-physical/src/planner.rs renamed to lib/graphfusion-engine/src/planner.rs

@@ -0,0 +1,23 @@
+use std::sync::Arc;
+use async_trait::async_trait;
+use datafusion::common::DataFusionError;
+use datafusion::datasource::TableProvider;
+use model::{Quad, QuadRef};
+
+#[async_trait]
+#[allow(clippy::len_without_is_empty)]
+pub trait QuadStorage {
+    /// Returns the table name of this [QuadStorage]. This name is used to register a table in the
+    /// DataFusion engine.
+    async fn table_name(&self) -> &str;
+
+    /// Returns the [TableProvider] for this [QuadStorage]. This provider is registered in the
+    /// DataFusion session and used for planning the execution of queries.
+    async fn table_provider(&self) -> Arc<dyn TableProvider>;
+
+    /// Loads the given quads into the storage.
+    async fn load_quads(&self, quads: Vec<Quad>) -> Result<usize, DataFusionError>;
+
+    /// Removes the given quad from the storage.
+    async fn remove<'a>(&self, quad: QuadRef<'_>) -> Result<bool, DataFusionError>;
+}

lib/graphfusion-engine/src/quad_storage.rs

@@ -1,6 +1,6 @@
 //! [SPARQL](https://www.w3.org/TR/sparql11-overview/) implementation.
 //!
-//! Stores execute SPARQL. See [`Store`](crate::store::Store::query()) for an example.
+//! Stores execute SPARQL. See [`Store`](graphfusion::store::Store::query()) for an example.
 
 mod algebra;
 pub mod error;

lib/graphfusion-engine/src/sparql/mod.rs

@@ -8,17 +8,18 @@ use spargebra::term::TermPattern;
 use std::cmp::Ordering;
 use std::collections::HashMap;
 use std::fmt;
+use std::fmt::Display;
 use std::sync::Arc;
 
 #[derive(PartialEq, Eq, Hash)]
 pub struct PatternNode {
     input: LogicalPlan,
-    patterns: Vec<TermPattern>,
+    patterns: Vec<dyn PatternElement>,
     schema: DFSchemaRef,
 }
 
 impl PatternNode {
-    pub fn try_new(input: LogicalPlan, patterns: Vec<TermPattern>) -> DFResult<Self> {
+    pub fn try_new(input: LogicalPlan, patterns: Vec<dyn PatternElement>) -> DFResult<Self> {
         if input.schema().columns().len() != patterns.len() {
             return plan_err!("Patterns must match the number of column of inner.");
         }

lib/graphfusion-engine/src/triple_store.rs

@@ -1,14 +1,7 @@
 mod logical;
 mod rewrite;
+mod pattern_element;
 
 pub use logical::PatternNode;
+pub use rewrite::compute_filters_for_pattern;
 pub use rewrite::PatternToProjectionRule;
-use spargebra::term::TermPattern;
-
-fn pattern_to_variable_name(pattern: &TermPattern) -> Option<String> {
-    match pattern {
-        TermPattern::BlankNode(bnode) => Some(format!("_:{}", bnode.as_ref().as_str())),
-        TermPattern::Variable(var) => Some(var.as_str().into()),
-        _ => None,
-    }
-}