laskoviymishka commented on code in PR #17:
URL:
https://github.com/apache/iceberg-verification/pull/17#discussion_r4536787137
##########
dev/schema/cases.partition-transforms.schema.json:
##########
@@ -0,0 +1,99 @@
+{
+ "$schema": "https://json-schema.org/draft/2020-12/schema";,
+ "$id":
"https://iceberg.apache.org/verification/cases.partition-transforms.schema.json";,
+ "title": "Conformance cases (partition-transforms surface extension)",
+ "description": "Extra structural constraints for
table-spec/partition-transforms. Applied on top of cases.base.schema.json by
dev/validate-fixtures.py.",
+ "type": "object",
+ "required": ["cases"],
+ "additionalProperties": false,
+ "properties": {
+ "cases": {
+ "type": "array",
+ "items": { "$ref": "#/$defs/case" }
+ }
+ },
+ "$defs": {
+ "case": {
+ "type": "object",
+ "required": ["id", "valid", "input", "decoded", "clause", "spec_ref"],
+ "additionalProperties": false,
+ "properties": {
+ "id": { "type": "string", "minLength": 1 },
+ "valid": { "const": true },
+ "input": { "$ref": "#/$defs/input" },
+ "decoded": { "$ref": "#/$defs/decoded" },
+ "clause": { "type": "string", "minLength": 1 },
+ "spec_ref": { "type": "string", "minLength": 1 },
+ "normative_level": { "enum": ["must", "should"] }
+ }
+ },
+ "input": {
+ "type": "object",
+ "required": ["transform", "source_type", "values"],
+ "additionalProperties": false,
+ "properties": {
+ "transform": {
+ "type": "string",
+ "pattern":
"^(identity|bucket\\[[0-9]+\\]|truncate\\[[0-9]+\\]|year|month|day|hour|void)$"
Review Comment:
`truncate[0]` gets through this pattern and the validator: a `truncate/`
case using it validates `OK`, and so does `bucket[0]` outside `bucket/`. Java's
[`Truncate`](https://github.com/apache/iceberg/blob/ea73769d120dc3819d8cfa2253b6dba4bb394886/api/src/main/java/org/apache/iceberg/transforms/Truncate.java#L41-L53)
rejects width <= 0 at construction, so such a fixture would expect results
from a transform that implementations refuse to build.
##########
dev/validate-fixtures.py:
##########
@@ -47,10 +52,317 @@ def _is_types_surface(path):
return "/table-spec/types/" in p or p.startswith("table-spec/types/")
+def _is_partition_transforms_surface(path):
+ p = _rel(path)
+ return "/table-spec/partition-transforms/" in p or p.startswith(
+ "table-spec/partition-transforms/"
+ )
+
+
+IDENTITY_TYPES = {
+ "unknown",
+ "boolean",
+ "int",
+ "long",
+ "float",
+ "double",
+ "decimal",
+ "date",
+ "time",
+ "timestamp",
+ "timestamptz",
+ "timestamp_ns",
+ "timestamptz_ns",
+ "string",
+ "uuid",
+ "fixed",
+ "binary",
+}
+BUCKET_TYPES = {
+ "int",
+ "long",
+ "decimal",
+ "date",
+ "time",
+ "timestamp",
+ "timestamptz",
+ "timestamp_ns",
+ "timestamptz_ns",
+ "string",
+ "uuid",
+ "fixed",
+ "binary",
+}
+TRUNCATE_TYPES = {"int", "long", "decimal", "string", "binary"}
+DATE_TRANSFORM_TYPES = {
+ "date",
+ "timestamp",
+ "timestamptz",
+ "timestamp_ns",
+ "timestamptz_ns",
+}
+HOUR_TYPES = {"timestamp", "timestamptz", "timestamp_ns", "timestamptz_ns"}
+VOID_TYPES = IDENTITY_TYPES | {"geometry", "geography"}
+
+
+def _type_family(type_name):
+ if not isinstance(type_name, str):
+ return None
+ if type_name.startswith("decimal("):
+ return "decimal"
+ if type_name.startswith("fixed["):
+ return "fixed"
+ if type_name.startswith("geometry("):
+ return "geometry"
+ if type_name.startswith("geography("):
+ return "geography"
+ return type_name
+
+
+def _transform_family(transform):
+ if not isinstance(transform, str):
+ return None
+ if transform.startswith("bucket["):
+ return "bucket"
+ if transform.startswith("truncate["):
+ return "truncate"
+ return transform
+
+
+def _json_scalar_equal(left, right):
+ return type(left) is type(right) and left == right
+
+
+def _partition_transform_errors(path, doc, check_bucket=True):
+ errors = []
+ cases = doc.get("cases") if isinstance(doc, dict) else None
+ if not isinstance(cases, list):
+ return errors
+
+ allowed_sources = {
+ "identity": IDENTITY_TYPES,
+ "bucket": BUCKET_TYPES,
+ "truncate": TRUNCATE_TYPES,
+ "year": DATE_TRANSFORM_TYPES,
+ "month": DATE_TRANSFORM_TYPES,
+ "day": DATE_TRANSFORM_TYPES,
+ "hour": HOUR_TYPES,
+ "void": VOID_TYPES,
+ }
+
+ for index, case in enumerate(cases):
+ if not isinstance(case, dict):
+ continue
+ input_value = case.get("input")
+ decoded = case.get("decoded")
+ if not isinstance(input_value, dict) or not isinstance(decoded, dict):
+ continue
+
+ transform = input_value.get("transform")
+ source_type = input_value.get("source_type")
+ values = input_value.get("values")
+ result_types = decoded.get("result_types")
+ expected = decoded.get("values")
+ transform_family = _transform_family(transform)
+ source_family = _type_family(source_type)
+ label = f"{path}: cases/{index}"
+
+ permitted = allowed_sources.get(transform_family)
+ if permitted is None:
+ errors.append(f"{label}: unknown transform '{transform}'")
+ elif source_family not in permitted:
+ errors.append(
+ f"{label}: transform '{transform}' does not support source
type "
+ f"'{source_type}'"
+ )
+
+ if isinstance(values, list) and isinstance(expected, list):
+ if len(values) != len(expected):
+ errors.append(
+ f"{label}: input.values and decoded.values must have equal
lengths"
+ )
+ if None not in values:
+ errors.append(f"{label}: input.values must include null")
+ for value_index, (value, output) in enumerate(zip(values,
expected)):
+ if value is None and output is not None:
+ errors.append(
+ f"{label}: decoded.values/{value_index} must be null
for a "
+ "null input"
+ )
+ if transform_family == "void" and output is not None:
+ errors.append(
+ f"{label}: decoded.values/{value_index} must be null
for void"
+ )
+ if transform_family == "identity" and not _json_scalar_equal(
+ value, output
+ ):
+ errors.append(
+ f"{label}: decoded.values/{value_index} must equal its
"
+ "identity input"
+ )
+
+ required_result_types = None
+ if transform_family in {"bucket", "year", "month", "hour"}:
+ required_result_types = {"int"}
+ elif transform_family == "day":
+ required_result_types = {"date"}
+ elif transform_family in {"identity", "truncate"} and isinstance(
+ source_type, str
+ ):
+ required_result_types = {source_type}
+ elif transform_family == "void" and isinstance(source_type, str):
+ required_result_types = {source_type, "int"}
+
+ if (
+ required_result_types is not None
+ and isinstance(result_types, list)
+ and all(isinstance(result_type, str) for result_type in
result_types)
+ ):
+ if set(result_types) != required_result_types:
+ errors.append(
+ f"{label}: decoded.result_types must be exactly "
+ f"{sorted(required_result_types)}"
+ )
+
+ bucket_path = path.replace("\\", "/").endswith(
+ "/partition-transforms/bucket/cases.json"
+ )
+ if check_bucket and not errors and bucket_path:
+ errors.extend(_bucket_errors(path, doc))
+
+ return errors
+
+
+EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)
+TIMESTAMP = re.compile(
+ r"^(\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d)(?:\.(\d{1,9}))?(Z|[+-]\d\d:\d\d)?$"
+)
+MASK = 0xFFFFFFFF
+
+
+def murmur3_x86_32(data):
+ """MurmurHash3 x86 32-bit with seed 0, returned as an unsigned integer."""
+ h = 0
+ for offset in range(0, len(data) // 4 * 4, 4):
+ k = int.from_bytes(data[offset : offset + 4], "little")
+ k = (k * 0xCC9E2D51) & MASK
+ k = ((k << 15) | (k >> 17)) & MASK
+ k = (k * 0x1B873593) & MASK
+ h ^= k
+ h = ((h << 13) | (h >> 19)) & MASK
+ h = (h * 5 + 0xE6546B64) & MASK
+
+ tail = data[len(data) // 4 * 4 :]
+ if tail:
+ k = int.from_bytes(tail, "little")
+ k = (k * 0xCC9E2D51) & MASK
+ k = ((k << 15) | (k >> 17)) & MASK
+ k = (k * 0x1B873593) & MASK
+ h ^= k
+
+ h ^= len(data)
+ h ^= h >> 16
+ h = (h * 0x85EBCA6B) & MASK
+ h ^= h >> 13
+ h = (h * 0xC2B2AE35) & MASK
+ return (h ^ (h >> 16)) & MASK
+
+
+def epoch_units(value, nanos=False):
+ match = TIMESTAMP.fullmatch(value)
+ if not match:
+ raise ValueError(f"invalid timestamp: {value}")
+ base, fraction, zone = match.groups()
+ instant = datetime.fromisoformat(base + ("+00:00" if zone == "Z" else zone
or ""))
+ instant = instant.astimezone(timezone.utc) if zone else
instant.replace(tzinfo=timezone.utc)
+ delta = instant - EPOCH
+ seconds = delta.days * 86400 + delta.seconds
+ digits = (fraction or "").ljust(9, "0")
+ nanoseconds = seconds * 1_000_000_000 + int(digits or "0")
+ if nanos and nanoseconds < 0 and nanoseconds % 1000:
+ raise ValueError("negative nanoseconds between microseconds need a
spec rounding rule")
+ return nanoseconds // 1000
+
+
+def decimal_bytes(value):
+ decimal = Decimal(value)
+ digits = decimal.as_tuple()
+ unscaled = int("".join(map(str, digits.digits)))
+ if digits.sign:
+ unscaled = -unscaled
+ length = 1
+ while not -(1 << (length * 8 - 1)) <= unscaled < (1 << (length * 8 - 1)):
+ length += 1
+ return unscaled.to_bytes(length, "big", signed=True)
+
+
+def encode_for_hash(source_type, value):
+ if source_type in ("int", "long"):
+ return struct.pack("<q", int(value))
Review Comment:
For `int`, `long` and decimal the bucket check hashes whatever it's given,
so values that break the [README
encoding](https://github.com/apache/iceberg-verification/blob/a45307a7e7fa019fc339d03ba317387038030b12/table-spec/partition-transforms/README.md?plain=1#L100-L104)
are accepted: `int` `34.5` (hashed as 34), `true` (hashed as 1), a bare-number
`long` `34`, and `"14.2"` under `decimal(9, 2)`, since `decimal_bytes` never
reads `P` or `S`. The committed fixtures all follow the encoding.
##########
dev/validate-fixtures.py:
##########
@@ -47,10 +52,317 @@ def _is_types_surface(path):
return "/table-spec/types/" in p or p.startswith("table-spec/types/")
+def _is_partition_transforms_surface(path):
+ p = _rel(path)
+ return "/table-spec/partition-transforms/" in p or p.startswith(
+ "table-spec/partition-transforms/"
+ )
+
+
+IDENTITY_TYPES = {
+ "unknown",
+ "boolean",
+ "int",
+ "long",
+ "float",
+ "double",
+ "decimal",
+ "date",
+ "time",
+ "timestamp",
+ "timestamptz",
+ "timestamp_ns",
+ "timestamptz_ns",
+ "string",
+ "uuid",
+ "fixed",
+ "binary",
+}
+BUCKET_TYPES = {
+ "int",
+ "long",
+ "decimal",
+ "date",
+ "time",
+ "timestamp",
+ "timestamptz",
+ "timestamp_ns",
+ "timestamptz_ns",
+ "string",
+ "uuid",
+ "fixed",
+ "binary",
+}
+TRUNCATE_TYPES = {"int", "long", "decimal", "string", "binary"}
+DATE_TRANSFORM_TYPES = {
+ "date",
+ "timestamp",
+ "timestamptz",
+ "timestamp_ns",
+ "timestamptz_ns",
+}
+HOUR_TYPES = {"timestamp", "timestamptz", "timestamp_ns", "timestamptz_ns"}
+VOID_TYPES = IDENTITY_TYPES | {"geometry", "geography"}
+
+
+def _type_family(type_name):
+ if not isinstance(type_name, str):
+ return None
+ if type_name.startswith("decimal("):
+ return "decimal"
+ if type_name.startswith("fixed["):
+ return "fixed"
+ if type_name.startswith("geometry("):
+ return "geometry"
+ if type_name.startswith("geography("):
+ return "geography"
+ return type_name
+
+
+def _transform_family(transform):
+ if not isinstance(transform, str):
+ return None
+ if transform.startswith("bucket["):
+ return "bucket"
+ if transform.startswith("truncate["):
+ return "truncate"
+ return transform
+
+
+def _json_scalar_equal(left, right):
+ return type(left) is type(right) and left == right
+
+
+def _partition_transform_errors(path, doc, check_bucket=True):
+ errors = []
+ cases = doc.get("cases") if isinstance(doc, dict) else None
+ if not isinstance(cases, list):
+ return errors
+
+ allowed_sources = {
+ "identity": IDENTITY_TYPES,
+ "bucket": BUCKET_TYPES,
+ "truncate": TRUNCATE_TYPES,
+ "year": DATE_TRANSFORM_TYPES,
+ "month": DATE_TRANSFORM_TYPES,
+ "day": DATE_TRANSFORM_TYPES,
+ "hour": HOUR_TYPES,
+ "void": VOID_TYPES,
+ }
+
+ for index, case in enumerate(cases):
+ if not isinstance(case, dict):
+ continue
+ input_value = case.get("input")
+ decoded = case.get("decoded")
+ if not isinstance(input_value, dict) or not isinstance(decoded, dict):
+ continue
+
+ transform = input_value.get("transform")
+ source_type = input_value.get("source_type")
+ values = input_value.get("values")
+ result_types = decoded.get("result_types")
+ expected = decoded.get("values")
+ transform_family = _transform_family(transform)
+ source_family = _type_family(source_type)
+ label = f"{path}: cases/{index}"
+
+ permitted = allowed_sources.get(transform_family)
+ if permitted is None:
+ errors.append(f"{label}: unknown transform '{transform}'")
+ elif source_family not in permitted:
+ errors.append(
+ f"{label}: transform '{transform}' does not support source
type "
+ f"'{source_type}'"
+ )
+
+ if isinstance(values, list) and isinstance(expected, list):
+ if len(values) != len(expected):
+ errors.append(
+ f"{label}: input.values and decoded.values must have equal
lengths"
+ )
+ if None not in values:
+ errors.append(f"{label}: input.values must include null")
+ for value_index, (value, output) in enumerate(zip(values,
expected)):
+ if value is None and output is not None:
+ errors.append(
+ f"{label}: decoded.values/{value_index} must be null
for a "
+ "null input"
+ )
+ if transform_family == "void" and output is not None:
+ errors.append(
+ f"{label}: decoded.values/{value_index} must be null
for void"
+ )
+ if transform_family == "identity" and not _json_scalar_equal(
+ value, output
+ ):
+ errors.append(
+ f"{label}: decoded.values/{value_index} must equal its
"
+ "identity input"
+ )
+
+ required_result_types = None
+ if transform_family in {"bucket", "year", "month", "hour"}:
+ required_result_types = {"int"}
+ elif transform_family == "day":
+ required_result_types = {"date"}
+ elif transform_family in {"identity", "truncate"} and isinstance(
+ source_type, str
+ ):
+ required_result_types = {source_type}
+ elif transform_family == "void" and isinstance(source_type, str):
+ required_result_types = {source_type, "int"}
+
+ if (
+ required_result_types is not None
+ and isinstance(result_types, list)
+ and all(isinstance(result_type, str) for result_type in
result_types)
+ ):
+ if set(result_types) != required_result_types:
+ errors.append(
+ f"{label}: decoded.result_types must be exactly "
+ f"{sorted(required_result_types)}"
+ )
+
+ bucket_path = path.replace("\\", "/").endswith(
+ "/partition-transforms/bucket/cases.json"
+ )
+ if check_bucket and not errors and bucket_path:
+ errors.extend(_bucket_errors(path, doc))
+
+ return errors
+
+
+EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)
+TIMESTAMP = re.compile(
+ r"^(\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d)(?:\.(\d{1,9}))?(Z|[+-]\d\d:\d\d)?$"
+)
+MASK = 0xFFFFFFFF
+
+
+def murmur3_x86_32(data):
+ """MurmurHash3 x86 32-bit with seed 0, returned as an unsigned integer."""
+ h = 0
+ for offset in range(0, len(data) // 4 * 4, 4):
+ k = int.from_bytes(data[offset : offset + 4], "little")
+ k = (k * 0xCC9E2D51) & MASK
+ k = ((k << 15) | (k >> 17)) & MASK
+ k = (k * 0x1B873593) & MASK
+ h ^= k
+ h = ((h << 13) | (h >> 19)) & MASK
+ h = (h * 5 + 0xE6546B64) & MASK
+
+ tail = data[len(data) // 4 * 4 :]
+ if tail:
+ k = int.from_bytes(tail, "little")
+ k = (k * 0xCC9E2D51) & MASK
+ k = ((k << 15) | (k >> 17)) & MASK
+ k = (k * 0x1B873593) & MASK
+ h ^= k
+
+ h ^= len(data)
+ h ^= h >> 16
+ h = (h * 0x85EBCA6B) & MASK
+ h ^= h >> 13
+ h = (h * 0xC2B2AE35) & MASK
+ return (h ^ (h >> 16)) & MASK
+
+
+def epoch_units(value, nanos=False):
+ match = TIMESTAMP.fullmatch(value)
+ if not match:
+ raise ValueError(f"invalid timestamp: {value}")
+ base, fraction, zone = match.groups()
+ instant = datetime.fromisoformat(base + ("+00:00" if zone == "Z" else zone
or ""))
+ instant = instant.astimezone(timezone.utc) if zone else
instant.replace(tzinfo=timezone.utc)
+ delta = instant - EPOCH
+ seconds = delta.days * 86400 + delta.seconds
+ digits = (fraction or "").ljust(9, "0")
+ nanoseconds = seconds * 1_000_000_000 + int(digits or "0")
+ if nanos and nanoseconds < 0 and nanoseconds % 1000:
+ raise ValueError("negative nanoseconds between microseconds need a
spec rounding rule")
+ return nanoseconds // 1000
+
+
+def decimal_bytes(value):
+ decimal = Decimal(value)
+ digits = decimal.as_tuple()
+ unscaled = int("".join(map(str, digits.digits)))
+ if digits.sign:
+ unscaled = -unscaled
+ length = 1
+ while not -(1 << (length * 8 - 1)) <= unscaled < (1 << (length * 8 - 1)):
+ length += 1
+ return unscaled.to_bytes(length, "big", signed=True)
+
+
+def encode_for_hash(source_type, value):
+ if source_type in ("int", "long"):
+ return struct.pack("<q", int(value))
+ if source_type.startswith("decimal("):
+ return decimal_bytes(value)
+ if source_type == "date":
+ days = (date.fromisoformat(value) - date(1970, 1, 1)).days
+ return struct.pack("<q", days)
+ if source_type == "time":
+ clock = time.fromisoformat(value)
+ micros = ((clock.hour * 60 + clock.minute) * 60 + clock.second) *
1_000_000
+ return struct.pack("<q", micros + clock.microsecond)
+ if source_type in ("timestamp", "timestamptz", "timestamp_ns",
"timestamptz_ns"):
+ return struct.pack("<q", epoch_units(value,
source_type.endswith("_ns")))
+ if source_type == "string":
+ return value.encode("utf-8")
+ if source_type == "uuid":
+ return uuid.UUID(value).bytes
+ if source_type == "binary" or source_type.startswith("fixed["):
+ return bytes.fromhex(value)
+ raise ValueError(f"unsupported bucket source type: {source_type}")
+
+
+def _bucket_errors(path, doc):
+ vectors = (
+ (struct.pack("<q", 34), 2017239379),
+ (struct.pack("<q", 17486), -653330422 & MASK),
+ (b"iceberg", 1210000089),
+ )
+ for data, expected_hash in vectors:
+ if murmur3_x86_32(data) != expected_hash:
+ raise AssertionError("Murmur3 does not reproduce an Appendix B
vector")
+
+ errors = []
+ for case_index, case in enumerate(doc["cases"]):
+ width = re.fullmatch(r"bucket\[(\d+)\]", case["input"]["transform"])
+ if not width or int(width[1]) <= 0:
+ errors.append(f"{path}: cases/{case_index}: invalid bucket
transform")
+ continue
+ count = int(width[1])
+ source_type = case["input"]["source_type"]
+ values = case["input"]["values"]
+ expected = case["decoded"]["values"]
+ for index, (value, wanted) in enumerate(zip(values, expected)):
+ try:
+ actual = None
+ if value is not None:
+ hashed = murmur3_x86_32(encode_for_hash(source_type,
value))
+ actual = (hashed & 0x7FFFFFFF) % count
+ except (ValueError, OverflowError, struct.error) as exc:
Review Comment:
A value that doesn't match `source_type` crashes the validator here instead
of producing a `FAILED:` report. With `"source_type": "string", "values":
[null, 5]` under `bucket/`, the run exits 1 with `AttributeError: 'int' object
has no attribute 'encode'` and no `cases/N` location; decimal, date, time,
timestamp, uuid and binary behave the same. The [selftest
contract](https://github.com/apache/iceberg-verification/blob/a45307a7e7fa019fc339d03ba317387038030b12/Makefile#L52-L55)
asks for a FAILED report, not a crash.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
