MaxStorage

MaxCompute MaxStorage is a high-throughput data read/write interface provided by MaxCompute. Compared to the Tunnel-based data channel, MaxStorage offers finer-grained session management, supports Arrow and Blob format read/write, incremental reads, and table preview, making it suitable for large-scale parallel data read/write scenarios.

Note

MaxStorage requires server-side support. Make sure the relevant features are enabled on your MaxCompute cluster.

Core concepts

The core concepts of MaxStorage include:

• Session: The transactional context for read/write operations. A read session manages data splits; a write session guarantees data atomicity.

• Split: A read session divides data into multiple splits by size, parallelism, etc. Each split can be read independently, enabling parallel processing.

• Stream: A data upload channel within a write session. A write session can create multiple streams to support parallel writes.

• Compression: The read path supports UNCOMPRESSED (default), LZ4, and ZSTD; the write path uses Arrow IPC built-in compression, reducing network transfer size.

• Route Token: A routing identifier returned by the server, used for session affinity to ensure subsequent requests are routed to the same node.

• Exactly-Once Mode: Write streams support exactly-once semantics, achieving idempotent writes via access_token and row_offset.

• API Version: The api_version parameter (default "2") selects the URL path api/storage/v2 or api/storage/v3; v3-era features are gated behind _supports_v3().

Client initialization

To use MaxStorage, create a MaxStorageClient instance, passing in the ODPS entry object.

from odps import ODPS
from odps.maxstorage import MaxStorageClient

# Initialize the ODPS entry object
odps = ODPS(
    access_id="your_access_id",
    secret_access_key="your_secret_access_key",
    project="your_project",
    endpoint="your_endpoint",
)

# Create a MaxStorage client; default api_version="2"
client = MaxStorageClient(odps)

# Use API v3 (enables WriteMode, nested Blob, and other advanced features)
client_v3 = MaxStorageClient(odps, api_version="3")

You can also manually specify the tunnel endpoint via the tunnel_endpoint parameter, bypassing auto-discovery:

client = MaxStorageClient(odps, tunnel_endpoint="http://your-tunnel-endpoint")

You can also specify a resource quota via the quota_name parameter:

client = MaxStorageClient(odps, quota_name="your_quota")

Reading data

Complete read flow

Reading data with MaxStorage requires the following steps:

1. Create a read session

2. Read data split by split

3. Close the read session (sessions expire automatically; manual close is optional)

Create a read session

Create a read session via create_table_read_session(). The session determines the split strategy, returned columns, and partitions. After creation it polls until the session status becomes NORMAL.

from odps.maxstorage import MaxStorageClient, SplitOptions, SplitMode

client = MaxStorageClient(odps)

# Create a read session with default split options (split by size)
read_session = client.create_table_read_session("your_table")
print(f"Session ID: {read_session.id}")
print(f"Number of splits: {len(read_session.splits)}")
print(f"Total record count: {read_session.record_count}")
print(f"Expiration time: {read_session.expiration_time}")

When creating a read session you can specify columns, partitions, and split options:

# Read only specific columns and partitions
read_session = client.create_table_read_session(
    "your_table",
    columns=["id", "name", "value"],
    partitions=["pt=20230101"],
)

# Split by row offset, 1 million rows per split
split_opts = SplitOptions(
    split_mode=SplitMode.ROW_OFFSET,
    split_number=1000000,
)
read_session = client.create_table_read_session(
    "your_table",
    split_options=split_opts,
)

Note

The split type returned by read_session.splits depends on the SplitMode: under SplitMode.SIZE it is IndexedInputSplit (addressed by index), and under SplitMode.ROW_OFFSET it is RowRangeInputSplit (addressed by row-offset range). Both can be passed directly to open_arrow_reader().

Note

For append2.0 / transactional tables, the server only supports the SplitMode.ROW_OFFSET split mode.

Reading data in Arrow format

Use open_arrow_reader() to read directly as Arrow RecordBatches:

import pyarrow as pa

read_session = client.create_table_read_session("your_table")

# Iterate over all splits to read data
for split in read_session.splits:
    reader = read_session.open_arrow_reader(split)
    while True:
        batch = reader.read()
        if batch is None:
            break
        df = batch.to_pandas()
        # Process the DataFrame
    reader.close()

