“

Series: S3 Data Foundation · Article 12 | Level: Advanced | Reading Time: 20 min

Analytics Engine: Ontology Wrapper for OLAP Capabilities

Tags: #AnalyticsEngine #OLAP #Doris #Aggregation #Dashboard #coomia-dip

#TL;DR

The coomia-dip platform wraps Apache Doris's OLAP analytical capabilities into an Ontology-semantic analysis query interface. Instead of writing complex SQL aggregation queries, users analyze data through OQL's AGGREGATE statement using entity types and properties as dimensions. This article fully dissects the analytics engine's architecture, including OQL AGGREGATE to Doris SQL compilation, multidimensional aggregation (Cube/Rollup) implementation, pre-aggregation materialized view auto-creation and matching, the Metric System's definition and expansion mechanism, real-time dashboard data push, and analytics query performance optimization strategies.

#1. Analytics Engine Positioning

#1.1 Ontology Analytics vs Traditional BI

Code

Traditional BI vs Ontology Analytics:

Traditional BI:
  User -> SQL query -> Data Warehouse -> Results
  Pain: Users must understand table schemas and write SQL

Ontology Analytics:
  User -> OQL AGGREGATE -> Analytics Engine -> Results
  Advantage: Users only need to understand entity types and properties

Comparison:
  SQL:  SELECT department, AVG(age), COUNT(*)
        FROM entity_common
        WHERE entity_type = 'Person'
        AND JSON_EXTRACT(properties, '$.status') = 'active'
        GROUP BY JSON_EXTRACT(properties, '$.department')

  OQL:  AGGREGATE Person
        GROUP BY department
        SELECT AVG(age), COUNT(*)
        WHERE status = 'active'

#1.2 Architecture Position

Code

Analytics Engine in the Data Foundation:

+------------------------------------------+
|              OQL Query Interface          |
+------------------------------------------+
|  +--------+  +--------+  +------------+  |
|  | FETCH  |  |TRAVERSE|  | AGGREGATE  |  |
|  | Entity |  | Graph  |  | Analytics  |  |
|  +--------+  +--------+  +------------+  |
+------------------------------------------+
|         Query Optimization Layer          |
|  (Pushdown, MV matching, Agg strategy)   |
+------------------------------------------+
|         Apache Doris OLAP Engine          |
|  (Columnar, Vectorized, Materialized)    |
+------------------------------------------+

#2. AGGREGATE Statement Compilation

#2.1 OQL AGGREGATE Syntax

BNF

aggregate_stmt ::= 'AGGREGATE' entity_type
                   'GROUP' 'BY' group_columns
                   'SELECT' agg_expressions
                   ['WHERE' predicate]
                   ['HAVING' having_predicate]
                   ['ORDER' 'BY' order_columns]
                   ['LIMIT' number]

agg_expressions ::= agg_expr (',' agg_expr)*
agg_expr ::= agg_function '(' column ')' ['AS' alias]
agg_function ::= 'COUNT' | 'SUM' | 'AVG' | 'MIN' | 'MAX'
                | 'PERCENTILE' | 'STDDEV' | 'VARIANCE'
                | 'DISTINCT_COUNT' | 'TOP_N'

#2.2 Compilation to Doris SQL

Python

class AggregateCompiler:
    """AGGREGATE statement to Doris SQL compiler"""

    def compile(self, stmt: AggregateStatement) -> str:
        entity_type = stmt.entity_type
        schema = self._registry.get_entity_type(entity_type)

        select_parts = []
        for group_col in stmt.group_by:
            select_parts.append(
                f"JSON_EXTRACT(properties, '$.{group_col}') "
                f"AS {group_col}"
            )

        for agg in stmt.aggregations:
            sql_agg = self._compile_aggregation(agg, schema)
            select_parts.append(sql_agg)

        where_parts = [f"entity_type = '{entity_type}'"]
        if stmt.where_clause:
            where_parts.append(
                self._compile_predicate(stmt.where_clause, schema)
            )

        group_parts = [
            f"JSON_EXTRACT(properties, '$.{col}')"
            for col in stmt.group_by
        ]

        sql = f"""
        SELECT {', '.join(select_parts)}
        FROM entity_common
        WHERE {' AND '.join(where_parts)}
        GROUP BY {', '.join(group_parts)}
        """

        if stmt.having_clause:
            sql += f" HAVING {self._compile_predicate(stmt.having_clause, schema)}"

        if stmt.order_by:
            order_parts = [
                f"{col} {direction}"
                for col, direction in stmt.order_by
            ]
            sql += f" ORDER BY {', '.join(order_parts)}"

        if stmt.limit:
            sql += f" LIMIT {stmt.limit}"

        return sql

    def _compile_aggregation(
        self, agg: AggregationExpr, schema: EntitySchema
    ) -> str:
        col = agg.column
        func = agg.function.upper()
        alias = agg.alias or f"{func.lower()}_{col}"
        col_expr = f"JSON_EXTRACT(properties, '$.{col}')"

        if func == 'DISTINCT_COUNT':
            return f"COUNT(DISTINCT {col_expr}) AS {alias}"
        elif func == 'TOP_N':
            return f"TOPN({col_expr}, {agg.n}) AS {alias}"
        elif func == 'PERCENTILE':
            return (f"PERCENTILE_APPROX({col_expr}, "
                    f"{agg.percentile}) AS {alias}")
        else:
            return f"{func}({col_expr}) AS {alias}"

