[Analytics Engine] Eliminate string/utf8 views & timestamp field coercion across the data-node→coordinator pipeline

[sandeshkr419]

Co-Authored-By: Claude Opus 4.7 (1M context)

Description

Producer (data node) emits Utf8View strings and Timestamp(Millisecond) timestamps. The coordinator's StreamingTable was registered with Utf8 and Timestamp(Second) multiple places, forcing a per-cell coercion at the wire and a per-batch cast at every UDF call site to bridge the schema mismatch.

This PR aligns the coordinator's declared schema with what the producer actually emits. Schema is now derived in Rust at registration time from the producer's substrait, schema crosses the FFI once. Scalar UDFs accept all three string variants natively via a single dispatch enum.

Net effect: the data path is fully zero-copy for strings and timestamps across the FFI, and string-heavy aggregates / UDF queries skip an O(rows × bytes) memcpy + buffer reallocation per batch. Downstream operators consume the producer's emitted types unchanged.

Related Issues

Resolves #[Issue number to be closed when this PR is merged]

Check List

• Functionality includes testing.
• API changes companion pull request created, if applicable.
• Public documentation issue/PR created, if applicable.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 5b48bff)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Possible Issue derive_schema_from_partial_plan registers MemTables with empty batch vectors (vec![vec![]]). If the producer plan contains aggregates or filters that rely on actual data to determine nullability or physical types, the derived schema may diverge from what a real data-bearing execution would produce. This could cause schema mismatches at runtime when the coordinator feeds actual batches. let table = MemTable::try_new(Arc::new(arrow_schema), vec![vec![]])?; // Plan may scan the same table twice; the second register is a no-op. Possible Issue collect_reads recursively walks the substrait plan but does not handle all possible RelType variants. If the plan contains a RelType not explicitly matched (e.g., Cross, Window, ExtensionSingle, etc.), those branches are silently ignored. If such a branch contains a ReadRel, it will not be registered, leading to a missing table error when the plan is lowered. fn collect_reads(rel: &substrait::proto::Rel, out: &mut Vec<substrait::proto::ReadRel>) { use substrait::proto::rel::RelType; match rel.rel_type.as_ref() { Some(RelType::Read(r)) => out.push((**r).clone()), Some(RelType::Filter(f)) => { if let Some(input) = &f.input { collect_reads(input, out); } } Some(RelType::Project(p)) => { if let Some(input) = &p.input { collect_reads(input, out); } } Some(RelType::Aggregate(a)) => { if let Some(input) = &a.input { collect_reads(input, out); } } Some(RelType::Sort(s)) => { if let Some(input) = &s.input { collect_reads(input, out); } } Some(RelType::Fetch(f)) => { if let Some(input) = &f.input { collect_reads(input, out); } } Some(RelType::Join(j)) => { if let Some(left) = &j.left { collect_reads(left, out); } if let Some(right) = &j.right { collect_reads(right, out); } } Some(RelType::Set(s)) => { for input in &s.inputs { collect_reads(input, out); } } _ => {} } } Possible Issue coerce_unsupported_timestamp_precision only coerces Timestamp(Second) to Timestamp(Millisecond). If the producer emits Timestamp(Microsecond) or Timestamp(Nanosecond) and the coordinator's declared schema is Timestamp(Second), no coercion occurs. The comment claims parquet only supports MILLIS/MICROS/NANOS, but the code does not handle MICROS or NANOS coercion, potentially causing a mismatch. fn coerce_unsupported_timestamp_precision( schema: &arrow::datatypes::Schema, ) -> arrow::datatypes::Schema { use arrow::datatypes::{DataType, Field, TimeUnit}; let fields: Vec<Field> = schema .fields() .iter() .map(|f| match f.data_type() { DataType::Timestamp(TimeUnit::Second, tz) => Field::new( f.name(), DataType::Timestamp(TimeUnit::Millisecond, tz.clone()), f.is_nullable(), ) .with_metadata(f.metadata().clone()), _ => f.as_ref().clone(), }) .collect(); arrow::datatypes::Schema::new_with_metadata(fields, schema.metadata().clone()) } Possible Issue typesMatch ignores nullability and all Timestamp precision/timezone parameters. If the producer emits Timestamp(Millisecond, "UTC") and the declared schema is Timestamp(Millisecond, null), the types match here but Arrow's C Data import may reject the batch due to timezone mismatch. The comment claims divergence is harmless, but timezone mismatches can cause runtime errors in Arrow FFI. private static boolean typesMatch(Schema actual, Schema declared) { List<Field> a = actual.getFields(); List<Field> d = declared.getFields(); if (a.size() != d.size()) { return false; } for (int i = 0; i < a.size(); i++) { ArrowType at = a.get(i).getType(); ArrowType dt = d.get(i).getType(); if (at.getTypeID() == ArrowType.ArrowTypeID.Timestamp && dt.getTypeID() == ArrowType.ArrowTypeID.Timestamp) { continue; } if (!at.equals(dt)) { return false; } } return true; }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 9a0c6b0

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Verify field names match The typesMatch method should also verify field names match between schemas. Currently it only checks types and field count, but mismatched field names could cause data corruption when columns are accessed by name rather than index. Add a name equality check alongside the type comparison. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DatafusionReduceSink.java [263-280] private static boolean typesMatch(Schema actual, Schema declared) { List<Field> a = actual.getFields(); List<Field> d = declared.getFields(); if (a.size() != d.size()) { return false; } for (int i = 0; i < a.size(); i++) { + if (!a.get(i).getName().equals(d.get(i).getName())) { + return false; + } ArrowType at = a.get(i).getType(); ArrowType dt = d.get(i).getType(); if (at.getTypeID() == ArrowType.ArrowTypeID.Timestamp && dt.getTypeID() == ArrowType.ArrowTypeID.Timestamp) { continue; } if (!at.equals(dt)) { return false; } } return true; } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that typesMatch only validates field types and count but not field names. Since Arrow schemas can be accessed by name, mismatched names could lead to incorrect column mapping. Adding a name check would strengthen the validation, though the current implementation may work if columns are accessed by index only.	Medium

