import com.aerospike.client.query.RecordSet; //导入依赖的package包/类
@Override
public long getCommittedWindowId(String appId, int operatorId)
{
  try {
    lastWindowFetchCommand.setFilters(Filter.equal(metaTableOperatorIdColumn, operatorId));
    lastWindowFetchCommand.setFilters(Filter.equal(metaTableAppIdColumn, appId));
    long lastWindow = -1;
    RecordSet recordSet = client.query(null, lastWindowFetchCommand);
    while (recordSet.next()) {
      lastWindow = Long.parseLong(recordSet.getRecord().getValue(metaTableWindowColumn).toString());
    }
    return lastWindow;
  } catch (AerospikeException ex) {
    throw new RuntimeException(ex);
  }
}
 

import com.aerospike.client.query.RecordSet; //导入依赖的package包/类
/**
 * This executes the query to retrieve result from database.
 * It then converts each row into tuple and emit that into output port.
 */
@Override
public void emitTuples()
{
  Statement query = queryToRetrieveData();
  logger.debug(String.format("select statement: %s", query.toString()));
  RecordSet rs;
  try {
    rs = store.getClient().query(null, query);
    while (rs.next()) {
      Record rec = rs.getRecord();
      T tuple = getTuple(rec);
      outputPort.emit(tuple);
    }
  } catch (Exception ex) {
    store.disconnect();
    DTThrowable.rethrow(ex);
  }
}
 

import com.aerospike.client.query.RecordSet; //导入依赖的package包/类
private void queryCircle(Location droneLocation, double radius, Predicate<? super Job> action) {
    Statement stmt = new Statement();
    stmt.setNamespace(database.namespace);
    stmt.setSetName(setName);
    String binName = Job.State.Waiting.name();
//    stmt.setBinNames(); // all
    stmt.setFilters(Filter.geoWithinRadius(binName, droneLocation.x, droneLocation.y, radius));

    QueryPolicy policy = new QueryPolicy();
    ++Metering.jobQueryWithinRadius;
    try (RecordSet rs = database.client.query(policy, stmt)) {
      while (rs.next()) {
        ++Metering.jobRadiusResults;
        Job job = get(rs.getKey(), rs.getRecord());
        if (!action.test(job)) {
          break;
        }
      }
    }
  }
 

import com.aerospike.client.query.RecordSet; //导入依赖的package包/类
@Override
  public void foreach(Job.State state, Predicate<? super Job> action) {
    if (true) {
      throw new Error("not tested");
    } else {
      String[] bins = { state.name() };
      Statement stmt = new Statement();
      stmt.setNamespace(database.namespace);
      stmt.setSetName(setName);
      stmt.setIndexName("username_index");
      stmt.setBinNames(bins);
//    stmt.setFilters(Filter.equal(state.name(), ___));

      RecordSet rs = database.client.query(null, stmt);
      while (rs.next()) {
        Record r = rs.getRecord();
        Job job = new Job(this);
        job.setLocation((Location) r.getValue("location"));
        if (!action.test(job)) {
          break;
        }
      }
    }
  }
 

import com.aerospike.client.query.RecordSet; //导入依赖的package包/类
/**
 * Select records filtered by Qualifiers
 *
 * @param stmt	   A Statement object containing Namespace, Set and the Bins to be returned.
 * @param sortMap	<STRONG>NOT IMPLEMENTED</STRONG>
 * @param qualifiers Zero or more Qualifiers for the update query
 * @return A KeyRecordIterator to iterate over the results
 */
public KeyRecordIterator select(Statement stmt, Map<String, String> sortMap, Qualifier... qualifiers) {
	KeyRecordIterator results = null;

	if (qualifiers != null && qualifiers.length > 0) {
		Map<String, Object> originArgs = new HashMap<String, Object>();
		originArgs.put("includeAllFields", 1);
		String filterFuncStr = buildFilterFunction(qualifiers);
		originArgs.put("filterFuncStr", filterFuncStr);
		String sortFuncStr = buildSortFunction(sortMap);
		originArgs.put("sortFuncStr", sortFuncStr);
		stmt.setAggregateFunction(this.getClass().getClassLoader(), AS_UTILITY_PATH, QUERY_MODULE, "select_records", Value.get(originArgs));
		ResultSet resultSet = this.client.queryAggregate(queryPolicy, stmt);
		results = new KeyRecordIterator(stmt.getNamespace(), resultSet);
	} else {
		RecordSet recordSet = this.client.query(queryPolicy, stmt);
		results = new KeyRecordIterator(stmt.getNamespace(), recordSet);
	}
	return results;
}
 

import com.aerospike.client.query.RecordSet; //导入依赖的package包/类
/**
 * Creates entities from the given {@link RecordSet} for the passed class.
 * 
 * @param record
 * @param clazz
 * @return
 */
@PerformanceMonitor
public <T extends Serializable> List<T> reverseMap(RecordSet recordSet, Class clazz) {
	List<T> values = new ArrayList<T>();
	try {
		while (recordSet != null && recordSet.next()) {
			T value = (T) reverseMap(recordSet.getRecord(), clazz);
			values.add(value);
		}
	} finally {
		recordSet.close();
	}
	if (values.isEmpty()) {
		return Collections.EMPTY_LIST;
	}
	return values;
}
 

import com.aerospike.client.query.RecordSet; //导入依赖的package包/类
public KeyRecordIterator(String namespace, RecordSet recordSet) {
	this(namespace);
	this.recordSet = recordSet;
	this.recordSetIterator = recordSet.iterator();
}
 