#3. Multidimensional Analysis

#3.1 Cube and Rollup

Python

class MultiDimensionalAnalysis:
    """Multidimensional analysis engine"""

    def cube_query(
        self,
        entity_type: str,
        dimensions: list[str],
        measures: list[AggregationExpr]
    ) -> str:
        """Generate CUBE query — all dimension combinations"""
        schema = self._registry.get_entity_type(entity_type)

        dim_exprs = [
            f"JSON_EXTRACT(properties, '$.{d}') AS {d}"
            for d in dimensions
        ]
        measure_exprs = [
            self._compile_aggregation(m, schema) for m in measures
        ]

        return f"""
        SELECT {', '.join(dim_exprs + measure_exprs)}
        FROM entity_common
        WHERE entity_type = '{entity_type}'
        GROUP BY CUBE({', '.join(
            f"JSON_EXTRACT(properties, '$.{d}')"
            for d in dimensions
        )})
        """

    def rollup_query(
        self,
        entity_type: str,
        hierarchy: list[str],
        measures: list[AggregationExpr]
    ) -> str:
        """Generate ROLLUP query — hierarchical aggregation"""
        schema = self._registry.get_entity_type(entity_type)

        dim_exprs = [
            f"JSON_EXTRACT(properties, '$.{d}') AS {d}"
            for d in hierarchy
        ]
        measure_exprs = [
            self._compile_aggregation(m, schema) for m in measures
        ]

        return f"""
        SELECT {', '.join(dim_exprs + measure_exprs)}
        FROM entity_common
        WHERE entity_type = '{entity_type}'
        GROUP BY ROLLUP({', '.join(
            f"JSON_EXTRACT(properties, '$.{d}')"
            for d in hierarchy
        )})
        """

#3.2 Multidimensional Analysis Example

Code

CUBE Analysis Example:

OQL:
  AGGREGATE Device
  CUBE BY region, type, status
  SELECT COUNT(*), AVG(uptime)

Result (8 dimension combinations):
+----------+----------+----------+-------+----------+
| region    | type      | status   | count | avg_up   |
+----------+----------+----------+-------+----------+
| East      | Sensor    | Running  | 150   | 99.2%    |
| East      | Sensor    | NULL     | 200   | 95.1%    |
| East      | NULL      | Running  | 300   | 98.5%    |
| East      | NULL      | NULL     | 500   | 94.2%    |
| NULL      | Sensor    | Running  | 400   | 99.0%    |
| NULL      | Sensor    | NULL     | 600   | 96.3%    |
| NULL      | NULL      | Running  | 800   | 98.8%    |
| NULL      | NULL      | NULL     | 1200  | 95.5%    |
+----------+----------+----------+-------+----------+
  (NULL means that dimension is aggregated)

#4. Metric System

#4.1 Metric Definitions

Python

@dataclass
class MetricDefinition:
    """Metric definition"""
    name: str
    display_name: str
    description: str
    entity_type: str
    formula: MetricFormula
    dimensions: list[str]
    filters: list[MetricFilter] | None
    time_granularity: str | None
    cache_ttl: timedelta | None

    def compile_to_sql(self, schema: EntitySchema) -> str:
        return self.formula.compile(schema)


# Registration examples
metric_registry.register(MetricDefinition(
    name='equipment_utilization',
    display_name='Equipment Utilization',
    description='Running hours / total hours',
    entity_type='Device',
    formula=MetricFormula(
        type='DERIVED',
        expression='SUM(running_hours) / SUM(total_hours) * 100'
    ),
    dimensions=['region', 'device_type'],
    time_granularity='DAILY',
    cache_ttl=timedelta(minutes=5),
))

