You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@xtable.apache.org by "mikedias (via GitHub)" <gi...@apache.org> on 2025/12/30 03:11:39 UTC
[PR] Implement Paimon Source Incremental Sync [incubator-xtable]
mikedias opened a new pull request, #780:
URL: https://github.com/apache/incubator-xtable/pull/780
Implements https://github.com/apache/incubator-xtable/issues/754
## What is the purpose of the pull request
Implement Paimon Source Incremental Sync
## Brief change log
• Implemented getTableChangeForCommit() to extract file changes (added/removed) from delta manifests for
incremental sync
• Implemented getCommitsBacklog() to identify snapshots that need to be processed since the last sync instant
• Implemented isIncrementalSyncSafeFrom() to validate if incremental sync is safe from a given instant by checking
snapshot availability
## Verify this pull request
• Added tests in TestPaimonDataFileExtractor to cover the `extractFilesDiff()` logic.
• Added tests in TestPaimonConversionSource to cover the incremental sync methods.
• Existing integration tests in `ITConversionController` verify end-to-end incremental sync behavior
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement Paimon Source Incremental Sync [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown merged PR #780:
URL: https://github.com/apache/incubator-xtable/pull/780
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement Paimon Source Incremental Sync [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement Paimon Source Incremental Sync [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #780:
URL: https://github.com/apache/incubator-xtable/pull/780#discussion_r2659996404
##########
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)
Review Comment:
Paimon can trigger compaction at different times depending on the settings. It wouldn't be the case for this particular test, but it would make it flaky if we change things and don't consider this scenario on the test.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement Paimon Source Incremental Sync [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #780:
URL: https://github.com/apache/incubator-xtable/pull/780#discussion_r2656712406
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonConversionSource.java:
##########
@@ -114,18 +116,125 @@ private Snapshot getLastSnapshot() {
@Override
public TableChange getTableChangeForCommit(Snapshot snapshot) {
- throw new UnsupportedOperationException("Incremental Sync is not supported yet.");
+ InternalTable tableAtSnapshot = getTable(snapshot);
+ InternalSchema internalSchema = tableAtSnapshot.getReadSchema();
+
+ InternalFilesDiff filesDiff =
+ dataFileExtractor.extractFilesDiff(paimonTable, snapshot, internalSchema);
+
+ return TableChange.builder()
+ .tableAsOfChange(tableAtSnapshot)
+ .filesDiff(filesDiff)
+ .sourceIdentifier(getCommitIdentifier(snapshot))
+ .build();
}
@Override
public CommitsBacklog<Snapshot> getCommitsBacklog(
InstantsForIncrementalSync instantsForIncrementalSync) {
- throw new UnsupportedOperationException("Incremental Sync is not supported yet.");
+ Instant lastSyncInstant = instantsForIncrementalSync.getLastSyncInstant();
+ long lastSyncTimeMillis = lastSyncInstant.toEpochMilli();
+
+ log.info(
+ "Getting commits backlog for Paimon table {} from instant {}",
+ paimonTable.name(),
+ lastSyncInstant);
+
+ Iterator<Snapshot> snapshotIterator;
+ try {
+ snapshotIterator = snapshotManager.snapshots();
+ } catch (IOException e) {
+ throw new ReadException("Could not iterate over the Paimon snapshot list", e);
+ }
+
+ List<Snapshot> snapshotsToProcess = new ArrayList<>();
+ while (snapshotIterator.hasNext()) {
+ Snapshot snapshot = snapshotIterator.next();
+ // Only include snapshots committed after the last sync
+ if (snapshot.timeMillis() > lastSyncTimeMillis) {
+ snapshotsToProcess.add(snapshot);
+ log.debug(
+ "Including snapshot {} (time={}, commitId={}) in backlog",
+ snapshot.id(),
+ snapshot.timeMillis(),
+ snapshot.commitIdentifier());
+ }
+ }
+
+ log.info("Found {} snapshots to process for incremental sync", snapshotsToProcess.size());
+
+ return CommitsBacklog.<Snapshot>builder()
+ .commitsToProcess(snapshotsToProcess)
+ .inFlightInstants(Collections.emptyList())
+ .build();
}
@Override
public boolean isIncrementalSyncSafeFrom(Instant instant) {
- return false; // Incremental sync is not supported yet
+ long timeInMillis = instant.toEpochMilli();
+
+ Long earliestSnapshotId = snapshotManager.earliestSnapshotId();
+ Long latestSnapshotId = snapshotManager.latestSnapshotId();
+ if (earliestSnapshotId == null || latestSnapshotId == null) {
+ log.warn("No snapshots found in table {}", paimonTable.name());
+ return false;
+ }
+
+ Snapshot earliestSnapshot = snapshotManager.snapshot(earliestSnapshotId);
+ Snapshot latestSnapshot = snapshotManager.snapshot(latestSnapshotId);
+
+ // Check 1: If instant is in the future (after latest snapshot), return false
+ if (timeInMillis > latestSnapshot.timeMillis()) {
+ log.warn(
+ "Instant {} is in the future. Latest snapshot {} has time {}",
+ instant,
+ latestSnapshot.id(),
+ latestSnapshot.timeMillis());
+ return false;
+ }
+
+ // Check 2: Has snapshot expiration affected this instant?
+ // If the earliest snapshot is newer than the requested instant,
+ // then snapshots have been expired and we can't do incremental sync
+ if (earliestSnapshot.timeMillis() > timeInMillis) {
+ log.warn(
+ "Incremental sync is not safe from instant {}. "
+ + "Earliest available snapshot {} (time={}) is newer than the requested instant. "
+ + "Snapshots may have been expired.",
+ instant,
+ earliestSnapshot.id(),
+ earliestSnapshot.timeMillis());
+ return false;
+ }
+
+ // Check 3: Verify a snapshot exists at or before the instant
Review Comment:
Can this check simply check that the `earliestSnapshot.timeMillis <= timeInMillis`?
##########
xtable-core/src/test/java/org/apache/xtable/paimon/TestPaimonDataFileExtractor.java:
##########
@@ -147,6 +147,99 @@ void testColumnStatsAreEmpty() {
}
}
+ @Test
+ void testExtractFilesDiffWithNewFiles() {
+ createUnpartitionedTable();
+
+ // Insert initial data
+ testTable.insertRows(5);
+ org.apache.paimon.Snapshot firstSnapshot = paimonTable.snapshotManager().latestSnapshot();
+ assertNotNull(firstSnapshot);
+
+ // Insert more data to create a second snapshot
+ testTable.insertRows(3);
+ org.apache.paimon.Snapshot secondSnapshot = paimonTable.snapshotManager().latestSnapshot();
+ assertNotNull(secondSnapshot);
+
+ org.apache.xtable.model.storage.InternalFilesDiff filesDiff =
+ extractor.extractFilesDiff(paimonTable, secondSnapshot, testSchema);
+
+ // Verify we have added files
+ assertNotNull(filesDiff);
+ assertNotNull(filesDiff.getFilesAdded());
+ assertTrue(filesDiff.getFilesAdded().size() > 0);
+
+ // Verify removed files collection exists (size may vary based on compaction behavior)
+ assertNotNull(filesDiff.getFilesRemoved());
Review Comment:
Is there any way to configure the test so that the sizes are predictable and we can assert on them?
Also should we assert the files removed is non-zero?
##########
xtable-core/src/test/java/org/apache/xtable/paimon/TestPaimonDataFileExtractor.java:
##########
@@ -147,6 +147,99 @@ void testColumnStatsAreEmpty() {
}
}
+ @Test
+ void testExtractFilesDiffWithNewFiles() {
+ createUnpartitionedTable();
+
+ // Insert initial data
+ testTable.insertRows(5);
+ org.apache.paimon.Snapshot firstSnapshot = paimonTable.snapshotManager().latestSnapshot();
+ assertNotNull(firstSnapshot);
+
+ // Insert more data to create a second snapshot
+ testTable.insertRows(3);
+ org.apache.paimon.Snapshot secondSnapshot = paimonTable.snapshotManager().latestSnapshot();
Review Comment:
Can we import `org.apache.paimon.Snapshot` here?
##########
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:
If we cannot know the size for certain upfront, we may want to assert that the first snapshot is not in the list of commits to process
##########
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)
Review Comment:
I'm not familiar with Paimon, what would cause a single round of inserts to create multiple snapshots?
##########
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:
Can we add a test where the `IsIncrementalSyncSafeFrom` returns false since the instant is before the first snapshot?
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement Paimon Source Incremental Sync [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #780:
URL: https://github.com/apache/incubator-xtable/pull/780#discussion_r2659997725
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonConversionSource.java:
##########
@@ -114,18 +116,125 @@ private Snapshot getLastSnapshot() {
@Override
public TableChange getTableChangeForCommit(Snapshot snapshot) {
- throw new UnsupportedOperationException("Incremental Sync is not supported yet.");
+ InternalTable tableAtSnapshot = getTable(snapshot);
+ InternalSchema internalSchema = tableAtSnapshot.getReadSchema();
+
+ InternalFilesDiff filesDiff =
+ dataFileExtractor.extractFilesDiff(paimonTable, snapshot, internalSchema);
+
+ return TableChange.builder()
+ .tableAsOfChange(tableAtSnapshot)
+ .filesDiff(filesDiff)
+ .sourceIdentifier(getCommitIdentifier(snapshot))
+ .build();
}
@Override
public CommitsBacklog<Snapshot> getCommitsBacklog(
InstantsForIncrementalSync instantsForIncrementalSync) {
- throw new UnsupportedOperationException("Incremental Sync is not supported yet.");
+ Instant lastSyncInstant = instantsForIncrementalSync.getLastSyncInstant();
+ long lastSyncTimeMillis = lastSyncInstant.toEpochMilli();
+
+ log.info(
+ "Getting commits backlog for Paimon table {} from instant {}",
+ paimonTable.name(),
+ lastSyncInstant);
+
+ Iterator<Snapshot> snapshotIterator;
+ try {
+ snapshotIterator = snapshotManager.snapshots();
+ } catch (IOException e) {
+ throw new ReadException("Could not iterate over the Paimon snapshot list", e);
+ }
+
+ List<Snapshot> snapshotsToProcess = new ArrayList<>();
+ while (snapshotIterator.hasNext()) {
+ Snapshot snapshot = snapshotIterator.next();
+ // Only include snapshots committed after the last sync
+ if (snapshot.timeMillis() > lastSyncTimeMillis) {
+ snapshotsToProcess.add(snapshot);
+ log.debug(
+ "Including snapshot {} (time={}, commitId={}) in backlog",
+ snapshot.id(),
+ snapshot.timeMillis(),
+ snapshot.commitIdentifier());
+ }
+ }
+
+ log.info("Found {} snapshots to process for incremental sync", snapshotsToProcess.size());
+
+ return CommitsBacklog.<Snapshot>builder()
+ .commitsToProcess(snapshotsToProcess)
+ .inFlightInstants(Collections.emptyList())
+ .build();
}
@Override
public boolean isIncrementalSyncSafeFrom(Instant instant) {
- return false; // Incremental sync is not supported yet
+ long timeInMillis = instant.toEpochMilli();
+
+ Long earliestSnapshotId = snapshotManager.earliestSnapshotId();
+ Long latestSnapshotId = snapshotManager.latestSnapshotId();
+ if (earliestSnapshotId == null || latestSnapshotId == null) {
+ log.warn("No snapshots found in table {}", paimonTable.name());
+ return false;
+ }
+
+ Snapshot earliestSnapshot = snapshotManager.snapshot(earliestSnapshotId);
+ Snapshot latestSnapshot = snapshotManager.snapshot(latestSnapshotId);
+
+ // Check 1: If instant is in the future (after latest snapshot), return false
+ if (timeInMillis > latestSnapshot.timeMillis()) {
+ log.warn(
+ "Instant {} is in the future. Latest snapshot {} has time {}",
+ instant,
+ latestSnapshot.id(),
+ latestSnapshot.timeMillis());
+ return false;
+ }
+
+ // Check 2: Has snapshot expiration affected this instant?
+ // If the earliest snapshot is newer than the requested instant,
+ // then snapshots have been expired and we can't do incremental sync
+ if (earliestSnapshot.timeMillis() > timeInMillis) {
+ log.warn(
+ "Incremental sync is not safe from instant {}. "
+ + "Earliest available snapshot {} (time={}) is newer than the requested instant. "
+ + "Snapshots may have been expired.",
+ instant,
+ earliestSnapshot.id(),
+ earliestSnapshot.timeMillis());
+ return false;
+ }
+
+ // Check 3: Verify a snapshot exists at or before the instant
Review Comment:
done, much better now!
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement Paimon Source Incremental Sync [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #780:
URL: https://github.com/apache/incubator-xtable/pull/780#discussion_r2659998204
##########
xtable-core/src/test/java/org/apache/xtable/paimon/TestPaimonDataFileExtractor.java:
##########
@@ -147,6 +147,99 @@ void testColumnStatsAreEmpty() {
}
}
+ @Test
+ void testExtractFilesDiffWithNewFiles() {
+ createUnpartitionedTable();
+
+ // Insert initial data
+ testTable.insertRows(5);
+ org.apache.paimon.Snapshot firstSnapshot = paimonTable.snapshotManager().latestSnapshot();
+ assertNotNull(firstSnapshot);
+
+ // Insert more data to create a second snapshot
+ testTable.insertRows(3);
+ org.apache.paimon.Snapshot secondSnapshot = paimonTable.snapshotManager().latestSnapshot();
+ assertNotNull(secondSnapshot);
+
+ org.apache.xtable.model.storage.InternalFilesDiff filesDiff =
+ extractor.extractFilesDiff(paimonTable, secondSnapshot, testSchema);
+
+ // Verify we have added files
+ assertNotNull(filesDiff);
+ assertNotNull(filesDiff.getFilesAdded());
+ assertTrue(filesDiff.getFilesAdded().size() > 0);
+
+ // Verify removed files collection exists (size may vary based on compaction behavior)
+ assertNotNull(filesDiff.getFilesRemoved());
Review Comment:
On this current setup it will always be 1. I've changed to reflect that.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement Paimon Source Incremental Sync [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #780:
URL: https://github.com/apache/incubator-xtable/pull/780#discussion_r2659998475
##########
xtable-core/src/test/java/org/apache/xtable/paimon/TestPaimonDataFileExtractor.java:
##########
@@ -147,6 +147,99 @@ void testColumnStatsAreEmpty() {
}
}
+ @Test
+ void testExtractFilesDiffWithNewFiles() {
+ createUnpartitionedTable();
+
+ // Insert initial data
+ testTable.insertRows(5);
+ org.apache.paimon.Snapshot firstSnapshot = paimonTable.snapshotManager().latestSnapshot();
+ assertNotNull(firstSnapshot);
+
+ // Insert more data to create a second snapshot
+ testTable.insertRows(3);
+ org.apache.paimon.Snapshot secondSnapshot = paimonTable.snapshotManager().latestSnapshot();
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org