This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 1498feeebe Spark: Compare filtered scan results by their unique record
IDs so tests do not depend on Spark task or row ordering (#18645)
1498feeebe is described below
commit 1498feeebe44c43070480fbfe462664897e9d731
Author: Simranjeet Singh <[email protected]>
AuthorDate: Fri Oct 30 00:06:22 2026 +0530
Spark: Compare filtered scan results by their unique record IDs so tests do
not depend on Spark task or row ordering (#18645)
---
.../iceberg/spark/source/TestFilteredScan.java | 26 +++++++---------------
.../iceberg/spark/source/TestFilteredScan.java | 26 +++++++---------------
.../iceberg/spark/source/TestFilteredScan.java | 26 +++++++---------------
.../iceberg/spark/source/TestFilteredScan.java | 26 +++++++---------------
4 files changed, 32 insertions(+), 72 deletions(-)
diff --git
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
index 26c2b6ab70..9334d77f08 100644
---
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
+++
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
@@ -31,6 +31,7 @@ import java.net.InetAddress;
import java.nio.file.Path;
import java.time.OffsetDateTime;
import java.util.Arrays;
+import java.util.Comparator;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;
@@ -61,7 +62,6 @@ import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
-import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
import org.apache.spark.sql.connector.expressions.filter.Predicate;
import org.apache.spark.sql.connector.read.Batch;
import org.apache.spark.sql.connector.read.InputPartition;
@@ -474,7 +474,7 @@ public class TestFilteredScan {
assertEqualsSafe(
SCHEMA.asStruct(),
- expected(8, 9, 7, 6, 5),
+ expected(5, 6, 7, 8, 9),
read(
table.location(), vectorized, "ts < cast('2017-12-22
00:00:00+00:00' as timestamp)"));
}
@@ -495,7 +495,7 @@ public class TestFilteredScan {
assertEqualsSafe(
SCHEMA.asStruct(),
- expected(2, 1),
+ expected(1, 2),
read(
table.location(),
vectorized,
@@ -654,24 +654,12 @@ public class TestFilteredScan {
return result;
}
- public static void assertEqualsUnsafe(
- Types.StructType struct, List<Record> expected, List<UnsafeRow> actual) {
- // TODO: match records by ID
- int numRecords = Math.min(expected.size(), actual.size());
- for (int i = 0; i < numRecords; i += 1) {
- GenericsHelpers.assertEqualsUnsafe(struct, expected.get(i),
actual.get(i));
- }
- assertThat(actual).as("Number of results should match
expected").hasSameSizeAs(expected);
- }
-
public static void assertEqualsSafe(
Types.StructType struct, List<Record> expected, List<Row> actual) {
- // TODO: match records by ID
- int numRecords = Math.min(expected.size(), actual.size());
- for (int i = 0; i < numRecords; i += 1) {
+ assertThat(actual).as("Should have the expected number of
rows").hasSameSizeAs(expected);
+ for (int i = 0; i < expected.size(); i += 1) {
GenericsHelpers.assertEqualsSafe(struct, expected.get(i), actual.get(i));
}
- assertThat(actual).as("Number of results should match
expected").hasSameSizeAs(expected);
}
private List<Record> expected(int... ordinals) {
@@ -745,7 +733,9 @@ public class TestFilteredScan {
.load(table)
.filter(expr)
.select(select0, selectN);
- return dataset.collectAsList();
+ List<Row> rows = dataset.collectAsList();
+ rows.sort(Comparator.comparingLong(row -> row.<Long>getAs("id")));
+ return rows;
}
private static OffsetDateTime parse(String timestamp) {
diff --git
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
index 26c2b6ab70..9334d77f08 100644
---
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
+++
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
@@ -31,6 +31,7 @@ import java.net.InetAddress;
import java.nio.file.Path;
import java.time.OffsetDateTime;
import java.util.Arrays;
+import java.util.Comparator;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;
@@ -61,7 +62,6 @@ import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
-import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
import org.apache.spark.sql.connector.expressions.filter.Predicate;
import org.apache.spark.sql.connector.read.Batch;
import org.apache.spark.sql.connector.read.InputPartition;
@@ -474,7 +474,7 @@ public class TestFilteredScan {
assertEqualsSafe(
SCHEMA.asStruct(),
- expected(8, 9, 7, 6, 5),
+ expected(5, 6, 7, 8, 9),
read(
table.location(), vectorized, "ts < cast('2017-12-22
00:00:00+00:00' as timestamp)"));
}
@@ -495,7 +495,7 @@ public class TestFilteredScan {
assertEqualsSafe(
SCHEMA.asStruct(),
- expected(2, 1),
+ expected(1, 2),
read(
table.location(),
vectorized,
@@ -654,24 +654,12 @@ public class TestFilteredScan {
return result;
}
- public static void assertEqualsUnsafe(
- Types.StructType struct, List<Record> expected, List<UnsafeRow> actual) {
- // TODO: match records by ID
- int numRecords = Math.min(expected.size(), actual.size());
- for (int i = 0; i < numRecords; i += 1) {
- GenericsHelpers.assertEqualsUnsafe(struct, expected.get(i),
actual.get(i));
- }
- assertThat(actual).as("Number of results should match
expected").hasSameSizeAs(expected);
- }
-
public static void assertEqualsSafe(
Types.StructType struct, List<Record> expected, List<Row> actual) {
- // TODO: match records by ID
- int numRecords = Math.min(expected.size(), actual.size());
- for (int i = 0; i < numRecords; i += 1) {
+ assertThat(actual).as("Should have the expected number of
rows").hasSameSizeAs(expected);
+ for (int i = 0; i < expected.size(); i += 1) {
GenericsHelpers.assertEqualsSafe(struct, expected.get(i), actual.get(i));
}
- assertThat(actual).as("Number of results should match
expected").hasSameSizeAs(expected);
}
private List<Record> expected(int... ordinals) {
@@ -745,7 +733,9 @@ public class TestFilteredScan {
.load(table)
.filter(expr)
.select(select0, selectN);
- return dataset.collectAsList();
+ List<Row> rows = dataset.collectAsList();
+ rows.sort(Comparator.comparingLong(row -> row.<Long>getAs("id")));
+ return rows;
}
private static OffsetDateTime parse(String timestamp) {
diff --git
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
index 24fecf4eb2..4b7ebec1e2 100644
---
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
+++
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
@@ -31,6 +31,7 @@ import java.net.InetAddress;
import java.nio.file.Path;
import java.time.OffsetDateTime;
import java.util.Arrays;
+import java.util.Comparator;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;
@@ -61,7 +62,6 @@ import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
-import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
import org.apache.spark.sql.connector.expressions.filter.Predicate;
import org.apache.spark.sql.connector.read.Batch;
import org.apache.spark.sql.connector.read.InputPartition;
@@ -457,7 +457,7 @@ public class TestFilteredScan {
assertEqualsSafe(
SCHEMA.asStruct(),
- expected(8, 9, 7, 6, 5),
+ expected(5, 6, 7, 8, 9),
read(
table.location(), vectorized, "ts < cast('2017-12-22
00:00:00+00:00' as timestamp)"));
}
@@ -478,7 +478,7 @@ public class TestFilteredScan {
assertEqualsSafe(
SCHEMA.asStruct(),
- expected(2, 1),
+ expected(1, 2),
read(
table.location(),
vectorized,
@@ -637,24 +637,12 @@ public class TestFilteredScan {
return result;
}
- public static void assertEqualsUnsafe(
- Types.StructType struct, List<Record> expected, List<UnsafeRow> actual) {
- // TODO: match records by ID
- int numRecords = Math.min(expected.size(), actual.size());
- for (int i = 0; i < numRecords; i += 1) {
- GenericsHelpers.assertEqualsUnsafe(struct, expected.get(i),
actual.get(i));
- }
- assertThat(actual).as("Number of results should match
expected").hasSameSizeAs(expected);
- }
-
public static void assertEqualsSafe(
Types.StructType struct, List<Record> expected, List<Row> actual) {
- // TODO: match records by ID
- int numRecords = Math.min(expected.size(), actual.size());
- for (int i = 0; i < numRecords; i += 1) {
+ assertThat(actual).as("Should have the expected number of
rows").hasSameSizeAs(expected);
+ for (int i = 0; i < expected.size(); i += 1) {
GenericsHelpers.assertEqualsSafe(struct, expected.get(i), actual.get(i));
}
- assertThat(actual).as("Number of results should match
expected").hasSameSizeAs(expected);
}
private List<Record> expected(int... ordinals) {
@@ -728,7 +716,9 @@ public class TestFilteredScan {
.load(table)
.filter(expr)
.select(select0, selectN);
- return dataset.collectAsList();
+ List<Row> rows = dataset.collectAsList();
+ rows.sort(Comparator.comparingLong(row -> row.<Long>getAs("id")));
+ return rows;
}
private static OffsetDateTime parse(String timestamp) {
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
index 57b833a16c..3b5f95cc69 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
@@ -31,6 +31,7 @@ import java.net.InetAddress;
import java.nio.file.Path;
import java.time.OffsetDateTime;
import java.util.Arrays;
+import java.util.Comparator;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;
@@ -62,7 +63,6 @@ import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.catalyst.InternalRow;
-import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
import org.apache.spark.sql.connector.expressions.NamedReference;
import org.apache.spark.sql.connector.expressions.PartitionFieldReference;
import org.apache.spark.sql.connector.expressions.filter.PartitionPredicate;
@@ -463,7 +463,7 @@ public class TestFilteredScan {
assertEqualsSafe(
SCHEMA.asStruct(),
- expected(8, 9, 7, 6, 5),
+ expected(5, 6, 7, 8, 9),
read(
table.location(), vectorized, "ts < cast('2017-12-22
00:00:00+00:00' as timestamp)"));
}
@@ -484,7 +484,7 @@ public class TestFilteredScan {
assertEqualsSafe(
SCHEMA.asStruct(),
- expected(2, 1),
+ expected(1, 2),
read(
table.location(),
vectorized,
@@ -690,24 +690,12 @@ public class TestFilteredScan {
return result;
}
- public static void assertEqualsUnsafe(
- Types.StructType struct, List<Record> expected, List<UnsafeRow> actual) {
- // TODO: match records by ID
- int numRecords = Math.min(expected.size(), actual.size());
- for (int i = 0; i < numRecords; i += 1) {
- GenericsHelpers.assertEqualsUnsafe(struct, expected.get(i),
actual.get(i));
- }
- assertThat(actual).as("Number of results should match
expected").hasSameSizeAs(expected);
- }
-
public static void assertEqualsSafe(
Types.StructType struct, List<Record> expected, List<Row> actual) {
- // TODO: match records by ID
- int numRecords = Math.min(expected.size(), actual.size());
- for (int i = 0; i < numRecords; i += 1) {
+ assertThat(actual).as("Should have the expected number of
rows").hasSameSizeAs(expected);
+ for (int i = 0; i < expected.size(); i += 1) {
GenericsHelpers.assertEqualsSafe(struct, expected.get(i), actual.get(i));
}
- assertThat(actual).as("Number of results should match
expected").hasSameSizeAs(expected);
}
private List<Record> expected(int... ordinals) {
@@ -814,7 +802,9 @@ public class TestFilteredScan {
.load(table)
.filter(expr)
.select(select0, selectN);
- return dataset.collectAsList();
+ List<Row> rows = dataset.collectAsList();
+ rows.sort(Comparator.comparingLong(row -> row.<Long>getAs("id")));
+ return rows;
}
private static OffsetDateTime parse(String timestamp) {
