diff --git a/internal/collector/generated/postgres_logs_transforms.json b/internal/collector/generated/postgres_logs_transforms.json
index f7409174e..066c06739 100644
--- a/internal/collector/generated/postgres_logs_transforms.json
+++ b/internal/collector/generated/postgres_logs_transforms.json
@@ -1 +1 @@
-[{"conditions":["body[\"format\"] == \"csv\""],"statements":["set(log.cache, ParseCSV(log.body[\"original\"], log.body[\"headers\"], delimiter=\",\", mode=\"strict\"))","merge_maps(log.cache, ExtractPatterns(log.cache[\"connection_from\"], \"(?:^[[]local[]]:(?\u003cremote_port\u003e.+)|:(?\u003cremote_port\u003e[^:]+))$\"), \"insert\") where Len(log.cache[\"connection_from\"]) \u003e 0","set(log.cache[\"remote_host\"], Substring(log.cache[\"connection_from\"], 0, Len(log.cache[\"connection_from\"]) - Len(log.cache[\"remote_port\"]) - 1)) where Len(log.cache[\"connection_from\"]) \u003e 0 and IsString(log.cache[\"remote_port\"])","set(log.cache[\"remote_host\"], log.cache[\"connection_from\"]) where Len(log.cache[\"connection_from\"]) \u003e 0 and not IsString(log.cache[\"remote_host\"])","merge_maps(log.cache, ExtractPatterns(log.cache[\"location\"], \"^(?:(?\u003cfunc_name\u003e[^,]+), )?(?\u003cfile_name\u003e[^:]+):(?\u003cfile_line_num\u003e\\\\d+)$\"), \"insert\") where Len(log.cache[\"location\"]) \u003e 0","set(log.cache[\"cursor_position\"], Double(log.cache[\"cursor_position\"])) where IsMatch(log.cache[\"cursor_position\"], \"^[0-9.]+$\")","set(log.cache[\"file_line_num\"], Double(log.cache[\"file_line_num\"])) where IsMatch(log.cache[\"file_line_num\"], \"^[0-9.]+$\")","set(log.cache[\"internal_position\"], Double(log.cache[\"internal_position\"])) where IsMatch(log.cache[\"internal_position\"], \"^[0-9.]+$\")","set(log.cache[\"leader_pid\"], Double(log.cache[\"leader_pid\"])) where IsMatch(log.cache[\"leader_pid\"], \"^[0-9.]+$\")","set(log.cache[\"line_num\"], Double(log.cache[\"line_num\"])) where IsMatch(log.cache[\"line_num\"], \"^[0-9.]+$\")","set(log.cache[\"pid\"], Double(log.cache[\"pid\"])) where IsMatch(log.cache[\"pid\"], \"^[0-9.]+$\")","set(log.cache[\"query_id\"], Double(log.cache[\"query_id\"])) where IsMatch(log.cache[\"query_id\"], \"^[0-9.]+$\")","set(log.cache[\"remote_port\"], Double(log.cache[\"remote_port\"])) where IsMatch(log.cache[\"remote_port\"], \"^[0-9.]+$\")","set(log.body[\"parsed\"], log.cache)"]},{"statements":["set(instrumentation_scope.name, \"postgres\")","set(instrumentation_scope.version, resource.attributes[\"db.version\"])","set(log.cache, log.body[\"parsed\"]) where log.body[\"format\"] == \"csv\"","set(log.cache, ParseJSON(log.body[\"original\"])) where log.body[\"format\"] == \"json\"","set(log.severity_text, log.cache[\"error_severity\"])","set(log.severity_number, SEVERITY_NUMBER_TRACE)  where log.severity_text == \"DEBUG5\"","set(log.severity_number, SEVERITY_NUMBER_TRACE2) where log.severity_text == \"DEBUG4\"","set(log.severity_number, SEVERITY_NUMBER_TRACE3) where log.severity_text == \"DEBUG3\"","set(log.severity_number, SEVERITY_NUMBER_TRACE4) where log.severity_text == \"DEBUG2\"","set(log.severity_number, SEVERITY_NUMBER_DEBUG)  where log.severity_text == \"DEBUG1\"","set(log.severity_number, SEVERITY_NUMBER_INFO)   where log.severity_text == \"INFO\" or log.severity_text == \"LOG\"","set(log.severity_number, SEVERITY_NUMBER_INFO2)  where log.severity_text == \"NOTICE\"","set(log.severity_number, SEVERITY_NUMBER_WARN)   where log.severity_text == \"WARNING\"","set(log.severity_number, SEVERITY_NUMBER_ERROR)  where log.severity_text == \"ERROR\"","set(log.severity_number, SEVERITY_NUMBER_FATAL)  where log.severity_text == \"FATAL\"","set(log.severity_number, SEVERITY_NUMBER_FATAL2) where log.severity_text == \"PANIC\"","set(log.time, Time(log.cache[\"timestamp\"], \"%F %T.%L %Z\")) where IsString(log.cache[\"timestamp\"])","set(instrumentation_scope.schema_url, \"https://opentelemetry.io/schemas/1.29.0\")","set(resource.attributes[\"db.system\"], \"postgresql\")","set(log.attributes[\"log.record.original\"], log.body[\"original\"])","set(log.body, log.cache)","set(log.attributes[\"client.address\"],  log.body[\"remote_host\"])  where IsString(log.body[\"remote_host\"])","set(log.attributes[\"client.port\"], Int(log.body[\"remote_port\"])) where IsDouble(log.body[\"remote_port\"])","set(log.attributes[\"code.filepath\"], log.body[\"file_name\"]) where IsString(log.body[\"file_name\"])","set(log.attributes[\"code.function\"], log.body[\"func_name\"]) where IsString(log.body[\"func_name\"])","set(log.attributes[\"code.lineno\"], Int(log.body[\"file_line_num\"])) where IsDouble(log.body[\"file_line_num\"])","set(log.attributes[\"db.namespace\"], log.body[\"dbname\"]) where IsString(log.body[\"dbname\"])","set(log.attributes[\"db.response.status_code\"], log.body[\"state_code\"]) where IsString(log.body[\"state_code\"])","set(log.attributes[\"process.creation.time\"], Concat([ Substring(log.body[\"session_start\"], 0, 10), \"T\", Substring(log.body[\"session_start\"], 11, 8), \"Z\"], \"\")) where IsMatch(log.body[\"session_start\"], \"^[^ ]{10} [^ ]{8} UTC$\")","set(log.attributes[\"process.pid\"], Int(log.body[\"pid\"])) where IsDouble(log.body[\"pid\"])","set(log.attributes[\"process.title\"], log.body[\"ps\"]) where IsString(log.body[\"ps\"])","set(log.attributes[\"user.name\"], log.body[\"user\"]) where IsString(log.body[\"user\"])"]},{"conditions":["Len(body[\"message\"]) \u003e 7 and Substring(body[\"message\"], 0, 7) == \"AUDIT: \""],"statements":["set(log.body[\"pgaudit\"], ParseCSV(Substring(log.body[\"message\"], 7, Len(log.body[\"message\"]) - 7), \"audit_type,statement_id,substatement_id,class,command,object_type,object_name,statement,parameter\", delimiter=\",\", mode=\"strict\"))","set(instrumentation_scope.name, \"pgaudit\") where Len(log.body[\"pgaudit\"]) \u003e 0"]}]
+[{"conditions":["log.body[\"format\"] == \"csv\""],"statements":["set(log.cache, ParseCSV(log.body[\"original\"], log.body[\"headers\"], delimiter=\",\", mode=\"strict\"))","merge_maps(log.cache, ExtractPatterns(log.cache[\"connection_from\"], \"(?:^[[]local[]]:(?\u003cremote_port\u003e.+)|:(?\u003cremote_port\u003e[^:]+))$\"), \"insert\") where Len(log.cache[\"connection_from\"]) \u003e 0","set(log.cache[\"remote_host\"], Substring(log.cache[\"connection_from\"], 0, Len(log.cache[\"connection_from\"]) - Len(log.cache[\"remote_port\"]) - 1)) where Len(log.cache[\"connection_from\"]) \u003e 0 and IsString(log.cache[\"remote_port\"])","set(log.cache[\"remote_host\"], log.cache[\"connection_from\"]) where Len(log.cache[\"connection_from\"]) \u003e 0 and not IsString(log.cache[\"remote_host\"])","merge_maps(log.cache, ExtractPatterns(log.cache[\"location\"], \"^(?:(?\u003cfunc_name\u003e[^,]+), )?(?\u003cfile_name\u003e[^:]+):(?\u003cfile_line_num\u003e\\\\d+)$\"), \"insert\") where Len(log.cache[\"location\"]) \u003e 0","set(log.cache[\"cursor_position\"], Double(log.cache[\"cursor_position\"])) where IsMatch(log.cache[\"cursor_position\"], \"^[0-9.]+$\")","set(log.cache[\"file_line_num\"], Double(log.cache[\"file_line_num\"])) where IsMatch(log.cache[\"file_line_num\"], \"^[0-9.]+$\")","set(log.cache[\"internal_position\"], Double(log.cache[\"internal_position\"])) where IsMatch(log.cache[\"internal_position\"], \"^[0-9.]+$\")","set(log.cache[\"leader_pid\"], Double(log.cache[\"leader_pid\"])) where IsMatch(log.cache[\"leader_pid\"], \"^[0-9.]+$\")","set(log.cache[\"line_num\"], Double(log.cache[\"line_num\"])) where IsMatch(log.cache[\"line_num\"], \"^[0-9.]+$\")","set(log.cache[\"pid\"], Double(log.cache[\"pid\"])) where IsMatch(log.cache[\"pid\"], \"^[0-9.]+$\")","set(log.cache[\"query_id\"], Double(log.cache[\"query_id\"])) where IsMatch(log.cache[\"query_id\"], \"^[0-9.]+$\")","set(log.cache[\"remote_port\"], Double(log.cache[\"remote_port\"])) where IsMatch(log.cache[\"remote_port\"], \"^[0-9.]+$\")","set(log.body[\"parsed\"], log.cache)"]},{"statements":["set(instrumentation_scope.name, \"postgres\")","set(instrumentation_scope.version, resource.attributes[\"db.version\"])","set(log.cache, log.body[\"parsed\"]) where log.body[\"format\"] == \"csv\"","set(log.cache, ParseJSON(log.body[\"original\"])) where log.body[\"format\"] == \"json\"","set(log.severity_text, log.cache[\"error_severity\"])","set(log.severity_number, SEVERITY_NUMBER_TRACE)  where log.severity_text == \"DEBUG5\"","set(log.severity_number, SEVERITY_NUMBER_TRACE2) where log.severity_text == \"DEBUG4\"","set(log.severity_number, SEVERITY_NUMBER_TRACE3) where log.severity_text == \"DEBUG3\"","set(log.severity_number, SEVERITY_NUMBER_TRACE4) where log.severity_text == \"DEBUG2\"","set(log.severity_number, SEVERITY_NUMBER_DEBUG)  where log.severity_text == \"DEBUG1\"","set(log.severity_number, SEVERITY_NUMBER_INFO)   where log.severity_text == \"INFO\" or log.severity_text == \"LOG\"","set(log.severity_number, SEVERITY_NUMBER_INFO2)  where log.severity_text == \"NOTICE\"","set(log.severity_number, SEVERITY_NUMBER_WARN)   where log.severity_text == \"WARNING\"","set(log.severity_number, SEVERITY_NUMBER_ERROR)  where log.severity_text == \"ERROR\"","set(log.severity_number, SEVERITY_NUMBER_FATAL)  where log.severity_text == \"FATAL\"","set(log.severity_number, SEVERITY_NUMBER_FATAL2) where log.severity_text == \"PANIC\"","set(log.time, Time(log.cache[\"timestamp\"], \"%F %T.%L %Z\")) where IsString(log.cache[\"timestamp\"])","set(instrumentation_scope.schema_url, \"https://opentelemetry.io/schemas/1.29.0\")","set(resource.attributes[\"db.system\"], \"postgresql\")","set(log.attributes[\"log.record.original\"], log.body[\"original\"])","set(log.body, log.cache)","set(log.attributes[\"client.address\"],  log.body[\"remote_host\"])  where IsString(log.body[\"remote_host\"])","set(log.attributes[\"client.port\"], Int(log.body[\"remote_port\"])) where IsDouble(log.body[\"remote_port\"])","set(log.attributes[\"code.filepath\"], log.body[\"file_name\"]) where IsString(log.body[\"file_name\"])","set(log.attributes[\"code.function\"], log.body[\"func_name\"]) where IsString(log.body[\"func_name\"])","set(log.attributes[\"code.lineno\"], Int(log.body[\"file_line_num\"])) where IsDouble(log.body[\"file_line_num\"])","set(log.attributes[\"db.namespace\"], log.body[\"dbname\"]) where IsString(log.body[\"dbname\"])","set(log.attributes[\"db.response.status_code\"], log.body[\"state_code\"]) where IsString(log.body[\"state_code\"])","set(log.attributes[\"process.creation.time\"], Concat([ Substring(log.body[\"session_start\"], 0, 10), \"T\", Substring(log.body[\"session_start\"], 11, 8), \"Z\"], \"\")) where IsMatch(log.body[\"session_start\"], \"^[^ ]{10} [^ ]{8} UTC$\")","set(log.attributes[\"process.pid\"], Int(log.body[\"pid\"])) where IsDouble(log.body[\"pid\"])","set(log.attributes[\"process.title\"], log.body[\"ps\"]) where IsString(log.body[\"ps\"])","set(log.attributes[\"user.name\"], log.body[\"user\"]) where IsString(log.body[\"user\"])"]},{"conditions":["Len(log.body[\"message\"]) \u003e 7 and Substring(log.body[\"message\"], 0, 7) == \"AUDIT: \""],"statements":["set(log.body[\"pgaudit\"], ParseCSV(Substring(log.body[\"message\"], 7, Len(log.body[\"message\"]) - 7), \"audit_type,statement_id,substatement_id,class,command,object_type,object_name,statement,parameter\", delimiter=\",\", mode=\"strict\"))","set(instrumentation_scope.name, \"pgaudit\") where Len(log.body[\"pgaudit\"]) \u003e 0"]}]
diff --git a/internal/collector/postgres_logs_transforms.yaml b/internal/collector/postgres_logs_transforms.yaml
index c8178f2d6..c58f1a1a7 100644
--- a/internal/collector/postgres_logs_transforms.yaml
+++ b/internal/collector/postgres_logs_transforms.yaml
@@ -8,7 +8,7 @@
 # TODO(postgres-14): We can stop parsing CSV logs when 14 is EOL.
 # https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/-/pkg/ottl/contexts/ottllog#readme
 - conditions:
-    - body["format"] == "csv"
+    - log.body["format"] == "csv"
   statements:
     # https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/-/pkg/ottl/ottlfuncs#parsecsv
     - set(log.cache, ParseCSV(log.body["original"], log.body["headers"], delimiter=",", mode="strict"))
@@ -196,7 +196,7 @@
     # https://github.com/pgaudit/pgaudit/blame/17.0/pgaudit.c#L876
     # TODO(postgres-18): Check this prefix and update the URL above.
     - >-
-      Len(body["message"]) > 7 and Substring(body["message"], 0, 7) == "AUDIT: "
+      Len(log.body["message"]) > 7 and Substring(log.body["message"], 0, 7) == "AUDIT: "
   statements:
     # https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/-/pkg/ottl/ottlfuncs#parsecsv
     - >-
diff --git a/internal/collector/postgres_test.go b/internal/collector/postgres_test.go
index 83deb349a..222b263e2 100644
--- a/internal/collector/postgres_test.go
+++ b/internal/collector/postgres_test.go
@@ -123,7 +123,7 @@ processors:
   transform/postgres_logs:
     log_statements:
     - conditions:
-      - body["format"] == "csv"
+      - log.body["format"] == "csv"
       statements:
       - set(log.cache, ParseCSV(log.body["original"], log.body["headers"], delimiter=",",
         mode="strict"))
