Connection Registry: 11 External Data Source Types

“

Series: S4 Ontology Modeling · Article 13 | Level: Intermediate | Reading Time: 18 min

#TL;DR

ConnectionRegistry is coomia-dip's unified data source integration layer, supporting 11 external data source types (relational databases, NoSQL, object storage, message queues, APIs) through declarative configuration for "register once, use everywhere."
Credential management and connection testing are the two pillars of production reliability — the platform provides built-in encrypted credential storage, periodic rotation, and connection health monitoring, eliminating the operational nightmare of scattered connection configurations.
Source Mapping maps external data structures to Ontology ObjectTypes, allowing existing systems to join the Ontology semantic ecosystem without migrating data.

#1. Introduction: A Unified Gateway to Data Silos

In enterprise digital transformation, a typical mid-size organization maintains 15-30 independent data sources: MySQL production databases, PostgreSQL analytics databases, MongoDB log stores, S3 object storage, Kafka message queues, third-party REST APIs...

The traditional approach requires writing separate connection code, authentication logic, and health checks for each data source. This leads to:

Connection configurations scattered across code repositories, config files, and environment variables
Chaotic credential management with passwords appearing in plaintext in config files
Lack of unified alerting and self-healing when connections fail
New data sources requiring development of new adapter code

Code

Traditional Approach:                    ConnectionRegistry Approach:

App A → MySQL config (hardcoded)        ┌─────────────────────┐
App B → MySQL config (another copy)     │  ConnectionRegistry  │
App C → PostgreSQL config               │                     │
App D → MongoDB config                  │  Unified registry   │
App E → S3 config                       │  Unified auth       │
App F → Kafka config                    │  Unified monitoring │
                                        │  Unified mapping    │
6 configs, 6 auth systems, 6 monitors  └─────────────────────┘
                                               │
                                        1 config, 1 auth, 1 monitor

ConnectionRegistry is coomia-dip's answer to this problem.

#2. ConnectionRegistry Core Concepts

#2.1 Architectural Position

ConnectionRegistry resides in the Control Layer (Control Layer) and serves as the "data bridge" for SchemaRegistry:

Code

┌─────────────────────────────────────────────────────┐
│                   Control Layer (B)                  │
│                                                     │
│  ┌──────────────┐      ┌──────────────────────┐    │
│  │SchemaRegistry │◄────►│ ConnectionRegistry   │    │
│  │              │      │                      │    │
│  │ ObjectType   │      │ Connection configs   │    │
│  │ RelationType │      │ Credential vault     │    │
│  │ ActionType   │      │ Health monitor       │    │
│  │ SourceMapping│      │ Source mapping        │    │
│  └──────────────┘      └──────────┬───────────┘    │
│                                    │                │
└────────────────────────────────────┼────────────────┘
                                     │ gRPC
                    ┌────────────────┼────────────────┐
                    │                │                │
               ┌────▼───┐     ┌─────▼────┐    ┌─────▼────┐
               │ MySQL   │     │PostgreSQL│    │ MongoDB  │
               │ Oracle  │     │ Doris    │    │ Redis    │
               │ SQL Svr │     │          │    │ ES       │
               └─────────┘     └──────────┘    └──────────┘

#2.2 Connection Data Model

A Connection comprises four parts:

YAML

apiVersion: ontology/v1
kind: Connection
metadata:
  name: manufacturing-mysql
  namespace: factory-alpha
  labels:
    environment: production
    team: data-engineering
spec:
  # 1. Connection type
  type: MYSQL

  # 2. Connection parameters
  config:
    host: db-prod-01.internal.company.com
    port: 3306
    database: manufacturing
    charset: utf8mb4
    maxPoolSize: 20
    connectionTimeout: 5000

  # 3. Credential reference (never stores passwords directly)
  credentialRef:
    name: mysql-prod-credentials
    vault: platform-vault

  # 4. Health check configuration
  healthCheck:
    enabled: true
    interval: 60s
    timeout: 5s
    query: "SELECT 1"

status:
  connected: true
  lastChecked: "2026-03-24T10:30:00Z"
  latency: 12ms
  version: "MySQL 8.0.35"

#2.3 Connection Lifecycle

Code

