<commit_before>package com.nzcorp.hbase.data_rippler;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
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.filter.PrefixFilter;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;

//remember to add the hbase dependencies to the pom file
@SuppressWarnings("unused")
public class DownstreamDataRippler extends BaseRegionObserver {

    private static final Log LOGGER = LogFactory.getLog(DownstreamDataRippler.class);
    private Connection conn;
    /**
     * The table into which the values from the current table should be written into
     */
    private String destinationTable;
    /**
     * The table from which the child table rowkeys should be retrieved from
     */ 
    private String secondaryIndexTable;
    /**
     * The column family name to use in the destination table
     */
    private String targetCf;
    /**
     * The column family name from which to collect values from
     */
    private String sourceCF;
    /**
     * Whether to write a ridiculously amount of logging information
     * Use with caution
     */ 
    private boolean f_debug;


    @Override
    public void start(CoprocessorEnvironment env) throws IOException {
        /**
         * The CoprocessorEnvironment.getConfiguration() will return a
         * Hadoop Configuration element as described here:
         * https://hadoop.apache.org/docs/r2.6.1/api/org/apache/hadoop/conf/Configuration.html
         *
         * The named arguments given after the last pipe separator when installing the coprocessor will be available on
         * the above configuration element
         */

        conn = ConnectionFactory.createConnection( env.getConfiguration() );

        destinationTable = env.getConfiguration().get("destination_table");
        LOGGER.info( String.format( "Using destination table %s", destinationTable ));
        try {
            conn.getTable(TableName.valueOf(destinationTable));
        } catch ( IOException ioe ){
            String err = "Table "+destinationTable+" does not exist";
            LOGGER.fatal(err, ioe);
            throw new IOException( err, ioe);
        }

        secondaryIndexColumnFamily = env.getConfiguration().get("index_column_family");

        // the column family name to take all values from
        secondaryIndexTable = env.getConfiguration().get("secondary_index_table");

        // the column family name to take all values from
        sourceCF = env.getConfiguration().get("source_column_family");

        // the column family name to put values into in the destinationTable
        targetCf = env.getConfiguration().get("target_column_family");

        //option to run *expensive* debugging
        f_debug = Boolean.parseBoolean( env.getConfiguration().get("full_debug"));

        LOGGER.info("Initializing data rippler copying values from column family "+ sourceCF +" to "+destinationTable+":"+targetCf );
        LOGGER.info("Using secondary index "+secondaryIndexTable);

    }

    @Override
    public void postPut(final ObserverContext<RegionCoprocessorEnvironment> observerContext,
                        final Put put,
                        final WALEdit edit,
                        final Durability durability_enum)
            throws IOException
    {
	Table table = null;
        try {
            long startTime = System.nanoTime();
            long lapTime = 0L;

            /**
             * The Mutation.getCellList is the method we need, as we want to get all qualifiers of a given column family
             * Unfortunately, the method is package-private, and we only have access to the subclass `Put` here, so we
             * access the method by using the Reflection API. Please note that this is a last resort tactic and that we
             * will try to argue with the HBase projects, that there is a use case for exposing this method in the
             * public API. As `Mutation::getCellList` is not in the public API, the method could disappear without
             * deprectation warnings (or warnings at all). It will be in 2.0.0, as far as we know.
             */
            final Method meth = Mutation.class.getDeclaredMethod("getCellList", byte[].class);
            meth.setAccessible(true);
            final List<Cell> list_of_cells = (List<Cell>) meth.invoke(put, sourceCF.getBytes());

            if (list_of_cells.isEmpty()) {
                LOGGER.info("No cells in this transaction");
                return;
            }

            LOGGER.info("Found "+Integer.toString(list_of_cells.size())+" cells in Put");

            if( f_debug )
            {
                for (Cell cell: list_of_cells) {
                    final byte[] rowKey = CellUtil.cloneRow(cell);
                    LOGGER.info(String.format("Found rowkey: %s", new String(rowKey)));
                }
            }

            table = conn.getTable(TableName.valueOf(destinationTable));
            Table secTable = conn.getTable(TableName.valueOf(secondaryIndexTable));

            for (Cell cell: list_of_cells) {
                final byte[] rowKey = CellUtil.cloneRow(cell);
                final byte[] family = targetCf.getBytes();
                final byte[] qualifier = CellUtil.cloneQualifier(cell);
                final byte[] value = CellUtil.cloneValue(cell);

                LOGGER.trace(String.format("Found rowkey: %s", new String(rowKey)));

                List<byte[]> targetRowkeys = getTargetRowkeys(rowKey, secTable);
                lapTime = (System.nanoTime() - startTime)/1000000;
                LOGGER.info( String.format( "Exiting postPut, took %d%n milliseconds", lapTime ));

                LOGGER.info( String.format( "Got %s targetKeys for rowKey %s in %d ms from start", targetRowkeys.size(), new String(rowKey), lapTime));

                for (byte[] targetKey : targetRowkeys) {
                    LOGGER.trace("Put'ing into " + destinationTable + ": " + new String(targetKey));
                    Put targetData = new Put(targetKey).addColumn(family, qualifier, value);
                    LOGGER.trace(String.format("Will insert %s:%s = %s", new String(family), new String(qualifier), new String(value)));
                    table.put(targetData);
                }
                lapTime = (System.nanoTime() - startTime)/1000000;
		LOGGER.info( String.format( "Wrote %s items to %s in %n%d" ) );
            }

            long endTime = System.nanoTime();
            long elapsedTime = (endTime - startTime)/1000000;
            LOGGER.info( String.format( "Exiting postPut, took %d ms from start", elapsedTime ));

        } catch (IllegalArgumentException ex) {
            LOGGER.fatal("During the postPut operation, something went horribly wrong", ex);
            if( table != null ) {
                table.close();
            }
            throw new IllegalArgumentException(ex);
        } catch (NoSuchMethodException e) {
            LOGGER.error("In trying to acquire reference to the Mutation::getCellList, an error occurred", e);
        } catch (IllegalAccessException e) {
            LOGGER.error("In trying to assign the reference to the Mutation::getCellList to a variable, an error occurred", e);
        } catch (InvocationTargetException e) {
            LOGGER.error("In trying to invoke the Mutation::getCellList, an error occurred", e);
        }

    }

