import org.rocksdb.DBOptions; //导入依赖的package包/类
RocksDBKeyedStateBackend2(
		final String operatorIdentifier,
		final ClassLoader userCodeClassLoader,
		final File instanceBasePath,
		final DBOptions dbOptions,
		final ColumnFamilyOptions columnFamilyOptions,
		final TaskKvStateRegistry kvStateRegistry,
		final TypeSerializer<K> keySerializer,
		final int numberOfKeyGroups,
		final KeyGroupRange keyGroupRange,
		final ExecutionConfig executionConfig) throws Exception {

	super(operatorIdentifier, userCodeClassLoader,
		instanceBasePath,
		dbOptions, columnFamilyOptions, kvStateRegistry, keySerializer,
		numberOfKeyGroups, keyGroupRange, executionConfig, false);
}
 

import org.rocksdb.DBOptions; //导入依赖的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.DBOptions; //导入依赖的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.DBOptions; //导入依赖的package包/类
/**
 * Tests list serialization and deserialization match.
 *
 * @see KvStateRequestSerializerTest#testListSerialization()
 * KvStateRequestSerializerTest#testListSerialization() using the heap state back-end
 * test
 */
@Test
public void testListSerialization() throws Exception {
	final long key = 0L;

	// objects for RocksDB state list serialisation
	DBOptions dbOptions = PredefinedOptions.DEFAULT.createDBOptions();
	dbOptions.setCreateIfMissing(true);
	ColumnFamilyOptions columnFamilyOptions = PredefinedOptions.DEFAULT.createColumnOptions();
	final RocksDBKeyedStateBackend2<Long> longHeapKeyedStateBackend =
		new RocksDBKeyedStateBackend2<>(
			"no-op",
			ClassLoader.getSystemClassLoader(),
			temporaryFolder.getRoot(),
			dbOptions,
			columnFamilyOptions,
			mock(TaskKvStateRegistry.class),
			LongSerializer.INSTANCE,
			1, new KeyGroupRange(0, 0),
			new ExecutionConfig()
		);
	longHeapKeyedStateBackend.restore(null);
	longHeapKeyedStateBackend.setCurrentKey(key);

	final InternalListState<VoidNamespace, Long> listState = longHeapKeyedStateBackend
		.createListState(VoidNamespaceSerializer.INSTANCE,
			new ListStateDescriptor<>("test", LongSerializer.INSTANCE));

	KvStateRequestSerializerTest.testListSerialization(key, listState);
}
 

import org.rocksdb.DBOptions; //导入依赖的package包/类
/**
 * Tests map serialization and deserialization match.
 *
 * @see KvStateRequestSerializerTest#testMapSerialization()
 * KvStateRequestSerializerTest#testMapSerialization() using the heap state back-end
 * test
 */
@Test
public void testMapSerialization() throws Exception {
	final long key = 0L;

	// objects for RocksDB state list serialisation
	DBOptions dbOptions = PredefinedOptions.DEFAULT.createDBOptions();
	dbOptions.setCreateIfMissing(true);
	ColumnFamilyOptions columnFamilyOptions = PredefinedOptions.DEFAULT.createColumnOptions();
	final RocksDBKeyedStateBackend<Long> longHeapKeyedStateBackend =
		new RocksDBKeyedStateBackend<>(
			"no-op",
			ClassLoader.getSystemClassLoader(),
			temporaryFolder.getRoot(),
			dbOptions,
			columnFamilyOptions,
			mock(TaskKvStateRegistry.class),
			LongSerializer.INSTANCE,
			1, new KeyGroupRange(0, 0),
			new ExecutionConfig(),
			false);
	longHeapKeyedStateBackend.restore(null);
	longHeapKeyedStateBackend.setCurrentKey(key);

	final InternalMapState<VoidNamespace, Long, String> mapState = (InternalMapState<VoidNamespace, Long, String>)
			longHeapKeyedStateBackend.getPartitionedState(
					VoidNamespace.INSTANCE,
					VoidNamespaceSerializer.INSTANCE,
					new MapStateDescriptor<>("test", LongSerializer.INSTANCE, StringSerializer.INSTANCE));

	KvStateRequestSerializerTest.testMapSerialization(key, mapState);
}
 

import org.rocksdb.DBOptions; //导入依赖的package包/类
/**
 * Gets the RocksDB {@link DBOptions} to be used for all RocksDB instances.
 */
public DBOptions getDbOptions() {
	// initial options from pre-defined profile
	DBOptions opt = predefinedOptions.createDBOptions();

	// add user-defined options, if specified
	if (optionsFactory != null) {
		opt = optionsFactory.createDBOptions(opt);
	}

	// add necessary default options
	opt = opt.setCreateIfMissing(true);

	return opt;
}
 

