This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 14482bdb0a [python] Read composite BTree indexes in data evolution
scans (#10802)
14482bdb0a is described below
commit 14482bdb0a16f90b45b6c5bbfb84e19b679b4146
Author: XiaoHongbo <[email protected]>
AuthorDate: Tue Oct 27 09:30:20 2026 +0800
[python] Read composite BTree indexes in data evolution scans (#10802)
---
.../python/globalindex/btree/composite_btree.py | 275 ++++++++++++
.../python/globalindex/btree/composite_key.py | 153 +++++++
.../data_evolution_global_index_coverage.py | 17 +-
.../data_evolution_global_index_scanner.py | 272 +++++++++++-
.../execution/python/read/scanner/file_scanner.py | 12 +-
.../globalindex/global_index_evaluation.py | 4 +-
.../tests/composite_btree_fallback_test.py | 463 ++++++++++++++++++++-
paimon-python/pypaimon/tests/global_index_test.py | 4 +-
.../resources/btree/BTreeFixtureGenerator.java | 121 ++++++
.../pypaimon/tests/resources/btree/README.md | 5 +
.../resources/btree/java-v2-composite-ac.btree | Bin 0 -> 232 bytes
.../btree/java-v2-composite-ac.btree.meta | Bin 0 -> 23 bytes
.../resources/btree/java-v2-composite-int.btree | Bin 0 -> 232 bytes
.../btree/java-v2-composite-int.btree.meta | Bin 0 -> 23 bytes
.../resources/btree/java-v2-composite-keys.txt | 18 +
.../btree/java-v2-composite-reversed.btree | Bin 0 -> 232 bytes
.../btree/java-v2-composite-reversed.btree.meta | Bin 0 -> 23 bytes
17 files changed, 1327 insertions(+), 17 deletions(-)
diff --git
a/paimon-python/pypaimon/execution/python/globalindex/btree/composite_btree.py
b/paimon-python/pypaimon/execution/python/globalindex/btree/composite_btree.py
new file mode 100644
index 0000000000..f369abbe17
--- /dev/null
+++
b/paimon-python/pypaimon/execution/python/globalindex/btree/composite_btree.py
@@ -0,0 +1,275 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# 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.
+
+"""Read Java composite BTree keys with left-prefix predicate bounds."""
+
+from dataclasses import dataclass
+import datetime
+from decimal import Decimal
+import numbers
+import struct
+
+from pypaimon.execution.python.globalindex.btree.btree_index_reader import
BTreeIndexReader
+from pypaimon.execution.python.globalindex.btree.btree_posting_list import
add_row_ids
+from pypaimon.execution.python.globalindex.btree.composite_key import (
+ CompositeKeySerializer, compare_nullable,
+)
+from pypaimon.globalindex.global_index_result import GlobalIndexResult
+from pypaimon.utils.roaring_bitmap import RoaringBitmap64
+
+
+MAX_INTERVALS = 256
+_POINT = frozenset(('equal', 'in', 'isNull'))
+_RANGE = frozenset(('greaterThan', 'greaterOrEqual', 'lessThan',
+ 'lessOrEqual', 'between', 'isNotNull'))
+
+
+@dataclass(frozen=True)
+class Bound:
+ values: tuple
+ after: bool
+
+
+@dataclass
+class CompositePlan:
+ intervals: list
+ bound_columns: int
+ equal_columns: int
+ field_count: int
+ covered_predicates: tuple
+
+ @property
+ def point_lookup(self):
+ return self.equal_columns == self.field_count
+
+
+def _compare_values(left, right, comparators):
+ for index, (a, b) in enumerate(zip(left, right)):
+ result = compare_nullable(a, b, comparators[index])
+ if result:
+ return result
+ return 0
+
+
+def compare_bounds(left, right, comparators):
+ difference = _compare_values(left.values, right.values, comparators)
+ if difference:
+ return difference
+ if len(left.values) != len(right.values):
+ if len(left.values) < len(right.values):
+ return 1 if left.after else -1
+ return -1 if right.after else 1
+ return (left.after > right.after) - (left.after < right.after)
+
+
+def compare_key_bound(key, bound, comparators):
+ difference = _compare_values(key, bound.values, comparators)
+ return difference if difference else (-1 if bound.after else 1)
+
+
+def _conjuncts(predicate):
+ if predicate.method != 'and':
+ return [predicate]
+ result = []
+ for child in predicate.literals:
+ result.extend(_conjuncts(child))
+ return result
+
+
+def _as_leaf(predicate):
+ if predicate.method != 'or':
+ return predicate
+ children = predicate.literals or []
+ if not children or any(child.method != 'equal' or child.field !=
children[0].field
+ for child in children):
+ return None
+ from pypaimon.common.predicate import Predicate
+ return Predicate('in', children[0].index, children[0].field,
+ [child.literals[0] for child in children])
+
+
+def _matches(value, predicate, comparator):
+ method = predicate.method
+ values = predicate.literals or []
+ if method == 'isNull':
+ return value is None
+ if method == 'isNotNull':
+ return value is not None
+ if value is None:
+ return False
+ if method == 'in':
+ return any(literal is not None and comparator(value, literal) == 0
+ for literal in values)
+ if any(literal is None for literal in values):
+ return False
+ if method == 'equal':
+ return comparator(value, values[0]) == 0
+ if method == 'between':
+ return (comparator(value, values[0]) >= 0
+ and comparator(value, values[1]) <= 0)
+ difference = comparator(value, values[0])
+ return {'greaterThan': difference > 0,
+ 'greaterOrEqual': difference >= 0,
+ 'lessThan': difference < 0,
+ 'lessOrEqual': difference <= 0}[method]
+
+
+def _literal_supported(type_name, literal):
+ if type_name == 'BOOLEAN':
+ return isinstance(literal, bool)
+ if type_name in ('TINYINT', 'SMALLINT', 'INT', 'INTEGER', 'BIGINT'):
+ return isinstance(literal, numbers.Integral) and not
isinstance(literal, bool)
+ if type_name.startswith(('CHAR', 'VARCHAR', 'STRING')):
+ return isinstance(literal, str)
+ if type_name in ('FLOAT', 'DOUBLE'):
+ # Java orders signed zeros separately, while Python row filtering does
not.
+ return (isinstance(literal, numbers.Real)
+ and not isinstance(literal, bool) and literal != 0)
+ if type_name.startswith(('DECIMAL', 'NUMERIC')):
+ return isinstance(literal, (Decimal, numbers.Integral)) and not
isinstance(literal, bool)
+ if type_name == 'DATE':
+ return ((isinstance(literal, datetime.date)
+ and not isinstance(literal, datetime.datetime))
+ or isinstance(literal, numbers.Integral)
+ and not isinstance(literal, bool))
+ if type_name.startswith('TIME') and not type_name.startswith('TIMESTAMP'):
+ return (isinstance(literal, datetime.time)
+ and literal.microsecond % 1000 == 0
+ or isinstance(literal, numbers.Integral)
+ and not isinstance(literal, bool))
+ if type_name.startswith('TIMESTAMP'):
+ return (isinstance(literal, datetime.datetime)
+ and not getattr(literal, 'nanosecond', 0))
+ return False
+
+
+def plan_composite(fields, predicate):
+ """Return None when the index cannot safely bound this predicate."""
+ try:
+ serializer = CompositeKeySerializer(fields)
+ except ValueError:
+ return None
+ comparators = serializer._comparators
+ conjuncts = _conjuncts(predicate)
+ prefixes = [()]
+ equal_columns = 0
+ range_predicates = []
+ covered_predicates = []
+ for field, comparator, scalar in zip(
+ fields, comparators, serializer._serializers):
+ column = []
+ sources = []
+ for child in conjuncts:
+ leaf = _as_leaf(child)
+ if leaf is not None and leaf.field == field.name and (
+ leaf.method in _POINT or leaf.method in _RANGE):
+ column.append(leaf)
+ sources.append(child)
+ if not column:
+ break
+ for leaf in column:
+ for literal in leaf.literals or []:
+ if literal is None:
+ continue
+ if not _literal_supported(field.type.type.upper(), literal):
+ return None
+ try:
+ converted = scalar.deserialize(scalar.serialize(literal))
+ if comparator(converted, literal) != 0:
+ return None
+ except (ArithmeticError, ValueError, TypeError, struct.error):
+ return None
+ point = [leaf for leaf in column if leaf.method in _POINT]
+ if not point:
+ range_predicates = column
+ covered_predicates.extend(sources)
+ break
+ selected = min(point, key=lambda leaf: len(leaf.literals or [None]))
+ candidates = ([None] if selected.method == 'isNull'
+ else selected.literals or [])
+ values = []
+ for value in candidates:
+ if value is None and (selected.method != 'isNull' or not
field.type.nullable):
+ continue
+ if all(_matches(value, leaf, comparator) for leaf in column):
+ if not any(compare_nullable(value, old, comparator) == 0 for
old in values):
+ values.append(value)
+ equal_columns += 1
+ covered_predicates.extend(sources)
+ if not values:
+ return CompositePlan([], equal_columns, equal_columns, len(fields),
+ tuple(covered_predicates))
+ if len(prefixes) * len(values) > MAX_INTERVALS:
+ return None
+ prefixes = [prefix + (value,) for prefix in prefixes for value in
values]
+ bound_columns = equal_columns + bool(range_predicates)
+ if not bound_columns:
+ return None
+ intervals = []
+ for prefix in prefixes:
+ lower, upper = Bound(prefix, False), Bound(prefix, True)
+ if range_predicates:
+ lower = Bound(prefix + (None,), True)
+ empty = False
+ for leaf in range_predicates:
+ if leaf.method == 'isNotNull':
+ continue
+ if None in (leaf.literals or []):
+ empty = True
+ break
+ if leaf.method in ('greaterThan', 'greaterOrEqual', 'between'):
+ candidate = Bound(prefix + (leaf.literals[0],),
+ leaf.method == 'greaterThan')
+ if compare_bounds(candidate, lower, comparators) > 0:
+ lower = candidate
+ if leaf.method in ('lessThan', 'lessOrEqual', 'between'):
+ candidate = Bound(prefix + (leaf.literals[-1],),
+ leaf.method != 'lessThan')
+ if compare_bounds(candidate, upper, comparators) < 0:
+ upper = candidate
+ if empty:
+ continue
+ if compare_bounds(lower, upper, comparators) < 0:
+ intervals.append((lower, upper))
+ return CompositePlan(intervals, bound_columns, equal_columns, len(fields),
+ tuple(covered_predicates))
+
+
+class CompositeBTreeIndexReader(BTreeIndexReader):
+ def query(self, plan):
+ result = RoaringBitmap64()
+ if self.min_key is None:
+ return GlobalIndexResult.create(result, is_exact=False)
+ for lower, upper in plan.intervals:
+ if (compare_key_bound(self.max_key, lower,
self.key_serializer._comparators) <= 0
+ or compare_key_bound(self.min_key, upper,
+ self.key_serializer._comparators) >=
0):
+ continue
+ if plan.point_lookup:
+ posting = self.reader.lookup(
+ self.key_serializer.serialize(lower.values))
+ if posting is not None:
+ add_row_ids(posting, self.footer.version, result)
+ continue
+ seek = lower.values + (None,) * (plan.field_count -
len(lower.values))
+ for entry in
self._entries_from(self.key_serializer.serialize(seek)):
+ key = self.key_serializer.deserialize(entry.key)
+ if compare_key_bound(key, upper,
self.key_serializer._comparators) >= 0:
+ break
+ if compare_key_bound(key, lower,
self.key_serializer._comparators) > 0:
+ add_row_ids(entry.value, self.footer.version, result)
+ return GlobalIndexResult.create(result, is_exact=False)
diff --git
a/paimon-python/pypaimon/execution/python/globalindex/btree/composite_key.py
b/paimon-python/pypaimon/execution/python/globalindex/btree/composite_key.py
new file mode 100644
index 0000000000..6e41c20292
--- /dev/null
+++ b/paimon-python/pypaimon/execution/python/globalindex/btree/composite_key.py
@@ -0,0 +1,153 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# 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.
+
+"""Java RowCompactedSerializer keys used by composite BTree indexes."""
+
+import math
+import struct
+
+from pypaimon.data.timestamp import Timestamp
+from pypaimon.execution.python.globalindex.key_serializer import (
+ KeySerializer, create_serializer, _read_var_len_int, _write_var_len_int,
+ _parse_precision,
+)
+
+
+class CompositeKeySerializer(KeySerializer):
+ def __init__(self, fields):
+ self._types = [field.type.type.upper() for field in fields]
+ self._serializers = [create_serializer(field.type) for field in fields]
+ self._comparators = [
+ _java_float_compare if type_name in ('FLOAT', 'DOUBLE')
+ else serializer.create_comparator()
+ for type_name, serializer in zip(self._types, self._serializers)
+ ]
+ self._header_size = (len(fields) + 15) // 8
+
+ def serialize(self, key):
+ if len(key) != len(self._serializers):
+ raise ValueError("Wrong composite key arity")
+ header = bytearray(self._header_size)
+ values = bytearray()
+ for index, (value, serializer, type_name) in enumerate(
+ zip(key, self._serializers, self._types)):
+ if value is None:
+ header[(index + 8) // 8] |= 1 << ((index + 8) % 8)
+ continue
+ if type_name == 'FLOAT' and math.isnan(float(value)):
+ encoded = struct.pack('<I', 0x7fc00000)
+ elif type_name == 'DOUBLE' and math.isnan(float(value)):
+ encoded = struct.pack('<Q', 0x7ff8000000000000)
+ else:
+ encoded = serializer.serialize(value)
+ if _variable_length(type_name):
+ values.extend(_write_var_len_int(len(encoded)))
+ values.extend(encoded)
+ return bytes(header + values)
+
+ def deserialize(self, data):
+ if len(data) < self._header_size:
+ raise ValueError("Truncated composite key")
+ position = self._header_size
+ result = []
+ for index, (serializer, type_name) in enumerate(
+ zip(self._serializers, self._types)):
+ if data[(index + 8) // 8] & (1 << ((index + 8) % 8)):
+ result.append(None)
+ continue
+ if _variable_length(type_name):
+ length, position = _read_var_len_int(data, position)
+ else:
+ length = _fixed_length(type_name)
+ if type_name.startswith('TIMESTAMP') and _parse_precision(
+ type_name, 6) > 3:
+ _, end = _read_var_len_int(data, position + 8)
+ length = end - position
+ end = position + length
+ if end > len(data):
+ raise ValueError("Truncated composite key value")
+ encoded = data[position:end]
+ if type_name.startswith('TIMESTAMP'):
+ milliseconds = struct.unpack('<q', encoded[:8])[0]
+ nanos = (_read_var_len_int(encoded, 8)[0]
+ if len(encoded) > 8 else 0)
+ result.append(Timestamp.from_epoch_millis(milliseconds, nanos))
+ elif type_name == 'DATE' or (type_name.startswith('TIME')
+ and not
type_name.startswith('TIMESTAMP')):
+ result.append(struct.unpack('<i', encoded)[0])
+ else:
+ result.append(serializer.deserialize(encoded))
+ position = end
+ if position != len(data):
+ raise ValueError("Trailing composite key bytes")
+ return tuple(result)
+
+ def create_comparator(self):
+ def compare(left, right):
+ for a, b, comparator in zip(left, right, self._comparators):
+ difference = compare_nullable(a, b, comparator)
+ if difference:
+ return difference
+ return 0
+ return compare
+
+
+def _variable_length(type_name):
+ return (type_name.startswith(('CHAR', 'VARCHAR', 'STRING'))
+ or type_name.startswith(('DECIMAL', 'NUMERIC'))
+ and _decimal_precision(type_name) > 18)
+
+
+def _decimal_precision(type_name):
+ from pypaimon.execution.python.globalindex.key_serializer import
_parse_decimal_params
+ return _parse_decimal_params(type_name)[0]
+
+
+def _fixed_length(type_name):
+ if type_name in ('BOOLEAN', 'TINYINT'):
+ return 1
+ if type_name == 'SMALLINT':
+ return 2
+ if type_name in ('INT', 'INTEGER', 'DATE', 'FLOAT') or (
+ type_name.startswith('TIME') and not
type_name.startswith('TIMESTAMP')):
+ return 4
+ if type_name in ('BIGINT', 'DOUBLE') or type_name.startswith(('DECIMAL',
'NUMERIC')):
+ return 8
+ if type_name.startswith('TIMESTAMP'):
+ # The caller handles the optional trailing varint separately.
+ return 8
+ raise ValueError("Unsupported composite key type: %s" % type_name)
+
+
+def compare_nullable(left, right, comparator):
+ if left is None:
+ return 0 if right is None else -1
+ if right is None:
+ return 1
+ return comparator(left, right)
+
+
+def _java_float_compare(left, right):
+ left, right = float(left), float(right)
+ if math.isnan(left):
+ return 0 if math.isnan(right) else 1
+ if math.isnan(right):
+ return -1
+ if left == right == 0:
+ return (math.copysign(1, left) > math.copysign(1, right)) - (
+ math.copysign(1, left) < math.copysign(1, right))
+ return (left > right) - (left < right)
diff --git
a/paimon-python/pypaimon/execution/python/globalindex/data_evolution_global_index_coverage.py
b/paimon-python/pypaimon/execution/python/globalindex/data_evolution_global_index_coverage.py
index 19aa1c251f..fe051153cf 100644
---
a/paimon-python/pypaimon/execution/python/globalindex/data_evolution_global_index_coverage.py
+++
b/paimon-python/pypaimon/execution/python/globalindex/data_evolution_global_index_coverage.py
@@ -45,11 +45,12 @@ class DataEvolutionGlobalIndexCoverage:
meta = index_file.global_index_meta
if meta is None:
continue
+ if getattr(index_file, 'index_type', None) == 'btree' and
meta.extra_field_ids:
+ continue
row_range = Range(meta.row_range_start, meta.row_range_end)
self._add_coverage(meta.index_field_id, row_range)
- if meta.extra_field_ids is not None:
- for extra_field_id in meta.extra_field_ids:
- self._add_coverage(extra_field_id, row_range)
+ for field_id in meta.extra_field_ids or []:
+ self._add_coverage(field_id, row_range)
def unindexed_ranges(
self,
@@ -70,6 +71,12 @@ class DataEvolutionGlobalIndexCoverage:
field_ids.add(field.id)
return self._unindexed_ranges(field_ids, search_mode)
+ def indexed_ranges(self, field_ids):
+ return self._indexed_ranges(field_ids)
+
+ def unindexed_ranges_from_indexed(self, indexed_ranges, search_mode=None):
+ return self._unindexed_ranges((), search_mode, indexed_ranges)
+
def _add_coverage(self, field_id: int, row_range: Range) -> None:
self._coverage_by_field.setdefault(field_id, []).append(row_range)
@@ -89,6 +96,7 @@ class DataEvolutionGlobalIndexCoverage:
self,
field_ids: Collection[int],
search_mode: Optional[GlobalIndexSearchMode] = None,
+ indexed_ranges=None,
) -> List[Range]:
if search_mode is None:
search_mode = _scalar_index_search_mode(self._table)
@@ -106,7 +114,8 @@ class DataEvolutionGlobalIndexCoverage:
data_ranges = [Range(0, next_row_id - 1)]
indexed_ranges = Range.sort_and_merge_overlap(
- self._indexed_ranges(field_ids), True)
+ self._indexed_ranges(field_ids) if indexed_ranges is None
+ else indexed_ranges, True)
unindexed = []
for data_range in Range.sort_and_merge_overlap(data_ranges, True):
unindexed.extend(data_range.exclude(indexed_ranges))
diff --git
a/paimon-python/pypaimon/execution/python/globalindex/data_evolution_global_index_scanner.py
b/paimon-python/pypaimon/execution/python/globalindex/data_evolution_global_index_scanner.py
index 6e9d562332..eb8627f5d8 100644
---
a/paimon-python/pypaimon/execution/python/globalindex/data_evolution_global_index_scanner.py
+++
b/paimon-python/pypaimon/execution/python/globalindex/data_evolution_global_index_scanner.py
@@ -18,6 +18,7 @@
"""Scanner for shard-based global indexes on data-evolution tables."""
from concurrent.futures import ThreadPoolExecutor
+import threading
from typing import Collection, List, Optional
from pypaimon.execution.python.globalindex.global_index_evaluator import (
@@ -39,6 +40,10 @@ from pypaimon.utils.range import Range
_SUPPORTED_SCALAR_INDEX_TYPES = frozenset(('btree', 'bitmap'))
+class CompositeIndexReadError(RuntimeError):
+ """A selected composite index is unreadable; falling back may hide
corruption."""
+
+
class DataEvolutionGlobalIndexScanner:
"""Scanner for shard-based global indexes."""
@@ -53,6 +58,7 @@ class DataEvolutionGlobalIndexScanner:
table=None,
snapshot=None,
partition_filter=None,
+ composite_files=None,
):
index_files = _supported_scalar_index_files(index_files)
self._options = options or CoreOptions(Options.from_none())
@@ -60,6 +66,12 @@ class DataEvolutionGlobalIndexScanner:
max_workers=thread_num or 32
)
self._fields = fields
+ self._file_io = file_io
+ self._index_path = index_path
+ self._composite_groups = _group_composite_files(composite_files or [])
+ self._composite_readers = {}
+ self._composite_reader_lock = threading.Lock()
+ self._composite_reader_locks = {}
self._coverage = (
DataEvolutionGlobalIndexCoverage(table, snapshot,
partition_filter, index_files)
if table is not None
@@ -139,7 +151,7 @@ class DataEvolutionGlobalIndexScanner:
@staticmethod
def create(table, index_files=None, partition_filter=None, predicate=None,
- snapshot=None) -> Optional['DataEvolutionGlobalIndexScanner']:
+ snapshot=None, allow_composite=False) ->
Optional['DataEvolutionGlobalIndexScanner']:
"""Create a DataEvolutionGlobalIndexScanner.
Can be called in two ways:
@@ -152,15 +164,19 @@ class DataEvolutionGlobalIndexScanner:
from pypaimon.index.index_file_handler import IndexFileHandler
if index_files is not None:
- index_files = _supported_scalar_index_files(index_files)
- if len(index_files) == 0:
+ index_files = list(index_files)
+ scalar_files = _supported_scalar_index_files(index_files)
+ composite_files = [file for file in index_files
+ if allow_composite and
_is_composite_btree(file)]
+ if not scalar_files and not composite_files:
return None
core_options = _core_options(table)
return DataEvolutionGlobalIndexScanner(
fields=table.fields,
file_io=table.file_io,
index_path=table.path_factory().global_index_path_factory().index_path(),
- index_files=index_files,
+ index_files=scalar_files,
+ composite_files=composite_files,
thread_num=core_options.global_index_thread_num(),
options=core_options,
table=table,
@@ -180,7 +196,8 @@ class DataEvolutionGlobalIndexScanner:
if partition_filter is not None:
if not partition_filter.test(entry.partition):
return False
- if not is_supported_scalar_index(entry.index_file):
+ if not (is_supported_scalar_index(entry.index_file)
+ or allow_composite and
_is_composite_btree(entry.index_file)):
return False
global_index_meta = entry.index_file.global_index_meta
if global_index_meta is None:
@@ -208,6 +225,8 @@ class DataEvolutionGlobalIndexScanner:
file_io=table.file_io,
index_path=table.path_factory().global_index_path_factory().index_path(),
index_files=scanned_index_files,
+ composite_files=[file for file in scanned_index_files
+ if allow_composite and _is_composite_btree(file)],
thread_num=core_options.global_index_thread_num(),
options=core_options,
table=table,
@@ -217,29 +236,238 @@ class DataEvolutionGlobalIndexScanner:
def scan(self, predicate: Optional[Predicate]) ->
Optional[GlobalIndexResult]:
"""Scan the global index with the given predicate."""
- return self._evaluator.evaluate(predicate)
+ evaluation = self.scan_with_coverage(predicate)
+ return evaluation.result if evaluation is not None else None
def scan_with_coverage(
self, predicate: Optional[Predicate]
) -> Optional[GlobalIndexEvaluation]:
+ if not self._composite_groups or predicate is None:
+ return self._evaluator.evaluate_with_contributing_fields(predicate)
+ return self._evaluate(predicate)
+
+ def _evaluate(self, predicate):
+ from pypaimon.execution.python.globalindex.btree.composite_btree
import (
+ _conjuncts,
+ plan_composite,
+ )
+
+ fields_by_id = {field.id: field for field in self._fields}
+ candidates = []
+ for field_ids, files in self._composite_groups.items():
+ if any(field_id not in fields_by_id for field_id in field_ids):
+ continue
+ fields = [fields_by_id[field_id] for field_id in field_ids]
+ plan = plan_composite(fields, predicate)
+ if plan is None:
+ continue
+ if plan.bound_columns == 1 and self._coverage is not None:
+ composite_ranges = Range.sort_and_merge_overlap(list(files),
True)
+ scalar_ranges = self._coverage.indexed_ranges([field_ids[0]])
+ if Range.and_(composite_ranges, scalar_ranges) ==
composite_ranges:
+ continue
+ try:
+ selected = self._select_composite_files(files, fields, plan)
+ except Exception as error:
+ raise CompositeIndexReadError("Cannot read composite index
metadata") from error
+ if selected is None:
+ continue
+ candidates.append((plan.bound_columns, plan.equal_columns,
+ -len(field_ids), field_ids, fields, files,
selected, plan))
+ if candidates:
+ _, _, _, field_ids, fields, files, selected, plan = max(
+ candidates, key=lambda candidate: candidate[:3])
+ if predicate.method == 'and':
+ key_fields = {field.name for field in fields}
+ remaining = []
+ for child in _conjuncts(predicate):
+ if any(child is used for used in plan.covered_predicates):
+ continue
+ referenced_fields = _get_all_fields(child)
+ # Key residuals stay in the final row filter.
+ if referenced_fields and referenced_fields <= key_fields:
+ continue
+ remaining.append(child)
+ if remaining:
+ rest = (remaining[0] if len(remaining) == 1
+ else Predicate('and', None, None, remaining))
+ if not self._may_use_composite(rest):
+ rest_future = self._evaluator._visit_async(rest)
+ try:
+ composite = self._read_composite(
+ field_ids, fields, files, selected, plan)
+ except BaseException:
+ try:
+ rest_future.result()
+ except BaseException:
+ pass
+ raise
+ return self._combine_evaluations(
+ [composite, rest_future.result()], 'and')
+ composite = self._read_composite(
+ field_ids, fields, files, selected, plan)
+ return self._combine_evaluations(
+ [composite, self._evaluate(rest)], 'and')
+ return self._read_composite(field_ids, fields, files, selected,
plan)
+
+ if predicate.method in ('and', 'or'):
+ scalar_children = []
+ composite_children = []
+ for child in predicate.literals:
+ if self._may_use_composite(child):
+ composite_children.append(child)
+ else:
+ scalar_children.append(child)
+ if not composite_children:
+ return
self._evaluator.evaluate_with_contributing_fields(predicate)
+ scalar_future = None
+ if scalar_children:
+ scalar = (scalar_children[0] if len(scalar_children) == 1
+ else Predicate(predicate.method, None, None,
scalar_children))
+ scalar_future = self._evaluator._visit_async(scalar)
+ evaluations = [self._evaluate(child) for child in
composite_children]
+ if scalar_future is not None:
+ evaluations.append(scalar_future.result())
+ return self._combine_evaluations(evaluations, predicate.method)
return self._evaluator.evaluate_with_contributing_fields(predicate)
+ def _may_use_composite(self, predicate):
+ from pypaimon.execution.python.globalindex.btree.composite_btree
import plan_composite
+
+ fields_by_id = {field.id: field for field in self._fields}
+ predicate_fields = _get_all_fields(predicate)
+ for field_ids in self._composite_groups:
+ if (all(field_id in fields_by_id for field_id in field_ids)
+ and fields_by_id[field_ids[0]].name in predicate_fields):
+ fields = [fields_by_id[field_id] for field_id in field_ids]
+ if plan_composite(fields, predicate) is not None:
+ return True
+ return (predicate.method in ('and', 'or')
+ and any(self._may_use_composite(child) for child in
predicate.literals))
+
+ def _combine_evaluations(self, evaluations, method):
+ if method == 'or' and any(item is None for item in evaluations):
+ return None
+ present = [item for item in evaluations if item is not None]
+ if not present:
+ return None
+ result = present[0].result
+ ranges = self._evaluation_ranges(present[0])
+ field_ids = set(present[0].contributing_field_ids)
+ for item in present[1:]:
+ result = (result.or_(item.result) if method == 'or'
+ else result.and_(item.result))
+ ranges = Range.and_(ranges, self._evaluation_ranges(item))
+ field_ids.update(item.contributing_field_ids)
+ if len(present) != len(evaluations):
+ result = GlobalIndexResult.create(result.results(), is_exact=False)
+ return GlobalIndexEvaluation(result, frozenset(field_ids),
tuple(ranges))
+
+ def _evaluation_ranges(self, evaluation):
+ if evaluation.indexed_ranges is not None:
+ return list(evaluation.indexed_ranges)
+ if self._coverage is None:
+ return []
+ return self._coverage.indexed_ranges(evaluation.contributing_field_ids)
+
+ def _select_composite_files(self, files, fields, plan):
+ from pypaimon.execution.python.globalindex.btree.composite_btree
import (
+ compare_key_bound,
+ )
+ from pypaimon.execution.python.globalindex.btree.composite_key import (
+ CompositeKeySerializer,
+ )
+ from pypaimon.globalindex.sorted_index_file_meta import
SortedIndexFileMeta
+
+ if not plan.intervals:
+ return []
+ serializer = CompositeKeySerializer(fields)
+ selected = []
+ budget = self._options.btree_index_fallback_scan_max_size()
+ for row_range, metas in files.items():
+ shard = []
+ for meta in metas:
+ if meta.metadata is None:
+ return None
+ index_meta = SortedIndexFileMeta.deserialize(meta.metadata)
+ if index_meta.first_key is not None and plan.intervals:
+ first = serializer.deserialize(index_meta.first_key)
+ last = serializer.deserialize(index_meta.last_key)
+ if not any(compare_key_bound(last, lower,
serializer._comparators) > 0
+ and compare_key_bound(first, upper,
+ serializer._comparators)
< 0
+ for lower, upper in plan.intervals):
+ continue
+ if plan.intervals:
+ shard.append((row_range, meta))
+ if not plan.point_lookup and sum(meta.file_size for _, meta in
shard) > budget:
+ return None
+ selected.extend(shard)
+ return selected
+
+ def _read_composite(self, field_ids, fields, files, selected, plan):
+ from pypaimon.execution.python.globalindex.btree.composite_btree
import (
+ CompositeBTreeIndexReader,
+ )
+ from pypaimon.execution.python.globalindex.btree.composite_key import (
+ CompositeKeySerializer,
+ )
+
+ serializer = CompositeKeySerializer(fields)
+ futures = []
+ for row_range, meta in selected:
+ def read(range_=row_range, io_meta=meta):
+ key = (field_ids, io_meta.external_path or io_meta.file_name)
+ with self._composite_reader_lock:
+ reader = self._composite_readers.get(key)
+ reader_lock = (self._composite_reader_locks.setdefault(
+ key, threading.Lock()) if reader is None else None)
+ if reader is None:
+ with reader_lock:
+ with self._composite_reader_lock:
+ reader = self._composite_readers.get(key)
+ if reader is None:
+ reader = CompositeBTreeIndexReader(
+ serializer, self._file_io, self._index_path,
io_meta)
+ with self._composite_reader_lock:
+ self._composite_readers[key] = reader
+ return reader.query(plan).offset(range_.from_)
+ futures.append(self._executor.submit(read))
+ result = GlobalIndexResult.create_empty()
+ failure = None
+ for future in futures:
+ try:
+ result = result.or_(future.result())
+ except Exception as error:
+ if failure is None:
+ failure = error
+ if failure is not None:
+ raise CompositeIndexReadError("Cannot read composite BTree file")
from failure
+ result = GlobalIndexResult.create(result.results(), is_exact=False)
+ return GlobalIndexEvaluation(
+ result, frozenset(field_ids[:plan.bound_columns]),
+ tuple(Range.sort_and_merge_overlap(list(files), True)))
+
def unindexed_rows(self, predicate: Optional[Predicate],
search_mode=None,
- contributing_field_ids=None) -> GlobalIndexResult:
+ contributing_field_ids=None, indexed_ranges=None) ->
GlobalIndexResult:
"""Return coarse row ids not covered by global indexes."""
return GlobalIndexResult.from_ranges(self.unindexed_ranges(
predicate,
search_mode=search_mode,
contributing_field_ids=contributing_field_ids,
+ indexed_ranges=indexed_ranges,
))
def unindexed_ranges(self, predicate: Optional[Predicate],
search_mode=None,
- contributing_field_ids=None) -> List[Range]:
+ contributing_field_ids=None, indexed_ranges=None) ->
List[Range]:
"""Return row ranges not covered by global indexes."""
if self._coverage is None:
return []
+ if indexed_ranges is not None:
+ return self._coverage.unindexed_ranges_from_indexed(
+ indexed_ranges, search_mode=search_mode)
if contributing_field_ids is not None:
return self._coverage.unindexed_ranges(
contributing_field_ids, search_mode=search_mode)
@@ -249,6 +477,8 @@ class DataEvolutionGlobalIndexScanner:
def close(self):
"""Close the scanner and release resources."""
self._evaluator.close()
+ for reader in self._composite_readers.values():
+ reader.close()
self._executor.shutdown(wait=False)
def __enter__(self) -> 'DataEvolutionGlobalIndexScanner':
@@ -311,6 +541,32 @@ def _supported_scalar_index_files(index_files):
if is_supported_scalar_index(index_file)]
+def _is_composite_btree(index_file):
+ meta = index_file.global_index_meta
+ return (index_file.index_type == 'btree' and meta is not None
+ and bool(meta.extra_field_ids)
+ and len(set((meta.index_field_id,) + tuple(meta.extra_field_ids)))
+ == 1 + len(meta.extra_field_ids))
+
+
+def _group_composite_files(index_files):
+ groups = {}
+ for index_file in index_files:
+ if not _is_composite_btree(index_file):
+ continue
+ meta = index_file.global_index_meta
+ field_ids = (meta.index_field_id,) + tuple(meta.extra_field_ids)
+ row_range = Range(meta.row_range_start, meta.row_range_end)
+ io_meta = GlobalIndexIOMeta(
+ file_name=index_file.file_name,
+ file_size=index_file.file_size,
+ metadata=meta.index_meta,
+ external_path=index_file.external_path,
+ )
+ groups.setdefault(field_ids, {}).setdefault(row_range,
[]).append(io_meta)
+ return groups
+
+
def _core_options(table):
options = getattr(table, "options", None)
if options is None:
diff --git
a/paimon-python/pypaimon/execution/python/read/scanner/file_scanner.py
b/paimon-python/pypaimon/execution/python/read/scanner/file_scanner.py
index 43a1493d5f..b3caa043ca 100755
--- a/paimon-python/pypaimon/execution/python/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/execution/python/read/scanner/file_scanner.py
@@ -562,6 +562,7 @@ class FileScanner:
partition_filter=self.partition_key_predicate,
predicate=predicate,
snapshot=snapshot,
+ allow_composite=True,
)
if scanner is None:
return None
@@ -570,6 +571,9 @@ class FileScanner:
if evaluation is None:
return None
scalar_mode = self.table.options.scalar_index_search_mode()
+ coverage_args = {}
+ if evaluation.indexed_ranges is not None:
+ coverage_args["indexed_ranges"] = evaluation.indexed_ranges
return _GlobalIndexPlanningResult(
evaluation.result,
scanner.unindexed_ranges(
@@ -577,9 +581,15 @@ class FileScanner:
search_mode=scalar_mode,
contributing_field_ids=(
evaluation.contributing_field_ids),
+ **coverage_args,
),
)
- except Exception:
+ except Exception as error:
+ from
pypaimon.execution.python.globalindex.data_evolution_global_index_scanner
import (
+ CompositeIndexReadError,
+ )
+ if isinstance(error, CompositeIndexReadError):
+ raise
return None
def read_manifest_entries(self, manifest_files: List[ManifestFileMeta],
diff --git a/paimon-python/pypaimon/globalindex/global_index_evaluation.py
b/paimon-python/pypaimon/globalindex/global_index_evaluation.py
index 92a33ff71b..bb4a4611a0 100644
--- a/paimon-python/pypaimon/globalindex/global_index_evaluation.py
+++ b/paimon-python/pypaimon/globalindex/global_index_evaluation.py
@@ -17,9 +17,10 @@
"""Global index evaluation result shared with scan planning."""
-from typing import FrozenSet, NamedTuple
+from typing import FrozenSet, NamedTuple, Optional, Tuple
from pypaimon.globalindex.global_index_result import GlobalIndexResult
+from pypaimon.utils.range import Range
class GlobalIndexEvaluation(NamedTuple):
@@ -27,3 +28,4 @@ class GlobalIndexEvaluation(NamedTuple):
result: GlobalIndexResult
contributing_field_ids: FrozenSet[int]
+ indexed_ranges: Optional[Tuple[Range, ...]] = None
diff --git a/paimon-python/pypaimon/tests/composite_btree_fallback_test.py
b/paimon-python/pypaimon/tests/composite_btree_fallback_test.py
index 2f887d6522..4cbce647c6 100644
--- a/paimon-python/pypaimon/tests/composite_btree_fallback_test.py
+++ b/paimon-python/pypaimon/tests/composite_btree_fallback_test.py
@@ -16,22 +16,42 @@
# under the License.
import unittest
+import shutil
+import tempfile
+import threading
+from datetime import time
from dataclasses import replace
+from decimal import Decimal
+from pathlib import Path
from unittest.mock import patch
import pyarrow as pa
import pytest
from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.data.timestamp import Timestamp
from pypaimon.execution.python.globalindex.data_evolution_global_index_scanner
import (
+ CompositeIndexReadError,
DataEvolutionGlobalIndexScanner,
is_supported_scalar_index,
)
+from pypaimon.execution.python.globalindex.btree.composite_btree import (
+ CompositeBTreeIndexReader,
+ plan_composite,
+)
+from pypaimon.execution.python.globalindex.btree.btree_index_writer import
BTreeIndexWriter
+from pypaimon.execution.python.globalindex.btree.btree_index_reader import
BTreeIndexReader
+from pypaimon.execution.python.globalindex.btree.composite_key import (
+ CompositeKeySerializer,
+)
from pypaimon.globalindex.global_index_meta import GlobalIndexMeta
+from pypaimon.execution.python.globalindex.key_serializer import
create_serializer
from pypaimon.globalindex.sorted_index_file_meta import SortedIndexFileMeta
from pypaimon.index.index_file_handler import IndexFileHandler
from pypaimon.index.index_file_meta import IndexFileMeta
from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
+from pypaimon.filesystem.local_file_io import LocalFileIO
+from pypaimon.schema.data_types import AtomicType, DataField
from pypaimon.execution.python.read.scanner.file_scanner import (FileScanner)
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.tests.data_evolution_test_helpers import (
@@ -53,6 +73,17 @@ def _composite_file():
)
+def _java_composite_file(name='java-v2-composite-int.btree', field_ids=(0, 1)):
+ path = Path(__file__).parent / 'resources/btree' / name
+ return IndexFileMeta(
+ index_type='btree', file_name=path.name,
+ external_path=str(path), file_size=path.stat().st_size, row_count=7,
+ global_index_meta=GlobalIndexMeta(
+ 0, 6, field_ids[0], list(field_ids[1:]),
+ Path(str(path) + '.meta').read_bytes()),
+ )
+
+
class CompositeBTreeFallbackTest(
BatchModeMixin, DataEvolutionTestBase, unittest.TestCase):
@@ -120,7 +151,10 @@ class CompositeBTreeFallbackTest(
table = self._table_with_partial_index()
snapshot = table.snapshot_manager().get_latest_snapshot()
scalar_entries = IndexFileHandler(table).scan(snapshot)
- composite_entry = IndexManifestEntry(0, GenericRow([], []), 0,
_composite_file())
+ unsupported = _composite_file()
+ unsupported = replace(unsupported, global_index_meta=replace(
+ unsupported.global_index_meta, extra_field_ids=[99]))
+ composite_entry = IndexManifestEntry(0, GenericRow([], []), 0,
unsupported)
b = table.new_read_builder().new_predicate_builder()
cases = [
(b.equal('a', 2), True, [2]),
@@ -180,3 +214,430 @@ class CompositeBTreeFallbackTest(
builder = table.new_read_builder().with_filter(predicate)
rows = builder.new_read().to_arrow(builder.new_scan().plan().splits())
return sorted(rows.column('a').to_pylist(), key=lambda value: (value
is not None, value))
+
+ @pytest.mark.python_plan
+ def test_java_composite_index_and_partial_coverage(self):
+ schema = pa.schema([('a', pa.int32()), ('b', pa.int32()), ('c',
pa.int32())])
+ table = self._create_table(pa_schema=schema)
+ self._write_arrow(table, pa.table({
+ 'a': [None, None, 1, 1, 1, 1, 2],
+ 'b': [None, 2, None, 1, 2, 2, 1],
+ 'c': [None, 2, None, 1, 2, 2, 1],
+ }, schema=schema))
+ self._write_arrow(table, pa.table({
+ 'a': [1, 2], 'b': [2, 1], 'c': [2, 1],
+ }, schema=schema))
+ entries = [IndexManifestEntry(0, GenericRow([], []), 0, file) for file
in (
+ _java_composite_file(),
+ _java_composite_file('java-v2-composite-ac.btree', (0, 2)),
+ _java_composite_file('java-v2-composite-reversed.btree', (1, 0)),
+ )]
+ table.create_global_index('a')
+ entries.extend(IndexFileHandler(table).scan(
+ table.snapshot_manager().get_latest_snapshot()))
+ builder = table.new_read_builder().new_predicate_builder()
+ cases = [
+ PredicateBuilder.and_predicates([
+ builder.equal('a', 1), builder.equal('b', 2)]),
+ PredicateBuilder.and_predicates([
+ builder.equal('a', 1), builder.equal('c', 2)]),
+ builder.equal('a', 1),
+ builder.is_null('a'),
+ PredicateBuilder.and_predicates([
+ builder.is_null('a'), builder.is_null('b')]),
+ PredicateBuilder.and_predicates([
+ builder.equal('a', 1), builder.greater_than('b', 1)]),
+ PredicateBuilder.and_predicates([
+ builder.equal('a', 1), builder.greater_than('b', 0),
+ builder.less_than('b', 2)]),
+ PredicateBuilder.and_predicates([
+ builder.equal('a', 1), builder.equal('b', 1),
+ builder.equal('c', 2)]),
+ builder.is_in('a', [1, 2]),
+ PredicateBuilder.and_predicates([
+ builder.equal('a', 1), builder.is_in('b', [2, None])]),
+ builder.greater_than('a', 1),
+ builder.is_not_null('b'),
+ builder.less_than('b', 1.5),
+ builder.is_in('b', list(range(257))),
+ builder.equal('a', 99),
+ builder.equal('b', 1),
+ builder.not_equal('a', 1),
+ PredicateBuilder.or_predicates([
+ builder.equal('a', 1), builder.equal('a', 2)]),
+ PredicateBuilder.or_predicates([
+ PredicateBuilder.and_predicates([
+ builder.equal('a', 1), builder.equal('b', 2)]),
+ builder.is_null('a')]),
+ PredicateBuilder.or_predicates([
+ builder.equal('b', 1), builder.equal('c', 2)]),
+ ]
+ reads = []
+ original = CompositeBTreeIndexReader.query
+
+ def observed(reader, plan):
+ reads.append(reader.io_meta.file_name)
+ return original(reader, plan)
+
+ def scan_indexes(handler, snapshot, entry_filter=None):
+ return [entry for entry in entries
+ if entry_filter is None or entry_filter(entry)]
+
+ with patch.object(IndexFileHandler, 'scan', scan_indexes),
patch.object(
+ CompositeBTreeIndexReader, 'query', observed):
+ for predicate in cases:
+ baseline = self._rows(
+ table.copy({'global-index.enabled': 'false'}), predicate)
+ for mode in ('full', 'detail'):
+ with self.subTest(predicate=predicate, mode=mode):
+ actual = self._rows(
+ table.copy({'scalar-index.search-mode': mode}),
predicate)
+ self.assertEqual(baseline, actual)
+ exact = PredicateBuilder.and_predicates([
+ builder.equal('a', 1), builder.equal('b', 2)])
+ with DataEvolutionGlobalIndexScanner.create(
+ table, predicate=exact, allow_composite=True) as scanner:
+ evaluation = scanner.scan_with_coverage(exact)
+ self.assertEqual([4, 5], evaluation.result.results().to_list())
+ self.assertEqual((Range(0, 6),), evaluation.indexed_ranges)
+ self.assertEqual([Range(7, 8)], scanner.unindexed_ranges(
+ exact, search_mode=table.copy({
+ 'scalar-index.search-mode': 'full'
+ }).options.scalar_index_search_mode(),
+ indexed_ranges=evaluation.indexed_ranges))
+ fast = self._rows(table.copy({'scalar-index.search-mode':
'fast'}), exact)
+ self.assertEqual(2, len(fast))
+ read_count = len(reads)
+ limited = table.copy({
+ 'scalar-index.search-mode': 'full',
+ 'btree-index.fallback-scan-max-size': '1B',
+ })
+ self.assertEqual(
+ self._rows(table.copy({'global-index.enabled': 'false'}),
+ builder.equal('b', 1)),
+ self._rows(limited, builder.equal('b', 1)))
+ self.assertEqual(read_count, len(reads))
+ self.assertTrue(reads, 'composite SST reader was not used')
+ self.assertIn('java-v2-composite-int.btree', reads)
+ self.assertIn('java-v2-composite-ac.btree', reads)
+ self.assertIn('java-v2-composite-reversed.btree', reads)
+
+ @pytest.mark.python_plan
+ def test_explicit_composite_files_and_remaining_scalar_predicate(self):
+ schema = pa.schema([('a', pa.int32()), ('b', pa.int32()), ('c',
pa.int32())])
+ table = self._create_table(pa_schema=schema)
+ self._write_arrow(table, pa.table({
+ 'a': [None, None, 1, 1, 1, 1, 2],
+ 'b': [None, 2, None, 1, 2, 2, 1],
+ 'c': [0, 0, 0, 0, 0, 1, 0],
+ }, schema=schema))
+ composite = _java_composite_file()
+ builder = table.new_read_builder().new_predicate_builder()
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+
+ self.assertIsNone(DataEvolutionGlobalIndexScanner.create(
+ table, index_files=[composite], snapshot=snapshot))
+ with DataEvolutionGlobalIndexScanner.create(
+ table, index_files=[composite], snapshot=snapshot,
+ allow_composite=True) as scanner:
+ evaluation = scanner.scan_with_coverage(builder.equal('a', 1))
+ self.assertEqual([2, 3, 4, 5],
evaluation.result.results().to_list())
+ self.assertEqual((Range(0, 6),), evaluation.indexed_ranges)
+
+ table.create_global_index('c')
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ scalar_files = [entry.index_file for entry in
IndexFileHandler(table).scan(snapshot)]
+ predicate = PredicateBuilder.and_predicates([
+ builder.equal('a', 1), builder.equal('b', 2), builder.equal('c',
1),
+ ])
+ gate = threading.Barrier(2, timeout=3)
+ original = BTreeIndexReader.__init__
+
+ def concurrent_init(reader, *args, **kwargs):
+ gate.wait()
+ original(reader, *args, **kwargs)
+
+ with patch.object(BTreeIndexReader, '__init__', concurrent_init):
+ with DataEvolutionGlobalIndexScanner.create(
+ table, index_files=[composite] + scalar_files,
snapshot=snapshot,
+ allow_composite=True) as scanner:
+ evaluation = scanner.scan_with_coverage(predicate)
+ self.assertEqual([5], evaluation.result.results().to_list())
+ self.assertEqual(frozenset([0, 1, 2]),
evaluation.contributing_field_ids)
+ self.assertEqual((Range(0, 6),), evaluation.indexed_ranges)
+
+ @pytest.mark.python_plan
+ def test_composite_and_scalar_use_intersected_coverage(self):
+ schema = pa.schema([('a', pa.int32()), ('b', pa.int32()), ('c',
pa.int32())])
+ table = self._create_table(pa_schema=schema)
+ self._write_arrow(table, pa.table({
+ 'a': [None, None, 1, 1], 'b': [None, 2, None, 1],
+ 'c': [0, 0, 0, 1],
+ }, schema=schema))
+ table.create_global_index('c')
+ self._write_arrow(table, pa.table({
+ 'a': [1, 1, 2], 'b': [2, 2, 1], 'c': [0, 1, 0],
+ }, schema=schema))
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ scalar_entries = IndexFileHandler(table).scan(snapshot)
+ composite = _java_composite_file()
+ predicate = PredicateBuilder.and_predicates([
+ table.new_read_builder().new_predicate_builder().equal('a', 1),
+ table.new_read_builder().new_predicate_builder().equal('c', 1),
+ ])
+
+ with DataEvolutionGlobalIndexScanner.create(
+ table, index_files=[composite] + [e.index_file for e in
scalar_entries],
+ snapshot=snapshot, allow_composite=True) as scanner:
+ evaluation = scanner.scan_with_coverage(predicate)
+ self.assertEqual([3], evaluation.result.results().to_list())
+ self.assertEqual((Range(0, 3),), evaluation.indexed_ranges)
+ self.assertEqual([Range(4, 6)], scanner.unindexed_ranges(
+ predicate, indexed_ranges=evaluation.indexed_ranges,
+ search_mode=table.copy({
+ 'scalar-index.search-mode': 'full'
+ }).options.scalar_index_search_mode()))
+
+ entries = [IndexManifestEntry(0, GenericRow([], []), 0, composite)] +
scalar_entries
+ with patch.object(IndexFileHandler, 'scan', return_value=entries):
+ actual = self._rows(table.copy({'scalar-index.search-mode':
'full'}), predicate)
+ expected = self._rows(table.copy({'global-index.enabled': 'false'}),
predicate)
+ self.assertEqual(expected, actual)
+
+ @pytest.mark.python_plan
+ def test_composite_key_residual_keeps_fast_coverage(self):
+ table = self._create_table()
+ self._write_arrow(table, pa.table({
+ 'a': [None, None, 1, 1], 'b': [None, 2, None, 1],
+ }, schema=self.pa_schema))
+ table.create_global_index('b')
+ self._write_arrow(table, pa.table({
+ 'a': [1, 1, 2], 'b': [2, 2, 1],
+ }, schema=self.pa_schema))
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ scalar_entries = IndexFileHandler(table).scan(snapshot)
+ composite = _java_composite_file()
+ entries = [IndexManifestEntry(0, GenericRow([], []), 0, composite)] +
scalar_entries
+ builder = table.new_read_builder().new_predicate_builder()
+ predicate = PredicateBuilder.and_predicates([
+ builder.equal('a', 1), builder.not_equal('b', 1),
+ ])
+ expected = self._rows(table.copy({'global-index.enabled': 'false'}),
predicate)
+ self.assertEqual([{'a': 1, 'b': 2}] * 2, expected)
+
+ with DataEvolutionGlobalIndexScanner.create(
+ table, index_files=[composite] + [entry.index_file for entry
in scalar_entries],
+ snapshot=snapshot, allow_composite=True) as scanner:
+ evaluation = scanner.scan_with_coverage(predicate)
+ self.assertEqual([2, 3, 4, 5],
evaluation.result.results().to_list())
+ self.assertEqual((Range(0, 6),), evaluation.indexed_ranges)
+
+ with patch.object(IndexFileHandler, 'scan', return_value=entries):
+ for mode in ('full', 'fast'):
+ with self.subTest(mode=mode):
+ actual = self._rows(
+ table.copy({'scalar-index.search-mode': mode}),
predicate)
+ self.assertEqual(expected, actual)
+
+ def test_composite_plan_requires_left_prefix_and_bounded_in(self):
+ fields = [
+ DataField(0, 'a', AtomicType('INT')),
+ DataField(1, 'b', AtomicType('INT')),
+ ]
+ builder = PredicateBuilder(fields)
+ self.assertIsNone(plan_composite(fields, builder.equal('b', 1)))
+ self.assertIsNone(plan_composite(
+ fields, builder.is_in('a', list(range(257)))))
+ self.assertEqual(256, len(plan_composite(
+ fields, builder.is_in('a', list(range(256)))).intervals))
+ self.assertEqual(1, len(plan_composite(
+ fields, builder.is_in('a', [1, None])).intervals))
+ self.assertIsNone(plan_composite(fields, builder.less_than('a', 1.5)))
+ self.assertIsNone(plan_composite(fields, builder.equal('a', True)))
+ unsupported = [DataField(0, 'a', AtomicType('BINARY')), fields[1]]
+ self.assertIsNone(plan_composite(unsupported, builder.equal('a',
b'x')))
+
+ def test_java_compacted_key_bytes_preserve_precision(self):
+ fields = [
+ DataField(0, 's', AtomicType('STRING')),
+ DataField(1, 'd', AtomicType('DECIMAL(38,18)')),
+ DataField(2, 't', AtomicType('TIMESTAMP(9)')),
+ ]
+ serializer = CompositeKeySerializer(fields)
+ expected = [
+ (None, None, None),
+ ('a', Decimal('1.230000000000000000'), Timestamp(0, 1000)),
+ ('x' * 130, Decimal('-99999999999999999999.999999999999999999'),
+ Timestamp(-1, 999999)),
+ ]
+ path = Path(__file__).parent /
'resources/btree/java-v2-composite-keys.txt'
+ encoded_keys = [line for line in path.read_text().splitlines()
+ if line and not line.startswith('#')]
+ self.assertEqual(len(expected), len(encoded_keys))
+ for row, encoded in zip(expected, encoded_keys):
+ with self.subTest(row=row):
+ self.assertEqual(row,
serializer.deserialize(bytes.fromhex(encoded)))
+ self.assertEqual(encoded, serializer.serialize(row).hex())
+
+ def test_float_keys_keep_java_order_and_zero_predicates_fall_back(self):
+ fields = [
+ DataField(0, 'a', AtomicType('FLOAT')),
+ DataField(1, 'b', AtomicType('INT')),
+ ]
+ serializer = CompositeKeySerializer(fields)
+ self.assertLess(serializer.create_comparator()((-0.0, 1), (0.0, 1)), 0)
+ builder = PredicateBuilder(fields)
+ self.assertIsNone(plan_composite(fields, builder.equal('a', 0.0)))
+ self.assertIsNone(plan_composite(fields, builder.greater_or_equal('a',
-0.0)))
+
+ def test_time_precision_does_not_break_prefix_reads(self):
+ fields = [
+ DataField(0, 'a', AtomicType('INT')),
+ DataField(1, 't', AtomicType('TIME(3)')),
+ ]
+ serializer = CompositeKeySerializer(fields)
+ self.assertEqual((1, 123), serializer.deserialize(
+ serializer.serialize((1, time(0, 0, 0, 123000)))))
+ builder = PredicateBuilder(fields)
+ self.assertIsNotNone(plan_composite(fields, builder.equal('a', 1)))
+ self.assertIsNotNone(plan_composite(fields,
PredicateBuilder.and_predicates([
+ builder.equal('a', 1), builder.equal('t', time(0, 0, 0,
123000))])))
+ self.assertIsNone(plan_composite(fields,
PredicateBuilder.and_predicates([
+ builder.equal('a', 1), builder.less_than('t', time(0, 0, 0,
123456))])))
+
+ @pytest.mark.python_plan
+ def test_time_submillisecond_bound_does_not_prune_matching_row(self):
+ schema = pa.schema([('a', pa.int32()), ('t', pa.time32('ms'))])
+ table = self._create_table(pa_schema=schema)
+ self._write_arrow(table, pa.table({
+ 'a': [1], 't': [time(0, 0, 0, 123000)],
+ }, schema=schema))
+ index_path =
table.path_factory().global_index_path_factory().index_path()
+ writer = BTreeIndexWriter(
+ table.file_io, index_path, CompositeKeySerializer(table.fields))
+ writer.write((1, 123), 0)
+ written = writer.finish()[0]
+ index_file = IndexFileMeta(
+ index_type='btree', file_name=written.file_name,
+ file_size=table.file_io.get_file_size(index_path + '/' +
written.file_name),
+ row_count=1,
+ global_index_meta=GlobalIndexMeta(0, 0, 0, [1], written.meta),
+ )
+ entry = IndexManifestEntry(0, GenericRow([], []), 0, index_file)
+ builder = table.new_read_builder().new_predicate_builder()
+ predicate = PredicateBuilder.and_predicates([
+ builder.equal('a', 1), builder.less_than('t', time(0, 0, 0,
123456)),
+ ])
+ expected = self._rows(table.copy({'global-index.enabled': 'false'}),
predicate)
+ self.assertEqual([{'a': 1, 't': time(0, 0, 0, 123000)}], expected)
+ with patch.object(IndexFileHandler, 'scan', return_value=[entry]):
+ actual = self._rows(table.copy({'scalar-index.search-mode':
'full'}), predicate)
+ self.assertEqual(expected, actual)
+
+ @pytest.mark.python_plan
+ def test_corrupt_selected_composite_metadata_is_not_hidden(self):
+ table = self._create_table()
+ self._write_arrow(table, pa.table({
+ 'a': [1], 'b': [2],
+ }, schema=self.pa_schema))
+ file = _java_composite_file()
+ file = replace(file, global_index_meta=replace(
+ file.global_index_meta, index_meta=b'broken'))
+ entry = IndexManifestEntry(0, GenericRow([], []), 0, file)
+ predicate = PredicateBuilder.and_predicates([
+ table.new_read_builder().new_predicate_builder().equal('a', 1),
+ table.new_read_builder().new_predicate_builder().equal('b', 2),
+ ])
+
+ with patch.object(IndexFileHandler, 'scan', return_value=[entry]):
+ with self.assertRaises(CompositeIndexReadError):
+ self._rows(table, predicate)
+
+ def test_unused_composite_preserves_scalar_read_parallelism(self):
+ fields = [DataField(i, name, AtomicType('INT'))
+ for i, name in enumerate(('a', 'b', 'x', 'y'))]
+ builder = PredicateBuilder(fields)
+ predicate = PredicateBuilder.and_predicates([
+ builder.equal('b', 1), builder.equal('x', 1), builder.equal('y',
1),
+ ])
+ file_io = LocalFileIO()
+ with tempfile.TemporaryDirectory() as index_path:
+ scalar_files = []
+ for field in fields[1:]:
+ writer = BTreeIndexWriter(
+ file_io, index_path, create_serializer(field.type))
+ writer.write(1, 0)
+ written = writer.finish()[0]
+ scalar_files.append(IndexFileMeta(
+ index_type='btree', file_name=written.file_name,
+ file_size=file_io.get_file_size(index_path + '/' +
written.file_name),
+ row_count=7,
+ global_index_meta=GlobalIndexMeta(
+ 0, 6, field.id, None, written.meta),
+ ))
+
+ gate = threading.Barrier(3, timeout=3)
+ original = BTreeIndexReader.__init__
+
+ def concurrent_init(reader, *args, **kwargs):
+ gate.wait()
+ original(reader, *args, **kwargs)
+
+ with patch.object(BTreeIndexReader, '__init__', concurrent_init):
+ with DataEvolutionGlobalIndexScanner(
+ fields, file_io, index_path, scalar_files,
+ thread_num=3,
composite_files=[_java_composite_file()]) as scanner:
+ evaluation = scanner.scan_with_coverage(predicate)
+ self.assertEqual([0], evaluation.result.results().to_list())
+ self.assertEqual(frozenset((1, 2, 3)),
evaluation.contributing_field_ids)
+
+ def test_distinct_composite_files_initialize_concurrently(self):
+ fields = [DataField(i, name, AtomicType('INT'))
+ for i, name in enumerate(('a', 'b'))]
+ builder = PredicateBuilder(fields)
+ predicate = PredicateBuilder.and_predicates([
+ builder.equal('a', 1), builder.equal('b', 2),
+ ])
+ source = _java_composite_file()
+ file_io = LocalFileIO()
+ with tempfile.TemporaryDirectory() as index_path:
+ files = []
+ for i in range(8):
+ path = str(Path(index_path) / ('shard-%d.btree' % i))
+ shutil.copyfile(source.external_path, path)
+ files.append(replace(
+ source, file_name=Path(path).name, external_path=path,
+ global_index_meta=replace(
+ source.global_index_meta,
+ row_range_start=i * 7, row_range_end=i * 7 + 6)))
+ files.append(replace(
+ files[0], global_index_meta=replace(
+ files[0].global_index_meta,
+ row_range_start=56, row_range_end=62)))
+
+ gate = threading.Barrier(8, timeout=3)
+ opened = []
+ lock = threading.Lock()
+ original = CompositeBTreeIndexReader.__init__
+
+ def concurrent_init(reader, *args, **kwargs):
+ with lock:
+ opened.append(args[-1].external_path)
+ gate.wait()
+ original(reader, *args, **kwargs)
+
+ with patch.object(CompositeBTreeIndexReader, '__init__',
concurrent_init):
+ with DataEvolutionGlobalIndexScanner(
+ fields, file_io, index_path, [], thread_num=8,
+ composite_files=files) as scanner:
+ evaluation = scanner.scan_with_coverage(predicate)
+ self.assertEqual(8, len(opened))
+ self.assertEqual(8, len(set(opened)))
+ self.assertEqual(sorted([offset + row for offset in range(0, 63, 7)
+ for row in (4, 5)]),
+ evaluation.result.results().to_list())
+
+ @staticmethod
+ def _rows(table, predicate):
+ builder = table.new_read_builder().with_filter(predicate)
+ return
builder.new_read().to_arrow(builder.new_scan().plan().splits()).to_pylist()
diff --git a/paimon-python/pypaimon/tests/global_index_test.py
b/paimon-python/pypaimon/tests/global_index_test.py
index ec889da3bf..f1532a291e 100644
--- a/paimon-python/pypaimon/tests/global_index_test.py
+++ b/paimon-python/pypaimon/tests/global_index_test.py
@@ -201,7 +201,7 @@ class
DataEvolutionGlobalIndexCoverageTest(unittest.TestCase):
coverage.unindexed_ranges(table.fields, predicate),
)
- def test_extra_fields_count_as_index_coverage(self):
+ def test_composite_fields_do_not_count_as_scalar_coverage(self):
from
pypaimon.execution.python.globalindex.data_evolution_global_index_coverage
import (
DataEvolutionGlobalIndexCoverage,
)
@@ -214,7 +214,7 @@ class
DataEvolutionGlobalIndexCoverageTest(unittest.TestCase):
[_coverage_index_file(0, 0, 9, extra_field_ids=[1])],
)
- self.assertEqual([], coverage.unindexed_ranges(1))
+ self.assertEqual([Range(0, 9)], coverage.unindexed_ranges(1))
def test_detail_mode_uses_table_data_ranges(self):
from
pypaimon.execution.python.globalindex.data_evolution_global_index_coverage
import (
diff --git
a/paimon-python/pypaimon/tests/resources/btree/BTreeFixtureGenerator.java
b/paimon-python/pypaimon/tests/resources/btree/BTreeFixtureGenerator.java
index 04a101491b..ae0c087b1e 100644
--- a/paimon-python/pypaimon/tests/resources/btree/BTreeFixtureGenerator.java
+++ b/paimon-python/pypaimon/tests/resources/btree/BTreeFixtureGenerator.java
@@ -19,17 +19,24 @@
package org.apache.paimon.globalindex.btree;
import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.Timestamp;
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.globalindex.KeySerializer;
+import org.apache.paimon.globalindex.CompositeKeySerializer;
import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
import org.apache.paimon.memory.MemorySlice;
import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
import org.apache.paimon.types.VarCharType;
import org.apache.paimon.utils.BloomFilter;
import org.apache.paimon.utils.LongArrayList;
import java.io.ByteArrayOutputStream;
+import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
@@ -157,5 +164,119 @@ public final class BTreeFixtureGenerator {
}
}
}
+ writeComposite(directory, "java-v2-composite-int.btree", false);
+ writeComposite(directory, "java-v2-composite-ac.btree", false);
+ writeComposite(directory, "java-v2-composite-reversed.btree", true);
+ writeCompositeKeys(directory);
+ }
+
+ private static void writeComposite(Path directory, String name, boolean
reversed)
+ throws Exception {
+ final ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ GlobalIndexFileWriter files =
+ new GlobalIndexFileWriter() {
+ @Override
+ public String newFileName(String prefix) {
+ return name;
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(String
fileName) {
+ return new PositionOutputStream() {
+ @Override
+ public long getPos() {
+ return bytes.size();
+ }
+
+ @Override
+ public void write(int value) {
+ bytes.write(value);
+ }
+
+ @Override
+ public void write(byte[] value) {
+ bytes.write(value, 0, value.length);
+ }
+
+ @Override
+ public void write(byte[] value, int offset, int
length) {
+ bytes.write(value, offset, length);
+ }
+
+ @Override
+ public void flush() {}
+
+ @Override
+ public void close() {}
+ };
+ }
+ };
+ BTreeIndexWriter writer =
+ new BTreeIndexWriter(
+ files,
+ new CompositeKeySerializer(RowType.of(DataTypes.INT(),
DataTypes.INT())),
+ 32,
+ null,
+ null,
+ 2);
+ Integer[][] keys = {
+ {null, null}, {null, 2}, {1, null}, {1, 1}, {1, 2}, {1, 2}, {2, 1}
+ };
+ int[] order = reversed ? new int[] {0, 2, 3, 6, 1, 4, 5} : new int[]
{0, 1, 2, 3, 4, 5, 6};
+ for (int rowId : order) {
+ Integer[] key = keys[rowId];
+ writer.write(GenericRow.of(
+ reversed ? key[1] : key[0], reversed ? key[0] : key[1]),
rowId);
+ }
+ ResultEntry entry = writer.finish().get(0);
+ Files.write(directory.resolve(name), bytes.toByteArray());
+ Files.write(directory.resolve(name + ".meta"), entry.meta());
+ }
+
+ private static void writeCompositeKeys(Path directory) throws Exception {
+ KeySerializer serializer =
+ new CompositeKeySerializer(
+ RowType.of(
+ DataTypes.STRING(), DataTypes.DECIMAL(38, 18),
+ DataTypes.TIMESTAMP(9)));
+ GenericRow[] rows = {
+ GenericRow.of(null, null, null),
+ GenericRow.of(
+ BinaryString.fromString("a"),
+ Decimal.fromBigDecimal(new
BigDecimal("1.230000000000000000"), 38, 18),
+ Timestamp.fromEpochMillis(0, 1000)),
+ GenericRow.of(
+ BinaryString.fromString(new String(new
char[130]).replace('\0', 'x')),
+ Decimal.fromBigDecimal(
+ new
BigDecimal("-99999999999999999999.999999999999999999"),
+ 38, 18),
+ Timestamp.fromEpochMillis(-1, 999999)),
+ };
+ StringBuilder result =
+ new StringBuilder(
+ "# Licensed to the Apache Software Foundation (ASF)
under one\n"
+ + "# or more contributor license agreements.
See the NOTICE file\n"
+ + "# distributed with this work for additional
information\n"
+ + "# regarding copyright ownership. The ASF
licenses this file\n"
+ + "# to you under the Apache License, Version
2.0 (the\n"
+ + "# \"License\"); you may not use this file
except in compliance\n"
+ + "# with the License. You may obtain a copy
of the License at\n"
+ + "#\n"
+ + "#
http://www.apache.org/licenses/LICENSE-2.0\n";
+ + "#\n"
+ + "# Unless required by applicable law or
agreed to in writing, software\n"
+ + "# distributed under the License is
distributed on an \"AS IS\" BASIS,\n"
+ + "# WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied.\n"
+ + "# See the License for the specific language
governing permissions and\n"
+ + "# limitations under the License.\n");
+ for (GenericRow row : rows) {
+ for (byte value : serializer.serialize(row)) {
+ result.append(String.format("%02x", value & 0xff));
+ }
+ result.append('\n');
+ }
+ Files.write(
+ directory.resolve("java-v2-composite-keys.txt"),
+ result.toString().getBytes(StandardCharsets.UTF_8));
}
}
diff --git a/paimon-python/pypaimon/tests/resources/btree/README.md
b/paimon-python/pypaimon/tests/resources/btree/README.md
index bf708a7ffc..6da0d7b726 100644
--- a/paimon-python/pypaimon/tests/resources/btree/README.md
+++ b/paimon-python/pypaimon/tests/resources/btree/README.md
@@ -79,3 +79,8 @@ with tempfile.TemporaryDirectory() as classes:
The fixtures were verified with Python 3.13 and pyroaring 1.1.0. These samples
cover 64-bit portable Roaring payloads, including multiple high-word values and
`Long.MAX_VALUE`. They do not establish compatibility with every pyroaring
version.
+
+The composite fixtures were generated by `BTreeFixtureGenerator.java` from
+Java source at `e2cdaf54ffbfe84a86df7d0ca3fb1da2e2d0d8ba`. Python tests
+attach field ID tuples for `(a,b)`, `(a,c)`, and `(b,a)`; the key bytes contain
+ordered values, not field IDs. `java-v2-composite-keys.txt` checks mixed-type
keys.
diff --git
a/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-ac.btree
b/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-ac.btree
new file mode 100644
index 0000000000..56d27480c0
Binary files /dev/null and
b/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-ac.btree differ
diff --git
a/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-ac.btree.meta
b/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-ac.btree.meta
new file mode 100644
index 0000000000..ca7d97c30a
Binary files /dev/null and
b/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-ac.btree.meta
differ
diff --git
a/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-int.btree
b/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-int.btree
new file mode 100644
index 0000000000..56d27480c0
Binary files /dev/null and
b/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-int.btree
differ
diff --git
a/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-int.btree.meta
b/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-int.btree.meta
new file mode 100644
index 0000000000..ca7d97c30a
Binary files /dev/null and
b/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-int.btree.meta
differ
diff --git
a/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-keys.txt
b/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-keys.txt
new file mode 100644
index 0000000000..f68d9d73c5
--- /dev/null
+++ b/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-keys.txt
@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# 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.
+0007
+00000161081111d67bb1bb00000000000000000000e807
+000082017878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787878787810b4c4b357a5793b85f675ddc000000001ffffffffffffffffbf843d
diff --git
a/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-reversed.btree
b/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-reversed.btree
new file mode 100644
index 0000000000..eb6e606cd2
Binary files /dev/null and
b/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-reversed.btree
differ
diff --git
a/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-reversed.btree.meta
b/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-reversed.btree.meta
new file mode 100644
index 0000000000..ca7d97c30a
Binary files /dev/null and
b/paimon-python/pypaimon/tests/resources/btree/java-v2-composite-reversed.btree.meta
differ
