voonhous commented on code in PR #19726:
URL: https://github.com/apache/hudi/pull/19726#discussion_r3888879103
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/testutils/HiveTestUtil.java:
##########
@@ -367,6 +367,30 @@ public static void createReplaceCommit(String instantTime,
String partitions, Wr
createReplaceCommitFile(replaceCommitMetadata, instantTime);
}
+ /**
+ * Writes the data files and the requested + inflight markers of an
INSERT_OVERWRITE replacecommit
+ * but not the completed file, i.e. a write that is still running. Complete
it with
+ * {@link #createReplaceCommitFile}.
+ */
+ public static HoodieReplaceCommitMetadata
startInsertOverwritePartition(String partitionPath, String instantTime)
+ throws IOException, URISyntaxException {
+ HoodieCommitMetadata commitMetadata = createPartition(partitionPath, true,
true, instantTime);
+ HoodieReplaceCommitMetadata replaceCommitMetadata = new
HoodieReplaceCommitMetadata();
+ commitMetadata.getPartitionToWriteStats().forEach((p, stats) ->
stats.forEach(s -> replaceCommitMetadata.addWriteStat(p, s)));
Review Comment:
**major:** This write-stat copy is load-bearing and nothing says so.
`TimelineUtils.isDeletePartition` counts `INSERT_OVERWRITE`
(`TimelineUtils.java:617-621`), so without it the new partition is resolved as
dropped rather than added, and it survives only via the delete-ts vs write-ts
filter at `TimelineUtils.java:167-170`.
This is also the one branch this suite protects that
`testHiveSyncWithMultiWriter` does not: flipping `GREATER_THAN` to
`GREATER_THAN_OR_EQUALS` on line 170 fails this test and passes all six legs of
that one.
Could a one-line comment record why the copy is required?
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncToolLongRunningWriteWatermark.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.model.HoodieReplaceCommitMetadata;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.hive.testutils.HiveTestUtil;
+import org.apache.hudi.sync.common.model.Partition;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+
+import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SYNC_MODE;
+import static org.apache.hudi.hive.testutils.HiveTestUtil.TABLE_NAME;
+import static org.apache.hudi.hive.testutils.HiveTestUtil.hiveSyncProps;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A partition written by a long-running INSERT_OVERWRITE must still be
registered when the incremental
+ * sync's watermark moved past the write's instant time while it was in
flight. That only works when the
+ * sync client persists and reads the commit completion-time watermark ({@code
last_commit_completion_time_sync}),
+ * which drives the "hollow instant" lookup in {@code
TimelineUtils.getCommitsTimelineAfter}; a client that
+ * tracks {@code last_commit_time_sync} alone silently and permanently drops
the partition.
+ */
+public class TestHiveSyncToolLongRunningWriteWatermark {
+
+ private static final String EXISTING_PARTITION_INSTANT = "100";
+ private static final String LONG_RUNNING_INSERT_OVERWRITE_INSTANT = "101";
+ private static final String EMPTY_COMMIT_DURING_WRITE = "102";
+ private static final String EMPTY_COMMIT_AFTER_WRITE = "103";
+ private static final String NEW_PARTITION = "2026/08/04";
+ /** {@link org.apache.hudi.hive.SlashEncodedDayPartitionValueExtractor} maps
{@link #NEW_PARTITION} to this datestr value. */
+ private static final List<String> NEW_PARTITION_VALUES =
Collections.singletonList("2026-08-04");
+
+ private HiveSyncTool hiveSyncTool;
+ private HoodieHiveSyncClient hiveClient;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ HiveTestUtil.setUp(Option.empty(), true);
+ hiveSyncProps.setProperty(HIVE_SYNC_MODE.key(), "hms");
+ }
+
+ @AfterEach
+ void teardown() throws Exception {
+ closeHiveSyncTool();
+ HiveTestUtil.clear();
+ }
+
+ @AfterAll
+ static void cleanUpClass() throws IOException {
+ HiveTestUtil.shutdown();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testPartitionOfWriteCompletingAfterWatermarkAdvanced(boolean
clientPersistsCompletionTime) throws Exception {
+ // t=100: table with one partition, synced. Watermark = 100.
+ HiveTestUtil.createCOWTable(EXISTING_PARTITION_INSTANT, 1, true);
+ reSyncHiveTable(clientPersistsCompletionTime);
+ assertEquals(1, hiveClient.getAllPartitions(TABLE_NAME).size());
+ assertEquals(EXISTING_PARTITION_INSTANT,
hiveClient.getLastCommitTimeSynced(TABLE_NAME).get());
+
+ // t=101: INSERT_OVERWRITE into a brand-new partition starts (requested +
inflight, files on storage)
+ // and keeps running for a long time.
+ HoodieReplaceCommitMetadata longRunningWrite =
+ HiveTestUtil.startInsertOverwritePartition(NEW_PARTITION,
LONG_RUNNING_INSERT_OVERWRITE_INSTANT);
+
+ // t=102: a concurrent writer lands an empty commit (no partitions) while
101 is still in flight.
+ HiveTestUtil.addEmptyCommit(EMPTY_COMMIT_DURING_WRITE);
+
+ // The periodic sync runs now: 101 is pending so it is invisible, and the
watermark jumps to 102.
+ reSyncHiveTable(clientPersistsCompletionTime);
+ assertEquals(1, hiveClient.getAllPartitions(TABLE_NAME).size(), "pending
write must not be synced yet");
+ assertEquals(EMPTY_COMMIT_DURING_WRITE,
hiveClient.getLastCommitTimeSynced(TABLE_NAME).get());
+
+ // 101 completes: its instant time is now BELOW the watermark, while its
completion time (stamped into
+ // the completed file's name at creation) is later than 102's.
+ HiveTestUtil.createReplaceCommitFile(longRunningWrite,
LONG_RUNNING_INSERT_OVERWRITE_INSTANT);
+ assertTrue(HiveTestUtil.fileSystem.exists(new Path(HiveTestUtil.basePath,
NEW_PARTITION)),
+ "partition data is on storage");
+
+ // Next sync cycle.
+ reSyncHiveTable(clientPersistsCompletionTime);
+ List<Partition> partitionsAfterCompletion =
hiveClient.getAllPartitions(TABLE_NAME);
+
+ // t=103: more unrelated commits keep arriving; every later sync advances
the watermark further.
+ HiveTestUtil.addEmptyCommit(EMPTY_COMMIT_AFTER_WRITE);
+ reSyncHiveTable(clientPersistsCompletionTime);
+ List<Partition> partitionsAfterLaterSyncs =
hiveClient.getAllPartitions(TABLE_NAME);
+ assertEquals(EMPTY_COMMIT_AFTER_WRITE,
hiveClient.getLastCommitTimeSynced(TABLE_NAME).get());
+
+ if (clientPersistsCompletionTime) {
+ // Stock Hive client: the completion-time watermark rescues the hollow
instant.
+ assertEquals(2, partitionsAfterCompletion.size(), "hollow instant
rescued by completion-time watermark");
+ assertTrue(containsNewPartition(partitionsAfterCompletion), "the
registered partition is the INSERT_OVERWRITE's");
+ assertEquals(2, partitionsAfterLaterSyncs.size());
+ assertTrue(containsNewPartition(partitionsAfterLaterSyncs));
+ } else {
+ // Instant-time-only client: the partition is on storage with a
completed replacecommit, yet it is
+ // never added - and the watermark keeps moving, so no future
incremental sync will add it.
+ assertEquals(1, partitionsAfterCompletion.size(), "partition of the
late-completing write was skipped");
+ assertFalse(containsNewPartition(partitionsAfterCompletion));
+ assertEquals(1, partitionsAfterLaterSyncs.size(), "partition is
permanently lost to incremental sync");
+ assertFalse(containsNewPartition(partitionsAfterLaterSyncs));
+ // read through a stock client to show the completion key was never
written to the metastore
+ try (HiveSyncTool stockTool = new HiveSyncTool(hiveSyncProps,
HiveTestUtil.getHiveConf())) {
+ assertFalse(((HoodieHiveSyncClient)
stockTool.syncClient).getLastCommitCompletionTimeSynced(TABLE_NAME).isPresent());
Review Comment:
**major:** The negative leg pins the behaviour of a test-local subclass, but
the same loss happens with the **stock** client on any table last synced before
HUDI-6182: the metastore then has `last_commit_time_sync` and no completion
key, so `isAlreadySynced` takes `.orElse(true)` (`HiveSyncTool.java:327`) and
short-circuits every later cycle. `AdbSyncTool.java:190` still hard-codes
`Option.empty()`, so the shape is live in-repo.
Could the last cycle re-run with a stock `HiveSyncTool` and assert the
partition is still absent, so the arm pins the upgrade path rather than the
fake?
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java:
##########
@@ -229,6 +229,15 @@ public void teardown() throws Exception {
HiveTestUtil.clear();
}
+ @Test
+ public void testUpdateLastCommitTimeSyncedWithNoCompletedCommit() {
+ hiveSyncProps.setProperty(HIVE_SYNC_MODE.key(), "hms");
+ // created-but-never-written table: the active timeline has no completed
instant, the update must no-op
+ reInitHiveSyncClient();
+ assertDoesNotThrow(() ->
hiveClient.updateLastCommitTimeSynced(HiveTestUtil.TABLE_NAME));
Review Comment:
**minor:** This pins the `!lastCommitSynced.isPresent()` early return, but
its JDBC-fallback sibling `updateLastCommitTimeSyncedViaJdbc`
(`HoodieHiveSyncClient.java:755-767`) writes both watermark keys and has no
coverage at all. The read half is already pinned at
`TestHoodieHiveSyncClientOperations.java:95`, and that file never calls
`updateLastCommitTimeSynced`. Not blocking.
Would it be worth one
`verify(fixture.jdbcMetadataOperator).setTableProperties(...)` on that existing
fallback fixture, so both halves are covered?
--
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]
