Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Sort by dw_last_updated before materializing #334

Draft
wants to merge 5 commits into
base: main
Choose a base branch
from
Draft

Sort by dw_last_updated before materializing #334

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
4de6b0a
sort
Jul 9, 2024
ef6e2d3
address comments
Jul 11, 2024
76cb74b
Unit test
Jul 15, 2024
589c05b
lint
Jul 15, 2024
9ff29ca
Sort on both exits.
Jul 16, 2024
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
64 changes: 35 additions & 29 deletions deltacat/compute/compactor_v2/steps/merge.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,10 @@
interface as unimplemented_deltacat_storage,
)
from deltacat.compute.compactor_v2.utils.dedupe import drop_duplicates
from deltacat.constants import BYTES_PER_GIBIBYTE
from deltacat.constants import (
BYTES_PER_GIBIBYTE,
DW_LAST_UPDATED_COLUMN_NAME,
)
from deltacat.compute.compactor_v2.constants import (
MERGE_TIME_IN_SECONDS,
MERGE_SUCCESS_COUNT,
Expand Down Expand Up @@ -125,43 +128,46 @@ def _merge_tables(

all_tables.append(table)

if not primary_keys or not can_drop_duplicates:
logger.info(
f"Not dropping duplicates for primary keys={primary_keys} "
f"and can_drop_duplicates={can_drop_duplicates}"
)
all_tables[incremental_idx] = _drop_delta_type_rows(
all_tables[incremental_idx], DeltaType.DELETE
)
# we need not drop duplicates
return pa.concat_tables(all_tables)
should_drop_duplicates = bool(primary_keys) and can_drop_duplicates
if should_drop_duplicates:
all_tables = generate_pk_hash_column(all_tables, primary_keys=primary_keys)

all_tables = generate_pk_hash_column(all_tables, primary_keys=primary_keys)
result_table_list = []

result_table_list = []
incremental_table = drop_duplicates(
all_tables[incremental_idx], on=sc._PK_HASH_STRING_COLUMN_NAME
)

incremental_table = drop_duplicates(
all_tables[incremental_idx], on=sc._PK_HASH_STRING_COLUMN_NAME
)
if compacted_table:
compacted_table = all_tables[0]

if compacted_table:
compacted_table = all_tables[0]

records_to_keep = pc.invert(
pc.is_in(
compacted_table[sc._PK_HASH_STRING_COLUMN_NAME],
incremental_table[sc._PK_HASH_STRING_COLUMN_NAME],
records_to_keep = pc.invert(
pc.is_in(
compacted_table[sc._PK_HASH_STRING_COLUMN_NAME],
incremental_table[sc._PK_HASH_STRING_COLUMN_NAME],
)
)
)

result_table_list.append(compacted_table.filter(records_to_keep))
result_table_list.append(compacted_table.filter(records_to_keep))

incremental_table = _drop_delta_type_rows(incremental_table, DeltaType.DELETE)
result_table_list.append(incremental_table)
incremental_table = _drop_delta_type_rows(incremental_table, DeltaType.DELETE)
result_table_list.append(incremental_table)

final_table = pa.concat_tables(result_table_list)
final_table = final_table.drop([sc._PK_HASH_STRING_COLUMN_NAME])
final_table = pa.concat_tables(result_table_list)
final_table = final_table.drop([sc._PK_HASH_STRING_COLUMN_NAME])
else:
logger.info(
f"Not dropping duplicates for primary keys={primary_keys} "
f"and can_drop_duplicates={can_drop_duplicates}"
)
all_tables[incremental_idx] = _drop_delta_type_rows(
all_tables[incremental_idx], DeltaType.DELETE
)
final_table = pa.concat_tables(all_tables)

# TODO: Retrieve sort order policy from the table version metadata instead of hard-coding.
if DW_LAST_UPDATED_COLUMN_NAME in final_table.column_names:
final_table = final_table.sort_by([(DW_LAST_UPDATED_COLUMN_NAME, "descending")])
Comment on lines +169 to +170
Copy link
Collaborator

Choose a reason for hiding this comment

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

Assuming the previously compacted table will already be sorted, we can further optimize by only sorting the incremental table.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

I think I understand. Would you mind linking a code reference?

Copy link
Contributor Author

@DmitriGekhtman DmitriGekhtman Jul 16, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

Ah, my bad, used my eyeballs.
Yes, I see. For the purposes of a POC, perhaps you can do that in a separate PR. Would indeed be good to avoid wasting time sorting a sorted table.

Copy link
Collaborator

Choose a reason for hiding this comment

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

I now realize we can use sort_keys argument to sort the incremental instead of hardcoding here. This was okay for POC but we want to limit this hard coding to our internal packages.

return final_table


Expand Down
2 changes: 2 additions & 0 deletions deltacat/constants.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,3 +53,5 @@
PYARROW_INFLATION_MULTIPLIER_ALL_COLUMNS = 6

MEMORY_TO_HASH_BUCKET_COUNT_RATIO = 0.0512 * BYTES_PER_TEBIBYTE

DW_LAST_UPDATED_COLUMN_NAME = "dw_last_updated"
60 changes: 55 additions & 5 deletions deltacat/tests/compute/compact_partition_rebase_test_cases.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
import datetime as dt
import pyarrow as pa
from deltacat.constants import DW_LAST_UPDATED_COLUMN_NAME
from deltacat.tests.compute.test_util_common import (
PartitionKey,
PartitionKeyType,
Expand Down Expand Up @@ -65,20 +67,68 @@ class RebaseCompactionTestCaseParams(BaseCompactorTestCase):
],
names=["pk_col_1", "sk_col_1", "sk_col_2", "col_1"],
),
expected_terminal_compact_partition_result=pa.Table.from_arrays(
expected_terminal_compact_partition_result=pa.Table.from_arrays([]),
expected_terminal_exception=None,
expected_terminal_exception_message=None,
do_create_placement_group=False,
records_per_compacted_file=DEFAULT_MAX_RECORDS_PER_FILE,
hash_bucket_count=DEFAULT_HASH_BUCKET_COUNT,
read_kwargs_provider=None,
drop_duplicates=True,
skip_enabled_compact_partition_drivers=[CompactorVersion.V1],
),
"2-rebase-sort": RebaseCompactionTestCaseParams(
primary_keys={"pk_col_1"},
sort_keys=[
SortKey.of(key_name="sk_col_1"),
SortKey.of(key_name="sk_col_2"),
],
partition_keys=[PartitionKey.of("region_id", PartitionKeyType.INT)],
partition_values=["1"],
input_deltas=pa.Table.from_arrays(
[
pa.array([str(i) for i in range(10)]),
pa.array([i for i in range(20, 30)]),
pa.array([i for i in range(0, 10)]),
pa.array(["foo"] * 10),
pa.array([i / 10 for i in range(40, 50)]),
pa.array([i / 10 for i in range(10, 20)]),
pa.array(dt.datetime(year, 1, 1) for year in range(2000, 2010)),
],
names=[
"pk_col_1",
"sk_col_1",
"sk_col_2",
"col_1",
DW_LAST_UPDATED_COLUMN_NAME,
],
),
input_deltas_delta_type=DeltaType.UPSERT,
# dw_last_update is in ascending order in the input table.
# Expect descending sort on dw_last_updated for each hash bucket.
# Since there is only one hash bucket, the order of input rows should be reversed.
rebase_expected_compact_partition_result=pa.Table.from_arrays(
[
pa.array([str(i) for i in reversed(range(10))]),
pa.array([i for i in reversed(range(0, 10))]),
pa.array(["foo"] * 10),
pa.array([i / 10 for i in reversed(range(10, 20))]),
pa.array(
dt.datetime(year, 1, 1) for year in reversed(range(2000, 2010))
),
],
names=[
"pk_col_1",
"sk_col_1",
"sk_col_2",
"col_1",
DW_LAST_UPDATED_COLUMN_NAME,
],
names=["pk_col_1", "sk_col_1", "sk_col_2", "col_1"],
),
expected_terminal_compact_partition_result=pa.Table.from_arrays([]),
expected_terminal_exception=None,
expected_terminal_exception_message=None,
do_create_placement_group=False,
records_per_compacted_file=DEFAULT_MAX_RECORDS_PER_FILE,
Copy link
Collaborator

Choose a reason for hiding this comment

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

max records per file is 4M but your test case doesn't have more than 4M records. So, it only tests sorting within a single file but not across multiple files.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Would you recommend reducing DEFAULT_MAX_RECORDS_PER_FILE to a number smaller than than the input_deltas table?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

(At least, for the new test case?)

hash_bucket_count=DEFAULT_HASH_BUCKET_COUNT,
hash_bucket_count=1,
read_kwargs_provider=None,
drop_duplicates=True,
skip_enabled_compact_partition_drivers=[CompactorVersion.V1],
Expand Down
29 changes: 19 additions & 10 deletions deltacat/tests/compute/test_compact_partition_rebase.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import boto3
from boto3.resources.base import ServiceResource
import pyarrow as pa
from deltacat.constants import DW_LAST_UPDATED_COLUMN_NAME
from deltacat.io.ray_plasma_object_store import RayPlasmaObjectStore
from pytest_benchmark.fixture import BenchmarkFixture

Expand Down Expand Up @@ -274,16 +275,24 @@ def test_compact_partition_rebase_same_source_and_destination(
compacted_delta_locator, storage_type=StorageType.LOCAL, **ds_mock_kwargs
)
actual_rebase_compacted_table = pa.concat_tables(tables)
Copy link
Collaborator

Choose a reason for hiding this comment

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

We can add one assertion if the result should yield multiple tables.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

If I understand correctly, the suggestion is to add an assertion on len(tables)?

# if no primary key is specified then sort by sort_key for consistent assertion
sorting_cols: List[Any] = (
[(val, "ascending") for val in primary_keys] if primary_keys else sort_keys
)
rebase_expected_compact_partition_result = (
rebase_expected_compact_partition_result.combine_chunks().sort_by(sorting_cols)
)
actual_rebase_compacted_table = (
actual_rebase_compacted_table.combine_chunks().sort_by(sorting_cols)
)
if DW_LAST_UPDATED_COLUMN_NAME in actual_rebase_compacted_table.column_names:
# If DW_LAST_UPDATED_COLUMN_NAME is present, don't sort expected and actual tables;
# we want to assert on the order of the rows in the table, to validate sorting on timestamp.
pass
else:
# If DW_LAST_UPDATED_COLUMN_NAME is absent, sort by primary key for consistent assertion.
# Sort by sort_key if no primary key is specified.
sorting_cols: List[Any] = (
[(val, "ascending") for val in primary_keys] if primary_keys else sort_keys
)
rebase_expected_compact_partition_result = (
rebase_expected_compact_partition_result.combine_chunks().sort_by(
sorting_cols
)
)
actual_rebase_compacted_table = (
actual_rebase_compacted_table.combine_chunks().sort_by(sorting_cols)
)
assert actual_rebase_compacted_table.equals(
rebase_expected_compact_partition_result
), f"{actual_rebase_compacted_table} does not match {rebase_expected_compact_partition_result}"
Loading