:- ensure_loaded(common).
cvs_id('$Id: index.pl,v 1.28 2005/04/14 22:17:40 sandro Exp $').

:- ensure_loaded(library(debug)).
:- ensure_loaded(qtest).
:- ensure_loaded(persistent_term_index).

:- ensure_loaded(persistence).
%   introduce   label_* 9r keyword_*  like word_*   for a stronger/better match
:- persistent(word_subj_graph/3, [duplicates(true), key(c_string)]).
:- persistent(word_subj_graph/1, [duplicates(false), key(c_string)]).
:- persistent(word_count/2, [value(c_long), key(c_string)]).
:- persistent(subj_graph/2, [duplicates(true)]).
:- persistent(pred_graph/2, [duplicates(true)]).
:- persistent(obj_graph/2, [duplicates(true)]).
:- persistent(address_ret/2, [duplicates(true)]).
:- persistent(address_latest_ret/2).
:- persistent(address_latest_successful_ret/2).

:- thread_local(graph_page/1).

:- multi_dynamic(index_queue/2).
:- multi_dynamic(indexing/2).
:- persistent(graph_index_done/1, [duplicates(false)]).

%%  rebuild_indexes
%%
%%  Delete all the index files and rebuild them from the rdfpage data. 
%%
%%  Limitations:
%%
%%       Not meant to be used in running system!
%%
%%       Call via "make index" or "make index-sandbox"
%%
rebuild_indexes :-
	% persistent_term_index_clear,
	load_retrievals,
	rebuild_retrieval_indexes,
	rebuild_graph_indexes.

rebuild_retrieval_indexes :-
	forall(  rdf(Ret, retr:initialResource, _),
		 index_retrieval_metadata(Ret)
	      ).

rebuild_graph_indexes :-
	request_indexing(_),
	debug(index),
	debug(index_detail),
	start_indexer,
	!,
	print_message(informational, format('waiting for indexing', [])),
	repeat,
	sleep(1),
	(  indexing(Ret, Graph)
	-> flag(index_count, Count, Count),
           print_message(informational, format('~w R=~q G=~q', [Count, Ret, Graph]))
	;  true
	),
	\+ once(index_queue(_,_)),
	print_message(informational, format('indexing done.', [])).

old_rebuild_retrieval_indexes :-
	% assume db files removed !
	rdf_reset_db,

	forall(  rdfpage_exists_storage(S),
		 (
		   (   rdf_statistics(triples(T0)),
		       format('T=~w considering page ~q: ', [T0, S]),
		       rdfpage_load_safe(S),
		       rdf_statistics(triples(T1)),
		       PT is T1-T0,
		       format('[loaded ~w triples] ', [PT]),
		       rdfpage_sub_uri(S, retrieval, Ret),
		       rdfpage(Ret, retr:initialResource, _)
		   ->  index_retrieval_metadata(Ret),
		       write('[indexed] ')
		   ;   write(' (not retrieval)')
		   ),
		   rdf_reset_db,
		   nl
	      )),
        persistence_close_all.

old_rebuild_graph_indexes :-

	rdf_reset_db,

	forall(  (   rdfpage_exists_storage(S),
		     \+graph_page(S)   % avoid 2nd look at data pages
		 ),
		 (   rdf_statistics(triples(T0)),
		     format('T=~w considering page ~q: ', [T0, S]),
		     % write('.'),
		     (   % is there actually an RDF graph stored for this page?
			 % (if so, load it)
			 rdfpage_load_safe(S),
			 rdf_statistics(triples(T1)),
			 PT is T1-T0,
			 format('[loaded ~w triples] ', [PT]),
			 %forall(   rdf(Subj,P,O,S),
			%	   format('~q ~q ~q.', [Subj, P, O])
			%       ),
		         % and does the page give essential Ret data?
			 once(rdf(Ret, retr:graph, Graph, S)),
			 write('[has graph] '),
			 % is it about itself?
			 rdfpage_sub_uri(S, _, Ret),
			 write('[self] ')
		     ->  % okay!   load the Graph, too
			 rdfpage_load_safe(Graph),
			 rdf_statistics(triples(T2)),
			 GT is T2-T1,
			 format('[graph loaded,  ~w triples] ', [GT]),
			 flush_output,
			 assert(graph_page(Graph)),
			 index_graph(Ret),
			 write('[indexed] ')
			 % rdfpage_unload(Graph)
		     ;   true
		     ),
		     % rdfpage_unload(S),
		     rdf_reset_db,
		     nl
		 )
	      ),
	persistence_close_all.

