import org.rocksdb.ColumnFamilyDescriptor; //导入依赖的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.ColumnFamilyDescriptor; //导入依赖的package包/类
private void prepareColumnFamilyDescriptors() throws RocksDBException {
    if (columnFamilies == null || columnFamilies.size() == 0) {
        if (columnFamilies == null) {
            columnFamilies = new HashSet<>();
        }
        String[] cfList = RocksDbUtils.getColumnFamilyList(directory.getAbsolutePath());
        for (String cf : cfList) {
            columnFamilies.add(RocksDbUtils.buildColumnFamilyDescriptor(cf));
        }
    }

    boolean hasDefaultCf = false;
    for (ColumnFamilyDescriptor cfd : columnFamilies) {
        if (Arrays.equals(cfd.columnFamilyName(), RocksDB.DEFAULT_COLUMN_FAMILY)) {
            hasDefaultCf = true;
            break;
        }
    }
    if (!hasDefaultCf) {
        columnFamilies.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
    }
}
 

import org.rocksdb.ColumnFamilyDescriptor; //导入依赖的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.ColumnFamilyDescriptor; //导入依赖的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.ColumnFamilyDescriptor; //导入依赖的package包/类
@Override
public void init() throws StorageException {
    try {
        collections = new ArrayList<>();
        descriptors = new ArrayList<>();

        final File f = new File(dataPath);
        boolean newDB = !f.exists();
        if (newDB) {
            f.mkdirs();
            logger.info("Created database folder : " + dataPath);
            descriptors.add(new ColumnFamilyDescriptor(
                    RocksDB.DEFAULT_COLUMN_FAMILY, new ColumnFamilyOptions()));

        } else {
            final List<byte[]> columns = RocksDB.listColumnFamilies(options, dataPath);
            if (columns != null) {
                for(byte[] column: columns) {
                    descriptors.add(new ColumnFamilyDescriptor(
                                    column, new ColumnFamilyOptions()));
                }
            }

        }

        dbOptions = new DBOptions()
                        .setCreateIfMissing(true);

        db = RocksDB.open(dbOptions, dataPath, descriptors, collections);

    } catch (RocksDBException e) {
        throw new StorageException("Error initialising Storage", e);
    }
}
 

import org.rocksdb.ColumnFamilyDescriptor; //导入依赖的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.ColumnFamilyDescriptor; //导入依赖的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.ColumnFamilyDescriptor; //导入依赖的package包/类
private RocksDB openDB(
	String path,
	List<ColumnFamilyDescriptor> stateColumnFamilyDescriptors,
	List<ColumnFamilyHandle> stateColumnFamilyHandles) throws IOException {

	List<ColumnFamilyDescriptor> columnFamilyDescriptors =
		new ArrayList<>(1 + stateColumnFamilyDescriptors.size());

	columnFamilyDescriptors.addAll(stateColumnFamilyDescriptors);

	// we add the required descriptor for the default CF in last position.
	columnFamilyDescriptors.add(new ColumnFamilyDescriptor(DEFAULT_COLUMN_FAMILY_NAME_BYTES, columnOptions));

	RocksDB dbRef;

	try {
		dbRef = RocksDB.open(
			Preconditions.checkNotNull(dbOptions),
			Preconditions.checkNotNull(path),
			columnFamilyDescriptors,
			stateColumnFamilyHandles);
	} catch (RocksDBException e) {
		throw new IOException("Error while opening RocksDB instance.", e);
	}

	// requested + default CF
	Preconditions.checkState(1 + stateColumnFamilyDescriptors.size() == stateColumnFamilyHandles.size(),
		"Not all requested column family handles have been created");

	return dbRef;
}
 

