From b5437732b174333a694b1dfc9d0247484278e971 Mon Sep 17 00:00:00 2001 From: Vinay Krishna Pudyodu Date: Tue, 12 May 2026 22:53:45 +0000 Subject: [PATCH 1/7] Added scan support for indexed fieldTypes Signed-off-by: Vinay Krishna Pudyodu --- .../schema/OpenSearchSchemaBuilder.java | 26 +- .../rust/src/api.rs | 1 + .../rust/src/indexed_executor.rs | 1 + .../rust/src/lib.rs | 1 + .../rust/src/query_executor.rs | 1 + .../rust/src/schema_coerce.rs | 206 ++++++++++ .../rust/src/session_context.rs | 3 + .../DataFusionAnalyticsBackendPlugin.java | 3 + .../engine/OpenSearchSchemaBuilderTests.java | 21 +- .../analytics/qa/FieldTypeCoverageIT.java | 360 ++++++++++++++++++ 10 files changed, 608 insertions(+), 15 deletions(-) create mode 100644 sandbox/plugins/analytics-backend-datafusion/rust/src/schema_coerce.rs create mode 100644 sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/FieldTypeCoverageIT.java diff --git a/sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/OpenSearchSchemaBuilder.java b/sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/OpenSearchSchemaBuilder.java index ff5dcff67b604..7cd3f5415d91a 100644 --- a/sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/OpenSearchSchemaBuilder.java +++ b/sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/OpenSearchSchemaBuilder.java @@ -68,18 +68,22 @@ public static SchemaPlus buildSchema(ClusterState clusterState) { * *

Type mapping: *

* * @param opensearchType the OpenSearch field type string @@ -88,9 +92,13 @@ 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. + case "scaled_float": return SqlTypeName.BIGINT; case "integer": return SqlTypeName.INTEGER; @@ -101,13 +109,17 @@ public static SqlTypeName mapFieldType(String opensearchType) { case "double": return SqlTypeName.DOUBLE; case "float": - return SqlTypeName.FLOAT; + return SqlTypeName.REAL; case "boolean": return SqlTypeName.BOOLEAN; case "date": + case "date_nanos": return SqlTypeName.TIMESTAMP; + case "ip": + case "binary": + return SqlTypeName.VARBINARY; default: - return SqlTypeName.VARCHAR; + throw new IllegalArgumentException("Unsupported OpenSearch field type: " + opensearchType); } } diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs index dcf8c6c523d55..c332582cbd771 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs @@ -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_for_substrait(schema); let config = ListingTableConfig::new(table_path) .with_listing_options(listing_options) .with_schema(schema); diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_executor.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_executor.rs index 4de9defb52814..f3d2d051f64a8 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_executor.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_executor.rs @@ -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_for_substrait(resolved_schema); let table_config = datafusion::datasource::listing::ListingTableConfig::new(shard_view.table_path.clone()) .with_listing_options(listing_options) .with_schema(resolved_schema); diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/lib.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/lib.rs index 6b2ba8f487bfd..fb522332de9a4 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/lib.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/lib.rs @@ -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; diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/query_executor.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/query_executor.rs index fead0c5341b20..9404c9c3eaa2a 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/query_executor.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/query_executor.rs @@ -126,6 +126,7 @@ pub async fn execute_query( error!("Failed to infer schema: {}", e); e })?; + let resolved_schema = crate::schema_coerce::coerce_for_substrait(resolved_schema); let table_config = ListingTableConfig::new(table_path) .with_listing_options(listing_options) diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/schema_coerce.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/schema_coerce.rs new file mode 100644 index 0000000000000..7c4ec7e92cd53 --- /dev/null +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/schema_coerce.rs @@ -0,0 +1,206 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +//! Schema coercion at the Substrait/DataFusion scan boundary. +//! +//! Substrait's type system is narrower than Arrow's. The DataFusion Substrait +//! consumer rejects scans whose table-provider schema is not +//! `datatype_is_logically_equal` to the schema declared in the plan. That +//! function in DataFusion 53 has hardcoded equivalences for `(Utf8, Utf8View)` +//! but not for several other Arrow/Substrait-incompatible pairs we hit: +//! +//! - `BinaryView` ← parquet emits this for variable-length byte columns +//! (and for fixed-width byte arrays like OpenSearch's 16-byte `ip`) when +//! `schema_force_view_types` is on. Substrait has no view types — isthmus +//! serializes `VARBINARY` as plain `binary`, which arrives as +//! `DataType::Binary`. +//! +//! - `UInt64` ← parquet emits this for `unsigned_long` columns. +//! Substrait integers are signed only — Calcite `BIGINT` serializes as +//! `i64`, which arrives as `DataType::Int64`. +//! +//! Until upstream patches add equivalence arms (mirroring the existing +//! Utf8/Utf8View pair), we normalize the inferred schema at the table-provider +//! boundary: substitute the Arrow-only types with their Substrait-compatible +//! counterparts before handing the schema to `ListingTableConfig`. The parquet +//! reader's `SchemaAdapter` inserts the per-batch cast at read time. +//! +//! Strings keep their `Utf8View` layout because the upstream Utf8/Utf8View +//! equivalence already lets them pass through without a coerce. + +use std::sync::Arc; + +use datafusion::arrow::datatypes::{DataType, Field, Fields, Schema, SchemaRef}; + +/// Rewrite the schema to forms Substrait can bind against: +/// - `BinaryView` → `Binary` +/// - `UInt64` → `Int64` +/// +/// Recurses into `List`, `LargeList`, `FixedSizeList`, `Map`, `Struct`, +/// `Union`, and `Dictionary`. Returns the input unchanged when no rewrite is +/// needed so callers avoid an unnecessary `Arc` reallocation in the common +/// case. +pub fn coerce_for_substrait(schema: SchemaRef) -> SchemaRef { + if !schema_needs_coerce(&schema) { + return schema; + } + let rewritten_fields: Vec = schema + .fields() + .iter() + .map(|f| rewrite_field(f)) + .collect(); + Arc::new(Schema::new_with_metadata( + rewritten_fields, + schema.metadata().clone(), + )) +} + +fn schema_needs_coerce(schema: &Schema) -> bool { + schema.fields().iter().any(|f| contains_incompatible(f.data_type())) +} + +fn contains_incompatible(dt: &DataType) -> bool { + match dt { + DataType::BinaryView | DataType::UInt64 => true, + DataType::List(f) | DataType::LargeList(f) | DataType::FixedSizeList(f, _) => { + contains_incompatible(f.data_type()) + } + DataType::Map(f, _) => contains_incompatible(f.data_type()), + DataType::Struct(fields) => fields.iter().any(|f| contains_incompatible(f.data_type())), + DataType::Union(fields, _) => fields.iter().any(|(_, f)| contains_incompatible(f.data_type())), + DataType::Dictionary(_, value_type) => contains_incompatible(value_type), + _ => false, + } +} + +fn rewrite_field(field: &Field) -> Field { + let new_type = rewrite_data_type(field.data_type()); + Field::new(field.name(), new_type, field.is_nullable()) + .with_metadata(field.metadata().clone()) +} + +fn rewrite_data_type(dt: &DataType) -> DataType { + match dt { + DataType::BinaryView => DataType::Binary, + DataType::UInt64 => DataType::Int64, + DataType::List(f) => DataType::List(Arc::new(rewrite_field(f))), + DataType::LargeList(f) => DataType::LargeList(Arc::new(rewrite_field(f))), + DataType::FixedSizeList(f, n) => DataType::FixedSizeList(Arc::new(rewrite_field(f)), *n), + DataType::Map(f, sorted) => DataType::Map(Arc::new(rewrite_field(f)), *sorted), + DataType::Struct(fields) => { + let new_fields: Vec = fields.iter().map(|f| rewrite_field(f)).collect(); + DataType::Struct(Fields::from(new_fields)) + } + DataType::Dictionary(key, value) => { + DataType::Dictionary(key.clone(), Box::new(rewrite_data_type(value))) + } + other => other.clone(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn top_level_binary_view_gets_rewritten() { + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, true), + Field::new("b", DataType::BinaryView, true), + ])); + let out = coerce_for_substrait(schema); + assert_eq!(out.field(0).data_type(), &DataType::Int64); + assert_eq!(out.field(1).data_type(), &DataType::Binary); + } + + #[test] + fn top_level_uint64_gets_rewritten() { + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::UInt64, true), + Field::new("b", DataType::Int64, true), + ])); + let out = coerce_for_substrait(schema); + assert_eq!(out.field(0).data_type(), &DataType::Int64); + assert_eq!(out.field(1).data_type(), &DataType::Int64); + } + + #[test] + fn schema_without_incompatible_types_is_returned_unchanged() { + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, true), + Field::new("b", DataType::Utf8View, true), + ])); + let before = Arc::as_ptr(&schema); + let out = coerce_for_substrait(schema); + assert_eq!(Arc::as_ptr(&out), before, "unchanged schema must not reallocate"); + } + + #[test] + fn nested_list_of_binary_view_gets_rewritten() { + let inner = Field::new("item", DataType::BinaryView, true); + let schema = Arc::new(Schema::new(vec![ + Field::new("xs", DataType::List(Arc::new(inner)), true), + ])); + let out = coerce_for_substrait(schema); + match out.field(0).data_type() { + DataType::List(f) => assert_eq!(f.data_type(), &DataType::Binary), + other => panic!("expected List, got {other:?}"), + } + } + + #[test] + fn nested_list_of_uint64_gets_rewritten() { + let inner = Field::new("item", DataType::UInt64, true); + let schema = Arc::new(Schema::new(vec![ + Field::new("xs", DataType::List(Arc::new(inner)), true), + ])); + let out = coerce_for_substrait(schema); + match out.field(0).data_type() { + DataType::List(f) => assert_eq!(f.data_type(), &DataType::Int64), + other => panic!("expected List, got {other:?}"), + } + } + + #[test] + fn field_metadata_and_nullability_preserved() { + let mut md = std::collections::HashMap::new(); + md.insert("key".to_string(), "value".to_string()); + let f = Field::new("b", DataType::BinaryView, false).with_metadata(md.clone()); + let schema = Arc::new(Schema::new(vec![f])); + let out = coerce_for_substrait(schema); + assert!(!out.field(0).is_nullable()); + assert_eq!(out.field(0).metadata(), &md); + } + + #[test] + fn utf8_view_is_left_alone() { + let schema = Arc::new(Schema::new(vec![ + Field::new("s", DataType::Utf8View, true), + Field::new("b", DataType::BinaryView, true), + ])); + let out = coerce_for_substrait(schema); + assert_eq!(out.field(0).data_type(), &DataType::Utf8View); + assert_eq!(out.field(1).data_type(), &DataType::Binary); + } + + #[test] + fn other_unsigned_ints_are_left_alone() { + // Only UInt64 ↔ Int64 needs coercion for OpenSearch's unsigned_long + // mapping today. Smaller unsigned widths aren't produced by any current + // OpenSearch field type, and Calcite has no SMALLINT-equivalent + // unsigned type to mismatch against, so we leave them untouched. + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::UInt8, true), + Field::new("b", DataType::UInt16, true), + Field::new("c", DataType::UInt32, true), + ])); + let before = Arc::as_ptr(&schema); + let out = coerce_for_substrait(schema); + assert_eq!(Arc::as_ptr(&out), before); + } +} diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/session_context.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/session_context.rs index 31a4f5a0028cc..e11581887b817 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/session_context.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/session_context.rs @@ -142,6 +142,9 @@ pub async unsafe fn create_session_context( error!("create_session_context: failed to infer schema: {}", e); e })?; + // Substrait's type system is narrower than Arrow's; normalize the inferred + // schema to forms the Substrait consumer can bind against. See crate::schema_coerce. + let resolved_schema = crate::schema_coerce::coerce_for_substrait(resolved_schema); let table_config = ListingTableConfig::new(shard_view.table_path.clone()) .with_listing_options(listing_options) diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java index cf5f3782922c3..39fac75e93295 100644 --- a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java +++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java @@ -62,6 +62,9 @@ public class DataFusionAnalyticsBackendPlugin implements AnalyticsSearchBackendP SUPPORTED_FIELD_TYPES.addAll(FieldType.date()); SUPPORTED_FIELD_TYPES.add(FieldType.BOOLEAN); SUPPORTED_FIELD_TYPES.add(FieldType.TEXT); + SUPPORTED_FIELD_TYPES.add(FieldType.BINARY); + SUPPORTED_FIELD_TYPES.add(FieldType.IP); + SUPPORTED_FIELD_TYPES.add(FieldType.MATCH_ONLY_TEXT); } // Filter-side scalar functions DataFusion can evaluate natively. Comparisons, arithmetic diff --git a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/engine/OpenSearchSchemaBuilderTests.java b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/engine/OpenSearchSchemaBuilderTests.java index 4ad9f66fdfb42..b8fa554ad5eaa 100644 --- a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/engine/OpenSearchSchemaBuilderTests.java +++ b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/engine/OpenSearchSchemaBuilderTests.java @@ -60,7 +60,7 @@ public void testBuildSchemaWithIntegerFloatBoolean() throws Exception { RelDataType rowType = table.getRowType(new org.apache.calcite.jdbc.JavaTypeFactoryImpl()); assertFieldType(rowType, "count", SqlTypeName.INTEGER); - assertFieldType(rowType, "ratio", SqlTypeName.FLOAT); + assertFieldType(rowType, "ratio", SqlTypeName.REAL); assertFieldType(rowType, "active", SqlTypeName.BOOLEAN); } @@ -79,7 +79,7 @@ public void testBuildSchemaWithDateIpTextShortByte() throws Exception { RelDataType rowType = table.getRowType(new org.apache.calcite.jdbc.JavaTypeFactoryImpl()); assertFieldType(rowType, "created", SqlTypeName.TIMESTAMP); - assertFieldType(rowType, "address", SqlTypeName.VARCHAR); + assertFieldType(rowType, "address", SqlTypeName.VARBINARY); assertFieldType(rowType, "content", SqlTypeName.VARCHAR); assertFieldType(rowType, "small_num", SqlTypeName.SMALLINT); assertFieldType(rowType, "tiny_num", SqlTypeName.TINYINT); @@ -137,23 +137,28 @@ public void testEmptyClusterStateProducesEmptySchema() { public void testMapFieldTypeForAllSupportedTypes() { assertEquals(SqlTypeName.VARCHAR, OpenSearchSchemaBuilder.mapFieldType("keyword")); assertEquals(SqlTypeName.VARCHAR, OpenSearchSchemaBuilder.mapFieldType("text")); + assertEquals(SqlTypeName.VARCHAR, OpenSearchSchemaBuilder.mapFieldType("match_only_text")); assertEquals(SqlTypeName.BIGINT, OpenSearchSchemaBuilder.mapFieldType("long")); + assertEquals(SqlTypeName.BIGINT, OpenSearchSchemaBuilder.mapFieldType("unsigned_long")); + assertEquals(SqlTypeName.BIGINT, OpenSearchSchemaBuilder.mapFieldType("scaled_float")); assertEquals(SqlTypeName.INTEGER, OpenSearchSchemaBuilder.mapFieldType("integer")); assertEquals(SqlTypeName.SMALLINT, OpenSearchSchemaBuilder.mapFieldType("short")); assertEquals(SqlTypeName.TINYINT, OpenSearchSchemaBuilder.mapFieldType("byte")); assertEquals(SqlTypeName.DOUBLE, OpenSearchSchemaBuilder.mapFieldType("double")); - assertEquals(SqlTypeName.FLOAT, OpenSearchSchemaBuilder.mapFieldType("float")); + assertEquals(SqlTypeName.REAL, OpenSearchSchemaBuilder.mapFieldType("float")); assertEquals(SqlTypeName.BOOLEAN, OpenSearchSchemaBuilder.mapFieldType("boolean")); assertEquals(SqlTypeName.TIMESTAMP, OpenSearchSchemaBuilder.mapFieldType("date")); - assertEquals(SqlTypeName.VARCHAR, OpenSearchSchemaBuilder.mapFieldType("ip")); + assertEquals(SqlTypeName.TIMESTAMP, OpenSearchSchemaBuilder.mapFieldType("date_nanos")); + assertEquals(SqlTypeName.VARBINARY, OpenSearchSchemaBuilder.mapFieldType("ip")); + assertEquals(SqlTypeName.VARBINARY, OpenSearchSchemaBuilder.mapFieldType("binary")); } /** - * Test that unknown field types default to VARCHAR. + * Test that unknown field types throw IllegalArgumentException naming the offending type. */ - public void testUnknownFieldTypeDefaultsToVarchar() { - assertEquals(SqlTypeName.VARCHAR, OpenSearchSchemaBuilder.mapFieldType("unknown_type")); - assertEquals(SqlTypeName.VARCHAR, OpenSearchSchemaBuilder.mapFieldType("geo_point")); + public void testUnknownFieldTypeThrows() { + IllegalArgumentException ex = expectThrows(IllegalArgumentException.class, () -> OpenSearchSchemaBuilder.mapFieldType("geo_point")); + assertTrue("Exception message should mention the unsupported type, was: " + ex.getMessage(), ex.getMessage().contains("geo_point")); } // --- helpers --- diff --git a/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/FieldTypeCoverageIT.java b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/FieldTypeCoverageIT.java new file mode 100644 index 0000000000000..01f6e5683161c --- /dev/null +++ b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/FieldTypeCoverageIT.java @@ -0,0 +1,360 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.analytics.qa; + +import org.opensearch.client.Request; +import org.opensearch.client.Response; +import org.opensearch.client.ResponseException; + +import java.io.IOException; +import java.util.List; +import java.util.Map; + +/** + * Per-type coverage test for composite-engine + parquet indexing and the analytics-engine + * PPL read path. One test method per OpenSearch mapping type the sandbox targets. + * + *

