Class: GroongaDelta::LocalWriter

Inherits:
Object
  • Object
show all
Defined in:
lib/groonga-delta/local-writer.rb

Instance Method Summary collapse

Constructor Details

#initialize(config) ⇒ LocalWriter

Returns a new instance of LocalWriter.



24
25
26
27
28
 
# File 'lib/groonga-delta/local-writer.rb', line 24

def initialize(config)
  @config = config
  @logger = @config.logger
  @dir = @config.delta_dir
end

Instance Method Details

#write_deletes(table, keys) ⇒ Object 



73
74
75
76
77
78
79
80
81
82
 
# File 'lib/groonga-delta/local-writer.rb', line 73

def write_deletes(table, keys)
  write_data(table, "delete", ".grn") do |output|
    delete = Groonga::Command::Delete.new
    delete[:table] = table
    keys.each do |key|
      delete[:key] = format_key(key)
      output.puts(delete.to_command_format)
    end
  end
end

#write_schema(command) ⇒ Object 



84
85
86
87
88
 
# File 'lib/groonga-delta/local-writer.rb', line 84

def write_schema(command)
  write_entry("schema", ".grn") do |output|
    output.puts(command.to_command_format)
  end
end

#write_upserts(table, records, packed: false) ⇒ Object 



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
 
# File 'lib/groonga-delta/local-writer.rb', line 30

def write_upserts(table, records, packed: false)
  if records.is_a?(Arrow::Table)
    write_data(table,
               "upsert",
               ".parquet",
               packed: packed,
               open_output: false) do |output|
      records.save(output, format: :parquet)
    end
  else
    write_data(table, "upsert", ".grn", packed: packed) do |output|
      first_record = true
      records.each do |record|
        if first_record
          output.puts("load --table #{table}")
          output.print("[")
          first_record = false
        else
          output.print(",")
        end
        output.puts
        json = "{"
        record.each_with_index do |(key, value), i|
          json << "," unless i.zero?
          json << "#{key.to_s.to_json}:"
          case value
          when Time
            json << value.dup.localtime.strftime("%Y-%m-%d %H:%M:%S").to_json
          else
            json << value.to_json
          end
        end
        json << "}"
        output.print(json)
      end
      unless first_record
        output.puts()
        output.puts("]")
      end
    end
  end
end