<commit_before>package com.splicemachine.derby.hbase;

import com.google.common.collect.Lists;
import com.splicemachine.async.Bytes;
import com.splicemachine.constants.SIConstants;
import com.splicemachine.constants.SpliceConstants;
import com.splicemachine.db.iapi.error.StandardException;
import com.splicemachine.derby.impl.temp.TempTable;
import com.splicemachine.hbase.KVPair;
import com.splicemachine.hbase.backup.Backup;
import com.splicemachine.hbase.backup.SnapshotUtils;
import com.splicemachine.hbase.backup.SnapshotUtilsFactory;
import com.splicemachine.mrio.MRConstants;
import com.splicemachine.mrio.api.core.MemstoreAware;
import com.splicemachine.hbase.backup.BackupUtils;
import com.splicemachine.pipeline.api.WriteContext;
import com.splicemachine.pipeline.writecontextfactory.WriteContextFactory;
import com.splicemachine.pipeline.constraint.Constraint;
import com.splicemachine.pipeline.constraint.ConstraintViolation;
import com.splicemachine.pipeline.impl.WriteResult;
import com.splicemachine.pipeline.writecontextfactory.WriteContextFactoryManager;
import com.splicemachine.si.api.TransactionalRegion;
import com.splicemachine.si.api.TxnOperationFactory;
import com.splicemachine.si.api.TxnView;
import com.splicemachine.si.impl.SimpleOperationFactory;
import com.splicemachine.si.impl.TransactionalRegions;
import com.splicemachine.si.impl.WriteConflict;
import com.splicemachine.utils.SpliceLogUtils;

import com.splicemachine.utils.ZkUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.io.HFileLink;
import org.apache.hadoop.hbase.regionserver.*;
import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequest;
import com.splicemachine.hbase.backup.BackupFileSet;
import org.apache.hadoop.hbase.zookeeper.RecoverableZooKeeper;
import org.apache.log4j.Logger;
import org.apache.zookeeper.KeeperException;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicReference;

/**
 * Region Observer for managing indices.
 *
 * @author <NAME>
 *         Created on: 2/28/13
 */
public abstract class AbstractSpliceIndexObserver extends BaseRegionObserver {
    private static final Logger LOG = Logger.getLogger(AbstractSpliceIndexObserver.class);
    protected AtomicReference<MemstoreAware> memstoreAware = new AtomicReference<MemstoreAware>(new MemstoreAware());     // Atomic Reference to memstore aware state handling
    protected DerbyFactory factory = DerbyFactoryDriver.derbyFactory;
    
    /**
     * Log component specific to compaction related code, so that this functionality
     * can be logged in isolation without having to also see the rest of the logging
     * from this class. To see additional temp table compaction logging,
     * also use log component: com.splicemachine.derby.impl.temp.TempTable.
     */
    private static final Logger LOG_COMPACT = Logger.getLogger(AbstractSpliceIndexObserver.class.getName() + ".Compaction");
    private static final Logger LOG_SPLIT = Logger.getLogger(AbstractSpliceIndexObserver.class.getName() + ".Split");
    private static final String LOG_COMPACT_PRE = TempTable.LOG_COMPACT_PRE;

    protected long conglomId;
    protected boolean isTemp;
    protected long blockingStoreFiles;
    protected TransactionalRegion region;
    protected TxnOperationFactory operationFactory;

    @Override
    public void stop(CoprocessorEnvironment e) throws IOException {
        super.stop(e);
        if (region != null)
            region.close();
    }

    @Override
    public void postOpen(final ObserverContext<RegionCoprocessorEnvironment> e) {
        //get the Conglomerate Id. If it's not a table that we can index (e.g. META, ROOT, SYS_TEMP,__TXN_LOG, etc)
        //then don't bother with setting things up
        String tableName = e.getEnvironment().getRegion().getTableDesc().getNameAsString();
        try {
            conglomId = Long.parseLong(tableName);
            operationFactory = new SimpleOperationFactory();
            region = TransactionalRegions.get(e.getEnvironment().getRegion());
        } catch (NumberFormatException nfe) {
            SpliceLogUtils.debug(LOG, "Unable to parse Conglomerate Id for table %s, indexing is will not be set up", tableName);
            conglomId = -1;
            blockingStoreFiles = SpliceConstants.config.getInt("hbase.hstore.blockingStoreFiles", 10);
            isTemp = SpliceConstants.TEMP_TABLE.equals(tableName);
            return;
        }

        super.postOpen(e);
    }