%%  index_retrieval_metadata(+Ret)
%%
%%  Update persistent predicates:
%%       address_ret/2
%%       address_latest_ret/2
%%       address_latest_successful_ret/2
%%
%%  Limitations:
%%
%%       This must be called in the order that Retrievals are made; it
%%       does not look at times.
%%
%%
%%  ISSUE -- what about ones on the QUEUE or in progres?  Nothing
%%  gets recorded for them, so how do we find them?  I think they're
%%  tracked in memory - and if memory is lost, so are they....
%%
index_retrieval_metadata(Ret) :-
	once(ret_get(Ret, retr:initialResource, Address)),
	debug(index, 'indexing retrieval metadata ~a (~a)', [Ret, Address]),
	passert(address_ret(Address, Ret)),
	passert(address_latest_ret(Address, Ret)),
	(   ret_get(Ret, rdf:type, retr:'SuccessfulRetrieval')
	->  passert(address_latest_successful_ret(Address, Ret))
	;   true
	).


start_indexer :-
	print_message(error, skipping_indexer).
start_indexer :-
        request_indexing(_),
	thread_create(index_loop, _, [detached(true),
					  alias(indexer),  %measured high water
					  global(32768),  %2160 max
					  local(32768),  %1511 max
					  trail(8192),    % 430 max
					  argument(1024),
					  stack(1024)
					 ]).


%%
%%  request_indexing(?Ret)
%%
%%  Add all the graph-chunks associated with all matching Retrievals
%%  to the index_queue, unless they are already done indexing or on
%%  the queue.
%%

request_indexing(Ret) :-
	forall(  (   rdfpage(Ret, retr:chunk, Graph),
		     \+ index_queue(Ret, Graph),
		     \+ indexing(Ret, Graph),
		     \+ graph_index_done(Graph)
		 ),
		 assertz(index_queue(Ret, Graph))
	      ).

index_loop :-
	thread_set_state(waiting),
	repeat,
	sleep(0.05),
	retract(index_queue(Ret, Graph)),
	flag(index_count, _, 0),
	assert(indexing(Ret, Graph)),
	protected_index_graph(Ret, Graph),
	retract(indexing(Ret, Graph)),
	thread_set_state(sleeping),
	thread_high_water(HW),
	log_term(HW),
	fail.

% maybe this should be  protected(index_graph(Ret, Graph))  ?
protected_index_graph(Ret, Graph) :-
	(  catch(index_graph(Ret, Graph), Error, true)
	-> true
	;  Error=failed
	),
	(   var(Error)
	->  passert(graph_index_done(Graph))
	;   log_term(index_error(Error, Ret, Graph)),
	    print_message(error, index_error(Error, Ret, Graph))

	),
	!.


%%  index_graph(Ret, Graph)
%%
%%  Update the indexes to include the RDF graph parsed from the given
%%  data from a given Retrieval
%%
%%  Side-effect: qsaves the Ret page, so we remember this has been
%%  indexed!

:- thread_local passert_once/1.
:- thread_local rdf_copy/3.

index_graph(Ret, Graph) :-
	debug(index, 'indexing graph: ~a', Graph),
	pagein_graph(Graph),
        get_time(StartTime),
	rdfpage_sub_uri(RetPage, _, Ret),
        rdf_transaction(rdf_assert(Graph, retr:indexingStarts, literal(unixtime(StartTime)), RetPage)),
	retractall(passert_once(_)),
	retractall(rdf_copy(_,_,_)),
	forall(rdf(S,P,O,Graph),
	       assert(rdf_copy(S,P,O))
	      ),
	forall(   rdf_copy(Subj, Pred, Obj),
		  index_triple(Subj, Pred, Obj, Graph)
	      ),
	findall(Clause,
		passert_once(Clause),
		Clauses),
	sort(Clauses, SortedClauses),
	checklist(passert, SortedClauses),
	retractall(passert_once(_)),
	get_time(EndTime),
        rdf_transaction(rdf_assert(Graph, retr:indexingEnds, literal(unixtime(EndTime)), RetPage)),
	rdfpage_qsave(RetPage),
	debug(index, 'done with retrieved graph ~a', Ret).

