--- old/usr/src/lib/libzfs/common/libzfs_sendrecv.c	Mon Aug  5 17:16:42 2019
+++ new/usr/src/lib/libzfs/common/libzfs_sendrecv.c	Mon Aug  5 17:16:42 2019
@@ -23,6 +23,7 @@
  * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
  * Copyright (c) 2011, 2015 by Delphix. All rights reserved.
  * Copyright (c) 2012, Joyent, Inc. All rights reserved.
+ * Copyright 2017 Nexenta Systems, Inc. All rights reserved.
  * Copyright (c) 2012 Pawel Jakub Dawidek. All rights reserved.
  * Copyright (c) 2013 Steven Hartland. All rights reserved.
  * Copyright 2015, OmniTI Computer Consulting, Inc. All rights reserved.
@@ -48,9 +49,11 @@
 #include <libzfs.h>
 #include <libzfs_core.h>
 
+#include "zfs_errno.h"
 #include "zfs_namecheck.h"
 #include "zfs_prop.h"
 #include "zfs_fletcher.h"
+#include "zfs_sendrecv.h"
 #include "libzfs_impl.h"
 #include <zlib.h>
 #include <sha2.h>
@@ -61,8 +64,8 @@
 extern void zfs_setprop_error(libzfs_handle_t *, zfs_prop_t, int, char *);
 
 static int zfs_receive_impl(libzfs_handle_t *, const char *, const char *,
-    recvflags_t *, int, const char *, nvlist_t *, avl_tree_t *, char **, int,
-    uint64_t *, const char *);
+    recvflags_t *, int, nvlist_t *, nvlist_t *, const char *, nvlist_t *,
+    avl_tree_t *, char **, int, uint64_t *, const char *);
 static int guid_to_name(libzfs_handle_t *, const char *,
     uint64_t, boolean_t, char *);
 
@@ -71,6 +74,8 @@
 typedef struct dedup_arg {
 	int	inputfd;
 	int	outputfd;
+	uint64_t	dedup_data_sz;
+	boolean_t	sendsize;
 	libzfs_handle_t  *dedup_hdl;
 } dedup_arg_t;
 
@@ -187,11 +192,15 @@
 }
 
 static int
-dump_record(dmu_replay_record_t *drr, void *payload, int payload_len,
-    zio_cksum_t *zc, int outfd)
+dump_record(dedup_arg_t *dda, dmu_replay_record_t *drr, void *payload,
+    int payload_len, zio_cksum_t *zc, int outfd)
 {
 	ASSERT3U(offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
 	    ==, sizeof (dmu_replay_record_t) - sizeof (zio_cksum_t));
+	if (dda != NULL) {
+		dda->dedup_data_sz +=
+		    sizeof (dmu_replay_record_t) + payload_len;
+	}
 	(void) fletcher_4_incremental_native(drr,
 	    offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum), zc);
 	if (drr->drr_type != DRR_BEGIN) {
@@ -302,7 +311,7 @@
 				if (ferror(stdin))
 					perror("fread");
 			}
-			if (dump_record(drr, buf, sz, &stream_cksum,
+			if (dump_record(dda, drr, buf, sz, &stream_cksum,
 			    outfd) != 0)
 				goto out;
 			break;
@@ -313,7 +322,7 @@
 			struct drr_end *drre = &drr->drr_u.drr_end;
 			/* use the recalculated checksum */
 			drre->drr_checksum = stream_cksum;
-			if (dump_record(drr, NULL, 0, &stream_cksum,
+			if (dump_record(dda, drr, NULL, 0, &stream_cksum,
 			    outfd) != 0)
 				goto out;
 			break;
@@ -327,7 +336,7 @@
 				    P2ROUNDUP((uint64_t)drro->drr_bonuslen, 8),
 				    ofp);
 			}
-			if (dump_record(drr, buf,
+			if (dump_record(dda, drr, buf,
 			    P2ROUNDUP((uint64_t)drro->drr_bonuslen, 8),
 			    &stream_cksum, outfd) != 0)
 				goto out;
@@ -338,7 +347,7 @@
 		{
 			struct drr_spill *drrs = &drr->drr_u.drr_spill;
 			(void) ssread(buf, drrs->drr_length, ofp);
-			if (dump_record(drr, buf, drrs->drr_length,
+			if (dump_record(dda, drr, buf, drrs->drr_length,
 			    &stream_cksum, outfd) != 0)
 				goto out;
 			break;
@@ -346,7 +355,7 @@
 
 		case DRR_FREEOBJECTS:
 		{
-			if (dump_record(drr, NULL, 0, &stream_cksum,
+			if (dump_record(dda, drr, NULL, 0, &stream_cksum,
 			    outfd) != 0)
 				goto out;
 			break;
@@ -420,12 +429,12 @@
 				wbr_drrr->drr_key.ddk_prop =
 				    drrw->drr_key.ddk_prop;
 
-				if (dump_record(&wbr_drr, NULL, 0,
+				if (dump_record(dda, &wbr_drr, NULL, 0,
 				    &stream_cksum, outfd) != 0)
 					goto out;
 			} else {
 				/* block not previously seen */
-				if (dump_record(drr, buf, payload_size,
+				if (dump_record(dda, drr, buf, payload_size,
 				    &stream_cksum, outfd) != 0)
 					goto out;
 			}
@@ -438,7 +447,7 @@
 			    &drr->drr_u.drr_write_embedded;
 			(void) ssread(buf,
 			    P2ROUNDUP((uint64_t)drrwe->drr_psize, 8), ofp);
-			if (dump_record(drr, buf,
+			if (dump_record(dda, drr, buf,
 			    P2ROUNDUP((uint64_t)drrwe->drr_psize, 8),
 			    &stream_cksum, outfd) != 0)
 				goto out;
@@ -447,7 +456,7 @@
 
 		case DRR_FREE:
 		{
-			if (dump_record(drr, NULL, 0, &stream_cksum,
+			if (dump_record(dda, drr, NULL, 0, &stream_cksum,
 			    outfd) != 0)
 				goto out;
 			break;
@@ -470,117 +479,6 @@
 }
 
 /*
- * Routines for dealing with the AVL tree of fs-nvlists
- */
-typedef struct fsavl_node {
-	avl_node_t fn_node;
-	nvlist_t *fn_nvfs;
-	char *fn_snapname;
-	uint64_t fn_guid;
-} fsavl_node_t;
-
-static int
-fsavl_compare(const void *arg1, const void *arg2)
-{
-	const fsavl_node_t *fn1 = arg1;
-	const fsavl_node_t *fn2 = arg2;
-
-	if (fn1->fn_guid > fn2->fn_guid)
-		return (+1);
-	else if (fn1->fn_guid < fn2->fn_guid)
-		return (-1);
-	else
-		return (0);
-}
-
-/*
- * Given the GUID of a snapshot, find its containing filesystem and
- * (optionally) name.
- */
-static nvlist_t *
-fsavl_find(avl_tree_t *avl, uint64_t snapguid, char **snapname)
-{
-	fsavl_node_t fn_find;
-	fsavl_node_t *fn;
-
-	fn_find.fn_guid = snapguid;
-
-	fn = avl_find(avl, &fn_find, NULL);
-	if (fn) {
-		if (snapname)
-			*snapname = fn->fn_snapname;
-		return (fn->fn_nvfs);
-	}
-	return (NULL);
-}
-
-static void
-fsavl_destroy(avl_tree_t *avl)
-{
-	fsavl_node_t *fn;
-	void *cookie;
-
-	if (avl == NULL)
-		return;
-
-	cookie = NULL;
-	while ((fn = avl_destroy_nodes(avl, &cookie)) != NULL)
-		free(fn);
-	avl_destroy(avl);
-	free(avl);
-}
-
-/*
- * Given an nvlist, produce an avl tree of snapshots, ordered by guid
- */
-static avl_tree_t *
-fsavl_create(nvlist_t *fss)
-{
-	avl_tree_t *fsavl;
-	nvpair_t *fselem = NULL;
-
-	if ((fsavl = malloc(sizeof (avl_tree_t))) == NULL)
-		return (NULL);
-
-	avl_create(fsavl, fsavl_compare, sizeof (fsavl_node_t),
-	    offsetof(fsavl_node_t, fn_node));
-
-	while ((fselem = nvlist_next_nvpair(fss, fselem)) != NULL) {
-		nvlist_t *nvfs, *snaps;
-		nvpair_t *snapelem = NULL;
-
-		VERIFY(0 == nvpair_value_nvlist(fselem, &nvfs));
-		VERIFY(0 == nvlist_lookup_nvlist(nvfs, "snaps", &snaps));
-
-		while ((snapelem =
-		    nvlist_next_nvpair(snaps, snapelem)) != NULL) {
-			fsavl_node_t *fn;
-			uint64_t guid;
-
-			VERIFY(0 == nvpair_value_uint64(snapelem, &guid));
-			if ((fn = malloc(sizeof (fsavl_node_t))) == NULL) {
-				fsavl_destroy(fsavl);
-				return (NULL);
-			}
-			fn->fn_nvfs = nvfs;
-			fn->fn_snapname = nvpair_name(snapelem);
-			fn->fn_guid = guid;
-
-			/*
-			 * Note: if there are multiple snaps with the
-			 * same GUID, we ignore all but one.
-			 */
-			if (avl_find(fsavl, fn, NULL) == NULL)
-				avl_add(fsavl, fn);
-			else
-				free(fn);
-		}
-	}
-
-	return (fsavl);
-}
-
-/*
  * Routines for dealing with the giant nvlist of fs-nvlists, etc.
  */
 typedef struct send_data {
@@ -624,7 +522,8 @@
 	 *	 "snaps" -> { name (lastname) -> number (guid) }
 	 *	 "snapprops" -> { name (lastname) -> { name -> value } }
 	 *
-	 *	 "origin" -> number (guid) (if clone)
+	 *	 "origin" -> number (guid of origin snapshot) (if clone)
+	 *	 "origin_fsname" -> string (full name of origin file system)
 	 *	 "sent" -> boolean (not on-disk)
 	 *	}
 	 *   }
@@ -687,6 +586,13 @@
 		zfs_prop_t prop = zfs_name_to_prop(propname);
 		nvlist_t *propnv;
 
+		/*
+		 * This property make sense only to this dataset,
+		 * so no reasons to include it into stream
+		 */
+                if (prop == ZFS_PROP_WBC_MODE)
+                        continue;
+
 		if (!zfs_prop_user(propname)) {
 			/*
 			 * Realistically, this should never happen.  However,
@@ -834,6 +740,7 @@
 	    sd->parent_fromsnap_guid));
 
 	if (zhp->zfs_dmustats.dds_origin[0]) {
+		char origin_fsname[ZFS_MAX_DATASET_NAME_LEN];
 		zfs_handle_t *origin = zfs_open(zhp->zfs_hdl,
 		    zhp->zfs_dmustats.dds_origin, ZFS_TYPE_SNAPSHOT);
 		if (origin == NULL) {
@@ -842,6 +749,12 @@
 		}
 		VERIFY(0 == nvlist_add_uint64(nvfs, "origin",
 		    origin->zfs_dmustats.dds_guid));
+		zfs_close(origin);
+		(void) strlcpy(origin_fsname, zhp->zfs_dmustats.dds_origin,
+		    sizeof (origin_fsname));
+		*strchr(origin_fsname, '@') = '\0';
+		VERIFY(0 == nvlist_add_string(nvfs, "origin_fsname",
+		    origin_fsname));
 	}
 
 	/* iterate over props */
@@ -854,7 +767,7 @@
 	sd->parent_fromsnap_guid = 0;
 	VERIFY(0 == nvlist_alloc(&sd->parent_snaps, NV_UNIQUE_NAME, 0));
 	VERIFY(0 == nvlist_alloc(&sd->snapprops, NV_UNIQUE_NAME, 0));
-	(void) zfs_iter_snapshots(zhp, B_FALSE, send_iterate_snap, sd);
+	(void) zfs_iter_snapshots_sorted(zhp, send_iterate_snap, sd);
 	VERIFY(0 == nvlist_add_nvlist(nvfs, "snaps", sd->parent_snaps));
 	VERIFY(0 == nvlist_add_nvlist(nvfs, "snapprops", sd->snapprops));
 	nvlist_free(sd->parent_snaps);
@@ -907,7 +820,7 @@
 		return (error);
 	}
 
-	if (avlp != NULL && (*avlp = fsavl_create(sd.fss)) == NULL) {
+	if (avlp != NULL && fsavl_create(sd.fss, avlp) != 0) {
 		nvlist_free(sd.fss);
 		*nvlp = NULL;
 		return (EZFS_NOMEM);
@@ -927,8 +840,11 @@
 	char prevsnap[ZFS_MAX_DATASET_NAME_LEN];
 	uint64_t prevsnap_obj;
 	boolean_t seenfrom, seento, replicate, doall, fromorigin;
-	boolean_t verbose, dryrun, parsable, progress, embed_data, std_out;
+	boolean_t verbose, dryrun, dedup, parsable, progress, embed_data, std_out;
 	boolean_t large_block, compress;
+	boolean_t sendsize;
+	uint32_t hdr_send_sz;
+	uint64_t send_sz;
 	int outfd;
 	boolean_t err;
 	nvlist_t *fss;
@@ -1010,7 +926,7 @@
 static int
 dump_ioctl(zfs_handle_t *zhp, const char *fromsnap, uint64_t fromsnap_obj,
     boolean_t fromorigin, int outfd, enum lzc_send_flags flags,
-    nvlist_t *debugnv)
+    nvlist_t *debugnv, boolean_t sendsize, uint64_t *sendcounter)
 {
 	zfs_cmd_t zc = { 0 };
 	libzfs_handle_t *hdl = zhp->zfs_hdl;
@@ -1025,6 +941,8 @@
 	zc.zc_sendobj = zfs_prop_get_int(zhp, ZFS_PROP_OBJSETID);
 	zc.zc_fromobj = fromsnap_obj;
 	zc.zc_flags = flags;
+	zc.zc_sendsize = sendsize;
+	zc.zc_sendcounter = 0;
 
 	VERIFY(0 == nvlist_alloc(&thisdbg, NV_UNIQUE_NAME, 0));
 	if (fromsnap && fromsnap[0] != '\0') {
@@ -1078,6 +996,7 @@
 		}
 	}
 
+	*sendcounter = (uint64_t)zc.zc_sendcounter;
 	if (debugnv)
 		VERIFY(0 == nvlist_add_nvlist(debugnv, zhp->zfs_name, thisdbg));
 	nvlist_free(thisdbg);
@@ -1275,6 +1194,7 @@
 	fromorigin = sdd->prevsnap[0] == '\0' &&
 	    (sdd->fromorigin || sdd->replicate);
 
+	/* print out to-from and approximate size in verbose mode */
 	if (sdd->verbose) {
 		uint64_t size = 0;
 		(void) estimate_ioctl(zhp, sdd->prevsnap_obj,
@@ -1287,11 +1207,14 @@
 	}
 
 	if (!sdd->dryrun) {
+		uint64_t sendcounter = 0;
+		boolean_t track_progress = (sdd->progress && !sdd->sendsize);
+		boolean_t sendsize = B_FALSE;
 		/*
 		 * If progress reporting is requested, spawn a new thread to
 		 * poll ZFS_IOC_SEND_PROGRESS at a regular interval.
 		 */
-		if (sdd->progress) {
+		if (track_progress) {
 			pa.pa_zhp = zhp;
 			pa.pa_fd = sdd->outfd;
 			pa.pa_parsable = sdd->parsable;
@@ -1303,10 +1226,25 @@
 			}
 		}
 
+		/*
+		 * We need to reset the sendsize flag being sent to
+		 * kernel if sdd->dedup is set. With dedup, the file
+		 * descriptor sent to kernel is one end of the pipe,
+		 * and we would want the data back in the pipe for
+		 * cksummer() to calculate the exact size of the dedup-ed
+		 * stream. So reset the sendsize flag such that
+		 * kernel writes to the pipe.
+		 */
+
+		sendsize = sdd->dedup ? B_FALSE : sdd->sendsize;
+
 		err = dump_ioctl(zhp, sdd->prevsnap, sdd->prevsnap_obj,
-		    fromorigin, sdd->outfd, flags, sdd->debugnv);
+		    fromorigin, sdd->outfd, flags, sdd->debugnv,
+		    sendsize, &sendcounter);
 
-		if (sdd->progress) {
+		sdd->send_sz += sendcounter;
+
+		if (track_progress) {
 			(void) pthread_cancel(tid);
 			(void) pthread_join(tid, NULL);
 		}
@@ -1491,78 +1429,44 @@
 nvlist_t *
 zfs_send_resume_token_to_nvlist(libzfs_handle_t *hdl, const char *token)
 {
-	unsigned int version;
-	int nread;
-	unsigned long long checksum, packed_len;
+	nvlist_t *nvl = NULL;
+	int error;
 
-	/*
-	 * Decode token header, which is:
-	 *   <token version>-<checksum of payload>-<uncompressed payload length>
-	 * Note that the only supported token version is 1.
-	 */
-	nread = sscanf(token, "%u-%llx-%llx-",
-	    &version, &checksum, &packed_len);
-	if (nread != 3) {
+	error = zfs_send_resume_token_to_nvlist_impl(token, &nvl);
+	switch (error) {
+	case EINVAL:
 		zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
 		    "resume token is corrupt (invalid format)"));
-		return (NULL);
-	}
-
-	if (version != ZFS_SEND_RESUME_TOKEN_VERSION) {
+		break;
+	case ENOTSUP:
 		zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
-		    "resume token is corrupt (invalid version %u)"),
-		    version);
-		return (NULL);
-	}
-
-	/* convert hexadecimal representation to binary */
-	token = strrchr(token, '-') + 1;
-	int len = strlen(token) / 2;
-	unsigned char *compressed = zfs_alloc(hdl, len);
-	for (int i = 0; i < len; i++) {
-		nread = sscanf(token + i * 2, "%2hhx", compressed + i);
-		if (nread != 1) {
-			free(compressed);
-			zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
-			    "resume token is corrupt "
-			    "(payload is not hex-encoded)"));
-			return (NULL);
-		}
-	}
-
-	/* verify checksum */
-	zio_cksum_t cksum;
-	fletcher_4_native(compressed, len, NULL, &cksum);
-	if (cksum.zc_word[0] != checksum) {
-		free(compressed);
+		    "resume token is corrupt (invalid version)"));
+		break;
+	case EBADMSG:
 		zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
+		    "resume token is corrupt "
+		    "(payload is not hex-encoded)"));
+		break;
+	case ECKSUM:
+		zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
 		    "resume token is corrupt (incorrect checksum)"));
-		return (NULL);
-	}
-
-	/* uncompress */
-	void *packed = zfs_alloc(hdl, packed_len);
-	uLongf packed_len_long = packed_len;
-	if (uncompress(packed, &packed_len_long, compressed, len) != Z_OK ||
-	    packed_len_long != packed_len) {
-		free(packed);
-		free(compressed);
+		break;
+	case ENOSR:
 		zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
 		    "resume token is corrupt (decompression failed)"));
-		return (NULL);
-	}
-
-	/* unpack nvlist */
-	nvlist_t *nv;
-	int error = nvlist_unpack(packed, packed_len, &nv, KM_SLEEP);
-	free(packed);
-	free(compressed);
-	if (error != 0) {
+		break;
+	case ENODATA:
 		zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
 		    "resume token is corrupt (nvlist_unpack failed)"));
-		return (NULL);
-	}
-	return (nv);
+		break;
+	case ENOMEM:
+		(void) no_memory(hdl);
+		break;
+	default:
+		break;
+	};
+
+	return (nvl);
 }
 
 int
@@ -1577,7 +1481,6 @@
 	int error = 0;
 	char name[ZFS_MAX_DATASET_NAME_LEN];
 	enum lzc_send_flags lzc_flags = 0;
-	FILE *fout = (flags->verbose && flags->dryrun) ? stdout : stderr;
 
 	(void) snprintf(errbuf, sizeof (errbuf), dgettext(TEXT_DOMAIN,
 	    "cannot resume send"));
@@ -1592,9 +1495,9 @@
 		return (zfs_error(hdl, EZFS_FAULT, errbuf));
 	}
 	if (flags->verbose) {
-		(void) fprintf(fout, dgettext(TEXT_DOMAIN,
+		(void) fprintf(stderr, dgettext(TEXT_DOMAIN,
 		    "resume token contents:\n"));
-		nvlist_print(fout, resume_nvl);
+		nvlist_print(stderr, resume_nvl);
 	}
 
 	if (nvlist_lookup_string(resume_nvl, "toname", &toname) != 0 ||
@@ -1651,7 +1554,7 @@
 		    lzc_flags, &size);
 		if (error == 0)
 			size = MAX(0, (int64_t)(size - bytes));
-		send_print_verbose(fout, zhp->zfs_name, fromname,
+		send_print_verbose(stderr, zhp->zfs_name, fromname,
 		    size, flags->parsable);
 	}
 
@@ -1781,6 +1684,7 @@
 		dda.outputfd = outfd;
 		dda.inputfd = pipefd[1];
 		dda.dedup_hdl = zhp->zfs_hdl;
+		dda.sendsize = flags->sendsize;
 		if ((err = pthread_create(&tid, NULL, cksummer, &dda)) != 0) {
 			(void) close(pipefd[0]);
 			(void) close(pipefd[1]);
@@ -1839,7 +1743,7 @@
 			    "%s@%s", zhp->zfs_name, tosnap);
 			drr.drr_payloadlen = buflen;
 
-			err = dump_record(&drr, packbuf, buflen, &zc, outfd);
+			err = dump_record(NULL, &drr, packbuf, buflen, &zc, outfd);
 			free(packbuf);
 			if (err != 0)
 				goto stderr_out;
@@ -1849,6 +1753,7 @@
 			drr.drr_type = DRR_END;
 			drr.drr_u.drr_end.drr_checksum = zc;
 			err = write(outfd, &drr, sizeof (drr));
+			sdd.hdr_send_sz += sizeof (drr);
 			if (err == -1) {
 				err = errno;
 				goto stderr_out;
@@ -1871,6 +1776,8 @@
 	sdd.fss = fss;
 	sdd.fsavl = fsavl;
 	sdd.verbose = flags->verbose;
+	sdd.dedup = flags->dedup;
+	sdd.sendsize = flags->sendsize;
 	sdd.parsable = flags->parsable;
 	sdd.progress = flags->progress;
 	sdd.dryrun = flags->dryrun;
@@ -1909,7 +1816,7 @@
 		sdd.cleanup_fd = -1;
 		sdd.snapholds = NULL;
 	}
-	if (flags->verbose || sdd.snapholds != NULL) {
+	if ((flags->verbose && !flags->sendsize) || sdd.snapholds != NULL) {
 		/*
 		 * Do a verbose no-op dry run to get all the verbose output
 		 * or to gather snapshot hold's before generating any data,
@@ -1969,6 +1876,7 @@
 			(void) pthread_cancel(tid);
 		(void) close(pipefd[0]);
 		(void) pthread_join(tid, NULL);
+		sdd.send_sz = dda.dedup_data_sz;
 	}
 
 	if (sdd.cleanup_fd != -1) {
@@ -1989,8 +1897,29 @@
 			return (zfs_standard_error(zhp->zfs_hdl,
 			    errno, errbuf));
 		}
+		sdd.hdr_send_sz += sizeof (drr);
 	}
 
+	if (flags->sendsize) {
+		if (flags->verbose) {
+			(void) fprintf(stderr,
+			"Send stream header size (bytes): %u\n",
+			    sdd.hdr_send_sz);
+			(void) fprintf(stderr,
+			"Send stream data size  (bytes):  %llu\n",
+			    (longlong_t)sdd.send_sz);
+			(void) fprintf(stderr,
+			"Total send stream size (bytes):  %llu\n",
+			    (longlong_t)(sdd.send_sz +
+			    (uint64_t)sdd.hdr_send_sz));
+		} else {
+			(void) fprintf(stderr,
+			"Total send stream size (bytes):  %llu\n",
+			    (longlong_t)(sdd.send_sz +
+			    (uint64_t)sdd.hdr_send_sz));
+		}
+	}
+
 	return (err || sdd.err);
 
 stderr_out:
@@ -2138,6 +2067,9 @@
 	prop_changelist_t *clp;
 	zfs_handle_t *zhp;
 
+	if (!zfs_dataset_exists(hdl, name, ZFS_TYPE_DATASET))
+		return (ENOENT);
+
 	zhp = zfs_open(hdl, name, ZFS_TYPE_DATASET);
 	if (zhp == NULL)
 		return (-1);
@@ -2155,16 +2087,26 @@
 
 	if (tryname) {
 		(void) strcpy(newname, tryname);
-
 		(void) strlcpy(zc.zc_value, tryname, sizeof (zc.zc_value));
+		err = ioctl(hdl->libzfs_fd, ZFS_IOC_RENAME, &zc);
 
 		if (flags->verbose) {
-			(void) printf("attempting rename %s to %s\n",
+			char errbuf[1024];
+			(void) snprintf(errbuf, sizeof (errbuf),
+			    dgettext(TEXT_DOMAIN,
+			    "attempting to rename '%s' to '%s': "),
 			    zc.zc_name, zc.zc_value);
+			if (err == 0) {
+				(void) fprintf(stderr, dgettext(TEXT_DOMAIN,
+				    "%s: success\n"), errbuf);
+			} else {
+				(void) zfs_standard_error(hdl, errno, errbuf);
+			}
 		}
-		err = ioctl(hdl->libzfs_fd, ZFS_IOC_RENAME, &zc);
+
 		if (err == 0)
 			changelist_rename(clp, name, tryname);
+
 	} else {
 		err = ENOENT;
 	}
@@ -2175,24 +2117,26 @@
 		(void) snprintf(newname, ZFS_MAX_DATASET_NAME_LEN,
 		    "%.*srecv-%u-%u", baselen, name, getpid(), seq);
 		(void) strlcpy(zc.zc_value, newname, sizeof (zc.zc_value));
+		err = ioctl(hdl->libzfs_fd, ZFS_IOC_RENAME, &zc);
 
 		if (flags->verbose) {
-			(void) printf("failed - trying rename %s to %s\n",
+			char errbuf[1024];
+			(void) snprintf(errbuf, sizeof (errbuf),
+			    dgettext(TEXT_DOMAIN,
+			    "attempting to temporarily rename '%s' to '%s': "),
 			    zc.zc_name, zc.zc_value);
+			if (err == 0) {
+				(void) fprintf(stderr, dgettext(TEXT_DOMAIN,
+				    "%s: success\n"), errbuf);
+			} else {
+				(void) zfs_standard_error(hdl, errno, errbuf);
+			}
 		}
-		err = ioctl(hdl->libzfs_fd, ZFS_IOC_RENAME, &zc);
+
 		if (err == 0)
 			changelist_rename(clp, name, newname);
-		if (err && flags->verbose) {
-			(void) printf("failed (%u) - "
-			    "will try again on next pass\n", errno);
-		}
+
 		err = EAGAIN;
-	} else if (flags->verbose) {
-		if (err == 0)
-			(void) printf("success\n");
-		else
-			(void) printf("failed (%u)\n", errno);
 	}
 
 	(void) changelist_postfix(clp);
@@ -2212,6 +2156,9 @@
 	boolean_t defer = B_FALSE;
 	int spa_version;
 
+	if (!zfs_dataset_exists(hdl, name, ZFS_TYPE_DATASET))
+		return (ENOENT);
+
 	zhp = zfs_open(hdl, name, ZFS_TYPE_DATASET);
 	if (zhp == NULL)
 		return (-1);
@@ -2231,16 +2178,23 @@
 	zc.zc_objset_type = DMU_OST_ZFS;
 	zc.zc_defer_destroy = defer;
 	(void) strlcpy(zc.zc_name, name, sizeof (zc.zc_name));
-
-	if (flags->verbose)
-		(void) printf("attempting destroy %s\n", zc.zc_name);
 	err = ioctl(hdl->libzfs_fd, ZFS_IOC_DESTROY, &zc);
-	if (err == 0) {
-		if (flags->verbose)
-			(void) printf("success\n");
-		changelist_remove(clp, zc.zc_name);
+
+	if (flags->verbose) {
+		char errbuf[1024];
+		(void) snprintf(errbuf, sizeof (errbuf), dgettext(TEXT_DOMAIN,
+		    "attempting to destroy '%s'"), zc.zc_name);
+		if (err == 0) {
+			(void) fprintf(stderr, dgettext(TEXT_DOMAIN,
+			    "%s: success\n"), errbuf);
+		} else {
+			(void) zfs_standard_error(hdl, errno, errbuf);
+		}
 	}
 
+	if (err == 0)
+		changelist_remove(clp, zc.zc_name);
+
 	(void) changelist_postfix(clp);
 	changelist_free(clp);
 
@@ -2347,61 +2301,52 @@
 }
 
 /*
- * Return +1 if guid1 is before guid2, 0 if they are the same, and -1 if
- * guid1 is after guid2.
+ * Returns a value:
+ * +1 - promote is reqired
+ *  0 - promote is not required
+ * -1 - an error is occured
  */
 static int
-created_before(libzfs_handle_t *hdl, avl_tree_t *avl,
+check_promote(libzfs_handle_t *hdl, avl_tree_t *avl,
     uint64_t guid1, uint64_t guid2)
 {
 	nvlist_t *nvfs;
 	char *fsname, *snapname;
-	char buf[ZFS_MAX_DATASET_NAME_LEN];
-	int rv;
-	zfs_handle_t *guid1hdl, *guid2hdl;
 	uint64_t create1, create2;
 
+	/* the local dataset is not cloned */
 	if (guid2 == 0)
 		return (0);
+
+	/* the stream dataset is not cloned */
 	if (guid1 == 0)
 		return (1);
 
 	nvfs = fsavl_find(avl, guid1, &snapname);
+	if (nvfs == NULL)
+		return (0);
 	VERIFY(0 == nvlist_lookup_string(nvfs, "name", &fsname));
-	(void) snprintf(buf, sizeof (buf), "%s@%s", fsname, snapname);
-	guid1hdl = zfs_open(hdl, buf, ZFS_TYPE_SNAPSHOT);
-	if (guid1hdl == NULL)
-		return (-1);
+	create1 = get_snap_txg(hdl, fsname, snapname);
 
 	nvfs = fsavl_find(avl, guid2, &snapname);
+	if (nvfs == NULL)
+		return (0);
 	VERIFY(0 == nvlist_lookup_string(nvfs, "name", &fsname));
-	(void) snprintf(buf, sizeof (buf), "%s@%s", fsname, snapname);
-	guid2hdl = zfs_open(hdl, buf, ZFS_TYPE_SNAPSHOT);
-	if (guid2hdl == NULL) {
-		zfs_close(guid1hdl);
+	create2 = get_snap_txg(hdl, fsname, snapname);
+
+	if (create1 == 0 || create2 == 0)
 		return (-1);
-	}
 
-	create1 = zfs_prop_get_int(guid1hdl, ZFS_PROP_CREATETXG);
-	create2 = zfs_prop_get_int(guid2hdl, ZFS_PROP_CREATETXG);
-
 	if (create1 < create2)
-		rv = -1;
-	else if (create1 > create2)
-		rv = +1;
-	else
-		rv = 0;
+		return (1);
 
-	zfs_close(guid1hdl);
-	zfs_close(guid2hdl);
-
-	return (rv);
+	return (0);
 }
 
 static int
 recv_incremental_replication(libzfs_handle_t *hdl, const char *tofs,
     recvflags_t *flags, nvlist_t *stream_nv, avl_tree_t *stream_avl,
-    nvlist_t *renamed)
+    nvlist_t *renamed, nvlist_t *limitds)
 {
 	nvlist_t *local_nv;
 	avl_tree_t *local_avl;
@@ -2440,6 +2385,8 @@
 		uint64_t stream_originguid = 0;
 		uint64_t parent_fromsnap_guid, stream_parent_fromsnap_guid;
 		char *fsname, *stream_fsname;
+		boolean_t stream_fs_exists = B_FALSE;
+		boolean_t stream_originfs_exists = B_FALSE;
 
 		nextfselem = nvlist_next_nvpair(local_nv, fselem);
 
@@ -2450,75 +2397,123 @@
 		    &parent_fromsnap_guid));
 		(void) nvlist_lookup_uint64(nvfs, "origin", &originguid);
 
+		if (!nvlist_empty(limitds) && !nvlist_exists(limitds, fsname)) {
+			if (flags->verbose) {
+				(void) fprintf(stderr, dgettext(TEXT_DOMAIN,
+				    "skip receiving for excluded '%s'\n"),
+				    fsname);
+			}
+			continue;
+		}
+
 		/*
 		 * First find the stream's fs, so we can check for
 		 * a different origin (due to "zfs promote")
+		 * and for preserving snapshots on the receiving side
 		 */
 		for (snapelem = nvlist_next_nvpair(snaps, NULL);
-		    snapelem; snapelem = nvlist_next_nvpair(snaps, snapelem)) {
-			uint64_t thisguid;
+		    snapelem != NULL; snapelem = nextsnapelem) {
+			uint64_t snapguid;
 
-			VERIFY(0 == nvpair_value_uint64(snapelem, &thisguid));
-			stream_nvfs = fsavl_find(stream_avl, thisguid, NULL);
+			nextsnapelem = nvlist_next_nvpair(snaps, snapelem);
+			VERIFY(0 == nvpair_value_uint64(snapelem, &snapguid));
+			stream_nvfs = fsavl_find(stream_avl, snapguid, NULL);
 
-			if (stream_nvfs != NULL)
+			if (stream_nvfs != NULL) {
+				stream_fs_exists = B_TRUE;
 				break;
+			}
 		}
 
+		/* Check the stream's fs for origin snapshot */
+		if (stream_fs_exists && originguid != 0) {
+			nvlist_t *stream_snaps;
+
+			VERIFY(0 == nvlist_lookup_nvlist(stream_nvfs, "snaps",
+			    &stream_snaps));
+
+			for (snapelem = nvlist_next_nvpair(stream_snaps, NULL);
+			    snapelem != NULL; snapelem = nextsnapelem) {
+				uint64_t stream_snapguid;
+
+				nextsnapelem = nvlist_next_nvpair(stream_snaps,
+				    snapelem);
+				VERIFY(0 == nvpair_value_uint64(snapelem,
+				    &stream_snapguid));
+
+				if (stream_snapguid == originguid) {
+					stream_originfs_exists = B_TRUE;
+					break;
+				}
+			}
+		}
+
 		/* check for promote */
 		(void) nvlist_lookup_uint64(stream_nvfs, "origin",
 		    &stream_originguid);
-		if (stream_nvfs && originguid != stream_originguid) {
-			switch (created_before(hdl, local_avl,
+		if (originguid != stream_originguid && stream_originfs_exists) {
+			switch (check_promote(hdl, local_avl,
 			    stream_originguid, originguid)) {
+			case 0:
+				break;
 			case 1: {
 				/* promote it! */
 				zfs_cmd_t zc = { 0 };
-				nvlist_t *origin_nvfs;
 				char *origin_fsname;
 
-				if (flags->verbose)
-					(void) printf("promoting %s\n", fsname);
-
-				origin_nvfs = fsavl_find(local_avl, originguid,
-				    NULL);
-				VERIFY(0 == nvlist_lookup_string(origin_nvfs,
-				    "name", &origin_fsname));
+				VERIFY(0 == nvlist_lookup_string(nvfs,
+				    "origin_fsname", &origin_fsname));
 				(void) strlcpy(zc.zc_value, origin_fsname,
 				    sizeof (zc.zc_value));
 				(void) strlcpy(zc.zc_name, fsname,
 				    sizeof (zc.zc_name));
 				error = zfs_ioctl(hdl, ZFS_IOC_PROMOTE, &zc);
+
+				if (flags->verbose) {
+					char errbuf[1024];
+					(void) snprintf(errbuf, sizeof (errbuf),
+					    dgettext(TEXT_DOMAIN,
+					    "attempting to promote '%s': "),
+					    zc.zc_name);
+					if (error == 0) {
+						(void) fprintf(stderr,
+						    dgettext(TEXT_DOMAIN,
+						    "%s: success\n"), errbuf);
+					} else {
+						(void) zfs_standard_error(hdl,
+						    errno, errbuf);
+					}
+				}
+
 				if (error == 0)
 					progress = B_TRUE;
-				break;
+
+				/*
+				 * We had/have the wrong origin, therefore our
+				 * list of snapshots is wrong. Need to handle
+				 * them on the next pass.
+				 */
+
+				needagain = B_TRUE;
+				goto out;
 			}
 			default:
-				break;
-			case -1:
-				fsavl_destroy(local_avl);
-				nvlist_free(local_nv);
-				return (-1);
+				progress = B_FALSE;
+				needagain = B_FALSE;
+				goto out;
 			}
-			/*
-			 * We had/have the wrong origin, therefore our
-			 * list of snapshots is wrong.  Need to handle
-			 * them on the next pass.
-			 */
-			needagain = B_TRUE;
-			continue;
 		}
 
 		for (snapelem = nvlist_next_nvpair(snaps, NULL);
-		    snapelem; snapelem = nextsnapelem) {
-			uint64_t thisguid;
+		    snapelem != NULL; snapelem = nextsnapelem) {
+			uint64_t snapguid;
 			char *stream_snapname;
 			nvlist_t *found, *props;
 
 			nextsnapelem = nvlist_next_nvpair(snaps, snapelem);
 
-			VERIFY(0 == nvpair_value_uint64(snapelem, &thisguid));
-			found = fsavl_find(stream_avl, thisguid,
+			VERIFY(0 == nvpair_value_uint64(snapelem, &snapguid));
+			found = fsavl_find(stream_avl, snapguid,
 			    &stream_snapname);
 
 			/* check for delete */
@@ -2525,18 +2520,53 @@
 			if (found == NULL) {
 				char name[ZFS_MAX_DATASET_NAME_LEN];
 
+				/*
+				 * Conventional force-receive (-F) behavior
+				 * combines two different steps:
+				 * 1. rollback the destination dataset to the
+				 *    most recent received snapshot
+				 * 2. destroy all those destination snapshots
+				 *    that are not present at the source
+				 * The keepsnap flag allows to effectively
+				 * separate 1 from 2 and perform forced receive
+				 * while still maintaining the destination
+				 * snapshots as per the corresponding snapshot
+				 * retention policy (at the destination).
+				 */
+
+				/*
+				 * When -F (force-receive) is not specified we
+				 * always keep snapshots at the destination
+				 * (i.e., this has always been zfs conventional
+				 * behavior). See also 'keepsnap' comment below
+				 */
 				if (!flags->force)
 					continue;
 
+				/*
+				 * keepsnap flag modifies the conventional
+				 * force-receive behavior not to destroy
+				 * destination snapshots that are not present
+				 * at the replication source
+				 */
+				if (flags->keepsnap && stream_fs_exists)
+					continue;
+
+				/*
+				 * Destroy destination snapshots that do
+				 * not exist at the replication source
+				 */
 				(void) snprintf(name, sizeof (name), "%s@%s",
 				    fsname, nvpair_name(snapelem));
 
 				error = recv_destroy(hdl, name,
 				    strlen(fsname)+1, newname, flags);
-				if (error)
-					needagain = B_TRUE;
-				else
+
+				if (error == 0)
 					progress = B_TRUE;
+				else
+					needagain = B_TRUE;
+
 				continue;
 			}
 
@@ -2571,14 +2601,15 @@
 
 				error = recv_rename(hdl, name, tryname,
 				    strlen(fsname)+1, newname, flags);
-				if (error)
-					needagain = B_TRUE;
-				else
+
+				if (error == 0)
 					progress = B_TRUE;
+				else
+					needagain = B_TRUE;
 			}
 
 			if (strcmp(stream_snapname, fromsnap) == 0)
-				fromguid = thisguid;
+				fromguid = snapguid;
 		}
 
 		/* check for delete */
@@ -2588,22 +2619,26 @@
 
 			error = recv_destroy(hdl, fsname, strlen(tofs)+1,
 			    newname, flags);
-			if (error)
-				needagain = B_TRUE;
-			else
+
+			switch (error) {
+			case 0:
 				progress = B_TRUE;
+				break;
+			case EAGAIN:
+				progress = B_TRUE;
+				needagain = B_TRUE;
+				goto out;
+			default:
+				needagain = B_TRUE;
+				break;
+			}
+
 			continue;
 		}
 
-		if (fromguid == 0) {
-			if (flags->verbose) {
-				(void) printf("local fs %s does not have "
-				    "fromsnap (%s in stream); must have "
-				    "been deleted locally; ignoring\n",
-				    fsname, fromsnap);
-			}
+		/* skip destroyed or re-created datasets */
+		if (fromguid == 0)
 			continue;
-		}
 
 		VERIFY(0 == nvlist_lookup_string(stream_nvfs,
 		    "name", &stream_fsname));
@@ -2645,8 +2680,10 @@
 			} else {
 				tryname[0] = '\0';
 				if (flags->verbose) {
-					(void) printf("local fs %s new parent "
-					    "not found\n", fsname);
+					(void) fprintf(stderr,
+					    dgettext(TEXT_DOMAIN,
+					    "parent dataset not found for "
+					    "local dataset '%s'\n"), fsname);
 				}
 			}
 
@@ -2660,13 +2697,14 @@
 				    newname));
 			}
 
-			if (error)
-				needagain = B_TRUE;
-			else
+			if (error == 0)
 				progress = B_TRUE;
+			else
+				needagain = B_TRUE;
 		}
 	}
 
+out:
 	fsavl_destroy(local_avl);
 	nvlist_free(local_nv);
 
@@ -2673,7 +2711,8 @@
 	if (needagain && progress) {
 		/* do another pass to fix up temporary names */
 		if (flags->verbose)
-			(void) printf("another pass:\n");
+			(void) fprintf(stderr, dgettext(TEXT_DOMAIN,
+			    "another pass for promote, destroy and rename:\n"));
 		goto again;
 	}
 
@@ -2682,8 +2721,9 @@
 
 static int
 zfs_receive_package(libzfs_handle_t *hdl, int fd, const char *destname,
-    recvflags_t *flags, dmu_replay_record_t *drr, zio_cksum_t *zc,
-    char **top_zfs, int cleanup_fd, uint64_t *action_handlep)
+    recvflags_t *flags, nvlist_t *exprops, nvlist_t *limitds,
+    dmu_replay_record_t *drr, zio_cksum_t *zc, char **top_zfs, int cleanup_fd,
+    uint64_t *action_handlep)
 {
 	nvlist_t *stream_nv = NULL;
 	avl_tree_t *stream_avl = NULL;
@@ -2764,10 +2804,15 @@
 
 		VERIFY(0 == nvlist_lookup_nvlist(stream_nv, "fss",
 		    &stream_fss));
-		if ((stream_avl = fsavl_create(stream_fss)) == NULL) {
+		error = fsavl_create(stream_fss, &stream_avl);
+		if (error != 0) {
 			zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
 			    "couldn't allocate avl tree"));
-			error = zfs_error(hdl, EZFS_NOMEM, errbuf);
+			if (error == ENOMEM)
+				error = zfs_error(hdl, EZFS_NOMEM, errbuf);
+			else
+				error = zfs_error(hdl, EZFS_BADSTREAM, errbuf);
+
 			goto out;
 		}
 
