jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/JdbcMultiRowRecordWriter.java
private void processQueue(
      LinkedList<Record> queue,
      List<OnRecordErrorException> errorRecords,
      Connection connection,
      int maxRowsPerBatch,
      int opCode
  ) throws StageException {
    if (queue.isEmpty()) {
      return;
    }

    int rowCount = 0;
    // Assume that columns are all same for the same operation to the same table
    // If some columns are missing in record, the record goes to error.
    final Record first = queue.getFirst();
    SortedMap<String, String> columnsToParameters = recordReader.getColumnsToParameters(
        first,
        opCode,
        getColumnsToParameters(),
        opCode == OperationType.UPDATE_CODE ? getColumnsToFieldNoPK() : getColumnsToFields()
    );

    if (columnsToParameters.isEmpty()) {
      // no parameters found for configured columns
      if (LOG.isWarnEnabled()) {
        LOG.warn("No parameters found for record with ID {}; skipping", first.getHeader().getSourceId());
      }
      return;
    }

    String query = generateQueryForMultiRow(
        opCode,
        columnsToParameters,
        getPrimaryKeyColumns(),
        // the next batch will have either the max number of records, or however many are left.
        Math.min(maxRowsPerBatch, queue.size())
    );

    // Need to store removed records from queue, because we might need to add newly generated columns
    // to records for Jdbc Tee Processor.
    LinkedList<Record> removed = new LinkedList<>();

    try (PreparedStatement statement = jdbcUtil.getPreparedStatement(getGeneratedColumnMappings(), query, connection)) {
      int paramIdx = 1;
      // Start processing records in queue. All records have the same operation to the same table.
      while (!queue.isEmpty()) {
        Record r = queue.removeFirst();
        if (opCode != DELETE_CODE) {
          paramIdx = setParamsToStatement(paramIdx, statement, columnsToParameters, r, connection, opCode);
        }
        if (opCode != OperationType.INSERT_CODE) {
          paramIdx = setPrimaryKeys(paramIdx, r, statement, opCode);
        }
        removed.add(r);
        ++rowCount;
        if (rowCount == maxRowsPerBatch) {
          // time to execute the current batch
          processBatch(removed, errorRecords, statement, connection);
          // reset our counters
          rowCount = 0;
          paramIdx = 1;
          removed.clear();
        }
      }
    } catch (SQLException e) {
      handleSqlException(e, removed, errorRecords);
    }

    // Process the rest of the records that are removed from queue but haven't processed yet
    // this happens when rowCount is still less than maxRowsPerBatch.
    // This is a bit of an ugly fix as its not very DRY but sufficient until there's a larger
    // refactoring of this code.
    if (!removed.isEmpty()) {
      query = generateQueryForMultiRow(
          opCode,
          columnsToParameters,
          getPrimaryKeyColumns(),
          removed.size() // always the remainder
      );

      try (PreparedStatement statement = jdbcUtil.getPreparedStatement(
          getGeneratedColumnMappings(),
          query,
          connection
      )) {
        int paramIdx = 1;
        for (Record r : removed) {
          if (opCode != DELETE_CODE) {
            paramIdx = setParamsToStatement(paramIdx, statement, columnsToParameters, r, connection, opCode);
          }
          if (opCode != OperationType.INSERT_CODE) {
            paramIdx = setPrimaryKeys(paramIdx, r, statement, opCode);
          }
        }
        processBatch(removed, errorRecords, statement, connection);
      } catch (SQLException e) {
        handleSqlException(e, removed, errorRecords);
      }
    }
  }
private void processQueue(
      LinkedList<Record> queue,
      List<OnRecordErrorException> errorRecords,
      Connection connection,
      int maxRowsPerBatch,
      int opCode
  ) throws StageException {
    if (queue.isEmpty()) {
      return;
    }

    int rowCount = 0;
    // Assume that columns are all same for the same operation to the same table
    // If some columns are missing in record, the record goes to error.
    final Record first = queue.getFirst();
    SortedMap<String, String> columnsToParameters = recordReader.getColumnsToParameters(
        first,
        opCode,
        getColumnsToParameters(),
        opCode == OperationType.UPDATE_CODE ? getColumnsToFieldNoPK() : getColumnsToFields()
    );

    if (columnsToParameters.isEmpty()) {
      // no parameters found for configured columns
      if (LOG.isWarnEnabled()) {
        LOG.warn("No parameters found for record with ID {}; skipping", first.getHeader().getSourceId());
      }
      return;
    }

    String query = generateQueryForMultiRow(
        opCode,
        columnsToParameters,
        getPrimaryKeyColumns(),
        // the next batch will have either the max number of records, or however many are left.
        Math.min(maxRowsPerBatch, queue.size())
    );

    // Need to store removed records from queue, because we might need to add newly generated columns
    // to records for Jdbc Tee Processor.
    LinkedList<Record> removed = new LinkedList<>();

    try (PreparedStatement statement = jdbcUtil.getPreparedStatement(getGeneratedColumnMappings(), query, connection)) {
      int paramIdx = 1;
      // Start processing records in queue. All records have the same operation to the same table.
      while (!queue.isEmpty()) {
        Record r = queue.removeFirst();
        if (opCode != DELETE_CODE) {
          paramIdx = setParamsToStatement(paramIdx, statement, columnsToParameters, r, connection, opCode);
        }
        if (opCode != OperationType.INSERT_CODE) {
          paramIdx = setPrimaryKeys(paramIdx, r, statement, opCode);
        }
        removed.add(r);
        ++rowCount;
        if (rowCount == maxRowsPerBatch) {
          // time to execute the current batch
          processBatch(removed, errorRecords, statement, connection);
          // reset our counters
          rowCount = 0;
          paramIdx = 1;
          removed.clear();
        }
      }
    } catch (SQLException e) {
      handleSqlException(e, removed, errorRecords);
    }

    // Process the rest of the records that are removed from queue but haven't processed yet
    // this happens when rowCount is still less than maxRowsPerBatch.
    // This is a bit of an ugly fix as its not very DRY but sufficient until there's a larger
    // refactoring of this code.
    if (!removed.isEmpty()) {
      query = generateQueryForMultiRow(
          opCode,
          columnsToParameters,
          getPrimaryKeyColumns(),
          removed.size() // always the remainder
      );

      try (PreparedStatement statement = jdbcUtil.getPreparedStatement(
          getGeneratedColumnMappings(),
          query,
          connection
      )) {
        int paramIdx = 1;
        for (Record r : removed) {
          if (opCode != DELETE_CODE) {
            paramIdx = setParamsToStatement(paramIdx, statement, columnsToParameters, r, connection, opCode);
          }
          if (opCode != OperationType.INSERT_CODE) {
            paramIdx = setPrimaryKeys(paramIdx, r, statement, opCode);
          }
        }
        processBatch(removed, errorRecords, statement, connection);
      } catch (SQLException e) {
        handleSqlException(e, removed, errorRecords);
      }
    }
  }
