diff --git a/publish/src/app.rs b/publish/src/app.rs index 27fac8a..aadf27d 100644 --- a/publish/src/app.rs +++ b/publish/src/app.rs @@ -19,8 +19,7 @@ use ldp::{RdfSource, RdfSourceUpdateResponse, ResourceRequestBuilder, Serializat use oxigraph::io::RdfFormat; use oxigraph::model::vocab::{rdf, rdfs}; use oxigraph::model::{BaseDirection, Dataset, NamedNode, Quad, Term}; -use tracing::{debug, debug_span, error, trace}; -use tracing::span::Span; +use tracing::{debug_span, error, field, trace}; use tracing_futures::Instrument; use crate::navigator::Navigator; use crate::rdf::curie::CurieHelper; @@ -37,7 +36,6 @@ pub(crate) enum Message { AddRdfSource(RdfSource), ConcludeTraversal, IndexQueryResults(HashMap), - CommitIndex, WindowClosed(window::Id), URLInputChanged(String), URLInputSubmitted, @@ -179,22 +177,26 @@ impl Publisher { match message { Message::RebuildIndex => { - if let Ok(writer) = self.index.writer() { - writer.remove_all().expect("Unable to clear index"); + let mut writer = self.index.writer().expect("Unable to obtain index writer"); + let clear_index_task = Task::perform(tokio::task::spawn_blocking(move || { + let _span = debug_span!("Clear Index").entered(); + writer.delete_all_documents().expect("Unable to clear index"); + writer.commit().expect("Unable to commit index operation"); + }), |_| Message::None); - let index_future = self.ontology - .index(&*language::ENGLISH_OR_UNTAGGED, None); - let index_task = Task::perform(index_future, |task_result| { - match task_result { - Ok(results) => Message::IndexQueryResults(results), - Err(err) => Message::ShowError(err.to_string()), - } - }); - task = Task::batch([ - index_task, - Task::done(Message::Traverse), - ]); - } + let index_future = self.ontology + .query_for_indexable_triples(&*language::ENGLISH_OR_UNTAGGED, None); + let index_task = Task::perform(index_future, |task_result| { + match task_result { + Ok(results) => Message::IndexQueryResults(results), + Err(err) => Message::ShowError(err.to_string()), + } + }); + + task = clear_index_task.chain(Task::batch([ + index_task, + Task::done(Message::Traverse), + ])); } Message::Traverse => { let client = self.http_client.clone(); @@ -216,7 +218,7 @@ impl Publisher { Message::ConcludeTraversal => { if let Some(traversal) = self.traversal.take() { let index_task = self.ontology - .index(&*language::ENGLISH_OR_UNTAGGED, Some(traversal)); + .query_for_indexable_triples(&*language::ENGLISH_OR_UNTAGGED, Some(traversal)); task = Task::perform(index_task, |task_result| { match task_result { Ok(results) => Message::IndexQueryResults(results), @@ -226,41 +228,38 @@ impl Publisher { } } Message::IndexQueryResults(results) => { - let mut index = self.index.clone(); + let mut writer = self.index.writer().expect("Unable to obtain index writer"); let curie_helper = self.curie_helper.clone(); task = Task::perform(tokio::task::spawn_blocking(move || { - let _span = debug_span!("Indexing").entered(); - if let Ok(writer) = index.writer() { - let mut counter = 0; - for (individual, entry) in results { - let mut document = doc!( - Schema::type_field() => entry.category_id, - Schema::iri_field() => individual.as_str(), - ); + let span = debug_span!("Index Query Results", documents = field::Empty).entered(); + let mut counter = 0usize; + for (individual, entry) in results { + let mut document = doc!( + Schema::type_field() => entry.category_id, + Schema::iri_field() => individual.as_str(), + ); - if let Some(curie) = curie_helper.abbreviate(individual.as_str()) { - document.add_text(Schema::curie_field(), curie); - } - - for (key, value) in entry.fields { - document.add_text(Schema::field(&key, language::ENGLISH_PRIMARY), value.as_str()); - } - writer.add(document).expect("Failed to add document to search index"); - counter += 1; + if let Some(curie) = curie_helper.abbreviate(individual.as_str()) { + document.add_text(Schema::curie_field(), curie); } - debug!("Added {counter} documents to search index"); + + for (key, value) in entry.fields { + document.add_text(Schema::field(&key, language::ENGLISH_PRIMARY), value.as_str()); + } + writer.add_document(document).expect("Failed to add document to search index"); + counter += 1; + } + + span.record("documents", counter); + writer.commit() + }), |result| { + if let Err(err) = result { + Message::ShowError(err.to_string()) + } else { + Message::None } - }), |result| if let Err(err) = result { - Message::ShowError(err.to_string()) - } else { - Message::CommitIndex }); } - Message::CommitIndex => { - if let Err(err) = self.index.commit() { - task = Task::done(Message::ShowError(err.to_string())); - } - } Message::WindowClosed(id) => { if self.window_id == id { task = iced::exit(); diff --git a/publish/src/rdf/materialize.rs b/publish/src/rdf/materialize.rs index fa64948..a0eac69 100644 --- a/publish/src/rdf/materialize.rs +++ b/publish/src/rdf/materialize.rs @@ -9,8 +9,7 @@ const INFERENCE_GRAPH: NamedNodeRef = NamedNodeRef::new_unchecked("https://graph const RDF_SCHEMA: NamedNodeRef = NamedNodeRef::new_unchecked("http://www.w3.org/2000/01/rdf-schema#"); pub fn same_as(store: &mut Store) -> error::Result<()> { - let span = debug_span!("Materialize owl:sameAs"); - let _enter = span.enter(); + let _span = debug_span!("Materialize owl:sameAs").entered(); let additional_quads = store .quads_for_pattern(None, Some(owl::SAME_AS), None, None) @@ -58,8 +57,7 @@ pub fn same_as(store: &mut Store) -> error::Result<()> { } pub fn super_properties(store: &mut Store) -> error::Result<()> { - let span = debug_span!("Materialize rdfs:subPropertyOf"); - let _enter = span.enter(); + let _span = debug_span!("Materialize rdfs:subPropertyOf").entered(); let update = SparqlEvaluator::new() .with_prefix("rdfs", RDF_SCHEMA.as_str())? @@ -88,8 +86,7 @@ pub fn super_properties(store: &mut Store) -> error::Result<()> { } pub fn super_classes(store: &mut Store) -> error::Result<()> { - let span = debug_span!("Materialize rdfs:subClassOf"); - let _enter = span.enter(); + let _span = debug_span!("Materialize rdfs:subClassOf").entered(); let update = SparqlEvaluator::new() .with_prefix("rdfs", RDF_SCHEMA.as_str())? diff --git a/publish/src/rdf/ontology.rs b/publish/src/rdf/ontology.rs index 9f60ddb..7db1869 100644 --- a/publish/src/rdf/ontology.rs +++ b/publish/src/rdf/ontology.rs @@ -9,7 +9,7 @@ use std::path::{Path, PathBuf}; use std::sync::LazyLock; use iced::futures::TryFutureExt; use oxigraph::store::Store; -use tracing::debug_span; +use tracing::{debug_span, field}; use crate::rdf::{conversion, materialize}; use crate::rdf::language::LanguageCondition; @@ -173,7 +173,7 @@ impl Ontology { .collect::() } - pub fn index(&self, language: &LanguageCondition, source: Option) -> impl Future>> + 'static { + pub fn query_for_indexable_triples(&self, language: &LanguageCondition, source: Option) -> impl Future>> + 'static { let language_filter = language.to_filter_expression("fieldValue"); let query = format!(r#"SELECT ?individual ?categoryId ?fieldName ?fieldValue WHERE {{ @@ -196,7 +196,7 @@ WHERE {{ let store = self.store.clone(); tokio::task::spawn_blocking(move || { - let _span = debug_span!("Index Query").entered(); + let span = debug_span!("Indexable Triples Query", solutions = field::Empty).entered(); let query_results = if let Some(source) = &source { sparql.on_queryable_dataset(source).execute() } else { @@ -205,6 +205,7 @@ WHERE {{ let mut results = HashMap::new(); if let QueryResults::Solutions(solutions) = query_results { + let mut counter = 0usize; for solution in solutions.filter_map(Result::ok) { let individual = solution.get("individual").and_then(conversion::term_to_named_node); let catalog_id = solution.get("categoryId").and_then(conversion::term_to_u64); @@ -219,7 +220,9 @@ WHERE {{ fields: HashMap::from_iter([(field_name.to_owned(), field_value.to_owned())]), }); } + counter += 1; } + span.record("solutions", counter); } else { unreachable!() } diff --git a/search/src/index.rs b/search/src/index.rs index 010d62b..bb3ca5e 100644 --- a/search/src/index.rs +++ b/search/src/index.rs @@ -1,14 +1,13 @@ use std::fs; -use crate::{error, IndexWriter}; +use crate::error; use crate::schema::Schema; use std::path::PathBuf; -use std::sync::Arc; use tantivy::collector::TopDocs; use tantivy::directory::{ManagedDirectory, MmapDirectory}; use tantivy::query::{BooleanQuery, Occur, QueryParser, TermQuery}; use tantivy::schema::{Field, IndexRecordOption}; use tantivy::tokenizer::{LowerCaser, NgramTokenizer, TextAnalyzer, TokenizerManager}; -use tantivy::{Index, IndexReader, ReloadPolicy, TantivyDocument, Term}; +use tantivy::{Index, IndexReader, IndexWriter, ReloadPolicy, TantivyDocument, Term}; use tracing::debug_span; #[derive(Default)] @@ -52,16 +51,13 @@ impl SearchIndexBuilder { Ok(SearchIndex { index, reader, - writer: None, }) } } -#[derive(Clone)] pub struct SearchIndex { index: Index, reader: IndexReader, - writer: Option>, } impl SearchIndex { @@ -69,23 +65,8 @@ impl SearchIndex { SearchIndexBuilder::default() } - pub fn writer(&mut self) -> crate::Result<&IndexWriter> { - if self.writer.is_none() { - let inner = self.index.writer(128 * 1024 * 1024)?; - self.writer = Some(Arc::new(IndexWriter { inner })); - } - - Ok(self.writer.as_ref().unwrap()) - } - - pub fn commit(&mut self) -> crate::Result<()> { - if let Some(writer) = self.writer.take() { - if let Some(mut writer) = Arc::into_inner(writer) { - writer.commit()?; - } - } - - Ok(()) + pub fn writer(&mut self) -> crate::Result { + Ok(self.index.writer(128 * 1024 * 1024)?) } pub fn query( @@ -95,7 +76,7 @@ impl SearchIndex { default_fields: Vec, limit: usize, ) -> error::Result> { - let _span = debug_span!("Search Query").entered(); + let _enter = debug_span!("Search Query").entered(); let parser = QueryParser::for_index(&self.index, default_fields); let (user_query, _) = parser.parse_query_lenient(user_query); diff --git a/search/src/lib.rs b/search/src/lib.rs index b8f34ad..5781210 100644 --- a/search/src/lib.rs +++ b/search/src/lib.rs @@ -1,7 +1,6 @@ mod error; mod index; mod schema; -mod update; pub use tantivy::TantivyDocument as SearchDocument; pub use tantivy::doc; @@ -9,5 +8,4 @@ pub use tantivy::schema::document::Value; pub use error::{Result, SearchError}; pub use index::{SearchIndex, SearchIndexBuilder}; -pub use schema::Schema; -pub use update::IndexWriter; \ No newline at end of file +pub use schema::Schema; \ No newline at end of file diff --git a/search/src/update.rs b/search/src/update.rs deleted file mode 100644 index 34750cc..0000000 --- a/search/src/update.rs +++ /dev/null @@ -1,28 +0,0 @@ -use tantivy::Term; -use crate::Schema; - -pub struct IndexWriter { - pub(crate) inner: tantivy::IndexWriter, -} - -impl IndexWriter { - pub fn add(&self, document: crate::SearchDocument) -> crate::Result<()> { - let _ = self.inner.add_document(document)?; - Ok(()) - } - - pub fn remove_all_of_type(&self, type_: u64) -> crate::Result<()> { - self.inner.delete_term(Term::from_field_u64(Schema::type_field(), type_)); - Ok(()) - } - - pub fn remove_all(&self) -> crate::Result<()> { - self.inner.delete_all_documents()?; - Ok(()) - } - - pub fn commit(&mut self) -> crate::Result<()> { - self.inner.commit()?; - Ok(()) - } -} \ No newline at end of file