Doris Query Path (even for small data):

Client --> FE --> Query Plan --> Distribute --> BE1,BE2,BE3
                                                    |
                                              Shuffle/Merge
                                                    |
                                              <-----+
                                              Return Result

Even querying 100 rows requires:
1. FE parses SQL (~2ms)
2. Generate distributed plan (~3ms)
3. Network transfer to BE (~5ms)
4. BE execution (~1ms)
5. Result return (~5ms)
--------------------------
Minimum overhead: ~16ms

DuckDB Embedded Query (same process):

+----------------------------+
|     Python/Java Process    |
|                            |
|  +------+    +----------+  |
|  | App  |--->| DuckDB   |  |
|  | Code |<---| Engine   |  |
|  +------+    +----------+  |
|                            |
|  Data passed in-process    |
|  No network, no serializ.  |
+----------------------------+

Same 100-row query:
1. DuckDB parses SQL (~0.1ms)
2. Columnar engine processes (~0.3ms)
3. Result returned in-memory (~0ms)
--------------------------
Minimum overhead: ~0.4ms (1/40th of Doris)

coomia-dip Dual-Engine Architecture:

+------------------------------------------+
|           QueryFederationService         |
|                                          |
|  +----------------------------------+    |
|  |         Query Router             |    |
|  |                                  |    |
|  |  Rows > 1M or needs indexes?    |    |
|  |  +--------+      +----------+   |    |
|  |  |  Yes   |      |    No    |   |    |
|  |  +---+----+      +----+-----+   |    |
|  +------+----------------+----------+    |
|         |                |               |
|    +----+-----+    +----+------+         |
|    |  Doris   |    |  DuckDB   |         |
|    | (Remote) |    | (Embedded)|         |
|    |          |    |           |         |
|    | gRPC/SQL |    | In-Process|         |
|    +----------+    +-----------+         |
+------------------------------------------+

Data Flow Between Doris and DuckDB:

1. Large datasets -> Doris storage + query
   entity_common (50M rows) -> Doris

2. Function context -> DuckDB temporary load
   During Function execution:
   +--------------+     +--------------+
   |   Doris      |     |   DuckDB     |
   |              |     |   (temp)     |
   | SELECT * FROM|---->| Load to mem  |
   | entity_common|     | Complex calc |
   | WHERE ...    |     | Return result|
   | LIMIT 10000  |     |              |
   +--------------+     +--------------+

3. Results -> Write back to Doris via Data Layer

class FunctionContext:
    """Ontology Function execution context"""

    def __init__(
        self,
        world_id: str,
        function_id: str,
        input_objects: list[OntologyObject],
        config: FunctionConfig
    ):
        self.world_id = world_id
        self.function_id = function_id
        self.input_objects = input_objects

        # Create embedded DuckDB instance
        self._duck = duckdb.connect(":memory:")
        self._setup_extensions()
        self._load_input_data()

    def _setup_extensions(self):
        """Load DuckDB extensions"""
        self._duck.execute("INSTALL httpfs; LOAD httpfs;")
        self._duck.execute("INSTALL json; LOAD json;")
        self._duck.execute("INSTALL parquet; LOAD parquet;")

        # Configure MinIO access for reading Iceberg data files
        self._duck.execute(f"""
            SET s3_endpoint = '{self.config.minio_endpoint}';
            SET s3_access_key_id = '{self.config.minio_access_key}';
            SET s3_secret_access_key = '{self.config.minio_secret_key}';
            SET s3_use_ssl = false;
            SET s3_url_style = 'path';
        """)

    def _load_input_data(self):
        """Load input data into DuckDB"""
        records = [obj.to_dict() for obj in self.input_objects]
        df = pd.DataFrame(records)
        self._duck.execute(
            "CREATE TABLE input_objects AS SELECT * FROM df"
        )

    def query(self, sql: str) -> list[dict]:
        """Execute SQL query in function context"""
        result = self._duck.execute(sql).fetchdf()
        return result.to_dict(orient='records')

    def execute_expression(self, expr: str) -> any:
        """Execute single-value expression"""
        result = self._duck.execute(f"SELECT {expr}").fetchone()
        return result[0] if result else None

    def close(self):
        """Release DuckDB resources"""
        self._duck.close()

