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.
| } | ||
|
|
||
| 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?
| @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 ..
| 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.
3 participants
Addition of Databricks Plugin