retrieve field index only once

Author: ChristianBeilschmidt

operators/src/source/ogr_source/mod.rs

@@ -5,8 +5,8 @@ use crate::engine::{
     CanonicOperatorName, OperatorData, OperatorName, QueryProcessor, WorkflowOperatorPath,
 };
 use crate::error::Error;
-use crate::util::Result;
 use crate::util::input::StringOrNumberRange;
+use crate::util::{Result, safe_lock_mutex};
 use crate::{
     engine::{
         InitializedVectorOperator, MetaData, QueryContext, SourceOperator,
@@ -560,7 +560,7 @@ where
     }
 }
 
-type TimeExtractorType = Box<dyn Fn(&Feature) -> Result<TimeInterval> + Send + Sync + 'static>;
+type TimeExtractorType = Box<dyn FnMut(&Feature) -> Result<TimeInterval> + Send + Sync + 'static>;
 
 #[pin_project(project = OgrSourceStreamProjection)]
 pub struct OgrSourceStream<G>
@@ -571,7 +571,7 @@ where
     dataset_iterator: Arc<Mutex<OgrDatasetIterator>>,
     data_types: Arc<HashMap<String, FeatureDataType>>,
     feature_collection_builder: FeatureCollectionBuilder<G>,
-    time_extractor: Arc<TimeExtractorType>,
+    time_extractor: Arc<std::sync::Mutex<TimeExtractorType>>,
     time_attribute_parser:
         Arc<Box<dyn Fn(FieldValue) -> Result<TimeInstance> + Send + Sync + 'static>>,
     query_rectangle: VectorQueryRectangle,
@@ -849,7 +849,7 @@ where
                 data_types: Arc::new(data_types),
                 feature_collection_builder,
                 query_rectangle,
-                time_extractor: Arc::new(time_extractor),
+                time_extractor: Arc::new(std::sync::Mutex::new(time_extractor)),
                 time_attribute_parser: Arc::new(time_attribute_parser),
                 chunk_byte_size,
                 future: None,
@@ -867,7 +867,7 @@ where
         feature_collection_builder: FeatureCollectionBuilder<G>,
         data_types: Arc<HashMap<String, FeatureDataType>>,
         query_rectangle: VectorQueryRectangle,
-        time_extractor: Arc<TimeExtractorType>,
+        time_extractor: Arc<std::sync::Mutex<TimeExtractorType>>,
         time_attribute_parser: Arc<Box<dyn Fn(FieldValue) -> Result<TimeInstance> + Send + Sync>>,
         chunk_byte_size: usize,
     ) -> Result<FeatureCollection<G>> {
@@ -880,7 +880,7 @@ where
                 &dataset_information,
                 &data_types,
                 &query_rectangle,
-                time_extractor.as_ref(),
+                safe_lock_mutex(&time_extractor).as_mut(),
                 time_attribute_parser.as_ref(),
                 chunk_byte_size,
             );
