opensearch-project · mch2 · May 16, 2026 · May 12, 2026 · May 13, 2026 · May 15, 2026
@@ -32,18 +32,8 @@ jobs:
         uses: dtolnay/rust-toolchain@stable
       - name: Install protobuf compiler
         run: sudo apt-get update && sudo apt-get install -y protobuf-compiler
-      - name: Check out SQL repo (mustang-ppl-integration)
-        uses: actions/checkout@v6
-        with:
-          repository: opensearch-project/sql
-          ref: feature/mustang-ppl-integration
-          path: sql
-      - name: Publish unified-query artifacts to maven local
-        working-directory: sql
-        continue-on-error: true
-        run: ./gradlew publishUnifiedQueryPublicationToMavenLocal
       - name: Run sandbox check
-        run: ./gradlew check -p sandbox -Dsandbox.enabled=true -Drepos.mavenLocal=true -PrustDebug
+        run: ./gradlew check -p sandbox -Dsandbox.enabled=true -PrustDebug
       - name: Upload test results
         if: always()
         uses: actions/upload-artifact@v4

@@ -68,18 +68,23 @@ public static SchemaPlus buildSchema(ClusterState clusterState) {
      *
      * <p>Type mapping:
      * <ul>
-     *   <li>keyword/text -> VARCHAR</li>
+     *   <li>keyword/text/match_only_text -> VARCHAR</li>
      *   <li>long -> BIGINT</li>
+     *   <li>unsigned_long -> BIGINT</li>
      *   <li>integer -> INTEGER</li>
      *   <li>short -> SMALLINT</li>
      *   <li>byte -> TINYINT</li>
      *   <li>double -> DOUBLE</li>
-     *   <li>float -> FLOAT</li>
+     *   <li>float -> REAL</li>
+     *   <li>half_float -> REAL</li>
+     *   <li>scaled_float -> BIGINT</li>
      *   <li>boolean -> BOOLEAN</li>
      *   <li>date -> TIMESTAMP</li>
-     *   <li>ip -> VARCHAR</li>
+     *   <li>date_nanos -> TIMESTAMP</li>
+     *   <li>ip -> VARBINARY</li>
+     *   <li>binary -> VARBINARY</li>
      *   <li>nested/object -> skip (not mapped)</li>
-     *   <li>unknown -> VARCHAR (default)</li>
+     *   <li>unknown -> throws IllegalArgumentException</li>
      * </ul>
      *
      * @param opensearchType the OpenSearch field type string
@@ -88,9 +93,15 @@ public static SqlTypeName mapFieldType(String opensearchType) {
         switch (opensearchType) {
             case "keyword":
             case "text":
-            case "ip":
+            case "match_only_text":
                 return SqlTypeName.VARCHAR;
             case "long":
+            case "unsigned_long":
+                // unsigned_long: values above 2^63 - 1 wrap into negatives because BIGINT is
+                // signed and Substrait has no unsigned integer types. Smaller values are safe.
+                // TODO: values above 2^63 - 1 wrap into negatives. Drop the UInt64 → Int64 narrowing
+                // (see schema_coerce.rs) when we have a proper solution.
+            case "scaled_float":
                 return SqlTypeName.BIGINT;
             case "integer":
                 return SqlTypeName.INTEGER;
@@ -101,13 +112,27 @@ public static SqlTypeName mapFieldType(String opensearchType) {
             case "double":
                 return SqlTypeName.DOUBLE;
             case "float":
-                return SqlTypeName.FLOAT;
+            case "half_float":
+                // half_float lands as Arrow Float16 on disk. Calcite has no fp16 type; widen to
+                // REAL so the planner sees the same shape as a regular float column. The parquet
+                // reader's SchemaAdapter casts Float16 → Float32 per batch.
+                // TODO: every record batch goes through a Float16 → Float32 cast (see
+                // schema_coerce.rs) and downstream operators see Float32. Drop the widening when
+                // we have a proper solution.
+                return SqlTypeName.REAL;
             case "boolean":
                 return SqlTypeName.BOOLEAN;
             case "date":
+            case "date_nanos":
                 return SqlTypeName.TIMESTAMP;
+            case "ip":
+            case "binary":
+                // TODO: differentiate ip and binary as separate UDTs instead of collapsing both
+                // to VARBINARY. With the type preserved, literals can be converted into the
+                // on-disk byte form the planner expects.
+                return SqlTypeName.VARBINARY;
             default:
-                return SqlTypeName.VARCHAR;
+                throw new IllegalArgumentException("Unsupported OpenSearch field type: " + opensearchType);
         }
     }
 

@@ -165,6 +165,9 @@ public enum ScalarFunction {
 
     EXTRACT(Category.SCALAR, SqlKind.EXTRACT),
 
+    // ── Conversion (placeholder UDFs rewritten by backend adapters) ──
+    BINARY(Category.SCALAR, SqlKind.OTHER_FUNCTION),
+
     // ── Datetime ────────────────────────────────────────────────────
     // fromSqlFunction resolves via valueOf(name.toUpperCase()), so the enum name IS
     // the wire contract. Aliases each need their own entry; the adapter map points

@@ -72,6 +72,9 @@ dependencies {
   // Substrait — Calcite RelNode to Substrait plan conversion for DataFusion native runtime
   implementation "io.substrait:isthmus:0.89.1"
   implementation "io.substrait:core:0.89.1"
+  // avatica ByteString is needed at compile time for IpBinaryFunctionAdapter's VARBINARY
+  // literal construction; runtime is provided by analytics-framework's runtimeOnly avatica.
+  compileOnly "org.apache.calcite.avatica:avatica-core:1.27.0"
   implementation "com.fasterxml.jackson.datatype:jackson-datatype-jdk8:${versions.jackson}"
   // jackson-datatype-jsr310 — added to arrow-flight-rpc (the parent plugin that bundles
   // arrow-vector). arrow-vector's JsonStringArrayList eagerly registers JavaTimeModule on

@@ -518,6 +518,7 @@ pub unsafe fn sql_to_substrait(
         let schema = listing_options
             .infer_schema(&ctx.state(), &table_path)
             .await?;
+        let schema = crate::schema_coerce::coerce_inferred_schema(schema);
         let config = ListingTableConfig::new(table_path)
             .with_listing_options(listing_options)
             .with_schema(schema);

@@ -156,6 +156,7 @@ pub async fn execute_indexed_query(
     let resolved_schema = listing_options
         .infer_schema(&ctx.state(), &shard_view.table_path)
         .await?;
+    let resolved_schema = crate::schema_coerce::coerce_inferred_schema(resolved_schema);
     let table_config = datafusion::datasource::listing::ListingTableConfig::new(shard_view.table_path.clone())
         .with_listing_options(listing_options)
         .with_schema(resolved_schema);

@@ -30,6 +30,7 @@ pub mod partition_stream;
 pub mod query_executor;
 pub mod query_tracker;
 pub mod runtime_manager;
+pub mod schema_coerce;
 pub mod session_context;
 pub mod statistics_cache;
 pub mod udf;

@@ -126,6 +126,7 @@ pub async fn execute_query(
             error!("Failed to infer schema: {}", e);
             e
         })?;
+    let resolved_schema = crate::schema_coerce::coerce_inferred_schema(resolved_schema);
 
     let table_config = ListingTableConfig::new(table_path)
         .with_listing_options(listing_options)