scovich commented on code in PR #8100:
URL: https://github.com/apache/arrow-rs/pull/8100#discussion_r2265096937
##########
arrow-avro/src/reader/mod.rs:
##########
@@ -162,29 +171,42 @@ impl Decoder {
///
/// Returns the number of bytes consumed.
pub fn decode(&mut self, data: &[u8]) -> Result<usize, ArrowError> {
- if self.expect_prefix
- && data.len() >= SINGLE_OBJECT_MAGIC.len()
- && !data.starts_with(&SINGLE_OBJECT_MAGIC)
- {
- return Err(ArrowError::ParseError(
- "Expected single‑object encoding fingerprint prefix for first
message \
- (writer_schema_store is set but active_fingerprint is None)"
- .into(),
- ));
- }
let mut total_consumed = 0usize;
- // The loop stops when the batch is full, a schema change is staged,
- // or handle_prefix indicates we need more bytes (Some(0)).
while total_consumed < data.len() && self.remaining_capacity > 0 {
- if let Some(n) = self.handle_prefix(&data[total_consumed..])? {
- // We either consumed a prefix (n > 0) and need a schema
switch, or we need
- // more bytes to make a decision. Either way, this decoding
attempt is finished.
- total_consumed += n;
+ if !self.awaiting_body {
+ if let Some(n) = self.handle_prefix(&data[total_consumed..])? {
+ if n == 0 {
+ break;
+ }
+ total_consumed += n;
+ self.awaiting_body = true;
+ if self.remaining_capacity == self.batch_size &&
self.pending_schema.is_some() {
+ self.apply_pending_schema_if_batch_empty();
+ }
Review Comment:
The callees already (re-)check both these conditions...
```suggestion
self.apply_pending_schema_if_batch_empty();
```
##########
arrow-avro/src/reader/mod.rs:
##########
@@ -282,6 +310,31 @@ impl Decoder {
pub fn batch_is_full(&self) -> bool {
self.remaining_capacity == 0
}
+
+ // Decode either the block count of remaining capacity from `data` (an OCF
block payload).
Review Comment:
```suggestion
// Decode either the block count or remaining capacity from `data` (an
OCF block payload).
```
?
##########
arrow-avro/src/reader/mod.rs:
##########
@@ -162,29 +171,42 @@ impl Decoder {
///
/// Returns the number of bytes consumed.
pub fn decode(&mut self, data: &[u8]) -> Result<usize, ArrowError> {
- if self.expect_prefix
- && data.len() >= SINGLE_OBJECT_MAGIC.len()
- && !data.starts_with(&SINGLE_OBJECT_MAGIC)
- {
- return Err(ArrowError::ParseError(
- "Expected single‑object encoding fingerprint prefix for first
message \
- (writer_schema_store is set but active_fingerprint is None)"
- .into(),
- ));
- }
let mut total_consumed = 0usize;
- // The loop stops when the batch is full, a schema change is staged,
- // or handle_prefix indicates we need more bytes (Some(0)).
while total_consumed < data.len() && self.remaining_capacity > 0 {
- if let Some(n) = self.handle_prefix(&data[total_consumed..])? {
- // We either consumed a prefix (n > 0) and need a schema
switch, or we need
- // more bytes to make a decision. Either way, this decoding
attempt is finished.
- total_consumed += n;
+ if !self.awaiting_body {
+ if let Some(n) = self.handle_prefix(&data[total_consumed..])? {
+ if n == 0 {
+ break;
+ }
+ total_consumed += n;
+ self.awaiting_body = true;
+ if self.remaining_capacity == self.batch_size &&
self.pending_schema.is_some() {
+ self.apply_pending_schema_if_batch_empty();
+ }
+ if self.remaining_capacity == 0 {
+ break;
+ }
+ }
+ }
+ match self.active_decoder.decode(&data[total_consumed..], 1) {
+ Ok(n) if n > 0 => {
+ self.remaining_capacity -= 1;
+ total_consumed += n;
+ self.awaiting_body = false;
Review Comment:
Is there always a fingerprint after each record? Or just a chance to see a
fingerprint?
##########
arrow-avro/src/reader/mod.rs:
##########
@@ -162,29 +171,42 @@ impl Decoder {
///
/// Returns the number of bytes consumed.
pub fn decode(&mut self, data: &[u8]) -> Result<usize, ArrowError> {
- if self.expect_prefix
- && data.len() >= SINGLE_OBJECT_MAGIC.len()
- && !data.starts_with(&SINGLE_OBJECT_MAGIC)
- {
- return Err(ArrowError::ParseError(
- "Expected single‑object encoding fingerprint prefix for first
message \
- (writer_schema_store is set but active_fingerprint is None)"
- .into(),
- ));
- }
let mut total_consumed = 0usize;
- // The loop stops when the batch is full, a schema change is staged,
- // or handle_prefix indicates we need more bytes (Some(0)).
while total_consumed < data.len() && self.remaining_capacity > 0 {
- if let Some(n) = self.handle_prefix(&data[total_consumed..])? {
- // We either consumed a prefix (n > 0) and need a schema
switch, or we need
- // more bytes to make a decision. Either way, this decoding
attempt is finished.
- total_consumed += n;
+ if !self.awaiting_body {
+ if let Some(n) = self.handle_prefix(&data[total_consumed..])? {
+ if n == 0 {
+ break;
Review Comment:
```suggestion
break; // insufficient bytes
```
##########
arrow-avro/src/reader/mod.rs:
##########
@@ -162,29 +171,42 @@ impl Decoder {
///
/// Returns the number of bytes consumed.
pub fn decode(&mut self, data: &[u8]) -> Result<usize, ArrowError> {
- if self.expect_prefix
- && data.len() >= SINGLE_OBJECT_MAGIC.len()
- && !data.starts_with(&SINGLE_OBJECT_MAGIC)
- {
- return Err(ArrowError::ParseError(
- "Expected single‑object encoding fingerprint prefix for first
message \
- (writer_schema_store is set but active_fingerprint is None)"
- .into(),
- ));
- }
let mut total_consumed = 0usize;
- // The loop stops when the batch is full, a schema change is staged,
- // or handle_prefix indicates we need more bytes (Some(0)).
while total_consumed < data.len() && self.remaining_capacity > 0 {
- if let Some(n) = self.handle_prefix(&data[total_consumed..])? {
- // We either consumed a prefix (n > 0) and need a schema
switch, or we need
- // more bytes to make a decision. Either way, this decoding
attempt is finished.
- total_consumed += n;
+ if !self.awaiting_body {
+ if let Some(n) = self.handle_prefix(&data[total_consumed..])? {
+ if n == 0 {
+ break;
+ }
+ total_consumed += n;
+ self.awaiting_body = true;
+ if self.remaining_capacity == self.batch_size &&
self.pending_schema.is_some() {
+ self.apply_pending_schema_if_batch_empty();
+ }
+ if self.remaining_capacity == 0 {
+ break;
+ }
+ }
+ }
+ match self.active_decoder.decode(&data[total_consumed..], 1) {
+ Ok(n) if n > 0 => {
+ self.remaining_capacity -= 1;
+ total_consumed += n;
+ self.awaiting_body = false;
+ }
+ Ok(_) => {
+ return Err(ArrowError::ParseError(
+ "Record decoder consumed 0 bytes".into(),
Review Comment:
I thought zero-byte records were legal, and we're supposed to keep looping
until the output batch is full?
##########
arrow-avro/src/reader/mod.rs:
##########
@@ -162,29 +171,42 @@ impl Decoder {
///
/// Returns the number of bytes consumed.
pub fn decode(&mut self, data: &[u8]) -> Result<usize, ArrowError> {
- if self.expect_prefix
- && data.len() >= SINGLE_OBJECT_MAGIC.len()
- && !data.starts_with(&SINGLE_OBJECT_MAGIC)
- {
- return Err(ArrowError::ParseError(
- "Expected single‑object encoding fingerprint prefix for first
message \
- (writer_schema_store is set but active_fingerprint is None)"
- .into(),
- ));
- }
let mut total_consumed = 0usize;
- // The loop stops when the batch is full, a schema change is staged,
- // or handle_prefix indicates we need more bytes (Some(0)).
while total_consumed < data.len() && self.remaining_capacity > 0 {
- if let Some(n) = self.handle_prefix(&data[total_consumed..])? {
- // We either consumed a prefix (n > 0) and need a schema
switch, or we need
- // more bytes to make a decision. Either way, this decoding
attempt is finished.
- total_consumed += n;
+ if !self.awaiting_body {
+ if let Some(n) = self.handle_prefix(&data[total_consumed..])? {
+ if n == 0 {
+ break;
+ }
+ total_consumed += n;
+ self.awaiting_body = true;
+ if self.remaining_capacity == self.batch_size &&
self.pending_schema.is_some() {
+ self.apply_pending_schema_if_batch_empty();
+ }
+ if self.remaining_capacity == 0 {
+ break;
Review Comment:
... but this whole control flow is a bit confusing. I wonder if it can be
made easier to understand?
```rust
while total_consumed < data.len() && self.remaining_capacity > 0 {
if self.awaiting_body {
let result = match self.active_decoder.decode(...) {
Ok(0) => Err(...),
Ok(n) => {
...
self.awaiting_body = false;
continue; // re-check input buffer before attempting to read
the row
}
Err(ref e) if is_incomplete_data(e) => break,
err => err,
}
return result;
}
match self.handle_prefix(...)? {
Some(0) => break, // insufficient bytes
Some(n) => {
total_consumed += n;
self.apply_pending_schema_if_batch_empty();
self.awaiting_body = true;
}
}
}
```
##########
arrow-avro/src/reader/mod.rs:
##########
@@ -282,6 +310,31 @@ impl Decoder {
pub fn batch_is_full(&self) -> bool {
self.remaining_capacity == 0
}
+
+ // Decode either the block count of remaining capacity from `data` (an OCF
block payload).
+ //
+ // Returns the number of bytes consumed from `data` along with the number
of records decoded.
+ fn decode_block(&mut self, data: &[u8], count: usize) -> Result<(usize,
usize), ArrowError> {
Review Comment:
Two same-type return values with very different meanings... is it worth
defining a struct for it so they have names? Or are there few enough (and
always internal) callers to keep track of it?
##########
arrow-avro/src/reader/mod.rs:
##########
@@ -162,29 +171,42 @@ impl Decoder {
///
/// Returns the number of bytes consumed.
pub fn decode(&mut self, data: &[u8]) -> Result<usize, ArrowError> {
- if self.expect_prefix
- && data.len() >= SINGLE_OBJECT_MAGIC.len()
- && !data.starts_with(&SINGLE_OBJECT_MAGIC)
- {
- return Err(ArrowError::ParseError(
- "Expected single‑object encoding fingerprint prefix for first
message \
- (writer_schema_store is set but active_fingerprint is None)"
- .into(),
- ));
- }
let mut total_consumed = 0usize;
- // The loop stops when the batch is full, a schema change is staged,
- // or handle_prefix indicates we need more bytes (Some(0)).
while total_consumed < data.len() && self.remaining_capacity > 0 {
- if let Some(n) = self.handle_prefix(&data[total_consumed..])? {
- // We either consumed a prefix (n > 0) and need a schema
switch, or we need
- // more bytes to make a decision. Either way, this decoding
attempt is finished.
- total_consumed += n;
+ if !self.awaiting_body {
+ if let Some(n) = self.handle_prefix(&data[total_consumed..])? {
+ if n == 0 {
+ break;
+ }
+ total_consumed += n;
+ self.awaiting_body = true;
+ if self.remaining_capacity == self.batch_size &&
self.pending_schema.is_some() {
+ self.apply_pending_schema_if_batch_empty();
+ }
+ if self.remaining_capacity == 0 {
+ break;
Review Comment:
```suggestion
break; // pending schema change ends the batch
```
##########
arrow-avro/src/reader/mod.rs:
##########
@@ -162,29 +171,42 @@ impl Decoder {
///
/// Returns the number of bytes consumed.
pub fn decode(&mut self, data: &[u8]) -> Result<usize, ArrowError> {
- if self.expect_prefix
- && data.len() >= SINGLE_OBJECT_MAGIC.len()
- && !data.starts_with(&SINGLE_OBJECT_MAGIC)
- {
- return Err(ArrowError::ParseError(
- "Expected single‑object encoding fingerprint prefix for first
message \
- (writer_schema_store is set but active_fingerprint is None)"
- .into(),
- ));
- }
let mut total_consumed = 0usize;
- // The loop stops when the batch is full, a schema change is staged,
- // or handle_prefix indicates we need more bytes (Some(0)).
while total_consumed < data.len() && self.remaining_capacity > 0 {
- if let Some(n) = self.handle_prefix(&data[total_consumed..])? {
- // We either consumed a prefix (n > 0) and need a schema
switch, or we need
- // more bytes to make a decision. Either way, this decoding
attempt is finished.
- total_consumed += n;
+ if !self.awaiting_body {
+ if let Some(n) = self.handle_prefix(&data[total_consumed..])? {
+ if n == 0 {
+ break;
+ }
+ total_consumed += n;
+ self.awaiting_body = true;
+ if self.remaining_capacity == self.batch_size &&
self.pending_schema.is_some() {
+ self.apply_pending_schema_if_batch_empty();
+ }
+ if self.remaining_capacity == 0 {
+ break;
+ }
+ }
+ }
+ match self.active_decoder.decode(&data[total_consumed..], 1) {
+ Ok(n) if n > 0 => {
+ self.remaining_capacity -= 1;
+ total_consumed += n;
+ self.awaiting_body = false;
+ }
+ Ok(_) => {
+ return Err(ArrowError::ParseError(
+ "Record decoder consumed 0 bytes".into(),
+ ));
+ }
+ Err(e) => {
+ return if is_incomplete_data(&e) {
+ Ok(total_consumed)
+ } else {
+ Err(e)
+ }
Review Comment:
nit: if/else inside match arms is often bulky and hard to read:
```suggestion
Err(ref e) if is_incomplete_data(&e) => return
Ok(total_consumed),
err => return err,
```
--
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]
