Spark schema <-> odcs conversion

odcs-spark/pom.xml

@@ -0,0 +1,38 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"
+         xmlns="http://maven.apache.org/POM/4.0.0">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>io.github.data-spec-labs</groupId>
+        <artifactId>odcs-java-sdk</artifactId>
+        <version>0.1.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>odcs-spark</artifactId>
+    <name>ODCS Java SDK - Spark Integration</name>
+    <packaging>jar</packaging>
+    <description>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.</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.github.data-spec-labs</groupId>
+            <artifactId>odcs-core</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-sql_${scala.binary.version}</artifactId>
+            <version>${spark.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

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<String> warnings;
+
+    private ConversionReport(List<String> 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<String> warnings() {
+        return warnings;
+    }
+
+    public boolean hasWarnings() {
+        return !warnings.isEmpty();
+    }
+
+    public static final class Builder {
+        private final List<String> 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);
+        }
+    }
+}

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 <T> converted value type
+ */
+public record ConversionResult<T>(T value, ConversionReport report) {
+
+    public ConversionResult {
+        Objects.requireNonNull(value, "value");
+        Objects.requireNonNull(report, "report");
+    }
+
+    public static <T> ConversionResult<T> of(T value, ConversionReport report) {
+        return new ConversionResult<>(value, report);
+    }
+
+    public boolean hasWarnings() {
+        return report.hasWarnings();
+    }
+}

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}.
+ *
+ * <p>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<StructType> convertWithReport(SchemaObject object) {
+        ConversionReport.Builder report = ConversionReport.builder();
+        StructType st = new OdcsToSparkConverter(options, report).toStructType(object);
+        return ConversionResult.of(st, report.build());
+    }
+
+    public ConversionResult<StructType> 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<SchemaObject> 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());
+    }
+}

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);
+        }
+    }
+}