diff --git a/core/trino-main/src/main/java/io/trino/SystemSessionProperties.java b/core/trino-main/src/main/java/io/trino/SystemSessionProperties.java
index 59be671b8ce1..40e7f1b41625 100644
--- a/core/trino-main/src/main/java/io/trino/SystemSessionProperties.java
+++ b/core/trino-main/src/main/java/io/trino/SystemSessionProperties.java
@@ -219,6 +219,7 @@ public final class SystemSessionProperties
     public static final String IDLE_WRITER_MIN_DATA_SIZE_THRESHOLD = "idle_writer_min_data_size_threshold";
     public static final String CLOSE_IDLE_WRITERS_TRIGGER_DURATION = "close_idle_writers_trigger_duration";
     public static final String COLUMNAR_FILTER_EVALUATION_ENABLED = "columnar_filter_evaluation_enabled";
+    public static final String SPOOLING_PROTOCOL_ENABLED = "spooling_protocol_enabled";
 
     private final List<PropertyMetadata<?>> sessionProperties;
 
@@ -1128,6 +1129,11 @@ public SystemSessionProperties(
                         ALLOW_UNSAFE_PUSHDOWN,
                         "Allow pushing down expressions that may fail for some inputs",
                         optimizerConfig.isUnsafePushdownAllowed(),
+                        true),
+                booleanProperty(
+                        SPOOLING_PROTOCOL_ENABLED,
+                        "Enable client spooling protocol",
+                        true,
                         true));
     }
 
@@ -2018,6 +2024,11 @@ public static boolean isColumnarFilterEvaluationEnabled(Session session)
         return session.getSystemProperty(COLUMNAR_FILTER_EVALUATION_ENABLED, Boolean.class);
     }
 
+    public static boolean isSpoolingProtocolEnabled(Session session)
+    {
+        return session.getSystemProperty(SPOOLING_PROTOCOL_ENABLED, Boolean.class);
+    }
+
     public static boolean isUnsafePushdownAllowed(Session session)
     {
         return session.getSystemProperty(ALLOW_UNSAFE_PUSHDOWN, Boolean.class);
diff --git a/core/trino-main/src/main/java/io/trino/execution/QueryStateMachine.java b/core/trino-main/src/main/java/io/trino/execution/QueryStateMachine.java
index 4b94bbdd1fa7..83a754e92193 100644
--- a/core/trino-main/src/main/java/io/trino/execution/QueryStateMachine.java
+++ b/core/trino-main/src/main/java/io/trino/execution/QueryStateMachine.java
@@ -88,6 +88,7 @@
 import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
 import static io.airlift.units.DataSize.succinctBytes;
 import static io.trino.SystemSessionProperties.getRetryPolicy;
+import static io.trino.SystemSessionProperties.isSpoolingProtocolEnabled;
 import static io.trino.execution.BasicStageStats.EMPTY_STAGE_STATS;
 import static io.trino.execution.QueryState.DISPATCHING;
 import static io.trino.execution.QueryState.FAILED;
@@ -309,7 +310,7 @@ static QueryStateMachine beginWithTicker(
             session = session.withExchangeEncryption(serializeAesEncryptionKey(createRandomAesEncryptionKey()));
         }
 
-        if (!queryType.map(SELECT::equals).orElse(false)) {
+        if (!queryType.map(SELECT::equals).orElse(false) || !isSpoolingProtocolEnabled(session)) {
             session = session.withoutSpooling();
         }