[Feature Request]: Unified read-only federated queries across heterogeneous data sources

### Is there an existing issue for the same feature request?

- [x] I have checked the existing issues.

This issue refines the broad framework request in #24033 into a product definition with an explicit first delivery profile and testable acceptance criteria.

### Is your feature request related to a problem?

MatrixOne users often need to work with data that remains in different systems:

- MatrixOne native tables;
- files and object storage;
- Iceberg tables;
- operational relational databases;
- analytical, search, streaming, or application systems.

MatrixOne already has source-specific ways to access some external data, including files and Iceberg, but it does not yet provide a unified federation product model for registering, governing, querying, explaining, and limiting work across heterogeneous sources.

For remote relational data specifically, users must currently build ETL or CDC before they can:

- evaluate MatrixOne against existing operational data;
- migrate incrementally instead of moving every table at once;
- join migrated MatrixOne tables with tables that have not moved;
- run selective, read-only cross-system analysis;
- access source-current data without first creating another copy.

The problem is therefore broader than adding a MySQL scan function. MatrixOne needs one SQL and governance surface for data that remains outside MatrixOne, while retaining source-specific execution semantics and making the cost and consistency boundaries visible.

### Describe the feature you'd like

Add unified, read-only federated queries across heterogeneous data sources.

The product should provide source-neutral concepts for:

- connections and credential references;
- catalogs and externally managed table objects;
- source capabilities and safe pushdown;
- cross-source SQL execution;
- consistency and freshness semantics;
- resource governance;
- security and auditing;
- `EXPLAIN` and runtime observability.

The product model must not require every source type to share one physical scan implementation. File, lakehouse, relational, analytical, search, and streaming systems have different planning, consistency, and execution contracts.

#### Delivery profiles

This feature defines a common product contract and delivers the first complete set of profiles:

1. **Native profile**
- MatrixOne native tables remain unchanged and participate in cross-source SQL.

2. **Lake/external profile**
- Existing file and Iceberg access remains compatible.
- These sources should adopt the common source identity, security, resource, and observability semantics where applicable.
- Existing optimized physical scan implementations do not need to be rewritten merely to share a product model.

3. **Remote relational profile — first new provider profile**
- MySQL and PostgreSQL are the initial providers.
- Read-only foreign tables, conservative pushdown, local cross-source joins, bounded execution, and complete failure handling are required.

Future analytical, search, stream, and application/API profiles should be possible without redesigning the common product model. They are not required for this feature to reach its first GA.

#### Illustrative user experience

The exact grammar should be finalized in a design document. The following examples describe the intended behavior rather than freeze SQL syntax.

Register a reusable connection:

```sql
CREATE CONNECTION crm_source
TYPE MYSQL
OPTIONS (
host = 'mysql.example.com',
port = '3306',
user = 'mo_reader',
credential = SECRET 'crm-readonly',
tls_mode = 'required'
);
```

Expose one remote object explicitly:

```sql
CREATE FOREIGN TABLE crm_customers
CONNECTION crm_source
REMOTE OBJECT 'crm.customers';
```

The foreign table should discover and validate the remote schema when it is created and store a MatrixOne schema binding. An incompatible remote schema change must produce a clear schema-drift error rather than silently remap columns.

Query it with normal MatrixOne SQL:

```sql
SELECT id, name
FROM crm_customers
WHERE region = 'CN'
LIMIT 100;
```

Combine heterogeneous sources:

```sql
SELECT o.id, c.name, e.event_type
FROM local_orders AS o
JOIN crm_customers AS c
ON o.customer_id = c.id
JOIN iceberg_events AS e
ON e.order_id = o.id
WHERE o.created_at >= '2026-07-01'
AND c.region = 'CN';
```

Existing `EXTERNAL TABLE` behavior for files and Iceberg should remain compatible. A design may keep `EXTERNAL TABLE` for externally stored data and introduce `FOREIGN TABLE` for remote services, provided both participate in the common federation semantics and normal SQL namespace.

