package org.synchronoss.cpo.jdbc;
   
   /*-
    * [[
    * jdbc
    * ==
    * Copyright (C) 2003 - 2026 Exaxis LLC, Synchronoss Technologies Inc
    * ==
    * This program is free software: you can redistribute it and/or modify
   * it under the terms of the GNU Lesser General Public License as
   * published by the Free Software Foundation, either version 3 of the
   * License, or (at your option) any later version.
   *
   * This program is distributed in the hope that it will be useful,
   * but WITHOUT ANY WARRANTY; without even the implied warranty of
   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   * GNU General Lesser Public License for more details.
   *
   * You should have received a copy of the GNU General Lesser Public
   * License along with this program.  If not, see
   * <http://www.gnu.org/licenses/lgpl-3.0.html>.
   * ]]
   */
  
  import java.lang.reflect.Constructor;
  import java.sql.CallableStatement;
  import java.sql.Connection;
  import java.sql.PreparedStatement;
  import java.sql.ResultSet;
  import java.sql.ResultSetMetaData;
  import java.sql.SQLException;
  import java.sql.Statement;
  import java.sql.Types;
  import java.util.ArrayList;
  import java.util.Collection;
  import java.util.List;
  import java.util.Spliterator;
  import java.util.Spliterators;
  import java.util.function.Consumer;
  import java.util.stream.Stream;
  import java.util.stream.StreamSupport;
  import javax.naming.Context;
  import javax.naming.InitialContext;
  import javax.naming.NamingException;
  import javax.sql.DataSource;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import org.synchronoss.cpo.core.*;
  import org.synchronoss.cpo.core.enums.Comparison;
  import org.synchronoss.cpo.core.enums.Crud;
  import org.synchronoss.cpo.core.enums.Logical;
  import org.synchronoss.cpo.core.helper.ExceptionHelper;
  import org.synchronoss.cpo.core.meta.CpoMetaDescriptor;
  import org.synchronoss.cpo.core.meta.DataTypeMapEntry;
  import org.synchronoss.cpo.core.meta.domain.CpoArgument;
  import org.synchronoss.cpo.core.meta.domain.CpoAttribute;
  import org.synchronoss.cpo.core.meta.domain.CpoClass;
  import org.synchronoss.cpo.core.meta.domain.CpoFunction;
  import org.synchronoss.cpo.jdbc.meta.JdbcCpoMetaDescriptor;
  import org.synchronoss.cpo.jdbc.meta.JdbcMethodMapper;
  import org.synchronoss.cpo.jdbc.meta.JdbcResultSetCpoData;
  
  /**
   * JdbcCpoAdapter is the JDBC implementation of {@link org.synchronoss.cpo.core.CpoAdapter}, a set
   * of routines that are responsible for managing value beans from a jdbc datasource.
   *
   * @author david berry
   */
  public class JdbcCpoAdapter extends CpoBaseAdapter<DataSource> {
  
    /** Version Id for this class. */
    private static final long serialVersionUID = 1L;
  
    private static final Logger logger = LoggerFactory.getLogger(JdbcCpoAdapter.class);
  
    /** The JNDI context used to resolve a {@code jndiName}-configured data source, if any. */
    private transient Context context_ = null;
  
    /** The capabilities of the database behind this adapter, probed once at construction. */
    private final JdbcDatabaseCapabilities capabilities;
  
    /** How this adapter obtains and releases its connections; see JdbcConnectionStrategy. */
    private transient JdbcConnectionStrategy connectionStrategy;
  
    /** CpoMetaDescriptor allows you to get the metadata for a class. */
    private JdbcCpoMetaDescriptor metaDescriptor = null;
  
    /**
     * Creates a JdbcCpoAdapter.
     *
     * @param metaDescriptor This datasource that identifies the cpo metadata datasource
     * @param jdsiTrx The datasoruce that identifies the transaction database.
     * @throws CpoException exception
     */
    protected JdbcCpoAdapter(JdbcCpoMetaDescriptor metaDescriptor, DataSourceInfo<DataSource> jdsiTrx)
        throws CpoException {
      super(jdsiTrx.getDataSourceName(), jdsiTrx.getFetchSize(), jdsiTrx.getBatchSize());
      this.metaDescriptor = metaDescriptor;
      setWriteDataSource(jdsiTrx.getDataSource());
     setReadDataSource(jdsiTrx.getDataSource());
     this.connectionStrategy =
         new JdbcPooledConnectionStrategy(getReadDataSource(), getWriteDataSource());
     this.capabilities = JdbcDatabaseCapabilities.probe(connectionStrategy);
   }
 
   /**
    * Creates a JdbcCpoAdapter.
    *
    * @param metaDescriptor This datasource that identifies the cpo metadata datasource
    * @param jdsiWrite The datasource that identifies the transaction database for write
    *     transactions.
    * @param jdsiRead The datasource that identifies the transaction database for read-only
    *     transactions.
    * @throws CpoException exception
    */
   protected JdbcCpoAdapter(
       JdbcCpoMetaDescriptor metaDescriptor,
       DataSourceInfo<DataSource> jdsiWrite,
       DataSourceInfo<DataSource> jdsiRead)
       throws CpoException {
     super(jdsiWrite.getDataSourceName(), jdsiWrite.getFetchSize(), jdsiWrite.getBatchSize());
     this.metaDescriptor = metaDescriptor;
     setWriteDataSource(jdsiWrite.getDataSource());
     setReadDataSource(jdsiRead.getDataSource());
     this.connectionStrategy =
         new JdbcPooledConnectionStrategy(getReadDataSource(), getWriteDataSource());
     this.capabilities = JdbcDatabaseCapabilities.probe(connectionStrategy);
   }
 
   /**
    * This constructor is used specifically to clone a JdbcCpoAdapter.
    *
    * @param jdbcCpoAdapter - the JdbcCpoAdapter to clone
    * @throws CpoException An exception occurred copying the datasource.
    */
   protected JdbcCpoAdapter(JdbcCpoAdapter jdbcCpoAdapter) throws CpoException {
     super(
         jdbcCpoAdapter.getDataSourceName(),
         jdbcCpoAdapter.getFetchSize(),
         jdbcCpoAdapter.getBatchSize());
     this.metaDescriptor = (JdbcCpoMetaDescriptor) jdbcCpoAdapter.getCpoMetaDescriptor();
     this.capabilities = jdbcCpoAdapter.capabilities;
     setWriteDataSource(jdbcCpoAdapter.getWriteDataSource());
     setReadDataSource(jdbcCpoAdapter.getReadDataSource());
     this.connectionStrategy =
         new JdbcPooledConnectionStrategy(getReadDataSource(), getWriteDataSource());
   }
 
   /**
    * Gets the connection strategy this adapter uses
    *
    * @return The connection strategy
    */
   JdbcConnectionStrategy getConnectionStrategy() {
     return connectionStrategy;
   }
 
   /**
    * Replaces the connection strategy; used by JdbcCpoTrxAdapter to pin a transaction connection
    *
    * @param connectionStrategy The connection strategy to use
    */
   void setConnectionStrategy(JdbcConnectionStrategy connectionStrategy) {
     this.connectionStrategy = connectionStrategy;
   }
 
   /**
    * Are batch updates supported
    *
    * @return true if batch updates are supported
    */
   protected boolean isBatchUpdatesSupported() {
     return capabilities.batchUpdatesSupported();
   }
 
   /**
    * Finds or Creates a JdbcCpoAdapter.
    *
    * @param metaDescriptor - The meta descriptor for the datasource
    * @param jdsiTrx - The datasurce info
    * @return A JdbcCpoAdapter
    * @throws CpoException - An error occurred finding of creating the datasource adapter
    */
   public static JdbcCpoAdapter getInstance(
       JdbcCpoMetaDescriptor metaDescriptor, DataSourceInfo<DataSource> jdsiTrx)
       throws CpoException {
     String adapterKey = metaDescriptor + ":" + jdsiTrx.getDataSourceName();
     JdbcCpoAdapter adapter = (JdbcCpoAdapter) findCpoAdapter(adapterKey);
     if (adapter == null) {
       adapter = new JdbcCpoAdapter(metaDescriptor, jdsiTrx);
       addCpoAdapter(adapterKey, adapter);
     }
     return adapter;
   }
 
   /**
    * Creates a JdbcCpoAdapter.
    *
    * @param metaDescriptor This datasource that identifies the cpo metadata datasource
    * @param jdsiWrite The datasource that identifies the transaction database for write
    *     transactions.
    * @param jdsiRead The datasource that identifies the transaction database for read-only
    *     transactions.
    * @return - A JdbcCpoAdapter
    * @throws CpoException exception
    */
   public static JdbcCpoAdapter getInstance(
       JdbcCpoMetaDescriptor metaDescriptor,
       DataSourceInfo<DataSource> jdsiWrite,
       DataSourceInfo<DataSource> jdsiRead)
       throws CpoException {
     String adapterKey =
         metaDescriptor + ":" + jdsiWrite.getDataSourceName() + ":" + jdsiRead.getDataSourceName();
     JdbcCpoAdapter adapter = (JdbcCpoAdapter) findCpoAdapter(adapterKey);
     if (adapter == null) {
       adapter = new JdbcCpoAdapter(metaDescriptor, jdsiWrite, jdsiRead);
       addCpoAdapter(adapterKey, adapter);
     }
     return adapter;
   }
 
   @Override
   public <T> long existsBean(CpoQuery query, T bean) throws CpoException {
     Connection c = null;
     long objCount = -1;
 
     try {
       c = getReadConnection();
 
       objCount = existsBean(query.groupName(), bean, c, query.wheres());
     } finally {
       // The read connection runs with autocommit off (required for cursor-based fetch), so
       // this read-only transaction must still be terminated before the connection returns
       // to the pool. Otherwise the open transaction pins a stale MVCC snapshot (and holds
       // any FOR UPDATE locks) for the next borrower. Rollback is used because no data was
       // changed; it releases the snapshot and locks just like commit would.
       rollbackLocalConnection(c);
       closeLocalConnection(c);
     }
 
     return objCount;
   }
 
   /**
    * The CpoAdapter will check to see if this bean exists in the datasource.
    *
    * @param <T> The type of the bean
    * @param groupName The groupName which identifies which EXISTS, INSERT, and UPDATE Function
    *     Groups to execute to upsert the bean.
    * @param bean This is a bean that has been defined within the metadata of the datasource. If the
    *     class is not defined an exception will be thrown.
    * @param con The datasource Connection with which to check if the bean exists
    * @param wheres A collection of where clauses
    * @return The number of matching beans: for a single-row result with one numeric column the
    *     column value is returned (count(*) style), otherwise the number of rows returned
    * @throws CpoException exception will be thrown if the Function Group has a function count != 1
    */
   protected <T> long existsBean(
       String groupName, T bean, Connection con, Collection<CpoWhere> wheres) throws CpoException {
     PreparedStatement ps = null;
     ResultSet rs = null;
     ResultSetMetaData rsmd;
     CpoClass cpoClass;
     long objCount = 0;
     Logger localLogger = logger;
 
     if (bean == null) {
       throw new CpoException("NULL Bean passed into existsBean");
     }
 
     try {
       cpoClass = metaDescriptor.getMetaClass(bean);
       List<CpoFunction> cpoFunctions =
           cpoClass.getFunctionGroup(Crud.EXIST, groupName).getFunctions();
       localLogger = LoggerFactory.getLogger(cpoClass.getMetaClass());
 
       for (CpoFunction cpoFunction : cpoFunctions) {
         localLogger.info(cpoFunction.getExpression());
         JdbcPreparedStatementFactory jpsf =
             new JdbcPreparedStatementFactory(
                 con, this, cpoClass, cpoFunction, bean, wheres, null, null);
         ps = jpsf.getPreparedStatement();
 
         long qCount = 0; // set the results for this function to 0
 
         rs = ps.executeQuery();
         jpsf.release();
         rsmd = rs.getMetaData();
 
         // A single-row result with one numeric column is interpreted as a count(*)
         // style EXIST function; everything else is counted row by row.
         if (rsmd.getColumnCount() == 1 && isCountResult(rsmd.getColumnType(1)) && rs.next()) {
           qCount = rs.getLong(1); // the number of beans that exist
           if (rs.next()) {
             // EXIST function has more than one record so not a count(*)
             qCount = 2;
           }
         }
 
         while (rs.next()) {
           qCount++;
         }
 
         objCount += qCount;
 
         rs.close();
         ps.close();
         rs = null;
         ps = null;
       }
     } catch (SQLException e) {
       String msg = "existsBean(groupName, bean, con) failed:";
       throw new CpoException(msg, e);
     } finally {
       resultSetClose(rs);
       statementClose(ps);
     }
 
     return objCount;
   }
 
   private static boolean isCountResult(int sqlType) {
     return switch (sqlType) {
       case Types.BIGINT,
           Types.INTEGER,
           Types.NUMERIC,
           Types.DECIMAL,
           Types.SMALLINT,
           Types.TINYINT ->
           true;
       default -> false;
     };
   }
 
   @Override
   public CpoWhere newWhere() {
     return new JdbcCpoWhere();
   }
 
   @Override
   public <T> CpoWhere newWhere(Logical logical, String attr, Comparison comp, T value) {
     return new JdbcCpoWhere(logical, attr, comp, value);
   }
 
   @Override
   public <T> CpoWhere newWhere(
       Logical logical, String attr, Comparison comp, T value, boolean not) {
     return new JdbcCpoWhere(logical, attr, comp, value, not);
   }
 
   /**
    * Sets the JNDI context
    *
    * @param context The JNDI Context
    * @throws CpoException An exception setting the context
    */
   protected void setContext(Context context) throws CpoException {
     try {
       if (context == null) {
         context_ = new InitialContext();
       } else {
         context_ = context;
       }
     } catch (NamingException e) {
       throw new CpoException("Error setting Context", e);
     }
   }
 
   /**
    * Gets the JNDI Context
    *
    * @return The current context
    */
   protected Context getContext() {
     return context_;
   }
 
   /**
    * Validates the crud of query being performed. If it is an UPSERT Group, it checks the database
    * to see if this is an update or an insert, and returns the query group. Otherwise, it sends back
    * the original query group. Upserts only work for single beans.
    *
    * @param <T> The crud of the bean
    * @param bean The bean to insert or update
    * @param crud The group crud
    * @param groupName The groupName
    * @param c The connection to use
    * @return The crud operation to use
    * @throws CpoException An exception occurred
    */
   protected <T> Crud adjustCrud(T bean, Crud crud, String groupName, Connection c)
       throws CpoException {
     Crud retType = crud;
     long objCount;
 
     if (Crud.UPSERT == retType) {
       objCount = existsBean(groupName, bean, c, null);
 
       if (objCount == 0) {
         retType = Crud.CREATE;
       } else if (objCount == 1) {
         retType = Crud.UPDATE;
       } else {
         throw new CpoException(
             "UPSERT can only UPDATE one record. Your EXISTS function returned 2 or more.");
       }
     }
 
     return retType;
   }
 
   /**
    * Gets the read connection for this CpoAdapter
    *
    * @return A read connection
    * @throws CpoException An error has occurred.
    */
   protected Connection getReadConnection() throws CpoException {
     return connectionStrategy.getReadConnection();
   }
 
   /**
    * Gets the write connection for this CpoAdapter
    *
    * @return A write connection
    * @throws CpoException An error has occurred.
    */
   protected Connection getWriteConnection() throws CpoException {
     return connectionStrategy.getWriteConnection();
   }
 
   /**
    * Closes the local connection.
    *
    * @param connection The connection to close
    */
   protected void closeLocalConnection(Connection connection) {
     connectionStrategy.closeLocalConnection(connection);
   }
 
   /**
    * Commits the local connection
    *
    * @param connection The connection to commit
    */
   protected void commitLocalConnection(Connection connection) {
     connectionStrategy.commitLocalConnection(connection);
   }
 
   /**
    * Rollback the local connection
    *
    * @param connection The connection to rollback
    */
   protected void rollbackLocalConnection(Connection connection) {
     connectionStrategy.rollbackLocalConnection(connection);
   }
 
   @Override
   protected <T, C> T processExecuteGroup(String groupName, C criteria, T result)
       throws CpoException {
     Connection c = null;
     T bean = null;
 
     try {
       c = getWriteConnection();
       bean = processExecuteGroup(groupName, criteria, result, c);
       commitLocalConnection(c);
     } catch (Exception e) {
       // Any exception has to try to rollback the work;
       rollbackLocalConnection(c);
       ExceptionHelper.reThrowCpoException(
           e, "processExecuteGroup(String groupName, C criteria, T result) failed");
     } finally {
       closeLocalConnection(c);
     }
 
     return bean;
   }
 
   /**
    * Executes a Bean whose MetaData contains a stored procedure. An assumption is that the bean
    * exists in the datasource.
    *
    * @param <T> The result bean type
    * @param <C> The criteria bean type
    * @param groupName The filter groupName which tells the datasource which beans should be
    *     returned. The groupName also signifies what data in the bean will be populated.
    * @param criteria This is a bean that has been defined within the metadata of the datasource. If
    *     the class is not defined an exception will be thrown. If the bean does not exist in the
    *     datasource, an exception will be thrown. This bean is used to populate the IN arguments
    *     used to retrieve the collection of beans.
    * @param conn The connection to use to execute this group
    * @param result This is a bean that has been defined within the metadata of the datasource. If
    *     the class is not defined an exception will be thrown. If the bean does not exist in the
    *     datasource, an exception will be thrown. This bean defines the bean type that will be
    *     returned in the
    * @return A result bean populate with the OUT arguments
    * @throws CpoException if the criteria or result bean's metadata cannot be resolved, the callable
    *     statement fails to execute, or the result bean cannot be instantiated
    */
   protected <T, C> T processExecuteGroup(String groupName, C criteria, T result, Connection conn)
       throws CpoException {
     CallableStatement cstmt = null;
     CpoClass criteriaClass;
     CpoClass resultClass;
     T returnBean = null;
     Logger localLogger = criteria == null ? logger : LoggerFactory.getLogger(criteria.getClass());
 
     JdbcCallableStatementFactory jcsf = null;
 
     if (criteria == null || result == null) {
       throw new CpoException("NULL Bean passed into executeBean");
     }
 
     try {
       criteriaClass = metaDescriptor.getMetaClass(criteria);
       resultClass = metaDescriptor.getMetaClass(result);
 
       List<CpoFunction> functions =
           criteriaClass.getFunctionGroup(Crud.EXECUTE, groupName).getFunctions();
       localLogger.info(
           "===================processExecuteGroup ("
               + groupName
               + ") Count<"
               + functions.size()
               + ">=========================");
 
       try {
         returnBean = (T) result.getClass().getDeclaredConstructor().newInstance();
       } catch (IllegalAccessException iae) {
         throw new CpoException("Unable to access the constructor of the Return Bean", iae);
       } catch (InstantiationException iae) {
         throw new CpoException("Unable to instantiate Return Bean", iae);
       }
 
       // Loop through the queries and process each one
       for (CpoFunction function : functions) {
 
         localLogger.debug("Executing Call:" + criteriaClass.getName() + ":" + groupName);
 
         jcsf = new JdbcCallableStatementFactory(conn, this, function, criteria, resultClass);
         cstmt = jcsf.getCallableStatement();
         cstmt.execute();
         jcsf.release();
 
         localLogger.debug("Processing Call:" + criteriaClass.getName() + ":" + groupName);
 
         // Todo: Add Code here to go through the arguments, find record sets,
         // and process them
         // Process the non-record set out params and make it the first
         // bean in the collection
 
         // Loop through the OUT Parameters and set them in the result
         // bean
         int j = 1;
         for (CpoArgument cpoArgument : jcsf.getOutArguments()) {
           JdbcCpoArgument jdbcArgument = (JdbcCpoArgument) cpoArgument;
           if (jdbcArgument.isOutParameter()) {
             JdbcCpoAttribute jdbcAttribute = jdbcArgument.getAttribute();
             if (jdbcAttribute == null) {
               jdbcAttribute =
                   (JdbcCpoAttribute) resultClass.getAttributeJava(jdbcArgument.getName());
               if (jdbcAttribute == null) {
                 throw new CpoException(
                     "Attribute <"
                         + jdbcArgument.getName()
                         + "> does not exist on class <"
                         + resultClass.getName()
                         + ">");
               }
             }
             jdbcAttribute.invokeSetter(
                 returnBean, new CallableStatementCpoData(cstmt, jdbcAttribute, j));
           }
           j++;
         }
 
         cstmt.close();
       }
     } catch (Throwable t) {
       String msg =
           "processExecuteGroup(String groupName, C criteria, T result, Connection conn)"
               + " failed. SQL=";
       localLogger.error(msg, t);
       throw new CpoException(msg, t);
     } finally {
       statementClose(cstmt);
       if (jcsf != null) {
         jcsf.release();
       }
     }
 
     return returnBean;
   }
 
   @Override
   protected <T> T processSelectGroup(
       T bean,
       String groupName,
       Collection<CpoWhere> wheres,
       Collection<CpoOrderBy> orderBy,
       Collection<CpoNativeFunction> nativeExpressions)
       throws CpoException {
     Connection c = null;
     T result = null;
 
     try {
       c = getReadConnection();
       result = processSelectGroup(bean, groupName, wheres, orderBy, nativeExpressions, c);
 
       // The select may have a for update clause on it
       // Since the connection is cached we need to get rid of this
       commitLocalConnection(c);
     } catch (Exception e) {
       // Any exception has to try to rollback the work;
       rollbackLocalConnection(c);
       ExceptionHelper.reThrowCpoException(e, "processSelectGroup(T bean, String groupName) failed");
     } finally {
       closeLocalConnection(c);
     }
 
     return result;
   }
 
   /**
    * Retrieves the Bean from the datasource.
    *
    * @param <T> The bean type
    * @param bean This is a bean that has been defined within the metadata of the datasource. If the
    *     class is not defined an exception will be thrown. The input bean is used to specify the
    *     search criteria.
    * @param groupName The name which identifies which RETRIEVE Function Group to execute to retrieve
    *     the bean.
    * @param wheres A collection of CpoWhere beans to be used by the function
    * @param orderBy A collection of CpoOrderBy beans to be used by the function
    * @param nativeExpressions A collection of CpoNativeFunction beans to be used by the function
    * @param con The connection to use for this select
    * @return A populated bean of the same type as the Bean passed in as a argument. If no beans
    *     match the criteria a NULL will be returned.
    * @throws CpoException the retrieve function defined for this beans returns more than one row, an
    *     exception will be thrown.
    */
   protected <T> T processSelectGroup(
       T bean,
       String groupName,
       Collection<CpoWhere> wheres,
       Collection<CpoOrderBy> orderBy,
       Collection<CpoNativeFunction> nativeExpressions,
       Connection con)
       throws CpoException {
     PreparedStatement ps = null;
     ResultSet rs = null;
     ResultSetMetaData rsmd;
     CpoClass cpoClass;
     JdbcCpoAttribute attribute;
     T criteriaObj = bean;
     boolean recordsExist = false;
     Logger localLogger = bean == null ? logger : LoggerFactory.getLogger(bean.getClass());
 
     int recordCount = 0;
     int attributesSet = 0;
 
     int k;
     T rObj = null;
 
     if (bean == null) {
       throw new CpoException("NULL Bean passed into retrieveBean");
     }
 
     try {
       cpoClass = metaDescriptor.getMetaClass(criteriaObj);
       List<CpoFunction> functions =
           cpoClass.getFunctionGroup(Crud.RETRIEVE, groupName).getFunctions();
 
       localLogger.info(buildCpoClassLogLine(criteriaObj.getClass(), Crud.RETRIEVE, groupName));
 
       try {
         rObj = (T) bean.getClass().getDeclaredConstructor().newInstance();
       } catch (IllegalAccessException iae) {
         localLogger.error(
             "=================== Could not access default constructor for Class=<"
                 + bean.getClass()
                 + "> ==================");
         throw new CpoException("Unable to access the constructor of the Return Bean", iae);
       } catch (InstantiationException iae) {
         throw new CpoException("Unable to instantiate Return Bean", iae);
       }
 
       for (CpoFunction cpoFunction : functions) {
 
         JdbcPreparedStatementFactory jpsf =
             new JdbcPreparedStatementFactory(
                 con, this, cpoClass, cpoFunction, criteriaObj, wheres, orderBy, nativeExpressions);
         ps = jpsf.getPreparedStatement();
 
         // insertions on
         // selectgroup
         rs = ps.executeQuery();
         jpsf.release();
 
         if (rs.isBeforeFirst()) {
           rsmd = rs.getMetaData();
 
           if ((rsmd.getColumnCount() == 2)
               && "CPO_ATTRIBUTE".equalsIgnoreCase(rsmd.getColumnLabel(1))
               && "CPO_VALUE".equalsIgnoreCase(rsmd.getColumnLabel(2))) {
             while (rs.next()) {
               recordsExist = true;
               recordCount++;
               attribute = (JdbcCpoAttribute) cpoClass.getAttributeData(rs.getString(1));
 
               if (attribute != null) {
                 attribute.invokeSetter(
                     rObj,
                     new JdbcResultSetCpoData(JdbcMethodMapper.getMethodMapper(), rs, attribute, 2));
                 attributesSet++;
               }
             }
           } else if (rs.next()) {
             recordsExist = true;
             recordCount++;
             for (k = 1; k <= rsmd.getColumnCount(); k++) {
               attribute = (JdbcCpoAttribute) cpoClass.getAttributeData(rsmd.getColumnLabel(k));
 
               if (attribute != null) {
                 attribute.invokeSetter(
                     rObj,
                     new JdbcResultSetCpoData(JdbcMethodMapper.getMethodMapper(), rs, attribute, k));
                 attributesSet++;
               }
             }
 
             if (rs.next()) {
               String msg =
                   "processSelectGroup(T bean, String groupName) failed: Multiple Records Returned";
               localLogger.error(msg);
               throw new CpoException(msg);
             }
           }
           criteriaObj = rObj;
         }
 
         rs.close();
         rs = null;
         ps.close();
         ps = null;
       }
 
       if (!recordsExist) {
         rObj = null;
         localLogger.info(
             buildRecordsLogLine(0, 0, criteriaObj.getClass(), Crud.RETRIEVE, groupName));
       } else {
         localLogger.info(
             buildRecordsLogLine(
                 recordCount, attributesSet, criteriaObj.getClass(), Crud.RETRIEVE, groupName));
       }
     } catch (Throwable t) {
       String msg = "processSeclectGroup(T bean) failed: " + ExceptionHelper.getLocalizedMessage(t);
       localLogger.error(msg, t);
       rObj = null;
       throw new CpoException(msg, t);
     } finally {
       resultSetClose(rs);
       statementClose(ps);
     }
 
     return rObj;
   }
 
   @Override
   protected <T, C> Stream<T> processSelectGroup(
       String groupName,
       C criteria,
       T result,
       Collection<CpoWhere> wheres,
       Collection<CpoOrderBy> orderBy,
       Collection<CpoNativeFunction> nativeExpressions,
       boolean useRetrieve)
       throws CpoException {
     Connection con = null;
 
     try {
       con = getReadConnection();
       return processSelectGroup(
           groupName, criteria, result, wheres, orderBy, nativeExpressions, con, useRetrieve);
     } catch (Exception e) {
       // Any exception has to try to rollback the work;
       rollbackLocalConnection(con);
       closeLocalConnection(con);
       ExceptionHelper.reThrowCpoException(
           e,
           "processSelectGroup(String groupName, C criteria, T result,C poWhere where,"
               + " Collection orderBy, boolean useRetrieve) failed");
     }
     return Stream.empty();
   }
 
   /**
    * Retrieves Beans from the datasource.
    *
    * @param <T> The result bean type
    * @param <C> The criteria bean type
    * @param groupName Query group groupName
    * @param criteria The criteria bean
    * @param result The result bean
    * @param wheres A collection of CpoWhere beans to be used by the function
    * @param orderBy A collection of CpoOrderBy beans to be used by the function
    * @param nativeExpressions A collection of CpoNativeFunction beans to be used by the function
    * @param useRetrieve Use the RETRIEVE_GROUP instead of the LIST_GROUP
    * @param con The connection to use for this select
    * @return A stream of T
    * @throws CpoException Any errors retrieving the data from the datasource
    */
   protected <T, C> Stream<T> processSelectGroup(
       String groupName,
       C criteria,
       T result,
       Collection<CpoWhere> wheres,
       Collection<CpoOrderBy> orderBy,
       Collection<CpoNativeFunction> nativeExpressions,
       Connection con,
       boolean useRetrieve)
       throws CpoException {
     Logger localLogger = criteria == null ? logger : LoggerFactory.getLogger(criteria.getClass());
     PreparedStatement ps = null;
     List<CpoFunction> cpoFunctions;
     CpoClass criteriaClass;
     CpoClass resultClass;
     ResultSet rs = null;
     ResultSetMetaData rsmd;
     int columnCount;
     JdbcCpoAttribute[] attributes;
     JdbcPreparedStatementFactory jpsf;
 
     if (criteria == null || result == null) {
       throw new CpoException("NULL Bean passed into retrieveBean or retrieveBeans");
     }
 
     try {
       criteriaClass = metaDescriptor.getMetaClass(criteria);
       resultClass = metaDescriptor.getMetaClass(result);
       if (useRetrieve) {
         localLogger.info(buildCpoClassLogLine(criteria.getClass(), Crud.RETRIEVE, groupName));
         cpoFunctions = criteriaClass.getFunctionGroup(Crud.RETRIEVE, groupName).getFunctions();
       } else {
         localLogger.info(buildCpoClassLogLine(criteria.getClass(), Crud.LIST, groupName));
         cpoFunctions = criteriaClass.getFunctionGroup(Crud.LIST, groupName).getFunctions();
       }
 
       CpoFunction cpoFunction = cpoFunctions.getFirst();
       jpsf =
           new JdbcPreparedStatementFactory(
               con, this, criteriaClass, cpoFunction, criteria, wheres, orderBy, nativeExpressions);
       ps = jpsf.getPreparedStatement();
 
       localLogger.debug("Retrieving Records");
 
       rs = ps.executeQuery();
       jpsf.release();
 
       localLogger.debug("Processing Records");
 
       rsmd = rs.getMetaData();
 
       columnCount = rsmd.getColumnCount();
 
       attributes = new JdbcCpoAttribute[columnCount + 1];
 
       for (int k = 1; k <= columnCount; k++) {
         attributes[k] = (JdbcCpoAttribute) resultClass.getAttributeData(rsmd.getColumnLabel(k));
       }
 
       // resolved once per query, not once per row: the bean constructor and one reusable
       // data wrapper per column (the wrapper's result set, attribute, and index are fixed)
       Constructor<?> resultConstructor = result.getClass().getDeclaredConstructor();
       JdbcResultSetCpoData[] columnData = new JdbcResultSetCpoData[columnCount + 1];
       for (int k = 1; k <= columnCount; k++) {
         if (attributes[k] != null) {
           columnData[k] =
               new JdbcResultSetCpoData(JdbcMethodMapper.getMethodMapper(), rs, attributes[k], k);
         }
       }
 
       ResultSet finalRs = rs;
       PreparedStatement finalPs = ps;
       return StreamSupport.stream(
               new Spliterators.AbstractSpliterator<T>(Long.MAX_VALUE, Spliterator.ORDERED) {
                 @Override
                 public boolean tryAdvance(Consumer<? super T> action) {
                   try {
                     if (!finalRs.next()) return false;
                     T bean = null;
                     try {
                       bean = (T) resultConstructor.newInstance();
                     } catch (IllegalAccessException iae) {
                       localLogger.error(
                           "=================== Could not access default constructor for Class=<"
                               + result.getClass()
                               + "> ==================");
                       throw new CpoException(
                           "Unable to access the constructor of the Return Bean", iae);
                     } catch (InstantiationException iae) {
                       throw new CpoException("Unable to instantiate Return Bean", iae);
                     }
 
                     for (int k = 1; k <= columnCount; k++) {
                       if (columnData[k] != null) {
                         attributes[k].invokeSetter(bean, columnData[k]);
                       }
                     }
                     action.accept(bean);
                     return true;
                   } catch (Exception ex) {
                     throw new RuntimeException(ex);
                   }
                 }
               },
               false)
           .onClose(
               () -> {
                 try {
                   // The select may have a for update clause on it
                   // Since the connection is cached we need to get rid of this
                   resultSetClose(finalRs);
                   statementClose(finalPs);
                   commitLocalConnection(con);
                   closeLocalConnection(con);
                 } catch (Exception e) {
                   resultSetClose(finalRs);
                   statementClose(finalPs);
                   rollbackLocalConnection(con);
                   closeLocalConnection(con);
                   throw new RuntimeException(e);
                 }
               });
     } catch (Throwable t) {
       resultSetClose(rs);
       statementClose(ps);
       rollbackLocalConnection(con);
       closeLocalConnection(con);
       String msg =
           "processSelectGroup(String groupName, C criteria, T result, CpoWhere where,"
               + " Collection orderBy, Connection con) failed. Error:";
       localLogger.error(msg, t);
       throw new CpoException(msg, t);
     }
   }
 
   @Override
   protected <T> long processUpdateGroup(
       T bean,
       Crud crud,
       String groupName,
       Collection<CpoWhere> wheres,
       Collection<CpoOrderBy> orderBy,
       Collection<CpoNativeFunction> nativeExpressions)
       throws CpoException {
     Connection c = null;
     long updateCount = 0;
 
     try {
       c = getWriteConnection();
       updateCount =
           processUpdateGroup(bean, crud, groupName, wheres, orderBy, nativeExpressions, c);
       commitLocalConnection(c);
     } catch (Exception e) {
       // Any exception has to try to rollback the work;
       rollbackLocalConnection(c);
       ExceptionHelper.reThrowCpoException(
           e, "processUdateGroup(T bean, String crud, String groupName) failed");
     } finally {
       closeLocalConnection(c);
     }
 
     return updateCount;
   }
 
   /**
    * Updates beans in the datasource
    *
    * @param <T> The bean type
    * @param bean The bean instance
    * @param crud The query group type
    * @param groupName The query group type
    * @param wheres A collection of CpoWhere beans to be used by the function
    * @param orderBy A collection of CpoOrderBy beans to be used by the function
    * @param nativeExpressions A collection of CpoNativeFunction beans to be used by the function
    * @param con The connection to use for the update
    * @return The number of records updated
    * @throws CpoException any errors processing the update
    */
   protected <T> long processUpdateGroup(
       T bean,
       Crud crud,
       String groupName,
       Collection<CpoWhere> wheres,
       Collection<CpoOrderBy> orderBy,
       Collection<CpoNativeFunction> nativeExpressions,
       Connection con)
       throws CpoException {
     Logger localLogger = bean == null ? logger : LoggerFactory.getLogger(bean.getClass());
     CpoClass cpoClass;
     PreparedStatement ps = null;
 
     JdbcPreparedStatementFactory jpsf = null;
     long updateCount = 0;
 
     if (bean == null) {
       throw new CpoException(
           "NULL Bean passed into insertBean, deleteBean, updateBean, or upsertBean");
     }
 
     try {
       cpoClass = metaDescriptor.getMetaClass(bean);
       List<CpoFunction> cpoFunctions =
           cpoClass
               .getFunctionGroup(adjustCrud(bean, crud, groupName, con), groupName)
               .getFunctions();
       localLogger.info(buildCpoClassLogLine(bean.getClass(), crud, groupName));
 
       int numRows = 0;
 
       for (CpoFunction cpoFunction : cpoFunctions) {
         jpsf =
             new JdbcPreparedStatementFactory(
                 con, this, cpoClass, cpoFunction, bean, wheres, orderBy, nativeExpressions);
         ps = jpsf.getPreparedStatement();
         numRows += ps.executeUpdate();
         jpsf.release();
         ps.close();
       }
       localLogger.info(buildUpdatesLogLine(numRows, bean.getClass(), crud, groupName));
 
       if (numRows > 0) {
         updateCount++;
       }
     } catch (Throwable t) {
       String msg =
           "ProcessUpdateGroup failed:"
               + crud.operation
               + ","
               + groupName
               + ","
               + bean.getClass().getName();
       // TODO FIX THIS
       // localLogger.error("bound values:" + this.parameterToString(jq));
       localLogger.error(msg, t);
       throw new CpoException(msg, t);
     } finally {
       statementClose(ps);
       if (jpsf != null) {
         jpsf.release();
       }
     }
 
     return updateCount;
   }
 
   /**
    * Updates beans in the datasource
    *
    * @param <T> The bean type
    * @param beans The list of T to update
    * @param crud The query group type
    * @param groupName The query group type
    * @param wheres A collection of CpoWhere beans to be used by the function
    * @param orderBy A collection of CpoOrderBy beans to be used by the function
    * @param nativeExpressions A collection of CpoNativeFunction beans to be used by the function
    * @param con The connection to use for the update
    * @return The number of records updated
    * @throws CpoException any errors processing the update
    */
   protected <T> long processBatchUpdateGroup(
       List<T> beans,
       Crud crud,
       String groupName,
       Collection<CpoWhere> wheres,
       Collection<CpoOrderBy> orderBy,
       Collection<CpoNativeFunction> nativeExpressions,
       Connection con)
       throws CpoException {
     CpoClass jmc;
     List<CpoFunction> cpoFunctions;
     PreparedStatement ps = null;
     CpoFunction cpoFunction;
     JdbcPreparedStatementFactory jpsf = null;
     long updateCount = 0;
     Logger localLogger = logger;
 
     if (beans.isEmpty()) return updateCount;
 
     T firstBean = beans.getFirst();
     try {
       jmc = metaDescriptor.getMetaClass(firstBean);
       cpoFunctions =
           jmc.getFunctionGroup(adjustCrud(firstBean, crud, groupName, con), groupName)
               .getFunctions();
       localLogger = LoggerFactory.getLogger(jmc.getMetaClass());
 
       // Only Batch if there is only one function
       if (cpoFunctions.size() == 1) {
         localLogger.info(buildBatchLogLine(firstBean.getClass(), crud, groupName));
         cpoFunction = cpoFunctions.get(0);
         jpsf =
             new JdbcPreparedStatementFactory(
                 con, this, jmc, cpoFunction, firstBean, wheres, orderBy, nativeExpressions);
         updateCount =
             new JdbcBatchExecutor(getBatchSize()).executeBatchedUpdates(jpsf, cpoFunction, beans);
         jpsf.release();
         localLogger.info(buildUpdatesLogLine(updateCount, firstBean.getClass(), crud, groupName));
       } else {
         localLogger.info(buildCpoClassLogLine(firstBean.getClass(), crud, groupName));
         for (T bean : beans) {
           for (CpoFunction function : cpoFunctions) {
             jpsf =
                 new JdbcPreparedStatementFactory(
                     con, this, jmc, function, bean, wheres, orderBy, nativeExpressions);
             ps = jpsf.getPreparedStatement();
             updateCount += ps.executeUpdate();
             jpsf.release();
             ps.close();
           }
         }
         localLogger.info(buildUpdatesLogLine(updateCount, firstBean.getClass(), crud, groupName));
       }
 
     } catch (Throwable t) {
       String msg =
           "ProcessUpdateGroup failed:"
               + crud.operation
               + ","
               + groupName
               + ","
               + firstBean.getClass().getName();
       // TODO FIX This
       // localLogger.error("bound values:" + this.parameterToString(jq));
       localLogger.error(msg, t);
       throw new CpoException(msg, t);
     } finally {
       statementClose(ps);
       if (jpsf != null) {
         jpsf.release();
       }
     }
 
     return updateCount;
   }
 
   @Override
   protected <T> long processUpdateGroup(
       List<T> beans,
       Crud crud,
       String groupName,
       Collection<CpoWhere> wheres,
       Collection<CpoOrderBy> orderBy,
       Collection<CpoNativeFunction> nativeExpressions)
       throws CpoException {
     Connection c = null;
     long updateCount = 0;
 
     try {
       c = getWriteConnection();
 
       updateCount =
           processUpdateGroup(beans, crud, groupName, wheres, orderBy, nativeExpressions, c);
       commitLocalConnection(c);
     } catch (Exception e) {
       // Any exception has to try to rollback the work;
       rollbackLocalConnection(c);
       ExceptionHelper.reThrowCpoException(
           e, "processUpdateGroup(Collection beans, String crud, String groupName) failed");
     } finally {
       closeLocalConnection(c);
     }
 
     return updateCount;
   }
 
   /**
    * Updates beans in the datasource
    *
    * @param <T> The bean type
    * @param beans The collection of T to update
    * @param crud The query group type
    * @param groupName The query group type
    * @param wheres A collection of CpoWhere beans to be used by the function
    * @param orderBy A collection of CpoOrderBy beans to be used by the function
    * @param nativeExpressions A collection of CpoNativeFunction beans to be used by the function
    * @param con The connection to use for the update
    * @return The number of records updated
    * @throws CpoException any errors processing the update
    */
   protected <T> long processUpdateGroup(
       List<T> beans,
       Crud crud,
       String groupName,
       Collection<CpoWhere> wheres,
       Collection<CpoOrderBy> orderBy,
       Collection<CpoNativeFunction> nativeExpressions,
       Connection con)
       throws CpoException {
     long updateCount = 0;
 
     if (beans.isEmpty()) return updateCount;
 
     if (capabilities.batchUpdatesSupported() && !Crud.UPSERT.equals(crud)) {
       updateCount =
           processBatchUpdateGroup(beans, crud, groupName, wheres, orderBy, nativeExpressions, con);
     } else {
       for (T bean : beans) {
         updateCount +=
             processUpdateGroup(bean, crud, groupName, wheres, orderBy, nativeExpressions, con);
       }
     }
 
     return updateCount;
   }
 
   private void statementClose(Statement s) {
     if (s != null) {
       try {
         s.close();
       } catch (Exception e) {
         if (logger.isTraceEnabled()) {
           logger.trace(e.getMessage());
         }
       }
     }
   }
 
   private void resultSetClose(ResultSet rs) {
     if (rs != null) {
       try {
         rs.close();
       } catch (Exception e) {
         if (logger.isTraceEnabled()) {
           logger.trace(e.getMessage());
         }
       }
     }
   }
 
   @Override
   public CpoMetaDescriptor getCpoMetaDescriptor() {
     return metaDescriptor;
   }
 
   @Override
   public List<CpoAttribute> getCpoAttributes(String expression) throws CpoException {
     List<CpoAttribute> attributes = new ArrayList<>();
 
     if (expression != null && !expression.isEmpty()) {
       Connection c = null;
       PreparedStatement ps = null;
       ResultSet rs = null;
       try {
         c = getReadConnection();
         ps = c.prepareStatement(expression);
         rs = ps.executeQuery();
         ResultSetMetaData rsmd = rs.getMetaData();
         for (int i = 1; i <= rsmd.getColumnCount(); i++) {
           JdbcCpoAttribute attribute = new JdbcCpoAttribute();
           attribute.setDataName(rsmd.getColumnLabel(i));
           attribute.setDbColumn(rsmd.getColumnName(i));
           try {
             attribute.setDbTable(rsmd.getTableName(i));
           } catch (Exception e) {
             // do nothing if this call is not supported
             logger.info("Could not get table name", e);
           }
 
           DataTypeMapEntry<?> dataTypeMapEntry =
               metaDescriptor.getDataTypeMapEntry(rsmd.getColumnType(i));
           attribute.setDataType(dataTypeMapEntry.dataTypeName());
           attribute.setDataTypeInt(dataTypeMapEntry.dataTypeInt());
           attribute.setJavaType(dataTypeMapEntry.javaClass().getName());
           attribute.setJavaName(dataTypeMapEntry.makeJavaName(rsmd.getColumnLabel(i)));
 
           attributes.add(attribute);
         }
       } catch (Throwable t) {
         logger.error(ExceptionHelper.getLocalizedMessage(t), t);
         throw new CpoException("Error Generating Attributes", t);
       } finally {
         resultSetClose(rs);
         statementClose(ps);
         // terminate the read-only transaction before pooling the connection (see existsBean)
         rollbackLocalConnection(c);
         closeLocalConnection(c);
       }
     }
     return attributes;
   }
 }