class DerivedPropertyEngine:
    """Compute derived properties using DuckDB"""

    def __init__(self):
        self._duck = duckdb.connect(":memory:")

    async def compute_derived_properties(
        self,
        entity: OntologyObject,
        derived_defs: list[DerivedPropertyDef],
        related_entities: list[OntologyObject]
    ) -> dict[str, any]:
        """Compute all derived properties for an entity"""
        results = {}
        self._load_entity(entity)
        self._load_related(related_entities)

        for prop_def in derived_defs:
            value = await self._compute_single(prop_def)
            results[prop_def.name] = value

        return results

    async def _compute_single(self, prop_def: DerivedPropertyDef) -> any:
        """Compute a single derived property"""
        match prop_def.computation_type:
            case ComputationType.EXPRESSION:
                return self._duck.execute(
                    f"SELECT {prop_def.expression} FROM current_entity"
                ).fetchone()[0]

            case ComputationType.AGGREGATION:
                return self._duck.execute(f"""
                    SELECT {prop_def.aggregation_func}({prop_def.source_field})
                    FROM related_entities
                    WHERE relation_type = '{prop_def.relation_type}'
                """).fetchone()[0]

            case ComputationType.WINDOW:
                return self._duck.execute(f"""
                    SELECT {prop_def.window_expression}
                    FROM related_entities
                    WHERE entity_id = (SELECT entity_id FROM current_entity)
                """).fetchone()[0]

            case ComputationType.CONDITIONAL:
                return self._duck.execute(f"""
                    SELECT {prop_def.case_expression}
                    FROM current_entity
                """).fetchone()[0]

# Example: Customer entity derived properties

derived_properties = [
    DerivedPropertyDef(
        name="total_order_amount",
        computation_type=ComputationType.AGGREGATION,
        expression="SUM(amount)",
        source_field="amount",
        relation_type="HAS_ORDER",
        aggregation_func="SUM"
    ),
    DerivedPropertyDef(
        name="order_count",
        computation_type=ComputationType.AGGREGATION,
        expression="COUNT(*)",
        source_field="*",
        relation_type="HAS_ORDER",
        aggregation_func="COUNT"
    ),
    DerivedPropertyDef(
        name="avg_order_amount",
        computation_type=ComputationType.EXPRESSION,
        expression="total_order_amount / NULLIF(order_count, 0)"
    ),
    DerivedPropertyDef(
        name="customer_tier",
        computation_type=ComputationType.CONDITIONAL,
        case_expression="""
            CASE
                WHEN total_order_amount > 100000 THEN 'platinum'
                WHEN total_order_amount > 50000  THEN 'gold'
                WHEN total_order_amount > 10000  THEN 'silver'
                ELSE 'bronze'
            END
        """
    ),
]

WITH order_stats AS (
    SELECT
        SUM(amount) AS total_order_amount,
        COUNT(*) AS order_count,
        SUM(CASE WHEN order_date >= CURRENT_DATE - INTERVAL '30 days'
            THEN amount ELSE 0 END) AS last_month_amount,
        SUM(CASE WHEN order_date >= CURRENT_DATE - INTERVAL '60 days'
                  AND order_date < CURRENT_DATE - INTERVAL '30 days'
            THEN amount ELSE 0 END) AS prev_month_amount
    FROM related_entities
    WHERE relation_type = 'HAS_ORDER'
)
SELECT
    total_order_amount,
    order_count,
    total_order_amount / NULLIF(order_count, 0) AS avg_order_amount,
    CASE
        WHEN total_order_amount > 100000 THEN 'platinum'
        WHEN total_order_amount > 50000  THEN 'gold'
        WHEN total_order_amount > 10000  THEN 'silver'
        ELSE 'bronze'
    END AS customer_tier,
    (last_month_amount - prev_month_amount) /
    NULLIF(prev_month_amount, 0) * 100 AS purchase_trend
FROM order_stats;

class IcebergDuckDBReader:
    """Read Iceberg data using DuckDB"""

    def __init__(self, minio_config: MinIOConfig):
        self._duck = duckdb.connect(":memory:")
        self._configure_s3(minio_config)

    def _configure_s3(self, config: MinIOConfig):
        self._duck.execute(f"""
            INSTALL httpfs; LOAD httpfs;
            INSTALL iceberg; LOAD iceberg;
            SET s3_endpoint = '{config.endpoint}';
            SET s3_access_key_id = '{config.access_key}';
            SET s3_secret_access_key = '{config.secret_key}';
            SET s3_use_ssl = false;
            SET s3_url_style = 'path';
        """)

    def read_iceberg_table(
        self,
        table_path: str,
        filters: str | None = None,
        columns: list[str] | None = None,
        limit: int | None = None
    ) -> pd.DataFrame:
        """Read Iceberg table data"""
        col_expr = ", ".join(columns) if columns else "*"
        sql = f"SELECT {col_expr} FROM iceberg_scan('{table_path}')"
        if filters:
            sql += f" WHERE {filters}"
        if limit:
            sql += f" LIMIT {limit}"
        return self._duck.execute(sql).fetchdf()

    def read_iceberg_snapshot(
        self, table_path: str, snapshot_id: int
    ) -> pd.DataFrame:
        """Read specific Iceberg snapshot (time-travel)"""
        return self._duck.execute(f"""
            SELECT * FROM iceberg_scan(
                '{table_path}',
                allow_moved_paths = true,
                version = '{snapshot_id}'
            )
        """).fetchdf()

