diff --git a/CHANGES.txt b/CHANGES.txt index 002ca6001d..57c0667730 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,3 +1,12 @@ +0.4.1 + * Fix FlushPeriod columnfamily configuration regression + (CASSANDRA-455) + * Fix long column name support (CASSANDRA-460) + * Fix for serializing a row that only contains tombstones + (CASSANDRA-458) + * Fix for discarding unneeded commitlog segments (CASSANDRA-459) + + 0.4.0 * fix get_key_range problems when a node is down (CASSANDRA-440) and add UnavailableException to more Thrift methods diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index c376a42e57..2df5b61c48 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -538,12 +538,14 @@ public class DatabaseDescriptor CFMetaData data = new CFMetaData(); data.columnType = "Standard"; data.comparator = new UTF8Type(); + data.flushPeriodInMinutes = 1; systemMetadata.put(SystemTable.LOCATION_CF, data); data = new CFMetaData(); data.columnType = "Super"; data.comparator = new UTF8Type(); data.subcolumnComparator = new BytesType(); + data.flushPeriodInMinutes = 10; systemMetadata.put(HintedHandOffManager.HINTS_CF, data); tableToCFMetaDataMap_.put(Table.SYSTEM_TABLE, systemMetadata); diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index c7c696a529..d9d8c807f4 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -165,6 +165,8 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean void onStart() throws IOException { + if (logger_.isDebugEnabled()) + logger_.debug("Starting CFS " + columnFamily_); // scan for data files corresponding to this CF List sstableFiles = new ArrayList(); String[] dataFileDirectories = DatabaseDescriptor.getAllDataFileLocationsForTable(table_); diff --git a/src/java/org/apache/cassandra/db/CommitLog.java b/src/java/org/apache/cassandra/db/CommitLog.java index b862fab213..8c72cf90da 100644 --- a/src/java/org/apache/cassandra/db/CommitLog.java +++ b/src/java/org/apache/cassandra/db/CommitLog.java @@ -35,8 +35,6 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; -import org.apache.commons.lang.StringUtils; - /* * Commit Log tracks every write operation into the system. The aim * of the commit log is to be able to successfully recover data that was @@ -96,6 +94,15 @@ public class CommitLog { return (position != -1L); } + + @Override + public String toString() + { + return "CommitLogContext(" + + "file='" + file + '\'' + + ", position=" + position + + ')'; + } } public static class CommitLogFileComparator implements Comparator @@ -359,7 +366,7 @@ public class CommitLog for (ColumnFamily columnFamily : rm.getColumnFamilies()) { int id = table.getColumnFamilyId(columnFamily.name()); - if (!clHeader_.isDirty(id) || (clHeader_.isDirty(id) && clHeader_.getPosition(id) == 0)) + if (!clHeader_.isDirty(id)) { clHeader_.turnOn(id, logWriter_.getFilePointer()); seekAndWriteCommitLogHeader(clHeader_.toByteArray()); @@ -455,77 +462,83 @@ public class CommitLog */ private void discardCompletedSegments(CommitLog.CommitLogContext cLogCtx, int id) throws IOException { + if (logger_.isDebugEnabled()) + logger_.debug("discard completed log segments for " + cLogCtx + ", column family " + id + ". CFIDs are " + Table.TableMetadata.getColumnFamilyIDString()); /* retrieve the commit log header associated with the file in the context */ - CommitLogHeader commitLogHeader = clHeaders_.get(cLogCtx.file); - if(commitLogHeader == null ) + if (clHeaders_.get(cLogCtx.file) == null) { - if( logFile_.equals(cLogCtx.file) ) + if (logFile_.equals(cLogCtx.file)) { /* this means we are dealing with the current commit log. */ - commitLogHeader = clHeader_; clHeaders_.put(cLogCtx.file, clHeader_); } else + { + logger_.error("Unknown commitlog file " + cLogCtx.file); return; + } } /* * log replay assumes that we only have to look at entries past the last * flush position, so verify that this flush happens after the last. - * (Currently Memtables are flushed on a single thread so this should be fine.) */ - assert cLogCtx.position >= commitLogHeader.getPosition(id); + assert cLogCtx.position >= clHeaders_.get(cLogCtx.file).getPosition(id); - commitLogHeader.turnOff(id); /* Sort the commit logs based on creation time */ List oldFiles = new ArrayList(clHeaders_.keySet()); Collections.sort(oldFiles, new CommitLogFileComparator()); - List listOfDeletedFiles = new ArrayList(); + /* * Loop through all the commit log files in the history. Now process * all files that are older than the one in the context. For each of - * these files the header needs to modified by performing a bitwise & - * of the header with the header of the file in the context. If we - * encounter the file in the context in our list of old commit log files - * then we update the header and write it back to the commit log. + * these files the header needs to modified by resetting the dirty + * bit corresponding to the flushed CF. */ for (String oldFile : oldFiles) { + CommitLogHeader header = clHeaders_.get(oldFile); if (oldFile.equals(cLogCtx.file)) { - /* - * We need to turn on again. This is because we always keep - * the bit turned on and the position indicates from where the - * commit log needs to be read. When a flush occurs we turn off - * perform & operation and then turn on with the new position. - */ - commitLogHeader.turnOn(id, cLogCtx.position); - seekAndWriteCommitLogHeader(commitLogHeader.toByteArray()); - break; - } - else - { - CommitLogHeader oldCommitLogHeader = clHeaders_.get(oldFile); - oldCommitLogHeader.and(commitLogHeader); - if (oldCommitLogHeader.isSafeToDelete()) + // we can't just mark the segment where the flush happened clean, + // since there may have been writes to it between when the flush + // started and when it finished. so mark the flush position as + // the replay point for this CF, instead. + if (logger_.isDebugEnabled()) + logger_.debug("Marking replay position " + cLogCtx.position + " on commit log " + oldFile); + header.turnOn(id, cLogCtx.position); + if (oldFile.equals(logFile_)) { - logger_.info("Deleting obsolete commit log:" + oldFile); - FileUtils.deleteAsync(oldFile); - listOfDeletedFiles.add(oldFile); + seekAndWriteCommitLogHeader(header.toByteArray()); } else { - BufferedRandomAccessFile logWriter = CommitLog.createWriter(oldFile); - writeCommitLogHeader(logWriter, oldCommitLogHeader.toByteArray()); - logWriter.close(); + writeOldCommitLogHeader(oldFile, header); } + break; + } + + header.turnOff(id); + if (header.isSafeToDelete()) + { + logger_.info("Deleting obsolete commit log:" + oldFile); + FileUtils.deleteAsync(oldFile); + clHeaders_.remove(oldFile); + } + else + { + if (logger_.isDebugEnabled()) + logger_.debug("Not safe to delete commit log " + oldFile + "; dirty is " + header.dirtyString()); + writeOldCommitLogHeader(oldFile, header); } } + } - for ( String deletedFile : listOfDeletedFiles) - { - clHeaders_.remove(deletedFile); - } + private void writeOldCommitLogHeader(String oldFile, CommitLogHeader header) throws IOException + { + BufferedRandomAccessFile logWriter = CommitLog.createWriter(oldFile); + writeCommitLogHeader(logWriter, header.toByteArray()); + logWriter.close(); } private boolean maybeRollLog() throws IOException @@ -541,10 +554,7 @@ public class CommitLog logWriter_ = CommitLog.createWriter(logFile_); /* squirrel away the old commit log header */ clHeaders_.put(oldLogFile, new CommitLogHeader(clHeader_)); - // we leave the old 'dirty' bits alone, so we can test for - // whether it's safe to remove a given log segment by and-ing its dirty - // with the current one. - clHeader_.zeroPositions(); + clHeader_.clear(); writeCommitLogHeader(logWriter_, clHeader_.toByteArray()); return true; } diff --git a/src/java/org/apache/cassandra/db/CommitLogHeader.java b/src/java/org/apache/cassandra/db/CommitLogHeader.java index 1b1cba6554..3832e9a79e 100644 --- a/src/java/org/apache/cassandra/db/CommitLogHeader.java +++ b/src/java/org/apache/cassandra/db/CommitLogHeader.java @@ -20,6 +20,7 @@ package org.apache.cassandra.db; import java.io.*; import java.util.BitSet; +import java.util.Arrays; import org.apache.cassandra.io.ICompactSerializer; import org.apache.cassandra.io.DataInputBuffer; @@ -35,17 +36,6 @@ class CommitLogHeader return serializer; } - public static BitSet and(byte[] bytes1, byte[] bytes2) throws IOException - { - DataInputBuffer bufIn = new DataInputBuffer(); - bufIn.reset(bytes1, 0, bytes1.length); - CommitLogHeader header1 = serializer.deserialize(bufIn); - bufIn.reset(bytes2, 0, bytes2.length); - CommitLogHeader header2 = serializer.deserialize(bufIn); - header1.and(header2); - return header1.dirty; - } - static int getLowestPosition(CommitLogHeader clHeader) { int minPosition = Integer.MAX_VALUE; @@ -116,17 +106,12 @@ class CommitLogHeader return dirty.isEmpty(); } - void zeroPositions() + void clear() { - int size = lastFlushedAt.length; - lastFlushedAt = new int[size]; + dirty.clear(); + Arrays.fill(lastFlushedAt, 0); } - - void and(CommitLogHeader commitLogHeader) - { - dirty.and(commitLogHeader.dirty); - } - + byte[] toByteArray() throws IOException { ByteArrayOutputStream bos = new ByteArrayOutputStream(); @@ -154,6 +139,19 @@ class CommitLogHeader return sb.toString(); } + public String dirtyString() + { + StringBuilder sb = new StringBuilder(); + for (int i = 0; i < dirty.length(); i++) + { + if (dirty.get(i)) + { + sb.append(i).append(", "); + } + } + return sb.toString(); + } + static class CommitLogHeaderSerializer implements ICompactSerializer { public void serialize(CommitLogHeader clHeader, DataOutputStream dos) throws IOException diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java index 391fe7180e..a7e0efe742 100644 --- a/src/java/org/apache/cassandra/db/Table.java +++ b/src/java/org/apache/cassandra/db/Table.java @@ -140,24 +140,18 @@ public class Table public String toString() { - StringBuilder sb = new StringBuilder(""); - Set cfNames = cfIdMap_.keySet(); - - for ( String cfName : cfNames ) - { - sb.append(cfName); - sb.append("---->"); - sb.append(cfIdMap_.get(cfName)); - sb.append(System.getProperty("line.separator")); - } - - return sb.toString(); + return "TableMetadata(" + FBUtilities.mapToString(cfIdMap_) + ")"; } public static int getColumnFamilyCount() { return idCfMap_.size(); } + + public static String getColumnFamilyIDString() + { + return FBUtilities.mapToString(tableMetadataMap_); + } } /** diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java index 0d6b40d2eb..b85f57ac45 100644 --- a/src/java/org/apache/cassandra/utils/FBUtilities.java +++ b/src/java/org/apache/cassandra/utils/FBUtilities.java @@ -25,11 +25,7 @@ import java.net.UnknownHostException; import java.security.MessageDigest; import java.text.DateFormat; import java.text.SimpleDateFormat; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Date; -import java.util.List; -import java.util.StringTokenizer; +import java.util.*; import java.util.zip.DataFormatException; import java.util.zip.Deflater; import java.util.zip.Inflater; @@ -412,4 +408,16 @@ public class FBUtilities } return new String(chars); } + + public static String mapToString(Map map) + { + StringBuilder sb = new StringBuilder(""); + + for (Map.Entry entry : map.entrySet()) + { + sb.append(entry.getKey()).append(": ").append(entry.getValue()).append(", "); + } + + return sb.append("}").toString(); + } }