feat: auto-update index mappings and validate document fields

queries/cdmq/add-run-worker.js

@@ -5,6 +5,19 @@ const { Readable } = require('stream');
 const cdm = require('./cdm');
 const fs = require('fs');
 
+function collectFieldPaths(obj, prefix) {
+  const paths = [];
+  for (const key of Object.keys(obj)) {
+    const path = prefix ? prefix + '.' + key : key;
+    if (typeof obj[key] === 'object' && obj[key] !== null && !Array.isArray(obj[key])) {
+      paths.push(...collectFieldPaths(obj[key], path));
+    } else {
+      paths.push(path);
+    }
+  }
+  return paths;
+}
+
 // Read an xz file and decompress to string
 async function decompressXzFile(filename) {
   debuglog('decompressXzFile: reading ' + filename);
@@ -22,7 +35,7 @@ async function decompressXzFile(filename) {
 module.exports = async ({ instance, filePath, docTypes, mode }) => {
   const maxLength = 4000000;
   const jsonArr = [];
-  const info = { runIds: {} };
+  const info = { runIds: {}, docFields: {} };
   const regExp = /\.ndjson\.xz$/;
   var docTypeCounts = {};
   if (mode == 'index') {
@@ -80,6 +93,16 @@ module.exports = async ({ instance, filePath, docTypes, mode }) => {
         const matches = regExp.exec(indexName);
         if (matches) {
           const cdmVer = matches[1];
+          const docType = matches[2];
+          const fieldPaths = collectFieldPaths(doc);
+          if (!info.docFields[docType]) {
+            info.docFields[docType] = [];
+          }
+          for (const fp of fieldPaths) {
+            if (!info.docFields[docType].includes(fp)) {
+              info.docFields[docType].push(fp);
+            }
+          }
           let yearDotMonth = '';
           let runId = '';
           if (cdmVer == 'v7dev') {

queries/cdmq/add-run.js

@@ -158,24 +158,47 @@ async function main() {
           process.exit(1);
         }
         instance['ver'] = cdmVer;
-        //if (!Object.keys(instance['indices']).includes(cdmVer)) {
-        //instance['indices'][cdmVer] = [];
-        //}
       } else {
         console.log('ERROR: there was not exactly one CDM version found in the data to be indexed:\n');
         console.log(Object.keys(info['indices']));
         console.log('info\n' + JSON.stringify(info['indices'], null, 2));
         process.exit(1);
       }
-      // For cdmv9 and newer, any time a document is to be indexed, it is imperitive
-      // that a check for the existence of the index is done, and if not found, create
-      // the index with the *correct* mappings and settings.  If this is not done, an index
-      // may be auto-created with the *incorrect* mappings and settings, and documents can
-      // be indexed, but not properly, and subsequent queries will *NOT* work.
+      if (info.docFields) {
+        var unknownFields = {};
+        var hasUnknown = false;
+        for (const docType of Object.keys(info.docFields)) {
+          if (!cdm.indexDefs[cdmVer] || !cdm.indexDefs[cdmVer][docType]) {
+            console.error('ERROR: no index definition found for ' + cdmVer + '/' + docType);
+            process.exit(1);
+          }
+          const validPaths = cdm.getMappingFieldPaths(cdm.indexDefs[cdmVer][docType]['mappings']['properties']);
+          for (const fieldPath of info.docFields[docType]) {
+            if (!validPaths.has(fieldPath)) {
+              if (!unknownFields[docType]) {
+                unknownFields[docType] = [];
+              }
+              unknownFields[docType].push(fieldPath);
+              hasUnknown = true;
+            }
+          }
+        }
+        if (hasUnknown) {
+          console.error('ERROR: documents contain fields not defined in indexDefs:');
+          for (const docType of Object.keys(unknownFields)) {
+            console.error('  ' + docType + ': ' + unknownFields[docType].join(', '));
+          }
+          console.error('These fields must be added to indexDefs in cdm.js before indexing.');
+          process.exit(1);
+        }
+        console.log('Document field validation passed');
+      }
+
       debuglog(JSON.stringify(info['runIds'][runId]['indices'][cdmVer], null, 2));
       for (var i = 0; i < info['runIds'][runId]['indices'][cdmVer].length; i++) {
         debuglog('checking for index ' + info['runIds'][runId]['indices'][cdmVer][i]);
         cdm.checkCreateIndex(instance, info['runIds'][runId]['indices'][cdmVer][i]);
+        cdm.updateIndexMappings(instance, info['runIds'][runId]['indices'][cdmVer][i]);
       }
       // Before indexing any documents, we must check for any existing ones.  Having duplicate
       // documents is really bad

queries/cdmq/cdm.js

@@ -399,6 +399,8 @@ indexDefs['v8dev']['metric_data']['mappings']['properties']['metric_data'] = {
 };
 indexDefs['v9dev']['metric_data'] = deepClone(indexDefs['v8dev']['metric_data']);
 
+exports.indexDefs = indexDefs;
+
 // --------------------------------------------------------------------------------------------------------------
 function memUsage() {
   if (debugOut == 0) return;
@@ -513,6 +515,62 @@ checkCreateIndex = function (instance, index) {
 };
 exports.checkCreateIndex = checkCreateIndex;
 
+// --------------------------------------------------------------------------------------------------------------
+getMappingFieldPaths = function (properties, prefix) {
+  const paths = new Set();
+  for (const key of Object.keys(properties)) {
+    const path = prefix ? prefix + '.' + key : key;
+    if (properties[key].properties) {
+      for (const subPath of getMappingFieldPaths(properties[key].properties, path)) {
+        paths.add(subPath);
+      }
+    } else {
+      paths.add(path);
+    }
+  }
+  return paths;
+};
+exports.getMappingFieldPaths = getMappingFieldPaths;
+
+// --------------------------------------------------------------------------------------------------------------
+updateIndexMappings = function (instance, index) {
+  var indices = index.includes(',') ? index.split(',') : [index];
+
+  for (var idx = 0; idx < indices.length; idx++) {
+    var thisIndex = indices[idx];
+
+    var resp = getCdmVerFromIndex(thisIndex);
+    if (resp['ret-code'] != 0) {
+      console.error('ERROR: updateIndexMappings: getCdmVerFromIndex returned ' + resp['ret-msg']);
+      return createResponse(1, resp['ret-msg']);
+    }
+    var cdmVer = resp['cdm-ver'];
+
+    resp = getDocType(thisIndex);
+    if (resp['ret-code'] != 0) {
+      console.error('ERROR: updateIndexMappings: getDocType returned ' + resp['ret-msg']);
+      return createResponse(1, resp['ret-msg']);
+    }
+    var docType = resp['doc-type'];
+
+    var url = 'http://' + instance['host'] + '/' + thisIndex + '/_mapping';
+    debuglog('updateIndexMappings: PUT ' + url);
+    resp = request('PUT', url, {
+      headers: instance['header'],
+      body: JSON.stringify(indexDefs[cdmVer][docType]['mappings'])
+    });
+    var data = JSON.parse(resp.getBody());
+    if (data['error']) {
+      console.error('ERROR: updateIndexMappings: ' + JSON.stringify(data['error']));
+      return createResponse(1, JSON.stringify(data['error']));
+    }
+    debuglog('updateIndexMappings response: ' + JSON.stringify(data, null, 2));
+  }
+
+  return createResponse(0, 'INFO: updateIndexMappings completed for ' + indices.length + ' index(es)');
+};
+exports.updateIndexMappings = updateIndexMappings;
+
 // --------------------------------------------------------------------------------------------------------------
 function getDocType(index) {
   var retMsg = '';
@@ -531,7 +589,7 @@ function getDocType(index) {
     if (matches) {
       docType = matches[1];
       if (docTypes[cdmVer].includes(docType)) {
-        return docType;
+        return createResponse(retCode, retMsg, { 'doc-type': docType });
       } else {
         retMsg = 'ERROR: index [' + index + '] does not match a docType: ' + docTypes[cdmVer];
         retCode = 1;