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 8588fd3306 [python] Verify native REST Blob view dependency reads
(#10471)
8588fd3306 is described below
commit 8588fd3306310078915bd447ddd0c666390b2fb2
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Oct 9 17:54:48 2026 +0800
[python] Verify native REST Blob view dependency reads (#10471)
---
.../pypaimon/tests/native_plan_rest_test.py | 338 +++++++++++++++++++++
1 file changed, 338 insertions(+)
diff --git a/paimon-python/pypaimon/tests/native_plan_rest_test.py
b/paimon-python/pypaimon/tests/native_plan_rest_test.py
index ef3294c2f3..b0c62de7c6 100644
--- a/paimon-python/pypaimon/tests/native_plan_rest_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_rest_test.py
@@ -361,3 +361,341 @@ def
test_repeated_plans_reuse_remote_file_sizes(rest_source, tmp_path):
server.shutdown()
server.server_close()
thread.join()
+
+
[email protected]
+def rest_blob_views(rest_catalog):
+ catalog, server = rest_catalog
+ schema = pa.schema([('id', pa.int32()), ('payload', pa.large_binary())])
+ common = {'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true'}
+ for name in ['upstream', 'middle', 'root']:
+ extra = {} if name == 'upstream' else {'blob-view-field': 'payload'}
+ catalog.create_table('default.' + name, Schema.from_pyarrow_schema(
+ schema, options=dict(common, **extra)), False)
+
+ def write(table, rows):
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ finally:
+ writer.close()
+
+ upstream = catalog.get_table('default.upstream')
+ write(upstream, [{'id': 1, 'payload': b'actual'}, {'id': 2, 'payload':
None}])
+ middle = catalog.get_table('default.middle')
+ write(middle, [dict(id=i, payload=BlobViewStruct(
+ 'default.upstream', upstream.field_dict['payload'].id, i).serialize())
for i in [0, 1]])
+ root = catalog.get_table('default.root')
+ write(root, [dict(id=i, payload=BlobViewStruct(
+ middle.identifier, middle.field_dict['payload'].id, i).serialize())
for i in [0, 1]])
+ return catalog, server, upstream, middle, root
+
+
+def _require_blob_view_read_via(server, expected):
+ from pypaimon.api.rest_api import RESTApi
+ from pypaimon.api.rest_util import RESTUtil
+ from pypaimon.common.identifier import Identifier
+ from pypaimon.common.json_util import JSON
+
+ route = server._route_request
+ requests = []
+
+ def require_read_via(method, resource_path, parameters, data, headers):
+ if any('/tables/' + name in resource_path for name in ['upstream',
'middle']
+ if name != expected.get_table_name()):
+ encoded = next((value for key, value in headers.items()
+ if key.lower() ==
RESTApi.READ_VIA_HEADER.lower()), None)
+ actual = JSON.from_json(RESTUtil.decode_string(encoded),
Identifier) if encoded else None
+ requests.append((resource_path, actual))
+ if actual != expected:
+ return server._mock_response(ErrorResponse(
+ 'TABLE', 'upstream', 'dependency needs root Read-Via',
403), 403)
+ return route(method, resource_path, parameters, data, headers)
+
+ return patch.object(server, '_route_request',
side_effect=require_read_via), requests
+
+
[email protected]_plan
[email protected]_read
[email protected]('native', [False, True])
[email protected]('nested', [False, True])
+def test_rest_blob_view_dependency_read_via(rest_blob_views, native, nested):
+ from contextlib import ExitStack
+ from pypaimon.table.row.blob import BlobDescriptor
+
+ _, server, _, middle, root = rest_blob_views
+ root = root if nested else middle
+ root = root.copy({'scan.native-plan.enabled': str(native).lower(),
+ 'read.native.enabled': str(native).lower()})
+ guard, requests = _require_blob_view_read_via(server, root.identifier)
+ with guard, ExitStack() as stack:
+ if native:
+
stack.enter_context(patch('pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('native Blob
view read fell back')))
+ builder = root.new_read_builder()
+ rows =
builder.new_read().to_arrow(builder.new_scan().plan().splits()).sort_by('id').to_pylist()
+ assert rows == [dict(id=0, payload=b'actual'), dict(id=1,
payload=None)]
+ descriptors = root.copy({'blob-as-descriptor':
'true'}).new_read_builder()
+ values =
descriptors.new_read().to_arrow(descriptors.new_scan().plan().splits()).sort_by('id')
+ assert
BlobDescriptor.is_blob_descriptor(values.column('payload')[0].as_py())
+ assert values.column('payload')[1].as_py() is None
+ assert requests and all(identifier == root.identifier for _, identifier in
requests)
+
+
[email protected]_plan
[email protected]_read
[email protected]('native', [False, True])
+def test_rest_blob_view_merge_source_keeps_dependency_context(rest_blob_views,
native):
+ from contextlib import ExitStack
+ from pypaimon.table.data_evolution_merge_into import WhenNotMatched
+
+ catalog, server, _, _, root = rest_blob_views
+ schema = pa.schema([('id', pa.int32()), ('payload', pa.large_binary())])
+ catalog.create_table('default.target', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': str(native).lower(),
+ 'read.native.enabled': str(native).lower(),
+ 'scan.native-plan.enabled': str(native).lower(),
+ }), False)
+ target = catalog.get_table('default.target')
+ builder = target.new_batch_write_builder()
+ guard, requests = _require_blob_view_read_via(server, root.identifier)
+ with guard, ExitStack() as stack:
+ if native:
+
stack.enter_context(patch('pypaimon.table.data_evolution_merge_into._normalize_source',
+ side_effect=AssertionError('Python MERGE
source materialization')))
+ messages = builder.new_update().merge_into(root, on=['id'],
when_not_matched=[WhenNotMatched('*')])
+ builder.new_commit().commit(messages)
+ reader = target.new_read_builder()
+ assert
reader.new_read().to_arrow(reader.new_scan().plan().splits()).sort_by('id').to_pylist()
== [
+ dict(id=0, payload=b'actual'), dict(id=1, payload=None)]
+ assert requests and all(identifier == root.identifier for _, identifier in
requests)
+
+
[email protected]_plan
[email protected]_read
[email protected]('native', [False, True])
+def test_rest_blob_view_payload_uses_upstream_token(rest_blob_views, native):
+ import json
+ import time
+ from contextlib import ExitStack
+ from collections import Counter
+ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+ from pathlib import Path
+ from threading import Thread
+ from urllib.parse import urlparse
+
+ from pypaimon.api.api_response import GetTableTokenResponse
+ from pypaimon.catalog.rest.rest_token import RESTToken
+ from pypaimon.catalog.rest.rest_token_file_io import RESTTokenFileIO
+ from pypaimon.common.identifier import Identifier
+
+ catalog, rest_server, upstream, middle, _ = rest_blob_views
+ roots = {table.identifier.get_table_name():
Path(urlparse(table.table_path).path)
+ for table in [upstream, middle]}
+ requests = []
+
+ class ObjectStore(BaseHTTPRequestHandler):
+ def do_HEAD(self):
+ self.serve()
+
+ def do_GET(self):
+ self.serve()
+
+ def serve(self):
+ path = urlparse(self.path).path
+ parts = path.split('/', 3)
+ if len(parts) != 4 or parts[1] != 'bucket' or parts[2] not in
roots:
+ self.send_error(404)
+ return
+ name, relative = parts[2:]
+ auth = self.headers.get('Authorization', '')
+ requests.append((name, relative, auth))
+ if 'Credential=' + name + '-key/' not in auth:
+ self.send_error(403, 'wrong table token')
+ return
+ file = roots[name] / relative
+ if not file.is_file():
+ self.send_error(404)
+ return
+ data = file.read_bytes()
+ size = len(data)
+ byte_range = self.headers.get('Range')
+ self.send_response(206 if byte_range else 200)
+ if byte_range:
+ start, end = byte_range[6:].split('-')
+ start, end = int(start), int(end) if end else size - 1
+ data = data[start:end + 1]
+ self.send_header('Content-Range', 'bytes %s-%s/%s' % (start,
end, size))
+ self.send_header('Content-Length', str(len(data)))
+ self.send_header('ETag', '"immutable-fixture"')
+ self.end_headers()
+ if self.command == 'GET':
+ self.wfile.write(data)
+
+ def log_message(self, *args):
+ pass
+
+ store = ThreadingHTTPServer(('127.0.0.1', 0), ObjectStore)
+ thread = Thread(target=store.serve_forever, daemon=True)
+ thread.start()
+ table_handle = rest_server._table_handle
+ token_handle = rest_server._table_token_handle
+ token_requests = Counter()
+
+ def remote_table(method, data, identifier, response_identifier=None):
+ body, status = table_handle(method, data, identifier,
response_identifier)
+ if method == 'GET' and identifier.get_table_name() in roots:
+ response = json.loads(body)
+ response['path'] = 's3://bucket/' + identifier.get_table_name()
+ body = json.dumps(response)
+ return body, status
+
+ def token(method, identifier):
+ name = identifier.get_table_name()
+ token_requests[name] += 1
+ # Force an upstream refresh without sleeps; it must keep Read-Via.
+ if name == 'upstream' and token_requests[name] == 1:
+ return rest_server._mock_response(GetTableTokenResponse(
+ token={'s3.access-key': name + '-key', 's3.secret-key':
'test-secret'},
+ expires_at_millis=0), 200)
+ return token_handle(method, identifier)
+
+ try:
+ options = dict(catalog.context.options.to_map(), **{
+ 's3.endpoint': 'http://127.0.0.1:%s' % store.server_port,
+ 's3.region': 'us-east-1', 's3.path.style.access': 'true',
+ 'data-token.enabled': 'true'})
+ for name in roots:
+ rest_server.set_table_token(Identifier('default', name),
RESTToken({
+ 's3.access-key': name + '-key', 's3.secret-key':
'test-secret'},
+ int(time.time() * 1000) + 7_200_000))
+ with patch.object(rest_server, '_table_handle',
side_effect=remote_table), \
+ patch.object(rest_server, '_table_token_handle',
side_effect=token), \
+ patch.object(RESTTokenFileIO, '_TOKEN_CACHE', {}):
+ view =
CatalogFactory.create(options).get_table('default.middle').copy({
+ 'scan.native-plan.enabled': str(native).lower(),
+ 'read.native.enabled': str(native).lower()})
+ guard, dependency_requests =
_require_blob_view_read_via(rest_server, view.identifier)
+ with guard, ExitStack() as stack:
+ if native:
+
stack.enter_context(patch('pypaimon.read.table_read.TableRead._create_split_read',
+
side_effect=AssertionError('token view read fell back')))
+ builder = view.new_read_builder()
+ result =
builder.new_read().to_arrow(builder.new_scan().plan().splits()).sort_by('id')
+ assert result.to_pylist() == [dict(id=0, payload=b'actual'),
dict(id=1, payload=None)]
+ assert dependency_requests
+ assert token_requests['upstream'] >= 2
+ assert any(name == 'upstream' and path.endswith('.blob') for name,
path, _ in requests)
+ assert all('Credential=' + name + '-key/' in auth for name, _, auth in
requests)
+ finally:
+ store.shutdown()
+ thread.join()
+ store.server_close()
+
+
[email protected]_plan
[email protected]_read
+def test_native_rest_blob_views_group_upstream_fields_by_id(rest_catalog):
+ from pypaimon.schema.schema_change import SchemaChange
+
+ catalog, server = rest_catalog
+ source_schema = pa.schema([('id', pa.int32()), ('other', pa.int32()),
('picture', pa.large_binary()),
+ ('thumbnail', pa.large_binary())])
+ options = {'row-tracking.enabled': 'true', 'data-evolution.enabled':
'true'}
+ catalog.create_table('default.upstream',
Schema.from_pyarrow_schema(source_schema, options=options), False)
+ source = catalog.get_table('default.upstream')
+ writer = source.new_batch_write_builder().new_write()
+ try:
+ writer.write_arrow(pa.Table.from_pylist([
+ dict(id=1, other=10, picture=b'picture-0', thumbnail=b'thumb-0'),
+ dict(id=2, other=20, picture=b'picture-1', thumbnail=None),
+ ], schema=source_schema))
+
source.new_batch_write_builder().new_commit().commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ schema = pa.schema([('id', pa.int32()), ('left', pa.large_binary()),
('right', pa.large_binary())])
+ catalog.create_table('default.grouped', Schema.from_pyarrow_schema(
+ schema, options=dict(options, **{'blob-view-field': 'left,right'})),
False)
+ view = catalog.get_table('default.grouped').copy({
+ 'scan.native-plan.enabled': 'true', 'read.native.enabled': 'true'})
+ picture_id = source.field_dict['picture'].id
+ thumbnail_id = source.field_dict['thumbnail'].id
+ refs = [dict(id=i, left=BlobViewStruct('default.upstream', picture_id, 1 -
i).serialize(),
+ right=BlobViewStruct('default.upstream', thumbnail_id,
i).serialize()) for i in [0, 1]]
+ writer = view.new_batch_write_builder().new_write()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(refs, schema=schema))
+
view.new_batch_write_builder().new_commit().commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ # Drop an ordinary column so BLOB field IDs differ from their positions.
+ catalog.alter_table('default.upstream', [SchemaChange.drop_column('id')])
+ guard, requests = _require_blob_view_read_via(server, view.identifier)
+ with guard, patch('pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('multi-field view read fell
back')):
+ builder = view.new_read_builder()
+ rows =
builder.new_read().to_arrow(builder.new_scan().plan().splits()).sort_by('id').to_pylist()
+ assert rows == [dict(id=0, left=b'picture-1', right=b'thumb-0'),
+ dict(id=1, left=b'picture-0', right=None)]
+ metadata = [path for path, _ in requests if
path.endswith('/tables/upstream')]
+ snapshots = [path for path, _ in requests if
path.endswith('/tables/upstream/snapshot')]
+ assert len(metadata) == len(snapshots) == 1
+
+
[email protected]_plan
[email protected]_read
[email protected]('native', [False, True])
[email protected]('branch', [None, 'dev'])
[email protected]('explicit', [False, True])
+def
test_rest_blob_view_keeps_branch_and_explicit_outer_context(rest_blob_views,
native, branch, explicit):
+ from contextlib import ExitStack
+ from pypaimon.api.rest_api import RESTApi
+ from pypaimon.api.rest_util import RESTUtil
+ from pypaimon.common.identifier import Identifier
+ from pypaimon.common.json_util import JSON
+
+ catalog, server, _, _, root = rest_blob_views
+ if branch:
+ catalog.create_tag(root.identifier, 'base')
+ catalog.create_branch(root.identifier, branch, tag_name='base')
+ identifier = Identifier('default', 'root', branch=branch)
+ outer = Identifier('outer 空间', 'parent.with.dots')
+ header = RESTUtil.encode_string(JSON.to_json(outer, separators=(',', ':')))
+ options = dict(catalog.context.options.to_map())
+ if explicit:
+ options[RESTApi.HEADER_PREFIX + RESTApi.READ_VIA_HEADER] = header
+ view = CatalogFactory.create(options).get_table(identifier).copy({
+ 'scan.native-plan.enabled': str(native).lower(),
'read.native.enabled': str(native).lower()})
+ expected = outer if explicit else view.identifier
+ guard, requests = _require_blob_view_read_via(server, expected)
+ with guard, ExitStack() as stack:
+ if native:
+
stack.enter_context(patch('pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('branch view
read fell back')))
+ builder = view.new_read_builder()
+ result =
builder.new_read().to_arrow(builder.new_scan().plan().splits()).sort_by('id')
+ assert result.to_pylist() == [dict(id=0, payload=b'actual'), dict(id=1,
payload=None)]
+ assert requests and all(actual == expected for _, actual in requests)
+
+
[email protected]_plan
[email protected]_read
+def
test_native_rest_blob_view_does_not_bypass_denied_dependency(rest_blob_views):
+ _, server, _, _, root = rest_blob_views
+ root = root.copy({'scan.native-plan.enabled': 'true',
'read.native.enabled': 'true'})
+ route = server._route_request
+
+ def deny_upstream(method, path, parameters, data, headers):
+ if '/tables/upstream' in path:
+ return server._mock_response(ErrorResponse('TABLE', 'upstream',
'upstream denied', 403), 403)
+ return route(method, path, parameters, data, headers)
+
+ with patch.object(server, '_route_request', side_effect=deny_upstream), \
+ patch('pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('denied dependency fell back')):
+ builder = root.new_read_builder()
+ with pytest.raises(Exception, match='upstream denied'):
+ builder.new_read().to_arrow(builder.new_scan().plan().splits())
