SQOOP_CONFIG_FILE=$1
FILE_LIST=$2
SPLICE_SCHEMA=$3
IMPORT_PATH=$4
QUERY_PATH=$5
LOG_FILE=$6
HADOOP_SBIN=$7
SPLICE_BIN=$8

SQLSERVER_DBO="dbo"
COLON=":"
DEFAULT_MAPPERS="1"
LOCAL="false"
TUTORIAL_HOME="/tmp"

# check for correct number of parameters
if [ $# -eq 8 ]; then
	# check if the sqoop config file exists
	if [ ! -f $SQOOP_CONFIG_FILE ]; then
		echo $SQOOP_CONFIG_FILE not found
		exit 1
	fi
	
	# check if the file list exists
	if [ ! -f $FILE_LIST ]; then
		echo $FILE_LIST not found
		exit 1
	else
		# loop through th file list and parse each line	
		while IFS=. read schema table column
		do
			echo Exporting $schema.$table
			if [ $column ]; then echo Splitting on $column
			fi

			# comment out this section as the example is for mysql
			# the commented code applies to sqlserver/oracle
			# check if the schema is dbo
			#if [ "$schema" == "$SQLSERVER_DBO" ]; then
			#	echo Has dbo as schema
			#	TABLE=$table
			#else
			#	echo Does NOT have dbo as schema
			#	TABLE=$schema.$table
			#fi
			# for mysql just set the TABLE var to the actual table name in the file
			TABLE=$table

			# Determine if there was a column specified then call sqoop accordingly optionally passing in the --split-by value
			# Also checks if the number of mappers is specified
			if [ $column ]; then
				#COLON_INDEX=$(expr index $column $COLON)
				COLON_INDEX=$(echo $column | sed -n "s/[$COLON].*//p" | wc -c)
				echo Colon Index: $COLON_INDEX
				if [ $COLON_INDEX -gt 0 ]; then
					COLUMN_END_POS=$(($COLON_INDEX-1))
					START_MAPPER_POS=$(($COLON_INDEX+1))
					END_MAPPER_POS=$((${#column}-$COLON_INDEX))
					echo Full column length: ${#column}
					echo Start Mapper Pos: $START_MAPPER_POS
					echo End Mapper Pos: $END_MAPPER_POS
					#COLUMN=$(expr substr $column 1 $COLUMN_END_POS)
					COLUMM=$(echo $column | cut -c 1-($COLUMN_END_POS-1))
					#MAPPERS=$(expr substr $column $START_MAPPER_POS $END_MAPPER_POS)
					MAPPERS=$(echo $column | cut -c $START_MAPPER_POS-($END_MAPPER_POS-1))
				else
					COLUMN=$column
					MAPPERS=$DEFAULT_MAPPERS
				fi
				echo Column: $COLUMN
				echo Mappers: $MAPPERS
			else
				#COLON_INDEX=$(expr index $table $COLON)
				COLON_INDEX=$(echo $table | sed -n "s/[$COLON].*//p" | wc -c)
				echo Colon Index: $COLON_INDEX
				if [ $COLON_INDEX -gt 0 ]; then
					COLUMN_END_POS=$(($COLON_INDEX-1))
					START_MAPPER_POS=$(($COLON_INDEX+1))
					END_MAPPER_POS=$((${#table}-$COLON_INDEX))
					echo Full column length: ${#column}
					echo Start Mapper Pos: $START_MAPPER_POS
					echo End Mapper Pos: $END_MAPPER_POS
					#TABLE=$(expr substr $table 1 $COLUMN_END_POS)
					#MAPPERS=$(expr substr $table $START_MAPPER_POS $END_MAPPER_POS)
					TABLE=$(echo $table | cut -c 1-$COLUMN_END_POS-1)
					MAPPERS=$(echo $table | cut -c $START_MAPPER_POS-($END_MAPPER_POS-1))
					table=$TABLE
				else
					MAPPERS=$DEFAULT_MAPPERS
				fi
				# comment out as this code block only applies to sqlserver/oracle
				#if [ "$schema" != "$SQLSERVER_DBO" ]; then
				#	TABLE=$schema.$table
				#fi
				echo Table: $TABLE
				echo Mappers: $MAPPERS

			fi

			# check if there is a query file for this table
			queryFile=$QUERY_PATH/query-$SPLICE_SCHEMA-$table.sql
			echo The query file is: $queryFile
			if [ -f "$queryFile" ]; then
				query=$(cat $queryFile)
				echo $query
			else
				query=""
				echo Query file not found. $query
			fi
			
			# This code is only required for this standalone example.
			# In a cluster setup this would not be needed as hadoop would already be running along with Splice
			if [ "$LOCAL" == "true" ]; then
				echo "Starting Hadoop (sqoop)"
				$HADOOP_SBIN/start-dfs.sh
				$HADOOP_SBIN/start-yarn.sh
				echo "Sleeping..."
				sleep 15s
				echo "Continuing..."
			fi


			# Make sure the directories exist in hdfs, clear the directory, and set permissions
			hadoop fs -mkdir -p /data/$SPLICE_SCHEMA/$table
			hadoop fs -chmod 777 /data/$SPLICE_SCHEMA/$table
			hadoop fs -rm -skipTrash /data/$SPLICE_SCHEMA/$table/*
			hadoop fs -mkdir -p /status/$SPLICE_SCHEMA/$table
			hadoop fs -chmod 777 /status/$SPLICE_SCHEMA/$table

			# add a section to create local directories
			if [ "$LOCAL" == "true" ]; then
				mkdir -p $TUTORIAL_HOME/data/$SPLICE_SCHEMA/$table
				mkdir -p $TUTORIAL_HOME/status/$SPLICE_SCHEMA/$table
				chmod 777 $TUTORIAL_HOME/data/$SPLICE_SCHEMA/$table
				chmod 777 $TUTORIAL_HOME/status/$SPLICE_SCHEMA/$table
				rm $TUTORIAL_HOME/data/$SPLICE_SCHEMA/$table/*
			fi

			# exit script if an error is encountered
			set -e

			echo The query param is: $query!
			echo The column param is: $column!
 
			if [ -z "$query" ]; then
				if [ $column ]; then
					echo Execute sqoop with split-by
					sqoop import --options-file $SQOOP_CONFIG_FILE --append --table $TABLE --schema $schema --split-by $COLUMN --target-dir /data/$SPLICE_SCHEMA/$table --m $MAPPERS
				else
					echo Execute sqoop without split-by
					sqoop import --options-file $SQOOP_CONFIG_FILE --append --table $TABLE --schema $schema --target-dir /data/$SPLICE_SCHEMA/$table --m $MAPPERS
				fi
			else 
				if [ $column ]; then
					echo Execute sqoop with split-by
					sqoop import --options-file $SQOOP_CONFIG_FILE --append --query "$query" --split-by $COLUMN --target-dir /data/$SPLICE_SCHEMA/$table --m $MAPPERS
				else
					echo Execute sqoop with query and without split-by
					sqoop import --options-file $SQOOP_CONFIG_FILE --append --query "$query" --target-dir /data/$SPLICE_SCHEMA/$table --m $MAPPERS
					#echo Sqoop extract for $SCHEMA.$TABLE failed because a query file is present but no column specified for the split-by.
					#exit 1
				fi
			fi
			
			if [ $? -gt 0 ]; then
				echo Sqoop extract failed
				exit 1
			else
				if [ "$LOCAL" == "true" ]; then
					# add command to copy the files from hdfs to local fs
					hadoop fs -copyToLocal /data/$SPLICE_SCHEMA/$table/* $TUTORIAL_HOME/data/$SPLICE_SCHEMA/$table/
					echo Export of $schema.$table successful
				fi
			fi

			# Kill hadoop so Splice import will work
			# This is just a workaround for standalone environments
			# In a cluster environment this would not be necessary
			if [ "$LOCAL" == "true" ]; then
				echo "Stopping Hadoop (sqoop)"
				$HADOOP_SBIN/stop-yarn.sh
				$HADOOP_SBIN/stop-dfs.sh
				# Make sure Splice is stopped then start Splice
				echo Stopping Splice
				echo Starting Splice
				$SPLICE_BIN/stop-splice.sh
				$SPLICE_BIN/start-splice.sh
			fi

			echo Importing $SPLICE_SCHEMA.$table to Splice Machine	
			
			# Import data from HDFS to Splice
			IMPORT_FILE=$IMPORT_PATH/import-$SPLICE_SCHEMA-$table.sql
			if [ -f $IMPORT_FILE ]; then
				$SPLICE_HOME/bin/sqlshell.sh -f $IMPORT_FILE
				STATUS=$(grep "ERROR SE\|ERROR 08006\|ClosedChannelException" $LOG_FILE | wc -l)
				if [ $STATUS -gt 0 ]; then
					echo Splice import failed
					exit 1
				else
					echo Import of $SPLICE_SCHEMA.$table completed
				fi
			else
				echo $IMPORT_FILE not found.... skipping import to Splice Machine
				exit 1
			fi

			# do not exit script if there are errors
			set +e
		done <$FILE_LIST
	fi
else
	echo Incorrect number of parameters specified to execute script
	echo "Usage: ./run-sqoop-full.sh <config file> <table list file> <splice schema name> <import file path> <log file name> <hadoop sbin path> <splice install bin path>"
	exit 1
fi

echo Sqoop job completed.
exit 0
