<filename>ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/QueryCompactor.java
begin_unit|revision:0.9.5;language:Java;cregit-version:0.0.1
begin_comment
comment|/*  * Licensed to the Apache Software Foundation (ASF) under one  * or more contributor license agreements.  See the NOTICE file  * distributed with this work for additional information  * regarding copyright ownership.  The ASF licenses this file  * to you under the Apache License, Version 2.0 (the  * "License"); you may not use this file except in compliance  * with the License.  You may obtain a copy of the License at  *  *     http://www.apache.org/licenses/LICENSE-2.0  *  * Unless required by applicable law or agreed to in writing, software  * distributed under the License is distributed on an "AS IS" BASIS,  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  * See the License for the specific language governing permissions and  * limitations under the License.  */
end_comment

begin_package
package|package
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|hive
operator|.
name|ql
operator|.
name|txn
operator|.
name|compactor
package|;
end_package

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|commons
operator|.
name|lang3
operator|.
name|exception
operator|.
name|ExceptionUtils
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|fs
operator|.
name|FileSystem
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|fs
operator|.
name|Path
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|hive
operator|.
name|common
operator|.
name|ValidWriteIdList
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|hive
operator|.
name|conf
operator|.
name|HiveConf
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|hive
operator|.
name|metastore
operator|.
name|api
operator|.
name|AlreadyExistsException
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|hive
operator|.
name|metastore
operator|.
name|api
operator|.
name|FieldSchema
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|hive
operator|.
name|metastore
operator|.
name|api
operator|.
name|Partition
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|hive
operator|.
name|metastore
operator|.
name|api
operator|.
name|Table
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|hive
operator|.
name|metastore
operator|.
name|api
operator|.
name|StorageDescriptor
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|hive
operator|.
name|metastore
operator|.
name|txn
operator|.
name|CompactionInfo
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|hive
operator|.
name|ql
operator|.
name|DriverUtils
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|hive
operator|.
name|ql
operator|.
name|metadata
operator|.
name|Hive
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|hive
operator|.
name|ql
operator|.
name|metadata
operator|.
name|HiveException
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|hive
operator|.
name|ql
operator|.
name|session
operator|.
name|SessionState
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|security
operator|.
name|UserGroupInformation
import|;
end_import

begin_import
import|import
name|org
operator|.
name|slf4j
operator|.
name|Logger
import|;
end_import

begin_import
import|import
name|org
operator|.
name|slf4j
operator|.
name|LoggerFactory
import|;
end_import

begin_import
import|import
name|java
operator|.
name|io
operator|.
name|IOException
import|;
end_import

begin_import
import|import
name|java
operator|.
name|util
operator|.
name|List
import|;
end_import

begin_import
import|import
name|java
operator|.
name|util
operator|.
name|UUID
import|;
end_import

begin_comment
comment|/**  * Common interface for query based compactions.  */
end_comment

