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