misc: Refactor ShuffleTest (prestodb#26643)

Summary:

Completely replace TestShuffleWriter/Reader with local shuffle, and refactor the ShuffleTest class

Differential Revision: D87160791

Author: duxiao1212

presto-native-execution/presto_cpp/main/operators/LocalShuffle.cpp

@@ -356,6 +356,10 @@ void LocalShuffleWriter::collect(
       "key '{}' must be empty for non-sorted shuffle",
       key);
 
+  // Testvalue injection point for exception testing
+  velox::common::testutil::TestValue::adjust(
+      "facebook::presto::operators::LocalShuffleWriter::collect", this);
+
   const auto rowSize = this->rowSize(key.size(), data.size());
   auto& buffer = inProgressPartitions_[partition];
   if (buffer == nullptr) {
@@ -520,6 +524,10 @@ LocalShuffleReader::next(uint64_t maxBytes) {
       initialized_,
       "LocalShuffleReader::initialize() must be called before next()");
 
+  // Testvalue injection point for exception testing
+  velox::common::testutil::TestValue::adjust(
+      "facebook::presto::operators::LocalShuffleReader::next", this);
+
   return folly::makeSemiFuture(
       sortedShuffle_ ? nextSorted(maxBytes) : nextUnsorted(maxBytes));
 }
@@ -617,13 +625,4 @@ std::shared_ptr<ShuffleWriter> LocalPersistentShuffleFactory::createWriter(
       sortedShuffle,
       pool);
 }
-
-// Testing function to expose extractRowMetadata for tests.
-// This will be removed after reader changes.
-std::vector<RowMetadata> testingExtractRowMetadata(
-    const char* buffer,
-    size_t bufferSize,
-    bool sortedShuffle) {
-  return extractRowMetadata(buffer, bufferSize, sortedShuffle);
-}
 } // namespace facebook::presto::operators

presto-native-execution/presto_cpp/main/operators/LocalShuffle.h

@@ -53,12 +53,6 @@ inline bool compareKeys(std::string_view key1, std::string_view key2) noexcept {
   return key1.size() < key2.size();
 }
 
-// Testing function to expose extractRowMetadata for tests.
-std::vector<RowMetadata> testingExtractRowMetadata(
-    const char* buffer,
-    size_t bufferSize,
-    bool sortedShuffle);
-
 // LocalShuffleWriteInfo is used for containing shuffle write information.
 // This struct is a 1:1 strict API mapping to
 // presto-spark-base/src/main/java/com/facebook/presto/spark/execution/PrestoSparkLocalShuffleWriteInfo.java
@@ -75,7 +69,6 @@ struct LocalShuffleWriteInfo {
   /// Structures are assumed to be encoded in JSON format.
   static LocalShuffleWriteInfo deserialize(const std::string& info);
 };
-
 // LocalShuffleReadInfo is used for containing shuffle read metadata
 // This struct is a 1:1 strict API mapping to
 // presto-spark-base/src/main/java/com/facebook/presto/spark/execution/PrestoSparkLocalShuffleReadInfo.java.
@@ -87,8 +80,6 @@ struct LocalShuffleReadInfo {
   std::string queryId;
   std::vector<std::string> partitionIds;
 
-  /// Deserializes shuffle information that is used by LocalPersistentShuffle.
-  /// Structures are assumed to be encoded in JSON format.
   static LocalShuffleReadInfo deserialize(const std::string& info);
 };