JingsongLi commented on code in PR #9760:
URL: https://github.com/apache/paimon/pull/9760#discussion_r3998882720
##########
paimon-python/pypaimon/multimodal/temporal.py:
##########
@@ -484,6 +525,233 @@ def build_arrays(self, anchor_rows, fetcher):
return arrays
+class _WindowJoinRight(_AsOfJoinRight):
+
+ _SUPPORTED_AGGREGATIONS = {
+ "count", "first", "last", "max", "mean", "min",
+ }
+
+ def __init__(self, label, query, anchor_on, by, preceding, following,
+ aggregations, closed, right_on, suffix):
+ super().__init__(
+ label, query, anchor_on, by, "nearest", None,
+ right_on, suffix)
+ self._preceding_key = _window_bound_key(
+ "preceding", preceding, self.time_type)
+ if following is None:
+ following = (
+ timedelta(0) if pa.types.is_timestamp(self.time_type) else 0)
+ self._following_key = _window_bound_key(
+ "following", following, self.time_type)
+ if closed not in ("both", "left", "neither", "right"):
+ raise ValueError(
+ "closed must be 'both', 'left', 'right', or 'neither'.")
+ self.closed = closed
+ self.aggregations = _normalize_aggregations(
+ aggregations, self.payload_schema, self.label,
+ self._SUPPORTED_AGGREGATIONS)
+ source_names = {
+ specification[1] for specification in self.aggregations
+ }
+ self._fetch_names = tuple(
+ field.name for field in self.payload_schema
+ if field.name in source_names)
+ self.payload_schema = pa.schema([
+ field for field in self.payload_schema
+ if field.name in source_names
+ ], metadata=self.payload_schema.metadata)
+
+ def output_fields(self, payload_schema, effective=True):
+ fields = []
+ for output_name, source_name, aggregation in self.aggregations:
+ source = payload_schema.field(source_name)
+ try:
+ output_type = _aggregate_output_type(
+ source.type, aggregation)
+ except TypeError:
+ if effective:
+ raise
+ output_type = source.type
+ fields.append(pa.field(
+ output_name, output_type, nullable=True,
+ metadata=source.metadata))
+ return fields
+
+ def match(self, anchor_row):
+ key = tuple(anchor_row[name] for name in self.by)
+ bounds = self._index.get(key)
+ if bounds is None:
+ return []
+ target = anchor_row[_TIME_KEY]
+ start, end = bounds
+ left = target - self._preceding_key
+ right = target + self._following_key
+ if (pa.types.is_integer(self.time_type)
+ or pa.types.is_timestamp(self.time_type)):
+ first_key = (
+ math.ceil(left)
+ if self.closed in ("both", "left")
+ else math.floor(left) + 1
+ )
+ last_key = (
+ math.floor(right)
+ if self.closed in ("both", "right")
+ else math.ceil(right) - 1
+ )
+ # Avoid comparing NumPy keys with out-of-range Python integers.
+ first_key = max(first_key, _python_scalar(self._time_keys[start]))
+ last_key = min(last_key, _python_scalar(self._time_keys[end - 1]))
+ if first_key > last_key:
+ return []
+ first = bisect_left(
+ self._time_keys, first_key, start, end)
+ last = bisect_right(
+ self._time_keys, last_key, first, end)
+ else:
+ first = (
+ bisect_left(self._time_keys, left, start, end)
+ if self.closed in ("both", "left")
+ else bisect_right(self._time_keys, left, start, end)
+ )
+ last = (
+ bisect_right(self._time_keys, right, first, end)
+ if self.closed in ("both", "right")
+ else bisect_left(self._time_keys, right, first, end)
+ )
+ return [self._row_ids[index].as_py()
+ for index in range(first, last)]
+
+ def build_arrays(self, anchor_rows, fetcher):
+ matches = [self.match(row) for row in anchor_rows]
+ unique_ids = list(dict.fromkeys(
+ row_id for match in matches for row_id in match))
+ values = fetcher.fetch(unique_ids)
+ positions = {
+ row_id: index for index, row_id in enumerate(unique_ids)
+ }
+ indices = [
+ [positions[row_id] for row_id in match]
+ for match in matches
+ ]
+ arrays = []
+ for _, source_name, aggregation in self.aggregations:
+ effective = fetcher.schema.field(source_name)
+ output_type = _aggregate_output_type(
+ effective.type, aggregation)
+ arrays.append(pa.array([
+ _aggregate_values(
+ values[source_name], row_indices, aggregation)
+ for row_indices in indices
+ ], type=output_type))
+ return arrays
+
+
+def _normalize_aggregations(aggregations, schema, label, supported):
+ if not isinstance(aggregations, dict) or not aggregations:
+ raise ValueError("aggregations must be a non-empty dict.")
+ normalized = []
+ missing = []
+ for output_name, specification in aggregations.items():
+ if not isinstance(output_name, str) or not output_name:
+ raise ValueError(
+ "Aggregation output names must be non-empty strings.")
+ if isinstance(specification, str):
+ source_name = output_name
+ aggregation = specification
+ elif isinstance(specification, tuple) and len(specification) == 2:
+ source_name, aggregation = specification
+ else:
+ raise ValueError(
+ "Aggregation %r must be an operation or a "
+ "(source column, operation) pair." % output_name)
+ if not isinstance(source_name, str) or not source_name:
+ raise ValueError(
+ "Aggregation source columns must be non-empty strings.")
+ if source_name not in schema.names:
+ missing.append(source_name)
+ if not isinstance(aggregation, str) or aggregation not in supported:
+ raise ValueError(
+ "Unsupported aggregation %r for output %r; expected one of "
+ "%r." % (aggregation, output_name, sorted(supported)))
+ normalized.append((output_name, source_name, aggregation))
+ if missing:
+ raise ValueError(
+ "%s is missing aggregation columns %r." % (label, missing))
+ return tuple(normalized)
+
+
+def _aggregate_output_type(data_type, aggregation):
+ if aggregation == "count":
+ return pa.int64()
+ if aggregation in ("first", "last"):
+ return data_type
+ if not (pa.types.is_integer(data_type)
+ or pa.types.is_floating(data_type)):
+ raise TypeError(
+ "Window %s aggregation requires an integer or floating-point "
+ "scalar column; got %s." % (aggregation, data_type))
+ if aggregation == "mean":
+ return pa.float64()
+ return data_type
+
+
+def _aggregate_values(values, indices, aggregation):
+ if not indices:
+ return 0 if aggregation == "count" else None
+ selected = pc.take(values, pa.array(indices, type=pa.int64()))
+ if aggregation == "count":
+ return pc.count(selected).as_py()
+ if aggregation == "mean":
+ items = [item for item in selected.to_pylist()
+ if item is not None]
+ if not items:
+ return None
+ if pa.types.is_integer(values.type):
+ return sum(items) / len(items)
+ if not all(math.isfinite(item) for item in items):
+ return pc.mean(selected).as_py()
+ try:
+ return math.fsum(items) / len(items)
+ except OverflowError:
+ pass
+ scale = max(abs(item) for item in items)
+ if scale == 0:
+ return 0.0
+ return (math.fsum(item / scale for item in items) / len(items)) * scale
+ if aggregation == "min":
+ return pc.min(selected).as_py()
+ if aggregation == "max":
+ return pc.max(selected).as_py()
+ items = selected.to_pylist()
+ if aggregation == "first":
+ return next((item for item in items if item is not None), None)
+ return next((item for item in reversed(items) if item is not None), None)
+
+
+def _window_bound_key(name, value, data_type):
+ if isinstance(value, bool) or not isinstance(value, (Real, timedelta)):
+ raise TypeError(
+ "%s must be numeric or datetime.timedelta." % name)
+ if isinstance(value, Integral):
+ value = int(value)
+ if (isinstance(value, Real) and not isinstance(value, Integral)
+ and not math.isfinite(value)):
+ raise ValueError("%s must be finite." % name)
+ zero = timedelta(0) if isinstance(value, timedelta) else 0
+ if value < zero:
+ raise ValueError("%s must be non-negative." % name)
+ _validate_tolerance(value, data_type)
+ if pa.types.is_integer(data_type) and not isinstance(value, Integral):
+ try:
+ exact = Fraction(value)
+ except TypeError:
+ exact = Fraction(float(value))
+ if exact.denominator == 1:
+ return exact.numerator
+ return exact
+ return _time_tolerance_key(value, data_type)
Review Comment:
**[P2] Preserve fractional timestamp units until applying endpoint closure**
`_time_tolerance_key()` truncates a `timedelta` to the timestamp column's
unit, which changes the meaning of open window endpoints. For a
`timestamp("ms")` column, an anchor and a right sample at the same timestamp,
`preceding=timedelta(microseconds=500)`, `following=timedelta(0)`, and
`closed="right"` should include that sample, but `count` returns 0: the
preceding bound becomes zero and the interval becomes `(t, t]`. Likewise, a
window open at both ends with 1500 microseconds on each side should include
samples at `t-1ms`, `t`, and `t+1ms`, but only includes `t`. Please convert
timestamp bounds to an exact fractional number of timestamp units and retain
that precision until `match()` applies its ceiling/floor logic.
##########
paimon-python/pypaimon/multimodal/temporal.py:
##########
@@ -484,6 +525,233 @@ def build_arrays(self, anchor_rows, fetcher):
return arrays
+class _WindowJoinRight(_AsOfJoinRight):
+
+ _SUPPORTED_AGGREGATIONS = {
+ "count", "first", "last", "max", "mean", "min",
+ }
+
+ def __init__(self, label, query, anchor_on, by, preceding, following,
+ aggregations, closed, right_on, suffix):
+ super().__init__(
+ label, query, anchor_on, by, "nearest", None,
+ right_on, suffix)
+ self._preceding_key = _window_bound_key(
+ "preceding", preceding, self.time_type)
+ if following is None:
+ following = (
+ timedelta(0) if pa.types.is_timestamp(self.time_type) else 0)
+ self._following_key = _window_bound_key(
+ "following", following, self.time_type)
+ if closed not in ("both", "left", "neither", "right"):
+ raise ValueError(
+ "closed must be 'both', 'left', 'right', or 'neither'.")
+ self.closed = closed
+ self.aggregations = _normalize_aggregations(
+ aggregations, self.payload_schema, self.label,
+ self._SUPPORTED_AGGREGATIONS)
+ source_names = {
+ specification[1] for specification in self.aggregations
+ }
+ self._fetch_names = tuple(
+ field.name for field in self.payload_schema
+ if field.name in source_names)
+ self.payload_schema = pa.schema([
+ field for field in self.payload_schema
+ if field.name in source_names
+ ], metadata=self.payload_schema.metadata)
+
+ def output_fields(self, payload_schema, effective=True):
+ fields = []
+ for output_name, source_name, aggregation in self.aggregations:
+ source = payload_schema.field(source_name)
+ try:
+ output_type = _aggregate_output_type(
+ source.type, aggregation)
+ except TypeError:
+ if effective:
+ raise
+ output_type = source.type
+ fields.append(pa.field(
+ output_name, output_type, nullable=True,
+ metadata=source.metadata))
+ return fields
+
+ def match(self, anchor_row):
+ key = tuple(anchor_row[name] for name in self.by)
+ bounds = self._index.get(key)
+ if bounds is None:
+ return []
+ target = anchor_row[_TIME_KEY]
+ start, end = bounds
+ left = target - self._preceding_key
+ right = target + self._following_key
Review Comment:
**[P2] Normalize NumPy floating-point bounds before computing endpoints**
`_window_bound_key()` normalizes NumPy integers, but leaves NumPy
floating-point bounds unchanged for floating-point time columns. With NumPy 2,
adding a Python float timestamp to `np.float32` can narrow the calculation to
float32. For example, when both timestamps are the float64 value
`1700000000.001`, `preceding=0.0` and `following=np.float32(0)` with the
default `closed="both"` return `count=0`, whereas Python float bounds correctly
return 1. The right endpoint rounds down to `1700000000`, excluding the exact
match. Please normalize these bounds to Python scalars before computing the
endpoints and add a regression for NumPy floating-point bounds.
##########
paimon-python/pypaimon/multimodal/temporal.py:
##########
@@ -769,13 +1037,29 @@ def _validate_metadata(query, metadata, key_columns):
class _RowIdFetcher:
- def __init__(self, query, row_group_cache):
+ def __init__(self, query, row_group_cache, output_names=None):
_validate_pinned_tag(query)
- self._schema = _query_schema(query)
+ query_schema, query_paths = _query_schema_and_paths(query)
+ visible_projection = query._effective_projection()
+ if output_names is None:
+ self._schema = query_schema
+ visible_paths = query_paths
+ else:
+ output_names = set(output_names)
+ selected = [
+ (field, path)
+ for field, path in zip(query_schema, query_paths)
+ if field.name in output_names
+ ]
+ self._schema = pa.schema(
+ [field for field, unused in selected],
+ metadata=query_schema.metadata,
+ )
+ visible_paths = [path for unused, path in selected]
+ visible_projection = [".".join(path) for path in visible_paths]
Review Comment:
**[P2] Preserve masking target identity when pruning nested projections**
For a table containing both top-level `a_b` and nested `a.b`,
`select(["a_b", "a.b"])` names the nested output `a_b__0`. If a window
aggregates only `a_b__0`, this pruning rebuilds the projection as `["a.b"]`,
assigning it the alias `a_b`. When the authorization response contains a mask
for the real top-level `a_b`, the later nested-mask check mistakes that rule
for an unbound nested mask and raises `ValueError`. Returning masks for
unprojected columns is supported by the existing authorization contract; the
same input succeeds with `join_asof`, and disabling this pruning makes the
window aggregation succeed as well. Please match masks using the original table
fields/paths so a pruned top-level target cannot be confused with a regenerated
nested alias, while retaining the check for genuinely unbound masking rules.
--
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]
