Use new data-stream-list server script data structure

- add to report method saves new large-hash parameter
- delete from report updated
- split download updated
- resume operation (both full and split files) updated
- warn conflicts updated
- update dataset operation updated
- rename operation (download and ingest)
- download discard changes updated
- stage feature works properly (add/remove local changes)
- ingestion reportUpToDate updated
- split ingestion updated
- dataset utils remove hardcoded url
- report up to date method updated

ebulk-data/embulk-wendelin-dataset-tool/lib/embulk/input/fif.rb

@@ -138,9 +138,9 @@ module Embulk
 		@logger.error("Your current dataset is outdated. Please, run a download to update it before ingest your changes.", print=TRUE)
 		puts
 		@logger.abortExecution(error=FALSE)
-	      end
-	    end
-	  end
+        end
+      end
+    end
 
 	  @logger.info("Supplier: #{task['supplier']}")
 	  @logger.info("Dataset name: #{task['data_set']}")
@@ -215,7 +215,7 @@ module Embulk
 	super
 	@supplier = task['supplier']
 	@dataset = task['data_set']
-	@chunk_size = task['chunk_size']
+	@chunk_size = DatasetUtils::CHUNK_SIZE
 	@data_set_directory = task['data_set_directory']
 	@logger = LogManager.instance()
 	@dataset_utils = DatasetUtils.new(@data_set_directory)
@@ -239,13 +239,19 @@ module Embulk
 	  filename, extension, reference = @dataset_utils.getPathInfo(path, @dataset)
 	  operation = rename ? DatasetUtils::RENAME : DatasetUtils::INGESTION
 	  @dataset_utils.saveCurrentOperation(operation, reference, new_reference)
-	  resume_split = @dataset_utils.splitOperationFileExist(reference) ? @dataset_utils.getLastSplitOperation(operation, reference, hash, @chunk_size) : 0
+	  resume_split, large_hash = 0, ""
+	  if @dataset_utils.splitOperationFileExist(reference)
+	    resume_split, large_hash = @dataset_utils.getLastSplitOperation(operation, reference, hash, @chunk_size, large_hash=TRUE)
+    end
 	  each_chunk(path, filename, extension, size, hash, schema[1..-1].map{|elm| elm.name}, @chunk_size, delete, new_reference, resume_split) do |entry|
 	    @dataset_utils.createSplitOperationControlFile(reference) if split
+	    large_hash += entry[8]
+	    #no need to send large hash to server
+	    entry.pop()
 	    @page_builder.add(entry)
 	    if ! delete && ! rename && entry[5] != ""
 	      split = TRUE
-	      @dataset_utils.saveSplitOperation(operation, reference, entry[5], hash, @chunk_size)
+	      @dataset_utils.saveSplitOperation(operation, reference, entry[5], hash, @chunk_size, large_hash)
 	    end
 	  end
 	  @page_builder.finish
@@ -269,7 +275,7 @@ module Embulk
 	    end
 	  else
 	    if @dataset_utils.reportFileExist()
-	      @dataset_utils.addToReport(reference, return_value, size, hash, task['data_set'], new_reference)
+	      @dataset_utils.addToReport(reference, return_value, size, hash, large_hash, task['data_set'], new_reference)
 	    end
 	  end
 	end
@@ -282,11 +288,11 @@ module Embulk
       def each_chunk(path, filename, extension, size, hash, fields, chunk_size=DatasetUtils::CHUNK_SIZE, delete=FALSE, new_reference=FALSE, resume_split=0)
 	if delete
 	  File.delete(path) if File.exist?(path)
-	  values = [@supplier, @dataset, filename, extension, "", DatasetUtils::DELETE, "", ""]
+	  values = [@supplier, @dataset, filename, extension, "", DatasetUtils::DELETE, "", "", ""]
 	  yield(values)
 	elsif new_reference
 	  File.delete(path) if File.exist?(path)
-	  values = [@supplier, @dataset, filename, extension, new_reference, DatasetUtils::RENAME, "", ""]
+	  values = [@supplier, @dataset, filename, extension, new_reference, DatasetUtils::RENAME, "", "", ""]
 	  yield(values)
 	else
 	  file_object = File.open(path, "rb")
@@ -297,7 +303,7 @@ module Embulk
 	      data = next_byte
 	      if not next_byte
 		if first # this means this is an empty file
-		  values = [@supplier, @dataset, filename, extension, "", "", size, hash]
+		  values = [@supplier, @dataset, filename, extension, "", "", size, hash, hash]
 		  yield(values)
 		end
 		break
@@ -320,7 +326,8 @@ module Embulk
 		eof = npart.to_s.rjust(3, "0")
 	      end
 	      content = Base64.encode64(data)
-	      values = [@supplier, @dataset, filename, extension, content, eof, size, hash]
+	      chunk_hash = @dataset_utils.getHashFromChunk(data)
+	      values = [@supplier, @dataset, filename, extension, content, eof, size, hash, chunk_hash]
 	      first = FALSE
 	      yield(values)
 	  end

ebulk-data/embulk-wendelin-dataset-tool/lib/embulk/input/wendelin.rb

@@ -242,7 +242,7 @@ module Embulk
       def initialize(task, schema, index, page_builder)
 	super
 	@data_set = task['data_set']
