[PATCH 09/12] vinyl: introduce statement environment

[Vladimir Davydov <vdavydov.dev@gmail.com>]

Store tuple_format_vtab, max_tuple_size, and key_format there.
This will allow us to determine a statement type (key or tuple)
by checking its format against key_format.
---
 src/box/vinyl.c                 | 29 ++++++++++++++++-------------
 src/box/vy_lsm.c                | 28 +++++++++++-----------------
 src/box/vy_lsm.h                |  9 +++++----
 src/box/vy_stmt.c               | 35 +++++++++++++++++++++++++++++------
 src/box/vy_stmt.h               | 34 +++++++++++++++++++++++++++-------
 test/unit/vy_iterators_helper.c | 21 +++++++--------------
 test/unit/vy_iterators_helper.h |  3 +--
 test/unit/vy_mem.c              |  6 ++----
 test/unit/vy_point_lookup.c     | 11 +++++------
 9 files changed, 103 insertions(+), 73 deletions(-)

diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index f2f37734..c469077e 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -105,6 +105,8 @@ struct vy_env {
 	struct mempool iterator_pool;
 	/** Memory quota */
 	struct vy_quota     quota;
+	/** Statement environment. */
+	struct vy_stmt_env stmt_env;
 	/** Common LSM tree environment. */
 	struct vy_lsm_env lsm_env;
 	/** Environment for cache subsystem */
@@ -602,6 +604,7 @@ static struct space *
 vinyl_engine_create_space(struct engine *engine, struct space_def *def,
 			  struct rlist *key_list)
 {
+	struct vy_env *env = vy_env(engine);
 	struct space *space = malloc(sizeof(*space));
 	if (space == NULL) {
 		diag_set(OutOfMemory, sizeof(*space),
@@ -624,11 +627,10 @@ vinyl_engine_create_space(struct engine *engine, struct space_def *def,
 	rlist_foreach_entry(index_def, key_list, link)
 		keys[key_count++] = index_def->key_def;
 
-	struct tuple_format *format =
-		tuple_format_new(&vy_tuple_format_vtab, NULL, keys, key_count,
-				 def->fields, def->field_count,
-				 def->exact_field_count, def->dict, false,
-				 false);
+	struct tuple_format *format;
+	format = vy_stmt_format_new(&env->stmt_env, keys, key_count,
+				    def->fields, def->field_count,
+				    def->exact_field_count, def->dict);
 	if (format == NULL) {
 		free(space);
 		return NULL;
@@ -718,9 +720,9 @@ vinyl_space_create_index(struct space *space, struct index_def *index_def)
 		pk = vy_lsm(space_index(space, 0));
 		assert(pk != NULL);
 	}
-	struct vy_lsm *lsm = vy_lsm_new(&env->lsm_env, &env->cache_env,
-					&env->mem_env, index_def,
-					space->format, pk,
+	struct vy_lsm *lsm = vy_lsm_new(&env->lsm_env, &env->stmt_env,
+					&env->cache_env, &env->mem_env,
+					index_def, space->format, pk,
 					space_group_id(space));
 	if (lsm == NULL) {
 		free(index);
@@ -2515,6 +2517,7 @@ vy_env_new(const char *path, size_t memory,
 	if (e->squash_queue == NULL)
 		goto error_squash_queue;
 
+	vy_stmt_env_create(&e->stmt_env);
 	vy_mem_env_create(&e->mem_env, memory);
 	vy_scheduler_create(&e->scheduler, write_threads,
 			    vy_env_dump_complete_cb,
@@ -2522,6 +2525,7 @@ vy_env_new(const char *path, size_t memory,
 
 	if (vy_lsm_env_create(&e->lsm_env, e->path,
 			      &e->scheduler.generation,
+			      e->stmt_env.key_format,
 			      vy_squash_schedule, e) != 0)
 		goto error_lsm_env;
 
@@ -2563,6 +2567,7 @@ vy_env_delete(struct vy_env *e)
 	vy_lsm_env_destroy(&e->lsm_env);
 	vy_mem_env_destroy(&e->mem_env);
 	vy_cache_env_destroy(&e->cache_env);
+	vy_stmt_env_destroy(&e->stmt_env);
 	vy_quota_destroy(&e->quota);
 	if (e->recovery != NULL)
 		vy_recovery_delete(e->recovery);
@@ -2635,8 +2640,7 @@ vinyl_engine_set_memory(struct vinyl_engine *vinyl, size_t size)
 void
 vinyl_engine_set_max_tuple_size(struct vinyl_engine *vinyl, size_t max_size)
 {
-	(void)vinyl;
-	vy_max_tuple_size = max_size;
+	vinyl->env->stmt_env.max_tuple_size = max_size;
 }
 
 void
@@ -3051,9 +3055,8 @@ vy_send_lsm(struct vy_join_ctx *ctx, struct vy_lsm_recovery_info *lsm_info)
 				   lsm_info->key_part_count);
 	if (ctx->key_def == NULL)
 		goto out;
-	ctx->format = tuple_format_new(&vy_tuple_format_vtab, NULL,
-				       &ctx->key_def, 1, NULL, 0, 0, NULL,
-				       false, false);
+	ctx->format = vy_stmt_format_new(&ctx->env->stmt_env, &ctx->key_def, 1,
+					 NULL, 0, 0, NULL);
 	if (ctx->format == NULL)
 		goto out_free_key_def;
 	tuple_format_ref(ctx->format);
diff --git a/src/box/vy_lsm.c b/src/box/vy_lsm.c
index 73d369fb..eca8a334 100644
--- a/src/box/vy_lsm.c
+++ b/src/box/vy_lsm.c
@@ -70,23 +70,17 @@ static const int64_t VY_MAX_RANGE_SIZE = 2LL * 1024 * 1024 * 1024;
 
 int
 vy_lsm_env_create(struct vy_lsm_env *env, const char *path,
-		  int64_t *p_generation,
+		  int64_t *p_generation, struct tuple_format *key_format,
 		  vy_upsert_thresh_cb upsert_thresh_cb,
 		  void *upsert_thresh_arg)
 {
-	env->key_format = tuple_format_new(&vy_tuple_format_vtab, NULL,
-					   NULL, 0, NULL, 0, 0, NULL, false,
-					   false);
-	if (env->key_format == NULL)
+	env->empty_key = vy_stmt_new_select(key_format, NULL, 0);
+	if (env->empty_key == NULL)
 		return -1;
-	tuple_format_ref(env->key_format);
-	env->empty_key = vy_stmt_new_select(env->key_format, NULL, 0);
-	if (env->empty_key == NULL) {
-		tuple_format_unref(env->key_format);
-		return -1;
-	}
 	env->path = path;
 	env->p_generation = p_generation;
+	env->key_format = key_format;
+	tuple_format_ref(key_format);
 	env->upsert_thresh_cb = upsert_thresh_cb;
 	env->upsert_thresh_arg = upsert_thresh_arg;
 	env->too_long_threshold = TIMEOUT_INFINITY;
@@ -124,9 +118,10 @@ vy_lsm_mem_tree_size(struct vy_lsm *lsm)
 }
 
 struct vy_lsm *
-vy_lsm_new(struct vy_lsm_env *lsm_env, struct vy_cache_env *cache_env,
-	   struct vy_mem_env *mem_env, struct index_def *index_def,
-	   struct tuple_format *format, struct vy_lsm *pk, uint32_t group_id)
+vy_lsm_new(struct vy_lsm_env *lsm_env, struct vy_stmt_env *stmt_env,
+	   struct vy_cache_env *cache_env, struct vy_mem_env *mem_env,
+	   struct index_def *index_def, struct tuple_format *format,
+	   struct vy_lsm *pk, uint32_t group_id)
 {
 	static int64_t run_buckets[] = {
 		0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 15, 20, 25, 50, 100,
@@ -161,9 +156,8 @@ vy_lsm_new(struct vy_lsm_env *lsm_env, struct vy_cache_env *cache_env,
 		 */
 		lsm->disk_format = format;
 	} else {
-		lsm->disk_format = tuple_format_new(&vy_tuple_format_vtab, NULL,
-						    &cmp_def, 1, NULL, 0, 0,
-						    NULL, false, false);
+		lsm->disk_format = vy_stmt_format_new(stmt_env, &cmp_def, 1,
+						      NULL, 0, 0, NULL);
 		if (lsm->disk_format == NULL)
 			goto fail_format;
 	}
diff --git a/src/box/vy_lsm.h b/src/box/vy_lsm.h
index 1bb447fb..b94e7a43 100644
--- a/src/box/vy_lsm.h
+++ b/src/box/vy_lsm.h
@@ -130,7 +130,7 @@ struct vy_lsm_env {
 /** Create a common LSM tree environment. */
 int
 vy_lsm_env_create(struct vy_lsm_env *env, const char *path,
-		  int64_t *p_generation,
+		  int64_t *p_generation, struct tuple_format *key_format,
 		  vy_upsert_thresh_cb upsert_thresh_cb,
 		  void *upsert_thresh_arg);
 
@@ -333,9 +333,10 @@ vy_lsm_mem_tree_size(struct vy_lsm *lsm);
 
 /** Allocate a new LSM tree object. */
 struct vy_lsm *
-vy_lsm_new(struct vy_lsm_env *lsm_env, struct vy_cache_env *cache_env,
-	   struct vy_mem_env *mem_env, struct index_def *index_def,
-	   struct tuple_format *format, struct vy_lsm *pk, uint32_t group_id);
+vy_lsm_new(struct vy_lsm_env *lsm_env, struct vy_stmt_env *stmt_env,
+	   struct vy_cache_env *cache_env, struct vy_mem_env *mem_env,
+	   struct index_def *index_def, struct tuple_format *format,
+	   struct vy_lsm *pk, uint32_t group_id);
 
 /** Free an LSM tree object. */
 void
diff --git a/src/box/vy_stmt.c b/src/box/vy_stmt.c
index 9d9f2c75..57abad23 100644
--- a/src/box/vy_stmt.c
+++ b/src/box/vy_stmt.c
@@ -112,12 +112,34 @@ vy_tuple_delete(struct tuple_format *format, struct tuple *tuple)
 	free(tuple);
 }
 
-struct tuple_format_vtab vy_tuple_format_vtab = {
-	vy_tuple_delete,
-	vy_tuple_new,
-};
+void
+vy_stmt_env_create(struct vy_stmt_env *env)
+{
+	env->tuple_format_vtab.tuple_new = vy_tuple_new;
+	env->tuple_format_vtab.tuple_delete = vy_tuple_delete;
+	env->max_tuple_size = 1024 * 1024;
+	env->key_format = vy_stmt_format_new(env, NULL, 0, NULL, 0, 0, NULL);
+	if (env->key_format == NULL)
+		panic("failed to create vinyl key format");
+	tuple_format_ref(env->key_format);
+}
 
-size_t vy_max_tuple_size = 1024 * 1024;
+void
+vy_stmt_env_destroy(struct vy_stmt_env *env)
+{
+	tuple_format_unref(env->key_format);
+}
+
+struct tuple_format *
+vy_stmt_format_new(struct vy_stmt_env *env, struct key_def *const *keys,
+		   uint16_t key_count, const struct field_def *fields,
+		   uint32_t field_count, uint32_t exact_field_count,
+		   struct tuple_dictionary *dict)
+{
+	return tuple_format_new(&env->tuple_format_vtab, env, keys, key_count,
+				fields, field_count, exact_field_count, dict,
+				false, false);
+}
 
 /**
  * Allocate a vinyl statement object on base of the struct tuple
@@ -132,9 +154,10 @@ size_t vy_max_tuple_size = 1024 * 1024;
 static struct tuple *
 vy_stmt_alloc(struct tuple_format *format, uint32_t bsize)
 {
+	struct vy_stmt_env *env = format->engine;
 	uint32_t total_size = sizeof(struct vy_stmt) + format->field_map_size +
 		bsize;
-	if (unlikely(total_size > vy_max_tuple_size)) {
+	if (unlikely(total_size > env->max_tuple_size)) {
 		diag_set(ClientError, ER_VINYL_MAX_TUPLE_SIZE,
 			 (unsigned) total_size);
 		error_log(diag_last_error(diag_get()));
diff --git a/src/box/vy_stmt.h b/src/box/vy_stmt.h
index ab352906..3881a2f4 100644
--- a/src/box/vy_stmt.h
+++ b/src/box/vy_stmt.h
@@ -49,6 +49,7 @@ extern "C" {
 struct xrow_header;
 struct region;
 struct tuple_format;
+struct tuple_dictionary;
 struct iovec;
 
 #define MAX_LSN (INT64_MAX / 2)
@@ -61,14 +62,33 @@ static_assert(VY_UPSERT_THRESHOLD <= UINT8_MAX, "n_upserts max value");
 static_assert(VY_UPSERT_INF == VY_UPSERT_THRESHOLD + 1,
 	      "inf must be threshold + 1");
 
-/** Vinyl statement vtable. */
-extern struct tuple_format_vtab vy_tuple_format_vtab;
+/** Vinyl statement environment. */
+struct vy_stmt_env {
+	/** Vinyl statement vtable. */
+	struct tuple_format_vtab tuple_format_vtab;
+	/**
+	 * Max tuple size
+	 * @see box.cfg.vinyl_max_tuple_size
+	 */
+	size_t max_tuple_size;
+	/** Tuple format for keys. */
+	struct tuple_format *key_format;
+};
 
-/**
- * Max tuple size
- * @see box.cfg.vinyl_max_tuple_size
- */
-extern size_t vy_max_tuple_size;
+/** Initialize a vinyl statement environment. */
+void
+vy_stmt_env_create(struct vy_stmt_env *env);
+
+/** Destroy a vinyl statement environment. */
+void
+vy_stmt_env_destroy(struct vy_stmt_env *env);
+
+/** Create a vinyl statement format. */
+struct tuple_format *
+vy_stmt_format_new(struct vy_stmt_env *env, struct key_def *const *keys,
+		   uint16_t key_count, const struct field_def *fields,
+		   uint32_t field_count, uint32_t exact_field_count,
+		   struct tuple_dictionary *dict);
 
 /** Statement flags. */
 enum {
diff --git a/test/unit/vy_iterators_helper.c b/test/unit/vy_iterators_helper.c
index 55d8504b..bcb60001 100644
--- a/test/unit/vy_iterators_helper.c
+++ b/test/unit/vy_iterators_helper.c
@@ -6,7 +6,7 @@
 
 struct tt_uuid INSTANCE_UUID;
 
-struct tuple_format *vy_key_format = NULL;
+struct vy_stmt_env stmt_env;
 struct vy_mem_env mem_env;
 struct vy_cache_env cache_env;
 
@@ -19,12 +19,9 @@ vy_iterator_C_test_init(size_t cache_size)
 	memory_init();
 	fiber_init(fiber_c_invoke);
 	tuple_init(NULL);
+	vy_stmt_env_create(&stmt_env);
 	vy_cache_env_create(&cache_env, cord_slab_cache());
 	vy_cache_env_set_quota(&cache_env, cache_size);
-	vy_key_format = tuple_format_new(&vy_tuple_format_vtab, NULL, NULL, 0,
-					 NULL, 0, 0, NULL, false, false);
-	tuple_format_ref(vy_key_format);
-
 	size_t mem_size = 64 * 1024 * 1024;
 	vy_mem_env_create(&mem_env, mem_size);
 }
@@ -33,8 +30,8 @@ void
 vy_iterator_C_test_finish()
 {
 	vy_mem_env_destroy(&mem_env);
-	tuple_format_unref(vy_key_format);
 	vy_cache_env_destroy(&cache_env);
+	vy_stmt_env_destroy(&stmt_env);
 	tuple_free();
 	fiber_free();
 	memory_free();
@@ -122,7 +119,7 @@ vy_new_simple_stmt(struct tuple_format *format,
 	case IPROTO_SELECT: {
 		const char *key = buf;
 		uint part_count = mp_decode_array(&key);
-		ret = vy_stmt_new_select(vy_key_format, key, part_count);
+		ret = vy_stmt_new_select(stmt_env.key_format, key, part_count);
 		fail_if(ret == NULL);
 		break;
 	}
@@ -199,11 +196,8 @@ struct vy_mem *
 create_test_mem(struct key_def *def)
 {
 	/* Create format */
-	struct key_def * const defs[] = { def };
-	struct tuple_format *format =
-		tuple_format_new(&vy_tuple_format_vtab, NULL, defs,
-				 def->part_count, NULL, 0, 0, NULL, false,
-				 false);
+	struct tuple_format *format;
+	format = vy_stmt_format_new(&stmt_env, &def, 1, NULL, 0, 0, NULL);
 	fail_if(format == NULL);
 
 	/* Create mem */
@@ -220,8 +214,7 @@ create_test_cache(uint32_t *fields, uint32_t *types,
 	*def = box_key_def_new(fields, types, key_cnt);
 	assert(*def != NULL);
 	vy_cache_create(cache, &cache_env, *def, true);
-	*format = tuple_format_new(&vy_tuple_format_vtab, NULL, def, 1, NULL, 0,
-				   0, NULL, false, false);
+	*format = vy_stmt_format_new(&stmt_env, def, 1, NULL, 0, 0, NULL);
 	tuple_format_ref(*format);
 }
 
diff --git a/test/unit/vy_iterators_helper.h b/test/unit/vy_iterators_helper.h
index 9690f684..49b4f4fd 100644
--- a/test/unit/vy_iterators_helper.h
+++ b/test/unit/vy_iterators_helper.h
@@ -51,8 +51,7 @@
 #define STMT_TEMPLATE_DEFERRED_DELETE(lsn, type, ...) \
 STMT_TEMPLATE_FLAGS(lsn, type, VY_STMT_DEFERRED_DELETE, __VA_ARGS__)
 
-extern struct tuple_format_vtab vy_tuple_format_vtab;
-extern struct tuple_format *vy_key_format;
+extern struct vy_stmt_env stmt_env;
 extern struct vy_mem_env mem_env;
 extern struct vy_cache_env cache_env;
 
diff --git a/test/unit/vy_mem.c b/test/unit/vy_mem.c
index b8272eac..0226a62a 100644
--- a/test/unit/vy_mem.c
+++ b/test/unit/vy_mem.c
@@ -76,10 +76,8 @@ test_iterator_restore_after_insertion()
 	assert(key_def != NULL);
 
 	/* Create format */
-	struct tuple_format *format = tuple_format_new(&vy_tuple_format_vtab,
-						       NULL, &key_def, 1, NULL,
-						       0, 0, NULL, false,
-						       false);
+	struct tuple_format *format = vy_stmt_format_new(&stmt_env, &key_def, 1,
+							 NULL, 0, 0, NULL);
 	assert(format != NULL);
 	tuple_format_ref(format);
 
diff --git a/test/unit/vy_point_lookup.c b/test/unit/vy_point_lookup.c
index 8ed16f6c..5f20d09b 100644
--- a/test/unit/vy_point_lookup.c
+++ b/test/unit/vy_point_lookup.c
@@ -66,7 +66,8 @@ test_basic()
 
 	int rc;
 	struct vy_lsm_env lsm_env;
-	rc = vy_lsm_env_create(&lsm_env, ".", &generation, NULL, NULL);
+	rc = vy_lsm_env_create(&lsm_env, ".", &generation,
+			       stmt_env.key_format, NULL, NULL);
 	is(rc, 0, "vy_lsm_env_create");
 
 	struct vy_run_env run_env;
@@ -83,10 +84,8 @@ test_basic()
 	isnt(key_def, NULL, "key_def is not NULL");
 
 	vy_cache_create(&cache, &cache_env, key_def, true);
-	struct tuple_format *format = tuple_format_new(&vy_tuple_format_vtab,
-						       NULL, &key_def, 1, NULL,
-						       0, 0, NULL, false,
-						       false);
+	struct tuple_format *format = vy_stmt_format_new(&stmt_env, &key_def, 1,
+							 NULL, 0, 0, NULL);
 	isnt(format, NULL, "tuple_format_new is not NULL");
 	tuple_format_ref(format);
 
@@ -95,7 +94,7 @@ test_basic()
 		index_def_new(512, 0, "primary", sizeof("primary") - 1, TREE,
 			      &index_opts, key_def, NULL);
 
-	struct vy_lsm *pk = vy_lsm_new(&lsm_env, &cache_env, &mem_env,
+	struct vy_lsm *pk = vy_lsm_new(&lsm_env, &stmt_env, &cache_env, &mem_env,
 				       index_def, format, NULL, 0);
 	isnt(pk, NULL, "lsm is not NULL")
 
-- 
2.11.0