diff --git a/components/cli/cli/command/stack/deploy.go b/components/cli/cli/command/stack/deploy.go
index d18a43484d..8e14b70d2f 100644
--- a/components/cli/cli/command/stack/deploy.go
+++ b/components/cli/cli/command/stack/deploy.go
@@ -7,6 +7,7 @@ import (
 	"github.com/docker/cli/cli/command"
 	"github.com/docker/cli/cli/compose/convert"
 	"github.com/docker/docker/api/types/swarm"
+	"github.com/docker/docker/api/types/versions"
 	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
 	"golang.org/x/net/context"
@@ -14,12 +15,16 @@ import (
 
 const (
 	defaultNetworkDriver = "overlay"
+	resolveImageAlways   = "always"
+	resolveImageChanged  = "changed"
+	resolveImageNever    = "never"
 )
 
 type deployOptions struct {
 	bundlefile       string
 	composefile      string
 	namespace        string
+	resolveImage     string
 	sendRegistryAuth bool
 	prune            bool
 }
@@ -44,12 +49,19 @@ func newDeployCommand(dockerCli command.Cli) *cobra.Command {
 	addRegistryAuthFlag(&opts.sendRegistryAuth, flags)
 	flags.BoolVar(&opts.prune, "prune", false, "Prune services that are no longer referenced")
 	flags.SetAnnotation("prune", "version", []string{"1.27"})
+	flags.StringVar(&opts.resolveImage, "resolve-image", resolveImageAlways,
+		`Query the registry to resolve image digest and supported platforms ("`+resolveImageAlways+`"|"`+resolveImageChanged+`"|"`+resolveImageNever+`")`)
+	flags.SetAnnotation("resolve-image", "version", []string{"1.30"})
 	return cmd
 }
 
 func runDeploy(dockerCli command.Cli, opts deployOptions) error {
 	ctx := context.Background()
 
+	if err := validateResolveImageFlag(dockerCli, &opts); err != nil {
+		return err
+	}
+
 	switch {
 	case opts.bundlefile == "" && opts.composefile == "":
 		return errors.Errorf("Please specify either a bundle file (with --bundle-file) or a Compose file (with --compose-file).")
@@ -62,6 +74,20 @@ func runDeploy(dockerCli command.Cli, opts deployOptions) error {
 	}
 }
 
+// validateResolveImageFlag validates the opts.resolveImage command line option
+// and also turns image resolution off if the version is older than 1.30
+func validateResolveImageFlag(dockerCli command.Cli, opts *deployOptions) error {
+	if opts.resolveImage != resolveImageAlways && opts.resolveImage != resolveImageChanged && opts.resolveImage != resolveImageNever {
+		return errors.Errorf("Invalid option %s for flag --resolve-image", opts.resolveImage)
+	}
+	// client side image resolution should not be done when the supported
+	// server version is older than 1.30
+	if versions.LessThan(dockerCli.Client().ClientVersion(), "1.30") {
+		opts.resolveImage = resolveImageNever
+	}
+	return nil
+}
+
 // checkDaemonIsSwarmManager does an Info API call to verify that the daemon is
 // a swarm manager. This is necessary because we must create networks before we
 // create services, but the API call for creating a network does not return a
diff --git a/components/cli/cli/command/stack/deploy_bundlefile.go b/components/cli/cli/command/stack/deploy_bundlefile.go
index 2f2a9aa042..1074210e97 100644
--- a/components/cli/cli/command/stack/deploy_bundlefile.go
+++ b/components/cli/cli/command/stack/deploy_bundlefile.go
@@ -87,5 +87,5 @@ func deployBundle(ctx context.Context, dockerCli command.Cli, opts deployOptions
 	if err := createNetworks(ctx, dockerCli, namespace, networks); err != nil {
 		return err
 	}
-	return deployServices(ctx, dockerCli, services, namespace, opts.sendRegistryAuth)
+	return deployServices(ctx, dockerCli, services, namespace, opts.sendRegistryAuth, opts.resolveImage)
 }
diff --git a/components/cli/cli/command/stack/deploy_composefile.go b/components/cli/cli/command/stack/deploy_composefile.go
index 642867cba9..1cafca58e7 100644
--- a/components/cli/cli/command/stack/deploy_composefile.go
+++ b/components/cli/cli/command/stack/deploy_composefile.go
@@ -92,7 +92,7 @@ func deployCompose(ctx context.Context, dockerCli command.Cli, opts deployOption
 	if err != nil {
 		return err
 	}
-	return deployServices(ctx, dockerCli, services, namespace, opts.sendRegistryAuth)
+	return deployServices(ctx, dockerCli, services, namespace, opts.sendRegistryAuth, opts.resolveImage)
 }
 
 func getServicesDeclaredNetworks(serviceConfigs []composetypes.ServiceConfig) map[string]struct{} {
@@ -282,6 +282,7 @@ func deployServices(
 	services map[string]swarm.ServiceSpec,
 	namespace convert.Namespace,
 	sendAuth bool,
+	resolveImage string,
 ) error {
 	apiClient := dockerCli.Client()
 	out := dockerCli.Out()
@@ -300,9 +301,9 @@ func deployServices(
 		name := namespace.Scope(internalName)
 
 		encodedAuth := ""
+		image := serviceSpec.TaskTemplate.ContainerSpec.Image
 		if sendAuth {
 			// Retrieve encoded auth token from the image reference
-			image := serviceSpec.TaskTemplate.ContainerSpec.Image
 			encodedAuth, err = command.RetrieveAuthTokenFromImage(ctx, dockerCli, image)
 			if err != nil {
 				return err
@@ -312,10 +313,12 @@ func deployServices(
 		if service, exists := existingServiceMap[name]; exists {
 			fmt.Fprintf(out, "Updating service %s (id: %s)\n", name, service.ID)
 
-			updateOpts := types.ServiceUpdateOptions{}
-			if sendAuth {
-				updateOpts.EncodedRegistryAuth = encodedAuth
+			updateOpts := types.ServiceUpdateOptions{EncodedRegistryAuth: encodedAuth}
+
+			if resolveImage == resolveImageAlways || (resolveImage == resolveImageChanged && image != service.Spec.Labels[convert.LabelImage]) {
+				updateOpts.QueryRegistry = true
 			}
+
 			response, err := apiClient.ServiceUpdate(
 				ctx,
 				service.ID,
@@ -333,10 +336,13 @@ func deployServices(
 		} else {
 			fmt.Fprintf(out, "Creating service %s\n", name)
 
-			createOpts := types.ServiceCreateOptions{}
-			if sendAuth {
-				createOpts.EncodedRegistryAuth = encodedAuth
+			createOpts := types.ServiceCreateOptions{EncodedRegistryAuth: encodedAuth}
+
+			// query registry if flag disabling it was not set
+			if resolveImage == resolveImageAlways || resolveImage == resolveImageChanged {
+				createOpts.QueryRegistry = true
 			}
+
 			if _, err := apiClient.ServiceCreate(ctx, serviceSpec, createOpts); err != nil {
 				return err
 			}
diff --git a/components/cli/cli/compose/convert/service.go b/components/cli/cli/compose/convert/service.go
index 233a376293..2280d57c79 100644
--- a/components/cli/cli/compose/convert/service.go
+++ b/components/cli/cli/compose/convert/service.go
@@ -18,7 +18,11 @@ import (
 	"github.com/pkg/errors"
 )
 
-const defaultNetwork = "default"
+const (
+	defaultNetwork = "default"
+	// LabelImage is the label used to store image name provided in the compose file
+	LabelImage = "com.docker.stack.image"
+)
 
 // Services from compose-file types to engine API types
 func Services(
@@ -158,6 +162,9 @@ func convertService(
 		UpdateConfig: convertUpdateConfig(service.Deploy.UpdateConfig),
 	}
 
+	// add an image label to serviceSpec
+	serviceSpec.Labels[LabelImage] = service.Image
+
 	// ServiceSpec.Networks is deprecated and should not have been used by
 	// this package. It is possible to update TaskTemplate.Networks, but it
 	// is not possible to update ServiceSpec.Networks. Unfortunately, we
diff --git a/components/engine/hack/integration-cli-on-swarm/host/dockercmd.go b/components/engine/hack/integration-cli-on-swarm/host/dockercmd.go
index 10ea0ecc24..c539876213 100644
--- a/components/engine/hack/integration-cli-on-swarm/host/dockercmd.go
+++ b/components/engine/hack/integration-cli-on-swarm/host/dockercmd.go
@@ -37,6 +37,7 @@ func deployStack(unusedCli *client.Client, stackName, composeFilePath string) er
 		{"docker", "stack", "deploy",
 			"--compose-file", composeFilePath,
 			"--with-registry-auth",
+			"--resolve-image", "never",
 			stackName},
 	})
 }
diff --git a/components/engine/integration-cli/docker_cli_stack_test.go b/components/engine/integration-cli/docker_cli_stack_test.go
index 91fe4d75c0..51b98ff187 100644
--- a/components/engine/integration-cli/docker_cli_stack_test.go
+++ b/components/engine/integration-cli/docker_cli_stack_test.go
@@ -62,6 +62,7 @@ func (s *DockerSwarmSuite) TestStackDeployComposeFile(c *check.C) {
 	testStackName := "testdeploy"
 	stackArgs := []string{
 		"stack", "deploy",
+		"--resolve-image", "never",
 		"--compose-file", "fixtures/deploy/default.yaml",
 		testStackName,
 	}
@@ -88,6 +89,7 @@ func (s *DockerSwarmSuite) TestStackDeployWithSecretsTwice(c *check.C) {
 	testStackName := "testdeploy"
 	stackArgs := []string{
 		"stack", "deploy",
+		"--resolve-image", "never",
 		"--compose-file", "fixtures/deploy/secrets.yaml",
 		testStackName,
 	}
@@ -120,6 +122,7 @@ func (s *DockerSwarmSuite) TestStackRemove(c *check.C) {
 	stackName := "testdeploy"
 	stackArgs := []string{
 		"stack", "deploy",
+		"--resolve-image", "never",
 		"--compose-file", "fixtures/deploy/remove.yaml",
 		stackName,
 	}
@@ -179,6 +182,7 @@ func (s *DockerSwarmSuite) TestStackDeployWithDAB(c *check.C) {
 	// deploy
 	stackArgs := []string{
 		"stack", "deploy",
+		"--resolve-image", "never",
 		"--bundle-file", testDABFileName,
 		testStackName,
 	}