ryux1 commented on code in PR #19849:
URL: https://github.com/apache/hudi/pull/19849#discussion_r3945240667
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala:
##########
@@ -75,16 +74,6 @@ trait HoodieFileSplit {}
case class HoodieTableSchema(structTypeSchema: StructType, schema:
HoodieSchema, internalSchema: Option[InternalSchema] = None)
-case class HoodieTableState(tablePath: String,
- latestCommitTimestamp: Option[String],
- recordKeyField: String,
- orderingFields: List[String],
- usesVirtualKeys: Boolean,
- metadataConfig: HoodieMetadataConfig,
- recordMergeImplClasses: List[String],
- recordMergeStrategyId: String)
Review Comment:
Updated the Impact and Risk sections to state explicitly that removing this
publicly scoped type is a source- and binary-compatibility change for
out-of-tree consumers. I chose the documented-breakage option rather than
retaining a deprecated shell; the changelog now calls out the removal.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala:
##########
@@ -252,19 +241,7 @@ abstract class HoodieBaseRelation(val sqlContext:
SQLContext,
HoodieFileIndex(sparkSession, metaClient, Some(tableStructSchema),
optParams,
FileStatusCache.getOrCreate(sparkSession), shouldIncludeLogFiles())
- lazy val tableState: HoodieTableState = {
- val recordMergerImpls =
optParams.get(HoodieWriteConfig.RECORD_MERGE_IMPL_CLASSES.key()).map(impls =>
ConfigUtils.split2List(impls).asScala.toList).getOrElse(List.empty)
- // Subset of the state of table's configuration as of at the time of the
query
- HoodieTableState(tablePath = basePath.toString,
- latestCommitTimestamp = queryTimestamp,
- recordKeyField = recordKeyField,
- orderingFields = orderingFields,
- usesVirtualKeys = !tableConfig.populateMetaFields(),
- metadataConfig = fileIndex.getMetadataConfig,
- recordMergeImplClasses = recordMergerImpls,
- recordMergeStrategyId = tableConfig.getRecordMergeStrategyId
- )
- }
+ protected lazy val latestCommitTimestamp: Option[String] = queryTimestamp
Review Comment:
Both fields remain live. `recordKeyField` and `orderingFields` feed
`mandatoryFieldsForMerging` in `MergeOnReadSnapshotRelation`, and
`orderingFields` also feeds `mandatoryFields` in both incremental relation
variants, which is then consumed by projection and schema pruning. I am leaving
them in place.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala:
##########
@@ -252,19 +241,7 @@ abstract class HoodieBaseRelation(val sqlContext:
SQLContext,
HoodieFileIndex(sparkSession, metaClient, Some(tableStructSchema),
optParams,
FileStatusCache.getOrCreate(sparkSession), shouldIncludeLogFiles())
- lazy val tableState: HoodieTableState = {
- val recordMergerImpls =
optParams.get(HoodieWriteConfig.RECORD_MERGE_IMPL_CLASSES.key()).map(impls =>
ConfigUtils.split2List(impls).asScala.toList).getOrElse(List.empty)
- // Subset of the state of table's configuration as of at the time of the
query
- HoodieTableState(tablePath = basePath.toString,
- latestCommitTimestamp = queryTimestamp,
- recordKeyField = recordKeyField,
- orderingFields = orderingFields,
- usesVirtualKeys = !tableConfig.populateMetaFields(),
- metadataConfig = fileIndex.getMetadataConfig,
- recordMergeImplClasses = recordMergerImpls,
- recordMergeStrategyId = tableConfig.getRecordMergeStrategyId
- )
- }
+ protected lazy val latestCommitTimestamp: Option[String] = queryTimestamp
Review Comment:
Kept the frozen lazy value to preserve the behavior of `master`: the old
lazy `tableState` captured `queryTimestamp` once, while widening the `def`
would recompute it on later `composeRDD` calls. Commit `32e71c0` renames the
captured value to `targetInstantTime` so the distinction is explicit without
changing the timing.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala:
##########
@@ -252,19 +241,7 @@ abstract class HoodieBaseRelation(val sqlContext:
SQLContext,
HoodieFileIndex(sparkSession, metaClient, Some(tableStructSchema),
optParams,
FileStatusCache.getOrCreate(sparkSession), shouldIncludeLogFiles())
- lazy val tableState: HoodieTableState = {
- val recordMergerImpls =
optParams.get(HoodieWriteConfig.RECORD_MERGE_IMPL_CLASSES.key()).map(impls =>
ConfigUtils.split2List(impls).asScala.toList).getOrElse(List.empty)
- // Subset of the state of table's configuration as of at the time of the
query
- HoodieTableState(tablePath = basePath.toString,
- latestCommitTimestamp = queryTimestamp,
- recordKeyField = recordKeyField,
- orderingFields = orderingFields,
- usesVirtualKeys = !tableConfig.populateMetaFields(),
- metadataConfig = fileIndex.getMetadataConfig,
Review Comment:
Confirmed that the reflective test is the only remaining in-repo caller. I
am leaving the getter unchanged here because reducing its visibility would
introduce another public API break unrelated to the RDD serialization change.
That cleanup is better isolated in a follow-up.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala:
##########
@@ -89,7 +89,7 @@ private[hudi] case class
HoodieMergeOnReadBaseFileReaders(fullSchemaReader: Base
* @param fileReaders suite of base file readers
* @param tableSchema table's full schema
* @param requiredSchema expected (potentially) projected schema
- * @param tableState table's state
+ * @param latestCommitTimestamp latest completed commit timestamp for the query
Review Comment:
Fixed in `32e71c0`: the parameter is now `targetInstantTime`, documented as
the as-of instant or the last instant in the query timeline.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala:
##########
@@ -89,7 +89,7 @@ private[hudi] case class
HoodieMergeOnReadBaseFileReaders(fullSchemaReader: Base
* @param fileReaders suite of base file readers
* @param tableSchema table's full schema
* @param requiredSchema expected (potentially) projected schema
- * @param tableState table's state
+ * @param latestCommitTimestamp latest completed commit timestamp for the query
Review Comment:
Fixed in `32e71c0`: the descriptions are aligned and `sqlConf`,
`optionalFilters`, `metaClient`, and `options` are now documented.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala:
##########
@@ -100,7 +100,7 @@ class HoodieMergeOnReadRDDV2(@transient sc: SparkContext,
fileReaders: HoodieMergeOnReadBaseFileReaders,
tableSchema: HoodieTableSchema,
requiredSchema: HoodieTableSchema,
- tableState: HoodieTableState,
+ latestCommitTimestamp: Option[String],
Review Comment:
Keeping `Option[String]` preserves the existing empty-table path:
`queryTimestamp` can be `None`, while `listLatestFileSlices` returns no splits.
Resolving a mandatory `String` at relation construction would turn that case
into an error. The renamed parameter documents what the optional value
represents.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV2.scala:
##########
@@ -271,4 +271,3 @@ trait HoodieIncrementalRelationV2Trait extends
HoodieBaseRelation {
optParams.getOrElse(DataSourceReadOptions.INCR_PATH_GLOB.key,
DataSourceReadOptions.INCR_PATH_GLOB.defaultValue)
}
-
Review Comment:
Restored the trailing blank line in `32e71c0`.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala:
##########
@@ -252,19 +241,7 @@ abstract class HoodieBaseRelation(val sqlContext:
SQLContext,
HoodieFileIndex(sparkSession, metaClient, Some(tableStructSchema),
optParams,
FileStatusCache.getOrCreate(sparkSession), shouldIncludeLogFiles())
- lazy val tableState: HoodieTableState = {
- val recordMergerImpls =
optParams.get(HoodieWriteConfig.RECORD_MERGE_IMPL_CLASSES.key()).map(impls =>
ConfigUtils.split2List(impls).asScala.toList).getOrElse(List.empty)
- // Subset of the state of table's configuration as of at the time of the
query
- HoodieTableState(tablePath = basePath.toString,
- latestCommitTimestamp = queryTimestamp,
- recordKeyField = recordKeyField,
- orderingFields = orderingFields,
- usesVirtualKeys = !tableConfig.populateMetaFields(),
- metadataConfig = fileIndex.getMetadataConfig,
Review Comment:
Good catch. The Impact section now calls out that removing `tableState`
stops forcing `fileIndex` during RDD construction and may defer or avoid the
listing when catalog statistics satisfy planning; paths that consult
`sizeInBytes` still force it. I did not add a test because whether Spark
requests relation statistics is planner/catalog behavior rather than a stable
invariant of this refactor.
--
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]
