Cherry-pick from oracle-structsupport

[vanshikaagupta22]

• cherry-pick Provide Struct data type support for oracle plugin #659

• update to 1.11.15-SNAPSHOT version

[gemini-code-assist]

Code Review

This pull request introduces support for Oracle STRUCT (User-Defined Types) in the Oracle database plugin, enabling the mapping of structured database types to CDAP StructuredRecord schemas and records. It implements schema reading, attribute mapping, and data conversion for nested structures and Oracle-specific types, along with comprehensive unit tests. The review feedback identifies several critical robustness and performance issues, including potential NullPointerException risks when retrieving connections from result set statements or loading classes via the bootstrap classloader, SQL query ambiguity when extracting type owners, missing validation for null type owners, and performance bottlenecks from repeated reflection calls.

[gemini-code-assist]

There are two issues here:

1. this.connection can be null if getSchema is called directly without getSchemaFields being called first, leading to a NullPointerException.
2. The query SELECT DATA_TYPE_OWNER FROM ALL_TAB_COLS WHERE COLUMN_NAME = ? AND DATA_TYPE = ? is too loose because it queries all tables in the database. If multiple tables have the same column name and type name but different owners, it will return multiple rows and non-deterministically pick the first one. We should restrict the query by TABLE_NAME and OWNER (table schema) using metadata.getTableName(index) and metadata.getSchemaName(index) if they are available.

    String owner = extractOwnerName(metadata, index, columnTypeName, sqlType);
    ColumnMetadata columnMetadata = new ColumnMetadata(sqlType, metadata.getColumnClassName(index),
            metadata.getPrecision(index), metadata.getScale(index), metadata.getColumnName(index),
            columnTypeName, metadata.isSigned(index), owner, 0);
    return getSchemaMapping(columnMetadata);
  }

  private String extractOwnerName(ResultSetMetaData metadata, int index, String columnTypeName, int sqlType) throws SQLException {
    if (columnTypeName != null && columnTypeName.contains(".")) {
      return columnTypeName.substring(0, columnTypeName.lastIndexOf('.'));
    }
    if (sqlType == Types.STRUCT) {
      if (this.connection == null) {
        throw new IllegalStateException("Database connection is not initialized. Connection is required to resolve STRUCT type owner.");
      }
      String columnName = metadata.getColumnName(index);
      String tableName = metadata.getTableName(index);
      String tableSchema = metadata.getSchemaName(index);
      StringBuilder sql = new StringBuilder("SELECT DATA_TYPE_OWNER FROM ALL_TAB_COLS WHERE COLUMN_NAME = ? AND DATA_TYPE = ?");
      if (tableName != null && !tableName.isEmpty()) {
        sql.append(" AND TABLE_NAME = ?");
      }
      if (tableSchema != null && !tableSchema.isEmpty()) {
        sql.append(" AND OWNER = ?");
      }
      try (PreparedStatement stmt = this.connection.prepareStatement(sql.toString())) {
        stmt.setString(1, columnName);
        stmt.setString(2, columnTypeName);
        int paramIndex = 3;
        if (tableName != null && !tableName.isEmpty()) {
          stmt.setString(paramIndex++, tableName);
        }
        if (tableSchema != null && !tableSchema.isEmpty()) {
          stmt.setString(paramIndex++, tableSchema);
        }
        try (ResultSet rs = stmt.executeQuery()) {
          if (rs.next()) {
            return rs.getString("DATA_TYPE_OWNER");
          }
        }
      } catch (SQLException e) {
        throw new IllegalArgumentException(String.format("Column '%s' is not a primitive type,"
                + "but it lacks a type owner. Therefore, it cannot be resolved as "
                + "a STRUCT type. ", columnName), e);
      }
    }
    return null;
  }

[gemini-code-assist]

Calling resultSet.getStatement().getConnection() directly can throw a NullPointerException if resultSet.getStatement() returns null. According to the JDBC specification, getStatement() may return null if the ResultSet was not created by a Statement (e.g., when created via database metadata or in certain mock/test environments). It is safer to guard against a null statement before retrieving the connection.

    java.sql.Statement statement = resultSet.getStatement();
    java.sql.Connection connection = statement != null ? statement.getConnection() : null;
    populateRecordField(connection, recordBuilder, field, fieldValue);

[gemini-code-assist]

If the class of bfile is loaded by the bootstrap classloader, getClassLoader() can return null. Calling classLoader.loadClass(...) directly will then throw a NullPointerException. We should provide a fallback classloader if getClassLoader() returns null.

      ClassLoader classLoader = bfile.getClass().getClassLoader();
      if (classLoader == null) {
        classLoader = OracleSourceDBRecord.class.getClassLoader();
      }

[gemini-code-assist]

Calling resultSet.getStatement().getConnection() directly can throw a NullPointerException if resultSet.getStatement() returns null. It is safer to guard against a null statement before retrieving the connection.

Suggested change

	if (structValue != null) {
	recordBuilder.set(field.getName(), convertStructToRecord(structValue, nonNullSchema,
	resultSet.getStatement().getConnection()));
	} else {
	if (structValue != null) {
	java.sql.Statement statement = resultSet.getStatement();
	java.sql.Connection connection = statement != null ? statement.getConnection() : null;
	recordBuilder.set(field.getName(), convertStructToRecord(structValue, nonNullSchema, connection));
	} else {

[gemini-code-assist]

Using reflection (getMethod and invoke) for every single row/record can cause a significant performance bottleneck when processing large datasets. Consider caching the getDescriptor and getMetaData Method objects in static or instance fields to avoid the overhead of repeatedly looking them up via reflection.

[gemini-code-assist]

If metadata.getOwner() (which is owner) is null or empty, the query in getStructSchema will use OWNER = NULL which always evaluates to false in SQL, returning no rows. This will eventually throw a generic and confusing "No attributes found..." exception. We should validate that the owner is resolved successfully before calling getStructSchema.

        String owner = metadata.getOwner();
        if (owner == null || owner.isEmpty()) {
          throw new IllegalArgumentException(String.format("Type owner for STRUCT column '%s' of type '%s' could not be resolved.", columnName, columnTypeName));
        }
        return getStructSchema(connection, columnTypeName, owner,
                columnName, nestingLevel);