Revert "Added catalog and schema level access checks in USE statement"

presto-main/src/main/java/com/facebook/presto/execution/UseTask.java

@@ -14,34 +14,24 @@
 package com.facebook.presto.execution;
 
 import com.facebook.presto.Session;
-import com.facebook.presto.common.CatalogSchemaName;
-import com.facebook.presto.common.transaction.TransactionId;
 import com.facebook.presto.metadata.Metadata;
 import com.facebook.presto.spi.PrestoException;
-import com.facebook.presto.spi.analyzer.MetadataResolver;
 import com.facebook.presto.spi.security.AccessControl;
-import com.facebook.presto.spi.security.AccessControlContext;
-import com.facebook.presto.spi.security.AccessDeniedException;
-import com.facebook.presto.spi.security.Identity;
 import com.facebook.presto.sql.analyzer.SemanticException;
 import com.facebook.presto.sql.tree.Expression;
 import com.facebook.presto.sql.tree.Use;
 import com.facebook.presto.transaction.TransactionManager;
-import com.google.common.collect.ImmutableSet;
 import com.google.common.util.concurrent.ListenableFuture;
 
 import java.util.List;
 
 import static com.facebook.presto.spi.StandardErrorCode.NOT_FOUND;
-import static com.facebook.presto.spi.security.AccessDeniedException.denyCatalogAccess;
 import static com.facebook.presto.sql.analyzer.SemanticErrorCode.CATALOG_NOT_SPECIFIED;
 import static com.google.common.util.concurrent.Futures.immediateFuture;
-import static java.util.Locale.ENGLISH;
 
 public class UseTask
         implements SessionTransactionControlTask<Use>
 {
-    String catalog;
     @Override
     public String getName()
     {
@@ -59,38 +49,11 @@ public ListenableFuture<?> execute(
     {
         Session session = stateMachine.getSession();
 
-        TransactionId transactionId = session.getTransactionId().get();
-
-        Identity identity = session.getIdentity();
-
-        AccessControlContext context = session.getAccessControlContext();
-
         checkCatalogAndSessionPresent(statement, session);
 
         checkAndSetCatalog(statement, metadata, stateMachine, session);
 
-        String schema = statement.getSchema().getValue();
-
-        stateMachine.setSetSchema(schema);
-
-        if (!hasCatalogAccess(identity, context, catalog, accessControl)) {
-            denyCatalogAccess(catalog);
-        }
-
-        CatalogSchemaName name = new CatalogSchemaName(catalog, schema);
-
-        MetadataResolver metadataresolver = metadata.getMetadataResolver(session);
-        if (!metadataresolver.schemaExists(name)) {
-            throw new PrestoException(NOT_FOUND, "Schema does not exist: " + name);
-        }
-
-        if (!hasSchemaAccess(transactionId, identity, context, catalog, schema, accessControl)) {
-            throw new AccessDeniedException("Cannot access schema: " + name);
-        }
-
-        if (statement.getCatalog().isPresent()) {
-            stateMachine.setSetCatalog(catalog);
-        }
+        stateMachine.setSetSchema(statement.getSchema().getValueLowerCase());
 
         return immediateFuture(null);
     }
@@ -104,24 +67,12 @@ private void checkCatalogAndSessionPresent(Use statement, Session session)
 
     private void checkAndSetCatalog(Use statement, Metadata metadata, QueryStateMachine stateMachine, Session session)
     {
-        if (statement.getCatalog().isPresent() || session.getCatalog().isPresent()) {
-            catalog = statement.getCatalog()
-                    .map(identifier -> identifier.getValue().toLowerCase(ENGLISH))
-                    .orElseGet(() -> session.getCatalog().get());
+        if (statement.getCatalog().isPresent()) {
+            String catalog = statement.getCatalog().get().getValueLowerCase();
             if (!metadata.getCatalogHandle(session, catalog).isPresent()) {
                 throw new PrestoException(NOT_FOUND, "Catalog does not exist: " + catalog);
             }
             stateMachine.setSetCatalog(catalog);
         }
     }
-
-    private boolean hasCatalogAccess(Identity identity, AccessControlContext context, String catalog, AccessControl accessControl)
-    {
-        return !accessControl.filterCatalogs(identity, context, ImmutableSet.of(catalog)).isEmpty();
-    }
-
-    private boolean hasSchemaAccess(TransactionId transactionId, Identity identity, AccessControlContext context, String catalog, String schema, AccessControl accessControl)
-    {
-        return !accessControl.filterSchemas(transactionId, identity, context, catalog, ImmutableSet.of(schema)).isEmpty();
-    }
 }