@@ -1604,3 +1604,152 @@ TEST_F(TabletRawChunkStripTest, largePayloadMultipleChunks) {
16041604 ASSERT_EQ (result.size (), 1 );
16051605 EXPECT_EQ (result[0 ].second , payload);
16061606}
1607+
1608+ // Tests for ZSTD_DCtx reuse in StreamData and StreamDataReader (D99456594).
1609+ // Inherits from TabletRawChunkStripTest for shared helpers
1610+ // (buildCompressedChunk, pool_, etc.).
1611+
1612+ class ZstdDCtxReuseTest : public TabletRawChunkStripTest {
1613+ protected:
1614+ void SetUp () override {
1615+ TabletRawChunkStripTest::SetUp ();
1616+ dctx_.reset (ZSTD_createDCtx ());
1617+ }
1618+
1619+ // Build legacy-format compressed stream data for a non-string scalar:
1620+ // [CompressionType::Zstd (1B)][ZSTD-compressed payload]
1621+ static std::string buildLegacyCompressedData (std::string_view payload) {
1622+ std::string result;
1623+ result.push_back (static_cast <char >(CompressionType::Zstd));
1624+ const auto maxSize = ZSTD_compressBound (payload.size ());
1625+ const auto offset = result.size ();
1626+ result.resize (offset + maxSize);
1627+ const auto compressedSize = ZSTD_compress (
1628+ result.data () + offset, maxSize, payload.data (), payload.size (), 1 );
1629+ NIMBLE_CHECK (!ZSTD_isError (compressedSize));
1630+ result.resize (offset + compressedSize);
1631+ return result;
1632+ }
1633+
1634+ // Like TabletRawChunkStripTest::deserializeTabletRaw, but passes
1635+ // the fixture's shared dctx to StreamDataReader.
1636+ std::vector<std::pair<uint32_t , std::string>> deserializeTabletRawWithDCtx (
1637+ uint32_t rowCount,
1638+ const std::vector<std::pair<uint32_t , std::string>>& streams) {
1639+ uint32_t maxOffset = 0 ;
1640+ for (const auto & [offset, _] : streams) {
1641+ maxOffset = std::max (maxOffset, offset);
1642+ }
1643+ std::vector<uint32_t > sizes (streams.empty () ? 0 : maxOffset + 1 , 0 );
1644+ std::string streamData;
1645+ for (const auto & [offset, data] : streams) {
1646+ sizes[offset] = static_cast <uint32_t >(data.size ());
1647+ streamData.append (data);
1648+ }
1649+
1650+ std::string buffer;
1651+ serde::detail::writeHeader (
1652+ buffer, SerializationVersion::kTabletRaw , rowCount);
1653+ buffer.append (streamData);
1654+ serde::detail::writeRawTrailer (sizes, EncodingType::Trivial, buffer);
1655+
1656+ DeserializerOptions options{.hasHeader = true };
1657+ StreamDataReader reader (pool_.get (), options, dctx_.get ());
1658+ auto actualRows = reader.initialize (std::string_view (buffer));
1659+ EXPECT_EQ (actualRows, rowCount);
1660+
1661+ std::vector<std::pair<uint32_t , std::string>> result;
1662+ reader.iterateStreams ([&](uint32_t offset, std::string_view data) {
1663+ result.emplace_back (offset, std::string (data));
1664+ });
1665+ return result;
1666+ }
1667+
1668+ struct DCtxDeleter {
1669+ void operator ()(ZSTD_DCtx* ctx) const {
1670+ ZSTD_freeDCtx (ctx);
1671+ }
1672+ };
1673+
1674+ std::unique_ptr<ZSTD_DCtx, DCtxDeleter> dctx_;
1675+ };
1676+
1677+ TEST_F (ZstdDCtxReuseTest, streamDataLegacyZstdWithDCtx) {
1678+ const std::vector<int32_t > expected = {10 , 20 , 30 , 40 };
1679+ std::string_view payload (
1680+ reinterpret_cast <const char *>(expected.data ()),
1681+ expected.size () * sizeof (int32_t ));
1682+ auto compressed = buildLegacyCompressedData (payload);
1683+
1684+ serde::StreamData sd (
1685+ ScalarKind::Int32,
1686+ SerializationVersion::kLegacy ,
1687+ compressed,
1688+ pool_.get (),
1689+ dctx_.get ());
1690+
1691+ std::vector<int32_t > output (expected.size ());
1692+ sd.copyTo (
1693+ reinterpret_cast <char *>(output.data ()), output.size () * sizeof (int32_t ));
1694+ EXPECT_EQ (output, expected);
1695+ }
1696+
1697+ TEST_F (ZstdDCtxReuseTest, streamDataDCtxReusedAcrossReset) {
1698+ const std::vector<int32_t > values1 = {1 , 2 , 3 };
1699+ std::string_view payload1 (
1700+ reinterpret_cast <const char *>(values1.data ()),
1701+ values1.size () * sizeof (int32_t ));
1702+ auto compressed1 = buildLegacyCompressedData (payload1);
1703+
1704+ const std::vector<int32_t > values2 = {100 , 200 };
1705+ std::string_view payload2 (
1706+ reinterpret_cast <const char *>(values2.data ()),
1707+ values2.size () * sizeof (int32_t ));
1708+ auto compressed2 = buildLegacyCompressedData (payload2);
1709+
1710+ // First decompression with dctx.
1711+ serde::StreamData sd (
1712+ ScalarKind::Int32,
1713+ SerializationVersion::kLegacy ,
1714+ compressed1,
1715+ pool_.get (),
1716+ dctx_.get ());
1717+
1718+ std::vector<int32_t > output1 (values1.size ());
1719+ sd.copyTo (
1720+ reinterpret_cast <char *>(output1.data ()),
1721+ output1.size () * sizeof (int32_t ));
1722+ EXPECT_EQ (output1, values1);
1723+
1724+ // Reset with new data, same dctx.
1725+ sd.reset (compressed2, SerializationVersion::kLegacy , dctx_.get ());
1726+
1727+ std::vector<int32_t > output2 (values2.size ());
1728+ sd.copyTo (
1729+ reinterpret_cast <char *>(output2.data ()),
1730+ output2.size () * sizeof (int32_t ));
1731+ EXPECT_EQ (output2, values2);
1732+ }
1733+
1734+ TEST_F (ZstdDCtxReuseTest, streamDataReaderCompressedChunkWithDCtx) {
1735+ std::string payload = " compressed data that should be zstd encoded for test"
1736+ auto chunk = buildCompressedChunk (payload);
1737+ auto result = deserializeTabletRawWithDCtx (10 , {{0 , chunk}});
1738+
1739+ ASSERT_EQ (result.size (), 1 );
1740+ EXPECT_EQ (result[0 ].first , 0 );
1741+ EXPECT_EQ (result[0 ].second , payload);
1742+ }
1743+
1744+ TEST_F (ZstdDCtxReuseTest, streamDataReaderDCtxReusedAcrossStreams) {
1745+ std::string payload1 = " first compressed stream payload data"
1746+ std::string payload2 = " second compressed stream payload data"
1747+ auto chunk1 = buildCompressedChunk (payload1);
1748+ auto chunk2 = buildCompressedChunk (payload2);
1749+
1750+ auto result = deserializeTabletRawWithDCtx (10 , {{0 , chunk1}, {1 , chunk2}});
1751+
1752+ ASSERT_EQ (result.size (), 2 );
1753+ EXPECT_EQ (result[0 ].second , payload1);
1754+ EXPECT_EQ (result[1 ].second , payload2);
1755+ }
![meta-codesync[bot]](https://avatars.githubusercontent.com/in/1373969?v=4&size=40){kind=avatar}
0 commit comments