feat(dataframe): expose sort and repartition

core/src/main/java/org/apache/datafusion/DataFrame.java

@@ -334,6 +334,87 @@ public DataFrame unnestColumns(UnnestOptions options, String... columns) {
     return new DataFrame(unnestColumns(nativeHandle, columns, options.preserveNulls()));
   }
 
+  /**
+   * Order the rows by the supplied sort keys. Each {@link SortExpr} names a column and a direction
+   * ({@link SortExpr#asc(String)} / {@link SortExpr#desc(String)}); call {@link
+   * SortExpr#nullsFirst(boolean)} to override null placement.
+   *
+   * <p>An empty {@code exprs} array is a no-op (matches DataFusion's {@code sort(vec![])}). The
+   * receiver remains usable and must still be closed independently.
+   *
+   * @throws IllegalArgumentException if {@code exprs} or any element is {@code null}.
+   * @throws RuntimeException if a sort column does not exist in this DataFrame's schema.
+   */
+  public DataFrame sort(SortExpr... exprs) {
+    if (nativeHandle == 0) {
+      throw new IllegalStateException("DataFrame is closed or already collected");
+    }
+    if (exprs == null) {
+      throw new IllegalArgumentException("sort exprs must be non-null");
+    }
+    String[] columns = new String[exprs.length];
+    boolean[] ascending = new boolean[exprs.length];
+    boolean[] nullsFirst = new boolean[exprs.length];
+    for (int i = 0; i < exprs.length; i++) {
+      SortExpr e = exprs[i];
+      if (e == null) {
+        throw new IllegalArgumentException("sort exprs[" + i + "] must be non-null");
+      }
+      columns[i] = e.column();
+      ascending[i] = e.ascending();
+      nullsFirst[i] = e.nullsFirst();
+    }
+    return new DataFrame(sortRows(nativeHandle, columns, ascending, nullsFirst));
+  }
+
+  /**
+   * Repartition this DataFrame using a round-robin scheme across {@code numPartitions} output
+   * partitions. The receiver remains usable and must still be closed independently.
+   *
+   * @throws IllegalArgumentException if {@code numPartitions <= 0}.
+   * @throws RuntimeException if the underlying repartition plan rejects the request.
+   */
+  public DataFrame repartitionRoundRobin(int numPartitions) {
+    if (nativeHandle == 0) {
+      throw new IllegalStateException("DataFrame is closed or already collected");
+    }
+    if (numPartitions <= 0) {
+      throw new IllegalArgumentException("numPartitions must be positive, was " + numPartitions);
+    }
+    return new DataFrame(repartitionRoundRobinRows(nativeHandle, numPartitions));
+  }
+
+  /**
+   * Repartition this DataFrame by hashing the named columns into {@code numPartitions} output
+   * partitions. v1 supports column-name keys only; expression keys are deferred until the Java
+   * binding gains an {@code Expr} builder. The receiver remains usable and must still be closed
+   * independently.
+   *
+   * @throws IllegalArgumentException if {@code numPartitions <= 0}, {@code columns} is {@code null}
+   *     or empty, or any element of {@code columns} is {@code null}.
+   * @throws RuntimeException if a partition column does not exist in this DataFrame's schema.
+   */
+  public DataFrame repartitionHash(int numPartitions, String... columns) {
+    if (nativeHandle == 0) {
+      throw new IllegalStateException("DataFrame is closed or already collected");
+    }
+    if (numPartitions <= 0) {
+      throw new IllegalArgumentException("numPartitions must be positive, was " + numPartitions);
+    }
+    if (columns == null) {
+      throw new IllegalArgumentException("repartitionHash columns must be non-null");
+    }
+    if (columns.length == 0) {
+      throw new IllegalArgumentException("repartitionHash requires at least one column");
+    }
+    for (int i = 0; i < columns.length; i++) {
+      if (columns[i] == null) {
+        throw new IllegalArgumentException("repartitionHash columns[" + i + "] must be non-null");
+      }
+    }
+    return new DataFrame(repartitionHashRows(nativeHandle, numPartitions, columns));
+  }
+
   /**
    * Equi-join this DataFrame with {@code right} on the named columns, using the given {@link
    * JoinType}. The receiver and {@code right} both remain usable and must still be closed
@@ -587,6 +668,13 @@ public void close() {
 
   private static native long unnestColumns(long handle, String[] columns, boolean preserveNulls);
 
+  private static native long sortRows(
+      long handle, String[] columns, boolean[] ascending, boolean[] nullsFirst);
+
+  private static native long repartitionRoundRobinRows(long handle, int numPartitions);
+
+  private static native long repartitionHashRows(long handle, int numPartitions, String[] columns);
+
   private static native long joinDataFrame(
       long leftHandle,
       long rightHandle,

core/src/main/java/org/apache/datafusion/SortExpr.java

@@ -0,0 +1,82 @@
+/*
+ * 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.datafusion;
+
+/**
+ * A single sort key passed to {@link DataFrame#sort(SortExpr...)}, mirroring DataFusion's {@code
+ * expr::Sort{ expr, asc, nulls_first }}. Build via {@link #asc(String)} or {@link #desc(String)};
+ * tweak null placement via {@link #nullsFirst(boolean)}.
+ *
+ * <p>v1 accepts column names only -- the {@code column} string is interpreted as a column
+ * reference, not a SQL expression. Complex sort keys (e.g. {@code "a + b"}) are intentionally
+ * deferred until the Java binding gains an {@code Expr} builder.
+ *
+ * <p>Defaults match DataFusion: {@link #asc(String)} sorts ascending with NULLs last; {@link
+ * #desc(String)} sorts descending with NULLs first.
+ */
+public final class SortExpr {
+
+  private final String column;
+  private final boolean ascending;
+  private boolean nullsFirst;
+
+  private SortExpr(String column, boolean ascending, boolean nullsFirst) {
+    this.column = column;
+    this.ascending = ascending;
+    this.nullsFirst = nullsFirst;
+  }
+
+  /** Sort the given column ascending, with NULLs placed last. */
+  public static SortExpr asc(String column) {
+    if (column == null) {
+      throw new IllegalArgumentException("SortExpr column must be non-null");
+    }
+    return new SortExpr(column, true, false);
+  }
+
+  /** Sort the given column descending, with NULLs placed first. */
+  public static SortExpr desc(String column) {
+    if (column == null) {
+      throw new IllegalArgumentException("SortExpr column must be non-null");
+    }
+    return new SortExpr(column, false, true);
+  }
+
+  /** Override the default NULL placement. {@code true} = NULLs first, {@code false} = last. */
+  public SortExpr nullsFirst(boolean v) {
+    this.nullsFirst = v;
+    return this;
+  }
+
+  /** The column name this sort key references. */
+  public String column() {
+    return column;
+  }
+
+  /** {@code true} if ascending, {@code false} if descending. */
+  public boolean ascending() {
+    return ascending;
+  }
+
+  /** {@code true} if NULLs are placed first, {@code false} if last. */
+  public boolean nullsFirst() {
+    return nullsFirst;
+  }
+}