Skip to content

Commit c24c756

Browse files
rahil-crahil-c
committed
add interface method for needsRollbackPendingCommitAndCompact and move logic before upgrade/downgrade on mdt and dt tables
1 parent a77b623 commit c24c756

File tree

8 files changed

+132
-27
lines changed

8 files changed

+132
-27
lines changed

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/DowngradeHandler.java

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,4 +44,14 @@ public interface DowngradeHandler {
4444
Pair<Map<ConfigProperty, String>, List<ConfigProperty>> downgrade(
4545
HoodieWriteConfig config, HoodieEngineContext context, String instantTime,
4646
SupportsUpgradeDowngrade upgradeDowngradeHelper);
47+
48+
/**
49+
* Indicates whether this downgrade handler needs rollback of pending commits and compaction
50+
* to be executed before the downgrade logic runs.
51+
*
52+
* @return true if rollback and compaction should be performed before downgrade, false otherwise.
53+
*/
54+
default boolean needsRollbackPendingCommitAndCompact() {
55+
return false;
56+
}
4757
}

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/EightToNineUpgradeHandler.java

Lines changed: 6 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -22,31 +22,28 @@
2222
import org.apache.hudi.common.engine.HoodieEngineContext;
2323
import org.apache.hudi.common.model.HoodieIndexMetadata;
2424
import org.apache.hudi.common.table.HoodieTableMetaClient;
25+
import org.apache.hudi.common.table.HoodieTableVersion;
2526
import org.apache.hudi.common.util.Option;
2627
import org.apache.hudi.config.HoodieWriteConfig;
2728
import org.apache.hudi.metadata.HoodieIndexVersion;
28-
import org.apache.hudi.common.model.HoodieTableType;
29-
import org.apache.hudi.common.table.HoodieTableVersion;
3029
import org.apache.hudi.table.HoodieTable;
3130

3231
import java.util.Collections;
3332
import java.util.Map;
3433

