import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
public RocksDBStore(String name, ColumnFamilyDescriptor family, ColumnFamilyHandle handle, RocksDB db, int stripes) {
  super();
  this.family = family;
  this.name = name;
  this.db = db;
  this.parallel = stripes;
  this.handle = handle;
  this.sharedLocks = new AutoCloseableLock[stripes];
  this.exclusiveLocks = new AutoCloseableLock[stripes];

  for (int i = 0; i < stripes; i++) {
    ReadWriteLock core = new ReentrantReadWriteLock();
    sharedLocks[i] = new AutoCloseableLock(core.readLock());
    exclusiveLocks[i] = new AutoCloseableLock(core.writeLock());
  }
}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
public FindByRangeIterator(RocksDB db, ColumnFamilyHandle handle, FindByRange<byte[]> range) {
  this.iter = db.newIterator(handle);
  this.end = range == null ? null : range.getEnd();
  this.endInclusive = range == null ? false : range.isEndInclusive();

  // position at beginning of cursor.
  if (range != null && range.getStart() != null) {
    iter.seek(range.getStart());
    if (!range.isStartInclusive() && Arrays.equals(iter.key(), range.getStart())) {
      iter.next();
    }
  } else {
    iter.seekToFirst();
  }

  populateNext();

}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
@Override
protected NodeTable createEmptyNodeTable() {
    FileOps.ensureDir(testRDB);
    FileOps.clearDirectory(testRDB);
    
    //XXX BuildR.createNodeTable.
    
    BuildR.formatDatabase(testRDB);
    db = BuildR.openDB(testRDB);
    ColumnFamilyHandle h = BuildR.findHandle(db, BuildR.cfNodeId2Node);
    
    //TDBuilder.
    // In-memory.
    RecordFactory recordFactory = new RecordFactory(SystemTDB.LenNodeHash, SystemTDB.SizeOfNodeId) ;
    Index nodeToId = new IndexMap(recordFactory);
    return new RocksNodeTable(db.rdb, h, nodeToId, ByteCodecThrift.create());
}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
@VisibleForTesting
@SuppressWarnings("unchecked")
@Override
public int numStateEntries() {
	int count = 0;

	for (Tuple2<ColumnFamilyHandle, RegisteredKeyedBackendStateMetaInfo<?, ?>> column : kvStateInformation.values()) {
		try (RocksIterator rocksIterator = db.newIterator(column.f0)) {
			rocksIterator.seekToFirst();

			while (rocksIterator.isValid()) {
				count++;
				rocksIterator.next();
			}
		}
	}

	return count;
}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
/**
 * Creates a new RocksDB backed state.
 *  @param namespaceSerializer The serializer for the namespace.
 */
protected AbstractRocksDBState(
		ColumnFamilyHandle columnFamily,
		TypeSerializer<N> namespaceSerializer,
		SD stateDesc,
		RocksDBKeyedStateBackend<K> backend) {

	this.namespaceSerializer = namespaceSerializer;
	this.backend = backend;

	this.columnFamily = columnFamily;

	writeOptions = new WriteOptions();
	writeOptions.setDisableWAL(true);
	this.stateDesc = Preconditions.checkNotNull(stateDesc, "State Descriptor");

	this.keySerializationStream = new ByteArrayOutputStreamWithPos(128);
	this.keySerializationDataOutputView = new DataOutputViewStreamWrapper(keySerializationStream);
	this.ambiguousKeyPossible = (backend.getKeySerializer().getLength() < 0)
			&& (namespaceSerializer.getLength() < 0);
}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
public AccessControl(RocksDB db, ColumnFamilyHandle ruleCf, ColumnFamilyHandle policyCf) throws RocksDBException {
    this.db = db;
    this.ruleCf = ruleCf;
    this.policyCf = policyCf;

    rules = loadRules(db, ruleCf);
    policies = loadPolicies(db, policyCf);

    rulesBySurt = new RulesBySsurt(rules.values());

    nextRuleId = new AtomicLong(calculateNextId(db, ruleCf));
    nextPolicyId = new AtomicLong(calculateNextId(db, policyCf));

    if (policies.isEmpty()) {
        // create some default policies
        put(new AccessPolicy("Public", "public", "staff"));
        put(new AccessPolicy("Staff Only", "staff"));
        put(new AccessPolicy("No Access"));
    }
}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
/**
 * Obtains an iterator for a column family.
 * 
 * <p>
 * Iterators will be automatically closed by this wrapper.
 * </p>
 * 
 * @param cfName
 * @return
 * @throws RocksDbException
 */
