[Delta-Standalone] Compute Snapshot protocol and metadata more efficiently (#533)

Author: scottsand-db

standalone/src/main/scala/io/delta/standalone/internal/SnapshotImpl.scala

@@ -17,7 +17,6 @@
 package io.delta.standalone.internal
 
 import java.net.URI
-import java.util.concurrent.{Executors, ExecutorService}
 
 import scala.collection.JavaConverters._
 import scala.collection.parallel.ExecutionContextTaskSupport
@@ -40,12 +39,30 @@ import io.delta.standalone.internal.logging.Logging
 import io.delta.standalone.internal.scan.{DeltaScanImpl, FilteredDeltaScanImpl}
 import io.delta.standalone.internal.util.{ConversionUtils, FileNames, JsonUtils}
 
+/**
+ * Contains the protocol, metadata, and corresponding table version. The protocol and metadata
+ * will be used as the defaults in the Snapshot if no newer values are found in logs > `version`.
+ */
+case class SnapshotProtocolMetadataHint(protocol: Protocol, metadata: Metadata, version: Long)
+
+/**
+ * Visible for testing.
+ *
+ * Will contain various metrics collected while finding/loading the latest protocol and metadata
+ * for this Snapshot. This can be used to verify that the minimal log replay occurred.
+ */
+case class ProtocolMetadataLoadMetrics(fileVersions: Seq[Long])
+
 /**
  * Scala implementation of Java interface [[Snapshot]].
  *
  * @param timestamp The timestamp of the latest commit in milliseconds. Can also be set to -1 if the
  * timestamp of the commit is unknown or the table has not been initialized, i.e.
  * `version = -1`.
+ * @param protocolMetadataHint The optional protocol, metadata, and table version that can be used
+ * to speed up loading *this* Snapshot's protocol and metadata (P&M).
+ * Essentially, when computing *this* Snapshot's P&M, we only need to
+ * look at the log files *newer* than the hint version.
  */
 private[internal] class SnapshotImpl(
  val hadoopConf: Configuration,
@@ -54,7 +71,14 @@ private[internal] class SnapshotImpl(
  val logSegment: LogSegment,
  val minFileRetentionTimestamp: Long,
  val deltaLog: DeltaLogImpl,
- val timestamp: Long) extends Snapshot with Logging {
+ val timestamp: Long,
+ protocolMetadataHint: Option[SnapshotProtocolMetadataHint] = Option.empty)
+ extends Snapshot with Logging {
+
+ protocolMetadataHint.foreach { hint =>
+ require(hint.version <= version, s"Cannot use a protocolMetadataHint with a version newer " +
+ s"than that of this Snapshot. Hint version: ${hint.version}, Snapshot version: $version")
+ }
 
  import SnapshotImpl._
 
@@ -124,36 +148,85 @@ private[internal] class SnapshotImpl(
  lazy val transactions: Map[String, Long] =
  setTransactionsScala.map(t => t.appId -> t.version).toMap
 
- // These values need to be declared lazy. In Scala, strict values (i.e. non-lazy) in superclasses
- // (e.g. SnapshotImpl) are fully initialized before subclasses (e.g. InitialSnapshotImpl).
- // If these were 'strict', or 'eager', vals, then `loadTableProtocolAndMetadata` would be called
- // for all new InitialSnapshotImpl instances, causing an exception.
- lazy val (protocolScala, metadataScala) = loadTableProtocolAndMetadata()
+ /**
+ * protocolScala, metadataScala are internals APIs.
+ * protocolMetadataLoadMetrics is visible for testing only.
+ *
+ * NOTE: These values need to be declared lazy. In Scala, strict values (i.e. non-lazy) in
+ * superclasses (e.g. SnapshotImpl) are fully initialized before subclasses
+ * (e.g. InitialSnapshotImpl). If these were 'strict', or 'eager', vals, then
+ * `loadTableProtocolAndMetadata` would be called for all new InitialSnapshotImpl instances,
+ * causing an exception.
+ */
+ lazy val (protocolScala, metadataScala, protocolMetadataLoadMetrics) =
+ loadTableProtocolAndMetadata()
+
+ private def loadTableProtocolAndMetadata(): (Protocol, Metadata, ProtocolMetadataLoadMetrics) = {
+ val fileVersionsScanned = scala.collection.mutable.Set[Long]()
+ def createMetrics = ProtocolMetadataLoadMetrics(fileVersionsScanned.toSeq.sorted.reverse)
 
- private def loadTableProtocolAndMetadata(): (Protocol, Metadata) = {
  var protocol: Protocol = null
  var metadata: Metadata = null
+
  val iter = memoryOptimizedLogReplay.getReverseIterator
 
  try {
  // We replay logs from newest to oldest and will stop when we find the latest Protocol and
- // Metadata.
- iter.asScala.foreach { case (action, _) =>
+ // Metadata (P&M).
+ //
+ // If the protocolMetadataHint is defined, then we will only look at log files strictly newer
+ // (>) than the protocolMetadataHint's version. If we don't find any new P&M, then we will
+ // default to those from the protocolMetadataHint.
+ //
+ // If the protocolMetadataHint is not defined, then we must look at all log files. If no
+ // P&M is found, then we fail.
+ iter.asScala.foreach { case (action, _, actionTableVersion) =>
+
+ // We have not yet found the latest P&M. If we had found BOTH, we would have returned
+ // already. Note that we may have already found ONE of them.
+ protocolMetadataHint.foreach { hint =>
+ if (actionTableVersion == hint.version) {
+ // Furthermore, we have already looked at all the actions in all the log files strictly
+ // newer (>) than the hint version. Thus, we can short circuit early and use the P&M
+ // from the hint.
+
+ val newestProtocol = if (protocol == null) {
+ logInfo(s"Using the protocol from the protocolMetadataHint: ${hint.protocol}")
+ hint.protocol
+ } else {
+ logInfo(s"Found a newer protocol: $protocol")
+ protocol
+ }
+
+ val newestMetadata = if (metadata == null) {
+ logInfo(s"Using the metadata from the protocolMetadataHint: ${hint.metadata}")
+ hint.metadata
+ } else {
+ logInfo(s"Found a newer metadata: $metadata")
+ metadata
+ }
+
+ return (newestProtocol, newestMetadata, createMetrics)
+ }
+ }
+
+ fileVersionsScanned += actionTableVersion
+
  action match {
  case p: Protocol if null == protocol =>
  // We only need the latest protocol
  protocol = p
 
  if (protocol != null && metadata != null) {
  // Stop since we have found the latest Protocol and metadata.
- return (protocol, metadata)
+ return (protocol, metadata, createMetrics)
  }
  case m: Metadata if null == metadata =>
  metadata = m
 
  if (protocol != null && metadata != null) {
  // Stop since we have found the latest Protocol and metadata.
- return (protocol, metadata)
+ return (protocol, metadata, createMetrics)
  }
  case _ => // do nothing
  }
@@ -330,6 +403,9 @@ private class InitialSnapshotImpl(
 
  override lazy val metadataScala: Metadata = Metadata()
 
+ override lazy val protocolMetadataLoadMetrics: ProtocolMetadataLoadMetrics =
+ ProtocolMetadataLoadMetrics(Seq.empty)
+
  override def scan(): DeltaScan = new DeltaScanImpl(memoryOptimizedLogReplay)
 
  override def scan(predicate: Expression): DeltaScan =

standalone/src/main/scala/io/delta/standalone/internal/SnapshotManagement.scala

@@ -64,13 +64,18 @@ private[internal] trait SnapshotManagement { self: DeltaLogImpl =>
  */
  private def updateInternal(): SnapshotImpl = {
  try {
- val segment = getLogSegmentForVersion(currentSnapshot.logSegment.checkpointVersion)
- if (segment != currentSnapshot.logSegment) {
- val startingFrom = segment.checkpointVersion
+ val newSegment = getLogSegmentForVersion(
+ startCheckpoint = currentSnapshot.logSegment.checkpointVersion)
+ if (newSegment != currentSnapshot.logSegment) {
+ val startingFrom = newSegment.checkpointVersion
  .map(v => s" starting from checkpoint version $v.").getOrElse(".")
- logInfo(s"Loading version ${segment.version}$startingFrom")
+ logInfo(s"Loading version ${newSegment.version}$startingFrom")
 
- val newSnapshot = createSnapshot(segment, segment.lastCommitTimestamp)
+ val newSnapshot = createSnapshot(
+ newSegment,
+ newSegment.lastCommitTimestamp,
+ previousSnapshotOpt = Some(currentSnapshot) // We are updating to the newSegment!
+ )
 
  if (currentSnapshot.version > -1 &&
  currentSnapshot.metadataScala.id != newSnapshot.metadataScala.id) {
@@ -221,7 +226,11 @@ private[internal] trait SnapshotManagement { self: DeltaLogImpl =>
  .map(v => s" starting from checkpoint $v.").getOrElse(".")
  logInfo(s"Loading version ${logSegment.version}$startCheckpoint")
 
- val snapshot = createSnapshot(logSegment, logSegment.lastCommitTimestamp)
+ val snapshot = createSnapshot(
+ logSegment,
+ logSegment.lastCommitTimestamp,
+ previousSnapshotOpt = None // This is the `init`. There's no previous snapshot.
+ )
 
  logInfo(s"Returning initial snapshot $snapshot")
 
@@ -240,21 +249,46 @@ private[internal] trait SnapshotManagement { self: DeltaLogImpl =>
  val startingCheckpoint = findLastCompleteCheckpoint(CheckpointInstance(version, None))
  val segment = getLogSegmentForVersion(startingCheckpoint.map(_.version), Some(version))
 
+ // In practice, this will always be None because all callers of this method have already called
+ // deltaLog.update() (to determine the full list of versions, to understand the full history).
+ // Thus, `snapshot.version` will always be > version. (If they were equal, we would have already
+ // returned early above).
+ val previousSnapshotOpt =
+ if (currentSnapshot.version <= version) Some(currentSnapshot) else None
+
  createSnapshot(
  segment,
- segment.lastCommitTimestamp
+ segment.lastCommitTimestamp,
+ previousSnapshotOpt
  )
  }
 
- private def createSnapshot(segment: LogSegment, lastCommitTimestamp: Long): SnapshotImpl = {
+ private def createSnapshot(
+ segment: LogSegment,
+ lastCommitTimestamp: Long,
+ previousSnapshotOpt: Option[SnapshotImpl]): SnapshotImpl = {
+
+ previousSnapshotOpt.foreach { previousSnapshot =>
+ assert(
+ previousSnapshot.version <= segment.version,
+ s"Trying to create a Snapshot at version ${segment.version} yet you are passing a " +
+ s"newer `previousSnapshotOpt` with version ${previousSnapshot.version}."
+ )
+ }
+
  new SnapshotImpl(
  hadoopConf,
  logPath,
  segment.version,
  segment,
  minFileRetentionTimestamp,
  this,
- lastCommitTimestamp)
+ lastCommitTimestamp,
+ previousSnapshotOpt.map { previousSnapshot =>
+ SnapshotProtocolMetadataHint(
+ previousSnapshot.protocolScala, previousSnapshot.metadataScala, previousSnapshot.version)
+ }
+ )
  }
 
  private def verifyDeltaVersions(versions: Array[Long]): Unit = {

standalone/src/main/scala/io/delta/standalone/internal/actions/MemoryOptimizedLogReplay.scala

@@ -23,7 +23,7 @@ import io.delta.storage.{CloseableIterator, LogStore}
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
 
-import io.delta.standalone.internal.util.JsonUtils
+import io.delta.standalone.internal.util.{FileNames, JsonUtils}
 
 /**
  * Used to replay the transaction logs from the newest log file to the oldest log file, in a
@@ -36,28 +36,30 @@ private[internal] class MemoryOptimizedLogReplay(
  timeZone: TimeZone) {
 
  /**
- * @return a [[CloseableIterator]] of tuple (Action, isLoadedFromCheckpoint) in reverse
- * transaction log order
+ * @return a [[CloseableIterator]] of tuple (Action, isLoadedFromCheckpoint, tableVersion) in
+ * reverse transaction log order
  */
- def getReverseIterator: CloseableIterator[(Action, Boolean)] =
- new CloseableIterator[(Action, Boolean)] {
+ def getReverseIterator: CloseableIterator[(Action, Boolean, Long)] =
+ new CloseableIterator[(Action, Boolean, Long)] {
  private val reverseFilesIter: Iterator[Path] = files.sortWith(_.getName > _.getName).iterator
- private var actionIter: Option[CloseableIterator[(Action, Boolean)]] = None
+ private var actionIter: Option[CloseableIterator[(Action, Boolean, Long)]] = None
 
  /**
  * Requires that `reverseFilesIter.hasNext` is true
  */
- private def getNextIter: Option[CloseableIterator[(Action, Boolean)]] = {
+ private def getNextIter: Option[CloseableIterator[(Action, Boolean, Long)]] = {
  val nextFile = reverseFilesIter.next()
 
  if (nextFile.getName.endsWith(".json")) {
- Some(new CustomJsonIterator(logStore.read(nextFile, hadoopConf)))
+ val fileVersion = FileNames.deltaVersion(nextFile)
+ Some(new CustomJsonIterator(logStore.read(nextFile, hadoopConf), fileVersion))
  } else if (nextFile.getName.endsWith(".parquet")) {
+ val fileVersion = FileNames.checkpointVersion(nextFile)
  val parquetIterable = ParquetReader.read[Parquet4sSingleActionWrapper](
  nextFile.toString,
  ParquetReader.Options(timeZone, hadoopConf)
  )
- Some(new CustomParquetIterator(parquetIterable))
+ Some(new CustomParquetIterator(parquetIterable, fileVersion))
  } else {
  throw new IllegalStateException(s"unexpected log file path: $nextFile")
  }
@@ -94,7 +96,7 @@ private[internal] class MemoryOptimizedLogReplay(
  actionIter.isDefined
  }
 
- override def next(): (Action, Boolean) = {
+ override def next(): (Action, Boolean, Long) = {
  if (!hasNext()) throw new NoSuchElementException
 
  if (actionIter.isEmpty) throw new IllegalStateException("Impossible")
@@ -112,27 +114,29 @@ private[internal] class MemoryOptimizedLogReplay(
 // Helper Classes
 ///////////////////////////////////////////////////////////////////////////
 
-private class CustomJsonIterator(iter: CloseableIterator[String])
- extends CloseableIterator[(Action, Boolean)] {
+private class CustomJsonIterator(iter: CloseableIterator[String], version: Long)
+ extends CloseableIterator[(Action, Boolean, Long)] {
 
  override def hasNext: Boolean = iter.hasNext
 
- override def next(): (Action, Boolean) = {
- (JsonUtils.mapper.readValue[SingleAction](iter.next()).unwrap, false)
+ override def next(): (Action, Boolean, Long) = {
+ (JsonUtils.mapper.readValue[SingleAction](iter.next()).unwrap, false, version)
  }
 
  override def close(): Unit = iter.close()
 }
 
-private class CustomParquetIterator(iterable: ParquetIterable[Parquet4sSingleActionWrapper])
- extends CloseableIterator[(Action, Boolean)] {
+private class CustomParquetIterator(
+ iterable: ParquetIterable[Parquet4sSingleActionWrapper],
+ version: Long)
+ extends CloseableIterator[(Action, Boolean, Long)] {
 
  private val iter = iterable.iterator
 
  override def hasNext: Boolean = iter.hasNext
 
- override def next(): (Action, Boolean) = {
- (iter.next().unwrap.unwrap, true)
+ override def next(): (Action, Boolean, Long) = {
+ (iter.next().unwrap.unwrap, true, version)
  }
 
  override def close(): Unit = iterable.close()

standalone/src/main/scala/io/delta/standalone/internal/scan/DeltaScanImpl.scala

@@ -82,7 +82,7 @@ private[internal] class DeltaScanImpl(replay: MemoryOptimizedLogReplay) extends
  */
  private def findNextValid(): Option[AddFile] = {
  while (iter.hasNext) {
- val (action, isCheckpoint) = iter.next()
+ val (action, isCheckpoint, _) = iter.next()
 
  action match {
  case add: AddFile =>