import org.rocksdb.DBOptions; //导入依赖的package包/类
@Test
public void testCorrectMergeOperatorSet() throws IOException {

	final ColumnFamilyOptions columnFamilyOptions = spy(new ColumnFamilyOptions());
	RocksDBKeyedStateBackend<Integer> test = null;
	try {
		test = new RocksDBKeyedStateBackend<>(
			"test",
			Thread.currentThread().getContextClassLoader(),
			tempFolder.newFolder(),
			mock(DBOptions.class),
			columnFamilyOptions,
			mock(TaskKvStateRegistry.class),
			IntSerializer.INSTANCE,
			1,
			new KeyGroupRange(0, 0),
			new ExecutionConfig(),
			enableIncrementalCheckpointing);

		verify(columnFamilyOptions, Mockito.times(1))
			.setMergeOperatorName(RocksDBKeyedStateBackend.MERGE_OPERATOR_NAME);
	} finally {
		if (test != null) {
			IOUtils.closeQuietly(test);
			test.dispose();
		}
		columnFamilyOptions.close();
	}
}
 

import org.rocksdb.DBOptions; //导入依赖的package包/类
@Test
public void testPredefinedOptionsEnum() {
	for (PredefinedOptions o : PredefinedOptions.values()) {
		try (DBOptions opt = o.createDBOptions()) {
			assertNotNull(opt);
		}
	}
}
 

import org.rocksdb.DBOptions; //导入依赖的package包/类
/**
 * Tests list serialization and deserialization match.
 *
 * @see KvStateRequestSerializerTest#testListSerialization()
 * KvStateRequestSerializerTest#testListSerialization() using the heap state back-end
 * test
 */
@Test
public void testListSerialization() throws Exception {
	final long key = 0L;

	// objects for RocksDB state list serialisation
	DBOptions dbOptions = PredefinedOptions.DEFAULT.createDBOptions();
	dbOptions.setCreateIfMissing(true);
	ColumnFamilyOptions columnFamilyOptions = PredefinedOptions.DEFAULT.createColumnOptions();
	final RocksDBKeyedStateBackend2<Long> longHeapKeyedStateBackend =
		new RocksDBKeyedStateBackend2<>(
			"no-op",
			ClassLoader.getSystemClassLoader(),
			temporaryFolder.getRoot(),
			dbOptions,
			columnFamilyOptions,
			mock(TaskKvStateRegistry.class),
			LongSerializer.INSTANCE,
			1, new KeyGroupRange(0, 0),
			new ExecutionConfig()
		);
	longHeapKeyedStateBackend.restore(null);
	longHeapKeyedStateBackend.setCurrentKey(key);

	final InternalListState<VoidNamespace, Long> listState = longHeapKeyedStateBackend
		.createListState(VoidNamespaceSerializer.INSTANCE,
			new ListStateDescriptor<>("test", LongSerializer.INSTANCE));

	KvStateRequestSerializerTest.testListSerialization(key, listState);
	longHeapKeyedStateBackend.dispose();
}
 

import org.rocksdb.DBOptions; //导入依赖的package包/类
/**
 * Tests map serialization and deserialization match.
 *
 * @see KvStateRequestSerializerTest#testMapSerialization()
 * KvStateRequestSerializerTest#testMapSerialization() using the heap state back-end
 * test
 */
@Test
public void testMapSerialization() throws Exception {
	final long key = 0L;

	// objects for RocksDB state list serialisation
	DBOptions dbOptions = PredefinedOptions.DEFAULT.createDBOptions();
	dbOptions.setCreateIfMissing(true);
	ColumnFamilyOptions columnFamilyOptions = PredefinedOptions.DEFAULT.createColumnOptions();
	final RocksDBKeyedStateBackend<Long> longHeapKeyedStateBackend =
		new RocksDBKeyedStateBackend<>(
			"no-op",
			ClassLoader.getSystemClassLoader(),
			temporaryFolder.getRoot(),
			dbOptions,
			columnFamilyOptions,
			mock(TaskKvStateRegistry.class),
			LongSerializer.INSTANCE,
			1, new KeyGroupRange(0, 0),
			new ExecutionConfig(),
			false);
	longHeapKeyedStateBackend.restore(null);
	longHeapKeyedStateBackend.setCurrentKey(key);

	final InternalMapState<VoidNamespace, Long, String> mapState = (InternalMapState<VoidNamespace, Long, String>)
			longHeapKeyedStateBackend.getPartitionedState(
					VoidNamespace.INSTANCE,
					VoidNamespaceSerializer.INSTANCE,
					new MapStateDescriptor<>("test", LongSerializer.INSTANCE, StringSerializer.INSTANCE));

	KvStateRequestSerializerTest.testMapSerialization(key, mapState);
	longHeapKeyedStateBackend.dispose();
}
 

