Package com.linkedin.davinci.store.rocksdb

Class RocksDBStoragePartition

java.lang.Object

com.linkedin.davinci.store.AbstractStoragePartition

com.linkedin.davinci.store.rocksdb.RocksDBStoragePartition

Direct Known Subclasses:

ReplicationMetadataRocksDBStoragePartition

@NotThreadSafe
public class RocksDBStoragePartition
extends AbstractStoragePartition

In RocksDBStoragePartition, it assumes the update(insert/delete) will happen sequentially. If the batch push is bytewise-sorted by key, this class is leveraging SstFileWriter to generate the SST file directly and ingest all the generated SST files into the RocksDB database at the end of the push. If the ingestion is unsorted, this class is using the regular RocksDB interface to support update operations.

Field Summary

Fields

Modifier and Type	Field	Description
protected final boolean	blobTransferEnabled
protected final List<org.rocksdb.ColumnFamilyDescriptor>	columnFamilyDescriptors
protected final List<org.rocksdb.ColumnFamilyHandle>	columnFamilyHandleList	Column Family is the concept in RocksDB to create isolation between different value for the same key.
protected final boolean	deferredWrite	Whether the input is sorted or not.
protected final int	partitionId
protected static final org.rocksdb.ReadOptions	READ_OPTIONS_DEFAULT
protected final ReentrantReadWriteLock	readCloseRWLock	Since all the modification functions are synchronized, we don't need any other synchronization for the update path to guard RocksDB closing behavior.
protected final boolean	readOnly	Whether the database is read only or not.
protected final boolean	readWriteLeaderForDefaultCF
protected final boolean	readWriteLeaderForRMDCF
protected final String	replicaId
protected org.rocksdb.RocksDB	rocksDB
protected final String	storeName
protected final String	storeNameAndVersion
protected final boolean	writeOnly
protected final org.rocksdb.WriteOptions	writeOptions	Here RocksDB disables WAL, but relies on the 'flush', which will be invoked through sync() to avoid data loss during recovery.

Constructor Summary

Constructors

Modifier	Constructor	Description
	RocksDBStoragePartition(StoragePartitionConfig storagePartitionConfig, RocksDBStorageEngineFactory factory, String dbDir, RocksDBMemoryStats rocksDBMemoryStats, RocksDBThrottler rocksDbThrottler, RocksDBServerConfig rocksDBServerConfig, VeniceStoreVersionConfig storeConfig)
protected	RocksDBStoragePartition(StoragePartitionConfig storagePartitionConfig, RocksDBStorageEngineFactory factory, String dbDir, RocksDBMemoryStats rocksDBMemoryStats, RocksDBThrottler rocksDbThrottler, RocksDBServerConfig rocksDBServerConfig, List<byte[]> columnFamilyNameList, VeniceStoreVersionConfig storeConfig)

Method Summary

Modifier and Type	Method	Description
void	beginBatchWrite(Map<String,String> checkpointedInfo, Optional<Supplier<byte[]>> expectedChecksumSupplier)
boolean	checkDatabaseIntegrity(Map<String,String> checkpointedInfo)	checks whether the current state of the database is valid during the start of ingestion.
void	close()	Close the specific partition
void	createSnapshot()	Creates a snapshot of the current state of the storage if the blob transfer feature is enabled via the store configuration
void	delete(byte[] key)	Delete a key from the partition database
void	deleteFilesInDirectory(String fullPath)
void	drop()	Drop when it is not required anymore.
void	endBatchWrite()
byte[]	get(byte[] key)	Get a value from the partition database
ByteBuffer	get(byte[] key, ByteBuffer valueToBePopulated)
byte[]	get(ByteBuffer keyBuffer)
<K, V> V	get(K key)	Get a Value from the partition database
Map<org.rocksdb.MemoryUsageType,Long>	getApproximateMemoryUsageByType(Set<org.rocksdb.Cache> caches)
protected Boolean	getBlobTransferEnabled()
void	getByKeyPrefix(byte[] keyPrefix, BytesStreamingCallback callback)	Populate provided callback with key-value pairs from the partition database where the keys have provided prefix.
protected List<org.rocksdb.ColumnFamilyHandle>	getColumnFamilyHandleList()
protected org.rocksdb.EnvOptions	getEnvOptions()
String	getFullPathForTempSSTFileDir()
AbstractStorageIterator	getIterator()
protected org.rocksdb.Options	getOptions()
long	getPartitionSizeInBytes()	Get the partition database size in bytes
long	getRmdByteUsage()
RocksDBSstFileWriter	getRocksDBSstFileWriter()
long	getRocksDBStatValue(String statName)
protected org.rocksdb.Options	getStoreOptions(StoragePartitionConfig storagePartitionConfig, boolean isRMD)
protected void	makeSureRocksDBIsStillOpen()
List<byte[]>	multiGet(List<byte[]> keys)
List<ByteBuffer>	multiGet(List<ByteBuffer> keys, List<ByteBuffer> values)
void	put(byte[] key, byte[] value)	Puts a value into the partition database
void	put(byte[] key, ByteBuffer valueBuffer)
<K, V> void	put(K key, V value)
void	reopen()	Reopen the underlying RocksDB database, and this operation will unload the data cached in memory.
Map<String,String>	sync()	Sync current database.
boolean	validateBatchIngestion()
boolean	verifyConfig(StoragePartitionConfig partitionConfig)	Check AbstractStoragePartition.verifyConfig(StoragePartitionConfig).