Each test is a two-phase script: + *

    + *
  1. Ingest phase — {@link #ingest} creates the index, posts N docs, and returns + * the raw bulk response without asserting on it.
    2. + *
  2. Assertion phase — one or more of: + *
      + *
    • {@link #assertBulkSucceeded} — {@code errors=false} and flush. Use first when ingest is expected to succeed.
      • + *
    • {@link #assertBulkErrored} — {@code errors=true}. For types whose parquet writer throws.
      • + *
    • {@link #assertScanSucceeds} — bare {@code source=} scan; materializes every column.
      • + *
    • {@link #assertScanFails} — bare scan must throw. Known-bug guard for types DataFusion can't materialize.
      • + *
    + *
    3. + *
+ */ +public class FieldTypeCoverageIT extends AnalyticsRestTestCase { + + // ── Numeric ─────────────────────────────────────────────────────────────────── + + public void testByte() throws IOException { + // Known bug: byte ingest fails at the parquet writer (ByteParquetField throws). + // The bulk response reports errors=true with unsupported_operation_exception. + Map bulk = ingest("ft_byte", "byte", 1, 2, 3); + assertBulkErrored(bulk, "byte"); + } + + public void testShort() throws IOException { + Map bulk = ingest("ft_short", "short", 100, 200, 300); + assertBulkSucceeded(bulk, "ft_short"); + assertScanSucceeds("ft_short", 3); + } + + public void testInteger() throws IOException { + Map bulk = ingest("ft_integer", "integer", 1000, 2000, 3000); + assertBulkSucceeded(bulk, "ft_integer"); + assertScanSucceeds("ft_integer", 3); + } + + public void testLong() throws IOException { + Map bulk = ingest("ft_long", "long", 100000000L, 200000000L, 300000000L); + assertBulkSucceeded(bulk, "ft_long"); + assertScanSucceeds("ft_long", 3); + } + + public void testUnsignedLong() throws IOException { + // Test values stay below 2^63 - 1 because BIGINT is signed; values above wrap. + Map bulk = ingest( + "ft_unsigned_long", + "unsigned_long", + 12345678901234567L, + 23456789012345678L, + 34567890123456789L + ); + assertBulkSucceeded(bulk, "ft_unsigned_long"); + assertScanSucceeds("ft_unsigned_long", 3); + } + + public void testHalfFloat() throws IOException { + // Known bug: half_float ingest fails (HalfFloatParquetField throws). Same shape as byte. + Map bulk = ingest("ft_half_float", "half_float", 47.6, 45.5, 52.1); + assertBulkErrored(bulk, "half_float"); + } + + public void testFloat() throws IOException { + Map bulk = ingest("ft_float", "float", 47.6, 45.5, 52.1); + assertBulkSucceeded(bulk, "ft_float"); + assertScanSucceeds("ft_float", 3); + } + + public void testDouble() throws IOException { + Map bulk = ingest("ft_double", "double", 47.6, 45.5, 52.1); + assertBulkSucceeded(bulk, "ft_double"); + assertScanSucceeds("ft_double", 3); + } + + public void testScaledFloat() throws IOException { + // Scan binds as BIGINT (storage is Int64). Projections / predicates / aggregates + // expose the *scaled* long, not the original float. + Map bulk = ingestWithMapping("ft_scaled_float", "scaled_float", ", \"scaling_factor\": 100", 19.99, 25.50, 99.00); + assertBulkSucceeded(bulk, "ft_scaled_float"); + assertScanSucceeds("ft_scaled_float", 3); + } + + // ── Text / keyword ──────────────────────────────────────────────────────────── + + public void testKeyword() throws IOException { + Map bulk = ingest("ft_keyword", "keyword", "alice", "bob", "carol"); + assertBulkSucceeded(bulk, "ft_keyword"); + assertScanSucceeds("ft_keyword", 3); + } + + public void testText() throws IOException { + Map bulk = ingest("ft_text", "text", "connection refused", "host unreachable", "peer reset"); + assertBulkSucceeded(bulk, "ft_text"); + assertScanSucceeds("ft_text", 3); + } + + public void testMatchOnlyText() throws IOException { + // Scan unsupported: match_only_text has no doc_values, so the parquet writer skips + // the val column entirely (verified by inspecting the parquet footer schema). The + // value lives only in the Lucene secondary's inverted index, which is search-only + // — not readable as a column. PPL scan therefore fails at DataFusion with "No field + // named val". Coverage stops at ingest. + Map bulk = ingest( + "ft_match_only_text", + "match_only_text", + "timeout on socket", + "dns lookup failed", + "tls handshake error" + ); + assertBulkSucceeded(bulk, "ft_match_only_text"); + assertScanFails("ft_match_only_text"); + } + + // ── Temporal ───────────────────────────────────────────────────────────────── + + public void testDate() throws IOException { + Map bulk = ingest("ft_date", "date", "\"1990-01-15\"", "\"1995-05-20\"", "\"1988-03-10\""); + assertBulkSucceeded(bulk, "ft_date"); + assertScanSucceeds("ft_date", 3); + } + + public void testDateNanos() throws IOException { + // Scan binds as TIMESTAMP. Sub-millisecond precision is silently truncated because + // Calcite TIMESTAMP is millisecond-precision. + Map bulk = ingest( + "ft_date_nanos", + "date_nanos", + "\"1990-01-15T10:00:00.123456789Z\"", + "\"1995-05-20T11:00:00.987654321Z\"", + "\"1988-03-10T12:00:00.111222333Z\"" + ); + assertBulkSucceeded(bulk, "ft_date_nanos"); + assertScanSucceeds("ft_date_nanos", 3); + } + + // ── Other ──────────────────────────────────────────────────────────────────── + + public void testBoolean() throws IOException { + Map bulk = ingest("ft_boolean", "boolean", true, false, true); + assertBulkSucceeded(bulk, "ft_boolean"); + assertScanSucceeds("ft_boolean", 3); + } + + public void testBinary() throws IOException { + // Scan binds as VARBINARY; the BinaryView → Binary rewrite happens in + // schema_coerce::coerce_for_substrait. Predicates on binary columns still fail at + // planning ("Unsupported conversion for Relational Data type: VARBINARY"), but + // count() / projections / group-by work. + Map bulk = ingest("ft_binary", "binary", "\"YWxpY2U=\"", "\"Ym9i\"", "\"Y2Fyb2w=\""); + assertBulkSucceeded(bulk, "ft_binary"); + assertScanSucceeds("ft_binary", 3); + } + + public void testIp() throws IOException { + // Scan binds as VARBINARY; the BinaryView → Binary rewrite happens in + // schema_coerce::coerce_for_substrait. Projections return raw 16-byte IPv6-mapped + // bytes. Predicates fail at planning until VARBINARY is wired through the PPL + // expression planner. + Map bulk = ingest("ft_ip", "ip", "\"192.168.1.1\"", "\"10.0.0.1\"", "\"172.16.0.1\""); + assertBulkSucceeded(bulk, "ft_ip"); + assertScanSucceeds("ft_ip", 3); + } + + // ── Phase 1: Ingest ────────────────────────────────────────────────────────── + + /** + * Create a single-field index and post N docs (named a/b/c/...). Returns the parsed + * bulk response without asserting on it — the caller decides whether to expect + * success ({@link #assertBulkSucceeded}) or failure ({@link #assertBulkErrored}). + */ + private Map ingest(String index, String type, Object... values) throws IOException { + return ingestWithMapping(index, type, "", values); + } + + /** + * Variant of {@link #ingest} that lets the caller inject extra mapping properties into + * the field definition (e.g. {@code ", \"scaling_factor\": 100"} for {@code scaled_float}). + * The fragment must be a leading-comma-prefixed JSON snippet or empty. + * + *

Distinct method name (rather than another {@code ingest} overload) because + * {@code ingest(idx, type, "...string-value...", ...)} would otherwise bind to the + * extra-mapping overload — Java picks {@code String} over {@code Object} as more specific + * — silently eating the first row value as a mapping fragment. + */ + private Map ingestWithMapping(String index, String type, String extraMappingJson, Object... values) + throws IOException { + String mapping = "\"val\": { \"type\": \"" + type + "\"" + extraMappingJson + " }"; + createIndex(index, createBody(index, mapping)); + return bulkRaw(index, asJsonArray(values)); + } + + // ── Phase 2: Query assertions ──────────────────────────────────────────────── + + /** + * Bulk response must report {@code errors=false}. Flushes so the parquet rowgroup is + * sealed before any subsequent scan assertion runs. + */ + private void assertBulkSucceeded(Map bulkResponse, String index) throws IOException { + assertEquals("Expected bulk ingest to succeed for [" + index + "]", Boolean.FALSE, bulkResponse.get("errors")); + flush(index); + } + + /** + * Bulk response must report {@code errors=true}. For types where the primary writer's + * {@code ParquetField} throws (byte, half_float). + */ + private void assertBulkErrored(Map bulkResponse, String type) { + assertEquals( + "Type [" + type + "] unexpectedly succeeded at ingest. Known-bug guard: update the test if this is now fixed.", + Boolean.TRUE, + bulkResponse.get("errors") + ); + } + + /** + * Bare {@code source=} scan must return {@code expected} rows. Materializes every + * column, so this catches per-column read-path bugs that {@code count()} misses. + */ + private void assertScanSucceeds(String index, int expected) throws IOException { + Map resp = executePpl("source=" + index); + @SuppressWarnings("unchecked") + List> rows = (List>) resp.get("rows"); + assertNotNull("source=" + index + " response missing rows", rows); + assertEquals("source=" + index + " row count", expected, rows.size()); + } + + /** + * Bare {@code source=} scan must throw. Known-bug guard for types where the + * column never lands in parquet (e.g. match_only_text has no parquet writer). When + * the underlying gap is closed and the scan starts succeeding, this assertion will + * fail and prompt the test to be flipped to {@link #assertScanSucceeds}. + */ + private void assertScanFails(String index) { + Request req = new Request("POST", "/_analytics/ppl"); + req.setJsonEntity("{\"query\": \"source=" + index + "\"}"); + try { + Response resp = client().performRequest(req); + fail( + "Expected source=" + index + " to fail, got status " + + resp.getStatusLine().getStatusCode() + + ". Known-bug guard: update the test if this is now fixed." + ); + } catch (ResponseException expected) { + assertTrue( + "Expected 5xx for source=" + index + ", got " + expected.getResponse().getStatusLine().getStatusCode(), + expected.getResponse().getStatusLine().getStatusCode() >= 500 + ); + } catch (IOException io) { + // Transport-level error — also counts as "scan failed". + } + } + + // ── Lower-level helpers ────────────────────────────────────────────────────── + + private static String createBody(String index, String valMappingFragment) { + return "{" + + "\"settings\": {" + + " \"number_of_shards\": 1," + + " \"number_of_replicas\": 0," + + " \"index.pluggable.dataformat.enabled\": true," + + " \"index.pluggable.dataformat\": \"composite\"," + + " \"index.composite.primary_data_format\": \"parquet\"," + + " \"index.composite.secondary_data_formats\": [\"lucene\"]" + + "}," + + "\"mappings\": {" + + " \"properties\": {" + + " \"name\": { \"type\": \"keyword\" }," + + " " + valMappingFragment + + " }" + + "}" + + "}"; + } + + private void createIndex(String index, String body) throws IOException { + // Be forgiving of leftover state from previous runs (preserveIndicesUponCompletion = true). + try { + client().performRequest(new Request("DELETE", "/" + index)); + } catch (ResponseException ignored) { + // expected on first run + } + Request create = new Request("PUT", "/" + index); + create.setJsonEntity(body); + Map resp = assertOkAndParse(client().performRequest(create), "Create index " + index); + assertEquals(Boolean.TRUE, resp.get("acknowledged")); + } + + /** Convert each value to its JSON literal form. Strings get wrapped in quotes unless already pre-quoted. */ + private static String[] asJsonArray(Object... values) { + String[] out = new String[values.length]; + for (int i = 0; i < values.length; i++) { + Object v = values[i]; + if (v instanceof String) { + String s = (String) v; + out[i] = s.startsWith("\"") ? s : "\"" + s + "\""; + } else { + out[i] = String.valueOf(v); + } + } + return out; + } + + /** + * Fire an N-doc bulk with sequential names {@code a, b, c, …} and {@code val=jsonValues[i]}, + * returning the parsed response. + */ + private Map bulkRaw(String index, String[] jsonValues) throws IOException { + StringBuilder body = new StringBuilder(); + for (int i = 0; i < jsonValues.length; i++) { + body.append("{\"index\":{}}\n"); + body.append("{\"name\":\"").append(rowName(i)).append("\",\"val\":").append(jsonValues[i]).append("}\n"); + } + Request bulk = new Request("POST", "/" + index + "/_bulk"); + bulk.addParameter("refresh", "true"); + bulk.setOptions(bulk.getOptions().toBuilder().addHeader("Content-Type", "application/x-ndjson").build()); + bulk.setJsonEntity(body.toString()); + return assertOkAndParse(client().performRequest(bulk), "Bulk " + index); + } + + /** Generate a stable row name for index i: a, b, …, z, aa, ab, … */ + private static String rowName(int i) { + StringBuilder sb = new StringBuilder(); + do { + sb.insert(0, (char) ('a' + (i % 26))); + i = i / 26 - 1; + } while (i >= 0); + return sb.toString(); + } + + private void flush(String index) throws IOException { + client().performRequest(new Request("POST", "/" + index + "/_flush?force=true")); + } + + private Map executePpl(String ppl) throws IOException { + Request request = new Request("POST", "/_analytics/ppl"); + request.setJsonEntity("{\"query\": \"" + escapeJson(ppl) + "\"}"); + return assertOkAndParse(client().performRequest(request), "PPL: " + ppl); + } +} From 6e19a2968b20ff3e44a777c6b602f34cb35e3c4b Mon Sep 17 00:00:00 2001 From: Vinay Krishna Pudyodu Date: Wed, 13 May 2026 17:23:40 +0000 Subject: [PATCH 2/7] Updated the javadocs Signed-off-by: Vinay Krishna Pudyodu --- .../rust/src/api.rs | 2 +- .../rust/src/indexed_executor.rs | 2 +- .../rust/src/query_executor.rs | 2 +- .../rust/src/schema_coerce.rs | 70 +++++++++++++------ .../rust/src/session_context.rs | 2 +- 5 files changed, 54 insertions(+), 24 deletions(-) diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs index c332582cbd771..7a5cefa33567f 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs @@ -518,7 +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_for_substrait(schema); + let schema = crate::schema_coerce::coerce_inferred_schema(schema); let config = ListingTableConfig::new(table_path) .with_listing_options(listing_options) .with_schema(schema); diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_executor.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_executor.rs index f3d2d051f64a8..1940ffa1c42fb 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_executor.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_executor.rs @@ -156,7 +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_for_substrait(resolved_schema); + 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); diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/query_executor.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/query_executor.rs index 9404c9c3eaa2a..869a8e42e6805 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/query_executor.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/query_executor.rs @@ -126,7 +126,7 @@ pub async fn execute_query( error!("Failed to infer schema: {}", e); e })?; - let resolved_schema = crate::schema_coerce::coerce_for_substrait(resolved_schema); + let resolved_schema = crate::schema_coerce::coerce_inferred_schema(resolved_schema); let table_config = ListingTableConfig::new(table_path) .with_listing_options(listing_options) diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/schema_coerce.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/schema_coerce.rs index 7c4ec7e92cd53..39135ca908f5d 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/schema_coerce.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/schema_coerce.rs @@ -10,12 +10,11 @@ //! //! Substrait's type system is narrower than Arrow's. The DataFusion Substrait //! consumer rejects scans whose table-provider schema is not -//! `datatype_is_logically_equal` to the schema declared in the plan. That -//! function in DataFusion 53 has hardcoded equivalences for `(Utf8, Utf8View)` -//! but not for several other Arrow/Substrait-incompatible pairs we hit: +//! `datatype_is_logically_equal` to the schema declared in the plan. The pairs +//! we hit on this path: //! //! - `BinaryView` ← parquet emits this for variable-length byte columns -//! (and for fixed-width byte arrays like OpenSearch's 16-byte `ip`) when +//! (including OpenSearch's 16-byte `ip` and `binary` mappings) when //! `schema_force_view_types` is on. Substrait has no view types — isthmus //! serializes `VARBINARY` as plain `binary`, which arrives as //! `DataType::Binary`. @@ -24,14 +23,45 @@ //! Substrait integers are signed only — Calcite `BIGINT` serializes as //! `i64`, which arrives as `DataType::Int64`. //! -//! Until upstream patches add equivalence arms (mirroring the existing -//! Utf8/Utf8View pair), we normalize the inferred schema at the table-provider -//! boundary: substitute the Arrow-only types with their Substrait-compatible -//! counterparts before handing the schema to `ListingTableConfig`. The parquet -//! reader's `SchemaAdapter` inserts the per-batch cast at read time. +//! `Utf8View` doesn't need coercing: DataFusion 53's +//! `DFSchema::datatype_is_logically_equal` (in `datafusion-common/src/dfschema.rs`) +//! has hardcoded match arms `(Utf8, Utf8View) => true` and `(Utf8View, Utf8) => true`, +//! so a Substrait plan declaring `Utf8` binds cleanly against a table column +//! reporting `Utf8View`. The two cases above are different in nature: //! -//! Strings keep their `Utf8View` layout because the upstream Utf8/Utf8View -//! equivalence already lets them pass through without a coerce. +//! - `(Binary, BinaryView)` is a missing equivalence in DataFusion. The two +//! types are semantically identical, but `datatype_is_logically_equal` has +//! no arm for them today (DF 53). If it becomes available in DataFusion, +//! the `BinaryView → Binary` rewrite here can be removed. +//! +//! - `(Int64, UInt64)` is a Substrait + Calcite gap. Substrait's integer +//! types are signed-only and Calcite has no unsigned `BIGINT`, so the +//! unsigned semantics are lost before the plan reaches DataFusion. Values +//! above `2^63 - 1` wrap into negatives — documented in +//! `OpenSearchSchemaBuilder.mapFieldType`. Narrowing at the scan boundary +//! is the only fix until Substrait grows unsigned types (see Substrait +//! `proto/type.proto`). +//! +//! We rewrite the inferred schema at the table-provider boundary: substitute the +//! Arrow-only types with their Substrait-compatible counterparts before handing +//! the schema to `ListingTableConfig`. The parquet reader's `SchemaAdapter` then +//! inserts the per-batch cast at read time (zero-copy bit reinterpret for +//! `UInt64 → Int64`; cheap buffer relabeling for `BinaryView → Binary`). +//! +//! Alternatives considered: +//! +//! - Disable view types entirely with +//! `execution.parquet.schema_force_view_types = false` (threaded into +//! `ParquetFormat::with_options`). Makes the reader emit `Utf8/Binary` instead +//! of `Utf8View/BinaryView`. Removes the BinaryView mismatch but also strips +//! `Utf8View` from string columns, giving up the inline-prefix optimization +//! on filter/group-by/hash paths. +//! +//! - Skip `infer_schema` entirely and construct the Arrow schema directly from +//! the OpenSearch mapping (the same source Calcite reads). Single source of +//! truth, no post-process step. Costs the cross-language plumbing to ship +//! the schema from Java to Rust and adds a second mapping table to keep in +//! sync with `OpenSearchSchemaBuilder.mapFieldType`. use std::sync::Arc; @@ -45,7 +75,7 @@ use datafusion::arrow::datatypes::{DataType, Field, Fields, Schema, SchemaRef}; /// `Union`, and `Dictionary`. Returns the input unchanged when no rewrite is /// needed so callers avoid an unnecessary `Arc` reallocation in the common /// case. -pub fn coerce_for_substrait(schema: SchemaRef) -> SchemaRef { +pub fn coerce_inferred_schema(schema: SchemaRef) -> SchemaRef { if !schema_needs_coerce(&schema) { return schema; } @@ -113,7 +143,7 @@ mod tests { Field::new("a", DataType::Int64, true), Field::new("b", DataType::BinaryView, true), ])); - let out = coerce_for_substrait(schema); + let out = coerce_inferred_schema(schema); assert_eq!(out.field(0).data_type(), &DataType::Int64); assert_eq!(out.field(1).data_type(), &DataType::Binary); } @@ -124,7 +154,7 @@ mod tests { Field::new("a", DataType::UInt64, true), Field::new("b", DataType::Int64, true), ])); - let out = coerce_for_substrait(schema); + let out = coerce_inferred_schema(schema); assert_eq!(out.field(0).data_type(), &DataType::Int64); assert_eq!(out.field(1).data_type(), &DataType::Int64); } @@ -136,7 +166,7 @@ mod tests { Field::new("b", DataType::Utf8View, true), ])); let before = Arc::as_ptr(&schema); - let out = coerce_for_substrait(schema); + let out = coerce_inferred_schema(schema); assert_eq!(Arc::as_ptr(&out), before, "unchanged schema must not reallocate"); } @@ -146,7 +176,7 @@ mod tests { let schema = Arc::new(Schema::new(vec![ Field::new("xs", DataType::List(Arc::new(inner)), true), ])); - let out = coerce_for_substrait(schema); + let out = coerce_inferred_schema(schema); match out.field(0).data_type() { DataType::List(f) => assert_eq!(f.data_type(), &DataType::Binary), other => panic!("expected List, got {other:?}"), @@ -159,7 +189,7 @@ mod tests { let schema = Arc::new(Schema::new(vec![ Field::new("xs", DataType::List(Arc::new(inner)), true), ])); - let out = coerce_for_substrait(schema); + let out = coerce_inferred_schema(schema); match out.field(0).data_type() { DataType::List(f) => assert_eq!(f.data_type(), &DataType::Int64), other => panic!("expected List, got {other:?}"), @@ -172,7 +202,7 @@ mod tests { md.insert("key".to_string(), "value".to_string()); let f = Field::new("b", DataType::BinaryView, false).with_metadata(md.clone()); let schema = Arc::new(Schema::new(vec![f])); - let out = coerce_for_substrait(schema); + let out = coerce_inferred_schema(schema); assert!(!out.field(0).is_nullable()); assert_eq!(out.field(0).metadata(), &md); } @@ -183,7 +213,7 @@ mod tests { Field::new("s", DataType::Utf8View, true), Field::new("b", DataType::BinaryView, true), ])); - let out = coerce_for_substrait(schema); + let out = coerce_inferred_schema(schema); assert_eq!(out.field(0).data_type(), &DataType::Utf8View); assert_eq!(out.field(1).data_type(), &DataType::Binary); } @@ -200,7 +230,7 @@ mod tests { Field::new("c", DataType::UInt32, true), ])); let before = Arc::as_ptr(&schema); - let out = coerce_for_substrait(schema); + let out = coerce_inferred_schema(schema); assert_eq!(Arc::as_ptr(&out), before); } } diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/session_context.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/session_context.rs index e11581887b817..060fbcdad7a6a 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/session_context.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/session_context.rs @@ -144,7 +144,7 @@ pub async unsafe fn create_session_context( })?; // Substrait's type system is narrower than Arrow's; normalize the inferred // schema to forms the Substrait consumer can bind against. See crate::schema_coerce. - let resolved_schema = crate::schema_coerce::coerce_for_substrait(resolved_schema); + let resolved_schema = crate::schema_coerce::coerce_inferred_schema(resolved_schema); let table_config = ListingTableConfig::new(shard_view.table_path.clone()) .with_listing_options(listing_options) From 9d408d95f84dfc741ab04df51d22024c8dd0f5b2 Mon Sep 17 00:00:00 2001 From: Vinay Krishna Pudyodu Date: Fri, 15 May 2026 01:56:24 +0000 Subject: [PATCH 3/7] Added IP and Binary predicate support Signed-off-by: Vinay Krishna Pudyodu --- .../analytics/spi/ScalarFunction.java | 3 + .../analytics-backend-datafusion/build.gradle | 3 + .../be/datafusion/BinaryFunctionAdapter.java | 115 ++++++++++++++++++ .../DataFusionAnalyticsBackendPlugin.java | 1 + .../planner/rel/OpenSearchFilter.java | 14 ++- .../analytics/qa/FieldTypeCoverageIT.java | 66 +++++++++- 6 files changed, 196 insertions(+), 6 deletions(-) create mode 100644 sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/BinaryFunctionAdapter.java diff --git a/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/ScalarFunction.java b/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/ScalarFunction.java index 9aa9086e3b0eb..f319d0890611c 100644 --- a/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/ScalarFunction.java +++ b/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/ScalarFunction.java @@ -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 diff --git a/sandbox/plugins/analytics-backend-datafusion/build.gradle b/sandbox/plugins/analytics-backend-datafusion/build.gradle index 5e5175ac2a8f3..9d989aef64bb8 100644 --- a/sandbox/plugins/analytics-backend-datafusion/build.gradle +++ b/sandbox/plugins/analytics-backend-datafusion/build.gradle @@ -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 diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/BinaryFunctionAdapter.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/BinaryFunctionAdapter.java new file mode 100644 index 0000000000000..99f4e9bdb5345 --- /dev/null +++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/BinaryFunctionAdapter.java @@ -0,0 +1,115 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.be.datafusion; + +import org.apache.calcite.avatica.util.ByteString; +import org.apache.calcite.plan.RelOptCluster; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rex.RexBuilder; +import org.apache.calcite.rex.RexCall; +import org.apache.calcite.rex.RexLiteral; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.sql.type.SqlTypeName; +import org.opensearch.analytics.spi.FieldStorageInfo; +import org.opensearch.analytics.spi.ScalarFunctionAdapter; + +import java.net.InetAddress; +import java.net.UnknownHostException; +import java.util.Base64; +import java.util.List; + +/** + * Converts {@code BINARY(varchar_literal)} into a VARBINARY literal whose bytes + * match the on-disk encoding of {@code ip} / {@code binary} fields. + * + *

The SQL plugin emits {@code BINARY(varchar)} whenever a VARCHAR literal is + * compared to a VARBINARY column (see ExtendedRexBuilder.makeCast in the sql repo). + * This adapter resolves the placeholder so DataFusion's native byte-comparison + * operators can run against the parquet column. + * + *

Disambiguation between {@code ip} and {@code binary} encoding is by the + * literal's text shape: try IP first, fall back to base64. IP literals contain + * {@code .} or {@code :} which are invalid base64; base64 literals don't contain + * those characters so they fail {@code InetAddress.getByName} for IPv4/IPv6 forms. + * + *

Output is a {@code SqlTypeName.VARBINARY} literal — not BINARY. Calcite's + * {@code makeBinaryLiteral} produces BINARY with a fixed precision which isthmus + * serializes as Substrait FixedBinary(N), mismatching the parquet column's + * variable-length Binary type. + * + *

TODO: Frontends (SQL plugin / PPL parser) should hand the analytics core a + * plan whose literals are already in the correct on-disk byte form, with the + * field-type distinction (ip vs binary) preserved end-to-end. This adapter + * exists today because OpenSearchSchemaBuilder collapses both ip and binary + * fields to plain VARBINARY before the plan reaches the SQL plugin's coercion + * layer, so the encoding decision can't be made there. The adapter recovers + * the necessary context here in the analytics backend (via FieldStorageInfo) + * and rewrites the placeholder accordingly. + * + * @opensearch.internal + */ +class BinaryFunctionAdapter implements ScalarFunctionAdapter { + + @Override + public RexNode adapt(RexCall original, List fieldStorage, RelOptCluster cluster) { + if (original.getOperands().size() != 1 + || !(original.getOperands().get(0) instanceof RexLiteral lit) + || lit.getType().getSqlTypeName() != SqlTypeName.VARCHAR) { + return original; + } + String value = lit.getValueAs(String.class); + if (value == null) { + return original; + } + + byte[] bytes = encodeAsIp(value); + if (bytes == null) { + bytes = encodeAsBinary(value); + } + if (bytes == null) { + throw new IllegalArgumentException( + "BINARY operand '" + value + "' is neither a valid IP address nor a valid base64-encoded binary value" + ); + } + + RexBuilder rexBuilder = cluster.getRexBuilder(); + RelDataType varbinary = cluster.getTypeFactory().createSqlType(SqlTypeName.VARBINARY); + return rexBuilder.makeLiteral(new ByteString(bytes), varbinary, false); + } + + /** + * IPv6-mapped 16-byte encoding matching Lucene's {@code InetAddressPoint.encode} — + * the same encoding the parquet writer uses for {@code ip} fields. IPv4 is encoded + * as 10 zero bytes + {@code 0xff 0xff} + 4 IPv4 bytes (RFC 4291 §2.5.5.2). + * IPv6 is the raw 16 bytes. + */ + private static byte[] encodeAsIp(String value) { + try { + byte[] addr = InetAddress.getByName(value).getAddress(); + if (addr.length == 16) { + return addr; + } + byte[] mapped = new byte[16]; + mapped[10] = (byte) 0xff; + mapped[11] = (byte) 0xff; + System.arraycopy(addr, 0, mapped, 12, 4); + return mapped; + } catch (UnknownHostException e) { + return null; + } + } + + private static byte[] encodeAsBinary(String value) { + try { + return Base64.getDecoder().decode(value); + } catch (IllegalArgumentException e) { + return null; + } + } +} diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java index 39fac75e93295..2542fb3b16944 100644 --- a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java +++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java @@ -472,6 +472,7 @@ public Map scalarFunctionAdapters() { Map.entry(ScalarFunction.MVFIND, new MvfindAdapter()), Map.entry(ScalarFunction.MVZIP, new MvzipAdapter()), Map.entry(ScalarFunction.MVAPPEND, new MvappendAdapter()), + Map.entry(ScalarFunction.BINARY, new BinaryFunctionAdapter()), Map.entry(ScalarFunction.CONCAT, new ConcatFunctionAdapter()), Map.entry(ScalarFunction.CONVERT_TZ, new ConvertTzAdapter()), Map.entry(ScalarFunction.COSH, new HyperbolicOperatorAdapter(SqlLibraryOperators.COSH)), diff --git a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rel/OpenSearchFilter.java b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rel/OpenSearchFilter.java index fc93cf1f78133..1fdf715f6238b 100644 --- a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rel/OpenSearchFilter.java +++ b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rel/OpenSearchFilter.java @@ -19,6 +19,7 @@ import org.apache.calcite.rel.metadata.RelMetadataQuery; import org.apache.calcite.rex.RexCall; import org.apache.calcite.rex.RexNode; +import org.apache.calcite.rex.RexUtil; import org.opensearch.analytics.planner.RelNodeUtils; import org.opensearch.analytics.spi.FieldStorageInfo; @@ -102,7 +103,18 @@ public RelNode stripAnnotations(List strippedChildren) { @Override public RelNode stripAnnotations(List strippedChildren, Function annotationResolver) { - return LogicalFilter.create(strippedChildren.getFirst(), resolveCondition(getCondition(), annotationResolver)); + RexNode resolved = resolveCondition(getCondition(), annotationResolver); + // Defensive flatten just before LogicalFilter. asserts isFlat on the condition. + // Two known re-nesting sources to guard against: + // 1. Calcite's SARG → AND/OR decomposition. A flat input like + // AND(p1, p2, SEARCH(col, Sarg[100..5000])) gets expanded into + // AND(p1, p2, AND(col>=100, col<=5000)) by downstream rules without re-flattening. + // 2. Annotation unwrap (resolveCondition above) when the unwrapped RexNode is itself + // an AND/OR — e.g. when a function adapter rewrote a predicate into a conjunction. + // Both shapes pass the parent AND's clone() unchanged but trip the Filter constructor's + // assertion. RexUtil.flatten canonicalizes the tree so the assertion holds. + RexNode flattened = RexUtil.flatten(getCluster().getRexBuilder(), resolved); + return LogicalFilter.create(strippedChildren.getFirst(), flattened); } private RexNode replaceAnnotations(RexNode node, ListIterator annotationIterator) { diff --git a/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/FieldTypeCoverageIT.java b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/FieldTypeCoverageIT.java index 01f6e5683161c..27795e2eac9c4 100644 --- a/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/FieldTypeCoverageIT.java +++ b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/FieldTypeCoverageIT.java @@ -165,9 +165,10 @@ public void testBoolean() throws IOException { public void testBinary() throws IOException { // Scan binds as VARBINARY; the BinaryView → Binary rewrite happens in - // schema_coerce::coerce_for_substrait. Predicates on binary columns still fail at - // planning ("Unsupported conversion for Relational Data type: VARBINARY"), but - // count() / projections / group-by work. + // schema_coerce::coerce_for_substrait. Predicates on binary columns now work via the + // BINARY(varchar) placeholder (BinaryFunctionAdapter rewrites it into a VARBINARY + // literal that DataFusion compares natively). Filter coverage lives in testIpFilters + // — binary columns share the same code path. Map bulk = ingest("ft_binary", "binary", "\"YWxpY2U=\"", "\"Ym9i\"", "\"Y2Fyb2w=\""); assertBulkSucceeded(bulk, "ft_binary"); assertScanSucceeds("ft_binary", 3); @@ -176,13 +177,55 @@ public void testBinary() throws IOException { public void testIp() throws IOException { // Scan binds as VARBINARY; the BinaryView → Binary rewrite happens in // schema_coerce::coerce_for_substrait. Projections return raw 16-byte IPv6-mapped - // bytes. Predicates fail at planning until VARBINARY is wired through the PPL - // expression planner. + // bytes. Filter / aggregation coverage on `ip` columns is in testIpFilters. Map bulk = ingest("ft_ip", "ip", "\"192.168.1.1\"", "\"10.0.0.1\"", "\"172.16.0.1\""); assertBulkSucceeded(bulk, "ft_ip"); assertScanSucceeds("ft_ip", 3); } + /** + * End-to-end coverage of filter / aggregation shapes against an {@code ip} column. + * Each predicate shape exercises a different code path: + *

    + *
  • {@code =} — comparator with VARBINARY column + VARCHAR literal, expanded via + * {@code ExtendedRexBuilder.makeCast} to {@code BINARY(varchar)} and resolved by + * {@code BinaryFunctionAdapter}.
    • + *
  • {@code !=}, {@code >} — same path, different comparator.
    • + *
  • {@code IN} — exercised the {@code OpenSearchTypeFactory.leastRestrictive} override + * for VARBINARY ⇄ VARCHAR.
    • + *
  • {@code BETWEEN} — same {@code leastRestrictive} path with a 3-arg shape.
    • + *
  • {@code cidrmatch} — exercised the inline expansion in + * {@code PPLFuncImpTable.populate} that rewrites cidr literals into byte-range AND.
    • + *
  • {@code AND}-combination — exercised the {@code RexUtil.flatten} guard in + * {@code OpenSearchFilter.stripAnnotations}.
    • + *
+ */ + public void testIpFilters() throws IOException { + Map bulk = ingest("ft_ip_filters", "ip", "\"192.168.1.1\"", "\"10.0.0.1\"", "\"172.16.0.1\""); + assertBulkSucceeded(bulk, "ft_ip_filters"); + assertScanSucceeds("ft_ip_filters", 3); + + assertFilterRowCount("source=ft_ip_filters | where val = '192.168.1.1'", 1); + assertFilterRowCount("source=ft_ip_filters | where val != '192.168.1.1'", 2); + assertFilterRowCount("source=ft_ip_filters | where val > '10.0.0.50'", 2); + assertFilterRowCount("source=ft_ip_filters | where val < '172.16.0.1'", 1); + assertFilterRowCount("source=ft_ip_filters | where val in ('192.168.1.1', '10.0.0.1')", 2); + assertFilterRowCount("source=ft_ip_filters | where val between '10.0.0.0' and '10.255.255.255'", 1); + + assertFilterRowCount("source=ft_ip_filters | where cidrmatch(val, '192.168.0.0/16')", 1); + assertFilterRowCount("source=ft_ip_filters | where cidrmatch(val, '10.0.0.0/8')", 1); + assertFilterRowCount("source=ft_ip_filters | where cidrmatch(val, '0.0.0.0/0')", 3); + assertFilterRowCount("source=ft_ip_filters | where NOT cidrmatch(val, '192.168.0.0/16')", 2); + + // AND-combination exercises the flatten guard in OpenSearchFilter.stripAnnotations + // (>= 4 conjuncts after SARG/cidr expansion). + assertFilterRowCount( + "source=ft_ip_filters | where val > '10.0.0.0' AND val < '200.0.0.0'" + + " AND cidrmatch(val, '0.0.0.0/0')", + 3 + ); + } + // ── Phase 1: Ingest ────────────────────────────────────────────────────────── /** @@ -246,6 +289,19 @@ private void assertScanSucceeds(String index, int expected) throws IOException { assertEquals("source=" + index + " row count", expected, rows.size()); } + /** + * Runs a PPL query and asserts the row count matches {@code expected}. Used by tests + * that exercise filter / aggregation shapes (e.g. {@link #testIpFilters}) where the + * row count is the meaningful signal. + */ + private void assertFilterRowCount(String ppl, int expected) throws IOException { + Map resp = executePpl(ppl); + @SuppressWarnings("unchecked") + List> rows = (List>) resp.get("rows"); + assertNotNull("[" + ppl + "] response missing rows", rows); + assertEquals("[" + ppl + "] row count", expected, rows.size()); + } + /** * Bare {@code source=} scan must throw. Known-bug guard for types where the * column never lands in parquet (e.g. match_only_text has no parquet writer). When From 4eca3082b634604aee781659aef9ef328083ac08 Mon Sep 17 00:00:00 2001 From: Vinay Krishna Pudyodu Date: Fri, 15 May 2026 21:46:07 +0000 Subject: [PATCH 4/7] fixed type issues and ip project functions Signed-off-by: Vinay Krishna Pudyodu --- .../be/datafusion/BinaryFunctionAdapter.java | 5 ++-- .../DataFusionAnalyticsBackendPlugin.java | 1 + .../planner/rel/OpenSearchFilter.java | 13 ++++------- .../analytics/qa/FieldTypeCoverageIT.java | 23 +++++++++++++++++++ 4 files changed, 31 insertions(+), 11 deletions(-) diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/BinaryFunctionAdapter.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/BinaryFunctionAdapter.java index 99f4e9bdb5345..ee562b3ad0310 100644 --- a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/BinaryFunctionAdapter.java +++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/BinaryFunctionAdapter.java @@ -79,8 +79,9 @@ public RexNode adapt(RexCall original, List fieldStorage, RelO } RexBuilder rexBuilder = cluster.getRexBuilder(); - RelDataType varbinary = cluster.getTypeFactory().createSqlType(SqlTypeName.VARBINARY); - return rexBuilder.makeLiteral(new ByteString(bytes), varbinary, false); + RelDataType varbinary = cluster.getTypeFactory() + .createTypeWithNullability(cluster.getTypeFactory().createSqlType(SqlTypeName.VARBINARY), true); + return rexBuilder.makeAbstractCast(varbinary, rexBuilder.makeLiteral(new ByteString(bytes), varbinary, false), false); } /** diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java index 2542fb3b16944..90f7ba6d9a794 100644 --- a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java +++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java @@ -280,6 +280,7 @@ public class DataFusionAnalyticsBackendPlugin implements AnalyticsSearchBackendP // by a custom Rust UDF on the DataFusion session context (`udf::mvfind`), routed via // {@link MvfindAdapter}. ScalarFunction.MVFIND, + ScalarFunction.BINARY, // Logical connectives — emitted in projections where boolean expressions are composed: // `case(a = 0 and b = 0, …)`, `eval x = a or b`, `eval x = NOT y`. DataFusion's substrait // consumer handles them natively. diff --git a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rel/OpenSearchFilter.java b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rel/OpenSearchFilter.java index 1fdf715f6238b..edf9d89d8dd73 100644 --- a/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rel/OpenSearchFilter.java +++ b/sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rel/OpenSearchFilter.java @@ -104,15 +104,10 @@ public RelNode stripAnnotations(List strippedChildren) { @Override public RelNode stripAnnotations(List strippedChildren, Function annotationResolver) { RexNode resolved = resolveCondition(getCondition(), annotationResolver); - // Defensive flatten just before LogicalFilter. asserts isFlat on the condition. - // Two known re-nesting sources to guard against: - // 1. Calcite's SARG → AND/OR decomposition. A flat input like - // AND(p1, p2, SEARCH(col, Sarg[100..5000])) gets expanded into - // AND(p1, p2, AND(col>=100, col<=5000)) by downstream rules without re-flattening. - // 2. Annotation unwrap (resolveCondition above) when the unwrapped RexNode is itself - // an AND/OR — e.g. when a function adapter rewrote a predicate into a conjunction. - // Both shapes pass the parent AND's clone() unchanged but trip the Filter constructor's - // assertion. RexUtil.flatten canonicalizes the tree so the assertion holds. + // LogicalFilter. asserts isFlat on the condition: an AND must not contain an + // AND child, and an OR must not contain an OR child. Adapter substitutions in + // BackendPlanAdapter.adaptRex can introduce that nesting (e.g. SargAdapter expands + // SEARCH into AND(>=, <=) under a parent AND). Flatten canonicalizes the tree. RexNode flattened = RexUtil.flatten(getCluster().getRexBuilder(), resolved); return LogicalFilter.create(strippedChildren.getFirst(), flattened); } diff --git a/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/FieldTypeCoverageIT.java b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/FieldTypeCoverageIT.java index 27795e2eac9c4..2f95fb1aea6da 100644 --- a/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/FieldTypeCoverageIT.java +++ b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/FieldTypeCoverageIT.java @@ -226,6 +226,29 @@ public void testIpFilters() throws IOException { ); } + /** + * Project-side coverage of {@code BinaryFunctionAdapter}. PPL {@code eval if(col=lit, …)}, + * {@code case(col=lit, …)}, and {@code count(eval(col=lit))} all lower to a + * {@code BINARY('lit':VARCHAR)} placeholder inside a {@code LogicalProject} (or a CASE in + * the project tree above an aggregate). + */ + public void testIpAndBinaryProjectExpressions() throws IOException { + Map ipBulk = ingest("ft_ip_project", "ip", "\"192.168.1.1\"", "\"10.0.0.1\"", "\"172.16.0.1\""); + assertBulkSucceeded(ipBulk, "ft_ip_project"); + assertFilterRowCount("source=ft_ip_project | eval is_local=if(val='192.168.1.1','y','n')", 3); + assertFilterRowCount( + "source=ft_ip_project | eval cls=case(val='192.168.1.1','a',val='10.0.0.1','b',true,'c')", + 3 + ); + assertFilterRowCount("source=ft_ip_project | stats count(eval(val='192.168.1.1')) as cnt", 1); + + Map binBulk = ingest("ft_binary_project", "binary", "\"YWxpY2U=\"", "\"Ym9i\"", "\"Y2Fyb2w=\""); + assertBulkSucceeded(binBulk, "ft_binary_project"); + assertFilterRowCount("source=ft_binary_project | eval is_alice=if(val='YWxpY2U=','y','n')", 3); + assertFilterRowCount("source=ft_binary_project | stats count(eval(val='YWxpY2U=')) as c", 1); + assertFilterRowCount("source=ft_ip_project | where val='192.168.1.1' | fields val", 1); + } + // ── Phase 1: Ingest ────────────────────────────────────────────────────────── /** From 5e722bb8d3ced0c9ba6761bd379ec733f8217c4e Mon Sep 17 00:00:00 2001 From: Vinay Krishna Pudyodu Date: Fri, 15 May 2026 22:58:44 +0000 Subject: [PATCH 5/7] remove local publish of sql repo Signed-off-by: Vinay Krishna Pudyodu --- .github/workflows/sandbox-check.yml | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) diff --git a/.github/workflows/sandbox-check.yml b/.github/workflows/sandbox-check.yml index f5aa63315bb80..fa4ddb079ea7f 100644 --- a/.github/workflows/sandbox-check.yml +++ b/.github/workflows/sandbox-check.yml @@ -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 From 9b6987fc31cf9b71ad55e2b6344db9e688780f83 Mon Sep 17 00:00:00 2001 From: Vinay Krishna Pudyodu Date: Fri, 15 May 2026 23:22:30 +0000 Subject: [PATCH 6/7] added todos Signed-off-by: Vinay Krishna Pudyodu --- .../opensearch/analytics/schema/OpenSearchSchemaBuilder.java | 3 +++ .../analytics-backend-datafusion/rust/src/schema_coerce.rs | 5 +++++ 2 files changed, 8 insertions(+) diff --git a/sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/OpenSearchSchemaBuilder.java b/sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/OpenSearchSchemaBuilder.java index 7cd3f5415d91a..3acb2868881d4 100644 --- a/sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/OpenSearchSchemaBuilder.java +++ b/sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/OpenSearchSchemaBuilder.java @@ -117,6 +117,9 @@ public static SqlTypeName mapFieldType(String opensearchType) { 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: throw new IllegalArgumentException("Unsupported OpenSearch field type: " + opensearchType); diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/schema_coerce.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/schema_coerce.rs index 39135ca908f5d..c1b364dc83c69 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/schema_coerce.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/schema_coerce.rs @@ -34,6 +34,11 @@ //! no arm for them today (DF 53). If it becomes available in DataFusion, //! the `BinaryView → Binary` rewrite here can be removed. //! +//! TODO: every record batch goes through a `BinaryView → Binary` cast in the +//! SchemaAdapter (offset+data buffer copy), and downstream operators see +//! `Binary` rather than `BinaryView`. Drop this arm when we have a proper +//! solution. +//! //! - `(Int64, UInt64)` is a Substrait + Calcite gap. Substrait's integer //! types are signed-only and Calcite has no unsigned `BIGINT`, so the //! unsigned semantics are lost before the plan reaches DataFusion. Values From a1062c0e1b0c176bb698e9a449330c077f27c148 Mon Sep 17 00:00:00 2001 From: Vinay Krishna Pudyodu Date: Sat, 16 May 2026 00:30:01 +0000 Subject: [PATCH 7/7] added half_float support Signed-off-by: Vinay Krishna Pudyodu --- .../schema/OpenSearchSchemaBuilder.java | 10 +++++++++ .../rust/src/schema_coerce.rs | 21 ++++++++++++++++++- .../engine/OpenSearchSchemaBuilderTests.java | 1 + .../analytics/qa/FieldTypeCoverageIT.java | 19 +++++++---------- 4 files changed, 39 insertions(+), 12 deletions(-) diff --git a/sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/OpenSearchSchemaBuilder.java b/sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/OpenSearchSchemaBuilder.java index 3acb2868881d4..05b238f5dbd3d 100644 --- a/sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/OpenSearchSchemaBuilder.java +++ b/sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/OpenSearchSchemaBuilder.java @@ -76,6 +76,7 @@ public static SchemaPlus buildSchema(ClusterState clusterState) { *
  • byte -> TINYINT
    • *
  • double -> DOUBLE
    • *
  • float -> REAL
    • + *
  • half_float -> REAL
    • *
  • scaled_float -> BIGINT
    • *
  • boolean -> BOOLEAN
    • *
  • date -> TIMESTAMP
    • @@ -98,6 +99,8 @@ public static SqlTypeName mapFieldType(String opensearchType) { 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": @@ -109,6 +112,13 @@ public static SqlTypeName mapFieldType(String opensearchType) { case "double": return SqlTypeName.DOUBLE; case "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; diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/schema_coerce.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/schema_coerce.rs index c1b364dc83c69..5bb0e9355c1df 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/schema_coerce.rs +++ b/sandbox/plugins/analytics-backend-datafusion/rust/src/schema_coerce.rs @@ -23,6 +23,10 @@ //! Substrait integers are signed only — Calcite `BIGINT` serializes as //! `i64`, which arrives as `DataType::Int64`. //! +//! - `Float16` ← parquet emits this for `half_float` columns. Calcite has +//! no fp16 type; `OpenSearchSchemaBuilder` maps it to `REAL` which Substrait +//! serializes as `fp32`, arriving as `DataType::Float32`. +//! //! `Utf8View` doesn't need coercing: DataFusion 53's //! `DFSchema::datatype_is_logically_equal` (in `datafusion-common/src/dfschema.rs`) //! has hardcoded match arms `(Utf8, Utf8View) => true` and `(Utf8View, Utf8) => true`, @@ -47,6 +51,19 @@ //! is the only fix until Substrait grows unsigned types (see Substrait //! `proto/type.proto`). //! +//! TODO: values above `2^63 - 1` wrap into negatives. Drop this arm when we +//! have a proper solution. +//! +//! - `(Float32, Float16)` is a Calcite-side gap: Calcite has no fp16 type, so +//! `half_float` columns are widened to `REAL` (fp32) at the Java planner. +//! Every record batch goes through a `Float16 → Float32` cast in the +//! SchemaAdapter, and downstream operators see `Float32` rather than +//! `Float16` — so we lose the half-precision storage benefit at compute time. +//! +//! TODO: every record batch goes through a `Float16 → Float32` cast and +//! downstream operators see `Float32`. Drop this arm when we have a proper +//! solution. +//! //! We rewrite the inferred schema at the table-provider boundary: substitute the //! Arrow-only types with their Substrait-compatible counterparts before handing //! the schema to `ListingTableConfig`. The parquet reader's `SchemaAdapter` then @@ -75,6 +92,7 @@ use datafusion::arrow::datatypes::{DataType, Field, Fields, Schema, SchemaRef}; /// Rewrite the schema to forms Substrait can bind against: /// - `BinaryView` → `Binary` /// - `UInt64` → `Int64` +/// - `Float16` → `Float32` /// /// Recurses into `List`, `LargeList`, `FixedSizeList`, `Map`, `Struct`, /// `Union`, and `Dictionary`. Returns the input unchanged when no rewrite is @@ -101,7 +119,7 @@ fn schema_needs_coerce(schema: &Schema) -> bool { fn contains_incompatible(dt: &DataType) -> bool { match dt { - DataType::BinaryView | DataType::UInt64 => true, + DataType::BinaryView | DataType::UInt64 | DataType::Float16 => true, DataType::List(f) | DataType::LargeList(f) | DataType::FixedSizeList(f, _) => { contains_incompatible(f.data_type()) } @@ -123,6 +141,7 @@ fn rewrite_data_type(dt: &DataType) -> DataType { match dt { DataType::BinaryView => DataType::Binary, DataType::UInt64 => DataType::Int64, + DataType::Float16 => DataType::Float32, DataType::List(f) => DataType::List(Arc::new(rewrite_field(f))), DataType::LargeList(f) => DataType::LargeList(Arc::new(rewrite_field(f))), DataType::FixedSizeList(f, n) => DataType::FixedSizeList(Arc::new(rewrite_field(f)), *n), diff --git a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/engine/OpenSearchSchemaBuilderTests.java b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/engine/OpenSearchSchemaBuilderTests.java index b8fa554ad5eaa..b5f33407f1f3c 100644 --- a/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/engine/OpenSearchSchemaBuilderTests.java +++ b/sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/engine/OpenSearchSchemaBuilderTests.java @@ -146,6 +146,7 @@ public void testMapFieldTypeForAllSupportedTypes() { assertEquals(SqlTypeName.TINYINT, OpenSearchSchemaBuilder.mapFieldType("byte")); assertEquals(SqlTypeName.DOUBLE, OpenSearchSchemaBuilder.mapFieldType("double")); assertEquals(SqlTypeName.REAL, OpenSearchSchemaBuilder.mapFieldType("float")); + assertEquals(SqlTypeName.REAL, OpenSearchSchemaBuilder.mapFieldType("half_float")); assertEquals(SqlTypeName.BOOLEAN, OpenSearchSchemaBuilder.mapFieldType("boolean")); assertEquals(SqlTypeName.TIMESTAMP, OpenSearchSchemaBuilder.mapFieldType("date")); assertEquals(SqlTypeName.TIMESTAMP, OpenSearchSchemaBuilder.mapFieldType("date_nanos")); diff --git a/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/FieldTypeCoverageIT.java b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/FieldTypeCoverageIT.java index 2f95fb1aea6da..cc5ad29d29a1c 100644 --- a/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/FieldTypeCoverageIT.java +++ b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/FieldTypeCoverageIT.java @@ -39,10 +39,9 @@ public class FieldTypeCoverageIT extends AnalyticsRestTestCase { // ── Numeric ─────────────────────────────────────────────────────────────────── public void testByte() throws IOException { - // Known bug: byte ingest fails at the parquet writer (ByteParquetField throws). - // The bulk response reports errors=true with unsupported_operation_exception. Map bulk = ingest("ft_byte", "byte", 1, 2, 3); - assertBulkErrored(bulk, "byte"); + assertBulkSucceeded(bulk, "ft_byte"); + assertScanSucceeds("ft_byte", 3); } public void testShort() throws IOException { @@ -77,9 +76,12 @@ public void testUnsignedLong() throws IOException { } public void testHalfFloat() throws IOException { - // Known bug: half_float ingest fails (HalfFloatParquetField throws). Same shape as byte. + // half_float lands as Arrow Float16. OpenSearchSchemaBuilder maps it to REAL and + // schema_coerce widens Float16 → Float32 before substrait binds; SchemaAdapter + // inserts the IEEE half→single cast per batch on read. Map bulk = ingest("ft_half_float", "half_float", 47.6, 45.5, 52.1); - assertBulkErrored(bulk, "half_float"); + assertBulkSucceeded(bulk, "ft_half_float"); + assertScanSucceeds("ft_half_float", 3); } public void testFloat() throws IOException { @@ -117,11 +119,6 @@ public void testText() throws IOException { } public void testMatchOnlyText() throws IOException { - // Scan unsupported: match_only_text has no doc_values, so the parquet writer skips - // the val column entirely (verified by inspecting the parquet footer schema). The - // value lives only in the Lucene secondary's inverted index, which is search-only - // — not readable as a column. PPL scan therefore fails at DataFusion with "No field - // named val". Coverage stops at ingest. Map bulk = ingest( "ft_match_only_text", "match_only_text", @@ -130,7 +127,7 @@ public void testMatchOnlyText() throws IOException { "tls handshake error" ); assertBulkSucceeded(bulk, "ft_match_only_text"); - assertScanFails("ft_match_only_text"); + assertScanSucceeds("ft_match_only_text", 3); } // ── Temporal ─────────────────────────────────────────────────────────────────