diff --git a/cmd/controller/controller.go b/cmd/controller/controller.go
index 847f58a..0569284 100644
--- a/cmd/controller/controller.go
+++ b/cmd/controller/controller.go
@@ -80,7 +80,7 @@ func (c *ZVController) syncZV(zv *apis.ZFSVolume) error {
 	var err error
 	// ZFS Volume should be deleted. Check if deletion timestamp is set
 	if c.isDeletionCandidate(zv) {
-		err = zvol.DestroyZvol(zv)
+		err = zvol.DestroyVolume(zv)
 		if err == nil {
 			zvol.RemoveZvolFinalizer(zv)
 		}
@@ -91,7 +91,7 @@ func (c *ZVController) syncZV(zv *apis.ZFSVolume) error {
 		if zv.Finalizers != nil {
 			err = zvol.SetZvolProp(zv)
 		} else {
-			err = zvol.CreateZvol(zv)
+			err = zvol.CreateVolume(zv)
 			if err == nil {
 				err = zvol.UpdateZvolInfo(zv)
 			}
diff --git a/deploy/sample/fio.yaml b/deploy/sample/fio.yaml
index 5ef5c60..cae01da 100644
--- a/deploy/sample/fio.yaml
+++ b/deploy/sample/fio.yaml
@@ -4,13 +4,14 @@ metadata:
   name: openebs-zfspv
 allowVolumeExpansion: true
 parameters:
-  blocksize: "4k"
+  recordsize: "4k"
   compression: "on"
   dedup: "on"
   thinprovision: "yes"
   #encryption: "on"
   #keyformat: "raw"
   #keylocation: "file:///home/pawan/key"
+  fstype: "zfs"
   poolname: "zfspv-pool"
 provisioner: zfs.csi.openebs.io
 allowedTopologies:
@@ -32,22 +33,33 @@ spec:
     requests:
       storage: 4Gi
 ---
-apiVersion: v1
-kind: Pod
+apiVersion: apps/v1
+kind: Deployment
 metadata:
   name: fio
+  labels:
+    name: fio
 spec:
-  restartPolicy: Never
-  containers:
-  - name: perfrunner
-    image: openebs/tests-fio
-    command: ["/bin/bash"]
-    args: ["-c", "while true ;do sleep 50; done"]
-    volumeMounts:
-       - mountPath: /datadir
-         name: fio-vol
-    tty: true
-  volumes:
-  - name: fio-vol
-    persistentVolumeClaim:
-      claimName: csi-zfspv
+  replicas: 1
+  selector:
+    matchLabels:
+      name: fio
+  template:
+    metadata:
+      labels:
+        name: fio
+    spec:
+      containers:
+        - resources:
+          name: perfrunner
+          image: openebs/tests-fio
+          imagePullPolicy: IfNotPresent
+          command: ["/bin/bash"]
+          args: ["-c", "while true ;do sleep 50; done"]
+          volumeMounts:
+            - mountPath: /datadir
+              name: fio-vol
+      volumes:
+        - name: fio-vol
+          persistentVolumeClaim:
+            claimName: csi-zfspv
diff --git a/deploy/sample/mongo-statefulset.yaml b/deploy/sample/mongo-statefulset.yaml
index 6e9c70e..a141c1f 100644
--- a/deploy/sample/mongo-statefulset.yaml
+++ b/deploy/sample/mongo-statefulset.yaml
@@ -7,9 +7,9 @@ kind: StorageClass
 metadata:
   name: mongo-pv-az
 parameters:
-  blocksize: "4k"
+  volblocksize: "4k"
   poolname: "zfspv-pool"
-  fsType: "xfs"
+  fstype: "xfs"
 provisioner: zfs.csi.openebs.io
 ---
 # Headless service for stable DNS entries of StatefulSet members.
diff --git a/deploy/sample/percona.yaml b/deploy/sample/percona.yaml
index 04609cc..0f2ca4a 100644
--- a/deploy/sample/percona.yaml
+++ b/deploy/sample/percona.yaml
@@ -4,7 +4,7 @@ metadata:
   name: openebs-zfspv
 allowVolumeExpansion: true
 parameters:
-  blocksize: "4k"
+  volblocksize: "4k"
   compression: "on"
   dedup: "on"
   thinprovision: "yes"
diff --git a/deploy/sample/zfspvcr.yaml b/deploy/sample/zfspvcr.yaml
index d877d65..6858343 100644
--- a/deploy/sample/zfspvcr.yaml
+++ b/deploy/sample/zfspvcr.yaml
@@ -4,7 +4,6 @@ metadata:
   name: pvc-37b07ad6-db68-11e9-bbb6-000c296e38d9
   namespace: openebs
 spec:
-  blocksize: 4k
   capacity: "4294967296"
   compression: "off"
   dedup: "off"
@@ -13,5 +12,6 @@ spec:
   keylocation: ""
   ownerNodeID: zfspv-node1
   poolName: zfspv-pool
+  recordsize: 8k
   thinProvision: "off"
-
+  volumeType: DATASET
diff --git a/deploy/zfs-operator.yaml b/deploy/zfs-operator.yaml
index 4da936e..2c72a39 100644
--- a/deploy/zfs-operator.yaml
+++ b/deploy/zfs-operator.yaml
@@ -20,7 +20,7 @@ spec:
     singular: zfsvolume
     kind: ZFSVolume
     shortNames:
-    - zvol
+    - zfsvol
     - zv
   additionalPrinterColumns:
   - JSONPath: .spec.poolName
@@ -35,6 +35,18 @@ spec:
     name: Size
     description: Size of the volume
     type: string
+  - JSONPath: .spec.volblocksize
+    name: volblocksize
+    description: volblocksize for the created zvol
+    type: string
+  - JSONPath: .spec.recordsize
+    name: recordsize
+    description: recordsize for the created zfs dataset
+    type: string
+  - JSONPath: .spec.fsType
+    name: Filesystem
+    description: filesystem created on the volume
+    type: string
 
 ---
 ##############################################
diff --git a/pkg/apis/openebs.io/core/v1alpha1/zfsvolume.go b/pkg/apis/openebs.io/core/v1alpha1/zfsvolume.go
index fb3c806..a94c979 100644
--- a/pkg/apis/openebs.io/core/v1alpha1/zfsvolume.go
+++ b/pkg/apis/openebs.io/core/v1alpha1/zfsvolume.go
@@ -82,31 +82,42 @@ type VolumeInfo struct {
 	// Capacity of the volume
 	Capacity string `json:"capacity"`
 
-	// BlockSize specifies the blocksize
-	// which we should use to create the zvol
-	BlockSize string `json:"blocksize"`
+	// RecordSize specifies the record size
+	// for the zfs dataset
+	RecordSize string `json:"recordsize,omitempty"`
+
+	// VolBlockSize specifies the block size for the zvol
+	VolBlockSize string `json:"volblocksize,omitempty"`
 
 	// Compression specifies if the it should
 	// enabled on the zvol
-	Compression string `json:"compression"`
+	Compression string `json:"compression,omitempty"`
 
 	// Dedup specifies the deduplication
 	// should be enabled on the zvol
-	Dedup string `json:"dedup"`
+	Dedup string `json:"dedup,omitempty"`
 
 	// Encryption specifies the encryption
 	// should be enabled on the zvol
-	Encryption string `json:"encryption"`
+	Encryption string `json:"encryption,omitempty"`
 
 	// KeyLocation is the location of key
 	// for the encryption
-	KeyLocation string `json:"keylocation"`
+	KeyLocation string `json:"keylocation,omitempty"`
 
 	// KeyFormat specifies format of the
 	// encryption key
-	KeyFormat string `json:"keyformat"`
+	KeyFormat string `json:"keyformat,omitempty"`
 
 	// Thinprovision specifies if we should
 	// thin provisioned the volume or not
-	ThinProvision string `json:"thinProvision"`
+	ThinProvision string `json:"thinProvision,omitempty"`
+
+	// VolumeType specifies whether the volume is
+	// zvol or a dataset
+	VolumeType string `json:"volumeType"`
+
+	// FsType specifies filesystem type for the
+	// zfs volume/dataset
+	FsType string `json:"fsType,omitempty"`
 }
diff --git a/pkg/builder/build.go b/pkg/builder/build.go
index 5533069..24ff03a 100644
--- a/pkg/builder/build.go
+++ b/pkg/builder/build.go
@@ -142,13 +142,27 @@ func (b *Builder) WithOwnerNode(host string) *Builder {
 	return b
 }
 
-// WithBlockSize sets blocksize of ZFSVolume
-func (b *Builder) WithBlockSize(blockSize string) *Builder {
-	bs := "4k"
-	if len(blockSize) > 0 {
-		bs = blockSize
-	}
-	b.volume.Object.Spec.BlockSize = bs
+// WithRecordSize sets the recordsize of ZFSVolume
+func (b *Builder) WithRecordSize(rs string) *Builder {
+	b.volume.Object.Spec.RecordSize = rs
+	return b
+}
+
+// WithVolBlockSize sets the volblocksize of ZFSVolume
+func (b *Builder) WithVolBlockSize(bs string) *Builder {
+	b.volume.Object.Spec.VolBlockSize = bs
+	return b
+}
+
+// WithVolumeType sets if ZFSVolume needs to be thin provisioned
+func (b *Builder) WithVolumeType(vtype string) *Builder {
+	b.volume.Object.Spec.VolumeType = vtype
+	return b
+}
+
+// WithFsType sets filesystem for the ZFSVolume
+func (b *Builder) WithFsType(fstype string) *Builder {
+	b.volume.Object.Spec.FsType = fstype
 	return b
 }
 
diff --git a/pkg/driver/agent.go b/pkg/driver/agent.go
index 6d35d48..5fc6ace 100644
--- a/pkg/driver/agent.go
+++ b/pkg/driver/agent.go
@@ -22,7 +22,7 @@ import (
 	ctrl "github.com/openebs/zfs-localpv/cmd/controller"
 	apis "github.com/openebs/zfs-localpv/pkg/apis/openebs.io/core/v1alpha1"
 	"github.com/openebs/zfs-localpv/pkg/builder"
-	zvol "github.com/openebs/zfs-localpv/pkg/zfs"
+	zfs "github.com/openebs/zfs-localpv/pkg/zfs"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -65,7 +65,7 @@ func GetVolAndMountInfo(
 
 	getOptions := metav1.GetOptions{}
 	vol, err := builder.NewKubeclient().
-		WithNamespace(zvol.OpenEBSNamespace).
+		WithNamespace(zfs.OpenEBSNamespace).
 		Get(req.GetVolumeId(), getOptions)
 
 	if err != nil {
@@ -96,8 +96,8 @@ func (ns *node) NodePublishVolume(
 	if err != nil {
 		goto PublishVolumeResponse
 	}
-	// Create the zfs volume and attempt mount operation on the requested path
-	if err = zvol.CreateAndMountZvol(vol, mountInfo); err != nil {
+	// attempt mount operation on the requested path
+	if err = zfs.MountVolume(vol, mountInfo); err != nil {
 		goto PublishVolumeResponse
 	}
 
@@ -120,6 +120,7 @@ func (ns *node) NodeUnpublishVolume(
 	var (
 		err           error
 		vol           *apis.ZFSVolume
+		devpath       string
 		currentMounts []string
 	)
 
@@ -130,22 +131,19 @@ func (ns *node) NodeUnpublishVolume(
 	targetPath := req.GetTargetPath()
 	volumeID := req.GetVolumeId()
 
-	getOptions := metav1.GetOptions{}
-	vol, err = builder.NewKubeclient().
-		WithNamespace(zvol.OpenEBSNamespace).
-		Get(volumeID, getOptions)
-
-	if err != nil {
+	if vol, err = zfs.GetZFSVolume(volumeID); err != nil {
 		return nil, err
 	}
 
-	zfsvolume := vol.Spec.PoolName + "/" + vol.Name
-	devpath := zvol.ZFS_DEVPATH + zfsvolume
-	currentMounts, err = zvol.GetMounts(devpath)
+	if devpath, err = zfs.GetVolumeDevPath(vol); err != nil {
+		goto NodeUnpublishResponse
+	}
+
+	currentMounts, err = zfs.GetMounts(devpath)
 	if err != nil {
 		return nil, err
 	} else if len(currentMounts) == 0 {
-		goto NodeUnpublishResponse
+		return nil, status.Error(codes.Internal, "umount request for not mounted volume")
 	} else if len(currentMounts) == 1 {
 		if currentMounts[0] != targetPath {
 			return nil, status.Error(codes.Internal, "device not mounted at right path")
@@ -158,15 +156,14 @@ func (ns *node) NodeUnpublishVolume(
 		return nil, status.Error(codes.Internal, "device not mounted at rightpath")
 	}
 
-	if vol, err = zvol.GetZFSVolume(volumeID); (err != nil) || (vol == nil) {
-		goto NodeUnpublishResponse
-	}
-
-	if err = zvol.UmountVolume(vol, req.GetTargetPath()); err != nil {
+	if err = zfs.UmountVolume(vol, req.GetTargetPath()); err != nil {
 		goto NodeUnpublishResponse
 	}
 
 NodeUnpublishResponse:
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
 	logrus.Infof("hostpath: volume %s path: %s has been unmounted.",
 		volumeID, targetPath)
 
@@ -181,7 +178,7 @@ func (ns *node) NodeGetInfo(
 	req *csi.NodeGetInfoRequest,
 ) (*csi.NodeGetInfoResponse, error) {
 
-	topology := map[string]string{zvol.ZFSTopologyKey: ns.driver.config.NodeID}
+	topology := map[string]string{zfs.ZFSTopologyKey: ns.driver.config.NodeID}
 	return &csi.NodeGetInfoResponse{
 		NodeId: ns.driver.config.NodeID,
 		AccessibleTopology: &csi.Topology{
diff --git a/pkg/driver/controller.go b/pkg/driver/controller.go
index 1f5d600..56a3c7e 100644
--- a/pkg/driver/controller.go
+++ b/pkg/driver/controller.go
@@ -25,7 +25,7 @@ import (
 	"github.com/openebs/zfs-localpv/pkg/builder"
 	errors "github.com/openebs/zfs-localpv/pkg/common/errors"
 	csipayload "github.com/openebs/zfs-localpv/pkg/response"
-	zvol "github.com/openebs/zfs-localpv/pkg/zfs"
+	zfs "github.com/openebs/zfs-localpv/pkg/zfs"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -69,7 +69,8 @@ func (cs *controller) CreateVolume(
 
 	volName := req.GetName()
 	size := req.GetCapacityRange().RequiredBytes
-	bs := req.GetParameters()["blocksize"]
+	rs := req.GetParameters()["recordsize"]
+	bs := req.GetParameters()["volblocksize"]
 	compression := req.GetParameters()["compression"]
 	dedup := req.GetParameters()["dedup"]
 	encr := req.GetParameters()["encryption"]
@@ -78,6 +79,9 @@ func (cs *controller) CreateVolume(
 	pool := req.GetParameters()["poolname"]
 	tp := req.GetParameters()["thinprovision"]
 	schld := req.GetParameters()["scheduler"]
+	fstype := req.GetParameters()["fstype"]
+
+	vtype := zfs.GetVolumeType(fstype)
 
 	selected := scheduler(req.AccessibilityRequirements, schld, pool)
 
@@ -90,7 +94,8 @@ func (cs *controller) CreateVolume(
 	volObj, err := builder.NewBuilder().
 		WithName(volName).
 		WithCapacity(strconv.FormatInt(int64(size), 10)).
-		WithBlockSize(bs).
+		WithRecordSize(rs).
+		WithVolBlockSize(bs).
 		WithPoolName(pool).
 		WithDedup(dedup).
 		WithEncryption(encr).
@@ -98,18 +103,20 @@ func (cs *controller) CreateVolume(
 		WithKeyLocation(kl).
 		WithThinProv(tp).
 		WithOwnerNode(selected).
+		WithVolumeType(vtype).
+		WithFsType(fstype).
 		WithCompression(compression).Build()
 
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
-	err = zvol.ProvisionVolume(size, volObj)
+	err = zfs.ProvisionVolume(size, volObj)
 	if err != nil {
 		return nil, status.Error(codes.Internal, "not able to provision the volume")
 	}
 
-	topology := map[string]string{zvol.ZFSTopologyKey: selected}
+	topology := map[string]string{zfs.ZFSTopologyKey: selected}
 
 	return csipayload.NewCreateVolumeResponseBuilder().
 		WithName(volName).
@@ -136,13 +143,13 @@ func (cs *controller) DeleteVolume(
 	volumeID := req.GetVolumeId()
 
 	// verify if the volume has already been deleted
-	vol, err := zvol.GetVolume(volumeID)
+	vol, err := zfs.GetVolume(volumeID)
 	if vol != nil && vol.DeletionTimestamp != nil {
 		goto deleteResponse
 	}
 
 	// Delete the corresponding ZV CR
-	err = zvol.DeleteVolume(volumeID)
+	err = zfs.DeleteVolume(volumeID)
 	if err != nil {
 		return nil, errors.Wrapf(
 			err,
diff --git a/pkg/driver/scheduler.go b/pkg/driver/scheduler.go
index 184da51..21cfc9e 100644
--- a/pkg/driver/scheduler.go
+++ b/pkg/driver/scheduler.go
@@ -76,7 +76,8 @@ func volumeWeightedScheduler(topo *csi.TopologyRequirement, pool string) string
 // the given zfs pool.
 func scheduler(topo *csi.TopologyRequirement, schld string, pool string) string {
 
-	if len(topo.Preferred) == 0 {
+	if topo == nil ||
+		len(topo.Preferred) == 0 {
 		logrus.Errorf("topology information not provided")
 		return ""
 	}
diff --git a/pkg/zfs/mount.go b/pkg/zfs/mount.go
index 3f0c1eb..46802aa 100644
--- a/pkg/zfs/mount.go
+++ b/pkg/zfs/mount.go
@@ -3,7 +3,6 @@ package zfs
 import (
 	"fmt"
 	"os"
-	"path/filepath"
 
 	"github.com/Sirupsen/logrus"
 	apis "github.com/openebs/zfs-localpv/pkg/apis/openebs.io/core/v1alpha1"
@@ -25,7 +24,6 @@ func FormatAndMountZvol(devicePath string, mountInfo *apis.MountInfo) error {
 		return err
 	}
 
-	logrus.Infof("created zvol %v and mounted %v fs %v", devicePath, mountInfo.MountPath, mountInfo.FSType)
 	return nil
 }
 
@@ -53,12 +51,22 @@ func UmountVolume(vol *apis.ZFSVolume, targetPath string,
 		return nil
 	}
 
-	if err = mounter.Unmount(targetPath); err != nil {
-		logrus.Errorf(
-			"zfspv umount volume: failed to unmount: %s\nError: %v",
-			targetPath, err,
-		)
-		return err
+	if vol.Spec.VolumeType == VOLTYPE_DATASET {
+		if err = UmountZFSDataset(vol); err != nil {
+			logrus.Errorf(
+				"zfspv failed to umount dataset: path %s Error: %v",
+				targetPath, err,
+			)
+			return err
+		}
+	} else {
+		if err = mounter.Unmount(targetPath); err != nil {
+			logrus.Errorf(
+				"zfspv failed to unmount zvol: path %s Error: %v",
+				targetPath, err,
+			)
+			return err
+		}
 	}
 
 	if err := os.RemoveAll(targetPath); err != nil {
@@ -72,7 +80,7 @@ func UmountVolume(vol *apis.ZFSVolume, targetPath string,
 }
 
 // GetMounts gets mountpoints for the specified volume
-func GetMounts(devicepath string) ([]string, error) {
+func GetMounts(dev string) ([]string, error) {
 
 	var (
 		currentMounts []string
@@ -80,10 +88,6 @@ func GetMounts(devicepath string) ([]string, error) {
 		mountList     []mount.MountPoint
 	)
 
-	dev, err := filepath.EvalSymlinks(devicepath)
-	if err != nil {
-		return nil, err
-	}
 	mounter := mount.New("")
 	// Get list of mounted paths present with the node
 	if mountList, err = mounter.List(); err != nil {
@@ -97,10 +101,8 @@ func GetMounts(devicepath string) ([]string, error) {
 	return currentMounts, nil
 }
 
-// CreateAndMountZvol creates the zfs Volume
-// and mounts the disk to the specified path
-func CreateAndMountZvol(vol *apis.ZFSVolume, mount *apis.MountInfo) error {
-	if len(mount.MountPath) == 0 {
+func verifyMountRequest(vol *apis.ZFSVolume, mountpath string) error {
+	if len(mountpath) == 0 {
 		return status.Error(codes.InvalidArgument, "mount path missing in request")
 	}
 
@@ -109,9 +111,11 @@ func CreateAndMountZvol(vol *apis.ZFSVolume, mount *apis.MountInfo) error {
 		return status.Error(codes.Internal, "volume is owned by different node")
 	}
 
-	devicePath, err := GetDevicePath(vol)
+	devicePath, err := GetVolumeDevPath(vol)
 	if err != nil {
-		return status.Error(codes.Internal, "not able to get the device path")
+		logrus.Errorf("can not get device for volume:%s dev %s err: %v",
+			vol.Name, devicePath, err.Error())
+		return err
 	}
 
 	/*
@@ -123,18 +127,63 @@ func CreateAndMountZvol(vol *apis.ZFSVolume, mount *apis.MountInfo) error {
 	 */
 	currentMounts, err := GetMounts(devicePath)
 	if err != nil {
+		logrus.Errorf("can not get mounts for volume:%s dev %s err: %v",
+			vol.Name, devicePath, err.Error())
 		return err
 	} else if len(currentMounts) >= 1 {
 		logrus.Errorf(
-			"can not mount, more than one mounts for volume:%s dev %s mounts: %v",
+			"can not mount, volume:%s already mounted dev %s mounts: %v",
 			vol.Name, devicePath, currentMounts,
 		)
 		return status.Error(codes.Internal, "device already mounted")
 	}
+	return nil
+}
+
+// MountZvol mounts the disk to the specified path
+func MountZvol(vol *apis.ZFSVolume, mount *apis.MountInfo) error {
+	volume := vol.Spec.PoolName + "/" + vol.Name
+	err := verifyMountRequest(vol, mount.MountPath)
+	if err != nil {
+		return status.Error(codes.Internal, "zvol can not be mounted")
+	}
+
+	devicePath := ZFS_DEVPATH + volume
+
 	err = FormatAndMountZvol(devicePath, mount)
 	if err != nil {
-		return status.Error(codes.Internal, "not able to mount the volume")
+		return status.Error(codes.Internal, "not able to format and mount the zvol")
 	}
 
+	logrus.Infof("zvol %v mounted %v fs %v", volume, mount.MountPath, mount.FSType)
+
 	return err
 }
+
+// MountDataset mounts the zfs dataset to the specified path
+func MountDataset(vol *apis.ZFSVolume, mount *apis.MountInfo) error {
+	volume := vol.Spec.PoolName + "/" + vol.Name
+	err := verifyMountRequest(vol, mount.MountPath)
+	if err != nil {
+		return status.Error(codes.Internal, "dataset can not be mounted")
+	}
+
+	err = MountZFSDataset(vol, mount.MountPath)
+	if err != nil {
+		return status.Error(codes.Internal, "not able to mount the dataset")
+	}
+
+	logrus.Infof("dataset %v mounted %v", volume, mount.MountPath)
+
+	return nil
+}
+
+// MountVolume mounts the disk to the specified path
+func MountVolume(vol *apis.ZFSVolume, mount *apis.MountInfo) error {
+	switch vol.Spec.VolumeType {
+	case VOLTYPE_DATASET:
+		return MountDataset(vol, mount)
+	default:
+		return MountZvol(vol, mount)
+	}
+}
diff --git a/pkg/zfs/zfs_util.go b/pkg/zfs/zfs_util.go
index afdfaed..cb42edc 100644
--- a/pkg/zfs/zfs_util.go
+++ b/pkg/zfs/zfs_util.go
@@ -17,188 +17,337 @@ limitations under the License.
 package zfs
 
 import (
-	"os"
 	"os/exec"
+	"path/filepath"
 
 	"github.com/Sirupsen/logrus"
 	apis "github.com/openebs/zfs-localpv/pkg/apis/openebs.io/core/v1alpha1"
 )
 
+// zfs related constants
+const (
+	ZFS_DEVPATH = "/dev/zvol/"
+	FSTYPE_ZFS  = "zfs"
+)
+
+// zfs command related constants
 const (
-	ZFS_DEVPATH   = "/dev/zvol/"
 	ZFSVolCmd     = "zfs"
 	ZFSCreateArg  = "create"
 	ZFSDestroyArg = "destroy"
 	ZFSSetArg     = "set"
+	ZFSListArg    = "list"
+)
+
+// constants to define volume type
+const (
+	VOLTYPE_DATASET = "DATASET"
+	VOLTYPE_ZVOL    = "ZVOL"
 )
 
 func PropertyChanged(oldVol *apis.ZFSVolume, newVol *apis.ZFSVolume) bool {
+	if oldVol.Spec.VolumeType == VOLTYPE_DATASET &&
+		newVol.Spec.VolumeType == VOLTYPE_DATASET &&
+		oldVol.Spec.RecordSize != newVol.Spec.RecordSize {
+		return true
+	}
+
 	return oldVol.Spec.Compression != newVol.Spec.Compression ||
-		oldVol.Spec.Dedup != newVol.Spec.Dedup ||
-		oldVol.Spec.Capacity != newVol.Spec.Capacity
+		oldVol.Spec.Dedup != newVol.Spec.Dedup
 }
 
-// builldVolumeCreateArgs returns zvol create command along with attributes as a string array
-func buildVolumeCreateArgs(vol *apis.ZFSVolume) []string {
-	var ZFSVolCmd []string
+// GetVolumeType returns the volume type
+// whether it is a zvol or dataset
+func GetVolumeType(fstype string) string {
+	/*
+	 * if fstype is provided as zfs then a zfs dataset will be created
+	 * otherwise a zvol will be created
+	 */
+	switch fstype {
+	case FSTYPE_ZFS:
+		return VOLTYPE_DATASET
+	default:
+		return VOLTYPE_ZVOL
+	}
+}
 
-	zvol := vol.Spec.PoolName + "/" + vol.Name
+// builldZvolCreateArgs returns zfs create command for zvol along with attributes as a string array
+func buildZvolCreateArgs(vol *apis.ZFSVolume) []string {
+	var ZFSVolArg []string
 
-	ZFSVolCmd = append(ZFSVolCmd, ZFSCreateArg)
+	volume := vol.Spec.PoolName + "/" + vol.Name
+
+	ZFSVolArg = append(ZFSVolArg, ZFSCreateArg)
 
 	if vol.Spec.ThinProvision == "yes" {
-		ZFSVolCmd = append(ZFSVolCmd, "-s")
+		ZFSVolArg = append(ZFSVolArg, "-s")
 	}
 	if len(vol.Spec.Capacity) != 0 {
-		ZFSVolCmd = append(ZFSVolCmd, "-V", vol.Spec.Capacity)
+		ZFSVolArg = append(ZFSVolArg, "-V", vol.Spec.Capacity)
 	}
-	if len(vol.Spec.BlockSize) != 0 {
-		ZFSVolCmd = append(ZFSVolCmd, "-b", vol.Spec.BlockSize)
+	if len(vol.Spec.VolBlockSize) != 0 {
+		ZFSVolArg = append(ZFSVolArg, "-b", vol.Spec.VolBlockSize)
 	}
 	if len(vol.Spec.Dedup) != 0 {
 		dedupProperty := "dedup=" + vol.Spec.Dedup
-		ZFSVolCmd = append(ZFSVolCmd, "-o", dedupProperty)
+		ZFSVolArg = append(ZFSVolArg, "-o", dedupProperty)
 	}
 	if len(vol.Spec.Compression) != 0 {
 		compressionProperty := "compression=" + vol.Spec.Compression
-		ZFSVolCmd = append(ZFSVolCmd, "-o", compressionProperty)
+		ZFSVolArg = append(ZFSVolArg, "-o", compressionProperty)
 	}
 	if len(vol.Spec.Encryption) != 0 {
 		encryptionProperty := "encryption=" + vol.Spec.Encryption
-		ZFSVolCmd = append(ZFSVolCmd, "-o", encryptionProperty)
+		ZFSVolArg = append(ZFSVolArg, "-o", encryptionProperty)
 	}
 	if len(vol.Spec.KeyLocation) != 0 {
 		keyLocation := "keylocation=" + vol.Spec.KeyLocation
-		ZFSVolCmd = append(ZFSVolCmd, "-o", keyLocation)
+		ZFSVolArg = append(ZFSVolArg, "-o", keyLocation)
 	}
 	if len(vol.Spec.KeyFormat) != 0 {
 		keyFormat := "keyformat=" + vol.Spec.KeyFormat
-		ZFSVolCmd = append(ZFSVolCmd, "-o", keyFormat)
+		ZFSVolArg = append(ZFSVolArg, "-o", keyFormat)
 	}
 
-	ZFSVolCmd = append(ZFSVolCmd, zvol)
+	ZFSVolArg = append(ZFSVolArg, volume)
 
-	return ZFSVolCmd
+	return ZFSVolArg
 }
 
-// builldVolumeSetArgs returns zvol set command along with attributes as a string array
-// TODO(pawan) need to find a way to identify which property has changed
-func buildVolumeSetArgs(vol *apis.ZFSVolume) []string {
-	var ZFSVolCmd []string
+// builldDatasetCreateArgs returns zfs create command for dataset along with attributes as a string array
+func buildDatasetCreateArgs(vol *apis.ZFSVolume) []string {
+	var ZFSVolArg []string
 
-	zvol := vol.Spec.PoolName + "/" + vol.Name
+	volume := vol.Spec.PoolName + "/" + vol.Name
 
-	ZFSVolCmd = append(ZFSVolCmd, ZFSSetArg)
+	ZFSVolArg = append(ZFSVolArg, ZFSCreateArg)
 
 	if len(vol.Spec.Capacity) != 0 {
-		volsize := "volsize=" + vol.Spec.Capacity
-		ZFSVolCmd = append(ZFSVolCmd, volsize)
+		quotaProperty := "quota=" + vol.Spec.Capacity
+		ZFSVolArg = append(ZFSVolArg, "-o", quotaProperty)
+	}
+	if len(vol.Spec.RecordSize) != 0 {
+		recordsizeProperty := "recordsize=" + vol.Spec.RecordSize
+		ZFSVolArg = append(ZFSVolArg, "-o", recordsizeProperty)
+	}
+	if vol.Spec.ThinProvision == "no" {
+		reservationProperty := "reservation=" + vol.Spec.Capacity
+		ZFSVolArg = append(ZFSVolArg, "-o", reservationProperty)
 	}
 	if len(vol.Spec.Dedup) != 0 {
 		dedupProperty := "dedup=" + vol.Spec.Dedup
-		ZFSVolCmd = append(ZFSVolCmd, dedupProperty)
+		ZFSVolArg = append(ZFSVolArg, "-o", dedupProperty)
 	}
 	if len(vol.Spec.Compression) != 0 {
 		compressionProperty := "compression=" + vol.Spec.Compression
-		ZFSVolCmd = append(ZFSVolCmd, compressionProperty)
+		ZFSVolArg = append(ZFSVolArg, "-o", compressionProperty)
+	}
+	if len(vol.Spec.Encryption) != 0 {
+		encryptionProperty := "encryption=" + vol.Spec.Encryption
+		ZFSVolArg = append(ZFSVolArg, "-o", encryptionProperty)
+	}
+	if len(vol.Spec.KeyLocation) != 0 {
+		keyLocation := "keylocation=" + vol.Spec.KeyLocation
+		ZFSVolArg = append(ZFSVolArg, "-o", keyLocation)
+	}
+	if len(vol.Spec.KeyFormat) != 0 {
+		keyFormat := "keyformat=" + vol.Spec.KeyFormat
+		ZFSVolArg = append(ZFSVolArg, "-o", keyFormat)
 	}
 
-	ZFSVolCmd = append(ZFSVolCmd, zvol)
+	// set the mount path to none, by default zfs mounts it to the default dataset path
+	ZFSVolArg = append(ZFSVolArg, "-o", "mountpoint=none", volume)
 
-	return ZFSVolCmd
+	return ZFSVolArg
 }
 
-// builldVolumeDestroyArgs returns zvol destroy command along with attributes as a string array
+// builldVolumeSetArgs returns volume set command along with attributes as a string array
+// TODO(pawan) need to find a way to identify which property has changed
+func buildVolumeSetArgs(vol *apis.ZFSVolume) []string {
+	var ZFSVolArg []string
+
+	volume := vol.Spec.PoolName + "/" + vol.Name
+
+	ZFSVolArg = append(ZFSVolArg, ZFSSetArg)
+
+	if vol.Spec.VolumeType == VOLTYPE_DATASET &&
+		len(vol.Spec.RecordSize) != 0 {
+		recordsizeProperty := "recordsize=" + vol.Spec.RecordSize
+		ZFSVolArg = append(ZFSVolArg, recordsizeProperty)
+	}
+
+	if len(vol.Spec.Dedup) != 0 {
+		dedupProperty := "dedup=" + vol.Spec.Dedup
+		ZFSVolArg = append(ZFSVolArg, dedupProperty)
+	}
+	if len(vol.Spec.Compression) != 0 {
+		compressionProperty := "compression=" + vol.Spec.Compression
+		ZFSVolArg = append(ZFSVolArg, compressionProperty)
+	}
+
+	ZFSVolArg = append(ZFSVolArg, volume)
+
+	return ZFSVolArg
+}
+
+// builldVolumeDestroyArgs returns volume destroy command along with attributes as a string array
 func buildVolumeDestroyArgs(vol *apis.ZFSVolume) []string {
-	var ZFSVolCmd []string
+	var ZFSVolArg []string
 
-	zvol := vol.Spec.PoolName + "/" + vol.Name
+	volume := vol.Spec.PoolName + "/" + vol.Name
 
-	ZFSVolCmd = append(ZFSVolCmd, ZFSDestroyArg, "-R", zvol)
+	ZFSVolArg = append(ZFSVolArg, ZFSDestroyArg, "-R", volume)
 
-	return ZFSVolCmd
+	return ZFSVolArg
 }
 
-// CreateZvol creates the zvol and returns the corresponding diskPath
-// of the volume which gets created on the node
-func CreateZvol(vol *apis.ZFSVolume) error {
-	zvol := vol.Spec.PoolName + "/" + vol.Name
-	devicePath := ZFS_DEVPATH + zvol
+func getVolume(volume string) error {
+	var ZFSVolArg []string
 
-	if _, err := os.Stat(devicePath); os.IsNotExist(err) {
+	ZFSVolArg = append(ZFSVolArg, ZFSListArg, volume)
 
-		args := buildVolumeCreateArgs(vol)
+	cmd := exec.Command(ZFSVolCmd, ZFSVolArg...)
+	_, err := cmd.CombinedOutput()
+	return err
+}
+
+// CreateVolume creates the zvol/dataset as per
+// info provided in ZFSVolume object
+func CreateVolume(vol *apis.ZFSVolume) error {
+	volume := vol.Spec.PoolName + "/" + vol.Name
+
+	if err := getVolume(volume); err != nil {
+		var args []string
+		if vol.Spec.VolumeType == VOLTYPE_DATASET {
+			args = buildDatasetCreateArgs(vol)
+		} else {
+			args = buildZvolCreateArgs(vol)
+		}
 		cmd := exec.Command(ZFSVolCmd, args...)
 		out, err := cmd.CombinedOutput()
 
 		if err != nil {
 			logrus.Errorf(
-				"zfs: could not create zvol %v cmd %v error: %s", zvol, args, string(out),
+				"zfs: could not create volume %v cmd %v error: %s", volume, args, string(out),
 			)
 			return err
 		}
-		logrus.Infof("created zvol %s", zvol)
+		logrus.Infof("created volume %s", volume)
 	} else if err == nil {
-		logrus.Infof("using existing zvol %v", zvol)
-	} else {
-		return err
+		logrus.Infof("using existing volume %v", volume)
 	}
 
 	return nil
 }
 
-// SetZvolProp sets the zvol property
+// SetDatasetMountProp sets mountpoint for the volume
+func SetDatasetMountProp(volume string, mountpath string) error {
+	var ZFSVolArg []string
+
+	mountProperty := "mountpoint=" + mountpath
+	ZFSVolArg = append(ZFSVolArg, ZFSSetArg, mountProperty, volume)
+
+	cmd := exec.Command(ZFSVolCmd, ZFSVolArg...)
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		logrus.Errorf("zfs: could not set mountpoint on dataset %v cmd %v error: %s",
+			volume, ZFSVolArg, string(out))
+	}
+	return err
+}
+
+// MountZFSDataset mounts the dataset to the given mountpoint
+func MountZFSDataset(vol *apis.ZFSVolume, mountpath string) error {
+	volume := vol.Spec.PoolName + "/" + vol.Name
+
+	return SetDatasetMountProp(volume, mountpath)
+}
+
+// UmountZFSDataset umounts the dataset
+func UmountZFSDataset(vol *apis.ZFSVolume) error {
+	volume := vol.Spec.PoolName + "/" + vol.Name
+
+	return SetDatasetMountProp(volume, "none")
+}
+
+// SetZvolProp sets the volume property
 func SetZvolProp(vol *apis.ZFSVolume) error {
 	var err error
-	zvol := vol.Spec.PoolName + "/" + vol.Name
-	devicePath := ZFS_DEVPATH + zvol
+	volume := vol.Spec.PoolName + "/" + vol.Name
 
-	if _, err = os.Stat(devicePath); err == nil {
-		args := buildVolumeSetArgs(vol)
-		cmd := exec.Command(ZFSVolCmd, args...)
-		out, err := cmd.CombinedOutput()
-
-		if err != nil {
-			logrus.Errorf(
-				"zfs: could not set property on zvol %v cmd %v error: %s", zvol, args, string(out),
-			)
-			return err
-		}
-		logrus.Infof("property set on zvol %s", zvol)
+	if len(vol.Spec.Compression) == 0 &&
+		len(vol.Spec.Dedup) == 0 &&
+		(vol.Spec.VolumeType != VOLTYPE_DATASET ||
+			len(vol.Spec.RecordSize) == 0) {
+		//nothing to set, just return
+		return nil
 	}
+	/* Case: Restart =>
+	 * In this case we get the add event but here we don't know which
+	 * property has changed when we were down, so firing the zfs set
+	 * command with the all property present on the ZFSVolume.
+
+	 * Case: Property Change =>
+	 * TODO(pawan) When we get the update event, we make sure at least
+	 * one property has changed before adding it to the event queue for
+	 * handling. At this stage, since we haven't stored the
+	 * ZFSVolume object as it will be too heavy, we are firing the set
+	 * command with the all property preset in the ZFSVolume object since
+	 * it is guaranteed that at least one property has changed.
+	 */
+
+	args := buildVolumeSetArgs(vol)
+	cmd := exec.Command(ZFSVolCmd, args...)
+	out, err := cmd.CombinedOutput()
+
+	if err != nil {
+		logrus.Errorf(
+			"zfs: could not set property on volume %v cmd %v error: %s", volume, args, string(out),
+		)
+		return err
+	}
+	logrus.Infof("property set on volume %s", volume)
 
 	return err
 }
 
-// DestroyZvol deletes the zvol
-func DestroyZvol(vol *apis.ZFSVolume) error {
-	zvol := vol.Spec.PoolName + "/" + vol.Name
-	devicePath := ZFS_DEVPATH + zvol
+// DestroyVolume deletes the zfs volume
+func DestroyVolume(vol *apis.ZFSVolume) error {
+	volume := vol.Spec.PoolName + "/" + vol.Name
 
-	if _, err := os.Stat(devicePath); err == nil {
-		args := buildVolumeDestroyArgs(vol)
-		cmd := exec.Command(ZFSVolCmd, args...)
-		out, err := cmd.CombinedOutput()
-
-		if err != nil {
-			logrus.Errorf(
-				"zfs: could not destroy zvol %v cmd %v error: %s", zvol, args, string(out),
-			)
-			return err
-		}
-		logrus.Infof("destroyed zvol %s", zvol)
+	if err := getVolume(volume); err != nil {
+		return nil
 	}
 
+	args := buildVolumeDestroyArgs(vol)
+	cmd := exec.Command(ZFSVolCmd, args...)
+	out, err := cmd.CombinedOutput()
+
+	if err != nil {
+		logrus.Errorf(
+			"zfs: could not destroy volume %v cmd %v error: %s", volume, args, string(out),
+		)
+		return err
+	}
+	logrus.Infof("destroyed volume %s", volume)
+
 	return nil
 }
 
-// GetDevicePath returns device path for zvol if it exists
-func GetDevicePath(vol *apis.ZFSVolume) (string, error) {
-	zvol := vol.Spec.PoolName + "/" + vol.Name
-	devicePath := ZFS_DEVPATH + zvol
+// GetVolumeDevPath returns devpath for the given volume
+func GetVolumeDevPath(vol *apis.ZFSVolume) (string, error) {
+	volume := vol.Spec.PoolName + "/" + vol.Name
+	if vol.Spec.VolumeType == VOLTYPE_DATASET {
+		return volume, nil
+	}
 
-	if _, err := os.Stat(devicePath); os.IsNotExist(err) {
+	devicePath := ZFS_DEVPATH + volume
+
+	// evaluate the symlink to get the dev path for zvol
+	dev, err := filepath.EvalSymlinks(devicePath)
+	if err != nil {
 		return "", err
 	}
-	return devicePath, nil
+
+	return dev, nil
 }