[
https://issues.apache.org/jira/browse/FLINK-35560?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
dongwoo.kim updated FLINK-35560:
--------------------------------
Summary: Support custom query operation validator for sql gateway (was:
Add query validator support to flink sql gateway via spi pattern)
> Support custom query operation validator for sql gateway
> --------------------------------------------------------
>
> Key: FLINK-35560
> URL: https://issues.apache.org/jira/browse/FLINK-35560
> Project: Flink
> Issue Type: Improvement
> Components: Table SQL / Gateway
> Reporter: dongwoo.kim
> Priority: Major
> Labels: pull-request-available
>
> h3. Summary
> Hello I'd like to suggest query validator support in flink sql gateway via
> spi pattern.
> As an sql gateway operator, there is need for query validation to only
> execute safe queries and drop unsafe queries.
> To address this need, I propose adding a {{QueryValidator}} interface in
> flink sql gateway api package.
> This interface will allow users to implement their own query validation
> logic, providing benefits to flink sql gateway operators.
> h3. Interface
> Below is a draft for the interface.
> It takes Operation and check whether the query is valid or not.
> {code:java}
> package org.apache.flink.table.gateway.api.validator;
> import org.apache.flink.annotation.Public;
> import org.apache.flink.table.operations.Operation;
> /**
> * Interface for implementing a validator that checks the safety of executing
> queries.
> */
> @Public
> public interface QueryValidator {
> boolean validateQuery(Operation op);
> }
> {code}
> h3. Example implementation
> Below is an example implementation that inspects Kafka table options,
> specifically {{{}scan.startup.timestamp-millis{}}}, and reject when the value
> is too small, which can cause high disk I/O load.
> {code:java}
> package org.apache.flink.table.gateway.api.validator;
> import org.apache.flink.table.gateway.api.validator.QueryValidator;
> import org.apache.flink.table.operations.Operation;
> import org.apache.flink.table.operations.ddl.CreateTableOperation;
> public class KafkaTimestampValidator implements QueryValidator {
> private static final long ONE_DAY = 24 * 60 * 60 * 1000L;
> @Override
> public boolean validateQuery(Operation op) {
> if (op instanceof CreateTableOperation) {
> CreateTableOperation createTableOp = (CreateTableOperation) op;
> String connector =
> createTableOp.getCatalogTable().getOptions().get("connector");
> if ("kafka".equals(connector)) {
> String startupTimestamp =
> createTableOp.getCatalogTable().getOptions().get("scan.startup.timestamp-millis");
> if (startupTimestamp != null &&
> Long.parseLong(startupTimestamp) < System.currentTimeMillis() - ONE_DAY) {
> return false;
> }
> }
> }
> return true;
> }
> }{code}
> I'd be happy to implement this feature, if we can reach on agreement.
> Thanks.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
