diff --git a/backupstore.go b/backupstore.go
index 9e79e8c8e..321727487 100644
--- a/backupstore.go
+++ b/backupstore.go
@@ -23,6 +23,7 @@ type Volume struct {
 	CompressionMethod    string `json:",string"`
 	StorageClassName     string `json:",string"`
 	DataEngine           string `json:",string"`
+	BackupTimes          int64  `json:",string"`
 }
 
 type Snapshot struct {
@@ -74,6 +75,9 @@ func addVolume(driver BackupStoreDriver, volume *Volume) error {
 		return fmt.Errorf("invalid volume name %v", volume.Name)
 	}
 
+	// first time create backup volume means first time create backup on this backupstore
+	volume.BackupTimes = 0
+
 	if err := saveVolume(driver, volume); err != nil {
 		log.WithError(err).Errorf("Failed to add volume %v", volume.Name)
 		return err
diff --git a/deltablock.go b/deltablock.go
index 73851e53b..185a2a919 100644
--- a/deltablock.go
+++ b/deltablock.go
@@ -175,7 +175,7 @@ func CreateDeltaBlockBackup(backupName string, config *DeltaBackupConfig) (isInc
 	}
 
 	backupRequest := &backupRequest{}
-	if volume.LastBackupName != "" {
+	if volume.LastBackupName != "" && !isFullBackup(config) {
 		lastBackupName := volume.LastBackupName
 		var backup, err = loadBackup(bsDriver, lastBackupName, volume.Name)
 		if err != nil {
@@ -208,6 +208,9 @@ func CreateDeltaBlockBackup(backupName string, config *DeltaBackupConfig) (isInc
 		}
 	}
 
+	logrus.Infof("[DEBUG] isFullBackup(config): %v", isFullBackup(config))
+	logrus.Infof("[DEBUG] backupRequest.lastBackup: %v", backupRequest.lastBackup)
+
 	log.WithFields(logrus.Fields{
 		LogFieldReason:       LogReasonStart,
 		LogFieldObject:       LogObjectSnapshot,
@@ -368,7 +371,7 @@ func backupBlock(bsDriver BackupStoreDriver, config *DeltaBackupConfig,
 	}()
 
 	blkFile := getBlockFilePath(volume.Name, checksum)
-	if bsDriver.FileExists(blkFile) {
+	if bsDriver.FileExists(blkFile) && !isFullBackup(config) {
 		log.Debugf("Found existing block matching at %v", blkFile)
 		return nil
 	}
@@ -554,6 +557,7 @@ func performBackup(bsDriver BackupStoreDriver, config *DeltaBackupConfig, delta
 	volume.CompressionMethod = config.Volume.CompressionMethod
 	volume.StorageClassName = config.Volume.StorageClassName
 	volume.DataEngine = config.Volume.DataEngine
+	volume.BackupTimes = volume.BackupTimes + 1
 
 	if err := saveVolume(bsDriver, volume); err != nil {
 		return progress.progress, "", err
@@ -1303,3 +1307,12 @@ func getBlockNamesForVolume(driver BackupStoreDriver, volumeName string) ([]stri
 
 	return util.ExtractNames(names, "", BLK_SUFFIX), nil
 }
+
+func isFullBackup(config *DeltaBackupConfig) bool {
+	if config.Labels != nil {
+		if backupMode, exist := config.Labels[types.GetLonghornLabelKey(types.LonghornBackupOptionBackupMode)]; exist {
+			return backupMode == types.LonghornBackupModeFull
+		}
+	}
+	return false
+}
diff --git a/types/types.go b/types/types.go
index bc626cc5f..b71880a02 100644
--- a/types/types.go
+++ b/types/types.go
@@ -1,5 +1,7 @@
 package types
 
+import "fmt"
+
 type ProgressState string
 
 const (
@@ -29,6 +31,16 @@ const (
 	NOProxy    = "NO_PROXY"
 
 	VirtualHostedStyle = "VIRTUAL_HOSTED_STYLE"
+
+	LonghornLabelKeyPrefix = "longhorn.io"
+)
+
+const (
+	LonghornBackupOptionBackupMode = "backup-mode"
+	LonghornBackupModeFull         = "full"
+	LonghornBackupModeIncremental  = "incremental"
+
+	LonghornBackupOptionFullBackupInterval = "full-backup-interval"
 )
 
 type Mapping struct {
@@ -65,3 +77,7 @@ const (
 const (
 	ErrorMsgRestoreCancelled = "backup restoration is cancelled"
 )
+
+func GetLonghornLabelKey(name string) string {
+	return fmt.Sprintf("%s/%s", LonghornLabelKeyPrefix, name)
+}