Batch the per-feature database round trips in feature import

app/models/abstract_feature.rb

@@ -11,6 +11,12 @@ class AbstractFeature < ActiveRecord::Base
 
   attr_writer :make_valid
 
+  # Set by `::precompute_geometry_validation`. Nil means the geometry is valid.
+  def precomputed_geometry_validation=(message)
+    @geometry_validation_precomputed = true
+    @precomputed_geometry_validation = message
+  end
+
   FEATURE_TYPES = %w(polygon point line)
 
   validates_presence_of :geog
@@ -136,6 +142,32 @@ def envelope(buffer_in_meters = 0)
     return envelope_json.values_at(0,2)
   end
 
+  # Reads the geometry validation message for many records in one query, so validating a
+  # batch costs one round trip rather than one per record. Each record holds its own
+  # answer afterwards and `valid?` reads it instead of querying.
+  #
+  # Records whose geometry is later replaced (a repair) discard the answer and fall back
+  # to querying, so a batch stays correct when some of its members change.
+  #
+  # @param records [Array<AbstractFeature>] records with `geog` assigned.
+  # @return [void]
+  def self.precompute_geometry_validation(records)
+    records = records.select {|record| record.geog.present? }
+    return if records.empty?
+
+    values = records.each_with_index.map {|record, index| "(#{index}, #{connection.quote(record.geog.to_s)})" }
+    messages = connection.select_rows(<<~SQL).to_h
+      SELECT t.i, ST_IsValidReason(x.geog)
+      FROM (VALUES #{values.join(',')}) AS t(i, wkt),
+      LATERAL (SELECT t.wkt::geography::geometry AS geog) AS x
+      WHERE NOT ST_IsValid(x.geog)
+    SQL
+
+    records.each_with_index do |record, index|
+      record.precomputed_geometry_validation = messages[index]
+    end
+  end
+
   def self.without_caching_derivatives(&block)
     old = automatically_cache_derivatives
     self.automatically_cache_derivatives = false
@@ -276,6 +308,7 @@ def make_valid?
   private
 
   def make_valid
+    @geometry_validation_precomputed = false # the geometry is about to change, so the batch's answer no longer applies
     self.geog = SpatialFeatures::Utils.select_db_value("SELECT ST_Buffer('#{sanitize}', 0)")
   end
 
@@ -307,6 +340,8 @@ def validate_geometry
   end
 
   def geometry_validation_message
+    return @precomputed_geometry_validation if @geometry_validation_precomputed
+
     klass = self.class.base_class # Use the base class because we don't want to have to include a type column in our select
     error = klass.connection.select_one(klass.unscoped.invalid.from("(SELECT '#{sanitize_input_for_sql(self.geog)}'::geography::geometry AS geog) #{klass.table_name}")) # Ensure we cast to geography because the geog attribute value may not have been coerced to geography yet, so we want it to apply the +-180/90 bounds to any odd geometry that will happen when we save to the database
     return error.fetch('invalid_geometry_message') if error

lib/spatial_features/has_spatial_features/feature_import.rb

@@ -10,6 +10,11 @@ module FeatureImport
     # method failed. Any per-file reasons are appended after it.
     EMPTY_IMPORT_MESSAGE = "No mapped areas could be imported.".freeze
 
+    # How many features have their geometry validity read in one query. A batch is sent as a
+    # literal list of geometries, so the ceiling is the size of that statement rather than
+    # the record count, and a file of few but very large geometries reaches it first.
+    FEATURE_VALIDATION_BATCH_SIZE = 500
+
     included do
       extend ActiveModel::Callbacks
       define_model_callbacks :update_features
@@ -153,15 +158,7 @@ def import_features(imports, skip_invalid)
       features.delete_all
       valid, invalid = Feature.defer_aggregate_refresh do
         Feature.without_caching_derivatives do
-          imports.flat_map {|import| features_from(import) }.partition do |feature|
-            feature.spatial_model = self
-            if feature.save
-              handle_images(feature)
-              true
-            else
-              false
-            end
-          end
+          save_features(imports.flat_map {|import| features_from(import) })
         end
       end
 
@@ -190,6 +187,33 @@ def import_features(imports, skip_invalid)
     # Parse failures surface lazily, when an importer's features are first read (e.g. a
     # shapefile archive missing its `.shx`), so they need the same containment as a file
     # that couldn't be opened at all: record the reason against that source and keep going.
+    # Saves each feature, reading the geometry validity of a batch at a time rather than of
+    # one record at a time. Saving is unchanged and still runs every callback.
+    #
+    # @param new_features [Array<Feature>] unsaved features, in the order they were read.
+    # @return [Array(Array<Feature>, Array<Feature>)] those that saved and those that did
+    #   not, each in the order given.
+    def save_features(new_features)
+      saved = []
+      rejected = []
+
+      new_features.each_slice(FEATURE_VALIDATION_BATCH_SIZE) do |batch|
+        batch.each {|feature| feature.spatial_model = self }
+        Feature.precompute_geometry_validation(batch)
+
+        batch.each do |feature|
+          if feature.save
+            handle_images(feature)
+            saved << feature
+          else
+            rejected << feature
+          end
+        end
+      end
+
+      [saved, rejected]
+    end
+
     def features_from(import)
       import.features
     rescue ImportError => e

lib/spatial_features/importers/kml.rb

@@ -20,6 +20,11 @@ class KML < Base
       UNPLACED_GEOMETRY_XPATH =
         GEOMETRY_TYPES.map {|type| "//#{type}[not(ancestor::Placemark)]" }.join(' | ').freeze
 
+      # How many geometry elements are parsed in one query. A batch is sent as a literal list
+      # of KML fragments, so the ceiling is the size of that statement rather than the
+      # element count, and a file of few but very large geometries reaches it first.
+      GEOMETRY_BATCH_SIZE = 200
+
       # matches a coordinate pair with an optional altitude, including invalid altitudes like NaN
       #   -118.1,50.9,NaN
       #   -118.1,50.9,0
@@ -34,20 +39,24 @@ def initialize(data, base_dir: nil, **options)
       private
 
       def each_record(&block)
+        pending = []
+
         kml_document.css('Placemark').each do |placemark|
           metadata = extract_metadata(placemark)
           importable_image_paths = images_from_metadata(metadata)
           name = placemark.css('name').text
 
           geometries_in(placemark).each do |geometry|
             # A hash of its own per feature, since each is stored on a separate record.
-            yield_feature(geometry, name, metadata.dup, importable_image_paths, &block)
+            hold(pending, geometry, name, metadata.dup, importable_image_paths, &block)
           end
         end
 
         kml_document.xpath(UNPLACED_GEOMETRY_XPATH).each do |geometry|
-          yield_feature(geometry, nil, {}, [], &block)
+          hold(pending, geometry, nil, {}, [], &block)
         end
+
+        yield_batch(pending, &block)
       end
 
       # Returns the geometry elements belonging to a Placemark.
@@ -76,20 +85,64 @@ def nested_placemarks?
       end
 
       # Yields the feature built from a geometry element.
+
+      # Holds a geometry element until there are enough of them to parse in one query.
       #
       # @param geometry [Nokogiri::XML::Element] a Polygon, LineString or Point node.
       # @param metadata [Hash] stored on the feature as it stands, so it must already have
       #   had its image keys removed.
-      # @yield [OpenStruct] nothing is yielded when the element holds no coordinates, or
-      #   when PostGIS cannot read it.
-      def yield_feature(geometry, name, metadata, importable_image_paths, &block)
+      # @return [void] an element holding no coordinates is dropped rather than held.
+      def hold(pending, geometry, name, metadata, importable_image_paths, &block)
         return if blank_feature?(geometry)
 
-        geog = geom_from_kml(geometry)
-        return if geog.blank?
+        pending << [geometry, name, metadata, importable_image_paths]
+        yield_batch(pending, &block) if pending.size >= GEOMETRY_BATCH_SIZE
+      end
+
+      # Yields a feature for each held element, in the order they were held.
+      #
+      # @yield [OpenStruct] an element PostGIS cannot read yields nothing.
+      def yield_batch(pending, &block)
+        return if pending.empty?
+
+        geographies = geom_from_kml_batch(pending.map(&:first))
+
+        pending.each_with_index do |(_, name, metadata, importable_image_paths), index|
+          geog = geographies[index]
+          next if geog.blank?
+
+          block.call OpenStruct.new(geog: geog, name: name, metadata: metadata,
+                                    importable_image_paths: importable_image_paths)
+        end
+
+        pending.clear
+      end
+
+      # Parses many KML geometry elements in one query.
+      #
+      # @param geometries [Array<Nokogiri::XML::Element>]
+      # @return [Array<String, nil>] one geography per element, in the order given, nil
+      #   where the element could not be read. A batch holding an element PostGIS rejects
+      #   fails as a whole, so it is re-read one element at a time and only that element
+      #   is lost.
+      def geom_from_kml_batch(geometries)
+        return [geom_from_kml(geometries.first)] if geometries.one?
+
+        geometries.each {|geometry| strip_altitude(geometry) }
+        connection = ActiveRecord::Base.connection
+        values = geometries.each_with_index.map {|geometry, index| "(#{index}, #{connection.quote(geometry.to_s)})" }
+
+        rows = ActiveRecord::Base.transaction(requires_new: true) do
+          connection.select_rows(<<~SQL)
+            SELECT t.i, ST_GeomFromKML(t.kml)
+            FROM (VALUES #{values.join(',')}) AS t(i, kml)
+            ORDER BY t.i
+          SQL
+        end
 
-        block.call OpenStruct.new(geog: geog, name: name, metadata: metadata,
-                                  importable_image_paths: importable_image_paths)
+        rows.map(&:last)
+      rescue ActiveRecord::StatementInvalid
+        geometries.map {|geometry| geom_from_kml(geometry) }
       end
 
       def kml_document

spec/lib/spatial_features/importers/kml_spec.rb

@@ -176,6 +176,24 @@
     end
   end
 
+  context 'when a batch holds an element PostGIS cannot read' do
+    # A single-vertex LineString is not a geometry PostGIS will parse, and it sits beside
+    # two placemarks that are fine.
+    let(:data) { kml_file_with_invalid_placemark.read }
+
+    describe '#features' do
+      it 'returns the elements it could read' do
+        expect(subject.features.count).to eq(2)
+      end
+
+      it 'reads the batch again one element at a time' do
+        expect(subject).to receive(:geom_from_kml).exactly(3).times.and_call_original
+
+        subject.features
+      end
+    end
+  end
+
   context 'when the input is xml but not kml' do
     let(:data) { "<html><body>hi</body></html>" }
 

spec/models/feature_spec.rb

@@ -150,4 +150,37 @@
       expect { house.features.first.refresh_aggregate }.to change { house.reload.aggregate_feature.cache_key }
     end
   end
+  describe '::precompute_geometry_validation' do
+    let(:valid_geog) { 'POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))' }
+    let(:invalid_geog) { 'POLYGON((0 0, 2 2, 2 0, 0 2, 0 0))' } # a bowtie, which self-intersects
+
+    it 'gives each record the answer the per-record query gives' do
+      batched = [Feature.new(geog: valid_geog), Feature.new(geog: invalid_geog)]
+      Feature.precompute_geometry_validation(batched)
+
+      individually = [Feature.new(geog: valid_geog), Feature.new(geog: invalid_geog)]
+
+      expect(batched.map {|f| f.send(:geometry_validation_message) })
+        .to eq(individually.map {|f| f.send(:geometry_validation_message) })
+    end
+
+    it 'reads nothing further while the answer holds' do
+      feature = Feature.new(geog: valid_geog)
+      Feature.precompute_geometry_validation([feature])
+
+      expect(Feature.connection).not_to receive(:select_one)
+
+      expect(feature).to be_valid
+    end
+
+    it 'discards the answer once a repair changes the geometry' do
+      feature = Feature.new(geog: invalid_geog)
+      feature.make_valid = true
+      Feature.precompute_geometry_validation([feature])
+
+      expect(feature).to be_valid # repaired, so it is asked again rather than trusting the batch
+      expect(feature.geog).not_to eq(invalid_geog)
+    end
+  end
+
 end