diff --git a/.rubocop.yml b/.rubocop.yml index 7de7a3b..a76ce27 100644 --- a/.rubocop.yml +++ b/.rubocop.yml @@ -1,7 +1,7 @@ AllCops: NewCops: enable - TargetRubyVersion: 3.0 + TargetRubyVersion: 3.2 Exclude: - 'vendor/**/*' - 'spec/**/*' diff --git a/CHANGELOG.md b/CHANGELOG.md index 5290a48..22825a5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,16 @@ All notable changes to this project will be documented in this file. ### Added * Support named and positional bind values through `QueryRequest#params`. +* `ColumnDescriptor` value objects (`Data`) for query columns, with `name` and `type` readers. + +### Changed + +* Require Ruby 3.2 or newer (`ColumnDescriptor` is a `Data` class). CI already only tested 3.2+. + +### Fixed + +* `QueryResult#columns` and `query_all[:columns]` return typed column descriptors again instead of bare names; rows stay keyed by column name. +* A malformed `/query` column line (anything other than an array of `{ "name", "type" }` string objects) now raises `ParseError` instead of passing through untyped. ## [0.8.1](https://github.com/altertable-ai/altertable-lakehouse-ruby/compare/altertable-lakehouse/v0.8.0...altertable-lakehouse/v0.8.1) (2026-10-07) diff --git a/README.md b/README.md index e9e9201..6eb13b1 100644 --- a/README.md +++ b/README.md @@ -124,7 +124,8 @@ result = client.query_all( ) puts result[:metadata] # Hash -puts result[:columns] # Array of columns +puts result[:columns] # Array of ColumnDescriptor (#name, #type) +puts result[:columns].first.type # e.g. "INTEGER" result[:rows].each do |row| puts row end diff --git a/altertable-lakehouse.gemspec b/altertable-lakehouse.gemspec index 7cae9b3..ea56430 100644 --- a/altertable-lakehouse.gemspec +++ b/altertable-lakehouse.gemspec @@ -12,7 +12,7 @@ Gem::Specification.new do |spec| spec.description = "Official Ruby client for Altertable Lakehouse API." spec.homepage = "https://github.com/altertable-ai/altertable-lakehouse-ruby" spec.license = "MIT" - spec.required_ruby_version = ">= 3.0.0" + spec.required_ruby_version = ">= 3.2.0" spec.metadata["homepage_uri"] = spec.homepage spec.metadata["source_code_uri"] = "https://github.com/altertable-ai/altertable-lakehouse-ruby" diff --git a/lib/altertable/lakehouse/adapters.rb b/lib/altertable/lakehouse/adapters.rb index 43ad404..afc1090 100644 --- a/lib/altertable/lakehouse/adapters.rb +++ b/lib/altertable/lakehouse/adapters.rb @@ -146,16 +146,16 @@ def initialize(base_url:, timeout:, open_timeout:, headers: {}, **options) @uri = URI.parse(@base_url) end - def get(path, body: nil, params: {}, headers: {}, &block) # rubocop:disable Lint/UnusedMethodArgument - request(Net::HTTP::Get, path, params: params, headers: headers, &block) + def get(path, body: nil, params: {}, headers: {}, &) # rubocop:disable Lint/UnusedMethodArgument + request(Net::HTTP::Get, path, params: params, headers: headers, &) end - def post(path, body: nil, params: {}, headers: {}, &block) - request(Net::HTTP::Post, path, body: body, params: params, headers: headers, &block) + def post(path, body: nil, params: {}, headers: {}, &) + request(Net::HTTP::Post, path, body: body, params: params, headers: headers, &) end - def delete(path, body: nil, params: {}, headers: {}, &block) # rubocop:disable Lint/UnusedMethodArgument - request(Net::HTTP::Delete, path, params: params, headers: headers, &block) + def delete(path, body: nil, params: {}, headers: {}, &) # rubocop:disable Lint/UnusedMethodArgument + request(Net::HTTP::Delete, path, params: params, headers: headers, &) end private diff --git a/lib/altertable/lakehouse/client.rb b/lib/altertable/lakehouse/client.rb index 173ca0a..709fa59 100644 --- a/lib/altertable/lakehouse/client.rb +++ b/lib/altertable/lakehouse/client.rb @@ -12,7 +12,7 @@ class Client DEFAULT_TIMEOUT = 10 DEFAULT_OPEN_TIMEOUT = 5 - def initialize(username: nil, password: nil, basic_auth_token: nil, base_url: nil, timeout: nil, open_timeout: nil, user_agent: nil, adapter: nil, headers: {}, **options) + def initialize(username: nil, password: nil, basic_auth_token: nil, base_url: nil, timeout: nil, open_timeout: nil, user_agent: nil, adapter: nil, headers: {}, **) # 1. Try passed basic_auth_token # 2. Try passed username/password # 3. Try ENV["ALTERTABLE_BASIC_AUTH_TOKEN"] @@ -40,7 +40,7 @@ def initialize(username: nil, password: nil, basic_auth_token: nil, base_url: ni "User-Agent" => @user_agent } - @adapter = select_adapter(adapter, base_url: @base_url, timeout: @timeout, open_timeout: @open_timeout, headers: default_headers.merge(headers), **options) + @adapter = select_adapter(adapter, base_url: @base_url, timeout: @timeout, open_timeout: @open_timeout, headers: default_headers.merge(headers), **) end # POST /append @@ -59,8 +59,8 @@ def get_task(task_id, headers: {}) end # POST /query (streamed) - def query(statement:, headers: {}, **options) - req_body = Models::QueryRequest.new(statement: statement, **options).to_h.to_json + def query(statement:, headers: {}, **) + req_body = Models::QueryRequest.new(statement: statement, **).to_h.to_json enum = Enumerator.new do |yielder| buffer = "" @@ -77,8 +77,8 @@ def query(statement:, headers: {}, **options) end # POST /query (accumulated) - def query_all(statement:, headers: {}, **options) - result = query(statement: statement, headers: headers, **options) + def query_all(statement:, headers: {}, **) + result = query(statement: statement, headers: headers, **) rows = result.to_a # Accumulate { metadata: result.metadata, @@ -278,17 +278,20 @@ def handle_response(resp) end end + ColumnDescriptor = Data.define(:name, :type) + class QueryResult include Enumerable # metadata: the stream header object (first NDJSON line) - # columns: array of column names or typed column descriptors (second NDJSON line) + # columns: ColumnDescriptor array (second NDJSON line) attr_reader :metadata, :columns def initialize(enum) @enum = enum @metadata = nil @columns = nil + @column_names = nil end def each(&block) @@ -314,10 +317,11 @@ def each(&block) when 0 @metadata = item when 1 - @columns = normalize_columns(item) + @columns = column_descriptors(item) + @column_names = @columns.map(&:name) else - if @columns.is_a?(Array) && item.is_a?(Array) - block.call(@columns.zip(item).to_h) + if item.is_a?(Array) + block.call(@column_names.zip(item).to_h) else block.call(item) end @@ -337,16 +341,30 @@ def stream_error_message(item) message if message.is_a?(String) end - # The query service emits typed column descriptors, such as - # { "name" => "id", "type" => "INTEGER" }. Keep the public result - # shape stable by exposing names and using them as row keys. - def normalize_columns(columns) - return columns unless columns.is_a?(Array) + def column_descriptors(columns) + unless columns.is_a?(Array) + raise column_parse_error("expected an array of column objects, got #{columns.inspect}") + end columns.map do |column| - column.is_a?(Hash) && column.key?("name") ? column["name"] : column + name = column["name"] if column.is_a?(Hash) + type = column["type"] if column.is_a?(Hash) + unless name.is_a?(String) && type.is_a?(String) + raise column_parse_error("expected a { \"name\", \"type\" } column object, got #{column.inspect}") + end + + ColumnDescriptor.new(name: name, type: type) end end + + def column_parse_error(message) + ParseError.new( + "Malformed /query column line: #{message}", + operation: "query", + http_method: "POST", + http_path: "/query" + ) + end end end diff --git a/rbi/altertable/lakehouse.rbi b/rbi/altertable/lakehouse.rbi index e01764e..963f5ff 100644 --- a/rbi/altertable/lakehouse.rbi +++ b/rbi/altertable/lakehouse.rbi @@ -110,7 +110,7 @@ module Altertable sig { params(statement: String, headers: T::Hash[String, String], options: T.untyped).returns(::Altertable::Lakehouse::QueryResult) } def query(statement:, headers: {}, **options); end - sig { params(statement: String, headers: T::Hash[String, String], options: T.untyped).returns(T::Hash[Symbol, T.untyped]).returns(T::Hash[Symbol, T.untyped]) } + sig { params(statement: String, headers: T::Hash[String, String], options: T.untyped).returns(T::Hash[Symbol, T.untyped]) } def query_all(statement:, headers: {}, **options); end sig do @@ -204,13 +204,21 @@ module Altertable def handle_response(resp); end end + class ColumnDescriptor < Data + sig { returns(String) } + def name; end + + sig { returns(String) } + def type; end + end + class QueryResult include Enumerable sig { returns(T.untyped) } def metadata; end - sig { returns(T.untyped) } + sig { returns(T.nilable(T::Array[::Altertable::Lakehouse::ColumnDescriptor])) } def columns; end sig { params(enum: T.untyped).void } diff --git a/sig/altertable/lakehouse/client.rbs b/sig/altertable/lakehouse/client.rbs index 169c8ce..3e2bf02 100644 --- a/sig/altertable/lakehouse/client.rbs +++ b/sig/altertable/lakehouse/client.rbs @@ -103,18 +103,34 @@ module Altertable def handle_response: (Adapters::Response resp) -> untyped end + class ColumnDescriptor < ::Data + attr_reader name: String + attr_reader type: String + + def self.new: (name: String, type: String) -> instance + | (String name, String type) -> instance + def to_h: () -> { name: String, type: String } + end + class QueryResult include Enumerable @enum: Enumerator[untyped, untyped] @metadata: untyped - @columns: untyped + @columns: Array[ColumnDescriptor]? + @column_names: Array[String]? attr_reader metadata: untyped - attr_reader columns: untyped + attr_reader columns: Array[ColumnDescriptor]? def initialize: (Enumerator[untyped, untyped] enum) -> void def each: () { (untyped) -> void } -> void + + private + + def stream_error_message: (untyped item) -> String? + def column_descriptors: (untyped columns) -> Array[ColumnDescriptor] + def column_parse_error: (String message) -> ParseError end end end diff --git a/spec/altertable/lakehouse_spec.rb b/spec/altertable/lakehouse_spec.rb index dbbca5f..fd3ebaf 100644 --- a/spec/altertable/lakehouse_spec.rb +++ b/spec/altertable/lakehouse_spec.rb @@ -143,7 +143,7 @@ # ── #query ─────────────────────────────────────────────────────────────────── describe "#query (streaming)" do - it "normalizes typed column descriptors to column names" do + it "exposes typed column descriptors and keys rows by column name" do stream = [ { "statement" => "SELECT 1 AS id" }, [{ "name" => "id", "type" => "INTEGER" }], @@ -152,16 +152,50 @@ result = Altertable::Lakehouse::QueryResult.new(stream.each) expect(result.to_a).to eq([{ "id" => 1 }]) - expect(result.columns).to eq(["id"]) + expect(result.columns).to eq([Altertable::Lakehouse::ColumnDescriptor.new(name: "id", type: "INTEGER")]) + + column = result.columns.first + expect(column).to be_a(Altertable::Lakehouse::ColumnDescriptor) + expect(column).to be_frozen + expect(column.name).to eq("id") + expect(column.type).to eq("INTEGER") + end + + it "keeps only name and type from column objects" do + stream = [ + { "statement" => "SELECT 1 AS id" }, + [{ "name" => "id", "type" => "INTEGER", "nullable" => false }] + ] + result = Altertable::Lakehouse::QueryResult.new(stream.each) + result.to_a + + expect(result.columns).to eq([Altertable::Lakehouse::ColumnDescriptor.new(name: "id", type: "INTEGER")]) + end + + [ + ["a bare column name", ["id"]], + ["a column object without type", [{ "name" => "id" }]], + ["a column object with a non-string name", [{ "name" => 1, "type" => "INTEGER" }]], + ["a non-array column line", { "name" => "id", "type" => "INTEGER" }] + ].each do |description, columns_line| + it "raises ParseError on #{description}" do + stream = [{ "statement" => "SELECT 1 AS id" }, columns_line, [1]] + result = Altertable::Lakehouse::QueryResult.new(stream.each) + + expect { result.to_a }.to raise_error(Altertable::Lakehouse::ParseError, /Malformed \/query column line/) { |error| + expect(error.operation).to eq("query") + expect(error.http_path).to eq("/query") + } + end end - it "parses the header, column names and data rows from a SELECT" do + it "parses the header, typed columns and data rows from a SELECT" do result = client.query(statement: "SELECT 42 AS answer") rows = result.to_a expect(result.metadata).to be_a(Hash) expect(result.metadata["statement"]).to eq("SELECT 42 AS answer") - expect(result.columns).to eq(["answer"]) + expect(result.columns).to eq([Altertable::Lakehouse::ColumnDescriptor.new(name: "answer", type: "INTEGER")]) expect(rows.length).to eq(1) expect(rows[0]).to eq({ "answer" => 42 }) end @@ -172,7 +206,10 @@ ) rows = result.to_a - expect(result.columns).to eq(["id", "val"]) + expect(result.columns).to eq([ + Altertable::Lakehouse::ColumnDescriptor.new(name: "id", type: "INTEGER"), + Altertable::Lakehouse::ColumnDescriptor.new(name: "val", type: "VARCHAR") + ]) expect(rows.length).to eq(3) expect(rows[0]).to eq({ "id" => 1, "val" => "a" }) expect(rows[1]).to eq({ "id" => 2, "val" => "b" }) @@ -186,7 +223,7 @@ rows = result.to_a # Column metadata is present even when the query produces no rows. - expect(result.columns).to eq(["n"]) + expect(result.columns).to eq([Altertable::Lakehouse::ColumnDescriptor.new(name: "n", type: "INTEGER")]) expect(rows).to be_empty end @@ -224,7 +261,7 @@ } expect(yielded).to eq([{ "id" => 1 }]) - expect(result.columns).to eq(["id"]) + expect(result.columns).to eq([Altertable::Lakehouse::ColumnDescriptor.new(name: "id", type: "INTEGER")]) expect(result.metadata).to include("statement" => "SELECT id") end end @@ -239,7 +276,11 @@ expect(result[:metadata]).to be_a(Hash) expect(result[:metadata]["statement"]).to eq("SELECT 7 AS x, 'hello' AS y") - expect(result[:columns]).to eq(["x", "y"]) + expect(result[:columns]).to eq([ + Altertable::Lakehouse::ColumnDescriptor.new(name: "x", type: "INTEGER"), + Altertable::Lakehouse::ColumnDescriptor.new(name: "y", type: "VARCHAR") + ]) + expect(result[:columns].map(&:type)).to eq(["INTEGER", "VARCHAR"]) expect(result[:rows]).to eq([{ "x" => 7, "y" => "hello" }]) end @@ -288,7 +329,7 @@ it "forwards per-request headers on #query" do stream_response = Altertable::Lakehouse::Adapters::Response.new( 200, - "{\"statement\":\"SELECT 1\"}\n[\"n\"]\n[1]\n", + "{\"statement\":\"SELECT 1\"}\n[{\"name\":\"n\",\"type\":\"INTEGER\"}]\n[1]\n", {} ) @@ -296,7 +337,7 @@ "/query", hash_including(headers: hash_including("Content-Type" => "application/json", "X-Trace" => "trace-1")) ).and_yield("{\"statement\":\"SELECT 1\"}\n", nil) - .and_yield("[\"n\"]\n", nil) + .and_yield("[{\"name\":\"n\",\"type\":\"INTEGER\"}]\n", nil) .and_yield("[1]\n", nil) .and_return(stream_response) @@ -365,7 +406,10 @@ ) result = client.query_all(statement: "SELECT * FROM #{table_name} ORDER BY id") - expect(result[:columns]).to eq(["id", "score"]) + expect(result[:columns]).to eq([ + Altertable::Lakehouse::ColumnDescriptor.new(name: "id", type: "BIGINT"), + Altertable::Lakehouse::ColumnDescriptor.new(name: "score", type: "BIGINT") + ]) expect(result[:rows]).to eq([ { "id" => 10, "score" => 100 }, { "id" => 20, "score" => 200 } diff --git a/specs b/specs index c55685e..7f11faa 160000 --- a/specs +++ b/specs @@ -1 +1 @@ -Subproject commit c55685efe831466bf8ceb6953ecb13cbb19780ab +Subproject commit 7f11faacc087aac6cf30ac2cf6ad7b7b73d6d038