You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@carbondata.apache.org by GitBox <gi...@apache.org> on 2020/10/01 10:54:25 UTC
[GitHub] [carbondata] marchpure commented on a change in pull request #3965: [CARBONDATA-4016] NPE and FileNotFound in Show Segments and Insert Stage
marchpure commented on a change in pull request #3965:
URL: https://github.com/apache/carbondata/pull/3965#discussion_r498155406
##########
File path: integration/spark/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertFromStageCommand.scala
##########
@@ -477,13 +479,23 @@ case class CarbonInsertFromStageCommand(
stageFiles.map { stage =>
executorService.submit(new Runnable {
override def run(): Unit = {
- val filePath = stage._1.getAbsolutePath
- val stream = FileFactory.getDataInputStream(filePath)
+ val filePath = tableStagePath + CarbonCommonConstants.FILE_SEPARATOR + stage._1.getName
+ var stream: DataInputStream = null
try {
- val stageInput = gson.fromJson(new InputStreamReader(stream), classOf[StageInput])
- output.add(stageInput)
+ stream = FileFactory.getDataInputStream(filePath)
+ var retry = CarbonInsertFromStageCommand.DELETE_FILES_RETRY_TIMES
+ breakable (while (retry > 0) try {
Review comment:
I have modified code according to your suggestion
##########
File path: integration/spark/src/main/scala/org/apache/carbondata/api/CarbonStore.scala
##########
@@ -96,20 +100,31 @@ object CarbonStore {
* Read stage files and return input files
*/
def readStageInput(
+ tableStagePath: String,
stageFiles: Seq[CarbonFile],
status: StageInput.StageStatus): Seq[StageInput] = {
val gson = new Gson()
val output = Collections.synchronizedList(new util.ArrayList[StageInput]())
stageFiles.map { stage =>
- val filePath = stage.getAbsolutePath
- val stream = FileFactory.getDataInputStream(filePath)
+ val filePath = tableStagePath + CarbonCommonConstants.FILE_SEPARATOR + stage.getName
+ var stream: DataInputStream = null
try {
- val stageInput = gson.fromJson(new InputStreamReader(stream), classOf[StageInput])
- stageInput.setCreateTime(stage.getLastModifiedTime)
- stageInput.setStatus(status)
- output.add(stageInput)
+ stream = FileFactory.getDataInputStream(filePath)
+ var retry = READ_FILE_RETRY_TIMES
+ breakable { while (retry > 0) { try {
+ val stageInput = gson.fromJson(new InputStreamReader(stream), classOf[StageInput])
+ stageInput.setCreateTime(stage.getLastModifiedTime)
+ stageInput.setStatus(status)
+ output.add(stageInput)
Review comment:
I have modified code according to your suggestion
##########
File path: integration/spark/src/main/scala/org/apache/carbondata/api/CarbonStore.scala
##########
@@ -96,20 +100,31 @@ object CarbonStore {
* Read stage files and return input files
*/
def readStageInput(
+ tableStagePath: String,
stageFiles: Seq[CarbonFile],
status: StageInput.StageStatus): Seq[StageInput] = {
val gson = new Gson()
val output = Collections.synchronizedList(new util.ArrayList[StageInput]())
stageFiles.map { stage =>
- val filePath = stage.getAbsolutePath
- val stream = FileFactory.getDataInputStream(filePath)
+ val filePath = tableStagePath + CarbonCommonConstants.FILE_SEPARATOR + stage.getName
+ var stream: DataInputStream = null
try {
- val stageInput = gson.fromJson(new InputStreamReader(stream), classOf[StageInput])
- stageInput.setCreateTime(stage.getLastModifiedTime)
- stageInput.setStatus(status)
- output.add(stageInput)
+ stream = FileFactory.getDataInputStream(filePath)
+ var retry = READ_FILE_RETRY_TIMES
+ breakable { while (retry > 0) { try {
+ val stageInput = gson.fromJson(new InputStreamReader(stream), classOf[StageInput])
+ stageInput.setCreateTime(stage.getLastModifiedTime)
+ stageInput.setStatus(status)
+ output.add(stageInput)
+ } catch {
+ case _ : FileNotFoundException => breakable()
Review comment:
I have modified code according to your suggestion
##########
File path: integration/spark/src/main/scala/org/apache/carbondata/api/CarbonStore.scala
##########
@@ -96,20 +100,31 @@ object CarbonStore {
* Read stage files and return input files
*/
def readStageInput(
+ tableStagePath: String,
stageFiles: Seq[CarbonFile],
status: StageInput.StageStatus): Seq[StageInput] = {
val gson = new Gson()
val output = Collections.synchronizedList(new util.ArrayList[StageInput]())
stageFiles.map { stage =>
- val filePath = stage.getAbsolutePath
- val stream = FileFactory.getDataInputStream(filePath)
+ val filePath = tableStagePath + CarbonCommonConstants.FILE_SEPARATOR + stage.getName
+ var stream: DataInputStream = null
try {
- val stageInput = gson.fromJson(new InputStreamReader(stream), classOf[StageInput])
- stageInput.setCreateTime(stage.getLastModifiedTime)
- stageInput.setStatus(status)
- output.add(stageInput)
+ stream = FileFactory.getDataInputStream(filePath)
+ var retry = READ_FILE_RETRY_TIMES
+ breakable { while (retry > 0) { try {
Review comment:
I have modified code according to your suggestion
##########
File path: integration/spark/src/main/scala/org/apache/carbondata/api/CarbonStore.scala
##########
@@ -96,20 +100,31 @@ object CarbonStore {
* Read stage files and return input files
*/
def readStageInput(
+ tableStagePath: String,
stageFiles: Seq[CarbonFile],
status: StageInput.StageStatus): Seq[StageInput] = {
val gson = new Gson()
val output = Collections.synchronizedList(new util.ArrayList[StageInput]())
stageFiles.map { stage =>
Review comment:
I have modified code according to your suggestion
----------------------------------------------------------------
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