eugenegujing opened a new pull request, #6053:
URL: https://github.com/apache/texera/pull/6053
### What changes were proposed in this PR?
**Problem.** Pandas-based Python operators (e.g. Sort via `TableOperator`)
build a DataFrame from input tuples. When an INT/LONG column contains nulls,
pandas promotes the whole column to float64 because an int column cannot hold
NaN, so `119` becomes `119.0`. On output, the worker's strict schema validation
in `Tuple.finalize()` then fails with `TypeError: Unmatched type for field
'weight', expected AttributeType.INT, got 119.0 (<class 'float'>) instead.`
This crashed every workflow whose CSV had an integer column with at least one
missing value, and the only workaround was manually inserting a Type Casting
operator for each affected column.
**Why fix it in the Python worker (option (a) of the issue).** The CSV
schema inference is correct (an all-integer column with nulls *is* INTEGER; the
JVM side handles null ints fine), and a UI per-column override (option (c))
would not remove the crash. The type contract is broken by pandas at the
Python-worker boundary, so the fix belongs at that boundary's single
chokepoint: `Tuple.cast_to_schema()`, which already performs safe casts (NaN ->
None, object -> pickled bytes) right before `validate_schema()`.
**The fix** (in `Tuple.cast_to_schema()` only; `validate_schema()` is
unchanged): when the target type is INT or LONG and the value is a float
(including `np.float64`) with a zero fractional part, cast it back to int — but
only when the result is provably the original integer:
- INT window: Arrow int32 capacity `[-2^31, 2^31 - 1]`. int32 values are
always exactly representable in float64, so capacity is the only constraint.
- LONG window: the float64 exact-integer range `[-(2^53) + 1, 2^53 - 1]`
instead of int64 capacity. Above 2^53, float64 rounds, so the received float
may already be a corrupted rendition of the original integer; coercing it would
turn a loud validation error into silent data corruption. The endpoint 2^53
itself is excluded because it is ambiguous (`2^53 + 1` also rounds to float
`2^53`).
- The range check compares the converted int rather than floats, to avoid
float rounding at the window endpoints.
- Non-integral, infinite, and out-of-window floats are left untouched so
`validate_schema()` still rejects them: lossy coercion must never happen
silently. An out-of-window integral float additionally logs an actionable
warning suggesting a cast to STRING or DOUBLE (or LONG for large integers in an
INT field).
**Deliberate behavior change for reviewers to note.** Restructuring the
if-chain in `cast_to_schema()` also fixes a pre-existing stale-variable bug: a
NaN destined for a BINARY field was first set to None and then re-pickled from
the stale local variable, producing pickled-NaN bytes instead of None. NaN in a
BINARY field now correctly finalizes to None (guarded by a dedicated test).
**The changed logic in `core/models/tuple.py`.** A new module-level constant
defines the safely coercible window per integral type:
```python
# Signed value ranges of the integral AttributeTypes within which an
# integral float can be safely cast back to int. INT is bounded by Arrow
# int32 capacity. LONG is bounded by the float64 exact-integer window
# rather than int64 capacity: above 2**53 float64 rounds, so the received
# float may already be a corrupted rendition of the original integer. The
# endpoint 2**53 itself is excluded because it is ambiguous (2**53 + 1
# also rounds to float 2**53).
INTEGRAL_TYPE_RANGES = {
AttributeType.INT: (-(2**31), 2**31 - 1),
AttributeType.LONG: (-(2**53) + 1, 2**53 - 1),
}
```
`cast_to_schema()`'s per-field loop is restructured from two independent
`if`s into mutually exclusive branches (null handling / integral-float coercion
/ BINARY pickling), which both hosts the new coercion and eliminates the
stale-variable read described above:
```python
# convert NaN to None to support null value conversion
if checknull(field_value):
self[field_name] = None
elif field_value is not None:
field_type = schema.get_attr_type(field_name)
if (
field_type in INTEGRAL_TYPE_RANGES
and isinstance(field_value, float)
and field_value.is_integer()
):
# pandas promotes an int column holding nulls to float64
# (119 -> 119.0), so convert integral floats destined for
# INT/LONG back to int -- but only when the result fits the
# safe range. Compare on the int result to avoid float
# rounding at the endpoints.
min_value, max_value = INTEGRAL_TYPE_RANGES[field_type]
int_value = int(field_value)
if min_value <= int_value <= max_value:
self[field_name] = int_value
else:
logger.warning(...) # actionable guidance, see diff
elif field_type == AttributeType.BINARY and not isinstance(
field_value, bytes
):
self[field_name] = b"pickle " + pickle.dumps(field_value)
```
The outer per-field `try/except` (keep the value unchanged if a cast fails,
continue with the next field) is preserved, and `validate_schema()` is
untouched, so anything the coercion deliberately skips still fails validation
loudly.
### Any related issues, documentation, discussions?
Fixes #5935
### How was this PR tested?
TDD: the tests were written first and confirmed to reproduce the crash
(red), then the fix turned them green.
- 34 new test cases in `amber/src/test/python/core/models/test_tuple.py` (59
total in the file, all passing): coercion cases including the int32 and
float64-exact-window boundaries and `np.float64`; rejection of non-integral /
infinite / out-of-window floats; the out-of-window warning; NaN/None handling;
DOUBLE and STRING fields staying untouched; tests pinning the coercion into
`cast_to_schema` rather than `validate_schema`; and an integration-style test
reproducing the full pipeline (`Table.from_tuple_likes` -> float64 promotion ->
`as_tuples` -> `finalize`).
- Full Python worker suite: `cd amber && pytest -m "not integration"` — all
pass.
- `ruff check` and `ruff format --check` clean on both changed files.
- `sbt "scalafixAll --check"` and `sbt scalafmtCheckAll` pass.
- Backend `AMBER_TEST_FILTER=skip-integration sbt test`: the full suite
passes — 0 failed, 0 aborted (WorkflowCore 1570, amber 1076, all other service
modules green). Run against a clean iceberg catalog, matching how CI provisions
one per run.
- Manual reproduction of the issue scenario (CSV with an integer column
containing blanks -> Sort) is covered by the integration-style unit test above,
which exercises the same `Table` -> `finalize` code path the worker uses.
### Was this PR authored or co-authored using generative AI tooling?
Co-authored by: Claude Code (Claude Fable 5)
--
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]
