timsaucer commented on code in PR #24973:
URL: https://github.com/apache/datafusion/pull/24973#discussion_r4066390065
##########
datafusion/proto/src/physical_plan/mod.rs:
##########
@@ -799,6 +799,92 @@ mod tests {
}
}
+ /// Codec that rejects function serde outright, as a codec which only
+ /// knows its own plan nodes does.
+ #[derive(Debug)]
+ struct RejectsFunctionsCodec;
Review Comment:
🤖 review:
RejectsFunctionsCodec doesn't do what its name says. It only overrides
try_encode_udwf; try_encode_udf and try_encode_udaf fall through to the Ok(())
default, meaning it silently accepts those. In
composed_codec_keeps_by_name_encoding_after_a_rejecting_codec that happens not
to matter, but the name will mislead the next person. Either implement all the
function hooks as errors or rename to RejectsUdwfCodec.
##########
datafusion/proto/src/physical_plan/mod.rs:
##########
@@ -2093,6 +2334,56 @@ impl ComposedPhysicalExtensionCodec {
.encode(buf)
.map_err(|e| internal_datafusion_err!("{e}"))
}
+
+ /// Like [`Self::encode_protobuf`], but for the function hooks whose trait
+ /// default is `Ok(())` rather than an error.
+ ///
+ /// Those hooks treat an empty buffer as "no custom payload, encode by
+ /// name", and the decode side only consults the function registry when the
+ /// payload is absent. Wrapping an empty blob in a [`DataEncoderTuple`]
+ /// would make a by-name function look codec-encoded and strand it at
+ /// decode time, so a codec that writes nothing must leave `buf` untouched.
+ fn encode_protobuf_by_name_aware(
+ &self,
+ buf: &mut Vec<u8>,
+ mut encode: impl FnMut(&dyn PhysicalExtensionCodec, &mut Vec<u8>) ->
Result<()>,
+ ) -> Result<()> {
Review Comment:
This function feels duplicative. I am concerned that it could lead to drift
over time requiring maintenance of two call sites that are almost identical.
What about something along the lines of
```
fn encode_with(
&self,
buf: &mut Vec<u8>,
encode_by_name_on_empty: bool,
mut encode: impl FnMut(&dyn PhysicalExtensionCodec, &mut Vec<u8>) ->
Result<()>,
) -> Result<()> {
// ...
Ok(()) => {
accepted = true;
if !encode_by_name_on_empty || !data.is_empty() {
encoder_position = Some(position as u32);
break;
}
}
// ...
}
```
##########
datafusion/proto/src/physical_plan/mod.rs:
##########
@@ -2093,6 +2334,56 @@ impl ComposedPhysicalExtensionCodec {
.encode(buf)
.map_err(|e| internal_datafusion_err!("{e}"))
}
+
+ /// Like [`Self::encode_protobuf`], but for the function hooks whose trait
Review Comment:
🤖 review note:
Position semantics now differ per hook, and new()'s doc doesn't say so. The
comment says "Position in this codecs list is important... If new codec is
added it should go to last position." With the by-name hooks, a codec that
accepts without writing no longer claims its position — a later codec can.
That's a real behavioral contract for anyone assembling a codec list, and it
should be in the new() doc, not only in a private helper's doc comment.
##########
datafusion/proto/src/physical_plan/mod.rs:
##########
@@ -2093,6 +2334,56 @@ impl ComposedPhysicalExtensionCodec {
.encode(buf)
.map_err(|e| internal_datafusion_err!("{e}"))
}
+
+ /// Like [`Self::encode_protobuf`], but for the function hooks whose trait
+ /// default is `Ok(())` rather than an error.
+ ///
+ /// Those hooks treat an empty buffer as "no custom payload, encode by
+ /// name", and the decode side only consults the function registry when the
+ /// payload is absent. Wrapping an empty blob in a [`DataEncoderTuple`]
+ /// would make a by-name function look codec-encoded and strand it at
+ /// decode time, so a codec that writes nothing must leave `buf` untouched.
+ fn encode_protobuf_by_name_aware(
+ &self,
+ buf: &mut Vec<u8>,
+ mut encode: impl FnMut(&dyn PhysicalExtensionCodec, &mut Vec<u8>) ->
Result<()>,
+ ) -> Result<()> {
Review Comment:
The idea for ^ was mine, but my agent suggested a test along the lines of
```
#[test]
fn composed_codec_by_name_decode_reaches_later_codecs() -> Result<()> {
let composed = ComposedPhysicalExtensionCodec::new(vec![
Arc::new(RejectsFunctionsCodec), // position 0, cannot resolve
Arc::new(EmptyPayloadOnlyCodec), // position 1, resolves by name
]);
assert_eq!(composed.try_decode_udwf("test_udwf", &[])?.name(),
"test_udwf");
Ok(())
}
```
I haven't verified this though.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
