import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
/**
 * Returns a map with column name as key and column date type as value.
 *
 * The value might be as simple as "Boolean" or more complex like
 *  - "Set|Boolean"
 *  - "List|String"
 *  - "Map|String|Integer"
 *  these are cases when the data type is a container of primitive data types.
 *
 * @param tableName
 * @return
 * @throws DbException
 */
public Map<String, String> getColumnInfo(
                                          String tableName ) throws DbException {

    connect();

    ResultSet results = session.execute("SELECT * FROM " + this.dbName + "." + tableName + " LIMIT 1");

    Map<String, String> columnInfo = new HashMap<String, String>();
    for (Definition columnDefinition : results.getColumnDefinitions()) {
        DataType dataType = columnDefinition.getType();
        String dataTypeName = dataType.getName().name();
        if ("Set".equalsIgnoreCase(dataTypeName)) {
            dataTypeName = dataTypeName + "|" + dataType.getTypeArguments().get(0);
        } else if ("List".equalsIgnoreCase(dataTypeName)) {
            dataTypeName = dataTypeName + "|" + dataType.getTypeArguments().get(0);
        } else if ("Map".equalsIgnoreCase(dataTypeName)) {
            dataTypeName = dataTypeName + "|" + dataType.getTypeArguments().get(0) + "|"
                           + dataType.getTypeArguments().get(1);
        }
        columnInfo.put(columnDefinition.getName(), dataTypeName);
    }

    return columnInfo;
}
 

import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
public RawNumericMetric map(Row row) {
    RawNumericMetric metricRow =  new RawNumericMetric(row.getString(0), row.getString(1), row.getDate(2).getTime(), row.getInt(3));
    ColumnDefinitions columeDef = row.getColumnDefinitions();
    List<Definition> columeDefList = columeDef.asList();
    Map<String, String> tagMap = new HashMap<String, String>();
    for(Definition def: columeDefList){
        if(def.getName().startsWith("tag_")){
            tagMap.put(def.getName(), row.getString(def.getName()));
        }
    }
    
    if(tagMap.size()>0){
        metricRow.setTagMap(tagMap);
    }
    return metricRow;
}
 

import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
@Override
public String toString() {
    final ToStringHelper toStringHelper = MoreObjects.toStringHelper(this);
    
    for (Definition definition : getRow().getColumnDefinitions().asList()) {
        toStringHelper.add(definition.getName(), toString(definition.getName(), definition.getType()));
    }

    String s = "[" + result.getExecutionInfo().getQueriedHost() + "] "+  toStringHelper.toString();

    if (result.getExecutionInfo().getQueryTrace() != null) {
        StringBuilder sb = new StringBuilder("\n");
        for (Event event : result.getExecutionInfo().getQueryTrace().getEvents()) {
            sb.append(event.getSource() + " - " + event.getSourceElapsedMicros() + ": " + event.getDescription() +"\n");
        }
        s = s + sb.toString();
    }
    
    return s;
}
 

import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public static JSONArray marshallResultSet(ResultSet resultSet) {
    JSONArray resultJson = new JSONArray();
    for (Row row : resultSet){
        JSONArray rowJson = new JSONArray();
        resultJson.add(rowJson);
        Iterator<Definition> definitionIterator = row.getColumnDefinitions().iterator();
        while(definitionIterator.hasNext()){
            Definition definition = definitionIterator.next();
            logger.debug("Marshalling [" + definition.getName() + "] of type [" + definition.getType() + "]");
            if (definition.getType() == DataType.text()){     
                String value = row.getString(definition.getName());
                rowJson.add(value);
                logger.debug("Adding [" + value + "]");
            }
        }
    }
    return resultJson;
}
 

