better error messages in connectors (#6422)

GitOrigin-RevId: fda4716c96d0d1f6017175596f100a3395619b73

python/pathway/tests/test_errors.py

@@ -1149,10 +1149,10 @@ class InputSchema(pw.Schema):
     expected_errors = T(
         """
         message
-        failed to parse field "b" with type Int from the following json payload: "x"
-        failed to parse field "b" with type Int from the following json payload: "1"
-        failed to parse field "c" with type Int / None from the following json payload: "t"
-        failed to parse field "c" with type Int / None from the following json payload: "y"
+        failed to create a field "b" with type Int from the following json payload: "x"
+        failed to create a field "b" with type Int from the following json payload: "1"
+        failed to create a field "c" with type Int / None from the following json payload: "t"
+        failed to create a field "c" with type Int / None from the following json payload: "y"
     """,
         split_on_whitespace=False,
     ).select(message=pw.this.message.str.replace("/", "|"))
@@ -1188,11 +1188,11 @@ class InputSchema(pw.Schema):
     expected_errors = T(
         """
         message
-        error in primary key, skipping the row: failed to parse field "b" \
+        error in primary key, skipping the row: failed to create a field "b" \
 with type Int from the following json payload: "x"
-        error in primary key, skipping the row: failed to parse field "b" \
+        error in primary key, skipping the row: failed to create a field "b" \
 with type Int from the following json payload: "1"
-        failed to parse field "c" with type Int / None from the following json payload: "y"
+        failed to create a field "c" with type Int / None from the following json payload: "y"
     """,
         split_on_whitespace=False,
     ).select(message=pw.this.message.str.replace("/", "|"))

src/connectors/data_format.rs

@@ -120,6 +120,15 @@ impl From<ParsedEvent> for ParsedEventWithErrors {
     }
 }
 
+fn limit_length(mut s: String, max_length: usize) -> String {
+    if s.len() > max_length {
+        s.truncate(max_length);
+        s + "..."
+    } else {
+        s
+    }
+}
+
 #[derive(Debug, thiserror::Error)]
 #[non_exhaustive]
 pub enum ParseError {
@@ -140,7 +149,7 @@ pub enum ParseError {
     #[error("too small number of csv tokens in the line: {0}")]
     UnexpectedNumberOfCsvTokens(usize),
 
-    #[error("failed to parse field {field_name:?} with type {type_:?}{} from the following json payload: {payload}", if *is_optional {" | None"} else {""})]
+    #[error("failed to create a field {field_name:?} with type {type_:?}{} from the following json payload: {}", if *is_optional {" | None"} else {""}, limit_length(format!("{payload}"), 500))]
     FailedToParseFromJson {
         field_name: String,
         payload: JsonValue,

src/connectors/mod.rs

@@ -760,8 +760,13 @@ impl Connector {
             let entry = match entry.remove_errors(&error_handling_logic) {
                 Ok(entry) => entry,
                 Err(err) => {
-                    // if there is an error in key
-                    self.log_parse_error(ParseError::ErrorInKey(err).into());
+                    let err = if self.skip_all_errors {
+                        err
+                    } else {
+                        // if there is an error in key
+                        ParseError::ErrorInKey(err).into()
+                    };
+                    self.log_parse_error(err);
                     continue;
                 }
             };

tests/integration/test_jsonlines.rs

@@ -403,7 +403,7 @@ fn test_jsonlines_failed_to_parse_field() -> eyre::Result<()> {
     assert_error_shown(
         Box::new(reader),
         Box::new(parser),
-        r#"failed to parse field "pet" with type Any from the following json payload: {"animal":"dog","measurements":[200,400,600],"name":"Alice"}"#,
+        r#"failed to create a field "pet" with type Any from the following json payload: {"animal":"dog","measurements":[200,400,600],"name":"Alice"}"#,
         ErrorPlacement::Value(0),
     );