accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ktur...@apache.org
Subject [1/4] accumulo git commit: ACCUMULO-4066 Speed up condition checking for conditional mutations
Date Thu, 28 Jan 2016 02:01:20 GMT
Repository: accumulo
Updated Branches:
  refs/heads/1.7 0faf8b9c0 -> d4882a15f


ACCUMULO-4066 Speed up condition checking for conditional mutations


Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/21d2f615
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/21d2f615
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/21d2f615

Branch: refs/heads/1.7
Commit: 21d2f61524de1f26ed88c4baaddee70c01081196
Parents: ff1e003
Author: Keith Turner <kturner@apache.org>
Authored: Wed Jan 27 14:13:40 2016 -0500
Committer: Keith Turner <kturner@apache.org>
Committed: Wed Jan 27 19:32:29 2016 -0500

----------------------------------------------------------------------
 .../core/client/impl/CompressedIterators.java   |  14 +-
 .../core/client/impl/ConditionalWriterImpl.java |  44 ++++-
 .../accumulo/core/iterators/IteratorUtil.java   |  93 ++++++-----
 .../client/impl/ConditionalComparatorTest.java  |  53 ++++++
 .../tserver/ConditionCheckerContext.java        | 164 +++++++++++++++++++
 .../org/apache/accumulo/tserver/Tablet.java     |  37 ++++-
 .../apache/accumulo/tserver/TabletServer.java   |  95 ++++-------
 .../accumulo/test/ConditionalWriterIT.java      | 137 +++++++++++++++-
 8 files changed, 521 insertions(+), 116 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/21d2f615/core/src/main/java/org/apache/accumulo/core/client/impl/CompressedIterators.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/CompressedIterators.java
b/core/src/main/java/org/apache/accumulo/core/client/impl/CompressedIterators.java
index 3fcce90..96d58a7 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/CompressedIterators.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/CompressedIterators.java
@@ -24,15 +24,12 @@ import java.util.Map;
 import java.util.Map.Entry;
 
 import org.apache.accumulo.core.client.IteratorSetting;
