<commit_before>// ============================================================================
//
// Copyright (C) 2006-2017 Talend Inc. - www.talend.com
//
// This source code is available under agreement available at
// %InstallDIR%\features\org.talend.rcp.branding.%PRODUCTNAME%\%PRODUCTNAME%license.txt
//
// You should have received a copy of the agreement
// along with this program; if not, write to Talend SA
// 9 rue Pages 92150 Suresnes, France
//
// ============================================================================
package org.talend.components.jdbc.runtime.writer;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.IndexedRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.talend.components.api.component.runtime.Result;
import org.talend.components.api.component.runtime.WriteOperation;
import org.talend.components.api.component.runtime.WriterWithFeedback;
import org.talend.components.api.container.RuntimeContainer;
import org.talend.components.api.properties.ComponentProperties;
import org.talend.components.common.avro.JDBCAvroRegistry;
import org.talend.components.jdbc.CommonUtils;
import org.talend.components.jdbc.RuntimeSettingProvider;
import org.talend.components.jdbc.runtime.JDBCRowSink;
import org.talend.components.jdbc.runtime.JdbcRuntimeUtils;
import org.talend.components.jdbc.runtime.setting.AllSetting;
import org.talend.daikon.avro.converter.IndexedRecordConverter;

/**
 * the JDBC writer for JDBC row
 *
 */
public class JDBCRowWriter implements WriterWithFeedback<Result, IndexedRecord, IndexedRecord> {

    private transient static final Logger LOG = LoggerFactory.getLogger(JDBCRowWriter.class);

    private WriteOperation<Result> writeOperation;

    private Connection conn;

    private JDBCRowSink sink;

    private AllSetting setting;

    private RuntimeSettingProvider properties;

    private RuntimeContainer runtime;

    private Result result;

    private final List<IndexedRecord> successfulWrites = new ArrayList<>();

    private final List<IndexedRecord> rejectedWrites = new ArrayList<>();

    private int successCount;

    private int rejectCount;

    private boolean useExistedConnection;

    private boolean dieOnError;

    private PreparedStatement prepared_statement;

    private Statement statement;

    private ResultSet resultSet;

    // private int insertCount;
    //
    // private int updateCount;
    //
    // private int deleteCount;

    private boolean usePreparedStatement;

    private String sql;

    private boolean propagateQueryResultSet;

    private boolean useCommit;

    private int commitCount;

    private int commitEvery;

    private Schema outSchema;

    private Schema rejectSchema;

    public JDBCRowWriter(WriteOperation<Result> writeOperation, RuntimeContainer runtime) {
        this.writeOperation = writeOperation;
        this.runtime = runtime;
        sink = (JDBCRowSink) writeOperation.getSink();
        setting = sink.properties.getRuntimeSetting();
        properties = sink.properties;

        useExistedConnection = setting.getReferencedComponentId() != null;

        Integer commitEveryNumber = setting.getCommitEvery();
        useCommit = !useExistedConnection && commitEveryNumber != null && commitEveryNumber != 0;
        if (useCommit) {
            commitEvery = commitEveryNumber;
        }

        dieOnError = setting.getDieOnError();

        propagateQueryResultSet = setting.getPropagateQueryResultSet();

        result = new Result();

        outSchema = CommonUtils.getOutputSchema((ComponentProperties) properties);

        rejectSchema = CommonUtils.getRejectSchema((ComponentProperties) properties);
    }

    public void open(String uId) throws IOException {
        try {
            conn = sink.getConnection(runtime);

            usePreparedStatement = setting.getUsePreparedStatement();
            sql = setting.getSql();

            if (usePreparedStatement) {
                prepared_statement = conn.prepareStatement(sql);
            } else {
                statement = conn.createStatement();
            }
        } catch (SQLException | ClassNotFoundException e) {
            throw CommonUtils.newComponentException(e);
        }
    }