metric_registry.register(MetricDefinition(
    name='direct_reports',
    display_name='Direct Reports',
    description='Count of entities connected via Manages edge',
    entity_type='Person',
    formula=MetricFormula(
        type='SIMPLE',
        expression='COUNT(edge WHERE edge_type = "Manages")'
    ),
    dimensions=['department'],
    cache_ttl=timedelta(minutes=30),
))

#4.2 Metric Expansion Mechanism

Python

class MetricExpansionEngine:
    """Metric expansion engine"""

    def expand_metric(
        self, entity_type: str, metric_name: str,
        context: QueryContext
    ) -> str:
        metric = self._registry.get_metric(metric_name)

        if metric is None:
            raise MetricNotFoundError(
                f"Unknown metric: {metric_name}"
            )

        if metric.entity_type != entity_type:
            raise MetricTypeMismatchError(
                f"Metric '{metric_name}' is defined on "
                f"'{metric.entity_type}', not '{entity_type}'"
            )

        if metric.cache_ttl:
            cached = self._cache.get(metric_name, context)
            if cached:
                return cached

        if metric.formula.type == 'SIMPLE':
            sql = self._expand_simple(metric, context)
        elif metric.formula.type == 'DERIVED':
            sql = self._expand_derived(metric, context)
        elif metric.formula.type == 'COMPOSITE':
            sql = self._expand_composite(metric, context)

        return sql

    def _expand_simple(
        self, metric: MetricDefinition, context: QueryContext
    ) -> str:
        if 'edge' in metric.formula.expression.lower():
            edge_type = self._extract_edge_type(
                metric.formula.expression
            )
            agg_func = self._extract_agg_func(
                metric.formula.expression
            )

            return f"""
            (SELECT {agg_func}(*)
             FROM entity_edge
             WHERE source_id = t.entity_id
               AND edge_type = '{edge_type}')
            """

        return metric.formula.compile(context.schema)

#5. Pre-Aggregation Materialized Views

#5.1 Auto-Creation Strategy

Python

class AutoMaterializedViewManager:
    """Automatic materialized view manager"""

    def analyze_and_create(
        self, query_log: list[QueryLogEntry]
    ):
        patterns = self._extract_aggregation_patterns(query_log)

        for pattern in patterns:
            if self._should_materialize(pattern):
                self._create_materialized_view(pattern)

    def _should_materialize(
        self, pattern: AggregationPattern
    ) -> bool:
        if pattern.daily_executions < 100:
            return False
        if pattern.avg_execution_ms < 500:
            return False
        if pattern.estimated_result_size_mb > 100:
            return False
        if pattern.source_update_frequency_per_minute > 1:
            return False
        return True

#5.2 Doris MV Auto-Matching

Code

Doris Materialized View Auto-Matching:

Original query:
  SELECT region, COUNT(*), AVG(uptime)
  FROM entity_common
  WHERE entity_type = 'Device'
  GROUP BY JSON_EXTRACT(properties, '$.region')

MV definition:
  CREATE MATERIALIZED VIEW mv_device_region AS
  SELECT entity_type,
         JSON_EXTRACT(properties, '$.region') AS region,
         COUNT(*) AS cnt,
         SUM(JSON_EXTRACT(properties, '$.uptime')) AS sum_uptime,
         COUNT(JSON_EXTRACT(properties, '$.uptime')) AS cnt_uptime
  FROM entity_common
  GROUP BY entity_type, JSON_EXTRACT(properties, '$.region')

Doris auto-matching:
  1. Detects GROUP BY dimensions match
  2. AVG(uptime) derivable from SUM(uptime)/COUNT(uptime)
  3. Auto-rewrites query to use MV

Rewritten:
  SELECT region, cnt, sum_uptime / cnt_uptime
  FROM mv_device_region
  WHERE entity_type = 'Device'

Effect: Query time from 2s -> 20ms (100x speedup)

#6. Real-Time Dashboards

#6.1 Data Push Architecture

Code

Real-Time Dashboard Data Push:

+----------+   +----------+   +----------+
| Flink CDC|-->| Doris    |-->| MV       |
| (ingest) |   | (storage)|   | (refresh)|
+----------+   +----------+   +----------+
                                    |
                                    v
                              +----------+
                              | Dashboard|
                              | Query    |
                              | Service  |
                              +----------+
                                    |
                    +---------------+---------------+
                    |               |               |
                    v               v               v
              +----------+  +----------+  +----------+
              | WebSocket|  | SSE      |  | Polling  |
              | Push     |  | Push     |  | Poll     |
              +----------+  +----------+  +----------+

