<commit_before>/*
 * Copyright (c) 2016, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. licenses this file to you 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 io.siddhi.core.table.record;

import io.siddhi.core.config.SiddhiAppContext;
import io.siddhi.core.config.SiddhiQueryContext;
import io.siddhi.core.event.ComplexEventChunk;
import io.siddhi.core.event.state.StateEvent;
import io.siddhi.core.event.stream.StreamEvent;
import io.siddhi.core.event.stream.StreamEventCloner;
import io.siddhi.core.event.stream.StreamEventFactory;
import io.siddhi.core.exception.ConnectionUnavailableException;
import io.siddhi.core.exception.SiddhiAppRuntimeException;
import io.siddhi.core.executor.ExpressionExecutor;
import io.siddhi.core.executor.VariableExpressionExecutor;
import io.siddhi.core.query.processor.ProcessingMode;
import io.siddhi.core.table.CompiledUpdateSet;
import io.siddhi.core.table.Table;
import io.siddhi.core.util.collection.AddingStreamEventExtractor;
import io.siddhi.core.util.collection.operator.CompiledCondition;
import io.siddhi.core.util.collection.operator.CompiledExpression;
import io.siddhi.core.util.collection.operator.MatchingMetaInfoHolder;
import io.siddhi.core.util.config.ConfigReader;
import io.siddhi.core.util.error.handler.util.ErrorOccurrence;
import io.siddhi.core.util.parser.ExpressionParser;
import io.siddhi.core.util.transport.DynamicOptions;
import io.siddhi.query.api.definition.TableDefinition;
import io.siddhi.query.api.execution.query.output.stream.UpdateSet;
import io.siddhi.query.api.expression.Expression;
import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * An abstract implementation of table. Abstract implementation will handle {@link ComplexEventChunk} so that
 * developer can directly work with event data.
 */
public abstract class AbstractRecordTable extends Table {
    private static final Logger log = Logger.getLogger(AbstractRecordTable.class);
    private ThreadLocal<DynamicOptions> trpDynamicOptions;
    protected StreamEventFactory storeEventPool;
    protected RecordTableHandler recordTableHandler;

    @Override
    public void init(TableDefinition tableDefinition, StreamEventFactory storeEventPool,
                     StreamEventCloner storeEventCloner, ConfigReader configReader, SiddhiAppContext
                             siddhiAppContext, RecordTableHandler recordTableHandler) {
        if (recordTableHandler != null) {
            recordTableHandler.init(tableDefinition, new RecordTableHandlerCallback(this),
                    siddhiAppContext);
        }
        this.recordTableHandler = recordTableHandler;
        this.storeEventPool = storeEventPool;
        init(tableDefinition, configReader);
        initCache(tableDefinition, siddhiAppContext, storeEventCloner, configReader);
    }

    protected void initCache(TableDefinition tableDefinition, SiddhiAppContext siddhiAppContext,
                             StreamEventCloner storeEventCloner, ConfigReader configReader) {

    }

    /**
     * Initializing the Record Table
     *
     * @param tableDefinition definition of the table with annotations if any
     * @param configReader    this hold the {@link AbstractRecordTable} configuration reader.
     */
    protected abstract void init(TableDefinition tableDefinition, ConfigReader configReader);

    @Override
    public TableDefinition getTableDefinition() {
        return tableDefinition;
    }

    @Override
    public void add(ComplexEventChunk<StreamEvent> addingEventChunk) {
        List<Object[]> records = new ArrayList<>();
        addingEventChunk.reset();
        long timestamp = 0L;
        while (addingEventChunk.hasNext()) {
            StreamEvent event = addingEventChunk.next();
            records.add(event.getOutputData());
            timestamp = event.getTimestamp();
        }
        try {
            if (recordTableHandler != null) {
                recordTableHandler.add(timestamp, records);
            } else {
                add(records);
            }
        } catch (ConnectionUnavailableException e) { // TODO: 2020-09-25 search about handling other DB errors 
            onAddError(addingEventChunk, e, ErrorOccurrence.STORE_ON_TABLE_ADD, true);
        } catch (SiddhiAppRuntimeException e) { // TODO: 2020-09-28 introduce a new runtime error 
            onAddError(addingEventChunk, e, ErrorOccurrence.STORE_ON_TABLE_ADD, false);
        }
        // TODO: 2020-09-25 catch  SiddhiAppRuntimeException and validate for RDBMS constraint violation exception
        //  and handle it. Don't give replay option for these types of errors
    }