import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
private static Object getValueFromType(DataType type, Row row, Definition column, Class<?> columnType) {
	ByteBuffer rawValue = row.getBytesUnsafe(column.getName());
	if (null == rawValue) {
		return null;
	}			
	Object o = type.deserialize(rawValue);		
	
	if (Integer.class == o.getClass()) {
		return Long.valueOf((long)(Integer)o); // appengine smacks all Integers to Longs so do the same to remain compatibility
	} else if (String.class == o.getClass() && Key.class == columnType) {
		return Key.parse((String)o);
	} else if (o instanceof ByteBuffer) {
		try {
			BinarySerializer<Object> b = new BinarySerializer<Object>();
			return b.fromByteBuffer(rawValue);			
		} catch(Exception e) {	
			Log.e(LOG_TAG, e);
		}			
	}
	return o;
}
 

import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
@Override
public DbRecordValuesList[] select(
                                    DbQuery dbQuery,
                                    DbReturnModes dbReturnMode ) throws DbException {

    connect();

    ArrayList<DbRecordValuesList> dbRecords = new ArrayList<DbRecordValuesList>();

    String sqlQuery = dbQuery.getQuery();
    if (allowFiltering) {
        sqlQuery += " ALLOW FILTERING";
    }

    if (log.isDebugEnabled()) {
        log.debug(sqlQuery);
    }

    ResultSet results = session.execute(sqlQuery);

    int currentRow = 0;
    Iterator<Row> it = results.iterator();
    while (it.hasNext()) {
        Row row = it.next();

        currentRow++;
        if (log.isDebugEnabled()) {
            log.debug("Result row number: " + currentRow);
        }

        DbRecordValuesList recordList = new DbRecordValuesList();

        for (Definition columnDefinition : row.getColumnDefinitions()) {
            DbColumn dbColumn = new DbColumn(columnDefinition.getTable(), columnDefinition.getName());
            dbColumn.setColumnType(columnDefinition.getType().getName().toString());

            Object value = extractObjectFromResultSet(row, columnDefinition);

            DbRecordValue recordValue = new DbRecordValue(dbColumn, value);
            recordList.add(recordValue);
        }
        dbRecords.add(recordList);
    }

    return dbRecords.toArray(new DbRecordValuesList[]{});
}
 

import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
private static void syncQuorum(String key){
	logger.info("Performing sync operation---");
	String[] splitString = key.split("\\.");
	String keyspaceName = splitString[0];
	String tableName = splitString[1];
	String primaryKeyValue = splitString[2];
	
	//get the primary key d
	TableMetadata tableInfo = returnColumnMetadata(keyspaceName, tableName);
	String primaryKeyName = tableInfo.getPrimaryKey().get(0).getName();//we only support single primary key
	DataType primaryKeyType = tableInfo.getPrimaryKey().get(0).getType();
	String cqlFormattedPrimaryKeyValue = convertToCQLDataType(primaryKeyType, primaryKeyValue);
	
	//get the row of data from a quorum
	String selectQuery =  "SELECT *  FROM "+keyspaceName+"."+tableName+ " WHERE "+primaryKeyName+"="+cqlFormattedPrimaryKeyValue+";"; 
	ResultSet results = getDSHandle().executeCriticalGet(selectQuery);
	
	//write it back to a quorum
	Row row = results.one();
	ColumnDefinitions colInfo = row.getColumnDefinitions();
	int totalColumns = colInfo.size();
	int counter =1;
	String fieldValueString="";
	for (Definition definition : colInfo){
		String colName = definition.getName();
		if(colName.equals(primaryKeyName))
			continue; 
		DataType colType = definition.getType();
		Object valueObj = getDSHandle().getColValue(row, colName, colType);	
		String valueString = convertToCQLDataType(colType,valueObj);	
		fieldValueString = fieldValueString+ colName+"="+valueString;
		if(counter!=(totalColumns-1))
			fieldValueString = fieldValueString+",";
		counter = counter +1;
	}
	
	String updateQuery =  "UPDATE "+keyspaceName+"."+tableName+" SET "+fieldValueString+" WHERE "+primaryKeyName+"="+cqlFormattedPrimaryKeyValue+";";
	getDSHandle().executePut(updateQuery, "critical");
}
 

