Optimize PagedInputStream::Skip (#6699)

Summary:

Currently when we skip bytes in `PagedInputStream`, we do the decompression unconditionally and it is expensive.  Some optimizations are added to address this:
1. Skip decompression of the whole block (frame in case of ZSTD) if
   1. We can get the precise decompressed size, and
   2. The decompressed size is no larger than the bytes need to skip
2. Accumulate contiguous skip calls to create larger skip region (delayed skipping)
3. Fix `ByteRleDecoder::skipBytes` to avoid reading data and breaking contiguous skips

Differential Revision: D49501856

Author: Yuhta

velox/dwio/common/CacheInputStream.cpp

@@ -212,7 +212,7 @@ void CacheInputStream::loadSync(Region region) {
     } else {
       // Hit memory cache.
       if (!entry->getAndClearFirstUseFlag()) {
-        ioStats_->ramHit().increment(entry->size());
+        ioStats_->ramHit().increment(region.length);
       }
       return;
     }
@@ -269,7 +269,7 @@ bool CacheInputStream::loadFromSsd(
     throw;
   }
   pin_ = std::move(pins[0]);
-  ioStats_->ssdRead().increment(entry.size());
+  ioStats_->ssdRead().increment(region.length);
   ioStats_->queryThreadIoLatency().increment(usec);
   entry.setExclusiveToShared();
   return true;

velox/dwio/common/compression/Compression.cpp

@@ -246,24 +246,31 @@ class ZstdDecompressor : public Decompressor {
   explicit ZstdDecompressor(
       uint64_t blockSize,
       const std::string& streamDebugInfo)
-      : Decompressor{blockSize, streamDebugInfo} {}
+      : Decompressor{blockSize, streamDebugInfo}, context_{ZSTD_createDCtx()} {}
+
+  ~ZstdDecompressor() override {
+    ZSTD_freeDCtx(context_);
+  }
 
   uint64_t decompress(
       const char* src,
       uint64_t srcLength,
       char* dest,
       uint64_t destLength) override;
 
-  uint64_t getUncompressedLength(const char* src, uint64_t srcLength)
+  int64_t getDecompressedLength(const char* src, uint64_t srcLength)
       const override;
+
+ private:
+  ZSTD_DCtx* context_;
 };
 
 uint64_t ZstdDecompressor::decompress(
     const char* src,
     uint64_t srcLength,
     char* dest,
     uint64_t destLength) {
-  auto ret = ZSTD_decompress(dest, destLength, src, srcLength);
+  auto ret = ZSTD_decompressDCtx(context_, dest, destLength, src, srcLength);
   DWIO_ENSURE(
       !ZSTD_isError(ret),
       "ZSTD returned an error: ",
@@ -273,15 +280,15 @@ uint64_t ZstdDecompressor::decompress(
   return ret;
 }
 
-uint64_t ZstdDecompressor::getUncompressedLength(
+int64_t ZstdDecompressor::getDecompressedLength(
     const char* src,
     uint64_t srcLength) const {
   auto uncompressedLength = ZSTD_getFrameContentSize(src, srcLength);
   // in the case when decompression size is not available, return the upper
   // bound
   if (uncompressedLength == ZSTD_CONTENTSIZE_UNKNOWN ||
       uncompressedLength == ZSTD_CONTENTSIZE_ERROR) {
-    return blockSize_;
+    return -blockSize_;
   }
   DWIO_ENSURE_LE(
       uncompressedLength,
@@ -304,7 +311,7 @@ class SnappyDecompressor : public Decompressor {
       char* dest,
       uint64_t destLength) override;
 
-  uint64_t getUncompressedLength(const char* src, uint64_t srcLength)
+  int64_t getDecompressedLength(const char* src, uint64_t srcLength)
       const override;
 };
 
@@ -313,7 +320,7 @@ uint64_t SnappyDecompressor::decompress(
     uint64_t srcLength,
     char* dest,
     uint64_t destLength) {
-  auto length = getUncompressedLength(src, srcLength);
+  auto length = std::abs(getDecompressedLength(src, srcLength));
   DWIO_ENSURE_GE(destLength, length);
   DWIO_ENSURE(
       snappy::RawUncompress(src, srcLength, dest),
@@ -322,14 +329,14 @@ uint64_t SnappyDecompressor::decompress(
   return length;
 }
 
-uint64_t SnappyDecompressor::getUncompressedLength(
+int64_t SnappyDecompressor::getDecompressedLength(
     const char* src,
     uint64_t srcLength) const {
   size_t uncompressedLength;
   // in the case when decompression size is not available, return the upper
   // bound
   if (!snappy::GetUncompressedLength(src, srcLength, &uncompressedLength)) {
-    return blockSize_;
+    return -blockSize_;
   }
   DWIO_ENSURE_LE(
       uncompressedLength,
@@ -339,6 +346,7 @@ uint64_t SnappyDecompressor::getUncompressedLength(
   return uncompressedLength;
 }
 
+// TODO: Is this really needed?
 class ZlibDecompressionStream : public PagedInputStream,
                                 private ZlibDecompressor {
  public:
@@ -355,13 +363,18 @@ class ZlibDecompressionStream : public PagedInputStream,
         ZlibDecompressor{blockSize, windowBits, streamDebugInfo, isGzip} {}
   ~ZlibDecompressionStream() override = default;
 
-  bool Next(const void** data, int32_t* size) override;
+  bool readOrSkip(const void** data, int32_t* size) override;
 };
 
-bool ZlibDecompressionStream::Next(const void** data, int32_t* size) {
+bool ZlibDecompressionStream::readOrSkip(const void** data, int32_t* size) {
+  if (data) {
+    VELOX_CHECK_EQ(pendingSkip_, 0);
+  }
   // if the user pushed back, return them the partial buffer
   if (outputBufferLength_) {
-    *data = outputBufferPtr_;
+    if (data) {
+      *data = outputBufferPtr_;
+    }
     *size = static_cast<int32_t>(outputBufferLength_);
     outputBufferPtr_ += outputBufferLength_;
     bytesReturned_ += outputBufferLength_;
@@ -381,7 +394,9 @@ bool ZlibDecompressionStream::Next(const void** data, int32_t* size) {
       static_cast<size_t>(inputBufferPtrEnd_ - inputBufferPtr_),
       remainingLength_);
   if (state_ == State::ORIGINAL) {
-    *data = inputBufferPtr_;
+    if (data) {
+      *data = inputBufferPtr_;
+    }
     *size = static_cast<int32_t>(availSize);
     outputBufferPtr_ = inputBufferPtr_ + availSize;
     outputBufferLength_ = 0;
@@ -393,7 +408,8 @@ bool ZlibDecompressionStream::Next(const void** data, int32_t* size) {
         getName(),
         " Info: ",
         ZlibDecompressor::streamDebugInfo_);
-    prepareOutputBuffer(getUncompressedLength(inputBufferPtr_, availSize));
+    prepareOutputBuffer(
+        std::abs(getDecompressedLength(inputBufferPtr_, availSize)));
 
     reset();
     zstream_.next_in =
@@ -432,7 +448,9 @@ bool ZlibDecompressionStream::Next(const void** data, int32_t* size) {
       }
     } while (result != Z_STREAM_END);
     *size = static_cast<int32_t>(blockSize_ - zstream_.avail_out);
-    *data = outputBufferPtr_;
+    if (data) {
+      *data = outputBufferPtr_;
+    }
     outputBufferLength_ = 0;
     outputBufferPtr_ += *size;
   }

velox/dwio/common/compression/Compression.h

@@ -46,14 +46,17 @@ class Compressor {
 class Decompressor {
  public:
   explicit Decompressor(uint64_t blockSize, const std::string& streamDebugInfo)
-      : blockSize_{blockSize}, streamDebugInfo_{streamDebugInfo} {}
+      : blockSize_{static_cast<int64_t>(blockSize)},
+        streamDebugInfo_{streamDebugInfo} {}
 
   virtual ~Decompressor() = default;
 
-  virtual uint64_t getUncompressedLength(
-      const char* /* unused */,
-      uint64_t /* unused */) const {
-    return blockSize_;
+  // Return negative number indicating this is an estimation and should not be
+  // used for compressed skipping.
+  virtual int64_t getDecompressedLength(
+      const char* /* src */,
+      uint64_t /* srcLength */) const {
+    return -blockSize_;
   }
 
   virtual uint64_t decompress(
@@ -63,7 +66,7 @@ class Decompressor {
       uint64_t destLength) = 0;
 
  protected:
-  uint64_t blockSize_;
+  int64_t blockSize_;
   const std::string streamDebugInfo_;
 };
 

velox/dwio/common/compression/PagedInputStream.cpp

@@ -104,9 +104,21 @@ const char* PagedInputStream::ensureInput(size_t availableInputBytes) {
 }
 
 bool PagedInputStream::Next(const void** data, int32_t* size) {
+  VELOX_CHECK_NOT_NULL(data);
+  skipAllPending();
+  return readOrSkip(data, size);
+}
+
+// Read into `data' if it is not null; otherwise skip some of the pending.
+bool PagedInputStream::readOrSkip(const void** data, int32_t* size) {
+  if (data) {
+    VELOX_CHECK_EQ(pendingSkip_, 0);
+  }
   // if the user pushed back, return them the partial buffer
   if (outputBufferLength_) {
-    *data = outputBufferPtr_;
+    if (data) {
+      *data = outputBufferPtr_;
+    }
     *size = static_cast<int32_t>(outputBufferLength_);
     outputBufferPtr_ += outputBufferLength_;
     bytesReturned_ += outputBufferLength_;
@@ -139,7 +151,9 @@ bool PagedInputStream::Next(const void** data, int32_t* size) {
   // if no decompression or decryption is needed, simply adjust the output
   // pointer. Otherwise, make sure we have continuous block
   if (original) {
-    *data = inputBufferPtr_;
+    if (data) {
+      *data = inputBufferPtr_;
+    }
     *size = static_cast<int32_t>(availSize);
     outputBufferPtr_ = inputBufferPtr_ + availSize;
     inputBufferPtr_ += availSize;
@@ -154,24 +168,36 @@ bool PagedInputStream::Next(const void** data, int32_t* size) {
         decrypter_->decrypt(folly::StringPiece{input, remainingLength_});
     input = reinterpret_cast<const char*>(decryptionBuffer_->data());
     remainingLength_ = decryptionBuffer_->length();
-    *data = input;
+    if (data) {
+      *data = input;
+    }
     *size = remainingLength_;
     outputBufferPtr_ = input + remainingLength_;
   }
 
   // perform decompression
   if (state_ == State::START) {
     DWIO_ENSURE_NOT_NULL(decompressor_.get(), "invalid stream state");
-    prepareOutputBuffer(
-        decompressor_->getUncompressedLength(input, remainingLength_));
-    outputBufferLength_ = decompressor_->decompress(
-        input,
-        remainingLength_,
-        outputBuffer_->data(),
-        outputBuffer_->capacity());
-    *data = outputBuffer_->data();
-    *size = static_cast<int32_t>(outputBufferLength_);
-    outputBufferPtr_ = outputBuffer_->data() + outputBufferLength_;
+    DWIO_ENSURE_NOT_NULL(input);
+    auto decompressedLength =
+        decompressor_->getDecompressedLength(input, remainingLength_);
+    if (!data && 0 <= decompressedLength &&
+        decompressedLength <= pendingSkip_) {
+      *size = decompressedLength;
+      outputBufferPtr_ = nullptr;
+    } else {
+      prepareOutputBuffer(std::abs(decompressedLength));
+      outputBufferLength_ = decompressor_->decompress(
+          input,
+          remainingLength_,
+          outputBuffer_->data(),
+          outputBuffer_->capacity());
+      if (data) {
+        *data = outputBuffer_->data();
+      }
+      *size = static_cast<int32_t>(outputBufferLength_);
+      outputBufferPtr_ = outputBuffer_->data() + outputBufferLength_;
+    }
     // release decryption buffer
     decryptionBuffer_ = nullptr;
   }
@@ -188,6 +214,14 @@ bool PagedInputStream::Next(const void** data, int32_t* size) {
 }
 
 void PagedInputStream::BackUp(int32_t count) {
+  if (pendingSkip_ > 0) {
+    auto len = std::min<int64_t>(count, pendingSkip_);
+    pendingSkip_ -= len;
+    count -= len;
+    if (count == 0) {
+      return;
+    }
+  }
   DWIO_ENSURE(
       outputBufferPtr_ != nullptr,
       "Backup without previous Next in ",
@@ -207,25 +241,29 @@ void PagedInputStream::BackUp(int32_t count) {
   bytesReturned_ -= count;
 }
 
-bool PagedInputStream::Skip(int32_t count) {
-  // this is a stupid implementation for now.
-  // should skip entire blocks without decompressing
-  while (count > 0) {
-    const void* ptr;
+bool PagedInputStream::skipAllPending() {
+  while (pendingSkip_ > 0) {
     int32_t len;
-    if (!Next(&ptr, &len)) {
+    if (!readOrSkip(nullptr, &len)) {
       return false;
     }
-    if (len > count) {
-      BackUp(len - count);
-      count = 0;
+    if (len > pendingSkip_) {
+      auto toBackUp = len - pendingSkip_;
+      pendingSkip_ = 0;
+      BackUp(toBackUp);
     } else {
-      count -= len;
+      pendingSkip_ -= len;
     }
   }
   return true;
 }
 
+bool PagedInputStream::Skip(int32_t count) {
+  pendingSkip_ += count;
+  // We never use the return value of this function so this is OK.
+  return true;
+}
+
 void PagedInputStream::clearDecompressionState() {
   state_ = State::HEADER;
   outputBufferLength_ = 0;
@@ -243,7 +281,8 @@ void PagedInputStream::seekToPosition(
   // to the beginning of the last view or last header, whichever is
   // later. If we are returning views into the decompression buffer,
   // we can backup to the beginning of the decompressed buffer
-  auto alreadyRead = bytesReturned_ - bytesReturnedAtLastHeaderOffset_;
+  auto alreadyRead =
+      bytesReturned_ - bytesReturnedAtLastHeaderOffset_ + pendingSkip_;
 
   // outsideOriginalWindow is true if we are returning views into
   // the input stream's buffer and we are seeking below the start of the last
@@ -260,12 +299,12 @@ void PagedInputStream::seekToPosition(
     auto provider = dwio::common::PositionProvider(positions);
     input_->seekToPosition(provider);
     clearDecompressionState();
-    Skip(uncompressedOffset);
+    pendingSkip_ = uncompressedOffset;
   } else {
     if (uncompressedOffset < alreadyRead) {
       BackUp(alreadyRead - uncompressedOffset);
     } else {
-      Skip(uncompressedOffset - alreadyRead);
+      pendingSkip_ += uncompressedOffset - alreadyRead;
     }
   }
 }