You can control the rows per batch with max_batch_rows and skip rows with skip_row_num:

reader = read_session.open_arrow_reader(
    read_session.splits[0],
    max_batch_rows=1024,
    skip_row_num=100,
)

You can also use get_as_record_reader() to get a row-oriented iterator of Record. The returned ArrowRecordReader exposes the following attributes in addition to iteration:

• schema: returns an OdpsSchema with column names and types.

• count: returns the number of records read so far; after iteration completes it equals the total row count of the split.

reader = read_session.open_arrow_reader(read_session.splits[0])
rr = reader.get_as_record_reader()
print(rr.schema.columns)  # column info
for record in rr:
    print(record[0], record[1])
print(rr.count)           # number of records read

Parallel reads

MaxStorage’s split mechanism natively supports parallel reads. Each split can be read independently, making it suitable for multi-threaded scenarios:

from concurrent.futures import ThreadPoolExecutor

read_session = client.create_table_read_session("your_table")

def read_split(split):
    reader = read_session.open_arrow_reader(split)
    batches = []
    while True:
        batch = reader.read()
        if batch is None:
            break
        batches.append(batch)
    reader.close()
    return pa.concat_batches(batches) if batches else None

# Read all splits in parallel using a thread pool
with ThreadPoolExecutor(max_workers=len(read_session.splits)) as pool:
    futures = [pool.submit(read_split, s) for s in read_session.splits]
    results = [f.result() for f in futures]

Incremental reads

Incremental reads retrieve only data added or changed since a given version:

from odps.maxstorage import IncrementalReadOptions

incr_opts = IncrementalReadOptions(
    version="v1",
    from_=100,
    to=200,
)
read_session = client.create_table_read_session(
    "your_table",
    incremental_read_enabled=True,
    incremental_read_options=incr_opts,
)

Compressed reads

Read-path compression is enabled via the compress_option parameter; the default is uncompressed. ZSTD and LZ4_FRAME are supported.

Note

Install the corresponding library before using compression:

• ZSTD: pip install zstandard

• LZ4: pip install lz4

from odps.tunnel import CompressOption

compress_option = CompressOption(
    CompressOption.CompressAlgorithm.ODPS_ZSTD,
)
reader = read_session.open_arrow_reader(
    read_session.splits[0],
    compress_option=compress_option,
)

You can also use the compress_algo shorthand:

reader = read_session.open_arrow_reader(
    read_session.splits[0],
    compress_algo="zstd",
)

Reading instance results

Use create_instance_read_session() to read the results of a SQL instance:

instance = odps.execute_sql("SELECT * FROM your_table LIMIT 100")
instance_session = client.create_instance_read_session(instance)

reader = instance_session.open_arrow_reader(offset=0, count=100)
while True:
    batch = reader.read()
    if batch is None:
        break
    print(batch.to_pandas())
reader.close()

Table preview

Use preview_table() to quickly preview the first rows of a table without creating a read session:

reader = client.preview_table("your_table", limit=10)
while True:
    batch = reader.read()
    if batch is None:
        break
    print(batch.to_pandas())
reader.close()

Writing data

Complete write flow

Writing data with MaxStorage requires the following steps:

1. Create a write session

2. Create a write stream

3. Write data (batch or record-oriented)

4. Commit the session

Create a write session

Create a write session via create_table_write_session():

from odps.maxstorage import MaxStorageClient, WriteMode

client = MaxStorageClient(odps)

# 创建批处理写会话（默认）
write_session = client.create_table_write_session("your_table")

# 创建流式写会话
write_session = client.create_table_write_session(
    "your_table",
    write_mode=WriteMode.STREAMING,
)

# 创建兼容模式写会话
write_session = client.create_table_write_session(
    "your_table",
    write_mode=WriteMode.BATCH_COMPATIBLE,
)

# 写入指定分区
write_session = client.create_table_write_session(
    "your_table",
    partition_spec="pt=20230101",
)

Batch-writing Arrow RecordBatches

Create a writer via open_arrow_writer(), then call write_batch to write:

import pyarrow as pa

write_session = client.create_table_write_session("your_table")
writer = write_session.open_arrow_writer(stream_id="0")

