package pimport

import (
	"bytes"
	"compress/gzip"
	"context"
	"fmt"
	"time"

	"github.com/dustin/go-humanize"
	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	dataflowpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/internal/bizutils"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	cluster_model "github.com/tidbcloud/dataflow-service/internal/model/cluster"
	import_model "github.com/tidbcloud/dataflow-service/internal/model/import"
	"github.com/tidbcloud/dataflow-service/internal/openapi/service"
	importsvc "github.com/tidbcloud/dataflow-service/internal/service/import"
	"github.com/tidbcloud/dataflow-service/internal/utils"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	"github.com/tidbcloud/idl/pbgen/proto/openapi/dp"
	"github.com/tidbcloud/pkgv2/tracing"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.opentelemetry.io/otel"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
	"gorm.io/gorm"
)

var Tracer = otel.Tracer("central/service/import")

const (
	uploadStubExpirationDuration = 10 * time.Minute
)

type Service struct {
	dp.UnimplementedImportServer
	service.BaseService
	importModel             import_model.Model
	clusterModel            cluster_model.Model
	stubManager             import_model.StubManager
	mgmtSvcGrpcClient       grpcli.MgmtSvcGrpcClient
	serverlessSvcGrpcClient grpcli.ServerlessSvcGrpcClient
}

func NewService(importModel import_model.Model, clusterModel cluster_model.Model,
	mgmtSvcGrpcClient grpcli.MgmtSvcGrpcClient, serverlessSvcGrpcClient grpcli.ServerlessSvcGrpcClient) *Service {
	return &Service{
		importModel:             importModel,
		clusterModel:            clusterModel,
		stubManager:             &import_model.StubManagerImpl{},
		mgmtSvcGrpcClient:       mgmtSvcGrpcClient,
		serverlessSvcGrpcClient: serverlessSvcGrpcClient,
	}
}

func (s *Service) ListImportTasks(ctx context.Context, req *dp.ListImportTasksReq) (_ *dp.ListImportTasksResp, gerr error) {
	ctx, span := Tracer.Start(ctx, "openapi.import.ListImportTasks")
	defer tracing.EndSpan(span, gerr)
	userContext := s.CurrentUser(ctx)
	logger := userContext.Logger()

	if err := checkListImportTasks(userContext, req, logger); err != nil {
		return nil, utils.WrapAsErrorCodeErr(err, "check get import request parameter error")
	}
	userImportModel := s.importModel.User(ctx, userContext.User())

	orgID := userContext.OrgID()
	projectID := entity.ID(req.GetProjectId())
	clusterID := entity.ID(req.GetClusterId())

	reqPage := int32(req.GetPage())
	reqPageSize := int32(req.GetPageSize())
	pager := utils.AdjustedPager(&reqPage, &reqPageSize)

	cluster, bizErr := s.mgmtSvcGrpcClient.GetClusterByID(ctx, req.GetClusterId())
	if bizErr != nil {
		logger.Error("get cluster info error", zap.Error(bizErr))
		return nil, utils.WrapAsErrorCodeErr(bizErr, "get cluster info error")
	}
	if cluster.DeployType == mgmt.DeployType_DEPLOY_TYPE_SERVERLESS {
		resp, err := s.ListServerlessImports(ctx, cluster, pager, req, userContext)
		if err != nil {
			logger.Error("list serverless imports error", zap.Error(err))
			return nil, utils.WrapAsErrorCodeErr(err, "list serverless imports error")
		}
		return resp, nil
	}

	total, imports, err := userImportModel.ListImport(ctx, orgID, projectID, clusterID, pager, nil)
	if err != nil {
		logger.Error("failed to list import", zap.Uint64("cluster id", req.GetClusterId()), zap.Error(err))
		if errors.Cause(err) == gorm.ErrRecordNotFound {
			return nil, errcode.NewNotFoundErr(err)
		}
		return nil, utils.WrapAsErrorCodeErr(err, "failed to list imports")
	}
	resultItems := []*dp.ImportItem{}
	for _, imp := range imports {
		impItem := ImportEntity2ImportItem(logger, imp)
		resultItems = append(resultItems, impItem)
	}
	return &dp.ListImportTasksResp{
		Items: resultItems,
		Total: uint32(total),
	}, nil
}

