hubcio commented on code in PR #4739:
URL: https://github.com/apache/iggy/pull/4739#discussion_r4469033687
##########
core/connectors/sinks/iceberg_sink/src/router/static_router.rs:
##########
@@ -64,33 +67,35 @@ impl StaticRouter {
error!("No valid tables found. Can't initiate Iceberg connector");
return Err(Error::InvalidConfig);
}
- Ok(StaticRouter { tables, catalog })
+ Ok(StaticRouter {
+ tables,
+ catalog,
+ invalid_rows,
+ })
}
}
#[async_trait]
impl Router for StaticRouter {
async fn route_data(
&self,
- messages_metadata: MessagesMetadata,
+ origin: &BatchOrigin<'_>,
+ messages_schema: Schema,
messages: Vec<ConsumedMessage>,
) -> Result<(), crate::Error> {
- let data: Vec<Payload> = messages
- .into_iter()
- .map(|m: ConsumedMessage| m.payload)
- .collect();
-
for table in &self.tables {
write_data(
- &data,
+ origin,
+ &messages,
table,
self.catalog.as_ref(),
- messages_metadata.schema,
+ messages_schema,
+ self.invalid_rows,
)
.await?;
info!(
"Routed {} messages to iceberg table {} successfully",
- data.len(),
+ messages.len(),
Review Comment:
warning: both routers report every input row as successfully routed when
`invalid_rows = "skip"` discards rows, including batches with no commit. label
both counts as input rows, or log the number of committed rows.
##########
core/connectors/sinks/iceberg_sink/src/router/mod.rs:
##########
@@ -230,6 +258,64 @@ async fn write_data_files(
Ok(data_files)
}
+/// 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,
+/// origin and table and left out, and the others are decoded.
+fn decode_rows(
+ origin: &BatchOrigin<'_>,
+ table: &TableIdent,
+ 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} ({origin}) for table:
{table}, 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}");
+ })
+}
+
+/// The reader allocates its buffers for a full batch, so the batch size is
the row count: a row
+/// checked on its own is read without the buffers of a default-sized batch.
+fn read_batches(
+ schema: &SchemaRef,
+ rows: &[&simd_json::OwnedValue],
+) -> Result<Vec<RecordBatch>, Error> {
Review Comment:
simplification: this helper returns a vector although its batch size covers
all supplied rows. return `Option<RecordBatch>` through
`reader.next().transpose()`, preserve error conversion, and use `if let` at the
writer.
##########
core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs:
##########
@@ -28,20 +29,18 @@ use tracing::{info, warn};
pub struct DynamicRouter {
catalog: Box<dyn Catalog>,
route_field: String,
+ invalid_rows: InvalidRows,
}
+/// Each table the run routes to, keyed by route field value, with the
messages routed to it.
pub struct DynamicWriter {
Review Comment:
simplification: `DynamicWriter` now wraps one map and exists only within
`DynamicRouter::route_data`. use a local map and its `Entry` API in
`route_data`.
--
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]
