public static Type fromHudiType(Schema avroSchema) {
        Schema.Type columnType = avroSchema.getType();
        LogicalType logicalType = avroSchema.getLogicalType();
        PrimitiveType primitiveType = null;
        boolean isConvertedFailed = false;

        switch (columnType) {
            case BOOLEAN:
                primitiveType = PrimitiveType.BOOLEAN;
                break;
            case INT:
                if (logicalType instanceof LogicalTypes.Date) {
                    primitiveType = PrimitiveType.DATE;
                } else if (logicalType instanceof LogicalTypes.TimeMillis) {
                    primitiveType = PrimitiveType.TIME;
                } else {
                    primitiveType = PrimitiveType.INT;
                }
                break;
            case LONG:
                if (logicalType instanceof LogicalTypes.TimeMicros) {
                    primitiveType = PrimitiveType.TIME;
                } else if (logicalType instanceof LogicalTypes.TimestampMillis
                        || logicalType instanceof LogicalTypes.TimestampMicros) {
                    primitiveType = PrimitiveType.DATETIME;
                } else {
                    primitiveType = PrimitiveType.BIGINT;
                }
                break;
            case FLOAT:
                primitiveType = PrimitiveType.FLOAT;
                break;
            case DOUBLE:
                primitiveType = PrimitiveType.DOUBLE;
                break;
            case STRING:
                return ScalarType.createDefaultCatalogString();
            case ARRAY:
                Type type = new ArrayType(fromHudiType(avroSchema.getElementType()));
                if (type.isArrayType()) {
                    return type;
                } else {
                    isConvertedFailed = true;
                    break;
                }
            case FIXED:
            case BYTES:
                if (logicalType instanceof LogicalTypes.Decimal) {
                    int precision = ((LogicalTypes.Decimal) logicalType).getPrecision();
                    int scale = ((LogicalTypes.Decimal) logicalType).getScale();
                    return ScalarType.createUnifiedDecimalType(precision, scale);
                } else {
                    primitiveType = PrimitiveType.VARCHAR;
                    break;
                }
            case RECORD:
                // Struct type
                List<Schema.Field> fields = avroSchema.getFields();
                Preconditions.checkArgument(fields.size() > 0);
                ArrayList<StructField> structFields = new ArrayList<>(fields.size());
                for (Schema.Field field : fields) {
                    String fieldName = field.name();
                    Type fieldType = fromHudiType(field.schema());
                    if (fieldType.isUnknown()) {
                        isConvertedFailed = true;
                        break;
                    }
                    structFields.add(new StructField(fieldName, fieldType));
                }

                if (!isConvertedFailed) {
                    return new StructType(structFields);
                }
                break;
            case MAP:
                Schema value = avroSchema.getValueType();
                Type valueType = fromHudiType(value);
                if (valueType.isUnknown()) {
                    isConvertedFailed = true;
                    break;
                }

                if (!isConvertedFailed) {
                    // Hudi map's key must be string
                    return new MapType(ScalarType.createDefaultCatalogString(), valueType);
                }
                break;
            case UNION:
                List<Schema> nonNullMembers = avroSchema.getTypes().stream()
                        .filter(schema -> !Schema.Type.NULL.equals(schema.getType()))
                        .collect(Collectors.toList());

                if (nonNullMembers.size() == 1) {
                    return fromHudiType(nonNullMembers.get(0));
                } else {
                    isConvertedFailed = true;
                    break;
                }
            case ENUM:
            default:
                isConvertedFailed = true;
                break;
        }

        if (isConvertedFailed) {
            primitiveType = PrimitiveType.UNKNOWN_TYPE;
        }

        return ScalarType.createType(primitiveType);
    }
@Test
    public void testMapHudiSchema() {
        Schema.Field field1 = new Schema.Field("field1", Schema.create(Schema.Type.INT), null, null);
        Schema.Field field2 = new Schema.Field("field2", Schema.create(Schema.Type.STRING), null, null);
        List<Schema.Field> fields = new LinkedList<>();
        fields.add(field1);
        fields.add(field2);
        Schema structSchema = Schema.createRecord(fields);

        Schema mapSchema = Schema.createMap(structSchema);

        StructField structField1 = new StructField("field1", ScalarType.createType(PrimitiveType.INT));
        StructField structField2 = new StructField("field2", ScalarType.createDefaultCatalogString());
        ArrayList<StructField> structFields = new ArrayList<>();
        structFields.add(structField1);
        structFields.add(structField2);
        StructType structType = new StructType(structFields);

        MapType mapType = new MapType(ScalarType.createDefaultCatalogString(), structType);

        Assert.assertEquals(mapType, fromHudiType(mapSchema));

        mapSchema = Schema.createMap(Schema.create(Schema.Type.NULL));
        Assert.assertEquals(UNKNOWN_TYPE, fromHudiType(mapSchema));
    }