[Discussion] Multi-Region Data Locality + Federated Execution

[Shekharrajak]

Hi team,

I am looking into guarantee where compute happening in region, result to FE, no raw data crosses borders and user should be able to see aggregated result.
Please let me know if we have some work/discussion already done like this :

Client
  │
  ▼
FE Leader  ──── Parses SQL → generates Logical Plan → Physical Query Plan 
  │
  │  PQP = DAG of fragments, each assigned to specific BE/CN nodes
  │
  ├─► Fragment 1 (Scan + partial agg) ──► BE/CN-A  [data local]
  ├─► Fragment 2 (Scan + partial agg) ──► BE/CN-B  [data local]
  ├─► Fragment 3 (Scan + partial agg) ──► BE/CN-C  [data local]
  │         │              │              │
  │         └──────────────┴──────────────┘
  │              Exchange nodes (HASH / BROADCAST / UNPARTITIONED)
  │                         │
  └─► Final Fragment (merge agg, top-N, ORDER BY) ──► single BE or FE itself
                            │
                         Result Set ──► back to FE ──► Client

[morningman]

Thanks for the detailed use case. Let me make sure I understand the requirement correctly.

One possible approach would be to implement this as a variant of federated query. We could potentially build it on top of the [ADBC Catalog](https://doris.apache.org/docs/dev/lakehouse/catalogs/adbc-catalog), which is planned to replace the current Doris Catalog for this type of cross-Doris query.

The basic idea would be:

• The optimizer pushes as much computation as possible, especially aggregation and joins, to the remote Doris cluster, so that only the processed/aggregated results are returned to the local cluster.
• For this use case, pushdown would need to be mandatory rather than just an optimization. If the optimizer cannot generate a remote plan that satisfies the data residency requirement—for example, if row-level/raw data would need to be returned from the remote cluster—the query should fail instead of falling back to fetching the raw data.

At the moment, the optimizer does not yet support aggregation pushdown or join pushdown for this federated query path, but these are capabilities we can consider adding.

Would this model satisfy your requirements? In particular, is it acceptable for a query to fail when Doris cannot generate a plan that keeps all raw data processing inside the remote region?

[Shekharrajak]

Would this model satisfy your requirements?
Yes, the only part is data present in one region will be isolated and should not transfer(no raw data movements over the network) . We are expecting Compute node to do the work in local region only and send the results back to leader FE node if possible.

is it acceptable for a query to fail when Doris cannot generate a plan that keeps all raw data processing inside the remote region?
Data could be in TBs but if it is configurable e.g. fail-fast in planning mode would help to know the cost of data movements across network and for simpler queries we will be getting results at least.

I need to check how we can simplify the table schemas so that query planning will be easier.