Function sync

cli/cmd/sync.go:177–572

(cmd *cobra.Command, args []string)

}

func sync(cmd *cobra.Command, args []string) error {
	cqDir, err := cmd.Flags().GetString("cq-dir")
	if err != nil {
		return err
	}

	noMigrate, err := cmd.Flags().GetBool("no-migrate")
	if err != nil {
		return err
	}

	licenseFile, err := cmd.Flags().GetString("license")
	if err != nil {
		return err
	}

	shard, err := parseShard(cmd)
	if err != nil {
		return err
	}

	cqColumnsNotNull, err := cmd.Flags().GetBool("cq-columns-not-null")
	if err != nil {
		return err
	}

	// in the cloud sync environment, we pass only the relevant environment variables to the plugin
	isolatePluginEnvironment := env.IsCloud()

	ctx := cmd.Context()
	log.Info().Strs("args", args).Msg("Loading spec(s)")
	fmt.Printf("Loading spec(s) from %s\n", strings.Join(args, ", "))
	// Validate after injection so a source-only spec isn't rejected before the
	// platform destination is added.
	specReader, err := specs.NewSpecReaderWithoutValidation(args)
	if err != nil {
		return fmt.Errorf("failed to load spec(s) from %s. Error: %w", strings.Join(args, ", "), err)
	}

	sources := specReader.Sources
	destinations := specReader.Destinations
	transformers := specReader.Transformers

	tableMetricsLocation, err := cmd.Flags().GetString("tables-metrics-location")
	if err != nil {
		return err
	}

	summaryLocation, err := cmd.Flags().GetString("summary-location")
	if err != nil {
		return err
	}

	var otelReceiver *otel.OtelReceiver

	// dlToken/teamName authenticate plugin download + usage against cloud
	// (headless cqpd_ token, else cloud login). See cqplatform.DownloadAuth.
	dlToken, teamName, err := cqplatform.DownloadAuth(ctx, log.Logger, sources, destinations, transformers)