[HUDI-6315] Feature flag for disabling prepped merge. - #9203
Conversation
nsivabalan
left a comment
There was a problem hiding this comment.
Lets also parametrize some tests for MIT as well
|
We can set ENABLE_PREPPED_MERGE_WRITES in MergeIntoHoodieTableCommand using hoodie.spark.sql.writes.optimized.enable. We don't need to introduce another public config that way |
Parameterized test case to test with both
Moved |
|
@hudi-bot run azure |
|
@hudi-bot run azure |
There was a problem hiding this comment.
I realized we have two diff configs across MIT and UPDATE/DELETEs. not a good user exp imo. lets unfiy them.
So, we will have one config "hoodie.spark.sql.optimized.writes.enable" in HoodieWriteConfig for users to enable to disable the optimized flow. default value is true.
But internally, we can use two diff internal configs, one for MIT (_hoodie.datasource.merge.into.prepped) and one for UPDATE and DELETEs (_hoodie.datasource.writes.prepped).
| .withDocumentation("Controls whether spark sql optimized update is enabled.") | ||
| .withDocumentation("Controls whether spark sql prepped update and delete is enabled.") | ||
|
|
||
| val ENABLE_OPTIMIZED_SQL_MERGE_WRITES: ConfigProperty[String] = ConfigProperty |
There was a problem hiding this comment.
I am also thinking, from a user standpoint we should have just 1 config to enable or disable the optimized flow (irrespective of whether its mIT or updates or deletes).
but internally we can use diff configs if we wish to differentiate MIT and rest.
| + "The class must be a subclass of `org.apache.hudi.callback.HoodieClientInitCallback`." | ||
| + "By default, no Hudi client init callback is executed."); | ||
|
|
||
| public static final String WRITE_PREPPED_MERGE_KEY = "_hoodie.datasource.merge.into.prepped"; |
There was a problem hiding this comment.
can we add java docs calling out the purpose of this
There was a problem hiding this comment.
sorry. lets name this
_hoodie.spark.sql.merge.into.prepped
There was a problem hiding this comment.
lets name the variable
SPARK_SQL_MERGE_INTO_PREPPED
| val df = if (optParams.getOrDefault(DATASOURCE_WRITE_PREPPED_KEY, | ||
| optParams.getOrDefault(SQL_MERGE_INTO_WRITES.key(), SQL_MERGE_INTO_WRITES.defaultValue().toString)) | ||
| .equalsIgnoreCase("true")) { | ||
| val df = if (optParams.getOrDefault(DATASOURCE_WRITE_PREPPED_KEY, "false").toBoolean || optParams.getOrDefault(WRITE_PREPPED_MERGE_KEY, "false").toBoolean) { |
There was a problem hiding this comment.
lets also rename the variable
DATASOURCE_WRITE_PREPPED_KEY to
SPARK_SQL_WRITE_PREPPED_KEY
There was a problem hiding this comment.
and the config key be
_hoodie.spark.sql.writes.prepped
| test("Test pkless complex merge cond") { | ||
| withRecordType()(withTempDir { tmp => | ||
| spark.sql("set hoodie.payload.combined.schema.validate = true") | ||
| spark.sql("set hoodie.spark.sql.optimized.merge.enable=true") |
There was a problem hiding this comment.
don't we need to fix all these?
|
@hudi-bot run azure |
| public static HoodieIndex createIndex(HoodieWriteConfig config) { | ||
| boolean mergeIntoWrites = config.getProps().getBoolean(HoodieInternalConfig.SQL_MERGE_INTO_WRITES.key(), | ||
| HoodieInternalConfig.SQL_MERGE_INTO_WRITES.defaultValue()); | ||
| boolean mergeIntoWrites = config.getProps().getBoolean(HoodieWriteConfig.SPARK_SQL_MERGE_INTO_PREPPED_KEY, false); |
There was a problem hiding this comment.
lets align the var name w/ the config.
boolean sqlMergeIntoPrepped
There was a problem hiding this comment.
Replaced all instances of mergeIntoWrites to sqlMergeIntoPrepped (except for the original reference in deduceWriterSchema)
| * Config key with boolean value that indicates whether record being written is already prepped. | ||
| */ | ||
| val DATASOURCE_WRITE_PREPPED_KEY = "_hoodie.datasource.write.prepped"; | ||
| val SPARK_SQL_WRITE_PREPPED_KEY = "_hoodie.spark.sql.writes.prepped"; |
There was a problem hiding this comment.
minor
"SPARK_SQL_WRITES_PREPPED_KEY"
| val mergeIntoWrites = parameters.getOrDefault(SQL_MERGE_INTO_WRITES.key(), | ||
| SQL_MERGE_INTO_WRITES.defaultValue.toString).toBoolean | ||
| val isPrepped = hoodieConfig.getBooleanOrDefault(SPARK_SQL_WRITE_PREPPED_KEY, false) | ||
| val mergeIntoWrites = parameters.getOrDefault(SPARK_SQL_MERGE_INTO_PREPPED_KEY, "false").toBoolean |
| hoodieConfig.setDefaultValue(DROP_PARTITION_COLUMNS) | ||
| hoodieConfig.setDefaultValue(KEYGENERATOR_CONSISTENT_LOGICAL_TIMESTAMP_ENABLED) | ||
| hoodieConfig.setDefaultValue(ENABLE_OPTIMIZED_SQL_WRITES) | ||
| hoodieConfig.setDefaultValue(SPARK_SQL_OPTIMIZED_WRITES) |
There was a problem hiding this comment.
lets check if this is really required.
There was a problem hiding this comment.
Removed. From what I can see it is not really needed.
| .option(DataSourceWriteOptions.ASYNC_CLUSTERING_ENABLE().key(), "true") | ||
| .option(DataSourceWriteOptions.ASYNC_CLUSTERING_ENABLE().key(), "true") | ||
| .option(DataSourceWriteOptions.ENABLE_OPTIMIZED_SQL_WRITES().key(), "true") | ||
| .option(DataSourceWriteOptions.SPARK_SQL_OPTIMIZED_WRITES().key(), "true") |
There was a problem hiding this comment.
for defaults, lets try to use the actual default instead of hard coding it
DataSourceWriteOptions.SPARK_SQL_OPTIMIZED_WRITES().default()
|
|
||
| // test with optimized sql writes enabled / disabled. | ||
| spark.sql(s"set hoodie.spark.sql.writes.optimized.enable=$optimizedSqlEnabled") | ||
| spark.sql(s"set hoodie.spark.sql.optimized.writes.enable=$optimizedSqlEnabled") |
There was a problem hiding this comment.
lets see if we can use the variable to avoid any mis-steps
|
@hudi-bot run azure |
|
CI failed due to known flaky tests. going ahead w/ merging |
Change Logs
Add user-defined feature flag for disabling prepped merge.
Impact
New feature flag
ENABLE_OPTIMIZED_MERGE_WRITESRisk level (write none, low medium or high below)
Low
Documentation Update
Describe any necessary documentation update if there is any new feature, config, or user-facing change
ticket number here and follow the instruction to make
changes to the website.
Contributor's checklist