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