Fix more bugs

Author: magbak

lib/maplib/src/mapping/expansion/validation.rs

@@ -263,7 +263,7 @@ fn infer_validate_mapping_column_type_from_ptype(
                     ))
                 }
             } else if ptype_is_blank(nn.as_ref()) {
-                if datatype.is_string()  {
+                if datatype.is_string() {
                     Ok(MappingColumnType::Flat(
                         BaseRDFNodeType::BlankNode.into_default_input_rdf_node_state(),
                     ))
@@ -284,9 +284,7 @@ fn infer_validate_mapping_column_type_from_ptype(
                 } else {
                     let ptype_rdf_node_type = BaseRDFNodeType::Literal(nn.clone());
                     let ptype_dt = ptype_rdf_node_type.default_input_polars_data_type();
-                    if ptype_dt.is_string() && datatype.is_string()
-                        || &ptype_dt == datatype
-                    {
+                    if ptype_dt.is_string() && datatype.is_string() || &ptype_dt == datatype {
                         Ok(MappingColumnType::Flat(
                             ptype_rdf_node_type.into_default_input_rdf_node_state(),
                         ))

lib/query_processing/src/expressions/functions.rs

@@ -1625,7 +1625,8 @@ pub fn func_expression(
                                 col(first_context.as_str())
                                     .struct_()
                                     .field_by_name(&t.field_col_name())
-                                    .is_null().not(),
+                                    .is_null()
+                                    .not(),
                             )
                             .then(lit(rdf_named_node_to_polars_literal_value(l)))
                             .otherwise(lit(LiteralValue::untyped_null()).cast(iri_dtype.clone())),

lib/query_processing/src/graph_patterns/cats.rs

@@ -1,12 +1,12 @@
-use polars::prelude::{by_name, col, LazyFrame};
+use polars::prelude::{by_name, col, JoinType, LazyFrame};
 use representation::cats::{maybe_decode_expr, optional_maybe_decode_expr, CatReEnc, Cats};
 use representation::solution_mapping::BaseCatState;
 use representation::{BaseRDFNodeType, RDFNodeState};
 use std::sync::Arc;
 
 pub enum CatOperation {
     Decode,
-    ReEnc(CatReEnc, bool),
+    ReEnc(CatReEnc),
 }
 
 impl CatOperation {
@@ -44,14 +44,14 @@ impl CatOperation {
                     ));
                 }
             }
-            CatOperation::ReEnc(cat_re_enc, forget_others) => {
+            CatOperation::ReEnc(cat_re_enc) => {
                 if !t.is_multi() {
-                    mappings = cat_re_enc.re_encode(mappings, c, forget_others);
+                    mappings = cat_re_enc.re_encode(mappings, c, false);
                 } else {
                     let tmp = uuid::Uuid::new_v4().to_string();
                     let n = base_t.field_col_name();
                     mappings = mappings.with_column(col(c).struct_().field_by_name(&n).alias(&tmp));
-                    mappings = cat_re_enc.re_encode(mappings, &tmp, forget_others);
+                    mappings = cat_re_enc.re_encode(mappings, &tmp, false);
                     mappings = mappings
                         .with_column(col(c).struct_().with_fields(vec![col(&tmp).alias(&n)]));
                     mappings = mappings.drop(by_name([tmp], true));
@@ -93,7 +93,7 @@ pub fn create_compatible_cats(
                 (
                     BaseCatState::CategoricalNative(false, Some(left_local_cats.clone())),
                     None,
-                    Some(CatOperation::ReEnc(re_enc, true)),
+                    Some(CatOperation::ReEnc(re_enc)),
                 )
             }
             BaseCatState::String | BaseCatState::NonString => {

lib/representation/Cargo.toml

@@ -20,6 +20,7 @@ thiserror.workspace = true
 pyo3 = {workspace = true, features = ["chrono-tz", "chrono"]}
 oxsdatatypes.workspace = true
 rayon.workspace = true
+nohash-hasher.workspace = true
 uuid.workspace = true
 
 [lints.rust]

lib/representation/src/cats.rs

@@ -8,15 +8,18 @@ mod split;
 pub use decode::*;
 pub use encode::*;
 pub use globalize::*;
+pub use image::*;
 pub use re_encode::*;
 pub use split::*;
-pub use image::*;
 
 use crate::BaseRDFNodeType;
+use nohash_hasher::NoHashHasher;
 use oxrdf::NamedNode;
 use polars::prelude::DataFrame;
 use std::collections::{BTreeMap, HashMap};
+use std::hash::BuildHasherDefault;
 use std::sync::Arc;
+use uuid::Uuid;
 
 const SUBJECT_PREFIX_COL_NAME: &str = "subject_prefix";
 const OBJECT_PREFIX_COL_NAME: &str = "object_prefix";
@@ -62,7 +65,7 @@ pub struct EncodedTriples {
 pub struct CatEncs {
     // We use a BTree map to keep strings sorted
     pub map: BTreeMap<String, u32>,
-    pub rev_map: HashMap<u32, String>,
+    pub rev_map: HashMap<u32, String, BuildHasherDefault<NoHashHasher<u32>>>,
 }
 
 #[derive(Debug, Clone)]
@@ -101,7 +104,7 @@ impl Cats {
             iri_height: 0,
             blank_height: 0,
             literal_height_map: Default::default(),
-            uuid: uuid::Uuid::new_v4().to_string(),
+            uuid: Uuid::new_v4().to_string(),
         };
         cats.iri_height = cats.calc_new_iri_height();
         cats.blank_height = cats.calc_new_blank_height();

lib/representation/src/cats/decode.rs

@@ -3,9 +3,10 @@ use crate::solution_mapping::BaseCatState;
 use crate::BaseRDFNodeType;
 use polars::datatypes::{DataType, Field, PlSmallStr};
 use polars::frame::DataFrame;
-use polars::prelude::{col, Column, Expr, IntoColumn, IntoLazy};
+use polars::prelude::{col, Column, Expr, IntoColumn, IntoLazy, IntoSeries, StringChunked, UInt32Chunked};
 use polars::series::Series;
 use std::sync::Arc;
+use rayon::iter::{IntoParallelRefIterator, ParallelBridge, ParallelIterator};
 
 impl CatEncs {
     pub fn decode(&self, ser: &Series, cat_type: &CatType) -> Series {
@@ -94,8 +95,8 @@ impl Cats {
             encs.extend(local_cats.get_encs(t))
         }
         let u32s = ser.u32().unwrap();
-        let mut strings = Vec::with_capacity(ser.len());
-        for u in u32s {
+        let us:Vec<_> = u32s.iter().collect();
+        let strings: Vec<_> = us.par_iter().map(|u| {
             let s = if let Some(u) = u {
                 let mut s = None;
                 for (t, e) in &encs {
@@ -108,9 +109,9 @@ impl Cats {
             } else {
                 None
             };
-            strings.push(s)
-        }
-        Series::from_iter(strings.into_iter())
+            s
+        }).collect();
+        Series::from_iter(strings)
     }
 }
 
@@ -172,7 +173,7 @@ pub fn decode_expr(
         move |x| {
             let original_name = x.name().to_string();
             let mut s = global_cats.decode(
-                &x.as_materialized_series(),
+                x.as_materialized_series(),
                 &base_rdf_node_type,
                 local_cats.as_ref().map(|x| x.clone()),
             );

lib/representation/src/cats/encode.rs

@@ -10,13 +10,14 @@ use polars::frame::DataFrame;
 use polars::prelude::{col, lit, IntoLazy, Series};
 use rayon::iter::{IntoParallelIterator, IntoParallelRefIterator, ParallelIterator};
 use std::collections::HashMap;
+use std::hash::BuildHasherDefault;
 use std::sync::Arc;
 
 impl CatEncs {
     pub fn new_empty() -> CatEncs {
         CatEncs {
             map: Default::default(),
-            rev_map: Default::default(),
+            rev_map: HashMap::with_capacity_and_hasher(2, BuildHasherDefault::default()),
         }
     }
 
@@ -532,42 +533,61 @@ pub fn encode_triples(
         let out: Vec<_> = dfs
             .into_par_iter()
             .map(|mut df| {
-                let subject_out =
-                    if let Some(subject_prefix_map) = prefix_maps.get(SUBJECT_COL_NAME) {
-                        let u = df
-                            .drop_in_place(SUBJECT_PREFIX_COL_NAME)
-                            .unwrap()
-                            .as_materialized_series_maintain_scalar()
-                            .u32()
-                            .unwrap()
-                            .first()
-                            .unwrap();
-                        Some(subject_prefix_map.get(&u).unwrap().clone())
-                    } else {
-                        None
-                    };
-                let object_out = if let Some(object_prefix_map) = prefix_maps.get(OBJECT_COL_NAME) {
-                    let u = df
-                        .drop_in_place(OBJECT_PREFIX_COL_NAME)
-                        .unwrap()
-                        .as_materialized_series_maintain_scalar()
-                        .u32()
-                        .unwrap()
-                        .first()
-                        .unwrap();
-                    Some(object_prefix_map.get(&u).unwrap().clone())
-                } else {
+                if df.height() == 0 {
                     None
-                };
-                (df, subject_out, object_out)
+                } else {
+                    let subject_out =
+                        if let Some(subject_prefix_map) = prefix_maps.get(SUBJECT_COL_NAME) {
+                            let ser = df
+                                .drop_in_place(SUBJECT_PREFIX_COL_NAME)
+                                .unwrap()
+                                .as_materialized_series_maintain_scalar();
+                            let u32ch = ser.u32().unwrap();
+                            let i = u32ch.first_non_null();
+                            if let Some(i) = i {
+                                let u = u32ch.get(i);
+                                if let Some(u) = u {
+                                    Some(subject_prefix_map.get(&u).unwrap().clone())
+                                } else {
+                                    None
+                                }
+                            } else {
+                                None
+                            }
+                        } else {
+                            None
+                        };
+                    let object_out =
+                        if let Some(object_prefix_map) = prefix_maps.get(OBJECT_COL_NAME) {
+                            let ser = df
+                                .drop_in_place(OBJECT_PREFIX_COL_NAME)
+                                .unwrap()
+                                .as_materialized_series_maintain_scalar();
+                            let u32ch = ser.u32().unwrap();
+                            let i = u32ch.first_non_null();
+                            if let Some(i) = i {
+                                let u = u32ch.get(i);
+                                if let Some(u) = u {
+                                    Some(object_prefix_map.get(&u).unwrap().clone())
+                                } else {
+                                    None
+                                }
+                            } else {
+                                None
+                            }
+                        } else {
+                            None
+                        };
+                    Some((df, subject_out, object_out))
+                }
             })
             .collect();
         out
     } else {
-        vec![(mappings, None, None)]
+        vec![Some((mappings, None, None))]
     };
     let mut new_out = vec![];
-    for (df, subject, object) in out {
+    for (df, subject, object) in out.into_iter().filter(|x| x.is_some()).map(|x| x.unwrap()) {
         let subject = if matches!(subject_cat_state, BaseCatState::CategoricalNative(..)) {
             if let Some(subject) = subject {
                 Some(CatType::Prefix(NamedNode::new_unchecked(subject)))

lib/representation/src/cats/image.rs

@@ -2,8 +2,10 @@ use super::{reencode_solution_mappings, CatEncs};
 use super::{CatReEnc, Cats};
 use crate::solution_mapping::{BaseCatState, EagerSolutionMappings, SolutionMappings};
 use crate::{BaseRDFNodeType, RDFNodeState};
+use nohash_hasher::NoHashHasher;
 use rayon::iter::{IntoParallelIterator, IntoParallelRefIterator, ParallelIterator};
 use std::collections::{BTreeMap, HashMap, HashSet};
+use std::hash::BuildHasherDefault;
 use std::sync::Arc;
 
 impl Cats {
@@ -139,7 +141,7 @@ impl Cats {
 
 impl CatEncs {
     pub fn image(&self, s: &HashSet<u32>) -> Option<CatEncs> {
-        let rev_map: HashMap<_, _> = s
+        let rev_map: HashMap<_, _, BuildHasherDefault<NoHashHasher<u32>>> = s
             .par_iter()
             .map(|x| {
                 if let Some(s) = self.rev_map.get(x) {
@@ -160,23 +162,30 @@ impl CatEncs {
     }
 }
 
-pub fn new_solution_mapping_cats(sms: Vec<EagerSolutionMappings>, global_cats:&Cats) -> (Vec<EagerSolutionMappings>, Cats) {
+pub fn new_solution_mapping_cats(
+    sms: Vec<EagerSolutionMappings>,
+    global_cats: &Cats,
+) -> (Vec<EagerSolutionMappings>, Cats) {
     let smsref: Vec<_> = sms.iter().collect();
     let mut cats = global_cats.mappings_cat_image(&smsref);
 
     let reenc = cats.merge_solution_mappings_locals(&smsref);
-    let new_sms:Vec<_> = sms.into_par_iter().map(|sm| {
-        reencode_solution_mappings(sm, &reenc)
-    }).collect();
+    let new_sms: Vec<_> = sms
+        .into_par_iter()
+        .map(|sm| reencode_solution_mappings(sm, &reenc))
+        .collect();
     (new_sms, cats)
 }
 
-pub fn set_global_cats_as_local(rdf_node_types:&mut HashMap<String, RDFNodeState>, cats:Arc<Cats>)  {
+pub fn set_global_cats_as_local(
+    rdf_node_types: &mut HashMap<String, RDFNodeState>,
+    cats: Arc<Cats>,
+) {
     for (_, s) in rdf_node_types {
         for v in s.map.values_mut() {
             if matches!(v, BaseCatState::CategoricalNative(_, _)) {
                 *v = BaseCatState::CategoricalNative(false, Some(cats.clone()));
             }
         }
     }
-}
+}

lib/representation/src/rdf_to_polars.rs

@@ -1,5 +1,6 @@
 use crate::{LANG_STRING_LANG_FIELD, LANG_STRING_VALUE_FIELD};
 use chrono::{DateTime, NaiveDate, NaiveDateTime, Utc};
+use chrono_tz::Tz;
 use log::warn;
 use oxrdf::vocab::{rdf, xsd};
 use oxrdf::{BlankNode, Literal, NamedNode, NamedNodeRef, Term};
@@ -9,7 +10,6 @@ use polars::prelude::{
 };
 use std::ops::Deref;
 use std::str::FromStr;
-use chrono_tz::Tz;
 
 pub fn rdf_term_to_polars_expr(term: &Term) -> Expr {
     match term {
@@ -399,7 +399,10 @@ pub fn polars_literal_values_to_series(literal_values: Vec<LiteralValue>, name:
                         })
                         .collect::<Vec<Option<i64>>>(),
                 )
-                .cast(&DataType::Datetime(*t, Some(tz.unwrap_or(&TimeZone::UTC).clone())))
+                .cast(&DataType::Datetime(
+                    *t,
+                    Some(tz.unwrap_or(&TimeZone::UTC).clone()),
+                ))
                 .unwrap(),
                 AnyValue::DatetimeOwned(_, t, tz) => {
                     let tz = tz.as_ref().map(|x| x.deref().clone());
@@ -417,7 +420,10 @@ pub fn polars_literal_values_to_series(literal_values: Vec<LiteralValue>, name:
                             })
                             .collect::<Vec<Option<i64>>>(),
                     )
-                    .cast(&DataType::Datetime(*t, Some(tz.unwrap_or(TimeZone::UTC).clone())))
+                    .cast(&DataType::Datetime(
+                        *t,
+                        Some(tz.unwrap_or(TimeZone::UTC).clone()),
+                    ))
                     .unwrap()
                 }
                 AnyValue::Date(_) => Series::new(