Methods inherited from class com.linkedin.davinci.store.AbstractStoragePartition

deleteWithReplicationMetadata, getPartitionId, getReplicationMetadata, putReplicationMetadata, putWithReplicationMetadata, putWithReplicationMetadata

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

READ_OPTIONS_DEFAULT

protected static final org.rocksdb.ReadOptions READ_OPTIONS_DEFAULT

writeOptions

protected final org.rocksdb.WriteOptions writeOptions

Here RocksDB disables WAL, but relies on the 'flush', which will be invoked through sync() to avoid data loss during recovery.

replicaId

protected final String replicaId

storeName

protected final String storeName

storeNameAndVersion

protected final String storeNameAndVersion

blobTransferEnabled

protected final boolean blobTransferEnabled

partitionId

protected final int partitionId

readCloseRWLock

protected final ReentrantReadWriteLock readCloseRWLock

Since all the modification functions are synchronized, we don't need any other synchronization for the update path to guard RocksDB closing behavior. The following readCloseRWLock is only used to guard get(byte[]) since we don't want to synchronize get requests.

rocksDB

protected org.rocksdb.RocksDB rocksDB

deferredWrite

protected final boolean deferredWrite

Whether the input is sorted or not.
deferredWrite = sortedInput => ingested via batch push which is sorted in VPJ, can use RocksDBSstFileWriter to ingest the input data to RocksDB
!deferredWrite = !sortedInput => can not use RocksDBSstFileWriter for ingestion

readOnly

protected final boolean readOnly

Whether the database is read only or not.

writeOnly

protected final boolean writeOnly

readWriteLeaderForDefaultCF

protected final boolean readWriteLeaderForDefaultCF

readWriteLeaderForRMDCF

protected final boolean readWriteLeaderForRMDCF

columnFamilyHandleList

protected final List<org.rocksdb.ColumnFamilyHandle> columnFamilyHandleList

Column Family is the concept in RocksDB to create isolation between different value for the same key. All KVs are stored in `DEFAULT` column family, if no column family is specified. If we stores replication metadata in the RocksDB, we stored it in a separated column family. We will insert all the column family descriptors into columnFamilyDescriptors and pass it to RocksDB when opening the store, and it will fill the columnFamilyHandles with handles which will be used when we want to put/get/delete from different RocksDB column families.

columnFamilyDescriptors

protected final List<org.rocksdb.ColumnFamilyDescriptor> columnFamilyDescriptors

Constructor Details

RocksDBStoragePartition

protected RocksDBStoragePartition(StoragePartitionConfig storagePartitionConfig,
 RocksDBStorageEngineFactory factory,
 String dbDir,
 RocksDBMemoryStats rocksDBMemoryStats,
 RocksDBThrottler rocksDbThrottler,
 RocksDBServerConfig rocksDBServerConfig,
 List<byte[]> columnFamilyNameList,
 VeniceStoreVersionConfig storeConfig)

RocksDBStoragePartition

public RocksDBStoragePartition(StoragePartitionConfig storagePartitionConfig,
 RocksDBStorageEngineFactory factory,
 String dbDir,
 RocksDBMemoryStats rocksDBMemoryStats,
 RocksDBThrottler rocksDbThrottler,
 RocksDBServerConfig rocksDBServerConfig,
 VeniceStoreVersionConfig storeConfig)

Method Details

makeSureRocksDBIsStillOpen

protected void makeSureRocksDBIsStillOpen()

getEnvOptions

protected org.rocksdb.EnvOptions getEnvOptions()

getBlobTransferEnabled

protected Boolean getBlobTransferEnabled()

getStoreOptions

protected org.rocksdb.Options getStoreOptions(StoragePartitionConfig storagePartitionConfig,
 boolean isRMD)

getColumnFamilyHandleList

protected List<org.rocksdb.ColumnFamilyHandle> getColumnFamilyHandleList()

getRmdByteUsage

public long getRmdByteUsage()

Overrides:

getRmdByteUsage in class AbstractStoragePartition

checkDatabaseIntegrity

public boolean checkDatabaseIntegrity(Map<String,String> checkpointedInfo)