import org.rocksdb.ColumnFamilyDescriptor; //导入依赖的package包/类
public void initDb(List<Integer> list) throws Exception {
    LOG.info("Begin to init rocksDB of {}", rootDir);

    DBOptions dbOptions = null;

    List<ColumnFamilyDescriptor> columnFamilyNames = new ArrayList<ColumnFamilyDescriptor>();
    columnFamilyNames.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
    for (Integer timeout : list) {
        columnFamilyNames.add(new ColumnFamilyDescriptor(String.valueOf(timeout).getBytes()));
    }

    List<Integer> ttlValues = new ArrayList<Integer>();
    // Default column family with infinite lifetime
    // ATTENSION, the first must be 0, RocksDB.java API has this limitation
    ttlValues.add(0);
    // new column family with list second ttl
    ttlValues.addAll(list);

    try {
        dbOptions = new DBOptions().setCreateMissingColumnFamilies(true).setCreateIfMissing(true);

        List<ColumnFamilyHandle> columnFamilyHandleList = new ArrayList<ColumnFamilyHandle>();

        ttlDB = TtlDB.open(dbOptions, rootDir, columnFamilyNames, columnFamilyHandleList, ttlValues, false);

        for (int i = 0; i < ttlValues.size(); i++) {
            windowHandlers.put(ttlValues.get(i), columnFamilyHandleList.get(i));
        }

        LOG.info("Successfully init rocksDB of {}", rootDir);
    } finally {

        if (dbOptions != null) {
            dbOptions.dispose();
        }
    }
}
 

import org.rocksdb.ColumnFamilyDescriptor; //导入依赖的package包/类
/**
 * Builds a list of {@link ColumnFamilyDescriptor}s, specifying options.
 * 
 * @param cfOptions
 * @param cfNames
 * @return
 */
public static List<ColumnFamilyDescriptor> buildColumnFamilyDescriptors(
        ColumnFamilyOptions cfOptions, String... cfNames) {
    List<ColumnFamilyDescriptor> result = new ArrayList<>();
    if (cfNames != null) {
        for (String cfName : cfNames) {
            result.add(buildColumnFamilyDescriptor(cfOptions, cfName));
        }
    }
    return result;
}
 

import org.rocksdb.ColumnFamilyDescriptor; //导入依赖的package包/类
synchronized public RocksDbWrapper setColumnFamilies(
        Collection<ColumnFamilyDescriptor> columnFamilies) {
    this.columnFamilies.clear();
    this.columnFamilyNames.clear();
    for (Entry<String, ColumnFamilyHandle> entry : this.columnFamilyHandles.entrySet()) {
        RocksDbUtils.closeRocksObjects(entry.getValue());
    }
    if (columnFamilies != null) {
        this.columnFamilies.addAll(columnFamilies);
        this.columnFamilies.iterator().forEachRemaining(x -> {
            this.columnFamilyNames.add(new String(x.columnFamilyName(), RocksDbUtils.UTF8));
        });
    }
    return this;
}
 

import org.rocksdb.ColumnFamilyDescriptor; //导入依赖的package包/类
protected void prepareColumnFamilyDescriptors() throws RocksDbException {
    if (columnFamilies == null || columnFamilies.size() == 0) {
        if (columnFamilies == null) {
            columnFamilies = new HashSet<>();
        }
        try {
            String[] cfList = RocksDbUtils.getColumnFamilyList(directory.getAbsolutePath());
            for (String cf : cfList) {
                columnFamilies.add(RocksDbUtils.buildColumnFamilyDescriptor(cf));
            }
        } catch (Exception e) {
            throw e instanceof RocksDbException ? (RocksDbException) e
                    : new RocksDbException(e);
        }
    }

    boolean hasDefaultCf = false;
    for (ColumnFamilyDescriptor cfd : columnFamilies) {
        if (Arrays.equals(cfd.columnFamilyName(), RocksDB.DEFAULT_COLUMN_FAMILY)) {
            hasDefaultCf = true;
            break;
        }
    }
    if (!hasDefaultCf) {
        columnFamilies.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
    }
}
 

