class CreateSchemaTask implements DataDefinitionTask<CreateSchema> {

    @VisibleForTesting
    ListenableFuture<?> internalExecute(CreateSchema statement, Metadata metadata, AccessControl accessControl, Session session, List<Expression> parameters)
    {
        CatalogSchemaName schema = createCatalogSchemaName(session, statement, Optional.of(statement.getSchemaName()));

        // TODO: validate that catalog exists

        accessControl.checkCanCreateSchema(session.toSecurityContext(), schema);

        if (metadata.schemaExists(session, schema)) {
            if (!statement.isNotExists()) {
                throw semanticException(SCHEMA_ALREADY_EXISTS, statement, "Schema '%s' already exists", schema);
            }
            return immediateFuture(null);
        }

        CatalogName catalogName = metadata.getCatalogHandle(session, schema.getCatalogName())
                .orElseThrow(() -> new PrestoException(NOT_FOUND, "Catalog does not exist: " + schema.getCatalogName()));

        Map<String, Object> properties = metadata.getSchemaPropertyManager().getProperties(
                catalogName,
                schema.getCatalogName(),
                mapFromProperties(statement.getProperties()),
                session,
                metadata,
                accessControl,
                parameterExtractor(statement, parameters));

        PrestoPrincipal principal = getCreatePrincipal(statement, session, metadata);
        try {
            metadata.createSchema(session, schema, properties, principal);
        }
        catch (PrestoException e) {
            // connectors are not required to handle the ignoreExisting flag
            if (!e.getErrorCode().equals(ALREADY_EXISTS.toErrorCode()) || !statement.isNotExists()) {
                throw e;
            }
        }

        return immediateFuture(null);
    }

}