[analytics-engine] Keep binary in scan set; dedup strip type-list; preserve untouched bulk docs

binary maps to VARBINARY in the analytics-engine schema builder (same as ip)
and is a registered parquet field type, so it scans fine - drop it from the
unsupported-field strip set, which is now {nested, geo_point, geo_shape, alias}.
Stripping binary needlessly mutated fixtures and risked masking a real
binary-handling bug instead of letting the assertion surface it.

Make UNSUPPORTED_FIELD_TYPES the single source of truth (public) and have
AnalyticsUnsupportedFieldStripVerifyIT import it instead of re-listing, so the
stripper and the verifier cannot drift. Add a byte-identity guard to
stripBulkFields: a bulk doc line is only re-serialized when it actually carried
a dropped key, so untouched docs pass through verbatim.

Signed-off-by: Eric Wei <mengwei.eric@gmail.com>

Author: mengweieric

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

@@ -25,7 +25,7 @@
  *
  * <ol>
  *   <li><b>Create succeeded</b> — the index exists (a mapping that still contained
- *       nested/geo_point/geo_shape/binary/alias would have been rejected by the parquet/composite
+ *       nested/geo_point/geo_shape/alias would have been rejected by the parquet/composite
  *       store at PUT time, so existence proves the strip removed them).
  *   <li><b>Mapping is clean</b> — the live mapping pulled back from the cluster contains none of
  *       the unsupported types at any depth.