index_triple(Subj, Pred, Obj, Graph) :-
	assert(passert_once(subj_graph(Subj, Graph))),
	assert(passert_once(pred_graph(Pred, Graph))),
	(   atom(Obj)
	->  assert(passert_once(obj_graph(Obj, Graph)))
	;   (   (  Obj=literal(AtomText)
		;  Obj=literal(lang(_LangTag, AtomText))
		),
		atom(AtomText),
		!, 
		atom_codes(AtomText, Text)
	    ->  debug(index_triple, 'indexing "~q"', AtomText),
		persistent_term_index(values(Subj, Graph), Usage),
		persistence_handle(word_subj_graph, 3, WSR3),
		persistence_handle(word_subj_graph, 1, WSR1),
		break_text_no_dups(Text, index_word(Usage, WSR3, WSR1))
	    ;   true    % don't index datatyped things yet....
	    )
	),
	%  hack: add the last word in the subject URI as a word 
	(   rdf_split_url(_, Word, Subj),
	    \+ Word = ''
	->  	persistent_term_index(values(Subj, Graph), Usage),
		persistence_handle(word_subj_graph, 3, WSR3),
		persistence_handle(word_subj_graph, 1, WSR1),
	        index_word(Usage, WSR3, WSR1, Word)
	;       true
	),
	flag(index_count, N, N+1).


	

index_word(Usage, WSR3, WSR1, Word) :-
	(   word_count(Word, Count)
	->  true
	;   Count=0
	),
	CountPlusOne is Count+1,
	passert(word_count(Word, CountPlusOne)),
	debug(index_word, 'count(~q) = ~q', [Word, CountPlusOne]),
	(   stopword(Word)
	->  debug(index_word, '(stopword, not indexed)', [])
	;   %fast version of: passert(word_subj_graph(Word, Subj, Graph)),
	    db_put(WSR3, Word, Usage),
	    atom_number(UsageAtom, Usage),
	    concat_atom([Word, '/', UsageAtom], Combined),
	    db_put(WSR1, Combined, 1)
	).

% there's a little window where we could count the current retrieval
% as done and in progress, but this doesn't seem worth a mutex to
% solve....
ret_index_done_fraction(Ret, Frac) :-
	findall(Triples,
		(   rdfpage(Ret, retr:chunk, Chunk),
		    rdf(Chunk, retr:triples, literal(Triples))
		), TotalParts),
	sumlist(TotalParts, Total),
	Total > 0,
	findall(Triples,
		(   rdfpage(Ret, retr:chunk, Chunk),
		    graph_index_done(Chunk),
		    rdf(Chunk, retr:triples, literal(Triples))
		), DoneParts),
	sumlist(DoneParts, Done1),
	(   indexing(Ret, _)
        ->  flag(index_count, N, N),
            Done is Done1 + N
        ;   Done is Done1
	),
	Frac is Done / Total.



/* Stop word....   Hmm.   See

   http://www.grey-cat.com/tunabreath-arch/2003/09/20/stop_words.html
and
   http://www.snowball.tartarus.org/archives/snowball-discuss/0318.html

   to get some ideas about how this is a hard problem.

   We can always filter for stopwords at search time, so maybe we
should be more aggressive than this.  But let's start with the list
attributed to Google.

   G="a about an and are as at be by en for from how i in is it of on or that the this to was what when where which who why will with",
   break_text(G, format('stopword(~q).~n')).
*/

stopword(a).
stopword(about).
stopword(an).
stopword(and).
stopword(are).
stopword(as).
stopword(at).
stopword(be).
stopword(by).
stopword(en).
stopword(for).
stopword(from).
stopword(how).
stopword(i).
stopword(in).
stopword(is).
stopword(it).
stopword(of).
stopword(on).
stopword(or).
stopword(that).
stopword(the).
stopword(this).
stopword(to).
stopword(was).
stopword(what).
stopword(when).
stopword(where).
stopword(which).
stopword(who).
stopword(why).
stopword(will).
stopword(with).


