DatabaseConsolidator.java

1 
package com.renomad.minum.database;
2
3 
import com.renomad.minum.logging.ILogger;
4 
import com.renomad.minum.state.Context;
5 
import com.renomad.minum.utils.FileUtils;
6
7 
import java.io.IOException;
8 
import java.nio.charset.StandardCharsets;
9 
import java.nio.file.Files;
10 
import java.nio.file.Path;
11 
import java.text.ParseException;
12 
import java.util.*;
13 
import java.util.stream.Collectors;
14
15 
import static com.renomad.minum.database.ChecksumUtility.buildChecksum;
16 
import static com.renomad.minum.database.ChecksumUtility.compareWithChecksum;
17 
import static com.renomad.minum.database.DatabaseAppender.simpleDateFormat;
18
19 
/**
20 
 * Consolidates the database append logs.
21 
 * <br>
22 
 * As the append logs get filled up, the consolidator comes
23 
 * along after to analyze those changes and determine a
24 
 * consolidated version.  For example, if the append logs
25 
 * have three updates for a particular element, then the consolidated file
26 
 * will have just the last update.
27 
 */
28 
final class DatabaseConsolidator {
29
30 
    /**
31 
     * This is the path to the append-only files, where incoming
32 
     * changes to the data are quickly stored.
33 
     */
34 
    private final Path appendLogDirectory;
35
36 
    /**
37 
     * This is the path to where we store consolidated data, so that
38 
     * database startup is as fast as possible.
39 
     */
40 
    private final Path consolidatedDataDirectory;
41
42 
    private final ILogger logger;
43
44 
    private final int maxLinesPerFile;
45
46 
    /**
47 
     * This represents an instruction for how to change the overall consolidated
48 
     * database files on disk.  Instructions are either to UPDATE or DELETE. This
49 
     * also encapsulates the data we're updating.
50 
     */
51 
    private record DatabaseChangeInstruction(DatabaseChangeAction action, long dataIndex, String data) {}
52
53 
    DatabaseConsolidator(Path persistenceDirectory, Context context) {
54 
        this.appendLogDirectory = persistenceDirectory.resolve("append_logs");
55 
        this.consolidatedDataDirectory = persistenceDirectory.resolve("consolidated_data");
56 
        var constants = context.getConstants();
57 
        this.logger = context.getLogger();
58 
        FileUtils fileUtils = new FileUtils(logger, constants);
59 1 1. <init> : removed call to com/renomad/minum/utils/FileUtils::makeDirectory → KILLED
 
        fileUtils.makeDirectory(this.consolidatedDataDirectory);
60 
        this.maxLinesPerFile = constants.maxLinesPerConsolidatedDatabaseFile;
61 
    }
62
63 
    /**
64 
     * Loop through all the append-only files
65 
     */
66 
    void consolidate() throws IOException {
67 
        logger.logDebug(() -> "Starting database consolidator");
68 
        List<Date> sortedList = getSortedAppendLogs(appendLogDirectory);
69 1 1. consolidate : negated conditional → TIMED_OUT
 
        if (sortedList.isEmpty()) {
70 
            logger.logDebug(() -> "No database files found to consolidate - exiting");
71 
            return;
72 
        } else {
73 
            logger.logDebug(() -> "Files to consolidate: " + sortedList.stream().map(simpleDateFormat::format).collect(Collectors.joining(";")));
74 
        }
75
76 
        // process the files in order.  This does potentially cause
77 
        // multiple updates for the consolidated files, but that's
78 
        // safer than building up too large a structure in memory
79 
        // before writing, and in any case, we're prioritizing efficiency
80 
        // so there should only be one write to each file per loop.
81 
        //
82 
        // after each append-only file is fully processed, it gets deleted.
83 
        for (Date date : sortedList) {
84 
            String filename = simpleDateFormat.format(date);
85 
            logger.logDebug(() -> "consolidator processing file " + filename + " in " + appendLogDirectory);
86 1 1. consolidate : removed call to com/renomad/minum/database/DatabaseConsolidator::processAppendLogFile → KILLED
 
            processAppendLogFile(filename);
87 
            logger.logDebug(() -> "consolidator finished with file " + filename + " in " + appendLogDirectory);
88 
        }
89 
        logger.logDebug(() -> "Database consolidation finished");
90 
    }
91
92
93 
    /**
94 
     * The expectation is that after we finish reading the X lines in
95 
     * this append log, we will have a set of clear instructions to
96 
     * apply to our previously consolidated files. There should end up
97 
     * being just one action for each id - update or delete.
98 
     * <br>
99 
     * Build a data structure holding instructions for the next step.
100 
     */
101 
    private void processAppendLogFile(String filename) throws IOException {
102 
        Path fullPathToFile = this.appendLogDirectory.resolve(filename);
103 
        List<String> lines = Files.readAllLines(fullPathToFile);
104 
        Map<Long, DatabaseChangeInstruction> resultingInstructions = new HashMap<>();
105
106 
        // process each line from the file
107
108 
        for (String line : lines) {
109 
            DatabaseChangeInstruction databaseChange = parseDatabaseChangeInstructionString(line, filename);
110
111 
            // the trick here is that by using a Map, only the last item added will remain at the end
112 
            resultingInstructions.put(databaseChange.dataIndex(), databaseChange);
113 
        }
114
115 
        // now we have the concise list of state changes, but the next step is figuring out how
116 
        // to organize them by their destination.  consolidated files will be grouped somehow.
117 
        // For example, indexes 1 - 1000, 1001-2000, etc (there may be more than 1000 per file).
118 
        // <br>
119 
        // So, we will group our data that way,
120 
        // and then efficiently update the files (a bad outcome, in contrast, would be updating
121 
        // the files multiple times each).
122
123 
        Map<Long, Collection<DatabaseChangeInstruction>> groupedInstructions = groupInstructionsByPartition(resultingInstructions);
124
125 1 1. processAppendLogFile : removed call to com/renomad/minum/database/DatabaseConsolidator::rewriteFiles → KILLED
 
        rewriteFiles(groupedInstructions);
126
127 
        // delete the file
128 1 1. processAppendLogFile : removed call to java/nio/file/Files::delete → KILLED
 
        Files.delete(fullPathToFile);
129 
    }
130
131 
    /**
132 
     * Given a {@link Map} of database change instructions, grouped by keys representing the
133 
     * first index in a group of indexes (like 1 to 100, or 101 to 200, etc), write the
134 
     * data to files, with consideration for what might already exist.  That is to say,
135 
     * if we are adding grouped instructions to an existing file such as "1_to_100", then
136 
     * we want to merge our incoming data with what is already there.  Otherwise, we are just
137 
     * creating a new file.
138 
     */
139 
    private void rewriteFiles(Map<Long, Collection<DatabaseChangeInstruction>> groupedInstructions) throws IOException {
140 
        for (Map.Entry<Long, Collection<DatabaseChangeInstruction>> instructions : groupedInstructions.entrySet()) {
141 2 1. rewriteFiles : Replaced long addition with subtraction → TIMED_OUT
2. rewriteFiles : Replaced integer subtraction with addition → KILLED
 
            String filename = String.format("%d_to_%d", instructions.getKey(), instructions.getKey() + (maxLinesPerFile - 1));
142 
            logger.logTrace(() -> "Writing consolidated data to " + filename);
143 
            List<String> data;
144 
            // if the file doesn't exist, we'll just start with an empty list. If it
145 
            // does exist, read its lines into a List data structure.
146 
            Path fullPathToConsolidatedFile = this.consolidatedDataDirectory.resolve(filename);
147 1 1. rewriteFiles : negated conditional → KILLED
 
            if (!Files.exists(fullPathToConsolidatedFile)) {
148 
                data = new ArrayList<>();
149 
            } else {
150 
                data = readConsolidatedFileWithChecksum(fullPathToConsolidatedFile);
151 
            }
152
153 
            // update the data in memory per the instructions
154 
            Collection<String> updatedData = updateData(filename, data, instructions.getValue());
155
156 
            String checksumString = buildChecksum(updatedData);
157
158 
            // write the data to disk
159 
            Files.write(fullPathToConsolidatedFile, updatedData, StandardCharsets.US_ASCII);
160
161 
            // write a hash of the data to use as a checksum.  This value will be checked
162 
            // when reading the data later on, to confirm nothing has changed since writing.
163 
            Path fullPathToChecksumFile = this.consolidatedDataDirectory.resolve(filename + ".checksum");
164 
            Files.writeString(fullPathToChecksumFile, checksumString);
165 
        }
166 
    }
167
168 
    /**
169 
     * Reads data from consolidated file, confirming the checksum in the process.
170 
     */
171 
    private static List<String> readConsolidatedFileWithChecksum(Path fullPathToConsolidatedFile) throws IOException {
172 
        // get all the data from the consolidated file
173 
        List<String> data = Files.readAllLines(fullPathToConsolidatedFile);
174
175 
        compareWithChecksum(fullPathToConsolidatedFile, data);
176
177 1 1. readConsolidatedFileWithChecksum : replaced return value with Collections.emptyList for com/renomad/minum/database/DatabaseConsolidator::readConsolidatedFileWithChecksum → KILLED
 
        return data;
178 
    }
179
180
181 
    /**
182 
     * Here, we have raw lines of data from a file, and a list of instructions for updating
183 
     * that data.  We will organize the raw data better, apply the instructions, and return
184 
     * the updated data
185 
     *
186 
     * @param linesOfData  raw lines of data from a file
187 
     * @param instructions details of how to change the data in the file, either UPDATE or DELETE
188 
     * @return an updated and sorted list of strings (sorted by index, which is the first value on each line)
189 
     */
190 
    static Collection<String> updateData(String filename, List<String> linesOfData, Collection<DatabaseChangeInstruction> instructions) {
191 
        SortedMap<Long, String> result = new TreeMap<>();
192 
        // put the original data into a map
193 
        for (String data : linesOfData) {
194 
            // the first pipe symbol is where the index number ends.  Apologies for
195 
            // the overlap of terms here, index and index.
196 
            int indexOfFirstPipe = data.indexOf('|');
197 1 1. updateData : negated conditional → KILLED
 
            if (indexOfFirstPipe == -1) {
198 
                throw new DbException(String.format("Error parsing line in file.  File: %s line: %s", filename, data));
199 
            }
200 
            String dataIndexString = data.substring(0, indexOfFirstPipe);
201 
            long dataIndexLong;
202 
            try {
203 
                dataIndexLong = Long.parseLong(dataIndexString);
204 
            } catch (NumberFormatException ex) {
205 
                throw new DbException(String.format("Failed to parse index from line in file. File: %s line: %s", filename, data), ex);
206 
            }
207 
            result.put(dataIndexLong, data);
208 
        }
209
210 
        // change that data per instructions
211 
        for (DatabaseChangeInstruction instruction : instructions) {
212 1 1. updateData : negated conditional → KILLED
 
            if (DatabaseChangeAction.UPDATE.equals(instruction.action())) {
213 
                result.put(instruction.dataIndex(), instruction.data());
214 
            } else {
215 
                // only other option is DELETE
216 
                result.remove(instruction.dataIndex());
217 
            }
218 
        }
219 1 1. updateData : replaced return value with Collections.emptyList for com/renomad/minum/database/DatabaseConsolidator::updateData → KILLED
 
        return result.values();
220 
    }
221
222 
    /**
223 
     * This method will group the instructions for changes to the database by which
224 
     * consolidated files they apply to, so that we only need to make one change
225 
     * to each file.  Files are named like this: 1, 1001, etc., or
226 
     * in other words, the starting index of each set of consolidated data.
227 
     * @param databaseChangeInstructionMap this is a map between keys representing the
228 
     *                                     index of the data, and the data itself.
229 
     * @return a map consisting of keys representing the target file for the data, and
230 
     * a collection of DatabaseChangeInstruction data to place in that file.
231 
     */
232 
    private Map<Long, Collection<DatabaseChangeInstruction>> groupInstructionsByPartition(
233 
            Map<Long, DatabaseChangeInstruction> databaseChangeInstructionMap) {
234
235 
        // initialize a data structure to store our results
236 
        Map<Long, Collection<DatabaseChangeInstruction>> instructionsGroupedByPartition = new HashMap<>();
237
238 
        // loop through the incoming data, grouping and ordering as necessary
239 
        for (var databaseChangeInstruction : databaseChangeInstructionMap.entrySet()) {
240
241 
            // determine the expected filename for this file.  For example, if the index is 1234, then
242 
            // the filename should be 1001
243 4 1. groupInstructionsByPartition : Replaced long subtraction with addition → TIMED_OUT
2. groupInstructionsByPartition : Replaced long multiplication with division → TIMED_OUT
3. groupInstructionsByPartition : Replaced long division with multiplication → TIMED_OUT
4. groupInstructionsByPartition : Replaced long addition with subtraction → KILLED
 
            long expectedFilename = (((databaseChangeInstruction.getKey() - 1) / maxLinesPerFile) * maxLinesPerFile) + 1;
244
245 
            // If there is no key found, we need to add one, and add a new collection
246 1 1. lambda$groupInstructionsByPartition$7 : replaced return value with Collections.emptyList for com/renomad/minum/database/DatabaseConsolidator::lambda$groupInstructionsByPartition$7 → KILLED
 
            instructionsGroupedByPartition.computeIfAbsent(expectedFilename, x -> new ArrayList<>());
247
248 
            // add a new item to the collection for this filename
249 
            instructionsGroupedByPartition.get(expectedFilename).add(databaseChangeInstruction.getValue());
250 
        }
251
252 1 1. groupInstructionsByPartition : replaced return value with Collections.emptyMap for com/renomad/minum/database/DatabaseConsolidator::groupInstructionsByPartition → KILLED
 
        return instructionsGroupedByPartition;
253 
    }
254
255 
    /**
256 
     * read first 6 characters - is it update or delete?
257 
     * skip a character
258 
     * read digits until we hit a pipe symbol, that's our index.
259 
     * read the rest of the content
260 
     */
261 
    static DatabaseChangeInstruction parseDatabaseChangeInstructionString(String databaseInstructionString, String filename) {
262 
        String actionString = databaseInstructionString.substring(0, 6);
263 
        DatabaseChangeAction action;
264 1 1. parseDatabaseChangeInstructionString : negated conditional → KILLED
 
        if ("UPDATE".equals(actionString)) {
265 
            action = DatabaseChangeAction.UPDATE;
266 1 1. parseDatabaseChangeInstructionString : negated conditional → KILLED
 
        } else if ("DELETE".equals(actionString)) {
267 
            action = DatabaseChangeAction.DELETE;
268 
        } else {
269 
            throw new DbException("Line in append-only log was missing an action (UPDATE or DELETE) in the first characters. Line was: " + databaseInstructionString);
270 
        }
271 
        // confusing overlap of terms - index is used here to mean two things:
272 
        // a) where we find the first pipe symbol
273 
        // b) the index value of the data
274 
        int indexOfPipe = databaseInstructionString.indexOf('|', 7);
275 1 1. parseDatabaseChangeInstructionString : negated conditional → KILLED
 
        if (indexOfPipe == -1) {
276 
            throw new DbException(
277 
                    "Failed to find index of the first pipe in the file %s, with content %s".formatted(filename, databaseInstructionString));
278 
        }
279 
        String dataIndex = databaseInstructionString.substring(7, indexOfPipe);
280 
        long dataIndexLong = Long.parseLong(dataIndex);
281
282 1 1. parseDatabaseChangeInstructionString : replaced return value with null for com/renomad/minum/database/DatabaseConsolidator::parseDatabaseChangeInstructionString → KILLED
 
        return new DatabaseChangeInstruction(action, dataIndexLong, databaseInstructionString.substring(7));
283 
    }
284
285 
    /**
286 
     * Given a directory, convert the list of files into a sorted
287 
     * list of dates.
288 
     * @return a sorted list of dates, or an empty list if nothing found
289 
     */
290 
    static List<Date> getSortedAppendLogs(Path appendLogDirectory) {
291 
        // get the list of file names, which are date-time stamps
292 
        String[] fileList = appendLogDirectory.toFile().list();
293
294 
        // if there aren't any append-only files, bail out with an empty list
295 1 1. getSortedAppendLogs : negated conditional → KILLED
 
        if (fileList == null) {
296 
            return List.of();
297 
        }
298
299 
        List<Date> appendLogDates = convertFileListToDateList(fileList);
300
301 
        // sort
302 1 1. getSortedAppendLogs : replaced return value with Collections.emptyList for com/renomad/minum/database/DatabaseConsolidator::getSortedAppendLogs → KILLED
 
        return appendLogDates.stream().sorted().toList();
303 
    }
304
305 
    /**
306 
     * Convert a list of filenames to a list of dates
307 
     */
308 
    static List<Date> convertFileListToDateList(String[] listOfFiles) {
309 
        // initialize a list which will hold the dates associated with each file name
310 
        List<Date> appendLogDates = new ArrayList<>();
311
312 
        // convert the names to dates
313 
        for (String file : listOfFiles) {
314 
            Date date;
315 
            try {
316 
                date = simpleDateFormat.parse(file);
317 
            } catch (ParseException e) {
318 
                throw new DbException(e);
319 
            }
320 
            appendLogDates.add(date);
321 
        }
322
323 1 1. convertFileListToDateList : replaced return value with Collections.emptyList for com/renomad/minum/database/DatabaseConsolidator::convertFileListToDateList → KILLED
 
        return appendLogDates;
324 
    }
325 
}

