tinydbquery.java

来自「nesC写的heed算法」· Java 代码 · 共 824 行 · 第 1/2 页

	 a currently running query.
	 */
    public Message setRateMessage(int rate) {
	QueryMsg m = new QueryMsg();
	initCommonFields(m);
	m.set_msgType(SET_RATE_MSG);
	this.epochDur = (short)(rate / MS_PER_EPOCH_DUR_UNIT);
	m.set_epochDuration(epochDur);
	return m;
    }

    public void setDropTables() {
	this.dropTables = true;
	
    }
	
    /* Return a vector of strings containing the
	 headings for the columns in this query
	 */
    public Vector getColumnHeadings() {
		Vector cols = new Vector();
		boolean addedGroupCol = false;
		
		cols.addElement("Epoch");
		
		if (isAgg()) {
			//it's an agg; the columns that are returned
			//are the group and the aggregate value
			for (int i = 0; i < numExprs(); i++) {
				QueryExpr e = getExpr(i);
				
				//System.out.println("Expression " + i + " is " + e.isAgg() + "   " + e);
				
				if (e.isAgg()) {
					AggExpr ae = (AggExpr)e;
					if (ae.getGroupField() != -1 && !addedGroupCol) {
						cols.addElement(groupColName());
						addedGroupCol = true;
					}
					String aggString = "";
					aggString += ae.getAgg().toString() + "(" + getField(ae.getField()).getName();
					
					if (ae.getFieldOp() != ArithOps.NO_OP) {
						aggString += ArithOps.getStringValue(ae.getFieldOp()) + " ";
						aggString += ae.getFieldConst();
					}
					
					aggString += ")";
					
					cols.addElement(aggString);
					
				}
				if (TinyDBMain.debug) System.out.println(cols);
			}
			
		} else {
			//its a selection; the columns that are returned
			//are the exprs
			for (int i =0; i < numFields(); i++) {
				QueryField qf = getField(i);
				if (qf != null) cols.addElement(qf.getName());
			}
			
		}
		
		return cols;
    }
	
    //Returns the type (as defined in QueryField) of the specified
    //column in the result set.
    public byte getFieldType(int idx) throws ArrayIndexOutOfBoundsException {
		boolean addedGroupCol = false;
		if (idx == 0) return QueryField.INTTWO;
		if (isAgg()) {
			for (int i = 0; i < numExprs(); i++) {
				QueryExpr e = getExpr(i);
				
				if (e.isAgg()) {
					AggExpr ae = (AggExpr)e;
					if (ae.getGroupField() != -1 && !addedGroupCol) {
						if (--idx == 0) return getField(groupExpr.getGroupField()).getType();
						addedGroupCol = true;
					}
					if (--idx == 0) return getField(ae.getField()).getType();
				}
				
			}
			throw new ArrayIndexOutOfBoundsException();
		} else {
			for (int i = 0; i < numFields(); i++) {
				QueryField qf = getField(i);
				if (--idx == 0) return qf.getType();
			}
			throw new ArrayIndexOutOfBoundsException();
		}
		
    }
	
    
    /** Return an Iterator over messages to be sent
	 to start sensors running this
	 query
	 */
    public Iterator messageIterator() {
		ArrayList messages = new ArrayList();
		Message msg;
		QueryMsg qrm;
		
		//first, set up all the fields
		for (int i = 0; i < fields.size(); i++) {
			QueryField f = (QueryField)fields.get(i);
			
			qrm = new QueryMsg();
			msg = qrm;
			initCommonFields(qrm);
			qrm.set_type(FIELD);
			qrm.set_idx((byte)i);
			qrm.set_u_field_op(f.getOp());
			qrm.setString_u_field_name(f.getName());
			qrm.set_u_field_type(f.getType());
			if (TinyDBMain.debug) System.out.println(qrm.toString());
			
			messages.add(msg);
		}
		
		//then all the exprs
		for (int i = 0; i < exprs.size(); i++) {
			QueryExpr e = (QueryExpr)exprs.get(i);
			
			qrm = new QueryMsg();
			msg = qrm;
			initCommonFields(qrm);
			
			qrm.set_type(EXPR);
			qrm.set_idx((byte)i);
			qrm.set_u_expr_opType((byte)(e.isAgg()?(((AggExpr)e).isTemporalAgg()?TEMPORAL_AGG_EXPR:AGG_EXPR):SEL_EXPR));
			
			
			qrm.set_u_expr_fieldOp(e.getFieldOp());
			qrm.set_u_expr_fieldConst(e.getFieldConst());
			if (e.isAgg()) {
				AggExpr a = (AggExpr)e;
				
				qrm.set_u_expr_ex_agg_field(e.getField());
				qrm.set_u_expr_ex_agg_op(a.getAggOpCode());
				qrm.set_u_expr_ex_agg_groupingField(a.getGroupField());
				qrm.set_u_expr_ex_agg_groupFieldOp(a.getGroupFieldOp());
				qrm.set_u_expr_ex_agg_groupFieldConst(a.getGroupFieldConst());
			
				// set up arguments, for both temporal and non-temporal ones
				AggOp ag = a.getAgg();//needs renaming?
				for (int j=0; j < ag.getArguments().size(); j++) {
					qrm.setElement_u_expr_ex_tagg_args(j, ag.getArgument(j));
				}
			} else {
				SelExpr s = (SelExpr)e;
				
				if (s.isString()) {
					qrm.set_u_expr_isStringExp((byte)1);
					qrm.set_u_expr_ex_sexp_field(e.getField());
					qrm.set_u_expr_ex_sexp_op((byte)s.getSelOpCode());
					qrm.setString_u_expr_ex_sexp_s(s.getStringConst());
				} else {
					qrm.set_u_expr_isStringExp((byte)0);
					qrm.set_u_expr_ex_opval_field(e.getField());
					qrm.set_u_expr_ex_opval_op((byte)s.getSelOpCode());
					qrm.set_u_expr_ex_opval_value(s.getValue());
				}
			}
			if (TinyDBMain.debug) {
				System.out.println("expr msg: ");
				System.out.print(qrm.toString());
			}
			messages.add(msg);
		}
		
		//the command, if this is a command buffer
		if (hasCmd) {
		    qrm = new QueryMsg();
		    msg = qrm;
			initCommonFields(qrm);
			qrm.set_type(BUFFER);
			qrm.setString_u_buf_cmd_name(cmdName);
			qrm.set_u_buf_cmd_hasParam((short)(hasParam?1:0));
			qrm.set_u_buf_cmd_param(paramVal);
			if (TinyDBMain.debug) System.out.println("command msg: " + qrm.toString());
			messages.add(msg);
		} else if (ramBuffer) { //or, might be a ram buffer
		    qrm = new QueryMsg();
		    msg = qrm;
			initCommonFields(qrm);
			qrm.set_type(BUFFER);
			qrm.set_u_buf_ram_numRows(bufSize);
			qrm.set_u_buf_ram_policy(EVICT_OLDEST_POLICY);
			qrm.set_u_buf_ram_create(createTable?(byte)1:(byte)0);
			qrm.set_u_buf_ram_hasOutput(hasName?(byte)1:(byte)0);
			qrm.setString_u_buf_ram_outBufName(queryName);
			qrm.set_u_buf_ram_hasInput(hasInputBuf?(byte)1:(byte)0);
			qrm.setString_u_buf_ram_inBufName(inputBufferName);
			if (TinyDBMain.debug) System.out.println("ram buffer msg: " + qrm.toString());
			messages.add(msg);
		}
		
		//if this query is triggered by an event, send in the event
		if (hasEvent) {
		    qrm = new QueryMsg();
		    msg = qrm;
			initCommonFields(qrm);
			qrm.set_type(EVENT);
			qrm.setString_u_eventName(eventName);
			if (TinyDBMain.debug) System.out.println("event message: " + qrm.toString());
			messages.add(msg);
		}
		
		//if this query is for a fixed number of epochs, send the number of epochs
		if (numEpochs > 0) {
		    qrm = new QueryMsg();
		    msg = qrm;
			initCommonFields(qrm);
			qrm.set_type(N_EPOCHS);
			qrm.set_u_numEpochs(numEpochs);
			if (TinyDBMain.debug) System.out.println("num epochs message: " + qrm.toString());
			messages.add(msg);
		}

		//if this is a drop message, that's it
		if (dropTables) {
		    qrm = new QueryMsg();
		    msg = qrm;
		    initCommonFields(qrm);
		    qrm.set_msgType(DROP_TABLE);			
		    if (TinyDBMain.debug) System.out.println("drop message: " + qrm.toString());
		    qrm.set_u_ttl(DEL_MSG_TTL);
		    messages.add(msg);
		}

		
		return messages.iterator();
    }
	
    // set up common fields in radio messages
    private void initCommonFields(QueryMsg m) {

      m.set_msgType(ADD_MSG);
      m.set_qid(qid);
      m.set_fwdNode(TinyDBNetwork.UART_ADDR); 
      m.set_numFields((byte)fields.size());
      m.set_numExprs((byte)exprs.size());
      m.set_fromCatalogBuffer(fromCatalogBuf?(byte)1:(byte)0);
      m.set_fromBuffer(fromCatalogBuf?catalogTableId:from_qid);
      
      m.set_bufferType(hasCmd?COMMAND_BUFFER:(ramBuffer?EEPROM_BUFFER:RADIO_BUFFER));

      m.set_epochDuration(epochDur);
      if (hasEvent)
	  m.set_hasEvent((byte)1);
      else
	  m.set_hasEvent((byte)0);
      if (numEpochs > 0)
	  m.set_hasForClause((byte)1);
      else
	  m.set_hasForClause((byte)0);
      for (int i = 0; i < 5; i++) {
	  m.setElement_timeSyncData(i, (short)0);
      }
      m.set_clockCount((short)0);
    }
	
    /* Return the id fo the query this query reads results from */
    public byte getFromQid() {
		return from_qid;
    }
    
    /** Set the id of the query this query reads results from */
    public void setFromQid(byte qid) {
		this.from_qid = qid;
    }
	
	public void setFromCatalogTable(byte catalogTable) {
		this.fromCatalogBuf = true;
		this.catalogTableId = catalogTable;
	}
	
    public boolean isFromCatalogTable() {
		return fromCatalogBuf;
    }
    
    /** Specify that this query should output results to a RAM based buffer */
	public void useRamBuffer(short size) {
		bufSize = size;
		ramBuffer = true;
	}
    
    /** Specify the name of the buffer this query outputs results to -- other queries
	 may refer to this buffer name.  If share is true, associate this name with
	 this query (globally), so that other queries can reference the local schema.
	 */
    public void setBufferName(String name, boolean share) {
	//overwrite old values, if they exist...
	if (share) nameHashMap.put(name.toLowerCase(), this);
	this.queryName = name;
	hasName = true;
    }
	
    public void setInputBufferName(String name) {
		this.inputBufferName = name;
		hasInputBuf = true;
    }
	
    /** Given a buffer name, lookup the query which corresponds to it. */
    public static TinyDBQuery getQueryForBufName(String name) {
		return (TinyDBQuery)nameHashMap.get(name.toLowerCase());
    }
	
    public void setBufferCreateTable(boolean create) {
		createTable = create;
    }
	
    public boolean getBufferCreateTable() {
		return createTable;
    }
	
	
    /** Inactive queries have been "stopped" -- e.g. cancelled on the motes,
	 but we may still want to keep state about them so the can be
	 restarted at the previous epoch
	 */
    public boolean active() {
		return isRunning;
    }
	
    public void setActive(boolean active) {
		isRunning = active;
    }
	
    
    
    public byte qid,from_qid;
    private short epochDur;
	public short numEpochs;
	
	private boolean fromCatalogBuf = false;
	private byte catalogTableId;
	
    private static HashMap nameHashMap = new HashMap();
	
    private boolean ramBuffer = false;
    private short bufSize;
	
    private boolean isGrouped = false;
    private AggExpr groupExpr = null;
	
    private ArrayList fields;
    private ArrayList exprs;
    private int lastSelExpr = -1;
	
    private String sql = "";
	
    private boolean hasCmd = false;
    private String cmdName;
    private short paramVal = 0;
    private boolean hasParam = false;
    private String queryName = "";
    private String inputBufferName = "";
    private boolean hasName = false;
    private boolean hasInputBuf = false;
	
    private boolean hasEvent = false;
    private boolean createTable = false;
    
    private boolean dropTables = false;

    private String eventName;
    
    static final byte FIELD = 0;
    static final byte EXPR = 1;
    static final byte BUFFER = 2;
    static final byte EVENT = 3;
    static final byte N_EPOCHS = 4;
    static final byte DROP_TABLE = 5;

    static final byte ADD_MSG = 0;
    static final byte DEL_MSG = 1;
    static final byte MODIFY_MSG = 2;
    static final byte SET_RATE_MSG = 3;
	
    static final byte DEL_MSG_TTL = 3; //ttl on delete messages
    
    public static final byte NO_FROM_QUERY = (byte)0xFF;
	
    static final byte SEL_EXPR = 0;
    static final byte AGG_EXPR = 1;
    static final byte TEMPORAL_AGG_EXPR = 2;
	
    static final byte RADIO_BUFFER = 0;
    static final byte RAM_BUFFER = 1;
    static final byte EEPROM_BUFFER = 2;
    static final byte COMMAND_BUFFER = 3;
    static final byte ATTRLIST = 4;
    static final byte EVENTLIST = 5;
    static final byte COMMANDLIST = 6;
    static final byte QUERYLIST = 7;
    static final byte EVICT_OLDEST_POLICY = 0;

    static final short MS_PER_EPOCH_DUR_UNIT =10;
    
    public static final short kEPOCH_DUR_ONE_SHOT=(short)0x7FFF;
	
    private boolean isRunning = false; //have we seen any results for this query lately?
	
}