Gabriel39 · Nov 12, 2026
@@ -98,9 +98,36 @@ find/getMore batches. Projection is materialized locally, and every SQL filter,
 LIKE/REGEXP, join, aggregation, sort and LIMIT remains a Doris operation. Range
 splitting, predicate/LIMIT pushdown and shared snapshots are separate work.
 
-`batch_size` and `query_timeout` are session variables. URI read preference settings
-take precedence over the catalog mode, retaining ordered tag sets and maximum staleness
-in both drivers. Without URI routing settings, `mongodb.read_preference` supplies the mode.
+`batch_size` and `query_timeout` are session variables. The following session variables
+also support statement-level `SET_VAR` hints; values are case-insensitive:
+
+| Session variable | Default | Supported values |
+|---|---|---|
+| `mongodb_read_preference` | `inherit` | `inherit`, `primary`, `primaryPreferred`, `secondary`, `secondaryPreferred`, `nearest` |
+| `mongodb_read_concern` | `inherit` | `inherit`, `local`, `majority`, `available` |
+
+`inherit` uses the catalog's effective policy: URI `readPreference` / `readConcernLevel`
+first, then `mongodb.read_preference` / `mongodb.read_concern`. An explicit session or
+statement value overrides that policy for data reads. Each scan pins its effective
+values without changing the catalog, metadata discovery, or other sessions.
+URI routing criteria (ordered tag sets, maximum staleness and hedge options) remain
+in effect when overriding the mode. A `primary` override with incompatible URI
+criteria fails at planning instead of silently dropping those criteria.
+
+```sql
+SET mongodb_read_preference = 'secondaryPreferred';
+SET mongodb_read_concern = 'local';
+SELECT /*+ SET_VAR(mongodb_read_preference='primary', mongodb_read_concern='majority') */
+    count(*) FROM mongo_native.sample.records;
+SET mongodb_read_preference = 'inherit';
+SET mongodb_read_concern = 'inherit';
+```
+
+Selecting a secondary can return stale data. `primaryPreferred` and
+`secondaryPreferred` permit fallback to the other role; `secondary` requires a
+secondary in a replica set. A read concern controls read visibility independently
+of node selection and does not establish a snapshot across Doris scans.
+
 Catalog connection defaults:
 
 | Property | Default |
@@ -131,8 +158,9 @@ bash thirdparty/build-thirdparty.sh mongo_c_driver
 
 FE unit tests reside in this module, BSON SerDe unit tests in
 `be/test/core/data_type_serde/data_type_serde_bson_test.cpp`, and reader lifecycle tests
-in `be/test/format_v2/table/mongodb_reader_test.cpp`. The regression suite
-`external_table_p0/mongodb/test_mongodb_native.groovy` uses the disposable MongoDB
+in `be/test/format_v2/table/mongodb_reader_test.cpp`. The regression suites
+`external_table_p0/mongodb/test_mongodb_native.groovy` and
+`external_table_p0/mongodb/test_mongodb_read_policy.groovy` use the disposable MongoDB
 fixture under `docker/thirdparties/docker-compose/mongodb/`, enabled with
 `enableMongoDBTest`, `externalEnvIp`, `mongodb_port`, `mongodb_user`, and
 `mongodb_password` regression settings.
@@ -26,6 +26,7 @@
 import com.mongodb.ReadConcern;
 import com.mongodb.ReadConcernLevel;
 import com.mongodb.ReadPreference;
+import org.bson.BsonDocument;
 import org.bson.Document;
 
 import java.nio.charset.StandardCharsets;
@@ -144,6 +145,38 @@ private static void require(boolean condition, String property) {
         }
     }
 