Mutations

59

1.1
Location : <init>
Killed by : com.renomad.minum.database.DbEngine2Tests.testWriteDeserializationComplaints(com.renomad.minum.database.DbEngine2Tests) removed call to com/renomad/minum/utils/FileUtils::makeDirectory → KILLED

69

1.1
Location : consolidate
Killed by : none negated conditional → TIMED_OUT

86

1.1
Location : consolidate
Killed by : com.renomad.minum.database.DbFileConverterTests.testConvertDbEngine2FolderStructureToDbClassicForm_EdgeCase_CorruptData(com.renomad.minum.database.DbFileConverterTests) removed call to com/renomad/minum/database/DatabaseConsolidator::processAppendLogFile → KILLED

125

1.1
Location : processAppendLogFile
Killed by : com.renomad.minum.database.DbEngine2Tests.test_LoadingData_MultipleThreads(com.renomad.minum.database.DbEngine2Tests) removed call to com/renomad/minum/database/DatabaseConsolidator::rewriteFiles → KILLED

128

1.1
Location : processAppendLogFile
Killed by : com.renomad.minum.database.DbEngine2Tests.testChecksums_ChecksumsMissingAtLoad(com.renomad.minum.database.DbEngine2Tests) removed call to java/nio/file/Files::delete → KILLED

