hudi-agent commented on code in PR #19378:
URL: https://github.com/apache/hudi/pull/19378#discussion_r3744040634
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java:
##########
@@ -1544,13 +1545,80 @@ protected boolean loadActiveTimelineOnTableInit() {
return true;
}
+ /**
+ * Adopt the table's {@code hoodie.meta.fields.mode} when this writer did
not state one.
+ *
+ * <p>The mode is a table property: it is settable at table creation,
through hudi-cli, or by an
+ * upgrade, and never by an ordinary write. Callers routinely build a write
config without
+ * restating the table's meta-field settings — table services do, and so
does a restarted
+ * StreamSync — and such a writer must write what the table already
advertises rather than silently
+ * narrowing it. Without this, an unstated writer resolves to {@code NONE}
(via the deprecated
+ * {@code hoodie.populate.meta.fields} fallback) and writes null meta
columns into a table whose
+ * earlier files have them populated.
+ *
+ * <p>Inheritance applies only when the writer states <em>neither</em>
property. If it explicitly
+ * set the mode or the deprecated boolean, that value is left alone so the
comparison below can
+ * reject it on a mismatch: a user who deliberately passed {@code
populate.meta.fields=false}
+ * against an {@code ALL} table should be told the setting conflicts, not
have it silently
+ * overridden.
+ *
+ * @return true when the writer stated neither property, i.e. the mode was
inherited and cannot
+ * disagree with the table.
+ */
+ private static boolean inferMetaFieldsModeFromTable(HoodieTableConfig
tableConfig, HoodieWriteConfig writeConfig) {
+ boolean statedMode =
writeConfig.contains(HoodieTableConfig.META_FIELDS_MODE)
+ &&
!StringUtils.isNullOrEmpty(writeConfig.getString(HoodieTableConfig.META_FIELDS_MODE));
+ boolean statedLegacyBoolean =
writeConfig.contains(HoodieTableConfig.POPULATE_META_FIELDS);
+ if (statedMode || statedLegacyBoolean) {
+ return false;
+ }
+ MetaFieldsMode tableMode = tableConfig.getMetaFieldsMode();
+ writeConfig.setValue(HoodieTableConfig.META_FIELDS_MODE, tableMode.name());
+ // Keep the derived boolean in step, so the ~55 call sites still reading
populateMetaFields()
+ // observe an answer consistent with the mode.
+ writeConfig.setValue(HoodieTableConfig.POPULATE_META_FIELDS,
+ Boolean.toString(tableMode.toLegacyPopulateMetaFields()));
+ return true;
+ }
+
public void validateAgainstTableProperties(HoodieTableConfig tableConfig,
HoodieWriteConfig writeConfig) {
// mismatch of table versions.
CommonClientUtils.validateTableVersion(tableConfig, writeConfig);
- // Once meta fields are disabled, it cant be re-enabled for a given table.
- if (!tableConfig.populateMetaFields() && writeConfig.populateMetaFields())
{
- throw new HoodieException(HoodieTableConfig.POPULATE_META_FIELDS.key() +
" already disabled for the table. Can't be re-enabled back");
+ // A writer that stated neither meta-field property inherits the table's
mode here, which makes
+ // the comparison below a no-op for it. A writer that stated either keeps
its value and is
+ // compared.
+ inferMetaFieldsModeFromTable(tableConfig, writeConfig);
+
+ // Meta-field population is physical, so a writer must not disagree with
the table about which
Review Comment:
🤖 The CoW-only / Spark-only guards in `HoodieWriteConfig.validate()` only
fire when a writer *states* a selective mode, but
`inferMetaFieldsModeFromTable` makes an unstated writer inherit the table's
mode with no re-validation. Since `HoodieFlinkWriteClient` calls this same
`final initTable(...)`, wouldn't a Flink job against a Spark-created
`COMMIT_TIME_ONLY` CoW table inherit the selective mode and then write null
meta columns (the Flink writer ignores the mode)? Should the engine/table-type
check be re-applied after inheritance here?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieAvroFileWriterFactory.java:
##########
@@ -123,7 +125,13 @@ protected HoodieFileWriter newHFileFileWriter(
HoodieAvroHFileReaderImplBase.KEY_FIELD_NAME,
filter,
config.getBoolean(HFILE_WRITER_TO_ALLOW_DUPLICATES));
- return new HoodieAvroHFileWriter(instantTime, path, hfileConfig, schema,
taskContextSupplier, config.getBoolean(HoodieTableConfig.POPULATE_META_FIELDS));
+ // Resolve through the mode like the parquet path above. HFile does not
populate meta columns
+ // selectively, so a selective mode is treated as "record key not
populated". Note getBoolean
Review Comment:
🤖 nit: `MetaFieldsMode` is already imported at the top of the file — could
you drop the fully-qualified
`org.apache.hudi.common.model.MetaFieldsMode.resolve(…)` here (and in the
similar Vortex/Lance blocks in `HoodieSparkFileWriterFactory`) to match the
style used on the parquet path just above?
<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.
+ *
Review Comment:
🤖 nit: could you swap out the reviewer handles (`cshuo`, `voonhous`) for a
description of the scenario they raised? Names from a PR thread won't mean
anything to someone reading this six months from now.
<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,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.
Review Comment:
🤖 nit: the `%s, %s, %s, %s, %s` list will silently go stale if a sixth mode
is ever added. Could you replace it with
`Arrays.stream(values()).map(Enum::name).collect(Collectors.joining(", "))` so
the message auto-updates?
<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);
+ } else {
+ assertNull(first.get(0), "commit_time must be null in mode " +
expectedMode);
Review Comment:
🤖 nit: the positional `first.get(0)` / `first.get(4)` etc. are fragile — if
the column order in the `select()` above ever changes, these assertions
silently test the wrong fields.
`first.getAs(HoodieRecord.COMMIT_TIME_METADATA_FIELD)` (and similarly for the
other fields) ties each check to the named column rather than a magic index.
<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]