┌─────────┐   register   ┌──────────┐   test    ┌──────────┐
│ PENDING  ├─────────────►│ TESTING  ├──────────►│  ACTIVE  │
└─────────┘              └────┬─────┘           └────┬─────┘
                              │ fail                  │
                              ▼                  disconnect
                         ┌──────────┐                 │
                         │  FAILED  │                 ▼
                         └──────────┘           ┌──────────┐
                                                │ INACTIVE │
                                                └────┬─────┘
                                                     │ reconnect
                                                     ▼
                                                ┌──────────┐
                                                │  ACTIVE  │
                                                └──────────┘

#3. 11 Supported Data Source Types

#3.1 Relational Databases (4 Types)

Data Source	Type ID	Default Port	Driver
MySQL	`MYSQL`	3306	mysql-connector-j 8.x
PostgreSQL	`POSTGRESQL`	5432	postgresql 42.x
Oracle	`ORACLE`	1521	ojdbc11
SQL Server	`SQLSERVER`	1433	mssql-jdbc 12.x

Relational database connection configuration examples:

Python

from ontology_sdk import OntologyClient, ConnectionSpec

client = OntologyClient(base_url="http://control-Layer:8080")

# MySQL connection
mysql_conn = ConnectionSpec(
    name="erp-mysql",
    type="MYSQL",
    config={
        "host": "mysql-prod.internal",
        "port": 3306,
        "database": "erp_production",
        "charset": "utf8mb4",
        "maxPoolSize": 20,
        "connectionTimeout": 5000,
        "ssl": True,
        "sslMode": "VERIFY_IDENTITY",
    },
    credential_ref="erp-mysql-cred",
)

result = client.connection.register(mysql_conn)
print(f"Registered: {result.name}, status: {result.status}")

Python

# PostgreSQL connection (with schema specification)
pg_conn = ConnectionSpec(
    name="analytics-pg",
    type="POSTGRESQL",
    config={
        "host": "pg-analytics.internal",
        "port": 5432,
        "database": "analytics",
        "schema": "public",
        "maxPoolSize": 30,
        "statementTimeout": 30000,
        "ssl": True,
    },
    credential_ref="pg-analytics-cred",
)

# Oracle connection (Service Name mode)
oracle_conn = ConnectionSpec(
    name="legacy-oracle",
    type="ORACLE",
    config={
        "host": "oracle-prod.internal",
        "port": 1521,
        "serviceName": "ORCL",
        "maxPoolSize": 10,
        "connectionTimeout": 10000,
    },
    credential_ref="oracle-prod-cred",
)

# SQL Server connection
sqlserver_conn = ConnectionSpec(
    name="hr-sqlserver",
    type="SQLSERVER",
    config={
        "host": "sqlsvr-hr.internal",
        "port": 1433,
        "database": "HumanResources",
        "instanceName": "MSSQLSERVER",
        "encrypt": True,
        "trustServerCertificate": False,
    },
    credential_ref="sqlsvr-hr-cred",
)

#3.2 NoSQL Databases (3 Types)

Data Source	Type ID	Protocol	Characteristics
MongoDB	`MONGODB`	mongodb://	Document store, auto schema inference
Redis	`REDIS`	redis://	Key-value store, cache mapping
Elasticsearch	`ELASTICSEARCH`	HTTP/HTTPS	Full-text search, search index mapping

Python

# MongoDB connection (replica set mode)
mongo_conn = ConnectionSpec(
    name="iot-mongodb",
    type="MONGODB",
    config={
        "hosts": [
            "mongo-01.internal:27017",
            "mongo-02.internal:27017",
            "mongo-03.internal:27017",
        ],
        "database": "iot_events",
        "replicaSet": "rs0",
        "authSource": "admin",
        "readPreference": "secondaryPreferred",
        "maxPoolSize": 50,
    },
    credential_ref="mongo-iot-cred",
)

# Elasticsearch connection
es_conn = ConnectionSpec(
    name="log-elasticsearch",
    type="ELASTICSEARCH",
    config={
        "hosts": [
            "https://es-01.internal:9200",
            "https://es-02.internal:9200",
        ],
        "indexPrefix": "app-logs-",
        "numberOfShards": 5,
        "numberOfReplicas": 1,
    },
    credential_ref="es-log-cred",
)

#3.3 Object Storage (1 Type)

Data Source	Type ID	Protocol	Supported File Formats
S3/MinIO	`S3`	S3 API	Parquet, CSV, JSON, Avro, ORC

Python