def read_parquet_from_minio(self, file_paths: list[str]) -> pd.DataFrame:
    """Read Parquet files directly from MinIO"""
    paths_str = ", ".join(f"'{p}'" for p in file_paths)
    return self._duck.execute(f"""
        SELECT * FROM read_parquet([{paths_str}])
    """).fetchdf()

class ActionExecutor:
    """Action executor"""

    async def execute_action(
        self, action_def: ActionDefinition, context: ActionContext
    ) -> ActionResult:
        duck = duckdb.connect(":memory:")
        try:
            await self._load_action_data(duck, action_def, context)

            match action_def.action_type:
                case ActionType.BATCH_UPDATE:
                    result = await self._batch_update(duck, action_def, context)
                case ActionType.COMPUTATION:
                    result = await self._compute(duck, action_def, context)
                case ActionType.VALIDATION:
                    result = await self._validate(duck, action_def, context)
            return result
        finally:
            duck.close()

    async def _batch_update(self, duck, action_def, context) -> ActionResult:
        """Batch update Action"""
        updated = duck.execute("""
            WITH credit_scores AS (
                SELECT
                    entity_id,
                    total_order_amount,
                    payment_delay_avg,
                    order_count,
                    (total_order_amount / 1000.0) * 0.4
                    + (1.0 / (1 + payment_delay_avg)) * 0.3
                    + LEAST(order_count / 10.0, 1.0) * 0.3 AS credit_score
                FROM customers
            )
            SELECT
                entity_id,
                credit_score,
                CASE
                    WHEN credit_score >= 0.8 THEN 'A'
                    WHEN credit_score >= 0.6 THEN 'B'
                    WHEN credit_score >= 0.4 THEN 'C'
                    ELSE 'D'
                END AS credit_rating
            FROM credit_scores
        """).fetchdf()

        return ActionResult(
            updates=updated.to_dict(orient='records'),
            affected_count=len(updated)
        )

-- Complex window analysis in DuckDB
-- Scenario: Equipment fault trend and anomaly detection

WITH fault_series AS (
    SELECT
        entity_id,
        fault_date,
        fault_count,
        AVG(fault_count) OVER (
            PARTITION BY entity_id
            ORDER BY fault_date
            ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
        ) AS moving_avg_7d,
        AVG(fault_count) OVER (
            PARTITION BY entity_id
            ORDER BY fault_date
            ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
        ) AS moving_avg_30d,
        STDDEV(fault_count) OVER (
            PARTITION BY entity_id
            ORDER BY fault_date
            ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
        ) AS stddev_30d,
        fault_count - LAG(fault_count, 1) OVER (
            PARTITION BY entity_id
            ORDER BY fault_date
        ) AS day_over_day,
        RANK() OVER (ORDER BY fault_count DESC) AS fault_rank
    FROM equipment_faults
)
SELECT
    entity_id, fault_date, fault_count,
    moving_avg_7d, moving_avg_30d,
    CASE
        WHEN fault_count > moving_avg_30d + 2 * stddev_30d THEN 'ANOMALY'
        WHEN fault_count > moving_avg_30d + stddev_30d THEN 'WARNING'
        ELSE 'NORMAL'
    END AS anomaly_status,
    day_over_day, fault_rank
FROM fault_series
WHERE fault_date >= CURRENT_DATE - INTERVAL '90 days'
ORDER BY entity_id, fault_date;

Test Environment:
- Doris: 3 FE + 3 BE (16C/64GB each)
- DuckDB: Embedded (on 16C/32GB app node)
- Data: entity_common table subsets

Dimensions:
1. Data sizes: 100 / 1K / 10K / 100K / 1M / 10M rows
2. Query complexity: simple agg / window / multi-join