#### Product behavior

##### Data location and freshness

- A federated object is read from its source when the query executes unless the user has explicitly configured a separate cache, materialization, or synchronization feature.
- Federation must never silently copy or serve stale data.
- The source type, data location, and applicable freshness/snapshot semantics must be inspectable.
- Foreign tables are read-only in the first remote-relational profile.

##### Query and consistency semantics

- MatrixOne native tables use the MatrixOne transaction snapshot.
- Each external source uses the snapshot or statement semantics provided by that source/profile.
- A federated query does not imply a globally consistent snapshot or distributed transaction across sources.
- Cross-source joins execute in MatrixOne unless a future provider profile explicitly supports an equivalent pushed subplan.
- Only operations proven to preserve MatrixOne SQL semantics may be pushed down.
- Unsupported or inexact operations remain MatrixOne residual operations.
- Source-specific collation, time, decimal, NULL, JSON, and identifier behavior must not be silently treated as MatrixOne semantics.

##### Pushdown

The first remote-relational profile should support:

- column projection;
- a conservative set of exact predicates;
- bound parameters;
- literal, bounded `LIMIT`.

The common capability model may later represent aggregate, ordering, join, search, snapshot, or parallel-read capabilities, but providers must only advertise capabilities they can execute correctly.

##### Explainability and observability

`EXPLAIN` should identify source boundaries and distinguish pushed work from MatrixOne residual work.

`EXPLAIN ANALYZE` should report profile-appropriate metrics. For a remote relational scan, it should expose at least:

```text
Federated Scan: crm_source.crm.customers
Profile: relational
Provider: mysql
Pushed columns: id, name
Pushed predicates: region = ?
Residual predicates: none
Pushed limit: 100
Source rows read: 100
Source bytes read: ...
Source execution time: ...
Source wait time: ...
Decode time: ...
```

Equivalent source identity, selected-object/file, row/byte, planning, and scan metrics should be available for existing file/Iceberg scans where meaningful.

Credentials and sensitive parameter values must never appear in plans or metrics.

##### Resource and failure behavior

- Source connection timeout, query timeout, maximum rows/bytes, and concurrency must be configurable where applicable.
- A source must not create unbounded connections, requests, readers, goroutines, buffers, or memory use across CNs.
- MatrixOne query cancellation must promptly cancel or close active external work.
- Downstream early termination, including `LIMIT`, must release source resources.
- A slow or unavailable source must not destabilize the MatrixOne cluster.
- Connection pools must be isolated by account, connection identity, endpoint, and credential version.
- Credential rotation must invalidate or retire sessions created with the old credential.
- Source errors must use source-neutral MatrixOne error categories while retaining useful, redacted provider context.

##### Security and governance

- Credentials must use secret references and be encrypted at rest.
- Plaintext credentials must never appear in DDL output, plans, statement history, logs, metrics, profiles, traces, or errors.
- Connections, catalogs, external/foreign tables, and their privileges must be account-scoped.
- Audit records should identify the MatrixOne account, connection/source, external object, operation, status, and transferred rows/bytes without recording secrets.
- TLS and certificate verification behavior must be explicit.

#### Initial GA scope

- A common source identity, capability, security, resource, and observability contract.
- Compatibility with existing file and Iceberg external-table behavior.
- Normal SQL combining MatrixOne native, Iceberg, and remote relational tables.
- Explicit registration of MySQL and PostgreSQL connections and foreign tables.
- Read-only MySQL and PostgreSQL scans.
- Safe projection, exact-predicate, parameter, and bounded-limit pushdown.
- Documented mappings for common numeric, decimal, string, binary, date/time, boolean, JSON, and nullable types.
- Multi-CN MatrixOne deployments, with each first-phase remote relational scan assigned to one selected CN/session.
- Source-neutral cancellation, failure, quota, audit, and diagnostic behavior.

