Packages

class RocksDBFileManager extends Logging

Class responsible for syncing RocksDB checkpoint files from local disk to DFS. For each version, checkpoint is saved in specific directory structure that allows successive versions to reuse to SST data files and archived log files. This allows each commit to be incremental, only new SST files and archived log files generated by RocksDB will be uploaded. The directory structures on local disk and in DFS are as follows.

Local checkpoint dir structure ------------------------------ RocksDB generates a bunch of files in the local checkpoint directory. The most important among them are the SST files; they are the actual log structured data files. Rest of the files contain the metadata necessary for RocksDB to read the SST files and start from the checkpoint. Note that the SST files are hard links to files in the RocksDB's working directory, and therefore successive checkpoints can share some of the SST files. So these SST files have to be copied to DFS in shared directory such that different committed versions can save them.

We consider both SST files and archived log files as immutable files which can be shared between different checkpoints.

localCheckpointDir | +-- OPTIONS-000005 +-- MANIFEST-000008 +-- CURRENT +-- 00007.sst +-- 00011.sst +-- archive | +-- 00008.log | +-- 00013.log ...

DFS directory structure after saving to DFS as version 10 ----------------------------------------------------------- The SST and archived log files are given unique file names and copied to the shared subdirectory. Every version maintains a mapping of local immutable file name to the unique file name in DFS. This mapping is saved in a JSON file (named metadata), which is zipped along with other checkpoint files into a single file [version].zip.

dfsRootDir | +-- SSTs | +-- 00007-[uuid1].sst | +-- 00011-[uuid2].sst +-- logs | +-- 00008-[uuid3].log | +-- 00013-[uuid4].log +-- 10.zip | +-- metadata <--- contains mapping between 00007.sst and [uuid1].sst, and the mapping between 00008.log and [uuid3].log | +-- OPTIONS-000005 | +-- MANIFEST-000008 | +-- CURRENT | ... | +-- 9.zip +-- 8.zip ...

Note the following. - Each [version].zip is a complete description of all the data and metadata needed to recover a RocksDB instance at the corresponding version. The SST files and log files are not included in the zip files, they can be shared cross different versions. This is unlike the [version].delta files of HDFSBackedStateStore where previous delta files needs to be read to be recovered. - This is safe wrt speculatively executed tasks running concurrently in different executors as each task would upload a different copy of the generated immutable files and atomically update the [version].zip. - Immutable files are identified uniquely based on their file name and file size. - Immutable files can be reused only across adjacent checkpoints/versions. - This class is thread-safe. Specifically, it is safe to concurrently delete old files from a different thread than the task thread saving files.

Linear Supertypes
Logging, AnyRef, Any
Ordering
  1. Alphabetic
  2. By Inheritance
Inherited
  1. RocksDBFileManager
  2. Logging
  3. AnyRef
  4. Any
  1. Hide All
  2. Show All
Visibility
  1. Public
  2. Protected

Instance Constructors

  1. new RocksDBFileManager(dfsRootDir: String, localTempDir: File, hadoopConf: Configuration, codecName: String = CompressionCodec.ZSTD, loggingId: String = "", storeConf: StateStoreConf = StateStoreConf.empty, fileChecksumEnabled: Boolean = false, fileChecksumThreadPoolSize: Option[Int] = None)

    dfsRootDir

    Directory where the [version].zip files will be stored

    localTempDir

    Local directory for temporary work

    hadoopConf

    Hadoop configuration for talking to DFS

    loggingId

    Id that will be prepended in logs for isolating concurrent RocksDBs

    fileChecksumEnabled

    Whether file checksum generation and verification is enabled

    fileChecksumThreadPoolSize

    Number of threads used to concurrently operate on the main and checksum files

Type Members

  1. implicit class LogStringContext extends AnyRef
    Definition Classes
    Logging

