Implementation:ArroyoSystems Arroyo Table Catalog

Overview

Table Catalog defines the core table abstractions used by the Arroyo SQL planner, including ConnectorTable for source/sink/lookup tables, FieldSpec for column definitions, Table for the overall table enum, and produce_optimized_plan for applying DataFusion optimizer rules to SQL statements.

Description

The module defines:

• ConnectorTable: The primary table representation containing:
  • id: Optional connection ID
  • connector, name: Connector type and table name
  • connection_type: Source, Sink, or Both
  • fields: Column definitions as Vec<FieldSpec>
  • config: Serialized connector configuration
  • format: Optional data format (JSON, Avro, Protobuf, etc.)
  • event_time_field, watermark_field: Event time and watermark configuration
  • idle_time: Source idle timeout
  • primary_keys: Primary key columns for CDC and lookup tables
  • Methods: from_options() for creating from SQL WITH clause, is_updating() for CDC detection, has_virtual_fields()

• FieldSpec: An enum for column types:
  • Struct(Field): Regular data field
  • Metadata { field, key }: Connector metadata field (e.g., Kafka partition, offset)
  • Virtual { field, expression }: Computed field with SQL expression

• Table: An enum encompassing all table types:
  • ConnectorTable: Source/sink connector table
  • MemoryTable: In-memory table for intermediate results
  • TableFromQuery: Table created from a subquery (CREATE TABLE AS)
  • PreviewSink: Built-in preview output sink
  • LookupTable: External lookup table

• Insert: Represents INSERT statements:
  • InsertQuery: INSERT INTO with a query
  • Anonymous: Anonymous sink for preview

• produce_optimized_plan: Applies a curated set of DataFusion optimizer rules including: SimplifyExpressions, ReplaceDistinctWithAggregate, EliminateJoin, DecorrelatePredicateSubquery, ScalarSubqueryToJoin, PushDownFilter, PushDownLimit, EliminateFilter, and many others.

Usage

Tables are created from SQL DDL (CREATE TABLE, CREATE VIEW) and registered in the ArroyoSchemaProvider. The produce_optimized_plan function is the first step in query compilation.

Code Reference

Source Location

crates/arroyo-planner/src/tables.rs

Signature

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum FieldSpec {
    Struct(Field),
    Metadata { field: Field, key: String },
    Virtual { field: Field, expression: Box<Expr> },
}

pub struct ConnectorTable {
    pub id: Option<i64>,
    pub connector: String,
    pub name: String,
    pub connection_type: ConnectionType,
    pub fields: Vec<FieldSpec>,
    pub config: String,
    pub format: Option<Format>,
    pub event_time_field: Option<String>,
    pub watermark_field: Option<String>,
    pub idle_time: Option<Duration>,
    pub primary_keys: Vec<String>,
    // ...
}

pub enum Table {
    ConnectorTable(ConnectorTable),
    MemoryTable { /* ... */ },
    TableFromQuery { /* ... */ },
    PreviewSink { /* ... */ },
    LookupTable { /* ... */ },
}

pub enum Insert {
    InsertQuery { /* ... */ },
    Anonymous { /* ... */ },
}

fn produce_optimized_plan(
    statement: &Statement,
    schema_provider: &ArroyoSchemaProvider,
) -> Result<LogicalPlan>

Import

use crate::tables::{ConnectorTable, FieldSpec, Table, Insert};

I/O Contract

Inputs

Name	Type	Description
statement	&Statement	Parsed SQL statement (from sqlparser)
schema_provider	&ArroyoSchemaProvider	Schema provider with table registry and analyzer
options	SQL WITH clause	Key-value configuration options for connectors

Outputs

Name	Type	Description
ConnectorTable	struct	Fully configured connector table with fields, format, and watermark settings
LogicalPlan	struct	Optimized DataFusion logical plan ready for Arroyo rewriting

Usage Examples

-- Creating a connector table (parsed by the planner into ConnectorTable)
CREATE TABLE orders (
    id INT,
    customer_id INT,
    amount FLOAT,
    event_time TIMESTAMP
) WITH (
    connector = 'kafka',
    topic = 'orders',
    format = 'json',
    event_time_field = 'event_time',
    watermark_field = 'event_time - INTERVAL 5 SECONDS'
);

Related Pages

• ArroyoSystems_Arroyo_Sql_Rewriters - SourceRewriter that processes ConnectorTable definitions
• ArroyoSystems_Arroyo_Plan_Rewriter - ArroyoRewriter that transforms the optimized plan
• ArroyoSystems_Arroyo_Debezium_Extension - CDC support triggered by updating ConnectorTables