diff --git a/cmd/zdb/zdb.c b/cmd/zdb/zdb.c
index 782eb224a771..2fa559cc1972 100644
--- a/cmd/zdb/zdb.c
+++ b/cmd/zdb/zdb.c
@@ -2050,6 +2050,32 @@ dump_all_ddts(spa_t *spa)
 	}
 
 	dump_dedup_ratio(&dds_total);
+
+	/*
+	 * Dump a histogram of unique class entry age
+	 */
+	if (dump_opt['D'] == 3 && getenv("ZDB_DDT_UNIQUE_AGE_HIST") != NULL) {
+		ddt_age_histo_t histogram;
+
+		(void) printf("DDT walk unique, building age histogram...\n");
+		ddt_prune_walk(spa, 0, &histogram);
+
+		/*
+		 * print out histogram for unique entry class birth
+		 */
+		if (histogram.dah_entries > 0) {
+			(void) printf("%5s  %9s  %4s\n",
+			    "age", "blocks", "amnt");
+			(void) printf("%5s  %9s  %4s\n",
+			    "-----", "---------", "----");
+			for (int i = 0; i < HIST_BINS; i++) {
+				(void) printf("%5d  %9d %4d%%\n", 1 << i,
+				    (int)histogram.dah_age_histo[i],
+				    (int)((histogram.dah_age_histo[i] * 100) /
+				    histogram.dah_entries));
+			}
+		}
+	}
 }
 
 static void
@@ -5808,6 +5834,9 @@ zdb_count_block(zdb_cb_t *zcb, zilog_t *zilog, const blkptr_t *bp,
 		ddt_exit(ddt);
 	}
 
+	/*
+	 * XXX - djb this fails during ztest runs (ENOENT)
+	 */
 	VERIFY3U(zio_wait(zio_claim(NULL, zcb->zcb_spa,
 	    refcnt ? 0 : spa_min_claim_txg(zcb->zcb_spa),
 	    bp, NULL, NULL, ZIO_FLAG_CANFAIL)), ==, 0);
diff --git a/cmd/zpool/zpool_main.c b/cmd/zpool/zpool_main.c
index 1d4f59c4b863..5c83df58af06 100644
--- a/cmd/zpool/zpool_main.c
+++ b/cmd/zpool/zpool_main.c
@@ -130,6 +130,8 @@ static int zpool_do_version(int, char **);
 
 static int zpool_do_wait(int, char **);
 
+static int zpool_do_ddt_prune(int, char **);
+
 static int zpool_do_help(int argc, char **argv);
 
 static zpool_compat_status_t zpool_do_load_compat(
@@ -167,6 +169,7 @@ typedef enum {
 	HELP_CLEAR,
 	HELP_CREATE,
 	HELP_CHECKPOINT,
+	HELP_DDT_PRUNE,
 	HELP_DESTROY,
 	HELP_DETACH,
 	HELP_EXPORT,
@@ -343,6 +346,8 @@ static zpool_command_t command_table[] = {
 	{ "sync",	zpool_do_sync,		HELP_SYNC		},
 	{ NULL },
 	{ "wait",	zpool_do_wait,		HELP_WAIT		},
+	{ NULL },
+	{ "ddtprune",	zpool_do_ddt_prune,	HELP_DDT_PRUNE		},
 };
 
 #define	NCOMMAND	(ARRAY_SIZE(command_table))
@@ -456,6 +461,8 @@ get_usage(zpool_help_t idx)
 	case HELP_WAIT:
 		return (gettext("\twait [-Hp] [-T d|u] [-t <activity>[,...]] "
 		    "<pool> [interval]\n"));
+	case HELP_DDT_PRUNE:
+		return (gettext("\tddtprune -d|-p <amount> <pool>\n"));
 	default:
 		__builtin_unreachable();
 	}
@@ -11667,6 +11674,88 @@ found:;
 	return (error);
 }
 
+/*
+ * zpool ddtprune -d|-p <amount> <pool>
+ *
+ *       -d <days>	Prune entries <days> old and older
+ *       -p <percent>	Prune <percent> amount of entries
+ *
+ * Prune single reference entries from DDT to satisfy the amount specified.
+ */
+int
+zpool_do_ddt_prune(int argc, char **argv)
+{
+	zpool_ddt_prune_unit_t unit = ZPOOL_DDT_PRUNE_NONE;
+	uint64_t amount = 0;
+	zpool_handle_t *zhp;
+	char *endptr;
+	int c;
+
+	while ((c = getopt(argc, argv, "d:p:")) != -1) {
+		switch (c) {
+		case 'd':
+			if (unit == ZPOOL_DDT_PRUNE_PERCENTAGE) {
+				(void) fprintf(stderr, gettext("-d cannot be "
+				    "combined with -p option\n"));
+				usage(B_FALSE);
+			}
+			errno = 0;
+			amount = strtoull(optarg, &endptr, 0);
+			if (errno != 0 || *endptr != '\0' || amount == 0) {
+				(void) fprintf(stderr,
+				    gettext("invalid days value\n"));
+				usage(B_FALSE);
+			}
+			amount *= 86400;	/* convert days to seconds */
+			unit = ZPOOL_DDT_PRUNE_AGE;
+			break;
+		case 'p':
+			if (unit == ZPOOL_DDT_PRUNE_AGE) {
+				(void) fprintf(stderr, gettext("-p cannot be "
+				    "combined with -d option\n"));
+				usage(B_FALSE);
+			}
+			errno = 0;
+			amount = strtoull(optarg, &endptr, 0);
+			if (errno != 0 || *endptr != '\0' ||
+			    amount == 0 || amount > 100) {
+				(void) fprintf(stderr,
+				    gettext("invalid percentage value\n"));
+				usage(B_FALSE);
+			}
+			unit = ZPOOL_DDT_PRUNE_PERCENTAGE;
+			break;
+		case '?':
+			(void) fprintf(stderr, gettext("invalid option '%c'\n"),
+			    optopt);
+			usage(B_FALSE);
+		}
+	}
+	argc -= optind;
+	argv += optind;
+
+	if (unit == ZPOOL_DDT_PRUNE_NONE) {
+		(void) fprintf(stderr,
+		    gettext("missing amount option (-d|-p <value>)\n"));
+		usage(B_FALSE);
+	} else if (argc < 1) {
+		(void) fprintf(stderr, gettext("missing pool argument\n"));
+		usage(B_FALSE);
+	} else if (argc > 1) {
+		(void) fprintf(stderr, gettext("too many arguments\n"));
+		usage(B_FALSE);
+	}
+	zhp = zpool_open(g_zfs, argv[0]);
+	if (zhp == NULL)
+		return (-1);
+
+	int error = zpool_ddt_prune(zhp, unit, amount);
+
+	zpool_close(zhp);
+
+	return (error);
+}
+
 static int
 find_command_idx(const char *command, int *idx)
 {
diff --git a/cmd/ztest.c b/cmd/ztest.c
index 6a9264ddcc4c..467e2c722bca 100644
--- a/cmd/ztest.c
+++ b/cmd/ztest.c
@@ -276,6 +276,8 @@ extern unsigned long zio_decompress_fail_fraction;
 extern unsigned long zfs_reconstruct_indirect_damage_fraction;
 extern uint64_t raidz_expand_max_reflow_bytes;
 extern uint_t raidz_expand_pause_point;
+extern boolean_t ddt_prune_artificial_age;
+extern boolean_t ddt_dump_prune_histogram;
 
 
 static ztest_shared_opts_t *ztest_shared_opts;
@@ -446,6 +448,7 @@ ztest_func_t ztest_fletcher;
 ztest_func_t ztest_fletcher_incr;
 ztest_func_t ztest_verify_dnode_bt;
 ztest_func_t ztest_pool_prefetch_ddt;
+ztest_func_t ztest_ddt_prune;
 
 static uint64_t zopt_always = 0ULL * NANOSEC;		/* all the time */
 static uint64_t zopt_incessant = 1ULL * NANOSEC / 10;	/* every 1/10 second */
@@ -502,6 +505,7 @@ static ztest_info_t ztest_info[] = {
 	ZTI_INIT(ztest_fletcher_incr, 1, &zopt_rarely),
 	ZTI_INIT(ztest_verify_dnode_bt, 1, &zopt_sometimes),
 	ZTI_INIT(ztest_pool_prefetch_ddt, 1, &zopt_rarely),
+	ZTI_INIT(ztest_ddt_prune, 1, &zopt_rarely),
 };
 
 #define	ZTEST_FUNCS	(sizeof (ztest_info) / sizeof (ztest_info_t))
@@ -7288,6 +7292,17 @@ ztest_trim(ztest_ds_t *zd, uint64_t id)
 	mutex_exit(&ztest_vdev_lock);
 }
 
