Implement bookmark copying

This feature allows copying existing bookmarks using

    zfs bookmark fs#target fs#newbookmark

There are some niche use cases for such functionality,
e.g. when using bookmarks as markers for replication progress.

Copying redaction bookmarks produces a normal bookmark that
cannot be used for redacted send (we are not duplicating
the redaction object).

ZCP support for bookmarking (both creation and copying) will be
implemented in a separate patch based on this work.

Overview:

- Terminology:
    - source = existing snapshot or bookmark
    - new/bmark = new bookmark
- Implement bookmark copying in `dsl_bookmark.c`
  - create new bookmark node
  - copy source's `zbn_phys` to new's `zbn_phys`
  - zero-out redaction object id in copy
- Extend existing bookmark ioctl nvlist schema to accept
  bookmarks as sources
  - => `dsl_bookmark_create_nvl_validate` is authoritative
- use `dsl_dataset_is_before` check for both snapshot
  and bookmark sources
- Adjust CLI
  - refactor shortname expansion logic in `zfs_do_bookmark`
- Update man pages
  - warn about redaction bookmark handling
- Add test cases
  - CLI
  - pyyzfs libzfs_core bindings

Reviewed-by: Matt Ahrens <[email protected]>
Reviewed-by: Paul Dagnelie <[email protected]>
Reviewed-by: Brian Behlendorf <[email protected]>
Signed-off-by: Christian Schwarz <[email protected]>
Closes #9571

19 files changed, 671 insertions, 141 deletions

diff --git a/cmd/zfs/zfs_main.c b/cmd/zfs/zfs_main.c
index 20ecb3031..ce8ed880c 100644
--- a/cmd/zfs/zfs_main.c
+++ b/cmd/zfs/zfs_main.c
@@ -30,6 +30,7 @@
  * Copyright (c) 2019 Datto Inc.
  * Copyright (c) 2019, loli10K <[email protected]>
  * Copyright 2019 Joyent, Inc.
+ * Copyright (c) 2019, 2020 by Christian Schwarz. All rights reserved.
  */
 
 #include <assert.h>
@@ -383,7 +384,8 @@ get_usage(zfs_help_t idx)
 		return (gettext("\tdiff [-FHt] <snapshot> "
 		    "[snapshot|filesystem]\n"));
 	case HELP_BOOKMARK:
