You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by mi...@apache.org on 2024/11/08 17:40:10 UTC
(impala) branch master updated (7369ebb8b -> 4255926b1)
This is an automated email from the ASF dual-hosted git repository.
michaelsmith pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
from 7369ebb8b IMPALA-13415: Add a special testing mode to track Calcite progress
new 453c9519b IMPALA-13482: Bug fixes for lag/coalesce in analytic function.
new 4255926b1 IMPALA-12554: Create one Ranger policy for multi-column GRANT
The 2 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:
.../ranger/RangerCatalogdAuthorizationManager.java | 59 +++++++++++++++++-
.../ranger/RangerImpalaResourceBuilder.java | 2 +
.../calcite/functions/AnalyzedAnalyticExpr.java | 10 ++++
.../functions/AnalyzedFunctionCallExpr.java | 4 +-
...ggFunction.java => ImpalaCoalesceFunction.java} | 40 +++++--------
.../operators/ImpalaCustomOperatorTable.java | 3 +
.../calcite/operators/ImpalaOperatorTable.java | 2 +
tests/authorization/test_ranger.py | 70 +++++++++++++++++++++-
8 files changed, 161 insertions(+), 29 deletions(-)
copy java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/{ImpalaAvgAggFunction.java => ImpalaCoalesceFunction.java} (51%)
(impala) 01/02: IMPALA-13482: Bug fixes for lag/coalesce in analytic function.
Posted by mi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
michaelsmith pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 453c9519bcadb64d8ad5a131c8d031a457dd4d22
Author: Steve Carlin <sc...@cloudera.com>
AuthorDate: Mon Oct 21 12:42:51 2024 -0700
IMPALA-13482: Bug fixes for lag/coalesce in analytic function.
The following SQL query in analytics.test ...
select lag(coalesce(505, 1 + NULL), 1) over (order by int_col desc)
from alltypestiny
... had a couple of issues
1) The coalesce function needed a special operator. This function
derives its return type from a common type that works for all
parameters.
2) The function was not being saved when being reset. This is
needed for when resetAnalysisState is called.
3) createNullLiteral needed to be overriden for similar reasons.
The null literal type needs to be saved for when
resetAnalysisState is called.
Change-Id: Ic54d955a73cec4b5f421099a74df4172a1b7dd8b
Reviewed-on: http://gerrit.cloudera.org:8080/22024
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Michael Smith <mi...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../calcite/functions/AnalyzedAnalyticExpr.java | 10 +++++
.../functions/AnalyzedFunctionCallExpr.java | 4 +-
.../calcite/operators/ImpalaCoalesceFunction.java | 48 ++++++++++++++++++++++
.../operators/ImpalaCustomOperatorTable.java | 3 ++
.../calcite/operators/ImpalaOperatorTable.java | 2 +
5 files changed, 65 insertions(+), 2 deletions(-)
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/AnalyzedAnalyticExpr.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/AnalyzedAnalyticExpr.java
index b9b452eb0..225f32ce7 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/AnalyzedAnalyticExpr.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/AnalyzedAnalyticExpr.java
@@ -72,4 +72,14 @@ public class AnalyzedAnalyticExpr extends AnalyticExpr {
}
}
}
+
+ // Overriding this method because a null literal may be created during the
+ // standardize() call and we want it to use the AnalyzedNullLiteral instead of
+ // base NullLiteral. The reason is that the base NullLiteral's
+ // resetAnalysisState() resets the type to NULL whereas we want any type that
+ // was previously assigned to the null literal to be preserved.
+ @Override
+ protected Expr createNullLiteral() {
+ return new AnalyzedNullLiteral(getFnCall().getParams().exprs().get(0).getType());
+ }
}
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/AnalyzedFunctionCallExpr.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/AnalyzedFunctionCallExpr.java
index 0dd0e0157..83307f35f 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/AnalyzedFunctionCallExpr.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/AnalyzedFunctionCallExpr.java
@@ -47,7 +47,7 @@ public class AnalyzedFunctionCallExpr extends FunctionCallExpr {
// resetAnalyzeState() method can be called at various points which could
// set the fn_ member to null. So we save the function in the savedFunction_
// variable so it can be properly set in analyzeImpl()
- private final Function savedFunction_;
+ private Function savedFunction_;
// c'tor that takes a list of Exprs that eventually get converted to FunctionParams
public AnalyzedFunctionCallExpr(Function fn, List<Expr> params, Type retType) {
@@ -132,11 +132,11 @@ public class AnalyzedFunctionCallExpr extends FunctionCallExpr {
}
fn_ = FunctionResolver.getExactFunction(getFnName().getFunction(),
ImpalaTypeConverter.createRelDataTypes(operandTypes));
+ this.savedFunction_ = fn_;
break;
default:
throw new AnalysisException("Unsupported aggregate function.");
}
}
}
-
}
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaCoalesceFunction.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaCoalesceFunction.java
new file mode 100644
index 000000000..e9bc58eac
--- /dev/null
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaCoalesceFunction.java
@@ -0,0 +1,48 @@
+/*
+ * 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.calcite.operators;
+
+import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.sql.SqlOperatorBinding;
+import org.apache.impala.calcite.type.ImpalaTypeConverter;
+import org.apache.impala.calcite.type.ImpalaTypeSystemImpl;
+
+import java.util.List;
+
+/**
+ * Implementation of coalesce which generates a return type
+ * that is common to all parameters.
+ */
+public class ImpalaCoalesceFunction extends ImpalaOperator {
+
+ public ImpalaCoalesceFunction() {
+ super("COALESCE");
+ }
+
+ @Override
+ public RelDataType inferReturnType(SqlOperatorBinding opBinding) {
+ RexBuilder rexBuilder =
+ new RexBuilder(new JavaTypeFactoryImpl(new ImpalaTypeSystemImpl()));
+ RelDataTypeFactory factory = rexBuilder.getTypeFactory();
+
+ List<RelDataType> operands = CommonOperatorFunctions.getOperandTypes(opBinding);
+ return ImpalaTypeConverter.getCompatibleType(operands, factory);
+ }
+}
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaCustomOperatorTable.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaCustomOperatorTable.java
index fc8741fc5..22006f7fd 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaCustomOperatorTable.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaCustomOperatorTable.java
@@ -192,6 +192,9 @@ public class ImpalaCustomOperatorTable extends ReflectiveSqlOperatorTable {
public static final ImpalaAdjustScaleFunction DTRUNC =
new ImpalaAdjustScaleFunction("DTRUNC");
+ public static final ImpalaCoalesceFunction COALESCE =
+ new ImpalaCoalesceFunction();
+
public static final ImpalaGroupingFunction GROUPING =
new ImpalaGroupingFunction();
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaOperatorTable.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaOperatorTable.java
index f9dbab1ee..f7da2c637 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaOperatorTable.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/operators/ImpalaOperatorTable.java
@@ -74,6 +74,8 @@ public class ImpalaOperatorTable extends ReflectiveSqlOperatorTable {
.add("corr")
.add("covar_pop")
.add("covar_samp")
+ .add("coalesce")
+ .add("lag")
.build();
private static ImpalaOperatorTable INSTANCE;
(impala) 02/02: IMPALA-12554: Create one Ranger policy for multi-column GRANT
Posted by mi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
michaelsmith pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 4255926b126039fad81c3f1107f2b94c3846c9d2
Author: Fang-Yu Rao <fa...@cloudera.com>
AuthorDate: Wed Nov 29 13:38:16 2023 -0800
IMPALA-12554: Create one Ranger policy for multi-column GRANT
This patch makes Impala create only one Ranger policy for the GRANT
statement when there are multiple columns specified to reduce the number
of policies created on the Ranger server.
Note that this patch relies on RANGER-4585 and RANGER-4638.
Testing:
- Manually verified that Impala's catalog daemon only sends one
GrantRevokeRequest to the Ranger plug-in and that the value of the
key 'column' is a comma-separated list of column names involved in
the GRANT statement.
- Added an end-to-end test to verify only one Ranger policy will be
created in a multi-column GRANT statement.
Change-Id: I2b0ebba256c7135b4b0d2160856202292d720c6d
Reviewed-on: http://gerrit.cloudera.org:8080/21940
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../ranger/RangerCatalogdAuthorizationManager.java | 59 +++++++++++++++++-
.../ranger/RangerImpalaResourceBuilder.java | 2 +
tests/authorization/test_ranger.py | 70 +++++++++++++++++++++-
3 files changed, 128 insertions(+), 3 deletions(-)
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 809f2928b..3c9a02d59 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
@@ -28,6 +28,7 @@ import org.apache.impala.catalog.AuthzCacheInvalidation;
import org.apache.impala.catalog.CatalogServiceCatalog;
import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.InternalException;
+import org.apache.impala.common.Pair;
import org.apache.impala.common.UnsupportedFeatureException;
import org.apache.impala.thrift.TCatalogServiceRequestHeader;
import org.apache.impala.thrift.TCreateDropRoleParams;
@@ -49,7 +50,9 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -308,6 +311,7 @@ public class RangerCatalogdAuthorizationManager implements AuthorizationManager
@VisibleForTesting
public void grantPrivilege(List<GrantRevokeRequest> requests, String sqlStmt,
String clientIp) throws ImpalaException {
+ long startTime = System.currentTimeMillis();
try {
for (GrantRevokeRequest request : requests) {
try (AutoFlush auditHandler = RangerBufferAuditHandler.autoFlush(sqlStmt,
@@ -319,12 +323,16 @@ public class RangerCatalogdAuthorizationManager implements AuthorizationManager
LOG.error("Error granting a privilege in Ranger: ", e);
throw new InternalException("Error granting a privilege in Ranger. " +
"Ranger error message: " + e.getMessage());
+ } finally {
+ LOG.debug("Handling granting privilege(s) took {} ms",
+ (System.currentTimeMillis() - startTime));
}
}
@VisibleForTesting
public void revokePrivilege(List<GrantRevokeRequest> requests, String sqlStmt,
String clientIp) throws ImpalaException {
+ long startTime = System.currentTimeMillis();
try {
for (GrantRevokeRequest request : requests) {
try (AutoFlush auditHandler = RangerBufferAuditHandler.autoFlush(sqlStmt,
@@ -336,6 +344,9 @@ public class RangerCatalogdAuthorizationManager implements AuthorizationManager
LOG.error("Error revoking a privilege in Ranger: ", e);
throw new InternalException("Error revoking a privilege in Ranger. " +
"Ranger error message: " + e.getMessage());
+ } finally {
+ LOG.debug("Handling revoking privilege(s) took {} ms",
+ (System.currentTimeMillis() - startTime));
}
}
@@ -419,7 +430,7 @@ public class RangerCatalogdAuthorizationManager implements AuthorizationManager
}
}
- return requests;
+ return consolidateGrantRevokeRequests(requests);
}
private static GrantRevokeRequest createGrantRevokeRequest(String grantor, String user,
@@ -480,6 +491,52 @@ public class RangerCatalogdAuthorizationManager implements AuthorizationManager
return request;
}
+ /**
+ * This method combines GrantRevokeRequest's for specified columns in the same table
+ * into one GrantRevokeRequest to reduce the number of Ranger policies created on the
+ * Ranger server. This method implicitly assumes that for those GrantRevokeRequest's
+ * with specified columns on the given List, the fields other than database, table, and
+ * column names are the same.
+ */
+ private static List<GrantRevokeRequest> consolidateGrantRevokeRequests(
+ List<GrantRevokeRequest> requests) {
+ List<GrantRevokeRequest> combinedRequests = new LinkedList<>();
+ Map<Pair<String, String>, GrantRevokeRequest> consolidatedColumnRequests =
+ new HashMap<>();
+ List<GrantRevokeRequest> unconsolidatedRequests = new LinkedList<>();
+
+ for (GrantRevokeRequest request : requests) {
+ Map<String, String> resource = request.getResource();
+ String column = resource.get(RangerImpalaResourceBuilder.COLUMN);
+ if (column != null &&
+ !column.equals(RangerImpalaResourceBuilder.WILDCARD_ASTERISK)) {
+ // If 'column' is the name of a specified column, the database name and the table
+ // name must have been specified too.
+ String database = resource.get(RangerImpalaResourceBuilder.DATABASE);
+ String table = resource.get(RangerImpalaResourceBuilder.TABLE);
+ Pair<String, String> key = new Pair<>(database, table);
+ if (!consolidatedColumnRequests.containsKey(key)) {
+ consolidatedColumnRequests.put(key, request);
+ } else {
+ // Combine 'column' into the GrantRevokeRequest for other specified columns in
+ // the same table.
+ Map<String, String> consolidatedResource =
+ consolidatedColumnRequests.get(key).getResource();
+ String consolidatedColumns = consolidatedResource
+ .get(RangerImpalaResourceBuilder.COLUMN);
+ consolidatedResource.put(RangerImpalaResourceBuilder.COLUMN,
+ consolidatedColumns + RangerImpalaResourceBuilder.COLUMN_SEP + column);
+ }
+ } else {
+ unconsolidatedRequests.add(request);
+ }
+ }
+
+ combinedRequests.addAll(consolidatedColumnRequests.values());
+ combinedRequests.addAll(unconsolidatedRequests);
+ return combinedRequests;
+ }
+
/**
* The caller of this method calls Ranger's REST API to grant/revoke roles
* corresponding to 'targetRoleNames' to/from groups associated with 'groupNames'.
diff --git a/fe/src/main/java/org/apache/impala/authorization/ranger/RangerImpalaResourceBuilder.java b/fe/src/main/java/org/apache/impala/authorization/ranger/RangerImpalaResourceBuilder.java
index e743c0c5f..ce98b65e9 100644
--- a/fe/src/main/java/org/apache/impala/authorization/ranger/RangerImpalaResourceBuilder.java
+++ b/fe/src/main/java/org/apache/impala/authorization/ranger/RangerImpalaResourceBuilder.java
@@ -27,10 +27,12 @@ public class RangerImpalaResourceBuilder {
public static final String DATABASE = "database";
public static final String TABLE = "table";
public static final String COLUMN = "column";
+ public static final String COLUMN_SEP = ",";
public static final String UDF = "udf";
public static final String URL = "url";
public static final String STORAGE_TYPE = "storage-type";
public static final String STORAGE_URL = "storage-url";
+ public static final String WILDCARD_ASTERISK = "*";
private final RangerAccessResourceImpl rangerAccessResource =
new RangerAccessResourceImpl();
diff --git a/tests/authorization/test_ranger.py b/tests/authorization/test_ranger.py
index c2f06cb44..51f78fa64 100644
--- a/tests/authorization/test_ranger.py
+++ b/tests/authorization/test_ranger.py
@@ -1051,7 +1051,7 @@ class TestRanger(CustomClusterTestSuite):
@staticmethod
def _get_ranger_privileges_db(user, db):
- policies = TestRanger._get_ranger_privileges(user)
+ policies = TestRanger._get_ranger_privileges()
result = []
for policy in policies:
@@ -1065,7 +1065,7 @@ class TestRanger(CustomClusterTestSuite):
return result
@staticmethod
- def _get_ranger_privileges(user):
+ def _get_ranger_privileges():
r = requests.get("{0}/service/plugins/policies"
.format(RANGER_HOST),
auth=RANGER_AUTH, headers=REST_HEADERS)
@@ -1271,6 +1271,72 @@ class TestRanger(CustomClusterTestSuite):
impala_client, query, user=username, query_options={'sync_ddl': 1})
return self.execute_query_expect_failure(impala_client, query, user=username)
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ impalad_args=IMPALAD_ARGS, catalogd_args=CATALOGD_ARGS, reset_ranger=True)
+ def test_grant_multiple_columns(self):
+ admin_client = self.create_impala_client()
+ access_type = "select"
+ db = "functional"
+ tbl = "alltypes"
+ cols = ["id", "bool_col", "tinyint_col", "smallint_col", "int_col", "bigint_col",
+ "float_col", "double_col", "date_string_col", "string_col", "timestamp_col",
+ "year", "month"]
+ cols_str = ""
+ for col in cols:
+ if not cols_str:
+ cols_str = col
+ else:
+ cols_str = cols_str + ", " + col
+ test_data = [("user", "non_owner", "users"), ("group", "non_owner", "groups"),
+ ("role", "test_role", "roles")]
+
+ for data in test_data:
+ kw = data[0]
+ principal_name = data[1]
+ principal_key = data[2]
+ try:
+ policy_ids = set()
+ if kw == "role":
+ admin_client.execute("create role {0}".format(principal_name), user=ADMIN)
+ admin_client.execute("grant {0}({1}) on table {2}.{3} to {4} {5}"
+ .format(access_type, cols_str, db, tbl, kw, principal_name), user=ADMIN)
+ policies = TestRanger._get_ranger_privileges()
+ for col in cols:
+ policy_ids = policy_ids \
+ .union(TestRanger._get_ranger_policy_ids(policies, principal_name,
+ principal_key, db, tbl, col, access_type))
+ # After the GRANT statement above, there should be only one single Ranger policy
+ # that grants the privilege of 'access_type' on the column 'db'.'tbl'.'col' to
+ # the principal 'principal_name' for each column in 'cols'.
+ assert len(policy_ids) == 1
+ finally:
+ admin_client.execute("revoke {0}({1}) on table {2}.{3} from {4} {5}"
+ .format(access_type, cols_str, db, tbl, kw, principal_name), user=ADMIN)
+ if kw == "role":
+ admin_client.execute("drop role {0}".format(principal_name), user=ADMIN)
+
+ @staticmethod
+ def _get_ranger_policy_ids(policies, principal_name, principal_key, db, tbl, col,
+ access_type):
+ """Returns the set of Ranger policy id's that grant the privilege of 'access_type' on
+ the column 'db'.'tbl'.'col'. to the principal 'principal_name'."""
+ result = set()
+
+ for policy in policies:
+ id = policy["id"]
+ resources = policy["resources"]
+ if "database" in resources and db in resources["database"]["values"] \
+ and "table" in resources and tbl in resources["table"]["values"] \
+ and "column" in resources and col in resources["column"]["values"]:
+ for policy_items in policy["policyItems"]:
+ if principal_name in policy_items[principal_key]:
+ for access in policy_items["accesses"]:
+ if access_type in access["type"] and access["isAllowed"] is True:
+ result.add(id)
+ break
+ return result
+
@pytest.mark.execute_serially
@CustomClusterTestSuite.with_args(
impalad_args=IMPALAD_ARGS, catalogd_args=CATALOGD_ARGS)