schema = pa.schema([
    ("id", pa.int64()),
    ("name", pa.string()),
])
batch = pa.RecordBatch.from_arrays(
    [pa.array([1, 2, 3], type=pa.int64()),
     pa.array(["a", "b", "c"], type=pa.string())],
    schema=schema,
)
writer.write_batch(batch)
writer.close()
write_session.commit()

Asynchronous flush

Use flush_async() to flush buffers asynchronously, with max_pending_buffers controlling backpressure:

writer = write_session.open_arrow_writer(stream_id="0", max_pending_buffers=4)
for batch in batches:
    writer.write_batch(batch)
    writer.flush_async()
writer.close()
write_session.commit()

Record-oriented writes

Use get_as_record_writer() to write via a row-oriented Record interface without manually constructing Arrow batches. This method works with any TableArrowWriter (plain tables and BLOB tables).

Plain table write (no BLOB columns):

from odps.models import Record

write_session = client.create_table_write_session("your_table")
writer = write_session.open_arrow_writer(stream_id="0")
record_writer = writer.get_as_record_writer()

record_writer.write(Record(columns=["id", "name"], values=[1, "alice"]))
record_writer.write(Record(columns=["id", "name"], values=[2, "bob"]))
record_writer.close()
write_session.commit()

Delta table write (with __operation column, supports UPSERT/DELETE):

from odps.models import Record

write_session = client.create_table_write_session("delta_table")
writer = write_session.open_arrow_writer(stream_id="0")
record_writer = writer.get_as_record_writer()  # returns DeltaTableRecordWriter

record_writer.write(Record(columns=["id", "name"], values=[1, "alice"]))  # UPSERT
record_writer.write(Record(columns=["id", "name"], values=[2, "bob"]))    # UPSERT
record_writer.delete(Record(columns=["id", "name"], values=[2, None]))    # DELETE
record_writer.close()
write_session.commit()

Note

get_as_record_writer() selects the writer type based on the schema: if the table has an __operation column, it returns DeltaTableRecordWriter, whose write() method stamps UPSERT and delete() method stamps DELETE — both operations can be freely interleaved on a single writer instance. Otherwise it returns AppendTableRecordWriter. For Record API writes on BLOB tables, see Auto-upload.

Compressed writes

The write path uses Arrow IPC built-in compression, enabled via the compress_option parameter:

from odps.tunnel import CompressOption

compress_option = CompressOption(
    CompressOption.CompressAlgorithm.ODPS_ZSTD,
)
writer = write_session.open_arrow_writer(
    stream_id="0",
    compress_option=compress_option,
)

Commit and abort

After writing, call commit() to finalize the data. To discard the data, call abort():

# Commit
write_session.commit()

# Abort (discard all uploaded data)
write_session.abort()

Note

close() does not auto-commit: if neither commit nor abort was called explicitly, close() will abort the session.

Batch-compatible write (BatchCompatible)

WriteMode.BATCH_COMPATIBLE 使用 block_number + attempt_number 标识 Writer， 并以强类型 BlockWriteResult 汇总提交，适合迁移依赖 Block 写入语义的任务。

兼容模式下不使用 open_arrow_writer，而是通过 open_block_writer() 创建 TableBlockWriter：

from odps.maxstorage import MaxStorageClient, WriteMode, BatchCompatibleOptions

client = MaxStorageClient(odps)
write_session = client.create_table_write_session(
    "your_table",
    write_mode=WriteMode.BATCH_COMPATIBLE,
    batch_compatible_options=BatchCompatibleOptions(
        enhance_write_check=True,
        max_field_size=8 * 1024 * 1024,
        dynamic_partition_limit=-1,
    ),
)

results = []
for block in range(block_count):
    writer = write_session.open_block_writer(block, 0)
    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3], type=pa.int64())],
        schema=writer.schema,
    )
    writer.write_batch(batch)
    results.append(writer.commit())

write_session.commit(block_results=results)

Note

• block_number 和 attempt_number 均从 0 开始。重试同一份 Block 数据时保持 block_number 不变并递增 attempt_number，最终只提交成功 attempt 的结果。

• 每个 TableBlockWriter 实例对应一个 block_number + attempt_number 对， 会在内存中缓存该 Block 的所有批次，直到 commit() 或 close() 上传完整的 Arrow IPC Stream。大任务应拆为多个 Block。

