[SPARK-31181][SQL][TESTS] Remove the default value assumption on CREATE TABLE test cases

### What changes were proposed in this pull request?

A few `CREATE TABLE` test cases have some assumption on the default value of `LEGACY_CREATE_HIVE_TABLE_BY_DEFAULT_ENABLED`. This PR (SPARK-31181) makes the test cases more explicit from test-case side.

The configuration change was tested via #27894 during discussing SPARK-31136. This PR has only the test case part from that PR.

### Why are the changes needed?

This makes our test case more robust in terms of the default value of `LEGACY_CREATE_HIVE_TABLE_BY_DEFAULT_ENABLED`. Even in the case where we switch the conf value, that will be one-liner with no test case changes.

### Does this PR introduce any user-facing change?

No.

### How was this patch tested?

Pass the Jenkins with the existing tests.

Closes #27946 from dongjoon-hyun/SPARK-EXPLICIT-TEST.

Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: dongjoon-hyun

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala

@@ -26,6 +26,7 @@ import org.apache.spark.sql.catalyst.expressions.{EqualTo, Literal}
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.connector.catalog.TableChange.ColumnPosition.{after, first}
 import org.apache.spark.sql.connector.expressions.{ApplyTransform, BucketTransform, DaysTransform, FieldReference, HoursTransform, IdentityTransform, LiteralValue, MonthsTransform, Transform, YearsTransform}
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructType, TimestampType}
 import org.apache.spark.unsafe.types.UTF8String
 
@@ -2163,18 +2164,20 @@ class DDLParserSuite extends AnalysisTest {
   }
 
   test("create table - without using") {
-    val sql = "CREATE TABLE 1m.2g(a INT)"
-    val expectedTableSpec = TableSpec(
-      Seq("1m", "2g"),
-      Some(new StructType().add("a", IntegerType)),
-      Seq.empty[Transform],
-      None,
-      Map.empty[String, String],
-      None,
-      Map.empty[String, String],
-      None,
-      None)
+    withSQLConf(SQLConf.LEGACY_CREATE_HIVE_TABLE_BY_DEFAULT_ENABLED.key -> "false") {
+      val sql = "CREATE TABLE 1m.2g(a INT)"
+      val expectedTableSpec = TableSpec(
+        Seq("1m", "2g"),
+        Some(new StructType().add("a", IntegerType)),
+        Seq.empty[Transform],
+        None,
+        Map.empty[String, String],
+        None,
+        Map.empty[String, String],
+        None,
+        None)
 
-    testCreateOrReplaceDdl(sql, expectedTableSpec, expectedIfNotExists = false)
+      testCreateOrReplaceDdl(sql, expectedTableSpec, expectedIfNotExists = false)
+    }
   }
 }

sql/core/src/test/resources/sql-tests/inputs/describe-table-column.sql

@@ -52,7 +52,7 @@ DROP TABLE desc_complex_col_table;
 
 --Test case insensitive
 
-CREATE TABLE customer(CName STRING);
+CREATE TABLE customer(CName STRING) USING PARQUET;
 
 INSERT INTO customer VALUES('Maria');
 

sql/core/src/test/resources/sql-tests/inputs/postgreSQL/create_view.sql

@@ -41,7 +41,7 @@ DROP TABLE emp;
 -- These views are left around mainly to exercise special cases in pg_dump.
 
 -- [SPARK-19842] Informational Referential Integrity Constraints Support in Spark
-CREATE TABLE view_base_table (key int /* PRIMARY KEY */, data varchar(20));
+CREATE TABLE view_base_table (key int /* PRIMARY KEY */, data varchar(20)) USING PARQUET;
 --
 CREATE VIEW key_dependent_view AS
    SELECT * FROM view_base_table GROUP BY key;

sql/core/src/test/resources/sql-tests/results/describe-table-column.sql.out

@@ -267,7 +267,7 @@ struct<>
 
 
 -- !query
-CREATE TABLE customer(CName STRING)
+CREATE TABLE customer(CName STRING) USING PARQUET
 -- !query schema
 struct<>
 -- !query output

