voonhous commented on code in PR #19205:
URL: https://github.com/apache/hudi/pull/19205#discussion_r3773362078
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java:
##########
@@ -1535,7 +1570,50 @@ public Properties build() {
tableConfig.setValue(HoodieTableConfig.CDC_SUPPLEMENTAL_LOGGING_MODE,
cdcSupplementalLoggingMode);
}
}
- if (null != populateMetaFields) {
+ // hoodie.meta.fields.mode is the source of truth, and hoodie.properties
must never contradict
+ // it: a table written selectively that still recorded
populate.meta.fields=true would be read
+ // as ALL by a pre-1.3.0 reader, which ignores the mode property
entirely. For NONE that is
+ // actively unsafe — an older incremental reader would run against
all-null commit times and
+ // silently return no rows.
+ //
+ // A caller that states both and disagrees is rejected rather than
silently overridden. Half
+ // their request would otherwise be discarded without a word, and it
would be inconsistent with
+ // BaseHoodieWriteClient#validateAgainstTableProperties, which already
rejects an explicitly-set
+ // boolean that disagrees with the table. Only a genuine contradiction
fails: ALL + true and
+ // NONE + false are coherent restatements and pass.
+ if (null != metaFieldsMode) {
Review Comment:
**[blocker] Neither bootstrap path can create a table with any mode other
than `ALL`.**
Both bootstrap table builders call `.fromProperties(props)` -- which now
sets the mode from `hoodie.meta.fields.mode`
(`HoodieTableMetaClient.java:1430-1432`) -- and then hand this check a
hard-coded `true` whenever the user never mentioned the deprecated boolean:
-
`hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/BootstrapExecutor.java:212`
-
`hudi-spark-datasource/hudi-spark/src/main/java/org/apache/hudi/cli/BootstrapExecutorUtils.java:239`
both do:
```java
.setPopulateMetaFields(props.getBoolean(
POPULATE_META_FIELDS.key(), POPULATE_META_FIELDS.defaultValue()))
```
`TypedProperties.getBoolean(String, boolean)` returns a primitive, so the
argument is never null. Bootstrapping with
`hoodie.meta.fields.mode=COMMIT_TIME_ONLY` (or `NONE`) therefore always throws
`Conflicting meta-field settings at table creation`.
This is the exact trap the PR already identified and fixed elsewhere --
`StreamSync.java:482-483` and `HoodieSparkSqlWriter.scala:290-296` both carry a
comment explaining that `getBooleanOrDefault` "would hand it the `true` default
... turning a plain `hoodie.meta.fields.mode=COMMIT_TIME_ONLY` run into a
spurious conflict." `git grep -n setPopulateMetaFields -- '*/src/main/*'` shows
3 of 5 call sites pass `null` when unstated; these 2 were missed.
Please apply the same pattern at both sites:
```java
.setPopulateMetaFields(props.containsKey(POPULATE_META_FIELDS.key())
? props.getBoolean(POPULATE_META_FIELDS.key()) : null)
```
and add one bootstrap test that creates with `hoodie.meta.fields.mode=NONE`
and asserts the table config round-trips.
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/common/model/HoodieSparkRecord.java:
##########
@@ -296,9 +307,14 @@ public HoodieRecord
wrapIntoHoodieRecordPayloadWithKeyGen(HoodieSchema recordSch
StructType structType =
HoodieInternalRowUtils.getCachedSchema(recordSchema);
String key;
String partition;
- boolean populateMetaFields =
Boolean.parseBoolean(props.getOrDefault(POPULATE_META_FIELDS.key(),
- POPULATE_META_FIELDS.defaultValue().toString()).toString());
- if (!populateMetaFields && keyGen.isPresent()) {
+ // Resolve via hoodie.meta.fields.mode — reading the deprecated boolean
alone would report
+ // "populated" for a selective-mode table (whose _hoodie_record_key column
is null), sending us
+ // down the meta-column branch below and NPE-ing on the null ordinal.
+ boolean recordKeyPopulated = MetaFieldsMode.resolve(
Review Comment:
**[major] Two guards still read the deprecated boolean out of a raw params
map, so a config that states only the mode gets the wrong answer -- including
for `NONE`, which this PR documents as exactly equivalent to
`populate.meta.fields=false`.**
This call site was fixed to resolve through `MetaFieldsMode`; these two were
not:
**1. `AutoRecordKeyGenerationUtils.scala:47`** -- the guard that blocks auto
record-key generation on virtual-key tables:
```scala
if (!parameters.getOrElse(HoodieTableConfig.POPULATE_META_FIELDS.key(),
HoodieTableConfig.POPULATE_META_FIELDS.defaultValue().toString).toBoolean) {
```
`parameters` never contains `hoodie.populate.meta.fields` when the user
states only the mode, so this reads the `true` default and does not fire. This
is a *tested* invariant being bypassed:
`TestAutoGenerationOfRecordKeys.testRecordKeysAutoGenInvalidParams`
(`:172-196`) asserts a `HoodieKeyGeneratorException` for
`hoodie.populate.meta.fields,false`, added by `a34067826c2f` (#8107). The write
then succeeds, `HoodieDatasetBulkInsertHelper` takes its non-populate branch
and never runs the key generator, and the rows land with no identity at all.
Called from `HoodieSparkSqlWriter.scala:404` and `:486`.
**2. `DataSourceOptions.scala:552`** -- the `ENABLE_ROW_WRITER` infer
function stops disabling the row writer for `BULK_INSERT` + non-populated meta
fields + `COMBINE_BEFORE_INSERT`, so `COMBINE_BEFORE_INSERT` is silently
ignored. Its own inline comment says that must not happen.
Please change both to resolve the mode, matching this line:
```scala
if (!MetaFieldsMode.resolve(hoodieConfig).isRecordKeyPopulated) {
```
and extend
`TestAutoGenerationOfRecordKeys.testRecordKeysAutoGenInvalidParams`'s
`@CsvSource` with `"hoodie.meta.fields.mode,NONE"` and
`"hoodie.meta.fields.mode,COMMIT_TIME_ONLY"`.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java:
##########
@@ -3889,6 +3974,27 @@ private void validate() {
checkArgument(ttlStatsMaxParallelism > 0,
String.format("%s must be positive, but was %d",
HoodieTTLConfig.STATS_MAX_PARALLELISM.key(),
ttlStatsMaxParallelism));
+
+ // hoodie.meta.fields.mode is the source of truth for meta-column
population; the deprecated
+ // populate.meta.fields boolean is consulted only when the mode is
absent. There is therefore
+ // no ambiguous combination to reject here — MetaFieldsMode.resolve
throws on unrecognized
+ // values.
+ MetaFieldsMode metaFieldsMode = writeConfig.getMetaFieldsMode();
+ // Selective meta-field modes are CoW-only in this release. MoR
log-write path does not yet
+ // respect the mode, which would silently produce log records with null
meta columns.
+ boolean isSelective = metaFieldsMode.isSelective();
Review Comment:
**[major] Nothing rejects the default BLOOM index on a selective table, and
the bloom filter is never written, so the first upsert NPEs.**
A selective mode makes `populateMetaFields()` false, which disables
bloom-filter creation at write time -- `enableBloomFilter(populateMetaFields,
hoodieConfig)` at `HoodieAvroFileWriterFactory.java:80` and
`HoodieSparkFileWriterFactory.java:74`. On read,
`FileFormatUtils.readBloomFilterFromMetadata` initialises `BloomFilter toReturn
= null` and returns it unchanged when the footer key is absent (`:162-183`).
`HoodieKeyLookupHandle.getBloomFilter()` (`:58-77`) propagates that null, and
`addKey` (`:82-89`) calls `bloomFilter.mightContain(recordKey)` with no null
check.
`hoodie.index.type` defaults to BLOOM and nothing in this file's
`validate()` or in `validateAgainstTableProperties` rejects it, so a
`COMMIT_TIME_ONLY` table created with defaults NPEs on its first upsert.
The mechanism predates this PR -- it is identical under
`populate.meta.fields=false` -- but that setting is documented as "only meant
to be used for append only/immutable data", so the combination was implausible.
`COMMIT_TIME_ONLY` is pitched at tables that keep taking writes, which is what
makes it reachable. Note `TestHoodieIndex.indexTypeParams` (`:116-132`)
deliberately pairs BLOOM/GLOBAL_BLOOM only with `populateMetaFields=true`.
Please reject BLOOM and GLOBAL_BLOOM when the record key is not populated,
next to the existing keygen check in `validateAgainstTableProperties`:
```java
if (!tableConfig.isRecordKeyPopulated()
&& (writeConfig.getIndexType() == HoodieIndex.IndexType.BLOOM
|| writeConfig.getIndexType() ==
HoodieIndex.IndexType.GLOBAL_BLOOM)) {
throw new HoodieException(...);
}
```
and add a `{IndexType.BLOOM, false, false}` row to
`TestHoodieIndex.indexTypeParams`.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java:
##########
@@ -3587,11 +3619,64 @@ public Builder withCanIgnorePostCommitFailures(boolean
canIgnorePostCommitFailur
return this;
}
+ /**
+ * @deprecated since 1.3.0, use {@link
#withMetaFieldsMode(MetaFieldsMode)} instead
+ * ({@code true} maps to {@link MetaFieldsMode#ALL}, {@code false} to
{@link MetaFieldsMode#NONE}).
+ */
+ @Deprecated
public Builder withPopulateMetaFields(boolean populateMetaFields) {
writeConfig.setValue(HoodieTableConfig.POPULATE_META_FIELDS,
Boolean.toString(populateMetaFields));
return this;
}
+ public Builder withMetaFieldsMode(MetaFieldsMode metaFieldsMode) {
+ // Leaving the mode unset defers to the deprecated populate.meta.fields
boolean. The legacy
+ // boolean is derived from the mode in build() rather than here, so the
two cannot be made to
+ // disagree by calling the setters in either order.
+ writeConfig.setValue(HoodieTableConfig.META_FIELDS_MODE,
+ metaFieldsMode == null ? "" : metaFieldsMode.name());
+ return this;
+ }
+
+ /**
+ * Rewrite the deprecated {@code populate.meta.fields} boolean from {@code
meta.fields.mode}
+ * whenever a mode is set, so the two can never disagree on the resulting
config.
+ *
+ * <p>Done at build time, not in the setter: {@code
withPopulateMetaFields} does not re-derive
+ * the mode, so deriving in {@link #withMetaFieldsMode} alone would make
the invariant depend on
+ * call order. {@code
withMetaFieldsMode(COMMIT_TIME_ONLY).withPopulateMetaFields(true)} would
+ * leave a selective mode sitting next to {@code
populate.meta.fields=true} — a config that
+ * resolves correctly (the mode wins) but carries the contradiction to
disk on any path that
+ * copies raw write-config props into {@code hoodie.properties},
misleading pre-1.3.0 readers
+ * into treating the table as ALL.
+ */
+ private void deriveLegacyPopulateMetaFieldsFromMode() {
+ String rawMode =
writeConfig.getString(HoodieTableConfig.META_FIELDS_MODE);
+ if (StringUtils.isNullOrEmpty(rawMode)) {
+ return;
+ }
+ boolean derived =
MetaFieldsMode.parse(rawMode).toLegacyPopulateMetaFields();
+ // A caller that explicitly set the boolean to something the mode
contradicts is rejected rather
+ // than silently overridden — otherwise half their request is discarded
without a word. Only a
+ // genuine contradiction fails; restating the derived value (ALL + true,
NONE + false) passes.
+ // An absent boolean is the ordinary case and simply takes the derived
value.
+ checkArgument(
Review Comment:
**[blocker] This `checkArgument` fires during archival on every table that
has been upgraded to v10.**
`LSMTimelineWriter.getOrCreateWriterConfig()`
(`LSMTimelineWriter.java:446-448`) builds its writer config by inheriting a
props blob and then overriding one of the two meta-field properties:
```java
HoodieWriteConfig.newBuilder()
.withProperties(this.config.getProps())
.withPopulateMetaFields(false).build();
```
`withProperties` is `writeConfig.getProps().putAll(properties)`
(`HoodieWriteConfig.java:3716`), so an inherited `hoodie.meta.fields.mode=ALL`
sits next to the explicitly-set `false`. `derived` is `true`, the boolean reads
`false`, and this throws `IllegalArgumentException: Conflicting meta-field
settings on the write config`.
The chain that puts `ALL` on the parent config is entirely internal -- no
user sets anything:
1. `NineToTenUpgradeHandler` writes `hoodie.meta.fields.mode=ALL` into
`hoodie.properties` for every v9 table. `HoodieTableVersion.current()` is
`TEN`, so this runs on the first write to any existing table, and
`TestNineToTenUpgradeHandler` asserts it.
2. `HoodieSparkSqlWriter.scala:1113-1118` copies every
`tableConfig.getProps` entry into the write params when `mode != Overwrite`.
3. `TimelineArchiverV2.java:88` hands that config to `LSMTimelineWriter`.
Note the asymmetry: selective modes survive (`derived=false` matches the
explicit `false`). It is `ALL` tables -- the default, and the overwhelming
majority -- that break. It surfaces only once archival actually writes an LSM
file, i.e. past `hoodie.keep.min.commits`, which is why nothing here catches
it: `grep -c archiv TestMetaFieldsModeE2E.java` returns 0.
`SevenToEightUpgradeHandler.java:259` has the same shape, as does
`RunTimelineCompactionProcedure` via `HoodieCLIUtils.createHoodieWriteClient`,
which also merges table props.
Swapping the call to `.withMetaFieldsMode(MetaFieldsMode.NONE)` does **not**
fix it -- the upgrade keeps `hoodie.populate.meta.fields=true` on disk, so that
value is inherited too and the check throws in the other direction.
The root cause is that this check cannot distinguish a value the caller
stated on *this* builder from one that arrived in a props blob. Suggested fix:
track the explicit setter calls, and only reject a caller-stated contradiction:
```java
// in Builder
private Boolean statedPopulateMetaFields; // set by withPopulateMetaFields
private boolean statedMetaFieldsMode; // set by withMetaFieldsMode
```
and gate the `checkArgument` on `statedPopulateMetaFields != null`, treating
an inherited boolean as something the mode overrides rather than contradicts.
As a minimal unblock if you would rather keep the check as-is,
`LSMTimelineWriter` must state both consistently:
```java
this.writeConfig = HoodieWriteConfig.newBuilder()
.withProperties(this.config.getProps())
.withMetaFieldsMode(MetaFieldsMode.NONE)
.withPopulateMetaFields(false).build();
```
Either way, please add a test that runs `TimelineArchiverV2` past
`keep.min.commits` on a table upgraded v9 -> v10.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/IncrementalRelationV1.scala:
##########
@@ -89,8 +89,12 @@ class IncrementalRelationV1(val sqlContext: SQLContext,
s"option ${DataSourceReadOptions.START_COMMIT.key}")
}
- if (!metaClient.getTableConfig.populateMetaFields()) {
- throw new HoodieException("Incremental queries are not supported when meta
fields are disabled")
+ if (!metaClient.getTableConfig.isCommitTimePopulated()) {
Review Comment:
**[major] This guard is relaxed from `populateMetaFields()` to
`isCommitTimePopulated()` with no test on either side.**
The change newly *permits* Spark structured-streaming incremental reads on
`COMMIT_TIME_ONLY` and `COMMIT_TIME_AND_FILE_NAME` tables. But
`IncrementalRelationV1`/`V2` back only the streaming source -- `grep -rn "new
IncrementalRelationV1\|new IncrementalRelationV2" --include=*.scala */src/main`
returns only `HoodieStreamSourceV1.scala:186` and
`HoodieStreamSourceV2.scala:162`. The datasource incremental tests added in
this PR route through `HoodieCopyOnWriteIncrementalHadoopFsRelationFactory`
instead, and `grep -c readStream TestMetaFieldsModeE2E.java` is 0.
The same is true of the narrowed branch in
`MergeOnReadIncrementalRelationV1.scala:273` / `V2.scala:263`
(`metaClient.getTableType == MERGE_ON_READ && !populateMetaFields()`), which is
untested in both directions.
This is a load-bearing branch on this feature: the read path has already
broken twice here -- `13b39db9fb22` (zero-row incremental reads under a
selective mode, shipped with 0 tests) and the
MoR-guard-rejects-every-selective-CoW-table bug found while writing the
datasource tests. These two classes are also the subject of two recent
data-loss fixes, `180592a0ad71` (#17514) and `859a7ee4fb61` (HUDI-9540).
Please add one `spark.readStream.format("hudi")` test on a
`COMMIT_TIME_ONLY` table asserting rows come back, and one asserting a MoR
table with `populate.meta.fields=false` is still rejected.
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestMetaFieldsModeE2E.java:
##########
@@ -0,0 +1,830 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.functional;
+
+import org.apache.hudi.DataSourceReadOptions;
+import org.apache.hudi.DataSourceWriteOptions;
+import org.apache.hudi.SparkAdapterSupport$;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.MetaFieldsMode;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.testutils.SparkClientFunctionalTestHarness;
+
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+import org.apache.spark.sql.SaveMode;
+import org.apache.spark.sql.functions;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Spark-datasource end-to-end tests for the {@code hoodie.meta.fields.mode}
property on CoW tables.
+ * Every {@link MetaFieldsMode} value is exercised via a write / re-read round
trip; on-disk column
+ * population is verified by reading the parquet files back and inspecting the
meta-column values.
+ */
+class TestMetaFieldsModeE2E extends SparkClientFunctionalTestHarness {
+
+ private static StructType simpleSchema() {
+ return DataTypes.createStructType(new StructField[]{
+ DataTypes.createStructField("column1", DataTypes.StringType, true),
+ DataTypes.createStructField("column2", DataTypes.StringType, true),
+ DataTypes.createStructField("column3", DataTypes.StringType, true)
+ }).asNullable();
+ }
+
+ private Map<String, String> baseOptions() {
+ Map<String, String> opts = new HashMap<>();
+ opts.put(DataSourceWriteOptions.RECORDKEY_FIELD().key(), "column1");
Review Comment:
**[major] Every test in this class uses a single-field record key on a
single partition field with the default keygen, which leaves the
historically-fragile virtual-key read branches unexercised.**
`grep -cE "ComplexKeyGenerator|Nonpartitioned|hive_style|HIVE_STYLE"
TestMetaFieldsModeE2E.java` returns 0.
The writer-side allowlist for non-populated meta fields permits COMPLEX and
NONPARTITIONED (`KeyGeneratorType.NO_METAFIELDS_KEYGEN_ALLOWLIST`), and
`validateAgainstTableProperties` enforces exactly that list -- but the read
path asserts `checkState(keyFields.length == 1)` at
`HoodieBaseRelation.scala:128`. So a selective table with a two-field
`ComplexKeyGenerator` is reachable through validation and unexercised by any
test.
This is the bug class that has bitten this exact combination twice before:
- `4f6fc726d0d3` -- hive-style partitioning and the default partition broke
bulk-insert row writer with SimpleKeyGen + virtual keys (#5664)
- `7da97c8096ad` -- non-partitioned with virtual keys broke the read path
(#5747)
Selective modes make all of those combinations newly reachable, and they
route through the same `populateMetaFields()==false` branches.
Please parameterize at least one write/read round trip under
`COMMIT_TIME_ONLY` over {simple keygen, two-field complex keygen,
non-partitioned keygen} x {`hive_style_partitioning` on/off}.
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestMetaFieldsModeE2E.java:
##########
@@ -0,0 +1,830 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.functional;
+
+import org.apache.hudi.DataSourceReadOptions;
+import org.apache.hudi.DataSourceWriteOptions;
+import org.apache.hudi.SparkAdapterSupport$;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.MetaFieldsMode;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.testutils.SparkClientFunctionalTestHarness;
+
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+import org.apache.spark.sql.SaveMode;
+import org.apache.spark.sql.functions;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Spark-datasource end-to-end tests for the {@code hoodie.meta.fields.mode}
property on CoW tables.
+ * Every {@link MetaFieldsMode} value is exercised via a write / re-read round
trip; on-disk column
+ * population is verified by reading the parquet files back and inspecting the
meta-column values.
+ */
+class TestMetaFieldsModeE2E extends SparkClientFunctionalTestHarness {
+
+ private static StructType simpleSchema() {
+ return DataTypes.createStructType(new StructField[]{
+ DataTypes.createStructField("column1", DataTypes.StringType, true),
+ DataTypes.createStructField("column2", DataTypes.StringType, true),
+ DataTypes.createStructField("column3", DataTypes.StringType, true)
+ }).asNullable();
+ }
+
+ private Map<String, String> baseOptions() {
+ Map<String, String> opts = new HashMap<>();
+ opts.put(DataSourceWriteOptions.RECORDKEY_FIELD().key(), "column1");
+ opts.put(DataSourceWriteOptions.PARTITIONPATH_FIELD().key(), "column2");
+ opts.put(DataSourceWriteOptions.ORDERING_FIELDS().key(), "column3");
+ opts.put(HoodieTableConfig.NAME.key(), "test_meta_fields_mode");
+ opts.put(DataSourceWriteOptions.TABLE_TYPE().key(), "COPY_ON_WRITE");
+ opts.put(HoodieMetadataConfig.ENABLE.key(), "false");
+ return opts;
+ }
+
+ private void writeRows(List<Row> records, StructType schema, Map<String,
String> options, String path, SaveMode mode) {
+ spark().createDataset(records,
+
SparkAdapterSupport$.MODULE$.sparkAdapter().getCatalystExpressionUtils().getEncoder(schema))
+ .write()
+ .format("hudi")
+ .options(options)
+ .mode(mode)
+ .save(path);
+ }
+
+ private HoodieTableConfig writeSampleAndGetTableConfig(Map<String, String>
options, String path) {
+ writeRows(Arrays.asList(
+ RowFactory.create("k1", "p1", "v1"),
+ RowFactory.create("k2", "p1", "v2")),
+ simpleSchema(), options, path, SaveMode.Overwrite);
+ HoodieTableMetaClient metaClient =
+
HoodieTableMetaClient.builder().setBasePath(path).setConf(storageConf()).build();
+ return metaClient.getTableConfig();
+ }
+
+ /**
+ * End-to-end assertion of the on-disk meta columns after a write. Reads the
parquet files back
+ * (bypassing Hudi's own read path so we see the raw column values) and
asserts which meta
+ * columns are non-null.
+ */
+ private void assertMetaColumnPopulation(String path, MetaFieldsMode
expectedMode) {
+ Dataset<Row> raw = spark().read().parquet(path + "/*/*.parquet");
+ Row first = raw.select(
+ HoodieRecord.COMMIT_TIME_METADATA_FIELD,
+ HoodieRecord.COMMIT_SEQNO_METADATA_FIELD,
+ HoodieRecord.RECORD_KEY_METADATA_FIELD,
+ HoodieRecord.PARTITION_PATH_METADATA_FIELD,
+ HoodieRecord.FILENAME_METADATA_FIELD).first();
Review Comment:
**[major] This helper inspects exactly one row, so the failure mode the
feature exists to prevent passes.**
`.first()` on an unordered glob means a writer that populates a meta column
on some rows and not others -- mixed-mode output, which is the whole hazard
behind the mode being immutable -- is never detected. 12 of the 28 methods in
this class assert only through this helper.
Concretely: make `HoodieRowCreateHandle.writeRowSelectiveMetaFields`
populate the opted-in columns for the first record only, and all 12 stay green.
This is the same `.first()` pattern I flagged on the old clustering helper
(`TestMetaFieldsMode.java:343`, 07-28); it was fixed there and survives here in
the helper that most of the class depends on. The PR already added a correct
version of this assertion in
`HoodieSparkWriterTestBase.assertNoMetaFieldsPopulated` (`:135-141`), which
counts over every row.
Please assert over the whole dataset instead:
```java
long total = raw.count();
for (int ord : populatedOrdinals) {
assertEquals(total,
raw.filter(col(HoodieRecord.HOODIE_META_COLUMNS.get(ord)).isNotNull()).count());
}
for (int ord : unpopulatedOrdinals) {
assertEquals(0,
raw.filter(col(HoodieRecord.HOODIE_META_COLUMNS.get(ord)).isNotNull()).count());
}
```
`TestHoodieStreamerMetaFieldsMode.assertOnDiskMetaColumns` (`:247-276`) has
the same defect and the same fix.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/TenToNineDowngradeHandler.java:
##########
@@ -18,25 +18,144 @@
package org.apache.hudi.table.upgrade;
+import org.apache.hudi.common.config.ConfigProperty;
import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.MetaFieldsMode;
import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieUpgradeDowngradeException;
-import java.util.Collections;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
/**
* Version 10 writes native log files by default. Downgrading to version 9
requires
* full compaction of native data/delete logs before the downgrade completes.
+ *
+ * <p>Version 10 also introduced {@code hoodie.meta.fields.mode}. {@code
hoodie.populate.meta.fields}
+ * is always written back from it ({@code ALL -> true}, every other mode
{@code -> false}), mirroring
+ * how {@link NineToEightDowngradeHandler} restores {@code
hoodie.table.payload.class}. That restate
+ * is load-bearing: {@code POPULATE_META_FIELDS} defaults to {@code true}, so
a table carrying only
+ * the mode would otherwise downgrade to {@code ALL} and claim {@code
_hoodie_record_key} is
+ * populated on files where it is null.
+ *
+ * <p>What happens to the mode itself depends on whether the legacy boolean
can express it:
+ *
+ * <ul>
+ * <li>{@link MetaFieldsMode#ALL} / {@link MetaFieldsMode#NONE} — dropped.
These are exactly the
+ * two states the boolean expresses, so the mode carries nothing the
downgraded table lacks.</li>
+ * <li>A selective mode, restated by the writer — <b>retained</b>. Restating
it is the operator
+ * asserting that every reader of this table honors the mode rather than
the boolean alone.
+ * Keeping it is also what makes the round trip lossless: a later
re-upgrade finds the mode
+ * intact rather than deriving {@code NONE} from the boolean.</li>
+ * <li>A selective mode, not restated (or restated as a different value) —
<b>rejected</b>.
+ * Dropping it would collapse the table to {@code NONE} irreversibly,
and that is not a call to
+ * make on the operator's behalf.</li>
+ * </ul>
+ *
+ * <p>Retaining the mode on a version 9 table is safe mechanically: the
property carries no
+ * {@code sinceVersion}, so {@code dropInvalidConfigs} does not strip it on
load.
*/
public class TenToNineDowngradeHandler implements DowngradeHandler {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(TenToNineDowngradeHandler.class);
+
+ /**
+ * The meta-fields mode the writer explicitly asked for, if any.
+ *
+ * <p>Empty when the property is absent or blank -- which is the ordinary
case, since almost no
+ * caller restates meta-field settings. Restating it during a downgrade is
therefore a deliberate
+ * signal rather than something that happens by accident, which is what
makes it usable as consent.
+ */
+ private static Option<MetaFieldsMode> statedMetaFieldsMode(HoodieWriteConfig
config) {
+ if (config == null ||
!config.contains(HoodieTableConfig.META_FIELDS_MODE)) {
+ return Option.empty();
+ }
+ String raw = config.getString(HoodieTableConfig.META_FIELDS_MODE);
+ return StringUtils.isNullOrEmpty(raw) ? Option.empty() :
Option.of(MetaFieldsMode.parse(raw));
+ }
+
@Override
public UpgradeDowngrade.TableConfigChangeSet downgrade(
HoodieWriteConfig config,
HoodieEngineContext context,
String instantTime,
SupportsUpgradeDowngrade upgradeDowngradeHelper) {
+ Set<ConfigProperty> propertiesToDelete = new HashSet<>();
+ propertiesToDelete.add(HoodieTableConfig.TABLE_STORAGE_LAYOUT);
+
+ Map<ConfigProperty, String> propertiesToUpdate = new HashMap<>();
+ if (upgradeDowngradeHelper != null) {
+ MetaFieldsMode metaFieldsMode =
+ upgradeDowngradeHelper.getTable(config,
context).getMetaClient().getTableConfig().getMetaFieldsMode();
+
+ // Always restate the legacy boolean from the mode. Version 9 readers
understand only that
+ // property, and without it POPULATE_META_FIELDS falls back to its
`true` default -- i.e. the
+ // table silently downgrades to ALL and claims meta columns it does not
have. For ALL / NONE
+ // this restates what was already there; for a selective mode it writes
`false`, so a reader
+ // that does not honor the mode under-claims rather than over-claims.
+ propertiesToUpdate.put(HoodieTableConfig.POPULATE_META_FIELDS,
+ String.valueOf(metaFieldsMode.toLegacyPopulateMetaFields()));
+
+ if (!metaFieldsMode.isSelective()) {
+ // ALL and NONE are exactly what the boolean can express, so the mode
carries no information
+ // the downgraded table lacks. Drop it.
+ propertiesToDelete.add(HoodieTableConfig.META_FIELDS_MODE);
+ } else if
(statedMetaFieldsMode(config).map(metaFieldsMode::equals).orElse(false)) {
Review Comment:
**[major] This consent gate is auto-satisfied on the hudi-cli path and
unsatisfiable on the Spark SQL path, so it never does what it was added to do.**
The premise stated at `:70-74` is that "almost no caller restates meta-field
settings. Restating it during a downgrade is therefore a deliberate signal
rather than something that happens by accident." Both halves of this PR make
that false:
**hudi-cli auto-consents.** `SparkMain.upgradeOrDowngradeTable`
(`SparkMain.java:561`) builds its config through the `getWriteConfig` added in
`a3c753d`, which unconditionally calls `withMetaFieldsModeOf(...)` ->
`builder.withMetaFieldsMode(metaClient.getTableConfig().getMetaFieldsMode())`
(`SparkMain.java:600-620`). The mode then survives the
`withProps(config.getProps())` copy at `SparkMain.java:570`. So
`statedMetaFieldsMode(config)` always equals the table's mode and this branch
is taken on **every** hudi-cli downgrade -- a selective table is silently
retained with only a `LOG.warn`, with the operator having asserted nothing.
**Spark SQL cannot consent.**
`UpgradeOrDowngradeProcedure.getWriteConfigWithTrue` (`:85-95`) builds a bare
config, and the procedure's only parameters are `table` and `to_version`
(`:40-50`). So `call downgrade_table` on a selective table always throws, and
the message tells the operator to "Set `hoodie.meta.fields.mode=X` on the
writer" -- an action that procedure gives them no way to perform.
Separately, the new strict-equality rule in `validateAgainstTableProperties`
already *requires* every writer against a non-`ALL` table to state the mode, so
"the writer restated it" cannot carry consent semantics anywhere.
The test that covers the throw mocks
`writeConfig.contains(META_FIELDS_MODE)` to `false`
(`TestTenToNineDowngradeHandler:79-81`), a state the CLI cannot produce, and
its comment asserts the now-false premise verbatim.
Please decouple consent from restatement -- e.g. an explicit
`hoodie.downgrade.allow.meta.fields.mode.retention` flag checked here, with
`SparkMain`'s auto-adopted mode tracked separately so it does not count -- and
add a `meta_fields_mode` parameter to `UpgradeOrDowngradeProcedure` so that
path can satisfy whatever the gate becomes. (Follow-on to the thread on `:72`,
where the write-back itself is settled.)
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestMetaFieldsModeE2E.java:
##########
@@ -0,0 +1,830 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.functional;
+
+import org.apache.hudi.DataSourceReadOptions;
+import org.apache.hudi.DataSourceWriteOptions;
+import org.apache.hudi.SparkAdapterSupport$;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.MetaFieldsMode;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.testutils.SparkClientFunctionalTestHarness;
+
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+import org.apache.spark.sql.SaveMode;
+import org.apache.spark.sql.functions;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Spark-datasource end-to-end tests for the {@code hoodie.meta.fields.mode}
property on CoW tables.
+ * Every {@link MetaFieldsMode} value is exercised via a write / re-read round
trip; on-disk column
+ * population is verified by reading the parquet files back and inspecting the
meta-column values.
+ */
+class TestMetaFieldsModeE2E extends SparkClientFunctionalTestHarness {
+
+ private static StructType simpleSchema() {
+ return DataTypes.createStructType(new StructField[]{
+ DataTypes.createStructField("column1", DataTypes.StringType, true),
+ DataTypes.createStructField("column2", DataTypes.StringType, true),
+ DataTypes.createStructField("column3", DataTypes.StringType, true)
+ }).asNullable();
+ }
+
+ private Map<String, String> baseOptions() {
+ Map<String, String> opts = new HashMap<>();
+ opts.put(DataSourceWriteOptions.RECORDKEY_FIELD().key(), "column1");
+ opts.put(DataSourceWriteOptions.PARTITIONPATH_FIELD().key(), "column2");
+ opts.put(DataSourceWriteOptions.ORDERING_FIELDS().key(), "column3");
+ opts.put(HoodieTableConfig.NAME.key(), "test_meta_fields_mode");
+ opts.put(DataSourceWriteOptions.TABLE_TYPE().key(), "COPY_ON_WRITE");
+ opts.put(HoodieMetadataConfig.ENABLE.key(), "false");
+ return opts;
+ }
+
+ private void writeRows(List<Row> records, StructType schema, Map<String,
String> options, String path, SaveMode mode) {
+ spark().createDataset(records,
+
SparkAdapterSupport$.MODULE$.sparkAdapter().getCatalystExpressionUtils().getEncoder(schema))
+ .write()
+ .format("hudi")
+ .options(options)
+ .mode(mode)
+ .save(path);
+ }
+
+ private HoodieTableConfig writeSampleAndGetTableConfig(Map<String, String>
options, String path) {
+ writeRows(Arrays.asList(
+ RowFactory.create("k1", "p1", "v1"),
+ RowFactory.create("k2", "p1", "v2")),
+ simpleSchema(), options, path, SaveMode.Overwrite);
+ HoodieTableMetaClient metaClient =
+
HoodieTableMetaClient.builder().setBasePath(path).setConf(storageConf()).build();
+ return metaClient.getTableConfig();
+ }
+
+ /**
+ * End-to-end assertion of the on-disk meta columns after a write. Reads the
parquet files back
+ * (bypassing Hudi's own read path so we see the raw column values) and
asserts which meta
+ * columns are non-null.
+ */
+ private void assertMetaColumnPopulation(String path, MetaFieldsMode
expectedMode) {
+ Dataset<Row> raw = spark().read().parquet(path + "/*/*.parquet");
+ Row first = raw.select(
+ HoodieRecord.COMMIT_TIME_METADATA_FIELD,
+ HoodieRecord.COMMIT_SEQNO_METADATA_FIELD,
+ HoodieRecord.RECORD_KEY_METADATA_FIELD,
+ HoodieRecord.PARTITION_PATH_METADATA_FIELD,
+ HoodieRecord.FILENAME_METADATA_FIELD).first();
+
+ if (expectedMode.isCommitTimePopulated()) {
+ assertNotNull(first.get(0), "expected _hoodie_commit_time to be
populated for mode " + expectedMode);
+ } else {
+ assertNull(first.get(0), "expected _hoodie_commit_time to be null for
mode " + expectedMode);
+ }
+ if (expectedMode.isFileNamePopulated()) {
+ assertNotNull(first.get(4), "expected _hoodie_file_name to be populated
for mode " + expectedMode);
+ } else {
+ assertNull(first.get(4), "expected _hoodie_file_name to be null for mode
" + expectedMode);
+ }
+ // Record key, partition path, and commit seq no are ALL-only.
+ if (expectedMode == MetaFieldsMode.ALL) {
+ assertNotNull(first.get(2), "record key must be populated in ALL mode");
+ assertNotNull(first.get(3), "partition path must be populated in ALL
mode");
+ assertNotNull(first.get(1), "commit seq no must be populated in ALL
mode");
+ } else {
+ assertNull(first.get(2), "record key must be null outside ALL mode, got:
" + first.get(2));
+ assertNull(first.get(3), "partition path must be null outside ALL mode,
got: " + first.get(3));
+ assertNull(first.get(1), "commit seq no must be null outside ALL mode,
got: " + first.get(1));
+ }
+ }
+
+ @Test
+ void allModePersistsAndPopulatesAllColumns() {
+ Map<String, String> options = baseOptions();
+ // ALL is the default; no need to set the mode explicitly.
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+
+ assertTrue(tc.populateMetaFields());
+ assertEquals(MetaFieldsMode.ALL, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.ALL);
+ }
+
+ @Test
+ void noneModePersistsAndLeavesAllColumnsNull() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+
+ assertFalse(tc.populateMetaFields());
+ assertEquals(MetaFieldsMode.NONE, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.NONE);
+ }
+
+ @Test
+ void commitTimeOnlyModePopulatesOnlyCommitTime() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.COMMIT_TIME_ONLY.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+
+ assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY.name(),
+ tc.getProps().getProperty(HoodieTableConfig.META_FIELDS_MODE.key()));
+ assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.COMMIT_TIME_ONLY);
+ }
+
+ @Test
+ void fileNameOnlyModePopulatesOnlyFileName() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.FILE_NAME_ONLY.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+
+ assertEquals(MetaFieldsMode.FILE_NAME_ONLY, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.FILE_NAME_ONLY);
+ }
+
+ @Test
+ void commitTimeAndFileNameModePopulatesBoth() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+
+ assertEquals(MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME,
tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(),
MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME);
+ }
+
+ @Test
+ void explicitlyContradictingTheModeIsRejectedAtTableCreation() {
+ // A selective mode implies populate.meta.fields=false. Stating the
boolean as true alongside it
+ // is a contradiction, and the user is told rather than having half their
request discarded.
+ // This is the datasource end of the check in
HoodieTableMetaClient.TableBuilder.
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "true");
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.COMMIT_TIME_ONLY.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ Throwable thrown = assertThrows(Throwable.class, () ->
+ writeSampleAndGetTableConfig(options, basePath()));
+
+ String rootMessage = rootMessageOf(thrown);
+ assertTrue(rootMessage.contains(HoodieTableConfig.META_FIELDS_MODE.key())
+ &&
rootMessage.contains(HoodieTableConfig.POPULATE_META_FIELDS.key()),
+ "the error must name both properties so the user knows which to drop,
got: " + rootMessage);
+ }
+
+ @Test
+ void selectiveModeWithoutTheLegacyBooleanDerivesItAsFalse() {
+ // The ordinary case: state only the mode. The boolean is derived, never
carried through
+ // verbatim -- a pre-1.3.0 reader ignores the mode property, so leaving
populate=true would make
+ // it treat a selectively-written table as ALL.
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.COMMIT_TIME_ONLY.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+
+ assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.COMMIT_TIME_ONLY);
+ assertFalse(tc.populateMetaFields(),
+ "legacy populate.meta.fields must be derived from the mode");
+ }
+
+ @Test
+ void noneModePersistsLegacyBooleanAsFalse() {
+ // The unsafe case this invariant protects: an old incremental reader that
saw
+ // populate.meta.fields=true on a NONE table would run against all-null
commit times and
+ // silently return zero rows. Stating only the mode -- the ordinary case
-- must derive false.
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.NONE.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+
+ assertEquals(MetaFieldsMode.NONE, tc.getMetaFieldsMode());
+ assertFalse(tc.populateMetaFields(),
+ "NONE must persist populate.meta.fields=false so pre-1.3.0 readers do
not treat it as ALL");
+ }
+
+ @Test
+ void allModePersistsLegacyBooleanAsTrue() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.ALL.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+
+ assertEquals(MetaFieldsMode.ALL, tc.getMetaFieldsMode());
+ assertTrue(tc.populateMetaFields(),
+ "ALL must persist populate.meta.fields=true for pre-1.3.0 readers");
+ }
+
+ @Test
+ void unknownModeValueIsRejected() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(), "SOMETHING_BOGUS");
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ Throwable thrown = assertThrows(Throwable.class, () ->
+ writeRows(Collections.singletonList(RowFactory.create("k1", "p1",
"v1")),
+ simpleSchema(), options, basePath(), SaveMode.Overwrite));
+
+ String rootMessage = rootMessageOf(thrown);
+ assertTrue(rootMessage.contains("SOMETHING_BOGUS"),
+ "Expected error to name the rejected value, got: " + rootMessage);
+ }
+
+ // -------------------------------------------------------------------------
+ // Non-row-writer path coverage. Bulk insert with row.writer.enable=false
forces the
+ // HoodieAvroParquetWriter path (via HoodieCreateHandle) instead of the
internal-row writer path.
+ // Both paths must respect the mode identically.
+ // -------------------------------------------------------------------------
+
+ @Test
+ void nonRowWriterPathAllMode() {
+ Map<String, String> options = baseOptions();
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL());
+ options.put("hoodie.datasource.write.row.writer.enable", "false");
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+ assertEquals(MetaFieldsMode.ALL, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.ALL);
+ }
+
+ @Test
+ void nonRowWriterPathNoneMode() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL());
+ options.put("hoodie.datasource.write.row.writer.enable", "false");
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+ assertEquals(MetaFieldsMode.NONE, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.NONE);
+ }
+
+ @Test
+ void nonRowWriterPathCommitTimeOnly() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.COMMIT_TIME_ONLY.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL());
+ options.put("hoodie.datasource.write.row.writer.enable", "false");
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+ assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.COMMIT_TIME_ONLY);
+ }
+
+ @Test
+ void nonRowWriterPathFileNameOnly() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.FILE_NAME_ONLY.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL());
+ options.put("hoodie.datasource.write.row.writer.enable", "false");
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+ assertEquals(MetaFieldsMode.FILE_NAME_ONLY, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.FILE_NAME_ONLY);
+ }
+
+ @Test
+ void nonRowWriterPathCommitTimeAndFileName() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL());
+ options.put("hoodie.datasource.write.row.writer.enable", "false");
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+ assertEquals(MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME,
tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(),
MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME);
+ }
+
+ // -------------------------------------------------------------------------
+ // Clustering coverage.
+ //
+ // These target 358fbfdd717a, where HoodieRowCreateHandle's selective path
copied the *source
+ // row's* _hoodie_file_name during clustering, leaving records pointing at a
file clustering had
+ // just replaced. asserting only assertNotNull cannot catch that — the stale
value is non-null too
+ // — so the assertion here compares the column against the file actually
holding the row.
+ //
+ // Only the selective modes are covered: ALL and NONE route through writeRow
/
+ // writeRowNoMetaFields and never enter the branch the fix touched.
+ // -------------------------------------------------------------------------
+
+ private Map<String, String> inlineClusteringOptions(MetaFieldsMode mode) {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(), mode.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+ options.put("hoodie.clustering.inline", "true");
+ options.put("hoodie.clustering.inline.max.commits", "1");
+ options.put("hoodie.clustering.plan.strategy.target.file.max.bytes",
"10485760");
+ options.put("hoodie.clustering.plan.strategy.small.file.limit",
"10485760");
+ return options;
+ }
+
+ /**
+ * Asserts clustering actually ran and that every surviving row's {@code
_hoodie_file_name} names
+ * the file holding it.
+ *
+ * <p>Reads through Hudi rather than globbing the parquet directly: after
inline clustering the
+ * pre-clustering file is still on disk (no cleaning has run), so a raw glob
would also inspect
+ * rows that were replaced and are no longer served.
+ */
+ private void assertClusteredFileNamesPointAtTheirOwnFile(String path,
MetaFieldsMode mode) {
+ HoodieTableMetaClient metaClient =
+
HoodieTableMetaClient.builder().setBasePath(path).setConf(storageConf()).build();
+ assertEquals(mode, metaClient.getTableConfig().getMetaFieldsMode());
+ assertEquals(1,
metaClient.getActiveTimeline().getCompletedReplaceTimeline().countInstants(),
+ "clustering must have produced a replacecommit, otherwise this test
proves nothing");
+
+ List<Row> rows = spark().read().format("hudi").load(path)
+ .withColumn("__containing_file", functions.input_file_name())
+ .collectAsList();
+ assertFalse(rows.isEmpty(), "expected the clustered table to still serve
rows");
+
+ for (Row row : rows) {
+ String fileName = row.getAs(HoodieRecord.FILENAME_METADATA_FIELD);
+ String containingFile = row.getAs("__containing_file").toString();
+ if (mode.isFileNamePopulated()) {
+ assertNotNull(fileName, "file name is opted in, so clustered rows must
carry one");
+ assertTrue(containingFile.endsWith("/" + fileName),
+ "_hoodie_file_name must name the file holding the row after
clustering, not the "
+ + "pre-clustering file it was read from; got " + fileName + "
inside " + containingFile);
+ } else {
+ assertNull(fileName,
+ "file name is not opted in, so clustering must not populate it;
got " + fileName);
+ }
+ }
+ }
+
+ @Test
+ void clusteringWritesTheNewFileNameUnderFileNameOnly() {
+ Map<String, String> options =
inlineClusteringOptions(MetaFieldsMode.FILE_NAME_ONLY);
+ writeRows(Arrays.asList(
+ RowFactory.create("k1", "p1", "v1"),
+ RowFactory.create("k2", "p1", "v2"),
+ RowFactory.create("k3", "p1", "v3")),
+ simpleSchema(), options, basePath(), SaveMode.Overwrite);
+
+ assertClusteredFileNamesPointAtTheirOwnFile(basePath(),
MetaFieldsMode.FILE_NAME_ONLY);
+ }
+
+ @Test
+ void clusteringWritesTheNewFileNameUnderCommitTimeAndFileName() {
+ Map<String, String> options =
inlineClusteringOptions(MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME);
+ writeRows(Arrays.asList(
+ RowFactory.create("k1", "p1", "v1"),
+ RowFactory.create("k2", "p1", "v2"),
+ RowFactory.create("k3", "p1", "v3")),
+ simpleSchema(), options, basePath(), SaveMode.Overwrite);
+
+ assertClusteredFileNamesPointAtTheirOwnFile(basePath(),
MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME);
+ }
+
+ @Test
+ void clusteringLeavesFileNameNullUnderCommitTimeOnly() {
+ Map<String, String> options =
inlineClusteringOptions(MetaFieldsMode.COMMIT_TIME_ONLY);
+ writeRows(Arrays.asList(
+ RowFactory.create("k1", "p1", "v1"),
+ RowFactory.create("k2", "p1", "v2"),
+ RowFactory.create("k3", "p1", "v3")),
+ simpleSchema(), options, basePath(), SaveMode.Overwrite);
+
+ // Also guards HoodieParquetBinaryCopyBase's unconditional file-name mask
from leaking into the
Review Comment:
**[major] This comment claims coverage the test does not provide, and the
path it names is still unfixed.**
`inlineClusteringOptions()` (`:369-379`) never sets
`hoodie.clustering.execution.strategy.class`, so this test runs the row-writer
clustering path and never reaches `HoodieParquetBinaryCopyBase`.
That path is untouched by this PR (`git diff apache/master...HEAD --stat --
hudi-hadoop-common/src/main/java/org/apache/hudi/parquet/io/` is empty) and
still masks the column unconditionally:
```java
// HoodieParquetBinaryCopyBase.java:132-134
// For meta column '_hoodie_file_name', rewriter will mask value with output
file name
maskColumns.put(ColumnPath.fromDotString(HoodieRecord.FILENAME_METADATA_FIELD),
maskValue);
```
So a `COMMIT_TIME_ONLY` or `NONE` table clustered with
`SparkBinaryCopyClusteringExecutionStrategy` gains a populated
`_hoodie_file_name`, contradicting the mode the table advertises. I raised the
production half in my review body on 07-28 and it was not picked up; the
misleading comment is new.
The test class that actually drives this strategy and asserts on
`_hoodie_file_name` already exists:
`TestSparkBinaryCopyClusteringAndValidationMeta.java:226`, strategy set at
`:349`.
Please gate the mask on the table's `isFileNamePopulated()` and add a
`COMMIT_TIME_ONLY` case to `TestSparkBinaryCopyClusteringAndValidationMeta`. If
you would rather defer the fix, delete this comment and state in
`MetaFieldsMode`'s javadoc that the binary-copy strategy is unsupported under
selective modes -- leaving a comment that claims a guard nothing enforces is
the worst of the three options.
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestMetaFieldsModeE2E.java:
##########
@@ -0,0 +1,830 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.functional;
+
+import org.apache.hudi.DataSourceReadOptions;
+import org.apache.hudi.DataSourceWriteOptions;
+import org.apache.hudi.SparkAdapterSupport$;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.MetaFieldsMode;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.testutils.SparkClientFunctionalTestHarness;
+
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+import org.apache.spark.sql.SaveMode;
+import org.apache.spark.sql.functions;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Spark-datasource end-to-end tests for the {@code hoodie.meta.fields.mode}
property on CoW tables.
+ * Every {@link MetaFieldsMode} value is exercised via a write / re-read round
trip; on-disk column
+ * population is verified by reading the parquet files back and inspecting the
meta-column values.
+ */
+class TestMetaFieldsModeE2E extends SparkClientFunctionalTestHarness {
+
+ private static StructType simpleSchema() {
+ return DataTypes.createStructType(new StructField[]{
+ DataTypes.createStructField("column1", DataTypes.StringType, true),
+ DataTypes.createStructField("column2", DataTypes.StringType, true),
+ DataTypes.createStructField("column3", DataTypes.StringType, true)
+ }).asNullable();
+ }
+
+ private Map<String, String> baseOptions() {
+ Map<String, String> opts = new HashMap<>();
+ opts.put(DataSourceWriteOptions.RECORDKEY_FIELD().key(), "column1");
+ opts.put(DataSourceWriteOptions.PARTITIONPATH_FIELD().key(), "column2");
+ opts.put(DataSourceWriteOptions.ORDERING_FIELDS().key(), "column3");
+ opts.put(HoodieTableConfig.NAME.key(), "test_meta_fields_mode");
+ opts.put(DataSourceWriteOptions.TABLE_TYPE().key(), "COPY_ON_WRITE");
+ opts.put(HoodieMetadataConfig.ENABLE.key(), "false");
+ return opts;
+ }
+
+ private void writeRows(List<Row> records, StructType schema, Map<String,
String> options, String path, SaveMode mode) {
+ spark().createDataset(records,
+
SparkAdapterSupport$.MODULE$.sparkAdapter().getCatalystExpressionUtils().getEncoder(schema))
+ .write()
+ .format("hudi")
+ .options(options)
+ .mode(mode)
+ .save(path);
+ }
+
+ private HoodieTableConfig writeSampleAndGetTableConfig(Map<String, String>
options, String path) {
+ writeRows(Arrays.asList(
+ RowFactory.create("k1", "p1", "v1"),
+ RowFactory.create("k2", "p1", "v2")),
+ simpleSchema(), options, path, SaveMode.Overwrite);
+ HoodieTableMetaClient metaClient =
+
HoodieTableMetaClient.builder().setBasePath(path).setConf(storageConf()).build();
+ return metaClient.getTableConfig();
+ }
+
+ /**
+ * End-to-end assertion of the on-disk meta columns after a write. Reads the
parquet files back
+ * (bypassing Hudi's own read path so we see the raw column values) and
asserts which meta
+ * columns are non-null.
+ */
+ private void assertMetaColumnPopulation(String path, MetaFieldsMode
expectedMode) {
+ Dataset<Row> raw = spark().read().parquet(path + "/*/*.parquet");
+ Row first = raw.select(
+ HoodieRecord.COMMIT_TIME_METADATA_FIELD,
+ HoodieRecord.COMMIT_SEQNO_METADATA_FIELD,
+ HoodieRecord.RECORD_KEY_METADATA_FIELD,
+ HoodieRecord.PARTITION_PATH_METADATA_FIELD,
+ HoodieRecord.FILENAME_METADATA_FIELD).first();
+
+ if (expectedMode.isCommitTimePopulated()) {
+ assertNotNull(first.get(0), "expected _hoodie_commit_time to be
populated for mode " + expectedMode);
+ } else {
+ assertNull(first.get(0), "expected _hoodie_commit_time to be null for
mode " + expectedMode);
+ }
+ if (expectedMode.isFileNamePopulated()) {
+ assertNotNull(first.get(4), "expected _hoodie_file_name to be populated
for mode " + expectedMode);
+ } else {
+ assertNull(first.get(4), "expected _hoodie_file_name to be null for mode
" + expectedMode);
+ }
+ // Record key, partition path, and commit seq no are ALL-only.
+ if (expectedMode == MetaFieldsMode.ALL) {
+ assertNotNull(first.get(2), "record key must be populated in ALL mode");
+ assertNotNull(first.get(3), "partition path must be populated in ALL
mode");
+ assertNotNull(first.get(1), "commit seq no must be populated in ALL
mode");
+ } else {
+ assertNull(first.get(2), "record key must be null outside ALL mode, got:
" + first.get(2));
+ assertNull(first.get(3), "partition path must be null outside ALL mode,
got: " + first.get(3));
+ assertNull(first.get(1), "commit seq no must be null outside ALL mode,
got: " + first.get(1));
+ }
+ }
+
+ @Test
+ void allModePersistsAndPopulatesAllColumns() {
+ Map<String, String> options = baseOptions();
+ // ALL is the default; no need to set the mode explicitly.
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+
+ assertTrue(tc.populateMetaFields());
+ assertEquals(MetaFieldsMode.ALL, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.ALL);
+ }
+
+ @Test
+ void noneModePersistsAndLeavesAllColumnsNull() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+
+ assertFalse(tc.populateMetaFields());
+ assertEquals(MetaFieldsMode.NONE, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.NONE);
+ }
+
+ @Test
+ void commitTimeOnlyModePopulatesOnlyCommitTime() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.COMMIT_TIME_ONLY.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+
+ assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY.name(),
+ tc.getProps().getProperty(HoodieTableConfig.META_FIELDS_MODE.key()));
+ assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.COMMIT_TIME_ONLY);
+ }
+
+ @Test
+ void fileNameOnlyModePopulatesOnlyFileName() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.FILE_NAME_ONLY.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+
+ assertEquals(MetaFieldsMode.FILE_NAME_ONLY, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.FILE_NAME_ONLY);
+ }
+
+ @Test
+ void commitTimeAndFileNameModePopulatesBoth() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+
+ assertEquals(MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME,
tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(),
MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME);
+ }
+
+ @Test
+ void explicitlyContradictingTheModeIsRejectedAtTableCreation() {
+ // A selective mode implies populate.meta.fields=false. Stating the
boolean as true alongside it
+ // is a contradiction, and the user is told rather than having half their
request discarded.
+ // This is the datasource end of the check in
HoodieTableMetaClient.TableBuilder.
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "true");
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.COMMIT_TIME_ONLY.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ Throwable thrown = assertThrows(Throwable.class, () ->
+ writeSampleAndGetTableConfig(options, basePath()));
+
+ String rootMessage = rootMessageOf(thrown);
+ assertTrue(rootMessage.contains(HoodieTableConfig.META_FIELDS_MODE.key())
+ &&
rootMessage.contains(HoodieTableConfig.POPULATE_META_FIELDS.key()),
+ "the error must name both properties so the user knows which to drop,
got: " + rootMessage);
+ }
+
+ @Test
+ void selectiveModeWithoutTheLegacyBooleanDerivesItAsFalse() {
+ // The ordinary case: state only the mode. The boolean is derived, never
carried through
+ // verbatim -- a pre-1.3.0 reader ignores the mode property, so leaving
populate=true would make
+ // it treat a selectively-written table as ALL.
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.COMMIT_TIME_ONLY.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+
+ assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.COMMIT_TIME_ONLY);
+ assertFalse(tc.populateMetaFields(),
+ "legacy populate.meta.fields must be derived from the mode");
+ }
+
+ @Test
+ void noneModePersistsLegacyBooleanAsFalse() {
+ // The unsafe case this invariant protects: an old incremental reader that
saw
+ // populate.meta.fields=true on a NONE table would run against all-null
commit times and
+ // silently return zero rows. Stating only the mode -- the ordinary case
-- must derive false.
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.NONE.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+
+ assertEquals(MetaFieldsMode.NONE, tc.getMetaFieldsMode());
+ assertFalse(tc.populateMetaFields(),
+ "NONE must persist populate.meta.fields=false so pre-1.3.0 readers do
not treat it as ALL");
+ }
+
+ @Test
+ void allModePersistsLegacyBooleanAsTrue() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.ALL.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+
+ assertEquals(MetaFieldsMode.ALL, tc.getMetaFieldsMode());
+ assertTrue(tc.populateMetaFields(),
+ "ALL must persist populate.meta.fields=true for pre-1.3.0 readers");
+ }
+
+ @Test
+ void unknownModeValueIsRejected() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(), "SOMETHING_BOGUS");
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+
+ Throwable thrown = assertThrows(Throwable.class, () ->
+ writeRows(Collections.singletonList(RowFactory.create("k1", "p1",
"v1")),
+ simpleSchema(), options, basePath(), SaveMode.Overwrite));
+
+ String rootMessage = rootMessageOf(thrown);
+ assertTrue(rootMessage.contains("SOMETHING_BOGUS"),
+ "Expected error to name the rejected value, got: " + rootMessage);
+ }
+
+ // -------------------------------------------------------------------------
+ // Non-row-writer path coverage. Bulk insert with row.writer.enable=false
forces the
+ // HoodieAvroParquetWriter path (via HoodieCreateHandle) instead of the
internal-row writer path.
+ // Both paths must respect the mode identically.
+ // -------------------------------------------------------------------------
+
+ @Test
+ void nonRowWriterPathAllMode() {
+ Map<String, String> options = baseOptions();
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL());
+ options.put("hoodie.datasource.write.row.writer.enable", "false");
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+ assertEquals(MetaFieldsMode.ALL, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.ALL);
+ }
+
+ @Test
+ void nonRowWriterPathNoneMode() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL());
+ options.put("hoodie.datasource.write.row.writer.enable", "false");
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+ assertEquals(MetaFieldsMode.NONE, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.NONE);
+ }
+
+ @Test
+ void nonRowWriterPathCommitTimeOnly() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.COMMIT_TIME_ONLY.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL());
+ options.put("hoodie.datasource.write.row.writer.enable", "false");
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+ assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.COMMIT_TIME_ONLY);
+ }
+
+ @Test
+ void nonRowWriterPathFileNameOnly() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.FILE_NAME_ONLY.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL());
+ options.put("hoodie.datasource.write.row.writer.enable", "false");
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+ assertEquals(MetaFieldsMode.FILE_NAME_ONLY, tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(), MetaFieldsMode.FILE_NAME_ONLY);
+ }
+
+ @Test
+ void nonRowWriterPathCommitTimeAndFileName() {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false");
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(),
MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL());
+ options.put("hoodie.datasource.write.row.writer.enable", "false");
+
+ HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath());
+ assertEquals(MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME,
tc.getMetaFieldsMode());
+ assertMetaColumnPopulation(basePath(),
MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME);
+ }
+
+ // -------------------------------------------------------------------------
+ // Clustering coverage.
+ //
+ // These target 358fbfdd717a, where HoodieRowCreateHandle's selective path
copied the *source
+ // row's* _hoodie_file_name during clustering, leaving records pointing at a
file clustering had
+ // just replaced. asserting only assertNotNull cannot catch that — the stale
value is non-null too
+ // — so the assertion here compares the column against the file actually
holding the row.
+ //
+ // Only the selective modes are covered: ALL and NONE route through writeRow
/
+ // writeRowNoMetaFields and never enter the branch the fix touched.
+ // -------------------------------------------------------------------------
+
+ private Map<String, String> inlineClusteringOptions(MetaFieldsMode mode) {
+ Map<String, String> options = baseOptions();
+ options.put(HoodieTableConfig.META_FIELDS_MODE.key(), mode.name());
+ options.put(DataSourceWriteOptions.OPERATION().key(),
DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL());
+ options.put("hoodie.clustering.inline", "true");
+ options.put("hoodie.clustering.inline.max.commits", "1");
+ options.put("hoodie.clustering.plan.strategy.target.file.max.bytes",
"10485760");
+ options.put("hoodie.clustering.plan.strategy.small.file.limit",
"10485760");
+ return options;
+ }
+
+ /**
+ * Asserts clustering actually ran and that every surviving row's {@code
_hoodie_file_name} names
+ * the file holding it.
+ *
+ * <p>Reads through Hudi rather than globbing the parquet directly: after
inline clustering the
+ * pre-clustering file is still on disk (no cleaning has run), so a raw glob
would also inspect
+ * rows that were replaced and are no longer served.
+ */
+ private void assertClusteredFileNamesPointAtTheirOwnFile(String path,
MetaFieldsMode mode) {
+ HoodieTableMetaClient metaClient =
+
HoodieTableMetaClient.builder().setBasePath(path).setConf(storageConf()).build();
+ assertEquals(mode, metaClient.getTableConfig().getMetaFieldsMode());
+ assertEquals(1,
metaClient.getActiveTimeline().getCompletedReplaceTimeline().countInstants(),
+ "clustering must have produced a replacecommit, otherwise this test
proves nothing");
+
+ List<Row> rows = spark().read().format("hudi").load(path)
+ .withColumn("__containing_file", functions.input_file_name())
+ .collectAsList();
+ assertFalse(rows.isEmpty(), "expected the clustered table to still serve
rows");
+
+ for (Row row : rows) {
+ String fileName = row.getAs(HoodieRecord.FILENAME_METADATA_FIELD);
+ String containingFile = row.getAs("__containing_file").toString();
+ if (mode.isFileNamePopulated()) {
+ assertNotNull(fileName, "file name is opted in, so clustered rows must
carry one");
+ assertTrue(containingFile.endsWith("/" + fileName),
+ "_hoodie_file_name must name the file holding the row after
clustering, not the "
+ + "pre-clustering file it was read from; got " + fileName + "
inside " + containingFile);
+ } else {
+ assertNull(fileName,
+ "file name is not opted in, so clustering must not populate it;
got " + fileName);
+ }
+ }
+ }
+
+ @Test
+ void clusteringWritesTheNewFileNameUnderFileNameOnly() {
+ Map<String, String> options =
inlineClusteringOptions(MetaFieldsMode.FILE_NAME_ONLY);
+ writeRows(Arrays.asList(
+ RowFactory.create("k1", "p1", "v1"),
+ RowFactory.create("k2", "p1", "v2"),
+ RowFactory.create("k3", "p1", "v3")),
+ simpleSchema(), options, basePath(), SaveMode.Overwrite);
+
+ assertClusteredFileNamesPointAtTheirOwnFile(basePath(),
MetaFieldsMode.FILE_NAME_ONLY);
+ }
+
+ @Test
+ void clusteringWritesTheNewFileNameUnderCommitTimeAndFileName() {
+ Map<String, String> options =
inlineClusteringOptions(MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME);
+ writeRows(Arrays.asList(
+ RowFactory.create("k1", "p1", "v1"),
+ RowFactory.create("k2", "p1", "v2"),
+ RowFactory.create("k3", "p1", "v3")),
+ simpleSchema(), options, basePath(), SaveMode.Overwrite);
+
+ assertClusteredFileNamesPointAtTheirOwnFile(basePath(),
MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME);
+ }
+
+ @Test
+ void clusteringLeavesFileNameNullUnderCommitTimeOnly() {
Review Comment:
**[major] The commit-time preservation branch cannot fail here -- this is
the same bug class `358fbfdd717a` fixed for `_hoodie_file_name` on this
branch.**
`HoodieRowCreateHandle.writeRowSelectiveMetaFields` preserves the source
row's commit time during clustering:
```java
metaFields[COMMIT_TIME_METADATA_FIELD_ORD] = shouldPreserveHoodieMetadata
? row.getUTF8String(COMMIT_TIME_METADATA_FIELD_ORD) : commitTime;
```
Drop the ternary and always stamp `commitTime` (the replacecommit instant)
and all three clustering tests stay green, because the only commit-time
assertion they reach is `assertNotNull(first.get(0))` inside
`assertMetaColumnPopulation` -- and the replacecommit instant is also non-null.
That is exactly the shape of `358fbfdd717a`, which fixed row-writer
clustering copying the wrong `_hoodie_file_name`. The file-name assertion was
strengthened in response (`assertClusteredFileNamesPointAtTheirOwnFile`); the
commit-time one was not, so clustering silently rewriting every row's commit
time -- which breaks incremental queries, the feature's entire premise -- would
ship undetected.
The repo already has the right shape at
`HoodieWriterClientTestHarness.verifyRecordsWrittenWithPreservedMetadata`
(`:490-492`), which groups by commit time and asserts it is one of the original
insert commits.
Please capture the insert instant before clustering and assert every
post-clustering row's `_hoodie_commit_time` equals it and is not the
replacecommit instant.
--
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]
