Slow Query - no results in DDErl

[c-bik]

QUERY

    SELECT *
        FROM TABLE(
            SLOW_TEXT_TABLE (
                :SQLT_STR_TEXT,
                :SQLT_INT_ROWS,
                :SQLT_INT_DELAY
            )
        )

CRASH

[info] [_DDRL_] {dderl_session,183} Query: 
<<"SELECT\r\n    *\r\nFROM\r\n    TABLE (SLOW_TEXT_TABLE(:SQLT_STR_TEXT, :SQLT_INT_ROWS, :SQLT_INT_DELAY))">>
With parameters:
[{<<":SQLT_INT_DELAY">>,in,'SQLT_INT',1},{<<":SQLT_INT_ROWS">>,in,'SQLT_INT',1000},{<<":SQLT_STR_TEXT">>,in,'SQLT_STR',<<"FOO_BAR_">>}]
Executed.
[error] gen_statem <0.2081.0> in state filling terminated with reason:
 {timeout,{gen_server,call,[<0.2080.0>,close]}} in gen_server:call/2 line 206
[error] CRASH REPORT Process <0.2081.0> with 0 neighbours exited with reason:
 {timeout,{gen_server,call,[<0.2080.0>,close]}} in gen_server:call/2 line 206
[error] [_DDRL_] {dderl_session,648} Problem processing command: exit:
 {{timeout,{gen_server,call,[<0.2080.0>,close]}},{gen_statem,call,[<0.2081.0>,get_count,infinity]}}
[{<<"button">>,[{<<"connection">>,<<"g2gDZAAIb2NpX3BvcnRnZAAQZGRlcmwxQDEyNy4wLjAuMQAAB8QAAAAAA24GAKDTBycnAg==">>},{<<"statement">>,<<"g2gCZAAJZGRlcmxfZnNtZ2QAEGRkZXJsMUAxMjcuMC4wLjEAAAghAAAAAAM=">>},{<<"binds">>,[{<<":SQLT_INT_DELAY">>,[{<<"typ">>,<<"SQLT_INT">>},{<<"dir">>,<<"in">>},{<<"val">>,<<"1">>},{<<"name">>,<<":SQLT_INT_DELAY">>}]},{<<":SQLT_INT_ROWS">>,[{<<"typ">>,<<"SQLT_INT">>},{<<"dir">>,<<"in">>},{<<"val">>,<<"1000">>},{<<"name">>,<<":SQLT_INT_ROWS">>}]},{<<":SQLT_STR_TEXT">>,[{<<"typ">>,<<"SQLT_STR">>},{<<"dir">>,<<"in">>},{<<"val">>,<<"FOO_BAR_">>},{<<"name">>,<<":SQLT_STR_TEXT">>}]}]},{<<"btn">>,<<"restart">>}]}]
[{gen,do_call,4,[{file,"gen.erl"},{line,177}]},
{gen_statem,call,3,[{file,"gen_statem.erl"},{line,461}]},
{oci_adapter,process_cmd,6,[{file,"_build/default/lib/dderl/src/oci_adapter.erl"},{line,499}]},
{dderl_session,spawn_process_call,8,[{file,"_build/default/lib/dderl/src/dderl_session.erl"},{line,645}]}]

[error] [_DDRL_] {dderl_session,191} dderl_session received unknown msg
 {'EXIT',<0.2081.0>,{timeout,{gen_server,call,[<0.2080.0>,close]}}} for <<"system">>

Query Details

CREATE OR REPLACE TYPE T_TEXT_ROW AS OBJECT (text_line  VARCHAR2(4000));
CREATE OR REPLACE TYPE T_TEXT_TAB IS TABLE OF T_TEXT_ROW;
create or replace function SLOW_TEXT_TABLE (
     SQLT_STR_TEXT  IN Varchar2,
     SQLT_INT_ROWS  IN INTEGER,
     SQLT_VNR_DELAY IN NUMBER
 ) RETURN T_TEXT_TAB PIPELINED
 is
 begin
     FOR i IN 1..SQLT_INT_ROWS LOOP
         PIPE ROW(T_TEXT_ROW(SQLT_STR_TEXT || i));
         sys.dbms_lock.sleep(SQLT_VNR_DELAY);
     END LOOP;
 end SLOW_TEXT_TABLE;

ERLOCI Code

Rows can be fetched with direct erloci APIs successfully!

expand for details

Tns = <<"(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=tcp)(HOST=192.168.1.43)(PORT=1521)))(CONNECT_DATA=(SERVICE_NAME=XE)))">>.
Pswd= <<"****">>.
User= <<"scott">>.

f(OciPort).
OciPort = erloci:new([{logging, true}, {env, []}]).
f(OciSession).
OciSession = OciPort:get_session(Tns, User, Pswd).

