Revert optimization to reorder columns in parquet writer

Some files produced by this optimization were ignored by Apache Spark.
Some versions of Databricks Runtime produce an exception when reading
files with re-ordered columns.

Author: raunaqmorarka

lib/trino-parquet/src/main/java/io/trino/parquet/writer/ParquetWriter.java

@@ -29,7 +29,6 @@
 import io.trino.parquet.writer.ColumnWriter.BufferData;
 import io.trino.spi.Page;
 import io.trino.spi.type.Type;
-import it.unimi.dsi.fastutil.ints.IntArrays;
 import org.apache.parquet.column.ParquetProperties;
 import org.apache.parquet.format.ColumnMetaData;
 import org.apache.parquet.format.CompressionCodec;
@@ -46,7 +45,6 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.io.OutputStream;
-import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -312,27 +310,16 @@ private void flush()
                 .map(BufferData::getMetaData)
                 .collect(toImmutableList());
 
-        // Since the reader coalesces nearby small reads, it is beneficial to
-        // reorder data streams to group columns with small size together
-        int[] indexes = new int[columns.size()];
-        Arrays.setAll(indexes, index -> index);
-        IntArrays.quickSort(indexes, (index, otherIndex) ->
-                Long.compare(columns.get(index).getTotal_compressed_size(), columns.get(otherIndex).getTotal_compressed_size()));
-
-        // Ordering of columns in the metadata should remain unchanged.
-        // Only the offsets in file at which the columns start may change as a result
-        // of reordering column data streams by their compressed size
         long currentOffset = stripeStartOffset;
-        for (int index : indexes) {
-            ColumnMetaData columnMetaData = columns.get(index);
+        for (ColumnMetaData columnMetaData : columns) {
             columnMetaData.setData_page_offset(currentOffset);
             currentOffset += columnMetaData.getTotal_compressed_size();
         }
         updateRowGroups(columns);
 
         // flush pages
-        for (int index : indexes) {
-            bufferDataList.get(index).getData()
+        for (BufferData bufferData : bufferDataList) {
+            bufferData.getData()
                     .forEach(data -> data.writeData(outputStream));
         }
     }

lib/trino-parquet/src/test/java/io/trino/parquet/writer/TestParquetWriter.java

@@ -35,7 +35,6 @@
 import org.testng.annotations.Test;
 
 import java.io.IOException;
-import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -137,12 +136,8 @@ public void testColumnReordering()
         ParquetMetadata parquetMetadata = MetadataReader.readFooter(dataSource, Optional.empty());
         assertThat(parquetMetadata.getBlocks().size()).isGreaterThanOrEqualTo(10);
         for (BlockMetaData blockMetaData : parquetMetadata.getBlocks()) {
-            // Sort columns by size in file
-            List<ColumnChunkMetaData> columns = blockMetaData.getColumns().stream()
-                    .sorted(Comparator.comparingLong(ColumnChunkMetaData::getTotalUncompressedSize))
-                    .collect(toImmutableList());
-            // Verify that the columns are stored in the same order
-            List<Long> offsets = columns.stream()
+            // Verify that the columns are stored in the same order as the metadata
+            List<Long> offsets = blockMetaData.getColumns().stream()
                     .map(ColumnChunkMetaData::getFirstDataPageOffset)
                     .collect(toImmutableList());
             assertThat(offsets).isSorted();