Handle node.sql.read_only setting in SQLOperations

sql/src/main/java/io/crate/action/sql/SQLOperations.java

@@ -29,11 +29,13 @@
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import io.crate.analyze.Analysis;
+import io.crate.analyze.AnalyzedStatement;
 import io.crate.analyze.Analyzer;
 import io.crate.analyze.ParameterContext;
 import io.crate.analyze.symbol.Field;
 import io.crate.analyze.symbol.Symbols;
 import io.crate.concurrent.CompletionListener;
+import io.crate.exceptions.ReadOnlyException;
 import io.crate.executor.Executor;
 import io.crate.executor.TaskResult;
 import io.crate.planner.Plan;
@@ -48,13 +50,15 @@
 import org.elasticsearch.common.inject.Singleton;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.logging.Loggers;
+import org.elasticsearch.common.settings.Settings;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 import java.util.*;
 
 import static io.crate.action.sql.SQLBulkRequest.EMPTY_BULK_ARGS;
 import static io.crate.action.sql.SQLRequest.EMPTY_ARGS;
+import static io.crate.action.sql.TransportBaseSQLAction.NODE_READ_ONLY_SETTING;
 
 @Singleton
 public class SQLOperations {
@@ -64,14 +68,17 @@ public class SQLOperations {
     private final Analyzer analyzer;
     private final Planner planner;
     private final Provider<Executor> executorProvider;
+    private final boolean isReadOnly;
 
     @Inject
     public SQLOperations(Analyzer analyzer,
                          Planner planner,
-                         Provider<Executor> executorProvider) {
+                         Provider<Executor> executorProvider,
+                         Settings settings) {
         this.analyzer = analyzer;
         this.planner = planner;
         this.executorProvider = executorProvider;
+        this.isReadOnly = settings.getAsBoolean(NODE_READ_ONLY_SETTING, false);
     }
 
     public Session createSession(@Nullable String defaultSchema) {
@@ -153,6 +160,7 @@ public void simpleQuery(String query,
 
             Statement statement = SqlParser.createStatement(query);
             Analysis analysis = analyzer.analyze(statement, new ParameterContext(EMPTY_ARGS, EMPTY_BULK_ARGS, defaultSchema));
+            validateReadOnly();
             Plan plan = planner.plan(analysis, UUID.randomUUID(), 0, 0);
 
             ResultReceiver resultReceiver;
@@ -235,6 +243,7 @@ public List<Field> describe(char type, String portalOrStatement) {
         public void execute(String portalName, int maxRows, ResultReceiver rowReceiver) {
             LOGGER.debug("method=describe portalName={} maxRows={}", portalName, maxRows);
             checkError();
+            validateReadOnly();
 
             resultReceivers.add(rowReceiver);
             this.maxRows = maxRows;
@@ -319,6 +328,12 @@ private void applySessionSettings(Plan plan) {
             }
         }
 
+        private void validateReadOnly() {
+            if (analysis.analyzedStatement().isWriteOperation() && isReadOnly) {
+                throw new ReadOnlyException();
+            }
+        }
+
         private void cleanup() {
             analysis = null;
             bulkParams.clear();

sql/src/main/java/io/crate/analyze/SetAnalyzedStatement.java

@@ -55,6 +55,6 @@ public <C, R> R accept(AnalyzedStatementVisitor<C, R> analyzedStatementVisitor,
 
     @Override
     public boolean isWriteOperation() {
-        return true;
+        return SetStatement.Scope.GLOBAL.equals(scope);
     }
 }

sql/src/test/java/io/crate/integrationtests/PostgresITest.java

@@ -22,9 +22,12 @@
 
 package io.crate.integrationtests;
 
+import io.crate.action.sql.TransportBaseSQLAction;
+import io.crate.action.sql.TransportSQLAction;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.hamcrest.Matchers;
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -35,22 +38,29 @@
 
 import static org.hamcrest.core.Is.is;
 
-@ESIntegTestCase.ClusterScope(numDataNodes = 1, numClientNodes = 0)
+@ESIntegTestCase.ClusterScope(numDataNodes = 2, numClientNodes = 0)
 public class PostgresITest extends SQLTransportIntegrationTest {
 
     private static final String JDBC_POSTGRESQL_URL = "jdbc:postgresql://127.0.0.1:4242/";
+    private static final String JDBC_POSTGRESQL_URL_READ_ONLY = "jdbc:postgresql://127.0.0.1:4243/";
 
     @Rule
     public ExpectedException expectedException = ExpectedException.none();
 
     @Override
     protected Settings nodeSettings(int nodeOrdinal) {
-        return Settings.builder()
-            .put(super.nodeSettings(nodeOrdinal))
+        Settings.Builder builder = Settings.builder();
+        builder.put(super.nodeSettings(nodeOrdinal))
             .put("network.psql", true)
-            .put("psql.host", "127.0.0.1")
-            .put("psql.port", "4242")
-            .build();
+            .put("psql.host", "127.0.0.1");
+
+        if ((nodeOrdinal + 1) % 2 == 0) {
+            builder.put("psql.port", "4242");
+        } else {
+            builder.put(TransportBaseSQLAction.NODE_READ_ONLY_SETTING, true);
+            builder.put("psql.port", "4243");
+        }
+        return builder.build();
     }
 
     @Before
@@ -188,6 +198,17 @@ public void testCustomSchemaAndAnalyzerFailure() throws Exception {
         }
     }
 
+    @Test
+    public void testStatementReadOnlyFailure() throws Exception {
+        try (Connection conn = DriverManager.getConnection(JDBC_POSTGRESQL_URL_READ_ONLY)) {
+            conn.setAutoCommit(true);
+            PreparedStatement stmt = conn.prepareStatement("create table test(a integer)");
+            expectedException.expect(PSQLException.class);
+            expectedException.expectMessage("ERROR: Only read operations are allowed on this node");
+            stmt.executeQuery();
+        }
+    }
+
     @Test
     public void testErrorRecoveryFromErrorsOutsideSqlOperations() throws Exception {
         try (Connection conn = DriverManager.getConnection(JDBC_POSTGRESQL_URL)) {