Resolve conflicts with upstream/master

Author: cfbber

build.sh

@@ -19,6 +19,8 @@
 ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")" &>/dev/null && pwd)"
 
 rm -rf version.go
+# Formatting Go code with gofmt
+find . -name '*.go' -exec gofmt -w {} \;
 go generate
 go build
 echo "Build success. Output: ${ROOT}/doris-streamloader"

loader/stream_loader.go

@@ -18,11 +18,13 @@
 package loader
 
 import (
+	"doris-streamloader/report"
 	"encoding/json"
 	"fmt"
 	"io"
 	"io/ioutil"
 	"net/http"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -31,7 +33,6 @@ import (
 
 	"github.com/pierrec/lz4/v4"
 	log "github.com/sirupsen/logrus"
-	"doris-streamloader/report"
 )
 
 type StreamLoadOption struct {
@@ -143,7 +144,7 @@ func (s *StreamLoad) createUrl() string {
 }
 
 // stream load create http request with string data
-func (s *StreamLoad) createRequest(url string, reader io.Reader) (req *http.Request, err error) {
+func (s *StreamLoad) createRequest(url string, reader io.Reader, workerIndex int, taskIndex int) (req *http.Request, err error) {
 	req, err = http.NewRequest("PUT", url, reader)
 	if err != nil {
 		return
@@ -155,6 +156,19 @@ func (s *StreamLoad) createRequest(url string, reader io.Reader) (req *http.Requ
 	req.Header.Set("Content-Type", "text/plain")
 	for k, v := range s.headers {
 		req.Header.Set(k, v)
+		// If a label has already been set in the headers, to prevent conflicts,
+		//generate a unique label by combining the original label, worker index, and task index.
+		if k == "label" {
+			var builder strings.Builder
+			builder.WriteString(v)
+			builder.WriteString("_")
+			builder.WriteString(strconv.Itoa(workerIndex))
+			builder.WriteString("_")
+			builder.WriteString(strconv.Itoa(taskIndex))
+
+			req.Header.Set("label", builder.String())
+		}
+
 	}
 
 	if s.Compress {
@@ -228,10 +242,10 @@ func (s *StreamLoad) readData(isEOS *atomic.Bool, rawWriter *io.PipeWriter, read
 	}
 }
 
-func (s *StreamLoad) send(url string, reader io.Reader) (*http.Response, error) {
+func (s *StreamLoad) send(url string, reader io.Reader, workerIndex int, taskIndex int) (*http.Response, error) {
 	realUrl := url
 	for {
-		req, err := s.createRequest(realUrl, reader)
+		req, err := s.createRequest(realUrl, reader, workerIndex, taskIndex)
 		if err != nil {
 			if req == nil {
 				return nil, err
@@ -347,7 +361,7 @@ func (s *StreamLoad) executeGetAndSend(maxRowsPerTask int, maxBytesPerTask int,
 				workerIndex:     workerIndex,
 				taskIndex:       taskIndex,
 			})
-		if resp, err := s.send(url, NopCloser(pr)); err != nil {
+		if resp, err := s.send(url, NopCloser(pr), workerIndex, taskIndex); err != nil {
 			s.handleSendError(workerIndex, taskIndex)
 			log.Errorf("Send error, resp: %v error message: %v", resp, err)
 			return

reader/reader.go

@@ -19,6 +19,8 @@ package file
 
 import (
 	"bufio"
+	"doris-streamloader/loader"
+	"doris-streamloader/report"
 	"io"
 	"os"
 	"path/filepath"
@@ -27,9 +29,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	loader "doris-streamloader/loader"
-	report "doris-streamloader/report"
-
 	log "github.com/sirupsen/logrus"
 )
 
@@ -138,7 +137,7 @@ func (f *FileReader) Read(reporter *report.Reporter, workers int, maxBytesPerTas
 				break
 			} else if err != nil {
 				log.Errorf("Read file failed, error message: %v, before retrying, we suggest:\n1.Check the input data files and fix if there is any problem.\n2.Do select count(*) to check whether data is partially loaded.\n3.If the data is partially loaded and duplication is unacceptable, consider dropping the table (with caution that all data in the table will be lost) and retry.\n4.Otherwise, just retry.\n", err)
-				if len(line) !=0 {
+				if len(line) != 0 {
 					log.Error("5.When using a specified line delimiter, the file must end with that delimiter.")
 				}
 				os.Exit(1)