perf: [durable_buffer] Avoid unnecessary copy of OTLP bytes in OtlpBytesAdapter::new() (open-telemetry#2726)

Replace `BinaryArray::from_vec()` which deep-copies the entire OTLP
payload with a zero-copy construction using
`Buffer::from(bytes::Bytes)`. The `clone_bytes()` call is just an Arc
refcount bump, and `Buffer::from(Bytes)` wraps the data without copying,
eliminating a full memcpy on the ingest hot path.
 
 # Change Summary
 
- **Zero-copy wrapping**: Use `Buffer::from(bytes::Bytes)` instead of
`BinaryArray::from_vec()` to avoid deep-copying the OTLP payload into
Arrow.
- **Bounds check**: Added explicit `i32::try_from` guard on payload
length for a clear error on oversized payloads.
- **New tests**: `test_otlp_bytes_adapter_zero_copy` (pointer equality
assertion) and `test_otlp_bytes_adapter_empty_payload`.
 
 ## What issue does this PR close?
 
 * Closes open-telemetry#2703
 
 ## How are these changes tested?
 
- Existing tests (`test_otlp_bytes_adapter`, `test_extract_otlp_bytes`)
verify correctness is preserved.
- New `test_otlp_bytes_adapter_zero_copy` asserts the Arrow buffer
points to the same memory as the original `OtlpProtoBytes` (no copy).
- New `test_otlp_bytes_adapter_empty_payload` verifies empty payload
handling.
 
 ## Are there any user-facing changes?
 
No. This is a transparent performance improvement with no API or
behavior changes.

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: gscalderon

rust/otap-dataflow/crates/core-nodes/src/processors/durable_buffer_processor/bundle_adapter.rs

@@ -45,6 +45,7 @@ use std::sync::{Arc, LazyLock};
 use std::time::SystemTime;
 
 use arrow::array::{BinaryArray, RecordBatch};
+use arrow::buffer::{Buffer, OffsetBuffer, ScalarBuffer};
 use arrow::datatypes::{DataType, Field, Schema};
 use quiver::record_bundle::{
     BundleDescriptor, PayloadRef, RecordBundle, SchemaFingerprint, SlotDescriptor, SlotId,
@@ -348,8 +349,22 @@ impl OtlpBytesAdapter {
             OtlpProtoBytes::ExportTracesRequest(_) => SignalType::Traces,
         };
 
-        // Create a record batch with a single binary column containing the OTLP bytes
-        let binary_array = BinaryArray::from_vec(vec![bytes.as_bytes()]);
+        // Create a record batch with a single binary column containing the OTLP bytes.
+        // Use zero-copy wrapping: clone_bytes() is just an Arc refcount bump,
+        // and Buffer::from(Bytes) wraps without copying the payload data.
+        let data_bytes = bytes.clone_bytes();
+        let len = i32::try_from(data_bytes.len()).map_err(|_| {
+            (
+                BundleConversionError::RecordBatchCreationError(format!(
+                    "OTLP payload too large for BinaryArray: {} bytes exceeds i32::MAX",
+                    data_bytes.len()
+                )),
+                bytes.clone(),
+            )
+        })?;
+        let data_buffer = Buffer::from(data_bytes);
+        let offsets = OffsetBuffer::new(ScalarBuffer::from(vec![0i32, len]));
+        let binary_array = BinaryArray::new(offsets, data_buffer, None);
         let batch = match RecordBatch::try_new(otlp_binary_schema(), vec![Arc::new(binary_array)]) {
             Ok(batch) => batch,
             Err(e) => {
@@ -736,6 +751,41 @@ mod tests {
         assert!(adapter.payload(wrong_slot).is_none());
     }
 
+    #[test]
+    fn test_otlp_bytes_adapter_zero_copy() {
+        let test_bytes = b"zero-copy verification payload".to_vec();
+        let otlp = OtlpProtoBytes::new_from_bytes(SignalType::Logs, test_bytes);
+
+        let adapter = OtlpBytesAdapter::new(otlp).map_err(|(e, _)| e).unwrap();
+
+        let slot = to_otlp_slot_id(SignalType::Logs);
+        let payload = adapter.payload(slot).unwrap();
+        let column = payload.batch.column(0);
+        let binary_array = column.as_any().downcast_ref::<BinaryArray>().unwrap();
+
+        // The Arrow buffer should alias the original bytes (zero-copy).
+        // Compare the pointer of the stored value with the original OtlpProtoBytes.
+        let arrow_value_ptr = binary_array.value(0).as_ptr();
+        let original_ptr = adapter.bytes.as_bytes().as_ptr();
+        assert_eq!(
+            arrow_value_ptr, original_ptr,
+            "BinaryArray value should point to the same memory as OtlpProtoBytes (zero-copy)"
+        );
+    }
+
+    #[test]
+    fn test_otlp_bytes_adapter_empty_payload() {
+        let otlp = OtlpProtoBytes::new_from_bytes(SignalType::Metrics, vec![]);
+
+        let adapter = OtlpBytesAdapter::new(otlp).map_err(|(e, _)| e).unwrap();
+
+        let slot = to_otlp_slot_id(SignalType::Metrics);
+        let payload = adapter.payload(slot).unwrap();
+        let column = payload.batch.column(0);
+        let binary_array = column.as_any().downcast_ref::<BinaryArray>().unwrap();
+        assert_eq!(binary_array.value(0), b"");
+    }
+
     #[test]
     fn test_extract_otlp_bytes() {
         let original_bytes = b"original OTLP data".to_vec();