| /* |
| * 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.drill.exec.store.parquet2; |
| |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Optional; |
| import java.util.function.BiFunction; |
| import java.util.function.Function; |
| import java.util.function.Supplier; |
| |
| import io.netty.buffer.DrillBuf; |
| import org.apache.drill.common.exceptions.DrillRuntimeException; |
| import org.apache.drill.common.expression.PathSegment; |
| import org.apache.drill.common.expression.SchemaPath; |
| import org.apache.drill.exec.ExecConstants; |
| import org.apache.drill.exec.expr.holders.BigIntHolder; |
| import org.apache.drill.exec.expr.holders.BitHolder; |
| import org.apache.drill.exec.expr.holders.DateHolder; |
| import org.apache.drill.exec.expr.holders.Float4Holder; |
| import org.apache.drill.exec.expr.holders.Float8Holder; |
| import org.apache.drill.exec.expr.holders.IntHolder; |
| import org.apache.drill.exec.expr.holders.IntervalHolder; |
| import org.apache.drill.exec.expr.holders.TimeHolder; |
| import org.apache.drill.exec.expr.holders.TimeStampHolder; |
| import org.apache.drill.exec.expr.holders.VarBinaryHolder; |
| import org.apache.drill.exec.expr.holders.VarCharHolder; |
| import org.apache.drill.exec.expr.holders.VarDecimalHolder; |
| import org.apache.drill.exec.physical.impl.OutputMutator; |
| import org.apache.drill.exec.server.options.OptionManager; |
| import org.apache.drill.exec.store.parquet.ParquetReaderUtility; |
| import org.apache.drill.exec.store.parquet.columnreaders.ParquetColumnMetadata; |
| import org.apache.drill.exec.vector.complex.impl.AbstractRepeatedMapWriter; |
| import org.apache.drill.exec.vector.complex.impl.SingleMapWriter; |
| import org.apache.drill.exec.vector.complex.writer.BaseWriter; |
| import org.apache.drill.exec.vector.complex.writer.BaseWriter.ListWriter; |
| import org.apache.drill.exec.vector.complex.writer.BaseWriter.MapWriter; |
| import org.apache.drill.exec.vector.complex.writer.BaseWriter.DictWriter; |
| import org.apache.drill.exec.vector.complex.writer.BigIntWriter; |
| import org.apache.drill.exec.vector.complex.writer.BitWriter; |
| import org.apache.drill.exec.vector.complex.writer.DateWriter; |
| import org.apache.drill.exec.vector.complex.writer.Float4Writer; |
| import org.apache.drill.exec.vector.complex.writer.Float8Writer; |
| import org.apache.drill.exec.vector.complex.writer.IntWriter; |
| import org.apache.drill.exec.vector.complex.writer.IntervalWriter; |
| import org.apache.drill.exec.vector.complex.writer.TimeStampWriter; |
| import org.apache.drill.exec.vector.complex.writer.TimeWriter; |
| import org.apache.drill.exec.vector.complex.writer.VarBinaryWriter; |
| import org.apache.drill.exec.vector.complex.writer.VarCharWriter; |
| import org.apache.drill.exec.vector.complex.writer.VarDecimalWriter; |
| import com.google.common.primitives.Ints; |
| import com.google.common.primitives.Longs; |
| import org.apache.parquet.io.api.Binary; |
| import org.apache.parquet.io.api.Converter; |
| import org.apache.parquet.io.api.GroupConverter; |
| import org.apache.parquet.io.api.PrimitiveConverter; |
| import org.apache.parquet.schema.GroupType; |
| import org.apache.parquet.schema.LogicalTypeAnnotation; |
| import org.apache.parquet.schema.PrimitiveType; |
| import org.apache.parquet.schema.Type; |
| import org.apache.parquet.schema.Type.Repetition; |
| import org.joda.time.DateTimeConstants; |
| |
| import static org.apache.drill.common.expression.SchemaPath.DYNAMIC_STAR; |
| import static org.apache.drill.exec.store.parquet.ParquetReaderUtility.NanoTimeUtils.getDateTimeValueFromBinary; |
| |
| public class DrillParquetGroupConverter extends GroupConverter { |
| |
| protected final List<Converter> converters; |
| |
| private final BaseWriter baseWriter; |
| private final OutputMutator mutator; |
| private final OptionManager options; |
| // See DRILL-4203 |
| private final ParquetReaderUtility.DateCorruptionStatus containsCorruptedDates; |
| |
| /** |
| * Debugging information in form of "parent">fieldName[WriterClassName-hashCode()], |
| * where "parent" is parent converterName. |
| */ |
| private String converterName; |
| |
| /** |
| * Constructor is responsible for creation of converter without creation of child converters. |
| * |
| * @param mutator output mutator, used to share managed buffer with primitive converters |
| * @param baseWriter map or list writer associated with the group converter |
| * @param options option manager used to check enabled option when necessary |
| * @param containsCorruptedDates allows to select strategy for dates handling |
| */ |
| protected DrillParquetGroupConverter(OutputMutator mutator, BaseWriter baseWriter, OptionManager options, |
| ParquetReaderUtility.DateCorruptionStatus containsCorruptedDates) { |
| this.mutator = mutator; |
| this.baseWriter = baseWriter; |
| this.options = options; |
| this.containsCorruptedDates = containsCorruptedDates; |
| converters = new ArrayList<>(); |
| } |
| |
| /** |
| * The constructor is responsible for creation of converters tree and may invoke itself for |
| * creation of child converters when nested field is group type field too. Assumed that ordering of |
| * fields from schema parameter matches ordering of paths in columns list. Though columns may have fields |
| * which aren't present in schema. |
| * |
| * @param mutator output mutator, used to share managed buffer with primitive converters |
| * @param baseWriter map or list writer associated with the group converter |
| * @param schema group type of the converter |
| * @param columns columns to project |
| * @param options option manager used to check enabled option when necessary |
| * @param containsCorruptedDates allows to select strategy for dates handling |
| * @param skipRepeated true only if parent field in schema detected as list and current schema is repeated group type |
| * @param parentName name of group converter which called the constructor |
| */ |
| public DrillParquetGroupConverter(OutputMutator mutator, BaseWriter baseWriter, GroupType schema, |
| Collection<SchemaPath> columns, OptionManager options, |
| ParquetReaderUtility.DateCorruptionStatus containsCorruptedDates, |
| boolean skipRepeated, String parentName) { |
| this(mutator, baseWriter, options, containsCorruptedDates); |
| this.converterName = String.format("%s>%s[%s-%d]", parentName, schema.getName(), baseWriter.getClass().getSimpleName(), baseWriter.hashCode()); |
| |
| Iterator<SchemaPath> colIterator = columns.iterator(); |
| |
| for (final Type type : schema.getFields()) { |
| |
| // Match the name of the field in the schema definition to the name of the field in the query. |
| String name = type.getName(); |
| PathSegment colNextChild = null; |
| while (colIterator.hasNext()) { |
| PathSegment colPath = colIterator.next().getRootSegment(); |
| String colPathName; |
| if (colPath.isNamed() && !DYNAMIC_STAR.equals(colPathName = colPath.getNameSegment().getPath()) && colPathName.equalsIgnoreCase(name)) { |
| name = colPathName; |
| colNextChild = colPath.getChild(); |
| break; |
| } |
| } |
| |
| Converter converter = createFieldConverter(skipRepeated, type, name, colNextChild); |
| converters.add(converter); |
| } |
| } |
| |
| private Converter createFieldConverter(boolean skipRepeated, Type fieldType, String name, PathSegment colNextChild) { |
| Converter converter; |
| if (fieldType.isPrimitive()) { |
| converter = getConverterForType(name, fieldType.asPrimitiveType()); |
| } else { |
| while (colNextChild != null && !colNextChild.isNamed()) { |
| colNextChild = colNextChild.getChild(); |
| } |
| |
| Collection<SchemaPath> columns = colNextChild == null |
| ? Collections.emptyList() |
| : Collections.singletonList(new SchemaPath(colNextChild.getNameSegment())); |
| |
| BaseWriter writer; |
| GroupType fieldGroupType = fieldType.asGroupType(); |
| if (ParquetReaderUtility.isLogicalListType(fieldGroupType)) { |
| writer = getWriter(name, MapWriter::list, ListWriter::list); |
| converter = new DrillParquetGroupConverter(mutator, writer, fieldGroupType, columns, options, |
| containsCorruptedDates, true, converterName); |
| } else if (options.getOption(ExecConstants.PARQUET_READER_ENABLE_MAP_SUPPORT_VALIDATOR) |
| && ParquetReaderUtility.isLogicalMapType(fieldGroupType)) { |
| writer = getWriter(name, MapWriter::dict, ListWriter::dict); |
| converter = new DrillParquetMapGroupConverter( |
| mutator, (DictWriter) writer, fieldGroupType, options, containsCorruptedDates); |
| } else if (fieldType.isRepetition(Repetition.REPEATED)) { |
| if (skipRepeated) { |
| converter = new DrillIntermediateParquetGroupConverter(mutator, baseWriter, fieldGroupType, columns, options, |
| containsCorruptedDates, false, converterName); |
| } else { |
| writer = getWriter(name, (m, s) -> m.list(s).map(), l -> l.list().map()); |
| converter = new DrillParquetGroupConverter(mutator, writer, fieldGroupType, columns, options, |
| containsCorruptedDates, false, converterName); |
| } |
| } else { |
| writer = getWriter(name, MapWriter::map, ListWriter::map); |
| converter = new DrillParquetGroupConverter(mutator, writer, fieldGroupType, columns, options, |
| containsCorruptedDates, false, converterName); |
| } |
| |
| } |
| return converter; |
| } |
| |
| protected PrimitiveConverter getConverterForType(String name, PrimitiveType type) { |
| switch(type.getPrimitiveTypeName()) { |
| case INT32: { |
| if (type.getOriginalType() == null) { |
| return getIntConverter(name, type); |
| } |
| switch(type.getOriginalType()) { |
| case UINT_8 : |
| case UINT_16: |
| case UINT_32: |
| case INT_8 : |
| case INT_16 : |
| case INT_32 : { |
| return getIntConverter(name, type); |
| } |
| case DECIMAL: { |
| ParquetReaderUtility.checkDecimalTypeEnabled(options); |
| return getVarDecimalConverter(name, type); |
| } |
| case DATE: { |
| DateWriter writer = type.isRepetition(Repetition.REPEATED) |
| ? getWriter(name, (m, f) -> m.list(f).date(), l -> l.list().date()) |
| : getWriter(name, (m, f) -> m.date(f), l -> l.date()); |
| |
| switch(containsCorruptedDates) { |
| case META_SHOWS_CORRUPTION: |
| return new DrillCorruptedDateConverter(writer); |
| case META_SHOWS_NO_CORRUPTION: |
| return new DrillDateConverter(writer); |
| case META_UNCLEAR_TEST_VALUES: |
| return new CorruptionDetectingDateConverter(writer); |
| default: |
| throw new DrillRuntimeException( |
| String.format("Issue setting up parquet reader for date type, " + |
| "unrecognized date corruption status %s. See DRILL-4203 for more info.", |
| containsCorruptedDates)); |
| } |
| } |
| case TIME_MILLIS: { |
| TimeWriter writer = type.isRepetition(Repetition.REPEATED) |
| ? getWriter(name, (m, f) -> m.list(f).time(), l -> l.list().time()) |
| : getWriter(name, (m, f) -> m.time(f), l -> l.time()); |
| return new DrillTimeConverter(writer); |
| } |
| default: { |
| throw new UnsupportedOperationException("Unsupported type: " + type.getOriginalType()); |
| } |
| } |
| } |
| case INT64: { |
| if (type.getOriginalType() == null) { |
| return getBigIntConverter(name, type); |
| } |
| switch(type.getOriginalType()) { |
| case UINT_64: |
| case INT_64: |
| return getBigIntConverter(name, type); |
| case TIMESTAMP_MICROS: { |
| TimeStampWriter writer = getTimeStampWriter(name, type); |
| return new DrillTimeStampMicrosConverter(writer); |
| } |
| case TIME_MICROS: { |
| TimeWriter writer = type.isRepetition(Repetition.REPEATED) |
| ? getWriter(name, (m, f) -> m.list(f).time(), l -> l.list().time()) |
| : getWriter(name, MapWriter::time, ListWriter::time); |
| return new DrillTimeMicrosConverter(writer); |
| } |
| case DECIMAL: { |
| ParquetReaderUtility.checkDecimalTypeEnabled(options); |
| return getVarDecimalConverter(name, type); |
| } |
| case TIMESTAMP_MILLIS: { |
| TimeStampWriter writer = getTimeStampWriter(name, type); |
| return new DrillTimeStampConverter(writer); |
| } |
| default: { |
| throw new UnsupportedOperationException("Unsupported type " + type.getOriginalType()); |
| } |
| } |
| } |
| case INT96: { |
| // TODO: replace null with TIMESTAMP_NANOS once parquet support such type annotation. |
| if (type.getOriginalType() == null) { |
| if (options.getOption(ExecConstants.PARQUET_READER_INT96_AS_TIMESTAMP).bool_val) { |
| TimeStampWriter writer = getTimeStampWriter(name, type); |
| return new DrillFixedBinaryToTimeStampConverter(writer); |
| } else { |
| VarBinaryWriter writer = type.isRepetition(Repetition.REPEATED) |
| ? getWriter(name, (m, f) -> m.list(f).varBinary(), l -> l.list().varBinary()) |
| : getWriter(name, (m, f) -> m.varBinary(f), listWriter -> listWriter.varBinary()); |
| return new DrillFixedBinaryToVarbinaryConverter(writer, ParquetColumnMetadata.getTypeLengthInBits(type.getPrimitiveTypeName()) / 8, mutator.getManagedBuffer()); |
| } |
| } |
| |
| } |
| case FLOAT: { |
| Float4Writer writer = type.isRepetition(Repetition.REPEATED) |
| ? getWriter(name, (m, f) -> m.list(f).float4(), l -> l.list().float4()) |
| : getWriter(name, (m, f) -> m.float4(f), l -> l.float4()); |
| return new DrillFloat4Converter(writer); |
| } |
| case DOUBLE: { |
| Float8Writer writer = type.isRepetition(Repetition.REPEATED) |
| ? getWriter(name, (m, f) -> m.list(f).float8(), l -> l.list().float8()) |
| : getWriter(name, (m, f) -> m.float8(f), l -> l.float8()); |
| return new DrillFloat8Converter(writer); |
| } |
| case BOOLEAN: { |
| BitWriter writer = type.isRepetition(Repetition.REPEATED) |
| ? getWriter(name, (m, f) -> m.list(f).bit(), l -> l.list().bit()) |
| : getWriter(name, (m, f) -> m.bit(f), l -> l.bit()); |
| return new DrillBoolConverter(writer); |
| } |
| case BINARY: { |
| LogicalTypeAnnotation.LogicalTypeAnnotationVisitor<PrimitiveConverter> typeAnnotationVisitor = new LogicalTypeAnnotation.LogicalTypeAnnotationVisitor<PrimitiveConverter>() { |
| @Override |
| public Optional<PrimitiveConverter> visit(LogicalTypeAnnotation.DecimalLogicalTypeAnnotation decimalLogicalType) { |
| ParquetReaderUtility.checkDecimalTypeEnabled(options); |
| return Optional.of(getVarDecimalConverter(name, type)); |
| } |
| |
| @Override |
| public Optional<PrimitiveConverter> visit(LogicalTypeAnnotation.StringLogicalTypeAnnotation stringLogicalType) { |
| return Optional.of(getVarCharConverter(name, type)); |
| } |
| |
| @Override |
| public Optional<PrimitiveConverter> visit(LogicalTypeAnnotation.EnumLogicalTypeAnnotation stringLogicalType) { |
| return Optional.of(getVarCharConverter(name, type)); |
| } |
| }; |
| Supplier<PrimitiveConverter> converterSupplier = () -> { |
| VarBinaryWriter writer = type.isRepetition(Repetition.REPEATED) |
| ? getWriter(name, (m, f) -> m.list(f).varBinary(), l -> l.list().varBinary()) |
| : getWriter(name, MapWriter::varBinary, ListWriter::varBinary); |
| return new DrillVarBinaryConverter(writer, mutator.getManagedBuffer()); |
| }; |
| return Optional.ofNullable(type.getLogicalTypeAnnotation()) |
| .map(typeAnnotation -> typeAnnotation.accept(typeAnnotationVisitor)) |
| .flatMap(Function.identity()) |
| .orElseGet(converterSupplier); |
| } |
| case FIXED_LEN_BYTE_ARRAY: |
| LogicalTypeAnnotation.LogicalTypeAnnotationVisitor<PrimitiveConverter> typeAnnotationVisitor = new LogicalTypeAnnotation.LogicalTypeAnnotationVisitor<PrimitiveConverter>() { |
| @Override |
| public Optional<PrimitiveConverter> visit(LogicalTypeAnnotation.DecimalLogicalTypeAnnotation decimalLogicalType) { |
| ParquetReaderUtility.checkDecimalTypeEnabled(options); |
| return Optional.of(getVarDecimalConverter(name, type)); |
| } |
| |
| @Override |
| public Optional<PrimitiveConverter> visit(LogicalTypeAnnotation.IntervalLogicalTypeAnnotation intervalLogicalType) { |
| IntervalWriter writer = type.isRepetition(Repetition.REPEATED) |
| ? getWriter(name, (m, f) -> m.list(f).interval(), l -> l.list().interval()) |
| : getWriter(name, MapWriter::interval, ListWriter::interval); |
| return Optional.of(new DrillFixedLengthByteArrayToInterval(writer)); |
| } |
| }; |
| |
| Supplier<PrimitiveConverter> converterSupplier = () -> { |
| VarBinaryWriter writer = type.isRepetition(Repetition.REPEATED) |
| ? getWriter(name, (m, f) -> m.list(f).varBinary(), l -> l.list().varBinary()) |
| : getWriter(name, MapWriter::varBinary, ListWriter::varBinary); |
| return new DrillFixedBinaryToVarbinaryConverter(writer, type.getTypeLength(), mutator.getManagedBuffer()); |
| }; |
| return Optional.ofNullable(type.getLogicalTypeAnnotation()) |
| .map(typeAnnotation -> typeAnnotation.accept(typeAnnotationVisitor)) |
| .flatMap(Function.identity()) |
| .orElseGet(converterSupplier); |
| default: |
| throw new UnsupportedOperationException("Unsupported type: " + type.getPrimitiveTypeName()); |
| } |
| } |
| |
| private PrimitiveConverter getVarCharConverter(String name, PrimitiveType type) { |
| VarCharWriter writer = type.isRepetition(Repetition.REPEATED) |
| ? getWriter(name, (m, f) -> m.list(f).varChar(), l -> l.list().varChar()) |
| : getWriter(name, (m, f) -> m.varChar(f), l -> l.varChar()); |
| return new DrillVarCharConverter(writer, mutator.getManagedBuffer()); |
| } |
| |
| private TimeStampWriter getTimeStampWriter(String name, PrimitiveType type) { |
| return type.isRepetition(Repetition.REPEATED) |
| ? getWriter(name, (m, f) -> m.list(f).timeStamp(), l -> l.list().timeStamp()) |
| : getWriter(name, (m, f) -> m.timeStamp(f), l -> l.timeStamp()); |
| } |
| |
| private PrimitiveConverter getBigIntConverter(String name, PrimitiveType type) { |
| BigIntWriter writer = type.isRepetition(Repetition.REPEATED) |
| ? getWriter(name, (m, f) -> m.list(f).bigInt(), l -> l.list().bigInt()) |
| : getWriter(name, (m, f) -> m.bigInt(f), l -> l.bigInt()); |
| return new DrillBigIntConverter(writer); |
| } |
| |
| private PrimitiveConverter getIntConverter(String name, PrimitiveType type) { |
| IntWriter writer = type.isRepetition(Repetition.REPEATED) |
| ? getWriter(name, (m, f) -> m.list(f).integer(), l -> l.list().integer()) |
| : getWriter(name, (m, f) -> m.integer(f), l -> l.integer()); |
| return new DrillIntConverter(writer); |
| } |
| |
| private PrimitiveConverter getVarDecimalConverter(String name, PrimitiveType type) { |
| int scale = type.getDecimalMetadata().getScale(); |
| int precision = type.getDecimalMetadata().getPrecision(); |
| VarDecimalWriter writer = type.isRepetition(Repetition.REPEATED) |
| ? getWriter(name, (m, f) -> m.list(f).varDecimal(precision, scale), l -> l.list().varDecimal(precision, scale)) |
| : getWriter(name, (m, f) -> m.varDecimal(f, precision, scale), l -> l.varDecimal(precision, scale)); |
| return new DrillVarDecimalConverter(writer, precision, scale, mutator.getManagedBuffer()); |
| } |
| |
| @Override |
| public Converter getConverter(int i) { |
| return converters.get(i); |
| } |
| |
| @Override |
| public void start() { |
| if (isMapWriter()) { |
| ((MapWriter) baseWriter).start(); |
| } else { |
| ((ListWriter) baseWriter).startList(); |
| } |
| } |
| |
| boolean isMapWriter() { |
| return baseWriter instanceof SingleMapWriter |
| || baseWriter instanceof AbstractRepeatedMapWriter; |
| } |
| |
| @Override |
| public void end() { |
| if (isMapWriter()) { |
| ((MapWriter) baseWriter).end(); |
| } else { |
| ((ListWriter) baseWriter).endList(); |
| } |
| } |
| |
| @Override |
| public String toString() { |
| return converterName; |
| } |
| |
| private <T> T getWriter(String name, BiFunction<MapWriter, String, T> fromMap, Function<BaseWriter.ListWriter, T> fromList) { |
| if (isMapWriter()) { |
| return fromMap.apply((MapWriter) baseWriter, name); |
| } else if (baseWriter instanceof ListWriter) { |
| return fromList.apply((ListWriter) baseWriter); |
| } else { |
| throw new IllegalStateException(String.format("Parent writer with type [%s] is unsupported", baseWriter.getClass())); |
| } |
| } |
| |
| public static class DrillIntConverter extends PrimitiveConverter { |
| private IntWriter writer; |
| private IntHolder holder = new IntHolder(); |
| |
| public DrillIntConverter(IntWriter writer) { |
| super(); |
| this.writer = writer; |
| } |
| |
| @Override |
| public void addInt(int value) { |
| holder.value = value; |
| writer.write(holder); |
| } |
| } |
| |
| public static class CorruptionDetectingDateConverter extends PrimitiveConverter { |
| private DateWriter writer; |
| private DateHolder holder = new DateHolder(); |
| |
| public CorruptionDetectingDateConverter(DateWriter writer) { |
| this.writer = writer; |
| } |
| |
| @Override |
| public void addInt(int value) { |
| if (value > ParquetReaderUtility.DATE_CORRUPTION_THRESHOLD) { |
| holder.value = (value - ParquetReaderUtility.CORRECT_CORRUPT_DATE_SHIFT) * DateTimeConstants.MILLIS_PER_DAY; |
| } else { |
| holder.value = value * (long) DateTimeConstants.MILLIS_PER_DAY; |
| } |
| writer.write(holder); |
| } |
| } |
| |
| public static class DrillCorruptedDateConverter extends PrimitiveConverter { |
| private DateWriter writer; |
| private DateHolder holder = new DateHolder(); |
| |
| public DrillCorruptedDateConverter(DateWriter writer) { |
| this.writer = writer; |
| } |
| |
| @Override |
| public void addInt(int value) { |
| holder.value = (value - ParquetReaderUtility.CORRECT_CORRUPT_DATE_SHIFT) * DateTimeConstants.MILLIS_PER_DAY; |
| writer.write(holder); |
| } |
| } |
| |
| public static class DrillDateConverter extends PrimitiveConverter { |
| private DateWriter writer; |
| private DateHolder holder = new DateHolder(); |
| |
| public DrillDateConverter(DateWriter writer) { |
| this.writer = writer; |
| } |
| |
| @Override |
| public void addInt(int value) { |
| holder.value = value * (long) DateTimeConstants.MILLIS_PER_DAY; |
| writer.write(holder); |
| } |
| } |
| |
| public static class DrillTimeConverter extends PrimitiveConverter { |
| private TimeWriter writer; |
| private TimeHolder holder = new TimeHolder(); |
| |
| public DrillTimeConverter(TimeWriter writer) { |
| this.writer = writer; |
| } |
| |
| @Override |
| public void addInt(int value) { |
| holder.value = value; |
| writer.write(holder); |
| } |
| } |
| |
| public static class DrillTimeMicrosConverter extends PrimitiveConverter { |
| private final TimeWriter writer; |
| private final TimeHolder holder = new TimeHolder(); |
| |
| public DrillTimeMicrosConverter(TimeWriter writer) { |
| this.writer = writer; |
| } |
| |
| @Override |
| public void addLong(long value) { |
| holder.value = (int) (value / 1000); |
| writer.write(holder); |
| } |
| } |
| |
| public static class DrillBigIntConverter extends PrimitiveConverter { |
| private BigIntWriter writer; |
| private BigIntHolder holder = new BigIntHolder(); |
| |
| public DrillBigIntConverter(BigIntWriter writer) { |
| this.writer = writer; |
| } |
| |
| @Override |
| public void addLong(long value) { |
| holder.value = value; |
| writer.write(holder); |
| } |
| } |
| |
| public static class DrillTimeStampConverter extends PrimitiveConverter { |
| private TimeStampWriter writer; |
| private TimeStampHolder holder = new TimeStampHolder(); |
| |
| public DrillTimeStampConverter(TimeStampWriter writer) { |
| this.writer = writer; |
| } |
| |
| @Override |
| public void addLong(long value) { |
| holder.value = value; |
| writer.write(holder); |
| } |
| } |
| |
| public static class DrillTimeStampMicrosConverter extends PrimitiveConverter { |
| private final TimeStampWriter writer; |
| private final TimeStampHolder holder = new TimeStampHolder(); |
| |
| public DrillTimeStampMicrosConverter(TimeStampWriter writer) { |
| this.writer = writer; |
| } |
| |
| @Override |
| public void addLong(long value) { |
| holder.value = value / 1000; |
| writer.write(holder); |
| } |
| } |
| |
| public static class DrillFloat4Converter extends PrimitiveConverter { |
| private Float4Writer writer; |
| private Float4Holder holder = new Float4Holder(); |
| |
| public DrillFloat4Converter(Float4Writer writer) { |
| this.writer = writer; |
| } |
| |
| @Override |
| public void addFloat(float value) { |
| holder.value = value; |
| writer.write(holder); |
| } |
| } |
| |
| public static class DrillFloat8Converter extends PrimitiveConverter { |
| private Float8Writer writer; |
| private Float8Holder holder = new Float8Holder(); |
| |
| public DrillFloat8Converter(Float8Writer writer) { |
| this.writer = writer; |
| } |
| |
| @Override |
| public void addDouble(double value) { |
| holder.value = value; |
| writer.write(holder); |
| } |
| } |
| |
| public static class DrillBoolConverter extends PrimitiveConverter { |
| private BitWriter writer; |
| private BitHolder holder = new BitHolder(); |
| |
| public DrillBoolConverter(BitWriter writer) { |
| this.writer = writer; |
| } |
| |
| @Override |
| public void addBoolean(boolean value) { |
| holder.value = value ? 1 : 0; |
| writer.write(holder); |
| } |
| } |
| |
| public static class DrillVarBinaryConverter extends PrimitiveConverter { |
| private VarBinaryWriter writer; |
| private DrillBuf buf; |
| private VarBinaryHolder holder = new VarBinaryHolder(); |
| |
| public DrillVarBinaryConverter(VarBinaryWriter writer, DrillBuf buf) { |
| this.writer = writer; |
| this.buf = buf; |
| } |
| |
| @Override |
| public void addBinary(Binary value) { |
| holder.buffer = buf = buf.reallocIfNeeded(value.length()); |
| buf.setBytes(0, value.toByteBuffer()); |
| holder.start = 0; |
| holder.end = value.length(); |
| writer.write(holder); |
| } |
| } |
| |
| public static class DrillVarCharConverter extends PrimitiveConverter { |
| private VarCharWriter writer; |
| private VarCharHolder holder = new VarCharHolder(); |
| private DrillBuf buf; |
| |
| public DrillVarCharConverter(VarCharWriter writer, DrillBuf buf) { |
| this.writer = writer; |
| this.buf = buf; |
| } |
| |
| @Override |
| public void addBinary(Binary value) { |
| holder.buffer = buf = buf.reallocIfNeeded(value.length()); |
| buf.setBytes(0, value.toByteBuffer()); |
| holder.start = 0; |
| holder.end = value.length(); |
| writer.write(holder); |
| } |
| } |
| |
| public static class DrillVarDecimalConverter extends PrimitiveConverter { |
| private VarDecimalWriter writer; |
| private VarDecimalHolder holder = new VarDecimalHolder(); |
| private DrillBuf buf; |
| |
| public DrillVarDecimalConverter(VarDecimalWriter writer, int precision, int scale, DrillBuf buf) { |
| this.writer = writer; |
| holder.scale = scale; |
| holder.precision = precision; |
| this.buf = buf; |
| } |
| |
| @Override |
| public void addBinary(Binary value) { |
| holder.buffer = buf.reallocIfNeeded(value.length()); |
| holder.buffer.setBytes(0, value.toByteBuffer()); |
| holder.start = 0; |
| holder.end = value.length(); |
| writer.write(holder); |
| } |
| |
| @Override |
| public void addInt(int value) { |
| byte[] bytes = Ints.toByteArray(value); |
| holder.buffer = buf.reallocIfNeeded(bytes.length); |
| holder.buffer.setBytes(0, bytes); |
| holder.start = 0; |
| holder.end = bytes.length; |
| writer.write(holder); |
| } |
| |
| @Override |
| public void addLong(long value) { |
| byte[] bytes = Longs.toByteArray(value); |
| holder.buffer = buf.reallocIfNeeded(bytes.length); |
| holder.buffer.setBytes(0, bytes); |
| holder.start = 0; |
| holder.end = bytes.length; |
| writer.write(holder); |
| } |
| } |
| |
| public static class DrillFixedLengthByteArrayToInterval extends PrimitiveConverter { |
| final private IntervalWriter writer; |
| final private IntervalHolder holder = new IntervalHolder(); |
| |
| public DrillFixedLengthByteArrayToInterval(IntervalWriter writer) { |
| this.writer = writer; |
| } |
| |
| @Override |
| public void addBinary(Binary value) { |
| final byte[] input = value.getBytes(); |
| holder.months = ParquetReaderUtility.getIntFromLEBytes(input, 0); |
| holder.days = ParquetReaderUtility.getIntFromLEBytes(input, 4); |
| holder.milliseconds = ParquetReaderUtility.getIntFromLEBytes(input, 8); |
| writer.write(holder); |
| } |
| } |
| /** |
| * Parquet currently supports a fixed binary type, which is not implemented in Drill. For now this |
| * data will be read in a s varbinary and the same length will be recorded for each value. |
| */ |
| public static class DrillFixedBinaryToVarbinaryConverter extends PrimitiveConverter { |
| private VarBinaryWriter writer; |
| private VarBinaryHolder holder = new VarBinaryHolder(); |
| |
| public DrillFixedBinaryToVarbinaryConverter(VarBinaryWriter writer, int length, DrillBuf buf) { |
| this.writer = writer; |
| holder.buffer = buf = buf.reallocIfNeeded(length); |
| holder.start = 0; |
| holder.end = length; |
| } |
| |
| @Override |
| public void addBinary(Binary value) { |
| holder.buffer.setBytes(0, value.toByteBuffer()); |
| writer.write(holder); |
| } |
| } |
| |
| /** |
| * Parquet currently supports a fixed binary type INT96 for storing hive, impala timestamp |
| * with nanoseconds precision. |
| */ |
| public static class DrillFixedBinaryToTimeStampConverter extends PrimitiveConverter { |
| private TimeStampWriter writer; |
| private TimeStampHolder holder = new TimeStampHolder(); |
| |
| public DrillFixedBinaryToTimeStampConverter(TimeStampWriter writer) { |
| this.writer = writer; |
| } |
| |
| @Override |
| public void addBinary(Binary value) { |
| holder.value = getDateTimeValueFromBinary(value, true); |
| writer.write(holder); |
| } |
| } |
| |
| /** |
| * Converter for field which is present in schema but don't need any actions to be performed by writer. |
| * For this purpose the converter is added to converter's chain but simply does nothing and actual writing |
| * will be performed by other converters in the chain. |
| */ |
| private static class DrillIntermediateParquetGroupConverter extends DrillParquetGroupConverter { |
| |
| DrillIntermediateParquetGroupConverter(OutputMutator mutator, BaseWriter baseWriter, GroupType schema, |
| Collection<SchemaPath> columns, OptionManager options, |
| ParquetReaderUtility.DateCorruptionStatus containsCorruptedDates, |
| boolean skipRepeated, String parentName) { |
| super(mutator, baseWriter, schema, columns, options, containsCorruptedDates, skipRepeated, parentName); |
| } |
| |
| @Override |
| public void start() {} |
| |
| @Override |
| public void end() {} |
| } |
| |
| |
| } |