    /**
     * Add records to the Table
     *
     * @param records records that need to be added to the table, each Object[] represent a record and it will match
     *                the attributes of the Table Definition.
     * @throws ConnectionUnavailableException
     */
    protected abstract void add(List<Object[]> records) throws ConnectionUnavailableException;
<commit_msg>Use DatabaseConstraintViolationException to store DB errors<commit_after>/*
 * Copyright (c) 2016, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. licenses this file to you 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 io.siddhi.core.table.record;

import io.siddhi.core.config.SiddhiAppContext;
import io.siddhi.core.config.SiddhiQueryContext;
import io.siddhi.core.event.ComplexEventChunk;
import io.siddhi.core.event.state.StateEvent;
import io.siddhi.core.event.stream.StreamEvent;
import io.siddhi.core.event.stream.StreamEventCloner;
import io.siddhi.core.event.stream.StreamEventFactory;
import io.siddhi.core.exception.ConnectionUnavailableException;
import io.siddhi.core.exception.DatabaseConstraintViolationException;
import io.siddhi.core.executor.ExpressionExecutor;
import io.siddhi.core.executor.VariableExpressionExecutor;
import io.siddhi.core.query.processor.ProcessingMode;
import io.siddhi.core.table.CompiledUpdateSet;
import io.siddhi.core.table.Table;
import io.siddhi.core.util.collection.AddingStreamEventExtractor;
import io.siddhi.core.util.collection.operator.CompiledCondition;
import io.siddhi.core.util.collection.operator.CompiledExpression;
import io.siddhi.core.util.collection.operator.MatchingMetaInfoHolder;
import io.siddhi.core.util.config.ConfigReader;
import io.siddhi.core.util.error.handler.util.ErrorOccurrence;
import io.siddhi.core.util.parser.ExpressionParser;
import io.siddhi.core.util.transport.DynamicOptions;
import io.siddhi.query.api.definition.TableDefinition;
import io.siddhi.query.api.execution.query.output.stream.UpdateSet;
import io.siddhi.query.api.expression.Expression;
import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * An abstract implementation of table. Abstract implementation will handle {@link ComplexEventChunk} so that
 * developer can directly work with event data.
 */
public abstract class AbstractRecordTable extends Table {
    private static final Logger log = Logger.getLogger(AbstractRecordTable.class);
    private ThreadLocal<DynamicOptions> trpDynamicOptions;
    protected StreamEventFactory storeEventPool;
    protected RecordTableHandler recordTableHandler;

    @Override
    public void init(TableDefinition tableDefinition, StreamEventFactory storeEventPool,
                     StreamEventCloner storeEventCloner, ConfigReader configReader, SiddhiAppContext
                             siddhiAppContext, RecordTableHandler recordTableHandler) {
        if (recordTableHandler != null) {
            recordTableHandler.init(tableDefinition, new RecordTableHandlerCallback(this),
                    siddhiAppContext);
        }
        this.recordTableHandler = recordTableHandler;
        this.storeEventPool = storeEventPool;
        init(tableDefinition, configReader);
        initCache(tableDefinition, siddhiAppContext, storeEventCloner, configReader);
    }

    protected void initCache(TableDefinition tableDefinition, SiddhiAppContext siddhiAppContext,
                             StreamEventCloner storeEventCloner, ConfigReader configReader) {

    }

    /**
     * Initializing the Record Table
     *
     * @param tableDefinition definition of the table with annotations if any
     * @param configReader    this hold the {@link AbstractRecordTable} configuration reader.
     */
    protected abstract void init(TableDefinition tableDefinition, ConfigReader configReader);

    @Override
    public TableDefinition getTableDefinition() {
        return tableDefinition;
    }

    @Override
    public void add(ComplexEventChunk<StreamEvent> addingEventChunk) {
        List<Object[]> records = new ArrayList<>();
        addingEventChunk.reset();
        long timestamp = 0L;
        while (addingEventChunk.hasNext()) {
            StreamEvent event = addingEventChunk.next();
            records.add(event.getOutputData());
            timestamp = event.getTimestamp();
        }
        try {
            if (recordTableHandler != null) {
                recordTableHandler.add(timestamp, records);
            } else {
                add(records);
            }
        } catch (ConnectionUnavailableException e) {
            onAddError(addingEventChunk, e, ErrorOccurrence.STORE_ON_TABLE_ADD, true);
        } catch (DatabaseConstraintViolationException e) {
            onAddError(addingEventChunk, e, ErrorOccurrence.STORE_ON_TABLE_ADD, false);
        }
        // TODO: 2020-09-25 catch  SiddhiAppRuntimeException and validate for RDBMS constraint violation exception
        //  and handle it. Don't give replay option for these types of errors
    }

    /**
     * Add records to the Table
     *
     * @param records records that need to be added to the table, each Object[] represent a record and it will match
     *                the attributes of the Table Definition.
     * @throws ConnectionUnavailableException
     */
    protected abstract void add(List<Object[]> records) throws ConnectionUnavailableException;

{'fraction_non_alphanumeric': 0.08035555555555556, 'fraction_numerical': 0.005155555555555556, 'mean_word_length': 3.5828920570264766, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, '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': '2501731', 'n_tokens_mistral': 3027, 'n_tokens_neox': 2947, 'n_words': 786}