Latency (ms) - Simple Aggregation (COUNT + AVG + GROUP BY)

Data Size     Doris    DuckDB    Winner
----------------------------------------------
100 rows      18ms     0.3ms     DuckDB (60x)
1K rows       20ms     0.8ms     DuckDB (25x)
10K rows      22ms     3ms       DuckDB (7x)
100K rows     28ms     15ms      DuckDB (2x)
1M rows       45ms     120ms     Doris (2.7x)
10M rows      150ms    1200ms    Doris (8x)
100M rows     350ms    OOM       Doris (inf)

Crossover point: ~500K rows

Latency (ms) - Complex Window Functions (multi-layer + CTE)

Data Size     Doris    DuckDB    Winner
----------------------------------------------
100 rows      35ms     0.5ms     DuckDB (70x)
1K rows       42ms     2ms       DuckDB (21x)
10K rows      65ms     12ms      DuckDB (5x)
100K rows     180ms    80ms      DuckDB (2.2x)
1M rows       850ms    650ms     DuckDB (1.3x)
10M rows      2500ms   5800ms    Doris (2.3x)

Crossover: ~3M rows (DuckDB advantage extends for complex queries)

Latency Chart (simple agg, log scale):

10000ms |                                          D
        |                                       D
 1000ms |                                    d
        |                              D
  100ms |                          d
        |                    D
   10ms | D  D  D  D    d
        |          d
    1ms |    d
        | d
  0.1ms |d
        +--------------------------------------
          100  1K  10K  100K  1M   10M  100M

D = Doris,  d = DuckDB
Crossover ~ 500K rows

class QueryRouter:
    """Query router: auto-select Doris or DuckDB"""

    DUCK_DB_MAX_ROWS = 500_000
    DUCK_DB_MAX_DATA_SIZE_MB = 512

    async def route_query(
        self, query: ParsedQuery, context: QueryContext
    ) -> EngineChoice:
        """Decide which engine to use"""

        # Rule 1: Vector/text search -> Doris
        if query.has_vector_search or query.has_text_search:
            return EngineChoice.DORIS

        # Rule 2: Estimate data volume
        estimated_rows = await self._estimate_row_count(query)

        # Rule 3: Large data -> Doris
        if estimated_rows > self.DUCK_DB_MAX_ROWS:
            return EngineChoice.DORIS

        # Rule 4: Complex windows + medium data -> DuckDB
        if query.has_complex_windows and estimated_rows < 3_000_000:
            return EngineChoice.DUCKDB

        # Rule 5: In function context -> DuckDB
        if context.is_function_context:
            return EngineChoice.DUCKDB

        # Rule 6: Small data -> DuckDB
        if estimated_rows < self.DUCK_DB_MAX_ROWS:
            return EngineChoice.DUCKDB

        return EngineChoice.DORIS

class QueryExecutorWithFallback:
    """Query executor with automatic fallback"""

    async def execute(self, query: str, context: QueryContext) -> QueryResult:
        parsed = self.parser.parse(query)
        engine = await self.router.route_query(parsed, context)

        try:
            if engine == EngineChoice.DUCKDB:
                return await self._execute_duckdb(parsed, context)
            else:
                return await self._execute_doris(parsed, context)
        except DuckDBOutOfMemoryError:
            logger.warning(f"DuckDB OOM, falling back to Doris")
            return await self._execute_doris(parsed, context)
        except DorisConnectionError:
            if parsed.estimated_rows < self.DUCK_DB_MAX_ROWS * 2:
                logger.warning(f"Doris unavailable, falling back to DuckDB")
                return await self._execute_duckdb(parsed, context)
            raise

class AgentAnalyticsTool:
    """Agent local analytics tool"""

    def __init__(self):
        self._duck = duckdb.connect(":memory:")

    async def load_context_data(
        self, world_id: str, object_types: list[str],
        limit_per_type: int = 10000
    ) -> None:
        for obj_type in object_types:
            data = await self.data_client.fetch_objects(
                world_id=world_id,
                object_type_id=obj_type,
                limit=limit_per_type
            )
            df = pd.DataFrame([obj.to_dict() for obj in data])
            self._duck.execute(
                f"CREATE TABLE {obj_type.lower()} AS SELECT * FROM df"
            )

    def analyze(self, sql: str) -> dict:
        try:
            result = self._duck.execute(sql).fetchdf()
            return {
                "status": "success",
                "data": result.to_dict(orient='records'),
                "row_count": len(result),
                "columns": list(result.columns)
            }
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def describe_tables(self) -> dict:
        tables = self._duck.execute(
            "SELECT table_name FROM information_schema.tables "
            "WHERE table_schema = 'main'"
        ).fetchdf()
        result = {}
        for table_name in tables['table_name']:
            columns = self._duck.execute(f"DESCRIBE {table_name}").fetchdf()
            result[table_name] = columns.to_dict(orient='records')
        return result