@@ -975,10 +975,12 @@ where
             } => {
                 let time_start_parser = Self::create_time_parser(start_format);
 
+                let mut field_index = None;
                 Box::new(move |feature: &Feature| {
-                    let field_value = feature
-                        .field_index(&start_field)
-                        .and_then(|i| feature.field(i))?;
+                    let field_index =
+                        get_or_insert_field_index(&mut field_index, feature, &start_field)?;
+
+                    let field_value = feature.field(field_index)?;
                     if let Some(field_value) = field_value {
                         let time_start = time_start_parser(field_value)?;
                         TimeInterval::new(time_start, (time_start + duration)?).map_err(Into::into)
@@ -997,13 +999,17 @@ where
                 let time_start_parser = Self::create_time_parser(start_format);
                 let time_end_parser = Self::create_time_parser(end_format);
 
+                let mut start_field_index = None;
+                let mut end_field_index = None;
+
                 Box::new(move |feature: &Feature| {
-                    let start_field_value = feature
-                        .field_index(&start_field)
-                        .and_then(|i| feature.field(i))?;
-                    let end_field_value = feature
-                        .field_index(&end_field)
-                        .and_then(|i| feature.field(i))?;
+                    let start_field_index =
+                        get_or_insert_field_index(&mut start_field_index, feature, &start_field)?;
+                    let end_field_index =
+                        get_or_insert_field_index(&mut end_field_index, feature, &end_field)?;
+
+                    let start_field_value = feature.field(start_field_index)?;
+                    let end_field_value = feature.field(end_field_index)?;
 
                     if let (Some(start_field_value), Some(end_field_value)) =
                         (start_field_value, end_field_value)
@@ -1025,13 +1031,20 @@ where
             } => {
                 let time_start_parser = Self::create_time_parser(start_format);
 
+                let mut start_field_index = None;
+                let mut duration_field_index = None;
+
                 Box::new(move |feature: &Feature| {
-                    let start_field_value = feature
-                        .field_index(&start_field)
-                        .and_then(|i| feature.field(i))?;
-                    let duration_field_value = feature
-                        .field_index(&duration_field)
-                        .and_then(|i| feature.field(i))?;
+                    let start_field_index =
+                        get_or_insert_field_index(&mut start_field_index, feature, &start_field)?;
+                    let duration_field_index = get_or_insert_field_index(
+                        &mut duration_field_index,
+                        feature,
+                        &duration_field,
+                    )?;
+
+                    let start_field_value = feature.field(start_field_index)?;
+                    let duration_field_value = feature.field(duration_field_index)?;
 
                     if let (Some(start_field_value), Some(duration_field_value)) =
                         (start_field_value, duration_field_value)
@@ -1110,7 +1123,7 @@ where
         dataset_information: &OgrSourceDataset,
         data_types: &HashMap<String, FeatureDataType>,
         query_rectangle: &VectorQueryRectangle,
-        time_extractor: &dyn Fn(&Feature) -> Result<TimeInterval>,
+        time_extractor: &mut dyn FnMut(&Feature) -> Result<TimeInterval>,
         time_attribute_parser: &dyn Fn(FieldValue) -> Result<TimeInstance>,
         chunk_byte_size: usize,
     ) -> Result<FeatureCollection<G>> {
@@ -1265,7 +1278,7 @@ where
         default_geometry: &Option<G>,
         data_types: &HashMap<String, FeatureDataType>,
         query_rectangle: &VectorQueryRectangle,
-        time_extractor: &dyn Fn(&Feature) -> Result<TimeInterval, Error>,
+        time_extractor: &mut dyn FnMut(&Feature) -> Result<TimeInterval, Error>,
         time_attribute_parser: &dyn Fn(FieldValue) -> Result<TimeInstance>,
         builder: &mut FeatureCollectionRowBuilder<G>,
         feature: &Feature,
@@ -1306,8 +1319,14 @@ where
         builder.push_generic_geometry(geometry);
         builder.push_time_interval(time_interval);
 
+        let mut field_indices = HashMap::with_capacity(data_types.len());
+
         for (column, data_type) in data_types {
-            let field = feature.field_index(column).and_then(|i| feature.field(i));
+            let field_index = field_indices
+                .entry(column.as_str())
+                .or_insert_with(|| feature.field_index(column));
+
+            let field = field_index.clone().and_then(|i| feature.field(i));
             let value =
                 Self::convert_field_value(*data_type, field, time_attribute_parser, error_spec)?;
             builder.push_data(column, value)?;
@@ -1319,6 +1338,20 @@ where
     }
 }
 
+fn get_or_insert_field_index(
+    field_index: &mut Option<usize>,
+    feature: &Feature,
+    field_name: &str,
+) -> Result<usize> {
+    if let Some(i) = field_index {
+        return Ok(*i);
+    }
+
+    let i = feature.field_index(field_name)?;
+    *field_index = Some(i);
+    Ok(i)
+}
+
 impl<G> Stream for OgrSourceStream<G>
 where
     G: Geometry + ArrowTyped + 'static + std::marker::Unpin + TryFromOgrGeometry,