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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
