Repository navigation
sync with open source how - #118
Draft
lesterhaynes wants to merge 10000 commits into
Draft
lesterhaynes wants to merge 10000 commits into
lesterhaynes wants to merge 10000 commits into
Conversation
njeevan233-spec
approved these changes
Feb 1, 2026
Add QdrantWriteConfig.content_payload_key. When set, the converter stores EmbeddableItem.content_string under that key in the point payload next to the metadata, so points retrieved from a collection written by Beam carry the chunk text. The default (None) keeps the current behavior of writing only the metadata. A key that collides with a metadata key raises ValueError instead of overwriting it, an empty key is rejected at construction time, and the item's metadata dict is copied rather than mutated. Fixes #40476 Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
* Replace typing.ByteString with collections.abc.Buffer * yapf * Replace ByteString in coders_property_based_test.py * update snowflakeio test to use buffer * Swap BQ schema tests to collections.abc.Sequence * corresponding change * missing import
Add RpcQosOptions.Builder.withRampupThrottlingDisabled(), mirroring DatastoreV1.Write.withRampupThrottlingDisabled(). When set, writes no longer wait for the 500/50/5 ramp-up budget and batch sizes are no longer capped by it; adaptive throttling on failures still applies. Addresses #27170. Claude-Session: https://claude.ai/code/session_01BofSrVTSEGJvEzsSNXfU2o Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: Yi Hu <yathu@google.com>
* update more nodes * update resolved address
* Correct pickling error check. * yapf
ReaderInvocationUtil pushed Beam metrics to Flink after every call to Reader#start and Reader#advance. Each push builds the metric results of all steps of the task and filters them down to one step. In a profiled batch job this took 37.6% of the CPU of the read stage, more than reading the records. Push metrics to Flink on the first call, at most once per second after that, and whenever the reader has no data. Metrics are still recorded on every call, and the metrics of the pipeline result, which come from the Flink accumulator, are not affected. Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
* Add gcs_avro_to_cloud_bigtable yaml blueprint example and e2e test * update blueprint to follow similar pattern as e2e * remove template link
* Add support for zoneinfo UTC timestamps * review feedback
… SpannerIO.Write (#40489) * [SpannerIO] Support excludeTxnFromChangeStreams and transactionTag in SpannerIO.Write * Added negative test case * Formatting changes
* Partly revert #40595: setCompileAndRuntimeJavaVersion logic wasn't stale it's needed to resolve errorprone x Gradle known issue compatibility * Clean up Java8 leftovers in setJavaVerOptions.
…0619) In Python's Secret.parse_secret_option, split each parameter on the first ':' only, so values such as ARNs and URLs are kept intact instead of being silently dropped. Java's Secret.parseSecretOption already did this. In both SDKs, a parameter without ':' now raises an error instead of being ignored, and empty parameters (e.g. from a trailing ';') are skipped.
Bumps [google.golang.org/api](https://github.com/googleapis/google-api-go-client) from 0.301.0 to 0.302.0. - [Release notes](https://github.com/googleapis/google-api-go-client/releases) - [Changelog](https://github.com/googleapis/google-api-go-client/blob/main/CHANGES.md) - [Commits](googleapis/google-api-go-client@v0.301.0...v0.302.0) --- updated-dependencies: - dependency-name: google.golang.org/api dependency-version: 0.302.0 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
…40626) * [Prism] Guard ClearOrderedListState against uninitialized state maps Problem: In TentativeData.ClearOrderedListState (sdks/go/pkg/beam/runners/prism/internal/engine/data.go), Prism indexed directly into d.state[stateID][w] and wrote the cleared slice back to kMap[string(uKey)] without checking whether d.state, d.state[stateID], or winMap[w] had been initialized: winMap := d.state[stateID] w := d.toWindow(wKey) kMap := winMap[w] data := kMap[string(uKey)] ... kMap[string(uKey)] = StateData{Bag: cleared} // panic: assignment to entry in nil map SDK harnesses (such as the Python SDK's SynchronousOrderedListRuntimeState.commit and the Rust SDK's OrderedListCell) buffer ordered-list mutations in memory during a bundle and flush pending StateRequest_Clear range requests before StateRequest_Append requests at bundle commit. When a bundle clears an ordered-list state cell (or range) in a window for which no prior bundle has persisted state on the runner, TentativeData.state has no map for (stateID, window), kMap is nil, and Prism panics on line 336. Fix: Use the existing d.clearState(stateID, wKey) helper in ClearOrderedListState and return early when kMap is nil, matching ClearBagState, ClearMultimapState, and ClearMultimapKeysState. Reproduction (Python SDK on PrismRunner): ```python import apache_beam as beam from apache_beam.coders import VarIntCoder from apache_beam.options.pipeline_options import PipelineOptions from apache_beam.testing.util import assert_that, equal_to from apache_beam.transforms.userstate import OrderedListStateSpec from apache_beam.utils.timestamp import Timestamp class ClearOrderedListDoFn(beam.DoFn): EVENTS = OrderedListStateSpec("events", VarIntCoder()) def process(self, element, events=beam.DoFn.StateParam(EVENTS)): key, (ts, op) = element if op == "CLEAR": events.clear() yield f"cleared:{key}" else: events.add((Timestamp(micros=ts), op)) entries = [f"{int(t.micros)}={v}" for t, v in events.read()] yield f"entry:{key}:{','.join(entries)}" if __name__ == "__main__": options = PipelineOptions(["--runner=PrismRunner", "--environment_type=LOOPBACK"]) with beam.Pipeline(options=options) as p: out = ( p | beam.Create([("k", (30, 300)), ("k", (10, 100)), ("k", (0, "CLEAR"))]) | beam.ParDo(ClearOrderedListDoFn()) ) assert_that( out, equal_to(["entry:k:30=300", "entry:k:10=100,30=300", "cleared:k"]), ) ``` * Refactor comments in data_test.go Updated comments for clarity in data_test.go.
…NSTRUCTOR_THROW (#40444)
…l-python-connector 1.23.0 (#40623)
* Restore Java 11 support in setJavaVerOptions the '11' branch. Workflows that still pass -PtestJavaVersion=11 now fail while Gradle configures the projects: Unknown Java Version 11 for setting additional java options * Trigger test workflow
* Enable throttler. * Preserve while loop instead of if.
…mestamp (#40599) The default start_timestamp of PeriodicImpulse and MatchContinuously is evaluated once, when the module is imported. As discussed in #40565 the default stays as is, so document it and point to an explicit timestamp or rebase for starting relative to pipeline execution. Fixes #40564 Co-authored-by: JanakiRamChittineni <341781521+raams95@users.noreply.github.com>
* [typescript] - fix typescript failures * fix test call
* [yaml] - fix post commit yaml test * trigger postcommit
* Add examples for AWS credentials from a secret manager Add examples under apache_beam/examples/secret_manager that set --aws_credentials_secret_manager to read AWS credentials from Google Cloud Secret Manager: * aws_credentials_s3_read.py: reads an S3 file outside a pipeline. * aws_credentials_s3_pipeline.py: reads an S3 file in a pipeline (DirectRunner or Dataflow). * aws_credentials_secret_read.py: reads a secret from AWS Secrets Manager with Secret.parse_secret_option and Secret.from_json. * Add a use case of arn for reading AWS secrets * Reformat * Added a requirements file for the example.
…delHandler (#40500) * [Python] Fix max_models_per_worker_hint not being enforced in KeyedModelHandler * Guard increment_max_models with a lock on the model handler manager
* [Prism] Finalize bundles of DoFns fused after another transform
Problem:
In finalizeStage (sdks/go/pkg/beam/runners/prism/internal/preprocess.go), Prism sets
stage.finalize only while it inspects the consumers of the stage's input PCollections.
The loop skips every PCollection that a transform of the same stage produces:
for pid, plinks := range stageFacts.PcolConsumers {
if _, ok := stageFacts.PcolProducers[pid]; ok {
continue
}
...
if pardo.GetRequestsFinalization() {
stg.finalize = true
}
So a ParDo that sets requests_finalization but is fused after another transform (a Map
after Create, the expand step of a Reshuffle, any element-wise chain) never marks its
stage. The SDK harness registers the callbacks and returns requires_finalization, but Prism
never sends FinalizeBundleRequest, and the callbacks never run. This affects every SDK. The
existing TestParDoBundleFinalizer cases apply the finalizing DoFn directly to Impulse, so the
DoFn always consumes the stage input and the tests pass.
Fix:
Set stage.finalize from every ParDo-like transform of the stage (ParDo and the splittable
DoFn stage URNs), not only from the consumers of the stage inputs. Statefulness, the
on-window-expiration timer and side inputs still come from the input consumers. Add a
TestParDoBundleFinalizer case with the finalizing DoFn fused after another ParDo.
Reproduction (Python SDK on PrismRunner):
```python
import os
import tempfile
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
class MarkFinalization(beam.DoFn):
"""Registers a bundle finalization callback that creates the file `path`."""
def __init__(self, path):
self._path = path
def process(self, element, bundle_finalizer=beam.DoFn.BundleFinalizerParam):
path = self._path
bundle_finalizer.register(lambda: open(path, "w").close())
yield element
if __name__ == "__main__":
markers = tempfile.mkdtemp()
at_input = os.path.join(markers, "at_stage_input")
fused = os.path.join(markers, "after_fused_map")
options = PipelineOptions(["--runner=PrismRunner", "--environment_type=LOOPBACK"])
with beam.Pipeline(options=options) as p:
impulse = p | beam.Impulse()
# Consumes the input of its stage: Prism finalizes the bundle.
_ = impulse | "AtStageInput" >> beam.ParDo(MarkFinalization(at_input))
# Fused after a Map in the same stage: Prism never finalizes the bundle.
_ = (
impulse
| "Map" >> beam.Map(lambda x: x)
| "AfterFusedMap" >> beam.ParDo(MarkFinalization(fused))
)
print("finalized:", sorted(os.listdir(markers)))
assert os.path.exists(at_input), "the DoFn at the stage input was not finalized"
assert os.path.exists(fused), "the DoFn fused after Map was not finalized"
```
Before the fix: `finalized: ['at_stage_input']`, and the second assertion fails.
After the fix: `finalized: ['after_fused_map', 'at_stage_input']`.
* [Prism] Test the finalization flag of a stage, and drop an unreachable error
Move the check of requests_finalization into requestsFinalization. finalizeStage runs
computeFacts first, which already rejects a ParDo payload that does not decode, so the
error return of the earlier loop could not run.
Add TestFinalizeStage_RequestsFinalization: the flag stays false when no transform asks for
it, and is set when the transform at the stage input, a ParDo fused after another
transform, or a fused splittable DoFn process stage asks for it.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
20 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Please add a meaningful description for your change here
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI.