From 3b6ea7b1d0cfce648353fed11d2ccf141e5bf6ab Mon Sep 17 00:00:00 2001 From: Rajat Goel Date: Sun, 9 Aug 2026 22:39:21 +0530 Subject: [PATCH] Spark schema <-> odcs conversion --- odcs-spark/pom.xml | 38 ++ .../odcs/spark/ConversionReport.java | 56 +++ .../odcs/spark/ConversionResult.java | 24 ++ .../odcs/spark/OdcsSparkSchema.java | 122 ++++++ .../odcs/spark/SparkSchemaOptions.java | 125 ++++++ .../odcs/spark/TypeMappingException.java | 39 ++ .../odcs/spark/schema/MapTypeCodec.java | 189 +++++++++ .../odcs/spark/schema/OdcsMetadata.java | 134 ++++++ .../spark/schema/OdcsToSparkConverter.java | 216 ++++++++++ .../odcs/spark/schema/PhysicalTypeParser.java | 140 +++++++ .../spark/schema/SparkToOdcsConverter.java | 384 ++++++++++++++++++ .../odcs/spark/schema/TypeMappings.java | 244 +++++++++++ .../odcs/spark/all-data-types.odcs.yaml | 84 ++++ pom.xml | 3 + 14 files changed, 1798 insertions(+) create mode 100644 odcs-spark/pom.xml create mode 100644 odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/ConversionReport.java create mode 100644 odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/ConversionResult.java create mode 100644 odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/OdcsSparkSchema.java create mode 100644 odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/SparkSchemaOptions.java create mode 100644 odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/TypeMappingException.java create mode 100644 odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/MapTypeCodec.java create mode 100644 odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/OdcsMetadata.java create mode 100644 odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/OdcsToSparkConverter.java create mode 100644 odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/PhysicalTypeParser.java create mode 100644 odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/SparkToOdcsConverter.java create mode 100644 odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/TypeMappings.java create mode 100644 odcs-spark/src/test/resources/odcs/spark/all-data-types.odcs.yaml diff --git a/odcs-spark/pom.xml b/odcs-spark/pom.xml new file mode 100644 index 0000000..8e054ab --- /dev/null +++ b/odcs-spark/pom.xml @@ -0,0 +1,38 @@ + + + 4.0.0 + + + io.github.data-spec-labs + odcs-java-sdk + 0.1.0-SNAPSHOT + + + odcs-spark + ODCS Java SDK - Spark Integration + jar + Apache Spark integration for the Open Data Contract Standard (ODCS v3). Bi-directional schema conversion between ODCS contracts and Spark StructType, with planned DataFrame enforcement. + + + + io.github.data-spec-labs + odcs-core + ${project.version} + + + + org.apache.spark + spark-sql_${scala.binary.version} + ${spark.version} + provided + + + + org.junit.jupiter + junit-jupiter + test + + + diff --git a/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/ConversionReport.java b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/ConversionReport.java new file mode 100644 index 0000000..27642d2 --- /dev/null +++ b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/ConversionReport.java @@ -0,0 +1,56 @@ +package io.github.dataspeclabs.odcs.spark; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +/** + * Warnings collected during an ODCS ↔ Spark schema conversion (typically in LENIENT mode). + */ +public final class ConversionReport { + + private final List warnings; + + private ConversionReport(List warnings) { + this.warnings = Collections.unmodifiableList(new ArrayList<>(warnings)); + } + + public static ConversionReport empty() { + return new ConversionReport(List.of()); + } + + public static Builder builder() { + return new Builder(); + } + + public List warnings() { + return warnings; + } + + public boolean hasWarnings() { + return !warnings.isEmpty(); + } + + public static final class Builder { + private final List warnings = new ArrayList<>(); + + public Builder warn(String path, String message) { + Objects.requireNonNull(message, "message"); + if (path == null || path.isBlank()) { + warnings.add(message); + } else { + warnings.add(path + ": " + message); + } + return this; + } + + public Builder warn(String message) { + return warn(null, message); + } + + public ConversionReport build() { + return new ConversionReport(warnings); + } + } +} diff --git a/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/ConversionResult.java b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/ConversionResult.java new file mode 100644 index 0000000..287a27e --- /dev/null +++ b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/ConversionResult.java @@ -0,0 +1,24 @@ +package io.github.dataspeclabs.odcs.spark; + +import java.util.Objects; + +/** + * Conversion outcome plus any warnings collected under LENIENT mode. + * + * @param converted value type + */ +public record ConversionResult(T value, ConversionReport report) { + + public ConversionResult { + Objects.requireNonNull(value, "value"); + Objects.requireNonNull(report, "report"); + } + + public static ConversionResult of(T value, ConversionReport report) { + return new ConversionResult<>(value, report); + } + + public boolean hasWarnings() { + return report.hasWarnings(); + } +} diff --git a/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/OdcsSparkSchema.java b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/OdcsSparkSchema.java new file mode 100644 index 0000000..27ae31f --- /dev/null +++ b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/OdcsSparkSchema.java @@ -0,0 +1,122 @@ +package io.github.dataspeclabs.odcs.spark; + +import io.github.dataspeclabs.odcs.core.model.v3.DataContract; +import io.github.dataspeclabs.odcs.core.model.v3.SchemaObject; +import io.github.dataspeclabs.odcs.core.model.v3.SchemaProperty; +import io.github.dataspeclabs.odcs.spark.schema.OdcsToSparkConverter; +import io.github.dataspeclabs.odcs.spark.schema.SparkToOdcsConverter; +import org.apache.spark.sql.types.DataType; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; + +import java.util.Objects; + +/** + * Public facade for bi-directional conversion between ODCS v3 schema definitions + * and Spark SQL {@link StructType}. + * + *

Static methods use {@link SparkSchemaOptions#defaults()}. Prefer + * {@link #using(SparkSchemaOptions)} when customizing naming, strictness, or metadata. + */ +public final class OdcsSparkSchema { + + private final SparkSchemaOptions options; + + private OdcsSparkSchema(SparkSchemaOptions options) { + this.options = Objects.requireNonNull(options, "options"); + } + + public static OdcsSparkSchema using(SparkSchemaOptions options) { + return new OdcsSparkSchema(options); + } + + // ---- Static convenience (defaults) ---- + + public static StructType toStructType(SchemaObject object) { + return using(SparkSchemaOptions.defaults()).convert(object); + } + + public static StructType toStructType(DataContract contract, String objectName) { + return using(SparkSchemaOptions.defaults()).convert(contract, objectName); + } + + public static SchemaObject toSchemaObject(StructType structType, String name) { + return using(SparkSchemaOptions.defaults()).convert(structType, name); + } + + public static StructField toStructField(SchemaProperty property) { + return using(SparkSchemaOptions.defaults()).convert(property); + } + + public static DataType toDataType(SchemaProperty property) { + return using(SparkSchemaOptions.defaults()).convertDataType(property); + } + + public static SchemaProperty toSchemaProperty(StructField field) { + return using(SparkSchemaOptions.defaults()).convert(field); + } + + public static SchemaProperty toSchemaProperty(String name, DataType dataType, boolean nullable) { + return using(SparkSchemaOptions.defaults()).convert(name, dataType, nullable); + } + + // ---- Instance API ---- + + public StructType convert(SchemaObject object) { + return convertWithReport(object).value(); + } + + public StructType convert(DataContract contract, String objectName) { + return convertWithReport(contract, objectName).value(); + } + + public StructField convert(SchemaProperty property) { + ConversionReport.Builder report = ConversionReport.builder(); + return new OdcsToSparkConverter(options, report).toStructField(property); + } + + public DataType convertDataType(SchemaProperty property) { + ConversionReport.Builder report = ConversionReport.builder(); + return new OdcsToSparkConverter(options, report).toDataType(property); + } + + public SchemaObject convert(StructType structType, String name) { + return convertWithReport(structType, name).value(); + } + + public SchemaProperty convert(StructField field) { + ConversionReport.Builder report = ConversionReport.builder(); + return new SparkToOdcsConverter(options, report).toSchemaProperty(field); + } + + public SchemaProperty convert(String name, DataType dataType, boolean nullable) { + ConversionReport.Builder report = ConversionReport.builder(); + return new SparkToOdcsConverter(options, report).toSchemaProperty(name, dataType, nullable); + } + + public ConversionResult convertWithReport(SchemaObject object) { + ConversionReport.Builder report = ConversionReport.builder(); + StructType st = new OdcsToSparkConverter(options, report).toStructType(object); + return ConversionResult.of(st, report.build()); + } + + public ConversionResult convertWithReport(DataContract contract, String objectName) { + Objects.requireNonNull(contract, "contract"); + Objects.requireNonNull(objectName, "objectName"); + if (contract.schema() == null) { + throw new TypeMappingException("", "DataContract has no schema objects"); + } + SchemaObject match = contract.schema().stream() + .filter(o -> objectName.equals(o.name()) || objectName.equals(o.physicalName())) + .findFirst() + .orElseThrow(() -> new TypeMappingException( + objectName, "no schema object named '" + objectName + "' in contract")); + return convertWithReport(match); + } + + public ConversionResult convertWithReport(StructType structType, String name) { + ConversionReport.Builder report = ConversionReport.builder(); + SchemaObject object = new SparkToOdcsConverter(options, report).toSchemaObject(structType, name); + return ConversionResult.of(object, report.build()); + } +} diff --git a/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/SparkSchemaOptions.java b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/SparkSchemaOptions.java new file mode 100644 index 0000000..b0b441e --- /dev/null +++ b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/SparkSchemaOptions.java @@ -0,0 +1,125 @@ +package io.github.dataspeclabs.odcs.spark; + +import java.util.Objects; + +/** + * Immutable configuration for {@link OdcsSparkSchema} conversions. + */ +public final class SparkSchemaOptions { + + public enum FieldNameSource { + /** Use {@code physicalName} when set, otherwise {@code name}. */ + PHYSICAL_THEN_LOGICAL, + /** Always use the ODCS logical {@code name}. */ + LOGICAL_ONLY + } + + public enum Strictness { + /** Unresolved or unsupported types throw {@link TypeMappingException}. */ + STRICT, + /** Unresolved types fall back to {@code StringType} and record a warning. */ + LENIENT + } + + public enum TimeTypeMapping { + /** Map ODCS {@code time} to Spark {@code StringType}, preserving {@code physicalType: time}. */ + STRING, + /** Map ODCS {@code time} to Spark {@code LongType} (microseconds since midnight). */ + LONG_MICROS + } + + private final FieldNameSource fieldNameSource; + private final Strictness strictness; + private final boolean emitMetadata; + private final boolean readMetadata; + private final TimeTypeMapping timeTypeMapping; + private final boolean requireDecimalPrecision; + + private SparkSchemaOptions(Builder builder) { + this.fieldNameSource = builder.fieldNameSource; + this.strictness = builder.strictness; + this.emitMetadata = builder.emitMetadata; + this.readMetadata = builder.readMetadata; + this.timeTypeMapping = builder.timeTypeMapping; + this.requireDecimalPrecision = builder.requireDecimalPrecision; + } + + public static SparkSchemaOptions defaults() { + return builder().build(); + } + + public static Builder builder() { + return new Builder(); + } + + public FieldNameSource fieldNameSource() { + return fieldNameSource; + } + + public Strictness strictness() { + return strictness; + } + + public boolean emitMetadata() { + return emitMetadata; + } + + public boolean readMetadata() { + return readMetadata; + } + + public TimeTypeMapping timeTypeMapping() { + return timeTypeMapping; + } + + public boolean requireDecimalPrecision() { + return requireDecimalPrecision; + } + + public boolean isStrict() { + return strictness == Strictness.STRICT; + } + + public static final class Builder { + private FieldNameSource fieldNameSource = FieldNameSource.PHYSICAL_THEN_LOGICAL; + private Strictness strictness = Strictness.STRICT; + private boolean emitMetadata = true; + private boolean readMetadata = true; + private TimeTypeMapping timeTypeMapping = TimeTypeMapping.STRING; + private boolean requireDecimalPrecision = true; + + public Builder fieldNameSource(FieldNameSource fieldNameSource) { + this.fieldNameSource = Objects.requireNonNull(fieldNameSource, "fieldNameSource"); + return this; + } + + public Builder strictness(Strictness strictness) { + this.strictness = Objects.requireNonNull(strictness, "strictness"); + return this; + } + + public Builder emitMetadata(boolean emitMetadata) { + this.emitMetadata = emitMetadata; + return this; + } + + public Builder readMetadata(boolean readMetadata) { + this.readMetadata = readMetadata; + return this; + } + + public Builder timeTypeMapping(TimeTypeMapping timeTypeMapping) { + this.timeTypeMapping = Objects.requireNonNull(timeTypeMapping, "timeTypeMapping"); + return this; + } + + public Builder requireDecimalPrecision(boolean requireDecimalPrecision) { + this.requireDecimalPrecision = requireDecimalPrecision; + return this; + } + + public SparkSchemaOptions build() { + return new SparkSchemaOptions(this); + } + } +} diff --git a/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/TypeMappingException.java b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/TypeMappingException.java new file mode 100644 index 0000000..386c12b --- /dev/null +++ b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/TypeMappingException.java @@ -0,0 +1,39 @@ +package io.github.dataspeclabs.odcs.spark; + +/** + * Thrown when an ODCS ↔ Spark type conversion cannot be performed under + * {@link SparkSchemaOptions.Strictness#STRICT} mode. + */ +public class TypeMappingException extends RuntimeException { + + private final String path; + private final String detail; + + public TypeMappingException(String path, String message) { + super(format(path, message)); + this.path = path; + this.detail = message; + } + + public TypeMappingException(String path, String message, Throwable cause) { + super(format(path, message), cause); + this.path = path; + this.detail = message; + } + + public String path() { + return path; + } + + /** Message without the path prefix. */ + public String detail() { + return detail; + } + + private static String format(String path, String message) { + if (path == null || path.isBlank()) { + return message; + } + return path + ": " + message; + } +} diff --git a/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/MapTypeCodec.java b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/MapTypeCodec.java new file mode 100644 index 0000000..3766b60 --- /dev/null +++ b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/MapTypeCodec.java @@ -0,0 +1,189 @@ +package io.github.dataspeclabs.odcs.spark.schema; + +import io.github.dataspeclabs.odcs.core.model.v3.CustomProperty; +import io.github.dataspeclabs.odcs.core.model.v3.LogicalType; +import io.github.dataspeclabs.odcs.core.model.v3.SchemaProperty; +import io.github.dataspeclabs.odcs.spark.ConversionReport; +import io.github.dataspeclabs.odcs.spark.SparkSchemaOptions; +import io.github.dataspeclabs.odcs.spark.TypeMappingException; +import org.apache.spark.sql.types.DataType; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.MapType; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; + +import java.util.ArrayList; +import java.util.List; +import java.util.Locale; +import java.util.Optional; +import java.util.function.BiFunction; + +/** + * A+C hybrid map codec: {@code physicalType: map} + {@code mapKeyType}/{@code mapValueType} + * custom properties, with {@code map<k,v>} as a secondary parse path. + * Isolated here so ODCS v3.2 {@code logicalType: map} is a one-class change. + */ +public final class MapTypeCodec { + + public static final String MAP_KEY_TYPE = "mapKeyType"; + public static final String MAP_VALUE_TYPE = "mapValueType"; + public static final String MAP_VALUE_REQUIRED = "mapValueRequired"; + + private MapTypeCodec() { + } + + public static boolean isMapProperty(SchemaProperty property) { + if (property == null) { + return false; + } + if (PhysicalTypeParser.parseMap(property.physicalType()).isPresent()) { + return true; + } + return OdcsMetadata.customString(property, MAP_KEY_TYPE).isPresent() + || OdcsMetadata.customString(property, MAP_VALUE_TYPE).isPresent(); + } + + public static DataType toMapType( + SchemaProperty property, + SparkSchemaOptions options, + String path, + ConversionReport.Builder report, + BiFunction, String, StructType> nestedStructConverter + ) { + Optional keyToken = OdcsMetadata.customString(property, MAP_KEY_TYPE); + Optional valueToken = OdcsMetadata.customString(property, MAP_VALUE_TYPE); + Optional mapSpec = PhysicalTypeParser.parseMap(property.physicalType()); + + if (keyToken.isEmpty() && mapSpec.isPresent() && !mapSpec.get().isBare()) { + keyToken = Optional.ofNullable(mapSpec.get().keyType()); + } + if (valueToken.isEmpty() && mapSpec.isPresent() && !mapSpec.get().isBare()) { + valueToken = Optional.ofNullable(mapSpec.get().valueType()); + } + + if (keyToken.isEmpty()) { + throwOrWarn(options, report, path, "map property missing mapKeyType", DataTypes.StringType); + keyToken = Optional.of("string"); + } + if (valueToken.isEmpty()) { + throwOrWarn(options, report, path, "map property missing mapValueType", DataTypes.StringType); + valueToken = Optional.of("string"); + } + + DataType keyType = resolveToken( + keyToken.get(), property, options, path + ".key", report, nestedStructConverter, false); + DataType valueType = resolveToken( + valueToken.get(), property, options, path + ".value", report, nestedStructConverter, true); + + boolean valueContainsNull = true; + Optional required = OdcsMetadata.customString(property, MAP_VALUE_REQUIRED); + if (required.isPresent() && "true".equalsIgnoreCase(required.get())) { + valueContainsNull = false; + } + + return DataTypes.createMapType(keyType, valueType, valueContainsNull); + } + + private static DataType resolveToken( + String token, + SchemaProperty property, + SparkSchemaOptions options, + String path, + ConversionReport.Builder report, + BiFunction, String, StructType> nestedStructConverter, + boolean allowObject + ) { + String t = token.trim().toLowerCase(Locale.ROOT); + if (allowObject && ("object".equals(t) || "struct".equals(t))) { + List props = property.properties(); + if (props == null || props.isEmpty()) { + throwOrWarn(options, report, path, + "map value type object requires nested properties", DataTypes.StringType); + return DataTypes.StringType; + } + return nestedStructConverter.apply(props, path); + } + return PhysicalTypeParser.parseScalarToken(token, options.requireDecimalPrecision()) + .orElseGet(() -> { + throwOrWarn(options, report, path, "unsupported map type token: " + token, DataTypes.StringType); + return DataTypes.StringType; + }); + } + + /** + * Import Spark {@link MapType} into the round-trippable ODCS convention. + */ + public static SchemaProperty fromMapType( + String name, + MapType mapType, + boolean nullable, + String description + ) { + String keyPhysical = PhysicalTypeParser.sparkPhysicalToken(mapType.keyType()); + String valuePhysical; + List nestedProps = null; + + if (mapType.valueType() instanceof StructType st) { + valuePhysical = "object"; + nestedProps = new ArrayList<>(); + for (StructField field : st.fields()) { + nestedProps.add(SparkToOdcsConverter.inferProperty(field)); + } + } else { + valuePhysical = PhysicalTypeParser.sparkPhysicalToken(mapType.valueType()); + } + + List custom = new ArrayList<>(); + custom.add(new CustomProperty(null, MAP_KEY_TYPE, keyPhysical, null)); + custom.add(new CustomProperty(null, MAP_VALUE_TYPE, valuePhysical, null)); + if (!mapType.valueContainsNull()) { + custom.add(new CustomProperty(null, MAP_VALUE_REQUIRED, "true", null)); + } + + Boolean required = nullable ? null : Boolean.TRUE; + + return new SchemaProperty( + null, + name, + null, + "map", + description, + null, + LogicalType.OBJECT, + null, + null, + null, + required, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + nestedProps, + null, + null, + null, + null, + custom + ); + } + + private static void throwOrWarn( + SparkSchemaOptions options, + ConversionReport.Builder report, + String path, + String message, + DataType fallback + ) { + if (options.isStrict()) { + throw new TypeMappingException(path, message); + } + report.warn(path, message + "; falling back to " + fallback.simpleString()); + } +} diff --git a/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/OdcsMetadata.java b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/OdcsMetadata.java new file mode 100644 index 0000000..b8596f2 --- /dev/null +++ b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/OdcsMetadata.java @@ -0,0 +1,134 @@ +package io.github.dataspeclabs.odcs.spark.schema; + +import io.github.dataspeclabs.odcs.core.model.v3.CustomProperty; +import io.github.dataspeclabs.odcs.core.model.v3.LogicalType; +import io.github.dataspeclabs.odcs.core.model.v3.SchemaProperty; +import org.apache.spark.sql.types.Metadata; +import org.apache.spark.sql.types.MetadataBuilder; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Optional; + +/** + * Codec for ODCS provenance stored on Spark {@link org.apache.spark.sql.types.StructField} metadata. + */ +public final class OdcsMetadata { + + public static final String COMMENT = "comment"; + public static final String NAME = "odcs.name"; + public static final String LOGICAL_TYPE = "odcs.logicalType"; + public static final String PHYSICAL_TYPE = "odcs.physicalType"; + public static final String PRIMARY_KEY = "odcs.primaryKey"; + public static final String PRIMARY_KEY_POSITION = "odcs.primaryKeyPosition"; + public static final String CLASSIFICATION = "odcs.classification"; + public static final String CRITICAL_DATA_ELEMENT = "odcs.criticalDataElement"; + public static final String TAGS = "odcs.tags"; + public static final String MAP_KEY_TYPE = "odcs.mapKeyType"; + public static final String MAP_VALUE_TYPE = "odcs.mapValueType"; + public static final String MAP_VALUE_REQUIRED = "odcs.mapValueRequired"; + + private OdcsMetadata() { + } + + public static Metadata build(SchemaProperty property, String sparkFieldName, boolean usedPhysicalName) { + MetadataBuilder builder = new MetadataBuilder(); + + if (property.description() != null && !property.description().isBlank()) { + builder.putString(COMMENT, property.description()); + } + if (usedPhysicalName && property.name() != null) { + builder.putString(NAME, property.name()); + } + if (property.logicalType() != null) { + builder.putString(LOGICAL_TYPE, property.logicalType().value()); + } + if (property.physicalType() != null) { + builder.putString(PHYSICAL_TYPE, property.physicalType()); + } + if (property.primaryKey() != null) { + builder.putBoolean(PRIMARY_KEY, property.primaryKey()); + } + if (property.primaryKeyPosition() != null) { + builder.putLong(PRIMARY_KEY_POSITION, property.primaryKeyPosition()); + } + if (property.classification() != null) { + builder.putString(CLASSIFICATION, property.classification()); + } + if (property.criticalDataElement() != null) { + builder.putBoolean(CRITICAL_DATA_ELEMENT, property.criticalDataElement()); + } + if (property.tags() != null && !property.tags().isEmpty()) { + builder.putStringArray(TAGS, property.tags().toArray(String[]::new)); + } + + customString(property, MapTypeCodec.MAP_KEY_TYPE) + .ifPresent(v -> builder.putString(MAP_KEY_TYPE, v)); + customString(property, MapTypeCodec.MAP_VALUE_TYPE) + .ifPresent(v -> builder.putString(MAP_VALUE_TYPE, v)); + customString(property, MapTypeCodec.MAP_VALUE_REQUIRED) + .ifPresent(v -> builder.putString(MAP_VALUE_REQUIRED, v)); + + return builder.build(); + } + + public static boolean hasOdcsKeys(Metadata metadata) { + if (metadata == null) { + return false; + } + return metadata.contains(LOGICAL_TYPE) + || metadata.contains(PHYSICAL_TYPE) + || metadata.contains(NAME) + || metadata.contains(MAP_KEY_TYPE); + } + + public static Optional getString(Metadata metadata, String key) { + if (metadata == null || !metadata.contains(key)) { + return Optional.empty(); + } + return Optional.ofNullable(metadata.getString(key)); + } + + public static Optional getBoolean(Metadata metadata, String key) { + if (metadata == null || !metadata.contains(key)) { + return Optional.empty(); + } + return Optional.of(metadata.getBoolean(key)); + } + + public static Optional getInt(Metadata metadata, String key) { + if (metadata == null || !metadata.contains(key)) { + return Optional.empty(); + } + return Optional.of((int) metadata.getLong(key)); + } + + public static List getTags(Metadata metadata) { + if (metadata == null || !metadata.contains(TAGS)) { + return List.of(); + } + String[] arr = metadata.getStringArray(TAGS); + if (arr == null || arr.length == 0) { + return List.of(); + } + return Collections.unmodifiableList(new ArrayList<>(Arrays.asList(arr))); + } + + public static Optional getLogicalType(Metadata metadata) { + return getString(metadata, LOGICAL_TYPE).map(LogicalType::fromValue); + } + + public static Optional customString(SchemaProperty property, String key) { + if (property.customProperties() == null) { + return Optional.empty(); + } + for (CustomProperty cp : property.customProperties()) { + if (cp != null && key.equals(cp.property()) && cp.value() != null) { + return Optional.of(String.valueOf(cp.value())); + } + } + return Optional.empty(); + } +} diff --git a/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/OdcsToSparkConverter.java b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/OdcsToSparkConverter.java new file mode 100644 index 0000000..4e0aebc --- /dev/null +++ b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/OdcsToSparkConverter.java @@ -0,0 +1,216 @@ +package io.github.dataspeclabs.odcs.spark.schema; + +import io.github.dataspeclabs.odcs.core.model.v3.LogicalType; +import io.github.dataspeclabs.odcs.core.model.v3.SchemaObject; +import io.github.dataspeclabs.odcs.core.model.v3.SchemaProperty; +import io.github.dataspeclabs.odcs.spark.ConversionReport; +import io.github.dataspeclabs.odcs.spark.SparkSchemaOptions; +import io.github.dataspeclabs.odcs.spark.TypeMappingException; +import org.apache.spark.sql.types.DataType; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.Metadata; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; + +import java.util.ArrayList; +import java.util.List; +import java.util.Locale; +import java.util.Optional; + +/** + * Converts ODCS schema definitions to Spark {@link StructType} / {@link DataType}. + */ +public final class OdcsToSparkConverter { + + private final SparkSchemaOptions options; + private final ConversionReport.Builder report; + + public OdcsToSparkConverter(SparkSchemaOptions options, ConversionReport.Builder report) { + this.options = options; + this.report = report; + } + + public StructType toStructType(SchemaObject object) { + if (object == null) { + throw new TypeMappingException("", "SchemaObject must not be null"); + } + String root = object.name() != null ? object.name() : ""; + return propertiesToStruct(object.properties(), root); + } + + public StructField toStructField(SchemaProperty property) { + return toStructField(property, propertyPath(null, property)); + } + + public DataType toDataType(SchemaProperty property) { + return resolveDataType(property, propertyPath(null, property)); + } + + StructType propertiesToStruct(List properties, String path) { + if (properties == null || properties.isEmpty()) { + failOrWarn(path, "object/schema has no properties", new StructType()); + return new StructType(); + } + List fields = new ArrayList<>(properties.size()); + for (SchemaProperty property : properties) { + fields.add(toStructField(property, propertyPath(path, property))); + } + return new StructType(fields.toArray(StructField[]::new)); + } + + private StructField toStructField(SchemaProperty property, String path) { + if (property == null) { + throw new TypeMappingException(path, "SchemaProperty must not be null"); + } + String fieldName = resolveFieldName(property, path); + boolean usedPhysicalName = options.fieldNameSource() == SparkSchemaOptions.FieldNameSource.PHYSICAL_THEN_LOGICAL + && property.physicalName() != null + && !property.physicalName().isBlank(); + + DataType dataType = resolveDataType(property, path); + boolean nullable = !Boolean.TRUE.equals(property.required()); + + Metadata metadata = options.emitMetadata() + ? OdcsMetadata.build(property, fieldName, usedPhysicalName) + : Metadata.empty(); + + return new StructField(fieldName, dataType, nullable, metadata); + } + + private DataType resolveDataType(SchemaProperty property, String path) { + // 1. Map codec + if (MapTypeCodec.isMapProperty(property)) { + return MapTypeCodec.toMapType( + property, + options, + path, + report, + this::propertiesToStruct); + } + + LogicalType logicalType = property.logicalType(); + + // Nested object / array before physicalType scalar parse + if (logicalType == LogicalType.ARRAY) { + return resolveArray(property, path); + } + if (logicalType == LogicalType.OBJECT) { + return propertiesToStruct(property.properties(), path); + } + + // 2. physicalType refinement that fully determines the type + Optional fromPhysical = tryPhysicalRefinement(property, path); + if (fromPhysical.isPresent()) { + return fromPhysical.get(); + } + + // 3–4. logicalType + options / defaults + if (logicalType == null) { + failOrWarn(path, "missing logicalType", DataTypes.StringType); + return DataTypes.StringType; + } + + return TypeMappings.fromLogicalType( + logicalType, + property.logicalTypeOptions(), + property.physicalType(), + options, + path, + report); + } + + private Optional tryPhysicalRefinement(SchemaProperty property, String path) { + String physical = property.physicalType(); + if (physical == null || physical.isBlank()) { + return Optional.empty(); + } + + Optional decimal = PhysicalTypeParser.parseDecimal(physical); + if (decimal.isPresent()) { + PhysicalTypeParser.DecimalSpec d = decimal.get(); + if (d.isBare()) { + if (options.requireDecimalPrecision()) { + failOrWarn(path, "decimal physicalType requires precision and scale, e.g. decimal(18,2)", + DataTypes.createDecimalType(10, 0)); + return Optional.of(DataTypes.createDecimalType(10, 0)); + } + return Optional.of(DataTypes.createDecimalType(10, 0)); + } + return Optional.of(DataTypes.createDecimalType(d.precision(), d.scale())); + } + + String p = physical.trim().toLowerCase(Locale.ROOT); + return switch (p) { + case "timestamp_ntz", "timestampntz" -> Optional.of(DataTypes.TimestampNTZType); + case "timestamp", "datetime" -> + property.logicalType() == LogicalType.TIMESTAMP + ? Optional.of(DataTypes.TimestampType) + : Optional.empty(); + case "binary", "bytes", "bytea" -> Optional.of(DataTypes.BinaryType); + case "tinyint", "byte" -> Optional.of(DataTypes.ByteType); + case "smallint", "short" -> Optional.of(DataTypes.ShortType); + case "int", "integer" -> Optional.of(DataTypes.IntegerType); + case "bigint", "long" -> Optional.of(DataTypes.LongType); + case "float", "real" -> Optional.of(DataTypes.FloatType); + case "double" -> Optional.of(DataTypes.DoubleType); + default -> { + if ((property.logicalType() == null || property.logicalType() == LogicalType.STRING) + && PhysicalTypeParser.parseVarcharLength(physical).isPresent()) { + yield Optional.of(DataTypes.StringType); + } + yield Optional.empty(); + } + }; + } + + private DataType resolveArray(SchemaProperty property, String path) { + SchemaProperty items = property.items(); + if (items == null) { + failOrWarn(path, "array logicalType requires items", DataTypes.StringType); + return DataTypes.createArrayType(DataTypes.StringType, true); + } + DataType elementType = resolveDataType(items, path + ".items"); + boolean containsNull = !Boolean.TRUE.equals(items.required()); + return DataTypes.createArrayType(elementType, containsNull); + } + + private String resolveFieldName(SchemaProperty property, String path) { + if (options.fieldNameSource() == SparkSchemaOptions.FieldNameSource.LOGICAL_ONLY) { + if (property.name() == null || property.name().isBlank()) { + throw new TypeMappingException(path, "property name is required"); + } + return property.name(); + } + if (property.physicalName() != null && !property.physicalName().isBlank()) { + return property.physicalName(); + } + if (property.name() == null || property.name().isBlank()) { + throw new TypeMappingException(path, "property name is required"); + } + return property.name(); + } + + private static String propertyPath(String parent, SchemaProperty property) { + String name = property == null ? "?" + : (property.name() != null ? property.name() + : (property.physicalName() != null ? property.physicalName() : "?")); + if (parent == null || parent.isBlank()) { + return name; + } + return parent + "." + name; + } + + private void failOrWarn(String path, String message, DataType fallback) { + if (options.isStrict()) { + throw new TypeMappingException(path, message); + } + report.warn(path, message + "; falling back to " + fallback.simpleString()); + } + + private void failOrWarn(String path, String message, StructType fallback) { + if (options.isStrict()) { + throw new TypeMappingException(path, message); + } + report.warn(path, message + "; falling back to empty struct"); + } +} diff --git a/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/PhysicalTypeParser.java b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/PhysicalTypeParser.java new file mode 100644 index 0000000..1906445 --- /dev/null +++ b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/PhysicalTypeParser.java @@ -0,0 +1,140 @@ +package io.github.dataspeclabs.odcs.spark.schema; + +import org.apache.spark.sql.types.DataType; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.DecimalType; + +import java.util.Locale; +import java.util.Optional; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * Parses free-form ODCS {@code physicalType} strings into Spark {@link DataType}s + * or structured tokens (decimal, map<k,v>, varchar). + */ +public final class PhysicalTypeParser { + + private static final Pattern DECIMAL = + Pattern.compile("(?i)^decimal\\s*\\(\\s*(\\d+)\\s*,\\s*(\\d+)\\s*\\)$"); + private static final Pattern DECIMAL_BARE = + Pattern.compile("(?i)^decimal$"); + private static final Pattern VARCHAR = + Pattern.compile("(?i)^(?:var)?char\\s*\\(\\s*(\\d+)\\s*\\)$"); + private static final Pattern MAP = + Pattern.compile("(?i)^map\\s*<\\s*(.+?)\\s*,\\s*(.+?)\\s*>$"); + + private PhysicalTypeParser() { + } + + public static Optional parseDecimal(String physicalType) { + if (physicalType == null || physicalType.isBlank()) { + return Optional.empty(); + } + String trimmed = physicalType.trim(); + Matcher m = DECIMAL.matcher(trimmed); + if (m.matches()) { + return Optional.of(new DecimalSpec(Integer.parseInt(m.group(1)), Integer.parseInt(m.group(2)))); + } + if (DECIMAL_BARE.matcher(trimmed).matches()) { + return Optional.of(DecimalSpec.unspecified()); + } + return Optional.empty(); + } + + public static Optional parseVarcharLength(String physicalType) { + if (physicalType == null || physicalType.isBlank()) { + return Optional.empty(); + } + Matcher m = VARCHAR.matcher(physicalType.trim()); + if (m.matches()) { + return Optional.of(Integer.parseInt(m.group(1))); + } + return Optional.empty(); + } + + public static Optional parseMap(String physicalType) { + if (physicalType == null || physicalType.isBlank()) { + return Optional.empty(); + } + String trimmed = physicalType.trim(); + if ("map".equalsIgnoreCase(trimmed)) { + return Optional.of(MapSpec.unspecified()); + } + Matcher m = MAP.matcher(trimmed); + if (m.matches()) { + return Optional.of(new MapSpec(m.group(1).trim(), m.group(2).trim())); + } + return Optional.empty(); + } + + public static boolean isExactMap(String physicalType) { + return physicalType != null && "map".equalsIgnoreCase(physicalType.trim()); + } + + /** + * Resolve a scalar Spark physical type token (no nesting). + */ + public static Optional parseScalarToken(String token, boolean requireDecimalPrecision) { + if (token == null || token.isBlank()) { + return Optional.empty(); + } + String t = token.trim().toLowerCase(Locale.ROOT); + + Optional decimal = parseDecimal(token); + if (decimal.isPresent()) { + DecimalSpec d = decimal.get(); + if (d.isBare()) { + if (requireDecimalPrecision) { + return Optional.empty(); + } + return Optional.of(DataTypes.createDecimalType(10, 0)); + } + return Optional.of(DataTypes.createDecimalType(d.precision(), d.scale())); + } + + return switch (t) { + case "string", "str", "text", "varchar", "char" -> Optional.of(DataTypes.StringType); + case "boolean", "bool" -> Optional.of(DataTypes.BooleanType); + case "byte", "tinyint", "int8" -> Optional.of(DataTypes.ByteType); + case "short", "smallint", "int16" -> Optional.of(DataTypes.ShortType); + case "int", "integer", "int32" -> Optional.of(DataTypes.IntegerType); + case "long", "bigint", "int64" -> Optional.of(DataTypes.LongType); + case "float", "real", "float32" -> Optional.of(DataTypes.FloatType); + case "double", "float64" -> Optional.of(DataTypes.DoubleType); + case "date" -> Optional.of(DataTypes.DateType); + case "timestamp", "datetime" -> Optional.of(DataTypes.TimestampType); + case "timestamp_ntz", "timestampntz" -> Optional.of(DataTypes.TimestampNTZType); + case "binary", "bytes", "bytea" -> Optional.of(DataTypes.BinaryType); + case "time" -> Optional.of(DataTypes.StringType); + default -> Optional.empty(); + }; + } + + public static String sparkPhysicalToken(DataType dataType) { + if (dataType instanceof DecimalType dt) { + return "decimal(" + dt.precision() + "," + dt.scale() + ")"; + } + return dataType.simpleString().toLowerCase(Locale.ROOT); + } + + public record DecimalSpec(int precision, int scale, boolean isBare) { + public DecimalSpec(int precision, int scale) { + this(precision, scale, false); + } + + public static DecimalSpec unspecified() { + return new DecimalSpec(-1, -1, true); + } + } + + public record MapSpec(String keyType, String valueType, boolean isBare) { + public MapSpec(String keyType, String valueType) { + this(keyType, valueType, false); + } + + public static MapSpec unspecified() { + return new MapSpec(null, null, true); + } + } +} diff --git a/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/SparkToOdcsConverter.java b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/SparkToOdcsConverter.java new file mode 100644 index 0000000..c2afbe4 --- /dev/null +++ b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/SparkToOdcsConverter.java @@ -0,0 +1,384 @@ +package io.github.dataspeclabs.odcs.spark.schema; + +import io.github.dataspeclabs.odcs.core.model.v3.LogicalType; +import io.github.dataspeclabs.odcs.core.model.v3.LogicalTypeOptions; +import io.github.dataspeclabs.odcs.core.model.v3.SchemaObject; +import io.github.dataspeclabs.odcs.core.model.v3.SchemaProperty; +import io.github.dataspeclabs.odcs.spark.ConversionReport; +import io.github.dataspeclabs.odcs.spark.SparkSchemaOptions; +import io.github.dataspeclabs.odcs.spark.TypeMappingException; +import org.apache.spark.sql.types.ArrayType; +import org.apache.spark.sql.types.DataType; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.DecimalType; +import org.apache.spark.sql.types.MapType; +import org.apache.spark.sql.types.Metadata; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; + +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +/** + * Converts Spark {@link StructType} / {@link DataType} to ODCS schema definitions. + */ +public final class SparkToOdcsConverter { + + private final SparkSchemaOptions options; + private final ConversionReport.Builder report; + + public SparkToOdcsConverter(SparkSchemaOptions options, ConversionReport.Builder report) { + this.options = options; + this.report = report; + } + + public SchemaObject toSchemaObject(StructType structType, String name) { + if (structType == null) { + throw new TypeMappingException("", "StructType must not be null"); + } + if (name == null || name.isBlank()) { + throw new TypeMappingException("", "schema object name is required"); + } + List properties = new ArrayList<>(); + for (StructField field : structType.fields()) { + properties.add(toSchemaProperty(field)); + } + return new SchemaObject( + null, + name, + null, + "table", + LogicalType.OBJECT, + null, + null, + null, + properties, + null, + null, + null, + null, + null + ); + } + + public SchemaProperty toSchemaProperty(StructField field) { + if (field == null) { + throw new TypeMappingException("", "StructField must not be null"); + } + Metadata metadata = field.metadata(); + if (options.readMetadata() && OdcsMetadata.hasOdcsKeys(metadata)) { + return restoreFromMetadata(field); + } + try { + return inferProperty(field); + } catch (TypeMappingException ex) { + if (options.isStrict()) { + throw ex; + } + report.warn(ex.path(), ex.detail() + "; falling back to string"); + Boolean required = field.nullable() ? null : Boolean.TRUE; + return property(field.name(), null, "string", LogicalType.STRING, null, required, + OdcsMetadata.getString(field.metadata(), OdcsMetadata.COMMENT).orElse(null), + null, null, null); + } + } + + public SchemaProperty toSchemaProperty(String name, DataType dataType, boolean nullable) { + StructField field = new StructField(name, dataType, nullable, Metadata.empty()); + return toSchemaProperty(field); + } + + /** + * Infer ODCS property from Spark type alone (no metadata). Package-visible for {@link MapTypeCodec}. + */ + static SchemaProperty inferProperty(StructField field) { + return inferProperty(field.name(), field.dataType(), field.nullable(), + OdcsMetadata.getString(field.metadata(), OdcsMetadata.COMMENT).orElse(null)); + } + + private static SchemaProperty inferProperty( + String name, + DataType dataType, + boolean nullable, + String description + ) { + Boolean required = nullable ? null : Boolean.TRUE; + + if (dataType instanceof MapType mapType) { + return MapTypeCodec.fromMapType(name, mapType, nullable, description); + } + if (dataType instanceof ArrayType arrayType) { + SchemaProperty items = inferProperty( + "items", + arrayType.elementType(), + arrayType.containsNull(), + null); + // items.required=true means containsNull=false + if (!arrayType.containsNull()) { + items = copyWithRequired(items, true); + } else { + items = copyWithRequired(items, null); + } + return property(name, null, "array", LogicalType.ARRAY, null, required, description, items, null, null); + } + if (dataType instanceof StructType structType) { + List nested = new ArrayList<>(); + for (StructField f : structType.fields()) { + nested.add(inferProperty(f)); + } + return property(name, null, "struct", LogicalType.OBJECT, null, required, description, null, nested, null); + } + if (DataTypes.TimestampNTZType.sameType(dataType)) { + LogicalTypeOptions opts = new LogicalTypeOptions( + null, null, null, null, null, null, null, null, null, + "false", null, null, null, null, null, null, null); + return property(name, null, "timestamp_ntz", LogicalType.TIMESTAMP, opts, required, description, null, null, null); + } + if (DataTypes.TimestampType.sameType(dataType)) { + return property(name, null, "timestamp", LogicalType.TIMESTAMP, null, required, description, null, null, null); + } + if (DataTypes.DateType.sameType(dataType)) { + return property(name, null, "date", LogicalType.DATE, null, required, description, null, null, null); + } + if (DataTypes.BooleanType.sameType(dataType)) { + return property(name, null, "boolean", LogicalType.BOOLEAN, null, required, description, null, null, null); + } + if (DataTypes.BinaryType.sameType(dataType)) { + LogicalTypeOptions opts = new LogicalTypeOptions( + "binary", null, null, null, null, null, null, null, null, + null, null, null, null, null, null, null, null); + return property(name, null, "binary", LogicalType.STRING, opts, required, description, null, null, null); + } + if (dataType instanceof DecimalType dt) { + String physical = "decimal(" + dt.precision() + "," + dt.scale() + ")"; + return property(name, null, physical, LogicalType.NUMBER, null, required, description, null, null, null); + } + if (DataTypes.FloatType.sameType(dataType)) { + LogicalTypeOptions opts = optionsWithFormat("f32"); + return property(name, null, "float", LogicalType.NUMBER, opts, required, description, null, null, null); + } + if (DataTypes.DoubleType.sameType(dataType)) { + LogicalTypeOptions opts = optionsWithFormat("f64"); + return property(name, null, "double", LogicalType.NUMBER, opts, required, description, null, null, null); + } + if (DataTypes.ByteType.sameType(dataType)) { + return property(name, null, "tinyint", LogicalType.INTEGER, optionsWithFormat("i8"), required, description, null, null, null); + } + if (DataTypes.ShortType.sameType(dataType)) { + return property(name, null, "smallint", LogicalType.INTEGER, optionsWithFormat("i16"), required, description, null, null, null); + } + if (DataTypes.IntegerType.sameType(dataType)) { + return property(name, null, "int", LogicalType.INTEGER, optionsWithFormat("i32"), required, description, null, null, null); + } + if (DataTypes.LongType.sameType(dataType)) { + return property(name, null, "bigint", LogicalType.INTEGER, optionsWithFormat("i64"), required, description, null, null, null); + } + if (DataTypes.StringType.sameType(dataType)) { + return property(name, null, "string", LogicalType.STRING, null, required, description, null, null, null); + } + + // Unsupported — callers using instance methods handle STRICT/LENIENT; static infer throws. + throw new TypeMappingException(name, "unsupported Spark type: " + dataType.simpleString()); + } + + private SchemaProperty restoreFromMetadata(StructField field) { + Metadata md = field.metadata(); + String logicalName = OdcsMetadata.getString(md, OdcsMetadata.NAME).orElse(field.name()); + String physicalName = OdcsMetadata.getString(md, OdcsMetadata.NAME).isPresent() ? field.name() : null; + // When odcs.name is present, Spark field was physicalName; logical name is odcs.name + if (OdcsMetadata.getString(md, OdcsMetadata.NAME).isPresent()) { + physicalName = field.name(); + } else { + physicalName = null; + } + + Optional metaLogical = OdcsMetadata.getLogicalType(md); + String physicalType = OdcsMetadata.getString(md, OdcsMetadata.PHYSICAL_TYPE).orElse(null); + String description = OdcsMetadata.getString(md, OdcsMetadata.COMMENT).orElse(null); + Boolean primaryKey = OdcsMetadata.getBoolean(md, OdcsMetadata.PRIMARY_KEY).orElse(null); + Integer primaryKeyPosition = OdcsMetadata.getInt(md, OdcsMetadata.PRIMARY_KEY_POSITION).orElse(null); + String classification = OdcsMetadata.getString(md, OdcsMetadata.CLASSIFICATION).orElse(null); + Boolean critical = OdcsMetadata.getBoolean(md, OdcsMetadata.CRITICAL_DATA_ELEMENT).orElse(null); + List tags = OdcsMetadata.getTags(md); + if (tags.isEmpty()) { + tags = null; + } + + Boolean required = field.nullable() ? null : Boolean.TRUE; + + DataType dataType = field.dataType(); + + // Map via metadata / type + if (dataType instanceof MapType mapType + || (physicalType != null && PhysicalTypeParser.parseMap(physicalType).isPresent()) + || OdcsMetadata.getString(md, OdcsMetadata.MAP_KEY_TYPE).isPresent()) { + SchemaProperty mapProp = MapTypeCodec.fromMapType( + logicalName, + dataType instanceof MapType mt ? mt : DataTypes.createMapType(DataTypes.StringType, DataTypes.StringType), + field.nullable(), + description); + // Overlay metadata-sourced names / governance + return overlayGovernance(mapProp, physicalName, physicalType, primaryKey, primaryKeyPosition, + classification, critical, tags, md); + } + + if (dataType instanceof ArrayType arrayType) { + SchemaProperty items = toSchemaProperty(new StructField( + "items", + arrayType.elementType(), + arrayType.containsNull(), + Metadata.empty())); + if (!arrayType.containsNull()) { + items = copyWithRequired(items, true); + } + LogicalType lt = metaLogical.orElse(LogicalType.ARRAY); + return new SchemaProperty( + null, logicalName, physicalName, + physicalType != null ? physicalType : "array", + description, null, lt, null, + primaryKey, primaryKeyPosition, required, null, null, null, + classification, null, null, null, null, null, critical, + items, null, null, null, null, tags, null); + } + + if (dataType instanceof StructType structType) { + List nested = new ArrayList<>(); + for (StructField f : structType.fields()) { + nested.add(toSchemaProperty(f)); + } + LogicalType lt = metaLogical.orElse(LogicalType.OBJECT); + return new SchemaProperty( + null, logicalName, physicalName, + physicalType != null ? physicalType : "struct", + description, null, lt, null, + primaryKey, primaryKeyPosition, required, null, null, null, + classification, null, null, null, null, null, critical, + null, nested, null, null, null, tags, null); + } + + // Scalars: prefer metadata logical/physical, reconstruct options for formats / ntz + SchemaProperty inferred = inferProperty(field.name(), dataType, field.nullable(), description); + LogicalType lt = metaLogical.orElse(inferred.logicalType()); + String phys = physicalType != null ? physicalType : inferred.physicalType(); + LogicalTypeOptions opts = inferred.logicalTypeOptions(); + + // Preserve varchar maxLength from physical if present in metadata physicalType + if (phys != null) { + Optional varcharLen = PhysicalTypeParser.parseVarcharLength(phys); + if (varcharLen.isPresent()) { + opts = new LogicalTypeOptions( + opts == null ? null : opts.format(), + null, + varcharLen.get(), + null, null, null, null, null, null, + opts == null ? null : opts.timezone(), + null, null, null, null, null, null, null); + } + } + + return new SchemaProperty( + null, logicalName, physicalName, phys, description, null, lt, opts, + primaryKey, primaryKeyPosition, required, null, null, null, + classification, null, null, null, null, null, critical, + null, null, null, null, null, tags, null); + } + + private SchemaProperty overlayGovernance( + SchemaProperty base, + String physicalName, + String physicalType, + Boolean primaryKey, + Integer primaryKeyPosition, + String classification, + Boolean critical, + List tags, + Metadata md + ) { + // Merge map custom props from metadata if present + List custom = base.customProperties(); + Optional metaKey = OdcsMetadata.getString(md, OdcsMetadata.MAP_KEY_TYPE); + Optional metaVal = OdcsMetadata.getString(md, OdcsMetadata.MAP_VALUE_TYPE); + if (metaKey.isPresent() || metaVal.isPresent()) { + List merged = new ArrayList<>(); + merged.add(new io.github.dataspeclabs.odcs.core.model.v3.CustomProperty( + null, MapTypeCodec.MAP_KEY_TYPE, + metaKey.orElse(OdcsMetadata.customString(base, MapTypeCodec.MAP_KEY_TYPE).orElse("string")), + null)); + merged.add(new io.github.dataspeclabs.odcs.core.model.v3.CustomProperty( + null, MapTypeCodec.MAP_VALUE_TYPE, + metaVal.orElse(OdcsMetadata.customString(base, MapTypeCodec.MAP_VALUE_TYPE).orElse("string")), + null)); + OdcsMetadata.getString(md, OdcsMetadata.MAP_VALUE_REQUIRED) + .ifPresent(v -> merged.add(new io.github.dataspeclabs.odcs.core.model.v3.CustomProperty( + null, MapTypeCodec.MAP_VALUE_REQUIRED, v, null))); + custom = merged; + } + + return new SchemaProperty( + base.id(), + base.name(), + physicalName, + physicalType != null ? physicalType : base.physicalType(), + base.description(), + base.businessName(), + base.logicalType(), + base.logicalTypeOptions(), + primaryKey, + primaryKeyPosition, + base.required(), + base.unique(), + base.partitioned(), + base.partitionKeyPosition(), + classification, + base.encryptedName(), + base.transformSourceObjects(), + base.transformLogic(), + base.transformDescription(), + base.examples(), + critical, + base.items(), + base.properties(), + base.relationships(), + base.authoritativeDefinitions(), + base.quality(), + tags, + custom + ); + } + + private static LogicalTypeOptions optionsWithFormat(String format) { + return new LogicalTypeOptions( + format, null, null, null, null, null, null, null, null, + null, null, null, null, null, null, null, null); + } + + private static SchemaProperty property( + String name, + String physicalName, + String physicalType, + LogicalType logicalType, + LogicalTypeOptions options, + Boolean required, + String description, + SchemaProperty items, + List properties, + List custom + ) { + return new SchemaProperty( + null, name, physicalName, physicalType, description, null, + logicalType, options, null, null, required, null, null, null, + null, null, null, null, null, null, null, + items, properties, null, null, null, null, custom); + } + + private static SchemaProperty copyWithRequired(SchemaProperty p, Boolean required) { + return new SchemaProperty( + p.id(), p.name(), p.physicalName(), p.physicalType(), p.description(), p.businessName(), + p.logicalType(), p.logicalTypeOptions(), p.primaryKey(), p.primaryKeyPosition(), + required, p.unique(), p.partitioned(), p.partitionKeyPosition(), p.classification(), + p.encryptedName(), p.transformSourceObjects(), p.transformLogic(), p.transformDescription(), + p.examples(), p.criticalDataElement(), p.items(), p.properties(), p.relationships(), + p.authoritativeDefinitions(), p.quality(), p.tags(), p.customProperties()); + } +} diff --git a/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/TypeMappings.java b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/TypeMappings.java new file mode 100644 index 0000000..5cc42f5 --- /dev/null +++ b/odcs-spark/src/main/java/io/github/dataspeclabs/odcs/spark/schema/TypeMappings.java @@ -0,0 +1,244 @@ +package io.github.dataspeclabs.odcs.spark.schema; + +import io.github.dataspeclabs.odcs.core.model.v3.LogicalType; +import io.github.dataspeclabs.odcs.core.model.v3.LogicalTypeOptions; +import io.github.dataspeclabs.odcs.spark.ConversionReport; +import io.github.dataspeclabs.odcs.spark.SparkSchemaOptions; +import io.github.dataspeclabs.odcs.spark.TypeMappingException; +import org.apache.spark.sql.types.DataType; +import org.apache.spark.sql.types.DataTypes; + +import java.util.Locale; +import java.util.Optional; + +/** + * Canonical ODCS {@link LogicalType} ↔ Spark {@link DataType} defaults and format-driven refinements. + */ +public final class TypeMappings { + + private TypeMappings() { + } + + public static DataType fromLogicalType( + LogicalType logicalType, + LogicalTypeOptions options, + String physicalType, + SparkSchemaOptions schemaOptions, + String path, + ConversionReport.Builder report + ) { + if (logicalType == null) { + failOrWarn(schemaOptions, report, path, "missing logicalType", DataTypes.StringType); + return DataTypes.StringType; + } + + return switch (logicalType) { + case STRING -> mapString(options, physicalType); + case BOOLEAN -> DataTypes.BooleanType; + case INTEGER -> mapInteger(options, physicalType, path, report); + case NUMBER -> mapNumber(options, physicalType, schemaOptions, path, report); + case DATE -> DataTypes.DateType; + case TIMESTAMP -> mapTimestamp(options, physicalType); + case TIME -> mapTime(schemaOptions); + case ARRAY, OBJECT -> throw new TypeMappingException( + path, + "logicalType " + logicalType.value() + " must be handled by nested converters"); + }; + } + + private static DataType mapString(LogicalTypeOptions options, String physicalType) { + if (physicalType != null) { + String p = physicalType.trim().toLowerCase(Locale.ROOT); + if ("binary".equals(p) || "bytes".equals(p) || "bytea".equals(p)) { + return DataTypes.BinaryType; + } + } + if (options != null && options.format() != null) { + String format = options.format().trim().toLowerCase(Locale.ROOT); + if ("binary".equals(format) || "byte".equals(format)) { + return DataTypes.BinaryType; + } + } + return DataTypes.StringType; + } + + private static DataType mapInteger( + LogicalTypeOptions options, + String physicalType, + String path, + ConversionReport.Builder report + ) { + Optional fromPhysical = PhysicalTypeParser.parseScalarToken(physicalType, true); + if (fromPhysical.isPresent() && isIntegral(fromPhysical.get())) { + return fromPhysical.get(); + } + + String format = options == null ? null : options.format(); + if (format == null || format.isBlank()) { + return DataTypes.IntegerType; + } + return switch (format.trim().toLowerCase(Locale.ROOT)) { + case "i8" -> DataTypes.ByteType; + case "i16" -> DataTypes.ShortType; + case "i32" -> DataTypes.IntegerType; + case "i64" -> DataTypes.LongType; + case "u8" -> { + report.warn(path, "unsigned format u8 widened to ShortType"); + yield DataTypes.ShortType; + } + case "u16" -> { + report.warn(path, "unsigned format u16 widened to IntegerType"); + yield DataTypes.IntegerType; + } + case "u32" -> { + report.warn(path, "unsigned format u32 widened to LongType"); + yield DataTypes.LongType; + } + case "u64" -> { + report.warn(path, "unsigned format u64 mapped to DecimalType(20,0)"); + yield DataTypes.createDecimalType(20, 0); + } + case "i128", "u128" -> { + report.warn(path, "format " + format + " mapped to DecimalType(38,0)"); + yield DataTypes.createDecimalType(38, 0); + } + default -> DataTypes.IntegerType; + }; + } + + private static DataType mapNumber( + LogicalTypeOptions options, + String physicalType, + SparkSchemaOptions schemaOptions, + String path, + ConversionReport.Builder report + ) { + Optional decimal = PhysicalTypeParser.parseDecimal(physicalType); + if (decimal.isPresent()) { + PhysicalTypeParser.DecimalSpec d = decimal.get(); + if (d.isBare()) { + if (schemaOptions.requireDecimalPrecision()) { + failOrWarn(schemaOptions, report, path, + "decimal physicalType requires precision and scale, e.g. decimal(18,2)", + DataTypes.createDecimalType(10, 0)); + return DataTypes.createDecimalType(10, 0); + } + return DataTypes.createDecimalType(10, 0); + } + return DataTypes.createDecimalType(d.precision(), d.scale()); + } + + Optional fromPhysical = PhysicalTypeParser.parseScalarToken(physicalType, true); + if (fromPhysical.isPresent() && isFloating(fromPhysical.get())) { + return fromPhysical.get(); + } + + String format = options == null ? null : options.format(); + if (format != null) { + return switch (format.trim().toLowerCase(Locale.ROOT)) { + case "f32" -> DataTypes.FloatType; + case "f64" -> DataTypes.DoubleType; + default -> DataTypes.DoubleType; + }; + } + return DataTypes.DoubleType; + } + + static DataType mapTimestamp(LogicalTypeOptions options, String physicalType) { + if (physicalType != null) { + String p = physicalType.trim().toLowerCase(Locale.ROOT); + if ("timestamp_ntz".equals(p) || "timestampntz".equals(p)) { + return DataTypes.TimestampNTZType; + } + if ("timestamp".equals(p) || "datetime".equals(p)) { + return DataTypes.TimestampType; + } + } + if (isTimezoneAware(options)) { + return DataTypes.TimestampType; + } + if (isTimezoneNtz(options)) { + return DataTypes.TimestampNTZType; + } + return DataTypes.TimestampType; + } + + private static DataType mapTime(SparkSchemaOptions schemaOptions) { + return switch (schemaOptions.timeTypeMapping()) { + case STRING -> DataTypes.StringType; + case LONG_MICROS -> DataTypes.LongType; + }; + } + + /** + * Lenient parse of {@code logicalTypeOptions.timezone}: + * {@code "true"}/{@code "false"} decide awareness; any other non-blank value is a zone id (aware). + */ + public static boolean isTimezoneAware(LogicalTypeOptions options) { + if (options == null || options.timezone() == null || options.timezone().isBlank()) { + return false; + } + String tz = options.timezone().trim(); + if ("false".equalsIgnoreCase(tz)) { + return false; + } + return "true".equalsIgnoreCase(tz) || !tz.isBlank(); + } + + public static boolean isTimezoneNtz(LogicalTypeOptions options) { + if (options == null || options.timezone() == null) { + return false; + } + return "false".equalsIgnoreCase(options.timezone().trim()); + } + + public static String integerFormatFor(DataType dataType) { + if (DataTypes.ByteType.sameType(dataType)) { + return "i8"; + } + if (DataTypes.ShortType.sameType(dataType)) { + return "i16"; + } + if (DataTypes.IntegerType.sameType(dataType)) { + return "i32"; + } + if (DataTypes.LongType.sameType(dataType)) { + return "i64"; + } + return null; + } + + public static String numberFormatFor(DataType dataType) { + if (DataTypes.FloatType.sameType(dataType)) { + return "f32"; + } + if (DataTypes.DoubleType.sameType(dataType)) { + return "f64"; + } + return null; + } + + private static boolean isIntegral(DataType dt) { + return DataTypes.ByteType.sameType(dt) + || DataTypes.ShortType.sameType(dt) + || DataTypes.IntegerType.sameType(dt) + || DataTypes.LongType.sameType(dt); + } + + private static boolean isFloating(DataType dt) { + return DataTypes.FloatType.sameType(dt) || DataTypes.DoubleType.sameType(dt); + } + + private static void failOrWarn( + SparkSchemaOptions options, + ConversionReport.Builder report, + String path, + String message, + DataType fallback + ) { + if (options.isStrict()) { + throw new TypeMappingException(path, message); + } + report.warn(path, message + "; falling back to " + fallback.simpleString()); + } +} diff --git a/odcs-spark/src/test/resources/odcs/spark/all-data-types.odcs.yaml b/odcs-spark/src/test/resources/odcs/spark/all-data-types.odcs.yaml new file mode 100644 index 0000000..b5923d1 --- /dev/null +++ b/odcs-spark/src/test/resources/odcs/spark/all-data-types.odcs.yaml @@ -0,0 +1,84 @@ +apiVersion: v3.1.0 +kind: DataContract +id: 53581432-6c55-4ba2-a65f-72344a91553a +version: 1.0.0 +status: active +name: spark_all_data_types +schema: + - name: transactions + physicalType: table + logicalType: object + properties: + - name: account_id + logicalType: string + physicalType: string + required: true + logicalTypeOptions: + minLength: 11 + maxLength: 11 + pattern: ACC[0-9]{8} + - name: txn_ref_date + logicalType: date + physicalType: date + - name: txn_timestamp + logicalType: timestamp + physicalType: timestamp + - name: txn_timestamp_ntz + logicalType: timestamp + physicalType: timestamp_ntz + logicalTypeOptions: + timezone: "false" + - name: txn_time + logicalType: time + physicalType: time + - name: amount + logicalType: number + physicalType: decimal(18,2) + required: true + - name: rate + logicalType: number + physicalType: double + logicalTypeOptions: + format: f64 + - name: age + logicalType: integer + physicalType: bigint + logicalTypeOptions: + format: i64 + - name: is_open + logicalType: boolean + physicalType: boolean + - name: payload + logicalType: string + physicalType: binary + logicalTypeOptions: + format: binary + - name: street_lines + logicalType: array + physicalType: array + items: + logicalType: string + physicalType: string + required: true + - name: customer_details + logicalType: object + physicalType: struct + properties: + - name: num_children + logicalType: integer + physicalType: int + required: true + - name: date_of_birth + logicalType: date + physicalType: date + - name: attributes + logicalType: object + physicalType: map + customProperties: + - property: mapKeyType + value: string + - property: mapValueType + value: bigint + - name: tags_map + logicalType: object + physicalType: map diff --git a/pom.xml b/pom.xml index 51dcd89..5bfc8ac 100644 --- a/pom.xml +++ b/pom.xml @@ -50,6 +50,7 @@ odcs-core + odcs-spark @@ -58,6 +59,8 @@ 2.22.1 6.1.2 2.0.4 + 3.5.8 + 2.13 3.14.0 3.5.3 data-spec-labs