import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
public Map<String, HashMap<String, Object>> marshalData(ResultSet results){
	Map<String, HashMap<String, Object>> resultMap = new HashMap<String, HashMap<String,Object>>();
	int counter =0;
	for (Row row : results) {
		ColumnDefinitions colInfo = row.getColumnDefinitions();
		HashMap<String,Object> resultOutput = new HashMap<String, Object>();
		for (Definition definition : colInfo) {
			if(!definition.getName().equals("vector_ts"))
				resultOutput.put(definition.getName(), getColValue(row, definition.getName(), definition.getType()));
		}
		resultMap.put("row "+counter, resultOutput);
		counter++;
	}
	return resultMap;
}
 

import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
public List<Map<String, Object>> selectRows(final String tablename, Map<String, String> cols) {
	String ns = "";
	String tbl = tablename;
	int ix = tbl.indexOf('.');
	if (ix >= 0) {
		ns = tablename.substring(0, ix);
		tbl = tablename.substring(ix+1);
	}
	Select sel = QueryBuilder.select().all().from(ns, tbl);
	Statement stmt = sel;
	if (cols.size() == 1) {
		// only handles 1 WHERE value right now
		String k = cols.keySet().iterator().next();
		Clause eqclause = QueryBuilder.eq(k, cols.get(k));
		stmt = sel.where(eqclause);
	}
	ResultSet resultset = session.execute(stmt);
	List<Map<String, Object>> results = new ArrayList<Map<String,Object>>();
	for (Row row : resultset) {
		Map<String, Object> map = new HashMap<String, Object>();
		for (Definition definition : row.getColumnDefinitions()) {
			map.put(definition.getName(), readRow(row, definition.getName(), definition.getType()));
		}
		results.add(map);
	}
	return results;
}
 

import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
@Deprecated
	public List<Map<String, String>> OLDselectRows(String tablename, Map<String, String> cols) {
		String query = String.format("SELECT * FROM %s", tablename);
		if (cols.size() > 0) {
			// add WHERE clause
//			String[] parts = tablename.split("\\.");
//			KeyspaceMetadata ks = cluster.getMetadata().getKeyspace(parts[0]);
//			TableMetadata tableInfo = ks.getTable(parts[1]);
			String whereclause = " WHERE";
			String prefix = "";
			for (String key : cols.keySet()) {
				String val = cols.get(key);
//				DataType colType = tableInfo.getColumn(key).getType();
				whereclause = String.format("%s%s %s = '%s'", whereclause, prefix, key, val);
				prefix = " AND";
			}
			query += whereclause;
		}
		LOG.debug(query);
		ResultSet resultset = session.execute(query);
		List<Map<String, String>> results = new ArrayList<Map<String,String>>();
		for (Row row : resultset) {
			ColumnDefinitions colInfo = row.getColumnDefinitions();
			Map<String, String> map = new HashMap<String, String>();
			for (Definition definition : colInfo) {
			//	map.put(definition.getName(), (String)MusicDataStore.readRow(row, definition.getName(), definition.getType()));
			}
			results.add(map);
		}
		return results;
	}
 

import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
protected CassandraResultSet(BaseCassandraStatement statement, CassandraCqlStatement parsedStmt, ResultSet rs) {
    super(statement, parsedStmt);

    if (rs != null) {
        for (Definition def : rs.getColumnDefinitions()) {
            CassandraColumnDefinition d = new CassandraColumnDefinition(
                    def.getKeyspace(), def.getTable(), def.getName(), def
                    .getType().getName().toString(), false);
            metadata.addColumnDefinition(d);
        }
    }

    _resultSet = rs;
}
 

