Addition of Databricks Plugin

database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnector.java

@@ -104,8 +104,12 @@ public InputFormatProvider getInputFormatProvider(ConnectorContext context, Samp
     String tableQuery = getTableQuery(path.getDatabase(), path.getSchema(), path.getTable(), request.getLimit(),
       request.getProperties().get("sampleType"), request.getProperties().get("strata"), sessionID);
     DataDrivenETLDBInputFormat.setInput(connectionConfigAccessor.getConfiguration(), getDBRecordType(),
-      tableQuery, null, false);
+      tableQuery, null, isAutoCommitEnabled());
     connectionConfigAccessor.setConnectionArguments(Maps.fromProperties(config.getConnectionArgumentsProperties()));
+    String isolationLevel = getTransactionIsolationLevel();
+    if (isolationLevel != null) {
+      connectionConfigAccessor.setTransactionIsolationLevel(isolationLevel);
+    }
     connectionConfigAccessor.getConfiguration().setInt(MRJobConfig.NUM_MAPS, 1);
     Map<String, String> additionalArguments = config.getAdditionalArguments();
     for (Map.Entry<String, String> argument : additionalArguments.entrySet()) {
@@ -221,4 +225,19 @@ protected Schema getTableSchema(Connection connection, String database,
   protected String generateSessionID() {
     return UUID.randomUUID().toString().replace('-', '_');
   }
+
+  /**
+   * Returns whether auto-commit should be enabled for this connector.
+   * By default, it is false.
+   */
+  protected boolean isAutoCommitEnabled() {
+    return false;
+  }
+  /**
+   * Returns the default transaction isolation level for this connector.
+   * If null, it falls back to the database driver's default or serializable.
+   */
+  protected String getTransactionIsolationLevel() {
+    return null;
+  }
 }

databricks-plugin/docs/Databricks-batchsource.md

@@ -0,0 +1,15 @@
+# Databricks Batch Source
+
+Description
+-----------
+Reads data from a Databricks table using a configurable SQL query.
+
+Properties
+----------
+* **Use Connection**: Whether to use an existing Databricks connection.
+* **Host**: Server Hostname of the Databricks cluster or SQL warehouse.
+* **Port**: Database port (default is 443).
+* **HTTP Path**: The HTTP Path for the Databricks cluster or SQL warehouse.
+* **Reference Name**: Name used to identify this source for lineage.
+* **Database / Catalog**: Optional catalog or database name.
+* **Import Query**: SQL query to execute against Databricks.

databricks-plugin/docs/Databricks-connector.md

@@ -0,0 +1,15 @@
+# Databricks Database Connector
+
+Description
+-----------
+Connects to Databricks database / Lakehouse via JDBC.
+
+Properties
+----------
+* **Host**: Server Hostname of the Databricks cluster or SQL warehouse.
+* **Port**: Database port (default is 443).
+* **HTTP Path**: The HTTP Path for the Databricks cluster or SQL warehouse.
+* **Database / Catalog**: Optional catalog or database name to connect to.
+* **Username**: Username / token user.
+* **Password / Token**: Personal Access Token (PAT) or password.
+* **Connection Arguments**: Arbitrary key-value pairs to pass as connection arguments to the JDBC driver (e.g. `AuthMech=11;Auth_Flow=2`).

databricks-plugin/pom.xml

@@ -0,0 +1,127 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  Copyright © 2026 CDAP
+
+  Licensed under the Apache License, Version 2.0 (the "License"); you may not
+  use this file except in compliance with the License. You may obtain a copy of
+  the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+  License for the specific language governing permissions and limitations under
+  the License.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <parent>
+    <artifactId>database-plugins-parent</artifactId>
+    <groupId>io.cdap.plugin</groupId>
+    <version>1.13.0-SNAPSHOT</version>
+  </parent>
+
+  <name>Databricks plugin</name>
+  <artifactId>databricks-plugin</artifactId>
+  <modelVersion>4.0.0</modelVersion>
+
+  <properties>
+    <databricks-jdbc.version>3.4.1</databricks-jdbc.version>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>io.cdap.cdap</groupId>
+      <artifactId>cdap-etl-api</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.cdap.plugin</groupId>
+      <artifactId>database-commons</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>io.cdap.plugin</groupId>
+      <artifactId>hydrator-common</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.google.guava</groupId>
+      <artifactId>guava</artifactId>
+    </dependency>
+
+    <!-- test dependencies -->
+    <dependency>
+      <groupId>com.databricks</groupId>
+      <artifactId>databricks-jdbc</artifactId>
+      <version>${databricks-jdbc.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.cdap.plugin</groupId>
+      <artifactId>database-commons</artifactId>
+      <version>${project.version}</version>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.cdap.cdap</groupId>
+      <artifactId>hydrator-test</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.cdap.cdap</groupId>
+      <artifactId>cdap-data-pipeline3_2.12</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.cdap.cdap</groupId>
+      <artifactId>cdap-api</artifactId>
+      <scope>provided</scope>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>io.cdap</groupId>
+        <artifactId>cdap-maven-plugin</artifactId>
+      </plugin>
+      <plugin>
+        <groupId>org.apache.felix</groupId>
+        <artifactId>maven-bundle-plugin</artifactId>
+        <version>5.1.2</version>
+        <extensions>true</extensions>
+        <configuration>
+          <instructions>
+            <_exportcontents>
+              io.cdap.plugin.databricks.*;
+              io.cdap.plugin.db.source.*;
+              org.apache.commons.lang;
+              org.apache.commons.logging.*;
+              org.codehaus.jackson.*
+            </_exportcontents>
+            <Embed-Dependency>*;inline=false;scope=compile</Embed-Dependency>
+            <Embed-Transitive>true</Embed-Transitive>
+            <Embed-Directory>lib</Embed-Directory>
+          </instructions>
+        </configuration>
+        <executions>
+          <execution>
+            <phase>package</phase>
+            <goals>
+              <goal>bundle</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
+</project>

databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnector.java

@@ -0,0 +1,162 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.plugin.databricks;
+
+import io.cdap.cdap.api.annotation.Category;
+import io.cdap.cdap.api.annotation.Description;
+import io.cdap.cdap.api.annotation.Name;
+import io.cdap.cdap.api.annotation.Plugin;
+import io.cdap.cdap.api.data.format.StructuredRecord;
+import io.cdap.cdap.etl.api.batch.BatchSource;
+import io.cdap.cdap.etl.api.connector.Connector;
+import io.cdap.cdap.etl.api.connector.ConnectorSpec;
+import io.cdap.cdap.etl.api.connector.ConnectorSpecRequest;
+import io.cdap.cdap.etl.api.connector.PluginSpec;
+import io.cdap.plugin.common.Constants;
+import io.cdap.plugin.common.ReferenceNames;
+import io.cdap.plugin.common.db.DBConnectorPath;
+import io.cdap.plugin.db.NoOpCommitConnection;
+import io.cdap.plugin.db.SchemaReader;
+import io.cdap.plugin.db.TransactionIsolationLevel;
+import io.cdap.plugin.db.connector.AbstractDBSpecificConnector;
+import io.cdap.plugin.db.connector.DBSpecificPath;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.mapreduce.lib.db.DBWritable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Databricks Database Connector that connects to Databricks database via JDBC.
+ */
+@Plugin(type = Connector.PLUGIN_TYPE)
+@Name(DatabricksConstants.PLUGIN_NAME)
+@Description("Connection to access data in Databricks using JDBC.")
+@Category("Database")
+public class DatabricksConnector extends AbstractDBSpecificConnector<DatabricksDBRecord> {
+  public static final String NAME = DatabricksConstants.PLUGIN_NAME;
+  private final DatabricksConnectorConfig config;
+
+  private static final Logger LOG = LoggerFactory.getLogger(DatabricksConnector.class);
+
+  public DatabricksConnector(DatabricksConnectorConfig config) {
+    super(config);
+    this.config = config;
+  }
+
+  @Override
+  protected DBConnectorPath getDBConnectorPath(String path) throws IOException {
+    return DBSpecificPath.of(path, supportSchema());
+  }
+
+  @Override
+  protected Connection getConnection(DBConnectorPath path) {
+    Connection connection = super.getConnection(path);
+    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);
+  }
+
+  @Override
+  protected Connection getConnection() {
+    Connection connection = super.getConnection();
+    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);
+  }
+
+  @Override
+  public boolean supportSchema() {
+    return true;
+  }
+
+  @Override
+  protected Class<? extends DBWritable> getDBRecordType() {
+    return DatabricksDBRecord.class;
+  }
+
+  @Override
+  public StructuredRecord transform(LongWritable longWritable, DatabricksDBRecord record) {
+    return record.getRecord();
+  }
+
+  @Override
+  protected SchemaReader getSchemaReader(String sessionID) {
+    return new DatabricksSchemaReader(sessionID);
+  }
+
+  @Override
+  protected String getTableName(String database, String schema, String table) {
+    if (database == null && schema == null) {
+      return String.format("`%s`", table);
+    }
+    if (database == null) {
+      return String.format("`%s`.`%s`", schema, table);
+    }
+    if (schema == null) {
+      return String.format("`%s`.`%s`", database, table);
+    }
+    return String.format("`%s`.`%s`.`%s`", database, schema, table);
+  }
+
+  @Override
+  protected String getRandomQuery(String tableName, int limit) {
+    return String.format("SELECT * FROM %s LIMIT %d", tableName, limit);
+  }
+
+  @Override
+  protected void setConnectorSpec(ConnectorSpecRequest request, DBConnectorPath path,
+                                  ConnectorSpec.Builder builder) {
+    Map<String, String> sourceProperties = new HashMap<>();
+    setConnectionProperties(sourceProperties, request);
+    builder.addRelatedPlugin(new PluginSpec(DatabricksConstants.PLUGIN_NAME,
+                                            BatchSource.PLUGIN_TYPE, sourceProperties));
+
+    String schema = path.getSchema();
+    sourceProperties.put(DatabricksSource.DatabricksSourceConfig.NUM_SPLITS, "1");
+    sourceProperties.put(DatabricksSource.DatabricksSourceConfig.FETCH_SIZE,
+                         DatabricksSource.DatabricksSourceConfig.DEFAULT_FETCH_SIZE);
+    String table = path.getTable();
+    if (table == null) {
+      return;
+    }
+    sourceProperties.put(DatabricksSource.DatabricksSourceConfig.IMPORT_QUERY,
+                         getTableQuery(path.getDatabase(), schema, table));
+    sourceProperties.put(Constants.Reference.REFERENCE_NAME, ReferenceNames.cleanseReferenceName(table));
+  }
+
+  @Override
+  protected boolean isAutoCommitEnabled() {
+    return true;
+  }
+
+  @Override
+  protected String getTransactionIsolationLevel() {
+    return TransactionIsolationLevel.Level.TRANSACTION_READ_UNCOMMITTED.name();
+  }
+}