Skip to content

Commit 59493dc

Browse files
cbb330cbb330
committed
Add e2e tests for Iceberg create_changelog_view procedure (Spark 3.5)
Tests the create_changelog_view stored procedure for CDC/incremental change tracking, available in Iceberg 1.5 (Spark 3.5 only): - Appends: verifies INSERT change types between snapshots - Overwrite: captures both DELETE (old rows) and INSERT (new rows) - Delete: verifies DELETE change type for removed rows (v2 format) - Net changes: collapses intermediate changes with identifier columns - Multi-snapshot span: changelog across multiple append snapshots
1 parent b813bc7 commit 59493dc

1 file changed

Lines changed: 254 additions & 0 deletions

File tree

  • integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/e2e

‎integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/e2e/ChangelogViewTest.java‎

Lines changed: 254 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,254 @@
1+
package com.linkedin.openhouse.spark.catalogtest.e2e;
2+
3+
import static org.junit.jupiter.api.Assertions.*;
4+
5+
import com.linkedin.openhouse.tablestest.OpenHouseSparkITest;
6+
import java.util.List;
7+
import java.util.Set;
8+
import java.util.stream.Collectors;
9+
import org.apache.spark.sql.Row;
10+
import org.apache.spark.sql.SparkSession;
11+
import org.junit.jupiter.api.Test;
12+
13+
public class ChangelogViewTest extends OpenHouseSparkITest {
14+
15+
private static final String DATABASE = "d1_changelog";
16+
17+
@Test
18+
void testChangelogViewForAppends() throws Exception {
19+
try (SparkSession spark = getSparkSession()) {
20+
String name = DATABASE + ".changelog_appends";
21+
spark.sql("DROP TABLE IF EXISTS openhouse." + name);
22+
spark.sql("CREATE TABLE openhouse." + name + " (id int, data string)");
23+
24+
spark.sql("INSERT INTO openhouse." + name + " VALUES (1, 'a'), (2, 'b')");
25+
spark.sql("INSERT INTO openhouse." + name + " VALUES (3, 'c')");
26+
27+
// Get snapshot IDs
28+
List<Row> snapshots =
29+
spark
30+
.sql("SELECT snapshot_id FROM openhouse." + name + ".snapshots ORDER BY committed_at")
31+
.collectAsList();
32+
assertEquals(2, snapshots.size());
33+
long snap1 = snapshots.get(0).getLong(0);
34+
long snap2 = snapshots.get(1).getLong(0);
35+
36+
// Create changelog view between the two snapshots
37+
spark.sql(
38+
String.format(
39+
"CALL openhouse.system.create_changelog_view("
40+
+ "table => '%s', "
41+
+ "options => map('start-snapshot-id', '%d', 'end-snapshot-id', '%d'))",
42+
name, snap1, snap2));
43+
44+
// The default view name is <table>_changes
45+
List<Row> changes = spark.sql("SELECT * FROM changelog_appends_changes").collectAsList();
46+
assertEquals(1, changes.size(), "Should have 1 change (the appended row)");
47+
48+
// Verify change type and data
49+
Row change = changes.get(0);
50+
assertEquals("INSERT", change.getAs("_change_type"));
51+
assertEquals(3, (int) change.getAs("id"));
52+
assertEquals("c", change.getAs("data"));
53+
54+
spark.sql("DROP TABLE openhouse." + name);
55+
}
56+
}
57+
58+
@Test
59+
void testChangelogViewForOverwrite() throws Exception {
60+
try (SparkSession spark = getSparkSession()) {
61+
String name = DATABASE + ".changelog_overwrite";
62+
spark.sql("DROP TABLE IF EXISTS openhouse." + name);
63+
spark.sql("CREATE TABLE openhouse." + name + " (id int, data string)");
64+
65+
spark.sql("INSERT INTO openhouse." + name + " VALUES (1, 'a'), (2, 'b')");
66+
spark.sql("INSERT OVERWRITE openhouse." + name + " VALUES (3, 'c')");
67+
68+
List<Row> snapshots =
69+
spark
70+
.sql("SELECT snapshot_id FROM openhouse." + name + ".snapshots ORDER BY committed_at")
71+
.collectAsList();
72+
assertEquals(2, snapshots.size());
73+
long snap1 = snapshots.get(0).getLong(0);
74+
long snap2 = snapshots.get(1).getLong(0);
75+
76+
spark.sql(
77+
String.format(
78+
"CALL openhouse.system.create_changelog_view("
79+
+ "table => '%s', "
80+
+ "options => map('start-snapshot-id', '%d', 'end-snapshot-id', '%d'))",
81+
name, snap1, snap2));
82+
83+
List<Row> changes = spark.sql("SELECT * FROM changelog_overwrite_changes").collectAsList();
84+
85+
// Overwrite should produce DELETEs for old rows and INSERTs for new rows
86+
Set<String> changeTypes =
87+
changes.stream().map(r -> r.getAs("_change_type").toString()).collect(Collectors.toSet());
88+
assertTrue(changeTypes.contains("DELETE"), "Should have DELETE changes for overwritten rows");
89+
assertTrue(changeTypes.contains("INSERT"), "Should have INSERT changes for new rows");
90+
91+
// Verify the deletes are for the original rows
92+
List<Row> deletes =
93+
changes.stream()
94+
.filter(r -> "DELETE".equals(r.getAs("_change_type")))
95+
.collect(Collectors.toList());
96+
assertEquals(2, deletes.size(), "Should delete both original rows");
97+
Set<Integer> deletedIds =
98+
deletes.stream().map(r -> (int) r.getAs("id")).collect(Collectors.toSet());
99+
assertTrue(deletedIds.containsAll(Set.of(1, 2)));
100+
101+
// Verify the insert is the new row
102+
List<Row> inserts =
103+
changes.stream()
104+
.filter(r -> "INSERT".equals(r.getAs("_change_type")))
105+
.collect(Collectors.toList());
106+
assertEquals(1, inserts.size(), "Should insert one new row");
107+
assertEquals(3, (int) inserts.get(0).getAs("id"));
108+
assertEquals("c", inserts.get(0).getAs("data"));
109+
110+
spark.sql("DROP TABLE openhouse." + name);
111+
}
112+
}
113+
114+
@Test
115+
void testChangelogViewForDelete() throws Exception {
116+
try (SparkSession spark = getSparkSession()) {
117+
String name = DATABASE + ".changelog_delete";
118+
spark.sql("DROP TABLE IF EXISTS openhouse." + name);
119+
spark.sql(
120+
"CREATE TABLE openhouse."
121+
+ name
122+
+ " (id int, data string) TBLPROPERTIES ('format-version'='2')");
123+
124+
spark.sql("INSERT INTO openhouse." + name + " VALUES (1, 'a'), (2, 'b'), (3, 'c')");
125+
spark.sql("DELETE FROM openhouse." + name + " WHERE id = 2");
126+
127+
List<Row> snapshots =
128+
spark
129+
.sql("SELECT snapshot_id FROM openhouse." + name + ".snapshots ORDER BY committed_at")
130+
.collectAsList();
131+
assertEquals(2, snapshots.size());
132+
long snap1 = snapshots.get(0).getLong(0);
133+
long snap2 = snapshots.get(1).getLong(0);
134+
135+
spark.sql(
136+
String.format(
137+
"CALL openhouse.system.create_changelog_view("
138+
+ "table => '%s', "
139+
+ "options => map('start-snapshot-id', '%d', 'end-snapshot-id', '%d'))",
140+
name, snap1, snap2));
141+
142+
List<Row> changes = spark.sql("SELECT * FROM changelog_delete_changes").collectAsList();
143+
144+
// DELETE should produce a DELETE change for the removed row
145+
assertEquals(1, changes.size(), "Should have 1 change for the deleted row");
146+
Row change = changes.get(0);
147+
assertEquals("DELETE", change.getAs("_change_type"));
148+
assertEquals(2, (int) change.getAs("id"));
149+
assertEquals("b", change.getAs("data"));
150+
151+
spark.sql("DROP TABLE openhouse." + name);
152+
}
153+
}
154+
155+
@Test
156+
void testChangelogViewWithNetChanges() throws Exception {
157+
try (SparkSession spark = getSparkSession()) {
158+
String name = DATABASE + ".changelog_net";
159+
spark.sql("DROP TABLE IF EXISTS openhouse." + name);
160+
spark.sql(
161+
"CREATE TABLE openhouse."
162+
+ name
163+
+ " (id int, data string) TBLPROPERTIES ('format-version'='2')");
164+
165+
// Insert, then delete, then re-insert the same id across multiple snapshots
166+
spark.sql("INSERT INTO openhouse." + name + " VALUES (1, 'a'), (2, 'b')");
167+
spark.sql("DELETE FROM openhouse." + name + " WHERE id = 1");
168+
spark.sql("INSERT INTO openhouse." + name + " VALUES (1, 'a_updated'), (3, 'c')");
169+
170+
List<Row> snapshots =
171+
spark
172+
.sql("SELECT snapshot_id FROM openhouse." + name + ".snapshots ORDER BY committed_at")
173+
.collectAsList();
174+
assertEquals(3, snapshots.size());
175+
long snap1 = snapshots.get(0).getLong(0);
176+
long snap3 = snapshots.get(2).getLong(0);
177+
178+
// Use net_changes to collapse intermediate changes.
179+
// compute_updates defaults to true when identifier_columns is set, but
180+
// net_changes and compute_updates cannot both be true, so disable compute_updates.
181+
spark.sql(
182+
String.format(
183+
"CALL openhouse.system.create_changelog_view("
184+
+ "table => '%s', "
185+
+ "options => map('start-snapshot-id', '%d', 'end-snapshot-id', '%d'), "
186+
+ "net_changes => true, "
187+
+ "compute_updates => false, "
188+
+ "identifier_columns => array('id'))",
189+
name, snap1, snap3));
190+
191+
List<Row> changes =
192+
spark
193+
.sql("SELECT * FROM changelog_net_changes ORDER BY _change_type, id")
194+
.collectAsList();
195+
196+
// Net changes: id=1 was deleted then re-inserted → net UPDATE
197+
// id=2 unchanged → no change
198+
// id=3 inserted → net INSERT
199+
Set<String> changeTypes =
200+
changes.stream().map(r -> r.getAs("_change_type").toString()).collect(Collectors.toSet());
201+
202+
// id=3 should appear as INSERT
203+
List<Row> inserts =
204+
changes.stream()
205+
.filter(r -> "INSERT".equals(r.getAs("_change_type")))
206+
.collect(Collectors.toList());
207+
assertTrue(
208+
inserts.stream().anyMatch(r -> (int) r.getAs("id") == 3), "id=3 should be a net INSERT");
209+
210+
spark.sql("DROP TABLE openhouse." + name);
211+
}
212+
}
213+
214+
@Test
215+
void testChangelogViewMultipleSnapshotsSpan() throws Exception {
216+
try (SparkSession spark = getSparkSession()) {
217+
String name = DATABASE + ".changelog_multi";
218+
spark.sql("DROP TABLE IF EXISTS openhouse." + name);
219+
spark.sql("CREATE TABLE openhouse." + name + " (id int, data string)");
220+
221+
spark.sql("INSERT INTO openhouse." + name + " VALUES (1, 'a')");
222+
spark.sql("INSERT INTO openhouse." + name + " VALUES (2, 'b')");
223+
spark.sql("INSERT INTO openhouse." + name + " VALUES (3, 'c')");
224+
225+
List<Row> snapshots =
226+
spark
227+
.sql("SELECT snapshot_id FROM openhouse." + name + ".snapshots ORDER BY committed_at")
228+
.collectAsList();
229+
assertEquals(3, snapshots.size());
230+
long snap1 = snapshots.get(0).getLong(0);
231+
long snap3 = snapshots.get(2).getLong(0);
232+
233+
spark.sql(
234+
String.format(
235+
"CALL openhouse.system.create_changelog_view("
236+
+ "table => '%s', "
237+
+ "options => map('start-snapshot-id', '%d', 'end-snapshot-id', '%d'))",
238+
name, snap1, snap3));
239+
240+
List<Row> changes =
241+
spark.sql("SELECT * FROM changelog_multi_changes ORDER BY id").collectAsList();
242+
243+
// Should see INSERTs for rows added in snapshots 2 and 3
244+
assertEquals(2, changes.size(), "Should see 2 inserts spanning snapshots 2 and 3");
245+
assertTrue(changes.stream().allMatch(r -> "INSERT".equals(r.getAs("_change_type"))));
246+
247+
Set<Integer> insertedIds =
248+
changes.stream().map(r -> (int) r.getAs("id")).collect(Collectors.toSet());
249+
assertEquals(Set.of(2, 3), insertedIds);
250+
251+
spark.sql("DROP TABLE openhouse." + name);
252+
}
253+
}
254+
}

0 commit comments

Comments
 (0)