#139 - Support checksums and conditional loading for a given CSV row (#322)

Author: mseaton

README.md

@@ -231,6 +231,7 @@ See the [documentation on Initializer's logging properties](readme/rtprops.md#lo
 * Fix conceptsets domain to prevent incorrect unretiring of associated concept
 * Fix to ensure concept lookups correctly handle concepts with % characters
 * Fix for performance to only process displays preloaders if they are present within csvs
+* Add support for row-level checksums in CSV domains to enable only re-loading those rows that have changed
 
 #### Version 2.11.0
 * Added support for patient flags (flags, flagpriorities, flagtags) domains

api/src/main/java/org/openmrs/module/initializer/InitializerConfig.java

@@ -32,6 +32,7 @@
 
 import static org.openmrs.module.initializer.InitializerConstants.PROPS_DOMAINS;
 import static org.openmrs.module.initializer.InitializerConstants.PROPS_EXCLUDE;
+import static org.openmrs.module.initializer.InitializerConstants.PROPS_ROW_CHECKSUMS_ENABLED;
 import static org.openmrs.module.initializer.InitializerConstants.PROPS_SKIPCHECKSUMS;
 import static org.openmrs.module.initializer.InitializerConstants.PROPS_STARTUP_LOAD;
 import static org.openmrs.module.initializer.InitializerConstants.PROPS_STARTUP_LOAD_CONTINUE_ON_ERROR;
@@ -53,6 +54,8 @@ public class InitializerConfig implements InitializingBean {
 
 	private Boolean skipChecksums = false;
 
+	private Boolean rowChecksumsEnabled = false;
+	
 	private String startupLoadingMode = "";
 
 	@Override
@@ -90,6 +93,8 @@ public void init() {
 
 		// checksums
 		skipChecksums = BooleanUtils.toBoolean(Optional.ofNullable(getPropertyValue(PROPS_SKIPCHECKSUMS)).orElse(""));
+		rowChecksumsEnabled = BooleanUtils
+		        .toBoolean(Optional.ofNullable(getPropertyValue(PROPS_ROW_CHECKSUMS_ENABLED)).orElse(""));
 
 		// Startup Loading Configuration
 		startupLoadingMode = getPropertyValue(PROPS_STARTUP_LOAD);
@@ -138,6 +143,14 @@ public boolean skipChecksums() {
 		return skipChecksums;
 	}
 
+	/**
+	 * @return true to enable per-row checksum tracking for CSV loaders, false (default) to use only
+	 *         file-level checksums
+	 */
+	public boolean isRowChecksumsEnabled() {
+		return rowChecksumsEnabled;
+	}
+	
 	/**
 	 * @return how configuration should be loaded in at startup in the module activator
 	 */

api/src/main/java/org/openmrs/module/initializer/InitializerConstants.java

@@ -29,6 +29,8 @@ public class InitializerConstants {
 
 	public static final String PROPS_SKIPCHECKSUMS = MODULE_ARTIFACT_ID + "." + "skip.checksums";
 
+	public static final String PROPS_ROW_CHECKSUMS_ENABLED = MODULE_ARTIFACT_ID + "." + "row.checksums.enabled";
+	
 	/*
 	 * Startup properties
 	 */

api/src/main/java/org/openmrs/module/initializer/api/ConfigDirUtil.java

@@ -4,14 +4,19 @@
 import java.io.FileInputStream;
 import java.io.FilenameFilter;
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
+import java.util.TreeMap;
 import java.util.stream.Stream;
 
 import org.apache.commons.codec.digest.DigestUtils;
@@ -36,6 +41,8 @@ public class ConfigDirUtil {
 
 	public static final String CHECKSUM_FILE_EXT = "checksum";
 
+	public static final String ROW_CHECKSUM_FILE_EXT = "rows.checksum";
+	
 	protected static final Logger log = LoggerFactory.getLogger(ConfigDirUtil.class);
 
 	protected boolean skipChecksums = false;
@@ -303,6 +310,109 @@ public void writeChecksum(File configFile, String checksum) {
 		}
 	}
 
+	/**
+	 * Computes a stable hash for a single CSV row. Each populated column header is paired with its
+	 * (possibly empty) value, and pairs are sorted by header name. This means column reordering does
+	 * not change the hash, but column addition/removal/rename or any value change — including a
+	 * transition between "absent" and "present-but-empty" — does. The latter distinction matters
+	 * because some line processors (e.g. NestedConceptLineProcessor) treat a present-but-empty cell as
+	 * an explicit clear, distinct from an absent column.
+	 * 
+	 * @param headerLine The CSV header line (column names).
+	 * @param line The CSV data row.
+	 * @return The MD5 hex hash of the normalized row.
+	 */
+	public static String computeRowChecksum(String[] headerLine, String[] line) {
+		if (headerLine == null || line == null) {
+			return NOT_COMPUTABLE_CHECKSUM;
+		}
+		Map<String, String> sorted = new TreeMap<>();
+		for (int i = 0; i < headerLine.length; i++) {
+			String header = headerLine[i];
+			if (StringUtils.isEmpty(header)) {
+				continue;
+			}
+			String value = (i < line.length && line[i] != null) ? line[i] : "";
+			sorted.put(header, value);
+		}
+		StringBuilder sb = new StringBuilder();
+		for (Map.Entry<String, String> e : sorted.entrySet()) {
+			sb.append(e.getKey()).append('=').append(e.getValue()).append('\n');
+		}
+		return DigestUtils.md5Hex(sb.toString().getBytes(StandardCharsets.UTF_8));
+	}
+	
+	/**
+	 * Reads the previously-saved set of row hashes for a CSV configuration file.
+	 * 
+	 * @param configFile The CSV configuration file.
+	 * @return The set of row hashes from the previous successful processing, or an empty set if no row
+	 *         checksum file exists yet.
+	 */
+	public Set<String> readRowChecksums(File configFile) {
+		final String rowChecksumFilename = getLocatedFilename(domainDirPath, configFile) + "." + ROW_CHECKSUM_FILE_EXT;
+		final Path rowChecksumPath = Paths.get(domainChecksumsDirPath, rowChecksumFilename);
+		Set<String> hashes = new HashSet<>();
+		if (!rowChecksumPath.toFile().exists()) {
+			return hashes;
+		}
+		try {
+			for (String line : Files.readAllLines(rowChecksumPath, StandardCharsets.UTF_8)) {
+				String trimmed = line.trim();
+				if (!trimmed.isEmpty()) {
+					hashes.add(trimmed);
+				}
+			}
+		}
+		catch (IOException e) {
+			log.warn("Error reading row checksum file: " + rowChecksumPath, e);
+		}
+		return hashes;
+	}
+	
+	/**
+	 * Writes the given set of row hashes for a CSV configuration file. Overwrites any previously saved
+	 * row checksums for the same file.
+	 * 
+	 * @param configFile The CSV configuration file.
+	 * @param rowChecksums The row hashes to persist.
+	 */
+	public void writeRowChecksums(File configFile, Set<String> rowChecksums) {
+		if (skipChecksums) {
+			return;
+		}
+		final String rowChecksumFilename = getLocatedFilename(domainDirPath, configFile) + "." + ROW_CHECKSUM_FILE_EXT;
+		final Path rowChecksumPath = Paths.get(domainChecksumsDirPath, rowChecksumFilename);
+		try {
+			Files.deleteIfExists(rowChecksumPath);
+			if (rowChecksums == null || rowChecksums.isEmpty()) {
+				return;
+			}
+			List<String> sorted = new ArrayList<>(rowChecksums);
+			Collections.sort(sorted);
+			FileUtils.writeLines(rowChecksumPath.toFile(), "UTF-8", sorted);
+		}
+		catch (IOException e) {
+			log.error("Error writing row checksum file at: " + rowChecksumPath, e);
+		}
+	}
+	
+	/**
+	 * Removes the row checksum file for a CSV configuration file, if any.
+	 * 
+	 * @param configFile The CSV configuration file.
+	 */
+	public void deleteRowChecksums(File configFile) {
+		final String rowChecksumFilename = getLocatedFilename(domainDirPath, configFile) + "." + ROW_CHECKSUM_FILE_EXT;
+		final Path rowChecksumPath = Paths.get(domainChecksumsDirPath, rowChecksumFilename);
+		try {
+			Files.deleteIfExists(rowChecksumPath);
+		}
+		catch (IOException e) {
+			log.warn("Error deleting row checksum file at: " + rowChecksumPath, e);
+		}
+	}
+	
 	/**
 	 * Removes the specified checksum file in the specified checksums directory.
 	 * 

api/src/main/java/org/openmrs/module/initializer/api/loaders/BaseCsvLoader.java

@@ -5,7 +5,10 @@
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 import org.openmrs.OpenmrsObject;
@@ -88,8 +91,32 @@ protected void load(InputStream is) throws Exception {
 		//
 
 		final CsvParser<T, BaseLineProcessor<T>> parser = getParser(is);
-		List<String[]> remainingLines = parser.getLines();
-		int totalCount = remainingLines.size();
+		final List<String[]> allLines = parser.getLines();
+		final int totalCount = allLines.size();
+		final String[] headerLine = parser.getHeaderLine();
+		
+		final boolean rowChecksumsEnabled = cfg.isRowChecksumsEnabled() && !cfg.skipChecksums();
+		final File file = getLoadedFile();
+		final ConfigDirUtil dirUtil = getDirUtil();
+		
+		final Set<String> previousRowHashes = rowChecksumsEnabled ? dirUtil.readRowChecksums(file) : new HashSet<>();
+		
+		List<String[]> remainingLines;
+		if (rowChecksumsEnabled) {
+			remainingLines = new ArrayList<>();
+			for (String[] line : allLines) {
+				if (!previousRowHashes.contains(ConfigDirUtil.computeRowChecksum(headerLine, line))) {
+					remainingLines.add(line);
+				}
+			}
+			int skipped = totalCount - remainingLines.size();
+			if (skipped > 0) {
+				log.info(skipped + " of " + totalCount + " CSV rows in " + file.getName()
+				        + " are unchanged since last load and will be skipped.");
+			}
+		} else {
+			remainingLines = new ArrayList<>(allLines);
+		}
 
 		int lastFailCount = 0;
 		CsvFailingLines result = new CsvFailingLines();
@@ -100,11 +127,28 @@ protected void load(InputStream is) throws Exception {
 			remainingLines = result.getFailingLines();
 		}
 
+		// Row checksum bookkeeping: persist hashes for rows considered processed — i.e. all rows in
+		// the file except those that failed on this run. Failing rows are excluded so they will be
+		// retried on the next load.
+		if (rowChecksumsEnabled) {
+			Set<String> failingHashes = new HashSet<>();
+			for (String[] failingLine : result.getFailingLines()) {
+				failingHashes.add(ConfigDirUtil.computeRowChecksum(headerLine, failingLine));
+			}
+			Set<String> newRowHashes = new HashSet<>();
+			for (String[] line : allLines) {
+				String rowHash = ConfigDirUtil.computeRowChecksum(headerLine, line);
+				if (!failingHashes.contains(rowHash)) {
+					newRowHashes.add(rowHash);
+				}
+			}
+			dirUtil.writeRowChecksums(file, newRowHashes);
+		}
+		
 		//
 		// logging
 		//
 
-		final File file = getLoadedFile();
 		// success logging
 		if (isEmpty(result.getFailingLines())) {
 			log.info(file.getName() + " ('" + getDomainName() + "' domain) was entirely successfully processed.");

api/src/test/java/org/openmrs/module/initializer/api/ConfigDirUtilTest.java

@@ -1,10 +1,13 @@
 package org.openmrs.module.initializer.api;
 
 import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.not;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.empty;
 import static org.openmrs.module.initializer.api.ConfigDirUtil.CHECKSUM_FILE_EXT;
+import static org.openmrs.module.initializer.api.ConfigDirUtil.ROW_CHECKSUM_FILE_EXT;
+import static org.openmrs.module.initializer.api.ConfigDirUtil.computeRowChecksum;
 import static org.openmrs.module.initializer.api.ConfigDirUtil.getLocatedFilename;
 
 import java.io.File;
@@ -14,8 +17,10 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 import org.apache.commons.codec.digest.DigestUtils;
@@ -136,6 +141,100 @@ public void getFiles_shouldHandleChecksumsWhenNestedFiles() throws IOException {
 		}
 	}
 
+	@Test
+	public void computeRowChecksum_shouldBeStableUnderColumnReordering() {
+		String[] header1 = { "uuid", "name", "description" };
+		String[] row1 = { "abc-123", "Acme", "A clinic" };
+		String[] header2 = { "name", "description", "uuid" };
+		String[] row2 = { "Acme", "A clinic", "abc-123" };
+		Assert.assertEquals(computeRowChecksum(header1, row1), computeRowChecksum(header2, row2));
+	}
+	
+	@Test
+	public void computeRowChecksum_shouldChangeWhenAValueChanges() {
+		String[] header = { "uuid", "name" };
+		String[] row1 = { "abc-123", "Acme" };
+		String[] row2 = { "abc-123", "Beta" };
+		assertThat(computeRowChecksum(header, row1), is(not(computeRowChecksum(header, row2))));
+	}
+	
+	@Test
+	public void computeRowChecksum_shouldDistinguishAbsentFromPresentButEmpty() {
+		// Adding a column — even when its cell is empty — must change the hash, because some line
+		// processors (e.g. NestedConceptLineProcessor) treat a present-but-empty cell as a directive
+		// to clear an existing field and treat an absent column as "leave the field alone".
+		String[] header1 = { "uuid", "name" };
+		String[] row1 = { "abc-123", "Acme" };
+		String[] header2 = { "uuid", "name", "description" };
+		String[] row2 = { "abc-123", "Acme", null };
+		String[] row3 = { "abc-123", "Acme", "" };
+		assertThat(computeRowChecksum(header1, row1), is(not(computeRowChecksum(header2, row2))));
+		assertThat(computeRowChecksum(header1, row1), is(not(computeRowChecksum(header2, row3))));
+		// However null and empty values within an existing column should be treated identically,
+		// since CsvParser normalizes blank cells to null on read.
+		Assert.assertEquals(computeRowChecksum(header2, row2), computeRowChecksum(header2, row3));
+	}
+	
+	@Test
+	public void computeRowChecksum_shouldChangeWhenAColumnIsRenamed() {
+		String[] header1 = { "uuid", "name" };
+		String[] header2 = { "uuid", "label" };
+		String[] row = { "abc-123", "Acme" };
+		assertThat(computeRowChecksum(header1, row), is(not(computeRowChecksum(header2, row))));
+	}
+	
+	@Test
+	public void rowChecksums_shouldRoundTripThroughDisk() throws IOException {
+		String configDirPath = getClass().getClassLoader().getResource("org/openmrs/module/initializer/include").getPath();
+		String checksumsDirPath = Files.createTempDirectory("configuration_checksums_rows").toString();
+		String domain = "file_patterns";
+		
+		ConfigDirUtil dirUtil = new ConfigDirUtil(configDirPath, checksumsDirPath, domain);
+		File configFile = new File(Paths.get(configDirPath, domain, "diagnoses.csv").toString());
+		
+		Set<String> hashes = new HashSet<>();
+		hashes.add("hash-a");
+		hashes.add("hash-b");
+		hashes.add("hash-c");
+		
+		// Writing then reading should round-trip the set.
+		dirUtil.writeRowChecksums(configFile, hashes);
+		File rowsFile = Paths.get(checksumsDirPath, domain,
+		    getLocatedFilename(Paths.get(configDirPath, domain).toString(), configFile) + "." + ROW_CHECKSUM_FILE_EXT)
+		        .toFile();
+		assertThat(rowsFile.exists(), is(true));
+		assertThat(dirUtil.readRowChecksums(configFile), is(hashes));
+		
+		// Writing an empty set should remove the file.
+		dirUtil.writeRowChecksums(configFile, new HashSet<>());
+		assertThat(rowsFile.exists(), is(false));
+		
+		// Reading when no file exists returns an empty set.
+		assertThat(dirUtil.readRowChecksums(configFile), is(empty()));
+	}
+	
+	@Test
+	public void deleteRowChecksums_shouldRemoveTheRowChecksumFile() throws IOException {
+		String configDirPath = getClass().getClassLoader().getResource("org/openmrs/module/initializer/include").getPath();
+		String checksumsDirPath = Files.createTempDirectory("configuration_checksums_rows_delete").toString();
+		String domain = "file_patterns";
+		
+		ConfigDirUtil dirUtil = new ConfigDirUtil(configDirPath, checksumsDirPath, domain);
+		File configFile = new File(Paths.get(configDirPath, domain, "diagnoses.csv").toString());
+		
+		Set<String> hashes = new HashSet<>();
+		hashes.add("hash-a");
+		dirUtil.writeRowChecksums(configFile, hashes);
+		
+		File rowsFile = Paths.get(checksumsDirPath, domain,
+		    getLocatedFilename(Paths.get(configDirPath, domain).toString(), configFile) + "." + ROW_CHECKSUM_FILE_EXT)
+		        .toFile();
+		assertThat(rowsFile.exists(), is(true));
+		
+		dirUtil.deleteRowChecksums(configFile);
+		assertThat(rowsFile.exists(), is(false));
+	}
+	
 	/*
 	 * One of the CSV files has a non-parseable _order.
 	 * The resulting exception is logged as an error.