fix(shuffle): Reset outputPos after flushing output buffer in CompressInternal

[zhangxffff]

What problem does this PR solve?

Issue Number: close #270

AdaptiveParallelZstdCodec::CompressInternal had a stream-state bug: after writing compressed bytes to outputStream, outputPos was not reset to 0. This could make the next compression iteration run with an effectively exhausted output window (outputLen - outputPos == 0) while keeping stale cursor state, which may produce malformed compressed stream output. On read path, this surfaced as ZSTD_decompressStream failures (Destination buffer is too small / Data corruption detected) and then INVALID_STATE. This PR also adds a regression test that reproduces the bug path.

Type of Change

• 🐛 Bug fix (non-breaking change which fixes an issue)
• ✨ New feature (non-breaking change which adds functionality)
• 🚀 Performance improvement (optimization)
• ⚠️ Breaking change (fix or feature that would cause existing functionality to change)
• 🔨 Refactoring (no logic changes)
• 🔧 Build/CI or Infrastructure changes
• 📝 Documentation only

Description

Code changes:

• bolt/shuffle/sparksql/CompressionStream.h
  • In CompressInternal, reset outputPos = 0 immediately after outputStream->Write(output, outputPos).
    Test changes:
• bolt/shuffle/sparksql/tests/AdaptiveParallelZstdCodecTest.cpp
  • Added incompressible payload builders.
  • Added CompressAndFlushStressRoundTripWithoutCorruption to stress near ZSTD_CStreamInSize() boundaries across multiple rounds/rows.
  • Test validates full public round-trip (CompressAndFlush + Decompress) and guards against stream corruption regressions.

Performance Impact

• No Impact: This change does not affect the critical path (e.g., build system, doc, error handling).

• Positive Impact: I have run benchmarks.

  Click to view Benchmark Results

  Paste your google-benchmark or TPC-H results here.
  Before: 10.5s
  After:   8.2s  (+20%)
  

• Negative Impact: Explained below (e.g., trade-off for correctness).

Checklist (For Author)

• I have added/updated unit tests (ctest).
• I have verified the code with local build (Release/Debug).
• I have run clang-format / linters.
• (Optional) I have run Sanitizers (ASAN/TSAN) locally for complex C++ changes.
• No need to test or manual test.

Breaking Changes

• No

• Yes (Description: ...)

  Click to view Breaking Changes

  Breaking Changes:
  - Description of the breaking change.
  - Possible solutions or workarounds.
  - Any other relevant information.
  

[fzhedu]

LGTM