anchor.diff:      end;        
   handle_info({mnesia_table_event,{write,Record,_ActivityId}}, #state{reply=Sock,fetchCtx=FetchCtx0,statement=Stmt}=State) ->
       %io:format(user, "~p - received mnesia subscription event ~p ~p~n", [?MODULE, write, Record]),
  -    #fetchCtx{status=Status,metarec=MetaRec,remaining=Remaining0}=FetchCtx0,
  +    #fetchCtx{status=Status,metarec=MetaRec,remaining=Remaining0,tailSpec=TailSpec}=FetchCtx0,
       case Status of
  -        tailing ->  
  -            case length(Stmt#statement.tables) of
  -                1 ->    
  -                    Wrap = fun(X) -> {X, MetaRec} end,
  -                    if  
  -                        Remaining0 > 1 ->
  -                            send_reply_to_client(Sock, {lists:map(Wrap, [Record]),false}),
  -                            {noreply, State#state{fetchCtx=FetchCtx0#fetchCtx{remaining=Remaining0-1}}};
  -                        true ->
  -                            send_reply_to_client(Sock, {lists:map(Wrap, [Record]),true}),
  -                            unsubscribe(Stmt),
  -                            {noreply, State#state{fetchCtx=#fetchCtx{}}}
  -                    end;
  -                _N ->    
  -                    ?UnimplementedException({"Joins not supported",Stmt#statement.tables})
  +        tailing ->
  +            case ets:match_spec_run([Record],TailSpec) of
  +                [] ->  
  +                    {noreply, State};
  +                [Rec] ->       
  +                    case length(Stmt#statement.tables) of
  +                        1 ->    
  +                            Wrap = fun(X) -> {X, MetaRec} end,
  +                            if  
  +                                Remaining0 > 1 ->
  +                                    send_reply_to_client(Sock, {lists:map(Wrap, [Rec]),false}),
  +                                    {noreply, State#state{fetchCtx=FetchCtx0#fetchCtx{remaining=Remaining0-1}}};
  +                                true ->
  +                                    send_reply_to_client(Sock, {lists:map(Wrap, [Rec]),true}),
  +                                    unsubscribe(Stmt),
  +                                    {noreply, State#state{fetchCtx=#fetchCtx{}}}
  +                            end;
  +                        _N ->    
  +                            case join_rows([Rec], FetchCtx0, Stmt) of
  +                                [] ->
  +                                    {noreply, State};
  +                                Result ->    
  +                                    Complete = (Remaining0 =< length(Result)),
  +                                    % io:format(user, "sending rows ~p~n", [Result]),
  +                                    send_reply_to_client(Sock, {Result, Complete}),
  +                                    case Complete of
  +                                        true -> 
  +                                            unsubscribe(Stmt),
  +                                            {noreply, State#state{fetchCtx=#fetchCtx{}}};
  +                                        false ->
  +                                            {noreply, State#state{fetchCtx=FetchCtx0#fetchCtx{remaining=Remaining0-length(Result)}}}
  +                                    end
  +                            end
  +                    end
               end;
           _ ->
               {noreply, State}
  
positive.code:                     , blockSize=100 ::integer()     %% could be adaptive
                      , remaining ::integer()         %% rows remaining to be fetched. initialized to Limit and decremented
                      , opts = [] ::list()            %% fetch options like {tail_mode,true}
                      }).
  
  -record(state,                  %% state for statment process, including fetch subprocess
  
negative.code:         ?assertEqual(ok, insert_range(SKey, 5, def, 'Imem', IsSec)),
          ?assertEqual(5,imem_meta:table_size(def)),
  
          Sql3 = "select col1, col2 from def where col2 <= 5;",
          io:format(user, "Query: ~p~n", [Sql3]),
          {ok, _Clm3, _RowFun3, StmtRef3} = imem_sql:exec(SKey, Sql3, 100, 'Imem', IsSec),
          try
              ?assertEqual(ok, imem_statement:fetch_recs_async(SKey, StmtRef3, self(), [{tail_mode,true}], IsSec)),
  