<commit_before>/*
 * The contents of this file are subject to the MonetDB Public License
 * Version 1.1 (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.monetdb.org/Legal/MonetDBLicense
 *
 * Software distributed under the License is distributed on an "AS IS"
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
 * License for the specific language governing rights and limitations
 * under the License.
 *
 * The Original Code is the MonetDB Database System.
 *
 * The Initial Developer of the Original Code is CWI.
 * Portions created by CWI are Copyright (C) 1997-July 2008 CWI.
 * Copyright August 2008-2015 MonetDB B.V.
 * All Rights Reserved.
 */

/* (c) <NAME>
 * Progress indicator
 * tachograph -d demo 
 * which connects to the demo database server and presents a server progress bar.
*/

#include "monetdb_config.h"
#include "monet_options.h"
#include <mapi.h>
#include <stream.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <sys/stat.h>
#include <errno.h>
#include <signal.h>
#include <math.h>
#include <unistd.h>
#include "mprompt.h"
#include "dotmonetdb.h"
#include "eventparser.h"

#ifndef HAVE_GETOPT_LONG
# include "monet_getopt.h"
#else
# ifdef HAVE_GETOPT_H
#  include "getopt.h"
# endif
#endif

#ifdef TIME_WITH_SYS_TIME
# include <sys/time.h>
# include <time.h>
#else
# ifdef HAVE_SYS_TIME_H
#  include <sys/time.h>
# else
#  include <time.h>
# endif
#endif

#define die(dbh, hdl)						\
	do {							\
		(hdl ? mapi_explain_query(hdl, stderr) :	\
		 dbh ? mapi_explain(dbh, stderr) :		\
		 fprintf(stderr, "!! command failed\n"));	\
		goto stop_disconnect;				\
	} while (0)

#define doQ(X)								\
	do {								\
		if ((hdl = mapi_query(dbh, X)) == NULL ||	\
		    mapi_error(dbh) != MOK)			\
			die(dbh, hdl);			\
	} while (0)

static stream *conn = NULL;
static char hostname[128];
static char *basefilename = "tacho";
static char *dbname;
static int beat = 5000;
static int delay = 0; // ms
static Mapi dbh;
static MapiHdl hdl = NULL;
static int interactive = 1;
static int capturing=0;
static int lastpc;
static int pccount;

#define RUNNING 1
#define FINISHED 2
typedef struct{
	int state;
	lng etc;
	lng actual;
	char *stmt;
} Event;

Event *events;

typedef struct {
	char *varname;
	char *source;
} Source;
Source *sources;	// original column name
int srctop, srcmax;

static void
addSource(char *varname, char *tbl, char *col)
{
	char buf[BUFSIZ];

	if(srctop == srcmax){
		if( srcmax == 0)
			sources = (Source *) malloc(2048 * sizeof(Source));
		else
			sources = (Source *) realloc(sources, srcmax+2048);
		srcmax+= 2048;
	}
	assert(sources);
	sources[srctop].varname = strdup(varname);
	snprintf(buf,BUFSIZ,"%s%s%s",tbl,(col?".":""),col?col:"");
	sources[srctop].source = strdup(buf);
	//fprintf(stderr,"addSource %s at %d  %s\n",varname, srctop, buf);
	srctop++;
}

static void
addSourcePair(char *varname, char *name)
{
	int i;

	if(srctop == srcmax){
		if( srcmax == 0)
			sources = (Source *) malloc(1024 * sizeof(Source));
		else
			sources = (Source *) realloc(sources, srcmax+1024);
		srcmax+= 1024;
	}
	for( i=0; i< srctop; i++)
	if( strcmp(name, sources[i].varname)==0){
		sources[srctop].varname = strdup(varname);
		sources[srctop].source = strdup(sources[i].source);
		//fprintf(stderr,"addSourcePair %s  %s\n",varname,sources[i].source);
		srctop++;
		return;
	}
}
static char *
fndSource(char *varname)
{
	int i;

	if(debug)
		fprintf(stderr,"fndSource %s\n",varname);
	for( i=0; i< srctop; i++)
	if( strcmp(varname, sources[i].varname)==0)
		return sources[i].source;
	return "TEMP";
}
/*
 * Parsing the argument list of a MAL call to obtain un-quoted string values
 */

static void
usageTachograph(void)
{
    fprintf(stderr, "tachograph [options] \n");
    fprintf(stderr, "  -d | --dbname=<database_name>\n");
    fprintf(stderr, "  -u | --user=<user>\n");
    fprintf(stderr, "  -p | --port=<portnr>\n");
    fprintf(stderr, "  -h | --host=<hostname>\n");
	fprintf(stderr, "  -b | --beat=<delay> in milliseconds (default 5000)\n");
	fprintf(stderr, "  -i | --interactive=<o | 1> show trace on stdout\n");
    fprintf(stderr, "  -o | --output=<webfile>\n");
    fprintf(stderr, "  -w | --wait=<delay time> in milliseconds\n");
    fprintf(stderr, "  -? | --help\n");
	exit(-1);
}

/* Any signal should be captured and turned into a graceful
 * termination of the profiling session. */

static void
stopListening(int i)
{
	fprintf(stderr,"signal %d received\n",i);
	if( dbh)
		doQ("profiler.stop();");
stop_disconnect:
	// show follow up action only once
	if(dbh)
		mapi_disconnect(dbh);
	exit(0);
}

char *currentfunction= 0;
char *currentquery= 0;
lng starttime = 0;
lng finishtime = 0;
lng duration =0;
int malsize = 0;
char *prevquery= 0;
int prevprogress =0;
int prevlevel =0; 
size_t txtlength=0;

static FILE *tachofd;