# S3 / MinIO connection
s3_conn = ConnectionSpec(
    name="data-lake-s3",
    type="S3",
    config={
        "endpoint": "https://s3.amazonaws.com",
        "region": "us-east-1",
        "bucket": "company-data-lake",
        "pathPrefix": "raw/",
        "fileFormat": "PARQUET",
        "compressionCodec": "SNAPPY",
    },
    credential_ref="s3-datalake-cred",
)

#3.4 Message Queues (1 Type)

Data Source	Type ID	Protocol	Use Case
Kafka	`KAFKA`	Kafka Protocol	Streaming data ingestion, CDC

Python

# Kafka connection
kafka_conn = ConnectionSpec(
    name="cdc-kafka",
    type="KAFKA",
    config={
        "bootstrapServers": "kafka-01:9092,kafka-02:9092,kafka-03:9092",
        "groupId": "coomia-dip-consumer",
        "autoOffsetReset": "earliest",
        "securityProtocol": "SASL_SSL",
        "saslMechanism": "SCRAM-SHA-256",
        "schemaRegistryUrl": "http://schema-registry:8081",
    },
    credential_ref="kafka-cdc-cred",
)

#3.5 External APIs (2 Types)

Data Source	Type ID	Protocol	Auth Methods
REST API	`REST_API`	HTTP/HTTPS	Bearer Token, API Key, OAuth2
gRPC Service	`GRPC_SERVICE`	gRPC	mTLS, Token

Python

# REST API connection
rest_conn = ConnectionSpec(
    name="weather-api",
    type="REST_API",
    config={
        "baseUrl": "https://api.weather.example.com/v2",
        "authType": "BEARER_TOKEN",
        "timeout": 10000,
        "retryPolicy": {
            "maxRetries": 3,
            "backoffMs": 1000,
        },
        "rateLimiting": {
            "requestsPerSecond": 10,
            "burstSize": 20,
        },
        "headers": {
            "Accept": "application/json",
        },
    },
    credential_ref="weather-api-cred",
)

# gRPC Service connection
grpc_conn = ConnectionSpec(
    name="pricing-service",
    type="GRPC_SERVICE",
    config={
        "host": "pricing-service.internal",
        "port": 9090,
        "useTls": True,
        "protoPackage": "com.company.pricing.v1",
        "serviceName": "PricingService",
        "loadBalancingPolicy": "round_robin",
    },
    credential_ref="pricing-grpc-cred",
)

#4. Connection Testing Mechanism

#4.1 Registration-Time Testing

Each data source type has built-in connection testing logic:

Python

# Register and test connection
result = client.connection.register_and_test(mysql_conn)

print(f"Connection: {result.name}")
print(f"Status: {result.status}")          # ACTIVE or FAILED
print(f"Latency: {result.latency_ms}ms")   # Initial connection latency
print(f"Version: {result.server_version}") # Server version
print(f"Features: {result.features}")      # Supported features list

# Test result details
if result.status == "FAILED":
    print(f"Error: {result.error_message}")
    print(f"Error Code: {result.error_code}")
    print(f"Suggestion: {result.suggestion}")

Test methods by data source type:

Code

Source Type          Test Method                    Checks
──────────────────────────────────────────────────────────
MYSQL              SELECT 1                       Connectivity, auth, perms
POSTGRESQL         SELECT 1                       Connectivity, schema exists
ORACLE             SELECT 1 FROM DUAL             Connectivity, service name
SQLSERVER          SELECT 1                       Connectivity, instance name
MONGODB            db.runCommand({ping:1})         Connectivity, replica status
REDIS              PING                           Connectivity, auth
ELASTICSEARCH      GET /_cluster/health           Cluster status, index access
S3                 HeadBucket                     Bucket existence, permissions
KAFKA              listTopics()                   Broker connectivity
REST_API           GET /health                    Endpoint reachability
GRPC_SERVICE       grpc.health.v1.Check           Service health status

#4.2 Runtime Health Checks

ConnectionRegistry continuously monitors the health of all active connections:

Python

# Configure health check policy
health_config = HealthCheckConfig(
    enabled=True,
    interval=60,              # Check every 60 seconds
    timeout=5,                # 5-second timeout marks failure
    failure_threshold=3,      # 3 consecutive failures marks INACTIVE
    success_threshold=1,      # 1 success recovers to ACTIVE
    alert_channels=["slack-ops", "pagerduty"],
)

client.connection.update_health_check("erp-mysql", health_config)

Health check state machine:

Code

                    success
              ┌──────────────┐
              │              │
              ▼              │
