hudi-agent commented on code in PR #18726: URL: https://github.com/apache/hudi/pull/18726#discussion_r3818967888
########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/analysis/TestHoodieIncrementalRelationIdentifier.scala: ########## @@ -0,0 +1,182 @@ +/* + * 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.spark.sql.hudi.analysis + +import org.apache.hudi.{HoodieFileIndex, HoodieIncrementalFileIndex, ScalaAssertionSupport} +import org.apache.hudi.HoodieConversionUtils.toJavaOption +import org.apache.hudi.common.table.HoodieTableMetaClient +import org.apache.hudi.common.table.timeline.HoodieInstant +import org.apache.hudi.hadoop.fs.HadoopFSUtils +import org.apache.hudi.testutils.HoodieClientTestBase +import org.apache.hudi.util.JFunction + +import org.apache.spark.sql.{SparkSession, SparkSessionExtensions} +import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation} +import org.apache.spark.sql.hudi.HoodieSparkSessionExtension +import org.junit.jupiter.api.{BeforeEach, Test} +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue} +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.ValueSource + +import java.util.function.Consumer + +/** + * Verifies that [[HoodieIncrementalRelationIdentifier]] enriches path-based incremental + * reads (and only those), so that lineage tooling sees a real table identifier instead + * of falling back to the relation's class name. + */ +class TestHoodieIncrementalRelationIdentifier extends HoodieClientTestBase with ScalaAssertionSupport { + + private var spark: SparkSession = _ + + @BeforeEach + override def setUp() { + setTableName("hoodie_incr_id_test") + initPath() + initSparkContexts() + spark = sqlContext.sparkSession + } + + override def getSparkSessionExtensionsInjector: org.apache.hudi.common.util.Option[Consumer[SparkSessionExtensions]] = + toJavaOption( + Some( + JFunction.toJavaConsumer((receiver: SparkSessionExtensions) => new HoodieSparkSessionExtension().apply(receiver))) + ) + + @ParameterizedTest + @ValueSource(strings = Array("cow", "mor")) + def testPathBasedIncrementalReadGetsCatalogTable(tableType: String): Unit = { + val tablePath = s"$basePath/$tableName" + createAndPopulateTable(tableType, tablePath) + + val firstInstant = firstCompletedInstantTime(tablePath) + + val df = spark.read.format("hudi") + .option("hoodie.datasource.query.type", "incremental") + .option("hoodie.datasource.read.begin.instanttime", firstInstant) + .load(tablePath) + + val analyzed = df.queryExecution.analyzed + + val lrOpt = analyzed.collectFirst { case lr: LogicalRelation => lr } + assertTrue(lrOpt.isDefined, + s"Expected a LogicalRelation in analyzed plan, got:\n$analyzed") + val lr = lrOpt.get + + // Sanity: confirm the path under test really exercises the incremental file index. + val location = lr.relation.asInstanceOf[HadoopFsRelation].location + assertTrue(location.isInstanceOf[HoodieIncrementalFileIndex], + s"Expected HoodieIncrementalFileIndex, got: ${location.getClass.getName}") + + assertTrue(lr.catalogTable.isDefined, + s"Expected catalogTable to be populated by HoodieIncrementalRelationIdentifier, got None") + + val ct = lr.catalogTable.get + assertEquals(tableName, ct.identifier.table, + s"Expected catalogTable identifier to use Hudi table name '$tableName', got '${ct.identifier.table}'") + assertEquals(Some("hudi"), ct.provider, + s"Expected provider 'hudi', got '${ct.provider}'") + assertTrue(ct.storage.locationUri.isDefined, + "Expected catalogTable.storage.locationUri to be populated from metaClient.getBasePath") + assertTrue(ct.schema.fields.nonEmpty, + "Expected catalogTable.schema to mirror the relation's resolved output schema") + } + + @Test + def testCatalogRegisteredIncrementalReadIsNotMutated(): Unit = { + spark.sql( + s""" Review Comment: 🤖 nit: the method name says `IncrementalRead` but the test body runs a plain `SELECT * FROM tableName` snapshot query, not an incremental one. Could you rename it to something like `testCatalogRegisteredSnapshotReadIsNotEnriched` so the name matches what the test actually exercises? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala: ########## @@ -620,3 +624,48 @@ case class HoodiePostAnalysisRule(sparkSession: SparkSession) extends Rule[Logic } } } + +/** + * Stamps a synthesized [[CatalogTable]] (table name, base path, schema) onto path-based + * Hudi reads whose underlying file index is incremental or CDC. Without it, lineage and + * governance tooling sees `LogicalRelation.catalogTable = None` and falls back to the + * relation's class name as the dataset identifier -- useless for tracking which table + * an incremental query came from. + * + * Scope is intentionally limited to incremental and CDC reads: + * - Catalog-registered reads already have `catalogTable` populated. + * - Path-based snapshot reads have a working file-path-based fallback in existing + * lineage tooling; changing their behavior is a separate decision. + */ +object HoodieIncrementalRelationIdentifier extends Rule[LogicalPlan] { + override def apply(plan: LogicalPlan): LogicalPlan = + AnalysisHelper.allowInvokingTransformsInAnalyzer { Review Comment: 🤖 Confirming this with concrete evidence — the arity really does differ by profile, and `hudi-spark` is compiled against each Spark profile so a hardcoded 4-arg pattern won't survive the Spark 4 builds: - Spark 3.x: 4-arg, e.g. `HoodieSpark35Analysis.scala:111` — `case LogicalRelation(h: HadoopFsRelation, _, catalogTable, _)` - Spark 4.0/4.1/4.2: 5-arg, e.g. `HoodieSpark40Analysis.scala:92`, `HoodieSpark41Analysis.scala:92`, `Spark4HoodiePruneFileSourcePartitions.scala:44` — all `LogicalRelation(..., _, _, _, _)` So `LogicalRelation(hfsr, _, None, _)` here will compile under 3.x but fail under the 4.x profiles. The type-pattern-plus-guard approach (`case lr: LogicalRelation if lr.catalogTable.isEmpty && lr.relation.isInstanceOf[HadoopFsRelation]`) sidesteps the arity difference and matches how `BaseSpark4Adapter`/`BaseSpark3Adapter` already stay profile-agnostic. -- 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]