-	@chunk_size = task['chunk_size']
+	@chunk_size = DatasetUtils::CHUNK_SIZE
 	@data_set_directory = task['data_set_directory']
 	@wendelin = WendelinClient.new(task['erp5_url'], task['user'], task['password'])
 	@logger = LogManager.instance()
@@ -250,46 +250,52 @@ module Embulk
       end
 
       def run
-	data_stream = task['data_streams'][@index]
-	id = data_stream["id"]
-	reference = data_stream["reference"]
-	size = data_stream["size"]
-	hash = data_stream["hash"]
-	renamed = data_stream["status"] == DatasetUtils::STATUS_RENAMED
-	deleted = hash.to_s == DatasetUtils::DELETE
+	remote_file = task['data_streams'][@index]
+	reference = remote_file["reference"]
+	size = remote_file["full-size"]
+	large_hash = remote_file["large-hash"]
+	data_stream_chunk_list = remote_file["data-stream-list"]
+	renamed = remote_file["status"] == DatasetUtils::STATUS_RENAMED
+	deleted = large_hash.to_s == DatasetUtils::DELETE
 	begin
 	  if deleted
 	    entry = [reference, "", @data_set, DatasetUtils::DELETE, renamed]
 	    page_builder.add(entry)
 	  elsif renamed
-	    new_reference = data_stream["new_reference"]
+	    new_reference = remote_file["new_reference"]
 	    entry = [reference, new_reference, @data_set, TRUE, renamed]
   	    page_builder.add(entry)
 	  else
-	    @logger.info("Discarding local change on '#{data_stream["path"]}'", print=TRUE) if task['discard_changes']
+	    @logger.info("Discarding local change on '#{remote_file["path"]}'", print=TRUE) if task['discard_changes']
 	    @logger.info("Getting content from remote #{reference}", print=TRUE)
 	    @logger.info("Downloading...", print=TRUE)
-	    resume_split = @dataset_utils.splitOperationFileExist(reference) ? @dataset_utils.getLastSplitOperation(DatasetUtils::DOWNLOAD, reference, hash, @chunk_size) : 0
-	    n_chunk = resume_split == 0 ? 0 : resume_split+1
-	    split = n_chunk > 0
+	    resume_split = @dataset_utils.splitOperationFileExist(reference) ? @dataset_utils.getLastSplitOperation(DatasetUtils::DOWNLOAD, reference, large_hash, @chunk_size) : 0
+      n_chunk = resume_split == 0 ? 0 : resume_split+1
+      split = n_chunk > 0
 	    @logger.info("Resuming interrupted split download...", print=TRUE) if split
-	    @wendelin.eachDataStreamContentChunk(id, @chunk_size, n_chunk) do |chunk|
-	      content = chunk.nil? || chunk.empty? ? "" : Base64.encode64(chunk)
-	      begin_of_file = n_chunk == 0
-	      split = n_chunk > 0
-	      @dataset_utils.createSplitOperationControlFile(reference) if split
-	      entry = [reference, content, @data_set, begin_of_file, renamed]
-	      page_builder.add(entry)
-	      @dataset_utils.saveSplitOperation(DatasetUtils::DOWNLOAD, reference, n_chunk, hash, @chunk_size) if split
-	      n_chunk += 1
-	    end
+      data_stream_chunk_list.each_with_index do |data_stream_chunk, index|
+        #skip datastreams/chunks already downloaded
+        if n_chunk == index
+          content = ""
+          @wendelin.eachDataStreamContentChunk(data_stream_chunk["id"], @chunk_size, 0) do |chunk|
+            content = chunk.nil? || chunk.empty? ? "" : Base64.encode64(chunk)
+          end
+          begin_of_file = n_chunk == 0
+          split = n_chunk > 0
+          @dataset_utils.createSplitOperationControlFile(reference) if split
+          entry = [reference, content, @data_set, begin_of_file, renamed]
+          page_builder.add(entry)
+          @dataset_utils.saveSplitOperation(DatasetUtils::DOWNLOAD, reference, n_chunk, large_hash, @chunk_size) if split
+          n_chunk += 1
+        end
+      end
 	  end
 	  page_builder.finish
 	  @dataset_utils.deleteSplitOperationFile(reference) if split
 	rescue java.lang.OutOfMemoryError
 	  @logger.logOutOfMemoryError(reference)
 	  return_value = DatasetUtils::RUN_ABORTED
-	rescue Exception => e  
+	rescue Exception => e
 	  @logger.error(e.to_s, print=TRUE)
 	  @logger.error(e.backtrace)
 	  puts "[INFO] For more detailed information, please refer to the log file: " + @logger.getLogPath()
@@ -302,7 +308,9 @@ module Embulk
 	  if deleted
 	    @dataset_utils.deleteFromReport(reference, return_value)
 	  else
-	    @dataset_utils.addToReport(reference, return_value, size, hash, task['data_set'], new_reference)
+      file_path = renamed ? @dataset_utils.referenceToPath(new_reference, @data_set_directory, @data_set) : @dataset_utils.referenceToPath(reference, @data_set_directory, @data_set)
+      hash = @dataset_utils.getHash(file_path).to_s
+	    @dataset_utils.addToReport(reference, return_value, size, hash, large_hash, task['data_set'], new_reference)
 	  end
 	end
 	return {return_value => reference}