[vishwasvaidya-cloudsufi]

[🍒]

🍒 [cherrypick]

PR : (#680)

Description:

Addition of Databricks Plugin

[gemini-code-assist]

Code Review

This pull request introduces a new Databricks plugin, including connector and source implementations, along with necessary configuration and test files. It also updates the project version to 1.12.6-SNAPSHOT across all modules. The review feedback identified a critical issue in the DatabricksSource configuration where connection properties are not correctly bound when not using a reusable connection, as well as potential NullPointerExceptions in the DatabricksConnector and an opportunity to improve the DatabricksDBRecord field handling using standard JDBC methods.

[gemini-code-assist]

When useConnection is false (i.e., the user configures connection details directly in the source instead of using a reusable connection), DatabricksSourceConfig fails to bind and use the connection properties correctly:

1. The httpPath field is missing from DatabricksSourceConfig, meaning the configured HTTP Path is completely lost.
2. getConnection() returns null because it only returns the connection field (which is null when useConnection is false). This causes createConnectionString() to return null and breaks lineage recording (host and database are recorded as null).
3. getTransactionIsolationLevel() falls back to the default TRANSACTION_REPEATABLE_READ even if a custom level was configured in the source.

To fix this, we should add the httpPath field, update getConnection() to construct and return a new DatabricksConnectorConfig when useConnection is false, and update getTransactionIsolationLevel() to use getConnection() instead of directly accessing the connection field.

    @Name(ConfigUtil.NAME_CONNECTION)
    @Macro
    @Nullable
    @Description("The existing connection to use.")
    private DatabricksConnectorConfig connection;

    @Name(DatabricksConnectorConfig.HTTP_PATH)
    @Description("The HTTP Path for the Databricks cluster or SQL warehouse.")
    @Macro
    @Nullable
    private String httpPath;

    @Override
    public Map<String, String> getDBSpecificArguments() {
      return Collections.emptyMap();
    }

    @VisibleForTesting
    public DatabricksSourceConfig(@Nullable Boolean useConnection,
                                  @Nullable DatabricksConnectorConfig connection) {
      this.useConnection = useConnection;
      this.connection = connection;
    }

    @Override
    public String getTransactionIsolationLevel() {
      DatabricksConnectorConfig conn = getConnection();
      return conn == null
        ? TransactionIsolationLevel.Level.TRANSACTION_REPEATABLE_READ.name()
        : conn.getTransactionIsolationLevel();
    }

    @Override
    public Integer getFetchSize() {
      Integer fetchSize = super.getFetchSize();
      return fetchSize == null ? Integer.parseInt(DEFAULT_FETCH_SIZE) : fetchSize;
    }

    @Override
    protected DatabricksConnectorConfig getConnection() {
      if (useConnection != null && useConnection) {
        return connection;
      }
      return new DatabricksConnectorConfig(
        user, password, jdbcPluginName, connectionArguments,
        host, httpPath, database, port, super.getTransactionIsolationLevel()
      );
    }

[gemini-code-assist]

Using resultSet.getString(columnIndex) is standard JDBC and directly returns the string representation (or null if SQL NULL), avoiding the overhead of getObject() and manual null-checking/toString conversion. It is also safer because some JDBC drivers might return custom objects for complex types whose toString() does not yield the raw string/JSON representation, whereas getString() is guaranteed to return the string representation.

Suggested change

	Object value = resultSet.getObject(columnIndex);
	recordBuilder.set(field.getName(), value != null ? value.toString() : null);
	recordBuilder.set(field.getName(), resultSet.getString(columnIndex));

[gemini-code-assist]

Add a null check for connection before calling setTransactionIsolation to prevent potential NullPointerException if super.getConnection(path) returns null.

    Connection connection = super.getConnection(path);
    if (connection != null) {
      try {
        connection.setTransactionIsolation(
          TransactionIsolationLevel.getLevel(getTransactionIsolationLevel()));
      } catch (SQLException e) {
        LOG.warn("Failed to set transaction isolation level to {}", getTransactionIsolationLevel(), e);
      }
    }
    return new NoOpCommitConnection(connection);

[gemini-code-assist]

Add a null check for connection before calling setTransactionIsolation to prevent potential NullPointerException if super.getConnection() returns null.

    Connection connection = super.getConnection();
    if (connection != null) {
      try {
        connection.setTransactionIsolation(
          TransactionIsolationLevel.getLevel(getTransactionIsolationLevel()));
      } catch (SQLException e) {
        LOG.warn("Failed to set transaction isolation level to {}", getTransactionIsolationLevel(), e);
      }
    }
    return new NoOpCommitConnection(connection);