Repository navigation
[GLUTEN-12597][CORE] Remodel Kafka read onto ReadRel.ExtensionTable (Substrait 0.98) - #12841
Conversation
550be87 to
bedfb96
Compare
|
Run Gluten Clickhouse CI on x86 |
3 similar comments
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
bedfb96 to
0f4d98c
Compare
|
Run Gluten Clickhouse CI on x86 |
There was a problem hiding this comment.
Pull request overview
This PR continues the Substrait 0.98 ReadRel.read_type migration by removing Gluten’s legacy Kafka graft from the vendored Substrait proto and modeling ClickHouse Kafka reads via the official ReadRel.ExtensionTable path, carrying a Gluten-owned gluten.StreamKafka payload in google.protobuf.Any.
Changes:
- Remove
stream_kafka/ReadRel.StreamKafkafromalgebra.protoand rely onReadRel.ExtensionTable+Anytype discrimination for Kafka. - Introduce
gluten-ownedkafka.proto(gluten.StreamKafka) and update JVM producer + ClickHouse native parser accordingly. - Add a shared JVM helper to pack
ReadRel.ExtensionTableand adjust build wiring (protobuf dependency + CMake glob).
Reviewed changes
Copilot reviewed 15 out of 15 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| gluten-substrait/src/main/scala/org/apache/gluten/utils/SubstraitUtil.scala | Adds helper to pack Gluten payloads into ReadRel.ExtensionTable.detail. |
| gluten-substrait/src/main/scala/org/apache/gluten/utils/SubstraitPlanPrinterUtil.scala | Extends TypeRegistry so Any-carried Gluten messages can print/parse as JSON. |
| gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto | Makes ReadRel.read_type match upstream (removes Kafka graft; keeps official fields). |
| gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/StreamKafkaSourceNode.java | Emits split-info payload as ReadRel.ExtensionTable containing gluten.StreamKafka in Any. |
| gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/ReadRelNode.java | Replaces boolean Kafka discriminator with an extension_table.detail Any. |
| gluten-kafka/src/main/scala/org/apache/gluten/execution/MicroBatchScanExecTransformer.scala | Marks Kafka reads by stamping gluten.StreamKafka type into the in-plan extension_table. |
| gluten-kafka/pom.xml | Moves protobuf-java to provided scope for compilation/runtime needs. |
| gluten-core/src/main/resources/org/apache/gluten/proto/kafka.proto | Adds Gluten-owned StreamKafka message definition for Any payloads. |
| docs/developers/SubstraitModifications.md | Updates Substrait modification inventory (removes old StreamKafka entry). |
| cpp-ch/local-engine/proto/CMakeLists.txt | Makes proto glob reconfigure-aware (CONFIGURE_DEPENDS). |
| cpp-ch/local-engine/Parser/SerializedPlanParser.cpp | Ensures Kafka reads always consume a split-info entry (asserts extension_table present). |
| cpp-ch/local-engine/Parser/RelParsers/StreamKafkaRelParser.cpp | Parses split-info as ReadRel::ExtensionTable and unpacks gluten.StreamKafka from Any. |
| cpp-ch/local-engine/Parser/RelParsers/ReadRelParser.cpp | Detects Kafka reads by extension_table.detail().Is<gluten::StreamKafka>(). |
| backends-clickhouse/src/main/java/org/apache/spark/sql/execution/datasources/clickhouse/ExtensionTableNode.java | Reuses the new helper to build extension tables. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
0f4d98c to
54cd984
Compare
|
Run Gluten Clickhouse CI on x86 |
54cd984 to
3264a7f
Compare
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI |
3264a7f to
14f322f
Compare
|
Run Gluten Clickhouse CI on x86 |
…Substrait 0.98) Part of apache#12597. Stacked on apache#12832 (the iceberg_table half of the ReadRel read_type migration). Makes ReadRel.read_type verbatim upstream Substrait 0.98 by removing Gluten's last graft on it -- the `bool stream_kafka = 1000` discriminator and the nested `ReadRel.StreamKafka` message. read_type now matches upstream exactly, which never allocated field 1000, so the retired tag is not reserved. Gluten's Kafka streaming read is remodeled onto the official `extension_table = 8` path, the mechanism MergeTree and Range already use. The StreamKafka payload moves to a new Gluten-owned `kafka.proto` (`package gluten`, `org.apache.gluten.proto`), packed into a `google.protobuf.Any` and carried in `ReadRel.ExtensionTable.detail`. The native consumer discriminates a Kafka read by the detail's type_url (`detail().Is<gluten::StreamKafka>()`) -- mirroring the already-merged Velox Iceberg idiom (`enhancement().Is<gluten::IcebergReadExtension>()`). This Any-in-an-official-extension-field pattern is preferred over grafting new fields; `WriteRel.bucket_spec` (field 1000) remains the one legacy graft. Gluten plans are transient and the JAR + native library are generated from one proto source and ship together, so there is no wire-compatibility constraint. Two side effects: the Kafka split-info payload changes from a bare StreamKafka to a ReadRel.ExtensionTable wrapping it, and read_type -- previously set on every ReadRel by the unconditional stream_kafka flag -- is now left unset for non-Kafka scans (no native code reads read_type_case). A JAR and native library must therefore be rebuilt together. ClickHouse-only; Kafka has no Velox path. Generated-by: Claude Code (Claude Opus 4.8)
14f322f to
e31e44a
Compare
|
Run Gluten Clickhouse CI on x86 |
|
Rebased and fixed conflicts |
|
LGTM |
What changes are proposed in this pull request?
Part of #12597.
This makes
ReadRel.read_typeverbatim upstream Substrait 0.98 by removing Gluten's last graft on it -- thebool stream_kafka = 1000discriminator and the nestedReadRel.StreamKafkamessage -- and remodels Gluten's ClickHouse Kafka streaming read onto the officialextension_table = 8path, the same mechanism MergeTree and Range already use.read_typenow matches upstream exactly, which never allocated field 1000, so the retired tag is not reserved.The
StreamKafkapayload moves out ofalgebra.protointo a new Gluten-ownedkafka.proto(package gluten,org.apache.gluten.proto), packed into agoogle.protobuf.Anyand carried inReadRel.ExtensionTable.detail. The native consumer discriminates a Kafka read by the detail's type_url viadetail().Is<gluten::StreamKafka>()-- mirroring the already-merged Velox Iceberg idiomenhancement().Is<gluten::IcebergReadExtension>(). ThisAny-in-an-official-extension-field pattern is the preferred way to carry Gluten payloads going forward, rather than grafting new fields onto the vendored Substrait messages;WriteRel.bucket_spec(field 1000) remains the one legacy graft.Compatibility: this is a JVM-producer + ClickHouse-consumer change with no wire-compatibility constraint -- Gluten plans are transient, and the JAR and native library are generated from one proto source and ship together. Two side effects are worth calling out. The Kafka split-info payload changes from a bare
StreamKafkato aReadRel.ExtensionTablewrapping it; andread_type, which the unconditionalstream_kafkaflag used to set on everyReadRel, is now left unset for non-Kafka scans (no native code readsread_type_case). A JAR and native library must therefore be rebuilt together. This path is ClickHouse-only -- Kafka has no Velox path.How was this patch tested?
Locally: standalone
protocconfirms the post-deleteread_typeoneof is well-formed and that the relocatedgluten.StreamKafkagenerates cleanly under both native codegen styles (flatkafka.pb.h). The JVM producer builds green --mvn -Pspark-3.5 -Pkafka -pl gluten-core,gluten-substrait,gluten-kafka -am -DskipTests clean install(-Pkafkais required;cleanclears stale generated.javafrom the removed message) -- with no lingering references toReadRel.StreamKafka/setStreamKafkaand scalastyle/checkstyle passing in-phase.The native ClickHouse
local-enginebuild (ReadRelParser.cpp,StreamKafkaRelParser.cpp, generatedkafka.pb.*) needs the Linux/Docker toolchain and is verified here by inspection only. The Kafka runtime suites (GlutenKafkaScanSuite/ClickhouseGlutenKafkaScanSuite) require-Pkafkaplus an externallocalhost:9092broker and are not run by any in-repo GitHub Actions job, so the end-to-end Kafka path is not exercised in CI. A maintainer running-Pkafkaagainst a broker before merge would be the real end-to-end check; the JVMtoProtobufand the CH deserialize are kept as literal mirrors of the existing extension-table readers to minimize that risk.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
🤖 Generated with AI