import com.aerospike.client.query.RecordSet; //导入依赖的package包/类
public void run() {
    try {
        AerospikeClient client =
            AerospikeClientSingleton.getInstance(new ClientPolicy(),
                                                 host, port);
        log.info(String.format("queryNode %s:%d %s:%s:%s[%d:%d]",
                               host, port, namespace, setName,
                               numrangeBin, numrangeBegin,
                               numrangeEnd));
        Statement stmt = new Statement();
        stmt.setNamespace(namespace);
        stmt.setSetName(setName);
        stmt.setFilters(Filter.range(numrangeBin,
                                     numrangeBegin,
                                     numrangeEnd));
        if (binNames != null)
            stmt.setBinNames(binNames);
        QueryPolicy queryPolicy = new QueryPolicy();
        RecordSet rs = client.queryNode(queryPolicy,
                                        stmt,
                                        client.getNode(node));
        isRunning = true;
        try {
            log.info("query starting");
            while (rs.next()) {
                Key key = rs.getKey();
                Record record = rs.getRecord();
                queue.put(new KeyRecPair(new AerospikeKey(key),
                                         new AerospikeRecord(record)));
            }
        }
        finally {
            rs.close();
            isFinished = true;
            log.info("query finished");
        }
    }
    catch (Exception ex) {
        isError = true;
        return;
    }
}
 

import com.aerospike.client.query.RecordSet; //导入依赖的package包/类
@Test
public void testDataSet() {

    DeepSparkContext context = new DeepSparkContext("local", "deepSparkContextTest");
    try {

        Statement stmnt = new Statement();
        stmnt.setNamespace(AerospikeJavaRDDFT.NAMESPACE_ENTITY);
        stmnt.setSetName("input");

        stmnt.setFilters(Filter.equal("id", "messageTest"));

        RecordSet recordSet = AerospikeJavaRDDFT.aerospike.query(null, stmnt);
        Record record = null;
        while (recordSet.next()) {
            record = recordSet.getRecord();
        }
        AerospikeRecord aerospikeRecord = new AerospikeRecord(record);

        Map<String, Object> bins = aerospikeRecord.bins;

        ExtractorConfig<MessageTestEntity> inputConfigEntity = new ExtractorConfig(MessageTestEntity.class);
        inputConfigEntity.putValue(ExtractorConstants.HOST, AerospikeJavaRDDFT.HOST)
                .putValue(ExtractorConstants.PORT, AerospikeJavaRDDFT.PORT)
                .putValue(ExtractorConstants.NAMESPACE, AerospikeJavaRDDFT.NAMESPACE_ENTITY)
                .putValue(ExtractorConstants.SET, "input");
        inputConfigEntity.setExtractorImplClass(AerospikeEntityExtractor.class);

        RDD<MessageTestEntity> inputRDDEntity = context.createRDD(inputConfigEntity);

        //Import dataSet was OK and we could read it
        assertEquals(1, inputRDDEntity.count(), "There should be 1 read entity.");

        List<MessageTestEntity> messages = inputRDDEntity.toJavaRDD().collect();

        MessageTestEntity message = messages.get(0);

        assertEquals((String) bins.get("id"), message.getId(), "Read message id should be the same as expected.");
        assertEquals((String) bins.get("message"), message.getMessage(),
                "Read message text should be the same as expected.");
    } finally {
        context.stop();
    }
}
 

import com.aerospike.client.query.RecordSet; //导入依赖的package包/类
/**
 * {@inheritDoc}
 *
 * @param query the query to execute.
 * @return the query result
 */
@Override
public Result<K, T> execute(Query<K, T> query) {

  List<AerospikeResultRecord> resultRecords = new ArrayList<>();
  String namespace = aerospikeParameters.getAerospikeMapping().getNamespace();
  String set = aerospikeParameters.getAerospikeMapping().getSet();

  // Query execution without any keys
  if (query.getStartKey() == null && query.getEndKey() == null) {

    try (RecordSet recordSet = aerospikeClient.query(null, getStatement(namespace, set))) {
      while (recordSet.next()) {
        AerospikeResultRecord aerospikeRecord = new AerospikeResultRecord(recordSet.getKey(),
                recordSet.getRecord());
        resultRecords.add(aerospikeRecord);
      }
    }
  }

  // Query execution for single key
  else if (query.getKey() != null) {
    Key key = getAerospikeKey(query.getKey());
    Record record = aerospikeClient.get(null, key);
    if (record != null) {
      resultRecords.add(new AerospikeResultRecord(key, record));
    }
  }

  // Query execution for key ranges
  // ToDo: Implement Query execution for key ranges
  //    else if (query.getStartKey() != null && query.getEndKey() != null) {
  //
  //      // the key range filtering at the gora side, which is not a better solution
  //      String lowerBound  = query.getStartKey().toString();
  //      String upperBound  = query.getEndKey().toString();
  //
  //      try (RecordSet recordSet = aerospikeClient.query(null, getStatement(namespace, set))) {
  //        while (recordSet.next()) {
  //          Key key = recordSet.getKey();
  //          Record record = recordSet.getRecord();
  //
  //          String input = key.userKey.toString();
  //          boolean isSpecifiedRange =  input.compareToIgnoreCase(lowerBound) >= 0 && input
  //                  .compareToIgnoreCase(upperBound) <= 0;
  //
  //          if (isSpecifiedRange) {
  //            AerospikeResultRecord aerospikeRecord = new AerospikeResultRecord(key, record);
  //            resultRecords.add(aerospikeRecord);
  //          }
  //
  //        }
  //      }
  //    }
  return new AerospikeQueryResult<>(this, query, resultRecords, getFieldsToQuery(null));
}