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.calcite.avatica.jdbc;
  
  import org.apache.calcite.avatica.AvaticaParameter;
  import org.apache.calcite.avatica.AvaticaPreparedStatement;
  import org.apache.calcite.avatica.AvaticaUtils;
  import org.apache.calcite.avatica.ColumnMetaData;
  import org.apache.calcite.avatica.ConnectionPropertiesImpl;
  import org.apache.calcite.avatica.Meta;
  import org.apache.calcite.avatica.MetaImpl;
  import org.apache.calcite.avatica.MissingResultsException;
  import org.apache.calcite.avatica.NoSuchConnectionException;
  import org.apache.calcite.avatica.NoSuchStatementException;
  import org.apache.calcite.avatica.QueryState;
  import org.apache.calcite.avatica.SqlType;
  import org.apache.calcite.avatica.metrics.Gauge;
  import org.apache.calcite.avatica.metrics.MetricsSystem;
  import org.apache.calcite.avatica.metrics.noop.NoopMetricsSystem;
  import org.apache.calcite.avatica.proto.Common;
  import org.apache.calcite.avatica.proto.Requests;
  import org.apache.calcite.avatica.remote.ProtobufMeta;
  import org.apache.calcite.avatica.remote.TypedValue;
  import org.apache.calcite.avatica.util.Unsafe;
  
  import com.google.common.base.Optional;
  import com.google.common.cache.Cache;
  import com.google.common.cache.CacheBuilder;
  import com.google.common.cache.RemovalListener;
  import com.google.common.cache.RemovalNotification;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.lang.reflect.InvocationTargetException;
  import java.sql.Connection;
  import java.sql.DatabaseMetaData;
  import java.sql.DriverManager;
  import java.sql.ParameterMetaData;
  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.Calendar;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Properties;
  import java.util.concurrent.ConcurrentMap;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicInteger;
  
  import static org.apache.calcite.avatica.remote.MetricsHelper.concat;
  
  /** Implementation of {@link Meta} upon an existing JDBC data source. */
  public class JdbcMeta implements ProtobufMeta {
    private static final Logger LOG = LoggerFactory.getLogger(JdbcMeta.class);
  
    private static final String CONN_CACHE_KEY_BASE = "avatica.connectioncache";
  
    private static final String STMT_CACHE_KEY_BASE = "avatica.statementcache";
  
    /** Special value for {@code Statement#getLargeMaxRows()} that means fetch
     * an unlimited number of rows in a single batch.
     *
     * <p>Any other negative value will return an unlimited number of rows but
     * will do it in the default batch size, namely 100. */
    public static final int UNLIMITED_COUNT = -2;
  
    // End of constants, start of member variables
  
    final Calendar calendar = Unsafe.localCalendar();
  
    /** Generates ids for statements. The ids are unique across all connections
     * created by this JdbcMeta. */
    private final AtomicInteger statementIdGenerator = new AtomicInteger();
  
    private final String url;
    private final Properties info;
    private final Cache<String, Connection> connectionCache;
    private final Cache<Integer, StatementInfo> statementCache;
    private final MetricsSystem metrics;
  
    /**
     * Creates a JdbcMeta.
     *
     * @param url a database url of the form
     *  <code>jdbc:<em>subprotocol</em>:<em>subname</em></code>
     */
    public JdbcMeta(String url) throws SQLException {
      this(url, new Properties());
    }
  
    /**
     * Creates a JdbcMeta.
     *
     * @param url a database url of the form
     * <code>jdbc:<em>subprotocol</em>:<em>subname</em></code>
     * @param user the database user on whose behalf the connection is being
     *   made
     * @param password the user's password
     */
    public JdbcMeta(final String url, final String user, final String password)
        throws SQLException {
      this(url, new Properties() {
        {
          put("user", user);
          put("password", password);
        }
      });
    }
  
    public JdbcMeta(String url, Properties info) throws SQLException {
      this(url, info, NoopMetricsSystem.getInstance());
    }
  
    /**
     * Creates a JdbcMeta.
     *
     * @param url a database url of the form
     * <code> jdbc:<em>subprotocol</em>:<em>subname</em></code>
     * @param info a list of arbitrary string tag/value pairs as
     * connection arguments; normally at least a "user" and
     * "password" property should be included
     */
    public JdbcMeta(String url, Properties info, MetricsSystem metrics)
        throws SQLException {
      this.url = url;
      this.info = info;
      this.metrics = Objects.requireNonNull(metrics);
  
      int concurrencyLevel = Integer.parseInt(
          info.getProperty(ConnectionCacheSettings.CONCURRENCY_LEVEL.key(),
              ConnectionCacheSettings.CONCURRENCY_LEVEL.defaultValue()));
      int initialCapacity = Integer.parseInt(
          info.getProperty(ConnectionCacheSettings.INITIAL_CAPACITY.key(),
              ConnectionCacheSettings.INITIAL_CAPACITY.defaultValue()));
      long maxCapacity = Long.parseLong(
          info.getProperty(ConnectionCacheSettings.MAX_CAPACITY.key(),
              ConnectionCacheSettings.MAX_CAPACITY.defaultValue()));
      long connectionExpiryDuration = Long.parseLong(
          info.getProperty(ConnectionCacheSettings.EXPIRY_DURATION.key(),
              ConnectionCacheSettings.EXPIRY_DURATION.defaultValue()));
      TimeUnit connectionExpiryUnit = TimeUnit.valueOf(
          info.getProperty(ConnectionCacheSettings.EXPIRY_UNIT.key(),
              ConnectionCacheSettings.EXPIRY_UNIT.defaultValue()));
      this.connectionCache = CacheBuilder.newBuilder()
          .concurrencyLevel(concurrencyLevel)
          .initialCapacity(initialCapacity)
          .maximumSize(maxCapacity)
          .expireAfterAccess(connectionExpiryDuration, connectionExpiryUnit)
          .removalListener(new ConnectionExpiryHandler())
          .build();
      LOG.debug("instantiated connection cache: {}", connectionCache.stats());
  
      concurrencyLevel = Integer.parseInt(
          info.getProperty(StatementCacheSettings.CONCURRENCY_LEVEL.key(),
              StatementCacheSettings.CONCURRENCY_LEVEL.defaultValue()));
      initialCapacity = Integer.parseInt(
          info.getProperty(StatementCacheSettings.INITIAL_CAPACITY.key(),
              StatementCacheSettings.INITIAL_CAPACITY.defaultValue()));
      maxCapacity = Long.parseLong(
          info.getProperty(StatementCacheSettings.MAX_CAPACITY.key(),
              StatementCacheSettings.MAX_CAPACITY.defaultValue()));
      connectionExpiryDuration = Long.parseLong(
          info.getProperty(StatementCacheSettings.EXPIRY_DURATION.key(),
              StatementCacheSettings.EXPIRY_DURATION.defaultValue()));
      connectionExpiryUnit = TimeUnit.valueOf(
          info.getProperty(StatementCacheSettings.EXPIRY_UNIT.key(),
              StatementCacheSettings.EXPIRY_UNIT.defaultValue()));
      this.statementCache = CacheBuilder.newBuilder()
          .concurrencyLevel(concurrencyLevel)
          .initialCapacity(initialCapacity)
          .maximumSize(maxCapacity)
          .expireAfterAccess(connectionExpiryDuration, connectionExpiryUnit)
          .removalListener(new StatementExpiryHandler())
          .build();
  
      LOG.debug("instantiated statement cache: {}", statementCache.stats());
  
      // Register some metrics
      this.metrics.register(concat(JdbcMeta.class, "ConnectionCacheSize"), new Gauge<Long>() {
        @Override public Long getValue() {
          return connectionCache.size();
        }
      });
  
      this.metrics.register(concat(JdbcMeta.class, "StatementCacheSize"), new Gauge<Long>() {
        @Override public Long getValue() {
          return statementCache.size();
        }
      });
    }
  
    // For testing purposes
    protected AtomicInteger getStatementIdGenerator() {
      return statementIdGenerator;
    }
  
    // For testing purposes
    protected Cache<String, Connection> getConnectionCache() {
      return connectionCache;
    }
  
    // For testing purposes
    protected Cache<Integer, StatementInfo> getStatementCache() {
      return statementCache;
    }
  
    /**
     * Converts from JDBC metadata to Avatica columns.
     */
    protected static List<ColumnMetaData>
        columns(ResultSetMetaData metaData) throws SQLException {
      if (metaData == null) {
        return Collections.emptyList();
      }
      final List<ColumnMetaData> columns = new ArrayList<>();
      for (int i = 1; i <= metaData.getColumnCount(); i++) {
        final SqlType sqlType = SqlType.valueOf(metaData.getColumnType(i));
        final ColumnMetaData.Rep rep = ColumnMetaData.Rep.of(sqlType.internal);
        final ColumnMetaData.AvaticaType t;
        if (sqlType == SqlType.ARRAY || sqlType == SqlType.STRUCT || sqlType == SqlType.MULTISET) {
          ColumnMetaData.AvaticaType arrayValueType = ColumnMetaData.scalar(Types.JAVA_OBJECT,
              metaData.getColumnTypeName(i), ColumnMetaData.Rep.OBJECT);
          t = ColumnMetaData.array(arrayValueType, metaData.getColumnTypeName(i), rep);
        } else {
          t = ColumnMetaData.scalar(metaData.getColumnType(i), metaData.getColumnTypeName(i), rep);
        }
        ColumnMetaData md =
            new ColumnMetaData(i - 1, metaData.isAutoIncrement(i),
                metaData.isCaseSensitive(i), metaData.isSearchable(i),
                metaData.isCurrency(i), metaData.isNullable(i),
                metaData.isSigned(i), metaData.getColumnDisplaySize(i),
                metaData.getColumnLabel(i), metaData.getColumnName(i),
                metaData.getSchemaName(i), metaData.getPrecision(i),
                metaData.getScale(i), metaData.getTableName(i),
                metaData.getCatalogName(i), t, metaData.isReadOnly(i),
                metaData.isWritable(i), metaData.isDefinitelyWritable(i),
                metaData.getColumnClassName(i));
        columns.add(md);
      }
      return columns;
    }
  
    /**
     * Converts from JDBC metadata to Avatica parameters
     */
    protected static List<AvaticaParameter> parameters(ParameterMetaData metaData)
        throws SQLException {
      if (metaData == null) {
        return Collections.emptyList();
      }
      final List<AvaticaParameter> params = new ArrayList<>();
      for (int i = 1; i <= metaData.getParameterCount(); i++) {
        params.add(
            new AvaticaParameter(metaData.isSigned(i), metaData.getPrecision(i),
                metaData.getScale(i), metaData.getParameterType(i),
                metaData.getParameterTypeName(i),
                metaData.getParameterClassName(i), "?" + i));
      }
      return params;
    }
  
    protected static Signature signature(ResultSetMetaData metaData,
        ParameterMetaData parameterMetaData, String sql,
        Meta.StatementType statementType) throws  SQLException {
      final CursorFactory cf = CursorFactory.LIST;  // because JdbcResultSet#frame
      return new Signature(columns(metaData), sql, parameters(parameterMetaData),
          null, cf, statementType);
    }
  
    protected static Signature signature(ResultSetMetaData metaData)
        throws SQLException {
      return signature(metaData, null, null, null);
    }
  
    public Map<DatabaseProperty, Object> getDatabaseProperties(ConnectionHandle ch) {
      try {
        final Map<DatabaseProperty, Object> map = new HashMap<>();
        final Connection conn = getConnection(ch.id);
        final DatabaseMetaData metaData = conn.getMetaData();
        for (DatabaseProperty p : DatabaseProperty.values()) {
          addProperty(map, metaData, p);
        }
        return map;
      } catch (SQLException e) {
        throw new RuntimeException(e);
      }
    }
  
    private static Object addProperty(Map<DatabaseProperty, Object> map,
        DatabaseMetaData metaData, DatabaseProperty p) throws SQLException {
      Object propertyValue;
      if (p.isJdbc) {
        try {
          propertyValue = p.method.invoke(metaData);
        } catch (IllegalAccessException | InvocationTargetException e) {
          throw new RuntimeException(e);
        }
      } else {
        propertyValue = p.defaultValue;
      }
  
      return map.put(p, propertyValue);
    }
  
    public MetaResultSet getTables(ConnectionHandle ch, String catalog, Pat schemaPattern,
        Pat tableNamePattern, List<String> typeList) {
      try {
        final ResultSet rs =
            getConnection(ch.id).getMetaData().getTables(catalog, schemaPattern.s,
                tableNamePattern.s, toArray(typeList));
        int stmtId = registerMetaStatement(rs);
        return JdbcResultSet.create(ch.id, stmtId, rs);
      } catch (SQLException e) {
        throw new RuntimeException(e);
      }
    }
  
    /**
      * Registers a StatementInfo for the given ResultSet, returning the id under
      * which it is registered. This should be used for metadata ResultSets, which
      * have an implicit statement created.
      */
    private int registerMetaStatement(ResultSet rs) throws SQLException {
      final int id = statementIdGenerator.getAndIncrement();
      StatementInfo statementInfo = new StatementInfo(rs.getStatement());
      statementInfo.setResultSet(rs);
      statementCache.put(id, statementInfo);
      return id;
    }
  
    public MetaResultSet getColumns(ConnectionHandle ch, String catalog, Pat schemaPattern,
        Pat tableNamePattern, Pat columnNamePattern) {
      try {
        final ResultSet rs =
            getConnection(ch.id).getMetaData().getColumns(catalog, schemaPattern.s,
                tableNamePattern.s, columnNamePattern.s);
        int stmtId = registerMetaStatement(rs);
        return JdbcResultSet.create(ch.id, stmtId, rs);
      } catch (SQLException e) {
        throw new RuntimeException(e);
      }
    }
  
    public MetaResultSet getSchemas(ConnectionHandle ch, String catalog, Pat schemaPattern) {
      try {
        final ResultSet rs =
            getConnection(ch.id).getMetaData().getSchemas(catalog, schemaPattern.s);
        int stmtId = registerMetaStatement(rs);
        return JdbcResultSet.create(ch.id, stmtId, rs);
      } catch (SQLException e) {
        throw new RuntimeException(e);
      }
    }
  
    public MetaResultSet getCatalogs(ConnectionHandle ch) {
      try {
        final ResultSet rs = getConnection(ch.id).getMetaData().getCatalogs();
        int stmtId = registerMetaStatement(rs);
        return JdbcResultSet.create(ch.id, stmtId, rs);
      } catch (SQLException e) {
        throw new RuntimeException(e);
      }
    }
  
    public MetaResultSet getTableTypes(ConnectionHandle ch) {
      try {
        final ResultSet rs = getConnection(ch.id).getMetaData().getTableTypes();
        int stmtId = registerMetaStatement(rs);
        return JdbcResultSet.create(ch.id, stmtId, rs);
      } catch (SQLException e) {
        throw new RuntimeException(e);
      }
    }
  
    public MetaResultSet getProcedures(ConnectionHandle ch, String catalog, Pat schemaPattern,
        Pat procedureNamePattern) {
      try {
        final ResultSet rs =
            getConnection(ch.id).getMetaData().getProcedures(catalog, schemaPattern.s,
                procedureNamePattern.s);
        int stmtId = registerMetaStatement(rs);
        return JdbcResultSet.create(ch.id, stmtId, rs);
      } catch (SQLException e) {
        throw new RuntimeException(e);
      }
    }
  
    public MetaResultSet getProcedureColumns(ConnectionHandle ch, String catalog, Pat schemaPattern,
        Pat procedureNamePattern, Pat columnNamePattern) {
      try {
        final ResultSet rs =
            getConnection(ch.id).getMetaData().getProcedureColumns(catalog,
                schemaPattern.s, procedureNamePattern.s, columnNamePattern.s);
        int stmtId = registerMetaStatement(rs);
        return JdbcResultSet.create(ch.id, stmtId, rs);
      } catch (SQLException e) {
        throw new RuntimeException(e);
      }
    }
  
    public MetaResultSet getColumnPrivileges(ConnectionHandle ch, String catalog, String schema,
        String table, Pat columnNamePattern) {
      try {
        final ResultSet rs =
            getConnection(ch.id).getMetaData().getColumnPrivileges(catalog, schema,
                table, columnNamePattern.s);
        int stmtId = registerMetaStatement(rs);
        return JdbcResultSet.create(ch.id, stmtId, rs);
      } catch (SQLException e) {
        throw new RuntimeException(e);
      }
    }
  
    public MetaResultSet getTablePrivileges(ConnectionHandle ch, String catalog, Pat schemaPattern,
        Pat tableNamePattern) {
      try {
        final ResultSet rs =
            getConnection(ch.id).getMetaData().getTablePrivileges(catalog,
                schemaPattern.s, tableNamePattern.s);
        int stmtId = registerMetaStatement(rs);
        return JdbcResultSet.create(ch.id, stmtId, rs);
      } catch (SQLException e) {
        throw new RuntimeException(e);
      }
    }
  
    public MetaResultSet getBestRowIdentifier(ConnectionHandle ch, String catalog, String schema,
        String table, int scope, boolean nullable) {
      LOG.trace("getBestRowIdentifier catalog:{} schema:{} table:{} scope:{} nullable:{}", catalog,
          schema, table, scope, nullable);
      try {
        final ResultSet rs =
            getConnection(ch.id).getMetaData().getBestRowIdentifier(catalog, schema,
                table, scope, nullable);
        int stmtId = registerMetaStatement(rs);
        return JdbcResultSet.create(ch.id, stmtId, rs);
      } catch (SQLException e) {
        throw new RuntimeException(e);
      }
    }
  
    public MetaResultSet getVersionColumns(ConnectionHandle ch, String catalog, String schema,
        String table) {
      LOG.trace("getVersionColumns catalog:{} schema:{} table:{}", catalog, schema, table);
      try {
        final ResultSet rs =
            getConnection(ch.id).getMetaData().getVersionColumns(catalog, schema, table);
        int stmtId = registerMetaStatement(rs);
        return JdbcResultSet.create(ch.id, stmtId, rs);
      } catch (SQLException e) {
        throw new RuntimeException(e);
      }
    }
  
    public MetaResultSet getPrimaryKeys(ConnectionHandle ch, String catalog, String schema,
        String table) {
      LOG.trace("getPrimaryKeys catalog:{} schema:{} table:{}", catalog, schema, table);
      try {
        final ResultSet rs =
            getConnection(ch.id).getMetaData().getPrimaryKeys(catalog, schema, table);
        int stmtId = registerMetaStatement(rs);
        return JdbcResultSet.create(ch.id, stmtId, rs);
      } catch (SQLException e) {
        throw new RuntimeException(e);
      }
    }
  
    public MetaResultSet getImportedKeys(ConnectionHandle ch, String catalog, String schema,
        String table) {
      return null;
    }
  
    public MetaResultSet getExportedKeys(ConnectionHandle ch, String catalog, String schema,
        String table) {
      return null;
    }
  
    public MetaResultSet getCrossReference(ConnectionHandle ch, String parentCatalog,
        String parentSchema, String parentTable, String foreignCatalog,
        String foreignSchema, String foreignTable) {
      return null;
    }
  
    public MetaResultSet getTypeInfo(ConnectionHandle ch) {
      try {
        final ResultSet rs = getConnection(ch.id).getMetaData().getTypeInfo();
        int stmtId = registerMetaStatement(rs);
        return JdbcResultSet.create(ch.id, stmtId, rs);
      } catch (SQLException e) {
        throw new RuntimeException(e);
      }
    }
  
    public MetaResultSet getIndexInfo(ConnectionHandle ch, String catalog, String schema,
        String table, boolean unique, boolean approximate) {
      return null;
    }
  
    public MetaResultSet getUDTs(ConnectionHandle ch, String catalog, Pat schemaPattern,
        Pat typeNamePattern, int[] types) {
      return null;
    }
  
    public MetaResultSet getSuperTypes(ConnectionHandle ch, String catalog, Pat schemaPattern,
        Pat typeNamePattern) {
      return null;
    }
  
    public MetaResultSet getSuperTables(ConnectionHandle ch, String catalog, Pat schemaPattern,
        Pat tableNamePattern) {
      return null;
    }
  
    public MetaResultSet getAttributes(ConnectionHandle ch, String catalog, Pat schemaPattern,
        Pat typeNamePattern, Pat attributeNamePattern) {
      return null;
    }
  
    public MetaResultSet getClientInfoProperties(ConnectionHandle ch) {
      return null;
    }
  
    public MetaResultSet getFunctions(ConnectionHandle ch, String catalog, Pat schemaPattern,
        Pat functionNamePattern) {
      return null;
    }
  
    public MetaResultSet getFunctionColumns(ConnectionHandle ch, String catalog, Pat schemaPattern,
        Pat functionNamePattern, Pat columnNamePattern) {
      return null;
    }
  
    public MetaResultSet getPseudoColumns(ConnectionHandle ch, String catalog, Pat schemaPattern,
        Pat tableNamePattern, Pat columnNamePattern) {
      return null;
    }
  
    public Iterable<Object> createIterable(StatementHandle handle, QueryState state,
        Signature signature, List<TypedValue> parameterValues, Frame firstFrame) {
      return null;
    }
  
    protected Connection getConnection(String id) throws SQLException {
      if (id == null) {
        throw new NullPointerException("Connection id is null.");
      }
      Connection conn = connectionCache.getIfPresent(id);
      if (conn == null) {
        throw new NoSuchConnectionException("Connection not found: invalid id, closed, or expired: "
            + id);
      }
      return conn;
    }
  
    public StatementHandle createStatement(ConnectionHandle ch) {
      try {
        final Connection conn = getConnection(ch.id);
        final Statement statement = conn.createStatement();
        final int id = statementIdGenerator.getAndIncrement();
        statementCache.put(id, new StatementInfo(statement));
        StatementHandle h = new StatementHandle(ch.id, id, null);
        LOG.trace("created statement {}", h);
        return h;
      } catch (SQLException e) {
        throw propagate(e);
      }
    }
  
    @Override public void closeStatement(StatementHandle h) {
      StatementInfo info = statementCache.getIfPresent(h.id);
      if (info == null || info.statement == null) {
        LOG.debug("client requested close unknown statement {}", h);
        return;
      }
      LOG.trace("closing statement {}", h);
      try {
        ResultSet results = info.getResultSet();
        if (info.isResultSetInitialized() && null != results) {
          results.close();
        }
        info.statement.close();
      } catch (SQLException e) {
        throw propagate(e);
      } finally {
        statementCache.invalidate(h.id);
      }
    }
  
    @Override public void openConnection(ConnectionHandle ch,
        Map<String, String> info) {
      Properties fullInfo = new Properties();
      fullInfo.putAll(this.info);
      if (info != null) {
        fullInfo.putAll(info);
      }
  
      final ConcurrentMap<String, Connection> cacheAsMap = connectionCache.asMap();
      if (cacheAsMap.containsKey(ch.id)) {
        throw new RuntimeException("Connection already exists: " + ch.id);
      }
      // Avoid global synchronization of connection opening
      try {
        Connection conn = createConnection(url, fullInfo);
        Connection loadedConn = cacheAsMap.putIfAbsent(ch.id, conn);
        // Race condition: someone beat us to storing the connection in the cache.
        if (loadedConn != null) {
          conn.close();
          throw new RuntimeException("Connection already exists: " + ch.id);
        }
      } catch (SQLException e) {
        throw new RuntimeException(e);
      }
    }
  
    // Visible for testing
    protected Connection createConnection(String url, Properties info) throws SQLException {
      // Allows simpler testing of openConnection
      return DriverManager.getConnection(url, info);
    }
  
    @Override public void closeConnection(ConnectionHandle ch) {
      Connection conn = connectionCache.getIfPresent(ch.id);
      if (conn == null) {
        LOG.debug("client requested close unknown connection {}", ch);
        return;
      }
      LOG.trace("closing connection {}", ch);
      try {
        conn.close();
      } catch (SQLException e) {
        throw propagate(e);
      } finally {
        connectionCache.invalidate(ch.id);
      }
    }
  
    protected void apply(Connection conn, ConnectionProperties connProps)
        throws SQLException {
      if (connProps.isAutoCommit() != null) {
        conn.setAutoCommit(connProps.isAutoCommit());
      }
      if (connProps.isReadOnly() != null) {
        conn.setReadOnly(connProps.isReadOnly());
      }
      if (connProps.getTransactionIsolation() != null) {
        conn.setTransactionIsolation(connProps.getTransactionIsolation());
      }
      if (connProps.getCatalog() != null) {
        conn.setCatalog(connProps.getCatalog());
      }
      if (connProps.getSchema() != null) {
        conn.setSchema(connProps.getSchema());
      }
    }
  
    @Override public ConnectionProperties connectionSync(ConnectionHandle ch,
        ConnectionProperties connProps) {
      LOG.trace("syncing properties for connection {}", ch);
      try {
        Connection conn = getConnection(ch.id);
        ConnectionPropertiesImpl props = new ConnectionPropertiesImpl(conn).merge(connProps);
        if (props.isDirty()) {
          apply(conn, props);
          props.setDirty(false);
        }
        return props;
      } catch (SQLException e) {
        throw propagate(e);
      }
    }
  
    RuntimeException propagate(Throwable e) {
      if (e instanceof RuntimeException) {
        throw (RuntimeException) e;
      } else if (e instanceof Error) {
        throw (Error) e;
      } else {
        throw new RuntimeException(e);
      }
    }
  
    public StatementHandle prepare(ConnectionHandle ch, String sql,
        long maxRowCount) {
      try {
        final Connection conn = getConnection(ch.id);
        final PreparedStatement statement = conn.prepareStatement(sql);
        final int id = getStatementIdGenerator().getAndIncrement();
        Meta.StatementType statementType = null;
        if (statement.isWrapperFor(AvaticaPreparedStatement.class)) {
          final AvaticaPreparedStatement avaticaPreparedStatement;
          avaticaPreparedStatement =
              statement.unwrap(AvaticaPreparedStatement.class);
          statementType = avaticaPreparedStatement.getStatementType();
        }
        // Set the maximum number of rows
        setMaxRows(statement, maxRowCount);
        getStatementCache().put(id, new StatementInfo(statement));
        StatementHandle h = new StatementHandle(ch.id, id,
            signature(statement.getMetaData(), statement.getParameterMetaData(),
                sql, statementType));
        LOG.trace("prepared statement {}", h);
        return h;
      } catch (SQLException e) {
        throw propagate(e);
      }
    }
  
    @SuppressWarnings("deprecation")
    public ExecuteResult prepareAndExecute(StatementHandle h, String sql,
        long maxRowCount, PrepareCallback callback) throws NoSuchStatementException {
      return prepareAndExecute(h, sql, maxRowCount, AvaticaUtils.toSaturatedInt(maxRowCount),
          callback);
    }
  
    public ExecuteResult prepareAndExecute(StatementHandle h, String sql, long maxRowCount,
        int maxRowsInFirstFrame, PrepareCallback callback) throws NoSuchStatementException {
      try {
        final StatementInfo info = getStatementCache().getIfPresent(h.id);
        if (info == null) {
          throw new NoSuchStatementException(h);
        }
        final Statement statement = info.statement;
        // Make sure that we limit the number of rows for the query
        setMaxRows(statement, maxRowCount);
        boolean ret = statement.execute(sql);
        info.setResultSet(statement.getResultSet());
        // Either execute(sql) returned true or the resultSet was null
        assert ret || null == info.getResultSet();
        final List<MetaResultSet> resultSets = new ArrayList<>();
        if (null == info.getResultSet()) {
          // Create a special result set that just carries update count
          resultSets.add(
              JdbcResultSet.count(h.connectionId, h.id,
                  AvaticaUtils.getLargeUpdateCount(statement)));
        } else {
          resultSets.add(
              JdbcResultSet.create(h.connectionId, h.id, info.getResultSet(), maxRowsInFirstFrame));
        }
        LOG.trace("prepAndExec statement {}", h);
        // TODO: review client to ensure statementId is updated when appropriate
        return new ExecuteResult(resultSets);
      } catch (SQLException e) {
        throw propagate(e);
      }
    }
  
    /**
     * Sets the provided maximum number of rows on the given statement.
     *
     * @param statement The JDBC Statement to operate on
     * @param maxRowCount The maximum number of rows which should be returned for the query
     */
    void setMaxRows(Statement statement, long maxRowCount) throws SQLException {
      // Special handling of maxRowCount as JDBC 0 is unlimited, our meta 0 row
      if (maxRowCount > 0) {
        AvaticaUtils.setLargeMaxRows(statement, maxRowCount);
      } else if (maxRowCount < 0) {
        statement.setMaxRows(0);
      }
    }
  
    public boolean syncResults(StatementHandle sh, QueryState state, long offset)
        throws NoSuchStatementException {
      try {
        final Connection conn = getConnection(sh.connectionId);
        final StatementInfo info = statementCache.getIfPresent(sh.id);
        if (null == info) {
          throw new NoSuchStatementException(sh);
        }
        final Statement statement = info.statement;
        // Let the state recreate the necessary ResultSet on the Statement
        info.setResultSet(state.invoke(conn, statement));
  
        if (null != info.getResultSet()) {
          // If it is non-null, try to advance to the requested offset.
          return info.advanceResultSetToOffset(info.getResultSet(), offset);
        }
  
        // No results, nothing to do. Client can move on.
        return false;
      } catch (SQLException e) {
        throw propagate(e);
      }
    }
  
    public Frame fetch(StatementHandle h, long offset, int fetchMaxRowCount) throws
        NoSuchStatementException, MissingResultsException {
      LOG.trace("fetching {} offset:{} fetchMaxRowCount:{}", h, offset, fetchMaxRowCount);
      try {
        final StatementInfo statementInfo = statementCache.getIfPresent(h.id);
        if (null == statementInfo) {
          // Statement might have expired, or never existed on this server.
          throw new NoSuchStatementException(h);
        }
  
        if (!statementInfo.isResultSetInitialized()) {
          // The Statement exists, but the results are missing. Need to call syncResults(...)
          throw new MissingResultsException(h);
        }
        if (statementInfo.getResultSet() == null) {
          return Frame.EMPTY;
        } else {
          return JdbcResultSet.frame(statementInfo, statementInfo.getResultSet(), offset,
              fetchMaxRowCount, calendar, Optional.<Meta.Signature>absent());
        }
      } catch (SQLException e) {
        throw propagate(e);
      }
    }
  
    private static String[] toArray(List<String> typeList) {
      if (typeList == null) {
        return null;
      }
      return typeList.toArray(new String[typeList.size()]);
    }
  
    @SuppressWarnings("deprecation")
    @Override public ExecuteResult execute(StatementHandle h, List<TypedValue> parameterValues,
        long maxRowCount) throws NoSuchStatementException {
      return execute(h, parameterValues, AvaticaUtils.toSaturatedInt(maxRowCount));
    }
  
    @Override public ExecuteResult execute(StatementHandle h,
        List<TypedValue> parameterValues, int maxRowsInFirstFrame) throws NoSuchStatementException {
      try {
        if (parameterValues != null && MetaImpl.checkParameterValueHasNull(parameterValues)) {
          throw new SQLException("exception while executing query: unbound parameter");
        }
  
        final StatementInfo statementInfo = getStatementCache().getIfPresent(h.id);
        if (null == statementInfo) {
          throw new NoSuchStatementException(h);
        }
        final List<MetaResultSet> resultSets;
        final PreparedStatement preparedStatement =
            (PreparedStatement) statementInfo.statement;
  
        if (parameterValues != null) {
          for (int i = 0; i < parameterValues.size(); i++) {
            TypedValue o = parameterValues.get(i);
            preparedStatement.setObject(i + 1, o.toJdbc(calendar));
          }
        }
  
        if (preparedStatement.execute()) {
          final Signature signature2;
          if (preparedStatement.isWrapperFor(AvaticaPreparedStatement.class)) {
            signature2 = h.signature;
          } else {
            h.signature = signature(preparedStatement.getMetaData(),
                preparedStatement.getParameterMetaData(), h.signature.sql,
                Meta.StatementType.SELECT);
            signature2 = h.signature;
          }
  
          // Make sure we set this for subsequent fetch()'s to find the result set.
          statementInfo.setResultSet(preparedStatement.getResultSet());
  
          if (statementInfo.getResultSet() == null) {
            resultSets = Collections.<MetaResultSet>singletonList(
                JdbcResultSet.empty(h.connectionId, h.id, signature2));
          } else {
            resultSets = Collections.<MetaResultSet>singletonList(
                JdbcResultSet.create(h.connectionId, h.id, statementInfo.getResultSet(),
                    maxRowsInFirstFrame, signature2));
          }
        } else {
          resultSets = Collections.<MetaResultSet>singletonList(
              JdbcResultSet.count(h.connectionId, h.id, preparedStatement.getUpdateCount()));
        }
  
        return new ExecuteResult(resultSets);
      } catch (SQLException e) {
        throw propagate(e);
      }
    }
  
    @Override public void commit(ConnectionHandle ch) {
      try {
        final Connection conn = getConnection(ch.id);
        conn.commit();
      } catch (SQLException e) {
        throw propagate(e);
      }
    }
  
    @Override public void rollback(ConnectionHandle ch) {
      try {
        final Connection conn = getConnection(ch.id);
        conn.rollback();
      } catch (SQLException e) {
        throw propagate(e);
      }
    }
  
    @Override public ExecuteBatchResult prepareAndExecuteBatch(StatementHandle h,
        List<String> sqlCommands) throws NoSuchStatementException {
      try {
        // Get the statement
        final StatementInfo info = statementCache.getIfPresent(h.id);
        if (info == null) {
          throw new NoSuchStatementException(h);
        }
  
        // addBatch() for each sql command
        final Statement stmt = info.statement;
        for (String sqlCommand : sqlCommands) {
          stmt.addBatch(sqlCommand);
        }
  
        // Execute the batch and return the results
        return new ExecuteBatchResult(AvaticaUtils.executeLargeBatch(stmt));
      } catch (SQLException e) {
        throw propagate(e);
      }
    }
  
    @Override public ExecuteBatchResult executeBatch(StatementHandle h,
        List<List<TypedValue>> updateBatches) throws NoSuchStatementException {
      try {
        final StatementInfo info = statementCache.getIfPresent(h.id);
        if (null == info) {
          throw new NoSuchStatementException(h);
        }
  
        final PreparedStatement preparedStmt = (PreparedStatement) info.statement;
        int rowUpdate = 1;
        for (List<TypedValue> batch : updateBatches) {
          int i = 1;
          for (TypedValue value : batch) {
            // Set the TypedValue in the PreparedStatement
            try {
              preparedStmt.setObject(i, value.toJdbc(calendar));
              i++;
            } catch (SQLException e) {
              throw new RuntimeException("Failed to set value on row #" + rowUpdate
                  + " and column #" + i, e);
            }
            // Track the update number for better error messages
            rowUpdate++;
          }
          preparedStmt.addBatch();
        }
        return new ExecuteBatchResult(AvaticaUtils.executeLargeBatch(preparedStmt));
      } catch (SQLException e) {
        throw propagate(e);
      }
    }
  
    @Override public ExecuteBatchResult executeBatchProtobuf(StatementHandle h,
        List<Requests.UpdateBatch> updateBatches) throws NoSuchStatementException {
      try {
        final StatementInfo info = statementCache.getIfPresent(h.id);
        if (null == info) {
          throw new NoSuchStatementException(h);
        }
  
        final PreparedStatement preparedStmt = (PreparedStatement) info.statement;
        for (Requests.UpdateBatch update : updateBatches) {
          int i = 1;
          for (Common.TypedValue value : update.getParameterValuesList()) {
            // Use the value and then increment
            preparedStmt.setObject(i++, TypedValue.protoToJdbc(value, calendar));
          }
          preparedStmt.addBatch();
        }
        return new ExecuteBatchResult(AvaticaUtils.executeLargeBatch(preparedStmt));
      } catch (SQLException e) {
        throw propagate(e);
      }
    }
  
    /** Configurable statement cache settings. */
    public enum StatementCacheSettings {
      /** JDBC connection property for setting connection cache concurrency level. */
      CONCURRENCY_LEVEL(STMT_CACHE_KEY_BASE + ".concurrency", "100"),
  
      /** JDBC connection property for setting connection cache initial capacity. */
      INITIAL_CAPACITY(STMT_CACHE_KEY_BASE + ".initialcapacity", "1000"),
  
      /** JDBC connection property for setting connection cache maximum capacity. */
      MAX_CAPACITY(STMT_CACHE_KEY_BASE + ".maxcapacity", "10000"),
  
      /** JDBC connection property for setting connection cache expiration duration.
       *
       * <p>Used in conjunction with {@link #EXPIRY_UNIT}.</p>
       */
      EXPIRY_DURATION(STMT_CACHE_KEY_BASE + ".expiryduration", "5"),
  
      /** JDBC connection property for setting connection cache expiration unit.
       *
       * <p>Used in conjunction with {@link #EXPIRY_DURATION}.</p>
       */
      EXPIRY_UNIT(STMT_CACHE_KEY_BASE + ".expiryunit", TimeUnit.MINUTES.name());
  
      private final String key;
      private final String defaultValue;
  
      StatementCacheSettings(String key, String defaultValue) {
        this.key = key;
        this.defaultValue = defaultValue;
      }
  
      /** The configuration key for specifying this setting. */
      public String key() {
        return key;
      }
  
      /** The default value for this setting. */
      public String defaultValue() {
        return defaultValue;
      }
    }
  
    /** Configurable connection cache settings. */
    public enum ConnectionCacheSettings {
      /** JDBC connection property for setting connection cache concurrency level. */
      CONCURRENCY_LEVEL(CONN_CACHE_KEY_BASE + ".concurrency", "10"),
  
      /** JDBC connection property for setting connection cache initial capacity. */
      INITIAL_CAPACITY(CONN_CACHE_KEY_BASE + ".initialcapacity", "100"),
  
      /** JDBC connection property for setting connection cache maximum capacity. */
      MAX_CAPACITY(CONN_CACHE_KEY_BASE + ".maxcapacity", "1000"),
  
      /** JDBC connection property for setting connection cache expiration duration. */
      EXPIRY_DURATION(CONN_CACHE_KEY_BASE + ".expiryduration", "10"),
  
      /** JDBC connection property for setting connection cache expiration unit. */
      EXPIRY_UNIT(CONN_CACHE_KEY_BASE + ".expiryunit", TimeUnit.MINUTES.name());
  
      private final String key;
      private final String defaultValue;
  
      ConnectionCacheSettings(String key, String defaultValue) {
        this.key = key;
        this.defaultValue = defaultValue;
      }
  
      /** The configuration key for specifying this setting. */
      public String key() {
        return key;
      }
  
      /** The default value for this setting. */
      public String defaultValue() {
        return defaultValue;
      }
    }
  
    /** Callback for {@link #connectionCache} member expiration. */
    private class ConnectionExpiryHandler
        implements RemovalListener<String, Connection> {
  
      public void onRemoval(RemovalNotification<String, Connection> notification) {
        String connectionId = notification.getKey();
        Connection doomed = notification.getValue();
        LOG.debug("Expiring connection {} because {}", connectionId, notification.getCause());
        try {
          if (doomed != null) {
            doomed.close();
          }
        } catch (Throwable t) {
          LOG.info("Exception thrown while expiring connection {}", connectionId, t);
        }
      }
    }
  
    /** Callback for {@link #statementCache} member expiration. */
    private class StatementExpiryHandler
        implements RemovalListener<Integer, StatementInfo> {
      public void onRemoval(RemovalNotification<Integer, StatementInfo> notification) {
        Integer stmtId = notification.getKey();
        StatementInfo doomed = notification.getValue();
        if (doomed == null) {
          // log/throw?
          return;
        }
        LOG.debug("Expiring statement {} because {}", stmtId, notification.getCause());
        try {
          if (doomed.getResultSet() != null) {
            doomed.getResultSet().close();
          }
          if (doomed.statement != null) {
            doomed.statement.close();
          }
        } catch (Throwable t) {
          LOG.info("Exception thrown while expiring statement {}", stmtId, t);
        }
      }
    }
  }
  
  // End JdbcMeta.java
  