Trino 3 — Connector & Catalog

Ở Trino 2 — Kiến trúc ta đã thấy connector là "cửa" mà worker gọi qua để đọc dữ liệu, và nó hiện thực giao diện SPI (Service Provider Interface). Bài này trả lời câu hỏi thực dụng hơn: có những connector nào, mỗi cái đọc được nguồn gì, cấu hình ra sao, và giới hạn ở đâu. Đây là kiến thức bạn cần khi phải cắm Trino vào một hệ sinh thái dữ liệu ngân hàng thật — nơi cùng lúc tồn tại data lake trên object storage, các core banking chạy PostgreSQL/Oracle, và Kafka bắn sự kiện realtime.

Connector là gì

Connector là một plugin của Trino, mỗi cái biết cách nói chuyện với một loại nguồn dữ liệu. Nó chịu trách nhiệm ba việc (đã nêu ở bài kiến trúc): cung cấp metadata (danh sách schema/bảng/cột, kiểu, thống kê), liệt kê split (cắt dữ liệu thành đơn vị đọc song song), và truy cập dữ liệu + pushdown (đọc, đồng thời đẩy bớt filter/cột xuống nguồn nếu nguồn đỡ được). Vì mọi connector nói cùng một "ngôn ngữ" SPI, engine Trino không cần biết bên dưới là Hive hay MongoDB — nó chỉ gọi giao diện, connector lo phần dịch sang giao thức của nguồn.

Điểm cần khắc cốt: Trino không sở hữu dữ liệu. Connector chỉ là lớp đọc/ghi tại chỗ — chính điều này cho phép Trino truy vấn liên hợp (federation) nhiều nguồn trong cùng một câu SQL.

Mô hình định danh: Catalog → Schema → Table

Trino tổ chức không gian tên ba cấp:

• Catalog: một cấu hình cụ thể của một connector trỏ tới một nguồn. Đây là điểm hay bị hiểu nhầm — catalog KHÔNG phải là "loại connector", mà là một instance đã cấu hình. Bạn có thể có hai catalog cùng dùng connector PostgreSQL nhưng trỏ tới hai server khác nhau (ví dụ core_prod và core_dr). Mỗi catalog tương ứng một file .properties.
• Schema: nhóm bảng trong catalog. Với RDBMS, schema ánh xạ thẳng sang schema/database của nguồn; với Hive/Iceberg, schema ứng với "database" trong metastore.
• Table: bảng, view, hoặc topic (Kafka) — tùy connector.

Tên bảng đầy đủ (fully-qualified) là catalog.schema.table. Ví dụ lakehouse.core.transactions nghĩa là: catalog lakehouse (một cấu hình Iceberg), schema core, bảng transactions. Bạn có thể SET SESSION hoặc dùng USE catalog.schema để đặt mặc định rồi viết tên bảng ngắn gọn.

Một câu SQL duy nhất có thể tham chiếu lakehouse.core.transactions, core.public.products và events.default.card_swipe cùng lúc — engine sẽ gọi ba connector khác nhau và ghép kết quả (chi tiết ở Trino 4 — SQL Federation).

Cấu hình một catalog: properties file

Mỗi catalog là một file etc/catalog/<tên_catalog>.properties. Tên file (bỏ đuôi) chính là tên catalog trong SQL. Cấu trúc tối thiểu: dòng connector.name chọn connector, phần còn lại là tham số của connector đó.

Ví dụ catalog PostgreSQL tên core (file etc/catalog/core.properties):

# etc/catalog/core.properties  — minh họa, không phải SQL
connector.name=postgresql
connection-url=jdbc:postgresql://core-db.ncb.internal:5432/corebank
connection-user=trino_ro
connection-password=${ENV:CORE_DB_PWD}
# bật pushdown gộp (aggregate) xuống Postgres
postgresql.experimental.enable-string-pushdown-with-collate=true