---

Previous suggestions

Suggestions up to commit 5b48bff

Category	Suggestion	Impact
Possible issue	Capture schema before closing batch The batch is closed before throwing the exception, but the exception message includes batch.getSchema() which accesses the closed batch. This can lead to undefined behavior or errors when accessing a closed VectorSchemaRoot. Capture the schema before closing the batch. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DatafusionReduceSink.java [210-219] if (!typesMatch(batch.getSchema(), declaredSchema)) { + Schema batchSchema = batch.getSchema(); batch.close(); throw new IllegalStateException( "DatafusionReduceSink: batch schema types do not match declared schema. " + "declared=" + declaredSchema + " batch=" - + batch.getSchema() + + batchSchema ); } Suggestion importance[1-10]: 9 __ Why: Accessing batch.getSchema() after closing the batch can lead to undefined behavior. The suggestion correctly identifies this critical issue and provides a proper fix by capturing the schema before closing.	High
General	Add cleanup for partial registration failures If registerPartitionStream throws an exception after successfully registering some inputs, the partially-registered senders are not cleaned up. This can lead to resource leaks. Wrap the registration loop in a try-catch block and close any registered senders on failure. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DatafusionReduceSink.java [113-123] -NativeBridge.RegisteredInput registered = NativeBridge.registerPartitionStream( - session.getPointer(), - inputIdFor(childStageId), - producerPlanBytes -); -senders.put(childStageId, new DatafusionPartitionSender(registered.pointer())); -childSchemas.put(childStageId, ArrowSchemaIpc.fromBytes(registered.schemaIpc())); +try { + for (Map.Entry<Integer, byte[]> child : childInputs.entrySet()) { + int childStageId = child.getKey(); + byte[] producerPlanBytes = child.getValue(); + NativeBridge.RegisteredInput registered = NativeBridge.registerPartitionStream( + session.getPointer(), + inputIdFor(childStageId), + producerPlanBytes + ); + senders.put(childStageId, new DatafusionPartitionSender(registered.pointer())); + childSchemas.put(childStageId, ArrowSchemaIpc.fromBytes(registered.schemaIpc())); + } +} catch (Exception e) { + for (DatafusionPartitionSender sender : senders.values()) { + sender.close(); + } + throw e; +} Suggestion importance[1-10]: 7 __ Why: The suggestion identifies a valid resource leak scenario where partial registration failures could leave senders unclosed. However, the existing code already has a try-catch block at the outer level (line 95) that handles cleanup, making this less critical than initially appears.	Medium