import org.rocksdb.ColumnFamilyDescriptor; //导入依赖的package包/类
private static void rocksDbTest(RocksDB db, List<ColumnFamilyHandle> handlers) {
    try {
        ColumnFamilyHandle handler1 = null;
        ColumnFamilyHandle handler2 = null;
        if (handlers.size() > 0) {
            // skip default column family
            handler1 = handlers.get(1);
            handler2 = handlers.get(2);
        } else {
            handler1 = db.createColumnFamily(new ColumnFamilyDescriptor("test1".getBytes()));
            handler2 = db.createColumnFamily(new ColumnFamilyDescriptor("test2".getBytes()));
        }
        int startValue1 = getStartValue(db, handler1);
        int startValue2 = getStartValue(db, handler2);;

        Checkpoint cp = Checkpoint.create(db);
   
        if (isCompaction) {
            db.compactRange();
            LOG.info("Compaction!");
        }

        long flushWaitTime = System.currentTimeMillis() + flushInterval;
        for (int i = 0; i < putNum || putNum == -1; i++) {
            db.put(handler1, String.valueOf(i % 1000).getBytes(), String.valueOf(startValue1 + i).getBytes());
            db.put(handler2, String.valueOf(i % 1000).getBytes(), String.valueOf(startValue2 + i).getBytes());
            if (isFlush && flushWaitTime <= System.currentTimeMillis()) {
                db.flush(new FlushOptions());
                if (isCheckpoint) {
                    cp.createCheckpoint(cpPath + "/" + i);
                }
                flushWaitTime = System.currentTimeMillis() + flushInterval;
            }
        }
    } catch (RocksDBException e) {
        LOG.error("Failed to put or flush", e);
    }
}
 

import org.rocksdb.ColumnFamilyDescriptor; //导入依赖的package包/类
private ColumnFamilyHandle getColumnFamilyHandle(TimeWindow window) throws RocksDBException {
    ColumnFamilyHandle handler = null;
    if (window == null) {
        handler = rocksDb.getDefaultColumnFamily();
    } else {
        handler = windowToCFHandler.get(window);
        if (handler == null) {
            handler = rocksDb.createColumnFamily(new ColumnFamilyDescriptor(serializer.serialize(window)));
            windowToCFHandler.put(window, handler);
        }
    }
    return handler;
}
 

import org.rocksdb.ColumnFamilyDescriptor; //导入依赖的package包/类
public void initDb(List<Integer> list) throws Exception {
    LOG.info("Begin to init rocksDB of {}", rootDir);

    DBOptions dbOptions = null;

    List<ColumnFamilyDescriptor> columnFamilyNames = new ArrayList<>();
    columnFamilyNames.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
    for (Integer timeout : list) {
        columnFamilyNames.add(new ColumnFamilyDescriptor(String.valueOf(timeout).getBytes()));
    }

    List<Integer> ttlValues = new ArrayList<>();
    // Default column family with infinite TTL
    // NOTE that the first must be 0, RocksDB.java API has this limitation
    ttlValues.add(0);
    // new column family with list second ttl
    ttlValues.addAll(list);

    try {
        dbOptions = new DBOptions().setCreateMissingColumnFamilies(true).setCreateIfMissing(true);
        List<ColumnFamilyHandle> columnFamilyHandleList = new ArrayList<>();
        ttlDB = TtlDB.open(dbOptions, rootDir, columnFamilyNames, columnFamilyHandleList, ttlValues, false);
        for (int i = 0; i < ttlValues.size(); i++) {
            windowHandlers.put(ttlValues.get(i), columnFamilyHandleList.get(i));
        }
        LOG.info("Successfully init rocksDB of {}", rootDir);
    } finally {

        if (dbOptions != null) {
            dbOptions.dispose();
        }
    }
}
 

