SimpleTwoScalar.java

package edu.mit.ll.graphulo.simplemult;

import com.google.common.collect.Iterators;
import edu.mit.ll.graphulo.ewise.EWiseOp;
import edu.mit.ll.graphulo.reducer.Reducer;
import edu.mit.ll.graphulo.rowmult.MultiplyOp;
import edu.mit.ll.graphulo.util.GraphuloUtil;
import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.IteratorEnvironment;
//import org.apache.log4j.LogManager;
//import org.apache.log4j.Logger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.AbstractMap.SimpleImmutableEntry;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;


/**
 * A simple abstract class for an operation acting on two scalars.
 * This includes matrix multiplication, element-wise multiplication,
 * element-wise sum, Combiner-style sum, unary Apply, and Reductions.
 * <p>
 * The requirement is that this operation logic must return zero or one entry per multiply
 * and depend only on Values, not Keys.  More advanced operations should not use this class.
 * Certain methods are marked as final to prevent misuse/confusion.
 * Please take care that your operator is commutative for the Combiner and Reduce usages.
 * Your operator should always be associative.
 * Finally, in the case of EWiseOp, the operation logic must not need to act on non-matching Values.
 * <p>
 *   Can be used for unary Apply if passed a fixed Value operand
 *   on the left or right of the multiplication inside iterator options.
 *   Defaults to the left side of the multiplication; pass {@value #REVERSE} if the right is desired.
 * <p>
 * Has a newVisibility option for new Keys created with MultiplyOp or EWiseOp usage.
 */
public abstract class SimpleTwoScalar extends KeyTwoScalar implements MultiplyOp, EWiseOp, Reducer {
  private static final Logger log = LoggerFactory.getLogger(SimpleTwoScalar.class);

  public static final String NEW_VISIBILITY ="newVisibility", USE_NEW_VISIBILITY = "useNewVisibility";
  protected boolean useNewVisibility = false;
  protected byte[] newVisibility = null;

  //////////////////////////////////////////////////////////////////////////////////
  /** Implements simple multiply logic. Returning null means no entry is emitted. */
  public abstract Value multiply(Value Aval, Value Bval);
  //////////////////////////////////////////////////////////////////////////////////

//  protected static IteratorSetting addOptionsToIteratorSetting(IteratorSetting itset, boolean reverse, Value fixedValue,
//                                                               String newVisibility) {
//    KeyTwoScalar.addOptionsToIteratorSetting(itset, reverse, fixedValue);
//    if (newVisibility != null && !newVisibility.isEmpty())
//      itset.addOption(ApplyIterator.APPLYOP + GraphuloUtil.OPT_SUFFIX + NEW_VISIBILITY, newVisibility);
//    return itset;
//  }

  @Override
  public void init(Map<String, String> options, IteratorEnvironment env) throws IOException {
    Map<String,String> parentOpts = new HashMap<>();
    for (Map.Entry<String, String> entry : options.entrySet()) {
      String k = entry.getKey(), v = entry.getValue();
      switch (k) {
        case USE_NEW_VISIBILITY: useNewVisibility = Boolean.parseBoolean(v); break;
        case NEW_VISIBILITY:
          newVisibility = v.getBytes(StandardCharsets.UTF_8);
          break;
        default:
          parentOpts.put(k, v);
          break;
      }
    }
    super.init(parentOpts, env);  // Setup parent.
  }

  @Override
  public SimpleTwoScalar deepCopy(IteratorEnvironment env) {
    SimpleTwoScalar copy = (SimpleTwoScalar) super.deepCopy(env);
    copy.useNewVisibility = useNewVisibility;
    copy.newVisibility = useNewVisibility ? Arrays.copyOf(newVisibility, newVisibility.length) : null ;
    return copy;
  }