Suggestions up to commit da94c42

Category	Suggestion	Impact
Possible issue	Avoid accessing closed batch schema The batch is closed before throwing the exception, but the exception message references batch.getSchema() which accesses the closed batch. This could lead to undefined behavior or errors when constructing the exception message. Move the schema extraction before closing the batch. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DatafusionReduceSink.java [210-219] if (!typesMatch(batch.getSchema(), declaredSchema)) { + String batchSchemaStr = batch.getSchema().toString(); batch.close(); throw new IllegalStateException( "DatafusionReduceSink: batch schema types do not match declared schema. " + "declared=" + declaredSchema + " batch=" - + batch.getSchema() + + batchSchemaStr ); } Suggestion importance[1-10]: 9 __ Why: Accessing batch.getSchema() after batch.close() is a critical bug that could cause undefined behavior or exceptions. The schema should be extracted before closing the batch.	High
General	Handle nested timestamp precision coercion The function only handles top-level Timestamp(Second) fields but doesn't recursively process nested struct or list fields that might contain timestamps. If the schema contains nested timestamp fields with unsupported precision, they won't be coerced, potentially causing schema mismatches. sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs [642-660] fn coerce_unsupported_timestamp_precision( schema: &arrow::datatypes::Schema, ) -> arrow::datatypes::Schema { use arrow::datatypes::{DataType, Field, TimeUnit}; - let fields: Vec<Field> = schema - .fields() - .iter() - .map(|f| match f.data_type() { - DataType::Timestamp(TimeUnit::Second, tz) => Field::new( - f.name(), + fn coerce_field(f: &Field) -> Field { + let coerced_type = match f.data_type() { + DataType::Timestamp(TimeUnit::Second, tz) => DataType::Timestamp(TimeUnit::Millisecond, tz.clone()), - f.is_nullable(), - ) - .with_metadata(f.metadata().clone()), - _ => f.as_ref().clone(), - }) - .collect(); + DataType::Struct(fields) => + DataType::Struct(fields.iter().map(coerce_field).collect()), + DataType::List(inner) => + DataType::List(Arc::new(coerce_field(inner))), + _ => f.data_type().clone(), + }; + Field::new(f.name(), coerced_type, f.is_nullable()) + .with_metadata(f.metadata().clone()) + } + let fields: Vec<Field> = schema.fields().iter().map(coerce_field).collect(); arrow::datatypes::Schema::new_with_metadata(fields, schema.metadata().clone()) } Suggestion importance[1-10]: 6 __ Why: Valid observation that nested timestamp fields aren't handled, but the current implementation may be sufficient for the actual data structures used. The suggestion adds complexity that may not be needed unless nested timestamps are actually encountered in production.	Low
	Clarify schema transformation ownership The coerce_unsupported_timestamp_precision function returns a new Schema but the result is assigned to a variable with the same name, shadowing the previous binding. This pattern works but could be clearer by using a mutable variable or a different name to indicate the transformation. sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs [607] -let arrow_schema = coerce_unsupported_timestamp_precision(&arrow_schema); +let arrow_schema = Arc::new(coerce_unsupported_timestamp_precision(arrow_schema.as_ref())); Suggestion importance[1-10]: 3 __ Why: The variable shadowing pattern is valid Rust and commonly used for transformations. The suggestion to wrap in Arc::new changes the type unnecessarily since coerce_unsupported_timestamp_precision already returns a Schema, not an Arc<Schema>.	Low

Suggestions up to commit 9510f74

