Make data-sources responsible for their own metadata entries.

Add standard metadata functions to the extension API.

12 files changed, 334 insertions, 346 deletions

diff --git a/dist/filelist b/dist/filelist
index 1b5fc9bd4ed..0fd79caab1a 100644
--- a/dist/filelist
+++ b/dist/filelist
@@ -79,9 +79,9 @@ src/lsm/lsm_meta.c
 src/lsm/lsm_stat.c
 src/lsm/lsm_tree.c
 src/lsm/lsm_worker.c
-src/meta/meta_api.c
 src/meta/meta_apply.c
 src/meta/meta_ckpt.c
+src/meta/meta_ext.c
 src/meta/meta_table.c
 src/meta/meta_track.c
 src/meta/meta_turtle.c
diff --git a/ext/test/memrata/memrata.c b/ext/test/memrata/memrata.c
index bfad539c323..53d0687a838 100644
--- a/ext/test/memrata/memrata.c
+++ b/ext/test/memrata/memrata.c
@@ -68,26 +68,19 @@
 } while (0)
 
 /*
- * I think it's pretty unlikely we'll ever need to revise the version, but
- * it can't hurt to be cautious.
+ * Version each file, out of sheer raging paranoia.
  */
 #define	KVS_MAJOR	1			/* KVS major, minor version */
 #define	KVS_MINOR	0
 
 /*
  * We partition the flat space into a set of objects based on a packed, 8B
- * leading byte.
- *
- * Object ID 1 is for master records; object ID from 1 to KVS_MAXID_BASE are
- * reserved for future use (and, I suppose object ID 0 would work as well).
- * I can't think of any reason I'd need them, but it's a cheap and easy backup
+ * leading byte.  Object IDs from 0 to KVS_MAXID_BASE are reserved for future
+ * use -- I can't think of any reason I'd need them, but it's a cheap backup
  * plan.
  */
-#define	KVS_MASTER_ID		1		/* Master object ID */
-#define	KVS_MASTER_VALUE_MAX	256		/* Maximum master value */
-
-#define	KVS_MAXID	"WiredTiger.maxid"	/* Maximum object ID key */
-#define	KVS_MAXID_BASE	6			/* First object ID */
+#define	KVS_MAXID	"memrata:WiredTiger.maxid"	/* max object ID key */
+#define	KVS_MAXID_BASE	6				/* first object ID */
 
 /*
  * Each KVS source supports a set of URIs (named objects).  Cursors reference
@@ -97,7 +90,7 @@ typedef struct __uri_source {
 	char *name;				/* Unique name */
 	pthread_rwlock_t lock;			/* Lock */
 
-	int	configured;			/* If configured */
+	int	configured;			/* If URI source configured */
 	u_int	ref;				/* Active reference count */
 
 	/*
@@ -1336,142 +1329,117 @@ uri_truncate(WT_DATA_SOURCE *wtds, WT_SESSION *session, URI_SOURCE *us)
 }
 
 /*
- * master_key_set --
- *	Set a master key from a key, checking the length.
+ * master_id_get --
+ *	Return the maximum file ID in the system.
  */
 static int
-master_key_set(WT_DATA_SOURCE *wtds,
-    WT_SESSION *session, const char *key, struct kvs_record *r)
+master_id_get(WT_DATA_SOURCE *wtds, WT_SESSION *session, uint64_t *maxidp)
 {
 	DATA_SOURCE *ds;
 	WT_EXTENSION_API *wtext;
-	size_t idlen, len;
 	int ret = 0;
-	uint8_t id[KVS_MAX_PACKED_8B];
+	const char *value;
 
 	ds = (DATA_SOURCE *)wtds;
 	wtext = ds->wtext;
 
-	if ((ret = wtext->struct_size(wtext,
-	    session, &idlen, "r", KVS_MASTER_ID)) != 0 ||
-	    (ret = wtext->struct_pack(wtext,
-	    session, id, sizeof(id), "r", KVS_MASTER_ID)) != 0)
-		return (ret);
+	if ((ret =
+	    wtext->metadata_read(wtext, session, KVS_MAXID, &value)) == 0) {
+		*maxidp = strtouq(value, NULL, 10);
+		return (0);
+	}
+	if (ret == WT_NOTFOUND) {
+		*maxidp = KVS_MAXID_BASE;
+		return (0);
+	}
+	ERET(wtext, session, ret, "%s: %s", KVS_MAXID, wtext->strerror(ret));
+}
+
+/*
+ * master_id_set --
+ *	Increment the maximum file ID in the system.
+ */
+static int
+master_id_set(WT_DATA_SOURCE *wtds, WT_SESSION *session, uint64_t maxid)
+{
+	DATA_SOURCE *ds;
+	WT_EXTENSION_API *wtext;
+	int ret = 0;
+	char value[32];			/* Large enough for any 8B value. */
 
-	/* Check to see if the ID and key can fit into a key. */
-	len = strlen(key);
-	if (idlen + len > KVS_MAX_KEY_LEN)
-		ERET(wtext, session, EINVAL, "%s: too large for a Memrata key");
+	ds = (DATA_SOURCE *)wtds;
+	wtext = ds->wtext;
 
-	memcpy((uint8_t *)r->key, id, idlen);
-	memcpy((uint8_t *)r->key + idlen, key, len);
-	r->key_len = idlen + len;
-	return (0);
+	(void)snprintf(value, sizeof(value), "%" PRIu64, maxid);
+	if ((ret =
+	    wtext->metadata_update(wtext, session, KVS_MAXID, value)) == 0)
+		return (0);
+	ERET(wtext, session, ret, "%s: %s", KVS_MAXID, wtext->strerror(ret));
 }
 
 /*
- * master_id_get --
- *	Return the maximum file ID in the system.
+ * master_uri_get --
+ *	Get the KVS master record for a URI.
  */
 static int