@@ -42,8 +42,10 @@
  */
 public class AnalyticsUnsupportedFieldStripVerifyIT extends PPLIntegTestCase {
 
+  // Single source of truth — the same set the load-path strip uses (TestUtils.AnalyticsIndexConfig).
+  // Importing it (rather than re-listing) guarantees the verifier and the stripper can't drift.
   private static final Set<String> UNSUPPORTED =
-      Set.of("nested", "geo_point", "geo_shape", "binary", "alias");
+      org.opensearch.sql.legacy.TestUtils.AnalyticsIndexConfig.UNSUPPORTED_FIELD_TYPES;
 
   /**
    * Field types the parquet/composite store also rejects but that are out of scope for the strip

integ-test/src/test/java/org/opensearch/sql/legacy/AnalyticsFieldStripTests.java

@@ -17,8 +17,9 @@
 
 /**
  * Pure-logic coverage for the analytics-engine field strip (no filesystem / cluster). Verifies that
- * nested/geo_point/geo_shape/binary/alias fields are removed from mappings and bulk data on the AE
- * route, and that everything is a no-op when the route is off.
+ * nested/geo_point/geo_shape/alias fields are removed from mappings and bulk data on the AE route,
+ * that a supported {@code binary} field is left intact, and that everything is a no-op when the
+ * route is off.
  */
 public class AnalyticsFieldStripTests {
 
@@ -37,6 +38,7 @@ private void enable() {
           + "\"nested_value\":{\"type\":\"nested\"},"
           + "\"geo_point_value\":{\"type\":\"geo_point\"},"
           + "\"geo_shape_value\":{\"type\":\"geo_shape\"},"
+          // binary is a SUPPORTED scan type (binary -> VARBINARY); it must be kept, not stripped.
           + "\"binary_value\":{\"type\":\"binary\"},"
           + "\"alias_value\":{\"type\":\"alias\",\"path\":\"keep_text\"},"
           + "\"obj\":{\"properties\":{"
@@ -51,15 +53,14 @@ public void mappingStrip_removesUnsupportedRecursively_andReportsTopLevel() {
     Set<String> dropped = AnalyticsIndexConfig.stripUnsupportedMappingFields(json);
 
     assertEquals(
-        Set.of("nested_value", "geo_point_value", "geo_shape_value", "binary_value", "alias_value"),
-        dropped);
+        Set.of("nested_value", "geo_point_value", "geo_shape_value", "alias_value"), dropped);
 
     JSONObject props = json.getJSONObject("mappings").getJSONObject("properties");
     assertTrue("supported scalar kept", props.has("keep_text"));
+    assertTrue("binary is a supported scan type — must be kept", props.has("binary_value"));
     assertFalse(props.has("nested_value"));
     assertFalse(props.has("geo_point_value"));
     assertFalse(props.has("geo_shape_value"));
-    assertFalse(props.has("binary_value"));
     assertFalse(props.has("alias_value"));
 
     // object field kept, but its unsupported sub-property stripped recursively
@@ -82,12 +83,13 @@ public void bulkStrip_removesDroppedKeysFromSourceLinesOnly() {
     enable();
     String bulk =
         "{\"index\":{\"_id\":\"1\"}}\n"
-            + "{\"keep_text\":\"x\",\"geo_point_value\":{\"lat\":1,\"lon\":2},\"binary_value\":\"AA==\"}\n"
+            + "{\"keep_text\":\"x\",\"geo_point_value\":{\"lat\":1,\"lon\":2},\"geo_shape_value\":\"POINT(1"
+            + " 2)\"}\n"
             + "{\"index\":{\"_id\":\"2\"}}\n"
             + "{\"keep_text\":\"y\",\"nested_value\":[{\"a\":1}]}\n";
     String out =
         AnalyticsIndexConfig.stripBulkFields(
-            bulk, Set.of("geo_point_value", "binary_value", "nested_value"));
+            bulk, Set.of("geo_point_value", "geo_shape_value", "nested_value"));
 
     String[] lines = out.split("\n");
     // action lines untouched
@@ -97,12 +99,33 @@ public void bulkStrip_removesDroppedKeysFromSourceLinesOnly() {
     JSONObject doc1 = new JSONObject(lines[1]);
     assertTrue(doc1.has("keep_text"));
     assertFalse(doc1.has("geo_point_value"));
-    assertFalse(doc1.has("binary_value"));
+    assertFalse(doc1.has("geo_shape_value"));
     JSONObject doc2 = new JSONObject(lines[3]);
     assertTrue(doc2.has("keep_text"));
     assertFalse(doc2.has("nested_value"));
   }
 
+  @Test
+  public void bulkStrip_leavesUntouchedSourceLinesByteForByte() {
+    enable();
+    // doc1 carries a dropped key (gets rewritten); doc2 does not (must pass through verbatim).
+    String docWithDrop = "{\"keep_text\":\"x\",\"nested_value\":[{\"a\":1}]}";
+    String docNoDrop = "{\"keep_text\":\"y\",\"age\":  30,\"z\":1}";
+    String bulk =
+        "{\"index\":{\"_id\":\"1\"}}\n"
+            + docWithDrop
+            + "\n"
+            + "{\"index\":{\"_id\":\"2\"}}\n"
+            + docNoDrop
+            + "\n";
+    String out = AnalyticsIndexConfig.stripBulkFields(bulk, Set.of("nested_value"));
+    String[] lines = out.split("\n", -1);
+    // The doc that had no dropped key is byte-for-byte identical (odd spacing/key order preserved).
+    assertEquals(docNoDrop, lines[3]);
+    // The doc that had a dropped key lost it.
+    assertFalse(new JSONObject(lines[1]).has("nested_value"));
+  }
+
   @Test
   public void bulkStrip_noopWhenDisabledOrEmptyDropSet() {
     String bulk = "{\"index\":{}}\n{\"geo_point_value\":{\"lat\":1}}\n";

integ-test/src/test/java/org/opensearch/sql/legacy/TestUtils.java

@@ -146,14 +146,34 @@ static String bulkLoadRefreshParam() {
     }
 
     /**
-     * Field types the analytics-engine (DataFusion) backend cannot read. Any property of one of
-     * these types is removed from the mapping (and the matching key from each bulk doc) before a
-     * test index is created/loaded on the analytics-engine route, so seeding a dataset that happens
-     * to use one of them doesn't fail the whole test. Applied uniformly to every dataset — no
-     * per-index variant files to author or keep in sync.
+     * Field types the analytics-engine (DataFusion) backend cannot read, and which therefore must be
+     * removed from a test index's mapping (and the matching key from each bulk doc) before the index
+     * is created/loaded on the analytics-engine route. Seeding a dataset that happens to use one of
+     * them would otherwise fail ingestion and surface as an unrelated {@code expected:<1> but
+     * was:<0>} downstream — so we strip uniformly at load time across every dataset, no per-index
+     * variant files to author or keep in sync.
+     *
+     * <p><b>This set is the single source of truth.</b> The strip triggers purely on a mapping's
+     * field <em>types</em> (not on which IT is running), so any IT — existing or newly added — that
+     * loads one of these datasets through {@link SQLIntegTestCase#loadIndex} is handled out of the
+     * box. {@code AnalyticsUnsupportedFieldStripVerifyIT} imports this same constant and proves no
+     * listed type survives in any live mapping, so the list cannot silently drift.
+     *
+     * <p><b>What is and isn't here, and why:</b>
+     *
+     * <ul>
+     *   <li>{@code nested} — entire subtree dropped (the engine has no nested-document support).
+     *   <li>{@code geo_point}, {@code geo_shape} — fall through the engine's type whitelist
+     *       ({@code OpenSearchSchemaBuilder.mapFieldType} default → null), so the column can't scan.
+     *   <li>{@code alias} — indirection the scan-row-type builder doesn't resolve.
+     *   <li><b>{@code binary} is intentionally NOT here</b> — the engine maps {@code binary →
+     *       VARBINARY} (same as {@code ip}, see {@code OpenSearchSchemaBuilder.mapFieldType}), so it
+     *       is a supported scan type. Stripping it would needlessly mutate fixtures and could mask a
+     *       real binary-handling bug rather than let the assertion surface it.
+     * </ul>
      */
-    private static final Set<String> UNSUPPORTED_FIELD_TYPES =
-        Set.of("nested", "geo_point", "geo_shape", "binary", "alias");
+    public static final Set<String> UNSUPPORTED_FIELD_TYPES =
+        Set.of("nested", "geo_point", "geo_shape", "alias");
 
     /**
      * Remove every {@link #UNSUPPORTED_FIELD_TYPES} property (recursively, including object
@@ -209,6 +229,11 @@ private static boolean removeIfUnsupported(JSONObject properties, String field)
      * alternates an action line ({@code {"index":{...}}}) with a source line; only source lines
      * (those without a bulk action key) are rewritten. No-op when disabled or {@code droppedFields}
      * is empty.
+     *
+     * <p>A source line is re-serialized <em>only when it actually carried one of the dropped
+     * keys</em>; every other line (action lines and docs that never had the field) is appended
+     * byte-for-byte unchanged. This keeps the payload as close to the on-disk fixture as possible —
+     * we don't reorder keys or reformat numbers on docs we aren't modifying.
      */
     static String stripBulkFields(String bulkBody, Set<String> droppedFields) {
       if (!isEnabled() || droppedFields.isEmpty()) {
@@ -224,10 +249,18 @@ static String stripBulkFields(String bulkBody, Set<String> droppedFields) {
           boolean isActionLine =
               doc.has("index") || doc.has("create") || doc.has("update") || doc.has("delete");
           if (!isActionLine) {
+            boolean removedAny = false;
             for (String f : droppedFields) {
-              doc.remove(f);
+              if (doc.has(f)) {
+                doc.remove(f);
+                removedAny = true;
+              }
+            }
+            // Only rewrite the line if we actually removed something; otherwise leave it verbatim
+            // so untouched docs stay byte-for-byte identical to the fixture.
+            if (removedAny) {
+              line = doc.toString();
             }
-            line = doc.toString();
           }
         }
         out.append(line);