Catalog Iceberg tên lakehouse (file etc/catalog/lakehouse.properties):

# etc/catalog/lakehouse.properties  — minh họa
connector.name=iceberg
iceberg.catalog.type=hive_metastore
hive.metastore.uri=thrift://hms.ncb.internal:9083
fs.native-s3.enabled=true
s3.endpoint=https://minio.ncb.internal
s3.region=ap-southeast-1

Trên các bản Trino gần đây, thêm/sửa catalog có thể làm động qua catalog management API mà không cần restart cả cụm; các bản cũ thì phải khởi động lại từng node để nạp file mới.

Các connector quan trọng

Hive (Hive Metastore + object storage) — kinh điển

Connector Hive là "cửa ngõ data lake" cổ điển: đọc/ghi file cột (Parquet, ORC) và cả text/CSV/JSON trên object storage (S3, GCS, Azure Blob) hoặc HDFS. Nó tra Hive Metastore (HMS) để lấy vị trí thư mục, danh sách partition, schema cột. Điểm mạnh: đọc song song cực tốt (mỗi file/khối là một split), pushdown filter theo partition (partition pruning) và theo cột. Điểm yếu: layout Hive không có ACID thực sự — sửa/xóa dòng lẻ rất khó, dễ gặp file nhỏ, không có time travel.

Iceberg & Delta Lake — table format hiện đại (lakehouse)

Đây là bước tiến giải quyết đúng các điểm yếu trên. Apache Iceberg và Delta Lake là các table format: vẫn lưu dữ liệu dạng Parquet trên object storage, nhưng thêm một lớp metadata có phiên bản (manifest, snapshot) mô tả chính xác file nào thuộc bảng ở thời điểm nào. Nhờ đó có:

• ACID transaction trên data lake — INSERT/UPDATE/DELETE/MERGE an toàn, đọc-ghi đồng thời không thấy dữ liệu dở dang.
• Time travel — truy vấn bảng "như tại" một snapshot/timestamp trong quá khứ (FOR VERSION AS OF / FOR TIMESTAMP AS OF).
• Schema evolution — thêm/xóa/đổi tên cột an toàn, không phải viết lại toàn bộ dữ liệu.
• Hidden partitioning (Iceberg) — partition theo biểu thức mà truy vấn không cần biết, tránh lỗi partition-key thủ công.

Đây chính là nền của kiến trúc lakehouse (xem data-engineering-07-lakehouse và cách khai thác ở Trino 5 — Lakehouse Query). Trino hỗ trợ Iceberg rất trưởng thành (đọc, ghi, DDL, maintenance như OPTIMIZE, expire snapshot); Delta Lake được hỗ trợ đọc tốt và ghi ở mức tùy phiên bản.

PostgreSQL / MySQL / JDBC — đọc database nghiệp vụ

Nhóm connector JDBC-based (PostgreSQL, MySQL, Oracle, SQL Server, ClickHouse, Redshift…) cho phép Trino đọc thẳng các database nghiệp vụ đang chạy — core banking, CRM, hệ thống thẻ — để join dữ liệu lịch sử trong lake với dữ liệu "nóng" đang sống trong OLTP mà không cần copy trước. Chúng pushdown mạnh: filter, một số aggregate, join giữa các bảng cùng nguồn, LIMIT, TOP N… được đẩy hẳn xuống DB gốc. Lưu ý vận hành: query nặng qua Trino vẫn chạy trên chính DB production, nên dùng user chỉ đọc và cân nhắc replica để không đè tải hệ thống lõi.

Kafka — đọc topic như bảng

Connector Kafka ánh xạ mỗi topic thành một bảng, mỗi message thành một dòng. Bạn khai báo schema (Avro/JSON/Protobuf/raw) để Trino giải mã payload thành cột. Điều này cho phép chạy SQL ad-hoc trên luồng sự kiện gần realtime — ví dụ đếm giao dịch thẻ theo phút trực tiếp từ topic. Giới hạn: đây là snapshot tại thời điểm quét (offset hiện có), không phải streaming liên tục; và pushdown chủ yếu giới hạn ở lọc theo partition/offset/timestamp, không đẩy được filter tùy ý xuống broker.