@@ -2804,7 +2849,7 @@
 			}
 
 			softerr = recv_incremental_replication(hdl, tofs, flags,
-			    stream_nv, stream_avl, renamed);
+			    stream_nv, stream_avl, renamed, limitds);
 
 			/* Unmount renamed filesystems before receiving. */
 			while ((pair = nvlist_next_nvpair(renamed,
@@ -2858,8 +2903,8 @@
 		 * recv_skip() and return 0).
 		 */
 		error = zfs_receive_impl(hdl, destname, NULL, flags, fd,
-		    sendfs, stream_nv, stream_avl, top_zfs, cleanup_fd,
-		    action_handlep, sendsnap);
+		    exprops, limitds, sendfs, stream_nv, stream_avl, top_zfs,
+		    cleanup_fd, action_handlep, sendsnap);
 		if (error == ENODATA) {
 			error = 0;
 			break;
@@ -2873,7 +2918,7 @@
 		 * renames again.
 		 */
 		softerr = recv_incremental_replication(hdl, tofs, flags,
-		    stream_nv, stream_avl, NULL);
+		    stream_nv, stream_avl, NULL, limitds);
 	}
 
 out:
@@ -3023,14 +3068,110 @@
 }
 
 /*
+ * Calculate a list of properties for the current dataset taking into account
+ * stream properties (props) and the properties specified on the command line
+ * using -x and/or -o options (exprops)
+ *
+ * This calculation:
+ * - Removes excluded properties (booleans)
+ * - Changes the values of overridden properties (strings)
+ *
+ */
+static int
+props_override(char *dsname, nvlist_t *props, nvlist_t *exprops,
+    nvlist_t **merged_propsp, recvflags_t *flags, libzfs_handle_t *hdl,
+    zfs_type_t type, uint64_t zoned, zfs_handle_t *zhp,
+    zpool_handle_t *zpool_hdl, const char *errbuf)
+{
+	nvlist_t *goprops, *gxprops, *merged_props, *vprops;
+	nvpair_t *pair;
+	int ret = 0;
+
+	if (nvlist_empty(props) || nvlist_empty(exprops))
+		return (0); /* No properties */
+
+	if (nvlist_dup(props, &merged_props, 0) != 0)
+		return (-1);
+
+	VERIFY(nvlist_alloc(&goprops, NV_UNIQUE_NAME, 0) == 0);
+	VERIFY(nvlist_alloc(&gxprops, NV_UNIQUE_NAME, 0) == 0);
+
+	/* build lists to process in order */
+	for (pair = nvlist_next_nvpair(exprops, NULL); pair != NULL;
+	    pair = nvlist_next_nvpair(exprops, pair)) {
+		const char *propname = nvpair_name(pair);
+		switch (nvpair_type(pair)) {
+		case DATA_TYPE_BOOLEAN:
+			VERIFY0(nvlist_add_nvpair(gxprops, pair));
+			break;
+		case DATA_TYPE_STRING:
+			VERIFY0(nvlist_add_nvpair(goprops, pair));
+			break;
+		default:
+			(void) fprintf(stderr, dgettext(TEXT_DOMAIN,
+			    "property '%s' must be a string or boolean"),
+			    propname);
+			/* should never happen, so assert */
+			assert(B_FALSE);
+		}
+	}
+
+	/* convert override properties e.g. strings to native */
+	if ((vprops = zfs_valid_proplist(hdl, type, goprops, zoned, zhp,
+	    zpool_hdl, errbuf)) == NULL)
+		goto error;
+
+	nvlist_free(goprops);
+	goprops = vprops;
+
+	/* override / set properties */
+	for (nvpair_t *pair = nvlist_next_nvpair(goprops, NULL); pair != NULL;
+	    pair = nvlist_next_nvpair(goprops, pair)) {
+		const char *pname = nvpair_name(pair);
+		if (!nvlist_exists(gxprops, pname)) {
+			if (flags->verbose) {
+				(void) printf("%s %s property from %s\n",
+				    nvlist_exists(merged_props, pname) ?
+				    "overriding" : "setting", pname, dsname);
+			}
+			VERIFY0(nvlist_add_nvpair(merged_props, pair));
+		}
+	}
+
+	/* exclude properties */
+	for (nvpair_t *pair = nvlist_next_nvpair(gxprops, NULL); pair != NULL;
+	    pair = nvlist_next_nvpair(gxprops, pair)) {
+		const char *pname = nvpair_name(pair);
+		if (nvlist_exists(merged_props, pname)) {
+			if (flags->verbose) {
+				(void) printf("excluding %s property "
+				    "from %s\n", pname, dsname);
+			}
+			VERIFY0(nvlist_remove_all(merged_props, pname));
+		}
+	}
+
+	*merged_propsp = merged_props;
+
+error:
+	if (0 != ret)
+		nvlist_free(merged_props);
+	nvlist_free(goprops);
+	nvlist_free(gxprops);
+
+	return (ret);
+}
+
+/*
  * Restores a backup of tosnap from the file descriptor specified by infd.
  */
 static int
 zfs_receive_one(libzfs_handle_t *hdl, int infd, const char *tosnap,
-    const char *originsnap, recvflags_t *flags, dmu_replay_record_t *drr,
-    dmu_replay_record_t *drr_noswap, const char *sendfs, nvlist_t *stream_nv,
-    avl_tree_t *stream_avl, char **top_zfs, int cleanup_fd,
-    uint64_t *action_handlep, const char *finalsnap)
+    const char *originsnap, recvflags_t *flags, nvlist_t *exprops,
+    nvlist_t *limitds, dmu_replay_record_t *drr,
+    dmu_replay_record_t *drr_noswap,
+    const char *sendfs, nvlist_t *stream_nv, avl_tree_t *stream_avl,
+    char **top_zfs, int cleanup_fd, uint64_t *action_handlep, const char *finalsnap)
 {
 	zfs_cmd_t zc = { 0 };
 	time_t begin_time;
@@ -3037,6 +3178,7 @@
 	int ioctl_err, ioctl_errno, err;
 	char *cp;
 	struct drr_begin *drrb = &drr->drr_u.drr_begin;
+	char dsname[ZFS_MAX_DATASET_NAME_LEN];
 	char errbuf[1024];
 	char prop_errbuf[1024];
 	const char *chopprefix;
@@ -3044,9 +3186,9 @@
 	boolean_t stream_wantsnewfs;
 	uint64_t parent_snapguid = 0;
 	prop_changelist_t *clp = NULL;
-	nvlist_t *snapprops_nvlist = NULL;
+	nvlist_t *snapprops_nvlist = NULL, *props = NULL, *merged_props = NULL;
 	zprop_errflags_t prop_errflags;
-	boolean_t recursive;
+	boolean_t recursive, skip;
 	char *snapname = NULL;
 
 	begin_time = time(NULL);
@@ -3058,10 +3200,9 @@
 	    ENOENT);
 
 	if (stream_avl != NULL) {
+		nvlist_t *snapprops;
 		nvlist_t *fs = fsavl_find(stream_avl, drrb->drr_toguid,
 		    &snapname);
-		nvlist_t *props;
-		int ret;
 
 		(void) nvlist_lookup_uint64(fs, "parentfromsnap",
 		    &parent_snapguid);
@@ -3073,17 +3214,16 @@
 			VERIFY(0 == nvlist_add_uint64(props,
 			    zfs_prop_to_name(ZFS_PROP_CANMOUNT), 0));
 		}
