hubcio commented on code in PR #4739:
URL: https://github.com/apache/iggy/pull/4739#discussion_r4468647759
##########
core/connectors/sinks/iceberg_sink/src/router/mod.rs:
##########
@@ -232,6 +229,59 @@ async fn write_data_files(
/// Files finalized by a failed batch are never committed, so they are removed
/// instead of being left as orphans on the object store.
+/// Decodes the JSON documents into record batches of `schema`. A row that
does not match fails
+/// the whole batch, unless `invalid_rows` skips it: then each such row is
logged with its offset
+/// and left out, and the others are decoded.
+fn decode_rows(
+ schema: &SchemaRef,
+ documents: &[(u64, Cow<'_, simd_json::OwnedValue>)],
+ invalid_rows: InvalidRows,
+) -> Result<Vec<RecordBatch>, Error> {
+ let rows: Vec<&simd_json::OwnedValue> = documents
+ .iter()
+ .map(|(_, document)| document.as_ref())
+ .collect();
+ let error = match read_batches(schema, &rows) {
+ Ok(batches) => return Ok(batches),
+ Err(error) => error,
+ };
+ if invalid_rows == InvalidRows::Fail {
+ error!("Error while getting record batch: {error}");
+ return Err(error);
+ }
+ // The batch read only says that some row does not match, so each row is
read on its own.
+ let mut valid = Vec::with_capacity(rows.len());
+ for (offset, document) in documents {
+ match read_batches(schema, &[document.as_ref()]) {
+ Ok(_) => valid.push(document.as_ref()),
+ Err(row_error) => warn!(
Review Comment:
warning: skipped-row warnings omit source and table identifiers, so repeated
offsets across partitions leave operators unable to locate discarded rows.
include connector, stream, topic, partition, and destination table identifiers.
##########
core/connectors/sinks/iceberg_sink/src/router/mod.rs:
##########
@@ -232,6 +229,59 @@ async fn write_data_files(
/// Files finalized by a failed batch are never committed, so they are removed
/// instead of being left as orphans on the object store.
+/// Decodes the JSON documents into record batches of `schema`. A row that
does not match fails
+/// the whole batch, unless `invalid_rows` skips it: then each such row is
logged with its offset
+/// and left out, and the others are decoded.
+fn decode_rows(
+ schema: &SchemaRef,
+ documents: &[(u64, Cow<'_, simd_json::OwnedValue>)],
+ invalid_rows: InvalidRows,
+) -> Result<Vec<RecordBatch>, Error> {
+ let rows: Vec<&simd_json::OwnedValue> = documents
+ .iter()
+ .map(|(_, document)| document.as_ref())
+ .collect();
+ let error = match read_batches(schema, &rows) {
+ Ok(batches) => return Ok(batches),
+ Err(error) => error,
+ };
+ if invalid_rows == InvalidRows::Fail {
+ error!("Error while getting record batch: {error}");
+ return Err(error);
+ }
+ // The batch read only says that some row does not match, so each row is
read on its own.
+ let mut valid = Vec::with_capacity(rows.len());
+ for (offset, document) in documents {
+ match read_batches(schema, &[document.as_ref()]) {
+ Ok(_) => valid.push(document.as_ref()),
+ Err(row_error) => warn!(
+ "Skipping the row at offset: {offset}, which does not match
the table schema. {row_error}"
+ ),
+ }
+ }
+ if valid.is_empty() {
+ return Ok(Vec::new());
+ }
+ read_batches(schema, &valid).inspect_err(|error| {
+ error!("Error while getting record batch: {error}");
+ })
+}
+
+fn read_batches(
+ schema: &SchemaRef,
+ rows: &[&simd_json::OwnedValue],
+) -> Result<Vec<RecordBatch>, Error> {
+ let reader = ReaderBuilder::new(Arc::clone(schema))
+ .build(JsonArrowReader::new(rows))
+ .map_err(|err| {
+ error!("Error while building Iceberg reader from message payload:
{err}");
+ Error::InitError(err.to_string())
+ })?;
+ reader
+ .map(|batch| batch.map_err(|err|
Error::InvalidRecordValue(format_error_chain(&err))))
+ .collect()
Review Comment:
warning: `read_batches` retains every decoded batch before the first write,
increasing peak memory for larger batches in both modes. decode and write
bounded chunks while preserving cleanup and row recovery.
##########
core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs:
##########
@@ -28,6 +29,7 @@ use tracing::{info, warn};
pub struct DynamicRouter {
catalog: Box<dyn Catalog>,
route_field: String,
+ invalid_rows: InvalidRows,
}
pub struct DynamicWriter {
Review Comment:
simplification: the existing `DynamicWriter` keeps identical table keys in
two maps. store each table and its message vector together in one map.
##########
core/connectors/sinks/iceberg_sink/src/router/mod.rs:
##########
@@ -232,6 +229,59 @@ async fn write_data_files(
/// Files finalized by a failed batch are never committed, so they are removed
/// instead of being left as orphans on the object store.
+/// Decodes the JSON documents into record batches of `schema`. A row that
does not match fails
+/// the whole batch, unless `invalid_rows` skips it: then each such row is
logged with its offset
+/// and left out, and the others are decoded.
+fn decode_rows(
+ schema: &SchemaRef,
+ documents: &[(u64, Cow<'_, simd_json::OwnedValue>)],
+ invalid_rows: InvalidRows,
+) -> Result<Vec<RecordBatch>, Error> {
+ let rows: Vec<&simd_json::OwnedValue> = documents
+ .iter()
+ .map(|(_, document)| document.as_ref())
+ .collect();
+ let error = match read_batches(schema, &rows) {
+ Ok(batches) => return Ok(batches),
+ Err(error) => error,
+ };
+ if invalid_rows == InvalidRows::Fail {
+ error!("Error while getting record batch: {error}");
+ return Err(error);
+ }
+ // The batch read only says that some row does not match, so each row is
read on its own.
+ let mut valid = Vec::with_capacity(rows.len());
+ for (offset, document) in documents {
+ match read_batches(schema, &[document.as_ref()]) {
Review Comment:
warning: the per-row fallback allocates buffers for a full batch for each
isolated row after a schema error. set the reader batch size to one for
single-row validation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
