ahmedabu98 opened a new issue, #41313:
URL: https://github.com/apache/beam/issues/41313
### What would you like to happen?
Today when you run a SELECT on a streaming (unbounded) source, like a Kafka
or Pub/Sub topic, in the Beam SQL shell, you never see a row. The shell runs
the query on the DirectRunner and collects its rows in memory. It only prints
them once the pipeline finishes, and a streaming pipeline never finishes. So
the shell sits there until you cancel the query, and then it throws away the
rows it collected.
### Proposal
For unbounded SELECTs, show rows while the pipeline produces them, in a
full-screen view much like the changelog mode in Flink's SQL client:
- The column names stay pinned at the top, and new rows come in at the
bottom.
- An `op` column shows each row's change kind: `+I` for an insert, `-U` and
`+U` for the row before and after an update, and `-D` for a delete.
- You can scroll back through the newest 10,000 rows with the arrow keys,
PgUp and PgDn, Home and End, or vi keys. Scrolling up pauses the view, and End
goes back to following new rows.
- `q`, Esc or Ctrl+C closes the view and cancels the pipeline.
Only unbounded SELECTs change. Bounded queries, `LIMIT` queries and INSERTs
work as before. Like any SELECT in the shell, this needs the DirectRunner.
### Examples
```sql
CREATE EXTERNAL TABLE orders (id BIGINT, item VARCHAR, amount DOUBLE)
TYPE kafka
LOCATION 'localhost:9092/orders'
TBLPROPERTIES '{"format": "json"}';
SELECT * FROM orders WHERE amount > 100;
```
The view opens right away, and rows show up as messages arrive on the topic:
```
Beam SQL live results RUNNING 00:42
op │ id │ item │ amount
─────┼──────┼──────────┼───────
+I │ 101 │ keyboard │ 149.0
+I │ 102 │ monitor │ 329.5
+I │ 103 │ webcam │ 119.9
3 rows · following ↑↓ PgUp PgDn Home End ←→ scroll · q quit
```
On an Iceberg table, a streaming `CHANGES()` read shows each change with its
kind:
```sql
SELECT * FROM CHANGES(customers) WITH ('streaming' = true);
```
```
Beam SQL live results RUNNING 01:15
op │ id │ name │ tier
─────┼──────┼───────┼───────
+I │ 1 │ alice │ silver
+I │ 2 │ bob │ gold
-U │ 1 │ alice │ silver
+U │ 1 │ alice │ gold
-D │ 2 │ bob │ gold
5 rows · following ↑↓ PgUp PgDn Home End ←→ scroll · q quit
```
Customer 1 moved from silver to gold, and customer 2 was deleted.
### Where `op` comes from
The `op` column shows each element's `ValueKind` metadata. It doesn't read a
change-type column. Iceberg's `CHANGES()` sets the kind on every element it
reads, which is where the `-U`, `+U` and `-D` rows above come from. If a source
doesn't set a kind, it defaults to `INSERT`. That's why every Kafka row is
`+I`. Filters and projections keep the kind. Aggregations build new rows, so
their results show as `+I`.
### Plan
This splits into two PRs:
- [ ] In the sql module, an unbounded SELECT returns its rows through JDBC
as they arrive. The shell already prints JDBC rows one at a time, so this alone
gets streaming rows on screen, just without the `op` column. This PR also adds
a hook the shell can use to show the rows itself.
- [ ] In the jdbc module, the shell registers the full-screen view when it
runs in a terminal. Without a terminal, for example when output is piped, rows
keep printing the plain way.
### Issue Priority
Priority: 2 (default / most feature requests should be filed as P2)
### Issue Components
- [ ] Component: Python SDK
- [ ] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Prism Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