• open_block_writer 时 SDK 会自动预留一次上传 quota，调用方无需配置或传递 token。若无法完成 quota 预留，open_block_writer 会直接失败。

• 兼容模式必须调用 commit(block_results=...) 提交，无参 commit() 会报错。

• BatchCompatibleOptions 的 max_field_size 设置后必须至少为 1024 字节， dynamic_partition_limit 使用 -1 表示服务端默认。

• 可通过 max_block_number 获取 Block 上限，合法范围为 [0, max_block_number)。

Cross-process read/write

The MaxStorage session ID is a server-assigned global identifier that can be passed between processes. This enables an architecture where one process creates the session and multiple processes read and write in parallel.

Cross-process read

The main process creates a read session and distributes the session_id and split information to worker processes. Each worker reloads the session via session_id and independently reads its assigned splits:

# ---- Main process: create read session, distribute splits ----
from odps.maxstorage import MaxStorageClient

client = MaxStorageClient(odps)
read_session = client.create_table_read_session("your_table")

session_id = read_session.id          # pass to worker processes
splits = read_session.splits          # split list, each independently readable

# Distribute session_id and splits to workers (e.g. via multiprocessing.Queue,
# Redis, files, etc.)

# ---- Worker process: reload session, read assigned splits ----
from odps.maxstorage import MaxStorageClient

client = MaxStorageClient(odps)

# Reload the existing session via session_id (does not create a new session)
read_session = client.create_table_read_session("your_table", session_id=session_id)

# Each worker reads only its assigned splits
for split in my_splits:
    reader = read_session.open_arrow_reader(split)
    while True:
        batch = reader.read()
        if batch is None:
            break
        process(batch)
    reader.close()

Cross-process write

The main process creates a write session and distributes the session_id to worker processes. Each worker reloads the session via session_id, creates a write stream with resume=True, and writes data independently. Once all workers finish, the main process commits the session:

# ---- Main process: create write session ----
from odps.maxstorage import MaxStorageClient, WriteMode

client = MaxStorageClient(odps)
write_session = client.create_table_write_session("your_table")
session_id = write_session.id          # pass to worker processes

# Wait for all worker processes to finish writing...

# Main process commits the session (reload via session_id, then commit)
write_session = client.create_table_write_session("your_table", session_id=session_id)
write_session.commit()

# ---- Worker process: reload session, resume writing ----
from odps.maxstorage import MaxStorageClient

client = MaxStorageClient(odps)
write_session = client.create_table_write_session("your_table", session_id=session_id)

# resume=True calls getWriteStream to retrieve the existing stream state
writer = write_session.open_arrow_writer(stream_id=my_stream_id, resume=True)

writer.write_batch(my_batch)
writer.close()   # closes the write stream (does not commit the session)

# Workers only close their streams; the main process commits globally

• Each worker process uses a different stream_id (“0”, “1”, “2”, …) to write to the same session in parallel.

  • A worker’s writer.close() only closes the write stream; it does not commit data. The main process must call write_session.commit() to make the data visible.

  • If a worker process exits abnormally, the main process can call write_session.abort() to discard all uploaded data.

  • When reloading a write session via session_id, the latest route_token is fetched automatically, ensuring requests are routed to the correct server node.

Exactly-Once mode

Enable exactly-once semantics with exactly_once_mode=True:

writer = write_session.open_arrow_writer(
    stream_id="0",
    exactly_once_mode=True,
)
# Each flush carries the current row_offset; the server returns a new ExactlyOnceRowOffset
writer.write_batch(batch)
writer.flush()

# On resume, use get_row_offset() to fetch the latest offset
writer = write_session.open_arrow_writer(
    stream_id="0",
    exactly_once_mode=True,
    resume=True,
)
offset = writer.get_row_offset()

Blob I/O

What is a Blob

A Blob (Binary Large Object) is a MaxCompute column type (BLOB) for storing binary data, suitable for images, audio, video, documents, and other unstructured data. In MaxStorage, a BLOB column’s value is represented at the Arrow level as bytes (i.e. pa.binary()).

What is a blob reference

Because a single blob can be large, MaxStorage does not inline raw binary content directly in the Arrow batch when writing to storage. Instead it uses a two-phase write:

1. Upload the blob data: use write_blob_batch / write_blob_stream to upload the raw binary data to the server’s blob storage area; the server returns a blob reference.

2. Write the reference: place the blob reference as bytes into the BLOB column’s slot in the Arrow batch, then submit it together with the regular data via write_batch. The server resolves the reference back to the actual binary content.

A blob reference is an opaque bytes (or str) value generated by the server at upload time; its internal encoding is not visible to the client. The client only needs to write it verbatim into the BLOB column — do not attempt to decode, modify, or concatenate references. On read, the BLOB column likewise returns blob references; you must call BlobManager.read_blobs with the references to retrieve the actual binary data.

In short: a BLOB column stores references, not raw binary content; the reference is the ``bytes`` you get back after upload, and you use ``read_blobs`` to exchange it for the original data.

MaxStorage does not provide a Blob wrapper class — blob references are plain bytes/str, which makes them integrate seamlessly with the Arrow / pandas ecosystem.

MaxStorage provides two BLOB write modes:

• Manual upload (see below): explicitly call write_blob_batch / write_blob_stream to upload, then place the returned references into BLOB columns yourself. Suitable for scenarios requiring precise control over upload timing or metadata.

• Auto-upload (see Auto-upload): create the writer with auto_upload_blobs=True and pass raw bytes / file-like objects directly into BLOB columns; the writer uploads and replaces them with references automatically. Suitable for scenarios where manual reference management is unnecessary.

Manual upload

In manual upload mode, you explicitly call write_blob_batch / write_blob_stream to upload BLOB data, then place the returned references into the BLOB columns of the Arrow batch yourself. This mode is suitable when you need precise control over upload timing, batch size, or metadata. No auto_upload_blobs is needed when creating the writer (the default plain writer accepts reference bytes in BLOB columns).

The data parameter of build_blob_write_item accepts bytes, bytearray, or any readable file-like object (implementing read/seek/tell). When passing a file-like object, the data is read streamingly without loading the entire Blob into memory:

from io import BytesIO

# bytes / bytearray — pass directly
item1 = writer.build_blob_write_item(b"image-bytes", column_name="img")
item2 = writer.build_blob_write_item(bytearray(b"more-bytes"), column_name="img")

# File-like object — supports streaming reads, avoids loading the full blob
item3 = writer.build_blob_write_item(open("large.jpg", "rb"), column_name="img")
item4 = writer.build_blob_write_item(BytesIO(b"in-memory-stream"), column_name="img")

import pyarrow as pa

write_session = client.create_table_write_session("your_table")
writer = write_session.open_arrow_writer(stream_id="0")

# 1. Batch-upload blobs and obtain references (resp.blob_references is already list[bytes])
items = [
    writer.build_blob_write_item(b"image1-bytes", column_name="img"),
    writer.build_blob_write_item(b"image2-bytes", column_name="img"),
]
resp = writer.write_blob_batch(items)
refs = resp.blob_references  # list[bytes]

# 2. Write the references into the BLOB column, submitted with the regular data
batch = pa.RecordBatch.from_arrays(
    [pa.array([0, 1], pa.int64()), pa.array(refs, pa.binary())],
    schema=pa.schema([("id", pa.int64()), ("img", pa.binary())]),
)
writer.write_batch(batch)

writer.close()
write_session.commit()

Streaming upload of a single blob:

import pyarrow as pa

# 1. Stream-upload a single blob and obtain its reference (resp.blob_reference is already bytes)
blob_writer = writer.write_blob_stream(column_name="img")
with open("image.jpg", "rb") as f:
    while True:
        chunk = f.read(65536)
        if not chunk:
            break
        blob_writer.write(chunk)
resp = blob_writer.finish()
ref = resp.blob_reference  # bytes

# 2. Write the reference into the BLOB column
batch = pa.RecordBatch.from_arrays(
    [pa.array([0], pa.int64()), pa.array([ref], pa.binary())],
    schema=pa.schema([("id", pa.int64()), ("img", pa.binary())]),
)
writer.write_batch(batch)

writer.close()
write_session.commit()

Auto-upload

