Add 'zfs send --saved' flag

This commit adds the --saved (-S) to the 'zfs send' command.
This flag allows a user to send a partially received dataset,
which can be useful when migrating a backup server to new
hardware. This flag is compatible with resumable receives, so
even if the saved send is interrupted, it can be resumed.
The flag does not require any user / kernel ABI changes or any
new feature flags in the send stream format.

Reviewed-by: Paul Dagnelie <[email protected]>
Reviewed-by: Alek Pinchuk <[email protected]>
Reviewed-by: Paul Zuchowski <[email protected]>
Reviewed-by: Christian Schwarz <[email protected]>
Reviewed-by: Matt Ahrens <[email protected]>
Reviewed-by: Brian Behlendorf <[email protected]>
Signed-off-by: Tom Caputi <[email protected]>
Closes #9007 

16 files changed, 501 insertions, 76 deletions

diff --git a/cmd/zfs/zfs_main.c b/cmd/zfs/zfs_main.c
index e2e7f7bb7..2bf2b6cf9 100644
--- a/cmd/zfs/zfs_main.c
+++ b/cmd/zfs/zfs_main.c
@@ -318,7 +318,8 @@ get_usage(zfs_help_t idx)
 		    "<filesystem|volume|snapshot>\n"
 		    "\tsend [-DnPpvLec] [-i bookmark|snapshot] "
 		    "--redact <bookmark> <snapshot>\n"
-		    "\tsend [-nvPe] -t <receive_resume_token>\n"));
+		    "\tsend [-nvPe] -t <receive_resume_token>\n"
+		    "\tsend [-Pnv] --saved filesystem\n"));
 	case HELP_SET:
 		return (gettext("\tset <property=value> ... "
 		    "<filesystem|volume|snapshot> ...\n"));
@@ -4207,11 +4208,12 @@ zfs_do_send(int argc, char **argv)
 		{"raw",		no_argument,		NULL, 'w'},
 		{"backup",	no_argument,		NULL, 'b'},
 		{"holds",	no_argument,		NULL, 'h'},
+		{"saved",	no_argument,		NULL, 'S'},
 		{0, 0, 0, 0}
 	};
 
 	/* check options */