-		ret = zcmd_write_src_nvlist(hdl, &zc, props);
-		if (err)
+
+		if (err) {
 			nvlist_free(props);
+			props = NULL;
+		}
 
-		if (0 == nvlist_lookup_nvlist(fs, "snapprops", &props)) {
-			VERIFY(0 == nvlist_lookup_nvlist(props,
+		if (0 == nvlist_lookup_nvlist(fs, "snapprops", &snapprops)) {
+			VERIFY(0 == nvlist_lookup_nvlist(snapprops,
 			    snapname, &snapprops_nvlist));
 		}
-
-		if (ret != 0)
-			return (-1);
 	}
 
 	cp = NULL;
@@ -3192,6 +3332,9 @@
 			(void) printf("found clone origin %s\n", zc.zc_string);
 	}
 
+	(void) strcpy(dsname, drrb->drr_toname);
+	*strchr(dsname, '@') = '\0';
+
 	boolean_t resuming = DMU_GET_FEATUREFLAGS(drrb->drr_versioninfo) &
 	    DMU_BACKUP_FEATURE_RESUMING;
 	stream_wantsnewfs = (drrb->drr_fromguid == NULL ||
@@ -3316,6 +3459,16 @@
 			}
 		}
 
+		/* convert override properties e.g. strings to native */
+		if (!nvlist_empty(exprops) && props_override(dsname, props,
+		    exprops, &merged_props, flags, hdl, zhp->zfs_type,
+		    zfs_prop_get_int(zhp, ZFS_PROP_ZONED), zhp, zhp->zpool_hdl,
+		    errbuf) != 0) {
+			zfs_close(zhp);
+			zcmd_free_nvlists(&zc);
+			return (-1);
+		}
+
 		/*
 		 * If we are resuming a newfs, set newfs here so that we will
 		 * mount it if the recv succeeds this time.  We can tell
@@ -3357,6 +3510,38 @@
 		}
 
 		newfs = B_TRUE;
+
+		if (!nvlist_empty(exprops)) {
+			/* Create an override set of properties if needed */
+			uint64_t zoned = 0;
+			char zp_name[MAXNAMELEN];
+			zpool_handle_t *zp_handle;
+			if (flags->isprefix && !flags->istail &&
+			    !flags->dryrun) {
+				/* Check if we're zoned or not */
+				if (check_parents(hdl, zc.zc_value, &zoned,
+				    B_FALSE, NULL) != 0) {
+					zcmd_free_nvlists(&zc);
+					return (-1);
+				}
+			}
+
+			(void) strlcpy(zp_name, zc.zc_name, sizeof (zp_name));
+			cp = strchr(zp_name, '/');
+			if (cp != NULL)
+				*cp = '\0';
+			zp_handle = zpool_open(hdl, zp_name);
+			if (zp_handle != NULL &&
+			    props_override(dsname, props,exprops,
+			    &merged_props, flags, hdl, ZFS_TYPE_DATASET,
+			    zoned, NULL, zp_handle, errbuf) != 0) {
+				zcmd_free_nvlists(&zc);
+				zpool_close(zp_handle);
+				return (-1);
+			}
+			if (zp_handle != NULL)
+				zpool_close(zp_handle);
+		}
 	}
 
 	zc.zc_begin_record = *drr_noswap;