Agent Interaction Flow:

User: "Analyze equipment fault trends over the last 30 days"

Agent executes:

Step 1: load_context_data(
    world_id="world-prod-001",
    object_types=["Equipment", "MaintenanceRecord"]
)

Step 2: analyze("""
    SELECT
        DATE_TRUNC('day', fault_date) AS day,
        COUNT(*) AS fault_count,
        COUNT(DISTINCT equipment_id) AS affected_equipment,
        AVG(repair_hours) AS avg_repair_time
    FROM maintenancerecord
    WHERE fault_date >= CURRENT_DATE - INTERVAL '30 days'
    GROUP BY DATE_TRUNC('day', fault_date)
    ORDER BY day
""")

Step 3: analyze("""
    SELECT
        equipment_type,
        COUNT(*) AS total_faults,
        AVG(repair_hours) AS avg_repair,
        MAX(fault_date) AS last_fault
    FROM maintenancerecord m
    JOIN equipment e ON m.equipment_id = e.entity_id
    WHERE m.fault_date >= CURRENT_DATE - INTERVAL '30 days'
    GROUP BY equipment_type
    ORDER BY total_faults DESC
""")

Agent responds: "Equipment fault trend analysis for the past 30 days..."

import duckdb

duck = duckdb.connect(":memory:", config={
    'memory_limit': '4GB',
    'threads': 8,
    'temp_directory': '/tmp/duckdb',
    'max_temp_directory_size': '10GB'
})

duck.execute("PRAGMA enable_progress_bar;")
duck.execute("SELECT * FROM duckdb_settings()").fetchdf()

class DuckDBDataLoader:

    @staticmethod
    def load_from_arrow(duck, table_name: str, arrow_table) -> None:
        """Zero-copy load via Apache Arrow (fastest)"""
        duck.execute(f"CREATE TABLE {table_name} AS SELECT * FROM arrow_table")

    @staticmethod
    def load_from_parquet(duck, table_name: str, parquet_path: str) -> None:
        """Direct Parquet file read"""
        duck.execute(f"""
            CREATE TABLE {table_name} AS
            SELECT * FROM read_parquet('{parquet_path}')
        """)

    @staticmethod
    def load_from_csv(duck, table_name: str, csv_path: str, delimiter=','):
        """Read CSV file"""
        duck.execute(f"""
            CREATE TABLE {table_name} AS
            SELECT * FROM read_csv_auto('{csv_path}', delim='{delimiter}', header=true)
        """)

-- DuckDB query optimization tips

-- 1. Use EXPLAIN ANALYZE
EXPLAIN ANALYZE
SELECT object_type_id, COUNT(*) FROM entity_common GROUP BY object_type_id;

-- 2. Tuning PRAGMAs
PRAGMA force_parallelism;
PRAGMA perfect_ht_threshold = 12;

-- 3. Column pruning (select only needed columns)
SELECT entity_id, title, status    -- Good
-- SELECT * FROM entity_common     -- Bad

-- 4. DuckDB auto-vectorizes all operations
-- No manual optimization needed, but avoid scalar UDFs

DuckDB's Position in coomia-dip:

+-------------------------------------------------+
|                 Query Layer                      |
|                                                  |
|  +------------------------------------------+   |
|  |         QueryFederationService            |   |
|  |                                           |   |
|  |  +-------------------------------------+  |   |
|  |  |          Query Router               |  |   |
|  |  |  Data size? Index? Windows? Context?|  |   |
|  |  +--------+----------------+-----------+  |   |
|  |           |                |               |   |
|  |     +-----+-----+   +-----+------+       |   |
|  |     |   Doris   |   |  DuckDB    |       |   |
|  |     |           |   |            |       |   |
|  |     | >500K rows|   | <500K rows |       |   |
|  |     | Vector    |   | Window fns |       |   |
|  |     | Full-text |   | Ad-hoc     |       |   |
|  |     | Persistent|   | Embedded   |       |   |
|  |     +-----------+   +------------+       |   |
|  +------------------------------------------+   |
+-------------------------------------------------+