This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 34e15860082b fix(reader): give the file group reader schema handler
the merged record-merge properties (#19389)
34e15860082b is described below
commit 34e15860082b118c098dccdb84034b8fb36bf14b
Author: Vova Kolmakov <[email protected]>
AuthorDate: Tue Jul 28 20:08:37 2026 +0700
fix(reader): give the file group reader schema handler the merged
record-merge properties (#19389)
* fix(reader): give the file group reader schema handler the merged
record-merge properties
* fix(reader): merge the record-merge properties into the constructor
parameter
Assigning the merged properties back to the parameter keeps the parameter
and the field pointing at the same object, so the schema handler, the record
merger and the merge-type lookup all see the merged set regardless of which one
a reference binds to. This replaces the mergedProps local with a two-line
change against master.
Renaming the parameter is not an option: both constructors carry Lombok's
@Builder(setterPrefix = "with"), which derives the builder setter name from the
parameter, so oriProps would rename withProps and break its 28 call sites
across hudi-trino, hudi-flink, hudi-spark-common, hudi-client-common, hudi-cli,
hudi-hadoop-mr and the metadata readers.
---------
Co-authored-by: Vova Kolmakov <[email protected]>
---
.../common/table/read/HoodieFileGroupReader.java | 3 +-
.../table/read/lsm/HoodieLsmFileGroupReader.java | 3 +-
.../read/TestFileGroupReaderDeleteMarkerProps.java | 219 +++++++++++++++++++++
.../read/TestHoodieFileGroupReaderOnSpark.scala | 37 ++--
4 files changed, 248 insertions(+), 14 deletions(-)
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodieFileGroupReader.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodieFileGroupReader.java
index 0d6b59c907cd..c95e5c5acd78 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodieFileGroupReader.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodieFileGroupReader.java
@@ -197,7 +197,8 @@ public final class HoodieFileGroupReader<T> implements
HoodieRecordReader<T> {
throw new IllegalArgumentException("Filegroup reader is doing log file
merge but not reading from the start of the base file");
}
HoodieTableConfig tableConfig = hoodieTableMetaClient.getTableConfig();
- this.props = ConfigUtils.getMergeProps(props, tableConfig);
+ props = ConfigUtils.getMergeProps(props, tableConfig);
+ this.props = props;
this.partitionPathFields = tableConfig.getPartitionFields();
readerContext.initRecordMerger(props);
readerContext.setTablePath(tablePath);
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/lsm/HoodieLsmFileGroupReader.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/lsm/HoodieLsmFileGroupReader.java
index da4154993f18..7fc28a4f3559 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/lsm/HoodieLsmFileGroupReader.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/lsm/HoodieLsmFileGroupReader.java
@@ -167,7 +167,8 @@ public final class HoodieLsmFileGroupReader<T> implements
HoodieRecordReader<T>
throw new IllegalArgumentException("LSM file group reader is doing log
file merge but not reading from the start of the base file");
}
HoodieTableConfig tableConfig = hoodieTableMetaClient.getTableConfig();
- this.props = ConfigUtils.getMergeProps(props, tableConfig);
+ props = ConfigUtils.getMergeProps(props, tableConfig);
+ this.props = props;
readerContext.initRecordMerger(props);
readerContext.setTablePath(tablePath);
readerContext.setLatestCommitTime(latestCommitTime);
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestFileGroupReaderDeleteMarkerProps.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestFileGroupReaderDeleteMarkerProps.java
new file mode 100644
index 000000000000..5646e34aaa20
--- /dev/null
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestFileGroupReaderDeleteMarkerProps.java
@@ -0,0 +1,219 @@
+/*
+ * 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.table.read;
+
+import org.apache.hudi.common.config.RecordMergeMode;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.common.util.ConfigUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.storage.StoragePath;
+
+import org.apache.avro.SchemaBuilder;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static
org.apache.hudi.common.model.DefaultHoodieRecordPayload.DELETE_KEY;
+import static
org.apache.hudi.common.model.DefaultHoodieRecordPayload.DELETE_MARKER;
+import static
org.apache.hudi.common.table.HoodieTableConfig.RECORD_MERGE_PROPERTY_PREFIX;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * A table at version 9 or later persists a custom delete marker only under the
+ * {@code hoodie.record.merge.property.} prefix, and {@link
ConfigUtils#getMergeProps} is what strips that
+ * prefix back to the plain {@code hoodie.payload.delete.field} / {@code
.marker} keys that
+ * {@link DeleteContext} reads. Read-path callers hand {@link
HoodieFileGroupReader} the properties as they
+ * come, so the reader is the only thing that can perform that merge before
the schema handler is built.
+ */
+public class TestFileGroupReaderDeleteMarkerProps {
+
+ private static final String DELETE_FIELD = "op";
+ private static final String DELETE_VALUE = "D";
+
+ private static final HoodieSchema TABLE_SCHEMA = HoodieSchema.fromAvroSchema(
+ SchemaBuilder.record("rec").fields()
+ .requiredString(HoodieRecord.RECORD_KEY_METADATA_FIELD)
+ .requiredString("key")
+ .requiredLong("ts")
+ .requiredString(DELETE_FIELD)
+ .endRecord());
+
+ private static final HoodieSchema REQUESTED_SCHEMA =
HoodieSchema.fromAvroSchema(
+ SchemaBuilder.record("rec").fields()
+ .requiredString("key")
+ .endRecord());
+
+ private static final HoodieSchema REQUESTED_SCHEMA_WITH_DELETE_FIELD =
HoodieSchema.fromAvroSchema(
+ SchemaBuilder.record("rec").fields()
+ .requiredString("key")
+ .requiredString(DELETE_FIELD)
+ .endRecord());
+
+ private static HoodieTableConfig
versionNineTableConfigWithCustomDeleteMarker() {
+ HoodieTableConfig tableConfig = new HoodieTableConfig();
+ tableConfig.setValue(HoodieTableConfig.VERSION,
String.valueOf(HoodieTableVersion.NINE.versionCode()));
+ tableConfig.setValue(HoodieTableConfig.RECORD_MERGE_MODE,
RecordMergeMode.COMMIT_TIME_ORDERING.name());
+ tableConfig.setValue(RECORD_MERGE_PROPERTY_PREFIX + DELETE_KEY,
DELETE_FIELD);
+ tableConfig.setValue(RECORD_MERGE_PROPERTY_PREFIX + DELETE_MARKER,
DELETE_VALUE);
+ return tableConfig;
+ }
+
+ /**
+ * The properties a query engine hands the reader: the table's own
properties, in which the custom delete
+ * marker only exists in its prefixed form. Nothing un-prefixes them before
the reader is built.
+ */
+ private static TypedProperties readerProps(HoodieTableConfig tableConfig) {
+ return TypedProperties.copy(tableConfig.getProps());
+ }
+
+ /**
+ * Builds a file group reader over a log-file-bearing split and returns the
schema handler it installed on
+ * the reader context.
+ */
+ private static FileGroupReaderSchemaHandler<String>
schemaHandlerOfReader(TypedProperties properties,
+
HoodieTableConfig tableConfig) {
+ return schemaHandlerOfReader(properties, tableConfig, REQUESTED_SCHEMA);
+ }
+
+ private static FileGroupReaderSchemaHandler<String>
schemaHandlerOfReader(TypedProperties properties,
+
HoodieTableConfig tableConfig,
+
HoodieSchema requestedSchema) {
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class,
RETURNS_DEEP_STUBS);
+ when(metaClient.getTableConfig()).thenReturn(tableConfig);
+ when(metaClient.getBasePath()).thenReturn(new
StoragePath("file:///tmp/hoodie_test_table"));
+
+ AtomicReference<FileGroupReaderSchemaHandler<String>> installedHandler =
new AtomicReference<>();
+ HoodieReaderContext<String> readerContext =
mock(HoodieReaderContext.class, RETURNS_DEEP_STUBS);
+ // The reader context reports log files so the schema handler takes the
merge path; the split itself stays
+ // empty because the reader is only built, never iterated - no file is
ever opened.
+ when(readerContext.getHasLogFiles()).thenReturn(true);
+ when(readerContext.getHasBootstrapBaseFile()).thenReturn(false);
+ when(readerContext.getInstantRange()).thenReturn(Option.empty());
+
when(readerContext.getMergeMode()).thenReturn(RecordMergeMode.COMMIT_TIME_ORDERING);
+
when(readerContext.getRecordContext().supportsParquetRowIndex()).thenReturn(false);
+ doAnswer(invocation -> {
+ installedHandler.set(invocation.getArgument(0));
+ return null;
+ }).when(readerContext).setSchemaHandler(any());
+ when(readerContext.getSchemaHandler()).thenAnswer(invocation ->
installedHandler.get());
+
+ HoodieFileGroupReader.<String>builder()
+ .withReaderContext(readerContext)
+ .withHoodieTableMetaClient(metaClient)
+ .withLatestCommitTime("001")
+ .withDataSchema(TABLE_SCHEMA)
+ .withRequestedSchema(requestedSchema)
+ .withProps(properties)
+ .withLogFiles(Stream.empty())
+ .withPartitionPath("")
+ .withStart(0L)
+ .withLength(Long.MAX_VALUE)
+ .build();
+
+ return installedHandler.get();
+ }
+
+ private static List<String>
requiredFieldNames(FileGroupReaderSchemaHandler<String> handler) {
+ return handler.getRequiredSchema().getFields().stream().map(field ->
field.name()).collect(Collectors.toList());
+ }
+
+ /**
+ * Sanity check on the premise: the un-prefixing lives in getMergeProps, so
the properties the reader is
+ * handed do not carry the plain delete keys at all.
+ */
+ @Test
+ public void mergePropsIsWhatUnprefixesTheDeleteMarker() {
+ HoodieTableConfig tableConfig =
versionNineTableConfigWithCustomDeleteMarker();
+ TypedProperties props = readerProps(tableConfig);
+
+ assertNull(props.getProperty(DELETE_KEY), "reader props must not carry the
unprefixed delete key");
+ assertEquals(DELETE_FIELD, ConfigUtils.getMergeProps(props,
tableConfig).getProperty(DELETE_KEY));
+
+ assertTrue(new DeleteContext(props,
TABLE_SCHEMA).getCustomDeleteMarkerKeyValue().isEmpty(),
+ "DeleteContext built from the reader props sees no custom delete
marker");
+ assertTrue(new DeleteContext(ConfigUtils.getMergeProps(props,
tableConfig), TABLE_SCHEMA)
+ .getCustomDeleteMarkerKeyValue().isPresent(),
+ "DeleteContext built from merged props sees the custom delete marker");
+ }
+
+ /**
+ * The reader must merge the table's record-merge properties in before
building the schema handler.
+ * Otherwise the handler's DeleteContext carries no marker, and since
FileGroupRecordBuffer takes its
+ * DeleteContext from that very handler, custom deletes stop being
recognised on the whole read path.
+ */
+ @Test
+ public void readerResolvesTheCustomDeleteMarkerFromPrefixedTableProps() {
+ HoodieTableConfig tableConfig =
versionNineTableConfigWithCustomDeleteMarker();
+
+ FileGroupReaderSchemaHandler<String> handler =
schemaHandlerOfReader(readerProps(tableConfig), tableConfig);
+
+
assertTrue(handler.getDeleteContext().getCustomDeleteMarkerKeyValue().isPresent(),
+ "the schema handler the reader installs must resolve the custom delete
marker");
+ assertTrue(requiredFieldNames(handler).contains(DELETE_FIELD),
+ "required schema must contain the custom delete column " +
DELETE_FIELD);
+ }
+
+ /**
+ * The delete column becomes a mandatory field, so it must not be appended
twice when the query already
+ * asked for it.
+ */
+ @Test
+ public void deleteColumnIsNotDuplicatedWhenAlreadyRequested() {
+ HoodieTableConfig tableConfig =
versionNineTableConfigWithCustomDeleteMarker();
+
+ FileGroupReaderSchemaHandler<String> handler =
+ schemaHandlerOfReader(readerProps(tableConfig), tableConfig,
REQUESTED_SCHEMA_WITH_DELETE_FIELD);
+
+ assertEquals(1,
requiredFieldNames(handler).stream().filter(DELETE_FIELD::equals).count(),
+ "the custom delete column must appear exactly once in the required
schema");
+ }
+
+ /**
+ * Control: a reader handed properties that already carry the plain delete
keys - which is how the existing
+ * engine tests are configured - resolves the marker either way, which is
why this defect stayed hidden.
+ */
+ @Test
+ public void readerAlsoResolvesAMarkerSuppliedInPlainForm() {
+ HoodieTableConfig tableConfig =
versionNineTableConfigWithCustomDeleteMarker();
+ TypedProperties props = readerProps(tableConfig);
+ props.setProperty(DELETE_KEY, DELETE_FIELD);
+ props.setProperty(DELETE_MARKER, DELETE_VALUE);
+
+ FileGroupReaderSchemaHandler<String> handler =
schemaHandlerOfReader(props, tableConfig);
+
+
assertTrue(handler.getDeleteContext().getCustomDeleteMarkerKeyValue().isPresent());
+ assertTrue(requiredFieldNames(handler).contains(DELETE_FIELD));
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderOnSpark.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderOnSpark.scala
index acfd3d306fb2..db2488c7c8e9 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderOnSpark.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderOnSpark.scala
@@ -233,7 +233,8 @@ class TestHoodieFileGroupReaderOnSpark extends
TestHoodieFileGroupReaderBase[Int
def testCustomDelete(useFgReader: String,
tableType: String,
positionUsed: String,
- mergeMode: String): Unit = {
+ mergeMode: String,
+ markerFromTableConfigOnly: String): Unit = {
val payloadClass =
"org.apache.hudi.common.table.read.CustomPayloadForTesting"
val fgReaderOpts: Map[String, String] = Map(
HoodieWriteConfig.MERGE_SMALL_FILE_GROUP_CANDIDATES_LIMIT.key -> "0",
@@ -243,13 +244,17 @@ class TestHoodieFileGroupReaderOnSpark extends
TestHoodieFileGroupReaderBase[Int
)
val deleteOpts: Map[String, String] = Map(
DELETE_KEY -> "op", DELETE_MARKER -> "d")
- val readOpts = if (mergeMode.equals("CUSTOM")) {
- fgReaderOpts ++ deleteOpts ++ Map(
- HoodieWriteConfig.WRITE_PAYLOAD_CLASS_NAME.key -> payloadClass)
+ val payloadOpts = if (mergeMode.equals("CUSTOM")) {
+ Map(HoodieWriteConfig.WRITE_PAYLOAD_CLASS_NAME.key -> payloadClass)
} else {
- fgReaderOpts ++ deleteOpts
+ Map.empty[String, String]
}
- val opts = readOpts
+ val opts = fgReaderOpts ++ deleteOpts ++ payloadOpts
+ // The write persists the marker on the table under the record-merge
property prefix. When the query does
+ // not restate the delete options, the table config is the only place the
reader can learn about them -
+ // which is what a query that just loads the path looks like.
+ val tableConfigOnly = markerFromTableConfigOnly.equals("true")
+ val readOpts = if (tableConfigOnly) fgReaderOpts ++ payloadOpts else opts
val columns = Seq("ts", "key", "rider", "driver", "fare", "op")
val data = Seq(
@@ -271,6 +276,11 @@ class TestHoodieFileGroupReaderOnSpark extends
TestHoodieFileGroupReaderBase[Int
val metaClient = HoodieTableMetaClient
.builder().setConf(getStorageConf).setBasePath(getBasePath).build
assertEquals((1, 0), getFileCount(metaClient, getBasePath))
+ if (tableConfigOnly) {
+ assertTrue(metaClient.getTableConfig.getProps.containsKey(
+ HoodieTableConfig.RECORD_MERGE_PROPERTY_PREFIX + DELETE_KEY),
+ "the delete marker must be persisted on the table, otherwise the read
side has nothing to pick up")
+ }
// Delete using delete markers.
val updateData = Seq(
@@ -458,12 +468,15 @@ class TestHoodieFileGroupReaderOnSpark extends
TestHoodieFileGroupReaderBase[Int
object TestHoodieFileGroupReaderOnSpark {
def customDeleteTestParams(): java.util.List[Arguments] = {
java.util.Arrays.asList(
- Arguments.of("true", "MERGE_ON_READ", "false", "EVENT_TIME_ORDERING"),
- Arguments.of("true", "MERGE_ON_READ", "true", "EVENT_TIME_ORDERING"),
- Arguments.of("true", "MERGE_ON_READ", "false", "COMMIT_TIME_ORDERING"),
- Arguments.of("true", "MERGE_ON_READ", "true", "COMMIT_TIME_ORDERING"),
- Arguments.of("true", "MERGE_ON_READ", "false", "CUSTOM"),
- Arguments.of("true", "MERGE_ON_READ", "true", "CUSTOM"))
+ Arguments.of("true", "MERGE_ON_READ", "false", "EVENT_TIME_ORDERING",
"false"),
+ Arguments.of("true", "MERGE_ON_READ", "true", "EVENT_TIME_ORDERING",
"false"),
+ Arguments.of("true", "MERGE_ON_READ", "false", "COMMIT_TIME_ORDERING",
"false"),
+ Arguments.of("true", "MERGE_ON_READ", "true", "COMMIT_TIME_ORDERING",
"false"),
+ Arguments.of("true", "MERGE_ON_READ", "false", "CUSTOM", "false"),
+ Arguments.of("true", "MERGE_ON_READ", "true", "CUSTOM", "false"),
+ // The query does not restate the delete options, so the table config is
the reader's only source.
+ Arguments.of("true", "MERGE_ON_READ", "false", "EVENT_TIME_ORDERING",
"true"),
+ Arguments.of("true", "MERGE_ON_READ", "false", "COMMIT_TIME_ORDERING",
"true"))
}
def getFileCount(metaClient: HoodieTableMetaClient, basePath: String):
(Long, Long) = {
