chore: remove RUN SCRIPT from the server implementation (MINOR)

ksql-parser/src/main/antlr4/io/confluent/ksql/parser/SqlBase.g4

@@ -66,7 +66,6 @@ statement
     | DROP TABLE (IF EXISTS)? sourceName (DELETE TOPIC)?                    #dropTable
     | DROP CONNECTOR identifier                                             #dropConnector
     | EXPLAIN  (statement | identifier)                                     #explain
-    | RUN SCRIPT STRING                                                     #runScript
     | CREATE TYPE identifier AS type                                        #registerType
     | DROP TYPE identifier                                                  #dropType
     ;

ksql-parser/src/main/java/io/confluent/ksql/parser/AstBuilder.java

@@ -109,7 +109,6 @@
 import io.confluent.ksql.parser.tree.RegisterType;
 import io.confluent.ksql.parser.tree.Relation;
 import io.confluent.ksql.parser.tree.ResultMaterialization;
-import io.confluent.ksql.parser.tree.RunScript;
 import io.confluent.ksql.parser.tree.Select;
 import io.confluent.ksql.parser.tree.SelectItem;
 import io.confluent.ksql.parser.tree.SetProperty;
@@ -571,11 +570,6 @@ public Node visitSelectSingle(final SqlBaseParser.SelectSingleContext context) {
       }
     }
 
