<commit_before>%% -------------------------------------------------------------------
%%
%% riak_kv_ts_util: supporting functions for timeseries code paths
%%
%% Copyright (c) 2016 Basho Technologies, Inc.  All Rights Reserved.
%%
%% This file is provided to you under the Apache License,
%% Version 2.0 (the "License"); you may not use this file
%% except in compliance with the License.  You may obtain
%% a copy of the License at
%%
%%   http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing,
%% software distributed under the License is distributed on an
%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
%% KIND, either express or implied.  See the License for the
%% specific language governing permissions and limitations
%% under the License.
%%
%% -------------------------------------------------------------------

%% @doc Various functions related to timeseries.

-module(riak_kv_ts_util).

-export([
         apply_timeseries_bucket_props/3,
         build_sql_record/3,
         check_table_feature_supported/2,
         is_table_supported/2,
         encode_typeval_key/1,
         explain_query/1, explain_query/2,
         get_column_types/2,
         get_table_ddl/1,
         lk/1,
         local_to_partition_key/2,
         maybe_parse_table_def/2,
         pk/1,
         queried_table/1,
         rm_rf/1,
         sql_record_to_tuple/1,
         sql_to_cover/6,
         table_to_bucket/1,
         validate_rows/2,
         varchar_to_timestamp/1
        ]).

-type ts_service_req() ::
    {ok, riak_kv_ts_svc:ts_requests(), {PermSpec::string(), Table::binary()}}.

-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
-compile(export_all).
-endif.

%% NOTE on table_to_bucket/1: Clients will work with table
%% names. Those names map to a bucket type/bucket name tuple in Riak,
%% with both the type name and the bucket name matching the table.
%%
%% Thus, as soon as code transitions from dealing with timeseries
%% concepts to Riak KV concepts, the table name must be converted to a
%% bucket tuple. This function is a convenient mechanism for doing so
%% and making that transition more obvious.

%%-include("riak_kv_wm_raw.hrl").
-include_lib("riak_ql/include/riak_ql_ddl.hrl").
-include("riak_kv_ts.hrl").

-spec sql_record_to_tuple(?SQL_SELECT{}) -> {binary(), [tuple(2) | tuple(3)], list(), [tuple(2)]}.
%% riak_ql_ddl:is_query_valid expects a tuple, not a SQL record
sql_record_to_tuple(?SQL_SELECT{'FROM'   = From,
                                'SELECT' = #riak_sel_clause_v1{clause=Select},
                                'WHERE'  = Where,
                                'ORDER BY' = OrderBy}) ->
    {From, Select, Where, OrderBy}.

-spec build_sql_record(riak_kv_qry:query_type(), proplists:proplist(), proplists:proplist()) ->
                              {ok, ?SQL_SELECT{}} | {error, any()}.
build_sql_record(Command, SQL, Options) ->
    try build_sql_record_int(Command, SQL, Options) of
        {ok, Record} ->
            {ok, Record};
        ER ->
            ER
    catch
        throw:Atom ->
            {error, Atom}
    end.


%% Convert the proplist obtained from the QL parser
build_sql_record_int(select, SQL, _Options = undefined) ->
    build_sql_record_int(select, SQL, []);
build_sql_record_int(select, SQL, Options) ->
    Cover  = proplists:get_value(cover, Options),
    if not (Cover == undefined orelse is_binary(Cover)) ->
            {error, bad_coverage_context};
       el/=se ->
            Tables  = proplists:get_value(tables, SQL),  %% plural is an historical reference, not a typo
            Fields  = proplists:get_value(fields, SQL),
            Where   = proplists:get_value(where,  SQL),
            Limit   = proplists:get_value(limit,  SQL),
            Offset  = proplists:get_value(offset, SQL),
            OrderBy = proplists:get_value(order_by, SQL),
            GroupBy = proplists:get_value(group_by, SQL),
            case is_binary(Tables) of
                true ->
                    Mod = riak_ql_ddl:make_module_name(Tables),
                    {ok,
                     ?SQL_SELECT{'SELECT'   = #riak_sel_clause_v1{clause = Fields},
                                 'FROM'     = Tables,
                                 'WHERE'    = convert_where_timestamps(Mod, Where),
                                 'LIMIT'    = Limit,
                                 'OFFSET'   = Offset,
                                 'ORDER BY' = OrderBy,
                                 helper_mod = Mod,
                                 cover_context = Cover,
                                 group_by      = GroupBy }
                    };
                false ->
                    {error, <<"Must provide exactly one table name">>}
            end
    end;

build_sql_record_int(explain, SQL, Options) ->
    case build_sql_record_int(select, SQL, Options) of
        {ok, Select} ->
            {ok, #riak_sql_explain_query_v1{'EXPLAIN' = Select}};
        Error -> Error
    end;

build_sql_record_int(describe, SQL, _Options) ->
    D = proplists:get_value(identifier, SQL),
    {ok, #riak_sql_describe_v1{'DESCRIBE' = D}};

build_sql_record_int(insert, SQL, _Options) ->
    Table = proplists:get_value(table, SQL),
    case is_binary(Table) of
        true ->
            %% To support non- and partial specification of columns, the DDL
            %% Module, Fields specified in the query, and Values specified in
            %% the query are sent to the riak_ql_ddl module to flesh out the
            %% missing or partially specified column to value mappings.
            Mod = riak_ql_ddl:make_module_name(Table),
            FieldsInQuery = proplists:get_value(fields, SQL, []),
            ValuesInQuery = proplists:get_value(values, SQL),
            {Fields, Values} = riak_ql_ddl:insert_sql_columns(Mod, FieldsInQuery, ValuesInQuery),

            %% In order to convert timestamps given as strings, rather
            %% than doing this here, we pass Values as is and
            %% relegate the timestamp conversion to
            %% riak_kv_qry:do_insert. The conversion will take place
            %% after checking for too many values wrt the number of
            %% Fields. Otherwise, this check would need to be done
            %% twice as the timestamp conversion code needs to have
            %% the Types available (and check that length(Types) ==
            %% length(Values)).
            {ok, #riak_sql_insert_v1{'INSERT'   = Table,
                                     fields     = Fields,
                                     values     = Values,
                                     helper_mod = Mod
                                    }};
        false ->
            {error, <<"Must provide exactly one table name">>}
    end;

build_sql_record_int(delete, SQL, _Cover) ->
    T = proplists:get_value(table, SQL),
    case is_binary(T) of
        true ->
            Mod = riak_ql_ddl:make_module_name(T),
            W = proplists:get_value(where, SQL),
            {ok, #riak_sql_delete_query_v1{'FROM'     = T,
                                           'WHERE'    = W,
                                           helper_mod = Mod}};
         false ->
            {error, <<"Must provide exactly one table name">>}
    end;
