natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

merge from 0.4 branch 

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@819826 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-09-29 03:42:42 +00:00
parent de8318315f
commit fb5082ab3b
7 changed files with 104 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
CHANGES.txt
View File

@ -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

2
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -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);

2
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -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<File> sstableFiles = new ArrayList<File>();
String[] dataFileDirectories = DatabaseDescriptor.getAllDataFileLocationsForTable(table_);

98
src/java/org/apache/cassandra/db/CommitLog.java
View File

@ -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<String>
@ -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<String> oldFiles = new ArrayList<String>(clHeaders_.keySet());
Collections.sort(oldFiles, new CommitLogFileComparator());
List<String> listOfDeletedFiles = new ArrayList<String>();
/*
* 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;
}

38
src/java/org/apache/cassandra/db/CommitLogHeader.java
View File

@ -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<CommitLogHeader>
{
public void serialize(CommitLogHeader clHeader, DataOutputStream dos) throws IOException

18
src/java/org/apache/cassandra/db/Table.java
View File

@ -140,24 +140,18 @@ public class Table
public String toString()
{
StringBuilder sb = new StringBuilder("");
Set<String> 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_);
}
}
/**

18
src/java/org/apache/cassandra/utils/FBUtilities.java
View File

@ -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();
}
}