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

move data files into per-table subdirectories. patch by Arin Sarkissian; reviewed by Michael Greene and jbellis 

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@794784 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-07-16 19:10:06 +00:00
parent 715b60d147
commit 840a64dcb9
9 changed files with 131 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -406,6 +406,10 @@ public class DatabaseDescriptor
}
}
/* make sure we have a directory for each table */
createTableDirectories();
/* Load the seeds for node contact points */
String[] seeds = xmlUtils.getNodeValues("/Storage/Seeds/Seed");
for( int i = 0; i < seeds.length; ++i )
@ -425,8 +429,22 @@ public class DatabaseDescriptor
}
}
/**
* Create the table directory in each data directory
*/
public static void createTableDirectories() throws IOException
{
for (String dataFile : dataFileDirectories_)
{
FileUtils.createDirectory(dataFile + File.separator + Table.SYSTEM_TABLE);
for (String table : tables_)
{
FileUtils.createDirectory(dataFile + File.separator + table);
}
}
}
/*
/**
* Create the metadata tables. This table has information about
* the table name and the column families that make up the table.
* Each column family also has an associated ID which is an int.
@ -483,7 +501,6 @@ public class DatabaseDescriptor
return columnIndexSizeInKB_ * 1024;
}
public static int getMemtableLifetime()
{
return memtableLifetime_;
@ -670,16 +687,22 @@ public class DatabaseDescriptor
return dataFileDirectories_;
}
public static String getDataFileLocation()
public static String[] getAllDataFileLocationsForTable(String table)
{
String dataFileDirectory = dataFileDirectories_[currentIndex_];
return dataFileDirectory;
String[] tableLocations = new String[dataFileDirectories_.length];
for (int i = 0; i < dataFileDirectories_.length; i++)
{
tableLocations[i] = dataFileDirectories_[i] + File.separator + table;
}
return tableLocations;
}
public static String getCompactionFileLocation()
public static String getDataFileLocationForTable(String table)
{
String dataFileDirectory = dataFileDirectories_[currentIndex_];
currentIndex_ = (currentIndex_ + 1 )%dataFileDirectories_.length ;
String dataFileDirectory = dataFileDirectories_[currentIndex_] + File.separator + table;
currentIndex_ = (currentIndex_ + 1) % dataFileDirectories_.length;
return dataFileDirectory;
}
@ -728,14 +751,16 @@ public class DatabaseDescriptor
* compacted file is greater than the max disk space available return null, we cannot
* do compaction in this case.
*/
public static String getCompactionFileLocation(long expectedCompactedFileSize)
public static String getDataFileLocationForTable(String table, long expectedCompactedFileSize)
{
long maxFreeDisk = 0;
int maxDiskIndex = 0;
String dataFileDirectory = null;
for ( int i = 0 ; i < dataFileDirectories_.length ; i++ )
String[] dataDirectoryForTable = getAllDataFileLocationsForTable(table);
for ( int i = 0 ; i < dataDirectoryForTable.length ; i++ )
{
File f = new File(dataFileDirectories_[i]);
File f = new File(dataDirectoryForTable[i]);
if( maxFreeDisk < f.getUsableSpace())
{
maxFreeDisk = f.getUsableSpace();
@ -746,8 +771,8 @@ public class DatabaseDescriptor
maxFreeDisk = (long)(0.9 * maxFreeDisk);
if( expectedCompactedFileSize < maxFreeDisk )
{
dataFileDirectory = dataFileDirectories_[maxDiskIndex];
currentIndex_ = (maxDiskIndex + 1 )%dataFileDirectories_.length ;
dataFileDirectory = dataDirectoryForTable[maxDiskIndex];
currentIndex_ = (maxDiskIndex + 1 )%dataDirectoryForTable.length ;
}
else
{

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

@ -110,18 +110,18 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
* index.
*/
List<Integer> indices = new ArrayList<Integer>();
String[] dataFileDirectories = DatabaseDescriptor.getAllDataFileLocations();
String[] dataFileDirectories = DatabaseDescriptor.getAllDataFileLocationsForTable(table);
for (String directory : dataFileDirectories)
{
File fileDir = new File(directory);
File[] files = fileDir.listFiles();
for (File file : files)
{
String filename = file.getName();
String[] tblCfName = getTableAndColumnFamilyName(filename);
String cfName = getColumnFamilyFromFileName(filename);
if (tblCfName[0].equals(table)
&& tblCfName[1].equals(columnFamily))
if (cfName.equals(columnFamily))
{
int index = getIndexFromFileName(filename);
indices.add(index);
@ -151,7 +151,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
// scan for data files corresponding to this CF
List<File> sstableFiles = new ArrayList<File>();
String[] dataFileDirectories = DatabaseDescriptor.getAllDataFileLocations();
String[] dataFileDirectories = DatabaseDescriptor.getAllDataFileLocationsForTable(table_);
for (String directory : dataFileDirectories)
{
File fileDir = new File(directory);
@ -165,9 +165,8 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
continue;
}
String[] tblCfName = getTableAndColumnFamilyName(filename);
if (tblCfName[0].equals(table_)
&& tblCfName[1].equals(columnFamily_)
String cfName = getColumnFamilyFromFileName(filename);
if (cfName.equals(columnFamily_)
&& filename.contains("-Data.db"))
{
sstableFiles.add(file.getAbsoluteFile());
@ -311,25 +310,9 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
return columnFamily_;
}
private static String[] getTableAndColumnFamilyName(String filename)
{
StringTokenizer st = new StringTokenizer(filename, "-");
String[] values = new String[2];
int i = 0;
while (st.hasMoreElements())
{
if (i == 0)
private static String getColumnFamilyFromFileName(String filename)
{
values[i] = (String) st.nextElement();
}
else if (i == 1)
{
values[i] = (String) st.nextElement();
break;
}
++i;
}
return values;
return filename.split("-")[0];
}
protected static int getIndexFromFileName(String filename)
@ -374,14 +357,15 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
// increment twice so that we do not generate consecutive numbers
String fname = getTempSSTableFileName();
return new File(DatabaseDescriptor.getDataFileLocation(), fname).getAbsolutePath();
return new File(DatabaseDescriptor.getDataFileLocationForTable(table_), fname).getAbsolutePath();
}
String getTempSSTableFileName()
{
fileIndexGenerator_.incrementAndGet();
return String.format("%s-%s-%s-%s-Data.db",
table_, columnFamily_, SSTable.TEMPFILE_MARKER, fileIndexGenerator_.incrementAndGet());
return String.format("%s-%s-%s-Data.db",
columnFamily_, SSTable.TEMPFILE_MARKER, fileIndexGenerator_.incrementAndGet());
}
/*
@ -405,8 +389,8 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
index = lowestIndex + 1;
return String.format("%s-%s-%s-%s-Data.db",
table_, columnFamily_, SSTable.TEMPFILE_MARKER, index);
return String.format("%s-%s-%s-Data.db",
columnFamily_, SSTable.TEMPFILE_MARKER, index);
}
void switchMemtable()
@ -924,7 +908,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
long expectedRangeFileSize = getExpectedCompactedFileSize(files);
/* in the worst case a node will be giving out half of its data so we take a chance */
expectedRangeFileSize = expectedRangeFileSize / 2;
rangeFileLocation = DatabaseDescriptor.getCompactionFileLocation(expectedRangeFileSize);
rangeFileLocation = DatabaseDescriptor.getDataFileLocationForTable(table_, expectedRangeFileSize);
// If the compaction file path is null that means we have no space left for this compaction.
if (rangeFileLocation == null)
{
@ -1084,12 +1068,13 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
}
logger_.debug("Total time taken for range split ..."
+ (System.currentTimeMillis() - startTime));
if (logger_.isDebugEnabled())
{
logger_.debug("Total time taken for range split ..." + (System.currentTimeMillis() - startTime));
logger_.debug("Total bytes Read for range split ..." + totalBytesRead);
logger_.debug("Total bytes written for range split ..."
+ totalBytesWritten + " Total keys read ..." + totalkeysRead);
}
return result;
}
@ -1112,7 +1097,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
private int doFileCompaction(List<String> files, int minBufferSize) throws IOException
{
logger_.info("Compacting [" + StringUtils.join(files, ",") + "]");
String compactionFileLocation = DatabaseDescriptor.getCompactionFileLocation(getExpectedCompactedFileSize(files));
String compactionFileLocation = DatabaseDescriptor.getDataFileLocationForTable(table_, getExpectedCompactedFileSize(files));
// If the compaction file path is null that means we have no space left for this compaction.
// try again w/o the largest one.
if (compactionFileLocation == null)

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

@ -188,6 +188,7 @@ public class Table
* sampler.
*/
SSTableReader sstable = SSTableReader.open(streamContext.getTargetFile());
if (logger_.isDebugEnabled())
logger_.debug("Merging the counting bloom filter in the sampler ...");
String[] peices = FBUtilities.strip(fileName, "-");
Table.open(peices[0]).getColumnFamilyStore(peices[1]).addToList(sstable);
@ -236,7 +237,7 @@ public class Table
String[] peices = FBUtilities.strip(sourceFile.getName(), "-");
String newFileName = fileNames.get( peices[1] + "-" + peices[2] );
String file = DatabaseDescriptor.getDataFileLocation() + File.separator + newFileName + "-Data.db";
String file = DatabaseDescriptor.getDataFileLocationForTable(streamContext.getTable()) + File.separator + newFileName + "-Data.db";
if (logger_.isDebugEnabled())
logger_.debug("Received Data from : " + message.getFrom() + " " + streamContext.getTargetFile() + " " + file);
streamContext.setTargetFile(file);
@ -245,6 +246,7 @@ public class Table
StreamContextManager.registerStreamCompletionHandler(message.getFrom().getHost(), new Table.BootstrapCompletionHandler());
/* Send a bootstrap initiation done message to execute on default stage. */
if (logger_.isDebugEnabled())
logger_.debug("Sending a bootstrap initiate done message ...");
Message doneMessage = new Message( StorageService.getLocalStorageEndPoint(), "", StorageService.bootStrapInitiateDoneVerbHandler_, new byte[0] );
MessagingService.getMessagingInstance().sendOneWay(doneMessage, message.getFrom());

11
src/java/org/apache/cassandra/dht/BootstrapMetadataVerbHandler.java
View File

@ -32,7 +32,6 @@ import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.io.StreamContextManager;
import org.apache.cassandra.service.StreamManager;
import org.apache.cassandra.utils.BloomFilter;
import org.apache.cassandra.utils.LogUtil;
import org.apache.log4j.Logger;
@ -61,6 +60,7 @@ public class BootstrapMetadataVerbHandler implements IVerbHandler
*/
for ( BootstrapMetadata bsmd : bsMetadata )
{
if (logger_.isDebugEnabled())
logger_.debug(bsmd.toString());
}
@ -68,6 +68,7 @@ public class BootstrapMetadataVerbHandler implements IVerbHandler
{
long startTime = System.currentTimeMillis();
doTransfer(bsmd.target_, bsmd.ranges_);
if (logger_.isDebugEnabled())
logger_.debug("Time taken to boostrap " +
bsmd.target_ +
" is " +
@ -101,6 +102,7 @@ public class BootstrapMetadataVerbHandler implements IVerbHandler
StringBuilder sb = new StringBuilder("");
sb.append(range.toString());
sb.append(" ");
if (logger_.isDebugEnabled())
logger_.debug("Beginning transfer process to " + target + " for ranges " + sb.toString());
}
@ -124,7 +126,7 @@ public class BootstrapMetadataVerbHandler implements IVerbHandler
/* Get the counting bloom filter for each endpoint and the list of files that need to be streamed */
List<String> fileList = new ArrayList<String>();
boolean bVal = table.forceCompaction(ranges, target, fileList);
doHandoff(target, fileList);
doHandoff(target, fileList, tName);
}
}
@ -132,7 +134,7 @@ public class BootstrapMetadataVerbHandler implements IVerbHandler
* Stream the files in the bootstrap directory over to the
* node being bootstrapped.
*/
private void doHandoff(EndPoint target, List<String> fileList) throws IOException
private void doHandoff(EndPoint target, List<String> fileList, String table) throws IOException
{
List<File> filesList = new ArrayList<File>();
for(String file : fileList)
@ -144,7 +146,7 @@ public class BootstrapMetadataVerbHandler implements IVerbHandler
int i = 0;
for ( File file : files )
{
streamContexts[i] = new StreamContextManager.StreamContext(file.getAbsolutePath(), file.length());
streamContexts[i] = new StreamContextManager.StreamContext(file.getAbsolutePath(), file.length(), table);
if (logger_.isDebugEnabled())
logger_.debug("Stream context metadata " + streamContexts[i]);
++i;
@ -163,6 +165,7 @@ public class BootstrapMetadataVerbHandler implements IVerbHandler
if (logger_.isDebugEnabled())
logger_.debug("Waiting for transfer to " + target + " to complete");
StreamManager.instance(target).waitForStreamCompletion();
if (logger_.isDebugEnabled())
logger_.debug("Done with transfer to " + target);
}
}

5
src/java/org/apache/cassandra/io/SSTable.java
View File

@ -69,8 +69,7 @@ public abstract class SSTable
static String parseTableName(String filename)
{
String[] parts = new File(filename).getName().split("-"); // table, cf, index, [filetype]
return parts[0];
return new File(filename).getParentFile().getName();
}
/**
@ -101,4 +100,4 @@ public abstract class SSTable
return key + ":" + position;
}
}
}
}

19
src/java/org/apache/cassandra/net/io/StreamContextManager.java
View File

@ -24,14 +24,8 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.io.Serializable;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import javax.xml.bind.annotation.XmlElement;
import org.apache.cassandra.db.Table;
import org.apache.cassandra.dht.BootstrapInitiateMessage;
import org.apache.cassandra.io.DataInputBuffer;
import org.apache.cassandra.io.ICompactSerializer;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.service.StorageService;
import org.apache.log4j.Logger;
@ -67,11 +61,18 @@ public class StreamContextManager
private String targetFile_;
private long expectedBytes_;
private String table_;
public StreamContext(String targetFile, long expectedBytes)
public StreamContext(String targetFile, long expectedBytes, String table)
{
targetFile_ = targetFile;
expectedBytes_ = expectedBytes;
table_ = table;
}
public String getTable()
{
return table_;
}
public String getTargetFile()
@ -115,13 +116,15 @@ public class StreamContextManager
{
dos.writeUTF(sc.targetFile_);
dos.writeLong(sc.expectedBytes_);
dos.writeUTF(sc.table_);
}
public StreamContextManager.StreamContext deserialize(DataInputStream dis) throws IOException
{
String targetFile = dis.readUTF();
long expectedBytes = dis.readLong();
return new StreamContext(targetFile, expectedBytes);
String table = dis.readUTF();
return new StreamContext(targetFile, expectedBytes, table);
}
}

34
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -694,34 +694,37 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
}
}
public void forceHandoff(String directories, String host) throws IOException
public void forceHandoff(List<String> dataDirectories, String host) throws IOException
{
List<File> filesList = new ArrayList<File>();
String[] sources = directories.split(":");
for (String source : sources)
List<StreamContextManager.StreamContext> streamContexts = new ArrayList<StreamContextManager.StreamContext>();
for (String dataDir : dataDirectories)
{
File directory = new File(source);
File directory = new File(dataDir);
Collections.addAll(filesList, directory.listFiles());
}
File[] files = filesList.toArray(new File[0]);
StreamContextManager.StreamContext[] streamContexts = new StreamContextManager.StreamContext[files.length];
int i = 0;
for ( File file : files )
for (File tableDir : directory.listFiles())
{
String tableName = tableDir.getName();
for (File file : tableDir.listFiles())
{
streamContexts[i] = new StreamContextManager.StreamContext(file.getAbsolutePath(), file.length());
streamContexts.add(new StreamContextManager.StreamContext(file.getAbsolutePath(), file.length(), tableName));
if (logger_.isDebugEnabled())
logger_.debug("Stream context metadata " + streamContexts[i]);
++i;
logger_.debug("Stream context metadata " + streamContexts);
}
}
}
if ( files.length > 0 )
if ( streamContexts.size() > 0 )
{
EndPoint target = new EndPoint(host, DatabaseDescriptor.getStoragePort());
/* Set up the stream manager with the files that need to streamed */
StreamManager.instance(target).addFilesToStream(streamContexts);
StreamManager.instance(target).addFilesToStream((StreamContextManager.StreamContext[]) streamContexts.toArray());
/* Send the bootstrap initiate message */
BootstrapInitiateMessage biMessage = new BootstrapInitiateMessage(streamContexts);
BootstrapInitiateMessage biMessage = new BootstrapInitiateMessage((StreamContextManager.StreamContext[]) streamContexts.toArray());
Message message = BootstrapInitiateMessage.makeBootstrapInitiateMessage(biMessage);
if (logger_.isDebugEnabled())
logger_.debug("Sending a bootstrap initiate message to " + target + " ...");
@ -729,6 +732,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
if (logger_.isDebugEnabled())
logger_.debug("Waiting for transfer to " + target + " to complete");
StreamManager.instance(target).waitForStreamCompletion();
if (logger_.isDebugEnabled())
logger_.debug("Done with transfer to " + target);
}
}

2
src/java/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -65,5 +65,5 @@ public interface StorageServiceMBean
* files need to be picked up.
* @param target endpoint receiving data.
*/
public void forceHandoff(String directories, String target) throws IOException;
public void forceHandoff(List<String> directories, String target) throws IOException;
}

32
test/unit/org/apache/cassandra/CleanupHelper.java
View File

@ -37,7 +37,6 @@ public class CleanupHelper
String[] directoryNames = {
DatabaseDescriptor.getBootstrapFileLocation(),
DatabaseDescriptor.getLogFileLocation(),
DatabaseDescriptor.getDataFileLocation(),
};
for (String dirName : directoryNames)
@ -49,9 +48,36 @@ public class CleanupHelper
}
for (File f : dir.listFiles())
{
if (logger.isDebugEnabled())
logger.debug("deleting " + f);
f.delete();
if (!f.delete()) {
logger.error("could not delete " + f);
}
}
}
}
// cleanup data directory which are stored as data directory/table/data files
for (String dirName : DatabaseDescriptor.getAllDataFileLocations())
{
File dir = new File(dirName);
if (!dir.exists())
{
throw new RuntimeException("No such directory: " + dir.getAbsolutePath());
}
for (File tableFile : dir.listFiles())
{
// table directory
if (tableFile.isDirectory()) {
for (File dataFile : tableFile.listFiles()) {
if (logger.isDebugEnabled())
logger.debug("deleting " + dataFile);
if (!dataFile.delete()) {
logger.error("could not delete " + dataFile);
}
}
}
}
}
}
}