import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
@SuppressWarnings("rawtypes")
public int getColumnDisplaySize(int column) throws SQLException
      {
          //checkIndex(column);
          Definition col = null;
          if(currentRow!=null){
          	col = currentRow.getColumnDefinitions().asList().get(column-1);
          }else{
          	col = driverResultSet.getColumnDefinitions().asList().get(column-1);
          }
          try{
          	
          	int length = -1;
           AbstractJdbcType jtype = TypesMap.getTypeForComparator(col.getType().toString());
           if (jtype instanceof JdbcBytes) length = Integer.MAX_VALUE / 2;
           if (jtype instanceof JdbcAscii || jtype instanceof JdbcUTF8) length = Integer.MAX_VALUE;
           if (jtype instanceof JdbcUUID) length = 36;
           if (jtype instanceof JdbcInt32) length = 4;
           if (jtype instanceof JdbcLong) length = 8;
          	// String stringValue = getObject(column).toString();
          	//return (stringValue == null ? -1 : stringValue.length());
           
           return length;
          }catch(Exception e){
          	return -1;
          }
          //return -1;
      }
 

import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
private List<String> getColumnName(final Session session,
		final String eventType) {
	List<String> currentlist = null;
	if (eventType.equals("ObjectEvent")) {
		currentlist = ObjectEventColumnName;
	} else if (eventType.equals("QuantityEvent")) {
		currentlist = QuantityEventColumnName;
	} else if (eventType.equals("TransactionEvent")) {
		currentlist = TransactionEventColumnName;
	} else if (eventType.equals("AggregationEvent")) {
		currentlist = AggregationEventColumnName;
	}
	if (currentlist == null) {
		LOG.debug(eventType + " column is null");
		ResultSet rs = session.execute("SELECT * FROM " + eventType
				+ " WHERE epc = 'TESTKEY';");
		List<Definition> cf = rs.getColumnDefinitions().asList();
		currentlist = new ArrayList<String>(cf.size());
		for (Definition def : cf) {
			currentlist.add(def.getName().toLowerCase());
		}
	}

	if (eventType.equals("ObjectEvent")) {
		ObjectEventColumnName = currentlist;
	} else if (eventType.equals("QuantityEvent")) {
		QuantityEventColumnName = currentlist;
	} else if (eventType.equals("TransactionEvent")) {
		TransactionEventColumnName = currentlist;
	} else if (eventType.equals("AggregationEvent")) {
		AggregationEventColumnName = currentlist;
	}

	return currentlist;
}
 

import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
private String getKeyColumn() {
	// TODO: Ideally we'd use ColumnDefinitions.contains...but it's throwing out of bounds exception
	for (Definition definition : rs.getColumnDefinitions()) {
		for (String keyColumn : new String[] { "KEY", "key" }) {
			if (keyColumn.equals(definition.getName())) {
				return keyColumn;
			}
		}
	}

	throw new RuntimeException("Can't determine key column from metadata");
}
 

import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
/**
 * Instantiates a new data store row from a row @see <a href="http://www.datastax.com/doc-source/developer/java-apidocs/com/datastax/driver/core/class-use/Row.html">Row</a>
 *
 * @param row the row
 */
public DataStoreRow(Row row) {
	for (Definition column : row.getColumnDefinitions()) {
		DataStoreColumn c = DataStoreColumn.create(column.getName());
		c.setValue(DataStoreRow.getValueFromType(column.getType(), row, column, c.getType()));
		if (null != c.getValue()) {
			this.addColumn(c);
		}
	}
}
 

