fix: Fix sorting when field ids are not provided in parquet files

ice/src/main/java/com/altinity/ice/cli/internal/cmd/Insert.java

@@ -76,6 +76,7 @@
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.mapping.MappingUtil;
 import org.apache.iceberg.mapping.NameMapping;
 import org.apache.iceberg.mapping.NameMappingParser;
 import org.apache.iceberg.parquet.Parquet;
@@ -452,9 +453,17 @@ private static List<DataFile> processFile(
       throw e;
     }
 
+    MessageType type = metadata.getFileMetaData().getSchema();
+    Map<String, String> tableProps = table.properties();
+    String nameMappingString = tableProps.get(TableProperties.DEFAULT_NAME_MAPPING);
+    NameMapping nameMapping =
+        !Strings.isNullOrEmpty(nameMappingString)
+            ? NameMappingParser.fromJson(nameMappingString)
+            : MappingUtil.create(tableSchema);
+
     boolean sorted = options.assumeSorted;
     if (!sorted && sortOrder.isSorted()) {
-      var sortCheck = Sorting.checkSorted(inputFile, tableSchema, sortOrder);
+      var sortCheck = Sorting.checkSorted(inputFile, tableSchema, sortOrder, nameMapping);
       sorted = sortCheck.ok();
       if (!sorted) {
         if (options.noCopy || options.s3CopyObject) {
@@ -490,13 +499,6 @@ private static List<DataFile> processFile(
       }
     }
 
-    MessageType type = metadata.getFileMetaData().getSchema();
-    Map<String, String> tableProps = table.properties();
-    String nameMappingString = tableProps.get(TableProperties.DEFAULT_NAME_MAPPING);
-    NameMapping nameMapping =
-        !Strings.isNullOrEmpty(nameMappingString)
-            ? NameMappingParser.fromJson(nameMappingString)
-            : null;
     Schema fileSchema = MessageTypeToSchema.convert(type, nameMapping);
 
     if (!SchemaEvolution.isSubset(fileSchema, table.schema())) {
@@ -576,7 +578,8 @@ private static List<DataFile> processFile(
             inputFile,
             dstDataFileSource,
             table.properties(),
-            compressionCodecOverride);
+            compressionCodecOverride,
+            nameMapping);
       } else if (sortOrder.isSorted() && !sorted) {
         return Collections.singletonList(
             copySorted(
@@ -591,7 +594,8 @@ private static List<DataFile> processFile(
                 dataFileNamingStrategy,
                 partitionKey,
                 table.properties(),
-                compressionCodecOverride));
+                compressionCodecOverride,
+                nameMapping));
       } else {
         // Table isn't partitioned or sorted. Copy as is.
         String dstDataFile;
@@ -614,6 +618,7 @@ private static List<DataFile> processFile(
             Parquet.read(inputFile)
                 .createReaderFunc(s -> GenericParquetReaders.buildReader(tableSchema, s))
                 .project(tableSchema)
+                .withNameMapping(nameMapping)
                 .reuseContainers();
 
         Parquet.WriteBuilder writeBuilder =
@@ -671,13 +676,14 @@ private static List<DataFile> copyPartitionedAndSorted(
       InputFile inputFile,
       DataFileNamingStrategy dstDataFileSource,
       Map<String, String> tableProperties,
-      @Nullable String compressionCodecOverride)
+      @Nullable String compressionCodecOverride,
+      @Nullable NameMapping nameMapping)
       throws IOException {
     logger.info("{}: partitioning{}", file, sortOrder.isSorted() ? "+sorting" : "");
 
     // FIXME: stream to reduce memory usage
     Map<PartitionKey, List<Record>> partitionedRecords =
-        Partitioning.partition(inputFile, tableSchema, partitionSpec);
+        Partitioning.partition(inputFile, tableSchema, partitionSpec, nameMapping);
 
     // Create a comparator based on table.sortOrder()
     RecordComparator comparator =
@@ -758,7 +764,8 @@ private static DataFile copySorted(
       DataFileNamingStrategy.Name dataFileNamingStrategy,
       PartitionKey partitionKey,
       Map<String, String> tableProperties,
-      @Nullable String compressionCodecOverride)
+      @Nullable String compressionCodecOverride,
+      @Nullable NameMapping nameMapping)
       throws IOException {
     logger.info("{}: copying (sorted) to {}", file, dstDataFile);
 
@@ -770,7 +777,8 @@ private static DataFile copySorted(
     Parquet.ReadBuilder readBuilder =
         Parquet.read(inputFile)
             .createReaderFunc(s -> GenericParquetReaders.buildReader(tableSchema, s))
-            .project(tableSchema);
+            .project(tableSchema)
+            .withNameMapping(nameMapping);
 
     // Read records into memory
     List<Record> records = new ArrayList<>();

ice/src/main/java/com/altinity/ice/cli/internal/iceberg/Partitioning.java

@@ -39,6 +39,8 @@
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.mapping.MappingUtil;
+import org.apache.iceberg.mapping.NameMapping;
 import org.apache.iceberg.parquet.Parquet;
 import org.apache.iceberg.transforms.Transform;
 import org.apache.iceberg.types.Type;
@@ -315,13 +317,25 @@ private static Object decodeStatValue(Object parquetStatValue, Type icebergType)
 
   public static Map<PartitionKey, List<org.apache.iceberg.data.Record>> partition(
       InputFile inputFile, Schema tableSchema, PartitionSpec partitionSpec) throws IOException {
+    return partition(inputFile, tableSchema, partitionSpec, MappingUtil.create(tableSchema));
+  }
+
+  public static Map<PartitionKey, List<org.apache.iceberg.data.Record>> partition(
+      InputFile inputFile,
+      Schema tableSchema,
+      PartitionSpec partitionSpec,
+      @Nullable NameMapping nameMapping)
+      throws IOException {
     PartitionKey partitionKeyMold = new PartitionKey(partitionSpec, tableSchema);
     Map<PartitionKey, List<org.apache.iceberg.data.Record>> partitionedRecords = new HashMap<>();
 
     Parquet.ReadBuilder readBuilder =
         Parquet.read(inputFile)
             .createReaderFunc(s -> GenericParquetReaders.buildReader(tableSchema, s))
             .project(tableSchema);
+    if (nameMapping != null) {
+      readBuilder = readBuilder.withNameMapping(nameMapping);
+    }
 
     try (CloseableIterable<org.apache.iceberg.data.Record> records = readBuilder.build()) {
       org.apache.iceberg.data.Record partitionRecord = GenericRecord.create(tableSchema);

ice/src/main/java/com/altinity/ice/cli/internal/iceberg/Sorting.java

@@ -25,6 +25,8 @@
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.CloseableIterator;
 import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.mapping.MappingUtil;
+import org.apache.iceberg.mapping.NameMapping;
 import org.apache.iceberg.parquet.Parquet;
 import org.apache.iceberg.types.Types;
 
@@ -104,9 +106,28 @@ public static boolean isSorted(InputFile inputFile, Schema tableSchema, SortOrde
     return checkSorted(inputFile, tableSchema, sortOrder).ok;
   }
 
-  // TODO: check metadata first to avoid full scan when unsorted
+  public static boolean isSorted(
+      InputFile inputFile,
+      Schema tableSchema,
+      SortOrder sortOrder,
+      @Nullable NameMapping nameMapping)
+      throws IOException {
+    return checkSorted(inputFile, tableSchema, sortOrder, nameMapping).ok;
+  }
+
   public static SortCheckResult checkSorted(
       InputFile inputFile, Schema tableSchema, SortOrder sortOrder) throws IOException {
+    // Resolve columns by name for files that have no embedded field IDs (e.g. ClickHouse exports).
+    return checkSorted(inputFile, tableSchema, sortOrder, MappingUtil.create(tableSchema));
+  }
+
+  // TODO: check metadata first to avoid full scan when unsorted
+  public static SortCheckResult checkSorted(
+      InputFile inputFile,
+      Schema tableSchema,
+      SortOrder sortOrder,
+      @Nullable NameMapping nameMapping)
+      throws IOException {
     if (sortOrder.isUnsorted()) {
       return new SortCheckResult(false);
     }
@@ -135,11 +156,15 @@ public static SortCheckResult checkSorted(
     }
     Schema projectedSchema = new Schema(projection);
 
-    try (CloseableIterable<Record> records =
+    Parquet.ReadBuilder readBuilder =
         Parquet.read(inputFile)
             .createReaderFunc(s -> GenericParquetReaders.buildReader(projectedSchema, s))
-            .project(projectedSchema)
-            .build()) {
+            .project(projectedSchema);
+    if (nameMapping != null) {
+      readBuilder = readBuilder.withNameMapping(nameMapping);
+    }
+
+    try (CloseableIterable<Record> records = readBuilder.build()) {
 
       CloseableIterator<Record> iter = records.iterator();
       if (iter.hasNext()) {

ice/src/test/java/com/altinity/ice/cli/internal/iceberg/SortingTest.java

@@ -9,17 +9,30 @@
  */
 package com.altinity.ice.cli.internal.iceberg;
 
+import static org.apache.iceberg.types.Types.NestedField.required;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import com.altinity.ice.cli.internal.iceberg.parquet.Metadata;
 import com.altinity.ice.test.Resource;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import org.apache.avro.SchemaBuilder;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.iceberg.NullOrder;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SortDirection;
 import org.apache.iceberg.SortOrder;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.parquet.ParquetSchemaUtil;
+import org.apache.iceberg.types.Types;
+import org.apache.parquet.avro.AvroParquetWriter;
+import org.apache.parquet.hadoop.ParquetWriter;
 import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.io.LocalOutputFile;
 import org.apache.parquet.schema.MessageType;
 import org.testng.annotations.Test;
 
@@ -44,4 +57,81 @@ public void testIsSorted() throws Exception {
                     .build()))
         .isTrue();
   }
+
+  // A Parquet file without Iceberg field IDs where the sort
+  // column's physical position differs from its declared field ID. Without a name mapping, the
+  // field-ID projection falls back to physical position and reads the wrong column, so a file that
+  // is actually sorted by "b" is wrongly reported as unsorted.
+  @Test
+  public void testIsSortedNoFieldIdsMovedSortColumn() throws Exception {
+    Path file = Files.createTempFile("sorting-no-id-moved", ".parquet");
+    Files.deleteIfExists(file);
+    // Physical column order is [b, a]: fallback IDs map id 1 -> b, id 2 -> a.
+    writeNoIdParquet(
+        file,
+        List.of("b", "a"),
+        List.of(
+            Map.of("b", "1", "a", "9"), Map.of("b", "2", "a", "5"), Map.of("b", "3", "a", "1")));
+
+    // Schema declares a=id1, b=id2, so field IDs do NOT match physical positions.
+    Schema schema =
+        new Schema(
+            required(1, "a", Types.StringType.get()), required(2, "b", Types.StringType.get()));
+    SortOrder sortByB =
+        SortOrder.builderFor(schema).sortBy("b", SortDirection.ASC, NullOrder.NULLS_FIRST).build();
+
+    InputFile in = org.apache.iceberg.Files.localInput(file.toFile());
+
+    // The file is genuinely sorted by "b" (values 1,2,3). Correct name resolution must return true.
+    assertThat(Sorting.isSorted(in, schema, sortByB)).isTrue();
+  }
+
+  // Control: when physical order matches the field IDs, positional fallback happens to be correct,
+  // so this passes both before and after the fix, isolating the mismatch as the cause of failure.
+  @Test
+  public void testIsSortedNoFieldIdsAlignedSortColumn() throws Exception {
+    Path file = Files.createTempFile("sorting-no-id-aligned", ".parquet");
+    Files.deleteIfExists(file);
+    // Physical column order is [a, b]: fallback IDs map id 1 -> a, id 2 -> b.
+    writeNoIdParquet(
+        file,
+        List.of("a", "b"),
+        List.of(
+            Map.of("a", "9", "b", "1"), Map.of("a", "5", "b", "2"), Map.of("a", "1", "b", "3")));
+
+    Schema schema =
+        new Schema(
+            required(1, "a", Types.StringType.get()), required(2, "b", Types.StringType.get()));
+    SortOrder sortByB =
+        SortOrder.builderFor(schema).sortBy("b", SortDirection.ASC, NullOrder.NULLS_FIRST).build();
+
+    InputFile in = org.apache.iceberg.Files.localInput(file.toFile());
+
+    assertThat(Sorting.isSorted(in, schema, sortByB)).isTrue();
+  }
+
+  private static void writeNoIdParquet(
+      Path file, List<String> columnOrder, List<Map<String, String>> rows) throws Exception {
+    SchemaBuilder.FieldAssembler<org.apache.avro.Schema> fields =
+        SchemaBuilder.record("row").namespace("com.altinity.ice.test").fields();
+    for (String column : columnOrder) {
+      fields = fields.name(column).type().stringType().noDefault();
+    }
+    org.apache.avro.Schema avroSchema = fields.endRecord();
+
+    try (ParquetWriter<GenericRecord> writer =
+        AvroParquetWriter.<GenericRecord>builder(new LocalOutputFile(file))
+            .withSchema(avroSchema)
+            .withConf(new Configuration())
+            .withWriteMode(org.apache.parquet.hadoop.ParquetFileWriter.Mode.OVERWRITE)
+            .build()) {
+      for (Map<String, String> row : rows) {
+        GenericRecord record = new GenericData.Record(avroSchema);
+        for (String column : columnOrder) {
+          record.put(column, row.get(column));
+        }
+        writer.write(record);
+      }
+    }
+  }
 }