hudi-agent commented on code in PR #19390:
URL: https://github.com/apache/hudi/pull/19390#discussion_r3677209304
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java:
##########
@@ -166,13 +166,13 @@ public static boolean isCowTable(Configuration conf) {
/**
* 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.
+ * <p>Insert operations preserve duplicate record keys and therefore default
to the regular
+ * storage layout. Other operations use Flink's LSM tree default.
*/
public static HoodieTableConfig.TableStorageLayout
getTableStorageLayout(Configuration conf) {
return HoodieTableConfig.TableStorageLayout.fromConfigValue(conf.getString(
HoodieTableConfig.TABLE_STORAGE_LAYOUT.key(),
- (isInsertOperation(conf) || isBulkInsertOperation(conf))
+ isInsertOperation(conf)
Review Comment:
🤖 Bulk insert now defaults to LSM, but the LSM reader collapses all versions
of a key within a sorted run (`LsmFileGroupRecordIterator#nextMergedRecord`
pops consecutive same-key records and merges them), so duplicate keys that bulk
insert writes into a base file get merged into one on snapshot read.
`checkStorageLayout` still blocks `insert` "because they allow duplicate record
keys" — bulk insert has the same property. @danny0405 is silently collapsing
bulk-insert duplicates on read the intended behavior? The new test only checks
the sort row, not an end-to-end read.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/WriterHelpers.java:
##########
@@ -32,8 +33,18 @@
public class WriterHelpers {
public static BulkInsertWriterHelper getWriterHelper(Configuration conf,
HoodieTable<?, ?, ?, ?> hoodieTable, HoodieWriteConfig writeConfig,
String instantTime, int
taskPartitionId, long taskId, long taskEpochId, RowType rowType) {
- return OptionsResolver.isBucketIndexType(conf)
- ? new BucketBulkInsertWriterHelper(conf, hoodieTable, writeConfig,
instantTime, taskPartitionId, taskId, taskEpochId, rowType)
- : new BulkInsertWriterHelper(conf, hoodieTable, writeConfig,
instantTime, taskPartitionId, taskId, taskEpochId, rowType);
+ if (OptionsResolver.isLsmTreeStorageLayout(conf)) {
+ return OptionsResolver.isBucketIndexType(conf)
+ ? new LsmBucketBulkInsertWriterHelper(
+ conf, hoodieTable, writeConfig, instantTime, taskPartitionId,
taskId, taskEpochId, rowType)
+ : new LsmBulkInsertWriterHelper(
+ conf, hoodieTable, writeConfig, instantTime, taskPartitionId,
taskId, taskEpochId, rowType);
+ } else {
+ return OptionsResolver.isBucketIndexType(conf)
+ ? new BucketBulkInsertWriterHelper(
+ conf, hoodieTable, writeConfig, instantTime, taskPartitionId,
taskId, taskEpochId, rowType)
Review Comment:
🤖 nit: the constructor arguments for `BucketBulkInsertWriterHelper` are
indented at the `?` level rather than four spaces past it, unlike every other
constructor in this block — could you align it with the
`LsmBucketBulkInsertWriterHelper` args above (i.e. add four more spaces)?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
