feat: new populatedFields option

If `true`, fetches a set of random documents to identify which fields are actually used by
documents. Can be useful for indices with lots of field mappings to increase query/reindex
performance. Defaults to `false`.

README.md

@@ -118,6 +118,7 @@ transformer({
 - `mappings`: Optional Elasticsearch document mappings. If not set and you're reindexing from another index, the mappings from the existing index will be used.
 - `mappingsOverride`: If you're reindexing and this is set to `true`, `mappings` will be applied on top of the source index's mappings. Defaults to `false`.
 - `indexMappingTotalFieldsLimit`: Optional field limit for the target index to be created that will be passed on as the `index.mapping.total_fields.limit` setting.
+- `populatedFields`: If `true`, fetches a set of random documents to identify which fields are actually used by documents. Can be useful for indices with lots of field mappings to increase query/reindex performance. Defaults to `false`.
 - `query`: Optional Elasticsearch [DSL query](https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl.html) to filter documents from the source index.
 - `skipHeader`: If true, skips the first line of the source file. Defaults to `false`.
 - `transform(line)`: A callback function which allows the transformation of a source line into one or several documents.

src/_index-reader.js

@@ -14,37 +14,68 @@ export default function indexReaderFactory(
   client,
   query,
   bufferSize = DEFAULT_BUFFER_SIZE,
+  populatedFields = false,
 ) {
   return async function indexReader() {
     const responseQueue = [];
     let docsNum = 0;
 
-    function search() {
+    async function fetchPopulatedFields() {
+      try {
+        const response = await client.search({
+          index: sourceIndexName,
+          size: bufferSize,
+          query: {
+            function_score: {
+              query,
+              random_score: {},
+            },
+          },
+        });
+
+        // Get all field names for each returned doc and flatten it
+        // to a list of unique field names used across all docs.
+        return new Set(response.hits.hits.map(d => Object.keys(d._source)).flat(1));
+      } catch (e) {
+        console.log('error', e);
+      }
+    }
+
+    function search(fields) {
       return client.search({
         index: sourceIndexName,
-        scroll: '30s',
+        scroll: '600s',
         size: bufferSize,
         query,
+        ...(fields ? { _source: fields } : {}),
       });
     }
 
     function scroll(id) {
       return client.scroll({
         scroll_id: id,
-        scroll: '30s',
+        scroll: '600s',
       });
     }
 
+    let fieldsWithData;
+
+    // identify populated fields
+    if (populatedFields) {
+      fieldsWithData = await fetchPopulatedFields();
+    }
+
     // start things off by searching, setting a scroll timeout, and pushing
     // our first response into the queue to be processed
-    const se = await search();
+    const se = await search(fieldsWithData);
     responseQueue.push(se);
     progressBar.start(se.hits.total.value, 0);
 
     function processHit(hit) {
       docsNum += 1;
       try {
         const doc = typeof transform === 'function' ? transform(hit._source) : hit._source; // eslint-disable-line no-underscore-dangle
+
         // if doc is undefined we'll skip indexing it
         if (typeof doc === 'undefined') {
           return;

src/main.js

@@ -18,6 +18,7 @@ export default async function transformer({
   mappings,
   mappingsOverride = false,
   indexMappingTotalFieldsLimit,
+  populatedFields = false,
   query,
   skipHeader = false,
   transform,
@@ -75,6 +76,7 @@ export default async function transformer({
         sourceClient,
         query,
         bufferSize,
+        populatedFields,
       );
     }