feat(table): expose `registerStreamingTable` for push-mode batch ingest
Personne n'a encore pris cette issue.
Évaluation
- Difficulté
- 5/5
- Temps estimé
- Plus d'une semaine
- Accessibilité débutants
- 38/100
Piste de recherche
Commencez par le chemin existant SessionContext.registerTable/TableProvider de PR #65 et le point d’entrée des implémentations antérieures rust/src/api.rs:572, register_partition_stream. Suivez l’intégration de Data.exportVectorSchemaRoot et de StreamingTable/PartitionStream décrite dans l’issue ; le travail est considéré comme terminé lorsque les producteurs peuvent écrire, se fermer ou échouer concurremment avec les lectures des requêtes, avec la backpressure, l’annulation, la propagation des erreurs et la sémantique de single-scan documentées et appliquées.
Rédigé par le modèle d'indexation à partir du texte de l'issue.
Description
Is your feature request related to a problem or challenge?
PR #65 shipped a Java-implemented TableProvider and SessionContext.registerTable(String, TableProvider). That covered the pull shape: DataFusion calls scan(BufferAllocator) and reads the returned ArrowReader.
But it does not cover the push shape that event-driven batch sources need:
- A coordinator that reduces over shard responses arriving incrementally -- the producer can't materialise an
ArrowReaderbecause the next batch hasn't arrived yet. - A Flight stream feeding into a query -- same problem; the producer is event-driven.
- Any in-process producer that emits batches as side-effects of other work and doesn't know in advance how many will arrive.
To bridge these into PR #65 today, callers have to write a BlockingArrowReader adapter that buffers pushed batches and serves them through the pull interface. That's a serialisation point: the producer blocks waiting for loadNextBatch() to be called, or DataFusion blocks waiting for the next batch -- the two ends can never run truly concurrently. The adapter also has to invent its own backpressure semantics, error propagation, end-of-stream signalling, and thread-safety story.
DataFusion itself solves this on the Rust side with StreamingTable + PartitionStream plus an mpsc channel: producer pushes Result<RecordBatch> into the sender, the consumer (DataFusion's StreamingTableExec) polls the receiver as part of normal query execution. The two ends decouple via the channel buffer, with the runtime providing backpressure and cancellation propagation.
Describe the solution you'd like
One new method on SessionContext returning a TableSink:
TableSink sink = ctx.registerStreamingTable("shard_results", schema, capacity);
// Producer thread (any thread, including outside any Tokio runtime):
try {
while (hasMoreInput()) {
sink.write(batch); // backpressures when channel is full
}
sink.close(); // EOF: queries see end-of-stream cleanly
} catch (Throwable t) {
sink.fail(t); // signal error: queries see RuntimeException
}
public final class TableSink implements AutoCloseable {
void write(VectorSchemaRoot batch); // exports via Data.exportVectorSchemaRoot
void close(); // EOF
void fail(Throwable cause); // error propagated to readers
}
After registration the table can be referenced like any other registered table:
DataFrame df = ctx.sql("SELECT count(*) FROM shard_results");
ArrowReader r = df.executeStream(allocator);
// Producer thread continues writing as r.loadNextBatch() drains.
Single-scan semantics. The registered table can only be queried once. After that scan completes (or is cancelled), the sink is no longer usable and the table cannot be re-scanned. This is the natural semantic for an event-driven producer -- the data is consumed as it arrives -- and matches what every downstream Substrait/Calcite plan that uses streaming tables already assumes. Documented loudly on registerStreamingTable's Javadoc; trying to re-execute against the same registration throws.
This is intentional. Re-scannable streaming would require buffering every batch internally, which defeats the streaming use case. Callers who need to re-scan the same data should use the existing registerTable / SimpleTableProvider pull shape (PR #65) instead.
Describe alternatives you've considered
BlockingArrowReaderadapter on top of PR #65'sregisterTable. What every caller currently has to hand-roll. It works but pushes the channel + backpressure + EOF + error story onto every embedder. Bridging via the upstream-canonicalStreamingTableshape is strictly less code and gets cancellation propagation for free.- Backpressure-free
try_write. A non-blocking variant that returnsfalsewhen the channel is full. Easy to add later as a follow-up if anyone wants it; not in scope here. Defaultwriteblocks, which is the contract every Java I/O caller expects. - Reuse PR #65's
TableProviderinterface and wrap mpsc internally. Considered. The problem: PR #65'sscan(BufferAllocator) -> ArrowReaderreturns synchronously, so a mpsc-backed implementation has to block onloadNextBatch()waiting for the producer -- exactly the serialisation point we're trying to avoid. Going direct toStreamingTable+PartitionStreamis the right layer.
Additional context
- The OpenSearch backend's
rust/src/api.rs:572register_partition_streamis the prior-art template; it does almost exactly this. The Java side there uses a hand-rolled FFM bridge (sender_send) that can be replaced with this surface as soon as it lands.
- Langage dominant
- Java
- Étoiles
- 32
- Forks
- 12
- Métriques de merge des PR
- Aucune PR mergée en 30 j
Guide de contribution
Ouvrir le guide de contribution
Par où commencer
- Lisez l'issue en entier, puis le guide de contribution du projet.
- Signalez en commentaire que vous la prenez — cela évite que deux personnes fassent le même travail.
- Forkez le dépôt et travaillez sur une branche.
- Ouvrez une pull request qui référence le numéro de l'issue.
Autres issues de apache/datafusion-java
-
Difficulté 5/5 Plus d'une semaine Accessibilité débutants 35/100
apache/datafusion-java#116 ·
-
Difficulté 5/5 Plus d'une semaine Accessibilité débutants 25/100
apache/datafusion-java#112 ·
-
enhancement
Difficulté 5/5 Plus d'une semaine Accessibilité débutants 42/100
apache/datafusion-java#96 ·
-
Create first release Ouverteenhancement
Difficulté 4/5 3-5 jours Accessibilité débutants 35/100
apache/datafusion-java#86 · 3 commentaires ·
-
enhancement
Difficulté 5/5 Plus d'une semaine Accessibilité débutants 45/100
apache/datafusion-java#68 ·
Toutes les issues de apache/datafusion-java
Issues similaires
-
bug untriaged
Difficulté 2/5 1-3 heures Accessibilité débutants 84/100
opensearch-project/ml-commons#5094 ·
-
bug
Difficulté 2/5 1-3 heures Accessibilité débutants 85/100
-
emitter:client:csharp feature
Difficulté 2/5 1-3 heures Accessibilité débutants 72/100
-
affects/8.10 affects/8.9 component/clients kind/bug likelihood/mid severity/mid
Difficulté 2/5 1-3 heures Accessibilité débutants 78/100
-
Two open-case totals on one screen: the Programs tile says 15,858 and the nav badge says 15,868 Ouvertebug frontend maui-pilot
Difficulté 2/5 1-3 heures Accessibilité débutants 72/100