You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2019/05/01 00:03:04 UTC
[impala] branch master updated (66fda62 -> c1b0a07)
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git.
from 66fda62 IMPALA-8419 : Validate event processing related configurations
new 3ad5b3f IMPALA-7973: Add support for fine grained events processing for partition level HMS events.
new d533cb1 IMPALA-8280, IMPALA-8281: Add support for show grant user/group with Ranger
new c1b0a07 IMPALA-8455: Gracefully handle failed loads in GET_TABLES req
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
fe/src/main/cup/sql-parser.cup | 10 +
.../impala/analysis/ShowGrantPrincipalStmt.java | 38 +-
.../ranger/RangerAuthorizationFactory.java | 2 +-
.../ranger/RangerCatalogdAuthorizationManager.java | 46 +--
.../ranger/RangerImpaladAuthorizationManager.java | 431 +++++++++++++++++++++
.../impala/authorization/ranger/RangerUtil.java | 73 ++++
.../impala/catalog/CatalogServiceCatalog.java | 29 +-
.../java/org/apache/impala/catalog/FeCatalog.java | 3 +
.../java/org/apache/impala/catalog/Principal.java | 17 +-
.../impala/catalog/events/MetastoreEvents.java | 197 ++++++++--
.../apache/impala/catalog/local/LocalCatalog.java | 10 +
.../java/org/apache/impala/service/MetadataOp.java | 7 +-
.../impala/analysis/AnalyzeAuthStmtsTest.java | 4 +-
.../org/apache/impala/analysis/ParserTest.java | 5 +-
.../events/MetastoreEventsProcessorTest.java | 61 ++-
.../impala/catalog/local/LocalCatalogTest.java | 25 +-
tests/authorization/test_ranger.py | 301 +++++++++++---
17 files changed, 1056 insertions(+), 203 deletions(-)
create mode 100644 fe/src/main/java/org/apache/impala/authorization/ranger/RangerImpaladAuthorizationManager.java
create mode 100644 fe/src/main/java/org/apache/impala/authorization/ranger/RangerUtil.java
[impala] 02/03: IMPALA-8280,
IMPALA-8281: Add support for show grant user/group with Ranger
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit d533cb1d43879cfa3892726bcae6316dea212754
Author: Austin Nobis <an...@cloudera.com>
AuthorDate: Mon Apr 22 09:13:19 2019 -0500
IMPALA-8280, IMPALA-8281: Add support for show grant user/group with Ranger
Add support for SHOW GRANT statements for Apache Ranger. This patch also
adds the RangerImpaladAuthorizationManager as the show grant statement
is called from impalad. The new supported syntax is:
SHOW GRANT USER/GROUP <id> ON server;
SHOW GRANT USER/GROUP <id> ON database <db>;
SHOW GRANT USER/GROUP <id> ON uri <uri>;
SHOW GRANT USER/GROUP <id> ON table <db>.<table>;
SHOW GRANT USER/GROUP <id> ON column <db>.<table>.<column>;
The following syntax is valid SQL, but is not supported currently by the
Apache Ranger integration with Impala:
SHOW GRANT USER/GROUP <id>
Testing:
- Ran all FE unit tests
- Ran authorization E2E tests
- Updated test_ranger to use show grant statement for verification of
granted privileges
Change-Id: Ic46fb9fc36c9e11ec78d5840d22eb0668150c2a4
Reviewed-on: http://gerrit.cloudera.org:8080/13074
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
fe/src/main/cup/sql-parser.cup | 10 +
.../impala/analysis/ShowGrantPrincipalStmt.java | 38 +-
.../ranger/RangerAuthorizationFactory.java | 2 +-
.../ranger/RangerCatalogdAuthorizationManager.java | 46 +--
.../ranger/RangerImpaladAuthorizationManager.java | 431 +++++++++++++++++++++
.../impala/authorization/ranger/RangerUtil.java | 73 ++++
.../java/org/apache/impala/catalog/Principal.java | 17 +-
.../impala/analysis/AnalyzeAuthStmtsTest.java | 4 +-
.../org/apache/impala/analysis/ParserTest.java | 5 +-
tests/authorization/test_ranger.py | 301 +++++++++++---
10 files changed, 804 insertions(+), 123 deletions(-)
diff --git a/fe/src/main/cup/sql-parser.cup b/fe/src/main/cup/sql-parser.cup
index 909d20e..17e7dd4 100644
--- a/fe/src/main/cup/sql-parser.cup
+++ b/fe/src/main/cup/sql-parser.cup
@@ -960,6 +960,14 @@ show_grant_principal_stmt ::=
RESULT = new ShowGrantPrincipalStmt(name, type,
PrivilegeSpec.createTableScopedPriv(TPrivilegeLevel.ALL, tbl_name));
:}
+ | KW_SHOW KW_GRANT principal_type:type ident_or_default:name KW_ON KW_COLUMN
+ column_name:col_name
+ {:
+ RESULT = new ShowGrantPrincipalStmt(name, type,
+ PrivilegeSpec.createColumnScopedPriv(TPrivilegeLevel.SELECT,
+ col_name.getTableName(),
+ Collections.singletonList(col_name.getColumnName())));
+ :}
| KW_SHOW KW_GRANT principal_type:type ident_or_default:name KW_ON uri_ident:uri_kw
STRING_LITERAL:uri
{:
@@ -1060,6 +1068,8 @@ privilege ::=
principal_type ::=
KW_ROLE
{: RESULT = TPrincipalType.ROLE; :}
+ | KW_GROUP
+ {: RESULT = TPrincipalType.GROUP; :}
| IDENT:user
{:
parser.checkIdentKeyword("USER", user);
diff --git a/fe/src/main/java/org/apache/impala/analysis/ShowGrantPrincipalStmt.java b/fe/src/main/java/org/apache/impala/analysis/ShowGrantPrincipalStmt.java
index be967ae..34f15b7 100644
--- a/fe/src/main/java/org/apache/impala/analysis/ShowGrantPrincipalStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/ShowGrantPrincipalStmt.java
@@ -57,29 +57,25 @@ public class ShowGrantPrincipalStmt extends AuthorizationStmt {
"empty.", Principal.toString(principalType_),
Principal.toString(principalType_).toUpperCase()));
}
- principal_ = analyzer.getCatalog().getAuthPolicy().getPrincipal(name_,
- principalType_);
- // If it's a role, we can determine if it doesn't exist here. For a user
- // it's considered non-existent if it doesn't have any groups, but we cannot
- // access the group information from analysis. The group check for a user
- // is done in AuthorizationPolicy.getPrincipalPrivileges
- if (principal_ == null) {
- switch (principalType_) {
- case ROLE:
- throw new AnalysisException(String.format("%s '%s' does not exist.",
- Principal.toString(principalType_), name_));
- case USER:
- // Create a user object here because it's possible the user does not exist in
- // Sentry, but still exists according to the OS, or Hadoop, or other custom
- // group mapping provider.
- principal_ = Principal.newInstance(name_, principalType_, new HashSet<>());
- break;
- default:
- throw new AnalysisException(String.format("Unexpected TPrincipalType: %s",
- Principal.toString(principalType_)));
- }
+ switch(principalType_) {
+ case ROLE:
+ principal_ = analyzer.getCatalog().getAuthPolicy().getPrincipal(name_,
+ principalType_);
+ if (principal_ == null) {
+ throw new AnalysisException(String.format("%s '%s' " +
+ "does not exist.", Principal.toString(principalType_), name_));
+ }
+ break;
+ case USER:
+ case GROUP:
+ principal_ = Principal.newInstance(name_, principalType_, new HashSet<>());
+ break;
+ default:
+ throw new AnalysisException(String.format("Unexpected TPrincipalType %s",
+ principalType_.name()));
}
+
if (privilegeSpec_ != null) privilegeSpec_.analyze(analyzer);
}
diff --git a/fe/src/main/java/org/apache/impala/authorization/ranger/RangerAuthorizationFactory.java b/fe/src/main/java/org/apache/impala/authorization/ranger/RangerAuthorizationFactory.java
index 8126408..f7248cc 100644
--- a/fe/src/main/java/org/apache/impala/authorization/ranger/RangerAuthorizationFactory.java
+++ b/fe/src/main/java/org/apache/impala/authorization/ranger/RangerAuthorizationFactory.java
@@ -80,7 +80,7 @@ public class RangerAuthorizationFactory implements AuthorizationFactory {
Supplier<? extends AuthorizationChecker> authzChecker) {
Preconditions.checkArgument(authzChecker.get() instanceof RangerAuthorizationChecker);
- return new RangerCatalogdAuthorizationManager(() ->
+ return new RangerImpaladAuthorizationManager(() ->
((RangerAuthorizationChecker) authzChecker.get()).getRangerImpalaPlugin());
}
diff --git a/fe/src/main/java/org/apache/impala/authorization/ranger/RangerCatalogdAuthorizationManager.java b/fe/src/main/java/org/apache/impala/authorization/ranger/RangerCatalogdAuthorizationManager.java
index 28a9ba4..f772c00 100644
--- a/fe/src/main/java/org/apache/impala/authorization/ranger/RangerCatalogdAuthorizationManager.java
+++ b/fe/src/main/java/org/apache/impala/authorization/ranger/RangerCatalogdAuthorizationManager.java
@@ -217,18 +217,18 @@ public class RangerCatalogdAuthorizationManager implements AuthorizationManager
// is common to other resources, we need to grant that privilege to those
// resources.
if (p.getColumn_name() != null || p.getTable_name() != null) {
- requests.add(createRequest.apply(createColumnResource(p)));
+ requests.add(createRequest.apply(RangerUtil.createColumnResource(p)));
} else if (p.getUri() != null) {
- requests.add(createRequest.apply(createUriResource(p)));
+ requests.add(createRequest.apply(RangerUtil.createUriResource(p)));
} else if (p.getDb_name() != null) {
// DB is used by column and function resources.
- requests.add(createRequest.apply(createColumnResource(p)));
- requests.add(createRequest.apply(createFunctionResource(p)));
+ requests.add(createRequest.apply(RangerUtil.createColumnResource(p)));
+ requests.add(createRequest.apply(RangerUtil.createFunctionResource(p)));
} else {
// Server is used by column, function, and URI resources.
- requests.add(createRequest.apply(createColumnResource(p)));
- requests.add(createRequest.apply(createFunctionResource(p)));
- requests.add(createRequest.apply(createUriResource(p)));
+ requests.add(createRequest.apply(RangerUtil.createColumnResource(p)));
+ requests.add(createRequest.apply(RangerUtil.createFunctionResource(p)));
+ requests.add(createRequest.apply(RangerUtil.createUriResource(p)));
}
}
@@ -260,36 +260,4 @@ public class RangerCatalogdAuthorizationManager implements AuthorizationManager
return request;
}
-
- private static Map<String, String> createColumnResource(TPrivilege privilege) {
- Map<String, String> resource = new HashMap<>();
-
- resource.put(RangerImpalaResourceBuilder.DATABASE, getOrAll(privilege.getDb_name()));
- resource.put(RangerImpalaResourceBuilder.TABLE, getOrAll(privilege.getTable_name()));
- resource.put(RangerImpalaResourceBuilder.COLUMN,
- getOrAll(privilege.getColumn_name()));
-
- return resource;
- }
-
- private static Map<String, String> createUriResource(TPrivilege privilege) {
- Map<String, String> resource = new HashMap<>();
- String uri = privilege.getUri();
- resource.put(RangerImpalaResourceBuilder.URL, uri == null ? "*" : uri);
-
- return resource;
- }
-
- private static Map<String, String> createFunctionResource(TPrivilege privilege) {
- Map<String, String> resource = new HashMap<>();
-
- resource.put(RangerImpalaResourceBuilder.DATABASE, getOrAll(privilege.getDb_name()));
- resource.put(RangerImpalaResourceBuilder.UDF, "*");
-
- return resource;
- }
-
- private static String getOrAll(String resource) {
- return (resource == null) ? "*" : resource;
- }
}
diff --git a/fe/src/main/java/org/apache/impala/authorization/ranger/RangerImpaladAuthorizationManager.java b/fe/src/main/java/org/apache/impala/authorization/ranger/RangerImpaladAuthorizationManager.java
new file mode 100644
index 0000000..244d273
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/authorization/ranger/RangerImpaladAuthorizationManager.java
@@ -0,0 +1,431 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.authorization.ranger;
+
+import com.google.common.collect.Sets;
+import org.apache.hadoop.hive.metastore.api.PrincipalType;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.impala.authorization.AuthorizationDelta;
+import org.apache.impala.authorization.AuthorizationManager;
+import org.apache.impala.authorization.User;
+import org.apache.impala.catalog.Type;
+import org.apache.impala.common.ImpalaException;
+import org.apache.impala.thrift.TColumn;
+import org.apache.impala.thrift.TCreateDropRoleParams;
+import org.apache.impala.thrift.TDdlExecResponse;
+import org.apache.impala.thrift.TGrantRevokePrivParams;
+import org.apache.impala.thrift.TGrantRevokeRoleParams;
+import org.apache.impala.thrift.TPrincipalType;
+import org.apache.impala.thrift.TPrivilege;
+import org.apache.impala.thrift.TPrivilegeLevel;
+import org.apache.impala.thrift.TResultRow;
+import org.apache.impala.thrift.TResultSet;
+import org.apache.impala.thrift.TResultSetMetadata;
+import org.apache.impala.thrift.TShowGrantPrincipalParams;
+import org.apache.impala.thrift.TShowRolesParams;
+import org.apache.impala.thrift.TShowRolesResult;
+import org.apache.impala.util.ClassUtil;
+import org.apache.impala.util.TResultRowBuilder;
+import org.apache.ranger.plugin.model.RangerPolicy;
+import org.apache.ranger.plugin.policyengine.RangerAccessRequest;
+import org.apache.ranger.plugin.policyengine.RangerAccessRequestImpl;
+import org.apache.ranger.plugin.policyengine.RangerAccessResourceImpl;
+import org.apache.ranger.plugin.policyengine.RangerPolicyEngine;
+import org.apache.ranger.plugin.policyengine.RangerResourceACLs;
+import org.apache.ranger.plugin.policyengine.RangerResourceACLs.AccessResult;
+import org.apache.ranger.plugin.service.RangerAuthContext;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+/**
+ * An implementation of {@link AuthorizationManager} for Impalad using Ranger.
+ *
+ * Operations here make requests to Ranger via the {@link RangerImpalaPlugin} to
+ * manage privileges for users.
+ *
+ * Operations not supported by Ranger will throw an {@link UnsupportedOperationException}.
+ */
+public class RangerImpaladAuthorizationManager implements AuthorizationManager {
+ private static final String ANY = "*";
+
+ private final Supplier<RangerImpalaPlugin> plugin_;
+ private final Supplier<RangerAuthContext> authContext_;
+
+ public RangerImpaladAuthorizationManager(Supplier<RangerImpalaPlugin> pluginSupplier) {
+ plugin_ = pluginSupplier;
+ authContext_ = () -> plugin_.get().createRangerAuthContext();
+ }
+
+ @Override
+ public void createRole(User requestingUser, TCreateDropRoleParams params,
+ TDdlExecResponse response) throws ImpalaException {
+ throw new UnsupportedOperationException(String.format(
+ "%s is not supported in Impalad", ClassUtil.getMethodName()));
+ }
+
+ @Override
+ public void dropRole(User requestingUser, TCreateDropRoleParams params,
+ TDdlExecResponse response) throws ImpalaException {
+ throw new UnsupportedOperationException(String.format(
+ "%s is not supported in Impalad", ClassUtil.getMethodName()));
+ }
+
+ @Override
+ public TShowRolesResult getRoles(TShowRolesParams params) throws ImpalaException {
+ throw new UnsupportedOperationException(String.format(
+ "%s is not supported in Impalad", ClassUtil.getMethodName()));
+ }
+
+ @Override
+ public void grantRoleToGroup(User requestingUser, TGrantRevokeRoleParams params,
+ TDdlExecResponse response) throws ImpalaException {
+ throw new UnsupportedOperationException(String.format(
+ "%s is not supported in Impalad", ClassUtil.getMethodName()));
+ }
+
+ @Override
+ public void revokeRoleFromGroup(User requestingUser, TGrantRevokeRoleParams params,
+ TDdlExecResponse response) throws ImpalaException {
+ throw new UnsupportedOperationException(String.format(
+ "%s is not supported in Impalad", ClassUtil.getMethodName()));
+ }
+
+ @Override
+ public void grantPrivilegeToRole(User requestingUser, TGrantRevokePrivParams params,
+ TDdlExecResponse response) throws ImpalaException {
+ throw new UnsupportedOperationException(String.format(
+ "%s is not supported in Impalad", ClassUtil.getMethodName()));
+ }
+
+ @Override
+ public void revokePrivilegeFromRole(User requestingUser, TGrantRevokePrivParams params,
+ TDdlExecResponse response) throws ImpalaException {
+ throw new UnsupportedOperationException(String.format(
+ "%s is not supported in Impalad", ClassUtil.getMethodName()));
+ }
+
+ @Override
+ public void grantPrivilegeToUser(User requestingUser, TGrantRevokePrivParams params,
+ TDdlExecResponse response) throws ImpalaException {
+ throw new UnsupportedOperationException(String.format(
+ "%s is not supported in Impalad", ClassUtil.getMethodName()));
+ }
+
+ @Override
+ public void revokePrivilegeFromUser(User requestingUser, TGrantRevokePrivParams params,
+ TDdlExecResponse response) throws ImpalaException {
+ throw new UnsupportedOperationException(String.format(
+ "%s is not supported in Impalad", ClassUtil.getMethodName()));
+ }
+
+ @Override
+ public void grantPrivilegeToGroup(User requestingUser, TGrantRevokePrivParams params,
+ TDdlExecResponse response) throws ImpalaException {
+ throw new UnsupportedOperationException(String.format(
+ "%s is not supported in Impalad", ClassUtil.getMethodName()));
+ }
+
+ @Override
+ public void revokePrivilegeFromGroup(User requestingUser, TGrantRevokePrivParams params,
+ TDdlExecResponse response) throws ImpalaException {
+ throw new UnsupportedOperationException(String.format(
+ "%s is not supported in Impalad", ClassUtil.getMethodName()));
+ }
+
+ private static Set<String> getGroups(String principal) {
+ UserGroupInformation ugi = UserGroupInformation.createRemoteUser(principal);
+ return Sets.newHashSet(ugi.getGroupNames());
+ }
+
+ private static Optional<String> getResourceName(String resourceType,
+ String resourceName, AccessResult accessResult) {
+ RangerPolicy.RangerPolicyResource rangerPolicyResource =
+ accessResult.getPolicy().getResources().get(resourceType);
+
+ if (rangerPolicyResource == null) {
+ return Optional.empty();
+ }
+ boolean nameIsPresent = rangerPolicyResource.getValues().contains(resourceName);
+
+ return nameIsPresent ? Optional.of(resourceName) : Optional.of(ANY);
+ }
+
+ private static boolean isDelegateAdmin(AccessResult accessResult, String privilegeLevel,
+ String principal, TPrincipalType type) {
+ for (RangerPolicy.RangerPolicyItem item : accessResult.getPolicy().getPolicyItems()) {
+ switch (type) {
+ case USER:
+ if (item.getUsers().contains(principal) &&
+ item.getAccesses().stream()
+ .anyMatch(rpia -> rpia.getType().equals(privilegeLevel))) {
+ return item.getDelegateAdmin();
+ }
+ break;
+ case GROUP:
+ if (item.getGroups().contains(principal)) {
+ return item.getDelegateAdmin();
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException(String.format("Unsupported principal " +
+ "type %s", type));
+ }
+ }
+ return false;
+ }
+
+ private static RangerResultRow toResultRow(String rangerPrivilegeLevel,
+ String principal, TPrincipalType type, AccessResult accessResult,
+ TPrivilege privilege) {
+ TPrivilege rangerPrivilege = new TPrivilege();
+ rangerPrivilege.setScope(privilege.getScope());
+ boolean grantOption = isDelegateAdmin(accessResult, rangerPrivilegeLevel, principal,
+ type);
+
+ // Ignore hive privileges which may not exist.
+ TPrivilegeLevel level;
+ try {
+ level = TPrivilegeLevel.valueOf(rangerPrivilegeLevel.toUpperCase());
+ } catch (IllegalArgumentException e) {
+ if (rangerPrivilegeLevel.equals(RangerAuthorizationChecker.UPDATE_ACCESS_TYPE)) {
+ level = TPrivilegeLevel.INSERT;
+ } else {
+ return null;
+ }
+ }
+
+ Date createTime = accessResult.getPolicy().getCreateTime();
+ Long longTime = createTime == null ? null : createTime.getTime();
+
+ Optional<String> database = getResourceName(RangerImpalaResourceBuilder.DATABASE,
+ privilege.getDb_name(), accessResult);
+ Optional<String> table = getResourceName(RangerImpalaResourceBuilder.TABLE,
+ privilege.getTable_name(), accessResult);
+ Optional<String> column = getResourceName(RangerImpalaResourceBuilder.COLUMN,
+ privilege.getColumn_name(), accessResult);
+ Optional<String> uri = getResourceName(RangerImpalaResourceBuilder.URL,
+ privilege.getUri(), accessResult);
+ Optional<String> udf = getResourceName(RangerImpalaResourceBuilder.UDF, ANY,
+ accessResult);
+
+ switch (privilege.getScope()) {
+ case COLUMN:
+ if (!column.isPresent() || column.get().equals("*")) return null;
+ case TABLE:
+ if (!table.isPresent() || table.get().equals("*")) return null;
+ case DATABASE:
+ if (!database.isPresent() || database.get().equals("*")) return null;
+ break;
+ case URI:
+ if (!uri.isPresent() || uri.get().equals("*")) return null;
+ break;
+ case SERVER:
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported privilege scope " +
+ privilege.getScope());
+ }
+
+ return new RangerResultRow(type, principal, database.orElse(""), table.orElse(""),
+ column.orElse(""), uri.orElse(""), udf.orElse(""), level, grantOption, longTime);
+ }
+
+ private static List<RangerAccessRequest> buildAccessRequests(TPrivilege privilege) {
+ List<Map<String, String>> resources = new ArrayList<>();
+
+ if (privilege == null) {
+ throw new UnsupportedOperationException("SHOW GRANT is not supported without a" +
+ " defined resource in Ranger.");
+ }
+
+ if (privilege.getColumn_name() != null || privilege.getTable_name() != null) {
+ resources.add(RangerUtil.createColumnResource(privilege));
+ } else if (privilege.getUri() != null) {
+ resources.add(RangerUtil.createUriResource(privilege));
+ } else if (privilege.getDb_name() != null) {
+ // DB is used by column and function resources.
+ resources.add(RangerUtil.createColumnResource(privilege));
+ resources.add(RangerUtil.createFunctionResource(privilege));
+ } else {
+ // Server is used by column, function, and URI resources.
+ resources.add(RangerUtil.createColumnResource(privilege));
+ resources.add(RangerUtil.createUriResource(privilege));
+ resources.add(RangerUtil.createFunctionResource(privilege));
+ }
+
+ List<RangerAccessRequest> requests = new ArrayList<>();
+ for (Map<String, String> resource : resources) {
+ requests.add(new RangerAccessRequestImpl(
+ new RangerAccessResourceImpl(Collections.unmodifiableMap(resource)),
+ RangerPolicyEngine.ANY_ACCESS, null, null));
+ }
+ return requests;
+ }
+
+ private static List<RangerResultRow> aclToPrivilege(Map<String, AccessResult> acls,
+ String principal, TPrivilege privilege, TPrincipalType type) {
+ return acls.entrySet()
+ .stream()
+ .map(en -> toResultRow(en.getKey(), principal, type, en.getValue(), privilege))
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public TResultSet getPrivileges(TShowGrantPrincipalParams params)
+ throws ImpalaException {
+ List<RangerAccessRequest> requests = buildAccessRequests(params.privilege);
+ Set<TResultRow> resultSet = new TreeSet<>();
+ TResultSet result = new TResultSet();
+
+ result.setSchema(RangerResultRow.getSchema());
+ result.setRows(new ArrayList<>());
+
+ for (RangerAccessRequest request : requests) {
+ List<RangerResultRow> resultRows;
+ RangerResourceACLs acls = authContext_.get().getResourceACLs(request);
+
+ switch (params.principal_type) {
+ case USER:
+ resultRows = new ArrayList<>(aclToPrivilege(
+ acls.getUserACLs().getOrDefault(params.name, Collections.emptyMap()),
+ params.name, params.privilege, TPrincipalType.USER));
+ for (String group : getGroups(params.name)) {
+ resultRows.addAll(aclToPrivilege(
+ acls.getGroupACLs().getOrDefault(group, Collections.emptyMap()),
+ params.name, params.privilege, TPrincipalType.GROUP));
+ }
+ break;
+ case GROUP:
+ resultRows = new ArrayList<>(aclToPrivilege(
+ acls.getGroupACLs().getOrDefault(params.name, Collections.emptyMap()),
+ params.name, params.privilege, TPrincipalType.GROUP));
+ break;
+ default:
+ throw new UnsupportedOperationException(String.format("Unsupported principal " +
+ "type %s.", params.principal_type));
+ }
+
+ boolean all = resultRows.stream().anyMatch(row ->
+ row.privilege_ == TPrivilegeLevel.ALL);
+
+ List<RangerResultRow> rows = all ? resultRows.stream()
+ .filter(row -> row.privilege_ == TPrivilegeLevel.ALL)
+ .collect(Collectors.toList()) : resultRows;
+
+ rows.forEach(principal -> resultSet.add(principal.toResultRow()));
+ }
+ resultSet.forEach(result::addToRows);
+
+ return result;
+ }
+
+ @Override
+ public void updateDatabaseOwnerPrivilege(String serverName, String databaseName,
+ String oldOwner, PrincipalType oldOwnerType, String newOwner,
+ PrincipalType newOwnerType, TDdlExecResponse response) throws ImpalaException {
+ }
+
+ @Override
+ public void updateTableOwnerPrivilege(String serverName, String databaseName,
+ String tableName, String oldOwner, PrincipalType oldOwnerType, String newOwner,
+ PrincipalType newOwnerType, TDdlExecResponse response) throws ImpalaException {
+ }
+
+ @Override
+ public AuthorizationDelta refreshAuthorization(boolean resetVersions) {
+ // TODO: IMPALA-8293 (part 2)
+ return new AuthorizationDelta();
+ }
+
+ private static class RangerResultRow {
+ private final TPrincipalType principalType_;
+ private final String principalName_;
+ private final String database_;
+ private final String table_;
+ private final String column_;
+ private final String uri_;
+ private final String udf_;
+ private final TPrivilegeLevel privilege_;
+ private final boolean grantOption_;
+ private final Long createTime_;
+
+ public RangerResultRow(TPrincipalType principalType, String principalName,
+ String database, String table, String column, String uri, String udf,
+ TPrivilegeLevel privilege, boolean grantOption, Long createTime) {
+ this.principalType_ = principalType;
+ this.principalName_ = principalName;
+ this.database_ = database;
+ this.table_ = table;
+ this.column_ = column;
+ this.uri_ = uri;
+ this.udf_ = udf;
+ this.privilege_ = privilege;
+ this.grantOption_ = grantOption;
+ this.createTime_ = createTime;
+ }
+
+ public static TResultSetMetadata getSchema() {
+ TResultSetMetadata schema = new TResultSetMetadata();
+
+ schema.addToColumns(new TColumn("principal_type", Type.STRING.toThrift()));
+ schema.addToColumns(new TColumn("principal_name", Type.STRING.toThrift()));
+ schema.addToColumns(new TColumn("database", Type.STRING.toThrift()));
+ schema.addToColumns(new TColumn("table", Type.STRING.toThrift()));
+ schema.addToColumns(new TColumn("column", Type.STRING.toThrift()));
+ schema.addToColumns(new TColumn("uri", Type.STRING.toThrift()));
+ schema.addToColumns(new TColumn("udf", Type.STRING.toThrift()));
+ schema.addToColumns(new TColumn("privilege", Type.STRING.toThrift()));
+ schema.addToColumns(new TColumn("grant_option", Type.BOOLEAN.toThrift()));
+ schema.addToColumns(new TColumn("create_time", Type.STRING.toThrift()));
+
+ return schema;
+ }
+
+ public TResultRow toResultRow() {
+ TResultRowBuilder rowBuilder = new TResultRowBuilder();
+
+ rowBuilder.add(principalType_.name().toUpperCase());
+ rowBuilder.add(principalName_);
+ rowBuilder.add(database_);
+ rowBuilder.add(table_);
+ rowBuilder.add(column_);
+ rowBuilder.add(uri_);
+ rowBuilder.add(udf_);
+ rowBuilder.add(privilege_.name().toLowerCase());
+ rowBuilder.add(grantOption_);
+ if (createTime_ == null) {
+ rowBuilder.add(null);
+ } else {
+ rowBuilder.add(createTime_);
+ }
+
+ return rowBuilder.get();
+ }
+ }
+}
diff --git a/fe/src/main/java/org/apache/impala/authorization/ranger/RangerUtil.java b/fe/src/main/java/org/apache/impala/authorization/ranger/RangerUtil.java
new file mode 100644
index 0000000..2fa61d9
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/authorization/ranger/RangerUtil.java
@@ -0,0 +1,73 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.authorization.ranger;
+
+import org.apache.impala.thrift.TPrivilege;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Collection of static functions to support Apache Ranger implementation
+ */
+public class RangerUtil {
+ private RangerUtil() { }
+
+ /**
+ * Creates a column resource for Ranger. Column resources also include
+ * database and table information.
+ */
+ public static Map<String, String> createColumnResource(TPrivilege privilege) {
+ Map<String, String> resource = new HashMap<>();
+
+ resource.put(RangerImpalaResourceBuilder.DATABASE, getOrAll(privilege.getDb_name()));
+ resource.put(RangerImpalaResourceBuilder.TABLE, getOrAll(privilege.getTable_name()));
+ resource.put(RangerImpalaResourceBuilder.COLUMN,
+ getOrAll(privilege.getColumn_name()));
+
+ return resource;
+ }
+
+ /**
+ * Creates a URI resource for Ranger. In Ranger a URI is known as a URL.
+ */
+ public static Map<String, String> createUriResource(TPrivilege privilege) {
+ Map<String, String> resource = new HashMap<>();
+ String uri = privilege.getUri();
+ resource.put(RangerImpalaResourceBuilder.URL, uri == null ? "*" : uri);
+
+ return resource;
+ }
+
+ /**
+ * Creates a function resource for Ranger. Function resources also include
+ * database information.
+ */
+ public static Map<String, String> createFunctionResource(TPrivilege privilege) {
+ Map<String, String> resource = new HashMap<>();
+
+ resource.put(RangerImpalaResourceBuilder.DATABASE, getOrAll(privilege.getDb_name()));
+ resource.put(RangerImpalaResourceBuilder.UDF, "*");
+
+ return resource;
+ }
+
+ private static String getOrAll(String resource) {
+ return (resource == null) ? "*" : resource;
+ }
+}
diff --git a/fe/src/main/java/org/apache/impala/catalog/Principal.java b/fe/src/main/java/org/apache/impala/catalog/Principal.java
index 1012fb2..39aeef1 100644
--- a/fe/src/main/java/org/apache/impala/catalog/Principal.java
+++ b/fe/src/main/java/org/apache/impala/catalog/Principal.java
@@ -170,6 +170,21 @@ public abstract class Principal extends CatalogObjectImpl {
public TPrincipalType getPrincipalType() { return principal_.getPrincipal_type(); }
public static String toString(TPrincipalType type) {
- return type == TPrincipalType.ROLE ? "Role" : "User";
+ String principal;
+ switch (type) {
+ case ROLE:
+ principal = "Role";
+ break;
+ case USER:
+ principal = "User";
+ break;
+ case GROUP:
+ principal = "Group";
+ break;
+ default:
+ throw new IllegalStateException(String.format("Unsupported principal type " +
+ "%s.", type));
+ }
+ return principal;
}
}
diff --git a/fe/src/test/java/org/apache/impala/analysis/AnalyzeAuthStmtsTest.java b/fe/src/test/java/org/apache/impala/analysis/AnalyzeAuthStmtsTest.java
index e4ec95b..e3ce330 100644
--- a/fe/src/test/java/org/apache/impala/analysis/AnalyzeAuthStmtsTest.java
+++ b/fe/src/test/java/org/apache/impala/analysis/AnalyzeAuthStmtsTest.java
@@ -101,11 +101,12 @@ public class AnalyzeAuthStmtsTest extends FrontendTestBase {
@Test
public void AnalyzeShowGrantPrincipal() {
- for (String type: new String[]{"ROLE myRole", "USER myUser"}) {
+ for (String type: new String[]{"ROLE myRole", "USER myUser", "GROUP myGroup"}) {
AnalyzesOk(String.format("SHOW GRANT %s", type));
AnalyzesOk(String.format("SHOW GRANT %s ON SERVER", type));
AnalyzesOk(String.format("SHOW GRANT %s ON DATABASE functional", type));
AnalyzesOk(String.format("SHOW GRANT %s ON TABLE functional.alltypes", type));
+ AnalyzesOk(String.format("SHOW GRANT %s ON COLUMN functional.alltypes.year", type));
AnalyzesOk(String.format("SHOW GRANT %s ON URI 'hdfs:////test-warehouse//foo'",
type));
@@ -117,6 +118,7 @@ public class AnalyzeAuthStmtsTest extends FrontendTestBase {
}
AnalysisError("SHOW GRANT ROLE does_not_exist",
"Role 'does_not_exist' does not exist.");
+
AnalysisError("SHOW GRANT ROLE does_not_exist ON SERVER",
"Role 'does_not_exist' does not exist.");
diff --git a/fe/src/test/java/org/apache/impala/analysis/ParserTest.java b/fe/src/test/java/org/apache/impala/analysis/ParserTest.java
index d9db2b0..cc30926 100644
--- a/fe/src/test/java/org/apache/impala/analysis/ParserTest.java
+++ b/fe/src/test/java/org/apache/impala/analysis/ParserTest.java
@@ -3739,7 +3739,7 @@ public class ParserTest extends FrontendTestBase {
@Test
public void TestShowGrantPrincipal() {
- for (String type: new String[]{"ROLE", "USER"}) {
+ for (String type: new String[]{"ROLE", "USER", "GROUP"}) {
// Show all grants on a particular principal type.
ParsesOk(String.format("SHOW GRANT %s foo", type));
@@ -3748,12 +3748,15 @@ public class ParserTest extends FrontendTestBase {
ParsesOk(String.format("SHOW GRANT %s foo ON DATABASE foo", type));
ParsesOk(String.format("SHOW GRANT %s foo ON TABLE foo", type));
ParsesOk(String.format("SHOW GRANT %s foo ON TABLE foo.bar", type));
+ ParsesOk(String.format("SHOW GRANT %s foo ON COLUMN bar.baz", type));
+ ParsesOk(String.format("SHOW GRANT %s foo ON COLUMN foo.bar.baz", type));
ParsesOk(String.format("SHOW GRANT %s foo ON URI '/abc/123'", type));
ParserError(String.format("SHOW GRANT %s", type));
ParserError(String.format("SHOW GRANT %s foo ON SERVER foo", type));
ParserError(String.format("SHOW GRANT %s foo ON DATABASE", type));
ParserError(String.format("SHOW GRANT %s foo ON TABLE", type));
+ ParserError(String.format("SHOW GRANT %s foo ON COLUMN", type));
ParserError(String.format("SHOW GRANT %s foo ON URI abc", type));
}
ParserError("SHOW GRANT FOO bar");
diff --git a/tests/authorization/test_ranger.py b/tests/authorization/test_ranger.py
index 7e14a76..f2900b0 100644
--- a/tests/authorization/test_ranger.py
+++ b/tests/authorization/test_ranger.py
@@ -26,6 +26,7 @@ import time
from getpass import getuser
from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
+ADMIN = "admin"
RANGER_AUTH = ("admin", "admin")
RANGER_HOST = "http://localhost:6080"
IMPALAD_ARGS = "--server-name=server1 --ranger_service_type=hive " \
@@ -38,32 +39,31 @@ class TestRanger(CustomClusterTestSuite):
"""
Tests for Apache Ranger integration with Apache Impala.
"""
+
@pytest.mark.execute_serially
@CustomClusterTestSuite.with_args(
- impalad_args=IMPALAD_ARGS, catalogd_args=CATALOGD_ARGS)
+ impalad_args=IMPALAD_ARGS, catalogd_args=CATALOGD_ARGS)
def test_grant_revoke_with_catalog_v1(self, unique_name):
"""Tests grant/revoke with catalog v1."""
self._test_grant_revoke(unique_name)
@pytest.mark.execute_serially
@CustomClusterTestSuite.with_args(
- impalad_args="{0} {1}".format(IMPALAD_ARGS, "--use_local_catalog=true"),
- catalogd_args="{0} {1}".format(CATALOGD_ARGS,
- "--use_local_catalog=true "
- "--catalog_topic_mode=minimal"))
+ impalad_args="{0} {1}".format(IMPALAD_ARGS, "--use_local_catalog=true"),
+ catalogd_args="{0} {1}".format(CATALOGD_ARGS,
+ "--use_local_catalog=true "
+ "--catalog_topic_mode=minimal"))
def test_grant_revoke_with_local_catalog(self, unique_name):
"""Tests grant/revoke with catalog v2 (local catalog)."""
self._test_grant_revoke(unique_name)
def _test_grant_revoke(self, unique_name):
user = getuser()
- admin = "admin"
admin_client = self.create_impala_client()
- user_client = self.create_impala_client()
unique_database = unique_name + "_db"
unique_table = unique_name + "_tbl"
group = grp.getgrnam(getuser()).gr_name
- test_data = [(user, "user"), (group, "group")]
+ test_data = [(user, "USER"), (group, "GROUP")]
for data in test_data:
ident = data[0]
@@ -72,96 +72,271 @@ class TestRanger(CustomClusterTestSuite):
try:
# Set-up temp database/table
admin_client.execute("drop database if exists {0} cascade"
- .format(unique_database), user=admin)
- admin_client.execute("create database {0}".format(unique_database), user=admin)
+ .format(unique_database), user=ADMIN)
+ admin_client.execute("create database {0}".format(unique_database), user=ADMIN)
admin_client.execute("create table {0}.{1} (x int)"
- .format(unique_database, unique_table), user=admin)
+ .format(unique_database, unique_table), user=ADMIN)
self.execute_query_expect_success(admin_client,
"grant select on database {0} to {1} {2}"
- .format(unique_database, kw, ident), user=admin)
+ .format(unique_database, kw, ident), user=ADMIN)
# TODO: IMPALA-8293 use refresh authorization
time.sleep(10)
- self.execute_query_expect_success(user_client, "show tables in {0}"
- .format(unique_database), user=user)
+ result = self.execute_query("show grant {0} {1} on database {2}"
+ .format(kw, ident, unique_database))
+ TestRanger._check_privileges(result, [
+ [kw, ident, unique_database, "", "", "", "*", "select", "false"],
+ [kw, ident, unique_database, "*", "*", "", "", "select", "false"]])
self.execute_query_expect_success(admin_client,
"revoke select on database {0} from {1} "
"{2}".format(unique_database, kw, ident),
- user=admin)
+ user=ADMIN)
# TODO: IMPALA-8293 use refresh authorization
time.sleep(10)
- self.execute_query_expect_failure(user_client, "show tables in {0}"
- .format(unique_database))
+ result = self.execute_query("show grant {0} {1} on database {2}"
+ .format(kw, ident, unique_database))
+ TestRanger._check_privileges(result, [])
finally:
admin_client.execute("revoke select on database {0} from {1} {2}"
- .format(unique_database, kw, ident), user=admin)
+ .format(unique_database, kw, ident), user=ADMIN)
admin_client.execute("drop database if exists {0} cascade"
- .format(unique_database), user=admin)
+ .format(unique_database), user=ADMIN)
@CustomClusterTestSuite.with_args(
- impalad_args=IMPALAD_ARGS, catalogd_args=CATALOGD_ARGS)
+ impalad_args=IMPALAD_ARGS, catalogd_args=CATALOGD_ARGS)
def test_grant_option(self, unique_name):
user1 = getuser()
- user2 = unique_name + "_user"
- admin = "admin"
admin_client = self.create_impala_client()
- user1_client = self.create_impala_client()
- user2_client = self.create_impala_client()
unique_database = unique_name + "_db"
unique_table = unique_name + "_tbl"
- id = self._add_ranger_user(user2)
try:
# Set-up temp database/table
admin_client.execute("drop database if exists {0} cascade".format(unique_database),
- user=admin)
- admin_client.execute("create database {0}".format(unique_database), user=admin)
+ user=ADMIN)
+ admin_client.execute("create database {0}".format(unique_database), user=ADMIN)
admin_client.execute("create table {0}.{1} (x int)"
- .format(unique_database, unique_table), user=admin)
+ .format(unique_database, unique_table), user=ADMIN)
# Give user 1 the ability to grant select privileges on unique_database
self.execute_query_expect_success(admin_client,
"grant select on database {0} to user {1} with "
"grant option".format(unique_database, user1),
- user=admin)
- # TODO: IMPALA-8293 use refresh authorization
- time.sleep(10)
- # User 1 grants select privilege to user 2
- self.execute_query_expect_success(user1_client,
- "grant select on database {0} to user {1}"
- .format(unique_database, user2), user=user1)
- # TODO: IMPALA-8293 use refresh authorization
- time.sleep(10)
- # User 2 exercises select privilege
- self.execute_query_expect_success(user2_client, "show tables in {0}"
- .format(unique_database), user=user2)
- # User 1 revokes select privilege from user 2
- self.execute_query_expect_success(user1_client,
- "revoke select on database {0} from user "
- "{1}".format(unique_database, user2), user=user1)
+ user=ADMIN)
# TODO: IMPALA-8293 use refresh authorization
time.sleep(10)
- # User 2 can no longer select because the privilege was revoked
- self.execute_query_expect_failure(user2_client, "show tables in {0}"
- .format(unique_database))
+ # Verify user 1 has with_grant privilege on unique_database
+ result = self.execute_query("show grant user {0} on database {1}"
+ .format(user1, unique_database))
+ TestRanger._check_privileges(result, [
+ ["USER", user1, unique_database, "", "", "", "*", "select", "true"],
+ ["USER", user1, unique_database, "*", "*", "", "", "select", "true"]])
+
# Revoke privilege granting from user 1
self.execute_query_expect_success(admin_client, "revoke grant option for select "
- "on database {0} from user {1}"
- .format(unique_database, user1), user=admin)
+ "on database {0} from user {1}"
+ .format(unique_database, user1), user=ADMIN)
# TODO: IMPALA-8293 use refresh authorization
time.sleep(10)
# User 1 can no longer grant privileges on unique_database
- self.execute_query_expect_failure(user1_client,
- "grant select on database {0} to user {1}"
- .format(unique_database, user2), user=user1)
+ result = self.execute_query("show grant user {0} on database {1}"
+ .format(user1, unique_database))
+ TestRanger._check_privileges(result, [])
finally:
- admin_client.execute("revoke select on database {0} from user {1}"
- .format(unique_database, user2), user=admin)
admin_client.execute("revoke grant option for select on database {0} from user {1}"
- .format(unique_database, user1), user=admin)
+ .format(unique_database, user1), user=ADMIN)
admin_client.execute("drop database if exists {0} cascade".format(unique_database),
- user=admin)
- self._remove_ranger_user(id)
+ user=ADMIN)
+
+ @CustomClusterTestSuite.with_args(
+ impalad_args=IMPALAD_ARGS, catalogd_args=CATALOGD_ARGS)
+ def test_show_grant(self, unique_name):
+ user = getuser()
+ group = grp.getgrnam(getuser()).gr_name
+ test_data = [(user, "USER"), (group, "GROUP")]
+ admin_client = self.create_impala_client()
+ unique_db = unique_name + "_db"
+ unique_table = unique_name + "_tbl"
+
+ try:
+ # Create test database/table
+ admin_client.execute("drop database if exists {0} cascade".format(unique_db),
+ user=ADMIN)
+ admin_client.execute("create database {0}".format(unique_db), user=ADMIN)
+ admin_client.execute("create table {0}.{1} (x int)"
+ .format(unique_db, unique_table), user=ADMIN)
+
+ for data in test_data:
+ # Test basic show grant functionality for user/group
+ self._test_show_grant_basic(admin_client, data[1], data[0], unique_db,
+ unique_table)
+ # Test that omitting ON <resource> results in failure
+ self._test_show_grant_without_on(data[1], data[0])
+
+ # Test ALL privilege hides other privileges
+ self._test_show_grant_mask(admin_client, user)
+
+ # Test USER inherits privileges for their GROUP
+ self._test_show_grant_user_group(admin_client, user, group, unique_db)
+
+ # Test that show grant without ON a resource fails
+
+ finally:
+ admin_client.execute("drop database if exists {0} cascade".format(unique_db),
+ user=ADMIN)
+
+ def _test_show_grant_without_on(self, kw, ident):
+ self.execute_query_expect_failure(self.client, "show grant {0} {1}".format(kw, ident))
+
+ def _test_show_grant_user_group(self, admin_client, user, group, unique_db):
+ try:
+ result = self.client.execute("show grant user {0} on database {1}"
+ .format(user, unique_db))
+ TestRanger._check_privileges(result, [])
+
+ admin_client.execute("grant select on database {0} to group {1}"
+ .format(unique_db, group))
+ time.sleep(10)
+ result = self.client.execute("show grant user {0} on database {1}"
+ .format(user, unique_db))
+ TestRanger._check_privileges(result, [
+ ["GROUP", user, unique_db, "", "", "", "*", "select", "false"],
+ ["GROUP", user, unique_db, "*", "*", "", "", "select", "false"]])
+ finally:
+ admin_client.execute("revoke select on database {0} from group {1}"
+ .format(unique_db, group))
+
+ def _test_show_grant_mask(self, admin_client, user):
+ privileges = ["select", "insert", "create", "alter", "drop", "refresh"]
+ try:
+ for privilege in privileges:
+ admin_client.execute("grant {0} on server to user {1}".format(privilege, user))
+ time.sleep(10)
+ result = self.client.execute("show grant user {0} on server".format(user))
+ TestRanger._check_privileges(result, [
+ ["USER", user, "", "", "", "*", "", "alter", "false"],
+ ["USER", user, "", "", "", "*", "", "create", "false"],
+ ["USER", user, "", "", "", "*", "", "drop", "false"],
+ ["USER", user, "", "", "", "*", "", "insert", "false"],
+ ["USER", user, "", "", "", "*", "", "select", "false"],
+ ["USER", user, "*", "", "", "", "*", "alter", "false"],
+ ["USER", user, "*", "", "", "", "*", "create", "false"],
+ ["USER", user, "*", "", "", "", "*", "drop", "false"],
+ ["USER", user, "*", "", "", "", "*", "insert", "false"],
+ ["USER", user, "*", "", "", "", "*", "select", "false"],
+ ["USER", user, "*", "*", "*", "", "", "alter", "false"],
+ ["USER", user, "*", "*", "*", "", "", "create", "false"],
+ ["USER", user, "*", "*", "*", "", "", "drop", "false"],
+ ["USER", user, "*", "*", "*", "", "", "insert", "false"],
+ ["USER", user, "*", "*", "*", "", "", "select", "false"]])
+
+ admin_client.execute("grant all on server to user {0}".format(user))
+ time.sleep(10)
+ result = self.client.execute("show grant user {0} on server".format(user))
+ TestRanger._check_privileges(result, [
+ ["USER", user, "", "", "", "*", "", "all", "false"],
+ ["USER", user, "*", "", "", "", "*", "all", "false"],
+ ["USER", user, "*", "*", "*", "", "", "all", "false"]])
+ finally:
+ for privilege in privileges:
+ admin_client.execute("revoke {0} on server from user {1}".format(privilege, user))
+
+ def _test_show_grant_basic(self, admin_client, kw, id, unique_database, unique_table):
+ uri = '/tmp'
+ try:
+ # Grant server privileges and verify
+ admin_client.execute("grant all on server to {0} {1}".format(kw, id), user=ADMIN)
+ time.sleep(10)
+ result = self.client.execute("show grant {0} {1} on server".format(kw, id))
+ TestRanger._check_privileges(result, [
+ [kw, id, "", "", "", "*", "", "all", "false"],
+ [kw, id, "*", "", "", "", "*", "all", "false"],
+ [kw, id, "*", "*", "*", "", "", "all", "false"]])
+
+ # Revoke server privileges and verify
+ admin_client.execute("revoke all on server from {0} {1}".format(kw, id))
+ time.sleep(10)
+ result = self.client.execute("show grant {0} {1} on server".format(kw, id))
+ TestRanger._check_privileges(result, [])
+
+ # Grant uri privileges and verify
+ admin_client.execute("grant all on uri '{0}' to {1} {2}"
+ .format(uri, kw, id))
+ time.sleep(10)
+ result = self.client.execute("show grant {0} {1} on uri '{2}'"
+ .format(kw, id, uri))
+ TestRanger._check_privileges(result, [
+ [kw, id, "", "", "", "hdfs://localhost:20500" + uri, "", "all", "false"]])
+
+ # Revoke uri privileges and verify
+ admin_client.execute("revoke all on uri '{0}' from {1} {2}"
+ .format(uri, kw, id))
+ time.sleep(10)
+ result = self.client.execute("show grant {0} {1} on uri '{2}'"
+ .format(kw, id, uri))
+ TestRanger._check_privileges(result, [])
+
+ # Grant database privileges and verify
+ admin_client.execute("grant select on database {0} to {1} {2}"
+ .format(unique_database, kw, id))
+ time.sleep(10)
+ result = self.client.execute("show grant {0} {1} on database {2}"
+ .format(kw, id, unique_database))
+ TestRanger._check_privileges(result, [
+ [kw, id, unique_database, "", "", "", "*", "select", "false"],
+ [kw, id, unique_database, "*", "*", "", "", "select", "false"]])
+
+ # Revoke database privileges and verify
+ admin_client.execute("revoke select on database {0} from {1} {2}"
+ .format(unique_database, kw, id))
+ time.sleep(10)
+ result = self.client.execute("show grant {0} {1} on database {2}"
+ .format(kw, id, unique_database))
+ TestRanger._check_privileges(result, [])
+
+ # Grant table privileges and verify
+ admin_client.execute("grant select on table {0}.{1} to {2} {3}"
+ .format(unique_database, unique_table, kw, id))
+ time.sleep(10)
+ result = self.client.execute("show grant {0} {1} on table {2}.{3}"
+ .format(kw, id, unique_database, unique_table))
+ TestRanger._check_privileges(result, [
+ [kw, id, unique_database, unique_table, "*", "", "", "select", "false"]])
+
+ # Revoke table privileges and verify
+ admin_client.execute("revoke select on table {0}.{1} from {2} {3}"
+ .format(unique_database, unique_table, kw, id))
+ time.sleep(10)
+ result = self.client.execute("show grant {0} {1} on table {2}.{3}"
+ .format(kw, id, unique_database, unique_table))
+ TestRanger._check_privileges(result, [])
+
+ # Grant column privileges and verify
+ admin_client.execute("grant select(x) on table {0}.{1} to {2} {3}"
+ .format(unique_database, unique_table, kw, id))
+ time.sleep(10)
+ result = self.client.execute("show grant {0} {1} on column {2}.{3}.x"
+ .format(kw, id, unique_database, unique_table))
+ TestRanger._check_privileges(result, [
+ [kw, id, unique_database, unique_table, "x", "", "", "select", "false"]])
+
+ # Revoke column privileges and verify
+ admin_client.execute("revoke select(x) on table {0}.{1} from {2} {3}"
+ .format(unique_database, unique_table, kw, id))
+ time.sleep(10)
+ result = self.client.execute("show grant {0} {1} on column {2}.{3}.x"
+ .format(kw, id, unique_database, unique_table))
+ TestRanger._check_privileges(result, [])
+ finally:
+ admin_client.execute("revoke all on server from {0} {1}".format(kw, id))
+ admin_client.execute("revoke all on uri '{0}' from {1} {2}"
+ .format(uri, kw, id))
+ admin_client.execute("revoke select on database {0} from {1} {2}"
+ .format(unique_database, kw, id))
+ admin_client.execute("revoke select on table {0}.{1} from {2} {3}"
+ .format(unique_database, unique_table, kw, id))
+ admin_client.execute("revoke select(x) on table {0}.{1} from {2} {3}"
+ .format(unique_database, unique_table, kw, id))
def _add_ranger_user(self, user):
data = {"name": user, "password": "password123", "userRoleList": ["ROLE_USER"]}
@@ -170,9 +345,17 @@ class TestRanger(CustomClusterTestSuite):
r = requests.post("{0}/service/xusers/secure/users".format(RANGER_HOST),
auth=RANGER_AUTH,
json=data, headers=headers)
- return json.loads(r.content)['id']
+ return json.loads(r.content)["id"]
def _remove_ranger_user(self, id):
r = requests.delete("{0}/service/xusers/users/{1}?forceDelete=true"
.format(RANGER_HOST, id), auth=RANGER_AUTH)
- assert r.status_code < 300 and r.status_code >= 200
+ assert 300 > r.status_code >= 200
+
+ @staticmethod
+ def _check_privileges(result, expected):
+ def columns(row):
+ cols = row.split("\t")
+ return cols[0:len(cols) - 1]
+
+ assert map(columns, result.data) == expected
[impala] 03/03: IMPALA-8455: Gracefully handle failed loads in
GET_TABLES req
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit c1b0a073938c144e9bf33901bd4df6dcda0f09ec
Author: Bharath Vissapragada <bh...@cloudera.com>
AuthorDate: Mon Apr 29 21:30:05 2019 -0700
IMPALA-8455: Gracefully handle failed loads in GET_TABLES req
Some HS2 clients (like Hue) rely on the GET_TABLES request to
populate table meta information. In local catalog mode, for any
table loading failures, this call throws a cryptic exception
that the callers do not expect.
The original ImpaladCatalog implementation gracefully handled this
by silently ignoring such tables. This patch replicates that
behavior in LocalCatalog mode.
Testing: Added a unit test that consistently fails without the patch.
Change-Id: I89cc2c585aa83a9aea5343f0d0383ba72bafd618
Reviewed-on: http://gerrit.cloudera.org:8080/13186
Reviewed-by: Bharath Vissapragada <bh...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../java/org/apache/impala/catalog/FeCatalog.java | 3 +++
.../apache/impala/catalog/local/LocalCatalog.java | 10 +++++++++
.../java/org/apache/impala/service/MetadataOp.java | 7 ++++--
.../impala/catalog/local/LocalCatalogTest.java | 25 +++++++++++++++++++++-
4 files changed, 42 insertions(+), 3 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/catalog/FeCatalog.java b/fe/src/main/java/org/apache/impala/catalog/FeCatalog.java
index 10396f9..e0e63c2 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FeCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FeCatalog.java
@@ -43,6 +43,9 @@ public interface FeCatalog {
FeTable getTable(String db_name, String table_name)
throws DatabaseNotFoundException;
+ /** @see Catalog#getTableNoThrow(String, String) */
+ FeTable getTableNoThrow(String db_name, String table_name);
+
/** @see Catalog#getTCatalogObject(TCatalogObject) */
TCatalogObject getTCatalogObject(TCatalogObject objectDesc)
throws CatalogException;
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalog.java b/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalog.java
index b2f3f9c..1868073 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/LocalCatalog.java
@@ -121,6 +121,16 @@ public class LocalCatalog implements FeCatalog {
}
@Override
+ public FeTable getTableNoThrow(String dbName, String tableName) {
+ try {
+ return getTable(dbName, tableName);
+ } catch (Exception e) {
+ // pass
+ }
+ return null;
+ }
+
+ @Override
public TCatalogObject getTCatalogObject(TCatalogObject objectDesc)
throws CatalogException {
// TODO(todd): this probably makes the /catalog page not load with an error.
diff --git a/fe/src/main/java/org/apache/impala/service/MetadataOp.java b/fe/src/main/java/org/apache/impala/service/MetadataOp.java
index fa4a0e7..3d85aa9 100644
--- a/fe/src/main/java/org/apache/impala/service/MetadataOp.java
+++ b/fe/src/main/java/org/apache/impala/service/MetadataOp.java
@@ -301,8 +301,11 @@ public class MetadataOp {
List<String> tableComments = Lists.newArrayList();
List<String> tableTypes = Lists.newArrayList();
for (String tabName: fe.getTableNames(db.getName(), tablePatternMatcher, user)) {
- FeTable table = catalog.getTable(db.getName(), tabName);
- if (table == null) continue;
+ FeTable table = catalog.getTableNoThrow(db.getName(), tabName);
+ if (table == null) {
+ result.missingTbls.add(new TableName(db.getName(), tabName));
+ continue;
+ }
String comment = null;
List<Column> columns = Lists.newArrayList();
diff --git a/fe/src/test/java/org/apache/impala/catalog/local/LocalCatalogTest.java b/fe/src/test/java/org/apache/impala/catalog/local/LocalCatalogTest.java
index a9f0d9d..990577b 100644
--- a/fe/src/test/java/org/apache/impala/catalog/local/LocalCatalogTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/local/LocalCatalogTest.java
@@ -22,8 +22,10 @@ import static org.junit.Assert.*;
import java.util.List;
import java.util.Set;
+import org.apache.hive.service.rpc.thrift.TGetTablesReq;
import org.apache.impala.analysis.Expr;
import org.apache.impala.analysis.ToSqlUtils;
+import org.apache.impala.authorization.NoopAuthorizationFactory;
import org.apache.impala.catalog.CatalogTest;
import org.apache.impala.catalog.ColumnStats;
import org.apache.impala.catalog.FeCatalogUtils;
@@ -36,7 +38,11 @@ import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
import org.apache.impala.catalog.Type;
import org.apache.impala.service.BackendConfig;
import org.apache.impala.service.FeSupport;
+import org.apache.impala.service.Frontend;
import org.apache.impala.thrift.TCatalogObjectType;
+import org.apache.impala.thrift.TMetadataOpRequest;
+import org.apache.impala.thrift.TMetadataOpcode;
+import org.apache.impala.thrift.TResultRow;
import org.apache.impala.thrift.TResultSet;
import org.apache.impala.util.MetaStoreUtil;
import org.apache.impala.util.PatternMatcher;
@@ -52,12 +58,14 @@ import com.google.common.collect.Iterables;
public class LocalCatalogTest {
private CatalogdMetaProvider provider_;
private LocalCatalog catalog_;
+ private Frontend fe_;
@Before
- public void setupCatalog() {
+ public void setupCatalog() throws Exception {
FeSupport.loadLibrary();
provider_ = new CatalogdMetaProvider(BackendConfig.INSTANCE.getBackendCfg());
catalog_ = new LocalCatalog(provider_, /*defaultKuduMasterHosts=*/null);
+ fe_ = new Frontend(new NoopAuthorizationFactory(), catalog_);
}
@Test
@@ -348,4 +356,19 @@ public class LocalCatalogTest {
assertEquals(table_with_header.parseSkipHeaderLineCount(error), 1);
assertEquals(error.length(), 0);
}
+
+ /**
+ * Test GET_TABLES request on an Impala incompatible table. It should be silently
+ * ignored.
+ */
+ @Test
+ public void testGetTables() throws Exception {
+ TMetadataOpRequest req = new TMetadataOpRequest();
+ req.opcode = TMetadataOpcode.GET_TABLES;
+ req.get_tables_req = new TGetTablesReq();
+ req.get_tables_req.setSchemaName("functional");
+ req.get_tables_req.setTableName("bad_serde");
+ TResultSet resp = fe_.execHiveServer2MetadataOp(req);
+ assertEquals(0, resp.rows.size());
+ }
}
[impala] 01/03: IMPALA-7973: Add support for fine grained events
processing for partition level HMS events.
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 3ad5b3fba202bf6809986a86f5e041da656f0a88
Author: Anurag Mantripragada <an...@cloudera.com>
AuthorDate: Fri Apr 12 18:17:22 2019 -0700
IMPALA-7973: Add support for fine grained events processing for
partition level HMS events.
This patch adds support for fine grained updates for add/drop/alter
partition events.
Currently, partition events invalidate the table. This can be
expensive for large tables. Here, we refresh affected partitions
in case of add/drop/alter partition events. HMS processes add/drop
partitions in a transaction, which means there may be multiple
partitions affected in a single add/drop event. We try to refresh all
these partitions in a loop. If any of the partition refresh fails,
we throw MetastoreNotificationNeedsInvalidateException to mandate a
manual invalidate for event processing to continue.
Testing:
Modified pre-existing tests for partition events to instead test if
partitions are added/dropped/altered when event processing is enabled.
Change-Id: I213401329f3965dd81055197792ccf8a05368af5
Reviewed-on: http://gerrit.cloudera.org:8080/13111
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../impala/catalog/CatalogServiceCatalog.java | 29 ++-
.../impala/catalog/events/MetastoreEvents.java | 197 +++++++++++++++++----
.../events/MetastoreEventsProcessorTest.java | 61 ++++---
3 files changed, 210 insertions(+), 77 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
index ea4486e..62f4b50 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
@@ -35,6 +35,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.apache.hadoop.fs.Hdfs;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.protocol.CachePoolEntry;
@@ -42,6 +43,7 @@ import org.apache.hadoop.hdfs.protocol.CachePoolInfo;
import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
+import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.UnknownDBException;
import org.apache.impala.analysis.TableName;
import org.apache.impala.authorization.AuthorizationDelta;
@@ -2107,32 +2109,29 @@ public class CatalogServiceCatalog extends Catalog {
}
/**
- * Refresh table if exists. Returns true if reloadTable() succeeds, false
- * otherwise. Throws CatalogException if reloadTable() is unsuccessful. Throws
- * DatabaseNotFoundException if Db doesn't exist.
+ * Refresh partition if it exists. Returns true if reload of the partition succeeds,
+ * false otherwise.
+ * @throws CatalogException if partition reload is unsuccessful.
+ * @throws DatabaseNotFoundException if Db doesn't exist.
*/
- public boolean refreshTableIfExists(String dbName, String tblName)
- throws CatalogException {
+ public boolean reloadPartitionIfExists(String dbName, String tblName,
+ List<TPartitionKeyValue> tPartSpec) throws CatalogException {
Table table = getTable(dbName, tblName);
if (table == null || table instanceof IncompleteTable) return false;
- reloadTable(table);
+ reloadPartition(table, tPartSpec);
return true;
}
/**
- * Refresh partition if exists. Returns true if reloadPartitition() succeeds, false
- * otherwise. Throws CatalogException if reloadPartition() is unsuccessful. Throws
+ * Refresh table if exists. Returns true if reloadTable() succeeds, false
+ * otherwise. Throws CatalogException if reloadTable() is unsuccessful. Throws
* DatabaseNotFoundException if Db doesn't exist.
*/
- public boolean refreshPartitionIfExists(String dbName, String tblName,
- Map<String, String> partSpec) throws CatalogException {
+ public boolean refreshTableIfExists(String dbName, String tblName)
+ throws CatalogException {
Table table = getTable(dbName, tblName);
if (table == null || table instanceof IncompleteTable) return false;
- List<TPartitionKeyValue> tPartSpec = new ArrayList<>(partSpec.size());
- for (Map.Entry<String, String> entry : partSpec.entrySet()) {
- tPartSpec.add(new TPartitionKeyValue(entry.getKey(), entry.getValue()));
- }
- reloadPartition(table, tPartSpec);
+ reloadTable(table);
return true;
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEvents.java b/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEvents.java
index c3c3280..493b637 100644
--- a/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEvents.java
+++ b/fe/src/main/java/org/apache/impala/catalog/events/MetastoreEvents.java
@@ -18,13 +18,11 @@
package org.apache.impala.catalog.events;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
@@ -53,10 +51,12 @@ import org.apache.impala.catalog.TableNotFoundException;
import org.apache.impala.common.Metrics;
import org.apache.impala.common.Pair;
import org.apache.impala.common.Reference;
+import org.apache.impala.thrift.TPartitionKeyValue;
import org.apache.impala.thrift.TTableName;
import org.apache.impala.util.ClassUtil;
import org.slf4j.LoggerFactory;
import org.slf4j.Logger;
+import org.apache.hadoop.hive.common.FileUtils;
/**
* Main class which provides Metastore event objects for various event types. Also
@@ -566,6 +566,38 @@ public class MetastoreEvents {
if (val == null || val.isEmpty()) return null;
return Boolean.valueOf(val);
}
+
+ /**
+ * Util method to create partition key-value map from HMS Partition objects.
+ */
+ protected static List<TPartitionKeyValue> getTPartitionSpecFromHmsPartition(
+ org.apache.hadoop.hive.metastore.api.Table msTbl, Partition partition) {
+ List<TPartitionKeyValue> tPartSpec = new ArrayList<>();
+ List<org.apache.hadoop.hive.metastore.api.FieldSchema> fsList =
+ msTbl.getPartitionKeys();
+ List<String> partVals = partition.getValues();
+ Preconditions.checkNotNull(partVals);
+ Preconditions.checkState(fsList.size() == partVals.size());
+ for (int i = 0; i < fsList.size(); i++) {
+ tPartSpec.add(new TPartitionKeyValue(fsList.get(i).getName(), partVals.get(i)));
+ }
+ return tPartSpec;
+ }
+
+ /**
+ * Util method to create a partition spec string out of a TPartitionKeyValue objects.
+ */
+ protected static String constructPartitionStringFromTPartitionSpec(
+ List<TPartitionKeyValue> tPartSpec) {
+ List<String> partitionCols = new ArrayList<>();
+ List<String> partitionVals = new ArrayList<>();
+ for (TPartitionKeyValue kv: tPartSpec) {
+ partitionCols.add(kv.getName());
+ partitionVals.add(kv.getValue());
+ }
+ String partString = FileUtils.makePartName(partitionCols, partitionVals);
+ return partString;
+ }
}
/**
@@ -736,39 +768,31 @@ public class MetastoreEvents {
private void processPartitionInserts() throws MetastoreNotificationException {
// For partitioned table, refresh the partition only.
Preconditions.checkNotNull(insertPartition_);
- Map<String, String> partSpec = new HashMap<>();
- List<org.apache.hadoop.hive.metastore.api.FieldSchema> fsList =
- msTbl_.getPartitionKeys();
- List<String> partVals = insertPartition_.getValues();
- Preconditions.checkNotNull(partVals);
- Preconditions.checkState(fsList.size() == partVals.size());
- for (int i = 0; i < fsList.size(); i++) {
- partSpec.put(fsList.get(i).getName(), partVals.get(i));
- }
+ List<TPartitionKeyValue> tPartSpec = getTPartitionSpecFromHmsPartition(msTbl_,
+ insertPartition_);
try {
// Ignore event if table or database is not in catalog. Throw exception if
// refresh fails.
- if (!catalog_.refreshPartitionIfExists(dbName_, tblName_, partSpec)) {
+ if (!catalog_.reloadPartitionIfExists(dbName_, tblName_, tPartSpec)) {
debugLog("Refresh of table {} partition {} after insert "
+ "event failed as the table is not present in the catalog.",
- getFullyQualifiedTblName(), Joiner.on(",").withKeyValueSeparator("=")
- .join(partSpec));
+ getFullyQualifiedTblName(), (tPartSpec));
} else {
infoLog("Table {} partition {} has been refreshed after insert.",
- getFullyQualifiedTblName(), Joiner.on(",").withKeyValueSeparator("=")
- .join(partSpec));
+ getFullyQualifiedTblName(),
+ constructPartitionStringFromTPartitionSpec(tPartSpec));
}
} catch (DatabaseNotFoundException e) {
debugLog("Refresh of table {} partition {} for insert "
+ "event failed as the database is not present in the catalog.",
- getFullyQualifiedTblName(), Joiner.on(",").withKeyValueSeparator("=")
- .join(partSpec));
+ getFullyQualifiedTblName(),
+ constructPartitionStringFromTPartitionSpec(tPartSpec));
} catch (CatalogException e) {
throw new MetastoreNotificationNeedsInvalidateException(debugString("Refresh "
+ "partition on table {} partition {} failed. Event processing cannot "
+ "continue. Issue and invalidate command to reset the event processor "
+ "state.", getFullyQualifiedTblName(),
- Joiner.on(",").withKeyValueSeparator("=").join(partSpec)));
+ constructPartitionStringFromTPartitionSpec(tPartSpec)));
}
}
@@ -1280,6 +1304,7 @@ public class MetastoreEvents {
public static class AddPartitionEvent extends TableInvalidatingEvent {
private final Partition lastAddedPartition_;
+ private final List<Partition> addedPartitions_;
/**
* Prevent instantiation from outside should use MetastoreEventFactory instead
@@ -1296,16 +1321,14 @@ public class MetastoreEvents {
MetastoreEventsProcessor.getMessageFactory()
.getDeserializer()
.getAddPartitionMessage(event.getMessage());
- List<Map<String, String>> keyValues = addPartitionMessage_.getPartitions();
- Preconditions.checkState(keyValues.size() > 0);
- ArrayList<Partition> addedPartitions =
+ addedPartitions_ =
Lists.newArrayList(addPartitionMessage_.getPartitionObjs());
- Preconditions.checkState(addedPartitions.size() > 0);
+ Preconditions.checkState(addedPartitions_.size() > 0);
// when multiple partitions are added in HMS they are all added as one transaction
// Hence all the partitions which are present in the message must have the same
// serviceId and version if it is set. hence it is fine to just look at the
// last added partition in the list and use it for the self-event ids
- lastAddedPartition_ = addedPartitions.get(addedPartitions.size() - 1);
+ lastAddedPartition_ = addedPartitions_.get(addedPartitions_.size() - 1);
msTbl_ = addPartitionMessage_.getTableObj();
} catch (Exception ex) {
throw new MetastoreNotificationException(ex);
@@ -1313,6 +1336,49 @@ public class MetastoreEvents {
}
@Override
+ public void process() throws MetastoreNotificationException, CatalogException {
+ if (isSelfEvent()) {
+ infoLog("Not processing the event as it is a self-event");
+ return;
+ }
+ // Notification is created for newly created partitions only. We need not worry
+ // about "IF NOT EXISTS".
+ try {
+ boolean success = true;
+ // HMS adds partitions in a transactional way. This means there may be multiple
+ // HMS partition objects in an add_partition event. We try to do the same here by
+ // refreshing all those partitions in a loop. If any partition refresh fails, we
+ // throw MetastoreNotificationNeedsInvalidateException exception. We skip
+ // refresh of the partitions if the table is not present in the catalog.
+ infoLog("Trying to refresh {} partitions added to table {} in the event",
+ addedPartitions_.size(), getFullyQualifiedTblName());
+ for (Partition partition : addedPartitions_) {
+ List<TPartitionKeyValue> tPartSpec =
+ getTPartitionSpecFromHmsPartition(msTbl_, partition);
+ if (!catalog_.reloadPartitionIfExists(dbName_, tblName_, tPartSpec)) {
+ debugLog("Refresh partitions on table {} failed "
+ + "as table was not present in the catalog.", getFullyQualifiedTblName());
+ success = false;
+ break;
+ }
+ }
+ if (success) {
+ infoLog("Refreshed {} partitions of table {}", addedPartitions_.size(),
+ getFullyQualifiedTblName());
+ }
+ } catch (DatabaseNotFoundException e) {
+ debugLog("Refresh partitions on table {} after add_partitions event failed as "
+ + "the database was not present in the catalog.",
+ getFullyQualifiedTblName());
+ } catch (CatalogException e) {
+ throw new MetastoreNotificationNeedsInvalidateException(debugString("Failed to "
+ + "refresh newly added partitions of table {}. Event processing cannot "
+ + "continue. Issue an invalidate command to reset event processor.",
+ getFullyQualifiedTblName()));
+ }
+ }
+
+ @Override
protected void initSelfEventIdentifiersFromEvent() {
versionNumberFromEvent_ = Long.parseLong(getStringProperty(
lastAddedPartition_.getParameters(),
@@ -1363,6 +1429,42 @@ public class MetastoreEvents {
}
@Override
+ public void process() throws MetastoreNotificationException, CatalogException {
+ if (isSelfEvent()) {
+ infoLog("Not processing the event as it is a self-event");
+ return;
+ }
+ // Refresh the partition that was altered.
+ Preconditions.checkNotNull(partitionAfter_);
+ List<TPartitionKeyValue> tPartSpec = getTPartitionSpecFromHmsPartition(msTbl_,
+ partitionAfter_);
+ try {
+ // Ignore event if table or database is not in catalog. Throw exception if
+ // refresh fails.
+
+ if (!catalog_.reloadPartitionIfExists(dbName_, tblName_, tPartSpec)) {
+ debugLog("Refresh of table {} partition {} failed as the table "
+ + "is not present in the catalog.", getFullyQualifiedTblName(),
+ constructPartitionStringFromTPartitionSpec(tPartSpec));
+ } else {
+ infoLog("Table {} partition {} has been refreshed", getFullyQualifiedTblName(),
+ constructPartitionStringFromTPartitionSpec(tPartSpec));
+ }
+ } catch (DatabaseNotFoundException e) {
+ debugLog("Refresh of table {} partition {} "
+ + "event failed as the database is not present in the catalog.",
+ getFullyQualifiedTblName(),
+ constructPartitionStringFromTPartitionSpec(tPartSpec));
+ } catch (CatalogException e) {
+ throw new MetastoreNotificationNeedsInvalidateException(debugString("Refresh "
+ + "partition on table {} partition {} failed. Event processing cannot "
+ + "continue. Issue and invalidate command to reset the event processor "
+ + "state.", getFullyQualifiedTblName(),
+ constructPartitionStringFromTPartitionSpec(tPartSpec)));
+ }
+ }
+
+ @Override
protected void initSelfEventIdentifiersFromEvent() {
versionNumberFromEvent_ = Long.parseLong(getStringProperty(
partitionAfter_.getParameters(),
@@ -1383,6 +1485,7 @@ public class MetastoreEvents {
}
public static class DropPartitionEvent extends TableInvalidatingEvent {
+ private final List<Map<String, String>> droppedPartitions_;
/**
* Prevent instantiation from outside should use MetastoreEventFactory instead
@@ -1397,7 +1500,10 @@ public class MetastoreEvents {
.getDeserializer()
.getDropPartitionMessage(event.getMessage());
try {
- msTbl_ = dropPartitionMessage.getTableObj();
+ msTbl_ = Preconditions.checkNotNull(dropPartitionMessage.getTableObj());
+ droppedPartitions_ = dropPartitionMessage.getPartitions();
+ Preconditions.checkNotNull(droppedPartitions_);
+ Preconditions.checkState(droppedPartitions_.size() > 0);
} catch (Exception ex) {
throw new MetastoreNotificationException(
debugString("Could not parse event message. "
@@ -1408,12 +1514,41 @@ public class MetastoreEvents {
}
@Override
- protected boolean isSelfEvent() {
- // TODO currently we don't have a way to determine if drop_partition is a self-event
- // since this NotificationEvent does not contain the partition objects
- // However, detecting this as a self-event will not be needed once we have
- // IMPALA-7973
- return false;
+ public void process() throws MetastoreNotificationException {
+ // We do not need self event as dropPartition() call is a no-op if the directory
+ // doesn't exist.
+ try {
+ boolean success = true;
+ // We refresh all the partitions that were dropped from HMS. If a refresh
+ // fails, we throw a MetastoreNotificationNeedsInvalidateException
+ infoLog("{} partitions dropped from table {}. Trying "
+ + "to refresh.", droppedPartitions_.size(), getFullyQualifiedTblName());
+ for (Map<String, String> partSpec : droppedPartitions_) {
+ List<TPartitionKeyValue> tPartSpec = new ArrayList<>(partSpec.size());
+ for (Map.Entry<String, String> entry : partSpec.entrySet()) {
+ tPartSpec.add(new TPartitionKeyValue(entry.getKey(), entry.getValue()));
+ }
+ if (!catalog_.reloadPartitionIfExists(dbName_, tblName_, tPartSpec)) {
+ debugLog("Could not refresh partition {} of table {} as table "
+ + "was not present in the catalog.",
+ getFullyQualifiedTblName());
+ success = false;
+ break;
+ }
+ }
+ if (success) {
+ infoLog("Refreshed {} partitions of table {}", droppedPartitions_.size(),
+ getFullyQualifiedTblName());
+ }
+ } catch (DatabaseNotFoundException e) {
+ debugLog("Could not refresh partitions of table {}"
+ + "as database was not present in the catalog.", getFullyQualifiedTblName());
+ } catch (CatalogException e) {
+ throw new MetastoreNotificationNeedsInvalidateException(debugString("Failed to "
+ + "drop some partitions from table {} after a drop partitions event. Event "
+ + "processing cannot continue. Issue an invalidate command to reset "
+ + "event processor state.", getFullyQualifiedTblName()));
+ }
}
@Override
diff --git a/fe/src/test/java/org/apache/impala/catalog/events/MetastoreEventsProcessorTest.java b/fe/src/test/java/org/apache/impala/catalog/events/MetastoreEventsProcessorTest.java
index f6ff4ad..bb2caed 100644
--- a/fe/src/test/java/org/apache/impala/catalog/events/MetastoreEventsProcessorTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/events/MetastoreEventsProcessorTest.java
@@ -645,12 +645,6 @@ public class MetastoreEventsProcessorTest {
addPartitions(TEST_DB_NAME, testTblName, partVals);
eventsProcessor_.processEvents();
- // after ADD_PARTITION event is received currently we just invalidate the table
- assertTrue("Table should have been invalidated after add partition event",
- catalog_.getTable(TEST_DB_NAME, testTblName)
- instanceof IncompleteTable);
-
- loadTable(testTblName);
assertEquals("Unexpected number of partitions fetched for the loaded table", 4,
((HdfsTable) catalog_.getTable(TEST_DB_NAME, testTblName))
.getPartitions()
@@ -664,10 +658,6 @@ public class MetastoreEventsProcessorTest {
dropPartitions(testTblName, partVals);
eventsProcessor_.processEvents();
- assertTrue("Table should have been invalidated after drop partition event",
- catalog_.getTable(TEST_DB_NAME, testTblName)
- instanceof IncompleteTable);
- loadTable(testTblName);
assertEquals("Unexpected number of partitions fetched for the loaded table", 1,
((HdfsTable) catalog_.getTable(TEST_DB_NAME, testTblName))
.getPartitions().size());
@@ -675,13 +665,16 @@ public class MetastoreEventsProcessorTest {
// issue alter partition ops
partVals.clear();
partVals.add(Arrays.asList("4"));
- Map<String, String> newParams = new HashMap<>(2);
- newParams.put("alterKey1", "alterVal1");
- alterPartitions(testTblName, partVals, newParams);
+ String newLocation = "/path/to/location/";
+ alterPartitions(testTblName, partVals, newLocation);
eventsProcessor_.processEvents();
- assertTrue("Table should have been invalidated after alter partition event",
- catalog_.getTable(TEST_DB_NAME, testTblName)
- instanceof IncompleteTable);
+
+ Collection<? extends FeFsPartition> parts =
+ FeCatalogUtils.loadAllPartitions((HdfsTable)
+ catalog_.getTable(TEST_DB_NAME, testTblName));
+ FeFsPartition singlePartition =
+ Iterables.getOnlyElement(parts);
+ assertTrue(newLocation.equals(singlePartition.getLocation()));
}
/**
@@ -1749,8 +1742,10 @@ public class MetastoreEventsProcessorTest {
assertEquals(1, eventsProcessor_.getNextMetastoreEvents().size());
eventsProcessor_.processEvents();
if (shouldEventBeProcessed) {
- assertTrue("Table should have been invalidated after add partition event",
- catalog_.getTable(dbName, tblName) instanceof IncompleteTable);
+ Collection<? extends FeFsPartition> partsAfterAdd =
+ FeCatalogUtils.loadAllPartitions((HdfsTable)
+ catalog_.getTable(dbName, tblName));
+ assertTrue("Partitions should have been added.", partsAfterAdd.size() == 6);
} else {
assertFalse("Table should still have been in loaded state since sync is "
+ "disabled",
@@ -1769,8 +1764,10 @@ public class MetastoreEventsProcessorTest {
assertEquals(1, eventsProcessor_.getNextMetastoreEvents().size());
eventsProcessor_.processEvents();
if (shouldEventBeProcessed) {
- assertTrue("Table should have been invalidated after alter partition event",
- catalog_.getTable(dbName, tblName) instanceof IncompleteTable);
+ Collection<? extends FeFsPartition> partsAfterDrop =
+ FeCatalogUtils.loadAllPartitions((HdfsTable) catalog_.getTable(dbName,
+ tblName));
+ assertTrue("Partitions should have been dropped", partsAfterDrop.size() == 2);
} else {
assertFalse("Table should still have been in loaded state since sync is "
+ "disabled",
@@ -1784,16 +1781,21 @@ public class MetastoreEventsProcessorTest {
eventsProcessor_.processEvents();
loadTable(tblName);
List<List<String>> partValues = new ArrayList<>(1);
+ partValues.add(Arrays.asList("3"));
partValues.add(Arrays.asList("2"));
partValues.add(Arrays.asList("1"));
- Map<String, String> newParams = new HashMap<>();
- newParams.put("newParamk1", "newParamv1");
- alterPartitions(tblName, partValues, newParams);
- assertEquals(2, eventsProcessor_.getNextMetastoreEvents().size());
+ String location = "/path/to/partition";
+ alterPartitions(tblName, partValues, location);
+ assertEquals(3, eventsProcessor_.getNextMetastoreEvents().size());
eventsProcessor_.processEvents();
if (shouldEventBeProcessed) {
- assertTrue("Table should have been invalidated after alter partition event",
- catalog_.getTable(dbName, tblName) instanceof IncompleteTable);
+ Collection<? extends FeFsPartition> partsAfterAlter =
+ FeCatalogUtils.loadAllPartitions((HdfsTable)
+ catalog_.getTable(dbName, tblName));
+ for (FeFsPartition part : partsAfterAlter) {
+ assertTrue("Partition location should have been modified by alter.",
+ location.equals(part.getLocation()));
+ }
} else {
assertFalse("Table should still have been in loaded state since sync is "
+ "disabled",
@@ -2070,8 +2072,6 @@ public class MetastoreEventsProcessorTest {
Arrays.asList(partitionKeyValue1, partitionKeyValue2));
eventsProcessor_.processEvents();
assertNotNull(catalog_.getTable(TEST_DB_NAME, testTblName));
- assertTrue(catalog_.getTable(TEST_DB_NAME, testTblName)
- instanceof IncompleteTable);
}
/**
@@ -2641,7 +2641,7 @@ public class MetastoreEventsProcessorTest {
}
private void alterPartitions(String tblName, List<List<String>> partValsList,
- Map<String, String> newParams)
+ String location)
throws TException {
GetPartitionsRequest request = new GetPartitionsRequest();
request.setDbName(TEST_DB_NAME);
@@ -2651,10 +2651,9 @@ public class MetastoreEventsProcessorTest {
Partition partition = metaStoreClient.getHiveClient().getPartition(TEST_DB_NAME,
tblName,
partVal);
- partition.setParameters(newParams);
+ partition.getSd().setLocation(location);
partitions.add(partition);
}
-
metaStoreClient.getHiveClient().alter_partitions(TEST_DB_NAME, tblName, partitions);
}
}