From 3ea610bf47f100c5613fc79614b937c7381aa506 Mon Sep 17 00:00:00 2001 From: xicm Date: Fri, 26 May 2023 15:29:49 +0800 Subject: [PATCH 1/3] [HUDI-5189] Make HiveAvroSerializer compatible with hive3 and support timestamp --- .../testutils/HoodieMergeOnReadTestUtils.java | 2 +- .../hadoop/HoodieColumnProjectionUtils.java | 18 +++ .../hudi/hadoop/HoodieParquetInputFormat.java | 41 ++++++- .../hadoop/avro/HoodieAvroParquetReader.java | 105 ++++++++++++++++ ...oodieTimestampAwareParquetInputFormat.java | 44 +++++++ .../AbstractRealtimeRecordReader.java | 9 ++ .../RealtimeCompactedRecordReader.java | 2 +- .../RealtimeUnmergedRecordReader.java | 2 +- .../hudi/hadoop/utils/HiveAvroSerializer.java | 15 +-- .../hudi/hadoop/utils/HoodieHiveUtils.java | 41 +++++++ .../HoodieRealtimeRecordReaderUtils.java | 40 ++++--- .../hudi/hadoop/utils/shims/Hive2Shims.java | 45 +++++++ .../hudi/hadoop/utils/shims/Hive3Shims.java | 113 ++++++++++++++++++ .../hadoop/TestHoodieParquetInputFormat.java | 66 ++++++++++ .../hadoop/testutils/InputFormatTestUtil.java | 2 +- .../hadoop/utils/TestHiveAvroSerializer.java | 7 +- .../src/test/resources/test_timetype.avsc | 45 +++++++ 17 files changed, 563 insertions(+), 34 deletions(-) create mode 100644 hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/avro/HoodieAvroParquetReader.java create mode 100644 hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/avro/HoodieTimestampAwareParquetInputFormat.java create mode 100644 hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive2Shims.java create mode 100644 hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive3Shims.java create mode 100644 hudi-hadoop-mr/src/test/resources/test_timetype.avsc diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieMergeOnReadTestUtils.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieMergeOnReadTestUtils.java index ab8ecf2431789..9275cd88c43d7 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieMergeOnReadTestUtils.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieMergeOnReadTestUtils.java @@ -211,7 +211,7 @@ private static void setPropsForInputFormat(FileInputFormat inputFormat, JobConf conf.set(hive_metastoreConstants.META_TABLE_PARTITION_COLUMNS, "datestr"); conf.set(hive_metastoreConstants.META_TABLE_COLUMN_TYPES, hiveColumnTypesWithDatestr); conf.set(IOConstants.COLUMNS, hiveColumnNames); - conf.get(IOConstants.COLUMNS_TYPES, hiveColumnTypesWithDatestr); + conf.set(IOConstants.COLUMNS_TYPES, hiveColumnTypesWithDatestr); // Hoodie Input formats are also configurable Configurable configurable = (Configurable)inputFormat; diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieColumnProjectionUtils.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieColumnProjectionUtils.java index 9ca99c41888b1..a2134643c3805 100644 --- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieColumnProjectionUtils.java +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieColumnProjectionUtils.java @@ -109,4 +109,22 @@ public static List> getIOColumnNameAndTypes(Configuration co .collect(Collectors.toList()); } + /** + * if schema contains timestamp columns, this method is used for compatibility when there is no timestamp fields + * We expect 3 cases to use parquet-avro reader to read timestamp column: + * 1. read columns contain timestamp type + * 2. no read columns and exists original columns contain timestamp type + * 3. no read columns and no original columns, but avro schema contains type + */ + public static boolean supportTimestamp(Configuration conf) { + List reads = Arrays.asList(getReadColumnNames(conf)); + if (reads.isEmpty()) { + return getIOColumnTypes(conf).contains("timestamp"); + } + List names = getIOColumns(conf); + List types = getIOColumnTypes(conf); + return types.isEmpty() || IntStream.range(0, names.size()).filter(i -> reads.contains(names.get(i))) + .anyMatch(i -> types.get(i).equals("timestamp")); + } + } \ No newline at end of file diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieParquetInputFormat.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieParquetInputFormat.java index c3588ef4e30f5..e9c013a316c97 100644 --- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieParquetInputFormat.java +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieParquetInputFormat.java @@ -21,7 +21,11 @@ import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.util.collection.Pair; import org.apache.hudi.hadoop.utils.HoodieHiveUtils; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.hadoop.avro.HoodieTimestampAwareParquetInputFormat; + import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants; +import org.apache.hadoop.hive.ql.io.parquet.read.ParquetRecordReaderWrapper; import org.apache.hadoop.hive.ql.io.sarg.ConvertAstToSearchArg; import org.apache.hadoop.hive.ql.plan.TableScanDesc; import org.apache.hadoop.io.ArrayWritable; @@ -32,11 +36,14 @@ import org.apache.hadoop.mapred.RecordReader; import org.apache.hadoop.mapred.Reporter; import org.apache.hudi.hadoop.utils.HoodieRealtimeInputFormatUtils; +import org.apache.parquet.hadoop.ParquetInputFormat; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; +import java.lang.reflect.Constructor; +import java.util.Arrays; import java.util.List; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -52,12 +59,36 @@ public class HoodieParquetInputFormat extends HoodieParquetInputFormatBase { private static final Logger LOG = LoggerFactory.getLogger(HoodieParquetInputFormat.class); + private boolean supportAvroRead = true; + public HoodieParquetInputFormat() { super(new HoodieCopyOnWriteTableInputFormat()); + initAvroInputFormat(); } protected HoodieParquetInputFormat(HoodieCopyOnWriteTableInputFormat delegate) { super(delegate); + initAvroInputFormat(); + } + + /** + * Spark2 use `parquet.hadoopParquetInputFormat` in `com.twitter:parquet-hadoop-bundle`. + * So that we need to distinguish the constructions of classes with + * `parquet.hadoopParquetInputFormat` or `org.apache.parquet.hadoop.ParquetInputFormat`. + * If we use `org.apache.parquet:parquet-hadoop`, we can use `HudiAvroParquetInputFormat` + * in Hive or Spark3 to get timestamp with correct type. + */ + private void initAvroInputFormat() { + try { + Constructor[] constructors = ParquetRecordReaderWrapper.class.getConstructors(); + if (Arrays.stream(constructors) + .anyMatch(c -> c.getParameterCount() > 0 && c.getParameterTypes()[0] + .getName().equals(ParquetInputFormat.class.getName()))) { + supportAvroRead = true; + } + } catch (SecurityException e) { + throw new HoodieException("Failed to check if support avro reader: " + e.getMessage(), e); + } } @Override @@ -93,7 +124,15 @@ public RecordReader getRecordReader(final InputSpli private RecordReader getRecordReaderInternal(InputSplit split, JobConf job, Reporter reporter) throws IOException { - return super.getRecordReader(split, job, reporter); + try { + if (supportAvroRead && HoodieColumnProjectionUtils.supportTimestamp(job)) { + return new ParquetRecordReaderWrapper(new HoodieTimestampAwareParquetInputFormat(), split, job, reporter); + } else { + return super.getRecordReader(split, job, reporter); + } + } catch (final InterruptedException | IOException e) { + throw new RuntimeException("Cannot create a RecordReaderWrapper", e); + } } private RecordReader createBootstrappingRecordReader(InputSplit split, diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/avro/HoodieAvroParquetReader.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/avro/HoodieAvroParquetReader.java new file mode 100644 index 0000000000000..045dd79340f96 --- /dev/null +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/avro/HoodieAvroParquetReader.java @@ -0,0 +1,105 @@ +/* + * 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.hudi.hadoop.avro; + +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericRecord; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.io.ArrayWritable; +import org.apache.hadoop.mapreduce.InputSplit; +import org.apache.hadoop.mapreduce.RecordReader; +import org.apache.hadoop.mapreduce.TaskAttemptContext; +import org.apache.hudi.hadoop.HoodieColumnProjectionUtils; +import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils; +import org.apache.parquet.avro.AvroReadSupport; +import org.apache.parquet.avro.AvroSchemaConverter; +import org.apache.parquet.format.converter.ParquetMetadataConverter; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.ParquetInputSplit; +import org.apache.parquet.hadoop.ParquetRecordReader; +import org.apache.parquet.hadoop.metadata.ParquetMetadata; +import org.apache.parquet.schema.MessageType; + +import java.io.IOException; +import java.util.Arrays; +import java.util.List; +import java.util.stream.Collectors; + +import static org.apache.parquet.hadoop.ParquetInputFormat.getFilter; + +public class HoodieAvroParquetReader extends RecordReader { + + private final ParquetRecordReader parquetRecordReader; + private Schema baseSchema; + + public HoodieAvroParquetReader(InputSplit inputSplit, Configuration conf) throws IOException { + AvroReadSupport avroReadSupport = new AvroReadSupport<>(); + // if exists read columns, we need to filter columns. + List readColNames = Arrays.asList(HoodieColumnProjectionUtils.getReadColumnNames(conf)); + if (!readColNames.isEmpty()) { + // get base schema + ParquetMetadata fileFooter = + ParquetFileReader.readFooter(conf, ((ParquetInputSplit) inputSplit).getPath(), ParquetMetadataConverter.NO_FILTER); + MessageType messageType = fileFooter.getFileMetaData().getSchema(); + baseSchema = new AvroSchemaConverter(conf).convert(messageType); + // filter schema for reading + final Schema filterSchema = Schema.createRecord(baseSchema.getName(), + baseSchema.getDoc(), baseSchema.getNamespace(), baseSchema.isError(), + baseSchema.getFields().stream() + .filter(f -> readColNames.contains(f.name())) + .map(f -> new Schema.Field(f.name(), f.schema(), f.doc(), f.defaultVal())) + .collect(Collectors.toList())); + avroReadSupport.setAvroReadSchema(conf, filterSchema); + avroReadSupport.setRequestedProjection(conf, filterSchema); + } + parquetRecordReader = new ParquetRecordReader<>(avroReadSupport, getFilter(conf)); + } + + @Override + public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException { + parquetRecordReader.initialize(split, context); + } + + @Override + public boolean nextKeyValue() throws IOException, InterruptedException { + return parquetRecordReader.nextKeyValue(); + } + + @Override + public Void getCurrentKey() throws IOException, InterruptedException { + return parquetRecordReader.getCurrentKey(); + } + + @Override + public ArrayWritable getCurrentValue() throws IOException, InterruptedException { + GenericRecord record = parquetRecordReader.getCurrentValue(); + return (ArrayWritable) HoodieRealtimeRecordReaderUtils.avroToArrayWritable(record, baseSchema, true); + } + + @Override + public float getProgress() throws IOException, InterruptedException { + return parquetRecordReader.getProgress(); + } + + @Override + public void close() throws IOException { + parquetRecordReader.close(); + } +} \ No newline at end of file diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/avro/HoodieTimestampAwareParquetInputFormat.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/avro/HoodieTimestampAwareParquetInputFormat.java new file mode 100644 index 0000000000000..ac3a9e735b8a1 --- /dev/null +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/avro/HoodieTimestampAwareParquetInputFormat.java @@ -0,0 +1,44 @@ +/* + * 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.hudi.hadoop.avro; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.io.ArrayWritable; +import org.apache.hadoop.mapreduce.InputSplit; +import org.apache.hadoop.mapreduce.RecordReader; +import org.apache.hadoop.mapreduce.TaskAttemptContext; +import org.apache.parquet.hadoop.ParquetInputFormat; +import org.apache.parquet.hadoop.util.ContextUtil; + +import java.io.IOException; + +/** + * To resolve issue Fix Timestamp/Date type read by Hive3, + * we need to handle timestamp types separately based on the parquet-avro approach + */ +public class HoodieTimestampAwareParquetInputFormat extends ParquetInputFormat { + + @Override + public RecordReader createRecordReader( + InputSplit inputSplit, + TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException { + Configuration conf = ContextUtil.getConfiguration(taskAttemptContext); + return new HoodieAvroParquetReader(inputSplit, conf); + } +} diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/AbstractRealtimeRecordReader.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/AbstractRealtimeRecordReader.java index 49ba2601a306d..85d4a92311bb6 100644 --- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/AbstractRealtimeRecordReader.java +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/AbstractRealtimeRecordReader.java @@ -25,6 +25,7 @@ import org.apache.hudi.common.table.TableSchemaResolver; import org.apache.hudi.common.util.Option; import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.hadoop.HoodieColumnProjectionUtils; import org.apache.hudi.hadoop.SchemaEvolutionContext; import org.apache.hudi.hadoop.utils.HiveAvroSerializer; import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils; @@ -72,6 +73,7 @@ public abstract class AbstractRealtimeRecordReader { protected boolean supportPayload = true; // handle hive type to avro record protected HiveAvroSerializer serializer; + private boolean supportTimestamp; public AbstractRealtimeRecordReader(RealtimeSplit split, JobConf job) { this.split = split; @@ -158,6 +160,9 @@ private void init() throws Exception { readerSchema = HoodieRealtimeRecordReaderUtils.generateProjectionSchema(writerSchema, schemaFieldsMap, projectionFields); LOG.info(String.format("About to read compacted logs %s for base split %s, projecting cols %s", split.getDeltaLogPaths(), split.getPath(), projectionFields)); + + // get timestamp columns + supportTimestamp = HoodieColumnProjectionUtils.supportTimestamp(jobConf); } public Schema constructHiveOrderedSchema(Schema writerSchema, Map schemaFieldsMap, String hiveColumnString) { @@ -200,6 +205,10 @@ public Schema getHiveSchema() { return hiveSchema; } + public boolean isSupportTimestamp() { + return supportTimestamp; + } + public RealtimeSplit getSplit() { return split; } diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeCompactedRecordReader.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeCompactedRecordReader.java index 0d12576310848..8701939ee25de 100644 --- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeCompactedRecordReader.java +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeCompactedRecordReader.java @@ -161,7 +161,7 @@ private void setUpWritable(Option rec, ArrayWritable ar // we assume, a later safe record in the log, is newer than what we have in the map & // replace it. Since we want to return an arrayWritable which is the same length as the elements in the latest // schema, we use writerSchema to create the arrayWritable from the latest generic record - ArrayWritable aWritable = (ArrayWritable) HoodieRealtimeRecordReaderUtils.avroToArrayWritable(recordToReturn, getHiveSchema()); + ArrayWritable aWritable = (ArrayWritable) HoodieRealtimeRecordReaderUtils.avroToArrayWritable(recordToReturn, getHiveSchema(), isSupportTimestamp()); Writable[] replaceValue = aWritable.get(); if (LOG.isDebugEnabled()) { LOG.debug(String.format("key %s, base values: %s, log values: %s", key, HoodieRealtimeRecordReaderUtils.arrayWritableToString(arrayWritable), diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeUnmergedRecordReader.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeUnmergedRecordReader.java index 15aa719809a63..427b758db346a 100644 --- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeUnmergedRecordReader.java +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeUnmergedRecordReader.java @@ -111,7 +111,7 @@ private List> getParallelProducers( scannerBuilder.withLogRecordScannerCallback(record -> { // convert Hoodie log record to Hadoop AvroWritable and buffer GenericRecord rec = (GenericRecord) record.toIndexedRecord(getReaderSchema(), payloadProps).get().getData(); - ArrayWritable aWritable = (ArrayWritable) HoodieRealtimeRecordReaderUtils.avroToArrayWritable(rec, getHiveSchema()); + ArrayWritable aWritable = (ArrayWritable) HoodieRealtimeRecordReaderUtils.avroToArrayWritable(rec, getHiveSchema(), isSupportTimestamp()); queue.insertRecord(aWritable); }) .build(); diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HiveAvroSerializer.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HiveAvroSerializer.java index 4455c4e96c2c1..9d5572a59020a 100644 --- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HiveAvroSerializer.java +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HiveAvroSerializer.java @@ -34,7 +34,6 @@ import org.apache.hadoop.hive.common.type.HiveVarchar; import org.apache.hadoop.hive.serde2.avro.AvroSerdeUtils; import org.apache.hadoop.hive.serde2.avro.InstanceCache; -import org.apache.hadoop.hive.serde2.io.DateWritable; import org.apache.hadoop.hive.serde2.objectinspector.ListObjectInspector; import org.apache.hadoop.hive.serde2.objectinspector.MapObjectInspector; import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector; @@ -42,20 +41,19 @@ import org.apache.hadoop.hive.serde2.objectinspector.StructField; import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector; import org.apache.hadoop.hive.serde2.objectinspector.UnionObjectInspector; -import org.apache.hadoop.hive.serde2.objectinspector.primitive.DateObjectInspector; -import org.apache.hadoop.hive.serde2.objectinspector.primitive.TimestampObjectInspector; import org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableDateObjectInspector; +import org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableTimestampObjectInspector; import org.apache.hadoop.hive.serde2.typeinfo.ListTypeInfo; import org.apache.hadoop.hive.serde2.typeinfo.MapTypeInfo; import org.apache.hadoop.hive.serde2.typeinfo.StructTypeInfo; import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo; import org.apache.hadoop.hive.serde2.typeinfo.UnionTypeInfo; import org.apache.hadoop.io.ArrayWritable; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.math.BigDecimal; -import java.sql.Timestamp; import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; @@ -301,14 +299,13 @@ private Object serializePrimitive(PrimitiveObjectInspector fieldOI, Object struc String string = (String)fieldOI.getPrimitiveJavaObject(structFieldData); return new Utf8(string); case DATE: - return DateWritable.dateToDays(((DateObjectInspector)fieldOI).getPrimitiveJavaObject(structFieldData)); + return HoodieHiveUtils.getDays(structFieldData); case TIMESTAMP: - Timestamp timestamp = - ((TimestampObjectInspector) fieldOI).getPrimitiveJavaObject(structFieldData); - return timestamp.getTime(); + Object timestamp = ((WritableTimestampObjectInspector) fieldOI).getPrimitiveJavaObject(structFieldData); + return HoodieHiveUtils.getMills(timestamp); case INT: if (schema.getLogicalType() != null && schema.getLogicalType().getName().equals("date")) { - return DateWritable.dateToDays(new WritableDateObjectInspector().getPrimitiveJavaObject(structFieldData)); + return new WritableDateObjectInspector().getPrimitiveWritableObject(structFieldData).getDays(); } return fieldOI.getPrimitiveJavaObject(structFieldData); case UNKNOWN: diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieHiveUtils.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieHiveUtils.java index 465aa96638779..9da27eaf91b40 100644 --- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieHiveUtils.java +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieHiveUtils.java @@ -20,11 +20,16 @@ import org.apache.hudi.common.util.CollectionUtils; import org.apache.hudi.common.util.Option; +import org.apache.hudi.hadoop.utils.shims.Hive2Shims; +import org.apache.hudi.hadoop.utils.shims.Hive3Shims; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.Writable; import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.mapreduce.JobContext; + +import org.apache.hive.common.util.HiveVersionInfo; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -75,6 +80,8 @@ public class HoodieHiveUtils { public static final Pattern HOODIE_CONSUME_MODE_PATTERN_STRING = Pattern.compile("hoodie\\.(.*)\\.consume\\.mode"); public static final String GLOBALLY_CONSISTENT_READ_TIMESTAMP = "last_replication_timestamp"; + private static final boolean isHive3 = isHive3(); + public static boolean shouldIncludePendingCommits(JobConf job, String tableName) { return job.getBoolean(String.format(HOODIE_CONSUME_PENDING_COMMITS, tableName), false); } @@ -147,4 +154,38 @@ public static List getIncrementalTableNames(JobContext job) { public static boolean isIncrementalUseDatabase(Configuration conf) { return conf.getBoolean(HOODIE_INCREMENTAL_USE_DATABASE, false); } + + public static boolean isHive3() { + return HiveVersionInfo.getShortVersion().startsWith("3"); + } + + public static boolean isHive2() { + return HiveVersionInfo.getShortVersion().startsWith("2"); + } + + /** + * Get timestamp writeable object from long value. + * Hive3 use TimestampWritableV2 to build timestamp objects and Hive2 use TimestampWritable. + * So that we need to initialize timestamp according to the version of Hive. + */ + public static Writable getTimestampWriteable(long value, boolean timestampMillis) { + return isHive3 ? Hive3Shims.getTimestampWriteable(value, timestampMillis) : Hive2Shims.getTimestampWriteable(value, timestampMillis); + } + + /** + * Get date writeable object from int value. + * Hive3 use DateWritableV2 to build date objects and Hive2 use DateWritable. + * So that we need to initialize date according to the version of Hive. + */ + public static Writable getDateWriteable(int value) { + return isHive3 ? Hive3Shims.getDateWriteable(value) : Hive2Shims.getDateWriteable(value); + } + + public static int getDays(Object dateWritable) { + return isHive3 ? Hive3Shims.getDays(dateWritable) : Hive2Shims.getDays(dateWritable); + } + + public static long getMills(Object timestamp) { + return isHive3 ? Hive3Shims.getMills(timestamp) : Hive2Shims.getMills(timestamp); + } } diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java index 19441e17a59b4..6d010e12cdd01 100644 --- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java @@ -24,15 +24,14 @@ import org.apache.avro.AvroRuntimeException; import org.apache.avro.JsonProperties; +import org.apache.avro.LogicalType; import org.apache.avro.LogicalTypes; import org.apache.avro.Schema; import org.apache.avro.generic.GenericArray; import org.apache.avro.generic.GenericFixed; import org.apache.avro.generic.GenericRecord; -import org.apache.hadoop.hive.serde2.io.DateWritable; import org.apache.hadoop.hive.serde2.io.DoubleWritable; import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable; -import org.apache.hadoop.hive.serde2.io.TimestampWritable; import org.apache.hadoop.hive.serde2.typeinfo.DecimalTypeInfo; import org.apache.hadoop.hive.serde2.typeinfo.HiveDecimalUtils; import org.apache.hadoop.io.ArrayWritable; @@ -48,7 +47,6 @@ import org.slf4j.LoggerFactory; import java.nio.ByteBuffer; -import java.sql.Timestamp; import java.util.ArrayList; import java.util.Arrays; import java.util.LinkedHashSet; @@ -147,6 +145,10 @@ public static Map getNameToFieldMap(Schema schema) { * Convert the projected read from delta record into an array writable. */ public static Writable avroToArrayWritable(Object value, Schema schema) { + return avroToArrayWritable(value, schema, false); + } + + public static Writable avroToArrayWritable(Object value, Schema schema, boolean supportTimestamp) { if (value == null) { return null; @@ -159,20 +161,26 @@ public static Writable avroToArrayWritable(Object value, Schema schema) { return new BytesWritable(((ByteBuffer)value).array()); case INT: if (schema.getLogicalType() != null && schema.getLogicalType().getName().equals("date")) { - return new DateWritable((Integer) value); + return HoodieHiveUtils.getDateWriteable((Integer) value); } - return new IntWritable((Integer) value); + return new IntWritable(Integer.parseInt(value.toString())); case LONG: - if (schema.getLogicalType() != null && "timestamp-micros".equals(schema.getLogicalType().getName())) { - return new TimestampWritable(new Timestamp((Long) value)); + LogicalType logicalType = schema.getLogicalType(); + // If there is a specified timestamp or under normal cases, we will process it + if (supportTimestamp) { + if (LogicalTypes.timestampMillis().equals(logicalType)) { + return HoodieHiveUtils.getTimestampWriteable((Long) value, true); + } else if (LogicalTypes.timestampMicros().equals(logicalType)) { + return HoodieHiveUtils.getTimestampWriteable((Long) value, false); + } } - return new LongWritable((Long) value); + return new LongWritable(Long.parseLong(value.toString())); case FLOAT: - return new FloatWritable((Float) value); + return new FloatWritable(Float.parseFloat(value.toString())); case DOUBLE: - return new DoubleWritable((Double) value); + return new DoubleWritable(Double.parseDouble(value.toString())); case BOOLEAN: - return new BooleanWritable((Boolean) value); + return new BooleanWritable(Boolean.parseBoolean(value.toString())); case NULL: return null; case RECORD: @@ -187,7 +195,7 @@ public static Writable avroToArrayWritable(Object value, Schema schema) { } catch (AvroRuntimeException e) { LOG.debug("Field:" + field.name() + "not found in Schema:" + schema); } - recordValues[recordValueIndex++] = avroToArrayWritable(fieldValue, field.schema()); + recordValues[recordValueIndex++] = avroToArrayWritable(fieldValue, field.schema(), supportTimestamp); } return new ArrayWritable(Writable.class, recordValues); case ENUM: @@ -197,7 +205,7 @@ public static Writable avroToArrayWritable(Object value, Schema schema) { Writable[] arrayValues = new Writable[arrayValue.size()]; int arrayValueIndex = 0; for (Object obj : arrayValue) { - arrayValues[arrayValueIndex++] = avroToArrayWritable(obj, schema.getElementType()); + arrayValues[arrayValueIndex++] = avroToArrayWritable(obj, schema.getElementType(), supportTimestamp); } // Hive 1.x will fail here, it requires values2 to be wrapped into another ArrayWritable return new ArrayWritable(Writable.class, arrayValues); @@ -209,7 +217,7 @@ public static Writable avroToArrayWritable(Object value, Schema schema) { Map.Entry mapEntry = (Map.Entry) entry; Writable[] nestedMapValues = new Writable[2]; nestedMapValues[0] = new Text(mapEntry.getKey().toString()); - nestedMapValues[1] = avroToArrayWritable(mapEntry.getValue(), schema.getValueType()); + nestedMapValues[1] = avroToArrayWritable(mapEntry.getValue(), schema.getValueType(), supportTimestamp); mapValues[mapValueIndex++] = new ArrayWritable(Writable.class, nestedMapValues); } // Hive 1.x will fail here, it requires values3 to be wrapped into another ArrayWritable @@ -222,9 +230,9 @@ public static Writable avroToArrayWritable(Object value, Schema schema) { Schema s1 = types.get(0); Schema s2 = types.get(1); if (s1.getType() == Schema.Type.NULL) { - return avroToArrayWritable(value, s2); + return avroToArrayWritable(value, s2, supportTimestamp); } else if (s2.getType() == Schema.Type.NULL) { - return avroToArrayWritable(value, s1); + return avroToArrayWritable(value, s1, supportTimestamp); } else { throw new IllegalArgumentException("Only support union with null"); } diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive2Shims.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive2Shims.java new file mode 100644 index 0000000000000..925ec5798313f --- /dev/null +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive2Shims.java @@ -0,0 +1,45 @@ +/* + * 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.hudi.hadoop.utils.shims; + +import org.apache.hadoop.hive.serde2.io.DateWritable; +import org.apache.hadoop.hive.serde2.io.TimestampWritable; +import org.apache.hadoop.io.Writable; + +import java.sql.Timestamp; + +public class Hive2Shims { + + public static Writable getTimestampWriteable(long value, boolean timestampMillis) { + Timestamp timestamp = new Timestamp(timestampMillis ? value : value / 1000); + return new TimestampWritable(timestamp); + } + + public static Writable getDateWriteable(int value) { + return new DateWritable(value); + } + + public static int getDays(Object dateWritable) { + return ((DateWritable) dateWritable).getDays(); + } + + public static long getMills(Object timestamp) { + return ((Timestamp) timestamp).getTime(); + } +} diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive3Shims.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive3Shims.java new file mode 100644 index 0000000000000..485d0e00ad3f2 --- /dev/null +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive3Shims.java @@ -0,0 +1,113 @@ +/* + * 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.hudi.hadoop.utils.shims; + +import org.apache.hudi.exception.HoodieException; + +import org.apache.hadoop.io.Writable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.lang.reflect.Constructor; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.sql.Timestamp; + +public class Hive3Shims { + + public static final Logger LOG = LoggerFactory.getLogger(Hive3Shims.class); + + public static final String HIVE_TIMESTAMP_TYPE_CLASS = "org.apache.hadoop.hive.common.type.Timestamp"; + public static final String TIMESTAMP_WRITEABLE_V2_CLASS = "org.apache.hadoop.hive.serde2.io.TimestampWritableV2"; + public static final String DATE_WRITEABLE_V2_CLASS = "org.apache.hadoop.hive.serde2.io.DateWritableV2"; + + private static Class TIMESTAMP_CLASS = null; + private static Method SET_TIME_IN_MILLIS = null; + private static Method TO_SQL_TIMESTAMP = null; + private static Constructor TIMESTAMP_WRITEABLE_V2_CONSTRUCTOR = null; + + private static Class DATE_WRITEABLE_CLASS = null; + private static Method GET_DAYS = null; + private static Constructor DATE_WRITEABLE_V2_CONSTRUCTOR = null; + + static { + // timestamp + try { + TIMESTAMP_CLASS = Class.forName(HIVE_TIMESTAMP_TYPE_CLASS); + SET_TIME_IN_MILLIS = TIMESTAMP_CLASS.getDeclaredMethod("setTimeInMillis", long.class); + TO_SQL_TIMESTAMP = TIMESTAMP_CLASS.getDeclaredMethod("toSqlTimestamp"); + TIMESTAMP_WRITEABLE_V2_CONSTRUCTOR = Class.forName(TIMESTAMP_WRITEABLE_V2_CLASS).getConstructor(TIMESTAMP_CLASS); + } catch (ClassNotFoundException | NoSuchMethodException e) { + LOG.trace("can not find hive3 timestampv2 class or method, use hive2 class!", e); + } + + // date + try { + DATE_WRITEABLE_CLASS = Class.forName(DATE_WRITEABLE_V2_CLASS); + GET_DAYS = DATE_WRITEABLE_CLASS.getDeclaredMethod("getDays"); + DATE_WRITEABLE_V2_CONSTRUCTOR = DATE_WRITEABLE_CLASS.getConstructor(int.class); + } catch (ClassNotFoundException | NoSuchMethodException e) { + LOG.trace("can not find hive3 datev2 class or method, use hive2 class!", e); + } + } + + /** + * Get timestamp writeable object from long value. + * Hive3 use TimestampWritableV2 to build timestamp objects and Hive2 use TimestampWritable. + * So that we need to initialize timestamp according to the version of Hive. + */ + public static Writable getTimestampWriteable(long value, boolean timestampMillis) { + try { + Object timestamp = TIMESTAMP_CLASS.newInstance(); + SET_TIME_IN_MILLIS.invoke(timestamp, timestampMillis ? value : value / 1000); + return (Writable) TIMESTAMP_WRITEABLE_V2_CONSTRUCTOR.newInstance(timestamp); + } catch (IllegalAccessException | InstantiationException | InvocationTargetException e) { + throw new HoodieException("can not create writable v2 class!", e); + } + } + + /** + * Get date writeable object from int value. + * Hive3 use DateWritableV2 to build date objects and Hive2 use DateWritable. + * So that we need to initialize date according to the version of Hive. + */ + public static Writable getDateWriteable(int value) { + try { + return (Writable) DATE_WRITEABLE_V2_CONSTRUCTOR.newInstance(value); + } catch (IllegalAccessException | InstantiationException | InvocationTargetException e) { + throw new HoodieException("can not create writable v2 class!", e); + } + } + + public static int getDays(Object dateWritable) { + try { + return (int)GET_DAYS.invoke(dateWritable); + } catch (IllegalAccessException | InvocationTargetException e) { + throw new HoodieException("can not create writable v2 class!", e); + } + } + + public static long getMills(Object timestamp) { + try { + return ((Timestamp) TO_SQL_TIMESTAMP.invoke(timestamp)).getTime(); + } catch (IllegalAccessException | InvocationTargetException e) { + throw new HoodieException("can not create writable v2 class!", e); + } + } +} diff --git a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/TestHoodieParquetInputFormat.java b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/TestHoodieParquetInputFormat.java index ccc57d0f6185c..32690f12096d1 100644 --- a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/TestHoodieParquetInputFormat.java +++ b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/TestHoodieParquetInputFormat.java @@ -19,8 +19,12 @@ package org.apache.hudi.hadoop; import org.apache.avro.Schema; +import org.apache.avro.generic.GenericData; import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.ql.io.IOConstants; import org.apache.hadoop.io.ArrayWritable; +import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.mapred.FileInputFormat; import org.apache.hadoop.mapred.InputSplit; @@ -50,6 +54,9 @@ import org.apache.hudi.hadoop.testutils.InputFormatTestUtil; import org.apache.hudi.hadoop.utils.HoodieHiveUtils; import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils; + +import org.apache.hive.common.util.HiveVersionInfo; +import org.apache.parquet.avro.AvroParquetWriter; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; @@ -60,11 +67,17 @@ import java.io.IOException; import java.nio.charset.StandardCharsets; import java.nio.file.Paths; +import java.sql.Timestamp; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.ZoneOffset; import java.util.ArrayList; import java.util.Collections; import java.util.List; import static org.apache.hudi.common.testutils.SchemaTestUtil.getSchemaFromResource; +import static org.apache.hudi.hadoop.HoodieColumnProjectionUtils.READ_COLUMN_NAMES_CONF_STR; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -704,4 +717,57 @@ private void ensureRecordsInCommit(String msg, String commit, int expectedNumber assertEquals(expectedNumberOfRecordsInCommit, actualCount, msg); assertEquals(totalExpected, totalCount, msg); } + + @Test + public void testHoodieParquetInputFormatReadTimeType() throws IOException { + long testTimestampLong = System.currentTimeMillis(); + int testDate = 19116;// 2022-05-04 + + Schema schema = SchemaTestUtil.getSchemaFromResource(getClass(), "/test_timetype.avsc"); + String commit = "20160628071126"; + HoodieTestUtils.init(HoodieTestUtils.getDefaultHadoopConf(), basePath.toString(), + HoodieTableType.COPY_ON_WRITE, HoodieFileFormat.PARQUET); + java.nio.file.Path partitionPath = basePath.resolve(Paths.get("2016", "06", "28")); + String fileId = FSUtils.makeBaseFileName(commit, "1-0-1", "fileid1", + HoodieFileFormat.PARQUET.getFileExtension()); + try (AvroParquetWriter parquetWriter = new AvroParquetWriter( + new Path(partitionPath.resolve(fileId).toString()), schema)) { + GenericData.Record record = new GenericData.Record(schema); + record.put("test_timestamp", testTimestampLong * 1000); + record.put("test_long", testTimestampLong * 1000); + record.put("test_date", testDate); + record.put("_hoodie_commit_time", commit); + record.put("_hoodie_commit_seqno", commit + 1); + parquetWriter.write(record); + } + + jobConf.set(IOConstants.COLUMNS, "test_timestamp,test_long,test_date,_hoodie_commit_time,_hoodie_commit_seqno"); + jobConf.set(IOConstants.COLUMNS_TYPES, "timestamp,bigint,date,string,string"); + jobConf.set(READ_COLUMN_NAMES_CONF_STR, "test_timestamp,test_long,test_date,_hoodie_commit_time,_hoodie_commit_seqno"); + InputFormatTestUtil.setupPartition(basePath, partitionPath); + InputFormatTestUtil.commit(basePath, commit); + FileInputFormat.setInputPaths(jobConf, partitionPath.toFile().getPath()); + + InputSplit[] splits = inputFormat.getSplits(jobConf, 1); + for (InputSplit split : splits) { + RecordReader recordReader = inputFormat + .getRecordReader(split, jobConf, null); + NullWritable key = recordReader.createKey(); + ArrayWritable writable = recordReader.createValue(); + while (recordReader.next(key, writable)) { + // test timestamp + if (HiveVersionInfo.getShortVersion().startsWith("3")) { + LocalDateTime localDateTime = LocalDateTime.ofInstant( + Instant.ofEpochMilli(testTimestampLong), ZoneOffset.UTC); + assertEquals(Timestamp.valueOf(localDateTime).toString(), String.valueOf(writable.get()[0])); + } else { + assertEquals(new Timestamp(testTimestampLong).toString(), String.valueOf(writable.get()[0])); + } + // test long + assertEquals(testTimestampLong * 1000, ((LongWritable) writable.get()[1]).get()); + // test date + assertEquals(LocalDate.ofEpochDay(testDate).toString(), String.valueOf(writable.get()[2])); + } + } + } } diff --git a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/testutils/InputFormatTestUtil.java b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/testutils/InputFormatTestUtil.java index a34849c04c879..c79fe436f958a 100644 --- a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/testutils/InputFormatTestUtil.java +++ b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/testutils/InputFormatTestUtil.java @@ -480,7 +480,7 @@ public static void setPropsForInputFormat(JobConf jobConf, jobConf.addResource(conf); } - private static void setupPartition(java.nio.file.Path basePath, java.nio.file.Path partitionPath) throws IOException { + public static void setupPartition(java.nio.file.Path basePath, java.nio.file.Path partitionPath) throws IOException { Files.createDirectories(partitionPath); // Create partition metadata to properly setup table's partition diff --git a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/utils/TestHiveAvroSerializer.java b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/utils/TestHiveAvroSerializer.java index c71221a242832..bb580d801d499 100644 --- a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/utils/TestHiveAvroSerializer.java +++ b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/utils/TestHiveAvroSerializer.java @@ -77,12 +77,12 @@ public void testSerialize() { avroRecord.put("col4", HoodieAvroUtils.DECIMAL_CONVERSION.toFixed(bd, currentDecimalType, currentDecimalType.getLogicalType())); avroRecord.put("col5", "2011-01-01"); avroRecord.put("col6", 18987); - avroRecord.put("col7", 1640491505000000L); + avroRecord.put("col7", 1640491505111222L); avroRecord.put("col8", false); ByteBuffer bb = ByteBuffer.wrap(new byte[]{97, 48, 53}); avroRecord.put("col9", bb); assertTrue(GenericData.get().validate(avroSchema, avroRecord)); - ArrayWritable writable = (ArrayWritable) HoodieRealtimeRecordReaderUtils.avroToArrayWritable(avroRecord, avroSchema); + ArrayWritable writable = (ArrayWritable) HoodieRealtimeRecordReaderUtils.avroToArrayWritable(avroRecord, avroSchema, true); List writableList = Arrays.stream(writable.get()).collect(Collectors.toList()); writableList.remove(writableList.size() - 1); @@ -99,7 +99,6 @@ public void testSerialize() { StructTypeInfo rowTypeInfoClip = (StructTypeInfo) TypeInfoFactory.getStructTypeInfo(columnNameListClip, columnTypeListClip); GenericRecord testRecordClip = new HiveAvroSerializer(new ArrayWritableObjectInspector(rowTypeInfoClip), columnNameListClip, columnTypeListClip).serialize(clipWritable, avroSchema); assertTrue(GenericData.get().validate(avroSchema, testRecordClip)); - } @Test @@ -114,7 +113,7 @@ public void testNestedValueSerialize() { avroRecord.put("student", studentRecord); assertTrue(GenericData.get().validate(nestedSchema, avroRecord)); - ArrayWritable writable = (ArrayWritable) HoodieRealtimeRecordReaderUtils.avroToArrayWritable(avroRecord, nestedSchema); + ArrayWritable writable = (ArrayWritable) HoodieRealtimeRecordReaderUtils.avroToArrayWritable(avroRecord, nestedSchema, true); List columnTypeList = createHiveTypeInfoFrom("string,string,struct"); List columnNameList = createHiveColumnsFrom("firstname,lastname,student"); diff --git a/hudi-hadoop-mr/src/test/resources/test_timetype.avsc b/hudi-hadoop-mr/src/test/resources/test_timetype.avsc new file mode 100644 index 0000000000000..b1d5ad4377823 --- /dev/null +++ b/hudi-hadoop-mr/src/test/resources/test_timetype.avsc @@ -0,0 +1,45 @@ +/* + * 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. + */ +{ + "type" : "record", + "name" : "testTimestampRecord", + "fields" : [ { + "name" : "test_timestamp", + "type" : { + "type" : "long", + "logicalType" : "timestamp-micros" + } + },{ + "name" : "test_long", + "type" : "long" + }, + { + "name" : "test_date", + "type" : { + "type": "int", + "logicalType": "date" + } + }, + { + "name" : "_hoodie_commit_time", + "type" : "string" + }, { + "name" : "_hoodie_commit_seqno", + "type" : "string" + }] +} \ No newline at end of file From 1bb181dc44194a8b9f3fe4378e58c9ec228550a3 Mon Sep 17 00:00:00 2001 From: xicm Date: Fri, 26 May 2023 15:39:13 +0800 Subject: [PATCH 2/3] fix check style --- .../org/apache/hudi/hadoop/utils/HoodieHiveUtils.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieHiveUtils.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieHiveUtils.java index 9da27eaf91b40..c496b04b481f8 100644 --- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieHiveUtils.java +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieHiveUtils.java @@ -80,7 +80,7 @@ public class HoodieHiveUtils { public static final Pattern HOODIE_CONSUME_MODE_PATTERN_STRING = Pattern.compile("hoodie\\.(.*)\\.consume\\.mode"); public static final String GLOBALLY_CONSISTENT_READ_TIMESTAMP = "last_replication_timestamp"; - private static final boolean isHive3 = isHive3(); + private static final boolean IS_HIVE3 = isHive3(); public static boolean shouldIncludePendingCommits(JobConf job, String tableName) { return job.getBoolean(String.format(HOODIE_CONSUME_PENDING_COMMITS, tableName), false); @@ -169,7 +169,7 @@ public static boolean isHive2() { * So that we need to initialize timestamp according to the version of Hive. */ public static Writable getTimestampWriteable(long value, boolean timestampMillis) { - return isHive3 ? Hive3Shims.getTimestampWriteable(value, timestampMillis) : Hive2Shims.getTimestampWriteable(value, timestampMillis); + return IS_HIVE3 ? Hive3Shims.getTimestampWriteable(value, timestampMillis) : Hive2Shims.getTimestampWriteable(value, timestampMillis); } /** @@ -178,14 +178,14 @@ public static Writable getTimestampWriteable(long value, boolean timestampMillis * So that we need to initialize date according to the version of Hive. */ public static Writable getDateWriteable(int value) { - return isHive3 ? Hive3Shims.getDateWriteable(value) : Hive2Shims.getDateWriteable(value); + return IS_HIVE3 ? Hive3Shims.getDateWriteable(value) : Hive2Shims.getDateWriteable(value); } public static int getDays(Object dateWritable) { - return isHive3 ? Hive3Shims.getDays(dateWritable) : Hive2Shims.getDays(dateWritable); + return IS_HIVE3 ? Hive3Shims.getDays(dateWritable) : Hive2Shims.getDays(dateWritable); } public static long getMills(Object timestamp) { - return isHive3 ? Hive3Shims.getMills(timestamp) : Hive2Shims.getMills(timestamp); + return IS_HIVE3 ? Hive3Shims.getMills(timestamp) : Hive2Shims.getMills(timestamp); } } From cbeb002446bb52c72ccde03d280ffdd20d7c6923 Mon Sep 17 00:00:00 2001 From: xicm Date: Fri, 26 May 2023 23:07:57 +0800 Subject: [PATCH 3/3] abstract hive shims --- .../hadoop/HoodieColumnProjectionUtils.java | 21 ++++++----- .../hudi/hadoop/HoodieParquetInputFormat.java | 4 +-- ...oodieTimestampAwareParquetInputFormat.java | 4 +-- .../hudi/hadoop/utils/HoodieHiveUtils.java | 14 ++++---- .../shims/{Hive2Shims.java => Hive2Shim.java} | 21 ++++++++--- .../shims/{Hive3Shims.java => Hive3Shim.java} | 32 +++++++++++------ .../hudi/hadoop/utils/shims/HiveShim.java | 35 +++++++++++++++++++ .../hudi/hadoop/utils/shims/HiveShims.java | 28 +++++++++++++++ 8 files changed, 125 insertions(+), 34 deletions(-) rename hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/{Hive2Shims.java => Hive2Shim.java} (74%) rename hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/{Hive3Shims.java => Hive3Shim.java} (85%) create mode 100644 hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/HiveShim.java create mode 100644 hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/HiveShims.java diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieColumnProjectionUtils.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieColumnProjectionUtils.java index a2134643c3805..1f08fcde97022 100644 --- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieColumnProjectionUtils.java +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieColumnProjectionUtils.java @@ -110,21 +110,24 @@ public static List> getIOColumnNameAndTypes(Configuration co } /** - * if schema contains timestamp columns, this method is used for compatibility when there is no timestamp fields - * We expect 3 cases to use parquet-avro reader to read timestamp column: - * 1. read columns contain timestamp type - * 2. no read columns and exists original columns contain timestamp type - * 3. no read columns and no original columns, but avro schema contains type + * If schema contains timestamp columns, this method is used for compatibility when there is no timestamp fields. + * + *

We expect 3 cases to use parquet-avro reader {@link org.apache.hudi.hadoop.avro.HoodieAvroParquetReader} to read timestamp column: + * + *

    + *
  1. Read columns contain timestamp type;
    2. + *
  2. Empty original columns;
    3. + *
  3. Empty read columns but existing original columns contain timestamp type.
    4. + *
*/ public static boolean supportTimestamp(Configuration conf) { - List reads = Arrays.asList(getReadColumnNames(conf)); - if (reads.isEmpty()) { + List readCols = Arrays.asList(getReadColumnNames(conf)); + if (readCols.isEmpty()) { return getIOColumnTypes(conf).contains("timestamp"); } List names = getIOColumns(conf); List types = getIOColumnTypes(conf); - return types.isEmpty() || IntStream.range(0, names.size()).filter(i -> reads.contains(names.get(i))) + return types.isEmpty() || IntStream.range(0, names.size()).filter(i -> readCols.contains(names.get(i))) .anyMatch(i -> types.get(i).equals("timestamp")); } - } \ No newline at end of file diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieParquetInputFormat.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieParquetInputFormat.java index e9c013a316c97..8a26a5c139c0e 100644 --- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieParquetInputFormat.java +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieParquetInputFormat.java @@ -74,9 +74,9 @@ protected HoodieParquetInputFormat(HoodieCopyOnWriteTableInputFormat delegate) { /** * Spark2 use `parquet.hadoopParquetInputFormat` in `com.twitter:parquet-hadoop-bundle`. * So that we need to distinguish the constructions of classes with - * `parquet.hadoopParquetInputFormat` or `org.apache.parquet.hadoop.ParquetInputFormat`. + * `parquet.hadoopParquetInputFormat` or `org.apache.parquet.hadoop.ParquetInputFormat`. * If we use `org.apache.parquet:parquet-hadoop`, we can use `HudiAvroParquetInputFormat` - * in Hive or Spark3 to get timestamp with correct type. + * in Hive or Spark3 to get timestamp with correct type. */ private void initAvroInputFormat() { try { diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/avro/HoodieTimestampAwareParquetInputFormat.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/avro/HoodieTimestampAwareParquetInputFormat.java index ac3a9e735b8a1..8f9aae530e412 100644 --- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/avro/HoodieTimestampAwareParquetInputFormat.java +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/avro/HoodieTimestampAwareParquetInputFormat.java @@ -30,14 +30,14 @@ /** * To resolve issue Fix Timestamp/Date type read by Hive3, - * we need to handle timestamp types separately based on the parquet-avro approach + * we need to handle timestamp types separately based on the parquet-avro approach. */ public class HoodieTimestampAwareParquetInputFormat extends ParquetInputFormat { @Override public RecordReader createRecordReader( InputSplit inputSplit, - TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException { + TaskAttemptContext taskAttemptContext) throws IOException { Configuration conf = ContextUtil.getConfiguration(taskAttemptContext); return new HoodieAvroParquetReader(inputSplit, conf); } diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieHiveUtils.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieHiveUtils.java index c496b04b481f8..e257f96e44b26 100644 --- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieHiveUtils.java +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieHiveUtils.java @@ -20,8 +20,8 @@ import org.apache.hudi.common.util.CollectionUtils; import org.apache.hudi.common.util.Option; -import org.apache.hudi.hadoop.utils.shims.Hive2Shims; -import org.apache.hudi.hadoop.utils.shims.Hive3Shims; +import org.apache.hudi.hadoop.utils.shims.HiveShim; +import org.apache.hudi.hadoop.utils.shims.HiveShims; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; @@ -82,6 +82,8 @@ public class HoodieHiveUtils { private static final boolean IS_HIVE3 = isHive3(); + private static final HiveShim HIVE_SHIM = HiveShims.getInstance(IS_HIVE3); + public static boolean shouldIncludePendingCommits(JobConf job, String tableName) { return job.getBoolean(String.format(HOODIE_CONSUME_PENDING_COMMITS, tableName), false); } @@ -169,7 +171,7 @@ public static boolean isHive2() { * So that we need to initialize timestamp according to the version of Hive. */ public static Writable getTimestampWriteable(long value, boolean timestampMillis) { - return IS_HIVE3 ? Hive3Shims.getTimestampWriteable(value, timestampMillis) : Hive2Shims.getTimestampWriteable(value, timestampMillis); + return HIVE_SHIM.getTimestampWriteable(value, timestampMillis); } /** @@ -178,14 +180,14 @@ public static Writable getTimestampWriteable(long value, boolean timestampMillis * So that we need to initialize date according to the version of Hive. */ public static Writable getDateWriteable(int value) { - return IS_HIVE3 ? Hive3Shims.getDateWriteable(value) : Hive2Shims.getDateWriteable(value); + return HIVE_SHIM.getDateWriteable(value); } public static int getDays(Object dateWritable) { - return IS_HIVE3 ? Hive3Shims.getDays(dateWritable) : Hive2Shims.getDays(dateWritable); + return HIVE_SHIM.getDays(dateWritable); } public static long getMills(Object timestamp) { - return IS_HIVE3 ? Hive3Shims.getMills(timestamp) : Hive2Shims.getMills(timestamp); + return HIVE_SHIM.getMills(timestamp); } } diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive2Shims.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive2Shim.java similarity index 74% rename from hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive2Shims.java rename to hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive2Shim.java index 925ec5798313f..b8583799d6c02 100644 --- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive2Shims.java +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive2Shim.java @@ -24,22 +24,33 @@ import java.sql.Timestamp; -public class Hive2Shims { +/** + * Shim clazz for Hive2. + */ +public class Hive2Shim implements HiveShim { + private static final Hive2Shim INSTANCE = new Hive2Shim(); + + private Hive2Shim() { + } + + public static Hive2Shim getInstance() { + return INSTANCE; + } - public static Writable getTimestampWriteable(long value, boolean timestampMillis) { + public Writable getTimestampWriteable(long value, boolean timestampMillis) { Timestamp timestamp = new Timestamp(timestampMillis ? value : value / 1000); return new TimestampWritable(timestamp); } - public static Writable getDateWriteable(int value) { + public Writable getDateWriteable(int value) { return new DateWritable(value); } - public static int getDays(Object dateWritable) { + public int getDays(Object dateWritable) { return ((DateWritable) dateWritable).getDays(); } - public static long getMills(Object timestamp) { + public long getMills(Object timestamp) { return ((Timestamp) timestamp).getTime(); } } diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive3Shims.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive3Shim.java similarity index 85% rename from hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive3Shims.java rename to hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive3Shim.java index 485d0e00ad3f2..0329c464e5490 100644 --- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive3Shims.java +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/Hive3Shim.java @@ -29,22 +29,25 @@ import java.lang.reflect.Method; import java.sql.Timestamp; -public class Hive3Shims { +/** + * Shim clazz for Hive3. + */ +public class Hive3Shim implements HiveShim { - public static final Logger LOG = LoggerFactory.getLogger(Hive3Shims.class); + public static final Logger LOG = LoggerFactory.getLogger(Hive3Shim.class); public static final String HIVE_TIMESTAMP_TYPE_CLASS = "org.apache.hadoop.hive.common.type.Timestamp"; public static final String TIMESTAMP_WRITEABLE_V2_CLASS = "org.apache.hadoop.hive.serde2.io.TimestampWritableV2"; public static final String DATE_WRITEABLE_V2_CLASS = "org.apache.hadoop.hive.serde2.io.DateWritableV2"; - private static Class TIMESTAMP_CLASS = null; + private static Class TIMESTAMP_CLASS = null; private static Method SET_TIME_IN_MILLIS = null; private static Method TO_SQL_TIMESTAMP = null; - private static Constructor TIMESTAMP_WRITEABLE_V2_CONSTRUCTOR = null; + private static Constructor TIMESTAMP_WRITEABLE_V2_CONSTRUCTOR = null; - private static Class DATE_WRITEABLE_CLASS = null; + private static Class DATE_WRITEABLE_CLASS = null; private static Method GET_DAYS = null; - private static Constructor DATE_WRITEABLE_V2_CONSTRUCTOR = null; + private static Constructor DATE_WRITEABLE_V2_CONSTRUCTOR = null; static { // timestamp @@ -67,12 +70,21 @@ public class Hive3Shims { } } + private static final Hive3Shim INSTANCE = new Hive3Shim(); + + private Hive3Shim() { + } + + public static Hive3Shim getInstance() { + return INSTANCE; + } + /** * Get timestamp writeable object from long value. * Hive3 use TimestampWritableV2 to build timestamp objects and Hive2 use TimestampWritable. * So that we need to initialize timestamp according to the version of Hive. */ - public static Writable getTimestampWriteable(long value, boolean timestampMillis) { + public Writable getTimestampWriteable(long value, boolean timestampMillis) { try { Object timestamp = TIMESTAMP_CLASS.newInstance(); SET_TIME_IN_MILLIS.invoke(timestamp, timestampMillis ? value : value / 1000); @@ -87,7 +99,7 @@ public static Writable getTimestampWriteable(long value, boolean timestampMillis * Hive3 use DateWritableV2 to build date objects and Hive2 use DateWritable. * So that we need to initialize date according to the version of Hive. */ - public static Writable getDateWriteable(int value) { + public Writable getDateWriteable(int value) { try { return (Writable) DATE_WRITEABLE_V2_CONSTRUCTOR.newInstance(value); } catch (IllegalAccessException | InstantiationException | InvocationTargetException e) { @@ -95,7 +107,7 @@ public static Writable getDateWriteable(int value) { } } - public static int getDays(Object dateWritable) { + public int getDays(Object dateWritable) { try { return (int)GET_DAYS.invoke(dateWritable); } catch (IllegalAccessException | InvocationTargetException e) { @@ -103,7 +115,7 @@ public static int getDays(Object dateWritable) { } } - public static long getMills(Object timestamp) { + public long getMills(Object timestamp) { try { return ((Timestamp) TO_SQL_TIMESTAMP.invoke(timestamp)).getTime(); } catch (IllegalAccessException | InvocationTargetException e) { diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/HiveShim.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/HiveShim.java new file mode 100644 index 0000000000000..4b67514d24354 --- /dev/null +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/HiveShim.java @@ -0,0 +1,35 @@ +/* + * 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.hudi.hadoop.utils.shims; + +import org.apache.hadoop.io.Writable; + +/** + * Shim class to resolve Hive version compatibility. + */ +public interface HiveShim { + + Writable getTimestampWriteable(long value, boolean timestampMillis); + + Writable getDateWriteable(int value); + + int getDays(Object dateWritable); + + long getMills(Object timestamp); +} diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/HiveShims.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/HiveShims.java new file mode 100644 index 0000000000000..089c7a60b955b --- /dev/null +++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/shims/HiveShims.java @@ -0,0 +1,28 @@ +/* + * 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.hudi.hadoop.utils.shims; + +/** + * Factory clazz for {@link HiveShim}. + */ +public class HiveShims { + public static HiveShim getInstance(boolean hive3) { + return hive3 ? Hive3Shim.getInstance() : Hive2Shim.getInstance(); + } +}