You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by el...@apache.org on 2014/10/16 02:11:08 UTC
[1/4] git commit: ACCUMULO-3215 Ensure tablet directories are
absolute, not relative
Repository: accumulo
Updated Branches:
refs/heads/1.6 0348e3ea3 -> 6ad16a9c5
refs/heads/master a21d2d3cf -> f118b2211
ACCUMULO-3215 Ensure tablet directories are absolute, not relative
Move 'c-' into a constant, make the computed srv:dir in ImportTable
an absolute directory (choosing a volume as per policy for each tablet),
and add a test to make sure the path is constructed correctly.
ImportExportIT was extended to check that srv:dir, in addition to file:,
is an absolute path.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/6ad16a9c
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/6ad16a9c
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/6ad16a9c
Branch: refs/heads/1.6
Commit: 6ad16a9c50f48a98def23e5720218659a410dd08
Parents: 0348e3e
Author: Josh Elser <el...@apache.org>
Authored: Wed Oct 15 15:14:25 2014 -0400
Committer: Josh Elser <el...@apache.org>
Committed: Wed Oct 15 19:31:29 2014 -0400
----------------------------------------------------------------------
.../org/apache/accumulo/core/Constants.java | 5 +-
.../accumulo/server/util/MetadataTableUtil.java | 2 +-
.../accumulo/gc/SimpleGarbageCollector.java | 2 +-
server/master/pom.xml | 5 ++
.../accumulo/master/tableOps/ImportTable.java | 34 ++++++++++--
.../master/tableOps/ImportTableTest.java | 54 ++++++++++++++++++++
.../org/apache/accumulo/tserver/Tablet.java | 46 ++++++++---------
.../apache/accumulo/test/ImportExportIT.java | 35 ++++++++++---
8 files changed, 147 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6ad16a9c/core/src/main/java/org/apache/accumulo/core/Constants.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/Constants.java b/core/src/main/java/org/apache/accumulo/core/Constants.java
index 15cccf4..2a570f4 100644
--- a/core/src/main/java/org/apache/accumulo/core/Constants.java
+++ b/core/src/main/java/org/apache/accumulo/core/Constants.java
@@ -56,7 +56,7 @@ public class Constants {
public static final String ZMONITOR_LOCK = ZMONITOR + "/lock";
public static final String ZMONITOR_HTTP_ADDR = ZMONITOR + "/http_addr";
public static final String ZMONITOR_LOG4J_ADDR = ZMONITOR + "/log4j_addr";
-
+
public static final String ZCONFIG = "/config";
public static final String ZTSERVERS = "/tservers";
@@ -88,6 +88,9 @@ public class Constants {
public static final String BULK_PREFIX = "b-";
+ public static final String CLONE_PREFIX = "c-";
+ public static final byte[] CLONE_PREFIX_BYTES = CLONE_PREFIX.getBytes(UTF8);
+
// this affects the table client caching of metadata
public static final int SCAN_BATCH_SIZE = 1000;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6ad16a9c/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java b/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
index f978439..f908d25 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
@@ -881,7 +881,7 @@ public class MetadataTableUtil {
Mutation m = new Mutation(k.getRow());
m.putDelete(k.getColumnFamily(), k.getColumnQualifier());
String dir = volumeManager.choose(ServerConstants.getTablesDirs()) + "/" + tableId
- + new String(FastFormat.toZeroPaddedString(dirCount++, 8, 16, "/c-".getBytes(Constants.UTF8)));
+ + new String(FastFormat.toZeroPaddedString(dirCount++, 8, 16, Constants.CLONE_PREFIX_BYTES));
TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(dir.getBytes(Constants.UTF8)));
bw.addMutation(m);
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6ad16a9c/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
----------------------------------------------------------------------
diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java b/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
index 4617299..d04303b 100644
--- a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
@@ -421,7 +421,7 @@ public class SimpleGarbageCollector implements Iface {
TableState tableState = TableManager.getInstance().getTableState(tableId);
if (tableState != null && tableState != TableState.DELETING) {
// clone directories don't always exist
- if (!tabletDir.startsWith("c-"))
+ if (!tabletDir.startsWith(Constants.CLONE_PREFIX))
log.debug("File doesn't exist: " + fullPath);
}
} else {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6ad16a9c/server/master/pom.xml
----------------------------------------------------------------------
diff --git a/server/master/pom.xml b/server/master/pom.xml
index bfd6b68..0268680 100644
--- a/server/master/pom.xml
+++ b/server/master/pom.xml
@@ -77,6 +77,11 @@
<scope>test</scope>
</dependency>
<dependency>
+ <groupId>org.easymock</groupId>
+ <artifactId>easymock</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<scope>test</scope>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6ad16a9c/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java b/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
index e57c689..8de6bce 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
@@ -222,7 +222,9 @@ class PopulateMetadataTable extends MasterRepo {
log.info("importDir is " + tableInfo.importDir);
// This is a directory already prefixed with proper volume information e.g. hdfs://localhost:8020/path/to/accumulo/tables/...
- String bulkDir = tableInfo.importDir;
+ final String bulkDir = tableInfo.importDir;
+
+ final String[] tableDirs = ServerConstants.getTablesDirs();
ZipEntry zipEntry;
while ((zipEntry = zis.getNextEntry()) != null) {
@@ -260,15 +262,28 @@ class PopulateMetadataTable extends MasterRepo {
}
if (m == null) {
+ // Make a unique directory inside the table's dir. Cannot import multiple tables into one table, so don't need to use unique allocator
+ String tabletDir = new String(FastFormat.toZeroPaddedString(dirCount++, 8, 16, Constants.CLONE_PREFIX_BYTES), Constants.UTF8);
+
+ // Build up a full hdfs://localhost:8020/accumulo/tables/$id/c-XXXXXXX
+ String absolutePath = getClonedTabletDir(master, tableDirs, tabletDir);
+
m = new Mutation(metadataRow);
- TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(FastFormat.toZeroPaddedString(dirCount++, 8, 16, "/c-".getBytes(Constants.UTF8))));
+ TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(absolutePath.getBytes(Constants.UTF8)));
currentRow = metadataRow;
}
if (!currentRow.equals(metadataRow)) {
mbw.addMutation(m);
+
+ // Make a unique directory inside the table's dir. Cannot import multiple tables into one table, so don't need to use unique allocator
+ String tabletDir = new String(FastFormat.toZeroPaddedString(dirCount++, 8, 16, Constants.CLONE_PREFIX_BYTES), Constants.UTF8);
+
+ // Build up a full hdfs://localhost:8020/accumulo/tables/$id/c-XXXXXXX
+ String absolutePath = getClonedTabletDir(master, tableDirs, tabletDir);
+
m = new Mutation(metadataRow);
- TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(FastFormat.toZeroPaddedString(dirCount++, 8, 16, "/c-".getBytes(Constants.UTF8))));
+ TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(absolutePath.getBytes(Constants.UTF8)));
}
m.put(key.getColumnFamily(), cq, val);
@@ -303,6 +318,19 @@ class PopulateMetadataTable extends MasterRepo {
}
}
+ /**
+ * Given options for tables (across multiple volumes), construct an absolute path using the unique name within the chosen volume
+ *
+ * @return An absolute, unique path for the imported table
+ */
+ protected String getClonedTabletDir(Master master, String[] tableDirs, String tabletDir) {
+ // We can try to spread out the tablet dirs across all volumes
+ String tableDir = master.getFileSystem().choose(tableDirs);
+
+ // Build up a full hdfs://localhost:8020/accumulo/tables/$id/c-XXXXXXX
+ return tableDir + "/" + tableInfo.tableId + "/" + tabletDir;
+ }
+
@Override
public void undo(long tid, Master environment) throws Exception {
MetadataTableUtil.deleteTable(tableInfo.tableId, false, SystemCredentials.get(), environment.getMasterLock());
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6ad16a9c/server/master/src/test/java/org/apache/accumulo/master/tableOps/ImportTableTest.java
----------------------------------------------------------------------
diff --git a/server/master/src/test/java/org/apache/accumulo/master/tableOps/ImportTableTest.java b/server/master/src/test/java/org/apache/accumulo/master/tableOps/ImportTableTest.java
new file mode 100644
index 0000000..31f6bde
--- /dev/null
+++ b/server/master/src/test/java/org/apache/accumulo/master/tableOps/ImportTableTest.java
@@ -0,0 +1,54 @@
+/*
+ * 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.
+ */
+package org.apache.accumulo.master.tableOps;
+
+import org.apache.accumulo.master.Master;
+import org.apache.accumulo.server.fs.VolumeManager;
+import org.easymock.EasyMock;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ *
+ */
+public class ImportTableTest {
+
+ @Test
+ public void testTabletDir() {
+ Master master = EasyMock.createMock(Master.class);
+ VolumeManager volumeManager = EasyMock.createMock(VolumeManager.class);
+ ImportedTableInfo iti = new ImportedTableInfo();
+ iti.tableId = "5";
+
+ // Different volumes with different paths
+ String[] tableDirs = new String[] {"hdfs://nn1:8020/apps/accumulo1/tables", "hdfs://nn2:8020/applications/accumulo/tables"};
+ // This needs to be unique WRT the importtable command
+ String tabletDir = "/c-00000001";
+
+ EasyMock.expect(master.getFileSystem()).andReturn(volumeManager);
+ // Choose the 2nd element
+ EasyMock.expect(volumeManager.choose(tableDirs)).andReturn(tableDirs[1]);
+
+ EasyMock.replay(master, volumeManager);
+
+ PopulateMetadataTable pmt = new PopulateMetadataTable(iti);
+ Assert.assertEquals(tableDirs[1] + "/" + iti.tableId + "/" + tabletDir, pmt.getClonedTabletDir(master, tableDirs, tabletDir));
+
+ EasyMock.verify(master, volumeManager);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6ad16a9c/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java
index 53469d7..de930aa 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java
@@ -157,16 +157,16 @@ import com.google.common.annotations.VisibleForTesting;
* We need to be able to have the master tell a tabletServer to
* close this file, and the tablet server to handle all pending client reads
* before closing
- *
+ *
*/
/**
- *
+ *
* this class just provides an interface to read from a MapFile mostly takes care of reporting start and end keys
- *
+ *
* need this because a single row extent can have multiple columns this manages all the columns (each handled by a store) for a single row-extent
- *
- *
+ *
+ *
*/
public class Tablet {
@@ -422,7 +422,7 @@ public class Tablet {
private long lastFlushID = -1;
private long lastCompactID = -1;
-
+
private static class CompactionWaitInfo {
long flushID = -1;
long compactionID = -1;
@@ -523,7 +523,7 @@ public class Tablet {
}
if (files == null) {
- if (tabletDir.getName().startsWith("c-"))
+ if (tabletDir.getName().startsWith(Constants.CLONE_PREFIX))
log.debug("Tablet " + extent + " had no dir, creating " + tabletDir); // its a clone dir...
else
log.warn("Tablet " + extent + " had no dir, creating " + tabletDir);
@@ -833,7 +833,7 @@ public class Tablet {
log.warn("Target map file already exist " + newDatafile);
fs.deleteRecursively(newDatafile.path());
}
-
+
rename(fs, tmpDatafile.path(), newDatafile.path());
}
break;
@@ -966,7 +966,7 @@ public class Tablet {
// rename before putting in metadata table, so files in metadata table should
// always exist
rename(fs, tmpDatafile.path(), newDatafile.path());
-
+
if (dfv.getNumEntries() == 0) {
fs.deleteRecursively(newDatafile.path());
}
@@ -1081,7 +1081,7 @@ public class Tablet {
this.tabletDirectory = tabletDirectory;
this.logId = logId;
this.location = location;
- this.datafileManager = datafileManager;
+ this.datafileManager = datafileManager;
}
private Tablet(TabletServer tabletServer, Text location, KeyExtent extent, TabletResourceManager trm, Configuration conf,
@@ -1356,9 +1356,9 @@ public class Tablet {
}
});
-
+
acuTableConf.getNamespaceConfiguration().addObserver(configObserver);
-
+
// Force a load of any per-table properties
configObserver.propertiesChanged();
@@ -1706,7 +1706,7 @@ public class Tablet {
/**
* Determine if a JVM shutdown is in progress.
- *
+ *
*/
private boolean shutdownInProgress() {
try {
@@ -2044,7 +2044,7 @@ public class Tablet {
public void setInterruptFlag(AtomicBoolean flag) {
throw new UnsupportedOperationException();
}
-
+
}
private DataFileValue minorCompact(Configuration conf, VolumeManager fs, InMemoryMap memTable, FileRef tmpDatafile, FileRef newDatafile, FileRef mergeFile,
@@ -2768,7 +2768,7 @@ public class Tablet {
/**
* Returns a Path object representing the tablet's location on the DFS.
- *
+ *
* @return location
*/
public Path getLocation() {
@@ -2863,7 +2863,7 @@ public class Tablet {
/**
* Returns true if a major compaction should be performed on the tablet.
- *
+ *
*/
public boolean needsMajorCompaction(MajorCompactionReason reason) {
if (majorCompactionInProgress)
@@ -2875,7 +2875,7 @@ public class Tablet {
/**
* Returns an int representing the total block size of the mapfiles served by this tablet.
- *
+ *
* @return size
*/
// this is the size of just the mapfiles
@@ -3046,7 +3046,7 @@ public class Tablet {
/**
* Returns true if this tablet needs to be split
- *
+ *
*/
public synchronized boolean needsSplit() {
boolean ret;
@@ -3304,7 +3304,7 @@ public class Tablet {
return smallestFiles;
}
-
+
static void rename(VolumeManager fs, Path src, Path dst) throws IOException {
if (!fs.rename(src, dst)) {
throw new IOException("Rename " + src + " to " + dst + " returned false ");
@@ -3373,7 +3373,7 @@ public class Tablet {
/**
* Returns a KeyExtent object representing this tablet's key range.
- *
+ *
* @return extent
*/
public KeyExtent getExtent() {
@@ -3433,12 +3433,12 @@ public class Tablet {
/**
* operations are disallowed while we split which is ok since splitting is fast
- *
+ *
* a minor compaction should have taken place before calling this so there should be relatively little left to compact
- *
+ *
* we just need to make sure major compactions aren't occurring if we have the major compactor thread decide who needs splitting we can avoid synchronization
* issues with major compactions
- *
+ *
*/
static class SplitInfo {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6ad16a9c/test/src/test/java/org/apache/accumulo/test/ImportExportIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/ImportExportIT.java b/test/src/test/java/org/apache/accumulo/test/ImportExportIT.java
index c7e0f0e..a48ed9d 100644
--- a/test/src/test/java/org/apache/accumulo/test/ImportExportIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/ImportExportIT.java
@@ -42,7 +42,6 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.collect.Iterables;
import com.google.common.io.Files;
/**
@@ -57,8 +56,13 @@ public class ImportExportIT extends SimpleMacIT {
private static final Logger log = LoggerFactory.getLogger(ImportExportIT.class);
+ @Override
+ protected int defaultTimeoutSeconds() {
+ return 60;
+ }
+
@Test
- public void test() throws Exception {
+ public void testExportImportThenScan() throws Exception {
Connector conn = getConnector();
String[] tableNames = getUniqueNames(2);
@@ -128,19 +132,34 @@ public class ImportExportIT extends SimpleMacIT {
Scanner s = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
s.setRange(MetadataSchema.TabletsSection.getRange(tableId));
s.fetchColumnFamily(MetadataSchema.TabletsSection.DataFileColumnFamily.NAME);
+ MetadataSchema.TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.fetch(s);
// Should find a single entry
- Entry<Key,Value> fileEntry = Iterables.getOnlyElement(s);
-
- // The file should be an absolute URI (file:///...), not a relative path (/b-000.../I000001.rf)
- String fileUri = fileEntry.getKey().getColumnQualifier().toString();
- Assert.assertFalse("Imported files should have absolute URIs, not relative: " + fileUri, looksLikeRelativePath(fileUri));
+ for (Entry<Key,Value> fileEntry : s) {
+ Key k = fileEntry.getKey();
+ String value = fileEntry.getValue().toString();
+ if (k.getColumnFamily().equals(MetadataSchema.TabletsSection.DataFileColumnFamily.NAME)) {
+ // The file should be an absolute URI (file:///...), not a relative path (/b-000.../I000001.rf)
+ String fileUri = k.getColumnQualifier().toString();
+ Assert.assertFalse("Imported files should have absolute URIs, not relative: " + fileUri, looksLikeRelativePath(fileUri));
+ } else if (k.getColumnFamily().equals(MetadataSchema.TabletsSection.ServerColumnFamily.NAME)) {
+ Assert.assertFalse("Server directory should have absolute URI, not relative: " + value, looksLikeRelativePath(value));
+ } else {
+ Assert.fail("Got expected pair: " + k + "=" + fileEntry.getValue());
+ }
+ }
// Online the original table before we verify equivalence
conn.tableOperations().online(srcTable, true);
+ verifyTableEquality(conn, srcTable, destTable);
+ }
+
+ private void verifyTableEquality(Connector conn, String srcTable, String destTable) throws Exception {
Iterator<Entry<Key,Value>> src = conn.createScanner(srcTable, Authorizations.EMPTY).iterator(), dest = conn.createScanner(destTable, Authorizations.EMPTY)
.iterator();
+ Assert.assertTrue("Could not read any data from source table", src.hasNext());
+ Assert.assertTrue("Could not read any data from destination table", dest.hasNext());
while (src.hasNext() && dest.hasNext()) {
Entry<Key,Value> orig = src.next(), copy = dest.next();
Assert.assertEquals(orig.getKey(), copy.getKey());
@@ -155,6 +174,8 @@ public class ImportExportIT extends SimpleMacIT {
if ('/' == uri.charAt(10)) {
return true;
}
+ } else if (uri.startsWith("/" + Constants.CLONE_PREFIX)) {
+ return true;
}
return false;
[4/4] git commit: Merge branch '1.6'
Posted by el...@apache.org.
Merge branch '1.6'
Conflicts:
server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
server/master/pom.xml
server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/f118b221
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/f118b221
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/f118b221
Branch: refs/heads/master
Commit: f118b22116215aa1f7d816b31f56aafa1656b643
Parents: a21d2d3 6ad16a9
Author: Josh Elser <el...@apache.org>
Authored: Wed Oct 15 20:10:16 2014 -0400
Committer: Josh Elser <el...@apache.org>
Committed: Wed Oct 15 20:10:16 2014 -0400
----------------------------------------------------------------------
.../org/apache/accumulo/core/Constants.java | 4 ++
.../accumulo/server/util/MetadataTableUtil.java | 3 +-
.../accumulo/gc/SimpleGarbageCollector.java | 2 +-
.../accumulo/master/tableOps/ImportTable.java | 34 ++++++++++--
.../master/tableOps/ImportTableTest.java | 54 ++++++++++++++++++++
.../apache/accumulo/tserver/tablet/Tablet.java | 2 +-
.../apache/accumulo/test/ImportExportIT.java | 35 ++++++++++---
7 files changed, 121 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f118b221/core/src/main/java/org/apache/accumulo/core/Constants.java
----------------------------------------------------------------------
diff --cc core/src/main/java/org/apache/accumulo/core/Constants.java
index 76b86d4,2a570f4..e7162c2
--- a/core/src/main/java/org/apache/accumulo/core/Constants.java
+++ b/core/src/main/java/org/apache/accumulo/core/Constants.java
@@@ -16,6 -16,7 +16,7 @@@
*/
package org.apache.accumulo.core;
-import java.nio.charset.Charset;
++import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@@ -88,6 -88,9 +89,9 @@@ public class Constants
public static final String BULK_PREFIX = "b-";
+ public static final String CLONE_PREFIX = "c-";
- public static final byte[] CLONE_PREFIX_BYTES = CLONE_PREFIX.getBytes(UTF8);
++ public static final byte[] CLONE_PREFIX_BYTES = CLONE_PREFIX.getBytes(StandardCharsets.UTF_8);
+
// this affects the table client caching of metadata
public static final int SCAN_BATCH_SIZE = 1000;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f118b221/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
index 8d14cad,f908d25..5f34279
--- a/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
@@@ -881,8 -881,8 +882,8 @@@ public class MetadataTableUtil
Mutation m = new Mutation(k.getRow());
m.putDelete(k.getColumnFamily(), k.getColumnQualifier());
String dir = volumeManager.choose(ServerConstants.getTablesDirs()) + "/" + tableId
- + new String(FastFormat.toZeroPaddedString(dirCount++, 8, 16, "/c-".getBytes(StandardCharsets.UTF_8)));
+ + new String(FastFormat.toZeroPaddedString(dirCount++, 8, 16, Constants.CLONE_PREFIX_BYTES));
- TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(dir.getBytes(Constants.UTF8)));
+ TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(dir.getBytes(StandardCharsets.UTF_8)));
bw.addMutation(m);
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f118b221/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f118b221/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
----------------------------------------------------------------------
diff --cc server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
index 1129ede,8de6bce..44ca898
--- a/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
@@@ -261,15 -262,28 +263,28 @@@ class PopulateMetadataTable extends Mas
}
if (m == null) {
+ // Make a unique directory inside the table's dir. Cannot import multiple tables into one table, so don't need to use unique allocator
- String tabletDir = new String(FastFormat.toZeroPaddedString(dirCount++, 8, 16, Constants.CLONE_PREFIX_BYTES), Constants.UTF8);
++ String tabletDir = new String(FastFormat.toZeroPaddedString(dirCount++, 8, 16, Constants.CLONE_PREFIX_BYTES), StandardCharsets.UTF_8);
+
+ // Build up a full hdfs://localhost:8020/accumulo/tables/$id/c-XXXXXXX
+ String absolutePath = getClonedTabletDir(master, tableDirs, tabletDir);
+
m = new Mutation(metadataRow);
- TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(FastFormat.toZeroPaddedString(dirCount++, 8, 16, "/c-".getBytes(StandardCharsets.UTF_8))));
- TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(absolutePath.getBytes(Constants.UTF8)));
++ TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(absolutePath.getBytes(StandardCharsets.UTF_8)));
currentRow = metadataRow;
}
if (!currentRow.equals(metadataRow)) {
mbw.addMutation(m);
+
+ // Make a unique directory inside the table's dir. Cannot import multiple tables into one table, so don't need to use unique allocator
- String tabletDir = new String(FastFormat.toZeroPaddedString(dirCount++, 8, 16, Constants.CLONE_PREFIX_BYTES), Constants.UTF8);
++ String tabletDir = new String(FastFormat.toZeroPaddedString(dirCount++, 8, 16, Constants.CLONE_PREFIX_BYTES), StandardCharsets.UTF_8);
+
+ // Build up a full hdfs://localhost:8020/accumulo/tables/$id/c-XXXXXXX
+ String absolutePath = getClonedTabletDir(master, tableDirs, tabletDir);
+
m = new Mutation(metadataRow);
- TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(FastFormat.toZeroPaddedString(dirCount++, 8, 16, "/c-".getBytes(StandardCharsets.UTF_8))));
- TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(absolutePath.getBytes(Constants.UTF8)));
++ TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(absolutePath.getBytes(StandardCharsets.UTF_8)));
}
m.put(key.getColumnFamily(), cq, val);
[3/4] Merge branch '1.6'
Posted by el...@apache.org.
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f118b221/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
index 97902fe,0000000..cabf689
mode 100644,000000..100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
@@@ -1,2675 -1,0 +1,2675 @@@
+/*
+ * 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.
+ */
+package org.apache.accumulo.tserver.tablet;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.Durability;
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.impl.DurabilityImpl;
+import org.apache.accumulo.core.client.impl.ScannerImpl;
+import org.apache.accumulo.core.client.impl.Tables;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.ConfigurationCopy;
+import org.apache.accumulo.core.conf.ConfigurationObserver;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.constraints.Violations;
+import org.apache.accumulo.core.data.ByteSequence;
+import org.apache.accumulo.core.data.Column;
+import org.apache.accumulo.core.data.ColumnUpdate;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.KeyExtent;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.data.thrift.IterInfo;
+import org.apache.accumulo.core.data.thrift.MapFileInfo;
+import org.apache.accumulo.core.file.FileOperations;
+import org.apache.accumulo.core.file.FileSKVIterator;
+import org.apache.accumulo.core.iterators.IterationInterruptedException;
+import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.apache.accumulo.core.iterators.system.SourceSwitchingIterator;
+import org.apache.accumulo.core.master.thrift.TabletLoadState;
+import org.apache.accumulo.core.metadata.MetadataTable;
+import org.apache.accumulo.core.metadata.RootTable;
+import org.apache.accumulo.core.metadata.schema.DataFileValue;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.DataFileColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LogColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ScanFileColumnFamily;
+import org.apache.accumulo.core.protobuf.ProtobufUtil;
+import org.apache.accumulo.core.replication.ReplicationConfigurationUtil;
+import org.apache.accumulo.core.replication.StatusUtil;
+import org.apache.accumulo.core.replication.proto.Replication.Status;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.security.ColumnVisibility;
+import org.apache.accumulo.core.security.Credentials;
+import org.apache.accumulo.core.tabletserver.log.LogEntry;
+import org.apache.accumulo.core.tabletserver.thrift.TabletStats;
+import org.apache.accumulo.core.util.LocalityGroupUtil;
+import org.apache.accumulo.core.util.Pair;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.accumulo.server.client.HdfsZooInstance;
+import org.apache.accumulo.server.conf.TableConfiguration;
+import org.apache.accumulo.server.fs.FileRef;
+import org.apache.accumulo.server.fs.VolumeManager;
+import org.apache.accumulo.server.fs.VolumeManager.FileType;
+import org.apache.accumulo.server.fs.VolumeUtil;
+import org.apache.accumulo.server.fs.VolumeUtil.TabletFiles;
+import org.apache.accumulo.server.master.state.TServerInstance;
+import org.apache.accumulo.server.master.tableOps.CompactionIterators;
+import org.apache.accumulo.server.problems.ProblemReport;
+import org.apache.accumulo.server.problems.ProblemReports;
+import org.apache.accumulo.server.problems.ProblemType;
+import org.apache.accumulo.server.security.SystemCredentials;
+import org.apache.accumulo.server.tablets.TabletTime;
+import org.apache.accumulo.server.tablets.UniqueNameAllocator;
+import org.apache.accumulo.server.util.FileUtil;
+import org.apache.accumulo.server.util.MasterMetadataUtil;
+import org.apache.accumulo.server.util.MetadataTableUtil;
+import org.apache.accumulo.server.util.ReplicationTableUtil;
+import org.apache.accumulo.server.util.TabletOperations;
+import org.apache.accumulo.server.zookeeper.ZooReaderWriter;
+import org.apache.accumulo.start.classloader.vfs.AccumuloVFSClassLoader;
+import org.apache.accumulo.trace.instrument.Span;
+import org.apache.accumulo.trace.instrument.Trace;
+import org.apache.accumulo.tserver.InMemoryMap;
+import org.apache.accumulo.tserver.MinorCompactionReason;
+import org.apache.accumulo.tserver.TConstraintViolationException;
+import org.apache.accumulo.tserver.TLevel;
+import org.apache.accumulo.tserver.TabletServer;
+import org.apache.accumulo.tserver.TabletServerResourceManager.TabletResourceManager;
+import org.apache.accumulo.tserver.TabletStatsKeeper;
+import org.apache.accumulo.tserver.TabletStatsKeeper.Operation;
+import org.apache.accumulo.tserver.TooManyFilesException;
+import org.apache.accumulo.tserver.TservConstraintEnv;
+import org.apache.accumulo.tserver.compaction.CompactionPlan;
+import org.apache.accumulo.tserver.compaction.CompactionStrategy;
+import org.apache.accumulo.tserver.compaction.DefaultCompactionStrategy;
+import org.apache.accumulo.tserver.compaction.MajorCompactionReason;
+import org.apache.accumulo.tserver.compaction.MajorCompactionRequest;
+import org.apache.accumulo.tserver.compaction.WriteParameters;
+import org.apache.accumulo.tserver.constraints.ConstraintChecker;
+import org.apache.accumulo.tserver.log.DfsLogger;
+import org.apache.accumulo.tserver.log.MutationReceiver;
+import org.apache.accumulo.tserver.mastermessage.TabletStatusMessage;
+import org.apache.accumulo.tserver.metrics.TabletServerMinCMetrics;
+import org.apache.accumulo.tserver.tablet.Compactor.CompactionCanceledException;
+import org.apache.accumulo.tserver.tablet.Compactor.CompactionEnv;
+import org.apache.commons.codec.DecoderException;
+import org.apache.commons.codec.binary.Hex;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.log4j.Logger;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.NoNodeException;
+
+import com.google.common.annotations.VisibleForTesting;
+
+/**
+ *
+ * Provide access to a single row range in a living TabletServer.
+ *
+ */
+public class Tablet implements TabletCommitter {
+ static private final Logger log = Logger.getLogger(Tablet.class);
+ static private final List<LogEntry> NO_LOG_ENTRIES = Collections.emptyList();
+
+ private final TabletServer tabletServer;
+ private final KeyExtent extent;
+ private final TabletResourceManager tabletResources;
+ private final DatafileManager datafileManager;
+ private final TableConfiguration tableConfiguration;
+ private final String tabletDirectory;
+ private final Path location; // absolute path of this tablets dir
+
+ private final TabletMemory tabletMemory;
+
+ private final TabletTime tabletTime;
+ private final Object timeLock = new Object();
+ private long persistedTime;
+
+ private TServerInstance lastLocation;
+ private volatile boolean tableDirChecked = false;
+
+ private final AtomicLong dataSourceDeletions = new AtomicLong(0);
+
+ public long getDataSourceDeletions() {
+ return dataSourceDeletions.get();
+ }
+
+ private final Set<ScanDataSource> activeScans = new HashSet<ScanDataSource>();
+
+ private static enum CloseState {
+ OPEN, CLOSING, CLOSED, COMPLETE
+ }
+
+ private volatile CloseState closeState = CloseState.OPEN;
+
+ private boolean updatingFlushID = false;
+
+ private long lastFlushID = -1;
+ private long lastCompactID = -1;
+
+ private static class CompactionWaitInfo {
+ long flushID = -1;
+ long compactionID = -1;
+ }
+
+ // stores info about user initiated major compaction that is waiting on a minor compaction to finish
+ private CompactionWaitInfo compactionWaitInfo = new CompactionWaitInfo();
+
+ static enum CompactionState {
+ WAITING_TO_START, IN_PROGRESS
+ };
+
+ private volatile CompactionState minorCompactionState = null;
+ private volatile CompactionState majorCompactionState = null;
+
+ private final Set<MajorCompactionReason> majorCompactionQueued = Collections.synchronizedSet(EnumSet.noneOf(MajorCompactionReason.class));
+
+ private final AtomicReference<ConstraintChecker> constraintChecker = new AtomicReference<ConstraintChecker>();
+
+ private int writesInProgress = 0;
+
+ private final TabletStatsKeeper timer = new TabletStatsKeeper();
+
+ private final Rate queryRate = new Rate(0.2);
+ private long queryCount = 0;
+
+ private final Rate queryByteRate = new Rate(0.2);
+ private long queryBytes = 0;
+
+ private final Rate ingestRate = new Rate(0.2);
+ private long ingestCount = 0;
+
+ private final Rate ingestByteRate = new Rate(0.2);
+ private long ingestBytes = 0;
+
+ private byte[] defaultSecurityLabel = new byte[0];
+
+ private long lastMinorCompactionFinishTime = 0;
+ private long lastMapFileImportTime = 0;
+
+ private volatile long numEntries = 0;
+ private volatile long numEntriesInMemory = 0;
+
+ private final Rate scannedRate = new Rate(0.2);
+ private final AtomicLong scannedCount = new AtomicLong(0);
+
+ private final ConfigurationObserver configObserver;
+
+ private final int logId;
+
+ @Override
+ public int getLogId() {
+ return logId;
+ }
+
+ public static class LookupResult {
+ public List<Range> unfinishedRanges = new ArrayList<Range>();
+ public long bytesAdded = 0;
+ public long dataSize = 0;
+ public boolean closed = false;
+ }
+
+ FileRef getNextMapFilename(String prefix) throws IOException {
+ String extension = FileOperations.getNewFileExtension(tableConfiguration);
+ checkTabletDir();
+ return new FileRef(location.toString() + "/" + prefix + UniqueNameAllocator.getInstance().getNextName() + "." + extension);
+ }
+
+ private void checkTabletDir() throws IOException {
+ if (!tableDirChecked) {
+ FileStatus[] files = null;
+ try {
+ files = getTabletServer().getFileSystem().listStatus(location);
+ } catch (FileNotFoundException ex) {
+ // ignored
+ }
+
+ if (files == null) {
- if (location.getName().startsWith("c-"))
++ if (location.getName().startsWith(Constants.CLONE_PREFIX))
+ log.debug("Tablet " + extent + " had no dir, creating " + location); // its a clone dir...
+ else
+ log.warn("Tablet " + extent + " had no dir, creating " + location);
+
+ getTabletServer().getFileSystem().mkdirs(location);
+ }
+ tableDirChecked = true;
+ }
+ }
+
+ /**
+ * Only visible for testing
+ */
+ @VisibleForTesting
+ protected Tablet(TabletTime tabletTime, String tabletDirectory, int logId, Path location, DatafileManager datafileManager, TabletServer tabletServer,
+ TabletResourceManager tabletResources, TabletMemory tabletMemory, TableConfiguration tableConfiguration, KeyExtent extent,
+ ConfigurationObserver configObserver) {
+ this.tabletTime = tabletTime;
+ this.tabletDirectory = tabletDirectory;
+ this.logId = logId;
+ this.location = location;
+ this.datafileManager = datafileManager;
+ this.tabletServer = tabletServer;
+ this.tabletResources = tabletResources;
+ this.tabletMemory = tabletMemory;
+ this.tableConfiguration = tableConfiguration;
+ this.extent = extent;
+ this.configObserver = configObserver;
+ }
+
+ public Tablet(TabletServer tabletServer, KeyExtent extent, TabletResourceManager trm, SplitInfo info) throws IOException {
+ this(tabletServer, new Text(info.getDir()), extent, trm, info.getDatafiles(), info.getTime(), info.getInitFlushID(), info.getInitCompactID(), info
+ .getLastLocation());
+ splitCreationTime = System.currentTimeMillis();
+ }
+
+ private Tablet(TabletServer tabletServer, Text location, KeyExtent extent, TabletResourceManager trm, SortedMap<FileRef,DataFileValue> datafiles,
+ String time, long initFlushID, long initCompactID, TServerInstance lastLocation) throws IOException {
+ this(tabletServer, extent, location, trm, NO_LOG_ENTRIES, datafiles, time, lastLocation, new HashSet<FileRef>(), initFlushID, initCompactID);
+ }
+
+ private static String lookupTime(AccumuloConfiguration conf, KeyExtent extent, SortedMap<Key,Value> tabletsKeyValues) {
+ SortedMap<Key,Value> entries;
+
+ if (extent.isRootTablet()) {
+ return null;
+ } else {
+ entries = new TreeMap<Key,Value>();
+ Text rowName = extent.getMetadataEntry();
+ for (Entry<Key,Value> entry : tabletsKeyValues.entrySet()) {
+ if (entry.getKey().compareRow(rowName) == 0 && TabletsSection.ServerColumnFamily.TIME_COLUMN.hasColumns(entry.getKey())) {
+ entries.put(new Key(entry.getKey()), new Value(entry.getValue()));
+ }
+ }
+ }
+
+ if (entries.size() == 1)
+ return entries.values().iterator().next().toString();
+ return null;
+ }
+
+ private static SortedMap<FileRef,DataFileValue> lookupDatafiles(AccumuloConfiguration conf, VolumeManager fs, KeyExtent extent,
+ SortedMap<Key,Value> tabletsKeyValues) throws IOException {
+
+ TreeMap<FileRef,DataFileValue> datafiles = new TreeMap<FileRef,DataFileValue>();
+
+ if (extent.isRootTablet()) { // the meta0 tablet
+ Path location = new Path(MetadataTableUtil.getRootTabletDir());
+
+ // cleanUpFiles() has special handling for delete. files
+ FileStatus[] files = fs.listStatus(location);
+ Collection<String> goodPaths = RootFiles.cleanupReplacement(fs, files, true);
+ for (String good : goodPaths) {
+ Path path = new Path(good);
+ String filename = path.getName();
+ FileRef ref = new FileRef(location.toString() + "/" + filename, path);
+ DataFileValue dfv = new DataFileValue(0, 0);
+ datafiles.put(ref, dfv);
+ }
+ } else {
+
+ Text rowName = extent.getMetadataEntry();
+
+ String tableId = extent.isMeta() ? RootTable.ID : MetadataTable.ID;
+ ScannerImpl mdScanner = new ScannerImpl(HdfsZooInstance.getInstance(), SystemCredentials.get(), tableId, Authorizations.EMPTY);
+
+ // Commented out because when no data file is present, each tablet will scan through metadata table and return nothing
+ // reduced batch size to improve performance
+ // changed here after endKeys were implemented from 10 to 1000
+ mdScanner.setBatchSize(1000);
+
+ // leave these in, again, now using endKey for safety
+ mdScanner.fetchColumnFamily(DataFileColumnFamily.NAME);
+
+ mdScanner.setRange(new Range(rowName));
+
+ for (Entry<Key,Value> entry : mdScanner) {
+ if (entry.getKey().compareRow(rowName) != 0) {
+ break;
+ }
+
+ FileRef ref = new FileRef(fs, entry.getKey());
+ datafiles.put(ref, new DataFileValue(entry.getValue().get()));
+ }
+ }
+ return datafiles;
+ }
+
+ private static List<LogEntry> lookupLogEntries(KeyExtent ke, SortedMap<Key,Value> tabletsKeyValues) {
+ List<LogEntry> logEntries = new ArrayList<LogEntry>();
+
+ if (ke.isMeta()) {
+ try {
+ logEntries = MetadataTableUtil.getLogEntries(SystemCredentials.get(), ke);
+ } catch (Exception ex) {
+ throw new RuntimeException("Unable to read tablet log entries", ex);
+ }
+ } else {
+ log.debug("Looking at metadata " + tabletsKeyValues);
+ Text row = ke.getMetadataEntry();
+ for (Entry<Key,Value> entry : tabletsKeyValues.entrySet()) {
+ Key key = entry.getKey();
+ if (key.getRow().equals(row)) {
+ if (key.getColumnFamily().equals(LogColumnFamily.NAME)) {
+ logEntries.add(LogEntry.fromKeyValue(key, entry.getValue()));
+ }
+ }
+ }
+ }
+
+ log.debug("got " + logEntries + " for logs for " + ke);
+ return logEntries;
+ }
+
+ private static Set<FileRef> lookupScanFiles(KeyExtent extent, SortedMap<Key,Value> tabletsKeyValues, VolumeManager fs) throws IOException {
+ HashSet<FileRef> scanFiles = new HashSet<FileRef>();
+
+ Text row = extent.getMetadataEntry();
+ for (Entry<Key,Value> entry : tabletsKeyValues.entrySet()) {
+ Key key = entry.getKey();
+ if (key.getRow().equals(row) && key.getColumnFamily().equals(ScanFileColumnFamily.NAME)) {
+ scanFiles.add(new FileRef(fs, key));
+ }
+ }
+
+ return scanFiles;
+ }
+
+ private static long lookupFlushID(KeyExtent extent, SortedMap<Key,Value> tabletsKeyValues) {
+ Text row = extent.getMetadataEntry();
+ for (Entry<Key,Value> entry : tabletsKeyValues.entrySet()) {
+ Key key = entry.getKey();
+ if (key.getRow().equals(row) && TabletsSection.ServerColumnFamily.FLUSH_COLUMN.equals(key.getColumnFamily(), key.getColumnQualifier()))
+ return Long.parseLong(entry.getValue().toString());
+ }
+
+ return -1;
+ }
+
+ private static long lookupCompactID(KeyExtent extent, SortedMap<Key,Value> tabletsKeyValues) {
+ Text row = extent.getMetadataEntry();
+ for (Entry<Key,Value> entry : tabletsKeyValues.entrySet()) {
+ Key key = entry.getKey();
+ if (key.getRow().equals(row) && TabletsSection.ServerColumnFamily.COMPACT_COLUMN.equals(key.getColumnFamily(), key.getColumnQualifier()))
+ return Long.parseLong(entry.getValue().toString());
+ }
+
+ return -1;
+ }
+
+ private static TServerInstance lookupLastServer(KeyExtent extent, SortedMap<Key,Value> tabletsKeyValues) {
+ for (Entry<Key,Value> entry : tabletsKeyValues.entrySet()) {
+ if (entry.getKey().getColumnFamily().compareTo(TabletsSection.LastLocationColumnFamily.NAME) == 0) {
+ return new TServerInstance(entry.getValue(), entry.getKey().getColumnQualifier());
+ }
+ }
+ return null;
+ }
+
+ public Tablet(TabletServer tabletServer, KeyExtent extent, Text location, TabletResourceManager trm, SortedMap<Key,Value> tabletsKeyValues)
+ throws IOException {
+ this(tabletServer, extent, location, trm, lookupLogEntries(extent, tabletsKeyValues), lookupDatafiles(tabletServer.getConfiguration(),
+ tabletServer.getFileSystem(), extent, tabletsKeyValues), lookupTime(tabletServer.getConfiguration(), extent, tabletsKeyValues), lookupLastServer(
+ extent, tabletsKeyValues), lookupScanFiles(extent, tabletsKeyValues, tabletServer.getFileSystem()), lookupFlushID(extent, tabletsKeyValues),
+ lookupCompactID(extent, tabletsKeyValues));
+ }
+
+ /**
+ * yet another constructor - this one allows us to avoid costly lookups into the Metadata table if we already know the files we need - as at split time
+ */
+ private Tablet(final TabletServer tabletServer, final KeyExtent extent, final Text location, final TabletResourceManager trm,
+ final List<LogEntry> rawLogEntries, final SortedMap<FileRef,DataFileValue> rawDatafiles, String time, final TServerInstance lastLocation,
+ Set<FileRef> scanFiles, long initFlushID, long initCompactID) throws IOException {
+
+ TableConfiguration tblConf = tabletServer.getTableConfiguration(extent);
+ if (null == tblConf) {
+ Tables.clearCache(tabletServer.getInstance());
+ tblConf = tabletServer.getTableConfiguration(extent);
+ if (null == tblConf) {
+ // Not guaranteed to be non-null, but should be. A failed load will be re-assigned though..
+ log.warn("Could not get table configuration for " + extent.getTableId().toString());
+ }
+ }
+
+ this.tableConfiguration = tblConf;
+
+ TabletFiles tabletPaths = VolumeUtil.updateTabletVolumes(tabletServer.getLock(), tabletServer.getFileSystem(), extent, new TabletFiles(location.toString(),
+ rawLogEntries, rawDatafiles), ReplicationConfigurationUtil.isEnabled(extent, this.tableConfiguration));
+
+ Path locationPath;
+
+ if (tabletPaths.dir.contains(":")) {
+ locationPath = new Path(tabletPaths.dir.toString());
+ } else {
+ locationPath = tabletServer.getFileSystem().getFullPath(FileType.TABLE, extent.getTableId().toString() + tabletPaths.dir.toString());
+ }
+
+ final List<LogEntry> logEntries = tabletPaths.logEntries;
+ final SortedMap<FileRef,DataFileValue> datafiles = tabletPaths.datafiles;
+
+ this.location = locationPath;
+ this.lastLocation = lastLocation;
+ this.tabletDirectory = tabletPaths.dir;
+
+ this.extent = extent;
+ this.tabletResources = trm;
+
+ this.lastFlushID = initFlushID;
+ this.lastCompactID = initCompactID;
+
+ if (extent.isRootTablet()) {
+ long rtime = Long.MIN_VALUE;
+ for (FileRef ref : datafiles.keySet()) {
+ Path path = ref.path();
+ FileSystem ns = tabletServer.getFileSystem().getVolumeByPath(path).getFileSystem();
+ FileSKVIterator reader = FileOperations.getInstance().openReader(path.toString(), true, ns, ns.getConf(), tabletServer.getTableConfiguration(extent));
+ long maxTime = -1;
+ try {
+
+ while (reader.hasTop()) {
+ maxTime = Math.max(maxTime, reader.getTopKey().getTimestamp());
+ reader.next();
+ }
+
+ } finally {
+ reader.close();
+ }
+
+ if (maxTime > rtime) {
+ time = TabletTime.LOGICAL_TIME_ID + "" + maxTime;
+ rtime = maxTime;
+ }
+ }
+ }
+ if (time == null && datafiles.isEmpty() && extent.equals(RootTable.OLD_EXTENT)) {
+ // recovery... old root tablet has no data, so time doesn't matter:
+ time = TabletTime.LOGICAL_TIME_ID + "" + Long.MIN_VALUE;
+ }
+
+ this.tabletServer = tabletServer;
+ this.logId = tabletServer.createLogId(extent);
+
+ setupDefaultSecurityLabels(extent);
+
+ tabletMemory = new TabletMemory(this);
+ tabletTime = TabletTime.getInstance(time);
+ persistedTime = tabletTime.getTime();
+
+ tableConfiguration.addObserver(configObserver = new ConfigurationObserver() {
+
+ private void reloadConstraints() {
+ constraintChecker.set(new ConstraintChecker(tableConfiguration));
+ }
+
+ @Override
+ public void propertiesChanged() {
+ reloadConstraints();
+
+ try {
+ setupDefaultSecurityLabels(extent);
+ } catch (Exception e) {
+ log.error("Failed to reload default security labels for extent: " + extent.toString());
+ }
+ }
+
+ @Override
+ public void propertyChanged(String prop) {
+ if (prop.startsWith(Property.TABLE_CONSTRAINT_PREFIX.getKey()))
+ reloadConstraints();
+ else if (prop.equals(Property.TABLE_DEFAULT_SCANTIME_VISIBILITY.getKey())) {
+ try {
+ log.info("Default security labels changed for extent: " + extent.toString());
+ setupDefaultSecurityLabels(extent);
+ } catch (Exception e) {
+ log.error("Failed to reload default security labels for extent: " + extent.toString());
+ }
+ }
+
+ }
+
+ @Override
+ public void sessionExpired() {
+ log.debug("Session expired, no longer updating per table props...");
+ }
+
+ });
+
+ tableConfiguration.getNamespaceConfiguration().addObserver(configObserver);
+
+ // Force a load of any per-table properties
+ configObserver.propertiesChanged();
+
+ if (!logEntries.isEmpty()) {
+ log.info("Starting Write-Ahead Log recovery for " + this.extent);
+ // count[0] = entries used on tablet
+ // count[1] = track max time from walog entries wihtout timestamps
+ final long[] count = new long[2];
+ final CommitSession commitSession = getTabletMemory().getCommitSession();
+ count[1] = Long.MIN_VALUE;
+ try {
+ Set<String> absPaths = new HashSet<String>();
+ for (FileRef ref : datafiles.keySet())
+ absPaths.add(ref.path().toString());
+
+ tabletServer.recover(this.getTabletServer().getFileSystem(), extent, tableConfiguration, logEntries, absPaths, new MutationReceiver() {
+ @Override
+ public void receive(Mutation m) {
+ // LogReader.printMutation(m);
+ Collection<ColumnUpdate> muts = m.getUpdates();
+ for (ColumnUpdate columnUpdate : muts) {
+ if (!columnUpdate.hasTimestamp()) {
+ // if it is not a user set timestamp, it must have been set
+ // by the system
+ count[1] = Math.max(count[1], columnUpdate.getTimestamp());
+ }
+ }
+ getTabletMemory().mutate(commitSession, Collections.singletonList(m));
+ count[0]++;
+ }
+ });
+
+ if (count[1] != Long.MIN_VALUE) {
+ tabletTime.useMaxTimeFromWALog(count[1]);
+ }
+ commitSession.updateMaxCommittedTime(tabletTime.getTime());
+
+ if (count[0] == 0) {
+ log.debug("No replayed mutations applied, removing unused entries for " + extent);
+ MetadataTableUtil.removeUnusedWALEntries(extent, logEntries, tabletServer.getLock());
+
+ // Ensure that we write a record marking each WAL as requiring replication to make sure we don't abandon the data
+ if (ReplicationConfigurationUtil.isEnabled(extent, tabletServer.getTableConfiguration(extent))) {
+ Status status = StatusUtil.fileClosed();
+ for (LogEntry logEntry : logEntries) {
+ log.debug("Writing closed status to metadata table for " + logEntry.logSet + " " + ProtobufUtil.toString(status));
+ ReplicationTableUtil.updateFiles(SystemCredentials.get(), extent, logEntry.logSet, status);
+ }
+ }
+
+ logEntries.clear();
+ } else if (ReplicationConfigurationUtil.isEnabled(extent, tabletServer.getTableConfiguration(extent))) {
+ // The logs are about to be re-used, we need to record that they have data for this extent,
+ // but that they may get more data
+ Status status = StatusUtil.openWithUnknownLength();
+ for (LogEntry logEntry : logEntries) {
+ log.debug("Writing updated status to metadata table for " + logEntry.logSet + " " + ProtobufUtil.toString(status));
+ ReplicationTableUtil.updateFiles(SystemCredentials.get(), extent, logEntry.logSet, status);
+ }
+ }
+
+ } catch (Throwable t) {
+ if (tableConfiguration.getBoolean(Property.TABLE_FAILURES_IGNORE)) {
+ log.warn("Error recovering from log files: ", t);
+ } else {
+ throw new RuntimeException(t);
+ }
+ }
+ // make some closed references that represent the recovered logs
+ currentLogs = new HashSet<DfsLogger>();
+ for (LogEntry logEntry : logEntries) {
+ for (String log : logEntry.logSet) {
+ currentLogs.add(new DfsLogger(tabletServer.getServerConfig(), log, logEntry.getColumnQualifier().toString()));
+ }
+ }
+
+ log.info("Write-Ahead Log recovery complete for " + this.extent + " (" + count[0] + " mutations applied, " + getTabletMemory().getNumEntries()
+ + " entries created)");
+ }
+
+ String contextName = tableConfiguration.get(Property.TABLE_CLASSPATH);
+ if (contextName != null && !contextName.equals("")) {
+ // initialize context classloader, instead of possibly waiting for it to initialize for a scan
+ // TODO this could hang, causing other tablets to fail to load - ACCUMULO-1292
+ AccumuloVFSClassLoader.getContextManager().getClassLoader(contextName);
+ }
+
+ // do this last after tablet is completely setup because it
+ // could cause major compaction to start
+ datafileManager = new DatafileManager(this, datafiles);
+
+ computeNumEntries();
+
+ getDatafileManager().removeFilesAfterScan(scanFiles);
+
+ // look for hints of a failure on the previous tablet server
+ if (!logEntries.isEmpty() || needsMajorCompaction(MajorCompactionReason.NORMAL)) {
+ // look for any temp files hanging around
+ removeOldTemporaryFiles();
+ }
+
+ log.log(TLevel.TABLET_HIST, extent + " opened");
+ }
+
+ private void removeOldTemporaryFiles() {
+ // remove any temporary files created by a previous tablet server
+ try {
+ for (FileStatus tmp : getTabletServer().getFileSystem().globStatus(new Path(location, "*_tmp"))) {
+ try {
+ log.debug("Removing old temp file " + tmp.getPath());
+ getTabletServer().getFileSystem().delete(tmp.getPath());
+ } catch (IOException ex) {
+ log.error("Unable to remove old temp file " + tmp.getPath() + ": " + ex);
+ }
+ }
+ } catch (IOException ex) {
+ log.error("Error scanning for old temp files in " + location);
+ }
+ }
+
+ private void setupDefaultSecurityLabels(KeyExtent extent) {
+ if (extent.isMeta()) {
+ defaultSecurityLabel = new byte[0];
+ } else {
+ try {
+ ColumnVisibility cv = new ColumnVisibility(tableConfiguration.get(Property.TABLE_DEFAULT_SCANTIME_VISIBILITY));
+ this.defaultSecurityLabel = cv.getExpression();
+ } catch (Exception e) {
+ log.error(e, e);
+ this.defaultSecurityLabel = new byte[0];
+ }
+ }
+ }
+
+ private LookupResult lookup(SortedKeyValueIterator<Key,Value> mmfi, List<Range> ranges, HashSet<Column> columnSet, List<KVEntry> results, long maxResultsSize)
+ throws IOException {
+
+ LookupResult lookupResult = new LookupResult();
+
+ boolean exceededMemoryUsage = false;
+ boolean tabletClosed = false;
+
+ Set<ByteSequence> cfset = null;
+ if (columnSet.size() > 0)
+ cfset = LocalityGroupUtil.families(columnSet);
+
+ for (Range range : ranges) {
+
+ if (exceededMemoryUsage || tabletClosed) {
+ lookupResult.unfinishedRanges.add(range);
+ continue;
+ }
+
+ int entriesAdded = 0;
+
+ try {
+ if (cfset != null)
+ mmfi.seek(range, cfset, true);
+ else
+ mmfi.seek(range, LocalityGroupUtil.EMPTY_CF_SET, false);
+
+ while (mmfi.hasTop()) {
+ Key key = mmfi.getTopKey();
+
+ KVEntry kve = new KVEntry(key, mmfi.getTopValue());
+ results.add(kve);
+ entriesAdded++;
+ lookupResult.bytesAdded += kve.estimateMemoryUsed();
+ lookupResult.dataSize += kve.numBytes();
+
+ exceededMemoryUsage = lookupResult.bytesAdded > maxResultsSize;
+
+ if (exceededMemoryUsage) {
+ addUnfinishedRange(lookupResult, range, key, false);
+ break;
+ }
+
+ mmfi.next();
+ }
+
+ } catch (TooManyFilesException tmfe) {
+ // treat this as a closed tablet, and let the client retry
+ log.warn("Tablet " + getExtent() + " has too many files, batch lookup can not run");
+ handleTabletClosedDuringScan(results, lookupResult, exceededMemoryUsage, range, entriesAdded);
+ tabletClosed = true;
+ } catch (IOException ioe) {
+ if (shutdownInProgress()) {
+ // assume HDFS shutdown hook caused this exception
+ log.debug("IOException while shutdown in progress ", ioe);
+ handleTabletClosedDuringScan(results, lookupResult, exceededMemoryUsage, range, entriesAdded);
+ tabletClosed = true;
+ } else {
+ throw ioe;
+ }
+ } catch (IterationInterruptedException iie) {
+ if (isClosed()) {
+ handleTabletClosedDuringScan(results, lookupResult, exceededMemoryUsage, range, entriesAdded);
+ tabletClosed = true;
+ } else {
+ throw iie;
+ }
+ } catch (TabletClosedException tce) {
+ handleTabletClosedDuringScan(results, lookupResult, exceededMemoryUsage, range, entriesAdded);
+ tabletClosed = true;
+ }
+
+ }
+
+ return lookupResult;
+ }
+
+ private void handleTabletClosedDuringScan(List<KVEntry> results, LookupResult lookupResult, boolean exceededMemoryUsage, Range range, int entriesAdded) {
+ if (exceededMemoryUsage)
+ throw new IllegalStateException("tablet should not exceed memory usage or close, not both");
+
+ if (entriesAdded > 0)
+ addUnfinishedRange(lookupResult, range, results.get(results.size() - 1).getKey(), false);
+ else
+ lookupResult.unfinishedRanges.add(range);
+
+ lookupResult.closed = true;
+ }
+
+ private void addUnfinishedRange(LookupResult lookupResult, Range range, Key key, boolean inclusiveStartKey) {
+ if (range.getEndKey() == null || key.compareTo(range.getEndKey()) < 0) {
+ Range nlur = new Range(new Key(key), inclusiveStartKey, range.getEndKey(), range.isEndKeyInclusive());
+ lookupResult.unfinishedRanges.add(nlur);
+ }
+ }
+
+ public LookupResult lookup(List<Range> ranges, HashSet<Column> columns, Authorizations authorizations, List<KVEntry> results, long maxResultSize,
+ List<IterInfo> ssiList, Map<String,Map<String,String>> ssio, AtomicBoolean interruptFlag) throws IOException {
+
+ if (ranges.size() == 0) {
+ return new LookupResult();
+ }
+
+ ranges = Range.mergeOverlapping(ranges);
+ Collections.sort(ranges);
+
+ Range tabletRange = extent.toDataRange();
+ for (Range range : ranges) {
+ // do a test to see if this range falls within the tablet, if it does not
+ // then clip will throw an exception
+ tabletRange.clip(range);
+ }
+
+ ScanDataSource dataSource = new ScanDataSource(this, authorizations, this.defaultSecurityLabel, columns, ssiList, ssio, interruptFlag);
+
+ LookupResult result = null;
+
+ try {
+ SortedKeyValueIterator<Key,Value> iter = new SourceSwitchingIterator(dataSource);
+ result = lookup(iter, ranges, columns, results, maxResultSize);
+ return result;
+ } catch (IOException ioe) {
+ dataSource.close(true);
+ throw ioe;
+ } finally {
+ // code in finally block because always want
+ // to return mapfiles, even when exception is thrown
+ dataSource.close(false);
+
+ synchronized (this) {
+ queryCount += results.size();
+ if (result != null)
+ queryBytes += result.dataSize;
+ }
+ }
+ }
+
+ Batch nextBatch(SortedKeyValueIterator<Key,Value> iter, Range range, int num, Set<Column> columns) throws IOException {
+
+ // log.info("In nextBatch..");
+
+ List<KVEntry> results = new ArrayList<KVEntry>();
+ Key key = null;
+
+ Value value;
+ long resultSize = 0L;
+ long resultBytes = 0L;
+
+ long maxResultsSize = tableConfiguration.getMemoryInBytes(Property.TABLE_SCAN_MAXMEM);
+
+ if (columns.size() == 0) {
+ iter.seek(range, LocalityGroupUtil.EMPTY_CF_SET, false);
+ } else {
+ iter.seek(range, LocalityGroupUtil.families(columns), true);
+ }
+
+ Key continueKey = null;
+ boolean skipContinueKey = false;
+
+ boolean endOfTabletReached = false;
+ while (iter.hasTop()) {
+
+ value = iter.getTopValue();
+ key = iter.getTopKey();
+
+ KVEntry kvEntry = new KVEntry(key, value); // copies key and value
+ results.add(kvEntry);
+ resultSize += kvEntry.estimateMemoryUsed();
+ resultBytes += kvEntry.numBytes();
+
+ if (resultSize >= maxResultsSize || results.size() >= num) {
+ continueKey = new Key(key);
+ skipContinueKey = true;
+ break;
+ }
+
+ iter.next();
+ }
+
+ if (iter.hasTop() == false) {
+ endOfTabletReached = true;
+ }
+
+ if (endOfTabletReached) {
+ continueKey = null;
+ }
+
+ if (endOfTabletReached && results.size() == 0)
+ results = null;
+
+ return new Batch(skipContinueKey, results, continueKey, resultBytes);
+ }
+
+ /**
+ * Determine if a JVM shutdown is in progress.
+ *
+ */
+ boolean shutdownInProgress() {
+ try {
+ Runtime.getRuntime().removeShutdownHook(new Thread(new Runnable() {
+ @Override
+ public void run() {}
+ }));
+ } catch (IllegalStateException ise) {
+ return true;
+ }
+
+ return false;
+ }
+
+ public Scanner createScanner(Range range, int num, Set<Column> columns, Authorizations authorizations, List<IterInfo> ssiList,
+ Map<String,Map<String,String>> ssio, boolean isolated, AtomicBoolean interruptFlag) {
+ // do a test to see if this range falls within the tablet, if it does not
+ // then clip will throw an exception
+ extent.toDataRange().clip(range);
+
+ ScanOptions opts = new ScanOptions(num, authorizations, this.defaultSecurityLabel, columns, ssiList, ssio, interruptFlag, isolated);
+ return new Scanner(this, range, opts);
+ }
+
+ DataFileValue minorCompact(VolumeManager fs, InMemoryMap memTable, FileRef tmpDatafile, FileRef newDatafile, FileRef mergeFile, boolean hasQueueTime,
+ long queued, CommitSession commitSession, long flushId, MinorCompactionReason mincReason) {
+ boolean failed = false;
+ long start = System.currentTimeMillis();
+ timer.incrementStatusMinor();
+
+ long count = 0;
+
+ try {
+ Span span = Trace.start("write");
+ CompactionStats stats;
+ try {
+ count = memTable.getNumEntries();
+
+ DataFileValue dfv = null;
+ if (mergeFile != null)
+ dfv = getDatafileManager().getDatafileSizes().get(mergeFile);
+
+ MinorCompactor compactor = new MinorCompactor(this, memTable, mergeFile, dfv, tmpDatafile, mincReason, tableConfiguration);
+ stats = compactor.call();
+ } finally {
+ span.stop();
+ }
+ span = Trace.start("bringOnline");
+ try {
+ getDatafileManager().bringMinorCompactionOnline(tmpDatafile, newDatafile, mergeFile, new DataFileValue(stats.getFileSize(), stats.getEntriesWritten()),
+ commitSession, flushId);
+ } finally {
+ span.stop();
+ }
+ return new DataFileValue(stats.getFileSize(), stats.getEntriesWritten());
+ } catch (Exception e) {
+ failed = true;
+ throw new RuntimeException(e);
+ } catch (Error e) {
+ // Weird errors like "OutOfMemoryError" when trying to create the thread for the compaction
+ failed = true;
+ throw new RuntimeException(e);
+ } finally {
+ try {
+ getTabletMemory().finalizeMinC();
+ } catch (Throwable t) {
+ log.error("Failed to free tablet memory", t);
+ }
+
+ if (!failed) {
+ lastMinorCompactionFinishTime = System.currentTimeMillis();
+ }
+ TabletServerMinCMetrics minCMetrics = getTabletServer().getMinCMetrics();
+ if (minCMetrics.isEnabled())
+ minCMetrics.add(TabletServerMinCMetrics.minc, (lastMinorCompactionFinishTime - start));
+ if (hasQueueTime) {
+ timer.updateTime(Operation.MINOR, queued, start, count, failed);
+ if (minCMetrics.isEnabled())
+ minCMetrics.add(TabletServerMinCMetrics.queue, (start - queued));
+ } else
+ timer.updateTime(Operation.MINOR, start, count, failed);
+ }
+ }
+
+ private synchronized MinorCompactionTask prepareForMinC(long flushId, MinorCompactionReason mincReason) {
+ CommitSession oldCommitSession = getTabletMemory().prepareForMinC();
+ otherLogs = currentLogs;
+ currentLogs = new HashSet<DfsLogger>();
+
+ FileRef mergeFile = null;
+ if (mincReason != MinorCompactionReason.RECOVERY) {
+ mergeFile = getDatafileManager().reserveMergingMinorCompactionFile();
+ }
+
+ return new MinorCompactionTask(this, mergeFile, oldCommitSession, flushId, mincReason);
+
+ }
+
+ public void flush(long tableFlushID) {
+ boolean updateMetadata = false;
+ boolean initiateMinor = false;
+
+ try {
+
+ synchronized (this) {
+
+ // only want one thing at a time to update flush ID to ensure that metadata table and tablet in memory state are consistent
+ if (updatingFlushID)
+ return;
+
+ if (lastFlushID >= tableFlushID)
+ return;
+
+ if (isClosing() || isClosed() || getTabletMemory().memoryReservedForMinC())
+ return;
+
+ if (getTabletMemory().getMemTable().getNumEntries() == 0) {
+ lastFlushID = tableFlushID;
+ updatingFlushID = true;
+ updateMetadata = true;
+ } else
+ initiateMinor = true;
+ }
+
+ if (updateMetadata) {
+ Credentials creds = SystemCredentials.get();
+ // if multiple threads were allowed to update this outside of a sync block, then it would be
+ // a race condition
+ MetadataTableUtil.updateTabletFlushID(extent, tableFlushID, creds, getTabletServer().getLock());
+ } else if (initiateMinor)
+ initiateMinorCompaction(tableFlushID, MinorCompactionReason.USER);
+
+ } finally {
+ if (updateMetadata) {
+ synchronized (this) {
+ updatingFlushID = false;
+ this.notifyAll();
+ }
+ }
+ }
+
+ }
+
+ public boolean initiateMinorCompaction(MinorCompactionReason mincReason) {
+ if (isClosed()) {
+ // don't bother trying to get flush id if closed... could be closed after this check but that is ok... just trying to cut down on uneeded log messages....
+ return false;
+ }
+
+ // get the flush id before the new memmap is made available for write
+ long flushId;
+ try {
+ flushId = getFlushID();
+ } catch (NoNodeException e) {
+ log.info("Asked to initiate MinC when there was no flush id " + getExtent() + " " + e.getMessage());
+ return false;
+ }
+ return initiateMinorCompaction(flushId, mincReason);
+ }
+
+ public boolean minorCompactNow(MinorCompactionReason mincReason) {
+ long flushId;
+ try {
+ flushId = getFlushID();
+ } catch (NoNodeException e) {
+ log.info("Asked to initiate MinC when there was no flush id " + getExtent() + " " + e.getMessage());
+ return false;
+ }
+ MinorCompactionTask mct = createMinorCompactionTask(flushId, mincReason);
+ if (mct == null)
+ return false;
+ mct.run();
+ return true;
+ }
+
+ boolean initiateMinorCompaction(long flushId, MinorCompactionReason mincReason) {
+ MinorCompactionTask mct = createMinorCompactionTask(flushId, mincReason);
+ if (mct == null)
+ return false;
+ getTabletResources().executeMinorCompaction(mct);
+ return true;
+ }
+
+ private MinorCompactionTask createMinorCompactionTask(long flushId, MinorCompactionReason mincReason) {
+ MinorCompactionTask mct;
+ long t1, t2;
+
+ StringBuilder logMessage = null;
+
+ try {
+ synchronized (this) {
+ t1 = System.currentTimeMillis();
+
+ if (isClosing() || isClosed() || majorCompactionState == CompactionState.WAITING_TO_START || getTabletMemory().memoryReservedForMinC()
+ || getTabletMemory().getMemTable().getNumEntries() == 0 || updatingFlushID) {
+
+ logMessage = new StringBuilder();
+
+ logMessage.append(extent.toString());
+ logMessage.append(" closeState " + closeState);
+ logMessage.append(" majorCompactionState " + majorCompactionState);
+ if (getTabletMemory() != null)
+ logMessage.append(" tabletMemory.memoryReservedForMinC() " + getTabletMemory().memoryReservedForMinC());
+ if (getTabletMemory() != null && getTabletMemory().getMemTable() != null)
+ logMessage.append(" tabletMemory.getMemTable().getNumEntries() " + getTabletMemory().getMemTable().getNumEntries());
+ logMessage.append(" updatingFlushID " + updatingFlushID);
+
+ return null;
+ }
+
+ mct = prepareForMinC(flushId, mincReason);
+ t2 = System.currentTimeMillis();
+ }
+ } finally {
+ // log outside of sync block
+ if (logMessage != null && log.isDebugEnabled())
+ log.debug(logMessage);
+ }
+
+ log.debug(String.format("MinC initiate lock %.2f secs", (t2 - t1) / 1000.0));
+ return mct;
+ }
+
+ public long getFlushID() throws NoNodeException {
+ try {
+ String zTablePath = Constants.ZROOT + "/" + HdfsZooInstance.getInstance().getInstanceID() + Constants.ZTABLES + "/" + extent.getTableId()
+ + Constants.ZTABLE_FLUSH_ID;
+ return Long.parseLong(new String(ZooReaderWriter.getRetryingInstance().getData(zTablePath, null), StandardCharsets.UTF_8));
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ } catch (NumberFormatException nfe) {
+ throw new RuntimeException(nfe);
+ } catch (KeeperException ke) {
+ if (ke instanceof NoNodeException) {
+ throw (NoNodeException) ke;
+ } else {
+ throw new RuntimeException(ke);
+ }
+ }
+ }
+
+ long getCompactionCancelID() {
+ String zTablePath = Constants.ZROOT + "/" + HdfsZooInstance.getInstance().getInstanceID() + Constants.ZTABLES + "/" + extent.getTableId()
+ + Constants.ZTABLE_COMPACT_CANCEL_ID;
+
+ try {
+ return Long.parseLong(new String(ZooReaderWriter.getRetryingInstance().getData(zTablePath, null), StandardCharsets.UTF_8));
+ } catch (KeeperException e) {
+ throw new RuntimeException(e);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public Pair<Long,List<IteratorSetting>> getCompactionID() throws NoNodeException {
+ try {
+ String zTablePath = Constants.ZROOT + "/" + HdfsZooInstance.getInstance().getInstanceID() + Constants.ZTABLES + "/" + extent.getTableId()
+ + Constants.ZTABLE_COMPACT_ID;
+
+ String[] tokens = new String(ZooReaderWriter.getRetryingInstance().getData(zTablePath, null), StandardCharsets.UTF_8).split(",");
+ long compactID = Long.parseLong(tokens[0]);
+
+ CompactionIterators iters = new CompactionIterators();
+
+ if (tokens.length > 1) {
+ Hex hex = new Hex();
+ ByteArrayInputStream bais = new ByteArrayInputStream(hex.decode(tokens[1].split("=")[1].getBytes(StandardCharsets.UTF_8)));
+ DataInputStream dis = new DataInputStream(bais);
+
+ try {
+ iters.readFields(dis);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ KeyExtent ke = new KeyExtent(extent.getTableId(), iters.getEndRow(), iters.getStartRow());
+
+ if (!ke.overlaps(extent)) {
+ // only use iterators if compaction range overlaps
+ iters = new CompactionIterators();
+ }
+ }
+
+ return new Pair<Long,List<IteratorSetting>>(compactID, iters.getIterators());
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ } catch (NumberFormatException nfe) {
+ throw new RuntimeException(nfe);
+ } catch (KeeperException ke) {
+ if (ke instanceof NoNodeException) {
+ throw (NoNodeException) ke;
+ } else {
+ throw new RuntimeException(ke);
+ }
+ } catch (DecoderException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public synchronized void waitForMinC() {
+ getTabletMemory().waitForMinC();
+ }
+
+ private synchronized CommitSession finishPreparingMutations(long time) {
+ if (writesInProgress < 0) {
+ throw new IllegalStateException("waitingForLogs < 0 " + writesInProgress);
+ }
+
+ if (isClosed() || getTabletMemory() == null) {
+ return null;
+ }
+
+ writesInProgress++;
+ CommitSession commitSession = getTabletMemory().getCommitSession();
+ commitSession.incrementCommitsInProgress();
+ commitSession.updateMaxCommittedTime(time);
+ return commitSession;
+ }
+
+ public void checkConstraints() {
+ ConstraintChecker cc = constraintChecker.get();
+
+ if (cc.classLoaderChanged()) {
+ ConstraintChecker ncc = new ConstraintChecker(tableConfiguration);
+ constraintChecker.compareAndSet(cc, ncc);
+ }
+ }
+
+ public CommitSession prepareMutationsForCommit(TservConstraintEnv cenv, List<Mutation> mutations) throws TConstraintViolationException {
+
+ ConstraintChecker cc = constraintChecker.get();
+
+ List<Mutation> violators = null;
+ Violations violations = new Violations();
+ cenv.setExtent(extent);
+ for (Mutation mutation : mutations) {
+ Violations more = cc.check(cenv, mutation);
+ if (more != null) {
+ violations.add(more);
+ if (violators == null)
+ violators = new ArrayList<Mutation>();
+ violators.add(mutation);
+ }
+ }
+
+ long time = tabletTime.setUpdateTimes(mutations);
+
+ if (!violations.isEmpty()) {
+
+ HashSet<Mutation> violatorsSet = new HashSet<Mutation>(violators);
+ ArrayList<Mutation> nonViolators = new ArrayList<Mutation>();
+
+ for (Mutation mutation : mutations) {
+ if (!violatorsSet.contains(mutation)) {
+ nonViolators.add(mutation);
+ }
+ }
+
+ CommitSession commitSession = null;
+
+ if (nonViolators.size() > 0) {
+ // if everything is a violation, then it is expected that
+ // code calling this will not log or commit
+ commitSession = finishPreparingMutations(time);
+ if (commitSession == null)
+ return null;
+ }
+
+ throw new TConstraintViolationException(violations, violators, nonViolators, commitSession);
+ }
+
+ return finishPreparingMutations(time);
+ }
+
+ @Override
+ public synchronized void abortCommit(CommitSession commitSession, List<Mutation> value) {
+ if (writesInProgress <= 0) {
+ throw new IllegalStateException("waitingForLogs <= 0 " + writesInProgress);
+ }
+
+ if (isCloseComplete() || getTabletMemory() == null) {
+ throw new IllegalStateException("aborting commit when tablet is closed");
+ }
+
+ commitSession.decrementCommitsInProgress();
+ writesInProgress--;
+ if (writesInProgress == 0)
+ this.notifyAll();
+ }
+
+ @Override
+ public void commit(CommitSession commitSession, List<Mutation> mutations) {
+
+ int totalCount = 0;
+ long totalBytes = 0;
+
+ // write the mutation to the in memory table
+ for (Mutation mutation : mutations) {
+ totalCount += mutation.size();
+ totalBytes += mutation.numBytes();
+ }
+
+ getTabletMemory().mutate(commitSession, mutations);
+
+ synchronized (this) {
+ if (writesInProgress < 1) {
+ throw new IllegalStateException("commiting mutations after logging, but not waiting for any log messages");
+ }
+
+ if (isCloseComplete()) {
+ throw new IllegalStateException("tablet closed with outstanding messages to the logger");
+ }
+
+ getTabletMemory().updateMemoryUsageStats();
+
+ // decrement here in case an exception is thrown below
+ writesInProgress--;
+ if (writesInProgress == 0)
+ this.notifyAll();
+
+ commitSession.decrementCommitsInProgress();
+
+ numEntries += totalCount;
+ numEntriesInMemory += totalCount;
+ ingestCount += totalCount;
+ ingestBytes += totalBytes;
+ }
+ }
+
+ /**
+ * Closes the mapfiles associated with a Tablet. If saveState is true, a minor compaction is performed.
+ */
+ public void close(boolean saveState) throws IOException {
+ initiateClose(saveState, false, false);
+ completeClose(saveState, true);
+ }
+
+ void initiateClose(boolean saveState, boolean queueMinC, boolean disableWrites) {
+
+ if (!saveState && queueMinC) {
+ throw new IllegalArgumentException("Not saving state on close and requesting minor compactions queue does not make sense");
+ }
+
+ log.debug("initiateClose(saveState=" + saveState + " queueMinC=" + queueMinC + " disableWrites=" + disableWrites + ") " + getExtent());
+
+ MinorCompactionTask mct = null;
+
+ synchronized (this) {
+ if (isClosed() || isClosing()) {
+ String msg = "Tablet " + getExtent() + " already " + closeState;
+ throw new IllegalStateException(msg);
+ }
+
+ // enter the closing state, no splits, minor, or major compactions can start
+ // should cause running major compactions to stop
+ closeState = CloseState.CLOSING;
+ this.notifyAll();
+
+ // determines if inserts and queries can still continue while minor compacting
+ if (disableWrites) {
+ closeState = CloseState.CLOSED;
+ }
+
+ // wait for major compactions to finish, setting closing to
+ // true should cause any running major compactions to abort
+ while (isMajorCompactionRunning()) {
+ try {
+ this.wait(50);
+ } catch (InterruptedException e) {
+ log.error(e.toString());
+ }
+ }
+
+ while (updatingFlushID) {
+ try {
+ this.wait(50);
+ } catch (InterruptedException e) {
+ log.error(e.toString());
+ }
+ }
+
+ if (!saveState || getTabletMemory().getMemTable().getNumEntries() == 0) {
+ return;
+ }
+
+ getTabletMemory().waitForMinC();
+
+ try {
+ mct = prepareForMinC(getFlushID(), MinorCompactionReason.CLOSE);
+ } catch (NoNodeException e) {
+ throw new RuntimeException(e);
+ }
+
+ if (queueMinC) {
+ getTabletResources().executeMinorCompaction(mct);
+ return;
+ }
+
+ }
+
+ // do minor compaction outside of synch block so that tablet can be read and written to while
+ // compaction runs
+ mct.run();
+ }
+
+ private boolean closeCompleting = false;
+
+ synchronized void completeClose(boolean saveState, boolean completeClose) throws IOException {
+
+ if (!isClosing() || isCloseComplete() || closeCompleting) {
+ throw new IllegalStateException("closeState = " + closeState);
+ }
+
+ log.debug("completeClose(saveState=" + saveState + " completeClose=" + completeClose + ") " + getExtent());
+
+ // ensure this method is only called once, also guards against multiple
+ // threads entering the method at the same time
+ closeCompleting = true;
+ closeState = CloseState.CLOSED;
+
+ // modify dataSourceDeletions so scans will try to switch data sources and fail because the tablet is closed
+ dataSourceDeletions.incrementAndGet();
+
+ for (ScanDataSource activeScan : activeScans) {
+ activeScan.interrupt();
+ }
+
+ // wait for reads and writes to complete
+ while (writesInProgress > 0 || activeScans.size() > 0) {
+ try {
+ this.wait(50);
+ } catch (InterruptedException e) {
+ log.error(e.toString());
+ }
+ }
+
+ getTabletMemory().waitForMinC();
+
+ if (saveState && getTabletMemory().getMemTable().getNumEntries() > 0) {
+ try {
+ prepareForMinC(getFlushID(), MinorCompactionReason.CLOSE).run();
+ } catch (NoNodeException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ if (saveState) {
+ // at this point all tablet data is flushed, so do a consistency check
+ RuntimeException err = null;
+ for (int i = 0; i < 5; i++) {
+ try {
+ closeConsistencyCheck();
+ err = null;
+ } catch (RuntimeException t) {
+ err = t;
+ log.error("Consistency check fails, retrying " + t);
+ UtilWaitThread.sleep(500);
+ }
+ }
+ if (err != null) {
+ ProblemReports.getInstance().report(new ProblemReport(extent.getTableId().toString(), ProblemType.TABLET_LOAD, this.extent.toString(), err));
+ log.error("Tablet closed consistency check has failed for " + this.extent + " giving up and closing");
+ }
+ }
+
+ try {
+ getTabletMemory().getMemTable().delete(0);
+ } catch (Throwable t) {
+ log.error("Failed to delete mem table : " + t.getMessage(), t);
+ }
+
+ getTabletMemory().close();
+
+ // close map files
+ getTabletResources().close();
+
+ log.log(TLevel.TABLET_HIST, extent + " closed");
+
+ tableConfiguration.getNamespaceConfiguration().removeObserver(configObserver);
+ tableConfiguration.removeObserver(configObserver);
+
+ if (completeClose)
+ closeState = CloseState.COMPLETE;
+ }
+
+ private void closeConsistencyCheck() {
+
+ if (getTabletMemory().getMemTable().getNumEntries() != 0) {
+ String msg = "Closed tablet " + extent + " has " + getTabletMemory().getMemTable().getNumEntries() + " entries in memory";
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+
+ if (getTabletMemory().memoryReservedForMinC()) {
+ String msg = "Closed tablet " + extent + " has minor compacting memory";
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+
+ try {
+ Pair<List<LogEntry>,SortedMap<FileRef,DataFileValue>> fileLog = MetadataTableUtil.getFileAndLogEntries(SystemCredentials.get(), extent);
+
+ if (fileLog.getFirst().size() != 0) {
+ String msg = "Closed tablet " + extent + " has walog entries in " + MetadataTable.NAME + " " + fileLog.getFirst();
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+
+ if (extent.isRootTablet()) {
+ if (!fileLog.getSecond().keySet().equals(getDatafileManager().getDatafileSizes().keySet())) {
+ String msg = "Data file in " + RootTable.NAME + " differ from in memory data " + extent + " " + fileLog.getSecond().keySet() + " "
+ + getDatafileManager().getDatafileSizes().keySet();
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+ } else {
+ if (!fileLog.getSecond().equals(getDatafileManager().getDatafileSizes())) {
+ String msg = "Data file in " + MetadataTable.NAME + " differ from in memory data " + extent + " " + fileLog.getSecond() + " "
+ + getDatafileManager().getDatafileSizes();
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+ }
+
+ } catch (Exception e) {
+ String msg = "Failed to do close consistency check for tablet " + extent;
+ log.error(msg, e);
+ throw new RuntimeException(msg, e);
+
+ }
+
+ if (otherLogs.size() != 0 || currentLogs.size() != 0) {
+ String msg = "Closed tablet " + extent + " has walog entries in memory currentLogs = " + currentLogs + " otherLogs = " + otherLogs;
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+
+ // TODO check lastFlushID and lostCompactID - ACCUMULO-1290
+ }
+
+ /**
+ * Returns a Path object representing the tablet's location on the DFS.
+ *
+ * @return location
+ */
+ public Path getLocation() {
+ return location;
+ }
+
+ public synchronized boolean initiateMajorCompaction(MajorCompactionReason reason) {
+
+ if (isClosing() || isClosed() || !needsMajorCompaction(reason) || isMajorCompactionRunning() || majorCompactionQueued.contains(reason)) {
+ return false;
+ }
+
+ majorCompactionQueued.add(reason);
+
+ getTabletResources().executeMajorCompaction(getExtent(), new CompactionRunner(this, reason));
+
+ return false;
+ }
+
+ /**
+ * Returns true if a major compaction should be performed on the tablet.
+ *
+ */
+ public boolean needsMajorCompaction(MajorCompactionReason reason) {
+ if (isMajorCompactionRunning())
+ return false;
+ if (reason == MajorCompactionReason.CHOP || reason == MajorCompactionReason.USER)
+ return true;
+ return getTabletResources().needsMajorCompaction(getDatafileManager().getDatafileSizes(), reason);
+ }
+
+ /**
+ * Returns an int representing the total block size of the files served by this tablet.
+ *
+ * @return size
+ */
+ // this is the size of just the files
+ public long estimateTabletSize() {
+ long size = 0L;
+
+ for (DataFileValue sz : getDatafileManager().getDatafileSizes().values())
+ size += sz.getSize();
+
+ return size;
+ }
+
+ private boolean sawBigRow = false;
+ private long timeOfLastMinCWhenBigFreakinRowWasSeen = 0;
+ private long timeOfLastImportWhenBigFreakinRowWasSeen = 0;
+ private long splitCreationTime;
+
+ private SplitRowSpec findSplitRow(Collection<FileRef> files) {
+
+ // never split the root tablet
+ // check if we already decided that we can never split
+ // check to see if we're big enough to split
+
+ long splitThreshold = tableConfiguration.getMemoryInBytes(Property.TABLE_SPLIT_THRESHOLD);
+ long maxEndRow = tableConfiguration.getMemoryInBytes(Property.TABLE_MAX_END_ROW_SIZE);
+
+ if (extent.isRootTablet() || estimateTabletSize() <= splitThreshold) {
+ return null;
+ }
+
+ // have seen a big row before, do not bother checking unless a minor compaction or map file import has occurred.
+ if (sawBigRow) {
+ if (timeOfLastMinCWhenBigFreakinRowWasSeen != lastMinorCompactionFinishTime || timeOfLastImportWhenBigFreakinRowWasSeen != lastMapFileImportTime) {
+ // a minor compaction or map file import has occurred... check again
+ sawBigRow = false;
+ } else {
+ // nothing changed, do not split
+ return null;
+ }
+ }
+
+ SortedMap<Double,Key> keys = null;
+
+ try {
+ // we should make .25 below configurable
+ keys = FileUtil.findMidPoint(getTabletServer().getFileSystem(), getTabletServer().getConfiguration(), extent.getPrevEndRow(), extent.getEndRow(),
+ FileUtil.toPathStrings(files), .25);
+ } catch (IOException e) {
+ log.error("Failed to find midpoint " + e.getMessage());
+ return null;
+ }
+
+ // check to see if one row takes up most of the tablet, in which case we can not split
+ try {
+
+ Text lastRow;
+ if (extent.getEndRow() == null) {
+ Key lastKey = (Key) FileUtil.findLastKey(getTabletServer().getFileSystem(), getTabletServer().getConfiguration(), files);
+ lastRow = lastKey.getRow();
+ } else {
+ lastRow = extent.getEndRow();
+ }
+
+ // We expect to get a midPoint for this set of files. If we don't get one, we have a problem.
+ final Key mid = keys.get(.5);
+ if (null == mid) {
+ throw new IllegalStateException("Could not determine midpoint for files");
+ }
+
+ // check to see that the midPoint is not equal to the end key
+ if (mid.compareRow(lastRow) == 0) {
+ if (keys.firstKey() < .5) {
+ Key candidate = keys.get(keys.firstKey());
+ if (candidate.getLength() > maxEndRow) {
+ log.warn("Cannot split tablet " + extent + ", selected split point too long. Length : " + candidate.getLength());
+
+ sawBigRow = true;
+ timeOfLastMinCWhenBigFreakinRowWasSeen = lastMinorCompactionFinishTime;
+ timeOfLastImportWhenBigFreakinRowWasSeen = lastMapFileImportTime;
+
+ return null;
+ }
+ if (candidate.compareRow(lastRow) != 0) {
+ // we should use this ratio in split size estimations
+ if (log.isTraceEnabled())
+ log.trace(String.format("Splitting at %6.2f instead of .5, row at .5 is same as end row%n", keys.firstKey()));
+ return new SplitRowSpec(keys.firstKey(), candidate.getRow());
+ }
+
+ }
+
+ log.warn("Cannot split tablet " + extent + " it contains a big row : " + lastRow);
+
+ sawBigRow = true;
+ timeOfLastMinCWhenBigFreakinRowWasSeen = lastMinorCompactionFinishTime;
+ timeOfLastImportWhenBigFreakinRowWasSeen = lastMapFileImportTime;
+
+ return null;
+ }
+
+ Text text = mid.getRow();
+ SortedMap<Double,Key> firstHalf = keys.headMap(.5);
+ if (firstHalf.size() > 0) {
+ Text beforeMid = firstHalf.get(firstHalf.lastKey()).getRow();
+ Text shorter = new Text();
+ int trunc = longestCommonLength(text, beforeMid);
+ shorter.set(text.getBytes(), 0, Math.min(text.getLength(), trunc + 1));
+ text = shorter;
+ }
+
+ if (text.getLength() > maxEndRow) {
+ log.warn("Cannot split tablet " + extent + ", selected split point too long. Length : " + text.getLength());
+
+ sawBigRow = true;
+ timeOfLastMinCWhenBigFreakinRowWasSeen = lastMinorCompactionFinishTime;
+ timeOfLastImportWhenBigFreakinRowWasSeen = lastMapFileImportTime;
+
+ return null;
+ }
+
+ return new SplitRowSpec(.5, text);
+ } catch (IOException e) {
+ // don't split now, but check again later
+ log.error("Failed to find lastkey " + e.getMessage());
+ return null;
+ }
+
+ }
+
+ private static int longestCommonLength(Text text, Text beforeMid) {
+ int common = 0;
+ while (common < text.getLength() && common < beforeMid.getLength() && text.getBytes()[common] == beforeMid.getBytes()[common]) {
+ common++;
+ }
+ return common;
+ }
+
+ private Map<FileRef,Pair<Key,Key>> getFirstAndLastKeys(SortedMap<FileRef,DataFileValue> allFiles) throws IOException {
+ Map<FileRef,Pair<Key,Key>> result = new HashMap<FileRef,Pair<Key,Key>>();
+ FileOperations fileFactory = FileOperations.getInstance();
+ VolumeManager fs = getTabletServer().getFileSystem();
+ for (Entry<FileRef,DataFileValue> entry : allFiles.entrySet()) {
+ FileRef file = entry.getKey();
+ FileSystem ns = fs.getVolumeByPath(file.path()).getFileSystem();
+ FileSKVIterator openReader = fileFactory.openReader(file.path().toString(), true, ns, ns.getConf(), this.getTableConfiguration());
+ try {
+ Key first = openReader.getFirstKey();
+ Key last = openReader.getLastKey();
+ result.put(file, new Pair<Key,Key>(first, last));
+ } finally {
+ openReader.close();
+ }
+ }
+ return result;
+ }
+
+ List<FileRef> findChopFiles(KeyExtent extent, Map<FileRef,Pair<Key,Key>> firstAndLastKeys, Collection<FileRef> allFiles) throws IOException {
+ List<FileRef> result = new ArrayList<FileRef>();
+ if (firstAndLastKeys == null) {
+ result.addAll(allFiles);
+ return result;
+ }
+
+ for (FileRef file : allFiles) {
+ Pair<Key,Key> pair = firstAndLastKeys.get(file);
+ if (pair == null) {
+ // file was created or imported after we obtained the first and last keys... there
+ // are a few options here... throw an exception which will cause the compaction to
+ // retry and also cause ugly error message that the admin has to ignore... could
+ // go get the first and last key, but this code is called while the tablet lock
+ // is held... or just compact the file....
+ result.add(file);
+ } else {
+ Key first = pair.getFirst();
+ Key last = pair.getSecond();
+ // If first and last are null, it's an empty file. Add it to the compact set so it goes away.
+ if ((first == null && last == null) || !extent.contains(first.getRow()) || !extent.contains(last.getRow())) {
+ result.add(file);
+ }
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Returns true if this tablet needs to be split
+ *
+ */
+ public synchronized boolean needsSplit() {
+ if (isClosing() || isClosed())
+ return false;
+ return findSplitRow(getDatafileManager().getFiles()) != null;
+ }
+
+ // BEGIN PRIVATE METHODS RELATED TO MAJOR COMPACTION
+
+ private boolean isCompactionEnabled() {
+ return !isClosing() && !getTabletServer().isMajorCompactionDisabled();
+ }
+
+ private CompactionStats _majorCompact(MajorCompactionReason reason) throws IOException, CompactionCanceledException {
+
+ long t1, t2, t3;
+
+ // acquire file info outside of tablet lock
+ CompactionStrategy strategy = Property.createTableInstanceFromPropertyName(tableConfiguration, Property.TABLE_COMPACTION_STRATEGY,
+ CompactionStrategy.class, new DefaultCompactionStrategy());
+ strategy.init(Property.getCompactionStrategyOptions(tableConfiguration));
+
+ Map<FileRef,Pair<Key,Key>> firstAndLastKeys = null;
+ if (reason == MajorCompactionReason.CHOP) {
+ firstAndLastKeys = getFirstAndLastKeys(getDatafileManager().getDatafileSizes());
+ } else if (reason != MajorCompactionReason.USER) {
+ MajorCompactionRequest request = new MajorCompactionRequest(extent, reason, getTabletServer().getFileSystem(), tableConfiguration);
+ request.setFiles(getDatafileManager().getDatafileSizes());
+ strategy.gatherInformation(request);
+ }
+
+ Map<FileRef,DataFileValue> filesToCompact;
+
+ int maxFilesToCompact = tableConfiguration.getCount(Property.TSERV_MAJC_THREAD_MAXOPEN);
+
+ CompactionStats majCStats = new CompactionStats();
+ CompactionPlan plan = null;
+
+ boolean propogateDeletes = false;
+
+ synchronized (this) {
+ // plan all that work that needs to be done in the sync block... then do the actual work
+ // outside the sync block
+
+ t1 = System.currentTimeMillis();
+
+ majorCompactionState = CompactionState.WAITING_TO_START;
+
+ getTabletMemory().waitForMinC();
+
+ t2 = System.currentTimeMillis();
+
+ majorCompactionState = CompactionState.IN_PROGRESS;
+ notifyAll();
+
+ VolumeManager fs = getTabletServer().getFileSystem();
+ if (extent.isRootTablet()) {
+ // very important that we call this before doing major compaction,
+ // otherwise deleted compacted files could possible be brought back
+ // at some point if the file they were compacted to was legitimately
+ // removed by a major compaction
+ RootFiles.cleanupReplacement(fs, fs.listStatus(this.location), false);
+ }
+ SortedMap<FileRef,DataFileValue> allFiles = getDatafileManager().getDatafileSizes();
+ List<FileRef> inputFiles = new ArrayList<FileRef>();
+ if (reason == MajorCompactionReason.CHOP) {
+ // enforce rules: files with keys outside our range need to be compacted
+ inputFiles.addAll(findChopFiles(extent, firstAndLastKeys, allFiles.keySet()));
+ } else if (reason == MajorCompactionReason.USER) {
+ inputFiles.addAll(allFiles.keySet());
+ } else {
+ MajorCompactionRequest request = new MajorCompactionRequest(extent, reason, fs, tableConfiguration);
+ request.setFiles(allFiles);
+ plan = strategy.getCompactionPlan(request);
+ if (plan != null) {
+ plan.validate(allFiles.keySet());
+ inputFiles.addAll(plan.inputFiles);
+ }
+ }
+
+ if (inputFiles.isEmpty()) {
+ return majCStats;
+ }
+ // If no original files will exist at the end of the compaction, we do not have to propogate deletes
+ Set<FileRef> droppedFiles = new HashSet<FileRef>();
+ droppedFiles.addAll(inputFiles);
+ if (plan != null)
+ droppedFiles.addAll(plan.deleteFiles);
+ propogateDeletes = !(droppedFiles.equals(allFiles.keySet()));
+ log.debug("Major compaction plan: " + plan + " propogate deletes : " + propogateDeletes);
+ filesToCompact = new HashMap<FileRef,DataFileValue>(allFiles);
+ filesToCompact.keySet().retainAll(inputFiles);
+
+ t3 = System.currentTimeMillis();
+
+ getDatafileManager().reserveMajorCompactingFiles(filesToCompact.keySet());
+ }
+
+ try {
+
+ log.debug(String.format("MajC initiate lock %.2f secs, wait %.2f secs", (t3 - t2) / 1000.0, (t2 - t1) / 1000.0));
+
+ Pair<Long,List<IteratorSetting>> compactionId = null;
+ if (!propogateDeletes) {
+ // compacting everything, so update the compaction id in metadata
+ try {
+ compactionId = getCompactionID();
+ } catch (NoNodeException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ List<IteratorSetting> compactionIterators = new ArrayList<IteratorSetting>();
+ if (compactionId != null) {
+ if (reason == MajorCompactionReason.USER) {
+ if (getCompactionCancelID() >= compactionId.getFirst()) {
+ // compaction was canceled
+ return majCStats;
+ }
+
+ synchronized (this) {
+ if (lastCompactID >= compactionId.getFirst())
+ // already compacted
+ return majCStats;
+ }
+ }
+
+ compactionIterators = compactionId.getSecond();
+ }
+
+ // need to handle case where only one file is being major compacted
+ while (filesToCompact.size() > 0) {
+
+ int numToCompact = maxFilesToCompact;
+
+ if (filesToCompact.size() > maxFilesToCompact && filesToCompact.size() < 2 * maxFilesToCompact) {
+ // on the second to last compaction pass, compact the minimum amount of files possible
+ numToCompact = filesToCompact.size() - maxFilesToCompact + 1;
+ }
+
+ Set<FileRef> smallestFiles = removeSmallest(filesToCompact, numToCompact);
+
+ FileRef fileName = getNextMapFilename((filesToCompact.size() == 0 && !propogateDeletes) ? "A" : "C");
+ FileRef compactTmpName = new FileRef(fileName.path().toString() + "_tmp");
+
+ AccumuloConfiguration tableConf = createTableConfiguration(tableConfiguration, plan);
+
+ Span span = Trace.start("compactFiles");
+ try {
+
+ CompactionEnv cenv = new CompactionEnv() {
+ @Override
+ public boolean isCompactionEnabled() {
+ return Tablet.this.isCompactionEnabled();
+ }
+
+ @Override
+ public IteratorScope getIteratorScope() {
+ return IteratorScope.majc;
+ }
+ };
+
+ HashMap<FileRef,DataFileValue> copy = new HashMap<FileRef,DataFileValue>(getDatafileManager().getDatafileSizes());
+ if (!copy.keySet().containsAll(smallestFiles))
+ throw new IllegalStateException("Cannot find data file values for " + smallestFiles);
+
+ copy.keySet().retainAll(smallestFiles);
+
+ log.debug("Starting MajC " + extent + " (" + reason + ") " + copy.keySet() + " --> " + compactTmpName + " " + compactionIterators);
+
+ // always propagate deletes, unless last batch
+ boolean lastBatch = filesToCompact.isEmpty();
+ Compactor compactor = new Compactor(this, copy, null, compactTmpName, lastBatch ? propogateDeletes : true, cenv, compactionIterators,
+ reason.ordinal(), tableConf);
+
+ CompactionStats mcs = compactor.call();
+
+ span.data("files", "" + smallestFiles.size());
+ span.data("read", "" + mcs.getEntriesRead());
+ span.data("written", "" + mcs.getEntriesWritten());
+ majCStats.add(mcs);
+
+ if (lastBatch && plan != null && plan.deleteFiles != null) {
+ smallestFiles.addAll(plan.deleteFiles);
+ }
+ getDatafileManager().bringMajorCompactionOnline(smallestFiles, compactTmpName, fileName,
+ filesToCompact.size() == 0 && compactionId != null ? compactionId.getFirst() : null,
+ new DataFileValue(mcs.getFileSize(), mcs.getEntriesWritten()));
+
+ // when major compaction produces a file w/ zero entries, it will be deleted... do not want
+ // to add the deleted file
+ if (filesToCompact.size() > 0 && mcs.getEntriesWritten() > 0) {
+ filesToCompact.put(fileName, new DataFileValue(mcs.getFileSize(), mcs.getEntriesWritten()));
+ }
+ } finally {
+ span.stop();
+ }
+
+ }
+ return majCStats;
+ } finally {
+ synchronized (Tablet.this) {
+ getDatafileManager().clearMajorCompactingFile();
+ }
+ }
+ }
+
+ protected AccumuloConfiguration createTableConfiguration(TableConfiguration base, CompactionPlan plan) {
+ if (plan == null || plan.writeParameters == null)
+ return base;
+ WriteParameters p = plan.writeParameters;
+ ConfigurationCopy result = new ConfigurationCopy(base);
+ if (p.getHdfsBlockSize() > 0)
+ result.set(Property.TABLE_FILE_BLOCK_SIZE, "" + p.getHdfsBlockSize());
+ if (p.getBlockSize() > 0)
+ result.set(Property.TABLE_FILE_COMPRESSED_BLOCK_SIZE, "" + p.getBlockSize());
+ if (p.getIndexBlockSize() > 0)
+ result.set(Property.TABLE_FILE_COMPRESSED_BLOCK_SIZE_INDEX, "" + p.getIndexBlockSize());
+ if (p.getCompressType() != null)
+ result.set(Property.TABLE_FILE_COMPRESSION_TYPE, p.getCompressType());
+ if (p.getReplication() != 0)
+ result.set(Property.TABLE_FILE_REPLICATION, "" + p.getReplication());
+ return result;
+ }
+
+ private Set<FileRef> removeSmallest(Map<FileRef,DataFileValue> filesToCompact, int maxFilesToCompact) {
+ // ensure this method works properly when multiple files have the same size
+
+ PriorityQueue<Pair<FileRef,Long>> fileHeap = new PriorityQueue<Pair<FileRef,Long>>(filesToCompact.size(), new Comparator<Pair<FileRef,Long>>() {
+ @Override
+ public int compare(Pair<FileRef,Long> o1, Pair<FileRef,Long> o2) {
+ if (o1.getSecond() == o2.getSecond())
+ return o1.getFirst().compareTo(o2.getFirst());
+ if (o1.getSecond() < o2.getSecond())
+ return -1;
+ return 1;
+ }
+ });
+
+ for (Iterator<Entry<FileRef,DataFileValue>> iterator = filesToCompact.entrySet().iterator(); iterator.hasNext();) {
+ Entry<FileRef,DataFileValue> entry = iterator.next();
+ fileHeap.add(new Pair<FileRef,Long>(entry.getKey(), entry.getValue().getSize()));
+ }
+
+ Set<FileRef> smallestFiles = new HashSet<FileRef>();
+ while (smallestFiles.size() < maxFilesToCompact && fileHeap.size() > 0) {
+ Pair<FileRef,Long> pair = fileHeap.remove();
+ filesToCompact.remove(pair.getFirst());
+ smallestFiles.add(pair.getFirst());
+ }
+
+ return smallestFiles;
+ }
+
+ // END PRIVATE METHODS RELATED TO MAJOR COMPACTION
+
+ /**
+ * Performs a major compaction on the tablet. If needsSplit() returns true, the tablet is split and a reference to the new tablet is returned.
+ */
+
+ CompactionStats majorCompact(MajorCompactionReason reason, long queued) {
+
+ CompactionStats majCStats = null;
+ boolean success = false;
+ long start = System.currentTimeMillis();
+
+ // Always trace majC
+ Span span = Trace.on("majorCompaction");
+
+ try {
+ timer.incrementStatusMajor();
+
+ synchronized (this) {
+ // check that compaction is still needed - defer to splitting
+ majorCompactionQueued.remove(reason);
+
+ if (isClosing() || isClosed() || !needsMajorCompaction(reason) || isMajorCompactionRunning() || needsSplit()) {
+ return null;
+ }
+
+ majorCompactionState = CompactionState.WAITING_TO_START;
+ }
+
+ try {
+ majCStats = _majorCompact(reason);
+ if (reason == MajorCompactionReason.CHOP) {
+ MetadataTableUtil.chopped(getExtent(), this.getTabletServer().getLock());
+ getTabletServer().enqueueMasterMessage(new TabletStatusMessage(TabletLoadState.CHOPPED, extent));
+ }
+ success = true;
+ } catch (CompactionCanceledException mcce) {
+ log.debug("Major compaction canceled, extent = " + getExtent());
+ } catch (Throwable t) {
+ log.error("MajC Failed, extent = " + getExtent());
+ log.error("MajC Failed, message = " + (t.getMessage() == null ? t.getClass().getName() : t.getMessage()), t);
+ } finally {
+ // ensure we always reset boolean, even
+ // when an exception is thrown
+ synchronized (this) {
+ majorCompactionState = null;
+ this.notifyAll();
+ }
+
+ Span curr = Trace.currentTrace();
+ curr.data("extent", "" + getExtent());
+ if (majCStats != null) {
+ curr.data("read", "" + majCStats.getEntriesRead());
+ curr.data("written", "" + majCStats.getEntriesWritten());
+ }
+ }
+ } finally {
+ span.stop();
+ long count = 0;
+ if (majCStats != null)
+ count = majCStats.getEntriesRead();
+ timer.updateTime(Operation.MAJOR, queued, start, count, !success);
+ }
+
+ return majCStats;
+ }
+
+ /**
+ * Returns a KeyExtent object representing this tablet's key range.
+ *
+ * @return extent
+ */
+ @Override
+ public KeyExtent getExtent() {
+ return extent;
+ }
+
+ synchronized void computeNumEntries() {
+ Collection<DataFileValue> vals = getDatafileManager().getDatafileSizes().values();
+
+ long numEntries = 0;
+
+ for (DataFileValue tableValue : vals) {
+ numEntries += tableValue.getNumEntries();
+ }
+
+ this.numEntriesInMemory = getTabletMemory().getNumEntries();
+ numEntries += getTabletMemory().getNumEntries();
+
+ this.numEntries = numEntries;
+ }
+
+ public long getNumEntries() {
+ return numEntries;
+ }
+
+ public long getNumEntriesInMemory() {
+ return numEntriesInMemory;
+ }
+
+ public synchronized boolean isClosing() {
+ return closeState == CloseState.CLOSING;
+ }
+
+ public synchronized boolean isClosed() {
+ return closeState == CloseState.CLOSED || closeState == CloseState.COMPLETE;
+ }
+
+ public synchronized boolean isCloseComplete() {
+ return closeState == CloseState.COMPLETE;
+ }
+
+ public boolean isMajorCompactionRunning() {
+ return majorCompactionState != null;
+ }
+
+ public boolean isMinorCompactionQueued() {
+ return minorCompactionState == CompactionState.WAITING_TO_START;
+ }
+
+ public boolean isMinorCompactionRunning() {
+ return minorCompactionState == CompactionState.IN_PROGRESS;
+ }
+
+ public boolean isMajorCompactionQueued() {
+ return majorCompactionQueued.size() > 0;
+ }
+
+ public TreeMap<KeyExtent,SplitInfo> split(byte[] sp) throws IOException {
+
+ if (sp != null && extent.getEndRow() != null && extent.getEndRow().equals(new Text(sp))) {
+ throw new IllegalArgumentException();
+ }
+
+ if (sp != null && sp.length > tableConfiguration.getMemoryInBytes(Property.TABLE_MAX_END_ROW_SIZE)) {
+ String msg = "Cannot split tablet " + extent + ", selected split point too long. Length : " + sp.length;
+ log.warn(msg);
+ throw new IOException(msg);
+ }
+
+ if (extent.isRootTablet()) {
+ String msg = "Cannot split root tablet";
+ log.warn(msg);
+ throw new RuntimeException(msg);
+ }
+
+ try {
+ initiateClose(true, false, false);
+ } catch (IllegalStateException ise) {
+ log.debug("File " + extent + " not splitting : " + ise.getMessage());
+ return null;
+ }
+
+ // obtain this info outside of synch block since it will involve opening
+ // the map files... it is ok if the set of map files changes, because
+ // this info is used for optimization... it is ok if map files are missing
+ // from the set... can still query and insert into the tablet while this
+ // map file operation is happening
+ Map<FileRef,FileUtil.FileInfo> firstAndLastRows = FileUtil.tryToGetFirstAndLastRows(getTabletServer().getFileSystem(),
+ getTabletServer().getConfiguration(), getDatafileManager().getFiles());
+
+ synchronized (this) {
+ // java needs tuples ...
+ TreeMap<KeyExtent,SplitInfo> newTablets = new TreeMap<KeyExtent,SplitInfo>();
+
+ long t1 = System.currentTimeMillis();
+ // choose a split point
+ SplitRowSpec splitPoint;
+ if (sp == null)
+ splitPoint = findSplitRow(getDatafileManager().getFiles());
+ else {
+ Text tsp = new Text(sp);
+ splitPoint = new SplitRowSpec(FileUtil.estimatePercentageLTE(getTabletServer().getFileSystem(), getTabletServer().getConfiguration(),
+ extent.getPrevEndRow(), extent.getEndRow(), FileUtil.toPathStrings(getDatafileManager().getFiles()), tsp), tsp);
+ }
+
+ if (splitPoint == null || splitPoint.row == null) {
+ log.info("had to abort split because splitRow was null");
+ closeState = CloseState.OPEN;
+ return null;
+ }
+
+ closeState = CloseState.CLOSING;
+ completeClose(true, false);
+
+ Text midRow = splitPoint.row;
+ double splitRatio = splitPoint.splitRatio;
+
+ KeyExtent low = new KeyExtent(extent.getTableId(), midRow, extent.getPrevEndRow());
+ KeyExtent high = new KeyExtent(extent.getTableId(), extent.getEndRow(), midRow);
+
+ String lowDirectory = TabletOperations.createTabletDirectory(getTabletServer().getFileSystem(), extent.getTableId().toString(), midRow);
+
+ // write new tablet information to MetadataTable
+ SortedMap<FileRef,DataFileValue> lowDatafileSizes = new TreeMap<FileRef,DataFileValue>();
+ SortedMap<FileRef,DataFileValue> highDatafileSizes = new TreeMap<FileRef,DataFileValue>();
+ List<FileRef> highDatafilesToRemove = new ArrayList<FileRef>();
+
+ MetadataTableUtil.splitDatafiles(extent.getTableId(), midRow, splitRatio, firstAndLastRows, getDatafileManager().getDatafileSizes(), lowDatafileSizes,
+ highDatafileSizes, highDatafilesToRemove);
+
+ log.debug("Files for low split " + low + " " + lowDatafileSizes.keySet());
+ log.debug("Files for high split " + high + " " + highDatafileSizes.keySet());
+
+ String time = tabletTime.getMetadataValue();
+
+ // it is possible that some of the bulk loading flags will be deleted after being read below because the bulk load
+ // finishes.... therefore split could propagate load flags for a finished bulk load... there is a special iterator
+ // on the metadata table to clean up this type of garbage
+ Map<FileRef,Long> bulkLoadedFiles = MetadataTableUtil.getBulkFilesLoaded(SystemCredentials.get(), extent);
+
+ MetadataTableUtil.sp
<TRUNCATED>
[2/4] git commit: ACCUMULO-3215 Ensure tablet directories are
absolute, not relative
Posted by el...@apache.org.
ACCUMULO-3215 Ensure tablet directories are absolute, not relative
Move 'c-' into a constant, make the computed srv:dir in ImportTable
an absolute directory (choosing a volume as per policy for each tablet),
and add a test to make sure the path is constructed correctly.
ImportExportIT was extended to check that srv:dir, in addition to file:,
is an absolute path.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/6ad16a9c
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/6ad16a9c
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/6ad16a9c
Branch: refs/heads/master
Commit: 6ad16a9c50f48a98def23e5720218659a410dd08
Parents: 0348e3e
Author: Josh Elser <el...@apache.org>
Authored: Wed Oct 15 15:14:25 2014 -0400
Committer: Josh Elser <el...@apache.org>
Committed: Wed Oct 15 19:31:29 2014 -0400
----------------------------------------------------------------------
.../org/apache/accumulo/core/Constants.java | 5 +-
.../accumulo/server/util/MetadataTableUtil.java | 2 +-
.../accumulo/gc/SimpleGarbageCollector.java | 2 +-
server/master/pom.xml | 5 ++
.../accumulo/master/tableOps/ImportTable.java | 34 ++++++++++--
.../master/tableOps/ImportTableTest.java | 54 ++++++++++++++++++++
.../org/apache/accumulo/tserver/Tablet.java | 46 ++++++++---------
.../apache/accumulo/test/ImportExportIT.java | 35 ++++++++++---
8 files changed, 147 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6ad16a9c/core/src/main/java/org/apache/accumulo/core/Constants.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/Constants.java b/core/src/main/java/org/apache/accumulo/core/Constants.java
index 15cccf4..2a570f4 100644
--- a/core/src/main/java/org/apache/accumulo/core/Constants.java
+++ b/core/src/main/java/org/apache/accumulo/core/Constants.java
@@ -56,7 +56,7 @@ public class Constants {
public static final String ZMONITOR_LOCK = ZMONITOR + "/lock";
public static final String ZMONITOR_HTTP_ADDR = ZMONITOR + "/http_addr";
public static final String ZMONITOR_LOG4J_ADDR = ZMONITOR + "/log4j_addr";
-
+
public static final String ZCONFIG = "/config";
public static final String ZTSERVERS = "/tservers";
@@ -88,6 +88,9 @@ public class Constants {
public static final String BULK_PREFIX = "b-";
+ public static final String CLONE_PREFIX = "c-";
+ public static final byte[] CLONE_PREFIX_BYTES = CLONE_PREFIX.getBytes(UTF8);
+
// this affects the table client caching of metadata
public static final int SCAN_BATCH_SIZE = 1000;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6ad16a9c/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java b/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
index f978439..f908d25 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
@@ -881,7 +881,7 @@ public class MetadataTableUtil {
Mutation m = new Mutation(k.getRow());
m.putDelete(k.getColumnFamily(), k.getColumnQualifier());
String dir = volumeManager.choose(ServerConstants.getTablesDirs()) + "/" + tableId
- + new String(FastFormat.toZeroPaddedString(dirCount++, 8, 16, "/c-".getBytes(Constants.UTF8)));
+ + new String(FastFormat.toZeroPaddedString(dirCount++, 8, 16, Constants.CLONE_PREFIX_BYTES));
TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(dir.getBytes(Constants.UTF8)));
bw.addMutation(m);
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6ad16a9c/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
----------------------------------------------------------------------
diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java b/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
index 4617299..d04303b 100644
--- a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
@@ -421,7 +421,7 @@ public class SimpleGarbageCollector implements Iface {
TableState tableState = TableManager.getInstance().getTableState(tableId);
if (tableState != null && tableState != TableState.DELETING) {
// clone directories don't always exist
- if (!tabletDir.startsWith("c-"))
+ if (!tabletDir.startsWith(Constants.CLONE_PREFIX))
log.debug("File doesn't exist: " + fullPath);
}
} else {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6ad16a9c/server/master/pom.xml
----------------------------------------------------------------------
diff --git a/server/master/pom.xml b/server/master/pom.xml
index bfd6b68..0268680 100644
--- a/server/master/pom.xml
+++ b/server/master/pom.xml
@@ -77,6 +77,11 @@
<scope>test</scope>
</dependency>
<dependency>
+ <groupId>org.easymock</groupId>
+ <artifactId>easymock</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<scope>test</scope>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6ad16a9c/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java b/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
index e57c689..8de6bce 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
@@ -222,7 +222,9 @@ class PopulateMetadataTable extends MasterRepo {
log.info("importDir is " + tableInfo.importDir);
// This is a directory already prefixed with proper volume information e.g. hdfs://localhost:8020/path/to/accumulo/tables/...
- String bulkDir = tableInfo.importDir;
+ final String bulkDir = tableInfo.importDir;
+
+ final String[] tableDirs = ServerConstants.getTablesDirs();
ZipEntry zipEntry;
while ((zipEntry = zis.getNextEntry()) != null) {
@@ -260,15 +262,28 @@ class PopulateMetadataTable extends MasterRepo {
}
if (m == null) {
+ // Make a unique directory inside the table's dir. Cannot import multiple tables into one table, so don't need to use unique allocator
+ String tabletDir = new String(FastFormat.toZeroPaddedString(dirCount++, 8, 16, Constants.CLONE_PREFIX_BYTES), Constants.UTF8);
+
+ // Build up a full hdfs://localhost:8020/accumulo/tables/$id/c-XXXXXXX
+ String absolutePath = getClonedTabletDir(master, tableDirs, tabletDir);
+
m = new Mutation(metadataRow);
- TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(FastFormat.toZeroPaddedString(dirCount++, 8, 16, "/c-".getBytes(Constants.UTF8))));
+ TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(absolutePath.getBytes(Constants.UTF8)));
currentRow = metadataRow;
}
if (!currentRow.equals(metadataRow)) {
mbw.addMutation(m);
+
+ // Make a unique directory inside the table's dir. Cannot import multiple tables into one table, so don't need to use unique allocator
+ String tabletDir = new String(FastFormat.toZeroPaddedString(dirCount++, 8, 16, Constants.CLONE_PREFIX_BYTES), Constants.UTF8);
+
+ // Build up a full hdfs://localhost:8020/accumulo/tables/$id/c-XXXXXXX
+ String absolutePath = getClonedTabletDir(master, tableDirs, tabletDir);
+
m = new Mutation(metadataRow);
- TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(FastFormat.toZeroPaddedString(dirCount++, 8, 16, "/c-".getBytes(Constants.UTF8))));
+ TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(absolutePath.getBytes(Constants.UTF8)));
}
m.put(key.getColumnFamily(), cq, val);
@@ -303,6 +318,19 @@ class PopulateMetadataTable extends MasterRepo {
}
}
+ /**
+ * Given options for tables (across multiple volumes), construct an absolute path using the unique name within the chosen volume
+ *
+ * @return An absolute, unique path for the imported table
+ */
+ protected String getClonedTabletDir(Master master, String[] tableDirs, String tabletDir) {
+ // We can try to spread out the tablet dirs across all volumes
+ String tableDir = master.getFileSystem().choose(tableDirs);
+
+ // Build up a full hdfs://localhost:8020/accumulo/tables/$id/c-XXXXXXX
+ return tableDir + "/" + tableInfo.tableId + "/" + tabletDir;
+ }
+
@Override
public void undo(long tid, Master environment) throws Exception {
MetadataTableUtil.deleteTable(tableInfo.tableId, false, SystemCredentials.get(), environment.getMasterLock());
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6ad16a9c/server/master/src/test/java/org/apache/accumulo/master/tableOps/ImportTableTest.java
----------------------------------------------------------------------
diff --git a/server/master/src/test/java/org/apache/accumulo/master/tableOps/ImportTableTest.java b/server/master/src/test/java/org/apache/accumulo/master/tableOps/ImportTableTest.java
new file mode 100644
index 0000000..31f6bde
--- /dev/null
+++ b/server/master/src/test/java/org/apache/accumulo/master/tableOps/ImportTableTest.java
@@ -0,0 +1,54 @@
+/*
+ * 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.
+ */
+package org.apache.accumulo.master.tableOps;
+
+import org.apache.accumulo.master.Master;
+import org.apache.accumulo.server.fs.VolumeManager;
+import org.easymock.EasyMock;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ *
+ */
+public class ImportTableTest {
+
+ @Test
+ public void testTabletDir() {
+ Master master = EasyMock.createMock(Master.class);
+ VolumeManager volumeManager = EasyMock.createMock(VolumeManager.class);
+ ImportedTableInfo iti = new ImportedTableInfo();
+ iti.tableId = "5";
+
+ // Different volumes with different paths
+ String[] tableDirs = new String[] {"hdfs://nn1:8020/apps/accumulo1/tables", "hdfs://nn2:8020/applications/accumulo/tables"};
+ // This needs to be unique WRT the importtable command
+ String tabletDir = "/c-00000001";
+
+ EasyMock.expect(master.getFileSystem()).andReturn(volumeManager);
+ // Choose the 2nd element
+ EasyMock.expect(volumeManager.choose(tableDirs)).andReturn(tableDirs[1]);
+
+ EasyMock.replay(master, volumeManager);
+
+ PopulateMetadataTable pmt = new PopulateMetadataTable(iti);
+ Assert.assertEquals(tableDirs[1] + "/" + iti.tableId + "/" + tabletDir, pmt.getClonedTabletDir(master, tableDirs, tabletDir));
+
+ EasyMock.verify(master, volumeManager);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6ad16a9c/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java
index 53469d7..de930aa 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java
@@ -157,16 +157,16 @@ import com.google.common.annotations.VisibleForTesting;
* We need to be able to have the master tell a tabletServer to
* close this file, and the tablet server to handle all pending client reads
* before closing
- *
+ *
*/
/**
- *
+ *
* this class just provides an interface to read from a MapFile mostly takes care of reporting start and end keys
- *
+ *
* need this because a single row extent can have multiple columns this manages all the columns (each handled by a store) for a single row-extent
- *
- *
+ *
+ *
*/
public class Tablet {
@@ -422,7 +422,7 @@ public class Tablet {
private long lastFlushID = -1;
private long lastCompactID = -1;
-
+
private static class CompactionWaitInfo {
long flushID = -1;
long compactionID = -1;
@@ -523,7 +523,7 @@ public class Tablet {
}
if (files == null) {
- if (tabletDir.getName().startsWith("c-"))
+ if (tabletDir.getName().startsWith(Constants.CLONE_PREFIX))
log.debug("Tablet " + extent + " had no dir, creating " + tabletDir); // its a clone dir...
else
log.warn("Tablet " + extent + " had no dir, creating " + tabletDir);
@@ -833,7 +833,7 @@ public class Tablet {
log.warn("Target map file already exist " + newDatafile);
fs.deleteRecursively(newDatafile.path());
}
-
+
rename(fs, tmpDatafile.path(), newDatafile.path());
}
break;
@@ -966,7 +966,7 @@ public class Tablet {
// rename before putting in metadata table, so files in metadata table should
// always exist
rename(fs, tmpDatafile.path(), newDatafile.path());
-
+
if (dfv.getNumEntries() == 0) {
fs.deleteRecursively(newDatafile.path());
}
@@ -1081,7 +1081,7 @@ public class Tablet {
this.tabletDirectory = tabletDirectory;
this.logId = logId;
this.location = location;
- this.datafileManager = datafileManager;
+ this.datafileManager = datafileManager;
}
private Tablet(TabletServer tabletServer, Text location, KeyExtent extent, TabletResourceManager trm, Configuration conf,
@@ -1356,9 +1356,9 @@ public class Tablet {
}
});
-
+
acuTableConf.getNamespaceConfiguration().addObserver(configObserver);
-
+
// Force a load of any per-table properties
configObserver.propertiesChanged();
@@ -1706,7 +1706,7 @@ public class Tablet {
/**
* Determine if a JVM shutdown is in progress.
- *
+ *
*/
private boolean shutdownInProgress() {
try {
@@ -2044,7 +2044,7 @@ public class Tablet {
public void setInterruptFlag(AtomicBoolean flag) {
throw new UnsupportedOperationException();
}
-
+
}
private DataFileValue minorCompact(Configuration conf, VolumeManager fs, InMemoryMap memTable, FileRef tmpDatafile, FileRef newDatafile, FileRef mergeFile,
@@ -2768,7 +2768,7 @@ public class Tablet {
/**
* Returns a Path object representing the tablet's location on the DFS.
- *
+ *
* @return location
*/
public Path getLocation() {
@@ -2863,7 +2863,7 @@ public class Tablet {
/**
* Returns true if a major compaction should be performed on the tablet.
- *
+ *
*/
public boolean needsMajorCompaction(MajorCompactionReason reason) {
if (majorCompactionInProgress)
@@ -2875,7 +2875,7 @@ public class Tablet {
/**
* Returns an int representing the total block size of the mapfiles served by this tablet.
- *
+ *
* @return size
*/
// this is the size of just the mapfiles
@@ -3046,7 +3046,7 @@ public class Tablet {
/**
* Returns true if this tablet needs to be split
- *
+ *
*/
public synchronized boolean needsSplit() {
boolean ret;
@@ -3304,7 +3304,7 @@ public class Tablet {
return smallestFiles;
}
-
+
static void rename(VolumeManager fs, Path src, Path dst) throws IOException {
if (!fs.rename(src, dst)) {
throw new IOException("Rename " + src + " to " + dst + " returned false ");
@@ -3373,7 +3373,7 @@ public class Tablet {
/**
* Returns a KeyExtent object representing this tablet's key range.
- *
+ *
* @return extent
*/
public KeyExtent getExtent() {
@@ -3433,12 +3433,12 @@ public class Tablet {
/**
* operations are disallowed while we split which is ok since splitting is fast
- *
+ *
* a minor compaction should have taken place before calling this so there should be relatively little left to compact
- *
+ *
* we just need to make sure major compactions aren't occurring if we have the major compactor thread decide who needs splitting we can avoid synchronization
* issues with major compactions
- *
+ *
*/
static class SplitInfo {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/6ad16a9c/test/src/test/java/org/apache/accumulo/test/ImportExportIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/ImportExportIT.java b/test/src/test/java/org/apache/accumulo/test/ImportExportIT.java
index c7e0f0e..a48ed9d 100644
--- a/test/src/test/java/org/apache/accumulo/test/ImportExportIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/ImportExportIT.java
@@ -42,7 +42,6 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.collect.Iterables;
import com.google.common.io.Files;
/**
@@ -57,8 +56,13 @@ public class ImportExportIT extends SimpleMacIT {
private static final Logger log = LoggerFactory.getLogger(ImportExportIT.class);
+ @Override
+ protected int defaultTimeoutSeconds() {
+ return 60;
+ }
+
@Test
- public void test() throws Exception {
+ public void testExportImportThenScan() throws Exception {
Connector conn = getConnector();
String[] tableNames = getUniqueNames(2);
@@ -128,19 +132,34 @@ public class ImportExportIT extends SimpleMacIT {
Scanner s = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
s.setRange(MetadataSchema.TabletsSection.getRange(tableId));
s.fetchColumnFamily(MetadataSchema.TabletsSection.DataFileColumnFamily.NAME);
+ MetadataSchema.TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.fetch(s);
// Should find a single entry
- Entry<Key,Value> fileEntry = Iterables.getOnlyElement(s);
-
- // The file should be an absolute URI (file:///...), not a relative path (/b-000.../I000001.rf)
- String fileUri = fileEntry.getKey().getColumnQualifier().toString();
- Assert.assertFalse("Imported files should have absolute URIs, not relative: " + fileUri, looksLikeRelativePath(fileUri));
+ for (Entry<Key,Value> fileEntry : s) {
+ Key k = fileEntry.getKey();
+ String value = fileEntry.getValue().toString();
+ if (k.getColumnFamily().equals(MetadataSchema.TabletsSection.DataFileColumnFamily.NAME)) {
+ // The file should be an absolute URI (file:///...), not a relative path (/b-000.../I000001.rf)
+ String fileUri = k.getColumnQualifier().toString();
+ Assert.assertFalse("Imported files should have absolute URIs, not relative: " + fileUri, looksLikeRelativePath(fileUri));
+ } else if (k.getColumnFamily().equals(MetadataSchema.TabletsSection.ServerColumnFamily.NAME)) {
+ Assert.assertFalse("Server directory should have absolute URI, not relative: " + value, looksLikeRelativePath(value));
+ } else {
+ Assert.fail("Got expected pair: " + k + "=" + fileEntry.getValue());
+ }
+ }
// Online the original table before we verify equivalence
conn.tableOperations().online(srcTable, true);
+ verifyTableEquality(conn, srcTable, destTable);
+ }
+
+ private void verifyTableEquality(Connector conn, String srcTable, String destTable) throws Exception {
Iterator<Entry<Key,Value>> src = conn.createScanner(srcTable, Authorizations.EMPTY).iterator(), dest = conn.createScanner(destTable, Authorizations.EMPTY)
.iterator();
+ Assert.assertTrue("Could not read any data from source table", src.hasNext());
+ Assert.assertTrue("Could not read any data from destination table", dest.hasNext());
while (src.hasNext() && dest.hasNext()) {
Entry<Key,Value> orig = src.next(), copy = dest.next();
Assert.assertEquals(orig.getKey(), copy.getKey());
@@ -155,6 +174,8 @@ public class ImportExportIT extends SimpleMacIT {
if ('/' == uri.charAt(10)) {
return true;
}
+ } else if (uri.startsWith("/" + Constants.CLONE_PREFIX)) {
+ return true;
}
return false;