Backport engine-side changes (#244)

comnik · web-flow · commit d467c8460580 · 2026-04-10T23:57:46.000+02:00
diff --git a/sdks/julia/LogicalQueryProtocol.jl/src/properties.jl b/sdks/julia/LogicalQueryProtocol.jl/src/properties.jl
@@ -43,11 +43,22 @@ function is_write_only(epoch::Epoch)
 end
 
 """
-    read_transaction(path::String)::Transaction
+    read_lqp(path::String)::Transaction
 
-Utility function that reads the transaction from the given `path`.
+Utility function that reads the transaction from the given text `.lqp` file at `path`.
 """
-function read_transaction(path::String)
+function read_lqp(path::String)
+    contents = Base.read(path, String)
+    txn, _ = Parser.parse(contents)
+    return txn
+end
+
+"""
+    read_bin(path::String)::Transaction
+
+Utility function that reads the transaction from the given binary protobuf file at `path`.
+"""
+function read_bin(path::String)
     bytes = IOBuffer(read(path))
     return ProtoBuf.decode(ProtoBuf.ProtoDecoder(bytes), Transaction)
 end
@@ -98,6 +109,61 @@ function is_supported_decimal_bits(bits::Integer)
     return bits in [8, 16, 32, 64, 128]
 end
 
+"""
+    collect_demanded_relations(epoch::Epoch)::Set{LQPRelationId}
+
+Collect the set of relation IDs that are demanded by the reads of an epoch.
+"""
+function collect_demanded_relations(epoch::Epoch)
+    ids = Set{LQPRelationId}()
+    for read in epoch.reads
+        _collect_read_ids!(ids, read)
+    end
+    return ids
+end
+
+function _collect_read_ids!(ids::Set{LQPRelationId}, read::Read)
+    isnothing(read.read_type) && return nothing
+    if read.read_type.name == :demand
+        _collect_read_ids!(ids, read.read_type[]::Demand)
+    elseif read.read_type.name == :output
+        _collect_read_ids!(ids, read.read_type[]::Output)
+    elseif read.read_type.name == :abort
+        _collect_read_ids!(ids, read.read_type[]::Abort)
+    elseif read.read_type.name == :var"#export"
+        _collect_read_ids!(ids, read.read_type[]::Export)
+    else
+        @assert false
+    end
+    return nothing
+end
+function _collect_read_ids!(ids::Set{LQPRelationId}, demand::Demand)
+    !isnothing(demand.relation_id) && push!(ids, persistent_id(demand.relation_id))
+    return nothing
+end
+function _collect_read_ids!(ids::Set{LQPRelationId}, output::Output)
+    !isnothing(output.relation_id) && push!(ids, persistent_id(output.relation_id))
+    return nothing
+end
+function _collect_read_ids!(ids::Set{LQPRelationId}, abort::Abort)
+    !isnothing(abort.relation_id) && push!(ids, persistent_id(abort.relation_id))
+    return nothing
+end
+function _collect_read_ids!(ids::Set{LQPRelationId}, _export::Export)
+    isnothing(_export.export_config) && return nothing
+    config = _export.export_config
+    if config.name == :csv_config
+        csv_config = config[]::ExportCSVConfig
+        for column in csv_config.data_columns
+            !isnothing(column.column_data) && push!(ids, persistent_id(column.column_data))
+        end
+    elseif config.name == :iceberg_config
+        iceberg_config = config[]::ExportIcebergConfig
+        !isnothing(iceberg_config.table_def) && push!(ids, persistent_id(iceberg_config.table_def))
+    end
+    return nothing
+end
+
 persistent_id(fragment::Fragment) = persistent_id(fragment.id::FragmentId)
 persistent_id(id::FragmentId) = LQPFragmentId(id.id)
 persistent_id(id::RelationId) = UInt128(id.id_low) + (UInt128(id.id_high) << 64)
diff --git a/sdks/julia/LogicalQueryProtocol.jl/test/properties_tests.jl b/sdks/julia/LogicalQueryProtocol.jl/test/properties_tests.jl
@@ -117,3 +117,199 @@ end
     )
     @test persistent_id(def) == UInt128(7)
 end
