diff --git a/queries/cdmq/add-run-worker.js b/queries/cdmq/add-run-worker.js index 32197207..134d6167 100755 --- a/queries/cdmq/add-run-worker.js +++ b/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') { diff --git a/queries/cdmq/add-run.js b/queries/cdmq/add-run.js index 0b4128f3..70a32584 100755 --- a/queries/cdmq/add-run.js +++ b/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 diff --git a/queries/cdmq/cdm.js b/queries/cdmq/cdm.js index 401fdb45..cea01c4a 100644 --- a/queries/cdmq/cdm.js +++ b/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;