-		return (gettext("\tbookmark <snapshot> <bookmark>\n"));
+		return (gettext("\tbookmark <snapshot|bookmark> "
+		    "<newbookmark>\n"));
 	case HELP_CHANNEL_PROGRAM:
 		return (gettext("\tprogram [-jn] [-t <instruction limit>] "
 		    "[-m <memory limit (b)>]\n"
@@ -7535,16 +7537,17 @@ out:
 }
 
 /*
- * zfs bookmark <fs@snap> <fs#bmark>
+ * zfs bookmark <fs@source>|<fs#source> <fs#bookmark>
  *
- * Creates a bookmark with the given name from the given snapshot.
+ * Creates a bookmark with the given name from the source snapshot
+ * or creates a copy of an existing source bookmark.
  */
 static int
 zfs_do_bookmark(int argc, char **argv)
 {
-	char snapname[ZFS_MAX_DATASET_NAME_LEN];
-	char bookname[ZFS_MAX_DATASET_NAME_LEN];
-	zfs_handle_t *zhp;
+	char *source, *bookname;
+	char expbuf[ZFS_MAX_DATASET_NAME_LEN];
+	int source_type;
 	nvlist_t *nvl;
 	int ret = 0;
 	int c;
@@ -7564,7 +7567,7 @@ zfs_do_bookmark(int argc, char **argv)
 
 	/* check number of arguments */
 	if (argc < 1) {
-		(void) fprintf(stderr, gettext("missing snapshot argument\n"));
+		(void) fprintf(stderr, gettext("missing source argument\n"));
 		goto usage;
 	}
 	if (argc < 2) {
@@ -7572,50 +7575,72 @@ zfs_do_bookmark(int argc, char **argv)
 		goto usage;
 	}
 
-	if (strchr(argv[0], '@') == NULL) {
+	source = argv[0];
+	bookname = argv[1];
+
+	if (strchr(source, '@') == NULL && strchr(source, '#') == NULL) {
 		(void) fprintf(stderr,
-		    gettext("invalid snapshot name '%s': "
-		    "must contain a '@'\n"), argv[0]);
+		    gettext("invalid source name '%s': "
+		    "must contain a '@' or '#'\n"), source);
 		goto usage;
 	}
-	if (strchr(argv[1], '#') == NULL) {
+	if (strchr(bookname, '#') == NULL) {
 		(void) fprintf(stderr,
 		    gettext("invalid bookmark name '%s': "
-		    "must contain a '#'\n"), argv[1]);
+		    "must contain a '#'\n"), bookname);
 		goto usage;
 	}
 
-	if (argv[0][0] == '@') {
-		/*
-		 * Snapshot name begins with @.
-		 * Default to same fs as bookmark.
-		 */
-		(void) strlcpy(snapname, argv[1], sizeof (snapname));
-		*strchr(snapname, '#') = '\0';
-		(void) strlcat(snapname, argv[0], sizeof (snapname));
-	} else {
-		(void) strlcpy(snapname, argv[0], sizeof (snapname));
+	/*
+	 * expand source or bookname to full path:
+	 * one of them may be specified as short name
+	 */
+	{
+		char **expand;
+		char *source_short, *bookname_short;
+		source_short = strpbrk(source, "@#");
+		bookname_short = strpbrk(bookname, "#");
+		if (source_short == source &&
+		    bookname_short == bookname) {
+			(void) fprintf(stderr, gettext(
+			    "either source or bookmark must be specified as "
+			    "full dataset paths"));
+			goto usage;
+		} else if (source_short != source &&
+		    bookname_short != bookname) {
+			expand = NULL;
+		} else if (source_short != source) {
+			strlcpy(expbuf, source, sizeof (expbuf));
+			expand = &bookname;
+		} else if (bookname_short != bookname) {
+			strlcpy(expbuf, bookname, sizeof (expbuf));
+			expand = &source;
+		} else {
+			abort();
+		}
+		if (expand != NULL) {
+			*strpbrk(expbuf, "@#") = '\0'; /* dataset name in buf */
+			(void) strlcat(expbuf, *expand, sizeof (expbuf));
+			*expand = expbuf;
+		}
 	}
-	if (argv[1][0] == '#') {
-		/*
-		 * Bookmark name begins with #.
-		 * Default to same fs as snapshot.
-		 */
-		(void) strlcpy(bookname, argv[0], sizeof (bookname));
-		*strchr(bookname, '@') = '\0';
-		(void) strlcat(bookname, argv[1], sizeof (bookname));
-	} else {
-		(void) strlcpy(bookname, argv[1], sizeof (bookname));
+
+	/* determine source type */
+	switch (*strpbrk(source, "@#")) {
+		case '@': source_type = ZFS_TYPE_SNAPSHOT; break;
+		case '#': source_type = ZFS_TYPE_BOOKMARK; break;
+		default: abort();
 	}
 
-	zhp = zfs_open(g_zfs, snapname, ZFS_TYPE_SNAPSHOT);
+	/* test the source exists */
+	zfs_handle_t *zhp;
+	zhp = zfs_open(g_zfs, source, source_type);
 	if (zhp == NULL)
 		goto usage;
 	zfs_close(zhp);
 
-
 	nvl = fnvlist_alloc();
-	fnvlist_add_string(nvl, bookname, snapname);
+	fnvlist_add_string(nvl, bookname, source);
 	ret = lzc_bookmark(nvl, NULL);
 	fnvlist_free(nvl);
 
@@ -7631,6 +7656,10 @@ zfs_do_bookmark(int argc, char **argv)
 		case EXDEV:
 			err_msg = "bookmark is in a different pool";
 			break;
+		case ZFS_ERR_BOOKMARK_SOURCE_NOT_ANCESTOR:
+			err_msg = "source is not an ancestor of the "
+			    "new bookmark's dataset";
+			break;
 		case EEXIST:
 			err_msg = "bookmark exists";
 			break;
diff --git a/contrib/pyzfs/libzfs_core/_constants.py b/contrib/pyzfs/libzfs_core/_constants.py
index 55de55d42..16057e9b3 100644
--- a/contrib/pyzfs/libzfs_core/_constants.py
+++ b/contrib/pyzfs/libzfs_core/_constants.py
@@ -22,11 +22,15 @@ from __future__ import absolute_import, division, print_function
 
 
 # https://stackoverflow.com/a/1695250
-def enum(*sequential, **named):
-    enums = dict(((b, a) for a, b in enumerate(sequential)), **named)
+def enum_with_offset(offset, sequential, named):
+    enums = dict(((b, a + offset) for a, b in enumerate(sequential)), **named)
     return type('Enum', (), enums)
 
 
+def enum(*sequential, **named):
+    return enum_with_offset(0, sequential, named)
+
+
 #: Maximum length of any ZFS name.
 MAXNAMELEN = 255
 #: Default channel program limits
@@ -60,12 +64,34 @@ zio_encrypt = enum(
     'ZIO_CRYPT_AES_256_GCM'
 )
 # ZFS-specific error codes
-ZFS_ERR_CHECKPOINT_EXISTS = 1024
-ZFS_ERR_DISCARDING_CHECKPOINT = 1025
-ZFS_ERR_NO_CHECKPOINT = 1026
-ZFS_ERR_DEVRM_IN_PROGRESS = 1027
-ZFS_ERR_VDEV_TOO_BIG = 1028
-ZFS_ERR_WRONG_PARENT = 1033
+zfs_errno = enum_with_offset(1024, [
+        'ZFS_ERR_CHECKPOINT_EXISTS',
+        'ZFS_ERR_DISCARDING_CHECKPOINT',
+        'ZFS_ERR_NO_CHECKPOINT',
+        'ZFS_ERR_DEVRM_IN_PROGRESS',
+        'ZFS_ERR_VDEV_TOO_BIG',
+        'ZFS_ERR_IOC_CMD_UNAVAIL',
+        'ZFS_ERR_IOC_ARG_UNAVAIL',
+        'ZFS_ERR_IOC_ARG_REQUIRED',
+        'ZFS_ERR_IOC_ARG_BADTYPE',
+        'ZFS_ERR_WRONG_PARENT',
+        'ZFS_ERR_FROM_IVSET_GUID_MISSING',
+        'ZFS_ERR_FROM_IVSET_GUID_MISMATCH',
+        'ZFS_ERR_SPILL_BLOCK_FLAG_MISSING',
+        'ZFS_ERR_UNKNOWN_SEND_STREAM_FEATURE',
+        'ZFS_ERR_EXPORT_IN_PROGRESS',
+        'ZFS_ERR_BOOKMARK_SOURCE_NOT_ANCESTOR',
+    ],
+    {}
+)
+# compat before we used the enum helper for these values
+ZFS_ERR_CHECKPOINT_EXISTS = zfs_errno.ZFS_ERR_CHECKPOINT_EXISTS
+assert(ZFS_ERR_CHECKPOINT_EXISTS == 1024)
+ZFS_ERR_DISCARDING_CHECKPOINT = zfs_errno.ZFS_ERR_DISCARDING_CHECKPOINT
+ZFS_ERR_NO_CHECKPOINT = zfs_errno.ZFS_ERR_NO_CHECKPOINT
+ZFS_ERR_DEVRM_IN_PROGRESS = zfs_errno.ZFS_ERR_DEVRM_IN_PROGRESS
+ZFS_ERR_VDEV_TOO_BIG = zfs_errno.ZFS_ERR_VDEV_TOO_BIG
+ZFS_ERR_WRONG_PARENT = zfs_errno.ZFS_ERR_WRONG_PARENT
 
 
 # vim: softtabstop=4 tabstop=4 expandtab shiftwidth=4
diff --git a/contrib/pyzfs/libzfs_core/_error_translation.py b/contrib/pyzfs/libzfs_core/_error_translation.py
index cf52ac918..b97bdd89a 100644
--- a/contrib/pyzfs/libzfs_core/_error_translation.py
+++ b/contrib/pyzfs/libzfs_core/_error_translation.py
@@ -39,7 +39,8 @@ from ._constants import (
     ZFS_ERR_NO_CHECKPOINT,
     ZFS_ERR_DEVRM_IN_PROGRESS,
     ZFS_ERR_VDEV_TOO_BIG,
-    ZFS_ERR_WRONG_PARENT
+    ZFS_ERR_WRONG_PARENT,
+    zfs_errno
 )
 
 
@@ -147,21 +148,36 @@ def lzc_destroy_snaps_translate_errors(ret, errlist, snaps, defer):
 
 
 def lzc_bookmark_translate_errors(ret, errlist, bookmarks):
+
     if ret == 0:
         return
 
     def _map(ret, name):
+        source = bookmarks[name]
         if ret == errno.EINVAL:
             if name:
-                snap = bookmarks[name]
                 pool_names = map(_pool_name, bookmarks.keys())
-                if not _is_valid_bmark_name(name):
-                    return lzc_exc.BookmarkNameInvalid(name)
-                elif not _is_valid_snap_name(snap):
-                    return lzc_exc.SnapshotNameInvalid(snap)
-                elif _fs_name(name) != _fs_name(snap):
-                    return lzc_exc.BookmarkMismatch(name)
-                elif any(x != _pool_name(name) for x in pool_names):
+
+                # use _validate* functions for MAXNAMELEN check
+                try:
+                    _validate_bmark_name(name)
+                except lzc_exc.ZFSError as e:
+                    return e
+
+                try:
+                    _validate_snap_name(source)
+                    source_is_snap = True
+                except lzc_exc.ZFSError:
+                    source_is_snap = False
+                try:
+                    _validate_bmark_name(source)
+                    source_is_bmark = True
+                except lzc_exc.ZFSError:
+                    source_is_bmark = False
+                if not source_is_snap and not source_is_bmark:
+                    return lzc_exc.BookmarkSourceInvalid(source)
+
+                if any(x != _pool_name(name) for x in pool_names):
                     return lzc_exc.PoolsDiffer(name)
             else:
                 invalid_names = [
@@ -174,6 +190,8 @@ def lzc_bookmark_translate_errors(ret, errlist, bookmarks):
             return lzc_exc.SnapshotNotFound(name)
         if ret == errno.ENOTSUP:
             return lzc_exc.BookmarkNotSupported(name)
+        if ret == zfs_errno.ZFS_ERR_BOOKMARK_SOURCE_NOT_ANCESTOR:
+            return lzc_exc.BookmarkMismatch(source)
         return _generic_exception(ret, name, "Failed to create bookmark")
 
     _handle_err_list(
diff --git a/contrib/pyzfs/libzfs_core/_libzfs_core.py b/contrib/pyzfs/libzfs_core/_libzfs_core.py
index 06797b0f3..fcfa5be31 100644
--- a/contrib/pyzfs/libzfs_core/_libzfs_core.py
+++ b/contrib/pyzfs/libzfs_core/_libzfs_core.py
@@ -319,14 +319,15 @@ def lzc_bookmark(bookmarks):
     Create bookmarks.
 
     :param bookmarks: a dict that maps names of wanted bookmarks to names of
-        existing snapshots.
+        existing snapshots or bookmarks.
     :type bookmarks: dict of bytes to bytes
     :raises BookmarkFailure: if any of the bookmarks can not be created for any
         reason.
 
     The bookmarks `dict` maps from name of the bookmark
     (e.g. :file:`{pool}/{fs}#{bmark}`) to the name of the snapshot
-    (e.g. :file:`{pool}/{fs}@{snap}`).  All the bookmarks and snapshots must
+    (e.g. :file:`{pool}/{fs}@{snap}`) or existint bookmark
+    :file:`{pool}/{fs}@{snap}`. All the bookmarks and snapshots must
     be in the same pool.
     '''
     errlist = {}
diff --git a/contrib/pyzfs/libzfs_core/exceptions.py b/contrib/pyzfs/libzfs_core/exceptions.py
index f8a775433..2206c2f2c 100644
--- a/contrib/pyzfs/libzfs_core/exceptions.py
+++ b/contrib/pyzfs/libzfs_core/exceptions.py
@@ -227,7 +227,15 @@ class BookmarkNotFound(ZFSError):
 
 class BookmarkMismatch(ZFSError):
     errno = errno.EINVAL
-    message = "Bookmark is not in snapshot's filesystem"
+    message = "source is not an ancestor of the new bookmark's dataset"
+
+    def __init__(self, name):
+        self.name = name
+
+
+class BookmarkSourceInvalid(ZFSError):
+    errno = errno.EINVAL
+    message = "Bookmark source is not a valid snapshot or existing bookmark"
 
     def __init__(self, name):
         self.name = name
diff --git a/contrib/pyzfs/libzfs_core/test/test_libzfs_core.py b/contrib/pyzfs/libzfs_core/test/test_libzfs_core.py
index 613d5eccd..f47583b83 100644
--- a/contrib/pyzfs/libzfs_core/test/test_libzfs_core.py
+++ b/contrib/pyzfs/libzfs_core/test/test_libzfs_core.py
@@ -1032,17 +1032,37 @@ class ZFSTest(unittest.TestCase):
         bmarks = [ZFSTest.pool.makeName(
             b'fs1#bmark1'), ZFSTest.pool.makeName(b'fs2#bmark1')]
         bmark_dict = {x: y for x, y in zip(bmarks, snaps)}
-
         lzc.lzc_snapshot(snaps)
         lzc.lzc_bookmark(bmark_dict)
         lzc.lzc_destroy_snaps(snaps, defer=False)
 
     @skipUnlessBookmarksSupported
+    def test_bookmark_copying(self):
+        snaps = [ZFSTest.pool.makeName(s) for s in [
+            b'fs1@snap1', b'fs1@snap2', b'fs2@snap1']]
+        bmarks = [ZFSTest.pool.makeName(x) for x in [
+            b'fs1#bmark1', b'fs1#bmark2', b'fs2#bmark1']]
+        bmarks_copies = [ZFSTest.pool.makeName(x) for x in [
+            b'fs1#bmark1_copy', b'fs1#bmark2_copy', b'fs2#bmark1_copy']]
+        bmark_dict = {x: y for x, y in zip(bmarks, snaps)}
+        bmark_copies_dict = {x: y for x, y in zip(bmarks_copies, bmarks)}
+
+        for snap in snaps:
+            lzc.lzc_snapshot([snap])
+        lzc.lzc_bookmark(bmark_dict)
+
+        lzc.lzc_bookmark(bmark_copies_dict)
+        lzc.lzc_destroy_bookmarks(bmarks_copies)
+
+        lzc.lzc_destroy_bookmarks(bmarks)
+        lzc.lzc_destroy_snaps(snaps, defer=False)
+
+    @skipUnlessBookmarksSupported
     def test_bookmarks_empty(self):
         lzc.lzc_bookmark({})
 
     @skipUnlessBookmarksSupported
-    def test_bookmarks_mismatching_name(self):
+    def test_bookmarks_foregin_source(self):
         snaps = [ZFSTest.pool.makeName(b'fs1@snap1')]
         bmarks = [ZFSTest.pool.makeName(b'fs2#bmark1')]
         bmark_dict = {x: y for x, y in zip(bmarks, snaps)}
@@ -1107,7 +1127,7 @@ class ZFSTest(unittest.TestCase):
             self.assertIsInstance(e, lzc_exc.NameTooLong)
 
     @skipUnlessBookmarksSupported
-    def test_bookmarks_mismatching_names(self):
+    def test_bookmarks_foreign_sources(self):
         snaps = [ZFSTest.pool.makeName(
             b'fs1@snap1'), ZFSTest.pool.makeName(b'fs2@snap1')]
         bmarks = [ZFSTest.pool.makeName(
@@ -1122,7 +1142,7 @@ class ZFSTest(unittest.TestCase):
             self.assertIsInstance(e, lzc_exc.BookmarkMismatch)
 
     @skipUnlessBookmarksSupported
-    def test_bookmarks_partially_mismatching_names(self):
+    def test_bookmarks_partially_foreign_sources(self):
         snaps = [ZFSTest.pool.makeName(
             b'fs1@snap1'), ZFSTest.pool.makeName(b'fs2@snap1')]
         bmarks = [ZFSTest.pool.makeName(
@@ -1154,33 +1174,48 @@ class ZFSTest(unittest.TestCase):
 
     @skipUnlessBookmarksSupported
     def test_bookmarks_missing_snap(self):
+        fss = [ZFSTest.pool.makeName(b'fs1'), ZFSTest.pool.makeName(b'fs2')]
         snaps = [ZFSTest.pool.makeName(
             b'fs1@snap1'), ZFSTest.pool.makeName(b'fs2@snap1')]
         bmarks = [ZFSTest.pool.makeName(
             b'fs1#bmark1'), ZFSTest.pool.makeName(b'fs2#bmark1')]
         bmark_dict = {x: y for x, y in zip(bmarks, snaps)}
 
-        lzc.lzc_snapshot(snaps[0:1])
+        lzc.lzc_snapshot(snaps[0:1])  # only create fs1@snap1
+
         with self.assertRaises(lzc_exc.BookmarkFailure) as ctx:
             lzc.lzc_bookmark(bmark_dict)
 
         for e in ctx.exception.errors:
             self.assertIsInstance(e, lzc_exc.SnapshotNotFound)
 
+        # no new bookmarks are created if one or more sources do not exist
+        for fs in fss:
+            fsbmarks = lzc.lzc_get_bookmarks(fs)
+            self.assertEqual(len(fsbmarks), 0)
+
     @skipUnlessBookmarksSupported
     def test_bookmarks_missing_snaps(self):
+        fss = [ZFSTest.pool.makeName(b'fs1'), ZFSTest.pool.makeName(b'fs2')]
         snaps = [ZFSTest.pool.makeName(
             b'fs1@snap1'), ZFSTest.pool.makeName(b'fs2@snap1')]
         bmarks = [ZFSTest.pool.makeName(
             b'fs1#bmark1'), ZFSTest.pool.makeName(b'fs2#bmark1')]
         bmark_dict = {x: y for x, y in zip(bmarks, snaps)}
 
+        # do not create any snapshots
+
         with self.assertRaises(lzc_exc.BookmarkFailure) as ctx:
             lzc.lzc_bookmark(bmark_dict)
 
         for e in ctx.exception.errors:
             self.assertIsInstance(e, lzc_exc.SnapshotNotFound)
 
+        # no new bookmarks are created if one or more sources do not exist
+        for fs in fss:
+            fsbmarks = lzc.lzc_get_bookmarks(fs)
+            self.assertEqual(len(fsbmarks), 0)
+
     @skipUnlessBookmarksSupported
     def test_bookmarks_for_the_same_snap(self):
         snap = ZFSTest.pool.makeName(b'fs1@snap1')
diff --git a/include/sys/dsl_bookmark.h b/include/sys/dsl_bookmark.h
index 7e4b8f02f..ec3895d40 100644
--- a/include/sys/dsl_bookmark.h
+++ b/include/sys/dsl_bookmark.h
@@ -103,6 +103,7 @@ typedef struct redact_block_phys {
 typedef int (*rl_traverse_callback_t)(redact_block_phys_t *, void *);
 
 int dsl_bookmark_create(nvlist_t *, nvlist_t *);
+int dsl_bookmark_create_nvl_validate(nvlist_t *);
 int dsl_bookmark_create_redacted(const char *, const char *, uint64_t,
     uint64_t *, void *, redaction_list_t **);
 int dsl_get_bookmarks(const char *, nvlist_t *, nvlist_t *);
diff --git a/include/sys/fs/zfs.h b/include/sys/fs/zfs.h
index 3e2c00235..f919bd9b1 100644
--- a/include/sys/fs/zfs.h
+++ b/include/sys/fs/zfs.h
@@ -1310,6 +1310,8 @@ typedef enum zfs_ioc {
  * not described precisely by generic errno codes.
  *
  * These numbers should not change over time. New entries should be appended.
+ *
+ * (Keep in sync with contrib/pyzfs/libzfs_core/_constants.py)
  */
 typedef enum {
 	ZFS_ERR_CHECKPOINT_EXISTS = 1024,
@@ -1327,6 +1329,7 @@ typedef enum {
 	ZFS_ERR_SPILL_BLOCK_FLAG_MISSING,
 	ZFS_ERR_UNKNOWN_SEND_STREAM_FEATURE,
 	ZFS_ERR_EXPORT_IN_PROGRESS,
+	ZFS_ERR_BOOKMARK_SOURCE_NOT_ANCESTOR,
 } zfs_errno_t;
 
 /*
diff --git a/include/zfs_namecheck.h b/include/zfs_namecheck.h
index 56d3d36f0..197c40b56 100644
--- a/include/zfs_namecheck.h
+++ b/include/zfs_namecheck.h
@@ -46,6 +46,7 @@ typedef enum {
 	NAME_ERR_SELF_REF,		/* reserved self path name ('.') */
 	NAME_ERR_PARENT_REF,		/* reserved parent path name ('..') */
 	NAME_ERR_NO_AT,			/* permission set is missing '@' */
+	NAME_ERR_NO_POUND, 		/* permission set is missing '#' */
 } namecheck_err_t;
 
 #define	ZFS_PERMSET_MAXLEN	64
@@ -56,6 +57,8 @@ int get_dataset_depth(const char *);
 int pool_namecheck(const char *, namecheck_err_t *, char *);
 int entity_namecheck(const char *, namecheck_err_t *, char *);
 int dataset_namecheck(const char *, namecheck_err_t *, char *);
+int snapshot_namecheck(const char *, namecheck_err_t *, char *);
+int bookmark_namecheck(const char *, namecheck_err_t *, char *);
 int dataset_nestcheck(const char *);
 int mountpoint_namecheck(const char *, namecheck_err_t *);
 int zfs_component_namecheck(const char *, namecheck_err_t *, char *);
diff --git a/lib/libzfs_core/libzfs_core.c b/lib/libzfs_core/libzfs_core.c
index 5dd8976eb..f65db4ff4 100644
--- a/lib/libzfs_core/libzfs_core.c
+++ b/lib/libzfs_core/libzfs_core.c
@@ -25,6 +25,7 @@
  * Copyright (c) 2017 Datto Inc.
  * Copyright 2017 RackTop Systems.
  * Copyright (c) 2017 Open-E, Inc. All Rights Reserved.
+ * Copyright (c) 2019, 2020 by Christian Schwarz. All rights reserved.
  */
 
 /*
@@ -1127,11 +1128,13 @@ lzc_rollback_to(const char *fsname, const char *snapname)
 }
 
 /*
- * Creates bookmarks.
+ * Creates new bookmarks from existing snapshot or bookmark.
  *
- * The bookmarks nvlist maps from name of the bookmark (e.g. "pool/fs#bmark") to
- * the name of the snapshot (e.g. "pool/fs@snap").  All the bookmarks and
- * snapshots must be in the same pool.
+ * The bookmarks nvlist maps from the full name of the new bookmark to
+ * the full name of the source snapshot or bookmark.
+ * All the bookmarks and snapshots must be in the same pool.
+ * The new bookmarks names must be unique.
+ * => see function dsl_bookmark_create_nvl_validate
  *
  * The returned results nvlist will have an entry for each bookmark that failed.
  * The value will be the (int32) error code.
@@ -1146,7 +1149,7 @@ lzc_bookmark(nvlist_t *bookmarks, nvlist_t **errlist)
 	int error;
 	char pool[ZFS_MAX_DATASET_NAME_LEN];
 
-	/* determine the pool name */
+	/* determine pool name from first bookmark */
 	elem = nvlist_next_nvpair(bookmarks, NULL);
 	if (elem == NULL)
 		return (0);
diff --git a/man/man8/zfs-bookmark.8 b/man/man8/zfs-bookmark.8
index 04d4af556..c1b48516d 100644
--- a/man/man8/zfs-bookmark.8
+++ b/man/man8/zfs-bookmark.8
@@ -29,6 +29,7 @@
 .\" Copyright 2019 Richard Laager. All rights reserved.
 .\" Copyright 2018 Nexenta Systems, Inc.
 .\" Copyright 2019 Joyent, Inc.
+.\" Copyright (c) 2019, 2020 by Christian Schwarz. All Rights Reserved.
 .\"
 .Dd June 30, 2019
 .Dt ZFS-BOOKMARK 8 SMM
@@ -42,14 +43,19 @@
 .It Xo
 .Nm
 .Cm bookmark
-.Ar snapshot bookmark
+.Ar snapshot Ns | Ns Ar bookmark newbookmark
 .Xc
-Creates a bookmark of the given snapshot.
+Creates a new bookmark of the given snapshot or bookmark.
 Bookmarks mark the point in time when the snapshot was created, and can be used
 as the incremental source for a
 .Xr zfs-send 8
 command.
 .Pp
+When creating a bookmark from an existing redaction bookmark, the resulting
+bookmark is
+.Sy not
+a redaction bookmark.
+.Pp
 This feature must be enabled to be used.
 See
 .Xr zpool-features 5
diff --git a/man/man8/zfs.8 b/man/man8/zfs.8
index 62b7f1f8a..eb6e0e33e 100644
--- a/man/man8/zfs.8
+++ b/man/man8/zfs.8
@@ -200,7 +200,7 @@ Streams are created using the
 .Xr zfs-send 8
 subcommand, which by default creates a full stream.
 .It Xr zfs-bookmark 8
-Creates a bookmark of the given snapshot.
+Creates a new bookmark of the given snapshot or bookmark.
 Bookmarks mark the point in time when the snapshot was created, and can be used
 as the incremental source for a
 .Nm zfs Cm send
diff --git a/module/zcommon/zfs_namecheck.c b/module/zcommon/zfs_namecheck.c
index 3076b8d8b..f8625042a 100644
--- a/module/zcommon/zfs_namecheck.c
+++ b/module/zcommon/zfs_namecheck.c
@@ -183,6 +183,8 @@ entity_namecheck(const char *path, namecheck_err_t *why, char *what)
 {
 	const char *end;
 
+	EQUIV(why == NULL, what == NULL);
+
 	/*
 	 * Make sure the name is not too long.
 	 */
@@ -311,6 +313,44 @@ dataset_namecheck(const char *path, namecheck_err_t *why, char *what)
 }
 
 /*
+ * Assert path is a valid bookmark name
+ */
+int
+bookmark_namecheck(const char *path, namecheck_err_t *why, char *what)
+{
+	int ret = entity_namecheck(path, why, what);
+
+	if (ret == 0 && strchr(path, '#') == NULL) {
+		if (why != NULL) {
+			*why = NAME_ERR_NO_POUND;
+			*what = '#';
+		}
+		return (-1);
+	}
+
+	return (ret);
+}
+
+/*
+ * Assert path is a valid snapshot name
+ */
+int
+snapshot_namecheck(const char *path, namecheck_err_t *why, char *what)
+{
+	int ret = entity_namecheck(path, why, what);
+
+	if (ret == 0 && strchr(path, '@') == NULL) {
+		if (why != NULL) {
+			*why = NAME_ERR_NO_AT;
+			*what = '@';
+		}
+		return (-1);
+	}
+
+	return (ret);
+}
+
+/*
  * mountpoint names must be of the following form:
  *
  *	/[component][/]*[component][/]
@@ -420,6 +460,8 @@ pool_namecheck(const char *pool, namecheck_err_t *why, char *what)
 EXPORT_SYMBOL(entity_namecheck);
 EXPORT_SYMBOL(pool_namecheck);
 EXPORT_SYMBOL(dataset_namecheck);
+EXPORT_SYMBOL(bookmark_namecheck);
+EXPORT_SYMBOL(snapshot_namecheck);
 EXPORT_SYMBOL(zfs_component_namecheck);
 EXPORT_SYMBOL(dataset_nestcheck);
 EXPORT_SYMBOL(get_dataset_depth);
diff --git a/module/zfs/dsl_bookmark.c b/module/zfs/dsl_bookmark.c
index 4d5c601d6..5a73c5300 100644
--- a/module/zfs/dsl_bookmark.c
+++ b/module/zfs/dsl_bookmark.c
@@ -16,6 +16,7 @@
 /*
  * Copyright (c) 2013, 2018 by Delphix. All rights reserved.
  * Copyright 2017 Nexenta Systems, Inc.
+ * Copyright 2019, 2020 by Christian Schwarz. All rights reserved.
  */
 
 #include <sys/zfs_context.h>
@@ -55,6 +56,9 @@ dsl_bookmark_hold_ds(dsl_pool_t *dp, const char *fullname,
 }
 
 /*
+ * When reading BOOKMARK_V1 bookmarks, the BOOKMARK_V2 fields are guaranteed
+ * to be zeroed.
+ *
  * Returns ESRCH if bookmark is not found.
  * Note, we need to use the ZAP rather than the AVL to look up bookmarks
  * by name, because only the ZAP honors the casesensitivity setting.
@@ -116,6 +120,91 @@ dsl_bookmark_lookup(dsl_pool_t *dp, const char *fullname,
 	return (error);
 }
 
+/*
+ * Validates that
+ * - bmark is a full dataset path of a bookmark (bookmark_namecheck)
+ * - source is a full path of a snaphot or bookmark
+ *   ({bookmark,snapshot}_namecheck)
+ *
+ * Returns 0 if valid, -1 otherwise.
+ */
+static int
+dsl_bookmark_create_nvl_validate_pair(const char *bmark, const char *source)
+{
+	if (bookmark_namecheck(bmark, NULL, NULL) != 0)
+		return (-1);
+
+	int is_bmark, is_snap;
+	is_bmark = bookmark_namecheck(source, NULL, NULL) == 0;
+	is_snap = snapshot_namecheck(source, NULL, NULL) == 0;
+	if (!is_bmark && !is_snap)
+		return (-1);
+
+	return (0);
+}
+
+/*
+ * Check that the given nvlist corresponds to the following schema:
+ *  { newbookmark -> source, ... }
+ * where
+ * - each pair passes dsl_bookmark_create_nvl_validate_pair
+ * - all newbookmarks are in the same pool
+ * - all newbookmarks have unique names
+ *
+ * Note that this function is only validates above schema. Callers must ensure
+ * that the bookmarks can be created, e.g. that sources exist.
+ *
+ * Returns 0 if the nvlist adheres to above schema.
+ * Returns -1 if it doesn't.
+ */
+int
+dsl_bookmark_create_nvl_validate(nvlist_t *bmarks)
+{
+	char *first;
+	size_t first_len;
+
+	first = NULL;
+	for (nvpair_t *pair = nvlist_next_nvpair(bmarks, NULL);
+	    pair != NULL; pair = nvlist_next_nvpair(bmarks, pair)) {
+
+		char *bmark = nvpair_name(pair);
+		char *source;
+
+		/* list structure: values must be snapshots XOR bookmarks */
+		if (nvpair_value_string(pair, &source) != 0)
+			return (-1);
+		if (dsl_bookmark_create_nvl_validate_pair(bmark, source) != 0)
+			return (-1);
+
+		/* same pool check */
+		if (first == NULL) {
+			char *cp = strpbrk(bmark, "/#");
+			if (cp == NULL)
+				return (-1);
+			first = bmark;
+			first_len = cp - bmark;
+		}
+		if (strncmp(first, bmark, first_len) != 0)
+			return (-1);
+		switch (*(bmark + first_len)) {
+			case '/': /* fallthrough */
+			case '#':
+				break;
+			default:
+				return (-1);
+		}
+
+		/* unique newbookmark names; todo: O(n^2) */
+		for (nvpair_t *pair2 = nvlist_next_nvpair(bmarks, pair);
+		    pair2 != NULL; pair2 = nvlist_next_nvpair(bmarks, pair2)) {
+			if (strcmp(nvpair_name(pair), nvpair_name(pair2)) == 0)
+				return (-1);
+		}
+
+	}
+	return (0);
+}
+
 typedef struct dsl_bookmark_create_redacted_arg {
 	const char	*dbcra_bmark;
 	const char	*dbcra_snap;
@@ -130,36 +219,85 @@ typedef struct dsl_bookmark_create_arg {
 	nvlist_t *dbca_errors;
 } dsl_bookmark_create_arg_t;
 
+/*
+ * expects that newbm and source have been validated using
+ * dsl_bookmark_create_nvl_validate_pair
+ */
 static int
-dsl_bookmark_create_check_impl(dsl_dataset_t *snapds, const char *bookmark_name,
-    dmu_tx_t *tx)
+dsl_bookmark_create_check_impl(dsl_pool_t *dp,
+    const char *newbm, const char *source)
 {
-	dsl_pool_t *dp = dmu_tx_pool(tx);
-	dsl_dataset_t *bmark_fs;
-	char *shortname;
-	int error;
-	zfs_bookmark_phys_t bmark_phys = { 0 };
+	ASSERT0(dsl_bookmark_create_nvl_validate_pair(newbm, source));
+	/* defer source namecheck until we know it's a snapshot or bookmark */
 
-	if (!snapds->ds_is_snapshot)
-		return (SET_ERROR(EINVAL));
+	int error;
+	dsl_dataset_t *newbm_ds;
+	char *newbm_short;
+	zfs_bookmark_phys_t bmark_phys;
 
-	error = dsl_bookmark_hold_ds(dp, bookmark_name,
-	    &bmark_fs, FTAG, &shortname);
+	error = dsl_bookmark_hold_ds(dp, newbm, &newbm_ds, FTAG, &newbm_short);
 	if (error != 0)
 		return (error);
 
-	if (!dsl_dataset_is_before(bmark_fs, snapds, 0)) {
-		dsl_dataset_rele(bmark_fs, FTAG);
-		return (SET_ERROR(EINVAL));
+	/* Verify that the new bookmark does not already exist */
+	error = dsl_bookmark_lookup_impl(newbm_ds, newbm_short, &bmark_phys);
+	switch (error) {
+	case ESRCH:
+		/* happy path: new bmark doesn't exist, proceed after switch */
+		error = 0;
+		break;
+	case 0:
+		error = SET_ERROR(EEXIST);
+		goto eholdnewbmds;
+	default:
+		/* dsl_bookmark_lookup_impl already did SET_ERRROR */
+		goto eholdnewbmds;
 	}
 
-	error = dsl_bookmark_lookup_impl(bmark_fs, shortname,
-	    &bmark_phys);
-	dsl_dataset_rele(bmark_fs, FTAG);
-	if (error == 0)
-		return (SET_ERROR(EEXIST));
-	if (error == ESRCH)
-		return (0);
+	/* error is retval of the following if-cascade */
+	if (strchr(source, '@') != NULL) {
+		dsl_dataset_t *source_snap_ds;
+		ASSERT3S(snapshot_namecheck(source, NULL, NULL), ==, 0);
+		error = dsl_dataset_hold(dp, source, FTAG, &source_snap_ds);
+		if (error == 0) {
+			VERIFY(source_snap_ds->ds_is_snapshot);
+			/*
+			 * Verify that source snapshot is an earlier point in
+			 * newbm_ds's timeline (source may be newbm_ds's origin)
+			 */
+			if (!dsl_dataset_is_before(newbm_ds, source_snap_ds, 0))
+				error = SET_ERROR(
+				    ZFS_ERR_BOOKMARK_SOURCE_NOT_ANCESTOR);
+			dsl_dataset_rele(source_snap_ds, FTAG);
+		}
+	} else if (strchr(source, '#') != NULL) {
+		zfs_bookmark_phys_t source_phys;
+		ASSERT3S(bookmark_namecheck(source, NULL, NULL), ==, 0);
+		/*
+		 * Source must exists and be an earlier point in newbm_ds's
+		 * timeline (newbm_ds's origin may be a snap of source's ds)
+		 */
+		error = dsl_bookmark_lookup(dp, source, newbm_ds, &source_phys);
+		switch (error) {
+		case 0:
+			break; /* happy path */
+		case EXDEV:
+			error = SET_ERROR(ZFS_ERR_BOOKMARK_SOURCE_NOT_ANCESTOR);
+			break;
+		default:
+			/* dsl_bookmark_lookup already did SET_ERRROR */
+			break;
+		}
+	} else {
+		/*
+		 * dsl_bookmark_create_nvl_validate validates that source is
+		 * either snapshot or bookmark
+		 */
+		panic("unreachable code: %s", source);
+	}
+
+eholdnewbmds:
+	dsl_dataset_rele(newbm_ds, FTAG);
 	return (error);
 }
 
@@ -167,33 +305,37 @@ static int
 dsl_bookmark_create_check(void *arg, dmu_tx_t *tx)
 {
 	dsl_bookmark_create_arg_t *dbca = arg;
+	int rv = 0;
+	int schema_err = 0;
 	ASSERT3P(dbca, !=, NULL);
 	ASSERT3P(dbca->dbca_bmarks, !=, NULL);
+	/* dbca->dbca_errors is allowed to be NULL */
 
 	dsl_pool_t *dp = dmu_tx_pool(tx);
-	int rv = 0;
 
 	if (!spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_BOOKMARKS))
 		return (SET_ERROR(ENOTSUP));
 
+	if (dsl_bookmark_create_nvl_validate(dbca->dbca_bmarks) != 0)
+		rv = schema_err = SET_ERROR(EINVAL);
+
 	for (nvpair_t *pair = nvlist_next_nvpair(dbca->dbca_bmarks, NULL);
 	    pair != NULL; pair = nvlist_next_nvpair(dbca->dbca_bmarks, pair)) {
-		dsl_dataset_t *snapds;
-		int error;
+		char *new = nvpair_name(pair);
 
-		/* note: validity of nvlist checked by ioctl layer */
-		error = dsl_dataset_hold(dp, fnvpair_value_string(pair),
-		    FTAG, &snapds);
+		int error = schema_err;
 		if (error == 0) {
-			error = dsl_bookmark_create_check_impl(snapds,
-			    nvpair_name(pair), tx);
-			dsl_dataset_rele(snapds, FTAG);
+			char *source = fnvpair_value_string(pair);
+			error = dsl_bookmark_create_check_impl(dp, new, source);
+			if (error != 0)
+				error = SET_ERROR(error);
 		}
+
 		if (error != 0) {
 			rv = error;
 			if (dbca->dbca_errors != NULL)
 				fnvlist_add_int32(dbca->dbca_errors,
-				    nvpair_name(pair), error);
+				    new, error);
 		}
 	}
 
@@ -259,6 +401,10 @@ dsl_bookmark_set_phys(zfs_bookmark_phys_t *zbm, dsl_dataset_t *snap)
 	}
 }
 
+/*
+ * Add dsl_bookmark_node_t `dbn` to the given dataset and increment appropriate
+ * SPA feature counters.
+ */
 void
 dsl_bookmark_node_add(dsl_dataset_t *hds, dsl_bookmark_node_t *dbn,
     dmu_tx_t *tx)
@@ -308,7 +454,7 @@ dsl_bookmark_node_add(dsl_dataset_t *hds, dsl_bookmark_node_t *dbn,
  * list, and store the object number of the redaction list in redact_obj.
  */
 static void
-dsl_bookmark_create_sync_impl(const char *bookmark, const char *snapshot,
+dsl_bookmark_create_sync_impl_snap(const char *bookmark, const char *snapshot,
     dmu_tx_t *tx, uint64_t num_redact_snaps, uint64_t *redact_snaps, void *tag,
     redaction_list_t **redaction_list)
 {
@@ -381,7 +527,76 @@ dsl_bookmark_create_sync_impl(const char *bookmark, const char *snapshot,
 	dsl_dataset_rele(snapds, FTAG);
 }
 
+
 static void
+dsl_bookmark_create_sync_impl_book(
+    const char *new_name, const char *source_name, dmu_tx_t *tx)
+{
+	dsl_pool_t *dp = dmu_tx_pool(tx);
+	dsl_dataset_t *bmark_fs_source, *bmark_fs_new;
+	char *source_shortname, *new_shortname;
+	zfs_bookmark_phys_t source_phys;
+
+	VERIFY0(dsl_bookmark_hold_ds(dp, source_name, &bmark_fs_source, FTAG,
+	    &source_shortname));
+	VERIFY0(dsl_bookmark_hold_ds(dp, new_name, &bmark_fs_new, FTAG,
+	    &new_shortname));
+
+	/*
+	 * create a copy of the source bookmark by copying most of its members
+	 *
+	 * Caveat: bookmarking a redaction bookmark yields a normal bookmark
+	 * -----------------------------------------------------------------
+	 * Reasoning:
+	 * - The zbm_redaction_obj would be referred to by both source and new
+	 *   bookmark, but would be destroyed once either source or new is
+	 *   destroyed, resulting in use-after-free of the referrred object.
+	 * - User expectation when issuing the `zfs bookmark` command is that
+	 *   a normal bookmark of the source is created
+	 *
+	 * Design Alternatives For Full Redaction Bookmark Copying:
+	 * - reference-count the redaction object => would require on-disk
+	 *   format change for existing redaction objects
+	 * - Copy the redaction object => cannot be done in syncing context
+	 *   because the redaction object might be too large
+	 */
+
+	VERIFY0(dsl_bookmark_lookup_impl(bmark_fs_source, source_shortname,
+	    &source_phys));
+	dsl_bookmark_node_t *new_dbn = dsl_bookmark_node_alloc(new_shortname);
+
+	memcpy(&new_dbn->dbn_phys, &source_phys, sizeof (source_phys));
+	new_dbn->dbn_phys.zbm_redaction_obj = 0;
+
+	/* update feature counters */
+	if (new_dbn->dbn_phys.zbm_flags & ZBM_FLAG_HAS_FBN) {
+		spa_feature_incr(dp->dp_spa,
+		    SPA_FEATURE_BOOKMARK_WRITTEN, tx);
+	}
+	/* no need for redaction bookmark counter; nulled zbm_redaction_obj */
+	/* dsl_bookmark_node_add bumps bookmarks and v2-bookmarks counter */
+
+	/*
+	 * write new bookmark
+	 *
+	 * Note that dsl_bookmark_lookup_impl guarantees that, if source is a
+	 * v1 bookmark, the v2-only fields are zeroed.
+	 * And dsl_bookmark_node_add writes back a v1-sized bookmark if
+	 * v2 bookmarks are disabled and/or v2-only fields are zeroed.
+	 * => bookmark copying works on pre-bookmark-v2 pools
+	 */
+	dsl_bookmark_node_add(bmark_fs_new, new_dbn, tx);
+
+	spa_history_log_internal_ds(bmark_fs_source, "bookmark", tx,
+	    "name=%s creation_txg=%llu source_guid=%llu",
+	    new_shortname, (longlong_t)new_dbn->dbn_phys.zbm_creation_txg,
+	    (longlong_t)source_phys.zbm_guid);
+
+	dsl_dataset_rele(bmark_fs_source, FTAG);
+	dsl_dataset_rele(bmark_fs_new, FTAG);
+}
+
+void
 dsl_bookmark_create_sync(void *arg, dmu_tx_t *tx)
 {
 	dsl_bookmark_create_arg_t *dbca = arg;
@@ -391,8 +606,19 @@ dsl_bookmark_create_sync(void *arg, dmu_tx_t *tx)
 
 	for (nvpair_t *pair = nvlist_next_nvpair(dbca->dbca_bmarks, NULL);
 	    pair != NULL; pair = nvlist_next_nvpair(dbca->dbca_bmarks, pair)) {
-		dsl_bookmark_create_sync_impl(nvpair_name(pair),
-		    fnvpair_value_string(pair), tx, 0, NULL, NULL, NULL);
+
+		char *new = nvpair_name(pair);
+		char *source = fnvpair_value_string(pair);
+
+		if (strchr(source, '@') != NULL) {
+			dsl_bookmark_create_sync_impl_snap(new, source, tx,
+			    0, NULL, NULL, NULL);
+		} else if (strchr(source, '#') != NULL) {
+			dsl_bookmark_create_sync_impl_book(new, source, tx);
+		} else {
+			panic("unreachable code");
+		}
+
 	}
 }
 
@@ -422,7 +648,6 @@ dsl_bookmark_create_redacted_check(void *arg, dmu_tx_t *tx)
 {
 	dsl_bookmark_create_redacted_arg_t *dbcra = arg;
 	dsl_pool_t *dp = dmu_tx_pool(tx);
-	dsl_dataset_t *snapds;
 	int rv = 0;
 
 	if (!spa_feature_is_enabled(dp->dp_spa,
@@ -436,13 +661,12 @@ dsl_bookmark_create_redacted_check(void *arg, dmu_tx_t *tx)
 	    sizeof (redaction_list_phys_t)) / sizeof (uint64_t))
 		return (SET_ERROR(E2BIG));
 
-	rv = dsl_dataset_hold(dp, dbcra->dbcra_snap,
-	    FTAG, &snapds);
-	if (rv == 0) {
-		rv = dsl_bookmark_create_check_impl(snapds, dbcra->dbcra_bmark,
-		    tx);
-		dsl_dataset_rele(snapds, FTAG);
-	}
+	if (dsl_bookmark_create_nvl_validate_pair(
+	    dbcra->dbcra_bmark, dbcra->dbcra_snap) != 0)
+		return (SET_ERROR(EINVAL));
+
+	rv = dsl_bookmark_create_check_impl(dp,
+	    dbcra->dbcra_bmark, dbcra->dbcra_snap);
 	return (rv);
 }
 
@@ -450,9 +674,9 @@ static void
 dsl_bookmark_create_redacted_sync(void *arg, dmu_tx_t *tx)
 {
 	dsl_bookmark_create_redacted_arg_t *dbcra = arg;
-	dsl_bookmark_create_sync_impl(dbcra->dbcra_bmark, dbcra->dbcra_snap, tx,
-	    dbcra->dbcra_numsnaps, dbcra->dbcra_snaps, dbcra->dbcra_tag,
-	    dbcra->dbcra_rl);
+	dsl_bookmark_create_sync_impl_snap(dbcra->dbcra_bmark,
+	    dbcra->dbcra_snap, tx, dbcra->dbcra_numsnaps, dbcra->dbcra_snaps,
+	    dbcra->dbcra_tag, dbcra->dbcra_rl);
 }
 
 int
diff --git a/module/zfs/zfs_ioctl.c b/module/zfs/zfs_ioctl.c
index b2517d84f..7ef54a098 100644
--- a/module/zfs/zfs_ioctl.c
+++ b/module/zfs/zfs_ioctl.c
@@ -37,6 +37,7 @@
  * Copyright 2017 RackTop Systems.
  * Copyright (c) 2017 Open-E, Inc. All Rights Reserved.
  * Copyright (c) 2019 Datto Inc.
+ * Copyright (c) 2019, 2020 by Christian Schwarz. All rights reserved.
  */
 
 /*
@@ -3614,11 +3615,13 @@ zfs_ioc_destroy_snaps(const char *poolname, nvlist_t *innvl, nvlist_t *outnvl)
 }
 
 /*
- * Create bookmarks.  Bookmark names are of the form <fs>#<bmark>.
- * All bookmarks must be in the same pool.
+ * Create bookmarks. The bookmark names are of the form <fs>#<bmark>.
+ * All bookmarks and snapshots must be in the same pool.
+ * dsl_bookmark_create_nvl_validate describes the nvlist schema in more detail.
  *
  * innvl: {
- *     bookmark1 -> snapshot1, bookmark2 -> snapshot2
+ *     new_bookmark1 -> existing_snapshot,
+ *     new_bookmark2 -> existing_bookmark,
  * }
  *
  * outnvl: bookmark -> error code (int32)
@@ -3632,25 +3635,6 @@ static const zfs_ioc_key_t zfs_keys_bookmark[] = {
 static int
 zfs_ioc_bookmark(const char *poolname, nvlist_t *innvl, nvlist_t *outnvl)
 {
-	for (nvpair_t *pair = nvlist_next_nvpair(innvl, NULL);
-	    pair != NULL; pair = nvlist_next_nvpair(innvl, pair)) {
-		char *snap_name;
-
-		/*
-		 * Verify the snapshot argument.
-		 */
-		if (nvpair_value_string(pair, &snap_name) != 0)
-			return (SET_ERROR(EINVAL));
-
-
-		/* Verify that the keys (bookmarks) are unique */
-		for (nvpair_t *pair2 = nvlist_next_nvpair(innvl, pair);
-		    pair2 != NULL; pair2 = nvlist_next_nvpair(innvl, pair2)) {
-			if (strcmp(nvpair_name(pair), nvpair_name(pair2)) == 0)
-				return (SET_ERROR(EINVAL));
-		}
-	}
-
 	return (dsl_bookmark_create(innvl, outnvl));
 }
 
@@ -4164,7 +4148,7 @@ recursive_unmount(const char *fsname, void *arg)
  * snapname is the snapshot to redact.
  * innvl: {
  *     "bookname" -> (string)
- *         name of the redaction bookmark to generate
+ *         shortname of the redaction bookmark to generate
  *     "snapnv" -> (nvlist, values ignored)
  *         snapshots to redact snapname with respect to
  * }
diff --git a/tests/runfiles/common.run b/tests/runfiles/common.run
index 81cca7eed..691cae9cb 100644
--- a/tests/runfiles/common.run
+++ b/tests/runfiles/common.run
@@ -86,7 +86,10 @@ tests = ['tst.destroy_fs', 'tst.destroy_snap', 'tst.get_count_and_limit',
     'tst.list_user_props', 'tst.parse_args_neg','tst.promote_conflict',
     'tst.promote_multiple', 'tst.promote_simple', 'tst.rollback_mult',
     'tst.rollback_one', 'tst.snapshot_destroy', 'tst.snapshot_neg',
-    'tst.snapshot_recursive', 'tst.snapshot_simple', 'tst.terminate_by_signal']
+    'tst.snapshot_recursive', 'tst.snapshot_simple',
+    'tst.bookmark.create', 'tst.bookmark.clone',
+    'tst.terminate_by_signal'
+    ]
 tags = ['functional', 'channel_program', 'synctask_core']
 
 [tests/functional/checksum]
diff --git a/tests/zfs-tests/tests/functional/cli_root/zfs_bookmark/cleanup.ksh b/tests/zfs-tests/tests/functional/cli_root/zfs_bookmark/cleanup.ksh
index 6a4e7cfc6..f84ac43e6 100755
--- a/tests/zfs-tests/tests/functional/cli_root/zfs_bookmark/cleanup.ksh
+++ b/tests/zfs-tests/tests/functional/cli_root/zfs_bookmark/cleanup.ksh
@@ -26,4 +26,6 @@
 
 . $STF_SUITE/include/libtest.shlib
 
+log_must zfs destroy "$TESTPOOL/$TESTFS/child"
+log_must zfs destroy "$TESTPOOL/${TESTFS}_with_suffix"
 default_cleanup
diff --git a/tests/zfs-tests/tests/functional/cli_root/zfs_bookmark/setup.ksh b/tests/zfs-tests/tests/functional/cli_root/zfs_bookmark/setup.ksh
index 2a9de0535..40953415c 100755
--- a/tests/zfs-tests/tests/functional/cli_root/zfs_bookmark/setup.ksh
+++ b/tests/zfs-tests/tests/functional/cli_root/zfs_bookmark/setup.ksh
@@ -28,4 +28,8 @@
 
 DISK=${DISKS%% *}
 
-default_volume_setup $DISK
+default_setup_noexit $DISK
+log_must zfs create "$TESTPOOL/$TESTFS/child"
+log_must zfs create "$TESTPOOL/${TESTFS}_with_suffix"
+log_must zfs create "$TESTPOOL/$TESTFS/recv"
+log_pass
diff --git a/tests/zfs-tests/tests/functional/cli_root/zfs_bookmark/zfs_bookmark_cliargs.ksh b/tests/zfs-tests/tests/functional/cli_root/zfs_bookmark/zfs_bookmark_cliargs.ksh
index 4a1183729..f3d516e95 100755
--- a/tests/zfs-tests/tests/functional/cli_root/zfs_bookmark/zfs_bookmark_cliargs.ksh
+++ b/tests/zfs-tests/tests/functional/cli_root/zfs_bookmark/zfs_bookmark_cliargs.ksh
@@ -22,6 +22,7 @@
 
 #
 # Copyright 2017, loli10K <[email protected]>. All rights reserved.
+# Copyright 2019, 2020 by Christian Schwarz. All rights reserved.
 #
 
 . $STF_SUITE/include/libtest.shlib
@@ -32,12 +33,22 @@
 #
 # STRATEGY:
 # 1. Create initial snapshot
+#
 # 2. Verify we can create a bookmark specifying snapshot and bookmark full paths
-# 3. Verify we can create a bookmark specifying the snapshot name
-# 4. Verify we can create a bookmark specifying the bookmark name
+# 3. Verify we can create a bookmark specifying the short snapshot name
+# 4. Verify we can create a bookmark specifying the short bookmark name
 # 5. Verify at least a full dataset path is required and both snapshot and
 #    bookmark name must be valid
 #
+# 6. Verify we can copy a bookmark by specifying the source bookmark and new
+#    bookmark full paths.
+# 7. Verify we can copy a bookmark specifying the short source name
+# 8. Verify we can copy a bookmark specifying the short new name
+# 9. Verify two short paths are not allowed, and test empty paths
+# 10. Verify we cannot copy a bookmark if the new bookmark already exists
+# 11. Verify that copying a bookmark only works if new and source name
+#     have the same dataset
+#
 
 verify_runnable "both"
 
@@ -49,18 +60,29 @@ function cleanup
 	if bkmarkexists "$DATASET#$TESTBM"; then
 		log_must zfs destroy "$DATASET#$TESTBM"
 	fi
+	if bkmarkexists "$DATASET#$TESTBMCOPY"; then
+		log_must zfs destroy "$DATASET#$TESTBMCOPY"
+	fi
 }
 
 log_assert "'zfs bookmark' should work only when passed valid arguments."
 log_onexit cleanup
 
 DATASET="$TESTPOOL/$TESTFS"
+DATASET_TWO="$TESTPOOL/${TESTFS}_two"
 TESTSNAP='snapshot'
+TESTSNAP2='snapshot2'
 TESTBM='bookmark'
+TESTBMCOPY='bookmark_copy'
+
 
 # Create initial snapshot
 log_must zfs snapshot "$DATASET@$TESTSNAP"
 
+#
+# Bookmark creation tests
+#
+
 # Verify we can create a bookmark specifying snapshot and bookmark full paths
 log_must zfs bookmark "$DATASET@$TESTSNAP" "$DATASET#$TESTBM"
 log_must eval "bkmarkexists $DATASET#$TESTBM"
@@ -97,4 +119,120 @@ log_mustnot zfs bookmark "$TESTSNAP" "$DATASET#"
 log_mustnot zfs bookmark "$TESTSNAP" "$DATASET"
 log_mustnot eval "bkmarkexists $DATASET#$TESTBM"
 
-log_pass "'zfs bookmark' works as expected only when passed valid arguments."
+# Verify that we can create a bookmarks on another origin filesystem
+log_must zfs clone "$DATASET@$TESTSNAP" "$DATASET_TWO"
+log_must zfs bookmark "$DATASET@$TESTSNAP" "$DATASET_TWO#$TESTBM"
+log_must eval "destroy_dataset $DATASET_TWO"
+
+# Verify that we can cannot create bookmarks on a non-origin filesystem
+log_must zfs create "$DATASET_TWO"
+log_mustnot_expect "source is not an ancestor of the new bookmark's dataset" zfs bookmark "$DATASET@$TESTSNAP" "$DATASET_TWO#$TESTBM"
+log_must zfs destroy "$DATASET_TWO"
+
+# Verify that we can create bookmarks of snapshots on the pool dataset
+log_must zfs snapshot "$TESTPOOL@$TESTSNAP"
+log_must zfs bookmark "$TESTPOOL@$TESTSNAP" "$TESTPOOL#$TESTBM"
+log_must zfs destroy "$TESTPOOL#$TESTBM"
+log_must zfs destroy "$TESTPOOL@$TESTSNAP"
+
+#
+# Bookmark copying tests
+#
+
+# create the source bookmark
+log_must zfs bookmark "$DATASET@$TESTSNAP" "$DATASET#$TESTBM"
+
+# Verify we can copy a bookmark by specifying the source bookmark
+# and new bookmark full paths.
+log_must eval "bkmarkexists $DATASET#$TESTBM"
+log_must zfs bookmark "$DATASET#$TESTBM" "$DATASET#$TESTBMCOPY"
+log_must eval "bkmarkexists $DATASET#$TESTBMCOPY"
+## validate destroy once (should be truly independent bookmarks)
+log_must zfs destroy "$DATASET#$TESTBM"
+log_mustnot eval "bkmarkexists $DATASET#$TESTBM"
+log_must eval "bkmarkexists $DATASET#$TESTBMCOPY"
+log_must zfs destroy "$DATASET#$TESTBMCOPY"
+log_mustnot eval "bkmarkexists $DATASET#$TESTBMCOPY"
+log_mustnot eval "bkmarkexists $DATASET#$TESTBM"
+## recreate the source bookmark
+log_must zfs bookmark "$DATASET@$TESTSNAP" "$DATASET#$TESTBM"
+
+# Verify we can copy a bookmark specifying the short source name
+log_must zfs bookmark "#$TESTBM" "$DATASET#$TESTBMCOPY"
+log_must eval "bkmarkexists $DATASET#$TESTBMCOPY"
+log_must zfs destroy "$DATASET#$TESTBMCOPY"
+
+# Verify we can copy a bookmark specifying the short bookmark name
+log_must zfs bookmark "$DATASET#$TESTBM" "#$TESTBMCOPY"
+log_must eval "bkmarkexists $DATASET#$TESTBMCOPY"
+log_must zfs destroy "$DATASET#$TESTBMCOPY"
+
+# Verify two short paths are not allowed, and test empty paths
+log_mustnot zfs bookmark "#$TESTBM" "#$TESTBMCOPY"
+log_mustnot zfs bookmark "#$TESTBM" "#"
+log_mustnot zfs bookmark "#"        "#$TESTBMCOPY"
+log_mustnot zfs bookmark "#"        "#"
+log_mustnot zfs bookmark "#"        ""
+log_mustnot zfs bookmark ""         "#"
+log_mustnot zfs bookmark ""         ""
+
+# Verify that we can copy bookmarks on another origin filesystem
+log_must zfs clone "$DATASET@$TESTSNAP" "$DATASET_TWO"
+log_must zfs bookmark "$DATASET#$TESTBM" "$DATASET_TWO#$TESTBMCOPY"
+log_must zfs destroy "$DATASET_TWO"
+
+# Verify that we can cannot create bookmarks on another non-origin filesystem
+log_must zfs create "$DATASET_TWO"
+log_mustnot_expect "source is not an ancestor of the new bookmark's dataset" zfs bookmark "$DATASET#$TESTBM" "$DATASET_TWO#$TESTBMCOPY"
+log_must zfs destroy "$DATASET_TWO"
+
+# Verify that we can copy bookmarks on the pool dataset
+log_must zfs snapshot "$TESTPOOL@$TESTSNAP"
+log_must zfs bookmark "$TESTPOOL@$TESTSNAP" "$TESTPOOL#$TESTBM"
+log_must zfs bookmark "$TESTPOOL#$TESTBM" "$TESTPOOL#$TESTBMCOPY"
+log_must zfs destroy "$TESTPOOL#$TESTBM"
+log_must zfs destroy "$TESTPOOL#$TESTBMCOPY"
+log_must zfs destroy "$TESTPOOL@$TESTSNAP"
+
+# Verify that copied 'normal' bookmarks are independent of the source bookmark
+log_must zfs bookmark "$DATASET#$TESTBM" "$DATASET#$TESTBMCOPY"
+log_must zfs destroy "$DATASET#$TESTBM"
+log_must eval "zfs send $DATASET@$TESTSNAP > $TEST_BASE_DIR/zfstest_datastream.$$"
+log_must eval "destroy_dataset $TESTPOOL/$TESTFS/recv"
+log_must eval "zfs recv -o mountpoint=none $TESTPOOL/$TESTFS/recv < $TEST_BASE_DIR/zfstest_datastream.$$"
+log_must zfs snapshot "$DATASET@$TESTSNAP2"
+log_must eval "zfs send -i \#$TESTBMCOPY $DATASET@$TESTSNAP2 > $TEST_BASE_DIR/zfstest_datastream.$$"
+log_must eval "zfs recv $TESTPOOL/$TESTFS/recv < $TEST_BASE_DIR/zfstest_datastream.$$"
+# cleanup
+log_must eval "destroy_dataset $DATASET@$TESTSNAP2"
+log_must zfs destroy "$DATASET#$TESTBMCOPY"
+log_must zfs bookmark "$DATASET@$TESTSNAP" "$DATASET#$TESTBM"
+
+# Verify that copied redaction bookmarks are independent of the source bookmark
+## create redaction bookmark
+log_must zfs destroy "$DATASET#$TESTBM"
+log_must zfs destroy "$DATASET@$TESTSNAP"
+log_must eval "echo secret > $TESTDIR/secret"
+log_must zfs snapshot "$DATASET@$TESTSNAP"
+log_must eval "echo redacted > $TESTDIR/secret"
+log_must zfs snapshot "$DATASET@$TESTSNAP2" # TESTSNAP2 is the redaction snapshot
+log_must zfs list -t all -o name,createtxg,guid,mountpoint,written
+log_must zfs redact "$DATASET@$TESTSNAP" "$TESTBM" "$DATASET@$TESTSNAP2"
+# ensure our primitive for testing whether a bookmark is a redaction bookmark works
+log_must eval "zfs get all $DATASET#$TESTBM | grep redact_snaps"
+## copy the redaction bookmark
+log_must zfs bookmark "$DATASET#$TESTBM" "#$TESTBMCOPY"
+log_must eval "zfs send --redact "$TESTBMCOPY" -i $DATASET@$TESTSNAP $DATASET@$TESTSNAP2 2>&1 | head -n 100 | grep 'internal error: Invalid argument'"
+log_mustnot eval "zfs get all $DATASET#$TESTBMCOPY | grep redact_snaps"
+# try the above again after destroying the source bookmark, preventive measure for future work
+log_must zfs destroy "$DATASET#$TESTBM"
+log_must eval "zfs send --redact "$TESTBMCOPY" -i $DATASET@$TESTSNAP $DATASET@$TESTSNAP2 2>&1 | head -n 100 | grep 'internal error: Invalid argument'"
+log_mustnot eval "zfs get all $DATASET#$TESTBMCOPY | grep redact_snaps"
+## cleanup
+log_must eval "destroy_dataset $DATASET@$TESTSNAP2"
+log_must zfs destroy "$DATASET#$TESTBMCOPY"
+log_must eval "destroy_dataset $DATASET@$TESTSNAP"
+log_must zfs snapshot "$DATASET@$TESTSNAP"
+log_must zfs bookmark "$DATASET@$TESTSNAP" "$DATASET#$TESTBM"
+
+log_pass "'zfs bookmark' works as expected"