Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
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
13 changes: 12 additions & 1 deletion pkg/actions/lua/lakefs/catalogexport/hive.lua
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,11 @@ end
-- Hive format partition iterator each result set is a collection of files under the same partition
local function extract_partition_pager(client, repo_id, commit_id, base_path, partition_cols, page_size)
local target_partition = ""
-- hidden ("_") segments are checked only below the table path, as Spark does
local base_prefix = base_path
if base_prefix ~= "" and not strings.has_suffix(base_prefix, "/") then
base_prefix = base_prefix .. "/"
end
local pager = utils.lakefs_object_pager(client, repo_id, commit_id, "", base_path, "",
page_size or DEFAULT_PAGE_SIZE)
local page = pager()
Expand All @@ -50,7 +55,13 @@ local function extract_partition_pager(client, repo_id, commit_id, base_path, pa
end
local entry = page[1]
local partition_key = extract_partitions_path(partition_cols, entry.path)
if not pathlib.is_hidden(entry.path) and partition_key ~= nil then
local rel_path = entry.path
if entry.path == base_path then
rel_path = ""
elseif strings.has_prefix(entry.path, base_prefix) then
rel_path = entry.path:sub(#base_prefix + 1)
end
if not pathlib.is_hidden(rel_path) and partition_key ~= nil then
-- first time: if not set, assign current object partition as the target_partition key
if target_partition == "" then
target_partition = partition_key
Expand Down
5 changes: 5 additions & 0 deletions pkg/actions/lua_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -300,6 +300,11 @@ func TestLuaRunTable(t *testing.T) {
Input: "testdata/lua/catalogexport_hive_partition_pager.lua",
Output: "testdata/lua/catalogexport_hive_partition_pager.output",
},
{
Name: "catalogexport_hive_partition_pager_hidden_base",
Input: "testdata/lua/catalogexport_hive_partition_pager_hidden_base.lua",
Output: "testdata/lua/catalogexport_hive_partition_pager_hidden_base.output",
},
{
Name: "catalogexport_delta",
Input: "testdata/lua/catalogexport_delta.lua",
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
local hive = require("lakefs/catalogexport/hive")
local strings = require("strings")

-- lakefs mock package: lists the given (sorted) paths under prefix
local function mock_client(paths)
return {
list_objects = function(repo_id, commit_id, after, prefix, delimiter, page_size)
local matched = {}
for _, p in ipairs(paths) do
if strings.has_prefix(p, prefix) then
table.insert(matched, {
physical_address = "s3://bucket/" .. p,
path = p,
size_bytes = 1
})
end
end
local first = (after == "") and 1 or after
local last = math.min(first + page_size - 1, #matched)
local results = {}
for i = first, last do
table.insert(results, matched[i])
end
return 200, {
results = results,
pagination = {
has_more = last < #matched,
next_offset = last + 1
}
}
end
}
end

-- table under a "_" path: only segments below the table path are checked for "_"
local staging = {
"_staging/ev/_SUCCESS",
"_staging/ev/_tmp/dt=2024-01-01/part-9.parquet",
"_staging/ev/dt=2024-01-01/_SUCCESS",
"_staging/ev/dt=2024-01-01/part-0.parquet",
"_staging/ev/dt=2024-01-02/.part-1.parquet.crc",
"_staging/ev/dt=2024-01-02/_temporary/0/part-8.parquet",
"_staging/ev/dt=2024-01-02/part-1.parquet",
"_staging/ev2/dt=2024-01-03/part-0.parquet",
}

local cases = {
{base = "_staging/ev", cols = {"dt"}, paths = staging},
{base = "_staging/ev/", cols = {"dt"}, paths = staging},
{base = "", cols = {"dt"}, paths = {
"_tmp/dt=2024-01-01/part-1.parquet",
"dt=2024-01-01/_SUCCESS",
"dt=2024-01-01/part-0.parquet",
}},
{base = "_staging/raw", cols = {}, paths = {
"_staging/raw",
"_staging/raw/_SUCCESS",
"_staging/raw/f1.csv",
"_staging/raw/f2.csv",
}},
-- partition value starting with "_" is not hidden
{base = "_letters", cols = {"a"}, paths = {
"_letters/a=_foo/x.csv",
}},
-- partition column name starting with "_" stays hidden (as in Spark)
{base = "_letters", cols = {"_a"}, paths = {
"_letters/_a=1/x.csv",
}},
{base = "_staging/empty", cols = {"dt"}, paths = {
"_staging/empty/_SUCCESS",
}},
}

for _, c in ipairs(cases) do
print("# base: '" .. c.base .. "' columns: [" .. table.concat(c.cols, ",") .. "]")
local pager = hive.extract_partition_pager(mock_client(c.paths), "repo", "commit", c.base, c.cols, 2)
for part_key, entries in pager do
print("partition: '" .. part_key .. "'")
for _, entry in ipairs(entries) do
print("path: " .. entry.path)
end
end
end
print("# done")
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
# base: '_staging/ev' columns: [dt]
partition: '_staging/ev/dt=2024-01-01/'
path: _staging/ev/dt=2024-01-01/part-0.parquet
partition: '_staging/ev/dt=2024-01-02/'
path: _staging/ev/dt=2024-01-02/.part-1.parquet.crc
path: _staging/ev/dt=2024-01-02/part-1.parquet
# base: '_staging/ev/' columns: [dt]
partition: '_staging/ev/dt=2024-01-01/'
path: _staging/ev/dt=2024-01-01/part-0.parquet
partition: '_staging/ev/dt=2024-01-02/'
path: _staging/ev/dt=2024-01-02/.part-1.parquet.crc
path: _staging/ev/dt=2024-01-02/part-1.parquet
# base: '' columns: [dt]
partition: 'dt=2024-01-01/'
path: dt=2024-01-01/part-0.parquet
# base: '_staging/raw' columns: []
partition: ''
path: _staging/raw
path: _staging/raw/f1.csv
path: _staging/raw/f2.csv
# base: '_letters' columns: [a]
partition: '_letters/a=_foo/'
path: _letters/a=_foo/x.csv
# base: '_letters' columns: [_a]
partition: ''
# base: '_staging/empty' columns: [dt]
partition: ''
# done
Loading