<commit_before>package com.splicemachine.derby.impl.storage;

import com.splicemachine.async.*;
import com.splicemachine.async.AsyncScanner;
import com.splicemachine.constants.SpliceConstants;
import com.splicemachine.derby.hbase.SpliceDriver;
import com.splicemachine.derby.iapi.sql.execute.SpliceRuntimeContext;
import com.splicemachine.derby.metrics.OperationMetric;
import com.splicemachine.derby.metrics.OperationRuntimeStats;
import com.splicemachine.derby.utils.marshall.PairDecoder;
import com.splicemachine.hbase.regioninfocache.HBaseRegionCache;
import com.splicemachine.hbase.regioninfocache.RegionCache;
import com.splicemachine.hbase.ScanDivider;
import com.splicemachine.metrics.BaseIOStats;
import com.splicemachine.metrics.IOStats;
import com.splicemachine.metrics.TimeView;
import com.splicemachine.pipeline.exception.Exceptions;
import com.splicemachine.utils.SpliceLogUtils;

import com.splicemachine.db.iapi.error.StandardException;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.util.List;
import java.util.SortedSet;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;

/**
 * @author <NAME>
 *         Date: 7/18/14
 */
public class AsyncClientScanProvider extends AbstractAsyncScanProvider {
    private static final Logger LOG = Logger.getLogger(ClientScanProvider.class);
    private final byte[] tableName;
    private final Scan scan;
    private AsyncScanner scanner;
    private long stopExecutionTime;
    private long startExecutionTime;
    private final RegionCache regionCache;
    private final HBaseClient hbaseClient;

    private boolean opened = false;

    public AsyncClientScanProvider(String type,
                              byte[] tableName, Scan scan,
                              PairDecoder decoder,
                              SpliceRuntimeContext spliceRuntimeContext) {
        super(decoder, type, spliceRuntimeContext);
        SpliceLogUtils.trace(LOG, "instantiated");
        this.tableName = tableName;
        this.scan = scan;
        this.regionCache = HBaseRegionCache.getInstance();
        this.hbaseClient = AsyncHbase.HBASE_CLIENT;
    }

    @Override
    public List<KeyValue> getResult() throws StandardException, IOException {
        if(!opened) {
            scanner.open();
            opened=true;
        }
        try {
            return scanner.nextKeyValues();
        } catch (Exception e) {
            throw Exceptions.parseException(e);
        }
    }

    private static final int PARALLEL_SCAN_REGION_THRESHOLD = 2; //TODO -sf- make this configurable
    @Override
    public void open() {
        SpliceLogUtils.trace(LOG, "open");
        try {
            SortedSet<Pair<HRegionInfo,ServerName>> regions = regionCache.getRegionsInRange(tableName, scan.getStartRow(), scan.getStopRow());
            if (regions.size() < PARALLEL_SCAN_REGION_THRESHOLD) {
                scanner = new QueueingAsyncScanner(DerbyAsyncScannerUtils.convertScanner(scan, tableName, hbaseClient), spliceRuntimeContext);<commit_msg>Added Warning to Logs in case of an asyncprovider exception<commit_after>package com.splicemachine.derby.impl.storage;

import com.splicemachine.async.*;
import com.splicemachine.async.AsyncScanner;
import com.splicemachine.constants.SpliceConstants;
import com.splicemachine.derby.hbase.SpliceDriver;
import com.splicemachine.derby.iapi.sql.execute.SpliceRuntimeContext;
import com.splicemachine.derby.metrics.OperationMetric;
import com.splicemachine.derby.metrics.OperationRuntimeStats;
import com.splicemachine.derby.utils.marshall.PairDecoder;
import com.splicemachine.hbase.regioninfocache.HBaseRegionCache;
import com.splicemachine.hbase.regioninfocache.RegionCache;
import com.splicemachine.hbase.ScanDivider;
import com.splicemachine.metrics.BaseIOStats;
import com.splicemachine.metrics.IOStats;
import com.splicemachine.metrics.TimeView;
import com.splicemachine.pipeline.exception.Exceptions;
import com.splicemachine.utils.SpliceLogUtils;

import com.splicemachine.db.iapi.error.StandardException;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.util.List;
import java.util.SortedSet;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;

/**
 * @author <NAME>
 *         Date: 7/18/14
 */
public class AsyncClientScanProvider extends AbstractAsyncScanProvider {
    private static final Logger LOG = Logger.getLogger(ClientScanProvider.class);
    private final byte[] tableName;
    private final Scan scan;
    private AsyncScanner scanner;
    private long stopExecutionTime;
    private long startExecutionTime;
    private final RegionCache regionCache;
    private final HBaseClient hbaseClient;

    private boolean opened = false;

    public AsyncClientScanProvider(String type,
                              byte[] tableName, Scan scan,
                              PairDecoder decoder,
                              SpliceRuntimeContext spliceRuntimeContext) {
        super(decoder, type, spliceRuntimeContext);
        SpliceLogUtils.trace(LOG, "instantiated");
        this.tableName = tableName;
        this.scan = scan;
        this.regionCache = HBaseRegionCache.getInstance();
        this.hbaseClient = AsyncHbase.HBASE_CLIENT;
    }

    @Override
    public List<KeyValue> getResult() throws StandardException, IOException {
        if(!opened) {
            scanner.open();
            opened=true;
        }
        try {
            return scanner.nextKeyValues();
        } catch (Exception e) {
            SpliceLogUtils.warn(LOG,"AsyncClientScanProvider#getResult handled an exception",e);
            throw Exceptions.parseException(e);
        }
    }

    private static final int PARALLEL_SCAN_REGION_THRESHOLD = 2; //TODO -sf- make this configurable
    @Override
    public void open() {
        SpliceLogUtils.trace(LOG, "open");
        try {
            SortedSet<Pair<HRegionInfo,ServerName>> regions = regionCache.getRegionsInRange(tableName, scan.getStartRow(), scan.getStopRow());
            if (regions.size() < PARALLEL_SCAN_REGION_THRESHOLD) {
                scanner = new QueueingAsyncScanner(DerbyAsyncScannerUtils.convertScanner(scan, tableName, hbaseClient), spliceRuntimeContext);
{'fraction_non_alphanumeric': 0.08911777877295118, 'fraction_numerical': 0.0020898641588296763, 'mean_word_length': 3.6689895470383274, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2775072', 'n_tokens_mistral': 1974, 'n_tokens_neox': 1927, 'n_words': 364}