func (s *Service) GetImportTask(ctx context.Context, req *dp.GetImportTaskReq) (_ *dp.ImportItem, gerr error) {
	ctx, span := Tracer.Start(ctx, "openapi.import.GetImportTask")
	defer tracing.EndSpan(span, gerr)
	userContext := s.CurrentUser(ctx)
	logger := userContext.Logger()

	if err := checkGetImportTask(userContext, req, logger); err != nil {
		return nil, utils.WrapAsErrorCodeErr(err, "check get import request parameter error")
	}
	userImportModel := s.importModel.User(ctx, userContext.User())

	cluster, bizErr := s.mgmtSvcGrpcClient.GetClusterByID(ctx, req.GetClusterId())
	if bizErr != nil {
		logger.Error("get cluster info error", zap.Error(bizErr))
		return nil, utils.WrapAsErrorCodeErr(bizErr, "get cluster info error")
	}
	if cluster.DeployType == mgmt.DeployType_DEPLOY_TYPE_SERVERLESS {
		resp, err := s.GetServerlessImport(ctx, cluster, req, userContext)
		if err != nil {
			logger.Error("get serverless import error", zap.Error(err))
			return nil, utils.WrapAsErrorCodeErr(err, "get serverless import error")
		}
		return resp, nil
	}

	orgID := userContext.OrgID()
	projectID := entity.ID(req.GetProjectId())
	clusterID := entity.ID(req.GetClusterId())
	importID := entity.ID(req.GetImportId())

	imp, err := userImportModel.GetImport(ctx, orgID, projectID, clusterID, importID)
	if err != nil {
		logger.Error("failed to get import", zap.Uint64("cluster id", req.GetClusterId()), zap.Uint64("import id", req.GetImportId()), zap.Error(err))
		if errors.Cause(err) == gorm.ErrRecordNotFound {
			return nil, errcode.NewNotFoundErr(err)
		}
		return nil, utils.WrapAsErrorCodeErr(err, "failed to get import")
	}
	resp := ImportEntity2ImportItem(logger, imp)
	return resp, nil
}

func (s *Service) GetImportTaskRoleInfo(ctx context.Context, req *dp.GetImportTaskRoleInfoReq) (_ *dp.ImportTaskRoleInfo, gerr error) {
	ctx, span := Tracer.Start(ctx, "openapi.import.GetImportTaskRoleInfo")
	defer tracing.EndSpan(span, gerr)
	userContext := s.CurrentUser(ctx)
	logger := userContext.Logger()

	if err := checkGetImportTaskRoleInfo(userContext, req, logger); err != nil {
		return nil, utils.WrapAsErrorCodeErr(err, "check get import request parameter error")
	}
	cluster, bizErr := s.mgmtSvcGrpcClient.GetClusterByID(ctx, req.GetClusterId())
	if bizErr != nil {
		logger.Error("get cluster info error", zap.Error(bizErr))
		return nil, utils.WrapAsErrorCodeErr(bizErr, "get cluster info error")
	}
	if cluster.DeployType == mgmt.DeployType_DEPLOY_TYPE_SERVERLESS {
		resp, err := s.GetDataIngestionProfile(ctx, cluster, req, userContext)
		if err != nil {
			logger.Error("get data ingestion profile error", zap.Error(err))
			return nil, utils.WrapAsErrorCodeErr(err, "get data ingestion profile error")
		}
		return resp, nil
	}

	userClusterModel := s.clusterModel.User(ctx, userContext.User())
	profile, err := userClusterModel.GetDataIngestionProfile(ctx, entity.ID(req.GetClusterId()))
	if err != nil {
		logger.Error("failed to get data ingestion profile from cluster model", zap.Uint64("cluster id", req.GetClusterId()), zap.Error(err))
		return nil, utils.WrapAsErrorCodeErr(err, "failed to get import role information")
	}
	resp := new(dp.ImportTaskRoleInfo)
	switch profile.Type {
	case infraapi.ProviderTypeAWS, infraapi.ProviderTypeEKS:
		resp.RoleInfo = &dp.ImportTaskRoleInfo_AwsImportRole{
			AwsImportRole: &dp.AwsImportTaskRoleInfo{
				AccountId:  profile.AwsAccountId,
				ExternalId: profile.AwsExternalId,
			},
		}
	case infraapi.ProviderTypeGCP, infraapi.ProviderTypeGKE:
		resp.RoleInfo = &dp.ImportTaskRoleInfo_GcpImportRole{
			GcpImportRole: &dp.GcpImportTaskRoleInfo{
				AccountId: profile.GcpAccountId,
			},
		}
	}
	return resp, nil
}

