feat(source) truncate long value to 1024 in log.
// See the License for the specific language governing permissions and
// limitations under the License.
+use std::borrow::Cow;
+
use async_trait::async_trait;
use common_base::tokio::io::AsyncBufRead;
use common_base::tokio::io::AsyncBufReadExt;
@@ -75,6 +77,19 @@ where R: AsyncBufRead + Unpin + Send
}
}
+fn maybe_truncated(s: &str, limit: usize) -> Cow<'_, str> {
+ if s.len() > limit {
+ Cow::Owned(format!(
+ "(first {}B of {}B): {}",
+ limit,
+ s.len(),
+ &s[..limit]
+ ))
+ } else {
+ Cow::Borrowed(s)
+ }
+}
+
#[async_trait]
impl<R> Source for NDJsonSource<R>
where R: AsyncBufRead + Unpin + Send
@@ -127,13 +142,14 @@ where R: AsyncBufRead + Unpin + Send
for ((name, type_name), deser) in fields.iter().zip(packs.iter_mut()) {
let value = &json[name];
deser.de_json(value).map_err(|e| {
+ let value_str = format!("{:?}", value);
ErrorCode::BadBytes(format!(
- "error at row {} column {}: type={}, value={:?}: {}",
+ "error at row {} column {}: type={}, err={}, value={}",
rows,
name,
type_name,
- value,
- e.message()
+ e.message(),
+ maybe_truncated(&value_str, 1024),
))
})?;
}
