wehub-resource-sync
bf2343b7e4
Integration Tests - MySQL + Elasticsearch / Detect Changes (push) Has been cancelled
Integration Tests - MySQL + Elasticsearch / integration-tests-mysql-elasticsearch (push) Has been cancelled
Integration Tests - PostgreSQL + Elasticsearch + Redis / Detect Changes (push) Has been cancelled
Integration Tests - PostgreSQL + Elasticsearch + Redis / integration-tests-postgres-elasticsearch-redis (push) Has been cancelled
Integration Tests - PostgreSQL + OpenSearch / Detect Changes (push) Has been cancelled
Integration Tests - PostgreSQL + OpenSearch / integration-tests-postgres-opensearch (push) Has been cancelled
Java Checkstyle / java-checkstyle (push) Has been cancelled
Maven Collate Tests / maven-collate-ci (push) Has been cancelled
OpenMetadata Service Unit Tests / openmetadata-service-unit-tests-status (push) Has been cancelled
Publish Package to Maven Central Repository / publish-maven-packages (push) Has been cancelled
OpenMetadata Service Unit Tests / Detect Changes (push) Has been cancelled
OpenMetadata Service Unit Tests / openmetadata-service-unit-tests (push) Has been cancelled
OpenMetadata Service Unit Tests / k8s_operator-unit-tests (push) Has been cancelled
385 lines
14 KiB
Python
385 lines
14 KiB
Python
# Copyright 2025 Collate
|
|
# Licensed under the Collate Community License, Version 1.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
# https://github.com/open-metadata/OpenMetadata/blob/main/ingestion/LICENSE
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
"""
|
|
Tests for path_pattern.py — glob matching, partition detection, table grouping.
|
|
"""
|
|
|
|
from metadata.generated.schema.entity.data.table import DataType
|
|
from metadata.utils.path_pattern import (
|
|
detect_hive_partitions,
|
|
extract_static_prefix,
|
|
extract_table_root,
|
|
group_files_by_table,
|
|
infer_structure_format,
|
|
pattern_to_regex,
|
|
)
|
|
|
|
|
|
class TestExtractStaticPrefix:
|
|
def test_wildcard_at_second_level(self):
|
|
assert extract_static_prefix("data/*/events/*.parquet") == "data/"
|
|
|
|
def test_wildcard_at_first_level(self):
|
|
assert extract_static_prefix("*/*.csv") == ""
|
|
|
|
def test_no_wildcards(self):
|
|
assert extract_static_prefix("data/events/file.parquet") == "data/events/file.parquet"
|
|
|
|
def test_deep_static_prefix(self):
|
|
assert extract_static_prefix("data/events/*.parquet") == "data/events/"
|
|
|
|
def test_double_star(self):
|
|
assert extract_static_prefix("data/**/*.json") == "data/"
|
|
|
|
def test_empty_pattern(self):
|
|
assert extract_static_prefix("") == ""
|
|
|
|
def test_just_wildcard(self):
|
|
assert extract_static_prefix("*.csv") == ""
|
|
|
|
def test_question_mark_wildcard(self):
|
|
assert extract_static_prefix("data/202?/*.parquet") == "data/"
|
|
|
|
def test_bracket_not_treated_as_wildcard(self):
|
|
"""Bracket character classes are not supported in patterns."""
|
|
assert extract_static_prefix("data/[abc]/*.parquet") == "data/[abc]/"
|
|
|
|
|
|
class TestPatternToRegex:
|
|
def test_single_star_matches_one_level(self):
|
|
regex = pattern_to_regex("data/*/events/*.parquet")
|
|
assert regex.match("data/warehouse/events/file.parquet")
|
|
assert not regex.match("data/a/b/events/file.parquet")
|
|
|
|
def test_double_star_matches_multiple_levels(self):
|
|
regex = pattern_to_regex("data/**/*.json")
|
|
assert regex.match("data/a/b/c/file.json")
|
|
assert regex.match("data/file.json")
|
|
|
|
def test_exact_match(self):
|
|
regex = pattern_to_regex("data/events/file.parquet")
|
|
assert regex.match("data/events/file.parquet")
|
|
assert not regex.match("data/events/other.parquet")
|
|
|
|
def test_extension_filter(self):
|
|
regex = pattern_to_regex("data/*/*.parquet")
|
|
assert regex.match("data/folder/file.parquet")
|
|
assert not regex.match("data/folder/file.csv")
|
|
|
|
def test_question_mark(self):
|
|
regex = pattern_to_regex("data/202?/*.parquet")
|
|
assert regex.match("data/2024/file.parquet")
|
|
assert not regex.match("data/20245/file.parquet")
|
|
|
|
def test_does_not_match_partial(self):
|
|
regex = pattern_to_regex("data/*/events/*.parquet")
|
|
assert not regex.match("data/warehouse/events/file.parquet.bak")
|
|
|
|
def test_special_characters_escaped(self):
|
|
regex = pattern_to_regex("data/events.v2/*.parquet")
|
|
assert regex.match("data/events.v2/file.parquet")
|
|
assert not regex.match("data/eventsXv2/file.parquet")
|
|
|
|
# --- Edge cases from code review ---
|
|
|
|
def test_star_matches_zero_chars(self):
|
|
"""Bug fix: * should match zero or more chars (not one or more).
|
|
data*.parquet should match data.parquet."""
|
|
regex = pattern_to_regex("data*.parquet")
|
|
assert regex.match("data.parquet")
|
|
assert regex.match("data_v2.parquet")
|
|
|
|
def test_double_star_at_start_matches_root(self):
|
|
"""Bug fix: **/*.parquet at start should match file.parquet
|
|
(zero-depth path with no directory)."""
|
|
regex = pattern_to_regex("**/*.parquet")
|
|
assert regex.match("file.parquet")
|
|
assert regex.match("data/file.parquet")
|
|
assert regex.match("a/b/c/file.parquet")
|
|
|
|
def test_double_star_at_end(self):
|
|
"""** at end should match zero or more trailing segments."""
|
|
regex = pattern_to_regex("data/**")
|
|
assert regex.match("data/file.parquet")
|
|
assert regex.match("data/a/b/file.parquet")
|
|
# data/ alone is a directory marker, filtered by list_keys before matching
|
|
|
|
def test_star_matches_empty_segment_in_prefix(self):
|
|
"""prefix*suffix should match when wildcard portion is empty."""
|
|
regex = pattern_to_regex("logs/*.csv")
|
|
assert regex.match("logs/file.csv")
|
|
assert regex.match("logs/.csv") # empty name before .csv
|
|
|
|
|
|
class TestExtractTableRoot:
|
|
def test_with_hive_partitions(self):
|
|
assert extract_table_root("data/events/year=2024/month=01/file.parquet") == "data/events"
|
|
|
|
def test_with_multiple_partitions(self):
|
|
assert extract_table_root("data/events/year=2024/month=01/day=15/file.parquet") == "data/events"
|
|
|
|
def test_without_partitions(self):
|
|
assert extract_table_root("data/events/file.parquet") == "data/events"
|
|
|
|
def test_root_level_file(self):
|
|
assert extract_table_root("file.parquet") == ""
|
|
|
|
def test_single_directory(self):
|
|
assert extract_table_root("events/file.parquet") == "events"
|
|
|
|
def test_partition_at_root(self):
|
|
assert extract_table_root("year=2024/month=01/file.parquet") == ""
|
|
|
|
def test_deep_nesting_no_partition(self):
|
|
assert extract_table_root("a/b/c/d/file.parquet") == "a/b/c/d"
|
|
|
|
def test_matches_manifest_datapath(self):
|
|
"""Table root must match what users put in manifest dataPath."""
|
|
assert extract_table_root("data/events/year=2024/month=01/part-00000.parquet") == "data/events"
|
|
|
|
def test_date_prefix_partition(self):
|
|
"""Non-Hive date prefix like 20230412 should be treated as partition."""
|
|
assert extract_table_root("cities_multiple_simple/20230412/State=AL/file.parquet") == "cities_multiple_simple"
|
|
|
|
def test_date_with_dashes_partition(self):
|
|
"""Date with dashes like 2024-01-15 should be treated as partition."""
|
|
assert extract_table_root("data/events/2024-01-15/file.parquet") == "data/events"
|
|
|
|
def test_timestamp_partition(self):
|
|
"""Timestamp like 20240115T000000Z should be treated as partition."""
|
|
assert extract_table_root("data/logs/20240115T120000Z/file.json") == "data/logs"
|
|
|
|
def test_mixed_non_hive_and_hive(self):
|
|
"""Date prefix followed by Hive partition."""
|
|
assert extract_table_root("data/events/20230412/State=AL/file.parquet") == "data/events"
|
|
|
|
def test_short_number_not_treated_as_partition(self):
|
|
"""Short numbers like 'v2' or directory names should NOT be partitions."""
|
|
assert extract_table_root("data/v2/file.parquet") == "data/v2"
|
|
|
|
def test_four_digit_year_alone_not_partition(self):
|
|
"""Four digits alone like '2024' is ambiguous — could be year partition."""
|
|
# We treat 8+ digits as partition but not 4 digits alone
|
|
assert extract_table_root("data/2024/file.parquet") == "data/2024"
|
|
|
|
|
|
class TestDetectHivePartitions:
|
|
def test_basic_int_partitions(self):
|
|
keys = [
|
|
"root/year=2024/month=01/f.parquet",
|
|
"root/year=2023/month=12/f.parquet",
|
|
"root/year=2024/month=06/f.parquet",
|
|
]
|
|
columns = detect_hive_partitions(keys, "root")
|
|
assert columns is not None
|
|
assert len(columns) == 2
|
|
assert columns[0].name.root == "year"
|
|
assert columns[0].dataType == DataType.INT
|
|
assert columns[1].name.root == "month"
|
|
assert columns[1].dataType == DataType.INT
|
|
|
|
def test_date_partition(self):
|
|
keys = [
|
|
"data/date=2024-01-15/f.parquet",
|
|
"data/date=2024-02-20/f.parquet",
|
|
]
|
|
columns = detect_hive_partitions(keys, "data")
|
|
assert columns is not None
|
|
assert len(columns) == 1
|
|
assert columns[0].name.root == "date"
|
|
assert columns[0].dataType == DataType.DATE
|
|
|
|
def test_string_partition(self):
|
|
keys = [
|
|
"data/region=us-east-1/f.parquet",
|
|
"data/region=eu-west-1/f.parquet",
|
|
]
|
|
columns = detect_hive_partitions(keys, "data")
|
|
assert columns is not None
|
|
assert len(columns) == 1
|
|
assert columns[0].name.root == "region"
|
|
assert columns[0].dataType == DataType.VARCHAR
|
|
|
|
def test_mixed_int_and_string(self):
|
|
keys = [
|
|
"data/year=2024/country=US/f.parquet",
|
|
"data/year=2023/country=UK/f.parquet",
|
|
]
|
|
columns = detect_hive_partitions(keys, "data")
|
|
assert columns is not None
|
|
assert len(columns) == 2
|
|
assert columns[0].dataType == DataType.INT
|
|
assert columns[1].dataType == DataType.VARCHAR
|
|
|
|
def test_no_partitions(self):
|
|
keys = [
|
|
"data/subdir/f.parquet",
|
|
"data/other/f.parquet",
|
|
]
|
|
columns = detect_hive_partitions(keys, "data")
|
|
assert columns is None
|
|
|
|
def test_inconsistent_partitions_returns_none(self):
|
|
keys = [
|
|
"data/year=2024/month=01/f.parquet",
|
|
"data/country=US/f.parquet",
|
|
]
|
|
columns = detect_hive_partitions(keys, "data")
|
|
assert columns is None
|
|
|
|
def test_empty_keys(self):
|
|
assert detect_hive_partitions([], "root") is None
|
|
|
|
def test_deeply_nested_partitions(self):
|
|
keys = [
|
|
"lake/events/year=2024/month=01/day=15/hour=00/f.parquet",
|
|
"lake/events/year=2024/month=01/day=15/hour=12/f.parquet",
|
|
]
|
|
columns = detect_hive_partitions(keys, "lake/events")
|
|
assert columns is not None
|
|
assert len(columns) == 4
|
|
assert [c.name.root for c in columns] == ["year", "month", "day", "hour"]
|
|
|
|
def test_single_partition(self):
|
|
keys = [
|
|
"data/state=AL/cities.parquet",
|
|
"data/state=AZ/cities.parquet",
|
|
]
|
|
columns = detect_hive_partitions(keys, "data")
|
|
assert columns is not None
|
|
assert len(columns) == 1
|
|
assert columns[0].name.root == "state"
|
|
assert columns[0].dataType == DataType.VARCHAR
|
|
|
|
|
|
class TestGroupFilesByTable:
|
|
def test_groups_by_partition_root(self):
|
|
keys = [
|
|
("data/events/year=2024/month=01/a.parquet", 100),
|
|
("data/events/year=2024/month=02/b.parquet", 200),
|
|
("data/users/c.parquet", 300),
|
|
]
|
|
groups = group_files_by_table(keys)
|
|
assert len(groups) == 2
|
|
assert "data/events" in groups
|
|
assert "data/users" in groups
|
|
assert len(groups["data/events"]) == 2
|
|
assert len(groups["data/users"]) == 1
|
|
|
|
def test_root_level_files_grouped_separately(self):
|
|
keys = [
|
|
("a.parquet", 100),
|
|
("b.parquet", 200),
|
|
]
|
|
groups = group_files_by_table(keys)
|
|
assert len(groups) == 1
|
|
assert "" in groups
|
|
assert len(groups[""]) == 2
|
|
|
|
def test_mixed_partitioned_and_flat(self):
|
|
keys = [
|
|
("data/events/year=2024/f.parquet", 100),
|
|
("data/events/standalone.parquet", 200),
|
|
]
|
|
groups = group_files_by_table(keys)
|
|
# Both should group under "data/events"
|
|
assert len(groups) == 1
|
|
assert "data/events" in groups
|
|
|
|
def test_multiple_tables(self):
|
|
keys = [
|
|
("data/sales/region=US/f.parquet", 100),
|
|
("data/sales/region=EU/f.parquet", 200),
|
|
("data/orders/year=2024/f.parquet", 300),
|
|
("data/users/profile.parquet", 400),
|
|
]
|
|
groups = group_files_by_table(keys)
|
|
assert len(groups) == 3
|
|
assert set(groups.keys()) == {"data/sales", "data/orders", "data/users"}
|
|
|
|
|
|
class TestInferStructureFormat:
|
|
"""Format auto-detection from file extensions."""
|
|
|
|
def test_parquet(self):
|
|
assert infer_structure_format("data/events/file.parquet") == "parquet"
|
|
|
|
def test_parquet_pq(self):
|
|
assert infer_structure_format("data/file.pq") == "parquet"
|
|
|
|
def test_csv(self):
|
|
assert infer_structure_format("transactions/data.csv") == "csv"
|
|
|
|
def test_tsv(self):
|
|
assert infer_structure_format("data.tsv") == "tsv"
|
|
|
|
def test_json(self):
|
|
assert infer_structure_format("events/log.json") == "json"
|
|
|
|
def test_jsonl(self):
|
|
assert infer_structure_format("stream.jsonl") == "json"
|
|
|
|
def test_avro(self):
|
|
assert infer_structure_format("schema/data.avro") == "avro"
|
|
|
|
def test_csv_gz(self):
|
|
assert infer_structure_format("compressed/data.csv.gz") == "csv"
|
|
|
|
def test_json_gz(self):
|
|
assert infer_structure_format("logs/app.json.gz") == "json"
|
|
|
|
def test_unknown_extension(self):
|
|
assert infer_structure_format("image.png") is None
|
|
|
|
def test_no_extension(self):
|
|
assert infer_structure_format("README") is None
|
|
|
|
def test_case_insensitive(self):
|
|
assert infer_structure_format("Data.PARQUET") == "parquet"
|
|
|
|
def test_parquet_snappy_compound(self):
|
|
assert infer_structure_format("data.parquet.snappy") == "parquet"
|
|
|
|
def test_parquet_plain(self):
|
|
assert infer_structure_format("data.parquet") == "parquet"
|
|
|
|
|
|
class TestEndToEndDiscovery:
|
|
"""Simulate the full discovery flow: pattern match -> group -> partition detect."""
|
|
|
|
def test_full_flow_parquet_with_partitions(self):
|
|
pattern = "data/*/events/**/*.parquet"
|
|
regex = pattern_to_regex(pattern)
|
|
|
|
all_keys = [
|
|
"data/warehouse/events/year=2024/month=01/part-00000.parquet",
|
|
"data/warehouse/events/year=2024/month=02/part-00000.parquet",
|
|
"data/warehouse/events/year=2023/month=12/part-00000.parquet",
|
|
"data/warehouse/logs/app.log",
|
|
"data/archive/events/year=2022/month=06/part-00000.parquet",
|
|
"other/file.csv",
|
|
]
|
|
|
|
matched = [(k, 1000) for k in all_keys if regex.match(k)]
|
|
assert len(matched) == 4
|
|
|
|
groups = group_files_by_table(matched)
|
|
assert "data/warehouse/events" in groups
|
|
assert "data/archive/events" in groups
|
|
|
|
for table_root, files in groups.items():
|
|
partitions = detect_hive_partitions([k for k, _ in files], table_root)
|
|
assert partitions is not None
|
|
assert len(partitions) == 2
|
|
assert partitions[0].name.root == "year"
|
|
assert partitions[1].name.root == "month"
|