#### Non-goals

The following are not required by this feature:

- support for every possible data source;
- one physical operator or one low-level reader interface for all source categories;
- rewriting working file or Iceberg scan implementations solely for uniformity;
- remote `INSERT`, `UPDATE`, `DELETE`, `TRUNCATE`, or DDL;
- two-phase commit or cross-source transactional atomicity;
- a globally consistent snapshot across MatrixOne and external sources;
- arbitrary user-provided remote SQL strings;
- join, aggregate, window, `ORDER BY`, or general subplan pushdown in the first relational profile;
- pushing MatrixOne data or temporary tables into an external source;
- parallel remote-relational splits across multiple sessions in the first profile;
- automatic import of an entire remote catalog;
- transparent query-result caching;
- automatic materialization, refresh, or CDC;
- stream-table semantics, unbounded queries, or exactly-once stream processing;
- querying arbitrary REST/SaaS APIs as tables;
- analytical/search-provider implementations;
- identical feature behavior where source semantics fundamentally differ.

#### Acceptance criteria

##### Common product model

- [ ] Every federated object exposes its source/profile identity and data-location semantics.
- [ ] Source credentials and privileges are managed independently from table definitions.
- [ ] `EXPLAIN` shows source boundaries plus pushed and residual work.
- [ ] Runtime metrics distinguish external-source I/O and wait time from MatrixOne storage I/O.
- [ ] Existing file and Iceberg queries remain compatible and have no correctness or material performance regression.
- [ ] A MatrixOne query can combine native, Iceberg, and remote relational tables.
- [ ] Documentation clearly distinguishes live federation, external storage, materialization, and CDC.

##### Remote relational profile

- [ ] Users can create, inspect, update credential references for, and drop MySQL and PostgreSQL connections without exposing secrets.
- [ ] Users can create, inspect, query, and drop foreign tables backed by MySQL and PostgreSQL.
- [ ] Creating a foreign table discovers and validates its source schema.
- [ ] Projection, supported exact predicates, bound parameters, and bounded `LIMIT` are pushed down.
- [ ] Unsupported or inexact predicates remain residual and return the same result as non-pushed execution.
- [ ] NULL and documented common types have deterministic mappings for both providers.
- [ ] Incompatible source schema changes produce an actionable schema-drift error.
- [ ] Behavior and limitations are documented for multi-CN deployments.

##### Safety and unhappy paths

- [ ] Cancellation during connection, planning, execution, row reading, and decoding releases all owned resources.
- [ ] Downstream early stop releases the active statement/reader and pooled session.
- [ ] Connection refusal, authentication failure, permission failure, source restart, partial read, timeout, malformed value, and schema drift return deterministic errors without hanging a MatrixOne query.
- [ ] Repeated failure, cancellation, reset, and plan reuse do not grow goroutines, sessions, memory, or retained batches.
- [ ] Per-account and per-source concurrency, timeout, row, and byte limits are enforced where applicable.
- [ ] One account cannot access another account's connection, secret, or federated object.
- [ ] Plaintext credentials do not appear in any user-visible or internal diagnostic surface.

##### Validation

- [ ] MySQL and PostgreSQL pass the same remote-relational provider conformance suite.
- [ ] Differential tests compare pushed and non-pushed results.
- [ ] Integration tests run against real MySQL and PostgreSQL instances.
- [ ] Fault-injection tests cover cancellation, timeout, disconnect, partial reads, schema drift, and downstream early termination.
- [ ] Race/reuse tests cover prepare failure, read failure, EOF, cancellation, repeated reset/free, and cached-plan execution.
- [ ] Selective predicates demonstrate materially lower source rows and bytes than non-pushed scans.
- [ ] Regression tests cover native + Iceberg + remote-relational joins.

### Describe implementation you've considered

The following is a reference direction, not a required API, package, or DDL contract. A design document must be reviewed before implementation.

