[SPARK-25528][SQL] data source v2 API refactor (batch read) #23086
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| /* | ||
| * 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.sources.v2; | ||
|
|
||
| import org.apache.spark.annotation.Evolving; | ||
| import org.apache.spark.sql.sources.v2.reader.Scan; | ||
| import org.apache.spark.sql.sources.v2.reader.ScanBuilder; | ||
|
|
||
| /** | ||
| * An empty mix-in interface for {@link Table}, to indicate this table supports batch scan. | ||
| * <p> | ||
| * If a {@link Table} implements this interface, its {@link Table#newScanBuilder(DataSourceOptions)} | ||
| * must return a {@link ScanBuilder} that builds {@link Scan} with {@link Scan#toBatch()} | ||
| * implemented. | ||
| * </p> | ||
| */ | ||
| @Evolving | ||
| public interface SupportsBatchRead extends Table { } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| /* | ||
| * 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.sources.v2; | ||
|
|
||
| import org.apache.spark.annotation.Evolving; | ||
| import org.apache.spark.sql.sources.v2.reader.Scan; | ||
| import org.apache.spark.sql.sources.v2.reader.ScanBuilder; | ||
| import org.apache.spark.sql.types.StructType; | ||
|
|
||
| /** | ||
| * An interface representing a logical structured data set of a data source. For example, the | ||
| * implementation can be a directory on the file system, a topic of Kafka, or a table in the | ||
| * catalog, etc. | ||
| * <p> | ||
| * This interface can mixin the following interfaces to support different operations: | ||
| * </p> | ||
| * <ul> | ||
| * <li>{@link SupportsBatchRead}: this table can be read in batch queries.</li> | ||
| * </ul> | ||
| */ | ||
| @Evolving | ||
| public interface Table { | ||
|
There was a problem hiding this comment. It would be helpful for a |
||
|
|
||
| /** | ||
| * A name to identify this table. | ||
| * <p> | ||
| * By default this returns the class name of the implementation. Please override it to provide a | ||
| * meaningful name, like the database and table name from catalog, or the location of files for | ||
| * this table. | ||
| * </p> | ||
| */ | ||
| default String name() { | ||
|
There was a problem hiding this comment. Do you think it's better to just ask implementations to override There was a problem hiding this comment. I don't think this should have a default. Implementations should definitely implement this. I think there is a difference between |
||
| return this.getClass().toString(); | ||
| } | ||
|
|
||
| /** | ||
| * Returns the schema of this table. | ||
| */ | ||
| StructType schema(); | ||
|
|
||
| /** | ||
| * Returns a {@link ScanBuilder} which can be used to build a {@link Scan} later. Spark will call | ||
| * this method for each data scanning query. | ||
| * <p> | ||
| * The builder can take some query specific information to do operators pushdown, and keep these | ||
| * information in the created {@link Scan}. | ||
| * </p> | ||
| */ | ||
| ScanBuilder newScanBuilder(DataSourceOptions options); | ||
|
There was a problem hiding this comment.
We should to reconsider There was a problem hiding this comment. Makes sense to me - There was a problem hiding this comment. I agree with it. Since There was a problem hiding this comment. Either in a follow-up or you can add the class in this PR. Either way works for me. |
||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,64 @@ | ||
| /* | ||
| * 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.sources.v2; | ||
|
|
||
| import org.apache.spark.annotation.Evolving; | ||
| import org.apache.spark.sql.sources.DataSourceRegister; | ||
| import org.apache.spark.sql.types.StructType; | ||
|
|
||
| /** | ||
| * The base interface for v2 data sources which don't have a real catalog. Implementations must | ||
| * have a public, 0-arg constructor. | ||
| * <p> | ||
| * The major responsibility of this interface is to return a {@link Table} for read/write. | ||
| * </p> | ||
| */ | ||
| @Evolving | ||
| // TODO: do not extend `DataSourceV2`, after we finish the API refactor completely. | ||
| public interface TableProvider extends DataSourceV2 { | ||
|
|
||
| /** | ||
| * Return a {@link Table} instance to do read/write with user-specified options. | ||
| * | ||
| * @param options the user-specified options that can identify a table, e.g. file path, Kafka | ||
| * topic name, etc. It's an immutable case-insensitive string-to-string map. | ||
| */ | ||
| Table getTable(DataSourceOptions options); | ||
|
|
||
| /** | ||
| * Return a {@link Table} instance to do read/write with user-specified schema and options. | ||
| * <p> | ||
| * By default this method throws {@link UnsupportedOperationException}, implementations should | ||
|
There was a problem hiding this comment. Javadoc would normally also add There was a problem hiding this comment. what I learned is that, we should only declare checked exceptions. See http://www.javapractices.com/topic/TopicAction.do?Id=171 There was a problem hiding this comment. Strange, that page links to one with the opposite advice: http://www.javapractices.com/topic/TopicAction.do?Id=44 I think that The page you linked to makes the argument that unchecked exceptions aren't part of the method contract and cannot be relied on. But documenting this shows that it is part of the contract or expected behavior, so I think docs are appropriate. There was a problem hiding this comment. added the throw clause. |
||
| * override this method to handle user-specified schema. | ||
| * </p> | ||
| * @param options the user-specified options that can identify a table, e.g. file path, Kafka | ||
| * topic name, etc. It's an immutable case-insensitive string-to-string map. | ||
| * @param schema the user-specified schema. | ||
| * @throws UnsupportedOperationException | ||
| */ | ||
| default Table getTable(DataSourceOptions options, StructType schema) { | ||
|
There was a problem hiding this comment. I know that this is from prior DataSourceV2 semantics, but what's the difference between providing the There was a problem hiding this comment. Basically just saying we should just push down this requested schema into the There was a problem hiding this comment. It's a different thing. Think about you are reading a parquet file, and you know exactly what its physical schema is, and you don't want Spark to waste a job to infer the schema. Then you can specify the schema when reading. Next, Spark will analyze the query, and figure out what the required schema is. This step is automatic and driven by Spark. There was a problem hiding this comment. I agree with @cloud-fan. These are slightly different uses. Here, it is supplying a schema for how to interpret data files. Say you have CSV files with columns |
||
| String name; | ||
| if (this instanceof DataSourceRegister) { | ||
| name = ((DataSourceRegister) this).shortName(); | ||
| } else { | ||
| name = this.getClass().getName(); | ||
| } | ||
| throw new UnsupportedOperationException( | ||
| name + " source does not support user-specified schema"); | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| /* | ||
| * 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.sources.v2.reader; | ||
|
|
||
| import org.apache.spark.annotation.Evolving; | ||
|
|
||
| /** | ||
| * A physical representation of a data source scan for batch queries. This interface is used to | ||
| * provide physical information, like how many partitions the scanned data has, and how to read | ||
| * records from the partitions. | ||
| */ | ||
| @Evolving | ||
| public interface Batch { | ||
|
There was a problem hiding this comment. I don't have a strong preference. I feel it's a little more clear to distinguish between scan and batch |
||
|
|
||
| /** | ||
| * Returns a list of {@link InputPartition input partitions}. Each {@link InputPartition} | ||
| * represents a data split that can be processed by one Spark task. The number of input | ||
| * partitions returned here is the same as the number of RDD partitions this scan outputs. | ||
| * <p> | ||
| * If the {@link Scan} supports filter pushdown, this Batch is likely configured with a filter | ||
| * and is responsible for creating splits for that filter, which is not a full scan. | ||
| * </p> | ||
| * <p> | ||
| * This method will be called only once during a data source scan, to launch one Spark job. | ||
| * </p> | ||
| */ | ||
| InputPartition[] planInputPartitions(); | ||
|
|
||
| /** | ||
| * Returns a factory, which produces one {@link PartitionReader} for one {@link InputPartition}. | ||
| */ | ||
| PartitionReaderFactory createReaderFactory(); | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,38 @@ | ||
| /* | ||
| * 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.sources.v2.reader; | ||
|
|
||
| import org.apache.spark.annotation.Evolving; | ||
| import org.apache.spark.sql.sources.v2.reader.partitioning.Partitioning; | ||
|
|
||
| /** | ||
| * A mix in interface for {@link BatchReadSupport}. Data sources can implement this interface to | ||
| * report data partitioning and try to avoid shuffle at Spark side. | ||
| * | ||
| * Note that, when a {@link ReadSupport} implementation creates exactly one {@link InputPartition}, | ||
| * Spark may avoid adding a shuffle even if the reader does not implement this interface. | ||
| */ | ||
| @Evolving | ||
| // TODO: remove it, after we finish the API refactor completely. | ||
| public interface OldSupportsReportPartitioning extends ReadSupport { | ||
|
|
||
| /** | ||
| * Returns the output data partitioning that this reader guarantees. | ||
| */ | ||
| Partitioning outputPartitioning(ScanConfig config); | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,38 @@ | ||
| /* | ||
| * 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.sources.v2.reader; | ||
|
|
||
| import org.apache.spark.annotation.Evolving; | ||
|
|
||
| /** | ||
| * A mix in interface for {@link BatchReadSupport}. Data sources can implement this interface to | ||
| * report statistics to Spark. | ||
| * | ||
| * As of Spark 2.4, statistics are reported to the optimizer before any operator is pushed to the | ||
| * data source. Implementations that return more accurate statistics based on pushed operators will | ||
| * not improve query performance until the planner can push operators before getting stats. | ||
| */ | ||
| @Evolving | ||
| // TODO: remove it, after we finish the API refactor completely. | ||
| public interface OldSupportsReportStatistics extends ReadSupport { | ||
|
|
||
| /** | ||
| * Returns the estimated statistics of this data source scan. | ||
| */ | ||
| Statistics estimateStatistics(ScanConfig config); | ||
| } |
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.
#21306 (TableCatalog support) adds this class as
org.apache.spark.sql.catalog.v2.Tablein thespark-catalystmodule. I think it needs to be in the catalyst module and should probably be in theo.a.s.sql.catalog.v2package as well.The important one is moving this to the catalyst module. The analyzer is in catalyst and all of the v2 logical plans and analysis rules will be in catalyst as well, because we are standardizing behavior. The standard validation rules should be in catalyst, not in a source-specific or hive-specific package in the sql-core or hive modules.
Because the logical plans and validation rules are in the catalyst package, the
TableCatalogAPI needs to be there as well. For example, when a catalog table identifier is resolved for a read query, one of the results is aTableCataloginstance for the catalog portion of the identifier. That catalog is used to load the v2 table, which is then wrapped in a v2 relation for further analysis. Similarly, the write path should also validate that the catalog exists during analysis by loading it, and would then pass the catalog in a v2 logical plan forCreateTableorCreateTableAsSelect.I also think that it makes sense to use the
org.apache.spark.sql.catalog.v2package forTablebecauseTableis more closely tied to theTableCatalogAPI than to the data source API. The link to DSv2 is thatTablecarriesnewScanBuilder, but the rest of the methods exposed byTableare for catalog functions, like inspecting a table's partitioning or table properties.Moving this class would make adding
TableCatalogless intrusive.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.
Moving this to the Catalyst package would set a precedent for user-overridable behavior to live in the catalyst project. I'm not aware of anything in the Catalyst package being considered as public API right now. Are we allowed to start such a convention at this juncture?
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.
Everything in catalyst is considered private (although public visibility for debugging) and it's best to stay that way.
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.
why does this
TableAPI need to be in catalyst? It's not even a plan. We can define a table LogicalPlan interface in catalyst, and implement it in the SQL module with thisTableAPI.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 can understand wanting to keep everything in Catalyst private. That's fine with me, but I think that Catalyst does need to be able to interact with tables and catalogs that are supplied by users.
For example: Our tables support schema evolution. Specifically, reading files that were written before a column was added. When we add a column, Spark shouldn't start failing in analysis for an AppendData operation in a scheduled job (as it would today). We need to be able to signal to the validation rule that the table supports reading files that are missing columns, so that Spark can do the right validation and allow writes that used to work to continue.
How would that information -- support for reading missing columns -- be communicated to the analyzer?
Also, what about my example above: how will the analyzer load tables using a user-supplied catalog if catalyst can't use any user-supplied implementations?
We could move all of the v2 analysis rules, like ResolveRelations, into the core module, but it seems to me that this requirement is no longer providing value if we have to do that. I think that catalyst is the right place for common plans and analysis rules to live because it is the library of common SQL components.
Wherever the rules and plans end up, they will need to access to the
TableCatalogAPI.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.
It's unclear to me what would be the best choice:
Can we delay the discussion when we have a PR to add catalog support after the refactor?
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.
Yes, that works.
But, can we move
Tableto theorg.apache.spark.sql.catalog.v2package whereTableCatalogis defined in the other PR? I thinkTableshould be defined with the catalog API and moving that later would require import changes to any file that referencesTable.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.
for other reviewers: in the ds v2 community sync, we decided to move data source v2 into a new module
sql-api, and make catalyst depends on it. This will be done in a followup.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 just went to make this change, but it requires moving any SQL class from catalyst referenced by the API into the API module as well... Let's discuss the options more on the dev list thread.