Merge branch '2.1'

Author: keith-turner

core/src/main/java/org/apache/accumulo/core/client/ClientSideIteratorScanner.java

@@ -30,15 +30,20 @@
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.TreeSet;
+import java.util.function.Supplier;
 
 import org.apache.accumulo.core.Constants;
 import org.apache.accumulo.core.client.sample.SamplerConfiguration;
+import org.apache.accumulo.core.clientImpl.ClientContext;
+import org.apache.accumulo.core.clientImpl.ClientServiceEnvironmentImpl;
+import org.apache.accumulo.core.clientImpl.ScannerImpl;
 import org.apache.accumulo.core.clientImpl.ScannerOptions;
 import org.apache.accumulo.core.data.ArrayByteSequence;
 import org.apache.accumulo.core.data.ByteSequence;
 import org.apache.accumulo.core.data.Column;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.TableId;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.dataImpl.thrift.IterInfo;
 import org.apache.accumulo.core.iterators.IteratorAdapter;
@@ -71,13 +76,17 @@
  * server side) and to the client side scanner (which will execute client side).
  */
 public class ClientSideIteratorScanner extends ScannerOptions implements Scanner {
+
   private int size;
 
   private Range range;
   private boolean isolated = false;
   private long readaheadThreshold = Constants.SCANNER_DEFAULT_READAHEAD_THRESHOLD;
   private SamplerConfiguration iteratorSamplerConfig;
 
+  private final Supplier<ClientContext> context;
+  private final Supplier<TableId> tableId;
+
   private class ClientSideIteratorEnvironment implements IteratorEnvironment {
 
     private SamplerConfiguration samplerConfig;
@@ -95,7 +104,9 @@ public IteratorScope getIteratorScope() {
 
     @Override
     public boolean isFullMajorCompaction() {
-      return false;
+      // The javadocs state this method will throw an ISE when scope is not majc
+      throw new IllegalStateException(
+          "Asked about major compaction type when scope is " + getIteratorScope());
     }
 
     @Override
@@ -122,6 +133,16 @@ public boolean isSamplingEnabled() {
     public SamplerConfiguration getSamplerConfiguration() {
       return samplerConfig;
     }
+
+    @Override
+    public PluginEnvironment getPluginEnv() {
+      return new ClientServiceEnvironmentImpl(context.get());
+    }
+
+    @Override
+    public TableId getTableId() {
+      return tableId.get();
+    }
   }
 
   /**
@@ -221,6 +242,22 @@ public ClientSideIteratorScanner(final Scanner scanner) {
     if (samplerConfig != null) {
       setSamplerConfiguration(samplerConfig);
     }
+
+    if (scanner instanceof ScannerImpl) {
+      var scannerImpl = (ScannerImpl) scanner;
+      this.context = () -> scannerImpl.getClientContext();
+      this.tableId = () -> scannerImpl.getTableId();
+    } else {
+      // These may never be used, so only fail if an attempt is made to use them.
+      this.context = () -> {
+        throw new UnsupportedOperationException(
+            "Do not know how to obtain client context from " + scanner.getClass().getName());
+      };
+      this.tableId = () -> {
+        throw new UnsupportedOperationException(
+            "Do not know how to obtain tableId from " + scanner.getClass().getName());
+      };
+    }
   }
 
   /**

core/src/main/java/org/apache/accumulo/core/clientImpl/OfflineIterator.java

@@ -35,6 +35,7 @@
 
 import org.apache.accumulo.core.client.AccumuloException;
 import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.PluginEnvironment;
 import org.apache.accumulo.core.client.SampleNotPresentException;
 import org.apache.accumulo.core.client.TableNotFoundException;
 import org.apache.accumulo.core.client.sample.SamplerConfiguration;
@@ -79,9 +80,13 @@ static class OfflineIteratorEnvironment implements IteratorEnvironment {
     private final AccumuloConfiguration conf;
     private final boolean useSample;
     private final SamplerConfiguration sampleConf;
+    private final ClientContext context;
+    private final TableId tableId;
 
-    public OfflineIteratorEnvironment(Authorizations auths, AccumuloConfiguration acuTableConf,
-        boolean useSample, SamplerConfiguration samplerConf) {
+    public OfflineIteratorEnvironment(ClientContext context, TableId tableId, Authorizations auths,
+        AccumuloConfiguration acuTableConf, boolean useSample, SamplerConfiguration samplerConf) {
+      this.context = context;
+      this.tableId = tableId;
       this.authorizations = auths;
       this.conf = acuTableConf;
       this.useSample = useSample;
@@ -135,7 +140,18 @@ public IteratorEnvironment cloneWithSamplingEnabled() {
       if (sampleConf == null) {
         throw new SampleNotPresentException();
       }
-      return new OfflineIteratorEnvironment(authorizations, conf, true, sampleConf);
+      return new OfflineIteratorEnvironment(context, tableId, authorizations, conf, true,
+          sampleConf);
+    }
+
+    @Override
+    public PluginEnvironment getPluginEnv() {
+      return new ClientServiceEnvironmentImpl(context);
+    }
+
+    @Override
+    public TableId getTableId() {
+      return tableId;
     }
   }
 
@@ -311,8 +327,9 @@ private SortedKeyValueIterator<Key,Value> createIterator(KeyExtent extent,
 
     MultiIterator multiIter = new MultiIterator(readers, extent);
 
-    OfflineIteratorEnvironment iterEnv = new OfflineIteratorEnvironment(authorizations, tableCC,
-        false, samplerConfImpl == null ? null : samplerConfImpl.toSamplerConfiguration());
+    OfflineIteratorEnvironment iterEnv =
+        new OfflineIteratorEnvironment(context, tableId, authorizations, tableCC, false,
+            samplerConfImpl == null ? null : samplerConfImpl.toSamplerConfiguration());
 
     byte[] defaultSecurityLabel;
     ColumnVisibility cv =

core/src/main/java/org/apache/accumulo/core/clientImpl/ScannerImpl.java

@@ -120,6 +120,16 @@ public ScannerImpl(ClientContext context, TableId tableId, Authorizations author
     this.size = Constants.SCAN_BATCH_SIZE;
   }
 
+  public ClientContext getClientContext() {
+    ensureOpen();
+    return context;
+  }
+
+  public TableId getTableId() {
+    ensureOpen();
+    return tableId;
+  }
+
   @Override
   public synchronized void setRange(Range range) {
     ensureOpen();

test/src/main/java/org/apache/accumulo/test/ClientSideIteratorIT.java

@@ -22,20 +22,32 @@
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Predicate;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 import org.apache.accumulo.core.client.Accumulo;
 import org.apache.accumulo.core.client.AccumuloClient;
 import org.apache.accumulo.core.client.BatchWriter;
 import org.apache.accumulo.core.client.ClientSideIteratorScanner;
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.clientImpl.ClientContext;
+import org.apache.accumulo.core.clientImpl.OfflineScanner;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Mutation;
 import org.apache.accumulo.core.data.PartialKey;
 import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.Filter;
+import org.apache.accumulo.core.iterators.IteratorEnvironment;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 import org.apache.accumulo.core.iterators.user.IntersectingIterator;
 import org.apache.accumulo.core.iterators.user.VersioningIterator;
 import org.apache.accumulo.core.security.Authorizations;
@@ -154,4 +166,100 @@ public void testVersioning() throws Exception {
       assertFalse(csis.iterator().hasNext());
     }
   }
+
+  private static final AtomicBoolean initCalled = new AtomicBoolean(false);
+
+  public static class TestPropFilter extends Filter {
+
+    private Predicate<Key> keyPredicate;
+
+    private Predicate<Key> createRegexPredicate(String regex) {
+      Predicate<Key> kp = k -> true;
+      if (regex != null) {
+        var pattern = Pattern.compile(regex);
+        kp = k -> pattern.matcher(k.getRowData().toString()).matches();
+      }
+
+      return kp;
+    }
+
+    @Override
+    public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options,
+        IteratorEnvironment env) throws IOException {
+      super.init(source, options, env);
+      Predicate<Key> generalPredicate =
+          createRegexPredicate(env.getPluginEnv().getConfiguration().getCustom("testRegex"));
+      Predicate<Key> tablePredicate = createRegexPredicate(
+          env.getPluginEnv().getConfiguration(env.getTableId()).getTableCustom("testRegex"));
+      keyPredicate = generalPredicate.and(tablePredicate);
+      initCalled.set(true);
+    }
+
+    @Override
+    public boolean accept(Key k, Value v) {
+      return keyPredicate.test(k);
+    }
+  }
+
+  private void runPluginEnvTest(Set<String> expected) throws Exception {
+    try (var scanner = client.createScanner(tableName)) {
+      initCalled.set(false);
+      var csis = new ClientSideIteratorScanner(scanner);
+      csis.addScanIterator(new IteratorSetting(100, "filter", TestPropFilter.class));
+      assertEquals(expected,
+          csis.stream().map(e -> e.getKey().getRowData().toString()).collect(Collectors.toSet()));
+      // this check is here to ensure the iterator executed client side and not server side
+      assertTrue(initCalled.get());
+    }
+
+    // The offline scanner also runs iterators client side, so test its client side access to
+    // accumulo config from iterators also.
+    client.tableOperations().offline(tableName, true);
+    var context = (ClientContext) client;
+    try (OfflineScanner offlineScanner =
+        new OfflineScanner(context, context.getTableId(tableName), Authorizations.EMPTY)) {
+      initCalled.set(false);
+      offlineScanner.addScanIterator(new IteratorSetting(100, "filter", TestPropFilter.class));
+      assertEquals(expected, offlineScanner.stream().map(e -> e.getKey().getRowData().toString())
+          .collect(Collectors.toSet()));
+      assertTrue(initCalled.get());
+    }
+    client.tableOperations().online(tableName, true);
+  }
+
+  /**
+   * Test an iterators ability to access accumulo config in an iterator running client side.
+   */
+  @Test
+  public void testPluginEnv() throws Exception {
+    Set<String> rows = Set.of("1234", "abc", "xyz789");
+
+    client.tableOperations().create(tableName);
+    try (BatchWriter bw = client.createBatchWriter(tableName)) {
+      for (var row : rows) {
+        Mutation m = new Mutation(row);
+        m.put("f", "q", "v");
+        bw.addMutation(m);
+      }
+    }
+
+    runPluginEnvTest(rows);
+
+    // The iterator should see the following system property and filter based on it
+    client.instanceOperations().setProperty("general.custom.testRegex", ".*[a-z]+.*");
+    runPluginEnvTest(Set.of("abc", "xyz789"));
+
+    // The iterator should see the following table property and filter based on the table and system
+    // property
+    client.tableOperations().setProperty(tableName, "table.custom.testRegex", ".*[0-9]+.*");
+    runPluginEnvTest(Set.of("xyz789"));
+
+    // Remove the system property, so filtering should only happen based on the table property
+    client.instanceOperations().removeProperty("general.custom.testRegex");
+    runPluginEnvTest(Set.of("1234", "xyz789"));
+
+    // Iterator should do no filtering after removing this property
+    client.tableOperations().removeProperty(tableName, "table.custom.testRegex");
+    runPluginEnvTest(rows);
+  }
 }