┌──────────┐    ┌──────────┐
│  HEALTHY  │    │DEGRADED  │
└─────┬────┘    └────┬─────┘
      │ fail          │ fail (threshold)
      ▼               ▼
┌──────────┐    ┌──────────┐
│ CHECKING  │    │UNHEALTHY │
└──────────┘    └────┬─────┘
                     │ auto-reconnect
                     ▼
               ┌──────────┐
               │RECOVERING│
               └──────────┘

#4.3 Connection Pool Management

Each data source connection is equipped with a connection pool to avoid frequent creation and destruction:

Python

# View connection pool status
pool_status = client.connection.get_pool_status("erp-mysql")

print(f"Active connections: {pool_status.active}")
print(f"Idle connections: {pool_status.idle}")
print(f"Waiting requests: {pool_status.waiting}")
print(f"Total created: {pool_status.total_created}")
print(f"Total destroyed: {pool_status.total_destroyed}")
print(f"Avg acquire time: {pool_status.avg_acquire_ms}ms")

#5. Credential Management

#5.1 Credential Storage Architecture

ConnectionRegistry never stores passwords directly. All credentials are managed through the CredentialVault:

Code

┌─────────────────────────────────────────────┐
│              ConnectionRegistry              │
│                                             │
│  Connection A ──credentialRef──┐            │
│  Connection B ──credentialRef──┤            │
│  Connection C ──credentialRef──┤            │
│                                │            │
│                    ┌───────────▼──────────┐ │
│                    │   CredentialVault     │ │
│                    │                      │ │
│                    │  AES-256-GCM encrypt │ │
│                    │  Master Key: HSM/KMS │ │
│                    │  Audit Log: complete │ │
│                    └──────────────────────┘ │
└─────────────────────────────────────────────┘

#5.2 Credential Types

Python

from ontology_sdk import CredentialSpec

# Username/password credential
basic_cred = CredentialSpec(
    name="mysql-prod-cred",
    type="USERNAME_PASSWORD",
    data={
        "username": "onto_reader",
        "password": "encrypted:vault:xxxxx",
    },
    rotation_policy={
        "enabled": True,
        "interval_days": 90,
        "notify_before_days": 14,
    },
)

# API Key credential
apikey_cred = CredentialSpec(
    name="weather-api-cred",
    type="API_KEY",
    data={
        "apiKey": "encrypted:vault:xxxxx",
        "headerName": "X-API-Key",
    },
)

# OAuth2 credential
oauth_cred = CredentialSpec(
    name="salesforce-oauth-cred",
    type="OAUTH2",
    data={
        "clientId": "encrypted:vault:xxxxx",
        "clientSecret": "encrypted:vault:xxxxx",
        "tokenUrl": "https://login.salesforce.com/services/oauth2/token",
        "scope": "api refresh_token",
    },
)

# mTLS credential
mtls_cred = CredentialSpec(
    name="pricing-grpc-cred",
    type="MTLS",
    data={
        "certPath": "/certs/client.pem",
        "keyPath": "/certs/client-key.pem",
        "caPath": "/certs/ca.pem",
    },
)

# AWS IAM Role credential
iam_cred = CredentialSpec(
    name="s3-datalake-cred",
    type="AWS_IAM_ROLE",
    data={
        "roleArn": "arn:aws:iam::123456789:role/coomia-dip-s3-reader",
        "externalId": "coomia-dip-prod",
        "sessionDuration": 3600,
    },
)

#5.3 Credential Rotation

In production environments, credentials must be rotated periodically:

Python

# Manually trigger credential rotation
rotation_result = client.credential.rotate("mysql-prod-cred")

print(f"Old credential expired: {rotation_result.old_expired_at}")
print(f"New credential active: {rotation_result.new_active_at}")
print(f"Affected connections: {rotation_result.affected_connections}")

# View rotation history
history = client.credential.rotation_history("mysql-prod-cred")
for entry in history:
    print(f"  {entry.rotated_at} by {entry.rotated_by} - {entry.status}")

Automatic rotation flow:

Code

Day 76/90                     Day 90/90
(expiry notification)          (auto-rotation)

┌──────────────┐   ┌──────────────┐   ┌──────────────┐   ┌──────────────┐
│Generate new  │──►│Dual cred     │──►│Verify new    │──►│Retire old    │
│credential    │   │coexistence   │   │cred works    │   │Record audit  │
└──────────────┘   └──────────────┘   └──────────────┘   └──────────────┘

#5.4 Credential Auditing

All credential operations are recorded in audit logs:

Python

# Query credential audit logs
audit_logs = client.credential.audit_log(
    credential_name="mysql-prod-cred",
    since="2026-03-01T00:00:00Z",
)

for log in audit_logs:
    print(f"{log.timestamp} | {log.action} | {log.actor} | {log.result}")

# Example output:
# 2026-03-15 10:30:00 | READ   | service:data-pipeline | SUCCESS
# 2026-03-15 14:22:00 | ROTATE | user:admin@company    | SUCCESS
# 2026-03-16 09:00:00 | READ   | service:schema-sync   | SUCCESS

#6. Source Mapping

#6.1 From External Tables to ObjectTypes

Source Mapping is one of ConnectionRegistry's most critical features — mapping external database table structures to Ontology ObjectTypes:

Python

# Auto-discover external database table structures
discovery = client.connection.discover_schema("erp-mysql")

for table in discovery.tables:
    print(f"Table: {table.name}")
    print(f"  Columns: {len(table.columns)}")
    print(f"  Primary Key: {table.primary_key}")
    print(f"  Foreign Keys: {table.foreign_keys}")
    print(f"  Row Count: {table.estimated_row_count}")
    print()

#6.2 Mapping Configuration

YAML

apiVersion: ontology/v1
kind: SourceMapping
metadata:
  name: erp-equipment-mapping
spec:
  connection: erp-mysql
  source:
    table: equipment
    schema: manufacturing
  target:
    objectType: Equipment
  propertyMappings:
    - source: equip_id
      target: equipmentId
      primaryKey: true
    - source: equip_name
      target: name
    - source: equip_status
      target: status
      transform: "UPPER(value)"
    - source: install_date
      target: installDate
      transform: "CAST(value AS TIMESTAMP)"
    - source: line_id
      target: productionLineId
      foreignKeyMapping:
        relation: BelongsToLine
        targetType: ProductionLine
        targetProperty: lineId
  syncPolicy:
    mode: INCREMENTAL
    schedule: "*/5 * * * *"
    watermarkColumn: updated_at
    batchSize: 1000

#6.3 Mapping Modes

coomia-dip supports three mapping modes:

Code

1. VIRTUAL (Virtual Mapping)
   ┌──────────┐     query time    ┌──────────┐
   │Ontology  │────────────────►│External DB│
   │ObjectType│    live forward   │  source   │
   └──────────┘                  └──────────┘
   Traits: Zero delay, no data migration, depends on source availability

2. REPLICATED (Replicated Mapping)
   ┌──────────┐     periodic sync ┌──────────┐
   │Ontology  │◄───────────────│External DB│
   │Iceberg   │    incremental    │  source   │
   └──────────┘                  └──────────┘
   Traits: High-perf queries, sync delay, independent availability

3. FEDERATED (Federated Mapping)
   ┌──────────┐     query routing ┌──────────┐
   │Ontology  │────────────────►│External DB│
   │query eng │    with cache     │  source   │
   └──────────┘                  └──────────┘
   Traits: Query optimization, smart caching, balance perf & freshness

Python

# Create virtual mapping
virtual_mapping = SourceMappingSpec(
    name="erp-equipment-virtual",
    connection="erp-mysql",
    source_table="equipment",
    target_object_type="Equipment",
    mode="VIRTUAL",
    property_mappings=[
        PropertyMapping("equip_id", "equipmentId", primary_key=True),
        PropertyMapping("equip_name", "name"),
        PropertyMapping("equip_status", "status", transform="UPPER(value)"),
    ],
)

# Create replicated mapping
replicated_mapping = SourceMappingSpec(
    name="erp-equipment-replicated",
    connection="erp-mysql",
    source_table="equipment",
    target_object_type="Equipment",
    mode="REPLICATED",
    sync_policy=SyncPolicy(
        schedule="*/5 * * * *",
        watermark_column="updated_at",
        batch_size=1000,
        conflict_resolution="LATEST_WINS",
    ),
    property_mappings=[
        PropertyMapping("equip_id", "equipmentId", primary_key=True),
        PropertyMapping("equip_name", "name"),
    ],
)

client.connection.create_source_mapping(virtual_mapping)

#6.4 Automatic Schema Inference

For schema-less data sources like MongoDB, ConnectionRegistry provides automatic schema inference:

Python

# Infer schema from MongoDB collection
inferred_schema = client.connection.infer_schema(
    connection="iot-mongodb",
    collection="sensor_readings",
    sample_size=10000,    # Sample 10,000 documents
    confidence=0.95,      # 95% confidence level
)