import org.rocksdb.DBOptions; //导入依赖的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.DBOptions; //导入依赖的package包/类
/**
 * Builds RocksDb DBOptions.
 * 
 * @param maxBackgroundFlushes
 *            high priority threads, usually 1 or 2 would be enough
 * @param maxBackgroundCompactions
 *            low priority threads, between 1 - num_cpu_cores
 * @param maxBackgroundThreads
 *            {@code >= maxBackgroundFlushes, maxBackgroundCompactions}
 * @param maxLogFileSize
 * @return
 */
public static DBOptions buildDbOptions(int maxBackgroundFlushes, int maxBackgroundCompactions,
        int maxBackgroundThreads, long maxLogFileSize) {
    DBOptions dbOptions = new DBOptions();
    dbOptions.setCreateIfMissing(true).setCreateMissingColumnFamilies(true)
            .setErrorIfExists(false);
    dbOptions.setMaxBackgroundFlushes(maxBackgroundFlushes)
            .setMaxBackgroundCompactions(maxBackgroundCompactions)
            .setIncreaseParallelism(maxBackgroundThreads);
    dbOptions.setAllowMmapReads(true).setAllowMmapWrites(true);
    dbOptions.setMaxOpenFiles(-1);
    dbOptions.setKeepLogFileNum(100).setLogFileTimeToRoll(3600)
            .setMaxLogFileSize(maxLogFileSize);
    return dbOptions;
}
 

import org.rocksdb.DBOptions; //导入依赖的package包/类
public RocksDbWrapper setDbOptions(DBOptions dbOptions) {
    if (this.dbOptions != null) {
        RocksDbUtils.closeRocksObjects(this.dbOptions);
    }
    this.dbOptions = dbOptions;
    myOwnDbOptions = false;
    return this;
}
 

import org.rocksdb.DBOptions; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
    File storageDir = new File("/tmp/rocksdb");
    FileUtils.deleteQuietly(storageDir);
    storageDir.mkdirs();

    DBOptions dbOptions = RocksDbUtils.buildDbOptions().setMaxTotalWalSize(16);

    try (RocksDbWrapper rocksDbWrapper = RocksDbWrapper.openReadWrite(storageDir, dbOptions,
            null, null, null)) {
        System.out.println("ColumnFamilies: " + rocksDbWrapper.getColumnFamilyNames());

        final int NUM_ITEMS = 1000000;

        String cfName = RocksDbWrapper.DEFAULT_COLUMN_FAMILY;
        // ColumnFamilyHandle cfh = rocskDb.getColumnFamilyHandle(cfName);
        long t1 = System.currentTimeMillis();
        for (int i = 0; i < NUM_ITEMS; i++) {
            String key = QueueUtils.IDGEN.generateId128Hex();
            String value = String.valueOf(i + 1);
            rocksDbWrapper.put(cfName, key, value);
        }
        long t2 = System.currentTimeMillis();
        long d = t2 - t1;
        System.out.println("Wrote " + NUM_ITEMS + " in " + d + " ms ("
                + Math.round(NUM_ITEMS * 1000.0 / d) + " items/sec)");

        t1 = System.currentTimeMillis();
        rocksDbWrapper.compactRange();
        System.out.println(
                "Compact Range finished in " + (System.currentTimeMillis() - t1) + "ms");

        System.out.println("Num keys: " + rocksDbWrapper.getEstimateNumKeys("default"));
        System.out.println(rocksDbWrapper.getProperty("default", "rocksdb.stats"));
    }

    try (RocksDbWrapper rocksDbReadonly = RocksDbWrapper.openReadOnly(storageDir)) {
        System.out.println("Num keys: " + rocksDbReadonly.getEstimateNumKeys("default"));
    }
}
 

import org.rocksdb.DBOptions; //导入依赖的package包/类
/**
 * Builds RocksDb {@link DBOptions}.
 * 
 * @param maxBackgroundFlushes
 *            high priority threads, usually 1 or 2 would be enough
 * @param maxBackgroundCompactions
 *            low priority threads, between 1 - num_cpu_cores
 * @param maxBackgroundThreads
 *            {@code >= maxBackgroundFlushes, maxBackgroundCompactions}
 * @param maxLogFileSize
 *            if equal to {@code 0}, write all logs to one file and roll by time
 * @return
 */
public static DBOptions buildDbOptions(int maxBackgroundFlushes, int maxBackgroundCompactions,
        int maxBackgroundThreads, long maxLogFileSize) {
    DBOptions dbOptions = new DBOptions();
    dbOptions.setCreateIfMissing(true).setCreateMissingColumnFamilies(true)
            .setErrorIfExists(false);
    dbOptions.setMaxBackgroundFlushes(maxBackgroundFlushes)
            .setMaxBackgroundCompactions(maxBackgroundCompactions)
            .setIncreaseParallelism(maxBackgroundThreads);
    dbOptions.setAllowMmapReads(true).setAllowMmapWrites(true);
    dbOptions.setMaxOpenFiles(-1);
    dbOptions.setKeepLogFileNum(100).setLogFileTimeToRoll(3600)
            .setMaxLogFileSize(maxLogFileSize);
    return dbOptions;
}
 

