[refactor](load) Refactor the loader to expand the way data is written

spark-doris-connector/pom.xml

@@ -67,7 +67,7 @@
     </mailingLists>
 
     <properties>
-        <revision>1.3.0-SNAPSHOT</revision>
+        <revision>1.4.0-SNAPSHOT</revision>
         <spark.version>3.1.2</spark.version>
         <spark.major.version>3.1</spark.major.version>
         <scala.version>2.12</scala.version>

spark-doris-connector/src/main/java/org/apache/doris/spark/load/DorisStreamLoad.java

@@ -31,6 +31,7 @@
 import org.apache.doris.spark.rest.RestService;
 import org.apache.doris.spark.rest.models.BackendV2;
 import org.apache.doris.spark.rest.models.RespContent;
+import org.apache.doris.spark.sql.Utils;
 import org.apache.doris.spark.util.ResponseUtil;
 import org.apache.http.HttpHeaders;
 import org.apache.http.HttpResponse;
@@ -77,6 +78,7 @@
 /**
  * DorisStreamLoad
  **/
+@Deprecated
 public class DorisStreamLoad implements Serializable {
 
     private static final Logger LOG = LoggerFactory.getLogger(DorisStreamLoad.class);

spark-doris-connector/src/main/java/org/apache/doris/spark/rest/models/RespContent.java

@@ -21,9 +21,15 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
 @JsonIgnoreProperties(ignoreUnknown = true)
 public class RespContent {
 
+    private final static List<String> DORIS_SUCCESS_STATUS = new ArrayList<>(Arrays.asList("Success", "Publish Timeout"));
+
     @JsonProperty(value = "TxnId")
     private long TxnId;
 
@@ -97,4 +103,9 @@ public String toString() {
         }
 
     }
+
+    public boolean isSuccess() {
+        return DORIS_SUCCESS_STATUS.contains(getStatus());
+    }
+
 }

spark-doris-connector/src/main/scala/org/apache/doris/spark/load/Loader.scala

@@ -0,0 +1,58 @@
+// 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.doris.spark.load
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.types.StructType
+
+/**
+ * Loader, interface class for write data to doris
+ */
+trait Loader extends Serializable {
+
+  /**
+   * execute load
+   *
+   * @param iterator row data iterator
+   * @param schema row data schema
+   * @return commit message
+   */
+  def load(iterator: Iterator[InternalRow], schema: StructType): Option[CommitMessage]
+
+  /**
+   * commit transaction
+   *
+   * @param msg commit message
+   */
+  def commit(msg: CommitMessage): Unit
+
+  /**
+   * abort transaction
+   *
+   * @param msg commit message
+   */
+  def abort(msg: CommitMessage): Unit
+
+}
+
+/**
+ * Commit message class
+ *
+ * @param value message value
+ */
+case class CommitMessage(value: Any) extends Serializable