anuragmantri commented on code in PR #16750: URL: https://github.com/apache/iceberg/pull/16750#discussion_r4365059043
########## spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkOrderingUtil.java: ########## @@ -0,0 +1,167 @@ +/* + * 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.iceberg.spark.source; + +import java.util.List; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.ScanTask; +import org.apache.iceberg.ScanTaskGroup; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SortField; +import org.apache.iceberg.SortOrder; +import org.apache.iceberg.Table; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +class SortOrderAnalyzer { + + private static final Logger LOG = LoggerFactory.getLogger(SortOrderAnalyzer.class); + + private SortOrderAnalyzer() {} + + static int reportablePrefixLength(Table table, Schema projection) { + SortOrder sortOrder = table.sortOrder(); + if (sortOrder.isUnsorted()) { + LOG.debug("Cannot report ordering for table {}: table is unsorted", table.name()); + return 0; + } + + if (!hasMergeableSortKeys(sortOrder, table.schema(), table.name())) { + return 0; + } + + int prefixLength = countReportableFields(sortOrder, projection); + if (prefixLength == 0) { + LOG.debug( + "Cannot report ordering for table {}: first sort key is not projected or not identity", + table.name()); + } + + return prefixLength; + } + + static boolean canMerge(List<? extends ScanTask> tasks, int sortOrderId) { + if (tasks.isEmpty()) { + LOG.debug("Cannot report ordering: no tasks"); + return false; + } + + for (ScanTask task : tasks) { + if (!canMerge(task, sortOrderId)) { + return false; + } + } + + return true; + } + + static boolean withinFileLimit(List<? extends ScanTaskGroup<?>> taskGroups, int maxFilesPerTask) { + for (ScanTaskGroup<?> taskGroup : taskGroups) { + if (taskGroup.tasks().size() > maxFilesPerTask) { + LOG.debug( + "Cannot report ordering: a task group has {} tasks, more than the limit of {}", + taskGroup.tasks().size(), + maxFilesPerTask); + return false; + } + } + + return true; + } + + private static boolean hasMergeableSortKeys( + SortOrder sortOrder, Schema schema, String tableName) { + List<SortField> sortFields = sortOrder.fields(); + for (int i = 0; i < sortFields.size(); i++) { + SortField sortField = sortFields.get(i); + if (!TypeUtil.ancestorFields(schema, sortField.sourceId()).isEmpty()) { Review Comment: Extracted `isUuid` and `isFloatingPoint` into `SparkOrderingUtil` and used them in `MergingSortedRowDataReader.validateSortKeys`. I kept the nested check separate on purpose. The reader accepts a nested sort key when it is projected, and the util refuses every nested key. ########## spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java: ########## @@ -123,6 +167,17 @@ public Partitioning outputPartitioning() { } } + @Override + public synchronized SortOrder[] outputOrdering() { + taskGroups(); Review Comment: Added `orderedBy=` next to `groupedBy=` in the descriptions of `SparkBatchQueryScan`, `SparkCopyOnWriteScan` and `SparkIncrementalAppendScan`, with tests for both the reported and not reported cases. ########## spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java: ########## @@ -242,6 +292,49 @@ protected synchronized List<ScanTaskGroup<T>> taskGroups() { return taskGroups; } + private List<ScanTaskGroup<T>> planGroupedTaskGroups() { + if (reportedOrdering == null) { Review Comment: Restructured so `planGroupedTaskGroups` returns early when `reportedOrdering` is set, and `reportedOrdering` is only assigned once the decision completes. A failure while planning now leaves it undecided instead of cached as no ordering. Let me know if you meant something different by the default being more robust. ########## spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SortOrderAnalyzer.java: ########## @@ -0,0 +1,167 @@ +/* + * 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.iceberg.spark.source; + +import java.util.List; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.ScanTask; +import org.apache.iceberg.ScanTaskGroup; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SortField; +import org.apache.iceberg.SortOrder; +import org.apache.iceberg.Table; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +class SortOrderAnalyzer { Review Comment: Renamed to `SparkOrderingUtil`. ########## spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatch.java: ########## @@ -86,6 +89,9 @@ class SparkBatch implements Batch { this.executorCacheLocalityEnabled = readConf.executorCacheLocalityEnabled(); this.scanHashCode = scanHashCode; this.cacheDeleteFilesOnExecutors = readConf.cacheDeleteFilesOnExecutors(); + this.orderingEnabled = orderingEnabled; + this.anyTaskNeedsMerging = + orderingEnabled && taskGroups.stream().anyMatch(group -> group.tasks().size() > 1); Review Comment: Added `SparkOrderingUtil.needsMerging`. `SparkBatch` and `SparkRowReaderFactory` both use it now, so they cannot disagree about which groups need the merging reader. ########## spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSortOrderAnalyzer.java: ########## @@ -0,0 +1,214 @@ +/* + * 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.iceberg.spark.source; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.Collections; +import org.apache.iceberg.ChangelogScanTask; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.ScanTask; +import org.apache.iceberg.ScanTaskGroup; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SortOrder; +import org.apache.iceberg.Table; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.Test; + +class TestSortOrderAnalyzer { + + private static final Schema SCHEMA = + new Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.required(2, "data", Types.StringType.get()), + Types.NestedField.required(3, "uuid", Types.UUIDType.get()), + Types.NestedField.required(4, "score", Types.DoubleType.get()), + Types.NestedField.required( + 5, + "nested", + Types.StructType.of(Types.NestedField.required(6, "key", Types.IntegerType.get())))); + + private static final SortOrder SORT_ORDER = SortOrder.builderFor(SCHEMA).asc("id").build(); + private static final int SORT_ORDER_ID = 1; Review Comment: Done, it now uses `SORT_ORDER.orderId()`. ########## docs/docs/spark-configuration.md: ########## @@ -187,6 +187,8 @@ val spark = SparkSession.builder() | spark.sql.iceberg.check-nullability | true | Validate that the write schema's nullability matches the table's nullability | | spark.sql.iceberg.check-ordering | true | Validates the write schema column order matches the table schema order | | spark.sql.iceberg.planning.preserve-data-grouping | false | When true, co-locate scan tasks for the same partition in the same read split, used in Storage Partitioned Joins | +| spark.sql.iceberg.planning.preserve-data-ordering | false | When true, report the table sort order to Spark so it can skip sorts. Requires `preserve-data-grouping` and applies only to partitioned tables. Ignored when `spark.sql.sources.v2.bucketing.allowCompatibleTransforms.enabled` or `spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled` is true. Each partition is read by a single task, and vectorized reads are disabled when a task reads more than one file | Review Comment: Trimmed both rows to their purpose. This is meant for all Spark versions. It starts with 4.1 and the other versions will follow in ports, so the rows carry no version note. I also dropped the read option for the max files setting, so only the session setting is documented. ########## spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSupportsReportOrdering.java: ########## @@ -0,0 +1,635 @@ +/* + * 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.iceberg.spark.source; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.File; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.Files; +import org.apache.iceberg.ParameterizedTestExtension; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.ReplaceSortOrder; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableUtil; +import org.apache.iceberg.TestHelpers; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.FileHelpers; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.spark.Spark3Util; +import org.apache.iceberg.spark.SparkReadOptions; +import org.apache.iceberg.spark.SparkSQLProperties; +import org.apache.iceberg.spark.TestBaseWithCatalog; +import org.apache.iceberg.types.Conversions; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.Pair; +import org.apache.spark.sql.connector.expressions.Expression; +import org.apache.spark.sql.connector.expressions.LiteralValue; +import org.apache.spark.sql.connector.expressions.SortOrder; +import org.apache.spark.sql.connector.expressions.filter.Predicate; +import org.apache.spark.sql.connector.read.ScanBuilder; +import org.apache.spark.sql.execution.SortExec; +import org.apache.spark.sql.internal.SQLConf; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; +import org.apache.spark.unsafe.types.UTF8String; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.extension.ExtendWith; + +@ExtendWith(ParameterizedTestExtension.class) +class TestSupportsReportOrdering extends TestBaseWithCatalog { + + private static final String OTHER_TABLE = "other_table"; + private static final String BUCKETED_8 = "bucketed_8"; + private static final String BUCKETED_4 = "bucketed_4"; + private static final Map<String, String> ORDERING_ENABLED = orderingConf(true); + private static final Map<String, String> ORDERING_DISABLED = orderingConf(false); + + private static Map<String, String> orderingConf(boolean preserveOrdering) { + return ImmutableMap.<String, String>builder() + .put(SparkSQLProperties.PRESERVE_DATA_ORDERING, String.valueOf(preserveOrdering)) + .put(SparkSQLProperties.PRESERVE_DATA_GROUPING, "true") + .put(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD().key(), "-1") + .put(SQLConf.ADAPTIVE_EXECUTION_ENABLED().key(), "false") + .put(SQLConf.V2_BUCKETING_ENABLED().key(), "true") + .put(SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED().key(), "true") + .put(SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION().key(), "false") + .put(SQLConf.REPLACE_HASH_WITH_SORT_AGG_ENABLED().key(), "true") + .buildOrThrow(); + } + + private static Map<String, String> orderingEnabledWith(String key, String value) { + Map<String, String> conf = Maps.newHashMap(ORDERING_ENABLED); + conf.put(key, value); + return conf; + } + + @AfterEach + void removeTables() { + sql("DROP TABLE IF EXISTS %s", tableName); + sql("DROP TABLE IF EXISTS %s", tableName(OTHER_TABLE)); + sql("DROP TABLE IF EXISTS %s", tableName(BUCKETED_8)); + sql("DROP TABLE IF EXISTS %s", tableName(BUCKETED_4)); + } + + @TestTemplate + void windowUsesReportedOrdering() { + createPartitionedTable(tableName); + + assertSortRemovedAndResultUnchanged( + "SELECT p, id, row_number() OVER (PARTITION BY p ORDER BY id) AS rn FROM %s", tableName); + } + + @TestTemplate + void sortMergeJoinUsesReportedOrdering() { + createPartitionedTable(tableName); + createPartitionedTable(tableName(OTHER_TABLE)); + + assertSortRemovedAndResultUnchanged( + "SELECT a.p, a.id, b.data FROM %s a JOIN %s b ON a.p = b.p AND a.id = b.id", + tableName, tableName(OTHER_TABLE)); + } + + @TestTemplate + void sortAggregateUsesReportedOrdering() { + createPartitionedTable(tableName); + + assertSortRemovedAndResultUnchanged( + "SELECT p, id, max(data), count(*) FROM %s GROUP BY p, id", tableName); + } + + @TestTemplate + void scanReturnsRowsInSortOrder() { + createPartitionedTable(tableName); + List<Object[]> expected = sql("SELECT id, data FROM %s WHERE p = 'A'", tableName); + + List<Object[]> rows = + sqlWithConf( + ORDERING_ENABLED, String.format("SELECT id, data FROM %s WHERE p = 'A'", tableName)); + + assertThat(rows).containsExactlyInAnyOrderElementsOf(expected); + assertThat(rows.stream().map(row -> (Integer) row[0]).toList()).isSorted(); + } + + @TestTemplate + void unprojectedSortKeyIsCorrect() { + createPartitionedTable(tableName); + + assertSortRemovedAndResultUnchanged("SELECT p, max(data) FROM %s GROUP BY p", tableName); + } + + @TestTemplate + void positionDeletesKeepReportedOrdering() throws IOException { + createPartitionedTable(tableName); + sql("INSERT INTO %s VALUES (4, 'd', 'A'), (7, 'g', 'A')", tableName); + Table table = validationCatalog.loadTable(tableIdent); + DataFile firstFile = dataFile(table, "A", 1); + List<Pair<CharSequence, Long>> deletes = Lists.newArrayList(); + for (long pos = 0; pos < firstFile.recordCount(); pos += 1) { + deletes.add(Pair.of(firstFile.location(), pos)); + } + + DeleteFile posDeletes = + FileHelpers.writeDeleteFile( + table, + Files.localOutput(File.createTempFile("junit", null, temp.toFile())), + TestHelpers.Row.of("A"), + deletes, + TableUtil.formatVersion(table)) + .first(); + table.newRowDelta().addDeletes(posDeletes).commit(); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).hasSize(2)); + assertSortRemovedAndResultUnchanged( + "SELECT p, id, row_number() OVER (PARTITION BY p ORDER BY id) AS rn FROM %s", tableName); + } + + @TestTemplate + void unsortedTableKeepsSplitPlanning() { + sql( + "CREATE TABLE %s (id INT, data STRING, p STRING) USING iceberg PARTITIONED BY (p)", + tableName); + sql("INSERT INTO %s VALUES (1, 'a', 'A'), (10, 'j', 'B')", tableName); + sql("INSERT INTO %s VALUES (2, 'b', 'A'), (11, 'k', 'B')", tableName); + Map<String, String> options = ImmutableMap.of(SparkReadOptions.SPLIT_SIZE, "1"); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SparkPartitioningAwareScan<?> scan = buildScan(tableIdent, options, null); + assertThat(scan.outputOrdering()).isEmpty(); + assertThat(scan.taskGroups()).hasSize(4); + }); + } + + @TestTemplate + void bucketedSelfJoinUsesReportedOrdering() { + createBucketedTable(tableName(BUCKETED_8), 8); + + assertSortRemovedAndResultUnchanged( + "SELECT a.id, b.data FROM %s a JOIN %s b ON a.id = b.id", + tableName(BUCKETED_8), tableName(BUCKETED_8)); + } + + @TestTemplate + void bucketedThreeWayJoinUsesReportedOrdering() { + createBucketedTable(tableName(BUCKETED_8), 8); + createBucketedTable(tableName(OTHER_TABLE), 8); + + assertSortRemovedAndResultUnchanged( + "SELECT a.id, b.data, c.data FROM %s a JOIN %s b ON a.id = b.id JOIN %s c ON b.id = c.id", + tableName(BUCKETED_8), tableName(OTHER_TABLE), tableName(BUCKETED_8)); + } + + @TestTemplate + void bucketedJoinWithSkewedKeyIsCorrect() { + createBucketedTable(tableName(BUCKETED_8), 8); + createBucketedTable(tableName(OTHER_TABLE), 8); + // every multiple of 8 lands in the same bucket, so that bucket has many more files + for (int file = 0; file < 4; file += 1) { + sql("INSERT INTO %s VALUES (%d, 'skew'), (%d, 'skew')", tableName(OTHER_TABLE), 8, 16); + } + + assertResultUnchanged( + orderingEnabledWith( + SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED().key(), "true"), + "SELECT a.id, b.data FROM %s a JOIN %s b ON a.id = b.id", + tableName(BUCKETED_8), + tableName(OTHER_TABLE)); + } + + @TestTemplate + void compatibleBucketTransformsDisableOrdering() { + createBucketedTable(tableName(BUCKETED_8), 8); + createBucketedTable(tableName(BUCKETED_4), 4); + Map<String, String> conf = + orderingEnabledWith(SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS().key(), "true"); + + withSQLConf( + conf, () -> assertThat(reportedOrdering(identifier(tableName(BUCKETED_8)))).isEmpty()); + assertResultUnchanged( + conf, + "SELECT a.id, b.data FROM %s a JOIN %s b ON a.id = b.id", + tableName(BUCKETED_8), + tableName(BUCKETED_4)); + } + + @TestTemplate + void joinKeysSubsetOfPartitionKeysDisablesOrdering() { + createPartitionedTable(tableName); + Map<String, String> conf = + orderingEnabledWith( + SQLConf.V2_BUCKETING_ALLOW_JOIN_KEYS_SUBSET_OF_PARTITION_KEYS().key(), "true"); + + withSQLConf(conf, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void unpartitionedTableDoesNotReportOrdering() { + sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName); + setSortOrder(tableIdent, "id"); + sql("INSERT INTO %s VALUES (1, 'a'), (3, 'c')", tableName); + sql("INSERT INTO %s VALUES (2, 'b'), (4, 'd')", tableName); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void orderingWithoutGroupingIsIgnored() { + createPartitionedTable(tableName); + Map<String, String> conf = + orderingEnabledWith(SparkSQLProperties.PRESERVE_DATA_GROUPING, "false"); + + withSQLConf(conf, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + assertResultUnchanged( + conf, + "SELECT p, id, row_number() OVER (PARTITION BY p ORDER BY id) AS rn FROM %s", + tableName); + } + + @TestTemplate + void uuidSortKeyDoesNotReportOrdering() { + Schema schema = + new Schema( + Types.NestedField.optional(1, "u", Types.UUIDType.get()), + Types.NestedField.optional(2, "p", Types.StringType.get())); + validationCatalog.createTable( + tableIdent, schema, PartitionSpec.builderFor(schema).identity("p").build()); + setSortOrder(tableIdent, "p", "u"); + sql("INSERT INTO %s VALUES ('%s', 'A')", tableName, "00000000-0000-0000-0000-000000000001"); + sql("INSERT INTO %s VALUES ('%s', 'A')", tableName, "ffffffff-0000-0000-0000-000000000001"); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void floatingPointSortKeyNotLastDoesNotReportOrdering() { + sql( + "CREATE TABLE %s (id INT, score DOUBLE, p STRING) USING iceberg PARTITIONED BY (p)", + tableName); + setSortOrder(tableIdent, "p", "score", "id"); + sql("INSERT INTO %s VALUES (1, 0.0, 'A'), (2, 1.5, 'A')", tableName); + sql("INSERT INTO %s VALUES (3, -0.0, 'A'), (4, 0.5, 'A')", tableName); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void nestedSortKeyDoesNotReportOrdering() { + sql( + "CREATE TABLE %s (id INT, s STRUCT<k: INT, v: STRING>, p STRING) " + + "USING iceberg PARTITIONED BY (p)", + tableName); + setSortOrder(tableIdent, "p", "s.k"); + sql("INSERT INTO %s VALUES (1, named_struct('k', 1, 'v', 'a'), 'A')", tableName); + sql("INSERT INTO %s VALUES (2, named_struct('k', 2, 'v', 'b'), 'A')", tableName); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void olderSortOrderDoesNotReportOrdering() { + createPartitionedTable(tableName); + setSortOrder(tableIdent, "p", "data"); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void equalityDeletesDoNotReportOrdering() throws IOException { + createPartitionedTable(tableName); + Table table = validationCatalog.loadTable(tableIdent); + Schema deleteSchema = table.schema().select("data"); + Record delete = GenericRecord.create(deleteSchema); + DeleteFile eqDeletes = + FileHelpers.writeDeleteFile( + table, + Files.localOutput(File.createTempFile("junit", null, temp.toFile())), + TestHelpers.Row.of("A"), + ImmutableList.of(delete.copy("data", "c")), + deleteSchema); + table.newRowDelta().addDeletes(eqDeletes).commit(); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + assertResultUnchanged( + ORDERING_ENABLED, + "SELECT p, id, row_number() OVER (PARTITION BY p ORDER BY id) AS rn FROM %s", + tableName); + } + + @TestTemplate + void tooManyFilesPerTaskRestoresSplitPlanning() { + createPartitionedTable(tableName); + sql("INSERT INTO %s VALUES (7, 'g', 'A')", tableName); + Map<String, String> options = + ImmutableMap.of( + SparkReadOptions.SPLIT_SIZE, + "1", + SparkReadOptions.PRESERVE_DATA_ORDERING_MAX_FILES_PER_TASK, + "2"); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SparkPartitioningAwareScan<?> scan = buildScan(tableIdent, options, null); + assertThat(scan.outputOrdering()).isEmpty(); + // partition A has 3 files and partition B has 2, each planned separately at split size 1 + assertThat(scan.taskGroups()).hasSize(5); + }); + } + + @TestTemplate + void filesPerTaskAtLimitReportOrdering() { + createPartitionedTable(tableName); + Map<String, String> options = + ImmutableMap.of( + SparkReadOptions.SPLIT_SIZE, + "1", + SparkReadOptions.PRESERVE_DATA_ORDERING_MAX_FILES_PER_TASK, + "2"); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SparkPartitioningAwareScan<?> scan = buildScan(tableIdent, options, null); + assertThat(scan.outputOrdering()).hasSize(2); + assertThat(scan.taskGroups()).hasSize(2); + }); + } + + @TestTemplate + void unprojectedSortKeyTrimsReportedOrdering() { + sql( + "CREATE TABLE %s (id INT, data STRING, p STRING) USING iceberg PARTITIONED BY (p)", + tableName); + setSortOrder(tableIdent, "p", "id", "data"); + sql("INSERT INTO %s VALUES (1, 'a', 'A'), (3, 'c', 'A')", tableName); + sql("INSERT INTO %s VALUES (2, 'b', 'A'), (4, 'd', 'A')", tableName); + StructType projection = + new StructType().add("id", DataTypes.IntegerType).add("p", DataTypes.StringType); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SortOrder[] ordering = + buildScan(tableIdent, ImmutableMap.of(), projection).outputOrdering(); + assertThat(ordering) + .extracting(order -> order.expression().describe()) + .containsExactly("p", "id"); + }); + } + + @TestTemplate + void runtimeFilterKeepsReportedOrdering() { + createPartitionedTable(tableName); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SparkBatchQueryScan scan = + (SparkBatchQueryScan) buildScan(tableIdent, ImmutableMap.of(), null); + SortOrder[] ordering = scan.outputOrdering(); + assertThat(ordering).hasSize(2); + + scan.filter(new Predicate[] {partitionEquals("A")}); + + assertThat(scan.outputOrdering()).isEqualTo(ordering); + assertThat(scan.taskGroups()).hasSize(1); + assertThat(scan.taskGroups().get(0).tasks()).hasSize(2); + }); + } + + @TestTemplate + void runtimeFilterKeepsRejectedOrdering() { + createPartitionedTable(tableName); + sql("INSERT INTO %s VALUES (7, 'g', 'A')", tableName); + Map<String, String> options = + ImmutableMap.of(SparkReadOptions.PRESERVE_DATA_ORDERING_MAX_FILES_PER_TASK, "2"); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SparkBatchQueryScan scan = (SparkBatchQueryScan) buildScan(tableIdent, options, null); + assertThat(scan.outputOrdering()).isEmpty(); + + // partition B alone is within the limit, but the decision must not change + scan.filter(new Predicate[] {partitionEquals("B")}); + + assertThat(scan.outputOrdering()).isEmpty(); + assertThat(scan.isOrderingEnabled()).isFalse(); + }); + } + + @TestTemplate + void singleFileTasksKeepVectorizedReads() { + sql( + "CREATE TABLE %s (id INT, data STRING, p STRING) USING iceberg PARTITIONED BY (p)", + tableName); + setSortOrder(tableIdent, "p", "id"); + sql("INSERT INTO %s VALUES (1, 'a', 'A'), (2, 'b', 'A'), (10, 'j', 'B')", tableName); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SparkPartitioningAwareScan<?> scan = buildScan(tableIdent, ImmutableMap.of(), null); + assertThat(scan.outputOrdering()).hasSize(2); + assertThat(scan.toBatch().createReaderFactory()) + .isInstanceOf(SparkColumnarReaderFactory.class); + }); + } + + @TestTemplate + void multiFileTasksUseRowReads() { + createPartitionedTable(tableName); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SparkPartitioningAwareScan<?> scan = buildScan(tableIdent, ImmutableMap.of(), null); + assertThat(scan.outputOrdering()).hasSize(2); + assertThat(scan.toBatch().createReaderFactory()) + .isInstanceOf(SparkRowReaderFactory.class); + }); + } + + @TestTemplate + void copyOnWriteMergeIsCorrect() { + createBucketedTable(tableName, 4); + sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName(OTHER_TABLE)); + sql("INSERT INTO %s VALUES (1, 'new1'), (4, 'new4'), (100, 'new100')", tableName(OTHER_TABLE)); + List<Object[]> expected = + sql( + "SELECT coalesce(s.id, t.id), coalesce(s.data, t.data) " + + "FROM %s t FULL OUTER JOIN %s s ON t.id = s.id", + tableName, tableName(OTHER_TABLE)); + + withSQLConf( + ORDERING_ENABLED, + () -> + sql( + "MERGE INTO %s t USING %s s ON t.id = s.id " + + "WHEN MATCHED THEN UPDATE SET t.data = s.data " + + "WHEN NOT MATCHED THEN INSERT *", + tableName, tableName(OTHER_TABLE))); + + assertThat(sql("SELECT id, data FROM %s", tableName)) + .containsExactlyInAnyOrderElementsOf(expected); + } + + // rows of partition A are split across two files with interleaved and duplicate IDs + private void createPartitionedTable(String name) { + sql("CREATE TABLE %s (id INT, data STRING, p STRING) USING iceberg PARTITIONED BY (p)", name); + setSortOrder(identifier(name), "p", "id"); + sql("INSERT INTO %s VALUES (1, 'a', 'A'), (3, 'c', 'A'), (5, 'e', 'A'), (10, 'j', 'B')", name); + sql("INSERT INTO %s VALUES (2, 'b', 'A'), (3, 'x', 'A'), (6, 'f', 'A'), (11, 'k', 'B')", name); + } + + // each bucket has three files with interleaved IDs + private void createBucketedTable(String name, int numBuckets) { + sql( + "CREATE TABLE %s (id INT, data STRING) USING iceberg PARTITIONED BY (bucket(%d, id))", + name, numBuckets); + setSortOrder(identifier(name), "id"); + for (int file = 0; file < 3; file += 1) { + StringBuilder values = new StringBuilder(); + for (int id = file; id < 60; id += 3) { + values + .append(values.length() == 0 ? "" : ", ") + .append(String.format("(%d, 'd%d')", id, id)); + } + + sql("INSERT INTO %s VALUES %s", name, values); + } + } + + private TableIdentifier identifier(String name) { + return TableIdentifier.of("default", name.substring(name.lastIndexOf('.') + 1)); + } + + private void setSortOrder(TableIdentifier ident, String... columns) { + ReplaceSortOrder sortOrder = validationCatalog.loadTable(ident).replaceSortOrder(); + for (String column : columns) { + sortOrder = sortOrder.asc(column); Review Comment: Added tests for descending, nulls first and nulls last. A mismatched null order test checks that the sort is kept and the results are still correct. ########## spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestBase.java: ########## @@ -53,13 +53,18 @@ import org.apache.spark.sql.execution.QueryExecution; import org.apache.spark.sql.execution.SparkPlan; import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec; +import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper; import org.apache.spark.sql.internal.SQLConf; import org.apache.spark.sql.util.QueryExecutionListener; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; +import scala.PartialFunction; +import scala.collection.JavaConverters; Review Comment: Switched to `scala.jdk.javaapi.CollectionConverters`. ########## spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java: ########## @@ -242,6 +292,49 @@ protected synchronized List<ScanTaskGroup<T>> taskGroups() { return taskGroups; } + private List<ScanTaskGroup<T>> planGroupedTaskGroups() { + if (reportedOrdering == null) { + this.reportedOrdering = NO_ORDERING; + int prefixLength = sortOrderPrefixLength(); + if (prefixLength > 0) { + List<ScanTaskGroup<T>> orderedTaskGroups = planTaskGroupsByPartitionKey(Long.MAX_VALUE); Review Comment: Added `planOneTaskGroupPerPartition()` and used it wherever we plan with an unlimited split size. ########## spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SortOrderAnalyzer.java: ########## @@ -0,0 +1,167 @@ +/* + * 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.iceberg.spark.source; + +import java.util.List; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.ScanTask; +import org.apache.iceberg.ScanTaskGroup; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SortField; +import org.apache.iceberg.SortOrder; +import org.apache.iceberg.Table; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +class SortOrderAnalyzer { + + private static final Logger LOG = LoggerFactory.getLogger(SortOrderAnalyzer.class); + + private SortOrderAnalyzer() {} + + static int reportablePrefixLength(Table table, Schema projection) { Review Comment: Done. `reportablePrefixLength` now takes the sort order, the table schema and the projection, so the unit tests no longer mock `Table`. ########## spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSupportsReportOrdering.java: ########## @@ -0,0 +1,635 @@ +/* + * 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.iceberg.spark.source; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.File; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.Files; +import org.apache.iceberg.ParameterizedTestExtension; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.ReplaceSortOrder; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableUtil; +import org.apache.iceberg.TestHelpers; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.FileHelpers; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.spark.Spark3Util; +import org.apache.iceberg.spark.SparkReadOptions; +import org.apache.iceberg.spark.SparkSQLProperties; +import org.apache.iceberg.spark.TestBaseWithCatalog; +import org.apache.iceberg.types.Conversions; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.Pair; +import org.apache.spark.sql.connector.expressions.Expression; +import org.apache.spark.sql.connector.expressions.LiteralValue; +import org.apache.spark.sql.connector.expressions.SortOrder; +import org.apache.spark.sql.connector.expressions.filter.Predicate; +import org.apache.spark.sql.connector.read.ScanBuilder; +import org.apache.spark.sql.execution.SortExec; +import org.apache.spark.sql.internal.SQLConf; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; +import org.apache.spark.unsafe.types.UTF8String; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.extension.ExtendWith; + +@ExtendWith(ParameterizedTestExtension.class) +class TestSupportsReportOrdering extends TestBaseWithCatalog { + + private static final String OTHER_TABLE = "other_table"; + private static final String BUCKETED_8 = "bucketed_8"; + private static final String BUCKETED_4 = "bucketed_4"; + private static final Map<String, String> ORDERING_ENABLED = orderingConf(true); + private static final Map<String, String> ORDERING_DISABLED = orderingConf(false); + + private static Map<String, String> orderingConf(boolean preserveOrdering) { + return ImmutableMap.<String, String>builder() + .put(SparkSQLProperties.PRESERVE_DATA_ORDERING, String.valueOf(preserveOrdering)) + .put(SparkSQLProperties.PRESERVE_DATA_GROUPING, "true") + .put(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD().key(), "-1") + .put(SQLConf.ADAPTIVE_EXECUTION_ENABLED().key(), "false") + .put(SQLConf.V2_BUCKETING_ENABLED().key(), "true") + .put(SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED().key(), "true") + .put(SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION().key(), "false") + .put(SQLConf.REPLACE_HASH_WITH_SORT_AGG_ENABLED().key(), "true") + .buildOrThrow(); + } + + private static Map<String, String> orderingEnabledWith(String key, String value) { + Map<String, String> conf = Maps.newHashMap(ORDERING_ENABLED); + conf.put(key, value); + return conf; + } + + @AfterEach + void removeTables() { + sql("DROP TABLE IF EXISTS %s", tableName); + sql("DROP TABLE IF EXISTS %s", tableName(OTHER_TABLE)); + sql("DROP TABLE IF EXISTS %s", tableName(BUCKETED_8)); + sql("DROP TABLE IF EXISTS %s", tableName(BUCKETED_4)); + } + + @TestTemplate + void windowUsesReportedOrdering() { + createPartitionedTable(tableName); + + assertSortRemovedAndResultUnchanged( + "SELECT p, id, row_number() OVER (PARTITION BY p ORDER BY id) AS rn FROM %s", tableName); + } + + @TestTemplate + void sortMergeJoinUsesReportedOrdering() { + createPartitionedTable(tableName); + createPartitionedTable(tableName(OTHER_TABLE)); + + assertSortRemovedAndResultUnchanged( + "SELECT a.p, a.id, b.data FROM %s a JOIN %s b ON a.p = b.p AND a.id = b.id", + tableName, tableName(OTHER_TABLE)); + } + + @TestTemplate + void sortAggregateUsesReportedOrdering() { + createPartitionedTable(tableName); + + assertSortRemovedAndResultUnchanged( + "SELECT p, id, max(data), count(*) FROM %s GROUP BY p, id", tableName); + } + + @TestTemplate + void scanReturnsRowsInSortOrder() { + createPartitionedTable(tableName); + List<Object[]> expected = sql("SELECT id, data FROM %s WHERE p = 'A'", tableName); + + List<Object[]> rows = + sqlWithConf( + ORDERING_ENABLED, String.format("SELECT id, data FROM %s WHERE p = 'A'", tableName)); + + assertThat(rows).containsExactlyInAnyOrderElementsOf(expected); + assertThat(rows.stream().map(row -> (Integer) row[0]).toList()).isSorted(); + } + + @TestTemplate + void unprojectedSortKeyIsCorrect() { + createPartitionedTable(tableName); + + assertSortRemovedAndResultUnchanged("SELECT p, max(data) FROM %s GROUP BY p", tableName); + } + + @TestTemplate + void positionDeletesKeepReportedOrdering() throws IOException { + createPartitionedTable(tableName); + sql("INSERT INTO %s VALUES (4, 'd', 'A'), (7, 'g', 'A')", tableName); + Table table = validationCatalog.loadTable(tableIdent); + DataFile firstFile = dataFile(table, "A", 1); + List<Pair<CharSequence, Long>> deletes = Lists.newArrayList(); + for (long pos = 0; pos < firstFile.recordCount(); pos += 1) { + deletes.add(Pair.of(firstFile.location(), pos)); + } + + DeleteFile posDeletes = + FileHelpers.writeDeleteFile( + table, + Files.localOutput(File.createTempFile("junit", null, temp.toFile())), + TestHelpers.Row.of("A"), + deletes, + TableUtil.formatVersion(table)) + .first(); + table.newRowDelta().addDeletes(posDeletes).commit(); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).hasSize(2)); + assertSortRemovedAndResultUnchanged( + "SELECT p, id, row_number() OVER (PARTITION BY p ORDER BY id) AS rn FROM %s", tableName); + } + + @TestTemplate + void unsortedTableKeepsSplitPlanning() { + sql( + "CREATE TABLE %s (id INT, data STRING, p STRING) USING iceberg PARTITIONED BY (p)", + tableName); + sql("INSERT INTO %s VALUES (1, 'a', 'A'), (10, 'j', 'B')", tableName); + sql("INSERT INTO %s VALUES (2, 'b', 'A'), (11, 'k', 'B')", tableName); + Map<String, String> options = ImmutableMap.of(SparkReadOptions.SPLIT_SIZE, "1"); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SparkPartitioningAwareScan<?> scan = buildScan(tableIdent, options, null); + assertThat(scan.outputOrdering()).isEmpty(); + assertThat(scan.taskGroups()).hasSize(4); + }); + } + + @TestTemplate + void bucketedSelfJoinUsesReportedOrdering() { + createBucketedTable(tableName(BUCKETED_8), 8); + + assertSortRemovedAndResultUnchanged( + "SELECT a.id, b.data FROM %s a JOIN %s b ON a.id = b.id", + tableName(BUCKETED_8), tableName(BUCKETED_8)); + } + + @TestTemplate + void bucketedThreeWayJoinUsesReportedOrdering() { + createBucketedTable(tableName(BUCKETED_8), 8); + createBucketedTable(tableName(OTHER_TABLE), 8); + + assertSortRemovedAndResultUnchanged( + "SELECT a.id, b.data, c.data FROM %s a JOIN %s b ON a.id = b.id JOIN %s c ON b.id = c.id", + tableName(BUCKETED_8), tableName(OTHER_TABLE), tableName(BUCKETED_8)); + } + + @TestTemplate + void bucketedJoinWithSkewedKeyIsCorrect() { + createBucketedTable(tableName(BUCKETED_8), 8); + createBucketedTable(tableName(OTHER_TABLE), 8); + // every multiple of 8 lands in the same bucket, so that bucket has many more files + for (int file = 0; file < 4; file += 1) { + sql("INSERT INTO %s VALUES (%d, 'skew'), (%d, 'skew')", tableName(OTHER_TABLE), 8, 16); + } + + assertResultUnchanged( + orderingEnabledWith( + SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED().key(), "true"), + "SELECT a.id, b.data FROM %s a JOIN %s b ON a.id = b.id", + tableName(BUCKETED_8), + tableName(OTHER_TABLE)); + } + + @TestTemplate + void compatibleBucketTransformsDisableOrdering() { + createBucketedTable(tableName(BUCKETED_8), 8); + createBucketedTable(tableName(BUCKETED_4), 4); + Map<String, String> conf = + orderingEnabledWith(SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS().key(), "true"); + + withSQLConf( + conf, () -> assertThat(reportedOrdering(identifier(tableName(BUCKETED_8)))).isEmpty()); + assertResultUnchanged( + conf, + "SELECT a.id, b.data FROM %s a JOIN %s b ON a.id = b.id", + tableName(BUCKETED_8), + tableName(BUCKETED_4)); + } + + @TestTemplate + void joinKeysSubsetOfPartitionKeysDisablesOrdering() { + createPartitionedTable(tableName); + Map<String, String> conf = + orderingEnabledWith( + SQLConf.V2_BUCKETING_ALLOW_JOIN_KEYS_SUBSET_OF_PARTITION_KEYS().key(), "true"); + + withSQLConf(conf, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void unpartitionedTableDoesNotReportOrdering() { + sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName); + setSortOrder(tableIdent, "id"); + sql("INSERT INTO %s VALUES (1, 'a'), (3, 'c')", tableName); + sql("INSERT INTO %s VALUES (2, 'b'), (4, 'd')", tableName); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void orderingWithoutGroupingIsIgnored() { + createPartitionedTable(tableName); + Map<String, String> conf = + orderingEnabledWith(SparkSQLProperties.PRESERVE_DATA_GROUPING, "false"); + + withSQLConf(conf, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + assertResultUnchanged( + conf, + "SELECT p, id, row_number() OVER (PARTITION BY p ORDER BY id) AS rn FROM %s", + tableName); + } + + @TestTemplate + void uuidSortKeyDoesNotReportOrdering() { + Schema schema = + new Schema( + Types.NestedField.optional(1, "u", Types.UUIDType.get()), + Types.NestedField.optional(2, "p", Types.StringType.get())); + validationCatalog.createTable( + tableIdent, schema, PartitionSpec.builderFor(schema).identity("p").build()); + setSortOrder(tableIdent, "p", "u"); + sql("INSERT INTO %s VALUES ('%s', 'A')", tableName, "00000000-0000-0000-0000-000000000001"); + sql("INSERT INTO %s VALUES ('%s', 'A')", tableName, "ffffffff-0000-0000-0000-000000000001"); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void floatingPointSortKeyNotLastDoesNotReportOrdering() { + sql( + "CREATE TABLE %s (id INT, score DOUBLE, p STRING) USING iceberg PARTITIONED BY (p)", + tableName); + setSortOrder(tableIdent, "p", "score", "id"); + sql("INSERT INTO %s VALUES (1, 0.0, 'A'), (2, 1.5, 'A')", tableName); + sql("INSERT INTO %s VALUES (3, -0.0, 'A'), (4, 0.5, 'A')", tableName); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void nestedSortKeyDoesNotReportOrdering() { + sql( + "CREATE TABLE %s (id INT, s STRUCT<k: INT, v: STRING>, p STRING) " + + "USING iceberg PARTITIONED BY (p)", + tableName); + setSortOrder(tableIdent, "p", "s.k"); + sql("INSERT INTO %s VALUES (1, named_struct('k', 1, 'v', 'a'), 'A')", tableName); + sql("INSERT INTO %s VALUES (2, named_struct('k', 2, 'v', 'b'), 'A')", tableName); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void olderSortOrderDoesNotReportOrdering() { + createPartitionedTable(tableName); + setSortOrder(tableIdent, "p", "data"); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void equalityDeletesDoNotReportOrdering() throws IOException { + createPartitionedTable(tableName); + Table table = validationCatalog.loadTable(tableIdent); + Schema deleteSchema = table.schema().select("data"); + Record delete = GenericRecord.create(deleteSchema); + DeleteFile eqDeletes = + FileHelpers.writeDeleteFile( + table, + Files.localOutput(File.createTempFile("junit", null, temp.toFile())), + TestHelpers.Row.of("A"), + ImmutableList.of(delete.copy("data", "c")), + deleteSchema); + table.newRowDelta().addDeletes(eqDeletes).commit(); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + assertResultUnchanged( + ORDERING_ENABLED, + "SELECT p, id, row_number() OVER (PARTITION BY p ORDER BY id) AS rn FROM %s", + tableName); + } + + @TestTemplate + void tooManyFilesPerTaskRestoresSplitPlanning() { + createPartitionedTable(tableName); + sql("INSERT INTO %s VALUES (7, 'g', 'A')", tableName); + Map<String, String> options = + ImmutableMap.of( + SparkReadOptions.SPLIT_SIZE, + "1", + SparkReadOptions.PRESERVE_DATA_ORDERING_MAX_FILES_PER_TASK, + "2"); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SparkPartitioningAwareScan<?> scan = buildScan(tableIdent, options, null); + assertThat(scan.outputOrdering()).isEmpty(); + // partition A has 3 files and partition B has 2, each planned separately at split size 1 + assertThat(scan.taskGroups()).hasSize(5); + }); + } + + @TestTemplate + void filesPerTaskAtLimitReportOrdering() { + createPartitionedTable(tableName); + Map<String, String> options = + ImmutableMap.of( + SparkReadOptions.SPLIT_SIZE, + "1", + SparkReadOptions.PRESERVE_DATA_ORDERING_MAX_FILES_PER_TASK, + "2"); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SparkPartitioningAwareScan<?> scan = buildScan(tableIdent, options, null); + assertThat(scan.outputOrdering()).hasSize(2); + assertThat(scan.taskGroups()).hasSize(2); + }); + } + + @TestTemplate + void unprojectedSortKeyTrimsReportedOrdering() { + sql( + "CREATE TABLE %s (id INT, data STRING, p STRING) USING iceberg PARTITIONED BY (p)", + tableName); + setSortOrder(tableIdent, "p", "id", "data"); + sql("INSERT INTO %s VALUES (1, 'a', 'A'), (3, 'c', 'A')", tableName); + sql("INSERT INTO %s VALUES (2, 'b', 'A'), (4, 'd', 'A')", tableName); + StructType projection = + new StructType().add("id", DataTypes.IntegerType).add("p", DataTypes.StringType); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SortOrder[] ordering = + buildScan(tableIdent, ImmutableMap.of(), projection).outputOrdering(); + assertThat(ordering) + .extracting(order -> order.expression().describe()) + .containsExactly("p", "id"); + }); + } + + @TestTemplate + void runtimeFilterKeepsReportedOrdering() { + createPartitionedTable(tableName); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SparkBatchQueryScan scan = + (SparkBatchQueryScan) buildScan(tableIdent, ImmutableMap.of(), null); + SortOrder[] ordering = scan.outputOrdering(); + assertThat(ordering).hasSize(2); + + scan.filter(new Predicate[] {partitionEquals("A")}); + + assertThat(scan.outputOrdering()).isEqualTo(ordering); + assertThat(scan.taskGroups()).hasSize(1); + assertThat(scan.taskGroups().get(0).tasks()).hasSize(2); + }); + } + + @TestTemplate + void runtimeFilterKeepsRejectedOrdering() { + createPartitionedTable(tableName); + sql("INSERT INTO %s VALUES (7, 'g', 'A')", tableName); + Map<String, String> options = + ImmutableMap.of(SparkReadOptions.PRESERVE_DATA_ORDERING_MAX_FILES_PER_TASK, "2"); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SparkBatchQueryScan scan = (SparkBatchQueryScan) buildScan(tableIdent, options, null); + assertThat(scan.outputOrdering()).isEmpty(); + + // partition B alone is within the limit, but the decision must not change + scan.filter(new Predicate[] {partitionEquals("B")}); + + assertThat(scan.outputOrdering()).isEmpty(); + assertThat(scan.isOrderingEnabled()).isFalse(); + }); + } + + @TestTemplate + void singleFileTasksKeepVectorizedReads() { + sql( + "CREATE TABLE %s (id INT, data STRING, p STRING) USING iceberg PARTITIONED BY (p)", + tableName); + setSortOrder(tableIdent, "p", "id"); + sql("INSERT INTO %s VALUES (1, 'a', 'A'), (2, 'b', 'A'), (10, 'j', 'B')", tableName); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SparkPartitioningAwareScan<?> scan = buildScan(tableIdent, ImmutableMap.of(), null); + assertThat(scan.outputOrdering()).hasSize(2); + assertThat(scan.toBatch().createReaderFactory()) + .isInstanceOf(SparkColumnarReaderFactory.class); + }); + } + + @TestTemplate + void multiFileTasksUseRowReads() { + createPartitionedTable(tableName); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SparkPartitioningAwareScan<?> scan = buildScan(tableIdent, ImmutableMap.of(), null); + assertThat(scan.outputOrdering()).hasSize(2); + assertThat(scan.toBatch().createReaderFactory()) + .isInstanceOf(SparkRowReaderFactory.class); + }); + } + + @TestTemplate + void copyOnWriteMergeIsCorrect() { Review Comment: Added merge-on-read MERGE and DELETE tests next to the copy-on-write MERGE test. Each one also checks that ordering is reported for the table before running the DML. ########## spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSupportsReportOrdering.java: ########## @@ -0,0 +1,635 @@ +/* + * 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.iceberg.spark.source; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.File; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.Files; +import org.apache.iceberg.ParameterizedTestExtension; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.ReplaceSortOrder; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableUtil; +import org.apache.iceberg.TestHelpers; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.FileHelpers; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.spark.Spark3Util; +import org.apache.iceberg.spark.SparkReadOptions; +import org.apache.iceberg.spark.SparkSQLProperties; +import org.apache.iceberg.spark.TestBaseWithCatalog; +import org.apache.iceberg.types.Conversions; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.Pair; +import org.apache.spark.sql.connector.expressions.Expression; +import org.apache.spark.sql.connector.expressions.LiteralValue; +import org.apache.spark.sql.connector.expressions.SortOrder; +import org.apache.spark.sql.connector.expressions.filter.Predicate; +import org.apache.spark.sql.connector.read.ScanBuilder; +import org.apache.spark.sql.execution.SortExec; +import org.apache.spark.sql.internal.SQLConf; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; +import org.apache.spark.unsafe.types.UTF8String; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.extension.ExtendWith; + +@ExtendWith(ParameterizedTestExtension.class) +class TestSupportsReportOrdering extends TestBaseWithCatalog { + + private static final String OTHER_TABLE = "other_table"; + private static final String BUCKETED_8 = "bucketed_8"; + private static final String BUCKETED_4 = "bucketed_4"; + private static final Map<String, String> ORDERING_ENABLED = orderingConf(true); + private static final Map<String, String> ORDERING_DISABLED = orderingConf(false); + + private static Map<String, String> orderingConf(boolean preserveOrdering) { + return ImmutableMap.<String, String>builder() + .put(SparkSQLProperties.PRESERVE_DATA_ORDERING, String.valueOf(preserveOrdering)) + .put(SparkSQLProperties.PRESERVE_DATA_GROUPING, "true") + .put(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD().key(), "-1") + .put(SQLConf.ADAPTIVE_EXECUTION_ENABLED().key(), "false") + .put(SQLConf.V2_BUCKETING_ENABLED().key(), "true") + .put(SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED().key(), "true") + .put(SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION().key(), "false") + .put(SQLConf.REPLACE_HASH_WITH_SORT_AGG_ENABLED().key(), "true") + .buildOrThrow(); + } + + private static Map<String, String> orderingEnabledWith(String key, String value) { + Map<String, String> conf = Maps.newHashMap(ORDERING_ENABLED); + conf.put(key, value); + return conf; + } + + @AfterEach + void removeTables() { + sql("DROP TABLE IF EXISTS %s", tableName); + sql("DROP TABLE IF EXISTS %s", tableName(OTHER_TABLE)); + sql("DROP TABLE IF EXISTS %s", tableName(BUCKETED_8)); + sql("DROP TABLE IF EXISTS %s", tableName(BUCKETED_4)); + } + + @TestTemplate + void windowUsesReportedOrdering() { + createPartitionedTable(tableName); + + assertSortRemovedAndResultUnchanged( + "SELECT p, id, row_number() OVER (PARTITION BY p ORDER BY id) AS rn FROM %s", tableName); + } + + @TestTemplate + void sortMergeJoinUsesReportedOrdering() { + createPartitionedTable(tableName); + createPartitionedTable(tableName(OTHER_TABLE)); + + assertSortRemovedAndResultUnchanged( + "SELECT a.p, a.id, b.data FROM %s a JOIN %s b ON a.p = b.p AND a.id = b.id", + tableName, tableName(OTHER_TABLE)); + } + + @TestTemplate + void sortAggregateUsesReportedOrdering() { + createPartitionedTable(tableName); + + assertSortRemovedAndResultUnchanged( + "SELECT p, id, max(data), count(*) FROM %s GROUP BY p, id", tableName); + } + + @TestTemplate + void scanReturnsRowsInSortOrder() { + createPartitionedTable(tableName); + List<Object[]> expected = sql("SELECT id, data FROM %s WHERE p = 'A'", tableName); + + List<Object[]> rows = + sqlWithConf( + ORDERING_ENABLED, String.format("SELECT id, data FROM %s WHERE p = 'A'", tableName)); + + assertThat(rows).containsExactlyInAnyOrderElementsOf(expected); + assertThat(rows.stream().map(row -> (Integer) row[0]).toList()).isSorted(); + } + + @TestTemplate + void unprojectedSortKeyIsCorrect() { + createPartitionedTable(tableName); + + assertSortRemovedAndResultUnchanged("SELECT p, max(data) FROM %s GROUP BY p", tableName); + } + + @TestTemplate + void positionDeletesKeepReportedOrdering() throws IOException { + createPartitionedTable(tableName); + sql("INSERT INTO %s VALUES (4, 'd', 'A'), (7, 'g', 'A')", tableName); + Table table = validationCatalog.loadTable(tableIdent); + DataFile firstFile = dataFile(table, "A", 1); + List<Pair<CharSequence, Long>> deletes = Lists.newArrayList(); + for (long pos = 0; pos < firstFile.recordCount(); pos += 1) { + deletes.add(Pair.of(firstFile.location(), pos)); + } + + DeleteFile posDeletes = + FileHelpers.writeDeleteFile( + table, + Files.localOutput(File.createTempFile("junit", null, temp.toFile())), + TestHelpers.Row.of("A"), + deletes, + TableUtil.formatVersion(table)) + .first(); + table.newRowDelta().addDeletes(posDeletes).commit(); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).hasSize(2)); + assertSortRemovedAndResultUnchanged( + "SELECT p, id, row_number() OVER (PARTITION BY p ORDER BY id) AS rn FROM %s", tableName); + } + + @TestTemplate + void unsortedTableKeepsSplitPlanning() { + sql( + "CREATE TABLE %s (id INT, data STRING, p STRING) USING iceberg PARTITIONED BY (p)", + tableName); + sql("INSERT INTO %s VALUES (1, 'a', 'A'), (10, 'j', 'B')", tableName); + sql("INSERT INTO %s VALUES (2, 'b', 'A'), (11, 'k', 'B')", tableName); + Map<String, String> options = ImmutableMap.of(SparkReadOptions.SPLIT_SIZE, "1"); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SparkPartitioningAwareScan<?> scan = buildScan(tableIdent, options, null); + assertThat(scan.outputOrdering()).isEmpty(); + assertThat(scan.taskGroups()).hasSize(4); + }); + } + + @TestTemplate + void bucketedSelfJoinUsesReportedOrdering() { + createBucketedTable(tableName(BUCKETED_8), 8); + + assertSortRemovedAndResultUnchanged( + "SELECT a.id, b.data FROM %s a JOIN %s b ON a.id = b.id", + tableName(BUCKETED_8), tableName(BUCKETED_8)); + } + + @TestTemplate + void bucketedThreeWayJoinUsesReportedOrdering() { + createBucketedTable(tableName(BUCKETED_8), 8); + createBucketedTable(tableName(OTHER_TABLE), 8); + + assertSortRemovedAndResultUnchanged( + "SELECT a.id, b.data, c.data FROM %s a JOIN %s b ON a.id = b.id JOIN %s c ON b.id = c.id", + tableName(BUCKETED_8), tableName(OTHER_TABLE), tableName(BUCKETED_8)); + } + + @TestTemplate + void bucketedJoinWithSkewedKeyIsCorrect() { + createBucketedTable(tableName(BUCKETED_8), 8); + createBucketedTable(tableName(OTHER_TABLE), 8); + // every multiple of 8 lands in the same bucket, so that bucket has many more files + for (int file = 0; file < 4; file += 1) { + sql("INSERT INTO %s VALUES (%d, 'skew'), (%d, 'skew')", tableName(OTHER_TABLE), 8, 16); + } + + assertResultUnchanged( + orderingEnabledWith( + SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED().key(), "true"), + "SELECT a.id, b.data FROM %s a JOIN %s b ON a.id = b.id", + tableName(BUCKETED_8), + tableName(OTHER_TABLE)); + } + + @TestTemplate + void compatibleBucketTransformsDisableOrdering() { + createBucketedTable(tableName(BUCKETED_8), 8); + createBucketedTable(tableName(BUCKETED_4), 4); + Map<String, String> conf = + orderingEnabledWith(SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS().key(), "true"); + + withSQLConf( + conf, () -> assertThat(reportedOrdering(identifier(tableName(BUCKETED_8)))).isEmpty()); + assertResultUnchanged( + conf, + "SELECT a.id, b.data FROM %s a JOIN %s b ON a.id = b.id", + tableName(BUCKETED_8), + tableName(BUCKETED_4)); + } + + @TestTemplate + void joinKeysSubsetOfPartitionKeysDisablesOrdering() { + createPartitionedTable(tableName); + Map<String, String> conf = + orderingEnabledWith( + SQLConf.V2_BUCKETING_ALLOW_JOIN_KEYS_SUBSET_OF_PARTITION_KEYS().key(), "true"); + + withSQLConf(conf, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void unpartitionedTableDoesNotReportOrdering() { + sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName); + setSortOrder(tableIdent, "id"); + sql("INSERT INTO %s VALUES (1, 'a'), (3, 'c')", tableName); + sql("INSERT INTO %s VALUES (2, 'b'), (4, 'd')", tableName); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void orderingWithoutGroupingIsIgnored() { + createPartitionedTable(tableName); + Map<String, String> conf = + orderingEnabledWith(SparkSQLProperties.PRESERVE_DATA_GROUPING, "false"); + + withSQLConf(conf, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + assertResultUnchanged( + conf, + "SELECT p, id, row_number() OVER (PARTITION BY p ORDER BY id) AS rn FROM %s", + tableName); + } + + @TestTemplate + void uuidSortKeyDoesNotReportOrdering() { + Schema schema = + new Schema( + Types.NestedField.optional(1, "u", Types.UUIDType.get()), + Types.NestedField.optional(2, "p", Types.StringType.get())); + validationCatalog.createTable( + tableIdent, schema, PartitionSpec.builderFor(schema).identity("p").build()); + setSortOrder(tableIdent, "p", "u"); + sql("INSERT INTO %s VALUES ('%s', 'A')", tableName, "00000000-0000-0000-0000-000000000001"); + sql("INSERT INTO %s VALUES ('%s', 'A')", tableName, "ffffffff-0000-0000-0000-000000000001"); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void floatingPointSortKeyNotLastDoesNotReportOrdering() { + sql( + "CREATE TABLE %s (id INT, score DOUBLE, p STRING) USING iceberg PARTITIONED BY (p)", + tableName); + setSortOrder(tableIdent, "p", "score", "id"); + sql("INSERT INTO %s VALUES (1, 0.0, 'A'), (2, 1.5, 'A')", tableName); + sql("INSERT INTO %s VALUES (3, -0.0, 'A'), (4, 0.5, 'A')", tableName); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void nestedSortKeyDoesNotReportOrdering() { + sql( + "CREATE TABLE %s (id INT, s STRUCT<k: INT, v: STRING>, p STRING) " + + "USING iceberg PARTITIONED BY (p)", + tableName); + setSortOrder(tableIdent, "p", "s.k"); + sql("INSERT INTO %s VALUES (1, named_struct('k', 1, 'v', 'a'), 'A')", tableName); + sql("INSERT INTO %s VALUES (2, named_struct('k', 2, 'v', 'b'), 'A')", tableName); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void olderSortOrderDoesNotReportOrdering() { + createPartitionedTable(tableName); + setSortOrder(tableIdent, "p", "data"); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + } + + @TestTemplate + void equalityDeletesDoNotReportOrdering() throws IOException { + createPartitionedTable(tableName); + Table table = validationCatalog.loadTable(tableIdent); + Schema deleteSchema = table.schema().select("data"); + Record delete = GenericRecord.create(deleteSchema); + DeleteFile eqDeletes = + FileHelpers.writeDeleteFile( + table, + Files.localOutput(File.createTempFile("junit", null, temp.toFile())), + TestHelpers.Row.of("A"), + ImmutableList.of(delete.copy("data", "c")), + deleteSchema); + table.newRowDelta().addDeletes(eqDeletes).commit(); + + withSQLConf(ORDERING_ENABLED, () -> assertThat(reportedOrdering(tableIdent)).isEmpty()); + assertResultUnchanged( + ORDERING_ENABLED, + "SELECT p, id, row_number() OVER (PARTITION BY p ORDER BY id) AS rn FROM %s", + tableName); + } + + @TestTemplate + void tooManyFilesPerTaskRestoresSplitPlanning() { + createPartitionedTable(tableName); + sql("INSERT INTO %s VALUES (7, 'g', 'A')", tableName); + Map<String, String> options = + ImmutableMap.of( + SparkReadOptions.SPLIT_SIZE, + "1", + SparkReadOptions.PRESERVE_DATA_ORDERING_MAX_FILES_PER_TASK, + "2"); + + withSQLConf( + ORDERING_ENABLED, + () -> { + SparkPartitioningAwareScan<?> scan = buildScan(tableIdent, options, null); + assertThat(scan.outputOrdering()).isEmpty(); + // partition A has 3 files and partition B has 2, each planned separately at split size 1 + assertThat(scan.taskGroups()).hasSize(5); Review Comment: The expected count is now computed from the planned files. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
