This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 99ceae1b6459 fix(meta-sync): advance last commit time synced when it
trails the active timeline midpoint (#19239)
99ceae1b6459 is described below
commit 99ceae1b645967dd39c5aee0ed6e904b363027a1
Author: Y Ethan Guo <[email protected]>
AuthorDate: Mon Aug 3 10:50:04 2026 -0700
fix(meta-sync): advance last commit time synced when it trails the active
timeline midpoint (#19239)
* fix(meta-sync): advance last commit time synced when it trails the active
timeline midpoint
* docs(meta-sync): clarify the marker advances on catalog-visible change,
not on data write
* docs(meta-sync): say "last commit time synced" instead of "marker" in the
javadoc
* test(meta-sync): unit-test the timeline-midpoint helper and take the
midpoint over completed commits
Compute the midpoint over completed commit instants only so an inflight
instant cannot shift it, and add TestHiveSyncToolTimelineMidpoint covering
the present/empty guards and the inflight-vs-completed boundary.
* test(meta-sync): narrow the timeline midpoint to completed commits
Use getCommitsTimeline().filterCompletedInstants() so clean, rollback, and
other non-commit instants cannot shift the midpoint, and add a test that
pins the narrowing to the commits timeline.
* refactor(meta-sync): read completed commits via
metaClient.getCommitsTimeline()
Keep the helper and its test identical across lines by reading the commits
timeline from metaClient.getCommitsTimeline(), which is available
everywhere,
rather than the HoodieTimeline interface method.
* test(meta-sync): rename midpoint test for clarity
Rename midpointIsTakenOverCompletedCommitsOnly to
midpointIsComputedFromCompletedCommitsOnly.
---
.../java/org/apache/hudi/hive/HiveSyncTool.java | 28 ++++-
.../org/apache/hudi/hive/TestHiveSyncTool.java | 12 ++-
.../hive/TestHiveSyncToolTimelineMidpoint.java | 114 +++++++++++++++++++++
3 files changed, 149 insertions(+), 5 deletions(-)
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
index 80b83816be8b..4c8aca377089 100644
---
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
@@ -26,6 +26,7 @@ import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.util.ConfigUtils;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.VisibleForTesting;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.InvalidTableException;
import org.apache.hudi.sync.common.HoodieSyncClient;
@@ -51,6 +52,8 @@ import java.util.Properties;
import java.util.Set;
import java.util.stream.Collectors;
+import static
org.apache.hudi.common.table.timeline.InstantComparison.LESSER_THAN;
+import static
org.apache.hudi.common.table.timeline.InstantComparison.compareTimestamps;
import static org.apache.hudi.common.util.StringUtils.nonEmpty;
import static
org.apache.hudi.hadoop.utils.HoodieInputFormatUtils.getInputFormatClassName;
import static
org.apache.hudi.hadoop.utils.HoodieInputFormatUtils.getOutputFormatClassName;
@@ -276,7 +279,8 @@ public class HiveSyncTool extends HoodieSyncTool implements
AutoCloseable {
boolean partitionsChanged = validateAndSyncPartitions(tableName,
tableExists);
boolean meetSyncConditions = schemaChanged || propertiesChanged ||
partitionsChanged;
- if (!config.getBoolean(META_SYNC_CONDITIONAL_SYNC) ||
meetSyncConditions) {
+ if (!config.getBoolean(META_SYNC_CONDITIONAL_SYNC) || meetSyncConditions
+ || isLastCommitTimeSyncedBehindTimelineMidpoint(tableName)) {
syncClient.updateLastCommitTimeSynced(tableName);
}
syncClient.updateHoodieWriterVersion(tableName);
@@ -291,6 +295,28 @@ public class HiveSyncTool extends HoodieSyncTool
implements AutoCloseable {
}
}
+ /**
+ * Whether last commit time synced trails the midpoint of the completed
commit instants.
+ * Advancing it at the midpoint bounds how far it can fall behind, capping
the archived-timeline
+ * scans a stale value forces on every conditional-sync round.
+ */
+ @VisibleForTesting
+ boolean isLastCommitTimeSyncedBehindTimelineMidpoint(String tableName) {
+ Option<String> lastCommitTimeSynced =
syncClient.getLastCommitTimeSynced(tableName);
+ if (!lastCommitTimeSynced.isPresent()) {
+ return false;
+ }
+ // Completed commits only: getCommitsTimeline() excludes non-commit
actions (clean, rollback),
+ // and filterCompletedInstants() excludes inflight instants.
+ List<HoodieInstant> completedCommits =
+
syncClient.getMetaClient().getCommitsTimeline().filterCompletedInstants().getInstants();
+ if (completedCommits.isEmpty()) {
+ return false;
+ }
+ String midpointInstantTime = completedCommits.get(completedCommits.size()
/ 2).requestedTime();
+ return compareTimestamps(lastCommitTimeSynced.get(), LESSER_THAN,
midpointInstantTime);
+ }
+
private boolean isAlreadySynced(String tableName) {
return syncClient.getLastCommitTimeSynced(tableName)
.map(lastCommit -> {
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
index 9f5211ba5efe..baf90079ada7 100644
---
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
@@ -2536,12 +2536,16 @@ public class TestHiveSyncTool {
HiveTestUtil.addMORPartitions(0, true, true, true,
ZonedDateTime.now().plusDays(2), commitTime2, commitTime3);
+ // No sync condition is met, but the sync marker trails the midpoint of
the active commits
+ // timeline ([100, 101, 102, 103] with midpoint 102), so it advances to
the last commit.
+ reInitHiveSyncClient();
reSyncHiveTable();
- assertEquals(commitTime1,
hiveClient.getLastCommitTimeSynced(tableName).get());
+ assertEquals(commitTime3,
hiveClient.getLastCommitTimeSynced(tableName).get());
// Let the last commit time synced to be before the start of the active
timeline,
- // to trigger the fallback of listing all partitions. There is no
partition change
- // and the last commit time synced should still be the same.
+ // to trigger the fallback of listing all partitions. There is no
partition change,
+ // and the sync marker again trails the timeline midpoint, so it advances
to the
+ // last commit instead of aging out further.
HiveTestUtil.addMORPartitions(0, true, true, true,
ZonedDateTime.now().plusDays(2), commitTime4, commitTime5);
HiveTestUtil.removeCommitFromActiveTimeline(commitTime0, COMMIT_ACTION);
HiveTestUtil.removeCommitFromActiveTimeline(commitTime1,
DELTA_COMMIT_ACTION);
@@ -2549,7 +2553,7 @@ public class TestHiveSyncTool {
HiveTestUtil.removeCommitFromActiveTimeline(commitTime3,
DELTA_COMMIT_ACTION);
reInitHiveSyncClient();
reSyncHiveTable();
- assertEquals(commitTime1,
hiveClient.getLastCommitTimeSynced(tableName).get());
+ assertEquals(commitTime5,
hiveClient.getLastCommitTimeSynced(tableName).get());
}
@ParameterizedTest
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncToolTimelineMidpoint.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncToolTimelineMidpoint.java
new file mode 100644
index 000000000000..3d0b5b3dd431
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncToolTimelineMidpoint.java
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.hive;
+
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.testutils.MockHoodieTimeline;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.sync.common.HoodieSyncClient;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link
HiveSyncTool#isLastCommitTimeSyncedBehindTimelineMidpoint}, which decides
+ * whether a no-change conditional sync should still advance last commit time
synced. The helper
+ * reads the completed commits timeline, so these mock {@code
getMetaClient().getCommitsTimeline()};
+ * a call to any other timeline would hit an unstubbed mock and fail.
+ */
+class TestHiveSyncToolTimelineMidpoint {
+
+ private static final String TABLE_NAME = "table";
+
+ @Test
+ void midpointIsComputedFromCompletedCommitsOnly() {
+ // Completed commits [100, 102, 104], midpoint 102; the later inflight 106
must not shift it to 104.
+ HoodieSyncClient syncClient = mockSyncClient(new
MockHoodieTimeline(Stream.of("100", "102", "104"), Stream.of("106")));
+ HiveSyncTool tool = toolWith(syncClient);
+
+ // Not synced yet: nothing to advance.
+ stubLastCommitTimeSynced(syncClient, Option.empty());
+ assertFalse(tool.isLastCommitTimeSyncedBehindTimelineMidpoint(TABLE_NAME));
+
+ // Trails the midpoint.
+ stubLastCommitTimeSynced(syncClient, Option.of("101"));
+ assertTrue(tool.isLastCommitTimeSyncedBehindTimelineMidpoint(TABLE_NAME));
+
+ // At the midpoint is not behind it.
+ stubLastCommitTimeSynced(syncClient, Option.of("102"));
+ assertFalse(tool.isLastCommitTimeSyncedBehindTimelineMidpoint(TABLE_NAME));
+
+ // Past 102 but below 104: behind only if the inflight 106 is wrongly
counted.
+ stubLastCommitTimeSynced(syncClient, Option.of("103"));
+ assertFalse(tool.isLastCommitTimeSyncedBehindTimelineMidpoint(TABLE_NAME));
+ }
+
+ @Test
+ void midpointHandlesSmallTimelines() {
+ // Size 1: the sole commit is the midpoint.
+ HoodieSyncClient syncClient = mockSyncClient(new
MockHoodieTimeline(Stream.of("101"), Stream.empty()));
+ HiveSyncTool tool = toolWith(syncClient);
+ stubLastCommitTimeSynced(syncClient, Option.of("100"));
+ assertTrue(tool.isLastCommitTimeSyncedBehindTimelineMidpoint(TABLE_NAME));
+ stubLastCommitTimeSynced(syncClient, Option.of("101"));
+ assertFalse(tool.isLastCommitTimeSyncedBehindTimelineMidpoint(TABLE_NAME));
+
+ // Size 2: the midpoint (index 1) is the newer commit.
+ syncClient = mockSyncClient(new MockHoodieTimeline(Stream.of("100",
"102"), Stream.empty()));
+ tool = toolWith(syncClient);
+ stubLastCommitTimeSynced(syncClient, Option.of("101"));
+ assertTrue(tool.isLastCommitTimeSyncedBehindTimelineMidpoint(TABLE_NAME));
+ stubLastCommitTimeSynced(syncClient, Option.of("102"));
+ assertFalse(tool.isLastCommitTimeSyncedBehindTimelineMidpoint(TABLE_NAME));
+ }
+
+ @Test
+ void emptyCommitsTimelineIsNotBehind() {
+ HoodieSyncClient syncClient = mockSyncClient(new
MockHoodieTimeline(Stream.empty(), Stream.empty()));
+ HiveSyncTool tool = toolWith(syncClient);
+ stubLastCommitTimeSynced(syncClient, Option.of("103"));
+ assertFalse(tool.isLastCommitTimeSyncedBehindTimelineMidpoint(TABLE_NAME));
+ }
+
+ private static HoodieSyncClient mockSyncClient(HoodieTimeline
commitsTimeline) {
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ when(metaClient.getCommitsTimeline()).thenReturn(commitsTimeline);
+ HoodieSyncClient syncClient = mock(HoodieSyncClient.class);
+ when(syncClient.getMetaClient()).thenReturn(metaClient);
+ return syncClient;
+ }
+
+ private static HiveSyncTool toolWith(HoodieSyncClient syncClient) {
+ HiveSyncTool tool = mock(HiveSyncTool.class, CALLS_REAL_METHODS);
+ tool.syncClient = syncClient;
+ return tool;
+ }
+
+ private static void stubLastCommitTimeSynced(HoodieSyncClient syncClient,
Option<String> value) {
+ when(syncClient.getLastCommitTimeSynced(TABLE_NAME)).thenReturn(value);
+ }
+}
