import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
/**
 * statements where the datetime is exactly the same as the queryInstant.
 */
@Override
public CloseableIteration<Statement, QueryEvaluationException> queryInstantEqualsInstant(
        final TemporalInstant queryInstant, final StatementConstraints constraints)
        throws QueryEvaluationException {
        // get rows where the repository time is equal to the given time in queryInstant.
        final Query query = new Query() {
            @Override
            public Range getRange(final KeyParts keyParts) {
                //System.out.println("Scanning queryInstantEqualsInstant: prefix:" + KeyParts.toHumanString(keyParts.getQueryKey()));
                return Range.prefix(keyParts.getQueryKey()); // <-- specific logic
            }
        };
        final ScannerBase scanner = query.doQuery(queryInstant, constraints);
        // TODO currently context constraints are filtered on the client.
        return getContextIteratorWrapper(scanner, constraints.getContext());
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
/**
 * get statements where the db row ID is BEFORE the given queryInstant.
 */
@Override
public CloseableIteration<Statement, QueryEvaluationException> queryInstantBeforeInstant(
        final TemporalInstant queryInstant, final StatementConstraints constraints)
        throws QueryEvaluationException {
    // get rows where the repository time is before the given time.
    final Query query = new Query() {
        @Override
        public Range getRange(final KeyParts keyParts) {
            Text start= null;
            if (keyParts.constraintPrefix != null ) {
                start = keyParts.constraintPrefix;   // <-- start specific logic
            } else {
                start = new Text(KeyParts.HASH_PREFIX_FOLLOWING);
            }
            final Text endAt = keyParts.getQueryKey();                   // <-- end specific logic
            //System.out.println("Scanning queryInstantBeforeInstant: from:" + KeyParts.toHumanString(start) + " up to:" + KeyParts.toHumanString(endAt));
            return new Range(start, true, endAt, false);
        }
    };
    final ScannerBase scanner = query.doQuery(queryInstant, constraints);
    return getContextIteratorWrapper(scanner, constraints.getContext());
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
/**
 * get statements where the date object is after the given queryInstant.
 */
@Override
public CloseableIteration<Statement, QueryEvaluationException> queryInstantAfterInstant(
        final TemporalInstant queryInstant, final StatementConstraints constraints)
        throws QueryEvaluationException {
    final Query query = new Query() {
        @Override
        public Range getRange(final KeyParts keyParts) {
            final Text start = Range.followingPrefix(keyParts.getQueryKey());  // <-- specific logic
            Text endAt = null;  // no constraints                            // <-- specific logic
            if (keyParts.constraintPrefix != null ) {
                endAt = Range.followingPrefix(keyParts.constraintPrefix);
            }
            //System.out.println("Scanning queryInstantAfterInstant from after:" + KeyParts.toHumanString(start) + " up to:" + KeyParts.toHumanString(endAt));
            return new Range(start, true, endAt, false);
        }
    };
    final ScannerBase scanner = query.doQuery(queryInstant, constraints);
    return getContextIteratorWrapper(scanner, constraints.getContext());
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
/**
 * Get instances inside a given interval.
 * Returns after interval's beginning time, and before ending time,
 * exclusive (don't match the beginning and ending).
 */
@Override
public CloseableIteration<Statement, QueryEvaluationException> queryInstantInsideInterval(
        final TemporalInterval queryInterval, final StatementConstraints constraints)
        throws QueryEvaluationException {
    // get rows where the time is after the given interval's beginning time and before the ending time.
    final TemporalInterval theQueryInterval = queryInterval;
    final Query query = new Query() {
        private final TemporalInterval queryInterval = theQueryInterval;
        @Override
        public Range getRange(final KeyParts keyParts) {
            final Text start = Range.followingPrefix(new Text(keyParts.getQueryKey(queryInterval.getHasBeginning())));
            final Text endAt = new Text(keyParts.getQueryKey(queryInterval.getHasEnd())); // <-- end specific logic
            //System.out.println("Scanning queryInstantInsideInterval: from excluding:" + KeyParts.toHumanString(start) + " up to:" + KeyParts.toHumanString(endAt));
            return new Range(start, false, endAt, false);
        }
    };
    final ScannerBase scanner = query.doQuery(queryInterval.getHasBeginning(), constraints);
    return getContextIteratorWrapper(scanner, constraints.getContext());
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
protected void addIndexFilterToIterator(
		final ScannerBase scanner ) {
	final List<MultiDimensionalCoordinateRangesArray> coords = base.getCoordinateRanges();
	if (!coords.isEmpty()) {
		final IteratorSetting iteratorSetting = new IteratorSetting(
				NumericIndexStrategyFilterIterator.IDX_FILTER_ITERATOR_PRIORITY,
				NumericIndexStrategyFilterIterator.IDX_FILTER_ITERATOR_NAME,
				NumericIndexStrategyFilterIterator.class);

		iteratorSetting.addOption(
				NumericIndexStrategyFilterIterator.INDEX_STRATEGY_KEY,
				ByteArrayUtils.byteArrayToString(AccumuloUtils.toBinary(index.getIndexStrategy())));

		iteratorSetting.addOption(
				NumericIndexStrategyFilterIterator.COORDINATE_RANGE_KEY,
				ByteArrayUtils.byteArrayToString(new ArrayOfArrays(
						coords.toArray(new MultiDimensionalCoordinateRangesArray[] {})).toBinary()));
		scanner.addScanIterator(iteratorSetting);
	}
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
protected void addFieldSubsettingToIterator(
		final ScannerBase scanner ) {
	if ((fieldIdsAdapterPair != null) && !isAggregation()) {
		final List<String> fieldIds = fieldIdsAdapterPair.getLeft();
		final DataAdapter<?> associatedAdapter = fieldIdsAdapterPair.getRight();
		if ((fieldIds != null) && (!fieldIds.isEmpty()) && (associatedAdapter != null)) {
			final IteratorSetting iteratorSetting = AttributeSubsettingIterator.getIteratorSetting();

			AttributeSubsettingIterator.setFieldIds(
					iteratorSetting,
					associatedAdapter,
					fieldIds,
					index.getIndexModel());

			iteratorSetting.addOption(
					AttributeSubsettingIterator.WHOLE_ROW_ENCODED_KEY,
					Boolean.toString(useWholeRowIterator()));
			scanner.addScanIterator(iteratorSetting);
		}
	}
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
public CloseableIterator<T> query(
		final AccumuloOperations accumuloOperations,
		final double[] maxResolutionSubsamplingPerDimension,
		final AdapterStore adapterStore ) {
	final ScannerBase scanner = getScanner(
			accumuloOperations,
			maxResolutionSubsamplingPerDimension,
			getScannerLimit());
	if (scanner == null) {
		LOGGER.error("Unable to get a new scanner instance, getScanner returned null");
		return null;
	}
	addScanIteratorSettings(scanner);
	return initCloseableIterator(
			scanner,
			initIterator(
					adapterStore,
					scanner));
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
/**
 * Converts a {@link ScannerBase} into a {@link CloseableIterable}
 */
public static CloseableIterable<Entry<Key, Value>> closeableIterable(final ScannerBase scanner) {
    checkNotNull(scanner);
    return new FluentCloseableIterable<Entry<Key, Value>>() {
        @Override
        protected void doClose() throws IOException {
            if (scanner instanceof BatchScanner)
                ((BatchScanner) scanner).close();
        }

        @Override
        protected Iterator<Entry<Key, Value>> retrieveIterator() {
            return scanner.iterator();
        }
    };
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
protected ScannerBase metricScanner(AccumuloFeatureConfig xform, Date start, Date end, String group, Iterable<String> types, String name, TimeUnit timeUnit, Auths auths) {
    checkNotNull(xform);

    try {
        group = defaultString(group);
        timeUnit = (timeUnit == null ? TimeUnit.MINUTES : timeUnit);

        BatchScanner scanner = connector.createBatchScanner(tableName + REVERSE_SUFFIX, auths.getAuths(), config.getMaxQueryThreads());

        Collection<Range> typeRanges = new ArrayList();
        for (String type : types)
            typeRanges.add(buildRange(type, start, end, timeUnit));

        scanner.setRanges(typeRanges);
        scanner.fetchColumn(new Text(combine(timeUnit.toString(), xform.featureName())), new Text(combine(group, name)));

        return scanner;

    } catch (TableNotFoundException e) {
        throw new RuntimeException(e);
    }
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
private Map<String, byte[]> streamingPropertyValueTableDatas(List<String> dataRowKeys) {
    try {
        if (dataRowKeys.size() == 0) {
            return Collections.emptyMap();
        }

        List<org.apache.accumulo.core.data.Range> ranges = dataRowKeys.stream()
                .map(RangeUtils::createRangeFromString)
                .collect(Collectors.toList());

        final long timerStartTime = System.currentTimeMillis();
        ScannerBase scanner = graph.createBatchScanner(graph.getDataTableName(), ranges, new org.apache.accumulo.core.security.Authorizations());

        graph.getGraphLogger().logStartIterator(scanner);
        Span trace = Trace.start("streamingPropertyValueTableData");
        trace.data("dataRowKeyCount", Integer.toString(dataRowKeys.size()));
        try {
            Map<String, byte[]> results = new HashMap<>();
            for (Map.Entry<Key, Value> col : scanner) {
                results.put(col.getKey().getRow().toString(), col.getValue().get());
            }
            return results;
        } finally {
            scanner.close();
            trace.stop();
            graph.getGraphLogger().logEndIterator(System.currentTimeMillis() - timerStartTime);
        }
    } catch (Exception ex) {
        throw new VertexiumException(ex);
    }
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
private ScannerBase getScanner() {
    if (scanner != null) {
        return scanner;
    }
    ArrayList<Range> ranges = Lists.newArrayList(RangeUtils.createRangeFromString(dataRowKey));

    timerStartTime = System.currentTimeMillis();
    try {
        scanner = graph.createBatchScanner(graph.getDataTableName(), ranges, new org.apache.accumulo.core.security.Authorizations());
    } catch (TableNotFoundException ex) {
        throw new VertexiumException("Could not create scanner", ex);
    }

    IteratorSetting iteratorSetting = new IteratorSetting(
            80,
            TimestampFilter.class.getSimpleName(),
            TimestampFilter.class
    );
    TimestampFilter.setStart(iteratorSetting, timestamp, true);
    TimestampFilter.setEnd(iteratorSetting, timestamp, true);
    scanner.addScanIterator(iteratorSetting);

    graph.getGraphLogger().logStartIterator(scanner);
    trace = Trace.start("streamingPropertyValueTableData");
    trace.data("dataRowKeyCount", Integer.toString(1));
    return scanner;
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
public ScannerBase doQuery(final TemporalInstant queryInstant, final StatementConstraints constraints) throws QueryEvaluationException {
    // key is contraintPrefix + time, or just time.
    // Any constraints handled here, if the constraints are empty, the
    // thisKeyParts.contraintPrefix will be null.
    final List<KeyParts> keyParts = KeyParts.keyPartsForQuery(queryInstant, constraints);
    ScannerBase scanner = null;
    if (keyParts.size() > 1) {
        scanner = getBatchScanner();
    } else {
        scanner = getScanner();
    }

    final Collection<Range> ranges = new HashSet<Range>();
    KeyParts lastKeyParts = null;
    Range range = null;
    for (final KeyParts thisKeyParts : keyParts) {
        range = getRange(thisKeyParts);
        ranges.add(range);
        lastKeyParts = thisKeyParts;
    }
    if (lastKeyParts == null || scanner == null) {
        throw new NullPointerException("lastkeyParts or scanner is null, impossible! keyParts.size()= " + keyParts.size() + " scanner= " + scanner);
    }
    //System.out.println("Scanning columns, cf:" + lastKeyParts.cf + "CQ:" + lastKeyParts.cq);
    scanner.fetchColumn(new Text(lastKeyParts.cf), new Text(lastKeyParts.cq));
    if (scanner instanceof BatchScanner) {
        ((BatchScanner) scanner).setRanges(ranges);
    } else if (range != null) {
        ((Scanner) scanner).setRange(range);
    }
    return scanner;
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
/**
 * An iteration wrapper for a loaded scanner that is returned for queries above.
 * Currently, this temporal index supports contexts only on the client, using this filter.
 *
 * @param scanner  the results to iterate, then close.
 * @param constraints  limit statements returned by next() to those matching the constraints.
 * @return an anonymous object that will iterate the resulting statements from a given scanner.
 * @throws QueryEvaluationException
 */
private static CloseableIteration<Statement, QueryEvaluationException> getContextIteratorWrapper(final ScannerBase scanner, final Resource context) {
    if (context==null) {
        return getIteratorWrapper(scanner);
    }
    return new ConstrainedIteratorWrapper(scanner) {
        @Override
        public boolean allowedBy(final Statement statement) {
            return allowedByContext(statement,  context);
        }
    };
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
@Override
protected Iterator initIterator(
		final AdapterStore adapterStore,
		final ScannerBase scanner ) {
	if (isAggregation()) {
		// aggregate the stats to a single value here

		try {
			final Iterator<Entry<Key, Value>> it = scanner.iterator();
			Mergeable mergedAggregationResult = null;
			if (!it.hasNext()) {
				return Iterators.emptyIterator();
			}
			else {
				while (it.hasNext()) {
					final Entry<Key, Value> input = it.next();
					if (input.getValue() != null) {
						if (mergedAggregationResult == null) {
							mergedAggregationResult = (Mergeable) AccumuloUtils.fromBinary(input.getValue().get());
						}
						else {
							mergedAggregationResult.merge((Mergeable) AccumuloUtils.fromBinary(input
									.getValue()
									.get()));
						}
					}
				}
			}
			return Iterators.singletonIterator(mergedAggregationResult);
		}
		finally {
			scanner.close();
		}
	}
	else {
		return super.initIterator(
				adapterStore,
				scanner);
	}
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
@Override
protected CloseableIterator<T> initCloseableIterator(
		ScannerBase scanner,
		Iterator it ) {
	return new CloseableIteratorWrapper(
			new Closeable() {
				boolean closed = false;

				@Override
				public void close()
						throws IOException {
					if (!closed) {
						if (scanner instanceof BatchDeleter) {
							try {
								((BatchDeleter) scanner).delete();
							}
							catch (MutationsRejectedException | TableNotFoundException e) {
								LOGGER.warn(
										"Unable to delete rows by query constraints",
										e);
							}
						}
						scanner.close();
					}
					closed = true;
				}
			},
			it);
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
@Override
protected ScannerBase createScanner(
		AccumuloOperations accumuloOperations,
		String tableName,
		boolean batchScanner,
		String... authorizations )
		throws TableNotFoundException {
	BatchDeleter deleter = accumuloOperations.createBatchDeleter(
			tableName,
			authorizations);
	deleter.removeScanIterator(BatchDeleter.class.getName() + ".NOVALUE");
	return deleter;
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
protected ScannerBase createScanner(
		final AccumuloOperations accumuloOperations,
		String tableName,
		boolean batchScanner,
		String... authorizations )
		throws TableNotFoundException {
	if (batchScanner) {
		return accumuloOperations.createBatchScanner(
				tableName,
				authorizations);
	}
	return accumuloOperations.createScanner(
			tableName,
			authorizations);
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
@Override
protected ScannerBase getScanner(
		final AccumuloOperations accumuloOperations,
		final double[] maxResolutionSubsamplingPerDimension,
		final Integer limit ) {
	final String tableName = index.getId().getString();
	Scanner scanner;
	try {
		scanner = accumuloOperations.createScanner(
				tableName,
				getAdditionalAuthorizations());
		scanner.setRange(accumuloRange);
		if ((adapterIds != null) && !adapterIds.isEmpty()) {
			for (final ByteArrayId adapterId : adapterIds) {
				scanner.fetchColumnFamily(new Text(
						adapterId.getBytes()));
			}
		}
		return scanner;
	}
	catch (final TableNotFoundException e) {
		LOGGER.warn(
				"Unable to query table '" + tableName + "'.  Table does not exist.",
				e);
		return null;
	}
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
@Override
protected Iterator initIterator(
		final AdapterStore adapterStore,
		final ScannerBase scanner ) {
	return new InputFormatIteratorWrapper(
			useWholeRowIterator(),
			adapterStore,
			index,
			scanner.iterator(),
			isOutputWritable,
			clientFilters.isEmpty() ? null : clientFilters.size() == 1 ? clientFilters.get(0)
					: new FilterList<QueryFilter>(
							clientFilters));
}
 

import org.apache.accumulo.core.client.ScannerBase; //导入依赖的package包/类
@Override
protected CloseableIterator<Object> initCloseableIterator(
		ScannerBase scanner,
		Iterator it ) {
	return new CloseableIteratorWrapper(
			new Closeable() {
				boolean closed = false;

				@Override
				public void close()
						throws IOException {
					if (!closed) {
						if (scanner instanceof BatchDeleter) {
							try {
								((BatchDeleter) scanner).delete();
							}
							catch (MutationsRejectedException | TableNotFoundException e) {
								LOGGER.warn(
										"Unable to delete rows by query constraints",
										e);
							}
						}
						scanner.close();
					}
					closed = true;
				}
			},
			it);
}