141

1.1
Location : rewriteFiles
Killed by : none Replaced long addition with subtraction → TIMED_OUT

2.2
Location : rewriteFiles
Killed by : com.renomad.minum.database.DbEngine2Tests.testChecksums_FailureDuringRewrite(com.renomad.minum.database.DbEngine2Tests) Replaced integer subtraction with addition → KILLED

147

1.1
Location : rewriteFiles
Killed by : com.renomad.minum.database.DbEngine2Tests.test_LoadingData_MultipleThreads(com.renomad.minum.database.DbEngine2Tests) negated conditional → KILLED

177

1.1
Location : readConsolidatedFileWithChecksum
Killed by : com.renomad.minum.security.TheBrigTests replaced return value with Collections.emptyList for com/renomad/minum/database/DatabaseConsolidator::readConsolidatedFileWithChecksum → KILLED

197

1.1
Location : updateData
Killed by : com.renomad.minum.database.DatabaseConsolidatorTests.testUpdatingData_EdgeCase_ParsingErrorForIndex(com.renomad.minum.database.DatabaseConsolidatorTests) negated conditional → KILLED

212

1.1
Location : updateData
Killed by : com.renomad.minum.database.DbEngine2Tests.test_LoadingData_MultipleThreads(com.renomad.minum.database.DbEngine2Tests) negated conditional → KILLED