f(CreateType).
CreateType = OciSession:prep_sql(<<"CREATE OR REPLACE TYPE T_TEXT_ROW AS OBJECT (text_line  VARCHAR2(4000));">>).
{executed, 0} = CreateType:exec_stmt().
ok = CreateType:close().
f(CreateType).
CreateType = OciSession:prep_sql(<<"CREATE OR REPLACE TYPE T_TEXT_TAB IS TABLE OF T_TEXT_ROW;">>).
{executed, 0} = CreateType:exec_stmt().
ok = CreateType:close().

f(CreateFunction).
CreateFunction = OciSession:prep_sql(<<"
 create or replace function SLOW_TEXT_TABLE (
     SQLT_STR_TEXT  IN Varchar2,
     SQLT_INT_ROWS  IN INTEGER,
     SQLT_VNR_DELAY IN NUMBER
 ) RETURN T_TEXT_TAB PIPELINED
 is
 begin
     FOR i IN 1..SQLT_INT_ROWS LOOP
         PIPE ROW(T_TEXT_ROW(SQLT_STR_TEXT || i));
         sys.dbms_lock.sleep(SQLT_VNR_DELAY);
     END LOOP;
 end SLOW_TEXT_TABLE;
">>).
{executed, 0} = CreateFunction:exec_stmt().
ok = CreateFunction:close().

f(SelectSlow).
SelectSlow = OciSession:prep_sql(<<"
    SELECT *
        FROM TABLE(
            SLOW_TEXT_TABLE (
                :SQLT_STR_TEXT,
                :SQLT_INT_ROWS,
                :SQLT_INT_DELAY
            )
        )
">>).
ok = SelectSlow:bind_vars([
    {<<":SQLT_STR_TEXT">>, 'SQLT_STR'},
    {<<":SQLT_INT_ROWS">>, 'SQLT_INT'},
    {<<":SQLT_INT_DELAY">>, 'SQLT_INT'}]).
f(Cols).
{cols, Cols} = SelectSlow:exec_stmt([{<<"FOO_BAR_">>, 1000, 1}], 1).
f(F).
(fun F() ->
    case timer:tc(fun() -> SelectSlow:fetch_rows(1) end)of
        {Time, {{rows, []}, true}} -> io:format("[~8.3.0f ms] EOT~n", [Time / 1000]);
        {Time, {{rows, Rows}, true}} ->
             io:format("[~8.3.0f ms] Last ~p~n", [Time / 1000, Rows]);
        {Time, {{rows, Rows}, _}} ->
             io:format("[~8.3.0f ms] Rows ~p~n", [Time / 1000, Rows]),
             F()
    end
 end)().

ok = SelectSlow:close().

OciSession:close().
OciPort:close().

FYI: @stoch

[acautin]

@c-bik I tested the query and indeed we have a timeout when pressing restart which I am looking into right now. However in your example using the driver directly you are fetching only one row SelectSlow:fetch_rows(1) in the normal dderl case we are fetching by blocks of 100, if I replace that fetch SelectSlow:fetch_rows(100) I will get not get any response until the 100 rows are produced, isn't the idea the driver responds with 1 row as soon as it is produced by the pipeline ?

[c-bik]

@acautin

in the normal dderl case we are fetching by blocks of 100

This is hardcoded in DDErl. I propose a way to optionally override it (ddConfig or a input in SQL Editor GUI). Having said that this is no longer a bug but a new feature proposal. What do you think?

isn't the idea the driver responds with 1 row as soon as it is produced by the pipeline ?

The way this fetch_rows(N) feature is implemented in driver is:

// 4GB is the maximum size you can write stdout as we are framing with 32 bit length header
if ( rows_fetched < N && erl_rows_term_size < 4GB ) {
    // fetch more rows
} else {
   // reply to erlang with fetched rows
}

There are no fetch timeout in current driver and packing multiple rows in one payload increases pipe throughput.

However, a new idea can be fetch_rows_async/0 API which will send rows (one or more) as messages as soon as they arrive to directly to the process (as long as the process lives). Will that help in this case? If it does, can you please give me a feature request in erloci with this?

[acautin]

@c-bik probably will have to implement a configurable fetch block size at the end in any case so the user can tweak it as his preference, but still I would like to find a better solution for it to work even if not optimally with the default settings. As a solution on the fsm we will buffer rows for a configurable period of time defaulting to 1 second and serve what we have at that point even if it is less than the buffer size. A similar solution probably can be implemented in oci so the fetch_rows(N) is implemented more or less as:

// 4GB is the maximum size you can write stdout as we are framing with 32 bit length header
if ( rows_fetched < N && erl_rows_term_size < 4GB && timerNotExpired ) {
    // fetch more rows
} else {
   // reply to erlang with fetched rows
}

Will add an issue to erloci with this feature request.

FYI: @stoch