Function copyDataToDst

backend/plugins/starrocks/tasks/tasks.go:263–378

put data to final dst database

(dc *DataConfigParams, columnMap map[string]string, orderBy string)

// put data to final dst database
func copyDataToDst(dc *DataConfigParams, columnMap map[string]string, orderBy string) error {
	c := dc.Ctx
	logger := dc.Ctx.GetLogger()
	config := dc.Config
	db := dc.SrcDb
	starrocksDb := dc.DestDb
	table := dc.SrcTableName
	starrocksTable := dc.DestTableName
	starrocksTmpTable := fmt.Sprintf("%s_tmp", starrocksTable)
	tableConfig, ok := config.TableConfigs[table]
	where := ""
	if ok {
		where = tableConfig.Where
	}
	var offset int
	var err error
	var rows dal.Rows
	rows, err = db.Cursor(
		dal.From(table),
		dal.Orderby(orderBy),
		dal.Where(where),
	)
	if err != nil {
		if strings.Contains(err.Error(), "cached plan must not change result type") {
			logger.Warn(err, "skip err: cached plan must not change result type")
			rows, err = db.Cursor(
				dal.From(table),
				dal.Orderby(orderBy),
				dal.Where(where),
			)
			if err != nil {
				return err
			}
		} else {
			return err
		}
	}
	defer rows.Close()

	var data []map[string]interface{}
	cols, err := (rows).Columns()
	if err != nil {
		return err
	}

	var batchCount int
	for rows.Next() {
		select {
		case <-c.GetContext().Done():
			return c.GetContext().Err()
		default:
		}
		row := make(map[string]interface{})
		columns := make([]interface{}, len(cols))
		columnPointers := make([]interface{}, len(cols))
		for i := range columns {
			dataType := columnMap[cols[i]]
			if strings.HasPrefix(dataType, "array") {