Elasticsearch, MongoDB, ClickHouse

• Elasticsearch: coi index như bảng, đẩy được nhiều predicate (full-text, range) xuống ES; hữu ích để join log/search với dữ liệu nghiệp vụ (đối chiếu es-01-overview).
• MongoDB: đọc collection như bảng, map document (kể cả nested) sang cột; pushdown filter cơ bản.
• ClickHouse: connector JDBC tới ClickHouse — hữu ích khi cần ghép dữ liệu từ một kho OLAP tốc độ cao vào truy vấn liên hợp (xem ch-01-overview).

Bảng tóm tắt đặc điểm:

Metastore: nơi lưu metadata bảng lake

Connector đọc file trên object storage cần biết bảng gồm những file nào, cột gì, partition ra sao — thông tin đó nằm trong metastore. Ba lựa chọn:

• Hive Metastore (HMS): dịch vụ Thrift kinh điển, lưu metadata trong một RDBMS (thường chính là PostgreSQL/MySQL). Dùng chung được cho Hive và Iceberg.
• AWS Glue Data Catalog: metastore managed trên AWS, tương thích API Hive — bỏ được gánh nặng tự vận hành HMS.
• REST catalog / JDBC catalog (Iceberg): các kiểu catalog mới của Iceberg, không phụ thuộc HMS.

Điểm cần nhớ: metastore chỉ lưu metadata (con trỏ), dữ liệu thật vẫn nằm trên object storage. Metastore chết thì không truy vấn được bảng lake dù file vẫn còn nguyên — nên nó là thành phần cần HA.

Predicate & projection pushdown — mấu chốt hiệu năng và chi phí

Đây là kiến thức tối quan trọng cho cả tốc độ lẫn hóa đơn cloud:

• Projection pushdown (đẩy cột): chỉ đọc đúng cột mà query cần. Với định dạng cột như Parquet/ORC, đọc 3 cột trong bảng 200 cột nghĩa là I/O giảm gần 60 lần — vì Parquet lưu theo cột nên bỏ qua được phần không cần.
• Predicate pushdown (đẩy filter): đẩy điều kiện WHERE xuống gần nguồn nhất có thể. Có nhiều mức:
  • Với file cột: dùng thống kê min/max mỗi row-group để bỏ qua cả khối không thỏa (predicate/statistics pruning), và partition pruning để bỏ qua cả thư mục partition.
  • Với JDBC/RDBMS: dịch WHERE thành mệnh đề SQL gửi thẳng xuống DB gốc — DB chỉ trả về dòng đã lọc, cắt mạnh dữ liệu qua mạng.
  • Với Kafka/Elasticsearch: đẩy được ở mức giới hạn (offset/timestamp, hoặc query DSL của ES).

Vì sao "cực quan trọng"? Trên object storage tính tiền theo lượng byte quét, và mọi byte đọc thừa đều phải chảy qua mạng vào worker rồi ăn RAM. Một query thiết kế tốt (chỉ cột cần + filter đúng partition) có thể quét vài GB thay vì vài TB. Mức pushdown phụ thuộc connector: JDBC đẩy được nhiều (kể cả aggregate/join cùng nguồn), Hive/Iceberg đẩy filter + partition + cột, còn Kafka gần như chỉ đọc rồi lọc tại Trino. Dùng EXPLAIN để xác minh filter có thực sự được đẩy xuống hay không (kỹ hơn ở Trino 6 — Performance Tuning).

Type mapping

Mỗi nguồn có hệ kiểu riêng; connector phải ánh xạ sang kiểu chuẩn của Trino. Vài điểm hay vấp:

