[#1750] feat(remote merge): Support Spark.

client-spark/common/pom.xml

@@ -89,6 +89,12 @@
             <groupId>net.jpountz.lz4</groupId>
             <artifactId>lz4</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.uniffle</groupId>
+            <artifactId>rss-common</artifactId>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

client-spark/common/src/main/java/org/apache/spark/shuffle/RssSparkConfig.java

@@ -447,6 +447,27 @@ public class RssSparkConfig {
                           + "sequence number, the partition id and the task attempt id."))
           .createWithDefault(1048576);
 
+  public static final ConfigEntry<Boolean> RSS_REMOTE_MERGE_ENABLE =
+      createBooleanBuilder(
+              new ConfigBuilder(SPARK_RSS_CONFIG_PREFIX + RssClientConfig.RSS_REMOTE_MERGE_ENABLE)
+                  .doc("Whether to enable remote merge"))
+          .createWithDefault(false);
+
+  public static final ConfigEntry<Integer> RSS_MERGED_BLOCK_SZIE =
+      createIntegerBuilder(
+              new ConfigBuilder(SPARK_RSS_CONFIG_PREFIX + RssClientConfig.RSS_MERGED_BLOCK_SZIE)
+                  .internal()
+                  .doc("The merged block size."))
+          .createWithDefault(RssClientConfig.RSS_MERGED_BLOCK_SZIE_DEFAULT);
+
+  public static final ConfigEntry<String> RSS_REMOTE_MERGE_CLASS_LOADER =
+      createStringBuilder(
+              new ConfigBuilder(
+                      SPARK_RSS_CONFIG_PREFIX + RssClientConfig.RSS_REMOTE_MERGE_CLASS_LOADER)
+                  .internal()
+                  .doc("The class loader label for remote merge"))
+          .createWithDefault(null);
+
   // spark2 doesn't have this key defined
   public static final String SPARK_SHUFFLE_COMPRESS_KEY = "spark.shuffle.compress";
 

client-spark/common/src/main/java/org/apache/spark/shuffle/SparkCombiner.java

@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.spark.Aggregator;
+
+import org.apache.uniffle.client.record.Record;
+import org.apache.uniffle.client.record.writer.Combiner;
+
+public class SparkCombiner<K, V, C> extends Combiner<K, V, C> {
+
+  private Aggregator<K, V, C> agg;
+
+  public SparkCombiner(Aggregator<K, V, C> agg) {
+    this.agg = agg;
+  }
+
+  @Override
+  public List<Record> combineValues(Iterator<Map.Entry<K, List<Record>>> recordIterator) {
+    List<Record> ret = new ArrayList<>();
+    while (recordIterator.hasNext()) {
+      Map.Entry<K, List<Record>> entry = recordIterator.next();
+      List<Record> records = entry.getValue();
+      Record<K, C> current = null;
+      for (Record<K, V> record : records) {
+        if (current == null) {
+          // Handle new Key
+          current = Record.create(record.getKey(), agg.createCombiner().apply(record.getValue()));
+        } else {
+          // Combine the values
+          C newValue = agg.mergeValue().apply(current.getValue(), record.getValue());
+          current.setValue(newValue);
+        }
+      }
+      ret.add(current);
+    }
+    return ret;
+  }
+
+  @Override
+  public List<Record> combineCombiners(Iterator<Map.Entry<K, List<Record>>> recordIterator) {
+    List<Record> ret = new ArrayList<>();
+    while (recordIterator.hasNext()) {
+      Map.Entry<K, List<Record>> entry = recordIterator.next();
+      List<Record> records = entry.getValue();
+      Record<K, C> current = null;
+      for (Record<K, C> record : records) {
+        if (current == null) {
+          // Handle new Key
+          current = record;
+        } else {
+          // Combine the values
+          C newValue = agg.mergeCombiners().apply(current.getValue(), record.getValue());
+          current.setValue(newValue);
+        }
+      }
+      ret.add(current);
+    }
+    return ret;
+  }
+}

client-spark/common/src/main/java/org/apache/spark/shuffle/reader/RMRssShuffleDataIterator.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle.reader;
+
+import java.io.IOException;
+
+import scala.collection.AbstractIterator;
+import scala.runtime.BoxedUnit;
+
+import org.apache.uniffle.client.record.Record;
+import org.apache.uniffle.client.record.reader.KeyValueReader;
+import org.apache.uniffle.client.record.reader.RMRecordsReader;
+import org.apache.uniffle.common.exception.RssException;
+
+public class RMRssShuffleDataIterator<K, C> extends AbstractIterator<Record<K, C>> {
+
+  private RMRecordsReader<K, ?, C> reader;
+  private KeyValueReader<K, C> keyValueReader;
+
+  public RMRssShuffleDataIterator(RMRecordsReader<K, ?, C> reader) {
+    this.reader = reader;
+    this.keyValueReader = reader.keyValueReader();
+  }
+
+  @Override
+  public boolean hasNext() {
+    try {
+      return this.keyValueReader.hasNext();
+    } catch (IOException e) {
+      throw new RssException(e);
+    }
+  }
+
+  @Override
+  public Record<K, C> next() {
+    try {
+      return this.keyValueReader.next();
+    } catch (IOException e) {
+      throw new RssException(e);
+    }
+  }
+
+  public BoxedUnit cleanup() {
+    reader.close();
+    return BoxedUnit.UNIT;
+  }
+}