voonhous commented on code in PR #19732: URL: https://github.com/apache/hudi/pull/19732#discussion_r3863046790
########## 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: Done in 5da2dc0: close() now closes every iterator, clears the list, then rethrows the first failure with the later ones attached via addSuppressed. ########## 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: Done in 5da2dc0: the constructor's open path (length and footer read) and createParquetReader, data-source open included, throw through handleException; the constructor no longer declares IOException and the factory's catch that wrapped it is gone. `testCorruptFileFailsWithBadData` builds the reader over a three-byte file and expects HUDI_BAD_DATA with the ParquetCorruptionException as its cause. One correction to the chain: the factory's catch was attaching the cause, as a HoodieIOException the loader's catch never saw, but the error code was lost either way. ########## 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: Done in 5da2dc0: the iterator takes the HoodieSchema projection, builds its handles with `HudiUtil.toColumnHandle`, and the Trino columns and VARBINARY positions come off those handles. `avroTypeToTrinoType` and the handle loop are gone. ########## 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: Done in 5da2dc0: `@ValueSource(booleans = {true, false})`, the table is named in the body, the provider is gone. ########## 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: Done in 5da2dc0: the comment names `.hoodie/timeline/history`. ########## 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) + { + // Handle Avro's nullable fields, which are represented as a UNION of null and a type + if (fieldSchema.isUnion()) { + List<Schema> nonNullSchemas = fieldSchema.getTypes().stream() + .filter(schema -> schema.getType() != Schema.Type.NULL) + .toList(); + // A union of multiple non-null types is not supported + if (nonNullSchemas.size() != 1) { + throw new UnsupportedOperationException("Unsupported Avro union type: " + fieldSchema); + } + fieldSchema = nonNullSchemas.getFirst(); + } + + return switch (fieldSchema.getType()) { + case STRING -> VARCHAR; + case INT -> INTEGER; + case LONG -> BIGINT; + case FLOAT -> REAL; + case DOUBLE -> DOUBLE; + case BOOLEAN -> BOOLEAN; + case BYTES -> VARBINARY; + // Be explicit about unhandled types instead of a silent fallback to prevent subtle bugs if the schema contains + // types like MAP, ARRAY, FIXED, etc + default -> throw new UnsupportedOperationException("Unsupported Avro type: " + fieldSchema.getType()); + }; + } + + /** + * Positions of the {@code bytes} fields of the record, 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(Schema projectedSchema) + { + return projectedSchema.getFields().stream() + .filter(field -> avroTypeToTrinoType(field.schema()).equals(VARBINARY)) + .mapToInt(Schema.Field::pos) + .toArray(); + } + + private static TrinoException handleException(StoragePath path, Exception exception) + { Review Comment: Keeping the name: it mirrors Trino's `ParquetPageSource.handleException`, which has the same signature and body and is handed to `ParquetReader` in the same exception-transform slot, so it reads the same as the Hive connector code it sits next to. -- 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]
