src/main/java/de/ids_mannheim/korap/index/collector/MatchCollectorDB.java

package de.ids_mannheim.korap.index.collector;
import de.ids_mannheim.korap.KorapNode;
import de.ids_mannheim.korap.KorapMatch;
import de.ids_mannheim.korap.index.MatchCollector;
import com.fasterxml.jackson.annotation.*;
import java.sql.Connection;
import java.sql.PreparedStatement;
import javax.sql.DataSource;
import java.sql.SQLException;
import java.util.*;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MatchCollectorDB extends MatchCollector {

    // Logger
    private final static Logger log = LoggerFactory.getLogger(KorapNode.class);

    /*
     * Todo: In case there are multiple threads searching,
     * the list should be synchrinized Collections.synchronizedList()
     */
    private String databaseType;
    private List matchCollector;
    private int bufferSize, docCollect;
    private String resultID;

    // private Connection connection;
    private DataSource pool;
    private Connection connection;
    private PreparedStatement prepared;

    /*
     * Create a new collector for database connections
     */
    public MatchCollectorDB (int bufferSize, String resultID) {
	this.bufferSize = bufferSize;
	this.resultID = resultID;
	this.matchCollector = new ArrayList<int[]>(bufferSize + 2);
    };

    /*
     * Add matches till the bufferSize exceeds - then commit to the database.
     */
    public void add (int UID, int matchCount) {
	if (this.docCollect == bufferSize)
	    this.commit();

	this.incrTotalResultDocs(1);
	this.incrTotalResults(matchCount);
	this.matchCollector.add(new int[]{UID, matchCount});
	this.docCollect++;
    };

    @JsonIgnore
    public void setDatabaseType (String type) {
	this.databaseType = type;
    };

    @JsonIgnore
    public String getDatabaseType () {
	return this.databaseType;
    };

    @JsonIgnore
    public void setDBPool (String type, DataSource ds, Connection conn) throws SQLException {
	this.setDatabaseType(type);
	this.connection = conn;
	this.pool = ds;
    };


    @JsonIgnore
    public void setDBPool (String type, DataSource ds) throws SQLException {
	this.setDatabaseType(type);
	this.pool = ds;
    };
    /*
      Create prepared statement for multiple requests
      this.prepared = this.conn.prepareStatement(
      "INSERT INTO people VALUES (?, ?);"
      );
      Only prepare if commit > buffersize!
      Difference between mariadb and sqlite!
    */

    
    /* TODO: Ensure the commit was successful! */
    public void commit () {	
	if (this.pool == null)
	    return;

	try {
	    /*
	     * This should be heavily optimized! It's aweful!
	     * ARGHHHHHHH!
	     */

	    if (this.connection.isClosed())
		this.connection = this.pool.getConnection();

	    StringBuilder sb = new StringBuilder();
	    sb.append("INSERT INTO ")
		.append(this.resultID)
		.append(" (text_id, match_count) ");

	    // SQLite batch insertion idiom
	    if (this.getDatabaseType().equals("sqlite")) {
		for (int i = 1; i < this.docCollect; i++) {
		    sb.append("SELECT ?, ? UNION ");
		}
		if (this.docCollect == 1)
		    sb.append("VALUES (?, ?)");
		else
		    sb.append("SELECT ?, ?");
	    }

	    // MySQL batch insertion idiom
	    else if (this.getDatabaseType().equals("mysql")) {
		sb.append(" VALUES ");
		for (int i = 1; i < this.docCollect; i++) {
		    sb.append("(?,?),");
		};
		sb.append("(?,?)");
	    }

	    // Unknown idiom
	    else {
		log.error("Unsupported Database type");
		return;
	    };

	    // Prepare statement based on the string
	    PreparedStatement prep = this.connection.prepareStatement(sb.toString());

	    int i = 1;
	    ListIterator li = this.matchCollector.listIterator(); 
	    while (li.hasNext()) {
		int[] v = (int[]) li.next();
		prep.setInt(i++, v[0]);
		prep.setInt(i++, v[1]);
	    };

	    prep.addBatch();
	    prep.executeBatch();
	    this.connection.commit();
	}

	// An SQL error occured ...
	catch (SQLException e) {
	    log.error(e.getLocalizedMessage());
	};

	this.matchCollector.clear();
	this.docCollect = 0;
	return;
    };

    /*
     * Close collector and connection
     */
    public void close () {
	this.commit();
	try {
	    this.connection.close();
	}
       	catch (SQLException e) {
	    log.warn(e.getLocalizedMessage());
	}
    };

    /*
     * Close collector and probably connection
     */
    public void close (boolean close) {
	if (close)
	    this.close();
	else
	    this.commit();
    };
};