[KYUUBI #1592] [TEST][ICEBERG][DELTA] Introduce row level operation test for data lake format

### _Why are the changes needed?_

Introduce row level operation test for data lake format and remove redundant tests

### _How was this patch tested?_
- [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] [Run test](https://kyuubi.readthedocs.io/en/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #1592 from pan3793/test.

Closes #1592

892feb8 [Cheng Pan] Simplify test
72fd093 [Cheng Pan] Fix import
9d20839 [Cheng Pan] Add row level operation tests for Iceberg and Delta
7730b4d [Cheng Pan] Also test JDBC connection MetaData in Kyuubi server
7a41dfd [Cheng Pan] [TEST] Remove redundant type info test in DeltaMetadataTests

Authored-by: Cheng Pan <chengpan@apache.org>
Signed-off-by: Kent Yao <yao@apache.org>

Author: pan3793

externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/operation/SparkDeltaOperationSuite.scala

@@ -18,11 +18,13 @@
 package org.apache.kyuubi.engine.spark.operation
 
 import org.apache.kyuubi.engine.spark.WithSparkSQLEngine
-import org.apache.kyuubi.operation.DeltaMetadataTests
+import org.apache.kyuubi.operation.{DeltaMetadataTests, RowLevelOperationTests}
 import org.apache.kyuubi.tags.DeltaTest
 
 @DeltaTest
-class SparkDeltaOperationSuite extends WithSparkSQLEngine with DeltaMetadataTests {
+class SparkDeltaOperationSuite extends WithSparkSQLEngine
+  with DeltaMetadataTests
+  with RowLevelOperationTests {
 
   override protected def jdbcUrl: String = getJdbcUrl
 

externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/operation/SparkIcebergOperationSuite.scala

@@ -18,11 +18,13 @@
 package org.apache.kyuubi.engine.spark.operation
 
 import org.apache.kyuubi.engine.spark.WithSparkSQLEngine
-import org.apache.kyuubi.operation.IcebergMetadataTests
+import org.apache.kyuubi.operation.{IcebergMetadataTests, RowLevelOperationTests}
 import org.apache.kyuubi.tags.IcebergTest
 
 @IcebergTest
-class SparkIcebergOperationSuite extends WithSparkSQLEngine with IcebergMetadataTests {
+class SparkIcebergOperationSuite extends WithSparkSQLEngine
+  with IcebergMetadataTests
+  with RowLevelOperationTests {
 
   override protected def jdbcUrl: String = getJdbcUrl
 

externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/operation/SparkOperationSuite.scala

@@ -17,8 +17,6 @@
 
 package org.apache.kyuubi.engine.spark.operation
 
-import java.sql.{DatabaseMetaData, ResultSet, SQLException, SQLFeatureNotSupportedException}
-
 import scala.collection.JavaConverters._
 import scala.util.Random
 
@@ -32,7 +30,6 @@ import org.apache.spark.kyuubi.SparkContextHelper
 import org.apache.spark.sql.catalyst.analysis.FunctionRegistry
 import org.apache.spark.sql.types._
 
-import org.apache.kyuubi.Utils
 import org.apache.kyuubi.engine.spark.WithSparkSQLEngine
 import org.apache.kyuubi.engine.spark.shim.SparkCatalogShim
 import org.apache.kyuubi.operation.{HiveMetadataTests, SparkQueryTests}
@@ -240,7 +237,7 @@ class SparkOperationSuite extends WithSparkSQLEngine with HiveMetadataTests with
     }
   }
 
-  test("execute statement -  select decimal") {
+  test("execute statement - select decimal") {
     withJdbcStatement() { statement =>
       val resultSet = statement.executeQuery("SELECT 1.2BD as col1, 1.23BD AS col2")
       assert(resultSet.next())
@@ -256,7 +253,7 @@ class SparkOperationSuite extends WithSparkSQLEngine with HiveMetadataTests with
     }
   }
 
-  test("execute statement -  select column name with dots") {
+  test("execute statement - select column name with dots") {
     withJdbcStatement() { statement =>
       val resultSet = statement.executeQuery("select 'tmp.hello'")
       assert(resultSet.next())
@@ -305,190 +302,6 @@ class SparkOperationSuite extends WithSparkSQLEngine with HiveMetadataTests with
     }
   }
 
-  test("Hive JDBC Database MetaData API Auditing") {
-    withJdbcStatement() { statement =>
-      val metaData = statement.getConnection.getMetaData
-      Seq(
-        () => metaData.allProceduresAreCallable(),
-        () => metaData.getURL,
-        () => metaData.getUserName,
-        () => metaData.isReadOnly,
-        () => metaData.nullsAreSortedHigh,
-        () => metaData.nullsAreSortedLow,
-        () => metaData.nullsAreSortedAtStart(),
-        () => metaData.nullsAreSortedAtEnd(),
-        () => metaData.usesLocalFiles(),
-        () => metaData.usesLocalFilePerTable(),
-        () => metaData.supportsMixedCaseIdentifiers(),
-        () => metaData.supportsMixedCaseQuotedIdentifiers(),
-        () => metaData.storesUpperCaseIdentifiers(),
-        () => metaData.storesUpperCaseQuotedIdentifiers(),
-        () => metaData.storesLowerCaseIdentifiers(),
-        () => metaData.storesLowerCaseQuotedIdentifiers(),
-        () => metaData.storesMixedCaseIdentifiers(),
-        () => metaData.storesMixedCaseQuotedIdentifiers(),
-        () => metaData.getSQLKeywords,
-        () => metaData.nullPlusNonNullIsNull,
-        () => metaData.supportsConvert,
-        () => metaData.supportsTableCorrelationNames,
-        () => metaData.supportsDifferentTableCorrelationNames,
-        () => metaData.supportsExpressionsInOrderBy(),
-        () => metaData.supportsOrderByUnrelated,
-        () => metaData.supportsGroupByUnrelated,
-        () => metaData.supportsGroupByBeyondSelect,
-        () => metaData.supportsLikeEscapeClause,
-        () => metaData.supportsMultipleTransactions,
-        () => metaData.supportsMinimumSQLGrammar,
-        () => metaData.supportsCoreSQLGrammar,
-        () => metaData.supportsExtendedSQLGrammar,
-        () => metaData.supportsANSI92EntryLevelSQL,
-        () => metaData.supportsANSI92IntermediateSQL,
-        () => metaData.supportsANSI92FullSQL,
-        () => metaData.supportsIntegrityEnhancementFacility,
-        () => metaData.isCatalogAtStart,
-        () => metaData.supportsSubqueriesInComparisons,
-        () => metaData.supportsSubqueriesInExists,
-        () => metaData.supportsSubqueriesInIns,
-        () => metaData.supportsSubqueriesInQuantifieds,
-        // Spark support this, see https://issues.apache.org/jira/browse/SPARK-18455
-        () => metaData.supportsCorrelatedSubqueries,
-        () => metaData.supportsOpenCursorsAcrossCommit,
-        () => metaData.supportsOpenCursorsAcrossRollback,
-        () => metaData.supportsOpenStatementsAcrossCommit,
-        () => metaData.supportsOpenStatementsAcrossRollback,
-        () => metaData.getMaxBinaryLiteralLength,
-        () => metaData.getMaxCharLiteralLength,
-        () => metaData.getMaxColumnsInGroupBy,
-        () => metaData.getMaxColumnsInIndex,
-        () => metaData.getMaxColumnsInOrderBy,
-        () => metaData.getMaxColumnsInSelect,
-        () => metaData.getMaxColumnsInTable,
-        () => metaData.getMaxConnections,
-        () => metaData.getMaxCursorNameLength,
-        () => metaData.getMaxIndexLength,
-        () => metaData.getMaxSchemaNameLength,
-        () => metaData.getMaxProcedureNameLength,
-        () => metaData.getMaxCatalogNameLength,
-        () => metaData.getMaxRowSize,
-        () => metaData.doesMaxRowSizeIncludeBlobs,
-        () => metaData.getMaxStatementLength,
-        () => metaData.getMaxStatements,
-        () => metaData.getMaxTableNameLength,
-        () => metaData.getMaxTablesInSelect,
-        () => metaData.getMaxUserNameLength,
-        () => metaData.supportsTransactionIsolationLevel(1),
-        () => metaData.supportsDataDefinitionAndDataManipulationTransactions,
-        () => metaData.supportsDataManipulationTransactionsOnly,
-        () => metaData.dataDefinitionCausesTransactionCommit,
-        () => metaData.dataDefinitionIgnoredInTransactions,
-        () => metaData.getColumnPrivileges("", "%", "%", "%"),
-        () => metaData.getTablePrivileges("", "%", "%"),
-        () => metaData.getBestRowIdentifier("", "%", "%", 0, true),
-        () => metaData.getVersionColumns("", "%", "%"),
-        () => metaData.getExportedKeys("", "default", ""),
-        () => metaData.supportsResultSetConcurrency(ResultSet.TYPE_FORWARD_ONLY, 2),
-        () => metaData.ownUpdatesAreVisible(ResultSet.TYPE_FORWARD_ONLY),
-        () => metaData.ownDeletesAreVisible(ResultSet.TYPE_FORWARD_ONLY),
-        () => metaData.ownInsertsAreVisible(ResultSet.TYPE_FORWARD_ONLY),
-        () => metaData.othersUpdatesAreVisible(ResultSet.TYPE_FORWARD_ONLY),
-        () => metaData.othersDeletesAreVisible(ResultSet.TYPE_FORWARD_ONLY),
-        () => metaData.othersInsertsAreVisible(ResultSet.TYPE_FORWARD_ONLY),
-        () => metaData.updatesAreDetected(ResultSet.TYPE_FORWARD_ONLY),
-        () => metaData.deletesAreDetected(ResultSet.TYPE_FORWARD_ONLY),
-        () => metaData.insertsAreDetected(ResultSet.TYPE_FORWARD_ONLY),
-        () => metaData.supportsNamedParameters(),
-        () => metaData.supportsMultipleOpenResults,
-        () => metaData.supportsGetGeneratedKeys,
-        () => metaData.getSuperTypes("", "%", "%"),
-        () => metaData.getSuperTables("", "%", "%"),
-        () => metaData.getAttributes("", "%", "%", "%"),
-        () => metaData.getResultSetHoldability,
-        () => metaData.locatorsUpdateCopy,
-        () => metaData.supportsStatementPooling,
-        () => metaData.getRowIdLifetime,
-        () => metaData.supportsStoredFunctionsUsingCallSyntax,
-        () => metaData.autoCommitFailureClosesAllResultSets,
-        () => metaData.getClientInfoProperties,
-        () => metaData.getFunctionColumns("", "%", "%", "%"),
-        () => metaData.getPseudoColumns("", "%", "%", "%"),
-        () => metaData.generatedKeyAlwaysReturned).foreach { func =>
-        val e = intercept[SQLFeatureNotSupportedException](func())
-        assert(e.getMessage === "Method not supported")
-      }
-
-      import org.apache.kyuubi.KYUUBI_VERSION
-      assert(metaData.allTablesAreSelectable)
-      assert(metaData.getDatabaseProductName === "Apache Kyuubi (Incubating)")
-      assert(metaData.getDatabaseProductVersion === KYUUBI_VERSION)
-      assert(metaData.getDriverName === "Kyuubi Project Hive JDBC Shaded Client")
-      assert(metaData.getDriverVersion === KYUUBI_VERSION)
-      assert(metaData.getDatabaseMajorVersion === Utils.majorVersion(KYUUBI_VERSION))
-      assert(metaData.getDatabaseMinorVersion === Utils.minorVersion(KYUUBI_VERSION))
-      assert(
-        metaData.getIdentifierQuoteString === " ",
-        "This method returns a space \" \" if identifier quoting is not supported")
-      assert(metaData.getNumericFunctions === "")
-      assert(metaData.getStringFunctions === "")
-      assert(metaData.getSystemFunctions === "")
-      assert(metaData.getTimeDateFunctions === "")
-      assert(metaData.getSearchStringEscape === "\\")
-      assert(metaData.getExtraNameCharacters === "")
-      assert(metaData.supportsAlterTableWithAddColumn())
-      assert(!metaData.supportsAlterTableWithDropColumn())
-      assert(metaData.supportsColumnAliasing())
-      assert(metaData.supportsGroupBy)
-      assert(!metaData.supportsMultipleResultSets)
-      assert(!metaData.supportsNonNullableColumns)
-      assert(metaData.supportsOuterJoins)
-      assert(metaData.supportsFullOuterJoins)
-      assert(metaData.supportsLimitedOuterJoins)
-      assert(metaData.getSchemaTerm === "database")
-      assert(metaData.getProcedureTerm === "UDF")
-      assert(metaData.getCatalogTerm === "instance")
-      assert(metaData.getCatalogSeparator === ".")
-      assert(metaData.supportsSchemasInDataManipulation)
-      assert(!metaData.supportsSchemasInProcedureCalls)
-      assert(metaData.supportsSchemasInTableDefinitions)
-      assert(!metaData.supportsSchemasInIndexDefinitions)
-      assert(!metaData.supportsSchemasInPrivilegeDefinitions)
-      // This is actually supported, but hive jdbc package return false
-      assert(!metaData.supportsCatalogsInDataManipulation)
-      assert(!metaData.supportsCatalogsInProcedureCalls)
-      // This is actually supported, but hive jdbc package return false
-      assert(!metaData.supportsCatalogsInTableDefinitions)
-      assert(!metaData.supportsCatalogsInIndexDefinitions)
-      assert(!metaData.supportsCatalogsInPrivilegeDefinitions)
-      assert(!metaData.supportsPositionedDelete)
-      assert(!metaData.supportsPositionedUpdate)
-      assert(!metaData.supportsSelectForUpdate)
-      assert(!metaData.supportsStoredProcedures)
-      // This is actually supported, but hive jdbc package return false
-      assert(!metaData.supportsUnion)
-      assert(metaData.supportsUnionAll)
-      assert(metaData.getMaxColumnNameLength === 128)
-      assert(metaData.getDefaultTransactionIsolation === java.sql.Connection.TRANSACTION_NONE)
-      assert(!metaData.supportsTransactions)
-      assert(!metaData.getProcedureColumns("", "%", "%", "%").next())
-      intercept[SQLException](metaData.getPrimaryKeys("", "default", ""))
-      assert(!metaData.getImportedKeys("", "default", "").next())
-      intercept[SQLException] {
-        metaData.getCrossReference("", "default", "src", "", "default", "src2")
-      }
-      assert(!metaData.getIndexInfo("", "default", "src", true, true).next())
-
-      assert(metaData.supportsResultSetType(new Random().nextInt()))
-      assert(!metaData.supportsBatchUpdates)
-      assert(!metaData.getUDTs(",", "%", "%", null).next())
-      assert(!metaData.supportsSavepoints)
-      assert(!metaData.supportsResultSetHoldability(ResultSet.HOLD_CURSORS_OVER_COMMIT))
-      assert(metaData.getJDBCMajorVersion === 3)
-      assert(metaData.getJDBCMinorVersion === 0)
-      assert(metaData.getSQLStateType === DatabaseMetaData.sqlStateSQL)
-      assert(metaData.getMaxLogicalLobSize === 0)
-      assert(!metaData.supportsRefCursors)
-    }
-  }
-
   test("get operation status") {
     val sql = "select date_sub(date'2011-11-11', '1')"