begin_class
specifier|abstract
class|class
name|QueryCompactor
block|{
specifier|private
specifier|static
specifier|final
name|Logger
name|LOG
init|=
name|LoggerFactory
operator|.
name|getLogger
argument_list|(
name|QueryCompactor
operator|.
name|class
operator|.
name|getName
argument_list|()
argument_list|)
decl_stmt|;
specifier|private
specifier|static
specifier|final
name|String
name|TMPDIR
init|=
literal|"_tmp"
decl_stmt|;
comment|/**    * Start a query based compaction.    * @param hiveConf hive configuration    * @param table the table, where the compaction should run    * @param partition the partition, where the compaction should run    * @param storageDescriptor this is the resolved storage descriptor    * @param writeIds valid write IDs used to filter rows while they're being read for compaction    * @param compactionInfo provides info about the type of compaction    * @throws IOException compaction cannot be finished.    */
specifier|abstract
name|void
name|runCompaction
parameter_list|(
name|HiveConf
name|hiveConf
parameter_list|,
name|Table
name|table
parameter_list|,
name|Partition
name|partition
parameter_list|,
name|StorageDescriptor
name|storageDescriptor
parameter_list|,
name|ValidWriteIdList
name|writeIds
parameter_list|,
name|CompactionInfo
name|compactionInfo
parameter_list|)
throws|throws
name|IOException
throws|,
name|HiveException
function_decl|;
comment|/**    * This is the final step of the compaction, which can vary based on compaction type. Usually this involves some file    * operation.    * @param dest The final directory; basically an SD directory.    * @param tmpTableName The name of the temporary table.    * @param conf hive configuration.    * @param actualWriteIds valid write Ids used to fetch the high watermark Id.    * @param compactorTxnId transaction, that the compacter started.    * @throws IOException failed to execute file system operation.    * @throws HiveException failed to execute file operation within hive.    */
specifier|protected
specifier|abstract
name|void
name|commitCompaction
parameter_list|(
name|String
name|dest
parameter_list|,
name|String
name|tmpTableName
parameter_list|,
name|HiveConf
name|conf
parameter_list|,
name|ValidWriteIdList
name|actualWriteIds
parameter_list|,
name|long
name|compactorTxnId
parameter_list|)
throws|throws
name|IOException
throws|,
name|HiveException
function_decl|;
comment|/**    * Run all the queries which performs the compaction.    * @param conf hive configuration, must be not null.    * @param tmpTableName The name of the temporary table.    * @param storageDescriptor this is the resolved storage descriptor.    * @param writeIds valid write IDs used to filter rows while they're being read for compaction.    * @param compactionInfo provides info about the type of compaction.    * @param createQueries collection of queries which creates the temporary tables.    * @param compactionQueries collection of queries which uses data from the original table and writes in temporary    *                          tables.    * @param dropQueries queries which drops the temporary tables.    * @throws IOException error during the run of the compaction.    */
name|void
name|runCompactionQueries
parameter_list|(
name|HiveConf
name|conf
parameter_list|,
name|String
name|tmpTableName
parameter_list|,
name|StorageDescriptor
name|storageDescriptor
parameter_list|,
name|ValidWriteIdList
name|writeIds
parameter_list|,
name|CompactionInfo
name|compactionInfo
parameter_list|,
name|List
argument_list|<
name|String
argument_list|>
name|createQueries
parameter_list|,
name|List
argument_list|<
name|String
argument_list|>
name|compactionQueries
parameter_list|,
name|List
argument_list|<
name|String
argument_list|>
name|dropQueries
parameter_list|)
throws|throws
name|IOException
block|{
name|Util
operator|.
name|disableLlapCaching
argument_list|(
name|conf
argument_list|)
expr_stmt|;
name|String
name|user
init|=
name|UserGroupInformation
operator|.
name|getCurrentUser
argument_list|()
operator|.
name|getShortUserName
argument_list|()
decl_stmt|;
name|SessionState
name|sessionState
init|=
name|DriverUtils
operator|.
name|setUpSessionState
argument_list|(
name|conf
argument_list|,
name|user
argument_list|,
literal|true
argument_list|)
decl_stmt|;
name|long
name|compactorTxnId
init|=
name|CompactorMR
operator|.
name|CompactorMap
operator|.
name|getCompactorTxnId
argument_list|(
name|conf
argument_list|)
decl_stmt|;
try|try
block|{
for|for
control|(
name|String
name|query
range|:
name|createQueries
control|)
block|{
try|try
block|{
name|LOG
operator|.
name|info
argument_list|(
literal|"Running {} compaction query into temp table with query: {}"
argument_list|,
name|compactionInfo
operator|.
name|isMajorCompaction
argument_list|()
condition|?
literal|"major"
else|:
literal|"minor"
argument_list|,
name|query
argument_list|)
expr_stmt|;
name|DriverUtils
operator|.
name|runOnDriver
argument_list|(
name|conf
argument_list|,
name|user
argument_list|,
name|sessionState
argument_list|,
name|query
argument_list|)
expr_stmt|;
block|}
catch|catch
parameter_list|(
name|Exception
name|ex
parameter_list|)
block|{
name|Throwable
name|cause
init|=
name|ex
decl_stmt|;
while|while
condition|(
name|cause
operator|!=
literal|null
operator|&&
operator|!
operator|(
name|cause
operator|instanceof
name|AlreadyExistsException
operator|)
condition|)
block|{
name|cause
operator|=
name|cause
operator|.
name|getCause
argument_list|()
expr_stmt|;
block|}
if|if
condition|(
name|cause
operator|==
literal|null
condition|)
block|{
throw|throw
operator|new
name|IOException
argument_list|(
name|ex
argument_list|)
throw|;
block|}
block|}
block|}
for|for
control|(
name|String
name|query
range|:
name|compactionQueries
control|)
block|{
name|LOG
operator|.
name|info
argument_list|(
literal|"Running {} compaction via query: {}"
argument_list|,
name|compactionInfo
operator|.
name|isMajorCompaction
argument_list|()
condition|?
literal|"major"
else|:
literal|"minor"
argument_list|,
name|query
argument_list|)
expr_stmt|;
name|DriverUtils
operator|.
name|runOnDriver
argument_list|(
name|conf
argument_list|,
name|user
argument_list|,
name|sessionState
argument_list|,
name|query
argument_list|,
name|writeIds
argument_list|,
name|compactorTxnId
argument_list|)
expr_stmt|;
block|}
name|commitCompaction
argument_list|(
name|storageDescriptor
operator|.
name|getLocation
argument_list|()
argument_list|,
name|tmpTableName
argument_list|,
name|conf
argument_list|,
name|writeIds
argument_list|,
name|compactorTxnId
argument_list|)
expr_stmt|;
block|}
catch|catch
parameter_list|(
name|HiveException
name|e
parameter_list|)
block|{
name|LOG
operator|.
name|error
argument_list|(
literal|"Error doing query based {} compaction"
argument_list|,
name|compactionInfo
operator|.
name|isMajorCompaction
argument_list|()
condition|?
literal|"major"
else|:
literal|"minor"
argument_list|,
name|e
argument_list|)
expr_stmt|;
throw|throw
operator|new
name|IOException
argument_list|(
name|e
argument_list|)
throw|;
block|}
finally|finally
block|{
try|try
block|{
for|for
control|(
name|String
name|query
range|:
name|dropQueries
control|)
block|{
name|LOG
operator|.
name|info
argument_list|(
literal|"Running {} compaction query into temp table with query: {}"
argument_list|,
name|compactionInfo
operator|.
name|isMajorCompaction
argument_list|()
condition|?
literal|"major"
else|:
literal|"minor"
argument_list|,
name|query
argument_list|)
expr_stmt|;
name|DriverUtils
operator|.
name|runOnDriver
argument_list|(
name|conf
argument_list|,
name|user
argument_list|,
name|sessionState
argument_list|,
name|query
argument_list|)
expr_stmt|;
block|}
block|}
catch|catch
parameter_list|(
name|HiveException
name|e
parameter_list|)
block|{
name|LOG
operator|.
name|error
argument_list|(
literal|"Unable to drop temp table {} which was created for running {} compaction"
argument_list|,
name|tmpTableName
argument_list|,
name|compactionInfo
operator|.
name|isMajorCompaction
argument_list|()
condition|?
literal|"major"
else|:
literal|"minor"
argument_list|)
expr_stmt|;
name|LOG
operator|.
name|error
argument_list|(
name|ExceptionUtils
operator|.
name|getStackTrace
argument_list|(
name|e
argument_list|)
argument_list|)
expr_stmt|;
block|}
block|}
block|}
comment|/**    * Collection of some helper functions.    */
specifier|static
class|class
name|Util
block|{
comment|/**      * Generate a random tmp path, under the provided storage.      * @param sd storage descriptor, must be not null.      * @return path, always not null      */
specifier|static
name|String
name|generateTmpPath
parameter_list|(
name|StorageDescriptor
name|sd
parameter_list|)
block|{
return|return
name|sd
operator|.
name|getLocation
argument_list|()
operator|+
literal|"/"
operator|+
name|TMPDIR
operator|+
literal|"_"
operator|+
name|UUID
operator|.
name|randomUUID
argument_list|()
operator|.
name|toString
argument_list|()
return|;
block|}
comment|/**      * Unless caching is explicitly required for ETL queries this method disables it.      * LLAP cache content lookup is file based, and since compaction alters the file structure it is not beneficial to      * cache anything here, as it won't (and actually can't) ever be looked up later.      * @param conf the Hive configuration      */
specifier|private
specifier|static
name|void
name|disableLlapCaching
parameter_list|(
name|HiveConf
name|conf
parameter_list|)
block|{
name|String
name|llapIOETLSkipFormat
init|=
name|conf
operator|.
name|getVar
argument_list|(
name|HiveConf
operator|.
name|ConfVars
operator|.
name|LLAP_IO_ETL_SKIP_FORMAT
argument_list|)
decl_stmt|;
if|if
condition|(
operator|!
literal|"none"
operator|.
name|equals
argument_list|(
name|llapIOETLSkipFormat
argument_list|)
condition|)
block|{
comment|// Unless caching is explicitly required for ETL queries - disable it.
name|conf
operator|.
name|setVar
argument_list|(
name|HiveConf
operator|.
name|ConfVars
operator|.
name|LLAP_IO_ETL_SKIP_FORMAT
argument_list|,
literal|"all"
argument_list|)
expr_stmt|;
block|}
block|}
comment|/**        * Get a create temporary table query string with Orc ACID columns.        * @param tableName name of the new temporary table        * @param table the table where the compaction is running        * @return create query        */
specifier|static
name|String
name|getCreateTempTableQueryWithAcidColumns
parameter_list|(
name|String
name|tableName
parameter_list|,
name|Table
name|table
parameter_list|)
block|{
name|StringBuilder
name|query
init|=
operator|new
name|StringBuilder
argument_list|(
literal|"create temporary external table "
argument_list|)
operator|.
name|append
argument_list|(
name|tableName
argument_list|)
operator|.
name|append
argument_list|(
literal|" ("
argument_list|)
decl_stmt|;
comment|// Acid virtual columns
name|query
operator|.
name|append
argument_list|(
literal|"`operation` int, `originalTransaction` bigint, `bucket` int, `rowId` bigint, `currentTransaction` "
operator|+
literal|"bigint, `row` struct<"
argument_list|)
expr_stmt|;
name|List
argument_list|<
name|FieldSchema
argument_list|>
name|cols
init|=
name|table
operator|.
name|getSd
argument_list|()
operator|.
name|getCols
argument_list|()
decl_stmt|;
name|boolean
name|isFirst
init|=
literal|true
decl_stmt|;
comment|// Actual columns
for|for
control|(
name|FieldSchema
name|col
range|:
name|cols
control|)
block|{
if|if
condition|(
operator|!
name|isFirst
condition|)
block|{
name|query
operator|.
name|append
argument_list|(
literal|", "
argument_list|)
expr_stmt|;
block|}
name|isFirst
operator|=
literal|false
expr_stmt|;
name|query
operator|.
name|append
argument_list|(
literal|"`"
argument_list|)
operator|.
name|append
argument_list|(
name|col
operator|.
name|getName
argument_list|()
argument_list|)
operator|.
name|append
argument_list|(
literal|"` "
argument_list|)
operator|.
name|append
argument_list|(
literal|":"
argument_list|)
operator|.
name|append
argument_list|(
name|col
operator|.
name|getType
argument_list|()
argument_list|)
expr_stmt|;
block|}
name|query
operator|.
name|append
argument_list|(
literal|">)"
argument_list|)
expr_stmt|;
return|return
name|query
operator|.
name|toString
argument_list|()
return|;
block|}
comment|/**      * Remove the root directory of a table if it's empty.      * @param conf the Hive configuration      * @param tmpTableName name of the table      * @throws IOException the directory cannot be deleted      * @throws HiveException the table is not found      */
specifier|static
name|void
name|cleanupEmptyDir
parameter_list|(
name|HiveConf
name|conf
parameter_list|,
name|String
name|tmpTableName
parameter_list|)
throws|throws
name|IOException
throws|,
name|HiveException
block|{
name|org
operator|.
name|apache
operator|.
name|hadoop
operator|.
name|hive
operator|.
name|ql
operator|.
name|metadata
operator|.
name|Table
name|tmpTable
init|=
name|Hive
operator|.
name|get
argument_list|()
operator|.
name|getTable
argument_list|(
name|tmpTableName
argument_list|)
decl_stmt|;
if|if
condition|(
name|tmpTable
operator|!=
literal|null
condition|)
block|{
name|Path
name|path
init|=
operator|new
name|Path
argument_list|(
name|tmpTable
operator|.
name|getSd
argument_list|()
operator|.
name|getLocation
argument_list|()
argument_list|)
decl_stmt|;
name|FileSystem
name|fs
init|=
name|path
operator|.
name|getFileSystem
argument_list|(
name|conf
argument_list|)
decl_stmt|;
if|if
condition|(
operator|!
name|fs
operator|.
name|listFiles
argument_list|(
name|path
argument_list|,
literal|false
argument_list|)
operator|.
name|hasNext
argument_list|()
condition|)
block|{
name|fs
operator|.
name|delete
argument_list|(
name|path
argument_list|,
literal|true
argument_list|)
expr_stmt|;
block|}
block|}
block|}
block|}
block|}
end_class

end_unit

