hudi-agent commented on code in PR #19206: URL: https://github.com/apache/hudi/pull/19206#discussion_r3743362178
########## hudi-common/src/main/java/org/apache/hudi/common/model/MetaFieldsMode.java: ########## @@ -0,0 +1,213 @@ +/* + * 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); + } + + /** + * Parse a raw {@code hoodie.meta.fields.mode} value into an enum constant, with a message that + * lists the allowed values. Prefer this over {@link #valueOf(String)} for user-supplied input. + */ + public static MetaFieldsMode parse(String rawMode) { + try { + // Case-insensitive: users hand-editing hoodie.properties or passing write options should not Review Comment: 🤖 nit: could you derive the allowed-values list dynamically here instead of hardcoding it? Something like `Arrays.stream(values()).map(Enum::name).collect(Collectors.joining(", "))` — that way the message stays accurate if a new constant is added later without anyone remembering to update this string. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestTableCommand.java: ########## @@ -304,4 +305,316 @@ private String getFileContent(String fileToReadStr) throws IOException { fis.close(); return fromUTF8Bytes(data); } + + // --------------------------------------------------------------------------- + // set-meta-fields-mode + // --------------------------------------------------------------------------- + + @Test + public void testSetMetaFieldsModeOnFreshTableToCommitTimeOnly() { + assertTrue(prepareTable()); + // Default table is ALL — no commits yet, so the safety check must let this through. + Object result = shell.evaluate(() -> + "table set-meta-fields-mode --target-mode COMMIT_TIME_ONLY"); + assertTrue(ShellEvaluationResultUtil.isSuccess(result)); + // Rendered diff must surface the changed properties so the operator can confirm the write. + String rendered = result.toString(); + assertTrue(rendered.contains(HoodieTableConfig.POPULATE_META_FIELDS.key()), + "expected rendered diff to mention populate.meta.fields, got: " + rendered); + assertTrue(rendered.contains(HoodieTableConfig.META_FIELDS_MODE.key()), + "expected rendered diff to mention meta.fields.mode, got: " + rendered); + HoodieTableMetaClient client = HoodieCLI.getTableMetaClient(); + assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY, client.getTableConfig().getMetaFieldsMode()); + assertFalse(client.getTableConfig().populateMetaFields()); + } + + @Test + public void testSetMetaFieldsModeOnFreshTableToFileNameOnly() { + assertTrue(prepareTable()); + Object result = shell.evaluate(() -> + "table set-meta-fields-mode --target-mode FILE_NAME_ONLY"); + assertTrue(ShellEvaluationResultUtil.isSuccess(result)); + assertEquals(MetaFieldsMode.FILE_NAME_ONLY, + HoodieCLI.getTableMetaClient().getTableConfig().getMetaFieldsMode()); + } + + @Test + public void testSetMetaFieldsModeOnFreshTableToCombinedMode() { + assertTrue(prepareTable()); + Object result = shell.evaluate(() -> + "table set-meta-fields-mode --target-mode COMMIT_TIME_AND_FILE_NAME"); + assertTrue(ShellEvaluationResultUtil.isSuccess(result)); + assertEquals(MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME, + HoodieCLI.getTableMetaClient().getTableConfig().getMetaFieldsMode()); + } + + @Test + public void testSetMetaFieldsModeOnFreshTableToNone() { + assertTrue(prepareTable()); + Object result = shell.evaluate(() -> + "table set-meta-fields-mode --target-mode NONE"); + assertTrue(ShellEvaluationResultUtil.isSuccess(result)); + HoodieTableMetaClient client = HoodieCLI.getTableMetaClient(); + assertEquals(MetaFieldsMode.NONE, client.getTableConfig().getMetaFieldsMode()); + assertFalse(client.getTableConfig().populateMetaFields()); + } + + @Test + public void testSetMetaFieldsModeToAllWritesTheModeExplicitly() throws IOException { + assertTrue(prepareTable()); + // First move to a selective mode, then back to ALL. Legal here only because the table has no + // commits — on a populated table this would be a widening and refused outright. + shell.evaluate(() -> "table set-meta-fields-mode --target-mode COMMIT_TIME_ONLY"); + Object result = shell.evaluate(() -> "table set-meta-fields-mode --target-mode ALL"); + assertTrue(ShellEvaluationResultUtil.isSuccess(result)); + + HoodieTableMetaClient client = HoodieCLI.getTableMetaClient(); + assertEquals(MetaFieldsMode.ALL, client.getTableConfig().getMetaFieldsMode()); + assertTrue(client.getTableConfig().populateMetaFields()); + // The mode is written explicitly rather than deleted. Deleting it would leave the table + // resolving through the legacy fallback -- indistinguishable from a table predating the + // property -- and would make this command's effect invisible in hoodie.properties. The v10->v9 + // downgrade handler also reads the mode to derive the boolean it writes back. + assertEquals(MetaFieldsMode.ALL.name(), + client.getTableConfig().getString(HoodieTableConfig.META_FIELDS_MODE)); + } + + @Test + public void testSetMetaFieldsModeRefusesWideningOnPopulatedTableEvenWithForce() throws Exception { + assertTrue(prepareTable()); + shell.evaluate(() -> "table set-meta-fields-mode --target-mode COMMIT_TIME_ONLY"); + createDummyCommitFile("20260101000000000"); + HoodieCLI.refreshTableMetadata(); + + // Widening is one-way-forbidden and --force does not override it: existing files are not + // rewritten, so the table would advertise a column that is null for every row written so far, + // and incremental queries would then admit the table and silently skip those rows. There is no + // consequence for an operator to knowingly accept, so this is a hard failure. + for (String command : new String[] { + "table set-meta-fields-mode --target-mode ALL", + "table set-meta-fields-mode --target-mode ALL --force true", + "table set-meta-fields-mode --target-mode COMMIT_TIME_AND_FILE_NAME --force true"}) { + Object result = shell.evaluate(() -> command); + assertFalse(ShellEvaluationResultUtil.isSuccess(result), "expected refusal for: " + command); + assertTrue(result.toString().contains("widen"), + "expected a widening refusal for '" + command + "', got: " + result); + assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY, + HoodieCLI.getTableMetaClient().getTableConfig().getMetaFieldsMode(), + "mode must be unchanged after a refused widening"); + } + } + + @Test + public void testSetMetaFieldsModeAllowsNarrowingOnPopulatedTableWithForce() throws Exception { + assertTrue(prepareTable()); + shell.evaluate(() -> "table set-meta-fields-mode --target-mode COMMIT_TIME_AND_FILE_NAME"); + createDummyCommitFile("20260101000000000"); + HoodieCLI.refreshTableMetadata(); + + // Narrowing is the direction the CLI exists to allow. It still needs --force, because it leaves + // mixed-mode files, but it is not refused outright the way widening is. + Object refused = shell.evaluate(() -> + "table set-meta-fields-mode --target-mode COMMIT_TIME_ONLY"); + assertFalse(ShellEvaluationResultUtil.isSuccess(refused)); + + Object forced = shell.evaluate(() -> + "table set-meta-fields-mode --target-mode COMMIT_TIME_ONLY --force true"); + assertTrue(ShellEvaluationResultUtil.isSuccess(forced), "narrowing with --force must succeed: " + forced); + assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY, + HoodieCLI.getTableMetaClient().getTableConfig().getMetaFieldsMode()); + } + + @Test + public void testSetMetaFieldsModeAcceptsLowercaseTargetMode() { + assertTrue(prepareTable()); + // Routed through MetaFieldsMode.parse rather than valueOf, so an operator typing the mode in + // lower case is not rejected. + Object result = shell.evaluate(() -> + "table set-meta-fields-mode --target-mode commit_time_only"); + assertTrue(ShellEvaluationResultUtil.isSuccess(result), "expected lowercase to be accepted: " + result); + assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY, + HoodieCLI.getTableMetaClient().getTableConfig().getMetaFieldsMode()); + } + + @Test + public void testSetMetaFieldsModeKeepsBothPropertiesInAgreement() { + assertTrue(prepareTable()); + // hoodie.properties must never contradict itself: the legacy boolean is derived from the mode, + // never taken from the caller, so a pre-1.3.0 reader that sees only the boolean treats a + // selective table as NONE rather than assuming meta columns that are physically null. + for (MetaFieldsMode mode : MetaFieldsMode.values()) { + shell.evaluate(() -> "table set-meta-fields-mode --target-mode " + mode.name() + " --force true"); + HoodieTableConfig tableConfig = HoodieCLI.getTableMetaClient().getTableConfig(); + if (tableConfig.getMetaFieldsMode() == mode) { + assertEquals(mode.toLegacyPopulateMetaFields(), tableConfig.populateMetaFields(), + "populate.meta.fields must be the derived value for mode " + mode); + } + } + } + + @Test + public void testSetMetaFieldsModeNoOpWhenAlreadyInTargetMode() { + assertTrue(prepareTable()); + Object first = shell.evaluate(() -> + "table set-meta-fields-mode --target-mode COMMIT_TIME_ONLY"); + assertTrue(ShellEvaluationResultUtil.isSuccess(first)); + // Second call — same target — should be a no-op message. + Object second = shell.evaluate(() -> + "table set-meta-fields-mode --target-mode COMMIT_TIME_ONLY"); + assertTrue(ShellEvaluationResultUtil.isSuccess(second)); + assertTrue(second.toString().contains("already in COMMIT_TIME_ONLY"), + "expected no-op message, got: " + second); + } + + @Test + public void testSetMetaFieldsModeRejectsUnknownValue() { + assertTrue(prepareTable()); + Object result = shell.evaluate(() -> + "table set-meta-fields-mode --target-mode BOGUS_MODE"); + // Shell evaluate returns the exception object on failure. + assertFalse(ShellEvaluationResultUtil.isSuccess(result)); + assertTrue(result.toString().contains("BOGUS_MODE"), + "expected error message to name the rejected value, got: " + result); + } + + @Test + public void testSetMetaFieldsModeRefusesOnPopulatedTable() throws Exception { + assertTrue(prepareTable()); + createDummyCommitFile("20260101000000000"); + HoodieCLI.refreshTableMetadata(); + + Object result = shell.evaluate(() -> + "table set-meta-fields-mode --target-mode COMMIT_TIME_ONLY"); + assertFalse(ShellEvaluationResultUtil.isSuccess(result)); + assertTrue(result.toString().contains("Refusing to change") || result.toString().contains("--force"), + "expected refusal message, got: " + result); + + // Mode must not have changed. + assertEquals(MetaFieldsMode.ALL, + HoodieCLI.getTableMetaClient().getTableConfig().getMetaFieldsMode()); + } + + @Test + public void testSetMetaFieldsModeWithForceOnPopulatedTable() throws Exception { + assertTrue(prepareTable()); + createDummyCommitFile("20260101000000000"); + HoodieCLI.refreshTableMetadata(); + + Object result = shell.evaluate(() -> + "table set-meta-fields-mode --target-mode COMMIT_TIME_ONLY --force true"); + assertTrue(ShellEvaluationResultUtil.isSuccess(result)); + assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY, + HoodieCLI.getTableMetaClient().getTableConfig().getMetaFieldsMode()); + } + + /** + * Pins the whole migration matrix rather than spot-checking it. Twenty ordered pairs, each either + * a legal narrowing or a refused widening, on a table that already has commits. + * + * <p>The subtle entries are the mutually-wider siblings: {@code COMMIT_TIME_ONLY} and + * {@code FILE_NAME_ONLY} each populate a column the other does not, so neither can migrate to the + * other in either direction. A spot-check of "narrowing works, widening does not" would miss that + * the relation is a lattice rather than a chain. + * + * <p>Uses a fresh table per pair so the starting mode can be set without tripping the very guard + * under test -- setting the initial mode happens before any commit exists. + */ + @Test + public void testSetMetaFieldsModeMigrationMatrixOnPopulatedTable() throws Exception { + for (MetaFieldsMode from : MetaFieldsMode.values()) { + for (MetaFieldsMode to : MetaFieldsMode.values()) { + if (from == to) { + continue; + } + // Fresh table per pair; connect to it so HoodieCLI points at the right one. + String pairName = tableName + "_" + from.name() + "_to_" + to.name(); + String pairPath = tablePath(pairName); + assertTrue(ShellEvaluationResultUtil.isSuccess( + shell.evaluate(() -> "create --path " + pairPath + " --tableName " + pairName))); + + // Establish the starting mode while the table is still empty, then make it "populated". + assertTrue(ShellEvaluationResultUtil.isSuccess( + shell.evaluate(() -> "table set-meta-fields-mode --target-mode " + from.name())), + "setting the initial mode on an empty table must succeed: " + from); + createDummyCommitFileAt(pairPath, "20260101000000000"); + HoodieCLI.refreshTableMetadata(); + assertEquals(from, HoodieCLI.getTableMetaClient().getTableConfig().getMetaFieldsMode()); + + boolean widening = to.isWiderThan(from); + Object result = shell.evaluate(() -> + "table set-meta-fields-mode --target-mode " + to.name() + " --force true"); + + if (widening) { + assertFalse(ShellEvaluationResultUtil.isSuccess(result), + from + " -> " + to + " adds a meta column and must be refused even with --force"); + assertTrue(result.toString().contains("widen"), + "expected a widening refusal for " + from + " -> " + to + ", got: " + result); + assertEquals(from, HoodieCLI.getTableMetaClient().getTableConfig().getMetaFieldsMode(), + "a refused migration must leave the mode untouched: " + from + " -> " + to); + } else { + assertTrue(ShellEvaluationResultUtil.isSuccess(result), + from + " -> " + to + " drops meta columns and must be allowed with --force, got: " + result); + HoodieTableConfig tableConfig = HoodieCLI.getTableMetaClient().getTableConfig(); + assertEquals(to, tableConfig.getMetaFieldsMode(), from + " -> " + to); + assertEquals(to.toLegacyPopulateMetaFields(), tableConfig.populateMetaFields(), + "the derived boolean must follow the new mode for " + from + " -> " + to); + } + } + } + } + + /** Both mutually-wider directions between the two single-column modes are refused. */ + @Test + public void testSetMetaFieldsModeRefusesBothSiblingDirections() throws Exception { + assertTrue(prepareTable()); + shell.evaluate(() -> "table set-meta-fields-mode --target-mode COMMIT_TIME_ONLY"); + createDummyCommitFile("20260101000000000"); + HoodieCLI.refreshTableMetadata(); + + Object toSibling = shell.evaluate(() -> + "table set-meta-fields-mode --target-mode FILE_NAME_ONLY --force true"); + assertFalse(ShellEvaluationResultUtil.isSuccess(toSibling), + "COMMIT_TIME_ONLY -> FILE_NAME_ONLY adds _hoodie_file_name and must be refused"); + assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY, + HoodieCLI.getTableMetaClient().getTableConfig().getMetaFieldsMode()); + + // ...and the reverse, on a table that starts the other way round. + String otherName = tableName + "_sibling_reverse"; + String otherPath = tablePath(otherName); + assertTrue(ShellEvaluationResultUtil.isSuccess( + shell.evaluate(() -> "create --path " + otherPath + " --tableName " + otherName))); + shell.evaluate(() -> "table set-meta-fields-mode --target-mode FILE_NAME_ONLY"); + createDummyCommitFileAt(otherPath, "20260101000000000"); + HoodieCLI.refreshTableMetadata(); + + Object toOther = shell.evaluate(() -> + "table set-meta-fields-mode --target-mode COMMIT_TIME_ONLY --force true"); + assertFalse(ShellEvaluationResultUtil.isSuccess(toOther), + "FILE_NAME_ONLY -> COMMIT_TIME_ONLY adds _hoodie_commit_time and must be refused"); + assertEquals(MetaFieldsMode.FILE_NAME_ONLY, + HoodieCLI.getTableMetaClient().getTableConfig().getMetaFieldsMode()); + } + + private void createDummyCommitFileAt(String tableBasePath, String instantTime) throws IOException { + java.nio.file.Path timelineDir = + Paths.get(tableBasePath, METAFOLDER_NAME, "timeline"); + if (!timelineDir.toFile().exists()) { + timelineDir.toFile().mkdirs(); + } + String completionTime = instantTime + "1"; + java.nio.file.Files.createFile(timelineDir.resolve(instantTime + "_" + completionTime + ".commit")); Review Comment: 🤖 nit: `createDummyCommitFile` and `createDummyCommitFileAt` have nearly identical bodies — could `createDummyCommitFile` just delegate to `createDummyCommitFileAt` by passing the table base path derived from `metaPath`? Keeps the file-creation logic in one place. <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,232 @@ +/* + * 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 keep writing the table's 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>The rule now lives in {@code BaseHoodieWriteClient#validateAgainstTableProperties} for every + * engine rather than in StreamSync, so this asserts it end-to-end through the streamer: a run that + * states neither meta-field property inherits the table's mode. + */ + @Test + public void testRestartWithoutRestatingTheModeKeepsWritingCommitTimes() 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()); + + // 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"; + 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>These are different cases under the current rule. Stating neither meta-field property inherits + * the table's mode (above); stating the boolean is an explicit request that contradicts a + * {@code COMMIT_TIME_ONLY} table, so it is rejected rather than silently narrowing the write to + * {@code NONE}. 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: could you use `first.getAs(HoodieRecord.COMMIT_TIME_METADATA_FIELD)` (and similarly for indices 1–4) instead of positional `get(0)` / `get(4)`? The magic indices are only correct as long as the `select()` call stays in exactly this order, so a future reorder would silently test the wrong column. <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]
