content: /**
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF 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 org.apache.apex.malhar.contrib.misc.streamquery;
  
  import java.sql.Connection;
  import java.sql.DriverManager;
  import java.sql.PreparedStatement;
  import java.sql.ResultSet;
  import java.sql.ResultSetMetaData;
  import java.sql.SQLException;
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  
  import org.apache.apex.malhar.contrib.misc.streamquery.AbstractSqlStreamOperator.InputSchema.ColumnInfo;
  
  import com.datatorrent.api.Context.OperatorContext;
  
  /**
   * An implementation of AbstractSqlStreamOperator that provides embedded derby sql input operator.
   * <p>
   * @displayName Derby Sql Stream
   * @category Stream Manipulators
   * @tags sql, in-memory, input operator
   * @since 0.3.2
   * @deprecated
   */
  @Deprecated
  public class DerbySqlStreamOperator extends AbstractSqlStreamOperator
  {
    protected transient ArrayList<PreparedStatement> insertStatements = new ArrayList<PreparedStatement>(5);
    protected List<String> execStmtStringList = new ArrayList<String>();
    protected transient ArrayList<PreparedStatement> execStatements = new ArrayList<PreparedStatement>(5);
    protected transient ArrayList<PreparedStatement> deleteStatements = new ArrayList<PreparedStatement>(5);
    protected transient Connection db;
  
    public void addExecStatementString(String stmt)
    {
      this.execStmtStringList.add(stmt);
    }
  
  
    @Override
    public void setup(OperatorContext context)
    {
      System.setProperty("derby.stream.error.file", "/dev/null");
      try {
        Class.forName("org.apache.derby.jdbc.EmbeddedDriver").newInstance();
      } catch (Exception ex) {
        throw new RuntimeException(ex);
      }
  
      String connUrl = "jdbc:derby:memory:MALHAR_TEMP;create=true";
      PreparedStatement st;
  
      try {
        db = DriverManager.getConnection(connUrl);
        // create the temporary tables here
        for (int i = 0; i < inputSchemas.size(); i++) {
          InputSchema inputSchema = inputSchemas.get(i);
          if (inputSchema == null || inputSchema.columnInfoMap.isEmpty()) {
            continue;
          }
          String columnSpec = "";
          String columnNames = "";
          String insertQuestionMarks = "";
          int j = 0;
          for (Map.Entry<String, ColumnInfo> entry : inputSchema.columnInfoMap.entrySet()) {
            if (!columnSpec.isEmpty()) {
              columnSpec += ",";
              columnNames += ",";
              insertQuestionMarks += ",";
            }
            columnSpec += entry.getKey();
            columnSpec += " ";
            columnSpec += entry.getValue().type;
            columnNames += entry.getKey();
            insertQuestionMarks += "?";
            entry.getValue().bindIndex = ++j;
          }
          String createTempTableStmt =
              "DECLARE GLOBAL TEMPORARY TABLE SESSION." + inputSchema.name + "(" + columnSpec + ") NOT LOGGED";
          st = db.prepareStatement(createTempTableStmt);
          st.execute();
          st.close();
  
          String insertStmt = "INSERT INTO SESSION." + inputSchema.name + " (" + columnNames + ") VALUES ("
              + insertQuestionMarks + ")";
  
          insertStatements.add(i, db.prepareStatement(insertStmt));
          deleteStatements.add(i, db.prepareStatement("DELETE FROM SESSION." + inputSchema.name));
        }
        for (String stmtStr : execStmtStringList) {
          execStatements.add(db.prepareStatement(stmtStr));
        }
      } catch (SQLException ex) {
        throw new RuntimeException(ex);
      }
    }
  
    @Override
    public void beginWindow(long windowId)
    {
      try {
        db.setAutoCommit(false);
      } catch (SQLException ex) {
        throw new RuntimeException(ex);
      }
    }
  
    @Override
    public void processTuple(int tableNum, HashMap<String, Object> tuple)
    {
      InputSchema inputSchema = inputSchemas.get(tableNum);
  
      PreparedStatement insertStatement = insertStatements.get(tableNum);
      try {
        for (Map.Entry<String, Object> entry : tuple.entrySet()) {
          ColumnInfo t = inputSchema.columnInfoMap.get(entry.getKey());
          if (t != null && t.bindIndex != 0) {
            insertStatement.setString(t.bindIndex, entry.getValue().toString());
          }
        }
  
        insertStatement.executeUpdate();
        insertStatement.clearParameters();
      } catch (SQLException ex) {
        throw new RuntimeException(ex);
      }
    }
  
    @Override
    public void endWindow()
    {
      try {
        db.commit();
        if (bindings != null) {
          for (int i = 0; i < bindings.size(); i++) {
            for (PreparedStatement stmt : execStatements) {
              stmt.setString(i, bindings.get(i).toString());
            }
          }
        }
  
        for (PreparedStatement stmt : execStatements) {
          executePreparedStatement(stmt);
        }
        for (PreparedStatement st : deleteStatements) {
          st.executeUpdate();
          st.clearParameters();
        }
      } catch (SQLException ex) {
        throw new RuntimeException(ex);
      }
      bindings = null;
    }
  
    private void executePreparedStatement(PreparedStatement statement) throws SQLException
    {
      ResultSet res = statement.executeQuery();
      ResultSetMetaData resmeta = res.getMetaData();
      int columnCount = resmeta.getColumnCount();
      while (res.next()) {
        HashMap<String, Object> resultRow = new HashMap<String, Object>();
        for (int i = 1; i <= columnCount; i++) {
          resultRow.put(resmeta.getColumnName(i), res.getObject(i));
        }
        this.result.emit(resultRow);
      }
      statement.clearParameters();
    }
  
    @Override
    public void teardown()
    {
      try {
        db.close();
      } catch (SQLException ex) {
        throw new RuntimeException(ex);
      }
    }
  
  }
  