You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ru...@apache.org on 2026/10/07 21:03:10 UTC
(iceberg) branch main updated: Spark 3.5, 4.0, 4.1: Backport changelog binary-value cancellation fix to 3.5, 4.0 and 4.1 (#18415)
This is an automated email from the ASF dual-hosted git repository.
RussellSpitzer 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 bd97530de9 Spark 3.5, 4.0, 4.1: Backport changelog binary-value cancellation fix to 3.5, 4.0 and 4.1 (#18415)
bd97530de9 is described below
commit bd97530de918f552c39208f0b00c4a4e5d2080a6
Author: Saath <63...@users.noreply.github.com>
AuthorDate: Wed Oct 7 22:03:04 2026 +0100
Spark 3.5, 4.0, 4.1: Backport changelog binary-value cancellation fix to 3.5, 4.0 and 4.1 (#18415)
---
.../TestCreateChangelogViewProcedure.java | 28 ++
.../apache/iceberg/spark/ChangelogIterator.java | 48 +++-
.../iceberg/spark/TestChangelogIterator.java | 301 +++++++++++++++------
.../TestCreateChangelogViewProcedure.java | 28 ++
.../apache/iceberg/spark/ChangelogIterator.java | 48 +++-
.../iceberg/spark/TestChangelogIterator.java | 301 +++++++++++++++------
.../TestCreateChangelogViewProcedure.java | 28 ++
.../apache/iceberg/spark/ChangelogIterator.java | 48 +++-
.../iceberg/spark/TestChangelogIterator.java | 301 +++++++++++++++------
9 files changed, 888 insertions(+), 243 deletions(-)
diff --git a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
index 3734d24e96..51be436921 100644
--- a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
+++ b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
@@ -57,6 +57,22 @@ public class TestCreateChangelogViewProcedure extends ExtensionsTestBase {
sql("ALTER TABLE %s ADD PARTITION FIELD id", tableName);
}
+ /** The initial and updated values must encode 0x01 and 0x02 respectively. */
+ private void assertBinaryNetChanges(String initial, String updated, String projection) {
+ sql("INSERT INTO %s VALUES (1, %s), (2, %s)", tableName, initial, initial);
+ sql("INSERT OVERWRITE %s VALUES (1, %s), (2, %s)", tableName, updated, initial);
+
+ List<Object[]> result =
+ sql(
+ "CALL %s.system.create_changelog_view(table => '%s', net_changes => true)",
+ catalogName, tableName);
+ String viewName = (String) result.get(0)[0];
+ assertEquals(
+ "Net changes should contain only the final rows",
+ ImmutableList.of(row(1, "02", INSERT), row(2, "01", INSERT)),
+ sql("SELECT id, %s, _change_type FROM %s ORDER BY id", projection, viewName));
+ }
+
private void createTableWithIdentifierField() {
sql("CREATE TABLE %s (id INT NOT NULL, data STRING) USING iceberg", tableName);
sql("ALTER TABLE %s SET IDENTIFIER FIELDS id", tableName);
@@ -613,6 +629,18 @@ public class TestCreateChangelogViewProcedure extends ExtensionsTestBase {
sql("select * from %s order by _change_ordinal, id, data", viewName));
}
+ @TestTemplate
+ public void testNetChangesWithBinaryValues() {
+ sql("CREATE TABLE %s (id INT, data BINARY) USING iceberg", tableName);
+ assertBinaryNetChanges("X'01'", "X'02'", "hex(data)");
+ }
+
+ @TestTemplate
+ public void testNetChangesWithNestedBinaryValues() {
+ sql("CREATE TABLE %s (id INT, data ARRAY<ARRAY<BINARY>>) USING iceberg", tableName);
+ assertBinaryNetChanges("array(array(X'01'))", "array(array(X'02'))", "hex(data[0][0])");
+ }
+
@TestTemplate
public void testNetChangesWithRemoveCarryOvers() {
// partitioned by id
diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java
index cc44b1f399..12ed77651a 100644
--- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java
+++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java
@@ -27,6 +27,7 @@ import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Iterators;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.StructType;
+import scala.collection.Seq;
/** An iterator that transforms rows from changelog tables within a single Spark task. */
public abstract class ChangelogIterator implements Iterator<Row> {
@@ -109,7 +110,52 @@ public abstract class ChangelogIterator implements Iterator<Row> {
}
protected boolean isDifferentValue(Row currentRow, Row nextRow, int idx) {
- return !Objects.equals(nextRow.get(idx), currentRow.get(idx));
+ return !valuesEqual(currentRow.get(idx), nextRow.get(idx));
+ }
+
+ /**
+ * Compares values the way {@link Objects#deepEquals} does, which compares binary values by
+ * content, except that arrays and structs are also traversed so that binary values nested within
+ * them are compared by content too.
+ */
+ private static boolean valuesEqual(Object left, Object right) {
+ if (left instanceof Seq<?> leftSeq && right instanceof Seq<?> rightSeq) {
+ return seqsEqual(leftSeq, rightSeq);
+ } else if (left instanceof Row leftRow && right instanceof Row rightRow) {
+ return rowsEqual(leftRow, rightRow);
+ }
+
+ return Objects.deepEquals(left, right);
+ }
+
+ private static boolean seqsEqual(Seq<?> left, Seq<?> right) {
+ int length = left.length();
+ if (length != right.length()) {
+ return false;
+ }
+
+ for (int index = 0; index < length; index++) {
+ if (!valuesEqual(left.apply(index), right.apply(index))) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ private static boolean rowsEqual(Row left, Row right) {
+ int size = left.size();
+ if (size != right.size()) {
+ return false;
+ }
+
+ for (int index = 0; index < size; index++) {
+ if (!valuesEqual(left.get(index), right.get(index))) {
+ return false;
+ }
+ }
+
+ return true;
}
protected static int[] generateIndicesToIdentifySameRow(
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestChangelogIterator.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestChangelogIterator.java
index bd9832f7d6..4bf639f778 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestChangelogIterator.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestChangelogIterator.java
@@ -25,18 +25,28 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Stream;
import org.apache.iceberg.ChangelogOperation;
import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+import org.apache.spark.sql.catalyst.CatalystTypeConverters;
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema;
+import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
public class TestChangelogIterator extends SparkTestHelperBase {
+
+ private static final DataType BINARY_ARRAY = DataTypes.createArrayType(DataTypes.BinaryType);
private static final String DELETE = ChangelogOperation.DELETE.name();
private static final String INSERT = ChangelogOperation.INSERT.name();
private static final String UPDATE_BEFORE = ChangelogOperation.UPDATE_BEFORE.name();
@@ -85,73 +95,6 @@ public class TestChangelogIterator extends SparkTestHelperBase {
}
}
- private void validate(Object[] permutation) {
- List<Row> rows = Lists.newArrayList();
- List<Object[]> expectedRows = Lists.newArrayList();
- for (int i = 0; i < permutation.length; i++) {
- rows.addAll(toOriginalRows((RowType) permutation[i], i));
- expectedRows.addAll(toExpectedRows((RowType) permutation[i], i));
- }
-
- Iterator<Row> iterator =
- ChangelogIterator.computeUpdates(rows.iterator(), SCHEMA, IDENTIFIER_FIELDS);
- List<Row> result = Lists.newArrayList(iterator);
- assertEquals("Rows should match", expectedRows, rowsToJava(result));
- }
-
- private List<Row> toOriginalRows(RowType rowType, int index) {
- switch (rowType) {
- case DELETED:
- return Lists.newArrayList(
- new GenericRowWithSchema(new Object[] {index, "b", "data", DELETE, 0, 0}, null));
- case INSERTED:
- return Lists.newArrayList(
- new GenericRowWithSchema(new Object[] {index, "c", "data", INSERT, 0, 0}, null));
- case CARRY_OVER:
- return Lists.newArrayList(
- new GenericRowWithSchema(new Object[] {index, "d", "data", DELETE, 0, 0}, null),
- new GenericRowWithSchema(new Object[] {index, "d", "data", INSERT, 0, 0}, null));
- case UPDATED:
- return Lists.newArrayList(
- new GenericRowWithSchema(new Object[] {index, "a", "data", DELETE, 0, 0}, null),
- new GenericRowWithSchema(new Object[] {index, "a", "new_data", INSERT, 0, 0}, null));
- default:
- throw new IllegalArgumentException("Unknown row type: " + rowType);
- }
- }
-
- private List<Object[]> toExpectedRows(RowType rowType, int order) {
- switch (rowType) {
- case DELETED:
- List<Object[]> rows = Lists.newArrayList();
- rows.add(new Object[] {order, "b", "data", DELETE, 0, 0});
- return rows;
- case INSERTED:
- List<Object[]> insertedRows = Lists.newArrayList();
- insertedRows.add(new Object[] {order, "c", "data", INSERT, 0, 0});
- return insertedRows;
- case CARRY_OVER:
- return Lists.newArrayList();
- case UPDATED:
- return Lists.newArrayList(
- new Object[] {order, "a", "data", UPDATE_BEFORE, 0, 0},
- new Object[] {order, "a", "new_data", UPDATE_AFTER, 0, 0});
- default:
- throw new IllegalArgumentException("Unknown row type: " + rowType);
- }
- }
-
- private void permute(List<RowType> arr, int start, List<Object[]> pm) {
- for (int i = start; i < arr.size(); i++) {
- Collections.swap(arr, i, start);
- permute(arr, start + 1, pm);
- Collections.swap(arr, start, i);
- }
- if (start == arr.size() - 1) {
- pm.add(arr.toArray());
- }
- }
-
@Test
public void testRowsWithNullValue() {
final List<Row> rowsWithNull =
@@ -307,19 +250,6 @@ public class TestChangelogIterator extends SparkTestHelperBase {
validateIterators(rowsWithDuplication, expectedRows);
}
- private void validateIterators(List<Row> rowsWithDuplication, List<Object[]> expectedRows) {
- Iterator<Row> iterator =
- ChangelogIterator.removeCarryovers(rowsWithDuplication.iterator(), SCHEMA);
- List<Row> result = Lists.newArrayList(iterator);
-
- assertEquals("Rows should match.", expectedRows, rowsToJava(result));
-
- iterator = ChangelogIterator.removeNetCarryovers(rowsWithDuplication.iterator(), SCHEMA);
- result = Lists.newArrayList(iterator);
-
- assertEquals("Rows should match.", expectedRows, rowsToJava(result));
- }
-
@Test
public void testRemoveNetCarryovers() {
List<Row> rowsWithDuplication =
@@ -356,4 +286,215 @@ public class TestChangelogIterator extends SparkTestHelperBase {
assertEquals("Rows should match.", expectedRows, rowsToJava(result));
}
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("values")
+ public void testRemoveNetChangesWithEqualValues(DataType type, IntFunction<Object> values) {
+ Row insert = row(type, values.apply(1), INSERT, 0);
+ Row delete = row(type, values.apply(1), DELETE, 1);
+ Row latest = row(type, values.apply(2), INSERT, 1);
+ Iterator<Row> result =
+ ChangelogIterator.removeNetCarryovers(
+ List.of(insert, delete, latest).iterator(), schema(type));
+ assertThat(result).toIterable().containsExactly(latest);
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("values")
+ public void testRemoveCarryoversWithEqualValues(DataType type, IntFunction<Object> values) {
+ assertRemoved(
+ type,
+ List.of(row(type, values.apply(1), DELETE, 0), row(type, values.apply(1), INSERT, 0)));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("values")
+ public void testRetainChangesWithDifferentValues(DataType type, IntFunction<Object> values) {
+ assertRetained(
+ type,
+ List.of(row(type, values.apply(1), DELETE, 0), row(type, values.apply(2), INSERT, 0)));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("values")
+ public void testDistinguishNullFromNonNull(DataType type, IntFunction<Object> values) {
+ assertRetained(
+ type, List.of(row(type, null, DELETE, 0), row(type, values.apply(1), INSERT, 0)));
+ assertRetained(
+ type, List.of(row(type, values.apply(1), DELETE, 0), row(type, null, INSERT, 0)));
+ assertRemoved(type, List.of(row(type, null, DELETE, 0), row(type, null, INSERT, 0)));
+ }
+
+ @Test
+ public void testRetainChangesWithDifferentArrayLengths() {
+ assertRetained(
+ BINARY_ARRAY,
+ List.of(
+ row(BINARY_ARRAY, List.of(new byte[] {1}), DELETE, 0),
+ row(BINARY_ARRAY, List.of(new byte[] {1}, new byte[] {1}), INSERT, 0)));
+ }
+
+ @Test
+ public void testRetainChangesWithDifferentArrayOrder() {
+ assertRetained(
+ BINARY_ARRAY,
+ List.of(
+ row(BINARY_ARRAY, List.of(new byte[] {1}, new byte[] {2}), DELETE, 0),
+ row(BINARY_ARRAY, List.of(new byte[] {2}, new byte[] {1}), INSERT, 0)));
+ }
+
+ @Test
+ public void testRetainChangesWithDifferentSignedZerosInArrays() {
+ DataType type = DataTypes.createArrayType(DataTypes.DoubleType);
+ assertRetained(
+ type, List.of(row(type, List.of(-0.0), DELETE, 0), row(type, List.of(0.0), INSERT, 0)));
+ }
+
+ @Test
+ public void testRetainChangesWithDifferentSignedZerosInStructs() {
+ StructType type =
+ new StructType().add("binary", DataTypes.BinaryType).add("number", DataTypes.DoubleType);
+ assertRetained(
+ type,
+ List.of(
+ row(type, RowFactory.create(new byte[] {1}, -0.0), DELETE, 0),
+ row(type, RowFactory.create(new byte[] {1}, 0.0), INSERT, 0)));
+ }
+
+ static Stream<Arguments> values() {
+ StructType struct =
+ new StructType().add("patch", DataTypes.BinaryType).add("number", DataTypes.IntegerType);
+ return Stream.of(
+ Arguments.of(
+ DataTypes.BinaryType, (IntFunction<Object>) value -> new byte[] {(byte) value}),
+ Arguments.of(
+ BINARY_ARRAY,
+ (IntFunction<Object>) value -> Arrays.asList(new byte[] {(byte) value}, null)),
+ Arguments.of(
+ DataTypes.createArrayType(BINARY_ARRAY),
+ (IntFunction<Object>) value -> List.of(List.of(new byte[] {(byte) value}))),
+ Arguments.of(
+ struct, (IntFunction<Object>) value -> RowFactory.create(new byte[] {(byte) value}, 7)),
+ Arguments.of(
+ DataTypes.createArrayType(struct),
+ (IntFunction<Object>)
+ value -> List.of(RowFactory.create(new byte[] {(byte) value}, 7))),
+ Arguments.of(
+ new StructType().add("patches", BINARY_ARRAY),
+ (IntFunction<Object>) value -> RowFactory.create(List.of(new byte[] {(byte) value}))),
+ Arguments.of(
+ DataTypes.createArrayType(DataTypes.IntegerType),
+ (IntFunction<Object>) value -> Arrays.asList(value, null)));
+ }
+
+ private void validate(Object[] permutation) {
+ List<Row> rows = Lists.newArrayList();
+ List<Object[]> expectedRows = Lists.newArrayList();
+ for (int i = 0; i < permutation.length; i++) {
+ rows.addAll(toOriginalRows((RowType) permutation[i], i));
+ expectedRows.addAll(toExpectedRows((RowType) permutation[i], i));
+ }
+
+ Iterator<Row> iterator =
+ ChangelogIterator.computeUpdates(rows.iterator(), SCHEMA, IDENTIFIER_FIELDS);
+ List<Row> result = Lists.newArrayList(iterator);
+ assertEquals("Rows should match", expectedRows, rowsToJava(result));
+ }
+
+ private List<Row> toOriginalRows(RowType rowType, int index) {
+ switch (rowType) {
+ case DELETED:
+ return Lists.newArrayList(
+ new GenericRowWithSchema(new Object[] {index, "b", "data", DELETE, 0, 0}, null));
+ case INSERTED:
+ return Lists.newArrayList(
+ new GenericRowWithSchema(new Object[] {index, "c", "data", INSERT, 0, 0}, null));
+ case CARRY_OVER:
+ return Lists.newArrayList(
+ new GenericRowWithSchema(new Object[] {index, "d", "data", DELETE, 0, 0}, null),
+ new GenericRowWithSchema(new Object[] {index, "d", "data", INSERT, 0, 0}, null));
+ case UPDATED:
+ return Lists.newArrayList(
+ new GenericRowWithSchema(new Object[] {index, "a", "data", DELETE, 0, 0}, null),
+ new GenericRowWithSchema(new Object[] {index, "a", "new_data", INSERT, 0, 0}, null));
+ default:
+ throw new IllegalArgumentException("Unknown row type: " + rowType);
+ }
+ }
+
+ private List<Object[]> toExpectedRows(RowType rowType, int order) {
+ switch (rowType) {
+ case DELETED:
+ List<Object[]> rows = Lists.newArrayList();
+ rows.add(new Object[] {order, "b", "data", DELETE, 0, 0});
+ return rows;
+ case INSERTED:
+ List<Object[]> insertedRows = Lists.newArrayList();
+ insertedRows.add(new Object[] {order, "c", "data", INSERT, 0, 0});
+ return insertedRows;
+ case CARRY_OVER:
+ return Lists.newArrayList();
+ case UPDATED:
+ return Lists.newArrayList(
+ new Object[] {order, "a", "data", UPDATE_BEFORE, 0, 0},
+ new Object[] {order, "a", "new_data", UPDATE_AFTER, 0, 0});
+ default:
+ throw new IllegalArgumentException("Unknown row type: " + rowType);
+ }
+ }
+
+ private void permute(List<RowType> arr, int start, List<Object[]> pm) {
+ for (int i = start; i < arr.size(); i++) {
+ Collections.swap(arr, i, start);
+ permute(arr, start + 1, pm);
+ Collections.swap(arr, start, i);
+ }
+ if (start == arr.size() - 1) {
+ pm.add(arr.toArray());
+ }
+ }
+
+ private void validateIterators(List<Row> rowsWithDuplication, List<Object[]> expectedRows) {
+ Iterator<Row> iterator =
+ ChangelogIterator.removeCarryovers(rowsWithDuplication.iterator(), SCHEMA);
+ List<Row> result = Lists.newArrayList(iterator);
+
+ assertEquals("Rows should match.", expectedRows, rowsToJava(result));
+
+ iterator = ChangelogIterator.removeNetCarryovers(rowsWithDuplication.iterator(), SCHEMA);
+ result = Lists.newArrayList(iterator);
+
+ assertEquals("Rows should match.", expectedRows, rowsToJava(result));
+ }
+
+ private static void assertRemoved(DataType type, List<Row> rows) {
+ assertThat(ChangelogIterator.removeCarryovers(rows.iterator(), schema(type))).isExhausted();
+ assertThat(ChangelogIterator.removeNetCarryovers(rows.iterator(), schema(type))).isExhausted();
+ }
+
+ private static void assertRetained(DataType type, List<Row> rows) {
+ assertThat(ChangelogIterator.removeCarryovers(rows.iterator(), schema(type)))
+ .toIterable()
+ .containsExactlyElementsOf(rows);
+ assertThat(ChangelogIterator.removeNetCarryovers(rows.iterator(), schema(type)))
+ .toIterable()
+ .containsExactlyElementsOf(rows);
+ }
+
+ private static StructType schema(DataType type) {
+ return new StructType()
+ .add("id", DataTypes.IntegerType)
+ .add("value", type)
+ .add(MetadataColumns.CHANGE_TYPE.name(), DataTypes.StringType)
+ .add(MetadataColumns.CHANGE_ORDINAL.name(), DataTypes.IntegerType)
+ .add(MetadataColumns.COMMIT_SNAPSHOT_ID.name(), DataTypes.LongType);
+ }
+
+ private static Row row(DataType type, Object value, String operation, int ordinal) {
+ // round-trip to get Spark's external representation (ArraySeq, GenericRowWithSchema); a plain
+ // java.util.List would not exercise the Seq branch of the comparison
+ Object internal = CatalystTypeConverters.createToCatalystConverter(type).apply(value);
+ Object external = CatalystTypeConverters.createToScalaConverter(type).apply(internal);
+ return RowFactory.create(1, external, operation, ordinal, (long) ordinal);
+ }
}
diff --git a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java b/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
index 3734d24e96..51be436921 100644
--- a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
+++ b/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
@@ -57,6 +57,22 @@ public class TestCreateChangelogViewProcedure extends ExtensionsTestBase {
sql("ALTER TABLE %s ADD PARTITION FIELD id", tableName);
}
+ /** The initial and updated values must encode 0x01 and 0x02 respectively. */
+ private void assertBinaryNetChanges(String initial, String updated, String projection) {
+ sql("INSERT INTO %s VALUES (1, %s), (2, %s)", tableName, initial, initial);
+ sql("INSERT OVERWRITE %s VALUES (1, %s), (2, %s)", tableName, updated, initial);
+
+ List<Object[]> result =
+ sql(
+ "CALL %s.system.create_changelog_view(table => '%s', net_changes => true)",
+ catalogName, tableName);
+ String viewName = (String) result.get(0)[0];
+ assertEquals(
+ "Net changes should contain only the final rows",
+ ImmutableList.of(row(1, "02", INSERT), row(2, "01", INSERT)),
+ sql("SELECT id, %s, _change_type FROM %s ORDER BY id", projection, viewName));
+ }
+
private void createTableWithIdentifierField() {
sql("CREATE TABLE %s (id INT NOT NULL, data STRING) USING iceberg", tableName);
sql("ALTER TABLE %s SET IDENTIFIER FIELDS id", tableName);
@@ -613,6 +629,18 @@ public class TestCreateChangelogViewProcedure extends ExtensionsTestBase {
sql("select * from %s order by _change_ordinal, id, data", viewName));
}
+ @TestTemplate
+ public void testNetChangesWithBinaryValues() {
+ sql("CREATE TABLE %s (id INT, data BINARY) USING iceberg", tableName);
+ assertBinaryNetChanges("X'01'", "X'02'", "hex(data)");
+ }
+
+ @TestTemplate
+ public void testNetChangesWithNestedBinaryValues() {
+ sql("CREATE TABLE %s (id INT, data ARRAY<ARRAY<BINARY>>) USING iceberg", tableName);
+ assertBinaryNetChanges("array(array(X'01'))", "array(array(X'02'))", "hex(data[0][0])");
+ }
+
@TestTemplate
public void testNetChangesWithRemoveCarryOvers() {
// partitioned by id
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java
index cc44b1f399..12ed77651a 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java
@@ -27,6 +27,7 @@ import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Iterators;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.StructType;
+import scala.collection.Seq;
/** An iterator that transforms rows from changelog tables within a single Spark task. */
public abstract class ChangelogIterator implements Iterator<Row> {
@@ -109,7 +110,52 @@ public abstract class ChangelogIterator implements Iterator<Row> {
}
protected boolean isDifferentValue(Row currentRow, Row nextRow, int idx) {
- return !Objects.equals(nextRow.get(idx), currentRow.get(idx));
+ return !valuesEqual(currentRow.get(idx), nextRow.get(idx));
+ }
+
+ /**
+ * Compares values the way {@link Objects#deepEquals} does, which compares binary values by
+ * content, except that arrays and structs are also traversed so that binary values nested within
+ * them are compared by content too.
+ */
+ private static boolean valuesEqual(Object left, Object right) {
+ if (left instanceof Seq<?> leftSeq && right instanceof Seq<?> rightSeq) {
+ return seqsEqual(leftSeq, rightSeq);
+ } else if (left instanceof Row leftRow && right instanceof Row rightRow) {
+ return rowsEqual(leftRow, rightRow);
+ }
+
+ return Objects.deepEquals(left, right);
+ }
+
+ private static boolean seqsEqual(Seq<?> left, Seq<?> right) {
+ int length = left.length();
+ if (length != right.length()) {
+ return false;
+ }
+
+ for (int index = 0; index < length; index++) {
+ if (!valuesEqual(left.apply(index), right.apply(index))) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ private static boolean rowsEqual(Row left, Row right) {
+ int size = left.size();
+ if (size != right.size()) {
+ return false;
+ }
+
+ for (int index = 0; index < size; index++) {
+ if (!valuesEqual(left.get(index), right.get(index))) {
+ return false;
+ }
+ }
+
+ return true;
}
protected static int[] generateIndicesToIdentifySameRow(
diff --git a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/TestChangelogIterator.java b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/TestChangelogIterator.java
index bd9832f7d6..4bf639f778 100644
--- a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/TestChangelogIterator.java
+++ b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/TestChangelogIterator.java
@@ -25,18 +25,28 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Stream;
import org.apache.iceberg.ChangelogOperation;
import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+import org.apache.spark.sql.catalyst.CatalystTypeConverters;
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema;
+import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
public class TestChangelogIterator extends SparkTestHelperBase {
+
+ private static final DataType BINARY_ARRAY = DataTypes.createArrayType(DataTypes.BinaryType);
private static final String DELETE = ChangelogOperation.DELETE.name();
private static final String INSERT = ChangelogOperation.INSERT.name();
private static final String UPDATE_BEFORE = ChangelogOperation.UPDATE_BEFORE.name();
@@ -85,73 +95,6 @@ public class TestChangelogIterator extends SparkTestHelperBase {
}
}
- private void validate(Object[] permutation) {
- List<Row> rows = Lists.newArrayList();
- List<Object[]> expectedRows = Lists.newArrayList();
- for (int i = 0; i < permutation.length; i++) {
- rows.addAll(toOriginalRows((RowType) permutation[i], i));
- expectedRows.addAll(toExpectedRows((RowType) permutation[i], i));
- }
-
- Iterator<Row> iterator =
- ChangelogIterator.computeUpdates(rows.iterator(), SCHEMA, IDENTIFIER_FIELDS);
- List<Row> result = Lists.newArrayList(iterator);
- assertEquals("Rows should match", expectedRows, rowsToJava(result));
- }
-
- private List<Row> toOriginalRows(RowType rowType, int index) {
- switch (rowType) {
- case DELETED:
- return Lists.newArrayList(
- new GenericRowWithSchema(new Object[] {index, "b", "data", DELETE, 0, 0}, null));
- case INSERTED:
- return Lists.newArrayList(
- new GenericRowWithSchema(new Object[] {index, "c", "data", INSERT, 0, 0}, null));
- case CARRY_OVER:
- return Lists.newArrayList(
- new GenericRowWithSchema(new Object[] {index, "d", "data", DELETE, 0, 0}, null),
- new GenericRowWithSchema(new Object[] {index, "d", "data", INSERT, 0, 0}, null));
- case UPDATED:
- return Lists.newArrayList(
- new GenericRowWithSchema(new Object[] {index, "a", "data", DELETE, 0, 0}, null),
- new GenericRowWithSchema(new Object[] {index, "a", "new_data", INSERT, 0, 0}, null));
- default:
- throw new IllegalArgumentException("Unknown row type: " + rowType);
- }
- }
-
- private List<Object[]> toExpectedRows(RowType rowType, int order) {
- switch (rowType) {
- case DELETED:
- List<Object[]> rows = Lists.newArrayList();
- rows.add(new Object[] {order, "b", "data", DELETE, 0, 0});
- return rows;
- case INSERTED:
- List<Object[]> insertedRows = Lists.newArrayList();
- insertedRows.add(new Object[] {order, "c", "data", INSERT, 0, 0});
- return insertedRows;
- case CARRY_OVER:
- return Lists.newArrayList();
- case UPDATED:
- return Lists.newArrayList(
- new Object[] {order, "a", "data", UPDATE_BEFORE, 0, 0},
- new Object[] {order, "a", "new_data", UPDATE_AFTER, 0, 0});
- default:
- throw new IllegalArgumentException("Unknown row type: " + rowType);
- }
- }
-
- private void permute(List<RowType> arr, int start, List<Object[]> pm) {
- for (int i = start; i < arr.size(); i++) {
- Collections.swap(arr, i, start);
- permute(arr, start + 1, pm);
- Collections.swap(arr, start, i);
- }
- if (start == arr.size() - 1) {
- pm.add(arr.toArray());
- }
- }
-
@Test
public void testRowsWithNullValue() {
final List<Row> rowsWithNull =
@@ -307,19 +250,6 @@ public class TestChangelogIterator extends SparkTestHelperBase {
validateIterators(rowsWithDuplication, expectedRows);
}
- private void validateIterators(List<Row> rowsWithDuplication, List<Object[]> expectedRows) {
- Iterator<Row> iterator =
- ChangelogIterator.removeCarryovers(rowsWithDuplication.iterator(), SCHEMA);
- List<Row> result = Lists.newArrayList(iterator);
-
- assertEquals("Rows should match.", expectedRows, rowsToJava(result));
-
- iterator = ChangelogIterator.removeNetCarryovers(rowsWithDuplication.iterator(), SCHEMA);
- result = Lists.newArrayList(iterator);
-
- assertEquals("Rows should match.", expectedRows, rowsToJava(result));
- }
-
@Test
public void testRemoveNetCarryovers() {
List<Row> rowsWithDuplication =
@@ -356,4 +286,215 @@ public class TestChangelogIterator extends SparkTestHelperBase {
assertEquals("Rows should match.", expectedRows, rowsToJava(result));
}
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("values")
+ public void testRemoveNetChangesWithEqualValues(DataType type, IntFunction<Object> values) {
+ Row insert = row(type, values.apply(1), INSERT, 0);
+ Row delete = row(type, values.apply(1), DELETE, 1);
+ Row latest = row(type, values.apply(2), INSERT, 1);
+ Iterator<Row> result =
+ ChangelogIterator.removeNetCarryovers(
+ List.of(insert, delete, latest).iterator(), schema(type));
+ assertThat(result).toIterable().containsExactly(latest);
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("values")
+ public void testRemoveCarryoversWithEqualValues(DataType type, IntFunction<Object> values) {
+ assertRemoved(
+ type,
+ List.of(row(type, values.apply(1), DELETE, 0), row(type, values.apply(1), INSERT, 0)));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("values")
+ public void testRetainChangesWithDifferentValues(DataType type, IntFunction<Object> values) {
+ assertRetained(
+ type,
+ List.of(row(type, values.apply(1), DELETE, 0), row(type, values.apply(2), INSERT, 0)));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("values")
+ public void testDistinguishNullFromNonNull(DataType type, IntFunction<Object> values) {
+ assertRetained(
+ type, List.of(row(type, null, DELETE, 0), row(type, values.apply(1), INSERT, 0)));
+ assertRetained(
+ type, List.of(row(type, values.apply(1), DELETE, 0), row(type, null, INSERT, 0)));
+ assertRemoved(type, List.of(row(type, null, DELETE, 0), row(type, null, INSERT, 0)));
+ }
+
+ @Test
+ public void testRetainChangesWithDifferentArrayLengths() {
+ assertRetained(
+ BINARY_ARRAY,
+ List.of(
+ row(BINARY_ARRAY, List.of(new byte[] {1}), DELETE, 0),
+ row(BINARY_ARRAY, List.of(new byte[] {1}, new byte[] {1}), INSERT, 0)));
+ }
+
+ @Test
+ public void testRetainChangesWithDifferentArrayOrder() {
+ assertRetained(
+ BINARY_ARRAY,
+ List.of(
+ row(BINARY_ARRAY, List.of(new byte[] {1}, new byte[] {2}), DELETE, 0),
+ row(BINARY_ARRAY, List.of(new byte[] {2}, new byte[] {1}), INSERT, 0)));
+ }
+
+ @Test
+ public void testRetainChangesWithDifferentSignedZerosInArrays() {
+ DataType type = DataTypes.createArrayType(DataTypes.DoubleType);
+ assertRetained(
+ type, List.of(row(type, List.of(-0.0), DELETE, 0), row(type, List.of(0.0), INSERT, 0)));
+ }
+
+ @Test
+ public void testRetainChangesWithDifferentSignedZerosInStructs() {
+ StructType type =
+ new StructType().add("binary", DataTypes.BinaryType).add("number", DataTypes.DoubleType);
+ assertRetained(
+ type,
+ List.of(
+ row(type, RowFactory.create(new byte[] {1}, -0.0), DELETE, 0),
+ row(type, RowFactory.create(new byte[] {1}, 0.0), INSERT, 0)));
+ }
+
+ static Stream<Arguments> values() {
+ StructType struct =
+ new StructType().add("patch", DataTypes.BinaryType).add("number", DataTypes.IntegerType);
+ return Stream.of(
+ Arguments.of(
+ DataTypes.BinaryType, (IntFunction<Object>) value -> new byte[] {(byte) value}),
+ Arguments.of(
+ BINARY_ARRAY,
+ (IntFunction<Object>) value -> Arrays.asList(new byte[] {(byte) value}, null)),
+ Arguments.of(
+ DataTypes.createArrayType(BINARY_ARRAY),
+ (IntFunction<Object>) value -> List.of(List.of(new byte[] {(byte) value}))),
+ Arguments.of(
+ struct, (IntFunction<Object>) value -> RowFactory.create(new byte[] {(byte) value}, 7)),
+ Arguments.of(
+ DataTypes.createArrayType(struct),
+ (IntFunction<Object>)
+ value -> List.of(RowFactory.create(new byte[] {(byte) value}, 7))),
+ Arguments.of(
+ new StructType().add("patches", BINARY_ARRAY),
+ (IntFunction<Object>) value -> RowFactory.create(List.of(new byte[] {(byte) value}))),
+ Arguments.of(
+ DataTypes.createArrayType(DataTypes.IntegerType),
+ (IntFunction<Object>) value -> Arrays.asList(value, null)));
+ }
+
+ private void validate(Object[] permutation) {
+ List<Row> rows = Lists.newArrayList();
+ List<Object[]> expectedRows = Lists.newArrayList();
+ for (int i = 0; i < permutation.length; i++) {
+ rows.addAll(toOriginalRows((RowType) permutation[i], i));
+ expectedRows.addAll(toExpectedRows((RowType) permutation[i], i));
+ }
+
+ Iterator<Row> iterator =
+ ChangelogIterator.computeUpdates(rows.iterator(), SCHEMA, IDENTIFIER_FIELDS);
+ List<Row> result = Lists.newArrayList(iterator);
+ assertEquals("Rows should match", expectedRows, rowsToJava(result));
+ }
+
+ private List<Row> toOriginalRows(RowType rowType, int index) {
+ switch (rowType) {
+ case DELETED:
+ return Lists.newArrayList(
+ new GenericRowWithSchema(new Object[] {index, "b", "data", DELETE, 0, 0}, null));
+ case INSERTED:
+ return Lists.newArrayList(
+ new GenericRowWithSchema(new Object[] {index, "c", "data", INSERT, 0, 0}, null));
+ case CARRY_OVER:
+ return Lists.newArrayList(
+ new GenericRowWithSchema(new Object[] {index, "d", "data", DELETE, 0, 0}, null),
+ new GenericRowWithSchema(new Object[] {index, "d", "data", INSERT, 0, 0}, null));
+ case UPDATED:
+ return Lists.newArrayList(
+ new GenericRowWithSchema(new Object[] {index, "a", "data", DELETE, 0, 0}, null),
+ new GenericRowWithSchema(new Object[] {index, "a", "new_data", INSERT, 0, 0}, null));
+ default:
+ throw new IllegalArgumentException("Unknown row type: " + rowType);
+ }
+ }
+
+ private List<Object[]> toExpectedRows(RowType rowType, int order) {
+ switch (rowType) {
+ case DELETED:
+ List<Object[]> rows = Lists.newArrayList();
+ rows.add(new Object[] {order, "b", "data", DELETE, 0, 0});
+ return rows;
+ case INSERTED:
+ List<Object[]> insertedRows = Lists.newArrayList();
+ insertedRows.add(new Object[] {order, "c", "data", INSERT, 0, 0});
+ return insertedRows;
+ case CARRY_OVER:
+ return Lists.newArrayList();
+ case UPDATED:
+ return Lists.newArrayList(
+ new Object[] {order, "a", "data", UPDATE_BEFORE, 0, 0},
+ new Object[] {order, "a", "new_data", UPDATE_AFTER, 0, 0});
+ default:
+ throw new IllegalArgumentException("Unknown row type: " + rowType);
+ }
+ }
+
+ private void permute(List<RowType> arr, int start, List<Object[]> pm) {
+ for (int i = start; i < arr.size(); i++) {
+ Collections.swap(arr, i, start);
+ permute(arr, start + 1, pm);
+ Collections.swap(arr, start, i);
+ }
+ if (start == arr.size() - 1) {
+ pm.add(arr.toArray());
+ }
+ }
+
+ private void validateIterators(List<Row> rowsWithDuplication, List<Object[]> expectedRows) {
+ Iterator<Row> iterator =
+ ChangelogIterator.removeCarryovers(rowsWithDuplication.iterator(), SCHEMA);
+ List<Row> result = Lists.newArrayList(iterator);
+
+ assertEquals("Rows should match.", expectedRows, rowsToJava(result));
+
+ iterator = ChangelogIterator.removeNetCarryovers(rowsWithDuplication.iterator(), SCHEMA);
+ result = Lists.newArrayList(iterator);
+
+ assertEquals("Rows should match.", expectedRows, rowsToJava(result));
+ }
+
+ private static void assertRemoved(DataType type, List<Row> rows) {
+ assertThat(ChangelogIterator.removeCarryovers(rows.iterator(), schema(type))).isExhausted();
+ assertThat(ChangelogIterator.removeNetCarryovers(rows.iterator(), schema(type))).isExhausted();
+ }
+
+ private static void assertRetained(DataType type, List<Row> rows) {
+ assertThat(ChangelogIterator.removeCarryovers(rows.iterator(), schema(type)))
+ .toIterable()
+ .containsExactlyElementsOf(rows);
+ assertThat(ChangelogIterator.removeNetCarryovers(rows.iterator(), schema(type)))
+ .toIterable()
+ .containsExactlyElementsOf(rows);
+ }
+
+ private static StructType schema(DataType type) {
+ return new StructType()
+ .add("id", DataTypes.IntegerType)
+ .add("value", type)
+ .add(MetadataColumns.CHANGE_TYPE.name(), DataTypes.StringType)
+ .add(MetadataColumns.CHANGE_ORDINAL.name(), DataTypes.IntegerType)
+ .add(MetadataColumns.COMMIT_SNAPSHOT_ID.name(), DataTypes.LongType);
+ }
+
+ private static Row row(DataType type, Object value, String operation, int ordinal) {
+ // round-trip to get Spark's external representation (ArraySeq, GenericRowWithSchema); a plain
+ // java.util.List would not exercise the Seq branch of the comparison
+ Object internal = CatalystTypeConverters.createToCatalystConverter(type).apply(value);
+ Object external = CatalystTypeConverters.createToScalaConverter(type).apply(internal);
+ return RowFactory.create(1, external, operation, ordinal, (long) ordinal);
+ }
}
diff --git a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
index 3734d24e96..51be436921 100644
--- a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
+++ b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java
@@ -57,6 +57,22 @@ public class TestCreateChangelogViewProcedure extends ExtensionsTestBase {
sql("ALTER TABLE %s ADD PARTITION FIELD id", tableName);
}
+ /** The initial and updated values must encode 0x01 and 0x02 respectively. */
+ private void assertBinaryNetChanges(String initial, String updated, String projection) {
+ sql("INSERT INTO %s VALUES (1, %s), (2, %s)", tableName, initial, initial);
+ sql("INSERT OVERWRITE %s VALUES (1, %s), (2, %s)", tableName, updated, initial);
+
+ List<Object[]> result =
+ sql(
+ "CALL %s.system.create_changelog_view(table => '%s', net_changes => true)",
+ catalogName, tableName);
+ String viewName = (String) result.get(0)[0];
+ assertEquals(
+ "Net changes should contain only the final rows",
+ ImmutableList.of(row(1, "02", INSERT), row(2, "01", INSERT)),
+ sql("SELECT id, %s, _change_type FROM %s ORDER BY id", projection, viewName));
+ }
+
private void createTableWithIdentifierField() {
sql("CREATE TABLE %s (id INT NOT NULL, data STRING) USING iceberg", tableName);
sql("ALTER TABLE %s SET IDENTIFIER FIELDS id", tableName);
@@ -613,6 +629,18 @@ public class TestCreateChangelogViewProcedure extends ExtensionsTestBase {
sql("select * from %s order by _change_ordinal, id, data", viewName));
}
+ @TestTemplate
+ public void testNetChangesWithBinaryValues() {
+ sql("CREATE TABLE %s (id INT, data BINARY) USING iceberg", tableName);
+ assertBinaryNetChanges("X'01'", "X'02'", "hex(data)");
+ }
+
+ @TestTemplate
+ public void testNetChangesWithNestedBinaryValues() {
+ sql("CREATE TABLE %s (id INT, data ARRAY<ARRAY<BINARY>>) USING iceberg", tableName);
+ assertBinaryNetChanges("array(array(X'01'))", "array(array(X'02'))", "hex(data[0][0])");
+ }
+
@TestTemplate
public void testNetChangesWithRemoveCarryOvers() {
// partitioned by id
diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java
index cc44b1f399..12ed77651a 100644
--- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java
+++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/ChangelogIterator.java
@@ -27,6 +27,7 @@ import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Iterators;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.StructType;
+import scala.collection.Seq;
/** An iterator that transforms rows from changelog tables within a single Spark task. */
public abstract class ChangelogIterator implements Iterator<Row> {
@@ -109,7 +110,52 @@ public abstract class ChangelogIterator implements Iterator<Row> {
}
protected boolean isDifferentValue(Row currentRow, Row nextRow, int idx) {
- return !Objects.equals(nextRow.get(idx), currentRow.get(idx));
+ return !valuesEqual(currentRow.get(idx), nextRow.get(idx));
+ }
+
+ /**
+ * Compares values the way {@link Objects#deepEquals} does, which compares binary values by
+ * content, except that arrays and structs are also traversed so that binary values nested within
+ * them are compared by content too.
+ */
+ private static boolean valuesEqual(Object left, Object right) {
+ if (left instanceof Seq<?> leftSeq && right instanceof Seq<?> rightSeq) {
+ return seqsEqual(leftSeq, rightSeq);
+ } else if (left instanceof Row leftRow && right instanceof Row rightRow) {
+ return rowsEqual(leftRow, rightRow);
+ }
+
+ return Objects.deepEquals(left, right);
+ }
+
+ private static boolean seqsEqual(Seq<?> left, Seq<?> right) {
+ int length = left.length();
+ if (length != right.length()) {
+ return false;
+ }
+
+ for (int index = 0; index < length; index++) {
+ if (!valuesEqual(left.apply(index), right.apply(index))) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ private static boolean rowsEqual(Row left, Row right) {
+ int size = left.size();
+ if (size != right.size()) {
+ return false;
+ }
+
+ for (int index = 0; index < size; index++) {
+ if (!valuesEqual(left.get(index), right.get(index))) {
+ return false;
+ }
+ }
+
+ return true;
}
protected static int[] generateIndicesToIdentifySameRow(
diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestChangelogIterator.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestChangelogIterator.java
index bd9832f7d6..4bf639f778 100644
--- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestChangelogIterator.java
+++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestChangelogIterator.java
@@ -25,18 +25,28 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
+import java.util.function.IntFunction;
+import java.util.stream.Stream;
import org.apache.iceberg.ChangelogOperation;
import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+import org.apache.spark.sql.catalyst.CatalystTypeConverters;
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema;
+import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
public class TestChangelogIterator extends SparkTestHelperBase {
+
+ private static final DataType BINARY_ARRAY = DataTypes.createArrayType(DataTypes.BinaryType);
private static final String DELETE = ChangelogOperation.DELETE.name();
private static final String INSERT = ChangelogOperation.INSERT.name();
private static final String UPDATE_BEFORE = ChangelogOperation.UPDATE_BEFORE.name();
@@ -85,73 +95,6 @@ public class TestChangelogIterator extends SparkTestHelperBase {
}
}
- private void validate(Object[] permutation) {
- List<Row> rows = Lists.newArrayList();
- List<Object[]> expectedRows = Lists.newArrayList();
- for (int i = 0; i < permutation.length; i++) {
- rows.addAll(toOriginalRows((RowType) permutation[i], i));
- expectedRows.addAll(toExpectedRows((RowType) permutation[i], i));
- }
-
- Iterator<Row> iterator =
- ChangelogIterator.computeUpdates(rows.iterator(), SCHEMA, IDENTIFIER_FIELDS);
- List<Row> result = Lists.newArrayList(iterator);
- assertEquals("Rows should match", expectedRows, rowsToJava(result));
- }
-
- private List<Row> toOriginalRows(RowType rowType, int index) {
- switch (rowType) {
- case DELETED:
- return Lists.newArrayList(
- new GenericRowWithSchema(new Object[] {index, "b", "data", DELETE, 0, 0}, null));
- case INSERTED:
- return Lists.newArrayList(
- new GenericRowWithSchema(new Object[] {index, "c", "data", INSERT, 0, 0}, null));
- case CARRY_OVER:
- return Lists.newArrayList(
- new GenericRowWithSchema(new Object[] {index, "d", "data", DELETE, 0, 0}, null),
- new GenericRowWithSchema(new Object[] {index, "d", "data", INSERT, 0, 0}, null));
- case UPDATED:
- return Lists.newArrayList(
- new GenericRowWithSchema(new Object[] {index, "a", "data", DELETE, 0, 0}, null),
- new GenericRowWithSchema(new Object[] {index, "a", "new_data", INSERT, 0, 0}, null));
- default:
- throw new IllegalArgumentException("Unknown row type: " + rowType);
- }
- }
-
- private List<Object[]> toExpectedRows(RowType rowType, int order) {
- switch (rowType) {
- case DELETED:
- List<Object[]> rows = Lists.newArrayList();
- rows.add(new Object[] {order, "b", "data", DELETE, 0, 0});
- return rows;
- case INSERTED:
- List<Object[]> insertedRows = Lists.newArrayList();
- insertedRows.add(new Object[] {order, "c", "data", INSERT, 0, 0});
- return insertedRows;
- case CARRY_OVER:
- return Lists.newArrayList();
- case UPDATED:
- return Lists.newArrayList(
- new Object[] {order, "a", "data", UPDATE_BEFORE, 0, 0},
- new Object[] {order, "a", "new_data", UPDATE_AFTER, 0, 0});
- default:
- throw new IllegalArgumentException("Unknown row type: " + rowType);
- }
- }
-
- private void permute(List<RowType> arr, int start, List<Object[]> pm) {
- for (int i = start; i < arr.size(); i++) {
- Collections.swap(arr, i, start);
- permute(arr, start + 1, pm);
- Collections.swap(arr, start, i);
- }
- if (start == arr.size() - 1) {
- pm.add(arr.toArray());
- }
- }
-
@Test
public void testRowsWithNullValue() {
final List<Row> rowsWithNull =
@@ -307,19 +250,6 @@ public class TestChangelogIterator extends SparkTestHelperBase {
validateIterators(rowsWithDuplication, expectedRows);
}
- private void validateIterators(List<Row> rowsWithDuplication, List<Object[]> expectedRows) {
- Iterator<Row> iterator =
- ChangelogIterator.removeCarryovers(rowsWithDuplication.iterator(), SCHEMA);
- List<Row> result = Lists.newArrayList(iterator);
-
- assertEquals("Rows should match.", expectedRows, rowsToJava(result));
-
- iterator = ChangelogIterator.removeNetCarryovers(rowsWithDuplication.iterator(), SCHEMA);
- result = Lists.newArrayList(iterator);
-
- assertEquals("Rows should match.", expectedRows, rowsToJava(result));
- }
-
@Test
public void testRemoveNetCarryovers() {
List<Row> rowsWithDuplication =
@@ -356,4 +286,215 @@ public class TestChangelogIterator extends SparkTestHelperBase {
assertEquals("Rows should match.", expectedRows, rowsToJava(result));
}
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("values")
+ public void testRemoveNetChangesWithEqualValues(DataType type, IntFunction<Object> values) {
+ Row insert = row(type, values.apply(1), INSERT, 0);
+ Row delete = row(type, values.apply(1), DELETE, 1);
+ Row latest = row(type, values.apply(2), INSERT, 1);
+ Iterator<Row> result =
+ ChangelogIterator.removeNetCarryovers(
+ List.of(insert, delete, latest).iterator(), schema(type));
+ assertThat(result).toIterable().containsExactly(latest);
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("values")
+ public void testRemoveCarryoversWithEqualValues(DataType type, IntFunction<Object> values) {
+ assertRemoved(
+ type,
+ List.of(row(type, values.apply(1), DELETE, 0), row(type, values.apply(1), INSERT, 0)));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("values")
+ public void testRetainChangesWithDifferentValues(DataType type, IntFunction<Object> values) {
+ assertRetained(
+ type,
+ List.of(row(type, values.apply(1), DELETE, 0), row(type, values.apply(2), INSERT, 0)));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("values")
+ public void testDistinguishNullFromNonNull(DataType type, IntFunction<Object> values) {
+ assertRetained(
+ type, List.of(row(type, null, DELETE, 0), row(type, values.apply(1), INSERT, 0)));
+ assertRetained(
+ type, List.of(row(type, values.apply(1), DELETE, 0), row(type, null, INSERT, 0)));
+ assertRemoved(type, List.of(row(type, null, DELETE, 0), row(type, null, INSERT, 0)));
+ }
+
+ @Test
+ public void testRetainChangesWithDifferentArrayLengths() {
+ assertRetained(
+ BINARY_ARRAY,
+ List.of(
+ row(BINARY_ARRAY, List.of(new byte[] {1}), DELETE, 0),
+ row(BINARY_ARRAY, List.of(new byte[] {1}, new byte[] {1}), INSERT, 0)));
+ }
+
+ @Test
+ public void testRetainChangesWithDifferentArrayOrder() {
+ assertRetained(
+ BINARY_ARRAY,
+ List.of(
+ row(BINARY_ARRAY, List.of(new byte[] {1}, new byte[] {2}), DELETE, 0),
+ row(BINARY_ARRAY, List.of(new byte[] {2}, new byte[] {1}), INSERT, 0)));
+ }
+
+ @Test
+ public void testRetainChangesWithDifferentSignedZerosInArrays() {
+ DataType type = DataTypes.createArrayType(DataTypes.DoubleType);
+ assertRetained(
+ type, List.of(row(type, List.of(-0.0), DELETE, 0), row(type, List.of(0.0), INSERT, 0)));
+ }
+
+ @Test
+ public void testRetainChangesWithDifferentSignedZerosInStructs() {
+ StructType type =
+ new StructType().add("binary", DataTypes.BinaryType).add("number", DataTypes.DoubleType);
+ assertRetained(
+ type,
+ List.of(
+ row(type, RowFactory.create(new byte[] {1}, -0.0), DELETE, 0),
+ row(type, RowFactory.create(new byte[] {1}, 0.0), INSERT, 0)));
+ }
+
+ static Stream<Arguments> values() {
+ StructType struct =
+ new StructType().add("patch", DataTypes.BinaryType).add("number", DataTypes.IntegerType);
+ return Stream.of(
+ Arguments.of(
+ DataTypes.BinaryType, (IntFunction<Object>) value -> new byte[] {(byte) value}),
+ Arguments.of(
+ BINARY_ARRAY,
+ (IntFunction<Object>) value -> Arrays.asList(new byte[] {(byte) value}, null)),
+ Arguments.of(
+ DataTypes.createArrayType(BINARY_ARRAY),
+ (IntFunction<Object>) value -> List.of(List.of(new byte[] {(byte) value}))),
+ Arguments.of(
+ struct, (IntFunction<Object>) value -> RowFactory.create(new byte[] {(byte) value}, 7)),
+ Arguments.of(
+ DataTypes.createArrayType(struct),
+ (IntFunction<Object>)
+ value -> List.of(RowFactory.create(new byte[] {(byte) value}, 7))),
+ Arguments.of(
+ new StructType().add("patches", BINARY_ARRAY),
+ (IntFunction<Object>) value -> RowFactory.create(List.of(new byte[] {(byte) value}))),
+ Arguments.of(
+ DataTypes.createArrayType(DataTypes.IntegerType),
+ (IntFunction<Object>) value -> Arrays.asList(value, null)));
+ }
+
+ private void validate(Object[] permutation) {
+ List<Row> rows = Lists.newArrayList();
+ List<Object[]> expectedRows = Lists.newArrayList();
+ for (int i = 0; i < permutation.length; i++) {
+ rows.addAll(toOriginalRows((RowType) permutation[i], i));
+ expectedRows.addAll(toExpectedRows((RowType) permutation[i], i));
+ }
+
+ Iterator<Row> iterator =
+ ChangelogIterator.computeUpdates(rows.iterator(), SCHEMA, IDENTIFIER_FIELDS);
+ List<Row> result = Lists.newArrayList(iterator);
+ assertEquals("Rows should match", expectedRows, rowsToJava(result));
+ }
+
+ private List<Row> toOriginalRows(RowType rowType, int index) {
+ switch (rowType) {
+ case DELETED:
+ return Lists.newArrayList(
+ new GenericRowWithSchema(new Object[] {index, "b", "data", DELETE, 0, 0}, null));
+ case INSERTED:
+ return Lists.newArrayList(
+ new GenericRowWithSchema(new Object[] {index, "c", "data", INSERT, 0, 0}, null));
+ case CARRY_OVER:
+ return Lists.newArrayList(
+ new GenericRowWithSchema(new Object[] {index, "d", "data", DELETE, 0, 0}, null),
+ new GenericRowWithSchema(new Object[] {index, "d", "data", INSERT, 0, 0}, null));
+ case UPDATED:
+ return Lists.newArrayList(
+ new GenericRowWithSchema(new Object[] {index, "a", "data", DELETE, 0, 0}, null),
+ new GenericRowWithSchema(new Object[] {index, "a", "new_data", INSERT, 0, 0}, null));
+ default:
+ throw new IllegalArgumentException("Unknown row type: " + rowType);
+ }
+ }
+
+ private List<Object[]> toExpectedRows(RowType rowType, int order) {
+ switch (rowType) {
+ case DELETED:
+ List<Object[]> rows = Lists.newArrayList();
+ rows.add(new Object[] {order, "b", "data", DELETE, 0, 0});
+ return rows;
+ case INSERTED:
+ List<Object[]> insertedRows = Lists.newArrayList();
+ insertedRows.add(new Object[] {order, "c", "data", INSERT, 0, 0});
+ return insertedRows;
+ case CARRY_OVER:
+ return Lists.newArrayList();
+ case UPDATED:
+ return Lists.newArrayList(
+ new Object[] {order, "a", "data", UPDATE_BEFORE, 0, 0},
+ new Object[] {order, "a", "new_data", UPDATE_AFTER, 0, 0});
+ default:
+ throw new IllegalArgumentException("Unknown row type: " + rowType);
+ }
+ }
+
+ private void permute(List<RowType> arr, int start, List<Object[]> pm) {
+ for (int i = start; i < arr.size(); i++) {
+ Collections.swap(arr, i, start);
+ permute(arr, start + 1, pm);
+ Collections.swap(arr, start, i);
+ }
+ if (start == arr.size() - 1) {
+ pm.add(arr.toArray());
+ }
+ }
+
+ private void validateIterators(List<Row> rowsWithDuplication, List<Object[]> expectedRows) {
+ Iterator<Row> iterator =
+ ChangelogIterator.removeCarryovers(rowsWithDuplication.iterator(), SCHEMA);
+ List<Row> result = Lists.newArrayList(iterator);
+
+ assertEquals("Rows should match.", expectedRows, rowsToJava(result));
+
+ iterator = ChangelogIterator.removeNetCarryovers(rowsWithDuplication.iterator(), SCHEMA);
+ result = Lists.newArrayList(iterator);
+
+ assertEquals("Rows should match.", expectedRows, rowsToJava(result));
+ }
+
+ private static void assertRemoved(DataType type, List<Row> rows) {
+ assertThat(ChangelogIterator.removeCarryovers(rows.iterator(), schema(type))).isExhausted();
+ assertThat(ChangelogIterator.removeNetCarryovers(rows.iterator(), schema(type))).isExhausted();
+ }
+
+ private static void assertRetained(DataType type, List<Row> rows) {
+ assertThat(ChangelogIterator.removeCarryovers(rows.iterator(), schema(type)))
+ .toIterable()
+ .containsExactlyElementsOf(rows);
+ assertThat(ChangelogIterator.removeNetCarryovers(rows.iterator(), schema(type)))
+ .toIterable()
+ .containsExactlyElementsOf(rows);
+ }
+
+ private static StructType schema(DataType type) {
+ return new StructType()
+ .add("id", DataTypes.IntegerType)
+ .add("value", type)
+ .add(MetadataColumns.CHANGE_TYPE.name(), DataTypes.StringType)
+ .add(MetadataColumns.CHANGE_ORDINAL.name(), DataTypes.IntegerType)
+ .add(MetadataColumns.COMMIT_SNAPSHOT_ID.name(), DataTypes.LongType);
+ }
+
+ private static Row row(DataType type, Object value, String operation, int ordinal) {
+ // round-trip to get Spark's external representation (ArraySeq, GenericRowWithSchema); a plain
+ // java.util.List would not exercise the Seq branch of the comparison
+ Object internal = CatalystTypeConverters.createToCatalystConverter(type).apply(value);
+ Object external = CatalystTypeConverters.createToScalaConverter(type).apply(internal);
+ return RowFactory.create(1, external, operation, ordinal, (long) ordinal);
+ }
}