diff --git a/src/servers/src/otlp/logs.rs b/src/servers/src/otlp/logs.rs
index 879d6b6d63..a956c7977c 100644
--- a/src/servers/src/otlp/logs.rs
+++ b/src/servers/src/otlp/logs.rs
@@ -327,6 +327,28 @@ struct ExistingLogColumn {
     datatype: ColumnDataType,
 }
 
+impl ExistingLogColumn {
+    fn schema_for_request_type(&self, request_type: ColumnDataType) -> ColumnSchema {
+        let mut schema = self.schema.clone();
+        if request_type == ColumnDataType::Binary && self.is_json_binary() {
+            schema.datatype = ColumnDataType::Binary as i32;
+        }
+        schema
+    }
+
+    fn is_json_binary(&self) -> bool {
+        self.datatype == ColumnDataType::Json
+            && matches!(
+                self.schema
+                    .datatype_extension
+                    .as_ref()
+                    .and_then(|datatype_extension| datatype_extension.type_ext.as_ref()),
+                Some(TypeExt::JsonType(json_type))
+                     if *json_type == JsonTypeExtension::JsonBinary as i32
+            )
+    }
+}
+
 #[derive(Default)]
 struct ExistingLogSchema {
     columns: HashMap<String, ExistingLogColumn>,
@@ -581,6 +603,7 @@ fn decide_existing_column_schema_and_convert_value(
         existing_column.datatype,
         existing_column.schema.semantic_type(),
         request_type,
+        existing_column.is_json_binary(),
         column_name,
         table_name,
     )?;
@@ -646,6 +669,7 @@ fn align_rows_with_existing_schema(
 
         let target_type = existing_column.datatype;
         let semantic_type = existing_column.schema.semantic_type();
+        let target_is_json_binary = existing_column.is_json_binary();
         for row in rows.iter_mut() {
             let Some(value) = row.values.get_mut(column_idx) else {
                 continue;
@@ -655,11 +679,12 @@ fn align_rows_with_existing_schema(
                 target_type,
                 semantic_type,
                 request_type,
+                target_is_json_binary,
                 &schema.column_name,
                 table_name,
             )?;
         }
-        *schema = existing_column.schema.clone();
+        *schema = existing_column.schema_for_request_type(request_type);
     }
 
     Ok(())
@@ -670,6 +695,7 @@ fn coerce_log_value_data(
     target_type: ColumnDataType,
     _semantic_type: SemanticType,
     request_type: ColumnDataType,
+    target_is_json_binary: bool,
     column_name: &str,
     table_name: &str,
 ) -> Result<Option<ValueData>> {
@@ -681,6 +707,10 @@ fn coerce_log_value_data(
         return Ok(Some(value_data));
     }
 
+    if request_type == ColumnDataType::Binary && target_is_json_binary {
+        return Ok(Some(value_data));
+    }
+
     if is_timestamp_type(request_type)
         && let Some(target_unit) = timestamp_unit(target_type)
     {
diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs
index 91a683f937..640eea7072 100644
--- a/tests-integration/tests/http.rs
+++ b/tests-integration/tests/http.rs
@@ -6387,6 +6387,28 @@ pub async fn test_otlp_logs(store_type: StorageType) {
             "[[1]]",
         )
         .await;
+
+        // Write another batch after the table exists. This verifies existing-schema
+        // alignment accepts the built-in JSONB columns in direct OTLP logs.
+        let res = send_req(
+            &client,
+            vec![(
+                HeaderName::from_static("content-type"),
+                HeaderValue::from_static("application/x-protobuf"),
+            )],
+            "/v1/otlp/v1/logs?db=public",
+            body.clone(),
+            false,
+        )
+        .await;
+        assert_eq!(StatusCode::OK, res.status());
+        validate_data(
+            "otlp_logs_multiple_batches",
+            &client,
+            "select count(*) from opentelemetry_logs;",
+            "[[4]]",
+        )
+        .await;
     }
 
     {