Copilot commented on code in PR #13236:
URL: https://github.com/apache/gluten/pull/13236#discussion_r4326562174
##########
cpp/velox/substrait/SubstraitToVeloxPlan.cc:
##########
@@ -1511,6 +1521,58 @@ core::PlanNodePtr
SubstraitToVeloxPlanConverter::constructCudfValueStreamNode(
}
#endif
+#ifdef ENABLE_KAFKA
+core::PlanNodePtr
SubstraitToVeloxPlanConverter::constructKafkaStreamNode(const
::substrait::ReadRel& readRel) {
+ VELOX_USER_CHECK(!readRel.has_filter(), "Filter pushdown is not supported
for Kafka reads.");
+ // Like other table scans, a Kafka scan consumes the next split info. The
JVM sends it as a serialized
+ // ReadRel.ExtensionTable containing a gluten.StreamKafka rather than
LocalFiles.
+ auto splitInfo = std::make_shared<KafkaSplitInfo>();
+ splitInfo->leafType = SplitInfo::LeafType::TABLE_SCAN;
+ if (!validationMode_) {
+ VELOX_CHECK_LT(splitInfoIdx_, rawSplitInfos_.size(), "Plan must have
readRel and related split info.");
+ ::substrait::ReadRel_ExtensionTable extensionTable;
+ VELOX_CHECK(
+ extensionTable.ParseFromString(rawSplitInfos_[splitInfoIdx_++]),
+ "Failed to parse Kafka split info as ReadRel.ExtensionTable");
+ VELOX_CHECK(
+ extensionTable.detail().UnpackTo(&splitInfo->streamKafka),
+ "Kafka split info must contain a gluten.StreamKafka payload");
+ }
+
+ std::vector<std::string> colNameList;
+ std::vector<TypePtr> veloxTypeList;
+ bool asLowerCase = !veloxCfg_->get<bool>(kCaseSensitive, false);
+ if (readRel.has_base_schema()) {
+ const auto& baseSchema = readRel.base_schema();
+ colNameList.reserve(baseSchema.names().size());
+ for (const auto& name : baseSchema.names()) {
+ std::string fieldName = name;
+ if (asLowerCase) {
+ folly::toLowerAscii(fieldName);
+ }
+ colNameList.emplace_back(fieldName);
+ }
+ veloxTypeList = SubstraitParser::parseNamedStruct(baseSchema, asLowerCase);
+ }
Review Comment:
`colNameList.size()` is used to index `veloxTypeList[idx]` without verifying
that `parseNamedStruct()` returned the same number of types as there are names.
If the Substrait `base_schema` is malformed (or evolves), this can cause
out-of-bounds access. Add an explicit size check (and ideally throw a
user-facing error) before the loop; also consider using `size_t` for the loop
index to avoid signed/unsigned issues.
##########
cpp/velox/compute/VeloxPlanConverter.cc:
##########
@@ -270,9 +273,13 @@ void parseLocalFileNodes(
std::shared_ptr<const facebook::velox::core::PlanNode>
VeloxPlanConverter::toVeloxPlan(
const ::substrait::Plan& substraitPlan,
- std::vector<::substrait::ReadRel_LocalFiles> localFiles) {
+ std::vector<::substrait::ReadRel_LocalFiles> localFiles,
+ std::vector<std::string> rawSplitInfos) {
if (!validationMode_) {
- parseLocalFileNodes(&substraitVeloxPlanConverter_, veloxCfg_, localFiles);
+ if (rawSplitInfos.empty()) {
+ parseLocalFileNodes(&substraitVeloxPlanConverter_, veloxCfg_,
localFiles);
+ }
+ substraitVeloxPlanConverter_.setRawSplitInfos(std::move(rawSplitInfos));
}
Review Comment:
`toVeloxPlan()` now accepts both `localFiles` and `rawSplitInfos`, but if
`rawSplitInfos` is non-empty, `localFiles` is silently ignored. This can lead
to accidental misuse and hard-to-debug behavior if both are provided. Consider
enforcing mutual exclusivity (e.g., throw if both are non-empty) or refactoring
to a clearer API (separate overloads / `std::variant` / `std::optional`) so
callers cannot accidentally provide conflicting split sources.
##########
cpp/velox/compute/WholeStageResultIterator.cc:
##########
@@ -144,6 +147,14 @@ WholeStageResultIterator::WholeStageResultIterator(
for (size_t scanInfoIdx = 0; scanInfoIdx < scanInfos.size(); ++scanInfoIdx) {
const auto& scanInfo = scanInfos[scanInfoIdx];
+#ifdef ENABLE_KAFKA
+ if (auto kafkaSplitInfo =
std::dynamic_pointer_cast<KafkaSplitInfo>(scanInfo)) {
+ std::vector<velox::exec::Split> kafkaSplits;
+
kafkaSplits.emplace_back(kafkaSplitInfo->toConnectorSplit(connectorIds_.kafka));
+ splits_.emplace_back(std::move(kafkaSplits));
+ continue;
+ }
Review Comment:
This assumes `connectorIds_.kafka` is non-empty whenever a `KafkaSplitInfo`
is present. If it’s empty, the created split will carry an empty connector ID
and will likely fail at execution time (or route to the wrong config entry).
Add an explicit check here (user-facing) that `connectorIds_.kafka` is set when
encountering `KafkaSplitInfo`, and fail fast with a clear message if it isn’t.
##########
cpp/velox/substrait/SubstraitToVeloxPlan.cc:
##########
@@ -1511,6 +1521,58 @@ core::PlanNodePtr
SubstraitToVeloxPlanConverter::constructCudfValueStreamNode(
}
#endif
+#ifdef ENABLE_KAFKA
+core::PlanNodePtr
SubstraitToVeloxPlanConverter::constructKafkaStreamNode(const
::substrait::ReadRel& readRel) {
+ VELOX_USER_CHECK(!readRel.has_filter(), "Filter pushdown is not supported
for Kafka reads.");
+ // Like other table scans, a Kafka scan consumes the next split info. The
JVM sends it as a serialized
+ // ReadRel.ExtensionTable containing a gluten.StreamKafka rather than
LocalFiles.
+ auto splitInfo = std::make_shared<KafkaSplitInfo>();
+ splitInfo->leafType = SplitInfo::LeafType::TABLE_SCAN;
+ if (!validationMode_) {
+ VELOX_CHECK_LT(splitInfoIdx_, rawSplitInfos_.size(), "Plan must have
readRel and related split info.");
+ ::substrait::ReadRel_ExtensionTable extensionTable;
+ VELOX_CHECK(
+ extensionTable.ParseFromString(rawSplitInfos_[splitInfoIdx_++]),
+ "Failed to parse Kafka split info as ReadRel.ExtensionTable");
+ VELOX_CHECK(
+ extensionTable.detail().UnpackTo(&splitInfo->streamKafka),
+ "Kafka split info must contain a gluten.StreamKafka payload");
+ }
+
+ std::vector<std::string> colNameList;
+ std::vector<TypePtr> veloxTypeList;
+ bool asLowerCase = !veloxCfg_->get<bool>(kCaseSensitive, false);
+ if (readRel.has_base_schema()) {
+ const auto& baseSchema = readRel.base_schema();
+ colNameList.reserve(baseSchema.names().size());
+ for (const auto& name : baseSchema.names()) {
+ std::string fieldName = name;
+ if (asLowerCase) {
+ folly::toLowerAscii(fieldName);
+ }
+ colNameList.emplace_back(fieldName);
+ }
+ veloxTypeList = SubstraitParser::parseNamedStruct(baseSchema, asLowerCase);
+ }
+
+ std::vector<std::string> outNames;
+ outNames.reserve(colNameList.size());
+ connector::ColumnHandleMap assignments;
+ for (int idx = 0; idx < colNameList.size(); idx++) {
+ auto outName = SubstraitParser::makeNodeName(planNodeId_, idx);
+ assignments[outName] =
std::make_shared<KafkaColumnHandle>(colNameList[idx], veloxTypeList[idx]);
+ outNames.emplace_back(outName);
+ }
Review Comment:
`colNameList.size()` is used to index `veloxTypeList[idx]` without verifying
that `parseNamedStruct()` returned the same number of types as there are names.
If the Substrait `base_schema` is malformed (or evolves), this can cause
out-of-bounds access. Add an explicit size check (and ideally throw a
user-facing error) before the loop; also consider using `size_t` for the loop
index to avoid signed/unsigned issues.
##########
cpp/velox/substrait/SubstraitToVeloxPlan.cc:
##########
@@ -1511,6 +1521,58 @@ core::PlanNodePtr
SubstraitToVeloxPlanConverter::constructCudfValueStreamNode(
}
#endif
+#ifdef ENABLE_KAFKA
+core::PlanNodePtr
SubstraitToVeloxPlanConverter::constructKafkaStreamNode(const
::substrait::ReadRel& readRel) {
+ VELOX_USER_CHECK(!readRel.has_filter(), "Filter pushdown is not supported
for Kafka reads.");
+ // Like other table scans, a Kafka scan consumes the next split info. The
JVM sends it as a serialized
+ // ReadRel.ExtensionTable containing a gluten.StreamKafka rather than
LocalFiles.
+ auto splitInfo = std::make_shared<KafkaSplitInfo>();
+ splitInfo->leafType = SplitInfo::LeafType::TABLE_SCAN;
+ if (!validationMode_) {
+ VELOX_CHECK_LT(splitInfoIdx_, rawSplitInfos_.size(), "Plan must have
readRel and related split info.");
+ ::substrait::ReadRel_ExtensionTable extensionTable;
+ VELOX_CHECK(
+ extensionTable.ParseFromString(rawSplitInfos_[splitInfoIdx_++]),
+ "Failed to parse Kafka split info as ReadRel.ExtensionTable");
+ VELOX_CHECK(
+ extensionTable.detail().UnpackTo(&splitInfo->streamKafka),
+ "Kafka split info must contain a gluten.StreamKafka payload");
+ }
Review Comment:
These failures are caused by invalid/missing split payloads coming from the
caller, but they use `VELOX_CHECK*` (internal check) rather than user-facing
checks. Using `VELOX_USER_CHECK*`/`VELOX_USER_FAIL` here would produce a
clearer, non-internal error classification for bad inputs and avoid treating
malformed split payloads like invariant violations.
##########
cpp/velox/compute/WholeStageResultIterator.cc:
##########
@@ -268,6 +279,9 @@ std::shared_ptr<velox::core::QueryCtx>
WholeStageResultIterator::createNewVeloxQ
connectorConfigs[connectorIds_.iceberg] = hiveSessionConfig;
connectorConfigs[connectorIds_.delta] = hiveSessionConfig;
connectorConfigs[connectorIds_.iterator] = hiveSessionConfig;
+#ifdef ENABLE_KAFKA
+ connectorConfigs[connectorIds_.kafka] = hiveSessionConfig;
+#endif
Review Comment:
If `connectorIds_.kafka` is empty, this inserts a connector config entry
under an empty string key, which can cause confusing behavior (and may
override/overlap with other config resolution). Guard this with `if
(!connectorIds_.kafka.empty())` (or assert when Kafka is actually used) so the
connector config map remains well-defined.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