/*
  
?- [index_document].
?- time(break_text("   Hello, sweet world!   ", format('Got: ~q~n'))).
Got: hello
Got: sweet
Got: world
% 286 inferences, 0.00 CPU in 0.00 seconds (0% CPU, Infinite Lips)

bench(break_text("   hello, sweet world!   ", noop)).
Run 1: 70.688725us

... and it's linear.
  
*/



break_text(Text, Callback) :-
	append(Text, " ", TextPlus),
	brk(Callback, TextPlus, "").

:- thread_local(already_used/1).

break_text_no_dups(Text, Callback) :-
	retractall(already_used(_)),
	break_text(Text, nodups(Callback)),
	retractall(already_used(_)).

nodups(Callback, Item) :-
	(   already_used(Item)
	->  true
	;   assert(already_used(Item)),
	    call(Callback, Item)
	).

ychar(DL) -->
	[D], { code_type(D, alnum), code_type(D, to_upper(DL))}.
nchar -->
	[D], { \+ code_type(D, alnum) }.
ystr([D|T]) -->
	ychar(D), ystr(T)
	; ychar(D), nchar, {T=[]}.
nstr -->
	nchar, nstr
	; nchar.
w(C) -->
	ystr(Text),
	(nstr; ""),
	{ atom_codes(Atom, Text), call(C, Atom) }.
ylist(C) -->
	w(C), ylist(C)
	; w(C)
	; "".
brk(C) -->
	(nstr; ""), ylist(C), (nstr; "").

%
%  Tests for break_text
%

noop(_).
gather(L, I) :-
	memberchk(I, L).
qtest :-
	break_text("", gather(L)),
	length(L, _),   % terminate the list
	L == [].
qtest :-
	break_text(" ", gather(L)),
	length(L, _),   % terminate the list
	L == [].
qtest :-
	break_text("  ", gather(L)),
	length(L, _),   % terminate the list
	L == [].
qtest :-
	break_text("x", gather(L)),
	length(L, _),   % terminate the list
	L == [x].
qtest :-
	break_text(" x", gather(L)),
	length(L, _),   % terminate the list
	L == [x].
qtest :-
	break_text("x ", gather(L)),
	length(L, _),   % terminate the list
	L == [x].
qtest :-
	break_text(" x ", gather(L)),
	length(L, _),   % terminate the list
	L == [x].
qtest :-
	break_text("Hello sweet world", gather(L)),
	length(L, _),   % terminate the list
	L == [hello, sweet, world].
qtest :-
	break_text("   Hello, sweet world!   ", gather(L)),
	length(L, _),   % terminate the list
	L == [hello, sweet, world].

qtest :-
	break_text_no_dups("Hello, sweet world, world, world!!", gather(L)),
	length(L, _),   % terminate the list
	L == [hello, sweet, world].

print_word_table :-
	findall(cw(C,W), word_count(W,C), A),
	sort(A, AS),
	forall(member(cw(C,W), AS),  format('~q: ~q~n', [W,C])).







/*

  Performance:

      * profile a chunk of cyc
      * can we copy the triples into a list
               10k triples should be <1M stack
        quickly and be happy?
        NAH - findall/findsome uses the store anyway.
time((
     flag(x, _, 0),
     retractall(tmp_rdf(_,_,_,_)),
     rdf(S,P,O,G),
     assert(tmp_rdf(S,P,O,G)),
     flag(x, A, A+1),
     A > 10000
    )).
=== 0.13s on swada, full sys

  http://www.w3.org/2004/ontaria/page/1644
time((
     flag(x, _, 0),
     retractall(tmp_rdf(_,_,_,_)),
     rdf(S,P,O,'http://www.w3.org/2004/ontaria/page/1644'),
     assert(tmp_rdf(S,P,O,G)),
     flag(x, A, A+1),
     A > 20000
    )).

=== 0.54, 0.54, 0.54
(retractall == 0.06s)

=== 0.37 with 1000
      73 with 20000
*/