feat(dev) migration, int _id;

- Made _id field an integer, new unique field and code to update and re-index.
- Added CLI commands and util script for reusable code.

Author: JVickery-TBS

ckanext/datastore_search/backend/solr.py

@@ -8,9 +8,10 @@
 import hashlib
 import tempfile
 import math
+from datetime import datetime
 
-from typing import Any, Optional, Dict, cast, List
-from ckan.types import Context, DataDict
+from typing import Any, Optional, Dict, List
+from ckan.types import DataDict
 
 from ckan.plugins.toolkit import _, config, get_action, enqueue_job
 from ckan.lib.jobs import add_queue_name_prefix
@@ -19,6 +20,10 @@
 from ckanext.datastore.backend.postgres import identifier
 from ckanext.datastore.blueprint import dump_to
 
+from ckanext.datastore_search.utils import (
+    get_site_context,
+    get_datastore_create_dict
+)
 from ckanext.datastore_search.backend import (
     DatastoreSearchBackend,
     DatastoreSearchException
@@ -39,7 +44,7 @@ class DatastoreSolrBackend(DatastoreSearchBackend):
     SOLR class for datastore search backend.
     """
     timeout = config.get('solr_timeout')
-    default_search_fields = ['_id', '_version_', 'indexed_ts', '_text_']
+    default_search_fields = ['_id', '_version_', 'indexed_ts', 'index_id', '_text_']
     configset_name = config.get('ckanext.datastore_search.solr.configset',
                                 'datastore_resource')
 
@@ -58,13 +63,20 @@ def field_type_map(self):
             'smallint': 'int',
             'integer': 'int',
             'bigint': 'int',
+            'int2': 'int',
+            'int4': 'int',
+            'int8': 'int',
+            'int16': 'int',
             'decimal': 'float',
             'numeric': 'float',
             'real': 'double',
             'double precision': 'double',
             'smallserial': 'int',
             'serial': 'int',
             'bigserial': 'int',
+            'serial4': 'int',
+            'serial8': 'int',
+            'serial16': 'int',
             # monetary types
             'money': 'float',
             # char types
@@ -125,13 +137,6 @@ def _send_api_request(self,
                                 timeout=self.timeout)
         return resp.json()
 
-    def _get_site_context(self) -> Context:
-        """
-        Return a CKAN Context for the system user.
-        """
-        site_user = get_action('get_site_user')({'ignore_auth': True}, {})
-        return cast(Context, {'user': site_user['name']})
-
     def reindex(self,
                 resource_id: Optional[str] = None,
                 connection: Optional[pysolr.Solr] = None,
@@ -143,7 +148,7 @@ def reindex(self,
             return
         # FIXME: put this in a background task as a larger
         #        DS Resource could take a long time??
-        context = self._get_site_context()
+        context = get_site_context()
         core_name = f'{self.prefix}{resource_id}'
         conn = self._make_connection(resource_id) if not connection else connection
 
@@ -188,12 +193,21 @@ def reindex(self,
             if not solr_records:
                 gathering_solr_records = False
             # type_ignore_reason: checking solr_records
-            indexed_ids += [r['_id'] for r in solr_records]  # type: ignore
+            indexed_ids += [str(r['_id']) for r in solr_records]  # type: ignore
             if self.only_use_engine:
                 # type_ignore_reason: checking solr_records
                 indexed_records += solr_records  # type: ignore
             offset += 1000
         core_name = f'{self.prefix}{resource_id}'
+
+        # clear index
+        if not only_missing:
+            log.debug('Emptying SOLR index for DataStore Resource %s' %
+                      resource_id)
+            conn.delete(q='*:*', commit=False)
+            log.debug('Unindexed all DataStore records for DataStore Resource %s' %
+                      resource_id)
+
         errmsg = _('Failed to reindex records for %s' % core_name)
         if not self.only_use_engine:
             gathering_ds_records = True
@@ -203,7 +217,6 @@ def reindex(self,
                 indexed_ids=','.join(indexed_ids)) if \
                 only_missing and indexed_ids and \
                 int(solr_total) <= int(ds_total) else ''
-            existing_ids = []
             while gathering_ds_records:
                 sql_string = '''
                     SELECT {columns} FROM {table} {where_statement}
@@ -217,7 +230,6 @@ def reindex(self,
                 if not ds_result['records']:
                     gathering_ds_records = False
                 for r in ds_result['records']:
-                    existing_ids.append(str(r['_id']))
                     if only_missing and indexed_ids and str(r['_id']) in indexed_ids:
                         continue
                     try:
@@ -230,17 +242,6 @@ def reindex(self,
                         raise DatastoreSearchException(
                             errmsg if not DEBUG else e.args[0][:MAX_ERR_LEN])
                 offset += 1000
-            orphan_ids = set(indexed_ids) - set(existing_ids)
-            for orphan_id in orphan_ids:
-                try:
-                    conn.delete(q='_id:%s' % orphan_id, commit=False)
-                    if DEBUG:
-                        log.debug('Unindexed DataStore record '
-                                  '_id=%s for Resource %s' %
-                                  (orphan_id, resource_id))
-                except pysolr.SolrError as e:
-                    raise DatastoreSearchException(
-                        errmsg if not DEBUG else e.args[0][:MAX_ERR_LEN])
         else:
             for r in indexed_records:
                 try:
@@ -271,10 +272,10 @@ def _check_counts(self,
                 _('SOLR core does not exist for DataStore Resource %s') % resource_id)
 
         ds_result = get_action('datastore_search')(
-            self._get_site_context(), {'resource_id': resource_id,
-                                       'limit': 0,
-                                       'include_total': True,
-                                       'skip_search_engine': True})
+            get_site_context(), {'resource_id': resource_id,
+                                 'limit': 0,
+                                 'include_total': True,
+                                 'skip_search_engine': True})
         ds_total = ds_result['total']
         solr_result = conn.search(q='*:*', rows=0)
         solr_total = solr_result.hits
@@ -529,16 +530,7 @@ def create_callback(self, data_dict: DataDict) -> Any:
 
         resource_id = data_dict.get('core_name', '').replace(self.prefix, '')
         upload_file = data_dict.get('extras', {}).get('upload_file', False)
-        context = self._get_site_context()
-        ds_result = get_action('datastore_search')(
-            context, {'resource_id': resource_id,
-                      'limit': 0,
-                      'skip_search_engine': True})
-        create_dict = {
-            'resource_id': resource_id,
-            'fields': [f for f in ds_result['fields'] if
-                       f['id'] not in self.default_search_fields],
-            'upload_file': upload_file}
+        create_dict = get_datastore_create_dict(resource_id, upload_file)
         # call create again to get datastore fields and data types to build schema
         self.create(create_dict)
 
@@ -549,9 +541,10 @@ def create_callback(self, data_dict: DataDict) -> Any:
         ):
             # use datastore_upsert with insert to be able to use dry_run to get _id
             get_action('datastore_upsert')(
-                context, {'resource_id': resource_id,
-                          'records': data_dict.get('extras', {}).get('records'),
-                          'method': 'insert'})
+                get_site_context(),
+                {'resource_id': resource_id,
+                 'records': data_dict.get('extras', {}).get('records'),
+                 'method': 'insert'})
 
     def upsert(self,
                data_dict: DataDict,
@@ -569,8 +562,25 @@ def upsert(self,
 
         if data_dict['records']:
             for r in data_dict['records']:
+                # TODO: do any other value type transforms that might be needed...
+                for k, v in r.items():
+                    if isinstance(v, datetime):
+                        r[k] = v.isoformat() + 'Z'
+                solr_record = self.search({
+                    'resource_id': resource_id,
+                    'offset': 0,
+                    'limit': 1,
+                    'query': '_id:%s' % r['_id']},
+                    connection=conn)
+                _r = dict(r)
+                if solr_record:
+                    _r = dict(solr_record[0], **_r)
+                for f in self.default_search_fields:
+                    if f == '_id' or f == 'index_id':
+                        continue
+                    _r.pop(f, None)
                 try:
-                    conn.add(docs=[r], commit=False)
+                    conn.add(docs=[_r], commit=False)
                     if DEBUG:
                         log.debug('Indexed DataStore record _id=%s for Resource %s' %
                                   (r['_id'], resource_id))
@@ -627,6 +637,7 @@ def search(self,
             'q': q,
             'q.op': 'AND',
             'fq': fq,
+            'fl': data_dict.get('fl', None),
             'df': data_dict.get('df', '_text_'),
             'start': data_dict.get('offset', 0),
             'rows': data_dict.get('limit', 1000),

ckanext/datastore_search/cli.py

@@ -0,0 +1,88 @@
+import click
+
+from typing import Optional
+
+from ckan.plugins.toolkit import ObjectNotFound
+from ckanext.datastore_search.backend import (
+    DatastoreSearchBackend,
+    DatastoreSearchException
+)
+from ckanext.datastore_search.logic.action import SEARCH_INDEX_SKIP_MSG
+from ckanext.datastore_search.utils import (
+    get_datastore_count,
+    get_datastore_create_dict,
+    is_using_pusher,
+    get_datastore_tables
+)
+
+
+@click.group(short_help="DataStore Search commands")
+def datastore_search():
+    """
+    DataStore Search commands
+    """
+    pass
+
+
+@datastore_search.command(
+    short_help="Migrate DataStore Resources into Search Engine Indices.")
+@click.option(
+    '-r', '--resource-id',
+    required=False,
+    type=click.STRING,
+    default=None,
+    help='Resource ID to migrate, otherwise all resources.'
+)
+@click.option(
+    "-v",
+    "--verbose",
+    is_flag=True,
+    default=False,
+    help="Increase verbosity"
+)
+def migrate(resource_id: Optional[str] = None,
+            verbose: Optional[bool] = False):
+    """
+    Migrate DataStore Resources into Search Engine Indices.
+    """
+    backend = DatastoreSearchBackend.get_active_backend()
+    if not resource_id:
+        ds_resource_ids = get_datastore_tables()
+        if not ds_resource_ids:
+            click.echo("No DataStore Resources exist. Exiting...")
+            return
+        if verbose:
+            click.echo("Gathered %s table names from the DataStore." %
+                       len(ds_resource_ids))
+    else:
+        try:
+            get_datastore_count(resource_id)
+        except ObjectNotFound:
+            click.echo('Resource not found or not a DataStore resource: %s' %
+                       resource_id)
+        ds_resource_ids = [resource_id]
+    for ds_resource_id in ds_resource_ids:
+        if (
+            not backend.only_use_engine
+            and (
+                ds_count := get_datastore_count(ds_resource_id) <
+                backend.min_rows_for_index
+            )
+        ):
+            # do not try to create search index if not enough rows
+            if verbose:
+                click.echo(SEARCH_INDEX_SKIP_MSG %
+                           (ds_resource_id,
+                            ds_count,
+                            backend.min_rows_for_index))
+            continue
+        create_dict = get_datastore_create_dict(
+            ds_resource_id, upload_file=is_using_pusher(ds_resource_id))
+        try:
+            backend.create(create_dict)
+            if verbose:
+                click.echo('Migrated DataStore Resource %s into search index' %
+                           ds_resource_id)
+        except DatastoreSearchException:
+            pass
+    click.echo('Done!')

ckanext/datastore_search/config/solr/managed-schema.xml

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!-- Solr managed schema - automatically generated - DO NOT EDIT -->
 <schema name="ckan-datastore" version="1.0">
-  <uniqueKey>_id</uniqueKey>
+  <uniqueKey>index_id</uniqueKey>
   <fieldType name="binary" class="solr.BinaryField"/>
   <fieldType name="boolean" class="solr.BoolField" omitNorms="true" sortMissingLast="true"/>
   <fieldType name="booleans" class="solr.BoolField" sortMissingLast="true" multiValued="true"/>
@@ -65,7 +65,8 @@
   </fieldType>
   <field name="_version_" type="string" multiValued="false" indexed="true" stored="true"/>
   <field name="indexed_ts" type="date" default="NOW" multiValued="false" indexed="true" stored="true"/>
-  <field name="_id" type="string" multiValued="false" indexed="true" required="true" stored="true"/>
+  <field name="index_id" type="string" multiValued="false" indexed="true" required="true" stored="true"/>
+  <field name="_id" type="int" multiValued="false" indexed="true" required="true" stored="true"/>
   <field name="_text_" type="text" multiValued="true" indexed="true" stored="false"/>
   <copyField source="*" dest="_text_"/>
 </schema>