This is an automated email from the ASF dual-hosted git repository.
danny0405 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 d9e44c5c519a feat(reader): Adapt the HoodieFileGroupReader to read the
native format log files (#19072)
d9e44c5c519a is described below
commit d9e44c5c519a97031e4bb665190e8a5c2eb79ee7
Author: Shuo Cheng <[email protected]>
AuthorDate: Tue Jun 30 13:46:31 2026 +0800
feat(reader): Adapt the HoodieFileGroupReader to read the native format log
files (#19072)
---
.../hudi/io/HoodieNativeLogFormatWriter.java | 20 +-
.../java/org/apache/hudi/common/fs/FSUtils.java | 13 +-
.../hudi/common/table/TableSchemaResolver.java | 7 +
.../table/log/BaseHoodieLogRecordReader.java | 2 +-
.../common/table/log/HoodieLogFormatReader.java | 37 ++-
.../table/log/HoodieNativeLogFileReader.java | 139 +++++++++
.../common/table/log/NativeLogFooterMetadata.java | 171 +++++++++++
.../common/table/log/block/HoodieDataBlock.java | 2 +-
.../common/table/log/block/HoodieDeleteBlock.java | 12 +
.../table/log/block/HoodieNativeDataBlock.java | 133 ++++++++
.../table/log/block/HoodieNativeDeleteBlock.java | 106 +++++++
.../common/table/read/BufferedRecordMerger.java | 4 +
.../table/read/BufferedRecordMergerFactory.java | 22 ++
.../read/buffer/HoodieFileGroupRecordBuffer.java | 6 -
.../read/buffer/KeyBasedFileGroupRecordBuffer.java | 18 +-
.../buffer/PositionBasedFileGroupRecordBuffer.java | 16 +-
.../read/buffer/ReusableKeyBasedRecordBuffer.java | 6 -
.../read/buffer/UnmergedFileGroupRecordBuffer.java | 6 -
.../table/view/AbstractTableFileSystemView.java | 6 +-
.../table/log/TestHoodieNativeLogFileReader.java | 51 ++++
.../table/log/TestNativeLogFooterMetadata.java | 94 ++++++
.../read/buffer/BaseTestFileGroupRecordBuffer.java | 13 +
.../buffer/TestKeyBasedFileGroupRecordBuffer.java | 7 +-
.../TestSortedKeyBasedFileGroupRecordBuffer.java | 6 +-
.../hudi/common/table/TestTableSchemaResolver.java | 69 +++++
.../read/TestHoodieFileGroupReaderNativeLogs.java | 338 +++++++++++++++++++++
.../table/view/TestHoodieTableFileSystemView.java | 32 ++
.../reader/HoodieFileGroupReaderTestHarness.java | 7 +-
.../testutils/reader/HoodieFileSliceTestUtils.java | 187 ++++++++++++
29 files changed, 1446 insertions(+), 84 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogFormatWriter.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogFormatWriter.java
index 8c7867ec7ea8..1463b3b59f7a 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogFormatWriter.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogFormatWriter.java
@@ -30,11 +30,11 @@ import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.log.AppendResult;
import org.apache.hudi.common.table.log.HoodieLogFormat;
import org.apache.hudi.common.table.log.LogFileCreationCallback;
+import org.apache.hudi.common.table.log.NativeLogFooterMetadata;
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
import
org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
import org.apache.hudi.common.table.read.BufferedRecord;
import org.apache.hudi.common.table.read.BufferedRecords;
-import org.apache.hudi.common.util.JsonUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.OrderingValues;
import org.apache.hudi.common.util.collection.ArrayComparable;
@@ -47,8 +47,6 @@ import org.apache.hudi.storage.StoragePath;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collections;
-import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
@@ -61,8 +59,6 @@ import static
org.apache.hudi.common.model.LogExtensions.DELETE_LOG_EXTENSION;
*/
public class HoodieNativeLogFormatWriter extends HoodieLogFormat.Writer {
- private static final String LOG_FORMAT_METADATA_FOOTER_KEY =
"hudi.log.format.metadata";
-
private final HoodieWriteConfig writeConfig;
private final HoodieFileFormat nativeFileFormat;
private final HoodieSchema tableSchema;
@@ -215,22 +211,10 @@ public class HoodieNativeLogFormatWriter extends
HoodieLogFormat.Writer {
private void addFooterMetadata(Map<HeaderMetadataType, String> header)
throws IOException {
if (dataFileWriter != null) {
- dataFileWriter.addFooterMetadata(getFooterMetadata(header));
+
dataFileWriter.addFooterMetadata(NativeLogFooterMetadata.toFooterMetadata(header));
}
}
- private Map<String, String> getFooterMetadata(Map<HeaderMetadataType,
String> header) throws IOException {
- Map<String, String> logFormatMetadata = new LinkedHashMap<>();
- logFormatMetadata.put(HeaderMetadataType.VERSION.name(),
String.valueOf(HoodieLogFormat.CURRENT_VERSION));
- header.forEach((key, value) -> {
- if (value != null) {
- logFormatMetadata.put(key.name(), value);
- }
- });
- return Collections.singletonMap(
- LOG_FORMAT_METADATA_FOOTER_KEY,
JsonUtils.getObjectMapper().writeValueAsString(logFormatMetadata));
- }
-
private void ensureDataFileWriter(HoodieSchema recordSchema) throws
IOException {
ensureAppendVersion();
if (dataFileWriter == null) {
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java
index 7b5d0b7002ee..ff39c9f23358 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java
@@ -865,7 +865,18 @@ public class FSUtils {
private static Option<HoodieLogFile> getLatestLogFile(Stream<HoodieLogFile>
logFiles) {
return
Option.fromJavaOptional(logFiles.min(HoodieLogFile.getReverseLogFileComparator()));
}
-
+
+ /**
+ * Return the file size of a log file.
+ */
+ public static long getFileSize(HoodieStorage storage, HoodieLogFile logFile)
{
+ try {
+ return logFile.getFileSize() >= 0 ? logFile.getFileSize() :
storage.getPathInfo(logFile.getPath()).getLength();
+ } catch (IOException e) {
+ throw new HoodieIOException("Unable to get file size for " + logFile, e);
+ }
+ }
+
public static Map<String, Boolean> deleteFilesParallelize(
HoodieTableMetaClient metaClient,
List<String> paths,
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/TableSchemaResolver.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/TableSchemaResolver.java
index 157f9d4c79ca..2b98896fe6c7 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/TableSchemaResolver.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/TableSchemaResolver.java
@@ -29,6 +29,7 @@ import org.apache.hudi.common.schema.HoodieSchemaType;
import org.apache.hudi.common.schema.HoodieSchemaUtils;
import org.apache.hudi.common.table.log.HoodieLogFormat;
import org.apache.hudi.common.table.log.HoodieLogFormat.Reader;
+import org.apache.hudi.common.table.log.NativeLogFooterMetadata;
import org.apache.hudi.common.table.log.block.HoodieDataBlock;
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
import org.apache.hudi.common.table.timeline.HoodieInstant;
@@ -61,6 +62,7 @@ import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Supplier;
+import java.util.regex.Matcher;
import java.util.stream.Stream;
/**
@@ -286,6 +288,11 @@ public class TableSchemaResolver {
* @return
*/
public static HoodieSchema readSchemaFromLogFile(HoodieStorage storage,
StoragePath path) throws IOException {
+ Option<Matcher> nativeLogMatcherOpt =
FSUtils.matchNativeLogFile(path.getName());
+ if (nativeLogMatcherOpt.isPresent()) {
+ return NativeLogFooterMetadata.readSchemaFromNativeLogFile(storage,
path, nativeLogMatcherOpt.get());
+ }
+
// We only need to read the schema from the log block header,
// so we read the block lazily to avoid reading block content
// containing the records
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/BaseHoodieLogRecordReader.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/BaseHoodieLogRecordReader.java
index 159662f73bb9..4fbb1cf8dfc0 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/BaseHoodieLogRecordReader.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/BaseHoodieLogRecordReader.java
@@ -226,7 +226,7 @@ public abstract class BaseHoodieLogRecordReader<T> {
HoodieLogFormatReader logFormatReaderWrapper = null;
try {
// Iterate over the paths
- logFormatReaderWrapper = new HoodieLogFormatReader(storage, logFiles,
+ logFormatReaderWrapper = new HoodieLogFormatReader(storage,
readerContext, hoodieTableMetaClient, logFiles,
readerSchema, reverseReader, bufferSize, shouldLookupRecords(),
recordKeyField, internalSchema);
/**
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormatReader.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormatReader.java
index 1c342fecf39c..d501cbc4a1f7 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormatReader.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormatReader.java
@@ -18,10 +18,15 @@
package org.apache.hudi.common.table.log;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
+import org.apache.hudi.common.util.HoodieRecordUtils;
import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.exception.HoodieNotSupportedException;
import org.apache.hudi.internal.schema.InternalSchema;
import org.apache.hudi.storage.HoodieStorage;
@@ -37,10 +42,12 @@ import java.util.List;
public class HoodieLogFormatReader implements HoodieLogFormat.Reader {
private final List<HoodieLogFile> logFiles;
- private HoodieLogFileReader currentReader;
+ private HoodieLogFormat.Reader currentReader;
private final HoodieStorage storage;
private final HoodieSchema readerSchema;
private final InternalSchema internalSchema;
+ private final HoodieReaderContext<?> readerContext;
+ private final HoodieTableMetaClient metaClient;
private final String recordKeyField;
private final boolean enableInlineReading;
private final int bufferSize;
@@ -48,8 +55,18 @@ public class HoodieLogFormatReader implements
HoodieLogFormat.Reader {
HoodieLogFormatReader(HoodieStorage storage, List<HoodieLogFile> logFiles,
HoodieSchema readerSchema,
boolean reverseLogReader, int bufferSize, boolean
enableRecordLookups,
String recordKeyField, InternalSchema internalSchema)
throws IOException {
+ this(storage, null, null, logFiles, readerSchema, reverseLogReader,
bufferSize, enableRecordLookups,
+ recordKeyField, internalSchema);
+ }
+
+ HoodieLogFormatReader(HoodieStorage storage, HoodieReaderContext<?>
readerContext, HoodieTableMetaClient metaClient,
+ List<HoodieLogFile> logFiles, HoodieSchema
readerSchema,
+ boolean reverseLogReader, int bufferSize, boolean
enableRecordLookups,
+ String recordKeyField, InternalSchema internalSchema)
throws IOException {
this.logFiles = logFiles;
this.storage = storage;
+ this.readerContext = readerContext;
+ this.metaClient = metaClient;
this.readerSchema = readerSchema;
this.bufferSize = bufferSize;
this.recordKeyField = recordKeyField;
@@ -57,8 +74,7 @@ public class HoodieLogFormatReader implements
HoodieLogFormat.Reader {
this.internalSchema = internalSchema == null ?
InternalSchema.getEmptyInternalSchema() : internalSchema;
if (!logFiles.isEmpty()) {
HoodieLogFile nextLogFile = logFiles.remove(0);
- this.currentReader = new HoodieLogFileReader(storage, nextLogFile,
readerSchema, bufferSize, false,
- enableRecordLookups, recordKeyField, internalSchema);
+ this.currentReader = createReader(nextLogFile, reverseLogReader);
}
}
@@ -84,8 +100,7 @@ public class HoodieLogFormatReader implements
HoodieLogFormat.Reader {
try {
HoodieLogFile nextLogFile = logFiles.remove(0);
this.currentReader.close();
- this.currentReader = new HoodieLogFileReader(storage, nextLogFile,
readerSchema, bufferSize, false,
- enableInlineReading, recordKeyField, internalSchema);
+ this.currentReader = createReader(nextLogFile, false);
} catch (IOException io) {
throw new HoodieIOException("unable to initialize read with log file
", io);
}
@@ -118,4 +133,16 @@ public class HoodieLogFormatReader implements
HoodieLogFormat.Reader {
public HoodieLogBlock prev() throws IOException {
return this.currentReader.prev();
}
+
+ private HoodieLogFormat.Reader createReader(HoodieLogFile logFile, boolean
reverseLogReader) throws IOException {
+ if (FSUtils.matchNativeLogFile(logFile.getFileName()).isPresent()) {
+ if (readerContext == null || metaClient == null) {
+ throw new HoodieNotSupportedException("Native log files require
HoodieFileGroupReader based reader context");
+ }
+ return new HoodieNativeLogFileReader(storage, logFile, readerContext,
readerSchema,
+
HoodieRecordUtils.getOrderingFieldNames(readerContext.getMergeMode(),
metaClient));
+ }
+ return new HoodieLogFileReader(storage, logFile, readerSchema, bufferSize,
reverseLogReader,
+ enableInlineReading, recordKeyField, internalSchema);
+ }
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieNativeLogFileReader.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieNativeLogFileReader.java
new file mode 100644
index 000000000000..cbb7a6b514c4
--- /dev/null
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieNativeLogFileReader.java
@@ -0,0 +1,139 @@
+/*
+ * 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.log;
+
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemas;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock;
+import
org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.common.table.log.block.HoodieNativeDataBlock;
+import org.apache.hudi.common.table.log.block.HoodieNativeDeleteBlock;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.exception.HoodieNotSupportedException;
+import org.apache.hudi.io.storage.HoodieIOFactory;
+import org.apache.hudi.storage.HoodieStorage;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+
+/**
+ * Reader for native log files.
+ */
+public class HoodieNativeLogFileReader implements HoodieLogFormat.Reader {
+
+ private final HoodieStorage storage;
+ private final HoodieLogFile logFile;
+ private final HoodieReaderContext<?> readerContext;
+ private final HoodieSchema readerSchema;
+ private final List<String> orderingFieldNames;
+ private boolean consumed;
+
+ HoodieNativeLogFileReader(HoodieStorage storage, HoodieLogFile logFile,
HoodieReaderContext<?> readerContext,
+ HoodieSchema readerSchema, List<String>
orderingFieldNames) {
+ this.storage = storage;
+ this.logFile = logFile;
+ this.readerContext = readerContext;
+ this.readerSchema = readerSchema;
+ this.orderingFieldNames = orderingFieldNames;
+ }
+
+ @Override
+ public HoodieLogFile getLogFile() {
+ return logFile;
+ }
+
+ @Override
+ public boolean hasPrev() {
+ return false;
+ }
+
+ @Override
+ public HoodieLogBlock prev() {
+ throw new HoodieNotSupportedException("Reverse reading is not supported
for native log files");
+ }
+
+ @Override
+ public void close() {
+ // no-op
+ }
+
+ @Override
+ public boolean hasNext() {
+ return !consumed;
+ }
+
+ @Override
+ public HoodieLogBlock next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException("No more blocks in native log file " +
logFile);
+ }
+ consumed = true;
+ HoodieFileFormat fileFormat = getNativeFileFormat();
+ Map<HeaderMetadataType, String> header = getLogBlockHeader(fileFormat);
+ if (FSUtils.isNativeDeleteLogFile(logFile.getFileName())) {
+ HoodieSchema deleteLogSchema = HoodieSchemas.createDeleteLogSchema(
+ readerContext.getSchemaHandler().getTableSchema(),
orderingFieldNames);
+ return new HoodieNativeDeleteBlock(storage, logFile, readerContext,
deleteLogSchema,
+ orderingFieldNames, header, new HashMap<>());
+ }
+ validateDataBlockHeader(logFile, header);
+ return new HoodieNativeDataBlock(storage, logFile, fileFormat,
Option.ofNullable(readerSchema), header, new HashMap<>());
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException("Remove not supported for
HoodieNativeLogFileReader");
+ }
+
+ private Map<HeaderMetadataType, String> getLogBlockHeader(HoodieFileFormat
fileFormat) {
+ Map<String, String> keyValueMetadata =
HoodieIOFactory.getIOFactory(storage)
+ .getFileFormatUtils(fileFormat)
+ .readFooter(storage, false, logFile.getPath(),
NativeLogFooterMetadata.FOOTER_METADATA_KEY);
+ Map<HeaderMetadataType, String> header =
NativeLogFooterMetadata.fromFooterMetadata(keyValueMetadata);
+ header.putIfAbsent(HeaderMetadataType.INSTANT_TIME,
logFile.getDeltaCommitTime());
+ return header;
+ }
+
+ static void validateDataBlockHeader(HoodieLogFile logFile,
Map<HeaderMetadataType, String> header) {
+ if (!header.containsKey(HeaderMetadataType.SCHEMA)) {
+ throw new HoodieIOException("Missing required native log block header
metadata '"
+ + HeaderMetadataType.SCHEMA.name() + "' for " + logFile
+ + ". Native data logs must store log block metadata in footer key '"
+ + NativeLogFooterMetadata.FOOTER_METADATA_KEY + "'");
+ }
+ }
+
+ private HoodieFileFormat getNativeFileFormat() {
+ String suffix = logFile.getSuffix();
+ try {
+ return HoodieFileFormat.fromFileExtension("." + suffix);
+ } catch (IllegalArgumentException e) {
+ throw new HoodieNotSupportedException(
+ "Unsupported native log file format suffix '" + suffix + "' for " +
logFile);
+ }
+ }
+}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/NativeLogFooterMetadata.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/NativeLogFooterMetadata.java
new file mode 100644
index 000000000000..9e613a449747
--- /dev/null
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/NativeLogFooterMetadata.java
@@ -0,0 +1,171 @@
+/*
+ * 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.log;
+
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.schema.HoodieSchema;
+import
org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.common.util.JsonUtils;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.exception.HoodieNotSupportedException;
+import org.apache.hudi.io.storage.HoodieIOFactory;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+
+import java.io.IOException;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.regex.Matcher;
+
+/**
+ * Shared on-disk contract for RFC-103 native log files. The log block header
(schema, instant time,
+ * etc.) is persisted as a single JSON entry in the native file footer, keyed
by
+ * {@link #FOOTER_METADATA_KEY}. This is the single source of truth used by
both the write path
+ * ({@code HoodieNativeLogFormatWriter}) and the read path ({@code
HoodieNativeLogFileReader}) so the
+ * two cannot drift apart.
+ */
+public class NativeLogFooterMetadata {
+
+ /**
+ * Footer key under which the serialized log block header is stored in a
native log file.
+ */
+ public static final String FOOTER_METADATA_KEY = "hudi.log.format.metadata";
+
+ private static final TypeReference<LinkedHashMap<String, String>> MAP_TYPE =
+ new TypeReference<LinkedHashMap<String, String>>() {
+ };
+
+ private NativeLogFooterMetadata() {
+ }
+
+ /**
+ * Serializes the log block header into the footer key/value map written to
the native file.
+ * The {@link HeaderMetadataType#VERSION} entry is injected automatically.
+ */
+ public static Map<String, String> toFooterMetadata(Map<HeaderMetadataType,
String> header) {
+ Map<String, String> logFormatMetadata = new LinkedHashMap<>();
+ logFormatMetadata.put(HeaderMetadataType.VERSION.name(),
String.valueOf(HoodieLogFormat.CURRENT_VERSION));
+ header.forEach((key, value) -> {
+ if (value != null) {
+ logFormatMetadata.put(key.name(), value);
+ }
+ });
+ String serialized;
+ try {
+ serialized =
JsonUtils.getObjectMapper().writeValueAsString(logFormatMetadata);
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to serialize native log footer
metadata", e);
+ }
+ Map<String, String> footer = new LinkedHashMap<>();
+ footer.put(FOOTER_METADATA_KEY, serialized);
+ return footer;
+ }
+
+ /**
+ * Parses the log block header back from the native file footer key/value
map. Unknown header
+ * types are ignored so that files written by newer minor versions remain
readable.
+ */
+ public static Map<HeaderMetadataType, String> fromFooterMetadata(Map<String,
String> footerMetadata) {
+ Map<HeaderMetadataType, String> header = new LinkedHashMap<>();
+ String serialized = footerMetadata.get(FOOTER_METADATA_KEY);
+ if (serialized == null) {
+ return header;
+ }
+ Map<String, String> logFormatMetadata;
+ try {
+ logFormatMetadata = JsonUtils.getObjectMapper().readValue(serialized,
MAP_TYPE);
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to parse native log footer metadata:
" + serialized, e);
+ }
+ logFormatMetadata.forEach((name, value) -> {
+ HeaderMetadataType type = headerTypeOrNull(name);
+ if (type != null) {
+ header.put(type, value);
+ }
+ });
+ validateFormatVersion(header);
+ return header;
+ }
+
+ /**
+ * Reads the table schema embedded in a native data log file footer.
+ *
+ * <p>Native data logs persist the synthetic log-block header, including
{@link HeaderMetadataType#SCHEMA},
+ * in the footer metadata entry keyed by {@link #FOOTER_METADATA_KEY}. This
method is intended for
+ * schema discovery paths that do not have a {@code HoodieReaderContext} and
therefore cannot construct
+ * a {@code HoodieNativeLogFileReader}. Native delete logs do not expose a
generic table schema through
+ * this API because their schema depends on the table schema and configured
ordering fields; those files
+ * return {@code null}, matching the legacy behavior when no data-block
schema can be discovered.
+ *
+ * @param storage storage used to read the native log file footer
+ * @param path native log file path
+ * @param nativeLogMatcher matcher returned by {@code
FSUtils.matchNativeLogFile(path.getName())}
+ * @return table schema for native data logs, or {@code null} for native
non-data logs
+ */
+ public static HoodieSchema readSchemaFromNativeLogFile(
+ HoodieStorage storage, StoragePath path, Matcher nativeLogMatcher) {
+ String nativeLogType = nativeLogMatcher.group(8);
+ if (!"log".equals(nativeLogType)) {
+ return null;
+ }
+
+ HoodieFileFormat fileFormat = HoodieFileFormat.fromFileExtension("." +
nativeLogMatcher.group(9));
+ Map<String, String> footer = HoodieIOFactory.getIOFactory(storage)
+ .getFileFormatUtils(fileFormat)
+ .readFooter(storage, false, path, FOOTER_METADATA_KEY);
+ Map<HeaderMetadataType, String> header = fromFooterMetadata(footer);
+ String schema = header.get(HeaderMetadataType.SCHEMA);
+ if (StringUtils.isNullOrEmpty(schema)) {
+ throw new HoodieIOException("Missing required native log schema metadata
'"
+ + HeaderMetadataType.SCHEMA.name() + "' in footer key '"
+ + FOOTER_METADATA_KEY + "' for " + path);
+ }
+ return HoodieSchema.parse(schema);
+ }
+
+ private static void validateFormatVersion(Map<HeaderMetadataType, String>
header) {
+ String version = header.get(HeaderMetadataType.VERSION);
+ if (version == null) {
+ return;
+ }
+ int parsedVersion;
+ try {
+ parsedVersion = Integer.parseInt(version.trim());
+ } catch (NumberFormatException e) {
+ throw new HoodieIOException("Invalid native log format version in footer
metadata: " + version);
+ }
+ if (parsedVersion > HoodieLogFormat.CURRENT_VERSION) {
+ throw new HoodieNotSupportedException("Native log format version " +
parsedVersion
+ + " is newer than the supported version " +
HoodieLogFormat.CURRENT_VERSION
+ + ". Please upgrade Hudi to read this log file.");
+ }
+ }
+
+ private static HeaderMetadataType headerTypeOrNull(String name) {
+ try {
+ return HeaderMetadataType.valueOf(name);
+ } catch (IllegalArgumentException e) {
+ return null;
+ }
+ }
+}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDataBlock.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDataBlock.java
index dff48985f646..64e1ac881e22 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDataBlock.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDataBlock.java
@@ -469,7 +469,7 @@ public abstract class HoodieDataBlock extends
HoodieLogBlock {
*
* @param <T> The type of engine-specific record representation.
*/
- private static class FilteringEngineRecordIterator<T> implements
ClosableIterator<T> {
+ protected static class FilteringEngineRecordIterator<T> implements
ClosableIterator<T> {
private final ClosableIterator<T> nested; // nested iterator
private final Set<String> keys; // the filtering keys
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDeleteBlock.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDeleteBlock.java
index 0d69dab14258..cb2923f1f2fa 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDeleteBlock.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieDeleteBlock.java
@@ -21,9 +21,12 @@ package org.apache.hudi.common.table.log.block;
import org.apache.hudi.avro.HoodieAvroUtils;
import org.apache.hudi.avro.model.HoodieDeleteRecord;
import org.apache.hudi.avro.model.HoodieDeleteRecordList;
+import org.apache.hudi.common.engine.RecordContext;
import org.apache.hudi.common.fs.SizeAwareDataInputStream;
import org.apache.hudi.common.model.DeleteRecord;
import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.table.read.BufferedRecord;
+import org.apache.hudi.common.table.read.BufferedRecords;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.SerializationUtils;
import org.apache.hudi.common.util.collection.Pair;
@@ -131,6 +134,15 @@ public class HoodieDeleteBlock extends HoodieLogBlock {
}
}
+ /**
+ * Returns delete records in the same representation used by the file-group
record buffer.
+ */
+ public <T> List<BufferedRecord<T>> getRecordsToDelete(RecordContext<T>
recordContext) {
+ return Arrays.stream(getRecordsToDelete())
+ .map(deleteRecord -> BufferedRecords.fromDeleteRecord(deleteRecord,
recordContext))
+ .collect(Collectors.toList());
+ }
+
private byte[] serializeV2() throws IOException {
// Serialization for log block version 2
return SerializationUtils.serialize(getRecordsToDelete());
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieNativeDataBlock.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieNativeDataBlock.java
new file mode 100644
index 000000000000..805e84e89198
--- /dev/null
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieNativeDataBlock.java
@@ -0,0 +1,133 @@
+/*
+ * 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.log.block;
+
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.io.SeekableDataInputStream;
+import org.apache.hudi.io.storage.HoodieIOFactory;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+
+import static
org.apache.hudi.common.util.ConfigUtils.DEFAULT_HUDI_CONFIG_FOR_READER;
+
+/**
+ * Data block backed by a native log file.
+ */
+public class HoodieNativeDataBlock extends HoodieDataBlock {
+
+ private final HoodieStorage storage;
+ private final HoodieLogFile logFile;
+ private final HoodieFileFormat fileFormat;
+
+ public HoodieNativeDataBlock(HoodieStorage storage,
+ HoodieLogFile logFile,
+ HoodieFileFormat fileFormat,
+ Option<HoodieSchema> readerSchema,
+ Map<HeaderMetadataType, String> header,
+ Map<FooterMetadataType, String> footer) {
+ super(Option.empty(), null, true, getContentLocation(storage, logFile),
readerSchema,
+ header, footer, HoodieRecord.RECORD_KEY_METADATA_FIELD, false);
+ this.storage = storage;
+ this.logFile = logFile;
+ this.fileFormat = fileFormat;
+ }
+
+ @Override
+ public HoodieLogBlockType getBlockType() {
+ // This is a synthetic reader-side block for native log files. Returning
PARQUET_DATA_BLOCK
+ // keeps the existing data-block processing/statistics path; native file
decoding is driven
+ // by fileFormat and readRecordsFromBlockPayload, not by this block type.
+ return HoodieLogBlockType.PARQUET_DATA_BLOCK;
+ }
+
+ @Override
+ protected <T> ClosableIterator<HoodieRecord<T>>
readRecordsFromBlockPayload(HoodieRecord.HoodieRecordType type) throws
IOException {
+ StoragePath path = logFile.getPath();
+ return HoodieIOFactory.getIOFactory(storage)
+ .getReaderFactory(type)
+ .getFileReader(DEFAULT_HUDI_CONFIG_FOR_READER, path, fileFormat,
Option.empty())
+ .getRecordIterator(getSchemaFromHeader(), readerSchema);
+ }
+
+ @Override
+ protected <T> ClosableIterator<HoodieRecord<T>>
readRecordsFromBlockPayload(HoodieRecord.HoodieRecordType type, int
ignoredBufferSize) throws IOException {
+ return readRecordsFromBlockPayload(type);
+ }
+
+ @Override
+ protected <T> ClosableIterator<T>
readRecordsFromBlockPayload(HoodieReaderContext<T> readerContext) throws
IOException {
+ return readerContext.getFileRecordIterator(
+ logFile.getPath(), 0, FSUtils.getFileSize(storage, logFile),
+ getSchemaFromHeader(),
+ readerSchema,
+ storage);
+ }
+
+ @Override
+ protected <T> ClosableIterator<T> lookupEngineRecords(HoodieReaderContext<T>
readerContext, List<String> keys, boolean fullKey) throws IOException {
+ return FilteringEngineRecordIterator.getInstance(
+ readRecordsFromBlockPayload(readerContext),
+ new HashSet<>(keys),
+ fullKey,
+ record ->
Option.ofNullable(readerContext.getRecordContext().getRecordKey(record,
readerSchema)));
+ }
+
+ @Override
+ protected ByteArrayOutputStream serializeRecords(List<HoodieRecord> records,
HoodieStorage storage) {
+ throw new UnsupportedOperationException("Native log data blocks are
read-only");
+ }
+
+ @Override
+ protected <T> ClosableIterator<HoodieRecord<T>> deserializeRecords(byte[]
content, HoodieRecord.HoodieRecordType type) {
+ throw new UnsupportedOperationException("Native log data blocks read
records directly from native files");
+ }
+
+ @Override
+ protected <T> ClosableIterator<HoodieRecord<T>> deserializeRecords(
+ SeekableDataInputStream inputStream,
+ HoodieLogBlockContentLocation contentLocation,
+ HoodieRecord.HoodieRecordType type,
+ int bufferSize) {
+ throw new UnsupportedOperationException("Native log data blocks do not use
inline log block content");
+ }
+
+ @Override
+ protected <T> ClosableIterator<T> deserializeRecords(HoodieReaderContext<T>
readerContext, byte[] content) {
+ throw new UnsupportedOperationException("Native log data blocks read
records directly from native files");
+ }
+
+ private static Option<HoodieLogBlockContentLocation>
getContentLocation(HoodieStorage storage, HoodieLogFile logFile) {
+ long fileSize = FSUtils.getFileSize(storage, logFile);
+ return Option.of(new HoodieLogBlockContentLocation(storage, logFile, 0,
fileSize, fileSize));
+ }
+}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieNativeDeleteBlock.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieNativeDeleteBlock.java
new file mode 100644
index 000000000000..37064e6ce7bb
--- /dev/null
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieNativeDeleteBlock.java
@@ -0,0 +1,106 @@
+/*
+ * 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.log.block;
+
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.engine.RecordContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.DeleteRecord;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.table.read.BufferedRecord;
+import org.apache.hudi.common.table.read.BufferedRecords;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.exception.HoodieNotSupportedException;
+import org.apache.hudi.storage.HoodieStorage;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Delete block backed by a native delete log file.
+ */
+public class HoodieNativeDeleteBlock extends HoodieDeleteBlock {
+
+ private final HoodieStorage storage;
+ private final HoodieLogFile logFile;
+ private final HoodieReaderContext<?> readerContext;
+ private final HoodieSchema deleteLogSchema;
+ private final List<String> orderingFieldNames;
+ private List<BufferedRecord<?>> bufferedRecordsToDelete;
+
+ public HoodieNativeDeleteBlock(HoodieStorage storage,
+ HoodieLogFile logFile,
+ HoodieReaderContext<?> readerContext,
+ HoodieSchema deleteLogSchema,
+ List<String> orderingFieldNames,
+ Map<HeaderMetadataType, String> header,
+ Map<FooterMetadataType, String> footer) {
+ super(Option.empty(), null, true, getContentLocation(storage, logFile),
header, footer);
+ this.storage = storage;
+ this.logFile = logFile;
+ this.readerContext = readerContext;
+ this.deleteLogSchema = deleteLogSchema;
+ this.orderingFieldNames = orderingFieldNames;
+ }
+
+ @Override
+ public DeleteRecord[] getRecordsToDelete() {
+ throw new HoodieNotSupportedException("Native delete log files do not
support the legacy DeleteRecord[] API. "
+ + "Use getRecordsToDelete(RecordContext) instead. Log file: " +
logFile);
+ }
+
+ @Override
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ public <T> List<BufferedRecord<T>> getRecordsToDelete(RecordContext<T>
recordContext) {
+ if (bufferedRecordsToDelete == null) {
+ bufferedRecordsToDelete = (List)
readBufferedRecordsToDelete(recordContext);
+ }
+ return (List) bufferedRecordsToDelete;
+ }
+
+ @SuppressWarnings("unchecked")
+ private <T> List<BufferedRecord<T>>
readBufferedRecordsToDelete(RecordContext<T> recordContext) {
+ HoodieReaderContext<T> typedReaderContext = (HoodieReaderContext<T>)
readerContext;
+ List<BufferedRecord<T>> deleteRecords = new ArrayList<>();
+ try (ClosableIterator<T> recordIterator =
typedReaderContext.getFileRecordIterator(
+ logFile.getPath(), 0, FSUtils.getFileSize(storage, logFile),
deleteLogSchema, deleteLogSchema, storage)) {
+ while (recordIterator.hasNext()) {
+ T record = recordIterator.next();
+ Object recordKey = recordContext.getValue(record, deleteLogSchema,
HoodieRecord.RECORD_KEY_METADATA_FIELD);
+ Comparable orderingValue = recordContext.getOrderingValue(record,
deleteLogSchema, orderingFieldNames);
+ deleteRecords.add(BufferedRecords.createDelete(recordKey.toString(),
orderingValue));
+ }
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to read native delete log file " +
logFile, e);
+ }
+ return deleteRecords;
+ }
+
+ private static Option<HoodieLogBlockContentLocation>
getContentLocation(HoodieStorage storage, HoodieLogFile logFile) {
+ long fileSize = FSUtils.getFileSize(storage, logFile);
+ return Option.of(new HoodieLogBlockContentLocation(storage, logFile, 0,
fileSize, fileSize));
+ }
+}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMerger.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMerger.java
index e8c1d103575e..9c6d8f1ca5f8 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMerger.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMerger.java
@@ -50,11 +50,15 @@ public interface BufferedRecordMerger<T> extends
Serializable {
/**
* Merges incoming delete record from log file with existing record in
record buffer.
*
+ * @deprecated Delete records should be converted to {@link BufferedRecord}s
and merged through
+ * {@link #deltaMerge(BufferedRecord, BufferedRecord)}.
+ *
* @param deleteRecord Incoming delete record from log file
* @param existingRecord Existing record in record buffer
*
* @return The merged record.
*/
+ @Deprecated
Option<DeleteRecord> deltaMerge(DeleteRecord deleteRecord, BufferedRecord<T>
existingRecord);
/**
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMergerFactory.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMergerFactory.java
index 06e903dd1fa4..4484abaefee9 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMergerFactory.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMergerFactory.java
@@ -110,6 +110,7 @@ public class BufferedRecordMergerFactory {
}
@Override
+ @Deprecated
public Option<DeleteRecord> deltaMerge(DeleteRecord deleteRecord,
BufferedRecord<T> existingRecord) {
return Option.of(deleteRecord);
}
@@ -139,6 +140,9 @@ public class BufferedRecordMergerFactory {
@Override
public Option<BufferedRecord<T>> deltaMerge(BufferedRecord<T> newRecord,
BufferedRecord<T>
existingRecord) {
+ if (newRecord.isDelete()) {
+ return Option.of(newRecord);
+ }
if (existingRecord != null) {
HoodieSchema newSchema =
recordContext.getSchemaFromBufferRecord(newRecord);
newRecord = partialUpdateHandler.partialMerge(
@@ -154,6 +158,9 @@ public class BufferedRecordMergerFactory {
@Override
public BufferedRecord<T> finalMerge(BufferedRecord<T> olderRecord,
BufferedRecord<T> newerRecord) {
+ if (newerRecord.isDelete()) {
+ return newerRecord;
+ }
HoodieSchema newSchema =
recordContext.getSchemaFromBufferRecord(newerRecord);
newerRecord = partialUpdateHandler.partialMerge(
newerRecord,
@@ -185,6 +192,7 @@ public class BufferedRecordMergerFactory {
}
@Override
+ @Deprecated
public Option<DeleteRecord> deltaMerge(DeleteRecord deleteRecord,
BufferedRecord<T> existingRecord) {
return deltaMergeDeleteRecord(deleteRecord, existingRecord,
recordContext);
}
@@ -216,6 +224,9 @@ public class BufferedRecordMergerFactory {
@Override
public Option<BufferedRecord<T>> deltaMerge(BufferedRecord<T> newRecord,
BufferedRecord<T> existingRecord) {
+ if (newRecord.isDelete()) {
+ return super.deltaMerge(newRecord, existingRecord);
+ }
if (existingRecord == null) {
return Option.of(newRecord);
} else if (shouldKeepNewerRecord(existingRecord, newRecord)) {
@@ -242,6 +253,9 @@ public class BufferedRecordMergerFactory {
@Override
public BufferedRecord<T> finalMerge(BufferedRecord<T> olderRecord,
BufferedRecord<T> newerRecord) {
+ if (newerRecord.isDelete()) {
+ return super.finalMerge(olderRecord, newerRecord);
+ }
if (newerRecord.isCommitTimeOrderingDelete()) {
return newerRecord;
}
@@ -297,6 +311,9 @@ public class BufferedRecordMergerFactory {
@Override
public Option<BufferedRecord<T>> deltaMerge(BufferedRecord<T> newRecord,
BufferedRecord<T> existingRecord) throws IOException {
+ if (newRecord.isDelete()) {
+ return deleteRecordMerger.deltaMerge(newRecord, existingRecord);
+ }
if (existingRecord == null) {
return Option.of(newRecord);
}
@@ -318,12 +335,16 @@ public class BufferedRecordMergerFactory {
}
@Override
+ @Deprecated
public Option<DeleteRecord> deltaMerge(DeleteRecord deleteRecord,
BufferedRecord<T> existingRecord) {
return this.deleteRecordMerger.deltaMerge(deleteRecord, existingRecord);
}
@Override
public BufferedRecord<T> finalMerge(BufferedRecord<T> olderRecord,
BufferedRecord<T> newerRecord) throws IOException {
+ if (newerRecord.isDelete()) {
+ return deleteRecordMerger.finalMerge(olderRecord, newerRecord);
+ }
// TODO(HUDI-7843): decouple the merging logic from the merger
// and use the record merge mode to control how to merge partial updates
return recordMerger.get().partialMerge(olderRecord, newerRecord,
readerSchema, recordContext, props);
@@ -442,6 +463,7 @@ public class BufferedRecordMergerFactory {
public abstract Option<BufferedRecord<T>>
deltaMergeRecords(BufferedRecord<T> newRecord, BufferedRecord<T>
existingRecord) throws IOException;
@Override
+ @Deprecated
public Option<DeleteRecord> deltaMerge(DeleteRecord deleteRecord,
BufferedRecord<T> existingRecord) {
BufferedRecord<T> deleteBufferedRecord =
BufferedRecords.fromDeleteRecord(deleteRecord, recordContext);
try {
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/HoodieFileGroupRecordBuffer.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/HoodieFileGroupRecordBuffer.java
index f3ebd1813959..24594123e324 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/HoodieFileGroupRecordBuffer.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/HoodieFileGroupRecordBuffer.java
@@ -19,7 +19,6 @@
package org.apache.hudi.common.table.read.buffer;
-import org.apache.hudi.common.model.DeleteRecord;
import org.apache.hudi.common.table.log.KeySpec;
import org.apache.hudi.common.table.log.block.HoodieDataBlock;
import org.apache.hudi.common.table.log.block.HoodieDeleteBlock;
@@ -72,11 +71,6 @@ public interface HoodieFileGroupRecordBuffer<T> {
*/
void processDeleteBlock(HoodieDeleteBlock deleteBlock) throws IOException;
- /**
- * Process next delete record.
- */
- void processNextDeletedRecord(DeleteRecord record, Serializable index);
-
/**
* Check if a record exists in the buffered records.
*/
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/KeyBasedFileGroupRecordBuffer.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/KeyBasedFileGroupRecordBuffer.java
index f052f5cee844..afbdfd0eed5a 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/KeyBasedFileGroupRecordBuffer.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/KeyBasedFileGroupRecordBuffer.java
@@ -23,7 +23,6 @@ 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.engine.RecordContext;
-import org.apache.hudi.common.model.DeleteRecord;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaCache;
import org.apache.hudi.common.table.HoodieTableMetaClient;
@@ -42,8 +41,6 @@ import org.apache.hudi.common.util.collection.Pair;
import java.io.IOException;
import java.io.Serializable;
-import java.util.Arrays;
-import java.util.Iterator;
import java.util.List;
/**
@@ -111,22 +108,11 @@ public class KeyBasedFileGroupRecordBuffer<T> extends
FileGroupRecordBuffer<T> {
@Override
public void processDeleteBlock(HoodieDeleteBlock deleteBlock) throws
IOException {
- Iterator<DeleteRecord> it =
Arrays.stream(deleteBlock.getRecordsToDelete()).iterator();
- while (it.hasNext()) {
- DeleteRecord record = it.next();
- processNextDeletedRecord(record, record.getRecordKey());
+ for (BufferedRecord<T> record :
deleteBlock.getRecordsToDelete(readerContext.getRecordContext())) {
+ processNextDataRecord(record, record.getRecordKey());
}
}
- @Override
- public void processNextDeletedRecord(DeleteRecord deleteRecord, Serializable
recordIdentifier) {
- BufferedRecord<T> existingRecord = records.get(recordIdentifier);
- totalLogRecords++;
- Option<DeleteRecord> recordOpt =
bufferedRecordMerger.deltaMerge(deleteRecord, existingRecord);
- recordOpt.ifPresent(deleteRec ->
- records.put(recordIdentifier,
BufferedRecords.fromDeleteRecord(deleteRec, readerContext.getRecordContext())));
- }
-
@Override
public boolean containsLogRecord(String recordKey) {
return records.containsKey(recordKey);
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/PositionBasedFileGroupRecordBuffer.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/PositionBasedFileGroupRecordBuffer.java
index 5eaa54e033c8..62a851a4b9d3 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/PositionBasedFileGroupRecordBuffer.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/PositionBasedFileGroupRecordBuffer.java
@@ -22,7 +22,6 @@ package org.apache.hudi.common.table.read.buffer;
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.DeleteRecord;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaCache;
import org.apache.hudi.common.table.HoodieTableMetaClient;
@@ -47,7 +46,6 @@ import org.roaringbitmap.longlong.Roaring64NavigableMap;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
@@ -189,7 +187,7 @@ public class PositionBasedFileGroupRecordBuffer<T> extends
KeyBasedFileGroupReco
switch (recordMergeMode) {
case COMMIT_TIME_ORDERING:
int commitTimeBasedRecordIndex = 0;
- DeleteRecord[] deleteRecords = deleteBlock.getRecordsToDelete();
+ List<BufferedRecord<T>> deleteRecords =
deleteBlock.getRecordsToDelete(readerContext.getRecordContext());
for (Long recordPosition : recordPositions) {
// IMPORTANT:
// use #put for log files with regular order(see
HoodieLogFile.LOG_FILE_COMPARATOR);
@@ -200,20 +198,16 @@ public class PositionBasedFileGroupRecordBuffer<T>
extends KeyBasedFileGroupReco
// because under hybrid strategy in #doHasNextFallbackBaseRecord, if
the record keys are not set up,
// this delete-vector could be kept in the records cache(see the
check in #fallbackToKeyBasedBuffer),
// and these keys would be deleted no matter whether there are
following-up inserts/updates.
- DeleteRecord deleteRecord =
deleteRecords[commitTimeBasedRecordIndex++];
- BufferedRecord<T> record =
BufferedRecords.fromDeleteRecord(deleteRecord,
readerContext.getRecordContext());
- records.put(recordPosition, record);
+ records.put(recordPosition,
deleteRecords.get(commitTimeBasedRecordIndex++));
}
return;
case EVENT_TIME_ORDERING:
case CUSTOM:
default:
int recordIndex = 0;
- Iterator<DeleteRecord> it =
Arrays.stream(deleteBlock.getRecordsToDelete()).iterator();
- while (it.hasNext()) {
- DeleteRecord record = it.next();
+ for (BufferedRecord<T> record :
deleteBlock.getRecordsToDelete(readerContext.getRecordContext())) {
long recordPosition = recordPositions.get(recordIndex++);
- processNextDeletedRecord(record, recordPosition);
+ processNextDataRecord(record, recordPosition);
}
}
}
@@ -314,4 +308,4 @@ public class PositionBasedFileGroupRecordBuffer<T> extends
KeyBasedFileGroupReco
return blockPositions;
}
-}
\ No newline at end of file
+}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/ReusableKeyBasedRecordBuffer.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/ReusableKeyBasedRecordBuffer.java
index bce8c7f9e74f..80867641c1a4 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/ReusableKeyBasedRecordBuffer.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/ReusableKeyBasedRecordBuffer.java
@@ -21,7 +21,6 @@ package org.apache.hudi.common.table.read.buffer;
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.DeleteRecord;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.PartialUpdateMode;
import org.apache.hudi.common.table.log.KeySpec;
@@ -123,11 +122,6 @@ public class ReusableKeyBasedRecordBuffer<T> extends
FileGroupRecordBuffer<T> {
}
- @Override
- public void processNextDeletedRecord(DeleteRecord record, Serializable
index) {
- throw new HoodieNotSupportedException("Reusable record buffer does not
process the delete records from the logs");
- }
-
@Override
public boolean containsLogRecord(String recordKey) {
return validKeys.contains(recordKey) &&
existingRecords.containsKey(recordKey);
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/UnmergedFileGroupRecordBuffer.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/UnmergedFileGroupRecordBuffer.java
index b1c2a12a24ee..66a437f11ead 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/UnmergedFileGroupRecordBuffer.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/UnmergedFileGroupRecordBuffer.java
@@ -22,7 +22,6 @@ package org.apache.hudi.common.table.read.buffer;
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.DeleteRecord;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.PartialUpdateMode;
@@ -114,11 +113,6 @@ class UnmergedFileGroupRecordBuffer<T> extends
FileGroupRecordBuffer<T> {
// no-op
}
- @Override
- public void processNextDeletedRecord(DeleteRecord deleteRecord, Serializable
index) {
- // no-op
- }
-
@Override
public boolean containsLogRecord(String recordKey) {
throw new UnsupportedOperationException("Not supported for " +
this.getClass().getSimpleName());
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
index 0c68f9ee4215..75118b90bb2c 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
@@ -527,11 +527,7 @@ public abstract class AbstractTableFileSystemView
implements SyncableFileSystemV
* @param pathInfoList List of StoragePathInfo
*/
private Stream<HoodieLogFile>
convertFileStatusesToLogFiles(List<StoragePathInfo> pathInfoList) {
- String logFileExtension =
metaClient.getTableConfig().getLogFileFormat().getFileExtension();
- Predicate<StoragePathInfo> rtFilePredicate = pathInfo -> {
- String fileName = pathInfo.getPath().getName();
- return FSUtils.isLogFile(pathInfo.getPath()) &&
fileName.contains(logFileExtension);
- };
+ Predicate<StoragePathInfo> rtFilePredicate = pathInfo ->
FSUtils.isLogFile(pathInfo.getPath());
return
pathInfoList.stream().filter(rtFilePredicate).map(HoodieLogFile::new);
}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/log/TestHoodieNativeLogFileReader.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/log/TestHoodieNativeLogFileReader.java
new file mode 100644
index 000000000000..673739d2b9c5
--- /dev/null
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/log/TestHoodieNativeLogFileReader.java
@@ -0,0 +1,51 @@
+/*
+ * 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.log;
+
+import org.apache.hudi.common.model.HoodieLogFile;
+import
org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.storage.StoragePath;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestHoodieNativeLogFileReader {
+
+ @Test
+ void testValidateDataBlockHeaderRequiresSchema() {
+ HoodieLogFile logFile = new HoodieLogFile(new StoragePath(
+ "/tmp/file-id_1-0-1_001_1.log.parquet"));
+ Map<HeaderMetadataType, String> header = new HashMap<>();
+ header.put(HeaderMetadataType.INSTANT_TIME, "001");
+
+ HoodieIOException exception = assertThrows(HoodieIOException.class,
+ () -> HoodieNativeLogFileReader.validateDataBlockHeader(logFile,
header));
+
+
assertTrue(exception.getMessage().contains(HeaderMetadataType.SCHEMA.name()));
+
assertTrue(exception.getMessage().contains(NativeLogFooterMetadata.FOOTER_METADATA_KEY));
+ assertTrue(exception.getMessage().contains(logFile.toString()));
+ }
+}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/log/TestNativeLogFooterMetadata.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/log/TestNativeLogFooterMetadata.java
new file mode 100644
index 000000000000..a7527a0599da
--- /dev/null
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/log/TestNativeLogFooterMetadata.java
@@ -0,0 +1,94 @@
+/*
+ * 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.log;
+
+import
org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.exception.HoodieNotSupportedException;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestNativeLogFooterMetadata {
+
+ @Test
+ void testWriteThenReadRoundTrip() {
+ Map<HeaderMetadataType, String> header = new HashMap<>();
+ header.put(HeaderMetadataType.INSTANT_TIME, "001");
+ header.put(HeaderMetadataType.SCHEMA,
"{\"type\":\"record\",\"name\":\"test\",\"fields\":[]}");
+ header.put(HeaderMetadataType.IS_PARTIAL, "true");
+
+ // Footer produced by the write path must be readable by the read path.
+ Map<String, String> footer =
NativeLogFooterMetadata.toFooterMetadata(header);
+
assertTrue(footer.containsKey(NativeLogFooterMetadata.FOOTER_METADATA_KEY));
+
+ Map<HeaderMetadataType, String> parsed =
NativeLogFooterMetadata.fromFooterMetadata(footer);
+ assertEquals("001", parsed.get(HeaderMetadataType.INSTANT_TIME));
+ assertEquals("{\"type\":\"record\",\"name\":\"test\",\"fields\":[]}",
parsed.get(HeaderMetadataType.SCHEMA));
+ assertEquals("true", parsed.get(HeaderMetadataType.IS_PARTIAL));
+ // VERSION is injected on write and surfaced on read.
+ assertEquals(String.valueOf(HoodieLogFormat.CURRENT_VERSION),
+ parsed.get(HeaderMetadataType.VERSION));
+ }
+
+ @Test
+ void testNullHeaderValuesAreDropped() {
+ Map<HeaderMetadataType, String> header = new HashMap<>();
+ header.put(HeaderMetadataType.INSTANT_TIME, "001");
+ header.put(HeaderMetadataType.SCHEMA, null);
+
+ Map<HeaderMetadataType, String> parsed =
+
NativeLogFooterMetadata.fromFooterMetadata(NativeLogFooterMetadata.toFooterMetadata(header));
+ assertEquals("001", parsed.get(HeaderMetadataType.INSTANT_TIME));
+ assertFalse(parsed.containsKey(HeaderMetadataType.SCHEMA));
+ }
+
+ @Test
+ void testMissingFooterKeyReturnsEmptyHeader() {
+ assertTrue(NativeLogFooterMetadata.fromFooterMetadata(new
HashMap<>()).isEmpty());
+ }
+
+ @Test
+ void testUnknownHeaderTypesAreIgnored() {
+ Map<String, String> footer = new HashMap<>();
+ footer.put(NativeLogFooterMetadata.FOOTER_METADATA_KEY,
+
"{\"VERSION\":\"2\",\"INSTANT_TIME\":\"001\",\"SOME_FUTURE_KEY\":\"v\"}");
+
+ Map<HeaderMetadataType, String> parsed =
NativeLogFooterMetadata.fromFooterMetadata(footer);
+ assertEquals("001", parsed.get(HeaderMetadataType.INSTANT_TIME));
+ assertEquals(2, parsed.size());
+ }
+
+ @Test
+ void testNewerFormatVersionIsRejected() {
+ Map<String, String> footer = new HashMap<>();
+ footer.put(NativeLogFooterMetadata.FOOTER_METADATA_KEY,
+ "{\"VERSION\":\"" + (HoodieLogFormat.CURRENT_VERSION + 1) +
"\",\"INSTANT_TIME\":\"001\"}");
+
+ assertThrows(HoodieNotSupportedException.class,
+ () -> NativeLogFooterMetadata.fromFooterMetadata(footer));
+ }
+}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/BaseTestFileGroupRecordBuffer.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/BaseTestFileGroupRecordBuffer.java
index b2869df3310d..0f29ab2c30d4 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/BaseTestFileGroupRecordBuffer.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/BaseTestFileGroupRecordBuffer.java
@@ -24,6 +24,7 @@ import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.engine.HoodieReaderContext;
import org.apache.hudi.common.engine.RecordContext;
import org.apache.hudi.common.model.BaseAvroPayload;
+import org.apache.hudi.common.model.DeleteRecord;
import org.apache.hudi.common.model.HoodieAvroIndexedRecord;
import org.apache.hudi.common.model.HoodieEmptyRecord;
import org.apache.hudi.common.model.HoodieKey;
@@ -37,6 +38,7 @@ import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.schema.HoodieSchemaType;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.log.block.HoodieDeleteBlock;
import org.apache.hudi.common.table.read.BufferedRecord;
import org.apache.hudi.common.table.read.BufferedRecords;
import org.apache.hudi.common.table.read.DeleteContext;
@@ -67,6 +69,7 @@ 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.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -87,6 +90,16 @@ public class BaseTestFileGroupRecordBuffer {
return record;
}
+ protected static void mockDeleteRecords(HoodieDeleteBlock deleteBlock,
DeleteRecord... deleteRecords) {
+ when(deleteBlock.getRecordsToDelete()).thenReturn(deleteRecords);
+ when(deleteBlock.getRecordsToDelete(any())).thenAnswer(invocation -> {
+ RecordContext recordContext = invocation.getArgument(0);
+ return Arrays.stream(deleteRecords)
+ .map(deleteRecord -> BufferedRecords.fromDeleteRecord(deleteRecord,
recordContext))
+ .collect(Collectors.toList());
+ });
+ }
+
protected static List<HoodieRecord>
convertToHoodieRecordsList(List<IndexedRecord> indexedRecords) {
return indexedRecords.stream().map(rec -> new HoodieAvroIndexedRecord(new
HoodieKey(rec.get(0).toString(), ""), rec)).collect(Collectors.toList());
}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestKeyBasedFileGroupRecordBuffer.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestKeyBasedFileGroupRecordBuffer.java
index 36149f65e97d..87e8b827f37c 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestKeyBasedFileGroupRecordBuffer.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestKeyBasedFileGroupRecordBuffer.java
@@ -123,8 +123,7 @@ class TestKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecordBuffer {
when(dataBlock2.getEngineRecordIterator(readerContext)).thenReturn(ClosableIterator.wrap(Arrays.asList(testRecord2EarlierUpdate,
testRecord3Update).iterator()));
HoodieDeleteBlock deleteBlock = mock(HoodieDeleteBlock.class);
- when(deleteBlock.getRecordsToDelete()).thenReturn(new DeleteRecord[]
{DeleteRecord.create("3", ""), DeleteRecord.create("2", "", -1L),
- DeleteRecord.create("1", "", 2L)});
+ mockDeleteRecords(deleteBlock, DeleteRecord.create("3", ""),
DeleteRecord.create("2", "", -1L), DeleteRecord.create("1", "", 2L));
// process data block, then delete block, then another data block
fileGroupRecordBuffer.processDataBlock(dataBlock, Option.empty());
fileGroupRecordBuffer.processDeleteBlock(deleteBlock);
@@ -268,7 +267,7 @@ class TestKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecordBuffer {
.thenReturn(ClosableIterator.wrap(Arrays.asList(testRecord1UpdateWithSameTime,
testRecord2Delete, testRecord4Update).iterator()));
HoodieDeleteBlock deleteBlock = mock(HoodieDeleteBlock.class);
- when(deleteBlock.getRecordsToDelete()).thenReturn(new DeleteRecord[]
{DeleteRecord.create("3", "")});
+ mockDeleteRecords(deleteBlock, DeleteRecord.create("3", ""));
fileGroupRecordBuffer.processDataBlock(dataBlock1, Option.empty());
fileGroupRecordBuffer.processDataBlock(dataBlock2, Option.empty());
fileGroupRecordBuffer.processDeleteBlock(deleteBlock);
@@ -344,7 +343,7 @@ class TestKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecordBuffer {
.thenReturn(ClosableIterator.wrap(Arrays.asList(testRecord2Delete,
testRecord4Update).iterator()));
HoodieDeleteBlock deleteBlock = mock(HoodieDeleteBlock.class);
- when(deleteBlock.getRecordsToDelete()).thenReturn(new DeleteRecord[]
{DeleteRecord.create("3", "")});
+ mockDeleteRecords(deleteBlock, DeleteRecord.create("3", ""));
fileGroupRecordBuffer.processDataBlock(dataBlock1, Option.empty());
fileGroupRecordBuffer.processDataBlock(dataBlock2, Option.empty());
fileGroupRecordBuffer.processDeleteBlock(deleteBlock);
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestSortedKeyBasedFileGroupRecordBuffer.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestSortedKeyBasedFileGroupRecordBuffer.java
index 611128a96309..39ad05de7c5b 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestSortedKeyBasedFileGroupRecordBuffer.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestSortedKeyBasedFileGroupRecordBuffer.java
@@ -98,7 +98,7 @@ class TestSortedKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecordBuf
ClosableIterator.wrap(Arrays.asList(testRecord6, testRecord4,
testRecord1, testRecord6Update, testRecord2Update).iterator()));
HoodieDeleteBlock deleteBlock = mock(HoodieDeleteBlock.class);
- when(deleteBlock.getRecordsToDelete()).thenReturn(new DeleteRecord[]
{DeleteRecord.create("3", "")});
+ mockDeleteRecords(deleteBlock, DeleteRecord.create("3", ""));
fileGroupRecordBuffer.processDataBlock(dataBlock, Option.empty());
fileGroupRecordBuffer.processDeleteBlock(deleteBlock);
@@ -175,7 +175,7 @@ class TestSortedKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecordBuf
when(dataBlock2.getEngineRecordIterator(mockReaderContext)).thenReturn(ClosableIterator.wrap(Arrays.asList(testRecord2Update,
testRecord5, testRecord3, testRecord1).iterator()));
HoodieDeleteBlock deleteBlock = mock(HoodieDeleteBlock.class);
- when(deleteBlock.getRecordsToDelete()).thenReturn(new DeleteRecord[]
{DeleteRecord.create("3", "")});
+ mockDeleteRecords(deleteBlock, DeleteRecord.create("3", ""));
fileGroupRecordBuffer.processDataBlock(dataBlock1, Option.empty());
fileGroupRecordBuffer.processDataBlock(dataBlock2, Option.empty());
fileGroupRecordBuffer.processDeleteBlock(deleteBlock);
@@ -217,4 +217,4 @@ class TestSortedKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecordBuf
}
return actualRecords;
}
-}
\ No newline at end of file
+}
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestTableSchemaResolver.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestTableSchemaResolver.java
index f2c6d2b49b20..e296f0d86373 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestTableSchemaResolver.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestTableSchemaResolver.java
@@ -20,6 +20,7 @@ package org.apache.hudi.common.table;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.HoodieWriteStat;
@@ -28,6 +29,7 @@ import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.schema.HoodieSchemaType;
import org.apache.hudi.common.table.log.HoodieLogFormat;
import org.apache.hudi.common.table.log.HoodieLogFormatWriter;
+import org.apache.hudi.common.table.log.NativeLogFooterMetadata;
import org.apache.hudi.common.table.log.block.HoodieDataBlock;
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
import org.apache.hudi.common.table.timeline.HoodieInstant;
@@ -36,8 +38,10 @@ import
org.apache.hudi.common.table.timeline.versioning.v2.InstantComparatorV2;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.testutils.HoodieTestUtils;
import org.apache.hudi.common.testutils.SchemaTestUtil;
+import org.apache.hudi.common.testutils.reader.HoodieFileSliceTestUtils;
import org.apache.hudi.common.util.FileFormatUtils;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.internal.schema.HoodieSchemaException;
import org.apache.hudi.io.storage.HoodieIOFactory;
import org.apache.hudi.storage.HoodieStorage;
@@ -54,6 +58,8 @@ import org.mockito.MockedStatic;
import java.io.IOException;
import java.net.URISyntaxException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
@@ -65,6 +71,8 @@ import static
org.apache.hudi.common.testutils.HoodieCommonTestHarness.getDataBl
import static org.apache.hudi.common.testutils.SchemaTestUtil.getSimpleSchema;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
@@ -124,6 +132,67 @@ class TestTableSchemaResolver {
logFilePath));
}
+ @Test
+ void testReadSchemaFromNativeDataLogFile() throws IOException {
+ String testDir = initTestDir("read_schema_from_native_data_log_file");
+ StoragePath partitionPath = new StoragePath(testDir, "partition1");
+ Files.createDirectories(Paths.get(partitionPath.toString()));
+ HoodieSchema expectedSchema = HoodieTestDataGenerator.HOODIE_SCHEMA;
+ StoragePath logFilePath = new StoragePath(partitionPath,
FSUtils.makeNativeLogFileName(
+ "test-fileid1", "1-0-1", "100", 1, "log", HoodieFileFormat.PARQUET));
+ HoodieFileSliceTestUtils.createNativeDataLogFile(
+ logFilePath.toString(),
+ Arrays.asList(HoodieFileSliceTestUtils.DATA_GEN.generateGenericRecord(
+ "1", "partition1", "rider-1", "driver-1", 1L)),
+ expectedSchema,
+ "100");
+
+ assertEquals(expectedSchema, TableSchemaResolver.readSchemaFromLogFile(
+ HoodieStorageUtils.getStorage(logFilePath,
HoodieTestUtils.getDefaultStorageConf()),
+ logFilePath));
+ }
+
+ @Test
+ void testReadSchemaFromNativeDeleteLogFileReturnsNull() throws IOException {
+ String testDir = initTestDir("read_schema_from_native_delete_log_file");
+ StoragePath partitionPath = new StoragePath(testDir, "partition1");
+ Files.createDirectories(Paths.get(partitionPath.toString()));
+ HoodieSchema tableSchema = HoodieTestDataGenerator.HOODIE_SCHEMA;
+ StoragePath logFilePath = new StoragePath(partitionPath,
FSUtils.makeNativeLogFileName(
+ "test-fileid1", "1-0-1", "100", 1, "deletes",
HoodieFileFormat.PARQUET));
+ HoodieStorage storage = HoodieStorageUtils.getStorage(logFilePath,
HoodieTestUtils.getDefaultStorageConf());
+ HoodieFileSliceTestUtils.createNativeDeleteLogFile(
+ storage,
+ logFilePath.toString(),
+ Arrays.asList(HoodieFileSliceTestUtils.DATA_GEN.generateGenericRecord(
+ "1", "partition1", "rider-1", "driver-1", 1L)),
+ tableSchema,
+ "100");
+
+ assertNull(TableSchemaResolver.readSchemaFromLogFile(storage,
logFilePath));
+ }
+
+ @Test
+ void testReadSchemaFromNativeDataLogFileRequiresSchemaFooter() {
+ StoragePath logFilePath = new
StoragePath("/tmp/test-fileid1_1-0-1_100_1.log.parquet");
+ HoodieStorage storage = mock(HoodieStorage.class);
+ HoodieIOFactory ioFactory = mock(HoodieIOFactory.class);
+ FileFormatUtils fileFormatUtils = mock(FileFormatUtils.class);
+
when(ioFactory.getFileFormatUtils(HoodieFileFormat.PARQUET)).thenReturn(fileFormatUtils);
+ when(fileFormatUtils.readFooter(storage, false, logFilePath,
NativeLogFooterMetadata.FOOTER_METADATA_KEY))
+ .thenReturn(new HashMap<>());
+
+ try (MockedStatic<HoodieIOFactory> ioFactoryMockedStatic =
mockStatic(HoodieIOFactory.class)) {
+ ioFactoryMockedStatic.when(() ->
HoodieIOFactory.getIOFactory(storage)).thenReturn(ioFactory);
+
+ HoodieIOException exception = assertThrows(HoodieIOException.class,
+ () -> TableSchemaResolver.readSchemaFromLogFile(storage,
logFilePath));
+
assertTrue(exception.getMessage().contains(HoodieLogBlock.HeaderMetadataType.SCHEMA.name()));
+
assertTrue(exception.getMessage().contains(NativeLogFooterMetadata.FOOTER_METADATA_KEY));
+ assertTrue(exception.getMessage().contains(logFilePath.toString()));
+ }
+ }
+
@Test
void testHasOperationFieldFileInspectionOrdering() throws IOException {
StorageConfiguration conf = new HadoopStorageConfiguration(false);
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderNativeLogs.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderNativeLogs.java
new file mode 100644
index 000000000000..49ce1c344aac
--- /dev/null
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderNativeLogs.java
@@ -0,0 +1,338 @@
+/*
+ * 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.avro.HoodieAvroReaderContext;
+import org.apache.hudi.common.config.RecordMergeMode;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import
org.apache.hudi.common.table.log.block.HoodieLogBlock.HoodieLogBlockType;
+import org.apache.hudi.common.testutils.HoodieTestTable;
+import
org.apache.hudi.common.testutils.reader.HoodieFileGroupReaderTestHarness;
+import org.apache.hudi.common.testutils.reader.HoodieFileSliceTestUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.expression.Literal;
+import org.apache.hudi.expression.Predicate;
+import org.apache.hudi.expression.Predicates;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.HoodieStorageUtils;
+
+import org.apache.avro.generic.IndexedRecord;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.apache.hudi.common.table.HoodieTableConfig.ORDERING_FIELDS;
+import static
org.apache.hudi.common.table.log.block.HoodieLogBlock.HoodieLogBlockType.PARQUET_DATA_BLOCK;
+import static
org.apache.hudi.common.testutils.FileCreateUtilsLegacy.logFileName;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.AVRO_SCHEMA;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.HOODIE_SCHEMA;
+import static
org.apache.hudi.common.testutils.HoodieTestUtils.DEFAULT_WRITE_TOKEN;
+import static
org.apache.hudi.common.testutils.reader.HoodieFileSliceTestUtils.ROW_KEY;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class TestHoodieFileGroupReaderNativeLogs extends
HoodieFileGroupReaderTestHarness {
+
+ @Override
+ protected Properties getMetaProps() {
+ Properties metaProps = super.getMetaProps();
+ metaProps.setProperty(HoodieTableConfig.RECORD_MERGE_MODE.key(),
RecordMergeMode.EVENT_TIME_ORDERING.name());
+ metaProps.setProperty(ORDERING_FIELDS.key(), "timestamp");
+ return metaProps;
+ }
+
+ @BeforeAll
+ public static void setUp() throws IOException {
+ properties.setProperty("hoodie.write.record.merge.mode",
RecordMergeMode.EVENT_TIME_ORDERING.name());
+ instantTimes = Arrays.asList("001", "002", "003", "004");
+ }
+
+ @BeforeEach
+ public void initialize() throws Exception {
+ setTableName(TestHoodieFileGroupReaderNativeLogs.class.getName());
+ initPath(tableName);
+ initMetaClient();
+ initTestDataGenerator(new String[] {PARTITION_PATH});
+ testTable = HoodieTestTable.of(metaClient);
+ readerContext = new HoodieAvroReaderContext(
+ storageConf, metaClient.getTableConfig(), Option.empty(),
Option.empty());
+ setUpMockCommits();
+ }
+
+ @Test
+ public void testPureNativeLogFiles() throws IOException {
+ HoodieStorage hoodieStorage = HoodieStorageUtils.getStorage(basePath,
storageConf);
+ preparePartitionPath();
+
+ HoodieLogFile nativeDataLog1 = createNativeDataLogFile("001", 1,
records(Arrays.asList("1", "2", "3"), 2L, false));
+ HoodieLogFile nativeDataLog2 = createNativeDataLogFile("002", 2,
records(Arrays.asList("1", "4"), 4L, false));
+ HoodieLogFile nativeDeleteLog = createNativeDeleteLogFile(hoodieStorage,
"003", 3, records(Arrays.asList("2"), 3L, true));
+ FileSlice fileSlice = createFileSlice(null, nativeDataLog1,
nativeDataLog2, nativeDeleteLog);
+
+ assertFileGroupRecords(fileSlice,
+ Arrays.asList("1", "3", "4"),
+ Arrays.asList(4L, 2L, 4L));
+ }
+
+ @Test
+ public void testNativeLogFilesWithKeyFilters() throws IOException {
+ preparePartitionPath();
+
+ HoodieLogFile nativeDataLog1 = createNativeDataLogFile("001", 1,
records(Arrays.asList("1", "2", "30"), 2L, false));
+ HoodieLogFile nativeDataLog2 = createNativeDataLogFile("002", 2,
records(Arrays.asList("1", "4", "31"), 4L, false));
+ FileSlice fileSlice = createFileSlice(null, nativeDataLog1,
nativeDataLog2);
+
+ Predicate inPredicate = Predicates.in(
+ Literal.from(ROW_KEY),
+ Arrays.asList(Literal.from("1"), Literal.from("31")));
+ assertFileGroupRecords(fileSlice, inPredicate,
+ Arrays.asList("1", "31"),
+ Arrays.asList(4L, 4L));
+
+ Predicate prefixPredicate = Predicates.startsWithAny(
+ Literal.from(ROW_KEY),
+ Arrays.asList(Literal.from("3")));
+ assertFileGroupRecords(fileSlice, prefixPredicate,
+ Arrays.asList("30", "31"),
+ Arrays.asList(2L, 4L));
+ }
+
+ @Test
+ public void testBaseFileWithNativeLogFiles() throws IOException {
+ preparePartitionPath();
+
+ HoodieBaseFile baseFile = createBaseFile("001", records(Arrays.asList("1",
"2", "3"), 2L, false));
+ HoodieLogFile nativeDataLog1 = createNativeDataLogFile("002", 1,
records(Arrays.asList("2"), 4L, false));
+ HoodieLogFile nativeDataLog2 = createNativeDataLogFile("003", 2,
records(Arrays.asList("4"), 3L, false));
+ FileSlice fileSlice = createFileSlice(baseFile, nativeDataLog1,
nativeDataLog2);
+
+ assertFileGroupRecords(fileSlice,
+ Arrays.asList("1", "2", "3", "4"),
+ Arrays.asList(2L, 4L, 2L, 3L));
+ }
+
+ @Test
+ public void testBaseFileWithNativeDeleteFiles() throws IOException {
+ HoodieStorage hoodieStorage = HoodieStorageUtils.getStorage(basePath,
storageConf);
+ preparePartitionPath();
+
+ HoodieBaseFile baseFile = createBaseFile("001", records(Arrays.asList("1",
"2", "3", "4", "5"), 2L, false));
+ HoodieLogFile nativeDeleteLog1 = createNativeDeleteLogFile(hoodieStorage,
"002", 1, records(Arrays.asList("1", "2"), Arrays.asList(3L, 1L), true));
+ HoodieLogFile nativeDeleteLog2 = createNativeDeleteLogFile(hoodieStorage,
"003", 2, records(Arrays.asList("4"), 3L, true));
+ FileSlice fileSlice = createFileSlice(baseFile, nativeDeleteLog1,
nativeDeleteLog2);
+
+ assertFileGroupRecords(fileSlice,
+ Arrays.asList("2", "3", "5"),
+ Arrays.asList(2L, 2L, 2L));
+ }
+
+ @Test
+ public void testMixedNativeAndLegacyLogFiles() throws IOException,
InterruptedException {
+ HoodieStorage hoodieStorage = HoodieStorageUtils.getStorage(basePath,
storageConf);
+ preparePartitionPath();
+
+ HoodieBaseFile baseFile = createBaseFile("001", records(Arrays.asList("1",
"2", "3", "4"), 2L, false));
+ Map<String, Long> keyToPositionMap = keyToPositionMap("1", "2", "3", "4",
"5");
+ HoodieLogFile legacyDataLog = createLegacyLogFile(
+ hoodieStorage, "002", 1, records(Arrays.asList("1", "5"), 3L, false),
+ PARQUET_DATA_BLOCK, keyToPositionMap);
+ HoodieLogFile nativeDataLog = createNativeDataLogFile("003", 2,
records(Arrays.asList("2", "6"), 4L, false));
+ FileSlice fileSlice = createFileSlice(baseFile, legacyDataLog,
nativeDataLog);
+
+ assertFileGroupRecords(fileSlice,
+ Arrays.asList("1", "2", "3", "4", "5", "6"),
+ Arrays.asList(3L, 4L, 2L, 2L, 3L, 4L));
+ }
+
+ @Test
+ public void testMixedLegacyLogFileAndNativeDeleteFile() throws IOException,
InterruptedException {
+ HoodieStorage hoodieStorage = HoodieStorageUtils.getStorage(basePath,
storageConf);
+ preparePartitionPath();
+
+ HoodieBaseFile baseFile = createBaseFile("001", records(Arrays.asList("1",
"2", "3", "4"), 2L, false));
+ Map<String, Long> keyToPositionMap = keyToPositionMap("1", "2", "3", "4",
"5");
+ HoodieLogFile legacyDataLog = createLegacyLogFile(
+ hoodieStorage, "002", 1, records(Arrays.asList("1", "5"), 4L, false),
+ PARQUET_DATA_BLOCK, keyToPositionMap);
+ HoodieLogFile nativeDeleteLog = createNativeDeleteLogFile(
+ hoodieStorage, "003", 2, records(Arrays.asList("1", "3", "5"),
Arrays.asList(3L, 3L, 5L), true));
+ FileSlice fileSlice = createFileSlice(baseFile, legacyDataLog,
nativeDeleteLog);
+
+ assertFileGroupRecords(fileSlice,
+ Arrays.asList("1", "2", "4"),
+ Arrays.asList(4L, 2L, 2L));
+ }
+
+ @Test
+ public void testMixedLegacyAndNativeLogsPreserveEventTimeDeleteOrdering()
throws IOException, InterruptedException {
+ HoodieStorage hoodieStorage = HoodieStorageUtils.getStorage(basePath,
storageConf);
+ FileSlice fileSlice =
HoodieFileSliceTestUtils.getMixedLegacyAndNativeFileSlice(
+ hoodieStorage, basePath, PARTITION_PATH, FILE_ID)
+ .orElseThrow(() -> new IllegalArgumentException("FileSlice is not
present"));
+
+ assertFileGroupRecords(fileSlice,
+ Arrays.asList("1", "2", "6", "7", "8", "9", "10"),
+ Arrays.asList(4L, 4L, 2L, 2L, 2L, 2L, 2L));
+ }
+
+ private void assertFileGroupRecords(FileSlice fileSlice, List<String>
expectedKeys, List<Long> expectedTimestamps)
+ throws IOException {
+ try (ClosableIterator<IndexedRecord> iterator =
getFileGroupIterator(fileSlice, false, false)) {
+ assertRecords(iterator, expectedKeys, expectedTimestamps);
+ }
+ }
+
+ private void assertFileGroupRecords(FileSlice fileSlice, Predicate
predicate, List<String> expectedKeys, List<Long> expectedTimestamps)
+ throws IOException {
+ try (ClosableIterator<IndexedRecord> iterator =
getFileGroupIterator(fileSlice, predicate)) {
+ assertRecords(iterator, expectedKeys, expectedTimestamps);
+ }
+ }
+
+ private static void assertRecords(ClosableIterator<IndexedRecord> iterator,
List<String> expectedKeys, List<Long> expectedTimestamps) {
+ List<String> actualKeys = new ArrayList<>();
+ List<Long> actualTimestamps = new ArrayList<>();
+ while (iterator.hasNext()) {
+ IndexedRecord record = iterator.next();
+
actualKeys.add(record.get(AVRO_SCHEMA.getField(ROW_KEY).pos()).toString());
+ actualTimestamps.add((Long)
record.get(AVRO_SCHEMA.getField("timestamp").pos()));
+ }
+ assertEquals(expectedKeys, actualKeys);
+ assertEquals(expectedTimestamps, actualTimestamps);
+ }
+
+ private ClosableIterator<IndexedRecord> getFileGroupIterator(FileSlice
fileSlice, Predicate predicate)
+ throws IOException {
+ HoodieReaderContext<IndexedRecord> originalReaderContext = readerContext;
+ readerContext = new HoodieAvroReaderContext(
+ storageConf, metaClient.getTableConfig(), Option.empty(),
Option.of(predicate));
+ try {
+ return getFileGroupIterator(fileSlice, false, false);
+ } finally {
+ readerContext = originalReaderContext;
+ }
+ }
+
+ private static FileSlice createFileSlice(HoodieBaseFile baseFile,
HoodieLogFile... logFiles) {
+ return new FileSlice(new HoodieFileGroupId(PARTITION_PATH, FILE_ID),
+ baseFile == null ? null : baseFile.getCommitTime(),
+ baseFile,
+ Arrays.asList(logFiles));
+ }
+
+ private HoodieBaseFile createBaseFile(String instantTime,
List<IndexedRecord> records) throws IOException {
+ return HoodieFileSliceTestUtils.createBaseFile(
+ partitionBasePath() + "/" + FSUtils.makeBaseFileName(instantTime,
DEFAULT_WRITE_TOKEN, FILE_ID, ".parquet"),
+ records,
+ HOODIE_SCHEMA,
+ instantTime);
+ }
+
+ private HoodieLogFile createNativeDataLogFile(String instantTime, int
version, List<IndexedRecord> records)
+ throws IOException {
+ return HoodieFileSliceTestUtils.createNativeDataLogFile(
+ partitionBasePath() + "/" + nativeLogFileName(instantTime, version,
".log.parquet"),
+ records,
+ HOODIE_SCHEMA,
+ instantTime);
+ }
+
+ private HoodieLogFile createNativeDeleteLogFile(
+ HoodieStorage hoodieStorage, String instantTime, int version,
List<IndexedRecord> records) throws IOException {
+ return HoodieFileSliceTestUtils.createNativeDeleteLogFile(
+ hoodieStorage,
+ partitionBasePath() + "/" + nativeLogFileName(instantTime, version,
".deletes.parquet"),
+ records,
+ HOODIE_SCHEMA,
+ instantTime);
+ }
+
+ private HoodieLogFile createLegacyLogFile(
+ HoodieStorage hoodieStorage,
+ String instantTime,
+ int version,
+ List<IndexedRecord> records,
+ HoodieLogBlockType blockType,
+ Map<String, Long> keyToPositionMap
+ ) throws IOException, InterruptedException {
+ return HoodieFileSliceTestUtils.createLogFile(
+ hoodieStorage,
+ partitionBasePath() + "/" + logFileName(instantTime, FILE_ID, version),
+ records,
+ HOODIE_SCHEMA,
+ FILE_ID,
+ "001",
+ instantTime,
+ version,
+ blockType,
+ false,
+ keyToPositionMap);
+ }
+
+ private static Map<String, Long> keyToPositionMap(String... keys) {
+ Map<String, Long> keyToPositionMap = new HashMap<>();
+ for (int i = 0; i < keys.length; i++) {
+ keyToPositionMap.put(keys[i], (long) i);
+ }
+ return keyToPositionMap;
+ }
+
+ private void preparePartitionPath() {
+ new File(partitionBasePath()).mkdirs();
+ }
+
+ private String partitionBasePath() {
+ return basePath + "/" + PARTITION_PATH;
+ }
+
+ private static String nativeLogFileName(String instantTime, int version,
String extension) {
+ return String.format("%s_%s_%s_%d%s", FILE_ID, DEFAULT_WRITE_TOKEN,
instantTime, version, extension);
+ }
+
+ private static List<IndexedRecord> records(List<String> keys, long
timestamp, boolean isDelete) {
+ List<Long> timestamps = new ArrayList<>();
+ keys.forEach(key -> timestamps.add(timestamp));
+ return records(keys, timestamps, isDelete);
+ }
+
+ private static List<IndexedRecord> records(List<String> keys, List<Long>
timestamps, boolean isDelete) {
+ List<IndexedRecord> records = new ArrayList<>();
+ for (int i = 0; i < keys.size(); i++) {
+ records.add(HoodieFileSliceTestUtils.DATA_GEN.generateGenericRecord(
+ keys.get(i), PARTITION_PATH, "rider." + keys.get(i), "driver." +
keys.get(i), timestamps.get(i), isDelete, false));
+ }
+ return records;
+ }
+}
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
index ff9e303f7772..082cdbf2767b 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
@@ -2504,6 +2504,38 @@ public class TestHoodieTableFileSystemView extends
HoodieCommonTestHarness {
return writeStat;
}
+ @Test
+ public void testNativeLogFilesAreVisibleInFileSystemView() throws Exception {
+ String partitionPath = "2026/06/25";
+ new File(basePath + "/" + partitionPath).mkdirs();
+
+ String fileId = UUID.randomUUID().toString();
+ String instantTime = "001";
+ String nativeDataLogFileName = String.format("%s_%s_%s_%d.log.parquet",
fileId, TEST_WRITE_TOKEN, instantTime, 1);
+ String nativeDeleteLogFileName =
String.format("%s_%s_%s_%d.deletes.parquet", fileId, TEST_WRITE_TOKEN,
instantTime, 2);
+ String nativeCdcLogFileName = String.format("%s_%s_%s_%d.cdc.parquet",
fileId, TEST_WRITE_TOKEN, instantTime, 3);
+ new File(basePath + "/" + partitionPath + "/" +
nativeDataLogFileName).createNewFile();
+ new File(basePath + "/" + partitionPath + "/" +
nativeDeleteLogFileName).createNewFile();
+ new File(basePath + "/" + partitionPath + "/" +
nativeCdcLogFileName).createNewFile();
+
+ HoodieCommitMetadata commitMetadata = new HoodieCommitMetadata();
+ commitMetadata.addWriteStat(partitionPath,
getHoodieWriteStat(partitionPath, fileId, nativeDataLogFileName));
+ commitMetadata.addWriteStat(partitionPath,
getHoodieWriteStat(partitionPath, fileId, nativeDeleteLogFileName));
+ commitMetadata.addWriteStat(partitionPath,
getHoodieWriteStat(partitionPath, fileId, nativeCdcLogFileName));
+ HoodieInstant instant = INSTANT_GENERATOR.createNewInstant(State.INFLIGHT,
HoodieTimeline.DELTA_COMMIT_ACTION, instantTime);
+ saveAsComplete(metaClient.getActiveTimeline(), instant, commitMetadata);
+
+ try (SyncableFileSystemView fileSystemView =
getFileSystemView(metaClient.reloadActiveTimeline(), true)) {
+ Set<String> logFileNames = fileSystemView.getAllFileSlices(partitionPath)
+ .flatMap(FileSlice::getLogFiles)
+ .map(logFile -> logFile.getPath().getName())
+ .collect(Collectors.toSet());
+ assertTrue(logFileNames.contains(nativeDataLogFileName));
+ assertTrue(logFileNames.contains(nativeDeleteLogFileName));
+ assertTrue(logFileNames.contains(nativeCdcLogFileName));
+ }
+ }
+
@Test
public void testGetLatestMergedFileSlicesBeforeOrOnIncludingInflight()
throws IOException {
String partitionPath = "2023/01/01";
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/reader/HoodieFileGroupReaderTestHarness.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/reader/HoodieFileGroupReaderTestHarness.java
index 1ec6d41d5b87..da95c1d6acfc 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/reader/HoodieFileGroupReaderTestHarness.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/reader/HoodieFileGroupReaderTestHarness.java
@@ -133,11 +133,16 @@ public class HoodieFileGroupReaderTestHarness extends
HoodieCommonTestHarness {
FILE_ID
);
+ FileSlice fileSlice = fileSliceOpt.orElseThrow(() -> new
IllegalArgumentException("FileSlice is not present"));
+ return getFileGroupIterator(fileSlice, shouldReadPositions,
allowInflightCommits);
+ }
+
+ protected ClosableIterator<IndexedRecord> getFileGroupIterator(FileSlice
fileSlice, boolean shouldReadPositions, boolean allowInflightCommits)
+ throws IOException {
properties.setProperty(HoodieMemoryConfig.MAX_MEMORY_FOR_MERGE.key(),String.valueOf(1024
* 1024 * 1000));
properties.setProperty(HoodieMemoryConfig.SPILLABLE_MAP_BASE_PATH.key(),
basePath + "/" + HoodieTableMetaClient.TEMPFOLDER_NAME);
properties.setProperty(HoodieCommonConfig.SPILLABLE_DISK_MAP_TYPE.key(),
ExternalSpillableMap.DiskMapType.ROCKS_DB.name());
properties.setProperty(HoodieCommonConfig.DISK_MAP_BITCASK_COMPRESSION_ENABLED.key(),
"false");
- FileSlice fileSlice = fileSliceOpt.orElseThrow(() -> new
IllegalArgumentException("FileSlice is not present"));
HoodieFileGroupReader<IndexedRecord> fileGroupReader =
HoodieFileGroupReader.<IndexedRecord>builder()
.withReaderContext(readerContext)
.withHoodieTableMetaClient(metaClient)
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/reader/HoodieFileSliceTestUtils.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/reader/HoodieFileSliceTestUtils.java
index d0f5bde6f41e..0cd11d770c98 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/reader/HoodieFileSliceTestUtils.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/reader/HoodieFileSliceTestUtils.java
@@ -19,8 +19,10 @@
package org.apache.hudi.common.testutils.reader;
+import org.apache.hudi.avro.HoodieAvroWriteSupport;
import org.apache.hudi.common.bloom.BloomFilterTypeCode;
import org.apache.hudi.common.config.HoodieConfig;
+import org.apache.hudi.common.config.HoodieParquetConfig;
import org.apache.hudi.common.config.HoodieStorageConfig;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.engine.LocalTaskContextSupplier;
@@ -34,9 +36,11 @@ import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordLocation;
import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemas;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.log.HoodieLogFormat;
import org.apache.hudi.common.table.log.HoodieLogFormatWriter;
+import org.apache.hudi.common.table.log.NativeLogFooterMetadata;
import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
import org.apache.hudi.common.table.log.block.HoodieCDCDataBlock;
import org.apache.hudi.common.table.log.block.HoodieDataBlock;
@@ -50,16 +54,22 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.io.storage.HoodieAvroFileWriter;
import org.apache.hudi.io.storage.HoodieFileWriterFactory;
+import org.apache.hudi.io.storage.hadoop.HoodieAvroParquetWriter;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StoragePath;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.IndexedRecord;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -79,6 +89,7 @@ import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.AVRO_SCHE
import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.HOODIE_SCHEMA;
import static
org.apache.hudi.common.testutils.reader.DataGenerationPlan.OperationType.DELETE;
import static
org.apache.hudi.common.testutils.reader.DataGenerationPlan.OperationType.INSERT;
+import static
org.apache.parquet.avro.HoodieAvroParquetSchemaConverter.getAvroSchemaConverter;
public class HoodieFileSliceTestUtils {
public static final String FORWARD_SLASH = "/";
@@ -93,6 +104,7 @@ public class HoodieFileSliceTestUtils {
public static final HoodieTestDataGenerator DATA_GEN =
new HoodieTestDataGenerator(0xDEED);
public static final TypedProperties PROPERTIES = new TypedProperties();
+ private static final String WRITE_TOKEN = "1-0-1";
private static String[] orderingFields = new String[] {"timestamp"};
// We use a number to represent a record key, and a (start, end) range
@@ -127,6 +139,16 @@ public class HoodieFileSliceTestUtils {
instantTime, fileId, version));
}
+ private static Path generateNativeLogFilePath(
+ String basePath,
+ String fileId,
+ String instantTime,
+ int version,
+ String extension) {
+ return new Path(basePath + FORWARD_SLASH
+ + String.format("%s_%s_%s_%d%s", fileId, WRITE_TOKEN, instantTime,
version, extension));
+ }
+
// Note:
// "start < end" means start <= k <= end.
// "start == end" means k = start.
@@ -318,6 +340,85 @@ public class HoodieFileSliceTestUtils {
return new HoodieLogFile(logFilePath);
}
+ public static HoodieLogFile createNativeDataLogFile(
+ String logFilePath,
+ List<IndexedRecord> records,
+ HoodieSchema schema,
+ String logInstantTime
+ ) throws IOException {
+ HoodieStorage storage = HoodieTestUtils.getStorage(logFilePath);
+ try (HoodieAvroFileWriter writer = createNativeLogWriter(storage,
logFilePath, schema, logInstantTime)) {
+ for (IndexedRecord record : records) {
+ writer.writeAvro(
+ (String) record.get(schema.getField(ROW_KEY).get().pos()), record);
+ }
+ }
+ return new HoodieLogFile(logFilePath);
+ }
+
+ public static HoodieLogFile createNativeDeleteLogFile(
+ HoodieStorage storage,
+ String logFilePath,
+ List<IndexedRecord> records,
+ HoodieSchema tableSchema,
+ String logInstantTime
+ ) throws IOException {
+ HoodieSchema deleteLogSchema =
HoodieSchemas.createDeleteLogSchema(tableSchema, Arrays.asList(TIMESTAMP));
+ try (HoodieAvroFileWriter writer = createNativeLogWriter(storage,
logFilePath, deleteLogSchema, logInstantTime)) {
+ for (IndexedRecord record : records) {
+ String recordKey =
record.get(record.getSchema().getField(ROW_KEY).pos()).toString();
+ GenericRecord deleteRecord = new
GenericData.Record(deleteLogSchema.toAvroSchema());
+ deleteRecord.put(HoodieRecord.RECORD_KEY_METADATA_FIELD, recordKey);
+ deleteRecord.put(TIMESTAMP,
record.get(record.getSchema().getField(TIMESTAMP).pos()));
+ writer.writeAvro(recordKey, deleteRecord);
+ }
+ }
+ return new HoodieLogFile(logFilePath);
+ }
+
+ private static HoodieAvroFileWriter createNativeLogWriter(
+ HoodieStorage storage,
+ String logFilePath,
+ HoodieSchema schema,
+ String logInstantTime
+ ) throws IOException {
+ HoodieAvroWriteSupport writeSupport = new HoodieAvroWriteSupport(
+
getAvroSchemaConverter(storage.getConf().unwrapAs(Configuration.class)).convert(schema),
+ schema,
+ Option.empty(),
+ new Properties());
+ // Write the footer using the same shared serializer the real
HoodieNativeLogFormatWriter uses,
+ // so the read path is exercised against the actual on-disk contract.
+ NativeLogFooterMetadata.toFooterMetadata(getNativeLogBlockHeader(schema,
logInstantTime))
+ .forEach(writeSupport::addFooterMetadata);
+
+ HoodieParquetConfig<HoodieAvroWriteSupport> parquetConfig = new
HoodieParquetConfig<>(
+ writeSupport,
+ CompressionCodecName.GZIP,
+ ParquetWriter.DEFAULT_BLOCK_SIZE,
+ ParquetWriter.DEFAULT_PAGE_SIZE,
+ 1024 * 1024 * 1024,
+ storage.getConf(),
+ 0.1,
+ true);
+ return new HoodieAvroParquetWriter(
+ new StoragePath(logFilePath),
+ parquetConfig,
+ logInstantTime,
+ new LocalTaskContextSupplier(),
+ false);
+ }
+
+ private static Map<HoodieLogBlock.HeaderMetadataType, String>
getNativeLogBlockHeader(
+ HoodieSchema schema,
+ String logInstantTime
+ ) {
+ Map<HoodieLogBlock.HeaderMetadataType, String> header = new HashMap<>();
+ header.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME, logInstantTime);
+ header.put(HoodieLogBlock.HeaderMetadataType.SCHEMA, schema.toString());
+ return header;
+ }
+
/**
* Based on provided parameters to generate a {@link FileSlice} object.
*/
@@ -452,4 +553,90 @@ public class HoodieFileSliceTestUtils {
HOODIE_SCHEMA,
plans));
}
+
+ /**
+ * Generate a file slice mixing legacy inline logs with RFC-103 native
parquet data and delete logs.
+ */
+ public static Option<FileSlice> getMixedLegacyAndNativeFileSlice(
+ HoodieStorage storage,
+ String basePath,
+ String partitionPath,
+ String fileId
+ ) throws IOException, InterruptedException {
+ String partitionBasePath = basePath + FORWARD_SLASH + partitionPath;
+ new File(partitionBasePath).mkdirs();
+
+ Map<String, Long> keyToPositionMap = new HashMap<>();
+ DataGenerationPlan baseFilePlan = DataGenerationPlan.newBuilder()
+ .withOperationType(INSERT)
+ .withPartitionPath(partitionPath)
+ .withRecordKeys(generateKeys(new KeyRange(1, 10)))
+ .withTimeStamp(2L)
+ .withInstantTime("001")
+ .withWritePositions(false)
+ .build();
+ List<IndexedRecord> baseRecords = generateRecords(baseFilePlan);
+ HoodieBaseFile baseFile = createBaseFile(
+ generateBaseFilePath(partitionBasePath, fileId,
baseFilePlan.getInstantTime()).toString(),
+ baseRecords,
+ HOODIE_SCHEMA,
+ baseFilePlan.getInstantTime());
+ for (int i = 0; i < baseFilePlan.getRecordKeys().size(); i++) {
+ keyToPositionMap.put(baseFilePlan.getRecordKeys().get(i), (long) i);
+ }
+
+ List<HoodieLogFile> logFiles = new ArrayList<>();
+ DataGenerationPlan legacyDeletePlan = DataGenerationPlan.newBuilder()
+ .withOperationType(DELETE)
+ .withPartitionPath(partitionPath)
+ .withRecordKeys(generateKeys(new KeyRange(1, 5)))
+ .withTimeStamp(3L)
+ .withInstantTime("002")
+ .withWritePositions(false)
+ .build();
+ logFiles.add(createLogFile(
+ storage,
+ generateLogFilePath(partitionBasePath, fileId,
legacyDeletePlan.getInstantTime(), 1).toString(),
+ generateRecords(legacyDeletePlan),
+ HOODIE_SCHEMA,
+ fileId,
+ baseFilePlan.getInstantTime(),
+ legacyDeletePlan.getInstantTime(),
+ 1,
+ DELETE_BLOCK,
+ false,
+ keyToPositionMap));
+
+ DataGenerationPlan nativeDataPlan = DataGenerationPlan.newBuilder()
+ .withOperationType(INSERT)
+ .withPartitionPath(partitionPath)
+ .withRecordKeys(generateKeys(new KeyRange(1, 2)))
+ .withTimeStamp(4L)
+ .withInstantTime("003")
+ .withWritePositions(false)
+ .build();
+ logFiles.add(createNativeDataLogFile(
+ generateNativeLogFilePath(partitionBasePath, fileId,
nativeDataPlan.getInstantTime(), 2, ".log.parquet").toString(),
+ generateRecords(nativeDataPlan),
+ HOODIE_SCHEMA,
+ nativeDataPlan.getInstantTime()));
+
+ DataGenerationPlan nativeDeletePlan = DataGenerationPlan.newBuilder()
+ .withOperationType(DELETE)
+ .withPartitionPath(partitionPath)
+ .withRecordKeys(generateKeys(new KeyRange(6, 8)))
+ .withTimeStamp(1L)
+ .withInstantTime("004")
+ .withWritePositions(false)
+ .build();
+ logFiles.add(createNativeDeleteLogFile(
+ storage,
+ generateNativeLogFilePath(partitionBasePath, fileId,
nativeDeletePlan.getInstantTime(), 3, ".deletes.parquet").toString(),
+ generateRecords(nativeDeletePlan),
+ HOODIE_SCHEMA,
+ nativeDeletePlan.getInstantTime()));
+
+ return Option.of(new FileSlice(
+ new HoodieFileGroupId(partitionPath, fileId),
baseFile.getCommitTime(), baseFile, logFiles));
+ }
}
