Skip to content

[WIP] Guarantee temp table cleanup and fix its ordering vs schema update #111

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
chikamura wants to merge 2 commits into from
+196 −133
Closed

[WIP] Guarantee temp table cleanup and fix its ordering vs schema update #111

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a82c9f9
Guarantee temp table cleanup and fix its ordering vs schema update
chikamura Sep 8, 2026
412a48b
Test that the temp table is deleted even when paths is empty
chikamura Sep 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
184 changes: 91 additions & 93 deletions src/main/java/org/embulk/output/bigquery_java/BigqueryJavaOutputPlugin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,114 +50,112 @@ public ConfigDiff transaction(
BigqueryTaskBuilder.build(task);
BigqueryClient client = new BigqueryClient(task, schema);
autoCreate(task, client);
client.storeCachedSrcFieldsIfNeed();

control.run(task.dump());
this.writers.values().forEach(BigqueryFileWriter::close);
logger.info("embulk-output-bigquery: finish to create intermediate files");

try {
paths = BigqueryUtil.getIntermediateFiles(task);
} catch (Exception e) {
logger.info(e.getMessage());
throw new RuntimeException(e);
}
if (paths.isEmpty()) {
logger.info("embulk-output-bigquery: Nothing for transfer");
client.createTableIfNotExist(task.getTable());

switch (task.getMode()) {
case "merge":
case "append":
case "replace":
case "delete_in_advance":
if (task.getTempTable().isPresent()) {
client.deleteTable(task.getTempTable().get());
}
break;
}

return CONFIG_MAPPER_FACTORY.newConfigDiff();
}

logger.debug(
"embulk-output-bigquery: LOAD IN PARALLEL {}",
paths.stream().map(Path::toString).collect(Collectors.joining("\n")));

// transfer data to BQ from files
ExecutorService executor = Executors.newFixedThreadPool(paths.size());
List<Future<JobStatistics>> jobStatisticFutures = new ArrayList<>();
List<JobStatistics.LoadStatistics> statistics = new ArrayList<>();
client.storeCachedSrcFieldsIfNeed();

for (Path path : paths) {
Future<JobStatistics> jobStatisticsFuture =
executor.submit(new BigqueryJobRunner(task, schema, path));
jobStatisticFutures.add(jobStatisticsFuture);
}
control.run(task.dump());
this.writers.values().forEach(BigqueryFileWriter::close);
logger.info("embulk-output-bigquery: finish to create intermediate files");

for (Future<JobStatistics> jobStatisticFuture : jobStatisticFutures) {
try {
statistics.add((JobStatistics.LoadStatistics) jobStatisticFuture.get());
paths = BigqueryUtil.getIntermediateFiles(task);
} catch (Exception e) {
logger.info(e.getMessage());
throw new RuntimeException(e);
}
}
BigqueryTransactionReport report =
getTransactionReport(task, client, statistics, this.writers.values());
if (task.getAbortOnError().get() && !task.getIsSkipJobResultCheck()) {
if (report.getNumInputRows().compareTo(report.getNumOutputRows()) != 0) {
String msg =
String.format(
"ABORT: `num_input_rows (%d)` and `num_output_rows (%d)` does not match",
report.getNumInputRows(), report.getNumOutputRows());
throw new RuntimeException(msg);
if (paths.isEmpty()) {
logger.info("embulk-output-bigquery: Nothing for transfer");
client.createTableIfNotExist(task.getTable());
return CONFIG_MAPPER_FACTORY.newConfigDiff();
}
}

if (task.getMode().equals("append") && task.getBeforeLoad().isPresent()) {
logger.info("embulk-output-bigquery: before_load will be executed");
logger.info("embulk-output-bigquery: {}", task.getBeforeLoad().get());
client.executeQuery(task.getBeforeLoad().get());
}
logger.debug(
"embulk-output-bigquery: LOAD IN PARALLEL {}",
paths.stream().map(Path::toString).collect(Collectors.joining("\n")));

if (task.getMode().equals("replace_backup")) {
if (task.getOldTable().isPresent()) {
client.copy(
task.getTable(),
task.getOldTable().get(),
task.getOldDataset().orElse(client.destinationDataset),
JobInfo.WriteDisposition.WRITE_TRUNCATE);
// transfer data to BQ from files
ExecutorService executor = Executors.newFixedThreadPool(paths.size());
List<Future<JobStatistics>> jobStatisticFutures = new ArrayList<>();
List<JobStatistics.LoadStatistics> statistics = new ArrayList<>();

for (Path path : paths) {
Future<JobStatistics> jobStatisticsFuture =
executor.submit(new BigqueryJobRunner(task, schema, path));
jobStatisticFutures.add(jobStatisticsFuture);
}
}
if (task.getTempTable().isPresent()) {
if (task.getMode().equals("merge")) {
client.merge(
task.getTempTable().get(),
task.getTable(),
task.getMergeKeys().orElse(Collections.emptyList()),
task.getMergeRule().orElse(Collections.emptyList()));
} else if (task.getMode().equals("append")) {
client.copy(
task.getTempTable().get(), task.getTable(), JobInfo.WriteDisposition.WRITE_APPEND);
} else {
client.copy(
task.getTempTable().get(), task.getTable(), JobInfo.WriteDisposition.WRITE_TRUNCATE);

for (Future<JobStatistics> jobStatisticFuture : jobStatisticFutures) {
try {
statistics.add((JobStatistics.LoadStatistics) jobStatisticFuture.get());
} catch (Exception e) {
throw new RuntimeException(e);
}
}
BigqueryTransactionReport report =
getTransactionReport(task, client, statistics, this.writers.values());
if (task.getAbortOnError().get() && !task.getIsSkipJobResultCheck()) {
if (report.getNumInputRows().compareTo(report.getNumOutputRows()) != 0) {
String msg =
String.format(
"ABORT: `num_input_rows (%d)` and `num_output_rows (%d)` does not match",
report.getNumInputRows(), report.getNumOutputRows());
throw new RuntimeException(msg);
}
}
client.deleteTable(task.getTempTable().get());
}

client.updateTableIfNeed();
if (task.getMode().equals("append") && task.getBeforeLoad().isPresent()) {
logger.info("embulk-output-bigquery: before_load will be executed");
logger.info("embulk-output-bigquery: {}", task.getBeforeLoad().get());
client.executeQuery(task.getBeforeLoad().get());
}

if (task.getDeleteFromLocalWhenJobEnd()) {
paths.forEach(p -> p.toFile().delete());
} else {
paths.forEach(
p -> {
File intermediateFile = new File(p.toString());
if (intermediateFile.exists()) {
logger.info("embulk-output-bigquery: keep {}", p.toString());
}
});
if (task.getMode().equals("replace_backup")) {
if (task.getOldTable().isPresent()) {
client.copy(
task.getTable(),
task.getOldTable().get(),
task.getOldDataset().orElse(client.destinationDataset),
JobInfo.WriteDisposition.WRITE_TRUNCATE);
}
}
if (task.getTempTable().isPresent()) {
if (task.getMode().equals("merge")) {
client.merge(
task.getTempTable().get(),
task.getTable(),
task.getMergeKeys().orElse(Collections.emptyList()),
task.getMergeRule().orElse(Collections.emptyList()));
} else if (task.getMode().equals("append")) {
client.copy(
task.getTempTable().get(), task.getTable(), JobInfo.WriteDisposition.WRITE_APPEND);
} else {
client.copy(
task.getTempTable().get(), task.getTable(), JobInfo.WriteDisposition.WRITE_TRUNCATE);
}
}

client.updateTableIfNeed();
} finally {
try {
if (task.getTempTable().isPresent()) {

@chikamura chikamura Sep 8, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Removed the mode switch since it was redundant: BigqueryTaskBuilder.setTempTable() only auto-populates temp_table for exactly these four modes, so task.getTempTable().isPresent() alone already implies the same restriction. This also matches ruby's own unconditional if task['temp_table'] check in its ensure block, with no mode restriction.

https://github.com/trocco-io/embulk-output-bigquery/blob/b4f8d2f4baf1c431a1cee3631790526441c3f9ec/lib/embulk/output/bigquery.rb#L498-L500

client.deleteTable(task.getTempTable().get());
}
} finally {
if (paths != null) {
if (task.getDeleteFromLocalWhenJobEnd()) {
paths.forEach(p -> p.toFile().delete());
} else {
paths.forEach(
p -> {
File intermediateFile = new File(p.toString());
if (intermediateFile.exists()) {
logger.info("embulk-output-bigquery: keep {}", p.toString());
}
});
}
}
}
}

return CONFIG_MAPPER_FACTORY.newConfigDiff();
Expand Down
Loading
Loading