@@ -3363,15 +3548,17 @@
 	zc.zc_cookie = infd;
 	zc.zc_guid = flags->force;
 	zc.zc_resumable = flags->resumable;
+	skip = !nvlist_empty(limitds) && !nvlist_exists(limitds, dsname);
 	if (flags->verbose) {
 		(void) printf("%s %s stream of %s into %s\n",
-		    flags->dryrun ? "would receive" : "receiving",
+		    skip ? (flags->dryrun ? "would skip" : "skipping") :
+		    (flags->dryrun ? "would receive" : "receiving"),
 		    drrb->drr_fromguid ? "incremental" : "full",
 		    drrb->drr_toname, zc.zc_value);
 		(void) fflush(stdout);
 	}
 
-	if (flags->dryrun) {
+	if (flags->dryrun || skip) {
 		zcmd_free_nvlists(&zc);
 		return (recv_skip(hdl, infd, flags->byteswap));
 	}
@@ -3381,6 +3568,19 @@
 	zc.zc_cleanup_fd = cleanup_fd;
 	zc.zc_action_handle = *action_handlep;
 
+	/*
+	 * if we ended up overriding props, use the merged ones,
+	 * otherwise use the ones that we got from the send stream
+	 */
+	if (merged_props) {
+		if (zcmd_write_src_nvlist(hdl, &zc, merged_props) != 0) {
+			nvlist_free(merged_props);
+			return (-1);
+		}
+		nvlist_free(merged_props);
+	} else if (props && zcmd_write_src_nvlist(hdl, &zc, props) != 0)
+		return (-1);
+
 	err = ioctl_err = zfs_ioctl(hdl, ZFS_IOC_RECV, &zc);
 	ioctl_errno = errno;
 	prop_errflags = (zprop_errflags_t)zc.zc_obj;
@@ -3526,6 +3726,11 @@
 			    "destination %s space quota exceeded"), zc.zc_name);
 			(void) zfs_error(hdl, EZFS_NOSPC, errbuf);
 			break;
