Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,12 @@ All notable changes to this project will be documented in this file.
### Added

* Support named and positional bind values through `QueryRequest#params`.
* Typed `column.name` and `column.type` readers on query columns. Columns are still hashes, so `column["name"]` and `column["type"]` keep working.

### Fixed

* `QueryResult#columns` and `query_all[:columns]` return the typed column descriptors (`{ "name", "type" }`) 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)

Expand Down
3 changes: 2 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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" => String, "type" => String })
puts result[:columns].first.type # e.g. "INTEGER"; also result[:columns].first["type"]
result[:rows].each do |row|
puts row
end
Expand Down
46 changes: 36 additions & 10 deletions lib/altertable/lakehouse/client.rb
Original file line number Diff line number Diff line change
Expand Up @@ -278,17 +278,28 @@ def handle_response(resp)
end
end

class ColumnDescriptor < Hash
def name
self["name"]
end

def type
self["type"]
end
end

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)
Expand All @@ -314,10 +325,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
Expand All @@ -337,16 +349,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["name" => name, "type" => type]

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wouldn't even inherit from Hash it it's to only use it like this

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes I was afraid of silent re-serialization breaking, but it's probably overkill to be backward compatible for this

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
Expand Down
18 changes: 16 additions & 2 deletions rbi/altertable/lakehouse.rbi
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -204,13 +204,27 @@ module Altertable
def handle_response(resp); end
end

class ColumnDescriptor < Hash
extend T::Generic

K = type_member { { fixed: String } }
V = type_member { { fixed: String } }
Elem = type_member { { fixed: [String, String] } }

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 }
Expand Down
16 changes: 14 additions & 2 deletions sig/altertable/lakehouse/client.rbs
Original file line number Diff line number Diff line change
Expand Up @@ -103,18 +103,30 @@ module Altertable
def handle_response: (Adapters::Response resp) -> untyped
end

class ColumnDescriptor < ::Hash[String, String]
def name: () -> String
def 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
67 changes: 56 additions & 11 deletions spec/altertable/lakehouse_spec.rb
Original file line number Diff line number Diff line change
Expand Up @@ -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" }],
Expand All @@ -152,16 +152,51 @@
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([{ "name" => "id", "type" => "INTEGER" }])

column = result.columns.first
expect(column).to be_a(Altertable::Lakehouse::ColumnDescriptor)
expect(column).to be_a(Hash)
expect(column.name).to eq("id")
expect(column.type).to eq("INTEGER")
expect(column["name"]).to eq("id")
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([{ "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([{ "name" => "answer", "type" => "INTEGER" }])
expect(rows.length).to eq(1)
expect(rows[0]).to eq({ "answer" => 42 })
end
Expand All @@ -172,7 +207,10 @@
)
rows = result.to_a

expect(result.columns).to eq(["id", "val"])
expect(result.columns).to eq([
{ "name" => "id", "type" => "INTEGER" },
{ "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" })
Expand All @@ -186,7 +224,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([{ "name" => "n", "type" => "INTEGER" }])
expect(rows).to be_empty
end

Expand Down Expand Up @@ -224,7 +262,7 @@
}

expect(yielded).to eq([{ "id" => 1 }])
expect(result.columns).to eq(["id"])
expect(result.columns).to eq([{ "name" => "id", "type" => "INTEGER" }])
expect(result.metadata).to include("statement" => "SELECT id")
end
end
Expand All @@ -239,7 +277,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([
{ "name" => "x", "type" => "INTEGER" },
{ "name" => "y", "type" => "VARCHAR" }
])
expect(result[:columns].map(&:type)).to eq(["INTEGER", "VARCHAR"])
expect(result[:rows]).to eq([{ "x" => 7, "y" => "hello" }])
end

Expand Down Expand Up @@ -288,15 +330,15 @@
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",
{}
)

expect(adapter).to receive(:post).with(
"/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)

Expand Down Expand Up @@ -365,7 +407,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([
{ "name" => "id", "type" => "BIGINT" },
{ "name" => "score", "type" => "BIGINT" }
])
expect(result[:rows]).to eq([
{ "id" => 10, "score" => 100 },
{ "id" => 20, "score" => 200 }
Expand Down
Loading