-	while ((c = getopt_long(argc, argv, ":i:I:RDpvnPLeht:cwbd:",
+	while ((c = getopt_long(argc, argv, ":i:I:RDpvnPLeht:cwbd:S",
 	    long_options, NULL)) != -1) {
 		switch (c) {
 		case 'i':
@@ -4271,6 +4273,9 @@ zfs_do_send(int argc, char **argv)
 			flags.embed_data = B_TRUE;
 			flags.largeblock = B_TRUE;
 			break;
+		case 'S':
+			flags.saved = B_TRUE;
+			break;
 		case ':':
 			/*
 			 * If a parameter was not passed, optopt contains the
@@ -4321,7 +4326,7 @@ zfs_do_send(int argc, char **argv)
 	if (resume_token != NULL) {
 		if (fromname != NULL || flags.replicate || flags.props ||
 		    flags.backup || flags.dedup || flags.holds ||
-		    redactbook != NULL) {
+		    flags.saved || redactbook != NULL) {
 			(void) fprintf(stderr,
 			    gettext("invalid flags combined with -t\n"));
 			usage(B_FALSE);
@@ -4342,6 +4347,23 @@ zfs_do_send(int argc, char **argv)
 		}
 	}
 
+	if (flags.saved) {
+		if (fromname != NULL || flags.replicate || flags.props ||
+		    flags.doall || flags.backup || flags.dedup ||
+		    flags.holds || flags.largeblock || flags.embed_data ||
+		    flags.compress || flags.raw || redactbook != NULL) {
+			(void) fprintf(stderr, gettext("incompatible flags "
+			    "combined with saved send flag\n"));
+			usage(B_FALSE);
+		}
+		if (strchr(argv[0], '@') != NULL) {
+			(void) fprintf(stderr, gettext("saved send must "
+			    "specify the dataset with partially-received "
+			    "state\n"));
+			usage(B_FALSE);
+		}
+	}
+
 	if (flags.raw && redactbook != NULL) {
 		(void) fprintf(stderr,
 		    gettext("Error: raw sends may not be redacted.\n"));
@@ -4355,7 +4377,16 @@ zfs_do_send(int argc, char **argv)
 		return (1);
 	}
 
-	if (resume_token != NULL) {
+	if (flags.saved) {
+		zhp = zfs_open(g_zfs, argv[0], ZFS_TYPE_DATASET);
+		if (zhp == NULL)
+			return (1);
+
+		err = zfs_send_saved(zhp, &flags, STDOUT_FILENO,
+		    resume_token);
+		zfs_close(zhp);
+		return (err != 0);
+	} else if (resume_token != NULL) {
 		return (zfs_send_resume(g_zfs, &flags, STDOUT_FILENO,
 		    resume_token));
 	}
diff --git a/include/libzfs.h b/include/libzfs.h
index 0518c800e..8069d4cd4 100644
--- a/include/libzfs.h
+++ b/include/libzfs.h
@@ -677,6 +677,9 @@ typedef struct sendflags {
 
 	/* include snapshot holds in send stream */
 	boolean_t holds;
+
+	/* stream represents a partially received dataset */
+	boolean_t saved;
 } sendflags_t;
 
 typedef boolean_t (snapfilter_cb_t)(zfs_handle_t *, void *);
@@ -688,6 +691,7 @@ extern int zfs_send_one(zfs_handle_t *, const char *, int, sendflags_t *,
 extern int zfs_send_progress(zfs_handle_t *, int, uint64_t *, uint64_t *);
 extern int zfs_send_resume(libzfs_handle_t *, sendflags_t *, int outfd,
     const char *);
+extern int zfs_send_saved(zfs_handle_t *, sendflags_t *, int, const char *);
 extern nvlist_t *zfs_send_resume_token_to_nvlist(libzfs_handle_t *hdl,
     const char *token);
 
diff --git a/include/libzfs_core.h b/include/libzfs_core.h
index bd0b0c4f7..c4b4f8e71 100644
--- a/include/libzfs_core.h
+++ b/include/libzfs_core.h
@@ -79,6 +79,7 @@ enum lzc_send_flags {
 	LZC_SEND_FLAG_LARGE_BLOCK = 1 << 1,
 	LZC_SEND_FLAG_COMPRESS = 1 << 2,
 	LZC_SEND_FLAG_RAW = 1 << 3,
+	LZC_SEND_FLAG_SAVED = 1 << 4,
 };
 
 int lzc_send(const char *, const char *, int, enum lzc_send_flags);
diff --git a/include/sys/dmu_send.h b/include/sys/dmu_send.h
index 2f3dfc39f..d6d050e01 100644
--- a/include/sys/dmu_send.h
+++ b/include/sys/dmu_send.h
@@ -51,14 +51,16 @@ struct dmu_send_outparams;
 int
 dmu_send(const char *tosnap, const char *fromsnap, boolean_t embedok,
     boolean_t large_block_ok, boolean_t compressok, boolean_t rawok,
-    uint64_t resumeobj, uint64_t resumeoff, const char *redactbook, int outfd,
-    offset_t *off, struct dmu_send_outparams *dsop);
+    boolean_t savedok, uint64_t resumeobj, uint64_t resumeoff,
+    const char *redactbook, int outfd, offset_t *off,
+    struct dmu_send_outparams *dsop);
 int dmu_send_estimate_fast(struct dsl_dataset *ds, struct dsl_dataset *fromds,
     zfs_bookmark_phys_t *frombook, boolean_t stream_compressed,
-    uint64_t *sizep);
+    boolean_t saved, uint64_t *sizep);
 int dmu_send_obj(const char *pool, uint64_t tosnap, uint64_t fromsnap,
     boolean_t embedok, boolean_t large_block_ok, boolean_t compressok,
-    boolean_t rawok, int outfd, offset_t *off, struct dmu_send_outparams *dso);
+    boolean_t rawok, boolean_t savedok, int outfd, offset_t *off,
+    struct dmu_send_outparams *dso);
 
 typedef int (*dmu_send_outfunc_t)(objset_t *os, void *buf, int len, void *arg);
 typedef struct dmu_send_outparams {
diff --git a/lib/libzfs/libzfs_sendrecv.c b/lib/libzfs/libzfs_sendrecv.c
index 20d29f48c..de3bfdbf4 100644
--- a/lib/libzfs/libzfs_sendrecv.c
+++ b/lib/libzfs/libzfs_sendrecv.c
@@ -1815,6 +1815,8 @@ lzc_flags_from_sendflags(const sendflags_t *flags)
 		lzc_flags |= LZC_SEND_FLAG_COMPRESS;
 	if (flags->raw)
 		lzc_flags |= LZC_SEND_FLAG_RAW;
+	if (flags->saved)
+		lzc_flags |= LZC_SEND_FLAG_SAVED;
 	return (lzc_flags);
 }
 
@@ -1981,9 +1983,9 @@ find_redact_book(libzfs_handle_t *hdl, const char *path,
 	return (0);
 }
 
-int
-zfs_send_resume(libzfs_handle_t *hdl, sendflags_t *flags, int outfd,
-    const char *resume_token)
+static int
+zfs_send_resume_impl(libzfs_handle_t *hdl, sendflags_t *flags, int outfd,
+    nvlist_t *resume_nvl)
 {
 	char errbuf[1024];
 	char *toname;
@@ -2001,15 +2003,6 @@ zfs_send_resume(libzfs_handle_t *hdl, sendflags_t *flags, int outfd,
 	(void) snprintf(errbuf, sizeof (errbuf), dgettext(TEXT_DOMAIN,
 	    "cannot resume send"));
 
-	nvlist_t *resume_nvl =
-	    zfs_send_resume_token_to_nvlist(hdl, resume_token);
-	if (resume_nvl == NULL) {
-		/*
-		 * zfs_error_aux has already been set by
-		 * zfs_send_resume_token_to_nvlist
-		 */
-		return (zfs_error(hdl, EZFS_FAULT, errbuf));
-	}
 	if (flags->verbosity != 0) {
 		(void) fprintf(fout, dgettext(TEXT_DOMAIN,
 		    "resume token contents:\n"));
@@ -2036,19 +2029,27 @@ zfs_send_resume(libzfs_handle_t *hdl, sendflags_t *flags, int outfd,
 		lzc_flags |= LZC_SEND_FLAG_COMPRESS;
 	if (flags->raw || nvlist_exists(resume_nvl, "rawok"))
 		lzc_flags |= LZC_SEND_FLAG_RAW;
+	if (flags->saved || nvlist_exists(resume_nvl, "savedok"))
+		lzc_flags |= LZC_SEND_FLAG_SAVED;
 
-	if (guid_to_name(hdl, toname, toguid, B_FALSE, name) != 0) {
-		if (zfs_dataset_exists(hdl, toname, ZFS_TYPE_DATASET)) {
-			zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
-			    "'%s' is no longer the same snapshot used in "
-			    "the initial send"), toname);
-		} else {
-			zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
-			    "'%s' used in the initial send no longer exists"),
-			    toname);
+	if (flags->saved) {
+		(void) strcpy(name, toname);
+	} else {
+		error = guid_to_name(hdl, toname, toguid, B_FALSE, name);
+		if (error != 0) {
+			if (zfs_dataset_exists(hdl, toname, ZFS_TYPE_DATASET)) {
+				zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
+				    "'%s' is no longer the same snapshot "
+				    "used in the initial send"), toname);
+			} else {
+				zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
+				    "'%s' used in the initial send no "
+				    "longer exists"), toname);
+			}
+			return (zfs_error(hdl, EZFS_BADPATH, errbuf));
 		}
-		return (zfs_error(hdl, EZFS_BADPATH, errbuf));
 	}