print(f"Inferred properties: {len(inferred_schema.properties)}")
for prop in inferred_schema.properties:
    print(f"  {prop.name}: {prop.type} "
          f"(nullable: {prop.nullable}, "
          f"coverage: {prop.coverage:.1%})")

# Example output:
# Inferred properties: 8
#   sensorId: STRING (nullable: False, coverage: 100.0%)
#   timestamp: TIMESTAMP (nullable: False, coverage: 100.0%)
#   temperature: DOUBLE (nullable: True, coverage: 98.5%)
#   humidity: DOUBLE (nullable: True, coverage: 97.2%)
#   location: STRUCT (nullable: True, coverage: 85.3%)

#7. ConnectionRegistry gRPC API

ConnectionRegistry exposes its services via gRPC. Here are the core Protobuf definitions:

PROTOBUF

syntax = "proto3";
package onto.control.connection.v1;

service ConnectionRegistryService {
  // Connection management
  rpc RegisterConnection(RegisterConnectionRequest)
      returns (RegisterConnectionResponse);
  rpc TestConnection(TestConnectionRequest)
      returns (TestConnectionResponse);
  rpc UpdateConnection(UpdateConnectionRequest)
      returns (UpdateConnectionResponse);
  rpc DeleteConnection(DeleteConnectionRequest)
      returns (DeleteConnectionResponse);
  rpc ListConnections(ListConnectionsRequest)
      returns (ListConnectionsResponse);
  rpc GetConnectionStatus(GetConnectionStatusRequest)
      returns (GetConnectionStatusResponse);

  // Schema discovery
  rpc DiscoverSchema(DiscoverSchemaRequest)
      returns (DiscoverSchemaResponse);
  rpc InferSchema(InferSchemaRequest)
      returns (InferSchemaResponse);

  // Source mapping
  rpc CreateSourceMapping(CreateSourceMappingRequest)
      returns (CreateSourceMappingResponse);
  rpc UpdateSourceMapping(UpdateSourceMappingRequest)
      returns (UpdateSourceMappingResponse);
  rpc DeleteSourceMapping(DeleteSourceMappingRequest)
      returns (DeleteSourceMappingResponse);
  rpc SyncSourceMapping(SyncSourceMappingRequest)
      returns (SyncSourceMappingResponse);

  // Credential management
  rpc StoreCredential(StoreCredentialRequest)
      returns (StoreCredentialResponse);
  rpc RotateCredential(RotateCredentialRequest)
      returns (RotateCredentialResponse);
  rpc GetCredentialAuditLog(GetCredentialAuditLogRequest)
      returns (GetCredentialAuditLogResponse);

  // Health check
  rpc GetHealthStatus(GetHealthStatusRequest)
      returns (stream HealthStatusEvent);
}

message RegisterConnectionRequest {
  string name = 1;
  string namespace = 2;
  ConnectionType type = 3;
  map<string, string> config = 4;
  string credential_ref = 5;
  HealthCheckConfig health_check = 6;
}

enum ConnectionType {
  CONNECTION_TYPE_UNSPECIFIED = 0;
  MYSQL = 1;
  POSTGRESQL = 2;
  ORACLE = 3;
  SQLSERVER = 4;
  MONGODB = 5;
  REDIS = 6;
  ELASTICSEARCH = 7;
  S3 = 8;
  KAFKA = 9;
  REST_API = 10;
  GRPC_SERVICE = 11;
}

#8. Multi-Connection Orchestration and Query Routing

#8.1 Cross-Data-Source Queries

When Ontology ObjectTypes are mapped to different data sources, the platform automatically handles cross-source queries:

Python

# Equipment from MySQL, SensorReading from MongoDB,
# WorkOrder from PostgreSQL
# Users don't need to know where data lives

result = client.ontology.query(
    object_type="Equipment",
    filter="status == 'RUNNING'",
    expand=[
        "sensors.latestReading",   # MongoDB
        "workOrders.openCount",    # PostgreSQL
    ],
)

# Platform automatically:
# 1. Queries Equipment from MySQL
# 2. Queries SensorReading from MongoDB
# 3. Queries WorkOrder from PostgreSQL
# 4. JOINs in memory and returns unified results

#8.2 Query Routing Strategy

Code