Category	Suggestion	Impact
Possible issue	Avoid accessing closed batch schema The batch is closed before throwing the exception, but the exception message includes batch.getSchema() which accesses the already-closed batch. This could lead to undefined behavior or errors when constructing the exception message. Move the schema extraction before closing the batch. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DatafusionReduceSink.java [210-219] if (!typesMatch(batch.getSchema(), declaredSchema)) { + String batchSchemaStr = batch.getSchema().toString(); batch.close(); throw new IllegalStateException( "DatafusionReduceSink: batch schema types do not match declared schema. " + "declared=" + declaredSchema + " batch=" - + batch.getSchema() + + batchSchemaStr ); } Suggestion importance[1-10]: 9 __ Why: Accessing batch.getSchema() after batch.close() is a critical bug that could lead to undefined behavior or errors. The schema should be extracted before closing the batch.	High
General	Reduce schema cloning overhead The table_schema is cloned for every batch in the loop. For large batch counts, this repeated cloning of the schema (which includes metadata and field definitions) could impact performance. Consider using Arc::clone on a wrapped schema or restructure to avoid repeated clones. sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs [914-937] let table_schema = derive_schema_from_partial_plan(partial_plan_bytes)?; let schema_ipc = schema_to_ipc_bytes(table_schema.as_ref())?; +let table_schema_arc = Arc::new((*table_schema).clone()); ... for (&array_ptr, &schema_ptr) in array_ptrs.iter().zip(schema_ptrs.iter()) { ... - let aligned = RecordBatch::try_new(table_schema.clone(), columns)?; + let aligned = RecordBatch::try_new(Arc::clone(&table_schema_arc), columns)?; batches.push(aligned); } Suggestion importance[1-10]: 5 __ Why: Reducing repeated schema clones in a loop is a reasonable performance optimization, especially for large batch counts. However, the actual impact depends on batch size and schema complexity, making this a moderate improvement.	Low
	Optimize error path execution order If register_partition fails after schema_to_ipc_bytes succeeds, the function returns an error but the allocated schema_ipc buffer is already constructed. While this is not a memory leak (the buffer is owned and will be dropped), consider the ordering to avoid unnecessary work on the error path. sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs [769-773] let schema = derive_schema_from_partial_plan(partial_plan_bytes)?; +let sender = session.register_partition(input_id, schema.clone())?; 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)) Suggestion importance[1-10]: 4 __ Why: While reordering operations to avoid unnecessary work on error paths is a valid optimization, the impact is minimal since schema_to_ipc_bytes is relatively lightweight and errors from register_partition are expected to be rare.	Low

Suggestions up to commit 9510f74

Category	Suggestion	Impact
Possible issue	Avoid accessing closed batch schema The batch is closed before throwing the exception, but the exception message includes batch.getSchema() which accesses the now-closed batch. This could lead to undefined behavior or errors when constructing the error message. Move the schema extraction before closing the batch. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DatafusionReduceSink.java [210-219] if (!typesMatch(batch.getSchema(), declaredSchema)) { + String batchSchemaStr = batch.getSchema().toString(); batch.close(); throw new IllegalStateException( "DatafusionReduceSink: batch schema types do not match declared schema. " + "declared=" + declaredSchema + " batch=" - + batch.getSchema() + + batchSchemaStr ); } Suggestion importance[1-10]: 9 __ Why: Accessing batch.getSchema() after batch.close() is a critical bug that could lead to undefined behavior or errors. The schema should be extracted before closing the batch to ensure safe access in the error message.	High
General	Handle table registration errors explicitly The comment states that duplicate registration is a no-op, but the result is silently discarded with let _ =. If registration fails for reasons other than duplication (e.g., invalid table name), the error is ignored. Consider handling or at least logging registration failures. sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs [609-612] let table = MemTable::try_new(Arc::new(arrow_schema), vec![vec![]])?; // Plan may scan the same table twice; the second register is a no-op. -let _ = ctx.register_table(&table_name, Arc::new(table)); +if let Err(e) = ctx.register_table(&table_name, Arc::new(table)) { + log::debug!("Table registration for '{}' returned error (may be duplicate): {}", table_name, e); +} Suggestion importance[1-10]: 5 __ Why: The suggestion to log registration failures is reasonable for debugging, but the comment explicitly states duplicate registration is expected and harmless. Adding logging would improve observability without changing behavior.	Low
	Optimize error path execution order If register_partition fails after schema_to_ipc_bytes succeeds, the function returns an error but the allocated schema_ipc buffer is already constructed. While this is not a memory leak (Rust will drop it), consider whether the error path should be optimized to avoid unnecessary work. sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs [769-777] let schema = derive_schema_from_partial_plan(partial_plan_bytes)?; +let sender = session.register_partition(input_id, schema.clone())?; 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)) Suggestion importance[1-10]: 3 __ Why: While reordering operations could avoid unnecessary work in the error path, the current implementation is correct and the performance impact is negligible. The schema_ipc encoding is relatively cheap and the error path is exceptional.	Low

[github-actions]

Persistent review updated to latest commit 9510f74

[github-actions]

Persistent review updated to latest commit da94c42

[github-actions]

Persistent review updated to latest commit 5b48bff