[BEAM-2855] nexmark python suite implement implement query 3, 4, 5, 6, 7, 8, 11

[leiyiz]

implemented querys
made a little change to nexmark Launcher to fix a bug where the error is not passed out correctly

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Choose reviewer(s) and mention them in a comment (R: @username).
• Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

Post-Commit Tests Status (on master branch)

Lang	SDK	Dataflow	Flink	Samza	Spark	Twister2
Go		---		---		---
Java
Python				---		---
XLang		---		---		---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website
Non-portable
Portable	---		---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[leiyiz]

R: @y1chi
R: @pabloem

[y1chi]

LGTM, just some minor nits.

[leiyiz]

Run Python PreCommit

[pabloem]

Is this how this is implemented in Java? I am wondering if we should make bids comparable. If they were comparable, then you would be able to just return max_prices

e.g.:

@functools.total_ordering
class ComparableBidByPrice(object):
  def __init__(self, bid):
    self.bid = bid
  def __eq__(self, other):
    return self.bid == other.bid
  def __lt__(self, other):
    return self.bid.price < other.bid.price

And then you'd do:

  max_bids = (
      sliding_bids
      | beam.Map(ComparableBidByPrice)
      | beam.CombineGlobally(max).without_defaults())

thoughts? The main thing here is having one fewer stage, thus higher performance - but I think the best option is to do whatever Java does.

[y1chi]

I think one of the purpose would be to benchmark the performance of side inputs, thus some pipelines are choosing certain beam semantics that may not be the best way.

[pabloem]

that's reasonable. Than can you just add a comment @leiyiz ?

[leiyiz]

Yeah, I read the code and the reason to not use combiner, which is more efficient, is to utilize the side-input functionality

[pabloem]

can you add a comment on the file in a follow up PR please?

[pabloem]

This trigger is a little hard to wrap my head around : ) can you help me understand it? So I guess we simply accumulate fired panes and fire everything every time? (let's say that the stream contains a new element every second. Would we fire 1000 elements after 1000 seconds?)

[leiyiz]

yes, so every time an event arrives, it should fire and calculate

[leiyiz]

because it is calculating the mean, every time something arrives it calculates the mean 1 more time

[pabloem]

Can it happen that the same person creates new auctions after > max_auction_waiting_time ? Will there be a new person event? If we get new auctions after the person is expired, then we'll just keep adding them to auction_state forevetr, no?

[leiyiz]

I think it would, then the new auction for person is just added to the state and eventually dropped when the pipeline ends. but it is also specified in the nexmark spec as "clear the state after TTL" also the default timer is like 600 seconds long which is way longer than the test duration so I think it is less of a issue?