Skip to content

Commit cfcc9aa

Browse files
linliu-codelinliu-code
committed
update the logic
1 parent 5ecf00a commit cfcc9aa

File tree

4 files changed

+90
-69
lines changed

4 files changed

+90
-69
lines changed

‎hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java

Lines changed: 8 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1060,7 +1060,6 @@ public static class TableBuilder {
10601060
private String tableName;
10611061
private String tableCreateSchema;
10621062
private HoodieTableVersion tableVersion;
1063-
private boolean autoUpgrade = true;
10641063
private String recordKeyFields;
10651064
private String secondaryKeyFields;
10661065
private String archiveLogFolder;
@@ -1129,11 +1128,6 @@ public TableBuilder setTableVersion(HoodieTableVersion tableVersion) {
11291128
return this;
11301129
}
11311130

1132-
public TableBuilder setAutoUpgrade(boolean autoUpgrade) {
1133-
this.autoUpgrade = autoUpgrade;
1134-
return this;
1135-
}
1136-
11371131
public TableBuilder setTimelinePath(String timelinePath) {
11381132
this.timelinePath = timelinePath;
11391133
return this;
@@ -1596,21 +1590,16 @@ public Properties build() {
15961590
}
15971591

15981592
HoodieTableVersion getTableVersionProperly() {
1599-
if (tableVersion == null
1600-
|| tableVersion.greaterThanOrEquals(HoodieTableVersion.current())) {
1593+
if (tableVersion == null) {
16011594
return HoodieTableVersion.current();
1602-
} else {
1603-
if (autoUpgrade && (tableVersion.versionCode() == 6 || tableVersion.versionCode() == 8)) {
1604-
String errorMessage = String.format(
1605-
"Table version \"%d\" is set for table creation, which is lower than "
1606-
+ "the highest available version \"%d\" the current Hudi binary supports. Meanwhile, "
1607-
+ "the auto upgrade configuration \"hoodie.write.auto.upgrade\" is enabled. A Potential risk is that "
1608-
+ "future commits might upgrade this table without any awareness. "
1609-
+ "To prevent this, please either set \"hoodie.write.auto.upgrade=false\" or unset the table version.",
1610-
tableVersion.versionCode(), HoodieTableVersion.current().versionCode());
1611-
throw new HoodieNotSupportedException(errorMessage);
1612-
}
1595+
} else if (tableVersion.greaterThanOrEquals(HoodieTableVersion.SIX)) {
16131596
return tableVersion;
1597+
} else {
1598+
String errorMessage = String.format(
1599+
"Creating a table in version \"%d\" is not allowed in current Hudi binary. "
1600+
+ "Please set \"hoodie.write.table.version\" config with a value >= 6",
1601+
tableVersion.versionCode());
1602+
throw new HoodieNotSupportedException(errorMessage);
16141603
}
16151604
}
16161605

‎hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java

Lines changed: 15 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -150,7 +150,6 @@ void testTableVersion() throws IOException {
150150
.setTableType(HoodieTableType.MERGE_ON_READ.name())
151151
.setTableName("table-version-test")
152152
.setTableVersion(HoodieTableVersion.SIX.versionCode())
153-
.setAutoUpgrade(false)
154153
.initTable(this.metaClient.getStorageConf(), basePath);
155154
assertEquals(HoodieTableVersion.SIX, metaClient1.getTableConfig().getTableVersion());
156155

@@ -170,12 +169,10 @@ void testGenerateFromAnotherMetaClient() throws IOException {
170169
.setTableType(HoodieTableType.MERGE_ON_READ.name())
171170
.setTableName("table-version-test")
172171
.setTableVersion(HoodieTableVersion.SIX.versionCode())
173-
.setAutoUpgrade(false)
174172
.initTable(this.metaClient.getStorageConf(), basePath1);
175173

176174
HoodieTableMetaClient metaClient2 = HoodieTableMetaClient.newTableBuilder()
177175
.fromMetaClient(metaClient1)
178-
.setAutoUpgrade(false)
179176
.initTable(this.metaClient.getStorageConf(), basePath2);
180177

181178
assertEquals(metaClient1.getTableConfig().getTableType(), metaClient2.getTableConfig().getTableType());
@@ -292,8 +289,7 @@ void testDeleteDefinition() throws IOException {
292289
void testGetTableVersionProperlyWithNullTableVersion() {
293290
HoodieTableMetaClient.TableBuilder tableBuilder = HoodieTableMetaClient.newTableBuilder()
294291
.setTableName("test_table")
295-
.setTableType(HoodieTableType.COPY_ON_WRITE)
296-
.setAutoUpgrade(false);
292+
.setTableType(HoodieTableType.COPY_ON_WRITE);
297293

298294
// tableVersion is null by default
299295
HoodieTableVersion result = tableBuilder.getTableVersionProperly();
@@ -305,82 +301,54 @@ void testGetTableVersionProperlyWithCurrentVersion() {
305301
HoodieTableMetaClient.TableBuilder tableBuilder = HoodieTableMetaClient.newTableBuilder()
306302
.setTableName("test_table")
307303
.setTableType(HoodieTableType.COPY_ON_WRITE)
308-
.setTableVersion(HoodieTableVersion.current())
309-
.setAutoUpgrade(false);
310-
HoodieTableVersion result = tableBuilder.getTableVersionProperly();
311-
assertEquals(HoodieTableVersion.current(), result);
312-
}
313-
314-
@Test
315-
void testGetTableVersionProperlyWithHigherVersion() {
316-
HoodieTableMetaClient.TableBuilder tableBuilder = HoodieTableMetaClient.newTableBuilder()
317-
.setTableName("test_table")
318-
.setTableType(HoodieTableType.COPY_ON_WRITE)
319-
.setTableVersion(HoodieTableVersion.NINE)
320-
.setAutoUpgrade(false);
304+
.setTableVersion(HoodieTableVersion.current());
321305
HoodieTableVersion result = tableBuilder.getTableVersionProperly();
322306
assertEquals(HoodieTableVersion.current(), result);
323307
}
324308

325309
@Test
326-
void testGetTableVersionProperlyWithVariousLowerVersions() {
327-
// Test with different lower versions
310+
void testGetTableVersionProperlyWithValidVersions() {
328311
HoodieTableVersion[] lowerVersions = {
329-
HoodieTableVersion.ZERO,
330-
HoodieTableVersion.ONE,
331-
HoodieTableVersion.TWO,
332-
HoodieTableVersion.THREE,
333-
HoodieTableVersion.FOUR,
334-
HoodieTableVersion.FIVE,
335312
HoodieTableVersion.SIX,
336313
HoodieTableVersion.SEVEN,
337-
HoodieTableVersion.EIGHT
314+
HoodieTableVersion.EIGHT,
315+
HoodieTableVersion.NINE
338316
};
339317

340318
for (HoodieTableVersion version : lowerVersions) {
341319
HoodieTableMetaClient.TableBuilder tableBuilder = HoodieTableMetaClient.newTableBuilder()
342320
.setTableName("test_table")
343321
.setTableType(HoodieTableType.COPY_ON_WRITE)
344-
.setTableVersion(version)
345-
.setAutoUpgrade(false);
322+
.setTableVersion(version);
346323

347324
HoodieTableVersion result = tableBuilder.getTableVersionProperly();
348325
assertEquals(version, result);
349326
}
350327
}
351328

352329
@Test
353-
void testGetTableVersionProperlyWithLowerVersionsAndAutoUpgradeTrue() {
330+
void testGetTableVersionProperlyWithInvalidValues() {
354331
// Test with different lower versions and autoUpgrade=true
355332
HoodieTableVersion[] lowerVersions = {
356333
HoodieTableVersion.ZERO,
357334
HoodieTableVersion.ONE,
358335
HoodieTableVersion.TWO,
359336
HoodieTableVersion.THREE,
360337
HoodieTableVersion.FOUR,
361-
HoodieTableVersion.FIVE,
362-
HoodieTableVersion.SIX,
363-
HoodieTableVersion.SEVEN,
364-
HoodieTableVersion.EIGHT
338+
HoodieTableVersion.FIVE
365339
};
366340

367341
for (HoodieTableVersion version : lowerVersions) {
368342
HoodieTableMetaClient.TableBuilder tableBuilder = HoodieTableMetaClient.newTableBuilder()
369343
.setTableName("test_table")
370344
.setTableType(HoodieTableType.COPY_ON_WRITE)
371-
.setTableVersion(version)
372-
.setAutoUpgrade(true);
373-
374-
if (version.versionCode() == 6 || version.versionCode() == 8) {
375-
HoodieNotSupportedException exception = assertThrows(HoodieNotSupportedException.class,
376-
() -> tableBuilder.getTableVersionProperly());
377-
// Verify the error message contains the expected version information
378-
assertTrue(exception.getMessage().contains(
379-
String.format("Table version \"%d\"", version.versionCode())));
380-
} else {
381-
HoodieTableVersion result = tableBuilder.getTableVersionProperly();
382-
assertEquals(version, result);
383-
}
345+
.setTableVersion(version);
346+
347+
HoodieNotSupportedException exception = assertThrows(
348+
HoodieNotSupportedException.class,
349+
() -> tableBuilder.getTableVersionProperly());
350+
assertTrue(exception.getMessage().contains(
351+
String.format("Creating a table in version \"%d\"", version.versionCode())));
384352
}
385353
}
386354
}

‎hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -300,12 +300,9 @@ class HoodieSparkSqlWriterInternal {
300300
hoodieConfig.getString(DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME)
301301
else KeyGeneratorType.getKeyGeneratorClassName(hoodieConfig)
302302
val tableFormat = hoodieConfig.getStringOrDefault(HoodieTableConfig.TABLE_FORMAT)
303-
val autoUpgrade = hoodieConfig.getBooleanOrDefault(
304-
HoodieWriteConfig.AUTO_UPGRADE_VERSION, HoodieWriteConfig.AUTO_UPGRADE_VERSION.defaultValue())
305303
HoodieTableMetaClient.newTableBuilder()
306304
.setTableType(tableType)
307305
.setTableVersion(tableVersion)
308-
.setAutoUpgrade(autoUpgrade)
309306
.setTableFormat(tableFormat)
310307
.setDatabaseName(databaseName)
311308
.setTableName(tblName)

‎hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,9 +27,11 @@
2727
import org.apache.hudi.common.table.timeline.HoodieTimeline;
2828
import org.apache.hudi.common.table.timeline.InstantFileNameGenerator;
2929
import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
30+
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
3031
import org.apache.hudi.common.testutils.HoodieTestUtils;
3132
import org.apache.hudi.common.util.Option;
3233
import org.apache.hudi.config.HoodieWriteConfig;
34+
import org.apache.hudi.exception.HoodieNotSupportedException;
3335
import org.apache.hudi.exception.HoodieUpgradeDowngradeException;
3436
import org.apache.hudi.metadata.HoodieTableMetadata;
3537
import org.apache.hudi.storage.StoragePath;
@@ -38,14 +40,19 @@
3840
import org.apache.spark.sql.Row;
3941

4042
import java.util.Arrays;
43+
import java.util.HashMap;
4144
import java.util.HashSet;
4245
import java.util.List;
46+
import java.util.Map;
4347
import java.util.Set;
4448
import java.util.stream.Collectors;
4549

50+
import org.apache.spark.sql.SaveMode;
51+
import org.apache.spark.sql.SparkSession;
4652
import org.junit.jupiter.api.io.TempDir;
4753
import org.junit.jupiter.params.ParameterizedTest;
4854
import org.junit.jupiter.params.provider.Arguments;
55+
import org.junit.jupiter.params.provider.CsvSource;
4956
import org.junit.jupiter.params.provider.MethodSource;
5057
import org.slf4j.Logger;
5158
import org.slf4j.LoggerFactory;
@@ -55,6 +62,10 @@
5562
import java.util.Properties;
5663
import java.util.stream.Stream;
5764

65+
import scala.collection.JavaConverters;
66+
import scala.collection.Seq;
67+
68+
import static org.apache.hudi.common.testutils.RawTripTestPayload.recordsToStrings;
5869
import static org.junit.jupiter.api.Assertions.assertEquals;
5970
import static org.junit.jupiter.api.Assertions.assertNotNull;
6071
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -203,6 +214,62 @@ public void testMetadataTableUpgradeDowngradeFailure(HoodieTableVersion fromVers
203214
"Exception message should contain metadata table failure message");
204215
}
205216

217+
@ParameterizedTest
218+
@CsvSource({"1","2","3","4","5","6","7","8","9","null"})
219+
void testTableVersionDuringTableCreation(String targetTableVersion) {
220+
SparkSession spark = spark();
221+
Map<String, String> writeOptions = new HashMap<>();
222+
if (!targetTableVersion.equals("null")) {
223+
writeOptions.put(HoodieWriteConfig.WRITE_TABLE_VERSION.key(), targetTableVersion);
224+
}
225+
writeOptions.put(HoodieWriteConfig.TBL_NAME.key(), "testTableCreation");
226+
HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator();
227+
List<String> records = recordsToStrings(dataGen.generateInserts("001", 5));
228+
Seq<String> scalaRecords = JavaConverters.collectionAsScalaIterableConverter(records).asScala().toSeq();
229+
Dataset<Row> inputDF = spark.read().json(spark.sparkContext().parallelize(
230+
(Seq<String>)scalaRecords, 2, scala.reflect.ClassTag$.MODULE$.apply(String.class)));
231+
232+
// Create table, and validate.
233+
Set<String> failSet = new HashSet<>(Arrays.asList("1", "2", "3", "4", "5"));
234+
if (failSet.contains(targetTableVersion)) {
235+
// This fails at HoodieTableMetaClient.
236+
assertThrows(HoodieNotSupportedException.class, () -> {
237+
inputDF.write().format("org.apache.hudi")
238+
.partitionBy("partition")
239+
.options(writeOptions)
240+
.mode(SaveMode.Append)
241+
.save(basePath());
242+
});
243+
} else if (targetTableVersion.equals("7")) {
244+
// Fails at write table version check.
245+
assertThrows(IllegalArgumentException.class, () -> {
246+
inputDF.write().format("org.apache.hudi")
247+
.partitionBy("partition")
248+
.options(writeOptions)
249+
.mode(SaveMode.Append)
250+
.save(basePath());
251+
});
252+
} else {
253+
inputDF.write().format("org.apache.hudi")
254+
.partitionBy("partition")
255+
.options(writeOptions)
256+
.mode(SaveMode.Append)
257+
.save(basePath());
258+
metaClient = HoodieTableMetaClient.builder().setConf(storageConf()).setBasePath(basePath()).build();
259+
// If no write version is specified, use current.
260+
if (!targetTableVersion.equals("null")) {
261+
assertEquals(
262+
Integer.valueOf(targetTableVersion),
263+
metaClient.getTableConfig().getTableVersion().versionCode());
264+
} else {
265+
// Otherwise, the table version is the target table version.
266+
assertEquals(
267+
HoodieTableVersion.current(),
268+
metaClient.getTableConfig().getTableVersion());
269+
}
270+
}
271+
}
272+
206273
/**
207274
* Load a fixture table from resources and copy it to a temporary location for testing.
208275
*/

0 commit comments

Comments
 (0)