diff --git a/ice/src/main/java/com/altinity/ice/cli/internal/cmd/Insert.java b/ice/src/main/java/com/altinity/ice/cli/internal/cmd/Insert.java index af7ddc33..0eb6be29 100644 --- a/ice/src/main/java/com/altinity/ice/cli/internal/cmd/Insert.java +++ b/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 processFile( throw e; } + MessageType type = metadata.getFileMetaData().getSchema(); + Map 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 processFile( } } - MessageType type = metadata.getFileMetaData().getSchema(); - Map 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 processFile( inputFile, dstDataFileSource, table.properties(), - compressionCodecOverride); + compressionCodecOverride, + nameMapping); } else if (sortOrder.isSorted() && !sorted) { return Collections.singletonList( copySorted( @@ -591,7 +594,8 @@ private static List 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 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 copyPartitionedAndSorted( InputFile inputFile, DataFileNamingStrategy dstDataFileSource, Map 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> 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 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 records = new ArrayList<>(); diff --git a/ice/src/main/java/com/altinity/ice/cli/internal/iceberg/Partitioning.java b/ice/src/main/java/com/altinity/ice/cli/internal/iceberg/Partitioning.java index 14604134..aa6ce57b 100644 --- a/ice/src/main/java/com/altinity/ice/cli/internal/iceberg/Partitioning.java +++ b/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,6 +317,15 @@ private static Object decodeStatValue(Object parquetStatValue, Type icebergType) public static Map> partition( InputFile inputFile, Schema tableSchema, PartitionSpec partitionSpec) throws IOException { + return partition(inputFile, tableSchema, partitionSpec, MappingUtil.create(tableSchema)); + } + + public static Map> partition( + InputFile inputFile, + Schema tableSchema, + PartitionSpec partitionSpec, + @Nullable NameMapping nameMapping) + throws IOException { PartitionKey partitionKeyMold = new PartitionKey(partitionSpec, tableSchema); Map> partitionedRecords = new HashMap<>(); @@ -322,6 +333,9 @@ public static Map> partition( Parquet.read(inputFile) .createReaderFunc(s -> GenericParquetReaders.buildReader(tableSchema, s)) .project(tableSchema); + if (nameMapping != null) { + readBuilder = readBuilder.withNameMapping(nameMapping); + } try (CloseableIterable records = readBuilder.build()) { org.apache.iceberg.data.Record partitionRecord = GenericRecord.create(tableSchema); diff --git a/ice/src/main/java/com/altinity/ice/cli/internal/iceberg/Sorting.java b/ice/src/main/java/com/altinity/ice/cli/internal/iceberg/Sorting.java index 8471c784..79c62ff4 100644 --- a/ice/src/main/java/com/altinity/ice/cli/internal/iceberg/Sorting.java +++ b/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 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 records = readBuilder.build()) { CloseableIterator iter = records.iterator(); if (iter.hasNext()) { diff --git a/ice/src/test/java/com/altinity/ice/cli/internal/iceberg/SortingTest.java b/ice/src/test/java/com/altinity/ice/cli/internal/iceberg/SortingTest.java index 31a6ef59..ef4baa6e 100644 --- a/ice/src/test/java/com/altinity/ice/cli/internal/iceberg/SortingTest.java +++ b/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 columnOrder, List> rows) throws Exception { + SchemaBuilder.FieldAssembler 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 writer = + AvroParquetWriter.builder(new LocalOutputFile(file)) + .withSchema(avroSchema) + .withConf(new Configuration()) + .withWriteMode(org.apache.parquet.hadoop.ParquetFileWriter.Mode.OVERWRITE) + .build()) { + for (Map row : rows) { + GenericRecord record = new GenericData.Record(avroSchema); + for (String column : columnOrder) { + record.put(column, row.get(column)); + } + writer.write(record); + } + } + } }