hudi-agent commented on code in PR #19205:
URL: https://github.com/apache/hudi/pull/19205#discussion_r3778669403
##########
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)) {
+ // Selective mode, and the writer restated it. That restatement is the
operator asserting
+ // "I know this table is selective and every reader of it honors the
mode" -- so the mode is
+ // retained on the downgraded table rather than dropped. Two
consequences they are taking on:
+ // the property outlives the version that formally understands it
(harmless -- it carries no
+ // sinceVersion, so it is not stripped on load), and a reader that
ignores it sees
+ // populate.meta.fields=false and treats the table as NONE.
+ //
+ // Keeping it is also what makes the round trip lossless: a later
re-upgrade finds the mode
+ // intact instead of deriving NONE from the boolean.
+ LOG.warn("Downgrading a table on {}={} to table version 9. The mode is
being retained on the "
+ + "downgraded table because it was explicitly restated on the
writer. Version 9 does "
+ + "not formally understand it, so every reader of this table
must honor {} rather "
+ + "than relying on {} alone -- which now reads false.",
+ HoodieTableConfig.META_FIELDS_MODE.key(), metaFieldsMode,
+ HoodieTableConfig.META_FIELDS_MODE.key(),
+ HoodieTableConfig.POPULATE_META_FIELDS.key());
+ } else {
+ // Selective mode and the writer did not restate it (or restated a
different one). Dropping
+ // the mode here would silently collapse the table to NONE and could
not be undone: a
Review Comment:
🤖 nit: the four-line comment starting with "No helper means the table config
is unreachable" appears twice — once inside the `else` block and once again
immediately after its closing `}`, right before the `return`. The second copy
looks like a paste left-over; could you drop it?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/model/MetaFieldsMode.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.common.model;
+
+import org.apache.hudi.common.config.HoodieConfig;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.util.StringUtils;
+
+import java.util.Locale;
+
+/**
+ * Which of Hudi's meta columns are physically populated on disk.
+ *
+ * <p>Selective modes exist so that tables that opt out of the default {@code
populate.meta.fields=true}
+ * can still keep the two columns that matter for downstream operations
without paying for the other
+ * three:
+ *
+ * <ul>
+ * <li>{@code _hoodie_commit_time} — required for incremental queries.</li>
+ * <li>{@code _hoodie_file_name} — useful for file-level pruning /
investigation lookups.</li>
+ * </ul>
+ *
+ * <p>The remaining three meta columns ({@code _hoodie_commit_seqno}, {@code
_hoodie_record_key},
+ * {@code _hoodie_partition_path}) are all-or-nothing — either populate every
meta column ({@link #ALL})
+ * or none of them beyond the two selectable ones. If you need any of the
remaining columns, set
+ * {@code hoodie.populate.meta.fields=true}.
+ *
+ * <p>This enum is the single source of truth for meta-column population. The
legacy boolean
+ * {@code hoodie.populate.meta.fields} is deprecated and consulted only when
+ * {@code hoodie.meta.fields.mode} is absent, so that tables written before
the mode property
+ * existed keep their behavior:
+ *
+ * <ul>
+ * <li>{@code populate.meta.fields=true} (or absent) → {@link #ALL} —
today's default.</li>
+ * <li>{@code populate.meta.fields=false} → {@link #NONE}.</li>
+ * </ul>
+ *
+ * <p>On-disk representation: the enum {@link #name()} is persisted in {@code
hoodie.properties}
+ * under the property {@code hoodie.meta.fields.mode}.
+ */
+public enum MetaFieldsMode {
+ /**
+ * All five Hudi meta columns are populated — today's default.
+ */
+ ALL(true, true),
+
+ /**
+ * No Hudi meta columns are populated. Incremental queries are unsupported.
File-level pruning
+ * that depends on {@code _hoodie_file_name} is unsupported.
+ */
+ NONE(false, false),
+
+ /**
+ * Only {@code _hoodie_commit_time} is populated. Incremental queries remain
functional; other
+ * meta columns stay null on disk.
+ */
+ COMMIT_TIME_ONLY(true, false),
+
+ /**
+ * Only {@code _hoodie_file_name} is populated. Useful for file-level
lookups and debugging;
+ * incremental queries are unsupported.
+ */
+ FILE_NAME_ONLY(false, true),
+
+ /**
+ * Both {@code _hoodie_commit_time} and {@code _hoodie_file_name} are
populated.
+ */
+ COMMIT_TIME_AND_FILE_NAME(true, true);
+
+ private final boolean commitTimePopulated;
+ private final boolean fileNamePopulated;
+
+ MetaFieldsMode(boolean commitTimePopulated, boolean fileNamePopulated) {
+ this.commitTimePopulated = commitTimePopulated;
+ this.fileNamePopulated = fileNamePopulated;
+ }
+
+ public boolean isCommitTimePopulated() {
+ return commitTimePopulated;
+ }
+
+ public boolean isFileNamePopulated() {
+ return fileNamePopulated;
+ }
+
+ /**
+ * @return true when all five meta columns are populated (i.e. this is
{@link #ALL}). Selective
+ * modes never populate {@code _hoodie_record_key}, {@code
_hoodie_partition_path}, or
+ * {@code _hoodie_commit_seqno}.
+ */
+ public boolean isRecordKeyPopulated() {
+ return this == ALL;
+ }
+
+ /**
+ * @return true for the modes that populate some but not all meta columns,
i.e. everything except
+ * {@link #ALL} and {@link #NONE}.
+ *
+ * <p>These are the modes the deprecated {@code hoodie.populate.meta.fields}
boolean cannot
+ * express, so they are what callers gate on when a code path only
understands all-or-nothing meta
+ * fields — writer engines not yet wired for selective population, table
versions that predate the
+ * mode property, and validation that must not let a two-state writer speak
for a five-state table.
+ */
+ public boolean isSelective() {
+ return this != ALL && this != NONE;
+ }
+
+ /**
+ * Resolve the effective mode from any {@link HoodieConfig} that may carry
the two properties —
+ * a table config, a write config, or a bare config built from write
options. Preferred over the
+ * two-argument overload: it keeps the property keys and the precedence rule
in one place instead
+ * of repeating them at every call site.
+ */
+ public static MetaFieldsMode resolve(HoodieConfig config) {
+ return
resolve(config.getStringOrDefault(HoodieTableConfig.META_FIELDS_MODE),
+ config.getBooleanOrDefault(HoodieTableConfig.POPULATE_META_FIELDS));
+ }
+
+ /**
+ * Resolve the effective mode. {@code hoodie.meta.fields.mode} is the source
of truth; the
+ * deprecated {@code hoodie.populate.meta.fields} boolean is a fallback for
tables written before
+ * the mode property existed. Precedence:
+ *
+ * <ul>
+ * <li>non-empty mode → the parsed enum value (the legacy boolean is not
consulted).</li>
+ * <li>null/empty mode + {@code populateMetaFields=false} → {@link
#NONE}.</li>
+ * <li>null/empty mode + {@code populateMetaFields=true} → {@link
#ALL}.</li>
+ * </ul>
+ *
+ * @param rawMode raw {@code hoodie.meta.fields.mode} value; may
be null or empty.
+ * @param legacyPopulateMetaFields value of the deprecated {@code
hoodie.populate.meta.fields}.
+ * @throws IllegalArgumentException when the raw mode value does not match
any enum value. This
+ * includes the pre-enum comma-separated format — callers that
upgrade an old table must
+ * migrate the value through the hudi-cli.
+ */
+ public static MetaFieldsMode resolve(String rawMode, boolean
legacyPopulateMetaFields) {
+ if (StringUtils.isNullOrEmpty(rawMode)) {
+ return legacyPopulateMetaFields ? ALL : NONE;
+ }
+ return parse(rawMode);
+ }
+
+ /**
+ * Resolve the mode from a table config that may not be able to answer,
defaulting to {@link #ALL}.
+ *
+ * <p>For write handles, which read the mode from the table rather than the
write config. A real
+ * {@link
org.apache.hudi.common.table.HoodieTableConfig#getMetaFieldsMode()} never
returns null --
+ * it falls back through the deprecated boolean to {@code ALL}. But a handle
constructed against a
+ * partially-stubbed table (as several unit tests do) would otherwise take a
null here and NPE later,
+ * at the point of use, far from the cause. {@code ALL} is the safe default:
it is the pre-feature
+ * behavior, so a caller that cannot state a mode gets what it would have
got before this existed.
+ */
+ public static MetaFieldsMode orAllIfUnknown(MetaFieldsMode mode) {
+ return mode == null ? ALL : mode;
+ }
+
+ /**
Review Comment:
🤖 nit: the five enum names are hardcoded in the format string — `"%s, %s,
%s, %s, %s"` with explicit `ALL, NONE, COMMIT_TIME_ONLY, FILE_NAME_ONLY,
COMMIT_TIME_AND_FILE_NAME`. If a sixth mode is added later the error message
silently goes stale. Have you considered
`Arrays.stream(values()).map(Enum::name).collect(Collectors.joining(", "))` to
keep it self-maintaining?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieStreamerMetaFieldsMode.java:
##########
@@ -0,0 +1,285 @@
+/*
+ * 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.utilities.deltastreamer;
+
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.MetaFieldsMode;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.functions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+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;
+
+/**
+ * End-to-end coverage for {@code hoodie.meta.fields.mode} through the
HoodieStreamer entrypoint.
+ * Each parameterized invocation runs a single ingest cycle in the given
{@link MetaFieldsMode} and
+ * verifies both the persisted table property and the actual on-disk parquet
column population.
+ *
+ * <p>Rejection paths (unknown token, populate=true+mode, MoR+mode) are
exercised in the datasource
+ * test {@code TestMetaFieldsMode}; this fixture focuses on the streamer
control-flow.
+ */
+public class TestHoodieStreamerMetaFieldsMode extends
HoodieDeltaStreamerTestBase {
+
+ /**
+ * Only the selective modes are parameterized here. ALL and NONE add no mode
key at all, so they
+ * exercise none of the streamer-side plumbing this feature introduced —
they are covered by the
+ * datasource tests and by {@code TestHoodieTableConfig}'s resolution cases.
+ */
+ @ParameterizedTest
+ @EnumSource(value = MetaFieldsMode.class,
+ names = {"COMMIT_TIME_ONLY", "FILE_NAME_ONLY",
"COMMIT_TIME_AND_FILE_NAME"})
+ public void testStreamerRespectsMetaFieldsMode(MetaFieldsMode mode) throws
Exception {
+ String tablePath = basePath + "/streamer_meta_fields_mode_" + mode.name();
+ HoodieDeltaStreamer.Config cfg = TestHelpers.makeConfig(tablePath,
WriteOperationType.INSERT);
+ // Force CoW; selective modes are CoW-only until MoR log-write is wired.
+ cfg.tableType = "COPY_ON_WRITE";
+ // The mode alone — pairing it with populate.meta.fields would now be a
stated conflict, since the
+ // mode is authoritative and the boolean is only the fallback for
resolving an absent one.
+ cfg.configs.add(HoodieTableConfig.META_FIELDS_MODE.key() + "=" +
mode.name());
+ HoodieDeltaStreamer streamer = new HoodieDeltaStreamer(cfg, jsc);
+ streamer.getIngestionService().ingestOnce();
+ streamer.shutdownGracefully();
+
+ HoodieTableMetaClient metaClient =
HoodieTestUtils.createMetaClient(context, tablePath);
+ assertEquals(mode, metaClient.getTableConfig().getMetaFieldsMode(),
+ "streamer must persist mode=" + mode + " on hoodie.properties");
+ assertOnDiskMetaColumns(tablePath, mode);
+ }
+
+ /**
+ * The regression this fixture exists for, and the one cshuo raised in
review: a restarted streamer
+ * that does not restate the mode must not silently write the wrong meta
columns.
+ *
+ * <p>StreamSync builds its write config from {@code props} alone, and the
mode is persisted only by
+ * {@code initializeEmptyTable}, which runs solely when the base path does
not exist. So a second run
+ * used to resolve to {@link MetaFieldsMode#NONE} and write base files with
a null
+ * {@code _hoodie_commit_time} while {@code hoodie.properties} still
advertised
+ * {@code COMMIT_TIME_ONLY} — incremental queries were then admitted and
silently dropped every one
+ * of those rows.
+ *
+ * <p>A selective table now requires the writer to state the mode, so the
restart is <b>rejected</b>
+ * rather than inheriting. Inheritance would work here, but it cannot be
relied on everywhere: the
+ * write path reads the mode in factories and handles that hold no table
config at all, so the write
+ * config has to be right on its own. Failing loudly at init is what makes
that guarantee real —
+ * and a rejected run leaves the table exactly as it was, which silent
narrowing did not.
+ * See {@code testRestartRestatingTheModeSucceeds} for the migration path.
+ */
+ @Test
+ public void testRestartWithoutRestatingTheModeIsRejected() throws Exception {
+ String tablePath = basePath + "/streamer_restart_inherits_mode";
+
+ HoodieDeltaStreamer.Config first = TestHelpers.makeConfig(tablePath,
WriteOperationType.INSERT);
+ first.tableType = "COPY_ON_WRITE";
+ first.configs.add(HoodieTableConfig.META_FIELDS_MODE.key() + "=" +
MetaFieldsMode.COMMIT_TIME_ONLY.name());
+ HoodieDeltaStreamer streamer = new HoodieDeltaStreamer(first, jsc);
+ streamer.getIngestionService().ingestOnce();
+ streamer.shutdownGracefully();
+
+ assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY,
+ HoodieTestUtils.createMetaClient(context,
tablePath).getTableConfig().getMetaFieldsMode());
+
+ long commitsAfterFirstRun = HoodieTestUtils.createMetaClient(context,
tablePath)
+ .getActiveTimeline().filterCompletedInstants().countInstants();
+
+ // Restart against the existing table stating neither the mode nor the
legacy boolean.
+ HoodieDeltaStreamer.Config restart = TestHelpers.makeConfig(tablePath,
WriteOperationType.INSERT);
+ restart.tableType = "COPY_ON_WRITE";
+
+ Throwable thrown = assertThrows(Throwable.class, () -> {
+ HoodieDeltaStreamer restarted = new HoodieDeltaStreamer(restart, jsc);
+ restarted.getIngestionService().ingestOnce();
+ restarted.shutdownGracefully();
+ });
+
+ String rootMessage = rootMessageOf(thrown);
+ assertTrue(rootMessage.contains(HoodieTableConfig.META_FIELDS_MODE.key()),
+ "expected the writer to be told to state the mode, got: " +
rootMessage);
+
+ // The rejected run must have left the table untouched -- no mode change,
no extra commit, and no
+ // base file carrying a null commit time. That last one is the actual data
loss being prevented:
+ // such rows are admitted by incremental queries and then silently dropped.
+ HoodieTableMetaClient metaClient =
HoodieTestUtils.createMetaClient(context, tablePath);
+ assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY,
metaClient.getTableConfig().getMetaFieldsMode(),
+ "a rejected restart must leave the table's mode untouched");
+ assertEquals(commitsAfterFirstRun,
+
metaClient.getActiveTimeline().filterCompletedInstants().countInstants(),
+ "a rejected restart must not have committed anything");
+ Dataset<Row> raw = sparkSession.read().parquet(tablePath +
"/*/*/*/*.parquet");
+ assertEquals(0,
+
raw.filter(functions.col(HoodieRecord.COMMIT_TIME_METADATA_FIELD).isNull()).count(),
+ "no row may have a null _hoodie_commit_time on a COMMIT_TIME_ONLY
table");
+ }
+
+ /**
+ * The migration path for the case above: a restart that restates the
table's mode proceeds normally.
+ *
+ * <p>This is what every writer against a selective table must do, and it is
the assertion that keeps
+ * the requirement from being a dead end -- the mode is stateable, and
stating it is enough.
+ */
+ @Test
+ public void testRestartRestatingTheModeSucceeds() throws Exception {
+ String tablePath = basePath + "/streamer_restart_restates_mode";
+
+ HoodieDeltaStreamer.Config first = TestHelpers.makeConfig(tablePath,
WriteOperationType.INSERT);
+ first.tableType = "COPY_ON_WRITE";
+ first.configs.add(HoodieTableConfig.META_FIELDS_MODE.key() + "=" +
MetaFieldsMode.COMMIT_TIME_ONLY.name());
+ HoodieDeltaStreamer streamer = new HoodieDeltaStreamer(first, jsc);
+ streamer.getIngestionService().ingestOnce();
+ streamer.shutdownGracefully();
+
+ // Restart restating the same mode.
+ HoodieDeltaStreamer.Config restart = TestHelpers.makeConfig(tablePath,
WriteOperationType.INSERT);
+ restart.tableType = "COPY_ON_WRITE";
+ restart.configs.add(HoodieTableConfig.META_FIELDS_MODE.key() + "=" +
MetaFieldsMode.COMMIT_TIME_ONLY.name());
+ HoodieDeltaStreamer restarted = new HoodieDeltaStreamer(restart, jsc);
+ restarted.getIngestionService().ingestOnce();
+ restarted.shutdownGracefully();
+
+ HoodieTableMetaClient metaClient =
HoodieTestUtils.createMetaClient(context, tablePath);
+ assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY,
metaClient.getTableConfig().getMetaFieldsMode(),
+ "the restart must not have changed the table's mode");
+
assertTrue(metaClient.getActiveTimeline().filterCompletedInstants().countInstants()
>= 2,
+ "expected the restart to have produced a second commit");
+
+ // Every row across both commits carries a commit time. A row with a null
one is what incremental
+ // queries silently drop, so this is the assertion that catches the
regression.
+ Dataset<Row> raw = sparkSession.read().parquet(tablePath +
"/*/*/*/*.parquet");
+ assertEquals(0,
+
raw.filter(functions.col(HoodieRecord.COMMIT_TIME_METADATA_FIELD).isNull()).count(),
+ "no row may have a null _hoodie_commit_time on a COMMIT_TIME_ONLY
table");
+
assertTrue(raw.select(HoodieRecord.COMMIT_TIME_METADATA_FIELD).distinct().count()
>= 2,
+ "both commits must be represented, so the second run really did write
through this path");
+ }
+
+ /**
+ * The variant @voonhous asked for, and the one cshuo originally described:
the restart states the
+ * deprecated boolean rather than nothing at all.
+ *
+ * <p>Both cases are rejected against a selective table, for the same
reason: the writer has not
+ * stated the mode the table is on. Stating the boolean instead makes the
disagreement explicit
+ * rather than merely unstated, but the outcome is the same. Before this
rule it narrowed silently,
+ * writing base files with a null {@code _hoodie_commit_time} into a table
that still advertised the
+ * mode.
+ */
+ @Test
+ public void testRestartStatingTheLegacyBooleanIsRejected() throws Exception {
+ String tablePath = basePath + "/streamer_restart_legacy_boolean_conflict";
+
+ HoodieDeltaStreamer.Config first = TestHelpers.makeConfig(tablePath,
WriteOperationType.INSERT);
+ first.tableType = "COPY_ON_WRITE";
+ first.configs.add(HoodieTableConfig.META_FIELDS_MODE.key() + "=" +
MetaFieldsMode.COMMIT_TIME_ONLY.name());
+ HoodieDeltaStreamer streamer = new HoodieDeltaStreamer(first, jsc);
+ streamer.getIngestionService().ingestOnce();
+ streamer.shutdownGracefully();
+
+ HoodieDeltaStreamer.Config restart = TestHelpers.makeConfig(tablePath,
WriteOperationType.INSERT);
+ restart.tableType = "COPY_ON_WRITE";
+ restart.configs.add(HoodieTableConfig.POPULATE_META_FIELDS.key() +
"=false");
+
+ Throwable thrown = assertThrows(Throwable.class, () -> {
+ HoodieDeltaStreamer restarted = new HoodieDeltaStreamer(restart, jsc);
+ restarted.getIngestionService().ingestOnce();
+ restarted.shutdownGracefully();
+ });
+
+ String rootMessage = rootMessageOf(thrown);
+ assertTrue(rootMessage.contains(HoodieTableConfig.META_FIELDS_MODE.key())
+ ||
rootMessage.contains(HoodieTableConfig.POPULATE_META_FIELDS.key()),
+ "expected a meta-fields conflict, got: " + rootMessage);
+
+ // The failed run must not have changed the table, nor written rows with a
null commit time.
+ HoodieTableMetaClient metaClient =
HoodieTestUtils.createMetaClient(context, tablePath);
+ assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY,
metaClient.getTableConfig().getMetaFieldsMode(),
+ "a rejected restart must leave the table's mode untouched");
+ Dataset<Row> raw = sparkSession.read().parquet(tablePath +
"/*/*/*/*.parquet");
+ assertEquals(0,
raw.filter(functions.col(HoodieRecord.COMMIT_TIME_METADATA_FIELD).isNull()).count(),
+ "no row may have a null _hoodie_commit_time");
+ }
+
+ @Test
+ public void testStreamerRejectsMorWithSelectiveMode() throws Exception {
+ String tablePath = basePath + "/streamer_mor_selective_rejected";
+ HoodieDeltaStreamer.Config cfg = TestHelpers.makeConfig(tablePath,
WriteOperationType.BULK_INSERT);
+ cfg.tableType = "MERGE_ON_READ";
+ cfg.configs.add(HoodieTableConfig.META_FIELDS_MODE.key() + "=" +
MetaFieldsMode.COMMIT_TIME_ONLY.name());
+
+ Throwable thrown = assertThrows(Throwable.class, () -> {
+ HoodieDeltaStreamer streamer = new HoodieDeltaStreamer(cfg, jsc);
+ streamer.getIngestionService().ingestOnce();
+ streamer.shutdownGracefully();
+ });
+
+ String rootMessage = rootMessageOf(thrown);
+ assertTrue(rootMessage.contains("COPY_ON_WRITE") ||
rootMessage.contains("MERGE_ON_READ")
+ || rootMessage.contains("MoR") ||
rootMessage.contains(HoodieTableConfig.META_FIELDS_MODE.key()),
+ "Expected MoR-restriction error, got: " + rootMessage);
+ }
+
+ private void assertOnDiskMetaColumns(String tablePath, MetaFieldsMode
expectedMode) {
+ // Default HoodieTestDataGenerator partitions are YYYY/MM/DD (three
levels).
+ Dataset<Row> raw = sparkSession.read().parquet(tablePath +
"/*/*/*/*.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), "commit_time must be populated in mode " +
expectedMode);
Review Comment:
🤖 nit: positional access like `first.get(0)` / `first.get(4)` is fragile —
if someone reorders the `select()` call above, these assertions silently check
the wrong column. Could you use
`first.getAs(HoodieRecord.COMMIT_TIME_METADATA_FIELD)` and
`first.getAs(HoodieRecord.FILENAME_METADATA_FIELD)` (and similarly for indices
1–3) instead?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieStreamerMetaFieldsMode.java:
##########
@@ -0,0 +1,285 @@
+/*
+ * 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.utilities.deltastreamer;
+
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.MetaFieldsMode;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.functions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+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;
+
+/**
+ * End-to-end coverage for {@code hoodie.meta.fields.mode} through the
HoodieStreamer entrypoint.
+ * Each parameterized invocation runs a single ingest cycle in the given
{@link MetaFieldsMode} and
+ * verifies both the persisted table property and the actual on-disk parquet
column population.
+ *
+ * <p>Rejection paths (unknown token, populate=true+mode, MoR+mode) are
exercised in the datasource
+ * test {@code TestMetaFieldsMode}; this fixture focuses on the streamer
control-flow.
+ */
+public class TestHoodieStreamerMetaFieldsMode extends
HoodieDeltaStreamerTestBase {
+
+ /**
+ * Only the selective modes are parameterized here. ALL and NONE add no mode
key at all, so they
+ * exercise none of the streamer-side plumbing this feature introduced —
they are covered by the
+ * datasource tests and by {@code TestHoodieTableConfig}'s resolution cases.
+ */
+ @ParameterizedTest
+ @EnumSource(value = MetaFieldsMode.class,
+ names = {"COMMIT_TIME_ONLY", "FILE_NAME_ONLY",
"COMMIT_TIME_AND_FILE_NAME"})
+ public void testStreamerRespectsMetaFieldsMode(MetaFieldsMode mode) throws
Exception {
+ String tablePath = basePath + "/streamer_meta_fields_mode_" + mode.name();
+ HoodieDeltaStreamer.Config cfg = TestHelpers.makeConfig(tablePath,
WriteOperationType.INSERT);
+ // Force CoW; selective modes are CoW-only until MoR log-write is wired.
+ cfg.tableType = "COPY_ON_WRITE";
+ // The mode alone — pairing it with populate.meta.fields would now be a
stated conflict, since the
+ // mode is authoritative and the boolean is only the fallback for
resolving an absent one.
+ cfg.configs.add(HoodieTableConfig.META_FIELDS_MODE.key() + "=" +
mode.name());
+ HoodieDeltaStreamer streamer = new HoodieDeltaStreamer(cfg, jsc);
+ streamer.getIngestionService().ingestOnce();
+ streamer.shutdownGracefully();
+
+ HoodieTableMetaClient metaClient =
HoodieTestUtils.createMetaClient(context, tablePath);
+ assertEquals(mode, metaClient.getTableConfig().getMetaFieldsMode(),
+ "streamer must persist mode=" + mode + " on hoodie.properties");
+ assertOnDiskMetaColumns(tablePath, mode);
+ }
+
+ /**
+ * The regression this fixture exists for, and the one cshuo raised in
review: a restarted streamer
+ * that does not restate the mode must not silently write the wrong meta
columns.
Review Comment:
🤖 nit: reviewer handles (`cshuo`, `voonhous`) in Javadoc will be opaque to
anyone who wasn't in that review thread. Could you rephrase to describe the
scenario directly rather than attributing it to individuals?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
