fix(spark): fetch schema from HMS when it is not found in FileSystem - #19536
fix(spark): fetch schema from HMS when it is not found in FileSystem#19536nada-attia wants to merge 2 commits into
Conversation
In HoodieBaseRelation, schema is assumed to always be resolvable from the table's commit metadata or data files. For datasets that contain no hudi partitions (only non-hudi partitions), or that have no data written yet, schema cannot be resolved that way. Fall back to fetching the schema from HMS in that case instead of throwing. Adds a unit test covering select on a freshly created, empty table.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! The PR makes HoodieBaseRelation fall back to the HMS catalog schema when TableSchemaResolver can't resolve a schema from commit metadata or data files, so reads on empty (or non-Hudi-partition-only) tables no longer throw. A couple of edge cases around the breadth of the fallback and path-based / HMS-absent tables are worth double-checking in the inline comments. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. One minor redundancy below; overall the change is clean and the test comment is genuinely useful.
| Try(schemaResolver.getTableSchema) match { | ||
| case Success(schema) => schema | ||
| case Failure(e) => throw e | ||
| case Failure(_) => |
There was a problem hiding this comment.
🤖 This falls back to HMS on any getTableSchema failure, not just the intended empty/no-data case. If the table actually has data but resolution fails for another reason (transient IO, corruption, or a schema evolved after create), we'd silently read with the possibly-stale HMS create-schema instead of failing loudly. Could we gate the fallback on the table genuinely having no resolvable schema (e.g. no completed commits / no base files), and chain the original exception into anything we rethrow so the root cause isn't lost?
| case Failure(_) => | ||
| // Schema not found on the dataset, so fetching schema from HMS. | ||
| logWarning(s"Schema not found on the dataset for $tableName, so fetching schema from HMS.") | ||
| val catalogTable = sparkSession.sessionState.catalog.externalCatalog.getTable(metaClient.getDbName, metaClient.getTableName) |
There was a problem hiding this comment.
🤖 For path-based reads (or tables not registered in the current session catalog), metaClient.getDbName / getTableName may not correspond to an HMS entry, and externalCatalog.getTable(...) will throw NoSuchDatabase/NoSuchTable, replacing the original schema error with a more confusing one. Is the fallback guaranteed to run only when the table is HMS-registered? It may be worth catching the not-found case and rethrowing the original resolver failure.
| // Schema not found on the dataset, so fetching schema from HMS. | ||
| logWarning(s"Schema not found on the dataset for $tableName, so fetching schema from HMS.") | ||
| val catalogTable = sparkSession.sessionState.catalog.externalCatalog.getTable(metaClient.getDbName, metaClient.getTableName) | ||
| convertToHoodieSchema(catalogTable.schema, tableName) |
There was a problem hiding this comment.
🤖 The normal getTableSchema() path includes the hoodie* meta fields (when populateMetaFields is true). Spark-SQL-created tables do persist meta fields in the HMS schema, so this works for the added test, but tables registered via Hive sync / DataSource writer may store the catalog schema without meta fields. Have you checked that catalogTable.schema here always includes the meta fields, so tableStructSchema stays consistent with the non-fallback path?
| case Failure(e) => throw e | ||
| case Failure(_) => | ||
| // Schema not found on the dataset, so fetching schema from HMS. | ||
| logWarning(s"Schema not found on the dataset for $tableName, so fetching schema from HMS.") |
There was a problem hiding this comment.
🤖 nit: the inline comment repeats the log message verbatim — could you drop one of them? The logWarning already serves as self-documentation here.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #19536 +/- ##
============================================
+ Coverage 77.17% 77.50% +0.33%
+ Complexity 33948 33102 -846
============================================
Files 2575 2522 -53
Lines 143397 139187 -4210
Branches 17579 16896 -683
============================================
- Hits 110662 107878 -2784
+ Misses 24476 23742 -734
+ Partials 8259 7567 -692
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
| case Failure(_) => | ||
| // Schema not found on the dataset, so fetching schema from HMS. | ||
| logWarning(s"Schema not found on the dataset for $tableName, so fetching schema from HMS.") | ||
| val catalogTable = sparkSession.sessionState.catalog.externalCatalog.getTable(metaClient.getDbName, metaClient.getTableName) |
There was a problem hiding this comment.
if the table does not exist in the catalog, we should throw appropriate msg.
There was a problem hiding this comment.
yes sure updated the PR
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for addressing the feedback! The updated diff now guards the HMS fallback with externalCatalog.tableExists(...) and throws a HoodieException that chains the original resolver exception when the table isn't in the catalog — this resolves the path-based/HMS-absent concern raised by @nsivabalan and in the prior review. One prior concern remains partially open: when the table does exist in the catalog, the fallback still triggers on any resolution failure (not just the empty/no-data case), which could silently substitute a stale HMS schema — flagged inline. The redundant inline comment (nit) and the Hive-synced meta-fields question from the last pass are also still open. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here.
| case Failure(e) => | ||
| // Schema not found on the dataset, so fetching schema from HMS. | ||
| logWarning(s"Schema not found on the dataset for $tableName, so fetching schema from HMS.") | ||
| val databaseName = metaClient.getTableConfig.getDatabaseName |
There was a problem hiding this comment.
🤖 Nice improvement chaining the original exception on the not-found path. One thing still open from last round: when the table does exist in the catalog, we fall back to the HMS schema on any getTableSchema failure — not just the empty/no-data case. If a table with real data fails resolution transiently (IO) or after a write-path schema evolution, we'd silently read under the (possibly stale) HMS create-schema instead of surfacing e. Could we gate the fallback on the table genuinely having no resolvable schema (e.g. no completed commits / no base files)?
6 participants
Describe the issue this Pull Request addresses
In HoodieBaseRelation, schema is assumed to always be resolvable from the table's commit metadata or data files. For datasets that contain no hudi partitions (only non-hudi partitions), or that have no data written yet, schema cannot be
resolved that way. Fall back to fetching the schema from HMS in that case instead of throwing.
Adds a unit test covering select on a freshly created, empty table.
Summary and Changelog
Reads on a Hudi table would throw when
TableSchemaResolvercould not resolve a schema from commit metadata or data files on the file system — e.g. a table with no hudi partitions (only non-hudi partitions), or one with no data written yet.HoodieBaseRelationnow catches that failure and falls back to reading the table's schema from the Hive Metastore (HMS) catalog entry instead of throwing.HoodieBaseRelation.scala: onTableSchemaResolver.getTableSchemafailure, fetch the catalog table viasparkSession.sessionState.catalog.externalCatalog.getTable(...)and convert its schema instead of propagating the exception.TestCreateTable.scala: addedTest Select On Empty Table Falls Back To HMS Schema, which creates a table with no data written and verifies aselecton it succeeds (schema resolved from HMS) rather than throwing.Impact
No public API changes. Read behavior change: a
selecton a table whose schema can't be resolved from the file system (e.g. empty table, or non-hudi-only partitions) now succeeds using the HMS-registered schema instead of throwing.Risk Level
low
Documentation Update
none
Contributor's checklist