[
https://issues.apache.org/jira/browse/HUDI-8150?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17890770#comment-17890770
]
Y Ethan Guo edited comment on HUDI-8150 at 10/21/24 10:33 PM:
--------------------------------------------------------------
In Hudi 1.0, we'll maintain the way how the Spark SQL writer and reader paths
work regarding how the custom payload classes are used to fulfill specific DML
logic. Here are more details. HUDI-8203 also needs the following information
to use the merge mode and payload classes properly in the Spark SQL.
On master and existing behavior (before HUDI-8203): we don't write the payload
class name to hoodie.properties if not set explicitly by the user. The default
payload class in 0.x releases is DefaultHoodieRecordPayload.
On the writer side in Spark SQL: there are two cases that are handled
specifically: (1) deducing logic: INSERT INTO automatically deduces the custom
payload class to use in certain cases, see code and details below; (2) MERGE
INTO uses expression payload class (ExpressionPayload) for merge behavior
defined by the SQL. The generated records are directly upserted to the table,
and the reader honors the existing payload class for record merging. In other
cases, the writer path uses the payload classes directly if record merge is
required.
{code:java}
val payloadClassName = if (useLegacyInsertDropDupFlow) {
deducePayloadClassNameLegacy(operation, tableType, insertMode)
} else {
if (insertDupPolicy == FAIL_INSERT_DUP_POLICY) {
classOf[ValidateDuplicateKeyPayload].getCanonicalName
} else {
classOf[DefaultHoodieRecordPayload].getCanonicalName
}
}private def deducePayloadClassNameLegacy(operation: String, tableType: String,
insertMode: InsertMode): String = {
if (operation == UPSERT_OPERATION_OPT_VAL &&
tableType == COW_TABLE_TYPE_OPT_VAL && insertMode == InsertMode.STRICT) {
// Validate duplicate key for COW, for MOR it will do the merge with the
DefaultHoodieRecordPayload
// on reading.
// TODO use HoodieSparkValidateDuplicateKeyRecordMerger when
SparkRecordMerger is default
classOf[ValidateDuplicateKeyPayload].getCanonicalName
} else if (operation == INSERT_OPERATION_OPT_VAL && tableType ==
COW_TABLE_TYPE_OPT_VAL &&
insertMode == InsertMode.STRICT) {
// Validate duplicate key for inserts to COW table when using strict insert
mode.
classOf[ValidateDuplicateKeyPayload].getCanonicalName
} else {
classOf[DefaultHoodieRecordPayload].getCanonicalName
}
} {code}
- payload class set (user sets it) in hoodie.properties -> Spark SQL writer
uses it and skips payload deduction logic for INSERT INTO. This means that
INSERT INTO uses user specified payload class. If the payload class is not
explicitly set (using default payload), the payload deduction logic takes
effect for INSERT INTO. MERGE INTO always uses ExpressionPayload.
- payload class not set (using default) in hoodie.properties -> Spark SQL
deduces the payload class (can be different from the default payload class) to
use. MERGE INTO always uses ExpressionPayload regardless.
On the reader side in Spark SQL: the file group reader-based implementation
based on merger classes or custom payload class is used (see HUDI-7678). Any
custom payload classes mentioned above (e.g., payload class for dedup,
expression payload class) are never used on the reader side.
We'll still maintain the above merging behavior for Spark SQL. With HUDI-8203
introducing the merge mode to rule the merging behavior, we're writing the
merge mode to the table config. There are some nuances to consider here around
whether to write the inferred payload class based on the merge mode, for SQL
writer merge behavior:
- (fine with master and the HUDI-8203 PR) MIT should always use the expression
payload, because MIT statement dictates the update/merge behavior.
- (problem to solve for HUDI-82303) INSERT INTO has some complex logic on
deducing internal payload to use
-- master branch follows this
-- In the HUDI-8203 PR, there is currently a conflict between payload config
in table config vs the deducted payload, causing behavior change
- (fine with master and the PR) All other DMLs use custom payload available in
the table config.
In the new behavior / merge mode PR (HUDI-8203):
- Always merge mode config into hoodie.properties
Proposal 1
- Maintain the behavior, where we don't write the payload class name to
hoodie.properties if not set explicitly.
- Always set merge mode table config (CUSTOM: custom payload class is set
always; forOVERWRITE_WITH_LATEST and EVENT_TIME_ORDERING we can infer the
payload class from merge mode, i.e., we know the relationship between merge
mode and payload class)
- In table config,
- For Spark SQL, do not add the payload class name when loading the
HoodieTableConfig
- For other places, still attach the payload from the Table Configs
Drawback: this adds more bandaid just for Spark SQL to get it working.
Ideally, the table config should contain all the merge mode/payload classes to
dictate the record merge behavior, which should not be changed once the table
is created, unless the user wants to do one-time change or migration. Adding
bandaids around table config persistence and merging behavior just for Spark
SQL makes it hard to reason going forward.
Proposal 2 (go with this)
- (Prefer this) Always write the payload class config in hoodie.properties
(implemented in the current HUDI-8203 PR)
- For Spark SQL INSERT INTO,
- for CUSTOM payload, always use it which is the current behavior
- (change needed) if it's default hoodie payload, use deduction logic without
honoring the table config.
Proposal 2 streamlined the table configs to be easy to understand; while having
logic for Spark SQL write only for the cases where default hoodie payload is
used, so the bandaid scope is much smaller, and easier to deprecate in the
future.
was (Author: JIRAUSER280684):
In Hudi 1.0, we'll maintain the way how the Spark SQL writer and reader paths
work regarding how the custom payload classes are used to fulfill specific DML
logic. Here are more details. HUDI-8203 also needs the following information
to use the merge mode and payload classes properly in the Spark SQL.
On master and existing behavior (before HUDI-8203): we don't write the payload
class name to hoodie.properties if not set explicitly by the user. The default
payload class in 0.x releases is DefaultHoodieRecordPayload.
On the writer side in Spark SQL: there are two cases that are handled
specifically: (1) deducing logic: INSERT INTO automatically deduces the custom
payload class to use in certain cases, see code and details below; (2) MERGE
INTO uses expression payload class (ExpressionPayload) for merge behavior
defined by the SQL. The generated records are directly upserted to the table,
and the reader honors the existing payload class for record merging. In other
cases, the writer path uses the payload classes directly if record merge is
required.
{code:java}
val payloadClassName = if (useLegacyInsertDropDupFlow) {
deducePayloadClassNameLegacy(operation, tableType, insertMode)
} else {
if (insertDupPolicy == FAIL_INSERT_DUP_POLICY) {
classOf[ValidateDuplicateKeyPayload].getCanonicalName
} else {
classOf[DefaultHoodieRecordPayload].getCanonicalName
}
}private def deducePayloadClassNameLegacy(operation: String, tableType: String,
insertMode: InsertMode): String = {
if (operation == UPSERT_OPERATION_OPT_VAL &&
tableType == COW_TABLE_TYPE_OPT_VAL && insertMode == InsertMode.STRICT) {
// Validate duplicate key for COW, for MOR it will do the merge with the
DefaultHoodieRecordPayload
// on reading.
// TODO use HoodieSparkValidateDuplicateKeyRecordMerger when
SparkRecordMerger is default
classOf[ValidateDuplicateKeyPayload].getCanonicalName
} else if (operation == INSERT_OPERATION_OPT_VAL && tableType ==
COW_TABLE_TYPE_OPT_VAL &&
insertMode == InsertMode.STRICT) {
// Validate duplicate key for inserts to COW table when using strict insert
mode.
classOf[ValidateDuplicateKeyPayload].getCanonicalName
} else {
classOf[DefaultHoodieRecordPayload].getCanonicalName
}
} {code}
- payload class set (user sets it) in hoodie.properties -> Spark SQL writer
uses it and skips payload deduction logic for INSERT INTO. This means that
INSERT INTO uses user specified payload class. If the payload class is not
explicitly set (using default payload), the payload deduction logic takes
effect for INSERT INTO. MERGE INTO always uses
- payload class not set (using default) in hoodie.properties -> Spark SQL
deducts the payload class (can be different from default) to use
On the reader side in Spark SQL:
HUDI-8203 SQL writer merge behavior:
- (fine with master and the PR) MIT should always use the expression payload,
because MIT statement dictates the update/merge behavior.
- (problem to solve) INSERT INTO has some complex logic on determining
internal payload to use
- master branch follows this
- In the PR, there is currently a conflict between payload config in table
config vs the deducted payload, causing behavior change
- (fine with master and the PR) All other DMLs use custom payload avaiable.
In the new behavior / merge mode PR:
- Always merge mode config into hoodie.properties
Proposal 1
- Maintain the behavior, where we don't write the payload class name to
hoodie.properties if not set explicitly.
- always set merge mode table config others (CUSTOM: custom payload class is
set always, and OVERWRITE_WITH_LATEST and EVENT_TIME_ORDERING we can infer from
merge mode, i.e., we know the relationship between merge mode and payload class)
- In table config,
- For Spark SQL, do not add the payload class name when loading the
HoodieTableConfig
- For other places, still attach the payload from the Table Configs
Proposal 2 (go with this)
- (Prefer this) Always write the payload class config in hoodie.properties
(what in the current PR)
- For Spark SQL INSERT INTO,
- for CUSTOM payload, always use it which is the current behavior
- (change needed) if it's default hoodie payload, use deduction logic without
honoring the table config
> Review semantics of SQL and RecordMerger interplay
> --------------------------------------------------
>
> Key: HUDI-8150
> URL: https://issues.apache.org/jira/browse/HUDI-8150
> Project: Apache Hudi
> Issue Type: New Feature
> Reporter: Ethan Guo (this is the old account; please use "yihua")
> Assignee: Y Ethan Guo
> Priority: Blocker
> Fix For: 1.0.0
>
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
