deniskuzZ commented on code in PR #6793: URL: https://github.com/apache/hive/pull/6793#discussion_r4030125485
########## llap-server/src/java/org/apache/hadoop/hive/llap/io/decode/ParquetEncodedDataConsumer.java: ########## @@ -0,0 +1,225 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hive.llap.io.decode; + +import java.io.IOException; +import java.time.ZoneId; +import java.util.List; +import java.util.Map; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.common.type.DataTypePhysicalVariation; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.conf.HiveConf.ConfVars; +import org.apache.hadoop.hive.llap.counters.LlapIOCounters; +import org.apache.hadoop.hive.llap.counters.QueryFragmentCounters; +import org.apache.hadoop.hive.llap.io.api.impl.ColumnVectorBatch; +import org.apache.hadoop.hive.llap.io.api.impl.LlapIoImpl; +import org.apache.hadoop.hive.llap.io.decode.ColumnVectorProducer.Includes; +import org.apache.hadoop.hive.llap.io.encoded.ParquetEncodedColumnBatch; +import org.apache.hadoop.hive.llap.metrics.LlapDaemonIOMetrics; +import org.apache.hadoop.hive.ql.exec.vector.ColumnVector; +import org.apache.hadoop.hive.ql.exec.vector.VectorizedBatchUtil; +import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch; +import org.apache.hadoop.hive.ql.io.IOConstants; +import org.apache.hadoop.hive.ql.io.orc.encoded.Consumer; +import org.apache.hadoop.hive.ql.io.parquet.read.DataWritableReadSupport; +import org.apache.hadoop.hive.ql.io.parquet.vector.ParquetRowGroupDecoder; +import org.apache.hadoop.hive.ql.io.parquet.vector.VectorizedColumnReader; +import org.apache.hadoop.hive.serde2.ColumnProjectionUtils; +import org.apache.hadoop.hive.serde2.typeinfo.DecimalTypeInfo; +import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo; +import org.apache.orc.TypeDescription; +import org.apache.orc.impl.SchemaEvolution; +import org.apache.parquet.HadoopReadOptions; +import org.apache.parquet.ParquetReadOptions; +import org.apache.parquet.column.page.PageReadStore; +import org.apache.parquet.compression.CompressionCodecFactory; +import org.apache.parquet.format.converter.ParquetMetadataConverter; +import org.apache.parquet.hadoop.metadata.ParquetMetadata; +import org.apache.parquet.schema.MessageType; + +import com.google.common.base.Strings; + +/** + * The Parquet counterpart of {@link OrcEncodedDataConsumer}: it turns one row group's worth of + * cached column-chunk buffers ({@link ParquetEncodedColumnBatch}) into {@link ColumnVectorBatch}es + * and hands them downstream. + * + * <p>Rather than reimplementing Parquet value decoding, this consumer builds a + * {@link ParquetCachedPageReadStore} directly over the cached buffers (pages are parsed in place and + * decompressed lazily, no ParquetFileReader and no whole-chunk copy) and drives the very same + * {@link VectorizedColumnReader}s the non-cached vectorized reader uses (via {@link ParquetRowGroupDecoder}). + */ +public class ParquetEncodedDataConsumer + extends EncodedDataConsumer<Object, ParquetEncodedColumnBatch> { + + private final Configuration jobConf; + private final boolean useDecimal64ColumnVectors; + private final CompressionCodecFactory codecFactory; + private final ParquetMetadataConverter converter; + private ParquetMetadata footer; + private MessageType requestedSchema; + private Path path; + private Map<String, Object> initialDefaults; + + // Derived lazily (once) from the footer + job conf, then reused across row groups. + private List<TypeInfo> columnTypesList; + private List<Integer> colsToInclude; + private boolean readAllColumns; + private boolean skipTimestampConversion; + private boolean skipProlepticConversion; + private boolean legacyConversionEnabled; + private ZoneId writerTimezone; + private boolean schemaInitialized = false; + + public ParquetEncodedDataConsumer(Consumer<ColumnVectorBatch> consumer, Includes includes, + QueryFragmentCounters counters, LlapDaemonIOMetrics ioMetrics, Configuration jobConf) { + super(consumer, includes.getPhysicalColumnIds().size(), ioMetrics, counters); + this.jobConf = jobConf; + this.useDecimal64ColumnVectors = HiveConf.getVar(jobConf, + ConfVars.HIVE_VECTORIZED_INPUT_FORMAT_SUPPORTS_ENABLED).equalsIgnoreCase("decimal_64"); Review Comment: do we have constant for "decimal_64" ? -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