static void resetTachograph(void){
	int i;
	if (debug)
		fprintf(stderr, "RESET tachograph\n");
	if( prevprogress)
		printf("\n"); 
	for(i=0; i < malsize; i++)
	if( events[i].stmt)
		free(events[i].stmt);
	for(i=0; i< srctop; i++){
		free(sources[i].varname);
		free(sources[i].source);
	}
	srctop=0;
	malsize= 0;
	malargtop =0;
	currentfunction = 0;
	currentquery = 0;
	starttime = 0;
	finishtime = 0;
	duration =0;
	fclose(tachofd);
	tachofd = 0;
	prevprogress = 0;
	txtlength =0;
	prevlevel=0;
	lastpc = 0;
	pccount = 0;
	fflush(stdout);
	if( events){
		free(events);
		events = 0;
	}
}

static char stamp[BUFSIZ]={0};
static void
rendertime(lng ticks, int flg)
{
	int t, hr,min,sec;

	if( ticks == 0){
		strcpy(stamp,"unknown ");
		return;
	}
	t = (int) (ticks/1000000);
	sec = t % 60;
	min = (t /60) %60;
	hr = (t /3600);
	if( flg)
	snprintf(stamp,BUFSIZ,"%02d:%02d:%02d.%06d", hr,min,sec, (int) ticks %1000000); 
	else
	snprintf(stamp,BUFSIZ,"%02d:%02d:%02d", hr,min,sec); 
}

#define MSGLEN 100

/*
 * Render the output of the stethoscope into a more user-friendly format.
 * This involves removal of MAL variables and possibly renaming the MAL functions
 * by more general alternatives.
 * If mode is set then we go for a minimal base-table related display
 */
static struct{
	char *name;
	int length;
	char *alias;
	int newl;
	int mode;
}mapping[]={
	{"algebra.leftfetchjoinPath", 25, "join",4, 0},
	{"algebra.thetasubselect", 22, "select",6, 0},
	{"algebra.leftfetchjoin", 21, "join",4, 0},
	{"dataflow.language", 17,	"parallel", 8, 0},
	{"algebra.subselect", 17, "select",6, 0},
	{"sql.projectdelta", 16, "project",7, 0},
	{"algebra.subjoin", 15, "join",4, 0},
	{"language.pass(nil)", 18,	"pass", 4, 0},
	{"mat.packIncrement", 17, "pack",4, 0},
	{"language.pass", 13,	"pass", 4, 0},
	{"aggr.subcount", 13,	"count", 3, 0},
	{"sql.subdelta", 12, "project",7, 0},
	{"bat.append", 10,	"append", 6, 0},
	{"aggr.subavg", 11,	"average", 7, 0},
	{"aggr.subsum", 11,	"sum", 3, 0},
	{"aggr.submin", 11,	"minimum", 7, 0},
	{"aggr.submax", 11,	"maximum", 7, 0},
	{"aggr.count", 10,	"count", 3, 0},
	{"calc.lng", 8,	"long", 4, 0},
	{"sql.bind", 8,	"bind", 4, 0},
	{"batcalc.hge", 11, "hugeint", 7, 0},
	{"batcalc.dbl", 11, "real", 4, 0},
	{"batcalc.flt", 11, "real", 4, 0},
	{"batcalc.lng", 11, "bigint",6, 0},
	{"batcalc.", 8, "", 0, 0},
	{"calc.", 5, "", 0, 0},
	{"sql.", 4,	"", 0, 0},
	{"bat.", 4,	"", 0, 0},
	{"aggr.", 5,	"", 0, 0},
	{"group.sub", 9,	"", 0, 0},
	{"group.", 6,	"", 0, 0},
	{"mtime.", 6,	"", 0, 0},
	{0,0,0,0,0}};

static void
renderCall(char *line, int len, char *stmt, int state, int mode)
{
	char *limit= line + len, *l = line, *c = stmt;
	int i;
	char varname[BUFSIZ], *v=0;

	(void) state;
	// look for assignment
	c = strstr(c,":=");
	if( c) 
		c+=2;
	else c=stmt;

	while ( *c && isspace((int) *c)) c++;
	/* consider a function name remapping */
	if( mode){
		for(i=0; mapping[i].name; i++)
			if( strncmp(c,mapping[i].name, mapping[i].length) == 0){
				c+= mapping[i].length;
				sprintf(l,"%s",  mapping[i].alias);
				l += mapping[i].newl;
				*l=0;
				break;
			}
		while(*c && *c !='(') *l++= *c++;
	}  else
		while(*c && *c !='(') *l++= *c++;
	// handle argument list
	if( *c == '(') *l++ = *c++;
	for(; *c && *c !=')' && l < limit-1; ){
		v= 0;
		if(isalpha((int)*c) ){ 
			v= varname;
			while(*c && (isalnum((int)*c) || *c=='_')) *v++ = *c++;
			*v=0;
		}
		if( *c == '=') c++;
		if( *c == '<'){
			while(*c && *c !='>') c++;
			if(*c) c++;
			if (v && varname[0]){
				v= fndSource(varname);
				snprintf(l, len -strlen(line)-2,"%s",v);
				while(*l) l++;
			}
			while(*c && *c !=']' && l < limit -2) *l++ = *c++;
			if( *c == ']' && l < limit) *l++ = *c++;
		}
		if (*c == '"' ) {
			*l++ = *c++;
			while(*c && *c !='"' && *(c-1) !='\\' && l < limit-2 ) *l++ =*c++;
		}  else
			while(*c && *c !=':' && *c !=',' && l < limit-2) *l++ = *c++;
		if (*c == ':'){
			if( strncmp(c,":bat",4)== 0){
				while(*c && *c !=']') c++;
				if( *c == ']') c++;
			} else
				while(*c && *c != ',' && *c != ')') c++;
		}
		// literals
		if( strcmp(varname,"nil") == 0 || strcmp(varname,"true")==0 || strcmp(varname,"false")==0)
			for(v = varname; *v; ) *l++ = *v++;
		*l++= *c++;
	}
	*l++ = *c;
	*l=0;
}

