minihippo commented on code in PR #5629:
URL: https://github.com/apache/hudi/pull/5629#discussion_r914608155
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieParquetDataBlock.java:
##########
@@ -93,53 +89,41 @@ protected byte[] serializeRecords(List<HoodieRecord>
records) throws IOException
}
Schema writerSchema = new
Schema.Parser().parse(super.getLogBlockHeader().get(HeaderMetadataType.SCHEMA));
-
- HoodieAvroWriteSupport writeSupport = new HoodieAvroWriteSupport(
- new AvroSchemaConverter().convert(writerSchema), writerSchema,
Option.empty());
-
- HoodieParquetConfig<HoodieAvroWriteSupport> avroParquetConfig =
- new HoodieParquetConfig<>(
- writeSupport,
- compressionCodecName.get(),
- ParquetWriter.DEFAULT_BLOCK_SIZE,
- ParquetWriter.DEFAULT_PAGE_SIZE,
- 1024 * 1024 * 1024,
- new Configuration(),
-
Double.parseDouble(String.valueOf(0.1)));//HoodieStorageConfig.PARQUET_COMPRESSION_RATIO.defaultValue()));
-
ByteArrayOutputStream baos = new ByteArrayOutputStream();
-
try (FSDataOutputStream outputStream = new FSDataOutputStream(baos)) {
- try (HoodieParquetStreamWriter parquetWriter = new
HoodieParquetStreamWriter(outputStream, avroParquetConfig)) {
- for (HoodieRecord record : records) {
+ HoodieFileWriter parquetWriter = null;
+ HoodieStorageConfig storageConfig =
HoodieStorageConfig.newBuilder().build();
Review Comment:
A better way to do it
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkParquetReader.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.io.storage;
+
+import org.apache.avro.Schema;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.HoodieInternalRowUtils;
+import org.apache.hudi.common.bloom.BloomFilter;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.util.BaseFileUtils;
+import org.apache.hudi.common.util.ClosableIterator;
+import org.apache.hudi.common.util.ParquetReaderIterator;
+import org.apache.parquet.hadoop.ParquetReader;
+import org.apache.parquet.hadoop.api.ReadSupport;
+import org.apache.parquet.hadoop.util.HadoopInputFile;
+import org.apache.parquet.io.InputFile;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.execution.datasources.parquet.ParquetReadSupport;
+import org.apache.spark.sql.internal.SQLConf;
+import org.apache.spark.sql.types.StructType;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+public class HoodieSparkParquetReader implements HoodieSparkFileReader {
+
+ private final Path path;
+ private final Configuration conf;
+ private final BaseFileUtils parquetUtils;
+ private List<ParquetReaderIterator> readerIterators = new ArrayList<>();
+
+ public HoodieSparkParquetReader(Configuration conf, Path path) {
+ this.path = path;
+ this.conf = conf;
+ this.parquetUtils = BaseFileUtils.getInstance(HoodieFileFormat.PARQUET);
+ }
+
+ @Override
+ public String[] readMinMaxRecordKeys() {
+ return parquetUtils.readMinMaxRecordKeys(conf, path);
+ }
+
+ @Override
+ public BloomFilter readBloomFilter() {
+ return parquetUtils.readBloomFilterFromMetadata(conf, path);
+ }
+
+ @Override
+ public Set<String> filterRowKeys(Set<String> candidateRowKeys) {
+ return parquetUtils.filterRowKeys(conf, path, candidateRowKeys);
+ }
+
+ @Override
+ public ClosableIterator<InternalRow> getInternalRowIterator(Schema schema)
throws IOException {
+ StructType structType = HoodieInternalRowUtils.getCachedSchema(schema);
+ conf.set(ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA(),
structType.json());
+ // todo: get it from spark context
+ conf.setBoolean(SQLConf.PARQUET_BINARY_AS_STRING().key(),false);
+ conf.setBoolean(SQLConf.PARQUET_INT96_AS_TIMESTAMP().key(), true);
+ InputFile inputFile = HadoopInputFile.fromPath(path, conf);
+ ParquetReader reader = new ParquetReader.Builder<InternalRow>(inputFile) {
+ @Override
+ protected ReadSupport getReadSupport() {
+ return new ParquetReadSupport();
+ }
+ }.withConf(conf).build();
+ ParquetReaderIterator<InternalRow> parquetReaderIterator = new
ParquetReaderIterator<>(reader, InternalRow::copy);
Review Comment:
`InternalRow::copy` any better way to solve the problem
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/commmon/model/HoodieSparkRecord.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * 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.commmon.model;
+
+import org.apache.hudi.HoodieInternalRowUtils;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieOperation;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.keygen.BaseKeyGenerator;
+import org.apache.hudi.keygen.SparkKeyGeneratorInterface;
+import org.apache.hudi.keygen.factory.HoodieSparkKeyGeneratorFactory;
+import org.apache.hudi.util.HoodieSparkRecordUtils;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.spark.sql.catalyst.CatalystTypeConverters;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import scala.Tuple2;
+
+import static
org.apache.hudi.common.table.HoodieTableConfig.POPULATE_META_FIELDS;
+import static org.apache.spark.sql.types.DataTypes.BooleanType;
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/**
+ * Spark Engine-specific Implementations of `HoodieRecord`.
+ */
+public class HoodieSparkRecord extends HoodieRecord<InternalRow> {
+
+ // IndexedRecord hold its schema, InternalRow should also hold its schema
+ private final StructType structType;
+
+ public HoodieSparkRecord(InternalRow data, StructType schema) {
+ super(null, data);
+ this.structType = schema;
+ }
+
+ public HoodieSparkRecord(InternalRow data, StructType schema, Comparable
orderingVal) {
+ super(null, data, orderingVal);
+ this.structType = schema;
+ }
+
+ public HoodieSparkRecord(HoodieKey key, InternalRow data, StructType schema)
{
+ super(key, data);
+ this.structType = schema;
+ }
+
+ public HoodieSparkRecord(HoodieKey key, InternalRow data, StructType schema,
Comparable orderingVal) {
+ super(key, data, orderingVal);
+ this.structType = schema;
+ }
+
+ public HoodieSparkRecord(HoodieKey key, InternalRow data, StructType schema,
HoodieOperation operation, Comparable orderingVal) {
+ super(key, data, operation, orderingVal);
+ this.structType = schema;
+ }
+
+ public HoodieSparkRecord(HoodieSparkRecord record) {
+ super(record);
+ this.structType = record.structType;
+ }
+
+ @Override
+ public HoodieRecord<InternalRow> newInstance() {
+ return new HoodieSparkRecord(this);
+ }
+
+ @Override
+ public HoodieRecord<InternalRow> newInstance(HoodieKey key, HoodieOperation
op) {
+ return new HoodieSparkRecord(key, data, structType, op,
getOrderingValue());
+ }
+
+ @Override
+ public HoodieRecord<InternalRow> newInstance(HoodieKey key) {
+ return new HoodieSparkRecord(key, data, structType, getOrderingValue());
+ }
+
+ @Override
+ public void deflate() {
+ }
+
+ @Override
+ public String getRecordKey(Option<BaseKeyGenerator> keyGeneratorOpt) {
+ if (key != null) {
+ return getRecordKey();
+ }
+ return keyGeneratorOpt.isPresent() ? ((SparkKeyGeneratorInterface)
keyGeneratorOpt.get()).getRecordKey(data, structType) :
data.getString(HoodieMetadataField.RECORD_KEY_METADATA_FIELD.ordinal());
+ }
+
+ @Override
+ public String getRecordKey(String keyFieldName) {
+ if (key != null) {
+ return getRecordKey();
+ }
+ Tuple2<StructField, Object> tuple2 =
HoodieInternalRowUtils.getCachedSchemaPosMap(structType).get(keyFieldName).get();
+ DataType dataType = tuple2._1.dataType();
+ int pos = (Integer) tuple2._2;
+ return data.get(pos, dataType).toString();
+ }
+
+ @Override
+ public HoodieRecordType getRecordType() {
+ return HoodieRecordType.SPARK;
+ }
+
+ @Override
+ public Object getRecordColumnValues(String[] columns, Schema schema, boolean
consistentLogicalTimestampEnabled) {
+ return HoodieSparkRecordUtils.getRecordColumnValues(this, columns,
structType, consistentLogicalTimestampEnabled);
+ }
+
+ @Override
+ public HoodieRecord mergeWith(Schema schema, HoodieRecord other, Schema
otherSchema, Schema writerSchema) throws IOException {
+ StructType otherStructType =
HoodieInternalRowUtils.getCachedSchema(otherSchema);
+ StructType writerStructType =
HoodieInternalRowUtils.getCachedSchema(writerSchema);
+ InternalRow mergeRow = HoodieInternalRowUtils.stitchRecords(data,
structType, (InternalRow) other.getData(), otherStructType, writerStructType);
+ return new HoodieSparkRecord(getKey(), mergeRow, writerStructType,
getOperation());
+ }
+
+ @Override
+ public HoodieRecord rewriteRecord(Schema recordSchema, Schema targetSchema,
TypedProperties props) throws IOException {
+ StructType targetStructType =
HoodieInternalRowUtils.getCachedSchema(targetSchema);
+ InternalRow rewriteRow = HoodieInternalRowUtils.rewriteRecord(data,
structType, targetStructType);
+ return new HoodieSparkRecord(getKey(), rewriteRow, targetStructType,
getOperation());
+ }
+
+ @Override
+ public HoodieRecord rewriteRecord(Schema recordSchema, Properties prop,
boolean schemaOnReadEnabled, Schema writeSchemaWithMetaFields) throws
IOException {
+ StructType writeSchemaWithMetaFieldsStructType =
HoodieInternalRowUtils.getCachedSchema(writeSchemaWithMetaFields);
+ InternalRow rewriteRow = schemaOnReadEnabled ?
HoodieInternalRowUtils.rewriteRecordWithNewSchema(data, structType,
writeSchemaWithMetaFieldsStructType, new HashMap<>())
+ : HoodieInternalRowUtils.rewriteRecord(data, structType,
writeSchemaWithMetaFieldsStructType);
+ return new HoodieSparkRecord(getKey(), rewriteRow,
writeSchemaWithMetaFieldsStructType, getOperation());
+ }
+
+ @Override
+ public HoodieRecord rewriteRecordWithMetadata(Schema recordSchema,
Properties prop, boolean schemaOnReadEnabled, Schema writeSchemaWithMetaFields,
String fileName) throws IOException {
+ StructType writeSchemaWithMetaFieldsStructType =
HoodieInternalRowUtils.getCachedSchema(writeSchemaWithMetaFields);
+ InternalRow rewriteRow = schemaOnReadEnabled ?
HoodieInternalRowUtils.rewriteEvolutionRecordWithMetadata(data, structType,
writeSchemaWithMetaFieldsStructType, fileName)
+ : HoodieInternalRowUtils.rewriteRecordWithMetadata(data, structType,
writeSchemaWithMetaFieldsStructType, fileName);
+ return new HoodieSparkRecord(getKey(), rewriteRow,
writeSchemaWithMetaFieldsStructType, getOperation());
+ }
+
+ @Override
+ public HoodieRecord rewriteRecordWithNewSchema(Schema recordSchema,
Properties prop, Schema newSchema, Map<String, String> renameCols) throws
IOException {
+ StructType newStructType =
HoodieInternalRowUtils.getCachedSchema(newSchema);
+ InternalRow rewriteRow =
HoodieInternalRowUtils.rewriteRecordWithNewSchema(data, structType,
newStructType, renameCols);
+ return new HoodieSparkRecord(getKey(), rewriteRow, newStructType,
getOperation());
+ }
+
+ @Override
+ public HoodieRecord rewriteRecordWithNewSchema(Schema recordSchema,
Properties prop, Schema newSchema) throws IOException {
+ StructType newStructType =
HoodieInternalRowUtils.getCachedSchema(newSchema);
+ InternalRow rewriteRow = HoodieInternalRowUtils.rewriteRecord(data,
structType, newStructType);
+ return new HoodieSparkRecord(getKey(), rewriteRow, structType,
getOperation());
+ }
+
+ @Override
+ public HoodieRecord overrideMetadataFieldValue(Schema recordSchema,
Properties prop, int pos, String newValue) throws IOException {
+ data.update(pos, CatalystTypeConverters.convertToCatalyst(newValue));
+ return this;
+ }
+
+ @Override
+ public HoodieRecord addMetadataValues(Schema recordSchema, Properties prop,
Map<HoodieMetadataField, String> metadataValues) throws IOException {
+ Arrays.stream(HoodieMetadataField.values()).forEach(metadataField -> {
+ String value = metadataValues.get(metadataField);
+ if (value != null) {
+ data.update(recordSchema.getField(metadataField.getFieldName()).pos(),
CatalystTypeConverters.convertToCatalyst(value));
+ }
+ });
+ return this;
+ }
+
+ @Override
+ public HoodieRecord expansion(Schema schema, Properties prop, String
payloadClass,
+ String preCombineField,
+ Option<Pair<String, String>> simpleKeyGenFieldsOpt,
+ Boolean withOperation,
+ Option<String> partitionNameOp,
+ Option<Boolean> populateMetaFieldsOp) {
+ boolean populateMetaFields = populateMetaFieldsOp.orElse(false);
+ if (populateMetaFields) {
+ return HoodieSparkRecordUtils.convertToHoodieSparkRecord(structType,
data, preCombineField, withOperation);
+ } else if (simpleKeyGenFieldsOpt.isPresent()) {
+ return HoodieSparkRecordUtils.convertToHoodieSparkRecord(structType,
data, preCombineField, simpleKeyGenFieldsOpt.get(), withOperation,
Option.empty());
+ } else {
+ return HoodieSparkRecordUtils.convertToHoodieSparkRecord(structType,
data, preCombineField, withOperation, partitionNameOp);
+ }
+ }
+
+ @Override
+ public HoodieRecord transform(Schema schema, Properties prop, boolean
useKeygen) {
+ StructType structType = HoodieInternalRowUtils.getCachedSchema(schema);
+ Option<SparkKeyGeneratorInterface> keyGeneratorOpt = Option.empty();
+ if (useKeygen &&
!Boolean.parseBoolean(prop.getOrDefault(POPULATE_META_FIELDS.key(),
POPULATE_META_FIELDS.defaultValue().toString()).toString())) {
+ try {
+ keyGeneratorOpt = Option.of((SparkKeyGeneratorInterface)
HoodieSparkKeyGeneratorFactory.createKeyGenerator(new TypedProperties(prop)));
+ } catch (IOException e) {
+ throw new HoodieException("Only SparkKeyGeneratorInterface are
supported when meta columns are disabled ", e);
+ }
+ }
+ String key = keyGeneratorOpt.isPresent() ?
keyGeneratorOpt.get().getRecordKey(data, structType)
+ : data.get(HoodieMetadataField.RECORD_KEY_METADATA_FIELD.ordinal(),
StringType).toString();
+ String partition = keyGeneratorOpt.isPresent() ?
keyGeneratorOpt.get().getPartitionPath(data, structType)
+ :
data.get(HoodieMetadataField.PARTITION_PATH_METADATA_FIELD.ordinal(),
StringType).toString();
+ this.key = new HoodieKey(key, partition);
+
+ return this;
+ }
+
+ @Override
+ public Option<Map<String, String>> getMetadata() {
+ return Option.empty();
+ }
+
+ @Override
+ public boolean isPresent(Schema schema, Properties prop) throws IOException {
Review Comment:
will fix
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/commmon/model/HoodieSparkRecord.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * 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.commmon.model;
+
+import org.apache.hudi.HoodieInternalRowUtils;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieOperation;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.keygen.BaseKeyGenerator;
+import org.apache.hudi.keygen.SparkKeyGeneratorInterface;
+import org.apache.hudi.keygen.factory.HoodieSparkKeyGeneratorFactory;
+import org.apache.hudi.util.HoodieSparkRecordUtils;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.spark.sql.catalyst.CatalystTypeConverters;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import scala.Tuple2;
+
+import static
org.apache.hudi.common.table.HoodieTableConfig.POPULATE_META_FIELDS;
+import static org.apache.spark.sql.types.DataTypes.BooleanType;
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/**
+ * Spark Engine-specific Implementations of `HoodieRecord`.
+ */
+public class HoodieSparkRecord extends HoodieRecord<InternalRow> {
+
+ // IndexedRecord hold its schema, InternalRow should also hold its schema
+ private final StructType structType;
+
+ public HoodieSparkRecord(InternalRow data, StructType schema) {
+ super(null, data);
+ this.structType = schema;
+ }
+
+ public HoodieSparkRecord(InternalRow data, StructType schema, Comparable
orderingVal) {
+ super(null, data, orderingVal);
+ this.structType = schema;
+ }
+
+ public HoodieSparkRecord(HoodieKey key, InternalRow data, StructType schema)
{
+ super(key, data);
+ this.structType = schema;
+ }
+
+ public HoodieSparkRecord(HoodieKey key, InternalRow data, StructType schema,
Comparable orderingVal) {
+ super(key, data, orderingVal);
+ this.structType = schema;
+ }
+
+ public HoodieSparkRecord(HoodieKey key, InternalRow data, StructType schema,
HoodieOperation operation, Comparable orderingVal) {
+ super(key, data, operation, orderingVal);
+ this.structType = schema;
+ }
+
+ public HoodieSparkRecord(HoodieSparkRecord record) {
+ super(record);
+ this.structType = record.structType;
+ }
+
+ @Override
+ public HoodieRecord<InternalRow> newInstance() {
+ return new HoodieSparkRecord(this);
+ }
+
+ @Override
+ public HoodieRecord<InternalRow> newInstance(HoodieKey key, HoodieOperation
op) {
+ return new HoodieSparkRecord(key, data, structType, op,
getOrderingValue());
+ }
+
+ @Override
+ public HoodieRecord<InternalRow> newInstance(HoodieKey key) {
+ return new HoodieSparkRecord(key, data, structType, getOrderingValue());
+ }
+
+ @Override
+ public void deflate() {
+ }
+
+ @Override
+ public String getRecordKey(Option<BaseKeyGenerator> keyGeneratorOpt) {
+ if (key != null) {
+ return getRecordKey();
+ }
+ return keyGeneratorOpt.isPresent() ? ((SparkKeyGeneratorInterface)
keyGeneratorOpt.get()).getRecordKey(data, structType) :
data.getString(HoodieMetadataField.RECORD_KEY_METADATA_FIELD.ordinal());
+ }
+
+ @Override
+ public String getRecordKey(String keyFieldName) {
+ if (key != null) {
+ return getRecordKey();
+ }
+ Tuple2<StructField, Object> tuple2 =
HoodieInternalRowUtils.getCachedSchemaPosMap(structType).get(keyFieldName).get();
+ DataType dataType = tuple2._1.dataType();
+ int pos = (Integer) tuple2._2;
+ return data.get(pos, dataType).toString();
+ }
+
+ @Override
+ public HoodieRecordType getRecordType() {
+ return HoodieRecordType.SPARK;
+ }
+
+ @Override
+ public Object getRecordColumnValues(String[] columns, Schema schema, boolean
consistentLogicalTimestampEnabled) {
+ return HoodieSparkRecordUtils.getRecordColumnValues(this, columns,
structType, consistentLogicalTimestampEnabled);
+ }
+
+ @Override
+ public HoodieRecord mergeWith(Schema schema, HoodieRecord other, Schema
otherSchema, Schema writerSchema) throws IOException {
+ StructType otherStructType =
HoodieInternalRowUtils.getCachedSchema(otherSchema);
+ StructType writerStructType =
HoodieInternalRowUtils.getCachedSchema(writerSchema);
+ InternalRow mergeRow = HoodieInternalRowUtils.stitchRecords(data,
structType, (InternalRow) other.getData(), otherStructType, writerStructType);
+ return new HoodieSparkRecord(getKey(), mergeRow, writerStructType,
getOperation());
+ }
+
+ @Override
+ public HoodieRecord rewriteRecord(Schema recordSchema, Schema targetSchema,
TypedProperties props) throws IOException {
+ StructType targetStructType =
HoodieInternalRowUtils.getCachedSchema(targetSchema);
+ InternalRow rewriteRow = HoodieInternalRowUtils.rewriteRecord(data,
structType, targetStructType);
+ return new HoodieSparkRecord(getKey(), rewriteRow, targetStructType,
getOperation());
+ }
+
+ @Override
+ public HoodieRecord rewriteRecord(Schema recordSchema, Properties prop,
boolean schemaOnReadEnabled, Schema writeSchemaWithMetaFields) throws
IOException {
+ StructType writeSchemaWithMetaFieldsStructType =
HoodieInternalRowUtils.getCachedSchema(writeSchemaWithMetaFields);
+ InternalRow rewriteRow = schemaOnReadEnabled ?
HoodieInternalRowUtils.rewriteRecordWithNewSchema(data, structType,
writeSchemaWithMetaFieldsStructType, new HashMap<>())
+ : HoodieInternalRowUtils.rewriteRecord(data, structType,
writeSchemaWithMetaFieldsStructType);
+ return new HoodieSparkRecord(getKey(), rewriteRow,
writeSchemaWithMetaFieldsStructType, getOperation());
+ }
+
+ @Override
+ public HoodieRecord rewriteRecordWithMetadata(Schema recordSchema,
Properties prop, boolean schemaOnReadEnabled, Schema writeSchemaWithMetaFields,
String fileName) throws IOException {
+ StructType writeSchemaWithMetaFieldsStructType =
HoodieInternalRowUtils.getCachedSchema(writeSchemaWithMetaFields);
+ InternalRow rewriteRow = schemaOnReadEnabled ?
HoodieInternalRowUtils.rewriteEvolutionRecordWithMetadata(data, structType,
writeSchemaWithMetaFieldsStructType, fileName)
+ : HoodieInternalRowUtils.rewriteRecordWithMetadata(data, structType,
writeSchemaWithMetaFieldsStructType, fileName);
+ return new HoodieSparkRecord(getKey(), rewriteRow,
writeSchemaWithMetaFieldsStructType, getOperation());
+ }
+
+ @Override
+ public HoodieRecord rewriteRecordWithNewSchema(Schema recordSchema,
Properties prop, Schema newSchema, Map<String, String> renameCols) throws
IOException {
+ StructType newStructType =
HoodieInternalRowUtils.getCachedSchema(newSchema);
+ InternalRow rewriteRow =
HoodieInternalRowUtils.rewriteRecordWithNewSchema(data, structType,
newStructType, renameCols);
+ return new HoodieSparkRecord(getKey(), rewriteRow, newStructType,
getOperation());
+ }
+
+ @Override
+ public HoodieRecord rewriteRecordWithNewSchema(Schema recordSchema,
Properties prop, Schema newSchema) throws IOException {
+ StructType newStructType =
HoodieInternalRowUtils.getCachedSchema(newSchema);
+ InternalRow rewriteRow = HoodieInternalRowUtils.rewriteRecord(data,
structType, newStructType);
+ return new HoodieSparkRecord(getKey(), rewriteRow, structType,
getOperation());
+ }
+
+ @Override
+ public HoodieRecord overrideMetadataFieldValue(Schema recordSchema,
Properties prop, int pos, String newValue) throws IOException {
+ data.update(pos, CatalystTypeConverters.convertToCatalyst(newValue));
+ return this;
+ }
+
+ @Override
+ public HoodieRecord addMetadataValues(Schema recordSchema, Properties prop,
Map<HoodieMetadataField, String> metadataValues) throws IOException {
+ Arrays.stream(HoodieMetadataField.values()).forEach(metadataField -> {
+ String value = metadataValues.get(metadataField);
+ if (value != null) {
+ data.update(recordSchema.getField(metadataField.getFieldName()).pos(),
CatalystTypeConverters.convertToCatalyst(value));
+ }
+ });
+ return this;
+ }
+
+ @Override
+ public HoodieRecord expansion(Schema schema, Properties prop, String
payloadClass,
+ String preCombineField,
+ Option<Pair<String, String>> simpleKeyGenFieldsOpt,
+ Boolean withOperation,
+ Option<String> partitionNameOp,
+ Option<Boolean> populateMetaFieldsOp) {
+ boolean populateMetaFields = populateMetaFieldsOp.orElse(false);
+ if (populateMetaFields) {
+ return HoodieSparkRecordUtils.convertToHoodieSparkRecord(structType,
data, preCombineField, withOperation);
+ } else if (simpleKeyGenFieldsOpt.isPresent()) {
+ return HoodieSparkRecordUtils.convertToHoodieSparkRecord(structType,
data, preCombineField, simpleKeyGenFieldsOpt.get(), withOperation,
Option.empty());
+ } else {
+ return HoodieSparkRecordUtils.convertToHoodieSparkRecord(structType,
data, preCombineField, withOperation, partitionNameOp);
+ }
+ }
+
+ @Override
+ public HoodieRecord transform(Schema schema, Properties prop, boolean
useKeygen) {
+ StructType structType = HoodieInternalRowUtils.getCachedSchema(schema);
+ Option<SparkKeyGeneratorInterface> keyGeneratorOpt = Option.empty();
+ if (useKeygen &&
!Boolean.parseBoolean(prop.getOrDefault(POPULATE_META_FIELDS.key(),
POPULATE_META_FIELDS.defaultValue().toString()).toString())) {
+ try {
+ keyGeneratorOpt = Option.of((SparkKeyGeneratorInterface)
HoodieSparkKeyGeneratorFactory.createKeyGenerator(new TypedProperties(prop)));
+ } catch (IOException e) {
+ throw new HoodieException("Only SparkKeyGeneratorInterface are
supported when meta columns are disabled ", e);
+ }
+ }
+ String key = keyGeneratorOpt.isPresent() ?
keyGeneratorOpt.get().getRecordKey(data, structType)
+ : data.get(HoodieMetadataField.RECORD_KEY_METADATA_FIELD.ordinal(),
StringType).toString();
+ String partition = keyGeneratorOpt.isPresent() ?
keyGeneratorOpt.get().getPartitionPath(data, structType)
+ :
data.get(HoodieMetadataField.PARTITION_PATH_METADATA_FIELD.ordinal(),
StringType).toString();
+ this.key = new HoodieKey(key, partition);
+
+ return this;
+ }
+
+ @Override
+ public Option<Map<String, String>> getMetadata() {
+ return Option.empty();
+ }
+
+ @Override
+ public boolean isPresent(Schema schema, Properties prop) throws IOException {
+ if (null == data) {
+ return false;
+ }
+ if (schema.getField(HoodieRecord.HOODIE_IS_DELETED_FIELD) == null) {
+ return true;
+ }
+ Object deleteMarker =
data.get(schema.getField(HoodieRecord.HOODIE_IS_DELETED_FIELD).pos(),
BooleanType);
+ return !(deleteMarker instanceof Boolean && (boolean) deleteMarker);
+ }
+
+ @Override
+ public boolean shouldIgnore(Schema schema, Properties prop) throws
IOException {
+ // TODO SENTINEL should refactor SENTINEL without Avro(GenericRecord)
+ if (null != data && data.equals(SENTINEL)) {
Review Comment:
will fix
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordReader.java:
##########
@@ -380,27 +379,22 @@ private boolean isNewInstantBlock(HoodieLogBlock
logBlock) {
* handle it.
*/
private void processDataBlock(HoodieDataBlock dataBlock, Option<KeySpec>
keySpecOpt) throws Exception {
- HoodieRecord.Mapper mapper = (rec) -> createHoodieRecord(rec,
this.hoodieTableMetaClient.getTableConfig(),
- this.payloadClassFQN, this.preCombineField, this.withOperationField,
this.simpleKeyGenFields, this.partitionName);
-
- try (ClosableIterator<HoodieRecord> recordIterator =
getRecordsIterator(dataBlock, keySpecOpt, mapper)) {
+ try (ClosableIterator<HoodieRecord> recordIterator =
getRecordsIterator(dataBlock, keySpecOpt, recordType)) {
Option<Schema> schemaOption = getMergedSchema(dataBlock);
- Schema finalReadSchema;
- if (recordIterator instanceof RecordIterator) {
- finalReadSchema = ((RecordIterator)
recordIterator).getFinalReadSchema();
- } else {
- finalReadSchema = dataBlock.getSchema();
- }
while (recordIterator.hasNext()) {
HoodieRecord currentRecord = recordIterator.next();
- HoodieRecord record = schemaOption.isPresent()
- ? currentRecord.rewriteRecordWithNewSchema(finalReadSchema, new
Properties(), schemaOption.get(), new HashMap<>(), mapper) : currentRecord;
- processNextRecord(record);
+ HoodieRecord record = schemaOption.isPresent() ?
currentRecord.rewriteRecordWithNewSchema(dataBlock.getSchema(), new
Properties(), schemaOption.get(), new HashMap<>()) : currentRecord;
Review Comment:
unused schema here. Introduced by step1&2, should be deleted
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordScanner.java:
##########
@@ -317,16 +325,36 @@ public Builder withPartition(String partitionName) {
return this;
}
+ @Override
+ public Builder withRecordType(HoodieRecordType type) {
+ this.recordType = type;
+ return this;
+ }
+
+ @Override
+ public Builder withMergeClass(String mergeClass) {
+ this.mergeClass = mergeClass;
+ return this;
+ }
+
@Override
public HoodieMergedLogRecordScanner build() {
if (this.partitionName == null &&
CollectionUtils.nonEmpty(this.logFilePaths)) {
this.partitionName = getRelativePartitionPath(new Path(basePath), new
Path(this.logFilePaths.get(0)).getParent());
}
+ assert recordType != null;
+ assert mergeClass != null;
+
+ if (HoodieTableMetadata.isMetadataTable(basePath)) {
+ recordType = HoodieRecordType.AVRO;
+ mergeClass = HoodieAvroRecordMerge.class.getName();
+ }
Review Comment:
Duplicate code with `HoodieUnMergedLogRecordScanner`
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java:
##########
@@ -135,21 +131,22 @@ public void runMerge(HoodieTable<T,
HoodieData<HoodieRecord<T>>, HoodieData<Hood
readerIterator = getMergingIterator(table, mergeHandle, baseFile,
reader, readSchema, externalSchemaTransformation);
} else {
if (needToReWriteRecord) {
- readerIterator = new
RewriteIterator(reader.getRecordIterator(HoodieAvroIndexedRecord::new),
readSchema, readSchema, table.getConfig().getProps(), renameCols);
+ readerIterator = new RewriteIterator(reader.getRecordIterator(),
readSchema, readSchema, table.getConfig().getProps(), renameCols);
} else {
- readerIterator = reader.getRecordIterator(readSchema,
HoodieAvroIndexedRecord::new);
+ readerIterator = reader.getRecordIterator(readSchema);
}
}
- ThreadLocal<BinaryEncoder> encoderCache = new ThreadLocal<>();
- ThreadLocal<BinaryDecoder> decoderCache = new ThreadLocal<>();
wrapper = new
BoundedInMemoryExecutor(table.getConfig().getWriteBufferLimitBytes(),
readerIterator,
new UpdateHandler(mergeHandle), record -> {
if (!externalSchemaTransformation) {
return record;
}
- // TODO Other type of record need to change
- return transformRecordBasedOnNewSchema(gReader, gWriter, encoderCache,
decoderCache, (GenericRecord) ((HoodieRecord)record).getData());
+ try {
+ return ((HoodieRecord) record).rewriteRecord(writerSchema,
readerSchema, new TypedProperties());
+ } catch (IOException e) {
+ throw new HoodieException(e);
Review Comment:
will fix
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java:
##########
@@ -329,19 +329,16 @@ protected boolean writeRecord(HoodieRecord<T>
hoodieRecord, Option<HoodieRecord>
* Go through an old record. Here if we detect a newer version shows up, we
write the new one to the file.
*/
public void write(HoodieRecord<T> oldRecord) {
- String key = oldRecord.getRecordKey(keyGeneratorOpt);
- boolean copyOldRecord = true;
Schema schema = useWriterSchemaForCompaction ? tableSchemaWithMetaFields :
tableSchema;
+ boolean copyOldRecord = true;
+ String key = oldRecord.getRecordKey(keyGeneratorOpt);
Review Comment:
Revert the changes
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileWriter.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.io.storage;
+
+import org.apache.avro.Schema;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.spark.sql.catalyst.CatalystTypeConverters;
+import org.apache.spark.sql.catalyst.InternalRow;
+
+import java.io.IOException;
+import java.util.Properties;
+
+public interface HoodieSparkFileWriter extends HoodieFileWriter {
+ boolean canWrite();
+
+ void close() throws IOException;
+
+ void writeRowWithMetadata(HoodieKey recordKey, InternalRow row) throws
IOException;
+
+ void writeRow(String recordKey, InternalRow row) throws IOException;
+
+ @Override
+ default void write(String recordKey, HoodieRecord record, Schema schema,
Properties props) throws IOException {
+ writeRow(recordKey, (InternalRow) record.getData());
+ }
+
+ @Override
+ default void writeWithMetadata(HoodieKey key, HoodieRecord record, Schema
schema, Properties props) throws IOException {
+ writeRowWithMetadata(key, (InternalRow) record.getData());
+ }
+
+ default InternalRow prepRecordWithMetadata(HoodieKey key, InternalRow row,
String instantTime, Integer partitionId, long recordIndex, String fileName) {
+ String seqId = HoodieRecord.generateSequenceId(instantTime, partitionId,
recordIndex);
+
row.update(HoodieRecord.HoodieMetadataField.COMMIT_TIME_METADATA_FIELD.ordinal(),
CatalystTypeConverters.convertToCatalyst(instantTime));
+
row.update(HoodieRecord.HoodieMetadataField.COMMIT_SEQNO_METADATA_FIELD.ordinal(),
CatalystTypeConverters.convertToCatalyst(seqId));
+
row.update(HoodieRecord.HoodieMetadataField.RECORD_KEY_METADATA_FIELD.ordinal(),
CatalystTypeConverters.convertToCatalyst(key.getRecordKey()));
+
row.update(HoodieRecord.HoodieMetadataField.PARTITION_PATH_METADATA_FIELD.ordinal(),
CatalystTypeConverters.convertToCatalyst(key.getPartitionPath()));
+
row.update(HoodieRecord.HoodieMetadataField.FILENAME_METADATA_FIELD.ordinal(),
CatalystTypeConverters.convertToCatalyst(fileName));
+ return row;
+ // Object[] metadata = {instantTime, seqId, key.getRecordKey(),
key.getPartitionPath(), fileName};
+ // InternalRow metadataRow = new GenericInternalRow(Arrays.stream(metadata)
+ // .map(o -> CatalystTypeConverters.convertToCatalyst(o)).toArray());
+ // return new JoinedRow(metadataRow, row);
Review Comment:
will fix
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordScanner.java:
##########
@@ -189,8 +188,13 @@ protected void processNextDeletedRecord(DeleteRecord
deleteRecord) {
}
}
// Put the DELETE record
- records.put(key, SpillableMapUtils.generateEmptyPayload(key,
- deleteRecord.getPartitionPath(), deleteRecord.getOrderingValue(),
getPayloadClassFQN()));
+ if (recordType == HoodieRecordType.AVRO) {
+ records.put(key, SpillableMapUtils.generateEmptyPayload(key,
+ deleteRecord.getPartitionPath(), deleteRecord.getOrderingValue(),
getPayloadClassFQN()));
+ } else {
+ HoodieEmptyRecord record = new HoodieEmptyRecord<>(new HoodieKey(key,
deleteRecord.getPartitionPath()), deleteRecord.getOrderingValue(), recordType);
Review Comment:
moving into `SpillableMapUtils.generateEmptyPayload` is restraint
##########
hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/HoodieSparkDefaultRecordMerge.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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;
+
+import org.apache.hudi.common.model.HoodieEmptyRecord;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType;
+import org.apache.hudi.common.util.Option;
+
+import org.apache.avro.Schema;
+
+import java.io.IOException;
+import java.util.Properties;
+
+public class HoodieSparkDefaultRecordMerge extends HoodieSparkRecordMerge {
+
+ @Override
+ public Option<HoodieRecord> combineAndGetUpdateValue(HoodieRecord older,
HoodieRecord newer, Schema schema, Properties props) throws IOException {
+ assert older.getRecordType() == HoodieRecordType.SPARK;
+ assert newer.getRecordType() == HoodieRecordType.SPARK;
+
+ // Null check is needed here to support schema evolution. The record in
storage may be from old schema where
+ // the new ordering column might not be present and hence returns null.
+ if (!needUpdatingPersistedRecord(older, newer, props)) {
Review Comment:
Cloud precombine field be changed？
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/RDDSpatialCurveSortPartitioner.java:
##########
@@ -54,42 +63,64 @@
private final SerializableSchema schema;
private final HoodieClusteringConfig.LayoutOptimizationStrategy
layoutOptStrategy;
private final HoodieClusteringConfig.SpatialCurveCompositionStrategyType
curveCompositionStrategyType;
+ private final HoodieRecordType recordType;
public RDDSpatialCurveSortPartitioner(HoodieSparkEngineContext
sparkEngineContext,
- String[] orderByColumns,
-
HoodieClusteringConfig.LayoutOptimizationStrategy layoutOptStrategy,
-
HoodieClusteringConfig.SpatialCurveCompositionStrategyType
curveCompositionStrategyType,
- Schema schema) {
+ String[] orderByColumns,
+ LayoutOptimizationStrategy layoutOptStrategy,
+ SpatialCurveCompositionStrategyType curveCompositionStrategyType,
+ Schema schema, HoodieRecordType recordType) {
this.sparkEngineContext = sparkEngineContext;
this.orderByColumns = orderByColumns;
this.layoutOptStrategy = layoutOptStrategy;
this.curveCompositionStrategyType = curveCompositionStrategyType;
this.schema = new SerializableSchema(schema);
+ this.recordType = recordType;
}
@Override
public JavaRDD<HoodieRecord<T>> repartitionRecords(JavaRDD<HoodieRecord<T>>
records, int outputSparkPartitions) {
- JavaRDD<GenericRecord> genericRecordsRDD =
- records.map(f -> (GenericRecord) f.toIndexedRecord(schema.get(), new
Properties()).get());
-
- Dataset<Row> sourceDataset =
- AvroConversionUtils.createDataFrame(
- genericRecordsRDD.rdd(),
- schema.toString(),
- sparkEngineContext.getSqlContext().sparkSession()
- );
-
- Dataset<Row> sortedDataset = reorder(sourceDataset, outputSparkPartitions);
-
- return HoodieSparkUtils.createRdd(sortedDataset, schema.get().getName(),
schema.get().getNamespace(), false, Option.empty())
- .toJavaRDD()
- .map(record -> {
- String key =
record.get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString();
- String partition =
record.get(HoodieRecord.PARTITION_PATH_METADATA_FIELD).toString();
- HoodieKey hoodieKey = new HoodieKey(key, partition);
- HoodieRecord hoodieRecord = new HoodieAvroRecord(hoodieKey, new
RewriteAvroPayload(record));
- return hoodieRecord;
- });
+ if (recordType == HoodieRecordType.AVRO) {
+ JavaRDD<GenericRecord> genericRecordsRDD =
+ records.map(f -> (GenericRecord) f.toIndexedRecord(schema.get(), new
Properties()).get());
+
+ Dataset<Row> sourceDataset =
+ AvroConversionUtils.createDataFrame(
+ genericRecordsRDD.rdd(),
+ schema.toString(),
+ sparkEngineContext.getSqlContext().sparkSession()
+ );
+
+ Dataset<Row> sortedDataset = reorder(sourceDataset,
outputSparkPartitions);
+
+ return HoodieSparkUtils.createRdd(sortedDataset, schema.get().getName(),
schema.get().getNamespace(), false, Option.empty())
+ .toJavaRDD()
+ .map(record -> {
+ String key =
record.get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString();
+ String partition =
record.get(HoodieRecord.PARTITION_PATH_METADATA_FIELD).toString();
+ HoodieKey hoodieKey = new HoodieKey(key, partition);
+ HoodieRecord hoodieRecord = new HoodieAvroRecord(hoodieKey, new
RewriteAvroPayload(record));
+ return hoodieRecord;
+ });
+ } else if (recordType == HoodieRecordType.SPARK) {
+ StructType structType =
HoodieInternalRowUtils.getCachedSchema(schema.get());
+ Dataset<Row> sourceDataset =
SparkConversionUtils.createDataFrame(records.rdd(),
sparkEngineContext.getSqlContext().sparkSession(), structType);
+
+ Dataset<Row> sortedDataset = reorder(sourceDataset,
outputSparkPartitions);
+
+ return sortedDataset.queryExecution().toRdd()
+ .toJavaRDD()
+ .map(row -> {
+ InternalRow internalRow = row.copy();
Review Comment:
The root cause is in the `ParquetReaderIterator`.
```java
public T next() {
try {
// To handle case when next() is called before hasNext()
if (this.next == null) {
if (!hasNext()) {
throw new HoodieException("No more records left to read from
parquet file");
}
}
T retVal = this.next;
this.next = read();
return retVal;
} catch (Exception e) {
FileIOUtils.closeQuietly(parquetReader);
throw new HoodieException("unable to read next record from parquet
file ", e);
}
}
private T read() throws IOException {
T record = parquetReader.read();
if (mapper == null || record == null) {
return record;
} else {
return mapper.apply(record);
}
}
```
Caused by inner implementations of the internalRow, when go next and read
without copying, the value of `retVal` will be changed meanwhile.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java:
##########
@@ -84,21 +80,21 @@ public void runMerge(HoodieTable<T,
HoodieData<HoodieRecord<T>>, HoodieData<Hood
Configuration cfgForHoodieFile = new Configuration(table.getHadoopConf());
HoodieBaseFile baseFile = mergeHandle.baseFileForMerge();
- final GenericDatumWriter<GenericRecord> gWriter;
- final GenericDatumReader<GenericRecord> gReader;
Schema readSchema;
+ Schema readerSchema;
+ Schema writerSchema;
if (externalSchemaTransformation ||
baseFile.getBootstrapBaseFile().isPresent()) {
- readSchema =
HoodieFileReaderFactory.getFileReader(table.getHadoopConf(),
mergeHandle.getOldFilePath()).getSchema();
- gWriter = new GenericDatumWriter<>(readSchema);
- gReader = new GenericDatumReader<>(readSchema,
mergeHandle.getWriterSchemaWithMetaFields());
+ readSchema =
HoodieFileReaderFactory.getReaderFactory(table.getConfig().getRecordType()).getFileReader(table.getHadoopConf(),
mergeHandle.getOldFilePath()).getSchema();
+ writerSchema = readSchema;
+ readerSchema = mergeHandle.getWriterSchemaWithMetaFields();
} else {
- gReader = null;
- gWriter = null;
+ readerSchema = null;
+ writerSchema = null;
readSchema = mergeHandle.getWriterSchemaWithMetaFields();
}
BoundedInMemoryExecutor<GenericRecord, GenericRecord, Void> wrapper = null;
- HoodieFileReader reader =
HoodieFileReaderFactory.getFileReader(cfgForHoodieFile,
mergeHandle.getOldFilePath());
+ HoodieFileReader reader =
HoodieFileReaderFactory.getReaderFactory(table.getConfig().getRecordType()).getFileReader(cfgForHoodieFile,
mergeHandle.getOldFilePath());
Review Comment:
will fix
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDD.scala:
##########
@@ -302,14 +306,29 @@ class HoodieMergeOnReadRDD(@transient sc: SparkContext,
private def serialize(curRowRecord: InternalRow): GenericRecord =
serializer.serialize(curRowRecord).asInstanceOf[GenericRecord]
- private def merge(curAvroRecord: GenericRecord, newRecord: HoodieRecord[_
<: HoodieRecordPayload[_]]): Option[IndexedRecord] = {
+ private def merge(curRow: InternalRow, newRecord: HoodieRecord[_]):
Option[InternalRow] = {
// NOTE: We have to pass in Avro Schema used to read from Delta Log file
since we invoke combining API
// on the record from the Delta Log
- val combinedRecord = merge.combineAndGetUpdateValue(new
HoodieAvroIndexedRecord(curAvroRecord), newRecord, logFileReaderAvroSchema,
payloadProps)
- if (combinedRecord.isPresent) {
-
toScalaOption(combinedRecord.get.asInstanceOf[HoodieAvroIndexedRecord].toIndexedRecord)
- } else {
- Option.empty
+ newRecord.getRecordType match {
+ case HoodieRecordType.SPARK =>
+ // Get ordering value in curAvroRecord
+ var curRecord = new HoodieSparkRecord(curRow,
baseFileReaderSchema.structTypeSchema)
+ val orderField =
payloadProps.getProperty(HoodiePayloadProps.PAYLOAD_ORDERING_FIELD_PROP_KEY)
Review Comment:
move to sparkMerger
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileReader.java:
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.io.storage;
+
+import org.apache.hudi.HoodieInternalRowUtils;
+import org.apache.hudi.commmon.model.HoodieSparkRecord;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.util.ClosableIterator;
+import org.apache.hudi.common.util.MappingIterator;
+
+import org.apache.avro.Schema;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.types.StructType;
+
+import java.io.IOException;
+
+import static org.apache.hudi.TypeUtils.unsafeCast;
+
+public interface HoodieSparkFileReader extends HoodieFileReader<InternalRow> {
+
+ Logger LOG = LogManager.getLogger(HoodieSparkFileReader.class);
Review Comment:
will fix
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileWriter.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.io.storage;
+
+import org.apache.avro.Schema;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.spark.sql.catalyst.CatalystTypeConverters;
+import org.apache.spark.sql.catalyst.InternalRow;
+
+import java.io.IOException;
+import java.util.Properties;
+
+public interface HoodieSparkFileWriter extends HoodieFileWriter {
+ boolean canWrite();
+
+ void close() throws IOException;
+
+ void writeRowWithMetadata(HoodieKey recordKey, InternalRow row) throws
IOException;
+
+ void writeRow(String recordKey, InternalRow row) throws IOException;
+
+ @Override
+ default void write(String recordKey, HoodieRecord record, Schema schema,
Properties props) throws IOException {
+ writeRow(recordKey, (InternalRow) record.getData());
+ }
+
+ @Override
+ default void writeWithMetadata(HoodieKey key, HoodieRecord record, Schema
schema, Properties props) throws IOException {
+ writeRowWithMetadata(key, (InternalRow) record.getData());
+ }
+
+ default InternalRow prepRecordWithMetadata(HoodieKey key, InternalRow row,
String instantTime, Integer partitionId, long recordIndex, String fileName) {
+ String seqId = HoodieRecord.generateSequenceId(instantTime, partitionId,
recordIndex);
+
row.update(HoodieRecord.HoodieMetadataField.COMMIT_TIME_METADATA_FIELD.ordinal(),
CatalystTypeConverters.convertToCatalyst(instantTime));
+
row.update(HoodieRecord.HoodieMetadataField.COMMIT_SEQNO_METADATA_FIELD.ordinal(),
CatalystTypeConverters.convertToCatalyst(seqId));
+
row.update(HoodieRecord.HoodieMetadataField.RECORD_KEY_METADATA_FIELD.ordinal(),
CatalystTypeConverters.convertToCatalyst(key.getRecordKey()));
+
row.update(HoodieRecord.HoodieMetadataField.PARTITION_PATH_METADATA_FIELD.ordinal(),
CatalystTypeConverters.convertToCatalyst(key.getPartitionPath()));
+
row.update(HoodieRecord.HoodieMetadataField.FILENAME_METADATA_FIELD.ordinal(),
CatalystTypeConverters.convertToCatalyst(fileName));
Review Comment:
will fix
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/commmon/model/HoodieSparkRecord.java:
##########
@@ -0,0 +1,270 @@
+/*
+ * 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.commmon.model;
+
+import org.apache.hudi.HoodieInternalRowUtils;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieOperation;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.keygen.BaseKeyGenerator;
+import org.apache.hudi.keygen.SparkKeyGeneratorInterface;
+import org.apache.hudi.keygen.factory.HoodieSparkKeyGeneratorFactory;
+import org.apache.hudi.util.HoodieSparkRecordUtils;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.spark.sql.catalyst.CatalystTypeConverters;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import scala.Tuple2;
+
+import static
org.apache.hudi.common.table.HoodieTableConfig.POPULATE_META_FIELDS;
+import static org.apache.spark.sql.types.DataTypes.BooleanType;
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/**
+ * Spark Engine-specific Implementations of `HoodieRecord`.
+ */
+public class HoodieSparkRecord extends HoodieRecord<InternalRow> {
+
+ // IndexedRecord hold its schema, InternalRow should also hold its schema
+ private final StructType structType;
+
+ public HoodieSparkRecord(InternalRow data, StructType schema) {
+ super(null, data);
+ this.structType = schema;
+ }
+
+ public HoodieSparkRecord(InternalRow data, StructType schema, Comparable
orderingVal) {
+ super(null, data, orderingVal);
+ this.structType = schema;
+ }
+
+ public HoodieSparkRecord(HoodieKey key, InternalRow data, StructType schema)
{
+ super(key, data);
+ this.structType = schema;
+ }
+
+ public HoodieSparkRecord(HoodieKey key, InternalRow data, StructType schema,
Comparable orderingVal) {
+ super(key, data, orderingVal);
+ this.structType = schema;
+ }
+
+ public HoodieSparkRecord(HoodieKey key, InternalRow data, StructType schema,
HoodieOperation operation, Comparable orderingVal) {
+ super(key, data, operation, orderingVal);
+ this.structType = schema;
+ }
+
+ public HoodieSparkRecord(HoodieSparkRecord record) {
+ super(record);
+ this.structType = record.structType;
+ }
+
+ @Override
+ public HoodieRecord<InternalRow> newInstance() {
+ return new HoodieSparkRecord(this);
+ }
+
+ @Override
+ public HoodieRecord<InternalRow> newInstance(HoodieKey key, HoodieOperation
op) {
+ return new HoodieSparkRecord(key, data, structType, op,
getOrderingValue());
+ }
+
+ @Override
+ public HoodieRecord<InternalRow> newInstance(HoodieKey key) {
+ return new HoodieSparkRecord(key, data, structType, getOrderingValue());
+ }
+
+ @Override
+ public void deflate() {
+ }
+
+ @Override
+ public String getRecordKey(Option<BaseKeyGenerator> keyGeneratorOpt) {
+ if (key != null) {
+ return getRecordKey();
+ }
+ return keyGeneratorOpt.isPresent() ? ((SparkKeyGeneratorInterface)
keyGeneratorOpt.get()).getRecordKey(data, structType) :
data.getString(HoodieMetadataField.RECORD_KEY_METADATA_FIELD.ordinal());
+ }
+
+ @Override
+ public String getRecordKey(String keyFieldName) {
+ if (key != null) {
+ return getRecordKey();
+ }
+ Tuple2<StructField, Object> tuple2 =
HoodieInternalRowUtils.getCachedSchemaPosMap(structType).get(keyFieldName).get();
+ DataType dataType = tuple2._1.dataType();
+ int pos = (Integer) tuple2._2;
+ return data.get(pos, dataType).toString();
+ }
+
+ @Override
+ public HoodieRecordType getRecordType() {
+ return HoodieRecordType.SPARK;
+ }
+
+ @Override
+ public Object getRecordColumnValues(String[] columns, Schema schema, boolean
consistentLogicalTimestampEnabled) {
+ return HoodieSparkRecordUtils.getRecordColumnValues(this, columns,
structType, consistentLogicalTimestampEnabled);
+ }
+
+ @Override
+ public HoodieRecord mergeWith(Schema schema, HoodieRecord other, Schema
otherSchema, Schema writerSchema) throws IOException {
+ StructType otherStructType =
HoodieInternalRowUtils.getCachedSchema(otherSchema);
+ StructType writerStructType =
HoodieInternalRowUtils.getCachedSchema(writerSchema);
+ InternalRow mergeRow = HoodieInternalRowUtils.stitchRecords(data,
structType, (InternalRow) other.getData(), otherStructType, writerStructType);
+ return new HoodieSparkRecord(getKey(), mergeRow, writerStructType,
getOperation());
+ }
+
+ @Override
+ public HoodieRecord rewriteRecord(Schema recordSchema, Schema targetSchema,
TypedProperties props) throws IOException {
+ StructType targetStructType =
HoodieInternalRowUtils.getCachedSchema(targetSchema);
+ InternalRow rewriteRow = HoodieInternalRowUtils.rewriteRecord(data,
structType, targetStructType);
+ return new HoodieSparkRecord(getKey(), rewriteRow, targetStructType,
getOperation());
+ }
+
+ @Override
+ public HoodieRecord rewriteRecord(Schema recordSchema, Properties prop,
boolean schemaOnReadEnabled, Schema writeSchemaWithMetaFields) throws
IOException {
+ StructType writeSchemaWithMetaFieldsStructType =
HoodieInternalRowUtils.getCachedSchema(writeSchemaWithMetaFields);
+ InternalRow rewriteRow = schemaOnReadEnabled ?
HoodieInternalRowUtils.rewriteRecordWithNewSchema(data, structType,
writeSchemaWithMetaFieldsStructType, new HashMap<>())
+ : HoodieInternalRowUtils.rewriteRecord(data, structType,
writeSchemaWithMetaFieldsStructType);
+ return new HoodieSparkRecord(getKey(), rewriteRow,
writeSchemaWithMetaFieldsStructType, getOperation());
+ }
+
+ @Override
+ public HoodieRecord rewriteRecordWithMetadata(Schema recordSchema,
Properties prop, boolean schemaOnReadEnabled, Schema writeSchemaWithMetaFields,
String fileName) throws IOException {
+ StructType writeSchemaWithMetaFieldsStructType =
HoodieInternalRowUtils.getCachedSchema(writeSchemaWithMetaFields);
+ InternalRow rewriteRow = schemaOnReadEnabled ?
HoodieInternalRowUtils.rewriteEvolutionRecordWithMetadata(data, structType,
writeSchemaWithMetaFieldsStructType, fileName)
+ : HoodieInternalRowUtils.rewriteRecordWithMetadata(data, structType,
writeSchemaWithMetaFieldsStructType, fileName);
+ return new HoodieSparkRecord(getKey(), rewriteRow,
writeSchemaWithMetaFieldsStructType, getOperation());
+ }
+
+ @Override
+ public HoodieRecord rewriteRecordWithNewSchema(Schema recordSchema,
Properties prop, Schema newSchema, Map<String, String> renameCols) throws
IOException {
+ StructType newStructType =
HoodieInternalRowUtils.getCachedSchema(newSchema);
+ InternalRow rewriteRow =
HoodieInternalRowUtils.rewriteRecordWithNewSchema(data, structType,
newStructType, renameCols);
+ return new HoodieSparkRecord(getKey(), rewriteRow, newStructType,
getOperation());
+ }
+
+ @Override
+ public HoodieRecord rewriteRecordWithNewSchema(Schema recordSchema,
Properties prop, Schema newSchema) throws IOException {
+ StructType newStructType =
HoodieInternalRowUtils.getCachedSchema(newSchema);
+ InternalRow rewriteRow = HoodieInternalRowUtils.rewriteRecord(data,
structType, newStructType);
+ return new HoodieSparkRecord(getKey(), rewriteRow, structType,
getOperation());
+ }
+
+ @Override
+ public HoodieRecord overrideMetadataFieldValue(Schema recordSchema,
Properties prop, int pos, String newValue) throws IOException {
+ data.update(pos, CatalystTypeConverters.convertToCatalyst(newValue));
+ return this;
+ }
+
+ @Override
+ public HoodieRecord addMetadataValues(Schema recordSchema, Properties prop,
Map<HoodieMetadataField, String> metadataValues) throws IOException {
+ Arrays.stream(HoodieMetadataField.values()).forEach(metadataField -> {
+ String value = metadataValues.get(metadataField);
+ if (value != null) {
+ data.update(recordSchema.getField(metadataField.getFieldName()).pos(),
CatalystTypeConverters.convertToCatalyst(value));
+ }
+ });
+ return this;
+ }
+
+ @Override
+ public HoodieRecord expansion(Schema schema, Properties prop, String
payloadClass,
+ String preCombineField,
+ Option<Pair<String, String>> simpleKeyGenFieldsOpt,
+ Boolean withOperation,
+ Option<String> partitionNameOp,
+ Option<Boolean> populateMetaFieldsOp) {
+ boolean populateMetaFields = populateMetaFieldsOp.orElse(false);
+ if (populateMetaFields) {
+ return HoodieSparkRecordUtils.convertToHoodieSparkRecord(structType,
data, preCombineField, withOperation);
+ } else if (simpleKeyGenFieldsOpt.isPresent()) {
+ return HoodieSparkRecordUtils.convertToHoodieSparkRecord(structType,
data, preCombineField, simpleKeyGenFieldsOpt.get(), withOperation,
Option.empty());
+ } else {
+ return HoodieSparkRecordUtils.convertToHoodieSparkRecord(structType,
data, preCombineField, withOperation, partitionNameOp);
+ }
+ }
+
+ @Override
+ public HoodieRecord transform(Schema schema, Properties prop, boolean
useKeygen) {
+ StructType structType = HoodieInternalRowUtils.getCachedSchema(schema);
+ Option<SparkKeyGeneratorInterface> keyGeneratorOpt = Option.empty();
Review Comment:
will fix
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
