nsivabalan commented on PR #20273:
URL: https://github.com/apache/hudi/pull/20273#issuecomment-6157370012
Thanks @ad1happy2go, nice debugging on this one. The core fix looks right to
me. Treating "first instant on the active timeline" as the archival boundary
breaks as soon as a rollback or clean outlives the commits around it, and the
archiver archives those on a separate schedule. Moving
`BaseHoodieTableFileIndex#isBeforeTimelineStarts` and the
`CompletionTimeQueryViewV2` cursor to the write timeline fixes the reported bug.
Most of the remaining diff is about archival beyond savepoint:
`activeSavepointedInstants`, the new `isArchived` boundary, the per-instant
`splitInstantByActiveness`, and the savepoint tests. I dug into that feature,
and I don't think we should keep extending read paths for it.
The hole it leaves after a savepointed commit breaks much more than
incremental reads. `getFirstNonSavepointCommit()` only skips a savepointed
commit when the timeline it's given contains the SAVEPOINT instant. The file
system view (`AbstractTableFileSystemView#refreshTimeline` →
`getWriteTimeline()`), the cleaner and the index all pass timelines without it.
So a commit in the hole counts as uncommitted.
I reproduced this on master with a Spark functional test on COW. All four
combinations behave the same: table version 6 and 9, each with the BLOOM and
SIMPLE index.
- C1 inserts into partition 1 and is savepointed.
- C2 inserts 10 records into partition 2.
- Four more commits into partition 3 archive C2, while C1 stays active.
| Step | Expected | Actual |
|---|---|---|
| Snapshot read | 60 rows | **50**: C2's rows are missing |
| Upsert of C2's keys | update C2's file group | **new file group** in
partition 2 |
| Delete savepoint C1, archive past it | 100 rows / 100 keys | **110 rows /
100 keys**: 10 duplicates |
That's the same root cause as #11405, which has been open since 2024. With
the LSM timeline we're moving away from a separate archived timeline anyway, so
I've opened #20310 to remove support for `hoodie.archive.beyond.savepoint`. It
deprecates the config, fails fast when it's set to true, and replaces the
config-based restore guard with a check on the table itself.
Given that, could we narrow this PR to the stale rollback/clean fix?
- Keep the write-timeline boundary in `isBeforeTimelineStarts` and the
`CompletionTimeQueryViewV2` cursor, plus the stale-rollback tests
(`testIncrementalStartBeforeFirstWriteIsArchivedDespiteOlderNonWriteInstant`,
`testIncrementalStartIsArchivedWithoutActiveWriteInstant`,
`testArchivedInstantsAfterStaleNonWriteInstant`, and the Spark functional test).
- Drop the savepoint-hole handling: `activeSavepointedInstants`, the
`isArchived` / `isCompleted` split, and the `splitInstantByActiveness` rewrite
with its sorting.
That also settles the open Flink comment about
`QueryContext#getMaxCompletionTime`. That issue only exists because archived
and active instants can now interleave, which can't happen once the feature is
gone.
<details><summary>Repro test (TestArchiveBeyondSavepointReads,
hudi-spark)</summary>
```java
package org.apache.hudi.functional;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.config.HoodieArchivalConfig;
import org.apache.hudi.config.HoodieCleanConfig;
import org.apache.hudi.config.HoodieIndexConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.testutils.HoodieClientTestBase;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Stream;
import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH;
import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH;
import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_THIRD_PARTITION_PATH;
import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors;
import static org.junit.jupiter.api.Assertions.assertAll;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Repro: with hoodie.archive.beyond.savepoint=true, commits archived after
a savepointed commit leave a hole in the
* active timeline. File slices written only by commits in the hole are
treated as uncommitted by the file system view.
*/
@Tag("functional")
public class TestArchiveBeyondSavepointReads extends HoodieClientTestBase {
private static final int NUM_RECORDS = 10;
private static final String REPORT_FILE =
"target/archive-beyond-savepoint-repro.txt";
@BeforeEach
public void setUp() throws Exception {
initPath();
initSparkContexts();
initTestDataGenerator();
initHoodieStorage();
initTimelineService();
}
private static Stream<Arguments> tableVersionAndIndex() {
return Stream.of(
Arguments.of(HoodieTableVersion.SIX, HoodieIndex.IndexType.BLOOM),
Arguments.of(HoodieTableVersion.SIX, HoodieIndex.IndexType.SIMPLE),
Arguments.of(HoodieTableVersion.NINE, HoodieIndex.IndexType.BLOOM),
Arguments.of(HoodieTableVersion.NINE, HoodieIndex.IndexType.SIMPLE));
}
/**
* C1 inserts into partition 1 and is savepointed. C2 inserts into
partition 2, a file group no later commit touches.
* C3..C6 insert into partition 3, which archives C2 while C1 stays
active. Then:
* 1. a snapshot read should still return C2's rows;
* 2. an upsert of C2's keys should update them in place, not insert them
again;
* 3. once the savepoint is deleted and C1 is archived too, the table
should not contain duplicate keys.
*/
@ParameterizedTest
@MethodSource("tableVersionAndIndex")
void testHoleCommitsAfterSavepoint(HoodieTableVersion tableVersion,
HoodieIndex.IndexType indexType) throws Exception {
HoodieWriteConfig writeConfig = getWriteConfigAndInitTable(tableVersion,
indexType);
List<String> observations = new ArrayList<>();
try (SparkRDDWriteClient client = getHoodieWriteClient(writeConfig)) {
// given: savepointed C1, hole commit C2, and enough later commits to
archive C2
String c1 = insert(client, DEFAULT_FIRST_PARTITION_PATH);
client.savepoint(c1, "user", "savepoint C1");
List<HoodieRecord> holeRecords = new ArrayList<>();
String c2 = insert(client, DEFAULT_SECOND_PARTITION_PATH, holeRecords);
for (int i = 0; i < 4; i++) {
insert(client, DEFAULT_THIRD_PARTITION_PATH);
}
HoodieTimeline activeTimeline = metaClient.reloadActiveTimeline();
assertTrue(activeTimeline.containsInstant(c1), "precondition:
savepointed C1 stays active");
assertFalse(activeTimeline.containsInstant(c2), "precondition: C2 is
archived, leaving a hole after C1");
// when/then 1: snapshot read
long snapshotCount = snapshotRead().count();
observations.add("snapshot rows after archival (expected 60): " +
snapshotCount);
// when/then 2: upsert C2's keys
String newCommitTime = client.startCommit();
List<HoodieRecord> updates = dataGen.generateUpdates(newCommitTime,
holeRecords);
client.commit(newCommitTime, client.upsert(jsc.parallelize(updates,
1), newCommitTime));
Dataset<Row> afterUpsert = snapshotRead();
long upsertRows = afterUpsert.count();
long upsertKeys =
afterUpsert.select("_hoodie_record_key").distinct().count();
long holeFileGroupsInStorage =
countBaseFileGroups(DEFAULT_SECOND_PARTITION_PATH);
observations.add("after upsert of C2 keys: rows=" + upsertRows + "
distinctKeys=" + upsertKeys
+ " fileGroupsInPartition2=" + holeFileGroupsInStorage + "
(expected 60/60/1)");
// when/then 3: delete the savepoint and archive C1, so the hole closes
client.deleteSavepoint(c1);
for (int i = 0; i < 4; i++) {
insert(client, DEFAULT_THIRD_PARTITION_PATH);
}
assertFalse(metaClient.reloadActiveTimeline().containsInstant(c1),
"precondition: C1 archived after savepoint delete");
Dataset<Row> afterHoleCloses = snapshotRead();
long finalRows = afterHoleCloses.count();
long finalKeys =
afterHoleCloses.select("_hoodie_record_key").distinct().count();
observations.add("after savepoint delete and archival: rows=" +
finalRows + " distinctKeys=" + finalKeys
+ " (expected " + (60 + 4 * NUM_RECORDS) + "/" + (60 + 4 *
NUM_RECORDS) + ")");
report(tableVersion, indexType, observations);
long expectedFinal = 60 + 4 * NUM_RECORDS;
assertAll(
() -> assertEquals(60, snapshotCount, "snapshot read drops rows of
the archived hole commit"),
() -> assertEquals(60, upsertRows, "rows after upserting the hole
commit's keys"),
() -> assertEquals(60, upsertKeys, "distinct keys after upserting
the hole commit's keys"),
() -> assertEquals(1, holeFileGroupsInStorage, "upsert should
update the existing file group, not create a new one"),
() -> assertEquals(expectedFinal, finalRows, "rows once the hole
closes"),
() -> assertEquals(finalRows, finalKeys, "duplicate record keys
once the hole closes"));
}
}
private HoodieWriteConfig getWriteConfigAndInitTable(HoodieTableVersion
tableVersion, HoodieIndex.IndexType indexType) throws IOException {
HoodieWriteConfig writeConfig =
getConfigBuilder(HoodieFailedWritesCleaningPolicy.EAGER)
.withAutoUpgradeVersion(false)
.withWriteTableVersion(tableVersion.versionCode())
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(indexType).build())
.withCleanConfig(HoodieCleanConfig.newBuilder().retainCommits(1).build())
.withArchivalConfig(HoodieArchivalConfig.newBuilder()
.archiveCommitsWith(2, 3)
.withArchiveBeyondSavepoint(true)
.build())
.withMetadataConfig(HoodieMetadataConfig.newBuilder()
.enable(true)
.withMaxNumDeltaCommitsBeforeCompaction(1)
.build())
.build();
initMetaClient(HoodieTableType.COPY_ON_WRITE, writeConfig.getProps());
return writeConfig;
}
private String insert(SparkRDDWriteClient client, String partition) {
return insert(client, partition, new ArrayList<>());
}
private String insert(SparkRDDWriteClient client, String partition,
List<HoodieRecord> insertedRecords) {
String instantTime = client.startCommit();
List<HoodieRecord> records =
dataGen.generateInsertsForPartition(instantTime, NUM_RECORDS, partition);
insertedRecords.addAll(records);
List<org.apache.hudi.client.WriteStatus> statuses =
client.insert(jsc.parallelize(records, 1), instantTime).collect();
assertNoWriteErrors(statuses);
client.commit(instantTime, jsc.parallelize(statuses, 1));
return instantTime;
}
private Dataset<Row> snapshotRead() {
return sparkSession.read().format("hudi").load(basePath);
}
private long countBaseFileGroups(String partition) throws IOException {
try (Stream<java.nio.file.Path> files = Files.list(Paths.get(basePath,
partition))) {
return files.map(path -> path.getFileName().toString())
.filter(name -> name.endsWith(".parquet"))
.map(name -> name.substring(0, name.indexOf('_')))
.distinct()
.count();
}
}
private static void report(HoodieTableVersion tableVersion,
HoodieIndex.IndexType indexType, List<String> observations) throws IOException {
StringBuilder sb = new StringBuilder("== table version " +
tableVersion.versionCode() + ", index " + indexType + "\n");
observations.forEach(line -> sb.append(" ").append(line).append('\n'));
Files.write(Paths.get(REPORT_FILE),
sb.toString().getBytes(StandardCharsets.UTF_8),
StandardOpenOption.CREATE, StandardOpenOption.APPEND);
}
}
```
</details>
--
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]