Description copied from class: AbstractStoragePartition

checks whether the current state of the database is valid during the start of ingestion.

Overrides:

checkDatabaseIntegrity in class AbstractStoragePartition

beginBatchWrite

public void beginBatchWrite(Map<String,String> checkpointedInfo,
 Optional<Supplier<byte[]>> expectedChecksumSupplier)

Overrides:

beginBatchWrite in class AbstractStoragePartition

endBatchWrite

public void endBatchWrite()

Overrides:

endBatchWrite in class AbstractStoragePartition

createSnapshot

public void createSnapshot()

Description copied from class: AbstractStoragePartition

Creates a snapshot of the current state of the storage if the blob transfer feature is enabled via the store configuration

Specified by:

createSnapshot in class AbstractStoragePartition

put

public void put(byte[] key,
 byte[] value)

Description copied from class: AbstractStoragePartition

Puts a value into the partition database

Specified by:

put in class AbstractStoragePartition

put

public void put(byte[] key,
 ByteBuffer valueBuffer)

Specified by:

put in class AbstractStoragePartition

put

public <K,
V> void put(K key,
 V value)

Specified by:

put in class AbstractStoragePartition

get

public byte[] get(byte[] key)

Description copied from class: AbstractStoragePartition

Get a value from the partition database

Specified by:

get in class AbstractStoragePartition

Parameters:

key - key to be retrieved

Returns:

null if the key does not exist, byte[] value if it exists.

get

public ByteBuffer get(byte[] key,
 ByteBuffer valueToBePopulated)

Overrides:

get in class AbstractStoragePartition

get

public <K,
V> V get(K key)

Description copied from class: AbstractStoragePartition

Get a Value from the partition database

Specified by:

get in class AbstractStoragePartition

Type Parameters:

K - the type for Key

V - the type for the return value

Parameters:

key - key to be retrieved

Returns:

null if the key does not exist, V value if it exists

get

public byte[] get(ByteBuffer keyBuffer)

Specified by:

get in class AbstractStoragePartition

multiGet

public List<byte[]> multiGet(List<byte[]> keys)

multiGet

public List<ByteBuffer> multiGet(List<ByteBuffer> keys,
 List<ByteBuffer> values)

getByKeyPrefix

public void getByKeyPrefix(byte[] keyPrefix,
 BytesStreamingCallback callback)

Description copied from class: AbstractStoragePartition

Populate provided callback with key-value pairs from the partition database where the keys have provided prefix. If prefix is null, callback will be populated will all key-value pairs from the partition database.

Specified by:

getByKeyPrefix in class AbstractStoragePartition

validateBatchIngestion

public boolean validateBatchIngestion()

Overrides:

validateBatchIngestion in class AbstractStoragePartition

delete

public void delete(byte[] key)

Description copied from class: AbstractStoragePartition

Delete a key from the partition database

Specified by:

delete in class AbstractStoragePartition

sync

public Map<String,String> sync()

Description copied from class: AbstractStoragePartition

Sync current database.

Specified by:

sync in class AbstractStoragePartition

Returns:

Database related info, which is required to be checkpointed.

deleteFilesInDirectory

public void deleteFilesInDirectory(String fullPath)

drop

public void drop()

Description copied from class: AbstractStoragePartition

Drop when it is not required anymore.

Specified by:

drop in class AbstractStoragePartition

close

public void close()

Description copied from class: AbstractStoragePartition

Close the specific partition

Specified by:

close in class AbstractStoragePartition

reopen

public void reopen()

Reopen the underlying RocksDB database, and this operation will unload the data cached in memory.

Overrides:

reopen in class AbstractStoragePartition

getRocksDBStatValue

public long getRocksDBStatValue(String statName)

getApproximateMemoryUsageByType

public Map<org.rocksdb.MemoryUsageType,Long> getApproximateMemoryUsageByType(Set<org.rocksdb.Cache> caches)

verifyConfig

public boolean verifyConfig(StoragePartitionConfig partitionConfig)

Check AbstractStoragePartition.verifyConfig(StoragePartitionConfig).

Specified by:

verifyConfig in class AbstractStoragePartition

Parameters:

partitionConfig -

Returns:

getPartitionSizeInBytes

public long getPartitionSizeInBytes()

Description copied from class: AbstractStoragePartition

Get the partition database size in bytes

Specified by:

getPartitionSizeInBytes in class AbstractStoragePartition

Returns:

partition database size

getOptions

protected org.rocksdb.Options getOptions()

getFullPathForTempSSTFileDir

public String getFullPathForTempSSTFileDir()

getRocksDBSstFileWriter

public RocksDBSstFileWriter getRocksDBSstFileWriter()

getIterator

public AbstractStorageIterator getIterator()

Overrides:

getIterator in class AbstractStoragePartition