anchor.diff:  
   %% Returns a one arity fun which is stateless for example pulling a field from a
   %% row.
  +% -spec compile_select_col_stateless(#ddl_v1{}, selection()) ->
  +%         {ColTypes::[primitive_field_type()], IsValid::true|any(), function()}.
   compile_select_col_stateless(DDL, {identifier, [<<"*">>]}) ->
       ColTypes = [X#riak_field_v1.type || X <- DDL#ddl_v1.fields],
       {ColTypes, true, fun(Row) -> Row end};
  
positive.code: 
  %%
  compile_select_clause(DDL, #riak_sql_v1{'SELECT' = #riak_sel_clause_v1{ clause = Sel } } = Q) ->
      %% compile each select column and put all the calc types into a set, if
      %% any of the results are aggregate then aggregate is the calc type for the
      %% whole query
      CompileColFn = fun(ColX, #riak_sel_clause_v1{calc_type        = CalcTy,
                                                   initial_state    = InitX,
                                                   col_return_types = ColRetX,
                                                   clause           = RunFnX,
                                                   is_valid         = IsValid}) ->
  
                             S = compile_select_col(DDL, ColX),
  
                             CalcTy2 = sets:add_element(S#riak_sel_clause_v1.calc_type, CalcTy),
                             Init2   = [S#riak_sel_clause_v1.initial_state    | InitX],
                             ColRet2 = [S#riak_sel_clause_v1.col_return_types | ColRetX],
                             RunFn2  = [S#riak_sel_clause_v1.clause           | RunFnX],
                             IsV2    = merge_validation(S#riak_sel_clause_v1.is_valid, IsValid),
  
                             %% ColTypes are messy because <<"*">> represents many
                             %% so you need to flatten the list
                             #riak_sel_clause_v1{calc_type        = CalcTy2,
                                                 initial_state    = Init2,
                                                 col_return_types = lists:flatten(ColRet2),
                                                 clause           = RunFn2,
                                                 is_valid         = IsV2}
                     end,
      Acc = #riak_sel_clause_v1{calc_type = sets:new()},
      Q2 = lists:foldr(CompileColFn, Acc, Sel),
      ColNames = get_col_names(DDL, Q),
      case Q2#riak_sel_clause_v1.is_valid of
          true       -> CalcT = Q2#riak_sel_clause_v1.calc_type,
                        case sets:is_element(aggregate, CalcT) of
                            true  -> {ok, Q2#riak_sel_clause_v1{calc_type = aggregate,
                                                                col_names = ColNames}};
                            false -> {ok, Q2#riak_sel_clause_v1{calc_type = rows,
                                                                col_names = ColNames}}
                        end;
          {error, E} -> {error, E}
      end.
  
  get_col_names(DDL, Q) ->
      case riak_ql_to_string:col_names_from_select(Q) of
  
negative.code:                      {ok, InitialState::[any()], SubQueries::[#riak_sql_v1{}]} | {error, any()}.
  compile(#ddl_v1{}, #riak_sql_v1{is_executable = true}) ->
      {error, 'query is already compiled'};
  compile(#ddl_v1{}, #riak_sql_v1{'SELECT' = #riak_sel_clause_v1{ clause = [] } }) ->
      {error, 'full table scan not implmented'};
  compile(#ddl_v1{} = DDL, #riak_sql_v1{is_executable = false,
                                        type          = sql} = Q) ->
  