diff --git a/deploy/example/pvc-volume-clone.yaml b/deploy/example/pvc-volume-clone.yaml
new file mode 100644
index 000000000..5b1575f34
--- /dev/null
+++ b/deploy/example/pvc-volume-clone.yaml
@@ -0,0 +1,16 @@
+---
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: pvc-nfs-clone
+  namespace: default
+spec:
+  accessModes:
+    - ReadWriteMany
+  resources:
+    requests:
+      storage: 10Gi
+  storageClassName: nfs-csi
+  dataSource:
+    kind: PersistentVolumeClaim
+    name: pvc-nfs-dynamic
diff --git a/pkg/nfs/controllerserver.go b/pkg/nfs/controllerserver.go
index cf8f44d42..65a74a78c 100644
--- a/pkg/nfs/controllerserver.go
+++ b/pkg/nfs/controllerserver.go
@@ -19,6 +19,7 @@ package nfs
 import (
 	"fmt"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"regexp"
 	"strconv"
@@ -143,12 +144,19 @@ func (cs *ControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 		}
 	}
 
+	if req.GetVolumeContentSource() != nil {
+		if err := cs.copyVolume(ctx, req, nfsVol); err != nil {
+			return nil, err
+		}
+	}
+
 	setKeyValueInMap(parameters, paramSubDir, nfsVol.subDir)
 	return &csi.CreateVolumeResponse{
 		Volume: &csi.Volume{
 			VolumeId:      nfsVol.id,
 			CapacityBytes: 0, // by setting it to zero, Provisioner will use PVC requested size as PV size
 			VolumeContext: parameters,
+			ContentSource: req.GetVolumeContentSource(),
 		},
 	}, nil
 }
@@ -307,6 +315,58 @@ func (cs *ControllerServer) internalUnmount(ctx context.Context, vol *nfsVolume)
 	return err
 }
 
+func (cs *ControllerServer) copyFromVolume(ctx context.Context, req *csi.CreateVolumeRequest, dstVol *nfsVolume) error {
+	srcVol, err := getNfsVolFromID(req.GetVolumeContentSource().GetVolume().GetVolumeId())
+	if err != nil {
+		return status.Error(codes.NotFound, err.Error())
+	}
+	// Note that the source path must include trailing '/.', can't use 'filepath.Join()' as it performs path cleaning
+	srcPath := fmt.Sprintf("%v/.", getInternalVolumePath(cs.Driver.workingMountDir, srcVol))
+	dstPath := getInternalVolumePath(cs.Driver.workingMountDir, dstVol)
+	klog.V(2).Infof("copy volume from volume %v -> %v", srcPath, dstPath)
+
+	var volCap *csi.VolumeCapability
+	if len(req.GetVolumeCapabilities()) > 0 {
+		volCap = req.GetVolumeCapabilities()[0]
+	}
+	if err = cs.internalMount(ctx, srcVol, nil, volCap); err != nil {
+		return status.Errorf(codes.Internal, "failed to mount src nfs server: %v", err.Error())
+	}
+	defer func() {
+		if err = cs.internalUnmount(ctx, srcVol); err != nil {
+			klog.Warningf("failed to unmount nfs server: %v", err.Error())
+		}
+	}()
+	if err = cs.internalMount(ctx, dstVol, nil, volCap); err != nil {
+		return status.Errorf(codes.Internal, "failed to mount dst nfs server: %v", err.Error())
+	}
+	defer func() {
+		if err = cs.internalUnmount(ctx, dstVol); err != nil {
+			klog.Warningf("failed to unmount dst nfs server: %v", err.Error())
+		}
+	}()
+
+	// recursive 'cp' with '-a' to handle symlinks
+	out, err := exec.Command("cp", "-a", srcPath, dstPath).CombinedOutput()
+	if err != nil {
+		return status.Error(codes.Internal, fmt.Sprintf("%v: %v", err, string(out)))
+	}
+	klog.V(2).Infof("copied %s -> %s", srcPath, dstPath)
+	return nil
+}
+
+func (cs *ControllerServer) copyVolume(ctx context.Context, req *csi.CreateVolumeRequest, vol *nfsVolume) error {
+	vs := req.VolumeContentSource
+	switch vs.Type.(type) {
+	case *csi.VolumeContentSource_Snapshot:
+		return status.Error(codes.Unimplemented, "Currently only volume copy from another volume is supported")
+	case *csi.VolumeContentSource_Volume:
+		return cs.copyFromVolume(ctx, req, vol)
+	default:
+		return status.Errorf(codes.InvalidArgument, "%v not a proper volume source", vs)
+	}
+}
+
 // newNFSVolume Convert VolumeCreate parameters to an nfsVolume
 func newNFSVolume(name string, size int64, params map[string]string) (*nfsVolume, error) {
 	var server, baseDir, subDir string
diff --git a/pkg/nfs/controllerserver_test.go b/pkg/nfs/controllerserver_test.go
index 9f3c8c871..692539795 100644
--- a/pkg/nfs/controllerserver_test.go
+++ b/pkg/nfs/controllerserver_test.go
@@ -325,6 +325,13 @@ func TestControllerGetCapabilities(t *testing.T) {
 							},
 						},
 					},
+					{
+						Type: &csi.ControllerServiceCapability_Rpc{
+							Rpc: &csi.ControllerServiceCapability_RPC{
+								Type: csi.ControllerServiceCapability_RPC_CLONE_VOLUME,
+							},
+						},
+					},
 				},
 			},
 			expectedErr: nil,
diff --git a/pkg/nfs/nfs.go b/pkg/nfs/nfs.go
index 039aea070..921a0fd14 100644
--- a/pkg/nfs/nfs.go
+++ b/pkg/nfs/nfs.go
@@ -84,6 +84,7 @@ func NewDriver(options *DriverOptions) *Driver {
 	n.AddControllerServiceCapabilities([]csi.ControllerServiceCapability_RPC_Type{
 		csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
 		csi.ControllerServiceCapability_RPC_SINGLE_NODE_MULTI_WRITER,
+		csi.ControllerServiceCapability_RPC_CLONE_VOLUME,
 	})
 
 	n.AddNodeServiceCapabilities([]csi.NodeServiceCapability_RPC_Type{
diff --git a/test/external-e2e/testdriver.yaml b/test/external-e2e/testdriver.yaml
index 406d4c9dd..b63de8d90 100644
--- a/test/external-e2e/testdriver.yaml
+++ b/test/external-e2e/testdriver.yaml
@@ -12,6 +12,7 @@ DriverInfo:
     multipods: true
     RWX: true
     fsGroup: true
+    pvcDataSource: true
 InlineVolumes:
 - Attributes:
     server: nfs-server.default.svc.cluster.local