Function ExportData

backend/plugins/starrocks/tasks/tasks.go:64–132

(c plugin.SubTaskContext)

}

func ExportData(c plugin.SubTaskContext) errors.Error {
	logger := c.GetLogger()
	config := c.GetData().(*StarRocksConfig)

	// 1. Get db instance
	var db dal.Dal
	if config.SourceDsn != "" && config.SourceType != "" {
		o, err := getDbInstance(c)
		if err != nil {
			return errors.Convert(err)
		}
		db = dalgorm.NewDalgorm(o)
		sqlDB, err := o.DB()
		if err != nil {
			return errors.Convert(err)
		}
		defer sqlDB.Close()
	} else {
		db = c.GetDal()
	}

	// 2. Filter out the tables to export
	starrocksTables, err := getExportingTables(c, db)
	if err != nil {
		return errors.Convert(err)
	}
	// 3. copy devlake data to starrocks
	sr, err := gorm.Open(mysql.Open(fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8mb4&parseTime=True&loc=Local", config.User, config.Password, config.Host, config.Port, config.Database)))
	if err != nil {
		return errors.Convert(err)
	}
	starrocksDb := dalgorm.NewDalgorm(sr)
	sqlStarrocksDB, err := sr.DB()
	if err != nil {
		return errors.Convert(err)
	}
	defer sqlStarrocksDB.Close()

	for _, table := range starrocksTables {
		select {
		case <-c.GetContext().Done():
			return errors.Convert(c.GetContext().Err())
		default:
		}

		dc := DataConfigParams{
			Ctx:           c,
			Config:        config,
			SrcDb:         db,
			DestDb:        starrocksDb,
			SrcTableName:  table,
			DestTableName: table,
		}
		columnMap, orderBy, skip, err := createTmpTableInStarrocks(&dc)
		if skip {
			logger.Info(fmt.Sprintf("table %s is up to date, so skip it", table))
			continue
		}