From 255948a4a5f871fa0379c87d329f8aafb66a99ef Mon Sep 17 00:00:00 2001 From: JanakiRamChittineni <341781521+raams95@users.noreply.github.com> Date: Tue, 13 Oct 2026 01:29:01 -0500 Subject: [PATCH 1/4] [Python] Keep Nexmark Query 1 bid prices as integers Query 1 multiplied the bid price by a float, so the converted Bid could not be encoded by BidCoder, which writes the price as a varint. Use the same integer arithmetic as the Java Query 1. Fixes #24678 --- .../benchmarks/nexmark/queries/query1.py | 6 +-- .../benchmarks/nexmark/queries/query1_test.py | 50 +++++++++++++++++++ 2 files changed, 52 insertions(+), 4 deletions(-) create mode 100644 sdks/python/apache_beam/testing/benchmarks/nexmark/queries/query1_test.py diff --git a/sdks/python/apache_beam/testing/benchmarks/nexmark/queries/query1.py b/sdks/python/apache_beam/testing/benchmarks/nexmark/queries/query1.py index 5588bc3cb1eb..315b333c080f 100644 --- a/sdks/python/apache_beam/testing/benchmarks/nexmark/queries/query1.py +++ b/sdks/python/apache_beam/testing/benchmarks/nexmark/queries/query1.py @@ -29,8 +29,6 @@ from apache_beam.testing.benchmarks.nexmark.models import nexmark_model from apache_beam.testing.benchmarks.nexmark.queries import nexmark_query_util -USD_TO_EURO = 0.89 - def load(events, metadata=None, pipeline_options=None): return ( @@ -38,5 +36,5 @@ def load(events, metadata=None, pipeline_options=None): | nexmark_query_util.JustBids() | 'ConvertToEuro' >> beam.Map( lambda bid: nexmark_model.Bid( - bid.auction, bid.bidder, bid.price * USD_TO_EURO, bid.date_time, - bid.extra))) + bid.auction, bid.bidder, + (bid.price * 89) // 100, bid.date_time, bid.extra))) diff --git a/sdks/python/apache_beam/testing/benchmarks/nexmark/queries/query1_test.py b/sdks/python/apache_beam/testing/benchmarks/nexmark/queries/query1_test.py new file mode 100644 index 000000000000..fba2b3751f8b --- /dev/null +++ b/sdks/python/apache_beam/testing/benchmarks/nexmark/queries/query1_test.py @@ -0,0 +1,50 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Unit tests for Nexmark Query 1.""" + +# pytype: skip-file + +import unittest + +import apache_beam as beam +from apache_beam.testing.benchmarks.nexmark.models import nexmark_model +from apache_beam.testing.benchmarks.nexmark.queries import query1 +from apache_beam.testing.test_pipeline import TestPipeline +from apache_beam.testing.util import assert_that +from apache_beam.testing.util import equal_to +from apache_beam.utils.timestamp import Timestamp + + +class Query1Test(unittest.TestCase): + def test_converted_price_is_encodable_integer(self): + bids = [ + nexmark_model.Bid(1, 2, 100, Timestamp.of(0), ''), + nexmark_model.Bid(1, 3, 1234, Timestamp.of(1), ''), + ] + + def round_trip_price(bid): + coder = nexmark_model.Bid.CODER + return coder.decode(coder.encode(bid)).price + + with TestPipeline() as p: + prices = (query1.load(p | beam.Create(bids)) | beam.Map(round_trip_price)) + assert_that(prices, equal_to([89, 1098])) + + +if __name__ == '__main__': + unittest.main() From c21c6f340c37e5160e2abedbc9a93f25b012ee3d Mon Sep 17 00:00:00 2001 From: JanakiRamChittineni <341781521+raams95@users.noreply.github.com> Date: Tue, 13 Oct 2026 08:37:31 -0500 Subject: [PATCH 2/4] Add CHANGES.md entry --- CHANGES.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGES.md b/CHANGES.md index fda4c9edc218..3ea5d81c88c2 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -104,6 +104,7 @@ * (Python) `Duration` built from float seconds now rounds to the nearest microsecond instead of truncating, which could lose a microsecond ([#40263](https://github.com/apache/beam/issues/40263)). * (Python) `ReadFromText` with `escapechar` no longer skips a delimiter that directly follows an escaped delimiter, which merged two records into one ([#40459](https://github.com/apache/beam/issues/40459)). * (Python) DataFrame API: `agg` with a list that includes `'count'` or `'size'` now labels those rows `count`/`size` instead of `sum` ([#40504](https://github.com/apache/beam/issues/40504)). +* (Python) Nexmark Query 1 now keeps bid prices as integers, so its output can be encoded by `BidCoder` ([#24678](https://github.com/apache/beam/issues/24678)). * Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)). ## Security Fixes From afbaa0b502b902519bd81fd3fd104e6e6ecfa662 Mon Sep 17 00:00:00 2001 From: JanakiRamChittineni <341781521+raams95@users.noreply.github.com> Date: Tue, 13 Oct 2026 14:43:56 -0500 Subject: [PATCH 3/4] Re-enable Python Nexmark Query 1 in the Direct postcommit --- .../beam_PostCommit_Python_Nexmark_Direct.yml | 3 +- .../benchmarks/nexmark/queries/query1_test.py | 50 ------------------- 2 files changed, 1 insertion(+), 52 deletions(-) delete mode 100644 sdks/python/apache_beam/testing/benchmarks/nexmark/queries/query1_test.py diff --git a/.github/workflows/beam_PostCommit_Python_Nexmark_Direct.yml b/.github/workflows/beam_PostCommit_Python_Nexmark_Direct.yml index f553073b78dc..1665a86adddc 100644 --- a/.github/workflows/beam_PostCommit_Python_Nexmark_Direct.yml +++ b/.github/workflows/beam_PostCommit_Python_Nexmark_Direct.yml @@ -97,10 +97,9 @@ jobs: matrix: job_name: [beam_PostCommit_Python_Nexmark_Direct] # Query numbers are listed explicitly due to issues with Python Nexmark Query: - # query = 1 - https://github.com/apache/beam/issues/24678 # query = 4,6,9 - https://github.com/apache/beam/issues/24679 # query = 12 - https://github.com/apache/beam/issues/24680 - query: [0, 2, 3, 5, 7, 8, 10, 11] + query: [0, 1, 2, 3, 5, 7, 8, 10, 11] if: | github.event_name == 'workflow_dispatch' || github.event_name == 'pull_request' || diff --git a/sdks/python/apache_beam/testing/benchmarks/nexmark/queries/query1_test.py b/sdks/python/apache_beam/testing/benchmarks/nexmark/queries/query1_test.py deleted file mode 100644 index fba2b3751f8b..000000000000 --- a/sdks/python/apache_beam/testing/benchmarks/nexmark/queries/query1_test.py +++ /dev/null @@ -1,50 +0,0 @@ -# -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You under the Apache License, Version 2.0 -# (the "License"); you may not use this file except in compliance with -# the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -"""Unit tests for Nexmark Query 1.""" - -# pytype: skip-file - -import unittest - -import apache_beam as beam -from apache_beam.testing.benchmarks.nexmark.models import nexmark_model -from apache_beam.testing.benchmarks.nexmark.queries import query1 -from apache_beam.testing.test_pipeline import TestPipeline -from apache_beam.testing.util import assert_that -from apache_beam.testing.util import equal_to -from apache_beam.utils.timestamp import Timestamp - - -class Query1Test(unittest.TestCase): - def test_converted_price_is_encodable_integer(self): - bids = [ - nexmark_model.Bid(1, 2, 100, Timestamp.of(0), ''), - nexmark_model.Bid(1, 3, 1234, Timestamp.of(1), ''), - ] - - def round_trip_price(bid): - coder = nexmark_model.Bid.CODER - return coder.decode(coder.encode(bid)).price - - with TestPipeline() as p: - prices = (query1.load(p | beam.Create(bids)) | beam.Map(round_trip_price)) - assert_that(prices, equal_to([89, 1098])) - - -if __name__ == '__main__': - unittest.main() From 49fb2885868c9608a581c820c9e78c184d598963 Mon Sep 17 00:00:00 2001 From: JanakiRamChittineni <341781521+raams95@users.noreply.github.com> Date: Wed, 14 Oct 2026 19:49:30 -0500 Subject: [PATCH 4/4] Remove CHANGES.md entry for test-only change --- CHANGES.md | 1 - 1 file changed, 1 deletion(-) diff --git a/CHANGES.md b/CHANGES.md index 712e0c578098..7693e65d78a0 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -106,7 +106,6 @@ * (Python) `ReadFromText` with `escapechar` no longer skips a delimiter that directly follows an escaped delimiter, which merged two records into one ([#40459](https://github.com/apache/beam/issues/40459)). * (Python) DataFrame API: `agg` with a list that includes `'count'` or `'size'` now labels those rows `count`/`size` instead of `sum` ([#40504](https://github.com/apache/beam/issues/40504)). * (Python) DataFrame API: `Series.value_counts(bins=...)` no longer always fails with a `TypeError` ([#40503](https://github.com/apache/beam/issues/40503)). -* (Python) Nexmark Query 1 now keeps bid prices as integers, so its output can be encoded by `BidCoder` ([#24678](https://github.com/apache/beam/issues/24678)). * Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)). ## Security Fixes