+		case EKZFS_WBCNOTSUP:
+			zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
+			    "write back cached datasets do not support recv"));
+			(void) zfs_error(hdl, EZFS_WBCNOTSUP, errbuf);
+			break;
 		default:
 			(void) zfs_standard_error(hdl, ioctl_errno, errbuf);
 		}
@@ -3602,7 +3807,8 @@
 
 static int
 zfs_receive_impl(libzfs_handle_t *hdl, const char *tosnap,
-    const char *originsnap, recvflags_t *flags, int infd, const char *sendfs,
+    const char *originsnap, recvflags_t *flags,
+    int infd, nvlist_t *exprops, nvlist_t *limitds, const char *sendfs,
     nvlist_t *stream_nv, avl_tree_t *stream_avl, char **top_zfs, int cleanup_fd,
     uint64_t *action_handlep, const char *finalsnap)
 {
@@ -3705,13 +3911,14 @@
 			VERIFY(finalsnap == NULL);
 		}
 		return (zfs_receive_one(hdl, infd, tosnap, originsnap, flags,
-		    &drr, &drr_noswap, sendfs, stream_nv, stream_avl, top_zfs,
-		    cleanup_fd, action_handlep, finalsnap));
+		    exprops, limitds, &drr, &drr_noswap, sendfs, stream_nv,
+		    stream_avl, top_zfs, cleanup_fd, action_handlep, finalsnap));
 	} else {
 		assert(DMU_GET_STREAM_HDRTYPE(drrb->drr_versioninfo) ==
 		    DMU_COMPOUNDSTREAM);
-		return (zfs_receive_package(hdl, infd, tosnap, flags, &drr,
-		    &zcksum, top_zfs, cleanup_fd, action_handlep));
+		return (zfs_receive_package(hdl, infd, tosnap, flags, exprops,
+		    limitds, &drr, &zcksum, top_zfs, cleanup_fd,
+		    action_handlep));
 	}
 }
 
