[SPARK-30314] Add identifier and catalog information to DataSourceV2Relation #26957
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -44,15 +44,35 @@ class CatalogManager( | |
| import CatalogManager.SESSION_CATALOG_NAME | ||
|
|
||
| private val catalogs = mutable.HashMap.empty[String, CatalogPlugin] | ||
| // Map from catalog back to it's original name for easy name look up, we don't use the | ||
| // CatalogPlugin's name as it might be different from the catalog name depending on | ||
| // implementation. | ||
| private val catalogIdentifiers = mutable.HashMap.empty[CatalogPlugin, String] | ||
|
There was a problem hiding this comment. Do we need to handle There was a problem hiding this comment. no, that catalog doesn't have a name There was a problem hiding this comment. I'd add a comment here stating that we will have a new instance of a catalog for each catalog name, therefore this reverse map works as intended. We should also add a test for it There was a problem hiding this comment. Why is the reverse map needed? Can't we just call There was a problem hiding this comment. According to my understand There was a problem hiding this comment. The plugin name is intended to prevent needing to do this. While we do rely on the catalog not to report the wrong name, I think it is reasonable to use it. I'm not strongly against this, though. If you think this is cleaner we can do that. |
||
| catalogIdentifiers(defaultSessionCatalog) = SESSION_CATALOG_NAME | ||
|
There was a problem hiding this comment. Maybe you should do this as part of There was a problem hiding this comment. No, the concern here is that if a user set a custom session catalog implementation and spark fail to load it, it would fall back to use |
||
|
|
||
| def catalog(name: String): CatalogPlugin = synchronized { | ||
| if (name.equalsIgnoreCase(SESSION_CATALOG_NAME)) { | ||
| v2SessionCatalog | ||
| } else { | ||
| catalogs.getOrElseUpdate(name, { | ||
| val catalog = Catalogs.load(name, conf) | ||
| catalogIdentifiers(catalog) = name | ||
| catalog | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Returns the identifier string for the given catalog | ||
| * | ||
| * @param catalog catalog to look up | ||
| * @return string identifier for the given catalog. If the catalog hasn't be registered return | ||
| * None | ||
| */ | ||
| def catalogIdentifier(catalog: CatalogPlugin): Option[String] = synchronized { | ||
| catalogIdentifiers.get(catalog) | ||
| } | ||
|
|
||
| def isCatalogRegistered(name: String): Boolean = { | ||
| try { | ||
| catalog(name) | ||
|
|
@@ -83,7 +103,11 @@ class CatalogManager( | |
| private[sql] def v2SessionCatalog: CatalogPlugin = { | ||
| conf.getConf(SQLConf.V2_SESSION_CATALOG_IMPLEMENTATION).map { customV2SessionCatalog => | ||
| try { | ||
| catalogs.getOrElseUpdate(SESSION_CATALOG_NAME, { | ||
| val catalog = loadV2SessionCatalog() | ||
| catalogIdentifiers(catalog) = SESSION_CATALOG_NAME | ||
| catalog | ||
| }) | ||
| } catch { | ||
| case NonFatal(_) => | ||
| logError( | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -21,7 +21,7 @@ import org.apache.spark.sql.catalyst.analysis.{MultiInstanceRelation, NamedRelat | |
| import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference} | ||
| import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, LogicalPlan, Statistics} | ||
| import org.apache.spark.sql.catalyst.util.truncatedString | ||
| import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Identifier, Table, TableCapability} | ||
| import org.apache.spark.sql.connector.read.{Scan, ScanBuilder, Statistics => V2Statistics, SupportsReportStatistics} | ||
| import org.apache.spark.sql.connector.read.streaming.{Offset, SparkDataStream} | ||
| import org.apache.spark.sql.connector.write.WriteBuilder | ||
|
|
@@ -32,12 +32,18 @@ import org.apache.spark.util.Utils | |
| * A logical plan representing a data source v2 table. | ||
| * | ||
| * @param table The table that this relation represents. | ||
| * @param output the output attributes of this relation | ||
| * @param catalogIdentifier the string identifier for the catalog | ||
| * @param identifiers the identifiers for the v2 relation. For multipath dataframe, there could be | ||
| * more than one identifier | ||
|
There was a problem hiding this comment. or none if a V2 relation is instantiated using options |
||
| * @param options The options for this table operation. It's used to create fresh [[ScanBuilder]] | ||
| * and [[WriteBuilder]]. | ||
| */ | ||
| case class DataSourceV2Relation( | ||
| table: Table, | ||
| output: Seq[AttributeReference], | ||
| catalogIdentifier: Option[String], | ||
| identifiers: Seq[Identifier], | ||
|
There was a problem hiding this comment. I don't think it is a good idea to have multiple identifiers here. DSv2 doesn't yet cover how file-based tables should work and I think we need a design document for them. Adding multiple identifiers here in support of something that has undefined behavior seems premature. Design and behavior of path-based identifiers aside, a table should use one and only one identifier. When path-based tables are supported, I expect them to use a single There was a problem hiding this comment. I see. If the specification is that there should be one and only one identifier, shall I just define it as There was a problem hiding this comment. Two things:
|
||
| options: CaseInsensitiveStringMap) | ||
| extends LeafNode with MultiInstanceRelation with NamedRelation { | ||
|
|
||
|
|
@@ -137,12 +143,20 @@ case class StreamingDataSourceV2Relation( | |
| } | ||
|
|
||
| object DataSourceV2Relation { | ||
| def create( | ||
| table: Table, | ||
| catalogIdentifier: Option[String], | ||
| identifiers: Seq[Identifier], | ||
| options: CaseInsensitiveStringMap): DataSourceV2Relation = { | ||
| val output = table.schema().toAttributes | ||
| DataSourceV2Relation(table, output, catalogIdentifier, identifiers, options) | ||
| } | ||
|
|
||
| def create( | ||
| table: Table, | ||
| catalogIdentifier: Option[String], | ||
| identifiers: Seq[Identifier]): DataSourceV2Relation = | ||
| create(table, catalogIdentifier, identifiers, CaseInsensitiveStringMap.empty) | ||
|
|
||
| /** | ||
| * This is used to transform data source v2 statistics to logical.Statistics. | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,40 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.spark.sql.connector.catalog | ||
|
|
||
| import org.mockito.Mockito.{mock, when} | ||
|
|
||
| import org.apache.spark.SparkFunSuite | ||
| import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation | ||
| import org.apache.spark.sql.types.StructType | ||
|
|
||
| class CatalogV2UtilSuite extends SparkFunSuite { | ||
| test("Load relation should encode the identifiers for V2Relations") { | ||
| val testCatalog = mock(classOf[TableCatalog]) | ||
| val ident = mock(classOf[Identifier]) | ||
| val table = mock(classOf[Table]) | ||
| when(table.schema()).thenReturn(mock(classOf[StructType])) | ||
| when(testCatalog.loadTable(ident)).thenReturn(table) | ||
| val r = CatalogV2Util.loadRelation(testCatalog, Some("dummy"), ident) | ||
| assert(r.isDefined) | ||
| assert(r.get.isInstanceOf[DataSourceV2Relation]) | ||
| val v2Relation = r.get.asInstanceOf[DataSourceV2Relation] | ||
| assert(v2Relation.catalogIdentifier == Some("dummy")) | ||
| assert(v2Relation.identifiers == Seq(ident)) | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -195,6 +195,7 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging { | |
| } | ||
|
|
||
| DataSource.lookupDataSourceV2(source, sparkSession.sessionState.conf).map { provider => | ||
| val catalogManager = sparkSession.sessionState.catalogManager | ||
| val sessionOptions = DataSourceV2Utils.extractSessionConfigs( | ||
| source = provider, conf = sparkSession.sessionState.conf) | ||
| val pathsOption = if (paths.isEmpty) { | ||
|
|
@@ -206,27 +207,29 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging { | |
|
|
||
| val finalOptions = sessionOptions ++ extraOptions.toMap ++ pathsOption | ||
| val dsOptions = new CaseInsensitiveStringMap(finalOptions.asJava) | ||
| val (table, catalogOpt, ident) = provider match { | ||
| case _: SupportsCatalogOptions if userSpecifiedSchema.nonEmpty => | ||
| throw new IllegalArgumentException( | ||
| s"$source does not support user specified schema. Please don't specify the schema.") | ||
| case hasCatalog: SupportsCatalogOptions => | ||
| val ident = hasCatalog.extractIdentifier(dsOptions) | ||
| val catalog = CatalogV2Util.getTableProviderCatalog( | ||
| hasCatalog, | ||
| catalogManager, | ||
| dsOptions) | ||
| (catalog.loadTable(ident), catalogManager.catalogIdentifier(catalog), Seq(ident)) | ||
| case _ => | ||
| userSpecifiedSchema match { | ||
| case Some(schema) => (provider.getTable(dsOptions, schema), None, Nil) | ||
| case _ => (provider.getTable(dsOptions), None, Nil) | ||
|
There was a problem hiding this comment. There is no catalog in this case, do we need to care what's the table identifier? I.e. for here, looks like the There was a problem hiding this comment. They will be in the future, not yet though. This is fine for now |
||
| } | ||
| } | ||
| import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._ | ||
| table match { | ||
| case _: SupportsRead if table.supports(BATCH_READ) => | ||
| Dataset.ofRows( | ||
| sparkSession, | ||
| DataSourceV2Relation.create(table, catalogOpt, ident, dsOptions)) | ||
|
|
||
| case _ => loadV1Source(paths: _*) | ||
| } | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Would it be easier to have a separate util method for
UnresolvedV2Relation? The interesting thing aboutUnresolvedV2Relationis that you have access to theoriginalNameParts. If that doesn't contain the catalog identifier, then we shouldn't need to add it to theDataSourceV2RelationThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Sure, I will do the refactoring.
However in terms of using
originalNameParts, I think the major reason why we are adding the two additional fields to DSV2 relation is that both CatalogPlugin and Table implementation doesn't include the resolved unique identifier for the table. Even though the originalNameParts don't have a catalog name in it, it might point to some default catalog in which case we would still want it to be in the resolved unique identifier. Otherwise, we might have two V2Relations pointing to the same table but has different identifiers which would be inconsistent and pretty confusing.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think it's fine to reproduce the full identifier that was actually loaded by adding in catalog and namespace if they are missing. But I'm not sure it is a good idea to rely on this behavior since there is no guarantee.