+void
+ztest_ddt_prune(ztest_ds_t *zd, uint64_t id)
+{
+	(void) zd, (void) id;
+
+	spa_t *spa = ztest_spa;
+	uint64_t pct = ztest_random(15) + 1;
+
+	(void) ddt_prune_unique_entries(spa, ZPOOL_DDT_PRUNE_PERCENTAGE, pct);
+}
+
 /*
  * Verify pool integrity by running zdb.
  */
@@ -7469,6 +7484,13 @@ ztest_resume_thread(void *arg)
 {
 	spa_t *spa = arg;
 
+	/*
+	 * Synthesize aged DDT entries for ddt prune testing
+	 */
+	ddt_prune_artificial_age = B_TRUE;
+	if (ztest_opts.zo_verbose >= 3)
+		ddt_dump_prune_histogram = B_TRUE;
+
 	while (!ztest_exiting) {
 		if (spa_suspended(spa))
 			ztest_resume(spa);
@@ -8586,7 +8608,13 @@ ztest_init(ztest_shared_t *zs)
 		 */
 		if (i == SPA_FEATURE_LOG_SPACEMAP && ztest_random(4) == 0)
 			continue;
-
+#if 0
+		/*
+		 * split bewtween legacy and fast dedup
+		 */
+		if (i == SPA_FEATURE_FAST_DEDUP && ztest_random(2) != 0)
+			continue;
+#endif
 		VERIFY3S(-1, !=, asprintf(&buf, "feature@%s",
 		    spa_feature_table[i].fi_uname));
 		fnvlist_add_uint64(props, buf, 0);
diff --git a/contrib/debian/openzfs-zfsutils.install b/contrib/debian/openzfs-zfsutils.install
index 10083351abb5..d51e4ef003e6 100644
--- a/contrib/debian/openzfs-zfsutils.install
+++ b/contrib/debian/openzfs-zfsutils.install
@@ -100,6 +100,7 @@ usr/share/man/man8/zpool-clear.8
 usr/share/man/man8/zpool-create.8
 usr/share/man/man8/zpool-destroy.8
 usr/share/man/man8/zpool-detach.8
+usr/share/man/man8/zpool-ddtprune.8
 usr/share/man/man8/zpool-events.8
 usr/share/man/man8/zpool-export.8
 usr/share/man/man8/zpool-get.8
diff --git a/include/libzfs.h b/include/libzfs.h
index 979b919ce2fa..96bce6e8ea4b 100644
--- a/include/libzfs.h
+++ b/include/libzfs.h
@@ -304,6 +304,9 @@ _LIBZFS_H int zpool_reopen_one(zpool_handle_t *, void *);
 
 _LIBZFS_H int zpool_sync_one(zpool_handle_t *, void *);
 
+_LIBZFS_H int zpool_ddt_prune(zpool_handle_t *, zpool_ddt_prune_unit_t,
+    uint64_t);
+
 _LIBZFS_H int zpool_vdev_online(zpool_handle_t *, const char *, int,
     vdev_state_t *);
 _LIBZFS_H int zpool_vdev_offline(zpool_handle_t *, const char *, boolean_t);
diff --git a/include/libzfs_core.h b/include/libzfs_core.h
index 206e5e5c2bf6..b1d74fbbc8f5 100644
--- a/include/libzfs_core.h
+++ b/include/libzfs_core.h
@@ -161,6 +161,9 @@ _LIBZFS_CORE_H int lzc_set_vdev_prop(const char *, nvlist_t *, nvlist_t **);
 
 _LIBZFS_CORE_H int lzc_scrub(zfs_ioc_t, const char *, nvlist_t *, nvlist_t **);
 
+_LIBZFS_CORE_H int lzc_ddt_prune(const char *, zpool_ddt_prune_unit_t,
+    uint64_t);
+
 #ifdef	__cplusplus
 }
 #endif
diff --git a/include/sys/ddt.h b/include/sys/ddt.h
index 66ed17708684..71d1a3b03f90 100644
--- a/include/sys/ddt.h
+++ b/include/sys/ddt.h
@@ -406,6 +406,9 @@ extern int ddt_walk(spa_t *spa, ddt_bookmark_t *ddb,
 
 extern boolean_t ddt_addref(spa_t *spa, const blkptr_t *bp);
 
+extern int ddt_prune_unique_entries(spa_t *spa, zpool_ddt_prune_unit_t unit,
+    uint64_t amount);
+
 #ifdef	__cplusplus
 }
 #endif
diff --git a/include/sys/ddt_impl.h b/include/sys/ddt_impl.h
index b050c2c9cce2..a0de61f81160 100644
--- a/include/sys/ddt_impl.h
+++ b/include/sys/ddt_impl.h
@@ -189,8 +189,11 @@ extern void ddt_log_commit(ddt_t *ddt, ddt_log_update_t *dlu);
 
 extern boolean_t ddt_log_take_first(ddt_t *ddt, ddt_log_t *ddl,
     ddt_lightweight_entry_t *ddlwe);
-extern boolean_t ddt_log_take_key(ddt_t *ddt, ddt_log_t *ddl,
-    const ddt_key_t *ddk, ddt_lightweight_entry_t *ddlwe);
+
+extern boolean_t ddt_log_find_key(ddt_t *ddt, const ddt_key_t *ddk,
+    ddt_lightweight_entry_t *ddlwe);
+extern boolean_t ddt_log_remove_key(ddt_t *ddt, ddt_log_t *ddl,
+    const ddt_key_t *ddk);
 
 extern void ddt_log_checkpoint(ddt_t *ddt, ddt_lightweight_entry_t *ddlwe,
     dmu_tx_t *tx);
@@ -213,6 +216,44 @@ extern void ddt_log_fini(void);
  * them up.
  */
 
+/*
+ * We use a histogram to convert a percentage request into a
+ * cutoff value where entries older than the cutoff get pruned.
+ *
+ * The histogram bins represent hours in power-of-two increments.
+ * 16 bins covers up to four years.
+ */
+#define	HIST_BINS 16
+
+typedef struct ddt_age_histo {
+	uint64_t dah_entries;
+	uint64_t dah_age_histo[HIST_BINS];
+} ddt_age_histo_t;
+
+void ddt_prune_walk(spa_t *spa, uint64_t cutoff, ddt_age_histo_t *histogram);
+
+#if defined(_KERNEL) || !defined(ZFS_DEBUG)
+#define	ddt_dump_age_histogram(histo, cutoff)	((void)0)
+#else
+static inline void
+ddt_dump_age_histogram(ddt_age_histo_t *histogram, uint64_t cutoff)
+{
+	if (histogram->dah_entries == 0)
+		return;
+
+	(void) printf("DDT prune unique class age, %llu hour cutoff\n",
+	    (u_longlong_t)(gethrestime_sec() - cutoff)/3600);
+	(void) printf("%5s  %9s  %4s\n", "age", "blocks", "amnt");
+	(void) printf("%5s  %9s  %4s\n", "-----", "---------", "----");
+	for (int i = 0; i < HIST_BINS; i++) {
+		(void) printf("%5d  %9llu %4d%%\n", 1<<i,
+		    (u_longlong_t)histogram->dah_age_histo[i],
+		    (int)((histogram->dah_age_histo[i] * 100) /
+		    histogram->dah_entries));
+	}
+}
+#endif
+
 /*
  * Enough room to expand DMU_POOL_DDT format for all possible DDT
  * checksum/class/type combinations.
diff --git a/include/sys/fs/zfs.h b/include/sys/fs/zfs.h
index 4358654c7484..f3ab435d78f0 100644
--- a/include/sys/fs/zfs.h
+++ b/include/sys/fs/zfs.h
@@ -1419,7 +1419,7 @@ typedef enum {
  */
 typedef enum zfs_ioc {
 	/*
-	 * Core features - 88/128 numbers reserved.
+	 * Core features - 89/128 numbers reserved.
 	 */
 #ifdef __FreeBSD__
 	ZFS_IOC_FIRST =	0,
@@ -1516,6 +1516,7 @@ typedef enum zfs_ioc {
 	ZFS_IOC_VDEV_SET_PROPS,			/* 0x5a56 */
 	ZFS_IOC_POOL_SCRUB,			/* 0x5a57 */
 	ZFS_IOC_POOL_PREFETCH,			/* 0x5a58 */
+	ZFS_IOC_DDT_PRUNE,			/* 0x5a59 */
 
 	/*
 	 * Per-platform (Optional) - 8/128 numbers reserved.
@@ -1652,6 +1653,12 @@ typedef enum {
 	ZPOOL_PREFETCH_DDT
 } zpool_prefetch_type_t;
 
+typedef enum {
+	ZPOOL_DDT_PRUNE_NONE,
+	ZPOOL_DDT_PRUNE_AGE,		/* in seconds */
+	ZPOOL_DDT_PRUNE_PERCENTAGE,	/* 1 - 100 */
+} zpool_ddt_prune_unit_t;
+
 /*
  * Bookmark name values.
  */
@@ -1745,6 +1752,12 @@ typedef enum {
  */
 #define	ZPOOL_PREFETCH_TYPE		"prefetch_type"
 
+/*
+ * The following are names used when invoking ZFS_IOC_DDT_PRUNE.
+ */
+#define	DDT_PRUNE_UNIT		"ddt_prune_unit"
+#define	DDT_PRUNE_AMOUNT	"ddt_prune_amount"
+
 /*
  * Flags for ZFS_IOC_VDEV_SET_STATE
  */
diff --git a/include/sys/spa_impl.h b/include/sys/spa_impl.h
index 47f349327461..8cf1b808c960 100644
--- a/include/sys/spa_impl.h
+++ b/include/sys/spa_impl.h
@@ -411,6 +411,7 @@ struct spa {
 	uint64_t	spa_dedup_dspace;	/* Cache get_dedup_dspace() */
 	uint64_t	spa_dedup_checksum;	/* default dedup checksum */
 	uint64_t	spa_dspace;		/* dspace in normal class */
+	boolean_t	spa_active_ddt_prune;	/* ddt prune process active */
 	struct brt	*spa_brt;		/* in-core BRT */
 	kmutex_t	spa_vdev_top_lock;	/* dueling offline/remove */
 	kmutex_t	spa_proc_lock;		/* protects spa_proc* */
diff --git a/lib/libzfs/libzfs_pool.c b/lib/libzfs/libzfs_pool.c
index 3f2dc7314cf6..d4493625f9f6 100644
--- a/lib/libzfs/libzfs_pool.c
+++ b/lib/libzfs/libzfs_pool.c
@@ -5549,3 +5549,31 @@ zpool_set_vdev_prop(zpool_handle_t *zhp, const char *vdevname,
 
 	return (ret);
 }
+
+/*
+ *
+ */
+int
+zpool_ddt_prune(zpool_handle_t *zhp, zpool_ddt_prune_unit_t unit,
+    uint64_t amount)
+{
+	int error = lzc_ddt_prune(zhp->zpool_name, unit, amount);
+	if (error != 0) {
+		libzfs_handle_t *hdl = zhp->zpool_hdl;
+		char errbuf[ERRBUFLEN];
+
+		(void) snprintf(errbuf, sizeof (errbuf), dgettext(TEXT_DOMAIN,
+		    "cannot prune dedup table on '%s'"), zhp->zpool_name);
+
+		if (error == EALREADY) {
+			zfs_error_aux(hdl, dgettext(TEXT_DOMAIN,
+			    "a prune operation is already in progress"));
+			(void) zfs_error(hdl, EZFS_BUSY, errbuf);
+		} else {
+			(void) zpool_standard_error(hdl, errno, errbuf);
+		}
+		return (-1);
+	}
+
+	return (0);
+}
diff --git a/lib/libzfs_core/libzfs_core.c b/lib/libzfs_core/libzfs_core.c
index ec8b0ff4f61c..d07fca6cebad 100644
--- a/lib/libzfs_core/libzfs_core.c
+++ b/lib/libzfs_core/libzfs_core.c
@@ -1927,3 +1927,25 @@ lzc_get_bootenv(const char *pool, nvlist_t **outnvl)
 {
 	return (lzc_ioctl(ZFS_IOC_GET_BOOTENV, pool, NULL, outnvl));
 }
+
+/*
+ * Prune the specified amount from the pool's dedup table.
+ */
+int
+lzc_ddt_prune(const char *pool, zpool_ddt_prune_unit_t unit, uint64_t amount)
+{
+	int error;
+
+	nvlist_t *result = NULL;
+	nvlist_t *args = fnvlist_alloc();
+
+	fnvlist_add_int32(args, DDT_PRUNE_UNIT, unit);
+	fnvlist_add_uint64(args, DDT_PRUNE_AMOUNT, amount);
+
+	error = lzc_ioctl(ZFS_IOC_DDT_PRUNE, pool, args, &result);
+
+	fnvlist_free(args);
+	fnvlist_free(result);
+
+	return (error);
+}
diff --git a/man/Makefile.am b/man/Makefile.am
index 194bb4721619..fde704933764 100644
--- a/man/Makefile.am
+++ b/man/Makefile.am
@@ -72,6 +72,7 @@ dist_man_MANS = \
 	%D%/man8/zpool-create.8 \
 	%D%/man8/zpool-destroy.8 \
 	%D%/man8/zpool-detach.8 \
+	%D%/man8/zpool-ddtprune.8 \
 	%D%/man8/zpool-events.8 \
 	%D%/man8/zpool-export.8 \
 	%D%/man8/zpool-get.8 \
diff --git a/man/man8/zpool-ddtprune.8 b/man/man8/zpool-ddtprune.8
new file mode 100644
index 000000000000..1ab7d3982c3e
--- /dev/null
+++ b/man/man8/zpool-ddtprune.8
@@ -0,0 +1,48 @@
+.\"
+.\" CDDL HEADER START
+.\"
+.\" The contents of this file are subject to the terms of the
+.\" Common Development and Distribution License (the "License").
+.\" You may not use this file except in compliance with the License.
+.\"
+.\" You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
+.\" or http://www.opensolaris.org/os/licensing.
+.\" See the License for the specific language governing permissions
+.\" and limitations under the License.
+.\"
+.\" When distributing Covered Code, include this CDDL HEADER in each
+.\" file and include the License file at usr/src/OPENSOLARIS.LICENSE.
+.\" If applicable, add the following below this CDDL HEADER, with the
+.\" fields enclosed by brackets "[]" replaced with your own identifying
+.\" information: Portions Copyright [yyyy] [name of copyright owner]
+.\"
+.\" CDDL HEADER END
+.\"
+.\"
+.\" Copyright (c) 2024, Klara Inc.
+.\"
+.Dd June 17, 2024
+.Dt ZPOOL-DDTPRUNE 8
+.Os
+.
+.Sh NAME
+.Nm zpool-ddtprune
+.Nd Prunes the oldest entries from the single reference dedup table(s)
+.Sh SYNOPSIS
+.Nm zpool
+.Cm ddtprune
+.Fl d Ar days | Fl p Ar percentage
+.Ar pool
+.Sh DESCRIPTION
+This command prunes older unique entries from the dedup table.
+As a complement to the dedup quota feature,
+.Sy ddtprune
+allows removal of older non-duplicate entries to make room for
+newer duplicate entries.
+.Pp
+The amount to prune can be based on a target percentage of the unique entries
+or based on the age (i.e., every unique entry older than N days).
+.
+.Sh SEE ALSO
+.Xr zdb 8 ,
+.Xr zpool-status 8
diff --git a/man/man8/zpool.8 b/man/man8/zpool.8
index 2b966b72bf4c..feb3083dad72 100644
--- a/man/man8/zpool.8
+++ b/man/man8/zpool.8
@@ -589,6 +589,7 @@ don't wait.
 .Xr zpool-create 8 ,
 .Xr zpool-destroy 8 ,
 .Xr zpool-detach 8 ,
+.Xr zpool-ddtprune 8 ,
 .Xr zpool-events 8 ,
 .Xr zpool-export 8 ,
 .Xr zpool-get 8 ,
diff --git a/module/zfs/ddt.c b/module/zfs/ddt.c
index 7f102d5eed68..b47ad8de1849 100644
--- a/module/zfs/ddt.c
+++ b/module/zfs/ddt.c
@@ -125,6 +125,13 @@
  * without which, no space would be recovered and the DDT would continue to be
  * considered "over quota". See zap_shrink_enabled.
  *
+ * ## Dedup table pruning
+ *
+ * As a complement to the dedup quota feature, ddtprune allows removal of older
+ * non-duplicate entries to make room for newer duplicate entries. The amount
+ * to prune can be based on a target percentage of the unique entries or based
+ * on the age (i.e., prune unique entry older than N days).
+ *
  * ## Dedup log
  *
  * Historically, all entries modified on a txg were written back to dedup
@@ -228,6 +235,19 @@ int zfs_dedup_prefetch = 0;
  */
 uint_t dedup_class_wait_txgs = 5;
 
+/*
+ * How many DDT prune entries to add to the DDT sync AVL tree.
+ * Note these addtional entries have a memory footprint of a
+ * ddt_entry_t (216 bytes).
+ */
+static uint32_t zfs_ddt_prunes_per_txg = 50000;
+
+/*
+ * For testing, synthesize aged DDT entries
+ * (in global scope for ztest)
+ */
+boolean_t ddt_prune_artificial_age = B_FALSE;
+boolean_t ddt_dump_prune_histogram = B_FALSE;
 
 /*
  * Don't do more than this many incremental flush passes per txg.
@@ -716,6 +736,30 @@ ddt_phys_clear(ddt_univ_phys_t *ddp, ddt_phys_variant_t v)
 		memset(&ddp->ddp_trad[v], 0, DDT_TRAD_PHYS_SIZE / DDT_PHYS_MAX);
 }
 
+static uint64_t
+ddt_class_start(void)
+{
+	uint64_t start = gethrestime_sec();
+
+	if (ddt_prune_artificial_age) {
+		/*
+		 * debug aide -- simulate a wider distribution
+		 * so we don't have to wait for an aged DDT
+		 * to test prune.
+		 */
+		int range = 1 << 21;
+		int percent = random_in_range(100);
+		if (percent < 50) {
+			range = range >> 4;
+		} else if (percent > 75) {
+			range /= 2;
+		}
+		start -= random_in_range(range);
+	}
+
+	return (start);
+}
+
 void
 ddt_phys_addref(ddt_univ_phys_t *ddp, ddt_phys_variant_t v)
 {
@@ -1020,6 +1064,47 @@ ddt_prefetch_all(spa_t *spa)
 
 static int ddt_configure(ddt_t *ddt, boolean_t new);
 
+/*
+ * If the BP passed to ddt_lookup has valid DVAs, then we need to compare them
+ * to the ones in the entry. If they're different, then the passed-in BP is
+ * from a previous generation of this entry (ie was previously pruned) and we
+ * have to act like the entry doesn't exist at all.
+ *
+ * This should only happen during a lookup to free the block (zio_ddt_free()).
+ *
+ * XXX this is similar in spirit to ddt_phys_select(), maybe can combine
+ *       -- robn, 2024-02-09
+ */
+static boolean_t
+ddt_entry_lookup_is_valid(ddt_t *ddt, const blkptr_t *bp, ddt_entry_t *dde)
+{
+	/* If the BP has no DVAs, then this entry is good */
+	uint_t ndvas = BP_GET_NDVAS(bp);
+	if (ndvas == 0)
+		return (B_TRUE);
+
+	/*
+	 * Only checking the phys for the copies. For flat, there's only one;
+	 * for trad it'll be the one that has the matching set of DVAs.
+	 */
+	const dva_t *dvas = (ddt->ddt_flags & DDT_FLAG_FLAT) ?
+	    dde->dde_phys->ddp_flat.ddp_dva :
+	    dde->dde_phys->ddp_trad[ndvas].ddp_dva;
+
+	/*
+	 * Compare entry DVAs with the BP. They should all be there, but
+	 * there's not really anything we can do if its only partial anyway,
+	 * that's an error somewhere else, maybe long ago.
+	 */
+	uint_t d;
+	for (d = 0; d < ndvas; d++)
+		if (!DVA_EQUAL(&dvas[d], &bp->blk_dva[d]))
+			return (B_FALSE);
+	ASSERT3U(d, ==, ndvas);
+
+	return (B_TRUE);
+}
+
 ddt_entry_t *
 ddt_lookup(ddt_t *ddt, const blkptr_t *bp, boolean_t add)
 {
@@ -1055,8 +1140,11 @@ ddt_lookup(ddt_t *ddt, const blkptr_t *bp, boolean_t add)
 
 		/* If it's already loaded, we can just return it. */
 		DDT_KSTAT_BUMP(ddt, dds_lookup_live_hit);
-		if (dde->dde_flags & DDE_FLAG_LOADED)
-			return (dde);
+		if (dde->dde_flags & DDE_FLAG_LOADED) {
+			if (ddt_entry_lookup_is_valid(ddt, bp, dde))
+				return (dde);
+			return (NULL);
+		}
 
 		/* Someone else is loading it, wait for it. */
 		dde->dde_waiters++;
@@ -1075,7 +1163,11 @@ ddt_lookup(ddt_t *ddt, const blkptr_t *bp, boolean_t add)
 		}
 
 		DDT_KSTAT_BUMP(ddt, dds_lookup_existing);
-		return (dde);
+
+		/* Make sure the loaded entry matches the BP */
+		if (ddt_entry_lookup_is_valid(ddt, bp, dde))
+			return (dde);
+		return (NULL);
 	} else
 		DDT_KSTAT_BUMP(ddt, dds_lookup_live_miss);
 
@@ -1088,32 +1180,42 @@ ddt_lookup(ddt_t *ddt, const blkptr_t *bp, boolean_t add)
 
 	/* Record the time this class was created (used by ddt prune) */
 	if (ddt->ddt_flags & DDT_FLAG_FLAT)
-		dde->dde_phys->ddp_flat.ddp_class_start = gethrestime_sec();
+		dde->dde_phys->ddp_flat.ddp_class_start = ddt_class_start();
 
 	avl_insert(&ddt->ddt_tree, dde, where);
 
 	/* If its in the log tree, we can "load" it from there */
 	if (ddt->ddt_flags & DDT_FLAG_LOG) {
 		ddt_lightweight_entry_t ddlwe;
-		boolean_t found = B_FALSE;
-
-		if (ddt_log_take_key(ddt, ddt->ddt_log_active,
-		    &search, &ddlwe)) {
-			DDT_KSTAT_BUMP(ddt, dds_lookup_log_active_hit);
-			found = B_TRUE;
-		} else if (ddt_log_take_key(ddt, ddt->ddt_log_flushing,
-		    &search, &ddlwe)) {
-			DDT_KSTAT_BUMP(ddt, dds_lookup_log_flushing_hit);
-			found = B_TRUE;
-		}
-
-		if (found) {
-			dde->dde_flags = DDE_FLAG_LOADED | DDE_FLAG_LOGGED;
 
+		if (ddt_log_find_key(ddt, &search, &ddlwe)) {
+			/*
+			 * See if we have the key first, and if so, set up
+			 * the entry.
+			 */
 			dde->dde_type = ddlwe.ddlwe_type;
 			dde->dde_class = ddlwe.ddlwe_class;
 			memcpy(dde->dde_phys, &ddlwe.ddlwe_phys,
 			    DDT_PHYS_SIZE(ddt));
+			/* Whatever we found isn't valid for this BP, eject */
+			if (!ddt_entry_lookup_is_valid(ddt, bp, dde)) {
+				avl_remove(&ddt->ddt_tree, dde);
+				ddt_free(ddt, dde);
+				return (NULL);
+			}
+
+			/* Remove it and count it */
+			if (ddt_log_remove_key(ddt,
+			    ddt->ddt_log_active, &search)) {
+				DDT_KSTAT_BUMP(ddt, dds_lookup_log_active_hit);
+			} else {
+				VERIFY(ddt_log_remove_key(ddt,
+				    ddt->ddt_log_flushing, &search));
+				DDT_KSTAT_BUMP(ddt,
+				    dds_lookup_log_flushing_hit);
+			}
+
+			dde->dde_flags = DDE_FLAG_LOADED | DDE_FLAG_LOGGED;
 
 			DDT_KSTAT_BUMP(ddt, dds_lookup_log_hit);
 			DDT_KSTAT_BUMP(ddt, dds_lookup_existing);
@@ -1152,6 +1254,8 @@ ddt_lookup(ddt_t *ddt, const blkptr_t *bp, boolean_t add)
 	dde->dde_type = type;	/* will be DDT_TYPES if no entry found */
 	dde->dde_class = class;	/* will be DDT_CLASSES if no entry found */
 
+	boolean_t valid = B_TRUE;
+
 	if (dde->dde_type == DDT_TYPES &&
 	    dde->dde_class == DDT_CLASSES &&
 	    ddt_over_quota(spa)) {
@@ -1165,6 +1269,24 @@ ddt_lookup(ddt_t *ddt, const blkptr_t *bp, boolean_t add)
 		/* Flag cleanup required */
 		dde->dde_flags |= DDE_FLAG_OVERQUOTA;
 	} else if (error == 0) {
+		/*
+		 * If what we loaded is no good for this BP and there's no one
+		 * waiting for it, we can just remove it and get out. If its no
+		 * good but there are waiters, we have to leave it, because we
+		 * don't know what they want. If its not needed we'll end up
+		 * taking an entry log/sync, but it can only happen if more
+		 * than one previous version of this block is being deleted at
+		 * the same time. This is extremely unlikely to happen and not
+		 * worth the effort to deal with without taking an entry
+		 * update.
+		 */
+		valid = ddt_entry_lookup_is_valid(ddt, bp, dde);
+		if (!valid && dde->dde_waiters == 0) {
+			avl_remove(&ddt->ddt_tree, dde);
+			ddt_free(ddt, dde);
+			return (NULL);
+		}
+
 		DDT_KSTAT_BUMP(ddt, dds_lookup_stored_hit);
 		DDT_KSTAT_BUMP(ddt, dds_lookup_existing);
 
@@ -1193,7 +1315,10 @@ ddt_lookup(ddt_t *ddt, const blkptr_t *bp, boolean_t add)
 	dde->dde_flags |= DDE_FLAG_LOADED;
 	cv_broadcast(&dde->dde_cv);
 
-	return (dde->dde_flags & DDE_FLAG_OVERQUOTA ? NULL : dde);
+	if ((dde->dde_flags & DDE_FLAG_OVERQUOTA) || !valid)
+		return (NULL);
+
+	return (dde);
 }
 
 void
@@ -1300,9 +1425,8 @@ ddt_destroy_dir(ddt_t *ddt, dmu_tx_t *tx)
 }
 
 /*
- * Determine, flags and on-disk layout from what's already stored. If there's
- * nothing stored, then if new is false, returns ENOENT, and if true, selects
- * based on pool config.
+ * Determine, flags and on-disk layout from what's already stored, or
+ * if there is nothing stored, select based on pool config.
  */
 static int
 ddt_configure(ddt_t *ddt, boolean_t new)
@@ -1505,12 +1629,6 @@ ddt_load(spa_t *spa)
 
 	ddt_create(spa);
 
-	error = zap_lookup(spa->spa_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
-	    DMU_POOL_DDT_STATS, sizeof (uint64_t), 1,
-	    &spa->spa_ddt_stat_object);
-	if (error)
-		return (error == ENOENT ? 0 : error);
-
 	for (enum zio_checksum c = 0; c < ZIO_CHECKSUM_FUNCTIONS; c++) {
 		if (!DDT_CHECKSUM_VALID(c))
 			continue;
@@ -1521,7 +1639,19 @@ ddt_load(spa_t *spa)
 			continue;
 		if (error != 0)
 			return (error);
+	}
 
+	error = zap_lookup(spa->spa_meta_objset, DMU_POOL_DIRECTORY_OBJECT,
+	    DMU_POOL_DDT_STATS, sizeof (uint64_t), 1,
+	    &spa->spa_ddt_stat_object);
+	if (error)
+		return (error == ENOENT ? 0 : error);
+
+	for (enum zio_checksum c = 0; c < ZIO_CHECKSUM_FUNCTIONS; c++) {
+		if (!DDT_CHECKSUM_VALID(c))
+			continue;
+
+		ddt_t *ddt = spa->spa_ddt[c];
 		for (ddt_type_t type = 0; type < DDT_TYPES; type++) {
 			for (ddt_class_t class = 0; class < DDT_CLASSES;
 			    class++) {
@@ -1816,7 +1946,7 @@ ddt_sync_flush_entry(ddt_t *ddt, ddt_lightweight_entry_t *ddlwe,
 		uint64_t phys_refcnt = ddt_phys_refcnt(ddp, v);
 
 		if (ddt_phys_birth(ddp, v) == 0) {
-			ASSERT3U(phys_refcnt, ==, 0);
+			ASSERT0(phys_refcnt);
 			continue;
 		}
 		if (DDT_PHYS_IS_DITTO(ddt, p)) {
@@ -2290,8 +2420,9 @@ ddt_walk_ready(spa_t *spa)
 	return (B_TRUE);
 }
 
-int
-ddt_walk(spa_t *spa, ddt_bookmark_t *ddb, ddt_lightweight_entry_t *ddlwe)
+static int
+ddt_walk_impl(spa_t *spa, ddt_bookmark_t *ddb, ddt_lightweight_entry_t *ddlwe,
+    uint64_t flags, boolean_t wait)
 {
 	do {
 		do {
@@ -2300,7 +2431,11 @@ ddt_walk(spa_t *spa, ddt_bookmark_t *ddb, ddt_lightweight_entry_t *ddlwe)
 				if (ddt == NULL)
 					continue;
 
-				if (ddt->ddt_flush_force_txg > 0)
+				if (flags != 0 &&
+				    (ddt->ddt_flags & flags) != flags)
+					continue;
+
+				if (wait && ddt->ddt_flush_force_txg > 0)
 					return (EAGAIN);
 
 				int error = ENOENT;
@@ -2324,13 +2459,19 @@ ddt_walk(spa_t *spa, ddt_bookmark_t *ddb, ddt_lightweight_entry_t *ddlwe)
 	return (SET_ERROR(ENOENT));
 }
 
+int
+ddt_walk(spa_t *spa, ddt_bookmark_t *ddb, ddt_lightweight_entry_t *ddlwe)
+{
+	return (ddt_walk_impl(spa, ddb, ddlwe, 0, B_TRUE));
+}
+
 /*
  * This function is used by Block Cloning (brt.c) to increase reference
  * counter for the DDT entry if the block is already in DDT.
  *
  * Return false if the block, despite having the D bit set, is not present
- * in the DDT. Currently this is not possible but might be in the future.
- * See the comment below.
+ * in the DDT. This is possible when the DDT has been pruned by an admin
+ * or by the DDT quota mechanism.
  */
 boolean_t
 ddt_addref(spa_t *spa, const blkptr_t *bp)
@@ -2361,28 +2502,13 @@ ddt_addref(spa_t *spa, const blkptr_t *bp)
 		int p = DDT_PHYS_FOR_COPIES(ddt, BP_GET_NDVAS(bp));
 		ddt_phys_variant_t v = DDT_PHYS_VARIANT(ddt, p);
 
-		/*
-		 * This entry already existed (dde_type is real), so it must
-		 * have refcnt >0 at the start of this txg. We are called from
-		 * brt_pending_apply(), before frees are issued, so the refcnt
-		 * can't be lowered yet. Therefore, it must be >0. We assert
-		 * this because if the order of BRT and DDT interactions were
-		 * ever to change and the refcnt was ever zero here, then
-		 * likely further action is required to fill out the DDT entry,
-		 * and this is a place that is likely to be missed in testing.
-		 */
-		ASSERT3U(ddt_phys_refcnt(dde->dde_phys, v), >, 0);
-
 		ddt_phys_addref(dde->dde_phys, v);
 		result = B_TRUE;
 	} else {
 		/*
-		 * At the time of implementating this if the block has the
-		 * DEDUP flag set it must exist in the DEDUP table, but
-		 * there are many advocates that want ability to remove
-		 * entries from DDT with refcnt=1. If this will happen,
-		 * we may have a block with the DEDUP set, but which doesn't
-		 * have a corresponding entry in the DDT. Be ready.
+		 * If the block has the DEDUP flag set it still might not
+		 * exist in the DEDUP table due to DDT pruning of entries
+		 * where refcnt=1.
 		 */
 		ddt_remove(ddt, dde);
 		result = B_FALSE;
@@ -2394,6 +2520,264 @@ ddt_addref(spa_t *spa, const blkptr_t *bp)
 	return (result);
 }
 
+typedef struct ddt_prune_entry {
+	ddt_t		*dpe_ddt;
+	ddt_key_t	dpe_key;
+	list_node_t	dpe_node;
+	ddt_univ_phys_t	dpe_phys[];
+} ddt_prune_entry_t;
+
+typedef struct ddt_prune_info {
+	spa_t		*dpi_spa;
+	uint64_t	dpi_txg_syncs;
+	uint64_t	dpi_pruned;
+	list_t		dpi_candidates;
+} ddt_prune_info_t;
+
+/*
+ * Add prune candidates for ddt_sync during spa_sync
+ */
+static void
+prune_candidates_sync(void *arg, dmu_tx_t *tx)
+{
+	(void) tx;
+	ddt_prune_info_t *dpi = arg;
+	ddt_prune_entry_t *dpe;
+
+	spa_config_enter(dpi->dpi_spa, SCL_ZIO, FTAG, RW_READER);
+
+	/* Process the prune candidates collected so far */
+	while ((dpe = list_remove_head(&dpi->dpi_candidates)) != NULL) {
+		blkptr_t blk;
+		ddt_t *ddt = dpe->dpe_ddt;
+
+		ddt_enter(ddt);
+
+		/*
+		 * If it's on the live list, then it was loaded for update
+		 * this txg and is no longer stale; skip it.
+		 */
+		if (avl_find(&ddt->ddt_tree, &dpe->dpe_key, NULL)) {
+			ddt_exit(ddt);
+			kmem_free(dpe, sizeof (*dpe));
+			continue;
+		}
+
+		ddt_bp_create(ddt->ddt_checksum, &dpe->dpe_key,
+		    dpe->dpe_phys, DDT_PHYS_FLAT, &blk);
+
+		ddt_entry_t *dde = ddt_lookup(ddt, &blk, B_TRUE);
+		if (dde != NULL && !(dde->dde_flags & DDE_FLAG_LOGGED)) {
+			ASSERT(dde->dde_flags & DDE_FLAG_LOADED);
+			/*
+			 * Zero the physical, so we don't try to free DVAs
+			 * at flush nor try to reuse this entry.
+			 */
+			ddt_phys_clear(dde->dde_phys, DDT_PHYS_FLAT);
+
+			dpi->dpi_pruned++;
+		}
+
+		ddt_exit(ddt);
+		kmem_free(dpe, sizeof (*dpe));
+	}
+
+	spa_config_exit(dpi->dpi_spa, SCL_ZIO, FTAG);
+	dpi->dpi_txg_syncs++;
+}
+
+/*
+ * Prune candidates are collected in open context and processed
+ * in sync context as part of ddt_sync_table().
+ */
+static void
+ddt_prune_entry(list_t *list, ddt_t *ddt, const ddt_key_t *ddk,
+    const ddt_univ_phys_t *ddp)
+{
+	ASSERT(ddt->ddt_flags & DDT_FLAG_FLAT);
+
+	size_t dpe_size = sizeof (ddt_prune_entry_t) + DDT_FLAT_PHYS_SIZE;
+	ddt_prune_entry_t *dpe = kmem_alloc(dpe_size, KM_SLEEP);
+
+	dpe->dpe_ddt = ddt;
+	dpe->dpe_key = *ddk;
+	memcpy(dpe->dpe_phys, ddp, DDT_FLAT_PHYS_SIZE);
+	list_insert_head(list, dpe);
+}
+
+/*
+ * Interate over all the entries in the DDT unique class.
+ * The walk will perform one of the following operations:
+ *  (a) build a histogram than can be used when pruning
+ *  (b) prune entries older than the cutoff
+ *
+ *  Also called by zdb(8) to dump the age histogram
+ */
+void
+ddt_prune_walk(spa_t *spa, uint64_t cutoff, ddt_age_histo_t *histogram)
+{
+	ddt_bookmark_t ddb = {
+		.ddb_class = DDT_CLASS_UNIQUE,
+		.ddb_type = 0,
+		.ddb_checksum = 0,
+		.ddb_cursor = 0
+	};
+	ddt_lightweight_entry_t ddlwe = {0};
+	int error;
+	int total = 0, valid = 0;
+	int candidates = 0;
+	uint64_t now = gethrestime_sec();
+	ddt_prune_info_t dpi;
+	boolean_t pruning = (cutoff != 0);
+
+	if (pruning) {
+		dpi.dpi_txg_syncs = 0;
+		dpi.dpi_pruned = 0;
+		dpi.dpi_spa = spa;
+		list_create(&dpi.dpi_candidates, sizeof (ddt_prune_entry_t),
+		    offsetof(ddt_prune_entry_t, dpe_node));
+	}
+
+	if (histogram != NULL)
+		memset(histogram, 0, sizeof (ddt_age_histo_t));
+
+	while ((error =
+	    ddt_walk_impl(spa, &ddb, &ddlwe, DDT_FLAG_FLAT, B_FALSE)) == 0) {
+		ddt_t *ddt = spa->spa_ddt[ddb.ddb_checksum];
+		VERIFY(ddt);
+
+		if (spa_shutting_down(spa) || issig())
+			break;
+		total++;
+
+		ASSERT(ddt->ddt_flags & DDT_FLAG_FLAT);
+		ASSERT3U(ddlwe.ddlwe_nphys, ==, 1);
+		ASSERT3U(ddlwe.ddlwe_phys.ddp_flat.ddp_refcnt, <=, 1);
+
+		uint64_t class_start =
+		    ddlwe.ddlwe_phys.ddp_flat.ddp_class_start;
+		if (class_start == 0)
+			continue;
+
+		/*
+		 * If this entry is on the log, then the stored entry is stale
+		 * and we should skip it.
+		 */
+		if (ddt_log_find_key(ddt, &ddlwe.ddlwe_key, NULL))
+			continue;
+
+		/* prune older entries */
+		if (pruning && class_start < cutoff) {
+			if (candidates++ >= zfs_ddt_prunes_per_txg) {
+				/* sync prune candidates in batches */
+				VERIFY0(dsl_sync_task(spa_name(spa),
+				    NULL, prune_candidates_sync,
+				    &dpi, 0, ZFS_SPACE_CHECK_NONE));
+				candidates = 1;
+			}
+			ddt_prune_entry(&dpi.dpi_candidates, ddt,
+			    &ddlwe.ddlwe_key, &ddlwe.ddlwe_phys);
+		}
+
+		/* build a histogram */
+		if (histogram != NULL) {
+			uint64_t age = MAX(1, (now - class_start) / 3600);
+			int bin = MIN(highbit64(age) - 1, HIST_BINS - 1);
+			histogram->dah_entries++;
+			histogram->dah_age_histo[bin]++;
+		}
+
+		valid++;
+	}
+
+	if (pruning && valid > 0) {
+		if (!list_is_empty(&dpi.dpi_candidates)) {
+			/* sync out final batch of prune candidates */
+			VERIFY0(dsl_sync_task(spa_name(spa), NULL,
+			    prune_candidates_sync, &dpi, 0,
+			    ZFS_SPACE_CHECK_NONE));
+		}
+		list_destroy(&dpi.dpi_candidates);
+
+		zfs_dbgmsg("pruned %llu entries (%d%%) across %llu txg syncs",
+		    (u_longlong_t)dpi.dpi_pruned,
+		    (int)((dpi.dpi_pruned * 100) / valid),
+		    (u_longlong_t)dpi.dpi_txg_syncs);
+	}
+}
+
+static uint64_t
+ddt_total_entries(spa_t *spa)
+{
+	ddt_object_t ddo;
+	ddt_get_dedup_object_stats(spa, &ddo);
+
+	return (ddo.ddo_count);
+}
+
+int
+ddt_prune_unique_entries(spa_t *spa, zpool_ddt_prune_unit_t unit,
+    uint64_t amount)
+{
+	uint64_t cutoff;
+	uint64_t start_time = gethrtime();
+
+	if (spa->spa_active_ddt_prune)
+		return (EALREADY);
+	if (ddt_total_entries(spa) == 0)
+		return (0);
+
+	spa->spa_active_ddt_prune = B_TRUE;
+
+	zfs_dbgmsg("prune %llu %s", (u_longlong_t)amount,
+	    unit == ZPOOL_DDT_PRUNE_PERCENTAGE ? "%" : "seconds old or older");
+
+	if (unit == ZPOOL_DDT_PRUNE_PERCENTAGE) {
+		ddt_age_histo_t histogram;
+		uint64_t oldest = 0;
+
+		/* Make a pass over DDT to build a histogram */
+		ddt_prune_walk(spa, 0, &histogram);
+
+		int target = (histogram.dah_entries * amount) / 100;
+
+		/*
+		 * Figure out our cutoff date
+		 * (i.e., which bins to prune from)
+		 */
+		for (int i = HIST_BINS - 1; i >= 0 && target > 0; i--) {
+			if (histogram.dah_age_histo[i] != 0) {
+				/* less than this bucket remaining */
+				if (target < histogram.dah_age_histo[i]) {
+					oldest = MAX(1, (1<<i) * 3600);
+					target = 0;
+				} else {
+					target -= histogram.dah_age_histo[i];
+				}
+			}
+		}
+		cutoff = gethrestime_sec() - oldest;
+
+		if (ddt_dump_prune_histogram)
+			ddt_dump_age_histogram(&histogram, cutoff);
+	} else if (unit == ZPOOL_DDT_PRUNE_AGE) {
+		cutoff = gethrestime_sec() - amount;
+	} else {
+		return (EINVAL);
+	}
+
+	if (cutoff > 0 && !spa_shutting_down(spa) && !issig()) {
+		/* Traverse DDT to prune entries older that our cuttoff */
+		ddt_prune_walk(spa, cutoff, NULL);
+	}
+
+	zfs_dbgmsg("%s: prune completed in %llu ms",
+	    spa_name(spa), (u_longlong_t)NSEC2MSEC(gethrtime() - start_time));
+
+	spa->spa_active_ddt_prune = B_FALSE;
+	return (0);
+}
+
 ZFS_MODULE_PARAM(zfs_dedup, zfs_dedup_, prefetch, INT, ZMOD_RW,
 	"Enable prefetching dedup-ed blks");
 
diff --git a/module/zfs/ddt_log.c b/module/zfs/ddt_log.c
index cabe44db0917..eca209af7b9b 100644
--- a/module/zfs/ddt_log.c
+++ b/module/zfs/ddt_log.c
@@ -353,16 +353,15 @@ ddt_log_take_first(ddt_t *ddt, ddt_log_t *ddl, ddt_lightweight_entry_t *ddlwe)
 }
 
 boolean_t
-ddt_log_take_key(ddt_t *ddt, ddt_log_t *ddl, const ddt_key_t *ddk,
-    ddt_lightweight_entry_t *ddlwe)
+ddt_log_remove_key(ddt_t *ddt, ddt_log_t *ddl, const ddt_key_t *ddk)
 {
 	ddt_log_entry_t *ddle = avl_find(&ddl->ddl_tree, ddk, NULL);
 	if (ddle == NULL)
 		return (B_FALSE);
 
-	DDT_LOG_ENTRY_TO_LIGHTWEIGHT(ddt, ddle, ddlwe);
-
-	ddt_histogram_sub_entry(ddt, &ddt->ddt_log_histogram, ddlwe);
+	ddt_lightweight_entry_t ddlwe;
+	DDT_LOG_ENTRY_TO_LIGHTWEIGHT(ddt, ddle, &ddlwe);
+	ddt_histogram_sub_entry(ddt, &ddt->ddt_log_histogram, &ddlwe);
 
 	avl_remove(&ddl->ddl_tree, ddle);
 	kmem_cache_free(ddt->ddt_flags & DDT_FLAG_FLAT ?
@@ -371,6 +370,21 @@ ddt_log_take_key(ddt_t *ddt, ddt_log_t *ddl, const ddt_key_t *ddk,
 	return (B_TRUE);
 }
 
+boolean_t
+ddt_log_find_key(ddt_t *ddt, const ddt_key_t *ddk,
+    ddt_lightweight_entry_t *ddlwe)
+{
+	ddt_log_entry_t *ddle =
+	    avl_find(&ddt->ddt_log_active->ddl_tree, ddk, NULL);
+	if (!ddle)
+		ddle = avl_find(&ddt->ddt_log_flushing->ddl_tree, ddk, NULL);
+	if (!ddle)
+		return (B_FALSE);
+	if (ddlwe)
+		DDT_LOG_ENTRY_TO_LIGHTWEIGHT(ddt, ddle, ddlwe);
+	return (B_TRUE);
+}
+
 void
 ddt_log_checkpoint(ddt_t *ddt, ddt_lightweight_entry_t *ddlwe, dmu_tx_t *tx)
 {
diff --git a/module/zfs/zfs_ioctl.c b/module/zfs/zfs_ioctl.c
index 897335dd4e4f..6049fdc925bb 100644
--- a/module/zfs/zfs_ioctl.c
+++ b/module/zfs/zfs_ioctl.c
@@ -4314,6 +4314,51 @@ zfs_ioc_pool_trim(const char *poolname, nvlist_t *innvl, nvlist_t *outnvl)
 	return (total_errors > 0 ? SET_ERROR(EINVAL) : 0);
 }
 
+#define	DDT_PRUNE_UNIT		"ddt_prune_unit"
+#define	DDT_PRUNE_AMOUNT	"ddt_prune_amount"
+
+/*
+ * innvl: {
+ *     "ddt_prune_unit" -> uint32_t
+ *     "ddt_prune_amount" -> uint64_t
+ * }
+ *
+ * outnvl: "waited" -> boolean_t
+ */
+static const zfs_ioc_key_t zfs_keys_ddt_prune[] = {
+	{DDT_PRUNE_UNIT,	DATA_TYPE_INT32,	0},
+	{DDT_PRUNE_AMOUNT,	DATA_TYPE_UINT64,	0},
+};
+
+static int
+zfs_ioc_ddt_prune(const char *poolname, nvlist_t *innvl, nvlist_t *outnvl)
+{
+	int32_t unit;
+	uint64_t amount;
+
+	if (nvlist_lookup_int32(innvl, DDT_PRUNE_UNIT, &unit) != 0 ||
+	    nvlist_lookup_uint64(innvl, DDT_PRUNE_AMOUNT, &amount) != 0) {
+		return (EINVAL);
+	}
+
+	spa_t *spa;
+	int error = spa_open(poolname, &spa, FTAG);
+	if (error != 0)
+		return (error);
+
+	if (!spa_feature_is_enabled(spa, SPA_FEATURE_FAST_DEDUP)) {
+		spa_close(spa, FTAG);
+		return (SET_ERROR(ENOTSUP));
+	}
+
+	error = ddt_prune_unique_entries(spa, (zpool_ddt_prune_unit_t)unit,
+	    amount);
+
+	spa_close(spa, FTAG);
+
+	return (error);
+}
+
 /*
  * This ioctl waits for activity of a particular type to complete. If there is
  * no activity of that type in progress, it returns immediately, and the
@@ -7402,6 +7447,11 @@ zfs_ioctl_init(void)
 	    POOL_CHECK_NONE, B_FALSE, B_FALSE,
 	    zfs_keys_get_props, ARRAY_SIZE(zfs_keys_get_props));
 
+	zfs_ioctl_register("zpool_ddt_prune", ZFS_IOC_DDT_PRUNE,
+	    zfs_ioc_ddt_prune, zfs_secpolicy_config, POOL_NAME,
+	    POOL_CHECK_SUSPENDED | POOL_CHECK_READONLY, B_TRUE, B_TRUE,
+	    zfs_keys_ddt_prune, ARRAY_SIZE(zfs_keys_ddt_prune));
+
 	/* IOCTLS that use the legacy function signature */
 
 	zfs_ioctl_register_legacy(ZFS_IOC_POOL_FREEZE, zfs_ioc_pool_freeze,
diff --git a/module/zfs/zio.c b/module/zfs/zio.c
index 2e44aa5aaf50..0a1b636e1e59 100644
--- a/module/zfs/zio.c
+++ b/module/zfs/zio.c
@@ -3836,6 +3836,16 @@ zio_ddt_free(zio_t *zio)
 	}
 	ddt_exit(ddt);
 
+	/*
+	 * When no entry was found, it must have been pruned,
+	 * so we can free it now instead of decrementing the
+	 * refcount in the DDT.
+	 */
+	if (!dde) {
+		BP_SET_DEDUP(bp, 0);
+		zio->io_pipeline |= ZIO_STAGE_DVA_FREE;
+	}
+
 	return (zio);
 }