static void
showBar(int level, lng clk, char *stmt)
{
	lng i =0, nl;
	size_t stamplen=0;
	char line[BUFSIZ]={0};

	nl = level/2-prevlevel/2;
	if(nl == 0 ||  level/2 <= prevlevel/2)
		return;
	assert(MSGLEN < BUFSIZ);
	if(prevlevel == 0)
		printf("[");
	else
	for( i= 50 - prevlevel/2 +txtlength; i>0; i--)
		printf("\b \b");
	for( i=0 ; i< nl ; i++)
		putchar('#');
	for( ; i < 50-prevlevel/2; i++)
		putchar('.');
	putchar(level ==100?']':'>');
	printf(" %3d%%",level);
	if( level == 100 || duration == 0){
		rendertime(clk,0);
		printf("  %s     ",stamp);
		stamplen= strlen(stamp)+3;
	} else
	if( duration && duration- clk > 0){
		rendertime(duration - clk,0);
		printf(" %c%s ETC ", (level == 100? '-':' '),stamp);
		stamplen= strlen(stamp)+3;
	} else
	if( duration && duration- clk < 0){
		rendertime(clk - duration ,0);
		printf(" +%s ETC ",stamp);
		stamplen= strlen(stamp)+3;
	} 
	renderCall(line,MSGLEN,(stmt?stmt:""),0,1);
	printf("%s",line);
	fflush(stdout);
	txtlength = 10 + stamplen + strlen(line);
	prevlevel = level;
}

/* create the progressbar JSON file for pickup */
static void
progressBarInit(char *qry)
{
	char buf[BUFSIZ];

	snprintf(buf,BUFSIZ,"%s_%s.json",basefilename,dbname);
	tachofd= fopen(buf,"w");
	if( tachofd == NULL){
		fprintf(stderr,"Could not create %s\n",buf);
		exit(0);
	}
	fprintf(tachofd,"{ \"tachograph\":0.1,\n");
	fprintf(tachofd," \"qid\":\"%s\",\n",currentfunction?currentfunction:"");
	fprintf(tachofd," \"query\":\"%s\",\n",qry);
	fprintf(tachofd," \"started\": "LLFMT",\n",starttime);
	fprintf(tachofd," \"duration\":"LLFMT",\n",duration);
	fprintf(tachofd," \"instructions\":%d\n",malsize);
	fprintf(tachofd,"},\n");
	fflush(tachofd);
}

