wombatu-kun commented on code in PR #19732: URL: https://github.com/apache/hudi/pull/19732#discussion_r3862571368
########## hudi-trino/src/main/java/io/trino/plugin/hudi/io/TrinoParquetFileReader.java: ########## @@ -0,0 +1,460 @@ +/* + * 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.metastore.HiveType; +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.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 io.trino.spi.type.Type; +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 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.hive.util.HiveTypeTranslator.toHiveType; +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.BigintType.BIGINT; +import static io.trino.spi.type.BooleanType.BOOLEAN; +import static io.trino.spi.type.DoubleType.DOUBLE; +import static io.trino.spi.type.IntegerType.INTEGER; +import static io.trino.spi.type.RealType.REAL; +import static io.trino.spi.type.VarbinaryType.VARBINARY; +import static io.trino.spi.type.VarcharType.VARCHAR; +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 Schema avroSchema; + 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) + throws IOException + { + 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)); + this.fileLength = inputFile.length(); + this.parquetMetadata = readParquetMetadata(); + this.avroSchema = extractAvroSchema(parquetMetadata.getFileMetaData()); + this.hoodieSchema = HoodieSchema.fromAvroSchema(avroSchema); + this.totalRecords = parquetMetadata.getBlocks().stream() + .mapToLong(BlockMetadata::rowCount) + .sum(); + } + + @Override + public ClosableIterator<IndexedRecord> getIndexedRecordIterator(HoodieSchema readerSchema, HoodieSchema requestedSchema, Map<String, String> renamedColumns) + throws IOException + { + // Timeline files are never schema-evolved, so no column can have been renamed under them + Schema projectedSchema = requestedSchema != null ? requestedSchema.toAvroSchema() : avroSchema; + 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 + synchronized (openIterators) { + for (ParquetIndexedRecordIterator iterator : openIterators) { Review Comment: close() stops at the first iterator whose close() throws, so the rest stay open and openIterators is never cleared, though the comment above promises every iterator is released. Close them all and attach the later failures with addSuppressed, as HudiPageSource's constructor does - nit, not a blocker. ########## hudi-trino/src/main/java/io/trino/plugin/hudi/io/TrinoParquetFileReader.java: ########## @@ -0,0 +1,460 @@ +/* + * 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.metastore.HiveType; +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.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 io.trino.spi.type.Type; +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 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.hive.util.HiveTypeTranslator.toHiveType; +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.BigintType.BIGINT; +import static io.trino.spi.type.BooleanType.BOOLEAN; +import static io.trino.spi.type.DoubleType.DOUBLE; +import static io.trino.spi.type.IntegerType.INTEGER; +import static io.trino.spi.type.RealType.REAL; +import static io.trino.spi.type.VarbinaryType.VARBINARY; +import static io.trino.spi.type.VarcharType.VARCHAR; +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 Schema avroSchema; + 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) + throws IOException + { + 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)); + this.fileLength = inputFile.length(); + this.parquetMetadata = readParquetMetadata(); + this.avroSchema = extractAvroSchema(parquetMetadata.getFileMetaData()); + this.hoodieSchema = HoodieSchema.fromAvroSchema(avroSchema); + this.totalRecords = parquetMetadata.getBlocks().stream() + .mapToLong(BlockMetadata::rowCount) + .sum(); + } + + @Override + public ClosableIterator<IndexedRecord> getIndexedRecordIterator(HoodieSchema readerSchema, HoodieSchema requestedSchema, Map<String, String> renamedColumns) + throws IOException + { + // Timeline files are never schema-evolved, so no column can have been renamed under them + Schema projectedSchema = requestedSchema != null ? requestedSchema.toAvroSchema() : avroSchema; + 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 + synchronized (openIterators) { + for (ParquetIndexedRecordIterator iterator : openIterators) { + iterator.close(); + } + openIterators.clear(); + } + } + + 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()); Review Comment: MetadataReader.readFooter throws ParquetCorruptionException, an IOException, so it escapes raw instead of reaching handleException, and ArchivedTimelineLoaderV2's catch then rebuilds it without the cause. Route the constructor's open path and createParquetReader's catch through handleException the way hasNext and close already do. ########## hudi-trino/src/main/java/io/trino/plugin/hudi/io/TrinoParquetFileReader.java: ########## @@ -0,0 +1,460 @@ +/* + * 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.metastore.HiveType; +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.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 io.trino.spi.type.Type; +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 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.hive.util.HiveTypeTranslator.toHiveType; +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.BigintType.BIGINT; +import static io.trino.spi.type.BooleanType.BOOLEAN; +import static io.trino.spi.type.DoubleType.DOUBLE; +import static io.trino.spi.type.IntegerType.INTEGER; +import static io.trino.spi.type.RealType.REAL; +import static io.trino.spi.type.VarbinaryType.VARBINARY; +import static io.trino.spi.type.VarcharType.VARCHAR; +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 Schema avroSchema; + 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) + throws IOException + { + 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)); + this.fileLength = inputFile.length(); + this.parquetMetadata = readParquetMetadata(); + this.avroSchema = extractAvroSchema(parquetMetadata.getFileMetaData()); + this.hoodieSchema = HoodieSchema.fromAvroSchema(avroSchema); + this.totalRecords = parquetMetadata.getBlocks().stream() + .mapToLong(BlockMetadata::rowCount) + .sum(); + } + + @Override + public ClosableIterator<IndexedRecord> getIndexedRecordIterator(HoodieSchema readerSchema, HoodieSchema requestedSchema, Map<String, String> renamedColumns) + throws IOException + { + // Timeline files are never schema-evolved, so no column can have been renamed under them + Schema projectedSchema = requestedSchema != null ? requestedSchema.toAvroSchema() : avroSchema; + 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 + synchronized (openIterators) { + for (ParquetIndexedRecordIterator iterator : openIterators) { + iterator.close(); + } + openIterators.clear(); + } + } + + 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); + } + + private static List<Column> buildTrinoColumns(Schema projectedSchema, MessageColumnIO messageColumnIO) + { + ImmutableList.Builder<Column> columnsBuilder = ImmutableList.builder(); + for (Schema.Field field : projectedSchema.getFields()) { + Type trinoType = avroTypeToTrinoType(field.schema()); + Field parquetField = constructField(trinoType, lookupColumnByName(messageColumnIO, field.name())) + .orElseThrow(() -> new TrinoException(HUDI_SCHEMA_ERROR, "Could not find column: " + field.name())); + columnsBuilder.add(new Column(field.name(), parquetField)); + } + return columnsBuilder.build(); + } + + private static List<HiveColumnHandle> buildColumnHandles(Schema projectedSchema) + { + List<HiveColumnHandle> columnHandles = new ArrayList<>(); + List<Schema.Field> fields = projectedSchema.getFields(); + for (int i = 0; i < fields.size(); i++) { + Schema.Field field = fields.get(i); + Type trinoType = avroTypeToTrinoType(field.schema()); + HiveType hiveType = toHiveType(trinoType); + columnHandles.add(new HiveColumnHandle( + field.name(), + i, + hiveType, + trinoType, + Optional.empty(), + HiveColumnHandle.ColumnType.REGULAR, + Optional.empty())); + } + return columnHandles; + } + + private static Type avroTypeToTrinoType(Schema fieldSchema) Review Comment: avroTypeToTrinoType and the body of buildColumnHandles duplicate HudiUtil.toColumnHandle, which maps a HoodieSchemaField through NativeLogicalTypesAvroTypeBlockHandler and is what HudiTrinoReaderContext already uses for the same job. Pass the HoodieSchema into the iterator alongside the Avro one, build the handles with toColumnHandle, and take the VARBINARY positions off those handles. ########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java: ########## @@ -1454,6 +1478,13 @@ private static Stream<Arguments> comprehensiveTestParameters() .map(boolValue -> Arguments.of(table, boolValue))); } + private static Stream<Arguments> archivedTimelineTestParameters() Review Comment: archivedTimelineTestParameters only ever emits HUDI_MOR_ARCHIVED_TIMELINE twice, so the table parameter carries a constant; testHudiTimestampKeygenEpochMillisPartitionedTables covers the same RO/RT pair with @ValueSource(booleans = {true, false}). Drop the provider and name the table in the test body. ########## 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 .hoodie/archived directory Review Comment: The comment says .hoodie/archived but the line under it reads .hoodie/timeline/history, which is where a table-version-8 table keeps its LSM timeline. Fix the comment to name the directory the code checks. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
