You are viewing a plain text version of this content. The canonical link for it is here.
Posted to gitbox@hive.apache.org by GitBox <gi...@apache.org> on 2021/04/08 15:25:27 UTC
[GitHub] [hive] marton-bod opened a new pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
marton-bod opened a new pull request #2161:
URL: https://github.com/apache/hive/pull/2161
CI build will fail since Tez dependency is SNAPSHOT, using: https://github.com/apache/tez/pull/101
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609880455
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java
##########
@@ -250,9 +255,32 @@ public int execute() {
this.setException(new HiveException(monitor.getDiagnostics()));
}
- // fetch the counters
try {
Set<StatusGetOpts> statusGetOpts = EnumSet.of(StatusGetOpts.GET_COUNTERS);
+ // save useful commit information into session conf, e.g. for custom commit hooks
+ List<BaseWork> allWork = work.getAllWork();
+ boolean hasReducer = allWork.stream().map(workToVertex::get).anyMatch(v -> v.getName().startsWith("Reducer"));
+ for (BaseWork baseWork : allWork) {
+ Vertex vertex = workToVertex.get(baseWork);
+ if (!hasReducer || vertex.getName().startsWith("Reducer")) {
+ // construct the parsable job id
+ VertexStatus status = dagClient.getVertexStatus(vertex.getName(), statusGetOpts);
+ String[] jobIdParts = status.getId().split("_");
+ // status.getId() returns something like: vertex_1617722404520_0001_1_00
+ // this should be transformed to a parsable JobID: job_16177224045200_0001
+ int vertexId = Integer.parseInt(jobIdParts[jobIdParts.length - 1]);
+ String jobId = String.format("job_%s%d_%s", jobIdParts[1], vertexId, jobIdParts[2]);
+ // prefix with table name (for multi-table inserts), if available
+ String tableName = Optional.ofNullable(workToConf.get(baseWork)).map(c -> c.get("name")).orElse(null);
+ String jobIdKey = HIVE_TEZ_COMMIT_JOB_ID + (tableName == null ? "" : "." + tableName);;
+ String taskCountKey = HIVE_TEZ_COMMIT_TASK_COUNT + (tableName == null ? "" : "." + tableName);
+ // save info into session conf
+ HiveConf sessionConf = SessionState.get().getConf();
+ sessionConf.set(jobIdKey, jobId);
+ sessionConf.setInt(taskCountKey, status.getProgress().getSucceededTaskCount());
Review comment:
Good point, I wanted to unset these variables in a finally block during `commitInsertTable`, just forgot to add it.
I did not find a place to put these in `queryState` but maybe I missed something.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r616496048
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java
##########
@@ -256,4 +265,74 @@ private static PartitionSpec spec(Schema schema, Properties properties,
return PartitionSpec.unpartitioned();
}
}
+
+ @Override
+ public void commitInsertTable(org.apache.hadoop.hive.metastore.api.Table table, boolean overwrite)
+ throws MetaException {
+ String tableName = TableIdentifier.of(table.getDbName(), table.getTableName()).toString();
+
+ // check status to determine whether we need to commit or to abort
+ JobConf jobConf = new JobConf(conf);
+ String queryIdKey = jobConf.get("hive.query.id") + "." + tableName + ".result";
Review comment:
Could we use constants here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609855130
##########
File path: iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveShell.java
##########
@@ -154,6 +155,13 @@ public void closeSession() {
}
}
+ public String executeExplain(String statement) {
+ List<Object[]> objects = executeStatement(statement);
+ return objects.stream()
+ .map(o -> (String) o[0])
+ .collect(Collectors.joining("\n"));
+ }
+
Review comment:
Needed this during way too many debug sessions :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r617490443
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -239,6 +252,16 @@ public void abortJob(JobContext originalContext, int status) throws IOException
cleanup(jobContext, jobLocations);
}
+ private Set<FileStatus> listForCommits(JobConf jobConf, String jobLocation) throws IOException {
Review comment:
Right, good idea
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r617486952
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -346,25 +372,23 @@ private static ExecutorService tableExecutor(Configuration conf, int maxThreadNu
/**
* Get the committed data files for this table and job.
+ *
+ * @param numTasks Number of writer tasks that produced a forCommit file
* @param executor The executor used for reading the forCommit files parallel
* @param location The location of the table
* @param jobContext The job context
* @param io The FileIO used for reading a files generated for commit
* @param throwOnFailure If <code>true</code> then it throws an exception on failure
* @return The list of the committed data files
*/
- private static Collection<DataFile> dataFiles(ExecutorService executor, String location, JobContext jobContext,
- FileIO io, boolean throwOnFailure) {
+ private static Collection<DataFile> dataFiles(int numTasks, ExecutorService executor, String location,
+ JobContext jobContext, FileIO io, boolean throwOnFailure) {
Review comment:
nit: I have seen Iceberg reviewers asking for 4 space padding.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609854598
##########
File path: iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveShell.java
##########
@@ -154,6 +155,13 @@ public void closeSession() {
}
}
+ public String executeExplain(String statement) {
+ List<Object[]> objects = executeStatement(statement);
+ return objects.stream()
+ .map(o -> (String) o[0])
+ .collect(Collectors.joining("\n"));
+ }
+
Review comment:
👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609851060
##########
File path: iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -224,6 +224,11 @@ private void cleanup(JobContext jobContext) throws IOException {
// If there are reducers, then every reducer will generate a result file.
// If this is a map only task, then every mapper will generate a result file.
int expectedFiles = conf.getNumReduceTasks() > 0 ? conf.getNumReduceTasks() : conf.getNumMapTasks();
+ // For Tez, we can only reliably get the number of tasks from the Tez AM, not the job conf
Review comment:
Why not set this in `commitInsertTable` - it would keep the hacks in one place instead of spreading around?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
marton-bod commented on pull request #2161:
URL: https://github.com/apache/hive/pull/2161#issuecomment-815925533
@pvary @lcspinter can you please take a look?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609892146
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java
##########
@@ -250,9 +255,32 @@ public int execute() {
this.setException(new HiveException(monitor.getDiagnostics()));
}
- // fetch the counters
try {
Set<StatusGetOpts> statusGetOpts = EnumSet.of(StatusGetOpts.GET_COUNTERS);
+ // save useful commit information into session conf, e.g. for custom commit hooks
+ List<BaseWork> allWork = work.getAllWork();
+ boolean hasReducer = allWork.stream().map(workToVertex::get).anyMatch(v -> v.getName().startsWith("Reducer"));
+ for (BaseWork baseWork : allWork) {
+ Vertex vertex = workToVertex.get(baseWork);
+ if (!hasReducer || vertex.getName().startsWith("Reducer")) {
+ // construct the parsable job id
+ VertexStatus status = dagClient.getVertexStatus(vertex.getName(), statusGetOpts);
+ String[] jobIdParts = status.getId().split("_");
+ // status.getId() returns something like: vertex_1617722404520_0001_1_00
+ // this should be transformed to a parsable JobID: job_16177224045200_0001
+ int vertexId = Integer.parseInt(jobIdParts[jobIdParts.length - 1]);
+ String jobId = String.format("job_%s%d_%s", jobIdParts[1], vertexId, jobIdParts[2]);
+ // prefix with table name (for multi-table inserts), if available
+ String tableName = Optional.ofNullable(workToConf.get(baseWork)).map(c -> c.get("name")).orElse(null);
+ String jobIdKey = HIVE_TEZ_COMMIT_JOB_ID + (tableName == null ? "" : "." + tableName);;
+ String taskCountKey = HIVE_TEZ_COMMIT_TASK_COUNT + (tableName == null ? "" : "." + tableName);
+ // save info into session conf
+ HiveConf sessionConf = SessionState.get().getConf();
+ sessionConf.set(jobIdKey, jobId);
+ sessionConf.setInt(taskCountKey, status.getProgress().getSucceededTaskCount());
Review comment:
Yeah, I'll look into it. One question though: will the queryState be available in the meta hook? Is it available also via the SessionState?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609849281
##########
File path: iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java
##########
@@ -250,4 +259,43 @@ private static PartitionSpec spec(Schema schema, Properties properties,
return PartitionSpec.unpartitioned();
}
}
+
+ @Override
+ public void commitInsertTable(org.apache.hadoop.hive.metastore.api.Table table, boolean overwrite)
Review comment:
In case of multi-table insert this method is called for every table?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609850873
##########
File path: iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java
##########
@@ -250,4 +259,43 @@ private static PartitionSpec spec(Schema schema, Properties properties,
return PartitionSpec.unpartitioned();
}
}
+
+ @Override
+ public void commitInsertTable(org.apache.hadoop.hive.metastore.api.Table table, boolean overwrite)
Review comment:
yes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r617040976
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -105,13 +105,18 @@ public void commitTask(TaskAttemptContext originalContext) throws IOException {
.executeWith(tableExecutor)
.run(output -> {
Table table = HiveIcebergStorageHandler.table(context.getJobConf(), output);
- HiveIcebergRecordWriter writer = writers.get(output);
- DataFile[] closedFiles = writer != null ? writer.dataFiles() : new DataFile[0];
- String fileForCommitLocation = generateFileForCommitLocation(table.location(), jobConf,
- attemptID.getJobID(), attemptID.getTaskID().getId());
-
- // Creating the file containing the data files generated by this task for this table
- createFileForCommit(closedFiles, fileForCommitLocation, table.io());
+ if (table != null) {
Review comment:
Actually, this issue does not occur with mr, only with tez, therefore it currently does not come up in upstream Iceberg. Will fix this once Tez writes have been enabled there too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r616510710
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java
##########
@@ -256,4 +265,74 @@ private static PartitionSpec spec(Schema schema, Properties properties,
return PartitionSpec.unpartitioned();
}
}
+
+ @Override
+ public void commitInsertTable(org.apache.hadoop.hive.metastore.api.Table table, boolean overwrite)
+ throws MetaException {
+ String tableName = TableIdentifier.of(table.getDbName(), table.getTableName()).toString();
+
+ // check status to determine whether we need to commit or to abort
+ JobConf jobConf = new JobConf(conf);
+ String queryIdKey = jobConf.get("hive.query.id") + "." + tableName + ".result";
Review comment:
Makes sense. I was thinking of replacing the query id with a constant like `HIVE_TEZ_COMMIT_JOB_RESULT` so it would be `HIVE_TEZ_COMMIT_JOB_ID + "." + tableName`, like for job id and task num to make things consisent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
marton-bod commented on pull request #2161:
URL: https://github.com/apache/hive/pull/2161#issuecomment-822504138
This is roughly what the changes would need to look like once we have the new Tez version released:
https://github.com/marton-bod/hive/pull/1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609854017
##########
File path: iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -224,6 +224,11 @@ private void cleanup(JobContext jobContext) throws IOException {
// If there are reducers, then every reducer will generate a result file.
// If this is a map only task, then every mapper will generate a result file.
int expectedFiles = conf.getNumReduceTasks() > 0 ? conf.getNumReduceTasks() : conf.getNumMapTasks();
+ // For Tez, we can only reliably get the number of tasks from the Tez AM, not the job conf
Review comment:
I wanted to be considerate of MR (even though it's unsupported in upstream Hive), but my initial approach was to not touch it here but set the numReduceTasks to the correct value in the `commitInsertTable`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
marton-bod commented on pull request #2161:
URL: https://github.com/apache/hive/pull/2161#issuecomment-816766162
@pvary I've uploaded a set of new commits addressing your review comments + changing the long-term approach with the temporary listing solution, as discussed. The listing solution is mostly based on the current downstream TP solution just simplified and refined a bit to save the information into the session conf. This should make it easy to swap out later with the proper vertex id-based implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r616608877
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -105,13 +105,18 @@ public void commitTask(TaskAttemptContext originalContext) throws IOException {
.executeWith(tableExecutor)
.run(output -> {
Table table = HiveIcebergStorageHandler.table(context.getJobConf(), output);
- HiveIcebergRecordWriter writer = writers.get(output);
- DataFile[] closedFiles = writer != null ? writer.dataFiles() : new DataFile[0];
- String fileForCommitLocation = generateFileForCommitLocation(table.location(), jobConf,
- attemptID.getJobID(), attemptID.getTaskID().getId());
-
- // Creating the file containing the data files generated by this task for this table
- createFileForCommit(closedFiles, fileForCommitLocation, table.io());
+ if (table != null) {
Review comment:
This happens during task commit, so before the commitInsert hook is called.
The essential problem here is that `OUTPUT_TABLES` contains all the tables, however, only those tables are serialized into the jobconfig that are relevant for the given task. So it tries to iterate over 1...N tables, but only has access to serialized Table 1 (hence the if).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609854654
##########
File path: iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java
##########
@@ -121,6 +121,9 @@ public void configureJobConf(TableDesc tableDesc, JobConf jobConf) {
if (tableDesc != null && tableDesc.getProperties() != null &&
tableDesc.getProperties().get(WRITE_KEY) != null) {
jobConf.set("mapred.output.committer.class", HiveIcebergOutputCommitter.class.getName());
+ // this will turn off job committing on the Tez AM side, so that we can commit the write jobs
+ // on HS2 side instead using the HiveIcebergMetaHook
+ jobConf.set("hive.tez.mapreduce.output.committer.class", "");
Review comment:
Good idea, will try the NullOutCommitter
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609852322
##########
File path: iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java
##########
@@ -250,4 +259,43 @@ private static PartitionSpec spec(Schema schema, Properties properties,
return PartitionSpec.unpartitioned();
}
}
+
+ @Override
+ public void commitInsertTable(org.apache.hadoop.hive.metastore.api.Table table, boolean overwrite)
+ throws MetaException {
+ // construct the job context
+ JobConf jobConf = new JobConf(conf);
+ String tableName = TableIdentifier.of(table.getDbName(), table.getTableName()).toString();
Review comment:
Not sure yet, I'll look into it. The catalog name might be stored in the table params, or we might need to load the Iceberg table to find out if we can. @lcspinter might have some more insights
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r612455851
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java
##########
@@ -250,9 +255,32 @@ public int execute() {
this.setException(new HiveException(monitor.getDiagnostics()));
}
- // fetch the counters
try {
Set<StatusGetOpts> statusGetOpts = EnumSet.of(StatusGetOpts.GET_COUNTERS);
+ // save useful commit information into session conf, e.g. for custom commit hooks
+ List<BaseWork> allWork = work.getAllWork();
+ boolean hasReducer = allWork.stream().map(workToVertex::get).anyMatch(v -> v.getName().startsWith("Reducer"));
+ for (BaseWork baseWork : allWork) {
+ Vertex vertex = workToVertex.get(baseWork);
+ if (!hasReducer || vertex.getName().startsWith("Reducer")) {
+ // construct the parsable job id
+ VertexStatus status = dagClient.getVertexStatus(vertex.getName(), statusGetOpts);
+ String[] jobIdParts = status.getId().split("_");
+ // status.getId() returns something like: vertex_1617722404520_0001_1_00
+ // this should be transformed to a parsable JobID: job_16177224045200_0001
+ int vertexId = Integer.parseInt(jobIdParts[jobIdParts.length - 1]);
+ String jobId = String.format("job_%s%d_%s", jobIdParts[1], vertexId, jobIdParts[2]);
+ // prefix with table name (for multi-table inserts), if available
+ String tableName = Optional.ofNullable(workToConf.get(baseWork)).map(c -> c.get("name")).orElse(null);
+ String jobIdKey = HIVE_TEZ_COMMIT_JOB_ID + (tableName == null ? "" : "." + tableName);;
+ String taskCountKey = HIVE_TEZ_COMMIT_TASK_COUNT + (tableName == null ? "" : "." + tableName);
+ // save info into session conf
+ HiveConf sessionConf = SessionState.get().getConf();
+ sessionConf.set(jobIdKey, jobId);
+ sessionConf.setInt(taskCountKey, status.getProgress().getSucceededTaskCount());
Review comment:
I'll look into this in the following PR, once we've replaced the temporary listing solution with the permanent one and upgraded the Tez dependency.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary merged pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
pvary merged pull request #2161:
URL: https://github.com/apache/hive/pull/2161
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609854017
##########
File path: iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -224,6 +224,11 @@ private void cleanup(JobContext jobContext) throws IOException {
// If there are reducers, then every reducer will generate a result file.
// If this is a map only task, then every mapper will generate a result file.
int expectedFiles = conf.getNumReduceTasks() > 0 ? conf.getNumReduceTasks() : conf.getNumMapTasks();
+ // For Tez, we can only reliably get the number of tasks from the Tez AM, not the job conf
Review comment:
My initial approach was to set the `jobConf.numReduceTasks()` to the correct value in the `commitInsertTable`, but I didn't want to mess up MR, although since it's unsupported in upstream Hive we could do that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609847726
##########
File path: iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java
##########
@@ -250,4 +259,43 @@ private static PartitionSpec spec(Schema schema, Properties properties,
return PartitionSpec.unpartitioned();
}
}
+
+ @Override
+ public void commitInsertTable(org.apache.hadoop.hive.metastore.api.Table table, boolean overwrite)
+ throws MetaException {
+ // construct the job context
+ JobConf jobConf = new JobConf(conf);
+ String tableName = TableIdentifier.of(table.getDbName(), table.getTableName()).toString();
Review comment:
How this will get the `Catalog` information?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r616613187
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java
##########
@@ -256,4 +265,74 @@ private static PartitionSpec spec(Schema schema, Properties properties,
return PartitionSpec.unpartitioned();
}
}
+
+ @Override
+ public void commitInsertTable(org.apache.hadoop.hive.metastore.api.Table table, boolean overwrite)
+ throws MetaException {
+ String tableName = TableIdentifier.of(table.getDbName(), table.getTableName()).toString();
+
+ // check status to determine whether we need to commit or to abort
+ JobConf jobConf = new JobConf(conf);
+ String queryIdKey = jobConf.get("hive.query.id") + "." + tableName + ".result";
+ boolean success = jobConf.getBoolean(queryIdKey, false);
+
+ // construct the job context
+ JobID jobID = JobID.forName(jobConf.get(TezTask.HIVE_TEZ_COMMIT_JOB_ID + "." + tableName));
+ int numTasks = conf.getInt(TezTask.HIVE_TEZ_COMMIT_TASK_COUNT + "." + tableName, -1);
+ jobConf.setNumReduceTasks(numTasks);
+ JobContext jobContext = new JobContextImpl(jobConf, jobID, null);
+
+ // we should only commit this current table because
+ // for multi-table inserts, this hook method will be called sequentially for each target table
+ jobConf.set(InputFormatConfig.OUTPUT_TABLES, tableName);
+
+ OutputCommitter committer = new HiveIcebergOutputCommitter();
+ try {
+ if (success) {
+ try {
+ committer.commitJob(jobContext);
+ } catch (Exception commitExc) {
+ LOG.error("Error while trying to commit job (table: {}, jobID: {}). Will abort it now.",
+ tableName, jobID, commitExc);
+ abortJob(jobContext, committer, true);
+ throw new MetaException("Unable to commit job: " + commitExc.getMessage());
+ }
+ } else {
+ abortJob(jobContext, committer, false);
+ }
+ } finally {
+ // avoid config pollution with prefixed/suffixed keys
+ cleanCommitConfig(queryIdKey, tableName);
+ }
+ }
+
+ private void abortJob(JobContext jobContext, OutputCommitter committer, boolean suppressExc) throws MetaException {
+ try {
+ committer.abortJob(jobContext, JobStatus.State.FAILED);
+ } catch (IOException abortExc) {
+ LOG.error("Error while trying to abort failed job. There might be uncleaned data files.", abortExc);
+ if (!suppressExc) {
+ throw new MetaException("Unable to abort job: " + abortExc.getMessage());
+ }
+ }
+ }
+
+ private void cleanCommitConfig(String queryIdKey, String tableName) {
+ conf.unset(TezTask.HIVE_TEZ_COMMIT_JOB_ID + "." + tableName);
+ conf.unset(TezTask.HIVE_TEZ_COMMIT_TASK_COUNT + "." + tableName);
+ conf.unset(InputFormatConfig.SERIALIZED_TABLE_PREFIX + tableName);
+ conf.unset(queryIdKey);
+ }
+
+ @Override
+ public void preInsertTable(org.apache.hadoop.hive.metastore.api.Table table, boolean overwrite)
+ throws MetaException {
+ // do nothing
+ }
+
+ @Override
+ public void rollbackInsertTable(org.apache.hadoop.hive.metastore.api.Table table, boolean overwrite)
+ throws MetaException {
+ // do nothing
Review comment:
I didn't put it there because f there is an execution error we should get a non-0 return code from Tez AM, instead of an exception. But now that I'm thinking about it, I suppose Hive will throw an exception at the end if the code was non-0, but I will test it out to make sure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r616658639
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -105,13 +105,18 @@ public void commitTask(TaskAttemptContext originalContext) throws IOException {
.executeWith(tableExecutor)
.run(output -> {
Table table = HiveIcebergStorageHandler.table(context.getJobConf(), output);
- HiveIcebergRecordWriter writer = writers.get(output);
- DataFile[] closedFiles = writer != null ? writer.dataFiles() : new DataFile[0];
- String fileForCommitLocation = generateFileForCommitLocation(table.location(), jobConf,
- attemptID.getJobID(), attemptID.getTaskID().getId());
-
- // Creating the file containing the data files generated by this task for this table
- createFileForCommit(closedFiles, fileForCommitLocation, table.io());
+ if (table != null) {
Review comment:
Sure, will do!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609853621
##########
File path: iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergStorageHandlerWithEngine.java
##########
@@ -291,8 +290,6 @@ public void testSelectDistinctFromTable() throws IOException {
@Test
public void testInsert() throws IOException {
- Assume.assumeTrue("Tez write is not implemented yet", executionEngine.equals("mr"));
Review comment:
Ohh.. how I like these changes 😄 🥇
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609873171
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java
##########
@@ -250,9 +255,32 @@ public int execute() {
this.setException(new HiveException(monitor.getDiagnostics()));
}
- // fetch the counters
try {
Set<StatusGetOpts> statusGetOpts = EnumSet.of(StatusGetOpts.GET_COUNTERS);
+ // save useful commit information into session conf, e.g. for custom commit hooks
+ List<BaseWork> allWork = work.getAllWork();
+ boolean hasReducer = allWork.stream().map(workToVertex::get).anyMatch(v -> v.getName().startsWith("Reducer"));
+ for (BaseWork baseWork : allWork) {
+ Vertex vertex = workToVertex.get(baseWork);
+ if (!hasReducer || vertex.getName().startsWith("Reducer")) {
+ // construct the parsable job id
+ VertexStatus status = dagClient.getVertexStatus(vertex.getName(), statusGetOpts);
+ String[] jobIdParts = status.getId().split("_");
+ // status.getId() returns something like: vertex_1617722404520_0001_1_00
+ // this should be transformed to a parsable JobID: job_16177224045200_0001
+ int vertexId = Integer.parseInt(jobIdParts[jobIdParts.length - 1]);
+ String jobId = String.format("job_%s%d_%s", jobIdParts[1], vertexId, jobIdParts[2]);
+ // prefix with table name (for multi-table inserts), if available
+ String tableName = Optional.ofNullable(workToConf.get(baseWork)).map(c -> c.get("name")).orElse(null);
+ String jobIdKey = HIVE_TEZ_COMMIT_JOB_ID + (tableName == null ? "" : "." + tableName);;
+ String taskCountKey = HIVE_TEZ_COMMIT_TASK_COUNT + (tableName == null ? "" : "." + tableName);
+ // save info into session conf
+ HiveConf sessionConf = SessionState.get().getConf();
+ sessionConf.set(jobIdKey, jobId);
+ sessionConf.setInt(taskCountKey, status.getProgress().getSucceededTaskCount());
Review comment:
Wouldn't this pollute the conf forever (for the session)?
Why not use `queryState` instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609867548
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java
##########
@@ -250,9 +255,32 @@ public int execute() {
this.setException(new HiveException(monitor.getDiagnostics()));
}
- // fetch the counters
try {
Set<StatusGetOpts> statusGetOpts = EnumSet.of(StatusGetOpts.GET_COUNTERS);
+ // save useful commit information into session conf, e.g. for custom commit hooks
+ List<BaseWork> allWork = work.getAllWork();
+ boolean hasReducer = allWork.stream().map(workToVertex::get).anyMatch(v -> v.getName().startsWith("Reducer"));
+ for (BaseWork baseWork : allWork) {
+ Vertex vertex = workToVertex.get(baseWork);
+ if (!hasReducer || vertex.getName().startsWith("Reducer")) {
Review comment:
Do I understand correctly that we would like to identify if we wrote something or not with this vertex?
Would `Vertex.getDataSinks()` work here?
What about if we have a multi-table insert writing to Iceberg and non-Iceberg tables?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609856986
##########
File path: iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveShell.java
##########
@@ -154,6 +155,13 @@ public void closeSession() {
}
}
+ public String executeExplain(String statement) {
+ List<Object[]> objects = executeStatement(statement);
+ return objects.stream()
+ .map(o -> (String) o[0])
+ .collect(Collectors.joining("\n"));
+ }
+
Review comment:
Leave a comment there not to remove it even if it is not used 😄
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609899783
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java
##########
@@ -250,9 +255,32 @@ public int execute() {
this.setException(new HiveException(monitor.getDiagnostics()));
}
- // fetch the counters
try {
Set<StatusGetOpts> statusGetOpts = EnumSet.of(StatusGetOpts.GET_COUNTERS);
+ // save useful commit information into session conf, e.g. for custom commit hooks
+ List<BaseWork> allWork = work.getAllWork();
+ boolean hasReducer = allWork.stream().map(workToVertex::get).anyMatch(v -> v.getName().startsWith("Reducer"));
+ for (BaseWork baseWork : allWork) {
+ Vertex vertex = workToVertex.get(baseWork);
+ if (!hasReducer || vertex.getName().startsWith("Reducer")) {
+ // construct the parsable job id
+ VertexStatus status = dagClient.getVertexStatus(vertex.getName(), statusGetOpts);
+ String[] jobIdParts = status.getId().split("_");
+ // status.getId() returns something like: vertex_1617722404520_0001_1_00
+ // this should be transformed to a parsable JobID: job_16177224045200_0001
+ int vertexId = Integer.parseInt(jobIdParts[jobIdParts.length - 1]);
+ String jobId = String.format("job_%s%d_%s", jobIdParts[1], vertexId, jobIdParts[2]);
+ // prefix with table name (for multi-table inserts), if available
+ String tableName = Optional.ofNullable(workToConf.get(baseWork)).map(c -> c.get("name")).orElse(null);
+ String jobIdKey = HIVE_TEZ_COMMIT_JOB_ID + (tableName == null ? "" : "." + tableName);;
+ String taskCountKey = HIVE_TEZ_COMMIT_TASK_COUNT + (tableName == null ? "" : "." + tableName);
+ // save info into session conf
+ HiveConf sessionConf = SessionState.get().getConf();
+ sessionConf.set(jobIdKey, jobId);
+ sessionConf.setInt(taskCountKey, status.getProgress().getSucceededTaskCount());
Review comment:
Maybe we should make it available in a map of queryId/QueryState?
And we can clean that up when we finish a query
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r617272290
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -105,13 +105,18 @@ public void commitTask(TaskAttemptContext originalContext) throws IOException {
.executeWith(tableExecutor)
.run(output -> {
Table table = HiveIcebergStorageHandler.table(context.getJobConf(), output);
- HiveIcebergRecordWriter writer = writers.get(output);
- DataFile[] closedFiles = writer != null ? writer.dataFiles() : new DataFile[0];
- String fileForCommitLocation = generateFileForCommitLocation(table.location(), jobConf,
- attemptID.getJobID(), attemptID.getTaskID().getId());
-
- // Creating the file containing the data files generated by this task for this table
- createFileForCommit(closedFiles, fileForCommitLocation, table.io());
+ if (table != null) {
Review comment:
Yeah, I think you're right
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609885766
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java
##########
@@ -250,9 +255,32 @@ public int execute() {
this.setException(new HiveException(monitor.getDiagnostics()));
}
- // fetch the counters
try {
Set<StatusGetOpts> statusGetOpts = EnumSet.of(StatusGetOpts.GET_COUNTERS);
+ // save useful commit information into session conf, e.g. for custom commit hooks
+ List<BaseWork> allWork = work.getAllWork();
+ boolean hasReducer = allWork.stream().map(workToVertex::get).anyMatch(v -> v.getName().startsWith("Reducer"));
+ for (BaseWork baseWork : allWork) {
+ Vertex vertex = workToVertex.get(baseWork);
+ if (!hasReducer || vertex.getName().startsWith("Reducer")) {
+ // construct the parsable job id
+ VertexStatus status = dagClient.getVertexStatus(vertex.getName(), statusGetOpts);
+ String[] jobIdParts = status.getId().split("_");
+ // status.getId() returns something like: vertex_1617722404520_0001_1_00
+ // this should be transformed to a parsable JobID: job_16177224045200_0001
+ int vertexId = Integer.parseInt(jobIdParts[jobIdParts.length - 1]);
+ String jobId = String.format("job_%s%d_%s", jobIdParts[1], vertexId, jobIdParts[2]);
+ // prefix with table name (for multi-table inserts), if available
+ String tableName = Optional.ofNullable(workToConf.get(baseWork)).map(c -> c.get("name")).orElse(null);
+ String jobIdKey = HIVE_TEZ_COMMIT_JOB_ID + (tableName == null ? "" : "." + tableName);;
+ String taskCountKey = HIVE_TEZ_COMMIT_TASK_COUNT + (tableName == null ? "" : "." + tableName);
+ // save info into session conf
+ HiveConf sessionConf = SessionState.get().getConf();
+ sessionConf.set(jobIdKey, jobId);
+ sessionConf.setInt(taskCountKey, status.getProgress().getSucceededTaskCount());
Review comment:
I think we should extend the `queryState` for Iceberg queries. This could be a candidate to put there, and also it could be good to put the snapshots of the table the query is execution on.
What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609847726
##########
File path: iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java
##########
@@ -250,4 +259,43 @@ private static PartitionSpec spec(Schema schema, Properties properties,
return PartitionSpec.unpartitioned();
}
}
+
+ @Override
+ public void commitInsertTable(org.apache.hadoop.hive.metastore.api.Table table, boolean overwrite)
+ throws MetaException {
+ // construct the job context
+ JobConf jobConf = new JobConf(conf);
+ String tableName = TableIdentifier.of(table.getDbName(), table.getTableName()).toString();
Review comment:
How this will get the `Catalog` information when @lcspinter's catalog related changes get in?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r610638294
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java
##########
@@ -250,9 +255,32 @@ public int execute() {
this.setException(new HiveException(monitor.getDiagnostics()));
}
- // fetch the counters
try {
Set<StatusGetOpts> statusGetOpts = EnumSet.of(StatusGetOpts.GET_COUNTERS);
+ // save useful commit information into session conf, e.g. for custom commit hooks
+ List<BaseWork> allWork = work.getAllWork();
+ boolean hasReducer = allWork.stream().map(workToVertex::get).anyMatch(v -> v.getName().startsWith("Reducer"));
+ for (BaseWork baseWork : allWork) {
+ Vertex vertex = workToVertex.get(baseWork);
+ if (!hasReducer || vertex.getName().startsWith("Reducer")) {
+ // construct the parsable job id
+ VertexStatus status = dagClient.getVertexStatus(vertex.getName(), statusGetOpts);
+ String[] jobIdParts = status.getId().split("_");
+ // status.getId() returns something like: vertex_1617722404520_0001_1_00
+ // this should be transformed to a parsable JobID: job_16177224045200_0001
+ int vertexId = Integer.parseInt(jobIdParts[jobIdParts.length - 1]);
+ String jobId = String.format("job_%s%d_%s", jobIdParts[1], vertexId, jobIdParts[2]);
Review comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod edited a comment on pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
marton-bod edited a comment on pull request #2161:
URL: https://github.com/apache/hive/pull/2161#issuecomment-822504138
This is roughly what the changes would need to look like once we have the new Tez version released:
https://github.com/marton-bod/hive/pull/1
(using the new Tez API instead of the listing)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609862126
##########
File path: iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -224,6 +224,11 @@ private void cleanup(JobContext jobContext) throws IOException {
// If there are reducers, then every reducer will generate a result file.
// If this is a map only task, then every mapper will generate a result file.
int expectedFiles = conf.getNumReduceTasks() > 0 ? conf.getNumReduceTasks() : conf.getNumMapTasks();
+ // For Tez, we can only reliably get the number of tasks from the Tez AM, not the job conf
Review comment:
Sounds good. Will set the value in the `commitInsertTable`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r616603042
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java
##########
@@ -256,4 +265,74 @@ private static PartitionSpec spec(Schema schema, Properties properties,
return PartitionSpec.unpartitioned();
}
}
+
+ @Override
+ public void commitInsertTable(org.apache.hadoop.hive.metastore.api.Table table, boolean overwrite)
+ throws MetaException {
+ String tableName = TableIdentifier.of(table.getDbName(), table.getTableName()).toString();
+
+ // check status to determine whether we need to commit or to abort
+ JobConf jobConf = new JobConf(conf);
+ String queryIdKey = jobConf.get("hive.query.id") + "." + tableName + ".result";
+ boolean success = jobConf.getBoolean(queryIdKey, false);
+
+ // construct the job context
+ JobID jobID = JobID.forName(jobConf.get(TezTask.HIVE_TEZ_COMMIT_JOB_ID + "." + tableName));
+ int numTasks = conf.getInt(TezTask.HIVE_TEZ_COMMIT_TASK_COUNT + "." + tableName, -1);
+ jobConf.setNumReduceTasks(numTasks);
+ JobContext jobContext = new JobContextImpl(jobConf, jobID, null);
+
+ // we should only commit this current table because
+ // for multi-table inserts, this hook method will be called sequentially for each target table
+ jobConf.set(InputFormatConfig.OUTPUT_TABLES, tableName);
+
+ OutputCommitter committer = new HiveIcebergOutputCommitter();
+ try {
+ if (success) {
+ try {
+ committer.commitJob(jobContext);
+ } catch (Exception commitExc) {
+ LOG.error("Error while trying to commit job (table: {}, jobID: {}). Will abort it now.",
+ tableName, jobID, commitExc);
+ abortJob(jobContext, committer, true);
+ throw new MetaException("Unable to commit job: " + commitExc.getMessage());
+ }
+ } else {
+ abortJob(jobContext, committer, false);
+ }
+ } finally {
+ // avoid config pollution with prefixed/suffixed keys
+ cleanCommitConfig(queryIdKey, tableName);
+ }
+ }
+
+ private void abortJob(JobContext jobContext, OutputCommitter committer, boolean suppressExc) throws MetaException {
+ try {
+ committer.abortJob(jobContext, JobStatus.State.FAILED);
+ } catch (IOException abortExc) {
+ LOG.error("Error while trying to abort failed job. There might be uncleaned data files.", abortExc);
+ if (!suppressExc) {
+ throw new MetaException("Unable to abort job: " + abortExc.getMessage());
+ }
+ }
+ }
+
+ private void cleanCommitConfig(String queryIdKey, String tableName) {
+ conf.unset(TezTask.HIVE_TEZ_COMMIT_JOB_ID + "." + tableName);
+ conf.unset(TezTask.HIVE_TEZ_COMMIT_TASK_COUNT + "." + tableName);
+ conf.unset(InputFormatConfig.SERIALIZED_TABLE_PREFIX + tableName);
+ conf.unset(queryIdKey);
+ }
+
+ @Override
+ public void preInsertTable(org.apache.hadoop.hive.metastore.api.Table table, boolean overwrite)
+ throws MetaException {
+ // do nothing
+ }
+
+ @Override
+ public void rollbackInsertTable(org.apache.hadoop.hive.metastore.api.Table table, boolean overwrite)
+ throws MetaException {
+ // do nothing
Review comment:
Shouldn't we call abortJob here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r610624208
##########
File path: iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java
##########
@@ -121,6 +121,9 @@ public void configureJobConf(TableDesc tableDesc, JobConf jobConf) {
if (tableDesc != null && tableDesc.getProperties() != null &&
tableDesc.getProperties().get(WRITE_KEY) != null) {
jobConf.set("mapred.output.committer.class", HiveIcebergOutputCommitter.class.getName());
+ // this will turn off job committing on the Tez AM side, so that we can commit the write jobs
+ // on HS2 side instead using the HiveIcebergMetaHook
+ jobConf.set("hive.tez.mapreduce.output.committer.class", "");
Review comment:
I replaced the string key with the constant. As for using a null committer, that class would need to extend the `OutputCommitter` from the `tez-api` package. Currently iceberg-handler does not depend on Tez, so we can only achieve that if we introduce that dependency.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r616510710
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java
##########
@@ -256,4 +265,74 @@ private static PartitionSpec spec(Schema schema, Properties properties,
return PartitionSpec.unpartitioned();
}
}
+
+ @Override
+ public void commitInsertTable(org.apache.hadoop.hive.metastore.api.Table table, boolean overwrite)
+ throws MetaException {
+ String tableName = TableIdentifier.of(table.getDbName(), table.getTableName()).toString();
+
+ // check status to determine whether we need to commit or to abort
+ JobConf jobConf = new JobConf(conf);
+ String queryIdKey = jobConf.get("hive.query.id") + "." + tableName + ".result";
Review comment:
Makes sense. I was thinking of replacing the query id with a constant like `HIVE_TEZ_COMMIT_JOB_RESULT` so it would be `HIVE_TEZ_COMMIT_JOB_RESULT + "." + tableName`, like for job id and task num to make things consisent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r617487622
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -239,6 +252,16 @@ public void abortJob(JobContext originalContext, int status) throws IOException
cleanup(jobContext, jobLocations);
}
+ private Set<FileStatus> listForCommits(JobConf jobConf, String jobLocation) throws IOException {
Review comment:
nit: Maybe a javadoc that do not use it for anything else than abort.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609867972
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java
##########
@@ -250,9 +255,32 @@ public int execute() {
this.setException(new HiveException(monitor.getDiagnostics()));
}
- // fetch the counters
try {
Set<StatusGetOpts> statusGetOpts = EnumSet.of(StatusGetOpts.GET_COUNTERS);
+ // save useful commit information into session conf, e.g. for custom commit hooks
+ List<BaseWork> allWork = work.getAllWork();
+ boolean hasReducer = allWork.stream().map(workToVertex::get).anyMatch(v -> v.getName().startsWith("Reducer"));
+ for (BaseWork baseWork : allWork) {
+ Vertex vertex = workToVertex.get(baseWork);
+ if (!hasReducer || vertex.getName().startsWith("Reducer")) {
+ // construct the parsable job id
+ VertexStatus status = dagClient.getVertexStatus(vertex.getName(), statusGetOpts);
+ String[] jobIdParts = status.getId().split("_");
+ // status.getId() returns something like: vertex_1617722404520_0001_1_00
+ // this should be transformed to a parsable JobID: job_16177224045200_0001
+ int vertexId = Integer.parseInt(jobIdParts[jobIdParts.length - 1]);
+ String jobId = String.format("job_%s%d_%s", jobIdParts[1], vertexId, jobIdParts[2]);
Review comment:
nit: Maybe use a constant here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r617206293
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -105,13 +105,18 @@ public void commitTask(TaskAttemptContext originalContext) throws IOException {
.executeWith(tableExecutor)
.run(output -> {
Table table = HiveIcebergStorageHandler.table(context.getJobConf(), output);
- HiveIcebergRecordWriter writer = writers.get(output);
- DataFile[] closedFiles = writer != null ? writer.dataFiles() : new DataFile[0];
- String fileForCommitLocation = generateFileForCommitLocation(table.location(), jobConf,
- attemptID.getJobID(), attemptID.getTaskID().getId());
-
- // Creating the file containing the data files generated by this task for this table
- createFileForCommit(closedFiles, fileForCommitLocation, table.io());
+ if (table != null) {
Review comment:
I think this is the same situation that we have been with some other Tez patches. It does not hurt to be there an we keep the two codes from diverging. But it's up to you to decide.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r616608877
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -105,13 +105,18 @@ public void commitTask(TaskAttemptContext originalContext) throws IOException {
.executeWith(tableExecutor)
.run(output -> {
Table table = HiveIcebergStorageHandler.table(context.getJobConf(), output);
- HiveIcebergRecordWriter writer = writers.get(output);
- DataFile[] closedFiles = writer != null ? writer.dataFiles() : new DataFile[0];
- String fileForCommitLocation = generateFileForCommitLocation(table.location(), jobConf,
- attemptID.getJobID(), attemptID.getTaskID().getId());
-
- // Creating the file containing the data files generated by this task for this table
- createFileForCommit(closedFiles, fileForCommitLocation, table.io());
+ if (table != null) {
Review comment:
This happens during task commit, so before the commitInsert hook is called.
The essential problem here is that `OUTPUT_TABLES` contains all the tables, however, only those tables are serialized into the jobconfig that are relevant for the given task. So it tries to iterate over 1...N tables (based on `OUTPUT_TABLES`), but only has access to serialized Table 1 (hence the if). The whole parallel commit logic for multitable inserts on both the task commit and job commit side are broken I think, if there is more than one vertex writing to target tables. Currently the tests pass because it creates a single writer vertex, which will have both tables serialized into its config.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r616603785
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -105,13 +105,18 @@ public void commitTask(TaskAttemptContext originalContext) throws IOException {
.executeWith(tableExecutor)
.run(output -> {
Table table = HiveIcebergStorageHandler.table(context.getJobConf(), output);
- HiveIcebergRecordWriter writer = writers.get(output);
- DataFile[] closedFiles = writer != null ? writer.dataFiles() : new DataFile[0];
- String fileForCommitLocation = generateFileForCommitLocation(table.location(), jobConf,
- attemptID.getJobID(), attemptID.getTaskID().getId());
-
- // Creating the file containing the data files generated by this task for this table
- createFileForCommit(closedFiles, fileForCommitLocation, table.io());
+ if (table != null) {
Review comment:
Could we do this check in `HiveIcebergMetaHook.commitInsertTable`?
Then we would not need any change here...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r616497039
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java
##########
@@ -256,4 +265,74 @@ private static PartitionSpec spec(Schema schema, Properties properties,
return PartitionSpec.unpartitioned();
}
}
+
+ @Override
+ public void commitInsertTable(org.apache.hadoop.hive.metastore.api.Table table, boolean overwrite)
+ throws MetaException {
+ String tableName = TableIdentifier.of(table.getDbName(), table.getTableName()).toString();
+
+ // check status to determine whether we need to commit or to abort
+ JobConf jobConf = new JobConf(conf);
+ String queryIdKey = jobConf.get("hive.query.id") + "." + tableName + ".result";
Review comment:
Maybe even "hive.query.id.%.result"? Or "hive.query.id.result."+tableName?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r616649742
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -105,13 +105,18 @@ public void commitTask(TaskAttemptContext originalContext) throws IOException {
.executeWith(tableExecutor)
.run(output -> {
Table table = HiveIcebergStorageHandler.table(context.getJobConf(), output);
- HiveIcebergRecordWriter writer = writers.get(output);
- DataFile[] closedFiles = writer != null ? writer.dataFiles() : new DataFile[0];
- String fileForCommitLocation = generateFileForCommitLocation(table.location(), jobConf,
- attemptID.getJobID(), attemptID.getTaskID().getId());
-
- // Creating the file containing the data files generated by this task for this table
- createFileForCommit(closedFiles, fileForCommitLocation, table.io());
+ if (table != null) {
Review comment:
Got it, then we should push this change to the Iceberg code as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609852924
##########
File path: iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java
##########
@@ -121,6 +121,9 @@ public void configureJobConf(TableDesc tableDesc, JobConf jobConf) {
if (tableDesc != null && tableDesc.getProperties() != null &&
tableDesc.getProperties().get(WRITE_KEY) != null) {
jobConf.set("mapred.output.committer.class", HiveIcebergOutputCommitter.class.getName());
+ // this will turn off job committing on the Tez AM side, so that we can commit the write jobs
+ // on HS2 side instead using the HiveIcebergMetaHook
+ jobConf.set("hive.tez.mapreduce.output.committer.class", "");
Review comment:
nit: is there a constant for the key? Also I have seen `NullOutCommitter` class around somewhere. Could we use that here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609860590
##########
File path: iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -224,6 +224,11 @@ private void cleanup(JobContext jobContext) throws IOException {
// If there are reducers, then every reducer will generate a result file.
// If this is a map only task, then every mapper will generate a result file.
int expectedFiles = conf.getNumReduceTasks() > 0 ? conf.getNumReduceTasks() : conf.getNumMapTasks();
+ // For Tez, we can only reliably get the number of tasks from the Tez AM, not the job conf
Review comment:
If I understand correctly in Hive3 `commitInsertTable` is not called when the execution engine is MR.
If it would be called anywhere then we are in a big trouble anyway as `HiveIcebergOutputCommitter.commitJob` would be called twice.
So all-in-all I would go for this change
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r610629268
##########
File path: iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java
##########
@@ -121,6 +121,9 @@ public void configureJobConf(TableDesc tableDesc, JobConf jobConf) {
if (tableDesc != null && tableDesc.getProperties() != null &&
tableDesc.getProperties().get(WRITE_KEY) != null) {
jobConf.set("mapred.output.committer.class", HiveIcebergOutputCommitter.class.getName());
+ // this will turn off job committing on the Tez AM side, so that we can commit the write jobs
+ // on HS2 side instead using the HiveIcebergMetaHook
+ jobConf.set("hive.tez.mapreduce.output.committer.class", "");
Review comment:
Then using `NullOutCommitter` does not worth it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609888812
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java
##########
@@ -250,9 +255,32 @@ public int execute() {
this.setException(new HiveException(monitor.getDiagnostics()));
}
- // fetch the counters
try {
Set<StatusGetOpts> statusGetOpts = EnumSet.of(StatusGetOpts.GET_COUNTERS);
+ // save useful commit information into session conf, e.g. for custom commit hooks
+ List<BaseWork> allWork = work.getAllWork();
+ boolean hasReducer = allWork.stream().map(workToVertex::get).anyMatch(v -> v.getName().startsWith("Reducer"));
+ for (BaseWork baseWork : allWork) {
+ Vertex vertex = workToVertex.get(baseWork);
+ if (!hasReducer || vertex.getName().startsWith("Reducer")) {
Review comment:
Correct, the goal is to pick out only those vertices that wrote to a table. I will look into the `dataSinks()` method, that could be a better way to identify them.
In principle, the mixed table scenario you mentioned should work. If we had two reducer vertices (that wrote to tables), we'll check both of them here. The vertex that wrote to the Iceberg table will publish its info (jobId, task num) with the table suffix into the session conf, and the vertex that wrote the Hive table will do the same except without any table suffix. But since we need no commit for the Hive table one, that second part is more or less irrelevant for now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r617490157
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -346,25 +372,23 @@ private static ExecutorService tableExecutor(Configuration conf, int maxThreadNu
/**
* Get the committed data files for this table and job.
+ *
+ * @param numTasks Number of writer tasks that produced a forCommit file
* @param executor The executor used for reading the forCommit files parallel
* @param location The location of the table
* @param jobContext The job context
* @param io The FileIO used for reading a files generated for commit
* @param throwOnFailure If <code>true</code> then it throws an exception on failure
* @return The list of the committed data files
*/
- private static Collection<DataFile> dataFiles(ExecutorService executor, String location, JobContext jobContext,
- FileIO io, boolean throwOnFailure) {
+ private static Collection<DataFile> dataFiles(int numTasks, ExecutorService executor, String location,
+ JobContext jobContext, FileIO io, boolean throwOnFailure) {
Review comment:
They are a bit confusing about this. The 4 spaces padding is indeed the general rule for line continuation, but I've been asked earlier by Anton on other PRs not to do that for method parameter indentations and do it this way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #2161: HIVE-25006: Commit Iceberg writes in HiveMetaHook instead of TezAM
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r617491058
##########
File path: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -346,25 +372,23 @@ private static ExecutorService tableExecutor(Configuration conf, int maxThreadNu
/**
* Get the committed data files for this table and job.
+ *
+ * @param numTasks Number of writer tasks that produced a forCommit file
* @param executor The executor used for reading the forCommit files parallel
* @param location The location of the table
* @param jobContext The job context
* @param io The FileIO used for reading a files generated for commit
* @param throwOnFailure If <code>true</code> then it throws an exception on failure
* @return The list of the committed data files
*/
- private static Collection<DataFile> dataFiles(ExecutorService executor, String location, JobContext jobContext,
- FileIO io, boolean throwOnFailure) {
+ private static Collection<DataFile> dataFiles(int numTasks, ExecutorService executor, String location,
+ JobContext jobContext, FileIO io, boolean throwOnFailure) {
Review comment:
Fun stuff 😄
Whatever!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] marton-bod commented on a change in pull request #2161: WIP: HIVE-XXX: Move Iceberg job commit to HS2 side
Posted by GitBox <gi...@apache.org>.
marton-bod commented on a change in pull request #2161:
URL: https://github.com/apache/hive/pull/2161#discussion_r609854017
##########
File path: iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java
##########
@@ -224,6 +224,11 @@ private void cleanup(JobContext jobContext) throws IOException {
// If there are reducers, then every reducer will generate a result file.
// If this is a map only task, then every mapper will generate a result file.
int expectedFiles = conf.getNumReduceTasks() > 0 ? conf.getNumReduceTasks() : conf.getNumMapTasks();
+ // For Tez, we can only reliably get the number of tasks from the Tez AM, not the job conf
Review comment:
My initial approach was to set the numReduceTasks to the correct value in the `commitInsertTable`, but I didn't want to mess up MR, although since it's unsupported in upstream Hive we could do that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org