Databricks plugin addition

[vishwasvaidya-cloudsufi]

No description provided.

[gemini-code-assist]

Code Review

This pull request introduces a new Databricks plugin, which includes a database connector, a batch source, schema reader, record handler, configuration classes, unit tests, and documentation. It also updates the shared database-commons module to support auto-commit and transaction isolation levels. Feedback on the changes highlights a critical issue where direct connection configuration is broken in DatabricksSourceConfig due to a missing httpPath field and incomplete getConnection logic. Additionally, it is recommended to refactor duplicated connection setup in DatabricksConnector to prevent potential NullPointerExceptions, implement actual random sampling in getRandomQuery using ORDER BY rand(), and add defensive null/empty checks for host and httpPath in canConnect.

[gemini-code-assist]

The DatabricksSourceConfig class does not support direct connection configuration (useConnection == false) because it is missing the httpPath field (which is specific to Databricks and not inherited from parent classes), and its getConnection() method only returns the connection field (which is null when useConnection is false). This will cause the plugin to fail to connect when configured directly in the pipeline.

To fix this, add the httpPath field to DatabricksSourceConfig and update getConnection() to construct a new DatabricksConnectorConfig using the direct fields when useConnection is false.

  public static class DatabricksSourceConfig extends AbstractDBSpecificSourceConfig {

    @Name(ConfigUtil.NAME_USE_CONNECTION)
    @Nullable
    @Description("Whether to use an existing connection.")
    private Boolean useConnection;

    @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() {
      return TransactionIsolationLevel.Level.TRANSACTION_READ_UNCOMMITTED.name();
    }

    @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
      );
    }

    @Override
    public void validate(FailureCollector collector) {
      ConfigUtil.validateConnection(this, useConnection, connection, collector);
      super.validate(collector);
    }
  }

[gemini-code-assist]

The connection configuration logic (setting the transaction isolation level and wrapping in NoOpCommitConnection) is duplicated in both getConnection(DBConnectorPath path) and getConnection(). Additionally, there is no null check on the connection returned by super.getConnection(), which could lead to a NullPointerException if the connection fails.

Refactor this logic into a private helper method to improve maintainability and safety.

  @Override
  protected Connection getConnection(DBConnectorPath path) {
    return configureConnection(super.getConnection(path));
  }

  @Override
  protected Connection getConnection() {
    return configureConnection(super.getConnection());
  }

  private Connection configureConnection(Connection connection) {
    if (connection == null) {
      return null;
    }
    try {
      connection.setTransactionIsolation(Connection.TRANSACTION_READ_UNCOMMITTED);
    } catch (SQLException e) {
      LOG.warn("Failed to set transaction isolation level to READ_UNCOMMITTED", e);
    }
    return new NoOpCommitConnection(connection);
  }

[gemini-code-assist]

The getRandomQuery method currently returns a sequential query with a LIMIT clause, which does not perform actual random sampling. This is misleading if the user selects 'Random' sampling in the UI.

To perform actual random sampling in Databricks, use ORDER BY rand(). Alternatively, if random sampling is not supported or is too inefficient for large tables, consider removing this override entirely so that the connector correctly indicates that random sampling is not supported, rather than silently falling back to a sequential limit query.

Suggested change

	@Override
	protected String getRandomQuery(String tableName, int limit) {
	return String.format("SELECT * FROM %s LIMIT %d", tableName, limit);
	}
	@Override
	protected String getRandomQuery(String tableName, int limit) {
	return String.format("SELECT * FROM %s ORDER BY rand() LIMIT %d", tableName, limit);
	}

[gemini-code-assist]

The canConnect() method checks for macros but does not verify if the required fields host and httpPath are null or empty. If they are missing, the connector will still attempt to connect, resulting in an invalid connection string and a confusing error.

Add defensive null/empty checks for host and httpPath in canConnect().

  @Override
  public boolean canConnect() {
    return super.canConnect() &&
      !com.google.common.base.Strings.isNullOrEmpty(host) && !containsMacro(ConnectionConfig.HOST) &&
      !containsMacro(ConnectionConfig.PORT) &&
      !com.google.common.base.Strings.isNullOrEmpty(httpPath) && !containsMacro(HTTP_PATH) &&
      !containsMacro(ConnectionConfig.DATABASE);
  }

[vishwasvaidya-cloudsufi]

raised by mistake