[Spark] Hybrid Delta connector combining V1(default) and V2 connectors #5726
Conversation
huan233usc
requested review from
gengliangwang,
huan233usc,
raveeram-db and
tdas
| case Some(catalogTable) => | ||
| DeltaSourceUtils.isDeltaDataSourceName(s.sourceName) || | ||
| catalogTable.provider.exists(DeltaSourceUtils.isDeltaDataSourceName) || | ||
| CatalogTableUtils.isUnityCatalogManagedTable(catalogTable) |
There was a problem hiding this comment.
shall we skip checking CatalogTableUtils.isUnityCatalogManagedTable(catalogTable) here?
There was a problem hiding this comment.
the scope of the change is to handle only uc owned tables?
| // scalastyle:off caselocale | ||
| mode.toUpperCase match { | ||
| // scalastyle:on caselocale | ||
| case "STRICT" => |
There was a problem hiding this comment.
I think we can just return V2 table in DeltaCatalog in STRICT mode. The new analyzer rule only happens on AUTO mode
| case s: StreamingRelation if shouldApplyV2Streaming(s) => | ||
| // catalogTable is guaranteed to be defined because shouldApplyV2Streaming checks it | ||
| // via isDeltaStreamingRelation, but we use pattern matching for safety | ||
| s.dataSource.catalogTable match { |
There was a problem hiding this comment.
if shouldApplyV2Streaming returns yes, s.dataSource.catalogTable should be defined
| ScalaUtils.toJavaMap(catalogTable.properties)) | ||
|
|
||
| // Add a marker to indicate this schema comes from V2 streaming (SparkTable) | ||
| // This allows DeltaDataSource.sourceSchema to distinguish between: |
There was a problem hiding this comment.
DeltaDataSource.sourceSchema is called before this rule
| * UC-managed markers through Spark's normal catalog resolution paths, without manually patching | ||
| * CatalogTable objects in tests. | ||
| */ | ||
| class UCTableInjectingSessionCatalogInner extends DelegatingCatalogExtension { |
There was a problem hiding this comment.
why extending DelegatingCatalogExtension? customizing the loadTable method of UCTableInjectingSessionCatalog seems enough
3 participants
Which Delta project/connector is this regarding?
Description
This PR improves the Delta V2 streaming POC by:
AUTOas a validspark.databricks.delta.v2.enableModevalue and keeping production default asNONEwhile enablingAUTOby default in SQL tests.ApplyV2Streaming(extracted into its own rule) to rewrite V1StreamingRelationto V2StreamingRelationV2backed by KernelSparkTablewhen enabled.DeltaDataSource.sourceSchemaavoid redundant schema loading when the schema originated from the V2 streaming path, using a dedicated marker option.How was this patch tested?
./build/sbt "spark/testOnly org.apache.spark.sql.delta.V2StreamingConversionSuite"initialOffset with initial snapshot is not supported yet(TODO left in test to flip to a passing test once supported).Does this PR introduce any user-facing changes?
No.