yihua commented on code in PR #19304:
URL: https://github.com/apache/hudi/pull/19304#discussion_r4044020409
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/KeyGenUtils.java:
##########
@@ -436,15 +460,185 @@ public static String getComplexKeygenErrorMessage(String
operation) {
+ "`hoodie.write.complex.keygen.validation.enable=false` to skip this
validation.";
}
+ /**
+ * Whether a complex key generator with a single record key field prepends
the field name to the key.
+ *
+ * <p>{@link HoodieTableConfig#COMPLEX_KEYGEN_ENCODING} wins whenever it is
present in the properties, because
+ * it describes what the table's data carries. Otherwise the write table
version decides: 9 and above always
+ * prefix, 8 and below follow {@code
hoodie.write.complex.keygen.new.encoding}.
+ */
public static boolean
encodeSingleKeyFieldNameForComplexKeyGen(TypedProperties props) {
+ String tableEncoding =
props.getProperty(HoodieTableConfig.COMPLEX_KEYGEN_ENCODING.key());
+ if (!StringUtils.isNullOrEmpty(tableEncoding)) {
+ return
ComplexKeyGenEncoding.fromString(tableEncoding).encodesFieldName();
+ }
int tableVersionCode = ConfigUtils.getIntWithAltKeys(props,
WRITE_TABLE_VERSION);
HoodieTableVersion tableVersion =
HoodieTableVersion.fromVersionCode(tableVersionCode);
return tableVersion.greaterThanOrEquals(HoodieTableVersion.NINE)
|| !ConfigUtils.getBooleanWithAltKeys(props,
COMPLEX_KEYGEN_NEW_ENCODING);
}
- public static boolean mayUseNewEncodingForComplexKeyGen(HoodieTableConfig
tableConfig) {
- return tableConfig.getTableVersion().lesserThan(HoodieTableVersion.NINE)
- && isComplexKeyGeneratorWithSingleRecordKeyField(tableConfig);
+ /**
+ * Whether the table's record key encoding is tracked by {@link
HoodieTableConfig#COMPLEX_KEYGEN_ENCODING}:
+ * a complex key generator with a single record key field and a populated
{@code _hoodie_record_key}.
+ * Without the meta field there is no stored key whose encoding could
diverge from the key generator's.
+ */
+ public static boolean isComplexKeyGenEncodingTracked(HoodieTableConfig
tableConfig) {
+ return tableConfig.isComplexKeyGenWithSingleRecordKeyField() &&
tableConfig.isRecordKeyPopulated();
+ }
+
+ /**
+ * Resolves the record key encoding of a single-field complex key generator
table for readers, without
+ * modifying the table: the persisted property, otherwise the encoding
deduced from the table's data.
+ *
+ * @return empty when the table does not use a single-field complex key
generator, or when its encoding
+ * cannot be determined, in which case record-key based file pruning must be
skipped
+ */
+ public static Option<ComplexKeyGenEncoding>
resolveComplexKeyGenEncoding(HoodieTableMetaClient metaClient) {
+ HoodieTableConfig tableConfig = metaClient.getTableConfig();
+ if (!tableConfig.isComplexKeyGenWithSingleRecordKeyField()) {
+ return Option.empty();
+ }
+ if (!tableConfig.isRecordKeyPopulated()) {
+ // the keys are regenerated by the key generator, which follows the
table version
+ return
tableConfig.getTableVersion().greaterThanOrEquals(HoodieTableVersion.NINE)
+ ? Option.of(ComplexKeyGenEncoding.FIELD_PREFIXED) : Option.empty();
+ }
+ Option<ComplexKeyGenEncoding> persisted =
tableConfig.getComplexKeyGenEncoding();
+ return persisted.isPresent() ? persisted :
deduceComplexKeyGenEncodingFromData(metaClient);
+ }
+
+ /**
+ * Resolves the record key encoding a writer must use and persist on a table
whose encoding is tracked
+ * ({@link #isComplexKeyGenEncodingTracked}) but not yet recorded: the
encoding deduced from the data,
+ * otherwise the configured {@code hoodie.write.complex.keygen.new.encoding}
when
+ * {@code hoodie.write.complex.keygen.validation.enable} is false.
+ *
+ * @return empty when the encoding cannot be determined and the validation
is enabled; the caller fails
+ * the operation with {@link #getComplexKeygenErrorMessage}
+ */
+ public static Option<ComplexKeyGenEncoding>
resolveComplexKeyGenEncodingForWrite(HoodieTableMetaClient metaClient,
+
HoodieWriteConfig config) {
+ Option<ComplexKeyGenEncoding> deduced =
deduceComplexKeyGenEncodingFromData(metaClient);
+ if (deduced.isPresent() || config.enableComplexKeygenValidation()) {
+ return deduced;
+ }
+ ComplexKeyGenEncoding configured =
+
ComplexKeyGenEncoding.fromUseNewEncoding(config.getBooleanOrDefault(COMPLEX_KEYGEN_NEW_ENCODING));
+ LOG.warn("Could not determine the record key encoding of table {} from its
data; recording the configured {} "
+ + "because {} is disabled. If the existing keys are not {}, fix {}
and rerun, or the records written "
+ + "from now on will not match the existing ones.",
+ metaClient.getBasePath(), configured,
ENABLE_COMPLEX_KEYGEN_VALIDATION.key(), configured,
+ COMPLEX_KEYGEN_NEW_ENCODING.key());
+ return Option.of(configured);
+ }
+
+ /**
+ * Deduces the record key encoding of a single-field complex key generator
table from the
+ * {@code _hoodie_record_key} stored in its most recently written data file.
+ *
+ * @return {@link ComplexKeyGenEncoding#FIELD_PREFIXED} for a table that
never had data files, the encoding
+ * read from the first readable base or log file otherwise, or empty when
the table has (or had) data files
+ * but none of them yields a record key
+ */
+ public static Option<ComplexKeyGenEncoding>
deduceComplexKeyGenEncodingFromData(HoodieTableMetaClient metaClient) {
+ String expectedPrefix =
metaClient.getTableConfig().getRecordKeyFields().get()[0] +
DEFAULT_COLUMN_VALUE_SEPARATOR;
+ HoodieTimeline completedTimeline =
metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
+ boolean hasDataFiles = false;
+ for (HoodieInstant instant :
completedTimeline.getReverseOrderedInstants().collect(Collectors.toList())) {
+ for (HoodieWriteStat writeStat : getWriteStats(instant,
completedTimeline)) {
+ if (StringUtils.isNullOrEmpty(writeStat.getPath())) {
+ continue;
+ }
+ hasDataFiles = true;
+ StoragePath path = new StoragePath(metaClient.getBasePath(),
writeStat.getPath());
+ Option<String> recordKey = readFirstRecordKey(metaClient, path);
+ if (recordKey.isPresent()) {
+ ComplexKeyGenEncoding encoding =
+
ComplexKeyGenEncoding.fromUseNewEncoding(!recordKey.get().startsWith(expectedPrefix));
+ LOG.info("Deduced complex keygen record key encoding {} of table {}
from {}",
+ encoding, metaClient.getBasePath(), path);
+ return Option.of(encoding);
+ }
+ }
+ }
+ if (hasDataFiles || hasArchivedCommits(metaClient)) {
+ LOG.warn("No data file with a readable record key found in table {}; the
complex keygen record key "
+ + "encoding cannot be deduced", metaClient.getBasePath());
+ return Option.empty();
+ }
+ return Option.of(ComplexKeyGenEncoding.FIELD_PREFIXED);
+ }
+
+ /** Data files may have been written by commits that have since been
archived; only a table without any is new. */
+ private static boolean hasArchivedCommits(HoodieTableMetaClient metaClient) {
+ try {
+ return !metaClient.getArchivedTimeline().getCommitsTimeline().empty();
Review Comment:
Addressed: the new-table check now lists the archive folder instead of
loading the archived timeline.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/KeyGenUtils.java:
##########
@@ -436,15 +460,185 @@ public static String getComplexKeygenErrorMessage(String
operation) {
+ "`hoodie.write.complex.keygen.validation.enable=false` to skip this
validation.";
}
+ /**
+ * Whether a complex key generator with a single record key field prepends
the field name to the key.
+ *
+ * <p>{@link HoodieTableConfig#COMPLEX_KEYGEN_ENCODING} wins whenever it is
present in the properties, because
+ * it describes what the table's data carries. Otherwise the write table
version decides: 9 and above always
+ * prefix, 8 and below follow {@code
hoodie.write.complex.keygen.new.encoding}.
+ */
public static boolean
encodeSingleKeyFieldNameForComplexKeyGen(TypedProperties props) {
+ String tableEncoding =
props.getProperty(HoodieTableConfig.COMPLEX_KEYGEN_ENCODING.key());
+ if (!StringUtils.isNullOrEmpty(tableEncoding)) {
+ return
ComplexKeyGenEncoding.fromString(tableEncoding).encodesFieldName();
+ }
int tableVersionCode = ConfigUtils.getIntWithAltKeys(props,
WRITE_TABLE_VERSION);
HoodieTableVersion tableVersion =
HoodieTableVersion.fromVersionCode(tableVersionCode);
return tableVersion.greaterThanOrEquals(HoodieTableVersion.NINE)
|| !ConfigUtils.getBooleanWithAltKeys(props,
COMPLEX_KEYGEN_NEW_ENCODING);
}
- public static boolean mayUseNewEncodingForComplexKeyGen(HoodieTableConfig
tableConfig) {
- return tableConfig.getTableVersion().lesserThan(HoodieTableVersion.NINE)
- && isComplexKeyGeneratorWithSingleRecordKeyField(tableConfig);
+ /**
+ * Whether the table's record key encoding is tracked by {@link
HoodieTableConfig#COMPLEX_KEYGEN_ENCODING}:
+ * a complex key generator with a single record key field and a populated
{@code _hoodie_record_key}.
+ * Without the meta field there is no stored key whose encoding could
diverge from the key generator's.
+ */
+ public static boolean isComplexKeyGenEncodingTracked(HoodieTableConfig
tableConfig) {
+ return tableConfig.isComplexKeyGenWithSingleRecordKeyField() &&
tableConfig.isRecordKeyPopulated();
+ }
+
+ /**
+ * Resolves the record key encoding of a single-field complex key generator
table for readers, without
+ * modifying the table: the persisted property, otherwise the encoding
deduced from the table's data.
+ *
+ * @return empty when the table does not use a single-field complex key
generator, or when its encoding
+ * cannot be determined, in which case record-key based file pruning must be
skipped
+ */
+ public static Option<ComplexKeyGenEncoding>
resolveComplexKeyGenEncoding(HoodieTableMetaClient metaClient) {
+ HoodieTableConfig tableConfig = metaClient.getTableConfig();
+ if (!tableConfig.isComplexKeyGenWithSingleRecordKeyField()) {
+ return Option.empty();
+ }
+ if (!tableConfig.isRecordKeyPopulated()) {
+ // the keys are regenerated by the key generator, which follows the
table version
+ return
tableConfig.getTableVersion().greaterThanOrEquals(HoodieTableVersion.NINE)
+ ? Option.of(ComplexKeyGenEncoding.FIELD_PREFIXED) : Option.empty();
+ }
+ Option<ComplexKeyGenEncoding> persisted =
tableConfig.getComplexKeyGenEncoding();
+ return persisted.isPresent() ? persisted :
deduceComplexKeyGenEncodingFromData(metaClient);
+ }
+
+ /**
+ * Resolves the record key encoding a writer must use and persist on a table
whose encoding is tracked
+ * ({@link #isComplexKeyGenEncodingTracked}) but not yet recorded: the
encoding deduced from the data,
+ * otherwise the configured {@code hoodie.write.complex.keygen.new.encoding}
when
+ * {@code hoodie.write.complex.keygen.validation.enable} is false.
+ *
+ * @return empty when the encoding cannot be determined and the validation
is enabled; the caller fails
+ * the operation with {@link #getComplexKeygenErrorMessage}
+ */
+ public static Option<ComplexKeyGenEncoding>
resolveComplexKeyGenEncodingForWrite(HoodieTableMetaClient metaClient,
+
HoodieWriteConfig config) {
+ Option<ComplexKeyGenEncoding> deduced =
deduceComplexKeyGenEncodingFromData(metaClient);
+ if (deduced.isPresent() || config.enableComplexKeygenValidation()) {
+ return deduced;
+ }
+ ComplexKeyGenEncoding configured =
+
ComplexKeyGenEncoding.fromUseNewEncoding(config.getBooleanOrDefault(COMPLEX_KEYGEN_NEW_ENCODING));
+ LOG.warn("Could not determine the record key encoding of table {} from its
data; recording the configured {} "
+ + "because {} is disabled. If the existing keys are not {}, fix {}
and rerun, or the records written "
+ + "from now on will not match the existing ones.",
+ metaClient.getBasePath(), configured,
ENABLE_COMPLEX_KEYGEN_VALIDATION.key(), configured,
+ COMPLEX_KEYGEN_NEW_ENCODING.key());
+ return Option.of(configured);
+ }
+
+ /**
+ * Deduces the record key encoding of a single-field complex key generator
table from the
+ * {@code _hoodie_record_key} stored in its most recently written data file.
+ *
+ * @return {@link ComplexKeyGenEncoding#FIELD_PREFIXED} for a table that
never had data files, the encoding
+ * read from the first readable base or log file otherwise, or empty when
the table has (or had) data files
+ * but none of them yields a record key
+ */
+ public static Option<ComplexKeyGenEncoding>
deduceComplexKeyGenEncodingFromData(HoodieTableMetaClient metaClient) {
+ String expectedPrefix =
metaClient.getTableConfig().getRecordKeyFields().get()[0] +
DEFAULT_COLUMN_VALUE_SEPARATOR;
+ HoodieTimeline completedTimeline =
metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
+ boolean hasDataFiles = false;
+ for (HoodieInstant instant :
completedTimeline.getReverseOrderedInstants().collect(Collectors.toList())) {
+ for (HoodieWriteStat writeStat : getWriteStats(instant,
completedTimeline)) {
+ if (StringUtils.isNullOrEmpty(writeStat.getPath())) {
+ continue;
+ }
+ hasDataFiles = true;
+ StoragePath path = new StoragePath(metaClient.getBasePath(),
writeStat.getPath());
+ Option<String> recordKey = readFirstRecordKey(metaClient, path);
+ if (recordKey.isPresent()) {
+ ComplexKeyGenEncoding encoding =
+
ComplexKeyGenEncoding.fromUseNewEncoding(!recordKey.get().startsWith(expectedPrefix));
+ LOG.info("Deduced complex keygen record key encoding {} of table {}
from {}",
+ encoding, metaClient.getBasePath(), path);
+ return Option.of(encoding);
+ }
+ }
+ }
+ if (hasDataFiles || hasArchivedCommits(metaClient)) {
+ LOG.warn("No data file with a readable record key found in table {}; the
complex keygen record key "
+ + "encoding cannot be deduced", metaClient.getBasePath());
+ return Option.empty();
+ }
+ return Option.of(ComplexKeyGenEncoding.FIELD_PREFIXED);
+ }
+
+ /** Data files may have been written by commits that have since been
archived; only a table without any is new. */
+ private static boolean hasArchivedCommits(HoodieTableMetaClient metaClient) {
+ try {
+ return !metaClient.getArchivedTimeline().getCommitsTimeline().empty();
+ } catch (Exception e) {
+ LOG.warn("Could not read the archived timeline of table {}",
metaClient.getBasePath(), e);
+ return true;
+ }
+ }
+
+ /**
+ * Copies the resolved {@link HoodieTableConfig#COMPLEX_KEYGEN_ENCODING}
from a write config onto another
+ * config, for components that build their key generator from a separate
config.
+ */
+ public static void copyResolvedComplexKeyEncoding(HoodieConfig from,
HoodieConfig to) {
+ if (from.contains(HoodieTableConfig.COMPLEX_KEYGEN_ENCODING)) {
Review Comment:
Addressed: the write config is no longer touched; each site puts the table's
recorded encoding on the key generator props right before instantiating it
(KeyGenUtils#withComplexKeyGenEncoding), and the property itself is recorded on
the table when the commit starts.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java:
##########
@@ -726,11 +730,52 @@ public static Option<Partitioner>
getInsertPartitioner(Configuration conf) {
}
/**
- * Returns whether complex keygen encodes single record key with field name.
+ * Returns whether the configured key generator is the complex key generator
(Avro or Spark flavour).
*/
- public static boolean useComplexKeygenNewEncoding(Configuration conf) {
- return
Boolean.parseBoolean(conf.getString(HoodieWriteConfig.COMPLEX_KEYGEN_NEW_ENCODING.key(),
-
HoodieWriteConfig.COMPLEX_KEYGEN_NEW_ENCODING.defaultValue().toString()));
+ public static boolean isComplexKeyGenerator(Configuration conf) {
Review Comment:
Addressed: the key generator class checks are gone; HoodieTableFactory sets
the table option on the job configuration and RowDataKeyGen reads it, keeping
its previous rule when the option is absent.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java:
##########
@@ -1544,6 +1546,38 @@ protected boolean loadActiveTimelineOnTableInit() {
return true;
}
+ /**
+ * Resolves the record key encoding of a single-field {@code
ComplexKeyGenerator} table onto the write config,
+ * so that every key generator built from it keys records the way the
table's data is keyed. A table that does
+ * not carry {@link HoodieTableConfig#COMPLEX_KEYGEN_ENCODING} yet gets it
deduced from its data and backfilled
+ * under the transaction lock, or the write fails when the encoding cannot
be determined and
+ * {@code hoodie.write.complex.keygen.validation.enable} is on.
+ * Public because the streamer keys its records before {@link #initTable}
runs and has to call this itself.
+ */
+ public void resolveComplexKeygenEncoding(HoodieTableMetaClient metaClient) {
+ if
(!KeyGenUtils.isComplexKeyGenEncodingTracked(metaClient.getTableConfig())) {
+ return;
+ }
+ if (!metaClient.getTableConfig().getComplexKeyGenEncoding().isPresent()) {
+ executeUsingTxnManager(Option.empty(), () ->
backfillComplexKeygenEncoding(metaClient));
+ }
+ config.setValue(HoodieTableConfig.COMPLEX_KEYGEN_ENCODING,
metaClient.getTableConfig().getComplexKeyGenEncoding().get().name());
+ }
+
+ private void backfillComplexKeygenEncoding(HoodieTableMetaClient metaClient)
{
Review Comment:
Addressed: no more config injection; the encoding is recorded on the table
in startCommit (under the transaction lock, before any record is keyed) and
applied to the key generator props at instantiation.
--
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]