import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
private void readExtensionsFromResult(final Row row, final List<Object> extensions) {
    LOG.debug("Read extensions");
    Map<String, String> prefixMap = row.getMap("prefix", String.class, String.class);
    List <Definition> cf = row.getColumnDefinitions().asList();
    List<String> prefixlist = new ArrayList<String>(prefixMap.keySet());                    
    for (Definition def : cf)
        for (String prefix : prefixlist) {
            LOG.debug("prefix: " + prefix);
            String columnname = def.getName().toLowerCase();
            LOG.debug("column name: " + columnname);
            //convert prefix name to column name style, e.g. change . to dot_
            String prefixtemp = prefixMap.get(prefix).replace("#", "_").replace("://", "virgule_").replace(":","collon_").replace("/","slash_").replace(".", "dot_");
            if (columnname.startsWith(prefixtemp.toLowerCase())) {
                String value = "";
                try {
                    value = row.getString(columnname);
                } catch (InvalidTypeException e) {
                    try {
                        value = row.getDate(columnname).toString();
                    } catch (InvalidTypeException e2) {
                        try {
                            value = String.valueOf(row.getInt(columnname));
                        } catch (InvalidTypeException e3) {
                            value = String.valueOf(row.getFloat(columnname));
                        }
                    }
                }
                String[] parts = columnname.split("_");
                String localPart = parts[parts.length - 1];
                String pre_namespace = columnname.substring(0, columnname.length()-localPart.length()-1);
                String namespace = pre_namespace.replace("slash_","/").replace("virgule_", "://").replace("collon_",":").replace("dot_", ".");
                LOG.debug("Value: " + value);
                LOG.debug("localPart: " + localPart);
                LOG.debug("namespace: " + namespace);
                LOG.debug("prefix: " + prefix);
                if (value != null) {
                    JAXBElement<String> elem = new JAXBElement<String>(new QName(namespace, localPart, prefix), String.class,
                            value);
                    extensions.add(elem);
                }

            }
    }
}
 

import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
@Override
public Hit get(Entity entity) throws Exception {

	String cql = "SELECT * FROM "+entity.getName()+"_"+entity.getMappingHash()+" WHERE rendezvous_id = '" + entity.getId() + "'";

	logger.info(cql);
	
	ResultSet result = session.execute(cql);
	
	List<Value> values = new ArrayList<Value>();
	
	for(Row row : result.all()){
		for(Definition definition : row.getColumnDefinitions().asList()){
			
			String field = definition.getName();
			
			if(!field.equals("rendezvous_id")){
				values.add(new Value(entity.getName(),field,row.getString(field),new Type(DataType.STRING.toString()), this));
				
				logger.info("Added a value for the field "+field);
			}
		}			
	}
	
	Hit hit = new Hit();
	hit.setRelevance(1);
	hit.setValues(values);
	
	return hit;
}
 

import com.datastax.driver.core.ColumnDefinitions.Definition; //导入依赖的package包/类
@Override
public List<Hit> query(Entity entity, Set<Value> queryValues) throws Exception {
	
	logger.debug("Searching from Cassandra");
	
	if(queryValues==null || queryValues.isEmpty() || queryValues.size() == 0) throw new Exception("No parameter for the query");
	
	int index = 0;
	String cql = "SELECT * FROM "+entity.getName()+"_"+entity.getMappingHash()+" WHERE ";

	for(Value value:queryValues){

		++index;
		
		cql+=value.getField().getFieldName()+"='"+value.getValue()+"'";
		if(index<queryValues.size()) cql+=" AND ";
		else cql+=" ALLOW FILTERING;";			
	}
			
	logger.info(cql);
	
	ResultSet result = session.execute(cql);
	
	List<Hit> hits = new ArrayList<Hit>();
	
	for(Row row : result.all()){

		Hit hit = new Hit();
		hit.setRelevance(1);
		
		List<Value> values = new ArrayList<Value>();
		
		for(Definition definition : row.getColumnDefinitions().asList()){
			
			String field = definition.getName();
			
			if(!field.equals("rendezvous_id")){
				values.add(new Value(entity.getName(),field,row.getString(field),new Type(DataType.STRING.toString()), this));
			}
		}
		
		hit.setValues(values);
		hits.add(hit);
	}		
	
	logger.info("<"+hits.size()+"> hits where found in Cassandra");
	
	return hits;
}