@@ -203,8 +203,8 @@ processors:
       - set(log.attributes["process.title"], log.body["ps"]) where IsString(log.body["ps"])
       - set(log.attributes["user.name"], log.body["user"]) where IsString(log.body["user"])
     - conditions:
-      - 'Len(body["message"]) > 7 and Substring(body["message"], 0, 7) == "AUDIT:
-        "'
+      - 'Len(log.body["message"]) > 7 and Substring(log.body["message"], 0, 7) ==
+        "AUDIT: "'
       statements:
       - set(log.body["pgaudit"], ParseCSV(Substring(log.body["message"], 7, Len(log.body["message"])
         - 7), "audit_type,statement_id,substatement_id,class,command,object_type,object_name,statement,parameter",
@@ -383,7 +383,7 @@ processors:
   transform/postgres_logs:
     log_statements:
     - conditions:
-      - body["format"] == "csv"
+      - log.body["format"] == "csv"
       statements:
       - set(log.cache, ParseCSV(log.body["original"], log.body["headers"], delimiter=",",
         mode="strict"))
@@ -463,8 +463,8 @@ processors:
       - set(log.attributes["process.title"], log.body["ps"]) where IsString(log.body["ps"])
       - set(log.attributes["user.name"], log.body["user"]) where IsString(log.body["user"])
     - conditions:
-      - 'Len(body["message"]) > 7 and Substring(body["message"], 0, 7) == "AUDIT:
-        "'
+      - 'Len(log.body["message"]) > 7 and Substring(log.body["message"], 0, 7) ==
+        "AUDIT: "'
       statements:
       - set(log.body["pgaudit"], ParseCSV(Substring(log.body["message"], 7, Len(log.body["message"])
         - 7), "audit_type,statement_id,substatement_id,class,command,object_type,object_name,statement,parameter",