func (s *Service) CreateImportTask(ctx context.Context, req *dp.CreateImportTaskReq) (_ *dp.CreateImportTaskResp, gerr error) {
	var err error
	ctx, span := Tracer.Start(ctx, "openapi.import.CreateImport")
	defer tracing.EndSpan(span, gerr)
	userContext := s.CurrentUser(ctx)
	logger := userContext.Logger()

	parsedInfo, err := checkCreateImportTask(userContext, req, logger)
	if err != nil {
		return nil, utils.WrapAsErrorCodeErr(err, "check request parameter error")
	}
	// pre-create the tables first, because some precheck depends on the existence of target table if schema file is not provided
	for _, tableDef := range req.GetOptions().GetPreCreateTables() {
		if _, ok := parsedInfo.TargetTableMap[getFullTableName(tableDef.GetDatabaseName(), tableDef.GetTableName())]; !ok {
			logger.Warn("The table to be created is not a target table, skip it",
				zap.String("db_name", tableDef.GetDatabaseName()),
				zap.String("table_name", tableDef.GetTableName()),
			)
			continue
		}
		localImportCfg := tableDefinition2LocalImportConfig(tableDef)
		if localImportCfg == nil {
			continue
		}

		err := s.importModel.CreateTargetTable(ctx,
			uint64(userContext.OrgID()),
			req.GetProjectId(),
			req.GetClusterId(),
			localImportCfg,
		)
		if err != nil {
			return nil, utils.WrapAsErrorCodeErr(err,
				fmt.Sprintf("pre-create target table %s.%s error",
					tableDef.GetDatabaseName(),
					tableDef.GetTableName(),
				),
			)
		}
	}

	cluster, bizErr := s.mgmtSvcGrpcClient.GetClusterByID(ctx, req.GetClusterId())
	if bizErr != nil {
		logger.Error("get cluster info error", zap.Error(bizErr))
		return nil, utils.WrapAsErrorCodeErr(bizErr, "get cluster info error")
	}
	if cluster.DeployType == mgmt.DeployType_DEPLOY_TYPE_SERVERLESS {
		resp, err := s.CreateServerlessImport(ctx, cluster, req, userContext)
		if err != nil {
			logger.Error("create serverless import error", zap.Error(err))
			return nil, utils.WrapAsErrorCodeErr(err, "create serverless import error")
		}
		return resp, nil
	}

	var stub *entity.ImportUploadStub
	if req.GetSpec().GetSource().GetType() == dp.ImportSource_LOCAL_FILE {
		stub, err = s.stubManager.GetStubByID(ctx, entity.ID(parsedInfo.UploadStubID))
		if err != nil {
			return nil, utils.WrapAsErrorCodeErr(err, "Getting upload stub has errors.")
		}
		if stub.ExpiredTime == nil {
			logger.Error("cannot find the expired time of the upload stub")
			return nil, utils.WrapAsErrorCodeErr(err, "Getting upload stub has errors.")
		}
		currentTime := time.Now()
		if stub.ExpiredTime.Before(currentTime) {
			logger.Error("The upload stub has expired",
				zap.Time("expire_time", *stub.ExpiredTime),
				zap.Time("current_time", currentTime),
			)
			return nil, errcode.NewCodedError(errors.Errorf("The upload stub has expired at: %v", *stub.ExpiredTime), errcode.StateCode)
		}
	}
	imp, err := ImportSpec2Entity(ctx, req.GetSpec(), stub, logger)
	if err != nil {
		return nil, utils.WrapAsErrorCodeErr(err, "Prepare create import has errors.")
	}
	imp.ProjectID = entity.ID(req.GetProjectId())
	imp.ClusterID = entity.ID(req.GetClusterId())
	imp.TenantID = entity.ID(cluster.OrganizationId)
	imp.Name = req.GetName()

	logger.Info("dump converted import entity", zap.Any("import_entity", imp))
	userImportModel := s.importModel.User(ctx, userContext.User())
	if err := userImportModel.ValidateImport(ctx,
		userContext.OrgID(),
		entity.ID(req.GetProjectId()),
		imp,
		[]dataflowpb.ImportValidationItem{
			dataflowpb.ImportValidationItem_SOURCE_ACCESS_CHECK,
			dataflowpb.ImportValidationItem_IMPORT_PRECHECK,
		},
	); err != nil {
		translatedErr := importsvc.TranslateValidateImportError(err, imp, logger)
		// There are some situations that after the translation, the result error is nil.
		// This helps to ignore some kind of error and do the rest of work.
		if translatedErr != nil {
			return nil, translatedErr
		}
	}
	if err := tweakImportEntityForCreate(imp); err != nil {
		logger.Error("tweak import entity for create has errors", zap.Error(err))
		return nil, utils.WrapAsErrorCodeErr(err, "prepare creating import has errors.")
	}
	newImportID, err := userImportModel.CreateImport(ctx,
		userContext.OrgID(),
		entity.ID(req.GetProjectId()),
		imp,
	)
	if err != nil {
		logger.Error("create import record has errors", zap.Error(err))
		return nil, utils.WrapAsErrorCodeErr(err, "create import record has errors.")
	}
	return &dp.CreateImportTaskResp{
		Id: newImportID,
	}, nil
}

