[SPARK-34205][SQL][SS] Add pipe to Dataset to enable Streaming Dataset pipe

What changes were proposed in this pull request?

This patch proposes to add pipe API to Dataset to enable pipe feature for streaming Dataset.
Why are the changes needed?

Dataset doesn't have pipe API but RDD has it. Although for normal Dataset, user can convert a Dataset to RDD and call RDD.pipe and then convert it back to a Dataset, for streaming Dataset it is not possible. So that being said, this is actually a requirement from Structured Streaming, but we need to add pipe API to Dataset to enable it in Structured Streaming. From Dataset perspective, it is also easier to use Dataset API instead of converting between Dataset and RDD.
Does this PR introduce any user-facing change?

Yes, a new API pipe is added to Dataset.
How was this patch tested?

Unit test.

Test build #134378 has started for PR 31296 at commit d4f9457.

Test build #134378 has started for PR 31296 at commit d4f9457.

cc @xuanyuanking @HeartSaVioR @cloud-fan

cc @xuanyuanking @HeartSaVioR @cloud-fan

Thank you @dongjoon-hyun.

Thank you @dongjoon-hyun.

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38965/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38965/

Test build #134378 has finished for PR 31296 at commit d4f9457.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class PipeElements[T](
case class PipeElementsExec(

Test build #134378 has finished for PR 31296 at commit d4f9457.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class PipeElements[T](
case class PipeElementsExec(

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/134378/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/134378/

Test build #134386 has started for PR 31296 at commit c6e57c1.

Test build #134386 has started for PR 31296 at commit c6e57c1.

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38972/

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38972/

Test build #134389 has started for PR 31296 at commit ff0da84.

Test build #134389 has started for PR 31296 at commit ff0da84.

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38972/

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38972/

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38974/

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38974/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38972/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38972/

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38974/

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38974/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38974/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38974/

Test build #134386 has finished for PR 31296 at commit c6e57c1.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test build #134386 has finished for PR 31296 at commit c6e57c1.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/134386/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/134386/

Test build #134389 has finished for PR 31296 at commit ff0da84.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test build #134389 has finished for PR 31296 at commit ff0da84.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/134389/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/134389/

Could we change the description a little bit:
for streaming Dataset it is not possible since queries with streaming sources must be executed with writeStream.start().

Could we change the description a little bit:
for streaming Dataset it is not possible since queries with streaming sources must be executed with writeStream.start().

Could you please describe the actual use case? I would like to confirm this works with complicated schema like array/map/nested struct with some binary columns, and for the case how forked process can deserialize inputs properly and applies operations and serialize again (that wouldn't matter much as the return is Dataset[String] though).

Could you please describe the actual use case? I would like to confirm this works with complicated schema like array/map/nested struct with some binary columns, and for the case how forked process can deserialize inputs properly and applies operations and serialize again (that wouldn't matter much as the return is Dataset[String] though).

Could you please describe the actual use case? I would like to confirm this works with complicated schema like array/map/nested struct with some binary columns, and for the case how forked process can deserialize inputs properly and applies operations and serialize again (that wouldn't matter much as the return is Dataset[String] though).

Our internal client needs to pipe streaming read from Kafka through a forked process, and currently with SS users cannot do it. I think the above question also applied to RDD.pipe. It's users' responsibility to make sure the process can understand the input.

In addition, this would incur non-trivial serde cost on communication between Spark process and external process. Probably we also need to revisit which benefits this gives to us compared to what Spark provides now (UDF, or some others if I miss?).

I believe there are some cases that users have to use pipe. Users should consider it before choosing and using pipe API. I think it is well known issue for pipe. The point is, when users need to use pipe on streaming data like RDD and batch Dataset, but streaming Dataset cannot support it.

Could you please describe the actual use case? I would like to confirm this works with complicated schema like array/map/nested struct with some binary columns, and for the case how forked process can deserialize inputs properly and applies operations and serialize again (that wouldn't matter much as the return is Dataset[String] though).

Our internal client needs to pipe streaming read from Kafka through a forked process, and currently with SS users cannot do it. I think the above question also applied to RDD.pipe. It's users' responsibility to make sure the process can understand the input.

In addition, this would incur non-trivial serde cost on communication between Spark process and external process. Probably we also need to revisit which benefits this gives to us compared to what Spark provides now (UDF, or some others if I miss?).

I believe there are some cases that users have to use pipe. Users should consider it before choosing and using pipe API. I think it is well known issue for pipe. The point is, when users need to use pipe on streaming data like RDD and batch Dataset, but streaming Dataset cannot support it.

Could we change the description a little bit:
for streaming Dataset it is not possible since queries with streaming sources must be executed with writeStream.start().

Sure, thanks.

Could we change the description a little bit:
for streaming Dataset it is not possible since queries with streaming sources must be executed with writeStream.start().

Sure, thanks.

Agree with @HeartSaVioR here. I think it is more general if we just provide a UDF for this... it also doesn’t pollute the Dataset API with something so rarely used.

Agree with @HeartSaVioR here. I think it is more general if we just provide a UDF for this... it also doesn’t pollute the Dataset API with something so rarely used.

Agree with @HeartSaVioR here. I think it is more general if we just provide a UDF for this... it also doesn’t pollute the Dataset API with something so rarely used.

I'm not sure I understand your point here. So you mean when users have to use pipe, it is simply to ask them to write UDF instead of using pipe?

Agree with @HeartSaVioR here. I think it is more general if we just provide a UDF for this... it also doesn’t pollute the Dataset API with something so rarely used.

I'm not sure I understand your point here. So you mean when users have to use pipe, it is simply to ask them to write UDF instead of using pipe?

Our internal client needs to pipe streaming read from Kafka through a forked process, and currently with SS users cannot do it. I think the above question also applied to RDD.pipe. It's users' responsibility to make sure the process can understand the input.

Yes the question is also applied to RDD.pipe as well, but the serialization is done via OutputStreamWriter.println which is relatively "known"  - String.valueOf(T) and print it out. Easy to reason about, though it'd show bad performance and tricky to deserialize if toString is implemented human friendly.
Here we are serializing via Encoder which is like a black-box to others (Have we documented how encoder encodes the object?) so that can't be users' responsibility. And once we do this, the approach to encode will also become a kind of public API.

Our internal client needs to pipe streaming read from Kafka through a forked process, and currently with SS users cannot do it. I think the above question also applied to RDD.pipe. It's users' responsibility to make sure the process can understand the input.

Yes the question is also applied to RDD.pipe as well, but the serialization is done via OutputStreamWriter.println which is relatively "known"  - String.valueOf(T) and print it out. Easy to reason about, though it'd show bad performance and tricky to deserialize if toString is implemented human friendly.
Here we are serializing via Encoder which is like a black-box to others (Have we documented how encoder encodes the object?) so that can't be users' responsibility. And once we do this, the approach to encode will also become a kind of public API.

Yes the question is also applied to RDD.pipe as well, but the serialization is done via OutputStreamWriter.println which is relatively "known" - String.valueOf(T) and print it out. Easy to reason about, though it'd show bad performance and tricky to deserialize if toString is implemented human friendly.
Here we are serializing via Encoder which is like a black-box to others (Have we documented how encoder encodes the object?) so that can't be users' responsibility. And once we do this, the approach to encode will also become a kind of public API.

Encoder is the how we serialize external data to Dataset. Why this is an issue? This doesn't invent any new stuff. Encoder is nothing related to how we pipe here. This follows all other typed Dataset API (map, foreach...) approach to use Encoder to deserialize internal row to domain object T. Then it follows RDD pipe API to print T out to forked process.

Yes the question is also applied to RDD.pipe as well, but the serialization is done via OutputStreamWriter.println which is relatively "known" - String.valueOf(T) and print it out. Easy to reason about, though it'd show bad performance and tricky to deserialize if toString is implemented human friendly.
Here we are serializing via Encoder which is like a black-box to others (Have we documented how encoder encodes the object?) so that can't be users' responsibility. And once we do this, the approach to encode will also become a kind of public API.

Encoder is the how we serialize external data to Dataset. Why this is an issue? This doesn't invent any new stuff. Encoder is nothing related to how we pipe here. This follows all other typed Dataset API (map, foreach...) approach to use Encoder to deserialize internal row to domain object T. Then it follows RDD pipe API to print T out to forked process.

So what is the domain object T for PipeRDD when your type in Dataset is complicated? Can you reason about the output format of T.toString which PipeRDD will send to the external process? Is the output format known to end users?
If you'd like to propose a public API then you'll need to be a first user on the API at least. The easiest way to prove your theory is to create a project which can communicate with the API. In the project you'll need to show the forked process can deal with such complicated type, and also prove the deserialization is "known one".

So what is the domain object T for PipeRDD when your type in Dataset is complicated? Can you reason about the output format of T.toString which PipeRDD will send to the external process? Is the output format known to end users?
If you'd like to propose a public API then you'll need to be a first user on the API at least. The easiest way to prove your theory is to create a project which can communicate with the API. In the project you'll need to show the forked process can deal with such complicated type, and also prove the deserialization is "known one".

Agree with @HeartSaVioR here. I think it is more general if we just provide a UDF for this... it also doesn’t pollute the Dataset API with something so rarely used.

I'm not sure I understand your point here. So you mean when users have to use pipe, it is simply to ask them to write UDF instead of using pipe?

No. My suggestion is to create a built in expression that can do that, rather than having a top level api called pipe. The expression would also allow users to choose the fields they use and how to serialize / deserialize.

Agree with @HeartSaVioR here. I think it is more general if we just provide a UDF for this... it also doesn’t pollute the Dataset API with something so rarely used.

I'm not sure I understand your point here. So you mean when users have to use pipe, it is simply to ask them to write UDF instead of using pipe?

No. My suggestion is to create a built in expression that can do that, rather than having a top level api called pipe. The expression would also allow users to choose the fields they use and how to serialize / deserialize.

Yes the question is also applied to RDD.pipe as well, but the serialization is done via OutputStreamWriter.println which is relatively "known" - String.valueOf(T) and print it out. Easy to reason about, though it'd show bad performance and tricky to deserialize if toString is implemented human friendly.

Here we are serializing via Encoder which is like a black-box to others (Have we documented how encoder encodes the object?) so that can't be users' responsibility. And once we do this, the approach to encode will also become a kind of public API.

Encoder is the how we serialize external data to Dataset. Why this is an issue? This doesn't invent any new stuff. Encoder is nothing related to how we pipe here. This follows all other typed Dataset API (map, foreach...) approach to use Encoder to deserialize internal row to domain object T. Then it follows RDD pipe API to print T out to forked process.
So that being said, if you already have a Dataset, there is already Encoder use to serialize it. Using pipe with Dataset is just like RDD pipe.

It is an issue because encoder only specifies how an object would map to the internal physical structure of the row, and by exposing this pipe API, we are exposing the structure (serialization) itself.
We can never change this serialization anymore.

Yes the question is also applied to RDD.pipe as well, but the serialization is done via OutputStreamWriter.println which is relatively "known" - String.valueOf(T) and print it out. Easy to reason about, though it'd show bad performance and tricky to deserialize if toString is implemented human friendly.

Here we are serializing via Encoder which is like a black-box to others (Have we documented how encoder encodes the object?) so that can't be users' responsibility. And once we do this, the approach to encode will also become a kind of public API.

Encoder is the how we serialize external data to Dataset. Why this is an issue? This doesn't invent any new stuff. Encoder is nothing related to how we pipe here. This follows all other typed Dataset API (map, foreach...) approach to use Encoder to deserialize internal row to domain object T. Then it follows RDD pipe API to print T out to forked process.
So that being said, if you already have a Dataset, there is already Encoder use to serialize it. Using pipe with Dataset is just like RDD pipe.

It is an issue because encoder only specifies how an object would map to the internal physical structure of the row, and by exposing this pipe API, we are exposing the structure (serialization) itself.
We can never change this serialization anymore.

It is an issue because encoder only specifies how an object would map to the internal physical structure of the row, and by exposing this pipe API, we are exposing the structure (serialization) itself.
We can never change this serialization anymore.

So Dataset.map also exposes it? The pipe API only outputs the domain object T as string to forked process. The forked process doesn't touch the physical structure. Just like Dataset.map outputs domain object T to user-provided function. The user-provided function doesn't know about physical structure.

It is an issue because encoder only specifies how an object would map to the internal physical structure of the row, and by exposing this pipe API, we are exposing the structure (serialization) itself.
We can never change this serialization anymore.

So Dataset.map also exposes it? The pipe API only outputs the domain object T as string to forked process. The forked process doesn't touch the physical structure. Just like Dataset.map outputs domain object T to user-provided function. The user-provided function doesn't know about physical structure.

It is an issue because encoder only specifies how an object would map to the internal physical structure of the row, and by exposing this pipe API, we are exposing the structure (serialization) itself.

We can never change this serialization anymore.

So Dataset.map also exposes it? The pipe API only outputs the domain object T as string to forked process. The forked process doesn't know the physical structure. Just like Dataset.map outputs domain object T to user-provided function. The user-provided function doesn't know about physical structure.

Map doesn't expose it because object of type T is just passed into the map function.
Pipe would need to serialize T in order to pass it to the different process, wouldn't it? I think that's what @HeartSaVioR was trying to say.

It is an issue because encoder only specifies how an object would map to the internal physical structure of the row, and by exposing this pipe API, we are exposing the structure (serialization) itself.

We can never change this serialization anymore.

So Dataset.map also exposes it? The pipe API only outputs the domain object T as string to forked process. The forked process doesn't know the physical structure. Just like Dataset.map outputs domain object T to user-provided function. The user-provided function doesn't know about physical structure.

Map doesn't expose it because object of type T is just passed into the map function.
Pipe would need to serialize T in order to pass it to the different process, wouldn't it? I think that's what @HeartSaVioR was trying to say.

Please just create an executable which prints out stdin and passes to the pipe API... I think it's the easiest way to realize it.

Please just create an executable which prints out stdin and passes to the pipe API... I think it's the easiest way to realize it.

Map doesn't expose it because object of type T is just passed into the map function.
Pipe would need to serialize T in order to pass it to the different process, wouldn't it? I think that's what @HeartSaVioR was trying to say.

No~ As I mentioned many times above, Dataset.pipe works like RDD.pipe. For RDD.pipe, there is one parameter printRDDElement: (T, String => Unit) => Unit, which is similar to the map function and takes the domain object T. The function takes the object T and uses the second parameter, a print out function, to print the string data. So it is basically like Dataset.map. Here is I'm confused why @HeartSaVioR continues to mention that point.

Map doesn't expose it because object of type T is just passed into the map function.
Pipe would need to serialize T in order to pass it to the different process, wouldn't it? I think that's what @HeartSaVioR was trying to say.

No~ As I mentioned many times above, Dataset.pipe works like RDD.pipe. For RDD.pipe, there is one parameter printRDDElement: (T, String => Unit) => Unit, which is similar to the map function and takes the domain object T. The function takes the object T and uses the second parameter, a print out function, to print the string data. So it is basically like Dataset.map. Here is I'm confused why @HeartSaVioR continues to mention that point.

Please just create an executable which prints out stdin (serialized data) and passes to the pipe API... I think it's the easiest way to realize.

Ok, ok. I didn't like the make the pipe API take too many parameter like printRDDElement at beginning. So let me add it and create a test to take complicated type.

Please just create an executable which prints out stdin (serialized data) and passes to the pipe API... I think it's the easiest way to realize.

Ok, ok. I didn't like the make the pipe API take too many parameter like printRDDElement at beginning. So let me add it and create a test to take complicated type.

I'm sorry, now I understand how this approach avoids such issue. This converts InternalRow to object before passing to PipedRDD... At least that would work the same with PipedRDD if the type is NOT a representation of Spark and preferably custom one. I'd concern about the performance perspective, but I totally agree this is the same how PipedRDD does.
It's still a question which is the better approach, top level of API vs built-in, as the API haven't existed for such a long time (the lifetime of 2.x) and no one might have been complained for that.

I'm sorry, now I understand how this approach avoids such issue. This converts InternalRow to object before passing to PipedRDD... At least that would work the same with PipedRDD if the type is NOT a representation of Spark and preferably custom one. I'd concern about the performance perspective, but I totally agree this is the same how PipedRDD does.
It's still a question which is the better approach, top level of API vs built-in, as the API haven't existed for such a long time (the lifetime of 2.x) and no one might have been complained for that.

I can think of a few ways to make this less visible. I can't tell how useful this is in general.

Just define a function for Dataset[String] and/or Dataset[Binary/byte array]. They can appear more fluent using implicits.
Use a UDF/expression to do this, using the mapPartitions API ...

This can even be done without being added to Spark...

I can think of a few ways to make this less visible. I can't tell how useful this is in general.

Just define a function for Dataset[String] and/or Dataset[Binary/byte array]. They can appear more fluent using implicits.
Use a UDF/expression to do this, using the mapPartitions API ...

This can even be done without being added to Spark...

For the performance perspective, I think we are not going to change some currently existing API so it is not a performance regression. User may understand the performance meaning when using pipe. Actually pipe is the way if there is no other choices. In some cases user have to use it, e.g. they're unable to replace the forked process.
I'm not sure if someone ever has been complained for it. For batch Dataset, it is easy to convert to RDD and run RDD.pipe. For streaming one, it is simply impossible. Maybe it is also related to SS adaptation rate?
UDF/expression not work because pipe is not 1-to-1 relation between input and output. For example piping through wc -l gets single output per partition.
There will be some issues if just invoking forked process in mapPartitions API, e.g. interrupting stdin writer when the task is finished. We did some work in PipedRDD to make pipe work well with Spark. I think it is better if we reuse the work instead of re-inventing all stuff again per user.

For the performance perspective, I think we are not going to change some currently existing API so it is not a performance regression. User may understand the performance meaning when using pipe. Actually pipe is the way if there is no other choices. In some cases user have to use it, e.g. they're unable to replace the forked process.
I'm not sure if someone ever has been complained for it. For batch Dataset, it is easy to convert to RDD and run RDD.pipe. For streaming one, it is simply impossible. Maybe it is also related to SS adaptation rate?
UDF/expression not work because pipe is not 1-to-1 relation between input and output. For example piping through wc -l gets single output per partition.
There will be some issues if just invoking forked process in mapPartitions API, e.g. interrupting stdin writer when the task is finished. We did some work in PipedRDD to make pipe work well with Spark. I think it is better if we reuse the work instead of re-inventing all stuff again per user.

I understand the functionality is lacking on SS. There's a workaround like foreachBatch -> toRDD -> pipe but streaming operations can't be added after calling pipe. So I'd agree that it'd be better to address the gap in any way.
I feel default serialization logic on PipedRDD is also fragile and not well documented as well. (This actually makes me wondering, is PipedRDD widely adopted?) Is there any documentation/mention that T.toString is used as a serialization, and it doesn't escape line break so multiple lines string will be printed as multiple lines without any guard? The default implementation is too naive and even for primitive type it's not hard to find the hole. There's a parameter to customize the serialization and we can add it as well so it makes me less concerned, but default should be still reasonable and well explained for the limitations if any.
Adding this to top level API would be easiest way to do as this simply leverages PipedRDD and the diff of this PR except tests is very small. This is a major upside.
The major downside is that this only works if end users intend to send all columns as input and use the output as single string column. Suppose end users have 10 columns in their DataFrame and want to use pipe with only one column and also retain 9 columns for next operation (so 10 columns still remain including the output of pipe). How to do it?
And like I said above I don't think they'll be able to understand the serialized form for the multiple columns or complicated column types. They'll end up using custom class for type T which overrides toString.
Adding this to function for parameter type string or binary would require more work and it would force end users to provide the serialized form as input if they want to pass multiple columns or non primitive column. But at least they should know what they are doing on using the function, and there's no magic behind the curtain, so should have no issue on serialization. Some functions like to_json would help to do serialize easily. It's up to end users that which column(s) they will pass, and all columns are still retained.

UDF/expression not work because pipe is not 1-to-1 relation between input and output. For example piping through wc -l gets single output per partition.

I'd feel more natural if the relation is 1-to-1 for normal case and N-to-1 "across partitions" for aggregation (so the output will be array[string] unless merge function can be provided). It sounds a bit arbitrary to me if the forked process can decide whether it will aggregate or not.

I understand the functionality is lacking on SS. There's a workaround like foreachBatch -> toRDD -> pipe but streaming operations can't be added after calling pipe. So I'd agree that it'd be better to address the gap in any way.
I feel default serialization logic on PipedRDD is also fragile and not well documented as well. (This actually makes me wondering, is PipedRDD widely adopted?) Is there any documentation/mention that T.toString is used as a serialization, and it doesn't escape line break so multiple lines string will be printed as multiple lines without any guard? The default implementation is too naive and even for primitive type it's not hard to find the hole. There's a parameter to customize the serialization and we can add it as well so it makes me less concerned, but default should be still reasonable and well explained for the limitations if any.
Adding this to top level API would be easiest way to do as this simply leverages PipedRDD and the diff of this PR except tests is very small. This is a major upside.
The major downside is that this only works if end users intend to send all columns as input and use the output as single string column. Suppose end users have 10 columns in their DataFrame and want to use pipe with only one column and also retain 9 columns for next operation (so 10 columns still remain including the output of pipe). How to do it?
And like I said above I don't think they'll be able to understand the serialized form for the multiple columns or complicated column types. They'll end up using custom class for type T which overrides toString.
Adding this to function for parameter type string or binary would require more work and it would force end users to provide the serialized form as input if they want to pass multiple columns or non primitive column. But at least they should know what they are doing on using the function, and there's no magic behind the curtain, so should have no issue on serialization. Some functions like to_json would help to do serialize easily. It's up to end users that which column(s) they will pass, and all columns are still retained.

UDF/expression not work because pipe is not 1-to-1 relation between input and output. For example piping through wc -l gets single output per partition.

I'd feel more natural if the relation is 1-to-1 for normal case and N-to-1 "across partitions" for aggregation (so the output will be array[string] unless merge function can be provided). It sounds a bit arbitrary to me if the forked process can decide whether it will aggregate or not.

Also, when they can do arbitrary aggregation in forked process per partition, the aggregation is only bound to the micro-batch, even only bound to the single partition, which is not matched the concept of aggregation in SS which should be across the micro-batches.

Also, when they can do arbitrary aggregation in forked process per partition, the aggregation is only bound to the micro-batch, even only bound to the single partition, which is not matched the concept of aggregation in SS which should be across the micro-batches.

I understand the functionality is lacking on SS. There's a workaround like foreachBatch -> toRDD -> pipe but streaming operations can't be added after calling pipe. So I'd agree that it'd be better to address the gap in any way.
I feel default serialization logic on PipedRDD is also fragile and not well documented as well. (This actually makes me wondering, is PipedRDD widely adopted?) Is there any documentation/mention that T.toString is used as a serialization, and it doesn't escape line break so multiple lines string will be printed as multiple lines without any guard? The default implementation is too naive and even for primitive type it's not hard to find the hole. There's a parameter to customize the serialization and we can add it as well so it makes me less concerned, but default should be still reasonable and well explained for the limitations if any.

The current RDD.pipe doesn't explicitly mention we output the string of T. This is what it said "All elements of each input partition are written to a process's stdin as lines of input separated by a newline." If you think it is not enough, we can improve the API document. About the parameter, do you mean printRDDElement?

Adding this to top level API would be easiest way to do as this simply leverages PipedRDD and the diff of this PR except tests is very small. This is a major upside.
The major downside is that this only works if end users intend to send all columns as input and use the output as single string column. Suppose end users have 10 columns in their DataFrame and want to use pipe with only one column and also retain 9 columns for next operation (so 10 columns still remain including the output of pipe). How to do it?
And like I said above I don't think they'll be able to understand the serialized form for the multiple columns or complicated column types. They'll end up using custom class for type T which overrides toString. (And the output of toString shouldn't be multiple lines.)
Adding this to function for parameter type string or binary would require more work and it would force end users to provide the serialized form as input if they want to pass multiple columns or non primitive column. But at least they should know what they are doing on using the function, and there's no magic behind the curtain, so should have no issue on serialization. Some functions like to_json would help to do serialize easily. It's up to end users that which column(s) they will pass, and all columns are still retained.
I'd feel more natural if the relation is 1-to-1 for normal case and N-to-1 "across partitions" for aggregation (so the output will be array[string] unless merge function can be provided). It sounds a bit arbitrary to me if the forked process can decide whether it will aggregate or not.

I think some of the questions are over the scope of the pipe concept. For example the one about only pipe one column but retain 9 columns for next operation. User also cannot only pipe only field of object T and retain all others after by using RDD's pipe.

I understand the functionality is lacking on SS. There's a workaround like foreachBatch -> toRDD -> pipe but streaming operations can't be added after calling pipe. So I'd agree that it'd be better to address the gap in any way.
I feel default serialization logic on PipedRDD is also fragile and not well documented as well. (This actually makes me wondering, is PipedRDD widely adopted?) Is there any documentation/mention that T.toString is used as a serialization, and it doesn't escape line break so multiple lines string will be printed as multiple lines without any guard? The default implementation is too naive and even for primitive type it's not hard to find the hole. There's a parameter to customize the serialization and we can add it as well so it makes me less concerned, but default should be still reasonable and well explained for the limitations if any.

The current RDD.pipe doesn't explicitly mention we output the string of T. This is what it said "All elements of each input partition are written to a process's stdin as lines of input separated by a newline." If you think it is not enough, we can improve the API document. About the parameter, do you mean printRDDElement?

Adding this to top level API would be easiest way to do as this simply leverages PipedRDD and the diff of this PR except tests is very small. This is a major upside.
The major downside is that this only works if end users intend to send all columns as input and use the output as single string column. Suppose end users have 10 columns in their DataFrame and want to use pipe with only one column and also retain 9 columns for next operation (so 10 columns still remain including the output of pipe). How to do it?
And like I said above I don't think they'll be able to understand the serialized form for the multiple columns or complicated column types. They'll end up using custom class for type T which overrides toString. (And the output of toString shouldn't be multiple lines.)
Adding this to function for parameter type string or binary would require more work and it would force end users to provide the serialized form as input if they want to pass multiple columns or non primitive column. But at least they should know what they are doing on using the function, and there's no magic behind the curtain, so should have no issue on serialization. Some functions like to_json would help to do serialize easily. It's up to end users that which column(s) they will pass, and all columns are still retained.
I'd feel more natural if the relation is 1-to-1 for normal case and N-to-1 "across partitions" for aggregation (so the output will be array[string] unless merge function can be provided). It sounds a bit arbitrary to me if the forked process can decide whether it will aggregate or not.

I think some of the questions are over the scope of the pipe concept. For example the one about only pipe one column but retain 9 columns for next operation. User also cannot only pipe only field of object T and retain all others after by using RDD's pipe.

Also, when they can do arbitrary aggregation in forked process per partition, the aggregation is only bound to the micro-batch, even only bound to the single partition, which is no longer matched the concept of aggregation in SS which should be across the micro-batches. (That said, the output can be different between batch and streaming even if the input is same.)

Hmm, I'd say not to think it as aggregation, they are different things. Pipe is widely used in Unix/Linux command line and I don't think we should mix it with aggregation.

Also, when they can do arbitrary aggregation in forked process per partition, the aggregation is only bound to the micro-batch, even only bound to the single partition, which is no longer matched the concept of aggregation in SS which should be across the micro-batches. (That said, the output can be different between batch and streaming even if the input is same.)

Hmm, I'd say not to think it as aggregation, they are different things. Pipe is widely used in Unix/Linux command line and I don't think we should mix it with aggregation.

BTW, I know PipedRDD (RDD.pipe) is widely used. At least I heard many times people use pipe to adapt existing program with Spark.

BTW, I know PipedRDD (RDD.pipe) is widely used. At least I heard many times people use pipe to adapt existing program with Spark.

The current RDD.pipe doesn't explicitly mention we output the string of T. This is what it said "All elements of each input partition are written to a process's stdin as lines of input separated by a newline." If you think it is not enough, we can improve the API document. About the parameter, do you mean printRDDElement?

Yes. This is definitely not enough. This is only intuitive if the type T is primitive like integer, long, String, etc. If you have type T as Java bean and override toString with IDE toString generator, the format is depending on the IDE. case class is depending on Scala, and I don't think the representation of toString is something Scala should guarantee compatibility. Makes sense?

I think some of the questions are over the scope of the pipe concept. For example the one about only pipe one column but retain 9 columns for next operation. User also cannot only pipe only field of object T and retain all others after by using RDD's pipe.

Once you're adding the pipe to the one of DataFrame operations, the operation 'pipe' should be evaluated as a DataFrame operation. End users using pipe wouldn't use the trivial external process like "cat" or "wc -l" which completely ignore the structure of input, but I can't find any example beyond such thing.
(I don't think something is reasonable "just because" previous RDD works like so.)
That's why I want to hear the actual use case, what is the type of T Dataset, which task the external process does, what is the output of external process, should they need to break down the output to multiple columns after that.

Hmm, I'd say not to think it as aggregation, they are different things. Pipe is widely used in Unix/Linux command line and I don't think we should mix it with aggregation.

The thing is that the output can be different between batch and streaming, and that is entirely depending on the external process. Any external process does the aggregation ("wc -l" does still an aggregation in effect, whatever you say) breaks the concept and the result is varying how the input stream is split to multiple micro-batches.
I see the possibility existing APIs can also break such thing (like mapPartitions/flatMap with user function which doesn't consider the fact) so I'd be OK if everyone doesn't mind. I still think restricting the relation to 1-to-1 / N-to-1 would be ideal, but that requires external process to be implemented as Spark's requirement which might not be possible, so...

The current RDD.pipe doesn't explicitly mention we output the string of T. This is what it said "All elements of each input partition are written to a process's stdin as lines of input separated by a newline." If you think it is not enough, we can improve the API document. About the parameter, do you mean printRDDElement?

Yes. This is definitely not enough. This is only intuitive if the type T is primitive like integer, long, String, etc. If you have type T as Java bean and override toString with IDE toString generator, the format is depending on the IDE. case class is depending on Scala, and I don't think the representation of toString is something Scala should guarantee compatibility. Makes sense?

I think some of the questions are over the scope of the pipe concept. For example the one about only pipe one column but retain 9 columns for next operation. User also cannot only pipe only field of object T and retain all others after by using RDD's pipe.

Once you're adding the pipe to the one of DataFrame operations, the operation 'pipe' should be evaluated as a DataFrame operation. End users using pipe wouldn't use the trivial external process like "cat" or "wc -l" which completely ignore the structure of input, but I can't find any example beyond such thing.
(I don't think something is reasonable "just because" previous RDD works like so.)
That's why I want to hear the actual use case, what is the type of T Dataset, which task the external process does, what is the output of external process, should they need to break down the output to multiple columns after that.

Hmm, I'd say not to think it as aggregation, they are different things. Pipe is widely used in Unix/Linux command line and I don't think we should mix it with aggregation.

The thing is that the output can be different between batch and streaming, and that is entirely depending on the external process. Any external process does the aggregation ("wc -l" does still an aggregation in effect, whatever you say) breaks the concept and the result is varying how the input stream is split to multiple micro-batches.
I see the possibility existing APIs can also break such thing (like mapPartitions/flatMap with user function which doesn't consider the fact) so I'd be OK if everyone doesn't mind. I still think restricting the relation to 1-to-1 / N-to-1 would be ideal, but that requires external process to be implemented as Spark's requirement which might not be possible, so...

I don't have a strong opinion. It seems consistent with RDD.pipe but I don't like the fact that we're relying on non-standard string representation by default.

I don't have a strong opinion. It seems consistent with RDD.pipe but I don't like the fact that we're relying on non-standard string representation by default.

Yes. This is definitely not enough. This is only intuitive if the type T is primitive like integer, long, String, etc. If you have type T as Java bean and override toString with IDE toString generator, the format is depending on the IDE. case class is depending on Scala, and I don't think the representation of toString is something Scala should guarantee compatibility. Makes sense?

That is what printRDDElement should do. For complex type T, users can provide custom function and produce necessary output for their need.

Once you're adding the pipe to the one of DataFrame operations, the operation 'pipe' should be evaluated as a DataFrame operation. End users using pipe wouldn't use the trivial external process like "cat" or "wc -l" which completely ignore the structure of input, but I can't find any example beyond such thing.
(I don't think something is reasonable "just because" previous RDD works like so.)
That's why I want to hear the actual use case, what is the type of T Dataset, which task the external process does, what is the output of external process, should they need to break down the output to multiple columns after that.

I think these are mainly focus how object T is going to work. I said it is not different than RDD working because they both work on object T. Custom function provided by users should take the responsibility to produce necessary output to the forked process. (Note I have not add the parameter yet. Maybe do it tomorrow.)

I see the possibility existing APIs can also break such thing (like mapPartitions/flatMap with user function which doesn't consider the fact) so I'd be OK if everyone doesn't mind. I still think restricting the relation to 1-to-1 / N-to-1 would be ideal, but that requires external process to be implemented as Spark's requirement which might not be possible, so...

"pipe" is not invented by Spark. I don't think we should provide a half-baked pipe function. It is worse than nothing. Not to mention that the technical point you said. IMHO, it brings more inconsistency between RDD pipe, Dataset pipe and streaming Dataset pipe. I think what we can do is to explicitly clarify the effect of pipe on micro-batch streaming is only per micro-batch not cross entire stream.

Yes. This is definitely not enough. This is only intuitive if the type T is primitive like integer, long, String, etc. If you have type T as Java bean and override toString with IDE toString generator, the format is depending on the IDE. case class is depending on Scala, and I don't think the representation of toString is something Scala should guarantee compatibility. Makes sense?

That is what printRDDElement should do. For complex type T, users can provide custom function and produce necessary output for their need.

Once you're adding the pipe to the one of DataFrame operations, the operation 'pipe' should be evaluated as a DataFrame operation. End users using pipe wouldn't use the trivial external process like "cat" or "wc -l" which completely ignore the structure of input, but I can't find any example beyond such thing.
(I don't think something is reasonable "just because" previous RDD works like so.)
That's why I want to hear the actual use case, what is the type of T Dataset, which task the external process does, what is the output of external process, should they need to break down the output to multiple columns after that.

I think these are mainly focus how object T is going to work. I said it is not different than RDD working because they both work on object T. Custom function provided by users should take the responsibility to produce necessary output to the forked process. (Note I have not add the parameter yet. Maybe do it tomorrow.)

I see the possibility existing APIs can also break such thing (like mapPartitions/flatMap with user function which doesn't consider the fact) so I'd be OK if everyone doesn't mind. I still think restricting the relation to 1-to-1 / N-to-1 would be ideal, but that requires external process to be implemented as Spark's requirement which might not be possible, so...

"pipe" is not invented by Spark. I don't think we should provide a half-baked pipe function. It is worse than nothing. Not to mention that the technical point you said. IMHO, it brings more inconsistency between RDD pipe, Dataset pipe and streaming Dataset pipe. I think what we can do is to explicitly clarify the effect of pipe on micro-batch streaming is only per micro-batch not cross entire stream.

I don't have a strong opinion. It seems consistent with RDD.pipe but I don't like the fact that we're relying on non-standard string representation by default.

Just like RDD pipe, there will be custom function users can provide for complex type T to produce necessary output to the forked process. I will add it later.

I don't have a strong opinion. It seems consistent with RDD.pipe but I don't like the fact that we're relying on non-standard string representation by default.

Just like RDD pipe, there will be custom function users can provide for complex type T to produce necessary output to the forked process. I will add it later.

Is it too hard requirement to explain the actual use case, especially you've said you have internal customer claiming this feature? I don't think my request requires anything needed redaction. (If there's something you can abstract the details or do some redaction by yourself.) My first comment was asking about the actual use case and I have been asking consistently.
I don't think RDD.pipe and Dataset.pipe is exactly same, at least the usability of the default printRDDElement. There're lots of users using "untyped" Dataset (DataFrame) which the default printRDDElement would depend on the internal implementation (Row is just an interface). The default serializer implementation only works if Dataset has only one column which type is matched with Java/Scala type, otherwise they always want to provide the serializer implementation. Based on this I wonder we should allow default serializer - probably we want to require end users to provide serializer so that they should know what they are doing.

Is it too hard requirement to explain the actual use case, especially you've said you have internal customer claiming this feature? I don't think my request requires anything needed redaction. (If there's something you can abstract the details or do some redaction by yourself.) My first comment was asking about the actual use case and I have been asking consistently.
I don't think RDD.pipe and Dataset.pipe is exactly same, at least the usability of the default printRDDElement. There're lots of users using "untyped" Dataset (DataFrame) which the default printRDDElement would depend on the internal implementation (Row is just an interface). The default serializer implementation only works if Dataset has only one column which type is matched with Java/Scala type, otherwise they always want to provide the serializer implementation. Based on this I wonder we should allow default serializer - probably we want to require end users to provide serializer so that they should know what they are doing.

Is it too hard requirement to explain the actual use case, especially you've said you have internal customer claiming this feature? I don't think my request requires anything needed redaction. (If there's something you can abstract the details or do some redaction by yourself.) My first comment was asking about the actual use case and I have been asking consistently.

I'm not sure how much details you'd like to see? The script we will call? The data input it accepts? The use case is generally to pipe through an external process. But if you insist, I think I need to get more information for the client. I just wonder if it is necessary to provide details instead of general use case of pipe. To know what input/script the client will be helpful? We are not to provide the client-specific API.
Okay, let me try to get more details to make everyone happy here.

I don't think RDD.pipe and Dataset.pipe is exactly same, at least the usability of the default printRDDElement. There're lots of users using "untyped" Dataset (DataFrame) which the default printRDDElement would depend on the internal implementation (Row is just an interface). The default serializer implementation only works if Dataset has only one column which type is matched with Java/Scala type, otherwise they always want to provide the serializer implementation. Based on this I wonder we should allow default serializer - probably we want to require end users to provide serializer so that they should know what they are doing.

For DataFrame, we know it is actually Dataset[Row]. If users need custom print-out, printRDDElement will take Row as input type. I don't see there is a problem here. This is how other typed functions (map, foreach...) work with untyped Dataset. I don't really get the serializer point you just mentioned.

Is it too hard requirement to explain the actual use case, especially you've said you have internal customer claiming this feature? I don't think my request requires anything needed redaction. (If there's something you can abstract the details or do some redaction by yourself.) My first comment was asking about the actual use case and I have been asking consistently.

I'm not sure how much details you'd like to see? The script we will call? The data input it accepts? The use case is generally to pipe through an external process. But if you insist, I think I need to get more information for the client. I just wonder if it is necessary to provide details instead of general use case of pipe. To know what input/script the client will be helpful? We are not to provide the client-specific API.
Okay, let me try to get more details to make everyone happy here.

I don't think RDD.pipe and Dataset.pipe is exactly same, at least the usability of the default printRDDElement. There're lots of users using "untyped" Dataset (DataFrame) which the default printRDDElement would depend on the internal implementation (Row is just an interface). The default serializer implementation only works if Dataset has only one column which type is matched with Java/Scala type, otherwise they always want to provide the serializer implementation. Based on this I wonder we should allow default serializer - probably we want to require end users to provide serializer so that they should know what they are doing.

For DataFrame, we know it is actually Dataset[Row]. If users need custom print-out, printRDDElement will take Row as input type. I don't see there is a problem here. This is how other typed functions (map, foreach...) work with untyped Dataset. I don't really get the serializer point you just mentioned.

I'm not sure how much details you'd like to see?

I'm quoting my comment:

That's why I want to hear the actual use case, what is the type of T Dataset, which task the external process does, what is the output of external process, should they need to break down the output to multiple columns after that.

There's no "detail" on business logic. If the type T is Java bean or case class or something, just mention it as it's a Java bean/case class. If that's typed but bound to the non-primitive type like tuple please mention it. If that's untyped you need to mention it, including whether it has complex column(s).
Fair enough for you?

For DataFrame, we know it is actually Dataset[Row]. If users need custom print-out, printRDDElement will take Row as input type.

What is the output of Row.toString then? Is it consistent if we change the implementation of Row? Even more, should end users know about that?
I already pointed out earlier that the default serializer only makes sense if the type T is primitive for Java/Scala which the output of T.toString is relatively intuitive for end users. For other types the default serializer won't work (or even end users are able to infer it, still fragile) including "untyped".

This is how other typed functions (map, foreach...) work with untyped Dataset.

The problem is other typed functions get the Row as simply Row, and able to call Row.getString or something like that, even knowing which columns the Row instance has. It doesn't need to be serialized as some other form. Does it apply to the external process? No. Spark should serialize the Row instance to send to the external process, and the serialized form of Row instance is "unknown" to end users, unless they deal with crafting serializer by their hand using Row.getString or so on. That's why I said default serializer doesn't work with untyped Dataset.

I'm not sure how much details you'd like to see?

I'm quoting my comment:

That's why I want to hear the actual use case, what is the type of T Dataset, which task the external process does, what is the output of external process, should they need to break down the output to multiple columns after that.

There's no "detail" on business logic. If the type T is Java bean or case class or something, just mention it as it's a Java bean/case class. If that's typed but bound to the non-primitive type like tuple please mention it. If that's untyped you need to mention it, including whether it has complex column(s).
Fair enough for you?

For DataFrame, we know it is actually Dataset[Row]. If users need custom print-out, printRDDElement will take Row as input type.

What is the output of Row.toString then? Is it consistent if we change the implementation of Row? Even more, should end users know about that?
I already pointed out earlier that the default serializer only makes sense if the type T is primitive for Java/Scala which the output of T.toString is relatively intuitive for end users. For other types the default serializer won't work (or even end users are able to infer it, still fragile) including "untyped".

This is how other typed functions (map, foreach...) work with untyped Dataset.

The problem is other typed functions get the Row as simply Row, and able to call Row.getString or something like that, even knowing which columns the Row instance has. It doesn't need to be serialized as some other form. Does it apply to the external process? No. Spark should serialize the Row instance to send to the external process, and the serialized form of Row instance is "unknown" to end users, unless they deal with crafting serializer by their hand using Row.getString or so on. That's why I said default serializer doesn't work with untyped Dataset.

The problem is other typed functions get the Row as simply Row, and able to call Row.getString or something like that, even knowing which columns the Row instance has. It doesn't need to be serialized as some other form. Does it apply to the external process? No. Spark should serialize the Row instance to send to the external process, and the serialized form of Row instance is "unknown" to end users, unless they deal with crafting serializer by their hand using Row.getString or so on. That's why I said default serializer doesn't work with untyped Dataset.

This is why we need a custom function printRDDElement. For complex type the string representation isn't reliable, a custom function is used to produce necessary output to the process.

The problem is other typed functions get the Row as simply Row, and able to call Row.getString or something like that, even knowing which columns the Row instance has. It doesn't need to be serialized as some other form. Does it apply to the external process? No. Spark should serialize the Row instance to send to the external process, and the serialized form of Row instance is "unknown" to end users, unless they deal with crafting serializer by their hand using Row.getString or so on. That's why I said default serializer doesn't work with untyped Dataset.

This is why we need a custom function printRDDElement. For complex type the string representation isn't reliable, a custom function is used to produce necessary output to the process.

OK so you seem to agree default serializer doesn't work for untyped. And I also think we agree default serializer is problematic for non-primitive type T on typed. These cases sound majority, and end users only get benefits if their Dataset is Dataset[String], Dataset[Int], Dataset[Long], etc. Even for them, default serializer can be simply implemented by end users via _.toString, and they would know what they are doing on serializing the row.
That said, is it still beneficial to provide default serializer and lead end users be confused if they don't know about the details? default must be reasonable and I don't think current default serializer is reasonable on majority cases. I think that is a non-trivial difference between RDD.pipe and Dataset.pipe.

OK so you seem to agree default serializer doesn't work for untyped. And I also think we agree default serializer is problematic for non-primitive type T on typed. These cases sound majority, and end users only get benefits if their Dataset is Dataset[String], Dataset[Int], Dataset[Long], etc. Even for them, default serializer can be simply implemented by end users via _.toString, and they would know what they are doing on serializing the row.
That said, is it still beneficial to provide default serializer and lead end users be confused if they don't know about the details? default must be reasonable and I don't think current default serializer is reasonable on majority cases. I think that is a non-trivial difference between RDD.pipe and Dataset.pipe.

OK so you seem to agree default serializer doesn't work for untyped. And I also think we agree default serializer is problematic for non-primitive type T on typed. These cases sound majority, and end users only get benefits if their Dataset is Dataset[String], Dataset[Int], Dataset[Long], etc. Even for them, default serializer can be simply implemented by end users via _.toString, and they would know what they are doing on serializing the row.
That said, is it still beneficial to provide default serializer and lead end users be confused if they don't know about the details? default must be reasonable and I don't think current default serializer is reasonable on majority cases. I think that is a non-trivial difference between RDD.pipe and Dataset.pipe.

Hmm, I'm fine if you think we should always require a custom function to produce the output.

OK so you seem to agree default serializer doesn't work for untyped. And I also think we agree default serializer is problematic for non-primitive type T on typed. These cases sound majority, and end users only get benefits if their Dataset is Dataset[String], Dataset[Int], Dataset[Long], etc. Even for them, default serializer can be simply implemented by end users via _.toString, and they would know what they are doing on serializing the row.
That said, is it still beneficial to provide default serializer and lead end users be confused if they don't know about the details? default must be reasonable and I don't think current default serializer is reasonable on majority cases. I think that is a non-trivial difference between RDD.pipe and Dataset.pipe.

Hmm, I'm fine if you think we should always require a custom function to produce the output.

Test build #134424 has started for PR 31296 at commit 45bd4a7.

Test build #134424 has started for PR 31296 at commit 45bd4a7.

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/39010/

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/39010/

Test build #134424 has finished for PR 31296 at commit 45bd4a7.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test build #134424 has finished for PR 31296 at commit 45bd4a7.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/39010/

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/39010/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/39010/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/134424/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/134424/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/39010/

cc @AngersZhuuuu too FYI

cc @AngersZhuuuu too FYI

Test build #134493 has started for PR 31296 at commit ac7460b.

Test build #134493 has started for PR 31296 at commit ac7460b.

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/39079/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/39079/

Test build #134493 has finished for PR 31296 at commit ac7460b.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test build #134493 has finished for PR 31296 at commit ac7460b.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/134493/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/134493/


Support --diff-published<EMAIL_ADDRESS>to diff against crates.io

Closes #124
Example usage in our own git (note how it handles a crate part of a workspace) (output shortened for brevity):
$ cargo run -- -p rustdoc-json --diff-published<EMAIL_ADDRESS>--verbose
Processing "/Users/martin/.cargo/cargo-public-api_build-root-for-published-crates/rustdoc-json-0.4.2/target/doc/rustdoc_json.json"
Processing "/Users/martin/src/cargo-public-api/target/doc/rustdoc_json.json"
Removed items from the public API
=================================
-pub const fn rustdoc_json::BuildOptions::all_features(self, all_features: bool) -> Self
-pub fn rustdoc_json::BuildOptions::target_dir(self, target_dir: impl AsRef<Path>) -> Self
-pub fn rustdoc_json::BuildOptions::toolchain(self, toolchain: impl Into<Option<String>>) -> Self

Changed items in the public API
===============================
(none)

Added items to the public API
=============================
+impl Unpin for rustdoc_json::BuildOptions
+impl UnwindSafe for rustdoc_json::BuildOptions

I think using a shared target dir makes sense here

Thanks for the input. I wonder how well that will work in practice though? Since we use the nightly compiler, I don't think there is any guarantee about forward compatibility? I.e. that a new version of nightly will properly handle partial build artifacts from an old compiler? And what if the user specifies incompatible crate versions? Either directly or indirectly. Will the compiler always sort that out? Including in the face of build.rs scripts that builds e.g. C++ libraries?
I think it will be tricky for users to diagnose tricky incremental build problems, since the building happens very much under the hood and in a "mysterious" way. So I think it could be could to be a bit defensive here.

I ended up using the user cache dir instead of the .cargo dir, but I used your proposed dir structure. I think using the cache is better though.
Merging!


[fix] prevent double decoding of params

Closes #7515
Please don't delete this checklist! Before submitting the PR, please make sure you do the following:

[x] It's really useful if your PR references an issue where it is discussed ahead of time. In many cases, features are absent for a reason. For large changes, please create an RFC: https://github.com/sveltejs/rfcs
[x] This message body should clearly illustrate what problems it solves.
[ ] Ideally, include a test that fails without this PR but passes with it.

Tests

[x] Run the tests with pnpm test and lint the project with pnpm lint and pnpm check

Changesets

[x] If your PR makes a change that should be noted in one or more packages' changelogs, generate a changeset by running pnpm changeset and following the prompts. All changesets should be patch until SvelteKit 1.0

thank you!

Sorry I didn't get a chance to review this before it got merged, but it looks like this has a bug in it: https://github.com/sveltejs/kit/issues/7554. Any chance you'd be able to take a look at it @dmkret ?


Drawing an image on an image

Unfortunately, I have not found a way to do this yet.
Does a method like Graphics.DrawImage() exist in ImageProcessorCore?
E.g. it would be great if something like this would work:
using (Graphics g = Graphics.FromImage(_resultImage))                     {                         g.DrawImage(_otherImage, x, y);                     }
Thank you

@JimBobSquarePants We call this Compose in ImageMagick. Maybe we should add a Compose Sampler?

There's no 2D drawing in ImageProcessor yet.....
I have an open ticket #264 that would cover paths etc but that still has work to do. Whatever we make I want it to be feature rich, easy to use and powerful. The architecture would be difference from the standard samplers.

This is not drawing text/paths. Its drawing one image on top of another image. In other words copy pixels from the specified area of the source image to the destination image. I would prefer to do this in a Compose sampler so we could add features like only copying one channel.

@dlemstra I'd actually like to have a look at splitting channels. The blend processor can merge based on lerping but that's all or nothing.

using (Graphics g = Graphics.FromImage(_resultImage)) {  g.DrawImage(_otherImage, x, y); }
Create an ImageFactory using a Stream (wherever you got _otherImage, or save _otherImage to a MemoryStream)...
_otherImageFactory.Overlay(
    new ImageLayer
    {
        Image = _resultImage,
        Position = new System.Drawing.Point(x, y)
    }
);

Maybe Overlay needs more parameters?

THATS NOT CORE

No need to shout. Other projects make it a little more obvious my having distinct repositories.

        using (FileStream streamFoo = File.OpenRead(@"C:\Program Files (x86)\Microsoft Visual Studio 14.0\Common7\IDE\ProjectTemplates\Extensibility\1033\VSShellTemplate\VSShellStubExe\Documentation\Images\AppEnv.jpg"))
        using (FileStream streamBar = File.OpenRead(@"C:\Program Files (x86)\Microsoft Visual Studio 14.0\Common7\IDE\Extensions\InstallShield\InstallShieldProject\1033\IS-text-200wide.jpg"))
        using (FileStream output = File.OpenWrite(@"C:\Users\aam00\Desktop\foobar2.jpg"))
        {
            Image imageFoo = new Image(streamFoo);
            Image imageBar = new Image(streamBar);

            imageFoo.Blend(imageBar, 100);

            imageFoo.Save(output);
        }

This works, overlaying at (0,0).

This fails silently if the file already exists. Setting the stream length to 0 doesn't truncate it.
I also tried positioning the overlay:
imageFoo.Blend(imageBar, 100, new Rectangle(50, 30, imageBar.Width, imageBar.Height));

The Rectangle parameter doesn't behave as expected. Rather than being sizing for the positioned image, the width and height seem to crop the overlay relative to (0,0).

Sorry, didn't mean to. Was on my phone. Will have a proper look when I get to a desktop.

@Emyr42 You're absolutely correct. We need another parameter, most likely a rectangle, the first rectangle signifies the area of interest of the image to blend.  The second would represent to location to drop the image.
My thought would be that if the second rectangle dimensions do not match that of the source image re would scale it.
What do you think?

Oh @Emyr42 meant to ask,

This fails silently if the file already exists. Setting the stream length to 0 doesn't truncate it.

What do you mean by this?

This issue was moved to JimBobSquarePants/ImageSharp#14


Document operational aspects of Duffy

Story
As an operator of Duffy,
I want to have detailed information on how to run Duffy,
so I can do that with little to no problems.
Acceptance Criteria

[ ] An easily accessible document exists which describes

[ ] How to install Duffy
[ ] Its components and how they interact with each other (operational view)
[ ] Considerations on catering for legacy tenants
[ ] Potential pitfalls (if any)

https://github.com/CentOS/duffy/wiki/Usage
Could use some reading between the lines ;P

https://github.com/CentOS/duffy/wiki/Configuration
This extends the example config with some documentation.


Fix cast error on player touch bottom area.

Fix part of #136
Fix cast error on player touch bottom area.

Nothing too interesting. Change from Node -> Area2D since that's what the script wanted. Seems to be fine.

Player now respawns in ~3 seconds instead of 160 seconds.

Thank you this was driving me crazy!


Request watchOS support: Currently get 'System.TypeInitializationException'

Note: for support questions, please ask on StackOverflow: https://stackoverflow.com/questions/tagged/Akavache . This repository's issues are reserved for feature requests and bug reports.
Do you want to request a feature or report a bug?
Request support for watchOS
What is the current behavior?
watchOS not supported
If the current behavior is a bug, please provide the steps to reproduce and if possible a minimal demo of the problem
What is the expected behavior?
Support for watchOS
What is the motivation / use case for changing the behavior?
Using Akavache directly on an Apple Watch Series 3 w/ LTE not connected to an iPhone.
Which versions of Akavache, and which platform / OS are affected by this issue? Did this work in previous versions of Akavache? Please also test with the latest stable and snapshot (https://www.myget.org/feed/Akavache/package/nuget/Akavache) versions.
Other information (e.g. stacktraces, related issues, suggestions how to fix)
In ExtensionDelegate:
BlobCache.ApplicationName = "AkavacheTest";

Terminating app due to uncaught exception 'System.TypeInitializationException', reason: 'The type initializer for 'Akavache.BlobCache' threw an exception.'

In principle, I support adding this platform. In practice - might need some downstream ecosystem work to implement (does sqliteraw green ship impl for watching?)

Looks like it's there as placeholders (TODOs) https://github.com/ericsink/SQLitePCL.raw/blob/master/gen_build.cs#L92.
I'll put a request up over there to see if he plans to work on those TODO items or if I can help him with testing on it given the previously merged pull request: https://github.com/ericsink/SQLitePCL.raw/pull/96
Thanks,
John

@ericsink I appreciate you taking a look at this.


ng-drag-handle is not working

var dragHandles = element.find('[ng-drag-handle]');"
JLite 'find' calls findElementsByTagName under the hood, thus it will never find [ng-drag-handle] attribute.
"angular": "1.4.7",
"ngDraggable": "0.1.8",

change :
var dragHandles = element.find('[ng-drag-handle]');
by :
var dragHandles = angular.element(element[0].querySelectorAll('[ng-drag-handle]'));
line 48

Yeah, I've done this. Do you want me to make a pull request for this? You know, it's not as that good to have a custom build for library.

Yes i will do it when i'll have time :)

is there a pull request for this?


Error setting up entry for hon

Describe the bug
After installing this git repo in my HACS and installing latest version (v0.14.0-beta.14), restarting HA, then I am unable to setup my Haier h0n account. The error I get to see in the error log;
Logger: homeassistant.config_entries
Bron: config_entries.py:604
Eerst voorgekomen: 16:20:27 (1 gebeurtenissen)
Laatst gelogd: 16:20:27

Error setting up entry for hon
Traceback (most recent call last):
  File "/usr/src/homeassistant/homeassistant/config_entries.py", line 604, in async_setup
    result = await component.async_setup_entry(hass, self)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/config/custom_components/hon/__init__.py", line 38, in async_setup_entry
    hon = await Hon(
          ^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/pyhon/hon.py", line 51, in setup
    async for a in Appliance.create_from_data(self._api, appliance_data):
  File "/usr/local/lib/python3.12/site-packages/pyhon/appliances/_base.py", line 84, in create_from_data
    await appliance.load_commands()
  File "/usr/local/lib/python3.12/site-packages/pyhon/appliances/_base.py", line 292, in load_commands
    self.commands, self.additional_data = loader(self, payload)
                                          ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/pyhon/command_loader.py", line 37, in loader
    _clean_name(category_name): HonCommand(
                                ^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/pyhon/commands.py", line 37, in __init__
    self._load_parameters(attributes)
  File "/usr/local/lib/python3.12/site-packages/pyhon/commands.py", line 74, in _load_parameters
    self._create_parameters(data, name, key)
  File "/usr/local/lib/python3.12/site-packages/pyhon/commands.py", line 85, in _create_parameters
    self._rules.append(HonRuleSet(self, data["fixedValue"]))
                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/pyhon/rules.py", line 22, in __init__
    self._parse_rule(rule)
  File "/usr/local/lib/python3.12/site-packages/pyhon/rules.py", line 32, in _parse_rule
    self._parse_conditions(param_key, trigger_key, trigger_data)
  File "/usr/local/lib/python3.12/site-packages/pyhon/rules.py", line 54, in _parse_conditions
    self._parse_conditions(param_key, extra_key, extra_data, extra)
  File "/usr/local/lib/python3.12/site-packages/pyhon/rules.py", line 54, in _parse_conditions
    self._parse_conditions(param_key, extra_key, extra_data, extra)
  File "/usr/local/lib/python3.12/site-packages/pyhon/rules.py", line 43, in _parse_conditions
    for multi_trigger_value, param_data in trigger_data.items():
                                           ^^^^^^^^^^^^^^^^^^
AttributeError: 'str' object has no attribute 'items'

Expected behavior
Able to setup the account haha.
Environment (please complete the following information):

Home Assistant Version: 2024.9.1
hOn Integration Version v0.14.0-beta.14

Tracked here: https://github.com/IoTLabs-pl/pyhOn/issues/5


CORE-6875 Json deserializer interfaces

Comprises of:

The deserializer interface which custom Json deserializers implement.
The interface that can parse Json. The implementation will forward all calls to Jackson, this is part of a later story.

Notes:

Json can of course model a tree of nodes, where each node can have properties and other nodes and a relationship like this is exposed by Jackson’s deserializer interface. This interface does the same albeit in a simplified fashion, offering a decent set of reading functions without adding too much complexity at this stage.
Tests to ensure the interfaces are Java compliant.
These interfaces will be part of the Corda public application api, but there will be nothing anyone can do with them at the moment.
This is not a breaking change.

Have you had a chance to implement this yet? Just wondering if you'll run into the same problem as with the serializers. Though, I agree that we can update them as necessary (and not worry about version change) since they're new.

I've implemented them locally and not noticed any problems, but only tests use the new interfaces at the moment. In any case I've added the required OSGI export now.


Update AboutMenuFragment in Androidlibrary to replace AsyncTask

AsyncTask has been deprecated and should be replaced in AboutMenuFragment.java in Androidlibrary.

Can this issue be assigned to me @wbrunette ?

You currently have 3 issues assigned to you, one has a PR leaving you at 2 issues actively working on. To spread the work, we prefer to let everyone have two active issues. If you would like to be swapped, please let us know what issue you want to be unassigned to.

@wbrunette @Redeem-Grimm-Satoshi Can I work on this task? Since it is related toLicenseReaderTask

@juayuohcarineneng19 Go ahead!

Thanks, I just sent in a PR: https://github.com/odk-x/androidlibrary/pull/275


Missing GET treated differently than missing POST

I've just started a Resource, and the class currently empty. When I try a GET, i get a 500 error, but when I try a POST, i get a 405. They should both be 405s.

the class currently empty.

What would be the use case for this scenario?
The exact same thing happens with plain MethodViews too:
from flask import Flask
from flask.views import MethodView
from flask_restful import Api, Resource

app = Flask(__name__)
api = Api(app)

@api.resource('/resource')
class Example(Resource):
    pass

class Example2(MethodView):
    pass

app.add_url_rule('/methodview', view_func=Example2.as_view('methodview'))

app.run(port=5000, debug=True)

$ http :5000/resource
HTTP/1.0 500 INTERNAL SERVER ERROR
$ http :5000/methodview
HTTP/1.0 500 INTERNAL SERVER ERROR
$ http post :5000/resource
HTTP/1.0 405 METHOD NOT ALLOWED
$ http post :5000/methodview
HTTP/1.0 405 METHOD NOT ALLOWED

In the future, please try to include stacktraces from the console and other relevant info as text instead of screenshots. The ones you included are a bit difficult to read because they are so large.

on an empty class OPTIONS returns 200, GET returns 500, and everything else returns 405.
by adding a put(), the  GET magically goes from being a 500 to a 405
Traceback (most recent call last):
  File "/Users/rbp/Projects/jupiter/default/lib/python2.7/site-packages/flask/app.py", line 1836, in __call__
    return self.wsgi_app(environ, start_response)
  File "/Users/rbp/Projects/jupiter/default/lib/python2.7/site-packages/flask/app.py", line 1820, in wsgi_app
    response = self.make_response(self.handle_exception(e))
  File "/Users/rbp/Projects/jupiter/default/lib/python2.7/site-packages/flask_restful/__init__.py", line 270, in error_router
    return original_handler(e)
  File "/Users/rbp/Projects/jupiter/default/lib/python2.7/site-packages/flask/app.py", line 1403, in handle_exception
    reraise(exc_type, exc_value, tb)
  File "/Users/rbp/Projects/jupiter/default/lib/python2.7/site-packages/flask_restful/__init__.py", line 267, in error_router
    return self.handle_error(e)
  File "/Users/rbp/Projects/jupiter/default/lib/python2.7/site-packages/flask/app.py", line 1817, in wsgi_app
    response = self.full_dispatch_request()
  File "/Users/rbp/Projects/jupiter/default/lib/python2.7/site-packages/flask/app.py", line 1477, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/Users/rbp/Projects/jupiter/default/lib/python2.7/site-packages/flask_restful/__init__.py", line 270, in error_router
    return original_handler(e)
  File "/Users/rbp/Projects/jupiter/default/lib/python2.7/site-packages/flask/app.py", line 1381, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/Users/rbp/Projects/jupiter/default/lib/python2.7/site-packages/flask_restful/__init__.py", line 267, in error_router
    return self.handle_error(e)
  File "/Users/rbp/Projects/jupiter/default/lib/python2.7/site-packages/flask/app.py", line 1475, in full_dispatch_request
    rv = self.dispatch_request()
  File "/Users/rbp/Projects/jupiter/default/lib/python2.7/site-packages/flask/app.py", line 1461, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/Users/rbp/Projects/jupiter/default/lib/python2.7/site-packages/flask_restful/__init__.py", line 471, in wrapper
    resp = resource(*args, **kwargs)
  File "/Users/rbp/Projects/jupiter/default/lib/python2.7/site-packages/flask/views.py", line 84, in view
    return self.dispatch_request(*args, **kwargs)
  File "/Users/rbp/Projects/jupiter/default/lib/python2.7/site-packages/flask_restful/__init__.py", line 576, in dispatch_request
    assert meth is not None, 'Unimplemented method %r' % request.method
AssertionError: Unimplemented method 'GET'

Any cons for changing GET's default behavior? According to https://www.w3.org/Protocols/rfc2616/rfc2616-sec10.html#sec10.4.6 GET if GET is not implemented/allowed, a 405 should be returned with the appropriate headers. Seems pretty doable.

I ended up adding a resource class to my app that handles a missing get as a 405, regardless of the presence of other methods.  Making it just a little special with the 500 doesn't make much sense to me.
For an app with resources at /somepath/(id), but for which there's no response at /somepath, it seems nicer for /somepath to generate 405s instead of 404s is just a bit cleaner (perhaps a bit of flakiness on my part?).  But it requires registering an otherwise "empty" resource class.
(Yes, I could construct some response for /somepath, but that seems pretty arbitrary as well.)

just closed? no commit?

Interesting resolution for the issue.
Given the deprecation of reqparse and having to work around little stuff like this, I'm not sure flask-restful actually adds much aside from method-based dispatch for my projects. Might be time to change things out for me.

what resolution? it was just closed. maybe that's what you mean by 'intersting'?

That's exactly what I find interesting.  Not satisfactory, but interesting that this response is considered satisfactory to the maintainers.

Perhaps it wasn't clear enough in my previous comment that this behavior is inherited from Flask itself (Resource is a subclass of flask.views.MethodView). Therefore I don't think it is appropriate to fix it in Flask-RESTful. If this behavior actually causes issues for your applications, you should raise it upstream in the Flask project.
I had also asked if there were any practical scenarios where someone would want to use an empty Resource that had no methods defined, but received no response. A resource that supports only the OPTIONS method seems rather useless, so I'm not sure why it would even be added to the API.
I'm sorry you weren't happy with the resolution, I hope this explains it better.
You can also subclass Resource yourself to define get which raises 405 METHOD NOT ALLOWED:
class Resource(flask_restful.Resource):
    def get(self, *args, **kwargs):
        abort(405)

Which is essentially what I've done, as described (perhaps not clearly) in my comment on 8 Feb 2016.

@freddrake This is the relevant "buggy" code:
flask.view.MethodView
    def dispatch_request(self, *args, **kwargs):
        meth = getattr(self, request.method.lower(), None)
        # If the request method is HEAD and we don't have a handler for it
        # retry with GET.
        if meth is None and request.method == 'HEAD':
            meth = getattr(self, 'get', None)
        assert meth is not None, 'Unimplemented method %r' % request.method
        return meth(*args, **kwargs)


⚠ PrivateHD has degraded performance

In 158fd4b, PrivateHD ($PHD) experienced degraded performance:

HTTP code: 200
Response time: 2641 ms

Resolved: PrivateHD performance has improved in 48d81f0.


Catch "413 Request Entity Too Large" error

When you send too large email - you get error in logs.
How can it be catched correctly in code?
Sample code
$message =
            (new \Swift_Message('app:test:email'))
<EMAIL_ADDRESS>                ->setFrom($this->mailerSenderEmail, $this->mailerSenderName)
                ->setBody('test', 'text/plain');
        $message->attach((new \Swift_Attachment())
            ->setFilename('test.pdf')
            ->setBody(str_pad('', 30000000, '1'))
        );

        $result = $this->mailer->send($message);
        $output->writeln('Done! - '.$result);

Since Symfony now have improved his mailer component, I suggest to take a look at https://symfony.com/doc/current/mailer.html#using-a-3rd-party-transport to see if it fits your needs.
Otherwise, I think you can just get an error from the event Swift_Events_SendEvent::RESULT_FAILED but no details attached.
Hope this helps a bit.


Event-based personalized search + recommendations

I noticed the recent integration of LLMs, local or remote, including RAG. This is a great feature.
The most requested feature nowadays is a real-time recommendation system, based on user events, like clicks, add to wishlist, add to basket, or orders, views, video display ....
It is so difficult to set in motion that only paid SaaS services are considered in general. Open source/local recsys are either too slow or not good enough (lack of real time, lack of good models for specific goals ...).
Features in the API:

Ingest catalog
Ingest events
Ingest users
dashboards
create scenarios from a selection of models (items-to-item, items-to-user, ...), boosts, and filters
call scenarios to get recommendations
call scenarios to get personalized search (reranking?)

I forgot to mention search personalization! Done.

Perhaps integrating the new UBI format?

https://github.com/opensearch-project/user-behavior-insights
https://www.youtube.com/watch?v=8_4VRtMWtSY


Http client support for full url argument

Fixes # 812
Changes
Overloaded the session function to be able to accept full URLs as a parameter instead of just host and port
Removed the default port of 80 since it would be going to the overloaded function otherwise
Changed the CreateSession parameter of one test to test this change
All tests and functions pass, program builds examples without fault

Codecov Report

Merging #26 (022f748) into main (976b5ca) will decrease coverage by 0.46%.
The diff coverage is 89.29%.

@@            Coverage Diff             @@
##             main      #26      +/-   ##
==========================================
- Coverage   95.96%   95.51%   -0.45%
==========================================
  Files         176      156      -20
  Lines        7172     6627     -545
==========================================
- Hits         6882     6329     -553
- Misses        290      298       +8

Impacted Files
Coverage Δ

api/include/opentelemetry/baggage/baggage.h
97.30% <ø> (ø)

.../include/opentelemetry/common/key_value_iterable.h
100.00% <ø> (ø)

...ude/opentelemetry/common/key_value_iterable_view.h
88.89% <ø> (ø)

api/include/opentelemetry/common/kv_properties.h
98.89% <ø> (ø)

api/include/opentelemetry/common/spin_lock_mutex.h
31.25% <ø> (ø)

api/include/opentelemetry/common/string_util.h
100.00% <ø> (ø)

api/include/opentelemetry/common/timestamp.h
100.00% <ø> (ø)

api/include/opentelemetry/context/context.h
100.00% <ø> (ø)

...lemetry/context/propagation/composite_propagator.h
100.00% <ø> (ø)

...ntelemetry/context/propagation/global_propagator.h
100.00% <ø> (ø)

... and 167 more


Static CSS Function and Custom Media Queries…

Problem Description
In the React component's static css(props) {} function, we can use the device: key to target CSS to a particular media query. As far as I know, there are only two possible values for the device: key, tablet and phone.
How would we go about targeting our CSS to other media queries, such as:
desktop only: ie. @media only screen and (min-width:981px)
or
tablet only:   ie. @media only screen and (max-width:980px) and (min-width:768px)
Thanks.

@dtcblyth no presently; it'll come someday however :)

Hi! Any news on this?
How can we set styles to desktop only? Is there any string we can enter in the device key?
Thank you!


How to make stunserver differentiate between the listening IP and the external IP

We have an AWS EC2 instance running stunserver. Its internal IP1 is <IP_ADDRESS>, corresponding to the public (external) IP1 of <IP_ADDRESS> (BINAT), and internal IP2 is <IP_ADDRESS>, corresponding to the public (external) IP2 of <IP_ADDRESS> (also through aws igw binat).
Now, using the command stunserver --mode full --primaryinterface <IP_ADDRESS> --altinterface <IP_ADDRESS>, we can start the Stuntman server. However, due to its lack of awareness of its actual public (external) IP address, there are some issues when communicating with the STUN client. It requires the STUN client to send messages to the internal 172.31 subnet.
Is there any parameter that we haven't discovered yet, which can be used to inform stunserver of the external public IP addresses corresponding to its primary and secondary listening addresses?
Thanks :-)

You want to set the "advertised addresses". There is a pair of parameters
for that: --primaryadvertised and --altadvertised
Example:
stunserver --mode full  --primaryinterface <IP_ADDRESS> --altinterface
<IP_ADDRESS> --primaryadvertised <IP_ADDRESS> --altadvertised <IP_ADDRESS>
On Tue, Sep 12, 2023 at 5:15 AM asbai @.***> wrote:

We have an AWS EC2 instance running stunserver. Its internal IP1 is
<IP_ADDRESS>, corresponding to the public (external) IP1 of <IP_ADDRESS> (BINAT),
and internal IP2 is <IP_ADDRESS>, corresponding to the public (external) IP2
of <IP_ADDRESS> (also through aws igw binat).
Now, using the command stunserver --mode full --primaryinterface
<IP_ADDRESS> --altinterface <IP_ADDRESS>, we can start the Stuntman server.
However, due to its lack of awareness of its actual public (external) IP
address, there are some issues when communicating with the STUN client. It
requires the STUN client to send messages to the internal 172.31 subnet.
Is there any parameter that we haven't discovered yet, which can be used
to inform stunserver of the external public IP addresses corresponding to
its primary and secondary listening addresses?
Thanks :-)
—
Reply to this email directly, view it on GitHub
https://github.com/jselbie/stunserver/issues/53, or unsubscribe
https://github.com/notifications/unsubscribe-auth/AAHNSFZV3DFNMCWUTS2ICNLX2BG47ANCNFSM6AAAAAA4UYIEIM
.
You are receiving this because you are subscribed to this thread.Message
ID: @.***>

stunserver --mode full  --primaryinterface <IP_ADDRESS> --altinterface <IP_ADDRESS> --primaryadvertised <IP_ADDRESS> --altadvertised <IP_ADDRESS>


Add runtime handling for new rpc resource and FaaS contracts.

Codecov Report

Merging #204 (be282e1) into develop (801811b) will increase coverage by 4.20%.
The diff coverage is 59.34%.

@@             Coverage Diff             @@
##           develop     #204      +/-   ##
===========================================
+ Coverage    50.87%   55.07%   +4.20%
===========================================
  Files           38       43       +5
  Lines         4246     4456     +210
===========================================
+ Hits          2160     2454     +294
+ Misses        1953     1854      -99
- Partials       133      148      +15

Flag
Coverage Δ

unittests
55.07% <59.34%> (+4.20%)
:arrow_up:

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files
Coverage Δ

pkg/adapters/grpc/faas_grpc.go
0.00% <0.00%> (ø)

pkg/adapters/grpc/resources_grpc.go
0.00% <0.00%> (ø)

pkg/worker/faas_http_worker.go
0.00% <0.00%> (ø)

pkg/worker/faas_worker.go
0.00% <0.00%> (ø)

pkg/worker/http_worker.go
0.00% <0.00%> (ø)

pkg/worker/schedule_worker.go
0.00% <0.00%> (ø)

pkg/plugins/gateway/cloudrun/http.go
80.55% <55.55%> (-8.74%)
:arrow_down:

pkg/plugins/gateway/lambda/lambda.go
69.51% <57.14%> (+0.02%)
:arrow_up:

pkg/worker/subscription_worker.go
58.33% <58.33%> (ø)

pkg/worker/grpc_worker.go
62.37% <62.37%> (ø)

... and 10 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 801811b...be282e1. Read the comment docs.


What would be the implications of disabling validation in serialization?

In an application that uses boto to massively insert to DynamoDB we've been able to observe how a significant amount of CPU time is dedicated to serialization and a part of it is dedicated parameter validation.
Disabling validation (through forcing include_validation=False in create_serializer() at serialize.py) leads to a ~23% performance increase.
Now, the question is: how (un)safe would be to disable validation in a production system?
python 2.7
Amazon Linux AMI 2015.09
botocore 1.3.15

That depends on how confident you are that your inputs are correct, and what your upgrade process looks like. If the data you send is invalid, the service will reject it, producing a ClientError on your side after you've already sent all the data. Many services have per-request pricing, so disabling local validation could potentially cost more than the saved cpu time. Dynamo gives free data in, though I don't know how invalid requests count against write capacity, if at all. I would guess that it won't count against you. Additionally, if there is a minor service update that changes parameter validation for Dynamo and you upgrade botocore without testing, you may start to see requests failing since you're using the updated model. This is not likely to happen, however.
In summary, here are the implications:

You won't know about invalid data until after you've spent bandwidth sending it
There will be a different class of errors when invalid data is found
Depending on pricing model specifics, you could still be charged for invalid requests
Updating botocore will require that you ensure your data still passes validation

Thank you very much for your reply, Jordon. That's pretty much what we thought, but it always is reassuring to hear it from someone with deeper insights, thanks! :)


soc: it8xxx2: add soc variants

Currently there are two soc variants (IT81202BX and IT81302BX) on
it8xxx2 series. The IT81202BX is 128-pin package (GPIO K and L groups
aren't bonding with pad).
This makes soc variant configurable and apply corresponding configuration
for a soc.
Signed-off-by: Dino Li<EMAIL_ADDRESS>

@fkokosinski please revisit.