You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by kt...@apache.org on 2021/02/24 02:17:37 UTC
[accumulo] branch 1451-external-compactions-feature updated: Move
code to run compactions into server base
This is an automated email from the ASF dual-hosted git repository.
kturner pushed a commit to branch 1451-external-compactions-feature
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/1451-external-compactions-feature by this push:
new 81fe695 Move code to run compactions into server base
81fe695 is described below
commit 81fe695a1e4d0e00f46c9f2701990826e1e766c0
Author: Keith Turner <kt...@apache.org>
AuthorDate: Tue Feb 23 20:21:06 2021 -0500
Move code to run compactions into server base
Required moving some functionality into ComapctionEnv in order to avoid moving
too many types. The refactoring cleaned up the code a good bit.
---
.../server/compaction}/CompactionInfo.java | 35 ++----------
.../server/compaction}/CompactionStats.java | 4 +-
.../accumulo/server/compaction}/Compactor.java | 62 +++++++++-------------
.../server/compaction}/CountingIterator.java | 2 +-
.../accumulo/tserver/ThriftClientHandler.java | 4 +-
.../accumulo/tserver/tablet/CompactableImpl.java | 3 +-
.../accumulo/tserver/tablet/CompactableUtils.java | 42 ++++++++++++---
.../accumulo/tserver/tablet/CompactionWatcher.java | 2 +
.../accumulo/tserver/tablet/MinorCompactor.java | 40 ++++++++++++--
.../org/apache/accumulo/tserver/tablet/Tablet.java | 1 +
.../accumulo/tserver/CountingIteratorTest.java | 2 +-
11 files changed, 113 insertions(+), 84 deletions(-)
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactionInfo.java b/server/base/src/main/java/org/apache/accumulo/server/compaction/CompactionInfo.java
similarity index 81%
rename from server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactionInfo.java
rename to server/base/src/main/java/org/apache/accumulo/server/compaction/CompactionInfo.java
index a5c4224..6195c86 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactionInfo.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/compaction/CompactionInfo.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.accumulo.tserver.tablet;
+package org.apache.accumulo.server.compaction;
import java.util.ArrayList;
import java.util.HashMap;
@@ -38,11 +38,13 @@ public class CompactionInfo {
private final String localityGroup;
private final long entriesRead;
private final long entriesWritten;
+ private final CompactionReason reason;
CompactionInfo(Compactor compactor) {
this.localityGroup = compactor.getCurrentLocalityGroup();
this.entriesRead = compactor.getEntriesRead();
this.entriesWritten = compactor.getEntriesWritten();
+ this.reason = compactor.getReason();
this.compactor = compactor;
}
@@ -84,37 +86,6 @@ public class CompactionInfo {
else
type = CompactionType.MAJOR;
- CompactionReason reason;
-
- if (compactor.hasIMM()) {
- switch (compactor.getMinCReason()) {
- case USER:
- reason = CompactionReason.USER;
- break;
- case CLOSE:
- reason = CompactionReason.CLOSE;
- break;
- case SYSTEM:
- default:
- reason = CompactionReason.SYSTEM;
- break;
- }
- } else {
- switch (compactor.getMajorCompactionReason()) {
- case USER:
- reason = CompactionReason.USER;
- break;
- case CHOP:
- reason = CompactionReason.CHOP;
- break;
- case SELECTOR:
- case SYSTEM:
- default:
- reason = CompactionReason.SYSTEM;
- break;
- }
- }
-
List<IterInfo> iiList = new ArrayList<>();
Map<String,Map<String,String>> iterOptions = new HashMap<>();
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactionStats.java b/server/base/src/main/java/org/apache/accumulo/server/compaction/CompactionStats.java
similarity index 94%
rename from server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactionStats.java
rename to server/base/src/main/java/org/apache/accumulo/server/compaction/CompactionStats.java
index fbaad04..3df80c0 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactionStats.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/compaction/CompactionStats.java
@@ -16,14 +16,14 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.accumulo.tserver.tablet;
+package org.apache.accumulo.server.compaction;
public class CompactionStats {
private long entriesRead;
private long entriesWritten;
private long fileSize;
- CompactionStats(long er, long ew) {
+ public CompactionStats(long er, long ew) {
this.setEntriesRead(er);
this.setEntriesWritten(ew);
}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Compactor.java b/server/base/src/main/java/org/apache/accumulo/server/compaction/Compactor.java
similarity index 89%
rename from server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Compactor.java
rename to server/base/src/main/java/org/apache/accumulo/server/compaction/Compactor.java
index 8efde9f..b228335 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Compactor.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/compaction/Compactor.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.accumulo.tserver.tablet;
+package org.apache.accumulo.server.compaction;
import java.io.IOException;
import java.text.DateFormat;
@@ -38,11 +38,13 @@ import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.IterConfigUtil;
import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.file.FileOperations;
import org.apache.accumulo.core.file.FileSKVIterator;
import org.apache.accumulo.core.file.FileSKVWriter;
+import org.apache.accumulo.core.iterators.IteratorEnvironment;
import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
import org.apache.accumulo.core.iteratorsImpl.system.ColumnFamilySkippingIterator;
@@ -52,7 +54,7 @@ import org.apache.accumulo.core.iteratorsImpl.system.TimeSettingIterator;
import org.apache.accumulo.core.metadata.StoredTabletFile;
import org.apache.accumulo.core.metadata.TabletFile;
import org.apache.accumulo.core.metadata.schema.DataFileValue;
-import org.apache.accumulo.core.spi.compaction.CompactionKind;
+import org.apache.accumulo.core.tabletserver.thrift.CompactionReason;
import org.apache.accumulo.core.util.LocalityGroupUtil;
import org.apache.accumulo.core.util.LocalityGroupUtil.LocalityGroupConfigurationError;
import org.apache.accumulo.core.util.ratelimit.RateLimiter;
@@ -62,9 +64,6 @@ import org.apache.accumulo.server.problems.ProblemReport;
import org.apache.accumulo.server.problems.ProblemReportingIterator;
import org.apache.accumulo.server.problems.ProblemReports;
import org.apache.accumulo.server.problems.ProblemType;
-import org.apache.accumulo.tserver.InMemoryMap;
-import org.apache.accumulo.tserver.MinorCompactionReason;
-import org.apache.accumulo.tserver.TabletIteratorEnvironment;
import org.apache.hadoop.fs.FileSystem;
import org.apache.htrace.Trace;
import org.apache.htrace.TraceScope;
@@ -88,10 +87,16 @@ public class Compactor implements Callable<CompactionStats> {
RateLimiter getReadLimiter();
RateLimiter getWriteLimiter();
+
+ IteratorEnvironment createIteratorEnv(ServerContext context, AccumuloConfiguration acuTableConf,
+ TableId tableId);
+
+ SortedKeyValueIterator<Key,Value> getMinCIterator();
+
+ CompactionReason getReason();
}
private final Map<StoredTabletFile,DataFileValue> filesToCompact;
- private final InMemoryMap imm;
private final TabletFile outputFile;
private final boolean propogateDeletes;
private final AccumuloConfiguration acuTableConf;
@@ -104,8 +109,6 @@ public class Compactor implements Callable<CompactionStats> {
private String currentLocalityGroup = "";
private final long startTime;
- private int reason;
-
private final AtomicLong entriesRead = new AtomicLong(0);
private final AtomicLong entriesWritten = new AtomicLong(0);
private final DateFormat dateFormatter = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss.SSS");
@@ -147,20 +150,18 @@ public class Compactor implements Callable<CompactionStats> {
return compactions;
}
- public Compactor(ServerContext context, Tablet tablet, Map<StoredTabletFile,DataFileValue> files,
- InMemoryMap imm, TabletFile outputFile, boolean propogateDeletes, CompactionEnv env,
- List<IteratorSetting> iterators, int reason, AccumuloConfiguration tableConfiguation) {
+ public Compactor(ServerContext context, KeyExtent extent,
+ Map<StoredTabletFile,DataFileValue> files, TabletFile outputFile, boolean propogateDeletes,
+ CompactionEnv env, List<IteratorSetting> iterators, AccumuloConfiguration tableConfiguation) {
this.context = context;
- this.extent = tablet.getExtent();
+ this.extent = extent;
this.fs = context.getVolumeManager();
this.acuTableConf = tableConfiguation;
this.filesToCompact = files;
- this.imm = imm;
this.outputFile = outputFile;
this.propogateDeletes = propogateDeletes;
this.env = env;
this.iterators = iterators;
- this.reason = reason;
startTime = System.currentTimeMillis();
}
@@ -169,18 +170,14 @@ public class Compactor implements Callable<CompactionStats> {
return fs;
}
- KeyExtent getExtent() {
+ public KeyExtent getExtent() {
return extent;
}
- String getOutputFile() {
+ protected String getOutputFile() {
return outputFile.toString();
}
- CompactionKind getMajorCompactionReason() {
- return CompactionKind.values()[reason];
- }
-
protected Map<String,Set<ByteSequence>> getLocalityGroups(AccumuloConfiguration acuTableConf)
throws IOException {
try {
@@ -343,8 +340,8 @@ public class Compactor implements Callable<CompactionStats> {
long entriesCompacted = 0;
List<SortedKeyValueIterator<Key,Value>> iters = openMapDataFiles(readers);
- if (imm != null) {
- iters.add(imm.compactionIterator());
+ if (env.getIteratorScope() == IteratorScope.minc) {
+ iters.add(env.getMinCIterator());
}
CountingIterator citr =
@@ -355,18 +352,11 @@ public class Compactor implements Callable<CompactionStats> {
// if(env.getIteratorScope() )
- TabletIteratorEnvironment iterEnv;
- if (env.getIteratorScope() == IteratorScope.majc)
- iterEnv = new TabletIteratorEnvironment(context, IteratorScope.majc, !propogateDeletes,
- acuTableConf, getExtent().tableId(), getMajorCompactionReason());
- else if (env.getIteratorScope() == IteratorScope.minc)
- iterEnv = new TabletIteratorEnvironment(context, IteratorScope.minc, acuTableConf,
- getExtent().tableId());
- else
- throw new IllegalArgumentException();
+ IteratorEnvironment iterEnv =
+ env.createIteratorEnv(context, acuTableConf, getExtent().tableId());
- SortedKeyValueIterator<Key,Value> itr = iterEnv.getTopLevelIterator(IterConfigUtil
- .convertItersAndLoad(env.getIteratorScope(), cfsi, acuTableConf, iterators, iterEnv));
+ SortedKeyValueIterator<Key,Value> itr = IterConfigUtil
+ .convertItersAndLoad(env.getIteratorScope(), cfsi, acuTableConf, iterators, iterEnv);
itr.seek(extent.toDataRange(), columnFamilies, inclusive);
@@ -425,7 +415,7 @@ public class Compactor implements Callable<CompactionStats> {
}
boolean hasIMM() {
- return imm != null;
+ return env.getIteratorScope() == IteratorScope.minc;
}
boolean willPropogateDeletes() {
@@ -448,8 +438,8 @@ public class Compactor implements Callable<CompactionStats> {
return this.iterators;
}
- MinorCompactionReason getMinCReason() {
- return MinorCompactionReason.values()[reason];
+ public CompactionReason getReason() {
+ return env.getReason();
}
}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CountingIterator.java b/server/base/src/main/java/org/apache/accumulo/server/compaction/CountingIterator.java
similarity index 98%
rename from server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CountingIterator.java
rename to server/base/src/main/java/org/apache/accumulo/server/compaction/CountingIterator.java
index 8656fd7..514ce0d 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CountingIterator.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/compaction/CountingIterator.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.accumulo.tserver.tablet;
+package org.apache.accumulo.server.compaction;
import java.io.IOException;
import java.util.ArrayList;
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/ThriftClientHandler.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/ThriftClientHandler.java
index 90a837f..fb108e9 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/ThriftClientHandler.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/ThriftClientHandler.java
@@ -118,6 +118,8 @@ import org.apache.accumulo.core.util.threads.Threads;
import org.apache.accumulo.fate.zookeeper.ZooLock;
import org.apache.accumulo.fate.zookeeper.ZooUtil;
import org.apache.accumulo.server.client.ClientServiceHandler;
+import org.apache.accumulo.server.compaction.CompactionInfo;
+import org.apache.accumulo.server.compaction.Compactor;
import org.apache.accumulo.server.conf.TableConfiguration;
import org.apache.accumulo.server.data.ServerMutation;
import org.apache.accumulo.server.fs.VolumeManager;
@@ -135,8 +137,6 @@ import org.apache.accumulo.tserver.session.SingleScanSession;
import org.apache.accumulo.tserver.session.SummarySession;
import org.apache.accumulo.tserver.session.UpdateSession;
import org.apache.accumulo.tserver.tablet.CommitSession;
-import org.apache.accumulo.tserver.tablet.CompactionInfo;
-import org.apache.accumulo.tserver.tablet.Compactor;
import org.apache.accumulo.tserver.tablet.KVEntry;
import org.apache.accumulo.tserver.tablet.PreparedMutations;
import org.apache.accumulo.tserver.tablet.ScanBatch;
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactableImpl.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactableImpl.java
index d0d8669..cd6e373 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactableImpl.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactableImpl.java
@@ -56,11 +56,12 @@ import org.apache.accumulo.core.spi.compaction.CompactionServices;
import org.apache.accumulo.core.util.compaction.CompactionJobImpl;
import org.apache.accumulo.core.util.ratelimit.RateLimiter;
import org.apache.accumulo.server.ServiceEnvironmentImpl;
+import org.apache.accumulo.server.compaction.CompactionStats;
+import org.apache.accumulo.server.compaction.Compactor.CompactionCanceledException;
import org.apache.accumulo.server.util.MetadataTableUtil;
import org.apache.accumulo.tserver.compactions.Compactable;
import org.apache.accumulo.tserver.compactions.CompactionManager;
import org.apache.accumulo.tserver.managermessage.TabletStatusMessage;
-import org.apache.accumulo.tserver.tablet.Compactor.CompactionCanceledException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactableUtils.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactableUtils.java
index c4de249..6a54c65 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactableUtils.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactableUtils.java
@@ -58,6 +58,7 @@ import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.file.FileOperations;
import org.apache.accumulo.core.file.FileSKVIterator;
+import org.apache.accumulo.core.iterators.IteratorEnvironment;
import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
import org.apache.accumulo.core.metadata.CompactableFileImpl;
@@ -72,11 +73,18 @@ import org.apache.accumulo.core.summary.Gatherer;
import org.apache.accumulo.core.summary.SummarizerFactory;
import org.apache.accumulo.core.summary.SummaryCollection;
import org.apache.accumulo.core.summary.SummaryReader;
+import org.apache.accumulo.core.tabletserver.thrift.CompactionReason;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.core.util.ratelimit.RateLimiter;
+import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.ServiceEnvironmentImpl;
+import org.apache.accumulo.server.compaction.CompactionStats;
+import org.apache.accumulo.server.compaction.Compactor;
+import org.apache.accumulo.server.compaction.Compactor.CompactionCanceledException;
+import org.apache.accumulo.server.compaction.Compactor.CompactionEnv;
import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.util.MetadataTableUtil;
+import org.apache.accumulo.tserver.TabletIteratorEnvironment;
import org.apache.accumulo.tserver.compaction.CompactionPlan;
import org.apache.accumulo.tserver.compaction.CompactionStrategy;
import org.apache.accumulo.tserver.compaction.MajorCompactionReason;
@@ -84,8 +92,6 @@ import org.apache.accumulo.tserver.compaction.MajorCompactionRequest;
import org.apache.accumulo.tserver.compaction.WriteParameters;
import org.apache.accumulo.tserver.tablet.CompactableImpl.CompactionCheck;
import org.apache.accumulo.tserver.tablet.CompactableImpl.CompactionHelper;
-import org.apache.accumulo.tserver.tablet.Compactor.CompactionCanceledException;
-import org.apache.accumulo.tserver.tablet.Compactor.CompactionEnv;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -555,9 +561,33 @@ public class CompactableUtils {
public RateLimiter getWriteLimiter() {
return writeLimiter;
}
- };
- int reason = job.getKind().ordinal();
+ @Override
+ public IteratorEnvironment createIteratorEnv(ServerContext context,
+ AccumuloConfiguration acuTableConf, TableId tableId) {
+ return new TabletIteratorEnvironment(context, IteratorScope.majc, !propogateDeletes,
+ acuTableConf, tableId, job.getKind());
+ }
+
+ @Override
+ public SortedKeyValueIterator<Key,Value> getMinCIterator() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompactionReason getReason() {
+ switch (job.getKind()) {
+ case USER:
+ return CompactionReason.USER;
+ case CHOP:
+ return CompactionReason.CHOP;
+ case SELECTOR:
+ case SYSTEM:
+ default:
+ return CompactionReason.SYSTEM;
+ }
+ }
+ };
AccumuloConfiguration tableConfig =
getCompactionConfig(job.getKind(), tablet, helper, job.getFiles());
@@ -569,8 +599,8 @@ public class CompactableUtils {
TabletFile newFile = tablet.getNextMapFilename(!propogateDeletes ? "A" : "C");
TabletFile compactTmpName = new TabletFile(new Path(newFile.getMetaInsert() + "_tmp"));
- Compactor compactor = new Compactor(tablet.getContext(), tablet, compactFiles, null,
- compactTmpName, propogateDeletes, cenv, iters, reason, tableConfig);
+ Compactor compactor = new Compactor(tablet.getContext(), tablet.getExtent(), compactFiles,
+ compactTmpName, propogateDeletes, cenv, iters, tableConfig);
var mcs = compactor.call();
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactionWatcher.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactionWatcher.java
index 497a343..8b665d7 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactionWatcher.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactionWatcher.java
@@ -29,6 +29,8 @@ import java.util.concurrent.TimeUnit;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.util.threads.ThreadPools;
+import org.apache.accumulo.server.compaction.CompactionInfo;
+import org.apache.accumulo.server.compaction.Compactor;
import org.slf4j.LoggerFactory;
public class CompactionWatcher implements Runnable {
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/MinorCompactor.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/MinorCompactor.java
index 6615c3d..ab04649 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/MinorCompactor.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/MinorCompactor.java
@@ -31,17 +31,27 @@ import java.util.concurrent.TimeUnit;
import org.apache.accumulo.core.clientImpl.Tables;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.data.ByteSequence;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.IteratorEnvironment;
import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
import org.apache.accumulo.core.manager.state.tables.TableState;
import org.apache.accumulo.core.metadata.TabletFile;
+import org.apache.accumulo.core.tabletserver.thrift.CompactionReason;
import org.apache.accumulo.core.util.LocalityGroupUtil;
import org.apache.accumulo.core.util.ratelimit.RateLimiter;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.compaction.CompactionStats;
+import org.apache.accumulo.server.compaction.Compactor;
import org.apache.accumulo.server.conf.TableConfiguration;
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.tserver.InMemoryMap;
import org.apache.accumulo.tserver.MinorCompactionReason;
+import org.apache.accumulo.tserver.TabletIteratorEnvironment;
import org.apache.accumulo.tserver.TabletServer;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
@@ -55,9 +65,8 @@ public class MinorCompactor extends Compactor {
public MinorCompactor(TabletServer tabletServer, Tablet tablet, InMemoryMap imm,
TabletFile outputFile, MinorCompactionReason mincReason, TableConfiguration tableConfig) {
- super(tabletServer.getContext(), tablet, Collections.emptyMap(), imm, outputFile, true,
+ super(tabletServer.getContext(), tablet.getExtent(), Collections.emptyMap(), outputFile, true,
new CompactionEnv() {
-
@Override
public boolean isCompactionEnabled() {
return true;
@@ -77,7 +86,32 @@ public class MinorCompactor extends Compactor {
public RateLimiter getWriteLimiter() {
return null;
}
- }, Collections.emptyList(), mincReason.ordinal(), tableConfig);
+
+ @Override
+ public IteratorEnvironment createIteratorEnv(ServerContext context,
+ AccumuloConfiguration acuTableConf, TableId tableId) {
+ return new TabletIteratorEnvironment(context, IteratorScope.minc, acuTableConf,
+ tableId);
+ }
+
+ @Override
+ public SortedKeyValueIterator<Key,Value> getMinCIterator() {
+ return imm.compactionIterator();
+ }
+
+ @Override
+ public CompactionReason getReason() {
+ switch (mincReason) {
+ case USER:
+ return CompactionReason.USER;
+ case CLOSE:
+ return CompactionReason.CLOSE;
+ case SYSTEM:
+ default:
+ return CompactionReason.SYSTEM;
+ }
+ }
+ }, Collections.emptyList(), tableConfig);
this.tabletServer = tabletServer;
}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
index fc8c6fe..6a83ea5 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
@@ -91,6 +91,7 @@ import org.apache.accumulo.core.util.ShutdownUtil;
import org.apache.accumulo.core.volume.Volume;
import org.apache.accumulo.server.ServerConstants;
import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.compaction.CompactionStats;
import org.apache.accumulo.server.conf.TableConfiguration;
import org.apache.accumulo.server.fs.VolumeChooserEnvironmentImpl;
import org.apache.accumulo.server.fs.VolumeUtil;
diff --git a/server/tserver/src/test/java/org/apache/accumulo/tserver/CountingIteratorTest.java b/server/tserver/src/test/java/org/apache/accumulo/tserver/CountingIteratorTest.java
index c67bedc..fd39310 100644
--- a/server/tserver/src/test/java/org/apache/accumulo/tserver/CountingIteratorTest.java
+++ b/server/tserver/src/test/java/org/apache/accumulo/tserver/CountingIteratorTest.java
@@ -29,7 +29,7 @@ import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.SortedMapIterator;
-import org.apache.accumulo.tserver.tablet.CountingIterator;
+import org.apache.accumulo.server.compaction.CountingIterator;
import org.junit.Test;
public class CountingIteratorTest {