xiangfu0 commented on code in PR #19664:
URL: https://github.com/apache/pinot/pull/19664#discussion_r4224763934
##########
pinot-common/src/main/java/org/apache/pinot/sql/parsers/CalciteSqlParser.java:
##########
@@ -145,13 +141,21 @@ public static SqlNodeAndOptions
compileToSqlNodeAndOptions(String sql)
SqlNodeAndOptions sqlNodeAndOptions =
extractSqlNodeAndOptions(sqlNodeList);
// add legacy OPTIONS keyword-based options
if (!options.isEmpty()) {
- Map<String, String> optionMap = extractOptionsMap(options);
- if (sqlNodeAndOptions.getSqlType() == PinotSqlType.DQL) {
- // No-op unless the broker enables query option validation. DML (e.g.
- // INSERT INTO FILE OPTION(taskName=...)) carries free-form task/FS
properties, like DML SET.
- QueryOptionsUtils.validateSqlQueryOptions(optionMap);
+ if (legacyOptionSyntaxMode == SqlOptionsMode.IGNORE) {
+ // Dropping a DELETE option (e.g. dryRun) would change its effect.
+ if (sqlNodeAndOptions.getSqlNode() instanceof SqlDelete) {
Review Comment:
This only fails for `instanceof SqlDelete`, so in IGNORE mode an `INSERT
INTO ... FROM FILE` still loses its `OPTION(...)` task configuration
(`taskName`, `taskType`, `input.fs.className`, `input.fs.prop.*`) without an
error. The broker already behaved that way, but registering
`QueryOptionConfigListener` in `BaseControllerStarter` (added for the last
round's comment about legacy syntax on the controller) now applies it to the
controller too, which used to stay in ALLOW. For example, with
`pinot.broker.query.option.legacySyntaxMode=IGNORE`, a controller `POST /sql`
of `INSERT INTO myTable FROM FILE 'file:///data/in/' OPTION(taskName=load-1)`
used to submit task `load-1`; it now submits a default
`SegmentGenerationAndPushTask` named `myTable_<uuid>` and reports success. A
retry no longer fails with `TaskAlreadyExistsException` and ingests the files
again, and a custom `taskType` silently becomes the default.
An INSERT's options configure it just as a DELETE's do, so I'd key this
check, and the one in `RequestUtils.setOptions`, on `PinotSqlType.DML`, and
update the `SqlOptionsModeTest` assertions that pin the silent drop (lines 72
and 115-116). If INSERT has to keep dropping options on the broker, at least
don't apply the legacy syntax mode to the statements the controller executes.
##########
pinot-common/src/main/java/org/apache/pinot/sql/parsers/dml/DeleteStatement.java:
##########
@@ -0,0 +1,351 @@
+/**
+ * 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.pinot.sql.parsers.dml;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.calcite.avatica.util.Casing;
+import org.apache.calcite.sql.SqlDelete;
+import org.apache.calcite.sql.SqlDialect;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlSyntax;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.util.SqlShuttle;
+import org.apache.calcite.sql.validate.SqlValidatorUtil;
+import org.apache.pinot.common.config.provider.TableCache;
+import org.apache.pinot.common.request.Expression;
+import org.apache.pinot.common.request.ExpressionType;
+import org.apache.pinot.common.request.Function;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.common.utils.DatabaseUtils;
+import org.apache.pinot.spi.config.task.AdhocTaskConfig;
+import org.apache.pinot.spi.exception.QueryErrorCode;
+import org.apache.pinot.spi.exception.QueryException;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.sql.parsers.CalciteSqlParser;
+import org.apache.pinot.sql.parsers.SqlNodeAndOptions;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+
+/// A SQL `DELETE FROM <table> WHERE <predicate>` statement.
+///
+/// Pinot parses `DELETE` but does not delete rows itself: the default SQL
executor answers that it is not supported,
+/// and a deployment that can delete rows, e.g. by purging the matching rows
from the segments with a minion task,
+/// executes the parsed statement by overriding
`SqlQueryExecutor#executeDelete`. Its execution type is
+/// [ExecutionType#EXECUTOR]: the generic [#execute()], [#getResultSchema()]
and [#generateAdhocTaskConfig()] do not
+/// apply to it.
+///
+/// Before handing the statement to the executor, the broker and the
controller resolve its table with
+/// [#resolveTableName], authorize the caller to delete rows from that table,
and check that the table exists with
+/// [#tableExists()].
+///
+/// Its options are the `SET` statements, the legacy `OPTION(...)` suffix and
the request `queryOptions` of the
+/// statement (`SET` takes precedence). The `database` option only qualifies
the table name, see [#resolveTableName].
+/// Every other option, including query options such as `timeoutMs`, is passed
to the executor through
+/// [#getOptions()], so that an option the executor relies on (e.g. a dry run)
cannot be reclassified as a query option
+/// and silently dropped.
+///
+/// Instances are immutable and thread-safe.
+public class DeleteStatement implements DataManipulationStatement {
+ public static final String NOT_SUPPORTED_MESSAGE =
+ "DELETE is not supported by this endpoint, it requires a SQL executor
that implements row deletion";
+
+ /// Serializes the WHERE clause into SQL that the Pinot parser reads back
into the same expression. Combined with
+ /// `quoteAllIdentifiers = false`, identifiers are only quoted (with double
quotes) when they were quoted.
+ private static final SqlDialect PINOT_SQL_DIALECT = new
SqlDialect(SqlDialect.EMPTY_CONTEXT
+ .withIdentifierQuoteString("\"")
+ .withLiteralQuoteString("'")
+ .withLiteralEscapedQuoteString("''")
+ .withUnquotedCasing(Casing.UNCHANGED)
+ .withQuotedCasing(Casing.UNCHANGED)
+ .withCaseSensitive(true));
+
+ /// Quotes the unquoted identifiers named like a SQL function without
arguments, e.g. `user`, `pi` or `current_date`:
+ /// Calcite unparses them as upper-cased keywords, while Pinot reads them as
columns, so they would name another
+ /// column. Mirrors the check of `SqlUtil.unparseSqlIdentifierSyntax`.
+ private static final SqlShuttle KEYWORD_IDENTIFIER_QUOTER = new SqlShuttle()
{
+ @Override
+ public SqlNode visit(SqlIdentifier identifier) {
+ if (identifier.isSimple() && !identifier.getParserPosition().isQuoted())
{
+ SqlOperator operator =
+
SqlValidatorUtil.lookupSqlFunctionByID(SqlStdOperatorTable.instance(),
identifier, null);
+ if (operator != null && (operator.getSyntax() == SqlSyntax.FUNCTION_ID
+ || operator.getSyntax() == SqlSyntax.FUNCTION_ID_CONSTANT)) {
+ return new SqlIdentifier(identifier.names, null,
SqlParserPos.QUOTED_ZERO,
+ List.of(SqlParserPos.QUOTED_ZERO));
+ }
+ }
+ return identifier;
+ }
+ };
+
+ /// Canonical names (lower case, without underscores) of the functions that
read another table than the one the
+ /// statement deletes from, which the caller is not authorized to read.
+ private static final Set<String> CROSS_TABLE_FUNCTIONS = Set.of("lookup",
"insubquery", "inpartitionedsubquery");
+
+ /// Canonical names of the functions that run a script. Rejected because the
broker's groovy policy
+ /// (`pinot.broker.disable.query.groovy`, the per-table
`queryConfig.disableGroovy` override and the static
+ /// analyzer) only applies to queries, not to a `DELETE`, and the executor
may run the predicate on another role.
+ private static final Set<String> SCRIPT_FUNCTIONS = Set.of("groovy");
+
+ private final String _tableName;
+ private final String _predicate;
+ @Nullable
+ private final String _database;
+ private final Map<String, String> _options;
+ private final boolean _resolved;
+ private final boolean _tableExists;
+
+ private DeleteStatement(String tableName, String predicate, @Nullable String
database, Map<String, String> options) {
+ this(tableName, predicate, database, options, false, false);
+ }
+
+ private DeleteStatement(String tableName, String predicate, @Nullable String
database, Map<String, String> options,
+ boolean resolved, boolean tableExists) {
+ _tableName = tableName;
+ _predicate = predicate;
+ _database = database;
+ _options = Collections.unmodifiableMap(new HashMap<>(options));
+ _resolved = resolved;
+ _tableExists = tableExists;
+ }
+
+ /// Parses a `DELETE` statement.
+ ///
+ /// @throws IllegalArgumentException if the statement is not a supported
`DELETE`: it must have a WHERE clause, no
+ /// table alias, a WHERE clause that Pinot
can parse as an expression, that does
+ /// not read another table (e.g. with
`lookUp` or `IN_SUBQUERY`) and that does not
+ /// run a script (`groovy`), and set the
database with the `database` option only
+ public static DeleteStatement parse(SqlNodeAndOptions sqlNodeAndOptions) {
+ SqlNode sqlNode = sqlNodeAndOptions.getSqlNode();
+ checkArgument(sqlNode instanceof SqlDelete, "Not a DELETE statement: %s",
sqlNode.getKind());
+ SqlDelete sqlDelete = (SqlDelete) sqlNode;
+ String tableName = getTableName(sqlDelete.getTargetTable());
+ checkArgument(sqlDelete.getAlias() == null,
+ "DELETE does not support a table alias, reference the columns
directly");
+ SqlNode condition = sqlDelete.getCondition();
+ checkArgument(condition != null, "DELETE requires a WHERE clause to
prevent an accidental full-table delete; use "
+ + "an explicit always-true predicate or delete the table segments to
remove all rows");
+ String predicate = toPinotSql(condition);
+
+ String database = null;
+ Map<String, String> options = new HashMap<>();
+ for (Map.Entry<String, String> option :
sqlNodeAndOptions.getOptions().entrySet()) {
+ String key = option.getKey();
+ if (key.equals(CommonConstants.DATABASE)) {
+ database = option.getValue();
+ } else if (key.equalsIgnoreCase(CommonConstants.DATABASE)) {
+ // Queries only read the exact `database` option (where they read it
at all, see resolveTableName): fail
+ // rather than delete from another table
+ throw new IllegalArgumentException(
+ "Unsupported option: " + key + ", set the database with the '" +
CommonConstants.DATABASE + "' option");
+ } else {
+ options.put(key, option.getValue());
+ }
+ }
+ return new DeleteStatement(tableName, predicate, database, options);
+ }
+
+ private static String getTableName(SqlNode targetTable) {
+ // Table hints and EXTEND clauses parse into other node types
+ checkArgument(targetTable instanceof SqlIdentifier, "DELETE only supports
a plain table name, got: %s",
+ targetTable);
+ // A quoted name part may contain a dot, which splits it like the table
name of a query
+ String tableName = String.join(".", ((SqlIdentifier) targetTable).names);
+ // Empty parts, e.g. in "db."."t", are rejected rather than dropped, so
that the table name is the one authorized
+ String[] parts = tableName.split("\\.", -1);
+ checkArgument(parts.length <= 2 &&
Arrays.stream(parts).noneMatch(String::isEmpty),
+ "Invalid table name: %s, expected [database.]table", tableName);
+ return tableName;
+ }
+
+ /// Serializes a WHERE clause back into SQL, and verifies that Pinot parses
it into the same expression.
+ private static String toPinotSql(SqlNode condition) {
+ Expression expression;
+ Expression serializedExpression;
+ String predicate;
+ try {
+ expression = CalciteSqlParser.compileToExpression(condition);
+ predicate = condition.accept(KEYWORD_IDENTIFIER_QUOTER)
+ .toSqlString(config -> config.withDialect(PINOT_SQL_DIALECT)
+ .withQuoteAllIdentifiers(false)
+ .withIndentation(0))
+ .getSql();
+ serializedExpression = CalciteSqlParser.compileToExpression(predicate);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Unsupported WHERE clause in DELETE:
" + describe(condition), e);
+ }
+ // Fail rather than hand over a predicate that selects other rows than the
statement
+ checkArgument(serializedExpression.equals(expression),
Review Comment:
This round-trip check compares single-stage `Expression`s, which drop
information that the predicate string keeps, so a predicate whose `Expression`
selects other rows than the SQL still passes:
- An interval literal compiles to its bare value: `INTERVAL '30' DAY`,
`INTERVAL '30' MINUTE` and `INTERVAL -'30' DAY` all become the string literal
`'30'` (`RequestUtils.getLiteral` falls through to `node.toValue()`).
- `compileFunctionExpression` reads the negation of `BETWEEN` but not
`SYMMETRIC`, so `x NOT BETWEEN SYMMETRIC 10 AND 1` becomes `NOT(BETWEEN(x, 10,
1))`.
- Exact integer literals outside the long range wrap (the TODO in
`RequestUtils.getLiteral`): `9223372036854775808` becomes `Long.MIN_VALUE` and
`18446744073709551616` becomes `0`.
So an executor that evaluates `compileToExpression(getPredicate())`, which
is the parse the `getPredicate` Javadoc promises, deletes other rows than the
statement says. Replaying `toPinotSql` against the parser on master: `DELETE
FROM events WHERE ts < CAST(now() AS TIMESTAMP) - INTERVAL '30' DAY` passes the
check, and its filter compiles to `ts < now - 30` (milliseconds), which is
nearly every row. `x NOT BETWEEN SYMMETRIC 10 AND 1` matches every non-null row
on the single-stage engine (the inverted range is always false), while the SQL,
and the multi-stage engine, which expands `SYMMETRIC`, only match `x < 1 OR x >
10`. And `x > 9223372036854775808` matches no row in SQL but nearly every row
as an `Expression`.
A single-stage SELECT has the same gaps, but a DELETE can't be undone, so
I'd fail closed here as for `groovy`: reject `SqlIntervalLiteral`s, `BETWEEN
SYMMETRIC`, and exact integer literals that don't fit in a long, with a
`DeleteStatementTest` case for each. `TIMESTAMPADD(DAY, ...)` keeps its unit
and can stay.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/api/resources/PinotClientRequest.java:
##########
@@ -713,6 +739,137 @@ private BrokerResponse executeSqlQuery(ObjectNode
sqlRequestJson, HttpRequesterI
}
}
+ /// Executes a `DELETE` once the caller is authorized to delete rows from
its table.
+ ///
+ /// The first-step access control runs first, as for queries, followed by
the table check without a table, so that
+ /// an access control that only authenticates the caller in that check
rejects an unauthenticated caller before the
+ /// table is looked up. The table is then resolved with the database of the
+ /// request and in the case it is defined with, the caller is authorized to
delete rows from it (see
+ /// [#authorizeDelete]), and the executor deletes rows from that exact
table. A table the table cache does not know
+ /// fails with [QueryErrorCode#TABLE_DOES_NOT_EXIST], as a query does, but
only once the caller is authorized for
+ /// it, so that the existence of a table is not leaked to a caller who is
not authorized for it.
+ private BrokerResponse executeDelete(SqlNodeAndOptions sqlNodeAndOptions,
Map<String, String> headers,
+ HttpRequesterIdentity requesterIdentity, @Nullable HttpHeaders
httpHeaders) {
+ AccessControl accessControl = _accessControlFactory.create();
+ // The first-step access control runs before the table is looked up, as
for queries
+ AuthorizationResult authorizationResult =
accessControl.authorize(requesterIdentity);
+ if (authorizationResult.hasAccess()) {
+ // Access controls that authenticate the caller in their table check
rather than in the first step (e.g. the
+ // ZooKeeper basic auth) reject an unauthenticated caller here, before
the table is looked up
+ authorizationResult = accessControl.authorize(requesterIdentity,
Set.of());
Review Comment:
Two issues with this probe, both from how the `AccessControl` defaults
behave:
1. It and the table check in `authorizeDelete` go through the default
`authorize(RequesterIdentity, Set)`, which calls `hasAccess(RequesterIdentity,
Set)`, and that throws `UnsupportedOperationException` for an access control
that only overrides the deprecated `hasAccess(RequesterIdentity,
BrokerRequest)`, which still works for single-stage queries (see
`AccessControlBackwardCompatibleTest`). With one configured, a DELETE skips the
`WebApplicationException` catch, is logged at ERROR with the full request JSON,
is counted in `UNCAUGHT_POST_EXCEPTIONS` instead of
`REQUEST_DROPPED_DUE_TO_ACCESS_ERROR`, and gets an HTTP 500 "Both hasAccess()
and authorize() are not implemented" instead of a 403. Nothing is deleted, but
a denial is reported as a server error. I mentioned this in the last summary
and it's still open.
2. The probe can only deny by throwing.
`TableAuthorizationResult.hasAccess()` is `_failedTables.isEmpty()`, and the
SPI default turns a `false` from the boolean `hasAccess(identity, Set)` into
`new TableAuthorizationResult(tables)`, which grants access for `Set.of()` (the
`deleteChecks` data provider notes that "the probe without a table passes").
With a plugin that leaves the first step at its default `true` and reports
missing credentials as a denial, an unauthenticated `DELETE FROM salesLogical
WHERE x = 1` gets QUERY_VALIDATION "DELETE does not support logical tables"
while other names get a 403. None of the shipped factories behave like that, so
this one is low severity.
Treating an `UnsupportedOperationException` from the table-set checks as a
denial routed through `deleteAccessDenied`, and treating a logical table like
an unknown one (record it in `resolveTableName`, run `authorizeDelete`, then
return QUERY_VALIDATION, as `tableExists()` already does) would fix both
without depending on how the access control reports missing credentials. A
`PinotClientRequestTest` case with an access control that only overrides
`hasAccess(RequesterIdentity, BrokerRequest)` would cover the first.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/executor/sql/SqlQueryExecutor.java:
##########
@@ -76,16 +86,63 @@ private static String getControllerBaseUrl(HelixManager
helixManager) {
return controllerBaseUrl;
}
- /// Execute DML Statement
+ /// Parses and executes a DML statement.
+ ///
+ /// A `DELETE` is refused by [#executeStatement] with a
[QueryErrorCode#ACCESS_DENIED] error, since this method
+ /// cannot authorize the caller: it is executed once its table is resolved
and the caller authorized to delete rows
+ /// from it, as the query endpoints of the broker and the controller do.
///
/// @param sqlNodeAndOptions Parsed DML object
/// @param headers extra headers map for minion task submission
/// @return BrokerResponse is the DML executed response
public BrokerResponse executeDMLStatement(SqlNodeAndOptions
sqlNodeAndOptions,
@Nullable Map<String, String> headers) {
- DataManipulationStatement statement =
DataManipulationStatementParser.parse(sqlNodeAndOptions);
+ DataManipulationStatement statement;
+ try {
+ statement = DataManipulationStatementParser.parse(sqlNodeAndOptions);
+ } catch (QueryException e) {
+ // e.g. a DELETE without a WHERE clause
+ return new BrokerResponseNative(e.getErrorCode(), e.getMessage());
+ }
+ return executeStatement(statement, headers);
+ }
+
+ /// Executes a parsed DML statement, e.g. from
[DataManipulationStatementParser#parse], as its
+ /// [DataManipulationStatement#getExecutionType()] tells: in-process, as a
minion task, or through the hook of this
+ /// executor for the statement
([DataManipulationStatement.ExecutionType#EXECUTOR]), e.g. [#executeDelete] for
a
+ /// [DeleteStatement].
+ ///
+ /// It does not authorize the caller. The table of a [DeleteStatement] must
be resolved with
+ /// [DeleteStatement#resolveTableName] and the caller authorized to delete
rows from it before it is executed, as the
+ /// query endpoints of the broker and the controller do: an unresolved
`DELETE` is refused with a
+ /// [QueryErrorCode#ACCESS_DENIED] error.
+ ///
+ /// Execution failures are returned in the response, never thrown: a
[QueryException] thrown by the executing hook
+ /// is returned with its error code and message, any other exception as a
[QueryErrorCode#QUERY_EXECUTION] error.
+ ///
+ /// @param statement parsed statement
+ /// @param headers headers of the original request, e.g. for minion task
submission
+ /// @return the response of the statement
+ public BrokerResponse executeStatement(DataManipulationStatement statement,
@Nullable Map<String, String> headers) {
BrokerResponseNative result = new BrokerResponseNative();
switch (statement.getExecutionType()) {
+ case EXECUTOR:
+ if (statement instanceof DeleteStatement) {
+ DeleteStatement deleteStatement = (DeleteStatement) statement;
+ if (!deleteStatement.isResolved()) {
+ return new BrokerResponseNative(QueryErrorCode.ACCESS_DENIED,
UNAUTHORIZED_DELETE_MESSAGE);
+ }
+ try {
+ return executeDelete(deleteStatement, headers);
+ } catch (QueryException e) {
+ return new BrokerResponseNative(e.getErrorCode(), e.getMessage());
+ } catch (Exception e) {
+ return new BrokerResponseNative(QueryErrorCode.QUERY_EXECUTION,
e.getMessage());
Review Comment:
This catch turns any non-`QueryException` from an `executeDelete` override
into a `QUERY_EXECUTION` response carrying only `e.getMessage()`, and
`SqlQueryExecutor` has no logger, so the exception type, cause and stack trace
aren't recorded anywhere. Before this wrapping, the broker logged them in the
DML branch of `PinotClientRequest` (`LOGGER.error("Error handling DML
request...", sqlRequestJson, e)`). Take an override that submits purge tasks
for some segments and then fails `Preconditions.checkState(taskId != null)`:
the client gets `QueryExecutionError: null` (the
`BrokerResponseNative(QueryErrorCode, String)` constructor concatenates the
null, unlike the `QueryProcessingException` constructor the MINION and HTTP
branches use), the broker only logs `{200=QueryExecutionError: null}`, and the
controller logs nothing after its "Executing DELETE" line. That leaves no way
to diagnose a DELETE that may have partly run.
I'd add a logger and log the throwable with the table before returning, and
fall back to `e.toString()` when the message is null, with a
`SqlQueryExecutorTest` case for an exception without a message.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/executor/sql/SqlQueryExecutor.java:
##########
@@ -76,16 +86,63 @@ private static String getControllerBaseUrl(HelixManager
helixManager) {
return controllerBaseUrl;
}
- /// Execute DML Statement
+ /// Parses and executes a DML statement.
+ ///
+ /// A `DELETE` is refused by [#executeStatement] with a
[QueryErrorCode#ACCESS_DENIED] error, since this method
+ /// cannot authorize the caller: it is executed once its table is resolved
and the caller authorized to delete rows
+ /// from it, as the query endpoints of the broker and the controller do.
///
/// @param sqlNodeAndOptions Parsed DML object
/// @param headers extra headers map for minion task submission
/// @return BrokerResponse is the DML executed response
public BrokerResponse executeDMLStatement(SqlNodeAndOptions
sqlNodeAndOptions,
@Nullable Map<String, String> headers) {
- DataManipulationStatement statement =
DataManipulationStatementParser.parse(sqlNodeAndOptions);
+ DataManipulationStatement statement;
+ try {
+ statement = DataManipulationStatementParser.parse(sqlNodeAndOptions);
+ } catch (QueryException e) {
+ // e.g. a DELETE without a WHERE clause
+ return new BrokerResponseNative(e.getErrorCode(), e.getMessage());
+ }
+ return executeStatement(statement, headers);
+ }
+
+ /// Executes a parsed DML statement, e.g. from
[DataManipulationStatementParser#parse], as its
+ /// [DataManipulationStatement#getExecutionType()] tells: in-process, as a
minion task, or through the hook of this
+ /// executor for the statement
([DataManipulationStatement.ExecutionType#EXECUTOR]), e.g. [#executeDelete] for
a
+ /// [DeleteStatement].
+ ///
+ /// It does not authorize the caller. The table of a [DeleteStatement] must
be resolved with
+ /// [DeleteStatement#resolveTableName] and the caller authorized to delete
rows from it before it is executed, as the
+ /// query endpoints of the broker and the controller do: an unresolved
`DELETE` is refused with a
+ /// [QueryErrorCode#ACCESS_DENIED] error.
+ ///
+ /// Execution failures are returned in the response, never thrown: a
[QueryException] thrown by the executing hook
+ /// is returned with its error code and message, any other exception as a
[QueryErrorCode#QUERY_EXECUTION] error.
+ ///
+ /// @param statement parsed statement
+ /// @param headers headers of the original request, e.g. for minion task
submission
+ /// @return the response of the statement
+ public BrokerResponse executeStatement(DataManipulationStatement statement,
@Nullable Map<String, String> headers) {
BrokerResponseNative result = new BrokerResponseNative();
switch (statement.getExecutionType()) {
+ case EXECUTOR:
+ if (statement instanceof DeleteStatement) {
+ DeleteStatement deleteStatement = (DeleteStatement) statement;
+ if (!deleteStatement.isResolved()) {
Review Comment:
This guard only proves that the statement went through the public
`resolveTableName`, which accepts any `TableCache` (`SqlQueryExecutorTest`
resolves with a mock), not that the caller was authorized: the authorization
chain (first-step authorize, QUERY and DELETE_ROWS on the typed and raw names,
`authorizeDeleteRows`, RLS, `tableExists`) lives in private methods of
`PinotClientRequest` and `PinotQueryResource`. Nothing in-tree is exposed,
since both endpoints authorize before calling `executeStatement`, gRPC rejects
DELETE, and the default `executeDelete` deletes nothing. But a deployment that
overrides `executeDelete` and adds its own entry point (e.g. a resource in
`broker.restlet.api.resource.packages` that injects `SqlQueryExecutor`) can
resolve and run a DELETE with no authorization or existence check, and the
ACCESS_DENIED code and `UNAUTHORIZED_DELETE_MESSAGE` make this guard read like
one.
I'd reword the message and the `isResolved` Javadoc to say it only rejects
statements that didn't come through a query endpoint, and consider exposing
each role's DELETE authorization (with the `tableExists` check) as a reusable
helper, so that a new entry point doesn't have to copy the private chain.
##########
pinot-common/src/main/java/org/apache/pinot/sql/parsers/dml/DeleteStatement.java:
##########
@@ -0,0 +1,351 @@
+/**
+ * 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.pinot.sql.parsers.dml;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.calcite.avatica.util.Casing;
+import org.apache.calcite.sql.SqlDelete;
+import org.apache.calcite.sql.SqlDialect;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlSyntax;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.util.SqlShuttle;
+import org.apache.calcite.sql.validate.SqlValidatorUtil;
+import org.apache.pinot.common.config.provider.TableCache;
+import org.apache.pinot.common.request.Expression;
+import org.apache.pinot.common.request.ExpressionType;
+import org.apache.pinot.common.request.Function;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.common.utils.DatabaseUtils;
+import org.apache.pinot.spi.config.task.AdhocTaskConfig;
+import org.apache.pinot.spi.exception.QueryErrorCode;
+import org.apache.pinot.spi.exception.QueryException;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.sql.parsers.CalciteSqlParser;
+import org.apache.pinot.sql.parsers.SqlNodeAndOptions;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+
+/// A SQL `DELETE FROM <table> WHERE <predicate>` statement.
+///
+/// Pinot parses `DELETE` but does not delete rows itself: the default SQL
executor answers that it is not supported,
+/// and a deployment that can delete rows, e.g. by purging the matching rows
from the segments with a minion task,
+/// executes the parsed statement by overriding
`SqlQueryExecutor#executeDelete`. Its execution type is
+/// [ExecutionType#EXECUTOR]: the generic [#execute()], [#getResultSchema()]
and [#generateAdhocTaskConfig()] do not
+/// apply to it.
+///
+/// Before handing the statement to the executor, the broker and the
controller resolve its table with
+/// [#resolveTableName], authorize the caller to delete rows from that table,
and check that the table exists with
+/// [#tableExists()].
+///
+/// Its options are the `SET` statements, the legacy `OPTION(...)` suffix and
the request `queryOptions` of the
+/// statement (`SET` takes precedence). The `database` option only qualifies
the table name, see [#resolveTableName].
+/// Every other option, including query options such as `timeoutMs`, is passed
to the executor through
+/// [#getOptions()], so that an option the executor relies on (e.g. a dry run)
cannot be reclassified as a query option
+/// and silently dropped.
+///
+/// Instances are immutable and thread-safe.
+public class DeleteStatement implements DataManipulationStatement {
+ public static final String NOT_SUPPORTED_MESSAGE =
+ "DELETE is not supported by this endpoint, it requires a SQL executor
that implements row deletion";
+
+ /// Serializes the WHERE clause into SQL that the Pinot parser reads back
into the same expression. Combined with
+ /// `quoteAllIdentifiers = false`, identifiers are only quoted (with double
quotes) when they were quoted.
+ private static final SqlDialect PINOT_SQL_DIALECT = new
SqlDialect(SqlDialect.EMPTY_CONTEXT
+ .withIdentifierQuoteString("\"")
+ .withLiteralQuoteString("'")
+ .withLiteralEscapedQuoteString("''")
+ .withUnquotedCasing(Casing.UNCHANGED)
+ .withQuotedCasing(Casing.UNCHANGED)
+ .withCaseSensitive(true));
+
+ /// Quotes the unquoted identifiers named like a SQL function without
arguments, e.g. `user`, `pi` or `current_date`:
+ /// Calcite unparses them as upper-cased keywords, while Pinot reads them as
columns, so they would name another
+ /// column. Mirrors the check of `SqlUtil.unparseSqlIdentifierSyntax`.
+ private static final SqlShuttle KEYWORD_IDENTIFIER_QUOTER = new SqlShuttle()
{
+ @Override
+ public SqlNode visit(SqlIdentifier identifier) {
+ if (identifier.isSimple() && !identifier.getParserPosition().isQuoted())
{
+ SqlOperator operator =
+
SqlValidatorUtil.lookupSqlFunctionByID(SqlStdOperatorTable.instance(),
identifier, null);
+ if (operator != null && (operator.getSyntax() == SqlSyntax.FUNCTION_ID
+ || operator.getSyntax() == SqlSyntax.FUNCTION_ID_CONSTANT)) {
+ return new SqlIdentifier(identifier.names, null,
SqlParserPos.QUOTED_ZERO,
+ List.of(SqlParserPos.QUOTED_ZERO));
+ }
+ }
+ return identifier;
+ }
+ };
+
+ /// Canonical names (lower case, without underscores) of the functions that
read another table than the one the
+ /// statement deletes from, which the caller is not authorized to read.
+ private static final Set<String> CROSS_TABLE_FUNCTIONS = Set.of("lookup",
"insubquery", "inpartitionedsubquery");
+
+ /// Canonical names of the functions that run a script. Rejected because the
broker's groovy policy
+ /// (`pinot.broker.disable.query.groovy`, the per-table
`queryConfig.disableGroovy` override and the static
+ /// analyzer) only applies to queries, not to a `DELETE`, and the executor
may run the predicate on another role.
+ private static final Set<String> SCRIPT_FUNCTIONS = Set.of("groovy");
+
+ private final String _tableName;
+ private final String _predicate;
+ @Nullable
+ private final String _database;
+ private final Map<String, String> _options;
+ private final boolean _resolved;
+ private final boolean _tableExists;
+
+ private DeleteStatement(String tableName, String predicate, @Nullable String
database, Map<String, String> options) {
+ this(tableName, predicate, database, options, false, false);
+ }
+
+ private DeleteStatement(String tableName, String predicate, @Nullable String
database, Map<String, String> options,
+ boolean resolved, boolean tableExists) {
+ _tableName = tableName;
+ _predicate = predicate;
+ _database = database;
+ _options = Collections.unmodifiableMap(new HashMap<>(options));
+ _resolved = resolved;
+ _tableExists = tableExists;
+ }
+
+ /// Parses a `DELETE` statement.
+ ///
+ /// @throws IllegalArgumentException if the statement is not a supported
`DELETE`: it must have a WHERE clause, no
+ /// table alias, a WHERE clause that Pinot
can parse as an expression, that does
+ /// not read another table (e.g. with
`lookUp` or `IN_SUBQUERY`) and that does not
+ /// run a script (`groovy`), and set the
database with the `database` option only
+ public static DeleteStatement parse(SqlNodeAndOptions sqlNodeAndOptions) {
+ SqlNode sqlNode = sqlNodeAndOptions.getSqlNode();
+ checkArgument(sqlNode instanceof SqlDelete, "Not a DELETE statement: %s",
sqlNode.getKind());
+ SqlDelete sqlDelete = (SqlDelete) sqlNode;
+ String tableName = getTableName(sqlDelete.getTargetTable());
+ checkArgument(sqlDelete.getAlias() == null,
+ "DELETE does not support a table alias, reference the columns
directly");
+ SqlNode condition = sqlDelete.getCondition();
+ checkArgument(condition != null, "DELETE requires a WHERE clause to
prevent an accidental full-table delete; use "
+ + "an explicit always-true predicate or delete the table segments to
remove all rows");
+ String predicate = toPinotSql(condition);
+
+ String database = null;
+ Map<String, String> options = new HashMap<>();
+ for (Map.Entry<String, String> option :
sqlNodeAndOptions.getOptions().entrySet()) {
+ String key = option.getKey();
+ if (key.equals(CommonConstants.DATABASE)) {
+ database = option.getValue();
+ } else if (key.equalsIgnoreCase(CommonConstants.DATABASE)) {
+ // Queries only read the exact `database` option (where they read it
at all, see resolveTableName): fail
+ // rather than delete from another table
+ throw new IllegalArgumentException(
+ "Unsupported option: " + key + ", set the database with the '" +
CommonConstants.DATABASE + "' option");
+ } else {
+ options.put(key, option.getValue());
+ }
+ }
+ return new DeleteStatement(tableName, predicate, database, options);
+ }
+
+ private static String getTableName(SqlNode targetTable) {
+ // Table hints and EXTEND clauses parse into other node types
+ checkArgument(targetTable instanceof SqlIdentifier, "DELETE only supports
a plain table name, got: %s",
+ targetTable);
+ // A quoted name part may contain a dot, which splits it like the table
name of a query
+ String tableName = String.join(".", ((SqlIdentifier) targetTable).names);
+ // Empty parts, e.g. in "db."."t", are rejected rather than dropped, so
that the table name is the one authorized
+ String[] parts = tableName.split("\\.", -1);
+ checkArgument(parts.length <= 2 &&
Arrays.stream(parts).noneMatch(String::isEmpty),
+ "Invalid table name: %s, expected [database.]table", tableName);
+ return tableName;
+ }
+
+ /// Serializes a WHERE clause back into SQL, and verifies that Pinot parses
it into the same expression.
+ private static String toPinotSql(SqlNode condition) {
+ Expression expression;
+ Expression serializedExpression;
+ String predicate;
+ try {
+ expression = CalciteSqlParser.compileToExpression(condition);
+ predicate = condition.accept(KEYWORD_IDENTIFIER_QUOTER)
+ .toSqlString(config -> config.withDialect(PINOT_SQL_DIALECT)
+ .withQuoteAllIdentifiers(false)
+ .withIndentation(0))
+ .getSql();
+ serializedExpression = CalciteSqlParser.compileToExpression(predicate);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Unsupported WHERE clause in DELETE:
" + describe(condition), e);
+ }
+ // Fail rather than hand over a predicate that selects other rows than the
statement
+ checkArgument(serializedExpression.equals(expression),
+ "Unsupported WHERE clause in DELETE, it cannot be serialized back into
the same expression: %s", condition);
+ checkFunctions(expression);
+ return predicate;
+ }
+
+ /// Renders a node for an error message without failing: rendering it is
what may have failed, e.g. for an operator
+ /// without unparse support such as `AT TIME ZONE`, in which case the cause
of the error carries the detail.
+ private static String describe(SqlNode node) {
+ try {
+ return node.toString();
+ } catch (RuntimeException e) {
+ return "<" + node.getKind() + " expression that cannot be rendered>";
+ }
+ }
+
+ /// Rejects the functions that read another table, as the caller is only
authorized for the table it deletes from,
+ /// and the functions that run a script, see [#SCRIPT_FUNCTIONS].
+ private static void checkFunctions(Expression expression) {
Review Comment:
The deny-lists cover every function the query path treats as cross-table or
as a script today, but `checkFunctions` only walks the compiled `Expression`,
and `CalciteSqlParser.toExpression` turns an `OVER` window spec into a plain
string literal (`node instanceof SqlWindow` compiles to
`getLiteralExpression(node.toString())`), which this method skips. Replaying
`toPinotSql`, `DELETE FROM t WHERE RANK() OVER (ORDER BY lookUp('dimTable',
'name', 'id', userId)) > 0` passes the round-trip check and compiles to
`GREATER_THAN(over(rank(), <string literal>), 0)`, with the `lookUp(...)` call
only inside the string literal, so no `lookup` function reaches this method.
The same holds for a `groovy(...)` call in a `PARTITION BY`. `getPredicate()`
then hands the executor a `lookUp` or `groovy` call that its Javadoc says is
rejected.
Neither engine evaluates a window function in WHERE today, so this isn't
exploitable in-tree, but executors are told to rely on that guarantee. A window
function is never valid in a DELETE predicate, so I'd reject `over` here (or
run the deny-lists over the `SqlNode` with a visitor that also walks into
`SqlWindow`), with a test.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java:
##########
@@ -715,6 +714,11 @@ private void setUpPinotController() {
// max.segment.completion.time.millis is seeded from cluster config before
serving requests.
_clusterConfigChangeHandler.registerClusterConfigChangeListener(_pinotLLCRealtimeSegmentManager);
LOGGER.info("Registered PinotLLCRealtimeSegmentManager as cluster config
change listener");
+ // Keeps the query option cluster configs in effect for the statements the
controller parses on /sql, as the broker
+ // does: the legacy OPTION(...) syntax mode for every statement, including
the DML it executes itself (a DELETE
+ // carrying legacy options fails in IGNORE and REJECT mode), and the
validation mode for the queries it parses
+ // before forwarding them
+ _clusterConfigChangeHandler.registerClusterConfigChangeListener(new
QueryOptionConfigListener());
Review Comment:
Registering this listener makes the broker cluster configs
`pinot.broker.query.option.validationMode` and `legacySyntaxMode` apply to
every parse in the controller JVM, but the REJECT allowlist is per JVM: keys
that broker plugins add with `QueryOptionsUtils.registerSqlQueryOptionKey` are
never registered here. With `validationMode=REJECT`, a controller `POST /sql`
of `SET myPluginKey = 'x'; SELECT * FROM t` now fails the controller's parse
(`PinotQueryResource` line 403) with SQL_PARSING "Unsupported query option
'myPluginKey'" while the broker accepts it; before this PR the controller
stayed in NONE and forwarded it. Separately, `validateMultiStageQuery` parses
outside its per-query try/catch (`PinotQueryResource` line 228), so with
`legacySyntaxMode=REJECT` a single query using `OPTION(...)` in the batch (or
an unknown `SET` key with `validationMode=REJECT`) now fails the whole request
with HTTP 500 instead of a `compiledSuccessfully=false` entry for that query.
The broker validates every query the controller forwards, so I'd apply only
the legacy syntax mode on the controller, which is what the DELETE it executes
needs (e.g. a flag on `QueryOptionConfigListener`), and move the line-228 parse
inside the per-query try.
##########
pinot-common/src/main/java/org/apache/pinot/sql/parsers/dml/DeleteStatement.java:
##########
@@ -0,0 +1,351 @@
+/**
+ * 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.pinot.sql.parsers.dml;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.calcite.avatica.util.Casing;
+import org.apache.calcite.sql.SqlDelete;
+import org.apache.calcite.sql.SqlDialect;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlSyntax;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.util.SqlShuttle;
+import org.apache.calcite.sql.validate.SqlValidatorUtil;
+import org.apache.pinot.common.config.provider.TableCache;
+import org.apache.pinot.common.request.Expression;
+import org.apache.pinot.common.request.ExpressionType;
+import org.apache.pinot.common.request.Function;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.common.utils.DatabaseUtils;
+import org.apache.pinot.spi.config.task.AdhocTaskConfig;
+import org.apache.pinot.spi.exception.QueryErrorCode;
+import org.apache.pinot.spi.exception.QueryException;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.sql.parsers.CalciteSqlParser;
+import org.apache.pinot.sql.parsers.SqlNodeAndOptions;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+
+/// A SQL `DELETE FROM <table> WHERE <predicate>` statement.
+///
+/// Pinot parses `DELETE` but does not delete rows itself: the default SQL
executor answers that it is not supported,
+/// and a deployment that can delete rows, e.g. by purging the matching rows
from the segments with a minion task,
+/// executes the parsed statement by overriding
`SqlQueryExecutor#executeDelete`. Its execution type is
+/// [ExecutionType#EXECUTOR]: the generic [#execute()], [#getResultSchema()]
and [#generateAdhocTaskConfig()] do not
+/// apply to it.
+///
+/// Before handing the statement to the executor, the broker and the
controller resolve its table with
+/// [#resolveTableName], authorize the caller to delete rows from that table,
and check that the table exists with
+/// [#tableExists()].
+///
+/// Its options are the `SET` statements, the legacy `OPTION(...)` suffix and
the request `queryOptions` of the
+/// statement (`SET` takes precedence). The `database` option only qualifies
the table name, see [#resolveTableName].
+/// Every other option, including query options such as `timeoutMs`, is passed
to the executor through
+/// [#getOptions()], so that an option the executor relies on (e.g. a dry run)
cannot be reclassified as a query option
+/// and silently dropped.
+///
+/// Instances are immutable and thread-safe.
+public class DeleteStatement implements DataManipulationStatement {
+ public static final String NOT_SUPPORTED_MESSAGE =
+ "DELETE is not supported by this endpoint, it requires a SQL executor
that implements row deletion";
+
+ /// Serializes the WHERE clause into SQL that the Pinot parser reads back
into the same expression. Combined with
+ /// `quoteAllIdentifiers = false`, identifiers are only quoted (with double
quotes) when they were quoted.
+ private static final SqlDialect PINOT_SQL_DIALECT = new
SqlDialect(SqlDialect.EMPTY_CONTEXT
+ .withIdentifierQuoteString("\"")
+ .withLiteralQuoteString("'")
+ .withLiteralEscapedQuoteString("''")
+ .withUnquotedCasing(Casing.UNCHANGED)
+ .withQuotedCasing(Casing.UNCHANGED)
+ .withCaseSensitive(true));
+
+ /// Quotes the unquoted identifiers named like a SQL function without
arguments, e.g. `user`, `pi` or `current_date`:
+ /// Calcite unparses them as upper-cased keywords, while Pinot reads them as
columns, so they would name another
+ /// column. Mirrors the check of `SqlUtil.unparseSqlIdentifierSyntax`.
+ private static final SqlShuttle KEYWORD_IDENTIFIER_QUOTER = new SqlShuttle()
{
+ @Override
+ public SqlNode visit(SqlIdentifier identifier) {
+ if (identifier.isSimple() && !identifier.getParserPosition().isQuoted())
{
+ SqlOperator operator =
+
SqlValidatorUtil.lookupSqlFunctionByID(SqlStdOperatorTable.instance(),
identifier, null);
+ if (operator != null && (operator.getSyntax() == SqlSyntax.FUNCTION_ID
+ || operator.getSyntax() == SqlSyntax.FUNCTION_ID_CONSTANT)) {
+ return new SqlIdentifier(identifier.names, null,
SqlParserPos.QUOTED_ZERO,
+ List.of(SqlParserPos.QUOTED_ZERO));
+ }
+ }
+ return identifier;
+ }
+ };
+
+ /// Canonical names (lower case, without underscores) of the functions that
read another table than the one the
+ /// statement deletes from, which the caller is not authorized to read.
+ private static final Set<String> CROSS_TABLE_FUNCTIONS = Set.of("lookup",
"insubquery", "inpartitionedsubquery");
+
+ /// Canonical names of the functions that run a script. Rejected because the
broker's groovy policy
+ /// (`pinot.broker.disable.query.groovy`, the per-table
`queryConfig.disableGroovy` override and the static
+ /// analyzer) only applies to queries, not to a `DELETE`, and the executor
may run the predicate on another role.
+ private static final Set<String> SCRIPT_FUNCTIONS = Set.of("groovy");
+
+ private final String _tableName;
+ private final String _predicate;
+ @Nullable
+ private final String _database;
+ private final Map<String, String> _options;
+ private final boolean _resolved;
+ private final boolean _tableExists;
+
+ private DeleteStatement(String tableName, String predicate, @Nullable String
database, Map<String, String> options) {
+ this(tableName, predicate, database, options, false, false);
+ }
+
+ private DeleteStatement(String tableName, String predicate, @Nullable String
database, Map<String, String> options,
+ boolean resolved, boolean tableExists) {
+ _tableName = tableName;
+ _predicate = predicate;
+ _database = database;
+ _options = Collections.unmodifiableMap(new HashMap<>(options));
+ _resolved = resolved;
+ _tableExists = tableExists;
+ }
+
+ /// Parses a `DELETE` statement.
+ ///
+ /// @throws IllegalArgumentException if the statement is not a supported
`DELETE`: it must have a WHERE clause, no
+ /// table alias, a WHERE clause that Pinot
can parse as an expression, that does
+ /// not read another table (e.g. with
`lookUp` or `IN_SUBQUERY`) and that does not
+ /// run a script (`groovy`), and set the
database with the `database` option only
+ public static DeleteStatement parse(SqlNodeAndOptions sqlNodeAndOptions) {
+ SqlNode sqlNode = sqlNodeAndOptions.getSqlNode();
+ checkArgument(sqlNode instanceof SqlDelete, "Not a DELETE statement: %s",
sqlNode.getKind());
+ SqlDelete sqlDelete = (SqlDelete) sqlNode;
+ String tableName = getTableName(sqlDelete.getTargetTable());
+ checkArgument(sqlDelete.getAlias() == null,
+ "DELETE does not support a table alias, reference the columns
directly");
+ SqlNode condition = sqlDelete.getCondition();
+ checkArgument(condition != null, "DELETE requires a WHERE clause to
prevent an accidental full-table delete; use "
+ + "an explicit always-true predicate or delete the table segments to
remove all rows");
+ String predicate = toPinotSql(condition);
+
+ String database = null;
+ Map<String, String> options = new HashMap<>();
+ for (Map.Entry<String, String> option :
sqlNodeAndOptions.getOptions().entrySet()) {
+ String key = option.getKey();
+ if (key.equals(CommonConstants.DATABASE)) {
+ database = option.getValue();
+ } else if (key.equalsIgnoreCase(CommonConstants.DATABASE)) {
+ // Queries only read the exact `database` option (where they read it
at all, see resolveTableName): fail
+ // rather than delete from another table
+ throw new IllegalArgumentException(
+ "Unsupported option: " + key + ", set the database with the '" +
CommonConstants.DATABASE + "' option");
+ } else {
+ options.put(key, option.getValue());
+ }
+ }
+ return new DeleteStatement(tableName, predicate, database, options);
+ }
+
+ private static String getTableName(SqlNode targetTable) {
+ // Table hints and EXTEND clauses parse into other node types
+ checkArgument(targetTable instanceof SqlIdentifier, "DELETE only supports
a plain table name, got: %s",
+ targetTable);
+ // A quoted name part may contain a dot, which splits it like the table
name of a query
+ String tableName = String.join(".", ((SqlIdentifier) targetTable).names);
+ // Empty parts, e.g. in "db."."t", are rejected rather than dropped, so
that the table name is the one authorized
+ String[] parts = tableName.split("\\.", -1);
+ checkArgument(parts.length <= 2 &&
Arrays.stream(parts).noneMatch(String::isEmpty),
+ "Invalid table name: %s, expected [database.]table", tableName);
+ return tableName;
+ }
+
+ /// Serializes a WHERE clause back into SQL, and verifies that Pinot parses
it into the same expression.
+ private static String toPinotSql(SqlNode condition) {
+ Expression expression;
+ Expression serializedExpression;
+ String predicate;
+ try {
+ expression = CalciteSqlParser.compileToExpression(condition);
+ predicate = condition.accept(KEYWORD_IDENTIFIER_QUOTER)
+ .toSqlString(config -> config.withDialect(PINOT_SQL_DIALECT)
+ .withQuoteAllIdentifiers(false)
+ .withIndentation(0))
+ .getSql();
+ serializedExpression = CalciteSqlParser.compileToExpression(predicate);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Unsupported WHERE clause in DELETE:
" + describe(condition), e);
+ }
+ // Fail rather than hand over a predicate that selects other rows than the
statement
+ checkArgument(serializedExpression.equals(expression),
+ "Unsupported WHERE clause in DELETE, it cannot be serialized back into
the same expression: %s", condition);
+ checkFunctions(expression);
+ return predicate;
+ }
+
+ /// Renders a node for an error message without failing: rendering it is
what may have failed, e.g. for an operator
+ /// without unparse support such as `AT TIME ZONE`, in which case the cause
of the error carries the detail.
+ private static String describe(SqlNode node) {
+ try {
+ return node.toString();
+ } catch (RuntimeException e) {
+ return "<" + node.getKind() + " expression that cannot be rendered>";
+ }
+ }
+
+ /// Rejects the functions that read another table, as the caller is only
authorized for the table it deletes from,
+ /// and the functions that run a script, see [#SCRIPT_FUNCTIONS].
+ private static void checkFunctions(Expression expression) {
+ if (expression.getType() != ExpressionType.FUNCTION) {
+ return;
+ }
+ Function function = expression.getFunctionCall();
+ String functionName = function.getOperator().replace("_",
"").toLowerCase(Locale.ROOT);
Review Comment:
`RequestUtils.canonicalizeFunctionName` has already lower-cased the operator
with the JVM default locale, and `toLowerCase(Locale.ROOT)` here leaves a
dotless `ı` (U+0131) as it is. On a broker or controller whose default locale
is Turkish or Azerbaijani (`-Duser.language=tr`), `DELETE FROM myTable WHERE
IN_SUBQUERY(userId, 'SELECT ID_SET(userId) FROM other') = 1` compiles to the
operator `ınsubquery`, which isn't in `CROSS_TABLE_FUNCTIONS`, so `parse()`
accepts it (and the `IN_SUBQUERY` case in `DeleteStatementTest` fails under
that locale). In any locale, a caller can also type the `ı` directly
(`ın_subquery(...)`, `ınPartıtıonedSubquery(...)`), since the parser accepts
U+0100 to U+1FFF in unquoted identifiers. Whether that becomes a cross-table
read depends on the executor, but anything that resolves function names with
`equalsIgnoreCase`, as `ServerQueryExecutorV1Impl` does for
`IN_PARTITIONED_SUBQUERY`, maps `ı` to `I` and runs the real function.
I'd fold with `toUpperCase(Locale.ROOT)` (which maps `ı` to `I`) and compare
against upper-case names, or reject non-ASCII function names outright, and add
tests that run `parse()` under `Locale.forLanguageTag("tr-TR")` and with a
U+0131 spelling.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotQueryResource.java:
##########
@@ -424,6 +442,77 @@ private StreamingOutput executeSqlQuery(@Context
HttpHeaders httpHeaders, String
}
}
+ /// Executes a `DELETE` once the caller is authorized to delete rows from
its table, see [#authorizeDelete].
+ ///
+ /// The caller is first checked without a table (the `READ` access type with
the `/sql` endpoint, as for a
+ /// multi-stage query), before the statement is parsed and its table looked
up: with basic auth, an unauthenticated
+ /// caller gets HTTP 401 here, before the table cache is consulted. The
table is then resolved, with the database of
+ /// the request and in the case it is defined with, the caller authorized to
delete rows from it, and only then is
+ /// the table checked to exist, so that the existence of a table is not
leaked to a caller who is not authorized for
+ /// it. The executor deletes rows from that exact table.
+ ///
+ /// The executor runs before the response streams, so that an exception it
lets escape is mapped to an error
+ /// response by [#executeSqlQueryCatching] rather than failing the request
with HTTP 500.
+ ///
+ /// @throws QueryException with [QueryErrorCode#ACCESS_DENIED] if the caller
is not authorized, with
+ /// [QueryErrorCode#TABLE_DOES_NOT_EXIST] if the
table does not exist, and with the errors of
+ /// [DataManipulationStatementParser#parse] and
[DeleteStatement#resolveTableName]
+ private StreamingOutput executeDelete(SqlNodeAndOptions sqlNodeAndOptions,
HttpHeaders httpHeaders) {
+ AccessControl accessControl = _accessControlFactory.create();
+ if (!accessControl.hasAccess(AccessType.READ, httpHeaders, SQL_ENDPOINT)) {
+ throw QueryErrorCode.ACCESS_DENIED.asException("Permission denied to
delete rows");
+ }
+ DeleteStatement statement = ((DeleteStatement)
DataManipulationStatementParser.parse(sqlNodeAndOptions))
+
.resolveTableName(httpHeaders.getHeaderString(CommonConstants.DATABASE),
+ _pinotHelixResourceManager.getTableCache());
+ authorizeDelete(accessControl, statement.getTableName(), httpHeaders);
+ if (!statement.tableExists()) {
+ throw QueryErrorCode.TABLE_DOES_NOT_EXIST.asException("Table does not
exist: " + statement.getTableName());
+ }
+ LOGGER.info("Executing DELETE on table: {}, predicate: {}, options: {},
client: {}", statement.getTableName(),
+ statement.getPredicate(), statement.getOptions().keySet(),
getClientIp(httpHeaders));
+ BrokerResponse response = _sqlQueryExecutor.executeStatement(statement,
extractHeaders(httpHeaders));
+ return output -> {
+ try (OutputStream os = output) {
+ response.toOutputStream(os);
+ }
+ };
+ }
+
+ /// Client of the request for the log, from the proxy headers as
`HttpRequesterIdentity#getClientIp` reads them.
+ private static String getClientIp(HttpHeaders httpHeaders) {
+ String forwardedFor = httpHeaders.getHeaderString("X-Forwarded-For");
+ if (forwardedFor != null) {
+ return forwardedFor;
+ }
+ String realIp = httpHeaders.getHeaderString("X-Real-IP");
+ return realIp != null ? realIp : CommonConstants.UNKNOWN;
+ }
+
+ /// Authorizes the caller, who passed the caller-level check of
[#executeDelete], to delete rows from the table.
+ ///
+ /// It checks, on the raw table name and in this order, the `READ` access
type with the `/sql` endpoint (the request
+ /// path, as the authentication filter passes for the other endpoints), the
fine-grained [Actions.Table#QUERY]
+ /// action, the `DELETE` access type with the `/sql` endpoint and the
fine-grained [Actions.Table#DELETE_ROWS]
+ /// action. The `READ` and `QUERY` checks are there because the WHERE clause
reads the table, the `DELETE` and
+ /// `DELETE_ROWS` ones because rows are deleted, as the other deletions of
the controller (e.g. of segments) check
+ /// `DELETE` on the raw table name. This is stricter than the controller's
query path, which only checks the `READ`
+ /// access type with `"Query"` as the endpoint and no fine-grained action,
and it does not apply the row-level
+ /// security of the brokers. Unlike the broker, which has no access type to
tell a deletion from a query and denies
+ /// it by default, an access control that does not tell access types apart
and allows every fine-grained action
+ /// lets every reader of the table delete rows, as it lets them delete
segments.
+ ///
+ /// @throws QueryException with [QueryErrorCode#ACCESS_DENIED] if the caller
is not authorized, as for queries
+ private void authorizeDelete(AccessControl accessControl, String tableName,
HttpHeaders httpHeaders) {
+ String rawTableName = TableNameBuilder.extractRawTableName(tableName);
+ if (!accessControl.hasAccess(rawTableName, AccessType.READ, httpHeaders,
SQL_ENDPOINT)
+ || !accessControl.hasAccess(httpHeaders, TargetType.TABLE,
rawTableName, Actions.Table.QUERY)
+ || !accessControl.hasAccess(rawTableName, AccessType.DELETE,
httpHeaders, SQL_ENDPOINT)
+ || !accessControl.hasAccess(httpHeaders, TargetType.TABLE,
rawTableName, Actions.Table.DELETE_ROWS)) {
+ throw QueryErrorCode.ACCESS_DENIED.asException("Permission denied to
delete rows from table: " + tableName);
Review Comment:
With `enable.case.insensitive=true` (the default), this denial names the
table as `resolveTableName` resolved it, which breaks the promise in the
`executeDelete` Javadoc (and on `tableExists()`) that a caller who isn't
authorized for a table doesn't learn whether it exists. A principal that passes
the `/sql` READ check but not the table checks (e.g. BasicAuth
`tables=OtherTable, permissions=READ`) can post `DELETE FROM SECRETTABLE WHERE
1 = 1` and gets "Permission denied to delete rows from table: SecretTable" if
the table exists and "...: SECRETTABLE" if it doesn't, so one probe reveals
both existence and the real name, while the controller's query path returns a
bare ACCESS_DENIED. The logical-table rejection leaks one step earlier:
`resolveTableName` throws "DELETE does not support logical tables" before
`authorizeDelete` runs, so any READ caller can tell logical tables apart.
I'd keep authorizing the resolved name but put the name as the caller wrote
it (or none) in the denial, and raise the logical-table error only after the
table checks pass. The broker's `deleteAccessDenied` echoes the resolved name
the same way, and so does BasicAuth's reason ("Authorization Failed for tables:
[SecretTable]"), as it already does for queries, so either scrub both or narrow
the Javadoc to say only that TABLE_DOES_NOT_EXIST comes after authorization.
The impact is small where callers can list tables with `GET /tables`, but it
matters for access controls that hide tables.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotQueryResource.java:
##########
@@ -424,6 +442,77 @@ private StreamingOutput executeSqlQuery(@Context
HttpHeaders httpHeaders, String
}
}
+ /// Executes a `DELETE` once the caller is authorized to delete rows from
its table, see [#authorizeDelete].
+ ///
+ /// The caller is first checked without a table (the `READ` access type with
the `/sql` endpoint, as for a
+ /// multi-stage query), before the statement is parsed and its table looked
up: with basic auth, an unauthenticated
+ /// caller gets HTTP 401 here, before the table cache is consulted. The
table is then resolved, with the database of
+ /// the request and in the case it is defined with, the caller authorized to
delete rows from it, and only then is
+ /// the table checked to exist, so that the existence of a table is not
leaked to a caller who is not authorized for
+ /// it. The executor deletes rows from that exact table.
+ ///
+ /// The executor runs before the response streams, so that an exception it
lets escape is mapped to an error
+ /// response by [#executeSqlQueryCatching] rather than failing the request
with HTTP 500.
+ ///
+ /// @throws QueryException with [QueryErrorCode#ACCESS_DENIED] if the caller
is not authorized, with
+ /// [QueryErrorCode#TABLE_DOES_NOT_EXIST] if the
table does not exist, and with the errors of
+ /// [DataManipulationStatementParser#parse] and
[DeleteStatement#resolveTableName]
+ private StreamingOutput executeDelete(SqlNodeAndOptions sqlNodeAndOptions,
HttpHeaders httpHeaders) {
+ AccessControl accessControl = _accessControlFactory.create();
+ if (!accessControl.hasAccess(AccessType.READ, httpHeaders, SQL_ENDPOINT)) {
+ throw QueryErrorCode.ACCESS_DENIED.asException("Permission denied to
delete rows");
+ }
+ DeleteStatement statement = ((DeleteStatement)
DataManipulationStatementParser.parse(sqlNodeAndOptions))
+
.resolveTableName(httpHeaders.getHeaderString(CommonConstants.DATABASE),
+ _pinotHelixResourceManager.getTableCache());
+ authorizeDelete(accessControl, statement.getTableName(), httpHeaders);
+ if (!statement.tableExists()) {
+ throw QueryErrorCode.TABLE_DOES_NOT_EXIST.asException("Table does not
exist: " + statement.getTableName());
+ }
+ LOGGER.info("Executing DELETE on table: {}, predicate: {}, options: {},
client: {}", statement.getTableName(),
+ statement.getPredicate(), statement.getOptions().keySet(),
getClientIp(httpHeaders));
+ BrokerResponse response = _sqlQueryExecutor.executeStatement(statement,
extractHeaders(httpHeaders));
+ return output -> {
+ try (OutputStream os = output) {
+ response.toOutputStream(os);
+ }
+ };
+ }
+
+ /// Client of the request for the log, from the proxy headers as
`HttpRequesterIdentity#getClientIp` reads them.
Review Comment:
The Javadoc says this reads the proxy headers as
`HttpRequesterIdentity#getClientIp` does, but the output differs. With
`X-Forwarded-For: 203.0.113.7, 10.0.0.1`, the broker's "Executing DELETE" line
logs `client: 203.0.113.7; 10.0.0.1` (it rewrites `,` to `;`), while this logs
`client: 203.0.113.7, 10.0.0.1`. With both `X-Forwarded-For: 203.0.113.7` and
`X-Real-IP: 10.0.0.1`, the broker concatenates them as `203.0.113.710.0.0.1`
and this returns only the `X-Forwarded-For` value. So the same audit line has a
different client format depending on which role ran the DELETE.
Since pinot-controller doesn't depend on pinot-broker, consider one static
helper in pinot-spi or pinot-common, used by `HttpRequesterIdentity`,
`GrpcRequesterIdentity` and this resource, that prefers `X-Forwarded-For`,
falls back to `X-Real-IP` and turns `,` into `;`; that would also fix the
broker gluing both headers together. At minimum, apply the `;` substitution
here and drop the claim that it matches.
##########
pinot-common/src/main/java/org/apache/pinot/sql/parsers/CalciteSqlParser.java:
##########
@@ -165,8 +169,8 @@ private static SqlNodeAndOptions
extractSqlNodeAndOptions(SqlNodeList sqlNodeLis
SqlNode statementNode = null;
Map<String, String> options = new HashMap<>();
for (SqlNode sqlNode : sqlNodeList) {
- if (sqlNode instanceof SqlInsertFromFile) {
- // extract insert statement (execution statement)
+ if (sqlNode instanceof SqlInsertFromFile || sqlNode instanceof
SqlDelete) {
Review Comment:
Only a top-level `SqlDelete` is classified as DML here, but `EXPLAIN PLAN
FOR` accepts any `SqlQueryOrDml`, so `EXPLAIN PLAN FOR DELETE FROM myTable
WHERE a = 1` is still classified as DQL. It passes the `instanceof SqlDelete`
checks in `PinotClientRequest` and `BrokerRequestHandlerDelegate` and reaches
`compileWithoutRewrite`, which unwraps the explicandum and casts it to
`SqlSelect`. On the single-stage engine (broker `/query/sql`, GET included, and
controller `/sql`) the client gets SQL_PARSING `class
org.apache.calcite.sql.SqlDelete cannot be cast to class
org.apache.calcite.sql.SqlSelect`, the confusing cast error this PR removes for
a bare DELETE (the multi-stage engine reports "Unsupported SQL query, failed to
validate query" instead). The first commit rejected EXPLAIN of DML at parse
time with a `CalciteSqlCompilerTest` case, but "Simplify SQL DELETE handling"
dropped both. Please restore the check (e.g. throw
`SqlCompilationException("EXPLAIN is not supported for DELETE")
` when a `SqlExplain`'s explicandum is a `SqlDelete`) with its test.
##########
pinot-common/src/main/java/org/apache/pinot/sql/parsers/dml/DeleteStatement.java:
##########
@@ -0,0 +1,351 @@
+/**
+ * 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.pinot.sql.parsers.dml;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.calcite.avatica.util.Casing;
+import org.apache.calcite.sql.SqlDelete;
+import org.apache.calcite.sql.SqlDialect;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlSyntax;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.util.SqlShuttle;
+import org.apache.calcite.sql.validate.SqlValidatorUtil;
+import org.apache.pinot.common.config.provider.TableCache;
+import org.apache.pinot.common.request.Expression;
+import org.apache.pinot.common.request.ExpressionType;
+import org.apache.pinot.common.request.Function;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.common.utils.DatabaseUtils;
+import org.apache.pinot.spi.config.task.AdhocTaskConfig;
+import org.apache.pinot.spi.exception.QueryErrorCode;
+import org.apache.pinot.spi.exception.QueryException;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.sql.parsers.CalciteSqlParser;
+import org.apache.pinot.sql.parsers.SqlNodeAndOptions;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+
+/// A SQL `DELETE FROM <table> WHERE <predicate>` statement.
+///
+/// Pinot parses `DELETE` but does not delete rows itself: the default SQL
executor answers that it is not supported,
+/// and a deployment that can delete rows, e.g. by purging the matching rows
from the segments with a minion task,
+/// executes the parsed statement by overriding
`SqlQueryExecutor#executeDelete`. Its execution type is
+/// [ExecutionType#EXECUTOR]: the generic [#execute()], [#getResultSchema()]
and [#generateAdhocTaskConfig()] do not
+/// apply to it.
+///
+/// Before handing the statement to the executor, the broker and the
controller resolve its table with
+/// [#resolveTableName], authorize the caller to delete rows from that table,
and check that the table exists with
+/// [#tableExists()].
+///
+/// Its options are the `SET` statements, the legacy `OPTION(...)` suffix and
the request `queryOptions` of the
+/// statement (`SET` takes precedence). The `database` option only qualifies
the table name, see [#resolveTableName].
+/// Every other option, including query options such as `timeoutMs`, is passed
to the executor through
+/// [#getOptions()], so that an option the executor relies on (e.g. a dry run)
cannot be reclassified as a query option
+/// and silently dropped.
+///
+/// Instances are immutable and thread-safe.
+public class DeleteStatement implements DataManipulationStatement {
+ public static final String NOT_SUPPORTED_MESSAGE =
+ "DELETE is not supported by this endpoint, it requires a SQL executor
that implements row deletion";
+
+ /// Serializes the WHERE clause into SQL that the Pinot parser reads back
into the same expression. Combined with
+ /// `quoteAllIdentifiers = false`, identifiers are only quoted (with double
quotes) when they were quoted.
+ private static final SqlDialect PINOT_SQL_DIALECT = new
SqlDialect(SqlDialect.EMPTY_CONTEXT
+ .withIdentifierQuoteString("\"")
+ .withLiteralQuoteString("'")
+ .withLiteralEscapedQuoteString("''")
+ .withUnquotedCasing(Casing.UNCHANGED)
+ .withQuotedCasing(Casing.UNCHANGED)
+ .withCaseSensitive(true));
+
+ /// Quotes the unquoted identifiers named like a SQL function without
arguments, e.g. `user`, `pi` or `current_date`:
+ /// Calcite unparses them as upper-cased keywords, while Pinot reads them as
columns, so they would name another
+ /// column. Mirrors the check of `SqlUtil.unparseSqlIdentifierSyntax`.
+ private static final SqlShuttle KEYWORD_IDENTIFIER_QUOTER = new SqlShuttle()
{
+ @Override
+ public SqlNode visit(SqlIdentifier identifier) {
+ if (identifier.isSimple() && !identifier.getParserPosition().isQuoted())
{
+ SqlOperator operator =
+
SqlValidatorUtil.lookupSqlFunctionByID(SqlStdOperatorTable.instance(),
identifier, null);
+ if (operator != null && (operator.getSyntax() == SqlSyntax.FUNCTION_ID
+ || operator.getSyntax() == SqlSyntax.FUNCTION_ID_CONSTANT)) {
+ return new SqlIdentifier(identifier.names, null,
SqlParserPos.QUOTED_ZERO,
+ List.of(SqlParserPos.QUOTED_ZERO));
+ }
+ }
+ return identifier;
+ }
+ };
+
+ /// Canonical names (lower case, without underscores) of the functions that
read another table than the one the
+ /// statement deletes from, which the caller is not authorized to read.
+ private static final Set<String> CROSS_TABLE_FUNCTIONS = Set.of("lookup",
"insubquery", "inpartitionedsubquery");
+
+ /// Canonical names of the functions that run a script. Rejected because the
broker's groovy policy
+ /// (`pinot.broker.disable.query.groovy`, the per-table
`queryConfig.disableGroovy` override and the static
+ /// analyzer) only applies to queries, not to a `DELETE`, and the executor
may run the predicate on another role.
+ private static final Set<String> SCRIPT_FUNCTIONS = Set.of("groovy");
+
+ private final String _tableName;
+ private final String _predicate;
+ @Nullable
+ private final String _database;
+ private final Map<String, String> _options;
+ private final boolean _resolved;
+ private final boolean _tableExists;
+
+ private DeleteStatement(String tableName, String predicate, @Nullable String
database, Map<String, String> options) {
+ this(tableName, predicate, database, options, false, false);
+ }
+
+ private DeleteStatement(String tableName, String predicate, @Nullable String
database, Map<String, String> options,
+ boolean resolved, boolean tableExists) {
+ _tableName = tableName;
+ _predicate = predicate;
+ _database = database;
+ _options = Collections.unmodifiableMap(new HashMap<>(options));
+ _resolved = resolved;
+ _tableExists = tableExists;
+ }
+
+ /// Parses a `DELETE` statement.
+ ///
+ /// @throws IllegalArgumentException if the statement is not a supported
`DELETE`: it must have a WHERE clause, no
+ /// table alias, a WHERE clause that Pinot
can parse as an expression, that does
+ /// not read another table (e.g. with
`lookUp` or `IN_SUBQUERY`) and that does not
+ /// run a script (`groovy`), and set the
database with the `database` option only
+ public static DeleteStatement parse(SqlNodeAndOptions sqlNodeAndOptions) {
+ SqlNode sqlNode = sqlNodeAndOptions.getSqlNode();
+ checkArgument(sqlNode instanceof SqlDelete, "Not a DELETE statement: %s",
sqlNode.getKind());
+ SqlDelete sqlDelete = (SqlDelete) sqlNode;
+ String tableName = getTableName(sqlDelete.getTargetTable());
+ checkArgument(sqlDelete.getAlias() == null,
+ "DELETE does not support a table alias, reference the columns
directly");
+ SqlNode condition = sqlDelete.getCondition();
+ checkArgument(condition != null, "DELETE requires a WHERE clause to
prevent an accidental full-table delete; use "
+ + "an explicit always-true predicate or delete the table segments to
remove all rows");
+ String predicate = toPinotSql(condition);
+
+ String database = null;
+ Map<String, String> options = new HashMap<>();
+ for (Map.Entry<String, String> option :
sqlNodeAndOptions.getOptions().entrySet()) {
+ String key = option.getKey();
+ if (key.equals(CommonConstants.DATABASE)) {
+ database = option.getValue();
+ } else if (key.equalsIgnoreCase(CommonConstants.DATABASE)) {
+ // Queries only read the exact `database` option (where they read it
at all, see resolveTableName): fail
+ // rather than delete from another table
+ throw new IllegalArgumentException(
+ "Unsupported option: " + key + ", set the database with the '" +
CommonConstants.DATABASE + "' option");
+ } else {
+ options.put(key, option.getValue());
+ }
+ }
+ return new DeleteStatement(tableName, predicate, database, options);
+ }
+
+ private static String getTableName(SqlNode targetTable) {
+ // Table hints and EXTEND clauses parse into other node types
+ checkArgument(targetTable instanceof SqlIdentifier, "DELETE only supports
a plain table name, got: %s",
+ targetTable);
+ // A quoted name part may contain a dot, which splits it like the table
name of a query
+ String tableName = String.join(".", ((SqlIdentifier) targetTable).names);
+ // Empty parts, e.g. in "db."."t", are rejected rather than dropped, so
that the table name is the one authorized
+ String[] parts = tableName.split("\\.", -1);
+ checkArgument(parts.length <= 2 &&
Arrays.stream(parts).noneMatch(String::isEmpty),
+ "Invalid table name: %s, expected [database.]table", tableName);
+ return tableName;
+ }
+
+ /// Serializes a WHERE clause back into SQL, and verifies that Pinot parses
it into the same expression.
+ private static String toPinotSql(SqlNode condition) {
+ Expression expression;
+ Expression serializedExpression;
+ String predicate;
+ try {
+ expression = CalciteSqlParser.compileToExpression(condition);
+ predicate = condition.accept(KEYWORD_IDENTIFIER_QUOTER)
+ .toSqlString(config -> config.withDialect(PINOT_SQL_DIALECT)
+ .withQuoteAllIdentifiers(false)
+ .withIndentation(0))
+ .getSql();
+ serializedExpression = CalciteSqlParser.compileToExpression(predicate);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Unsupported WHERE clause in DELETE:
" + describe(condition), e);
+ }
+ // Fail rather than hand over a predicate that selects other rows than the
statement
+ checkArgument(serializedExpression.equals(expression),
+ "Unsupported WHERE clause in DELETE, it cannot be serialized back into
the same expression: %s", condition);
+ checkFunctions(expression);
+ return predicate;
+ }
+
+ /// Renders a node for an error message without failing: rendering it is
what may have failed, e.g. for an operator
+ /// without unparse support such as `AT TIME ZONE`, in which case the cause
of the error carries the detail.
+ private static String describe(SqlNode node) {
+ try {
+ return node.toString();
+ } catch (RuntimeException e) {
+ return "<" + node.getKind() + " expression that cannot be rendered>";
+ }
+ }
+
+ /// Rejects the functions that read another table, as the caller is only
authorized for the table it deletes from,
+ /// and the functions that run a script, see [#SCRIPT_FUNCTIONS].
+ private static void checkFunctions(Expression expression) {
+ if (expression.getType() != ExpressionType.FUNCTION) {
+ return;
+ }
+ Function function = expression.getFunctionCall();
+ String functionName = function.getOperator().replace("_",
"").toLowerCase(Locale.ROOT);
+ checkArgument(!CROSS_TABLE_FUNCTIONS.contains(functionName),
+ "Unsupported WHERE clause in DELETE, %s reads another table",
function.getOperator());
+ checkArgument(!SCRIPT_FUNCTIONS.contains(functionName),
+ "Unsupported WHERE clause in DELETE, %s is not supported in a DELETE
predicate", function.getOperator());
+ if (function.getOperands() != null) {
+ for (Expression operand : function.getOperands()) {
+ checkFunctions(operand);
+ }
+ }
+ }
+
+ /// Returns the statement with its table name resolved: qualified with the
database of the request, and in the case
+ /// the table is defined with (table names are case-insensitive by default).
The broker and the controller authorize
+ /// the caller for the resolved table name, check that the table exists with
[#tableExists()], and hand the resolved
+ /// statement to the executor.
+ ///
+ /// The database of the request is the `database` request header, else the
`database` option of the statement, as
+ /// for a multi-stage query (see
`DatabaseUtils#extractDatabaseFromQueryRequest`) and for the controller `/sql`
+ /// endpoint. They must match when both are set, and the database of a
`database.table` name must match them. Note
+ /// that the broker's single-stage engine resolves the table of a query from
the `database` header only, so a
+ /// preview of the predicate with `SET database` on that engine may read
another table than the one the `DELETE`
+ /// resolves to: preview with the `database` header or a qualified
`database.table` name instead.
+ ///
+ /// @param databaseHeader value of the `database` request header, if any
+ /// @param tableCache tables of the cluster, to resolve the case of the
table name. A table name it does not know
+ /// keeps the case of the statement, and the resolved
statement reports [#tableExists()] false.
+ /// @throws QueryException with [QueryErrorCode#QUERY_VALIDATION] if the
`database` header, the `database` option
+ /// and the database of a `database.table` name do
not match (a
+ /// `DatabaseConflictException`), or if the table is
a logical table, which `DELETE` does
+ /// not support
+ public DeleteStatement resolveTableName(@Nullable String databaseHeader,
TableCache tableCache)
+ throws QueryException {
+ String database =
DatabaseUtils.extractDatabaseFromOptionAndHeader(_database, databaseHeader);
Review Comment:
Following up on the last round: the Javadoc and the test now call this out
as intentional, but I don't think documenting it is enough for a destructive
statement. `resolveTableName` qualifies an unqualified table with the
`database` option when no `database` header is sent, while the default
single-stage engine resolves a query's table from the header only, and the
controller `/sql` forwards single-stage queries without adding the header. So
the same SQL text reads one table and deletes from another: with `orders` and
`sales.orders` on the same broker tenant and no `database` header, `SET
database = 'sales'; SELECT COUNT(*) FROM orders WHERE region = 'EU'` counts
rows of the default-database `orders`, while `SET database = 'sales'; DELETE
FROM orders WHERE region = 'EU'` deletes from `sales.orders`, whose rows the
user never previewed. It isn't an authorization bypass, since DELETE_ROWS is
checked on `sales.orders`.
I'd reject the `database` option on an unqualified table name unless the
header is also set (they already have to match), with an error asking for the
header or a qualified `sales.orders`, and change the `DeleteStatementTest` case
at line 104 to expect that error. Making the single-stage engine honor the
option instead would change the results of existing SELECTs, so that belongs in
a separate change.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/api/resources/PinotClientRequest.java:
##########
@@ -713,6 +739,137 @@ private BrokerResponse executeSqlQuery(ObjectNode
sqlRequestJson, HttpRequesterI
}
}
+ /// Executes a `DELETE` once the caller is authorized to delete rows from
its table.
+ ///
+ /// The first-step access control runs first, as for queries, followed by
the table check without a table, so that
+ /// an access control that only authenticates the caller in that check
rejects an unauthenticated caller before the
+ /// table is looked up. The table is then resolved with the database of the
+ /// request and in the case it is defined with, the caller is authorized to
delete rows from it (see
+ /// [#authorizeDelete]), and the executor deletes rows from that exact
table. A table the table cache does not know
+ /// fails with [QueryErrorCode#TABLE_DOES_NOT_EXIST], as a query does, but
only once the caller is authorized for
+ /// it, so that the existence of a table is not leaked to a caller who is
not authorized for it.
+ private BrokerResponse executeDelete(SqlNodeAndOptions sqlNodeAndOptions,
Map<String, String> headers,
+ HttpRequesterIdentity requesterIdentity, @Nullable HttpHeaders
httpHeaders) {
+ AccessControl accessControl = _accessControlFactory.create();
+ // The first-step access control runs before the table is looked up, as
for queries
+ AuthorizationResult authorizationResult =
accessControl.authorize(requesterIdentity);
+ if (authorizationResult.hasAccess()) {
+ // Access controls that authenticate the caller in their table check
rather than in the first step (e.g. the
+ // ZooKeeper basic auth) reject an unauthenticated caller here, before
the table is looked up
+ authorizationResult = accessControl.authorize(requesterIdentity,
Set.of());
+ }
+ if (!authorizationResult.hasAccess()) {
+ throw deleteAccessDenied(null, authorizationResult);
+ }
+ if (_tableCache == null) {
+ return new BrokerResponseNative(QueryErrorCode.QUERY_VALIDATION,
+ "DELETE is not supported by this broker: no table cache was
configured");
+ }
+ DeleteStatement statement;
+ try {
+ String databaseHeader = httpHeaders != null ?
httpHeaders.getHeaderString(CommonConstants.DATABASE) : null;
+ statement = ((DeleteStatement)
DataManipulationStatementParser.parse(sqlNodeAndOptions))
+ .resolveTableName(databaseHeader, _tableCache);
+ } catch (QueryException e) {
+ // e.g. an invalid statement, a logical table, or a database header that
does not match the statement
+ return new BrokerResponseNative(e.getErrorCode(), e.getMessage());
+ }
+ authorizeDelete(accessControl, statement.getTableName(),
requesterIdentity, httpHeaders);
+ if (!statement.tableExists()) {
+ return new BrokerResponseNative(QueryErrorCode.TABLE_DOES_NOT_EXIST,
+ "Table does not exist: " + statement.getTableName());
+ }
+ LOGGER.info("Executing DELETE on table: {}, predicate: {}, options: {},
client: {}", statement.getTableName(),
Review Comment:
This INFO line logs `getPredicate()` with its literal values and doesn't go
through `QueryLogger.redactQuery`, which the other broker query-logging sites
use. With `pinot.broker.query.log.sqlRedaction=full`, an authorized `DELETE
FROM users WHERE email = '[email protected]'` still writes `predicate: email =
'[email protected]'` to the broker log, which is exactly the record a
privacy-driven delete shouldn't leave behind. CR/LF aren't escaped either:
Calcite's `quoteStringLiteral` only doubles quotes, so a newline inside a
string literal or a quoted identifier splits the record, and the caller can
append a line that looks like a real broker log entry. The `client` value is
the raw `X-Forwarded-For`/`X-Real-IP` header, which the caller controls unless
a proxy overwrites it, and it is logged even though `QueryLogger` only logs
client IPs when `pinot.broker.request.client.ip.logging` is on (off by default).
I'd apply the redaction mode here (or log only the table and option keys),
and escape backslash, CR and LF as `QueryLogger.toSingleLine` does. The
controller's line in `PinotQueryResource.executeDelete` has the same newline
issue, though the controller has no redaction setting and already logs raw SQL
at INFO.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
