voonhous commented on code in PR #18726:
URL: https://github.com/apache/hudi/pull/18726#discussion_r3921176476
##########
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))
Review Comment:
**minor:** `lr.schema` is the plan output, not the table. For CDC it is the
`(op, ts_ms, before, after)` envelope
(`HoodieHadoopFsRelationFactory.scala:353,468`), while
`hudi_table_changes('db.tbl', 'cdc', ...)` attaches the real table schema to
the same relation (`HoodieSparkBaseAnalysis.scala:104-110`), so one table is
published under two schemas. And if the first analysis references `_metadata`,
`AddMetadataColumns` has already appended it to `lr.output`
(`LogicalRelation.withMetadataColumns`). Not blocking. Could we pass
`fsRelation.schema` here, and for CDC either resolve the table schema or say in
the scaladoc that the envelope is intentional?
##########
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:
**minor:** this line makes the stamped relation satisfy
`ResolvesToHudiTable`: `resolveHoodieTable` keys on `provider == "hudi"`
(`SparkAdapter.scala:151-153`, `BaseSpark3Adapter.scala:95`) and gates
MERGE/UPDATE/DELETE/index-DDL targets, not only INSERT as the description says.
It is unreachable today only because a DataFrame temp view resolves to `View`
and `PhysicalOperation` stops there. Not blocking. Could we add a negative test
(`createOrReplaceTempView` over the incremental read, then `UPDATE v SET ...`
still fails) so a future `View` change trips a test rather than routing a read
into `UpdateHoodieTableCommand`?
##########
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")
Review Comment:
**minor:** these two pass for any non-empty relation, and
`identifier.database` is never asserted although the rule has a dedicated
branch for it; SQL `CREATE TABLE` stamps `hoodie.database.name`
(`HoodieCatalogTable.scala:217-218,234`), so the expected value here is
`Some("default")`. Not blocking. Could these become equality checks against the
base path URI and `lr.schema`, plus `assertEquals(Some("default"),
ct.identifier.database)` and `assertEquals(CatalogTableType.EXTERNAL,
ct.tableType)`?
##########
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`.
Review Comment:
**minor:** the history behind "kept for Catalyst optimizer safety" is worth
naming: MERGE was promoted to `DataWritingCommand` with real children in #13239
(`b689a711ef9a`) and reverted in #13981 (`20f1585b1291`, `Couldn't find
operation#8969 in [...]`). The description's #12772 is the Spark 4 support PR;
the sibling moves were #13044/#13110/#13176, and the leaf-plus-`innerChildren`
precedent is #4894. Could we cite #13981 here (same comment in the
spark4-common copy) so nobody re-promotes MERGE, and fix the description?
```suggestion
// Required so that `HoodieLeafLike#children = Nil` (kept for Catalyst
optimizer
// safety: MERGE was promoted to `DataWritingCommand` in #13239 and
reverted in #13981
// after runtime attribute-binding failures) 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`.
```
##########
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.
Review Comment:
**nit:** the comment says it falls back to `default`, but `dbName = None`
leaves the identifier unqualified (`TableIdentifier.unquotedString` prints the
bare table name); nothing substitutes `default`. Feel free to ignore, but could
we reword so consumers do not expect `default.<tbl>`?
```suggestion
// Leaves the identifier unqualified when `hoodie.database.name` is
unset (no `default`
// substitution here) -- matches existing path-based DataFrame read
behavior.
```
##########
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(
Review Comment:
**nit:** `partitionColumnNames` stays empty, so a partitioned table reads as
unpartitioned. That is the right call -- `CatalogTable.partitionSchema` asserts
the partition columns are the trailing fields of `schema`
(`catalog/interface.scala:262-264`), which a relation schema does not guarantee
-- but a one-liner would stop someone "fixing" it later. Feel free to ignore.
```suggestion
// `partitionColumnNames` is intentionally empty:
`CatalogTable.partitionSchema` asserts the
// partition columns are the trailing fields of `schema`, which a
relation schema does not
// guarantee, so populating it naively would throw on first access.
CatalogTable(
```
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/analysis/TestMergeIntoHoodieTableCommandInnerChildren.scala:
##########
@@ -0,0 +1,225 @@
+/*
+ * 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.HoodieConversionUtils.toJavaOption
+import org.apache.hudi.ScalaAssertionSupport
+import org.apache.hudi.testutils.HoodieClientTestBase
+import org.apache.hudi.util.JFunction
+
+import org.apache.spark.sql.{SparkSession, SparkSessionExtensions}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.logical.{MergeIntoTable,
UpdateAction}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.hudi.HoodieSparkSessionExtension
+import org.apache.spark.sql.hudi.command.MergeIntoHoodieTableCommand
+import org.junit.jupiter.api.{BeforeEach, Test}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+
+import java.util.function.Consumer
+
+/**
+ * Guards both the leaf-ness of [[MergeIntoHoodieTableCommand]] (Catalyst
optimizer
+ * safety) and the new `innerChildren` exposure (lineage / EXPLAIN
reachability).
+ */
+class TestMergeIntoHoodieTableCommandInnerChildren extends
HoodieClientTestBase with ScalaAssertionSupport {
Review Comment:
**nit:** each of the 7 new tests boots a fresh `SparkContext`
(`initSparkContexts()` per `@BeforeEach`, ~30s in CI), while every MERGE SQL
suite uses the shared-session `HoodieSparkSqlTestBase` (68 suites vs 10 on
`HoodieClientTestBase`); the two plan-shape tests differ only in the merge SQL,
and `ScalaAssertionSupport` is unused. `TestHoodiePruneFileSourcePartitions` in
this directory uses the same base, so feel free to ignore -- but would
`dml/others/TestMergeIntoTable.scala` be a cheaper home for these?
##########
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:** relative to #18298, only the MERGE visitor gains a node:
`UpdateHoodieTableCommand.innerChildren` exposes the input `query`, not the
`UpdateTable` node, and `DeleteHoodieTableCommand` does not retain
`DeleteFromTable` at all, so OpenLineage-style `case u: UpdateTable` / `case d:
DeleteFromTable` visitors still find nothing. Feel free to ignore -- could the
description say so in one sentence, so readers of the issue do not assume all
three now work?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/analysis/TestMergeIntoHoodieTableCommandInnerChildren.scala:
##########
@@ -0,0 +1,225 @@
+/*
+ * 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.HoodieConversionUtils.toJavaOption
+import org.apache.hudi.ScalaAssertionSupport
+import org.apache.hudi.testutils.HoodieClientTestBase
+import org.apache.hudi.util.JFunction
+
+import org.apache.spark.sql.{SparkSession, SparkSessionExtensions}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.logical.{MergeIntoTable,
UpdateAction}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.hudi.HoodieSparkSessionExtension
+import org.apache.spark.sql.hudi.command.MergeIntoHoodieTableCommand
+import org.junit.jupiter.api.{BeforeEach, Test}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+
+import java.util.function.Consumer
+
+/**
+ * Guards both the leaf-ness of [[MergeIntoHoodieTableCommand]] (Catalyst
optimizer
+ * safety) and the new `innerChildren` exposure (lineage / EXPLAIN
reachability).
+ */
+class TestMergeIntoHoodieTableCommandInnerChildren extends
HoodieClientTestBase with ScalaAssertionSupport {
+
+ private var spark: SparkSession = _
+ private var sourceTableName: String = _
+
+ @BeforeEach
+ override def setUp() {
+ setTableName("hoodie_merge_target")
+ sourceTableName = "hoodie_merge_source"
+ 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)))
+ )
+
+ @Test
+ def testMergeIntoExposesAnalyzedMergeIntoTableViaInnerChildren(): Unit = {
+ spark.sql(
+ s"""
+ |CREATE TABLE $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) USING hudi
+ |TBLPROPERTIES (
+ | type = 'cow',
+ | primaryKey = 'id',
+ | orderingFields = 'ts'
+ |)
+ |LOCATION '$basePath/$tableName'
+ """.stripMargin)
+
+ spark.sql(s"INSERT INTO $tableName VALUES (1, 'a1', 10.0, 1000)")
+
+ spark.sql(s"CREATE OR REPLACE TEMPORARY VIEW $sourceTableName AS " +
+ s"SELECT 1 AS id, 'a1_updated' AS name, 99.0 AS price, 2000L AS ts " +
+ s"UNION ALL " +
+ s"SELECT 2, 'a2', 20.0, 2000L")
+
+ val mergeSql =
+ s"""
+ |MERGE INTO $tableName AS t
+ |USING $sourceTableName AS s
+ |ON t.id = s.id
+ |WHEN MATCHED THEN UPDATE SET t.name = s.name, t.price = s.price,
t.ts = s.ts
+ |WHEN NOT MATCHED THEN INSERT (id, name, price, ts) VALUES (s.id,
s.name, s.price, s.ts)
+ """.stripMargin
+
+ val analyzed = spark.sql(mergeSql).queryExecution.analyzed
+
+ val cmdOpt = analyzed.collectFirst { case c: MergeIntoHoodieTableCommand
=> c }
+ assertTrue(cmdOpt.isDefined,
+ s"Expected MergeIntoHoodieTableCommand in analyzed plan,
got:\n$analyzed")
+ val cmd = cmdOpt.get
+
+ assertEquals(0, cmd.children.size,
+ s"MergeIntoHoodieTableCommand must remain a Catalyst leaf, got children:
${cmd.children}")
+
+ assertEquals(1, cmd.innerChildren.size,
+ s"Expected innerChildren to expose exactly one node (MergeIntoTable),
got: ${cmd.innerChildren}")
+
+ val inner = cmd.innerChildren.head
+ assertTrue(inner.isInstanceOf[MergeIntoTable],
+ s"Expected innerChildren(0) to be MergeIntoTable, got:
${inner.getClass.getName}")
+
+ val mergeIntoTable = inner.asInstanceOf[MergeIntoTable]
+
+ val sourceLeaves = mergeIntoTable.sourceTable.collectLeaves()
+ assertTrue(sourceLeaves.nonEmpty,
+ s"Expected at least one leaf in source plan, got:
${mergeIntoTable.sourceTable}")
+
+ val targetLeaves = mergeIntoTable.targetTable.collectLeaves()
+ assertTrue(targetLeaves.exists(_.isInstanceOf[LogicalRelation]),
+ s"Expected target plan to contain a LogicalRelation, got:
${mergeIntoTable.targetTable}")
+
+ assertTrue(mergeIntoTable.mergeCondition != null, "Expected non-null
mergeCondition")
Review Comment:
**minor:** `collectLeaves().nonEmpty` and `mergeCondition != null` cannot be
false for a resolved `MergeIntoTable` (`mergeCondition` is a required
constructor arg the parser fills from `ON`), so neither guards anything. Not
blocking. Could the first become a check that the source leaf is the
`hoodie_merge_source` view, and the second
`assertTrue(mergeIntoTable.mergeCondition.resolved)` or an `EqualTo` shape
check?
##########
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)
Review Comment:
**minor:** this test cannot fail: `SELECT * FROM t` is a snapshot read, so
`incrementalOrCDCIndex` never matches and the `catalogTable.isEmpty` guard is
never reached; and `identifier.table == tableName` would also hold for a
synthesized replacement, since the rule builds it from `hoodie.table.name`. Not
blocking. Could it read through the identifier form of `hudi_table_changes`
instead (the one shape where the guard is load-bearing,
`HoodieSparkBaseAnalysis.scala:104-110`) and assert something only the catalog
copy carries, e.g. `ct.properties.nonEmpty`?
##########
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)
Review Comment:
**major:** the CDC arm has no test: `grep -rn HoodieCDCFileIndex
hudi-spark-datasource/*/src/test` is empty and the new suite never sets
`hoodie.datasource.query.incremental.format=cdc`, so this branch is
unexercised. Could we add one `catalogTable` assertion to the existing "Test
hudi_table_changes cdc" case in `TestHoodieTableValuedFunction.scala:262`? It
already loops `(cow|mor) x (tableId|path)`, so a single assertion pins CDC, the
`hudi_table_changes('<path>', ...)` form this rule now stamps, and the
identifier form the `isEmpty` guard protects.
##########
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
Review Comment:
**nit:** "exactly as written" is slightly off: `ResolveImplementations`
stores `ReplaceExpressions(mit)` (`HoodieAnalysis.scala:486`), so
`RuntimeReplaceable` expressions such as `nvl` are already rewritten to
`coalesce` in what `innerChildren` exposes. Feel free to ignore.
```suggestion
// NOTE: this exposes the *analyzed* `MergeIntoTable` (after
`ReplaceExpressions`, so
// `RuntimeReplaceable` nodes such as `nvl` are already rewritten),
carrying the statement's
// assignments as written -- deliberately not a post-alignment plan. Hudi
does expand partial
```
##########
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)))
Review Comment:
**minor:** on the interaction flagged in the description: with
`catalogTable` set, `Spark3/4HoodiePruneFileSourcePartitions` now attaches
`CatalogStatistics`, but `FilterEstimation.estimate` returns `None` when the
child has no `rowCount` (`FilterEstimation.scala:45-46`), so
`rowCount`/`colStats` stay empty and `toPlanStats` yields
`Statistics(sizeInBytes)`. The only delta vs `HadoopFsRelation.sizeInBytes` is
that `spark.sql.sources.fileCompressionFactor` (default 1.0) no longer applies;
the scan node also prints `Scan hudi default.<tbl>` now. Not blocking. Could we
pin this with one incremental case in `TestHoodiePruneFileSourcePartitions`,
which already asserts `lr.stats.sizeInBytes`?
--
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]
