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"

Reviewed By: singcha

Test Plan:
`ensure_consistent_read` is set to false for Select and Explain queries, and to true for all other type of queries

```
select * from test_adheer_delete limit 100
```

```
headers: [{thrift_metastore_client_info={
  "project" : "presto",
  "max_num_partitions_to_be_cached" : "1000",
  **"ensure_consistent_read" : "false",**
  "dw.jobId" : "20251027_045135_00005_ig79c",
  "dw.owner" : "prestoui",
  "dw.pipeline" : "presto-js-client"
}}]
```


```
CREATE TABLE test_adheer_delete (col1 bigint, col2 bigint) WITH (oncall = 'presto_batch', partitioned_by = ARRAY['col2'])
```

```
headers: [{thrift_metastore_client_info={
  "project" : "presto",
  "max_num_partitions_to_be_cached" : "1000",
  **"ensure_consistent_read" : "true",**
  "dw.jobId" : "20251027_043305_00004_ig79c",
  "dw.owner" : "prestoui",
  "dw.pipeline" : "presto-js-client"
}}]
```

built presto version `0.296-20251028.203517-312`
deployed on test cluster `atn1_adhoctest_t6_ec1`
set the session property `SET SESSION prism.force_metastore_primary_enabled=true;`

ran the query `select * from presto_query_statistics limit 1;` with the query id `20251028_234254_00028_yqy68`
validated the headers with the query `select headers, * from hive_metsatore_logger_logs_inc_archive where ds = '2025-10-28' and dw_environment = 'atn1_adhoctest_t6_ec1' and dw_job_id = '20251028_234254_00028_yqy68'`
{F1983081638}
can see `ensure_consistent_read: true`

ran the same query but without setting the session property, and validated the headers
{F1983082149}
can see `ensure_consistent_read: false`

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)
         {