Class: Purplelight::Manifest

Inherits:
Object
  • Object
show all
Defined in:
lib/purplelight/manifest.rb

Overview

Manifest persists snapshot run metadata and progress to a JSON file.

It records configuration, partition checkpoints, and per-part byte/row counts so interrupted runs can resume safely and completed runs are reproducible. Methods are thread-safe where mutation occurs.

Constant Summary collapse

DEFAULT_VERSION = 
2

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(path:, data: nil) ⇒ Manifest

Returns a new instance of Manifest.



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/purplelight/manifest.rb', line 26

def initialize(path:, data: nil)
  @path = path
  @directory = File.dirname(path)
  @temporary_path = "#{path}.tmp"
  FileUtils.mkdir_p(@directory)
  @data = data || {
    'version' => DEFAULT_VERSION,
    'run_id' => SecureRandom.uuid,
    'created_at' => Time.now.utc.iso8601,
    'collection' => nil,
    'format' => nil,
    'compression' => nil,
    'query_digest' => nil,
    'options' => {},
    'partition_filters' => nil,
    'parts' => [],
    'partitions' => []
  }
  @mutex = Mutex.new
  @last_save_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
end

Instance Attribute Details

#data ⇒ Object (readonly)

Returns the value of attribute data.



19
20
21
 
# File 'lib/purplelight/manifest.rb', line 19

def data
  @data
end

#path ⇒ Object (readonly)

Returns the value of attribute path.



19
20
21
 
# File 'lib/purplelight/manifest.rb', line 19

def path
  @path
end

Class Method Details

.load(path) ⇒ Object 



48
49
50
51
 
# File 'lib/purplelight/manifest.rb', line 48

def self.load(path)
  data = JSON.parse(File.read(path))
  new(path: path, data: data)
end

.query_digest(query, projection) ⇒ Object 



21
22
23
24
 
# File 'lib/purplelight/manifest.rb', line 21

def self.query_digest(query, projection)
  payload = { query: query, projection: projection }
  Digest::SHA256.hexdigest(JSON.generate(payload))
end

Instance Method Details

#add_progress_to_part!(index:, rows_delta:, bytes_delta:) ⇒ Object 



114
115
116
117
118
119
120
121
 
# File 'lib/purplelight/manifest.rb', line 114

def add_progress_to_part!(index:, rows_delta:, bytes_delta:)
  @mutex.synchronize do
    part = @data['parts'][index]
    part['rows'] += rows_delta
    part['bytes'] += bytes_delta
    save_maybe!
  end
end

#compatible_with?(collection:, format:, compression:, query_digest:) ⇒ Boolean

Returns:

  • (Boolean) 


70
71
72
73
74
75
76
77
 
# File 'lib/purplelight/manifest.rb', line 70

def compatible_with?(collection:, format:, compression:, query_digest:)
  return false unless @data['version'] == DEFAULT_VERSION

  @data['collection'] == collection &&
    @data['format'] == format.to_s &&
    @data['compression'] == compression.to_s &&
    @data['query_digest'] == query_digest
end

#complete_part!(index:, checksum: nil) ⇒ Object 



123
124
125
126
127
128
129
130
 
# File 'lib/purplelight/manifest.rb', line 123

def complete_part!(index:, checksum: nil)
  @mutex.synchronize do
    part = @data['parts'][index]
    part['complete'] = true
    part['checksum'] = checksum
    save!
  end
end

#configure!(collection:, format:, compression:, query_digest:, options: {}, partition_count: nil) ⇒ Object 



58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/purplelight/manifest.rb', line 58

def configure!(collection:, format:, compression:, query_digest:, options: {}, partition_count: nil)
  @mutex.synchronize do
    @data['collection'] = collection
    @data['format'] = format.to_s
    @data['compression'] = compression.to_s
    @data['query_digest'] = query_digest
    @data['options'] = options
    normalize_partition_data!(partition_count) if partition_count
    save!
  end
end

#configure_partition_filters!(filters) ⇒ Object 



158
159
160
161
162
 
# File 'lib/purplelight/manifest.rb', line 158

def configure_partition_filters!(filters)
  @mutex.synchronize do
    @data['partition_filters'] = filters.as_extended_json
  end
end

#ensure_partitions!(count) ⇒ Object 



79
80
81
82
83
84
85
86
 
# File 'lib/purplelight/manifest.rb', line 79

def ensure_partitions!(count)
  @mutex.synchronize do
    if @data['partitions'].empty?
      normalize_partition_data!(count)
      save!
    end
  end
end

#mark_partition_complete!(index) ⇒ Object 



96
97
98
99
100
101
102
 
# File 'lib/purplelight/manifest.rb', line 96

def mark_partition_complete!(index)
  @mutex.synchronize do
    part = @data['partitions'][index]
    part['completed'] = true
    save!
  end
end

#open_part!(path) ⇒ Object 



104
105
106
107
108
109
110
111
112
 
# File 'lib/purplelight/manifest.rb', line 104

def open_part!(path)
  @mutex.synchronize do
    idx = @data['parts'].size
    @data['parts'] << { 'index' => idx, 'path' => path, 'bytes' => 0, 'rows' => 0, 'complete' => false,
                        'checksum' => nil }
    save!
    idx
  end
end

#partition_checkpoint(index) ⇒ Object 



140
141
142
143
 
# File 'lib/purplelight/manifest.rb', line 140

def partition_checkpoint(index)
  raw_checkpoint = @data['partitions'][index]&.fetch('last_id_exclusive', nil)
  BSON::ExtJSON.parse_obj(raw_checkpoint) if raw_checkpoint
end

#partition_filters ⇒ Object 



145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'lib/purplelight/manifest.rb', line 145

def partition_filters
  raw_filters = @data['partition_filters']
  return unless raw_filters

  BSON::ExtJSON.parse_obj(raw_filters).map do |filter_spec|
    {
      filter: filter_spec[:filter] || filter_spec['filter'],
      sort: filter_spec[:sort] || filter_spec['sort'],
      hint: filter_spec[:hint] || filter_spec['hint']
    }
  end
end

#partitions ⇒ Object 



136
137
138
 
# File 'lib/purplelight/manifest.rb', line 136

def partitions
  @data['partitions']
end

#parts ⇒ Object 



132
133
134
 
# File 'lib/purplelight/manifest.rb', line 132

def parts
  @data['parts']
end

#save! ⇒ Object 



53
54
55
56
 
# File 'lib/purplelight/manifest.rb', line 53

def save!
  File.write(@temporary_path, JSON.pretty_generate(@data))
  File.rename(@temporary_path, path)
end

#update_partition_checkpoint!(index, last_id_exclusive) ⇒ Object 



88
89
90
91
92
93
94
 
# File 'lib/purplelight/manifest.rb', line 88

def update_partition_checkpoint!(index, last_id_exclusive)
  @mutex.synchronize do
    part = @data['partitions'][index]
    part['last_id_exclusive'] = serialize_checkpoint(last_id_exclusive)
    save_maybe!
  end
end