Repository: incubator-fluo Updated Branches: refs/heads/master c896fc16b -> f977834b8
Added integration test for timestamp skipping iter (TSI). Also made TSI more robust to Accumulo changing in future. Project: http://git-wip-us.apache.org/repos/asf/incubator-fluo/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-fluo/commit/f977834b Tree: http://git-wip-us.apache.org/repos/asf/incubator-fluo/tree/f977834b Diff: http://git-wip-us.apache.org/repos/asf/incubator-fluo/diff/f977834b Branch: refs/heads/master Commit: f977834b876133528a160b769427439282cb1d7c Parents: c896fc1 Author: Keith Turner <ktur...@apache.org> Authored: Mon Nov 21 20:57:12 2016 -0500 Committer: Keith Turner <ktur...@apache.org> Committed: Tue Nov 22 10:32:39 2016 -0500 ---------------------------------------------------------------------- .../iterators/TimestampSkippingIterator.java | 34 ++++-- .../accumulo/Skip100StampsIterator.java | 106 +++++++++++++++++++ .../integration/accumulo/TimeskippingIT.java | 71 +++++++++++++ 3 files changed, 205 insertions(+), 6 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/f977834b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/TimestampSkippingIterator.java ---------------------------------------------------------------------- diff --git a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/TimestampSkippingIterator.java b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/TimestampSkippingIterator.java index a09ac3b..ec2a83f 100644 --- a/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/TimestampSkippingIterator.java +++ b/modules/accumulo/src/main/java/org/apache/fluo/accumulo/iterators/TimestampSkippingIterator.java @@ -20,6 +20,7 @@ import java.lang.reflect.Field; import java.util.Collection; import java.util.Map; +import com.google.common.annotations.VisibleForTesting; import org.apache.accumulo.core.data.ByteSequence; import org.apache.accumulo.core.data.Key; import org.apache.accumulo.core.data.PartialKey; @@ -50,9 +51,12 @@ public class TimestampSkippingIterator implements SortedKeyValueIterator<Key, Va private boolean inclusive; private boolean hasSeeked = false; + private boolean removedDeletingIterator = false; + private int removalFailures = 0; + private static final Logger log = LoggerFactory.getLogger(TimestampSkippingIterator.class); - TimestampSkippingIterator(SortedKeyValueIterator<Key, Value> source) { + public TimestampSkippingIterator(SortedKeyValueIterator<Key, Value> source) { this.source = source; } @@ -78,7 +82,7 @@ public class TimestampSkippingIterator implements SortedKeyValueIterator<Key, Va while (source.hasTop() && curCol.equals(source.getTopKey(), PartialKey.ROW_COLFAM_COLQUAL_COLVIS) && timestamp < source.getTopKey().getTimestamp()) { - if (count == 10) { + if (count == 10 && shouldSeek()) { // seek to prefix Key seekKey = new Key(curCol); seekKey.setTimestamp(timestamp); @@ -148,20 +152,23 @@ public class TimestampSkippingIterator implements SortedKeyValueIterator<Key, Va } } - private static void setParent(SortedKeyValueIterator<Key, Value> iter, + private static boolean setParent(SortedKeyValueIterator<Key, Value> iter, SortedKeyValueIterator<Key, Value> newParent) { try { if (iter instanceof WrappingIterator) { Field field = WrappingIterator.class.getDeclaredField("source"); field.setAccessible(true); field.set(iter, newParent); + return true; } } catch (NoSuchFieldException | IllegalArgumentException | IllegalAccessException e) { log.debug(e.getMessage(), e); } + + return false; } - private static void removeDeletingIterator(SortedKeyValueIterator<Key, Value> source) { + private static boolean removeDeletingIterator(SortedKeyValueIterator<Key, Value> source) { SortedKeyValueIterator<Key, Value> prev = source; SortedKeyValueIterator<Key, Value> parent = getParent(source); @@ -174,9 +181,21 @@ public class TimestampSkippingIterator implements SortedKeyValueIterator<Key, Va if (parent != null && parent instanceof DeletingIterator) { SortedKeyValueIterator<Key, Value> delParent = getParent(parent); if (delParent != null) { - setParent(prev, delParent); + return setParent(prev, delParent); } } + + return false; + } + + @VisibleForTesting + public final boolean shouldSeek() { + /* + * This method is a saftey check to ensure the deleting iterator was removed. If this iterator + * was not removed for some reason, then the performance of seeking will be O(N^2). In the case + * where its not removed, it would be better to just scan forward. + */ + return !hasSeeked || removedDeletingIterator || removalFailures < 3; } private void seek(Range range) throws IOException { @@ -185,7 +204,10 @@ public class TimestampSkippingIterator implements SortedKeyValueIterator<Key, Va // up iterators until the 1st seek. Therefore can only remove the deleting iter after the 1st // seek. Also, Accumulo may switch data sources and re-setup the deleting iterator, thats why // this iterator keeps trying to remove it. - removeDeletingIterator(source); + removedDeletingIterator |= removeDeletingIterator(source); + if (!removedDeletingIterator) { + removalFailures++; + } } source.seek(range, fams, inclusive); hasSeeked = true; http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/f977834b/modules/integration/src/test/java/org/apache/fluo/integration/accumulo/Skip100StampsIterator.java ---------------------------------------------------------------------- diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/accumulo/Skip100StampsIterator.java b/modules/integration/src/test/java/org/apache/fluo/integration/accumulo/Skip100StampsIterator.java new file mode 100644 index 0000000..aa8b2ef --- /dev/null +++ b/modules/integration/src/test/java/org/apache/fluo/integration/accumulo/Skip100StampsIterator.java @@ -0,0 +1,106 @@ +/* + * 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.fluo.integration.accumulo; + +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.Range; +import org.apache.accumulo.core.data.Value; +import org.apache.accumulo.core.iterators.IteratorEnvironment; +import org.apache.accumulo.core.iterators.SortedKeyValueIterator; +import org.apache.fluo.accumulo.iterators.TimestampSkippingIterator; + +public class Skip100StampsIterator implements SortedKeyValueIterator<Key, Value> { + + private TimestampSkippingIterator source; + private boolean hasTop; + private int goodVal; + private int goodSeek; + + @Override + public void init(SortedKeyValueIterator<Key, Value> source, Map<String, String> options, + IteratorEnvironment env) throws IOException { + + this.source = new TimestampSkippingIterator(source); + + } + + @Override + public boolean hasTop() { + return hasTop; + } + + @Override + public void next() throws IOException { + hasTop = false; + } + + @Override + public void seek(Range range, Collection<ByteSequence> columnFamilies, boolean inclusive) + throws IOException { + + source.seek(range, columnFamilies, inclusive); + + Key k = new Key("r1", "f1", "q1"); + + long ts = 99999; + goodVal = 0; + hasTop = true; + + /* + * If the TimestampSkippingIterator is not able to remove the DeletingIterator, then the + * following loop will have O(N^2) performance. This happens because every time the following + * loop seeks forward a little bit, the DeletingIterator scans from the start of the column + * looking for deletes. I manually commented out the code that removes the DeletingIterator and + * the following code took 50 to 100 times longer to run. + */ + + while (source.hasTop() && ts > 0) { + source.skipToTimestamp(k, ts); + if (source.hasTop()) { + if (source.getTopValue().toString().equals("v" + ts)) { + goodVal++; + } + + if (source.shouldSeek()) { + goodSeek++; + } + } + + ts -= 100; + } + } + + @Override + public Key getTopKey() { + return new Key("r1", "f1", "q1", 42); + } + + @Override + public Value getTopValue() { + return new Value(("" + goodVal + " " + goodSeek).getBytes()); + } + + @Override + public SortedKeyValueIterator<Key, Value> deepCopy(IteratorEnvironment env) { + throw new UnsupportedOperationException(); + } + +} http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/f977834b/modules/integration/src/test/java/org/apache/fluo/integration/accumulo/TimeskippingIT.java ---------------------------------------------------------------------- diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/accumulo/TimeskippingIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/accumulo/TimeskippingIT.java new file mode 100644 index 0000000..68bb138 --- /dev/null +++ b/modules/integration/src/test/java/org/apache/fluo/integration/accumulo/TimeskippingIT.java @@ -0,0 +1,71 @@ +/* + * 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.fluo.integration.accumulo; + +import com.google.common.collect.Iterables; +import org.apache.accumulo.core.client.BatchWriter; +import org.apache.accumulo.core.client.BatchWriterConfig; +import org.apache.accumulo.core.client.IteratorSetting; +import org.apache.accumulo.core.client.Scanner; +import org.apache.accumulo.core.data.Mutation; +import org.apache.accumulo.core.security.Authorizations; +import org.apache.fluo.integration.ITBase; +import org.junit.Assert; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class TimeskippingIT extends ITBase { + + private static final Logger log = LoggerFactory.getLogger(TimeskippingIT.class); + + @Test + public void testTimestampSkippingIterPerformance() throws Exception { + + conn.tableOperations().create("ttsi", false); + + BatchWriter bw = conn.createBatchWriter("ttsi", new BatchWriterConfig()); + Mutation m = new Mutation("r1"); + for (int i = 0; i < 100000; i++) { + m.put("f1", "q1", i, "v" + i); + } + + bw.addMutation(m); + bw.close(); + + long t2 = System.currentTimeMillis(); + + Scanner scanner = conn.createScanner("ttsi", Authorizations.EMPTY); + scanner.addScanIterator(new IteratorSetting(10, Skip100StampsIterator.class)); + + Assert.assertEquals("999 1000", Iterables.getOnlyElement(scanner).getValue().toString()); + long t3 = System.currentTimeMillis(); + + if (t3 - t2 > 3000) { + log.warn("Timestamp skipping iterator took longer than expected " + (t3 - t2)); + } + + conn.tableOperations().flush("ttsi", null, null, true); + + long t4 = System.currentTimeMillis(); + Assert.assertEquals("999 1000", Iterables.getOnlyElement(scanner).getValue().toString()); + long t5 = System.currentTimeMillis(); + + if (t5 - t4 > 3000) { + log.warn("Timestamp skipping iterator took longer than expected " + (t5 - t4)); + } + } +}