    private List<byte[]> getTargetRowkeys(byte[] rowKey, Table secTable) throws IOException{
        Scan scan = new Scan();
        byte[] filter = Bytes.add(rowKey, "+".getBytes());
        scan.setFilter(new PrefixFilter(filter));
        ResultScanner resultScanner = secTable.getScanner(scan);

        List<byte[]> targetKeys = new ArrayList<byte[]>();
        // get assembly rowkey from the secondary index
        for (Result result : resultScanner) {
            byte[] indexKey = result.getRow();
            LOGGER.trace(String.format("indexKey: %s", new String(indexKey)));
            String[] bits = new String(indexKey).split("\\+");
            LOGGER.debug(String.format("assemblyKey %s", bits[bits.length - 1]));<commit_msg>fixing logging statements and removing superflouous configuration value<commit_after>package com.nzcorp.hbase.data_rippler;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
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.filter.PrefixFilter;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;

//remember to add the hbase dependencies to the pom file
@SuppressWarnings("unused")
public class DownstreamDataRippler extends BaseRegionObserver {

    private static final Log LOGGER = LogFactory.getLog(DownstreamDataRippler.class);
    private Connection conn;
    /**
     * The table into which the values from the current table should be written into
     */
    private String destinationTable;
    /**
     * The table from which the child table rowkeys should be retrieved from
     */ 
    private String secondaryIndexTable;
    /**
     * The column family name to use in the destination table
     */
    private String targetCf;
    /**
     * The column family name from which to collect values from
     */
    private String sourceCF;
    /**
     * Whether to write a ridiculously amount of logging information
     * Use with caution
     */ 
    private boolean f_debug;


    @Override
    public void start(CoprocessorEnvironment env) throws IOException {
        /**
         * The CoprocessorEnvironment.getConfiguration() will return a
         * Hadoop Configuration element as described here:
         * https://hadoop.apache.org/docs/r2.6.1/api/org/apache/hadoop/conf/Configuration.html
         *
         * The named arguments given after the last pipe separator when installing the coprocessor will be available on
         * the above configuration element
         */

        conn = ConnectionFactory.createConnection( env.getConfiguration() );

        destinationTable = env.getConfiguration().get("destination_table");
        LOGGER.info( String.format( "Using destination table %s", destinationTable ));
        try {
            conn.getTable(TableName.valueOf(destinationTable));
        } catch ( IOException ioe ){
            String err = "Table "+destinationTable+" does not exist";
            LOGGER.fatal(err, ioe);
            throw new IOException( err, ioe);
        }

        // the column family name to take all values from
        secondaryIndexTable = env.getConfiguration().get("secondary_index_table");

        // the column family name to take all values from
        sourceCF = env.getConfiguration().get("source_column_family");

        // the column family name to put values into in the destinationTable
        targetCf = env.getConfiguration().get("target_column_family");

        //option to run *expensive* debugging
        f_debug = Boolean.parseBoolean( env.getConfiguration().get("full_debug"));

        LOGGER.info("Initializing data rippler copying values from column family "+ sourceCF +" to "+destinationTable+":"+targetCf );
        LOGGER.info("Using secondary index "+secondaryIndexTable);

    }

    @Override
    public void postPut(final ObserverContext<RegionCoprocessorEnvironment> observerContext,
                        final Put put,
                        final WALEdit edit,
                        final Durability durability_enum)
            throws IOException
    {
	Table table = null;
        try {
            long startTime = System.nanoTime();
            long lapTime = 0L;

            /**
             * The Mutation.getCellList is the method we need, as we want to get all qualifiers of a given column family
             * Unfortunately, the method is package-private, and we only have access to the subclass `Put` here, so we
             * access the method by using the Reflection API. Please note that this is a last resort tactic and that we
             * will try to argue with the HBase projects, that there is a use case for exposing this method in the
             * public API. As `Mutation::getCellList` is not in the public API, the method could disappear without
             * deprectation warnings (or warnings at all). It will be in 2.0.0, as far as we know.
             */
            final Method meth = Mutation.class.getDeclaredMethod("getCellList", byte[].class);
            meth.setAccessible(true);
            final List<Cell> list_of_cells = (List<Cell>) meth.invoke(put, sourceCF.getBytes());

            if (list_of_cells.isEmpty()) {
                LOGGER.info("No cells in this transaction");
                return;
            }

            LOGGER.info("Found "+Integer.toString(list_of_cells.size())+" cells in Put");

            if( f_debug )
            {
                for (Cell cell: list_of_cells) {
                    final byte[] rowKey = CellUtil.cloneRow(cell);
                    LOGGER.info(String.format("Found rowkey: %s", new String(rowKey)));
                }
            }

            table = conn.getTable(TableName.valueOf(destinationTable));
            Table secTable = conn.getTable(TableName.valueOf(secondaryIndexTable));

            for (Cell cell: list_of_cells) {
                final byte[] rowKey = CellUtil.cloneRow(cell);
                final byte[] family = targetCf.getBytes();
                final byte[] qualifier = CellUtil.cloneQualifier(cell);
                final byte[] value = CellUtil.cloneValue(cell);

                LOGGER.debug(String.format("Found rowkey: %s", new String(rowKey)));

                List<byte[]> targetRowkeys = getTargetRowkeys(rowKey, secTable);
                lapTime = (System.nanoTime() - startTime)/1000000;

                LOGGER.debug( String.format( "Got %s targetKeys for rowKey %s in %d ms from start", targetRowkeys.size(), new String(rowKey), lapTime));

                for (byte[] targetKey : targetRowkeys) {
                    LOGGER.trace("Put'ing into " + destinationTable + ": " + new String(targetKey));
                    Put targetData = new Put(targetKey).addColumn(family, qualifier, value);
                    LOGGER.trace(String.format("Will insert %s:%s = %s", new String(family), new String(qualifier), new String(value)));
                    table.put(targetData);
                }
                lapTime = (System.nanoTime() - startTime)/1000000;
		LOGGER.info( String.format( "Wrote %s items to %s in %d ms", targetRowkeys.size(), new String(destinationTable), lapTime ) );
            }

            long endTime = System.nanoTime();
            long elapsedTime = (endTime - startTime)/1000000;
            LOGGER.info( String.format( "Exiting postPut, took %d ms from start", elapsedTime ));

        } catch (IllegalArgumentException ex) {
            LOGGER.fatal("During the postPut operation, something went horribly wrong", ex);
            if( table != null ) {
                table.close();
            }
            throw new IllegalArgumentException(ex);
        } catch (NoSuchMethodException e) {
            LOGGER.error("In trying to acquire reference to the Mutation::getCellList, an error occurred", e);
        } catch (IllegalAccessException e) {
            LOGGER.error("In trying to assign the reference to the Mutation::getCellList to a variable, an error occurred", e);
        } catch (InvocationTargetException e) {
            LOGGER.error("In trying to invoke the Mutation::getCellList, an error occurred", e);
        }

    }

    private List<byte[]> getTargetRowkeys(byte[] rowKey, Table secTable) throws IOException{
        Scan scan = new Scan();
        byte[] filter = Bytes.add(rowKey, "+".getBytes());
        scan.setFilter(new PrefixFilter(filter));
        ResultScanner resultScanner = secTable.getScanner(scan);

        List<byte[]> targetKeys = new ArrayList<byte[]>();
        // get assembly rowkey from the secondary index
        for (Result result : resultScanner) {
            byte[] indexKey = result.getRow();
            LOGGER.trace(String.format("RowKey: %s", new String(indexKey)));
            String[] bits = new String(indexKey).split("\\+");
            LOGGER.debug(String.format("assemblyKey %s", bits[bits.length - 1]));
{'fraction_non_alphanumeric': 0.09842452019478659, 'fraction_numerical': 0.003322830134631911, 'mean_word_length': 2.5472464946149156, 'pattern_counts': {'":': 4, '<': 17, '<?xml version=': 0, '>': 17, 'https://': 2, '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': '684197', 'n_tokens_mistral': 4646, 'n_tokens_neox': 4483, 'n_words': 1351}