voonhous commented on code in PR #19849:
URL: https://github.com/apache/hudi/pull/19849#discussion_r3944193962
##########
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:
`HoodieTableState` is plain `public` and ships in hudi-spark-bundle, so any
out-of-repo relation subclass, custom datasource or tool that constructs one or
reads `relation.tableState` breaks at compile time against the new source, and
with `NoClassDefFoundError`/`NoSuchMethodError` against the new bundle. The PR
body says "No public API or behavior change".
Its file-neighbour at `HoodieMergeOnReadRDDV2.scala:80` is `private[hudi]
case class HoodieMergeOnReadBaseFileReaders`, so the codebase does scope
internal case classes deliberately; this one never was.
Could we either keep the case class for one release marked deprecated, or
drop the "no public API change" claim from the description and flag the removal
in the release notes?
##########
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:
The value passed here is
`specifiedQueryTimestamp.orElse(timeline.lastInstant())`, which is not the
latest completed commit at two of the three call sites:
- under time travel (`as.of.instant` /
`hoodie.datasource.read.begin.instanttime`) it is the user's as-of instant;
- `MergeOnReadIncrementalRelationV1` overrides `timeline` to
`findInstantsInRange(startTimestamp, endTimestamp)` plus pending compaction,
and V2 to `queryContext.getActiveTimeline`, so it is the last instant of the
queried window.
Could we word it as the query's target instant -- the as-of instant, or the
last instant of the queried timeline -- instead? As written, a reader will
assume the table's newest commit is what reaches
`HoodieFileGroupReader.withLatestCommitTime` and mis-reason about merge and
log-block visibility.
##########
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:
Separately: dropping this call leaves
`BaseHoodieTableFileIndex#getMetadataConfig` with no in-repo production caller.
The only remaining reference is
`BaseHoodieTableFileIndexTest#testGetMetadataConfigReturnsFieldValue`, which
reflectively sets the field and then asserts the getter hands it back.
Its neighbours in that class are already scoped --
`@Getter(AccessLevel.PROTECTED)` on `partitionColumns` and `queryPaths`. Could
we drop `metadataConfig`'s `@Getter` to `AccessLevel.PROTECTED` and retire that
test with it?
##########
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:
This call was the only thing forcing the `lazy val fileIndex` (line 240),
whose own scaladoc warns it "initializes eagerly listing all of the files w/in
the given Hudi table".
For `MergeOnReadIncrementalRelationV1`/`V2` on the non-`fullTableScan`
branch, `collectFileSplits` builds its own `HoodieTableFileSystemView` from
`affectedFilesInCommits` and never touches `fileIndex` -- the only `fileIndex`
hits in those two files are V1:133 and V2:127, inside `listFileSplits`, a
different entry point. So an incremental MOR query with a start/end range used
to perform a full table listing at `composeRDD` time and now does not.
That reads like a win, but it is a real behavior delta against the
description, and it changes when the listing and any listing-time failure
surface. Could we call it out in the PR body and add an incremental-MOR test
that pins the new behavior?
##########
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:
Nit: every other entry aligns its description at a fixed column and this one
uses a single space, so the block no longer reads as a table.
While we are in here, the block is also missing `sqlConf`,
`optionalFilters`, `metaClient` and `options`, all present in the constructor
before this PR. Could we add them?
##########
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:
All three uses inside `compute()` are `latestCommitTimestamp.orNull` (lines
211, 229, 242), so the `Some`/`None` distinction is never read. Both builders
then do `ValidationUtils.checkArgument(latestCommitTime != null, "Latest commit
time is required")` (`HoodieFileGroupReader.java:121`,
`HoodieLsmFileGroupReader.java:108`), so a `None` fails per-task on the
executor rather than on the driver.
Since the PR is deliberately picking this parameter's type, could it be
`String`, resolved once on the driver? That removes the wrapper and turns an
executor-side `IllegalArgumentException` into a fail-fast at relation
construction.
##########
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:
This has nothing to do with collapsing `HoodieTableState` -- could we
restore the trailing blank line? It widens the conflict surface for anyone
cherry-picking onto release-1.2.x, and `git diff --check`, cited in the PR body
as the verification, does not catch it.
##########
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:
This is a second name for something the same class already exposes.
`queryTimestamp` (line 260) is a `def` recomputed from the overridable
`timeline` on every call, and `listLatestFileSlices` (line 345) uses it that
way; as a `lazy val`, `latestCommitTimestamp` freezes the first value. The two
disagree whenever the metaClient timeline is reloaded between
`collectFileSplits` and `composeRDD` on a reused relation.
It also lands on `HoodieBaseRelation`, so `BaseFileOnlyRelation`,
`HoodieBootstrapRelation` and `IncrementalRelation` all inherit a member only
the three MOR relations use, which widens the base class in a PR aimed at
narrowing surface.
Could we widen `private def queryTimestamp` to `protected def
queryTimestamp` and pass `queryTimestamp` at the three call sites instead? That
adds no member at all and keeps one accessor for the concept.
--
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]