  /** For {@link MultiplyOp}. Uses {@link System#currentTimeMillis()} for the timestamp. */
  @Override
  public final Iterator<? extends Entry<Key, Value>> multiply(
      ByteSequence Mrow, ByteSequence ATcolF, ByteSequence ATcolQ, ByteSequence ATcolVis, long ATtime,
      ByteSequence BcolF, ByteSequence BcolQ, ByteSequence BcolVis, long Btime,
      Value ATval, Value Bval) {
//    System.err.println("Mrow:"+Mrow+" ATcolQ:"+ATcolQ+" BcolQ:"+BcolQ+" ATval:"+ATval+" Bval:"+Bval);
    assert ATval != null || Bval != null;
    Key k = new Key(ATcolQ.toArray(), ATcolF.toArray(),
        BcolQ.toArray(), useNewVisibility ? newVisibility : GraphuloUtil.EMPTY_BYTES);
    Value v = reverse ? multiply(Bval, ATval) : multiply(ATval, Bval);
    return v == null ? Collections.<Entry<Key,Value>>emptyIterator() : Iterators.singletonIterator(new SimpleImmutableEntry<>(k, v));
  }

  /** For {@link EWiseOp}. Uses {@link System#currentTimeMillis()} for the timestamp. */
  @Override
  public final Iterator<? extends Entry<Key, Value>> multiply(
      ByteSequence Mrow, ByteSequence McolF, ByteSequence McolQ, ByteSequence McolVis,
      long Atime, long Btime,
      Value Aval, Value Bval) {
    // Important!  Aval xor Bval could be null, if emitNoMatchA or emitNoMatchB are true in TwoTableIterator.
    // Decision is to emit the non-matching entries untouched by the operation.  This is a *SIMPLETwoScalar* operator.
    assert Aval != null || Bval != null;
    final Key k = new Key(Mrow.toArray(), McolF.toArray(),
        McolQ.toArray(), useNewVisibility ? newVisibility : McolVis.toArray());
    if (Aval == null)
      return Iterators.singletonIterator(new SimpleImmutableEntry<>(k, Bval));
    if (Bval == null)
      return Iterators.singletonIterator(new SimpleImmutableEntry<>(k, Aval));

    Value v = reverse ? multiply(Bval, Aval) : multiply(Aval, Bval);
    return v == null ? Collections.<Entry<Key,Value>>emptyIterator() : Iterators.singletonIterator(new SimpleImmutableEntry<>(k, v));
  }

  @Override
  public final Value multiply(Key key, Value v1, Value v2) {
    return multiply(v1, v2);
  }


  @Override
  public IteratorOptions describeOptions() {
    IteratorOptions io = super.describeOptions();
    io.setName("SimpleTwoScalar");
    io.setDescription("A Combiner that operates on every pair of entries matching row through column visibility, that does not need to see the Key");
//    io.addNamedOption(NEW_VISIBILITY, "Authorizations to use for new Keys created with MultiplyOp or EWiseOp usage");
    return io;
  }

  @Override
  public boolean validateOptions(Map<String, String> options) {
    if (options.containsKey(NEW_VISIBILITY))
//      new Authorizations(options.get(NEW_VISIBILITY));
      log.warn("Combiner usage does not use newVisibility: "+options.get(NEW_VISIBILITY));
    return super.validateOptions(options);
  }

  private Value reducerV;

  @Override
  public final void reset() {
    reducerV = null;
  }

  @Override
  public final void update(Key k, Value v) {
    if (reducerV == null)
      reducerV = v == null ? null : new Value(v);
    else
      reducerV = reverse ? multiply(reducerV, v) : multiply(v, reducerV);
  }

  @Override
  public final void combine(byte[] another) {
    Value v = new Value(another);
    if (reducerV == null)
      reducerV = new Value(v);
    else
      reducerV = reverse ? multiply(reducerV, v) : multiply(v, reducerV);
  }

  @Override
  public final boolean hasTopForClient() {
    return reducerV != null;
  }

  @Override
  public final byte[] getForClient() {
    return reducerV.get();
  }

}