import org.rocksdb.ColumnFamilyDescriptor; //导入依赖的package包/类
public void start() throws Exception {
  if(inMemory){
    return;
  }

  final String baseDirectory =
      CoreStoreProviderImpl.MODE == ForcedMemoryMode.DISK && this.baseDirectory == null
      ? Files.createTempDir().toString() : this.baseDirectory.toString();

  final File dbDirectory = new File(baseDirectory, CATALOG_STORE_NAME);
  if (dbDirectory.exists()) {
    if (!dbDirectory.isDirectory()) {
      throw new DatastoreException(
          String.format("Invalid path %s for local catalog db, not a directory.", dbDirectory.getAbsolutePath()));
    }
  } else {
    if (!dbDirectory.mkdirs()) {
      throw new DatastoreException(
          String.format("Failed to create directory %s for local catalog db.", dbDirectory.getAbsolutePath()));
    }
  }

  RocksDB.loadLibrary();
  final String path = dbDirectory.toString();

  final List<byte[]> families;
  try(final Options options = new Options()) {
    options.setCreateIfMissing(true);
    // get a list of existing families.
    families = new ArrayList<>(RocksDB.listColumnFamilies(options, path));
  }


  // add the default family (we don't use this)
  families.add(RocksDB.DEFAULT_COLUMN_FAMILY);
  final Function<byte[], ColumnFamilyDescriptor> func = new Function<byte[], ColumnFamilyDescriptor>(){
    @Override
    public ColumnFamilyDescriptor apply(byte[] input) {
      return new ColumnFamilyDescriptor(input);
    }
  };

  List<ColumnFamilyHandle> familyHandles = new ArrayList<>();
  try(final DBOptions dboptions = new DBOptions()) {
    dboptions.setCreateIfMissing(true);
    db = RocksDB.open(dboptions, path.toString(), Lists.transform(families, func), familyHandles);
  }
  // create an output list to be populated when we open the db.

  // populate the local cache with the existing tables.
  for(int i =0; i < families.size(); i++){
    byte[] family = families.get(i);
    if(Arrays.equals(family, RocksDB.DEFAULT_COLUMN_FAMILY)){
      // we don't allow use of the default handle.
      defaultHandle = familyHandles.get(i);
    } else {
      String name = new String(family, UTF_8);
      RocksDBStore store = new RocksDBStore(name, new ColumnFamilyDescriptor(family), familyHandles.get(i), db, stripeCount);
      maps.put(name, store);
    }
  }
}
 

import org.rocksdb.ColumnFamilyDescriptor; //导入依赖的package包/类
/**
 * Restore the KV-state / ColumnFamily meta data for all key-groups referenced by the current state handle.
 *
 * @throws IOException
 * @throws ClassNotFoundException
 * @throws RocksDBException
 */
private void restoreKVStateMetaData() throws IOException, StateMigrationException, RocksDBException {

	KeyedBackendSerializationProxy<K> serializationProxy =
		new KeyedBackendSerializationProxy<>(rocksDBKeyedStateBackend.userCodeClassLoader);

	serializationProxy.read(currentStateHandleInView);

	// check for key serializer compatibility; this also reconfigures the
	// key serializer to be compatible, if it is required and is possible
	if (CompatibilityUtil.resolveCompatibilityResult(
		serializationProxy.getKeySerializer(),
		UnloadableDummyTypeSerializer.class,
		serializationProxy.getKeySerializerConfigSnapshot(),
		rocksDBKeyedStateBackend.keySerializer)
		.isRequiresMigration()) {

		// TODO replace with state migration; note that key hash codes need to remain the same after migration
		throw new StateMigrationException("The new key serializer is not compatible to read previous keys. " +
			"Aborting now since state migration is currently not available");
	}

	this.keygroupStreamCompressionDecorator = serializationProxy.isUsingKeyGroupCompression() ?
		SnappyStreamCompressionDecorator.INSTANCE : UncompressedStreamCompressionDecorator.INSTANCE;

	List<RegisteredKeyedBackendStateMetaInfo.Snapshot<?, ?>> restoredMetaInfos =
		serializationProxy.getStateMetaInfoSnapshots();
	currentStateHandleKVStateColumnFamilies = new ArrayList<>(restoredMetaInfos.size());
	//rocksDBKeyedStateBackend.restoredKvStateMetaInfos = new HashMap<>(restoredMetaInfos.size());

	for (RegisteredKeyedBackendStateMetaInfo.Snapshot<?, ?> restoredMetaInfo : restoredMetaInfos) {

		Tuple2<ColumnFamilyHandle, RegisteredKeyedBackendStateMetaInfo<?, ?>> registeredColumn =
			rocksDBKeyedStateBackend.kvStateInformation.get(restoredMetaInfo.getName());

		if (registeredColumn == null) {
			byte[] nameBytes = restoredMetaInfo.getName().getBytes(ConfigConstants.DEFAULT_CHARSET);

			ColumnFamilyDescriptor columnFamilyDescriptor = new ColumnFamilyDescriptor(
				nameBytes,
				rocksDBKeyedStateBackend.columnOptions);

			RegisteredKeyedBackendStateMetaInfo<?, ?> stateMetaInfo =
				new RegisteredKeyedBackendStateMetaInfo<>(
					restoredMetaInfo.getStateType(),
					restoredMetaInfo.getName(),
					restoredMetaInfo.getNamespaceSerializer(),
					restoredMetaInfo.getStateSerializer());

			rocksDBKeyedStateBackend.restoredKvStateMetaInfos.put(restoredMetaInfo.getName(), restoredMetaInfo);

			ColumnFamilyHandle columnFamily = rocksDBKeyedStateBackend.db.createColumnFamily(columnFamilyDescriptor);

			registeredColumn = new Tuple2<>(columnFamily, stateMetaInfo);
			rocksDBKeyedStateBackend.kvStateInformation.put(stateMetaInfo.getName(), registeredColumn);

		} else {
			// TODO with eager state registration in place, check here for serializer migration strategies
		}
		currentStateHandleKVStateColumnFamilies.add(registeredColumn.f0);
	}
}
 

