Skip to content

Commit 562ac79

Browse files
stephane-airbytestephane-airbyte
committed
fix compiler errors
1 parent c347f40 commit 562ac79

File tree

52 files changed

+3942
-2437
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

52 files changed

+3942
-2437
lines changed

‎airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties

+1-1
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
version=0.28.8
1+
version=0.28.9

‎airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/java/io/airbyte/cdk/integrations/destination/jdbc/typing_deduping/JdbcDestinationHandler.java

+19-18
Original file line numberDiff line numberDiff line change
@@ -96,23 +96,23 @@ protected DSLContext getDslContext() {
9696
}
9797

9898
private Optional<TableDefinition> findExistingTable(final StreamId id) throws Exception {
99-
return findExistingTable(jdbcDatabase, databaseName, id.finalNamespace(), id.finalName());
99+
return findExistingTable(jdbcDatabase, databaseName, id.getFinalNamespace(), id.getFinalName());
100100
}
101101

102102
private boolean isFinalTableEmpty(final StreamId id) throws Exception {
103103
return !jdbcDatabase.queryBoolean(
104104
getDslContext().select(
105105
field(exists(
106106
selectOne()
107-
.from(name(id.finalNamespace(), id.finalName()))
107+
.from(name(id.getFinalNamespace(), id.getFinalName()))
108108
.limit(1))))
109109
.getSQL(ParamType.INLINED));
110110
}
111111

112112
private InitialRawTableStatus getInitialRawTableState(final StreamId id) throws Exception {
113113
boolean tableExists = jdbcDatabase.executeMetadataQuery(dbmetadata -> {
114-
LOGGER.info("Retrieving table from Db metadata: {} {} {}", databaseName, id.rawNamespace(), id.rawName());
115-
try (final ResultSet table = dbmetadata.getTables(databaseName, id.rawNamespace(), id.rawName(), null)) {
114+
LOGGER.info("Retrieving table from Db metadata: {} {} {}", databaseName, id.getRawNamespace(), id.getRawName());
115+
try (final ResultSet table = dbmetadata.getTables(databaseName, id.getRawNamespace(), id.getRawName(), null)) {
116116
return table.next();
117117
} catch (SQLException e) {
118118
LOGGER.error("Failed to retrieve table info from metadata", e);
@@ -131,7 +131,7 @@ private InitialRawTableStatus getInitialRawTableState(final StreamId id) throws
131131
try (final Stream<Timestamp> timestampStream = jdbcDatabase.unsafeQuery(
132132
conn -> conn.prepareStatement(
133133
getDslContext().select(field("MIN(_airbyte_extracted_at)").as("min_timestamp"))
134-
.from(name(id.rawNamespace(), id.rawName()))
134+
.from(name(id.getRawNamespace(), id.getRawName()))
135135
.where(DSL.condition("_airbyte_loaded_at IS NULL"))
136136
.getSQL()),
137137
record -> record.getTimestamp("min_timestamp"))) {
@@ -150,7 +150,7 @@ record -> record.getTimestamp("min_timestamp"))) {
150150
try (final Stream<Timestamp> timestampStream = jdbcDatabase.unsafeQuery(
151151
conn -> conn.prepareStatement(
152152
getDslContext().select(field("MAX(_airbyte_extracted_at)").as("min_timestamp"))
153-
.from(name(id.rawNamespace(), id.rawName()))
153+
.from(name(id.getRawNamespace(), id.getRawName()))
154154
.getSQL()),
155155
record -> record.getTimestamp("min_timestamp"))) {
156156
// Filter for nonNull values in case the query returned NULL (i.e. no raw records at all).
@@ -161,7 +161,7 @@ record -> record.getTimestamp("min_timestamp"))) {
161161

162162
@Override
163163
public void execute(final Sql sql) throws Exception {
164-
final List<List<String>> transactions = sql.transactions();
164+
final List<List<String>> transactions = sql.transactions;
165165
final UUID queryId = UUID.randomUUID();
166166
for (final List<String> transaction : transactions) {
167167
final UUID transactionId = UUID.randomUUID();
@@ -255,20 +255,20 @@ private CompletionStage<DestinationInitialStatus<DestinationState>> retrieveStat
255255
final StreamConfig streamConfig) {
256256
return destinationStatesFuture.thenApply(destinationStates -> {
257257
try {
258-
final Optional<TableDefinition> finalTableDefinition = findExistingTable(streamConfig.id());
258+
final Optional<TableDefinition> finalTableDefinition = findExistingTable(streamConfig.getId());
259259
final boolean isSchemaMismatch;
260260
final boolean isFinalTableEmpty;
261261
if (finalTableDefinition.isPresent()) {
262262
isSchemaMismatch = !existingSchemaMatchesStreamConfig(streamConfig, finalTableDefinition.get());
263-
isFinalTableEmpty = isFinalTableEmpty(streamConfig.id());
263+
isFinalTableEmpty = isFinalTableEmpty(streamConfig.getId());
264264
} else {
265265
// If the final table doesn't exist, then by definition it doesn't have a schema mismatch and has no
266266
// records.
267267
isSchemaMismatch = false;
268268
isFinalTableEmpty = true;
269269
}
270-
final InitialRawTableStatus initialRawTableState = getInitialRawTableState(streamConfig.id());
271-
DestinationState destinationState = destinationStates.getOrDefault(streamConfig.id().asPair(), toDestinationState(Jsons.emptyObject()));
270+
final InitialRawTableStatus initialRawTableState = getInitialRawTableState(streamConfig.getId());
271+
DestinationState destinationState = destinationStates.getOrDefault(streamConfig.getId().asPair(), toDestinationState(Jsons.emptyObject()));
272272
return new DestinationInitialStatus<>(streamConfig, finalTableDefinition.isPresent(), initialRawTableState,
273273
isSchemaMismatch, isFinalTableEmpty, destinationState);
274274
} catch (Exception e) {
@@ -337,9 +337,9 @@ protected boolean existingSchemaMatchesStreamConfig(final StreamConfig stream, f
337337
// Missing AB meta columns from final table, we need them to do proper T+D so trigger soft-reset
338338
return false;
339339
}
340-
final LinkedHashMap<String, String> intendedColumns = stream.columns().entrySet().stream()
340+
final LinkedHashMap<String, String> intendedColumns = stream.getColumns().entrySet().stream()
341341
.collect(LinkedHashMap::new,
342-
(map, column) -> map.put(column.getKey().name(), toJdbcTypeName(column.getValue())),
342+
(map, column) -> map.put(column.getKey().getName(), toJdbcTypeName(column.getValue())),
343343
LinkedHashMap::putAll);
344344

345345
// Filter out Meta columns since they don't exist in stream config.
@@ -354,16 +354,16 @@ protected boolean existingSchemaMatchesStreamConfig(final StreamConfig stream, f
354354
}
355355

356356
@Override
357-
public void commitDestinationStates(final Map<StreamId, DestinationState> destinationStates) throws Exception {
357+
public void commitDestinationStates(final Map<StreamId, ? extends DestinationState> destinationStates) throws Exception {
358358
if (destinationStates.isEmpty()) {
359359
return;
360360
}
361361

362362
// Delete all state records where the stream name+namespace match one of our states
363363
String deleteStates = getDslContext().deleteFrom(table(quotedName(rawTableSchemaName, DESTINATION_STATE_TABLE_NAME)))
364364
.where(destinationStates.keySet().stream()
365-
.map(streamId -> field(quotedName(DESTINATION_STATE_TABLE_COLUMN_NAME)).eq(streamId.originalName())
366-
.and(field(quotedName(DESTINATION_STATE_TABLE_COLUMN_NAMESPACE)).eq(streamId.originalNamespace())))
365+
.map(streamId -> field(quotedName(DESTINATION_STATE_TABLE_COLUMN_NAME)).eq(streamId.getOriginalName())
366+
.and(field(quotedName(DESTINATION_STATE_TABLE_COLUMN_NAMESPACE)).eq(streamId.getOriginalNamespace())))
367367
.reduce(
368368
DSL.falseCondition(),
369369
Condition::or))
@@ -381,10 +381,11 @@ public void commitDestinationStates(final Map<StreamId, DestinationState> destin
381381
// and assume the destination can cast it appropriately.
382382
// Destination-specific timestamp syntax is weird and annoying.
383383
field(quotedName(DESTINATION_STATE_TABLE_COLUMN_UPDATED_AT), String.class));
384-
for (Map.Entry<StreamId, DestinationState> destinationState : destinationStates.entrySet()) {
384+
for (Map.Entry<StreamId, ? extends DestinationState> destinationState : destinationStates.entrySet()) {
385385
final StreamId streamId = destinationState.getKey();
386386
final String stateJson = Jsons.serialize(destinationState.getValue());
387-
insertStatesStep = insertStatesStep.values(streamId.originalName(), streamId.originalNamespace(), stateJson, OffsetDateTime.now().toString());
387+
insertStatesStep =
388+
insertStatesStep.values(streamId.getOriginalName(), streamId.getOriginalNamespace(), stateJson, OffsetDateTime.now().toString());
388389
}
389390
String insertStates = insertStatesStep.getSQL(ParamType.INLINED);
390391

‎airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/java/io/airbyte/cdk/integrations/destination/jdbc/typing_deduping/JdbcSqlGenerator.java

+26-26
Original file line numberDiff line numberDiff line change
@@ -195,7 +195,7 @@ List<Field<?>> buildFinalTableFields(final LinkedHashMap<ColumnId, AirbyteType>
195195
final List<Field<?>> fields =
196196
metaColumns.entrySet().stream().map(metaColumn -> field(quotedName(metaColumn.getKey()), metaColumn.getValue())).collect(toList());
197197
final List<Field<?>> dataFields =
198-
columns.entrySet().stream().map(column -> field(quotedName(column.getKey().name()), toDialectType(column.getValue()))).collect(
198+
columns.entrySet().stream().map(column -> field(quotedName(column.getKey().getName()), toDialectType(column.getValue()))).collect(
199199
toList());
200200
dataFields.addAll(fields);
201201
return dataFields;
@@ -258,16 +258,16 @@ public Sql createSchema(final String schema) {
258258
@Override
259259
public Sql createTable(final StreamConfig stream, final String suffix, final boolean force) {
260260
// TODO: Use Naming transformer to sanitize these strings with redshift restrictions.
261-
final String finalTableIdentifier = stream.id().finalName() + suffix.toLowerCase();
261+
final String finalTableIdentifier = stream.getId().getFinalName() + suffix.toLowerCase();
262262
if (!force) {
263263
return transactionally(Stream.concat(
264-
Stream.of(createTableSql(stream.id().finalNamespace(), finalTableIdentifier, stream.columns())),
264+
Stream.of(createTableSql(stream.getId().getFinalNamespace(), finalTableIdentifier, stream.getColumns())),
265265
createIndexSql(stream, suffix).stream()).toList());
266266
}
267267
return transactionally(Stream.concat(
268268
Stream.of(
269-
dropTableIfExists(quotedName(stream.id().finalNamespace(), finalTableIdentifier)).getSQL(ParamType.INLINED),
270-
createTableSql(stream.id().finalNamespace(), finalTableIdentifier, stream.columns())),
269+
dropTableIfExists(quotedName(stream.getId().getFinalNamespace(), finalTableIdentifier)).getSQL(ParamType.INLINED),
270+
createTableSql(stream.getId().getFinalNamespace(), finalTableIdentifier, stream.getColumns())),
271271
createIndexSql(stream, suffix).stream()).toList());
272272
}
273273

@@ -285,18 +285,18 @@ public Sql updateTable(final StreamConfig streamConfig,
285285
@Override
286286
public Sql overwriteFinalTable(final StreamId stream, final String finalSuffix) {
287287
return transactionally(
288-
dropTableIfExists(name(stream.finalNamespace(), stream.finalName())).getSQL(ParamType.INLINED),
289-
alterTable(name(stream.finalNamespace(), stream.finalName() + finalSuffix))
290-
.renameTo(name(stream.finalName()))
288+
dropTableIfExists(name(stream.getFinalNamespace(), stream.getFinalName())).getSQL(ParamType.INLINED),
289+
alterTable(name(stream.getFinalNamespace(), stream.getFinalName() + finalSuffix))
290+
.renameTo(name(stream.getFinalName()))
291291
.getSQL());
292292
}
293293

294294
@Override
295295
public Sql migrateFromV1toV2(final StreamId streamId, final String namespace, final String tableName) {
296-
final Name rawTableName = name(streamId.rawNamespace(), streamId.rawName());
296+
final Name rawTableName = name(streamId.getRawNamespace(), streamId.getRawName());
297297
final DSLContext dsl = getDslContext();
298298
return transactionally(
299-
dsl.createSchemaIfNotExists(streamId.rawNamespace()).getSQL(),
299+
dsl.createSchemaIfNotExists(streamId.getRawNamespace()).getSQL(),
300300
dsl.dropTableIfExists(rawTableName).getSQL(),
301301
DSL.createTable(rawTableName)
302302
.column(COLUMN_NAME_AB_RAW_ID, SQLDataType.VARCHAR(36).nullable(false))
@@ -315,7 +315,7 @@ public Sql migrateFromV1toV2(final StreamId streamId, final String namespace, fi
315315

316316
@Override
317317
public Sql clearLoadedAt(final StreamId streamId) {
318-
return Sql.of(update(table(name(streamId.rawNamespace(), streamId.rawName())))
318+
return Sql.of(update(table(name(streamId.getRawNamespace(), streamId.getRawName())))
319319
.set(field(COLUMN_NAME_AB_LOADED_AT), inline((String) null))
320320
.getSQL());
321321
}
@@ -350,28 +350,28 @@ private Sql insertAndDeleteTransaction(final StreamConfig streamConfig,
350350
final String finalSuffix,
351351
final Optional<Instant> minRawTimestamp,
352352
final boolean useExpensiveSaferCasting) {
353-
final String finalSchema = streamConfig.id().finalNamespace();
354-
final String finalTable = streamConfig.id().finalName() + (finalSuffix != null ? finalSuffix.toLowerCase() : "");
355-
final String rawSchema = streamConfig.id().rawNamespace();
356-
final String rawTable = streamConfig.id().rawName();
353+
final String finalSchema = streamConfig.getId().getFinalNamespace();
354+
final String finalTable = streamConfig.getId().getFinalName() + (finalSuffix != null ? finalSuffix.toLowerCase() : "");
355+
final String rawSchema = streamConfig.getId().getRawNamespace();
356+
final String rawTable = streamConfig.getId().getRawName();
357357

358358
// Poor person's guarantee of ordering of fields by using same source of ordered list of columns to
359359
// generate fields.
360360
final CommonTableExpression<Record> rawTableRowsWithCast = name(TYPING_CTE_ALIAS).as(
361-
selectFromRawTable(rawSchema, rawTable, streamConfig.columns(),
361+
selectFromRawTable(rawSchema, rawTable, streamConfig.getColumns(),
362362
getFinalTableMetaColumns(false),
363-
rawTableCondition(streamConfig.destinationSyncMode(),
364-
streamConfig.columns().containsKey(cdcDeletedAtColumn),
363+
rawTableCondition(streamConfig.getDestinationSyncMode(),
364+
streamConfig.getColumns().containsKey(cdcDeletedAtColumn),
365365
minRawTimestamp),
366366
useExpensiveSaferCasting));
367-
final List<Field<?>> finalTableFields = buildFinalTableFields(streamConfig.columns(), getFinalTableMetaColumns(true));
368-
final Field<Integer> rowNumber = getRowNumber(streamConfig.primaryKey(), streamConfig.cursor());
367+
final List<Field<?>> finalTableFields = buildFinalTableFields(streamConfig.getColumns(), getFinalTableMetaColumns(true));
368+
final Field<Integer> rowNumber = getRowNumber(streamConfig.getPrimaryKey(), streamConfig.getCursor());
369369
final CommonTableExpression<Record> filteredRows = name(NUMBERED_ROWS_CTE_ALIAS).as(
370370
select(asterisk(), rowNumber).from(rawTableRowsWithCast));
371371

372372
// Used for append-dedupe mode.
373373
final String insertStmtWithDedupe =
374-
insertIntoFinalTable(finalSchema, finalTable, streamConfig.columns(), getFinalTableMetaColumns(true))
374+
insertIntoFinalTable(finalSchema, finalTable, streamConfig.getColumns(), getFinalTableMetaColumns(true))
375375
.select(with(rawTableRowsWithCast)
376376
.with(filteredRows)
377377
.select(finalTableFields)
@@ -383,17 +383,17 @@ private Sql insertAndDeleteTransaction(final StreamConfig streamConfig,
383383

384384
// Used for append and overwrite modes.
385385
final String insertStmt =
386-
insertIntoFinalTable(finalSchema, finalTable, streamConfig.columns(), getFinalTableMetaColumns(true))
386+
insertIntoFinalTable(finalSchema, finalTable, streamConfig.getColumns(), getFinalTableMetaColumns(true))
387387
.select(with(rawTableRowsWithCast)
388388
.select(finalTableFields)
389389
.from(rawTableRowsWithCast))
390390
.getSQL(ParamType.INLINED);
391-
final String deleteStmt = deleteFromFinalTable(finalSchema, finalTable, streamConfig.primaryKey(), streamConfig.cursor());
391+
final String deleteStmt = deleteFromFinalTable(finalSchema, finalTable, streamConfig.getPrimaryKey(), streamConfig.getCursor());
392392
final String deleteCdcDeletesStmt =
393-
streamConfig.columns().containsKey(cdcDeletedAtColumn) ? deleteFromFinalTableCdcDeletes(finalSchema, finalTable) : "";
393+
streamConfig.getColumns().containsKey(cdcDeletedAtColumn) ? deleteFromFinalTableCdcDeletes(finalSchema, finalTable) : "";
394394
final String checkpointStmt = checkpointRawTable(rawSchema, rawTable, minRawTimestamp);
395395

396-
if (streamConfig.destinationSyncMode() != DestinationSyncMode.APPEND_DEDUP) {
396+
if (streamConfig.getDestinationSyncMode() != DestinationSyncMode.APPEND_DEDUP) {
397397
return transactionally(
398398
insertStmt,
399399
checkpointStmt);
@@ -470,7 +470,7 @@ private String deleteFromFinalTable(final String schemaName,
470470
private String deleteFromFinalTableCdcDeletes(final String schema, final String tableName) {
471471
final DSLContext dsl = getDslContext();
472472
return dsl.deleteFrom(table(quotedName(schema, tableName)))
473-
.where(field(quotedName(cdcDeletedAtColumn.name())).isNotNull())
473+
.where(field(quotedName(cdcDeletedAtColumn.getName())).isNotNull())
474474
.getSQL(ParamType.INLINED);
475475
}
476476

‎airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/java/io/airbyte/cdk/integrations/destination/jdbc/typing_deduping/JdbcV1V2Migrator.java

+7-7
Original file line numberDiff line numberDiff line change
@@ -35,9 +35,9 @@ public JdbcV1V2Migrator(final NamingConventionTransformer namingConventionTransf
3535

3636
@SneakyThrows
3737
@Override
38-
protected boolean doesAirbyteInternalNamespaceExist(final StreamConfig streamConfig) {
38+
public boolean doesAirbyteInternalNamespaceExist(final StreamConfig streamConfig) {
3939
final String retrievedSchema = database.executeMetadataQuery(dbMetadata -> {
40-
try (ResultSet columns = dbMetadata.getSchemas(databaseName, streamConfig.id().rawNamespace())) {
40+
try (ResultSet columns = dbMetadata.getSchemas(databaseName, streamConfig.getId().getRawNamespace())) {
4141
String schema = "";
4242
while (columns.next()) {
4343
// Catalog can be null, so don't do anything with it.
@@ -54,22 +54,22 @@ protected boolean doesAirbyteInternalNamespaceExist(final StreamConfig streamCon
5454
}
5555

5656
@Override
57-
protected boolean schemaMatchesExpectation(final TableDefinition existingTable, final Collection<String> columns) {
57+
public boolean schemaMatchesExpectation(final TableDefinition existingTable, final Collection<String> columns) {
5858
return existingTable.columns().keySet().containsAll(columns);
5959
}
6060

6161
@SneakyThrows
6262
@Override
63-
protected Optional<TableDefinition> getTableIfExists(final String namespace, final String tableName) throws Exception {
63+
public Optional<TableDefinition> getTableIfExists(final String namespace, final String tableName) throws Exception {
6464
return JdbcDestinationHandler.findExistingTable(database, databaseName, namespace, tableName);
6565
}
6666

6767
@Override
68-
protected NamespacedTableName convertToV1RawName(final StreamConfig streamConfig) {
68+
public NamespacedTableName convertToV1RawName(final StreamConfig streamConfig) {
6969
@SuppressWarnings("deprecation")
70-
final String tableName = this.namingConventionTransformer.getRawTableName(streamConfig.id().originalName());
70+
final String tableName = this.namingConventionTransformer.getRawTableName(streamConfig.getId().getOriginalName());
7171
return new NamespacedTableName(
72-
this.namingConventionTransformer.getIdentifier(streamConfig.id().originalNamespace()),
72+
this.namingConventionTransformer.getIdentifier(streamConfig.getId().getOriginalNamespace()),
7373
tableName);
7474
}
7575

‎airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/java/io/airbyte/cdk/integrations/destination/staging/SerialStagingConsumerFactory.java

+3-3
Original file line numberDiff line numberDiff line change
@@ -111,9 +111,9 @@ private static Function<ConfiguredAirbyteStream, WriteConfig> toWriteConfig(fina
111111
final String outputSchema;
112112
final String tableName;
113113
if (useDestinationsV2Columns) {
114-
final StreamId streamId = parsedCatalog.getStream(abStream.getNamespace(), streamName).id();
115-
outputSchema = streamId.rawNamespace();
116-
tableName = streamId.rawName();
114+
final StreamId streamId = parsedCatalog.getStream(abStream.getNamespace(), streamName).getId();
115+
outputSchema = streamId.getRawNamespace();
116+
tableName = streamId.getRawName();
117117
} else {
118118
outputSchema = getOutputSchema(abStream, config.get("schema").asText(), namingResolver);
119119
tableName = namingResolver.getRawTableName(streamName);

0 commit comments

Comments
 (0)