+    String resolveReadPreference(String policy) {
+        if ("inherit".equalsIgnoreCase(policy)) {
+            return readPreference;
+        }
+        String mode;
+        try {
+            mode = ReadPreference.valueOf(policy).getName();
+        } catch (IllegalArgumentException e) {
+            throw new IllegalArgumentException("Invalid mongodb_read_preference");
+        }
+        if ("primary".equals(mode) && connectionString.getReadPreference() != null) {
+            BsonDocument inherited = connectionString.getReadPreference().toDocument();
+            // BE preserves URI routing criteria when applying the statement's mode. Reject
+            // incompatible overrides here rather than silently dropping the catalog's routing.
+            if ((inherited.containsKey("tags") && !inherited.getArray("tags").isEmpty())
+                    || inherited.containsKey("maxStalenessSeconds") || inherited.containsKey("hedge")) {
+                throw new IllegalArgumentException("mongodb_read_preference=primary is incompatible with URI "
+                        + "read preference tags, max staleness or hedge options");
+            }
+        }
+        return mode;
+    }
+
+    String resolveReadConcern(String policy) {
+        if ("inherit".equalsIgnoreCase(policy)) {
+            return readConcern;
+        }
+        String level = policy.toLowerCase(Locale.ROOT);
+        require(Set.of("local", "majority", "available").contains(level), "mongodb_read_concern");
+        return level;
+    }
+
     MongoClientSettings clientSettings() {
         MongoClientSettings.Builder builder = MongoClientSettings.builder().applyConnectionString(connectionString)
                 .applyToSocketSettings(s -> s.connectTimeout(connectTimeoutMs, TimeUnit.MILLISECONDS)

@@ -60,10 +60,12 @@ public List<ConnectorScanRange> planScan(ConnectorSession session, ConnectorScan
         d.setUsername(properties.getUsername());
         d.setPassword(properties.getPassword());
         d.setAuthDatabase(properties.getAuthDatabase());
-        d.setReadPreference(properties.getReadPreference());
-        d.setReadConcern(properties.getReadConcern());
         d.setTypeMismatchAsNull("null".equals(properties.getTypeMismatch()));
         Map<String, String> policies = session.getSessionProperties();
+        // Pin effective policies in the range; never mutate the catalog or its shared metadata client.
+        d.setReadPreference(properties.resolveReadPreference(
+                policies.getOrDefault("mongodb_read_preference", "inherit")));
+        d.setReadConcern(properties.resolveReadConcern(policies.getOrDefault("mongodb_read_concern", "inherit")));
         d.setBatchSize(bounded(policies, "batch_size", 1024, 1, 65536));
         d.setTimeoutMs(1000L * bounded(policies, "query_timeout", 300, 1, Integer.MAX_VALUE / 1000));
         d.setConnectTimeoutMs(properties.getConnectTimeoutMs());

@@ -22,6 +22,7 @@
 import org.apache.doris.connector.spi.scan.ConnectorScanRequest;
 import org.apache.doris.connector.spi.scan.ScanNodePropertiesResult;
 import org.apache.doris.thrift.TFileRangeDesc;
+import org.apache.doris.thrift.TMongoScanDesc;
 import org.apache.doris.thrift.TTableFormatFileDesc;
 
 import org.bson.Document;
@@ -34,6 +35,16 @@
 
 class MongoScanPlanTest {
     private static final class Session implements ConnectorSession {
+        private final Map<String, String> policies;
+
+        Session() {
+            this(Map.of("batch_size", "128", "query_timeout", "30"));
+        }
+
+        Session(Map<String, String> policies) {
+            this.policies = policies;
+        }
+
         @Override
         public String getQueryId() {
             return "fixture-query";
@@ -76,7 +87,7 @@ public Map<String, String> getCatalogProperties() {
 
         @Override
         public Map<String, String> getSessionProperties() {
-            return Map.of("batch_size", "128", "query_timeout", "30");
+            return policies;
         }
     }
 
@@ -85,6 +96,89 @@ private static MongoCatalogProperties properties() {
                 "mongodb.username", "fixture-user", "mongodb.password", "fixture-secret"));
     }
 
+    private static TMongoScanDesc plan(MongoCatalogProperties properties, Map<String, String> policies) {
+        MongoTableHandle handle = new MongoTableHandle("sample",
+                new Document("name", "records").append("type", "collection"), properties);
+        ConnectorScanRange range = new MongoScanPlanProvider(properties).planScan(new Session(policies),
+                ConnectorScanRequest.builder(handle, List.of()).build()).get(0);
+        TTableFormatFileDesc descriptor = new TTableFormatFileDesc();
+        range.populateRangeParams(descriptor, new TFileRangeDesc());
+        return descriptor.getMongodbParams();
+    }
+
+    @Test
+    void queryPoliciesOverrideCatalogAndRemainPinnedToTheirScan() {
+        MongoCatalogProperties properties = properties();
+        TMongoScanDesc overridden = plan(properties, Map.of("mongodb_read_preference", "SECONDARYPREFERRED",
+                "mongodb_read_concern", "LOCAL"));
+        Assertions.assertEquals("secondaryPreferred", overridden.getReadPreference());
+        Assertions.assertEquals("local", overridden.getReadConcern());
+        TMongoScanDesc inherited = plan(properties, Map.of("mongodb_read_preference", "inherit",
+                "mongodb_read_concern", "inherit"));
+        Assertions.assertEquals("primary", inherited.getReadPreference());
+        Assertions.assertEquals("majority", inherited.getReadConcern());
+        Assertions.assertEquals("secondaryPreferred", overridden.getReadPreference());
+        Assertions.assertEquals(inherited.getConfigurationVersion(), overridden.getConfigurationVersion());
+    }
+
+    @Test
+    void inheritsUriPoliciesAndRejectsPrimaryWithSecondaryCriteria() {
+        MongoCatalogProperties properties = MongoCatalogProperties.of(Map.of("mongodb.uri",
+                "mongodb://localhost:27017/?readPreference=secondary&readPreferenceTags=zone:east"
+                        + "&maxStalenessSeconds=120&readConcernLevel=available"));
+        TMongoScanDesc inherited = plan(properties, Map.of());
+        Assertions.assertEquals("secondary", inherited.getReadPreference());
+        Assertions.assertEquals("available", inherited.getReadConcern());
+        TMongoScanDesc overridden = plan(properties, Map.of("mongodb_read_preference", "nearest",
+                "mongodb_read_concern", "majority"));
+        Assertions.assertEquals("nearest", overridden.getReadPreference());
+        Assertions.assertEquals(properties.getUri(), overridden.getUri());
+        Assertions.assertEquals("majority", overridden.getReadConcern());
+        Assertions.assertThrows(IllegalArgumentException.class,
+                () -> plan(properties, Map.of("mongodb_read_preference", "primary")));
+    }
+
+    @Test
+    void validatesEachInheritedRoutingConstraintWithoutChangingCatalogSettings() {
+        for (String suffix : List.of("&readPreferenceTags=zone:east", "&maxStalenessSeconds=120")) {
+            MongoCatalogProperties properties = MongoCatalogProperties.of(Map.of("mongodb.uri",
+                    "mongodb://localhost:27017/?readPreference=secondary" + suffix));
+            Assertions.assertThrows(IllegalArgumentException.class,
+                    () -> plan(properties, Map.of("mongodb_read_preference", "primary")));
+            Assertions.assertEquals("secondary", plan(properties, Map.of()).getReadPreference());
+        }
+        MongoCatalogProperties modeOnly = MongoCatalogProperties.of(Map.of("mongodb.uri",
+                "mongodb://localhost:27017/?readPreference=secondary&maxStalenessSeconds=-1"));
+        Assertions.assertEquals("primary",
+                plan(modeOnly, Map.of("mongodb_read_preference", "primary")).getReadPreference());
+        MongoCatalogProperties catalog = MongoCatalogProperties.of(Map.of("mongodb.uri", "mongodb://localhost:27017",
+                "mongodb.read_preference", "nearest", "mongodb.read_concern", "local"));
+        TMongoScanDesc inherited = plan(catalog, Map.of("mongodb_read_preference", "INHERIT",
+                "mongodb_read_concern", "INHERIT"));
+        Assertions.assertEquals("nearest", inherited.getReadPreference());
+        Assertions.assertEquals("local", inherited.getReadConcern());
+    }
+
+    @Test
+    void rejectsUnsupportedQueryPoliciesBeforeDispatch() {
+        for (String preference : List.of("", "invalid", "secondary_preferred")) {
+            Assertions.assertThrows(IllegalArgumentException.class,
+                    () -> plan(properties(), Map.of("mongodb_read_preference", preference)));
+        }
+        for (String concern : List.of("", "invalid", "snapshot", "linearizable")) {
+            Assertions.assertThrows(IllegalArgumentException.class,
+                    () -> plan(properties(), Map.of("mongodb_read_concern", concern)));
+        }
+        for (String preference : List.of("primary", "primaryPreferred", "secondary", "secondaryPreferred", "nearest")) {
+            Assertions.assertEquals(preference,
+                    plan(properties(), Map.of("mongodb_read_preference", preference)).getReadPreference());
+        }
+        for (String concern : List.of("local", "majority", "available")) {
+            Assertions.assertEquals(concern,
+                    plan(properties(), Map.of("mongodb_read_concern", concern)).getReadConcern());
+        }
+    }
+
     @Test
     void emitsOneStableNativeRangeForEachLogicalObjectKind() {
         MongoCatalogProperties properties = properties();

@@ -3086,6 +3086,22 @@ public void checkFetchSplitsMaxWaitTime(String value) throws Exception {
             description = "enable count(*) pushdown optimization for external table")
     private boolean enableCountPushDownForExternalTable = true;
 
+    public static final String MONGODB_READ_PREFERENCE = "mongodb_read_preference";
+    public static final String MONGODB_READ_CONCERN = "mongodb_read_concern";
+
+    @VarAttrDef.VarAttr(name = MONGODB_READ_PREFERENCE, needForward = true,
+            checker = "checkMongoDBReadPreference",
+            options = {"inherit", "primary", "primaryPreferred", "secondary", "secondaryPreferred", "nearest"},
+            description = "MongoDB data-read routing for this session or SET_VAR statement. "
+                    + "inherit follows the catalog policy, including URI routing criteria")
+    public String mongodbReadPreference = "inherit";
+
+    @VarAttrDef.VarAttr(name = MONGODB_READ_CONCERN, needForward = true,
+            checker = "checkMongoDBReadConcern", options = {"inherit", "local", "majority", "available"},
+            description = "MongoDB data-read visibility for this session or SET_VAR statement. "
+                    + "inherit follows the catalog policy; shared snapshots are not supported")
+    public String mongodbReadConcern = "inherit";
+
     public static final String FLUSS_UNION_READ_MODE = "fluss_union_read_mode";
 
     @VarAttrDef.VarAttr(name = FLUSS_UNION_READ_MODE,
@@ -4558,6 +4574,21 @@ public void checkBackendSelectionMode(String backendSelectionMode) {
         }
     }
 
+    public void checkMongoDBReadPreference(String value) {
+        if (!Set.of("inherit", "primary", "primarypreferred", "secondary", "secondarypreferred", "nearest")
+                .contains(value.toLowerCase(Locale.ROOT))) {
+            throw new UnsupportedOperationException("Invalid mongodb_read_preference; expected inherit, primary, "
+                    + "primaryPreferred, secondary, secondaryPreferred or nearest");
+        }
+    }
+
+    public void checkMongoDBReadConcern(String value) {
+        if (!Set.of("inherit", "local", "majority", "available").contains(value.toLowerCase(Locale.ROOT))) {
+            throw new UnsupportedOperationException("Invalid mongodb_read_concern; expected inherit, local, "
+                    + "majority or available");
+        }
+    }
+
     public void checkFlussUnionReadMode(String mode) {
         String normalized = Strings.nullToEmpty(mode).trim().toLowerCase(Locale.ROOT);
         if (!normalized.isEmpty() && !"auto".equals(normalized)

diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariableMongoReadPolicyTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariableMongoReadPolicyTest.java
@@ -0,0 +1,83 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.qe;
+
+import org.apache.doris.analysis.SetType;
+import org.apache.doris.analysis.SetVar;
+import org.apache.doris.analysis.StringLiteral;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.connector.ConnectorSessionBuilder;
+import org.apache.doris.connector.spi.ConnectorSession;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+class SessionVariableMongoReadPolicyTest {
+    private static void set(SessionVariable variables, String name, String value) throws DdlException {
+        VariableMgr.setVar(variables, new SetVar(SetType.SESSION, name, new StringLiteral(value)));
+    }
+
+    @Test
+    void defaultsInheritAndConnectorSnapshotsAreSessionIsolated() throws Exception {
+        ConnectContext context = new ConnectContext();
+        SessionVariable variables = context.getSessionVariable();
+        ConnectorSession inherited = ConnectorSessionBuilder.from(context).build();
+        Assertions.assertEquals("inherit", inherited.getSessionProperties().get("mongodb_read_preference"));
+        Assertions.assertEquals("inherit", inherited.getSessionProperties().get("mongodb_read_concern"));
+        set(variables, "mongodb_read_preference", "secondaryPreferred");
+        set(variables, "mongodb_read_concern", "local");
+        ConnectorSession overridden = ConnectorSessionBuilder.from(context).build();
+        Assertions.assertEquals("secondaryPreferred", overridden.getSessionProperties().get("mongodb_read_preference"));
+        Assertions.assertEquals("local", overridden.getSessionProperties().get("mongodb_read_concern"));
+        Assertions.assertEquals("inherit", inherited.getSessionProperties().get("mongodb_read_preference"));
+        Assertions.assertEquals("inherit", VariableMgr.toMap(new SessionVariable()).get("mongodb_read_concern"));
+    }
+
+    @Test
+    void acceptsSupportedValuesAndForwardsPolicies() throws Exception {
+        SessionVariable variables = new SessionVariable();
+        for (String value : List.of("inherit", "primary", "primaryPreferred", "secondary", "secondaryPreferred",
+                "nearest", "SECONDARY", "INHERIT")) {
+            set(variables, "mongodb_read_preference", value);
+            Assertions.assertEquals(value, VariableMgr.toMap(variables).get("mongodb_read_preference"));
+        }
+        for (String value : List.of("inherit", "local", "majority", "available", "LOCAL", "INHERIT")) {
+            set(variables, "mongodb_read_concern", value);
+            Assertions.assertEquals(value, VariableMgr.toMap(variables).get("mongodb_read_concern"));
+        }
+        Map<String, String> forwarded = variables.getForwardVariables();
+        Assertions.assertEquals("INHERIT", forwarded.get("mongodb_read_preference"));
+        Assertions.assertEquals("INHERIT", forwarded.get("mongodb_read_concern"));
+    }
+
+    @Test
+    void rejectsUnsupportedValuesWithoutChangingSession() {
+        SessionVariable variables = new SessionVariable();
+        for (String value : List.of("", "invalid", "secondary_preferred", " primary")) {
+            Assertions.assertThrows(DdlException.class, () -> set(variables, "mongodb_read_preference", value));
+        }
+        for (String value : List.of("", "snapshot", "linearizable", "unknown", " local")) {
+            Assertions.assertThrows(DdlException.class, () -> set(variables, "mongodb_read_concern", value));
+        }
+        Assertions.assertEquals("inherit", VariableMgr.toMap(variables).get("mongodb_read_preference"));
+        Assertions.assertEquals("inherit", VariableMgr.toMap(variables).get("mongodb_read_concern"));
+    }
+}