In auto-upload mode, create the writer with auto_upload_blobs=True to get a TableArrowBlobUploadWriter. BLOB columns can then accept raw bytes / file-like objects directly; the writer batch-uploads each BLOB cell and replaces it with a reference during write_batch or rw.write(). This mode is suitable when manual reference management is unnecessary.

Arrow API write (top-level BLOB):

import pyarrow as pa

write_session = client.create_table_write_session("blob_table")
writer = write_session.open_arrow_writer(stream_id="0", auto_upload_blobs=True)

# Pass bytes directly in the BLOB column; the writer auto-uploads
batch = pa.RecordBatch.from_arrays(
    [pa.array([0, 1], pa.int64()), pa.array([b"img0", b"img1"], pa.binary())],
    schema=pa.schema([("id", pa.int64()), ("img", pa.binary())]),
)
writer.write_batch(batch)
writer.close()
write_session.commit()

Record API write (no manual Arrow batch construction):

from io import BytesIO
from odps.models import Record

write_session = client.create_table_write_session("blob_table")
writer = write_session.open_arrow_writer(stream_id="0", auto_upload_blobs=True)
rw = writer.get_as_record_writer()

# Pass bytes / BytesIO / file-like for the BLOB column; the writer auto batch-uploads
rw.write(Record(columns=["id", "img"], values=[1, b"raw-bytes"]))
rw.write(Record(columns=["id", "img"], values=[2, BytesIO(b"file-like")]))
rw.close()
write_session.commit()

When a file-like object is passed, the writer reads its entire content and immediately closes the file handle, so the caller does not need to close it manually.

Nested BLOB write (ARRAY<BLOB>, requires API v3):

from io import BytesIO

# Nested ARRAY<BLOB> table: a BIGINT, b ARRAY<BLOB>
client_v3 = MaxStorageClient(odps, api_version="3")
write_session = client_v3.create_table_write_session("nested_blob_table")
writer = write_session.open_arrow_writer(stream_id="0", auto_upload_blobs=True)
rw = writer.get_as_record_writer()

# Pass list[bytes] for column b; each element is auto-uploaded
rw.write([0, [b"blob_0_a", b"blob_0_b"]])
rw.write([1, [BytesIO(b"blob_1_a"), b"blob_1_b", BytesIO(b"blob_1_c")]])
rw.close()
write_session.commit()

Note

With auto_upload_blobs=True a TableArrowBlobUploadWriter is returned; otherwise a plain TableArrowWriter is returned (BLOB columns must already contain reference bytes). Both writer types support get_as_record_writer() and the manual-upload helper methods (build_blob_write_item / write_blob_stream / write_blob_batch). When passing file-like objects into BLOB columns, the writer does not close them by default; set auto_close_files=True when creating the writer to close them automatically. This applies to the Arrow API (write_batch), manual batch upload (write_blob_batch), and the Record API (get_as_record_writer).

Reading blobs

Create a BlobManager via open_blob_manager(), then call read_blobs / read_blob to download:

blob_manager = client.open_blob_manager("your_table")

# Read multiple blobs; returns a BlobDataIterator
iterator = blob_manager.read_blobs([ref1, ref2])
for record in iterator:
    print(len(record.data))  # record.data is bytes

# Streaming read (avoids buffering the entire blob)
stream_reader = blob_manager.read_blobs([ref1], stream=True)
while True:
    chunk = stream_reader.read(4096)
    if not chunk:
        break
    # Process the chunk

# Read a single blob; returns a file-like object
fp = blob_manager.read_blob(ref1)
data = fp.read()

When reading via the Record API, BLOB columns return references that must be downloaded via BlobManager:

# Top-level BLOB
read_session = client.create_table_read_session("blob_table")
reader = read_session.open_arrow_reader(read_session.splits[0])
blob_manager = client.open_blob_manager("blob_table")

for record in reader.get_as_record_reader():
    ref = record[1]              # bytes — blob reference
    data = next(blob_manager.read_blobs([ref])).data
    print(record[0], len(data))  # a, len(b)

# Nested ARRAY<BLOB>
read_session = client_v3.create_table_read_session("nested_blob_table")
reader = read_session.open_arrow_reader(read_session.splits[0])
blob_manager = client_v3.open_blob_manager("nested_blob_table")

