func()

in reader/reader.go [111:193]

• 67 lines of code
• 19 McCabe index (conditional complexity)

func (f *FileReader) Read(reporter *report.Reporter, workers int, maxBytesPerTask int, retryInfo *map[int]int,
	loadResp *loader.Resp, retryCount int, lineDelimiter byte) {
	index := 0
	data := f.pool.Get().([]byte)
	count := f.batchRows
	// sendBytesMap used to record how many bytes a task has sent and whether it has ended
	// currentTaskMap used to record current task number
	// record these two to determine which data needs to be reload by using currentTaskMap[workerNumber] >= taskIndex
	sendBytesMap := make(map[int]int)
	currentTaskMap := make(map[int]int)
	for i := 0; i < workers; i++ {
		sendBytesMap[i] = maxBytesPerTask
		currentTaskMap[i] = 1
	}

	for _, file := range f.files {
		loadResp.LoadFiles = append(loadResp.LoadFiles, file.Name())
		reader := bufio.NewReaderSize(file, f.bufferSize)

		for {
			if atomic.LoadUint64(&reporter.FinishedWorkers) == atomic.LoadUint64(&reporter.TotalWorkers) {
				return
			}
			line, err := reader.ReadBytes(lineDelimiter)
			if err == io.EOF && len(line) == 0 {
				file.Close()
				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 {
					log.Error("5.When using a specified line delimiter, the file must end with that delimiter.")
				}
				os.Exit(1)
			}

			// copy line into data
			data = append(data, line...)

			count--
			if retryCount == 0 {
				loadResp.TotalRows++
			}
			if count == 0 || len(data) > f.batchBytes {
				reporter.Lock.Lock()
				_, needSkipWorker := reporter.FailedWorkers[index%workers]
				reporter.Lock.Unlock()
				// if the worker is not failed
				if !needSkipWorker {
					needSkipTask := false
					// need retry
					if len(*retryInfo) > 0 {
						// judge if need skip data of task
						taskIndex, isRetry := (*retryInfo)[index%workers]
						if !(isRetry && currentTaskMap[index%workers] >= taskIndex) {
							needSkipTask = true
						}
					}
					// if do not need skip task, send data to queue
					if !needSkipTask {
						f.sendToQueue(reporter, data, index%workers)
					}
					// update reporter
					reporter.IncrSendBytes(int64(len(data)))
					// if sendBytes less than 0,
					// add task number and reset send byte to maxBytesPerTask
					sendBytesMap[index%workers] -= len(data)
					if sendBytesMap[index%workers] <= 0 {
						sendBytesMap[index%workers] = maxBytesPerTask
						currentTaskMap[index%workers] += 1
					}
				}
				data = f.pool.Get().([]byte)
				count = f.batchRows
				index++
			}
		}
	}
	// send remain data
	if len(data) > 0 {
		reporter.IncrSendBytes(int64(len(data)))
		(*f.queues)[index%workers] <- data
	}
}