Tracking: Ad-hoc(batch) ingestion

[st1page]

We will enhance the ad-hoc ingestion capability in subsequent releases, with the expectation that it will eventually be possible for users to read ad-hoc data if it is persisted on an external system.

Streaming storage

for the streaming storage, the predicate pushdown with the "offset" is required

• kafka
  • select from source
  • TVF
• pulsar
  • select from source
  • TVF

lake

• iceberg
  • select from source
  • Support Iceberg TVF #18637
  • optimization
    • column pruning
    • predicate pushdown

file source(object store)

• select from source
• TVF
  • only support S3 currently
• optimization
  • column pruning
  • predicate pushdown

Database

Currently we only support Create table with primary key on the CDC connector. To support it, we need design and introduce new syntax that CREATE source with CDC connector. In that case, the source can only be ad-hoc queried.

• PG
  • select from source
  • Support PG TVF #18636
  • optimization
    • column pruning
    • range predicate pushdown
    • lookup
• MySQL
  • select from source
  • TVF
  • optimization
    • column pruning
    • range predicate pushdown
    • lookup
• MongoDB
  • select from source
  • TVF
  • optimization
    • column pruning
    • range predicate pushdown
    • lookup

[kwannoel]

Hi, I will help with this issue, starting with TVFs.

[xxchan]

Have we reached consensus to support TVFs? To me, their use cases are duplicated with Sources, so they seem to be unnecessary.

I’d like to see rationales and examples where they are more useful than sources before adding them

[st1page]

Have we reached consensus to support TVFs? To me, their use cases are duplicated with Sources, so they seem to be unnecessary.

I’d like to see rationales and examples where they are more useful than sources before adding them

• The original discussion of the discussion here https://risingwave-labs.slack.com/archives/C036F5Z3EMD/p1720166956241589
• Additionally, One case is for the Ad-hoc ingestion from databases. Currently we only support the CDC table and can not create a source on a external databases's table. So only TVF is clear defined method to do ad hoc ingest from Databases. We can refer to the grammer of duckDB for the cases
  • https://duckdb.org/docs/extensions/postgres.html#the-postgres_query-table-function
  • https://duckdb.org/docs/extensions/mysql#the-mysql_query-table-function

[xxchan]

Thanks for the explanation!

Currently we only support the CDC table and can not create a source on a external databases's table.

Makes me think whether also related with other shared source e.g., Kafka?

We can refer to the grammer of duckDB for the cases

Compared with duckDB

• They don't have source at all. So it might be a little different
• Their syntax contains a ATTACH, which looks like CREATE CONNECTION we might have in the future. So maybe we should design that first.

ATTACH 'dbname=postgresscanner' AS postgres_db (TYPE POSTGRES);
SELECT * FROM postgres_query('postgres_db', 'SELECT * FROM cars LIMIT 3');

[st1page]

Currently we only support the CDC table and can not create a source on a external databases's table.

Makes me think whether also related with other shared source e.g., Kafka?

The issue is not related to "shared" but it is beacuse the CDC source contains multiple tables' changes. Actually that is a "CONNECTION"

[st1page]

Compared with duckDB

• They don't have source at all. So it might be a little different
• Their syntax contains a ATTACH, which looks like CREATE CONNECTION we might have in the future. So maybe we should design that first.

ATTACH 'dbname=postgresscanner' AS postgres_db (TYPE POSTGRES);
SELECT * FROM postgres_query('postgres_db', 'SELECT * FROM cars LIMIT 3');

Agree with that. cc @chenzl25. do we have plan to simplify the syntax of the TVF with connection?

[chenzl25]

After the connection is supported, in my mind connection can be used in TVF directly like:

• read_parquet(s3_connection, 's3://bucket/path/xxxx.parquet')
• read_csv(s3_connection, 's3://bucket/path/xxxx.parquet')
• read_json(s3_connection, 's3://bucket/path/xxxx.parquet')
• iceberg_scan(iceberg_connection, 'database_name.table_name')
• postgres_query(pg_connection, 'select * from t')
• mysql_quert(my_connection, 'select * from t')

Connections contain the necessary information to allow TVF to query the external system.
I think @tabVersion will support Connection in this Q.