-    @Override
-    public Node visitRunScript(final SqlBaseParser.RunScriptContext context) {
-      return new RunScript(getLocation(context));
-    }
-
     @Override
     public Node visitListTopics(final SqlBaseParser.ListTopicsContext context) {
       return new ListTopics(getLocation(context), context.EXTENDED() != null);

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/InteractiveStatementExecutor.java

@@ -25,7 +25,6 @@
 import io.confluent.ksql.parser.tree.CreateTableAsSelect;
 import io.confluent.ksql.parser.tree.ExecutableDdlStatement;
 import io.confluent.ksql.parser.tree.InsertInto;
-import io.confluent.ksql.parser.tree.RunScript;
 import io.confluent.ksql.parser.tree.TerminateQuery;
 import io.confluent.ksql.planner.plan.ConfiguredKsqlPlan;
 import io.confluent.ksql.query.QueryId;
@@ -244,8 +243,6 @@ private void executeStatement(
     } else if (statement.getStatement() instanceof TerminateQuery) {
       terminateQuery((PreparedStatement<TerminateQuery>) statement);
       successMessage = "Query terminated.";
-    } else if (statement.getStatement() instanceof RunScript) {
-      handleLegacyRunScript(command, mode);
     } else {
       throw new KsqlException(String.format(
           "Unexpected statement type: %s",

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/execution/RequestHandler.java

@@ -15,20 +15,16 @@
 
 package io.confluent.ksql.rest.server.execution;
 
-import com.google.common.collect.Iterables;
 import io.confluent.ksql.engine.KsqlEngine;
 import io.confluent.ksql.parser.KsqlParser.ParsedStatement;
 import io.confluent.ksql.parser.KsqlParser.PreparedStatement;
-import io.confluent.ksql.parser.tree.RunScript;
 import io.confluent.ksql.parser.tree.Statement;
 import io.confluent.ksql.rest.entity.KsqlEntity;
 import io.confluent.ksql.rest.entity.KsqlEntityList;
 import io.confluent.ksql.rest.server.computation.DistributingExecutor;
 import io.confluent.ksql.services.ServiceContext;
 import io.confluent.ksql.statement.ConfiguredStatement;
 import io.confluent.ksql.util.KsqlConfig;
-import io.confluent.ksql.util.KsqlConstants;
-import io.confluent.ksql.util.KsqlStatementException;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -79,29 +75,16 @@ public KsqlEntityList execute(
     final KsqlEntityList entities = new KsqlEntityList();
     for (ParsedStatement parsed : statements) {
       final PreparedStatement<?> prepared = ksqlEngine.prepare(parsed);
-      if (prepared.getStatement() instanceof RunScript) {
-        final KsqlEntityList result = executeRunScript(
-            serviceContext,
-            prepared,
-            propertyOverrides
-        );
-        if (!result.isEmpty()) {
-          // This is to maintain backwards compatibility until we deprecate
-          // RunScript in the next major release - the expected behavior was
-          // to return only the last entity
-          entities.add(Iterables.getLast(result));
-        }
-      } else {
-        final ConfiguredStatement<?> configured = ConfiguredStatement.of(
-            prepared, scopedPropertyOverrides, ksqlConfig);
-        executeStatement(
-            serviceContext,
-            configured,
-            parsed,
-            scopedPropertyOverrides,
-            entities
-        ).ifPresent(entities::add);
-      }
+      final ConfiguredStatement<?> configured = ConfiguredStatement.of(
+          prepared, scopedPropertyOverrides, ksqlConfig);
+
+      executeStatement(
+          serviceContext,
+          configured,
+          parsed,
+          scopedPropertyOverrides,
+          entities
+      ).ifPresent(entities::add);
     }
     return entities;
   }
@@ -131,23 +114,4 @@ private <T extends Statement> Optional<KsqlEntity> executeStatement(
     );
   }
 
-  private KsqlEntityList executeRunScript(
-      final ServiceContext serviceContext,
-      final PreparedStatement<?> statement,
-      final Map<String, Object> propertyOverrides
-  ) {
-    final String sql = (String) propertyOverrides
-        .get(KsqlConstants.LEGACY_RUN_SCRIPT_STATEMENTS_CONTENT);
-
-    if (sql == null) {
-      throw new KsqlStatementException(
-          "Request is missing script content", statement.getStatementText());
-    }
-
-    return execute(
-        serviceContext,
-        ksqlEngine.parse(sql),
-        propertyOverrides
-    );
-  }
 }

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/validation/RequestValidator.java

@@ -24,23 +24,19 @@
 import io.confluent.ksql.parser.KsqlParser.PreparedStatement;
 import io.confluent.ksql.parser.tree.CreateAsSelect;
 import io.confluent.ksql.parser.tree.InsertInto;
-import io.confluent.ksql.parser.tree.RunScript;
 import io.confluent.ksql.parser.tree.Statement;
 import io.confluent.ksql.rest.util.QueryCapacityUtil;
 import io.confluent.ksql.services.ServiceContext;
 import io.confluent.ksql.statement.ConfiguredStatement;
 import io.confluent.ksql.statement.Injector;
 import io.confluent.ksql.util.KsqlConfig;
-import io.confluent.ksql.util.KsqlConstants;
 import io.confluent.ksql.util.KsqlException;
 import io.confluent.ksql.util.KsqlStatementException;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.function.BiFunction;
 import java.util.function.Function;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * Wraps an execution context and information about how to validate statements
@@ -49,8 +45,6 @@
  */
 public class RequestValidator {
 
-  private static final Logger LOG = LoggerFactory.getLogger(RequestValidator.class);
-
   private final Map<Class<? extends Statement>, StatementValidator<?>> customValidators;
   private final BiFunction<KsqlExecutionContext, ServiceContext, Injector> injectorFactory;
   private final Function<ServiceContext, KsqlExecutionContext> snapshotSupplier;
@@ -108,9 +102,8 @@ public int validate(
       final ConfiguredStatement<?> configured = ConfiguredStatement.of(
           prepared, scopedPropertyOverrides, ksqlConfig);
 
-      numPersistentQueries += (prepared.getStatement() instanceof RunScript)
-          ? validateRunScript(serviceContext, configured, scopedPropertyOverrides, ctx)
-          : validate(serviceContext, configured, scopedPropertyOverrides, ctx, injector);
+      numPersistentQueries +=
+          validate(serviceContext, configured, scopedPropertyOverrides, ctx, injector);
     }
 
     if (QueryCapacityUtil.exceedsPersistentQueryCapacity(ctx, ksqlConfig, numPersistentQueries)) {
@@ -154,24 +147,4 @@ private <T extends Statement> int validate(
     return (statement instanceof CreateAsSelect || statement instanceof InsertInto) ? 1 : 0;
   }
 
-  private int validateRunScript(
-      final ServiceContext serviceContext,
-      final ConfiguredStatement<?> statement,
-      final Map<String, Object> mutableScopedProperties,
-      final KsqlExecutionContext executionContext
-  ) {
-    final String sql = (String) statement.getOverrides()
-        .get(KsqlConstants.LEGACY_RUN_SCRIPT_STATEMENTS_CONTENT);
-
-    if (sql == null) {
-      throw new KsqlStatementException(
-          "Request is missing script content", statement.getStatementText());
-    }
-
-    LOG.warn("RUN SCRIPT statement detected. "
-        + "Note: RUN SCRIPT is deprecated and will be removed in the next major version. "
-        + "statement: " + statement.getStatementText());
-
-    return validate(serviceContext, executionContext.parse(sql), mutableScopedProperties, sql);
-  }
 }

ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/computation/InteractiveStatementExecutorTest.java

@@ -44,12 +44,9 @@
 import io.confluent.ksql.metastore.MetaStore;
 import io.confluent.ksql.metastore.MetaStoreImpl;
 import io.confluent.ksql.name.SourceName;
-import io.confluent.ksql.parser.KsqlParser.ParsedStatement;
 import io.confluent.ksql.parser.KsqlParser.PreparedStatement;
-import io.confluent.ksql.parser.SqlBaseParser.SingleStatementContext;
 import io.confluent.ksql.parser.tree.CreateStreamAsSelect;
 import io.confluent.ksql.parser.tree.DropStream;
-import io.confluent.ksql.parser.tree.RunScript;
 import io.confluent.ksql.parser.tree.Statement;
 import io.confluent.ksql.parser.tree.TerminateQuery;
 import io.confluent.ksql.planner.plan.ConfiguredKsqlPlan;
@@ -67,7 +64,6 @@
 import io.confluent.ksql.statement.ConfiguredStatement;
 import io.confluent.ksql.test.util.EmbeddedSingleNodeKafkaCluster;
 import io.confluent.ksql.util.KsqlConfig;
-import io.confluent.ksql.util.KsqlServerException;
 import io.confluent.ksql.util.KsqlStatementException;
 import io.confluent.ksql.util.Pair;
 import io.confluent.ksql.util.PersistentQueryMetadata;
@@ -461,30 +457,6 @@ public void shouldEnforceReferentialIntegrity() {
         CoreMatchers.equalTo(CommandStatus.Status.SUCCESS));
   }
 
-  @SuppressWarnings("unchecked")
-  private PersistentQueryMetadata mockReplayRunScript(
-      final String runScriptStatement,
-      final String queryStatement
-  ) {
-    final PersistentQueryMetadata mockQuery = mock(PersistentQueryMetadata.class);
-    final Statement mockRunScript = mock(RunScript.class);
-    final PreparedStatement<Statement> runScript =
-        PreparedStatement.of(queryStatement, mockRunScript);
-    when(mockParser.parseSingleStatement(runScriptStatement))
-        .thenReturn(runScript);
-    final ImmutableList<ParsedStatement> parsedStatements = ImmutableList
-        .of(ParsedStatement.of(queryStatement, mock(SingleStatementContext.class)));
-    final PreparedStatement<Statement> preparedStatement =
-        PreparedStatement.of(queryStatement, mock(Statement.class));
-    when(mockEngine.parse(eq(queryStatement))).thenReturn(parsedStatements);
-    when(mockEngine.prepare(parsedStatements.get(0)))
-        .thenReturn((PreparedStatement)preparedStatement);
-    when(mockEngine.execute(eq(serviceContext), eqConfiguredStatement(preparedStatement)))
-        .thenReturn(ExecuteResult.of(mockQuery));
-    when(mockEngine.getPersistentQueries()).thenReturn(ImmutableList.of());
-    return mockQuery;
-  }
-
   @Test
   public void shouldSkipStartWhenReplayingLog() {
     // Given:
@@ -567,55 +539,6 @@ public void shouldNotCascadeDropStreamCommand() {
     );
   }
 
-  @Test
-  public void shouldHandleLegacyRunScriptCommand() {
-    // Given:
-    final String runScriptStatement = "run script";
-    final String queryStatement = "a query";
-    final PersistentQueryMetadata mockQuery = mockReplayRunScript(runScriptStatement, queryStatement);
-
-    // When:
-    statementExecutorWithMocks.handleStatement(
-        new QueuedCommand(
-            new CommandId(CommandId.Type.STREAM, "RunScript", CommandId.Action.EXECUTE),
-            new Command(
-                runScriptStatement,
-                Collections.singletonMap("ksql.run.script.statements", queryStatement),
-                emptyMap()),
-            Optional.empty(),
-            0L
-        )
-    );
-
-    // Then:
-    verify(mockQuery, times(1)).start();
-  }
-
-  @Test
-  public void shouldRestoreLegacyRunScriptCommand() {
-    // Given:
-    final String runScriptStatement = "run script";
-    final String queryStatement = "a persistent query";
-    final PersistentQueryMetadata mockQuery = mockReplayRunScript(runScriptStatement, queryStatement);
-
-    // When:
-    statementExecutorWithMocks.handleRestore(
-        new QueuedCommand(
-            new CommandId(CommandId.Type.STREAM, "RunScript", CommandId.Action.EXECUTE),
-            new Command(
-                runScriptStatement,
-                Collections.singletonMap("ksql.run.script.statements", queryStatement),
-                emptyMap()
-            ),
-            Optional.empty(),
-            0L
-        )
-    );
-
-    // Then:
-    verify(mockQuery, times(0)).start();
-  }
-
   @Test
   public void shouldTerminateAll() {
     // Given: