Code cleanup in abstract classes (#18811)

* Refactor attempt 1 Increasing readability: - Performed audit of methods that are marked as public/protected but should be private - Performed audit of methods that are marked as public but should be protected - Removing @OverRide from methods that don't need them (are the very first) * Remove .sqlite files * Remove AbstractRelationDbSource class - Removing AbstractRelationalDbSource abstract class and moving contents of it to a Util class * Rename utility class * Add overrides back

Author: akashkulk

airbyte-integrations/connectors/source-bigquery/src/main/java/io/airbyte/integrations/source/bigquery/BigQuerySource.java

@@ -4,6 +4,10 @@
 
 package io.airbyte.integrations.source.bigquery;
 
+import static io.airbyte.integrations.source.relationaldb.RelationalDbQueryUtils.enquoteIdentifierList;
+import static io.airbyte.integrations.source.relationaldb.RelationalDbQueryUtils.getFullTableName;
+import static io.airbyte.integrations.source.relationaldb.RelationalDbQueryUtils.queryTable;
+
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.cloud.bigquery.QueryParameterValue;
 import com.google.cloud.bigquery.StandardSQLTypeName;
@@ -18,8 +22,9 @@
 import io.airbyte.db.bigquery.BigQuerySourceOperations;
 import io.airbyte.integrations.base.IntegrationRunner;
 import io.airbyte.integrations.base.Source;
-import io.airbyte.integrations.source.relationaldb.AbstractRelationalDbSource;
+import io.airbyte.integrations.source.relationaldb.AbstractDbSource;
 import io.airbyte.integrations.source.relationaldb.CursorInfo;
+import io.airbyte.integrations.source.relationaldb.RelationalDbQueryUtils;
 import io.airbyte.integrations.source.relationaldb.TableInfo;
 import io.airbyte.protocol.models.CommonField;
 import io.airbyte.protocol.models.JsonSchemaType;
@@ -34,7 +39,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class BigQuerySource extends AbstractRelationalDbSource<StandardSQLTypeName, BigQueryDatabase> implements Source {
+public class BigQuerySource extends AbstractDbSource<StandardSQLTypeName, BigQueryDatabase> implements Source {
 
  private static final Logger LOGGER = LoggerFactory.getLogger(BigQuerySource.class);
  private static final String QUOTE = "`";
@@ -76,7 +81,7 @@ public List<CheckedConsumer<BigQueryDatabase, Exception>> getCheckOperations(fin
  checkList.add(database -> {
  if (isDatasetConfigured(database)) {
  database.query(String.format("select 1 from %s where 1=0",
- getFullTableName(getConfigDatasetId(database), "INFORMATION_SCHEMA.TABLES")));
+ getFullTableName(getConfigDatasetId(database), "INFORMATION_SCHEMA.TABLES", getQuoteString())));
  LOGGER.info("The source passed the Dataset query test!");
  } else {
  LOGGER.info("The Dataset query test is skipped due to not configured datasetId!");
@@ -140,12 +145,23 @@ public AutoCloseableIterator<JsonNode> queryTableIncremental(final BigQueryDatab
  final CursorInfo cursorInfo,
  final StandardSQLTypeName cursorFieldType) {
  return queryTableWithParams(database, String.format("SELECT %s FROM %s WHERE %s > ?",
- enquoteIdentifierList(columnNames),
- getFullTableName(schemaName, tableName),
+ RelationalDbQueryUtils.enquoteIdentifierList(columnNames, getQuoteString()),
+ getFullTableName(schemaName, tableName, getQuoteString()),
  cursorInfo.getCursorField()),
  sourceOperations.getQueryParameter(cursorFieldType, cursorInfo.getCursor()));
  }
 
+ @Override
+ protected AutoCloseableIterator<JsonNode> queryTableFullRefresh(final BigQueryDatabase database,
+ final List<String> columnNames,
+ final String schemaName,
+ final String tableName) {
+ LOGGER.info("Queueing query for table: {}", tableName);
+ return queryTable(database, String.format("SELECT %s FROM %s",
+ enquoteIdentifierList(columnNames, getQuoteString()),
+ getFullTableName(schemaName, tableName, getQuoteString())));
+ }
+
  @Override
  public boolean isCursorType(final StandardSQLTypeName standardSQLTypeName) {
  return true;

airbyte-integrations/connectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java

@@ -20,6 +20,9 @@
 import static io.airbyte.db.jdbc.JdbcConstants.JDBC_COLUMN_TYPE_NAME;
 import static io.airbyte.db.jdbc.JdbcConstants.JDBC_IS_NULLABLE;
 import static io.airbyte.db.jdbc.JdbcUtils.EQUALS;
+import static io.airbyte.integrations.source.relationaldb.RelationalDbQueryUtils.enquoteIdentifierList;
+import static io.airbyte.integrations.source.relationaldb.RelationalDbQueryUtils.getFullTableName;
+import static io.airbyte.integrations.source.relationaldb.RelationalDbQueryUtils.queryTable;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.collect.ImmutableList;
@@ -40,7 +43,7 @@
 import io.airbyte.integrations.base.AirbyteStreamNameNamespacePair;
 import io.airbyte.integrations.base.Source;
 import io.airbyte.integrations.source.jdbc.dto.JdbcPrivilegeDto;
-import io.airbyte.integrations.source.relationaldb.AbstractRelationalDbSource;
+import io.airbyte.integrations.source.relationaldb.AbstractDbSource;
 import io.airbyte.integrations.source.relationaldb.CursorInfo;
 import io.airbyte.integrations.source.relationaldb.TableInfo;
 import io.airbyte.integrations.source.relationaldb.state.StateManager;
@@ -81,7 +84,7 @@
  * relational DB source which can be accessed via JDBC driver. If you are implementing a connector
  * for a relational DB which has a JDBC driver, make an effort to use this class.
  */
-public abstract class AbstractJdbcSource<Datatype> extends AbstractRelationalDbSource<Datatype, JdbcDatabase> implements Source {
+public abstract class AbstractJdbcSource<Datatype> extends AbstractDbSource<Datatype, JdbcDatabase> implements Source {
 
  public static final String SSL_MODE = "sslMode";
 
@@ -136,12 +139,23 @@ public AbstractJdbcSource(final String driverClass,
  this.sourceOperations = sourceOperations;
  }
 
+ @Override
+ protected AutoCloseableIterator<JsonNode> queryTableFullRefresh(final JdbcDatabase database,
+ final List<String> columnNames,
+ final String schemaName,
+ final String tableName) {
+ LOGGER.info("Queueing query for table: {}", tableName);
+ return queryTable(database, String.format("SELECT %s FROM %s",
+ enquoteIdentifierList(columnNames, getQuoteString()),
+ getFullTableName(schemaName, tableName, getQuoteString())));
+ }
+
  /**
  * Configures a list of operations that can be used to check the connection to the source.
  *
  * @return list of consumers that run queries for the check command.
  */
- public List<CheckedConsumer<JdbcDatabase, Exception>> getCheckOperations(final JsonNode config) throws Exception {
+ protected List<CheckedConsumer<JdbcDatabase, Exception>> getCheckOperations(final JsonNode config) throws Exception {
  return ImmutableList.of(database -> {
  LOGGER.info("Attempting to get metadata from the database to see if we can connect.");
  database.bufferedResultSetQuery(connection -> connection.getMetaData().getCatalogs(), sourceOperations::rowToJson);
@@ -258,7 +272,7 @@ private JsonNode getColumnMetadata(final ResultSet resultSet) throws SQLExceptio
  * @param field Essential column information returned from
  * {@link AbstractJdbcSource#getColumnMetadata}.
  */
- public Datatype getFieldType(final JsonNode field) {
+ private Datatype getFieldType(final JsonNode field) {
  return sourceOperations.getFieldType(field);
  }
 
@@ -515,7 +529,7 @@ public void close() {
  * @param config configuration
  * @return map containing relevant parsed values including location of keystore or an empty map
  */
- public Map<String, String> parseSSLConfig(final JsonNode config) {
+ protected Map<String, String> parseSSLConfig(final JsonNode config) {
  LOGGER.debug("source config: {}", config);
 
  final Map<String, String> additionalParameters = new HashMap<>();
@@ -572,7 +586,7 @@ public Map<String, String> parseSSLConfig(final JsonNode config) {
  * @param sslParams
  * @return SSL portion of JDBC question params or and empty string
  */
- public String toJDBCQueryParams(final Map<String, String> sslParams) {
+ protected String toJDBCQueryParams(final Map<String, String> sslParams) {
  return Objects.isNull(sslParams) ? ""
  : sslParams.entrySet()
  .stream()

airbyte-integrations/connectors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlSource.java

@@ -7,6 +7,10 @@
 import static io.airbyte.integrations.debezium.AirbyteDebeziumHandler.shouldUseCDC;
 import static io.airbyte.integrations.debezium.internals.DebeziumEventUtils.CDC_DELETED_AT;
 import static io.airbyte.integrations.debezium.internals.DebeziumEventUtils.CDC_UPDATED_AT;
+import static io.airbyte.integrations.source.relationaldb.RelationalDbQueryUtils.enquoteIdentifierList;
+import static io.airbyte.integrations.source.relationaldb.RelationalDbQueryUtils.getFullTableName;
+import static io.airbyte.integrations.source.relationaldb.RelationalDbQueryUtils.getIdentifierWithQuoting;
+import static io.airbyte.integrations.source.relationaldb.RelationalDbQueryUtils.queryTable;
 import static java.util.stream.Collectors.toList;
 
 import com.fasterxml.jackson.databind.JsonNode;
@@ -80,7 +84,7 @@ public AutoCloseableIterator<JsonNode> queryTableFullRefresh(final JdbcDatabase
  LOGGER.info("Queueing query for table: {}", tableName);
 
  final String newIdentifiers = getWrappedColumnNames(database, null, columnNames, schemaName, tableName);
- final String preparedSqlQuery = String.format("SELECT %s FROM %s", newIdentifiers, getFullTableName(schemaName, tableName));
+ final String preparedSqlQuery = String.format("SELECT %s FROM %s", newIdentifiers, getFullTableName(schemaName, tableName, getQuoteString()));
 
  LOGGER.info("Prepared SQL query for TableFullRefresh is: " + preparedSqlQuery);
  return queryTable(database, preparedSqlQuery);
@@ -107,8 +111,8 @@ protected String getWrappedColumnNames(final JdbcDatabase database,
  final SQLServerResultSetMetaData sqlServerResultSetMetaData = (SQLServerResultSetMetaData) database
  .queryMetadata(String
  .format("SELECT TOP 1 %s FROM %s", // only first row is enough to get field's type
- enquoteIdentifierList(columnNames),
- getFullTableName(schemaName, tableName)));
+ enquoteIdentifierList(columnNames, getQuoteString()),
+ getFullTableName(schemaName, tableName, getQuoteString())));
 
  // metadata will be null if table doesn't contain records
  if (sqlServerResultSetMetaData != null) {
@@ -127,7 +131,7 @@ protected String getWrappedColumnNames(final JdbcDatabase database,
  .map(
  el -> hierarchyIdColumns.contains(el) ? String
  .format("%s.ToString() as %s%s%s", el, identifierQuoteString, el, identifierQuoteString)
- : getIdentifierWithQuoting(el))
+ : getIdentifierWithQuoting(el, getQuoteString()))
  .toList());
  } catch (final SQLException e) {
  LOGGER.error("Failed to fetch metadata to prepare a proper request.", e);

airbyte-integrations/connectors/source-relational-db/src/main/java/io/airbyte/integrations/source/relationaldb/AbstractDbSource.java

@@ -7,6 +7,7 @@
 import static io.airbyte.integrations.base.errors.messages.ErrorMessage.getErrorMessage;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 import io.airbyte.commons.exceptions.ConnectionErrorException;
@@ -234,7 +235,7 @@ private void validateCursorFieldForIncrementalTables(
  }
  }
 
- protected List<TableInfo<CommonField<DataType>>> discoverWithoutSystemTables(
+ private List<TableInfo<CommonField<DataType>>> discoverWithoutSystemTables(
  final Database database) throws Exception {
  final Set<String> systemNameSpaces = getExcludedInternalNameSpaces();
  final List<TableInfo<CommonField<DataType>>> discoveredTables = discoverInternal(database);
@@ -244,7 +245,7 @@ protected List<TableInfo<CommonField<DataType>>> discoverWithoutSystemTables(
  Collectors.toList()));
  }
 
- protected List<AutoCloseableIterator<AirbyteMessage>> getFullRefreshIterators(
+ private List<AutoCloseableIterator<AirbyteMessage>> getFullRefreshIterators(
  final Database database,
  final ConfiguredAirbyteCatalog catalog,
  final Map<String, TableInfo<CommonField<DataType>>> tableNameToTable,
@@ -287,7 +288,7 @@ protected List<AutoCloseableIterator<AirbyteMessage>> getIncrementalIterators(
  * sync mode
  * @return List of AirbyteMessageIterators containing all iterators for a catalog
  */
- protected List<AutoCloseableIterator<AirbyteMessage>> getSelectedIterators(
+ private List<AutoCloseableIterator<AirbyteMessage>> getSelectedIterators(
  final Database database,
  final ConfiguredAirbyteCatalog catalog,
  final Map<String, TableInfo<CommonField<DataType>>> tableNameToTable,
@@ -331,7 +332,7 @@ protected List<AutoCloseableIterator<AirbyteMessage>> getSelectedIterators(
  * @param emittedAt Time when data was emitted from the Source database
  * @return
  */
- protected AutoCloseableIterator<AirbyteMessage> createReadIterator(final Database database,
+ private AutoCloseableIterator<AirbyteMessage> createReadIterator(final Database database,
  final ConfiguredAirbyteStream airbyteStream,
  final TableInfo<CommonField<DataType>> table,
  final StateManager stateManager,
@@ -414,7 +415,7 @@ protected AutoCloseableIterator<AirbyteMessage> createReadIterator(final Databas
  * @param emittedAt Time when data was emitted from the Source database
  * @return AirbyteMessage Iterator that
  */
- protected AutoCloseableIterator<AirbyteMessage> getIncrementalStream(final Database database,
+ private AutoCloseableIterator<AirbyteMessage> getIncrementalStream(final Database database,
  final ConfiguredAirbyteStream airbyteStream,
  final List<String> selectedDatabaseFields,
  final TableInfo<CommonField<DataType>> table,
@@ -456,7 +457,7 @@ protected AutoCloseableIterator<AirbyteMessage> getIncrementalStream(final Datab
  * @param emittedAt Time when data was emitted from the Source database
  * @return AirbyteMessageIterator with all records for a database source
  */
- protected AutoCloseableIterator<AirbyteMessage> getFullRefreshStream(final Database database,
+ private AutoCloseableIterator<AirbyteMessage> getFullRefreshStream(final Database database,
  final String streamName,
  final String namespace,
  final List<String> selectedDatabaseFields,
@@ -468,11 +469,11 @@ protected AutoCloseableIterator<AirbyteMessage> getFullRefreshStream(final Datab
  return getMessageIterator(queryStream, streamName, namespace, emittedAt.toEpochMilli());
  }
 
- protected String getFullyQualifiedTableName(final String nameSpace, final String tableName) {
+ private String getFullyQualifiedTableName(final String nameSpace, final String tableName) {
  return nameSpace != null ? nameSpace + "." + tableName : tableName;
  }
 
- public AutoCloseableIterator<AirbyteMessage> getMessageIterator(
+ private AutoCloseableIterator<AirbyteMessage> getMessageIterator(
  final AutoCloseableIterator<JsonNode> recordIterator,
  final String streamName,
  final String namespace,
@@ -493,7 +494,7 @@ public AutoCloseableIterator<AirbyteMessage> getMessageIterator(
  * @return list of table/data structure info
  * @throws Exception might throw an error during connection to database
  */
- protected List<TableInfo<Field>> getTables(final Database database) throws Exception {
+ private List<TableInfo<Field>> getTables(final Database database) throws Exception {
  final List<TableInfo<CommonField<DataType>>> tableInfos = discoverWithoutSystemTables(database);
  final Map<String, List<String>> fullyQualifiedTableNameToPrimaryKeys = discoverPrimaryKeys(
  database, tableInfos);
@@ -522,7 +523,7 @@ protected List<TableInfo<Field>> getTables(final Database database) throws Excep
  .collect(Collectors.toList());
  }
 
- protected Field toField(final CommonField<DataType> field) {
+ private Field toField(final CommonField<DataType> field) {
  if (getType(field.getType()) == JsonSchemaType.OBJECT && field.getProperties() != null
  && !field.getProperties().isEmpty()) {
  final var properties = field.getProperties().stream().map(this::toField).toList();
@@ -532,7 +533,7 @@ protected Field toField(final CommonField<DataType> field) {
  }
  }
 
- protected void assertColumnsWithSameNameAreSame(final String nameSpace, final String tableName,
+ private void assertColumnsWithSameNameAreSame(final String nameSpace, final String tableName,
  final List<CommonField<DataType>> columns) {
  columns.stream()
  .collect(Collectors.groupingBy(CommonField<DataType>::getName))
@@ -557,7 +558,7 @@ protected void assertColumnsWithSameNameAreSame(final String nameSpace, final St
  * for SELECT-ing the table with privileges. In some cases such SELECT doesn't require (e.g. in
  * Oracle DB - the schema is the user, you cannot REVOKE a privilege on a table from its owner).
  */
- public <T> Set<T> getPrivilegesTableForCurrentUser(final JdbcDatabase database,
+ protected <T> Set<T> getPrivilegesTableForCurrentUser(final JdbcDatabase database,
  final String schema) throws SQLException {
  return Collections.emptySet();
  }
@@ -585,7 +586,7 @@ public <T> Set<T> getPrivilegesTableForCurrentUser(final JdbcDatabase database,
  *
  * @return list of consumers that run queries for the check command.
  */
- public abstract List<CheckedConsumer<Database, Exception>> getCheckOperations(JsonNode config)
+ protected abstract List<CheckedConsumer<Database, Exception>> getCheckOperations(JsonNode config)
  throws Exception;
 
  /**
@@ -601,7 +602,7 @@ public abstract List<CheckedConsumer<Database, Exception>> getCheckOperations(Js
  *
  * @return set of system namespaces(schemas) to be excluded
  */
- public abstract Set<String> getExcludedInternalNameSpaces();
+ protected abstract Set<String> getExcludedInternalNameSpaces();
 
  /**
  * Discover all available tables in the source database.
@@ -653,7 +654,7 @@ protected abstract Map<String, List<String>> discoverPrimaryKeys(Database databa
  * @param tableName target table
  * @return iterator with read data
  */
- public abstract AutoCloseableIterator<JsonNode> queryTableFullRefresh(final Database database,
+ protected abstract AutoCloseableIterator<JsonNode> queryTableFullRefresh(final Database database,
  final List<String> columnNames,
  final String schemaName,
  final String tableName);
@@ -666,7 +667,7 @@ public abstract AutoCloseableIterator<JsonNode> queryTableFullRefresh(final Data
  *
  * @return iterator with read data
  */
- public abstract AutoCloseableIterator<JsonNode> queryTableIncremental(Database database,
+ protected abstract AutoCloseableIterator<JsonNode> queryTableIncremental(Database database,
  List<String> columnNames,
  String schemaName,
  String tableName,
@@ -685,7 +686,7 @@ protected int getStateEmissionFrequency() {
  /**
  * @return list of fields that could be used as cursors
  */
- public abstract boolean isCursorType(DataType type);
+ protected abstract boolean isCursorType(DataType type);
 
  private Database createDatabaseInternal(final JsonNode sourceConfig) throws Exception {
  final Database database = createDatabase(sourceConfig);