@Override public StatementHandle prepare(ConnectionHandle ch, String sql,
    long maxRowCount) {
  final StatementHandle h = createStatement(ch);
  final CalciteConnectionImpl calciteConnection = getConnection();

  final CalciteServerStatement statement;
  try {
    statement = calciteConnection.server.getStatement(h);
  } catch (NoSuchStatementException e) {
    // Not possible. We just created a statement.
    throw new AssertionError("missing statement", e);
  }
  final Context context = statement.createPrepareContext();
  final CalcitePrepare.Query<Object> query = toQuery(context, sql);
  h.signature = calciteConnection.parseQuery(query, context, maxRowCount);
  statement.setSignature(h.signature);
  return h;
}
 

@Override public Frame fetch(StatementHandle h, long offset,
    int fetchMaxRowCount) throws NoSuchStatementException {
  final CalciteConnectionImpl calciteConnection = getConnection();
  CalciteServerStatement stmt = calciteConnection.server.getStatement(h);
  final Signature signature = stmt.getSignature();
  final Iterator<Object> iterator;
  if (stmt.getResultSet() == null) {
    final Iterable<Object> iterable =
        _createIterable(h, signature, null, null);
    iterator = iterable.iterator();
    stmt.setResultSet(iterator);
  } else {
    iterator = stmt.getResultSet();
  }
  final List rows =
      MetaImpl.collect(signature.cursorFactory,
          LimitIterator.of(iterator, fetchMaxRowCount),
          new ArrayList<List<Object>>());
  boolean done = fetchMaxRowCount == 0 || rows.size() < fetchMaxRowCount;
  @SuppressWarnings("unchecked") List<Object> rows1 = (List<Object>) rows;
  return new Meta.Frame(offset, done, rows1);
}
 

@Override public ExecuteResult execute(StatementHandle h,
    List<TypedValue> parameterValues, int maxRowsInFirstFrame)
    throws NoSuchStatementException {
  final CalciteConnectionImpl calciteConnection = getConnection();
  CalciteServerStatement stmt = calciteConnection.server.getStatement(h);
  final Signature signature = stmt.getSignature();

  MetaResultSet metaResultSet;
  if (signature.statementType.canUpdate()) {
    final Iterable<Object> iterable =
        _createIterable(h, signature, parameterValues, null);
    final Iterator<Object> iterator = iterable.iterator();
    stmt.setResultSet(iterator);
    metaResultSet = MetaResultSet.count(h.connectionId, h.id,
        ((Number) iterator.next()).intValue());
  } else {
    // Don't populate the first frame.
    // It's not worth saving a round-trip, since we're local.
    final Meta.Frame frame =
        new Meta.Frame(0, false, Collections.emptyList());
    metaResultSet =
        MetaResultSet.create(h.connectionId, h.id, false, signature, frame);
  }

  return new ExecuteResult(ImmutableList.of(metaResultSet));
}
 

@Override public boolean syncResults(final StatementHandle h, final QueryState state,
    final long offset) throws NoSuchStatementException {
  try {
    return connection.invokeWithRetries(
        new CallableWithoutException<Boolean>() {
          public Boolean call() {
            final Service.SyncResultsResponse response =
                service.apply(
                    new Service.SyncResultsRequest(h.connectionId, h.id, state, offset));
            if (response.missingStatement) {
              throw new RuntimeException(new NoSuchStatementException(h));
            }
            return response.moreResults;
          }
        });
  } catch (RuntimeException e) {
    Throwable cause = e.getCause();
    if (cause instanceof NoSuchStatementException) {
      throw (NoSuchStatementException) cause;
    }
    throw e;
  }
}
 