public RocksIterator getIterator(String cfName) throws RocksDbException {
    synchronized (iterators) {
        RocksIterator it = iterators.get(cfName);
        if (it == null) {
            ColumnFamilyHandle cfh = getColumnFamilyHandle(cfName);
            if (cfh == null) {
                throw new RocksDbException.ColumnFamilyNotExists(cfName);
            }
            it = rocksDb.newIterator(cfh, readOptions);
            iterators.put(cfName, it);
        }
        return it;
    }

}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
/**
 * Puts a key/value.
 * 
 * @param cfh
 * @param writeOptions
 * @param key
 * @param value
 * @throws RocksDbException
 */
protected void put(ColumnFamilyHandle cfh, WriteOptions writeOptions, byte[] key, byte[] value)
        throws RocksDbException {
    if (readOnly) {
        throw new RocksDbException.ReadOnlyException("put");
    }
    if (cfh == null) {
        throw new RocksDbException.ColumnFamilyNotExists();
    }
    if (value == null) {
        delete(cfh, writeOptions, key);
    } else {
        try {
            rocksDb.put(cfh, writeOptions != null ? writeOptions : this.writeOptions, key,
                    value);
        } catch (Exception e) {
            throw e instanceof RocksDbException ? (RocksDbException) e
                    : new RocksDbException(e);
        }
    }
}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
@Override
public V get(TimeWindow window, K key) {
    try {
        ColumnFamilyHandle handler = getColumnFamilyHandle(window);
        V ret = null;
        if (key != null) {
            byte[] rawKey = serializer.serialize(key);
            byte[] rawData = rocksDb.get(handler, rawKey);
            ret = rawData != null ? (V) serializer.deserialize(rawData) : null;
        }
        return ret;
    } catch (RocksDBException e) {
        LOG.error("Failed to get value by key-{} for timeWindow={}", key, window);
        throw new RuntimeException(e.getMessage());
    }
}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
@Override
public Collection<K> getAllKeys(TimeWindow window) {
    try {
        ColumnFamilyHandle handler = getColumnFamilyHandle(window);
        Collection<K> keys = new ArrayList<K>();
        RocksIterator itr = rocksDb.newIterator(handler);
        itr.seekToFirst();
        while (itr.isValid()) {
            keys.add((K) serializer.deserialize(itr.key()));
            itr.next();
        }
        return keys;
    } catch (RocksDBException e) {
        LOG.error("Failed to get all keys for timeWindow={}", window);
        throw new RuntimeException(e.getMessage());
    }
}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
@Override
public Object get(String key) {
    for (Entry<Integer, ColumnFamilyHandle> entry : windowHandlers.entrySet()) {
        try {
            byte[] data = ttlDB.get(entry.getValue(), key.getBytes());
            if (data != null) {
                try {
                    return Utils.javaDeserialize(data);
                } catch (Exception e) {
                    LOG.error("Failed to deserialize obj of " + key);
                    ttlDB.remove(entry.getValue(), key.getBytes());
                    return null;
                }
            }
        } catch (Exception ignored) {
        }
    }

    return null;
}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
@Override
public NodeTable buildBaseNodeTable(String name) {
    String nDat = name+":NodeHash2Id";
    String nIndex = name+":NodeId2Node";
    
    ColumnFamilyHandle h = BuildR.findHandle(trdb, nDat);
    RecordFactory recordFactory = new RecordFactory(SystemTDB.LenNodeHash, SystemTDB.SizeOfNodeId) ;
    Index hashNodeIndex = buildRangeIndex(recordFactory, nIndex);
    NodeTable nt = new RocksNodeTable(trdb.rdb, h, hashNodeIndex, ByteCodecThrift.create()); 
    return nt;
}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
private ByteStore newDB(String name) throws RocksDBException {
  if(inMemory){
    return new MapStore(name);
  }else{
    final ColumnFamilyDescriptor columnFamilyDescriptor = new ColumnFamilyDescriptor(name.getBytes(UTF_8));
    ColumnFamilyHandle handle = db.createColumnFamily(columnFamilyDescriptor);
    return new RocksDBStore(name, columnFamilyDescriptor, handle, db, stripeCount);
  }
}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
@Before
public void setUpStore() {
  ColumnFamilyHandle handle = rocksDBResource.get().getDefaultColumnFamily();
  store = new RocksDBStore("test", new ColumnFamilyDescriptor("test".getBytes(UTF_8)), handle, rocksDBResource.get(), 4);

  // Making sure test is repeatable
  Random random = new Random(42);
  for(int i = 0; i < 1 << 16; i++ ) {
    store.put(newRandomValue(random), newRandomValue(random));
  }
}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
@Override
public void createCollection(final String databaseName, final String collectionName) throws StorageException {
    try {
        final ColumnFamilyHandle column = db.createColumnFamily(new ColumnFamilyDescriptor(collectionName
                .getBytes()));

        collections.add(column);

    } catch (RocksDBException e) {
        throw new StorageException("Collection not created", e);
    }
}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
public void open() throws RocksDBException {
    if (db != null) throw new IllegalStateException("Database already open");
    storagePath.toFile().mkdirs();
    defaultColumnFamily = addColumnFamily("default", Serdes.stringSerdes()); // must exist

    options = new DBOptions();
    options.setCreateIfMissing(true);
    options.setCreateMissingColumnFamilies(true);
    //options.setBytesPerSync(1024 * 1024);

    List<ColumnFamilyDescriptor> descriptors = new ArrayList<>();
    List<ColumnFamilyHandle> handles = new ArrayList<>();
    List<ColumnFamily<?, ?>> families = new ArrayList<>();
    try {
        for (Map.Entry<String, ColumnFamily<?, ?>> entry : columnFamilies.entrySet()) {
            descriptors.add(new ColumnFamilyDescriptor(entry.getKey().getBytes("UTF-8")));
            families.add(entry.getValue());
        }
    } catch (UnsupportedEncodingException e) {
        throw new RuntimeException(e);
    }

    db = RocksDB.open(options, storagePath.toString(), descriptors, handles);
    if (families.size() != handles.size()) {
        throw new IllegalStateException("Unexpected number of column family handles");
    }
    for (int i = 0; i < families.size(); i++) {
        families.get(i).setDBHandle(db, handles.get(i));
    }
    log.info("Opened database {} at path {}", name, storagePath);
}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
/**
 * Creates a new {@code RocksDBMapState}.
 *
 * @param namespaceSerializer The serializer for the namespace.
 * @param stateDesc The state identifier for the state.
 */
