You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "yihua (via GitHub)" <gi...@apache.org> on 2026/09/30 23:52:32 UTC
[PR] feat(spark): [DNM] read Hudi file slices natively through hudi-rs in the Comet bundle [hudi]
yihua opened a new pull request, #20170:
URL: https://github.com/apache/hudi/pull/20170
### Describe the issue this Pull Request addresses
issue: #19673
Comet leaves the Hudi scan on the JVM (`Unsupported file format HoodieFileGroupReaderBasedFileFormat`), so with #19674 Parquet decode stays on the JVM and a merge-on-read read pays a per-cell `CometSparkRowToColumnar` bridge before anything runs natively. RFC-113 (phase 2) covers the design.
**Do not merge.** Stacked on #19674, review that first; the diff shows the parent's three commits until it merges, and the code will be squashed and re-reviewed before this leaves draft. Design: RFC-113 (Phase 2). Opened to capture the current state.
### Summary and Changelog
Adds a native Hudi scan to the Comet bundle: the Hudi scan node is replaced by `CometHudiNativeScanExec`, whose native operator reads each task's file slices through hudi-rs (Parquet decode plus MOR log merge in Rust) and streams Arrow batches straight into the DataFusion operators above it. Planning stays in Hudi's existing Spark integration; the operator only receives the table base URI, the read options, the projected schema, and the slice paths.
- `hudi-spark-datasource/hudi-spark-comet`: the Spark-side glue (scan provider with eligibility gates, the exec node, the per-partition plan data injector), compiled only under `-Dcomet.hudi.build`.
- `packaging/hudi-native-spark-bundle/comet/`: Comet is built from a pinned commit (1.0.0) with a changed-files-only overlay that adds the `HudiScan` proto, the native operator, and a `ServiceLoader` hook for external scan providers; `build_comet.sh` verifies the overlay against the pinned upstream blob SHAs before applying it. hudi-rs is pinned by git rev to the arrow 58 / DataFusion 54.1 upgrade that shipped as 0.5.0.
- Bundle validation gains native-scan probes gated on `NATIVE_HUDI_SCAN=1`.
Runtime gate: `spark.comet.scan.hudi.enabled=true` (off by default). Any query shape outside the gates (incremental, CDC, bootstrap, nested types, table version outside 6/8/9, MOR below version 8, dropped partition columns, non-identity key generators with partition-column projection) leaves the plan untouched.
Still owed before this is ready for review: an adversarial self-review of the authored code, CI for the Comet source build, the S3 storage option passthrough, and a decision on whether the overlay should store patches instead of full file copies.
### Impact
New module and build profile only. Without `-Dcomet.hudi.build` the bundle is byte-identical to #19674. With it, the bundle depends on a locally built Comet at a distinct version (`1.0.0-hudi-<n>`), so the unmodified Central artifact can never satisfy the dependency silently.
### Risk Level
low
Opt-in at build time and at runtime, with the JVM scan as the fallback for every unsupported shape. Validated locally on Spark 3.5.5 (macOS arm64, Corretto 17, table version 8 tables): copy-on-write and merge-on-read snapshot rows and aggregates identical to the JVM read, no row bridge on the MOR path, time travel identical and native, and a CoW-join-MOR probe fully native with both sides through the native scan. Disabling the config leaves the JVM plan untouched, and the #19674 validation still passes without the build flag.
### Documentation Update
The module README under `packaging/hudi-native-spark-bundle/comet/` documents the build order and pins. Website docs for the native bundle and the new config are needed once this leaves draft.
### Contributor's checklist
- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [ ] Enough context is provided in the sections above
- [ ] Adequate tests were added if applicable
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): [DNM] read Hudi file slices natively through hudi-rs in the Comet bundle [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #20170:
URL: https://github.com/apache/hudi/pull/20170#issuecomment-5922238750
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d22b68962e65929e1a07279bbe14ebebfc151735",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17398",
"triggerID" : "d22b68962e65929e1a07279bbe14ebebfc151735",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d22b68962e65929e1a07279bbe14ebebfc151735 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17398)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): [DNM] read Hudi file slices natively through hudi-rs in the Comet bundle [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on PR #20170:
URL: https://github.com/apache/hudi/pull/20170#issuecomment-6366112498
IIUC, Planning is still done on JVM, so JVM is responsible for scanning, listing the files, prunes partitions, and groups them into file slices as of the query instant to be read.
Spark then builds the physical plan.
Comet's rules then swap Spark operators for Comet ones, and the provider swaps the Hudi scan for `CometHudiNativeScanExec`.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): read Hudi file slices natively through hudi-rs in the Comet bundle [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #20170:
URL: https://github.com/apache/hudi/pull/20170#discussion_r4404211577
##########
hudi-spark-datasource/hudi-spark-comet/src/main/scala/org/apache/hudi/comet/HudiCometScanProvider.scala:
##########
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.comet
+
+import org.apache.hudi.{HoodieFileIndex, HoodiePartitionCDCFileGroupMapping, HoodiePartitionFileSliceMapping}
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.{HoodieFileFormat, HoodieLogFile}
+import org.apache.hudi.storage.StoragePath
+
+import org.apache.comet.rules.CometExternalV1ScanProvider
+import org.apache.comet.serde.OperatorOuterClass
+import org.apache.comet.serde.QueryPlanSerde.serializeDataType
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.DynamicPruningExpression
+import org.apache.spark.sql.comet.{CometHudiNativeScanExec, SerializedPlan}
+import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.FileFormat
+import org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat
+import org.apache.spark.sql.types.{ArrayType, MapType, StructType}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
+
+/**
+ * Converts a Hudi data source V1 scan into a Comet native scan that reads file slices through
+ * hudi-rs. Registered with Comet's scan rule through the JDK ServiceLoader.
+ *
+ * Conversion is conservative: any shape outside the supported set (snapshot reads of Parquet
+ * tables whose partition columns are stored in the data files, with primitive output types)
+ * returns None, which leaves the plan on Hudi's JVM scan.
+ */
+class HudiCometScanProvider extends CometExternalV1ScanProvider with Logging {
+
+ override def supportsFormat(fileFormat: FileFormat): Boolean =
+ fileFormat.isInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+
+ override def convert(
+ session: SparkSession,
+ scanExec: FileSourceScanExec): Option[SparkPlan] = {
+ if (!session.conf
+ .getOption(HudiCometScanProvider.SCAN_ENABLED_KEY)
+ .exists(_.toBoolean)) {
+ return None
+ }
+ val format =
+ scanExec.relation.fileFormat.asInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+ if (format.isIncrementalQuery || format.isBootstrapTable ||
+ format.isMultipleBaseFileFormats || format.getRequiredFilters.nonEmpty) {
+ return None
+ }
+ if (scanExec.requiredSchema.isEmpty) {
+ return None
+ }
+ if (scanExec.partitionFilters.exists(
+ _.find(_.isInstanceOf[DynamicPruningExpression]).isDefined)) {
+ return None
+ }
Review Comment:
Fixed. Confirmed by repro: `where partitionpath = (select max(partitionpath) ...)` returned every partition natively. Spark lists partitions without any filter that holds a `PlanExpression` (scalar subquery or dynamic pruning) and applies those at execution time, so the provider now declines when a partition filter contains one; that replaces the narrower `DynamicPruningExpression` check.
##########
packaging/hudi-spark-native-bundle/comet/overlay/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -0,0 +1,1097 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.rules
+
+import java.lang.{Boolean => JBoolean}
+import java.net.URI
+import java.util.Locale
+import java.util.concurrent.ConcurrentHashMap
+
+import scala.collection.mutable
+import scala.collection.mutable.ListBuffer
+import scala.jdk.CollectionConverters._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{Attribute, DynamicPruningExpression, Expression, GenericInternalRow, InputFileBlockLength, InputFileBlockStart, InputFileName, PlanExpression}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.util.{sideBySide, ArrayBasedMapData, GenericArrayData, MetadataColumnHelper}
+import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getExistenceDefaultValues
+import org.apache.spark.sql.comet.{CometBatchScanExec, CometScanExec}
+import org.apache.spark.sql.execution.{FileSourceScanExec, InSubqueryExec, SparkPlan, SubqueryAdaptiveBroadcastExec}
+import org.apache.spark.sql.execution.datasources.HadoopFsRelation
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.datasources.v2.csv.CSVScan
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
+
+import org.apache.comet.{CometConf, DataTypeSupport, NativeBase}
+import org.apache.comet.CometConf._
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, isSpark35Plus, withFallbackReason, withFallbackReasons}
+import org.apache.comet.DataTypeSupport.isComplexType
+import org.apache.comet.iceberg.{CometIcebergNativeScanMetadata, IcebergReflection}
+import org.apache.comet.objectstore.NativeConfig
+import org.apache.comet.parquet.CometParquetUtils.{encryptionEnabled, isEncryptionConfigSupported}
+import org.apache.comet.serde.operator.{CometIcebergNativeScan, CometNativeScan}
+import org.apache.comet.shims.{CometTypeShim, ShimCometStreaming, ShimFileFormat, ShimSubqueryBroadcast}
+
+/**
+ * Spark physical optimizer rule for replacing Spark scans with Comet scans.
+ */
+case class CometScanRule(session: SparkSession)
+ extends Rule[SparkPlan]
+ with CometTypeShim
+ with ShimSubqueryBroadcast {
+
+ private lazy val showTransformations = CometConf.COMET_EXPLAIN_TRANSFORMATIONS.get()
+
+ override def apply(plan: SparkPlan): SparkPlan = {
+ val newPlan = _apply(plan)
+ if (showTransformations && !newPlan.fastEquals(plan)) {
+ logInfo(s"""
+ |=== Applying Rule $ruleName ===
+ |${sideBySide(plan.treeString, newPlan.treeString).mkString("\n")}
+ |""".stripMargin)
+ }
+ newPlan
+ }
+
+ private def _apply(plan: SparkPlan): SparkPlan = {
+ if (!isCometLoaded(conf)) return plan
+
+ // Comet does not support structured streaming. The parallel guard in
+ // CometExecRule only stops operator wrapping, so without this check we
+ // would still rewrite scans to CometScanExec in a streaming plan.
+ if (ShimCometStreaming.isStreamingPlan(plan)) return plan
+
+ def isSupportedScanNode(plan: SparkPlan): Boolean = plan match {
+ case _: FileSourceScanExec => true
+ case _: BatchScanExec => true
+ case _ => false
+ }
+
+ def isIcebergMetadataTable(scanExec: BatchScanExec): Boolean = {
+ // List of Iceberg metadata tables:
+ // https://iceberg.apache.org/docs/latest/spark-queries/#inspecting-tables
+ val metadataTableSuffix = Set(
+ "history",
+ "metadata_log_entries",
+ "snapshots",
+ "entries",
+ "files",
+ "manifests",
+ "partitions",
+ "position_deletes",
+ "all_data_files",
+ "all_delete_files",
+ "all_entries",
+ "all_manifests")
+
+ // Match case-insensitively: metadata tables surface lowercase via the path form
+ // (...metadata.json#all_manifests) but uppercase via the catalog-identifier form
+ // (db.table.ALL_DATA_FILES), and the latter must hit this gate too rather than fall through
+ // to reflection that fails on the metadata-table class.
+ val name = scanExec.table.name().toLowerCase(Locale.ROOT)
+ metadataTableSuffix.exists(name.endsWith)
+ }
+
+ val fullPlan = plan
+
+ def transformScan(scanNode: SparkPlan): SparkPlan = scanNode match {
+ // Tagged by CometSpark34AqeDppFallbackRule on Spark < 3.5 to keep a peer scan
+ // Spark-native for canonical symmetry in SMJ self-joins (SPARK-32509).
+ case scan if scan.getTagValue(CometScanRule.SKIP_COMET_SCAN_TAG).isDefined =>
+ withFallbackReason(scan, "AQE DPP region fallback (Spark < 3.5)")
+
+ case scan if !CometConf.COMET_NATIVE_SCAN_ENABLED.get(conf) =>
+ withFallbackReason(scan, "Comet Scan is not enabled")
+
+ // data source V1
+ case scanExec: FileSourceScanExec =>
+ transformV1Scan(fullPlan, scanExec)
+
+ // data source V2
+ case scanExec: BatchScanExec =>
+ if (isIcebergMetadataTable(scanExec)) {
+ withFallbackReason(scanExec, "Iceberg Metadata tables are not supported")
+ } else {
+ transformV2Scan(scanExec)
+ }
+ }
+
+ plan.transform {
+ case scan if isSupportedScanNode(scan) => transformScan(scan)
+ }
+ }
+
+ private def transformV1Scan(plan: SparkPlan, scanExec: FileSourceScanExec): SparkPlan = {
+ val metadataColNames = metadataCols(scanExec)
+ if (metadataColNames.nonEmpty) {
+ return withFallbackReason(
+ scanExec,
+ s"Metadata column(s) ${metadataColNames.mkString(", ")} is not supported")
+ }
+
+ // On Spark 3.4, injectQueryStageOptimizerRule is unavailable, so
+ // CometPlanAdaptiveDynamicPruningFilters cannot run. Fall back this scan to Spark so that
+ // Spark's PlanAdaptiveDynamicPruningFilters handles the SAB natively. Comet's narrower
+ // CometSpark34AqeDppFallbackRule (queryStagePrepRule on 3.4) then tags any matching BHJ's
+ // build-side broadcast so Spark's rule can match it via sameResult. See
+ // CometSpark34AqeDppFallbackRule's class docstring.
+ //
+ // On 3.5+, CometPlanAdaptiveDynamicPruningFilters rewrites SABs directly and this fallback
+ // is not needed.
+ if (!isSpark35Plus && scanExec.partitionFilters.exists(isAqeDynamicPruningFilter)) {
+ return withFallbackReason(scanExec, "AQE Dynamic Partition Pruning requires Spark 3.5+")
+ }
+
+ scanExec.relation match {
+ case r: HadoopFsRelation =>
+ if (!CometScanExec.isFileFormatSupported(r.fileFormat)) {
+ CometExternalV1ScanProvider.findForFormat(r.fileFormat) match {
Review Comment:
Addressed in the provider rather than the rule: it now checks `spark.comet.exec.enabled`, `ignoreCorruptFiles`/`ignoreMissingFiles` (session and relation options), the disabled-vectorized-reader opt-in, Parquet encryption, and that the table base URI uses a scheme object_store resolves. The remaining Comet guards are Parquet-reader specific (row index, default values, Variant) and do not apply to this operator, which reads through hudi-rs.
##########
hudi-spark-datasource/hudi-spark-comet/src/main/scala/org/apache/hudi/comet/HudiCometScanProvider.scala:
##########
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.comet
+
+import org.apache.hudi.{HoodieFileIndex, HoodiePartitionCDCFileGroupMapping, HoodiePartitionFileSliceMapping}
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.{HoodieFileFormat, HoodieLogFile}
+import org.apache.hudi.storage.StoragePath
+
+import org.apache.comet.rules.CometExternalV1ScanProvider
+import org.apache.comet.serde.OperatorOuterClass
+import org.apache.comet.serde.QueryPlanSerde.serializeDataType
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.DynamicPruningExpression
+import org.apache.spark.sql.comet.{CometHudiNativeScanExec, SerializedPlan}
+import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.FileFormat
+import org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat
+import org.apache.spark.sql.types.{ArrayType, MapType, StructType}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
+
+/**
+ * Converts a Hudi data source V1 scan into a Comet native scan that reads file slices through
+ * hudi-rs. Registered with Comet's scan rule through the JDK ServiceLoader.
+ *
+ * Conversion is conservative: any shape outside the supported set (snapshot reads of Parquet
+ * tables whose partition columns are stored in the data files, with primitive output types)
+ * returns None, which leaves the plan on Hudi's JVM scan.
+ */
+class HudiCometScanProvider extends CometExternalV1ScanProvider with Logging {
+
+ override def supportsFormat(fileFormat: FileFormat): Boolean =
+ fileFormat.isInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+
+ override def convert(
+ session: SparkSession,
+ scanExec: FileSourceScanExec): Option[SparkPlan] = {
+ if (!session.conf
+ .getOption(HudiCometScanProvider.SCAN_ENABLED_KEY)
+ .exists(_.toBoolean)) {
+ return None
+ }
+ val format =
+ scanExec.relation.fileFormat.asInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+ if (format.isIncrementalQuery || format.isBootstrapTable ||
+ format.isMultipleBaseFileFormats || format.getRequiredFilters.nonEmpty) {
+ return None
+ }
+ if (scanExec.requiredSchema.isEmpty) {
+ return None
+ }
+ if (scanExec.partitionFilters.exists(
+ _.find(_.isInstanceOf[DynamicPruningExpression]).isDefined)) {
+ return None
+ }
+ if (scanExec.output.exists(attr =>
+ attr.dataType.isInstanceOf[StructType] || attr.dataType.isInstanceOf[ArrayType] ||
+ attr.dataType.isInstanceOf[MapType])) {
+ return None
+ }
+
+ val fileIndex = scanExec.relation.location match {
+ case index: HoodieFileIndex => index
+ case _ => return None
+ }
+ val tableConfig = fileIndex.metaClient.getTableConfig
+ val tableVersion = tableConfig.getTableVersion.versionCode()
+ // The versions the hudi-rs file group reader accepts. Merge-on-read
+ // additionally requires 8+, where log files are sealed per commit, so the
+ // planned file paths fully determine the data the reader must see.
+ if (!HudiCometScanProvider.SUPPORTED_TABLE_VERSIONS.contains(tableVersion) ||
+ (format.isMORTable && tableVersion < 8)) {
+ return None
+ }
+ if (tableConfig.shouldDropPartitionColumns()) {
+ return None
+ }
+ if (tableConfig.getBaseFileFormat != HoodieFileFormat.PARQUET) {
+ return None
+ }
+ // Every output column must come out of the data files: the native reader
+ // has no partition-value append path. Partition columns are stored in the
+ // files (dropped partition columns are gated out above), but the JVM read
+ // appends the value parsed from the partition path instead, and the two
+ // agree only when the key generator writes values into the path unchanged.
+ // Generators that transform values (timestamp-based, custom) stay on the
+ // JVM scan whenever a partition column is projected.
+ val partitionColumns = scanExec.relation.partitionSchema.fieldNames.toSet
+ val availableColumns = scanExec.relation.dataSchema.fieldNames.toSet ++ partitionColumns
+ if (!scanExec.output.forall(attr => availableColumns.contains(attr.name))) {
+ return None
+ }
+ if (scanExec.output.exists(attr => partitionColumns.contains(attr.name)) &&
+ !HudiCometScanProvider.IDENTITY_KEY_GENERATORS.contains(
+ Option(tableConfig.getKeyGeneratorClassName)
+ .getOrElse("org.apache.hudi.keygen.SimpleKeyGenerator"))) {
+ return None
+ }
Review Comment:
It did not: hudi-rs refused the slice at task time, so the query failed instead of falling back. The provider now requires `COMMIT_TIME_ORDERING` or `EVENT_TIME_ORDERING` on merge-on-read tables, and declines any table with more than one ordering field (hudi-rs refuses to open those, copy-on-write included) or the LSM storage layout.
##########
hudi-spark-datasource/hudi-spark-comet/src/main/scala/org/apache/hudi/comet/HudiCometScanProvider.scala:
##########
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.comet
+
+import org.apache.hudi.{HoodieFileIndex, HoodiePartitionCDCFileGroupMapping, HoodiePartitionFileSliceMapping}
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.{HoodieFileFormat, HoodieLogFile}
+import org.apache.hudi.storage.StoragePath
+
+import org.apache.comet.rules.CometExternalV1ScanProvider
+import org.apache.comet.serde.OperatorOuterClass
+import org.apache.comet.serde.QueryPlanSerde.serializeDataType
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.DynamicPruningExpression
+import org.apache.spark.sql.comet.{CometHudiNativeScanExec, SerializedPlan}
+import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.FileFormat
+import org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat
+import org.apache.spark.sql.types.{ArrayType, MapType, StructType}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
+
+/**
+ * Converts a Hudi data source V1 scan into a Comet native scan that reads file slices through
+ * hudi-rs. Registered with Comet's scan rule through the JDK ServiceLoader.
+ *
+ * Conversion is conservative: any shape outside the supported set (snapshot reads of Parquet
+ * tables whose partition columns are stored in the data files, with primitive output types)
+ * returns None, which leaves the plan on Hudi's JVM scan.
+ */
+class HudiCometScanProvider extends CometExternalV1ScanProvider with Logging {
+
+ override def supportsFormat(fileFormat: FileFormat): Boolean =
+ fileFormat.isInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+
+ override def convert(
+ session: SparkSession,
+ scanExec: FileSourceScanExec): Option[SparkPlan] = {
+ if (!session.conf
+ .getOption(HudiCometScanProvider.SCAN_ENABLED_KEY)
+ .exists(_.toBoolean)) {
+ return None
+ }
+ val format =
+ scanExec.relation.fileFormat.asInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+ if (format.isIncrementalQuery || format.isBootstrapTable ||
+ format.isMultipleBaseFileFormats || format.getRequiredFilters.nonEmpty) {
+ return None
+ }
+ if (scanExec.requiredSchema.isEmpty) {
+ return None
+ }
+ if (scanExec.partitionFilters.exists(
+ _.find(_.isInstanceOf[DynamicPruningExpression]).isDefined)) {
+ return None
+ }
+ if (scanExec.output.exists(attr =>
+ attr.dataType.isInstanceOf[StructType] || attr.dataType.isInstanceOf[ArrayType] ||
+ attr.dataType.isInstanceOf[MapType])) {
+ return None
+ }
+
+ val fileIndex = scanExec.relation.location match {
+ case index: HoodieFileIndex => index
+ case _ => return None
+ }
+ val tableConfig = fileIndex.metaClient.getTableConfig
+ val tableVersion = tableConfig.getTableVersion.versionCode()
+ // The versions the hudi-rs file group reader accepts. Merge-on-read
+ // additionally requires 8+, where log files are sealed per commit, so the
+ // planned file paths fully determine the data the reader must see.
+ if (!HudiCometScanProvider.SUPPORTED_TABLE_VERSIONS.contains(tableVersion) ||
+ (format.isMORTable && tableVersion < 8)) {
+ return None
+ }
+ if (tableConfig.shouldDropPartitionColumns()) {
+ return None
+ }
+ if (tableConfig.getBaseFileFormat != HoodieFileFormat.PARQUET) {
+ return None
+ }
+ // Every output column must come out of the data files: the native reader
+ // has no partition-value append path. Partition columns are stored in the
+ // files (dropped partition columns are gated out above), but the JVM read
+ // appends the value parsed from the partition path instead, and the two
+ // agree only when the key generator writes values into the path unchanged.
+ // Generators that transform values (timestamp-based, custom) stay on the
+ // JVM scan whenever a partition column is projected.
+ val partitionColumns = scanExec.relation.partitionSchema.fieldNames.toSet
+ val availableColumns = scanExec.relation.dataSchema.fieldNames.toSet ++ partitionColumns
+ if (!scanExec.output.forall(attr => availableColumns.contains(attr.name))) {
+ return None
+ }
+ if (scanExec.output.exists(attr => partitionColumns.contains(attr.name)) &&
+ !HudiCometScanProvider.IDENTITY_KEY_GENERATORS.contains(
+ Option(tableConfig.getKeyGeneratorClassName)
+ .getOrElse("org.apache.hudi.keygen.SimpleKeyGenerator"))) {
+ return None
+ }
+
+ val baseUri = fileIndex.metaClient.getBasePath.toString
+ val slices = extractFileSlices(scanExec, baseUri, format.isMORTable) match {
+ case Some(s) => s
+ case None => return None
+ }
+
+ val scanHashCode = scanExec.canonicalized.hashCode()
+ val commonBuilder = OperatorOuterClass.HudiScanCommon
+ .newBuilder()
+ .setTableBaseUri(baseUri)
+ .setSessionTimezone(session.sessionState.conf.sessionLocalTimeZone)
+ .setScanHashCode(scanHashCode)
+ // Bounds the log-file scan to instants at or before the query instant,
+ // exactly as the JVM file group reader does with the same timestamp; a
+ // snapshot read passes the latest completed instant, a time-travel read
+ // the requested one. Base files need no bound: the driver's listing only
+ // names base files from admitted commits.
+ .putOptions("hoodie.read.end.timestamp", format.getQueryTimestamp)
Review Comment:
Fixed. The provider now declines when `hoodie.datasource.merge.type=skip_merge` (repro: 450 unmerged rows on the JVM vs 280 merged natively), `hoodie.datasource.read.extract.partition.values.from.path=true`, or `hoodie.schema.on.read.enable=true` is set, in the relation options or the session. Read-optimized queries were already correct since that relation lists base files only. Storage options (S3 credentials and endpoints) are still the open item named in the description.
##########
hudi-spark-datasource/hudi-spark-comet/src/main/scala/org/apache/hudi/comet/HudiCometScanProvider.scala:
##########
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.comet
+
+import org.apache.hudi.{HoodieFileIndex, HoodiePartitionCDCFileGroupMapping, HoodiePartitionFileSliceMapping}
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.{HoodieFileFormat, HoodieLogFile}
+import org.apache.hudi.storage.StoragePath
+
+import org.apache.comet.rules.CometExternalV1ScanProvider
+import org.apache.comet.serde.OperatorOuterClass
+import org.apache.comet.serde.QueryPlanSerde.serializeDataType
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.DynamicPruningExpression
+import org.apache.spark.sql.comet.{CometHudiNativeScanExec, SerializedPlan}
+import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.FileFormat
+import org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat
+import org.apache.spark.sql.types.{ArrayType, MapType, StructType}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
+
+/**
+ * Converts a Hudi data source V1 scan into a Comet native scan that reads file slices through
+ * hudi-rs. Registered with Comet's scan rule through the JDK ServiceLoader.
+ *
+ * Conversion is conservative: any shape outside the supported set (snapshot reads of Parquet
+ * tables whose partition columns are stored in the data files, with primitive output types)
+ * returns None, which leaves the plan on Hudi's JVM scan.
+ */
+class HudiCometScanProvider extends CometExternalV1ScanProvider with Logging {
+
+ override def supportsFormat(fileFormat: FileFormat): Boolean =
+ fileFormat.isInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+
+ override def convert(
+ session: SparkSession,
+ scanExec: FileSourceScanExec): Option[SparkPlan] = {
+ if (!session.conf
+ .getOption(HudiCometScanProvider.SCAN_ENABLED_KEY)
+ .exists(_.toBoolean)) {
+ return None
+ }
+ val format =
+ scanExec.relation.fileFormat.asInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+ if (format.isIncrementalQuery || format.isBootstrapTable ||
+ format.isMultipleBaseFileFormats || format.getRequiredFilters.nonEmpty) {
+ return None
+ }
+ if (scanExec.requiredSchema.isEmpty) {
+ return None
+ }
+ if (scanExec.partitionFilters.exists(
+ _.find(_.isInstanceOf[DynamicPruningExpression]).isDefined)) {
+ return None
+ }
+ if (scanExec.output.exists(attr =>
+ attr.dataType.isInstanceOf[StructType] || attr.dataType.isInstanceOf[ArrayType] ||
+ attr.dataType.isInstanceOf[MapType])) {
+ return None
+ }
+
+ val fileIndex = scanExec.relation.location match {
+ case index: HoodieFileIndex => index
+ case _ => return None
+ }
+ val tableConfig = fileIndex.metaClient.getTableConfig
+ val tableVersion = tableConfig.getTableVersion.versionCode()
+ // The versions the hudi-rs file group reader accepts. Merge-on-read
+ // additionally requires 8+, where log files are sealed per commit, so the
+ // planned file paths fully determine the data the reader must see.
+ if (!HudiCometScanProvider.SUPPORTED_TABLE_VERSIONS.contains(tableVersion) ||
+ (format.isMORTable && tableVersion < 8)) {
+ return None
+ }
+ if (tableConfig.shouldDropPartitionColumns()) {
+ return None
+ }
+ if (tableConfig.getBaseFileFormat != HoodieFileFormat.PARQUET) {
+ return None
+ }
+ // Every output column must come out of the data files: the native reader
+ // has no partition-value append path. Partition columns are stored in the
+ // files (dropped partition columns are gated out above), but the JVM read
+ // appends the value parsed from the partition path instead, and the two
+ // agree only when the key generator writes values into the path unchanged.
+ // Generators that transform values (timestamp-based, custom) stay on the
+ // JVM scan whenever a partition column is projected.
+ val partitionColumns = scanExec.relation.partitionSchema.fieldNames.toSet
+ val availableColumns = scanExec.relation.dataSchema.fieldNames.toSet ++ partitionColumns
+ if (!scanExec.output.forall(attr => availableColumns.contains(attr.name))) {
+ return None
+ }
+ if (scanExec.output.exists(attr => partitionColumns.contains(attr.name)) &&
+ !HudiCometScanProvider.IDENTITY_KEY_GENERATORS.contains(
+ Option(tableConfig.getKeyGeneratorClassName)
+ .getOrElse("org.apache.hudi.keygen.SimpleKeyGenerator"))) {
+ return None
+ }
Review Comment:
Good catch. Both write under `__HIVE_DEFAULT_PARTITION__`; the JVM read returns null for that partition's column while the data file (and so the native read) holds `''` for the empty value. The provider now declines when a partition column is projected and any selected partition has a null partition value. Without the partition column projected the read stays native and matches.
##########
hudi-spark-datasource/hudi-spark-comet/src/main/scala/org/apache/hudi/comet/HudiCometScanProvider.scala:
##########
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.comet
+
+import org.apache.hudi.{HoodieFileIndex, HoodiePartitionCDCFileGroupMapping, HoodiePartitionFileSliceMapping}
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.{HoodieFileFormat, HoodieLogFile}
+import org.apache.hudi.storage.StoragePath
+
+import org.apache.comet.rules.CometExternalV1ScanProvider
+import org.apache.comet.serde.OperatorOuterClass
+import org.apache.comet.serde.QueryPlanSerde.serializeDataType
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.DynamicPruningExpression
+import org.apache.spark.sql.comet.{CometHudiNativeScanExec, SerializedPlan}
+import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.FileFormat
+import org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat
+import org.apache.spark.sql.types.{ArrayType, MapType, StructType}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
+
+/**
+ * Converts a Hudi data source V1 scan into a Comet native scan that reads file slices through
+ * hudi-rs. Registered with Comet's scan rule through the JDK ServiceLoader.
+ *
+ * Conversion is conservative: any shape outside the supported set (snapshot reads of Parquet
+ * tables whose partition columns are stored in the data files, with primitive output types)
+ * returns None, which leaves the plan on Hudi's JVM scan.
+ */
+class HudiCometScanProvider extends CometExternalV1ScanProvider with Logging {
+
+ override def supportsFormat(fileFormat: FileFormat): Boolean =
+ fileFormat.isInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+
+ override def convert(
+ session: SparkSession,
+ scanExec: FileSourceScanExec): Option[SparkPlan] = {
+ if (!session.conf
+ .getOption(HudiCometScanProvider.SCAN_ENABLED_KEY)
+ .exists(_.toBoolean)) {
+ return None
+ }
+ val format =
+ scanExec.relation.fileFormat.asInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+ if (format.isIncrementalQuery || format.isBootstrapTable ||
+ format.isMultipleBaseFileFormats || format.getRequiredFilters.nonEmpty) {
+ return None
+ }
+ if (scanExec.requiredSchema.isEmpty) {
+ return None
+ }
+ if (scanExec.partitionFilters.exists(
+ _.find(_.isInstanceOf[DynamicPruningExpression]).isDefined)) {
+ return None
+ }
+ if (scanExec.output.exists(attr =>
+ attr.dataType.isInstanceOf[StructType] || attr.dataType.isInstanceOf[ArrayType] ||
+ attr.dataType.isInstanceOf[MapType])) {
+ return None
+ }
+
+ val fileIndex = scanExec.relation.location match {
+ case index: HoodieFileIndex => index
+ case _ => return None
+ }
+ val tableConfig = fileIndex.metaClient.getTableConfig
+ val tableVersion = tableConfig.getTableVersion.versionCode()
+ // The versions the hudi-rs file group reader accepts. Merge-on-read
+ // additionally requires 8+, where log files are sealed per commit, so the
+ // planned file paths fully determine the data the reader must see.
+ if (!HudiCometScanProvider.SUPPORTED_TABLE_VERSIONS.contains(tableVersion) ||
+ (format.isMORTable && tableVersion < 8)) {
+ return None
+ }
+ if (tableConfig.shouldDropPartitionColumns()) {
+ return None
+ }
+ if (tableConfig.getBaseFileFormat != HoodieFileFormat.PARQUET) {
+ return None
+ }
+ // Every output column must come out of the data files: the native reader
+ // has no partition-value append path. Partition columns are stored in the
+ // files (dropped partition columns are gated out above), but the JVM read
+ // appends the value parsed from the partition path instead, and the two
+ // agree only when the key generator writes values into the path unchanged.
+ // Generators that transform values (timestamp-based, custom) stay on the
+ // JVM scan whenever a partition column is projected.
+ val partitionColumns = scanExec.relation.partitionSchema.fieldNames.toSet
+ val availableColumns = scanExec.relation.dataSchema.fieldNames.toSet ++ partitionColumns
+ if (!scanExec.output.forall(attr => availableColumns.contains(attr.name))) {
+ return None
+ }
+ if (scanExec.output.exists(attr => partitionColumns.contains(attr.name)) &&
+ !HudiCometScanProvider.IDENTITY_KEY_GENERATORS.contains(
+ Option(tableConfig.getKeyGeneratorClassName)
+ .getOrElse("org.apache.hudi.keygen.SimpleKeyGenerator"))) {
+ return None
+ }
+
+ val baseUri = fileIndex.metaClient.getBasePath.toString
+ val slices = extractFileSlices(scanExec, baseUri, format.isMORTable) match {
+ case Some(s) => s
+ case None => return None
+ }
+
+ val scanHashCode = scanExec.canonicalized.hashCode()
+ val commonBuilder = OperatorOuterClass.HudiScanCommon
+ .newBuilder()
+ .setTableBaseUri(baseUri)
+ .setSessionTimezone(session.sessionState.conf.sessionLocalTimeZone)
+ .setScanHashCode(scanHashCode)
+ // Bounds the log-file scan to instants at or before the query instant,
+ // exactly as the JVM file group reader does with the same timestamp; a
+ // snapshot read passes the latest completed instant, a time-travel read
+ // the requested one. Base files need no bound: the driver's listing only
+ // names base files from admitted commits.
+ .putOptions("hoodie.read.end.timestamp", format.getQueryTimestamp)
+ for (attr <- scanExec.output) {
+ val field = OperatorOuterClass.SparkStructField
+ .newBuilder()
+ .setName(attr.name)
+ .setNullable(attr.nullable)
+ serializeDataType(attr.dataType) match {
+ case Some(dataType) => field.setDataType(dataType)
+ case None => return None
+ }
+ commonBuilder.addRequiredSchema(field)
+ }
+ val commonBytes = commonBuilder.build().toByteArray
+
+ val perPartitionBytes = slices.map { slice =>
Review Comment:
Fixed. Slices are now packed with the same `FilePartition` logic Spark's file scan uses (largest first, up to `FilePartition.maxSplitBytes`), so the 30k-row, 471-file probe table goes from 471 native tasks to the JVM's 17. A base file larger than the split size stays one task; byte-range splitting of a copy-on-write file would need row-group ranges in the hudi-rs API and is a follow-up.
##########
packaging/hudi-spark-native-bundle/comet/overlay/native/core/src/execution/operators/hudi_scan.rs:
##########
@@ -0,0 +1,245 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! A native Hudi table scan, reading file slices through hudi-rs.
+//!
+//! The Spark driver plans the read (file listing, partition pruning, slice
+//! grouping as of the query instant) and ships each task its file slices plus
+//! the table base URI. hudi-rs opens the table from `hoodie.properties` under
+//! that URI, so merge mode, ordering field, and table version never travel in
+//! the plan; this operator only reads the named slices and projects the
+//! requested columns.
+//!
+//! Every batch crosses from hudi-rs into this workspace through the Arrow C
+//! data interface, whose layout is fixed by the Arrow specification and is
+//! identical across arrow crate majors; the transfer is pointer-level, not a
+//! copy. hudi-rs currently pins the same arrow major, so the crossing is a
+//! same-crate round trip; if the lines diverge again, only the export side
+//! switches to a renamed dep on hudi-rs's line.
+
+use std::collections::HashMap;
+use std::fmt;
+use std::sync::Arc;
+
+use arrow::array::{Array as _, RecordBatch, RecordBatchOptions, StructArray};
+use arrow::compute::cast;
+use arrow::datatypes::SchemaRef;
+use datafusion::common::{DataFusionError, Result as DFResult};
+use datafusion::execution::{SendableRecordBatchStream, TaskContext};
+use datafusion::physical_expr::EquivalenceProperties;
+use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
+use datafusion::physical_plan::metrics::{
+ BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet,
+};
+use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
+use datafusion::physical_plan::{
+ DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PlanProperties,
+};
+use futures::{StreamExt, TryStreamExt};
+use hudi::file_group::reader::FileGroupReader;
+use hudi::table::ReadOptions;
+
+/// One file slice assigned to this partition: a base file and the log files
+/// whose records amend it.
+#[derive(Debug, Clone)]
+pub struct HudiFileSliceSpec {
+ pub base_file_path: String,
+ pub log_file_paths: Vec<String>,
+}
+
+#[derive(Debug)]
+pub struct HudiScanExec {
+ table_base_uri: String,
+ options: HashMap<String, String>,
+ output_schema: SchemaRef,
+ file_slices: Vec<HudiFileSliceSpec>,
+ plan_properties: Arc<PlanProperties>,
+ metrics: ExecutionPlanMetricsSet,
+}
+
+impl HudiScanExec {
+ pub fn new(
+ table_base_uri: String,
+ options: HashMap<String, String>,
+ output_schema: SchemaRef,
+ file_slices: Vec<HudiFileSliceSpec>,
+ ) -> Self {
+ let plan_properties = Arc::new(PlanProperties::new(
+ EquivalenceProperties::new(Arc::clone(&output_schema)),
+ Partitioning::UnknownPartitioning(1),
+ EmissionType::Incremental,
+ Boundedness::Bounded,
+ ));
+ Self {
+ table_base_uri,
+ options,
+ output_schema,
+ file_slices,
+ plan_properties,
+ metrics: ExecutionPlanMetricsSet::new(),
+ }
+ }
+}
+
+/// Moves a batch from hudi-rs's arrow into this workspace's arrow over the
+/// Arrow C data interface.
+fn batch_across_ffi(batch: RecordBatch) -> DFResult<RecordBatch> {
+ let data = StructArray::from(batch).into_data();
+ let (ffi_array, ffi_schema) = arrow::ffi::to_ffi(&data)
+ .map_err(|e| DataFusionError::Execution(format!("Hudi scan arrow export failed: {e}")))?;
+ // SAFETY: FFI_ArrowArray and FFI_ArrowSchema are the #[repr(C)] structs of
+ // the Arrow C data interface; their layout is defined by the Arrow
+ // specification and is identical across arrow crate versions. Ownership
+ // moves with the structs, so the release callbacks fire exactly once.
+ let data = unsafe { arrow::ffi::from_ffi(ffi_array, &ffi_schema) }
+ .map_err(|e| DataFusionError::Execution(format!("Hudi scan arrow import failed: {e}")))?;
+ Ok(RecordBatch::from(StructArray::from(data)))
+}
+
+/// Reorders and casts a batch's columns to the scan's output schema. hudi-rs
+/// projects the requested columns but keeps the table schema's column order
+/// and physical types; Spark expects its own order and logical types.
+fn align_to_schema(batch: RecordBatch, schema: &SchemaRef) -> DFResult<RecordBatch> {
+ let num_rows = batch.num_rows();
+ let mut columns = Vec::with_capacity(schema.fields().len());
+ for field in schema.fields() {
+ let index = batch.schema().index_of(field.name()).map_err(|_| {
+ DataFusionError::Execution(format!(
+ "Hudi scan did not produce required column {}",
+ field.name()
+ ))
+ })?;
+ let column = batch.column(index);
+ let column = if column.data_type() == field.data_type() {
+ Arc::clone(column)
+ } else {
+ cast(column, field.data_type()).map_err(|e| {
+ DataFusionError::Execution(format!(
+ "Hudi scan cannot cast column {} from {} to {}: {e}",
+ field.name(),
+ column.data_type(),
+ field.data_type()
+ ))
+ })?
+ };
+ columns.push(column);
+ }
+ RecordBatch::try_new_with_options(
+ Arc::clone(schema),
+ columns,
+ &RecordBatchOptions::new().with_row_count(Some(num_rows)),
+ )
+ .map_err(|e| DataFusionError::Execution(format!("Hudi scan batch assembly failed: {e}")))
+}
+
+impl DisplayAs for HudiScanExec {
+ fn fmt_as(&self, _t: DisplayFormatType, f: &mut fmt::Formatter) -> fmt::Result {
+ write!(
+ f,
+ "HudiScanExec: table={}, file_slices={}",
+ self.table_base_uri,
+ self.file_slices.len()
+ )
+ }
+}
+
+impl ExecutionPlan for HudiScanExec {
+ fn name(&self) -> &str {
+ "HudiScanExec"
+ }
+
+ fn schema(&self) -> SchemaRef {
+ Arc::clone(&self.output_schema)
+ }
+
+ fn properties(&self) -> &Arc<PlanProperties> {
+ &self.plan_properties
+ }
+
+ fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+ vec![]
+ }
+
+ fn with_new_children(
+ self: Arc<Self>,
+ _children: Vec<Arc<dyn ExecutionPlan>>,
+ ) -> DFResult<Arc<dyn ExecutionPlan>> {
+ Ok(self)
+ }
+
+ fn execute(
+ &self,
+ _partition: usize,
+ _context: Arc<TaskContext>,
+ ) -> DFResult<SendableRecordBatchStream> {
+ let base_uri = self.table_base_uri.clone();
+ let options = self.options.clone();
+ let schema = Arc::clone(&self.output_schema);
+ let slices = self.file_slices.clone();
+ let projection: Vec<String> = schema
+ .fields()
+ .iter()
+ .map(|f| f.name().clone())
+ .collect();
+ let baseline = BaselineMetrics::new(&self.metrics, 0);
+
+ let batches = futures::stream::once(async move {
+ let reader = FileGroupReader::new_with_options(
+ &base_uri,
+ options.iter().map(|(k, v)| (k.as_str(), v.as_str())),
+ )
+ .await
+ .map_err(|e| {
+ DataFusionError::Execution(format!("failed to open Hudi table {base_uri}: {e}"))
+ })?;
+ let read_options = ReadOptions {
+ projection: Some(projection),
+ ..Default::default()
+ };
+ let mut out = Vec::with_capacity(slices.len());
+ for slice in &slices {
+ let batch = reader
Review Comment:
Fixed. The operator now uses `read_file_slice_from_paths_stream`, which pushes the projection into the base file read and streams each slice in batches of the session batch size, with the per-slice streams chained lazily, so a task no longer collects all of its slices before emitting. It also gives the reader the table schema, so a column added after some base files were written is null-filled as on the JVM instead of failing merge-on-read slices. Today that schema comes from opening the hudi-rs `Table` once per task; apache/hudi-rs#846 adds a reader constructor that takes the table properties and schema the driver already has, after which the plan will ship both and tasks will touch storage only for the data files.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): read Hudi file slices natively through hudi-rs in the Comet bundle [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #20170:
URL: https://github.com/apache/hudi/pull/20170#discussion_r4419156063
##########
hudi-spark-datasource/hudi-spark-comet/src/main/scala/org/apache/hudi/comet/HudiCometScanProvider.scala:
##########
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.comet
+
+import org.apache.hudi.{HoodieFileIndex, HoodiePartitionCDCFileGroupMapping, HoodiePartitionFileSliceMapping}
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.{HoodieFileFormat, HoodieLogFile}
+import org.apache.hudi.storage.StoragePath
+
+import org.apache.comet.rules.CometExternalV1ScanProvider
+import org.apache.comet.serde.OperatorOuterClass
+import org.apache.comet.serde.QueryPlanSerde.serializeDataType
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.DynamicPruningExpression
+import org.apache.spark.sql.comet.{CometHudiNativeScanExec, SerializedPlan}
+import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.FileFormat
+import org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat
+import org.apache.spark.sql.types.{ArrayType, MapType, StructType}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
+
+/**
+ * Converts a Hudi data source V1 scan into a Comet native scan that reads file slices through
+ * hudi-rs. Registered with Comet's scan rule through the JDK ServiceLoader.
+ *
+ * Conversion is conservative: any shape outside the supported set (snapshot reads of Parquet
+ * tables whose partition columns are stored in the data files, with primitive output types)
+ * returns None, which leaves the plan on Hudi's JVM scan.
+ */
+class HudiCometScanProvider extends CometExternalV1ScanProvider with Logging {
+
+ override def supportsFormat(fileFormat: FileFormat): Boolean =
+ fileFormat.isInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+
+ override def convert(
+ session: SparkSession,
+ scanExec: FileSourceScanExec): Option[SparkPlan] = {
+ if (!session.conf
+ .getOption(HudiCometScanProvider.SCAN_ENABLED_KEY)
+ .exists(_.toBoolean)) {
+ return None
+ }
+ val format =
+ scanExec.relation.fileFormat.asInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+ if (format.isIncrementalQuery || format.isBootstrapTable ||
+ format.isMultipleBaseFileFormats || format.getRequiredFilters.nonEmpty) {
+ return None
+ }
+ if (scanExec.requiredSchema.isEmpty) {
+ return None
+ }
+ if (scanExec.partitionFilters.exists(
+ _.find(_.isInstanceOf[DynamicPruningExpression]).isDefined)) {
+ return None
+ }
+ if (scanExec.output.exists(attr =>
+ attr.dataType.isInstanceOf[StructType] || attr.dataType.isInstanceOf[ArrayType] ||
+ attr.dataType.isInstanceOf[MapType])) {
+ return None
+ }
+
+ val fileIndex = scanExec.relation.location match {
+ case index: HoodieFileIndex => index
+ case _ => return None
+ }
+ val tableConfig = fileIndex.metaClient.getTableConfig
+ val tableVersion = tableConfig.getTableVersion.versionCode()
+ // The versions the hudi-rs file group reader accepts. Merge-on-read
+ // additionally requires 8+, where log files are sealed per commit, so the
+ // planned file paths fully determine the data the reader must see.
+ if (!HudiCometScanProvider.SUPPORTED_TABLE_VERSIONS.contains(tableVersion) ||
+ (format.isMORTable && tableVersion < 8)) {
+ return None
+ }
+ if (tableConfig.shouldDropPartitionColumns()) {
+ return None
+ }
+ if (tableConfig.getBaseFileFormat != HoodieFileFormat.PARQUET) {
+ return None
+ }
+ // Every output column must come out of the data files: the native reader
+ // has no partition-value append path. Partition columns are stored in the
+ // files (dropped partition columns are gated out above), but the JVM read
+ // appends the value parsed from the partition path instead, and the two
+ // agree only when the key generator writes values into the path unchanged.
+ // Generators that transform values (timestamp-based, custom) stay on the
+ // JVM scan whenever a partition column is projected.
+ val partitionColumns = scanExec.relation.partitionSchema.fieldNames.toSet
+ val availableColumns = scanExec.relation.dataSchema.fieldNames.toSet ++ partitionColumns
+ if (!scanExec.output.forall(attr => availableColumns.contains(attr.name))) {
+ return None
+ }
+ if (scanExec.output.exists(attr => partitionColumns.contains(attr.name)) &&
+ !HudiCometScanProvider.IDENTITY_KEY_GENERATORS.contains(
+ Option(tableConfig.getKeyGeneratorClassName)
+ .getOrElse("org.apache.hudi.keygen.SimpleKeyGenerator"))) {
+ return None
+ }
Review Comment:
The comment at L184 says a scan of a partitioned table always outputs its partition columns, whatever the query projects. If that holds, the null-partition check at L204 declines every query on a table that has a `__HIVE_DEFAULT_PARTITION__` partition, including queries that don't select the partition column. That doesn't match "without the partition column projected the read stays native." Which one is right? If it's the comment, could we limit the check to partition columns the query actually references, e.g. via the parent project or `requiredSchema`? This only costs performance, so a follow-up is fine.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): [DNM] read Hudi file slices natively through hudi-rs in the Comet bundle [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #20170:
URL: https://github.com/apache/hudi/pull/20170#issuecomment-5922867426
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d22b68962e65929e1a07279bbe14ebebfc151735",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17398",
"triggerID" : "d22b68962e65929e1a07279bbe14ebebfc151735",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d22b68962e65929e1a07279bbe14ebebfc151735 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17398)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): read Hudi file slices natively through hudi-rs in the Comet bundle [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on PR #20170:
URL: https://github.com/apache/hudi/pull/20170#issuecomment-6476613139
That is right. Everything up to the file slices stays on the JVM: the Hudi relation lists partitions and files, prunes partitions, and the provider resolves the file slices as of the query instant and packs them into tasks. The plan that goes native carries, per task, the slice paths, the query instant, the projection and the filters. The hudi-rs side opens the base file and log files of each slice, merges them, null-fills columns added after the base file was written, and returns Arrow batches straight into Comet's operators. The illustration matches that boundary and would fit well in the RFC-133 doc.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): [DNM] read Hudi file slices natively through hudi-rs in the Comet bundle [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #20170:
URL: https://github.com/apache/hudi/pull/20170#discussion_r4360147328
##########
hudi-spark-datasource/hudi-spark-comet/src/main/scala/org/apache/hudi/comet/HudiCometScanProvider.scala:
##########
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.comet
+
+import org.apache.hudi.{HoodieFileIndex, HoodiePartitionCDCFileGroupMapping, HoodiePartitionFileSliceMapping}
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.{HoodieFileFormat, HoodieLogFile}
+import org.apache.hudi.storage.StoragePath
+
+import org.apache.comet.rules.CometExternalV1ScanProvider
+import org.apache.comet.serde.OperatorOuterClass
+import org.apache.comet.serde.QueryPlanSerde.serializeDataType
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.DynamicPruningExpression
+import org.apache.spark.sql.comet.{CometHudiNativeScanExec, SerializedPlan}
+import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.FileFormat
+import org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat
+import org.apache.spark.sql.types.{ArrayType, MapType, StructType}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
+
+/**
+ * Converts a Hudi data source V1 scan into a Comet native scan that reads file slices through
+ * hudi-rs. Registered with Comet's scan rule through the JDK ServiceLoader.
+ *
+ * Conversion is conservative: any shape outside the supported set (snapshot reads of Parquet
+ * tables whose partition columns are stored in the data files, with primitive output types)
+ * returns None, which leaves the plan on Hudi's JVM scan.
+ */
+class HudiCometScanProvider extends CometExternalV1ScanProvider with Logging {
+
+ override def supportsFormat(fileFormat: FileFormat): Boolean =
+ fileFormat.isInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+
+ override def convert(
+ session: SparkSession,
+ scanExec: FileSourceScanExec): Option[SparkPlan] = {
+ if (!session.conf
+ .getOption(HudiCometScanProvider.SCAN_ENABLED_KEY)
+ .exists(_.toBoolean)) {
+ return None
+ }
+ val format =
+ scanExec.relation.fileFormat.asInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+ if (format.isIncrementalQuery || format.isBootstrapTable ||
+ format.isMultipleBaseFileFormats || format.getRequiredFilters.nonEmpty) {
+ return None
+ }
+ if (scanExec.requiredSchema.isEmpty) {
+ return None
+ }
+ if (scanExec.partitionFilters.exists(
+ _.find(_.isInstanceOf[DynamicPruningExpression]).isDefined)) {
+ return None
+ }
Review Comment:
Partition filter that uses a scalar subquery is dropped, so rows from every partition come back.
E.g.
```
SELECT id FROM t WHERE dt = (SELECT max(dt) FROM t)
```
Not sure how to fix this though, maybe decline partition filter containing a `PlanExpression` for now? Then we figure out how to handle subqueries in the future?
##########
hudi-spark-datasource/hudi-spark-comet/src/main/scala/org/apache/hudi/comet/HudiCometScanProvider.scala:
##########
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.comet
+
+import org.apache.hudi.{HoodieFileIndex, HoodiePartitionCDCFileGroupMapping, HoodiePartitionFileSliceMapping}
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.{HoodieFileFormat, HoodieLogFile}
+import org.apache.hudi.storage.StoragePath
+
+import org.apache.comet.rules.CometExternalV1ScanProvider
+import org.apache.comet.serde.OperatorOuterClass
+import org.apache.comet.serde.QueryPlanSerde.serializeDataType
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.DynamicPruningExpression
+import org.apache.spark.sql.comet.{CometHudiNativeScanExec, SerializedPlan}
+import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.FileFormat
+import org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat
+import org.apache.spark.sql.types.{ArrayType, MapType, StructType}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
+
+/**
+ * Converts a Hudi data source V1 scan into a Comet native scan that reads file slices through
+ * hudi-rs. Registered with Comet's scan rule through the JDK ServiceLoader.
+ *
+ * Conversion is conservative: any shape outside the supported set (snapshot reads of Parquet
+ * tables whose partition columns are stored in the data files, with primitive output types)
+ * returns None, which leaves the plan on Hudi's JVM scan.
+ */
+class HudiCometScanProvider extends CometExternalV1ScanProvider with Logging {
+
+ override def supportsFormat(fileFormat: FileFormat): Boolean =
+ fileFormat.isInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+
+ override def convert(
+ session: SparkSession,
+ scanExec: FileSourceScanExec): Option[SparkPlan] = {
+ if (!session.conf
+ .getOption(HudiCometScanProvider.SCAN_ENABLED_KEY)
+ .exists(_.toBoolean)) {
+ return None
+ }
+ val format =
+ scanExec.relation.fileFormat.asInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+ if (format.isIncrementalQuery || format.isBootstrapTable ||
+ format.isMultipleBaseFileFormats || format.getRequiredFilters.nonEmpty) {
+ return None
+ }
+ if (scanExec.requiredSchema.isEmpty) {
+ return None
+ }
+ if (scanExec.partitionFilters.exists(
+ _.find(_.isInstanceOf[DynamicPruningExpression]).isDefined)) {
+ return None
+ }
+ if (scanExec.output.exists(attr =>
+ attr.dataType.isInstanceOf[StructType] || attr.dataType.isInstanceOf[ArrayType] ||
+ attr.dataType.isInstanceOf[MapType])) {
+ return None
+ }
+
+ val fileIndex = scanExec.relation.location match {
+ case index: HoodieFileIndex => index
+ case _ => return None
+ }
+ val tableConfig = fileIndex.metaClient.getTableConfig
+ val tableVersion = tableConfig.getTableVersion.versionCode()
+ // The versions the hudi-rs file group reader accepts. Merge-on-read
+ // additionally requires 8+, where log files are sealed per commit, so the
+ // planned file paths fully determine the data the reader must see.
+ if (!HudiCometScanProvider.SUPPORTED_TABLE_VERSIONS.contains(tableVersion) ||
+ (format.isMORTable && tableVersion < 8)) {
+ return None
+ }
+ if (tableConfig.shouldDropPartitionColumns()) {
+ return None
+ }
+ if (tableConfig.getBaseFileFormat != HoodieFileFormat.PARQUET) {
+ return None
+ }
+ // Every output column must come out of the data files: the native reader
+ // has no partition-value append path. Partition columns are stored in the
+ // files (dropped partition columns are gated out above), but the JVM read
+ // appends the value parsed from the partition path instead, and the two
+ // agree only when the key generator writes values into the path unchanged.
+ // Generators that transform values (timestamp-based, custom) stay on the
+ // JVM scan whenever a partition column is projected.
+ val partitionColumns = scanExec.relation.partitionSchema.fieldNames.toSet
+ val availableColumns = scanExec.relation.dataSchema.fieldNames.toSet ++ partitionColumns
+ if (!scanExec.output.forall(attr => availableColumns.contains(attr.name))) {
+ return None
+ }
+ if (scanExec.output.exists(attr => partitionColumns.contains(attr.name)) &&
+ !HudiCometScanProvider.IDENTITY_KEY_GENERATORS.contains(
+ Option(tableConfig.getKeyGeneratorClassName)
+ .getOrElse("org.apache.hudi.keygen.SimpleKeyGenerator"))) {
+ return None
+ }
Review Comment:
Can you please check if CUSTOM merge mode here works?
##########
hudi-spark-datasource/hudi-spark-comet/src/main/scala/org/apache/hudi/comet/HudiCometScanProvider.scala:
##########
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.comet
+
+import org.apache.hudi.{HoodieFileIndex, HoodiePartitionCDCFileGroupMapping, HoodiePartitionFileSliceMapping}
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.{HoodieFileFormat, HoodieLogFile}
+import org.apache.hudi.storage.StoragePath
+
+import org.apache.comet.rules.CometExternalV1ScanProvider
+import org.apache.comet.serde.OperatorOuterClass
+import org.apache.comet.serde.QueryPlanSerde.serializeDataType
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.DynamicPruningExpression
+import org.apache.spark.sql.comet.{CometHudiNativeScanExec, SerializedPlan}
+import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.FileFormat
+import org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat
+import org.apache.spark.sql.types.{ArrayType, MapType, StructType}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
+
+/**
+ * Converts a Hudi data source V1 scan into a Comet native scan that reads file slices through
+ * hudi-rs. Registered with Comet's scan rule through the JDK ServiceLoader.
+ *
+ * Conversion is conservative: any shape outside the supported set (snapshot reads of Parquet
+ * tables whose partition columns are stored in the data files, with primitive output types)
+ * returns None, which leaves the plan on Hudi's JVM scan.
+ */
+class HudiCometScanProvider extends CometExternalV1ScanProvider with Logging {
+
+ override def supportsFormat(fileFormat: FileFormat): Boolean =
+ fileFormat.isInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+
+ override def convert(
+ session: SparkSession,
+ scanExec: FileSourceScanExec): Option[SparkPlan] = {
+ if (!session.conf
+ .getOption(HudiCometScanProvider.SCAN_ENABLED_KEY)
+ .exists(_.toBoolean)) {
+ return None
+ }
+ val format =
+ scanExec.relation.fileFormat.asInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+ if (format.isIncrementalQuery || format.isBootstrapTable ||
+ format.isMultipleBaseFileFormats || format.getRequiredFilters.nonEmpty) {
+ return None
+ }
+ if (scanExec.requiredSchema.isEmpty) {
+ return None
+ }
+ if (scanExec.partitionFilters.exists(
+ _.find(_.isInstanceOf[DynamicPruningExpression]).isDefined)) {
+ return None
+ }
+ if (scanExec.output.exists(attr =>
+ attr.dataType.isInstanceOf[StructType] || attr.dataType.isInstanceOf[ArrayType] ||
+ attr.dataType.isInstanceOf[MapType])) {
+ return None
+ }
+
+ val fileIndex = scanExec.relation.location match {
+ case index: HoodieFileIndex => index
+ case _ => return None
+ }
+ val tableConfig = fileIndex.metaClient.getTableConfig
+ val tableVersion = tableConfig.getTableVersion.versionCode()
+ // The versions the hudi-rs file group reader accepts. Merge-on-read
+ // additionally requires 8+, where log files are sealed per commit, so the
+ // planned file paths fully determine the data the reader must see.
+ if (!HudiCometScanProvider.SUPPORTED_TABLE_VERSIONS.contains(tableVersion) ||
+ (format.isMORTable && tableVersion < 8)) {
+ return None
+ }
+ if (tableConfig.shouldDropPartitionColumns()) {
+ return None
+ }
+ if (tableConfig.getBaseFileFormat != HoodieFileFormat.PARQUET) {
+ return None
+ }
+ // Every output column must come out of the data files: the native reader
+ // has no partition-value append path. Partition columns are stored in the
+ // files (dropped partition columns are gated out above), but the JVM read
+ // appends the value parsed from the partition path instead, and the two
+ // agree only when the key generator writes values into the path unchanged.
+ // Generators that transform values (timestamp-based, custom) stay on the
+ // JVM scan whenever a partition column is projected.
+ val partitionColumns = scanExec.relation.partitionSchema.fieldNames.toSet
+ val availableColumns = scanExec.relation.dataSchema.fieldNames.toSet ++ partitionColumns
+ if (!scanExec.output.forall(attr => availableColumns.contains(attr.name))) {
+ return None
+ }
+ if (scanExec.output.exists(attr => partitionColumns.contains(attr.name)) &&
+ !HudiCometScanProvider.IDENTITY_KEY_GENERATORS.contains(
+ Option(tableConfig.getKeyGeneratorClassName)
+ .getOrElse("org.apache.hudi.keygen.SimpleKeyGenerator"))) {
+ return None
+ }
+
+ val baseUri = fileIndex.metaClient.getBasePath.toString
+ val slices = extractFileSlices(scanExec, baseUri, format.isMORTable) match {
+ case Some(s) => s
+ case None => return None
+ }
+
+ val scanHashCode = scanExec.canonicalized.hashCode()
+ val commonBuilder = OperatorOuterClass.HudiScanCommon
+ .newBuilder()
+ .setTableBaseUri(baseUri)
+ .setSessionTimezone(session.sessionState.conf.sessionLocalTimeZone)
+ .setScanHashCode(scanHashCode)
+ // Bounds the log-file scan to instants at or before the query instant,
+ // exactly as the JVM file group reader does with the same timestamp; a
+ // snapshot read passes the latest completed instant, a time-travel read
+ // the requested one. Base files need no bound: the driver's listing only
+ // names base files from admitted commits.
+ .putOptions("hoodie.read.end.timestamp", format.getQueryTimestamp)
Review Comment:
CMIIW, won't this cause the all read options on the query to be dropped except `"hoodie.read.end.timestamp"`.
##########
packaging/hudi-spark-native-bundle/comet/overlay/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -0,0 +1,1097 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.rules
+
+import java.lang.{Boolean => JBoolean}
+import java.net.URI
+import java.util.Locale
+import java.util.concurrent.ConcurrentHashMap
+
+import scala.collection.mutable
+import scala.collection.mutable.ListBuffer
+import scala.jdk.CollectionConverters._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{Attribute, DynamicPruningExpression, Expression, GenericInternalRow, InputFileBlockLength, InputFileBlockStart, InputFileName, PlanExpression}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.util.{sideBySide, ArrayBasedMapData, GenericArrayData, MetadataColumnHelper}
+import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getExistenceDefaultValues
+import org.apache.spark.sql.comet.{CometBatchScanExec, CometScanExec}
+import org.apache.spark.sql.execution.{FileSourceScanExec, InSubqueryExec, SparkPlan, SubqueryAdaptiveBroadcastExec}
+import org.apache.spark.sql.execution.datasources.HadoopFsRelation
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.datasources.v2.csv.CSVScan
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
+
+import org.apache.comet.{CometConf, DataTypeSupport, NativeBase}
+import org.apache.comet.CometConf._
+import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, isSpark35Plus, withFallbackReason, withFallbackReasons}
+import org.apache.comet.DataTypeSupport.isComplexType
+import org.apache.comet.iceberg.{CometIcebergNativeScanMetadata, IcebergReflection}
+import org.apache.comet.objectstore.NativeConfig
+import org.apache.comet.parquet.CometParquetUtils.{encryptionEnabled, isEncryptionConfigSupported}
+import org.apache.comet.serde.operator.{CometIcebergNativeScan, CometNativeScan}
+import org.apache.comet.shims.{CometTypeShim, ShimCometStreaming, ShimFileFormat, ShimSubqueryBroadcast}
+
+/**
+ * Spark physical optimizer rule for replacing Spark scans with Comet scans.
+ */
+case class CometScanRule(session: SparkSession)
+ extends Rule[SparkPlan]
+ with CometTypeShim
+ with ShimSubqueryBroadcast {
+
+ private lazy val showTransformations = CometConf.COMET_EXPLAIN_TRANSFORMATIONS.get()
+
+ override def apply(plan: SparkPlan): SparkPlan = {
+ val newPlan = _apply(plan)
+ if (showTransformations && !newPlan.fastEquals(plan)) {
+ logInfo(s"""
+ |=== Applying Rule $ruleName ===
+ |${sideBySide(plan.treeString, newPlan.treeString).mkString("\n")}
+ |""".stripMargin)
+ }
+ newPlan
+ }
+
+ private def _apply(plan: SparkPlan): SparkPlan = {
+ if (!isCometLoaded(conf)) return plan
+
+ // Comet does not support structured streaming. The parallel guard in
+ // CometExecRule only stops operator wrapping, so without this check we
+ // would still rewrite scans to CometScanExec in a streaming plan.
+ if (ShimCometStreaming.isStreamingPlan(plan)) return plan
+
+ def isSupportedScanNode(plan: SparkPlan): Boolean = plan match {
+ case _: FileSourceScanExec => true
+ case _: BatchScanExec => true
+ case _ => false
+ }
+
+ def isIcebergMetadataTable(scanExec: BatchScanExec): Boolean = {
+ // List of Iceberg metadata tables:
+ // https://iceberg.apache.org/docs/latest/spark-queries/#inspecting-tables
+ val metadataTableSuffix = Set(
+ "history",
+ "metadata_log_entries",
+ "snapshots",
+ "entries",
+ "files",
+ "manifests",
+ "partitions",
+ "position_deletes",
+ "all_data_files",
+ "all_delete_files",
+ "all_entries",
+ "all_manifests")
+
+ // Match case-insensitively: metadata tables surface lowercase via the path form
+ // (...metadata.json#all_manifests) but uppercase via the catalog-identifier form
+ // (db.table.ALL_DATA_FILES), and the latter must hit this gate too rather than fall through
+ // to reflection that fails on the metadata-table class.
+ val name = scanExec.table.name().toLowerCase(Locale.ROOT)
+ metadataTableSuffix.exists(name.endsWith)
+ }
+
+ val fullPlan = plan
+
+ def transformScan(scanNode: SparkPlan): SparkPlan = scanNode match {
+ // Tagged by CometSpark34AqeDppFallbackRule on Spark < 3.5 to keep a peer scan
+ // Spark-native for canonical symmetry in SMJ self-joins (SPARK-32509).
+ case scan if scan.getTagValue(CometScanRule.SKIP_COMET_SCAN_TAG).isDefined =>
+ withFallbackReason(scan, "AQE DPP region fallback (Spark < 3.5)")
+
+ case scan if !CometConf.COMET_NATIVE_SCAN_ENABLED.get(conf) =>
+ withFallbackReason(scan, "Comet Scan is not enabled")
+
+ // data source V1
+ case scanExec: FileSourceScanExec =>
+ transformV1Scan(fullPlan, scanExec)
+
+ // data source V2
+ case scanExec: BatchScanExec =>
+ if (isIcebergMetadataTable(scanExec)) {
+ withFallbackReason(scanExec, "Iceberg Metadata tables are not supported")
+ } else {
+ transformV2Scan(scanExec)
+ }
+ }
+
+ plan.transform {
+ case scan if isSupportedScanNode(scan) => transformScan(scan)
+ }
+ }
+
+ private def transformV1Scan(plan: SparkPlan, scanExec: FileSourceScanExec): SparkPlan = {
+ val metadataColNames = metadataCols(scanExec)
+ if (metadataColNames.nonEmpty) {
+ return withFallbackReason(
+ scanExec,
+ s"Metadata column(s) ${metadataColNames.mkString(", ")} is not supported")
+ }
+
+ // On Spark 3.4, injectQueryStageOptimizerRule is unavailable, so
+ // CometPlanAdaptiveDynamicPruningFilters cannot run. Fall back this scan to Spark so that
+ // Spark's PlanAdaptiveDynamicPruningFilters handles the SAB natively. Comet's narrower
+ // CometSpark34AqeDppFallbackRule (queryStagePrepRule on 3.4) then tags any matching BHJ's
+ // build-side broadcast so Spark's rule can match it via sameResult. See
+ // CometSpark34AqeDppFallbackRule's class docstring.
+ //
+ // On 3.5+, CometPlanAdaptiveDynamicPruningFilters rewrites SABs directly and this fallback
+ // is not needed.
+ if (!isSpark35Plus && scanExec.partitionFilters.exists(isAqeDynamicPruningFilter)) {
+ return withFallbackReason(scanExec, "AQE Dynamic Partition Pruning requires Spark 3.5+")
+ }
+
+ scanExec.relation match {
+ case r: HadoopFsRelation =>
+ if (!CometScanExec.isFileFormatSupported(r.fileFormat)) {
+ CometExternalV1ScanProvider.findForFormat(r.fileFormat) match {
Review Comment:
The hook runs before `#nativeScan()` guards, will this cause issues?
##########
hudi-spark-datasource/hudi-spark-comet/src/main/scala/org/apache/hudi/comet/HudiCometScanProvider.scala:
##########
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.comet
+
+import org.apache.hudi.{HoodieFileIndex, HoodiePartitionCDCFileGroupMapping, HoodiePartitionFileSliceMapping}
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.{HoodieFileFormat, HoodieLogFile}
+import org.apache.hudi.storage.StoragePath
+
+import org.apache.comet.rules.CometExternalV1ScanProvider
+import org.apache.comet.serde.OperatorOuterClass
+import org.apache.comet.serde.QueryPlanSerde.serializeDataType
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.DynamicPruningExpression
+import org.apache.spark.sql.comet.{CometHudiNativeScanExec, SerializedPlan}
+import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.FileFormat
+import org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat
+import org.apache.spark.sql.types.{ArrayType, MapType, StructType}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
+
+/**
+ * Converts a Hudi data source V1 scan into a Comet native scan that reads file slices through
+ * hudi-rs. Registered with Comet's scan rule through the JDK ServiceLoader.
+ *
+ * Conversion is conservative: any shape outside the supported set (snapshot reads of Parquet
+ * tables whose partition columns are stored in the data files, with primitive output types)
+ * returns None, which leaves the plan on Hudi's JVM scan.
+ */
+class HudiCometScanProvider extends CometExternalV1ScanProvider with Logging {
+
+ override def supportsFormat(fileFormat: FileFormat): Boolean =
+ fileFormat.isInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+
+ override def convert(
+ session: SparkSession,
+ scanExec: FileSourceScanExec): Option[SparkPlan] = {
+ if (!session.conf
+ .getOption(HudiCometScanProvider.SCAN_ENABLED_KEY)
+ .exists(_.toBoolean)) {
+ return None
+ }
+ val format =
+ scanExec.relation.fileFormat.asInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+ if (format.isIncrementalQuery || format.isBootstrapTable ||
+ format.isMultipleBaseFileFormats || format.getRequiredFilters.nonEmpty) {
+ return None
+ }
+ if (scanExec.requiredSchema.isEmpty) {
+ return None
+ }
+ if (scanExec.partitionFilters.exists(
+ _.find(_.isInstanceOf[DynamicPruningExpression]).isDefined)) {
+ return None
+ }
+ if (scanExec.output.exists(attr =>
+ attr.dataType.isInstanceOf[StructType] || attr.dataType.isInstanceOf[ArrayType] ||
+ attr.dataType.isInstanceOf[MapType])) {
+ return None
+ }
+
+ val fileIndex = scanExec.relation.location match {
+ case index: HoodieFileIndex => index
+ case _ => return None
+ }
+ val tableConfig = fileIndex.metaClient.getTableConfig
+ val tableVersion = tableConfig.getTableVersion.versionCode()
+ // The versions the hudi-rs file group reader accepts. Merge-on-read
+ // additionally requires 8+, where log files are sealed per commit, so the
+ // planned file paths fully determine the data the reader must see.
+ if (!HudiCometScanProvider.SUPPORTED_TABLE_VERSIONS.contains(tableVersion) ||
+ (format.isMORTable && tableVersion < 8)) {
+ return None
+ }
+ if (tableConfig.shouldDropPartitionColumns()) {
+ return None
+ }
+ if (tableConfig.getBaseFileFormat != HoodieFileFormat.PARQUET) {
+ return None
+ }
+ // Every output column must come out of the data files: the native reader
+ // has no partition-value append path. Partition columns are stored in the
+ // files (dropped partition columns are gated out above), but the JVM read
+ // appends the value parsed from the partition path instead, and the two
+ // agree only when the key generator writes values into the path unchanged.
+ // Generators that transform values (timestamp-based, custom) stay on the
+ // JVM scan whenever a partition column is projected.
+ val partitionColumns = scanExec.relation.partitionSchema.fieldNames.toSet
+ val availableColumns = scanExec.relation.dataSchema.fieldNames.toSet ++ partitionColumns
+ if (!scanExec.output.forall(attr => availableColumns.contains(attr.name))) {
+ return None
+ }
+ if (scanExec.output.exists(attr => partitionColumns.contains(attr.name)) &&
+ !HudiCometScanProvider.IDENTITY_KEY_GENERATORS.contains(
+ Option(tableConfig.getKeyGeneratorClassName)
+ .getOrElse("org.apache.hudi.keygen.SimpleKeyGenerator"))) {
+ return None
+ }
Review Comment:
We have quite a bit of issues regarding key generators in the past, can we test for null or empty partition values?
WE need to check empty string partitions like `dt=''` with `__HIVE_DEFAULT_PARTITION__` interactions here.
##########
packaging/hudi-spark-native-bundle/comet/overlay/native/core/src/execution/operators/hudi_scan.rs:
##########
@@ -0,0 +1,245 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! A native Hudi table scan, reading file slices through hudi-rs.
+//!
+//! The Spark driver plans the read (file listing, partition pruning, slice
+//! grouping as of the query instant) and ships each task its file slices plus
+//! the table base URI. hudi-rs opens the table from `hoodie.properties` under
+//! that URI, so merge mode, ordering field, and table version never travel in
+//! the plan; this operator only reads the named slices and projects the
+//! requested columns.
+//!
+//! Every batch crosses from hudi-rs into this workspace through the Arrow C
+//! data interface, whose layout is fixed by the Arrow specification and is
+//! identical across arrow crate majors; the transfer is pointer-level, not a
+//! copy. hudi-rs currently pins the same arrow major, so the crossing is a
+//! same-crate round trip; if the lines diverge again, only the export side
+//! switches to a renamed dep on hudi-rs's line.
+
+use std::collections::HashMap;
+use std::fmt;
+use std::sync::Arc;
+
+use arrow::array::{Array as _, RecordBatch, RecordBatchOptions, StructArray};
+use arrow::compute::cast;
+use arrow::datatypes::SchemaRef;
+use datafusion::common::{DataFusionError, Result as DFResult};
+use datafusion::execution::{SendableRecordBatchStream, TaskContext};
+use datafusion::physical_expr::EquivalenceProperties;
+use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
+use datafusion::physical_plan::metrics::{
+ BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet,
+};
+use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
+use datafusion::physical_plan::{
+ DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PlanProperties,
+};
+use futures::{StreamExt, TryStreamExt};
+use hudi::file_group::reader::FileGroupReader;
+use hudi::table::ReadOptions;
+
+/// One file slice assigned to this partition: a base file and the log files
+/// whose records amend it.
+#[derive(Debug, Clone)]
+pub struct HudiFileSliceSpec {
+ pub base_file_path: String,
+ pub log_file_paths: Vec<String>,
+}
+
+#[derive(Debug)]
+pub struct HudiScanExec {
+ table_base_uri: String,
+ options: HashMap<String, String>,
+ output_schema: SchemaRef,
+ file_slices: Vec<HudiFileSliceSpec>,
+ plan_properties: Arc<PlanProperties>,
+ metrics: ExecutionPlanMetricsSet,
+}
+
+impl HudiScanExec {
+ pub fn new(
+ table_base_uri: String,
+ options: HashMap<String, String>,
+ output_schema: SchemaRef,
+ file_slices: Vec<HudiFileSliceSpec>,
+ ) -> Self {
+ let plan_properties = Arc::new(PlanProperties::new(
+ EquivalenceProperties::new(Arc::clone(&output_schema)),
+ Partitioning::UnknownPartitioning(1),
+ EmissionType::Incremental,
+ Boundedness::Bounded,
+ ));
+ Self {
+ table_base_uri,
+ options,
+ output_schema,
+ file_slices,
+ plan_properties,
+ metrics: ExecutionPlanMetricsSet::new(),
+ }
+ }
+}
+
+/// Moves a batch from hudi-rs's arrow into this workspace's arrow over the
+/// Arrow C data interface.
+fn batch_across_ffi(batch: RecordBatch) -> DFResult<RecordBatch> {
+ let data = StructArray::from(batch).into_data();
+ let (ffi_array, ffi_schema) = arrow::ffi::to_ffi(&data)
+ .map_err(|e| DataFusionError::Execution(format!("Hudi scan arrow export failed: {e}")))?;
+ // SAFETY: FFI_ArrowArray and FFI_ArrowSchema are the #[repr(C)] structs of
+ // the Arrow C data interface; their layout is defined by the Arrow
+ // specification and is identical across arrow crate versions. Ownership
+ // moves with the structs, so the release callbacks fire exactly once.
+ let data = unsafe { arrow::ffi::from_ffi(ffi_array, &ffi_schema) }
+ .map_err(|e| DataFusionError::Execution(format!("Hudi scan arrow import failed: {e}")))?;
+ Ok(RecordBatch::from(StructArray::from(data)))
+}
+
+/// Reorders and casts a batch's columns to the scan's output schema. hudi-rs
+/// projects the requested columns but keeps the table schema's column order
+/// and physical types; Spark expects its own order and logical types.
+fn align_to_schema(batch: RecordBatch, schema: &SchemaRef) -> DFResult<RecordBatch> {
+ let num_rows = batch.num_rows();
+ let mut columns = Vec::with_capacity(schema.fields().len());
+ for field in schema.fields() {
+ let index = batch.schema().index_of(field.name()).map_err(|_| {
+ DataFusionError::Execution(format!(
+ "Hudi scan did not produce required column {}",
+ field.name()
+ ))
+ })?;
+ let column = batch.column(index);
+ let column = if column.data_type() == field.data_type() {
+ Arc::clone(column)
+ } else {
+ cast(column, field.data_type()).map_err(|e| {
+ DataFusionError::Execution(format!(
+ "Hudi scan cannot cast column {} from {} to {}: {e}",
+ field.name(),
+ column.data_type(),
+ field.data_type()
+ ))
+ })?
+ };
+ columns.push(column);
+ }
+ RecordBatch::try_new_with_options(
+ Arc::clone(schema),
+ columns,
+ &RecordBatchOptions::new().with_row_count(Some(num_rows)),
+ )
+ .map_err(|e| DataFusionError::Execution(format!("Hudi scan batch assembly failed: {e}")))
+}
+
+impl DisplayAs for HudiScanExec {
+ fn fmt_as(&self, _t: DisplayFormatType, f: &mut fmt::Formatter) -> fmt::Result {
+ write!(
+ f,
+ "HudiScanExec: table={}, file_slices={}",
+ self.table_base_uri,
+ self.file_slices.len()
+ )
+ }
+}
+
+impl ExecutionPlan for HudiScanExec {
+ fn name(&self) -> &str {
+ "HudiScanExec"
+ }
+
+ fn schema(&self) -> SchemaRef {
+ Arc::clone(&self.output_schema)
+ }
+
+ fn properties(&self) -> &Arc<PlanProperties> {
+ &self.plan_properties
+ }
+
+ fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+ vec![]
+ }
+
+ fn with_new_children(
+ self: Arc<Self>,
+ _children: Vec<Arc<dyn ExecutionPlan>>,
+ ) -> DFResult<Arc<dyn ExecutionPlan>> {
+ Ok(self)
+ }
+
+ fn execute(
+ &self,
+ _partition: usize,
+ _context: Arc<TaskContext>,
+ ) -> DFResult<SendableRecordBatchStream> {
+ let base_uri = self.table_base_uri.clone();
+ let options = self.options.clone();
+ let schema = Arc::clone(&self.output_schema);
+ let slices = self.file_slices.clone();
+ let projection: Vec<String> = schema
+ .fields()
+ .iter()
+ .map(|f| f.name().clone())
+ .collect();
+ let baseline = BaselineMetrics::new(&self.metrics, 0);
+
+ let batches = futures::stream::once(async move {
+ let reader = FileGroupReader::new_with_options(
+ &base_uri,
+ options.iter().map(|(k, v)| (k.as_str(), v.as_str())),
+ )
+ .await
+ .map_err(|e| {
+ DataFusionError::Execution(format!("failed to open Hudi table {base_uri}: {e}"))
+ })?;
+ let read_options = ReadOptions {
+ projection: Some(projection),
+ ..Default::default()
+ };
+ let mut out = Vec::with_capacity(slices.len());
+ for slice in &slices {
+ let batch = reader
Review Comment:
This is rust territory, so i might be wrong, do help to check if there's column pruning here where we only read the columns that are required.
I don't see any batch-sizing, so i assume the entire slice comes back as one batch, will that lead to memory or performance issues if comet decides to spill to disk?
##########
hudi-spark-datasource/hudi-spark-comet/src/main/scala/org/apache/hudi/comet/HudiCometScanProvider.scala:
##########
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.comet
+
+import org.apache.hudi.{HoodieFileIndex, HoodiePartitionCDCFileGroupMapping, HoodiePartitionFileSliceMapping}
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.model.{HoodieFileFormat, HoodieLogFile}
+import org.apache.hudi.storage.StoragePath
+
+import org.apache.comet.rules.CometExternalV1ScanProvider
+import org.apache.comet.serde.OperatorOuterClass
+import org.apache.comet.serde.QueryPlanSerde.serializeDataType
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.DynamicPruningExpression
+import org.apache.spark.sql.comet.{CometHudiNativeScanExec, SerializedPlan}
+import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.FileFormat
+import org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat
+import org.apache.spark.sql.types.{ArrayType, MapType, StructType}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
+
+/**
+ * Converts a Hudi data source V1 scan into a Comet native scan that reads file slices through
+ * hudi-rs. Registered with Comet's scan rule through the JDK ServiceLoader.
+ *
+ * Conversion is conservative: any shape outside the supported set (snapshot reads of Parquet
+ * tables whose partition columns are stored in the data files, with primitive output types)
+ * returns None, which leaves the plan on Hudi's JVM scan.
+ */
+class HudiCometScanProvider extends CometExternalV1ScanProvider with Logging {
+
+ override def supportsFormat(fileFormat: FileFormat): Boolean =
+ fileFormat.isInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+
+ override def convert(
+ session: SparkSession,
+ scanExec: FileSourceScanExec): Option[SparkPlan] = {
+ if (!session.conf
+ .getOption(HudiCometScanProvider.SCAN_ENABLED_KEY)
+ .exists(_.toBoolean)) {
+ return None
+ }
+ val format =
+ scanExec.relation.fileFormat.asInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+ if (format.isIncrementalQuery || format.isBootstrapTable ||
+ format.isMultipleBaseFileFormats || format.getRequiredFilters.nonEmpty) {
+ return None
+ }
+ if (scanExec.requiredSchema.isEmpty) {
+ return None
+ }
+ if (scanExec.partitionFilters.exists(
+ _.find(_.isInstanceOf[DynamicPruningExpression]).isDefined)) {
+ return None
+ }
+ if (scanExec.output.exists(attr =>
+ attr.dataType.isInstanceOf[StructType] || attr.dataType.isInstanceOf[ArrayType] ||
+ attr.dataType.isInstanceOf[MapType])) {
+ return None
+ }
+
+ val fileIndex = scanExec.relation.location match {
+ case index: HoodieFileIndex => index
+ case _ => return None
+ }
+ val tableConfig = fileIndex.metaClient.getTableConfig
+ val tableVersion = tableConfig.getTableVersion.versionCode()
+ // The versions the hudi-rs file group reader accepts. Merge-on-read
+ // additionally requires 8+, where log files are sealed per commit, so the
+ // planned file paths fully determine the data the reader must see.
+ if (!HudiCometScanProvider.SUPPORTED_TABLE_VERSIONS.contains(tableVersion) ||
+ (format.isMORTable && tableVersion < 8)) {
+ return None
+ }
+ if (tableConfig.shouldDropPartitionColumns()) {
+ return None
+ }
+ if (tableConfig.getBaseFileFormat != HoodieFileFormat.PARQUET) {
+ return None
+ }
+ // Every output column must come out of the data files: the native reader
+ // has no partition-value append path. Partition columns are stored in the
+ // files (dropped partition columns are gated out above), but the JVM read
+ // appends the value parsed from the partition path instead, and the two
+ // agree only when the key generator writes values into the path unchanged.
+ // Generators that transform values (timestamp-based, custom) stay on the
+ // JVM scan whenever a partition column is projected.
+ val partitionColumns = scanExec.relation.partitionSchema.fieldNames.toSet
+ val availableColumns = scanExec.relation.dataSchema.fieldNames.toSet ++ partitionColumns
+ if (!scanExec.output.forall(attr => availableColumns.contains(attr.name))) {
+ return None
+ }
+ if (scanExec.output.exists(attr => partitionColumns.contains(attr.name)) &&
+ !HudiCometScanProvider.IDENTITY_KEY_GENERATORS.contains(
+ Option(tableConfig.getKeyGeneratorClassName)
+ .getOrElse("org.apache.hudi.keygen.SimpleKeyGenerator"))) {
+ return None
+ }
+
+ val baseUri = fileIndex.metaClient.getBasePath.toString
+ val slices = extractFileSlices(scanExec, baseUri, format.isMORTable) match {
+ case Some(s) => s
+ case None => return None
+ }
+
+ val scanHashCode = scanExec.canonicalized.hashCode()
+ val commonBuilder = OperatorOuterClass.HudiScanCommon
+ .newBuilder()
+ .setTableBaseUri(baseUri)
+ .setSessionTimezone(session.sessionState.conf.sessionLocalTimeZone)
+ .setScanHashCode(scanHashCode)
+ // Bounds the log-file scan to instants at or before the query instant,
+ // exactly as the JVM file group reader does with the same timestamp; a
+ // snapshot read passes the latest completed instant, a time-travel read
+ // the requested one. Base files need no bound: the driver's listing only
+ // names base files from admitted commits.
+ .putOptions("hoodie.read.end.timestamp", format.getQueryTimestamp)
+ for (attr <- scanExec.output) {
+ val field = OperatorOuterClass.SparkStructField
+ .newBuilder()
+ .setName(attr.name)
+ .setNullable(attr.nullable)
+ serializeDataType(attr.dataType) match {
+ case Some(dataType) => field.setDataType(dataType)
+ case None => return None
+ }
+ commonBuilder.addRequiredSchema(field)
+ }
+ val commonBytes = commonBuilder.build().toByteArray
+
+ val perPartitionBytes = slices.map { slice =>
Review Comment:
IIUC, every slice now = one task? So 1k file slices = 1k tasks.
IIRC, Trino has bin-packing and splitting, so we need to pay attention to that to prevent unnecessary file lookup or file open/read operations.
IIUC, there's no splitting now, so a large base file will also = 1 task, where Trino will split (for performance) is this intended?
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): [DNM] read Hudi file slices natively through hudi-rs in the Comet bundle [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #20170:
URL: https://github.com/apache/hudi/pull/20170#issuecomment-5921938294
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "d22b68962e65929e1a07279bbe14ebebfc151735",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d22b68962e65929e1a07279bbe14ebebfc151735",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d22b68962e65929e1a07279bbe14ebebfc151735 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(spark): [DNM] read Hudi file slices natively through hudi-rs in the Comet bundle [hudi]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #20170:
URL: https://github.com/apache/hudi/pull/20170#issuecomment-6353858600
## [Codecov](https://app.codecov.io/gh/apache/hudi/pull/20170?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 76.46%. Comparing base ([`5cce63e`](https://app.codecov.io/gh/apache/hudi/commit/5cce63e83c482e8c0dcd686919d00a1527b5ec9e?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`f21b39e`](https://app.codecov.io/gh/apache/hudi/commit/f21b39e6367126cb8d3303ade61cc914f6567174?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 3 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #20170 +/- ##
=============================================
+ Coverage 56.75% 76.46% +19.70%
- Complexity 26066 33889 +7823
=============================================
Files 2573 2574 +1
Lines 146420 146723 +303
Branches 17929 17988 +59
=============================================
+ Hits 83107 112190 +29083
+ Misses 57397 26245 -31152
- Partials 5916 8288 +2372
```
| [Components](https://app.codecov.io/gh/apache/hudi/pull/20170/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hudi-common](https://app.codecov.io/gh/apache/hudi/pull/20170/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `82.05% <ø> (+2.78%)` | :arrow_up: |
| [hudi-client](https://app.codecov.io/gh/apache/hudi/pull/20170/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `77.81% <ø> (+13.43%)` | :arrow_up: |
| [hudi-flink](https://app.codecov.io/gh/apache/hudi/pull/20170/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.49% <ø> (-1.63%)` | :arrow_down: |
| [hudi-spark-datasource](https://app.codecov.io/gh/apache/hudi/pull/20170/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `61.54% <ø> (+46.30%)` | :arrow_up: |
| [hudi-utilities](https://app.codecov.io/gh/apache/hudi/pull/20170/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `77.50% <ø> (+76.36%)` | :arrow_up: |
| [hudi-cli](https://app.codecov.io/gh/apache/hudi/pull/20170/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.92% <ø> (+53.71%)` | :arrow_up: |
| [hudi-hadoop](https://app.codecov.io/gh/apache/hudi/pull/20170/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `68.86% <ø> (+1.70%)` | :arrow_up: |
| [hudi-sync](https://app.codecov.io/gh/apache/hudi/pull/20170/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `76.80% <ø> (+0.44%)` | :arrow_up: |
| [hudi-io](https://app.codecov.io/gh/apache/hudi/pull/20170/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.38% <ø> (+0.97%)` | :arrow_up: |
| [hudi-timeline-service](https://app.codecov.io/gh/apache/hudi/pull/20170/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `76.58% <ø> (-1.55%)` | :arrow_down: |
| [hudi-cloud](https://app.codecov.io/gh/apache/hudi/pull/20170/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.01% <ø> (ø)` | |
| [hudi-kafka-connect](https://app.codecov.io/gh/apache/hudi/pull/20170/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.90% <ø> (+57.90%)` | :arrow_up: |
| [Flag](https://app.codecov.io/gh/apache/hudi/pull/20170/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [flink-integration-tests](https://app.codecov.io/gh/apache/hudi/pull/20170/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `36.34% <ø> (-13.30%)` | :arrow_down: |
| [hadoop-mr-java-client](https://app.codecov.io/gh/apache/hudi/pull/20170/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `43.94% <ø> (+0.04%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/hudi/pull/20170?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...parquet/HoodieFileGroupReaderBasedFileFormat.scala](https://app.codecov.io/gh/apache/hudi/pull/20170?src=pr&el=tree&filepath=hudi-spark-datasource%2Fhudi-spark-common%2Fsrc%2Fmain%2Fscala%2Forg%2Fapache%2Fspark%2Fsql%2Fexecution%2Fdatasources%2Fparquet%2FHoodieFileGroupReaderBasedFileFormat.scala&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1zcGFyay1kYXRhc291cmNlL2h1ZGktc3BhcmstY29tbW9uL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvc3Bhcmsvc3FsL2V4ZWN1dGlvbi9kYXRhc291cmNlcy9wYXJxdWV0L0hvb2RpZUZpbGVHcm91cFJlYWRlckJhc2VkRmlsZUZvcm1hdC5zY2FsYQ==) | `67.21% <ø> (+14.75%)` | :arrow_up: |
... and [973 files with indirect coverage changes](https://app.codecov.io/gh/apache/hudi/pull/20170/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org