Address review: rename the partial-result setting and fold the fallback into pushDownAggregate

The setting is user-facing behavior, not a Calcite internal, so move it from
plugins.calcite.* to plugins.query.partial_result.on_mapping_conflict.enabled
and drop the CALCITE_ prefix from the key.

Fold tryPartialResultAggregate into pushDownAggregate so the planner rule keeps
a single entry point. The fallback is now private and gated by an
allowPartialFallback flag, so re-entering on the narrowed scan attempts it at
most once.

Signed-off-by: Kai Huang <ahkcs@amazon.com>

Author: ahkcs

common/src/main/java/org/opensearch/sql/common/setting/Settings.java

@@ -44,11 +44,11 @@ public enum Key {
     CALCITE_PUSHDOWN_ROWCOUNT_ESTIMATION_FACTOR(
         "plugins.calcite.pushdown.rowcount.estimation.factor"),
     CALCITE_SUPPORT_ALL_JOIN_TYPES("plugins.calcite.all_join_types.allowed"),
-    CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT(
-        "plugins.calcite.partial_result.on_mapping_conflict.enabled"),
 
     /** Query Settings. */
     FIELD_TYPE_TOLERANCE("plugins.query.field_type_tolerance"),
+    PARTIAL_RESULT_ON_MAPPING_CONFLICT(
+        "plugins.query.partial_result.on_mapping_conflict.enabled"),
 
     /** Common Settings for SQL and PPL. */
     QUERY_MEMORY_LIMIT("plugins.query.memory_limit"),

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePartialResultOnMappingConflictIT.java

@@ -30,7 +30,7 @@
  * another collapses to text-without-keyword across the wildcard pattern, which defeats aggregation
  * pushdown and forces a per-shard document scan that opens a Point-In-Time context on every shard.
  *
- * <p>When {@code plugins.calcite.partial_result.on_mapping_conflict.enabled} is on, the aggregation
+ * <p>When {@code plugins.query.partial_result.on_mapping_conflict.enabled} is on, the aggregation
  * is instead pushed down over just the aggregatable (keyword) index subset — no PIT — and the
  * response carries a {@code PARTIAL_RESULT} warning naming the excluded index.
  */
@@ -304,7 +304,7 @@ private void setPartialResult(boolean enabled) throws IOException {
     updateClusterSettings(
         new ClusterSetting(
             "persistent",
-            Settings.Key.CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT.getKeyValue(),
+            Settings.Key.PARTIAL_RESULT_ON_MAPPING_CONFLICT.getKeyValue(),
             Boolean.toString(enabled)));
   }
 

opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/AggregateIndexScanRule.java

@@ -98,12 +98,6 @@ protected void apply(
       @Nullable LogicalProject project,
       CalciteLogicalIndexScan scan) {
     AbstractRelNode newRelNode = scan.pushDownAggregate(aggregate, project);
-    if (newRelNode == null) {
-      // Normal pushdown failed (e.g. a text/keyword mapping conflict across a wildcard pattern that
-      // would otherwise fall back to a per-shard scan and exhaust PIT contexts). If partial results
-      // are enabled, push the aggregation over the aggregatable index subset instead and warn.
-      newRelNode = scan.tryPartialResultAggregate(aggregate, project);
-    }
     if (newRelNode != null) {
       call.transformTo(newRelNode);
       PlanUtils.tryPruneRelNodes(call);

opensearch/src/main/java/org/opensearch/sql/opensearch/setting/OpenSearchSettings.java

@@ -187,9 +187,9 @@ public class OpenSearchSettings extends Settings {
           Setting.Property.NodeScope,
           Setting.Property.Dynamic);
 
-  public static final Setting<?> CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT_SETTING =
+  public static final Setting<?> PARTIAL_RESULT_ON_MAPPING_CONFLICT_SETTING =
       Setting.boolSetting(
-          Key.CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT.getKeyValue(),
+          Key.PARTIAL_RESULT_ON_MAPPING_CONFLICT.getKeyValue(),
           false,
           Setting.Property.NodeScope,
           Setting.Property.Dynamic);
@@ -493,9 +493,9 @@ public OpenSearchSettings(ClusterSettings clusterSettings) {
     register(
         settingBuilder,
         clusterSettings,
-        Key.CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT,
-        CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT_SETTING,
-        new Updater(Key.CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT));
+        Key.PARTIAL_RESULT_ON_MAPPING_CONFLICT,
+        PARTIAL_RESULT_ON_MAPPING_CONFLICT_SETTING,
+        new Updater(Key.PARTIAL_RESULT_ON_MAPPING_CONFLICT));
     register(
         settingBuilder,
         clusterSettings,
@@ -700,7 +700,7 @@ public static List<Setting<?>> pluginSettings() {
         .add(CALCITE_PUSHDOWN_ENABLED_SETTING)
         .add(CALCITE_PUSHDOWN_ROWCOUNT_ESTIMATION_FACTOR_SETTING)
         .add(CALCITE_SUPPORT_ALL_JOIN_TYPES_SETTING)
-        .add(CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT_SETTING)
+        .add(PARTIAL_RESULT_ON_MAPPING_CONFLICT_SETTING)
         .add(DEFAULT_PATTERN_METHOD_SETTING)
         .add(DEFAULT_PATTERN_MODE_SETTING)
         .add(DEFAULT_PATTERN_MAX_SAMPLE_COUNT_SETTING)

opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java

@@ -375,6 +375,15 @@ public CalciteLogicalIndexScan pushDownRareTop(Project project, RareTopDigest di
   }
 
   public AbstractRelNode pushDownAggregate(Aggregate aggregate, @Nullable Project project) {
+    return pushDownAggregate(aggregate, project, true);
+  }
+
+  /**
+   * @param allowPartialFallback whether a failure may fall back to the partial-result path. False
+   *     when re-entering from that path, so the fallback is attempted at most once.
+   */
+  private AbstractRelNode pushDownAggregate(
+      Aggregate aggregate, @Nullable Project project, boolean allowPartialFallback) {
     try {
       CalciteLogicalIndexScan newScan =
           new CalciteLogicalIndexScan(
@@ -400,7 +409,7 @@ public AbstractRelNode pushDownAggregate(Aggregate aggregate, @Nullable Project
         if (LOG.isDebugEnabled()) {
           LOG.debug("Cannot pushdown the aggregate due to bucket contains array (nested) type");
         }
-        return null;
+        return allowPartialFallback ? tryPartialResultAggregate(aggregate, project) : null;
       }
       int queryBucketSize = osIndex.getQueryBucketSize();
       boolean bucketNullable = !PPLHintUtils.ignoreNullBucket(aggregate);
@@ -431,7 +440,7 @@ public AbstractRelNode pushDownAggregate(Aggregate aggregate, @Nullable Project
         LOG.debug("Cannot pushdown the aggregate {}", aggregate, e);
       }
     }
-    return null;
+    return allowPartialFallback ? tryPartialResultAggregate(aggregate, project) : null;
   }
 
   /**
@@ -447,9 +456,10 @@ public AbstractRelNode pushDownAggregate(Aggregate aggregate, @Nullable Project
    * warning ({@link QueryContext#isWarningsSupported}). The partitioning decision lives in {@link
    * PartialResultAggregatePushdown}; this method owns the plan-time wiring (settings gate, mapping
    * lookup, narrowed-scan construction, warning emission). Returns {@code null} — leaving the
-   * caller to fall back — whenever partial mode does not apply.
+   * aggregate un-pushed, as before — whenever partial mode does not apply.
    */
-  public AbstractRelNode tryPartialResultAggregate(Aggregate aggregate, @Nullable Project project) {
+  private AbstractRelNode tryPartialResultAggregate(
+      Aggregate aggregate, @Nullable Project project) {
     if (!isPartialResultEnabled()) {
       return null;
     }
@@ -484,9 +494,11 @@ public AbstractRelNode tryPartialResultAggregate(Aggregate aggregate, @Nullable
               narrowedIndex,
               getRowType(),
               pushDownContext.cloneWithOsIndex(narrowedIndex));
-      AbstractRelNode pushed = narrowedScan.pushDownAggregate(aggregate, project);
+      // allowPartialFallback = false: the subset is already narrowed, so a second attempt would be
+      // redundant. Keeps the fallback strictly one-shot.
+      AbstractRelNode pushed = narrowedScan.pushDownAggregate(aggregate, project, false);
       if (pushed == null) {
-        return null; // narrowed subset still can't push down -> fall back
+        return null; // narrowed subset still can't push down -> leave un-pushed
       }
 
       CalcitePlanContext.addWarning(plan.warning());
@@ -510,9 +522,7 @@ private boolean isPartialResultEnabled() {
       return override;
     }
     return (Boolean)
-        osIndex
-            .getSettings()
-            .getSettingValue(Settings.Key.CALCITE_PARTIAL_RESULT_ON_MAPPING_CONFLICT);
+        osIndex.getSettings().getSettingValue(Settings.Key.PARTIAL_RESULT_ON_MAPPING_CONFLICT);
   }
 
   public AbstractRelNode pushDownLimit(LogicalSort sort, Integer limit, Integer offset) {