func (s *Service) UpdateImportTask(ctx context.Context, req *dp.UpdateImportTaskReq) (_ *emptypb.Empty, gerr error) {
	ctx, span := Tracer.Start(ctx, "openapi.import.UpdateImport")
	defer tracing.EndSpan(span, gerr)
	userContext := s.CurrentUser(ctx)
	logger := userContext.Logger()

	if err := checkUpdateImportTask(userContext, req, logger); err != nil {
		return nil, utils.WrapAsErrorCodeErr(err, "check update import request parameter error")
	}
	cluster, bizErr := s.mgmtSvcGrpcClient.GetClusterByID(ctx, req.GetClusterId())
	if bizErr != nil {
		logger.Error("get cluster info error", zap.Error(bizErr))
		return nil, utils.WrapAsErrorCodeErr(bizErr, "get cluster info error")
	}
	if cluster.DeployType == mgmt.DeployType_DEPLOY_TYPE_SERVERLESS {
		err := s.UpdateServerlessImport(ctx, cluster, req, userContext)
		if err != nil {
			logger.Error("update serverless import error", zap.Error(err))
			return nil, utils.WrapAsErrorCodeErr(err, "update serverless import error")
		}
		return &emptypb.Empty{}, nil
	}
	if req.GetAction() == dp.UpdateImportTaskReq_CANCEL {
		userImportModel := s.importModel.User(ctx, userContext.User())
		if err := userImportModel.CancelImport(ctx,
			entity.ID(req.GetProjectId()),
			entity.ID(req.GetClusterId()),
			entity.ID(req.GetImportId()),
			false, /* waitCanceled */
		); err != nil {
			errMsg := "failed to cancel import"
			logger.Error(errMsg, zap.Error(err))
			switch realErr := err.(type) {
			case *import_model.CancelImportInvalidStatusError:
				return nil, errcode.NewCodedError(realErr, errcode.StateCode)
			default:
				return nil, utils.WrapAsErrorCodeErr(err, errMsg)
			}
		}
		logger.Info("cancel import success")
	}
	return &emptypb.Empty{}, nil
}

