Skip memmove in prepareInputBuffer when trailing space suffices

Summary:
Re-lands D95740314 with a bug fix.

In `prepareInputBuffer()`, when `inputData_` already points into `inputBuffer_` and there is enough trailing capacity for the requested size, we can skip the memmove that compacts data to the front of the buffer. This avoids unnecessary data movement in the common case where chunks are consumed incrementally.

The original D95740314 had a bug: `ensureInput()` appended new data at `inputBuffer_->asMutable<char>() + inputSize_` (buffer start + inputSize_), which is wrong when `inputData_` is mid-buffer after the memmove skip. Fixed by changing the memcpy destination to `const_cast<char*>(inputData_) + inputSize_`, which correctly appends relative to wherever `inputData_` currently points.

All usage sites of `inputData_`/`inputBuffer_` were audited:
- `loadNextChunk()`: reads from `inputData_`, no writes to buffer — safe
- `ensureInputIncremental_hack()`: arithmetic relative to `inputData_` — safe
- `prepareInputBuffer()` realloc path: copies from `inputData_`, updates pointer — safe
- `prepareInputBuffer()` memmove path: resets `inputData_` to buffer start — safe
- `seekToChunk()`: clean reset of `inputData_` and `inputSize_` — safe

Differential Revision: D97329725

Author: tanjialiang

dwio/nimble/velox/selective/ChunkedDecoder.cpp

@@ -80,7 +80,7 @@ bool ChunkedDecoder::ensureInput(int size) {
     } else {
       prepareInputBuffer(inputSize_ + len);
       // Append after the previous content.
-      memcpy(inputBuffer_->asMutable<char>() + inputSize_, buf, len);
+      memcpy(const_cast<char*>(inputData_) + inputSize_, buf, len);
     }
     inputSize_ += len;
   }
@@ -98,15 +98,26 @@ bool ChunkedDecoder::ensureInputIncremental_hack(int size, const char*& pos) {
 
 // After this function is called, we ensure these:
 // 1. `inputBuffer_' is allocated and at least `size' bytes large.
-// 2. The first `inputSize_' bytes in `inputData_' before the call are copied to
-//    the beginning of `inputBuffer_'.
-// 3. `inputData_' is pointing to `inputBuffer_'.
+// 2. The first `inputSize_' bytes in `inputData_' before the call are in
+//    `inputBuffer_' (either at their current position or compacted to the
+//    front).
+// 3. `inputData_' points into `inputBuffer_' with at least `size - inputSize_'
+//    bytes of trailing space available for appending.
 void ChunkedDecoder::prepareInputBuffer(int32_t size) {
   NIMBLE_DCHECK_LE(inputSize_, size);
   if (inputBuffer_ && size <= inputBuffer_->capacity()) {
     if (inputData_ == inputBuffer_->as<char>()) {
       return;
     }
+    // Check if we already point into the buffer and the remaining data plus
+    // the new data fit without compacting to the front.
+    const char* bufStart = inputBuffer_->as<char>();
+    const char* bufEnd = bufStart + inputBuffer_->capacity();
+    if (inputData_ >= bufStart && inputData_ < bufEnd &&
+        inputData_ + size <= bufEnd) {
+      // Enough trailing space — skip the memmove.
+      return;
+    }
     char* newInputData = inputBuffer_->asMutable<char>();
     if (inputSize_ > 0) {
       memmove(newInputData, inputData_, inputSize_);

dwio/nimble/velox/selective/tests/ChunkedDecoderTest.cpp

@@ -148,6 +148,52 @@ TEST_F(ChunkedDecoderTest, ensureInput) {
   ASSERT_FALSE(helper.ensureInput(1));
 }
 
+// Verify that ensureInput correctly appends data when inputData_ is mid-buffer
+// (the memmove-skip optimization in prepareInputBuffer fires).
+// Regression test: the old code used inputBuffer_->asMutable<char>() +
+// inputSize_ for the memcpy destination, which is wrong when inputData_ doesn't
+// point to the start of inputBuffer_. The fix uses
+// const_cast<char*>(inputData_)
+// + inputSize_.
+TEST_F(ChunkedDecoderTest, ensureInputMemmoveSkip) {
+  // Build 32 bytes of deterministic data.
+  std::string data(32, '\0');
+  for (size_t i = 0; i < data.size(); ++i) {
+    data[i] = 'a' + (i % 26);
+  }
+
+  // block_size = 3: Next() returns 3 bytes at a time, forcing ensureInput to
+  // loop and accumulate data in inputBuffer_.
+  ChunkedDecoder decoder(
+      std::make_unique<dwio::common::SeekableArrayInputStream>(
+          data.data(), data.size(), /*block_size=*/3),
+      false,
+      nullptr,
+      &pool());
+  ChunkedDecoderTestHelper helper(&decoder);
+
+  // Step 1: Fill the internal buffer with 9 bytes ("abcdefghi").
+  // This forces inputBuffer_ allocation; Velox AlignedBuffer rounds capacity
+  // up (typically to 64 bytes), so there is plenty of trailing space.
+  helper.ensureInput(9);
+  ASSERT_EQ(helper.inputData().substr(0, 9), data.substr(0, 9));
+
+  // Step 2: Consume 3 bytes. inputData_ advances to bufStart+3, inputSize_=6.
+  // Now inputData_ no longer points to the start of inputBuffer_.
+  helper.advanceInputData(3);
+  ASSERT_EQ(helper.inputData(), data.substr(3, 6));
+
+  // Step 3: Request 9 bytes total (have 6, need 3 more from the stream).
+  // prepareInputBuffer sees inputData_ mid-buffer with enough trailing space
+  // and skips the memmove. Then Next() returns 3 more bytes that must be
+  // appended right after the existing 6 bytes — i.e. at inputData_+6
+  // (= bufStart+9), NOT at bufStart+6.
+  // With the old bug, the memcpy destination was bufStart+inputSize_ =
+  // bufStart+6, which overwrote the tail of the existing data, corrupting it.
+  helper.ensureInput(9);
+  ASSERT_EQ(helper.inputData().substr(0, 9), data.substr(3, 9));
+}
+
 // Test fixture for ChunkedDecoder data operations with parameterized
 // stream index support.
 class ChunkedDecoderDataTest : public index::test::TabletIndexTestBase,