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 1 commit 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✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #19536 +/- ##
============================================
- Coverage 77.17% 77.16% -0.01%
- Complexity 33948 33953 +5
============================================
Files 2575 2576 +1
Lines 143397 143423 +26
Branches 17579 17581 +2
============================================
+ Hits 110662 110679 +17
- Misses 24476 24486 +10
+ Partials 8259 8258 -1
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.
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