+
 	zhp = zfs_open(hdl, name, ZFS_TYPE_DATASET);
 	if (zhp == NULL) {
 		zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
@@ -2199,6 +2200,122 @@ zfs_send_resume(libzfs_handle_t *hdl, sendflags_t *flags, int outfd,
 	return (error);
 }
 
+int
+zfs_send_resume(libzfs_handle_t *hdl, sendflags_t *flags, int outfd,
+    const char *resume_token)
+{
+	int ret;
+	char errbuf[1024];
+	nvlist_t *resume_nvl;
+
+	(void) snprintf(errbuf, sizeof (errbuf), dgettext(TEXT_DOMAIN,
+	    "cannot resume send"));
+
+	resume_nvl = zfs_send_resume_token_to_nvlist(hdl, resume_token);
+	if (resume_nvl == NULL) {
+		/*
+		 * zfs_error_aux has already been set by
+		 * zfs_send_resume_token_to_nvlist()
+		 */
+		return (zfs_error(hdl, EZFS_FAULT, errbuf));
+	}
+
+	ret = zfs_send_resume_impl(hdl, flags, outfd, resume_nvl);
+	nvlist_free(resume_nvl);
+
+	return (ret);
+}
+
+int
+zfs_send_saved(zfs_handle_t *zhp, sendflags_t *flags, int outfd,
+    const char *resume_token)
+{
+	int ret;
+	libzfs_handle_t *hdl = zhp->zfs_hdl;
+	nvlist_t *saved_nvl = NULL, *resume_nvl = NULL;
+	uint64_t saved_guid = 0, resume_guid = 0;
+	uint64_t obj = 0, off = 0, bytes = 0;
+	char token_buf[ZFS_MAXPROPLEN];
+	char errbuf[1024];
+
+	(void) snprintf(errbuf, sizeof (errbuf), dgettext(TEXT_DOMAIN,
+	    "saved send failed"));
+
+	ret = zfs_prop_get(zhp, ZFS_PROP_RECEIVE_RESUME_TOKEN,
+	    token_buf, sizeof (token_buf), NULL, NULL, 0, B_TRUE);
+	if (ret != 0)
+		goto out;
+
+	saved_nvl = zfs_send_resume_token_to_nvlist(hdl, token_buf);
+	if (saved_nvl == NULL) {
+		/*
+		 * zfs_error_aux has already been set by
+		 * zfs_send_resume_token_to_nvlist()
+		 */
+		ret = zfs_error(hdl, EZFS_FAULT, errbuf);
+		goto out;
+	}
+
+	/*
+	 * If a resume token is provided we use the object and offset
+	 * from that instead of the default, which starts from the
+	 * beginning.
+	 */
+	if (resume_token != NULL) {
+		resume_nvl = zfs_send_resume_token_to_nvlist(hdl,
+		    resume_token);
+		if (resume_nvl == NULL) {
+			ret = zfs_error(hdl, EZFS_FAULT, errbuf);
+			goto out;
+		}
+
+		if (nvlist_lookup_uint64(resume_nvl, "object", &obj) != 0 ||
+		    nvlist_lookup_uint64(resume_nvl, "offset", &off) != 0 ||
+		    nvlist_lookup_uint64(resume_nvl, "bytes", &bytes) != 0 ||
+		    nvlist_lookup_uint64(resume_nvl, "toguid",
+		    &resume_guid) != 0) {
+			zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
+			    "provided resume token is corrupt"));
+			ret = zfs_error(hdl, EZFS_FAULT, errbuf);
+			goto out;
+		}
+
+		if (nvlist_lookup_uint64(saved_nvl, "toguid",
+		    &saved_guid)) {
+			zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
+			    "dataset's resume token is corrupt"));
+			ret = zfs_error(hdl, EZFS_FAULT, errbuf);
+			goto out;
+		}
+
+		if (resume_guid != saved_guid) {
+			zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
+			    "provided resume token does not match dataset"));
+			ret = zfs_error(hdl, EZFS_BADBACKUP, errbuf);
+			goto out;
+		}
+	}
+
+	(void) nvlist_remove_all(saved_nvl, "object");
+	fnvlist_add_uint64(saved_nvl, "object", obj);
+
+	(void) nvlist_remove_all(saved_nvl, "offset");
+	fnvlist_add_uint64(saved_nvl, "offset", off);
+
+	(void) nvlist_remove_all(saved_nvl, "bytes");
+	fnvlist_add_uint64(saved_nvl, "bytes", bytes);
+
+	(void) nvlist_remove_all(saved_nvl, "toname");
+	fnvlist_add_string(saved_nvl, "toname", zhp->zfs_name);
+
+	ret = zfs_send_resume_impl(hdl, flags, outfd, saved_nvl);
+
+out:
+	nvlist_free(saved_nvl);
+	nvlist_free(resume_nvl);
+	return (ret);
+}
+
 /*
  * This function informs the target system that the recursive send is complete.
  * The record is also expected in the case of a send -p.
diff --git a/lib/libzfs_core/libzfs_core.c b/lib/libzfs_core/libzfs_core.c
index 2f31edcc2..5dd8976eb 100644
--- a/lib/libzfs_core/libzfs_core.c
+++ b/lib/libzfs_core/libzfs_core.c
@@ -675,6 +675,8 @@ lzc_send_resume_redacted(const char *snapname, const char *from, int fd,
 		fnvlist_add_boolean(args, "compressok");
 	if (flags & LZC_SEND_FLAG_RAW)
 		fnvlist_add_boolean(args, "rawok");
+	if (flags & LZC_SEND_FLAG_SAVED)
+		fnvlist_add_boolean(args, "savedok");
 	if (resumeobj != 0 || resumeoff != 0) {
 		fnvlist_add_uint64(args, "resume_object", resumeobj);
 		fnvlist_add_uint64(args, "resume_offset", resumeoff);
diff --git a/man/man8/zfs-send.8 b/man/man8/zfs-send.8
index 0171166eb..2561cb61c 100644
--- a/man/man8/zfs-send.8
+++ b/man/man8/zfs-send.8
@@ -60,6 +60,10 @@
 .Fl t
 .Ar receive_resume_token
 .Nm
+.Cm send
+.Op Fl Pnv
+.Fl S Ar filesystem
+.Nm
 .Cm redact
 .Ar snapshot redaction_bookmark
 .Ar redaction_snapshot Ns ...
@@ -503,6 +507,23 @@ See the documentation for
 for more details.
 .It Xo
 .Nm
+.Cm send
+.Op Fl Pnv
+.Op Fl i Ar snapshot Ns | Ns Ar bookmark
+.Fl S
+.Ar filesystem
+.Xc
+Generate a send stream from a dataset that has been partially received.
+.Bl -tag -width "-L"
+.It Fl S, -saved
+This flag requires that the specified filesystem previously received a resumable
+send that did not finish and was interrupted. In such scenarios this flag
+enables the user to send this partially received state. Using this flag will
+always use the last fully received snapshot as the incremental source if it
+exists.
+.El
+.It Xo
+.Nm
 .Cm redact
 .Ar snapshot redaction_bookmark
 .Ar redaction_snapshot Ns ...
diff --git a/module/zfs/dmu_recv.c b/module/zfs/dmu_recv.c
index f68419bfa..6f3545b7e 100644
--- a/module/zfs/dmu_recv.c
+++ b/module/zfs/dmu_recv.c
@@ -887,7 +887,6 @@ dmu_recv_begin_sync(void *arg, dmu_tx_t *tx)
 		drba->drba_cookie->drc_raw = B_TRUE;
 	}
 
-
 	if (featureflags & DMU_BACKUP_FEATURE_REDACTED) {
 		uint64_t *redact_snaps;
 		uint_t numredactsnaps;
diff --git a/module/zfs/dmu_send.c b/module/zfs/dmu_send.c
index b0a56650e..62de978d3 100644
--- a/module/zfs/dmu_send.c
+++ b/module/zfs/dmu_send.c
@@ -51,6 +51,7 @@
 #include <sys/ddt.h>
 #include <sys/zfs_onexit.h>
 #include <sys/dmu_send.h>
+#include <sys/dmu_recv.h>
 #include <sys/dsl_destroy.h>
 #include <sys/blkptr.h>
 #include <sys/dsl_bookmark.h>
@@ -1888,8 +1889,11 @@ struct dmu_send_params {
 	boolean_t embedok;
 	boolean_t large_block_ok;
 	boolean_t compressok;
+	boolean_t rawok;
+	boolean_t savedok;
 	uint64_t resumeobj;
 	uint64_t resumeoff;
+	uint64_t saved_guid;
 	zfs_bookmark_phys_t *redactbook;
 	/* Stream output params */
 	dmu_send_outparams_t *dso;