func (s *Service) UploadLocalFile(ctx context.Context, req *dp.UploadLocalFileReq) (_ *dp.UploadLocalFileResp, gerr error) {
	ctx, span := Tracer.Start(ctx, "openapi.import.UploadLocalFile")
	defer tracing.EndSpan(span, gerr)
	userContext := s.CurrentUser(ctx)
	logger := userContext.Logger()

	if err := checkUploadLocalFile(userContext, req, logger); err != nil {
		return nil, utils.WrapAsErrorCodeErr(err, "check upload local file request parameter error")
	}

	cluster, err := bizutils.GetCluster(ctx, uint64(userContext.OrgID()), req.GetProjectId(), req.GetClusterId())
	if err != nil {
		return nil, err
	}
	if cluster.IsDevTier {
		resp, err := s.UploadLoadFileForServerless(ctx, cluster, req, userContext)
		if err != nil {
			logger.Error("upload serverless import file error", zap.Error(err))
			return nil, utils.WrapAsErrorCodeErr(err, "upload serverless import file error")
		}
		return resp, nil
	}
	region := cluster.RegionName

	var uploader import_model.LocalFileUploader = import_model.NewLocalFileS3Uploader(userContext.User(), region, nil)
	payload := req.GetPayload()
	if payload == nil {
		return nil, errcode.NewInvalidInputErr(errors.New("payload is not specified"))
	}
	rawReader := bytes.NewReader(payload.GetContent())
	actualContentReader, err := gzip.NewReader(rawReader)
	if err != nil {
		logger.Error("construct gzip reader has error", zap.Error(err))
		return nil, utils.WrapAsErrorCodeErr(err, "Prepareing processing payload has errors.")
	}
	//nolint: errcheck
	defer actualContentReader.Close()

	url, err := uploader.PrepareTargetFile(ctx, req.GetLocalFileName(), payload.GetTotalSizeBytes())
	if err != nil {
		logger.Error("prepare target file error", zap.Error(err))
		return nil, utils.WrapAsErrorCodeErr(err, "prepare target file error")
	}
	logger.Info("target file URL", zap.String("target_file_url", url))
	if err := uploader.UploadData(ctx, actualContentReader); err != nil {
		logger.Error("upload payload data into target has errors", zap.Error(err))
		return nil, utils.WrapAsErrorCodeErr(err, "Uploading local file data has errors.")
	}
	uploadInfo, err := uploader.GetPreparedUploadInfo(ctx)
	if err != nil {
		logger.Error("get prepared upload info error", zap.Error(err))
		return nil, utils.WrapAsErrorCodeErr(err, "Generating the upload stub has errors.")
	}
	expireTime := time.Now().Add(uploadStubExpirationDuration)
	newID, err := s.stubManager.CreateStub(ctx, &entity.ImportUploadStub{
		TenantID:         userContext.OrgID(),
		ProjectID:        entity.ID(req.GetProjectId()),
		ClusterID:        entity.ID(req.GetClusterId()),
		LocalFileName:    uploadInfo.OriginalFileName,
		UploadedFileName: uploadInfo.TargetFileName,
		UploadedURL:      uploadInfo.TargetURL,
		UploadedRegion:   region,
		ExpiredTime:      &expireTime,
	})
	if err != nil {
		logger.Error("create stub for the upload has errors", zap.Error(err))
		return nil, utils.WrapAsErrorCodeErr(err, "Generating the upload stub has errors.")
	}
	return &dp.UploadLocalFileResp{
		UploadStubId: uint64(newID),
	}, nil
}

