You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by ji...@apache.org on 2008/01/13 02:36:10 UTC
svn commit: r611519 [1/3] - in /lucene/hadoop/trunk/src/contrib/hbase: ./
src/java/org/apache/hadoop/hbase/ src/java/org/apache/hadoop/hbase/util/
src/test/ src/test/org/apache/hadoop/hbase/
src/test/org/apache/hadoop/hbase/mapred/ src/test/org/apache/...
Author: jimk
Date: Sat Jan 12 17:36:08 2008
New Revision: 611519
URL: http://svn.apache.org/viewvc?rev=611519&view=rev
Log:
HADOOP-2478 Restructure how HBase lays out files in the file system (phase 2) Includes migration tool org.apache.hadoop.hbase.util.Migrate
Added:
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/util/Migrate.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/util/TestMigrate.java
Modified:
lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HColumnDescriptor.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HConstants.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HLog.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMaster.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMerge.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegion.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionInfo.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionServer.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HStore.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HStoreFile.java
lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HTableDescriptor.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/hbase-site.xml
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/AbstractMergeTestBase.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/HBaseClusterTestCase.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/HBaseTestCase.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/MiniHBaseCluster.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/MultiRegionTable.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestCompaction.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestDeleteAll.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestDeleteFamily.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestGet.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestGet2.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestHLog.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestHRegion.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestHStoreFile.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestInfoServers.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestLogRolling.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestMergeMeta.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestMergeTable.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestScanner.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestSplit.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/TestTimestamp.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/mapred/TestTableIndex.java
lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/mapred/TestTableMapReduce.java
Modified: lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt?rev=611519&r1=611518&r2=611519&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt Sat Jan 12 17:36:08 2008
@@ -10,6 +10,8 @@
HADOOP-2519 Performance improvements: Customized RPC serialization
HADOOP-2478 Restructure how HBase lays out files in the file system (phase 1)
(test input data)
+ HADOOP-2478 Restructure how HBase lays out files in the file system (phase 2)
+ Includes migration tool org.apache.hadoop.hbase.util.Migrate
NEW FEATURES
HADOOP-2061 Add new Base64 dialects
Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HColumnDescriptor.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HColumnDescriptor.java?rev=611519&r1=611518&r2=611519&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HColumnDescriptor.java (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HColumnDescriptor.java Sat Jan 12 17:36:08 2008
@@ -28,6 +28,8 @@
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.hbase.io.TextSequence;
+
/**
* An HColumnDescriptor contains information about a column family such as the
* number of versions, compression settings, etc.
@@ -101,6 +103,8 @@
private BloomFilterDescriptor bloomFilter;
// Version number of this class
private byte versionNumber;
+ // Family name without the ':'
+ private transient Text familyName = null;
/**
* Default constructor. Must be present for Writable.
@@ -172,6 +176,17 @@
public Text getName() {
return name;
}
+
+ /** @return name of column family without trailing ':' */
+ public synchronized Text getFamilyName() {
+ if (name != null) {
+ if (familyName == null) {
+ familyName = new TextSequence(name, 0, name.getLength() - 1).toText();
+ }
+ return familyName;
+ }
+ return null;
+ }
/** @return compression type being used for the column family */
public CompressionType getCompression() {
@@ -211,6 +226,7 @@
return this.bloomFilter;
}
+ /** {@inheritDoc} */
@Override
public String toString() {
// Output a name minus ':'.
Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HConstants.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HConstants.java?rev=611519&r1=611518&r2=611519&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HConstants.java (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HConstants.java Sat Jan 12 17:36:08 2008
@@ -78,9 +78,6 @@
/** Default HBase instance root directory */
static final String DEFAULT_HBASE_DIR = "/hbase";
- /** Used to construct the name of the directory in which a HRegion resides */
- static final String HREGIONDIR_PREFIX = "hregion_";
-
/** Used to construct the name of the log directory for a region server */
static final String HREGION_LOGDIR_NAME = "log";
Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HLog.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HLog.java?rev=611519&r1=611518&r2=611519&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HLog.java (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HLog.java Sat Jan 12 17:36:08 2008
@@ -131,88 +131,6 @@
private final Integer updateLock = new Integer(0);
/**
- * Split up a bunch of log files, that are no longer being written to, into
- * new files, one per region. Delete the old log files when finished.
- *
- * @param rootDir Root directory of the HBase instance
- * @param srcDir Directory of log files to split: e.g.
- * <code>${ROOTDIR}/log_HOST_PORT</code>
- * @param fs FileSystem
- * @param conf HBaseConfiguration
- * @throws IOException
- */
- static void splitLog(Path rootDir, Path srcDir, FileSystem fs,
- Configuration conf)
- throws IOException {
- Path logfiles[] = fs.listPaths(new Path[] { srcDir });
- LOG.info("splitting " + logfiles.length + " log(s) in " +
- srcDir.toString());
- Map<Text, SequenceFile.Writer> logWriters =
- new HashMap<Text, SequenceFile.Writer>();
- try {
- for (int i = 0; i < logfiles.length; i++) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Splitting " + i + " of " + logfiles.length + ": " +
- logfiles[i]);
- }
- // Check for empty file.
- if (fs.getFileStatus(logfiles[i]).getLen() <= 0) {
- LOG.info("Skipping " + logfiles[i].toString() +
- " because zero length");
- continue;
- }
- HLogKey key = new HLogKey();
- HLogEdit val = new HLogEdit();
- SequenceFile.Reader in = new SequenceFile.Reader(fs, logfiles[i], conf);
- try {
- int count = 0;
- for (; in.next(key, val); count++) {
- Text regionName = key.getRegionName();
- SequenceFile.Writer w = logWriters.get(regionName);
- if (w == null) {
- Path logfile = new Path(HRegion.getRegionDir(rootDir,
- HRegionInfo.encodeRegionName(regionName)),
- HREGION_OLDLOGFILE_NAME);
- if (LOG.isDebugEnabled()) {
- LOG.debug("Creating new log file writer for path " + logfile +
- "; map content " + logWriters.toString());
- }
- w = SequenceFile.createWriter(fs, conf, logfile, HLogKey.class,
- HLogEdit.class, getCompressionType(conf));
- // Use copy of regionName; regionName object is reused inside in
- // HStoreKey.getRegionName so its content changes as we iterate.
- logWriters.put(new Text(regionName), w);
- }
- if (count % 10000 == 0 && count > 0 && LOG.isDebugEnabled()) {
- LOG.debug("Applied " + count + " edits");
- }
- w.append(key, val);
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug("Applied " + count + " total edits");
- }
- } finally {
- in.close();
- }
- }
- } finally {
- for (SequenceFile.Writer w : logWriters.values()) {
- w.close();
- }
- }
-
- if (fs.exists(srcDir)) {
- if (!fs.delete(srcDir)) {
- LOG.error("Cannot delete: " + srcDir);
- if (!FileUtil.fullyDelete(new File(srcDir.toString()))) {
- throw new IOException("Cannot delete: " + srcDir);
- }
- }
- }
- LOG.info("log file splitting completed for " + srcDir.toString());
- }
-
- /**
* Create an edit log at the given <code>dir</code> location.
*
* You should never have to load an existing log. If there is a log at
@@ -222,9 +140,10 @@
* @param fs
* @param dir
* @param conf
+ * @param listener
* @throws IOException
*/
- HLog(final FileSystem fs, final Path dir, final Configuration conf,
+ public HLog(final FileSystem fs, final Path dir, final Configuration conf,
final LogRollListener listener) throws IOException {
this.fs = fs;
this.dir = dir;
@@ -381,7 +300,7 @@
*
* @throws IOException
*/
- void closeAndDelete() throws IOException {
+ public void closeAndDelete() throws IOException {
close();
fs.delete(dir);
}
@@ -557,6 +476,92 @@
*/
void abortCacheFlush() {
this.cacheFlushLock.unlock();
+ }
+
+ /**
+ * Split up a bunch of log files, that are no longer being written to, into
+ * new files, one per region. Delete the old log files when finished.
+ *
+ * @param rootDir qualified root directory of the HBase instance
+ * @param srcDir Directory of log files to split: e.g.
+ * <code>${ROOTDIR}/log_HOST_PORT</code>
+ * @param fs FileSystem
+ * @param conf HBaseConfiguration
+ * @throws IOException
+ */
+ static void splitLog(Path rootDir, Path srcDir, FileSystem fs,
+ Configuration conf) throws IOException {
+ Path logfiles[] = fs.listPaths(new Path[] { srcDir });
+ LOG.info("splitting " + logfiles.length + " log(s) in " +
+ srcDir.toString());
+ Map<Text, SequenceFile.Writer> logWriters =
+ new HashMap<Text, SequenceFile.Writer>();
+ try {
+ for (int i = 0; i < logfiles.length; i++) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Splitting " + i + " of " + logfiles.length + ": " +
+ logfiles[i]);
+ }
+ // Check for empty file.
+ if (fs.getFileStatus(logfiles[i]).getLen() <= 0) {
+ LOG.info("Skipping " + logfiles[i].toString() +
+ " because zero length");
+ continue;
+ }
+ HLogKey key = new HLogKey();
+ HLogEdit val = new HLogEdit();
+ SequenceFile.Reader in = new SequenceFile.Reader(fs, logfiles[i], conf);
+ try {
+ int count = 0;
+ for (; in.next(key, val); count++) {
+ Text tableName = key.getTablename();
+ Text regionName = key.getRegionName();
+ SequenceFile.Writer w = logWriters.get(regionName);
+ if (w == null) {
+ Path logfile = new Path(
+ HRegion.getRegionDir(
+ HTableDescriptor.getTableDir(rootDir, tableName),
+ HRegionInfo.encodeRegionName(regionName)
+ ),
+ HREGION_OLDLOGFILE_NAME
+ );
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Creating new log file writer for path " + logfile +
+ "; map content " + logWriters.toString());
+ }
+ w = SequenceFile.createWriter(fs, conf, logfile, HLogKey.class,
+ HLogEdit.class, getCompressionType(conf));
+ // Use copy of regionName; regionName object is reused inside in
+ // HStoreKey.getRegionName so its content changes as we iterate.
+ logWriters.put(new Text(regionName), w);
+ }
+ if (count % 10000 == 0 && count > 0 && LOG.isDebugEnabled()) {
+ LOG.debug("Applied " + count + " edits");
+ }
+ w.append(key, val);
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Applied " + count + " total edits");
+ }
+ } finally {
+ in.close();
+ }
+ }
+ } finally {
+ for (SequenceFile.Writer w : logWriters.values()) {
+ w.close();
+ }
+ }
+
+ if (fs.exists(srcDir)) {
+ if (!fs.delete(srcDir)) {
+ LOG.error("Cannot delete: " + srcDir);
+ if (!FileUtil.fullyDelete(new File(srcDir.toString()))) {
+ throw new IOException("Cannot delete: " + srcDir);
+ }
+ }
+ }
+ LOG.info("log file splitting completed for " + srcDir.toString());
}
private static void usage() {
Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMaster.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMaster.java?rev=611519&r1=611518&r2=611519&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMaster.java (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMaster.java Sat Jan 12 17:36:08 2008
@@ -95,7 +95,7 @@
volatile boolean shutdownRequested = false;
volatile AtomicInteger quiescedMetaServers = new AtomicInteger(0);
volatile boolean fsOk = true;
- final Path dir;
+ final Path rootdir;
final HBaseConfiguration conf;
final FileSystem fs;
final Random rand;
@@ -333,8 +333,7 @@
if (!hasReferencesA && !hasReferencesB) {
LOG.info("Deleting region " + parent.getRegionName() +
" because daughter splits no longer hold references");
- if (!HRegion.deleteRegion(fs, dir,
- HRegionInfo.encodeRegionName(parent.getRegionName()))) {
+ if (!HRegion.deleteRegion(fs, rootdir, parent)) {
LOG.warn("Deletion of " + parent.getRegionName() + " failed");
}
@@ -378,10 +377,11 @@
if (split == null) {
return result;
}
- for (Text family: split.getTableDesc().families().keySet()) {
- Path p = HStoreFile.getMapDir(fs.makeQualified(dir),
- HRegionInfo.encodeRegionName(split.getRegionName()),
- HStoreKey.extractFamily(family));
+ Path tabledir =
+ HTableDescriptor.getTableDir(rootdir, split.getTableDesc().getName());
+ for (HColumnDescriptor family: split.getTableDesc().families().values()) {
+ Path p = HStoreFile.getMapDir(tabledir, split.getEncodedName(),
+ family.getFamilyName());
// Look for reference files. Call listPaths with an anonymous
// instance of PathFilter.
@@ -389,7 +389,7 @@
Path [] ps = fs.listPaths(p,
new PathFilter () {
public boolean accept(Path path) {
- return HStoreFile.isReference(path);
+ return HStore.isReference(path);
}
}
);
@@ -481,12 +481,12 @@
if (!initialMetaScanComplete && serverName.length() != 0) {
StringBuilder dirName = new StringBuilder("log_");
dirName.append(serverName.replace(":", "_"));
- Path logDir = new Path(dir, dirName.toString());
+ Path logDir = new Path(rootdir, dirName.toString());
try {
if (fs.exists(logDir)) {
splitLogLock.lock();
try {
- HLog.splitLog(dir, logDir, fs, conf);
+ HLog.splitLog(rootdir, logDir, fs, conf);
} finally {
splitLogLock.unlock();
}
@@ -866,37 +866,38 @@
/**
* Build the HMaster
- * @param dir base directory
+ * @param rootdir base directory of this HBase instance
* @param address server address and port number
* @param conf configuration
*
* @throws IOException
*/
- public HMaster(Path dir, HServerAddress address, HBaseConfiguration conf)
+ public HMaster(Path rootdir, HServerAddress address, HBaseConfiguration conf)
throws IOException {
- this.dir = dir;
this.conf = conf;
this.fs = FileSystem.get(conf);
+ this.rootdir = fs.makeQualified(rootdir);
+ this.conf.set(HConstants.HBASE_DIR, this.rootdir.toString());
this.rand = new Random();
- Path rootRegionDir = HRegion.getRegionDir(dir,
- HRegionInfo.encodeRegionName(HRegionInfo.rootRegionInfo.getRegionName()));
+ Path rootRegionDir =
+ HRegion.getRegionDir(rootdir, HRegionInfo.rootRegionInfo);
LOG.info("Root region dir: " + rootRegionDir.toString());
try {
// Make sure the root directory exists!
- if(! fs.exists(dir)) {
- fs.mkdirs(dir);
+ if(! fs.exists(rootdir)) {
+ fs.mkdirs(rootdir);
}
if (!fs.exists(rootRegionDir)) {
LOG.info("bootstrap: creating ROOT and first META regions");
try {
HRegion root = HRegion.createHRegion(HRegionInfo.rootRegionInfo,
- this.dir, this.conf);
+ this.rootdir, this.conf);
HRegion meta = HRegion.createHRegion(HRegionInfo.firstMetaRegionInfo,
- this.dir, this.conf);
+ this.rootdir, this.conf);
// Add first region from the META table to the ROOT region.
HRegion.addRegionToMETA(root, meta);
@@ -990,7 +991,7 @@
* @return Hbase root dir.
*/
public Path getRootDir() {
- return this.dir;
+ return this.rootdir;
}
/**
@@ -1973,7 +1974,7 @@
dirName.append(serverInfo.getStartCode());
dirName.append("_");
dirName.append(deadServer.getPort());
- this.oldLogDir = new Path(dir, dirName.toString());
+ this.oldLogDir = new Path(rootdir, dirName.toString());
}
/** {@inheritDoc} */
@@ -2148,7 +2149,7 @@
return false;
}
try {
- HLog.splitLog(dir, oldLogDir, fs, conf);
+ HLog.splitLog(rootdir, oldLogDir, fs, conf);
} finally {
splitLogLock.unlock();
}
@@ -2415,8 +2416,7 @@
} else if (deleteRegion) {
try {
- HRegion.deleteRegion(fs, dir,
- HRegionInfo.encodeRegionName(regionInfo.getRegionName()));
+ HRegion.deleteRegion(fs, rootdir, regionInfo);
} catch (IOException e) {
e = RemoteExceptionHandler.checkIOException(e);
LOG.error("failed delete region " + regionInfo.getRegionName(), e);
@@ -2613,7 +2613,7 @@
// 2. Create the HRegion
HRegion region =
- HRegion.createHRegion(newRegion, this.dir, this.conf);
+ HRegion.createHRegion(newRegion, this.rootdir, this.conf);
// 3. Insert into meta
@@ -2999,8 +2999,7 @@
// Delete the region
try {
- HRegion.deleteRegion(fs, dir,
- HRegionInfo.encodeRegionName(i.getRegionName()));
+ HRegion.deleteRegion(fs, rootdir, i);
} catch (IOException e) {
LOG.error("failed to delete region " + i.getRegionName(),
@@ -3061,15 +3060,16 @@
protected void postProcessMeta(MetaRegion m, HRegionInterface server)
throws IOException {
+ Path tabledir = new Path(rootdir, tableName.toString());
for (HRegionInfo i: unservedRegions) {
i.getTableDesc().families().remove(columnName);
updateRegionInfo(server, m.getRegionName(), i);
// Delete the directories used by the column
- String encodedName = HRegionInfo.encodeRegionName(i.getRegionName());
- fs.delete(HStoreFile.getMapDir(dir, encodedName, columnName));
- fs.delete(HStoreFile.getInfoDir(dir, encodedName, columnName));
+ String encodedName = i.getEncodedName();
+ fs.delete(HStoreFile.getMapDir(tabledir, encodedName, columnName));
+ fs.delete(HStoreFile.getInfoDir(tabledir, encodedName, columnName));
}
}
}
Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMerge.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMerge.java?rev=611519&r1=611518&r2=611519&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMerge.java (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMerge.java Sat Jan 12 17:36:08 2008
@@ -19,20 +19,18 @@
*/
package org.apache.hadoop.hbase;
-import java.io.ByteArrayOutputStream;
-import java.io.DataOutputStream;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
import java.util.NoSuchElementException;
import java.util.Random;
import java.util.TreeMap;
-import java.util.TreeSet;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.util.Writables;
-import org.apache.hadoop.io.DataInputBuffer;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.hbase.io.BatchUpdate;
@@ -43,7 +41,6 @@
*/
class HMerge implements HConstants {
static final Log LOG = LogFactory.getLog(HMerge.class);
- static final Text[] META_COLS = {COL_REGIONINFO};
static final Random rand = new Random();
private HMerge() {
@@ -74,7 +71,7 @@
throw new IllegalStateException(
"Can not compact META table if instance is on-line");
}
- new OfflineMerger(conf, fs, META_TABLE_NAME).process();
+ new OfflineMerger(conf, fs).process();
} else {
if(!masterIsRunning) {
@@ -86,42 +83,40 @@
}
private static abstract class Merger {
- protected HBaseConfiguration conf;
- protected FileSystem fs;
- protected Text tableName;
- protected Path dir;
- protected Path basedir;
- protected HLog hlog;
- protected DataInputBuffer in;
- protected boolean more;
- protected HStoreKey key;
- protected HRegionInfo info;
+ protected final HBaseConfiguration conf;
+ protected final FileSystem fs;
+ protected final Path tabledir;
+ protected final HLog hlog;
+ private final long maxFilesize;
+
protected Merger(HBaseConfiguration conf, FileSystem fs, Text tableName)
throws IOException {
this.conf = conf;
this.fs = fs;
- this.tableName = tableName;
- this.in = new DataInputBuffer();
- this.more = true;
- this.key = new HStoreKey();
- this.info = new HRegionInfo();
- this.dir = new Path(conf.get(HBASE_DIR, DEFAULT_HBASE_DIR));
- this.basedir = new Path(dir, "merge_" + System.currentTimeMillis());
- fs.mkdirs(basedir);
+ this.maxFilesize =
+ conf.getLong("hbase.hregion.max.filesize", DEFAULT_MAX_FILE_SIZE);
+
+ this.tabledir = new Path(
+ fs.makeQualified(new Path(conf.get(HBASE_DIR, DEFAULT_HBASE_DIR))),
+ tableName.toString()
+ );
+ Path logdir = new Path(tabledir, "merge_" + System.currentTimeMillis() +
+ HREGION_LOGDIR_NAME);
this.hlog =
- new HLog(fs, new Path(basedir, HREGION_LOGDIR_NAME), conf, null);
+ new HLog(fs, logdir, conf, null);
}
void process() throws IOException {
try {
- while(more) {
- TreeSet<HRegionInfo> regionsToMerge = next();
- if(regionsToMerge == null) {
- break;
+ for(HRegionInfo[] regionsToMerge = next();
+ regionsToMerge != null;
+ regionsToMerge = next()) {
+
+ if (!merge(regionsToMerge)) {
+ return;
}
- merge(regionsToMerge.toArray(new HRegionInfo[regionsToMerge.size()]));
}
} finally {
try {
@@ -130,19 +125,13 @@
} catch(IOException e) {
LOG.error(e);
}
- try {
- fs.delete(basedir);
-
- } catch(IOException e) {
- LOG.error(e);
- }
}
}
- private void merge(HRegionInfo[] regions) throws IOException {
- if(regions.length < 2) {
+ private boolean merge(HRegionInfo[] info) throws IOException {
+ if(info.length < 2) {
LOG.info("only one region - nothing to merge");
- return;
+ return false;
}
HRegion currentRegion = null;
@@ -150,20 +139,18 @@
HRegion nextRegion = null;
long nextSize = 0;
Text midKey = new Text();
- for(int i = 0; i < regions.length - 1; i++) {
- if(currentRegion == null) {
+ for (int i = 0; i < info.length - 1; i++) {
+ if (currentRegion == null) {
currentRegion =
- new HRegion(dir, hlog, fs, conf, regions[i], null, null);
+ new HRegion(tabledir, hlog, fs, conf, info[i], null, null);
currentSize = currentRegion.largestHStore(midKey).getAggregate();
}
nextRegion =
- new HRegion(dir, hlog, fs, conf, regions[i + 1], null, null);
+ new HRegion(tabledir, hlog, fs, conf, info[i + 1], null, null);
nextSize = nextRegion.largestHStore(midKey).getAggregate();
- long maxFilesize =
- conf.getLong("hbase.hregion.max.filesize", DEFAULT_MAX_FILE_SIZE);
- if((currentSize + nextSize) <= (maxFilesize / 2)) {
+ if ((currentSize + nextSize) <= (maxFilesize / 2)) {
// We merge two adjacent regions if their total size is less than
// one half of the desired maximum size
@@ -175,10 +162,7 @@
updateMeta(currentRegion.getRegionName(), nextRegion.getRegionName(),
mergedRegion);
- currentRegion = null;
- i++;
- continue;
-
+ break;
}
LOG.info("not merging regions " + currentRegion.getRegionName()
+ " and " + nextRegion.getRegionName());
@@ -190,9 +174,10 @@
if(currentRegion != null) {
currentRegion.close();
}
+ return true;
}
- protected abstract TreeSet<HRegionInfo> next() throws IOException;
+ protected abstract HRegionInfo[] next() throws IOException;
protected abstract void updateMeta(Text oldRegion1, Text oldRegion2,
HRegion newRegion) throws IOException;
@@ -201,55 +186,63 @@
/** Instantiated to compact a normal user table */
private static class OnlineMerger extends Merger {
- private HTable table;
- private HScannerInterface metaScanner;
+ private final Text tableName;
+ private final HTable table;
+ private final HScannerInterface metaScanner;
private HRegionInfo latestRegion;
OnlineMerger(HBaseConfiguration conf, FileSystem fs, Text tableName)
throws IOException {
super(conf, fs, tableName);
+ this.tableName = tableName;
this.table = new HTable(conf, META_TABLE_NAME);
- this.metaScanner = table.obtainScanner(META_COLS, new Text());
+ this.metaScanner = table.obtainScanner(COL_REGIONINFO_ARRAY, tableName);
this.latestRegion = null;
}
private HRegionInfo nextRegion() throws IOException {
try {
+ HStoreKey key = new HStoreKey();
TreeMap<Text, byte[]> results = new TreeMap<Text, byte[]>();
- if(! metaScanner.next(key, results)) {
- more = false;
+ if (! metaScanner.next(key, results)) {
return null;
}
byte[] bytes = results.get(COL_REGIONINFO);
- if(bytes == null || bytes.length == 0) {
+ if (bytes == null || bytes.length == 0) {
throw new NoSuchElementException("meta region entry missing "
+ COL_REGIONINFO);
}
HRegionInfo region =
(HRegionInfo) Writables.getWritable(bytes, new HRegionInfo());
- if(!region.isOffline()) {
+ if (!region.getTableDesc().getName().equals(tableName)) {
+ return null;
+ }
+
+ if (!region.isOffline()) {
throw new TableNotDisabledException("region " + region.getRegionName()
+ " is not disabled");
}
return region;
- } catch(IOException e) {
+ } catch (IOException e) {
+ e = RemoteExceptionHandler.checkIOException(e);
+ LOG.error("meta scanner error", e);
try {
metaScanner.close();
- } catch(IOException ex) {
- LOG.error(ex);
+ } catch (IOException ex) {
+ ex = RemoteExceptionHandler.checkIOException(ex);
+ LOG.error("error closing scanner", ex);
}
- more = false;
throw e;
}
}
@Override
- protected TreeSet<HRegionInfo> next() throws IOException {
- TreeSet<HRegionInfo> regions = new TreeSet<HRegionInfo>();
+ protected HRegionInfo[] next() throws IOException {
+ List<HRegionInfo> regions = new ArrayList<HRegionInfo>();
if(latestRegion == null) {
latestRegion = nextRegion();
}
@@ -260,7 +253,7 @@
if(latestRegion != null) {
regions.add(latestRegion);
}
- return regions;
+ return regions.toArray(new HRegionInfo[regions.size()]);
}
@Override
@@ -280,6 +273,8 @@
table.delete(lockid, COL_REGIONINFO);
table.delete(lockid, COL_SERVER);
table.delete(lockid, COL_STARTCODE);
+ table.delete(lockid, COL_SPLITA);
+ table.delete(lockid, COL_SPLITB);
table.commit(lockid);
lockid = -1L;
@@ -292,14 +287,12 @@
}
}
}
- ByteArrayOutputStream byteValue = new ByteArrayOutputStream();
- DataOutputStream s = new DataOutputStream(byteValue);
newRegion.getRegionInfo().setOffline(true);
- newRegion.getRegionInfo().write(s);
long lockid = -1L;
try {
lockid = table.startUpdate(newRegion.getRegionName());
- table.put(lockid, COL_REGIONINFO, byteValue.toByteArray());
+ table.put(lockid, COL_REGIONINFO,
+ Writables.getBytes(newRegion.getRegionInfo()));
table.commit(lockid);
lockid = -1L;
@@ -317,31 +310,35 @@
/** Instantiated to compact the meta region */
private static class OfflineMerger extends Merger {
- private TreeSet<HRegionInfo> metaRegions;
- private TreeMap<Text, byte []> results;
+ private final List<HRegionInfo> metaRegions = new ArrayList<HRegionInfo>();
+ private final HRegion root;
- OfflineMerger(HBaseConfiguration conf, FileSystem fs, Text tableName)
+ OfflineMerger(HBaseConfiguration conf, FileSystem fs)
throws IOException {
- super(conf, fs, tableName);
- this.metaRegions = new TreeSet<HRegionInfo>();
- this.results = new TreeMap<Text, byte []>();
+ super(conf, fs, META_TABLE_NAME);
+
+ Path rootTableDir = HTableDescriptor.getTableDir(
+ fs.makeQualified(new Path(conf.get(HBASE_DIR, DEFAULT_HBASE_DIR))),
+ ROOT_TABLE_NAME);
// Scan root region to find all the meta regions
- HRegion root =
- new HRegion(dir, hlog,fs, conf, HRegionInfo.rootRegionInfo, null, null);
+ root = new HRegion(rootTableDir, hlog, fs, conf,
+ HRegionInfo.rootRegionInfo, null, null);
- HScannerInterface rootScanner =
- root.getScanner(META_COLS, new Text(), System.currentTimeMillis(), null);
+ HScannerInterface rootScanner = root.getScanner(COL_REGIONINFO_ARRAY,
+ new Text(), System.currentTimeMillis(), null);
try {
+ HStoreKey key = new HStoreKey();
+ TreeMap<Text, byte[]> results = new TreeMap<Text, byte[]>();
while(rootScanner.next(key, results)) {
for(byte [] b: results.values()) {
- in.reset(b, b.length);
- info.readFields(in);
- metaRegions.add(info);
- results.clear();
+ HRegionInfo info = Writables.getHRegionInfoOrNull(b);
+ if (info != null) {
+ metaRegions.add(info);
+ }
}
}
} finally {
@@ -356,18 +353,19 @@
}
@Override
- protected TreeSet<HRegionInfo> next() {
- more = false;
- return metaRegions;
+ protected HRegionInfo[] next() {
+ HRegionInfo[] results = null;
+ if (metaRegions.size() > 0) {
+ results = metaRegions.toArray(new HRegionInfo[metaRegions.size()]);
+ metaRegions.clear();
+ }
+ return results;
}
@Override
protected void updateMeta(Text oldRegion1, Text oldRegion2,
HRegion newRegion) throws IOException {
- HRegion root =
- new HRegion(dir, hlog, fs, conf, HRegionInfo.rootRegionInfo, null, null);
-
Text[] regionsToDelete = {
oldRegion1,
oldRegion2
@@ -379,6 +377,8 @@
b.delete(lockid, COL_REGIONINFO);
b.delete(lockid, COL_SERVER);
b.delete(lockid, COL_STARTCODE);
+ b.delete(lockid, COL_SPLITA);
+ b.delete(lockid, COL_SPLITB);
root.batchUpdate(System.currentTimeMillis(), b);
lockid = -1L;
@@ -386,14 +386,12 @@
LOG.debug("updated columns in row: " + regionsToDelete[r]);
}
}
- ByteArrayOutputStream byteValue = new ByteArrayOutputStream();
- DataOutputStream s = new DataOutputStream(byteValue);
- newRegion.getRegionInfo().setOffline(true);
- newRegion.getRegionInfo().write(s);
+ HRegionInfo newInfo = newRegion.getRegionInfo();
+ newInfo.setOffline(true);
long lockid = Math.abs(rand.nextLong());
BatchUpdate b = new BatchUpdate(lockid);
lockid = b.startUpdate(newRegion.getRegionName());
- b.put(lockid, COL_REGIONINFO, byteValue.toByteArray());
+ b.put(lockid, COL_REGIONINFO, Writables.getBytes(newInfo));
root.batchUpdate(System.currentTimeMillis(), b);
if(LOG.isDebugEnabled()) {
LOG.debug("updated columns in row: " + newRegion.getRegionName());
Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegion.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegion.java?rev=611519&r1=611518&r2=611519&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegion.java (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegion.java Sat Jan 12 17:36:08 2008
@@ -122,7 +122,7 @@
HBaseConfiguration conf = a.getConf();
HTableDescriptor tabledesc = a.getTableDesc();
HLog log = a.getLog();
- Path rootDir = a.getRootDir();
+ Path basedir = a.getBaseDir();
Text startKey = a.getStartKey();
Text endKey = b.getEndKey();
Path merges = new Path(a.getRegionDir(), MERGEDIR);
@@ -131,8 +131,8 @@
}
HRegionInfo newRegionInfo = new HRegionInfo(tabledesc, startKey, endKey);
- Path newRegionDir = HRegion.getRegionDir(merges,
- HRegionInfo.encodeRegionName(newRegionInfo.getRegionName()));
+ Path newRegionDir =
+ HRegion.getRegionDir(merges, newRegionInfo.getEncodedName());
if(fs.exists(newRegionDir)) {
throw new IOException("Cannot merge; target file collision at " +
newRegionDir);
@@ -148,15 +148,14 @@
for (Map.Entry<Text, List<HStoreFile>> es : byFamily.entrySet()) {
Text colFamily = es.getKey();
List<HStoreFile> srcFiles = es.getValue();
- HStoreFile dst = new HStoreFile(conf, merges,
- HRegionInfo.encodeRegionName(newRegionInfo.getRegionName()),
- colFamily, Math.abs(rand.nextLong()));
+ HStoreFile dst = new HStoreFile(conf, fs, merges,
+ newRegionInfo.getEncodedName(), colFamily, -1, null);
dst.mergeStoreFiles(srcFiles, fs, conf);
}
// Done
// Construction moves the merge files into place under region.
- HRegion dstRegion = new HRegion(rootDir, log, fs, conf, newRegionInfo,
+ HRegion dstRegion = new HRegion(basedir, log, fs, conf, newRegionInfo,
newRegionDir, null);
// Get rid of merges directory
@@ -199,7 +198,7 @@
final AtomicLong memcacheSize = new AtomicLong(0);
- final Path rootDir;
+ final Path basedir;
final HLog log;
final FileSystem fs;
final HBaseConfiguration conf;
@@ -228,7 +227,6 @@
private final Integer splitLock = new Integer(0);
private final long desiredMaxFileSize;
private final long minSequenceId;
- private final String encodedRegionName;
final AtomicInteger activeScannerCount = new AtomicInteger(0);
//////////////////////////////////////////////////////////////////////////////
@@ -245,7 +243,8 @@
* appropriate log info for this HRegion. If there is a previous log file
* (implying that the HRegion has been written-to before), then read it from
* the supplied path.
- * @param rootDir root directory for HBase instance
+ * @param basedir qualified path of directory where region should be located,
+ * usually the table directory.
* @param fs is the filesystem.
* @param conf is global configuration settings.
* @param regionInfo - HRegionInfo that describes the region
@@ -255,22 +254,17 @@
*
* @throws IOException
*/
- public HRegion(Path rootDir, HLog log, FileSystem fs, HBaseConfiguration conf,
+ public HRegion(Path basedir, HLog log, FileSystem fs, HBaseConfiguration conf,
HRegionInfo regionInfo, Path initialFiles, CacheFlushListener listener)
throws IOException {
- this.rootDir = rootDir;
+ this.basedir = basedir;
this.log = log;
this.fs = fs;
this.conf = conf;
this.regionInfo = regionInfo;
- this.encodedRegionName =
- HRegionInfo.encodeRegionName(this.regionInfo.getRegionName());
this.threadWakeFrequency = conf.getLong(THREAD_WAKE_FREQUENCY, 10 * 1000);
-
- // Declare the regionName. This is a unique string for the region, used to
- // build a unique filename.
- this.regiondir = HRegion.getRegionDir(rootDir, this.encodedRegionName);
+ this.regiondir = new Path(basedir, this.regionInfo.getEncodedName());
Path oldLogFile = new Path(regiondir, HREGION_OLDLOGFILE_NAME);
// Move prefab HStore files into place (if any). This picks up split files
@@ -281,14 +275,13 @@
// Load in all the HStores.
long maxSeqId = -1;
- for(Map.Entry<Text, HColumnDescriptor> e :
- this.regionInfo.getTableDesc().families().entrySet()) {
- Text colFamily = HStoreKey.extractFamily(e.getKey()).toText();
+ for(HColumnDescriptor c :
+ this.regionInfo.getTableDesc().families().values()) {
- HStore store = new HStore(rootDir, this.regionInfo.getRegionName(),
- this.encodedRegionName, e.getValue(), fs, oldLogFile, conf);
+ HStore store = new HStore(this.basedir, this.regionInfo, c, this.fs,
+ oldLogFile, this.conf);
- stores.put(colFamily, store);
+ stores.put(c.getFamilyName(), store);
long storeSeqId = store.getMaxSequenceId();
if (storeSeqId > maxSeqId) {
@@ -450,7 +443,7 @@
}
/** @return region id */
- public long getRegionId() {
+ long getRegionId() {
return this.regionInfo.getRegionId();
}
@@ -459,13 +452,8 @@
return this.regionInfo.getRegionName();
}
- /** @return root directory path */
- public Path getRootDir() {
- return rootDir;
- }
-
/** @return HTableDescriptor for this region */
- public HTableDescriptor getTableDesc() {
+ HTableDescriptor getTableDesc() {
return this.regionInfo.getTableDesc();
}
@@ -475,19 +463,17 @@
}
/** @return Configuration object */
- public HBaseConfiguration getConf() {
+ HBaseConfiguration getConf() {
return this.conf;
}
- /** @return region directory Path
- * @see HRegion#getRegionDir(Path, String)
- */
- public Path getRegionDir() {
+ /** @return region directory Path */
+ Path getRegionDir() {
return this.regiondir;
}
/** @return FileSystem being used by this region */
- public FileSystem getFilesystem() {
+ FileSystem getFilesystem() {
return this.fs;
}
@@ -550,18 +536,19 @@
return null;
}
long startTime = System.currentTimeMillis();
- Path splits = getSplitsDir();
+ Path splits = new Path(this.regiondir, SPLITDIR);
+ if(!this.fs.exists(splits)) {
+ this.fs.mkdirs(splits);
+ }
HRegionInfo regionAInfo = new HRegionInfo(this.regionInfo.getTableDesc(),
this.regionInfo.getStartKey(), midKey);
- Path dirA = getSplitRegionDir(splits,
- HRegionInfo.encodeRegionName(regionAInfo.getRegionName()));
+ Path dirA = new Path(splits, regionAInfo.getEncodedName());
if(fs.exists(dirA)) {
throw new IOException("Cannot split; target file collision at " + dirA);
}
HRegionInfo regionBInfo = new HRegionInfo(this.regionInfo.getTableDesc(),
midKey, this.regionInfo.getEndKey());
- Path dirB = getSplitRegionDir(splits,
- HRegionInfo.encodeRegionName(regionBInfo.getRegionName()));
+ Path dirB = new Path(splits, regionBInfo.getEncodedName());
if(this.fs.exists(dirB)) {
throw new IOException("Cannot split; target file collision at " + dirB);
}
@@ -592,18 +579,16 @@
for(HStoreFile h: hstoreFilesToSplit) {
// A reference to the bottom half of the hsf store file.
HStoreFile.Reference aReference = new HStoreFile.Reference(
- this.encodedRegionName, h.getFileId(), new HStoreKey(midKey),
- HStoreFile.Range.bottom);
- HStoreFile a = new HStoreFile(this.conf, splits,
- HRegionInfo.encodeRegionName(regionAInfo.getRegionName()),
- h.getColFamily(), Math.abs(rand.nextLong()), aReference);
+ this.regionInfo.getEncodedName(), h.getFileId(),
+ new HStoreKey(midKey), HStoreFile.Range.bottom);
+ HStoreFile a = new HStoreFile(this.conf, fs, splits,
+ regionAInfo.getEncodedName(), h.getColFamily(), -1, aReference);
// Reference to top half of the hsf store file.
HStoreFile.Reference bReference = new HStoreFile.Reference(
- this.encodedRegionName, h.getFileId(), new HStoreKey(midKey),
- HStoreFile.Range.top);
- HStoreFile b = new HStoreFile(this.conf, splits,
- HRegionInfo.encodeRegionName(regionBInfo.getRegionName()),
- h.getColFamily(), Math.abs(rand.nextLong()), bReference);
+ this.regionInfo.getEncodedName(), h.getFileId(),
+ new HStoreKey(midKey), HStoreFile.Range.top);
+ HStoreFile b = new HStoreFile(this.conf, fs, splits,
+ regionBInfo.getEncodedName(), h.getColFamily(), -1, bReference);
h.splitStoreFile(a, b, this.fs);
}
@@ -611,10 +596,10 @@
// Opening the region copies the splits files from the splits directory
// under each region.
HRegion regionA =
- new HRegion(rootDir, log, fs, conf, regionAInfo, dirA, null);
+ new HRegion(basedir, log, fs, conf, regionAInfo, dirA, null);
regionA.close();
HRegion regionB =
- new HRegion(rootDir, log, fs, conf, regionBInfo, dirB, null);
+ new HRegion(basedir, log, fs, conf, regionBInfo, dirB, null);
regionB.close();
// Cleanup
@@ -673,18 +658,6 @@
return split;
}
- private Path getSplitRegionDir(final Path splits, final String region) {
- return HRegion.getRegionDir(splits, region);
- }
-
- private Path getSplitsDir() throws IOException {
- Path splits = new Path(this.regiondir, SPLITDIR);
- if(!this.fs.exists(splits)) {
- this.fs.mkdirs(splits);
- }
- return splits;
- }
-
/**
* Only do a compaction if it is necessary
*
@@ -1511,7 +1484,11 @@
/** {@inheritDoc} */
@Override
public String toString() {
- return getRegionName().toString();
+ return regionInfo.getRegionName().toString();
+ }
+
+ private Path getBaseDir() {
+ return this.basedir;
}
/**
@@ -1707,13 +1684,13 @@
* @throws IOException
*/
static HRegion createHRegion(final HRegionInfo info, final Path rootDir,
- final HBaseConfiguration conf)
- throws IOException {
- Path regionDir = HRegion.getRegionDir(rootDir,
- HRegionInfo.encodeRegionName(info.getRegionName()));
+ final HBaseConfiguration conf) throws IOException {
+ Path tableDir =
+ HTableDescriptor.getTableDir(rootDir, info.getTableDesc().getName());
+ Path regionDir = HRegion.getRegionDir(tableDir, info.getEncodedName());
FileSystem fs = FileSystem.get(conf);
fs.mkdirs(regionDir);
- return new HRegion(rootDir,
+ return new HRegion(tableDir,
new HLog(fs, new Path(regionDir, HREGION_LOGDIR_NAME), conf, null),
fs, conf, info, null, null);
}
@@ -1749,15 +1726,14 @@
* Deletes all the files for a HRegion
*
* @param fs the file system object
- * @param baseDirectory base directory for HBase
- * @param name region file name ENCODED!
+ * @param rootdir qualified path of HBase root directory
+ * @param info HRegionInfo for region to be deleted
* @throws IOException
* @return True if deleted.
- * @see HRegionInfo#encodeRegionName(Text)
*/
- static boolean deleteRegion(FileSystem fs, Path baseDirectory, String name)
- throws IOException {
- Path p = HRegion.getRegionDir(fs.makeQualified(baseDirectory), name);
+ static boolean deleteRegion(FileSystem fs, Path rootdir, HRegionInfo info)
+ throws IOException {
+ Path p = HRegion.getRegionDir(rootdir, info);
if (LOG.isDebugEnabled()) {
LOG.debug("DELETING region " + p.toString());
}
@@ -1767,12 +1743,26 @@
/**
* Computes the Path of the HRegion
*
- * @param dir hbase home directory
+ * @param tabledir qualified path for table
* @param name region file name ENCODED!
* @return Path of HRegion directory
* @see HRegionInfo#encodeRegionName(Text)
*/
- public static Path getRegionDir(final Path dir, final String name) {
- return new Path(dir, new Path(HREGIONDIR_PREFIX + name));
+ static Path getRegionDir(final Path tabledir, final String name) {
+ return new Path(tabledir, name);
+ }
+
+ /**
+ * Computes the Path of the HRegion
+ *
+ * @param rootdir qualified path of HBase root directory
+ * @param info HRegionInfo for the region
+ * @return qualified path of region directory
+ */
+ static Path getRegionDir(final Path rootdir, final HRegionInfo info) {
+ return new Path(
+ HTableDescriptor.getTableDir(rootdir, info.getTableDesc().getName()),
+ info.getEncodedName()
+ );
}
}
Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionInfo.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionInfo.java?rev=611519&r1=611518&r2=611519&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionInfo.java (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionInfo.java Sat Jan 12 17:36:08 2008
@@ -79,6 +79,7 @@
private Text startKey;
private HTableDescriptor tableDesc;
private int hashCode;
+ private transient String encodedName = null;
private void setHashCode() {
int result = this.regionName.hashCode();
@@ -182,6 +183,14 @@
public Text getRegionName(){
return regionName;
}
+
+ /** @return the encoded region name */
+ public synchronized String getEncodedName() {
+ if (encodedName == null) {
+ encodedName = encodeRegionName(regionName);
+ }
+ return encodedName;
+ }
/** @return the startKey */
public Text getStartKey(){
@@ -242,10 +251,8 @@
@Override
public String toString() {
return "regionname: " + this.regionName.toString() + ", startKey: <" +
- this.startKey.toString() + ">, encodedName(" +
- encodeRegionName(this.regionName) + ")" +
- (isOffline()? " offline: true,": "") +
- (isSplit()? " split: true,": "") +
+ this.startKey.toString() + ">, encodedName(" + getEncodedName() + ")" +
+ (isOffline()? " offline: true,": "") + (isSplit()? " split: true,": "") +
" tableDesc: {" + this.tableDesc.toString() + "}";
}
Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionServer.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionServer.java?rev=611519&r1=611518&r2=611519&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionServer.java (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionServer.java Sat Jan 12 17:36:08 2008
@@ -92,6 +92,8 @@
protected final HServerInfo serverInfo;
protected final HBaseConfiguration conf;
+ private FileSystem fs;
+ private Path rootDir;
private final Random rand = new Random();
// region name -> HRegion
@@ -138,15 +140,18 @@
* is registered as a shutdown hook in the HRegionServer constructor and is
* only called when the HRegionServer receives a kill signal.
*/
- class ShutdownThread
- extends Thread {
-
+ class ShutdownThread extends Thread {
private final HRegionServer instance;
+ /**
+ * @param instance
+ */
public ShutdownThread(HRegionServer instance) {
this.instance = instance;
}
+ /** {@inheritDoc} */
+ @Override
public synchronized void start() {
LOG.info("Starting shutdown thread.");
@@ -914,6 +919,8 @@
}
this.conf.set(key, value);
}
+ this.fs = FileSystem.get(this.conf);
+ this.rootDir = new Path(this.conf.get(HConstants.HBASE_DIR));
this.log = setupHLog();
startServiceThreads();
} catch (IOException e) {
@@ -929,15 +936,12 @@
private HLog setupHLog() throws RegionServerRunningException,
IOException {
- String rootDir = this.conf.get(HConstants.HBASE_DIR);
- LOG.info("Root dir: " + rootDir);
- Path logdir = new Path(new Path(rootDir), "log" + "_" + getThisIP() + "_" +
+ Path logdir = new Path(rootDir, "log" + "_" + getThisIP() + "_" +
this.serverInfo.getStartCode() + "_" +
this.serverInfo.getServerAddress().getPort());
if (LOG.isDebugEnabled()) {
LOG.debug("Log dir " + logdir);
}
- FileSystem fs = FileSystem.get(this.conf);
if (fs.exists(logdir)) {
throw new RegionServerRunningException("region server already " +
"running at " + this.serverInfo.getServerAddress().toString() +
@@ -1209,9 +1213,12 @@
HRegion region = onlineRegions.get(regionInfo.getRegionName());
if(region == null) {
try {
- region = new HRegion(new Path(this.conf.get(HConstants.HBASE_DIR)),
- this.log, FileSystem.get(conf), conf, regionInfo, null,
- this.cacheFlusher);
+ region = new HRegion(
+ HTableDescriptor.getTableDir(rootDir,
+ regionInfo.getTableDesc().getName()
+ ),
+ this.log, this.fs, conf, regionInfo, null, this.cacheFlusher
+ );
} catch (IOException e) {
LOG.error("error opening region " + regionInfo.getRegionName(), e);
@@ -1651,9 +1658,7 @@
*/
protected boolean checkFileSystem() {
if (this.fsOk) {
- FileSystem fs = null;
try {
- fs = FileSystem.get(this.conf);
if (fs != null && !FSUtils.isFileSystemAvailable(fs)) {
LOG.fatal("Shutting down HRegionServer: file system not available");
this.abortRequested = true;
Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HStore.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HStore.java?rev=611519&r1=611518&r2=611519&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HStore.java (original)
+++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HStore.java Sat Jan 12 17:36:08 2008
@@ -19,7 +19,6 @@
*/
package org.apache.hadoop.hbase;
-import java.io.DataInputStream;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
@@ -28,13 +27,13 @@
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Random;
import java.util.SortedMap;
import java.util.TreeMap;
-import java.util.Vector;
import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -64,7 +63,7 @@
* Locking and transactions are handled at a higher level. This API should not
* be called directly by any writer, but rather by an HRegion manager.
*/
-class HStore implements HConstants {
+public class HStore implements HConstants {
static final Log LOG = LogFactory.getLog(HStore.class);
/**
@@ -474,50 +473,51 @@
}
}
- static final String COMPACTION_TO_REPLACE = "toreplace";
- static final String COMPACTION_DONE = "done";
+ /*
+ * Regex that will work for straight filenames and for reference names.
+ * If reference, then the regex has more than just one group. Group 1 is
+ * this files id. Group 2 the referenced region name, etc.
+ */
+ private static Pattern REF_NAME_PARSER =
+ Pattern.compile("^(\\d+)(?:\\.(.+))?$");
private static final String BLOOMFILTER_FILE_NAME = "filter";
final Memcache memcache = new Memcache();
- Path dir;
- Text regionName;
- String encodedRegionName;
- HColumnDescriptor family;
- Text familyName;
- SequenceFile.CompressionType compression;
- FileSystem fs;
- HBaseConfiguration conf;
- Path mapdir;
- Path loginfodir;
- Path filterDir;
- Filter bloomFilter;
- private String storeName;
+ private final Path basedir;
+ private final HRegionInfo info;
+ private final HColumnDescriptor family;
+ private final SequenceFile.CompressionType compression;
+ final FileSystem fs;
+ private final HBaseConfiguration conf;
+ private final Path filterDir;
+ final Filter bloomFilter;
private final Path compactionDir;
- Integer compactLock = new Integer(0);
- Integer flushLock = new Integer(0);
+ private final Integer compactLock = new Integer(0);
+ private final Integer flushLock = new Integer(0);
- final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+ private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
final AtomicInteger activeScanners = new AtomicInteger(0);
- /* Sorted Map of readers keyed by sequence id (Most recent should be last in
+ final String storeName;
+
+ /*
+ * Sorted Map of readers keyed by sequence id (Most recent should be last in
* in list).
*/
- SortedMap<Long, HStoreFile> storefiles =
+ final SortedMap<Long, HStoreFile> storefiles =
Collections.synchronizedSortedMap(new TreeMap<Long, HStoreFile>());
- /* Sorted Map of readers keyed by sequence id (Most recent should be last in
+ /*
+ * Sorted Map of readers keyed by sequence id (Most recent should be last in
* in list).
*/
- TreeMap<Long, MapFile.Reader> readers = new TreeMap<Long, MapFile.Reader>();
+ private final SortedMap<Long, MapFile.Reader> readers =
+ new TreeMap<Long, MapFile.Reader>();
- Random rand = new Random();
-
private volatile long maxSeqId;
-
private final int compactionThreshold;
-
private final ReentrantReadWriteLock newScannerLock =
new ReentrantReadWriteLock();
@@ -545,70 +545,72 @@
* <p>It's assumed that after this constructor returns, the reconstructionLog
* file will be deleted (by whoever has instantiated the HStore).
*
- * @param dir log file directory
- * @param regionName
- * @param encodedName
- * @param family name of column family
+ * @param basedir qualified path under which the region directory lives
+ * @param info HRegionInfo for this region
+ * @param family HColumnDescriptor for this column
* @param fs file system object
* @param reconstructionLog existing log file to apply if any
* @param conf configuration object
* @throws IOException
*/
- HStore(Path dir, Text regionName, String encodedName,
- HColumnDescriptor family, FileSystem fs, Path reconstructionLog,
- HBaseConfiguration conf) throws IOException {
-
- this.dir = dir;
- this.compactionDir = new Path(HRegion.getRegionDir(dir, encodedName),
- "compaction.dir");
- this.regionName = regionName;
- this.encodedRegionName = encodedName;
- this.family = family;
- this.familyName = HStoreKey.extractFamily(this.family.getName()).toText();
- this.compression = SequenceFile.CompressionType.NONE;
- this.storeName = this.encodedRegionName + "/" + this.familyName.toString();
-
- if(family.getCompression() != HColumnDescriptor.CompressionType.NONE) {
- if(family.getCompression() == HColumnDescriptor.CompressionType.BLOCK) {
- this.compression = SequenceFile.CompressionType.BLOCK;
- } else if(family.getCompression() ==
- HColumnDescriptor.CompressionType.RECORD) {
- this.compression = SequenceFile.CompressionType.RECORD;
- } else {
- assert(false);
- }
- }
+ HStore(Path basedir, HRegionInfo info, HColumnDescriptor family,
+ FileSystem fs, Path reconstructionLog, HBaseConfiguration conf)
+ throws IOException {
+ this.basedir = basedir;
+ this.info = info;
+ this.family = family;
this.fs = fs;
this.conf = conf;
- this.mapdir = HStoreFile.getMapDir(dir, encodedRegionName, familyName);
- fs.mkdirs(mapdir);
- this.loginfodir = HStoreFile.getInfoDir(dir, encodedRegionName, familyName);
- fs.mkdirs(loginfodir);
+
+ this.compactionDir = new Path(basedir, "compaction.dir");
+ this.storeName =
+ this.info.getEncodedName() + "/" + this.family.getFamilyName();
+
+ if (family.getCompression() == HColumnDescriptor.CompressionType.BLOCK) {
+ this.compression = SequenceFile.CompressionType.BLOCK;
+ } else if (family.getCompression() ==
+ HColumnDescriptor.CompressionType.RECORD) {
+ this.compression = SequenceFile.CompressionType.RECORD;
+ } else {
+ this.compression = SequenceFile.CompressionType.NONE;
+ }
+
+ Path mapdir = HStoreFile.getMapDir(basedir, info.getEncodedName(),
+ family.getFamilyName());
+ if (!fs.exists(mapdir)) {
+ fs.mkdirs(mapdir);
+ }
+ Path infodir = HStoreFile.getInfoDir(basedir, info.getEncodedName(),
+ family.getFamilyName());
+ if (!fs.exists(infodir)) {
+ fs.mkdirs(infodir);
+ }
+
if(family.getBloomFilter() == null) {
this.filterDir = null;
this.bloomFilter = null;
} else {
- this.filterDir =
- HStoreFile.getFilterDir(dir, encodedRegionName, familyName);
- fs.mkdirs(filterDir);
- loadOrCreateBloomFilter();
+ this.filterDir = HStoreFile.getFilterDir(basedir, info.getEncodedName(),
+ family.getFamilyName());
+ if (!fs.exists(filterDir)) {
+ fs.mkdirs(filterDir);
+ }
+ this.bloomFilter = loadOrCreateBloomFilter();
}
if(LOG.isDebugEnabled()) {
- LOG.debug("starting " + this.regionName + "/" + this.familyName + " ("
- + this.storeName +
+ LOG.debug("starting " + storeName +
((reconstructionLog == null || !fs.exists(reconstructionLog)) ?
- ") (no reconstruction log)": " with reconstruction log: (" +
- reconstructionLog.toString()));
+ " (no reconstruction log)" :
+ " with reconstruction log: " + reconstructionLog.toString()));
}
- // Go through the 'mapdir' and 'loginfodir' together, make sure that all
+ // Go through the 'mapdir' and 'infodir' together, make sure that all
// MapFiles are in a reliable state. Every entry in 'mapdir' must have a
// corresponding one in 'loginfodir'. Without a corresponding log info
// file, the entry in 'mapdir' must be deleted.
- List<HStoreFile> hstoreFiles = HStoreFile.loadHStoreFiles(conf, dir,
- encodedRegionName, familyName, fs);
+ List<HStoreFile> hstoreFiles = loadHStoreFiles(infodir, mapdir);
for(HStoreFile hsf: hstoreFiles) {
this.storefiles.put(Long.valueOf(hsf.loadInfo(fs)), hsf);
}
@@ -624,8 +626,8 @@
this.maxSeqId = getMaxSequenceId(hstoreFiles);
if (LOG.isDebugEnabled()) {
- LOG.debug("maximum sequence id for hstore " + regionName + "/" +
- familyName + " (" + storeName + ") is " + this.maxSeqId);
+ LOG.debug("maximum sequence id for hstore " + storeName + " is " +
+ this.maxSeqId);
}
doReconstructionLog(reconstructionLog, maxSeqId);
@@ -693,14 +695,14 @@
TreeMap<HStoreKey, byte []> reconstructedCache =
new TreeMap<HStoreKey, byte []>();
- SequenceFile.Reader login = new SequenceFile.Reader(this.fs,
+ SequenceFile.Reader logReader = new SequenceFile.Reader(this.fs,
reconstructionLog, this.conf);
try {
HLogKey key = new HLogKey();
HLogEdit val = new HLogEdit();
long skippedEdits = 0;
- while (login.next(key, val)) {
+ while (logReader.next(key, val)) {
maxSeqIdInLog = Math.max(maxSeqIdInLog, key.getLogSeqNum());
if (key.getLogSeqNum() <= maxSeqID) {
skippedEdits++;
@@ -714,14 +716,14 @@
// METACOLUMN info such as HBASE::CACHEFLUSH entries
Text column = val.getColumn();
if (column.equals(HLog.METACOLUMN)
- || !key.getRegionName().equals(regionName)
- || !HStoreKey.extractFamily(column).equals(this.familyName)) {
+ || !key.getRegionName().equals(info.getRegionName())
+ || !HStoreKey.extractFamily(column).equals(family.getFamilyName())) {
if (LOG.isDebugEnabled()) {
LOG.debug("Passing on edit " + key.getRegionName() + ", " +
column.toString() + ": " +
new String(val.getVal(), UTF8_ENCODING) +
- ", my region: " + regionName + ", my column: " +
- this.familyName);
+ ", my region: " + info.getRegionName() + ", my column: " +
+ family.getFamilyName());
}
continue;
}
@@ -733,7 +735,7 @@
reconstructedCache.put(k, val.getVal());
}
} finally {
- login.close();
+ logReader.close();
}
if (reconstructedCache.size() > 0) {
@@ -745,6 +747,76 @@
}
}
+ /*
+ * Creates a series of HStoreFiles loaded from the given directory.
+ * There must be a matching 'mapdir' and 'loginfo' pair of files.
+ * If only one exists, we'll delete it.
+ *
+ * @param infodir qualified path for info file directory
+ * @param mapdir qualified path for map file directory
+ * @throws IOException
+ */
+ private List<HStoreFile> loadHStoreFiles(Path infodir, Path mapdir)
+ throws IOException {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("infodir: " + infodir.toString() + " mapdir: " +
+ mapdir.toString());
+ }
+ // Look first at info files. If a reference, these contain info we need
+ // to create the HStoreFile.
+ Path infofiles[] = fs.listPaths(new Path[] {infodir});
+ ArrayList<HStoreFile> results = new ArrayList<HStoreFile>(infofiles.length);
+ ArrayList<Path> mapfiles = new ArrayList<Path>(infofiles.length);
+ for (Path p: infofiles) {
+ Matcher m = REF_NAME_PARSER.matcher(p.getName());
+ /*
+ * * * * * N O T E * * * * *
+ *
+ * We call isReference(Path, Matcher) here because it calls
+ * Matcher.matches() which must be called before Matcher.group(int)
+ * and we don't want to call Matcher.matches() twice.
+ *
+ * * * * * N O T E * * * * *
+ */
+ boolean isReference = isReference(p, m);
+ long fid = Long.parseLong(m.group(1));
+
+ HStoreFile curfile = null;
+ HStoreFile.Reference reference = null;
+ if (isReference) {
+ reference = readSplitInfo(p, fs);
+ }
+ curfile = new HStoreFile(conf, fs, basedir, info.getEncodedName(),
+ family.getFamilyName(), fid, reference);
+ Path mapfile = curfile.getMapFilePath();
+ if (!fs.exists(mapfile)) {
+ fs.delete(curfile.getInfoFilePath());
+ LOG.warn("Mapfile " + mapfile.toString() + " does not exist. " +
+ "Cleaned up info file. Continuing...");
+ continue;
+ }
+
+ // TODO: Confirm referent exists.
+
+ // Found map and sympathetic info file. Add this hstorefile to result.
+ results.add(curfile);
+ // Keep list of sympathetic data mapfiles for cleaning info dir in next
+ // section. Make sure path is fully qualified for compare.
+ mapfiles.add(mapfile);
+ }
+
+ // List paths by experience returns fully qualified names -- at least when
+ // running on a mini hdfs cluster.
+ Path datfiles[] = fs.listPaths(new Path[] {mapdir});
+ for (int i = 0; i < datfiles.length; i++) {
+ // If does not have sympathetic info file, delete.
+ if (!mapfiles.contains(fs.makeQualified(datfiles[i]))) {
+ fs.delete(datfiles[i]);
+ }
+ }
+ return results;
+ }
+
//////////////////////////////////////////////////////////////////////////////
// Bloom filters
//////////////////////////////////////////////////////////////////////////////
@@ -754,12 +826,12 @@
* If the HStore already exists, it will read in the bloom filter saved
* previously. Otherwise, it will create a new bloom filter.
*/
- private void loadOrCreateBloomFilter() throws IOException {
+ private Filter loadOrCreateBloomFilter() throws IOException {
Path filterFile = new Path(filterDir, BLOOMFILTER_FILE_NAME);
+ Filter bloomFilter = null;
if(fs.exists(filterFile)) {
if (LOG.isDebugEnabled()) {
- LOG.debug("loading bloom filter for " + this.regionName + "/" +
- this.familyName + " (" + this.storeName + ")");
+ LOG.debug("loading bloom filter for " + this.storeName);
}
BloomFilterDescriptor.BloomFilterType type =
@@ -777,6 +849,11 @@
case RETOUCHED_BLOOMFILTER:
bloomFilter = new RetouchedBloomFilter();
+ break;
+
+ default:
+ throw new IllegalArgumentException("unknown bloom filter type: " +
+ type);
}
FSDataInputStream in = fs.open(filterFile);
try {
@@ -786,8 +863,7 @@
}
} else {
if (LOG.isDebugEnabled()) {
- LOG.debug("creating bloom filter for " + this.regionName + "/" +
- this.familyName + " (" + this.storeName + ")");
+ LOG.debug("creating bloom filter for " + this.storeName);
}
BloomFilterDescriptor.BloomFilterType type =
@@ -812,6 +888,7 @@
family.getBloomFilter().nbHash);
}
}
+ return bloomFilter;
}
/**
@@ -821,8 +898,7 @@
*/
private void flushBloomFilter() throws IOException {
if (LOG.isDebugEnabled()) {
- LOG.debug("flushing bloom filter for " + this.regionName + "/" +
- this.familyName + " (" + this.storeName + ")");
+ LOG.debug("flushing bloom filter for " + this.storeName);
}
FSDataOutputStream out =
fs.create(new Path(filterDir, BLOOMFILTER_FILE_NAME));
@@ -832,8 +908,7 @@
out.close();
}
if (LOG.isDebugEnabled()) {
- LOG.debug("flushed bloom filter for " + this.regionName + "/" +
- this.familyName + " (" + this.storeName + ")");
+ LOG.debug("flushed bloom filter for " + this.storeName);
}
}
@@ -875,8 +950,7 @@
this.readers.clear();
result = new ArrayList<HStoreFile>(storefiles.values());
this.storefiles.clear();
- LOG.debug("closed " + this.regionName + "/" + this.familyName + " ("
- + this.storeName + ")");
+ LOG.debug("closed " + this.storeName);
return result;
} finally {
this.lock.writeLock().unlock();
@@ -918,8 +992,8 @@
synchronized(flushLock) {
// A. Write the Maps out to the disk
- HStoreFile flushedFile = HStoreFile.obtainNewHStoreFile(conf, dir,
- encodedRegionName, familyName, fs);
+ HStoreFile flushedFile = new HStoreFile(conf, fs, basedir,
+ info.getEncodedName(), family.getFamilyName(), -1L, null);
String name = flushedFile.toString();
MapFile.Writer out = flushedFile.getWriter(this.fs, this.compression,
this.bloomFilter);
@@ -940,7 +1014,7 @@
for (Map.Entry<HStoreKey, byte []> es: cache.entrySet()) {
HStoreKey curkey = es.getKey();
TextSequence f = HStoreKey.extractFamily(curkey.getColumn());
- if (f.equals(this.familyName)) {
+ if (f.equals(this.family.getFamilyName())) {
entries++;
out.append(curkey, new ImmutableBytesWritable(es.getValue()));
}
@@ -970,7 +1044,7 @@
LOG.debug("Added " + name + " with " + entries +
" entries, sequence id " + logCacheFlushId + ", and size " +
StringUtils.humanReadableInt(flushedFile.length()) + " for " +
- this.regionName + "/" + this.familyName);
+ this.storeName);
}
} finally {
this.lock.writeLock().unlock();
@@ -991,7 +1065,7 @@
if (this.storefiles != null) {
compactionNeeded = this.storefiles.size() >= this.compactionThreshold;
if (LOG.isDebugEnabled()) {
- LOG.debug("compaction for HStore " + regionName + "/" + familyName +
+ LOG.debug("compaction for HStore " + storeName +
(compactionNeeded ? " " : " not ") + "needed.");
}
}
@@ -1019,17 +1093,16 @@
*/
boolean compact() throws IOException {
synchronized (compactLock) {
- Path curCompactStore = getCompactionDir();
if (LOG.isDebugEnabled()) {
LOG.debug("started compaction of " + storefiles.size() +
- " files using " + curCompactStore.toString() + " for " +
- this.regionName + "/" + this.familyName);
+ " files using " + compactionDir.toString() + " for " +
+ this.storeName);
}
- if (this.fs.exists(curCompactStore)) {
+ if (this.fs.exists(compactionDir)) {
// Clean out its content in prep. for this new compaction. Has either
// aborted previous compaction or it has content of a previous
// compaction.
- Path [] toRemove = this.fs.listPaths(new Path [] {curCompactStore});
+ Path [] toRemove = this.fs.listPaths(new Path [] {compactionDir});
for (int i = 0; i < toRemove.length; i++) {
this.fs.delete(toRemove[i]);
}
@@ -1042,22 +1115,21 @@
if (filesToCompact.size() < 1 ||
(filesToCompact.size() == 1 && !filesToCompact.get(0).isReference())) {
if (LOG.isDebugEnabled()) {
- LOG.debug("nothing to compact for " + this.regionName + "/" +
- this.familyName + " (" + this.storeName + ")");
+ LOG.debug("nothing to compact for " + this.storeName);
}
return false;
}
- if (!fs.exists(curCompactStore) && !fs.mkdirs(curCompactStore)) {
- LOG.warn("Mkdir on " + curCompactStore.toString() + " for " +
- this.regionName + "/" +
- this.familyName + " failed");
+ if (!fs.exists(compactionDir) && !fs.mkdirs(compactionDir)) {
+ LOG.warn("Mkdir on " + compactionDir.toString() + " for " +
+ this.storeName + " failed");
return false;
}
- // Step through them, writing to the brand-new TreeMap
- HStoreFile compactedOutputFile = new HStoreFile(conf, this.compactionDir,
- encodedRegionName, familyName, -1);
+ // Step through them, writing to the brand-new MapFile
+ HStoreFile compactedOutputFile = new HStoreFile(conf, fs,
+ this.compactionDir, info.getEncodedName(), family.getFamilyName(),
+ -1L, null);
MapFile.Writer compactedOut = compactedOutputFile.getWriter(this.fs,
this.compression, this.bloomFilter);
try {
@@ -1071,24 +1143,8 @@
long maxId = getMaxSequenceId(filesToCompact);
compactedOutputFile.writeInfo(fs, maxId);
- // Write out a list of data files that we're replacing
- Path filesToReplace = new Path(curCompactStore, COMPACTION_TO_REPLACE);
- FSDataOutputStream out = fs.create(filesToReplace);
- try {
- out.writeInt(filesToCompact.size());
- for (HStoreFile hsf : filesToCompact) {
- hsf.write(out);
- }
- } finally {
- out.close();
- }
-
- // Indicate that we're done.
- Path doneFile = new Path(curCompactStore, COMPACTION_DONE);
- fs.create(doneFile).close();
-
// Move the compaction into place.
- completeCompaction(curCompactStore);
+ completeCompaction(filesToCompact, compactedOutputFile);
return true;
}
}
@@ -1121,7 +1177,7 @@
// culprit.
LOG.warn("Failed with " + e.toString() + ": " + hsf.toString() +
(hsf.isReference() ? " " + hsf.getReference().toString() : "") +
- " for " + this.regionName + "/" + this.familyName);
+ " for " + this.storeName);
closeCompactionReaders(rdrs);
throw e;
}
@@ -1221,8 +1277,7 @@
try {
rdrs[i].close();
} catch (IOException e) {
- LOG.warn("Exception closing reader for " + this.regionName + "/" +
- this.familyName, e);
+ LOG.warn("Exception closing reader for " + this.storeName, e);
}
}
}
@@ -1348,11 +1403,15 @@
* 9) Allow new scanners to proceed.
* </pre>
*
- * @param curCompactStore Compaction to complete.
+ * @param compactedFiles list of files that were compacted
+ * @param compactedFile HStoreFile that is the result of the compaction
+ * @throws IOException
*/
- private void completeCompaction(final Path curCompactStore)
- throws IOException {
+ private void completeCompaction(List<HStoreFile> compactedFiles,
+ HStoreFile compactedFile) throws IOException {
+
// 1. Wait for active scanners to exit
+
newScannerLock.writeLock().lock(); // prevent new scanners
try {
synchronized (activeScanners) {
@@ -1369,54 +1428,27 @@
}
try {
- Path doneFile = new Path(curCompactStore, COMPACTION_DONE);
- if (!fs.exists(doneFile)) {
- // The last execution didn't finish the compaction, so there's nothing
- // we can do. We'll just have to redo it. Abandon it and return.
- LOG.warn("Redo failed compaction (missing 'done' file) for " +
- this.regionName + "/" + this.familyName);
- return;
- }
-
- // 3. Load in the files to be deleted.
- Vector<HStoreFile> toCompactFiles = new Vector<HStoreFile>();
- Path filesToReplace = new Path(curCompactStore, COMPACTION_TO_REPLACE);
- DataInputStream in = new DataInputStream(fs.open(filesToReplace));
- try {
- int numfiles = in.readInt();
- for(int i = 0; i < numfiles; i++) {
- HStoreFile hsf = new HStoreFile(conf);
- hsf.readFields(in);
- toCompactFiles.add(hsf);
- }
- } finally {
- in.close();
- }
-
- // 4. Moving the new MapFile into place.
- HStoreFile compactedFile = new HStoreFile(conf, this.compactionDir,
- encodedRegionName, familyName, -1);
- // obtainNewHStoreFile does its best to generate a filename that does not
- // currently exist.
- HStoreFile finalCompactedFile = HStoreFile.obtainNewHStoreFile(conf, dir,
- encodedRegionName, familyName, fs);
+ // 3. Moving the new MapFile into place.
+
+ HStoreFile finalCompactedFile = new HStoreFile(conf, fs, basedir,
+ info.getEncodedName(), family.getFamilyName(), -1, null);
if(LOG.isDebugEnabled()) {
LOG.debug("moving " + compactedFile.toString() + " in " +
this.compactionDir.toString() + " to " +
- finalCompactedFile.toString() + " in " + dir.toString() +
- " for " + this.regionName + "/" + this.familyName);
+ finalCompactedFile.toString() + " in " + basedir.toString() +
+ " for " + this.storeName);
}
if (!compactedFile.rename(this.fs, finalCompactedFile)) {
LOG.error("Failed move of compacted file " +
- finalCompactedFile.toString() + " for " + this.regionName + "/" +
- this.familyName);
+ finalCompactedFile.toString() + " for " + this.storeName);
return;
}
- // 5. and 6. Unload all the replaced MapFiles, close and delete.
- Vector<Long> toDelete = new Vector<Long>(toCompactFiles.size());
+ // 4. and 5. Unload all the replaced MapFiles, close and delete.
+
+ List<Long> toDelete = new ArrayList<Long>();
for (Map.Entry<Long, HStoreFile> e: this.storefiles.entrySet()) {
- if (!toCompactFiles.contains(e.getValue())) {
+ if (!compactedFiles.contains(e.getValue())) {
continue;
}
Long key = e.getKey();
@@ -1433,24 +1465,24 @@
hsf.delete();
}
- // 7. Loading the new TreeMap.
+ // 6. Loading the new TreeMap.
Long orderVal = Long.valueOf(finalCompactedFile.loadInfo(fs));
this.readers.put(orderVal,
finalCompactedFile.getReader(this.fs, this.bloomFilter));
this.storefiles.put(orderVal, finalCompactedFile);
} catch (IOException e) {
- LOG.error("Failed replacing compacted files for " +
- this.regionName + "/" + this.familyName + ". Compacted file is " +
- finalCompactedFile.toString() + ". Files replaced are " +
- toCompactFiles.toString() +
+ e = RemoteExceptionHandler.checkIOException(e);
+ LOG.error("Failed replacing compacted files for " + this.storeName +
+ ". Compacted file is " + finalCompactedFile.toString() +
+ ". Files replaced are " + compactedFiles.toString() +
" some of which may have been already removed", e);
}
} finally {
- // 8. Releasing the write-lock
+ // 7. Releasing the write-lock
this.lock.writeLock().unlock();
}
} finally {
- // 9. Allow new scanners to proceed.
+ // 8. Allow new scanners to proceed.
newScannerLock.writeLock().unlock();
}
}
@@ -1505,18 +1537,7 @@
}
}
- /*
- * @return Path to the compaction directory for this column family.
- * Compaction dir is a subdirectory of the region. Needs to have the
- * same regiondir/storefamily path prefix; HStoreFile constructor presumes
- * it (TODO: Fix).
- */
- private Path getCompactionDir() {
- return HStoreFile.getHStoreDir(this.compactionDir,
- this.encodedRegionName, this.familyName);
- }
-
- private MapFile.Reader [] getReaders() {
+ MapFile.Reader [] getReaders() {
return this.readers.values().
toArray(new MapFile.Reader[this.readers.size()]);
}
@@ -1796,8 +1817,7 @@
midKey.set(((HStoreKey)midkey).getRow());
}
} catch(IOException e) {
- LOG.warn("Failed getting store size for " + this.regionName + "/" +
- this.familyName, e);
+ LOG.warn("Failed getting store size for " + this.storeName, e);
} finally {
this.lock.readLock().unlock();
}
@@ -1835,6 +1855,38 @@
return this.storeName;
}
+ /*
+ * @see writeSplitInfo(Path p, HStoreFile hsf, FileSystem fs)
+ */
+ static HStoreFile.Reference readSplitInfo(final Path p, final FileSystem fs)
+ throws IOException {
+ FSDataInputStream in = fs.open(p);
+ try {
+ HStoreFile.Reference r = new HStoreFile.Reference();
+ r.readFields(in);
+ return r;
+ } finally {
+ in.close();
+ }
+ }
+
+ /**
+ * @param p Path to check.
+ * @return True if the path has format of a HStoreFile reference.
+ */
+ public static boolean isReference(final Path p) {
+ return isReference(p, REF_NAME_PARSER.matcher(p.getName()));
+ }
+
+ private static boolean isReference(final Path p, final Matcher m) {
+ if (m == null || !m.matches()) {
+ LOG.warn("Failed match of store file name " + p.toString());
+ throw new RuntimeException("Failed match of store file name " +
+ p.toString());
+ }
+ return m.groupCount() > 1 && m.group(2) != null;
+ }
+
/**
* A scanner that iterates through the HStore files
*/
@@ -1939,7 +1991,7 @@
try {
readers[i].close();
} catch(IOException e) {
- LOG.error(regionName + "/" + familyName + " closing sub-scanner", e);
+ LOG.error(storeName + " closing sub-scanner", e);
}
}
@@ -1959,7 +2011,7 @@
try {
readers[i].close();
} catch(IOException e) {
- LOG.error(regionName + "/" + familyName + " closing scanner", e);
+ LOG.error(storeName + " closing scanner", e);
}
}
}
@@ -2195,8 +2247,7 @@
try {
scanners[i].close();
} catch (IOException e) {
- LOG.warn(regionName + "/" + familyName + " failed closing scanner "
- + i, e);
+ LOG.warn(storeName + " failed closing scanner " + i, e);
}
} finally {
scanners[i] = null;
@@ -2217,7 +2268,7 @@
synchronized (activeScanners) {
int numberOfScanners = activeScanners.decrementAndGet();
if (numberOfScanners < 0) {
- LOG.error(regionName + "/" + familyName +
+ LOG.error(storeName +
" number of active scanners less than zero: " +
numberOfScanners + " resetting to zero");
activeScanners.set(0);