On Sat, Apr 07, 2018 at 04:38:09PM +0300, Vladimir Davydov wrote:
static int
vinyl_space_check_format(struct space *space, struct tuple_format *format)
{
- (void)format;
struct vy_env *env = vy_env(space->engine);
- if (space->index_count == 0)
+
+ /*
+ * If this is local recovery, the space was checked before
+ * restart so there's nothing we need to do.
+ */
+ if (env->status == VINYL_INITIAL_RECOVERY_LOCAL ||
+ env->status == VINYL_FINAL_RECOVERY_LOCAL)
return 0;
+
+ if (space->index_count == 0)
+ return 0; /* space is empty, nothing to do */
+
+ /*
+ * Iterate over all tuples stored in the given space and
+ * check each of them for conformity to the new format.
+ * Since read iterator may yield, we install an on_replace
+ * trigger to check tuples inserted after we started the
+ * iteration.
+ */
struct vy_lsm *pk = vy_lsm(space->index[0]);
- if (env->status != VINYL_ONLINE)
- return 0;
- if (pk->stat.disk.count.rows == 0 && pk->stat.memory.count.rows == 0)
- return 0;
- diag_set(ClientError, ER_UNSUPPORTED, "Vinyl",
- "changing format of a non-empty space");
- return -1;
+
+ struct tuple *key = vy_stmt_new_select(pk->env->key_format, NULL, 0);
+ if (key == NULL)
+ return -1;
+
+ struct trigger on_replace;
+ struct vy_check_format_ctx ctx;
+ ctx.format = format;
+ ctx.is_failed = false;
+ diag_create(&ctx.diag);
+ trigger_create(&on_replace, vy_check_format_on_replace, &ctx, NULL);
+ trigger_add(&space->on_replace, &on_replace);
+
+ struct vy_read_iterator itr;
+ vy_read_iterator_open(&itr, pk, NULL, ITER_ALL, key,
+ &env->xm->p_global_read_view);
+ int rc;
+ struct tuple *tuple;
+ while ((rc = vy_read_iterator_next(&itr, &tuple)) == 0) {
+ if (tuple == NULL)
+ break;
+ if (ctx.is_failed) {
+ diag_move(&ctx.diag, diag_get());
+ rc = -1;
+ break;
+ }
+ rc = tuple_validate(format, tuple);
+ if (rc != 0)
+ break;
+ }
+ vy_read_iterator_close(&itr);
+ diag_destroy(&ctx.diag);
+ trigger_clear(&on_replace);
+ tuple_unref(key);
+ return rc;
}