Addition of Databricks Plugin - #680
vikasrathee-cs wants to merge 1 commit into
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a new Databricks plugin module, including a batch source, database connector, configuration classes, documentation, widgets, and unit tests. It also updates the database-commons module to support custom auto-commit and transaction isolation levels. The review feedback highlights several critical improvements: preventing potential NullPointerExceptions when the connection configuration or column names are null, handling unsupported transaction isolation levels gracefully instead of throwing RuntimeExceptions, and implementing actual random sampling in the getRandomQuery method using ORDER BY rand().
80cabb2 to
9a4d4fc
Compare
9a4d4fc to
e865f6d
Compare
| } | ||
|
|
||
| @Override | ||
| protected String getRandomQuery(String tableName, int limit) { |
There was a problem hiding this comment.
getRandomQuery() is only ever reached from AbstractDBSpecificConnector.getTableQuery(..., sampleType, ...), which is only invoked when the UI passes a sampleType.
The UI only offers a sample type that the connector declares via ConnectorSpec.Builder.addSupportedSampleType(...)
and DatabricksConnector.setConnectorSpec() never calls it (compare MysqlConnector, which calls .addSupportedSampleType(SampleType.RANDOM) and STRATIFIED).
So the override can never fire.
There was a problem hiding this comment.
Added SampleType.RANDOM and SampleType.STRATIFIED in setConnectorSpec() and updated getRandomQuery() and getStratifiedQuery() similar to MysqlConnector
| } | ||
|
|
||
| @Override | ||
| protected String getTableName(String database, String schema, String table) { |
There was a problem hiding this comment.
If the value has a backtick (`) in it, it will malform things going forward.
Please handle validation.
There was a problem hiding this comment.
Tested this on Databricks — Databricks does not allow creating catalog, schema, or table names with an unescaped backtick (`) and fails with [PARSE_SYNTAX_ERROR]. Also, getTableName() here is consistent with MysqlConnector and AbstractDBSpecificConnector, where the identifiers come directly from browsing the connection metadata.
| } | ||
|
|
||
| public static final String PLUGIN_NAME = "Databricks"; | ||
| public static final String DRIVER_CLASS_NAME = "com.databricks.client.jdbc.Driver"; |
There was a problem hiding this comment.
Is this used any where?
There was a problem hiding this comment.
Removed the unused DRIVER_CLASS_NAME constant from DatabricksConstants.
| @Test | ||
| public void testShouldIgnoreColumn() throws SQLException { | ||
| DatabricksSchemaReader schemaReader = new DatabricksSchemaReader("sessionID"); | ||
| Map<Integer, String> names = new java.util.HashMap<>(); |
There was a problem hiding this comment.
Please use import. Don't use FQNs unless necessary .
| typeNames.put(8, "ARRAY"); | ||
| typeNames.put(9, "MAP"); | ||
|
|
||
| ResultSetMetaData metadata = createMockMetadata(typeNames, java.util.Collections.emptyMap()); |
There was a problem hiding this comment.
Please use import. Don't use FQNs unless necessary .
|
|
||
| private ResultSetMetaData createMockMetadata(Map<Integer, String> columnTypeNames, | ||
| Map<Integer, String> columnNames) { | ||
| return (ResultSetMetaData) Proxy.newProxyInstance( |
There was a problem hiding this comment.
Why this approach ? is it not diable via Mockito ? Mockito.mock(ResultSetMetaData.class) with when(...).thenReturn(...) would be a lot more readable ..
There was a problem hiding this comment.
Updated DatabricksSchemaReaderTest to use Mockito.mock(ResultSetMetaData.class) with Mockito.when(...).thenReturn(...) instead of Proxy.newProxyInstance.
| if (typeName.equalsIgnoreCase("BIGINT")) { | ||
| return Schema.of(Schema.Type.LONG); | ||
| } | ||
| if (typeName.equalsIgnoreCase("TIMESTAMP") || typeName.equalsIgnoreCase("TIMESTAMP_NTZ") || |
There was a problem hiding this comment.
Databricks TIMESTAMP is timestamp-with-local-time-zone (an instant); TIMESTAMP_NTZ is the zone-less one. Mapping both to CDAP DATETIME drops the zone and will shift values based on the executor's default TZ. I think this should be TIMESTAMP → LogicalType.TIMESTAMP_MICROS and TIMESTAMP_NTZ → DATETIME. Also, TIMESTAMPTZ isn't a Databricks type name, and the BIGINT/DATE branches are redundant — CommonSchemaReader already maps those from the SQL type.
There was a problem hiding this comment.
Updated DatabricksSchemaReader so TIMESTAMP_NTZ maps to Schema.LogicalType.DATETIME, removed TIMESTAMPTZ and redundant SQL type branches (BIGINT, DATE, INT, SMALLINT, TINYINT, TIME, TIMESTAMP), and delegated standard JDBC types (including TIMESTAMP
|
Hi, I don't think you have pushed the changes yet . |
| request.getProperties().get("sampleType"), request.getProperties().get("strata"), sessionID); | ||
| DataDrivenETLDBInputFormat.setInput(connectionConfigAccessor.getConfiguration(), getDBRecordType(), | ||
| tableQuery, null, false); | ||
| tableQuery, null, isAutoCommitEnabled()); |
There was a problem hiding this comment.
getInputFormatProvider() runs for every DB connector in this repo — MySQL, Postgres, Oracle, SQL Server, Redshift — not just Databricks.
The defaulting looks correct to me (isAutoCommitEnabled() → false reproduces the old hard-coded false, and a null isolation level leaves TransactionIsolationLevel.CONF_KEY unset so getLevel(null) still yields SERIALIZABLE). But nothing asserts that, so a future change to either default would silently alter connection behaviour for every other plugin with no test failing.
Could you add coverage in database-commons for:
- the base hooks returning false / null;
- a non-overriding connector leaving AUTO_COMMIT_ENABLED and TransactionIsolationLevel.CONF_KEY exactly as before this change;
an overriding connector setting both.
database-commons/src/test/java/io/cdap/plugin/db/source/DataDrivenETLDBInputFormatTest.java is a reasonable place
| </dependency> | ||
|
|
||
| <!-- test dependencies --> | ||
| <dependency> |
There was a problem hiding this comment.
This dependency isn't referenced by any test in the PR — nothing loads the driver, so it's currently dead weight in the build.
That's really a symptom of the wider gap: there's no DatabricksPluginTestBase, DatabricksPluginTestSuite, DatabricksSourceTestRun, DatabricksFailedConnectionTest or DatabricksDBRecordUnitTest here. Every other plugin in the repo ships that set — see amazon-redshift-plugin/src/test/.../RedshiftPluginTestBase.java (218 lines) as the closest template. The two unit tests in this PR don't touch DatabricksSource, DatabricksDBRecord, or any connection path.
For a brand-new plugin I'd like at least a DatabricksPluginTestBase following the Redshift pattern, which would also give this dependency a purpose. Separately, could you attach evidence of a real end-to-end run (browse → get schema → sample → pipeline read) against a SQL warehouse, over a table containing TIMESTAMP, TIMESTAMP_NTZ, DATE, DECIMAL(38,10), ARRAY, MAP, STRUCT, VARIANT and a NULL-only column? That's the set most likely to break and none of it is covered today.
3 participants
Addition of Databricks Plugin