219

1.1
Location : updateData
Killed by : com.renomad.minum.database.DbEngine2Tests.test_LoadingData_MultipleThreads(com.renomad.minum.database.DbEngine2Tests) replaced return value with Collections.emptyList for com/renomad/minum/database/DatabaseConsolidator::updateData → KILLED

243

1.1
Location : groupInstructionsByPartition
Killed by : none Replaced long subtraction with addition → TIMED_OUT

2.2
Location : groupInstructionsByPartition
Killed by : none Replaced long multiplication with division → TIMED_OUT

3.3
Location : groupInstructionsByPartition
Killed by : none Replaced long division with multiplication → TIMED_OUT

4.4
Location : groupInstructionsByPartition
Killed by : com.renomad.minum.database.DbEngine2Tests.testChecksums_FailureDuringRewrite(com.renomad.minum.database.DbEngine2Tests) Replaced long addition with subtraction → KILLED

246

1.1
Location : lambda$groupInstructionsByPartition$7
Killed by : com.renomad.minum.database.DbEngine2Tests.test_LoadingData_MultipleThreads(com.renomad.minum.database.DbEngine2Tests) replaced return value with Collections.emptyList for com/renomad/minum/database/DatabaseConsolidator::lambda$groupInstructionsByPartition$7 → KILLED