#6.2 Dashboard Query Optimization

Python

class DashboardQueryService:
    """Dashboard query service"""

    async def get_dashboard_data(
        self, dashboard_id: str
    ) -> DashboardData:
        dashboard = self._registry.get_dashboard(dashboard_id)
        panels = dashboard.panels

        tasks = [
            self._execute_panel_query(panel) for panel in panels
        ]
        results = await asyncio.gather(*tasks)

        return DashboardData(
            dashboard_id=dashboard_id,
            panels={
                panel.id: result
                for panel, result in zip(panels, results)
            },
            timestamp=datetime.utcnow()
        )

    async def _execute_panel_query(
        self, panel: DashboardPanel
    ) -> PanelData:
        mv = self._mv_matcher.find_matching_mv(panel.query)
        if mv:
            return await self._execute_from_mv(mv, panel)

        cache_key = panel.query_hash
        cached = self._cache.get(cache_key)
        if cached and cached.age < panel.refresh_interval:
            return cached

        result = await self._execute_query(panel.query)
        self._cache.set(
            cache_key, result, ttl=panel.refresh_interval
        )
        return result

#7. Performance Benchmarks

Code

Analytics Engine Performance Benchmarks:

Test env: 1M entities, 5 types, 3 Doris BE nodes

+---------------------+----------+----------+----------+
| Query Type           | No MV    | With MV  | Speedup  |
+---------------------+----------+----------+----------+
| Single-dim COUNT     | 800ms    | 15ms     | 53x      |
| Single-dim AVG       | 1.2s     | 20ms     | 60x      |
| Two-dim GROUP BY     | 2.5s     | 30ms     | 83x      |
| CUBE (3 dimensions)  | 8.0s     | 50ms     | 160x     |
| ROLLUP (4 levels)    | 5.0s     | 40ms     | 125x     |
| PERCENTILE_APPROX    | 3.0s     | 100ms    | 30x      |
| Cross-type aggregate | 4.0s     | 200ms    | 20x      |
+---------------------+----------+----------+----------+

Conclusion: MVs are the decisive factor for analytics performance

#8. Testing Strategy

Python

class TestAnalyticsEngine:

    def test_aggregate_compilation(self):
        oql = """
        AGGREGATE Person
        GROUP BY department
        SELECT COUNT(*), AVG(age)
        WHERE status = 'active'
        """
        sql = compiler.compile(parse(oql))
        assert "GROUP BY" in sql
        assert "COUNT(*)" in sql
        assert "AVG" in sql

    def test_metric_expansion(self):
        result = execute(
            "FETCH Person WITH METRIC direct_reports LIMIT 10"
        )
        assert 'direct_reports' in result.columns
        assert all(
            isinstance(v, int) for v in result['direct_reports']
        )

    def test_mv_auto_matching(self):
        create_mv("mv_test",
            "SELECT type, COUNT(*) FROM entity_common GROUP BY type")
        plan = explain(
            "AGGREGATE Device GROUP BY type SELECT COUNT(*)"
        )
        assert "mv_test" in plan.used_views

    def test_cube_correctness(self):
        result = execute("""
        AGGREGATE Device CUBE BY region, type SELECT COUNT(*)
        """)
        null_rows = [r for r in result
                     if r['region'] is None or r['type'] is None]
        assert len(null_rows) > 0  # Should have aggregate rows

#Key Takeaways

Ontology-semantic analytics dramatically lowers the BI barrier: users analyze via OQL AGGREGATE using entity types and properties as dimensions, without understanding underlying table structures or SQL.
Materialized views are the decisive analytics performance factor: auto-created MVs can reduce aggregation queries from seconds to milliseconds (50-160x improvement).
The Metric System decouples metric definition from queries: define once, expand everywhere — metric formulas registered in the Metric Registry are referenced via the WITH METRIC clause in any query.
Multidimensional analysis (CUBE/ROLLUP) covers all dimension combinations: through Doris native CUBE/ROLLUP support, a single query yields aggregation results for all dimension combinations.
Real-time dashboards rely on layered caching and materialized views: MVs provide millisecond-level base queries, combined with result caching and WebSocket push for real-time data display.

#Next Article

Next up: S3-13 "Search Engine: Elasticsearch Ontology Integration" will demonstrate how Elasticsearch's full-text search capabilities are integrated into Ontology queries.

Tags: #AnalyticsEngine #OLAP #Doris #Aggregation #MaterializedView #MetricSystem #CUBE #ROLLUP #Dashboard #coomia-dip #DataFoundation