dongwoo.kim created FLINK-35560: ----------------------------------- Summary: Add query validator support to flink sql gateway via spi pattern Key: FLINK-35560 URL: https://issues.apache.org/jira/browse/FLINK-35560 Project: Flink Issue Type: Improvement Components: Table SQL / Gateway Reporter: dongwoo.kim
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:sql} 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 h4. -- This message was sent by Atlassian Jira (v8.20.10#820010)