You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by yi...@apache.org on 2025/11/25 21:38:19 UTC
(hudi) branch master updated: feat(schema): Internal Schema System Integration with HoodieSchema (#14314)
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new a06d409242dd feat(schema): Internal Schema System Integration with HoodieSchema (#14314)
a06d409242dd is described below
commit a06d409242dd2333c7d8513100278e317842a788
Author: Tim Brown <ti...@onehouse.ai>
AuthorDate: Tue Nov 25 16:38:12 2025 -0500
feat(schema): Internal Schema System Integration with HoodieSchema (#14314)
---
.../apache/hudi/client/BaseHoodieWriteClient.java | 13 +-
.../table/action/commit/HoodieMergeHelper.java | 8 +-
.../storage/row/HoodieRowParquetWriteSupport.java | 12 +-
.../java/org/apache/hudi/avro/AvroSchemaUtils.java | 7 +-
.../table/log/AbstractHoodieLogRecordScanner.java | 4 +-
.../table/read/FileGroupReaderSchemaHandler.java | 7 +-
.../table/read/buffer/FileGroupRecordBuffer.java | 4 +-
.../hudi/common/util/InternalSchemaCache.java | 7 +-
.../convert/AvroInternalSchemaConverter.java | 613 ---------------------
.../schema/convert/InternalSchemaConverter.java | 568 +++++++++++++++++++
.../schema/utils/AvroSchemaEvolutionUtils.java | 15 +-
.../org/apache/hudi/avro/TestHoodieAvroUtils.java | 2 +-
.../hudi/common/schema/HoodieSchemaTestUtils.java | 70 +++
.../read/TestFileGroupReaderSchemaHandler.java | 5 +-
...erter.java => TestInternalSchemaConverter.java} | 68 +--
.../schema/utils/TestAvroSchemaEvolutionUtils.java | 242 ++++----
.../hudi/table/catalog/HoodieCatalogUtil.java | 5 +-
.../hudi/table/format/InternalSchemaManager.java | 4 +-
.../apache/hudi/hadoop/SchemaEvolutionContext.java | 6 +-
.../hudi/hadoop/avro/HoodieAvroParquetReader.java | 12 +-
.../hive/TestHoodieCombineHiveInputFormat.java | 5 +-
.../scala/org/apache/hudi/HoodieBaseRelation.scala | 8 +-
.../hudi/HoodieHadoopFsRelationFactory.scala | 6 +-
.../scala/org/apache/hudi/HoodieSchemaUtils.scala | 12 +-
.../org/apache/hudi/HoodieSparkSqlWriter.scala | 7 +-
.../spark/sql/hudi/command/AlterTableCommand.scala | 7 +-
.../util/TestSparkInternalSchemaConverter.scala | 8 +-
.../hudi/common/TestHoodieInternalRowUtils.scala | 25 +-
28 files changed, 897 insertions(+), 853 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
index cc19970fb60c..52ae2eca5d51 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
@@ -44,6 +44,7 @@ import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.model.TableServiceType;
import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.HoodieTableVersion;
@@ -76,7 +77,7 @@ import org.apache.hudi.internal.schema.InternalSchema;
import org.apache.hudi.internal.schema.Type;
import org.apache.hudi.internal.schema.action.InternalSchemaChangeApplier;
import org.apache.hudi.internal.schema.action.TableChange;
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter;
import org.apache.hudi.internal.schema.io.FileBasedInternalSchemaStorageManager;
import org.apache.hudi.internal.schema.utils.AvroSchemaEvolutionUtils;
import org.apache.hudi.internal.schema.utils.InternalSchemaUtils;
@@ -349,7 +350,7 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient
InternalSchema internalSchema;
Schema avroSchema = HoodieAvroUtils.createHoodieWriteSchema(config.getSchema(), config.allowOperationMetadataField());
if (historySchemaStr.isEmpty()) {
- internalSchema = SerDeHelper.fromJson(config.getInternalSchema()).orElseGet(() -> AvroInternalSchemaConverter.convert(avroSchema));
+ internalSchema = SerDeHelper.fromJson(config.getInternalSchema()).orElseGet(() -> InternalSchemaConverter.convert(HoodieSchema.fromAvroSchema(avroSchema)));
internalSchema.setSchemaId(Long.parseLong(instantTime));
} else {
internalSchema = InternalSchemaUtils.searchSchema(Long.parseLong(instantTime),
@@ -367,7 +368,7 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient
schemasManager.persistHistorySchemaStr(instantTime, SerDeHelper.inheritSchemas(evolvedSchema, historySchemaStr));
}
// update SCHEMA_KEY
- metadata.addMetadata(SCHEMA_KEY, AvroInternalSchemaConverter.convert(evolvedSchema, avroSchema.getFullName()).toString());
+ metadata.addMetadata(SCHEMA_KEY, InternalSchemaConverter.convert(evolvedSchema, avroSchema.getFullName()).toString());
}
}
@@ -1570,7 +1571,7 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient
public void addColumn(String colName, Schema schema, String doc, String position, TableChange.ColumnPositionChange.ColumnPositionType positionType) {
Pair<InternalSchema, HoodieTableMetaClient> pair = getInternalSchemaAndMetaClient();
InternalSchema newSchema = new InternalSchemaChangeApplier(pair.getLeft())
- .applyAddChange(colName, AvroInternalSchemaConverter.convertToField(schema), doc, position, positionType);
+ .applyAddChange(colName, InternalSchemaConverter.convertToField(HoodieSchema.fromAvroSchema(schema)), doc, position, positionType);
commitTableChange(newSchema, pair.getRight());
}
@@ -1667,7 +1668,7 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient
TableSchemaResolver schemaUtil = new TableSchemaResolver(metaClient);
String historySchemaStr = schemaUtil.getTableHistorySchemaStrFromCommitMetadata().orElseGet(
() -> SerDeHelper.inheritSchemas(getInternalSchema(schemaUtil), ""));
- Schema schema = AvroInternalSchemaConverter.convert(newSchema, getAvroRecordQualifiedName(config.getTableName()));
+ Schema schema = InternalSchemaConverter.convert(newSchema, getAvroRecordQualifiedName(config.getTableName())).toAvroSchema();
String commitActionType = CommitUtils.getCommitActionType(WriteOperationType.ALTER_SCHEMA, metaClient.getTableType());
String instantTime = startCommit(commitActionType, metaClient);
config.setSchema(schema.toString());
@@ -1691,7 +1692,7 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient
private InternalSchema getInternalSchema(TableSchemaResolver schemaUtil) {
return schemaUtil.getTableInternalSchemaFromCommitMetadata().orElseGet(() -> {
try {
- return AvroInternalSchemaConverter.convert(schemaUtil.getTableAvroSchema());
+ return InternalSchemaConverter.convert(HoodieSchema.fromAvroSchema(schemaUtil.getTableAvroSchema()));
} catch (Exception e) {
throw new HoodieException(String.format("cannot find schema for current table: %s", config.getBasePath()));
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java
index 4a8695b97cd9..5cad7ad8e334 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java
@@ -32,7 +32,7 @@ import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.internal.schema.InternalSchema;
import org.apache.hudi.internal.schema.action.InternalSchemaMerger;
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter;
import org.apache.hudi.internal.schema.utils.AvroSchemaEvolutionUtils;
import org.apache.hudi.internal.schema.utils.InternalSchemaUtils;
import org.apache.hudi.internal.schema.utils.SerDeHelper;
@@ -182,7 +182,7 @@ public class HoodieMergeHelper<T> extends BaseMergeHelper {
if (fileSchema.isEmptySchema() && writeConfig.getBoolean(HoodieCommonConfig.RECONCILE_SCHEMA)) {
TableSchemaResolver tableSchemaResolver = new TableSchemaResolver(metaClient);
try {
- fileSchema = AvroInternalSchemaConverter.convert(tableSchemaResolver.getTableAvroSchema(true));
+ fileSchema = InternalSchemaConverter.convert(HoodieSchema.fromAvroSchema(tableSchemaResolver.getTableAvroSchema(true)));
} catch (Exception e) {
throw new HoodieException(String.format("Failed to get InternalSchema for given versionId: %s", commitInstantTime), e);
}
@@ -203,8 +203,8 @@ public class HoodieMergeHelper<T> extends BaseMergeHelper {
.collect(Collectors.toList());
InternalSchema mergedSchema = new InternalSchemaMerger(writeInternalSchema, querySchema,
true, false, false).mergeSchema();
- Schema newWriterSchema = AvroInternalSchemaConverter.convert(mergedSchema, writerSchema.getFullName());
- Schema writeSchemaFromFile = AvroInternalSchemaConverter.convert(writeInternalSchema, newWriterSchema.getFullName());
+ Schema newWriterSchema = InternalSchemaConverter.convert(mergedSchema, writerSchema.getFullName()).getAvroSchema();
+ Schema writeSchemaFromFile = InternalSchemaConverter.convert(writeInternalSchema, newWriterSchema.getFullName()).getAvroSchema();
boolean needToReWriteRecord = sameCols.size() != colNamesFromWriteSchema.size()
|| SchemaCompatibility.checkReaderWriterCompatibility(newWriterSchema, writeSchemaFromFile).getType() == org.apache.avro.SchemaCompatibility.SchemaCompatibilityType.COMPATIBLE;
if (needToReWriteRecord) {
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowParquetWriteSupport.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowParquetWriteSupport.java
index edbdab45243d..427cec23fbc6 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowParquetWriteSupport.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowParquetWriteSupport.java
@@ -20,15 +20,15 @@ package org.apache.hudi.io.storage.row;
import org.apache.hudi.HoodieSparkUtils;
import org.apache.hudi.SparkAdapterSupport$;
-import org.apache.hudi.avro.HoodieAvroUtils;
import org.apache.hudi.avro.HoodieBloomFilterWriteSupport;
import org.apache.hudi.common.bloom.BloomFilter;
import org.apache.hudi.common.config.HoodieConfig;
import org.apache.hudi.common.config.HoodieStorageConfig;
+import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hudi.common.util.ValidationUtils;
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter;
import org.apache.hudi.internal.schema.utils.SerDeHelper;
import org.apache.avro.LogicalType;
@@ -132,12 +132,12 @@ public class HoodieRowParquetWriteSupport extends WriteSupport<InternalRow> {
|| Boolean.parseBoolean(config.getStringOrDefault(AvroWriteSupport.WRITE_OLD_LIST_STRUCTURE, "false"));
this.structType = structType;
// The avro schema is used to determine the precision for timestamps
- this.avroSchema = SerDeHelper.fromJson(config.getString(INTERNAL_SCHEMA_STRING)).map(internalSchema -> AvroInternalSchemaConverter.convert(internalSchema, "spark_schema"))
+ this.avroSchema = SerDeHelper.fromJson(config.getString(INTERNAL_SCHEMA_STRING)).map(internalSchema -> InternalSchemaConverter.convert(internalSchema, "spark_schema"))
.orElseGet(() -> {
String schemaString = Option.ofNullable(config.getString(WRITE_SCHEMA_OVERRIDE)).orElseGet(() -> config.getString(AVRO_SCHEMA_STRING));
- Schema parsedSchema = new Schema.Parser().parse(schemaString);
- return HoodieAvroUtils.addMetadataFields(parsedSchema, config.getBooleanOrDefault(ALLOW_OPERATION_METADATA_FIELD));
- });
+ HoodieSchema parsedSchema = HoodieSchema.parse(schemaString);
+ return HoodieSchema.addMetadataFields(parsedSchema, config.getBooleanOrDefault(ALLOW_OPERATION_METADATA_FIELD));
+ }).toAvroSchema();
ParquetWriteSupport.setSchema(structType, hadoopConf);
this.rootFieldWriters = getFieldWriters(structType, avroSchema);
this.hadoopConf = hadoopConf;
diff --git a/hudi-common/src/main/java/org/apache/hudi/avro/AvroSchemaUtils.java b/hudi-common/src/main/java/org/apache/hudi/avro/AvroSchemaUtils.java
index a17b0472e7c0..0eae120a5cfd 100644
--- a/hudi-common/src/main/java/org/apache/hudi/avro/AvroSchemaUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/avro/AvroSchemaUtils.java
@@ -18,6 +18,7 @@
package org.apache.hudi.avro;
+import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.exception.HoodieAvroSchemaException;
@@ -48,7 +49,7 @@ import java.util.stream.Collectors;
import static org.apache.hudi.avro.HoodieAvroUtils.createNewSchemaField;
import static org.apache.hudi.common.util.CollectionUtils.reduce;
import static org.apache.hudi.common.util.ValidationUtils.checkState;
-import static org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter.convert;
+import static org.apache.hudi.internal.schema.convert.InternalSchemaConverter.convert;
/**
* Utils for Avro Schema.
@@ -722,10 +723,10 @@ public class AvroSchemaUtils {
if (filterCols.isEmpty()) {
return schema;
}
- InternalSchema internalSchema = convert(schema);
+ InternalSchema internalSchema = convert(HoodieSchema.fromAvroSchema(schema));
TableChanges.ColumnUpdateChange schemaChange = TableChanges.ColumnUpdateChange.get(internalSchema);
schemaChange = reduce(filterCols, schemaChange,
(change, field) -> change.updateColumnNullability(field, true));
- return convert(SchemaChangeUtils.applyTableChanges2Schema(internalSchema, schemaChange), schema.getFullName());
+ return convert(SchemaChangeUtils.applyTableChanges2Schema(internalSchema, schemaChange), schema.getFullName()).toAvroSchema();
}
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordScanner.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordScanner.java
index 97d0f41cca60..e5c6e17bf310 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordScanner.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordScanner.java
@@ -43,7 +43,7 @@ import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.internal.schema.InternalSchema;
import org.apache.hudi.internal.schema.action.InternalSchemaMerger;
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StoragePath;
@@ -843,7 +843,7 @@ public abstract class AbstractHoodieLogRecordScanner {
InternalSchema fileSchema = InternalSchemaCache.searchSchemaAndCache(currentInstantTime, hoodieTableMetaClient);
InternalSchema mergedInternalSchema = new InternalSchemaMerger(fileSchema, internalSchema,
true, false).mergeSchema();
- Schema mergedAvroSchema = AvroInternalSchemaConverter.convert(mergedInternalSchema, readerSchema.getFullName());
+ Schema mergedAvroSchema = InternalSchemaConverter.convert(mergedInternalSchema, readerSchema.getFullName()).toAvroSchema();
return Option.of(Pair.of((record) -> {
return record.rewriteRecordWithNewSchema(
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/FileGroupReaderSchemaHandler.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/FileGroupReaderSchemaHandler.java
index 86be346c8b76..c335dd2d576c 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/FileGroupReaderSchemaHandler.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/FileGroupReaderSchemaHandler.java
@@ -27,6 +27,7 @@ import org.apache.hudi.common.engine.HoodieReaderContext;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordMerger;
+import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.HoodieTableMetaClient;
@@ -38,7 +39,7 @@ import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.common.util.collection.Triple;
import org.apache.hudi.internal.schema.InternalSchema;
import org.apache.hudi.internal.schema.action.InternalSchemaMerger;
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter;
import org.apache.hudi.storage.StoragePath;
import org.apache.avro.Schema;
@@ -159,7 +160,7 @@ public class FileGroupReaderSchemaHandler<T> {
InternalSchema fileSchema = InternalSchemaCache.searchSchemaAndCache(commitInstantTime, metaClient);
Pair<InternalSchema, Map<String, String>> mergedInternalSchema = new InternalSchemaMerger(fileSchema, internalSchema,
true, false, false).mergeSchemaGetRenamed();
- Schema mergedAvroSchema = AvroSchemaCache.intern(AvroInternalSchemaConverter.convert(mergedInternalSchema.getLeft(), requiredSchema.getFullName()));
+ Schema mergedAvroSchema = AvroSchemaCache.intern(InternalSchemaConverter.convert(mergedInternalSchema.getLeft(), requiredSchema.getFullName()).toAvroSchema());
return Pair.of(mergedAvroSchema, mergedInternalSchema.getRight());
}
@@ -180,7 +181,7 @@ public class FileGroupReaderSchemaHandler<T> {
}
protected InternalSchema doPruneInternalSchema(Schema requiredSchema, InternalSchema internalSchema) {
- return AvroInternalSchemaConverter.pruneAvroSchemaToInternalSchema(requiredSchema, internalSchema);
+ return InternalSchemaConverter.pruneHoodieSchemaToInternalSchema(HoodieSchema.fromAvroSchema(requiredSchema), internalSchema);
}
@VisibleForTesting
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/FileGroupRecordBuffer.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/FileGroupRecordBuffer.java
index 5b3ff3bcae0d..27c7644ee859 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/FileGroupRecordBuffer.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/FileGroupRecordBuffer.java
@@ -47,7 +47,7 @@ import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.internal.schema.InternalSchema;
import org.apache.hudi.internal.schema.action.InternalSchemaMerger;
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter;
import org.apache.avro.Schema;
@@ -227,7 +227,7 @@ abstract class FileGroupRecordBuffer<T> implements HoodieFileGroupRecordBuffer<T
InternalSchema fileSchema = InternalSchemaCache.searchSchemaAndCache(currentInstantTime, hoodieTableMetaClient);
Pair<InternalSchema, Map<String, String>> mergedInternalSchema = new InternalSchemaMerger(fileSchema, internalSchema,
true, false, false).mergeSchemaGetRenamed();
- Schema mergedAvroSchema = AvroInternalSchemaConverter.convert(mergedInternalSchema.getLeft(), readerSchema.getFullName());
+ Schema mergedAvroSchema = InternalSchemaConverter.convert(mergedInternalSchema.getLeft(), readerSchema.getFullName()).getAvroSchema();
// `mergedAvroSchema` maybe not equal with `readerSchema`, case: drop a column `f_x`, and then add a new column with same name `f_x`,
// then the new added column in `mergedAvroSchema` will have a suffix: `f_xsuffix`, distinguished from the original column `f_x`, see
// InternalSchemaMerger#buildRecordType() for details.
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/InternalSchemaCache.java b/hudi-common/src/main/java/org/apache/hudi/common/util/InternalSchemaCache.java
index 907d89eedd0c..6348a10f899b 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/InternalSchemaCache.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/InternalSchemaCache.java
@@ -18,8 +18,8 @@
package org.apache.hudi.common.util;
-import org.apache.hudi.avro.HoodieAvroUtils;
import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.CommitMetadataSerDe;
@@ -33,7 +33,7 @@ import org.apache.hudi.common.table.timeline.TimelinePathProvider;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.internal.schema.InternalSchema;
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter;
import org.apache.hudi.internal.schema.io.FileBasedInternalSchemaStorageManager;
import org.apache.hudi.internal.schema.utils.InternalSchemaUtils;
import org.apache.hudi.internal.schema.utils.SerDeHelper;
@@ -43,7 +43,6 @@ import org.apache.hudi.storage.StoragePathInfo;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
-import org.apache.avro.Schema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -215,7 +214,7 @@ public class InternalSchemaCache {
return fileSchema.isEmptySchema()
? StringUtils.isNullOrEmpty(avroSchema)
? InternalSchema.getEmptyInternalSchema()
- : AvroInternalSchemaConverter.convert(HoodieAvroUtils.addMetadataFields(new Schema.Parser().parse(avroSchema)))
+ : InternalSchemaConverter.convert(HoodieSchema.createHoodieWriteSchema(avroSchema, false))
: fileSchema;
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/internal/schema/convert/AvroInternalSchemaConverter.java b/hudi-common/src/main/java/org/apache/hudi/internal/schema/convert/AvroInternalSchemaConverter.java
deleted file mode 100644
index f4dd2d8af167..000000000000
--- a/hudi-common/src/main/java/org/apache/hudi/internal/schema/convert/AvroInternalSchemaConverter.java
+++ /dev/null
@@ -1,613 +0,0 @@
-/*
- * 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.internal.schema.convert;
-
-import org.apache.hudi.avro.AvroSchemaUtils;
-import org.apache.hudi.common.util.Option;
-import org.apache.hudi.common.util.VisibleForTesting;
-import org.apache.hudi.exception.HoodieNullSchemaTypeException;
-import org.apache.hudi.internal.schema.HoodieSchemaException;
-import org.apache.hudi.internal.schema.InternalSchema;
-import org.apache.hudi.internal.schema.Type;
-import org.apache.hudi.internal.schema.Types;
-import org.apache.hudi.internal.schema.utils.InternalSchemaUtils;
-
-import org.apache.avro.JsonProperties;
-import org.apache.avro.LogicalType;
-import org.apache.avro.LogicalTypes;
-import org.apache.avro.Schema;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.Deque;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.stream.Collectors;
-
-import static org.apache.avro.Schema.Type.STRING;
-import static org.apache.avro.Schema.Type.UNION;
-
-/**
- * Auxiliary class.
- * Converts an avro schema into InternalSchema, or convert InternalSchema to an avro schema
- */
-public class AvroInternalSchemaConverter {
-
- // NOTE: We're using dot as field's name delimiter for nested fields
- // so that Avro is able to interpret qualified name as rather
- // the combination of the Avro's namespace and actual record's name.
- // For example qualified nested field's name "trip.fare.amount",
- // Avro will produce a record with
- // - Namespace: "trip.fare"
- // - Name: "amount"
- //
- // This is crucial aspect of maintaining compatibility b/w schemas, after
- // converting Avro [[Schema]]s to [[InternalSchema]]s and back
- private static final String AVRO_NAME_DELIMITER = ".";
-
- /**
- * Convert internalSchema to avro Schema.
- *
- * @param internalSchema internal schema.
- * @param name the record name.
- * @return an avro Schema.
- */
- public static Schema convert(InternalSchema internalSchema, String name) {
- return buildAvroSchemaFromInternalSchema(internalSchema, name);
- }
-
- public static InternalSchema pruneAvroSchemaToInternalSchema(Schema schema, InternalSchema originSchema) {
- List<String> pruneNames = collectColNamesFromSchema(schema);
- return InternalSchemaUtils.pruneInternalSchema(originSchema, pruneNames);
- }
-
- /**
- * Collect all the leaf nodes names.
- *
- * @param schema a avro schema.
- * @return leaf nodes full names.
- */
- @VisibleForTesting
- static List<String> collectColNamesFromSchema(Schema schema) {
- List<String> result = new ArrayList<>();
- Deque<String> visited = new LinkedList<>();
- collectColNamesFromAvroSchema(schema, visited, result);
- return result;
- }
-
- private static void collectColNamesFromAvroSchema(Schema schema, Deque<String> visited, List<String> resultSet) {
- switch (schema.getType()) {
- case RECORD:
- List<Schema.Field> fields = schema.getFields();
- for (Schema.Field f : fields) {
- visited.push(f.name());
- collectColNamesFromAvroSchema(f.schema(), visited, resultSet);
- visited.pop();
- addFullNameIfLeafNode(f.schema(), f.name(), visited, resultSet);
- }
- return;
-
- case UNION:
- collectColNamesFromAvroSchema(AvroSchemaUtils.getNonNullTypeFromUnion(schema), visited, resultSet);
- return;
-
- case ARRAY:
- visited.push("element");
- collectColNamesFromAvroSchema(schema.getElementType(), visited, resultSet);
- visited.pop();
- addFullNameIfLeafNode(schema.getElementType(), "element", visited, resultSet);
- return;
-
- case MAP:
- addFullNameIfLeafNode(STRING, "key", visited, resultSet);
- visited.push("value");
- collectColNamesFromAvroSchema(schema.getValueType(), visited, resultSet);
- visited.pop();
- addFullNameIfLeafNode(schema.getValueType(), "value", visited, resultSet);
- return;
-
- default:
- }
- }
-
- private static void addFullNameIfLeafNode(Schema schema, String name, Deque<String> visited, List<String> resultSet) {
- addFullNameIfLeafNode(AvroSchemaUtils.getNonNullTypeFromUnion(schema).getType(), name, visited, resultSet);
- }
-
- private static void addFullNameIfLeafNode(Schema.Type type, String name, Deque<String> visited, List<String> resultSet) {
- switch (type) {
- case RECORD:
- case ARRAY:
- case MAP:
- return;
- default:
- resultSet.add(InternalSchemaUtils.createFullName(name, visited));
- }
- }
-
- /**
- * Converting from avro -> internal schema -> avro
- * causes null to always be first in unions.
- * if we compare a schema that has not been converted to internal schema
- * at any stage, the difference in ordering can cause issues. To resolve this,
- * we order null to be first for any avro schema that enters into hudi.
- * AvroSchemaUtils.isProjectionOfInternal uses index based comparison for unions.
- * Spark and flink don't support complex unions so this would not be an issue
- * but for the metadata table HoodieMetadata.avsc uses a trick where we have a bunch of
- * different types wrapped in record for col stats.
- *
- * @param schema avro schema.
- * @return an avro Schema where null is the first.
- */
- public static Schema fixNullOrdering(Schema schema) {
- if (schema == null) {
- return Schema.create(Schema.Type.NULL);
- } else if (schema.getType() == Schema.Type.NULL) {
- return schema;
- }
- return convert(convert(schema), schema.getFullName());
- }
-
- /**
- * Convert RecordType to avro Schema.
- *
- * @param type internal schema.
- * @param name the record name.
- * @return an avro Schema.
- */
- public static Schema convert(Types.RecordType type, String name) {
- return buildAvroSchemaFromType(type, name);
- }
-
- /**
- * Convert internal type to avro Schema.
- *
- * @param type internal type.
- * @param name the record name.
- * @return an avro Schema.
- */
- public static Schema convert(Type type, String name) {
- return buildAvroSchemaFromType(type, name);
- }
-
- /** Convert an avro schema into internal type. */
- public static Type convertToField(Schema schema) {
- return buildTypeFromAvroSchema(schema, Collections.emptyMap());
- }
-
- private static Type convertToField(Schema schema, Map<String, Integer> existingFieldNameToPositionMapping) {
- return buildTypeFromAvroSchema(schema, existingFieldNameToPositionMapping);
- }
-
- /** Convert an avro schema into internalSchema. */
- public static InternalSchema convert(Schema schema, Map<String, Integer> existingFieldNameToPositionMapping) {
- return new InternalSchema((Types.RecordType) convertToField(schema, existingFieldNameToPositionMapping));
- }
-
- public static InternalSchema convert(Schema schema) {
- return new InternalSchema((Types.RecordType) convertToField(schema));
- }
-
- /** Check whether current avro schema is optional?. */
- public static boolean isOptional(Schema schema) {
- if (schema.getType() == UNION && schema.getTypes().size() == 2) {
- return schema.getTypes().get(0).getType() == Schema.Type.NULL || schema.getTypes().get(1).getType() == Schema.Type.NULL;
- }
- return false;
- }
-
- /** Returns schema with nullable true. */
- public static Schema nullableSchema(Schema schema) {
- if (schema.getType() == UNION) {
- if (!isOptional(schema)) {
- throw new HoodieSchemaException(String.format("Union schemas are not supported: %s", schema));
- }
- return schema;
- } else {
- return Schema.createUnion(Schema.create(Schema.Type.NULL), schema);
- }
- }
-
- /**
- * Build hudi type from avro schema.
- *
- * @param schema a avro schema.
- * @return a hudi type.
- */
- public static Type buildTypeFromAvroSchema(Schema schema, Map<String, Integer> existingNameToPositions) {
- // set flag to check this has not been visited.
- Deque<String> visited = new LinkedList<>();
- AtomicInteger nextId = new AtomicInteger(0);
- return visitAvroSchemaToBuildType(schema, visited, "", nextId, existingNameToPositions);
- }
-
- private static void checkNullType(Type fieldType, String fieldName, Deque<String> visited) {
- if (fieldType == null) {
- StringBuilder sb = new StringBuilder();
- sb.append("Field '");
- Iterator<String> visitedIterator = visited.descendingIterator();
- while (visitedIterator.hasNext()) {
- sb.append(visitedIterator.next());
- sb.append(".");
- }
- sb.append(fieldName);
- sb.append("' has type null");
- throw new HoodieNullSchemaTypeException(sb.toString());
- } else if (fieldType.typeId() == Type.TypeID.ARRAY) {
- visited.push(fieldName);
- checkNullType(((Types.ArrayType) fieldType).elementType(), InternalSchema.ARRAY_ELEMENT, visited);
- visited.pop();
- } else if (fieldType.typeId() == Type.TypeID.MAP) {
- visited.push(fieldName);
- checkNullType(((Types.MapType) fieldType).valueType(), InternalSchema.MAP_VALUE, visited);
- visited.pop();
- }
- }
-
- /**
- * Converts an avro schema into hudi type.
- *
- * @param schema a avro schema.
- * @param visited track the visit node when do traversal for avro schema; used to check if the name of avro record schema is correct.
- * @param currentFieldPath the dot-separated path to the current field; empty at the root and always ends in a '.' otherwise for ease of concatenation.
- * @param nextId an initial id which used to create id for all fields.
- * @return a hudi type match avro schema.
- */
- private static Type visitAvroSchemaToBuildType(Schema schema, Deque<String> visited, String currentFieldPath, AtomicInteger nextId, Map<String, Integer> existingNameToPosition) {
- switch (schema.getType()) {
- case RECORD:
- String name = schema.getFullName();
- if (visited.contains(name)) {
- throw new HoodieSchemaException(String.format("cannot convert recursive avro record %s", name));
- }
- visited.push(name);
- List<Schema.Field> fields = existingNameToPosition.isEmpty() ? schema.getFields() :
- schema.getFields().stream()
- .sorted(Comparator.comparing(field -> existingNameToPosition.getOrDefault(currentFieldPath + field.name(), Integer.MAX_VALUE)))
- .collect(Collectors.toList());
- List<Type> fieldTypes = new ArrayList<>(fields.size());
- int nextAssignId = nextId.get();
- nextId.set(nextAssignId + fields.size());
- fields.forEach(field -> {
- Type fieldType = visitAvroSchemaToBuildType(field.schema(), visited, currentFieldPath + field.name() + ".", nextId, existingNameToPosition);
- checkNullType(fieldType, field.name(), visited);
- fieldTypes.add(fieldType);
- });
- visited.pop();
- List<Types.Field> internalFields = new ArrayList<>(fields.size());
-
- for (int i = 0; i < fields.size(); i++) {
- Schema.Field field = fields.get(i);
- Type fieldType = fieldTypes.get(i);
- internalFields.add(Types.Field.get(nextAssignId, AvroInternalSchemaConverter.isOptional(field.schema()), field.name(), fieldType, field.doc()));
- nextAssignId += 1;
- }
- // NOTE: We're keeping a tab of full-name here to make sure we stay
- // compatible across various Spark (>= 2.4) and Avro (>= 1.8.2) versions;
- // Avro will be properly handling fully-qualified names on its own (splitting
- // them up into namespace/struct-name pair)
- return Types.RecordType.get(internalFields, schema.getFullName());
- case UNION:
- List<Type> fTypes = new ArrayList<>(2);
- schema.getTypes().forEach(t -> {
- fTypes.add(visitAvroSchemaToBuildType(t, visited, currentFieldPath, nextId, existingNameToPosition));
- });
- return fTypes.get(0) == null ? fTypes.get(1) : fTypes.get(0);
- case ARRAY:
- String elementPath = currentFieldPath + InternalSchema.ARRAY_ELEMENT + ".";
- Schema elementSchema = schema.getElementType();
- int elementId = nextId.get();
- nextId.set(elementId + 1);
- Type elementType = visitAvroSchemaToBuildType(elementSchema, visited, elementPath, nextId, existingNameToPosition);
- return Types.ArrayType.get(elementId, AvroInternalSchemaConverter.isOptional(schema.getElementType()), elementType);
- case MAP:
- int keyId = nextId.get();
- int valueId = keyId + 1;
- nextId.set(valueId + 1);
- String valuePath = currentFieldPath + InternalSchema.MAP_VALUE + ".";
- Type valueType = visitAvroSchemaToBuildType(schema.getValueType(), visited, valuePath, nextId, existingNameToPosition);
- return Types.MapType.get(keyId, valueId, Types.StringType.get(), valueType, AvroInternalSchemaConverter.isOptional(schema.getValueType()));
- default:
- return visitAvroPrimitiveToBuildInternalType(schema);
- }
- }
-
- private static Type visitAvroPrimitiveToBuildInternalType(Schema primitive) {
- LogicalType logical = primitive.getLogicalType();
- if (logical != null) {
- String name = logical.getName();
- if (logical instanceof LogicalTypes.Decimal) {
- if (primitive.getType() == Schema.Type.FIXED) {
- return Types.DecimalTypeFixed.get(((LogicalTypes.Decimal) logical).getPrecision(),
- ((LogicalTypes.Decimal) logical).getScale(), primitive.getFixedSize());
- } else if (primitive.getType() == Schema.Type.BYTES) {
- return Types.DecimalTypeBytes.get(
- ((LogicalTypes.Decimal) logical).getPrecision(),
- ((LogicalTypes.Decimal) logical).getScale());
- } else {
- throw new IllegalArgumentException("Unsupported primitive type for Decimal: " + primitive.getType().getName());
- }
- } else if (logical instanceof LogicalTypes.Date) {
- return Types.DateType.get();
- } else if (logical instanceof LogicalTypes.TimeMillis) {
- return Types.TimeMillisType.get();
- } else if (logical instanceof LogicalTypes.TimeMicros) {
- return Types.TimeType.get();
- } else if (logical instanceof LogicalTypes.TimestampMillis) {
- return Types.TimestampMillisType.get();
- } else if (logical instanceof LogicalTypes.TimestampMicros) {
- return Types.TimestampType.get();
- } else if (logical instanceof LogicalTypes.LocalTimestampMillis) {
- return Types.LocalTimestampMillisType.get();
- } else if (logical instanceof LogicalTypes.LocalTimestampMicros) {
- return Types.LocalTimestampMicrosType.get();
- } else if (LogicalTypes.uuid().getName().equals(name)) {
- return Types.UUIDType.get();
- }
- }
-
- switch (primitive.getType()) {
- case BOOLEAN:
- return Types.BooleanType.get();
- case INT:
- return Types.IntType.get();
- case LONG:
- return Types.LongType.get();
- case FLOAT:
- return Types.FloatType.get();
- case DOUBLE:
- return Types.DoubleType.get();
- case STRING:
- case ENUM:
- return Types.StringType.get();
- case FIXED:
- return Types.FixedType.getFixed(primitive.getFixedSize());
- case BYTES:
- return Types.BinaryType.get();
- case NULL:
- return null;
- default:
- throw new UnsupportedOperationException("Unsupported primitive type: " + primitive);
- }
- }
-
- /**
- * Converts hudi type into an Avro Schema.
- *
- * @param type a hudi type.
- * @param recordName the record name
- * @return an Avro schema match this type
- */
- public static Schema buildAvroSchemaFromType(Type type, String recordName) {
- Map<Type, Schema> cache = new HashMap<>();
- return visitInternalSchemaToBuildAvroSchema(type, cache, recordName);
- }
-
- /**
- * Converts hudi internal Schema into an Avro Schema.
- *
- * @param schema a hudi internal Schema.
- * @param recordName the record name
- * @return a Avro schema match hudi internal schema.
- */
- public static Schema buildAvroSchemaFromInternalSchema(InternalSchema schema, String recordName) {
- Map<Type, Schema> cache = new HashMap<>();
- return visitInternalSchemaToBuildAvroSchema(schema.getRecord(), cache, recordName);
- }
-
- /**
- * Converts hudi type into an Avro Schema.
- *
- * @param type a hudi type.
- * @param cache use to cache intermediate convert result to save cost.
- * @param recordName auto-generated record name used as a fallback, in case
- * {@link org.apache.hudi.internal.schema.Types.RecordType} doesn't bear original record-name
- * @return an Avro schema match this type
- */
- private static Schema visitInternalSchemaToBuildAvroSchema(Type type, Map<Type, Schema> cache, String recordName) {
- switch (type.typeId()) {
- case RECORD:
- Types.RecordType record = (Types.RecordType) type;
- List<Schema> schemas = new ArrayList<>();
- record.fields().forEach(f -> {
- String nestedRecordName = recordName + AVRO_NAME_DELIMITER + f.name();
- Schema tempSchema = visitInternalSchemaToBuildAvroSchema(f.type(), cache, nestedRecordName);
- // convert tempSchema
- Schema result = f.isOptional() ? AvroInternalSchemaConverter.nullableSchema(tempSchema) : tempSchema;
- schemas.add(result);
- });
- // check visited
- Schema recordSchema;
- recordSchema = cache.get(record);
- if (recordSchema != null) {
- return recordSchema;
- }
- recordSchema = visitInternalRecordToBuildAvroRecord(record, schemas, recordName);
- cache.put(record, recordSchema);
- return recordSchema;
- case ARRAY:
- Types.ArrayType array = (Types.ArrayType) type;
- Schema elementSchema;
- elementSchema = visitInternalSchemaToBuildAvroSchema(array.elementType(), cache, recordName);
- Schema arraySchema;
- arraySchema = cache.get(array);
- if (arraySchema != null) {
- return arraySchema;
- }
- arraySchema = visitInternalArrayToBuildAvroArray(array, elementSchema);
- cache.put(array, arraySchema);
- return arraySchema;
- case MAP:
- Types.MapType map = (Types.MapType) type;
- Schema keySchema;
- Schema valueSchema;
- keySchema = visitInternalSchemaToBuildAvroSchema(map.keyType(), cache, recordName);
- valueSchema = visitInternalSchemaToBuildAvroSchema(map.valueType(), cache, recordName);
- Schema mapSchema;
- mapSchema = cache.get(map);
- if (mapSchema != null) {
- return mapSchema;
- }
- mapSchema = visitInternalMapToBuildAvroMap(map, keySchema, valueSchema);
- cache.put(map, mapSchema);
- return mapSchema;
- default:
- Schema primitiveSchema = visitInternalPrimitiveToBuildAvroPrimitiveType((Type.PrimitiveType) type, recordName);
- cache.put(type, primitiveSchema);
- return primitiveSchema;
- }
- }
-
- /**
- * Converts hudi RecordType to Avro RecordType.
- * this is auxiliary function used by visitInternalSchemaToBuildAvroSchema
- */
- private static Schema visitInternalRecordToBuildAvroRecord(Types.RecordType recordType, List<Schema> fieldSchemas, String recordNameFallback) {
- List<Types.Field> fields = recordType.fields();
- List<Schema.Field> avroFields = new ArrayList<>();
- for (int i = 0; i < fields.size(); i++) {
- Types.Field f = fields.get(i);
- Schema.Field field = new Schema.Field(f.name(), fieldSchemas.get(i), f.doc(), f.isOptional() ? JsonProperties.NULL_VALUE : null);
- avroFields.add(field);
- }
- String recordName = Option.ofNullable(recordType.name()).orElse(recordNameFallback);
- return Schema.createRecord(recordName, null, null, false, avroFields);
- }
-
- /**
- * Converts hudi ArrayType to Avro ArrayType.
- * this is auxiliary function used by visitInternalSchemaToBuildAvroSchema
- */
- private static Schema visitInternalArrayToBuildAvroArray(Types.ArrayType array, Schema elementSchema) {
- Schema result;
- if (array.isElementOptional()) {
- result = Schema.createArray(AvroInternalSchemaConverter.nullableSchema(elementSchema));
- } else {
- result = Schema.createArray(elementSchema);
- }
- return result;
- }
-
- /**
- * Converts hudi MapType to Avro MapType.
- * this is auxiliary function used by visitInternalSchemaToBuildAvroSchema
- */
- private static Schema visitInternalMapToBuildAvroMap(Types.MapType map, Schema keySchema, Schema valueSchema) {
- Schema mapSchema;
- if (keySchema.getType() == Schema.Type.STRING) {
- mapSchema = Schema.createMap(map.isValueOptional() ? AvroInternalSchemaConverter.nullableSchema(valueSchema) : valueSchema);
- } else {
- throw new HoodieSchemaException("only support StringType key for avro MapType");
- }
- return mapSchema;
- }
-
- /**
- * Converts hudi PrimitiveType to Avro PrimitiveType.
- * this is auxiliary function used by visitInternalSchemaToBuildAvroSchema
- */
- private static Schema visitInternalPrimitiveToBuildAvroPrimitiveType(Type.PrimitiveType primitive, String recordName) {
- switch (primitive.typeId()) {
- case BOOLEAN:
- return Schema.create(Schema.Type.BOOLEAN);
-
- case INT:
- return Schema.create(Schema.Type.INT);
-
- case LONG:
- return Schema.create(Schema.Type.LONG);
-
- case FLOAT:
- return Schema.create(Schema.Type.FLOAT);
-
- case DOUBLE:
- return Schema.create(Schema.Type.DOUBLE);
-
- case DATE:
- return LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT));
-
- case TIME:
- return LogicalTypes.timeMicros().addToSchema(Schema.create(Schema.Type.LONG));
-
- case TIME_MILLIS:
- return LogicalTypes.timeMillis().addToSchema(Schema.create(Schema.Type.INT));
-
- case TIMESTAMP:
- return LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG));
-
- case TIMESTAMP_MILLIS:
- return LogicalTypes.timestampMillis().addToSchema(Schema.create(Schema.Type.LONG));
-
- case LOCAL_TIMESTAMP_MICROS:
- return LogicalTypes.localTimestampMicros().addToSchema(Schema.create(Schema.Type.LONG));
-
- case LOCAL_TIMESTAMP_MILLIS:
- return LogicalTypes.localTimestampMillis().addToSchema(Schema.create(Schema.Type.LONG));
-
- case STRING:
- return Schema.create(Schema.Type.STRING);
-
- case BINARY:
- return Schema.create(Schema.Type.BYTES);
-
- case UUID: {
- // NOTE: All schemas corresponding to Avro's type [[FIXED]] are generated
- // with the "fixed" name to stay compatible w/ [[SchemaConverters]]
- String name = recordName + AVRO_NAME_DELIMITER + "fixed";
- Schema fixedSchema = Schema.createFixed(name, null, null, 16);
- return LogicalTypes.uuid().addToSchema(fixedSchema);
- }
-
- case FIXED: {
- Types.FixedType fixed = (Types.FixedType) primitive;
- // NOTE: All schemas corresponding to Avro's type [[FIXED]] are generated
- // with the "fixed" name to stay compatible w/ [[SchemaConverters]]
- String name = recordName + AVRO_NAME_DELIMITER + "fixed";
- return Schema.createFixed(name, null, null, fixed.getFixedSize());
- }
-
- case DECIMAL:
- case DECIMAL_FIXED: {
- Types.DecimalTypeFixed decimal = (Types.DecimalTypeFixed) primitive;
- // NOTE: All schemas corresponding to Avro's type [[FIXED]] are generated
- // with the "fixed" name to stay compatible w/ [[SchemaConverters]]
- String name = recordName + AVRO_NAME_DELIMITER + "fixed";
- Schema fixedSchema = Schema.createFixed(name,
- null, null, decimal.getFixedSize());
- return LogicalTypes.decimal(decimal.precision(), decimal.scale())
- .addToSchema(fixedSchema);
- }
-
- case DECIMAL_BYTES: {
- Types.DecimalTypeBytes decimal = (Types.DecimalTypeBytes) primitive;
- return LogicalTypes.decimal(decimal.precision(), decimal.scale())
- .addToSchema(Schema.create(Schema.Type.BYTES));
- }
-
- default:
- throw new UnsupportedOperationException(
- "Unsupported type ID: " + primitive.typeId());
- }
- }
-}
diff --git a/hudi-common/src/main/java/org/apache/hudi/internal/schema/convert/InternalSchemaConverter.java b/hudi-common/src/main/java/org/apache/hudi/internal/schema/convert/InternalSchemaConverter.java
new file mode 100644
index 000000000000..d055e16c58ce
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/internal/schema/convert/InternalSchemaConverter.java
@@ -0,0 +1,568 @@
+/*
+ * 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.internal.schema.convert;
+
+import org.apache.hudi.common.schema.HoodieJsonProperties;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaField;
+import org.apache.hudi.common.schema.HoodieSchemaType;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.VisibleForTesting;
+import org.apache.hudi.exception.HoodieNullSchemaTypeException;
+import org.apache.hudi.internal.schema.HoodieSchemaException;
+import org.apache.hudi.internal.schema.InternalSchema;
+import org.apache.hudi.internal.schema.Type;
+import org.apache.hudi.internal.schema.Types;
+import org.apache.hudi.internal.schema.utils.InternalSchemaUtils;
+
+import org.apache.avro.LogicalTypes;
+import org.apache.avro.Schema;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+/**
+ * Auxiliary class.
+ * Converts a HoodieSchema into InternalSchema, or convert InternalSchema to a HoodieSchema.
+ */
+public class InternalSchemaConverter {
+ private static final String FIELD_NAME_DELIMITER = ".";
+
+ /**
+ * Convert internalSchema to HoodieSchema.
+ *
+ * @param internalSchema internal schema.
+ * @param name the record name.
+ * @return an HoodieSchema.
+ */
+ public static HoodieSchema convert(InternalSchema internalSchema, String name) {
+ return buildHoodieSchemaFromInternalSchema(internalSchema, name);
+ }
+
+ public static InternalSchema pruneHoodieSchemaToInternalSchema(HoodieSchema schema, InternalSchema originSchema) {
+ List<String> pruneNames = collectColNamesFromSchema(schema);
+ return InternalSchemaUtils.pruneInternalSchema(originSchema, pruneNames);
+ }
+
+ /**
+ * Collect all the leaf nodes names.
+ *
+ * @param schema a HoodieSchema.
+ * @return leaf nodes full names.
+ */
+ @VisibleForTesting
+ static List<String> collectColNamesFromSchema(HoodieSchema schema) {
+ List<String> result = new ArrayList<>();
+ Deque<String> visited = new LinkedList<>();
+ collectColNamesFromSchema(schema, visited, result);
+ return result;
+ }
+
+ private static void collectColNamesFromSchema(HoodieSchema schema, Deque<String> visited, List<String> resultSet) {
+ switch (schema.getType()) {
+ case RECORD:
+ List<HoodieSchemaField> fields = schema.getFields();
+ for (HoodieSchemaField f : fields) {
+ visited.push(f.name());
+ collectColNamesFromSchema(f.schema(), visited, resultSet);
+ visited.pop();
+ addFullNameIfLeafNode(f.schema(), f.name(), visited, resultSet);
+ }
+ return;
+
+ case UNION:
+ collectColNamesFromSchema(schema.getNonNullType(), visited, resultSet);
+ return;
+
+ case ARRAY:
+ visited.push("element");
+ collectColNamesFromSchema(schema.getElementType(), visited, resultSet);
+ visited.pop();
+ addFullNameIfLeafNode(schema.getElementType(), "element", visited, resultSet);
+ return;
+
+ case MAP:
+ addFullNameIfLeafNode(HoodieSchemaType.STRING, "key", visited, resultSet);
+ visited.push("value");
+ collectColNamesFromSchema(schema.getValueType(), visited, resultSet);
+ visited.pop();
+ addFullNameIfLeafNode(schema.getValueType(), "value", visited, resultSet);
+ return;
+
+ default:
+ }
+ }
+
+ private static void addFullNameIfLeafNode(HoodieSchema schema, String name, Deque<String> visited, List<String> resultSet) {
+ addFullNameIfLeafNode(schema.getNonNullType().getType(), name, visited, resultSet);
+ }
+
+ private static void addFullNameIfLeafNode(HoodieSchemaType type, String name, Deque<String> visited, List<String> resultSet) {
+ switch (type) {
+ case RECORD:
+ case ARRAY:
+ case MAP:
+ return;
+ default:
+ resultSet.add(InternalSchemaUtils.createFullName(name, visited));
+ }
+ }
+
+ /**
+ * Converting from HoodieSchema -> internal schema -> HoodieSchema
+ * causes null to always be first in unions.
+ * if we compare a schema that has not been converted to internal schema
+ * at any stage, the difference in ordering can cause issues. To resolve this,
+ * we order null to be first for any HoodieSchema that enters into hudi.
+ * AvroSchemaUtils.isProjectionOfInternal uses index based comparison for unions.
+ * Spark and flink don't support complex unions so this would not be an issue
+ * but for the metadata table HoodieMetadata.avsc uses a trick where we have a bunch of
+ * different types wrapped in record for col stats.
+ *
+ * @param schema HoodieSchema.
+ * @return a HoodieSchema where null is the first.
+ */
+ public static HoodieSchema fixNullOrdering(HoodieSchema schema) {
+ if (schema == null) {
+ return HoodieSchema.create(HoodieSchemaType.NULL);
+ } else if (schema.getType() == HoodieSchemaType.NULL) {
+ return schema;
+ }
+ return convert(convert(schema), schema.getFullName());
+ }
+
+ /**
+ * Convert RecordType to HoodieSchema.
+ *
+ * @param type internal schema.
+ * @param name the record name.
+ * @return a HoodieSchema.
+ */
+ public static HoodieSchema convert(Types.RecordType type, String name) {
+ return buildSchemaFromType(type, name);
+ }
+
+ /**
+ * Convert internal type to HoodieSchema.
+ *
+ * @param type internal type.
+ * @param name the record name.
+ * @return a HoodieSchema.
+ */
+ public static HoodieSchema convert(Type type, String name) {
+ return buildSchemaFromType(type, name);
+ }
+
+ /** Convert a HoodieSchema into internal type. */
+ public static Type convertToField(HoodieSchema schema) {
+ return buildTypeFromHoodieSchema(schema, Collections.emptyMap());
+ }
+
+ private static Type convertToField(HoodieSchema schema, Map<String, Integer> existingFieldNameToPositionMapping) {
+ return buildTypeFromHoodieSchema(schema, existingFieldNameToPositionMapping);
+ }
+
+ /** Convert a HoodieSchema into internalSchema. */
+ public static InternalSchema convert(HoodieSchema schema, Map<String, Integer> existingFieldNameToPositionMapping) {
+ return new InternalSchema((Types.RecordType) convertToField(schema, existingFieldNameToPositionMapping));
+ }
+
+ public static InternalSchema convert(HoodieSchema schema) {
+ return new InternalSchema((Types.RecordType) convertToField(schema));
+ }
+
+ /**
+ * Build hudi type from HoodieSchema.
+ *
+ * @param schema a HoodieSchema.
+ * @return a hudi type.
+ */
+ public static Type buildTypeFromHoodieSchema(HoodieSchema schema, Map<String, Integer> existingNameToPositions) {
+ // set flag to check this has not been visited.
+ Deque<String> visited = new LinkedList<>();
+ AtomicInteger nextId = new AtomicInteger(0);
+ return visitSchemaToBuildType(schema, visited, "", nextId, existingNameToPositions);
+ }
+
+ private static void checkNullType(Type fieldType, String fieldName, Deque<String> visited) {
+ if (fieldType == null) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("Field '");
+ Iterator<String> visitedIterator = visited.descendingIterator();
+ while (visitedIterator.hasNext()) {
+ sb.append(visitedIterator.next());
+ sb.append(".");
+ }
+ sb.append(fieldName);
+ sb.append("' has type null");
+ throw new HoodieNullSchemaTypeException(sb.toString());
+ } else if (fieldType.typeId() == Type.TypeID.ARRAY) {
+ visited.push(fieldName);
+ checkNullType(((Types.ArrayType) fieldType).elementType(), InternalSchema.ARRAY_ELEMENT, visited);
+ visited.pop();
+ } else if (fieldType.typeId() == Type.TypeID.MAP) {
+ visited.push(fieldName);
+ checkNullType(((Types.MapType) fieldType).valueType(), InternalSchema.MAP_VALUE, visited);
+ visited.pop();
+ }
+ }
+
+ /**
+ * Converts an HoodieSchema into internal type.
+ *
+ * @param schema a HoodieSchema.
+ * @param visited track the visit node when do traversal for HoodieSchema; used to check if the name of record schema is correct.
+ * @param currentFieldPath the dot-separated path to the current field; empty at the root and always ends in a '.' otherwise for ease of concatenation.
+ * @param nextId an initial id which used to create id for all fields.
+ * @return a hudi type match HoodieSchema.
+ */
+ private static Type visitSchemaToBuildType(HoodieSchema schema, Deque<String> visited, String currentFieldPath, AtomicInteger nextId, Map<String, Integer> existingNameToPosition) {
+ switch (schema.getType()) {
+ case RECORD:
+ String name = schema.getFullName();
+ if (visited.contains(name)) {
+ throw new HoodieSchemaException(String.format("cannot convert recursive record %s", name));
+ }
+ visited.push(name);
+ List<HoodieSchemaField> fields = existingNameToPosition.isEmpty() ? schema.getFields() :
+ schema.getFields().stream()
+ .sorted(Comparator.comparing(field -> existingNameToPosition.getOrDefault(currentFieldPath + field.name(), Integer.MAX_VALUE)))
+ .collect(Collectors.toList());
+ List<Type> fieldTypes = new ArrayList<>(fields.size());
+ int nextAssignId = nextId.get();
+ nextId.set(nextAssignId + fields.size());
+ fields.forEach(field -> {
+ Type fieldType = visitSchemaToBuildType(field.schema(), visited, currentFieldPath + field.name() + ".", nextId, existingNameToPosition);
+ checkNullType(fieldType, field.name(), visited);
+ fieldTypes.add(fieldType);
+ });
+ visited.pop();
+ List<Types.Field> internalFields = new ArrayList<>(fields.size());
+
+ for (int i = 0; i < fields.size(); i++) {
+ HoodieSchemaField field = fields.get(i);
+ Type fieldType = fieldTypes.get(i);
+ internalFields.add(Types.Field.get(nextAssignId, field.isNullable(), field.name(), fieldType, field.doc().orElse(null)));
+ nextAssignId += 1;
+ }
+ // NOTE: We're keeping a tab of full-name here to make sure we stay
+ // compatible across various Spark (>= 2.4) and Avro (>= 1.8.2) versions;
+ // Avro will be properly handling fully-qualified names on its own (splitting
+ // them up into namespace/struct-name pair)
+ return Types.RecordType.get(internalFields, name);
+ case UNION:
+ List<Type> fTypes = new ArrayList<>(2);
+ schema.getTypes().forEach(t -> {
+ fTypes.add(visitSchemaToBuildType(t, visited, currentFieldPath, nextId, existingNameToPosition));
+ });
+ return fTypes.get(0) == null ? fTypes.get(1) : fTypes.get(0);
+ case ARRAY:
+ String elementPath = currentFieldPath + InternalSchema.ARRAY_ELEMENT + ".";
+ HoodieSchema elementSchema = schema.getElementType();
+ int elementId = nextId.get();
+ nextId.set(elementId + 1);
+ Type elementType = visitSchemaToBuildType(elementSchema, visited, elementPath, nextId, existingNameToPosition);
+ return Types.ArrayType.get(elementId, schema.getElementType().isNullable(), elementType);
+ case MAP:
+ int keyId = nextId.get();
+ int valueId = keyId + 1;
+ nextId.set(valueId + 1);
+ String valuePath = currentFieldPath + InternalSchema.MAP_VALUE + ".";
+ Type valueType = visitSchemaToBuildType(schema.getValueType(), visited, valuePath, nextId, existingNameToPosition);
+ return Types.MapType.get(keyId, valueId, Types.StringType.get(), valueType, schema.getValueType().isNullable());
+ default:
+ return visitPrimitiveToBuildInternalType(schema);
+ }
+ }
+
+ private static Type visitPrimitiveToBuildInternalType(HoodieSchema schema) {
+ switch (schema.getType()) {
+ case BOOLEAN:
+ return Types.BooleanType.get();
+ case INT:
+ return Types.IntType.get();
+ case LONG:
+ return Types.LongType.get();
+ case FLOAT:
+ return Types.FloatType.get();
+ case DOUBLE:
+ return Types.DoubleType.get();
+ case STRING:
+ case ENUM:
+ return Types.StringType.get();
+ case FIXED:
+ return Types.FixedType.getFixed(schema.getFixedSize());
+ case BYTES:
+ return Types.BinaryType.get();
+ case UUID:
+ return Types.UUIDType.get();
+ case DECIMAL:
+ HoodieSchema.Decimal decimalSchema = (HoodieSchema.Decimal) schema;
+ if (decimalSchema.isFixed()) {
+ return Types.DecimalTypeFixed.get(decimalSchema.getPrecision(), decimalSchema.getScale(), decimalSchema.getFixedSize());
+ }
+ return Types.DecimalTypeBytes.get(decimalSchema.getPrecision(), decimalSchema.getScale());
+ case TIME:
+ HoodieSchema.Time timeSchema = (HoodieSchema.Time) schema;
+ if (timeSchema.getPrecision() == HoodieSchema.TimePrecision.MICROS) {
+ return Types.TimeType.get();
+ } else if (timeSchema.getPrecision() == HoodieSchema.TimePrecision.MILLIS) {
+ return Types.TimeMillisType.get();
+ } else {
+ throw new UnsupportedOperationException("Unsupported time precision: " + timeSchema.getPrecision());
+ }
+ case TIMESTAMP:
+ HoodieSchema.Timestamp timestampSchema = (HoodieSchema.Timestamp) schema;
+ if (timestampSchema.getPrecision() == HoodieSchema.TimePrecision.MICROS) {
+ return timestampSchema.isUtcAdjusted() ? Types.TimestampType.get() : Types.LocalTimestampMicrosType.get();
+ } else if (timestampSchema.getPrecision() == HoodieSchema.TimePrecision.MILLIS) {
+ return timestampSchema.isUtcAdjusted() ? Types.TimestampMillisType.get() : Types.LocalTimestampMillisType.get();
+ } else {
+ throw new UnsupportedOperationException("Unsupported timestamp precision: " + timestampSchema.getPrecision());
+ }
+ case DATE:
+ return Types.DateType.get();
+ case NULL:
+ return null;
+ default:
+ throw new UnsupportedOperationException("Unsupported primitive type: " + schema.getType());
+ }
+ }
+
+ /**
+ * Converts hudi type into a HoodieSchema.
+ *
+ * @param type a hudi type.
+ * @param recordName the record name
+ * @return a HoodieSchema match this type
+ */
+ private static HoodieSchema buildSchemaFromType(Type type, String recordName) {
+ Map<Type, HoodieSchema> cache = new HashMap<>();
+ return visitInternalSchemaToBuildHoodieSchema(type, cache, recordName);
+ }
+
+ /**
+ * Converts hudi internal Schema into a HoodieSchema.
+ *
+ * @param schema a hudi internal Schema.
+ * @param recordName the record name
+ * @return a HoodieSchema match hudi internal schema.
+ */
+ public static HoodieSchema buildHoodieSchemaFromInternalSchema(InternalSchema schema, String recordName) {
+ Map<Type, HoodieSchema> cache = new HashMap<>();
+ return visitInternalSchemaToBuildHoodieSchema(schema.getRecord(), cache, recordName);
+ }
+
+ /**
+ * Converts hudi type into a HoodieSchema.
+ *
+ * @param type a hudi type.
+ * @param cache use to cache intermediate convert result to save cost.
+ * @param recordName auto-generated record name used as a fallback, in case
+ * {@link org.apache.hudi.internal.schema.Types.RecordType} doesn't bear original record-name
+ * @return a HoodieSchema match this type
+ */
+ private static HoodieSchema visitInternalSchemaToBuildHoodieSchema(Type type, Map<Type, HoodieSchema> cache, String recordName) {
+ switch (type.typeId()) {
+ case RECORD:
+ Types.RecordType record = (Types.RecordType) type;
+ List<HoodieSchema> schemas = new ArrayList<>();
+ record.fields().forEach(f -> {
+ String nestedRecordName = recordName + FIELD_NAME_DELIMITER + f.name();
+ HoodieSchema tempSchema = visitInternalSchemaToBuildHoodieSchema(f.type(), cache, nestedRecordName);
+ // convert tempSchema
+ HoodieSchema result = f.isOptional() ? HoodieSchema.createNullable(tempSchema) : tempSchema;
+ schemas.add(result);
+ });
+ // check visited
+ HoodieSchema recordSchema;
+ recordSchema = cache.get(record);
+ if (recordSchema != null) {
+ return recordSchema;
+ }
+ recordSchema = visitInternalRecordToBuildHoodieRecord(record, schemas, recordName);
+ cache.put(record, recordSchema);
+ return recordSchema;
+ case ARRAY:
+ Types.ArrayType array = (Types.ArrayType) type;
+ HoodieSchema elementSchema;
+ elementSchema = visitInternalSchemaToBuildHoodieSchema(array.elementType(), cache, recordName);
+ HoodieSchema arraySchema;
+ arraySchema = cache.get(array);
+ if (arraySchema != null) {
+ return arraySchema;
+ }
+ arraySchema = visitInternalArrayToBuildHoodieArray(array, elementSchema);
+ cache.put(array, arraySchema);
+ return arraySchema;
+ case MAP:
+ Types.MapType map = (Types.MapType) type;
+ HoodieSchema keySchema = visitInternalSchemaToBuildHoodieSchema(map.keyType(), cache, recordName);
+ HoodieSchema valueSchema = visitInternalSchemaToBuildHoodieSchema(map.valueType(), cache, recordName);
+ HoodieSchema mapSchema = cache.get(map);
+ if (mapSchema != null) {
+ return mapSchema;
+ }
+ mapSchema = visitInternalMapToBuildHoodieMap(map, keySchema, valueSchema);
+ cache.put(map, mapSchema);
+ return mapSchema;
+ default:
+ HoodieSchema primitiveSchema = visitInternalPrimitiveToBuildHoodiePrimitiveType((Type.PrimitiveType) type, recordName);
+ cache.put(type, primitiveSchema);
+ return primitiveSchema;
+ }
+ }
+
+ /**
+ * Converts hudi RecordType to Hoodie RecordType.
+ * this is auxiliary function used by visitInternalSchemaToBuildHoodieSchema
+ */
+ private static HoodieSchema visitInternalRecordToBuildHoodieRecord(Types.RecordType recordType, List<HoodieSchema> fieldSchemas, String recordNameFallback) {
+ List<Types.Field> fields = recordType.fields();
+ List<HoodieSchemaField> schemaFields = new ArrayList<>(fields.size());
+ for (int i = 0; i < fields.size(); i++) {
+ Types.Field f = fields.get(i);
+ HoodieSchemaField field = HoodieSchemaField.of(f.name(), fieldSchemas.get(i), f.doc(), f.isOptional() ? HoodieJsonProperties.NULL_VALUE : null);
+ schemaFields.add(field);
+ }
+ String recordName = Option.ofNullable(recordType.name()).orElse(recordNameFallback);
+ return HoodieSchema.createRecord(recordName, null, null, false, schemaFields);
+ }
+
+ /**
+ * Converts hudi ArrayType to Hoodie ArrayType.
+ * this is auxiliary function used by visitInternalSchemaToBuildHoodieSchema
+ */
+ private static HoodieSchema visitInternalArrayToBuildHoodieArray(Types.ArrayType array, HoodieSchema elementSchema) {
+ HoodieSchema result;
+ if (array.isElementOptional()) {
+ result = HoodieSchema.createArray(HoodieSchema.createNullable(elementSchema));
+ } else {
+ result = HoodieSchema.createArray(elementSchema);
+ }
+ return result;
+ }
+
+ /**
+ * Converts hudi MapType to Hoodie MapType.
+ * this is auxiliary function used by visitInternalSchemaToBuildHoodieSchema
+ */
+ private static HoodieSchema visitInternalMapToBuildHoodieMap(Types.MapType map, HoodieSchema keySchema, HoodieSchema valueSchema) {
+ HoodieSchema mapSchema;
+ if (keySchema.getType() == HoodieSchemaType.STRING) {
+ mapSchema = HoodieSchema.createMap(map.isValueOptional() ? HoodieSchema.createNullable(valueSchema) : valueSchema);
+ } else {
+ throw new HoodieSchemaException("only support StringType key for Hoodie MapType");
+ }
+ return mapSchema;
+ }
+
+ /**
+ * Converts hudi PrimitiveType to Hoodie PrimitiveType.
+ * this is auxiliary function used by visitInternalSchemaToBuildHoodieSchema
+ */
+ private static HoodieSchema visitInternalPrimitiveToBuildHoodiePrimitiveType(Type.PrimitiveType primitive, String recordName) {
+ switch (primitive.typeId()) {
+ case BOOLEAN:
+ return HoodieSchema.create(HoodieSchemaType.BOOLEAN);
+
+ case INT:
+ return HoodieSchema.create(HoodieSchemaType.INT);
+
+ case LONG:
+ return HoodieSchema.create(HoodieSchemaType.LONG);
+
+ case FLOAT:
+ return HoodieSchema.create(HoodieSchemaType.FLOAT);
+
+ case DOUBLE:
+ return HoodieSchema.create(HoodieSchemaType.DOUBLE);
+
+ case DATE:
+ return HoodieSchema.fromAvroSchema(LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT)));
+
+ case TIME:
+ return HoodieSchema.fromAvroSchema(LogicalTypes.timeMicros().addToSchema(Schema.create(Schema.Type.LONG)));
+
+ case TIME_MILLIS:
+ return HoodieSchema.fromAvroSchema(LogicalTypes.timeMillis().addToSchema(Schema.create(Schema.Type.INT)));
+
+ case TIMESTAMP:
+ return HoodieSchema.fromAvroSchema(LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG)));
+
+ case TIMESTAMP_MILLIS:
+ return HoodieSchema.fromAvroSchema(LogicalTypes.timestampMillis().addToSchema(Schema.create(Schema.Type.LONG)));
+
+ case LOCAL_TIMESTAMP_MICROS:
+ return HoodieSchema.fromAvroSchema(LogicalTypes.localTimestampMicros().addToSchema(Schema.create(Schema.Type.LONG)));
+
+ case LOCAL_TIMESTAMP_MILLIS:
+ return HoodieSchema.fromAvroSchema(LogicalTypes.localTimestampMillis().addToSchema(Schema.create(Schema.Type.LONG)));
+
+ case STRING:
+ return HoodieSchema.create(HoodieSchemaType.STRING);
+
+ case BINARY:
+ return HoodieSchema.create(HoodieSchemaType.BYTES);
+
+ case UUID: {
+ // NOTE: All schemas corresponding to Hoodie's type [[FIXED]] are generated
+ // with the "fixed" name to stay compatible w/ [[SchemaConverters]]
+ String name = recordName + FIELD_NAME_DELIMITER + "fixed";
+ Schema fixedSchema = Schema.createFixed(name, null, null, 16);
+ return HoodieSchema.fromAvroSchema(LogicalTypes.uuid().addToSchema(fixedSchema));
+ }
+
+ case FIXED: {
+ Types.FixedType fixed = (Types.FixedType) primitive;
+ // NOTE: All schemas corresponding to Hoodie's type [[FIXED]] are generated
+ // with the "fixed" name to stay compatible w/ [[SchemaConverters]]
+ String name = recordName + FIELD_NAME_DELIMITER + "fixed";
+ return HoodieSchema.createFixed(name, null, null, fixed.getFixedSize());
+ }
+
+ case DECIMAL:
+ case DECIMAL_FIXED: {
+ Types.DecimalTypeFixed decimal = (Types.DecimalTypeFixed) primitive;
+ // NOTE: All schemas corresponding to Hoodie's type [[FIXED]] are generated
+ // with the "fixed" name to stay compatible w/ [[SchemaConverters]]
+ String name = recordName + FIELD_NAME_DELIMITER + "fixed";
+ return HoodieSchema.createDecimal(name, null, null, decimal.precision(), decimal.scale(), decimal.getFixedSize());
+ }
+
+ case DECIMAL_BYTES: {
+ Types.DecimalTypeBytes decimal = (Types.DecimalTypeBytes) primitive;
+ return HoodieSchema.createDecimal(decimal.precision(), decimal.scale());
+ }
+
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported type ID: " + primitive.typeId());
+ }
+ }
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/internal/schema/utils/AvroSchemaEvolutionUtils.java b/hudi-common/src/main/java/org/apache/hudi/internal/schema/utils/AvroSchemaEvolutionUtils.java
index 44d36fabe025..5ae6f203f0f3 100644
--- a/hudi-common/src/main/java/org/apache/hudi/internal/schema/utils/AvroSchemaEvolutionUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/internal/schema/utils/AvroSchemaEvolutionUtils.java
@@ -19,6 +19,7 @@
package org.apache.hudi.internal.schema.utils;
import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.internal.schema.InternalSchema;
import org.apache.hudi.internal.schema.action.TableChanges;
import org.apache.hudi.internal.schema.action.TableChangesHelper;
@@ -35,7 +36,7 @@ import java.util.TreeMap;
import java.util.stream.Collectors;
import static org.apache.hudi.common.util.CollectionUtils.reduce;
-import static org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter.convert;
+import static org.apache.hudi.internal.schema.convert.InternalSchemaConverter.convert;
/**
* Utility methods to support evolve old avro schema based on a given schema.
@@ -68,7 +69,7 @@ public class AvroSchemaEvolutionUtils {
if (incomingSchema.getType() == Schema.Type.NULL) {
return oldTableSchema;
}
- InternalSchema inComingInternalSchema = convert(incomingSchema, oldTableSchema.getNameToPosition());
+ InternalSchema inComingInternalSchema = convert(HoodieSchema.fromAvroSchema(incomingSchema), oldTableSchema.getNameToPosition());
// check column add/missing
List<String> colNamesFromIncoming = inComingInternalSchema.getAllColsFullName();
List<String> colNamesFromOldSchema = oldTableSchema.getAllColsFullName();
@@ -149,7 +150,7 @@ public class AvroSchemaEvolutionUtils {
}
public static Schema reconcileSchema(Schema incomingSchema, Schema oldTableSchema, boolean makeMissingFieldsNullable) {
- return convert(reconcileSchema(incomingSchema, convert(oldTableSchema), makeMissingFieldsNullable), oldTableSchema.getFullName());
+ return convert(reconcileSchema(incomingSchema, convert(HoodieSchema.fromAvroSchema(oldTableSchema)), makeMissingFieldsNullable), oldTableSchema.getFullName()).toAvroSchema();
}
/**
@@ -177,9 +178,9 @@ public class AvroSchemaEvolutionUtils {
return targetSchema;
}
- InternalSchema targetInternalSchema = convert(targetSchema);
+ InternalSchema targetInternalSchema = convert(HoodieSchema.fromAvroSchema(targetSchema));
// Use existing fieldIds for consistent field ordering between commits when shouldReorderColumns is true
- InternalSchema sourceInternalSchema = convert(sourceSchema, shouldReorderColumns ? targetInternalSchema.getNameToPosition() : Collections.emptyMap());
+ InternalSchema sourceInternalSchema = convert(HoodieSchema.fromAvroSchema(sourceSchema), shouldReorderColumns ? targetInternalSchema.getNameToPosition() : Collections.emptyMap());
List<String> colNamesSourceSchema = sourceInternalSchema.getAllColsFullName();
List<String> colNamesTargetSchema = targetInternalSchema.getAllColsFullName();
@@ -199,7 +200,7 @@ public class AvroSchemaEvolutionUtils {
if (nullableUpdateColsInSource.isEmpty() && typeUpdateColsInSource.isEmpty()) {
//standardize order of unions
- return convert(sourceInternalSchema, sourceSchema.getFullName());
+ return convert(sourceInternalSchema, sourceSchema.getFullName()).toAvroSchema();
}
TableChanges.ColumnUpdateChange schemaChange = TableChanges.ColumnUpdateChange.get(sourceInternalSchema);
@@ -217,7 +218,7 @@ public class AvroSchemaEvolutionUtils {
}
- return convert(SchemaChangeUtils.applyTableChanges2Schema(sourceInternalSchema, schemaChange), sourceSchema.getFullName());
+ return convert(SchemaChangeUtils.applyTableChanges2Schema(sourceInternalSchema, schemaChange), sourceSchema.getFullName()).toAvroSchema();
}
}
diff --git a/hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java b/hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java
index 050d2924de7c..f7a9529f5b78 100644
--- a/hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java
+++ b/hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java
@@ -923,7 +923,7 @@ public class TestHoodieAvroUtils {
@Test
void testConvertBytesToFixed() {
Random rand = new Random();
- //size calculated using AvroInternalSchemaConverter.computeMinBytesForPrecision
+ //size calculated using InternalSchemaConverter.computeMinBytesForPrecision
testConverBytesToFixedHelper(rand.nextDouble(), 13, 7, 6);
testConverBytesToFixedHelper(rand.nextDouble(), 4, 2, 2);
testConverBytesToFixedHelper(rand.nextDouble(), 32, 12, 14);
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/schema/HoodieSchemaTestUtils.java b/hudi-common/src/test/java/org/apache/hudi/common/schema/HoodieSchemaTestUtils.java
new file mode 100644
index 000000000000..387dfca41865
--- /dev/null
+++ b/hudi-common/src/test/java/org/apache/hudi/common/schema/HoodieSchemaTestUtils.java
@@ -0,0 +1,70 @@
+/*
+ * 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.common.schema;
+
+import org.apache.avro.JsonProperties;
+
+import java.util.Arrays;
+
+public class HoodieSchemaTestUtils {
+ public static HoodieSchemaField createNestedField(String name, HoodieSchemaType type) {
+ return createNestedField(name, HoodieSchema.create(type));
+ }
+
+ public static HoodieSchemaField createNestedField(String name, HoodieSchema schema) {
+ return HoodieSchemaField.of(name, createRecord(name, HoodieSchemaField.of("nested", schema, null, null)), null, null);
+ }
+
+ public static HoodieSchemaField createArrayField(String name, HoodieSchemaType type) {
+ return createArrayField(name, HoodieSchema.create(type));
+ }
+
+ public static HoodieSchemaField createArrayField(String name, HoodieSchema schema) {
+ return HoodieSchemaField.of(name, HoodieSchema.createArray(schema), null, null);
+ }
+
+ public static HoodieSchemaField createNullableArrayField(String name, HoodieSchema schema) {
+ return HoodieSchemaField.of(name, HoodieSchema.createNullable(HoodieSchema.createArray(schema)), null, HoodieJsonProperties.NULL_VALUE);
+ }
+
+ public static HoodieSchemaField createMapField(String name, HoodieSchemaType type) {
+ return createMapField(name, HoodieSchema.create(type));
+ }
+
+ public static HoodieSchemaField createMapField(String name, HoodieSchema schema) {
+ return HoodieSchemaField.of(name, HoodieSchema.createMap(schema), null, null);
+ }
+
+ public static HoodieSchemaField createPrimitiveField(String name, HoodieSchemaType type) {
+ return HoodieSchemaField.of(name, HoodieSchema.create(type), null, null);
+ }
+
+ public static HoodieSchemaField createNullablePrimitiveField(String name, HoodieSchemaType type) {
+ return HoodieSchemaField.of(name, HoodieSchema.createNullable(HoodieSchema.create(type)), null, JsonProperties.NULL_VALUE);
+ }
+
+ public static HoodieSchema createRecord(String name, HoodieSchemaField... fields) {
+ return HoodieSchema.createRecord(name, null, null, false, Arrays.asList(fields));
+ }
+
+ public static HoodieSchema createNullableRecord(String name, HoodieSchemaField... fields) {
+ return HoodieSchema.createNullable(HoodieSchema.createRecord(name, null, null, false, Arrays.asList(fields)));
+ }
+}
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestFileGroupReaderSchemaHandler.java b/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestFileGroupReaderSchemaHandler.java
index 5d88a929759f..a1e1b572c71b 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestFileGroupReaderSchemaHandler.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestFileGroupReaderSchemaHandler.java
@@ -29,6 +29,7 @@ import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordMerger;
import org.apache.hudi.common.model.OverwriteNonDefaultsWithLatestAvroPayload;
import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload;
+import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.testutils.SchemaTestUtil;
import org.apache.hudi.common.util.InternalSchemaCache;
@@ -37,7 +38,7 @@ import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.internal.schema.InternalSchema;
import org.apache.hudi.internal.schema.Types;
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter;
import org.apache.hudi.storage.StoragePath;
import org.apache.avro.Schema;
@@ -106,7 +107,7 @@ public class TestFileGroupReaderSchemaHandler extends SchemaHandlerTestBase {
HoodieReaderContext<String> readerContext = createReaderContext(hoodieTableConfig, false, false, true, false, null);
Schema requestedSchema = generateProjectionSchema("_hoodie_record_key", "timestamp", "rider");
- InternalSchema internalSchema = AvroInternalSchemaConverter.convert(DATA_SCHEMA);
+ InternalSchema internalSchema = InternalSchemaConverter.convert(HoodieSchema.fromAvroSchema(DATA_SCHEMA));
InternalSchema originalSchema = new InternalSchema(Types.RecordType.get(internalSchema.columns().stream().map(field -> {
if (field.name().equals("timestamp")) {
// rename timestamp to ts in file schema and change type to int, output schema names and types must match the requested schema
diff --git a/hudi-common/src/test/java/org/apache/hudi/internal/schema/convert/TestAvroInternalSchemaConverter.java b/hudi-common/src/test/java/org/apache/hudi/internal/schema/convert/TestInternalSchemaConverter.java
similarity index 54%
rename from hudi-common/src/test/java/org/apache/hudi/internal/schema/convert/TestAvroInternalSchemaConverter.java
rename to hudi-common/src/test/java/org/apache/hudi/internal/schema/convert/TestInternalSchemaConverter.java
index 8bf7eabce029..1ec7927813f0 100644
--- a/hudi-common/src/test/java/org/apache/hudi/internal/schema/convert/TestAvroInternalSchemaConverter.java
+++ b/hudi-common/src/test/java/org/apache/hudi/internal/schema/convert/TestInternalSchemaConverter.java
@@ -19,62 +19,64 @@
package org.apache.hudi.internal.schema.convert;
-import org.apache.hudi.avro.AvroSchemaTestUtils;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaField;
+import org.apache.hudi.common.schema.HoodieSchemaType;
-import org.apache.avro.Schema;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.List;
-import static org.apache.hudi.avro.AvroSchemaTestUtils.createArrayField;
-import static org.apache.hudi.avro.AvroSchemaTestUtils.createMapField;
-import static org.apache.hudi.avro.AvroSchemaTestUtils.createNullablePrimitiveField;
-import static org.apache.hudi.avro.AvroSchemaTestUtils.createNullableRecord;
-import static org.apache.hudi.avro.AvroSchemaTestUtils.createPrimitiveField;
-import static org.apache.hudi.avro.AvroSchemaTestUtils.createRecord;
+import static org.apache.hudi.common.schema.HoodieSchemaTestUtils.createArrayField;
+import static org.apache.hudi.common.schema.HoodieSchemaTestUtils.createMapField;
+import static org.apache.hudi.common.schema.HoodieSchemaTestUtils.createNestedField;
+import static org.apache.hudi.common.schema.HoodieSchemaTestUtils.createNullablePrimitiveField;
+import static org.apache.hudi.common.schema.HoodieSchemaTestUtils.createNullableRecord;
+import static org.apache.hudi.common.schema.HoodieSchemaTestUtils.createPrimitiveField;
+import static org.apache.hudi.common.schema.HoodieSchemaTestUtils.createRecord;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
-public class TestAvroInternalSchemaConverter {
+public class TestInternalSchemaConverter {
- public static Schema getSimpleSchema() {
+ public static HoodieSchema getSimpleSchema() {
return createRecord("simpleSchema",
- createPrimitiveField("field1", Schema.Type.INT),
- createPrimitiveField("field2", Schema.Type.STRING));
+ createPrimitiveField("field1", HoodieSchemaType.INT),
+ createPrimitiveField("field2", HoodieSchemaType.STRING));
}
public static List<String> getSimpleSchemaExpectedColumnNames() {
return Arrays.asList("field1", "field2");
}
- public static Schema getSimpleSchemaWithNullable() {
+ public static HoodieSchema getSimpleSchemaWithNullable() {
return createRecord("simpleSchemaWithNullable",
- createNullablePrimitiveField("field1", Schema.Type.INT),
- createPrimitiveField("field2", Schema.Type.STRING));
+ createNullablePrimitiveField("field1", HoodieSchemaType.INT),
+ createPrimitiveField("field2", HoodieSchemaType.STRING));
}
- public static Schema getComplexSchemaSingleLevel() {
+ public static HoodieSchema getComplexSchemaSingleLevel() {
return createRecord("complexSchemaSingleLevel",
- AvroSchemaTestUtils.createNestedField("field1", Schema.Type.INT),
- createArrayField("field2", Schema.Type.STRING),
- createMapField("field3", Schema.Type.DOUBLE));
+ createNestedField("field1", HoodieSchemaType.INT),
+ createArrayField("field2", HoodieSchemaType.STRING),
+ createMapField("field3", HoodieSchemaType.DOUBLE));
}
public static List<String> getComplexSchemaSingleLevelExpectedColumnNames() {
return Arrays.asList("field1.nested", "field2.element", "field3.key", "field3.value");
}
- public static Schema getDeeplyNestedFieldSchema() {
+ public static HoodieSchema getDeeplyNestedFieldSchema() {
return createRecord("deeplyNestedFieldSchema",
- createPrimitiveField("field1", Schema.Type.INT),
- new Schema.Field("field2",
+ createPrimitiveField("field1", HoodieSchemaType.INT),
+ HoodieSchemaField.of("field2",
createRecord("field2nest",
createArrayField("field2nestarray",
createNullableRecord("field2nestarraynest",
- createNullablePrimitiveField("field21", Schema.Type.INT),
- createNullablePrimitiveField("field22", Schema.Type.INT)))), null, null),
- createNullablePrimitiveField("field3", Schema.Type.INT));
+ createNullablePrimitiveField("field21", HoodieSchemaType.INT),
+ createNullablePrimitiveField("field22", HoodieSchemaType.INT)))), null, null),
+ createNullablePrimitiveField("field3", HoodieSchemaType.INT));
}
public static List<String> getDeeplyNestedFieldSchemaExpectedColumnNames() {
@@ -84,27 +86,27 @@ public class TestAvroInternalSchemaConverter {
@Test
public void testCollectColumnNames() {
- Schema simpleSchema = getSimpleSchema();
- List<String> fieldNames = AvroInternalSchemaConverter.collectColNamesFromSchema(simpleSchema);
+ HoodieSchema simpleSchema = getSimpleSchema();
+ List<String> fieldNames = InternalSchemaConverter.collectColNamesFromSchema(simpleSchema);
List<String> expectedOutput = getSimpleSchemaExpectedColumnNames();
assertEquals(expectedOutput.size(), fieldNames.size());
assertTrue(fieldNames.containsAll(expectedOutput));
- Schema simpleSchemaWithNullable = getSimpleSchemaWithNullable();
- fieldNames = AvroInternalSchemaConverter.collectColNamesFromSchema(simpleSchemaWithNullable);
+ HoodieSchema simpleSchemaWithNullable = getSimpleSchemaWithNullable();
+ fieldNames = InternalSchemaConverter.collectColNamesFromSchema(simpleSchemaWithNullable);
expectedOutput = getSimpleSchemaExpectedColumnNames();
assertEquals(expectedOutput.size(), fieldNames.size());
assertTrue(fieldNames.containsAll(expectedOutput));
- Schema complexSchemaSingleLevel = getComplexSchemaSingleLevel();
- fieldNames = AvroInternalSchemaConverter.collectColNamesFromSchema(complexSchemaSingleLevel);
+ HoodieSchema complexSchemaSingleLevel = getComplexSchemaSingleLevel();
+ fieldNames = InternalSchemaConverter.collectColNamesFromSchema(complexSchemaSingleLevel);
expectedOutput = getComplexSchemaSingleLevelExpectedColumnNames();
assertEquals(expectedOutput.size(), fieldNames.size());
assertTrue(fieldNames.containsAll(expectedOutput));
- Schema deeplyNestedFieldSchema = getDeeplyNestedFieldSchema();
- fieldNames = AvroInternalSchemaConverter.collectColNamesFromSchema(deeplyNestedFieldSchema);
+ HoodieSchema deeplyNestedFieldSchema = getDeeplyNestedFieldSchema();
+ fieldNames = InternalSchemaConverter.collectColNamesFromSchema(deeplyNestedFieldSchema);
expectedOutput = getDeeplyNestedFieldSchemaExpectedColumnNames();
assertEquals(expectedOutput.size(), fieldNames.size());
assertTrue(fieldNames.containsAll(expectedOutput));
diff --git a/hudi-common/src/test/java/org/apache/hudi/internal/schema/utils/TestAvroSchemaEvolutionUtils.java b/hudi-common/src/test/java/org/apache/hudi/internal/schema/utils/TestAvroSchemaEvolutionUtils.java
index 19598e278499..2cdf6ae292f1 100644
--- a/hudi-common/src/test/java/org/apache/hudi/internal/schema/utils/TestAvroSchemaEvolutionUtils.java
+++ b/hudi-common/src/test/java/org/apache/hudi/internal/schema/utils/TestAvroSchemaEvolutionUtils.java
@@ -19,6 +19,10 @@
package org.apache.hudi.internal.schema.utils;
import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.common.schema.HoodieJsonProperties;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaField;
+import org.apache.hudi.common.schema.HoodieSchemaType;
import org.apache.hudi.common.testutils.SchemaTestUtil;
import org.apache.hudi.exception.HoodieNullSchemaTypeException;
import org.apache.hudi.internal.schema.InternalSchema;
@@ -26,12 +30,10 @@ import org.apache.hudi.internal.schema.InternalSchemaBuilder;
import org.apache.hudi.internal.schema.Type;
import org.apache.hudi.internal.schema.Types;
import org.apache.hudi.internal.schema.action.TableChanges;
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter;
-import org.apache.avro.JsonProperties;
import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
-import org.apache.avro.SchemaBuilder;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.junit.jupiter.api.Assertions;
@@ -68,20 +70,20 @@ public class TestAvroSchemaEvolutionUtils {
@Test
public void testPrimitiveTypes() {
- Schema[] avroPrimitives = new Schema[] {
- Schema.create(Schema.Type.BOOLEAN),
- Schema.create(Schema.Type.INT),
- Schema.create(Schema.Type.LONG),
- Schema.create(Schema.Type.FLOAT),
- Schema.create(Schema.Type.DOUBLE),
- LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT)),
- LogicalTypes.timeMicros().addToSchema(Schema.create(Schema.Type.LONG)),
- LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG)),
- Schema.create(Schema.Type.STRING),
- LogicalTypes.uuid().addToSchema(Schema.createFixed("t1.fixed", null, null, 16)),
- Schema.createFixed("t1.fixed", null, null, 12),
- Schema.create(Schema.Type.BYTES),
- LogicalTypes.decimal(9, 4).addToSchema(Schema.createFixed("t1.fixed", null, null, 4))};
+ HoodieSchema[] schemaPrimitives = new HoodieSchema[] {
+ HoodieSchema.create(HoodieSchemaType.BOOLEAN),
+ HoodieSchema.create(HoodieSchemaType.INT),
+ HoodieSchema.create(HoodieSchemaType.LONG),
+ HoodieSchema.create(HoodieSchemaType.FLOAT),
+ HoodieSchema.create(HoodieSchemaType.DOUBLE),
+ HoodieSchema.fromAvroSchema(LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT))),
+ HoodieSchema.fromAvroSchema(LogicalTypes.timeMicros().addToSchema(Schema.create(Schema.Type.LONG))),
+ HoodieSchema.fromAvroSchema(LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG))),
+ HoodieSchema.create(HoodieSchemaType.STRING),
+ HoodieSchema.fromAvroSchema(LogicalTypes.uuid().addToSchema(Schema.createFixed("t1.fixed", null, null, 16))),
+ HoodieSchema.createFixed("t1.fixed", null, null, 12),
+ HoodieSchema.create(HoodieSchemaType.BYTES),
+ HoodieSchema.fromAvroSchema(LogicalTypes.decimal(9, 4).addToSchema(Schema.createFixed("t1.fixed", null, null, 4)))};
Type[] primitiveTypes = new Type[] {
Types.BooleanType.get(),
@@ -100,10 +102,10 @@ public class TestAvroSchemaEvolutionUtils {
};
for (int i = 0; i < primitiveTypes.length; i++) {
- Type convertPrimitiveResult = AvroInternalSchemaConverter.convertToField(avroPrimitives[i]);
+ Type convertPrimitiveResult = InternalSchemaConverter.convertToField(schemaPrimitives[i]);
Assertions.assertEquals(convertPrimitiveResult, primitiveTypes[i]);
- Schema convertResult = AvroInternalSchemaConverter.convert(primitiveTypes[i], "t1");
- Assertions.assertEquals(convertResult, avroPrimitives[i]);
+ HoodieSchema convertResult = InternalSchemaConverter.convert(primitiveTypes[i], "t1");
+ Assertions.assertEquals(convertResult, schemaPrimitives[i]);
}
}
@@ -125,29 +127,33 @@ public class TestAvroSchemaEvolutionUtils {
Types.Field.get(12, "decimal", Types.DecimalType.get(10, 2))
}));
- Schema schema = create("t1",
- new Schema.Field("bool", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.BOOLEAN)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("int", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.INT)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("long", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.LONG)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("float", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.FLOAT)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("double", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.DOUBLE)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("date", AvroInternalSchemaConverter.nullableSchema(LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT))), null, JsonProperties.NULL_VALUE),
- new Schema.Field("time", AvroInternalSchemaConverter.nullableSchema(LogicalTypes.timeMicros().addToSchema(Schema.create(Schema.Type.LONG))), null, JsonProperties.NULL_VALUE),
- new Schema.Field("timestamp", AvroInternalSchemaConverter.nullableSchema(LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG))), null, JsonProperties.NULL_VALUE),
- new Schema.Field("string", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.STRING)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("uuid", AvroInternalSchemaConverter.nullableSchema(LogicalTypes.uuid().addToSchema(Schema.createFixed("t1.uuid.fixed", null, null, 16))), null, JsonProperties.NULL_VALUE),
- new Schema.Field("fixed", AvroInternalSchemaConverter.nullableSchema(Schema.createFixed("t1.fixed.fixed", null, null, 10)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("binary", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.BYTES)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("decimal", AvroInternalSchemaConverter.nullableSchema(LogicalTypes.decimal(10, 2)
- .addToSchema(Schema.createFixed("t1.decimal.fixed", null, null, 5))), null, JsonProperties.NULL_VALUE));
- Schema convertedSchema = AvroInternalSchemaConverter.convert(record, "t1");
+ HoodieSchema schema = create("t1",
+ HoodieSchemaField.of("bool", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.BOOLEAN)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("int", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.INT)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("long", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.LONG)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("float", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.FLOAT)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("double", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.DOUBLE)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("date", HoodieSchema.createNullable(HoodieSchema.fromAvroSchema(LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT)))),
+ null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("time", HoodieSchema.createNullable(HoodieSchema.fromAvroSchema(LogicalTypes.timeMicros().addToSchema(Schema.create(Schema.Type.LONG)))),
+ null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("timestamp", HoodieSchema.createNullable(HoodieSchema.fromAvroSchema(LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG)))),
+ null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("string", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.STRING)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("uuid", HoodieSchema.createNullable(HoodieSchema.fromAvroSchema(LogicalTypes.uuid().addToSchema(
+ Schema.createFixed("t1.uuid.fixed", null, null, 16)))), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("fixed", HoodieSchema.createNullable(HoodieSchema.createFixed("t1.fixed.fixed", null, null, 10)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("binary", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.BYTES)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("decimal", HoodieSchema.createNullable(HoodieSchema.fromAvroSchema(LogicalTypes.decimal(10, 2)
+ .addToSchema(Schema.createFixed("t1.decimal.fixed", null, null, 5)))), null, HoodieJsonProperties.NULL_VALUE));
+ HoodieSchema convertedSchema = InternalSchemaConverter.convert(record, "t1");
Assertions.assertEquals(convertedSchema, schema);
- Types.RecordType convertedRecord = AvroInternalSchemaConverter.convert(schema).getRecord();
+ Types.RecordType convertedRecord = InternalSchemaConverter.convert(schema).getRecord();
Assertions.assertEquals(convertedRecord, record);
}
- private Schema create(String name, Schema.Field... fields) {
- return Schema.createRecord(name, null, null, false, Arrays.asList(fields));
+ private HoodieSchema create(String name, HoodieSchemaField... fields) {
+ return HoodieSchema.createRecord(name, null, null, false, Arrays.asList(fields));
}
@Test
@@ -156,18 +162,18 @@ public class TestAvroSchemaEvolutionUtils {
Types.RecordType.get(Arrays.asList(Types.Field.get(1, false, "a", Types.FloatType.get()),
Types.Field.get(2, false, "b", Types.FloatType.get()))));
- Schema schema = SchemaBuilder.array().items(create("t1",
- new Schema.Field("a", Schema.create(Schema.Type.FLOAT), null, null),
- new Schema.Field("b", Schema.create(Schema.Type.FLOAT), null, null)));
- Schema convertedSchema = AvroInternalSchemaConverter.convert(arrayNestRecordType, "t1");
+ HoodieSchema schema = HoodieSchema.createArray(create("t1",
+ HoodieSchemaField.of("a", HoodieSchema.create(HoodieSchemaType.FLOAT), null, null),
+ HoodieSchemaField.of("b", HoodieSchema.create(HoodieSchemaType.FLOAT), null, null)));
+ HoodieSchema convertedSchema = InternalSchemaConverter.convert(arrayNestRecordType, "t1");
Assertions.assertEquals(convertedSchema, schema);
- Types.ArrayType convertedRecord = (Types.ArrayType) AvroInternalSchemaConverter.convertToField(schema);
+ Types.ArrayType convertedRecord = (Types.ArrayType) InternalSchemaConverter.convertToField(schema);
Assertions.assertEquals(convertedRecord, arrayNestRecordType);
}
@Test
public void testComplexConvert() {
- Schema schema = new Schema.Parser().parse(schemaStr);
+ HoodieSchema schema = HoodieSchema.parse(schemaStr);
Types.RecordType recordType = Types.RecordType.get(Types.Field.get(0, false, "id", Types.IntType.get()),
Types.Field.get(1, true, "data", Types.StringType.get()),
@@ -183,39 +189,39 @@ public class TestAvroSchemaEvolutionUtils {
);
InternalSchema internalSchema = new InternalSchema(recordType);
- Type convertRecord = AvroInternalSchemaConverter.convert(schema).getRecord();
+ Type convertRecord = InternalSchemaConverter.convert(schema).getRecord();
Assertions.assertEquals(convertRecord, internalSchema.getRecord());
- Assertions.assertEquals(schema, AvroInternalSchemaConverter.convert(internalSchema, "newTableName"));
+ Assertions.assertEquals(schema, InternalSchemaConverter.convert(internalSchema, "newTableName"));
}
@Test
public void testNullFieldType() {
- Schema schema = create("t1",
- new Schema.Field("nullField", Schema.create(Schema.Type.NULL), null, JsonProperties.NULL_VALUE));
+ HoodieSchema schema = create("t1",
+ HoodieSchemaField.of("nullField", HoodieSchema.create(HoodieSchemaType.NULL), null, HoodieJsonProperties.NULL_VALUE));
Throwable t = assertThrows(HoodieNullSchemaTypeException.class,
- () -> AvroInternalSchemaConverter.convert(schema));
+ () -> InternalSchemaConverter.convert(schema));
assertTrue(t.getMessage().contains("'t1.nullField'"));
- Schema schemaArray = create("t2",
- new Schema.Field("nullArray", Schema.createArray(Schema.create(Schema.Type.NULL)), null, null));
+ HoodieSchema schemaArray = create("t2",
+ HoodieSchemaField.of("nullArray", HoodieSchema.createArray(HoodieSchema.create(HoodieSchemaType.NULL)), null, null));
t = assertThrows(HoodieNullSchemaTypeException.class,
- () -> AvroInternalSchemaConverter.convert(schemaArray));
+ () -> InternalSchemaConverter.convert(schemaArray));
assertTrue(t.getMessage().contains("'t2.nullArray.element'"));
- Schema schemaMap = create("t3",
- new Schema.Field("nullMap", Schema.createMap(Schema.create(Schema.Type.NULL)), null, null));
+ HoodieSchema schemaMap = create("t3",
+ HoodieSchemaField.of("nullMap", HoodieSchema.createMap(HoodieSchema.create(HoodieSchemaType.NULL)), null, null));
t = assertThrows(HoodieNullSchemaTypeException.class,
- () -> AvroInternalSchemaConverter.convert(schemaMap));
+ () -> InternalSchemaConverter.convert(schemaMap));
assertTrue(t.getMessage().contains("'t3.nullMap.value'"));
- Schema schemaComplex = create("t4",
- new Schema.Field("complexField", Schema.createMap(
+ HoodieSchema schemaComplex = create("t4",
+ HoodieSchemaField.of("complexField", HoodieSchema.createMap(
create("nestedStruct",
- new Schema.Field("nestedArray", Schema.createArray(Schema.createMap(Schema.create(Schema.Type.NULL))),
+ HoodieSchemaField.of("nestedArray", HoodieSchema.createArray(HoodieSchema.createMap(HoodieSchema.create(HoodieSchemaType.NULL))),
null, null))), null, null));
t = assertThrows(HoodieNullSchemaTypeException.class,
- () -> AvroInternalSchemaConverter.convert(schemaComplex));
+ () -> InternalSchemaConverter.convert(schemaComplex));
assertTrue(t.getMessage().contains("'t4.nestedStruct.nestedArray.element.value'"));
}
@@ -245,16 +251,16 @@ public class TestAvroSchemaEvolutionUtils {
@Test
public void testFixNullOrdering() {
- Schema schema = SchemaTestUtil.getSchemaFromResource(TestAvroSchemaEvolutionUtils.class, "/nullWrong.avsc");
- Schema expectedSchema = SchemaTestUtil.getSchemaFromResource(TestAvroSchemaEvolutionUtils.class, "/nullRight.avsc");
- Assertions.assertEquals(expectedSchema, AvroInternalSchemaConverter.fixNullOrdering(schema));
- Assertions.assertEquals(expectedSchema, AvroInternalSchemaConverter.fixNullOrdering(expectedSchema));
+ HoodieSchema schema = HoodieSchema.fromAvroSchema(SchemaTestUtil.getSchemaFromResource(TestAvroSchemaEvolutionUtils.class, "/nullWrong.avsc"));
+ HoodieSchema expectedSchema = HoodieSchema.fromAvroSchema(SchemaTestUtil.getSchemaFromResource(TestAvroSchemaEvolutionUtils.class, "/nullRight.avsc"));
+ Assertions.assertEquals(expectedSchema, InternalSchemaConverter.fixNullOrdering(schema));
+ Assertions.assertEquals(expectedSchema, InternalSchemaConverter.fixNullOrdering(expectedSchema));
}
@Test
public void testFixNullOrderingSameSchemaCheck() {
- Schema schema = SchemaTestUtil.getSchemaFromResource(TestAvroSchemaEvolutionUtils.class, "/source_evolved.avsc");
- Assertions.assertEquals(schema, AvroInternalSchemaConverter.fixNullOrdering(schema));
+ HoodieSchema schema = HoodieSchema.fromAvroSchema(SchemaTestUtil.getSchemaFromResource(TestAvroSchemaEvolutionUtils.class, "/source_evolved.avsc"));
+ Assertions.assertEquals(schema, InternalSchemaConverter.fixNullOrdering(schema));
}
public enum Enum {
@@ -275,7 +281,7 @@ public class TestAvroSchemaEvolutionUtils {
@Test
public void testReWriteRecordWithTypeChanged() {
String enumSchema = "{\"type\":\"enum\",\"name\":\"Enum\",\"namespace\":\"org.apache.hudi.internal.schema.utils.TestAvroSchemaEvolutionUtils\",\"symbols\":[\"ENUM1\",\"ENUM2\"]}";
- Schema avroSchema = new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"h0_record\",\"namespace\":\"hoodie.h0\",\"fields\""
+ HoodieSchema hoodieSchema = HoodieSchema.parse("{\"type\":\"record\",\"name\":\"h0_record\",\"namespace\":\"hoodie.h0\",\"fields\""
+ ":[{\"name\":\"id\",\"type\":[\"null\",\"int\"],\"default\":null},"
+ "{\"name\":\"comb\",\"type\":[\"null\",\"int\"],\"default\":null},"
+ "{\"name\":\"com1\",\"type\":[\"null\",\"int\"],\"default\":null},"
@@ -300,7 +306,7 @@ public class TestAvroSchemaEvolutionUtils {
+ "{\"name\":\"enum\",\"type\":[\"null\"," + enumSchema + "],\"default\":null}"
+ "]}");
// create a test record with avroSchema
- GenericData.Record avroRecord = new GenericData.Record(avroSchema);
+ GenericData.Record avroRecord = new GenericData.Record(hoodieSchema.toAvroSchema());
avroRecord.put("id", 1);
avroRecord.put("comb", 100);
avroRecord.put("com1", -100);
@@ -312,10 +318,10 @@ public class TestAvroSchemaEvolutionUtils {
avroRecord.put("col21", 5.001f);
avroRecord.put("col3", 12.999d);
avroRecord.put("col31", 9999.999d);
- Schema currentDecimalType = avroSchema.getField("col4").schema().getTypes().get(1);
+ Schema currentDecimalType = hoodieSchema.getField("col4").get().getAvroField().schema().getTypes().get(1);
BigDecimal bd = new BigDecimal("123.456").setScale(((LogicalTypes.Decimal) currentDecimalType.getLogicalType()).getScale());
avroRecord.put("col4", HoodieAvroUtils.DECIMAL_CONVERSION.toFixed(bd, currentDecimalType, currentDecimalType.getLogicalType()));
- Schema currentDecimalType1 = avroSchema.getField("col41").schema().getTypes().get(1);
+ Schema currentDecimalType1 = hoodieSchema.getField("col41").get().getAvroField().schema().getTypes().get(1);
BigDecimal bd1 = new BigDecimal("7890.456").setScale(((LogicalTypes.Decimal) currentDecimalType1.getLogicalType()).getScale());
avroRecord.put("col41", HoodieAvroUtils.DECIMAL_CONVERSION.toFixed(bd1, currentDecimalType1, currentDecimalType1.getLogicalType()));
@@ -327,8 +333,8 @@ public class TestAvroSchemaEvolutionUtils {
ByteBuffer bb = ByteBuffer.wrap(new byte[] {97, 48, 53});
avroRecord.put("col9", bb);
avroRecord.put("enum", new GenericData.EnumSymbol(new Schema.Parser().parse(enumSchema), Enum.ENUM1));
- Assertions.assertEquals(GenericData.get().validate(avroSchema, avroRecord), true);
- InternalSchema internalSchema = AvroInternalSchemaConverter.convert(avroSchema);
+ Assertions.assertEquals(GenericData.get().validate(hoodieSchema.toAvroSchema(), avroRecord), true);
+ InternalSchema internalSchema = InternalSchemaConverter.convert(hoodieSchema);
// do change type operation
TableChanges.ColumnUpdateChange updateChange = TableChanges.ColumnUpdateChange.get(internalSchema);
updateChange
@@ -350,11 +356,11 @@ public class TestAvroSchemaEvolutionUtils {
.updateColumnType("col6", Types.StringType.get())
.updateColumnType("enum", Types.StringType.get());
InternalSchema newSchema = SchemaChangeUtils.applyTableChanges2Schema(internalSchema, updateChange);
- Schema newAvroSchema = AvroInternalSchemaConverter.convert(newSchema, avroSchema.getFullName());
- GenericRecord newRecord = HoodieAvroUtils.rewriteRecordWithNewSchema(avroRecord, newAvroSchema, Collections.emptyMap());
+ HoodieSchema newHoodieSchema = InternalSchemaConverter.convert(newSchema, hoodieSchema.getFullName());
+ GenericRecord newRecord = HoodieAvroUtils.rewriteRecordWithNewSchema(avroRecord, newHoodieSchema.toAvroSchema(), Collections.emptyMap());
Assertions.assertEquals("ENUM1", newRecord.get("enum"));
- Assertions.assertEquals(GenericData.get().validate(newAvroSchema, newRecord), true);
+ Assertions.assertEquals(GenericData.get().validate(newHoodieSchema.toAvroSchema(), newRecord), true);
}
@Test
@@ -368,20 +374,20 @@ public class TestAvroSchemaEvolutionUtils {
Types.Field.get(4, false, "locations", Types.MapType.get(8, 9, Types.StringType.get(),
Types.RecordType.get(Types.Field.get(10, false, "lat", Types.FloatType.get()), Types.Field.get(11, false, "long", Types.FloatType.get())), false))
);
- Schema schema = AvroInternalSchemaConverter.convert(record, "test1");
- GenericData.Record avroRecord = new GenericData.Record(schema);
- GenericData.get().validate(schema, avroRecord);
+ HoodieSchema schema = InternalSchemaConverter.convert(record, "test1");
+ GenericData.Record avroRecord = new GenericData.Record(schema.toAvroSchema());
+ GenericData.get().validate(schema.toAvroSchema(), avroRecord);
avroRecord.put("id", 2);
avroRecord.put("data", "xs");
// fill record type
- GenericData.Record preferencesRecord = new GenericData.Record(AvroInternalSchemaConverter.convert(record.fieldType("preferences"), "test1.preferences"));
+ GenericData.Record preferencesRecord = new GenericData.Record(InternalSchemaConverter.convert(record.fieldType("preferences"), "test1.preferences").toAvroSchema());
preferencesRecord.put("feature1", false);
preferencesRecord.put("feature2", true);
- Assertions.assertEquals(GenericData.get().validate(AvroInternalSchemaConverter.convert(record.fieldType("preferences"), "test1.preferences"), preferencesRecord), true);
+ Assertions.assertEquals(GenericData.get().validate(InternalSchemaConverter.convert(record.fieldType("preferences"), "test1.preferences").toAvroSchema(), preferencesRecord), true);
avroRecord.put("preferences", preferencesRecord);
// fill mapType
Map<String, GenericData.Record> locations = new HashMap<>();
- Schema mapSchema = AvroInternalSchemaConverter.convert(((Types.MapType)record.fieldByNameCaseInsensitive("locations").type()).valueType(), "test1.locations");
+ Schema mapSchema = InternalSchemaConverter.convert(((Types.MapType)record.fieldByNameCaseInsensitive("locations").type()).valueType(), "test1.locations").toAvroSchema();
GenericData.Record locationsValue = new GenericData.Record(mapSchema);
locationsValue.put("lat", 1.2f);
locationsValue.put("long", 1.4f);
@@ -398,7 +404,7 @@ public class TestAvroSchemaEvolutionUtils {
avroRecord.put("doubles", doubles);
// do check
- Assertions.assertTrue(GenericData.get().validate(schema, avroRecord));
+ Assertions.assertTrue(GenericData.get().validate(schema.toAvroSchema(), avroRecord));
// create newSchema
Types.RecordType newRecord = Types.RecordType.get(
Types.Field.get(0, false, "id", Types.IntType.get()),
@@ -416,13 +422,13 @@ public class TestAvroSchemaEvolutionUtils {
)
);
- Schema newAvroSchema = AvroInternalSchemaConverter.convert(newRecord, schema.getName());
+ Schema newAvroSchema = InternalSchemaConverter.convert(newRecord, schema.getName()).toAvroSchema();
GenericRecord newAvroRecord = HoodieAvroUtils.rewriteRecordWithNewSchema(avroRecord, newAvroSchema, Collections.emptyMap());
// test the correctly of rewrite
Assertions.assertEquals(GenericData.get().validate(newAvroSchema, newAvroRecord), true);
// test rewrite with rename
- InternalSchema internalSchema = AvroInternalSchemaConverter.convert(schema);
+ InternalSchema internalSchema = InternalSchemaConverter.convert(schema);
// do change rename operation
TableChanges.ColumnUpdateChange updateChange = TableChanges.ColumnUpdateChange.get(internalSchema);
updateChange
@@ -432,11 +438,11 @@ public class TestAvroSchemaEvolutionUtils {
.renameColumn("preferences.feature2", "f2")
.renameColumn("locations.value.lat", "lt");
InternalSchema internalSchemaRename = SchemaChangeUtils.applyTableChanges2Schema(internalSchema, updateChange);
- Schema avroSchemaRename = AvroInternalSchemaConverter.convert(internalSchemaRename, schema.getFullName());
+ HoodieSchema hoodieSchemaRename = InternalSchemaConverter.convert(internalSchemaRename, schema.getFullName());
Map<String, String> renameCols = InternalSchemaUtils.collectRenameCols(internalSchema, internalSchemaRename);
- GenericRecord avroRecordRename = HoodieAvroUtils.rewriteRecordWithNewSchema(avroRecord, avroSchemaRename, renameCols);
+ GenericRecord avroRecordRename = HoodieAvroUtils.rewriteRecordWithNewSchema(avroRecord, hoodieSchemaRename.toAvroSchema(), renameCols);
// test the correctly of rewrite
- Assertions.assertEquals(GenericData.get().validate(avroSchemaRename, avroRecordRename), true);
+ assertTrue(GenericData.get().validate(hoodieSchemaRename.toAvroSchema(), avroRecordRename));
}
@Test
@@ -479,8 +485,8 @@ public class TestAvroSchemaEvolutionUtils {
Types.Field.get(5, true, "nest2", Types.BooleanType.get())))
);
evolvedRecord = (Types.RecordType)InternalSchemaBuilder.getBuilder().refreshNewId(evolvedRecord, new AtomicInteger(0));
- Schema evolvedAvroSchema = AvroInternalSchemaConverter.convert(evolvedRecord, "test1");
- InternalSchema result = AvroSchemaEvolutionUtils.reconcileSchema(evolvedAvroSchema, oldSchema, false);
+ HoodieSchema evolvedSchema = InternalSchemaConverter.convert(evolvedRecord, "test1");
+ InternalSchema result = AvroSchemaEvolutionUtils.reconcileSchema(evolvedSchema.getAvroSchema(), oldSchema, false);
Types.RecordType checkedRecord = Types.RecordType.get(
Types.Field.get(0, false, "id", Types.IntType.get()),
Types.Field.get(1, true, "data", Types.StringType.get()),
@@ -509,23 +515,23 @@ public class TestAvroSchemaEvolutionUtils {
public void testReconcileSchema() {
// simple schema test
// a: boolean, b: int, c: long, d: date
- Schema schema = create("simple",
- new Schema.Field("a", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.BOOLEAN)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("b", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.INT)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("c", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.LONG)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("d", AvroInternalSchemaConverter.nullableSchema(LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT))), null, JsonProperties.NULL_VALUE));
+ HoodieSchema schema = create("simple",
+ HoodieSchemaField.of("a", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.BOOLEAN)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("b", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.INT)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("c", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.LONG)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("d", HoodieSchema.createNullable(HoodieSchema.fromAvroSchema(LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT)))), null, HoodieJsonProperties.NULL_VALUE));
// a: boolean, c: long, c_1: long, d: date
- Schema incomingSchema = create("simpleIncoming",
- new Schema.Field("a", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.BOOLEAN)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("a1", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.LONG)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("c", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.LONG)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("c1", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.LONG)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("c2", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.LONG)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("d", AvroInternalSchemaConverter.nullableSchema(LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT))), null, JsonProperties.NULL_VALUE),
- new Schema.Field("d1", AvroInternalSchemaConverter.nullableSchema(LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT))), null, JsonProperties.NULL_VALUE),
- new Schema.Field("d2", AvroInternalSchemaConverter.nullableSchema(LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT))), null, JsonProperties.NULL_VALUE));
-
- Schema simpleCheckSchema = new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"simple\",\"fields\":[{\"name\":\"a\",\"type\":[\"null\",\"boolean\"],\"default\":null},"
+ HoodieSchema incomingSchema = create("simpleIncoming",
+ HoodieSchemaField.of("a", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.BOOLEAN)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("a1", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.LONG)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("c", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.LONG)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("c1", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.LONG)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("c2", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.LONG)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("d", HoodieSchema.createNullable(HoodieSchema.fromAvroSchema(LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT)))), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("d1", HoodieSchema.createNullable(HoodieSchema.fromAvroSchema(LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT)))), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("d2", HoodieSchema.createNullable(HoodieSchema.fromAvroSchema(LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT)))), null, HoodieJsonProperties.NULL_VALUE));
+
+ HoodieSchema simpleCheckSchema = HoodieSchema.parse("{\"type\":\"record\",\"name\":\"simple\",\"fields\":[{\"name\":\"a\",\"type\":[\"null\",\"boolean\"],\"default\":null},"
+ "{\"name\":\"b\",\"type\":[\"null\",\"int\"],\"default\":null},"
+ "{\"name\":\"c\",\"type\":[\"null\",\"long\"],\"default\":null},"
+ "{\"name\":\"d\",\"type\":[\"null\",{\"type\":\"int\",\"logicalType\":\"date\"}],\"default\":null},"
@@ -534,30 +540,30 @@ public class TestAvroSchemaEvolutionUtils {
+ "{\"name\":\"d1\",\"type\":[\"null\",{\"type\":\"int\",\"logicalType\":\"date\"}],\"default\":null},"
+ "{\"name\":\"d2\",\"type\":[\"null\",{\"type\":\"int\",\"logicalType\":\"date\"}],\"default\":null}]}");
- Schema simpleReconcileSchema = AvroInternalSchemaConverter.convert(AvroSchemaEvolutionUtils
- .reconcileSchema(incomingSchema, AvroInternalSchemaConverter.convert(schema), false), "schemaNameFallback");
+ HoodieSchema simpleReconcileSchema = InternalSchemaConverter.convert(AvroSchemaEvolutionUtils
+ .reconcileSchema(incomingSchema.getAvroSchema(), InternalSchemaConverter.convert(schema), false), "schemaNameFallback");
Assertions.assertEquals(simpleCheckSchema, simpleReconcileSchema);
}
@Test
public void testNotEvolveSchemaIfReconciledSchemaUnchanged() {
// a: boolean, c: long, c_1: long, d: date
- Schema oldSchema = create("simple",
- new Schema.Field("a", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.BOOLEAN)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("b", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.INT)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("c", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.LONG)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("d", AvroInternalSchemaConverter.nullableSchema(LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT))), null, JsonProperties.NULL_VALUE));
+ HoodieSchema oldSchema = create("simple",
+ HoodieSchemaField.of("a", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.BOOLEAN)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("b", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.INT)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("c", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.LONG)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("d", HoodieSchema.createNullable(HoodieSchema.fromAvroSchema(LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT)))), null, HoodieJsonProperties.NULL_VALUE));
// incoming schema is part of old schema
// a: boolean, b: int, c: long
- Schema incomingSchema = create("simple",
- new Schema.Field("a", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.BOOLEAN)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("b", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.INT)), null, JsonProperties.NULL_VALUE),
- new Schema.Field("c", AvroInternalSchemaConverter.nullableSchema(Schema.create(Schema.Type.LONG)), null, JsonProperties.NULL_VALUE));
+ HoodieSchema incomingSchema = create("simple",
+ HoodieSchemaField.of("a", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.BOOLEAN)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("b", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.INT)), null, HoodieJsonProperties.NULL_VALUE),
+ HoodieSchemaField.of("c", HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.LONG)), null, HoodieJsonProperties.NULL_VALUE));
- InternalSchema oldInternalSchema = AvroInternalSchemaConverter.convert(oldSchema);
+ InternalSchema oldInternalSchema = InternalSchemaConverter.convert(oldSchema);
// set a non-default schema id for old table schema, e.g., 2.
oldInternalSchema.setSchemaId(2);
- InternalSchema evolvedSchema = AvroSchemaEvolutionUtils.reconcileSchema(incomingSchema, oldInternalSchema, false);
+ InternalSchema evolvedSchema = AvroSchemaEvolutionUtils.reconcileSchema(incomingSchema.getAvroSchema(), oldInternalSchema, false);
// the evolved schema should be the old table schema, since there is no type change at all.
Assertions.assertEquals(oldInternalSchema, evolvedSchema);
}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/HoodieCatalogUtil.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/HoodieCatalogUtil.java
index fe08e56dc8e4..66e9d3dd24a8 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/HoodieCatalogUtil.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/HoodieCatalogUtil.java
@@ -22,6 +22,7 @@ import org.apache.hudi.adapter.Utils;
import org.apache.hudi.client.HoodieFlinkWriteClient;
import org.apache.hudi.common.model.PartitionBucketIndexHashingConfig;
import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.configuration.FlinkOptions;
@@ -29,7 +30,7 @@ import org.apache.hudi.configuration.HadoopConfigurations;
import org.apache.hudi.exception.HoodieCatalogException;
import org.apache.hudi.internal.schema.InternalSchema;
import org.apache.hudi.internal.schema.Type;
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter;
import org.apache.hudi.util.AvroSchemaConverter;
import org.apache.hudi.util.FlinkWriteClients;
import org.apache.hudi.util.StreamerUtil;
@@ -232,7 +233,7 @@ public class HoodieCatalogUtil {
HoodieFlinkWriteClient<?> writeClient = createWriteClient(tablePath, oldTable, hadoopConf, inferTablePathFunc);
Pair<InternalSchema, HoodieTableMetaClient> pair = writeClient.getInternalSchemaAndMetaClient();
InternalSchema oldSchema = pair.getLeft();
- Function<LogicalType, Type> convertFunc = (LogicalType logicalType) -> AvroInternalSchemaConverter.convertToField(AvroSchemaConverter.convertToSchema(logicalType));
+ Function<LogicalType, Type> convertFunc = (LogicalType logicalType) -> InternalSchemaConverter.convertToField(HoodieSchema.fromAvroSchema(AvroSchemaConverter.convertToSchema(logicalType)));
InternalSchema newSchema = Utils.applyTableChange(oldSchema, tableChanges, convertFunc);
if (!oldSchema.equals(newSchema)) {
writeClient.setOperationType(WriteOperationType.ALTER_SCHEMA);
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/InternalSchemaManager.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/InternalSchemaManager.java
index 6ee6a8d81d53..7429e4786bcb 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/InternalSchemaManager.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/InternalSchemaManager.java
@@ -33,7 +33,7 @@ import org.apache.hudi.internal.schema.InternalSchema;
import org.apache.hudi.internal.schema.Type;
import org.apache.hudi.internal.schema.Types;
import org.apache.hudi.internal.schema.action.InternalSchemaMerger;
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter;
import org.apache.hudi.internal.schema.utils.InternalSchemaUtils;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
@@ -160,7 +160,7 @@ public class InternalSchemaManager implements Serializable {
List<Integer> selectedFieldList = IntStream.of(selectedFields).boxed().collect(Collectors.toList());
// mergeSchema is built with useColumnTypeFromFileSchema = true
List<DataType> mergeSchemaAsDataTypes = AvroSchemaConverter.convertToDataType(
- AvroInternalSchemaConverter.convert(mergeSchema, "tableName")).getChildren();
+ InternalSchemaConverter.convert(mergeSchema, "tableName").toAvroSchema()).getChildren();
DataType[] fileFieldTypes = new DataType[queryFieldTypes.length];
for (int i = 0; i < queryFieldTypes.length; i++) {
// position of ChangedType in querySchema
diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/SchemaEvolutionContext.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/SchemaEvolutionContext.java
index 75436c2b5006..4796b8ed29f2 100644
--- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/SchemaEvolutionContext.java
+++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/SchemaEvolutionContext.java
@@ -36,7 +36,7 @@ import org.apache.hudi.internal.schema.InternalSchema;
import org.apache.hudi.internal.schema.Type;
import org.apache.hudi.internal.schema.Types;
import org.apache.hudi.internal.schema.action.InternalSchemaMerger;
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter;
import org.apache.hudi.internal.schema.utils.InternalSchemaUtils;
import org.apache.hudi.internal.schema.utils.SerDeHelper;
import org.apache.hudi.storage.HoodieStorage;
@@ -206,12 +206,12 @@ public class SchemaEvolutionContext {
String partitionFields = job.get(hive_metastoreConstants.META_TABLE_PARTITION_COLUMNS, "");
List<String> partitioningFields = !partitionFields.isEmpty() ? Arrays.stream(partitionFields.split("/")).collect(Collectors.toList())
: new ArrayList<>();
- Schema writerSchema = AvroInternalSchemaConverter.convert(internalSchemaOption.get(), tableAvroSchema.getName());
+ Schema writerSchema = InternalSchemaConverter.convert(internalSchemaOption.get(), tableAvroSchema.getName()).toAvroSchema();
writerSchema = HoodieRealtimeRecordReaderUtils.addPartitionFields(writerSchema, partitioningFields);
Map<String, Schema.Field> schemaFieldsMap = HoodieRealtimeRecordReaderUtils.getNameToFieldMap(writerSchema);
// we should get HoodieParquetInputFormat#HIVE_TMP_COLUMNS,since serdeConstants#LIST_COLUMNS maybe change by HoodieParquetInputFormat#setColumnNameList
Schema hiveSchema = realtimeRecordReader.constructHiveOrderedSchema(writerSchema, schemaFieldsMap, job.get(HIVE_TMP_COLUMNS));
- Schema readerSchema = AvroInternalSchemaConverter.convert(prunedInternalSchema, tableAvroSchema.getName());
+ Schema readerSchema = InternalSchemaConverter.convert(prunedInternalSchema, tableAvroSchema.getName()).toAvroSchema();
// setUp evolution schema
realtimeRecordReader.setWriterSchema(writerSchema);
realtimeRecordReader.setReaderSchema(readerSchema);
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
index aa52b108e005..4052098a07ff 100644
--- 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
@@ -19,9 +19,13 @@
package org.apache.hudi.hadoop.avro;
import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.hadoop.HoodieColumnProjectionUtils;
import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils;
+import org.apache.hudi.internal.schema.InternalSchema;
+import org.apache.hudi.internal.schema.action.InternalSchemaMerger;
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
@@ -31,10 +35,6 @@ 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.internal.schema.InternalSchema;
-import org.apache.hudi.internal.schema.action.InternalSchemaMerger;
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
-
import org.apache.parquet.avro.AvroReadSupport;
import org.apache.parquet.format.converter.ParquetMetadataConverter;
import org.apache.parquet.hadoop.ParquetFileReader;
@@ -68,11 +68,11 @@ public class HoodieAvroParquetReader extends RecordReader<Void, ArrayWritable> {
if (internalSchemaOption.isPresent()) {
// do schema reconciliation in case there exists read column which is not in the file schema.
InternalSchema mergedInternalSchema = new InternalSchemaMerger(
- AvroInternalSchemaConverter.convert(baseSchema),
+ InternalSchemaConverter.convert(HoodieSchema.fromAvroSchema(baseSchema)),
internalSchemaOption.get(),
true,
true).mergeSchema();
- baseSchema = AvroInternalSchemaConverter.convert(mergedInternalSchema, baseSchema.getFullName());
+ baseSchema = InternalSchemaConverter.convert(mergedInternalSchema, baseSchema.getFullName()).getAvroSchema();
}
// if exists read columns, we need to filter columns.
diff --git a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/hive/TestHoodieCombineHiveInputFormat.java b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/hive/TestHoodieCombineHiveInputFormat.java
index 8390aef86dac..a5f96aa28d9f 100644
--- a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/hive/TestHoodieCombineHiveInputFormat.java
+++ b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/hive/TestHoodieCombineHiveInputFormat.java
@@ -22,6 +22,7 @@ import org.apache.hudi.avro.HoodieAvroUtils;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.table.log.HoodieLogFormat;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.testutils.FileCreateUtilsLegacy;
@@ -35,7 +36,7 @@ import org.apache.hudi.hadoop.SchemaEvolutionContext;
import org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat;
import org.apache.hudi.hadoop.testutils.InputFormatTestUtil;
import org.apache.hudi.internal.schema.InternalSchema;
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter;
import org.apache.hudi.internal.schema.utils.SerDeHelper;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StorageConfiguration;
@@ -131,7 +132,7 @@ public class TestHoodieCombineHiveInputFormat extends HoodieCommonTestHarness {
HoodieCommitMetadata commitMetadataOne = CommitUtils.buildMetadata(Collections.emptyList(), Collections.emptyMap(), Option.empty(), WriteOperationType.UPSERT,
schema.toString(), HoodieTimeline.COMMIT_ACTION);
// mock the latest schema to the commit metadata
- InternalSchema internalSchema = AvroInternalSchemaConverter.convert(schema);
+ InternalSchema internalSchema = InternalSchemaConverter.convert(HoodieSchema.fromAvroSchema(schema));
commitMetadataOne.addMetadata(SerDeHelper.LATEST_SCHEMA, SerDeHelper.toJson(internalSchema));
FileCreateUtilsLegacy.createCommit(COMMIT_METADATA_SER_DE, path1.toString(), commitTime, Option.of(commitMetadataOne));
// Create 3 parquet files with 10 records each for partition 2
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
index 847f2464ae60..e1aee39f8231 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
@@ -43,7 +43,7 @@ import org.apache.hudi.exception.HoodieException
import org.apache.hudi.hadoop.fs.HadoopFSUtils
import org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToStoragePath
import org.apache.hudi.internal.schema.InternalSchema
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter
import org.apache.hudi.internal.schema.utils.{InternalSchemaUtils, SerDeHelper}
import org.apache.hudi.io.storage.HoodieSparkIOFactory
import org.apache.hudi.metadata.HoodieTableMetadata
@@ -171,8 +171,8 @@ abstract class HoodieBaseRelation(val sqlContext: SQLContext,
}
val (name, namespace) = AvroConversionUtils.getAvroRecordNameAndNamespace(tableName)
- val avroSchema = internalSchemaOpt.map { is =>
- AvroInternalSchemaConverter.convert(is, namespace + "." + name)
+ val avroSchema: Schema = internalSchemaOpt.map { is =>
+ InternalSchemaConverter.convert(is, namespace + "." + name).toAvroSchema
} orElse {
specifiedQueryTimestamp.map(schemaResolver.getTableAvroSchema)
} orElse {
@@ -801,7 +801,7 @@ object HoodieBaseRelation extends SparkAdapterSupport {
case Right(internalSchema) =>
checkState(!internalSchema.isEmptySchema)
val prunedInternalSchema = InternalSchemaUtils.pruneInternalSchema(internalSchema, requiredColumns.toList.asJava)
- val requiredAvroSchema = AvroInternalSchemaConverter.convert(prunedInternalSchema, "schema")
+ val requiredAvroSchema = InternalSchemaConverter.convert(prunedInternalSchema, "schema").toAvroSchema
val requiredStructSchema = AvroConversionUtils.convertAvroSchemaToStructType(requiredAvroSchema)
(requiredAvroSchema, requiredStructSchema, prunedInternalSchema)
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieHadoopFsRelationFactory.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieHadoopFsRelationFactory.scala
index 242f29f2c89c..bad337c18527 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieHadoopFsRelationFactory.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieHadoopFsRelationFactory.scala
@@ -27,7 +27,7 @@ import org.apache.hudi.common.table.log.InstantRange.RangeType
import org.apache.hudi.common.table.timeline.HoodieTimeline
import org.apache.hudi.common.util.StringUtils.isNullOrEmpty
import org.apache.hudi.internal.schema.InternalSchema
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter
import org.apache.hudi.keygen.{CustomAvroKeyGenerator, CustomKeyGenerator, TimestampBasedAvroKeyGenerator, TimestampBasedKeyGenerator}
import org.apache.hudi.storage.StoragePath
@@ -141,8 +141,8 @@ abstract class HoodieBaseHadoopFsRelationFactory(val sqlContext: SQLContext,
}
val (name, namespace) = AvroConversionUtils.getAvroRecordNameAndNamespace(tableName)
- val avroSchema = internalSchemaOpt.map { is =>
- AvroInternalSchemaConverter.convert(is, namespace + "." + name)
+ val avroSchema: Schema = internalSchemaOpt.map { is =>
+ InternalSchemaConverter.convert(is, namespace + "." + name).toAvroSchema
} orElse {
specifiedQueryTimestamp.map(schemaResolver.getTableAvroSchema)
} orElse {
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala
index 24183e05e1b7..c16a66631916 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala
@@ -25,12 +25,13 @@ import org.apache.hudi.avro.HoodieAvroUtils
import org.apache.hudi.avro.HoodieAvroUtils.removeMetadataFields
import org.apache.hudi.common.config.{HoodieCommonConfig, HoodieConfig, TypedProperties}
import org.apache.hudi.common.model.HoodieRecord
+import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
import org.apache.hudi.common.util.{ConfigUtils, StringUtils}
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.exception.{HoodieException, SchemaCompatibilityException}
import org.apache.hudi.internal.schema.InternalSchema
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter
import org.apache.hudi.internal.schema.utils.AvroSchemaEvolutionUtils
import org.apache.hudi.internal.schema.utils.AvroSchemaEvolutionUtils.reconcileSchemaRequirements
@@ -113,13 +114,14 @@ object HoodieSchemaUtils {
opts: Map[String, String]): Schema = {
latestTableSchemaOpt match {
// If table schema is empty, then we use the source schema as a writer's schema.
- case None => AvroInternalSchemaConverter.fixNullOrdering(sourceSchema)
+ case None => InternalSchemaConverter.fixNullOrdering(HoodieSchema.fromAvroSchema(sourceSchema)).toAvroSchema
// Otherwise, we need to make sure we reconcile incoming and latest table schemas
case Some(latestTableSchemaWithMetaFields) =>
// NOTE: Meta-fields will be unconditionally injected by Hudi writing handles, for the sake of deducing proper writer schema
// we're stripping them to make sure we can perform proper analysis
// add call to fix null ordering to ensure backwards compatibility
- val latestTableSchema = AvroInternalSchemaConverter.fixNullOrdering(removeMetadataFields(latestTableSchemaWithMetaFields))
+ val latestTableSchema = InternalSchemaConverter.fixNullOrdering(HoodieSchema.fromAvroSchema(
+ removeMetadataFields(latestTableSchemaWithMetaFields))).toAvroSchema
// Before validating whether schemas are compatible, we need to "canonicalize" source's schema
// relative to the table's one, by doing a (minor) reconciliation of the nullability constraints:
@@ -132,7 +134,7 @@ object HoodieSchemaUtils {
val canonicalizedSourceSchema = if (shouldCanonicalizeSchema) {
canonicalizeSchema(sourceSchema, latestTableSchema, opts, !shouldReconcileSchema)
} else {
- AvroInternalSchemaConverter.fixNullOrdering(sourceSchema)
+ InternalSchemaConverter.fixNullOrdering(HoodieSchema.fromAvroSchema(sourceSchema)).toAvroSchema
}
if (shouldReconcileSchema) {
@@ -200,7 +202,7 @@ object HoodieSchemaUtils {
val setNullForMissingColumns = opts.getOrElse(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS.key(),
HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS.defaultValue()).toBoolean
val mergedInternalSchema = AvroSchemaEvolutionUtils.reconcileSchema(canonicalizedSourceSchema, internalSchema, setNullForMissingColumns)
- val evolvedSchema = AvroInternalSchemaConverter.convert(mergedInternalSchema, latestTableSchema.getFullName)
+ val evolvedSchema = InternalSchemaConverter.convert(mergedInternalSchema, latestTableSchema.getFullName).toAvroSchema
val shouldRemoveMetaDataFromInternalSchema = sourceSchema.getFields().asScala.filter(f => f.name().equalsIgnoreCase(HoodieRecord.RECORD_KEY_METADATA_FIELD)).isEmpty
if (shouldRemoveMetaDataFromInternalSchema) HoodieAvroUtils.removeMetadataFields(evolvedSchema) else evolvedSchema
case None =>
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
index a726789d93ad..2d69d609b836 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
@@ -37,6 +37,7 @@ import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.model._
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
import org.apache.hudi.common.model.HoodieTableType.{COPY_ON_WRITE, MERGE_ON_READ}
+import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient, HoodieTableVersion, TableSchemaResolver}
import org.apache.hudi.common.table.log.block.HoodieLogBlock.HoodieLogBlockType
import org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator
@@ -52,7 +53,7 @@ import org.apache.hudi.hive.ddl.HiveSyncMode
import org.apache.hudi.index.HoodieIndex
import org.apache.hudi.index.bucket.partition.PartitionBucketIndexUtils
import org.apache.hudi.internal.schema.InternalSchema
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter
import org.apache.hudi.internal.schema.utils.SerDeHelper
import org.apache.hudi.keygen.{BaseKeyGenerator, TimestampBasedAvroKeyGenerator, TimestampBasedKeyGenerator}
import org.apache.hudi.keygen.constant.KeyGeneratorType
@@ -373,7 +374,7 @@ class HoodieSparkSqlWriterInternal {
// we will force-apply schema evolution to the writer's schema
if (shouldReconcileSchema && hoodieConfig.getBooleanOrDefault(DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED)) {
val allowOperationMetaDataField = parameters.getOrElse(HoodieWriteConfig.ALLOW_OPERATION_METADATA_FIELD.key(), "false").toBoolean
- Some(AvroInternalSchemaConverter.convert(HoodieAvroUtils.addMetadataFields(latestTableSchemaOpt.getOrElse(sourceSchema), allowOperationMetaDataField)))
+ Some(InternalSchemaConverter.convert(HoodieSchema.fromAvroSchema(HoodieAvroUtils.addMetadataFields(latestTableSchemaOpt.getOrElse(sourceSchema), allowOperationMetaDataField))))
} else {
None
}
@@ -657,7 +658,7 @@ class HoodieSparkSqlWriterInternal {
val correctInternalSchema = internalSchemaOpt.map { internalSchema =>
if (internalSchema.findField(HoodieRecord.RECORD_KEY_METADATA_FIELD) == null && writeSchemaOpt.isDefined) {
val allowOperationMetaDataField = parameters.getOrElse(HoodieWriteConfig.ALLOW_OPERATION_METADATA_FIELD.key(), "false").toBoolean
- AvroInternalSchemaConverter.convert(HoodieAvroUtils.addMetadataFields(writeSchemaOpt.get, allowOperationMetaDataField))
+ InternalSchemaConverter.convert(HoodieSchema.fromAvroSchema(HoodieAvroUtils.addMetadataFields(writeSchemaOpt.get, allowOperationMetaDataField)))
} else {
internalSchema
}
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/AlterTableCommand.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/AlterTableCommand.scala
index 6dfb5253a90d..b31eb10889c2 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/AlterTableCommand.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/AlterTableCommand.scala
@@ -21,6 +21,7 @@ import org.apache.hudi.{DataSourceUtils, HoodieWriterUtils}
import org.apache.hudi.avro.AvroSchemaUtils.getAvroRecordQualifiedName
import org.apache.hudi.client.utils.SparkInternalSchemaConverter
import org.apache.hudi.common.model.{HoodieCommitMetadata, HoodieFailedWritesCleaningPolicy, WriteOperationType}
+import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.common.table.{HoodieTableMetaClient, TableSchemaResolver}
import org.apache.hudi.common.table.timeline.HoodieInstant.State
import org.apache.hudi.common.util.{CommitUtils, Option}
@@ -29,7 +30,7 @@ import org.apache.hudi.hadoop.fs.HadoopFSUtils
import org.apache.hudi.internal.schema.InternalSchema
import org.apache.hudi.internal.schema.action.TableChange.ColumnChangeID
import org.apache.hudi.internal.schema.action.TableChanges
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter
import org.apache.hudi.internal.schema.io.FileBasedInternalSchemaStorageManager
import org.apache.hudi.internal.schema.utils.{SchemaChangeUtils, SerDeHelper}
import org.apache.hudi.table.HoodieSparkTable
@@ -219,7 +220,7 @@ case class AlterTableCommand(table: CatalogTable, changes: Seq[TableChange], cha
val schemaUtil = new TableSchemaResolver(metaClient)
val schema = schemaUtil.getTableInternalSchemaFromCommitMetadata().orElse {
- AvroInternalSchemaConverter.convert(schemaUtil.getTableAvroSchema)
+ InternalSchemaConverter.convert(HoodieSchema.fromAvroSchema(schemaUtil.getTableAvroSchema))
}
val historySchemaStr = schemaUtil.getTableHistorySchemaStrFromCommitMetadata.orElse("")
@@ -250,7 +251,7 @@ object AlterTableCommand extends Logging {
* @param sparkSession The spark session.
*/
def commitWithSchema(internalSchema: InternalSchema, historySchemaStr: String, table: CatalogTable, sparkSession: SparkSession): Unit = {
- val schema = AvroInternalSchemaConverter.convert(internalSchema, getAvroRecordQualifiedName(table.identifier.table))
+ val schema = InternalSchemaConverter.convert(internalSchema, getAvroRecordQualifiedName(table.identifier.table))
val path = getTableLocation(table, sparkSession)
val jsc = new JavaSparkContext(sparkSession.sparkContext)
val client = DataSourceUtils.createHoodieClient(
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/util/TestSparkInternalSchemaConverter.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/util/TestSparkInternalSchemaConverter.scala
index 6da4b317960c..557f20468652 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/util/TestSparkInternalSchemaConverter.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/util/TestSparkInternalSchemaConverter.scala
@@ -21,18 +21,18 @@ package org.apache.hudi.util
import org.apache.hudi.SparkAdapterSupport
import org.apache.hudi.client.utils.SparkInternalSchemaConverter.collectColNamesFromSparkStruct
-import org.apache.hudi.internal.schema.convert.TestAvroInternalSchemaConverter._
+import org.apache.hudi.common.schema.HoodieSchema
+import org.apache.hudi.internal.schema.convert.TestInternalSchemaConverter._
import org.apache.hudi.testutils.HoodieSparkClientTestHarness
-import org.apache.avro.Schema
import org.apache.spark.sql.types._
import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
import org.junit.jupiter.api.Test
class TestSparkInternalSchemaConverter extends HoodieSparkClientTestHarness with SparkAdapterSupport {
- private def getStructType(schema: Schema): DataType = {
- sparkAdapter.getAvroSchemaConverters.toSqlType(schema)._1
+ private def getStructType(schema: HoodieSchema): DataType = {
+ sparkAdapter.getAvroSchemaConverters.toSqlType(schema.toAvroSchema)._1
}
@Test
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieInternalRowUtils.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieInternalRowUtils.scala
index b60b3612c8fd..681e2fb6c4e2 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieInternalRowUtils.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestHoodieInternalRowUtils.scala
@@ -21,9 +21,10 @@ package org.apache.spark.sql.hudi.common
import org.apache.hudi.{AvroConversionUtils, SparkRowSerDe}
import org.apache.hudi.SparkAdapterSupport.sparkAdapter
import org.apache.hudi.avro.HoodieAvroUtils
+import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.internal.schema.Types
import org.apache.hudi.internal.schema.action.TableChanges
-import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter
+import org.apache.hudi.internal.schema.convert.InternalSchemaConverter
import org.apache.hudi.internal.schema.utils.SchemaChangeUtils
import org.apache.hudi.testutils.HoodieClientTestUtils
@@ -286,7 +287,7 @@ class TestHoodieInternalRowUtils extends FunSuite with Matchers with BeforeAndAf
val bb = ByteBuffer.wrap(Array[Byte](97, 48, 53))
avroRecord.put("col9", bb)
assert(GenericData.get.validate(avroSchema, avroRecord))
- val internalSchema = AvroInternalSchemaConverter.convert(avroSchema)
+ val internalSchema = InternalSchemaConverter.convert(HoodieSchema.fromAvroSchema(avroSchema))
// do change type operation
val updateChange = TableChanges.ColumnUpdateChange.get(internalSchema)
updateChange.updateColumnType("id", Types.LongType.get)
@@ -306,14 +307,14 @@ class TestHoodieInternalRowUtils extends FunSuite with Matchers with BeforeAndAf
.updateColumnType("col51", Types.DecimalType.get(18, 9))
.updateColumnType("col6", Types.StringType.get)
val newSchema = SchemaChangeUtils.applyTableChanges2Schema(internalSchema, updateChange)
- val newAvroSchema = AvroInternalSchemaConverter.convert(newSchema, avroSchema.getName)
- val newRecord = HoodieAvroUtils.rewriteRecordWithNewSchema(avroRecord, newAvroSchema, new HashMap[String, String])
- assert(GenericData.get.validate(newAvroSchema, newRecord))
+ val newHoodieSchema = InternalSchemaConverter.convert(newSchema, avroSchema.getName)
+ val newRecord = HoodieAvroUtils.rewriteRecordWithNewSchema(avroRecord, newHoodieSchema.toAvroSchema, new HashMap[String, String])
+ assert(GenericData.get.validate(newHoodieSchema.toAvroSchema, newRecord))
// Convert avro to internalRow
val structTypeSchema = HoodieInternalRowUtils.getCachedSchema(avroSchema)
- val newStructTypeSchema = HoodieInternalRowUtils.getCachedSchema(newAvroSchema)
+ val newStructTypeSchema = HoodieInternalRowUtils.getCachedSchema(newHoodieSchema.toAvroSchema)
val row = AvroConversionUtils.createAvroToInternalRowConverter(avroSchema, structTypeSchema).apply(avroRecord).get
- val newRowExpected = AvroConversionUtils.createAvroToInternalRowConverter(newAvroSchema, newStructTypeSchema)
+ val newRowExpected = AvroConversionUtils.createAvroToInternalRowConverter(newHoodieSchema.toAvroSchema, newStructTypeSchema)
.apply(newRecord).get
val rowWriter = HoodieInternalRowUtils.genUnsafeRowWriter(structTypeSchema, newStructTypeSchema, JCollections.emptyMap(), JCollections.emptyMap())
@@ -332,20 +333,20 @@ class TestHoodieInternalRowUtils extends FunSuite with Matchers with BeforeAndAf
Types.Field.get(4, false, "locations", Types.MapType.get(8, 9, Types.StringType.get(),
Types.RecordType.get(Types.Field.get(10, false, "lat", Types.FloatType.get()), Types.Field.get(11, false, "long", Types.FloatType.get())), false))
)
- val schema = AvroInternalSchemaConverter.convert(record, "test1")
+ val schema = InternalSchemaConverter.convert(record, "test1").toAvroSchema
val avroRecord = new GenericData.Record(schema)
GenericData.get.validate(schema, avroRecord)
avroRecord.put("id", 2)
avroRecord.put("data", "xs")
// fill record type
- val preferencesRecord = new GenericData.Record(AvroInternalSchemaConverter.convert(record.fieldType("preferences"), "test1_preferences"))
+ val preferencesRecord = new GenericData.Record(InternalSchemaConverter.convert(record.fieldType("preferences"), "test1_preferences").toAvroSchema)
preferencesRecord.put("feature1", false)
preferencesRecord.put("feature2", true)
- assert(GenericData.get.validate(AvroInternalSchemaConverter.convert(record.fieldType("preferences"), "test1_preferences"), preferencesRecord))
+ assert(GenericData.get.validate(InternalSchemaConverter.convert(record.fieldType("preferences"), "test1_preferences").toAvroSchema, preferencesRecord))
avroRecord.put("preferences", preferencesRecord)
// fill mapType
val locations = new HashMap[String, GenericData.Record]
- val mapSchema = AvroInternalSchemaConverter.convert(record.fieldByNameCaseInsensitive("locations").`type`.asInstanceOf[Types.MapType].valueType, "test1_locations")
+ val mapSchema = InternalSchemaConverter.convert(record.fieldByNameCaseInsensitive("locations").`type`.asInstanceOf[Types.MapType].valueType, "test1_locations").toAvroSchema
val locationsValue: GenericData.Record = new GenericData.Record(mapSchema)
locationsValue.put("lat", 1.2f)
locationsValue.put("long", 1.4f)
@@ -361,7 +362,7 @@ class TestHoodieInternalRowUtils extends FunSuite with Matchers with BeforeAndAf
avroRecord.put("doubles", doubles)
// create newSchema
val newRecord = Types.RecordType.get(Types.Field.get(0, false, "id", Types.IntType.get), Types.Field.get(1, true, "data", Types.StringType.get), Types.Field.get(2, true, "preferences", Types.RecordType.get(Types.Field.get(5, false, "feature1", Types.BooleanType.get), Types.Field.get(5, true, "featurex", Types.BooleanType.get), Types.Field.get(6, true, "feature2", Types.BooleanType.get))), Types.Field.get(3, false, "doubles", Types.ArrayType.get(7, false, Types.DoubleType.get)), Types [...]
- val newAvroSchema = AvroInternalSchemaConverter.convert(newRecord, schema.getName)
+ val newAvroSchema = InternalSchemaConverter.convert(newRecord, schema.getName).toAvroSchema
val newAvroRecord = HoodieAvroUtils.rewriteRecordWithNewSchema(avroRecord, newAvroSchema, new HashMap[String, String])
// test the correctly of rewrite
assert(GenericData.get.validate(newAvroSchema, newAvroRecord))