-master_id_get(
-    WT_DATA_SOURCE *wtds, WT_SESSION *session, kvs_t kvs, uint64_t *maxidp)
+master_uri_get(WT_DATA_SOURCE *wtds,
+    WT_SESSION *session, const char *uri, const char **valuep)
 {
-	struct kvs_record *r, _r;
 	DATA_SOURCE *ds;
 	WT_EXTENSION_API *wtext;
 	int ret = 0;
-	char key[KVS_MAX_KEY_LEN], val[KVS_MASTER_VALUE_MAX];
-
-	*maxidp = KVS_MAXID_BASE;
 
 	ds = (DATA_SOURCE *)wtds;
 	wtext = ds->wtext;
 
-	r = &_r;
-	memset(r, 0, sizeof(*r));
-	r->key = key;
-	if ((ret = master_key_set(wtds, session, KVS_MAXID, r)) != 0)
-		return (ret);
-	r->val = val;
-	r->val_len = sizeof(val);
-	switch (ret = kvs_get(kvs, r, 0UL, (unsigned long)sizeof(val))) {
-	case 0:
-		*maxidp = strtouq(r->val, NULL, 10);
-		break;
-	case KVS_E_KEY_NOT_FOUND:
-		ret = 0;
-		break;
-	default:
-		ERET(wtext, session,
-		    WT_ERROR, "kvs_get: %s: %s", KVS_MAXID, kvs_strerror(ret));
-		/* NOTREACHED */
-	}
-	return (ret);
+	return (wtext->metadata_read(wtext, session, uri, valuep));
 }
 
 /*
- * master_id_set --
- *	Increment the maximum file ID in the system.
+ * master_uri_drop --
+ *	Drop the KVS master record for a URI.
  */
 static int
-master_id_set(
-    WT_DATA_SOURCE *wtds, WT_SESSION *session, kvs_t kvs, uint64_t maxid)
+master_uri_drop(WT_DATA_SOURCE *wtds, WT_SESSION *session, const char *uri)
 {
-	struct kvs_record *r, _r;
 	DATA_SOURCE *ds;
 	WT_EXTENSION_API *wtext;
 	int ret = 0;
-	char key[KVS_MAX_KEY_LEN], val[KVS_MASTER_VALUE_MAX];
 
 	ds = (DATA_SOURCE *)wtds;
 	wtext = ds->wtext;
 
-	r = &_r;
-	memset(r, 0, sizeof(*r));
-	r->key = key;
-	if ((ret = master_key_set(wtds, session, KVS_MAXID, r)) != 0)
-		return (ret);
-	r->val = val;
-	r->val_len = (size_t)snprintf(val, sizeof(val), "%" PRIu64, maxid);
-	if ((ret = kvs_set(kvs, r)) != 0)
-		ERET(wtext, session,
-		    WT_ERROR, "kvs_set: %s: %s", KVS_MAXID, kvs_strerror(ret));
-	if ((ret = kvs_commit(kvs)) != 0)
-		ERET(wtext, session,
-		    WT_ERROR, "kvs_commit: %s", kvs_strerror(ret));
-	return (0);
+	return (wtext->metadata_remove(wtext, session, uri));
 }
 
 /*
- * master_uri_get --
- *	Get the KVS master record for a URI.
+ * master_uri_rename --
+ *	Rename the KVS master record for a URI.
  */
 static int