• Số thập phân tài chính: giữ ở DECIMAL(p,s) xuyên suốt — đừng để rơi về DOUBLE (mất chính xác tiền tệ). Postgres numeric → DECIMAL khớp tốt.
• Thời gian: Postgres timestamptz map sang TIMESTAMP WITH TIME ZONE; cần thống nhất kiểu có/không timezone để tránh lệch giờ.
• Kiểu không có ở Trino (ENUM, kiểu hình học, một số kiểu Mongo) có thể bị map thành VARCHAR hoặc bỏ qua — kiểm tra bằng DESCRIBE catalog.schema.table. Postgres jsonb thường map sang JSON (đối chiếu pg-03-jsonb).

Giới hạn từng connector

Không connector nào toàn năng, và biết trước giới hạn giúp tránh thiết kế sai:

• Write: Iceberg ghi đầy đủ; Hive ghi được nhưng không xóa/sửa dòng lẻ dễ dàng; Kafka/Elasticsearch về cơ bản chỉ đọc; JDBC ghi tùy cấu hình và tùy quyền user.
• Pushdown: JDBC đẩy mạnh nhất (filter + nhiều aggregate + join cùng nguồn); Hive/Iceberg đẩy filter/partition/cột; Kafka gần như không đẩy filter tùy ý.
• Transaction xuyên nguồn: Trino KHÔNG có giao dịch phân tán 2PC giữa các catalog khác nhau — bạn không thể INSERT nguyên tử đồng thời vào Iceberg và PostgreSQL trong một transaction.
• Độ trễ nguồn: query qua JDBC nhanh hay chậm phụ thuộc chính DB gốc; Kafka phụ thuộc số partition; ES phụ thuộc cluster tìm kiếm.

SQL minh họa qua connector

Trên Trino thật (cú pháp fully-qualified, KHÔNG chạy trong sandbox — chỉ minh họa), một truy vấn liên hợp ghép lake + core + events:

-- Trino — minh họa, KHÔNG chạy trong sandbox PostgreSQL
SELECT p.product_name,
       count(*)      AS so_gd,
       sum(t.amount) AS tong_tien
FROM lakehouse.core.transactions t                 -- Iceberg trên object storage
JOIN core.public.products         p ON t.product_id = p.id   -- PostgreSQL core
WHERE t.created_at >= TIMESTAMP '2026-06-01 00:00:00'        -- predicate pushdown xuống Iceberg
GROUP BY p.product_name
ORDER BY tong_tien DESC;

Câu này minh họa hai connector khác nhau trong một SQL, và filter theo created_at được đẩy xuống Iceberg để cắt partition trước khi dữ liệu về worker.

Để làm quen dạng truy vấn qua JDBC connector (Trino đọc một RDBMS như đọc bảng thường), ta có thể chạy trên sandbox PostgreSQL một câu tổng hợp tương đương — ghép 5 bảng, tính doanh số theo phòng ban của nhân viên phụ trách khách hàng theo thành phố (minh họa cùng dạng aggregate mà một JDBC connector sẽ pushdown xuống nguồn):

-- ▶ Chạy được
SELECT d.name AS department,
       c.city,
       COUNT(DISTINCT t.id)    AS so_giao_dich,
       SUM(t.amount)           AS tong_gia_tri,
       AVG(a.balance)          AS so_du_tb
FROM transactions t
JOIN accounts    a ON t.account_id = a.id
JOIN customers   c ON a.customer_id = c.id
JOIN departments d ON c.city = d.name          -- ghép minh họa city ↔ tên phòng ban
JOIN employees   e ON e.department_id = d.id
GROUP BY d.name, c.city
ORDER BY tong_gia_tri DESC;

Câu SELECT này chạy được trên sandbox: nó chỉ dùng 5 bảng cho phép, và về mặt logic giống hệt điều Trino làm khi một JDBC connector đọc và pushdown aggregate xuống PostgreSQL.

Use case thực tế