+
+@testitem "global_ids" tags = [:ring1, :unit] begin
+    using LogicalQueryProtocol: global_ids, persistent_id, LQPRelationId
+    using LogicalQueryProtocol.relationalai.lqp.v1
+    using ProtoBuf: OneOf
+
+    rid1 = RelationId(; id_low=UInt64(10), id_high=UInt64(0))
+    rid2 = RelationId(; id_low=UInt64(20), id_high=UInt64(0))
+    rid3 = RelationId(; id_low=UInt64(30), id_high=UInt64(0))
+    t1 = var"#Type"(; var"#type"=OneOf(:int_type, IntType()))
+
+    # Def declaration
+    def = Def(;
+        name=rid1,
+        body=Abstraction(; vars=Binding[], value=Formula()),
+    )
+    decl_def = Declaration(; declaration_type=OneOf(:def, def))
+    @test global_ids(decl_def) == [persistent_id(rid1)]
+
+    # Constraint declaration
+    fd = FunctionalDependency(; keys=Var[], values=Var[])
+    constraint = Constraint(;
+        name=rid2,
+        constraint_type=OneOf(:functional_dependency, fd),
+    )
+    decl_constraint = Declaration(; declaration_type=OneOf(:constraint, constraint))
+    @test global_ids(decl_constraint) == [persistent_id(rid2)]
+
+    # EDB data declaration
+    edb = EDB(; target_id=rid1)
+    data_edb = Data(; data_type=OneOf(:edb, edb))
+    decl_edb = Declaration(; declaration_type=OneOf(:data, data_edb))
+    @test global_ids(decl_edb) == [persistent_id(rid1)]
+
+    # CSVData declaration with multiple columns
+    col1 = GNFColumn(; column_path=["a"], target_id=rid1, types=[t1])
+    col2 = GNFColumn(; column_path=["b"], target_id=rid2, types=[t1])
+    csv = CSVData(; columns=[col1, col2])
+    data_csv = Data(; data_type=OneOf(:csv_data, csv))
+    @test global_ids(data_csv) == [persistent_id(rid1), persistent_id(rid2)]
+
+    # CSVData skips columns without target_id
+    col_no_target = GNFColumn(; column_path=["c"], types=[t1])
+    csv2 = CSVData(; columns=[col1, col_no_target, col2])
+    data_csv2 = Data(; data_type=OneOf(:csv_data, csv2))
+    @test global_ids(data_csv2) == [persistent_id(rid1), persistent_id(rid2)]
+
+    # IcebergData declaration with multiple columns
+    loc = IcebergLocator(; table_name="t", namespace=["n"], warehouse="w")
+    cfg = IcebergCatalogConfig(;
+        catalog_uri="uri", scope="", properties=Dict(), auth_properties=Dict(),
+    )
+    icol1 = GNFColumn(; column_path=["x"], target_id=rid1, types=[t1])
+    icol2 = GNFColumn(; column_path=["y"], target_id=rid2, types=[t1])
+    icol3 = GNFColumn(; column_path=["z"], target_id=rid3, types=[t1])
+    iceberg = IcebergData(;
+        locator=loc, config=cfg, columns=[icol1, icol2, icol3],
+        from_snapshot="s1", to_snapshot="s2", returns_delta=false,
+    )
+    data_iceberg = Data(; data_type=OneOf(:iceberg_data, iceberg))
+    @test global_ids(data_iceberg) == [
+        persistent_id(rid1), persistent_id(rid2), persistent_id(rid3),
+    ]
+
+    # IcebergData skips columns without target_id
+    icol_no_target = GNFColumn(; column_path=["w"], types=[t1])
+    iceberg2 = IcebergData(;
+        locator=loc, config=cfg, columns=[icol1, icol_no_target, icol3],
+        from_snapshot="", to_snapshot="", returns_delta=true,
+    )
+    data_iceberg2 = Data(; data_type=OneOf(:iceberg_data, iceberg2))
+    @test global_ids(data_iceberg2) == [persistent_id(rid1), persistent_id(rid3)]
+
+    # Data routed through Declaration
+    decl_iceberg = Declaration(; declaration_type=OneOf(:data, data_iceberg))
+    @test global_ids(decl_iceberg) == [
+        persistent_id(rid1), persistent_id(rid2), persistent_id(rid3),
+    ]
+end
+
+@testitem "collect_demanded_relations" tags = [:ring1, :unit] begin
+    using LogicalQueryProtocol: collect_demanded_relations, persistent_id, LQPRelationId
+    using LogicalQueryProtocol.relationalai.lqp.v1
+    using ProtoBuf: OneOf
+
+    rid1 = RelationId(; id_low=UInt64(1), id_high=UInt64(0))
+    rid2 = RelationId(; id_low=UInt64(2), id_high=UInt64(0))
+    rid3 = RelationId(; id_low=UInt64(3), id_high=UInt64(0))
+    rid4 = RelationId(; id_low=UInt64(4), id_high=UInt64(0))
+
+    # Empty epoch
+    epoch = Epoch(; writes=Write[], reads=Read[])
+    @test collect_demanded_relations(epoch) == Set{LQPRelationId}()
+
+    # Demand read
+    demand = Demand(; relation_id=rid1)
+    rd_demand = Read(; read_type=OneOf(:demand, demand))
+
+    # Output read
+    output = Output(; name="out", relation_id=rid2)
+    rd_output = Read(; read_type=OneOf(:output, output))
+
+    # Abort read
+    abort = Abort(; name="ab", relation_id=rid3)
+    rd_abort = Read(; read_type=OneOf(:abort, abort))
+
+    # Epoch with all three read types
+    epoch_mixed = Epoch(; writes=Write[], reads=[rd_demand, rd_output, rd_abort])
+    ids = collect_demanded_relations(epoch_mixed)
+    @test ids == Set([persistent_id(rid1), persistent_id(rid2), persistent_id(rid3)])
+
+    # Duplicate relation IDs are deduplicated
+    demand2 = Demand(; relation_id=rid1)
+    rd_demand2 = Read(; read_type=OneOf(:demand, demand2))
+    epoch_dup = Epoch(; writes=Write[], reads=[rd_demand, rd_demand2])
+    @test length(collect_demanded_relations(epoch_dup)) == 1
+
+    # CSV export collects column data relation IDs
+    col1 = ExportCSVColumn(; column_name="a", column_data=rid1)
+    col2 = ExportCSVColumn(; column_name="b", column_data=rid2)
+    csv_config = ExportCSVConfig(; path="/tmp/out", data_columns=[col1, col2])
+    export_csv = Export(; export_config=OneOf(:csv_config, csv_config))
+    rd_export_csv = Read(; read_type=OneOf(:var"#export", export_csv))
+    epoch_csv = Epoch(; writes=Write[], reads=[rd_export_csv])
+    csv_ids = collect_demanded_relations(epoch_csv)
+    @test csv_ids == Set([persistent_id(rid1), persistent_id(rid2)])
+
+    # CSV export skips columns without column_data
+    col_no_data = ExportCSVColumn(; column_name="c")
+    csv_config2 = ExportCSVConfig(; path="/tmp/out", data_columns=[col1, col_no_data])
+    export_csv2 = Export(; export_config=OneOf(:csv_config, csv_config2))
+    rd_export_csv2 = Read(; read_type=OneOf(:var"#export", export_csv2))
+    epoch_csv2 = Epoch(; writes=Write[], reads=[rd_export_csv2])
+    @test collect_demanded_relations(epoch_csv2) == Set([persistent_id(rid1)])
+
+    # Iceberg export collects table_def relation ID
+    loc = IcebergLocator(; table_name="t", namespace=["n"], warehouse="w")
+    cfg = IcebergCatalogConfig(;
+        catalog_uri="uri", scope="", properties=Dict(), auth_properties=Dict(),
+    )
+    iceberg_config = ExportIcebergConfig(; locator=loc, config=cfg, table_def=rid4)
+    export_iceberg = Export(; export_config=OneOf(:iceberg_config, iceberg_config))
+    rd_export_ice = Read(; read_type=OneOf(:var"#export", export_iceberg))
+    epoch_ice = Epoch(; writes=Write[], reads=[rd_export_ice])
+    @test collect_demanded_relations(epoch_ice) == Set([persistent_id(rid4)])
+
+    # Export with no export_config is handled gracefully
+    export_empty = Export()
+    rd_export_empty = Read(; read_type=OneOf(:var"#export", export_empty))
+    epoch_empty_export = Epoch(; writes=Write[], reads=[rd_export_empty])
+    @test collect_demanded_relations(epoch_empty_export) == Set{LQPRelationId}()
+end
+
+@testitem "read_lqp" tags = [:ring1, :unit] begin
+    using LogicalQueryProtocol: read_lqp
+    using LogicalQueryProtocol.relationalai.lqp.v1: Transaction
+
+    lqp_dir = joinpath(@__DIR__, "lqp")
+    @test isdir(lqp_dir)
+
+    txn = read_lqp(joinpath(lqp_dir, "arithmetic.lqp"))
+    @test txn isa Transaction
+    @test length(txn.epochs) >= 1
+end
+
+@testitem "read_bin" tags = [:ring1, :unit] begin
+    using LogicalQueryProtocol: read_bin
+    using LogicalQueryProtocol.relationalai.lqp.v1: Transaction
+
+    bin_dir = joinpath(@__DIR__, "bin")
+    @test isdir(bin_dir)
+
+    txn = read_bin(joinpath(bin_dir, "arithmetic.bin"))
+    @test txn isa Transaction
+    @test length(txn.epochs) >= 1
+end
+
+@testitem "read_lqp and read_bin agree" tags = [:ring1, :unit] begin
+    using LogicalQueryProtocol: read_lqp, read_bin
+
+    lqp_dir = joinpath(@__DIR__, "lqp")
+    bin_dir = joinpath(@__DIR__, "bin")
+
+    lqp_files = sort(filter(f -> endswith(f, ".lqp"), readdir(lqp_dir)))
+    @test !isempty(lqp_files)
+
+    for lqp_file in lqp_files
+        bin_file = replace(lqp_file, ".lqp" => ".bin")
+        bin_path = joinpath(bin_dir, bin_file)
+        isfile(bin_path) || continue
+
+        txn_lqp = read_lqp(joinpath(lqp_dir, lqp_file))
+        txn_bin = read_bin(bin_path)
+        @test txn_lqp == txn_bin
+    end
+end