for record in reader.get_as_record_reader():
    ref_list = record[1]         # list[bytes] — list of references
    blobs = [b.data for b in blob_manager.read_blobs(ref_list)]
    print(record[0], [len(b) for b in blobs])

Nested blobs

When a BLOB column is nested inside a complex type (ARRAY, STRUCT, MAP), use find_all_blob_column_ids() to resolve the server-assigned column ID for the nested column (as a dot-path such as b.element). This feature requires API version 3 or above (specify api_version="3" when creating the client).

The Python / Arrow representation of a BLOB column depends on the complex type that contains it. The table below lists common combinations:

Note

In a dot-path, .element denotes the ARRAY element level, .value the MAP value level, and .<field_name> the STRUCT field level. For multi-level nesting, segments are concatenated from outermost to innermost.

ARRAY<BLOB> example

Table schema: a BIGINT, b ARRAY<BLOB>. Each element of column b is an independent Blob.

Manual upload (batch-upload first, then write refs into Arrow):

client_v3 = MaxStorageClient(odps, api_version="3")
write_session = client_v3.create_table_write_session("nested_blob_table")
writer = write_session.open_arrow_writer(stream_id="0")

# Nested column name: column b of type array<blob> → "b.element"
# Batch-upload 2~3 blobs for two rows (resp.blob_references is already list[bytes])
items = [
    writer.build_blob_write_item(b"row0_blob0", column_name="b.element"),
    writer.build_blob_write_item(b"row0_blob1", column_name="b.element"),
    writer.build_blob_write_item(b"row1_blob0", column_name="b.element"),
]
resp = writer.write_blob_batch(items)
refs = resp.blob_references  # list[bytes]

# Organize refs into list[bytes] per row, write into Arrow list(binary)
row_refs = [refs[0:2], refs[2:3]]
batch = pa.RecordBatch.from_arrays(
    [pa.array([0, 1], pa.int64()), pa.array(row_refs, pa.list_(pa.binary()))],
    schema=pa.schema([("a", pa.int64()), ("b", pa.list_(pa.binary()))]),
)
writer.write_batch(batch)
writer.close()
write_session.commit()

On read, column b returns list[bytes] (each element is a reference); download them individually:

read_session = client_v3.create_table_read_session("nested_blob_table")
reader = read_session.open_arrow_reader(read_session.splits[0])
blob_manager = client_v3.open_blob_manager("nested_blob_table")

for batch in reader:
    for a_val, blob_ref_list in zip(batch.column(0).to_pylist(),
                                    batch.column(1).to_pylist()):
        # blob_ref_list is list[bytes] (a list of references)
        blobs = [b.data for b in blob_manager.read_blobs(blob_ref_list)]
        print(a_val, [len(b) for b in blobs])

STRUCT<f: BLOB> example

Table schema: id BIGINT, s STRUCT<f: BLOB>. Column s is a struct whose field f is a Blob.

items = [writer.build_blob_write_item(b"struct_blob_0", column_name="s.f")]
resp = writer.write_blob_batch(items)
ref = resp.blob_references[0]  # bytes

# Arrow row: {"f": <ref bytes>}
batch = pa.RecordBatch.from_arrays(
    [pa.array([0], pa.int64()),
     pa.array([{"f": ref}], pa.struct([("f", pa.binary())]))],
    schema=pa.schema([("id", pa.int64()),
                      ("s", pa.struct([("f", pa.binary())]))]),
)
writer.write_batch(batch)
writer.close()
write_session.commit()

On read, column s returns a dict whose "f" field is a reference:

for batch in reader:
    for id_val, s_val in zip(batch.column(0).to_pylist(),
                             batch.column(1).to_pylist()):
        ref = s_val["f"]
        data = next(blob_manager.read_blobs([ref])).data
        print(id_val, len(data))

Blob metadata

Each blob can carry two pieces of metadata: MIME type (mime_type) and custom file name (custom_file_name, API v3 only). Set them on BlobWriteItem at upload time; retrieve them from BlobRecord / BlobStreamReader at download time.

Full pipeline: upload blobs with metadata → write references → read references → download and verify metadata.

from odps.maxstorage import MaxStorageClient

client_v3 = MaxStorageClient(odps, api_version="3")

# --- Write: upload blobs with metadata ---
write_session = client_v3.create_table_write_session("blob_table")
writer = write_session.open_arrow_writer(stream_id="0")

