This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 0c3f00ee0958 fix(trino): read LSM archived-timeline parquet files
through a Trino-… (#19732)
0c3f00ee0958 is described below
commit 0c3f00ee0958eb1c72889d7b593dd10cf8f92b35
Author: voonhous <[email protected]>
AuthorDate: Thu Aug 27 18:02:10 2026 +0800
fix(trino): read LSM archived-timeline parquet files through a Trino-…
(#19732)
fix(trino): read LSM archived-timeline parquet files through a Trino-backed
reader (#19732)
HudiTrinoFileReaderFactory.newParquetFileReader threw
UnsupportedOperationException, and
hudi-common's ArchivedTimelineLoaderV2 obtains its reader for the LSM
history files under
.hoodie/timeline/history from that factory. Since #14019 the archived
timeline is loaded
lazily, so the throw surfaced whenever the completion time of an archived
instant was needed,
e.g. file slicing of a MOR table whose log file belongs to an archived
delta commit.
Add TrinoParquetFileReader, a HoodieAvroFileReader over Trino's
ParquetReader, and return it
from the factory. The constructor reads the footer once, taking the schema
from the file's
parquet.avro.schema property and the record count from its row groups. Each
getIndexedRecordIterator opens its own ParquetReader over the projection
the loader requests,
with column handles typed by HudiUtil.toColumnHandle, and turns every page
into IndexedRecords
through HudiAvroSerializer; VARBINARY values become the ByteBuffer that
ArchivedTimelineV2
casts the metadata and plan columns to. The iterator remembers exhaustion,
since on Trino 484
a nextPage() past the last row group throws instead of returning null
again, and closing the
reader releases every iterator it handed out, rethrowing the first close
failure with the
rest attached as suppressed. Failures at open, page read (checked or
unchecked) and close go
through the same handleException as Hive's ParquetPageSource: a
ParquetCorruptionException is
HUDI_BAD_DATA, anything else HUDI_CURSOR_ERROR, each with its cause.
Record-key, key-prefix and
row-key lookups, the bloom filter and the min/max record keys are
unsupported: a timeline
file carries none of the data-file footer metadata those rely on, and no
Trino path asks for
them. Only LoadMode.TIME is reachable from the connector today.
Ported from onehouseinc/trino#72 and adapted to the HoodieSchema-based
reader contract, the
PrefilledColumnValues serializer and Trino 484's ParquetReader API.
Tests. TestTrinoParquetFileReader reads archived_timeline.parquet, the
first history file of
a four-instant tv8 COW table written with Hudi 1.0.2 (create script in
hudi_cow_archived_timeline.md): the full read, the TIME, METADATA, PLAN and
FULL projections
(FULL orders plan before metadata, the reverse of the file, so it pins
name-based mapping and
the ByteBuffer conversion), a drained iterator staying drained, close()
releasing an open
iterator, a corrupt footer failing with HUDI_BAD_DATA, a corrupt data page
failing with
HUDI_CURSOR_ERROR and the ParquetDecodingException as its cause, and the
unsupported footer
lookups. TestHudiSmokeTest.testReadTableWithArchivedTimeline reads a tv8
MOR fixture
(hudi_mor_archived_timeline.zip, script in hudi_mor_archived_timeline.md)
whose oldest log
file belongs to archived delta commit 20250918122106595, read-optimized and
real-time; both
cases fail on master.
Fixes #13994
---
.../plugin/hudi/io/HudiTrinoFileReaderFactory.java | 2 +-
.../plugin/hudi/io/TrinoParquetFileReader.java | 443 +++++++++++++++++++++
.../io/trino/plugin/hudi/TestHudiSmokeTest.java | 25 ++
.../plugin/hudi/io/TestTrinoParquetFileReader.java | 252 ++++++++++++
.../testing/ResourceHudiTablesInitializer.java | 1 +
.../src/test/resources/archived_timeline.parquet | Bin 0 -> 3263 bytes
.../hudi_cow_archived_timeline.md | 124 ++++++
.../hudi_mor_archived_timeline.md | 96 +++++
.../hudi_mor_archived_timeline.zip | Bin 0 -> 242504 bytes
9 files changed, 942 insertions(+), 1 deletion(-)
diff --git
a/hudi-trino/src/main/java/io/trino/plugin/hudi/io/HudiTrinoFileReaderFactory.java
b/hudi-trino/src/main/java/io/trino/plugin/hudi/io/HudiTrinoFileReaderFactory.java
index cf666d5e9883..50be5b9f8066 100644
---
a/hudi-trino/src/main/java/io/trino/plugin/hudi/io/HudiTrinoFileReaderFactory.java
+++
b/hudi-trino/src/main/java/io/trino/plugin/hudi/io/HudiTrinoFileReaderFactory.java
@@ -38,7 +38,7 @@ public class HudiTrinoFileReaderFactory
@Override
protected HoodieFileReader newParquetFileReader(StoragePath path)
{
- throw new UnsupportedOperationException("HudiTrinoFileReaderFactory
does not support Parquet file reader");
+ return new TrinoParquetFileReader(storage, path);
}
@Override
diff --git
a/hudi-trino/src/main/java/io/trino/plugin/hudi/io/TrinoParquetFileReader.java
b/hudi-trino/src/main/java/io/trino/plugin/hudi/io/TrinoParquetFileReader.java
new file mode 100644
index 000000000000..5548763d4d5a
--- /dev/null
+++
b/hudi-trino/src/main/java/io/trino/plugin/hudi/io/TrinoParquetFileReader.java
@@ -0,0 +1,443 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.io;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.filesystem.TrinoFileSystem;
+import io.trino.filesystem.TrinoInputFile;
+import io.trino.memory.context.AggregatedMemoryContext;
+import io.trino.parquet.Column;
+import io.trino.parquet.Field;
+import io.trino.parquet.ParquetCorruptionException;
+import io.trino.parquet.ParquetDataSource;
+import io.trino.parquet.ParquetReaderOptions;
+import io.trino.parquet.metadata.BlockMetadata;
+import io.trino.parquet.metadata.FileMetadata;
+import io.trino.parquet.metadata.ParquetMetadata;
+import io.trino.parquet.predicate.TupleDomainParquetPredicate;
+import io.trino.parquet.reader.MetadataReader;
+import io.trino.parquet.reader.ParquetReader;
+import io.trino.parquet.reader.RowGroupInfo;
+import io.trino.plugin.base.metrics.FileFormatDataSourceStats;
+import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hive.parquet.ParquetReaderConfig;
+import io.trino.plugin.hudi.HudiUtil;
+import io.trino.plugin.hudi.storage.HudiTrinoStorage;
+import io.trino.plugin.hudi.util.HudiAvroSerializer;
+import io.trino.spi.Page;
+import io.trino.spi.TrinoException;
+import io.trino.spi.connector.SourcePage;
+import io.trino.spi.predicate.TupleDomain;
+import io.trino.spi.type.SqlVarbinary;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.hudi.common.bloom.BloomFilter;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.core.io.storage.HoodieAvroFileReader;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.io.MessageColumnIO;
+import org.apache.parquet.schema.MessageType;
+import org.joda.time.DateTimeZone;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static
io.trino.memory.context.AggregatedMemoryContext.newSimpleAggregatedMemoryContext;
+import static io.trino.parquet.ParquetTypeUtils.constructField;
+import static io.trino.parquet.ParquetTypeUtils.getColumnIO;
+import static io.trino.parquet.ParquetTypeUtils.getDescriptors;
+import static io.trino.parquet.ParquetTypeUtils.lookupColumnByName;
+import static io.trino.parquet.predicate.PredicateUtils.buildPredicate;
+import static io.trino.parquet.predicate.PredicateUtils.getFilteredRowGroups;
+import static
io.trino.plugin.hive.parquet.ParquetPageSourceFactory.createDataSource;
+import static
io.trino.plugin.hive.parquet.ParquetPageSourceFactory.getParquetMessageType;
+import static io.trino.plugin.hudi.HudiErrorCode.HUDI_BAD_DATA;
+import static io.trino.plugin.hudi.HudiErrorCode.HUDI_CURSOR_ERROR;
+import static io.trino.plugin.hudi.HudiErrorCode.HUDI_SCHEMA_ERROR;
+import static io.trino.spi.type.VarbinaryType.VARBINARY;
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Reads an LSM archived-timeline Parquet file through Trino's {@link
ParquetReader}, turning each
+ * {@link Page} it produces into an Avro {@link IndexedRecord} with {@link
HudiAvroSerializer}. Hudi's
+ * archived-timeline loader asks {@link HudiTrinoFileReaderFactory} for an
Avro file reader over the
+ * history files under {@code .hoodie/timeline/history}, and this is the
connector's answer to that
+ * request. It is not a general data-file reader: record-key, key-prefix and
row-key lookups are
+ * unsupported, and so are the bloom filter and min/max record key lookups --
a timeline file carries
+ * none of the data-file footer metadata those rely on.
+ */
+public class TrinoParquetFileReader
+ extends HoodieAvroFileReader
+{
+ private static final String PARQUET_AVRO_SCHEMA_KEY =
"parquet.avro.schema";
+ private static final DateTimeZone UTC_TIME_ZONE = DateTimeZone.UTC;
+ private static final int DOMAIN_COMPACTION_THRESHOLD = 1000;
+
+ private final StoragePath path;
+ private final ParquetReaderOptions readerOptions = new
ParquetReaderConfig().toParquetReaderOptions();
+ private final TrinoInputFile inputFile;
+ private final long fileLength;
+ private final ParquetMetadata parquetMetadata;
+ private final HoodieSchema hoodieSchema;
+ private final long totalRecords;
+ // Every iterator handed out by getIndexedRecordIterator, so that close()
can release the ParquetReader
+ // of one the caller left open. A reader instance is used by a single
thread, but the list is cheap to guard
+ private final List<ParquetIndexedRecordIterator> openIterators = new
ArrayList<>();
+
+ public TrinoParquetFileReader(HoodieStorage storage, StoragePath path)
+ {
+ this.path = requireNonNull(path, "path is null");
+ requireNonNull(storage, "storage is null");
+ checkArgument(storage instanceof HudiTrinoStorage, "storage must be an
instance of HudiTrinoStorage");
+ HudiTrinoStorage trinoStorage = (HudiTrinoStorage) storage;
+
+ // HudiTrinoStorage#getFileSystem is typed Object so that hudi-common
stays free of Trino types
+ TrinoFileSystem fileSystem = (TrinoFileSystem)
trinoStorage.getFileSystem();
+ this.inputFile =
fileSystem.newInputFile(HudiTrinoStorage.convertToLocation(path));
+ try {
+ this.fileLength = inputFile.length();
+ this.parquetMetadata = readParquetMetadata();
+ // The footer's row-group metadata is parsed lazily, so
getBlocks() can fail the way readFooter does
+ this.totalRecords = parquetMetadata.getBlocks().stream()
+ .mapToLong(BlockMetadata::rowCount)
+ .sum();
+ }
+ catch (IOException e) {
+ // Failing to open the file surfaces the way a failing read does:
a corrupt footer as HUDI_BAD_DATA,
+ // anything else as HUDI_CURSOR_ERROR, each with its cause attached
+ throw handleException(path, e);
+ }
+ Schema avroSchema =
extractAvroSchema(parquetMetadata.getFileMetaData());
+ this.hoodieSchema = HoodieSchema.fromAvroSchema(avroSchema);
+ }
+
+ @Override
+ public ClosableIterator<IndexedRecord>
getIndexedRecordIterator(HoodieSchema readerSchema, HoodieSchema
requestedSchema, Map<String, String> renamedColumns)
+ {
+ // Timeline files are never schema-evolved, so no column can have been
renamed under them
+ HoodieSchema projectedSchema = requestedSchema != null ?
requestedSchema : hoodieSchema;
+ ParquetIndexedRecordIterator iterator = new
ParquetIndexedRecordIterator(projectedSchema);
+ synchronized (openIterators) {
+ openIterators.add(iterator);
+ }
+ return iterator;
+ }
+
+ @Override
+ public ClosableIterator<IndexedRecord>
getIndexedRecordsByKeysIterator(List<String> keys, HoodieSchema readerSchema)
+ {
+ throw new UnsupportedOperationException("Reading records by keys is
not supported by this reader");
+ }
+
+ @Override
+ public ClosableIterator<IndexedRecord>
getIndexedRecordsByKeyPrefixIterator(List<String> sortedKeyPrefixes,
HoodieSchema readerSchema)
+ {
+ throw new UnsupportedOperationException("Reading records by key
prefixes is not supported by this reader");
+ }
+
+ @Override
+ public String[] readMinMaxRecordKeys()
+ {
+ throw new UnsupportedOperationException("Reading min/max record keys
is not supported by this reader");
+ }
+
+ @Override
+ public BloomFilter readBloomFilter()
+ {
+ throw new UnsupportedOperationException("Reading a bloom filter is not
supported by this reader");
+ }
+
+ @Override
+ public Set<Pair<String, Long>> filterRowKeys(Set<String> candidateRowKeys)
+ {
+ throw new UnsupportedOperationException("Filtering row keys is not
supported by this reader");
+ }
+
+ @Override
+ public ClosableIterator<String> getRecordKeyIterator()
+ {
+ throw new UnsupportedOperationException("Iterating over only record
keys is not supported by this reader");
+ }
+
+ @Override
+ public HoodieSchema getSchema()
+ {
+ return hoodieSchema;
+ }
+
+ @Override
+ public long getTotalRecords()
+ {
+ return totalRecords;
+ }
+
+ @Override
+ public void close()
+ {
+ // The only resource this reader opens outside the constructor is the
ParquetReader of an iterator.
+ // Closing the reader releases every iterator it handed out, including
any the caller left open;
+ // an iterator's close() is idempotent, so closing one that is already
closed is a no-op. One that fails
+ // to close does not stop the others from being closed: the first
failure is rethrown once every iterator
+ // has been released, with the later failures attached to it as
suppressed
+ synchronized (openIterators) {
+ RuntimeException failure = null;
+ for (ParquetIndexedRecordIterator iterator : openIterators) {
+ try {
+ iterator.close();
+ }
+ catch (RuntimeException e) {
+ if (failure == null) {
+ failure = e;
+ }
+ else {
+ failure.addSuppressed(e);
+ }
+ }
+ }
+ openIterators.clear();
+ if (failure != null) {
+ throw failure;
+ }
+ }
+ }
+
+ private ParquetMetadata readParquetMetadata()
+ throws IOException
+ {
+ // No estimated size: with one at or below the small-file threshold
createDataSource returns a
+ // MemoryParquetDataSource that slurps the whole file, where the
footer read only needs the tail
+ try (ParquetDataSource dataSource =
openDataSource(newSimpleAggregatedMemoryContext(), OptionalLong.empty())) {
+ return MetadataReader.readFooter(dataSource, Optional.empty());
+ }
+ }
+
+ private ParquetDataSource openDataSource(AggregatedMemoryContext
memoryContext, OptionalLong estimatedFileSize)
+ throws IOException
+ {
+ return createDataSource(inputFile, estimatedFileSize, readerOptions,
memoryContext, new FileFormatDataSourceStats());
+ }
+
+ private Schema extractAvroSchema(FileMetadata fileMetaData)
+ {
+ String avroSchemaStr =
fileMetaData.getKeyValueMetaData().get(PARQUET_AVRO_SCHEMA_KEY);
+ if (avroSchemaStr == null) {
+ throw new TrinoException(HUDI_SCHEMA_ERROR, "Parquet file does not
contain Avro schema in metadata: " + path);
+ }
+ return new Schema.Parser().parse(avroSchemaStr);
+ }
+
+ /**
+ * One {@link HiveColumnHandle} per field of the projection, typed from
the field's Avro schema the way
+ * {@link HudiUtil#toColumnHandle} types the handles of a data-file read.
Handle {@code i} is built from
+ * field {@code i}, so a handle's position in the returned list is its
field's position in the records this
+ * reader produces. That list position is what {@link
#binaryFieldPositions} relies on, not the handle's own
+ * column index, which {@code toColumnHandle} leaves at 0.
+ */
+ private static List<HiveColumnHandle> buildColumnHandles(HoodieSchema
projectedSchema)
+ {
+ return projectedSchema.getFields().stream()
+ .map(HudiUtil::toColumnHandle)
+ .toList();
+ }
+
+ private static List<Column> buildTrinoColumns(List<HiveColumnHandle>
columnHandles, MessageColumnIO messageColumnIO)
+ {
+ ImmutableList.Builder<Column> columnsBuilder = ImmutableList.builder();
+ for (HiveColumnHandle columnHandle : columnHandles) {
+ String name = columnHandle.getName();
+ Field parquetField = constructField(columnHandle.getType(),
lookupColumnByName(messageColumnIO, name))
+ .orElseThrow(() -> new TrinoException(HUDI_SCHEMA_ERROR,
"Could not find column: " + name));
+ columnsBuilder.add(new Column(name, parquetField));
+ }
+ return columnsBuilder.build();
+ }
+
+ /**
+ * Record positions of the projection's VARBINARY columns, if any. Trino
hands a VARBINARY value out as a
+ * {@link SqlVarbinary}, while Avro's in-memory representation of {@code
bytes} is a {@link ByteBuffer} --
+ * and that is what hudi-common casts to when it reads the {@code
metadata} and {@code plan} columns of an
+ * LSM instant, so those values have to be converted before the record
leaves this reader.
+ */
+ private static int[] binaryFieldPositions(List<HiveColumnHandle>
columnHandles)
+ {
+ return IntStream.range(0, columnHandles.size())
+ .filter(position ->
columnHandles.get(position).getType().equals(VARBINARY))
+ .toArray();
+ }
+
+ private static TrinoException handleException(StoragePath path, Exception
exception)
+ {
+ if (exception instanceof TrinoException trinoException) {
+ return trinoException;
+ }
+ if (exception instanceof ParquetCorruptionException) {
+ return new TrinoException(HUDI_BAD_DATA, exception);
+ }
+ return new TrinoException(HUDI_CURSOR_ERROR, "Failed to read Parquet
file: " + path, exception);
+ }
+
+ private class ParquetIndexedRecordIterator
+ implements ClosableIterator<IndexedRecord>
+ {
+ private final ParquetReader parquetReader;
+ private final HudiAvroSerializer avroSerializer;
+ private final int[] binaryFieldPositions;
+ private Page currentPage;
+ private int currentPosition;
+ private boolean exhausted;
+ private boolean closed;
+
+ ParquetIndexedRecordIterator(HoodieSchema projectedSchema)
+ {
+ List<HiveColumnHandle> columnHandles =
buildColumnHandles(projectedSchema);
+ this.parquetReader = createParquetReader(columnHandles);
+ // Null prefilled values: PrefilledColumnValues answers partition
and hidden metadata columns of a
+ // split, of which a timeline read has neither, and serialize()
only ever reads page values. There
+ // is no split here to build one from -- create(HudiSplit) is its
only factory.
+ this.avroSerializer = new HudiAvroSerializer(columnHandles, null,
projectedSchema.toAvroSchema());
+ this.binaryFieldPositions = binaryFieldPositions(columnHandles);
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+ if (closed || exhausted) {
+ return false;
+ }
+ if (currentPage != null && currentPosition <
currentPage.getPositionCount()) {
+ return true;
+ }
+ try {
+ loadNextPage();
+ return !exhausted;
+ }
+ catch (IOException | RuntimeException e) {
+ // loadNextPage() decodes the page eagerly, and a data page
that fails to decode surfaces as a
+ // ParquetDecodingException, which is unchecked: ParquetReader
routes only the IOExceptions of a
+ // page read through the exception transform it is handed
+ throw handleException(path, e);
+ }
+ }
+
+ @Override
+ public IndexedRecord next()
+ {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ IndexedRecord record = avroSerializer.serialize(currentPage,
currentPosition);
+ for (int fieldPosition : binaryFieldPositions) {
+ if (record.get(fieldPosition) instanceof SqlVarbinary value) {
+ record.put(fieldPosition,
ByteBuffer.wrap(value.getBytes()));
+ }
+ }
+ currentPosition++;
+ return record;
+ }
+
+ @Override
+ public void close()
+ {
+ if (!closed) {
+ closed = true;
+ currentPage = null;
+ try {
+ // Also closes the underlying ParquetDataSource
+ parquetReader.close();
+ }
+ catch (IOException e) {
+ throw handleException(path, e);
+ }
+ }
+ }
+
+ private ParquetReader createParquetReader(List<HiveColumnHandle>
columnHandles)
+ {
+ AggregatedMemoryContext memoryContext =
newSimpleAggregatedMemoryContext();
+ ParquetDataSource dataSource = null;
+ try {
+ dataSource = openDataSource(memoryContext,
OptionalLong.of(fileLength));
+ FileMetadata fileMetaData = parquetMetadata.getFileMetaData();
+ MessageType fileSchema = fileMetaData.getSchema();
+ MessageType requestedSchema =
getParquetMessageType(columnHandles, true, fileSchema)
+ .orElse(new MessageType(fileSchema.getName(),
ImmutableList.of()));
+ List<Column> columns = buildTrinoColumns(columnHandles,
getColumnIO(fileSchema, requestedSchema));
+
+ Map<List<String>, ColumnDescriptor> descriptorsByPath =
getDescriptors(fileSchema, requestedSchema);
+ TupleDomain<ColumnDescriptor> tupleDomain = TupleDomain.all();
+ TupleDomainParquetPredicate parquetPredicate =
buildPredicate(requestedSchema, tupleDomain, descriptorsByPath, UTC_TIME_ZONE);
+ List<RowGroupInfo> rowGroups = getFilteredRowGroups(
+ 0,
+ fileLength,
+ dataSource,
+ parquetMetadata,
+ ImmutableList.of(tupleDomain),
+ ImmutableList.of(parquetPredicate),
+ descriptorsByPath,
+ UTC_TIME_ZONE,
+ DOMAIN_COMPACTION_THRESHOLD,
+ readerOptions);
+
+ return new ParquetReader(
+ Optional.ofNullable(fileMetaData.getCreatedBy()),
+ columns,
+ false,
+ rowGroups,
+ dataSource,
+ UTC_TIME_ZONE,
+ memoryContext,
+ readerOptions,
+ exception -> handleException(path, exception),
+ Optional.of(parquetPredicate),
+ Optional.empty(),
+ parquetMetadata.getDecryptionContext());
+ }
+ catch (IOException | RuntimeException e) {
+ // The reader owns the data source only once it is constructed
+ if (dataSource != null) {
+ try {
+ dataSource.close();
+ }
+ catch (IOException _) {
+ }
+ }
+ throw handleException(path, e);
+ }
+ }
+
+ private void loadNextPage()
+ throws IOException
+ {
+ SourcePage sourcePage = parquetReader.nextPage();
+ // Once the reader has handed out its last page it must never be
asked again: a nextPage() past
+ // the end of the row groups throws IndexOutOfBoundsException
instead of returning null a second time
+ exhausted = sourcePage == null;
+ currentPage = exhausted ? null : sourcePage.getPage();
+ currentPosition = 0;
+ }
+ }
+}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
index 981191962f28..70da4b1249cd 100644
--- a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
+++ b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
@@ -87,6 +87,7 @@ import static
io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.Testing
import static
io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_COW_TABLE_WITH_FIELD_NAMES_IN_CAPS;
import static
io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_COW_TABLE_WITH_MULTI_KEYS_AND_FIELD_NAMES_IN_CAPS;
import static
io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_CUSTOM_KEYGEN_PT_V8_MOR;
+import static
io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_MOR_ARCHIVED_TIMELINE;
import static
io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_MULTI_PT_V8_MOR;
import static
io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_NON_EXTRACTABLE_PARTITION_PATH;
import static
io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_NON_PART_COW;
@@ -1386,6 +1387,30 @@ public class TestHudiSmokeTest
SchemaEvolutionHudiTablesInitializer.TABLE_NAME, predicate);
}
+ // The MOR fixture's oldest log file belongs to delta commit
20250918122106595, which has been archived: the
+ // first instant on the active timeline is 20250918122107347. File slicing
therefore has to look that log
+ // file's completion time up in the LSM archived timeline, a read that
goes through
+ // HudiTrinoFileReaderFactory#newParquetFileReader and threw
UnsupportedOperationException before
+ // TrinoParquetFileReader existed (apache/hudi#13994).
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testReadTableWithArchivedTimeline(boolean isRtTable)
+ {
+ String tableName = isRtTable ?
HUDI_MOR_ARCHIVED_TIMELINE.getRtTableName()
+ : HUDI_MOR_ARCHIVED_TIMELINE.getTableName();
+ @Language("SQL") String actualQuery = "SELECT id, name, price, ts FROM
" + tableName;
+ @Language("SQL") String expectedQuery;
+ if (isRtTable) {
+ // Real-time table, log files are merged onto the base files
+ expectedQuery = "VALUES (2, 'updated_user2', 20.0, 2000), (3,
'user3', 30.0, 3000), (4, 'user4', 40.0, 4000), (5, 'user5', 50.0, 5000)";
+ }
+ else {
+ // Read-optimized table, base files only (the fixture was written
with inline compaction disabled)
+ expectedQuery = "VALUES (1, 'user1', 10.0, 1000), (2, 'user2',
20.0, 2000), (3, 'user3', 30.0, 3000), (4, 'user4', 40.0, 4000), (5, 'user5',
50.0, 5000)";
+ }
+ assertQuery(actualQuery, expectedQuery);
+ }
+
private void testTimestampMicros(HiveTimestampPrecision
timestampPrecision, LocalDateTime expected)
throws Exception
{
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/io/TestTrinoParquetFileReader.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/io/TestTrinoParquetFileReader.java
new file mode 100644
index 000000000000..f1691cefb1cb
--- /dev/null
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/io/TestTrinoParquetFileReader.java
@@ -0,0 +1,252 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.io;
+
+import com.google.common.io.Resources;
+import io.trino.filesystem.local.LocalFileSystem;
+import io.trino.parquet.ParquetCorruptionException;
+import io.trino.plugin.hudi.storage.HudiTrinoStorage;
+import io.trino.plugin.hudi.storage.TrinoStorageConfiguration;
+import io.trino.spi.TrinoException;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.hudi.avro.model.HoodieLSMTimelineInstant;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.table.timeline.HoodieArchivedTimeline;
+import org.apache.hudi.common.table.timeline.LSMTimeline;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.parquet.io.ParquetDecodingException;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.NoSuchElementException;
+
+import static io.trino.plugin.hudi.HudiErrorCode.HUDI_BAD_DATA;
+import static io.trino.plugin.hudi.HudiErrorCode.HUDI_CURSOR_ERROR;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests {@link TrinoParquetFileReader} against a four-instant LSM
archived-timeline parquet file, the shape Hudi's
+ * archived-timeline loader reads through the connector. {@code
archived_timeline.parquet} is the first history file,
+ * {@code 20250918121953134_20250918122001506_0.parquet}, of the
table-version-8 COW table that the create script in
+ * {@code hudi-testing-data/hudi_cow_archived_timeline.md} produces with Hudi
1.0.2: four commit instants,
+ * 20250918121953134 through 20250918122001506. Only that file is checked in,
not the table.
+ */
+class TestTrinoParquetFileReader
+{
+ private static final String ARCHIVED_TIMELINE_PARQUET_FILE =
"archived_timeline.parquet";
+ // The fixture's first page is the only data page of instantTime: 25 bytes
of thrift-compact PageHeader from
+ // byte 4 (type, sizes, crc, then the DataPageHeader whose encoding field
is the byte at 22), then the payload.
+ // Thrift compact writes the encoding enum as a zigzag varint: PLAIN (0)
is 0x00, RLE_DICTIONARY (8) is 0x10
+ private static final int INSTANT_TIME_PAGE_ENCODING_OFFSET = 22;
+ private static final byte PLAIN_ENCODING = 0x00;
+ private static final byte RLE_DICTIONARY_ENCODING = 0x10;
+
+ @Test
+ void testReadArchivedTimelineFile()
+ throws Exception
+ {
+ HoodieSchema tableSchema =
HoodieSchema.fromAvroSchema(HoodieLSMTimelineInstant.getClassSchema());
+ Schema avroSchema = tableSchema.toAvroSchema();
+
+ try (TrinoParquetFileReader reader = createReader()) {
+
assertThat(reader.getSchema().toAvroSchema()).isEqualTo(HoodieLSMTimelineInstant.getClassSchema());
+ assertThat(reader.getTotalRecords()).isEqualTo(4);
+
+ List<IndexedRecord> records = new ArrayList<>();
+ try (ClosableIterator<IndexedRecord> iterator =
reader.getIndexedRecordIterator(tableSchema, tableSchema)) {
+ iterator.forEachRemaining(records::add);
+ }
+ assertThat(records).hasSize(4);
+
+ IndexedRecord firstRecord = records.getFirst();
+
assertThat(firstRecord.get(avroSchema.getField("instantTime").pos()).toString()).isEqualTo("20250918121953134");
+
assertThat(firstRecord.get(avroSchema.getField("completionTime").pos()).toString()).isEqualTo("20250918121957816");
+
assertThat(firstRecord.get(avroSchema.getField("action").pos()).toString()).isEqualTo("commit");
+
+ IndexedRecord secondRecord = records.get(1);
+
assertThat(secondRecord.get(avroSchema.getField("instantTime").pos()).toString()).isEqualTo("20250918121958100");
+
assertThat(secondRecord.get(avroSchema.getField("completionTime").pos()).toString()).isEqualTo("20250918121959081");
+
assertThat(secondRecord.get(avroSchema.getField("action").pos()).toString()).isEqualTo("commit");
+
+ IndexedRecord lastRecord = records.getLast();
+
assertThat(lastRecord.get(avroSchema.getField("instantTime").pos()).toString()).isEqualTo("20250918122001506");
+
assertThat(lastRecord.get(avroSchema.getField("completionTime").pos()).toString()).isEqualTo("20250918122002218");
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = HoodieArchivedTimeline.LoadMode.class, names =
{"TIME", "METADATA", "PLAN", "FULL"})
+ void testProjectedReadUsesRequestedSchema(HoodieArchivedTimeline.LoadMode
loadMode)
+ throws Exception
+ {
+ // The projections the archived-timeline readers request. TIME is what
CompletionTimeQueryViewV2 asks for when
+ // it only needs instant times; METADATA and PLAN are what
ArchivedTimelineV2 asks for when it needs a payload,
+ // and it casts that bytes column to ByteBuffer -- the SqlVarbinary ->
ByteBuffer conversion exists for those
+ // two. FULL is requested only by EightToSevenDowngradeHandler, but it
is the one projection that orders plan
+ // before metadata, the reverse of the file's metadata, plan, so a
passing FULL read proves columns are mapped
+ // by name and not by position
+ Schema projectedAvroSchema = LSMTimeline.getReadSchema(loadMode);
+ HoodieSchema tableSchema =
HoodieSchema.fromAvroSchema(HoodieLSMTimelineInstant.getClassSchema());
+ HoodieSchema projectedSchema =
HoodieSchema.fromAvroSchema(projectedAvroSchema);
+
+ try (TrinoParquetFileReader reader = createReader()) {
+ List<IndexedRecord> records = new ArrayList<>();
+ try (ClosableIterator<IndexedRecord> iterator =
reader.getIndexedRecordIterator(tableSchema, projectedSchema)) {
+ iterator.forEachRemaining(records::add);
+ }
+ assertThat(records).hasSize(4);
+ assertThat(records).allSatisfy(record ->
assertThat(record.getSchema()).isEqualTo(projectedAvroSchema));
+
+ int instantTimePos =
projectedAvroSchema.getField("instantTime").pos();
+ int completionTimePos =
projectedAvroSchema.getField("completionTime").pos();
+
assertThat(records.getFirst().get(instantTimePos).toString()).isEqualTo("20250918121953134");
+
assertThat(records.getFirst().get(completionTimePos).toString()).isEqualTo("20250918121957816");
+
assertThat(records.get(1).get(instantTimePos).toString()).isEqualTo("20250918121958100");
+
assertThat(records.get(1).get(completionTimePos).toString()).isEqualTo("20250918121959081");
+
+ // All four instants of the fixture are commits: every row has
metadata bytes and a null plan. METADATA and
+ // PLAN each carry only their own bytes column and FULL carries
both, so the two are checked independently
+ if (projectedAvroSchema.getField("metadata") != null) {
+ int metadataPos =
projectedAvroSchema.getField("metadata").pos();
+ assertThat(records).allSatisfy(record -> {
+
assertThat(record.get(metadataPos)).isInstanceOf(ByteBuffer.class);
+ assertThat(((ByteBuffer)
record.get(metadataPos)).remaining()).isGreaterThan(0);
+ });
+ }
+ if (projectedAvroSchema.getField("plan") != null) {
+ int planPos = projectedAvroSchema.getField("plan").pos();
+ assertThat(records).allSatisfy(record ->
assertThat(record.get(planPos)).isNull());
+ }
+ }
+ }
+
+ @Test
+ void testDrainedIteratorStaysDrained()
+ throws Exception
+ {
+ // Once the ParquetReader has handed out its last page it must not be
asked for another: a nextPage() past the
+ // end of the row groups throws instead of returning null again, so
the iterator has to remember it is done
+ HoodieSchema tableSchema =
HoodieSchema.fromAvroSchema(HoodieLSMTimelineInstant.getClassSchema());
+ try (TrinoParquetFileReader reader = createReader();
+ ClosableIterator<IndexedRecord> iterator =
reader.getIndexedRecordIterator(tableSchema, tableSchema)) {
+ List<IndexedRecord> records = new ArrayList<>();
+ iterator.forEachRemaining(records::add);
+ assertThat(records).hasSize(4);
+
+ assertThat(iterator.hasNext()).isFalse();
+ assertThat(iterator.hasNext()).isFalse();
+
assertThatThrownBy(iterator::next).isInstanceOf(NoSuchElementException.class);
+ }
+ }
+
+ @Test
+ void testCloseReleasesOpenIterator()
+ throws Exception
+ {
+ // Closing the reader closes an iterator the caller left open; closing
that iterator afterwards is a no-op
+ HoodieSchema tableSchema =
HoodieSchema.fromAvroSchema(HoodieLSMTimelineInstant.getClassSchema());
+ TrinoParquetFileReader reader = createReader();
+ ClosableIterator<IndexedRecord> iterator =
reader.getIndexedRecordIterator(tableSchema, tableSchema);
+ assertThat(iterator.hasNext()).isTrue();
+ assertThat(iterator.next()).isNotNull();
+
+ reader.close();
+ assertThat(iterator.hasNext()).isFalse();
+
assertThatThrownBy(iterator::next).isInstanceOf(NoSuchElementException.class);
+ iterator.close();
+ reader.close();
+ }
+
+ @Test
+ void testCorruptFileFailsWithBadData(@TempDir Path tempDir)
+ throws Exception
+ {
+ // A footer that does not parse is reported the way a failed page read
is: HUDI_BAD_DATA with the corruption as
+ // its cause. The ParquetCorruptionException it starts out as is an
IOException, and left raw it would reach
+ // hudi-common as a HoodieIOException instead
+ Path corruptFile = tempDir.resolve("corrupt.parquet");
+ Files.write(corruptFile, new byte[] {1, 2, 3});
+ StoragePath path = new
StoragePath(corruptFile.toFile().toURI().toString());
+
+ assertThatThrownBy(() -> new TrinoParquetFileReader(localStorage(),
path))
+ .isInstanceOf(TrinoException.class)
+ .hasCauseInstanceOf(ParquetCorruptionException.class)
+ .extracting(e -> ((TrinoException) e).getErrorCode())
+ .isEqualTo(HUDI_BAD_DATA.toErrorCode());
+ }
+
+ @Test
+ void testCorruptDataPageFailsWithCursorError(@TempDir Path tempDir)
+ throws Exception
+ {
+ // A data page that fails to decode is reported the way a failed
footer read is: as a TrinoException with the
+ // failure as its cause. Trino decodes a page eagerly and reports a
corrupt one with a ParquetDecodingException,
+ // which is unchecked, so left raw it would pass through hudi-common
untouched and reach the engine as an
+ // internal error instead of HUDI_CURSOR_ERROR. The footer is intact,
so the reader still opens and the failure
+ // surfaces from hasNext(). Switching the page's encoding from PLAIN
to RLE_DICTIONARY makes it claim a
+ // dictionary the column chunk does not carry, which is what the
column reader rejects
+ byte[] bytes =
Resources.toByteArray(Resources.getResource(ARCHIVED_TIMELINE_PARQUET_FILE));
+
assertThat(bytes[INSTANT_TIME_PAGE_ENCODING_OFFSET]).isEqualTo(PLAIN_ENCODING);
+ bytes[INSTANT_TIME_PAGE_ENCODING_OFFSET] = RLE_DICTIONARY_ENCODING;
+ Path corruptFile = tempDir.resolve("corrupt_page.parquet");
+ Files.write(corruptFile, bytes);
+ StoragePath path = new
StoragePath(corruptFile.toFile().toURI().toString());
+
+ HoodieSchema tableSchema =
HoodieSchema.fromAvroSchema(HoodieLSMTimelineInstant.getClassSchema());
+ try (TrinoParquetFileReader reader = new
TrinoParquetFileReader(localStorage(), path);
+ ClosableIterator<IndexedRecord> iterator =
reader.getIndexedRecordIterator(tableSchema, tableSchema)) {
+ assertThatThrownBy(iterator::hasNext)
+ .isInstanceOf(TrinoException.class)
+ .hasCauseInstanceOf(ParquetDecodingException.class)
+ .extracting(e -> ((TrinoException) e).getErrorCode())
+ .isEqualTo(HUDI_CURSOR_ERROR.toErrorCode());
+ }
+ }
+
+ @Test
+ void testFooterLookupsUnsupported()
+ throws Exception
+ {
+ // A timeline file is not a data file: it carries neither a bloom
filter nor min/max record keys
+ try (TrinoParquetFileReader reader = createReader()) {
+
assertThatThrownBy(reader::readBloomFilter).isInstanceOf(UnsupportedOperationException.class);
+
assertThatThrownBy(reader::readMinMaxRecordKeys).isInstanceOf(UnsupportedOperationException.class);
+ }
+ }
+
+ private static TrinoParquetFileReader createReader()
+ throws Exception
+ {
+ File parquetFile = new
File(Resources.getResource(ARCHIVED_TIMELINE_PARQUET_FILE).toURI());
+ return new TrinoParquetFileReader(localStorage(), new
StoragePath(parquetFile.toURI().toString()));
+ }
+
+ private static HoodieStorage localStorage()
+ {
+ return new HudiTrinoStorage(new LocalFileSystem(Paths.get("/")), new
TrinoStorageConfiguration());
+ }
+}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/ResourceHudiTablesInitializer.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/ResourceHudiTablesInitializer.java
index f430c7b556c2..821c0dae72ad 100644
---
a/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/ResourceHudiTablesInitializer.java
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/ResourceHudiTablesInitializer.java
@@ -359,6 +359,7 @@ public class ResourceHudiTablesInitializer
HUDI_TIMESTAMP_KEYGEN_PT_SCALAR_TO_YYYY_MM_DD_HH_V8_MOR(hudiTimestampKeygenColumns(),
hudiTimestampKeygenPartitionColumns(),
hudiTimestampKeygenPartitions("SCALAR"), true),
HUDI_CUSTOM_KEYGEN_PT_V8_MOR(hudiCustomKeyGenColumns(),
hudiCustomKeyGenPartitionColumns(), hudiCustomKeyGenPartitions(), false),
HUDI_NON_EXTRACTABLE_PARTITION_PATH(multiPartitionRegularColumns(),
multiPartitionColumns(), multiPartitionsWithNonExtractablePartitionPaths(),
false),
+ HUDI_MOR_ARCHIVED_TIMELINE(hudiMultiFgRegularColumns(),
ImmutableList.of(), ImmutableMap.of(), true),
/**/;
private static final List<Column> HUDI_META_COLUMNS = ImmutableList.of(
diff --git a/hudi-trino/src/test/resources/archived_timeline.parquet
b/hudi-trino/src/test/resources/archived_timeline.parquet
new file mode 100644
index 000000000000..c0c888d43fb1
Binary files /dev/null and
b/hudi-trino/src/test/resources/archived_timeline.parquet differ
diff --git
a/hudi-trino/src/test/resources/hudi-testing-data/hudi_cow_archived_timeline.md
b/hudi-trino/src/test/resources/hudi-testing-data/hudi_cow_archived_timeline.md
new file mode 100644
index 000000000000..01e71806eaaa
--- /dev/null
+++
b/hudi-trino/src/test/resources/hudi-testing-data/hudi_cow_archived_timeline.md
@@ -0,0 +1,124 @@
+<!--
+ 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.
+-->
+
+## Create script
+
+Structure of table:
+- COW table in table version 8 with an archived timeline
+- Using Hudi 1.0.2
+- Non-partitioned table
+
+The table itself is not checked in. Its first LSM history file,
+`.hoodie/timeline/history/20250918121953134_20250918122001506_0.parquet` (four
commit instants, 20250918121953134
+through 20250918122001506), is checked in as
`src/test/resources/archived_timeline.parquet` for
+`TestTrinoParquetFileReader`.
+
+```scala
+package org.apache.spark.sql.hudi.timeline
+
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration
+import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
+
+import java.io.File
+
+class TestCompactedTimelineTable extends HoodieSparkSqlTestBase {
+
+ test("Test COW Table with Compacted LSM Timeline") {
+ withRecordType()(withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+
+ // Create COW table with aggressive timeline archival settings
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'cow',
+ | preCombineField = 'ts',
+ | 'hoodie.keep.min.commits' = '3',
+ | 'hoodie.keep.max.commits' = '5',
+ | 'hoodie.cleaner.commits.retained' = '2',
+ | 'hoodie.archive.automatic' = 'true'
+ | )
+ """.stripMargin)
+
+ // Generate initial commits
+ spark.sql(s"insert into $tableName values(1, 'alice', 100.0,
1000)")
+ spark.sql(s"insert into $tableName values(2, 'bob', 200.0, 2000)")
+ spark.sql(s"insert into $tableName values(3, 'charlie', 300.0,
3000)")
+
+ // Update operations to create more timeline entries
+ spark.sql(s"update $tableName set price = 110.0 where id = 1")
+ spark.sql(s"update $tableName set name = 'robert' where id = 2")
+
+ // More commits to trigger archival
+ spark.sql(s"insert into $tableName values(4, 'david', 400.0,
4000)")
+ spark.sql(s"insert into $tableName values(5, 'eve', 500.0, 5000)")
+
+ // Delete operation
+ spark.sql(s"delete from $tableName where id = 3")
+
+ // Additional commits to exceed max commits threshold and trigger
archival
+ spark.sql(s"insert into $tableName values(6, 'frank', 600.0,
6000)")
+ spark.sql(s"update $tableName set price = price * 1.1 where id >
4")
+ spark.sql(s"insert into $tableName values(7, 'grace', 700.0,
7000)")
+
+ // Verify data correctness after all operations
+ checkAnswer(s"select id, name, price, ts from $tableName order by
id")(
+ Seq(1, "alice", 110.0, 1000),
+ Seq(2, "robert", 200.0, 2000),
+ Seq(4, "david", 400.0, 4000),
+ Seq(5, "eve", 550.0, 5000),
+ Seq(6, "frank", 660.0, 6000),
+ Seq(7, "grace", 700.0, 7000)
+ )
+
+ // Verify timeline archival occurred
+ val metaClient = HoodieTableMetaClient.builder()
+ .setConf(new
HadoopStorageConfiguration(spark.sparkContext.hadoopConfiguration))
+ .setBasePath(tablePath)
+ .build()
+
+ val timeline = metaClient.getActiveTimeline
+ val archivedTimeline = metaClient.getArchivedTimeline
+
+ // Check that archived timeline exists and has entries
+ assertResult(true)(archivedTimeline.reload().countInstants() > 0)
+
+ // Verify archived timeline files exist in the
.hoodie/timeline/history directory
+ val archivedDir = new File(tablePath, ".hoodie/timeline/history")
+ assertResult(true)(archivedDir.exists() &&
archivedDir.listFiles().nonEmpty)
+
+ // Check that archived files are parquet format
+ val archivedFiles =
archivedDir.listFiles().filter(_.getName.endsWith(".parquet"))
+ assertResult(true)(archivedFiles.nonEmpty)
+
+ println(s"Active timeline instants: ${timeline.countInstants()}")
+ println(s"Archived timeline instants:
${archivedTimeline.reload().countInstants()}")
+ println(s"Archived files:
${archivedFiles.map(_.getName).mkString(", ")}")
+ })
+ }
+}
+```
diff --git
a/hudi-trino/src/test/resources/hudi-testing-data/hudi_mor_archived_timeline.md
b/hudi-trino/src/test/resources/hudi-testing-data/hudi_mor_archived_timeline.md
new file mode 100644
index 000000000000..cf22eaaaab62
--- /dev/null
+++
b/hudi-trino/src/test/resources/hudi-testing-data/hudi_mor_archived_timeline.md
@@ -0,0 +1,96 @@
+<!--
+ 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.
+-->
+
+## Create script
+
+Structure of table:
+- MOR table in table version 8 with an archived timeline
+- Using Hudi 1.0.2
+- Non-partitioned table
+
+```scala
+package org.apache.spark.sql.hudi.timeline
+
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration
+import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
+
+import java.io.File
+
+class TestCompactedTimelineTable extends HoodieSparkSqlTestBase {
+
+ test("Test MOR Table with Compacted LSM Timeline") {
+ withRecordType()(withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath
+
+ // Create MOR table with timeline archival settings
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'mor',
+ | preCombineField = 'ts',
+ | 'hoodie.keep.min.commits' = '2',
+ | 'hoodie.keep.max.commits' = '4',
+ | 'hoodie.cleaner.commits.retained' = '1',
+ | 'hoodie.archive.automatic' = 'true',
+ | 'hoodie.compact.inline' = 'false'
+ | )
+ """.stripMargin)
+
+ // Generate commits and updates to create both base and log files
+ spark.sql(s"insert into $tableName values(1, 'user1', 10.0, 1000)")
+ spark.sql(s"insert into $tableName values(2, 'user2', 20.0, 2000)")
+ spark.sql(s"update $tableName set price = 15.0 where id = 1")
+ spark.sql(s"insert into $tableName values(3, 'user3', 30.0, 3000)")
+ spark.sql(s"update $tableName set name = 'updated_user2' where id
= 2")
+
+ // More operations to trigger archival
+ spark.sql(s"insert into $tableName values(4, 'user4', 40.0, 4000)")
+ spark.sql(s"delete from $tableName where id = 1")
+ spark.sql(s"insert into $tableName values(5, 'user5', 50.0, 5000)")
+
+ // Verify final data state
+ checkAnswer(s"select id, name, price, ts from $tableName order by
id")(
+ Seq(2, "updated_user2", 20.0, 2000),
+ Seq(3, "user3", 30.0, 3000),
+ Seq(4, "user4", 40.0, 4000),
+ Seq(5, "user5", 50.0, 5000)
+ )
+
+ // Check archived timeline creation
+ val metaClient = HoodieTableMetaClient.builder()
+ .setConf(new
HadoopStorageConfiguration(spark.sparkContext.hadoopConfiguration))
+ .setBasePath(tablePath)
+ .build()
+
+ val archivedTimeline = metaClient.getArchivedTimeline
+ assertResult(true)(archivedTimeline.reload().countInstants() > 0)
+
+ println(s"MOR Archived timeline instants:
${archivedTimeline.reload().countInstants()}")
+ })
+ }
+}
+```
diff --git
a/hudi-trino/src/test/resources/hudi-testing-data/hudi_mor_archived_timeline.zip
b/hudi-trino/src/test/resources/hudi-testing-data/hudi_mor_archived_timeline.zip
new file mode 100644
index 000000000000..3fa27a36ca69
Binary files /dev/null and
b/hudi-trino/src/test/resources/hudi-testing-data/hudi_mor_archived_timeline.zip
differ
