cshuo commented on code in PR #19307:
URL: https://github.com/apache/hudi/pull/19307#discussion_r3611875897
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/TenToNineDowngradeHandler.java:
##########
@@ -37,6 +38,8 @@ public UpgradeDowngrade.TableConfigChangeSet downgrade(
SupportsUpgradeDowngrade upgradeDowngradeHelper) {
return new UpgradeDowngrade.TableConfigChangeSet(
Collections.emptyMap(),
- Collections.singleton(HoodieTableConfig.TABLE_STORAGE_LAYOUT));
+ CollectionUtils.createImmutableSet(
+ HoodieTableConfig.TABLE_STORAGE_LAYOUT,
+ HoodieTableConfig.LOG_FILE_FORMAT));
Review Comment:
remove log format.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java:
##########
@@ -164,12 +164,24 @@ public static boolean isCowTable(Configuration conf) {
}
/**
- * Returns whether the table uses LSM tree storage layout.
+ * Returns the configured table storage layout.
+ *
+ * <p>Insert and bulk insert operations preserve duplicate record keys and
therefore default to
+ * the regular storage layout. Other operations use Flink's LSM tree default.
*/
- public static boolean isLsmTreeStorageLayout(Configuration conf) {
+ public static HoodieTableConfig.TableStorageLayout
getTableStorageLayout(Configuration conf) {
return HoodieTableConfig.TableStorageLayout.fromConfigValue(conf.getString(
HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(),
- HoodieTableConfig.TABLE_STORAGE_LAYOUT.defaultValue())) == LSM_TREE;
+ (isInsertOperation(conf) || isBulkInsertOperation(conf))
Review Comment:
todo: add key sort for bulk insert with lsm layout.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java:
##########
@@ -359,12 +368,26 @@ public static HoodieTableMetaClient initTableIfNotExists(
basePath, conf.get(FlinkOptions.TABLE_NAME));
}
- return StreamerUtil.createMetaClient(conf, hadoopConf);
+ HoodieTableMetaClient metaClient = StreamerUtil.createMetaClient(conf,
hadoopConf);
+ validateInsertOperationStorageLayout(conf,
metaClient.getTableConfig().getTableStorageLayout());
+ // The persisted layout is authoritative for existing tables. For newly
created tables this
+ // also propagates Flink's LSM default to write-side components that read
the runtime config.
+ conf.setString(HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(),
Review Comment:
move to HoodieTableSink
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/TenToNineDowngradeHandler.java:
##########
@@ -37,6 +38,8 @@ public UpgradeDowngrade.TableConfigChangeSet downgrade(
SupportsUpgradeDowngrade upgradeDowngradeHelper) {
return new UpgradeDowngrade.TableConfigChangeSet(
Collections.emptyMap(),
- Collections.singleton(HoodieTableConfig.TABLE_STORAGE_LAYOUT));
+ CollectionUtils.createImmutableSet(
+ HoodieTableConfig.TABLE_STORAGE_LAYOUT,
+ HoodieTableConfig.LOG_FILE_FORMAT));
Review Comment:
add doc to the config.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java:
##########
@@ -328,14 +328,23 @@ public static HoodieTableMetaClient initTableIfNotExists(
Configuration conf,
org.apache.hadoop.conf.Configuration hadoopConf) throws IOException {
final String basePath = conf.get(FlinkOptions.PATH);
- if (!tableExists(basePath, hadoopConf)) {
+ final boolean tableExists = tableExists(basePath, hadoopConf);
+ if (!tableExists) {
+ HoodieTableConfig.TableStorageLayout storageLayout =
OptionsResolver.getTableStorageLayout(conf);
+ String baseFileFormat = conf.getString(
+ HoodieTableConfig.BASE_FILE_FORMAT.key(),
+ HoodieTableConfig.BASE_FILE_FORMAT.defaultValue().name());
+ validateInsertOperationStorageLayout(conf, storageLayout);
HoodieTableMetaClient.newTableBuilder()
.setTableCreateSchema(conf.get(FlinkOptions.SOURCE_AVRO_SCHEMA))
.setTableType(conf.get(FlinkOptions.TABLE_TYPE))
.setTableName(conf.get(FlinkOptions.TABLE_NAME))
.setTableVersion(conf.get(FlinkOptions.WRITE_TABLE_VERSION))
.setTableFormat(conf.get(FlinkOptions.WRITE_TABLE_FORMAT))
.setBaseFileFormat(conf.getString(HoodieTableConfig.BASE_FILE_FORMAT.key(),
null))
+
.setLogFileFormat(HoodieTableVersion.fromVersionCode(conf.get(FlinkOptions.WRITE_TABLE_VERSION))
Review Comment:
remove
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/lsm/LsmFileGroupRecordIterator.java:
##########
@@ -314,32 +315,85 @@ private ClosableIterator<BufferedRecord<T>>
createFileIterator(StoragePathInfo p
if (FSUtils.isNativeDeleteLogFile(storagePath.getName())) {
return createNativeDeleteLogIterator(pathInfo, storagePath, fileSize);
}
- Pair<HoodieSchema, Map<String, String>> requiredSchemaAndRenamedColumns =
-
readerContext.getSchemaHandler().getRequiredSchemaForFileAndRenamedColumns(storagePath);
- HoodieSchema fileRequiredSchema =
requiredSchemaAndRenamedColumns.getLeft();
- ClosableIterator<T> recordIterator;
+ return FSUtils.isLogFile(storagePath)
+ ? createLogFileIterator(pathInfo, storagePath, fileSize)
+ : createBaseFileIterator(pathInfo, storagePath, fileSize);
+ }
+
+ /**
+ * Reads a base file using the engine's schema-evolution support, matching
+ * {@code HoodieFileGroupReader#makeBaseFileIterator}.
+ */
+ private ClosableIterator<BufferedRecord<T>>
createBaseFileIterator(StoragePathInfo pathInfo,
+
StoragePath storagePath,
+ long
fileSize) throws IOException {
+ ClosableIterator<T> recordIterator = getFileRecordIterator(
+ pathInfo,
+ storagePath,
+ fileSize,
+ readerContext.getSchemaHandler().getTableSchema(),
+ readerSchema);
+ return toBufferedRecordIterator(recordIterator, readerSchema);
+ }
+
+ /**
+ * Reads a native data log with the same schema flow as an inline log block.
+ *
+ * <p>With schema evolution enabled, records are decoded with the writer
schema stored in the
+ * native log footer and then transformed once to the evolved schema for
that instant. Without
+ * schema evolution, the file reader projects directly to the required
reader schema.
+ */
+ private ClosableIterator<BufferedRecord<T>>
createLogFileIterator(StoragePathInfo pathInfo,
Review Comment:
todo: unify schema evolution for base and log file.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java:
##########
@@ -1126,6 +1128,16 @@ public TableBuilder setBaseFileFormat(String
baseFileFormat) {
return this;
}
+ public TableBuilder setLogFileFormat(String logFileFormat) {
Review Comment:
remove
--
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]
