import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
@Override
public List<PartitionQuery<K, T>> getPartitions(Query<K, T> query){
  List<InputSplit> splits = null;
  List<PartitionQuery<K, T>> queries = null;
  try{
    splits = GoraMapReduceUtils.getSplits(getConf(), inputPath);
    queries = new ArrayList<>(splits.size());

    for(InputSplit split : splits) {
      queries.add(new FileSplitPartitionQuery<>(query, (FileSplit) split));
    }
  }catch(IOException ex){
    LOG.error(ex.getMessage(), ex);
  }
  return queries;
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
@Override
public List<PartitionQuery<K, T>> getPartitions(Query<K, T> query){
  List<InputSplit> splits = null;
  List<PartitionQuery<K, T>> queries = null;
  try{
    splits = GoraMapReduceUtils.getSplits(getConf(), inputPath);
    queries = new ArrayList<PartitionQuery<K,T>>(splits.size());

    for(InputSplit split : splits) {
      queries.add(new FileSplitPartitionQuery<K, T>(query, (FileSplit) split));
    }
  }catch(IOException ex){
    LOG.error(ex.getMessage());
    LOG.error(ex.getStackTrace().toString());
  }
  return queries;
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
@Override
public List<PartitionQuery<K, T>> getPartitions(Query<K, T> query)
    throws IOException {
  // TODO: implement this using Hadoop DB support

  ArrayList<PartitionQuery<K, T>> partitions = new ArrayList<>();
  PartitionQueryImpl<K, T> pqi = new PartitionQueryImpl<>(query);
  pqi.setConf(getConf());
  partitions.add(pqi);

  return partitions;
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
@Override
/**
 * Returns a single partition containing the original query
 */
public List<PartitionQuery<K, T>> getPartitions(Query<K, T> query){
  List<PartitionQuery<K, T>> list = new ArrayList<>();
  PartitionQueryImpl<K, T> pqi = new PartitionQueryImpl<>(query);
  pqi.setConf(getConf());
  list.add(pqi);
  return list;
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
@SuppressWarnings({ "rawtypes" })
private void setInputPath(PartitionQuery<K,T> partitionQuery
    , TaskAttemptContext context) throws IOException {
  //if the data store is file based
  if(partitionQuery instanceof FileSplitPartitionQuery) {
    FileSplit split = ((FileSplitPartitionQuery<K,T>)partitionQuery).getSplit();
    //set the input path to FileSplit's path.
    ((FileBackedDataStore)partitionQuery.getDataStore()).setInputPath(
        split.getPath().toString());
  }
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public RecordReader<K, T> createRecordReader(InputSplit split,
    TaskAttemptContext context) throws IOException, InterruptedException {
  PartitionQuery<K,T> partitionQuery = (PartitionQuery<K, T>)
    ((GoraInputSplit)split).getQuery();

  setInputPath(partitionQuery, context);
  return new GoraRecordReader<>(partitionQuery, context);
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
@Override
public List<PartitionQuery<String, MockPersistent>> getPartitions(
    Query<String, MockPersistent> query) throws IOException {

  ArrayList<PartitionQuery<String, MockPersistent>> list =
    new ArrayList<>();

  for(int i=0; i<NUM_PARTITIONS; i++) {
    list.add(new PartitionQueryImpl<>(query, LOCATIONS[i]));
  }

  return list;
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
/**
 * First, asserts that the attempt to obtain splits results in 
 * greater than 0 splits which can be used for computation.
 * We then check that the partition query (obtained by using the 
 * splits) has the same fields as we would expect by directly 
 * accessing the fields of an Employee object.
 * @throws IOException
 * @throws InterruptedException
 */
@Test
@SuppressWarnings("rawtypes")
public void testGetSplits() throws IOException, InterruptedException {
  List<InputSplit> splits = getInputSplits();

  assertTrue(splits.size() > 0);

  InputSplit split = splits.get(0);
  PartitionQuery query = ((GoraInputSplit)split).getQuery();
  assertTrue(Arrays.equals(getEmployeeFieldNames(), query.getFields()));
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
private List<PartitionQuery<String, MockPersistent>> 
  getPartitions() throws IOException {
  MockDataStore store = MockDataStore.get();
  MockQuery query = store.newQuery();

  List<PartitionQuery<String, MockPersistent>> partitions =
    store.getPartitions(query);
  return partitions;
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
@Test
public void testGetLocations() throws IOException {
  List<PartitionQuery<String, MockPersistent>> partitions = 
    getPartitions();

  int i=0;
  for(PartitionQuery<String, MockPersistent> partition : partitions) {
    GoraInputSplit split = new GoraInputSplit(conf, partition);
    assertEquals(split.getLocations().length, 1);
    assertEquals(split.getLocations()[0], MockDataStore.LOCATIONS[i++]);
  }
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
@Test
public void testReadWrite() throws Exception {
  
  List<PartitionQuery<String, MockPersistent>> partitions = 
    getPartitions();

  for(PartitionQuery<String, MockPersistent> partition : partitions) {
    GoraInputSplit split = new GoraInputSplit(conf, partition);
    TestWritable.testWritable(split);
  }
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
public static void testGetPartitions(DataStore<String, WebPage> store
    , Query<String, WebPage> query) throws Exception {
  List<PartitionQuery<String, WebPage>> partitions = store.getPartitions(query);

  assertNotNull(partitions);
  assertTrue(partitions.size() > 0);

  for(PartitionQuery<String, WebPage> partition:partitions) {
    assertNotNull(partition);
  }

  assertPartitions(store, query, partitions);
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
@Override
public List<PartitionQuery<K, T>> getPartitions(Query<K, T> query)
    throws IOException {
  // TODO GORA-298 Implement CassandraStore#getPartitions
  List<PartitionQuery<K,T>> partitions = new ArrayList<>();
  PartitionQueryImpl<K, T> pqi = new PartitionQueryImpl<>(query);
  pqi.setConf(getConf());
  partitions.add(pqi);
  return partitions;
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
/**
 * Partitions the given query and returns a list of PartitionQuerys, which
 * will execute on local data.
 */
@Override
public List<PartitionQuery<K, T>> getPartitions(final Query<K, T> query)
    throws IOException {
  // FIXME: for now, there is only one partition as we do not handle
  // MongoDB sharding configuration
  List<PartitionQuery<K, T>> partitions = new ArrayList<>();
  PartitionQueryImpl<K, T> partitionQuery = new PartitionQueryImpl<>(
      query);
  partitionQuery.setConf(getConf());
  partitions.add(partitionQuery);
  return partitions;
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
@SuppressWarnings({ "rawtypes" })
private void setInputPath(PartitionQuery<K, T> partitionQuery) throws IOException {
  //if the data store is file based
  if(partitionQuery instanceof FileSplitPartitionQuery) {
    FileSplit split = ((FileSplitPartitionQuery<K,T>)partitionQuery).getSplit();
    //set the input path to FileSplit's path.
    ((FileBackedDataStore)partitionQuery.getDataStore()).setInputPath(
        split.getPath().toString());
  }
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public RecordReader<K, T> createRecordReader(InputSplit split,
    TaskAttemptContext context) throws IOException, InterruptedException {
  PartitionQuery<K,T> partitionQuery = (PartitionQuery<K, T>)
    ((GoraInputSplit)split).getQuery();

  setInputPath(partitionQuery);
  return new GoraRecordReader<>(partitionQuery, context);
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public List<PartitionQuery<K, T>> getPartitions(Query<K, T> query) throws IOException {
  List<PartitionQuery<K, T>> partitions = new ArrayList<>();
  PartitionWSQueryImpl<K, T> pqi = new PartitionWSQueryImpl<>(query);
  pqi.setDataStore(this);
  partitions.add(pqi);
  return partitions;
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
@Override
public List<PartitionQuery<K, T>> getPartitions(Query<K, T> query) throws IOException {

  final List<PartitionQuery<K, T>> list = new ArrayList<>();
  final PartitionQueryImpl<K, T> pqi = new PartitionQueryImpl<>(query);

  pqi.setConf(getConf());
  list.add(pqi);
  return list;
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
/**
 * {@inheritDoc}
 * As the Aerospike does not support query key ranges as at the moment, only the single partition
 * is retrieved with this method.
 *
 * @param query the query to execute.
 * @return the list of partitions, one partion at the list as at the moment
 */
@Override
public List<PartitionQuery<K, T>> getPartitions(Query<K, T> query) throws IOException {
  List<PartitionQuery<K, T>> partitions = new ArrayList<>();
  PartitionQueryImpl<K, T> partitionQuery = new PartitionQueryImpl<>(
          query);
  partitionQuery.setConf(getConf());
  partitions.add(partitionQuery);
  return partitions;
}
 

import org.apache.gora.query.PartitionQuery; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public List<PartitionQuery<K, T>> getPartitions(Query<K, T> query) throws IOException {
  // TODO : Improve code on OrientDB clusters
  List<PartitionQuery<K, T>> partitions = new ArrayList<>();
  PartitionQueryImpl<K, T> partitionQuery = new PartitionQueryImpl<>(
          query);
  partitionQuery.setConf(this.getConf());
  partitions.add(partitionQuery);
  return partitions;
}