@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);
    }
@Test
    public void testStartTransaction()
    {
        Session session = sessionBuilder()
                .setClientTransactionSupport()
                .build();
        TransactionManager transactionManager = createTestTransactionManager();
        QueryStateMachine stateMachine = createQueryStateMachine("START TRANSACTION", session, true, transactionManager, executor, metadata);
        assertFalse(stateMachine.getSession().getTransactionId().isPresent());
        StartTransactionTask startTransactionTask = new StartTransactionTask();
        getFutureValue(startTransactionTask.execute(new StartTransaction(ImmutableList.of()), transactionManager, metadata, new AllowAllAccessControl(), stateMachine, emptyList()));
        assertFalse(stateMachine.getQueryInfo(Optional.empty()).isClearTransactionId());
        assertTrue(stateMachine.getQueryInfo(Optional.empty()).getStartedTransactionId().isPresent());
        assertEquals(transactionManager.getAllTransactionInfos().size(), 1);

        TransactionInfo transactionInfo = transactionManager.getTransactionInfo(stateMachine.getQueryInfo(Optional.empty()).getStartedTransactionId().get());
        assertFalse(transactionInfo.isAutoCommitContext());
    }