@@ -1897,7 +1901,7 @@ struct dmu_send_params {
 	/* Stream progress params */
 	offset_t *off;
 	int outfd;
-	boolean_t rawok;
+	char saved_toname[MAXNAMELEN];
 };
 
 static int
@@ -1984,10 +1988,15 @@ create_begin_record(struct dmu_send_params *dspp, objset_t *os,
 		drrb->drr_flags |= DRR_FLAG_FREERECORDS;
 	drr->drr_u.drr_begin.drr_flags |= DRR_FLAG_SPILL_BLOCK;
 
-	dsl_dataset_name(to_ds, drrb->drr_toname);
-	if (!to_ds->ds_is_snapshot) {
-		(void) strlcat(drrb->drr_toname, "@--head--",
-		    sizeof (drrb->drr_toname));
+	if (dspp->savedok) {
+		drrb->drr_toguid = dspp->saved_guid;
+		strcpy(drrb->drr_toname, dspp->saved_toname);
+	} else {
+		dsl_dataset_name(to_ds, drrb->drr_toname);
+		if (!to_ds->ds_is_snapshot) {
+			(void) strlcat(drrb->drr_toname, "@--head--",
+			    sizeof (drrb->drr_toname));
+		}
 	}
 	return (drr);
 }
@@ -2305,6 +2314,7 @@ dmu_send_impl(struct dmu_send_params *dspp)
 		dsl_pool_rele(dp, tag);
 		return (err);
 	}
+
 	/*
 	 * If this is a non-raw send of an encrypted ds, we can ensure that
 	 * the objset_phys_t is authenticated. This is safe because this is
@@ -2526,19 +2536,27 @@ dmu_send_impl(struct dmu_send_params *dspp)
 		goto out;
 	}
 
-	bzero(drr, sizeof (dmu_replay_record_t));
-	drr->drr_type = DRR_END;
-	drr->drr_u.drr_end.drr_checksum = dsc.dsc_zc;
-	drr->drr_u.drr_end.drr_toguid = dsc.dsc_toguid;
+	/*
+	 * Send the DRR_END record if this is not a saved stream.
+	 * Otherwise, the omitted DRR_END record will signal to
+	 * the receive side that the stream is incomplete.
+	 */
+	if (!dspp->savedok) {
+		bzero(drr, sizeof (dmu_replay_record_t));
+		drr->drr_type = DRR_END;
+		drr->drr_u.drr_end.drr_checksum = dsc.dsc_zc;
+		drr->drr_u.drr_end.drr_toguid = dsc.dsc_toguid;
 
-	if (dump_record(&dsc, NULL, 0) != 0)
-		err = dsc.dsc_err;
+		if (dump_record(&dsc, NULL, 0) != 0)
+			err = dsc.dsc_err;
+	}
 out:
 	mutex_enter(&to_ds->ds_sendstream_lock);
 	list_remove(&to_ds->ds_sendstreams, dssp);
 	mutex_exit(&to_ds->ds_sendstream_lock);
 
-	VERIFY(err != 0 || (dsc.dsc_sent_begin && dsc.dsc_sent_end));
+	VERIFY(err != 0 || (dsc.dsc_sent_begin &&
+	    (dsc.dsc_sent_end || dspp->savedok)));
 
 	kmem_free(drr, sizeof (dmu_replay_record_t));
 	kmem_free(dssp, sizeof (dmu_sendstatus_t));