252

1.1
Location : groupInstructionsByPartition
Killed by : com.renomad.minum.database.DbEngine2Tests.test_LoadingData_MultipleThreads(com.renomad.minum.database.DbEngine2Tests) replaced return value with Collections.emptyMap for com/renomad/minum/database/DatabaseConsolidator::groupInstructionsByPartition → KILLED

264

1.1
Location : parseDatabaseChangeInstructionString
Killed by : com.renomad.minum.database.DatabaseConsolidatorTests.testParsingDatabaseChangeStrings_EdgeCase_InvalidAction(com.renomad.minum.database.DatabaseConsolidatorTests) negated conditional → KILLED

266

1.1
Location : parseDatabaseChangeInstructionString
Killed by : com.renomad.minum.database.DatabaseConsolidatorTests.testParsingDatabaseChangeStrings_EdgeCase_InvalidAction(com.renomad.minum.database.DatabaseConsolidatorTests) negated conditional → KILLED

275

1.1
Location : parseDatabaseChangeInstructionString
Killed by : com.renomad.minum.database.DbFileConverterTests.testConvertDbEngine2FolderStructureToDbClassicForm_EdgeCase_CorruptData(com.renomad.minum.database.DbFileConverterTests) negated conditional → KILLED

282

1.1
Location : parseDatabaseChangeInstructionString
Killed by : com.renomad.minum.database.DbEngine2Tests.test_LoadingData_MultipleThreads(com.renomad.minum.database.DbEngine2Tests) replaced return value with null for com/renomad/minum/database/DatabaseConsolidator::parseDatabaseChangeInstructionString → KILLED

295

1.1
Location : getSortedAppendLogs
Killed by : com.renomad.minum.database.DatabaseConsolidatorTests.testGetAppendFiles_EdgeCase_NoFiles(com.renomad.minum.database.DatabaseConsolidatorTests) negated conditional → KILLED

302

1.1
Location : getSortedAppendLogs
Killed by : com.renomad.minum.database.DbFileConverterTests.testConvertDbEngine2FolderStructureToDbClassicForm_EdgeCase_CorruptData(com.renomad.minum.database.DbFileConverterTests) replaced return value with Collections.emptyList for com/renomad/minum/database/DatabaseConsolidator::getSortedAppendLogs → KILLED

323

1.1
Location : convertFileListToDateList
Killed by : com.renomad.minum.database.DbFileConverterTests.testConvertDbEngine2FolderStructureToDbClassicForm_EdgeCase_CorruptData(com.renomad.minum.database.DbFileConverterTests) replaced return value with Collections.emptyList for com/renomad/minum/database/DatabaseConsolidator::convertFileListToDateList → KILLED

Active mutators

Tests examined


Report generated by PIT 1.17.0