fix: fixed merge timestamp_key for nodes that get messages by take() on user code

src/caret_analyze/infra/lttng/records_provider_lttng.py

@@ -921,12 +921,10 @@ def _compose_inter_proc_comm_records(
             columns.append(COLUMN_NAME.RCL_PUBLISH_TIMESTAMP)
         if COLUMN_NAME.DDS_WRITE_TIMESTAMP in records.columns:
             columns.append(COLUMN_NAME.DDS_WRITE_TIMESTAMP)
-        columns.append(COLUMN_NAME.SOURCE_TIMESTAMP)
-
-        sub_records = self._source.sub_records(callback_object, None)
-        is_take_node = len(sub_records) == 0
-        if not is_take_node:
-            columns.append(COLUMN_NAME.CALLBACK_START_TIMESTAMP)
+        columns += [
+            COLUMN_NAME.SOURCE_TIMESTAMP,
+            COLUMN_NAME.CALLBACK_START_TIMESTAMP,
+        ]
 
         self._format(records, columns)
 
@@ -1379,9 +1377,15 @@ def fill_source_timestamp_with_latest_timestamp(records):
             f'{self._node_path.publish_topic_name}/rclcpp_publish_timestamp',
         ]
         left_key = sub_records.columns[0]
-        if COLUMN_NAME.RMW_TAKE_TIMESTAMP in columns:
-            columns.remove(COLUMN_NAME.RMW_TAKE_TIMESTAMP)
-            left_key = COLUMN_NAME.RMW_TAKE_TIMESTAMP
+
+        # Set left_key to rmw_take timestamp
+        # if sub_records are obtained by RecordsProviderLttng.subscription_take_records()
+        if is_take_node:
+            for column in sub_records.columns:
+                if column.endswith(COLUMN_NAME.RMW_TAKE_TIMESTAMP):
+                    columns.remove(column)
+                    left_key = column
+                    break
 
         pub_sub_records = merge_sequential(
             left_records=sub_records,

src/caret_analyze/runtime/path.py

@@ -163,6 +163,12 @@ def is_match_column(column: str, target_name: str) -> bool:
                 right_records)
             right_records.rename_columns(rename_rule)
 
+            # remove columns from included in communications not via callback
+            if is_match_column(right_records.columns[-1], 'callback_start_timestamp') and \
+                    all(x is None for x
+                        in right_records.get_column_series(right_records.columns[-1])):
+                right_records.drop_columns([right_records.columns[-1]])
+
             if left_records.columns[-1] != right_records.columns[0]:
                 raise InvalidRecordsError('left columns[-1] != right columns[0]')
 

src/test/infra/lttng/test_latency_definitions.py

@@ -1381,6 +1381,7 @@ def test_inter_proc_empty_data(
             columns=[
                 f'{communication.topic_name}/rclcpp_publish_timestamp',
                 f'{communication.topic_name}/source_timestamp',
+                f'{callback.callback_name}/callback_start_timestamp',
             ],
             dtype='Int64'
         )

src/test/infra/lttng/test_records_provider_lttng.py

@@ -608,18 +608,18 @@ def noop(*args, **kwargs):
         take_records_data = [
             RecordCppImpl({
                 f'{topic_name_1}/{COLUMN_NAME.SOURCE_TIMESTAMP}': 1,
-                COLUMN_NAME.RMW_TAKE_TIMESTAMP: 2,
+                f'{topic_name_1}/{COLUMN_NAME.RMW_TAKE_TIMESTAMP}': 2,
             }),
             RecordCppImpl({
                 f'{topic_name_1}/{COLUMN_NAME.SOURCE_TIMESTAMP}': 6,
-                COLUMN_NAME.RMW_TAKE_TIMESTAMP: 7,
+                f'{topic_name_1}/{COLUMN_NAME.RMW_TAKE_TIMESTAMP}': 7,
             }),
         ]
         take_records = RecordsCppImpl(
             take_records_data,
             [
                 ColumnValue(f'{topic_name_1}/{COLUMN_NAME.SOURCE_TIMESTAMP}'),
-                ColumnValue(COLUMN_NAME.RMW_TAKE_TIMESTAMP),
+                ColumnValue(f'{topic_name_1}/{COLUMN_NAME.RMW_TAKE_TIMESTAMP}'),
             ]
         )