@@ -3723,8 +3930,8 @@
  * transfer can be resumed if the sending side supports it).
  */
 int
-zfs_receive(libzfs_handle_t *hdl, const char *tosnap, nvlist_t *props,
-    recvflags_t *flags, int infd, avl_tree_t *stream_avl)
+zfs_receive(libzfs_handle_t *hdl, const char *tosnap, recvflags_t *flags,
+    int infd, nvlist_t *exprops, nvlist_t *limitds, avl_tree_t *stream_avl)
 {
 	char *top_zfs = NULL;
 	int err;
@@ -3731,8 +3938,8 @@
 	int cleanup_fd;
 	uint64_t action_handle = 0;
 	char *originsnap = NULL;
-	if (props) {
-		err = nvlist_lookup_string(props, "origin", &originsnap);
+	if (exprops) {
+		err = nvlist_lookup_string(exprops, "origin", &originsnap);
 		if (err && err != ENOENT)
 			return (err);
 	}
@@ -3740,8 +3947,9 @@
 	cleanup_fd = open(ZFS_DEV, O_RDWR|O_EXCL);
 	VERIFY(cleanup_fd >= 0);
 
-	err = zfs_receive_impl(hdl, tosnap, originsnap, flags, infd, NULL, NULL,
-	    stream_avl, &top_zfs, cleanup_fd, &action_handle, NULL);
+	err = zfs_receive_impl(hdl, tosnap, originsnap, flags, infd, exprops,
+	    limitds, NULL, NULL, stream_avl, &top_zfs, cleanup_fd,
+	    &action_handle, NULL);
 
 	VERIFY(0 == close(cleanup_fd));
 
