Author: ecn
Date: Tue Apr 24 17:07:30 2012
New Revision: 1329860
URL: http://svn.apache.org/viewvc?rev=1329860&view=rev
Log:
ACCUMULO-528 add support for merge on the !METADATA table
Added:
accumulo/trunk/test/system/auto/simple/mergeMetadata.py (with props)
Modified:
accumulo/trunk/core/src/main/java/org/apache/accumulo/core/Constants.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/Master.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/state/MergeStats.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/state/RootTabletStateStore.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/tableOps/TableRangeOp.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/util/TabletOperations.java
Modified:
accumulo/trunk/core/src/main/java/org/apache/accumulo/core/Constants.java
URL:
http://svn.apache.org/viewvc/accumulo/trunk/core/src/main/java/org/apache/accumulo/core/Constants.java?rev=1329860&r1=1329859&r2=1329860&view=diff
==============================================================================
--- accumulo/trunk/core/src/main/java/org/apache/accumulo/core/Constants.java
(original)
+++ accumulo/trunk/core/src/main/java/org/apache/accumulo/core/Constants.java
Tue Apr 24 17:07:30 2012
@@ -126,6 +126,8 @@ public class Constants {
public static final KeyExtent ROOT_TABLET_EXTENT = new KeyExtent(new
Text(METADATA_TABLE_ID), KeyExtent.getMetadataEntry(new
Text(METADATA_TABLE_ID), null),
null);
+ public static final Range METADATA_ROOT_TABLET_KEYSPACE = new
Range(ROOT_TABLET_EXTENT.getMetadataEntry(), false,
KeyExtent.getMetadataEntry(new Text(
+ METADATA_TABLE_ID), null), true);
public static final String VALUE_ENCODING = "UTF-8";
@@ -156,6 +158,7 @@ public class Constants {
public static final String OLD_PACKAGE_NAME = "cloudbase";
public static final String VALID_TABLE_NAME_REGEX = "^\\w+$";
public static final String MAPFILE_EXTENSION = "map";
+ public static final String GENERATED_TABLET_DIRECTORY_PREFIX = "t-";
public static String getBaseDir(AccumuloConfiguration conf) {
return conf.get(Property.INSTANCE_DFS_DIR);
Modified:
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/Master.java
URL:
http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/Master.java?rev=1329860&r1=1329859&r2=1329860&view=diff
==============================================================================
---
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/Master.java
(original)
+++
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/Master.java
Tue Apr 24 17:07:30 2012
@@ -982,12 +982,18 @@ public class Master implements LiveTServ
String tableName = ByteBufferUtil.toString(arguments.get(0));
Text startRow = ByteBufferUtil.toText(arguments.get(1));
Text endRow = ByteBufferUtil.toText(arguments.get(2));
-
final String tableId = checkTableId(tableName, TableOperation.MERGE);
- checkNotMetadataTable(tableName, TableOperation.MERGE);
+ if (tableName.equals(Constants.METADATA_TABLE_NAME)) {
+ if (startRow.compareTo(new Text("0")) < 0) {
+ startRow = new Text("0");
+ if (endRow.getLength() != 0 && endRow.compareTo(startRow) < 0)
+ throw new ThriftTableOperationException(null, tableName,
TableOperation.MERGE, TableOperationExceptionType.OTHER,
+ "end-row specification is in the root tablet, which cannot
be merged or split");
+ }
+ }
+ log.debug("Creating merge op: " + tableId + " " + startRow + " " +
endRow);
verify(c, tableId, TableOperation.MERGE,
check(c, SystemPermission.SYSTEM) || check(c,
SystemPermission.ALTER_TABLE) || check(c, tableId,
TablePermission.ALTER_TABLE));
-
fate.seedTransaction(opid, new TraceRepo<Master>(new
TableRangeOp(MergeInfo.Operation.MERGE, tableId, startRow, endRow)),
autoCleanup);
break;
}
@@ -1625,6 +1631,9 @@ public class Master implements LiveTServ
start = new Text();
}
Range scanRange = new
Range(KeyExtent.getMetadataEntry(range.getTableId(), start), false, stopRow,
false);
+ if (range.isMeta())
+ scanRange = scanRange.clip(Constants.METADATA_ROOT_TABLET_KEYSPACE);
+
BatchWriter bw = null;
try {
long fileCount = 0;
@@ -1651,14 +1660,18 @@ public class Master implements LiveTServ
} else if (Constants.METADATA_TIME_COLUMN.hasColumns(key)) {
maxLogicalTime = TabletTime.maxMetadataTime(maxLogicalTime,
value.toString());
} else if (Constants.METADATA_DIRECTORY_COLUMN.hasColumns(key)) {
-
bw.addMutation(MetadataTable.createDeleteMutation(range.getTableId().toString(),
entry.getValue().toString()));
+ if (!range.isMeta())
+
bw.addMutation(MetadataTable.createDeleteMutation(range.getTableId().toString(),
entry.getValue().toString()));
}
}
// read the logical time from the last tablet in the merge range, it
is not included in
// the loop above
scanner = conn.createScanner(Constants.METADATA_TABLE_NAME,
Constants.NO_AUTHS);
- scanner.setRange(new Range(stopRow));
+ Range last = new Range(stopRow);
+ if (range.isMeta())
+ last = last.clip(Constants.METADATA_ROOT_TABLET_KEYSPACE);
+ scanner.setRange(last);
ColumnFQ.fetch(scanner, Constants.METADATA_TIME_COLUMN);
for (Entry<Key,Value> entry : scanner) {
if (Constants.METADATA_TIME_COLUMN.hasColumns(entry.getKey())) {
@@ -2317,5 +2330,9 @@ public class Master implements LiveTServ
public ServerConfiguration getConfiguration() {
return serverConfig;
}
+
+ public FileSystem getFileSystem() {
+ return this.fs;
+ }
}
Modified:
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/state/MergeStats.java
URL:
http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/state/MergeStats.java?rev=1329860&r1=1329859&r2=1329860&view=diff
==============================================================================
---
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/state/MergeStats.java
(original)
+++
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/state/MergeStats.java
Tue Apr 24 17:07:30 2012
@@ -105,6 +105,11 @@ public class MergeStats {
log.info("failed to see any tablets for this range, ignoring " +
info.getRange());
return state;
}
+ if (total == 1 && info.getRange().isMeta()) {
+ // root tablet watcher trying to merge metadata tablets it won't even
scan
+ log.debug("ignoring merge of " + info.getRange());
+ return state;
+ }
if (state == MergeState.SPLITTING) {
log.info(hosted + " are hosted, total " + total);
if (!info.isDelete() && total == 1) {
@@ -178,6 +183,8 @@ public class MergeStats {
Text tableId = extent.getTableId();
Text first = KeyExtent.getMetadataEntry(tableId, start);
Range range = new Range(first, false, null, true);
+ if (extent.isMeta())
+ range = new Range(Constants.METADATA_ROOT_TABLET_KEYSPACE);
scanner.setRange(range);
KeyExtent prevExtent = null;
Modified:
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/state/RootTabletStateStore.java
URL:
http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/state/RootTabletStateStore.java?rev=1329860&r1=1329859&r2=1329860&view=diff
==============================================================================
---
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/state/RootTabletStateStore.java
(original)
+++
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/state/RootTabletStateStore.java
Tue Apr 24 17:07:30 2012
@@ -33,9 +33,7 @@ public class RootTabletStateStore extend
@Override
public Iterator<TabletLocationState> iterator() {
- Range range = new Range(Constants.ROOT_TABLET_EXTENT.getMetadataEntry(),
false, KeyExtent.getMetadataEntry(new Text(Constants.METADATA_TABLE_ID), null),
- true);
- return new MetaDataTableScanner(instance, auths, range, state);
+ return new MetaDataTableScanner(instance, auths,
Constants.METADATA_ROOT_TABLET_KEYSPACE, state);
}
@Override
Modified:
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/tableOps/TableRangeOp.java
URL:
http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/tableOps/TableRangeOp.java?rev=1329860&r1=1329859&r2=1329860&view=diff
==============================================================================
---
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/tableOps/TableRangeOp.java
(original)
+++
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/tableOps/TableRangeOp.java
Tue Apr 24 17:07:30 2012
@@ -16,16 +16,25 @@
*/
package org.apache.accumulo.server.master.tableOps;
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.impl.thrift.TableOperation;
import org.apache.accumulo.core.client.impl.thrift.TableOperationExceptionType;
import
org.apache.accumulo.core.client.impl.thrift.ThriftTableOperationException;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.data.KeyExtent;
import org.apache.accumulo.core.util.TextUtil;
import org.apache.accumulo.server.fate.Repo;
import org.apache.accumulo.server.master.Master;
import org.apache.accumulo.server.master.state.MergeInfo;
-import org.apache.accumulo.server.master.state.MergeState;
import org.apache.accumulo.server.master.state.MergeInfo.Operation;
+import org.apache.accumulo.server.master.state.MergeState;
+import org.apache.accumulo.server.security.SecurityConstants;
+import org.apache.accumulo.server.util.MetadataTable;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
/**
@@ -42,6 +51,31 @@ import org.apache.hadoop.io.Text;
* and prevent merges (writes) while they run. Merge operations will lock out
some operations while they run.
*/
+class MakeDeleteEntries extends MasterRepo {
+
+ private static final long serialVersionUID = 1L;
+ private static final long ARBITRARY_BUFFER_SIZE = 10000;
+ private static final long ARBITRARY_LATENCY = 1000;
+
+ @Override
+ public Repo<Master> call(long tid, Master master) throws Exception {
+ log.info("creating delete entries for merged metadata tablets");
+ Instance instance = master.getInstance();
+ Connector conn =
instance.getConnector(SecurityConstants.getSystemCredentials());
+ BatchWriter bw = conn.createBatchWriter(Constants.METADATA_TABLE_NAME,
ARBITRARY_BUFFER_SIZE, ARBITRARY_LATENCY, 1);
+ AccumuloConfiguration conf = instance.getConfiguration();
+ String tableDir = Constants.getMetadataTableDir(conf);
+ for (FileStatus fs : master.getFileSystem().listStatus(new
Path(tableDir))) {
+ // TODO: add the entries only if there are no !METADATA table references
+ if (fs.isDir() && fs.getPath().getName().matches("^" +
Constants.GENERATED_TABLET_DIRECTORY_PREFIX + ".*")) {
+
bw.addMutation(MetadataTable.createDeleteMutation(Constants.METADATA_TABLE_ID,
"/" + fs.getPath().getName()));
+ }
+ }
+ bw.close();
+ return null;
+ }
+}
+
class TableRangeOpWait extends MasterRepo {
private static final long serialVersionUID = 1L;
@@ -67,6 +101,13 @@ class TableRangeOpWait extends MasterRep
log.warn("removing merge information " + mergeInfo);
env.clearMergeState(tableIdText);
Utils.unreserveTable(tableId, tid, true);
+ // We can't add entries to the metadata table if it is offline for this
merge.
+ // If the delete entries for the metadata table were in the root tablet,
it would work just fine
+ // but all the delete entries go into the end of the metadata table. Work
around: add the
+ // delete entries after the merge completes.
+ if (mergeInfo.getOperation().equals(Operation.MERGE) &&
tableId.equals(Constants.METADATA_TABLE_ID)) {
+ return new MakeDeleteEntries();
+ }
return null;
}
Modified:
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/util/TabletOperations.java
URL:
http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/util/TabletOperations.java?rev=1329860&r1=1329859&r2=1329860&view=diff
==============================================================================
---
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/util/TabletOperations.java
(original)
+++
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/util/TabletOperations.java
Tue Apr 24 17:07:30 2012
@@ -45,7 +45,7 @@ public class TabletOperations {
return lowDirectory;
log.warn("Failed to create " + lowDirectoryPath + " for unknown
reason");
} else {
- lowDirectory = "/t-" + namer.getNextName();
+ lowDirectory = "/" + Constants.GENERATED_TABLET_DIRECTORY_PREFIX +
namer.getNextName();
Path lowDirectoryPath = new Path(tableDir + lowDirectory);
if (fs.exists(lowDirectoryPath))
throw new IllegalStateException("Dir exist when it should not " +
lowDirectoryPath);
Added: accumulo/trunk/test/system/auto/simple/mergeMetadata.py
URL:
http://svn.apache.org/viewvc/accumulo/trunk/test/system/auto/simple/mergeMetadata.py?rev=1329860&view=auto
==============================================================================
--- accumulo/trunk/test/system/auto/simple/mergeMetadata.py (added)
+++ accumulo/trunk/test/system/auto/simple/mergeMetadata.py Tue Apr 24 17:07:30
2012
@@ -0,0 +1,72 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+
+import logging
+import unittest
+
+from TestUtils import TestUtilsMixin
+
+log = logging.getLogger('test.auto')
+
+class MergeMeta(unittest.TestCase, TestUtilsMixin):
+ """Split and merge the !METADATA table"""
+
+ order = 30
+
+ settings = TestUtilsMixin.settings.copy()
+
+ def setUp(self):
+ TestUtilsMixin.setUp(self);
+
+ def runTest(self):
+ out, err, code = self.shell(self.masterHost(), '''
+addsplits -t !METADATA 1 2 3 4 5
+createtable a1
+createtable a2
+createtable a3
+createtable a4
+createtable a5
+merge -t !METADATA
+sleep 2
+scan -np -t !METADATA
+''')
+ assert code == 0
+ # look for delete entries for the abandoned directories
+ assert out.find('~del') >= 0
+
+class MergeMetaFail(unittest.TestCase, TestUtilsMixin):
+ """test a failed merge of the !METADATA table"""
+
+ order = 30
+
+ settings = TestUtilsMixin.settings.copy()
+
+ def setUp(self):
+ TestUtilsMixin.setUp(self);
+
+ def runTest(self):
+ out, err, code = self.shell(self.masterHost(), '''
+merge -t !METADATA -b ! -e !!
+''')
+ assert code != 0
+
+
+def suite():
+ result = unittest.TestSuite()
+ result.addTest(MergeMeta())
+ result.addTest(MergeMetaFail())
+ return result
Propchange: accumulo/trunk/test/system/auto/simple/mergeMetadata.py
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: accumulo/trunk/test/system/auto/simple/mergeMetadata.py
------------------------------------------------------------------------------
svn:executable = *