items = [
    writer.build_blob_write_item(
        b"\x89PNG image data",
        column_name="b",
        mime_type="image/png",
        custom_file_name="photo.png",
    ),
    writer.build_blob_write_item(
        b'{"key": "value"}',
        column_name="b",
        mime_type="application/json",
        custom_file_name="config.json",
    ),
]
resp = writer.write_blob_batch(items)
refs = resp.blob_references  # list[bytes]

# Write references into an Arrow batch
batch = pa.RecordBatch.from_arrays(
    [pa.array([0, 1], pa.int64()), pa.array(refs, pa.binary())],
    schema=pa.schema([("a", pa.int64()), ("b", pa.binary())]),
)
writer.write_batch(batch)
writer.close()
write_session.commit()

# --- Read: download blobs and retrieve metadata ---
read_session = client_v3.create_table_read_session("blob_table")
reader = read_session.open_arrow_reader(read_session.splits[0])
blob_manager = client_v3.open_blob_manager("blob_table")

for batch in reader:
    for a_val, ref in zip(batch.column(0).to_pylist(),
                          batch.column(1).to_pylist()):
        # Option 1: BlobRecord (one-shot read, includes metadata)
        record = next(blob_manager.read_blobs([ref]))
        print(record.data)              # bytes — raw content
        print(record.mime_type)         # "image/png" / "application/json"
        print(record.custom_file_name)  # "photo.png" / "config.json"

        # Option 2: BlobStreamReader (streaming read, includes metadata)
        stream_reader = blob_manager.read_blobs([ref], stream=True)
        print(stream_reader.mime_type)        # metadata available before reading
        print(stream_reader.custom_file_name)
        while True:
            chunk = stream_reader.read(4096)
            if not chunk:
                break
            # process chunk
        stream_reader.next()  # advance to the next blob (if any)

Note

custom_file_name is only available on API v3 and above; on v2 clients the field is silently ignored at upload time, and BlobRecord.custom_file_name is always None at download time. mime_type works on both v2 and v3.

Per-blob metadata callback

The “manual upload” above requires manually calling write_blob_batch and setting metadata on each BlobWriteItem. In auto-upload mode (putting raw BLOB data directly into an Arrow batch via write_batch, or writing via the Record API rw.write()), the writer auto-uploads each BLOB cell in batch. Use the blob_metadata_callback to assign mime_type and custom_file_name individually for every BLOB cell.

Signature: callback(row_index, column_name, blob_data) -> (mime_type, custom_file_name) | None

• row_index: Zero-based current row index.

• column_name: Dot-path of the BLOB column (top-level columns use the column name e.g. "b"; nested columns use "b.element", "s.f").

• blob_data: The original BLOB data. In the Record API path this is the original value the user passed (bytes / file-like object); the callback fires before the file is read into bytes. In the Arrow API path this is the deserialized bytes.

Note

• The callback is invoked exactly once per BLOB cell.

• None values and existing blob references (str) do not trigger the callback — they are never uploaded.

• When the callback returns None, it falls back to the session-level defaults (blob_mime_type / blob_custom_file_name passed to open_arrow_writer, which can also be used on their own without a callback); if those are also None, the blob carries no metadata.

• Stream uploads (write_blob_stream) use a wire protocol with no framing header, so the streaming path does not support metadata.

Example (inferring custom_file_name from the file name):

import os
from odps.models import Record

write_session = client.create_table_write_session("blob_table")

def metadata_fn(row_index, column_name, blob_data):
    # blob_data is the original file-like object passed by the user;
    # use its .name to derive a file name
    name = os.path.basename(getattr(blob_data, "name", f"blob_{row_index}"))
    return None, name

writer = write_session.open_arrow_writer(
    stream_id="0",
    auto_upload_blobs=True,
    blob_metadata_callback=metadata_fn,
)
rw = writer.get_as_record_writer()

rw.write(Record(columns=["a", "b"], values=[1, open("photo.jpg", "rb")]))
rw.write(Record(columns=["a", "b"], values=[2, open("doc.pdf", "rb")]))
rw.close()
write_session.commit()
# The two blobs' custom_file_name are "photo.jpg" and "doc.pdf"