import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
@Override
public List<Short> check(Environment env, Mutation mutation) {
  Set<Short> violations = null;

  if (!isAlphaNum(mutation.getRow()))
    violations = addViolation(violations, NON_ALPHA_NUM_ROW);

  Collection<ColumnUpdate> updates = mutation.getUpdates();
  for (ColumnUpdate columnUpdate : updates) {
    if (!isAlphaNum(columnUpdate.getColumnFamily()))
      violations = addViolation(violations, NON_ALPHA_NUM_COLF);

    if (!isAlphaNum(columnUpdate.getColumnQualifier()))
      violations = addViolation(violations, NON_ALPHA_NUM_COLQ);
  }

  return null == violations ? null : new ArrayList<>(violations);
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
@Override
public void write(Text table, Mutation mutation) throws IOException, InterruptedException {
  TreeMap<Key,Value> buffer = getBuffer(table);
  int mutationSize = 0;
  for (ColumnUpdate update : mutation.getUpdates()) {
    Key k = new Key(mutation.getRow(), update.getColumnFamily(), update.getColumnQualifier(), update.getColumnVisibility(), update.getTimestamp(),
        update.isDeleted());
    Value v = new Value(update.getValue());
    // TODO account for object overhead
    mutationSize += k.getSize();
    mutationSize += v.getSize();
    buffer.put(k, v);
  }
  size += mutationSize;
  long bufferSize = bufferSizes.get(table);
  
  // TODO use a MutableLong instead
  bufferSize += mutationSize;
  bufferSizes.put(table, bufferSize);
  
  while (size >= maxSize) {
    flushLargestTable();
  }
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
public static org.apache.accumulo.proxy.thrift.ColumnUpdate toThrift(ColumnUpdate update) {
  if (update == null) {
    return null;
  }

  org.apache.accumulo.proxy.thrift.ColumnUpdate tcu = new org.apache.accumulo.proxy.thrift.ColumnUpdate();
  tcu.setColFamily(update.getColumnFamily());
  tcu.setColQualifier(update.getColumnQualifier());
  tcu.setColVisibility(update.getColumnVisibility());
  tcu.setTimestamp(update.getTimestamp());
  tcu.setValue(update.getValue());
  // Work around for ACCUMULO-3474
  if (update.isDeleted()) {
    tcu.setDeleteCell(update.isDeleted());
  }
  return tcu;
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
/**
 * Ensure mutations pass constraints
 */
public List<Short> check(Environment env, Mutation mutation) {

	// Create a list to hold violation codes
	ArrayList<Short> violations = new ArrayList<Short>();

	// Get the updates from the mutation object
	Collection<ColumnUpdate> updates = mutation.getUpdates();

	// Iterate through each record determining if a record passes
	// constraints
	for (ColumnUpdate columnUpdate : updates) {
		// Ignore any record that contains a delete
		if (!columnUpdate.isDeleted()) {
			// If the record is greater than the maximum value, add a
			// violation code to the list of violations
			if (Long.parseLong(new String(columnUpdate.getValue())) >= maxValue) {
				violations.add(MAX_VALUE_EXCEEDED);
			}
		}
	}

	// Return the list of violation codes
	return violations;
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
/**
 * Signs the given mutation and then write it to Accumulo.
 *
 * @param mutation
 *          The mutation to sign.
 */
@Override
public void addMutation(Mutation mutation) throws MutationsRejectedException {
  Mutation signedMutation = new Mutation(mutation.getRow());

  // Sign the entries.
  for (ColumnUpdate update : mutation.getUpdates()) {
    if (update.isDeleted()) {
      if (signatureConfig.destination == Destination.COLUMN_VISIBILITY) {
        throw new IllegalArgumentException("cannot delete entries when the signature is stored in the column visibility");
      }

      if (update.hasTimestamp()) {
        signedMutation.putDelete(update.getColumnFamily(), update.getColumnQualifier(), new ColumnVisibility(update.getColumnVisibility()),
            update.getTimestamp());
      } else {
        signedMutation.putDelete(update.getColumnFamily(), update.getColumnQualifier(), new ColumnVisibility(update.getColumnVisibility()));
      }
    } else {
      Entry<Key,Value> signedEntry = signer.sign(new MutableEntry(mutation.getRow(), update).toEntry(), update.hasTimestamp());
      Key signedKey = signedEntry.getKey();

      if (update.hasTimestamp()) {
        signedMutation.put(signedKey.getColumnFamily(), signedKey.getColumnQualifier(), signedKey.getColumnVisibilityParsed(), signedKey.getTimestamp(),
            signedEntry.getValue());
      } else {
        signedMutation.put(signedKey.getColumnFamily(), signedKey.getColumnQualifier(), signedKey.getColumnVisibilityParsed(), signedEntry.getValue());
      }
    }
  }

  // Write the signed mutations.
  if (signatureTableWriter != null) {
    tableWriter.addMutation(mutation);
    signatureTableWriter.addMutation(signedMutation);
  } else {
    tableWriter.addMutation(signedMutation);
  }
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
/**
 * Creates a mutable entry for the given update.
 *
 * @param row
 *          The row that is being updated.
 * @param update
 *          Update to wrap.
 */
public MutableEntry(byte[] row, ColumnUpdate update) {
  this.row = row;
  this.colF = update.getColumnFamily();
  this.colQ = update.getColumnQualifier();
  this.colVis = update.getColumnVisibility();
  this.timestamp = update.getTimestamp();
  this.delete = update.isDeleted();
  this.value = update.getValue();
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
@Test
public void deleteTest() throws Exception {
  when(mockConnector.createBatchWriter(TEST_TABLE, null)).thenReturn(mockWriter);
  BatchWriter writer = new SignedBatchWriter(mockConnector, TEST_TABLE, null, getConfig("config1.ini"), aliceKeyContainers.get(ValueSigner.RSA_PSS));

  Mutation mutation = new Mutation("row".getBytes());
  mutation.putDelete("colF".getBytes(), "colQ".getBytes());
  mutation.putDelete("colF".getBytes(), "colQ".getBytes(), 0L);
  writer.addMutation(mutation);

  verify(mockWriter).addMutation(captor.capture());
  verify(mockSignatureWriter, never()).addMutation(any());

  List<Mutation> mutations = captor.getAllValues();
  assertThat("only a single mutation", mutations, iterableWithSize(1));

  Mutation signed = mutations.get(0);
  assertThat("row is unchanged", signed.getRow(), equalTo("row".getBytes()));

  List<ColumnUpdate> updates = signed.getUpdates();
  assertThat("has 2 updates", updates, hasSize(2));

  ColumnUpdate update = updates.get(0);
  assertThat("is delete operation", update.isDeleted(), is(true));
  assertThat("column family is unchanged", update.getColumnFamily(), equalTo("colF".getBytes()));
  assertThat("column qualifier is unchanged", update.getColumnQualifier(), equalTo("colQ".getBytes()));
  assertThat("timestamp is not set", update.hasTimestamp(), is(false));

  update = updates.get(1);
  assertThat("is delete operation", update.isDeleted(), is(true));
  assertThat("column family is unchanged", update.getColumnFamily(), equalTo("colF".getBytes()));
  assertThat("column qualifier is unchanged", update.getColumnQualifier(), equalTo("colQ".getBytes()));
  assertThat("timestamp is set", update.hasTimestamp(), is(true));
  assertThat("timestamp is correct", update.getTimestamp(), is(0L));
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
@Test
public void constructorWithUpdateTest() {
  ColumnUpdate update = new ColumnUpdate(colF, colQ, colVis, true, timestamp, delete, value);
  MutableEntry mutableEntry = new MutableEntry(row, update);

  assertThat("row should be unmodified", mutableEntry.row, is(equalTo(row)));
  assertThat("colF should be unmodified", mutableEntry.colF, is(equalTo(colF)));
  assertThat("colQ should be unmodified", mutableEntry.colQ, is(equalTo(colQ)));
  assertThat("colVis should be unmodified", mutableEntry.colVis, is(equalTo(colVis)));
  assertThat("timestamp should be unmodified", mutableEntry.timestamp, is(equalTo(timestamp)));
  assertThat("delete should be unmodified", mutableEntry.delete, is(equalTo(delete)));
  assertThat("value should be unmodified", mutableEntry.value, is(equalTo(value)));
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
private void createTestData1() {
    List<Tag> tags = Collections.singletonList(new Tag("host", "host1"));

    for (long i = 0; i < 1000; i += 100) {
        Metric m = new Metric("sys.loadAvg", i, .2, tags);
        Mutation mutation = MetricAdapter.toMutation(m);
        for (ColumnUpdate cu : mutation.getUpdates()) {
            Key key = new Key(mutation.getRow(), cu.getColumnFamily(), cu.getColumnQualifier(),
                    cu.getColumnVisibility(), cu.getTimestamp());
            testData1.put(key, new Value(cu.getValue()));
        }
    }
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
void put(Map<Key, Value> testData, Metric m) {
    Mutation mutation = MetricAdapter.toMutation(m);
    for (ColumnUpdate cu : mutation.getUpdates()) {
        Key key = new Key(mutation.getRow(), cu.getColumnFamily(), cu.getColumnQualifier(),
                cu.getColumnVisibility(), cu.getTimestamp());
        testData.put(key, new Value(cu.getValue()));
    }
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
@Override
public List<Short> check(Environment env, Mutation mutation) {
  Collection<ColumnUpdate> updates = mutation.getUpdates();

  for (ColumnUpdate columnUpdate : updates) {
    if (!isNumeric(columnUpdate.getValue()))
      return VIOLATION_LIST;
  }

  return null;
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
/**
 * Writes a mutation to the specified table.  If the mutation is meant to delete then the mutation will
 * be transformed to a delete mutation.
 * @param table the table to write to.
 * @param mutation the {@link mutation}.
 * @param context the {@link Context}.
 * @param isDelete {@code true} if the mutation should be a delete mutation.  {@code false} otherwise.
 * @throws IOException
 * @throws InterruptedException
 */
private static void writeMutation(final Text table, final Mutation mutation, final Context context, final boolean isDelete) throws IOException, InterruptedException {
    if (isDelete) {
        final List<ColumnUpdate> updates = mutation.getUpdates();
        final ColumnUpdate columnUpdate = updates.get(0);
        final ColumnVisibility cv = columnUpdate.getColumnVisibility() != null ? new ColumnVisibility(columnUpdate.getColumnVisibility()) : null;
        final Mutation deleteMutation = new Mutation(new Text(mutation.getRow()));
        deleteMutation.putDelete(columnUpdate.getColumnFamily(), columnUpdate.getColumnQualifier(), cv, columnUpdate.getTimestamp());
        context.write(table, deleteMutation);
    } else {
        context.write(table, mutation);
    }
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
/**
 * Converts each non-thrift ColumnUpdate in {@code list} to a Thrift ColumnUpdate (see {@link #toThrift(ColumnUpdate)}) and adds it to the provided
 * {@code thriftUpdates} list. If {@code updates} is null or empty, {@code thriftUpdates} will not be changed.
 * 
 * @param thriftUpdates
 *          the list where the newly created Thrift ColumnUpdates will be added.
 * @param updates
 *          the list of ColumnUpdates to be converted to Thrift ColumnUpdates
 */
public static void addThriftColumnUpdates(List<org.apache.accumulo.proxy.thrift.ColumnUpdate> thriftUpdates, List<ColumnUpdate> updates) {
  if (updates == null) {
    return;
  }

  for (ColumnUpdate cu : updates) {
    thriftUpdates.add(toThrift(cu));
  }
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
@Test
public void testWrite2TupleWithColumn() throws IOException, ParseException {
    AccumuloStorage storage = new AccumuloStorage("col");

    Tuple t = TupleFactory.getInstance().newTuple(2);
    t.set(0, "row");
    t.set(1, "value");

    Collection<Mutation> mutations = storage.getMutations(t);

    Assert.assertEquals(1, mutations.size());

    Mutation m = mutations.iterator().next();

    Assert.assertTrue("Rows not equal",
            Arrays.equals(m.getRow(), ((String) t.get(0)).getBytes()));

    List<ColumnUpdate> colUpdates = m.getUpdates();
    Assert.assertEquals(1, colUpdates.size());

    ColumnUpdate colUpdate = colUpdates.get(0);
    Assert.assertTrue("CF not equal",
            Arrays.equals(colUpdate.getColumnFamily(), "col".getBytes()));
    Assert.assertTrue("CQ not equal",
            Arrays.equals(colUpdate.getColumnQualifier(), new byte[0]));
    Assert.assertTrue("Values not equal",
            Arrays.equals(colUpdate.getValue(), "value".getBytes()));
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
@Test
public void testWrite2TupleWithColumnQual() throws IOException,
        ParseException {
    AccumuloStorage storage = new AccumuloStorage("col:qual");

    Tuple t = TupleFactory.getInstance().newTuple(2);
    t.set(0, "row");
    t.set(1, "value");

    Collection<Mutation> mutations = storage.getMutations(t);

    Assert.assertEquals(1, mutations.size());

    Mutation m = mutations.iterator().next();

    Assert.assertTrue("Rows not equal",
            Arrays.equals(m.getRow(), ((String) t.get(0)).getBytes()));

    List<ColumnUpdate> colUpdates = m.getUpdates();
    Assert.assertEquals(1, colUpdates.size());

    ColumnUpdate colUpdate = colUpdates.get(0);
    Assert.assertTrue("CF not equal",
            Arrays.equals(colUpdate.getColumnFamily(), "col".getBytes()));
    Assert.assertTrue("CQ not equal", Arrays.equals(
            colUpdate.getColumnQualifier(), "qual".getBytes()));
    Assert.assertTrue("Values not equal",
            Arrays.equals(colUpdate.getValue(), "value".getBytes()));
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
@Test
public void testWriteIgnoredExtraColumns() throws IOException,
        ParseException {
    AccumuloStorage storage = new AccumuloStorage("col");

    Tuple t = TupleFactory.getInstance().newTuple(3);
    t.set(0, "row");
    t.set(1, "value1");
    t.set(2, "value2");

    Collection<Mutation> mutations = storage.getMutations(t);

    Assert.assertEquals(1, mutations.size());

    Mutation m = mutations.iterator().next();

    Assert.assertTrue("Rows not equal",
            Arrays.equals(m.getRow(), ((String) t.get(0)).getBytes()));

    List<ColumnUpdate> colUpdates = m.getUpdates();
    Assert.assertEquals(1, colUpdates.size());

    ColumnUpdate colUpdate = colUpdates.get(0);
    Assert.assertTrue("CF not equal",
            Arrays.equals(colUpdate.getColumnFamily(), "col".getBytes()));
    Assert.assertTrue("CQ not equal",
            Arrays.equals(colUpdate.getColumnQualifier(), new byte[0]));
    Assert.assertTrue("Values not equal",
            Arrays.equals(colUpdate.getValue(), "value1".getBytes()));
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
@Test
public void testWriteIgnoredExtraMap() throws IOException, ParseException {
    AccumuloStorage storage = new AccumuloStorage("col1");

    Map<String, Object> map = Maps.newHashMap();

    map.put("mapcol1", "mapval1");
    map.put("mapcol2", "mapval2");
    map.put("mapcol3", "mapval3");
    map.put("mapcol4", "mapval4");

    Tuple t = TupleFactory.getInstance().newTuple(3);
    t.set(0, "row");
    t.set(1, "value1");
    t.set(2, map);

    Collection<Mutation> mutations = storage.getMutations(t);

    Assert.assertEquals(1, mutations.size());

    Mutation m = mutations.iterator().next();

    Assert.assertTrue("Rows not equal",
            Arrays.equals(m.getRow(), ((String) t.get(0)).getBytes()));

    List<ColumnUpdate> colUpdates = m.getUpdates();
    Assert.assertEquals(1, colUpdates.size());

    ColumnUpdate update = colUpdates.get(0);
    Assert.assertEquals("col1", new String(update.getColumnFamily()));
    Assert.assertEquals("", new String(update.getColumnQualifier()));
    Assert.assertEquals("value1", new String(update.getValue()));
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
@Test
public void testWriteMultipleColumnsWithNonExpandedMap()
        throws IOException, ParseException {
    AccumuloStorage storage = new AccumuloStorage("col1,col2");

    Map<String, Object> map = Maps.newHashMap();

    map.put("mapcol1", "mapval1");
    map.put("mapcol2", "mapval2");
    map.put("mapcol3", "mapval3");
    map.put("mapcol4", "mapval4");

    Tuple t = TupleFactory.getInstance().newTuple(3);
    t.set(0, "row");
    t.set(1, "value1");
    t.set(2, map);

    Collection<Mutation> mutations = storage.getMutations(t);

    Assert.assertEquals(1, mutations.size());

    Mutation m = mutations.iterator().next();

    Assert.assertTrue("Rows not equal",
            Arrays.equals(m.getRow(), ((String) t.get(0)).getBytes()));

    List<ColumnUpdate> colUpdates = m.getUpdates();
    Assert.assertEquals(2, colUpdates.size());

    ColumnUpdate update = colUpdates.get(0);
    Assert.assertEquals("col1", new String(update.getColumnFamily()));
    Assert.assertEquals("", new String(update.getColumnQualifier()));
    Assert.assertEquals("value1", new String(update.getValue()));

    update = colUpdates.get(1);
    Assert.assertEquals("col2", new String(update.getColumnFamily()));
    Assert.assertEquals("", new String(update.getColumnQualifier()));
    Assert.assertArrayEquals(storage.objToBytes(map, DataType.MAP),
            update.getValue());
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
@Test
public void addMutationValueTest() throws Exception {
  when(mockConnector.createBatchWriter(TEST_TABLE, null)).thenReturn(mockWriter);
  BatchWriter writer = new SignedBatchWriter(mockConnector, TEST_TABLE, null, getConfig("config1.ini"), aliceKeyContainers.get(ValueSigner.RSA_PSS));
  EntrySigner signer = new EntrySigner(getConfig("config1.ini"), aliceKeyContainers.get(ValueSigner.RSA_PSS));

  Mutation mutation = new Mutation("row".getBytes());
  mutation.put("colF".getBytes(), "colQ".getBytes(), "val1".getBytes());
  mutation.put("colF".getBytes(), "colQ".getBytes(), 0, "val2".getBytes());
  writer.addMutation(mutation);

  verify(mockWriter).addMutation(captor.capture());
  verify(mockSignatureWriter, never()).addMutation(any());

  List<Mutation> mutations = captor.getAllValues();
  assertThat("only a single mutation", mutations, hasSize(1));

  Mutation signed = mutations.get(0);
  assertThat("row is unchanged", signed.getRow(), equalTo("row".getBytes()));

  List<ColumnUpdate> updates = signed.getUpdates();
  assertThat("has 2 updates", updates, hasSize(2));

  ColumnUpdate update = updates.get(0);
  assertThat("column family is unchanged", update.getColumnFamily(), equalTo("colF".getBytes()));
  assertThat("column qualifier is unchanged", update.getColumnFamily(), equalTo("colF".getBytes()));
  assertThat("timestamp not set", update.hasTimestamp(), is(false));
  assertThat("signature in value", update.getColumnFamily(), not(equalTo("val1".getBytes())));

  MutableEntry verified = new MutableEntry(signer.verify(new MutableEntry("row".getBytes(), update).toEntry()));
  assertThat("value was correctly unwrapped", verified.value, equalTo("val1".getBytes()));

  update = updates.get(1);
  assertThat("column family is unchanged", update.getColumnFamily(), equalTo("colF".getBytes()));
  assertThat("column qualifier is unchanged", update.getColumnFamily(), equalTo("colF".getBytes()));
  assertThat("timestamp is set", update.hasTimestamp(), is(true));
  assertThat("timestamp is correct", update.getTimestamp(), is(0L));
  assertThat("signature in value", update.getColumnFamily(), not(equalTo("val2".getBytes())));

  verified = new MutableEntry(signer.verify(new MutableEntry("row".getBytes(), update).toEntry()));
  assertThat("value was correctly unwrapped", verified.value, equalTo("val2".getBytes()));
}
 

import org.apache.accumulo.core.data.ColumnUpdate; //导入依赖的package包/类
@Test
public void addMutationColVisTest() throws Exception {
  when(mockConnector.createBatchWriter(TEST_TABLE, null)).thenReturn(mockWriter);
  BatchWriter writer = new SignedBatchWriter(mockConnector, TEST_TABLE, null, getConfig("config2.ini"), aliceKeyContainers.get(ValueSigner.RSA_PSS));
  EntrySigner signer = new EntrySigner(getConfig("config2.ini"), aliceKeyContainers.get(ValueSigner.RSA_PSS));

  Mutation mutation = new Mutation("row".getBytes());
  mutation.put("colF".getBytes(), "colQ".getBytes(), "val1".getBytes());
  mutation.put("colF".getBytes(), "colQ".getBytes(), 0, "val2".getBytes());
  writer.addMutation(mutation);

  verify(mockWriter).addMutation(captor.capture());
  verify(mockSignatureWriter, never()).addMutation(any());

  List<Mutation> mutations = captor.getAllValues();
  assertThat("only a single mutation", mutations, iterableWithSize(1));

  Mutation signed = mutations.get(0);
  assertThat("row is unchanged", signed.getRow(), equalTo("row".getBytes()));

  List<ColumnUpdate> updates = signed.getUpdates();
  assertThat("has 2 updates", updates, hasSize(2));

  ColumnUpdate update = updates.get(0);
  assertThat("column family is unchanged", update.getColumnFamily(), equalTo("colF".getBytes()));
  assertThat("column qualifier is unchanged", update.getColumnQualifier(), equalTo("colQ".getBytes()));
  assertThat("colQualifier has the default visibility", new String(update.getColumnVisibility(), Utils.VISIBILITY_CHARSET), startsWith("(default)"));
  assertThat("timestamp not set", update.hasTimestamp(), is(false));
  assertThat("value is unchanged", update.getValue(), equalTo("val1".getBytes()));

  MutableEntry verified = new MutableEntry(signer.verify(new MutableEntry("row".getBytes(), update).toEntry()));
  assertThat("value was correctly unwrapped", verified.value, equalTo("val1".getBytes()));

  update = updates.get(1);
  assertThat("column family is unchanged", update.getColumnFamily(), equalTo("colF".getBytes()));
  assertThat("column qualifier is unchanged", update.getColumnQualifier(), equalTo("colQ".getBytes()));
  assertThat("colQualifier has the default visibility", new String(update.getColumnVisibility(), Utils.VISIBILITY_CHARSET), startsWith("(default)"));
  assertThat("timestamp is set", update.hasTimestamp(), is(true));
  assertThat("timestamp is correct", update.getTimestamp(), is(0L));
  assertThat("value is unchanged", update.getValue(), equalTo("val2".getBytes()));

  verified = new MutableEntry(signer.verify(new MutableEntry("row".getBytes(), update).toEntry()));
  assertThat("value was correctly unwrapped", verified.value, equalTo("val2".getBytes()));
}