import org.rocksdb.DBOptions; //导入依赖的package包/类
public RocksDbWrapper setDbOptions(DBOptions dbOptions) {
    if (this.dbOptions != null && myOwnDbOptions) {
        RocksDbUtils.closeRocksObjects(this.dbOptions);
    }
    this.dbOptions = dbOptions;
    myOwnDbOptions = false;
    return this;
}
 

import org.rocksdb.DBOptions; //导入依赖的package包/类
protected void initRocksDb() {
    RocksDbOptionsFactory optionFactory = new RocksDbOptionsFactory.Defaults();
    Options options = optionFactory.createOptions(null);
    DBOptions dbOptions = optionFactory.createDbOptions(null);
    ColumnFamilyOptions cfOptions = optionFactory.createColumnFamilyOptions(null);
    String optionsFactoryClass = (String) conf.get(ConfigExtension.ROCKSDB_OPTIONS_FACTORY_CLASS);
    if (optionsFactoryClass != null) {
        RocksDbOptionsFactory udfOptionFactory = (RocksDbOptionsFactory) Utils.newInstance(optionsFactoryClass);
        options = udfOptionFactory.createOptions(options);
        dbOptions = udfOptionFactory.createDbOptions(dbOptions);
        cfOptions = udfOptionFactory.createColumnFamilyOptions(cfOptions);
    }

    try {
        ttlTimeSec = ConfigExtension.getStateTtlTime(conf);
        if (ttlTimeSec > 0)
            rocksDb = TtlDB.open(options, rocksDbDir, ttlTimeSec, false);
        else
            rocksDb = RocksDB.open(options, rocksDbDir);
        // enable compaction
        rocksDb.compactRange();
        LOG.info("Finish the initialization of RocksDB");
    } catch (RocksDBException e) {
        LOG.error("Failed to open rocksdb located at " + rocksDbDir, e);
        throw new RuntimeException(e.getMessage());
    }

    lastCheckpointFiles = new HashSet<String>();
    lastCleanTime = System.currentTimeMillis();
    lastSuccessBatchId = -1;
}
 

import org.rocksdb.DBOptions; //导入依赖的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.DBOptions; //导入依赖的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.DBOptions; //导入依赖的package包/类
public RocksDBKeyedStateBackend(
	String operatorIdentifier,
	ClassLoader userCodeClassLoader,
	File instanceBasePath,
	DBOptions dbOptions,
	ColumnFamilyOptions columnFamilyOptions,
	TaskKvStateRegistry kvStateRegistry,
	TypeSerializer<K> keySerializer,
	int numberOfKeyGroups,
	KeyGroupRange keyGroupRange,
	ExecutionConfig executionConfig,
	boolean enableIncrementalCheckpointing
) throws IOException {

	super(kvStateRegistry, keySerializer, userCodeClassLoader, numberOfKeyGroups, keyGroupRange, executionConfig);

	this.operatorIdentifier = Preconditions.checkNotNull(operatorIdentifier);

	this.enableIncrementalCheckpointing = enableIncrementalCheckpointing;
	this.rocksDBResourceGuard = new ResourceGuard();

	// ensure that we use the right merge operator, because other code relies on this
	this.columnOptions = Preconditions.checkNotNull(columnFamilyOptions)
		.setMergeOperatorName(MERGE_OPERATOR_NAME);

	this.dbOptions = Preconditions.checkNotNull(dbOptions);

	this.instanceBasePath = Preconditions.checkNotNull(instanceBasePath);
	this.instanceRocksDBPath = new File(instanceBasePath, "db");

	if (instanceBasePath.exists()) {
		// Clear the base directory when the backend is created
		// in case something crashed and the backend never reached dispose()
		cleanInstanceBasePath();
	}

	if (!instanceBasePath.mkdirs()) {
		throw new IOException("Could not create RocksDB data directory.");
	}

	this.keyGroupPrefixBytes = getNumberOfKeyGroups() > (Byte.MAX_VALUE + 1) ? 2 : 1;
	this.kvStateInformation = new HashMap<>();
	this.restoredKvStateMetaInfos = new HashMap<>();
	this.materializedSstFiles = new TreeMap<>();
	this.backendUID = UUID.randomUUID();
	LOG.debug("Setting initial keyed backend uid for operator {} to {}.", this.operatorIdentifier, this.backendUID);
}