Value Members

  1. final def !=(arg0: Any): Boolean
    Definition Classes
    AnyRef → Any
  2. final def ##: Int
    Definition Classes
    AnyRef → Any
  3. final def ==(arg0: Any): Boolean
    Definition Classes
    AnyRef → Any
  4. def MDC(key: LogKey, value: Any): MDC
    Attributes
    protected
    Definition Classes
    Logging
  5. final def asInstanceOf[T0]: T0
    Definition Classes
    Any
  6. def clone(): AnyRef
    Attributes
    protected[lang]
    Definition Classes
    AnyRef
    Annotations
    @throws(classOf[java.lang.CloneNotSupportedException]) @IntrinsicCandidate() @native()
  7. def close(): Unit

    Close this file manager and release underlying resources.

  8. def deleteOldVersions(numVersionsToRetain: Int, maxVersionsToDeletePerMaintenance: Int = -1, minVersionsToDelete: Long = 0): Unit

    Delete old versions by deleting the associated version and SST files.

    Delete old versions by deleting the associated version and SST files. At a high-level, when enough stale version files are present for batch deletion, this method finds which versions to delete, and which SST files that were last used in those versions. It's safe to delete these SST files because a SST file can be reused only in successive versions. Therefore, if a SST file F was last used in version V, then it won't be used in version V+1 or later, and if version V can be deleted, then F can safely be deleted as well.

    First, it checks whether enough stale version files are present for batch deletion. If true, it does the following to find old files. - List all the existing [version].zip files - Find the min version that needs to be retained based on the given numVersionsToRetain. - Accordingly decide which versions should be deleted. - Resolve all SSTs files of all the existing versions, if not already resolved. - Find the files that were last used in the to-be-deleted versions as we will not need those files any more. - Find the orphan sst and log files whose zip files are not uploaded successfully or have been overwritten. To avoid deleting files of ongoing tasks, only delete orphan files that are older than all tracked files when there are at least 2 versions. - Delete sst and log files in to-be-deleted versions. - Delete orphan files. - Delete changelog files of to-be-deleted versions.

    Note that it only deletes files that it knows are safe to delete. It may not delete the following files. - Partially written SST files - SST files that were used in a version, but that version got overwritten with a different set of SST files.

    numVersionsToRetain

    the number of RocksDB versions to keep in object store after the deletion. Must be greater than 0, or -1 to retain all versions.

    maxVersionsToDeletePerMaintenance

    the max number of RocksDB versions to delete per maintenance operation. Must be greater than 0, or -1 to delete all stale versions.

    minVersionsToDelete

    the min number of stale versions required to trigger deletion. If its set to <= 0, then we will always perform list operations to determine deletion candidates. If set to a positive value, then we will skip deletion if the number of stale versions is less than this value.

  9. def dfsFileSuffix(immutableFile: RocksDBImmutableFile): String
  10. final def eq(arg0: AnyRef): Boolean
    Definition Classes
    AnyRef
  11. def equals(arg0: AnyRef): Boolean
    Definition Classes
    AnyRef → Any
  12. def existsSnapshotFile(version: Long, checkpointUniqueId: Option[String] = None): Boolean
  13. def findOrphanFiles(trackedFiles: Seq[String], allFiles: Seq[FileStatus]): Seq[String]

    Find orphan files which are not tracked by zip files.

    Find orphan files which are not tracked by zip files. Both sst files and log files can be orphan files. They are uploaded separately before the zip file of that version is uploaded. When the zip file of a version get overwritten, the referenced sst and log files become orphan. Be careful here since sst and log files of the ongoing version also appear to be orphan before their zip file is uploaded.

    trackedFiles

    files tracked by metadata in versioned zip file

    allFiles

    all sst or log files in the directory.

    returns

    filenames of orphan files

  14. def getChangeLogWriter(version: Long, useColumnFamilies: Boolean = false, checkpointUniqueId: Option[String] = None, stateStoreCheckpointIdLineage: Option[Array[LineageItem]] = None): StateStoreChangelogWriter
  15. def getChangelogReader(version: Long, checkpointUniqueId: Option[String] = None): StateStoreChangelogReader
  16. final def getClass(): Class[_ <: AnyRef]
    Definition Classes
    AnyRef → Any
    Annotations
    @IntrinsicCandidate() @native()
  17. def getFileSystem(myDfsRootDir: String, myHadoopConf: Configuration): FileSystem
    Attributes
    protected
  18. def getLatestSnapshotVersion(version: Long): Long
  19. def getLatestSnapshotVersionAndUniqueIdFromLineage(lineage: Array[LineageItem]): Option[(Long, String)]

    Based on the ground truth lineage loaded from changelog file (lineage), this function does file listing to find all snapshot (version, uniqueId) pairs, and finds the ground truth latest snapshot (version, uniqueId) the db instance needs to load.

    Based on the ground truth lineage loaded from changelog file (lineage), this function does file listing to find all snapshot (version, uniqueId) pairs, and finds the ground truth latest snapshot (version, uniqueId) the db instance needs to load.

    lineage

    The ground truth lineage loaded from changelog file, sorted by id

    returns

    The ground truth latest snapshot (version, uniqueId) the db instance needs to load, when the return value is None it means ther is no such snapshot found.

  20. def getLatestVersion(): Long

    Get the latest version available in the DFS directory.

    Get the latest version available in the DFS directory. If no data present, it returns 0.

  21. def hashCode(): Int
    Definition Classes
    AnyRef → Any
    Annotations
    @IntrinsicCandidate() @native()
  22. def initializeLogIfNecessary(isInterpreter: Boolean, silent: Boolean): Boolean
    Attributes
    protected
    Definition Classes
    Logging
  23. def initializeLogIfNecessary(isInterpreter: Boolean): Unit
    Attributes
    protected
    Definition Classes
    Logging
  24. final def isInstanceOf[T0]: Boolean
    Definition Classes
    Any
  25. def isTraceEnabled(): Boolean
    Attributes
    protected
    Definition Classes
    Logging
  26. def latestLoadCheckpointMetrics: RocksDBFileManagerMetrics
  27. def latestSaveCheckpointMetrics: RocksDBFileManagerMetrics
  28. def listRocksDBFiles(localDir: File): (Seq[File], Seq[File])

    List all the RocksDB files that need be synced or recovered.

  29. def loadCheckpointFromDfs(version: Long, localDir: File, rocksDBFileMapping: RocksDBFileMapping, checkpointUniqueId: Option[String] = None): RocksDBCheckpointMetadata

    Load all necessary files for specific checkpoint version from DFS to given local directory.

    Load all necessary files for specific checkpoint version from DFS to given local directory. If version is 0, then it will delete all files in the directory. For other versions, it ensures that only the exact files generated during checkpointing will be present in the local directory.

  30. def log: Logger
    Attributes
    protected
    Definition Classes
    Logging
  31. def logBasedOnLevel(level: Level)(f: => MessageWithContext): Unit
    Attributes
    protected
    Definition Classes
    Logging
  32. def logDebug(msg: => String, throwable: Throwable): Unit
    Attributes
    protected
    Definition Classes
    Logging
  33. def logDebug(entry: LogEntry, throwable: Throwable): Unit
    Attributes
    protected
    Definition Classes
    Logging
  34. def logDebug(entry: LogEntry): Unit
    Attributes
    protected
    Definition Classes
    Logging
  35. def logDebug(msg: => String): Unit
    Attributes
    protected
    Definition Classes
    Logging
  36. def logError(msg: => String, throwable: Throwable): Unit
    Attributes
    protected
    Definition Classes
    Logging
  37. def logError(entry: LogEntry, throwable: Throwable): Unit
    Attributes
    protected
    Definition Classes
    Logging
  38. def logError(entry: LogEntry): Unit
    Attributes
    protected
    Definition Classes
    Logging
  39. def logError(msg: => String): Unit
    Attributes
    protected
    Definition Classes
    Logging
  40. def logInfo(msg: => String, throwable: Throwable): Unit
    Attributes
    protected
    Definition Classes
    Logging
  41. def logInfo(entry: LogEntry, throwable: Throwable): Unit
    Attributes
    protected
    Definition Classes
    Logging
  42. def logInfo(entry: LogEntry): Unit
    Attributes
    protected
    Definition Classes
    Logging
  43. def logInfo(msg: => String): Unit
    Attributes
    protected
    Definition Classes
    Logging
  44. def logName: String
    Attributes
    protected
    Definition Classes
    RocksDBFileManager → Logging
  45. def logTrace(msg: => String, throwable: Throwable): Unit
    Attributes
    protected
    Definition Classes
    Logging
  46. def logTrace(entry: LogEntry, throwable: Throwable): Unit
    Attributes
    protected
    Definition Classes
    Logging
  47. def logTrace(entry: LogEntry): Unit
    Attributes
    protected
    Definition Classes
    Logging
  48. def logTrace(msg: => String): Unit
    Attributes
    protected
    Definition Classes
    Logging
  49. def logWarning(msg: => String, throwable: Throwable): Unit
    Attributes
    protected
    Definition Classes
    Logging
  50. def logWarning(entry: LogEntry, throwable: Throwable): Unit
    Attributes
    protected
    Definition Classes
    Logging
  51. def logWarning(entry: LogEntry): Unit
    Attributes
    protected
    Definition Classes
    Logging
  52. def logWarning(msg: => String): Unit
    Attributes
    protected
    Definition Classes
    Logging
  53. final def ne(arg0: AnyRef): Boolean
    Definition Classes
    AnyRef
  54. def newDFSFileName(localFileName: String, dfsFileSuffix: String): String
  55. final def notify(): Unit
    Definition Classes
    AnyRef
    Annotations
    @IntrinsicCandidate() @native()
  56. final def notifyAll(): Unit
    Definition Classes
    AnyRef
    Annotations
    @IntrinsicCandidate() @native()
  57. def saveCheckpointToDfs(checkpointDir: File, version: Long, numKeys: Long, numInternalKeys: Long, fileMapping: Map[String, RocksDBSnapshotFile], columnFamilyMapping: Option[Map[String, ColumnFamilyInfo]] = None, maxColumnFamilyId: Option[Short] = None, checkpointUniqueId: Option[String] = None, verifyNonEmptyFilesInZip: Boolean = false): Unit

    Save all the files in given local checkpoint directory as a committed version in DFS

  58. def setMaxSeenVersion(version: Long): Unit

    Set maxSeenVersion to max of itself and version we are uploading.

    Set maxSeenVersion to max of itself and version we are uploading. This is to ensure accuracy in the case the query has restarted from a particular version.

  59. final def synchronized[T0](arg0: => T0): T0
    Definition Classes
    AnyRef
  60. def toString(): String
    Definition Classes
    AnyRef → Any
  61. final def wait(arg0: Long, arg1: Int): Unit
    Definition Classes
    AnyRef
    Annotations
    @throws(classOf[java.lang.InterruptedException])
  62. final def wait(arg0: Long): Unit
    Definition Classes
    AnyRef
    Annotations
    @throws(classOf[java.lang.InterruptedException]) @native()
  63. final def wait(): Unit
    Definition Classes
    AnyRef
    Annotations
    @throws(classOf[java.lang.InterruptedException])
  64. def withLogContext(context: Map[String, String])(body: => Unit): Unit
    Attributes
    protected
    Definition Classes
    Logging

Deprecated Value Members

  1. def finalize(): Unit
    Attributes
    protected[lang]
    Definition Classes
    AnyRef
    Annotations
    @throws(classOf[java.lang.Throwable]) @Deprecated
    Deprecated

    (Since version 9)

Inherited from Logging

Inherited from AnyRef

Inherited from Any

Ungrouped