voonhous commented on code in PR #19205: URL: https://github.com/apache/hudi/pull/19205#discussion_r3794375023
########## hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestMetaFieldsModeE2E.java: ########## @@ -0,0 +1,830 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hudi.functional; + +import org.apache.hudi.DataSourceReadOptions; +import org.apache.hudi.DataSourceWriteOptions; +import org.apache.hudi.SparkAdapterSupport$; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.model.HoodieRecord; +import org.apache.hudi.common.model.MetaFieldsMode; +import org.apache.hudi.common.table.HoodieTableConfig; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.testutils.SparkClientFunctionalTestHarness; + +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.RowFactory; +import org.apache.spark.sql.SaveMode; +import org.apache.spark.sql.functions; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Spark-datasource end-to-end tests for the {@code hoodie.meta.fields.mode} property on CoW tables. + * Every {@link MetaFieldsMode} value is exercised via a write / re-read round trip; on-disk column + * population is verified by reading the parquet files back and inspecting the meta-column values. + */ +class TestMetaFieldsModeE2E extends SparkClientFunctionalTestHarness { + + private static StructType simpleSchema() { + return DataTypes.createStructType(new StructField[]{ + DataTypes.createStructField("column1", DataTypes.StringType, true), + DataTypes.createStructField("column2", DataTypes.StringType, true), + DataTypes.createStructField("column3", DataTypes.StringType, true) + }).asNullable(); + } + + private Map<String, String> baseOptions() { + Map<String, String> opts = new HashMap<>(); + opts.put(DataSourceWriteOptions.RECORDKEY_FIELD().key(), "column1"); + opts.put(DataSourceWriteOptions.PARTITIONPATH_FIELD().key(), "column2"); + opts.put(DataSourceWriteOptions.ORDERING_FIELDS().key(), "column3"); + opts.put(HoodieTableConfig.NAME.key(), "test_meta_fields_mode"); + opts.put(DataSourceWriteOptions.TABLE_TYPE().key(), "COPY_ON_WRITE"); + opts.put(HoodieMetadataConfig.ENABLE.key(), "false"); + return opts; + } + + private void writeRows(List<Row> records, StructType schema, Map<String, String> options, String path, SaveMode mode) { + spark().createDataset(records, + SparkAdapterSupport$.MODULE$.sparkAdapter().getCatalystExpressionUtils().getEncoder(schema)) + .write() + .format("hudi") + .options(options) + .mode(mode) + .save(path); + } + + private HoodieTableConfig writeSampleAndGetTableConfig(Map<String, String> options, String path) { + writeRows(Arrays.asList( + RowFactory.create("k1", "p1", "v1"), + RowFactory.create("k2", "p1", "v2")), + simpleSchema(), options, path, SaveMode.Overwrite); + HoodieTableMetaClient metaClient = + HoodieTableMetaClient.builder().setBasePath(path).setConf(storageConf()).build(); + return metaClient.getTableConfig(); + } + + /** + * End-to-end assertion of the on-disk meta columns after a write. Reads the parquet files back + * (bypassing Hudi's own read path so we see the raw column values) and asserts which meta + * columns are non-null. + */ + private void assertMetaColumnPopulation(String path, MetaFieldsMode expectedMode) { + Dataset<Row> raw = spark().read().parquet(path + "/*/*.parquet"); + Row first = raw.select( + HoodieRecord.COMMIT_TIME_METADATA_FIELD, + HoodieRecord.COMMIT_SEQNO_METADATA_FIELD, + HoodieRecord.RECORD_KEY_METADATA_FIELD, + HoodieRecord.PARTITION_PATH_METADATA_FIELD, + HoodieRecord.FILENAME_METADATA_FIELD).first(); + + if (expectedMode.isCommitTimePopulated()) { + assertNotNull(first.get(0), "expected _hoodie_commit_time to be populated for mode " + expectedMode); + } else { + assertNull(first.get(0), "expected _hoodie_commit_time to be null for mode " + expectedMode); + } + if (expectedMode.isFileNamePopulated()) { + assertNotNull(first.get(4), "expected _hoodie_file_name to be populated for mode " + expectedMode); + } else { + assertNull(first.get(4), "expected _hoodie_file_name to be null for mode " + expectedMode); + } + // Record key, partition path, and commit seq no are ALL-only. + if (expectedMode == MetaFieldsMode.ALL) { + assertNotNull(first.get(2), "record key must be populated in ALL mode"); + assertNotNull(first.get(3), "partition path must be populated in ALL mode"); + assertNotNull(first.get(1), "commit seq no must be populated in ALL mode"); + } else { + assertNull(first.get(2), "record key must be null outside ALL mode, got: " + first.get(2)); + assertNull(first.get(3), "partition path must be null outside ALL mode, got: " + first.get(3)); + assertNull(first.get(1), "commit seq no must be null outside ALL mode, got: " + first.get(1)); + } + } + + @Test + void allModePersistsAndPopulatesAllColumns() { + Map<String, String> options = baseOptions(); + // ALL is the default; no need to set the mode explicitly. + options.put(DataSourceWriteOptions.OPERATION().key(), DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL()); + + HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath()); + + assertTrue(tc.populateMetaFields()); + assertEquals(MetaFieldsMode.ALL, tc.getMetaFieldsMode()); + assertMetaColumnPopulation(basePath(), MetaFieldsMode.ALL); + } + + @Test + void noneModePersistsAndLeavesAllColumnsNull() { + Map<String, String> options = baseOptions(); + options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false"); + options.put(DataSourceWriteOptions.OPERATION().key(), DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL()); + + HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath()); + + assertFalse(tc.populateMetaFields()); + assertEquals(MetaFieldsMode.NONE, tc.getMetaFieldsMode()); + assertMetaColumnPopulation(basePath(), MetaFieldsMode.NONE); + } + + @Test + void commitTimeOnlyModePopulatesOnlyCommitTime() { + Map<String, String> options = baseOptions(); + options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false"); + options.put(HoodieTableConfig.META_FIELDS_MODE.key(), MetaFieldsMode.COMMIT_TIME_ONLY.name()); + options.put(DataSourceWriteOptions.OPERATION().key(), DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL()); + + HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath()); + + assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY.name(), + tc.getProps().getProperty(HoodieTableConfig.META_FIELDS_MODE.key())); + assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY, tc.getMetaFieldsMode()); + assertMetaColumnPopulation(basePath(), MetaFieldsMode.COMMIT_TIME_ONLY); + } + + @Test + void fileNameOnlyModePopulatesOnlyFileName() { + Map<String, String> options = baseOptions(); + options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false"); + options.put(HoodieTableConfig.META_FIELDS_MODE.key(), MetaFieldsMode.FILE_NAME_ONLY.name()); + options.put(DataSourceWriteOptions.OPERATION().key(), DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL()); + + HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath()); + + assertEquals(MetaFieldsMode.FILE_NAME_ONLY, tc.getMetaFieldsMode()); + assertMetaColumnPopulation(basePath(), MetaFieldsMode.FILE_NAME_ONLY); + } + + @Test + void commitTimeAndFileNameModePopulatesBoth() { + Map<String, String> options = baseOptions(); + options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false"); + options.put(HoodieTableConfig.META_FIELDS_MODE.key(), MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME.name()); + options.put(DataSourceWriteOptions.OPERATION().key(), DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL()); + + HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath()); + + assertEquals(MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME, tc.getMetaFieldsMode()); + assertMetaColumnPopulation(basePath(), MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME); + } + + @Test + void explicitlyContradictingTheModeIsRejectedAtTableCreation() { + // A selective mode implies populate.meta.fields=false. Stating the boolean as true alongside it + // is a contradiction, and the user is told rather than having half their request discarded. + // This is the datasource end of the check in HoodieTableMetaClient.TableBuilder. + Map<String, String> options = baseOptions(); + options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "true"); + options.put(HoodieTableConfig.META_FIELDS_MODE.key(), MetaFieldsMode.COMMIT_TIME_ONLY.name()); + options.put(DataSourceWriteOptions.OPERATION().key(), DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL()); + + Throwable thrown = assertThrows(Throwable.class, () -> + writeSampleAndGetTableConfig(options, basePath())); + + String rootMessage = rootMessageOf(thrown); + assertTrue(rootMessage.contains(HoodieTableConfig.META_FIELDS_MODE.key()) + && rootMessage.contains(HoodieTableConfig.POPULATE_META_FIELDS.key()), + "the error must name both properties so the user knows which to drop, got: " + rootMessage); + } + + @Test + void selectiveModeWithoutTheLegacyBooleanDerivesItAsFalse() { + // The ordinary case: state only the mode. The boolean is derived, never carried through + // verbatim -- a pre-1.3.0 reader ignores the mode property, so leaving populate=true would make + // it treat a selectively-written table as ALL. + Map<String, String> options = baseOptions(); + options.put(HoodieTableConfig.META_FIELDS_MODE.key(), MetaFieldsMode.COMMIT_TIME_ONLY.name()); + options.put(DataSourceWriteOptions.OPERATION().key(), DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL()); + + HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath()); + + assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY, tc.getMetaFieldsMode()); + assertMetaColumnPopulation(basePath(), MetaFieldsMode.COMMIT_TIME_ONLY); + assertFalse(tc.populateMetaFields(), + "legacy populate.meta.fields must be derived from the mode"); + } + + @Test + void noneModePersistsLegacyBooleanAsFalse() { + // The unsafe case this invariant protects: an old incremental reader that saw + // populate.meta.fields=true on a NONE table would run against all-null commit times and + // silently return zero rows. Stating only the mode -- the ordinary case -- must derive false. + Map<String, String> options = baseOptions(); + options.put(HoodieTableConfig.META_FIELDS_MODE.key(), MetaFieldsMode.NONE.name()); + options.put(DataSourceWriteOptions.OPERATION().key(), DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL()); + + HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath()); + + assertEquals(MetaFieldsMode.NONE, tc.getMetaFieldsMode()); + assertFalse(tc.populateMetaFields(), + "NONE must persist populate.meta.fields=false so pre-1.3.0 readers do not treat it as ALL"); + } + + @Test + void allModePersistsLegacyBooleanAsTrue() { + Map<String, String> options = baseOptions(); + options.put(HoodieTableConfig.META_FIELDS_MODE.key(), MetaFieldsMode.ALL.name()); + options.put(DataSourceWriteOptions.OPERATION().key(), DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL()); + + HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath()); + + assertEquals(MetaFieldsMode.ALL, tc.getMetaFieldsMode()); + assertTrue(tc.populateMetaFields(), + "ALL must persist populate.meta.fields=true for pre-1.3.0 readers"); + } + + @Test + void unknownModeValueIsRejected() { + Map<String, String> options = baseOptions(); + options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false"); + options.put(HoodieTableConfig.META_FIELDS_MODE.key(), "SOMETHING_BOGUS"); + options.put(DataSourceWriteOptions.OPERATION().key(), DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL()); + + Throwable thrown = assertThrows(Throwable.class, () -> + writeRows(Collections.singletonList(RowFactory.create("k1", "p1", "v1")), + simpleSchema(), options, basePath(), SaveMode.Overwrite)); + + String rootMessage = rootMessageOf(thrown); + assertTrue(rootMessage.contains("SOMETHING_BOGUS"), + "Expected error to name the rejected value, got: " + rootMessage); + } + + // ------------------------------------------------------------------------- + // Non-row-writer path coverage. Bulk insert with row.writer.enable=false forces the + // HoodieAvroParquetWriter path (via HoodieCreateHandle) instead of the internal-row writer path. + // Both paths must respect the mode identically. + // ------------------------------------------------------------------------- + + @Test + void nonRowWriterPathAllMode() { + Map<String, String> options = baseOptions(); + options.put(DataSourceWriteOptions.OPERATION().key(), DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL()); + options.put("hoodie.datasource.write.row.writer.enable", "false"); + + HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath()); + assertEquals(MetaFieldsMode.ALL, tc.getMetaFieldsMode()); + assertMetaColumnPopulation(basePath(), MetaFieldsMode.ALL); + } + + @Test + void nonRowWriterPathNoneMode() { + Map<String, String> options = baseOptions(); + options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false"); + options.put(DataSourceWriteOptions.OPERATION().key(), DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL()); + options.put("hoodie.datasource.write.row.writer.enable", "false"); + + HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath()); + assertEquals(MetaFieldsMode.NONE, tc.getMetaFieldsMode()); + assertMetaColumnPopulation(basePath(), MetaFieldsMode.NONE); + } + + @Test + void nonRowWriterPathCommitTimeOnly() { + Map<String, String> options = baseOptions(); + options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false"); + options.put(HoodieTableConfig.META_FIELDS_MODE.key(), MetaFieldsMode.COMMIT_TIME_ONLY.name()); + options.put(DataSourceWriteOptions.OPERATION().key(), DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL()); + options.put("hoodie.datasource.write.row.writer.enable", "false"); + + HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath()); + assertEquals(MetaFieldsMode.COMMIT_TIME_ONLY, tc.getMetaFieldsMode()); + assertMetaColumnPopulation(basePath(), MetaFieldsMode.COMMIT_TIME_ONLY); + } + + @Test + void nonRowWriterPathFileNameOnly() { + Map<String, String> options = baseOptions(); + options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false"); + options.put(HoodieTableConfig.META_FIELDS_MODE.key(), MetaFieldsMode.FILE_NAME_ONLY.name()); + options.put(DataSourceWriteOptions.OPERATION().key(), DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL()); + options.put("hoodie.datasource.write.row.writer.enable", "false"); + + HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath()); + assertEquals(MetaFieldsMode.FILE_NAME_ONLY, tc.getMetaFieldsMode()); + assertMetaColumnPopulation(basePath(), MetaFieldsMode.FILE_NAME_ONLY); + } + + @Test + void nonRowWriterPathCommitTimeAndFileName() { + Map<String, String> options = baseOptions(); + options.put(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false"); + options.put(HoodieTableConfig.META_FIELDS_MODE.key(), MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME.name()); + options.put(DataSourceWriteOptions.OPERATION().key(), DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL()); + options.put("hoodie.datasource.write.row.writer.enable", "false"); + + HoodieTableConfig tc = writeSampleAndGetTableConfig(options, basePath()); + assertEquals(MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME, tc.getMetaFieldsMode()); + assertMetaColumnPopulation(basePath(), MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME); + } + + // ------------------------------------------------------------------------- + // Clustering coverage. + // + // These target 358fbfdd717a, where HoodieRowCreateHandle's selective path copied the *source + // row's* _hoodie_file_name during clustering, leaving records pointing at a file clustering had + // just replaced. asserting only assertNotNull cannot catch that — the stale value is non-null too + // — so the assertion here compares the column against the file actually holding the row. + // + // Only the selective modes are covered: ALL and NONE route through writeRow / + // writeRowNoMetaFields and never enter the branch the fix touched. + // ------------------------------------------------------------------------- + + private Map<String, String> inlineClusteringOptions(MetaFieldsMode mode) { + Map<String, String> options = baseOptions(); + options.put(HoodieTableConfig.META_FIELDS_MODE.key(), mode.name()); + options.put(DataSourceWriteOptions.OPERATION().key(), DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL()); + options.put("hoodie.clustering.inline", "true"); + options.put("hoodie.clustering.inline.max.commits", "1"); + options.put("hoodie.clustering.plan.strategy.target.file.max.bytes", "10485760"); + options.put("hoodie.clustering.plan.strategy.small.file.limit", "10485760"); + return options; + } + + /** + * Asserts clustering actually ran and that every surviving row's {@code _hoodie_file_name} names + * the file holding it. + * + * <p>Reads through Hudi rather than globbing the parquet directly: after inline clustering the + * pre-clustering file is still on disk (no cleaning has run), so a raw glob would also inspect + * rows that were replaced and are no longer served. + */ + private void assertClusteredFileNamesPointAtTheirOwnFile(String path, MetaFieldsMode mode) { + HoodieTableMetaClient metaClient = + HoodieTableMetaClient.builder().setBasePath(path).setConf(storageConf()).build(); + assertEquals(mode, metaClient.getTableConfig().getMetaFieldsMode()); + assertEquals(1, metaClient.getActiveTimeline().getCompletedReplaceTimeline().countInstants(), + "clustering must have produced a replacecommit, otherwise this test proves nothing"); + + List<Row> rows = spark().read().format("hudi").load(path) + .withColumn("__containing_file", functions.input_file_name()) + .collectAsList(); + assertFalse(rows.isEmpty(), "expected the clustered table to still serve rows"); + + for (Row row : rows) { + String fileName = row.getAs(HoodieRecord.FILENAME_METADATA_FIELD); + String containingFile = row.getAs("__containing_file").toString(); + if (mode.isFileNamePopulated()) { + assertNotNull(fileName, "file name is opted in, so clustered rows must carry one"); + assertTrue(containingFile.endsWith("/" + fileName), + "_hoodie_file_name must name the file holding the row after clustering, not the " + + "pre-clustering file it was read from; got " + fileName + " inside " + containingFile); + } else { + assertNull(fileName, + "file name is not opted in, so clustering must not populate it; got " + fileName); + } + } + } + + @Test + void clusteringWritesTheNewFileNameUnderFileNameOnly() { + Map<String, String> options = inlineClusteringOptions(MetaFieldsMode.FILE_NAME_ONLY); + writeRows(Arrays.asList( + RowFactory.create("k1", "p1", "v1"), + RowFactory.create("k2", "p1", "v2"), + RowFactory.create("k3", "p1", "v3")), + simpleSchema(), options, basePath(), SaveMode.Overwrite); + + assertClusteredFileNamesPointAtTheirOwnFile(basePath(), MetaFieldsMode.FILE_NAME_ONLY); + } + + @Test + void clusteringWritesTheNewFileNameUnderCommitTimeAndFileName() { + Map<String, String> options = inlineClusteringOptions(MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME); + writeRows(Arrays.asList( + RowFactory.create("k1", "p1", "v1"), + RowFactory.create("k2", "p1", "v2"), + RowFactory.create("k3", "p1", "v3")), + simpleSchema(), options, basePath(), SaveMode.Overwrite); + + assertClusteredFileNamesPointAtTheirOwnFile(basePath(), MetaFieldsMode.COMMIT_TIME_AND_FILE_NAME); + } + + @Test + void clusteringLeavesFileNameNullUnderCommitTimeOnly() { Review Comment: Dug into the null. Short answer to your question: yes, `format("hudi")` does serve `_hoodie_commit_time` on a `COMMIT_TIME_ONLY` table, and this suite already proves it -- `appendWithoutRestatingTheModeKeepsTheTableSelective` reads through `format("hudi")` on a non-clustered `COMMIT_TIME_ONLY` table and asserts a non-null commit time on every row (`TestMetaFieldsModeE2E:832-838`), green at `6c8f179`. I also walked the snapshot read path looking for a meta-field special case and there is none: `FileGroupReaderSchemaHandler.generateRequiredSchema` returns the requested schema verbatim when the slice has no log files, and neither `HoodieFileGroupReaderBasedFileFormat` nor `SparkFileFormatInternalRowReaderContext` strips or null-pads meta columns. The read is faithful. So your assertion was not reading the wrong thing. The clustered files physically carry a null `_hoodie_commit_time`. This thread found a real bug. Two things point at the same place: 1. The glob check you removed was failing for the same reason, not because of stale files. The helper comment says pre-clustering files "may predate the current mode", but in this fixture they cannot: `clusteringLeavesFileNameNullUnderCommitTimeOnly` writes with `SaveMode.Overwrite` and `inlineClusteringOptions` sets the mode on that very first write, so every parquet file on disk was written under `COMMIT_TIME_ONLY`. The pre-clustering file passes both glob assertions; only post-clustering rows can fail it. The raw-parquet failure was independent confirmation of the null, and removing the call hid it. 2. The asymmetry you observed is exactly the shape of `writeRowSelectiveMetaFields` (`HoodieRowCreateHandle:184-201`): `_hoodie_file_name` is stamped fresh from the handle's own file, so it is always correct; `_hoodie_commit_time` is copied from the incoming row when `shouldPreserveHoodieMetadata` is set, which clustering defaults to true (`MultipleSparkJobExecutionStrategy:104`). For a clustered row to read back with a correct file name and a null commit time, `row.getUTF8String(COMMIT_TIME_METADATA_FIELD_ORD)` must already have been null -- the clustering input carried the null and preservation persisted it. The input rows come from `readRecordsForGroupAsRow`'s file-group read. Why that read returns null at ordinal 0 while the same machinery on the query path serves it populated is the one hop I could not settle by inspection. The two setups differ in exactly two ways, worth probing first: - Schema provenance: the query path prunes the schema resolved from commit metadata; clustering builds `addMetadataFields(TableSchemaResolver.getTableSchema(false), allowOperationMetadataField())` (`MultipleSparkJobExecutionStrategy:253-256`), i.e. synthesized meta fields, plus possibly `_hoodie_operation` at ordinal 5 which the file does not have. - Reader context: the query path hands the parquet reader `sparkRequiredSchema`; `SparkReaderContextFactory.getContext()` leaves it empty and uses the broadcast non-vectorized reader. `inputRecords.select(_hoodie_commit_time).show()` before `performClusteringWithRecordsAsRow` will split "FG read returned null" from "writer dropped it" in one run. Given the above: please restore the withdrawn direct-comparison assertion (it is the regression test for this), re-point `assertMetaColumnPopulation` at the clustered `COMMIT_TIME_ONLY` table (the physical half; with this fixture the stale-file caveat does not apply), and fix the clustering input read. As it stands clustering destroys the one column a `COMMIT_TIME_ONLY` table exists to keep, and any incremental read window that crosses the clustering commit silently loses those rows -- so this is a blocker for the mode, not a test gap. -- 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]