┌──────────────────────────────────────────────────────┐
│                   Query Router                        │
│                                                      │
│  1. Parse ObjectTypes involved in query               │
│  2. Look up SourceMapping for each ObjectType         │
│  3. Select query path based on mapping mode           │
│     - VIRTUAL  → query external source directly       │
│     - REPLICATED → query local Iceberg                │
│     - FEDERATED → check cache → fall back to source   │
│  4. Execute multi-source queries in parallel          │
│  5. Merge results and apply Ontology semantics        │
│                                                      │
│  Optimizations:                                      │
│  - Predicate pushdown                                │
│  - Projection pushdown                               │
│  - Join reordering                                   │
│  - Result caching                                    │
└──────────────────────────────────────────────────────┘

#9. Operations and Monitoring

#9.1 Connection Dashboard

Python

# Get overview of all connections
dashboard = client.connection.dashboard()

print(f"Total connections: {dashboard.total}")
print(f"Active: {dashboard.active}")
print(f"Inactive: {dashboard.inactive}")
print(f"Failed: {dashboard.failed}")

# Group by type
for type_group in dashboard.by_type:
    print(f"\n{type_group.type}:")
    print(f"  Count: {type_group.count}")
    print(f"  Avg latency: {type_group.avg_latency_ms}ms")
    print(f"  Error rate: {type_group.error_rate:.2%}")

#9.2 Connection Alert Rules

Python

# Configure alert rules
alert_rule = AlertRule(
    name="connection-high-latency",
    condition="connection.latency_ms > 500",
    duration="5m",
    severity="WARNING",
    channels=["slack-ops"],
    message="Connection {connection.name} latency exceeded 500ms",
)

client.connection.create_alert_rule(alert_rule)

# View alert history
alerts = client.connection.get_alerts(
    since="2026-03-20T00:00:00Z",
    severity="WARNING",
)
for alert in alerts:
    print(f"{alert.fired_at} | {alert.connection} | {alert.message}")

#9.3 Connection Migration

When you need to switch data sources (e.g., migrating from MySQL to PostgreSQL):

Python

# Create migration plan
migration = client.connection.create_migration(
    source_connection="erp-mysql",
    target_connection="erp-postgresql",
    strategy="BLUE_GREEN",    # Blue-green switch
    validation_queries=[
        "SELECT COUNT(*) FROM equipment",
        "SELECT MAX(updated_at) FROM equipment",
    ],
)

# Execute migration
migration.execute()

# Validate
validation = migration.validate()
print(f"Data consistency: {validation.consistency_check}")
print(f"Row count match: {validation.row_count_match}")
print(f"Latency comparison: {validation.latency_comparison}")

# Switch over
migration.switch_over()

#10. Security Best Practices

#10.1 Network Layer Security

Code

┌─────────────────────────────────────────────────┐
│                  coomia-dip VPC                   │
│                                                 │
│  ┌──────────────┐      ┌──────────────────┐    │
│  │Control Layer │      │ Connection Pool   │    │
│  │              │─────►│                  │    │
│  │Connection    │      │ TLS 1.3          │    │
│  │Registry      │      │ Connection limit │    │
│  └──────────────┘      │ IP whitelist     │    │
│                        └────────┬─────────┘    │
│                                 │              │
└─────────────────────────────────┼──────────────┘
                                  │ Encrypted
                           ┌──────▼──────┐
                           │ External DB  │
                           │ (DMZ/VPN)    │
                           └─────────────┘

#10.2 Principle of Least Privilege

Python

# Configure minimum permissions for a connection
permission_config = ConnectionPermission(
    connection="erp-mysql",
    allowed_operations=["SELECT"],          # Read-only
    allowed_tables=["equipment", "orders"], # Specific tables only
    row_filter="factory_id = 'F001'",       # Row-level filtering
    column_mask={
        "employee_ssn": "MASK_LAST_4",     # Column-level masking
        "salary": "REDACT",
    },
)

client.connection.set_permissions(permission_config)

#10.3 Audit Compliance

All connection operations are recorded in complete audit trails:

Code

Audit Event                      Recorded Content
────────────────────────────────────────────
CONNECTION_REGISTERED            Who registered, when, config snapshot
CONNECTION_TESTED               Test result, latency
CREDENTIAL_ACCESSED             Which service read the credential
CREDENTIAL_ROTATED              Pre/post rotation metadata
SCHEMA_DISCOVERED               Which tables were discovered
SOURCE_MAPPING_CREATED          Mapping configuration details
QUERY_ROUTED                    Which connection the query was routed to
CONNECTION_FAILED               Failure reason, impact scope