import org.rocksdb.ColumnFamilyDescriptor; //导入依赖的package包/类
public void ttlDbOpenWithColumnFamilies() throws RocksDBException, InterruptedException {
    DBOptions dbOptions = null;
    TtlDB ttlDB = null;
    List<ColumnFamilyDescriptor> cfNames = new ArrayList<ColumnFamilyDescriptor>();
    List<ColumnFamilyHandle> columnFamilyHandleList = new ArrayList<ColumnFamilyHandle>();
    cfNames.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
    cfNames.add(new ColumnFamilyDescriptor("new_cf".getBytes()));

    List<Integer> ttlValues = new ArrayList<Integer>();
    // Default column family with infinite lifetime
    ttlValues.add(0);
    // new column family with 1 second ttl
    ttlValues.add(1);

    try {
        System.out.println("Begin to open db");
        dbOptions = new DBOptions().setCreateMissingColumnFamilies(true).setCreateIfMissing(true);
        ttlDB = TtlDB.open(dbOptions, rootDir, cfNames, columnFamilyHandleList, ttlValues, false);

        System.out.println("Successfully open db " + rootDir);

        ttlDB.put("key".getBytes(), "value".getBytes());
        assertThat(ttlDB.get("key".getBytes())).isEqualTo("value".getBytes());
        ttlDB.put(columnFamilyHandleList.get(1), "key".getBytes(), "value".getBytes());
        assertThat(ttlDB.get(columnFamilyHandleList.get(1), "key".getBytes())).isEqualTo("value".getBytes());
        TimeUnit.SECONDS.sleep(2);

        ttlDB.compactRange();
        ttlDB.compactRange(columnFamilyHandleList.get(1));

        assertThat(ttlDB.get("key".getBytes())).isNotNull();
        assertThat(ttlDB.get(columnFamilyHandleList.get(1), "key".getBytes())).isNull();

    } finally {
        for (ColumnFamilyHandle columnFamilyHandle : columnFamilyHandleList) {
            columnFamilyHandle.dispose();
        }
        if (ttlDB != null) {
            ttlDB.close();
        }
        if (dbOptions != null) {
            dbOptions.dispose();
        }
    }
}
 

import org.rocksdb.ColumnFamilyDescriptor; //导入依赖的package包/类
public RocksDbWrapper setColumnFamilies(Collection<ColumnFamilyDescriptor> columnFamilies) {
    if (columnFamilies != null) {
        this.columnFamilies.addAll(columnFamilies);
    }
    return this;
}
 

import org.rocksdb.ColumnFamilyDescriptor; //导入依赖的package包/类
public Collection<ColumnFamilyDescriptor> getColumnFamilies() {
    return Collections.unmodifiableSet(columnFamilies);
}