From 840a64dcb9cefe927c681424e12284a1ef40452e Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Thu, 16 Jul 2009 19:10:06 +0000 Subject: [PATCH] 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 --- .../cassandra/config/DatabaseDescriptor.java | 53 +++++++++++++----- .../cassandra/db/ColumnFamilyStore.java | 55 +++++++------------ src/java/org/apache/cassandra/db/Table.java | 4 +- .../dht/BootstrapMetadataVerbHandler.java | 11 ++-- src/java/org/apache/cassandra/io/SSTable.java | 5 +- .../net/io/StreamContextManager.java | 19 ++++--- .../cassandra/service/StorageService.java | 34 +++++++----- .../service/StorageServiceMBean.java | 2 +- .../org/apache/cassandra/CleanupHelper.java | 32 ++++++++++- 9 files changed, 131 insertions(+), 84 deletions(-) diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index e8dfb41f12..b614c7b44c 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -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 { diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index dc9698e780..7d22208b65 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -110,18 +110,18 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean * index. */ List indices = new ArrayList(); - 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 sstableFiles = new ArrayList(); - 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 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) diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java index 16c44d386b..6e72c9b87f 100644 --- a/src/java/org/apache/cassandra/db/Table.java +++ b/src/java/org/apache/cassandra/db/Table.java @@ -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()); diff --git a/src/java/org/apache/cassandra/dht/BootstrapMetadataVerbHandler.java b/src/java/org/apache/cassandra/dht/BootstrapMetadataVerbHandler.java index 87d52e4f27..d5acc4c600 100644 --- a/src/java/org/apache/cassandra/dht/BootstrapMetadataVerbHandler.java +++ b/src/java/org/apache/cassandra/dht/BootstrapMetadataVerbHandler.java @@ -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 fileList = new ArrayList(); 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 fileList) throws IOException + private void doHandoff(EndPoint target, List fileList, String table) throws IOException { List filesList = new ArrayList(); 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); } } diff --git a/src/java/org/apache/cassandra/io/SSTable.java b/src/java/org/apache/cassandra/io/SSTable.java index 4b0b3e53f8..6948a576c3 100644 --- a/src/java/org/apache/cassandra/io/SSTable.java +++ b/src/java/org/apache/cassandra/io/SSTable.java @@ -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; } } -} +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/net/io/StreamContextManager.java b/src/java/org/apache/cassandra/net/io/StreamContextManager.java index 74302cd1fd..1482470c7b 100644 --- a/src/java/org/apache/cassandra/net/io/StreamContextManager.java +++ b/src/java/org/apache/cassandra/net/io/StreamContextManager.java @@ -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); } } diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 5355de4a06..6ad80dfe6d 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -694,34 +694,37 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto } } - public void forceHandoff(String directories, String host) throws IOException + public void forceHandoff(List dataDirectories, String host) throws IOException { List filesList = new ArrayList(); - String[] sources = directories.split(":"); - for (String source : sources) + List streamContexts = new ArrayList(); + + 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); } } diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index 1384aa2feb..20072ed411 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -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 directories, String target) throws IOException; } diff --git a/test/unit/org/apache/cassandra/CleanupHelper.java b/test/unit/org/apache/cassandra/CleanupHelper.java index 75a0584bae..41976381b2 100644 --- a/test/unit/org/apache/cassandra/CleanupHelper.java +++ b/test/unit/org/apache/cassandra/CleanupHelper.java @@ -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); } } } -} \ No newline at end of file + + // 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); + } + } + } + } + } + + } +}