cloud-fan commented on code in PR #59648: URL: https://github.com/apache/spark/pull/59648#discussion_r4332480593
########## sql/api/src/main/scala/org/apache/spark/sql/BinByBuilder.scala: ########## @@ -0,0 +1,152 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql + +import org.apache.spark.annotation.Experimental + +/** + * Builds a BIN BY operation on a Dataset. + * + * @since 4.4.0 + */ +@Experimental +final class BinByBuilder private[sql] (dataset: Dataset[_]) { + + /** + * Sets the timestamp columns for each row's half-open range. Both columns must have the same + * timestamp type. + * + * @since 4.4.0 + */ + def range(rangeStart: Column, rangeEnd: Column): BinByWithRange = { + new BinByWithRange(dataset = dataset, rangeStart = rangeStart, rangeEnd = rangeEnd) + } +} + +/** + * A BIN BY builder with its range columns set. + * + * @since 4.4.0 + */ +@Experimental +final class BinByWithRange private[sql] ( + dataset: Dataset[_], + rangeStart: Column, + rangeEnd: Column) { + + /** + * Sets the bin width to a positive, constant day-time interval expression. + * + * @since 4.4.0 + */ + def binWidth(width: Column): BinByWithWidth = { + new BinByWithWidth( + dataset, + BinBySpec(rangeStart = rangeStart, rangeEnd = rangeEnd, binWidth = width)) + } +} + +/** + * A BIN BY builder with its range and bin width set. + * + * @since 4.4.0 + */ +@Experimental +final class BinByWithWidth private[sql] (dataset: Dataset[_], spec: BinBySpec) { + + /** + * Sets a constant alignment origin with the same timestamp type as the range columns. The + * default origin is 1970-01-01 00:00:00, in the session time zone for TIMESTAMP. + * + * @since 4.4.0 + */ + def alignTo(origin: Column): BinByWithWidth = { + new BinByWithWidth(dataset, spec.copy(origin = Some(origin))) + } + + /** + * Sets the FLOAT or DOUBLE columns to scale by each bin's overlap ratio. At least one column is + * required. + * + * @since 4.4.0 + */ + @scala.annotation.varargs + def distributeUniform(cols: Column*): BinByWithDistribute = { + new BinByWithDistribute(dataset, spec.copy(distributeColumns = cols.toVector)) + } +} + +/** + * A BIN BY builder that can produce a DataFrame, with optional output column names. Each rename + * returns an independent builder. + * + * @since 4.4.0 + */ +@Experimental +final class BinByWithDistribute private[sql] (dataset: Dataset[_], spec: BinBySpec) { + + /** + * Sets the name of the bin-start column. The default is `bin_start`. + * + * @since 4.4.0 + */ + def renameBinStart(name: String): BinByWithDistribute = { + require(name != null, "Column name cannot be null") + + new BinByWithDistribute(dataset, spec.copy(binStartName = Some(name))) + } + + /** + * Sets the name of the bin-end column. The default is `bin_end`. + * + * @since 4.4.0 + */ + def renameBinEnd(name: String): BinByWithDistribute = { + require(name != null, "Column name cannot be null") + + new BinByWithDistribute(dataset, spec.copy(binEndName = Some(name))) + } + + /** + * Sets the name of the overlap-ratio column. The default is `bin_distribute_ratio`. + * + * @since 4.4.0 + */ + def renameBinDistributeRatio(name: String): BinByWithDistribute = { + require(name != null, "Column name cannot be null") + + new BinByWithDistribute(dataset, spec.copy(binDistributeRatioName = Some(name))) + } + + /** + * Returns the DataFrame with one row for each bin that overlaps an input range. Review Comment: **Nit (P3):** This contradicts the operator for NULL bounds. `BinByExec` emits exactly one row for an input row whose range start or end is NULL, with NULL `bin_start` / `bin_end` / ratio and NULL scaled values, even though no bin overlaps it. A zero-length range also yields one row with ratio 1.0, and an inverted range raises an error. The PySpark docstring Notes already describe all three cases, but Scala/Java users reading this ScalaDoc could assume NULL-bound rows are dropped or that bin columns are never NULL, for example when counting bins or joining on `bin_start`. Consider mirroring the PySpark Notes here or in the `Dataset.binBy` overview. See **Shared repair plan 1** in the review body. ########## sql/api/src/main/scala/org/apache/spark/sql/BinByBuilder.scala: ########## @@ -0,0 +1,152 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql + +import org.apache.spark.annotation.Experimental + +/** + * Builds a BIN BY operation on a Dataset. + * + * @since 4.4.0 + */ +@Experimental +final class BinByBuilder private[sql] (dataset: Dataset[_]) { + + /** + * Sets the timestamp columns for each row's half-open range. Both columns must have the same Review Comment: **Nit (P3):** In the Dataset API a `Column` parameter usually accepts any expression, but here only a direct reference to a top-level input column passes analysis. `col("start").cast("timestamp_ntz")`, `col("start").as("s")`, `col("value") + 1` and nested fields such as `col("s.f")` all fail with `BIN_BY_REQUIRES_TOP_LEVEL_COLUMN`, as the new `BinBySuite` cases assert. The ScalaDoc here and on `distributeUniform`, and the PySpark `rangeStart` / `rangeEnd` / `distributeUniform` parameter docs, should say so and point users to project the value into a column first, for example: "Each argument must reference a top-level column of this Dataset; project computed, cast, or nested values into a column first." The error itself is clear, so this is about discoverability in the new public docs. See **Shared repair plan 1** in the review body. -- 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]