    @Override
    public InternalScanner preCompactScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c,
                                                 Store store,
                                                 List<? extends KeyValueScanner> scanners,
                                                 ScanType scanType,
                                                 long earliestPutTs, InternalScanner s) throws IOException {
        if (!isTemp || (s == null && scanners.size() <= 0))
            return super.preCompactScannerOpen(c, store, scanners, scanType, earliestPutTs, s);

        if (blockingStoreFiles <= scanners.size()) {
            LOG_COMPACT.info(String.format("%s Falling back to normal HBase compaction for TEMP. Consider increasing setting for blockingStoreFiles.", LOG_COMPACT_PRE));
            return super.preCompactScannerOpen(c, store, scanners, scanType, earliestPutTs, s);
        } else {
            if (LOG_COMPACT.isTraceEnabled())
                LOG_COMPACT.trace(String.format("%s Compacting TEMP using the TEMP compaction mechanism", LOG_COMPACT_PRE));
            c.complete();
            return SpliceDriver.driver().getTempTable().getTempCompactionScanner();
        }
    }

    @Override
    public void preCompactSelection(ObserverContext<RegionCoprocessorEnvironment> c, Store store, List<StoreFile> candidates) throws IOException {
        filterFiles(c, candidates);
    }

    @Override
    public void preCompactSelection(ObserverContext<RegionCoprocessorEnvironment> c, Store store, List<StoreFile> candidates, CompactionRequest request) throws IOException {
        filterFiles(c, candidates);
    }

    @Override
    public void postCompact(ObserverContext<RegionCoprocessorEnvironment> e,
                            Store store, StoreFile resultFile, CompactionRequest request) throws IOException {

        if (LOG.isTraceEnabled())
            SpliceLogUtils.trace(LOG, "postCompact store=%s, storeFile=%s, request=%s", store, resultFile, request);

        recordCompactedFilesForBackup(e, request, resultFile);

        while (true) {
            MemstoreAware latest = memstoreAware.get();
            if (memstoreAware.compareAndSet(latest, MemstoreAware.decrementCompactionCount(latest))) ;
            break;
        }
        super.postCompact(e, store, resultFile, request);
    }
    
    @Override
    public void postSplit(ObserverContext<RegionCoprocessorEnvironment> e, HRegion l, HRegion r) throws IOException {
        recordRegionSplitForBackup(e, l, r);
    }
    
    /**
     * ***************************************************************************************************************
     */
    /*private helper methods*/

    private void updateFileSet(String tableName,
                               String parentRegionName,
                               Set<String> pathSet,
                               HashMap<byte[], Collection<StoreFile>> storeFileInfoMap,
                               HRegion r) throws StandardException{

        if (LOG_SPLIT.isTraceEnabled()) {
            SpliceLogUtils.trace(LOG_SPLIT, "updateFileSet: region = " + r.getRegionInfo().getEncodedName());
        }
        String childRegionName = r.getRegionInfo().getEncodedName();
        for(Collection<StoreFile> storeFiles : storeFileInfoMap.values()) {
            for (StoreFile storeFile : storeFiles) {
                String referenceFileName = storeFile.getPath().getName();
                String[] s = referenceFileName.split("\\.");
                if (s.length == 1)
                    continue;
                String fileName = s[0];<commit_msg>Keep track of finished splits<commit_after>package com.splicemachine.derby.hbase;

import com.google.common.collect.Lists;
import com.splicemachine.async.Bytes;
import com.splicemachine.constants.SIConstants;
import com.splicemachine.constants.SpliceConstants;
import com.splicemachine.db.iapi.error.StandardException;
import com.splicemachine.derby.impl.temp.TempTable;
import com.splicemachine.hbase.KVPair;
import com.splicemachine.hbase.backup.Backup;
import com.splicemachine.hbase.backup.SnapshotUtils;
import com.splicemachine.hbase.backup.SnapshotUtilsFactory;
import com.splicemachine.mrio.MRConstants;
import com.splicemachine.mrio.api.core.MemstoreAware;
import com.splicemachine.hbase.backup.BackupUtils;
import com.splicemachine.pipeline.api.WriteContext;
import com.splicemachine.pipeline.writecontextfactory.WriteContextFactory;
import com.splicemachine.pipeline.constraint.Constraint;
import com.splicemachine.pipeline.constraint.ConstraintViolation;
import com.splicemachine.pipeline.impl.WriteResult;
import com.splicemachine.pipeline.writecontextfactory.WriteContextFactoryManager;
import com.splicemachine.si.api.TransactionalRegion;
import com.splicemachine.si.api.TxnOperationFactory;
import com.splicemachine.si.api.TxnView;
import com.splicemachine.si.impl.SimpleOperationFactory;
import com.splicemachine.si.impl.TransactionalRegions;
import com.splicemachine.si.impl.WriteConflict;
import com.splicemachine.utils.SpliceLogUtils;

import com.splicemachine.utils.ZkUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.io.HFileLink;
import org.apache.hadoop.hbase.regionserver.*;
import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequest;
import com.splicemachine.hbase.backup.BackupFileSet;
import org.apache.hadoop.hbase.zookeeper.RecoverableZooKeeper;
import org.apache.log4j.Logger;
import org.apache.zookeeper.KeeperException;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicReference;

/**
 * Region Observer for managing indices.
 *
 * @author <NAME>
 *         Created on: 2/28/13
 */
public abstract class AbstractSpliceIndexObserver extends BaseRegionObserver {
    private static final Logger LOG = Logger.getLogger(AbstractSpliceIndexObserver.class);
    protected AtomicReference<MemstoreAware> memstoreAware = new AtomicReference<MemstoreAware>(new MemstoreAware());     // Atomic Reference to memstore aware state handling
    protected DerbyFactory factory = DerbyFactoryDriver.derbyFactory;
    
    /**
     * Log component specific to compaction related code, so that this functionality
     * can be logged in isolation without having to also see the rest of the logging
     * from this class. To see additional temp table compaction logging,
     * also use log component: com.splicemachine.derby.impl.temp.TempTable.
     */
    private static final Logger LOG_COMPACT = Logger.getLogger(AbstractSpliceIndexObserver.class.getName() + ".Compaction");
    private static final Logger LOG_SPLIT = Logger.getLogger(AbstractSpliceIndexObserver.class.getName() + ".Split");
    private static final String LOG_COMPACT_PRE = TempTable.LOG_COMPACT_PRE;

    protected long conglomId;
    protected boolean isTemp;
    protected long blockingStoreFiles;
    protected TransactionalRegion region;
    protected TxnOperationFactory operationFactory;

    @Override
    public void stop(CoprocessorEnvironment e) throws IOException {
        super.stop(e);
        if (region != null)
            region.close();
    }

    @Override
    public void postOpen(final ObserverContext<RegionCoprocessorEnvironment> e) {
        //get the Conglomerate Id. If it's not a table that we can index (e.g. META, ROOT, SYS_TEMP,__TXN_LOG, etc)
        //then don't bother with setting things up
        String tableName = e.getEnvironment().getRegion().getTableDesc().getNameAsString();
        try {
            conglomId = Long.parseLong(tableName);
            operationFactory = new SimpleOperationFactory();
            region = TransactionalRegions.get(e.getEnvironment().getRegion());
        } catch (NumberFormatException nfe) {
            SpliceLogUtils.debug(LOG, "Unable to parse Conglomerate Id for table %s, indexing is will not be set up", tableName);
            conglomId = -1;
            blockingStoreFiles = SpliceConstants.config.getInt("hbase.hstore.blockingStoreFiles", 10);
            isTemp = SpliceConstants.TEMP_TABLE.equals(tableName);
            return;
        }

        super.postOpen(e);
    }

    @Override
    public InternalScanner preCompactScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c,
                                                 Store store,
                                                 List<? extends KeyValueScanner> scanners,
                                                 ScanType scanType,
                                                 long earliestPutTs, InternalScanner s) throws IOException {
        if (!isTemp || (s == null && scanners.size() <= 0))
            return super.preCompactScannerOpen(c, store, scanners, scanType, earliestPutTs, s);

        if (blockingStoreFiles <= scanners.size()) {
            LOG_COMPACT.info(String.format("%s Falling back to normal HBase compaction for TEMP. Consider increasing setting for blockingStoreFiles.", LOG_COMPACT_PRE));
            return super.preCompactScannerOpen(c, store, scanners, scanType, earliestPutTs, s);
        } else {
            if (LOG_COMPACT.isTraceEnabled())
                LOG_COMPACT.trace(String.format("%s Compacting TEMP using the TEMP compaction mechanism", LOG_COMPACT_PRE));
            c.complete();
            return SpliceDriver.driver().getTempTable().getTempCompactionScanner();
        }
    }

    @Override
    public void preCompactSelection(ObserverContext<RegionCoprocessorEnvironment> c, Store store, List<StoreFile> candidates) throws IOException {
        filterFiles(c, candidates);
    }

    @Override
    public void preCompactSelection(ObserverContext<RegionCoprocessorEnvironment> c, Store store, List<StoreFile> candidates, CompactionRequest request) throws IOException {
        filterFiles(c, candidates);
    }

    @Override
    public void postCompact(ObserverContext<RegionCoprocessorEnvironment> e,
                            Store store, StoreFile resultFile, CompactionRequest request) throws IOException {

        if (LOG.isTraceEnabled())
            SpliceLogUtils.trace(LOG, "postCompact store=%s, storeFile=%s, request=%s", store, resultFile, request);

        recordCompactedFilesForBackup(e, request, resultFile);

        while (true) {
            MemstoreAware latest = memstoreAware.get();
            if (memstoreAware.compareAndSet(latest, MemstoreAware.decrementCompactionCount(latest))) ;
            break;
        }
        super.postCompact(e, store, resultFile, request);
    }
    
    @Override
    public void postSplit(ObserverContext<RegionCoprocessorEnvironment> e, HRegion l, HRegion r) throws IOException {
        while (true) {
            MemstoreAware latest = memstoreAware.get();
            if(memstoreAware.compareAndSet(latest, MemstoreAware.changeSplitMerge(latest, false)));
            break;
        }
        recordRegionSplitForBackup(e, l, r);
    }
    
    /**
     * ***************************************************************************************************************
     */
    /*private helper methods*/

    private void updateFileSet(String tableName,
                               String parentRegionName,
                               Set<String> pathSet,
                               HashMap<byte[], Collection<StoreFile>> storeFileInfoMap,
                               HRegion r) throws StandardException{

        if (LOG_SPLIT.isTraceEnabled()) {
            SpliceLogUtils.trace(LOG_SPLIT, "updateFileSet: region = " + r.getRegionInfo().getEncodedName());
        }
        String childRegionName = r.getRegionInfo().getEncodedName();
        for(Collection<StoreFile> storeFiles : storeFileInfoMap.values()) {
            for (StoreFile storeFile : storeFiles) {
                String referenceFileName = storeFile.getPath().getName();
                String[] s = referenceFileName.split("\\.");
                if (s.length == 1)
                    continue;
                String fileName = s[0];
{'fraction_non_alphanumeric': 0.10082351595562164, 'fraction_numerical': 0.0013725265927027337, 'mean_word_length': 3.3306092124814266, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2658603', 'n_tokens_mistral': 4767, 'n_tokens_neox': 4691, 'n_words': 1000}