@Override public ExecuteResult execute(StatementHandle h,
    List<TypedValue> parameterValues, int maxRowsInFirstFrame)
    throws NoSuchStatementException {
  final CalciteConnectionImpl calciteConnection = getConnection();
  CalciteServerStatement stmt = calciteConnection.server.getStatement(h);
  final Signature signature = stmt.getSignature();

  MetaResultSet metaResultSet;
  if (signature.statementType.canUpdate()) {
    final Iterable<Object> iterable =
        _createIterable(h, signature, parameterValues, null);
    final Iterator<Object> iterator = iterable.iterator();
    stmt.setResultSet(iterator);
    metaResultSet = MetaResultSet.count(h.connectionId, h.id,
        ((Number) iterator.next()).intValue());
  } else {
    // Don't populate the first frame.
    // It's not worth saving a round-trip, since we're local.
    final Meta.Frame frame =
        new Meta.Frame(0, false, Collections.emptyList());
    metaResultSet =
        MetaResultSet.create(h.connectionId, h.id, false, signature, frame);
  }

  return new ExecuteResult(ImmutableList.of(metaResultSet));
}
 

@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 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);
    }
}
 

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);
  }
}
 

@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 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);
  }
}
 

public ExecuteResponse apply(ExecuteRequest request) {
  try (final Context ignore = executeTimer.start()) {
    try {
      final Meta.ExecuteResult executeResult = meta.execute(request.statementHandle,
          request.parameterValues, AvaticaUtils.toSaturatedInt(request.maxRowCount));

      final List<ResultSetResponse> results = new ArrayList<>(executeResult.resultSets.size());
      for (Meta.MetaResultSet metaResultSet : executeResult.resultSets) {
        results.add(toResponse(metaResultSet));
      }
      return new ExecuteResponse(results, false, serverLevelRpcMetadata);
    } catch (NoSuchStatementException e) {
      return new ExecuteResponse(null, true, serverLevelRpcMetadata);
    }
  }
}
 

public ExecuteBatchResponse apply(ExecuteBatchRequest request) {
  final Meta.StatementHandle h = new Meta.StatementHandle(request.connectionId,
      request.statementId, null);
  try {
    ExecuteBatchResult result;
    if (request.hasProtoUpdateBatches() && meta instanceof ProtobufMeta) {
      result = ((ProtobufMeta) meta).executeBatchProtobuf(h, request.getProtoUpdateBatches());
    } else {
      result = meta.executeBatch(h, request.parameterValues);
    }
    return new ExecuteBatchResponse(request.connectionId, request.statementId,
        result.updateCounts, false, serverLevelRpcMetadata);
  } catch (NoSuchStatementException e) {
    return new ExecuteBatchResponse(request.connectionId, request.statementId, null, true,
        serverLevelRpcMetadata);
  }
}
 

@Override
public ExecuteResult prepareAndExecute(StatementHandle statementHandle, String sql,
                                       long maxRowCount,
                                       int maxRowsInFirstFrame,
                                       PrepareCallback prepareCallback)
        throws NoSuchStatementException {
  try {
    MetaResultSet metaResultSet;
    synchronized (prepareCallback.getMonitor()) {
      prepareCallback.clear();
      ParserResult result = getConnection().parse(sql);
      metaResultSet = new PlanExecutor(statementHandle, getConnection(),
          connectionCache, maxRowCount).execute(result);
      prepareCallback.assign(metaResultSet.signature, metaResultSet.firstFrame,
          metaResultSet.updateCount);
    }
    prepareCallback.execute();
    return new ExecuteResult(ImmutableList.of(metaResultSet));
  } catch (Exception e) {
    throw propagate(e);
  }

}
 

@Override public StatementHandle prepare(ConnectionHandle ch, String sql,
    long maxRowCount) {
  final StatementHandle h = createStatement(ch);
  final CalciteConnectionImpl calciteConnection = getConnection();

  final CalciteServerStatement statement;
  try {
    statement = calciteConnection.server.getStatement(h);
  } catch (NoSuchStatementException e) {
    // Not possible. We just created a statement.
    throw new AssertionError("missing statement", e);
  }
  final Context context = statement.createPrepareContext();
  final CalcitePrepare.Query<Object> query = toQuery(context, sql);
  h.signature = calciteConnection.parseQuery(query, context, maxRowCount);
  statement.setSignature(h.signature);
  return h;
}
 

@Override
public ExecuteResult prepareAndExecute(StatementHandle h,
                                       String sql, long maxRowCount, int maxRowsInFirstFrame,
                                       PrepareCallback callback) throws NoSuchStatementException {
    begin();
    try {
        return super.prepareAndExecute(h, sql, maxRowCount, maxRowsInFirstFrame, callback);
    } finally {
        if (isAutoCommit()) {
            commit(connection.handle);
        }
    }
}
 

@Override
public Frame fetch(StatementHandle h, long offset,
                   int fetchMaxRowCount) throws NoSuchStatementException {
    begin();
    try {
        return super.fetch(h, offset, fetchMaxRowCount);
    } finally {
        if (isAutoCommit()) {
            commit(connection.handle);
        }
    }
}
 

@Override
public ExecuteResult execute(StatementHandle h,
                             List<TypedValue> parameterValues, int maxRowsInFirstFrame)
    throws NoSuchStatementException {
    begin();
    try {
        return super.execute(h, parameterValues, maxRowsInFirstFrame);
    } finally {
        if (isAutoCommit()) {
            commit(connection.handle);
        }
    }
}
 

@Override
public ExecuteBatchResult executeBatch(StatementHandle h,
                                       List<List<TypedValue>> parameterValueLists) throws NoSuchStatementException {
    begin();
    try {
        return super.executeBatch(h, parameterValueLists);
    } finally {
        if (isAutoCommit()) {
            commit(connection.handle);
        }
    }
}
 

@Override
public ExecuteBatchResult prepareAndExecuteBatch(
    final StatementHandle h,
    List<String> sqlCommands) throws NoSuchStatementException {
    begin();
    try {
        return super.prepareAndExecuteBatch(h, sqlCommands);
    } finally {
        if (isAutoCommit()) {
            commit(connection.handle);
        }
    }
}
 

@Override public void closeStatement(StatementHandle h) {
  final CalciteConnectionImpl calciteConnection = getConnection();
  final CalciteServerStatement stmt;
  try {
    stmt = calciteConnection.server.getStatement(h);
  } catch (NoSuchStatementException e) {
    // statement is not valid; nothing to do
    return;
  }
  // stmt.close(); // TODO: implement
  calciteConnection.server.removeStatement(h);
}
 

@Override public ExecuteResult prepareAndExecute(StatementHandle h,
    String sql, long maxRowCount, int maxRowsInFirstFrame,
    PrepareCallback callback) throws NoSuchStatementException {
  final CalcitePrepare.CalciteSignature<Object> signature;
  try {
    synchronized (callback.getMonitor()) {
      callback.clear();
      final CalciteConnectionImpl calciteConnection = getConnection();
      final CalciteServerStatement statement =
          calciteConnection.server.getStatement(h);
      final Context context = statement.createPrepareContext();
      final CalcitePrepare.Query<Object> query = toQuery(context, sql);
      signature = calciteConnection.parseQuery(query, context, maxRowCount);
      statement.setSignature(signature);
      final int updateCount;
      switch (signature.statementType) {
      case CREATE:
      case DROP:
      case ALTER:
      case OTHER_DDL:
        updateCount = 0; // DDL produces no result set
        break;
      default:
        updateCount = -1; // SELECT and DML produces result set
        break;
      }
      callback.assign(signature, null, updateCount);
    }
    callback.execute();
    final MetaResultSet metaResultSet =
        MetaResultSet.create(h.connectionId, h.id, false, signature, null);
    return new ExecuteResult(ImmutableList.of(metaResultSet));
  } catch (SQLException e) {
    throw new RuntimeException(e);
  }
  // TODO: share code with prepare and createIterable
}
 

@Override public ExecuteBatchResult executeBatch(StatementHandle h,
    List<List<TypedValue>> parameterValueLists) throws NoSuchStatementException {
  final List<Long> updateCounts = new ArrayList<>();
  for (List<TypedValue> parameterValueList : parameterValueLists) {
    ExecuteResult executeResult = execute(h, parameterValueList, -1);
    final long updateCount =
        executeResult.resultSets.size() == 1
            ? executeResult.resultSets.get(0).updateCount
            : -1L;
    updateCounts.add(updateCount);
  }
  return new ExecuteBatchResult(Longs.toArray(updateCounts));
}
 

public CalciteServerStatement getStatement(Meta.StatementHandle h)
    throws NoSuchStatementException {
  CalciteServerStatement statement = statementMap.get(h.id);
  if (statement == null) {
    throw new NoSuchStatementException(h);
  }
  return statement;
}
 

@SuppressWarnings("deprecation")
@Override public ExecuteResult prepareAndExecute(StatementHandle h, String sql, long maxRowCount,
    PrepareCallback callback) throws NoSuchStatementException {
  // The old semantics were that maxRowCount was also treated as the maximum number of
  // elements in the first Frame of results. A value of -1 would also preserve this, but an
  // explicit (positive) number is easier to follow, IMO.
  return prepareAndExecute(h, sql, maxRowCount, AvaticaUtils.toSaturatedInt(maxRowCount),
      callback);
}
 

@Override public Frame fetch(final StatementHandle h, final long offset,
    final int fetchMaxRowCount) throws NoSuchStatementException, MissingResultsException {
  try {
    return connection.invokeWithRetries(
        new CallableWithoutException<Frame>() {
          public Frame call() {
            final Service.FetchResponse response =
                service.apply(
                    new Service.FetchRequest(h.connectionId, h.id, offset, fetchMaxRowCount));
            if (response.missingStatement) {
              throw new RuntimeException(new NoSuchStatementException(h));
            }
            if (response.missingResults) {
              throw new RuntimeException(new MissingResultsException(h));
            }
            return response.frame;
          }
        });
  } catch (RuntimeException e) {
    Throwable cause = e.getCause();
    if (cause instanceof NoSuchStatementException) {
      throw (NoSuchStatementException) cause;
    } else if (cause instanceof MissingResultsException) {
      throw (MissingResultsException) cause;
    }
    throw e;
  }
}
 

public CalciteServerStatement getStatement(Meta.StatementHandle h)
    throws NoSuchStatementException {
  CalciteServerStatement statement = statementMap.get(h.id);
  if (statement == null) {
    throw new NoSuchStatementException(h);
  }
  return statement;
}
 

@Override public ExecuteBatchResult executeBatch(final StatementHandle h,
    final List<List<TypedValue>> parameterValues) throws NoSuchStatementException {
  return connection.invokeWithRetries(new CallableWithoutException<ExecuteBatchResult>() {
    @Override public ExecuteBatchResult call() {
      Service.ExecuteBatchResponse response =
          service.apply(new Service.ExecuteBatchRequest(h.connectionId, h.id, parameterValues));
      return new ExecuteBatchResult(response.updateCounts);
    }
  });
}
 

@Override
public ExecuteResult execute(StatementHandle h,
    List<TypedValue> parameterValues, int maxRowsInFirstFrame) throws NoSuchStatementException {
    final StatementInfo statementInfo = statementCache.getIfPresent(h.id);
    if (null == statementInfo) {
        throw new NoSuchStatementException(h);
    }
    String sql = h.signature.sql;
    for (TypedValue value : parameterValues) {
        if (value.type == Rep.BYTE || value.type == Rep.SHORT || value.type == Rep.LONG || value.type == Rep.DOUBLE
            || value.type == Rep.INTEGER || value.type == Rep.FLOAT) {
            sql = sql.replaceFirst("\\?", value.value.toString());
        } else {
            sql = sql.replaceFirst("\\?", "'" + value.value.toString() + "'");
        }
    }
    ExecuteResult executeResult = null;
    try {
        QuicksqlConnectionImpl connection = (QuicksqlConnectionImpl) getConnection(h.connectionId);
        String jdbcUrl = connection.getInfoByName("jdbcUrl");
        if (StringUtils.isNotBlank(jdbcUrl)) {
            executeResult = jdbcExecute(h, jdbcUrl, connection.getInfoByName("user"), connection
                .getInfoByName("password"), sql);
        } else {
            executeResult = getExecuteResultSet(h, connection, sql);
        }
    } catch (Exception e) {
        e.printStackTrace();
        throw new RuntimeException(e);
    }
    return executeResult;
}
 

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);
  }
}