keith-turner closed pull request #588: ACCUMULO-4629 add exact deletes
URL: https://github.com/apache/accumulo/pull/588
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/core/src/main/java/org/apache/accumulo/core/Constants.java
b/core/src/main/java/org/apache/accumulo/core/Constants.java
index f8cf8e270c..567d201359 100644
--- a/core/src/main/java/org/apache/accumulo/core/Constants.java
+++ b/core/src/main/java/org/apache/accumulo/core/Constants.java
@@ -41,6 +41,7 @@
public static final String ZTABLE_COMPACT_ID = "/compact-id";
public static final String ZTABLE_COMPACT_CANCEL_ID = "/compact-cancel-id";
public static final String ZTABLE_NAMESPACE = "/namespace";
+ public static final String ZTABLE_EXACT_DELETE = "/exact-delete";
public static final String ZNAMESPACES = "/namespaces";
public static final String ZNAMESPACE_NAME = "/name";
diff --git
a/core/src/main/java/org/apache/accumulo/core/client/admin/NewTableConfiguration.java
b/core/src/main/java/org/apache/accumulo/core/client/admin/NewTableConfiguration.java
index 9c97f54b0d..ac6180b7f2 100644
---
a/core/src/main/java/org/apache/accumulo/core/client/admin/NewTableConfiguration.java
+++
b/core/src/main/java/org/apache/accumulo/core/client/admin/NewTableConfiguration.java
@@ -32,6 +32,7 @@
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.impl.TableOperationsHelper;
+import org.apache.accumulo.core.client.rfile.RFile.ScannerOptions;
import org.apache.accumulo.core.client.sample.SamplerConfiguration;
import org.apache.accumulo.core.client.summary.Summarizer;
import org.apache.accumulo.core.client.summary.SummarizerConfiguration;
@@ -56,6 +57,7 @@
private TimeType timeType = DEFAULT_TIME_TYPE;
private boolean limitVersion = true;
+ private boolean exactDelete = false;
private Map<String,String> properties = Collections.emptyMap();
private Map<String,String> samplerProps = Collections.emptyMap();
@@ -156,6 +158,34 @@ public NewTableConfiguration
enableSampling(SamplerConfiguration samplerConfigur
return this;
}
+ /**
+ * This setting determines how deletes are interpreted for a table. When
this setting is false,
+ * which is the default, deletes hide everything in a column where the
timestamp is less than or
+ * equal to the delete. When this setting is true, only versions in a column
with the same
+ * timestamp as a delete are hidden.
+ *
+ * <p>
+ * The Accumulo user manual contains a section about this feature.
+ *
+ * @see TableOperations#isExactDeleteEnabled(String)
+ * @see ScannerOptions#withExactDeletes()
+ * @since 2.0.0
+ * @return this
+ */
+ public NewTableConfiguration setExactDeleteEnabled(boolean b) {
+ exactDelete = b;
+ return this;
+ }
+
+ /**
+ * @since 2.0.0
+ * @return The value previously passed to {@link
#setExactDeleteEnabled(boolean)}. If
+ * {@link #setExactDeleteEnabled(boolean)} was never called then
returns false.
+ */
+ public boolean isExactDeleteEnabled() {
+ return exactDelete;
+ }
+
/**
* Enables creating summary statistics using {@link Summarizer}'s for the
new table.
*
diff --git
a/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperations.java
b/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperations.java
index 894b2fc60c..5129634cdb 100644
---
a/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperations.java
+++
b/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperations.java
@@ -1015,4 +1015,14 @@ void removeSummarizers(String tableName,
Predicate<SummarizerConfiguration> pred
*/
List<SummarizerConfiguration> listSummarizers(String tableName)
throws AccumuloException, TableNotFoundException,
AccumuloSecurityException;
+
+ /**
+ * Determines if exact deletes were enabled for a table when it was created.
By default exact
+ * deletes are not enabled.
+ *
+ * @see NewTableConfiguration#setExactDeleteEnabled(boolean)
+ * @since 2.0.0
+ */
+ public boolean isExactDeleteEnabled(String tableName)
+ throws AccumuloException, TableNotFoundException,
AccumuloSecurityException;
}
diff --git
a/core/src/main/java/org/apache/accumulo/core/client/impl/OfflineIterator.java
b/core/src/main/java/org/apache/accumulo/core/client/impl/OfflineIterator.java
index af5626d66c..a045f54c44 100644
---
a/core/src/main/java/org/apache/accumulo/core/client/impl/OfflineIterator.java
+++
b/core/src/main/java/org/apache/accumulo/core/client/impl/OfflineIterator.java
@@ -158,12 +158,14 @@ public IteratorEnvironment cloneWithSamplingEnabled() {
private ScannerOptions options;
private ArrayList<SortedKeyValueIterator<Key,Value>> readers;
private AccumuloConfiguration config;
+ private boolean exactDeletes;
public OfflineIterator(ScannerOptions options, ClientContext context,
- Authorizations authorizations, Text table, Range range) {
+ Authorizations authorizations, Text table, Range range, boolean
exactDeletes) {
this.options = new ScannerOptions(options);
this.context = context;
this.range = range;
+ this.exactDeletes = exactDeletes;
if (this.options.fetchedColumns.size() > 0) {
this.range = range.bound(this.options.fetchedColumns.first(),
@@ -383,7 +385,7 @@ private void nextTablet() throws TableNotFoundException,
AccumuloException, IOEx
defaultSecurityLabel = cv.getExpression();
SortedKeyValueIterator<Key,Value> visFilter =
IteratorUtil.setupSystemScanIterators(multiIter,
- new HashSet<>(options.fetchedColumns), authorizations,
defaultSecurityLabel);
+ new HashSet<>(options.fetchedColumns), authorizations,
defaultSecurityLabel, exactDeletes);
return iterEnv.getTopLevelIterator(
IteratorUtil.loadIterators(IteratorScope.scan, visFilter, extent,
acuTableConf,
diff --git
a/core/src/main/java/org/apache/accumulo/core/client/impl/OfflineScanner.java
b/core/src/main/java/org/apache/accumulo/core/client/impl/OfflineScanner.java
index d3229d2778..2eb0a6cb5a 100644
---
a/core/src/main/java/org/apache/accumulo/core/client/impl/OfflineScanner.java
+++
b/core/src/main/java/org/apache/accumulo/core/client/impl/OfflineScanner.java
@@ -38,6 +38,7 @@
private ClientContext context;
private Authorizations authorizations;
private Text tableId;
+ private boolean exactDeletes;
public OfflineScanner(ClientContext context, Table.ID tableId,
Authorizations authorizations) {
checkArgument(context != null, "context is null");
@@ -49,6 +50,7 @@ public OfflineScanner(ClientContext context, Table.ID
tableId, Authorizations au
this.authorizations = authorizations;
this.batchSize = Constants.SCAN_BATCH_SIZE;
this.timeOut = Integer.MAX_VALUE;
+ this.exactDeletes = Tables.isExactDeleteEnabled(context, tableId);
}
@Deprecated
@@ -95,7 +97,7 @@ public void disableIsolation() {
@Override
public Iterator<Entry<Key,Value>> iterator() {
- return new OfflineIterator(this, context, authorizations, tableId, range);
+ return new OfflineIterator(this, context, authorizations, tableId, range,
exactDeletes);
}
@Override
diff --git
a/core/src/main/java/org/apache/accumulo/core/client/impl/TableOperationsImpl.java
b/core/src/main/java/org/apache/accumulo/core/client/impl/TableOperationsImpl.java
index 2ca0ad6f98..b098f919ad 100644
---
a/core/src/main/java/org/apache/accumulo/core/client/impl/TableOperationsImpl.java
+++
b/core/src/main/java/org/apache/accumulo/core/client/impl/TableOperationsImpl.java
@@ -235,7 +235,8 @@ public void create(String tableName, NewTableConfiguration
ntc)
checkArgument(ntc != null, "ntc is null");
List<ByteBuffer> args =
Arrays.asList(ByteBuffer.wrap(tableName.getBytes(UTF_8)),
- ByteBuffer.wrap(ntc.getTimeType().name().getBytes(UTF_8)));
+ ByteBuffer.wrap(ntc.getTimeType().name().getBytes(UTF_8)),
+
ByteBuffer.wrap(Boolean.toString(ntc.isExactDeleteEnabled()).getBytes(UTF_8)));
Map<String,String> opts = ntc.getProperties();
@@ -1929,4 +1930,14 @@ public void removeSummarizers(String tableName,
Predicate<SummarizerConfiguratio
public ImportSourceArguments addFilesTo(String tableName) {
return new BulkImport(tableName, context);
}
+
+ @Override
+ public boolean isExactDeleteEnabled(String tableName) throws
TableNotFoundException {
+ checkArgument(tableName != null, "tableName is null");
+
+ Table.ID tableId = Tables.getTableId(context, tableName);
+
+ return Tables.isExactDeleteEnabled(context, tableId);
+
+ }
}
diff --git
a/core/src/main/java/org/apache/accumulo/core/client/impl/Tables.java
b/core/src/main/java/org/apache/accumulo/core/client/impl/Tables.java
index ddf77d8e5a..c1abcbb3ac 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/Tables.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/Tables.java
@@ -237,6 +237,16 @@ public static TableState getTableState(ClientContext
context, Table.ID tableId,
}
+ public static boolean isExactDeleteEnabled(ClientContext context, Table.ID
tableId) {
+ String path = ZooUtil.getRoot(context.getInstanceID()) + Constants.ZTABLES
+ "/"
+ + tableId.canonicalID() + Constants.ZTABLE_EXACT_DELETE;
+ ZooCache zc = getZooCache(context);
+
+ byte[] val = zc.get(path);
+
+ return Boolean.parseBoolean(new String(val, UTF_8));
+ }
+
public static String qualified(String tableName) {
return qualified(tableName, Namespace.DEFAULT);
}
diff --git
a/core/src/main/java/org/apache/accumulo/core/client/rfile/RFile.java
b/core/src/main/java/org/apache/accumulo/core/client/rfile/RFile.java
index cc5f23ef6a..be31e71a93 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/rfile/RFile.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/rfile/RFile.java
@@ -25,6 +25,7 @@
import java.util.function.Predicate;
import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.admin.NewTableConfiguration;
import org.apache.accumulo.core.client.admin.TableOperations;
import org.apache.accumulo.core.client.sample.SamplerConfiguration;
import org.apache.accumulo.core.client.summary.Summarizer;
@@ -169,6 +170,18 @@
*/
ScannerOptions withBounds(Range range);
+ /**
+ * This setting determines how delete marker are interpreted. By default
deletes hide everything
+ * in a column where the timestamp is less then or equal to the delete.
With this setting only
+ * versions of the column with the same timestamp as a delete are hidden.
+ *
+ * @see NewTableConfiguration#setExactDeleteEnabled(boolean)
+ * @see TableOperations#isExactDeleteEnabled(String)
+ * @since 2.0.0
+ * @return this
+ */
+ ScannerOptions withExactDeletes();
+
/**
* Construct the {@link Scanner} with iterators specified in a tables
properties. Properties for
* a table can be obtained by calling {@link
TableOperations#getProperties(String)}
diff --git
a/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScanner.java
b/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScanner.java
index bece0a1155..edf8296178 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScanner.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScanner.java
@@ -89,6 +89,7 @@
boolean useSystemIterators = true;
public HashMap<String,String> tableConfig;
Range bounds;
+ public boolean exactDeletes = false;
}
// This cache exist as a hack to avoid leaking decompressors. When the RFile
code is not given a
@@ -180,6 +181,11 @@ public void indexWeightChanged() {}
"Set authorizations and specified not to use system iterators");
}
+ if (opts.exactDeletes && !opts.useSystemIterators) {
+ throw new IllegalArgumentException(
+ "Requested exact deletes and specified not to use system iterators");
+ }
+
this.opts = opts;
if (null != opts.tableConfig && opts.tableConfig.size() > 0) {
ConfigurationCopy tableCC = new
ConfigurationCopy(DefaultConfiguration.getInstance());
@@ -377,7 +383,7 @@ public SamplerConfiguration getSamplerConfiguration() {
SortedSet<Column> cols = this.getFetchedColumns();
families = LocalityGroupUtil.families(cols);
iterator = IteratorUtil.setupSystemScanIterators(iterator, cols,
getAuthorizations(),
- EMPTY_BYTES);
+ EMPTY_BYTES, opts.exactDeletes);
}
try {
diff --git
a/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScannerBuilder.java
b/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScannerBuilder.java
index 769a199988..8b8dbf484a 100644
---
a/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScannerBuilder.java
+++
b/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScannerBuilder.java
@@ -149,4 +149,10 @@ public ScannerOptions withBounds(Range range) {
this.opts.bounds = range;
return this;
}
+
+ @Override
+ public ScannerOptions withExactDeletes() {
+ this.opts.exactDeletes = true;
+ return this;
+ }
}
diff --git
a/core/src/main/java/org/apache/accumulo/core/constraints/NoTimestampDeleteConstraint.java
b/core/src/main/java/org/apache/accumulo/core/constraints/NoTimestampDeleteConstraint.java
new file mode 100644
index 0000000000..11424f3704
--- /dev/null
+++
b/core/src/main/java/org/apache/accumulo/core/constraints/NoTimestampDeleteConstraint.java
@@ -0,0 +1,58 @@
+/*
+ * 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.core.constraints;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.accumulo.core.client.admin.NewTableConfiguration;
+import org.apache.accumulo.core.data.ColumnUpdate;
+import org.apache.accumulo.core.data.Mutation;
+
+/**
+ * This constraint ensures that all deletes in a mutation must specify a
timestamp.
+ *
+ * <p>
+ * This constraint is useful in conjunction with exact deletes.
+ *
+ * @see NewTableConfiguration#setExactDeleteEnabled(boolean)
+ * @since 2.0.0
+ */
+public class NoTimestampDeleteConstraint implements Constraint {
+
+ @Override
+ public String getViolationDescription(short violationCode) {
+ if (violationCode == 1) {
+ return "Delete did not specify a timestamp";
+ }
+ return null;
+ }
+
+ @Override
+ public List<Short> check(Environment env, Mutation mutation) {
+ List<ColumnUpdate> updates = mutation.getUpdates();
+ for (ColumnUpdate update : updates) {
+ if (!update.hasTimestamp()) {
+ return Collections.singletonList((short) 1);
+ }
+ }
+
+ return null;
+ }
+
+}
diff --git
a/core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
b/core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
index d6e904c208..9343d3d5a5 100644
--- a/core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
+++ b/core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
@@ -449,8 +449,8 @@ public static IteratorConfig
toIteratorConfig(List<IteratorSetting> iterators) {
public static SortedKeyValueIterator<Key,Value> setupSystemScanIterators(
SortedKeyValueIterator<Key,Value> source, Set<Column> cols,
Authorizations auths,
- byte[] defaultVisibility) throws IOException {
- DeletingIterator delIter = new DeletingIterator(source, false);
+ byte[] defaultVisibility, boolean exactDeletes) throws IOException {
+ SortedKeyValueIterator<Key,Value> delIter = DeletingIterator.wrap(source,
false, exactDeletes);
ColumnFamilySkippingIterator cfsi = new
ColumnFamilySkippingIterator(delIter);
SortedKeyValueIterator<Key,Value> colFilter =
ColumnQualifierFilter.wrap(cfsi, cols);
return VisibilityFilter.wrap(colFilter, auths, defaultVisibility);
diff --git
a/core/src/main/java/org/apache/accumulo/core/iterators/system/DeletingIterator.java
b/core/src/main/java/org/apache/accumulo/core/iterators/system/DeletingIterator.java
index edb28f5508..8c0c49cdcf 100644
---
a/core/src/main/java/org/apache/accumulo/core/iterators/system/DeletingIterator.java
+++
b/core/src/main/java/org/apache/accumulo/core/iterators/system/DeletingIterator.java
@@ -30,6 +30,8 @@
import org.apache.accumulo.core.iterators.ServerWrappingIterator;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import com.google.common.annotations.VisibleForTesting;
+
public class DeletingIterator extends ServerWrappingIterator {
private boolean propogateDeletes;
private Key workKey = new Key();
@@ -39,13 +41,13 @@ public DeletingIterator deepCopy(IteratorEnvironment env) {
return new DeletingIterator(this, env);
}
- public DeletingIterator(DeletingIterator other, IteratorEnvironment env) {
+ private DeletingIterator(DeletingIterator other, IteratorEnvironment env) {
super(other.source.deepCopy(env));
propogateDeletes = other.propogateDeletes;
}
- public DeletingIterator(SortedKeyValueIterator<Key,Value> iterator, boolean
propogateDeletes)
- throws IOException {
+ @VisibleForTesting
+ DeletingIterator(SortedKeyValueIterator<Key,Value> iterator, boolean
propogateDeletes) {
super(iterator);
this.propogateDeletes = propogateDeletes;
}
@@ -105,4 +107,13 @@ public void init(SortedKeyValueIterator<Key,Value> source,
Map<String,String> op
IteratorEnvironment env) {
throw new UnsupportedOperationException();
}
+
+ public static SortedKeyValueIterator<Key,Value>
wrap(SortedKeyValueIterator<Key,Value> iterator,
+ boolean propogateDeletes, boolean exactDeletes) {
+ if (exactDeletes) {
+ return new ExactDeletingIterator(iterator, propogateDeletes);
+ } else {
+ return new DeletingIterator(iterator, propogateDeletes);
+ }
+ }
}
diff --git
a/core/src/main/java/org/apache/accumulo/core/iterators/system/ExactDeletingIterator.java
b/core/src/main/java/org/apache/accumulo/core/iterators/system/ExactDeletingIterator.java
new file mode 100644
index 0000000000..9b6451e239
--- /dev/null
+++
b/core/src/main/java/org/apache/accumulo/core/iterators/system/ExactDeletingIterator.java
@@ -0,0 +1,116 @@
+/*
+ * 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.core.iterators.system;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Map;
+
+import org.apache.accumulo.core.data.ByteSequence;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.PartialKey;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.IteratorEnvironment;
+import org.apache.accumulo.core.iterators.ServerWrappingIterator;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+
+public class ExactDeletingIterator extends ServerWrappingIterator {
+ private boolean propogateDeletes;
+ private Key workKey = new Key();
+
+ @Override
+ public ExactDeletingIterator deepCopy(IteratorEnvironment env) {
+ return new ExactDeletingIterator(this, env);
+ }
+
+ public ExactDeletingIterator(ExactDeletingIterator other,
IteratorEnvironment env) {
+ super(other.source.deepCopy(env));
+ propogateDeletes = other.propogateDeletes;
+ }
+
+ public ExactDeletingIterator(SortedKeyValueIterator<Key,Value> iterator,
+ boolean propogateDeletes) {
+ super(iterator);
+ this.propogateDeletes = propogateDeletes;
+ }
+
+ @Override
+ public void next() throws IOException {
+ if (source.getTopKey().isDeleted())
+ skipRowColumnTime();
+ else
+ source.next();
+ findTop();
+ }
+
+ private static Range adjustRange(Range range) {
+ Range seekRange = range;
+
+ if (range.getStartKey() != null
+ && (range.isStartKeyInclusive() ^ range.getStartKey().isDeleted())) {
+ Key seekKey = new Key(seekRange.getStartKey());
+ seekKey.setDeleted(true);
+ seekRange = new Range(seekKey, true, range.getEndKey(),
range.isEndKeyInclusive());
+ }
+
+ return seekRange;
+ }
+
+ @Override
+ public void seek(Range range, Collection<ByteSequence> columnFamilies,
boolean inclusive)
+ throws IOException {
+ // do not want to seek to the middle of a row
+ Range seekRange = adjustRange(range);
+
+ source.seek(seekRange, columnFamilies, inclusive);
+ findTop();
+
+ if (propogateDeletes && range != seekRange) {
+ while (source.hasTop() && range.beforeStartKey(source.getTopKey())) {
+ next();
+ }
+ }
+ }
+
+ private void findTop() throws IOException {
+ if (!propogateDeletes) {
+ while (source.hasTop() && source.getTopKey().isDeleted()) {
+ skipRowColumnTime();
+ }
+ }
+ }
+
+ private void skipRowColumnTime() throws IOException {
+ workKey.set(source.getTopKey());
+
+ Key keyToSkip = workKey;
+ source.next();
+
+ while (source.hasTop()
+ && source.getTopKey().equals(keyToSkip,
PartialKey.ROW_COLFAM_COLQUAL_COLVIS_TIME)) {
+ source.next();
+ }
+ }
+
+ @Override
+ public void init(SortedKeyValueIterator<Key,Value> source,
Map<String,String> options,
+ IteratorEnvironment env) {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git
a/core/src/test/java/org/apache/accumulo/core/client/impl/TableOperationsHelperTest.java
b/core/src/test/java/org/apache/accumulo/core/client/impl/TableOperationsHelperTest.java
index 799c894131..8893a20711 100644
---
a/core/src/test/java/org/apache/accumulo/core/client/impl/TableOperationsHelperTest.java
+++
b/core/src/test/java/org/apache/accumulo/core/client/impl/TableOperationsHelperTest.java
@@ -309,6 +309,12 @@ public void removeSummarizers(String tableName,
Predicate<SummarizerConfiguratio
throws AccumuloException, TableNotFoundException,
AccumuloSecurityException {
throw new UnsupportedOperationException();
}
+
+ @Override
+ public boolean isExactDeleteEnabled(String tableName)
+ throws AccumuloException, TableNotFoundException,
AccumuloSecurityException {
+ throw new UnsupportedOperationException();
+ }
}
protected TableOperationsHelper getHelper() {
diff --git
a/core/src/test/java/org/apache/accumulo/core/client/rfile/RFileTest.java
b/core/src/test/java/org/apache/accumulo/core/client/rfile/RFileTest.java
index 30f9df6467..e86ed1787e 100644
--- a/core/src/test/java/org/apache/accumulo/core/client/rfile/RFileTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/client/rfile/RFileTest.java
@@ -346,6 +346,12 @@ public void testAuths() throws Exception {
Assert.assertEquals(ImmutableMap.of(k3, v3), toMap(scanner));
Assert.assertEquals(new Authorizations("B"), scanner.getAuthorizations());
scanner.close();
+
+ try {
+ RFile.newScanner().from(testFile).withFileSystem(localFs)
+ .withAuthorizations(new
Authorizations("B")).withoutSystemIterators().build();
+ Assert.fail();
+ } catch (IllegalArgumentException e) {}
}
@Test
@@ -838,4 +844,65 @@ public void testMultipleFilesAndCache() throws Exception {
Assert.assertEquals(testData, toMap(scanner));
scanner.close();
}
+
+ @Test
+ public void testDeletes() throws Exception {
+ LocalFileSystem localFs = FileSystem.getLocal(new Configuration());
+ String testFile = createTmpTestFile();
+
+ SortedMap<Key,Value> testData = new TreeMap<>();
+
+ testData.put(new Key("r1", "fam1", "q1", 5L), new Value("v1"));
+ testData.put(new Key("r1", "fam1", "q1", 6L), new Value("v2"));
+ testData.put(new Key("r1", "fam1", "q1", 7L), new Value("v3"));
+ testData.put(new Key("r1", "fam1", "q1", 8L), new Value("v4"));
+ testData.put(new Key("r1", "fam1", "q1", 9L), new Value("v5"));
+
+ SortedMap<Key,Value> deletes = new TreeMap<>();
+ Key k = new Key("r1", "fam1", "q1", 8L);
+ k.setDeleted(true);
+ deletes.put(k, new Value(""));
+ k = new Key("r1", "fam1", "q1", 6L);
+ k.setDeleted(true);
+ deletes.put(k, new Value(""));
+
+ try (RFileWriter writer =
RFile.newWriter().to(testFile).withFileSystem(localFs).build()) {
+ SortedMap<Key,Value> writeData = new TreeMap<>();
+ writeData.putAll(testData);
+ writeData.putAll(deletes);
+ writer.append(writeData.entrySet());
+ }
+
+ try (Scanner scanner =
RFile.newScanner().from(testFile).withFileSystem(localFs)
+ .withExactDeletes().build()) {
+ SortedMap<Key,Value> expected = new TreeMap<>();
+ expected.putAll(testData);
+ expected.remove(new Key("r1", "fam1", "q1", 8L));
+ expected.remove(new Key("r1", "fam1", "q1", 6L));
+
+ Assert.assertEquals(expected, toMap(scanner));
+ }
+
+ try (Scanner scanner =
RFile.newScanner().from(testFile).withFileSystem(localFs).build()) {
+ SortedMap<Key,Value> expected = new TreeMap<>();
+ expected.put(new Key("r1", "fam1", "q1", 9L), new Value("v5"));
+
+ Assert.assertEquals(expected, toMap(scanner));
+ }
+
+ try (Scanner scanner =
RFile.newScanner().from(testFile).withFileSystem(localFs)
+ .withoutSystemIterators().build()) {
+ SortedMap<Key,Value> expected = new TreeMap<>();
+ expected.putAll(testData);
+ expected.putAll(deletes);
+
+ Assert.assertEquals(expected, toMap(scanner));
+ }
+
+ try {
+
RFile.newScanner().from(testFile).withFileSystem(localFs).withoutSystemIterators()
+ .withExactDeletes().build();
+ Assert.fail();
+ } catch (IllegalArgumentException e) {}
+ }
}
diff --git
a/core/src/test/java/org/apache/accumulo/core/iterators/system/ExactDeletingIteratorTest.java
b/core/src/test/java/org/apache/accumulo/core/iterators/system/ExactDeletingIteratorTest.java
new file mode 100644
index 0000000000..c8dd594e5d
--- /dev/null
+++
b/core/src/test/java/org/apache/accumulo/core/iterators/system/ExactDeletingIteratorTest.java
@@ -0,0 +1,152 @@
+/*
+ * 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.core.iterators.system;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.TreeMap;
+
+import org.apache.accumulo.core.data.ByteSequence;
+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.junit.Assert;
+import org.junit.Test;
+
+public class ExactDeletingIteratorTest {
+
+ private static final Collection<ByteSequence> EMPTY_COL_FAMS = new
ArrayList<>();
+
+ private Key nk(Object... entry) {
+ Key k = new Key((String) entry[0], (String) entry[1], (String) entry[2]);
+ k.setTimestamp((int) entry[3]);
+ k.setDeleted((boolean) entry[4]);
+ return k;
+ }
+
+ private Range nr(Key key, boolean inclusive) {
+ return new Range(key, inclusive, null, false);
+ }
+
+ private TreeMap<Key,Value> parse(Object[][] data) {
+ TreeMap<Key,Value> tm = new TreeMap<>();
+
+ for (Object[] entry : data) {
+ Key k = nk(entry);
+ Value v = new Value((String) entry[5]);
+ tm.put(k, v);
+ }
+
+ return tm;
+ }
+
+ private void runTest(Object[][] testData, Range range, boolean propDel,
Object[][] expectedData)
+ throws IOException {
+ ExactDeletingIterator iter = new ExactDeletingIterator(new
SortedMapIterator(parse(testData)),
+ propDel);
+ TreeMap<Key,Value> actual = new TreeMap<>();
+
+ iter.seek(range, EMPTY_COL_FAMS, false);
+
+ while (iter.hasTop()) {
+ actual.put(iter.getTopKey(), iter.getTopValue());
+ iter.next();
+ }
+
+ Assert.assertEquals(parse(expectedData), actual);
+ }
+
+ private void runTest(Object[][] testData, boolean propDel, Object[][]
expectedData)
+ throws IOException {
+ runTest(testData, new Range(), propDel, expectedData);
+ }
+
+ @Test
+ public void testEverything() throws Exception {
+ // @formatter:off
+ Object[][] data = {
+ {"r01", "f01", "q01", 7, true, ""},
+ {"r01", "f01", "q01", 6, false, "v01"},
+ {"r01", "f01", "q01", 5, true, ""},
+ {"r01", "f01", "q01", 5, false, "v02"},
+ {"r01", "f01", "q01", 4, false, "v03"},
+ {"r01", "f01", "q01", 3, true, ""},
+ {"r01", "f01", "q01", 3, false, "v04"},
+ {"r01", "f01", "q01", 2, true, ""},
+ {"r01", "f01", "q01", 2, false, "v05"},
+ {"r01", "f01", "q01", 1, false, "v06"}};
+
+
+ Object[][] expected1 = {
+ {"r01", "f01", "q01", 6, false, "v01"},
+ {"r01", "f01", "q01", 4, false, "v03"},
+ {"r01", "f01", "q01", 1, false, "v06"}};
+
+ Object[][] expected2 = {
+ {"r01", "f01", "q01", 7, true, ""},
+ {"r01", "f01", "q01", 6, false, "v01"},
+ {"r01", "f01", "q01", 5, true, ""},
+ {"r01", "f01", "q01", 4, false, "v03"},
+ {"r01", "f01", "q01", 3, true, ""},
+ {"r01", "f01", "q01", 2, true, ""},
+ {"r01", "f01", "q01", 1, false, "v06"}};
+
+ Object[][] expected3 = {
+ {"r01", "f01", "q01", 4, false, "v03"},
+ {"r01", "f01", "q01", 1, false, "v06"}};
+
+ Object[][] expected4 = {
+ {"r01", "f01", "q01", 5, true, ""},
+ {"r01", "f01", "q01", 4, false, "v03"},
+ {"r01", "f01", "q01", 3, true, ""},
+ {"r01", "f01", "q01", 2, true, ""},
+ {"r01", "f01", "q01", 1, false, "v06"}};
+
+ Object[][] expected5 = {
+ {"r01", "f01", "q01", 4, false, "v03"},
+ {"r01", "f01", "q01", 3, true, ""},
+ {"r01", "f01", "q01", 2, true, ""},
+ {"r01", "f01", "q01", 1, false, "v06"}};
+ // @formatter:on
+
+ runTest(data, false, expected1);
+ runTest(data, true, expected2);
+
+ runTest(expected1, true, expected1);
+ runTest(expected1, false, expected1);
+
+ runTest(expected2, false, expected1);
+ runTest(expected2, true, expected2);
+
+ runTest(data, nr(nk("r01", "f01", "q01", 5, false), false), false,
expected3);
+ runTest(data, nr(nk("r01", "f01", "q01", 5, false), true), false,
expected3);
+ runTest(data, nr(nk("r01", "f01", "q01", 5, true), false), false,
expected3);
+ runTest(data, nr(nk("r01", "f01", "q01", 5, false), true), false,
expected3);
+
+ runTest(data, nr(nk("r01", "f01", "q01", 4, false), true), false,
expected3);
+ runTest(data, nr(nk("r01", "f01", "q01", 4, true), false), false,
expected3);
+ runTest(data, nr(nk("r01", "f01", "q01", 4, true), true), false,
expected3);
+
+ runTest(data, nr(nk("r01", "f01", "q01", 5, true), true), true, expected4);
+ runTest(data, nr(nk("r01", "f01", "q01", 5, false), true), true,
expected5);
+ runTest(data, nr(nk("r01", "f01", "q01", 5, true), false), true,
expected5);
+ runTest(data, nr(nk("r01", "f01", "q01", 5, false), false), true,
expected5);
+ }
+}
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/ServerConstants.java
b/server/base/src/main/java/org/apache/accumulo/server/ServerConstants.java
index 51a54f9d77..a6b39dc9d3 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/ServerConstants.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/ServerConstants.java
@@ -47,6 +47,11 @@
*/
public static final Integer WIRE_VERSION = 3;
+ /**
+ * version (9) Added exact deletes for 2.0 which a zookeeper setting
impacting how deletes are
+ * interpreted. Also added summary data to rfiles.
+ */
+ public static final int EXACT_DELETES_AND_SUMMARIES = 9;
/**
* version (8) reflects changes to RFile index (ACCUMULO-1124) in version
1.8.0
*/
@@ -58,7 +63,7 @@
/**
* this is the current data version
*/
- public static final int DATA_VERSION = SHORTEN_RFILE_KEYS;
+ public static final int DATA_VERSION = EXACT_DELETES_AND_SUMMARIES;
/**
* version (6) reflects the addition of a separate root table
(ACCUMULO-1481) in version 1.6.0
*/
@@ -73,7 +78,8 @@
public static final int LOGGING_TO_HDFS = 4;
public static final BitSet CAN_UPGRADE = new BitSet();
static {
- for (int i : new int[] {DATA_VERSION, MOVE_TO_REPLICATION_TABLE,
MOVE_TO_ROOT_TABLE}) {
+ for (int i : new int[] {DATA_VERSION, SHORTEN_RFILE_KEYS,
MOVE_TO_REPLICATION_TABLE,
+ MOVE_TO_ROOT_TABLE}) {
CAN_UPGRADE.set(i);
}
}
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java
b/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java
index 82007c692f..0d9abf3312 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java
@@ -617,11 +617,11 @@ private static void initZooKeeper(Opts opts, String uuid,
String instanceNamePat
TableManager.prepareNewNamespaceState(uuid, Namespace.ID.ACCUMULO,
Namespace.ACCUMULO,
NodeExistsPolicy.FAIL);
TableManager.prepareNewTableState(uuid, RootTable.ID,
Namespace.ID.ACCUMULO, RootTable.NAME,
- TableState.ONLINE, NodeExistsPolicy.FAIL);
+ TableState.ONLINE, false, NodeExistsPolicy.FAIL);
TableManager.prepareNewTableState(uuid, MetadataTable.ID,
Namespace.ID.ACCUMULO,
- MetadataTable.NAME, TableState.ONLINE, NodeExistsPolicy.FAIL);
+ MetadataTable.NAME, TableState.ONLINE, false, NodeExistsPolicy.FAIL);
TableManager.prepareNewTableState(uuid, ReplicationTable.ID,
Namespace.ID.ACCUMULO,
- ReplicationTable.NAME, TableState.OFFLINE, NodeExistsPolicy.FAIL);
+ ReplicationTable.NAME, TableState.OFFLINE, false,
NodeExistsPolicy.FAIL);
zoo.putPersistentData(zkInstanceRoot + Constants.ZTSERVERS,
EMPTY_BYTE_ARRAY,
NodeExistsPolicy.FAIL);
zoo.putPersistentData(zkInstanceRoot + Constants.ZPROBLEMS,
EMPTY_BYTE_ARRAY,
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/tables/TableManager.java
b/server/base/src/main/java/org/apache/accumulo/server/tables/TableManager.java
index 20d1358227..3210d8e296 100644
---
a/server/base/src/main/java/org/apache/accumulo/server/tables/TableManager.java
+++
b/server/base/src/main/java/org/apache/accumulo/server/tables/TableManager.java
@@ -77,11 +77,11 @@ public static void prepareNewNamespaceState(String
instanceId, Namespace.ID name
}
public static void prepareNewTableState(String instanceId, Table.ID tableId,
- Namespace.ID namespaceId, String tableName, TableState state,
NodeExistsPolicy existsPolicy)
- throws KeeperException, InterruptedException {
+ Namespace.ID namespaceId, String tableName, TableState state, boolean
exactDelete,
+ NodeExistsPolicy existsPolicy) throws KeeperException,
InterruptedException {
// state gets created last
- log.debug("Creating ZooKeeper entries for new table {} (ID: {}) in
namespace (ID: {})",
- tableName, tableId, namespaceId);
+ log.debug("Creating ZooKeeper entries for new table {} (ID: {}) in
namespace (ID: {})"
+ + " exactDelete: {}", tableName, tableId, namespaceId, exactDelete);
Pair<String,String> qualifiedTableName = Tables.qualify(tableName);
tableName = qualifiedTableName.getSecond();
String zTablePath = Constants.ZROOT + "/" + instanceId + Constants.ZTABLES
+ "/" + tableId;
@@ -97,6 +97,9 @@ public static void prepareNewTableState(String instanceId,
Table.ID tableId,
zoo.putPersistentData(zTablePath + Constants.ZTABLE_COMPACT_CANCEL_ID,
ZERO_BYTE, existsPolicy);
zoo.putPersistentData(zTablePath + Constants.ZTABLE_STATE,
state.name().getBytes(UTF_8),
existsPolicy);
+ zoo.putPersistentData(zTablePath + Constants.ZTABLE_EXACT_DELETE,
+ Boolean.toString(exactDelete).getBytes(UTF_8), existsPolicy);
+
}
public synchronized static TableManager getInstance() {
@@ -226,16 +229,19 @@ public TableState updateTableStateCache(Table.ID tableId)
{
}
public void addTable(Table.ID tableId, Namespace.ID namespaceId, String
tableName,
- NodeExistsPolicy existsPolicy)
+ boolean exactDelete, NodeExistsPolicy existsPolicy)
throws KeeperException, InterruptedException, NamespaceNotFoundException
{
- prepareNewTableState(instanceID, tableId, namespaceId, tableName,
TableState.NEW, existsPolicy);
+ prepareNewTableState(instanceID, tableId, namespaceId, tableName,
TableState.NEW, exactDelete,
+ existsPolicy);
updateTableStateCache(tableId);
}
public void cloneTable(Table.ID srcTableId, Table.ID tableId, String
tableName,
Namespace.ID namespaceId, Map<String,String> propertiesToSet,
Set<String> propertiesToExclude,
- NodeExistsPolicy existsPolicy) throws KeeperException,
InterruptedException {
- prepareNewTableState(instanceID, tableId, namespaceId, tableName,
TableState.NEW, existsPolicy);
+ boolean exactDelete, NodeExistsPolicy existsPolicy)
+ throws KeeperException, InterruptedException {
+ prepareNewTableState(instanceID, tableId, namespaceId, tableName,
TableState.NEW, exactDelete,
+ existsPolicy);
String srcTablePath = Constants.ZROOT + "/" + instanceID +
Constants.ZTABLES + "/" + srcTableId
+ Constants.ZTABLE_CONF;
diff --git
a/server/master/src/main/java/org/apache/accumulo/master/FateServiceHandler.java
b/server/master/src/main/java/org/apache/accumulo/master/FateServiceHandler.java
index cda934c44b..94bc426ed9 100644
---
a/server/master/src/main/java/org/apache/accumulo/master/FateServiceHandler.java
+++
b/server/master/src/main/java/org/apache/accumulo/master/FateServiceHandler.java
@@ -145,6 +145,7 @@ public void executeFateOperation(TInfo tinfo, TCredentials
c, long opid, FateOpe
TableOperation tableOp = TableOperation.CREATE;
String tableName = validateTableNameArgument(arguments.get(0),
tableOp, NOT_SYSTEM);
TimeType timeType =
TimeType.valueOf(ByteBufferUtil.toString(arguments.get(1)));
+ boolean exactDelete =
Boolean.parseBoolean(ByteBufferUtil.toString(arguments.get(2)));
Namespace.ID namespaceId;
@@ -159,10 +160,8 @@ public void executeFateOperation(TInfo tinfo, TCredentials
c, long opid, FateOpe
if (!master.security.canCreateTable(c, tableName, namespaceId))
throw new ThriftSecurityException(c.getPrincipal(),
SecurityErrorCode.PERMISSION_DENIED);
- master.fate.seedTransaction(opid,
- new TraceRepo<>(
- new CreateTable(c.getPrincipal(), tableName, timeType,
options, namespaceId)),
- autoCleanup);
+ master.fate.seedTransaction(opid, new TraceRepo<>(new
CreateTable(c.getPrincipal(),
+ tableName, timeType, exactDelete, options, namespaceId)),
autoCleanup);
break;
}
diff --git a/server/master/src/main/java/org/apache/accumulo/master/Master.java
b/server/master/src/main/java/org/apache/accumulo/master/Master.java
index 308212b4f9..7bbcc32b5f 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/Master.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/Master.java
@@ -350,11 +350,17 @@ private void upgradeZookeeper() {
log.debug("Initializing recovery area.");
zoo.putPersistentData(zooRoot + Constants.ZRECOVERY, zero,
NodeExistsPolicy.SKIP);
+ final byte[] falseBytes = Boolean.toString(false).getBytes(UTF_8);
+
for (String id : zoo.getChildren(zooRoot + Constants.ZTABLES)) {
log.debug("Prepping table {} for compaction cancellations.", id);
zoo.putPersistentData(
zooRoot + Constants.ZTABLES + "/" + id +
Constants.ZTABLE_COMPACT_CANCEL_ID, zero,
NodeExistsPolicy.SKIP);
+
+ zoo.putPersistentData(
+ zooRoot + Constants.ZTABLES + "/" + id +
Constants.ZTABLE_EXACT_DELETE, falseBytes,
+ NodeExistsPolicy.SKIP);
}
@SuppressWarnings("deprecation")
@@ -410,13 +416,13 @@ private void upgradeZookeeper() {
// create replication table in zk
log.debug("Upgrade creating table {} (ID: {})", ReplicationTable.NAME,
ReplicationTable.ID);
TableManager.prepareNewTableState(getInstanceID(), ReplicationTable.ID,
- Namespace.ID.ACCUMULO, ReplicationTable.NAME, TableState.OFFLINE,
+ Namespace.ID.ACCUMULO, ReplicationTable.NAME, TableState.OFFLINE,
false,
NodeExistsPolicy.SKIP);
// create root table
log.debug("Upgrade creating table {} (ID: {})", RootTable.NAME,
RootTable.ID);
TableManager.prepareNewTableState(getInstanceID(), RootTable.ID,
Namespace.ID.ACCUMULO,
- RootTable.NAME, TableState.ONLINE, NodeExistsPolicy.SKIP);
+ RootTable.NAME, TableState.ONLINE, false, NodeExistsPolicy.SKIP);
Initialize.initSystemTablesConfig();
// ensure root user can flush root table
security.grantTablePermission(context.rpcCreds(),
security.getRootUsername(), RootTable.ID,
diff --git
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/CloneZookeeper.java
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/CloneZookeeper.java
index 480081a91a..43ab947f4f 100644
---
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/CloneZookeeper.java
+++
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/CloneZookeeper.java
@@ -31,12 +31,14 @@
private static final long serialVersionUID = 1L;
private CloneInfo cloneInfo;
+ private boolean exactDelete;
public CloneZookeeper(CloneInfo cloneInfo, ClientContext context)
throws NamespaceNotFoundException {
this.cloneInfo = cloneInfo;
this.cloneInfo.namespaceId = Namespaces.getNamespaceId(context,
Tables.qualify(this.cloneInfo.tableName).getFirst());
+ this.exactDelete = Tables.isExactDeleteEnabled(context,
cloneInfo.srcTableId);
}
@Override
@@ -59,7 +61,7 @@ public long isReady(long tid, Master environment) throws
Exception {
TableManager.getInstance().cloneTable(cloneInfo.srcTableId,
cloneInfo.tableId,
cloneInfo.tableName, cloneInfo.namespaceId,
cloneInfo.propertiesToSet,
- cloneInfo.propertiesToExclude, NodeExistsPolicy.OVERWRITE);
+ cloneInfo.propertiesToExclude, exactDelete,
NodeExistsPolicy.OVERWRITE);
Tables.clearCache(environment.getContext());
return new CloneMetadata(cloneInfo);
diff --git
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/CreateTable.java
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/CreateTable.java
index c3ca32d3b9..61359dd5b8 100644
---
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/CreateTable.java
+++
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/CreateTable.java
@@ -31,14 +31,16 @@
private TableInfo tableInfo;
- public CreateTable(String user, String tableName, TimeType timeType,
Map<String,String> props,
- Namespace.ID namespaceId) {
+ public CreateTable(String user, String tableName, TimeType timeType, boolean
exactDelete,
+ Map<String,String> props, Namespace.ID namespaceId) {
tableInfo = new TableInfo();
tableInfo.tableName = tableName;
tableInfo.timeType = TabletTime.getTimeID(timeType);
tableInfo.user = user;
tableInfo.props = props;
tableInfo.namespaceId = namespaceId;
+ tableInfo.exactDelete = exactDelete;
+
}
@Override
diff --git
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/ExportTable.java
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/ExportTable.java
index f9c28db4b9..619970248a 100644
---
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/ExportTable.java
+++
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/ExportTable.java
@@ -52,9 +52,12 @@ public void undo(long tid, Master env) throws Exception {
Utils.unreserveHdfsDirectory(new Path(tableInfo.exportDir).toString(),
tid);
}
- public static final int VERSION = 1;
+ public static final int VERSION_1 = 1;
+ public static final int VERSION = 2;
public static final String DATA_VERSION_PROP = "srcDataVersion";
public static final String EXPORT_VERSION_PROP = "exportVersion";
+ public static final String EXACT_DELETE_PROP = "exactDelete";
+
}
diff --git
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportPopulateZookeeper.java
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportPopulateZookeeper.java
index d2c4238976..656d9d502a 100644
---
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportPopulateZookeeper.java
+++
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportPopulateZookeeper.java
@@ -79,7 +79,7 @@ public long isReady(long tid, Master environment) throws
Exception {
String namespace = Tables.qualify(tableInfo.tableName).getFirst();
Namespace.ID namespaceId = Namespaces.getNamespaceId(env.getContext(),
namespace);
TableManager.getInstance().addTable(tableInfo.tableId, namespaceId,
tableInfo.tableName,
- NodeExistsPolicy.OVERWRITE);
+ tableInfo.exactDelete, NodeExistsPolicy.OVERWRITE);
Tables.clearCache(env.getContext());
} finally {
diff --git
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
index 479e61b774..ade953fce3 100644
---
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
+++
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
@@ -60,7 +60,7 @@ public long isReady(long tid, Master environment) throws
Exception {
@Override
public Repo<Master> call(long tid, Master env) throws Exception {
- checkVersions(env);
+ tableInfo.exactDelete = checkVersions(env);
// first step is to reserve a table id.. if the machine fails during this
step
// it is ok to retry... the only side effect is that a table id may not be
used
@@ -77,11 +77,13 @@ public long isReady(long tid, Master environment) throws
Exception {
}
}
- public void checkVersions(Master env) throws
AcceptableThriftTableOperationException {
+ public boolean checkVersions(Master env) throws
AcceptableThriftTableOperationException {
Path path = new Path(tableInfo.exportDir, Constants.EXPORT_FILE);
Integer exportVersion = null;
Integer dataVersion = null;
+ Boolean exactDelete = null;
+
try (ZipInputStream zis = new
ZipInputStream(env.getFileSystem().open(path))) {
ZipEntry zipEntry;
while ((zipEntry = zis.getNextEntry()) != null) {
@@ -94,8 +96,11 @@ public void checkVersions(Master env) throws
AcceptableThriftTableOperationExcep
exportVersion = Integer.parseInt(sa[1]);
} else if (sa[0].equals(ExportTable.DATA_VERSION_PROP)) {
dataVersion = Integer.parseInt(sa[1]);
+ } else if (sa[0].equals(ExportTable.EXACT_DELETE_PROP)) {
+ exactDelete = Boolean.valueOf(sa[1]);
}
}
+ in.close();
break;
}
}
@@ -115,6 +120,16 @@ public void checkVersions(Master env) throws
AcceptableThriftTableOperationExcep
throw new AcceptableThriftTableOperationException(null,
tableInfo.tableName,
TableOperation.IMPORT, TableOperationExceptionType.OTHER,
"Incompatible data version " + dataVersion);
+
+ if (exactDelete == null && exportVersion == ExportTable.VERSION_1)
+ exactDelete = false;
+
+ if (exactDelete == null)
+ throw new AcceptableThriftTableOperationException(null,
tableInfo.tableName,
+ TableOperation.IMPORT, TableOperationExceptionType.OTHER,
+ "Missing expected exact delete property");
+
+ return exactDelete;
}
@Override
diff --git
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportedTableInfo.java
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportedTableInfo.java
index 365f693acc..499a59a845 100644
---
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportedTableInfo.java
+++
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportedTableInfo.java
@@ -31,4 +31,5 @@
public Table.ID tableId;
public String importDir;
public Namespace.ID namespaceId;
+ public boolean exactDelete;
}
diff --git
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/PopulateZookeeper.java
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/PopulateZookeeper.java
index 5b5b9d37fd..9a9bf5c5c2 100644
---
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/PopulateZookeeper.java
+++
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/PopulateZookeeper.java
@@ -52,7 +52,7 @@ public long isReady(long tid, Master environment) throws
Exception {
TableOperation.CREATE);
TableManager.getInstance().addTable(tableInfo.tableId,
tableInfo.namespaceId,
- tableInfo.tableName, NodeExistsPolicy.OVERWRITE);
+ tableInfo.tableName, tableInfo.exactDelete,
NodeExistsPolicy.OVERWRITE);
for (Entry<String,String> entry : tableInfo.props.entrySet())
TablePropUtil.setTableProperty(tableInfo.tableId, entry.getKey(),
entry.getValue());
diff --git
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/TableInfo.java
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/TableInfo.java
index 8854e4fd26..7ec217ae92 100644
---
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/TableInfo.java
+++
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/TableInfo.java
@@ -35,4 +35,6 @@
public Map<String,String> props;
public String dir = null;
+
+ public boolean exactDelete;
}
diff --git
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/WriteExportFiles.java
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/WriteExportFiles.java
index d3f84717ef..879b6448d7 100644
---
a/server/master/src/main/java/org/apache/accumulo/master/tableOps/WriteExportFiles.java
+++
b/server/master/src/main/java/org/apache/accumulo/master/tableOps/WriteExportFiles.java
@@ -166,6 +166,8 @@ public static void exportTable(VolumeManager fs,
ServerContext context, String t
osw.append("srcTableID:" + tableID.canonicalID() + "\n");
osw.append(ExportTable.DATA_VERSION_PROP + ":" +
ServerConstants.DATA_VERSION + "\n");
osw.append("srcCodeVersion:" + Constants.VERSION + "\n");
+ osw.append(
+ ExportTable.EXACT_DELETE_PROP + ":" +
Tables.isExactDeleteEnabled(context, tableID));
osw.flush();
dataOut.flush();
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index 28ccd5cfe3..46b6bec774 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@ -72,6 +72,7 @@
import org.apache.accumulo.core.client.impl.Namespace;
import org.apache.accumulo.core.client.impl.ScannerImpl;
import org.apache.accumulo.core.client.impl.Table;
+import org.apache.accumulo.core.client.impl.Table.ID;
import org.apache.accumulo.core.client.impl.Tables;
import org.apache.accumulo.core.client.impl.TabletLocator;
import org.apache.accumulo.core.client.impl.TabletType;
@@ -276,6 +277,9 @@
import org.slf4j.LoggerFactory;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
public class TabletServer implements Runnable {
@@ -2406,6 +2410,24 @@ public void run() {
}
}
+ // This information is immutable per table, so cache it with guava instead
of using ZooCache.
+ // ZooCache would leave a watch active on the node which consumes server
side resources. There is
+ // no need to consume server side resources for something that is not
expected to change.
+ private LoadingCache<Table.ID,Boolean> exactDeletesCache =
CacheBuilder.newBuilder()
+ .maximumSize(10000).build(new CacheLoader<Table.ID,Boolean>() {
+ @Override
+ public Boolean load(ID tid) throws Exception {
+ String zTablePath = Constants.ZROOT + "/" + getInstanceID() +
Constants.ZTABLES + "/"
+ + tid + Constants.ZTABLE_EXACT_DELETE;
+ byte[] data = ZooReaderWriter.getInstance().getData(zTablePath,
null);
+ return Boolean.parseBoolean(new String(data, UTF_8));
+ }
+ });
+
+ boolean isExactDeleteEnabled(Table.ID tableId) {
+ return exactDeletesCache.getUnchecked(tableId);
+ }
+
protected class AssignmentHandler implements Runnable {
private final KeyExtent extent;
private final int retryAttempt;
@@ -2513,9 +2535,11 @@ public void run() {
getTableConfiguration(extent));
TabletData data;
if (extent.isRootTablet()) {
- data = new TabletData(fs, ZooReaderWriter.getInstance(),
getTableConfiguration(extent));
+ data = new TabletData(fs, ZooReaderWriter.getInstance(),
getTableConfiguration(extent),
+ isExactDeleteEnabled(extent.getTableId()));
} else {
- data = new TabletData(extent, fs,
tabletsKeyValues.entrySet().iterator());
+ data = new TabletData(extent, fs,
tabletsKeyValues.entrySet().iterator(),
+ isExactDeleteEnabled(extent.getTableId()));
}
tablet = new Tablet(TabletServer.this, extent, trm, data);
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Compactor.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Compactor.java
index 904af5fc94..9ae7d366dd 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Compactor.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Compactor.java
@@ -112,6 +112,7 @@
private final long compactorID = nextCompactorID.getAndIncrement();
protected volatile Thread thread;
private final ServerContext context;
+ private final boolean exactDeletes;
public long getCompactorID() {
return compactorID;
@@ -147,7 +148,8 @@ private void clearStats() {
public Compactor(ServerContext context, Tablet tablet,
Map<FileRef,DataFileValue> files,
InMemoryMap imm, FileRef outputFile, boolean propogateDeletes,
CompactionEnv env,
- List<IteratorSetting> iterators, int reason, AccumuloConfiguration
tableConfiguation) {
+ List<IteratorSetting> iterators, int reason, AccumuloConfiguration
tableConfiguation,
+ boolean exactDeletes) {
this.context = context;
this.extent = tablet.getExtent();
this.fs = tablet.getTabletServer().getFileSystem();
@@ -159,6 +161,7 @@ public Compactor(ServerContext context, Tablet tablet,
Map<FileRef,DataFileValue
this.env = env;
this.iterators = iterators;
this.reason = reason;
+ this.exactDeletes = exactDeletes;
startTime = System.currentTimeMillis();
}
@@ -344,7 +347,8 @@ private void compactLocalityGroup(String lgName,
Set<ByteSequence> columnFamilie
CountingIterator citr = new CountingIterator(new MultiIterator(iters,
extent.toDataRange()),
entriesRead);
- DeletingIterator delIter = new DeletingIterator(citr, propogateDeletes);
+ SortedKeyValueIterator<Key,Value> delIter = DeletingIterator.wrap(citr,
propogateDeletes,
+ exactDeletes);
ColumnFamilySkippingIterator cfsi = new
ColumnFamilySkippingIterator(delIter);
// if(env.getIteratorScope() )
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 c41430c0c4..88272b632c 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
@@ -81,7 +81,8 @@ public RateLimiter getReadLimiter() {
public RateLimiter getWriteLimiter() {
return null;
}
- }, Collections.emptyList(), mincReason.ordinal(), tableConfig);
+ }, Collections.emptyList(), mincReason.ordinal(), tableConfig,
+ tablet.isExactDeleteEnabled());
this.tabletServer = tabletServer;
}
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/ScanDataSource.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/ScanDataSource.java
index 0ddd362546..38e517c118 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/ScanDataSource.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/ScanDataSource.java
@@ -200,7 +200,7 @@ public boolean isCurrent() {
SortedKeyValueIterator<Key,Value> visFilter =
IteratorUtil.setupSystemScanIterators(
statsIterator, options.getColumnSet(), options.getAuthorizations(),
- options.getDefaultLabels());
+ options.getDefaultLabels(), tablet.isExactDeleteEnabled());
if (!loadIters) {
return visFilter;
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 5d1f4fd410..1a958341d8 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
@@ -259,6 +259,8 @@ public long getDataSourceDeletions() {
private final int logId;
+ private final boolean exactDeletes;
+
@Override
public int getLogId() {
return logId;
@@ -306,8 +308,8 @@ private void checkTabletDir() throws IOException {
protected Tablet(TabletTime tabletTime, String tabletDirectory, int logId,
Path location,
DatafileManager datafileManager, TabletServer tabletServer,
TabletResourceManager tabletResources, TabletMemory tabletMemory,
- TableConfiguration tableConfiguration, KeyExtent extent,
- ConfigurationObserver configObserver) {
+ TableConfiguration tableConfiguration, KeyExtent extent,
ConfigurationObserver configObserver,
+ boolean exactDeletes) {
this.tabletTime = tabletTime;
this.tabletDirectory = tabletDirectory;
this.logId = logId;
@@ -321,6 +323,7 @@ protected Tablet(TabletTime tabletTime, String
tabletDirectory, int logId, Path
this.extent = extent;
this.configObserver = configObserver;
this.splitCreationTime = 0;
+ this.exactDeletes = exactDeletes;
}
public Tablet(final TabletServer tabletServer, final KeyExtent extent,
@@ -337,6 +340,7 @@ public Tablet(final TabletServer tabletServer, final
KeyExtent extent,
this.tabletTime = TabletTime.getInstance(data.getTime());
this.persistedTime = tabletTime.getTime();
this.logId = tabletServer.createLogId();
+ this.exactDeletes = data.isExactDeleteEnabled();
TableConfiguration tblConf = tabletServer.getTableConfiguration(extent);
if (null == tblConf) {
@@ -2010,7 +2014,7 @@ public RateLimiter getWriteLimiter() {
boolean lastBatch = filesToCompact.isEmpty();
Compactor compactor = new Compactor(context, this, copy, null,
compactTmpName,
lastBatch ? propogateDeletes : true, cenv, compactionIterators,
reason.ordinal(),
- tableConf);
+ tableConf, exactDeletes);
CompactionStats mcs = compactor.call();
@@ -2326,9 +2330,9 @@ public boolean isMajorCompactionQueued() {
log.debug("TABLET_HIST {} split {} {}", extent, low, high);
newTablets.put(high, new TabletData(tabletDirectory, highDatafileSizes,
time, lastFlushID,
- lastCompactID, lastLocation, getBulkIngestedFiles()));
+ lastCompactID, lastLocation, getBulkIngestedFiles(), exactDeletes));
newTablets.put(low, new TabletData(lowDirectory, lowDatafileSizes, time,
lastFlushID,
- lastCompactID, lastLocation, getBulkIngestedFiles()));
+ lastCompactID, lastLocation, getBulkIngestedFiles(), exactDeletes));
long t2 = System.currentTimeMillis();
@@ -2894,4 +2898,8 @@ public void cleanupBulkLoadedFiles(Set<Long> tids) {
}
}
+ public boolean isExactDeleteEnabled() {
+ return exactDeletes;
+ }
+
}
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/TabletData.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/TabletData.java
index 7774875a6d..f4f1057bbd 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/TabletData.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/TabletData.java
@@ -78,9 +78,12 @@
private Map<Long,List<FileRef>> bulkImported = new HashMap<>();
private long splitTime = 0;
private String directory = null;
+ private final boolean exactDeletes;
// Read tablet data from metadata tables
- public TabletData(KeyExtent extent, VolumeManager fs,
Iterator<Entry<Key,Value>> entries) {
+ public TabletData(KeyExtent extent, VolumeManager fs,
Iterator<Entry<Key,Value>> entries,
+ boolean exactDeletes) {
+ this.exactDeletes = exactDeletes;
final Text family = new Text();
Text rowName = extent.getMetadataEntry();
while (entries.hasNext()) {
@@ -132,8 +135,9 @@ public TabletData(KeyExtent extent, VolumeManager fs,
Iterator<Entry<Key,Value>>
}
// Read basic root table metadata from zookeeper
- public TabletData(VolumeManager fs, ZooReader rdr, AccumuloConfiguration
conf)
- throws IOException {
+ public TabletData(VolumeManager fs, ZooReader rdr, AccumuloConfiguration
conf,
+ boolean exactDeletes) throws IOException {
+ this.exactDeletes = exactDeletes;
directory =
VolumeUtil.switchRootTableVolume(MetadataTableUtil.getRootTabletDir());
Path location = new Path(directory);
@@ -175,7 +179,8 @@ public TabletData(VolumeManager fs, ZooReader rdr,
AccumuloConfiguration conf)
// Data pulled from an existing tablet to make a split
public TabletData(String tabletDirectory, SortedMap<FileRef,DataFileValue>
highDatafileSizes,
String time, long lastFlushID, long lastCompactID, TServerInstance
lastLocation,
- Map<Long,List<FileRef>> bulkIngestedFiles) {
+ Map<Long,List<FileRef>> bulkIngestedFiles, boolean exactDeletes) {
+ this.exactDeletes = exactDeletes;
this.directory = tabletDirectory;
this.dataFiles = highDatafileSizes;
this.time = time;
@@ -229,4 +234,8 @@ public String getDirectory() {
public long getSplitTime() {
return splitTime;
}
+
+ public boolean isExactDeleteEnabled() {
+ return exactDeletes;
+ }
}
diff --git
a/server/tserver/src/test/java/org/apache/accumulo/tserver/tablet/TabletTest.java
b/server/tserver/src/test/java/org/apache/accumulo/tserver/tablet/TabletTest.java
index 1ca32478b5..02fc12ceec 100644
---
a/server/tserver/src/test/java/org/apache/accumulo/tserver/tablet/TabletTest.java
+++
b/server/tserver/src/test/java/org/apache/accumulo/tserver/tablet/TabletTest.java
@@ -53,7 +53,7 @@ public void correctValuesSetForProperties() {
ConfigurationObserver obs =
EasyMock.createMock(ConfigurationObserver.class);
Tablet tablet = new Tablet(time, "", 0, new Path("/foo"), dfm, tserver,
tserverResourceManager,
- tabletMemory, tableConf, extent, obs);
+ tabletMemory, tableConf, extent, obs, false);
long hdfsBlockSize = 10000L, blockSize = 5000L, indexBlockSize = 500L;
int replication = 5;
diff --git a/test/src/main/java/org/apache/accumulo/test/ImportExportIT.java
b/test/src/main/java/org/apache/accumulo/test/ImportExportIT.java
index b28a27a96f..1f7f6476f8 100644
--- a/test/src/main/java/org/apache/accumulo/test/ImportExportIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/ImportExportIT.java
@@ -23,12 +23,14 @@
import java.util.Arrays;
import java.util.Iterator;
import java.util.Map.Entry;
+import java.util.Random;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.admin.NewTableConfiguration;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
@@ -69,7 +71,12 @@ public void testExportImportThenScan() throws Exception {
String[] tableNames = getUniqueNames(2);
String srcTable = tableNames[0], destTable = tableNames[1];
- conn.tableOperations().create(srcTable);
+
+ Random rand = new Random();
+ NewTableConfiguration ntc = new NewTableConfiguration();
+ boolean enableExactDeletes = rand.nextBoolean();
+ ntc.setExactDeleteEnabled(enableExactDeletes);
+ conn.tableOperations().create(srcTable, ntc);
BatchWriter bw = conn.createBatchWriter(srcTable, new BatchWriterConfig());
for (int row = 0; row < 1000; row++) {
@@ -170,6 +177,8 @@ public void testExportImportThenScan() throws Exception {
// Online the original table before we verify equivalence
conn.tableOperations().online(srcTable, true);
+ Assert.assertEquals(enableExactDeletes,
conn.tableOperations().isExactDeleteEnabled(destTable));
+
verifyTableEquality(conn, srcTable, destTable);
}
diff --git
a/test/src/main/java/org/apache/accumulo/test/functional/ExactDeleteIT.java
b/test/src/main/java/org/apache/accumulo/test/functional/ExactDeleteIT.java
new file mode 100644
index 0000000000..9948245ff2
--- /dev/null
+++ b/test/src/main/java/org/apache/accumulo/test/functional/ExactDeleteIT.java
@@ -0,0 +1,235 @@
+/*
+ * 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.test.functional;
+
+import java.util.Collections;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.MutationsRejectedException;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.admin.CompactionConfig;
+import org.apache.accumulo.core.client.admin.NewTableConfiguration;
+import org.apache.accumulo.core.client.impl.ClientContext;
+import org.apache.accumulo.core.client.impl.OfflineScanner;
+import org.apache.accumulo.core.client.impl.Table.ID;
+import org.apache.accumulo.core.client.impl.Tables;
+import org.apache.accumulo.core.constraints.NoTimestampDeleteConstraint;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.harness.AccumuloClusterHarness;
+import org.junit.Assert;
+import org.junit.Test;
+
+import com.google.common.collect.ImmutableSet;
+
+public class ExactDeleteIT extends AccumuloClusterHarness {
+
+ void verifyVals(Scanner scanner, String... expectedArray) {
+ Set<String> actual = StreamSupport.stream(scanner.spliterator(), false)
+ .map(e -> e.getValue().toString()).collect(Collectors.toSet());
+ Set<String> expected = ImmutableSet.copyOf(expectedArray);
+
+ Assert.assertEquals(expected, actual);
+ }
+
+ @Test
+ public void testScanAndCompact() throws Exception {
+ Connector c = getConnector();
+ String tableName = getUniqueNames(1)[0];
+ NewTableConfiguration ntc = new NewTableConfiguration();
+ ntc.setExactDeleteEnabled(true);
+ ntc.withoutDefaultIterators();
+ c.tableOperations().create(tableName, ntc);
+
+ Assert.assertTrue(c.tableOperations().isExactDeleteEnabled(tableName));
+
+ writeInitialData(c, tableName);
+
+ Scanner scanner = c.createScanner(tableName, Authorizations.EMPTY);
+
+ verifyVals(scanner, "v0", "v1", "v2", "v3", "v10", "v11", "v12", "v13");
+
+ try (BatchWriter writer = c.createBatchWriter(tableName)) {
+ Mutation m = new Mutation("1234");
+ m.putDelete("f1", "q1", 2L);
+ writer.addMutation(m);
+
+ m = new Mutation("1235");
+ m.putDelete("f1", "q1", 1L);
+ writer.addMutation(m);
+
+ m = new Mutation("1235");
+ m.putDelete("f1", "q1", 3L);
+ writer.addMutation(m);
+ }
+
+ verifyVals(scanner, "v0", "v1", "v3", "v10", "v12");
+
+ c.tableOperations().flush(tableName, null, null, true);
+
+ verifyVals(scanner, "v0", "v1", "v3", "v10", "v12");
+
+ try (BatchWriter writer = c.createBatchWriter(tableName)) {
+ Mutation m = new Mutation("1234");
+ m.putDelete("f1", "q1", 3L);
+ writer.addMutation(m);
+ }
+
+ // should have a delete in memory for data in files
+ verifyVals(scanner, "v0", "v1", "v10", "v12");
+
+ c.tableOperations().offline(tableName, true);
+ // Should be two files at this point, with a delete in one file for an
entry in another file.
+ // Having these two files is important for testing offline scan and
compaction. Do not want a
+ // single file where deletes were processed.
+ ClientContext ctx = new ClientContext(c.info());
+ ID tid = Tables.getTableId(ctx, tableName);
+ OfflineScanner offlineScanner = new OfflineScanner(new
ClientContext(c.info()), tid,
+ Authorizations.EMPTY);
+ verifyVals(offlineScanner, "v0", "v1", "v10", "v12");
+ c.tableOperations().online(tableName, true);
+
+ c.tableOperations().compact(tableName, new
CompactionConfig().setFlush(true).setWait(true));
+
+ verifyVals(scanner, "v0", "v1", "v10", "v12");
+ }
+
+ private void writeInitialData(Connector c, String tableName)
+ throws MutationsRejectedException, TableNotFoundException {
+ try (BatchWriter writer = c.createBatchWriter(tableName)) {
+ for (long l = 0; l < 4; l++) {
+ Mutation m = new Mutation("1234");
+ m.put("f1", "q1", l, "v" + l);
+ writer.addMutation(m);
+
+ m = new Mutation("1235");
+ m.put("f1", "q1", l, "v" + (10 + l));
+ writer.addMutation(m);
+ }
+ }
+ }
+
+ @Test
+ public void testClone() throws Exception {
+ Connector c = getConnector();
+ String[] tableNames = getUniqueNames(4);
+ String tableName = tableNames[0];
+ String normalTable = tableNames[1];
+ NewTableConfiguration ntc = new NewTableConfiguration();
+ ntc.setExactDeleteEnabled(true);
+ ntc.withoutDefaultIterators();
+ c.tableOperations().create(tableName, ntc);
+
+ ntc = new NewTableConfiguration();
+ ntc.withoutDefaultIterators();
+ c.tableOperations().create(normalTable, ntc);
+
+ Assert.assertTrue(c.tableOperations().isExactDeleteEnabled(tableName));
+ Assert.assertFalse(c.tableOperations().isExactDeleteEnabled(normalTable));
+
+ writeInitialData(c, tableName);
+ writeInitialData(c, normalTable);
+
+ String cloneExactDel = tableNames[2];
+ String cloneNormal = tableNames[3];
+ c.tableOperations().clone(tableName, cloneExactDel, true,
Collections.emptyMap(),
+ Collections.emptySet());
+
+ c.tableOperations().clone(normalTable, cloneNormal, true,
Collections.emptyMap(),
+ Collections.emptySet());
+
+ Assert.assertTrue(c.tableOperations().isExactDeleteEnabled(cloneExactDel));
+ Assert.assertFalse(c.tableOperations().isExactDeleteEnabled(cloneNormal));
+
+ Scanner scanner1 = c.createScanner(cloneExactDel, Authorizations.EMPTY);
+ Scanner scanner2 = c.createScanner(cloneNormal, Authorizations.EMPTY);
+
+ verifyVals(scanner1, "v0", "v1", "v2", "v3", "v10", "v11", "v12", "v13");
+ verifyVals(scanner2, "v0", "v1", "v2", "v3", "v10", "v11", "v12", "v13");
+
+ for (String table : new String[] {cloneExactDel, cloneNormal}) {
+ try (BatchWriter writer = c.createBatchWriter(table)) {
+ Mutation m = new Mutation("1234");
+ m.putDelete("f1", "q1", 2L);
+ writer.addMutation(m);
+
+ m = new Mutation("1235");
+ m.putDelete("f1", "q1", 1L);
+ writer.addMutation(m);
+
+ m = new Mutation("1235");
+ m.putDelete("f1", "q1", 3L);
+ writer.addMutation(m);
+ }
+ }
+
+ verifyVals(scanner1, "v0", "v1", "v3", "v10", "v12");
+ verifyVals(scanner2, "v3");
+ }
+
+ @Test
+ public void testConstraint() throws Exception {
+ Connector c = getConnector();
+ String tableName = getUniqueNames(1)[0];
+ NewTableConfiguration ntc = new NewTableConfiguration();
+ ntc.setProperties(Collections.singletonMap("table.constraint.1",
+ NoTimestampDeleteConstraint.class.getName()));
+ ntc.setExactDeleteEnabled(true);
+ ntc.withoutDefaultIterators();
+ c.tableOperations().create(tableName, ntc);
+
+ // Following write should not be affected by constraint
+ writeInitialData(c, tableName);
+
+ Scanner scanner = c.createScanner(tableName, Authorizations.EMPTY);
+
+ verifyVals(scanner, "v0", "v1", "v2", "v3", "v10", "v11", "v12", "v13");
+
+ // Following deletes should not be affected by constraint
+ try (BatchWriter writer = c.createBatchWriter(tableName)) {
+ Mutation m = new Mutation("1234");
+ m.putDelete("f1", "q1", 2L);
+ writer.addMutation(m);
+
+ m = new Mutation("1235");
+ m.putDelete("f1", "q1", 1L);
+ writer.addMutation(m);
+ }
+
+ verifyVals(scanner, "v0", "v1", "v3", "v10", "v12", "v13");
+
+ BatchWriter writer = c.createBatchWriter(tableName);
+ try {
+ Mutation m = new Mutation("1234");
+ m.putDelete("f1", "q1");
+ writer.addMutation(m);
+ writer.close();
+ Assert.fail("Expected write to fail because of constraint");
+ } catch (MutationsRejectedException mre) {
+ Assert.assertEquals(1, mre.getConstraintViolationSummaries().size());
+ String desc =
mre.getConstraintViolationSummaries().get(0).getViolationDescription();
+ Assert.assertEquals("Delete did not specify a timestamp", desc);
+ }
+
+ verifyVals(scanner, "v0", "v1", "v3", "v10", "v12", "v13");
+ }
+}
diff --git
a/test/src/main/java/org/apache/accumulo/test/performance/scan/CollectTabletStats.java
b/test/src/main/java/org/apache/accumulo/test/performance/scan/CollectTabletStats.java
index dd8a6d0724..f91e211b16 100644
---
a/test/src/main/java/org/apache/accumulo/test/performance/scan/CollectTabletStats.java
+++
b/test/src/main/java/org/apache/accumulo/test/performance/scan/CollectTabletStats.java
@@ -439,7 +439,7 @@ private static void reportHdfsBlockLocations(List<FileRef>
files) throws Excepti
iters.add(smi);
MultiIterator multiIter = new MultiIterator(iters, ke);
- DeletingIterator delIter = new DeletingIterator(multiIter, false);
+ SortedKeyValueIterator<Key,Value> delIter =
DeletingIterator.wrap(multiIter, false, false);
ColumnFamilySkippingIterator cfsi = new
ColumnFamilySkippingIterator(delIter);
SortedKeyValueIterator<Key,Value> colFilter =
ColumnQualifierFilter.wrap(cfsi, columnSet);
SortedKeyValueIterator<Key,Value> visFilter =
VisibilityFilter.wrap(colFilter, authorizations,
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
