import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
public MizoTitanHBaseRelationParser(Map<Long, MizoTitanRelationType> relationTypes, Cell cell) {
    this.relationTypes = relationTypes;

    this.vertexId = parseVertexId(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());

    Entry entry = createEntry(cell);
    extractRelationMetadata(entry.asReadBuffer());

    if (!isSystemType()) {
        RelationCache relationCache = TITAN_EDGE_SERIALIZER.parseRelation(entry, false, TITAN_TYPE_INSPECTOR);

        if (isProperty()) {
            propertyValue = relationCache.getValue();
        }
        else {
            relationId = relationCache.relationId;
            otherVertexId = relationCache.getOtherVertexId();
            propertiesIterator = relationCache.propertyIterator();
        }

        if (!isKnownType()) {
            log.warn("Unknown relation type (vertex-id={}, type-id={})", this.vertexId, this.typeId);
        }
    }
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
/**
 * Simplest case test of the lock cleaner.
 */
@Test
public void testDeleteSingleLock() throws BackendException {
    Instant now = Instant.ofEpochMilli(1L);

    Entry expiredLockCol = StaticArrayEntry.of(codec.toLockCol(now,
            defaultLockRid, TimestampProviders.MILLI), BufferUtil.getIntBuffer(0));
    EntryList expiredSingleton = StaticArrayEntryList.of(expiredLockCol);

    now = now.plusMillis(1);
    del = new StandardLockCleanerRunnable(store, kc, tx, codec, now, TimestampProviders.MILLI);

    expect(store.getSlice(eq(ksq), eq(tx)))
            .andReturn(expiredSingleton);

    store.mutate(
            eq(key),
            eq(ImmutableList.<Entry> of()),
            eq(ImmutableList.<StaticBuffer> of(expiredLockCol.getColumn())),
            anyObject(StoreTransaction.class));

    ctrl.replay();
    del.run();
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
private final Entry getIndexEntry(CompositeIndexType index, RecordEntry[] record, TitanElement element) {
    DataOutput out = serializer.getDataOutput(1+8+8*record.length+4*8);
    out.putByte(FIRST_INDEX_COLUMN_BYTE);
    if (index.getCardinality()!=Cardinality.SINGLE) {
        VariableLong.writePositive(out,element.longId());
        if (index.getCardinality()!=Cardinality.SET) {
            for (RecordEntry re : record) {
                VariableLong.writePositive(out,re.relationId);
            }
        }
    }
    int valuePosition=out.getPosition();
    if (element instanceof TitanVertex) {
        VariableLong.writePositive(out,element.longId());
    } else {
        assert element instanceof TitanRelation;
        RelationIdentifier rid = (RelationIdentifier)element.id();
        long[] longs = rid.getLongRepresentation();
        Preconditions.checkArgument(longs.length == 3 || longs.length == 4);
        for (int i = 0; i < longs.length; i++) VariableLong.writePositive(out, longs[i]);
    }
    return new StaticArrayEntry(out.getStaticBuffer(),valuePosition);
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
/**
 * Returns the list of adjacent vertex ids for this query. By reading those ids
 * from the entries directly (without creating objects) we get much better performance.
 *
 * @return
 */
public VertexList vertexIds() {
    LongArrayList list = new LongArrayList();
    long previousId = 0;
    for (Long id : Iterables.transform(this,new Function<Entry, Long>() {
        @Nullable
        @Override
        public Long apply(@Nullable Entry entry) {
            return edgeSerializer.readRelation(entry,true,tx).getOtherVertexId();
        }
    })) {
        list.add(id);
        if (id>=previousId && previousId>=0) previousId=id;
        else previousId=-1;
    }
    return new VertexLongList(tx,list,previousId>=0);
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
@Override
public boolean hasNext() {
    ensureOpen();

    if (null != nextRow)
        return true;

    while (rows.hasNext()) {
        nextRow = rows.next();
        List<Entry> ents = nextRow.getValue().getSlice(new KeySliceQuery(nextRow.getKey(), columnSlice), transaction);
        if (null != ents && 0 < ents.size())
            break;
    }

    return null != nextRow;
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
private Map<String,Object> toMap() {
    Map<String,Object> entries = Maps.newHashMap();
    List<Entry> result = BackendOperation.execute(new BackendOperation.Transactional<List<Entry>>() {
        @Override
        public List<Entry> call(StoreTransaction txh) throws BackendException {
            return store.getSlice(new KeySliceQuery(rowKey, BufferUtil.zeroBuffer(1), BufferUtil.oneBuffer(128)),txh);
        }

        @Override
        public String toString() {
            return "setConfiguration";
        }
    },txProvider, times, maxOperationWaitTime);

    for (Entry entry : result) {
        String key = staticBuffer2String(entry.getColumnAs(StaticBuffer.STATIC_FACTORY));
        Object value = staticBuffer2Object(entry.getValueAs(StaticBuffer.STATIC_FACTORY), Object.class);
        entries.put(key,value);
    }
    return entries;
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
private final Entry getIndexEntry(CompositeIndexType index, RecordEntry[] record, TitanElement element) {
    DataOutput out = serializer.getDataOutput(1+8+8*record.length+4*8);
    out.putByte(FIRST_INDEX_COLUMN_BYTE);
    if (index.getCardinality()!=Cardinality.SINGLE) {
        VariableLong.writePositive(out,element.getLongId());
        if (index.getCardinality()!=Cardinality.SET) {
            for (RecordEntry re : record) {
                VariableLong.writePositive(out,re.relationId);
            }
        }
    }
    int valuePosition=out.getPosition();
    if (element instanceof TitanVertex) {
        VariableLong.writePositive(out,element.getLongId());
    } else {
        assert element instanceof TitanRelation;
        RelationIdentifier rid = (RelationIdentifier)element.getId();
        long[] longs = rid.getLongRepresentation();
        Preconditions.checkArgument(longs.length == 3 || longs.length == 4);
        for (int i = 0; i < longs.length; i++) VariableLong.writePositive(out, longs[i]);
    }
    return new StaticArrayEntry(out.getStaticBuffer(),valuePosition);
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
public void testDataSequential() throws Exception {
    loadData(200000,2);
    close();
    KeyColumnValueStoreManager manager = openStorageManager();
    KeyColumnValueStore store = manager.openDatabase(Backend.EDGESTORE_NAME);
    SliceQuery query = new SliceQuery(BufferUtil.zeroBuffer(8),BufferUtil.oneBuffer(8));
    query.setLimit(2);
    Stopwatch watch = Stopwatch.createStarted();
    StoreTransaction txh = manager.beginTransaction(StandardBaseTransactionConfig.of(TimestampProviders.MILLI));
    KeyIterator iter = store.getKeys(query,txh);
    int numV = 0;
    while(iter.hasNext()) {
        StaticBuffer key = iter.next();
        RecordIterator<Entry> entries = iter.getEntries();
        assertEquals(2, Iterators.size(entries));
        numV++;
    }
    iter.close();
    txh.commit();
    System.out.println("Time taken: " + watch.elapsed(TimeUnit.MILLISECONDS));
    System.out.println("Num Vertices: " + numV);
    store.close();
    manager.close();

}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
private Map<String,Object> toMap() {
    Map<String,Object> entries = Maps.newHashMap();
    List<Entry> result = BackendOperation.execute(new BackendOperation.Transactional<List<Entry>>() {
        @Override
        public List<Entry> call(StoreTransaction txh) throws BackendException {
            return store.getSlice(new KeySliceQuery(rowKey, BufferUtil.zeroBuffer(128), BufferUtil.oneBuffer(128)),txh);
        }

        @Override
        public String toString() {
            return "setConfiguration";
        }
    },txProvider, times, maxOperationWaitTime);

    for (Entry entry : result) {
        String key = staticBuffer2String(entry.getColumnAs(StaticBuffer.STATIC_FACTORY));
        Object value = staticBuffer2Object(entry.getValueAs(StaticBuffer.STATIC_FACTORY), Object.class);
        entries.put(key,value);
    }
    return entries;
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
/**
 * Locks with timestamps equal to or numerically greater than the cleaner
 * cutoff timestamp must be preserved. Test that the cleaner reads locks by
 * slicing the store and then does <b>not</b> attempt to write.
 */
@Test
public void testPreservesLocksAtOrAfterCutoff() throws BackendException {
    final Instant cutoff = Instant.ofEpochMilli(10L);

    Entry currentLock = StaticArrayEntry.of(codec.toLockCol(cutoff,
            defaultLockRid, TimestampProviders.MILLI), BufferUtil.getIntBuffer(0));
    Entry futureLock = StaticArrayEntry.of(codec.toLockCol(cutoff.plusMillis(1),
            defaultLockRid, TimestampProviders.MILLI), BufferUtil.getIntBuffer(0));
    EntryList locks = StaticArrayEntryList.of(currentLock, futureLock);

    // Don't increment cutoff: lockCol is exactly at the cutoff timestamp
    del = new StandardLockCleanerRunnable(store, kc, tx, codec, cutoff, TimestampProviders.MILLI);

    expect(store.getSlice(eq(ksq), eq(tx))).andReturn(locks);

    ctrl.replay();
    del.run();
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
/**
 * Reads the configuration property for this StoreManager
 *
 * @param key Key identifying the configuration property
 * @return Value stored for the key or null if the configuration property has not (yet) been defined.
 * @throws com.thinkaurelius.titan.diskstorage.BackendException
 */
@Override
public <O> O get(final String key, final Class<O> datatype) {
    StaticBuffer column = string2StaticBuffer(key);
    final KeySliceQuery query = new KeySliceQuery(rowKey,column, BufferUtil.nextBiggerBuffer(column));
    StaticBuffer result = BackendOperation.execute(new BackendOperation.Transactional<StaticBuffer>() {
        @Override
        public StaticBuffer call(StoreTransaction txh) throws BackendException {
            List<Entry> entries = store.getSlice(query,txh);
            if (entries.isEmpty()) return null;
            return entries.get(0).getValueAs(StaticBuffer.STATIC_FACTORY);
        }

        @Override
        public String toString() {
            return "getConfiguration";
        }
    }, txProvider, times, maxOperationWaitTime);
    if (result==null) return null;
    return staticBuffer2Object(result, datatype);
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
/**
 * {@inheritDoc}
 * <p/>
 * This implementation supports locking when {@code lockStore} is non-null.
 */
@Override
public void mutate(StaticBuffer key, List<Entry> additions, List<StaticBuffer> deletions, StoreTransaction txh) throws BackendException {
    ExpectedValueCheckingTransaction etx = (ExpectedValueCheckingTransaction)txh;
    boolean hasAtLeastOneLock = etx.prepareForMutations();
    if (hasAtLeastOneLock) {
        // Force all mutations on this transaction to use strong consistency
        store.mutate(key, additions, deletions, getConsistentTx(txh));
    } else {
        store.mutate(key, additions, deletions, unwrapTx(txh));
    }
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
public void loadStore(int numKeys, int numCols) {
    StoreTransaction tx = getStoreTx();
    try {
        for (int i=1;i<=numKeys;i++) {
            List<Entry> adds = new ArrayList<Entry>(numCols);
            for (int j=1;j<=numCols;j++) adds.add(getEntry(j,j));
            store.mutate(BufferUtil.getIntBuffer(i),adds,KeyColumnValueStore.NO_DELETIONS,tx);
        }
        tx.commit();
    } catch (BackendException e) {
        throw new RuntimeException(e);
    }
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
@Override
public Iterator<Entry> iterator() {
    Iterator<Entry> iter;
    //If there is a limit we need to wrap the basic iterator in a LimitAdjustingIterator which ensures the right number
    //of elements is returned. Otherwise we just return the basic iterator.
    if (sliceQuery.hasLimit() && sliceQuery.getLimit()!=query.getLimit()) {
        iter = new LimitAdjustingIterator();
    } else {
        iter = getBasicIterator();
    }
    return iter;
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
@Override
public Entry next() {
    final Map.Entry<byte[], NavigableMap<Long, byte[]>> entry = iterator.next();
    byte[] col = entry.getKey();
    byte[] val = entry.getValue().lastEntry().getValue();
    return StaticArrayEntry.of(new StaticArrayBuffer(col), new StaticArrayBuffer(val));
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
/**
 * Retrieves the value for the specified column and key under the given transaction
 * from the store if such exists, otherwise returns NULL
 *
 * @param store  Store
 * @param key    Key
 * @param column Column
 * @param txh    Transaction
 * @return Value for key and column or NULL if such does not exist
 */
public static StaticBuffer get(KeyColumnValueStore store, StaticBuffer key, StaticBuffer column, StoreTransaction txh) throws BackendException {
    KeySliceQuery query = new KeySliceQuery(key, column, BufferUtil.nextBiggerBuffer(column)).setLimit(2);
    List<Entry> result = store.getSlice(query, txh);
    if (result.size() > 1)
        log.warn("GET query returned more than 1 result: store {} | key {} | column {}", new Object[]{store.getName(),
                key, column});

    if (result.isEmpty()) return null;
    else return result.get(0).getValueAs(StaticBuffer.STATIC_FACTORY);
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
@Override
public Iterable<Entry> getCurrentValue() throws IOException, InterruptedException {
    Result result = (Result)reader.getCurrentValue();
    NavigableMap<byte[],NavigableMap<byte[],NavigableMap<Long,byte[]>>> nm = result.getMap();
    return new HBaseMapIterable(nm.get(edgestoreFamilyBytes));
    // return new HBaseMapIterable(reader.getCurrentValue().getMap().get(edgestoreFamilyBytes));
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
@Override
public void add(StaticBuffer key, Entry cell) {
    try {
        kcvs.mutateEntries(key, Lists.newArrayList(cell), KCVSCache.NO_DELETIONS,tx);
    } catch (BackendException e) {
        throw new TitanException("Unexpected storage exception in log persistence against cache",e);
    }
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
@Override
protected void map(StaticBuffer key, Iterable<Entry> values, Context context) throws IOException, InterruptedException {
    EntryArrayList al = EntryArrayList.of(values);

    // KeyFilter check
    if (!keyFilter.test(key)) {
        log.debug("Skipping key {} based on KeyFilter", key);
        return;
    }

    // InitialQuery check (at least one match is required or else the key is ignored)
    EntryList initialQueryMatches = findEntriesMatchingQuery(initialQuery, al);
    if (0 == initialQueryMatches.size()) {
        log.debug("Skipping key {} based on InitialQuery ({}) match failure", key, initialQuery);
        return;
    }

    // Both conditions (KeyFilter && InitialQuery) for invoking process are satisfied

    // Create an entries parameter to be passed into the process method
    Map<SliceQuery, EntryList> matches = new HashMap<>();
    matches.put(initialQuery, initialQueryMatches);

    // Find matches (if any are present) for noninitial queries
    for (SliceQuery sq : subsequentQueries) {
        matches.put(sq, findEntriesMatchingQuery(sq, al));
    }

    // Process
    job.process(key, matches, metrics);
}
 

import com.thinkaurelius.titan.diskstorage.Entry; //导入依赖的package包/类
@Override
public RecordReader<StaticBuffer, Iterable<Entry>> createRecordReader(final InputSplit inputSplit, final TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
    tableReader =
            // DAVID
            // (TableRecordReader) tableInputFormat.createRecordReader(inputSplit, taskAttemptContext);
            (RecordReader) tableInputFormat.createRecordReader(inputSplit, taskAttemptContext);
    // DAVID
    titanRecordReader = new HBaseBinaryRecordReader(tableReader, inputCFBytes);
    return titanRecordReader;
}