-master_uri_get(WT_DATA_SOURCE *wtds,
-    WT_SESSION *session, const char *uri, kvs_t kvs, char *val)
+master_uri_rename(WT_DATA_SOURCE *wtds,
+    WT_SESSION *session, const char *uri, const char *newuri)
 {
-	struct kvs_record *r, _r;
 	DATA_SOURCE *ds;
 	WT_EXTENSION_API *wtext;
 	int ret = 0;
-	char key[KVS_MAX_KEY_LEN];
+	const char *value;
 
 	ds = (DATA_SOURCE *)wtds;
 	wtext = ds->wtext;
 
-	r = &_r;
-	memset(r, 0, sizeof(*r));
-	r->key = key;
-	if ((ret = master_key_set(wtds, session, uri, r)) != 0)
+	/* Insert the record under a new name. */
+	if ((ret = master_uri_get(wtds, session, uri, &value)) != 0 ||
+	    (ret = wtext->metadata_insert(wtext, session, newuri, value)) != 0)
 		return (ret);
-	r->val = val;
-	r->val_len = KVS_MASTER_VALUE_MAX;
-	if ((ret = kvs_get(kvs, r, 0UL, KVS_MASTER_VALUE_MAX)) == 0)
-		return (0);
-	if (ret == KVS_E_KEY_NOT_FOUND)
-		return (WT_NOTFOUND);
-	ERET(wtext, session, WT_ERROR, "kvs_get: %s", uri, kvs_strerror(ret));
+
+	/*
+	 * Remove the original record, and if that fails, attempt to remove
+	 * the new record.
+	 */
+	if ((ret = wtext->metadata_remove(wtext, session, uri)) != 0)
+		(void)wtext->metadata_remove(wtext, session, newuri);
+	return (ret);
 }
 
 /*
@@ -1480,66 +1448,68 @@ master_uri_get(WT_DATA_SOURCE *wtds,
  */
 static int
 master_uri_set(WT_DATA_SOURCE *wtds,
-    WT_SESSION *session, const char *uri, kvs_t kvs, WT_CONFIG_ARG *config)
+    WT_SESSION *session, const char *uri, WT_CONFIG_ARG *config)
 {
-	struct kvs_record *r, _r;
 	DATA_SOURCE *ds;
 	WT_CONFIG_ITEM a, b;
 	WT_EXTENSION_API *wtext;
 	uint64_t maxid;
-	int ret = 0;
-	char key[KVS_MAX_KEY_LEN], val[KVS_MASTER_VALUE_MAX];
+	int exclusive, ret = 0;
+	char value[1024];
 
 	ds = (DATA_SOURCE *)wtds;
 	wtext = ds->wtext;
 
 	/* Get the maximum file ID. */
-	if ((ret = master_id_get(wtds, session, kvs, &maxid)) != 0)
+	if ((ret = master_id_get(wtds, session, &maxid)) != 0)
 		return (ret);
 	++maxid;
 
+	exclusive = 0;
+	if ((ret =
+	    wtext->config_get(wtext, session, config, "exclusive", &a)) == 0)
+		exclusive = a.val != 0;
+	else if (ret != WT_NOTFOUND)
+		ERET(wtext, session, ret,
+		    "exclusive configuration: %s", wtext->strerror(ret));
+
 	/* Get the key/value format strings. */
 	if ((ret = wtext->config_get(
 	    wtext, session, config, "key_format", &a)) != 0)
-		ERET(wtext, session, ret,
-		    "key_format configuration: %s", wtext->strerror(ret));
+		if (ret == WT_NOTFOUND) {
+			a.str = "u";
+			a.len = 1;
+		} else
+			ERET(wtext, session, ret,
+			    "key_format configuration: %s",
+			    wtext->strerror(ret));
 	if ((ret = wtext->config_get(
 	    wtext, session, config, "value_format", &b)) != 0)
-		ERET(wtext, session, ret,
-		    "value_format configuration: %s", wtext->strerror(ret));
+		if (ret == WT_NOTFOUND) {
+			b.str = "u";
+			b.len = 1;
+		} else
+			ERET(wtext, session, ret,
+			    "value_format configuration: %s",
+			    wtext->strerror(ret));
 
 	/*
-	 * Create a new reference using kvs_add (which fails if the record
-	 * already exists).
+	 * Create a new reference using insert (which fails if the record
+	 * already exists).  If that succeeds, we just used up a unique ID,
+	 * update the master ID record.
 	 */
-	r = &_r;
-	memset(r, 0, sizeof(*r));
-	r->key = key;
-	if ((ret = master_key_set(wtds, session, uri, r)) != 0)
-		return (ret);
-	r->val = val;
-	r->val_len = (size_t)snprintf(val, sizeof(val),
-	    "key_generator=1,uid=%" PRIu64
-	    ",version=(major=%d,minor=%d),key_format=%.*s,value_format=%.*s",
+	(void)snprintf(value, sizeof(value),
+	    "uid=%" PRIu64
+	    ",version=(major=%d,minor=%d)"
+	    ",key_format=%.*s,value_format=%.*s",
 	    maxid, KVS_MAJOR, KVS_MINOR, (int)a.len, a.str, (int)b.len, b.str);
-	if (r->val_len >= sizeof(val))
-		ERET(wtext, session, WT_ERROR, "master URI value too large");
-	++r->val_len;				/* Include the trailing nul. */
-	switch (ret = kvs_add(kvs, r)) {
-	case 0:
-		if ((ret = master_id_set(wtds, session, kvs, maxid)) != 0)
-			return (ret);
-		if ((ret = kvs_commit(kvs)) != 0)
-			ERET(wtext, session,
-			    WT_ERROR, "kvs_commit: %s", kvs_strerror(ret));
-		break;
-	case KVS_E_KEY_EXISTS:
-		break;
-	default:
-		ERET(wtext,
-		    session, WT_ERROR, "kvs_add: %s", uri, kvs_strerror(ret));
-	}
-	return (0);
+	if ((ret = wtext->metadata_insert(wtext, session, uri, value)) == 0)
+		return (master_id_set(wtds, session, maxid));
+
+	if (ret == WT_DUPLICATE_KEY)
+		return (exclusive ? EEXIST : 0);
+
+	ERET(wtext, session, ret, "%s: %s", KVS_MAXID, wtext->strerror(ret));
 }
 
 /*
@@ -1553,7 +1523,6 @@ kvs_session_create(WT_DATA_SOURCE *wtds,
 	DATA_SOURCE *ds;
 	URI_SOURCE *us;
 	WT_EXTENSION_API *wtext;
-	kvs_t kvs;
 	int ret = 0;
 
 	ds = (DATA_SOURCE *)wtds;
@@ -1562,10 +1531,9 @@ kvs_session_create(WT_DATA_SOURCE *wtds,
 	/* Get a locked reference to the URI. */
 	if ((ret = uri_source_open(wtds, session, config, uri, 0, &us)) != 0)
 		return (ret);
-	kvs = us->ks->kvs;
 
 	/* Create the URI master record if it doesn't already exist. */
-	ret = master_uri_set(wtds, session, uri, kvs, config);
+	ret = master_uri_set(wtds, session, uri, config);
 
 	/* Unlock the URI. */
 	ETRET(unlock(wtext, session, &us->lock));
@@ -1581,14 +1549,11 @@ static int
 kvs_session_drop(WT_DATA_SOURCE *wtds,
     WT_SESSION *session, const char *uri, WT_CONFIG_ARG *config)
 {
-	struct kvs_record *r, _r;
 	DATA_SOURCE *ds;
 	KVS_SOURCE *ks;
 	URI_SOURCE **p, *us;
 	WT_EXTENSION_API *wtext;
-	kvs_t kvs;
 	int ret = 0;
-	char key[KVS_MAX_KEY_LEN];
 
 	ds = (DATA_SOURCE *)wtds;
 	wtext = ds->wtext;
@@ -1597,7 +1562,6 @@ kvs_session_drop(WT_DATA_SOURCE *wtds,
 	if ((ret = uri_source_open(wtds, session, config, uri, 1, &us)) != 0)
 		return (ret);
 	ks = us->ks;
-	kvs = us->ks->kvs;
 
 	/* If there are active references to the object, we're busy. */
 	if (us->ref != 0) {
@@ -1605,6 +1569,10 @@ kvs_session_drop(WT_DATA_SOURCE *wtds,
 		goto err;
 	}
 
+	/* Discard all of the rows in the object. */
+	if ((ret = uri_truncate(wtds, session, us)) != 0)
+		goto err;
+
 	/*
 	 * Remove the entry from the URI_SOURCE list -- it's a singly-linked
 	 * list, find the reference to it.
@@ -1613,8 +1581,8 @@ kvs_session_drop(WT_DATA_SOURCE *wtds,
 		if (*p == us)
 			break;
 	/*
-	 * We should be guaranteed to find an entry, after all, we just looked
-	 * it up, and everything is locked down.
+	 * We should be guaranteed to find an entry, we just looked it up and
+	 * everything is locked down.
 	 */
 	if (*p == NULL) {
 		ret = WT_NOTFOUND;
@@ -1622,24 +1590,8 @@ kvs_session_drop(WT_DATA_SOURCE *wtds,
 	}
 	*p = (*p)->next;
 
-	/*
-	 * Create a new reference to the newname, using kvs_add (which fails
-	 * if the record already exists).
-	 */
-	r = &_r;
-	memset(r, 0, sizeof(*r));
-	r->key = key;
-	if ((ret = master_key_set(wtds, session, uri, r)) != 0)
-		goto err;
-	r->val = NULL;
-	r->val_len = 0;
-	if ((ret = kvs_del(kvs, r)) != 0) {
-		ESET(wtext, session, WT_ERROR,
-		    "kvs_del: %s: %s", uri, kvs_strerror(ret));
-		goto err;
-	}
-
-	ret = uri_truncate(wtds, session, us);
+	/* Discard the metadata entry. */
+	ret = master_uri_drop(wtds, session, uri);
 
 err:	ETRET(unlock(wtext, session, &us->lock));
 	ETRET(unlock(wtext, session, &ds->global_lock));
@@ -1660,15 +1612,15 @@ kvs_session_open_cursor(WT_DATA_SOURCE *wtds, WT_SESSION *session,
 	WT_CONFIG_ITEM v;
 	WT_CURSOR *wtcursor;
 	WT_EXTENSION_API *wtext;
-	kvs_t kvs;
 	int ret = 0;
-	char val[KVS_MASTER_VALUE_MAX];
+	const char *value;
 
 	*new_cursor = NULL;
 
 	cursor = NULL;
 	ds = (DATA_SOURCE *)wtds;
 	wtext = ds->wtext;
+	value = NULL;
 
 	/* Allocate and initialize a cursor. */
 	if ((cursor = calloc(1, sizeof(CURSOR))) == NULL)
@@ -1710,7 +1662,6 @@ kvs_session_open_cursor(WT_DATA_SOURCE *wtds, WT_SESSION *session,
 	/* Get a locked reference to the URI. */
 	if ((ret = uri_source_open(wtds, session, config, uri, 0, &us)) != 0)
 		goto err;
-	kvs = us->ks->kvs;
 
 	/*
 	 * Finish initializing the cursor (if the URI_SOURCE structure requires
@@ -1726,10 +1677,10 @@ kvs_session_open_cursor(WT_DATA_SOURCE *wtds, WT_SESSION *session,
 	if (!us->configured) {
 		us->configured = 1;
 
-		if ((ret = master_uri_get(wtds, session, uri, kvs, val)) != 0)
+		if ((ret = master_uri_get(wtds, session, uri, &value)) != 0)
 			goto err;
 		if ((ret = wtext->config_strget(
-		    wtext, session, val, "uid", &v)) != 0) {
+		    wtext, session, value, "uid", &v)) != 0) {
 			ESET(wtext, session, ret,
 			    "WT_EXTENSION_API.config: uid: %s",
 			    wtext->strerror(ret));
@@ -1755,7 +1706,7 @@ kvs_session_open_cursor(WT_DATA_SOURCE *wtds, WT_SESSION *session,
 		}
 
 		if ((ret = wtext->config_strget(
-		    wtext, session, val, "key_format", &v)) != 0) {
+		    wtext, session, value, "key_format", &v)) != 0) {
 			ESET(wtext, session, ret,
 			    "key_format configuration: %s",
 			    wtext->strerror(ret));
@@ -1764,7 +1715,7 @@ kvs_session_open_cursor(WT_DATA_SOURCE *wtds, WT_SESSION *session,
 		us->config_recno = v.len == 1 && v.str[0] == 'r';
 
 		if ((ret = wtext->config_strget(
-		    wtext, session, val, "value_format", &v)) != 0) {
+		    wtext, session, value, "value_format", &v)) != 0) {
 			ESET(wtext, session, ret,
 			    "value_format configuration: %s",
 			    wtext->strerror(ret));
@@ -1805,6 +1756,7 @@ err:		if (cursor != NULL) {
 			free(cursor);
 		}
 	}
+	free((void *)value);
 	return (ret);
 }
 
@@ -1814,27 +1766,21 @@ err:		if (cursor != NULL) {
  */
 static int
 kvs_session_rename(WT_DATA_SOURCE *wtds, WT_SESSION *session,
-    const char *uri, const char *newname, WT_CONFIG_ARG *config)
+    const char *uri, const char *newuri, WT_CONFIG_ARG *config)
 {
-	struct kvs_record *r, _r;
 	DATA_SOURCE *ds;
 	URI_SOURCE *us;
 	WT_EXTENSION_API *wtext;
-	kvs_t kvs;
 	int ret = 0;
-	char *copy, key[KVS_MAX_KEY_LEN], val[KVS_MASTER_VALUE_MAX];
+	char *copy;
 
 	ds = (DATA_SOURCE *)wtds;
 	wtext = ds->wtext;
-
-	/* Get a copy of the new name. */
-	if ((copy = strdup(newname)) == NULL)
-		return (os_errno());
+	copy = NULL;
 
 	/* Get a locked reference to the URI. */
 	if ((ret = uri_source_open(wtds, session, config, uri, 1, &us)) != 0)
 		return (ret);
-	kvs = us->ks->kvs;
 
 	/* If there are active references to the object, we're busy. */
 	if (us->ref != 0) {
@@ -1842,65 +1788,18 @@ kvs_session_rename(WT_DATA_SOURCE *wtds, WT_SESSION *session,
 		goto err;
 	}
 
-	/* Get the master record for the original name. */
-	if ((ret = master_uri_get(wtds, session, uri, kvs, val)) != 0)
-		goto err;
+	/* Get a copy of the new name to simplify cleanup. */
+	if ((copy = strdup(newuri)) == NULL)
+		return (os_errno());
 
-	/*
-	 * Create a new reference to the newname, using kvs_add (which fails
-	 * if the record already exists).
-	 */
-	r = &_r;
-	memset(r, 0, sizeof(*r));
-	r->key = key;
-	if ((ret = master_key_set(wtds, session, newname, r)) != 0)
+	/* Update the metadata record. */
+	if ((ret = master_uri_rename(wtds, session, uri, newuri)) != 0)
 		goto err;
-	r->val = val;
-	r->val_len = strlen(val) + 1;		/* Include the trailing nul. */
-	switch (ret = kvs_add(kvs, r)) {
-	case 0:
-		/* Remove the original entry. */
-		if ((ret = master_key_set(wtds, session, uri, r)) != 0)
-			goto cleanup;
-		r->val = NULL;
-		r->val_len = 0;
-		if ((ret = kvs_del(kvs, r)) != 0) {
-			ESET(wtext, session, WT_ERROR,
-			    "kvs_del: %s: %s", uri, kvs_strerror(ret));
-
-cleanup:		/*
-			 * Try and remove the new entry if we can't remove the
-			 * old entry.
-			 */
-			if ((ret =
-			    master_key_set(wtds, session, newname, r)) == 0 &&
-			    (ret = kvs_del(kvs, r)) != 0)
-				ESET(wtext, session, WT_ERROR,
-				    "kvs_del: %s: %s",
-				    newname, kvs_strerror(ret));
-			goto err;
-		}
 
-		/* Copy in the new name. */
-		free(us->name);
-		us->name = copy;
-		copy = NULL;
-
-		/* Flush the change. */
-		if ((ret = kvs_commit(kvs)) != 0) {
-			ESET(wtext, session,
-			    WT_ERROR, "kvs_commit: %s", kvs_strerror(ret));
-			goto err;
-		}
-		break;
-	case KVS_E_KEY_EXISTS:
-		ESET(wtext, session, EEXIST, "%s", newname);
-		break;
-	default:
-		ESET(wtext, session,
-		    WT_ERROR, "kvs_add: %s", newname, kvs_strerror(ret));
-		break;
-	}
+	/* Swap our copy of the name. */
+	free(us->name);
+	us->name = copy;
+	copy = NULL;
 
 err:	ETRET(unlock(wtext, session, &us->lock));
 	ETRET(unlock(wtext, session, &ds->global_lock));
diff --git a/src/conn/conn_api.c b/src/conn/conn_api.c
index d5680b14941..82fdee0f7dc 100644
--- a/src/conn/conn_api.c
+++ b/src/conn/conn_api.c
@@ -31,6 +31,10 @@ __conn_get_extension_api(WT_CONNECTION *wt_conn)
 	conn->extension_api.config_scan_begin = __wt_ext_config_scan_begin;
 	conn->extension_api.config_scan_end = __wt_ext_config_scan_end;
 	conn->extension_api.config_scan_next = __wt_ext_config_scan_next;
+	conn->extension_api.metadata_insert = __wt_ext_metadata_insert;
+	conn->extension_api.metadata_read = __wt_ext_metadata_read;
+	conn->extension_api.metadata_remove = __wt_ext_metadata_remove;
+	conn->extension_api.metadata_update = __wt_ext_metadata_update;
 	conn->extension_api.struct_pack = __wt_ext_struct_pack;
 	conn->extension_api.struct_size = __wt_ext_struct_size;
 	conn->extension_api.struct_unpack = __wt_ext_struct_unpack;
diff --git a/src/include/extern.h b/src/include/extern.h
index 7db5dc06358..013123f2ba9 100644
--- a/src/include/extern.h
+++ b/src/include/extern.h
@@ -745,13 +745,6 @@ extern int __wt_lsm_tree_worker(WT_SESSION_IMPL *session,
 extern void *__wt_lsm_merge_worker(void *vargs);
 extern void *__wt_lsm_bloom_worker(void *arg);
 extern void *__wt_lsm_checkpoint_worker(void *arg);
-extern int __wt_metadata_get(WT_SESSION *session,
-    const char *uri,
-    const char **valuep);
-extern int __wt_metadata_get_ckptlist( WT_SESSION *session,
-    const char *name,
-    WT_CKPT **ckptbasep);
-extern void __wt_metadata_free_ckptlist(WT_SESSION *session, WT_CKPT *ckptbase);
 extern int __wt_meta_btree_apply(WT_SESSION_IMPL *session,
     int (*func)(WT_SESSION_IMPL *,
     const char *[]),
@@ -774,6 +767,25 @@ extern int __wt_meta_ckptlist_set( WT_SESSION_IMPL *session,
 extern void __wt_meta_ckptlist_free(WT_SESSION_IMPL *session,
     WT_CKPT *ckptbase);
 extern void __wt_meta_checkpoint_free(WT_SESSION_IMPL *session, WT_CKPT *ckpt);
+extern int __wt_ext_metadata_insert(WT_EXTENSION_API *wt_api,
+    WT_SESSION *wt_session,
+    const char *key,
+    const char *value);
+extern int __wt_ext_metadata_read(WT_EXTENSION_API *wt_api,
+    WT_SESSION *wt_session,
+    const char *key,
+    const char **valuep);
+extern int __wt_ext_metadata_remove( WT_EXTENSION_API *wt_api,
+    WT_SESSION *wt_session,
+    const char *key);
+extern int __wt_ext_metadata_update(WT_EXTENSION_API *wt_api,
+    WT_SESSION *wt_session,
+    const char *key,
+    const char *value);
+extern int __wt_metadata_get_ckptlist( WT_SESSION *session,
+    const char *name,
+    WT_CKPT **ckptbasep);
+extern void __wt_metadata_free_ckptlist(WT_SESSION *session, WT_CKPT *ckptbase);
 extern int __wt_metadata_open(WT_SESSION_IMPL *session);
 extern int __wt_metadata_cursor( WT_SESSION_IMPL *session,
     const char *config,
diff --git a/src/include/wiredtiger_ext.h b/src/include/wiredtiger_ext.h
index aeb446fd45c..5209934aff2 100644
--- a/src/include/wiredtiger_ext.h
+++ b/src/include/wiredtiger_ext.h
@@ -194,6 +194,53 @@ struct __wt_extension_api {
 	    WT_CONFIG_SCAN *scan, WT_CONFIG_ITEM *key, WT_CONFIG_ITEM *value);
 
 	/*!
+	 * Insert a row into the metadata.
+	 *
+	 * @param wt_api the extension handle
+	 * @param session the session handle (or NULL if none available)
+	 * @param key row key
+	 * @param value row value
+	 * @errors
+	 */
+	int (*metadata_insert)(WT_EXTENSION_API *wt_api,
+	    WT_SESSION *session, const char *key, const char *value);
+
+	/*!
+	 * Read a row from the metadata.
+	 *
+	 * @param wt_api the extension handle
+	 * @param session the session handle (or NULL if none available)
+	 * @param key row key
+	 * @param [out] valuep the row value
+	 * @errors
+	 */
+	int (*metadata_read)(WT_EXTENSION_API *wt_api,
+	    WT_SESSION *session, const char *key, const char **valuep);
+
+	/*!
+	 * Remove a row in the metadata.
+	 *
+	 * @param wt_api the extension handle
+	 * @param session the session handle (or NULL if none available)
+	 * @param key row key
+	 * @errors
+	 */
+	int (*metadata_remove)(
+	    WT_EXTENSION_API *wt_api, WT_SESSION *session, const char *key);
+
+	/*!
+	 * Update a row in the metadata.
+	 *
+	 * @param wt_api the extension handle
+	 * @param session the session handle (or NULL if none available)
+	 * @param key row key
+	 * @param value row value
+	 * @errors
+	 */
+	int (*metadata_update)(WT_EXTENSION_API *wt_api,
+	    WT_SESSION *session, const char *key, const char *value);
+
+	/*!
 	 * Pack a structure into a buffer.
 	 * See ::wiredtiger_struct_pack for details.
 	 *
diff --git a/src/meta/meta_api.c b/src/meta/meta_api.c
deleted file mode 100644
index 8aa3caa6fa2..00000000000
--- a/src/meta/meta_api.c
+++ /dev/null
@@ -1,40 +0,0 @@
-/*-
- * Copyright (c) 2008-2013 WiredTiger, Inc.
- *	All rights reserved.
- *
- * See the file LICENSE for redistribution information.
- */
-
-#include "wt_internal.h"
-
-/*
- * wt_metadata_get --
- *	Public entry point to __wt_metadata_read (for wt dump and list).
- */
-int
-__wt_metadata_get(WT_SESSION *session, const char *uri, const char **valuep)
-{
-	return (__wt_metadata_read((WT_SESSION_IMPL *)session, uri, valuep));
-}
-
-/*
- * __wt_metadata_get_ckptlist --
- *	Public entry point to __wt_meta_ckptlist_get (for wt list).
- */
-int
-__wt_metadata_get_ckptlist(
-    WT_SESSION *session, const char *name, WT_CKPT **ckptbasep)
-{
-	return (__wt_meta_ckptlist_get(
-	    (WT_SESSION_IMPL *)session, name, ckptbasep));
-}
-
-/*
- * __wt_metadata_free_ckptlist --
- *	Public entry point to __wt_meta_ckptlist_free (for wt list).
- */
-void
-__wt_metadata_free_ckptlist(WT_SESSION *session, WT_CKPT *ckptbase)
-{
-	__wt_meta_ckptlist_free((WT_SESSION_IMPL *)session, ckptbase);
-}
diff --git a/src/meta/meta_ext.c b/src/meta/meta_ext.c
new file mode 100644
index 00000000000..8af85c68381
--- /dev/null
+++ b/src/meta/meta_ext.c
@@ -0,0 +1,103 @@
+/*-
+ * Copyright (c) 2008-2013 WiredTiger, Inc.
+ *	All rights reserved.
+ *
+ * See the file LICENSE for redistribution information.
+ */
+
+#include "wt_internal.h"
+
+/*
+ * __wt_ext_metadata_insert --
+ *	Insert a row into the metadata (external API version).
+ */
+int
+__wt_ext_metadata_insert(WT_EXTENSION_API *wt_api,
+    WT_SESSION *wt_session, const char *key, const char *value)
+{
+	WT_CONNECTION_IMPL *conn;
+	WT_SESSION_IMPL *session;
+
+	conn = (WT_CONNECTION_IMPL *)wt_api->conn;
+	if ((session = (WT_SESSION_IMPL *)wt_session) == NULL)
+		session = conn->default_session;
+
+	return (__wt_metadata_insert(session, key, value));
+}
+
+/*
+ * __wt_ext_metadata_read --
+ *	Reads and copies a row from the metadata (external API version).
+ *	The caller is responsible for freeing the allocated memory.
+ */
+int
+__wt_ext_metadata_read(WT_EXTENSION_API *wt_api,
+    WT_SESSION *wt_session, const char *key, const char **valuep)
+{
+	WT_CONNECTION_IMPL *conn;
+	WT_SESSION_IMPL *session;
+
+	conn = (WT_CONNECTION_IMPL *)wt_api->conn;
+	if ((session = (WT_SESSION_IMPL *)wt_session) == NULL)
+		session = conn->default_session;
+
+	return (__wt_metadata_read(session, key, valuep));
+}
+
+/*
+ * __wt_ext_metadata_remove --
+ *	Removes a row from the metadata (external API version).
+ */
+int
+__wt_ext_metadata_remove(
+    WT_EXTENSION_API *wt_api, WT_SESSION *wt_session, const char *key)
+{
+	WT_CONNECTION_IMPL *conn;
+	WT_SESSION_IMPL *session;
+
+	conn = (WT_CONNECTION_IMPL *)wt_api->conn;
+	if ((session = (WT_SESSION_IMPL *)wt_session) == NULL)
+		session = conn->default_session;
+
+	return (__wt_metadata_remove(session, key));
+}
+
+/*
+ * __wt_ext_metadata_update --
+ *	Update a row in the metadata (external API version).
+ */
+int
+__wt_ext_metadata_update(WT_EXTENSION_API *wt_api,
+    WT_SESSION *wt_session, const char *key, const char *value)
+{
+	WT_CONNECTION_IMPL *conn;
+	WT_SESSION_IMPL *session;
+
+	conn = (WT_CONNECTION_IMPL *)wt_api->conn;
+	if ((session = (WT_SESSION_IMPL *)wt_session) == NULL)
+		session = conn->default_session;
+
+	return (__wt_metadata_update(session, key, value));
+}
+
+/*
+ * __wt_metadata_get_ckptlist --
+ *	Public entry point to __wt_meta_ckptlist_get (for wt list).
+ */
+int
+__wt_metadata_get_ckptlist(
+    WT_SESSION *session, const char *name, WT_CKPT **ckptbasep)
+{
+	return (__wt_meta_ckptlist_get(
+	    (WT_SESSION_IMPL *)session, name, ckptbasep));
+}
+
+/*
+ * __wt_metadata_free_ckptlist --
+ *	Public entry point to __wt_meta_ckptlist_free (for wt list).
+ */
+void
+__wt_metadata_free_ckptlist(WT_SESSION *session, WT_CKPT *ckptbase)
+{
+	__wt_meta_ckptlist_free((WT_SESSION_IMPL *)session, ckptbase);
+}
diff --git a/src/schema/schema_create.c b/src/schema/schema_create.c
index ea2801b549c..b1ca4b0d423 100644
--- a/src/schema/schema_create.c
+++ b/src/schema/schema_create.c
@@ -492,22 +492,11 @@ err:		if (table != NULL) {
 
 static int
 __create_data_source(WT_SESSION_IMPL *session,
-    const char *uri, int exclusive, const char *config, WT_DATA_SOURCE *dsrc)
+    const char *uri, const char *config, WT_DATA_SOURCE *dsrc)
 {
 	WT_CONFIG_ITEM cval;
-	WT_DECL_RET;
-	const char *cfg[4], *fileconf;
-
-	fileconf = NULL;
-
-	/* Check if the data-source already exists. */
-	if ((ret =
-	    __wt_metadata_read(session, uri, &fileconf)) != WT_NOTFOUND) {
-		__wt_free(session, fileconf);
-		if (exclusive)
-			WT_TRET(EEXIST);
-		return (ret);
-	}
+	const char *cfg[] = {
+	    WT_CONFIG_BASE(session, session_create), config, NULL };
 
 	/*
 	 * User-specified collators aren't supported for data-source objects.
@@ -518,34 +507,7 @@ __create_data_source(WT_SESSION_IMPL *session,
 		    "WT_DATA_SOURCE objects do not support WT_COLLATOR "
 		    "ordering");
 
-	/*
-	 * Set a default key/value format, and insert the configuration into
-	 * the metadata.
-	 *
-	 * XXX
-	 * Use the session_create information, even though it includes a ton of
-	 * things we don't care about (like checksum configuration).  We should
-	 * be stripping that information out.
-	 */
-	cfg[0] = WT_CONFIG_BASE(session, session_create);
-	cfg[1] = "key_format=u,value_format=u";
-	cfg[2] = config;
-	cfg[3] = NULL;
-	WT_RET(__wt_config_collapse(session, cfg, &fileconf));
-	if ((ret = __wt_metadata_insert(session, uri, fileconf)) == 0) {
-		cfg[0] = fileconf;
-		cfg[1] = NULL;
-		WT_ERR(dsrc->create(
-		    dsrc, &session->iface, uri, (WT_CONFIG_ARG *)cfg));
-	} else if (ret == WT_DUPLICATE_KEY)
-		ret = EEXIST;
-
-	if (0) {
-err:		WT_TRET(__wt_metadata_remove(session, uri));
-	}
-
-	__wt_free(session, fileconf);
-	return (ret);
+	return (dsrc->create(dsrc, &session->iface, uri, (WT_CONFIG_ARG *)cfg));
 }
 
 int
@@ -580,7 +542,7 @@ __wt_schema_create(
 	else if ((dsrc = __wt_schema_get_source(session, uri)) != NULL)
 		ret = dsrc->create == NULL ?
 		    __wt_object_unsupported(session, uri) :
-		    __create_data_source(session, uri, exclusive, config, dsrc);
+		    __create_data_source(session, uri, config, dsrc);
 	else
 		ret = __wt_bad_object_type(session, uri);
 
diff --git a/src/schema/schema_drop.c b/src/schema/schema_drop.c
index 100fe0067df..7b18a8e10fa 100644
--- a/src/schema/schema_drop.c
+++ b/src/schema/schema_drop.c
@@ -183,15 +183,12 @@ __wt_schema_drop(WT_SESSION_IMPL *session, const char *uri, const char *cfg[])
 		ret = __wt_lsm_tree_drop(session, uri, cfg);
 	else if (WT_PREFIX_MATCH(uri, "table:"))
 		ret = __drop_table(session, uri, force, cfg);
-	else if ((dsrc = __wt_schema_get_source(session, uri)) != NULL) {
-		if (dsrc->drop == NULL)
-			ret = __wt_object_unsupported(session, uri);
-		if (ret == 0)
-			WT_TRET_NOTFOUND_OK(__wt_metadata_remove(session, uri));
-		if (ret == 0)
-			ret = dsrc->drop(
-			    dsrc, &session->iface, uri, (WT_CONFIG_ARG *)cfg);
-	} else
+	else if ((dsrc = __wt_schema_get_source(session, uri)) != NULL)
+		ret = dsrc->drop == NULL ?
+		    __wt_object_unsupported(session, uri) :
+		    dsrc->drop(
+		    dsrc, &session->iface, uri, (WT_CONFIG_ARG *)cfg);
+	else
 		ret = __wt_bad_object_type(session, uri);
 
 	/*
diff --git a/src/schema/schema_rename.c b/src/schema/schema_rename.c
index e1b1b50e5f7..ca349abc519 100644
--- a/src/schema/schema_rename.c
+++ b/src/schema/schema_rename.c
@@ -258,15 +258,12 @@ __wt_schema_rename(WT_SESSION_IMPL *session,
 		ret = __wt_lsm_tree_rename(session, uri, newuri, cfg);
 	else if (WT_PREFIX_MATCH(uri, "table:"))
 		ret = __rename_table(session, uri, newuri, cfg);
-	else if ((dsrc = __wt_schema_get_source(session, uri)) != NULL) {
-		if (dsrc->rename == NULL)
-			ret = __wt_object_unsupported(session, uri);
-		if (ret == 0)
-			ret = __metadata_rename(session, uri, newuri);
-		if (ret == 0)
-			ret = dsrc->rename(dsrc,
-			    &session->iface, uri, newuri, (WT_CONFIG_ARG *)cfg);
-	} else
+	else if ((dsrc = __wt_schema_get_source(session, uri)) != NULL)
+		ret = dsrc->rename == NULL ?
+		    __wt_object_unsupported(session, uri) :
+		    dsrc->rename(dsrc,
+		    &session->iface, uri, newuri, (WT_CONFIG_ARG *)cfg);
+	else
 		ret = __wt_bad_object_type(session, uri);
 
 	/* Bump the schema generation so that stale data is ignored. */
diff --git a/src/utilities/util_dump.c b/src/utilities/util_dump.c
index 887859160fb..bfca5fafc6d 100644
--- a/src/utilities/util_dump.c
+++ b/src/utilities/util_dump.c
@@ -140,6 +140,7 @@ dump_config(WT_SESSION *session, const char *uri, int hex)
 {
 	WT_CURSOR *cursor;
 	WT_DECL_RET;
+	WT_EXTENSION_API *wtext;
 	int tret;
 	const char *value;
 
@@ -185,7 +186,10 @@ dump_config(WT_SESSION *session, const char *uri, int hex)
 		 * This where we find out a file object doesn't exist, use a
 		 * simple error message.
 		 */
-		if ((ret = __wt_metadata_get(session, uri, &value)) == 0) {
+		wtext = session->
+		    connection->get_extension_api(session->connection);
+		if ((ret =
+		    wtext->metadata_read(wtext, session, uri, &value)) == 0) {
 			if (dump_prefix(hex) != 0 ||
 			    print_config(session, uri, value, NULL) != 0 ||
 			    dump_suffix() != 0)
diff --git a/src/utilities/util_list.c b/src/utilities/util_list.c
index 55a5114899a..a392e7b2a3f 100644
--- a/src/utilities/util_list.c
+++ b/src/utilities/util_list.c
@@ -64,6 +64,7 @@ list_print(WT_SESSION *session, const char *name, int cflag, int vflag)
 {
 	WT_CURSOR *cursor;
 	WT_DECL_RET;
+	WT_EXTENSION_API *wtext;
 	int found;
 	const char *key, *value, *uri;
 
@@ -79,8 +80,10 @@ list_print(WT_SESSION *session, const char *name, int cflag, int vflag)
 		if (cflag && (ret = list_print_checkpoint(session, uri)) != 0)
 			return (ret);
 		if (vflag) {
-			if ((ret =
-			    __wt_metadata_get(session, uri, &value)) != 0)
+			wtext = session->
+			    connection->get_extension_api(session->connection);
+			if ((ret = wtext->
+			    metadata_read(wtext, session, uri, &value)) != 0)
 				return (
 				    util_err(ret, "metadata read: %s", uri));
 			printf("%s\n", value);