[Fixes #13478] Enhancement / bugfix of upsert functionality (#13482)

* [Fixes #13478] improvements on upsert
* [Fixes #13478] add bulk update and insert in upsert
* [Fixes #13478] add auto multi promotion for upsert

Author: mattiagiupponi

.env_dev

@@ -210,3 +210,4 @@ RESTART_POLICY_WINDOW=120s
 
 DEFAULT_MAX_PARALLEL_UPLOADS_PER_USER=5
 IMPORTER_ENABLE_DYN_MODELS=True
+UPSERT_CHUNK_SIZE= 10

geonode/settings.py

@@ -2319,5 +2319,8 @@ def get_geonode_catalogue_service():
 
 # Group default logo url
 GROUP_LOGO_URL = os.getenv("GROUP_LOGO_URL", "/geonode/img/group_logo.png")
+
+UPSERT_CHUNK_SIZE = ast.literal_eval(os.getenv("UPSERT_CHUNK_SIZE", "1000"))
+
 FILE_UPLOAD_DIRECTORY_PERMISSIONS = 0o777
 FILE_UPLOAD_PERMISSIONS = 0o777

geonode/upload/celery_tasks.py

@@ -841,10 +841,13 @@ def upsert_data(self, execution_id, /, handler_module_path, action, **kwargs):
         if not is_valid:
             raise UpsertException(errors)
 
-        result = _datastore.upsert_data(execution_id, **kwargs)
+        upsert_success, result = _datastore.upsert_data(execution_id, **kwargs)
 
         orchestrator.update_execution_request_obj(_exec, {"output_params": {"upsert": result}})
 
+        if not upsert_success:
+            raise UpsertException("Upsert has failed, please verify the log for more information")
+
         resource = ResourceBase.objects.get(pk=_exec.input_params.get("resource_pk"))
 
         task_params = (

geonode/upload/handlers/common/vector.py

@@ -17,6 +17,8 @@
 #
 #########################################################################
 import ast
+from datetime import datetime
+from itertools import islice
 from django.db import connections
 from geonode.security.permissions import _to_compact_perms
 from geonode.storage.manager import StorageManager
@@ -54,6 +56,7 @@
 from geonode.upload.models import ResourceHandlerInfo
 from geonode.upload.orchestrator import orchestrator
 from django.db.models import Q
+
 import pyproj
 from geonode.geoserver.security import delete_dataset_cache, set_geowebcache_invalidate_cache
 from geonode.geoserver.helpers import get_time_info
@@ -600,6 +603,7 @@ def create_dynamic_model_fields(
                         self.promote_to_multi(ogr.GeometryTypeToName(layer.GetGeomType()))
                     ),
                     "dim": (2 if not ogr.GeometryTypeToName(layer.GetGeomType()).lower().startswith("3d") else 3),
+                    "authority": self.identify_authority(layer),
                 }
             ]
 
@@ -627,6 +631,14 @@ def promote_to_multi(self, geometry_name: str):
         """
         return geometry_name
 
+    def promote_geom_to_multi(self, geom):
+        """
+        Convert the GetGeometryType object into Multi
+        example if is Point -> MultiPoint
+        Needed for the shapefiles
+        """
+        return geom
+
     def create_geonode_resource(
         self,
         layer_name: str,
@@ -963,19 +975,28 @@ def upsert_validation(self, files, execution_id, **kwargs: dict) -> Tuple[bool,
                 raise UpsertException(
                     f"The columns in the source and target do not match they must be equal. The following are not expected or missing: {differeces}"
                 )
+            skip_geom_eval = False
             for field in new_file_schema_fields:
                 # check if the field exists in the previous schema
                 target_field = target_schema_fields.filter(name=field["name"]).first()
                 if target_field:
                     # if is the primary key, we can skip the check
                     # If the field exists the class name should be the same
+                    if "authority" in field and not skip_geom_eval:
+                        if db_value := target_field.model_schema.as_model().objects.first():
+                            skip_geom_eval = True
+                            if not str(db_value.geometry.srid) in field["authority"]:
+                                message = f"The file provided have a different authority ({field['authority']}) compared to the one in the DB: {db_value}"
+                                raise UpsertException(message)
+
                     if not target_field.class_name == field["class_name"] and not target_field.kwargs.get(
                         "primary_key"
                     ):
                         # if the class changes, is marked as error
                         message = f"The type of the following field is changed and is prohibited: field: {field['name']}| current: {target_field.class_name}| new: {field['class_name']}"
                         errors.append(message)
                         logger.error(message)
+
             return not errors, errors
         else:
             raise UpsertException(
@@ -998,7 +1019,19 @@ def __get_new_and_original_schema(self, files, execution_id):
         layers = self._select_valid_layers(all_layers)
         if not layers:
             raise UpsertException("No valid layers found in the provided file for upsert.")
+
         layer = layers[0]
+        # evaluate if some of the ogc_fid entry is null. if is null we stop the workflow
+        # the user should provide the completed list with the ogc_fid set
+        sql_query = f'SELECT * FROM "{layer.GetName()}" WHERE "ogc_fid" IS NULL'
+
+        # Execute the SQL query to the layer
+        result = all_layers.ExecuteSQL(sql_query)
+        if not result or (result and result.GetFeatureCount() > 0):
+            raise UpsertException(
+                f"All the feature in the file must have the ogc_fid field correctly populated. Number of None value: {result.GetFeatureCount() if result else 'all'}"
+            )
+
         # Will generate the same schema as the target_resource_schema
         new_file_schema_fields = self.create_dynamic_model_fields(
             layer,
@@ -1048,39 +1081,39 @@ def upsert_data(self, files, execution_id, **kwargs):
         if not layers:
             raise UpsertException("No valid layers were found in the file provided")
         # we can upsert just 1 layer at time
-        for feature in layers[0]:
-            feature_as_dict = feature.items()
-            filter_dict = {field: value for field, value in feature_as_dict.items() if field == upsert_key}
-            if not filter_dict:
-                # if is not possible to extract the feature from ogr2ogr, better to ignore the layer
-                logger.error("was not possible to extract the feature of the dataset, skipping...")
-                update_error.append(f"Was not possible to manage the following data: {feature}")
-                continue
-            to_update = OriginalResource.objects.filter(**filter_dict)
-            geom = feature.GetGeometryRef()
-            feature_as_dict.update({self.default_geometry_column_name: geom.ExportToWkt()})
-            if to_update:
-                try:
-                    to_update.update(**feature_as_dict)
-                    valid_update += 1
-                except Exception as e:
-                    logger.error(f"Error during update of {feature_as_dict} in upsert: {e}")
-                    update_error.append(str(e))
-            else:
-                try:
-                    OriginalResource.objects.create(**feature_as_dict)
-                    valid_create += 1
-                except Exception as e:
-                    logger.error(f"Error during creation of {feature_as_dict} in upsert: {e}")
-                    create_error.append(str(e))
+        schema_fields = [f.name for f in model.fields.filter(kwargs__primary_key__isnull=True)]
+        total_feature = layers[0].GetFeatureCount()
+        layer_iterator = iter(layers[0])
+        while True:
+            # Create an iterator for the next chunk
+            data_chunk = list(islice(layer_iterator, settings.UPSERT_CHUNK_SIZE))
+
+            # If the chunk is empty, we've reached the end of the layer
+            if not data_chunk:
+                break
+
+            valid_create, valid_update, update_error, create_error = self._process_feature(
+                data_chunk,
+                model_instance=OriginalResource,
+                upsert_key=upsert_key,
+                update_error=update_error,
+                create_error=create_error,
+                valid_update=valid_update,
+                valid_create=valid_create,
+                schema_fields=schema_fields,
+            )
 
         # generating the resorucehandler infor to track the changes on ther resource
 
         self.create_resourcehandlerinfo(
             handler_module_path=str(self), resource=original_resource, execution_id=exec_obj
         )
 
+        if (total_feature - len(update_error) - len(create_error)) == 0:
+            raise UpsertException("All the entries provided raised error, execution is going to be stopped")
+
         return not update_error and not create_error, {
+            "success": not update_error and not create_error,
             "errors": {"create": create_error, "update": update_error},
             "data": {
                 "total": {
@@ -1092,6 +1125,61 @@ def upsert_data(self, files, execution_id, **kwargs):
             },
         }
 
+    def _process_feature(
+        self,
+        data_chunk,
+        model_instance,
+        upsert_key,
+        update_error,
+        create_error,
+        valid_update,
+        valid_create,
+        schema_fields,
+    ):
+        # getting all the upsert_key value from the data chunk
+        # retrieving the data from the DB
+        value_in_db = model_instance.objects.filter(
+            **{f"{upsert_key}__in": (getattr(feature, upsert_key) for feature in data_chunk)}
+        ).in_bulk(field_name=upsert_key)
+        update_bulk = []
+        create_bulk = []
+        # looping over the chunk data
+        for feature in data_chunk:
+            feature_as_dict = feature.items()
+            # need to simulate the "promote to multi" used by the upload process.
+            # here we cannot rely on ogr2ogr so we need to do it manually
+            geom = feature.GetGeometryRef()
+            feature_as_dict.update({self.default_geometry_column_name: self.promote_geom_to_multi(geom).ExportToWkt()})
+            if feature_as_dict.get(upsert_key) in value_in_db:
+                # if the key is present, we need to update the object
+                # the geometry must be treated manually
+                obj = value_in_db[feature_as_dict.get(upsert_key)]
+                for key, value in feature_as_dict.items():
+                    setattr(obj, key, value)
+                update_bulk.append(obj)
+            else:
+                # if the key is not present, we can create a new instance
+                create_bulk.append(model_instance(**feature_as_dict))
+
+        if update_bulk:
+            try:
+                # bulk update of the value. the schema is needed since is possible in django
+                # to update only specific columns of the schema. By default is updating all of them
+                model_instance.objects.bulk_update(update_bulk, fields=schema_fields)
+                valid_update += len(update_bulk)
+            except Exception as e:
+                logger.error("Error during update of feature in upsert")
+                update_error.append(str(e))
+
+        if create_bulk:
+            try:
+                model_instance.objects.bulk_create(create_bulk)
+                valid_create += len(update_bulk)
+            except Exception as e:
+                logger.error("Error during create of feature in upsert")
+                create_error.append(str(e))
+        return valid_create, valid_update, update_error, create_error
+
     def extract_upsert_key(self, exec_obj, dynamic_model_instance):
         # first we check if the upsert key is passed by the call
         key = exec_obj.input_params.get("upsert_key")
@@ -1120,11 +1208,18 @@ def refresh_geonode_resource(self, execution_id, asset=None, dataset=None, creat
                 resource=dataset, files=exec_obj.input_params["files"], action=exec_obj.action
             )
             # but we need to delete the previous one associated to the resource
-
+        start = datetime.now()
         delete_dataset_cache(dataset.alternate)
+        logging.debug(f"DATASET DELETE CACHE DONE {datetime.now() - start}")
+
         # recalculate featuretype info
+        start = datetime.now()
         DataPublisher(str(self)).recalculate_geoserver_featuretype(dataset)
+        logging.debug(f"recalculate_geoserver_featuretype DONE {datetime.now() - start}")
+
+        start = datetime.now()
         set_geowebcache_invalidate_cache(dataset_alternate=dataset.alternate)
+        logging.debug(f"set_geowebcache_invalidate_cache DONE {datetime.now() - start}")
 
         dataset = resource_manager.update(dataset.uuid, instance=dataset)
 

geonode/upload/handlers/shapefile/handler.py

@@ -210,7 +210,30 @@ def promote_to_multi(self, geometry_name):
         If needed change the name of the geometry, by promoting it to Multi
         example if is Point -> MultiPoint
         Needed for the shapefiles
+        Later this is used to map the geometry coming from ogr2ogr with a django class
         """
         if "Multi" not in geometry_name and "Point" not in geometry_name:
             return f"Multi {geometry_name.title()}"
         return geometry_name
+
+    def promote_geom_to_multi(self, geom):
+        """
+        Convert the GetGeometryType object into Multi
+        example if is Point -> MultiPoint
+        Needed for the shapefiles
+        """
+        match geom.GetGeometryType():
+            case ogr.wkbMultiLineString | ogr.wkbMultiPolygon:
+                # if is already multi, we dont need to do anything
+                return geom
+            case ogr.wkbLineString:
+                new_multi_geom = ogr.Geometry(ogr.wkbMultiLineString)
+                new_multi_geom.AddGeometry(geom)
+                return new_multi_geom
+            case ogr.wkbPolygon:
+                new_multi_geom = ogr.Geometry(ogr.wkbMultiPolygon)
+                new_multi_geom.AddGeometry(geom)
+                return new_multi_geom
+            case _:
+                # we dont convert points
+                return geom

geonode/upload/tests/end2end/test_env2end_upsert.py

@@ -54,16 +54,10 @@ def setUpClass(cls) -> None:
         super().setUpClass()
         cls.user = get_user_model().objects.exclude(username="Anonymous").first()
         cls.original = {
-            "base_file": f"{project_dir}/tests/fixture/upsert/original.shp",
-            "dbf_file": f"{project_dir}/tests/fixture/upsert/original.dbf",
-            "prj_file": f"{project_dir}/tests/fixture/upsert/original.prj",
-            "shx_file": f"{project_dir}/tests/fixture/upsert/original.shx",
+            "base_file": f"{project_dir}/tests/fixture/upsert/original.json",
         }
-        cls.upsert_shp = {
-            "base_file": f"{project_dir}/tests/fixture/upsert/upsert.shp",
-            "dbf_file": f"{project_dir}/tests/fixture/upsert/upsert.dbf",
-            "prj_file": f"{project_dir}/tests/fixture/upsert/upsert.prj",
-            "shx_file": f"{project_dir}/tests/fixture/upsert/upsert.shx",
+        cls.upsert_geojson = {
+            "base_file": f"{project_dir}/tests/fixture/upsert/upsert.json",
         }
         file_path = gisdata.VECTOR_DATA
         filename = os.path.join(file_path, "san_andres_y_providencia_natural.shp")
@@ -191,16 +185,15 @@ class ImporterShapefileImportTestUpsert(BaseImporterEndToEndTest):
         GEODATABASE_URL=f"{geourl.split('/geonode_data')[0]}/test_geonode_data", IMPORTER_ENABLE_DYN_MODELS=True
     )
     def test_import_shapefile_upsert(self):
-
         self._cleanup_layers(name="original")
         payload = {_filename: open(_file, "rb") for _filename, _file in self.original.items()}
         payload["action"] = "upload"
         initial_name = "original"
         prev_dataset = self._assertimport(payload, initial_name, keep_resource=True)
-        payload = {_filename: open(_file, "rb") for _filename, _file in self.upsert_shp.items()}
+        payload = {_filename: open(_file, "rb") for _filename, _file in self.upsert_geojson.items()}
         payload["resource_pk"] = prev_dataset.pk
         payload["action"] = "upsert"
-        payload["upsert_key"] = "unique"
+        payload["upsert_key"] = "ogc_fid"
 
         # time to upsert the data
         self.client.force_login(self.admin)
@@ -210,9 +203,10 @@ def test_import_shapefile_upsert(self):
         self.assertEqual(response.status_code, 201)
         exec_obj = ExecutionRequest.objects.get(exec_id=response.json().get("execution_id"))
         output_input = exec_obj.output_params
-        self.assertTrue(output_input.get("upsert")[0])
-        data = output_input.get("upsert")[1]
+        self.assertTrue(output_input.get("upsert", {}).get("success"))
+        data = output_input.get("upsert")
         expected = {
+            "success": True,
             "data": {
                 "error": {"create": 0, "update": 0},
                 "total": {"error": 0, "success": 2},
@@ -228,7 +222,7 @@ def test_import_shapefile_upsert(self):
         self.assertIsNotNone(key)
         self.assertTrue(len(key) == 1)
         # evaluate if the dynamic model is correctly upserted, we expect 2 rows
-        self.assertEqual(schema.as_model().objects.count(), 2)
+        self.assertEqual(schema.as_model().objects.count(), 3)
 
     @mock.patch.dict(os.environ, {"GEONODE_GEODATABASE": "test_geonode_data", "IMPORTER_ENABLE_DYN_MODELS": "True"})
     @override_settings(
@@ -244,7 +238,7 @@ def test_import_shapefile_upsert_raise_error_with_different_schemas(self):
         payload = {_filename: open(_file, "rb") for _filename, _file in self.default_shp.items()}
         payload["resource_pk"] = prev_dataset.pk
         payload["action"] = "upsert"
-        payload["upsert_key"] = "unique"
+        payload["upsert_key"] = "ogc_fid"
 
         # time to upsert the data
         self.client.force_login(self.admin)
@@ -255,6 +249,6 @@ def test_import_shapefile_upsert_raise_error_with_different_schemas(self):
         self.assertFalse(response.json().get("success", True))
         self.assertTrue("errors" in response.json())
         self.assertTrue(
-            "The columns in the source and target do not match they must be equal. The following are not expected or missing"
+            "All the feature in the file must have the ogc_fid field correctly populated. Number of None value: all"
             in response.json()["errors"][0]
         )

geonode/upload/tests/fixture/upsert/original.dbf

@@ -0,0 +1,25 @@
+{
+  "type": "FeatureCollection",
+  "features": [
+    {
+      "type": "Feature",
+      "geometry": {
+        "type": "Point",
+        "coordinates": [
+          -47.20138110814478,
+          -13.8281037379215
+        ]
+      },
+      "properties": {
+        "ogc_fid": 999,
+        "st_area_shape": 192656942.39615256,
+        "st_length_shape": 117499.50756699985,
+        "ra_cira": 9,
+        "ra_nome": "THIS IS A CREATE",
+        "ra_codigo": "THIS IS A CREATE",
+        "ra_path": "http://localhost:8080",
+        "ra_areakm2": 192.65694240000002
+      }
+    }
+  ]
+}