<commit_before>/*
 * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy
 * of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed
 * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
 * CONDITIONS OF ANY KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations under the License.
 */

package org.wso2.siddhi.extension.eventtable.rdbms;


import org.apache.hadoop.util.bloom.CountingBloomFilter;
import org.apache.hadoop.util.bloom.Key;
import org.apache.hadoop.util.hash.Hash;
import org.apache.log4j.Logger;
import org.wso2.siddhi.core.event.ComplexEventChunk;
import org.wso2.siddhi.core.event.stream.StreamEvent;
import org.wso2.siddhi.core.exception.ExecutionPlanRuntimeException;
import org.wso2.siddhi.extension.eventtable.cache.CachingTable;
import org.wso2.siddhi.query.api.definition.Attribute;
import org.wso2.siddhi.query.api.definition.TableDefinition;

import javax.sql.DataSource;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class DBHandler {

    private String tableName;
    private Map<String, String> elementMappings;
    private ExecutionInfo executionInfo;
    private List<Attribute> attributeList;
    private DataSource dataSource;
    private CountingBloomFilter[] bloomFilters;
    private boolean isBloomFilterEnabled;
    private TableDefinition tableDefinition;
    private int bloomFilterSize;
    private int bloomFilterHashFunction;
    private static final Logger log = Logger.getLogger(DBHandler.class);


    public DBHandler(DataSource dataSource, String tableName, List<Attribute> attributeList, TableDefinition tableDefinition) {

        Connection con = null;
        this.dataSource = dataSource;
        this.tableName = tableName;
        this.attributeList = attributeList;
        this.tableDefinition = tableDefinition;
        executionInfo = new ExecutionInfo();

        try {
            con = dataSource.getConnection();
            initializeDatabaseExecutionInfo(executionInfo);
            initializeConnection();


        } catch (SQLException e) {
            throw new ExecutionPlanRuntimeException("Error while initialising the connection ", e);
        } finally {
            cleanUpConnections(null, con);
        }

    }

    public ExecutionInfo getExecutionInfoInstance() {
        ExecutionInfo executionInfo = new ExecutionInfo();
        initializeDatabaseExecutionInfo(executionInfo);
        return executionInfo;
    }

    public Map<String, String> getElementMappings() {
        return elementMappings;
    }

    public String getTableName() {
        return tableName;
    }

    public List<Attribute> getAttributeList() {
        return attributeList;
    }

    public void setBloomFilterProperties(int bloomFilterSize, int bloomFilterHashFunction) {
        this.bloomFilterSize = bloomFilterSize;
        this.bloomFilterHashFunction = bloomFilterHashFunction;
    }

    public boolean isBloomFilterEnabled() {
        return isBloomFilterEnabled;
    }

    public CountingBloomFilter[] getBloomFilters() {
        return bloomFilters;
    }

    public void addEvent(ComplexEventChunk<StreamEvent> addingEventChunk, CachingTable cachingTable) {
        addingEventChunk.reset();
        PreparedStatement stmt = null;

        ArrayList<StreamEvent> bloomFilterInsertionList = null;
        if (isBloomFilterEnabled) {
            bloomFilterInsertionList = new ArrayList<StreamEvent>();
        }

        while (addingEventChunk.hasNext()) {
            StreamEvent streamEvent = addingEventChunk.next();
            Connection con = null;
            try {
                con = dataSource.getConnection();
                con.setAutoCommit(false);
                stmt = con.prepareStatement(executionInfo.getPreparedInsertStatement());
                populateStatement(streamEvent.getOutputData(), stmt, executionInfo.getInsertQueryColumnOrder());
                stmt.executeUpdate();
                con.commit();

                if (isBloomFilterEnabled && bloomFilterInsertionList != null) {
                    bloomFilterInsertionList.add(streamEvent);
                }
                if (cachingTable != null) {
                    cachingTable.add(streamEvent);
                }
            } catch (SQLException e) {
                throw new ExecutionPlanRuntimeException("Error while adding events to event table", e);
            } finally {
                cleanUpConnections(stmt, con);
            }
        }

        if (isBloomFilterEnabled) {
            addToBloomFilters(bloomFilterInsertionList);
        }
    }


    public void deleteEvent(Object[] obj, StreamEvent streamEvent, ExecutionInfo executionInfo) {

        PreparedStatement stmt = null;
        Connection con = null;
        try {

            con = dataSource.getConnection();
            con.setAutoCommit(false);
            stmt = con.prepareStatement(executionInfo.getPreparedDeleteStatement());
            populateStatement(obj, stmt, executionInfo.getDeleteQueryColumnOrder());
            int deletedRows = stmt.executeUpdate();
            con.commit();
            if (log.isDebugEnabled()) {
                log.debug(deletedRows + " rows deleted in table " + tableName);
            }

            if (isBloomFilterEnabled && deletedRows > 0) {
                removeFromBloomFilters(streamEvent);
            }

        } catch (SQLException e) {
            throw new ExecutionPlanRuntimeException("Error while deleting the event", e);
        } finally {
            cleanUpConnections(stmt, con);
        }
    }

    public void updateEvent(Object[] obj, ExecutionInfo executionInfo) {

        PreparedStatement stmt = null;
        Connection con = null;
        int updatedRows = 0;
        try {

            con = dataSource.getConnection();
            con.setAutoCommit(false);
            stmt = con.prepareStatement(executionInfo.getPreparedUpdateStatement());
            populateStatement(obj, stmt, executionInfo.getUpdateQueryColumnOrder());
            updatedRows = stmt.executeUpdate();
            con.commit();
            if (log.isDebugEnabled()) {
                log.debug(updatedRows + " updated deleted in table " + tableName);
            }
        } catch (SQLException e) {
            throw new ExecutionPlanRuntimeException("Error while updating the event", e);
        } finally {<commit_msg>Fixing the code refactoring changes for RDBMS event table<commit_after>/*
 * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy
 * of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed
 * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
 * CONDITIONS OF ANY KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations under the License.
 */

package org.wso2.siddhi.extension.eventtable.rdbms;


import org.apache.hadoop.util.bloom.CountingBloomFilter;
import org.apache.hadoop.util.bloom.Key;
import org.apache.hadoop.util.hash.Hash;
import org.apache.log4j.Logger;
import org.wso2.siddhi.core.event.ComplexEventChunk;
import org.wso2.siddhi.core.event.stream.StreamEvent;
import org.wso2.siddhi.core.exception.ExecutionPlanRuntimeException;
import org.wso2.siddhi.extension.eventtable.cache.CachingTable;
import org.wso2.siddhi.query.api.definition.Attribute;
import org.wso2.siddhi.query.api.definition.TableDefinition;

import javax.sql.DataSource;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class DBHandler {

    private String tableName;
    private Map<String, String> elementMappings;
    private ExecutionInfo executionInfo;
    private List<Attribute> attributeList;
    private DataSource dataSource;
    private CountingBloomFilter[] bloomFilters;
    private boolean isBloomFilterEnabled;
    private TableDefinition tableDefinition;
    private int bloomFilterSize;
    private int bloomFilterHashFunction;
    private static final Logger log = Logger.getLogger(DBHandler.class);


    public DBHandler(DataSource dataSource, String tableName, List<Attribute> attributeList, TableDefinition tableDefinition) {

        Connection con = null;
        this.dataSource = dataSource;
        this.tableName = tableName;
        this.attributeList = attributeList;
        this.tableDefinition = tableDefinition;
        executionInfo = new ExecutionInfo();

        try {
            con = dataSource.getConnection();
            initializeDatabaseExecutionInfo(executionInfo);
            initializeConnection();


        } catch (SQLException e) {
            throw new ExecutionPlanRuntimeException("Error while initialising the connection ", e);
        } finally {
            cleanUpConnections(null, con);
        }

    }

    public ExecutionInfo getExecutionInfoInstance() {
        ExecutionInfo executionInfo = new ExecutionInfo();
        initializeDatabaseExecutionInfo(executionInfo);
        return executionInfo;
    }

    public Map<String, String> getElementMappings() {
        return elementMappings;
    }

    public String getTableName() {
        return tableName;
    }

    public List<Attribute> getAttributeList() {
        return attributeList;
    }

    public void setBloomFilterProperties(int bloomFilterSize, int bloomFilterHashFunction) {
        this.bloomFilterSize = bloomFilterSize;
        this.bloomFilterHashFunction = bloomFilterHashFunction;
    }

    public boolean isBloomFilterEnabled() {
        return isBloomFilterEnabled;
    }

    public CountingBloomFilter[] getBloomFilters() {
        return bloomFilters;
    }

    public void addEvent(ComplexEventChunk<StreamEvent> addingEventChunk, CachingTable cachingTable) {
        addingEventChunk.reset();
        PreparedStatement stmt = null;

        ArrayList<StreamEvent> bloomFilterInsertionList = null;
        if (isBloomFilterEnabled) {
            bloomFilterInsertionList = new ArrayList<StreamEvent>();
        }

        while (addingEventChunk.hasNext()) {
            StreamEvent streamEvent = addingEventChunk.next();
            Connection con = null;
            try {
                con = dataSource.getConnection();
                con.setAutoCommit(false);
                stmt = con.prepareStatement(executionInfo.getPreparedInsertStatement());
                populateStatement(streamEvent.getOutputData(), stmt, executionInfo.getInsertQueryColumnOrder());
                stmt.executeUpdate();
                con.commit();

                if (isBloomFilterEnabled && bloomFilterInsertionList != null) {
                    bloomFilterInsertionList.add(streamEvent);
                }
                if (cachingTable != null) {
                    cachingTable.add(streamEvent);
                }
            } catch (SQLException e) {
                throw new ExecutionPlanRuntimeException("Error while adding events to event table", e);
            } finally {
                cleanUpConnections(stmt, con);
            }
        }

        if (isBloomFilterEnabled) {
            addToBloomFilters(bloomFilterInsertionList);
        }
    }


    public void deleteEvent(Object[] obj, StreamEvent streamEvent, ExecutionInfo executionInfo) {

        PreparedStatement stmt = null;
        Connection con = null;
        try {

            con = dataSource.getConnection();
            con.setAutoCommit(false);
            stmt = con.prepareStatement(executionInfo.getPreparedDeleteStatement());
            populateStatement(obj, stmt, executionInfo.getDeleteQueryColumnOrder());
            int deletedRows = stmt.executeUpdate();
            con.commit();
            if (log.isDebugEnabled()) {
                log.debug(deletedRows + " rows deleted in table " + tableName);
            }

            if (isBloomFilterEnabled && deletedRows > 0) {
                removeFromBloomFilters(streamEvent);
            }

        } catch (SQLException e) {
            throw new ExecutionPlanRuntimeException("Error while deleting the event", e);
        } finally {
            cleanUpConnections(stmt, con);
        }
    }

    public void updateEvent(Object[] obj, ExecutionInfo executionInfo) {

        PreparedStatement stmt = null;
        Connection con = null;
        int updatedRows = 0;
        try {

            con = dataSource.getConnection();
            con.setAutoCommit(false);
            stmt = con.prepareStatement(executionInfo.getPreparedUpdateStatement());
            populateStatement(obj, stmt, executionInfo.getUpdateQueryColumnOrder());
            updatedRows = stmt.executeUpdate();
            con.commit();
            if (log.isDebugEnabled()) {
                log.debug(updatedRows + " updated in table " + tableName);
            }
        } catch (SQLException e) {
            throw new ExecutionPlanRuntimeException("Error while updating the event", e);
        } finally {
{'fraction_non_alphanumeric': 0.08292682926829269, 'fraction_numerical': 0.0029563932002956393, 'mean_word_length': 2.651106314085267, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, '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': '7327399', 'n_tokens_mistral': 3507, 'n_tokens_neox': 3372, 'n_words': 835}