build_sql_record_int(show_tables, _SQL, _Options) ->
    {ok, #riak_sql_show_tables_v1{}};
build_sql_record_int(show_create_table, SQL, _Options) -><commit_msg>convert timestamps in WHERE for DELETE (#1579)<commit_after>%% -------------------------------------------------------------------
%%
%% riak_kv_ts_util: supporting functions for timeseries code paths
%%
%% Copyright (c) 2016 <NAME>, Inc.  All Rights Reserved.
%%
%% This file is provided to you under the Apache License,
%% Version 2.0 (the "License"); you may not use this file
%% except in compliance with the License.  You may obtain
%% a copy of the License at
%%
%%   http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing,
%% software distributed under the License is distributed on an
%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
%% KIND, either express or implied.  See the License for the
%% specific language governing permissions and limitations
%% under the License.
%%
%% -------------------------------------------------------------------

%% @doc Various functions related to timeseries.

-module(riak_kv_ts_util).

-export([
         apply_timeseries_bucket_props/3,
         build_sql_record/3,
         check_table_feature_supported/2,
         is_table_supported/2,
         encode_typeval_key/1,
         explain_query/1, explain_query/2,
         get_column_types/2,
         get_table_ddl/1,
         lk/1,
         local_to_partition_key/2,
         maybe_parse_table_def/2,
         pk/1,
         queried_table/1,
         rm_rf/1,
         sql_record_to_tuple/1,
         sql_to_cover/6,
         table_to_bucket/1,
         validate_rows/2,
         varchar_to_timestamp/1
        ]).

-type ts_service_req() ::
    {ok, riak_kv_ts_svc:ts_requests(), {PermSpec::string(), Table::binary()}}.

-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
-compile(export_all).
-endif.

%% NOTE on table_to_bucket/1: Clients will work with table
%% names. Those names map to a bucket type/bucket name tuple in Riak,
%% with both the type name and the bucket name matching the table.
%%
%% Thus, as soon as code transitions from dealing with timeseries
%% concepts to Riak KV concepts, the table name must be converted to a
%% bucket tuple. This function is a convenient mechanism for doing so
%% and making that transition more obvious.

%%-include("riak_kv_wm_raw.hrl").
-include_lib("riak_ql/include/riak_ql_ddl.hrl").
-include("riak_kv_ts.hrl").

-spec sql_record_to_tuple(?SQL_SELECT{}) -> {binary(), [tuple(2) | tuple(3)], list(), [tuple(2)]}.
%% riak_ql_ddl:is_query_valid expects a tuple, not a SQL record
sql_record_to_tuple(?SQL_SELECT{'FROM'   = From,
                                'SELECT' = #riak_sel_clause_v1{clause=Select},
                                'WHERE'  = Where,
                                'ORDER BY' = OrderBy}) ->
    {From, Select, Where, OrderBy}.

-spec build_sql_record(riak_kv_qry:query_type(), proplists:proplist(), proplists:proplist()) ->
                              {ok, ?SQL_SELECT{}} | {error, any()}.
build_sql_record(Command, SQL, Options) ->
    try build_sql_record_int(Command, SQL, Options) of
        {ok, Record} ->
            {ok, Record};
        ER ->
            ER
    catch
        throw:Atom ->
            {error, Atom}
    end.


%% Convert the proplist obtained from the QL parser
build_sql_record_int(select, SQL, _Options = undefined) ->
    build_sql_record_int(select, SQL, []);
build_sql_record_int(select, SQL, Options) ->
    Cover  = proplists:get_value(cover, Options),
    if not (Cover == undefined orelse is_binary(Cover)) ->
            {error, bad_coverage_context};
       el/=se ->
            Tables  = proplists:get_value(tables, SQL),  %% plural is an historical reference, not a typo
            Fields  = proplists:get_value(fields, SQL),
            Where   = proplists:get_value(where,  SQL),
            Limit   = proplists:get_value(limit,  SQL),
            Offset  = proplists:get_value(offset, SQL),
            OrderBy = proplists:get_value(order_by, SQL),
            GroupBy = proplists:get_value(group_by, SQL),
            case is_binary(Tables) of
                true ->
                    Mod = riak_ql_ddl:make_module_name(Tables),
                    {ok,
                     ?SQL_SELECT{'SELECT'   = #riak_sel_clause_v1{clause = Fields},
                                 'FROM'     = Tables,
                                 'WHERE'    = convert_where_timestamps(Mod, Where),
                                 'LIMIT'    = Limit,
                                 'OFFSET'   = Offset,
                                 'ORDER BY' = OrderBy,
                                 helper_mod = Mod,
                                 cover_context = Cover,
                                 group_by      = GroupBy }
                    };
                false ->
                    {error, <<"Must provide exactly one table name">>}
            end
    end;

build_sql_record_int(explain, SQL, Options) ->
    case build_sql_record_int(select, SQL, Options) of
        {ok, Select} ->
            {ok, #riak_sql_explain_query_v1{'EXPLAIN' = Select}};
        Error -> Error
    end;

build_sql_record_int(describe, SQL, _Options) ->
    D = proplists:get_value(identifier, SQL),
    {ok, #riak_sql_describe_v1{'DESCRIBE' = D}};

build_sql_record_int(insert, SQL, _Options) ->
    Table = proplists:get_value(table, SQL),
    case is_binary(Table) of
        true ->
            %% To support non- and partial specification of columns, the DDL
            %% Module, Fields specified in the query, and Values specified in
            %% the query are sent to the riak_ql_ddl module to flesh out the
            %% missing or partially specified column to value mappings.
            Mod = riak_ql_ddl:make_module_name(Table),
            FieldsInQuery = proplists:get_value(fields, SQL, []),
            ValuesInQuery = proplists:get_value(values, SQL),
            {Fields, Values} = riak_ql_ddl:insert_sql_columns(Mod, FieldsInQuery, ValuesInQuery),

            %% In order to convert timestamps given as strings, rather
            %% than doing this here, we pass Values as is and
            %% relegate the timestamp conversion to
            %% riak_kv_qry:do_insert. The conversion will take place
            %% after checking for too many values wrt the number of
            %% Fields. Otherwise, this check would need to be done
            %% twice as the timestamp conversion code needs to have
            %% the Types available (and check that length(Types) ==
            %% length(Values)).
            {ok, #riak_sql_insert_v1{'INSERT'   = Table,
                                     fields     = Fields,
                                     values     = Values,
                                     helper_mod = Mod
                                    }};
        false ->
            {error, <<"Must provide exactly one table name">>}
    end;

build_sql_record_int(delete, SQL, _Cover) ->
    T = proplists:get_value(table, SQL),
    case is_binary(T) of
        true ->
            Mod = riak_ql_ddl:make_module_name(T),
            W = proplists:get_value(where, SQL),
            {ok, #riak_sql_delete_query_v1{'FROM'     = T,
                                           'WHERE'    = convert_where_timestamps(Mod, W),
                                           helper_mod = Mod}};
         false ->
            {error, <<"Must provide exactly one table name">>}
    end;
build_sql_record_int(show_tables, _SQL, _Options) ->
    {ok, #riak_sql_show_tables_v1{}};
build_sql_record_int(show_create_table, SQL, _Options) ->
{'fraction_non_alphanumeric': 0.11721562875050007, 'fraction_numerical': 0.005467395652753701, 'mean_word_length': 1.938099902056807, 'pattern_counts': {'":': 0, '<': 16, '<?xml version=': 0, '>': 66, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 32, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '886905', 'n_tokens_mistral': 4433, 'n_tokens_neox': 4274, 'n_words': 1156}