[SPARK-50593][SQL] SPJ: Support truncate transform via generalized ReducibleFunction API

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/functions/ReducibleFunction.java

@@ -60,6 +60,42 @@
 @Evolving
 public interface ReducibleFunction<I, O> {
 
+  /**
+   * Generic reducer for parameterized functions (bucket, truncate, etc.).
+   *
+   * If this function is 'reducible' on another function, return the {@link Reducer}.
+   * <p>
+   * This method supports functions with any number of parameters of any type.
+   * <p>
+   * Examples:
+   * <ul>
+   *     <li>bucket(4, x) and bucket(2, x):
+   *         <br>thisParams = [4], otherParams = [2]
+   *         <br>Extract with: thisParams.getInt(0), otherParams.getInt(0)
+   *     </li>
+   *     <li>truncate(x, 3) and truncate(x, 5):
+   *         <br>thisParams = [3], otherParams = [5]
+   *         <br>Extract with: thisParams.getInt(0), otherParams.getInt(0)
+   *     </li>
+   *     <li>hypothetical range_bucket(x, 0L, 100L, 4):
+   *         <br>thisParams = [0L, 100L, 4]
+   *         <br>Extract with: thisParams.getLong(0), thisParams.getLong(1), thisParams.getInt(2)
+   *     </li>
+   * </ul>
+   *
+   * @param thisParams parameters for this function
+   * @param otherFunction the other parameterized function
+   * @param otherParams parameters for the other function
+   * @return a reduction function if reducible, null otherwise
+   * @since 5.0.0
+   */
+  default Reducer<I, O> reducer(
+          ReducibleParameters thisParams,
+          ReducibleFunction<?, ?> otherFunction,
+          ReducibleParameters otherParams) {
+    throw new UnsupportedOperationException();
+  }
+
   /**
    * This method is for the bucket function.
    *
@@ -78,7 +114,12 @@ public interface ReducibleFunction<I, O> {
    * @param otherBucketFunction the other parameterized function
    * @param otherNumBuckets parameter for the other function
    * @return a reduction function if it is reducible, null if not
+   * @deprecated as of 5.0.0. Please override
+   *     {@link #reducer(ReducibleParameters, ReducibleFunction, ReducibleParameters)} instead.
+   *     The new overload supports transforms with any number of parameters of any type
+   *     (e.g. truncate width, multi-arg range buckets), not just a single int.
    */
+  @Deprecated(since = "5.0.0")
   default Reducer<I, O> reducer(
       int thisNumBuckets,
       ReducibleFunction<?, ?> otherBucketFunction,
@@ -101,6 +142,6 @@ default Reducer<I, O> reducer(
    * @return a reduction function if it is reducible, null if not.
    */
   default Reducer<I, O> reducer(ReducibleFunction<?, ?> otherFunction) {
-    throw new UnsupportedOperationException();
+    return reducer(ReducibleParameters.EMPTY, otherFunction, ReducibleParameters.EMPTY);
   }
 }

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/functions/ReducibleParameters.java

@@ -0,0 +1,147 @@
+/*
+ * 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.connector.catalog.functions;
+
+import org.apache.spark.annotation.Evolving;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Container for reducible function literal parameters.
+ * Provides type-safe access to parameters of various types.
+ *
+ * Examples:
+ * <ul>
+ *   <li>bucket(4, col) → ReducibleParameters([4])</li>
+ *   <li>truncate(col, 3) → ReducibleParameters([3])</li>
+ *   <li>range_bucket(col, 0L, 100L, 10) → ReducibleParameters([0L, 100L, 10])</li>
+ *   <li>custom_transform(col, "param") → ReducibleParameters(["param"])</li>
+ * </ul>
+ *
+ * @since 5.0.0
+ */
+@Evolving
+public class ReducibleParameters {
+    public static final ReducibleParameters EMPTY = new ReducibleParameters();
+
+    private final List<Object> values;
+
+    private ReducibleParameters() {
+        this.values = new ArrayList<>();
+    }
+
+    public ReducibleParameters(List<Object> values) {
+        this.values = values;
+    }
+
+    public ReducibleParameters(Object... values) {
+        this.values = Arrays.asList(values);
+    }
+
+    /**
+     * Get the number of parameters.
+     */
+    public int count() {
+        return values.size();
+    }
+
+    /**
+     * Check if this container has parameters.
+     */
+    public boolean isEmpty() {
+        return values.isEmpty();
+    }
+
+    /**
+     * Get parameter at index as Integer.
+     * @throws ClassCastException if parameter is not an Integer
+     * @throws IndexOutOfBoundsException if index is invalid
+     */
+    public int getInt(int index) {
+        return (Integer) values.get(index);
+    }
+
+    /**
+     * Get parameter at index as Long.
+     * @throws ClassCastException if parameter is not a Long
+     * @throws IndexOutOfBoundsException if index is invalid
+     */
+    public long getLong(int index) {
+        return (Long) values.get(index);
+    }
+
+    /**
+     * Get parameter at index as String.
+     * @throws ClassCastException if parameter is not a String
+     * @throws IndexOutOfBoundsException if index is invalid
+     */
+    public String getString(int index) {
+        return (String) values.get(index);
+    }
+
+    /**
+     * Get parameter at index as Double.
+     * @throws ClassCastException if parameter is not a Double
+     * @throws IndexOutOfBoundsException if index is invalid
+     */
+    public double getDouble(int index) {
+        return (Double) values.get(index);
+    }
+
+    /**
+     * Get parameter at index as Float.
+     * @throws ClassCastException if parameter is not a Float
+     * @throws IndexOutOfBoundsException if index is invalid
+     */
+    public float getFloat(int index) {
+        return (Float) values.get(index);
+    }
+
+    /**
+     * Get raw parameter value at index.
+     */
+    public Object get(int index) {
+        return values.get(index);
+    }
+
+    /**
+     * Get all parameter values as a list.
+     */
+    public List<Object> getAll() {
+        return values;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ReducibleParameters that = (ReducibleParameters) o;
+        return values.equals(that.values);
+    }
+
+    @Override
+    public int hashCode() {
+        return values.hashCode();
+    }
+
+    @Override
+    public String toString() {
+        return "ReducibleParameters(" + values + ")";
+    }
+}