@@ -2564,7 +2582,8 @@ out:
 int
 dmu_send_obj(const char *pool, uint64_t tosnap, uint64_t fromsnap,
     boolean_t embedok, boolean_t large_block_ok, boolean_t compressok,
-    boolean_t rawok, int outfd, offset_t *off, dmu_send_outparams_t *dsop)
+    boolean_t rawok, boolean_t savedok, int outfd, offset_t *off,
+    dmu_send_outparams_t *dsop)
 {
 	int err;
 	dsl_dataset_t *fromds;
@@ -2578,6 +2597,7 @@ dmu_send_obj(const char *pool, uint64_t tosnap, uint64_t fromsnap,
 	dspp.dso = dsop;
 	dspp.tag = FTAG;
 	dspp.rawok = rawok;
+	dspp.savedok = savedok;
 
 	err = dsl_pool_hold(pool, FTAG, &dspp.dp);
 	if (err != 0)
@@ -2644,8 +2664,9 @@ dmu_send_obj(const char *pool, uint64_t tosnap, uint64_t fromsnap,
 int
 dmu_send(const char *tosnap, const char *fromsnap, boolean_t embedok,
     boolean_t large_block_ok, boolean_t compressok, boolean_t rawok,
-    uint64_t resumeobj, uint64_t resumeoff, const char *redactbook, int outfd,
-    offset_t *off, dmu_send_outparams_t *dsop)
+    boolean_t savedok, uint64_t resumeobj, uint64_t resumeoff,
+    const char *redactbook, int outfd, offset_t *off,
+    dmu_send_outparams_t *dsop)
 {
 	int err = 0;
 	ds_hold_flags_t dsflags = (rawok) ? 0 : DS_HOLD_FLAG_DECRYPT;
@@ -2653,6 +2674,7 @@ dmu_send(const char *tosnap, const char *fromsnap, boolean_t embedok,
 	dsl_dataset_t *fromds = NULL;
 	zfs_bookmark_phys_t book = {0};
 	struct dmu_send_params dspp = {0};
+
 	dspp.tosnap = tosnap;
 	dspp.embedok = embedok;
 	dspp.large_block_ok = large_block_ok;
@@ -2664,6 +2686,7 @@ dmu_send(const char *tosnap, const char *fromsnap, boolean_t embedok,
 	dspp.resumeobj = resumeobj;
 	dspp.resumeoff = resumeoff;
 	dspp.rawok = rawok;
+	dspp.savedok = savedok;
 
 	if (fromsnap != NULL && strpbrk(fromsnap, "@#") == NULL)
 		return (SET_ERROR(EINVAL));
@@ -2671,13 +2694,57 @@ dmu_send(const char *tosnap, const char *fromsnap, boolean_t embedok,
 	err = dsl_pool_hold(tosnap, FTAG, &dspp.dp);
 	if (err != 0)
 		return (err);
+
 	if (strchr(tosnap, '@') == NULL && spa_writeable(dspp.dp->dp_spa)) {
 		/*
 		 * We are sending a filesystem or volume.  Ensure
 		 * that it doesn't change by owning the dataset.
 		 */
-		err = dsl_dataset_own(dspp.dp, tosnap, dsflags, FTAG,
-		    &dspp.to_ds);
+
+		if (savedok) {
+			/*
+			 * We are looking for the dataset that represents the
+			 * partially received send stream. If this stream was
+			 * received as a new snapshot of an existing dataset,
+			 * this will be saved in a hidden clone named
+			 * "<pool>/<dataset>/%recv". Otherwise, the stream
+			 * will be saved in the live dataset itself. In
+			 * either case we need to use dsl_dataset_own_force()
+			 * because the stream is marked as inconsistent,
+			 * which would normally make it unavailable to be
+			 * owned.
+			 */
+			char *name = kmem_asprintf("%s/%s", tosnap,
+			    recv_clone_name);
+			err = dsl_dataset_own_force(dspp.dp, name, dsflags,
+			    FTAG, &dspp.to_ds);
+			if (err == ENOENT) {
+				err = dsl_dataset_own_force(dspp.dp, tosnap,
+				    dsflags, FTAG, &dspp.to_ds);
+			}
+
+			if (err == 0) {
+				err = zap_lookup(dspp.dp->dp_meta_objset,
+				    dspp.to_ds->ds_object,
+				    DS_FIELD_RESUME_TOGUID, 8, 1,
+				    &dspp.saved_guid);
+			}
+
+			if (err == 0) {
+				err = zap_lookup(dspp.dp->dp_meta_objset,
+				    dspp.to_ds->ds_object,
+				    DS_FIELD_RESUME_TONAME, 1,
+				    sizeof (dspp.saved_toname),
+				    dspp.saved_toname);
+			}
+			if (err != 0)
+				dsl_dataset_disown(dspp.to_ds, dsflags, FTAG);
+
+			kmem_strfree(name);
+		} else {
+			err = dsl_dataset_own(dspp.dp, tosnap, dsflags,
+			    FTAG, &dspp.to_ds);
+		}
 		owned = B_TRUE;
 	} else {
 		err = dsl_dataset_hold_flags(dspp.dp, tosnap, dsflags, FTAG,
@@ -2763,9 +2830,6 @@ dmu_send(const char *tosnap, const char *fromsnap, boolean_t embedok,
 				    0)) {
 					err = SET_ERROR(EXDEV);
 				} else {
-					ASSERT3U(dspp.is_clone, ==,
-					    (dspp.to_ds->ds_dir !=
-					    fromds->ds_dir));
 					zb->zbm_creation_txg =
 					    dsl_dataset_phys(fromds)->
 					    ds_creation_txg;
@@ -2870,37 +2934,80 @@ dmu_adjust_send_estimate_for_indirects(dsl_dataset_t *ds, uint64_t uncompressed,
 }
 
 int
-dmu_send_estimate_fast(dsl_dataset_t *ds, dsl_dataset_t *fromds,
-    zfs_bookmark_phys_t *frombook, boolean_t stream_compressed, uint64_t *sizep)
+dmu_send_estimate_fast(dsl_dataset_t *origds, dsl_dataset_t *fromds,
+    zfs_bookmark_phys_t *frombook, boolean_t stream_compressed,
+    boolean_t saved, uint64_t *sizep)
 {
 	int err;
+	dsl_dataset_t *ds = origds;
 	uint64_t uncomp, comp;
 
-	ASSERT(dsl_pool_config_held(ds->ds_dir->dd_pool));
+	ASSERT(dsl_pool_config_held(origds->ds_dir->dd_pool));
 	ASSERT(fromds == NULL || frombook == NULL);
 
-	/* tosnap must be a snapshot */
-	if (!ds->ds_is_snapshot)
+	/*
+	 * If this is a saved send we may actually be sending
+	 * from the %recv clone used for resuming.
+	 */
+	if (saved) {
+		objset_t *mos = origds->ds_dir->dd_pool->dp_meta_objset;
+		uint64_t guid;
+		char dsname[ZFS_MAX_DATASET_NAME_LEN + 6];
+
+		dsl_dataset_name(origds, dsname);
+		(void) strcat(dsname, "/");
+		(void) strcat(dsname, recv_clone_name);
+
+		err = dsl_dataset_hold(origds->ds_dir->dd_pool,
+		    dsname, FTAG, &ds);
+		if (err != ENOENT && err != 0) {
+			return (err);
+		} else if (err == ENOENT) {
+			ds = origds;
+		}
+
+		/* check that this dataset has partially received data */
+		err = zap_lookup(mos, ds->ds_object,
+		    DS_FIELD_RESUME_TOGUID, 8, 1, &guid);
+		if (err != 0) {
+			err = SET_ERROR(err == ENOENT ? EINVAL : err);
+			goto out;
+		}
+
+		err = zap_lookup(mos, ds->ds_object,
+		    DS_FIELD_RESUME_TONAME, 1, sizeof (dsname), dsname);
+		if (err != 0) {
+			err = SET_ERROR(err == ENOENT ? EINVAL : err);
+			goto out;
+		}
+	}
+
+	/* tosnap must be a snapshot or the target of a saved send */
+	if (!ds->ds_is_snapshot && ds == origds)
 		return (SET_ERROR(EINVAL));
 
 	if (fromds != NULL) {
 		uint64_t used;
-		if (!fromds->ds_is_snapshot)
-			return (SET_ERROR(EINVAL));
+		if (!fromds->ds_is_snapshot) {
+			err = SET_ERROR(EINVAL);
+			goto out;
+		}
 
-		if (!dsl_dataset_is_before(ds, fromds, 0))
-			return (SET_ERROR(EXDEV));
+		if (!dsl_dataset_is_before(ds, fromds, 0)) {
+			err = SET_ERROR(EXDEV);
+			goto out;
+		}
 
 		err = dsl_dataset_space_written(fromds, ds, &used, &comp,
 		    &uncomp);
 		if (err != 0)
-			return (err);
+			goto out;
 	} else if (frombook != NULL) {
 		uint64_t used;
 		err = dsl_dataset_space_written_bookmark(frombook, ds, &used,
 		    &comp, &uncomp);
 		if (err != 0)
-			return (err);
+			goto out;
 	} else {
 		uncomp = dsl_dataset_phys(ds)->ds_uncompressed_bytes;
 		comp = dsl_dataset_phys(ds)->ds_compressed_bytes;
@@ -2912,6 +3019,10 @@ dmu_send_estimate_fast(dsl_dataset_t *ds, dsl_dataset_t *fromds,
 	 * Add the size of the BEGIN and END records to the estimate.
 	 */
 	*sizep += 2 * sizeof (dmu_replay_record_t);
+
+out:
+	if (ds != origds)
+		dsl_dataset_rele(ds, FTAG);
 	return (err);
 }
 
diff --git a/module/zfs/zfs_ioctl.c b/module/zfs/zfs_ioctl.c
index 2c1f0aab6..fb8034f70 100644
--- a/module/zfs/zfs_ioctl.c
+++ b/module/zfs/zfs_ioctl.c
@@ -5291,6 +5291,7 @@ zfs_ioc_send(zfs_cmd_t *zc)
 	boolean_t large_block_ok = (zc->zc_flags & 0x2);
 	boolean_t compressok = (zc->zc_flags & 0x4);
 	boolean_t rawok = (zc->zc_flags & 0x8);
+	boolean_t savedok = (zc->zc_flags & 0x10);
 
 	if (zc->zc_obj != 0) {
 		dsl_pool_t *dp;
@@ -5340,7 +5341,7 @@ zfs_ioc_send(zfs_cmd_t *zc)
 		}
 
 		error = dmu_send_estimate_fast(tosnap, fromsnap, NULL,
-		    compressok || rawok, &zc->zc_objset_type);
+		    compressok || rawok, savedok, &zc->zc_objset_type);
 
 		if (fromsnap != NULL)
 			dsl_dataset_rele(fromsnap, FTAG);
@@ -5358,8 +5359,8 @@ zfs_ioc_send(zfs_cmd_t *zc)
 		out.dso_arg = fp;
 		out.dso_dryrun = B_FALSE;
 		error = dmu_send_obj(zc->zc_name, zc->zc_sendobj,
-		    zc->zc_fromobj, embedok, large_block_ok, compressok, rawok,
-		    zc->zc_cookie, &off, &out);
+		    zc->zc_fromobj, embedok, large_block_ok, compressok,
+		    rawok, savedok, zc->zc_cookie, &off, &out);
 
 		zfs_file_put(zc->zc_cookie);
 	}
@@ -6245,6 +6246,8 @@ zfs_ioc_space_snaps(const char *lastsnap, nvlist_t *innvl, nvlist_t *outnvl)
  *         presence indicates compressed DRR_WRITE records are permitted
  *     (optional) "rawok" -> (value ignored)
  *         presence indicates raw encrypted records should be used.
+ *     (optional) "savedok" -> (value ignored)
+ *         presence indicates we should send a partially received snapshot
  *     (optional) "resume_object" and "resume_offset" -> (uint64)
  *         if present, resume send stream from specified object and offset.
  *     (optional) "redactbook" -> (string)
@@ -6261,6 +6264,7 @@ static const zfs_ioc_key_t zfs_keys_send_new[] = {
 	{"embedok",		DATA_TYPE_BOOLEAN,	ZK_OPTIONAL},
 	{"compressok",		DATA_TYPE_BOOLEAN,	ZK_OPTIONAL},
 	{"rawok",		DATA_TYPE_BOOLEAN,	ZK_OPTIONAL},
+	{"savedok",		DATA_TYPE_BOOLEAN,	ZK_OPTIONAL},
 	{"resume_object",	DATA_TYPE_UINT64,	ZK_OPTIONAL},
 	{"resume_offset",	DATA_TYPE_UINT64,	ZK_OPTIONAL},
 	{"redactbook",		DATA_TYPE_STRING,	ZK_OPTIONAL},
@@ -6279,6 +6283,7 @@ zfs_ioc_send_new(const char *snapname, nvlist_t *innvl, nvlist_t *outnvl)
 	boolean_t embedok;
 	boolean_t compressok;
 	boolean_t rawok;
+	boolean_t savedok;
 	uint64_t resumeobj = 0;
 	uint64_t resumeoff = 0;
 	char *redactbook = NULL;
@@ -6291,6 +6296,7 @@ zfs_ioc_send_new(const char *snapname, nvlist_t *innvl, nvlist_t *outnvl)
 	embedok = nvlist_exists(innvl, "embedok");
 	compressok = nvlist_exists(innvl, "compressok");
 	rawok = nvlist_exists(innvl, "rawok");
+	savedok = nvlist_exists(innvl, "savedok");
 
 	(void) nvlist_lookup_uint64(innvl, "resume_object", &resumeobj);
 	(void) nvlist_lookup_uint64(innvl, "resume_offset", &resumeoff);
@@ -6306,8 +6312,9 @@ zfs_ioc_send_new(const char *snapname, nvlist_t *innvl, nvlist_t *outnvl)
 	out.dso_outfunc = dump_bytes;
 	out.dso_arg = fp;
 	out.dso_dryrun = B_FALSE;
-	error = dmu_send(snapname, fromname, embedok, largeblockok, compressok,
-	    rawok, resumeobj, resumeoff, redactbook, fd, &off, &out);
+	error = dmu_send(snapname, fromname, embedok, largeblockok,
+	    compressok, rawok, savedok, resumeobj, resumeoff,
+	    redactbook, fd, &off, &out);
 
 	zfs_file_put(fd);
 	return (error);
@@ -6372,6 +6379,7 @@ zfs_ioc_send_space(const char *snapname, nvlist_t *innvl, nvlist_t *outnvl)
 	boolean_t embedok;
 	boolean_t compressok;
 	boolean_t rawok;
+	boolean_t savedok;
 	uint64_t space = 0;
 	boolean_t full_estimate = B_FALSE;
 	uint64_t resumeobj = 0;
@@ -6395,6 +6403,7 @@ zfs_ioc_send_space(const char *snapname, nvlist_t *innvl, nvlist_t *outnvl)
 	embedok = nvlist_exists(innvl, "embedok");
 	compressok = nvlist_exists(innvl, "compressok");
 	rawok = nvlist_exists(innvl, "rawok");
+	savedok = nvlist_exists(innvl, "savedok");
 	boolean_t from = (nvlist_lookup_string(innvl, "from", &fromname) == 0);
 	boolean_t altbook = (nvlist_lookup_string(innvl, "redactbook",
 	    &redactlist_book) == 0);
@@ -6469,12 +6478,12 @@ zfs_ioc_send_space(const char *snapname, nvlist_t *innvl, nvlist_t *outnvl)
 		dsl_dataset_rele(tosnap, FTAG);
 		dsl_pool_rele(dp, FTAG);
 		error = dmu_send(snapname, fromname, embedok, largeblockok,
-		    compressok, rawok, resumeobj, resumeoff, redactlist_book,
-		    fd, &off, &out);
+		    compressok, rawok, savedok, resumeobj, resumeoff,
+		    redactlist_book, fd, &off, &out);
 	} else {
 		error = dmu_send_estimate_fast(tosnap, fromsnap,
 		    (from && strchr(fromname, '#') != NULL ? &zbm : NULL),
-		    compressok || rawok, &space);
+		    compressok || rawok, savedok, &space);
 		space -= resume_bytes;
 		if (fromsnap != NULL)
 			dsl_dataset_rele(fromsnap, FTAG);
diff --git a/tests/runfiles/common.run b/tests/runfiles/common.run
index 2c227875b..50330884d 100644
--- a/tests/runfiles/common.run
+++ b/tests/runfiles/common.run
@@ -786,7 +786,8 @@ tests = ['rsend_001_pos', 'rsend_002_pos', 'rsend_003_pos', 'rsend_004_pos',
     'send_encrypted_props', 'send_encrypted_truncated_files',
     'send_freeobjects', 'send_realloc_files',
     'send_realloc_encrypted_files', 'send_spill_block', 'send_holds',
-    'send_hole_birth', 'send_mixed_raw', 'send-wDR_encrypted_zvol']
+    'send_hole_birth', 'send_mixed_raw', 'send-wDR_encrypted_zvol',
+    'send_partial_dataset']
 tags = ['functional', 'rsend']
 
 [tests/functional/scrub_mirror]
diff --git a/tests/zfs-tests/cmd/libzfs_input_check/libzfs_input_check.c b/tests/zfs-tests/cmd/libzfs_input_check/libzfs_input_check.c
index 294c055f2..1a02aa515 100644
--- a/tests/zfs-tests/cmd/libzfs_input_check/libzfs_input_check.c
+++ b/tests/zfs-tests/cmd/libzfs_input_check/libzfs_input_check.c
@@ -508,6 +508,7 @@ test_send_new(const char *snapshot, int fd)
 	fnvlist_add_string(optional, "fromsnap", from);
 	fnvlist_add_uint64(optional, "resume_object", resumeobj);
 	fnvlist_add_uint64(optional, "resume_offset", offset);
+	fnvlist_add_boolean(optional, "savedok");
 #endif
 	IOC_INPUT_TEST(ZFS_IOC_SEND_NEW, snapshot, required, optional, 0);
 
diff --git a/tests/zfs-tests/tests/functional/redacted_send/redacted_resume.ksh b/tests/zfs-tests/tests/functional/redacted_send/redacted_resume.ksh
index 9a3c5329d..7e043e176 100755
--- a/tests/zfs-tests/tests/functional/redacted_send/redacted_resume.ksh
+++ b/tests/zfs-tests/tests/functional/redacted_send/redacted_resume.ksh
@@ -73,7 +73,7 @@ log_must eval "get_diff $send_mnt/f3 $recv_mnt/f3 >$tmpdir/get_diff.out"
 range=$(cat $tmpdir/get_diff.out)
 [[ "$RANGE10" = "$range" ]] || log_fail "Unexpected range: $range"
 
-# Test recv -A works properly
+# Test recv -A works properly and verify saved sends are not allowed
 log_mustnot zfs recv -A $recvfs
 log_must zfs destroy -R $recvfs
 log_mustnot zfs recv -A $recvfs
@@ -81,6 +81,7 @@ log_must eval "zfs send --redact book1 $sendfs@snap >$stream"
 dd if=$stream bs=64k count=1 | log_mustnot zfs receive -s $recvfs
 [[ "-" = $(get_prop receive_resume_token $recvfs) ]] && \
     log_fail "Receive token not found."
+log_mustnot eval "zfs send --saved --redact book1 $recvfs > /dev/null"
 log_must zfs recv -A $recvfs
 log_must datasetnonexists $recvfs
 
diff --git a/tests/zfs-tests/tests/functional/rsend/Makefile.am b/tests/zfs-tests/tests/functional/rsend/Makefile.am
index 585018ac2..8e16f8847 100644
--- a/tests/zfs-tests/tests/functional/rsend/Makefile.am
+++ b/tests/zfs-tests/tests/functional/rsend/Makefile.am
@@ -41,6 +41,7 @@ dist_pkgdata_SCRIPTS = \
 	send-c_zstreamdump.ksh \
 	send-cpL_varied_recsize.ksh \
 	send_freeobjects.ksh \
+	send_partial_dataset.ksh \
 	send_realloc_dnode_size.ksh \
 	send_realloc_files.ksh \
 	send_realloc_encrypted_files.ksh \
diff --git a/tests/zfs-tests/tests/functional/rsend/rsend.kshlib b/tests/zfs-tests/tests/functional/rsend/rsend.kshlib
index 8ffb133af..3961dbdcc 100644
--- a/tests/zfs-tests/tests/functional/rsend/rsend.kshlib
+++ b/tests/zfs-tests/tests/functional/rsend/rsend.kshlib
@@ -563,17 +563,31 @@ function churn_files
 }
 
 #
-# Mess up file contents
+# Mess up a send file's contents
 #
-# $1 The file path
+# $1 The send file path
 #
-function mess_file
+function mess_send_file
 {
 	file=$1
 
 	filesize=$(stat_size $file)
 
 	offset=$(($RANDOM * $RANDOM % $filesize))
+
+	# The random offset might truncate the send stream to be
+	# smaller than the DRR_BEGIN record. If this happens, then
+	# the receiving system won't have enough info to create the
+	# partial dataset at all. We use zstreamdump to check for
+	# this and retry in this case.
+	nr_begins=$(head -c $offset $file | zstreamdump | \
+	    grep DRR_BEGIN | awk '{ print $5 }')
+	while [ "$nr_begins" -eq 0 ]; do
+		offset=$(($RANDOM * $RANDOM % $filesize))
+		nr_begins=$(head -c $offset $file | zstreamdump | \
+		    grep DRR_BEGIN | awk '{ print $5 }')
+	done
+
 	if (($RANDOM % 7 <= 1)); then
 		#
 		# We corrupt 2 bytes to minimize the chance that we
@@ -626,7 +640,7 @@ function resume_test
 	log_must eval "$sendcmd >/$streamfs/$stream_num"
 
 	for ((i=0; i<2; i=i+1)); do
-		mess_file /$streamfs/$stream_num
+		mess_send_file /$streamfs/$stream_num
 		log_mustnot zfs recv -suv $recvfs </$streamfs/$stream_num
 		stream_num=$((stream_num+1))
 
diff --git a/tests/zfs-tests/tests/functional/rsend/send_partial_dataset.ksh b/tests/zfs-tests/tests/functional/rsend/send_partial_dataset.ksh
new file mode 100755
index 000000000..8a0971596
--- /dev/null
+++ b/tests/zfs-tests/tests/functional/rsend/send_partial_dataset.ksh
@@ -0,0 +1,110 @@
+#!/bin/ksh
+
+#
+# This file and its contents are supplied under the terms of the
+# Common Development and Distribution License ("CDDL"), version a.0.
+# You may only use this file in accordance with the terms of version
+# a.0 of the CDDL.
+#
+# A full copy of the text of the CDDL should have accompanied this
+# source.  A copy of the CDDL is also available via the Internet at
+# http://www.illumos.org/license/CDDL.
+#
+
+#
+# Copyright (c) 2019 Datto Inc.
+#
+
+. $STF_SUITE/include/libtest.shlib
+. $STF_SUITE/tests/functional/rsend/rsend.kshlib
+
+#
+# Description:
+# Verify that a partially received dataset can be sent with
+# 'zfs send --saved'.
+#
+# Strategy:
+# 1. Setup a pool with partially received filesystem
+# 2. Perform saved send without incremental
+# 3. Perform saved send with incremental
+# 4. Perform saved send with incremental, resuming from a token
+# 5. Perform negative tests for invalid command inputs
+#
+
+verify_runnable "both"
+
+log_assert "Verify that a partially received dataset can be sent with " \
+	"'zfs send --saved'."
+
+function cleanup
+{
+	destroy_dataset $POOL/testfs2 "-r"
+	destroy_dataset $POOL/stream "-r"
+	destroy_dataset $POOL/recvfs "-r"
+	destroy_dataset $POOL/partialfs "-r"
+}
+log_onexit cleanup
+
+log_must zfs create $POOL/testfs2
+log_must zfs create $POOL/stream
+mntpnt=$(get_prop mountpoint $POOL/testfs2)
+
+# Setup a pool with partially received filesystems
+log_must mkfile 1m $mntpnt/filea
+log_must zfs snap $POOL/testfs2@a
+log_must mkfile 1m $mntpnt/fileb
+log_must zfs snap $POOL/testfs2@b
+log_must eval "zfs send $POOL/testfs2@a | zfs recv $POOL/recvfs"
+log_must eval "zfs send -i $POOL/testfs2@a $POOL/testfs2@b > " \
+	"/$POOL/stream/inc.send"
+log_must eval "zfs send $POOL/testfs2@b > /$POOL/stream/full.send"
+mess_send_file /$POOL/stream/full.send
+mess_send_file /$POOL/stream/inc.send
+log_mustnot zfs recv -s $POOL/recvfullfs < /$POOL/stream/full.send
+log_mustnot zfs recv -s $POOL/recvfs < /$POOL/stream/inc.send
+
+# Perform saved send without incremental
+log_mustnot eval "zfs send --saved $POOL/recvfullfs | zfs recv -s " \
+	"$POOL/partialfs"
+token=$(zfs get -Hp -o value receive_resume_token $POOL/partialfs)
+log_must eval "zfs send -t $token | zfs recv -s $POOL/partialfs"
+file_check $POOL/recvfullfs $POOL/partialfs
+log_must zfs destroy -r $POOL/partialfs
+
+# Perform saved send with incremental
+log_must eval "zfs send $POOL/recvfs@a | zfs recv $POOL/partialfs"
+log_mustnot eval "zfs send --saved $POOL/recvfs | " \
+	"zfs recv -s $POOL/partialfs"
+token=$(zfs get -Hp -o value receive_resume_token $POOL/partialfs)
+log_must eval "zfs send -t $token | zfs recv -s $POOL/partialfs"
+file_check $POOL/recvfs $POOL/partialfs
+log_must zfs destroy -r $POOL/partialfs
+
+# Perform saved send with incremental, resuming from token
+log_must eval "zfs send $POOL/recvfs@a | zfs recv $POOL/partialfs"
+log_must eval "zfs send --saved $POOL/recvfs > " \
+	"/$POOL/stream/partial.send"
+mess_send_file /$POOL/stream/partial.send
+log_mustnot zfs recv -s $POOL/partialfs < /$POOL/stream/partial.send
+token=$(zfs get -Hp -o value receive_resume_token $POOL/partialfs)
+log_must eval "zfs send -t $token | zfs recv -s $POOL/partialfs"
+file_check $POOL/recvfs $POOL/partialfs
+
+# Perform negative tests for invalid command inputs
+set -A badargs \
+	"" \
+	"$POOL/recvfs@a" \
+	"-i $POOL/recvfs@a $POOL/recvfs@b" \
+	"-R $POOL/recvfs" \
+	"-p $POOL/recvfs" \
+	"-I $POOL/recvfs" \
+	"-D $POOL/recvfs" \
+	"-h $POOL/recvfs"
+
+while (( i < ${#badargs[*]} ))
+do
+	log_mustnot eval "zfs send --saved ${badargs[i]} >/dev/null"
+	(( i = i + 1 ))
+done
+
+log_pass "A partially received dataset can be sent with 'zfs send --saved'."