Fix reindex command and some es utils

corehq/apps/es/client.py

@@ -399,7 +399,8 @@ def _validate_single_index(index):
 
     def reindex(
             self, source, dest, wait_for_completion=False,
-            refresh=False, batch_size=1000, requests_per_second=None, copy_doc_ids=True, query=None,
+            refresh=False, batch_size=1000, requests_per_second=None,
+            copy_doc_ids=True, query=None, purge_ids=False
     ):
         """
         Starts the reindex process in elastic search cluster
@@ -416,6 +417,7 @@ def reindex(
                            and can be reduced if you encounter scroll timeouts.
         :param query: ``dict`` optional parameter to include a term query to filter which documents are included in
                       the reindex
+        :param purge_ids: ``bool`` if True, will remove the _id field from the documents
         :returns: None if wait_for_completion is True else would return task_id of reindex task
         """
 
@@ -435,27 +437,35 @@ def reindex(
             "conflicts": "proceed"
         }
 
-    # Should be removed after ES 5-6 migration
-        if copy_doc_ids and source == const.HQ_USERS_INDEX_NAME:
-            # Remove password from form index
+        if copy_doc_ids or purge_ids:
             reindex_body["script"] = {
                 "lang": "painless",
-                "source": """
-                ctx._source.remove('password');
-                if (!ctx._source.containsKey('doc_id')) {
-                    ctx._source['doc_id'] = ctx._id;
-                }
-                """
-            }
-        elif copy_doc_ids:
-            reindex_body["script"] = {
-                "lang": "painless",
-                "source": """
-                if (!ctx._source.containsKey('doc_id')) {
-                    ctx._source['doc_id'] = ctx._id;
-                }
-                """
+                "source": ""
             }
+            script_parts = []
+
+            if purge_ids:
+                script_parts.append("""
+                    if (ctx._source.containsKey('_id')) {
+                        ctx._source.remove('_id');
+                    }
+                """)
+
+            if source == const.HQ_USERS_INDEX_NAME:
+                # Remove password field from users index
+                script_parts.append("""
+                    ctx._source.remove('password');
+                """)
+
+            if copy_doc_ids:
+                # Add doc_id field to the documents
+                script_parts.append("""
+                    if (!ctx._source.containsKey('doc_id')) {
+                        ctx._source['doc_id'] = ctx._id;
+                    }
+                """)
+
+            reindex_body["script"]["source"] = " ".join(script_parts)
 
         reindex_kwargs = {
             "wait_for_completion": wait_for_completion,

corehq/apps/es/management/commands/elastic_sync_multiplexed.py

@@ -42,7 +42,7 @@ class ESSyncUtil:
     def __init__(self):
         self.es = get_client()
 
-    def start_reindex(self, cname, reindex_batch_size=1000, requests_per_second=None):
+    def start_reindex(self, cname, reindex_batch_size=1000, requests_per_second=None, purge_ids=False):
 
         adapter = doc_adapter_from_cname(cname)
 
@@ -58,7 +58,7 @@ def start_reindex(self, cname, reindex_batch_size=1000, requests_per_second=None
         logger.info("Starting ReIndex process")
         task_id = es_manager.reindex(
             source_index, destination_index,
-            requests_per_second=requests_per_second, batch_size=reindex_batch_size
+            requests_per_second=requests_per_second, batch_size=reindex_batch_size, purge_ids=purge_ids
         )
         logger.info(f"Copying docs from index {source_index} to index {destination_index}")
         task_number = task_id.split(':')[1]