feat: Send "ensureConsistentRead" to Metastore API for all SELECT queries (#26474)

Summary:
X-link: facebookexternal/presto-facebook#3436


X-link: facebookexternal/presto-facebook#3436

feat: Send "ensureConsistentRead" to Metastore API for all SELECT queries

BREAKING CHANGE: Default metastore replica to be read from for Select queries will be the read replica, and re-after-write users will have to set "prism.force_metastore_primary_enabled" session property to "true"

Differential Revision: D84520611

Pulled By: adheer-araokar

Author: adheer-araokar

presto-hive/src/test/java/com/facebook/presto/hive/AbstractTestHiveClient.java

@@ -29,6 +29,7 @@
 import com.facebook.presto.common.predicate.Range;
 import com.facebook.presto.common.predicate.TupleDomain;
 import com.facebook.presto.common.predicate.ValueSet;
+import com.facebook.presto.common.resourceGroups.QueryType;
 import com.facebook.presto.common.type.ArrayType;
 import com.facebook.presto.common.type.MapType;
 import com.facebook.presto.common.type.NamedTypeSignature;
@@ -1250,6 +1251,12 @@ public RuntimeStats getRuntimeStats()
                 return session.getRuntimeStats();
             }
 
+            @Override
+            public Optional<QueryType> getQueryType()
+            {
+                return session.getQueryType();
+            }
+
             @Override
             public ConnectorSession forConnectorId(ConnectorId connectorId)
             {

presto-main-base/src/main/java/com/facebook/presto/FullConnectorSession.java

@@ -15,6 +15,7 @@
 
 import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.common.function.SqlFunctionProperties;
+import com.facebook.presto.common.resourceGroups.QueryType;
 import com.facebook.presto.common.type.TimeZoneKey;
 import com.facebook.presto.metadata.SessionPropertyManager;
 import com.facebook.presto.spi.ConnectorId;
@@ -309,6 +310,12 @@ public RuntimeStats getRuntimeStats()
         return runtimeStats;
     }
 
+    @Override
+    public Optional<QueryType> getQueryType()
+    {
+        return session.getQueryType();
+    }
+
     @Override
     public ConnectorSession forConnectorId(ConnectorId connectorId)
     {

presto-main-base/src/main/java/com/facebook/presto/testing/TestingConnectorSession.java

@@ -16,6 +16,7 @@
 import com.facebook.presto.FullConnectorSession;
 import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.common.function.SqlFunctionProperties;
+import com.facebook.presto.common.resourceGroups.QueryType;
 import com.facebook.presto.common.type.TimeZoneKey;
 import com.facebook.presto.execution.QueryIdGenerator;
 import com.facebook.presto.spi.ConnectorId;
@@ -228,6 +229,12 @@ public RuntimeStats getRuntimeStats()
         return new RuntimeStats();
     }
 
+    @Override
+    public Optional<QueryType> getQueryType()
+    {
+        return Optional.of(QueryType.SELECT);
+    }
+
     @Override
     public ConnectorSession forConnectorId(ConnectorId connectorId)
     {

presto-spi/src/main/java/com/facebook/presto/spi/ConnectorSession.java

@@ -15,6 +15,7 @@
 
 import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.common.function.SqlFunctionProperties;
+import com.facebook.presto.common.resourceGroups.QueryType;
 import com.facebook.presto.common.type.TimeZoneKey;
 import com.facebook.presto.spi.function.SqlFunctionId;
 import com.facebook.presto.spi.function.SqlInvokedFunction;
@@ -67,6 +68,8 @@ default boolean isReadConstraints()
 
     RuntimeStats getRuntimeStats();
 
+    Optional<QueryType> getQueryType();
+
     /**
      * returns a ConnectorSession for a specific ConnectorId
      * @return

presto-spi/src/test/java/com/facebook/presto/spi/TestingSession.java

@@ -15,6 +15,7 @@
 
 import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.common.function.SqlFunctionProperties;
+import com.facebook.presto.common.resourceGroups.QueryType;
 import com.facebook.presto.common.type.TimeZoneKey;
 import com.facebook.presto.spi.function.SqlFunctionId;
 import com.facebook.presto.spi.function.SqlInvokedFunction;
@@ -131,6 +132,12 @@ public RuntimeStats getRuntimeStats()
             return new RuntimeStats();
         }
 
+        @Override
+        public Optional<QueryType> getQueryType()
+        {
+            return Optional.of(QueryType.SELECT);
+        }
+
         @Override
         public ConnectorSession forConnectorId(ConnectorId connectorId)
         {