class StartTransactionTask implements DataDefinitionTask<StartTransaction> {

    @Override
    public ListenableFuture<?> execute(StartTransaction statement, TransactionManager transactionManager, Metadata metadata, AccessControl accessControl, QueryStateMachine stateMachine, List<Expression> parameters)
    {
        Session session = stateMachine.getSession();
        if (!session.isClientTransactionSupport()) {
            throw new PrestoException(StandardErrorCode.INCOMPATIBLE_CLIENT, "Client does not support transactions");
        }
        if (session.getTransactionId().isPresent()) {
            throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Nested transactions not supported");
        }

        Optional<IsolationLevel> isolationLevel = extractIsolationLevel(statement);
        Optional<Boolean> readOnly = extractReadOnly(statement);

        TransactionId transactionId = transactionManager.beginTransaction(
                isolationLevel.orElse(TransactionManager.DEFAULT_ISOLATION),
                readOnly.orElse(TransactionManager.DEFAULT_READ_ONLY),
                false);

        stateMachine.setStartedTransactionId(transactionId);

        // Since the current session does not contain this new transaction ID, we need to manually mark it as inactive
        // when this statement completes.
        transactionManager.trySetInactive(transactionId);

        return immediateFuture(null);
    }

}