[SPARK-46272][SQL] Support CTAS using DSv2 sources

[allisonwang-db]

What changes were proposed in this pull request?

#43949 supports CREATE TABLE using DSv2 sources. This PR supports CREATE TABLE AS SELECT (CTAS) using DSv2 sources. It turns out that we don't need additional code changes. This PR simply adds more test cases for CTAS queries.

Why are the changes needed?

To add tests for CTAS for DSv2 sources.

Does this PR introduce any user-facing change?

No

How was this patch tested?

New tests

Was this patch authored or co-authored using generative AI tooling?

No

[dongjoon-hyun]

When we change src/main, you don't need to use [TESTS], @allisonwang-db .

[allisonwang-db]

I see! Thanks for letting me know!

[allisonwang-db]

cc @cloud-fan

[cloud-fan]

This is not related to data source and we probably don't need it here.

[cloud-fan]

I'm a bit confused here. So for this data source, we probably don't check anything when creating the table. But when load the table, and get the Table instance in V2SessionCatalog, shall we check if Table#schema match what we stored in HMS?

[allisonwang-db]

This is actually tricky, and the behavior depends on the implementation of the data source.

In createTable, we store the query's schema (col1, col2, col3) in HMS, but this data source's getTable method does not take into account the input schema.

So, when the DSv2 relation is being created, it uses table.columns.asSchema which is the default implementation of the schema (i, j).

spark/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala

Lines 210 to 211 in bacdb3b

	val schema = CharVarcharUtils.replaceCharVarcharWithStringInSchema(table.columns.asSchema)
	DataSourceV2Relation(table, toAttributes(schema), catalog, identifier, options)

However, if the data source utilize user-defined schema in getTable, it won't throw this exception.

[cloud-fan]

I think we should enforce something here. When calling TableProvider.getTable with user-specified schema, we should make sure the returned table reports the same schema as the user-specified schema (probably ignore nullability). It's actually a documented requirement

   * Return a {@link Table} instance with the specified table schema, partitioning and properties
   * to do read/write. The returned table should report the same schema and partitioning with the
   * specified ones, or Spark may fail the operation.

[cloud-fan]

sqlState is required now. maybe 42K02 is fine

[cloud-fan]

shall we do this check in a lower level? where we instantiate the v2 Table.

[allisonwang-db]

Make sense

[cloud-fan]

can we check it in loadTable? In case the data source is a bit random and only return wrong schema at second time.

[allisonwang-db]

I tried it but looks like loadTable is used in many other places, such as in tableExists. If we check the schema there, it will fail commands other than create table (e.g DROP TABLE t will fail because the schema in catalogTable does not match the table schema)

[cloud-fan]

LGTM if all tests pass

[allisonwang-db]

@cloud-fan the test failure seems unrelated

[cloud-fan]

Can you retrigger the GA jobs?

[cloud-fan]

thanks, merging to master!