better error handling in ix (#7704)

GitOrigin-RevId: a689209bd8ad20c929821506f415f24217a142af

Author: KamilPiechowiak

CHANGELOG.md

@@ -7,6 +7,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ### Changed
 - values of non-deterministic UDFs are not stored in tables that are `append_only`.
+- `pw.Table.ix` has better runtime error message that includes id of the missing row.
 
 ### Fixed
 - temporal behaviors in temporal operators (`windowby`, `interval_join`) now consume no CPU when no data passes through them.

python/pathway/tests/test_errors.py

@@ -617,6 +617,49 @@ def test_with_universe_of():
     )
 
 
+def test_ix():
+    t1 = pw.debug.table_from_markdown(
+        """
+          | a
+        1 | 1
+        2 | 3
+        3 | 2
+        4 | 2
+    """
+    ).with_columns(ap=pw.this.pointer_from(pw.this.a))
+
+    t2 = pw.debug.table_from_markdown(
+        """
+          | c
+        1 | 10
+        2 | 13
+    """
+    )
+    res = t1.select(pw.this.a, c=t2.ix(pw.this.ap).c)
+    res = res.select(pw.this.a, c=pw.fill_error(res.c, -1))
+    expected = pw.debug.table_from_markdown(
+        """
+          | a |  c
+        1 | 1 | 10
+        2 | 3 | -1
+        3 | 2 | 13
+        5 | 2 | 13
+    """
+    )
+    expected_errors = T(
+        """
+        message
+        key missing in output table: ^Z3QWT294JQSHPSR8KTPG9ECE4W
+        """,
+        split_on_whitespace=False,
+    )
+    assert_table_equality_wo_index(
+        (res, pw.global_error_log().select(pw.this.message)),
+        (expected, expected_errors),
+        terminate_on_error=False,
+    )
+
+
 def test_remove_errors():
     t1 = T(
         """

src/engine/dataflow.rs

@@ -1021,24 +1021,6 @@ impl<S: MaybeTotalScope> DataflowGraphInner<S> {
         self.static_column(universe_handle, Vec::new(), column_properties)
     }
 
-    #[track_caller]
-    fn assert_collections_same_size<V1: Data, V2: Data>(
-        &self,
-        left: &Collection<S, V1>,
-        right: &Collection<S, V2>,
-    ) -> Result<()> {
-        let error_logger = self.create_error_logger()?;
-        left.map_named("assert_collections_same_size::left", |_| ())
-            .negate()
-            .concat(&right.map_named("assert_collections_same_size::right", |_| ()))
-            .consolidate()
-            .inspect(move |(_data, _time, diff)| {
-                assert_ne!(diff, &0);
-                error_logger.log_error(DataError::ValueMissing);
-            });
-        Ok(())
-    }
-
     #[track_caller]
     fn assert_input_keys_match_output_keys(
         &self,
@@ -1064,6 +1046,38 @@ impl<S: MaybeTotalScope> DataflowGraphInner<S> {
         Ok(())
     }
 
+    fn make_output_keys_match_input_keys(
+        &self,
+        input_values: &Values<S>,
+        output_collection: &Collection<S, (Key, Value)>,
+    ) -> Result<Collection<S, (Key, Value)>> {
+        let leftover_values = input_values.concat(
+            &output_collection
+                .map_named(
+                    "restrict_or_override_table_universe::compare",
+                    |(key, values)| {
+                        (
+                            key,
+                            values.as_tuple().expect("values should be a tuple")[0].clone(),
+                        )
+                    },
+                )
+                .distinct()
+                .negate(),
+        );
+        let error_logger = self.create_error_logger()?;
+
+        Ok(
+            output_collection.concat(&leftover_values.consolidate().map_named(
+                "restrict_or_override_table_universe::fill",
+                move |(key, new_values)| {
+                    error_logger.log_error(DataError::KeyMissingInOutputTable(key));
+                    (key, Value::from([new_values, Value::Error].as_slice()))
+                },
+            )),
+        )
+    }
+
     fn static_universe(&mut self, keys: Vec<Key>) -> Result<UniverseHandle> {
         let worker_count = self.scope.peers();
         let worker_index = self.scope.index();
@@ -1701,29 +1715,7 @@ impl<S: MaybeTotalScope> DataflowGraphInner<S> {
                 ))
             },
         );
-        let leftover_values = new_table.values().concat(
-            &result
-                .map_named(
-                    "restrict_or_override_table_universe::compare",
-                    |(key, values)| {
-                        (
-                            key,
-                            values.as_tuple().expect("values should be a tuple")[0].clone(),
-                        )
-                    },
-                )
-                .distinct()
-                .negate(),
-        );
-        let error_logger = self.create_error_logger()?;
-
-        let result = result.concat(&leftover_values.consolidate().map_named(
-            "restrict_or_override_table_universe::fill",
-            move |(key, new_values)| {
-                error_logger.log_error(DataError::KeyMissingInOutputTable(key));
-                (key, Value::from([new_values, Value::Error].as_slice()))
-            },
-        ));
+        let result = self.make_output_keys_match_input_keys(new_table.values(), &result)?;
 
         if !self.ignore_asserts && same_universes {
             self.assert_input_keys_match_output_keys(original_table.keys(), &result)?;
@@ -2249,8 +2241,10 @@ impl<S: MaybeTotalScope> DataflowGraphInner<S> {
             }
             _ => new_table,
         };
-        if !self.ignore_asserts && ix_key_policy != IxKeyPolicy::SkipMissing {
-            self.assert_collections_same_size(key_table.values(), &new_table)?;
+        let new_table = if ix_key_policy == IxKeyPolicy::SkipMissing {
+            new_table
+        } else {
+            self.make_output_keys_match_input_keys(key_table.values(), &new_table)?
         };
 
         Ok(self