#### Separate the common product model from physical profiles

Use source-neutral catalog metadata for common behavior:

```text
source/connection identity
profile and provider type
external object identity
MatrixOne output schema
source type metadata
schema/capability version
credential reference
```

The optimizer should reason through source capabilities rather than source names. A planning result should distinguish:

- exact pushed operations;
- optional prefilters that retain a MatrixOne residual;
- unsupported operations;
- source estimates when available;
- source-specific physical scan specifications.

Do not add MySQL/PostgreSQL branches to the MatrixOne planner, compiler, or scan operator. Likewise, do not force file, Iceberg, relational, search, and stream systems into an abstraction that only fits one category.

#### Logical and physical planning

A dedicated logical federated/remote scan can contain canonical MatrixOne columns, expressions, limits, and source identity.

A provider-specific physical plan may contain a versioned, opaque, parameterized scan specification. Logical plans must not be raw concatenated remote SQL, and physical plans must contain credential references rather than credential values.

Existing file/Iceberg logical and physical paths may remain specialized. They should integrate with the common catalog, capability, governance, and observability contracts where doing so preserves their current behavior and performance.

#### Remote relational execution

Remote databases should not implement MatrixOne's native storage `Relation` contract.

A dedicated remote-relational scan execution should:

- resolve the source and current credential version at runtime;
- use one selected CN/session in the first profile;
- stream source rows into MatrixOne vectorized batches;
- produce at most one batch per operator `Call`;
- return normal end-of-stream without sending terminal pipeline signals from `Call`;
- close source execution idempotently on error, cancellation, `Reset`, and `Free`;
- preserve ownership and cleanup of partial and returned batches.

Parallel splits should be added only when a provider can define stable partitioning and compatible snapshot semantics.

#### Provider boundary

Provider-specific behavior should remain behind a federation provider/profile boundary:

- driver or protocol/session creation;
- source metadata discovery;
- source-to-MatrixOne type mapping;
- identifier, parameter, and expression translation;
- exact/pre-filter/unsupported pushdown decisions;
- physical request generation;
- statistics retrieval;
- cancellation;
- provider-error conversion.

The provider name should not reuse MatrixOne's existing SQL pipeline or stream `connector` concepts.

For capabilities already represented by the common model, adding another provider should require provider implementation and registration rather than planner/operator source-specific branches.

#### Validation strategy

One provider is insufficient to validate the remote-relational abstraction. MySQL and PostgreSQL intentionally exercise different naming, parameter, type, time, collation, metadata, snapshot, and cancellation semantics.

Both providers should use a shared conformance suite. A deterministic fake provider should cover lifecycle, cancellation, partial-batch, quota, schema-version, and fault-injection behavior.

Existing file and Iceberg suites should protect compatibility while common observability and governance are introduced.

### Documentation, Adoption, Use Case, Migration Strategy

Recommended uses:

- query externally managed data before building ingestion;
- evaluate MatrixOne against existing data;
- keep workflows operating during incremental migration;
- run low-frequency, selective cross-source analysis;
- access source-current data when copying is unnecessary;
- determine which external datasets should later be materialized or synchronized.

Not recommended:

- high-QPS, latency-critical application reads;
- repeated full scans of large external sources;
- large cross-source joins without selective pushdown;
- workflows requiring a globally consistent snapshot;
- cross-source transactional writes.

Documentation must make source dependency visible: federated query latency, availability, freshness, and consistency depend on the external source and network.

The feature should be introduced as an explicit preview. GA should require the common product, compatibility, relational-profile, unhappy-path, security, observability, and validation criteria above—not only successful happy-path scans.

### Additional information

- Related historical umbrella: #24033.
- This issue defines the product behavior, scope, and acceptance boundary.
- Exact interfaces, protobuf changes, catalog tables, DDL grammar, package layout, upgrade compatibility, rollout mechanics, and implementation subtasks belong in a follow-up design document.