nsivabalan commented on code in PR #19205:
URL: https://github.com/apache/hudi/pull/19205#discussion_r3782120899
##########
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:
You were right that the gate could not do its job, and the discussion it
prompted produced a better rule than the one I would have written.
Both breakages were mine, from changes made after the offline call that set
the restatement rule: `a3c753d` made hudi-cli adopt the table's mode
unconditionally (so it always "restated"), and the strict-equality rule in
`validateAgainstTableProperties` made restatement mandatory everywhere --
which, as you say, means it cannot carry consent semantics at all.
Rather than track auto-adoption separately, 83451a91 keys the decision on
`hoodie.table.initial.version`, which is recorded at creation and no write path
can produce accidentally:
- **created before v10** -- retain the mode and downgrade, no flag. The
table predates the mode, so returning it to a version that predates it is not a
new state for it.
- **created at v10 or later** -- throw unless
`hoodie.downgrade.allow.meta.fields.mode.retention=true`. Such a table has
never been read by an older version, so nothing outside the deployment has had
to honor the mode.
That keeps your flag, but only where it can mean something, and it drops the
need for a `meta_fields_mode` parameter on `UpgradeOrDowngradeProcedure` --
Spark SQL now satisfies the common case without one.
`TestTenToNineDowngradeHandler` rewritten accordingly: the mock no longer
stubs `contains(META_FIELDS_MODE)` to a state the CLI cannot produce, and the
three rule cases are keyed on initial version plus the flag. 17/17.
##########
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:
Confirmed and fixed in 10f9940b. I traced each link: `enableBloomFilter`
short-circuits on `populateMetaFields`, `readBloomFilterFromMetadata` returns
null when the footer key is absent, and `addKey:83` calls `mightContain` with
no null check. With BLOOM as the default index type, a `COMMIT_TIME_ONLY` table
created with defaults NPEs on its first upsert.
Your point about why this is newly reachable rather than newly broken is the
part that makes it worth fixing here: the mechanism is identical under
`populate.meta.fields=false`, but that setting is documented for append-only
data, so nobody paired it with mutation. `COMMIT_TIME_ONLY` targets tables that
keep taking writes.
Rejecting BLOOM and GLOBAL_BLOOM at init, next to the existing key-generator
restriction, so the failure is attributable rather than a
`NullPointerException` mid-write.
On the `TestHoodieIndex.indexTypeParams` row: I did not add `{BLOOM, false,
false}` there. That parameterization feeds functional tests that write and tag
records, so the row would abort in `setUp` rather than assert the rejection.
Put it in `TestBaseHoodieWriteClient` instead --
`validateAgainstTablePropertiesRejectsABloomIndexWithoutTheRecordKey` for
BLOOM/GLOBAL_BLOOM, plus an accept case for SIMPLE/GLOBAL_SIMPLE so the
restriction is pinned as bloom-specific rather than "selective tables cannot
have an index". Happy to add the parameterization row too if you would rather
have it in both places.
##########
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:
Both halves confirmed and fixed.
**Production** (2a802c21): the mask is now installed only when the table's
mode populates the column. The mode reaches `HoodieParquetBinaryCopyBase` from
`HoodieBinaryCopyHandle`, which already holds the table, via a defaulted setter
on `HoodieFileBinaryCopier`.
**Test** (583a089c): added
`testStreamCopyClusteringLeavesFileNameNullUnderCommitTimeOnly` to
`TestSparkBinaryCopyClusteringAndValidationMeta`, driving
`SPARK_STREAM_COPY_CLUSTERING_EXECUTION_STRATEGY` as you pointed out. Verified
non-vacuous -- removing the gate fails with `expected: <0> but was: <60>`, so
every row gains a file name the table never advertised.
You were also right that the comment was the worst part. It claimed the
row-writer clustering test guarded this path when that test never sets an
execution strategy, so it never reaches the binary copier. Deleted.
While there I found a second comment of mine making the same kind of
unfounded claim: `HoodieAvroParquetWriter` said bloom population is skipped
because it "requires the record-key column". It does not -- the ALL branch
takes the key from the in-memory `HoodieKey`. The real reason is that
`enableBloomFilter` short-circuits on `populateMetaFields`. Corrected.
--
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]
