001
002 /*
003 * Copyright (C) 2011 Archie L. Cobbs. All rights reserved.
004 *
005 * $Id: AbstractSchemaUpdater.java 273 2012-02-02 21:52:15Z archie.cobbs $
006 */
007
008 package org.dellroad.stuff.schema;
009
010 import java.util.ArrayList;
011 import java.util.Collection;
012 import java.util.Collections;
013 import java.util.Comparator;
014 import java.util.HashSet;
015 import java.util.Iterator;
016 import java.util.LinkedHashSet;
017 import java.util.List;
018 import java.util.Set;
019 import java.util.TreeMap;
020 import java.util.TreeSet;
021
022 import org.dellroad.stuff.graph.TopologicalSorter;
023 import org.slf4j.Logger;
024 import org.slf4j.LoggerFactory;
025
026 /**
027 * Handles the initialization and schema maintenance of a database.
028 *
029 * <p>
030 * In this class, a <b>database</b> is some stateful object whose structure and/or content may need to change over time.
031 * <b>Updates</b> are uniquely named objects capable of making such changes. Databases are also capable of storing the
032 * names of the already-applied updates.
033 *
034 * <p>
035 * Given a database and a set of current updates, this class will ensure that a database is initialized if necessary
036 * and up-to-date with respect to the updates.
037 *
038 * <p>
039 * The primary method is {@link #initializeAndUpdateDatabase initializeAndUpdateDatabase()}, which will:
040 * <ul>
041 * <li>Initialize an {@linkplain #databaseNeedsInitialization empty} database (if necessary);</li>
042 * <li>Apply any outstanding {@link SchemaUpdate}s as needed, ordered properly according to
043 * their {@linkplain SchemaUpdate#getRequiredPredecessors predecessor constraints}; and</li>
044 * <li>Keep track of which {@link SchemaUpdate}s have already been applied across restarts.</li>
045 * </ul>
046 * </p>
047 *
048 * @param <D> database type
049 * @param <T> database transaction type
050 */
051 public abstract class AbstractSchemaUpdater<D, T> {
052
053 protected final Logger log = LoggerFactory.getLogger(this.getClass());
054
055 private Collection<? extends SchemaUpdate<T>> updates;
056 private boolean ignoreUnrecognizedUpdates;
057
058 /**
059 * Get the configured updates. This property is required.
060 *
061 * @return configured updates
062 * @see #setUpdates setUpdates()
063 */
064 public Collection<? extends SchemaUpdate<T>> getUpdates() {
065 return this.updates;
066 }
067
068 /**
069 * Configure the updates.
070 * This should be the set of all updates that may need to be applied to the database.
071 *
072 * <p>
073 * For any given application, ideally this set should be "write only" in the sense that once an update is added to the set
074 * and applied to one or more actual databases, the update and its name should thereafter never change. Otherwise,
075 * it would be possible for different databases to have inconsistent schemas even though the same updates were recorded.
076 *
077 * <p>
078 * Furthermore, if not configured to {@linkplain #setIgnoreUnrecognizedUpdates ignore unrecognized updates already applied}
079 * (the default behavior), then updates must never be removed from this set as the application evolves;
080 * see {@link #setIgnoreUnrecognizedUpdates} for more information on the rationale.
081 *
082 * @param updates all updates; each update must have a unique {@link SchemaUpdate#getName name}.
083 * @see #getUpdates
084 * @see #setIgnoreUnrecognizedUpdates
085 */
086 public void setUpdates(Collection<? extends SchemaUpdate<T>> updates) {
087 this.updates = updates;
088 }
089
090 /**
091 * Determine whether unrecognized updates are ignored or cause an exception.
092 *
093 * @return true if unrecognized updates should be ignored, false if they should cause an exception to be thrown
094 * @see #setIgnoreUnrecognizedUpdates setIgnoreUnrecognizedUpdates()
095 */
096 public boolean isIgnoreUnrecognizedUpdates() {
097 return this.ignoreUnrecognizedUpdates;
098 }
099
100 /**
101 * Configure behavior when an unknown update is registered as having already been applied in the database.
102 *
103 * <p>
104 * The default behavior is <code>false</code>, which results in an exception being thrown. This protects against
105 * accidental downgrades (i.e., running older code against a database with a newer schema), which are not supported.
106 * However, this also requires that all updates that might ever possibly have been applied to the database be
107 * present in the set of configured updates.
108 *
109 * <p>
110 * Setting this to <code>true</code> will result in unrecognized updates simply being ignored.
111 * This setting loses the downgrade protection but allows obsolete schema updates to be dropped over time.
112 *
113 * @param ignoreUnrecognizedUpdates whether to ignore unrecognized updates
114 * @see #isIgnoreUnrecognizedUpdates
115 */
116 public void setIgnoreUnrecognizedUpdates(boolean ignoreUnrecognizedUpdates) {
117 this.ignoreUnrecognizedUpdates = ignoreUnrecognizedUpdates;
118 }
119
120 /**
121 * Perform database schema initialization and updates.
122 *
123 * <p>
124 * This method applies the following logic: if the {@linkplain #databaseNeedsInitialization database needs initialization},
125 * then {@linkplain #initializeDatabase initialize the database} and {@linkplain #recordUpdateApplied record} each update
126 * as having been applied; otherwise, {@linkplain #apply apply} any {@linkplain #getAppliedUpdateNames unapplied updates}
127 * as needed.
128 *
129 * <p>
130 * Note this implies the database initialization must initialize the database to its current, up-to-date state
131 * (with respect to the set of all available updates), not its original, pre-update state.
132 *
133 * <p>
134 * The database initialization step, and each of the update steps, is {@linkplain #applyInTransaction performed within
135 * its own transaction}.
136 *
137 * @param database the database to initialize (if necessary) and update
138 * @throws Exception if an update fails
139 * @throws IllegalStateException if this instance is not configured to {@linkplain #setIgnoreUnrecognizedUpdates ignore
140 * unrecognized updates} and an unrecognized update has already been applied
141 * @throws IllegalArgumentException if two configured updates have the same name
142 * @throws IllegalArgumentException if any configured update has a required predecessor which is not also a configured update
143 * (i.e., if the updates are not transitively closed under predecessors)
144 */
145 public synchronized void initializeAndUpdateDatabase(D database) throws Exception {
146
147 // Log
148 this.log.info("verifying database");
149
150 // First, initialize if necessary
151 this.applyInTransaction(database, new DatabaseAction<T>() {
152 @Override
153 public void apply(T transaction) throws Exception {
154
155 // Already initialized?
156 if (!AbstractSchemaUpdater.this.databaseNeedsInitialization(transaction)) {
157 AbstractSchemaUpdater.this.log.debug("detected initialized database");
158 return;
159 }
160
161 // Initialize database
162 AbstractSchemaUpdater.this.log.info("uninitialized database detected - initializing now");
163 AbstractSchemaUpdater.this.initializeDatabase(transaction);
164
165 // Record all schema updates as having already been applied
166 ArrayList<SchemaUpdate<T>> updateList = new ArrayList<SchemaUpdate<T>>(AbstractSchemaUpdater.this.getUpdates());
167 Collections.sort(updateList, new UpdateByNameComparator());
168 for (SchemaUpdate<T> update : updateList) {
169 for (String name : AbstractSchemaUpdater.this.getUpdateNames(update))
170 AbstractSchemaUpdater.this.recordUpdateApplied(transaction, name);
171 }
172 }
173 });
174
175 // Next, apply any new updates
176 this.applySchemaUpdates(database);
177
178 // Done
179 this.log.info("database verification complete");
180 }
181
182 /**
183 * Determine if the given schema update name is valid. Valid names are non-empty and
184 * have no leading or trailing whitespace.
185 */
186 public static boolean isValidUpdateName(String updateName) {
187 return updateName.length() > 0 && updateName.trim().length() == updateName.length();
188 }
189
190 /**
191 * Determine if the database needs initialization.
192 *
193 * <p>
194 * If so, {@link #initializeDatabase} will eventually be invoked.
195 *
196 * @param transaction open transaction
197 * @throws Exception if an error occurs while accessing the database
198 */
199 protected abstract boolean databaseNeedsInitialization(T transaction) throws Exception;
200
201 /**
202 * Initialize an uninitialized database. This should create and initialize the database schema and content,
203 * including whatever portion of that is used to track schema updates.
204 *
205 * @param transaction open transaction
206 * @throws Exception if an error occurs while accessing the database
207 */
208 protected abstract void initializeDatabase(T transaction) throws Exception;
209
210 /**
211 * Begin a transaction on the given database.
212 * The transaction will always eventually either be
213 * {@linkplain #commitTransaction committed} or {@linkplain #rollbackTransaction rolled back}.
214 *
215 * @param database database
216 * @return transaction handle
217 * @throws Exception if an error occurs while accessing the database
218 */
219 protected abstract T openTransaction(D database) throws Exception;
220
221 /**
222 * Commit a previously opened transaction.
223 *
224 * @param transaction open transaction previously returned from {@link #openTransaction openTransaction()}
225 * @throws Exception if an error occurs while accessing the database
226 */
227 protected abstract void commitTransaction(T transaction) throws Exception;
228
229 /**
230 * Roll back a previously opened transaction.
231 * This method will also be invoked if {@link #commitTransaction commitTransaction()} throws an exception.
232 *
233 * @param transaction open transaction previously returned from {@link #openTransaction openTransaction()}
234 * @throws Exception if an error occurs while accessing the database
235 */
236 protected abstract void rollbackTransaction(T transaction) throws Exception;
237
238 /**
239 * Determine which updates have already been applied to the database.
240 *
241 * @param transaction open transaction
242 * @throws Exception if an error occurs while accessing the database
243 */
244 protected abstract Set<String> getAppliedUpdateNames(T transaction) throws Exception;
245
246 /**
247 * Record an update as having been applied to the database.
248 *
249 * @param transaction open transaction
250 * @param name update name
251 * @throws IllegalStateException if the update has already been recorded in the database
252 * @throws Exception if an error occurs while accessing the database
253 */
254 protected abstract void recordUpdateApplied(T transaction, String name) throws Exception;
255
256 /**
257 * Determine the preferred ordering of two updates that do not have any predecessor constraints
258 * (including implied indirect constraints) between them.
259 *
260 * <p>
261 * The {@link Comparator} returned by the implementation in {@link AbstractSchemaUpdater} simply sorts updates by name.
262 * Subclasses may override if necessary.
263 *
264 * @return a {@link Comparator} that sorts incomparable updates in the order they should be applied
265 */
266 protected Comparator<SchemaUpdate<T>> getOrderingTieBreaker() {
267 return new UpdateByNameComparator();
268 }
269
270 /**
271 * Generate the update name for one action within a multi-action update.
272 *
273 * <p>
274 * The implementation in {@link AbstractSchemaUpdater} just adds a suffix using {@code index + 1}, padded to
275 * 5 digits, producing names like {@code name-00001}, {@code name-00002}, etc.
276 *
277 * @param update the schema update
278 * @param index the index of the action (zero based)
279 * @see SchemaUpdate#isSingleAction
280 */
281 protected String generateMultiUpdateName(SchemaUpdate<T> update, int index) {
282 return String.format("%s-%05d", update.getName(), index + 1);
283 }
284
285 /**
286 * Execute a database action within an existing transaction.
287 *
288 * <p>
289 * All database operations in {@link AbstractSchemaUpdater} are performed via this method;
290 * subclasses are encouraged to follow this pattern.
291 *
292 * <p>
293 * The implementation in {@link AbstractSchemaUpdater} simply invokes {@link DatabaseAction#apply action.apply()};
294 * subclasses may override if desired.
295 *
296 * @throws Exception if an error occurs while accessing the database
297 */
298 protected void apply(T transaction, DatabaseAction<T> action) throws Exception {
299 action.apply(transaction);
300 }
301
302 /**
303 * Execute a database action. A new transaction will be created, used, and closed.
304 * Delegates to {@link #apply apply()} for the actual execution of the action.
305 *
306 * <p>
307 * If the action or {@link #commitTransaction commitTransaction()} fails, the transaction
308 * is {@linkplain #rollbackTransaction rolled back}.
309 *
310 * @throws Exception if an error occurs while accessing the database
311 */
312 protected void applyInTransaction(D database, DatabaseAction<T> action) throws Exception {
313 T transaction = this.openTransaction(database);
314 boolean success = false;
315 try {
316 this.apply(transaction, action);
317 this.commitTransaction(transaction);
318 success = true;
319 } finally {
320 if (!success)
321 this.rollbackTransaction(transaction);
322 }
323 }
324
325 /**
326 * Apply schema updates to an initialized database.
327 */
328 private void applySchemaUpdates(D database) throws Exception {
329
330 // Sanity check
331 final HashSet<SchemaUpdate<T>> allUpdates = new HashSet<SchemaUpdate<T>>(this.getUpdates());
332 if (allUpdates == null)
333 throw new IllegalArgumentException("no updates configured");
334
335 // Create mapping from update name to update; multiple updates will have multiple names
336 TreeMap<String, SchemaUpdate<T>> updateMap = new TreeMap<String, SchemaUpdate<T>>();
337 for (SchemaUpdate<T> update : allUpdates) {
338 for (String updateName : this.getUpdateNames(update)) {
339 if (!isValidUpdateName(updateName))
340 throw new IllegalArgumentException("illegal schema update name `" + updateName + "'");
341 if (updateMap.put(updateName, update) != null)
342 throw new IllegalArgumentException("duplicate schema update name `" + updateName + "'");
343 }
344 }
345 this.log.debug("these are all known schema updates: " + updateMap.keySet());
346
347 // Verify updates are transitively closed under predecessor constraints
348 for (SchemaUpdate<T> update : allUpdates) {
349 for (SchemaUpdate<T> predecessor : update.getRequiredPredecessors()) {
350 if (!allUpdates.contains(predecessor)) {
351 throw new IllegalArgumentException("schema update `" + update.getName()
352 + "' has a required predecessor named `" + predecessor.getName() + "' that is not a configured update");
353 }
354 }
355 }
356
357 // Sort updates in the order we should to apply them
358 List<SchemaUpdate<T>> updateList = new TopologicalSorter<SchemaUpdate<T>>(allUpdates,
359 new SchemaUpdateEdgeLister<T>(), this.getOrderingTieBreaker()).sortEdgesReversed();
360
361 // Determine which updates have already been applied
362 final HashSet<String> appliedUpdateNames = new HashSet<String>();
363 this.applyInTransaction(database, new DatabaseAction<T>() {
364 @Override
365 public void apply(T transaction) throws Exception {
366 appliedUpdateNames.addAll(AbstractSchemaUpdater.this.getAppliedUpdateNames(transaction));
367 }
368 });
369 this.log.debug("these are the already-applied schema updates: " + appliedUpdateNames);
370
371 // Check whether any unknown updates have been applied
372 TreeSet<String> unknownUpdateNames = new TreeSet<String>(appliedUpdateNames);
373 unknownUpdateNames.removeAll(updateMap.keySet());
374 if (!unknownUpdateNames.isEmpty()) {
375 if (!this.isIgnoreUnrecognizedUpdates()) {
376 throw new IllegalStateException(unknownUpdateNames.size()
377 + " unrecognized update(s) have already been applied: " + unknownUpdateNames);
378 }
379 this.log.info("ignoring " + unknownUpdateNames.size()
380 + " unrecognized update(s) already applied: " + unknownUpdateNames);
381 }
382
383 // Remove the already-applied updates
384 updateMap.keySet().removeAll(appliedUpdateNames);
385 HashSet<SchemaUpdate<T>> remainingUpdates = new HashSet<SchemaUpdate<T>>(updateMap.values());
386 for (Iterator<SchemaUpdate<T>> i = updateList.iterator(); i.hasNext(); ) {
387 if (!remainingUpdates.contains(i.next()))
388 i.remove();
389 }
390
391 // Now are any updates needed?
392 if (updateList.isEmpty()) {
393 this.log.info("no schema updates are required");
394 return;
395 }
396
397 // Log which updates we're going to apply
398 final LinkedHashSet<String> remainingUpdateNames = new LinkedHashSet<String>(updateMap.size());
399 for (SchemaUpdate<T> update : updateList) {
400 ArrayList<String> updateNames = this.getUpdateNames(update);
401 updateNames.removeAll(appliedUpdateNames);
402 remainingUpdateNames.addAll(updateNames);
403 }
404 this.log.info("applying " + remainingUpdateNames.size() + " schema update(s): " + remainingUpdateNames);
405
406 // Apply and record each unapplied update
407 for (SchemaUpdate<T> nextUpdate : updateList) {
408 final RecordingUpdateHandler updateHandler = new RecordingUpdateHandler(nextUpdate, remainingUpdateNames);
409 this.applyInTransaction(database, new DatabaseAction<T>() {
410 @Override
411 public void apply(T transaction) throws Exception {
412 updateHandler.process(transaction);
413 }
414 });
415 }
416 }
417
418 // Get all update names, expanding multi-updates as necessary
419 private ArrayList<String> getUpdateNames(SchemaUpdate<T> update) throws Exception {
420 final ArrayList<String> names = new ArrayList<String>();
421 UpdateHandler updateHandler = new UpdateHandler(update) {
422 @Override
423 protected void handleSingleUpdate(T transaction, DatabaseAction<T> action) {
424 names.add(this.update.getName());
425 }
426 @Override
427 protected void handleMultiUpdate(T transaction, DatabaseAction<T> action, int index) {
428 names.add(AbstractSchemaUpdater.this.generateMultiUpdateName(this.update, index));
429 }
430 };
431 updateHandler.process(null);
432 return names;
433 }
434
435 // Apply and record an update, all within a single transaction
436 private void applyAndRecordUpdate(T transaction, String name, final DatabaseAction<T> action) throws Exception {
437 if (action != null) {
438 this.log.info("applying schema update `" + name + "'");
439 this.apply(transaction, action);
440 } else
441 this.log.info("recording empty schema update `" + name + "'");
442 this.recordUpdateApplied(transaction, name);
443 }
444
445 private class RecordingUpdateHandler extends UpdateHandler {
446
447 private final Set<String> remainingUpdateNames;
448
449 public RecordingUpdateHandler(SchemaUpdate<T> update, Set<String> remainingUpdateNames) {
450 super(update);
451 this.remainingUpdateNames = remainingUpdateNames;
452 }
453
454 @Override
455 protected void handleEmptyUpdate(T transaction) throws Exception {
456 assert this.remainingUpdateNames.contains(this.update.getName());
457 AbstractSchemaUpdater.this.applyAndRecordUpdate(transaction, this.update.getName(), null);
458 }
459
460 @Override
461 protected void handleSingleUpdate(T transaction, DatabaseAction<T> action) throws Exception {
462 assert this.remainingUpdateNames.contains(this.update.getName());
463 AbstractSchemaUpdater.this.applyAndRecordUpdate(transaction, this.update.getName(), action);
464 }
465
466 @Override
467 protected void handleSingleMultiUpdate(T transaction, final List<? extends DatabaseAction<T>> actions)
468 throws Exception {
469 assert this.remainingUpdateNames.contains(this.update.getName());
470 AbstractSchemaUpdater.this.applyAndRecordUpdate(transaction, this.update.getName(), new DatabaseAction<T>() {
471 @Override
472 public void apply(T transaction) throws Exception {
473 for (DatabaseAction<T> action : actions)
474 AbstractSchemaUpdater.this.apply(transaction, action);
475 }
476 });
477 }
478
479 @Override
480 protected void handleMultiUpdate(T transaction, DatabaseAction<T> action, int index) throws Exception {
481 String updateName = AbstractSchemaUpdater.this.generateMultiUpdateName(this.update, index);
482 if (!this.remainingUpdateNames.contains(updateName)) // a partially completed multi-update
483 return;
484 AbstractSchemaUpdater.this.applyAndRecordUpdate(transaction, updateName, action);
485 }
486 }
487
488 // Adapter class for handling updates of various types
489 private class UpdateHandler {
490
491 protected final SchemaUpdate<T> update;
492
493 private final List<? extends DatabaseAction<T>> actions;
494
495 public UpdateHandler(SchemaUpdate<T> update) {
496 this.update = update;
497 this.actions = update.getDatabaseActions();
498 }
499
500 public final void process(T transaction) throws Exception {
501 switch (this.actions.size()) {
502 case 0:
503 this.handleEmptyUpdate(transaction);
504 break;
505 case 1:
506 this.handleSingleUpdate(transaction, this.actions.get(0));
507 break;
508 default:
509 if (update.isSingleAction()) {
510 this.handleSingleMultiUpdate(transaction, actions);
511 break;
512 } else {
513 int index = 0;
514 for (DatabaseAction<T> action : this.actions)
515 this.handleMultiUpdate(transaction, action, index++);
516 }
517 break;
518 }
519 }
520
521 protected void handleEmptyUpdate(T transaction) throws Exception {
522 this.handleSingleUpdate(transaction, null);
523 }
524
525 protected void handleSingleUpdate(T transaction, DatabaseAction<T> action) throws Exception {
526 }
527
528 protected void handleSingleMultiUpdate(T transaction, List<? extends DatabaseAction<T>> actions) throws Exception {
529 this.handleSingleUpdate(transaction, null);
530 }
531
532 protected void handleMultiUpdate(T transaction, DatabaseAction<T> action, int index) throws Exception {
533 }
534 }
535
536 // Sorts updates by name
537 private class UpdateByNameComparator implements Comparator<SchemaUpdate<T>> {
538
539 @Override
540 public int compare(SchemaUpdate<T> update1, SchemaUpdate<T> update2) {
541 return update1.getName().compareTo(update2.getName());
542 }
543 }
544 }
545