sql/core/src/test/resources/sql-tests/results/postgreSQL/create_view.sql.out

@@ -42,7 +42,7 @@ struct<>
 
 
 -- !query
-CREATE TABLE view_base_table (key int /* PRIMARY KEY */, data varchar(20))
+CREATE TABLE view_base_table (key int /* PRIMARY KEY */, data varchar(20)) USING PARQUET
 -- !query schema
 struct<>
 -- !query output

sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala

@@ -257,6 +257,7 @@ class DataSourceV2SQLSuite
   }
 
   test("CreateTable: without USING clause") {
+    spark.conf.set(SQLConf.LEGACY_CREATE_HIVE_TABLE_BY_DEFAULT_ENABLED.key, "false")
     // unset this config to use the default v2 session catalog.
     spark.conf.unset(V2_SESSION_CATALOG_IMPLEMENTATION.key)
     val testCatalog = catalog("testcat").asTableCatalog
@@ -681,6 +682,7 @@ class DataSourceV2SQLSuite
   }
 
   test("CreateTableAsSelect: without USING clause") {
+    spark.conf.set(SQLConf.LEGACY_CREATE_HIVE_TABLE_BY_DEFAULT_ENABLED.key, "false")
     // unset this config to use the default v2 session catalog.
     spark.conf.unset(V2_SESSION_CATALOG_IMPLEMENTATION.key)
     val testCatalog = catalog("testcat").asTableCatalog

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLParserSuite.scala

@@ -40,7 +40,8 @@ import org.apache.spark.sql.test.SharedSparkSession
 import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
 
 class DDLParserSuite extends AnalysisTest with SharedSparkSession {
-  private lazy val parser = new SparkSqlParser(new SQLConf)
+  private lazy val parser = new SparkSqlParser(new SQLConf().copy(
+    SQLConf.LEGACY_CREATE_HIVE_TABLE_BY_DEFAULT_ENABLED -> false))
 
   private def assertUnsupported(sql: String, containsThesePhrases: Seq[String] = Seq()): Unit = {
     val e = intercept[ParseException] {

sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala

@@ -846,16 +846,16 @@ class InsertSuite extends DataSourceTest with SharedSparkSession {
   test("SPARK-29174 Support LOCAL in INSERT OVERWRITE DIRECTORY to data source") {
     withTempPath { dir =>
       val path = dir.toURI.getPath
-      sql(s"""create table tab1 ( a int) location '$path'""")
+      sql(s"""create table tab1 ( a int) using parquet location '$path'""")
       sql("insert into tab1 values(1)")
       checkAnswer(sql("select * from tab1"), Seq(1).map(i => Row(i)))
-      sql("create table tab2 ( a int)")
+      sql("create table tab2 ( a int) using parquet")
       sql("insert into tab2 values(2)")
       checkAnswer(sql("select * from tab2"), Seq(2).map(i => Row(i)))
       sql(s"""insert overwrite local directory '$path' using parquet select * from tab2""")
       sql("refresh table tab1")
       checkAnswer(sql("select * from tab1"), Seq(2).map(i => Row(i)))
-      }
+    }
   }
 
   test("SPARK-29174 fail LOCAL in INSERT OVERWRITE DIRECT remote path") {

sql/hive/src/test/scala/org/apache/spark/sql/hive/StatisticsSuite.scala

@@ -1520,10 +1520,12 @@ class StatisticsSuite extends StatisticsCollectionTestBase with TestHiveSingleto
     val ext_tbl = "SPARK_30269_external"
     withTempDir { dir =>
       withTable(tbl, ext_tbl) {
-        sql(s"CREATE TABLE $tbl (key INT, value STRING, ds STRING) PARTITIONED BY (ds)")
+        sql(s"CREATE TABLE $tbl (key INT, value STRING, ds STRING)" +
+          "USING parquet PARTITIONED BY (ds)")
         sql(
           s"""
              | CREATE TABLE $ext_tbl (key INT, value STRING, ds STRING)
+             | USING PARQUET
              | PARTITIONED BY (ds)
              | LOCATION '${dir.toURI}'
            """.stripMargin)