mikedias commented on code in PR #780:
URL: https://github.com/apache/incubator-xtable/pull/780#discussion_r2659998390
##########
xtable-core/src/test/java/org/apache/xtable/paimon/TestPaimonConversionSource.java:
##########
@@ -165,38 +167,105 @@ void
testGetCurrentSnapshotThrowsExceptionWhenNoSnapshot() {
}
@Test
- void testGetTableChangeForCommitThrowsUnsupportedOperationException() {
+ void testGetCommitsBacklogReturnsCommitsAfterLastSync() {
+ // Insert initial data to create first snapshot
+ testTable.insertRows(5);
+ Snapshot firstSnapshot = paimonTable.snapshotManager().latestSnapshot();
+ assertNotNull(firstSnapshot);
+
+ // Insert more data to create second snapshot
testTable.insertRows(3);
- Snapshot snapshot = paimonTable.snapshotManager().latestSnapshot();
+ Snapshot secondSnapshot = paimonTable.snapshotManager().latestSnapshot();
+ assertNotNull(secondSnapshot);
+ assertNotEquals(firstSnapshot.id(), secondSnapshot.id());
- UnsupportedOperationException exception =
- assertThrows(
- UnsupportedOperationException.class,
- () -> conversionSource.getTableChangeForCommit(snapshot));
+ // Get commits backlog from first snapshot time
+ InstantsForIncrementalSync instantsForSync =
+ InstantsForIncrementalSync.builder()
+ .lastSyncInstant(Instant.ofEpochMilli(firstSnapshot.timeMillis()))
+ .build();
+
+ CommitsBacklog<Snapshot> backlog =
conversionSource.getCommitsBacklog(instantsForSync);
+
+ // Verify we get at least the second snapshot (may get more if insertRows
creates multiple)
+ assertNotNull(backlog);
+ assertTrue(backlog.getCommitsToProcess().size() >= 1);
Review Comment:
done!
##########
xtable-core/src/test/java/org/apache/xtable/paimon/TestPaimonConversionSource.java:
##########
@@ -165,38 +167,105 @@ void
testGetCurrentSnapshotThrowsExceptionWhenNoSnapshot() {
}
@Test
- void testGetTableChangeForCommitThrowsUnsupportedOperationException() {
+ void testGetCommitsBacklogReturnsCommitsAfterLastSync() {
+ // Insert initial data to create first snapshot
+ testTable.insertRows(5);
+ Snapshot firstSnapshot = paimonTable.snapshotManager().latestSnapshot();
+ assertNotNull(firstSnapshot);
+
+ // Insert more data to create second snapshot
testTable.insertRows(3);
- Snapshot snapshot = paimonTable.snapshotManager().latestSnapshot();
+ Snapshot secondSnapshot = paimonTable.snapshotManager().latestSnapshot();
+ assertNotNull(secondSnapshot);
+ assertNotEquals(firstSnapshot.id(), secondSnapshot.id());
- UnsupportedOperationException exception =
- assertThrows(
- UnsupportedOperationException.class,
- () -> conversionSource.getTableChangeForCommit(snapshot));
+ // Get commits backlog from first snapshot time
+ InstantsForIncrementalSync instantsForSync =
+ InstantsForIncrementalSync.builder()
+ .lastSyncInstant(Instant.ofEpochMilli(firstSnapshot.timeMillis()))
+ .build();
+
+ CommitsBacklog<Snapshot> backlog =
conversionSource.getCommitsBacklog(instantsForSync);
+
+ // Verify we get at least the second snapshot (may get more if insertRows
creates multiple)
+ assertNotNull(backlog);
+ assertTrue(backlog.getCommitsToProcess().size() >= 1);
+ // Verify the last snapshot in the backlog is the second snapshot
+ assertEquals(
+ secondSnapshot.id(),
+ backlog.getCommitsToProcess().get(backlog.getCommitsToProcess().size()
- 1).id());
+ assertTrue(backlog.getInFlightInstants().isEmpty());
+ }
- assertEquals("Incremental Sync is not supported yet.",
exception.getMessage());
+ @Test
+ void testGetCommitsBacklogReturnsEmptyForFutureInstant() {
+ testTable.insertRows(5);
+
+ // Use a future instant
+ InstantsForIncrementalSync instantsForSync =
+ InstantsForIncrementalSync.builder()
+ .lastSyncInstant(Instant.now().plusSeconds(3600))
+ .build();
+
+ CommitsBacklog<Snapshot> backlog =
conversionSource.getCommitsBacklog(instantsForSync);
+
+ // Verify no snapshots are returned
+ assertNotNull(backlog);
+ assertTrue(backlog.getCommitsToProcess().isEmpty());
}
@Test
- void testGetCommitsBacklogThrowsUnsupportedOperationException() {
- InstantsForIncrementalSync mockInstants =
-
InstantsForIncrementalSync.builder().lastSyncInstant(Instant.now()).build();
+ void testGetTableChangeForCommitReturnsCorrectFilesDiff() {
+ // Insert initial data
+ testTable.insertRows(5);
+ Snapshot firstSnapshot = paimonTable.snapshotManager().latestSnapshot();
+ assertNotNull(firstSnapshot);
- UnsupportedOperationException exception =
- assertThrows(
- UnsupportedOperationException.class,
- () -> conversionSource.getCommitsBacklog(mockInstants));
+ // Insert more data to create second snapshot
+ testTable.insertRows(3);
+ Snapshot secondSnapshot = paimonTable.snapshotManager().latestSnapshot();
+ assertNotNull(secondSnapshot);
+
+ // Get table change for second snapshot
+ TableChange tableChange =
conversionSource.getTableChangeForCommit(secondSnapshot);
+
+ // Verify table change structure
+ assertNotNull(tableChange);
+ assertNotNull(tableChange.getFilesDiff());
+ assertNotNull(tableChange.getTableAsOfChange());
+ assertEquals(
+ Long.toString(secondSnapshot.commitIdentifier()),
tableChange.getSourceIdentifier());
- assertEquals("Incremental Sync is not supported yet.",
exception.getMessage());
+ // For append-only table, we should have added files and no removed files
+ assertTrue(tableChange.getFilesDiff().getFilesAdded().size() > 0);
}
@Test
- void testIsIncrementalSyncSafeFromReturnsFalse() {
- Instant testInstant = Instant.now();
+ void testIsIncrementalSyncSafeFromReturnsTrueForValidInstant() {
Review Comment:
done!
--
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]
