timsaucer commented on code in PR #1678:
URL:
https://github.com/apache/datafusion-python/pull/1678#discussion_r3936188655
##########
crates/core/src/codec.rs:
##########
@@ -321,30 +724,54 @@ impl LogicalExtensionCodec for PythonLogicalCodec {
node: Arc<dyn TableProvider>,
buf: &mut Vec<u8>,
) -> Result<()> {
- self.inner.try_encode_table_provider(table_ref, node, buf)
+ chain_encode(
+ &self.chain,
+ &self.terminal,
+ buf,
+ "a table provider",
+ |codec, buf| codec.try_encode_table_provider(table_ref,
Arc::clone(&node), buf),
+ )
}
fn try_decode_file_format(
&self,
buf: &[u8],
ctx: &TaskContext,
) -> Result<Arc<dyn FileFormatFactory>> {
- self.inner.try_decode_file_format(buf, ctx)
+ chain_decode(
+ &self.chain,
+ &self.terminal,
+ buf,
+ "a file format",
+ |codec, buf| codec.try_decode_file_format(buf, ctx),
+ )
}
fn try_encode_file_format(
&self,
buf: &mut Vec<u8>,
node: Arc<dyn FileFormatFactory>,
) -> Result<()> {
- self.inner.try_encode_file_format(buf, node)
+ chain_encode(
+ &self.chain,
+ &self.terminal,
+ buf,
+ "a file format",
+ |codec, buf| codec.try_encode_file_format(buf, Arc::clone(&node)),
+ )
}
fn try_encode_udf(&self, node: &ScalarUDF, buf: &mut Vec<u8>) ->
Result<()> {
if self.python_udf_inlining && try_encode_python_scalar_udf(node,
buf)? {
Review Comment:
I opened a follow up issue. great idea!
##########
crates/core/src/codec.rs:
##########
@@ -223,12 +233,339 @@ fn strip_wire_header<'a>(
Ok(Some(&buf[py_minor_idx + 1..]))
}
+/// Family prefix for the envelope wrapping a chained codec's payload.
+///
+/// A distinct magic is what makes "is this framed?" a definite test
+/// rather than a speculative decode. Probing by attempting to parse the
+/// envelope would reintroduce exactly the protobuf ambiguity this
+/// framing exists to remove: prost skips unknown fields and defaults
+/// missing ones, so a foreign payload can parse cleanly as an envelope.
+pub(crate) const CHAINED_PAYLOAD_FAMILY: &[u8] = b"DFPYCHN";
+
+/// Wire-format version for the chained-payload envelope. Independent of
+/// [`WIRE_VERSION_CURRENT`], which versions the cloudpickle framing.
+pub(crate) const CHAIN_WIRE_VERSION_CURRENT: u8 = 1;
+
+/// Oldest chained-payload envelope version this build decodes.
+pub(crate) const CHAIN_WIRE_VERSION_MIN_SUPPORTED: u8 = 1;
+
+/// Prefix for the synthetic id given to a codec installed from a bare
+/// PyCapsule, which exposes nothing stable to derive an identity from.
+/// The rest of the id is random per install, so no other session can
+/// mint it: a payload carrying one decodes within the installing
+/// session's lineage, which clones the id along with the chain, and
+/// fails with a pointed error anywhere else rather than resolving to a
+/// different codec. A counter or a chain position would not do — every
+/// session numbers from the same end, so the first bare capsule
+/// installed anywhere would answer for every other session's first.
Review Comment:
Tried to clean it up.
--
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]
