= 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_inferred_schema(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_inferred_schema(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_inferred_schema(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_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:?}"),
+ }
+ }
+
+ #[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_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:?}"),
+ }
+ }
+
+ #[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_inferred_schema(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_inferred_schema(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_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 31a4f5a0028cc..060fbcdad7a6a 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_inferred_schema(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/BinaryFunctionAdapter.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/BinaryFunctionAdapter.java
new file mode 100644
index 0000000000000..ee562b3ad0310
--- /dev/null
+++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/BinaryFunctionAdapter.java
@@ -0,0 +1,116 @@
+/*
+ * 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()
+ .createTypeWithNullability(cluster.getTypeFactory().createSqlType(SqlTypeName.VARBINARY), true);
+ return rexBuilder.makeAbstractCast(varbinary, rexBuilder.makeLiteral(new ByteString(bytes), varbinary, false), 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 cf5f3782922c3..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
@@ -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
@@ -277,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.
@@ -469,6 +473,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..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
@@ -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,13 @@ 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);
+ // 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);
}
private RexNode replaceAnnotations(RexNode node, ListIterator annotationIterator) {
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..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
@@ -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,29 @@ 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.REAL, OpenSearchSchemaBuilder.mapFieldType("half_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..cc5ad29d29a1c
--- /dev/null
+++ b/sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/FieldTypeCoverageIT.java
@@ -0,0 +1,436 @@
+/*
+ * 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:
+ *
+ * - Ingest phase — {@link #ingest} creates the index, posts N docs, and returns
+ * the raw bulk response without asserting on it.
+ * - 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.
+ *
+ *
+ *
+ */
+public class FieldTypeCoverageIT extends AnalyticsRestTestCase {
+
+ // ── Numeric ───────────────────────────────────────────────────────────────────
+
+ public void testByte() throws IOException {
+ Map bulk = ingest("ft_byte", "byte", 1, 2, 3);
+ assertBulkSucceeded(bulk, "ft_byte");
+ assertScanSucceeds("ft_byte", 3);
+ }
+
+ 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 {
+ // 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);
+ assertBulkSucceeded(bulk, "ft_half_float");
+ assertScanSucceeds("ft_half_float", 3);
+ }
+
+ 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 {
+ 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");
+ assertScanSucceeds("ft_match_only_text", 3);
+ }
+
+ // ── 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 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);
+ }
+
+ 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. 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
+ );
+ }
+
+ /**
+ * 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 ──────────────────────────────────────────────────────────
+
+ /**
+ * 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());
+ }
+
+ /**
+ * 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
+ * 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);
+ }
+}