Bối cảnh (NCB): Khối Phân tích cần một cửa truy vấn duy nhất phục vụ đội chống gian lận (fraud). Dữ liệu nằm rải ba nơi: (1) lịch sử giao dịch 24 tháng trên lakehouse Iceberg (object storage MinIO, ~2 tỷ dòng), (2) danh mục khách hàng và trạng thái rủi ro đang sống trong PostgreSQL core (cập nhật liên tục), và (3) sự kiện quẹt thẻ realtime trên Kafka. Trước đây mỗi lần điều tra, cán bộ phải xin trích xuất từ ba đội khác nhau, mất 1–2 ngày.

Giải pháp bằng Trino, đúng theo bài này:

1. Dựng ba catalog trên cùng một cụm Trino: lakehouse (connector Iceberg, metastore là HMS), core (connector PostgreSQL, user chỉ đọc trỏ vào read replica để không đè core prod), events (connector Kafka, khai báo schema Avro cho topic card_swipe).
2. Một câu SQL fraud ghép cả ba: join giao dịch lịch sử (Iceberg) với trạng thái rủi ro hiện tại (Postgres) và quẹt thẻ trong 15 phút gần nhất (Kafka).
3. Tối ưu pushdown: filter created_at >= now - interval '90' day được đẩy xuống Iceberg (partition pruning → chỉ quét ~90 ngày thay vì 24 tháng); SELECT chỉ 6 cột cần thiết (projection pushdown) → I/O trên object storage giảm mạnh; filter risk_flag = 'HIGH' đẩy thẳng xuống Postgres replica.

Kết quả: một truy vấn điều tra chạy trong 8–20 giây thay vì 1–2 ngày chờ trích xuất thủ công; lượng dữ liệu Iceberg quét mỗi query giảm từ ~1,8 TB xuống ~40 GB nhờ partition + projection pushdown, kéo theo chi phí đọc object storage giảm tương ứng. Đội fraud tự phục vụ, không còn phụ thuộc ba đội nguồn.

Ghi nhớ

• Connector là plugin (SPI) biết đọc một loại nguồn; nó cung cấp metadata + split + truy cập dữ liệu, nhưng Trino không sở hữu dữ liệu.
• Định danh ba cấp catalog.schema.table; catalog = một cấu hình connector trỏ tới một nguồn cụ thể (không phải "loại connector") — mỗi catalog là một file .properties.
• Connector nền tảng: Hive (Parquet/ORC trên S3/HDFS + Hive Metastore, kinh điển nhưng không ACID/time travel); Iceberg & Delta Lake (table format hiện đại → lakehouse, có ACID/time travel/schema evolution).
• JDBC (PostgreSQL/MySQL/Oracle…) đọc DB nghiệp vụ và pushdown mạnh nhất (filter + aggregate + join cùng nguồn); Kafka đọc topic như bảng nhưng chỉ là snapshot lúc quét; còn có Elasticsearch, MongoDB, ClickHouse.
• Metastore (Hive Metastore / AWS Glue / REST catalog) lưu metadata bảng lake — con trỏ tới file, không phải dữ liệu; cần HA vì mất metastore là mất truy vấn.
• Projection + predicate pushdown là chìa khóa hiệu năng và chi phí: chỉ đọc cột cần và đẩy filter/partition xuống nguồn để cắt byte quét — dùng EXPLAIN để xác minh filter có được đẩy xuống thật không.
• Type mapping cần lưu ý tiền tệ (DECIMAL, tránh DOUBLE), timestamp có/không timezone, và kiểu lạ bị map về VARCHAR.
• Giới hạn: mức write và pushdown khác nhau theo connector; Trino không có transaction phân tán 2PC giữa các catalog khác nhau.
• Một cụm Trino ngân hàng điển hình cắm đồng thời data lake (Iceberg) + Postgres core + Kafka — ghép chúng trong một SQL chính là truy vấn liên hợp (Trino 4 — SQL Federation).