public RocksDBMapState(ColumnFamilyHandle columnFamily,
		TypeSerializer<N> namespaceSerializer,
		MapStateDescriptor<UK, UV> stateDesc,
		RocksDBKeyedStateBackend<K> backend) {

	super(columnFamily, namespaceSerializer, stateDesc, backend);

	this.userKeySerializer = stateDesc.getKeySerializer();
	this.userValueSerializer = stateDesc.getValueSerializer();

	writeOptions = new WriteOptions();
	writeOptions.setDisableWAL(true);
}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
/**
 * Creates a new {@code RocksDBListState}.
 *
 * @param namespaceSerializer The serializer for the namespace.
 * @param stateDesc The state identifier for the state. This contains name
 *                     and can create a default state value.
 */
public RocksDBListState(ColumnFamilyHandle columnFamily,
		TypeSerializer<N> namespaceSerializer,
		ListStateDescriptor<V> stateDesc,
		RocksDBKeyedStateBackend<K> backend) {

	super(columnFamily, namespaceSerializer, stateDesc, backend);
	this.valueSerializer = stateDesc.getElementSerializer();

	writeOptions = new WriteOptions();
	writeOptions.setDisableWAL(true);
}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
/**
 * Creates a new {@code RocksDBValueState}.
 *
 * @param namespaceSerializer The serializer for the namespace.
 * @param stateDesc The state identifier for the state. This contains name
 *                           and can create a default state value.
 */
public RocksDBValueState(ColumnFamilyHandle columnFamily,
		TypeSerializer<N> namespaceSerializer,
		ValueStateDescriptor<V> stateDesc,
		RocksDBKeyedStateBackend<K> backend) {

	super(columnFamily, namespaceSerializer, stateDesc, backend);
	this.valueSerializer = stateDesc.getSerializer();

	writeOptions = new WriteOptions();
	writeOptions.setDisableWAL(true);
}
 

import org.rocksdb.ColumnFamilyHandle; //导入依赖的package包/类
/**
 * Creates a new {@code RocksDBReducingState}.
 *
 * @param namespaceSerializer The serializer for the namespace.
 * @param stateDesc The state identifier for the state. This contains name
 *                     and can create a default state value.
 */
public RocksDBReducingState(ColumnFamilyHandle columnFamily,
		TypeSerializer<N> namespaceSerializer,
		ReducingStateDescriptor<V> stateDesc,
		RocksDBKeyedStateBackend<K> backend) {

	super(columnFamily, namespaceSerializer, stateDesc, backend);
	this.valueSerializer = stateDesc.getSerializer();
	this.reduceFunction = stateDesc.getReduceFunction();

	writeOptions = new WriteOptions();
	writeOptions.setDisableWAL(true);
}