func (s *Service) PreviewImportData(ctx context.Context, req *dp.PreviewImportDataReq) (_ *dp.PreviewImportDataResp, gerr error) {
	var err error
	ctx, span := Tracer.Start(ctx, "openapi.import.PreviewImportData")
	defer tracing.EndSpan(span, gerr)
	userContext := s.CurrentUser(ctx)
	logger := userContext.Logger()

	parsedInfo, err := checkPreviewImportData(userContext, req, logger)
	if err != nil {
		return nil, utils.WrapAsErrorCodeErr(err, "check preview import data request parameter error")
	}

	cluster, bizErr := s.mgmtSvcGrpcClient.GetClusterByID(ctx, req.GetClusterId())
	if bizErr != nil {
		logger.Error("get cluster info error", zap.Error(bizErr))
		return nil, utils.WrapAsErrorCodeErr(bizErr, "get cluster info error")
	}
	if cluster.DeployType == mgmt.DeployType_DEPLOY_TYPE_SERVERLESS {
		resp, err := s.PreviewServerlessImportData(ctx, cluster, req, userContext)
		if err != nil {
			logger.Error("preview serverless import error", zap.Error(err))
			return nil, utils.WrapAsErrorCodeErr(err, "preview serverless import error")
		}
		return resp, nil
	}

	var stub *entity.ImportUploadStub
	if req.GetSpec().GetSource().GetType() == dp.ImportSource_LOCAL_FILE {
		stub, err = s.stubManager.GetStubByID(ctx, entity.ID(parsedInfo.UploadStubID))
		if err != nil {
			return nil, utils.WrapAsErrorCodeErr(err, "Getting upload stub has errors.")
		}
		// Check the expire time should not nil, in order not to panic the program.
		// In normal cases, this should never happen, because each time the upload stub is created, the expiratime has been filled.
		if stub.ExpiredTime == nil {
			logger.Error("cannot find the expired time of the upload stub")
			return nil, utils.WrapAsErrorCodeErr(err, "Getting upload stub has errors.")
		}
		currentTime := time.Now()
		if stub.ExpiredTime.Before(currentTime) {
			logger.Error("The upload stub has expired",
				zap.Time("expire_time", *stub.ExpiredTime),
				zap.Time("current_time", currentTime),
			)
			return nil, errcode.NewCodedError(errors.Errorf("The upload stub has expired at: %v", *stub.ExpiredTime), errcode.StateCode)
		}
	}
	sourceType := req.GetSpec().GetSource().GetType()
	// currently, only support preview local files
	if sourceType != dp.ImportSource_LOCAL_FILE {
		return nil, errcode.NewInvalidInputErr(errors.Errorf("The source type %v is not supported for preview", sourceType))
	}
	targetTable := req.GetSpec().GetTarget().GetTables()[0]
	entityCSVConfig, err := buildCSVConfigInEntity(req.GetSpec().GetSource().GetFormat().GetCsvConfig(), logger)
	if err != nil {
		logger.Error("build CSV config in entity error", zap.Error(err))
		return nil, utils.WrapAsErrorCodeErr(err, "parse CSV config error")
	}
	logger.Info("dump entity CSV config", zap.Any("entity_csv_config", entityCSVConfig))
	var previewer import_model.ImportPreviewer = &import_model.LocalImportPreviewer{
		CSVFormat:  entityCSVConfig,
		SourceURL:  stub.UploadedURL,
		DataFormat: specFormatType2EntityDataFormat(req.GetSpec().GetSource().GetFormat().GetType()),
		PreviewFileMappings: []*import_model.PreviewFileMapping{
			{
				DatabaseName: targetTable.GetDatabaseName(),
				TableName:    targetTable.GetTableName(),
				FileName:     stub.UploadedFileName,
			},
		},
	}
	dbPreviewResults, err := previewer.Preview(ctx, uint(req.GetLimitRowsCount()))
	if err != nil {
		logger.Error("preview table data has errors", zap.Error(err))
		switch concreteErr := errors.Cause(err).(type) {
		case *import_model.ErrCSVSyntax:
			return nil, errcode.NewInvalidInputErr(errors.New("CSV syntax error. Please check your csv configuration matches your csv content."))
		case *import_model.ErrCSVRowOverflow:
			return nil, errcode.NewInvalidInputErr(errors.Errorf("The CSV file is invalid, your %s row data has overflowed.", humanize.Ordinal(concreteErr.RowNum+1)))
		case *import_model.ErrHeaderDataColumnMismatch:
			return nil, errcode.NewInvalidInputErr(errors.Errorf("Invalid CSV file, your row data has %d columns but your headers have %d columns. Please modify the CSV files to keep them matched.",
				concreteErr.DataColumnsCount, concreteErr.HeaderColumnsCount))
		case *import_model.ErrPreviewHasWarning:
			logger.Warn("preview import has warnings", zap.String("warning_message", concreteErr.WarningMessage))
		default:
			return nil, utils.WrapAsErrorCodeErr(err, "Internal error when previewing table data.  Please seek support")
		}
	}

	tabPreviews := []*dp.TablePreview{}
	if len(dbPreviewResults) < 1 {
		return nil, errcode.NewCodedError(errors.New("no preview result for the database"), errcode.StateCode)
	}
	dbPreviewResult := dbPreviewResults[0]
	if len(dbPreviewResult.GetTables()) < 1 {
		return nil, errcode.NewCodedError(errors.New("no preview result for the table"), errcode.StateCode)
	}
	tablePreviewResult := dbPreviewResult.GetTables()[0]

	var resultSchema *dp.TableSchema
	targetTableSchema := func() *dp.TableSchema {
		tableInfoFromTarget, err := s.importModel.GetTableColumns(ctx, &dataflowpb.ImportsGetTableColumnsReq{
			OrgId:     uint64(userContext.OrgID()),
			ProjectId: req.GetProjectId(),
			ClusterId: req.GetClusterId(),
			Db:        tablePreviewResult.GetSchemaName(),
			Table:     tablePreviewResult.GetTableName(),
		})
		if err != nil {
			logger.Error("get target table columns has errors", zap.Error(err))
			return nil
		}
		if len(tableInfoFromTarget.GetRowHeaders()) <= 0 ||
			len(tableInfoFromTarget.GetRowHeaders()) != len(tableInfoFromTarget.GetRowDataTypes()) {
			return nil
		}
		targetTableSchema, err := previewResult2TableSchema(tableInfoFromTarget, logger)
		if err != nil {
			logger.Error("construct table schema result has errors for target table schema", zap.Error(err))
			return nil
		}
		return targetTableSchema
	}()
	if targetTableSchema != nil {
		resultSchema = targetTableSchema
	} else {
		previewTableSchema, err := previewResult2TableSchema(tablePreviewResult, logger)
		if err != nil {
			return nil, utils.WrapAsErrorCodeErr(err, "construct table schema result has errors.")
		}
		resultSchema = previewTableSchema
	}
	previewTableData, err := previewResult2TableData(tablePreviewResult, logger)
	if err != nil {
		return nil, utils.WrapAsErrorCodeErr(err, "construct table data result has errors.")
	}
	tabPreviews = append(tabPreviews, &dp.TablePreview{
		DatabaseName:  tablePreviewResult.GetSchemaName(),
		TableName:     tablePreviewResult.GetTableName(),
		SchemaPreview: resultSchema,
		DataPreview:   previewTableData,
	})

	return &dp.PreviewImportDataResp{
		TablePreviews: tabPreviews,
	}, nil
}