35-
import static org.apache.hudi.table.upgrade.UpgradeDowngradeUtils.rollbackFailedWritesAndCompact;
36-
3734
public class EightToNineUpgradeHandler implements UpgradeHandler {
3835

36+
@Override
37+
public boolean needsRollbackPendingCommitAndCompact() {
38+
return true;
39+
}
40+
3941
@Override
4042
public Map<ConfigProperty, String> upgrade(HoodieWriteConfig config,
4143
HoodieEngineContext context,
4244
String instantTime,
4345
SupportsUpgradeDowngrade upgradeDowngradeHelper) {
4446
final HoodieTable table = upgradeDowngradeHelper.getTable(config, context);
45-
// Rollback and run compaction in one step
46-
rollbackFailedWritesAndCompact(
47-
table, context, config, upgradeDowngradeHelper,
48-
HoodieTableType.MERGE_ON_READ.equals(table.getMetaClient().getTableType()),
49-
HoodieTableVersion.NINE);
5047

5148
// If auto upgrade is disabled, set writer version to 8 and return
5249
if (!config.autoUpgrade()) {

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/EightToSevenDowngradeHandler.java

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,6 @@
8787
import static org.apache.hudi.metadata.MetadataPartitionType.RECORD_INDEX;
8888
import static org.apache.hudi.table.upgrade.UpgradeDowngradeUtils.EIGHT_TO_SIX_TIMELINE_ACTION_MAP;
8989
import static org.apache.hudi.table.upgrade.UpgradeDowngradeUtils.convertCompletionTimeToEpoch;
90-
import static org.apache.hudi.table.upgrade.UpgradeDowngradeUtils.rollbackFailedWritesAndCompact;
9190

9291
/**
9392
* Version 7 is going to be placeholder version for bridge release 0.16.0.
@@ -96,14 +95,17 @@
9695
public class EightToSevenDowngradeHandler implements DowngradeHandler {
9796

9897
private static final Logger LOG = LoggerFactory.getLogger(EightToSevenDowngradeHandler.class);
98+
99+
@Override
100+
public boolean needsRollbackPendingCommitAndCompact() {
101+
return true;
102+
}
99103
private static final Set<String> SUPPORTED_METADATA_PARTITION_PATHS = getSupportedMetadataPartitionPaths();
100104

101105
@Override
102106
public Pair<Map<ConfigProperty, String>, List<ConfigProperty>> downgrade(HoodieWriteConfig config, HoodieEngineContext context, String instantTime, SupportsUpgradeDowngrade upgradeDowngradeHelper) {
103107
final HoodieTable table = upgradeDowngradeHelper.getTable(config, context);
104108
Map<ConfigProperty, String> tablePropsToAdd = new HashMap<>();
105-
// Rollback and run compaction in one step
106-
rollbackFailedWritesAndCompact(table, context, config, upgradeDowngradeHelper, HoodieTableType.MERGE_ON_READ.equals(table.getMetaClient().getTableType()), HoodieTableVersion.EIGHT);
107109

108110
HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder().setConf(context.getStorageConf().newInstance()).setBasePath(config.getBasePath()).build();
109111
// Handle timeline downgrade:

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/NineToEightDowngradeHandler.java

Lines changed: 6 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,6 @@
2121

2222
import org.apache.hudi.common.config.ConfigProperty;
2323
import org.apache.hudi.common.engine.HoodieEngineContext;
24-
import org.apache.hudi.common.model.HoodieTableType;
25-
import org.apache.hudi.common.table.HoodieTableVersion;
2624
import org.apache.hudi.common.util.collection.Pair;
2725
import org.apache.hudi.config.HoodieWriteConfig;
2826
import org.apache.hudi.table.HoodieTable;
@@ -31,20 +29,19 @@
3129
import java.util.List;
3230
import java.util.Map;
3331

34-
import static org.apache.hudi.table.upgrade.UpgradeDowngradeUtils.rollbackFailedWritesAndCompact;
35-
3632
public class NineToEightDowngradeHandler implements DowngradeHandler {
33+
34+
@Override
35+
public boolean needsRollbackPendingCommitAndCompact() {
36+
return true;
37+
}
38+
3739
@Override
3840
public Pair<Map<ConfigProperty, String>, List<ConfigProperty>> downgrade(HoodieWriteConfig config,
3941
HoodieEngineContext context,
4042
String instantTime,
4143
SupportsUpgradeDowngrade upgradeDowngradeHelper) {
4244
final HoodieTable table = upgradeDowngradeHelper.getTable(config, context);
43-
// Rollback and run compaction in one step
44-
rollbackFailedWritesAndCompact(
45-
table, context, config, upgradeDowngradeHelper,
46-
HoodieTableType.MERGE_ON_READ.equals(table.getMetaClient().getTableType()),
47-
HoodieTableVersion.NINE);
4845
UpgradeDowngradeUtils.dropNonV1SecondaryIndexPartitions(
4946
config, context, table, upgradeDowngradeHelper, "downgrading from table version nine to eight");
5047
return Pair.of(Collections.emptyMap(), Collections.emptyList());

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/SevenToEightUpgradeHandler.java

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,6 @@
7373
import static org.apache.hudi.common.table.timeline.TimelineLayout.TIMELINE_LAYOUT_V1;
7474
import static org.apache.hudi.common.table.timeline.TimelineLayout.TIMELINE_LAYOUT_V2;
7575
import static org.apache.hudi.table.upgrade.UpgradeDowngradeUtils.SIX_TO_EIGHT_TIMELINE_ACTION_MAP;
76-
import static org.apache.hudi.table.upgrade.UpgradeDowngradeUtils.rollbackFailedWritesAndCompact;
7776

7877
/**
7978
* Version 7 is going to be placeholder version for bridge release 0.16.0.
@@ -83,6 +82,11 @@ public class SevenToEightUpgradeHandler implements UpgradeHandler {
8382

8483
private static final Logger LOG = LoggerFactory.getLogger(SevenToEightUpgradeHandler.class);
8584

85+
@Override
86+
public boolean needsRollbackPendingCommitAndCompact() {
87+
return true;
88+
}
89+
8690
@Override
8791
public Map<ConfigProperty, String> upgrade(HoodieWriteConfig config, HoodieEngineContext context,
8892
String instantTime, SupportsUpgradeDowngrade upgradeDowngradeHelper) {
@@ -101,8 +105,6 @@ public Map<ConfigProperty, String> upgrade(HoodieWriteConfig config, HoodieEngin
101105
HoodieTableMetadataUtil.deleteMetadataTable(config.getBasePath(), context);
102106
}
103107

104-
// Rollback and run compaction in one step
105-
rollbackFailedWritesAndCompact(table, context, config, upgradeDowngradeHelper, HoodieTableType.MERGE_ON_READ.equals(table.getMetaClient().getTableType()), HoodieTableVersion.SIX);
106108
try {
107109
HoodieTableMetaClient.createTableLayoutOnStorage(context.getStorageConf(), new StoragePath(config.getBasePath()), config.getProps(), TimelineLayoutVersion.VERSION_2, false);
108110
} catch (IOException e) {

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/SixToFiveDowngradeHandler.java

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -49,15 +49,17 @@
4949
*/
5050
public class SixToFiveDowngradeHandler implements DowngradeHandler {
5151

52+
@Override
53+
public boolean needsRollbackPendingCommitAndCompact() {
54+
return true;
55+
}
56+
5257
@Override
5358
public Pair<Map<ConfigProperty, String>, List<ConfigProperty>> downgrade(HoodieWriteConfig config, HoodieEngineContext context, String instantTime, SupportsUpgradeDowngrade upgradeDowngradeHelper) {
5459
final HoodieTable table = upgradeDowngradeHelper.getTable(config, context);
5560

5661
// Since version 6 includes a new schema field for metadata table(MDT), the MDT needs to be deleted during downgrade to avoid column drop error.
5762
HoodieTableMetadataUtil.deleteMetadataTable(config.getBasePath(), context);
58-
// The log block version has been upgraded in version six so compaction is required for downgrade.
59-
UpgradeDowngradeUtils.rollbackFailedWritesAndCompact(table, context, config, upgradeDowngradeHelper, HoodieTableType.MERGE_ON_READ.equals(table.getMetaClient().getTableType()),
60-
HoodieTableVersion.SIX);
6163
UpgradeDowngradeUtils.syncCompactionRequestedFileToAuxiliaryFolder(table);
6264

6365
HoodieTableMetaClient metaClient = HoodieTableMetaClient.reload(table.getMetaClient());

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/UpgradeDowngrade.java

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -156,6 +156,43 @@ public void run(HoodieTableVersion toVersion, String instantTime) {
156156
return;
157157
}
158158

159+
// Check if any handlers in the upgrade/downgrade path need rollback and perform it once before starting
160+
boolean needsRollback = false;
161+
if (fromVersion.versionCode() < toVersion.versionCode()) {
162+
// Check upgrade handlers
163+
HoodieTableVersion checkVersion = fromVersion;
164+
while (checkVersion.versionCode() < toVersion.versionCode()) {
165+
HoodieTableVersion nextVersion = HoodieTableVersion.fromVersionCode(checkVersion.versionCode() + 1);
166+
UpgradeHandler handler = getUpgradeHandlerInstance(checkVersion, nextVersion);
167+
if (handler.needsRollbackPendingCommitAndCompact()) {
168+
needsRollback = true;
169+
break;
170+
}
171+
checkVersion = nextVersion;
172+
}
173+
} else {
174+
// Check downgrade handlers
175+
HoodieTableVersion checkVersion = fromVersion;
176+
while (checkVersion.versionCode() > toVersion.versionCode()) {
177+
HoodieTableVersion prevVersion = HoodieTableVersion.fromVersionCode(checkVersion.versionCode() - 1);
178+
DowngradeHandler handler = getDowngradeHandlerInstance(checkVersion, prevVersion);
179+
if (handler.needsRollbackPendingCommitAndCompact()) {
180+
needsRollback = true;
181+
break;
182+
}
183+
checkVersion = prevVersion;
184+
}
185+
}
186+
187+
// Perform rollback once if any handler needs it
188+
if (needsRollback) {
189+
LOG.info("Performing rollback and compaction before upgrade/downgrade operations");
190+
HoodieTable table = upgradeDowngradeHelper.getTable(config, context);
191+
UpgradeDowngradeUtils.rollbackFailedWritesAndCompact(table, context, config, upgradeDowngradeHelper,
192+
HoodieTableType.MERGE_ON_READ.equals(metaClient.getTableType()),
193+
metaClient.getTableConfig().getTableVersion());
194+
}
195+
159196
// Perform the actual upgrade/downgrade; this has to be idempotent, for now.
160197
LOG.info("Attempting to move table from version " + fromVersion + " to " + toVersion);
161198
Map<ConfigProperty, String> tablePropsToAdd = new Hashtable<>();
@@ -276,4 +313,52 @@ protected Pair<Map<ConfigProperty, String>, List<ConfigProperty>> downgrade(Hood
276313
throw new HoodieUpgradeDowngradeException(fromVersion.versionCode(), toVersion.versionCode(), false);
277314
}
278315
}
316+
317+
protected UpgradeHandler getUpgradeHandlerInstance(HoodieTableVersion fromVersion, HoodieTableVersion toVersion) {
318+
if (fromVersion == HoodieTableVersion.ZERO && toVersion == HoodieTableVersion.ONE) {
319+
return new ZeroToOneUpgradeHandler();
320+
} else if (fromVersion == HoodieTableVersion.ONE && toVersion == HoodieTableVersion.TWO) {
321+
return new OneToTwoUpgradeHandler();
322+
} else if (fromVersion == HoodieTableVersion.TWO && toVersion == HoodieTableVersion.THREE) {
323+
return new TwoToThreeUpgradeHandler();
324+
} else if (fromVersion == HoodieTableVersion.THREE && toVersion == HoodieTableVersion.FOUR) {
325+
return new ThreeToFourUpgradeHandler();
326+
} else if (fromVersion == HoodieTableVersion.FOUR && toVersion == HoodieTableVersion.FIVE) {
327+
return new FourToFiveUpgradeHandler();
328+
} else if (fromVersion == HoodieTableVersion.FIVE && toVersion == HoodieTableVersion.SIX) {
329+
return new FiveToSixUpgradeHandler();
330+
} else if (fromVersion == HoodieTableVersion.SIX && toVersion == HoodieTableVersion.SEVEN) {
331+
return new SixToSevenUpgradeHandler();
332+
} else if (fromVersion == HoodieTableVersion.SEVEN && toVersion == HoodieTableVersion.EIGHT) {
333+
return new SevenToEightUpgradeHandler();
334+
} else if (fromVersion == HoodieTableVersion.EIGHT && toVersion == HoodieTableVersion.NINE) {
335+
return new EightToNineUpgradeHandler();
336+
} else {
337+
throw new HoodieUpgradeDowngradeException(fromVersion.versionCode(), toVersion.versionCode(), true);
338+
}
339+
}
340+
341+
protected DowngradeHandler getDowngradeHandlerInstance(HoodieTableVersion fromVersion, HoodieTableVersion toVersion) {
342+
if (fromVersion == HoodieTableVersion.ONE && toVersion == HoodieTableVersion.ZERO) {
343+
return new OneToZeroDowngradeHandler();
344+
} else if (fromVersion == HoodieTableVersion.TWO && toVersion == HoodieTableVersion.ONE) {
345+
return new TwoToOneDowngradeHandler();
346+
} else if (fromVersion == HoodieTableVersion.THREE && toVersion == HoodieTableVersion.TWO) {
347+
return new ThreeToTwoDowngradeHandler();
348+
} else if (fromVersion == HoodieTableVersion.FOUR && toVersion == HoodieTableVersion.THREE) {
349+
return new FourToThreeDowngradeHandler();
350+
} else if (fromVersion == HoodieTableVersion.FIVE && toVersion == HoodieTableVersion.FOUR) {
351+
return new FiveToFourDowngradeHandler();
352+
} else if (fromVersion == HoodieTableVersion.SIX && toVersion == HoodieTableVersion.FIVE) {
353+
return new SixToFiveDowngradeHandler();
354+
} else if (fromVersion == HoodieTableVersion.SEVEN && toVersion == HoodieTableVersion.SIX) {
355+
return new SevenToSixDowngradeHandler();
356+
} else if (fromVersion == HoodieTableVersion.EIGHT && toVersion == HoodieTableVersion.SEVEN) {
357+
return new EightToSevenDowngradeHandler();
358+
} else if (fromVersion == HoodieTableVersion.NINE && toVersion == HoodieTableVersion.EIGHT) {
359+
return new NineToEightDowngradeHandler();
360+
} else {
361+
throw new HoodieUpgradeDowngradeException(fromVersion.versionCode(), toVersion.versionCode(), false);
362+
}
363+
}
279364
}

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/UpgradeHandler.java

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,4 +41,14 @@ public interface UpgradeHandler {
4141
Map<ConfigProperty, String> upgrade(
4242
HoodieWriteConfig config, HoodieEngineContext context, String instantTime,
4343
SupportsUpgradeDowngrade upgradeDowngradeHelper);
44+
45+
/**
46+
* Indicates whether this upgrade handler needs rollback of pending commits and compaction
47+
* to be executed before the upgrade logic runs.
48+
*
49+
* @return true if rollback and compaction should be performed before upgrade, false otherwise.
50+
*/
51+
default boolean needsRollbackPendingCommitAndCompact() {
52+
return false;
53+
}
4454
}

0 commit comments

Comments
 (0)