Class: DataDrain::Record Abstract

Inherits:
Object
  • Object
show all
Includes:
ActiveModel::Attributes, ActiveModel::Model
Defined in:
lib/data_drain/record.rb

Overview

This class is abstract.

Subclasifica este modelo para cada tabla archivada.

Clase base que actúa como un ORM (Object-Relational Mapper) de solo lectura y purga para interactuar con el Data Lake en formato Parquet utilizando DuckDB.

Examples:

class ArchivedVersion < DataDrain::Record
  self.folder_name = 'versions'
  self.partition_keys = [:year, :month, :isp_id]
  attribute :event, :string
end

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.connection ⇒ DuckDB::Connection

Retorna la conexión persistente a DuckDB en memoria para el hilo (Thread) actual. Esto previene tener que recargar extensiones (como httpfs) en cada consulta.

Returns:

  • (DuckDB::Connection) —

    Conexión activa a DuckDB.



29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/data_drain/record.rb', line 29

def self.connection
  Thread.current[:data_drain_duckdb_conn] ||= begin
    db = DuckDB::Database.open(":memory:")
    conn = db.connect

    config = DataDrain.configuration
    conn.query("SET max_memory='#{config.limit_ram}';") if config.limit_ram.present?
    conn.query("SET temp_directory='#{config.tmp_directory}'") if config.tmp_directory.present?

    DataDrain::Storage.adapter.setup_duckdb(conn)
    conn
  end
end

.destroy_all(**partitions) ⇒ Integer

Elimina físicamente los directorios o prefijos de S3.

Parameters:

  • partitions (Hash) —

    Particiones a eliminar.

Returns:

  • (Integer) —

    Cantidad de particiones físicas eliminadas.



86
87
88
89
90
91
 
# File 'lib/data_drain/record.rb', line 86

def self.destroy_all(**partitions)
  adapter = DataDrain::Storage.adapter
  DataDrain.configuration.logger.info "[DataDrain] 🗑️ Ejecutando destroy_all en #{folder_name} con: #{partitions.inspect}"

  adapter.destroy_partitions(bucket, folder_name, partition_keys, partitions)
end

.find(id, **partitions) ⇒ DataDrain::Record?

Busca un registro específico por su ID. Implementa sanitización básica para prevenir Inyección SQL.

Parameters:

  • id (String, Integer) —

    Identificador único del registro.

  • partitions (Hash) —

    Pares clave-valor de las particiones donde buscar.

Returns:



67
68
69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/data_drain/record.rb', line 67

def self.find(id, **partitions)
  path = build_query_path(partitions)
  # Sanitización básica: duplicar comillas simples para anular escapes SQL
  safe_id = id.to_s.gsub("'", "''")

  sql = <<~SQL
    SELECT #{attribute_names.join(', ')}
    FROM read_parquet('#{path}')
    WHERE id = '#{safe_id}'
    LIMIT 1
  SQL

  execute_and_instantiate(sql, attribute_names).first
end

.where(limit: 50, **partitions) ⇒ Array<DataDrain::Record>

Consulta registros en el Data Lake filtrando por claves de partición.

Parameters:

  • limit (Integer) (defaults to: 50) —

    Cantidad máxima de registros a retornar.

  • partitions (Hash) —

    Pares clave-valor correspondientes a las particiones.

Returns:



48
49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/data_drain/record.rb', line 48

def self.where(limit: 50, **partitions)
  path = build_query_path(partitions)

  sql = <<~SQL
    SELECT #{attribute_names.join(', ')}
    FROM read_parquet('#{path}')
    ORDER BY created_at DESC
    LIMIT #{limit}
  SQL

  execute_and_instantiate(sql, attribute_names)
end

Instance Method Details

#inspect ⇒ String

Returns Representación legible en consola.

Returns:

  • (String) —

    Representación legible en consola.



94
95
96
97
98
99
100
 
# File 'lib/data_drain/record.rb', line 94

def inspect
  inspection = attributes.map do |name, value|
    "#{name}: #{value.nil? ? 'nil' : value.inspect}"
  end.compact.join(", ")

  "#<#{self.class} #{inspection}>"
end