private void processQueue ( LinkedList < Record > queue , List < OnRecordErrorException > errorRecords , Connection connection , int maxRowsPerBatch , int opCode ) throws StageException { if ( queue . isEmpty ( ) ) { return ; } int rowCount = 0 ; // Assume that columns are all same for the same operation to the same table // If some columns are missing in record, the record goes to error. final Record first = queue . getFirst ( ) ; SortedMap < String , String > columnsToParameters = recordReader . getColumnsToParameters ( first , opCode , getColumnsToParameters ( ) , opCode == OperationType . UPDATE_CODE ? getColumnsToFieldNoPK ( ) : getColumnsToFields ( ) ) ; if ( columnsToParameters . isEmpty ( ) ) { // no parameters found for configured columns if ( LOG . isWarnEnabled ( ) ) { LOG . warn ( "No parameters found for record with ID {}; skipping" , first . getHeader ( ) . getSourceId ( ) ) ; } return ; } String query = generateQueryForMultiRow ( opCode , columnsToParameters , getPrimaryKeyColumns ( ) , // the next batch will have either the max number of records, or however many are left. Math . min ( maxRowsPerBatch , queue . size ( ) ) ) ; // Need to store removed records from queue, because we might need to add newly generated columns // to records for Jdbc Tee Processor. LinkedList < Record > removed = new LinkedList <> ( ) ; try ( PreparedStatement statement = jdbcUtil . getPreparedStatement ( getGeneratedColumnMappings ( ) , query , connection ) ) { int paramIdx = 1 ; // Start processing records in queue. All records have the same operation to the same table. while ( ! queue . isEmpty ( ) ) { Record r = queue . removeFirst ( ) ; if ( opCode != DELETE_CODE ) { paramIdx = setParamsToStatement ( paramIdx , statement , columnsToParameters , r , connection , opCode ) ; } if ( opCode != OperationType . INSERT_CODE ) { paramIdx = setPrimaryKeys ( paramIdx , r , statement , opCode ) ; } removed . add ( r ) ; ++ rowCount ; if ( rowCount == maxRowsPerBatch ) { // time to execute the current batch processBatch ( removed , errorRecords , statement , connection ) ; // reset our counters rowCount = 0 ; paramIdx = 1 ; removed . clear ( ) ; } } } catch ( SQLException e ) { handleSqlException ( e , removed , errorRecords ) ; } // Process the rest of the records that are removed from queue but haven't processed yet // this happens when rowCount is still less than maxRowsPerBatch. // This is a bit of an ugly fix as its not very DRY but sufficient until there's a larger // refactoring of this code. if ( ! removed . isEmpty ( ) ) { query = generateQueryForMultiRow ( opCode , columnsToParameters , getPrimaryKeyColumns ( ) , removed . size ( ) // always the remainder ) ; try ( PreparedStatement statement = jdbcUtil . getPreparedStatement ( getGeneratedColumnMappings ( ) , query , connection ) ) { int paramIdx = 1 ; for ( Record r : removed ) { if ( opCode != DELETE_CODE ) { paramIdx = setParamsToStatement ( paramIdx , statement , columnsToParameters , r , connection , opCode ) ; } if ( opCode != OperationType . INSERT_CODE ) { paramIdx = setPrimaryKeys ( paramIdx , r , statement , opCode ) ; } } processBatch ( removed , errorRecords , statement , connection ) ; } catch ( SQLException e ) { handleSqlException ( e , removed , errorRecords ) ; } } }
Process all records in queue. All records have same operation to same table.
Generate a query and set parameters from each record. INSERT and DELETE can be multi-row operation
but UPDATE is single-row operation.
If maxStatement
@param errorRecords
@param connection
@param maxRowsPerBatch
@param opCode
@param queue
@return
@throws OnRecordErrorException
Process all records in queue . All records have same operation to same table . Generate a query and set parameters from each record . INSERT and DELETE can be multi - row operation but UPDATE is single - row operation . If maxStatement