#11. Practical Exercise: Building a Multi-Source Ontology from Scratch

Here is a complete hands-on example — integrating three data sources into a unified Ontology:

Python

from ontology_sdk import OntologyClient

client = OntologyClient(base_url="http://control-Layer:8080")

# ============================================
# Step 1: Register data sources
# ============================================

# ERP system (MySQL)
client.connection.register(ConnectionSpec(
    name="erp-mysql",
    type="MYSQL",
    config={"host": "mysql.internal", "port": 3306, "database": "erp"},
    credential_ref="erp-cred",
))

# IoT platform (MongoDB)
client.connection.register(ConnectionSpec(
    name="iot-mongo",
    type="MONGODB",
    config={"hosts": ["mongo.internal:27017"], "database": "iot"},
    credential_ref="iot-cred",
))

# Data lake (S3)
client.connection.register(ConnectionSpec(
    name="datalake-s3",
    type="S3",
    config={"endpoint": "https://s3.internal", "bucket": "datalake"},
    credential_ref="s3-cred",
))

# ============================================
# Step 2: Test all connections
# ============================================

connections = client.connection.list()
for conn in connections:
    result = client.connection.test(conn.name)
    print(f"{conn.name}: {result.status} ({result.latency_ms}ms)")

# ============================================
# Step 3: Discover schema and create mappings
# ============================================

# Discover and map Equipment from MySQL
mysql_schema = client.connection.discover_schema("erp-mysql")
equipment_table = mysql_schema.get_table("equipment")

client.connection.create_source_mapping(SourceMappingSpec(
    name="equipment-mapping",
    connection="erp-mysql",
    source_table="equipment",
    target_object_type="Equipment",
    mode="REPLICATED",
    sync_policy=SyncPolicy(schedule="*/5 * * * *"),
    property_mappings=equipment_table.auto_map(),
))

# Infer and map SensorReading from MongoDB
mongo_schema = client.connection.infer_schema("iot-mongo", "sensor_readings")

client.connection.create_source_mapping(SourceMappingSpec(
    name="sensor-mapping",
    connection="iot-mongo",
    source_table="sensor_readings",
    target_object_type="SensorReading",
    mode="VIRTUAL",
    property_mappings=mongo_schema.auto_map(),
))

# Map historical analytics data from S3
client.connection.create_source_mapping(SourceMappingSpec(
    name="history-mapping",
    connection="datalake-s3",
    source_table="analytics/equipment_history/",
    target_object_type="EquipmentHistory",
    mode="FEDERATED",
    property_mappings=[
        PropertyMapping("equip_id", "equipmentId"),
        PropertyMapping("event_time", "eventTime"),
        PropertyMapping("event_type", "eventType"),
        PropertyMapping("details", "details"),
    ],
))

# ============================================
# Step 4: Verify unified queries
# ============================================

# Now you can query across three data sources
result = client.ontology.query(
    object_type="Equipment",
    filter="status == 'RUNNING'",
    expand=[
        "sensorReadings.latest",    # From MongoDB
        "history.last30Days",       # From S3
    ],
)

for equipment in result.objects:
    print(f"Equipment: {equipment.name}")
    print(f"  Latest sensor: {equipment.sensorReadings.latest}")
    print(f"  History events: {len(equipment.history.last30Days)}")

#Key Takeaways

ConnectionRegistry is the "unified registry" for data sources. 11 data source types cover all common enterprise data source scenarios, enabling one-stop registration, testing, and monitoring through declarative configuration.
Credential management must be separate from connection configuration. CredentialVault provides encrypted storage, automatic rotation, and operational auditing, eliminating the security risk of plaintext passwords scattered everywhere.
Source Mapping bridges Ontology and the external world. Three mapping modes (VIRTUAL/REPLICATED/FEDERATED) serve different scenarios, allowing existing systems to join the Ontology semantic ecosystem without data migration.
Cross-data-source queries are transparent to users. The Query Router automatically handles predicate pushdown, result merging, and cache optimization — users only need to think in terms of Ontology semantics.

#Next Article

S4-14: Ontology Modeling Best Practices: 6 Golden Rules — We will summarize six best practices for Ontology modeling: naming conventions, granularity control, relation direction, interface extraction, metric design, and versioning.

tags: connection-registry, data-source, credential-management, source-mapping, ontology, grpc, connection-testing, health-check, coomia-dip