    public void write(Object datum) throws IOException {
        result.totalCount++;

        cleanWrites();

        IndexedRecord input = this.getFactory(datum).convertToAvro(datum);

        //TODO low performance, but need to reset it by row level
        setting = sink.properties.getRuntimeSetting();
        
        try {
            if (usePreparedStatement) {
                JdbcRuntimeUtils.setPreparedStatement(prepared_statement, setting.getIndexs(), setting.getTypes(),
                        setting.getValues());

                if (propagateQueryResultSet) {
                    resultSet = prepared_statement.executeQuery();
                } else {
                    prepared_statement.execute();
                }
            } else {
                if (propagateQueryResultSet) {
                    resultSet = statement.executeQuery(sql);
                } else {
                    statement.execute(sql);
                }
            }

            handleSuccess(input);
        } catch (SQLException e) {
            if (dieOnError) {
                throw CommonUtils.newComponentException(e);
            } else {<commit_msg>fix(TDI-41611): tJDBCRow dynamic sql query (#1445)<commit_after>// ============================================================================
//
// Copyright (C) 2006-2017 Talend Inc. - www.talend.com
//
// This source code is available under agreement available at
// %InstallDIR%\features\org.talend.rcp.branding.%PRODUCTNAME%\%PRODUCTNAME%license.txt
//
// You should have received a copy of the agreement
// along with this program; if not, write to Talend SA
// 9 rue Pages 92150 Suresnes, France
//
// ============================================================================
package org.talend.components.jdbc.runtime.writer;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.IndexedRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.talend.components.api.component.runtime.Result;
import org.talend.components.api.component.runtime.WriteOperation;
import org.talend.components.api.component.runtime.WriterWithFeedback;
import org.talend.components.api.container.RuntimeContainer;
import org.talend.components.api.properties.ComponentProperties;
import org.talend.components.common.avro.JDBCAvroRegistry;
import org.talend.components.jdbc.CommonUtils;
import org.talend.components.jdbc.RuntimeSettingProvider;
import org.talend.components.jdbc.runtime.JDBCRowSink;
import org.talend.components.jdbc.runtime.JdbcRuntimeUtils;
import org.talend.components.jdbc.runtime.setting.AllSetting;
import org.talend.daikon.avro.converter.IndexedRecordConverter;

/**
 * the JDBC writer for JDBC row
 *
 */
public class JDBCRowWriter implements WriterWithFeedback<Result, IndexedRecord, IndexedRecord> {

    private transient static final Logger LOG = LoggerFactory.getLogger(JDBCRowWriter.class);

    private WriteOperation<Result> writeOperation;

    private Connection conn;

    private JDBCRowSink sink;

    private AllSetting setting;

    private RuntimeSettingProvider properties;

    private RuntimeContainer runtime;

    private Result result;

    private final List<IndexedRecord> successfulWrites = new ArrayList<>();

    private final List<IndexedRecord> rejectedWrites = new ArrayList<>();

    private int successCount;

    private int rejectCount;

    private boolean useExistedConnection;

    private boolean dieOnError;

    private PreparedStatement prepared_statement;

    private Statement statement;

    private ResultSet resultSet;

    // private int insertCount;
    //
    // private int updateCount;
    //
    // private int deleteCount;

    private boolean usePreparedStatement;

    private String sql;

    private boolean propagateQueryResultSet;

    private boolean useCommit;

    private int commitCount;

    private int commitEvery;

    private Schema outSchema;

    private Schema rejectSchema;

    public JDBCRowWriter(WriteOperation<Result> writeOperation, RuntimeContainer runtime) {
        this.writeOperation = writeOperation;
        this.runtime = runtime;
        sink = (JDBCRowSink) writeOperation.getSink();
        setting = sink.properties.getRuntimeSetting();
        properties = sink.properties;

        useExistedConnection = setting.getReferencedComponentId() != null;

        Integer commitEveryNumber = setting.getCommitEvery();
        useCommit = !useExistedConnection && commitEveryNumber != null && commitEveryNumber != 0;
        if (useCommit) {
            commitEvery = commitEveryNumber;
        }

        dieOnError = setting.getDieOnError();

        propagateQueryResultSet = setting.getPropagateQueryResultSet();

        result = new Result();

        outSchema = CommonUtils.getOutputSchema((ComponentProperties) properties);

        rejectSchema = CommonUtils.getRejectSchema((ComponentProperties) properties);
    }

    public void open(String uId) throws IOException {
        try {
            conn = sink.getConnection(runtime);

            usePreparedStatement = setting.getUsePreparedStatement();
            sql = setting.getSql();

            if (usePreparedStatement) {
                prepared_statement = conn.prepareStatement(sql);
            } else {
                statement = conn.createStatement();
            }
        } catch (SQLException | ClassNotFoundException e) {
            throw CommonUtils.newComponentException(e);
        }
    }

    public void write(Object datum) throws IOException {
        result.totalCount++;

        cleanWrites();

        IndexedRecord input = this.getFactory(datum).convertToAvro(datum);

        //TODO low performance, but need to reset it by row level
        setting = sink.properties.getRuntimeSetting();
        
        try {
            if (usePreparedStatement) {
                JdbcRuntimeUtils.setPreparedStatement(prepared_statement, setting.getIndexs(), setting.getTypes(),
                        setting.getValues());

                if (propagateQueryResultSet) {
                    resultSet = prepared_statement.executeQuery();
                } else {
                    prepared_statement.execute();
                }
            } else {
                //Need to get updated sql query in case of dynamic(row) values usage
                if (propagateQueryResultSet) {
                    resultSet = statement.executeQuery(setting.getSql());
                } else {
                    statement.execute(setting.getSql());
                }
            }

            handleSuccess(input);
        } catch (SQLException e) {
            if (dieOnError) {
                throw CommonUtils.newComponentException(e);
            } else {
{'fraction_non_alphanumeric': 0.11008461405629424, 'fraction_numerical': 0.0037126575720946298, 'mean_word_length': 2.995515695067265, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '383136', 'n_tokens_mistral': 3043, 'n_tokens_neox': 2854, 'n_words': 671}