voonhous commented on code in PR #18726:
URL: https://github.com/apache/hudi/pull/18726#discussion_r3922237105
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala:
##########
@@ -639,3 +643,70 @@ 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 {
+ plan transform {
+ // Type pattern + nested match avoids destructuring `LogicalRelation`,
whose
+ // case-class arity differs between Spark 3.x (4 args) and Spark 4.x
(5 args). This
+ // rule lives in `hudi-spark`, which is compiled against every
supported profile.
+ case lr: LogicalRelation if lr.catalogTable.isEmpty =>
+ lr.relation match {
+ case fsRelation: HadoopFsRelation =>
+ incrementalOrCDCIndex(fsRelation.location)
+ .flatMap(index => buildCatalogTable(index.metaClient,
lr.schema))
+ .map(catalogTable => lr.copy(catalogTable =
Some(catalogTable)))
+ .getOrElse(lr)
+ case _ => lr
+ }
+ }
+ }
+
+ /**
+ * Narrows a [[FileIndex]] to the incremental/CDC Hudi indexes this rule
handles, or
+ * `None` for anything else. Both are [[HoodieFileIndex]] subclasses, so the
widening is
+ * checked by the compiler rather than by an `asInstanceOf` that a future
third index
+ * type could silently break.
+ */
+ private def incrementalOrCDCIndex(location: FileIndex):
Option[HoodieFileIndex] =
+ location match {
+ case index: HoodieIncrementalFileIndex => Some(index)
+ case index: HoodieCDCFileIndex => Some(index)
+ case _ => None
+ }
+
+ private def buildCatalogTable(
+ metaClient: HoodieTableMetaClient,
+ schema: StructType): Option[CatalogTable] = {
+ val tableConfig = metaClient.getTableConfig
+ // `hoodie.table.name` is required for a valid Hudi table, but if it is
somehow unset
+ // leave `catalogTable` as `None` -- the pre-existing behavior -- rather
than synthesize
+ // a `TableIdentifier(null)` that would surface downstream as a garbage
dataset name.
+ Option(tableConfig.getTableName).filter(_.nonEmpty).map { tableName =>
+ // Falls back to Spark's `default` database when `hoodie.database.name`
is unset --
+ // matches existing path-based DataFrame read behavior.
+ val dbName = Option(tableConfig.getDatabaseName).filter(_.nonEmpty)
+ CatalogTable(
+ identifier = TableIdentifier(tableName, dbName),
Review Comment:
**minor:** this name is `hoodie.properties`-derived and can diverge from the
catalog identity: `CREATE TABLE ... LOCATION` over an existing table leaves
name/db untouched (`HoodieCatalogTable.scala:204,234-235` set them only when
`!hoodieTableExists`; `TestCreateTable.scala:240-313` registers one path under
three names), and DataFrame-written tables default `hoodie.database.name` to
`""` (`HoodieSparkSqlWriter.scala:241`), so they stamp unqualified. Not
blocking. Could the scaladoc say `storage.locationUri` is the stable key and
the identifier may not be catalog-resolvable?
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala:
##########
@@ -639,3 +643,70 @@ 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.
Review Comment:
**minor:** OpenLineage never falls back to a class name:
`LogicalRelationDatasetBuilder` checks `catalogTable.isDefined` first, then
handles any `HadoopFsRelation` via `relation.location().rootPaths()` (unhandled
relations throw). Since #17457 (`0c0c402a9a03`, 2025-12-07) incremental/CDC
reads are `HadoopFsRelation`s, so for OL the delta is: same path dataset, plus
a table symlink, minus the `size`/`fileCount` input facets (the catalogTable
branch takes them from `catalogTable.stats` / `numFiles`, both absent here).
The snapshot-exclusion rationale applies equally to incremental reads. Not
blocking. Could the premise be restated as that delta and the scope
re-justified on it?
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala:
##########
@@ -126,7 +129,8 @@ object HoodieAnalysis extends SparkAdapterSupport {
// NOTE: By default all commands are converted into corresponding Hudi
implementations during
// "post-hoc resolution" phase
session => ResolveImplementations(session),
- session => HoodiePostAnalysisRule(session)
+ session => HoodiePostAnalysisRule(session),
+ _ => HoodieIncrementalRelationIdentifier
Review Comment:
**major:** the two halves do not compose for a MERGE source:
`ResolveImplementations` runs first in this `Once` batch and turns the
statement into a leaf whose `mapChildren` is `final ... = this`
(`HoodieLeafLike.scala:27`), so `plan transform` never reaches a
`hudi_table_changes('/path', ...)` source under `MERGE INTO ... USING`, and
`innerChildren` then publishes that unstamped source. INSERT/CTAS sources are
reached (`DataWritingCommand.child = query`); only MERGE misses. The other two
post-hoc rules are root-only `plan match` on commands, so could this rule be
registered first, with a MERGE-with-TVF-source test asserting the inner source
relation is stamped?
##########
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 testCatalogRegisteredReadIsNotMutated(): Unit = {
+ spark.sql(
+ s"""
+ |CREATE TABLE $tableName (
+ | id int,
+ | name string,
+ | ts long
+ |) USING hudi
+ |TBLPROPERTIES (
+ | type = 'cow',
+ | primaryKey = 'id',
+ | orderingFields = 'ts'
+ |)
+ |LOCATION '$basePath/$tableName'
+ """.stripMargin)
+ spark.sql(s"INSERT INTO $tableName VALUES (1, 'a1', 1000)")
+
+ val analyzed = spark.sql(s"SELECT * FROM
$tableName").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
+
+ assertTrue(lr.catalogTable.isDefined, "Expected catalogTable from catalog
registration")
+ assertEquals(tableName, lr.catalogTable.get.identifier.table)
+ }
+
+ @Test
+ def testSnapshotPathBasedReadIsNotEnriched(): Unit = {
+ val tablePath = s"$basePath/$tableName"
+ createAndPopulateTable("cow", tablePath)
+
+ val df = spark.read.format("hudi").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
+
+ val location = lr.relation.asInstanceOf[HadoopFsRelation].location
+ assertFalse(location.isInstanceOf[HoodieIncrementalFileIndex],
+ s"Snapshot read should not produce a HoodieIncrementalFileIndex, got:
${location.getClass.getName}")
+ assertTrue(location.isInstanceOf[HoodieFileIndex],
+ s"Snapshot read should produce a HoodieFileIndex, got:
${location.getClass.getName}")
+ assertFalse(lr.catalogTable.isDefined,
+ "Snapshot path-based read must not be enriched by
HoodieIncrementalRelationIdentifier")
+ }
+
+ // --- helpers ---
+
+ private def createAndPopulateTable(tableType: String, tablePath: String):
Unit = {
+ spark.sql(
+ s"""
+ |CREATE TABLE $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) USING hudi
+ |TBLPROPERTIES (
+ | type = '$tableType',
+ | primaryKey = 'id',
+ | orderingFields = 'ts'
+ |)
+ |LOCATION '$tablePath'
+ """.stripMargin)
+ // Two commits so the incremental range is never empty.
+ spark.sql(s"INSERT INTO $tableName VALUES (1, 'a1', 10, 1000), (2, 'a2',
20, 2000)")
+ spark.sql(s"INSERT INTO $tableName VALUES (3, 'a3', 30, 3000)")
+ }
+
+ private def firstCompletedInstantTime(tablePath: String): String = {
+ val metaClient = HoodieTableMetaClient.builder()
+ .setBasePath(tablePath)
+
.setConf(HadoopFSUtils.getStorageConf(spark.sparkContext.hadoopConfiguration)).build()
+ val instants: java.util.List[HoodieInstant] =
+ metaClient.getCommitsTimeline.filterCompletedInstants.getInstants
+ assertFalse(instants.isEmpty, s"Expected at least one completed instant at
$tablePath")
+ instants.get(0).requestedTime
Review Comment:
**minor:** `hoodie.datasource.read.begin.instanttime` is a completion time
on table version 8+ (`DataSourceOptions.scala:117-119`;
`MergeOnReadIncrementalRelationV2.scala:200,208`, `OPEN_CLOSED`), and
requested(c1) < completion(c1), so this "incremental" read spans the whole
table; the "two commits so the range is never empty" comment at L168 is not
load-bearing either (`EmptyRelation` only at zero commits,
`DefaultSource.scala:332`). Not blocking. Could this use the completion time so
the fixture reads what its name says?
```suggestion
instants.get(0).getCompletionTime
```
##########
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() {
Review Comment:
**nit:** procedure syntax is deprecated in Scala 2.13, which the spark4
profiles compile under; same at
`TestMergeIntoHoodieTableCommandInnerChildren.scala:46`. Feel free to ignore.
```suggestion
override def setUp(): Unit = {
```
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala:
##########
@@ -639,3 +643,70 @@ 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 {
+ plan transform {
+ // Type pattern + nested match avoids destructuring `LogicalRelation`,
whose
+ // case-class arity differs between Spark 3.x (4 args) and Spark 4.x
(5 args). This
+ // rule lives in `hudi-spark`, which is compiled against every
supported profile.
+ case lr: LogicalRelation if lr.catalogTable.isEmpty =>
+ lr.relation match {
+ case fsRelation: HadoopFsRelation =>
Review Comment:
**nit:** on release-1.1.x, which lacks #17457, incremental/CDC reads are not
`HadoopFsRelation`s, so this arm never fires and the rule is a silent no-op
there; 1.2.x has #17457 and is fine. Only matters if 1.1.x is ever a backport
target. Feel free to ignore.
##########
hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/command/MergeIntoHoodieTableCommand.scala:
##########
@@ -113,6 +113,23 @@ case class MergeIntoHoodieTableCommand(mergeInto:
MergeIntoTable) extends Hoodie
with ProvidesHoodieConfig
with PredicateHelper {
+ // Required so that `HoodieLeafLike#children = Nil` (kept for Catalyst
optimizer
+ // safety) does not also hide the source/target/merge-condition from
EXPLAIN, lineage
+ // extractors, and other plan walkers. `innerChildren` is not traversed by
`transform`
+ // or `mapChildren`.
+ //
+ // NOTE: this exposes the *analyzed* `MergeIntoTable`, carrying the
statement's assignments
+ // exactly as written -- deliberately not a post-alignment plan. Hudi does
expand partial
+ // `UPDATE SET` / `INSERT (...)` clauses to the full target schema, in
`alignAssignments`,
+ // but that expansion is serialized into the write config
(`PAYLOAD_*_CONDITION_AND_ASSIGNMENTS`)
+ // and never materialized as a `MergeIntoTable`. It also would not be a
faithful description
+ // of the write if it were: the filler assignments are bound against the
source-joined-target
+ // payload rather than the target alone, and under `ORIGINAL_VALUE` on a MOR
table an untouched
+ // column is encoded as `Literal(null)` -- a plan walker would read that as
"written as NULL"
+ // when it means "left alone". Consumers needing whole-row fidelity should
fill in the
+ // unassigned target columns themselves.
+ override def innerChildren: Seq[QueryPlan[_]] = Seq(mergeInto)
Review Comment:
**nit:** "consumed only by EXPLAIN and opt-in plan walkers" undersells it:
`SQLExecution` computes `physicalPlanDescription =
explainString(spark.sql.ui.explainMode)` (default `formatted`, which recurses
`innerChildren`) on every execution, so each MERGE now ships its full
source+target subtree into the SQL tab and event log, one per micro-batch for
foreachBatch MERGE -- the same as INSERT already does. Feel free to ignore, but
could the description say so in one line?
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala:
##########
@@ -639,3 +643,70 @@ 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] {
Review Comment:
**nit:** there is no opt-out, and no Hudi precedent either way (no rule in
`HoodieAnalysis` reads a config). Given the OL facet change and the SQL-UI
payload change noted elsewhere, would a `hoodie.*` boolean (default true) be
worth adding as an escape hatch? Feel free to ignore.
##########
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 testCatalogRegisteredReadIsNotMutated(): Unit = {
+ spark.sql(
+ s"""
+ |CREATE TABLE $tableName (
+ | id int,
+ | name string,
+ | ts long
+ |) USING hudi
+ |TBLPROPERTIES (
+ | type = 'cow',
+ | primaryKey = 'id',
+ | orderingFields = 'ts'
+ |)
+ |LOCATION '$basePath/$tableName'
+ """.stripMargin)
+ spark.sql(s"INSERT INTO $tableName VALUES (1, 'a1', 1000)")
+
+ val analyzed = spark.sql(s"SELECT * FROM
$tableName").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
+
+ assertTrue(lr.catalogTable.isDefined, "Expected catalogTable from catalog
registration")
+ assertEquals(tableName, lr.catalogTable.get.identifier.table)
+ }
+
+ @Test
+ def testSnapshotPathBasedReadIsNotEnriched(): Unit = {
+ val tablePath = s"$basePath/$tableName"
+ createAndPopulateTable("cow", tablePath)
+
+ val df = spark.read.format("hudi").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
+
+ val location = lr.relation.asInstanceOf[HadoopFsRelation].location
+ assertFalse(location.isInstanceOf[HoodieIncrementalFileIndex],
+ s"Snapshot read should not produce a HoodieIncrementalFileIndex, got:
${location.getClass.getName}")
+ assertTrue(location.isInstanceOf[HoodieFileIndex],
+ s"Snapshot read should produce a HoodieFileIndex, got:
${location.getClass.getName}")
+ assertFalse(lr.catalogTable.isDefined,
+ "Snapshot path-based read must not be enriched by
HoodieIncrementalRelationIdentifier")
+ }
+
+ // --- helpers ---
+
+ private def createAndPopulateTable(tableType: String, tablePath: String):
Unit = {
+ spark.sql(
+ s"""
+ |CREATE TABLE $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) USING hudi
+ |TBLPROPERTIES (
+ | type = '$tableType',
+ | primaryKey = 'id',
+ | orderingFields = 'ts'
+ |)
+ |LOCATION '$tablePath'
+ """.stripMargin)
+ // Two commits so the incremental range is never empty.
+ spark.sql(s"INSERT INTO $tableName VALUES (1, 'a1', 10, 1000), (2, 'a2',
20, 2000)")
+ spark.sql(s"INSERT INTO $tableName VALUES (3, 'a3', 30, 3000)")
Review Comment:
**minor:** every fixture is a SQL-created, catalog-registered table read
back by path, so name == dir == catalog name (the `identifier.table` assertion
cannot tell where the name came from) and `hoodie.database.name` is always
`default`, so the `dbName = None` branch of `buildCatalogTable` is never
reached. The PR's actual scenario -- a
`df.write.format("hudi").option(TBL_NAME, ...).save(path)` table never
registered -- exercises both. Not blocking. Could one arm write the table
through the DataFrame writer with a name that differs from the directory?
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala:
##########
@@ -639,3 +643,70 @@ 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 {
+ plan transform {
+ // Type pattern + nested match avoids destructuring `LogicalRelation`,
whose
+ // case-class arity differs between Spark 3.x (4 args) and Spark 4.x
(5 args). This
+ // rule lives in `hudi-spark`, which is compiled against every
supported profile.
+ case lr: LogicalRelation if lr.catalogTable.isEmpty =>
+ lr.relation match {
+ case fsRelation: HadoopFsRelation =>
+ incrementalOrCDCIndex(fsRelation.location)
+ .flatMap(index => buildCatalogTable(index.metaClient,
lr.schema))
+ .map(catalogTable => lr.copy(catalogTable =
Some(catalogTable)))
+ .getOrElse(lr)
+ case _ => lr
+ }
+ }
+ }
+
+ /**
+ * Narrows a [[FileIndex]] to the incremental/CDC Hudi indexes this rule
handles, or
+ * `None` for anything else. Both are [[HoodieFileIndex]] subclasses, so the
widening is
+ * checked by the compiler rather than by an `asInstanceOf` that a future
third index
+ * type could silently break.
+ */
+ private def incrementalOrCDCIndex(location: FileIndex):
Option[HoodieFileIndex] =
+ location match {
+ case index: HoodieIncrementalFileIndex => Some(index)
+ case index: HoodieCDCFileIndex => Some(index)
+ case _ => None
+ }
+
+ private def buildCatalogTable(
+ metaClient: HoodieTableMetaClient,
+ schema: StructType): Option[CatalogTable] = {
+ val tableConfig = metaClient.getTableConfig
+ // `hoodie.table.name` is required for a valid Hudi table, but if it is
somehow unset
+ // leave `catalogTable` as `None` -- the pre-existing behavior -- rather
than synthesize
+ // a `TableIdentifier(null)` that would surface downstream as a garbage
dataset name.
+ Option(tableConfig.getTableName).filter(_.nonEmpty).map { tableName =>
+ // Falls back to Spark's `default` database when `hoodie.database.name`
is unset --
+ // matches existing path-based DataFrame read behavior.
+ val dbName = Option(tableConfig.getDatabaseName).filter(_.nonEmpty)
+ CatalogTable(
+ identifier = TableIdentifier(tableName, dbName),
+ tableType = CatalogTableType.EXTERNAL,
+ storage = CatalogStorageFormat.empty.copy(
+ locationUri = Some(metaClient.getBasePath.toUri)),
+ schema = schema,
+ provider = Some("hudi")
Review Comment:
Precedent worth having on this thread: the identifier form of
`hudi_table_changes` has attached a Hudi-provider `CatalogTable` to
`HoodieIncrementalFileIndex` / `HoodieCDCFileIndex` relations since #8729
(`243098c39aff`, Jun 2023), under test in `TestHoodieTableValuedFunction`
(`isTableId`), with no follow-up bug in that time. So the shape itself is not
new; the negative test would only pin the `View` block.
--
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]
