[jira] [Created] (SPARK-44512) dataset.sort.select.write.partitionBy does not return a sorted output if AQE is enabled

["Yiu-Chung Lee (Jira)" <ji...@apache.org>]

Yiu-Chung Lee created SPARK-44512:
-------------------------------------

             Summary: dataset.sort.select.write.partitionBy does not return a sorted output if AQE is enabled
                 Key: SPARK-44512
                 URL: https://issues.apache.org/jira/browse/SPARK-44512
             Project: Spark
          Issue Type: Bug
          Components: SQL
    Affects Versions: 3.4.1
         Environment: Code that replicates the problem:

 

import java.io.IOException;
import java.util.List;
import java.util.function.Function;
import org.apache.hadoop.fs.FileSystem;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import scala.Tuple3;

// To compile: javac Test.java && jar cvf Test.jar Test.class
// bug: spark-submit --class Test Test.jar
// no bug: spark-submit --class Test Test.jar workaround
// no bug: spark-submit --conf spark.sql.adaptive.enabled=false --class Test Test.jar (3 output files in each partition key)

public class Test {
    public static void main(String args[]) throws IOException {
        final var spark = SparkSession
            .builder()
            .config("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
            .getOrCreate();
        final var hadoopConf = spark.sparkContext().hadoopConfiguration();
        final var fs = FileSystem.get(hadoopConf);
        fs.setWriteChecksum(false);

        // create a minimal dataset that is enough to reproduce the bug
        // The three columns are named _1, _2, and _3 (the field names of Tuple3)
        var dataset = spark.createDataset(List.of(
                new Tuple3<Long, String, String>(3L, "a", "r"),
                new Tuple3<Long, String, String>(3L, "b", "r"),
                new Tuple3<Long, String, String>(2L, "b", "q"),
                new Tuple3<Long, String, String>(2L, "a", "q"),
                new Tuple3<Long, String, String>(1L, "a", "p"),
                new Tuple3<Long, String, String>(1L, "b", "p")
            ),
            Encoders.tuple(Encoders.LONG(), Encoders.STRING(), Encoders.STRING()))
        .sort("_1")
        .select("_2", "_3");

        // This is an identity mapper, i.e. returns itself
        // Enabled by adding an argument "workaround" when executing spark-submit.
        // With AQE enabled, .sort() will work as intended only if this identity mapper
        // is inserted between .sort() and .select() in the pipeline
        if (args.length > 0 && args[0].equals("workaround")) {
            dataset = dataset.map((MapFunction<Row, Row>) row -> row, dataset.encoder());
        }

        // output column _3 to text files, partitioned by column _2
        // _1 is only for sorting purpose, not used in output
        // output will not be sorted without the identity mapper
        dataset.write()
            .mode("overwrite")
            .partitionBy("_2")
            .text("output");
        dataset.explain();
        spark.close();
    }
}
            Reporter: Yiu-Chung Lee
         Attachments: Test.java

(In this example the dataset is of type Tuple3, and the columns are named _1, _2 and _3)

 

I found then when AQE is enabled, the following code does not produce an output

{{dataset.sort("_1")}}
{{.select("_2", "_3")}}
{{.write()}}
{{.partitionBy("_2")}}
{{.text("output");}}

 

However, if I insert an identity mapper between select and write, the output would be sorted as expected.

{{dataset.sort("_1")}}
{{.select("_2", "_3")}}
{{.map((MapFunction<Row, Row>) row -> row, dataset.encoder())}}
{{{}.write(){}}}{{{}.partitionBy("_2"){}}}
{{.text("output")}}

Below is the complete code that replicates the problem.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org