static void
update(EventRecord *ev)
{
	int progress=0;
	int i,j;
	char *v, *qry, *q = 0, *c;
	int uid = 0,qid = 0;
	char line[BUFSIZ];
	char prereq[BUFSIZ]={0};
	char number[BUFSIZ]={0};
 
	/* handle a ping event, keep the current instruction in focus */
	if (ev->state >= PING ) {
		// All state events are ignored
		return;
	}

	if (debug)
		fprintf(stderr, "Update %s input %s stmt %s time " LLFMT"\n",(ev->state>=0?statenames[ev->state]:"unknown"),(ev->fcn?ev->fcn:"(null)"),(currentfunction?currentfunction:""),ev->clkticks -starttime);

	if (starttime == 0) {
		if (ev->fcn == 0 ) {
			if (debug)
				fprintf(stderr, "Skip %s input %s\n",(ev->state>=0?statenames[ev->state]:"unknown"),ev->fcn);
			return;
		}
		if (debug)
			fprintf(stderr, "Start capturing updates %s\n",ev->fcn);
	}
	if (ev->clkticks < 0) {
		/* HACK: *TRY TO* compensate for the fact that the MAL
		 * profiler chops-off day information, and assume that
		 * clkticks is < starttime because the tomograph run
		 * crossed a day boundary (midnight);
		 * we simply add 1 day (24 hours) worth of microseconds.
		 * NOTE: this surely does NOT work correctly if the
		 * tomograph run takes 24 hours or more ...
		 */
		ev->clkticks += US_DD;
	}

	/* monitor top level function brackets, we restrict ourselves to SQL queries */
	if (!capturing && ev->state == START && ev->fcn && strncmp(ev->fcn, "function", 8) == 0) {
		if( (i = sscanf(ev->fcn + 9,"user.s%d_%d",&uid,&qid)) != 2){
			if( debug)
				fprintf(stderr,"Start phase parsing %d, uid %d qid %d\n",i,uid,qid);
			return;
		}
		if (capturing++ == 0){
			starttime = ev->clkticks;
			finishtime = ev->clkticks + ev->ticks;
			duration = ev->ticks;
		}
		if (currentfunction == 0)
			currentfunction = strdup(ev->fcn+9);
		if (debug)
			fprintf(stderr, "Enter function %s capture %d\n", currentfunction, capturing);
		return;
	}
	ev->clkticks -= starttime;

	if ( !capturing)
		return;

	/* start of instruction box */
	if (ev->state == START ) {
		if(ev->fcn && strstr(ev->fcn,"querylog.define") ){
			// extract a string argument
			currentquery = malarguments[0];
			malsize = atoi(malarguments[3]);
			events = (Event*) malloc(malsize * sizeof(Event));
			memset((char*)events, 0, malsize * sizeof(Event));
			// use the truncated query text, beware that the \ is already escaped in the call argument.
			q = qry = (char *) malloc(strlen(currentquery) * 2);
			for (c= currentquery; *c; ){
				if ( strncmp(c,"\\\\t",3) == 0){
					*q++ = '\t';
					c+=3;
				} else
				if ( strncmp(c,"\\\\n",3) == 0){
					*q++ = '\n';
					c+=3;
				} else if ( strncmp(c,"\\\\",2) == 0){
					c+= 2;
				} else *q++ = *c++;
			}
			*q =0;
			currentquery = qry;
			if( ! (prevquery && strcmp(currentquery,prevquery)== 0) && interactive )
				printf("%s\n",qry);
			prevquery = currentquery;
			progressBarInit(ev->stmt);
		}
		events[ev->pc].state = RUNNING;
		renderCall(line,BUFSIZ, ev->stmt,0,1);
		events[ev->pc].stmt = strdup(ev->stmt);
		events[ev->pc].etc = ev->ticks;
		if( ev->pc > lastpc)
			lastpc = ev->pc;
		// keep track of sources, pick first variable only
		// We should use the MAL semantics to pick to proper heritage path
		if ( strstr(ev->stmt,"sql.tid") && *ev->stmt != '('){
			char nme[BUFSIZ],*n=nme, *s= ev->stmt;
			while(*s && *s != '=') *n++ = *s++;
			*n = 0;
			addSource(nme, malarguments[2], "");
		} else 
		if ( strstr(ev->stmt,"sql.bind") && *ev->stmt != '('){
			char nme[BUFSIZ],*n=nme, *s= ev->stmt;
			while(*s && *s != '=') *n++ = *s++;
			*n = 0;
			addSource(nme, malarguments[2], malarguments[3]);
		} else
		if ( strstr(ev->stmt,"sql.projectdelta") && *ev->stmt != '(' ){
			char nme[BUFSIZ],*n=nme, *s= ev->stmt;
			while(*s && *s != '=') *n++ = *s++;
			*n = 0;
			addSourcePair(nme, malvariables[1]);
		} else
		if ( strstr(ev->stmt,"algebra.leftfetchjoin") && *ev->stmt != '(' ){
			char nme[BUFSIZ],*n=nme, *s= ev->stmt;
			while(*s && *s != '=') *n++ = *s++;
			*n = 0;
			addSourcePair(nme, malvariables[malvartop-1]);
		} else
		if ( strstr(ev->stmt,"algebra.subjoin") && *ev->stmt != '(' ){
			char nme[BUFSIZ],*n=nme, *s= ev->stmt;
			while(*s && *s != '=') *n++ = *s++;
			*n = 0;
			addSourcePair(nme, malvariables[malvartop-1]);
		} else {
			char nme[BUFSIZ],*n=nme, *s= ev->stmt;
			if( *s =='(') s++;
			while(*s && *s != '=') *n++ = *s++;
			*n = 0;
			// update the source direction
			for( i=0; i< malvartop;i++)
				addSourcePair(nme,malvariables[i]);
		}
		fprintf(tachofd,"{\n");
		fprintf(tachofd,"\"qid\":\"%s\",\n",currentfunction?currentfunction:"");
		fprintf(tachofd,"\"pc\":%d,\n",ev->pc);
		fprintf(tachofd,"\"time\": "LLFMT",\n",ev->clkticks);
		fprintf(tachofd,"\"status\": \"start\",\n");
		fprintf(tachofd,"\"estimate\": "LLFMT",\n",ev->ticks);
		fprintf(tachofd,"\"stmt\": \"%s\",\n",ev->stmt);
		fprintf(tachofd,"\"beautystmt\": \"%s\",\n",line);
		// collect all input producing PCs
		fprintf(tachofd,"\"prereq\":[");
		for( i=0; i < malvartop; i++){
			// remove duplicates
			for(j= ev->pc-1; j>=0;j --){
				//if(debug)
					//fprintf(stderr,"locate %s in %s\n",malvariables[i], events[j].stmt);
				if(events[j].stmt && (v = strstr(events[j].stmt, malvariables[i])) && v < strstr(events[j].stmt,":=")){
					snprintf(number,BUFSIZ,"%d",j);
					if( strstr(prereq,number) == 0)
						snprintf(prereq + strlen(prereq), BUFSIZ-1-strlen(prereq), "%s%d",(i?", ":""), j);
					//fprintf(tachofd,"%s%d",(i?", ":""), j);
					break;
				}
			}
		}
		fprintf(tachofd,"%s",prereq);
		fprintf(tachofd,"]\n");
		fprintf(tachofd,"},\n");
		fflush(tachofd);

		clearArguments();
		return;
	}
	if( tachofd == NULL){
		if( debug) fprintf(stderr,"No tachofd available\n");
		return;
	}
	/* end the instruction box */
	if (ev->state == DONE ){
			
		fprintf(tachofd,"{\n");
		fprintf(tachofd,"\"qid\":\"%s\",\n",currentfunction?currentfunction:"");
		fprintf(tachofd,"\"pc\":%d,\n",ev->pc);
		fprintf(tachofd,"\"time\": "LLFMT",\n",ev->clkticks);
		fprintf(tachofd,"\"status\": \"done\",\n");<commit_msg>Filter one query instance The code is not yet aware of multiple queries in flight. This could be deduced from the 'tag'. For now, we pick one random sample.<commit_after>/*
 * The contents of this file are subject to the MonetDB Public License
 * Version 1.1 (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.monetdb.org/Legal/MonetDBLicense
 *
 * Software distributed under the License is distributed on an "AS IS"
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
 * License for the specific language governing rights and limitations
 * under the License.
 *
 * The Original Code is the MonetDB Database System.
 *
 * The Initial Developer of the Original Code is CWI.
 * Portions created by CWI are Copyright (C) 1997-July 2008 CWI.
 * Copyright August 2008-2015 MonetDB B.V.
 * All Rights Reserved.
 */

/* (c) <NAME>
 * Progress indicator
 * tachograph -d demo 
 * which connects to the demo database server and presents a server progress bar.
*/

#include "monetdb_config.h"
#include "monet_options.h"
#include <mapi.h>
#include <stream.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <sys/stat.h>
#include <errno.h>
#include <signal.h>
#include <math.h>
#include <unistd.h>
#include "mprompt.h"
#include "dotmonetdb.h"
#include "eventparser.h"

#ifndef HAVE_GETOPT_LONG
# include "monet_getopt.h"
#else
# ifdef HAVE_GETOPT_H
#  include "getopt.h"
# endif
#endif

#ifdef TIME_WITH_SYS_TIME
# include <sys/time.h>
# include <time.h>
#else
# ifdef HAVE_SYS_TIME_H
#  include <sys/time.h>
# else
#  include <time.h>
# endif
#endif

#define die(dbh, hdl)						\
	do {							\
		(hdl ? mapi_explain_query(hdl, stderr) :	\
		 dbh ? mapi_explain(dbh, stderr) :		\
		 fprintf(stderr, "!! command failed\n"));	\
		goto stop_disconnect;				\
	} while (0)

#define doQ(X)								\
	do {								\
		if ((hdl = mapi_query(dbh, X)) == NULL ||	\
		    mapi_error(dbh) != MOK)			\
			die(dbh, hdl);			\
	} while (0)

static stream *conn = NULL;
static char hostname[128];
static char *basefilename = "tacho";
static char *dbname;
static int beat = 5000;
static int delay = 0; // ms
static Mapi dbh;
static MapiHdl hdl = NULL;
static int interactive = 1;
static int capturing=0;
static int lastpc;
static int pccount;

#define RUNNING 1
#define FINISHED 2
typedef struct{
	int state;
	lng etc;
	lng actual;
	char *stmt;
} Event;

Event *events;

typedef struct {
	char *varname;
	char *source;
} Source;
Source *sources;	// original column name
int srctop, srcmax;

static void
addSource(char *varname, char *tbl, char *col)
{
	char buf[BUFSIZ];

	if(srctop == srcmax){
		if( srcmax == 0)
			sources = (Source *) malloc(2048 * sizeof(Source));
		else
			sources = (Source *) realloc(sources, srcmax+2048);
		srcmax+= 2048;
	}
	assert(sources);
	sources[srctop].varname = strdup(varname);
	snprintf(buf,BUFSIZ,"%s%s%s",tbl,(col?".":""),col?col:"");
	sources[srctop].source = strdup(buf);
	//fprintf(stderr,"addSource %s at %d  %s\n",varname, srctop, buf);
	srctop++;
}

static void
addSourcePair(char *varname, char *name)
{
	int i;

	if(srctop == srcmax){
		if( srcmax == 0)
			sources = (Source *) malloc(1024 * sizeof(Source));
		else
			sources = (Source *) realloc(sources, srcmax+1024);
		srcmax+= 1024;
	}
	for( i=0; i< srctop; i++)
	if( strcmp(name, sources[i].varname)==0){
		sources[srctop].varname = strdup(varname);
		sources[srctop].source = strdup(sources[i].source);
		//fprintf(stderr,"addSourcePair %s  %s\n",varname,sources[i].source);
		srctop++;
		return;
	}
}
static char *
fndSource(char *varname)
{
	int i;

	if(debug)
		fprintf(stderr,"fndSource %s\n",varname);
	for( i=0; i< srctop; i++)
	if( strcmp(varname, sources[i].varname)==0)
		return sources[i].source;
	return "TEMP";
}
/*
 * Parsing the argument list of a MAL call to obtain un-quoted string values
 */

static void
usageTachograph(void)
{
    fprintf(stderr, "tachograph [options] \n");
    fprintf(stderr, "  -d | --dbname=<database_name>\n");
    fprintf(stderr, "  -u | --user=<user>\n");
    fprintf(stderr, "  -p | --port=<portnr>\n");
    fprintf(stderr, "  -h | --host=<hostname>\n");
	fprintf(stderr, "  -b | --beat=<delay> in milliseconds (default 5000)\n");
	fprintf(stderr, "  -i | --interactive=<o | 1> show trace on stdout\n");
    fprintf(stderr, "  -o | --output=<webfile>\n");
    fprintf(stderr, "  -w | --wait=<delay time> in milliseconds\n");
    fprintf(stderr, "  -? | --help\n");
	exit(-1);
}

/* Any signal should be captured and turned into a graceful
 * termination of the profiling session. */

static void
stopListening(int i)
{
	fprintf(stderr,"signal %d received\n",i);
	if( dbh)
		doQ("profiler.stop();");
stop_disconnect:
	// show follow up action only once
	if(dbh)
		mapi_disconnect(dbh);
	exit(0);
}

char *currentfunction= 0;
char *currentquery= 0;
int currenttag;		// to distinguish query invocations
lng starttime = 0;
lng finishtime = 0;
lng duration =0;
int malsize = 0;
char *prevquery= 0;
int prevprogress =0;
int prevlevel =0; 
size_t txtlength=0;

static FILE *tachofd;

static void resetTachograph(void){
	int i;
	if (debug)
		fprintf(stderr, "RESET tachograph\n");
	if( prevprogress)
		printf("\n"); 
	for(i=0; i < malsize; i++)
	if( events[i].stmt)
		free(events[i].stmt);
	for(i=0; i< srctop; i++){
		free(sources[i].varname);
		free(sources[i].source);
	}
	srctop=0;
	malsize= 0;
	malargtop =0;
	currentfunction = 0;
	currenttag = 0;
	currentquery = 0;
	starttime = 0;
	finishtime = 0;
	duration =0;
	fclose(tachofd);
	tachofd = 0;
	prevprogress = 0;
	txtlength =0;
	prevlevel=0;
	lastpc = 0;
	pccount = 0;
	fflush(stdout);
	if( events){
		free(events);
		events = 0;
	}
}

static char stamp[BUFSIZ]={0};
static void
rendertime(lng ticks, int flg)
{
	int t, hr,min,sec;

	if( ticks == 0){
		strcpy(stamp,"unknown ");
		return;
	}
	t = (int) (ticks/1000000);
	sec = t % 60;
	min = (t /60) %60;
	hr = (t /3600);
	if( flg)
	snprintf(stamp,BUFSIZ,"%02d:%02d:%02d.%06d", hr,min,sec, (int) ticks %1000000); 
	else
	snprintf(stamp,BUFSIZ,"%02d:%02d:%02d", hr,min,sec); 
}

#define MSGLEN 100

/*
 * Render the output of the stethoscope into a more user-friendly format.
 * This involves removal of MAL variables and possibly renaming the MAL functions
 * by more general alternatives.
 * If mode is set then we go for a minimal base-table related display
 */
static struct{
	char *name;
	int length;
	char *alias;
	int newl;
	int mode;
}mapping[]={
	{"algebra.leftfetchjoinPath", 25, "join",4, 0},
	{"algebra.thetasubselect", 22, "select",6, 0},
	{"algebra.leftfetchjoin", 21, "join",4, 0},
	{"dataflow.language", 17,	"parallel", 8, 0},
	{"algebra.subselect", 17, "select",6, 0},
	{"sql.projectdelta", 16, "project",7, 0},
	{"algebra.subjoin", 15, "join",4, 0},
	{"language.pass(nil)", 18,	"pass", 4, 0},
	{"mat.packIncrement", 17, "pack",4, 0},
	{"language.pass", 13,	"pass", 4, 0},
	{"aggr.subcount", 13,	"count", 3, 0},
	{"sql.subdelta", 12, "project",7, 0},
	{"bat.append", 10,	"append", 6, 0},
	{"aggr.subavg", 11,	"average", 7, 0},
	{"aggr.subsum", 11,	"sum", 3, 0},
	{"aggr.submin", 11,	"minimum", 7, 0},
	{"aggr.submax", 11,	"maximum", 7, 0},
	{"aggr.count", 10,	"count", 3, 0},
	{"calc.lng", 8,	"long", 4, 0},
	{"sql.bind", 8,	"bind", 4, 0},
	{"batcalc.hge", 11, "hugeint", 7, 0},
	{"batcalc.dbl", 11, "real", 4, 0},
	{"batcalc.flt", 11, "real", 4, 0},
	{"batcalc.lng", 11, "bigint",6, 0},
	{"batcalc.", 8, "", 0, 0},
	{"calc.", 5, "", 0, 0},
	{"sql.", 4,	"", 0, 0},
	{"bat.", 4,	"", 0, 0},
	{"aggr.", 5,	"", 0, 0},
	{"group.sub", 9,	"", 0, 0},
	{"group.", 6,	"", 0, 0},
	{"mtime.", 6,	"", 0, 0},
	{0,0,0,0,0}};

static void
renderCall(char *line, int len, char *stmt, int state, int mode)
{
	char *limit= line + len, *l = line, *c = stmt;
	int i;
	char varname[BUFSIZ], *v=0;

	(void) state;
	// look for assignment
	c = strstr(c,":=");
	if( c) 
		c+=2;
	else c=stmt;

	while ( *c && isspace((int) *c)) c++;
	/* consider a function name remapping */
	if( mode){
		for(i=0; mapping[i].name; i++)
			if( strncmp(c,mapping[i].name, mapping[i].length) == 0){
				c+= mapping[i].length;
				sprintf(l,"%s",  mapping[i].alias);
				l += mapping[i].newl;
				*l=0;
				break;
			}
		while(*c && *c !='(') *l++= *c++;
	}  else
		while(*c && *c !='(') *l++= *c++;
	// handle argument list
	if( *c == '(') *l++ = *c++;
	for(; *c && *c !=')' && l < limit-1; ){
		v= 0;
		if(isalpha((int)*c) ){ 
			v= varname;
			while(*c && (isalnum((int)*c) || *c=='_')) *v++ = *c++;
			*v=0;
		}
		if( *c == '=') c++;
		if( *c == '<'){
			while(*c && *c !='>') c++;
			if(*c) c++;
			if (v && varname[0]){
				v= fndSource(varname);
				snprintf(l, len -strlen(line)-2,"%s",v);
				while(*l) l++;
			}
			while(*c && *c !=']' && l < limit -2) *l++ = *c++;
			if( *c == ']' && l < limit) *l++ = *c++;
		}
		if (*c == '"' ) {
			*l++ = *c++;
			while(*c && *c !='"' && *(c-1) !='\\' && l < limit-2 ) *l++ =*c++;
		}  else
			while(*c && *c !=':' && *c !=',' && l < limit-2) *l++ = *c++;
		if (*c == ':'){
			if( strncmp(c,":bat",4)== 0){
				while(*c && *c !=']') c++;
				if( *c == ']') c++;
			} else
				while(*c && *c != ',' && *c != ')') c++;
		}
		// literals
		if( strcmp(varname,"nil") == 0 || strcmp(varname,"true")==0 || strcmp(varname,"false")==0)
			for(v = varname; *v; ) *l++ = *v++;
		*l++= *c++;
	}
	*l++ = *c;
	*l=0;
}

static void
showBar(int level, lng clk, char *stmt)
{
	lng i =0, nl;
	size_t stamplen=0;
	char line[BUFSIZ]={0};

	nl = level/2-prevlevel/2;
	if(nl == 0 ||  level/2 <= prevlevel/2)
		return;
	assert(MSGLEN < BUFSIZ);
	if(prevlevel == 0)
		printf("[");
	else
	for( i= 50 - prevlevel/2 +txtlength; i>0; i--)
		printf("\b \b");
	for( i=0 ; i< nl ; i++)
		putchar('#');
	for( ; i < 50-prevlevel/2; i++)
		putchar('.');
	putchar(level ==100?']':'>');
	printf(" %3d%%",level);
	if( level == 100 || duration == 0){
		rendertime(clk,0);
		printf("  %s    ",stamp);
		stamplen= strlen(stamp)+3;
	} else
	if( duration && duration- clk > 0){
		rendertime(duration - clk,0);
		printf(" ETC%c%s ", (level == 100? '-':' '),stamp);
		stamplen= strlen(stamp)+3;
	} else
	if( duration && duration- clk < 0){
		rendertime(clk - duration ,0);
		printf(" ETC+%s ",stamp);
		stamplen= strlen(stamp)+3;
	} 
	renderCall(line,MSGLEN,(stmt?stmt:""),0,1);
	printf("%s",line);
	fflush(stdout);
	txtlength = 9 + stamplen + strlen(line);
	prevlevel = level;
}

/* create the progressbar JSON file for pickup */
static void
progressBarInit(char *qry)
{
	char buf[BUFSIZ];

	snprintf(buf,BUFSIZ,"%s_%s.json",basefilename,dbname);
	tachofd= fopen(buf,"w");
	if( tachofd == NULL){
		fprintf(stderr,"Could not create %s\n",buf);
		exit(0);
	}
	fprintf(tachofd,"{ \"tachograph\":0.1,\n");
	fprintf(tachofd," \"qid\":\"%s\",\n",currentfunction?currentfunction:"");
	fprintf(tachofd," \"tag\":\"%d\",\n",currenttag);
	fprintf(tachofd," \"query\":\"%s\",\n",qry);
	fprintf(tachofd," \"started\": "LLFMT",\n",starttime);
	fprintf(tachofd," \"duration\":"LLFMT",\n",duration);
	fprintf(tachofd," \"instructions\":%d\n",malsize);
	fprintf(tachofd,"},\n");
	fflush(tachofd);
}

static void
update(EventRecord *ev)
{
	int progress=0;
	int i,j;
	char *v, *qry, *q = 0, *c;
	int uid = 0,qid = 0;
	char line[BUFSIZ];
	char prereq[BUFSIZ]={0};
	char number[BUFSIZ]={0};
 
	/* handle a ping event, keep the current instruction in focus */
	if (ev->state >= PING ) {
		// All state events are ignored
		return;
	}

	if (debug)
		fprintf(stderr, "Update %s input %s stmt %s time " LLFMT"\n",(ev->state>=0?statenames[ev->state]:"unknown"),(ev->fcn?ev->fcn:"(null)"),(currentfunction?currentfunction:""),ev->clkticks -starttime);

	if (starttime == 0) {
		if (ev->fcn == 0 ) {
			if (debug)
				fprintf(stderr, "Skip %s input %s\n",(ev->state>=0?statenames[ev->state]:"unknown"),ev->fcn);
			return;
		}
		if (debug)
			fprintf(stderr, "Start capturing updates %s\n",ev->fcn);
	}
	if (ev->clkticks < 0) {
		/* HACK: *TRY TO* compensate for the fact that the MAL
		 * profiler chops-off day information, and assume that
		 * clkticks is < starttime because the tomograph run
		 * crossed a day boundary (midnight);
		 * we simply add 1 day (24 hours) worth of microseconds.
		 * NOTE: this surely does NOT work correctly if the
		 * tomograph run takes 24 hours or more ...
		 */
		ev->clkticks += US_DD;
	}

	/* monitor top level function brackets, we restrict ourselves to SQL queries */
	if (!capturing && ev->state == START && ev->fcn && strncmp(ev->fcn, "function", 8) == 0) {
		if( (i = sscanf(ev->fcn + 9,"user.s%d_%d",&uid,&qid)) != 2){
			if( debug)
				fprintf(stderr,"Start phase parsing %d, uid %d qid %d\n",i,uid,qid);
			return;
		}
		if (capturing++ == 0){
			starttime = ev->clkticks;
			finishtime = ev->clkticks + ev->ticks;
			duration = ev->ticks;
		}
		if (currentfunction == 0){
			currentfunction = strdup(ev->fcn+9);
			currenttag = ev->tag;
		}
		if (debug)
			fprintf(stderr, "Enter function %s capture %d\n", currentfunction, capturing);
		return;
	}
	ev->clkticks -= starttime;

	if ( !capturing)
		return;

	/* start of instruction box */
	if (ev->state == START ) {
		if(ev->fcn && strstr(ev->fcn,"querylog.define") ){
			// extract a string argument
			currentquery = malarguments[0];
			malsize = atoi(malarguments[3]);
			events = (Event*) malloc(malsize * sizeof(Event));
			memset((char*)events, 0, malsize * sizeof(Event));
			// use the truncated query text, beware that the \ is already escaped in the call argument.
			q = qry = (char *) malloc(strlen(currentquery) * 2);
			for (c= currentquery; *c; ){
				if ( strncmp(c,"\\\\t",3) == 0){
					*q++ = '\t';
					c+=3;
				} else
				if ( strncmp(c,"\\\\n",3) == 0){
					*q++ = '\n';
					c+=3;
				} else if ( strncmp(c,"\\\\",2) == 0){
					c+= 2;
				} else *q++ = *c++;
			}
			*q =0;
			currentquery = qry;
			if( ! (prevquery && strcmp(currentquery,prevquery)== 0) && interactive )
				printf("%s\n",qry);
			prevquery = currentquery;
			progressBarInit(ev->stmt);
		}
		if( ev->tag != currenttag)
			return;	// forget all except one query
		events[ev->pc].state = RUNNING;
		renderCall(line,BUFSIZ, ev->stmt,0,1);
		events[ev->pc].stmt = strdup(ev->stmt);
		events[ev->pc].etc = ev->ticks;
		if( ev->pc > lastpc)
			lastpc = ev->pc;
		// keep track of sources, pick first variable only
		// We should use the MAL semantics to pick to proper heritage path
		if ( strstr(ev->stmt,"sql.tid") && *ev->stmt != '('){
			char nme[BUFSIZ],*n=nme, *s= ev->stmt;
			while(*s && *s != '=') *n++ = *s++;
			*n = 0;
			addSource(nme, malarguments[2], "");
		} else 
		if ( strstr(ev->stmt,"sql.bind") && *ev->stmt != '('){
			char nme[BUFSIZ],*n=nme, *s= ev->stmt;
			while(*s && *s != '=') *n++ = *s++;
			*n = 0;
			addSource(nme, malarguments[2], malarguments[3]);
		} else
		if ( strstr(ev->stmt,"sql.projectdelta") && *ev->stmt != '(' ){
			char nme[BUFSIZ],*n=nme, *s= ev->stmt;
			while(*s && *s != '=') *n++ = *s++;
			*n = 0;
			addSourcePair(nme, malvariables[1]);
		} else
		if ( strstr(ev->stmt,"algebra.leftfetchjoin") && *ev->stmt != '(' ){
			char nme[BUFSIZ],*n=nme, *s= ev->stmt;
			while(*s && *s != '=') *n++ = *s++;
			*n = 0;
			addSourcePair(nme, malvariables[malvartop-1]);
		} else
		if ( strstr(ev->stmt,"algebra.subjoin") && *ev->stmt != '(' ){
			char nme[BUFSIZ],*n=nme, *s= ev->stmt;
			while(*s && *s != '=') *n++ = *s++;
			*n = 0;
			addSourcePair(nme, malvariables[malvartop-1]);
		} else {
			char nme[BUFSIZ],*n=nme, *s= ev->stmt;
			if( *s =='(') s++;
			while(*s && *s != '=') *n++ = *s++;
			*n = 0;
			// update the source direction
			for( i=0; i< malvartop;i++)
				addSourcePair(nme,malvariables[i]);
		}
		fprintf(tachofd,"{\n");
		fprintf(tachofd,"\"qid\":\"%s\",\n",currentfunction?currentfunction:"");
		fprintf(tachofd,"\"tag\":%d,\n",ev->tag);
		fprintf(tachofd,"\"pc\":%d,\n",ev->pc);
		fprintf(tachofd,"\"time\": "LLFMT",\n",ev->clkticks);
		fprintf(tachofd,"\"status\": \"start\",\n");
		fprintf(tachofd,"\"estimate\": "LLFMT",\n",ev->ticks);
		fprintf(tachofd,"\"stmt\": \"%s\",\n",ev->stmt);
		fprintf(tachofd,"\"beautystmt\": \"%s\",\n",line);
		// collect all input producing PCs
		fprintf(tachofd,"\"prereq\":[");
		for( i=0; i < malvartop; i++){
			// remove duplicates
			for(j= ev->pc-1; j>=0;j --){
				//if(debug)
					//fprintf(stderr,"locate %s in %s\n",malvariables[i], events[j].stmt);
				if(events[j].stmt && (v = strstr(events[j].stmt, malvariables[i])) && v < strstr(events[j].stmt,":=")){
					snprintf(number,BUFSIZ,"%d",j);
					if( strstr(prereq,number) == 0)
						snprintf(prereq + strlen(prereq), BUFSIZ-1-strlen(prereq), "%s%d",(i?", ":""), j);
					//fprintf(tachofd,"%s%d",(i?", ":""), j);
					break;
				}
			}
		}
		fprintf(tachofd,"%s",prereq);
		fprintf(tachofd,"]\n");
		fprintf(tachofd,"},\n");
		fflush(tachofd);

		clearArguments();
		return;
	}
	if( tachofd == NULL){
		if( debug) fprintf(stderr,"No tachofd available\n");
		return;
	}
	/* end the instruction box */
	if (ev->state == DONE ){
			
		if( ev->tag != currenttag)
			return;	// forget all except one query
		fprintf(tachofd,"{\n");
		fprintf(tachofd,"\"qid\":\"%s\",\n",currentfunction?currentfunction:"");
		fprintf(tachofd,"\"tag\":%d,\n",ev->tag);
		fprintf(tachofd,"\"pc\":%d,\n",ev->pc);
		fprintf(tachofd,"\"time\": "LLFMT",\n",ev->clkticks);
		fprintf(tachofd,"\"status\": \"done\",\n");
{'fraction_non_alphanumeric': 0.22066826613611096, 'fraction_numerical': 0.02241207794487682, 'mean_word_length': 3.6839797176922024, 'pattern_counts': {'":': 51, '<': 89, '<?xml version=': 0, '>': 190, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1340052', 'n_tokens_mistral': 15462, 'n_tokens_neox': 13985, 'n_words': 3355}