anchor.diff:      CompileState = compiling,
       insert_previous(BucketType, DDL, CompileState).
   
  -convert_ddl_v2_to_ddl_v1(#ddl_v2{} = DDL) ->
  -    #ddl_v1{
  -       table = DDL#ddl_v2.table,
  -       fields = DDL#ddl_v2.fields,
  -       partition_key = DDL#ddl_v2.partition_key,
  -       local_key = DDL#ddl_v2.local_key
  -      }.
   insert_previous(BucketType, #ddl_v2{} = DDL, CompileState) ->
  -    DDL1 = convert_ddl_v2_to_ddl_v1(DDL),
  -    insert_previous(BucketType, DDL1, CompileState);
  +    case riak_ql_ddl:convert(v1, DDL) of
  +        Errors = [{error, _Reason}|_T] -> Errors;
  +        [DDL1] ->
  +            insert_previous(BucketType, DDL1, CompileState)
  +    end;
   insert_previous(BucketType, #ddl_v1{} = DDL, CompileState) ->
       %% the version is always 1 for the v2 table
       DDLVersion = 1,
  
positive.code:     end.
  
  %%
  -spec get_ddl(BucketType::binary(), Version::riak_ql_ddl:ddl_version()) ->
          {ok, term()} | notfound.
  get_ddl(BucketType, Version) when is_binary(BucketType), is_atom(Version) ->
  
negative.code:          get_table_status_pairs/0,
           get_table_status/1,
           get_compiled_ddl_versions/1,
           assert_compiled_ddl_versions_current/1,
           get_ddl/2,
           insert/2,
           new/1,
  