[Analytics Backend] UInt64 → Int64 bridge at partition-stream boundary

Resolves the 13 by-clause failures in CalciteTimechartCommandIT (5/18 → 18/18)
caused by DataFusion's row_number() physical op emitting UInt64 while
Calcite types ROW_NUMBER OVER as BIGINT (Int64 signed). Two coupled
changes mirror the existing schema_coerce::coerce_inferred_schema bridge
the parquet-scan path uses, applied at the partition-stream boundary
where the analytics-engine reduce stage reads from upstream PARTIAL
fragments via NativeBridge.registerPartitionStream.

1. api::register_partition_stream — coerce the producer's physical
   schema via schema_coerce before registering the StreamingTable and
   before writing the IPC bytes back to Java. The Substrait consumer
   plan's `ensure_schema_compatibility` (datafusion-substrait 53)
   strict-checks each ReadRel's base_schema against the registered
   table; the coercion makes Int64-declared plans bind against this
   table. The Java-side typesMatch tripwire then sees the coerced
   schema and validates incoming batches against it (next change).

2. api::sender_send — coerce incoming RecordBatches to match the
   sender's (now-coerced) schema via arrow::compute::cast where types
   differ. Without this, DataFusion's HashJoin / RepartitionExec
   panics with "primitive array" on as_primitive::<Int64Type>() of
   a UInt64 column when the producer's actual batch reaches downstream
   operators. cast() is the existing Arrow kernel — no-op when the
   schemas already match, single-cast on the mismatched columns
   otherwise. Same width preserves zero-copy semantics modulo the
   bit-reinterpret arrow handles internally.

3. DatafusionReduceSink.typesMatch — relax Java's tripwire to treat
   same-width Int (any signedness) as compatible, mirroring the existing
   Timestamp(any precision) tolerance. Same rationale: divergence is
   harmless because the sender_send coercion produces a downstream-
   compatible batch.

Pass-rate impact:
- CalciteTimechartCommandIT:    5/18  → 18/18 ✅ (+13)
- CalciteTimechartPerFunctionIT: 1/12 → 1/12  (residuals are opensearch-project#21584's
                                                multi-unit-SPAN gap +
                                                variable-month TIMESTAMPADD)
- Combined timechart:           6/30  → 19/30 (63%)

Signed-off-by: Kai Huang <huangkaics@gmail.com>

Author: ahkcs

sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs

@@ -793,6 +793,14 @@ pub unsafe fn register_partition_stream(
 ) -> Result<(i64, Vec<u8>), DataFusionError> {
     let session = &mut *(session_ptr as *mut LocalSession);
     let schema = derive_schema_from_partial_plan(partial_plan_bytes)?;
+    // Substrait-narrowing parity with the scan boundary (see schema_coerce). The producer's
+    // physical output schema may include Arrow-only types Substrait can't bind against —
+    // notably UInt64 from a DataFusion `row_number()` physical op (Calcite types ROW_NUMBER
+    // OVER as BIGINT / Int64, isthmus serializes accordingly, then `from_substrait_plan` of
+    // the consumer plan trips `ensure_schema_compatibility` against this registered table).
+    // Coerce up-front so the consumer plan binds; the channel batches still arrive with the
+    // producer's native types and get cast to the declared schema by the receiver adapter.
+    let schema = crate::schema_coerce::coerce_inferred_schema(schema);
     let schema_ipc = schema_to_ipc_bytes(schema.as_ref())?;
     let sender = session.register_partition(input_id, schema)?;
     Ok((Box::into_raw(Box::new(sender)) as i64, schema_ipc))
@@ -934,9 +942,52 @@ pub unsafe fn sender_send(
     let struct_array = StructArray::from(array_data);
     let batch = RecordBatch::from(struct_array);
 
+    // Mirror the schema_coerce step applied at registration (see register_partition_stream):
+    // when the producer's physical type differs from the consumer-declared type by an
+    // Arrow-only divergence (UInt64 ↔ Int64, Float16 → Float32, BinaryView ↔ Binary), the
+    // table is registered with the consumer-declared type and the StreamingTable / downstream
+    // operators downcast strictly to that type. The batch as received here still carries the
+    // producer's original type, so cast each mismatched column to the sender's schema before
+    // pushing into the channel.
+    let batch = coerce_batch_to_sender_schema(batch, sender.schema())?;
+
     sender.send_blocking(Ok(batch), io_handle)
 }
 
+/// Cast batch columns to match the sender's declared schema where they differ. No-op when
+/// schemas already match (the common case). Returns an error if a cast is not representable
+/// via Arrow's `cast` kernel — callers should not register a sender whose schema is not
+/// reachable from the batch's via cast (the registration-time `schema_coerce` only narrows
+/// types the cast kernel handles).
+fn coerce_batch_to_sender_schema(
+    batch: RecordBatch,
+    target_schema: &arrow::datatypes::SchemaRef,
+) -> Result<RecordBatch, DataFusionError> {
+    if batch.schema().as_ref() == target_schema.as_ref() {
+        return Ok(batch);
+    }
+    let mut columns: Vec<arrow::array::ArrayRef> = Vec::with_capacity(batch.num_columns());
+    for (i, target_field) in target_schema.fields().iter().enumerate() {
+        let src = batch.column(i);
+        if src.data_type() == target_field.data_type() {
+            columns.push(src.clone());
+        } else {
+            columns.push(arrow::compute::cast(src, target_field.data_type()).map_err(|e| {
+                DataFusionError::Execution(format!(
+                    "Failed to cast partition stream column '{}' from {:?} to {:?}: {}",
+                    target_field.name(),
+                    src.data_type(),
+                    target_field.data_type(),
+                    e
+                ))
+            })?);
+        }
+    }
+    RecordBatch::try_new(Arc::clone(target_schema), columns).map_err(|e| {
+        DataFusionError::Execution(format!("Failed to assemble coerced partition stream batch: {}", e))
+    })
+}
+
 /// Closes a partition stream sender. Dropping the sender closes the mpsc,
 /// which the receiver side (DataFusion's streaming table) interprets as
 /// end-of-input.

sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DatafusionReduceSink.java

@@ -234,6 +234,12 @@ private void feedToSender(DatafusionPartitionSender sender, VectorSchemaRoot bat
      * parameters are tolerated because the data-node parquet reader and physical
      * planner pick a precision the Java-side declaration does not predict, and the
      * chosen precision round-trips through Arrow C Data — divergence is harmless.
+     * Same-width Int signedness divergence is tolerated for the same reason as the
+     * Rust-side {@code schema_coerce::coerce_inferred_schema(UInt64 → Int64)} bridge:
+     * DataFusion's {@code row_number()} physical op emits UInt64, while Calcite types
+     * {@code ROW_NUMBER OVER} as BIGINT (Int64 signed). The Substrait check is satisfied
+     * on the registration side by the coercer; this tripwire mirrors the relaxation so
+     * the actual UInt64 batches from the producer pass through.
      */
     private static boolean typesMatch(Schema actual, Schema declared) {
         List<Field> a = actual.getFields();
@@ -247,6 +253,9 @@ private static boolean typesMatch(Schema actual, Schema declared) {
             if (at.getTypeID() == ArrowType.ArrowTypeID.Timestamp && dt.getTypeID() == ArrowType.ArrowTypeID.Timestamp) {
                 continue;
             }
+            if (at instanceof ArrowType.Int ai && dt instanceof ArrowType.Int di && ai.getBitWidth() == di.getBitWidth()) {
+                continue;
+            }
             if (!at.equals(dt)) {
                 return false;
             }