HyukjinKwon commented on code in PR #146:
URL:
https://github.com/apache/spark-connect-rust/pull/146#discussion_r4462607391
##########
crates/pyspark-rs/src/dataframe.rs:
##########
@@ -2491,3 +2519,40 @@ impl PyLocalRowIterator {
}
}
}
+
+/// Convert an observed-metric literal to a Python value (mirrors
+/// `LiteralExpression._to_value` for the scalar kinds observations produce).
Kinds
+/// without a natural Python scalar fall back to their rendered string.
+fn literal_to_py(
+ py: Python<'_>,
+ literal: &spark_connect_proto::expression::Literal,
+) -> PyResult<Py<PyAny>> {
+ use pyo3::IntoPyObjectExt;
+ use spark_connect_proto::expression::literal::LiteralType;
+ Ok(match &literal.literal_type {
+ None | Some(LiteralType::Null(_)) => py.None(),
+ Some(LiteralType::Boolean(b)) => b.into_py_any(py)?,
+ Some(LiteralType::Byte(v))
+ | Some(LiteralType::Short(v))
+ | Some(LiteralType::Integer(v)) => v.into_py_any(py)?,
+ Some(LiteralType::Long(v)) => v.into_py_any(py)?,
+ Some(LiteralType::Float(v)) => (*v as f64).into_py_any(py)?,
+ Some(LiteralType::Double(v)) => v.into_py_any(py)?,
+ Some(LiteralType::String(s)) => s.into_py_any(py)?,
+ Some(LiteralType::Binary(b)) => pyo3::types::PyBytes::new(py,
b).into_any().unbind(),
+ Some(LiteralType::Decimal(d)) => py
+ .import("decimal")?
+ .getattr("Decimal")?
+ .call1((&d.value,))?
+ .unbind(),
+ Some(LiteralType::Array(a)) => {
+ let items = a
+ .elements
+ .iter()
+ .map(|e| literal_to_py(py, e))
+ .collect::<PyResult<Vec<_>>>()?;
+ items.into_py_any(py)?
+ }
+ Some(_) =>
spark_connect::profiler::format_literal(literal).into_py_any(py)?,
Review Comment:
**Blocking (P1):** This fallback sends every literal kind not matched above
through `profiler::format_literal`, which is a display helper. For date,
timestamp, timestamp_ntz, time, the nanosecond timestamps, every interval kind,
map, struct and specialized arrays, it returns a fixed type name such as
`"timestamp"` or `"map"`, not a rendered value (profiler.rs:30-44). So
`df.observe("o", F.max("ts").alias("m"))` followed by an action gives
`executionInfo.flows == [("o", {"m": "timestamp"})]`, where PySpark's
`LiteralExpression._to_value` returns a `datetime`. The wrong value comes back
without any error, and the doc comment above says these kinds fall back to
their "rendered string", which is not what happens. These kinds need real
decoding (date, datetime, time, timedelta, dict, Row, list) to match the
reference client.
**Verification:**
- **Regression:** For each literal kind that previously fell back to a type
name (date, timestamp, timestamp_ntz, time, the nanosecond timestamps,
calendar, year-month and day-time intervals, map, struct, specialized array),
an observation whose server response carries that literal yields the reference
Python value in executionInfo.flows rather than a type-name string.
- **Behavior:** Nested arrays, maps and structs decode their elements
recursively, while existing scalar kinds (numbers, strings, binary, decimal)
keep their current values.
##########
crates/spark-connect/src/session.rs:
##########
@@ -713,6 +720,10 @@ fn datatype_to_arrow(dt: &DataType) ->
Result<arrow::datatypes::DataType> {
// TIMESTAMP (LTZ) and TIMESTAMP_NTZ are both micros; NTZ carries no
zone.
DataType::Timestamp => A::Timestamp(TimeUnit::Microsecond,
Some("UTC".into())),
DataType::TimestampNtz => A::Timestamp(TimeUnit::Microsecond, None),
+ DataType::TimestampLtzNanos { .. } => {
Review Comment:
**Non-blocking (P2):** The schema side now maps the nanosecond variants to
`Timestamp(Nanosecond, ..)`, but `values_to_arrow` still only has the
`DataType::Timestamp | DataType::TimestampNtz` arm (line 867). These variants
fall to the `Unsupported type for Arrow createDataFrame conversion` catch-all.
On the base branch a server-reported `TIMESTAMP_LTZ` nanos column decoded as
`DataType::Timestamp`, so `spark.createDataFrame(df.collect(), df.schema)`
worked. With this change it raises. `spark.createDataFrame([],
StructType([StructField("t", TimestampNTZNanosType())]))` and nested
array/struct/map element types fail the same way. A builder arm that scales the
row micros to a `TimestampNanosecondArray` (UTC only for LTZ) would keep it
consistent with this mapping.
**Verification:**
- **Regression:** Building a local relation from timestamp rows with a
timestamp_ltz(9) schema, as when re-creating a DataFrame from its own
server-reported nanos schema, succeeds and keeps the instant, as the base
branch did when this type decoded as a microsecond timestamp.
- **Behavior:** Empty and non-empty local data with timestamp_ntz(p) and
timestamp_ltz(p) columns, including inside an array, produce nanosecond Arrow
arrays whose zone matches the declared field.
##########
crates/pyspark-rs/src/pipelines.rs:
##########
@@ -162,12 +162,14 @@ pub fn pipeline_define_flow(
}
/// `PipelineCommand.DefineFlow` carrying an Auto-CDC flow body.
`stored_as_scd_type` is
-/// the `SCDType` enum value (0 unspecified, 1 SCD_TYPE_1, 2 SCD_TYPE_2).
+/// the `SCDType` enum value (0 unspecified, 1 SCD_TYPE_1, 2 SCD_TYPE_2). The
SCD2-only
+/// track-history lists and `sql_conf` are validated client-side in
`pipelines/api.py`.
Review Comment:
**Nit (P3):** Only the track-history lists are validated in
`pipelines/api.py` (CANNOT_SET_TOGETHER /
INVALID_MULTIPLE_ARGUMENT_CONDITIONS). `create_auto_cdc_flow` passes
`spark_conf=spark_conf or {}` with no check, and the registry forwards
`dict(flow.spark_conf)` as is, so the `sql_conf` part of this sentence is
inaccurate.
**Verification:**
- **Inspection:** The documented validation matches the checks
create_auto_cdc_flow actually performs on track-history lists and spark_conf.
##########
crates/pyspark-rs/src/errors.rs:
##########
@@ -75,6 +76,11 @@ pub fn spark_error_to_py_exception(err: SparkError) -> PyErr
{
// This creates a PyObject exception instance
match exc_class.call1((msg.clone(),)) {
Ok(exc_instance) => {
+ // Mirror the reference client, which stamps
`_operation_id` on
+ // errors raised for an ExecutePlan (read via
`operation_id`).
+ if let Some(op) = &operation_id {
+ let _ = exc_instance.setattr("_operation_id", op);
Review Comment:
**Nit (P3):** This `_operation_id` stamping is never exercised:
`test_execution_info_and_operation_id` sets `err._operation_id` by hand and
builds `ExecutionInfo` directly, so it only tests the Python property and the
vendored classes. Removing or misspelling this setattr, passing the wrong id
into `ExecutionInfo`, or swapping observed keys and values would keep every
suite green. Offline tests driven by the mock Connect server (one action whose
responses carry an operation id and observed metrics, and one failing
`ExecutePlan`) would cover both the getter and this conversion.
**Verification:**
- **Regression:** After a mocked action that returns an operation id and
observed metrics, executionInfo.operation_id equals that id and flows reports
the observed keys and decoded values.
- **Regression:** A mocked ExecutePlan failure raises a PySpark exception
whose operation_id equals the operation id of the failed request.
##########
crates/spark-connect/src/session.rs:
##########
@@ -31,6 +31,9 @@ pub struct ExecutionInfo {
/// Observed metrics collected during execution (also where `observe(...)`
/// and UDF-profiler results surface).
pub observed_metrics: Vec<proto::execute_plan_response::ObservedMetrics>,
+ /// The server-assigned ExecutePlan operation id
(`ExecutePlanResponse.operation_id`).
Review Comment:
**Nit (P3):** The id is not server-assigned.
`ExecutePlanResponseReattachableIterator::new` uses the request's
`operation_id` or generates `Uuid::new_v4()` on the client before sending
(crates/spark-connect-core/src/reattach.rs:38-44), and the server echoes it in
`ExecutePlanResponse.operation_id`. Describing it as client-assigned and echoed
by the server would avoid misleading callers about when the id is known.
**Verification:**
- **Inspection:** The field documentation matches the id assignment in the
reattachable ExecutePlan iterator and no longer calls it server-assigned.
##########
python/pyspark/pipelines/spark_connect_graph_element_registry.py:
##########
@@ -165,7 +167,12 @@ def register_flow(self, flow: Flow) -> None:
def register_auto_cdc_flow(self, flow: AutoCdcFlow) -> None:
scl_file, scl_line = _scl(flow.source_code_location)
- stored_as_scd_type = _SCD_TYPE_1 if flow.stored_as_scd_type is not
None else 0
+ # SCDType enum value: 0 unspecified, else SCD_TYPE_1 / SCD_TYPE_2.
+ stored_as_scd_type = (
+ _SCD_TYPE_BY_VALUE[str(flow.stored_as_scd_type)]
Review Comment:
**Nit (P3):** This SCD-type mapping and the positional forwarding of
`track_history_column_list`, `track_history_except_column_list` and
`spark_conf` below are fork-specific code with no failing signal.
`test_create_auto_cdc_flow_scd2_track_history` uses
`LocalGraphElementRegistry`, which only stores the `AutoCdcFlow`, and no Python
or Rust test inspects the `DefineFlow` this path builds. Mapping `"2"` to
SCD_TYPE_1, swapping the two track-history arguments, or dropping `sql_conf`
would all pass. A test through this registry that captures the `DefineFlow`
offline (as other offline tests do with the mock Connect server) would cover it.
**Verification:**
- **Regression:** An SCD2 flow registered with track_history_column_list and
spark_conf produces a DefineFlow carrying SCD_TYPE_2, the expected
track-history expressions in the include field only, and the given sql_conf.
- **Behavior:** A flow with no SCD type and no track-history arguments still
produces an unspecified SCD type, empty track-history lists and an empty
sql_conf.
--
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]
