Google Git
Sign in
apache / accumulo / ed97eac76da47f711f8f901810cc9d4d8fb27214 / . / server / tserver / src / main / java / org / apache / accumulo / tserver / tablet / Tablet.java
blob: 30bac63c270701af8ba4bb993cbbacdf8ad29c27 [file] [log] [blame]
/*
* 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 static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Objects.requireNonNull;
import static org.apache.accumulo.fate.util.UtilWaitThread.sleepUninterruptibly;
import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
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.Callable;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
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.admin.CompactionStrategyConfig;
import org.apache.accumulo.core.client.impl.DurabilityImpl;
import org.apache.accumulo.core.client.impl.Tables;
import org.apache.accumulo.core.client.sample.SamplerConfiguration;
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.Mutation;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.data.impl.KeyExtent;
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.YieldCallback;
import org.apache.accumulo.core.iterators.YieldingKeyValueIterator;
import org.apache.accumulo.core.iterators.system.SourceSwitchingIterator;
import org.apache.accumulo.core.master.thrift.BulkImportState;
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.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationConfigurationUtil;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.ColumnVisibility;
import org.apache.accumulo.core.tabletserver.log.LogEntry;
import org.apache.accumulo.core.tabletserver.thrift.TabletStats;
import org.apache.accumulo.core.trace.ProbabilitySampler;
import org.apache.accumulo.core.trace.Span;
import org.apache.accumulo.core.trace.Trace;
import org.apache.accumulo.core.util.LocalityGroupUtil;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.core.util.ratelimit.RateLimiter;
import org.apache.accumulo.server.ServerConstants;
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.UserCompactionConfig;
import org.apache.accumulo.server.metrics.Metrics;
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.replication.StatusUtil;
import org.apache.accumulo.server.replication.proto.Replication.Status;
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.zookeeper.ZooReaderWriter;
import org.apache.accumulo.start.classloader.vfs.AccumuloVFSClassLoader;
import org.apache.accumulo.tserver.ConditionCheckerContext.ConditionChecker;
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.metrics.TabletServerScanMetrics;
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 org.gaul.modernizer_maven_annotations.SuppressModernizer;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.ImmutableSet.Builder;
/**
*
* 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);
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 = null;
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<>();
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 final 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<>();
private int writesInProgress = 0;
private final TabletStatsKeeper timer = new TabletStatsKeeper();
private final Rate queryRate = new Rate(0.95);
private long queryCount = 0;
private final Rate queryByteRate = new Rate(0.95);
private long queryBytes = 0;
private final Rate ingestRate = new Rate(0.95);
private long ingestCount = 0;
private final Rate ingestByteRate = new Rate(0.95);
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.95);
private final AtomicLong scannedCount = new AtomicLong(0);
private final ConfigurationObserver configObserver;
// Files that are currently in the process of bulk importing. Access to this is protected by the
// tablet lock.
private final Set<FileRef> bulkImporting = new HashSet<>();
// Files that were successfully bulk imported.
private final Cache<Long,List<FileRef>> bulkImported = CacheBuilder.newBuilder().build();
private final int logId;
@Override
public int getLogId() {
return logId;
}
public static class LookupResult {
public List<Range> unfinishedRanges = new ArrayList<>();
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(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;
this.splitCreationTime = 0;
}
public Tablet(final TabletServer tabletServer, final KeyExtent extent,
final TabletResourceManager trm, TabletData data) throws IOException {
this.tabletServer = tabletServer;
this.extent = extent;
this.tabletResources = trm;
this.lastLocation = data.getLastLocation();
this.lastFlushID = data.getFlushID();
this.lastCompactID = data.getCompactID();
this.splitCreationTime = data.getSplitTime();
this.tabletTime = TabletTime.getInstance(data.getTime());
this.persistedTime = tabletTime.getTime();
this.logId = tabletServer.createLogId();
TableConfiguration tblConf = tabletServer.getTableConfiguration(extent);
if (null == tblConf) {
Tables.clearCache(tabletServer.getInstance());
tblConf = tabletServer.getTableConfiguration(extent);
requireNonNull(tblConf, "Could not get table configuration for " + extent.getTableId());
}
this.tableConfiguration = tblConf;
// translate any volume changes
VolumeManager fs = tabletServer.getFileSystem();
boolean replicationEnabled =
ReplicationConfigurationUtil.isEnabled(extent, this.tableConfiguration);
TabletFiles tabletPaths =
new TabletFiles(data.getDirectory(), data.getLogEntris(), data.getDataFiles());
tabletPaths = VolumeUtil.updateTabletVolumes(tabletServer, tabletServer.getLock(), fs, extent,
tabletPaths, replicationEnabled);
// deal with relative path for the directory
Path locationPath;
if (tabletPaths.dir.contains(":")) {
locationPath = new Path(tabletPaths.dir);
} else {
locationPath = tabletServer.getFileSystem().getFullPath(FileType.TABLE,
extent.getTableId() + tabletPaths.dir);
}
this.location = locationPath;
this.tabletDirectory = tabletPaths.dir;
for (Entry<Long,List<FileRef>> entry : data.getBulkImported().entrySet()) {
this.bulkImported.put(entry.getKey(), new CopyOnWriteArrayList<>(entry.getValue()));
}
setupDefaultSecurityLabels(extent);
final List<LogEntry> logEntries = tabletPaths.logEntries;
final SortedMap<FileRef,DataFileValue> datafiles = tabletPaths.datafiles;
tableConfiguration.addObserver(configObserver = new ConfigurationObserver() {
private void reloadConstraints() {
log.debug("Reloading constraints for extent: " + extent);
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.trace("Session expired, no longer updating per table props...");
}
});
tableConfiguration.getNamespaceConfiguration().addObserver(configObserver);
tabletMemory = new TabletMemory(this);
// Force a load of any per-table properties
configObserver.propertiesChanged();
if (!logEntries.isEmpty()) {
log.info("Starting Write-Ahead Log recovery for " + this.extent);
final AtomicLong entriesUsedOnTablet = new AtomicLong(0);
// track max time from walog entries without timestamps
final AtomicLong maxTime = new AtomicLong(Long.MIN_VALUE);
final CommitSession commitSession = getTabletMemory().getCommitSession();
try {
Set<String> absPaths = new HashSet<>();
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
maxTime.set(Math.max(maxTime.get(), columnUpdate.getTimestamp()));
}
}
getTabletMemory().mutate(commitSession, Collections.singletonList(m));
entriesUsedOnTablet.incrementAndGet();
}
});
if (maxTime.get() != Long.MIN_VALUE) {
tabletTime.useMaxTimeFromWALog(maxTime.get());
}
commitSession.updateMaxCommittedTime(tabletTime.getTime());
if (entriesUsedOnTablet.get() == 0) {
log.debug("No replayed mutations applied, removing unused entries for " + extent);
MetadataTableUtil.removeUnusedWALEntries(getTabletServer(), extent, logEntries,
tabletServer.getLock());
// No replication update to be made because the fact that this tablet didn't use any
// mutations
// from the WAL implies nothing about use of this WAL by other tablets. Do nothing.
logEntries.clear();
} else if (ReplicationConfigurationUtil.isEnabled(extent,
tabletServer.getTableConfiguration(extent))) {
// The logs are about to be re-used by this tablet, we need to record that they have data
// for this extent,
// but that they may get more data. logEntries is not cleared which will cause the
// elements
// in logEntries to be added to the currentLogs for this Tablet below.
//
// This update serves the same purpose as an update during a MinC. We know that the WAL
// was defined
// (written when the WAL was opened) but this lets us know there are mutations written to
// this WAL
// that could potentially be replicated. Because the Tablet is using this WAL, we can be
// sure that
// the WAL isn't closed (WRT replication Status) and thus we're safe to update its
// progress.
Status status = StatusUtil.openWithUnknownLength();
for (LogEntry logEntry : logEntries) {
log.debug("Writing updated status to metadata table for " + logEntry.filename + " "
+ ProtobufUtil.toString(status));
ReplicationTableUtil.updateFiles(tabletServer, extent, logEntry.filename, status);
}
}
} catch (Throwable t) {
String msg = "Error recovering tablet " + extent + " from log files";
if (tableConfiguration.getBoolean(Property.TABLE_FAILURES_IGNORE)) {
log.warn(msg, t);
} else {
throw new RuntimeException(msg, t);
}
}
// make some closed references that represent the recovered logs
currentLogs = new HashSet<>();
for (LogEntry logEntry : logEntries) {
currentLogs.add(new DfsLogger(tabletServer.getServerConfig(), logEntry.filename,
logEntry.getColumnQualifier().toString()));
}
rebuildReferencedLogs();
log.info(
"Write-Ahead Log recovery complete for " + this.extent + " (" + entriesUsedOnTablet.get()
+ " 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(data.getScanFiles());
// 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, long batchTimeOut)
throws IOException {
LookupResult lookupResult = new LookupResult();
boolean exceededMemoryUsage = false;
boolean tabletClosed = false;
Set<ByteSequence> cfset = null;
if (columnSet.size() > 0)
cfset = LocalityGroupUtil.families(columnSet);
long timeToRun = TimeUnit.MILLISECONDS.toNanos(batchTimeOut);
long startNanos = System.nanoTime();
if (batchTimeOut <= 0 || batchTimeOut == Long.MAX_VALUE) {
batchTimeOut = 0;
}
// determine if the iterator supported yielding
YieldCallback<Key> yield = new YieldCallback<>();
if (mmfi instanceof YieldingKeyValueIterator)
((YieldingKeyValueIterator<Key,Value>) mmfi).enableYielding(yield);
boolean yielded = false;
for (Range range : ranges) {
boolean timesUp = batchTimeOut > 0 && (System.nanoTime() - startNanos) > timeToRun;
if (exceededMemoryUsage || tabletClosed || timesUp || yielded) {
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()) {
if (yield.hasYielded()) {
throw new IOException("Coding error: hasTop returned true but has yielded at "
+ yield.getPositionAndReset());
}
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;
timesUp = batchTimeOut > 0 && (System.nanoTime() - startNanos) > timeToRun;
if (exceededMemoryUsage || timesUp) {
addUnfinishedRange(lookupResult, range, key, false);
break;
}
mmfi.next();
}
if (yield.hasYielded()) {
yielded = true;
Key yieldPosition = yield.getPositionAndReset();
if (!range.contains(yieldPosition)) {
throw new IOException("Underlying iterator yielded to a position outside of its range: "
+ yieldPosition + " not in " + range);
}
if (!results.isEmpty()
&& yieldPosition.compareTo(results.get(results.size() - 1).getKey()) <= 0) {
throw new IOException("Underlying iterator yielded to a position"
+ " that does not follow the last key returned: " + yieldPosition + " <= "
+ results.get(results.size() - 1).getKey());
}
addUnfinishedRange(lookupResult, range, yieldPosition, false);
log.debug("Scan yield detected at position " + yieldPosition);
Metrics scanMetrics = getTabletServer().getScanMetrics();
if (scanMetrics.isEnabled())
scanMetrics.add(TabletServerScanMetrics.YIELD, 1);
}
} 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 " + extent + "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 void checkConditions(ConditionChecker checker, Authorizations authorizations,
AtomicBoolean iFlag) throws IOException {
ScanDataSource dataSource =
new ScanDataSource(this, authorizations, this.defaultSecurityLabel, iFlag);
try {
SortedKeyValueIterator<Key,Value> iter = new SourceSwitchingIterator(dataSource);
checker.check(iter);
} 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);
}
}
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,
SamplerConfiguration samplerConfig, long batchTimeOut, String classLoaderContext)
throws IOException {
if (ranges.size() == 0) {
return new LookupResult();
}
ranges = Range.mergeOverlapping(ranges);
if (ranges.size() > 1) {
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, samplerConfig, batchTimeOut, classLoaderContext);
LookupResult result = null;
try {
SortedKeyValueIterator<Key,Value> iter = new SourceSwitchingIterator(dataSource);
result = lookup(iter, ranges, columns, results, maxResultSize, batchTimeOut);
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,
long batchTimeOut, boolean isolated) throws IOException {
// log.info("In nextBatch..");
long timeToRun = TimeUnit.MILLISECONDS.toNanos(batchTimeOut);
long startNanos = System.nanoTime();
if (batchTimeOut == Long.MAX_VALUE || batchTimeOut <= 0) {
batchTimeOut = 0;
}
List<KVEntry> results = new ArrayList<>();
Key key = null;
Value value;
long resultSize = 0L;
long resultBytes = 0L;
long maxResultsSize = tableConfiguration.getMemoryInBytes(Property.TABLE_SCAN_MAXMEM);
Key continueKey = null;
boolean skipContinueKey = false;
YieldCallback<Key> yield = new YieldCallback<>();
// we cannot yield if we are in isolation mode
if (!isolated) {
if (iter instanceof YieldingKeyValueIterator)
((YieldingKeyValueIterator<Key,Value>) iter).enableYielding(yield);
}
if (columns.size() == 0) {
iter.seek(range, LocalityGroupUtil.EMPTY_CF_SET, false);
} else {
iter.seek(range, LocalityGroupUtil.families(columns), true);
}
while (iter.hasTop()) {
if (yield.hasYielded()) {
throw new IOException(
"Coding error: hasTop returned true but has yielded at " + yield.getPositionAndReset());
}
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();
boolean timesUp = batchTimeOut > 0 && (System.nanoTime() - startNanos) >= timeToRun;
if (resultSize >= maxResultsSize || results.size() >= num || timesUp) {
continueKey = new Key(key);
skipContinueKey = true;
break;
}
iter.next();
}
if (yield.hasYielded()) {
continueKey = new Key(yield.getPositionAndReset());
skipContinueKey = true;
if (!range.contains(continueKey)) {
throw new IOException("Underlying iterator yielded to a position outside of its range: "
+ continueKey + " not in " + range);
}
if (!results.isEmpty()
&& continueKey.compareTo(results.get(results.size() - 1).getKey()) <= 0) {
throw new IOException(
"Underlying iterator yielded to a position that does not follow the last key returned: "
+ continueKey + " <= " + results.get(results.size() - 1).getKey());
}
log.debug("Scan yield detected at position " + continueKey);
Metrics scanMetrics = getTabletServer().getScanMetrics();
if (scanMetrics.isEnabled())
scanMetrics.add(TabletServerScanMetrics.YIELD, 1);
} else if (iter.hasTop() == false) {
// end of tablet has been reached
continueKey = null;
if (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, SamplerConfiguration samplerConfig,
long batchTimeOut, String classLoaderContext) {
// 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, samplerConfig, batchTimeOut, classLoaderContext);
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;
String oldName = Thread.currentThread().getName();
try {
Thread.currentThread().setName("Minor compacting " + this.extent);
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(tabletServer, 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("Exception occurred during minor compaction on " + extent, e);
} catch (Error e) {
// Weird errors like "OutOfMemoryError" when trying to create the thread for the compaction
failed = true;
throw new RuntimeException("Exception occurred during minor compaction on " + extent, e);
} finally {
Thread.currentThread().setName(oldName);
try {
getTabletMemory().finalizeMinC();
} catch (Throwable t) {
log.error("Failed to free tablet memory", t);
}
if (!failed) {
lastMinorCompactionFinishTime = System.currentTimeMillis();
}
Metrics 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) {
Preconditions.checkState(otherLogs.isEmpty());
Preconditions.checkState(referencedLogs.equals(currentLogs));
CommitSession oldCommitSession = getTabletMemory().prepareForMinC();
otherLogs = currentLogs;
currentLogs = new HashSet<>();
FileRef mergeFile = null;
if (mincReason != MinorCompactionReason.RECOVERY) {
mergeFile = getDatafileManager().reserveMergingMinorCompactionFile();
}
double tracePercent =
tabletServer.getConfiguration().getFraction(Property.TSERV_MINC_TRACE_PERCENT);
return new MinorCompactionTask(this, mergeFile, oldCommitSession, flushId, mincReason,
tracePercent);
}
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) {
// if multiple threads were allowed to update this outside of a sync block, then it would be
// a race condition
MetadataTableUtil.updateTabletFlushID(extent, tableFlushID, tabletServer,
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 + "/" + tabletServer.getInstance().getInstanceID()
+ Constants.ZTABLES + "/" + extent.getTableId() + Constants.ZTABLE_FLUSH_ID;
return Long
.parseLong(new String(ZooReaderWriter.getInstance().getData(zTablePath, null), UTF_8));
} catch (InterruptedException e) {
throw new RuntimeException("Exception on " + extent + " getting flush ID", e);
} catch (NumberFormatException nfe) {
throw new RuntimeException("Exception on " + extent + " getting flush ID", nfe);
} catch (KeeperException ke) {
if (ke instanceof NoNodeException) {
throw (NoNodeException) ke;
} else {
throw new RuntimeException("Exception on " + extent + " getting flush ID", ke);
}
}
}
long getCompactionCancelID() {
String zTablePath = Constants.ZROOT + "/" + tabletServer.getInstance().getInstanceID()
+ Constants.ZTABLES + "/" + extent.getTableId() + Constants.ZTABLE_COMPACT_CANCEL_ID;
try {
return Long
.parseLong(new String(ZooReaderWriter.getInstance().getData(zTablePath, null), UTF_8));
} catch (KeeperException e) {
throw new RuntimeException(e);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
public Pair<Long,UserCompactionConfig> getCompactionID() throws NoNodeException {
try {
String zTablePath = Constants.ZROOT + "/" + tabletServer.getInstance().getInstanceID()
+ Constants.ZTABLES + "/" + extent.getTableId() + Constants.ZTABLE_COMPACT_ID;
String[] tokens =
new String(ZooReaderWriter.getInstance().getData(zTablePath, null), UTF_8).split(",");
long compactID = Long.parseLong(tokens[0]);
UserCompactionConfig compactionConfig = new UserCompactionConfig();
if (tokens.length > 1) {
Hex hex = new Hex();
ByteArrayInputStream bais =
new ByteArrayInputStream(hex.decode(tokens[1].split("=")[1].getBytes(UTF_8)));
DataInputStream dis = new DataInputStream(bais);
try {
compactionConfig.readFields(dis);
} catch (IOException e) {
throw new RuntimeException(e);
}
KeyExtent ke = new KeyExtent(extent.getTableId(), compactionConfig.getEndRow(),
compactionConfig.getStartRow());
if (!ke.overlaps(extent)) {
// only use iterators if compaction range overlaps
compactionConfig = new UserCompactionConfig();
}
}
return new Pair<>(compactID, compactionConfig);
} catch (InterruptedException e) {
throw new RuntimeException("Exception on " + extent + " getting compaction ID", e);
} catch (NumberFormatException nfe) {
throw new RuntimeException("Exception on " + extent + " getting compaction ID", nfe);
} catch (KeeperException ke) {
if (ke instanceof NoNodeException) {
throw (NoNodeException) ke;
} else {
throw new RuntimeException("Exception on " + extent + " getting compaction ID", ke);
}
} catch (DecoderException e) {
throw new RuntimeException("Exception on " + extent + " getting compaction ID", e);
}
}
private synchronized CommitSession finishPreparingMutations(long time) {
if (writesInProgress < 0) {
throw new IllegalStateException("FATAL: Something really bad went wrong. Attempted to "
+ "increment a negative number of writes in progress " + writesInProgress + "on tablet "
+ extent);
}
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<>();
violators.add(mutation);
}
}
long time = tabletTime.setUpdateTimes(mutations);
if (!violations.isEmpty()) {
HashSet<Mutation> violatorsSet = new HashSet<>(violators);
ArrayList<Mutation> nonViolators = new ArrayList<>();
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("FATAL: Something really bad went wrong. Attempted to "
+ "decrement the number of writes in progress " + writesInProgress
+ " to < 0 on tablet " + extent);
}
if (isCloseComplete()) {
throw new IllegalStateException(
"Tablet " + extent + " 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("Bad state initiating close on " + extent
+ ". State not saved and requested minor compactions queue");
}
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("Exception on " + extent + " during prep for MinC", 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("Bad close state " + closeState + " on tablet " + extent);
}
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("Exception on " + extent + " during prep for MinC", 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);
sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
}
}
if (err != null) {
ProblemReports.getInstance(tabletServer).report(new ProblemReport(extent.getTableId(),
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() + " for tablet " + extent, 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(tabletServer, 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 || referencedLogs.size() != 0) {
String msg = "Closed tablet " + extent + " has walog entries in memory currentLogs = "
+ currentLogs + " otherLogs = " + otherLogs + " refererncedLogs = " + referencedLogs;
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 final 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;
}
if (keys.isEmpty()) {
log.info("Cannot split tablet " + extent + ", files contain no data for tablet.");
// set the following to keep tablet from attempting to split until the tablets set of files
// changes.
sawBigRow = true;
timeOfLastMinCWhenBigFreakinRowWasSeen = lastMinorCompactionFinishTime;
timeOfLastImportWhenBigFreakinRowWasSeen = lastMapFileImportTime;
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 (mid == null) {
throw new IllegalStateException("Could not determine midpoint for files on " + extent);
}
// 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<>();
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.newReaderBuilder().forFile(file.path().toString(), ns, ns.getConf())
.withTableConfiguration(this.getTableConfiguration()).seekToBeginning().build();
try {
Key first = openReader.getFirstKey();
Key last = openReader.getLastKey();
result.put(file, new Pair<>(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<>();
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) || (first != null && !extent.contains(first.getRow()))
|| (last != null && !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 CompactionStats _majorCompact(MajorCompactionReason reason)
throws IOException, CompactionCanceledException {
long t1, t2, t3;
Pair<Long,UserCompactionConfig> compactionId = null;
CompactionStrategy strategy = null;
Map<FileRef,Pair<Key,Key>> firstAndLastKeys = null;
if (reason == MajorCompactionReason.USER) {
try {
compactionId = getCompactionID();
strategy = createCompactionStrategy(compactionId.getSecond().getCompactionStrategy());
} catch (NoNodeException e) {
throw new RuntimeException("Exception on " + extent + " during MajC", e);
}
} else if (reason == MajorCompactionReason.NORMAL || reason == MajorCompactionReason.IDLE) {
strategy = Property.createTableInstanceFromPropertyName(tableConfiguration,
Property.TABLE_COMPACTION_STRATEGY, CompactionStrategy.class,
new DefaultCompactionStrategy());
strategy.init(Property.getCompactionStrategyOptions(tableConfiguration));
} else if (reason == MajorCompactionReason.CHOP) {
firstAndLastKeys = getFirstAndLastKeys(getDatafileManager().getDatafileSizes());
} else {
throw new IllegalArgumentException(
"Unknown compaction reason " + reason + " during MajC on " + extent);
}
if (strategy != null) {
MajorCompactionRequest request = new MajorCompactionRequest(extent, reason,
getTabletServer().getFileSystem(), tableConfiguration);
request.setFiles(getDatafileManager().getDatafileSizes());
strategy.gatherInformation(request);
}
Map<FileRef,DataFileValue> filesToCompact = null;
int maxFilesToCompact = tableConfiguration.getCount(Property.TSERV_MAJC_THREAD_MAXOPEN);
CompactionStats majCStats = new CompactionStats();
CompactionPlan plan = null;
boolean propogateDeletes = false;
boolean updateCompactionID = 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<>();
if (reason == MajorCompactionReason.CHOP) {
// enforce rules: files with keys outside our range need to be compacted
inputFiles.addAll(findChopFiles(extent, firstAndLastKeys, 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()) {
if (reason == MajorCompactionReason.USER) {
if (compactionId.getSecond().getIterators().isEmpty()) {
log.debug(
"No-op major compaction by USER on 0 input files because no iterators present.");
lastCompactID = compactionId.getFirst();
updateCompactionID = true;
} else {
log.debug("Major compaction by USER on 0 input files with iterators.");
filesToCompact = new HashMap<>();
}
} else {
return majCStats;
}
} else {
// If no original files will exist at the end of the compaction, we do not have to propogate
// deletes
Set<FileRef> droppedFiles = new HashSet<>();
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<>(allFiles);
filesToCompact.keySet().retainAll(inputFiles);
getDatafileManager().reserveMajorCompactingFiles(filesToCompact.keySet());
}
t3 = System.currentTimeMillis();
}
try {
log.debug(String.format("MajC initiate lock %.2f secs, wait %.2f secs", (t3 - t2) / 1000.0,
(t2 - t1) / 1000.0));
if (updateCompactionID) {
MetadataTableUtil.updateTabletCompactID(extent, compactionId.getFirst(), tabletServer,
getTabletServer().getLock());
return majCStats;
}
if (!propogateDeletes && compactionId == null) {
// compacting everything, so update the compaction id in metadata
try {
compactionId = getCompactionID();
if (compactionId.getSecond().getCompactionStrategy() != null) {
compactionId = null;
// TODO maybe return unless chop?
}
} catch (NoNodeException e) {
throw new RuntimeException("Exception on " + extent + " during MajC", e);
}
}
List<IteratorSetting> compactionIterators = new ArrayList<>();
if (compactionId != null) {
if (reason == MajorCompactionReason.USER) {
if (getCompactionCancelID() >= compactionId.getFirst()) {
// compaction was canceled
return majCStats;
}
compactionIterators = compactionId.getSecond().getIterators();
synchronized (this) {
if (lastCompactID >= compactionId.getFirst())
// already compacted
return majCStats;
}
}
}
// need to handle case where only one file is being major compacted
// ACCUMULO-3645 run loop at least once, even if filesToCompact.isEmpty()
do {
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() {
// avoid calling isClosing() because its synchronized and this is called frequently in
// compaction
return closeState != CloseState.CLOSING;
}
@Override
public IteratorScope getIteratorScope() {
return IteratorScope.majc;
}
@Override
public RateLimiter getReadLimiter() {
return getTabletServer().getMajorCompactionReadLimiter();
}
@Override
public RateLimiter getWriteLimiter() {
return getTabletServer().getMajorCompactionWriteLimiter();
}
};
HashMap<FileRef,DataFileValue> copy =
new HashMap<>(getDatafileManager().getDatafileSizes());
if (!copy.keySet().containsAll(smallestFiles))
throw new IllegalStateException("Cannot find data file values for " + smallestFiles
+ " on " + extent + " during MajC");
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(tabletServer, 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();
}
} while (filesToCompact.size() > 0);
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
// short-circuit; also handles zero files case
if (filesToCompact.size() <= maxFilesToCompact) {
Set<FileRef> smallestFiles = new HashSet<>(filesToCompact.keySet());
filesToCompact.clear();
return smallestFiles;
}
PriorityQueue<Pair<FileRef,Long>> fileHeap =
new PriorityQueue<>(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<>(entry.getKey(), entry.getValue().getSize()));
}
Set<FileRef> smallestFiles = new HashSet<>();
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();
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;
}
Span span = null;
try {
double tracePercent =
tabletServer.getConfiguration().getFraction(Property.TSERV_MAJC_TRACE_PERCENT);
ProbabilitySampler sampler = new ProbabilitySampler(tracePercent);
span = Trace.on("majorCompaction", sampler);
majCStats = _majorCompact(reason);
if (reason == MajorCompactionReason.CHOP) {
MetadataTableUtil.chopped(getTabletServer(), getExtent(), this.getTabletServer().getLock());
getTabletServer()
.enqueueMasterMessage(new TabletStatusMessage(TabletLoadState.CHOPPED, extent));
}
success = true;
} catch (CompactionCanceledException cce) {
log.debug("Major compaction canceled, extent = " + getExtent());
} catch (IOException ioe) {
log.error("MajC Failed, extent = " + getExtent(), ioe);
} catch (RuntimeException e) {
log.error("MajC Unexpected exception, extent = " + getExtent(), e);
} finally {
// ensure we always reset boolean, even
// when an exception is thrown
synchronized (this) {
majorCompactionState = null;
this.notifyAll();
}
if (span != null) {
span.data("extent", "" + getExtent());
if (majCStats != null) {
span.data("read", "" + majCStats.getEntriesRead());
span.data("written", "" + majCStats.getEntriesWritten());
}
span.stop();
}
}
long count = 0;
if (majCStats != null)
count = majCStats.getEntriesRead();
timer.updateTime(Operation.MAJOR, queued, start, count, !success);
return majCStats;
}
@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,TabletData> split(byte[] sp) throws IOException {
if (sp != null && extent.getEndRow() != null && extent.getEndRow().equals(new Text(sp))) {
throw new IllegalArgumentException(
"Attempting to split on EndRow " + extent.getEndRow() + " for " + extent);
}
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,TabletData> newTablets = new TreeMap<>();
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 =
createTabletDirectory(getTabletServer().getFileSystem(), extent.getTableId(), midRow);
// write new tablet information to MetadataTable
SortedMap<FileRef,DataFileValue> lowDatafileSizes = new TreeMap<>();
SortedMap<FileRef,DataFileValue> highDatafileSizes = new TreeMap<>();
List<FileRef> highDatafilesToRemove = new ArrayList<>();
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();
MetadataTableUtil.splitTablet(high, extent.getPrevEndRow(), splitRatio, getTabletServer(),
getTabletServer().getLock());
MasterMetadataUtil.addNewTablet(getTabletServer(), low, lowDirectory,
getTabletServer().getTabletSession(), lowDatafileSizes, getBulkIngestedFiles(), time,
lastFlushID, lastCompactID, getTabletServer().getLock());
MetadataTableUtil.finishSplit(high, highDatafileSizes, highDatafilesToRemove,
getTabletServer(), getTabletServer().getLock());
log.log(TLevel.TABLET_HIST, extent + " split " + low + " " + high);
newTablets.put(high, new TabletData(tabletDirectory, highDatafileSizes, time, lastFlushID,
lastCompactID, lastLocation, getBulkIngestedFiles()));
newTablets.put(low, new TabletData(lowDirectory, lowDatafileSizes, time, lastFlushID,
lastCompactID, lastLocation, getBulkIngestedFiles()));
long t2 = System.currentTimeMillis();
log.debug(String.format("offline split time : %6.2f secs", (t2 - t1) / 1000.0));
closeState = CloseState.COMPLETE;
return newTablets;
}
}
public SortedMap<FileRef,DataFileValue> getDatafiles() {
return getDatafileManager().getDatafileSizes();
}
public double queryRate() {
return queryRate.rate();
}
public double queryByteRate() {
return queryByteRate.rate();
}
public double ingestRate() {
return ingestRate.rate();
}
public double ingestByteRate() {
return ingestByteRate.rate();
}
public double scanRate() {
return scannedRate.rate();
}
public long totalQueries() {
return this.queryCount;
}
public long totalIngest() {
return this.ingestCount;
}
// synchronized?
public void updateRates(long now) {
queryRate.update(now, queryCount);
queryByteRate.update(now, queryBytes);
ingestRate.update(now, ingestCount);
ingestByteRate.update(now, ingestBytes);
scannedRate.update(now, scannedCount.get());
}
public long getSplitCreationTime() {
return splitCreationTime;
}
public void importMapFiles(long tid, Map<FileRef,MapFileInfo> fileMap, boolean setTime)
throws IOException {
Map<FileRef,DataFileValue> entries = new HashMap<>(fileMap.size());
List<String> files = new ArrayList<>();
for (Entry<FileRef,MapFileInfo> entry : fileMap.entrySet()) {
entries.put(entry.getKey(), new DataFileValue(entry.getValue().estimatedSize, 0l));
files.add(entry.getKey().path().toString());
}
// Clients timeout and will think that this operation failed.
// Don't do it if we spent too long waiting for the lock
long now = System.currentTimeMillis();
synchronized (this) {
if (isClosed()) {
throw new IOException("tablet " + extent + " is closed");
}
// TODO check seems uneeded now - ACCUMULO-1291
long lockWait = System.currentTimeMillis() - now;
if (lockWait
> getTabletServer().getConfiguration().getTimeInMillis(Property.GENERAL_RPC_TIMEOUT)) {
throw new IOException(
"Timeout waiting " + (lockWait / 1000.) + " seconds to get tablet lock");
}
List<FileRef> alreadyImported = bulkImported.getIfPresent(tid);
if (alreadyImported != null) {
for (FileRef entry : alreadyImported) {
if (fileMap.remove(entry) != null) {
log.info("Ignoring import of bulk file already imported: " + entry);
}
}
}
Iterator<FileRef> fiter = fileMap.keySet().iterator();
while (fiter.hasNext()) {
FileRef file = fiter.next();
if (bulkImporting.contains(file)) {
log.info("Ignoring import of bulk file currently importing: " + file);
fiter.remove();
}
}
if (fileMap.isEmpty()) {
return;
}
if (writesInProgress < 0) {
throw new IllegalStateException("FATAL: Something really bad went wrong. Attempted to "
+ "increment a negative number of writes in progress " + writesInProgress + "on tablet "
+ extent);
}
// prevent other threads from processing this file while its added to the metadata table.
bulkImporting.addAll(fileMap.keySet());
writesInProgress++;
}
try {
tabletServer.updateBulkImportState(files, BulkImportState.LOADING);
getDatafileManager().importMapFiles(tid, entries, setTime);
lastMapFileImportTime = System.currentTimeMillis();
if (needsSplit()) {
getTabletServer().executeSplit(this);
} else {
initiateMajorCompaction(MajorCompactionReason.NORMAL);
}
} finally {
synchronized (this) {
if (writesInProgress < 1)
throw new IllegalStateException("FATAL: Something really bad went wrong. Attempted to "
+ "decrement the number of writes in progress " + writesInProgress
+ " to < 0 on tablet " + extent);
writesInProgress--;
if (writesInProgress == 0)
this.notifyAll();
if (!bulkImporting.removeAll(fileMap.keySet())) {
throw new AssertionError(
"Likely bug in code, always expect to remove something. Please open an Accumulo issue.");
}
try {
bulkImported.get(tid, new Callable<List<FileRef>>() {
@Override
public List<FileRef> call() throws Exception {
return new ArrayList<>();
}
}).addAll(fileMap.keySet());
} catch (Exception ex) {
log.info(ex.toString(), ex);
}
tabletServer.removeBulkImportState(files);
}
}
}
private Set<DfsLogger> currentLogs = new HashSet<>();
private Set<DfsLogger> otherLogs = Collections.emptySet();
// An immutable copy of currentLogs + otherLogs. This exists so that removeInUseLogs() does not
// have to get the tablet lock. See #558
private volatile Set<DfsLogger> referencedLogs = Collections.emptySet();
private synchronized void rebuildReferencedLogs() {
/*
* Each tablet has the following sets of WALogs. While a WALog exists in one set, garbage
* collection must be avoided.
*
* 1. WALogs for the active in memory map
*
* 2. WAlogs for the minor compacting in memory map
*
* 3. WAlogs for a newly minor compacted file that is being added to the metadata table.
*
* Set 1 is currentLogs. Set 2 is otherLogs. Set 3 only exist in referenced logs as a side
* effect of not calling this method in beginClearingUnusedLogs() when otherLogs is cleared.
*
* Ensuring referencedLogs accurately tracks these sets ensures in use walogs are not GCed.
*/
Builder<DfsLogger> builder = ImmutableSet.builder();
builder.addAll(currentLogs);
builder.addAll(otherLogs);
referencedLogs = builder.build();
}
public void removeInUseLogs(Set<DfsLogger> candidates) {
candidates.removeAll(referencedLogs);
}
public void checkIfMinorCompactionNeededForLogs(List<DfsLogger> closedLogs) {
// grab this outside of tablet lock.
int maxLogs = tableConfiguration.getCount(Property.TABLE_MINC_LOGS_MAX);
String reason = null;
synchronized (this) {
if (currentLogs.size() >= maxLogs) {
reason = "referenced " + currentLogs.size() + " write ahead logs";
} else if (maxLogs < closedLogs.size()) {
// If many tablets reference a single WAL, but each tablet references a different WAL then
// this could result in the tablet server referencing many WALs. For recovery that would
// mean each tablet had to process lots of WAL. This check looks for a single use of an
// older WAL and compacts if one is found. The following check assumes the most recent WALs
// are at the end of the list and ignores these.
List<DfsLogger> oldClosed = closedLogs.subList(0, closedLogs.size() - maxLogs);
for (DfsLogger closedLog : oldClosed) {
if (currentLogs.contains(closedLog)) {
reason = "referenced at least one old write ahead log " + closedLog.getFileName();
break;
}
}
}
}
if (reason != null) {
// initiate and log outside of tablet lock
initiateMinorCompaction(MinorCompactionReason.SYSTEM);
log.debug("Initiating minor compaction for " + getExtent() + " because " + reason);
}
}
Set<String> beginClearingUnusedLogs() {
Set<String> unusedLogs = new HashSet<>();
ArrayList<String> otherLogsCopy = new ArrayList<>();
ArrayList<String> currentLogsCopy = new ArrayList<>();
// do not hold tablet lock while acquiring the log lock
logLock.lock();
synchronized (this) {
if (removingLogs)
throw new IllegalStateException("Attempted to clear logs when removal of logs in progress");
for (DfsLogger logger : otherLogs) {
otherLogsCopy.add(logger.toString());
unusedLogs.add(logger.getMeta());
}
for (DfsLogger logger : currentLogs) {
currentLogsCopy.add(logger.toString());
unusedLogs.remove(logger.getMeta());
}
otherLogs = Collections.emptySet();
// Intentionally NOT calling rebuildReferenedLogs() here as that could cause GC of in use
// walogs(see #539). The clearing of otherLogs is reflected in refererncedLogs when
// finishClearingUnusedLogs() calls rebuildReferenedLogs(). See the comments in
// rebuildReferenedLogs() for more info.
if (unusedLogs.size() > 0)
removingLogs = true;
}
// do debug logging outside tablet lock
for (String logger : otherLogsCopy) {
log.debug("Logs for memory compacted: " + getExtent() + " " + logger);
}
for (String logger : currentLogsCopy) {
log.debug("Logs for current memory: " + getExtent() + " " + logger);
}
for (String logger : unusedLogs) {
log.debug("Logs to be destroyed: " + getExtent() + " " + logger);
}
return unusedLogs;
}
synchronized void finishClearingUnusedLogs() {
removingLogs = false;
rebuildReferencedLogs();
logLock.unlock();
}
private boolean removingLogs = false;
// this lock is basically used to synchronize writing of log info to metadata
private final ReentrantLock logLock = new ReentrantLock();
// don't release the lock if this method returns true for success; instead, the caller should
// clean up by calling finishUpdatingLogsUsed()
@Override
public boolean beginUpdatingLogsUsed(InMemoryMap memTable, DfsLogger more, boolean mincFinish) {
boolean releaseLock = true;
// do not hold tablet lock while acquiring the log lock
logLock.lock();
try {
synchronized (this) {
if (isCloseComplete()) {
throw new IllegalStateException("Can not update logs of closed tablet " + extent);
}
boolean addToOther;
if (memTable == getTabletMemory().getMinCMemTable())
addToOther = true;
else if (memTable == getTabletMemory().getMemTable())
addToOther = false;
else
throw new IllegalArgumentException("Passed in memtable that is not in use for " + extent);
if (mincFinish) {
if (addToOther)
throw new IllegalStateException("Adding to other logs for mincFinish on " + extent);
if (otherLogs.size() != 0)
throw new IllegalStateException("Expect other logs to be 0 when minC finish, but its "
+ otherLogs + " for " + extent);
// when writing a minc finish event, there is no need to add the log to metadata
// if nothing has been logged for the tablet since the minor compaction started
if (currentLogs.size() == 0)
return !releaseLock;
}
int numAdded = 0;
int numContained = 0;
if (addToOther) {
if (otherLogs.add(more))
numAdded++;
if (currentLogs.contains(more))
numContained++;
} else {
if (currentLogs.add(more))
numAdded++;
if (otherLogs.contains(more))
numContained++;
}
if (numAdded > 0) {
rebuildReferencedLogs();
}
if (numAdded > 0 && numAdded != 1) {
// expect to add all or none
throw new IllegalArgumentException(
"Added subset of logs " + extent + " " + more + " " + currentLogs);
}
if (numContained > 0 && numContained != 1) {
// expect to contain all or none
throw new IllegalArgumentException(
"Other logs contained subset of logs " + extent + " " + more + " " + otherLogs);
}
if (numAdded > 0 && numContained == 0) {
releaseLock = false;
}
return !releaseLock;
}
} finally {
if (releaseLock)
logLock.unlock();
}
}
@Override
public void finishUpdatingLogsUsed() {
logLock.unlock();
}
synchronized public void chopFiles() {
initiateMajorCompaction(MajorCompactionReason.CHOP);
}
private CompactionStrategy createCompactionStrategy(CompactionStrategyConfig strategyConfig) {
String context = tableConfiguration.get(Property.TABLE_CLASSPATH);
String clazzName = strategyConfig.getClassName();
try {
Class<? extends CompactionStrategy> clazz;
if (context != null && !context.equals(""))
clazz = AccumuloVFSClassLoader.getContextManager().loadClass(context, clazzName,
CompactionStrategy.class);
else
clazz = AccumuloVFSClassLoader.loadClass(clazzName, CompactionStrategy.class);
CompactionStrategy strategy = clazz.newInstance();
strategy.init(strategyConfig.getOptions());
return strategy;
} catch (Exception e) {
throw new RuntimeException("Error creating compaction strategy on " + extent, e);
}
}
public void compactAll(long compactionId, UserCompactionConfig compactionConfig) {
boolean updateMetadata = false;
synchronized (this) {
if (lastCompactID >= compactionId)
return;
if (isMinorCompactionRunning()) {
// want to wait for running minc to finish before starting majc, see ACCUMULO-3041
if (compactionWaitInfo.compactionID == compactionId) {
if (lastFlushID == compactionWaitInfo.flushID)
return;
} else {
compactionWaitInfo.compactionID = compactionId;
compactionWaitInfo.flushID = lastFlushID;
return;
}
}
if (isClosing() || isClosed() || majorCompactionQueued.contains(MajorCompactionReason.USER)
|| isMajorCompactionRunning())
return;
CompactionStrategyConfig strategyConfig = compactionConfig.getCompactionStrategy();
CompactionStrategy strategy = createCompactionStrategy(strategyConfig);
MajorCompactionRequest request = new MajorCompactionRequest(extent,
MajorCompactionReason.USER, getTabletServer().getFileSystem(), tableConfiguration);
request.setFiles(getDatafileManager().getDatafileSizes());
try {
if (strategy.shouldCompact(request)) {
initiateMajorCompaction(MajorCompactionReason.USER);
} else {
majorCompactionState = CompactionState.IN_PROGRESS;
updateMetadata = true;
lastCompactID = compactionId;
}
} catch (IOException e) {
throw new RuntimeException("IOException on " + extent + " during compact all", e);
}
}
if (updateMetadata) {
try {
// if multiple threads were allowed to update this outside of a sync block, then it would be
// a race condition
MetadataTableUtil.updateTabletCompactID(extent, compactionId, getTabletServer(),
getTabletServer().getLock());
} finally {
synchronized (this) {
majorCompactionState = null;
this.notifyAll();
}
}
}
}
@Override
public TableConfiguration getTableConfiguration() {
return tableConfiguration;
}
@Override
public Durability getDurability() {
return DurabilityImpl.fromString(getTableConfiguration().get(Property.TABLE_DURABILITY));
}
@Override
public void updateMemoryUsageStats(long size, long mincSize) {
getTabletResources().updateMemoryUsageStats(this, size, mincSize);
}
public long incrementDataSourceDeletions() {
return dataSourceDeletions.incrementAndGet();
}
synchronized public void updateQueryStats(int size, long numBytes) {
queryCount += size;
queryBytes += numBytes;
}
TabletServer getTabletServer() {
return tabletServer;
}
public void updatePersistedTime(long bulkTime, Map<FileRef,DataFileValue> paths, long tid) {
synchronized (timeLock) {
if (bulkTime > persistedTime)
persistedTime = bulkTime;
MetadataTableUtil.updateTabletDataFile(tid, extent, paths,
tabletTime.getMetadataValue(persistedTime), getTabletServer(),
getTabletServer().getLock());
}
}
public void updateTabletDataFile(long maxCommittedTime, FileRef newDatafile, FileRef absMergeFile,
DataFileValue dfv, Set<String> unusedWalLogs, Set<FileRef> filesInUseByScans, long flushId) {
synchronized (timeLock) {
if (maxCommittedTime > persistedTime)
persistedTime = maxCommittedTime;
String time = tabletTime.getMetadataValue(persistedTime);
MasterMetadataUtil.updateTabletDataFile(getTabletServer(), extent, newDatafile, absMergeFile,
dfv, time, filesInUseByScans, tabletServer.getClientAddressString(),
tabletServer.getLock(), unusedWalLogs, lastLocation, flushId);
}
}
TabletResourceManager getTabletResources() {
return tabletResources;
}
DatafileManager getDatafileManager() {
return datafileManager;
}
TabletMemory getTabletMemory() {
return tabletMemory;
}
public long getAndUpdateTime() {
return tabletTime.getAndUpdateTime();
}
public void flushComplete(long flushId) {
lastLocation = null;
dataSourceDeletions.incrementAndGet();
tabletMemory.finishedMinC();
lastFlushID = flushId;
computeNumEntries();
}
public TServerInstance resetLastLocation() {
TServerInstance result = lastLocation;
lastLocation = null;
return result;
}
synchronized public void addActiveScans(ScanDataSource scanDataSource) {
activeScans.add(scanDataSource);
}
public int removeScan(ScanDataSource scanDataSource) {
activeScans.remove(scanDataSource);
return activeScans.size();
}
synchronized public void setLastCompactionID(Long compactionId) {
if (compactionId != null)
this.lastCompactID = compactionId;
}
public void removeMajorCompactionQueuedReason(MajorCompactionReason reason) {
majorCompactionQueued.remove(reason);
}
public void minorCompactionWaitingToStart() {
minorCompactionState = CompactionState.WAITING_TO_START;
}
public void minorCompactionStarted() {
minorCompactionState = CompactionState.IN_PROGRESS;
}
public void minorCompactionComplete() {
minorCompactionState = null;
}
public TabletStats getTabletStats() {
return timer.getTabletStats();
}
public AtomicLong getScannedCounter() {
return scannedCount;
}
@SuppressModernizer
private static String createTabletDirectory(VolumeManager fs, String tableId, Text endRow) {
String lowDirectory;
UniqueNameAllocator namer = UniqueNameAllocator.getInstance();
String volume = fs.choose(Optional.of(tableId), ServerConstants.getBaseUris())
+ Constants.HDFS_TABLES_DIR + Path.SEPARATOR;
while (true) {
try {
if (endRow == null) {
lowDirectory = Constants.DEFAULT_TABLET_LOCATION;
Path lowDirectoryPath = new Path(volume + "/" + tableId + "/" + lowDirectory);
if (fs.exists(lowDirectoryPath) || fs.mkdirs(lowDirectoryPath)) {
FileSystem pathFs = fs.getVolumeByPath(lowDirectoryPath).getFileSystem();
return lowDirectoryPath.makeQualified(pathFs.getUri(), pathFs.getWorkingDirectory())
.toString();
}
log.warn("Failed to create " + lowDirectoryPath + " for unknown reason");
} else {
lowDirectory = "/" + Constants.GENERATED_TABLET_DIRECTORY_PREFIX + namer.getNextName();
Path lowDirectoryPath = new Path(volume + "/" + tableId + "/" + lowDirectory);
if (fs.exists(lowDirectoryPath))
throw new IllegalStateException("Attempting to create tablet dir for tableID " + tableId
+ " and dir exists when it should not: " + lowDirectoryPath);
if (fs.mkdirs(lowDirectoryPath)) {
FileSystem lowDirectoryFs = fs.getVolumeByPath(lowDirectoryPath).getFileSystem();
return lowDirectoryPath
.makeQualified(lowDirectoryFs.getUri(), lowDirectoryFs.getWorkingDirectory())
.toString();
}
}
} catch (IOException e) {
log.warn(e);
}
log.warn("Failed to create dir for tablet in table " + tableId + " in volume " + volume
+ " + will retry ...");
sleepUninterruptibly(3, TimeUnit.SECONDS);
}
}
public Map<Long,List<FileRef>> getBulkIngestedFiles() {
return new HashMap<>(bulkImported.asMap());
}
public void cleanupBulkLoadedFiles(Set<Long> tids) {
for (Long tid : tids) {
bulkImported.invalidate(tid);
}
}
}