-import org.apache.accumulo.core.data.ArrayByteSequence;
-import org.apache.accumulo.core.data.ByteSequence;
 import org.apache.accumulo.core.data.thrift.IterInfo;
 import org.apache.accumulo.core.util.UnsynchronizedBuffer;
 
 public class CompressedIterators {
   private Map<String,Integer> symbolMap;
   private List<String> symbolTable;
-  private Map<ByteSequence,IterConfig> cache;
 
   public static class IterConfig {
     public List<IterInfo> ssiList = new ArrayList<IterInfo>();
@@ -46,7 +43,6 @@ public class CompressedIterators {
 
   public CompressedIterators(List<String> symbols) {
     this.symbolTable = symbols;
-    this.cache = new HashMap<ByteSequence,IterConfig>();
   }
 
   private int getSymbolID(String symbol) {
@@ -85,14 +81,7 @@ public class CompressedIterators {
   }
 
   public IterConfig decompress(ByteBuffer iterators) {
-
-    ByteSequence iterKey = new ArrayByteSequence(iterators);
-    IterConfig config = cache.get(iterKey);
-    if (config != null) {
-      return config;
-    }
-
-    config = new IterConfig();
+    IterConfig config = new IterConfig();
 
     UnsynchronizedBuffer.Reader in = new UnsynchronizedBuffer.Reader(iterators);
 
@@ -120,7 +109,6 @@ public class CompressedIterators {
 
     }
 
-    cache.put(iterKey, config);
     return config;
   }
 

http://git-wip-us.apache.org/repos/asf/accumulo/blob/21d2f615/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
b/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
index 730cf08..9030d77 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
@@ -19,7 +19,9 @@ package org.apache.accumulo.core.client.impl;
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -70,6 +72,7 @@ import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
 import org.apache.accumulo.core.util.BadArgumentException;
 import org.apache.accumulo.core.util.ByteBufferUtil;
 import org.apache.accumulo.core.util.LoggingRunnable;
+import org.apache.accumulo.core.util.NamingThreadFactory;
 import org.apache.accumulo.core.util.ThriftUtil;
 import org.apache.accumulo.core.util.UtilWaitThread;
 import org.apache.accumulo.core.zookeeper.ZooUtil;
@@ -375,7 +378,7 @@ class ConditionalWriterImpl implements ConditionalWriter {
     this.credentials = credentials;
     this.auths = config.getAuthorizations();
     this.ve = new VisibilityEvaluator(config.getAuthorizations());
-    this.threadPool = new ScheduledThreadPoolExecutor(config.getMaxWriteThreads());
+    this.threadPool = new ScheduledThreadPoolExecutor(config.getMaxWriteThreads(), new NamingThreadFactory(this.getClass().getSimpleName()));
     this.locator = TabletLocator.getLocator(instance, new Text(tableId));
     this.serverQueues = new HashMap<String,ServerQueue>();
     this.tableId = tableId;
@@ -741,10 +744,47 @@ class ConditionalWriterImpl implements ConditionalWriter {
     }
   }
 
+  static class ConditionComparator implements Comparator<Condition> {
+
+    private static final Long MAX = Long.valueOf(Long.MAX_VALUE);
+
+    @Override
+    public int compare(Condition c1, Condition c2) {
+      int comp = c1.getFamily().compareTo(c2.getFamily());
+      if (comp == 0) {
+        comp = c1.getQualifier().compareTo(c2.getQualifier());
+        if (comp == 0) {
+          comp = c1.getVisibility().compareTo(c2.getVisibility());
+          if (comp == 0) {
+            Long l1 = c1.getTimestamp();
+            Long l2 = c2.getTimestamp();
+            if (l1 == null) {
+              l1 = MAX;
+            }
+
+            if (l2 == null) {
+              l2 = MAX;
+            }
+
+            comp = l2.compareTo(l1);
+          }
+        }
+      }
+
+      return comp;
+    }
+  }
+
+  private static final ConditionComparator CONDITION_COMPARATOR = new ConditionComparator();
+
   private List<TCondition> convertConditions(ConditionalMutation cm, CompressedIterators
compressedIters) {
     List<TCondition> conditions = new ArrayList<TCondition>(cm.getConditions().size());
 
-    for (Condition cond : cm.getConditions()) {
+    // sort conditions inorder to get better lookup performance. Sort on client side so tserver
does not have to do it.
+    Condition[] ca = cm.getConditions().toArray(new Condition[cm.getConditions().size()]);
+    Arrays.sort(ca, CONDITION_COMPARATOR);
+
+    for (Condition cond : ca) {
       long ts = 0;
       boolean hasTs = false;
 

http://git-wip-us.apache.org/repos/asf/accumulo/blob/21d2f615/core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
----------------------------------------------------------------------
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 98392f6..6f76d77 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
@@ -26,6 +26,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Set;
 import java.util.TreeMap;
 
 import org.apache.accumulo.core.client.IteratorSetting;
@@ -108,21 +109,26 @@ public class IteratorUtil {
     return props;
   }
 
-  public static int getMaxPriority(IteratorScope scope, AccumuloConfiguration conf) {
-    List<IterInfo> iters = new ArrayList<IterInfo>();
-    parseIterConf(scope, iters, new HashMap<String,Map<String,String>>(), conf);
+  public static void mergeIteratorConfig(List<IterInfo> destList, Map<String,Map<String,String>>
destOpts, List<IterInfo> tableIters,
+      Map<String,Map<String,String>> tableOpts, List<IterInfo> ssi, Map<String,Map<String,String>>
ssio) {
+    destList.addAll(tableIters);
+    destList.addAll(ssi);
+    Collections.sort(destList, new IterInfoComparator());
 
-    int max = 0;
-
-    for (IterInfo iterInfo : iters) {
-      if (iterInfo.priority > max)
-        max = iterInfo.priority;
+    Set<Entry<String,Map<String,String>>> es = tableOpts.entrySet();
+    for (Entry<String,Map<String,String>> entry : es) {
+      if (entry.getValue() == null) {
+        destOpts.put(entry.getKey(), null);
+      } else {
+        destOpts.put(entry.getKey(), new HashMap<String,String>(entry.getValue()));
+      }
     }
 
-    return max;
+    IteratorUtil.mergeOptions(ssio, destOpts);
+
   }
 
-  protected static void parseIterConf(IteratorScope scope, List<IterInfo> iters, Map<String,Map<String,String>>
allOptions, AccumuloConfiguration conf) {
+  public static void parseIterConf(IteratorScope scope, List<IterInfo> iters, Map<String,Map<String,String>>
allOptions, AccumuloConfiguration conf) {
     final Property scopeProperty = IteratorScope.getProperty(scope);
     final String scopePropertyKey = scopeProperty.getKey();
 
@@ -155,24 +161,6 @@ public class IteratorUtil {
     Collections.sort(iters, new IterInfoComparator());
   }
 
-  public static String findIterator(IteratorScope scope, String className, AccumuloConfiguration
conf, Map<String,String> opts) {
-    ArrayList<IterInfo> iters = new ArrayList<IterInfo>();
-    Map<String,Map<String,String>> allOptions = new HashMap<String,Map<String,String>>();
-
-    parseIterConf(scope, iters, allOptions, conf);
-
-    for (IterInfo iterInfo : iters)
-      if (iterInfo.className.equals(className)) {
-        Map<String,String> tmpOpts = allOptions.get(iterInfo.iterName);
-        if (tmpOpts != null) {
-          opts.putAll(tmpOpts);
-        }
-        return iterInfo.iterName;
-      }
-
-    return null;
-  }
-
   public static <K extends WritableComparable<?>,V extends Writable> SortedKeyValueIterator<K,V>
loadIterators(IteratorScope scope,
       SortedKeyValueIterator<K,V> source, KeyExtent extent, AccumuloConfiguration conf,
IteratorEnvironment env) throws IOException {
     List<IterInfo> emptyList = Collections.emptyList();
@@ -209,6 +197,12 @@ public class IteratorUtil {
 
     parseIterConf(scope, iters, allOptions, conf);
 
+    mergeOptions(ssio, allOptions);
+
+    return loadIterators(source, iters, allOptions, env, useAccumuloClassLoader, conf.get(Property.TABLE_CLASSPATH));
+  }
+
+  private static void mergeOptions(Map<String,Map<String,String>> ssio, Map<String,Map<String,String>>
allOptions) {
     for (Entry<String,Map<String,String>> entry : ssio.entrySet()) {
       if (entry.getValue() == null)
         continue;
@@ -219,30 +213,35 @@ public class IteratorUtil {
         options.putAll(entry.getValue());
       }
     }
-
-    return loadIterators(source, iters, allOptions, env, useAccumuloClassLoader, conf.get(Property.TABLE_CLASSPATH));
   }
 
-  @SuppressWarnings("unchecked")
   public static <K extends WritableComparable<?>,V extends Writable> SortedKeyValueIterator<K,V>
loadIterators(SortedKeyValueIterator<K,V> source,
       Collection<IterInfo> iters, Map<String,Map<String,String>> iterOpts,
IteratorEnvironment env, boolean useAccumuloClassLoader, String context)
       throws IOException {
+    return loadIterators(source, iters, iterOpts, env, useAccumuloClassLoader, context, null);
+  }
+
+  public static <K extends WritableComparable<?>,V extends Writable> SortedKeyValueIterator<K,V>
loadIterators(SortedKeyValueIterator<K,V> source,
+      Collection<IterInfo> iters, Map<String,Map<String,String>> iterOpts,
IteratorEnvironment env, boolean useAccumuloClassLoader, String context,
+      Map<String,Class<? extends SortedKeyValueIterator<K,V>>> classCache)
throws IOException {
     // wrap the source in a SynchronizedIterator in case any of the additional configured
iterators want to use threading
     SortedKeyValueIterator<K,V> prev = new SynchronizedIterator<K,V>(source);
 
     try {
       for (IterInfo iterInfo : iters) {
 
-        Class<? extends SortedKeyValueIterator<K,V>> clazz;
-        if (useAccumuloClassLoader) {
-          if (context != null && !context.equals(""))
-            clazz = (Class<? extends SortedKeyValueIterator<K,V>>) AccumuloVFSClassLoader.getContextManager().loadClass(context,
iterInfo.className,
-                SortedKeyValueIterator.class);
-          else
-            clazz = (Class<? extends SortedKeyValueIterator<K,V>>) AccumuloVFSClassLoader.loadClass(iterInfo.className,
SortedKeyValueIterator.class);
+        Class<? extends SortedKeyValueIterator<K,V>> clazz = null;
+        if (classCache != null) {
+          clazz = classCache.get(iterInfo.className);
+
+          if (clazz == null) {
+            clazz = loadClass(useAccumuloClassLoader, context, iterInfo);
+            classCache.put(iterInfo.className, clazz);
+          }
         } else {
-          clazz = (Class<? extends SortedKeyValueIterator<K,V>>) Class.forName(iterInfo.className).asSubclass(SortedKeyValueIterator.class);
+          clazz = loadClass(useAccumuloClassLoader, context, iterInfo);
         }
+
         SortedKeyValueIterator<K,V> skvi = clazz.newInstance();
 
         Map<String,String> options = iterOpts.get(iterInfo.iterName);
@@ -266,6 +265,22 @@ public class IteratorUtil {
     return prev;
   }
 
+  @SuppressWarnings("unchecked")
+  private static <K extends WritableComparable<?>,V extends Writable> Class<?
extends SortedKeyValueIterator<K,V>> loadClass(boolean useAccumuloClassLoader,
+      String context, IterInfo iterInfo) throws ClassNotFoundException, IOException {
+    Class<? extends SortedKeyValueIterator<K,V>> clazz;
+    if (useAccumuloClassLoader) {
+      if (context != null && !context.equals(""))
+        clazz = (Class<? extends SortedKeyValueIterator<K,V>>) AccumuloVFSClassLoader.getContextManager().loadClass(context,
iterInfo.className,
+            SortedKeyValueIterator.class);
+      else
+        clazz = (Class<? extends SortedKeyValueIterator<K,V>>) AccumuloVFSClassLoader.loadClass(iterInfo.className,
SortedKeyValueIterator.class);
+    } else {
+      clazz = (Class<? extends SortedKeyValueIterator<K,V>>) Class.forName(iterInfo.className).asSubclass(SortedKeyValueIterator.class);
+    }
+    return clazz;
+  }
+
   public static Range maximizeStartKeyTimeStamp(Range range) {
     Range seekRange = range;
 

http://git-wip-us.apache.org/repos/asf/accumulo/blob/21d2f615/core/src/test/java/org/apache/accumulo/core/client/impl/ConditionalComparatorTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/client/impl/ConditionalComparatorTest.java
b/core/src/test/java/org/apache/accumulo/core/client/impl/ConditionalComparatorTest.java
new file mode 100644
index 0000000..5413893
--- /dev/null
+++ b/core/src/test/java/org/apache/accumulo/core/client/impl/ConditionalComparatorTest.java
@@ -0,0 +1,53 @@
+/*
+ * 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.client.impl;
+
+import org.apache.accumulo.core.client.impl.ConditionalWriterImpl.ConditionComparator;
+import org.apache.accumulo.core.data.Condition;
+import org.apache.accumulo.core.security.ColumnVisibility;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ConditionalComparatorTest {
+  @Test
+  public void testComparator() {
+    Condition c1 = new Condition("a", "b");
+    Condition c2 = new Condition("a", "c");
+    Condition c3 = new Condition("b", "c");
+    Condition c4 = new Condition("a", "b").setTimestamp(5);
+    Condition c5 = new Condition("a", "b").setTimestamp(6);
+    Condition c6 = new Condition("a", "b").setVisibility(new ColumnVisibility("A&B"));
+    Condition c7 = new Condition("a", "b").setVisibility(new ColumnVisibility("A&C"));
+
+    ConditionComparator comparator = new ConditionComparator();
+
+    Assert.assertTrue(comparator.compare(c1, c1) == 0);
+    Assert.assertTrue(comparator.compare(c1, c2) < 0);
+    Assert.assertTrue(comparator.compare(c2, c1) > 0);
+    Assert.assertTrue(comparator.compare(c1, c3) < 0);
+    Assert.assertTrue(comparator.compare(c3, c1) > 0);
+    Assert.assertTrue(comparator.compare(c1, c4) < 0);
+    Assert.assertTrue(comparator.compare(c4, c1) > 0);
+    Assert.assertTrue(comparator.compare(c5, c4) < 0);
+    Assert.assertTrue(comparator.compare(c4, c5) > 0);
+    Assert.assertTrue(comparator.compare(c1, c7) < 0);
+    Assert.assertTrue(comparator.compare(c7, c1) > 0);
+    Assert.assertTrue(comparator.compare(c6, c7) < 0);
+    Assert.assertTrue(comparator.compare(c7, c6) > 0);
+  }
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/21d2f615/server/tserver/src/main/java/org/apache/accumulo/tserver/ConditionCheckerContext.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/ConditionCheckerContext.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/ConditionCheckerContext.java
new file mode 100644
index 0000000..2e34f38
--- /dev/null
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/ConditionCheckerContext.java
@@ -0,0 +1,164 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.accumulo.tserver;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.accumulo.core.client.impl.CompressedIterators;
+import org.apache.accumulo.core.client.impl.CompressedIterators.IterConfig;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.ArrayByteSequence;
+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.data.thrift.IterInfo;
+import org.apache.accumulo.core.data.thrift.TCMResult;
+import org.apache.accumulo.core.data.thrift.TCMStatus;
+import org.apache.accumulo.core.data.thrift.TCondition;
+import org.apache.accumulo.core.iterators.IteratorUtil;
+import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.apache.accumulo.tserver.data.ServerConditionalMutation;
+import org.apache.hadoop.io.Text;
+
+import com.google.common.base.Preconditions;
+
+public class ConditionCheckerContext {
+  private CompressedIterators compressedIters;
+
+  private List<IterInfo> tableIters;
+  private Map<String,Map<String,String>> tableIterOpts;
+  private TabletIteratorEnvironment tie;
+  private String context;
+  private Map<String,Class<? extends SortedKeyValueIterator<Key,Value>>>
classCache;
+
+  private static class MergedIterConfig {
+    List<IterInfo> mergedIters;
+    Map<String,Map<String,String>> mergedItersOpts;
+
+    MergedIterConfig(List<IterInfo> mergedIters, Map<String,Map<String,String>>
mergedItersOpts) {
+      this.mergedIters = mergedIters;
+      this.mergedItersOpts = mergedItersOpts;
+    }
+  }
+
+  private Map<ByteSequence,MergedIterConfig> mergedIterCache = new HashMap<ByteSequence,MergedIterConfig>();
+
+  ConditionCheckerContext(CompressedIterators compressedIters, AccumuloConfiguration tableConf)
{
+    this.compressedIters = compressedIters;
+
+    tableIters = new ArrayList<IterInfo>();
+    tableIterOpts = new HashMap<String,Map<String,String>>();
+
+    // parse table iterator config once
+    IteratorUtil.parseIterConf(IteratorScope.scan, tableIters, tableIterOpts, tableConf);
+
+    context = tableConf.get(Property.TABLE_CLASSPATH);
+
+    classCache = new HashMap<String,Class<? extends SortedKeyValueIterator<Key,Value>>>();
+
+    tie = new TabletIteratorEnvironment(IteratorScope.scan, tableConf);
+  }
+
+  SortedKeyValueIterator<Key,Value> buildIterator(SortedKeyValueIterator<Key,Value>
systemIter, TCondition tc) throws IOException {
+
+    ArrayByteSequence key = new ArrayByteSequence(tc.iterators);
+    MergedIterConfig mic = mergedIterCache.get(key);
+    if (mic == null) {
+      IterConfig ic = compressedIters.decompress(tc.iterators);
+
+      List<IterInfo> mergedIters = new ArrayList<IterInfo>(tableIters.size()
+ ic.ssiList.size());
+      Map<String,Map<String,String>> mergedItersOpts = new HashMap<String,Map<String,String>>(tableIterOpts.size()
+ ic.ssio.size());
+
+      IteratorUtil.mergeIteratorConfig(mergedIters, mergedItersOpts, tableIters, tableIterOpts,
ic.ssiList, ic.ssio);
+
+      mic = new MergedIterConfig(mergedIters, mergedItersOpts);
+
+      mergedIterCache.put(key, mic);
+    }
+
+    return IteratorUtil.loadIterators(systemIter, mic.mergedIters, mic.mergedItersOpts, tie,
true, context, classCache);
+  }
+
+  boolean checkConditions(SortedKeyValueIterator<Key,Value> systemIter, ServerConditionalMutation
scm) throws IOException {
+    boolean add = true;
+
+    for (TCondition tc : scm.getConditions()) {
+
+      Range range;
+      if (tc.hasTimestamp)
+        range = Range.exact(new Text(scm.getRow()), new Text(tc.getCf()), new Text(tc.getCq()),
new Text(tc.getCv()), tc.getTs());
+      else
+        range = Range.exact(new Text(scm.getRow()), new Text(tc.getCf()), new Text(tc.getCq()),
new Text(tc.getCv()));
+
+      SortedKeyValueIterator<Key,Value> iter = buildIterator(systemIter, tc);
+
+      ByteSequence cf = new ArrayByteSequence(tc.getCf());
+      iter.seek(range, Collections.singleton(cf), true);
+      Value val = null;
+      if (iter.hasTop()) {
+        val = iter.getTopValue();
+      }
+
+      if ((val == null ^ tc.getVal() == null) || (val != null && !Arrays.equals(tc.getVal(),
val.get()))) {
+        add = false;
+        break;
+      }
+    }
+    return add;
+  }
+
+  public class ConditionChecker {
+
+    private List<ServerConditionalMutation> conditionsToCheck;
+    private List<ServerConditionalMutation> okMutations;
+    private List<TCMResult> results;
+    private boolean checked = false;
+
+    public ConditionChecker(List<ServerConditionalMutation> conditionsToCheck, List<ServerConditionalMutation>
okMutations, List<TCMResult> results) {
+      this.conditionsToCheck = conditionsToCheck;
+      this.okMutations = okMutations;
+      this.results = results;
+    }
+
+    public void check(SortedKeyValueIterator<Key,Value> systemIter) throws IOException
{
+      Preconditions.checkArgument(!checked, "check() method should only be called once");
+      checked = true;
+
+      for (ServerConditionalMutation scm : conditionsToCheck) {
+        if (checkConditions(systemIter, scm)) {
+          okMutations.add(scm);
+        } else {
+          results.add(new TCMResult(scm.getID(), TCMStatus.REJECTED));
+        }
+      }
+    }
+  }
+
+  public ConditionChecker newChecker(List<ServerConditionalMutation> conditionsToCheck,
List<ServerConditionalMutation> okMutations, List<TCMResult> results) {
+    return new ConditionChecker(conditionsToCheck, okMutations, results);
+  }
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/21d2f615/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java
index 3f00c0b..ca4bde6 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java
@@ -128,6 +128,7 @@ import org.apache.accumulo.trace.instrument.Span;
 import org.apache.accumulo.trace.instrument.Trace;
 import org.apache.accumulo.tserver.Compactor.CompactionCanceledException;
 import org.apache.accumulo.tserver.Compactor.CompactionEnv;
+import org.apache.accumulo.tserver.ConditionCheckerContext.ConditionChecker;
 import org.apache.accumulo.tserver.FileManager.ScanFileManager;
 import org.apache.accumulo.tserver.InMemoryMap.MemoryIterator;
 import org.apache.accumulo.tserver.TabletServer.TservConstraintEnv;
@@ -1597,6 +1598,23 @@ public class Tablet {
     void receive(List<KVEntry> matches) throws IOException;
   }
 
+  public void checkConditions(ConditionChecker checker, Authorizations authorizations, AtomicBoolean
iFlag) throws IOException {
+
+    ScanDataSource dataSource = new ScanDataSource(this, authorizations, this.defaultSecurityLabel,
iFlag);
+
+    try {
+      SortedKeyValueIterator<Key,Value> iter = new SourceSwitchingIterator(dataSource);
+      checker.check(iter);
+    } catch (IOException ioe) {
+      dataSource.close(true);
+      throw ioe;
+    } finally {
+      // code in finally block because always want
+      // to return mapfiles, even when exception is thrown
+      dataSource.close(false);
+    }
+  }
+
   class LookupResult {
     List<Range> unfinishedRanges = new ArrayList<Range>();
     long bytesAdded = 0;
@@ -1929,18 +1947,29 @@ public class Tablet {
     private StatsIterator statsIterator;
 
     ScanOptions options;
+    private final boolean loadIters;
 
     ScanDataSource(Authorizations authorizations, byte[] defaultLabels, HashSet<Column>
columnSet, List<IterInfo> ssiList, Map<String,Map<String,String>> ssio,
         AtomicBoolean interruptFlag) {
       expectedDeletionCount = dataSourceDeletions.get();
       this.options = new ScanOptions(-1, authorizations, defaultLabels, columnSet, ssiList,
ssio, interruptFlag, false);
       this.interruptFlag = interruptFlag;
+      this.loadIters = true;
     }
 
     ScanDataSource(ScanOptions options) {
       expectedDeletionCount = dataSourceDeletions.get();
       this.options = options;
       this.interruptFlag = options.interruptFlag;
+      this.loadIters = true;
+    }
+
+    ScanDataSource(Tablet tablet, Authorizations authorizations, byte[] defaultLabels, AtomicBoolean
iFlag) {
+      expectedDeletionCount = dataSourceDeletions.get();
+      Set<Column> emptycols = Collections.emptySet();
+      this.options = new ScanOptions(-1, authorizations, defaultLabels, emptycols, null,
null, iFlag, false);
+      this.interruptFlag = iFlag;
+      this.loadIters = false;
     }
 
     @Override
@@ -2035,8 +2064,12 @@ public class Tablet {
 
       VisibilityFilter visFilter = new VisibilityFilter(colFilter, options.authorizations,
options.defaultLabels);
 
-      return iterEnv.getTopLevelIterator(IteratorUtil
-          .loadIterators(IteratorScope.scan, visFilter, extent, acuTableConf, options.ssiList,
options.ssio, iterEnv));
+      if (!loadIters) {
+        return visFilter;
+      } else {
+        return iterEnv.getTopLevelIterator(IteratorUtil.loadIterators(IteratorScope.scan,
visFilter, extent, acuTableConf, options.ssiList, options.ssio,
+            iterEnv));
+      }
     }
 
     private void close(boolean sawErrors) {

http://git-wip-us.apache.org/repos/asf/accumulo/blob/21d2f615/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
----------------------------------------------------------------------
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 4e3e00b..6023ae3 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
@@ -67,7 +67,6 @@ import org.apache.accumulo.core.client.AccumuloException;
 import org.apache.accumulo.core.client.AccumuloSecurityException;
 import org.apache.accumulo.core.client.Instance;
 import org.apache.accumulo.core.client.impl.CompressedIterators;
-import org.apache.accumulo.core.client.impl.CompressedIterators.IterConfig;
 import org.apache.accumulo.core.client.impl.ScannerImpl;
 import org.apache.accumulo.core.client.impl.Tables;
 import org.apache.accumulo.core.client.impl.TabletLocator;
@@ -99,7 +98,6 @@ import org.apache.accumulo.core.data.thrift.ScanResult;
 import org.apache.accumulo.core.data.thrift.TCMResult;
 import org.apache.accumulo.core.data.thrift.TCMStatus;
 import org.apache.accumulo.core.data.thrift.TColumn;
-import org.apache.accumulo.core.data.thrift.TCondition;
 import org.apache.accumulo.core.data.thrift.TConditionalMutation;
 import org.apache.accumulo.core.data.thrift.TConditionalSession;
 import org.apache.accumulo.core.data.thrift.TKey;
@@ -199,6 +197,7 @@ import org.apache.accumulo.trace.instrument.Span;
 import org.apache.accumulo.trace.instrument.Trace;
 import org.apache.accumulo.trace.thrift.TInfo;
 import org.apache.accumulo.tserver.Compactor.CompactionInfo;
+import org.apache.accumulo.tserver.ConditionCheckerContext.ConditionChecker;
 import org.apache.accumulo.tserver.RowLocks.RowLock;
 import org.apache.accumulo.tserver.Tablet.CommitSession;
 import org.apache.accumulo.tserver.Tablet.KVEntry;
@@ -1930,79 +1929,57 @@ public class TabletServer extends AbstractMetricsImpl implements org.apache.accu
         List<String> symbols) throws IOException {
       Iterator<Entry<KeyExtent,List<ServerConditionalMutation>>> iter =
updates.entrySet().iterator();
 
-      CompressedIterators compressedIters = new CompressedIterators(symbols);
+      final CompressedIterators compressedIters = new CompressedIterators(symbols);
+      ConditionCheckerContext checkerContext = new ConditionCheckerContext(compressedIters,
ServerConfiguration.getTableConfiguration(instance, cs.tableId));
 
       while (iter.hasNext()) {
-        Entry<KeyExtent,List<ServerConditionalMutation>> entry = iter.next();
-        Tablet tablet = onlineTablets.get(entry.getKey());
+        final Entry<KeyExtent,List<ServerConditionalMutation>> entry = iter.next();
+        final Tablet tablet = onlineTablets.get(entry.getKey());
 
         if (tablet == null || tablet.isClosed()) {
           for (ServerConditionalMutation scm : entry.getValue())
             results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
           iter.remove();
         } else {
-          List<ServerConditionalMutation> okMutations = new ArrayList<ServerConditionalMutation>(entry.getValue().size());
+          final List<ServerConditionalMutation> okMutations = new ArrayList<ServerConditionalMutation>(entry.getValue().size());
+          final List<TCMResult> resultsSubList = results.subList(results.size(), results.size());
 
-          for (ServerConditionalMutation scm : entry.getValue()) {
-            if (checkCondition(results, cs, compressedIters, tablet, scm))
-              okMutations.add(scm);
-          }
-
-          entry.setValue(okMutations);
-        }
-
-      }
-    }
-
-    boolean checkCondition(ArrayList<TCMResult> results, ConditionalSession cs, CompressedIterators
compressedIters, Tablet tablet,
-        ServerConditionalMutation scm) throws IOException {
-      boolean add = true;
-
-      Set<Column> emptyCols = Collections.emptySet();
-
-      for (TCondition tc : scm.getConditions()) {
-
-        Range range;
-        if (tc.hasTimestamp)
-          range = Range.exact(new Text(scm.getRow()), new Text(tc.getCf()), new Text(tc.getCq()),
new Text(tc.getCv()), tc.getTs());
-        else
-          range = Range.exact(new Text(scm.getRow()), new Text(tc.getCf()), new Text(tc.getCq()),
new Text(tc.getCv()));
-
-        IterConfig ic = compressedIters.decompress(tc.iterators);
-
-        Scanner scanner = tablet.createScanner(range, 1, emptyCols, cs.auths, ic.ssiList,
ic.ssio, false, cs.interruptFlag);
+          ConditionChecker checker = checkerContext.newChecker(entry.getValue(), okMutations,
resultsSubList);
+          try {
+            tablet.checkConditions(checker, cs.auths, cs.interruptFlag);
 
-        try {
-          ScanBatch batch = scanner.read();
+            if (okMutations.size() > 0) {
+              entry.setValue(okMutations);
+            } else {
+              iter.remove();
+            }
+          } catch (TabletClosedException e) {
+            // clear anything added while checking conditions.
+            resultsSubList.clear();
 
-          Value val = null;
+            for (ServerConditionalMutation scm : entry.getValue()) {
+              results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
+            }
+            iter.remove();
+          } catch (IterationInterruptedException e) {
+            // clear anything added while checking conditions.
+            resultsSubList.clear();
 
-          for (KVEntry entry2 : batch.results) {
-            val = entry2.getValue();
-            break;
-          }
+            for (ServerConditionalMutation scm : entry.getValue()) {
+              results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
+            }
+            iter.remove();
+          } catch (TooManyFilesException e) {
+            // clear anything added while checking conditions.
+            resultsSubList.clear();
 
-          if ((val == null ^ tc.getVal() == null) || (val != null && !Arrays.equals(tc.getVal(),
val.get()))) {
-            results.add(new TCMResult(scm.getID(), TCMStatus.REJECTED));
-            add = false;
-            break;
+            for (ServerConditionalMutation scm : entry.getValue()) {
+              results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
+            }
+            iter.remove();
           }
-
-        } catch (TabletClosedException e) {
-          results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
-          add = false;
-          break;
-        } catch (IterationInterruptedException iie) {
-          results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
-          add = false;
-          break;
-        } catch (TooManyFilesException tmfe) {
-          results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
-          add = false;
-          break;
         }
       }
-      return add;
     }
 
     private void writeConditionalMutations(Map<KeyExtent,List<ServerConditionalMutation>>
updates, ArrayList<TCMResult> results, ConditionalSession sess) {

http://git-wip-us.apache.org/repos/asf/accumulo/blob/21d2f615/test/src/test/java/org/apache/accumulo/test/ConditionalWriterIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/ConditionalWriterIT.java b/test/src/test/java/org/apache/accumulo/test/ConditionalWriterIT.java
index 5bfaebf..d32bc43 100644
--- a/test/src/test/java/org/apache/accumulo/test/ConditionalWriterIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/ConditionalWriterIT.java
@@ -19,9 +19,11 @@ package org.apache.accumulo.test;
 
 import static org.junit.Assert.assertTrue;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -63,7 +65,11 @@ import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Mutation;
 import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.IteratorEnvironment;
+import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
 import org.apache.accumulo.core.iterators.LongCombiner.Type;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.apache.accumulo.core.iterators.WrappingIterator;
 import org.apache.accumulo.core.iterators.user.SummingCombiner;
 import org.apache.accumulo.core.iterators.user.VersioningIterator;
 import org.apache.accumulo.core.security.Authorizations;
@@ -90,6 +96,7 @@ import org.junit.Assert;
 import org.junit.Assume;
 import org.junit.Test;
 
+import com.google.common.base.Charsets;
 import com.google.common.collect.Iterables;
 
 /**
@@ -501,7 +508,135 @@ public class ConditionalWriterIT extends AccumuloClusterIT {
 
     Assert.assertEquals(expected, actual);
 
-    // TODO test w/ table that has iterators configured
+    cw.close();
+  }
+
+  public static class AddingIterator extends WrappingIterator {
+    long amount = 0;
+
+    @Override
+    public Value getTopValue() {
+      Value val = super.getTopValue();
+      long l = Long.parseLong(val.toString());
+      String newVal = (l + amount) + "";
+      return new Value(newVal.getBytes(Charsets.UTF_8));
+    }
+
+    @Override
+    public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String>
options, IteratorEnvironment env) throws IOException {
+      this.setSource(source);
+      amount = Long.parseLong(options.get("amount"));
+    }
+  }
+
+  public static class MultiplyingIterator extends WrappingIterator {
+    long amount = 0;
+
+    @Override
+    public Value getTopValue() {
+      Value val = super.getTopValue();
+      long l = Long.parseLong(val.toString());
+      String newVal = l * amount + "";
+      return new Value(newVal.getBytes(Charsets.UTF_8));
+    }
+
+    @Override
+    public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String>
options, IteratorEnvironment env) throws IOException {
+      this.setSource(source);
+      amount = Long.parseLong(options.get("amount"));
+    }
+  }
+
+  @Test
+  public void testTableAndConditionIterators() throws Exception {
+
+    // test w/ table that has iterators configured
+    Connector conn = getConnector();
+    String tableName = getUniqueNames(1)[0];
+
+    IteratorSetting aiConfig1 = new IteratorSetting(30, "AI1", AddingIterator.class);
+    aiConfig1.addOption("amount", "2");
+    IteratorSetting aiConfig2 = new IteratorSetting(35, "MI1", MultiplyingIterator.class);
+    aiConfig2.addOption("amount", "3");
+    IteratorSetting aiConfig3 = new IteratorSetting(40, "AI2", AddingIterator.class);
+    aiConfig3.addOption("amount", "5");
+
+    conn.tableOperations().create(tableName);
+
+    BatchWriter bw = conn.createBatchWriter(tableName, new BatchWriterConfig());
+
+    Mutation m = new Mutation("ACCUMULO-1000");
+    m.put("count", "comments", "6");
+    bw.addMutation(m);
+
+    m = new Mutation("ACCUMULO-1001");
+    m.put("count", "comments", "7");
+    bw.addMutation(m);
+
+    m = new Mutation("ACCUMULO-1002");
+    m.put("count", "comments", "8");
+    bw.addMutation(m);
+
+    bw.close();
+
+    conn.tableOperations().attachIterator(tableName, aiConfig1, EnumSet.of(IteratorScope.scan));
+    conn.tableOperations().offline(tableName, true);
+    conn.tableOperations().online(tableName, true);
+
+    ConditionalWriter cw = conn.createConditionalWriter(tableName, new ConditionalWriterConfig());
+
+    ConditionalMutation cm6 = new ConditionalMutation("ACCUMULO-1000", new Condition("count",
"comments").setValue("8"));
+    cm6.put("count", "comments", "7");
+    Assert.assertEquals(Status.ACCEPTED, cw.write(cm6).getStatus());
+
+    Scanner scanner = conn.createScanner(tableName, new Authorizations());
+    scanner.setRange(new Range("ACCUMULO-1000"));
+    scanner.fetchColumn(new Text("count"), new Text("comments"));
+
+    Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
+    Assert.assertEquals("9", entry.getValue().toString());
+
+    ConditionalMutation cm7 = new ConditionalMutation("ACCUMULO-1000", new Condition("count",
"comments").setIterators(aiConfig2).setValue("27"));
+    cm7.put("count", "comments", "8");
+    Assert.assertEquals(Status.ACCEPTED, cw.write(cm7).getStatus());
+
+    entry = Iterables.getOnlyElement(scanner);
+    Assert.assertEquals("10", entry.getValue().toString());
+
+    ConditionalMutation cm8 = new ConditionalMutation("ACCUMULO-1000", new Condition("count",
"comments").setIterators(aiConfig2, aiConfig3).setValue("35"));
+    cm8.put("count", "comments", "9");
+    Assert.assertEquals(Status.ACCEPTED, cw.write(cm8).getStatus());
+
+    entry = Iterables.getOnlyElement(scanner);
+    Assert.assertEquals("11", entry.getValue().toString());
+
+    ConditionalMutation cm3 = new ConditionalMutation("ACCUMULO-1000", new Condition("count",
"comments").setIterators(aiConfig2).setValue("33"));
+    cm3.put("count", "comments", "3");
+
+    ConditionalMutation cm4 = new ConditionalMutation("ACCUMULO-1001", new Condition("count",
"comments").setIterators(aiConfig3).setValue("14"));
+    cm4.put("count", "comments", "3");
+
+    ConditionalMutation cm5 = new ConditionalMutation("ACCUMULO-1002", new Condition("count",
"comments").setIterators(aiConfig3).setValue("10"));
+    cm5.put("count", "comments", "3");
+
+    Iterator<Result> results = cw.write(Arrays.asList(cm3, cm4, cm5).iterator());
+    Map<String,Status> actual = new HashMap<String,Status>();
+
+    while (results.hasNext()) {
+      Result result = results.next();
+      String k = new String(result.getMutation().getRow());
+      Assert.assertFalse("Did not expect to see multiple resultus for the row: " + k, actual.containsKey(k));
+      actual.put(k, result.getStatus());
+    }
+
+    cw.close();
+
+    Map<String,Status> expected = new HashMap<String,Status>();
+    expected.put("ACCUMULO-1000", Status.ACCEPTED);
+    expected.put("ACCUMULO-1001", Status.ACCEPTED);
+    expected.put("ACCUMULO-1002", Status.REJECTED);
+
+    Assert.assertEquals(expected, actual);
 
     cw.close();
   }


Mime
View raw message