dongjoon-hyun commented on PR #57834: URL: https://github.com/apache/spark/pull/57834#issuecomment-5228068759
The overall design looks solid: the same operation ID is consistently attached to `ExecutePlan`, `ReattachExecute`, and `ReleaseExecute` via a single interceptor on both paths, and the Scala test's interceptor-based verification genuinely exercises the builder-side filtering (without it, `Metadata.get` would return the user-supplied `"ignored"` value and the test would fail). A few comments: **1. Python still sends the user-supplied header on non-`ExecutePlan` RPCs (inconsistent with Scala and with the warning message)** Scala strips the user-provided `spark-connect-operation-id` from the channel metadata in `createChannel()`, so it is removed for all RPCs. Python only filters it in `_execute_plan_metadata()`, while `AnalyzePlan`, `Config`, `Interrupt`, `ReleaseSession`, `AddArtifacts`, etc. still pass `self._builder.metadata()` unfiltered. If a user sets `spark-connect-operation-id=foo` as a connection option: - the Python init-time warning says the option "is ignored", but it is actually still sent on non-`ExecutePlan` RPCs; - a gateway correlating on this header would see a bogus static value on those RPCs, partially reintroducing the problem this PR solves. It would be better to filter once centrally (e.g. cache the filtered metadata at client init and use it for all RPCs), matching the Scala behavior. **2. The Python test assertion does not actually verify the filtering** In `test_execute_plan_sends_operation_id_metadata`, `dict(mock.metadata)` keeps the last value for duplicate keys, and `_execute_plan_metadata` appends the operation ID last — so the test would still pass even if the user-supplied `"ignored"` entry were not filtered out. To realize the "connection options cannot override" intent, assert the key appears exactly once: ```python values = [v for k, v in mock.metadata if k == "spark-connect-operation-id"] self.assertEqual(values, [req.operation_id]) ``` **3. Nit: redundant stub construction in `CustomSparkConnectBlockingStub`** `executePlanStub` calls `SparkConnectServiceGrpc.newBlockingStub(channel)` again, but the class already has a `stub` member — `stub.withInterceptors(...)` would suffice. Stubs are lightweight so this is not a performance concern, just duplication. **4. Nit: guard against an empty operation ID** `executePlan(request)` attaches the header unconditionally, so an empty `request.getOperationId` would send an empty-valued header. The same method already uses `Option(request.getOperationId).filter(_.nonEmpty)`; skipping the header when the ID is empty would be consistent and safer. Low risk in practice since the only current caller always sets a validated UUID. Item 1 is worth fixing (or at least aligning the warning text with the actual behavior) before merging; the rest are minor. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
