diff --git a/checkpoint/orbax/checkpoint/_src/serialization/ocdbt_specs_e2e_test.py b/checkpoint/orbax/checkpoint/_src/serialization/ocdbt_specs_e2e_test.py new file mode 100644 index 000000000..8c9a47acb --- /dev/null +++ b/checkpoint/orbax/checkpoint/_src/serialization/ocdbt_specs_e2e_test.py @@ -0,0 +1,430 @@ +# Copyright 2026 The Orbax Authors. +# +# Licensed 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. + +"""End-to-end tests for distributed arrays serialization with OCDBT.""" + +import asyncio +import contextlib +import dataclasses +import tempfile +from typing import TypeAlias +import unittest + +from absl.testing import absltest +from absl.testing import parameterized +from etils import epath +import numpy as np +from orbax.checkpoint._src.arrays import fragments as fragments_lib +from orbax.checkpoint._src.arrays import subchunking +from orbax.checkpoint._src.arrays import types as arrays_types +from orbax.checkpoint._src.serialization import ocdbt_process_spec +from orbax.checkpoint._src.serialization import ocdbt_utils +from orbax.checkpoint._src.serialization import tensorstore_utils +import tensorstore as ts + + +@dataclasses.dataclass(frozen=True) +class TestArray: + """A test array. + + Attributes: + name: The name of the array. + value: The value (unsharded) of the array. + fragments_by_process_id: A mapping describing which array fragments should + be written by which test process. + """ + name: str + value: np.ndarray + fragments_by_process_id: dict[str, fragments_lib.NpFragments] + + +TestData: TypeAlias = tuple[TestArray, ...] + + +def build_test_data() -> TestData: + """Builds test arrays. Test setup simulates 4 test processes.""" + # Unsharded array. + a = np.arange(10) + + # Array sharded across 2 out of 4 test processes. + b = np.arange(4* 5).reshape(4, 5) + b_sharded_shape = (1, 5) + + # Array sharded across 2 out of 4 test processes. Large enough to trigger + # writing of an OCDBT "value" data file + b_large = np.arange(4 * 10 * 1024).reshape(4, 10, 1024) + b_large_sharded_shape = (1, 10, 1024) + + # Array sharded across 3 out of 4 test processes. + c = np.arange(6 * 13).reshape(6, 13) + c_sharded_shape = (2, 13) + + # Array sharded across all 4 test processes. + d = np.arange(3 * 12 * 8).reshape(3, 12, 8) + d_sharded_shape = (3, 6, 4) + + def _generate_fragments_by_process_id( + array: np.ndarray, + sharded_shape: arrays_types.Shape, + num_processes: int, + ) -> dict[str, fragments_lib.NpFragments]: + """Generates fragments_by_process_id for a given array.""" + sharded_fragments = subchunking.chunk_fragments( + fragments_lib.NpFragments.all_of(array), sharded_shape + ) + num_sharded_fragments = len(sharded_fragments.fragments) + assert num_sharded_fragments % num_processes == 0 + fragments_per_process = num_sharded_fragments // num_processes + return { + f"h{i}": fragments_lib.NpFragments( + shape=array.shape, + dtype=array.dtype, + fragments=sharded_fragments.fragments[ + (i * fragments_per_process) : (i + 1) * fragments_per_process + ], + ) + for i in range(num_processes) + } + + return ( + TestArray( + name="a", + value=a, + fragments_by_process_id={"h0": fragments_lib.NpFragments.all_of(a)}, + ), + TestArray( + name="b", + value=b, + fragments_by_process_id=_generate_fragments_by_process_id( + b, b_sharded_shape, num_processes=2 + ), + ), + TestArray( + name="b_large", + value=b_large, + fragments_by_process_id=_generate_fragments_by_process_id( + b_large, b_large_sharded_shape, num_processes=2 + ), + ), + TestArray( + name="c", + value=c, + fragments_by_process_id=_generate_fragments_by_process_id( + c, c_sharded_shape, num_processes=3 + ), + ), + TestArray( + name="d", + value=d, + fragments_by_process_id=_generate_fragments_by_process_id( + d, d_sharded_shape, num_processes=4 + ), + ), + ) + + +def all_process_ids(test_data: TestData) -> set[str]: + """Returns all unique process ids in the given test data.""" + return set( + [ + process_id # pylint: disable=g-complex-comprehension + for array in test_data + for process_id in array.fragments_by_process_id + ] + ) + + +def _should_create_ts(fragments: fragments_lib.NpFragments) -> bool: + """Determines if TensorStore array metadata needs to be written.""" + # Only do this if the fragments contain the "leading" array element + # (0th in array's flat index space). + return any((fragment.start == 0).all() for fragment in fragments.fragments) + + +async def _write_array( + name: str, + path: epath.Path, + array_fragments: fragments_lib.NpFragments, + process_id: str, + ts_context: ts.Context, + *, + store_ocdbt_metadata_and_values_separately: bool = False, + temporary_metadata_context: ( + tensorstore_utils.OcdbtTemporaryMetadataContext | None + ) = None, +) -> None: + """Writes array fragments to the given path with the given process id.""" + array_write_tspec = tensorstore_utils.ArrayWriteSpec( + directory=path.as_posix(), + relative_array_filename=name, + global_shape=array_fragments.shape, + write_shape=array_fragments.fragments[0].value.shape, + dtype=array_fragments.dtype, + use_ocdbt=True, + process_id=process_id, + store_ocdbt_metadata_and_values_separately=( + store_ocdbt_metadata_and_values_separately + ), + ocdbt_temporary_metadata_context=temporary_metadata_context, + ).json + + if _should_create_ts(array_fragments): + # Open with create=True once (`should_create` is supposed to be True for + # only one of the hosts), so that the metadata (.zarray) is written once. + array_ts = await ts.open( + array_write_tspec, + context=ts_context, + open=True, + create=True, + ) + else: + array_ts = await ts.open( + array_write_tspec, + context=ts_context, + open=True, + assume_metadata=True, + ) + + write_futures = [] + for fragment in array_fragments.fragments: + write_futures.append(array_ts[fragment.index].write(fragment.value)) + + await asyncio.gather(*write_futures) + + +async def _read_array( + name: str, + path: epath.Path, + ts_context: ts.Context, +) -> np.ndarray: + """Reads an array with the given name from the given path.""" + array_ts = await ts.open( + { + "driver": "zarr", + "kvstore": tensorstore_utils.build_kvstore_tspec( + path.as_posix(), + name, + use_ocdbt=True, + ), + }, + context=ts_context, + read=True, + ) + return await array_ts.read() + + +async def _verify_array_data(test_data: TestData, path: epath.Path) -> None: + """Verifies all test arrays' data in the given directory.""" + ts_context = tensorstore_utils.get_ts_context(use_ocdbt=True) + for array in test_data: + read_array_value = await _read_array(array.name, path, ts_context) + np.testing.assert_array_equal(read_array_value, array.value) + + +class OcdbtSpecsE2eTest( + unittest.IsolatedAsyncioTestCase, parameterized.TestCase +): + + def _verify_per_process_ocdbt_files( + self, + test_data: TestData, + path: epath.Path, + store_ocdbt_metadata_and_values_separately: bool, + ) -> None: + """Verifies key per-process OCDBT files at the given path.""" + for process_id in all_process_ids(test_data): + process_spec = ocdbt_process_spec.OcdbtProcessSpec(process_id=process_id) + process_dir = path / str(process_spec) + self.assertTrue((process_dir / "manifest.ocdbt").exists()) + # Values and metadata directory naming depends on the + # store_ocdbt_metadata_and_values_separately flag. + if store_ocdbt_metadata_and_values_separately: + self.assertFalse((process_dir / "d").is_dir()) + self.assertTrue((process_dir / "ocdbt_meta").is_dir()) + # b_large should have generated files written to ocdbt_data/ subdir. + if process_id in ("h0", "h1"): + self.assertTrue((process_dir / "ocdbt_data").is_dir()) + else: + self.assertTrue((process_dir / "d").is_dir()) + self.assertFalse((process_dir / "ocdbt_meta").is_dir()) + self.assertFalse((process_dir / "ocdbt_data").is_dir()) + + @parameterized.product( + store_ocdbt_metadata_and_values_separately=(False, True), + ) + async def test_write(self, store_ocdbt_metadata_and_values_separately: bool): + test_dir = epath.Path(self.create_tempdir()) / "test_data" + test_dir.mkdir(parents=True, exist_ok=True) + + test_data = build_test_data() + + # Create process-specific subdirectories. + for process_id in all_process_ids(test_data): + spec = ocdbt_process_spec.OcdbtProcessSpec(process_id=process_id) + (test_dir / str(spec)).mkdir(parents=False, exist_ok=False) + + ts_context = tensorstore_utils.get_ts_context(use_ocdbt=True) + + async def _write(): + write_futures = [] + for array in test_data: + for process_id, fragments in array.fragments_by_process_id.items(): + write_futures.append( + _write_array( + array.name, + test_dir, + fragments, + process_id, + ts_context, + store_ocdbt_metadata_and_values_separately=( + store_ocdbt_metadata_and_values_separately + ), + ) + ) + await asyncio.gather(*write_futures) + + await _write() + self._verify_per_process_ocdbt_files( + test_data, + test_dir, + store_ocdbt_metadata_and_values_separately, + ) + + await ocdbt_utils.merge_ocdbt_per_process_files( + test_dir, ts_context, use_zarr3=False, enable_validation=False + ) + await _verify_array_data(test_data, test_dir) + + @parameterized.product( + store_ocdbt_metadata_and_values_separately=(False, True), + ) + async def test_write_with_temporary_metadata_context( + self, + store_ocdbt_metadata_and_values_separately: bool, + ): + test_dir = epath.Path(self.create_tempdir()) / "test_data" + test_dir.mkdir(parents=True, exist_ok=True) + + test_data = build_test_data() + + # Create process-specific persistent subdirectories. + for process_id in all_process_ids(test_data): + spec = ocdbt_process_spec.OcdbtProcessSpec(process_id=process_id) + (test_dir / str(spec)).mkdir(parents=False, exist_ok=False) + + ts_context = tensorstore_utils.get_ts_context(use_ocdbt=True) + + async def _write(): + exit_stack = contextlib.ExitStack() + with exit_stack: + # Allocate temporary directories for each process's temporary metadata. + tmp_metadata_context_by_process_id: dict[ + str, tensorstore_utils.OcdbtTemporaryMetadataContext + ] = {} + + def _get_tmp_metadata_context_by_process_id( + process_id: str, + ) -> tensorstore_utils.OcdbtTemporaryMetadataContext: + if process_id not in tmp_metadata_context_by_process_id: + tmp_context = exit_stack.enter_context( + tempfile.TemporaryDirectory() + ) + tmp_metadata_context_by_process_id[process_id] = ( + tensorstore_utils.OcdbtTemporaryMetadataContext( + path=epath.Path(tmp_context) + ) + ) + return tmp_metadata_context_by_process_id[process_id] + + write_futures = [] + for array in test_data: + for process_id, fragments in array.fragments_by_process_id.items(): + write_futures.append( + _write_array( + array.name, + test_dir, + fragments, + process_id, + ts_context, + store_ocdbt_metadata_and_values_separately=( + store_ocdbt_metadata_and_values_separately + ), + temporary_metadata_context=( + _get_tmp_metadata_context_by_process_id(process_id) + ), + ) + ) + await asyncio.gather(*write_futures) + + # Verify that temporary metadata has been written as expected. + for process_id in all_process_ids(test_data): + process_spec = ocdbt_process_spec.OcdbtProcessSpec( + process_id=process_id + ) + process_dir = test_dir / str(process_spec) + tmp_metadata_dir = _get_tmp_metadata_context_by_process_id( + process_id + ).path + # Manifest should not have been written to the final destination. + self.assertFalse((process_dir / "manifest.ocdbt").exists()) + tmp_manifest_path = ( + tmp_metadata_dir / "_ocdbt_tmp_meta/manifest.ocdbt" + ) + self.assertTrue(tmp_manifest_path.exists()) + # Check that metadata is not written to the final destination. We can + # only check this easily if we're writing metadata and values + # separately. + if store_ocdbt_metadata_and_values_separately: + self.assertFalse((process_dir / "ocdbt_meta").is_dir()) + self.assertTrue( + (tmp_metadata_dir / "_ocdbt_tmp_meta").is_dir() + ) + # b_large should have generated files written to ocdbt_data/ subdir + # directly, bypassing the temporary metadata context. + if process_id in ("h0", "h1"): + self.assertTrue((process_dir / "ocdbt_data").is_dir()) + + # Commit temporary metadata to persistent storage. + commit_metadata_futures = [] + for process_id in all_process_ids(test_data): + process_spec = ocdbt_process_spec.OcdbtProcessSpec( + process_id=process_id + ) + commit_metadata_futures.append( + ocdbt_utils.commit_temporary_ocdbt_metadata( + test_dir / str(process_spec), + _get_tmp_metadata_context_by_process_id(process_id), + ts_context, + store_ocdbt_metadata_and_values_separately=( + store_ocdbt_metadata_and_values_separately + ), + ) + ) + await asyncio.gather(*commit_metadata_futures) + + await _write() + self._verify_per_process_ocdbt_files( + test_data, + test_dir, + store_ocdbt_metadata_and_values_separately, + ) + + await ocdbt_utils.merge_ocdbt_per_process_files( + test_dir, ts_context, use_zarr3=False, enable_validation=False + ) + await _verify_array_data(test_data, test_dir) + + +if __name__ == "__main__": + absltest.main() diff --git a/checkpoint/orbax/checkpoint/_src/serialization/ocdbt_utils.py b/checkpoint/orbax/checkpoint/_src/serialization/ocdbt_utils.py index 8d74ec420..e3c4aca9c 100644 --- a/checkpoint/orbax/checkpoint/_src/serialization/ocdbt_utils.py +++ b/checkpoint/orbax/checkpoint/_src/serialization/ocdbt_utils.py @@ -144,6 +144,7 @@ async def merge_ocdbt_per_process_files( ts_context: ts.Context, use_zarr3: bool, enable_validation: bool = True, + # local_temporary_metadata_path: epath.Path | None = None, ): """Merges OCDBT files written to per-process subdirectories. @@ -164,6 +165,8 @@ async def merge_ocdbt_per_process_files( validation. enable_validation: If True, validate params after merging. May have a performance impact. + # local_temporary_metadata_path: Path to the local temporary directory for + # storing metadata. """ start_time = time.time() open_ops = [] @@ -220,6 +223,37 @@ async def merge_ocdbt_per_process_files( ) +async def commit_temporary_ocdbt_metadata( + persistent_path: epath.Path, + temporary_metadata_context: ts_utils.OcdbtTemporaryMetadataContext, + ts_context: ts.Context, + *, + store_ocdbt_metadata_and_values_separately: bool = False, +) -> None: + """Commits temporary OCDBT metadata to the persistent metadata directory.""" + target_kvstore_tspec = ts_utils.build_kvstore_tspec( + persistent_path.as_posix(), + use_ocdbt=True, + ocdbt_write_options=ts_utils.OcdbtKvStoreWriteOptions( + mode=ts_utils.OcdbtWriteMode.COMMIT_TEMPORARY, + store_ocdbt_metadata_and_values_separately=( + store_ocdbt_metadata_and_values_separately + ), + ), + ocdbt_temporary_metadata_context=temporary_metadata_context, + ) + source_kvstore_tspec = ts_utils.build_kvstore_tspec( + persistent_path.as_posix(), + use_ocdbt=True, + ocdbt_temporary_metadata_context=temporary_metadata_context, + ) + target_kvstore, source_kvstore = await asyncio.gather( + ts_utils.open_kv_store(target_kvstore_tspec, ts_context), + ts_utils.open_kv_store(source_kvstore_tspec, ts_context), + ) + await source_kvstore.experimental_copy_range_to(target_kvstore) + + def get_process_index_for_subdir( use_ocdbt: bool, override_ocdbt_process_id: Optional[str] = None, diff --git a/checkpoint/orbax/checkpoint/_src/serialization/tensorstore_utils.py b/checkpoint/orbax/checkpoint/_src/serialization/tensorstore_utils.py index 9fe37b4c3..c990e1f26 100644 --- a/checkpoint/orbax/checkpoint/_src/serialization/tensorstore_utils.py +++ b/checkpoint/orbax/checkpoint/_src/serialization/tensorstore_utils.py @@ -69,6 +69,7 @@ # 'ocdbt_data/' subdirectory. _OCDBT_SPLIT_VALUE_DATA_PREFIX = 'ocdbt_data/' _OCDBT_SPLIT_META_DATA_PREFIX = 'ocdbt_meta/' +_OCDBT_TMP_METADATA_PREFIX = '_ocdbt_tmp_meta/' ZARR_VER2 = 'zarr' ZARR_VER3 = 'zarr3' @@ -155,10 +156,13 @@ class OcdbtWriteMode(enum.Enum): WRITE: Used when writing checkpoint data. MERGE: Used for target (parent) KvStore when merging OCDBT metadata from all per-process subdirectories. + COMMIT_TEMPORARY: Used when committing metadata accumulated in a temporary + metadata directory to the persistent directory. """ WRITE = 'write' MERGE = 'merge' + COMMIT_TEMPORARY = 'commit_temporary' @dataclasses.dataclass(frozen=True) @@ -178,6 +182,12 @@ class OcdbtKvStoreWriteOptions: store_ocdbt_metadata_and_values_separately: bool = False +@dataclasses.dataclass(frozen=True) +class OcdbtTemporaryMetadataContext: + """Options specific to OCDBT temporary metadata context.""" + path: epath.Path + + def _get_kvstore_for_gcs(ckpt_path: str) -> JsonSpec: """Constructs a TensorStore kvstore spec for a GCS path.""" m = re.fullmatch(_GCS_PATH_RE, ckpt_path, re.DOTALL) @@ -214,6 +224,7 @@ def _build_ocdbt_kvstore_tspec( *, process_spec: OcdbtProcessSpec | None = None, write_options: OcdbtKvStoreWriteOptions | None = None, + temporary_metadata_context: OcdbtTemporaryMetadataContext | None = None, ) -> JsonSpec: """Constructs a spec for a Tensorstore OCDBT KvStore. @@ -225,6 +236,7 @@ def _build_ocdbt_kvstore_tspec( name). write_options: Options specific to OCDBT KvStore write modes. Should be provided when the kvstore will be used for writing or merging. + temporary_metadata_context: Context for local temporary metadata directory. Returns: A Tensorstore KvStore spec in dictionary form. @@ -242,7 +254,11 @@ def _build_ocdbt_kvstore_tspec( if is_gcs_path: base_driver_spec = _get_kvstore_for_gcs(directory) else: - base_driver_spec = {'driver': DEFAULT_DRIVER, 'path': str(directory)} + trailing_slash = '/' if temporary_metadata_context is not None else '' + base_driver_spec = { + 'driver': DEFAULT_DRIVER, + 'path': str(directory) + trailing_slash, + } # For OCDBT on local filesystems (including GCSFuse), we can safely use # non-atomic writes for data files to avoid expensive renames. However, @@ -259,31 +275,76 @@ def _build_ocdbt_kvstore_tspec( ) resolved_base_spec = base_driver_spec + value_prefix = None + metadata_prefix = None + manifest_spec = None + if ( isinstance(resolved_base_spec, dict) and resolved_base_spec.get('driver') == 'file' ): - kv_spec = { - 'driver': 'ocdbt', - 'base': { - **resolved_base_spec, - 'file_io_locking': {'mode': 'non_atomic'}, - }, - 'manifest': base_driver_spec, + manifest_spec = base_driver_spec + base_driver_spec = { + **resolved_base_spec, + 'file_io_locking': {'mode': 'non_atomic'}, } - else: - kv_spec = { - 'driver': 'ocdbt', - 'base': base_driver_spec, + + if write_options is not None: + if ( + write_options.mode == OcdbtWriteMode.COMMIT_TEMPORARY + and temporary_metadata_context is None + ): + raise ValueError('OCDBT commit mode requires temporary metadata context.') + if write_options.store_ocdbt_metadata_and_values_separately: + value_prefix = _OCDBT_SPLIT_VALUE_DATA_PREFIX + metadata_prefix = _OCDBT_SPLIT_META_DATA_PREFIX + + if temporary_metadata_context is not None: + if write_options is not None and write_options.mode == OcdbtWriteMode.MERGE: + raise ValueError( + 'OCDBT merge mode does not support temporary metadata context.' + ) + if ( + write_options is None + or write_options.mode != OcdbtWriteMode.COMMIT_TEMPORARY + ): + manifest_spec = ( + f'file://{temporary_metadata_context.path}' + f'/{_OCDBT_TMP_METADATA_PREFIX}' + ) + if write_options is not None and write_options.mode == OcdbtWriteMode.WRITE: + metadata_prefix = _OCDBT_TMP_METADATA_PREFIX + + base_driver_spec = { + 'driver': 'kvstack', + 'layers': [ + # Write to the real persistent checkpoint directory by default. + {'base': base_driver_spec}, + # Per-process metadata is stored in the separate local + # temporary directory. + { + 'prefix': _OCDBT_TMP_METADATA_PREFIX, + 'base': ( + f'file://{temporary_metadata_context.path}/' + ), + }, + ], } + kv_spec = {'driver': 'ocdbt', 'base': base_driver_spec} + + if manifest_spec is not None: + kv_spec['manifest'] = manifest_spec # pyrefly: ignore[bad-assignment] + if value_prefix: + kv_spec['value_data_prefix'] = value_prefix + if metadata_prefix: + kv_spec['btree_node_data_prefix'] = metadata_prefix + kv_spec['version_tree_node_data_prefix'] = metadata_prefix + if write_options is not None: _add_ocdbt_write_options( kv_spec, target_data_file_size=write_options.target_data_file_size, - store_ocdbt_metadata_and_values_separately=( - write_options.store_ocdbt_metadata_and_values_separately - ), ) if name is not None: @@ -334,6 +395,9 @@ def build_kvstore_tspec( use_ocdbt: bool = True, ocdbt_process_spec: OcdbtProcessSpec | None = None, ocdbt_write_options: OcdbtKvStoreWriteOptions | None = None, + ocdbt_temporary_metadata_context: ( + OcdbtTemporaryMetadataContext | None + ) = None, ) -> JsonSpec: """Constructs a spec for a Tensorstore KvStore. @@ -346,6 +410,8 @@ def build_kvstore_tspec( name). ocdbt_write_options: Options specific to OCDBT KvStore write modes. Should be provided when the kvstore will be used for writing or merging. + ocdbt_temporary_metadata_context: Context for local temporary metadata + directory. Returns: A Tensorstore KvStore spec in dictionary form. @@ -356,6 +422,7 @@ def build_kvstore_tspec( name=name, process_spec=ocdbt_process_spec, write_options=ocdbt_write_options, + temporary_metadata_context=ocdbt_temporary_metadata_context, ) return _build_non_ocdbt_kvstore_tspec(directory=directory, name=name) @@ -387,8 +454,6 @@ def _get_backend_ocdbt_target_data_file_size( def _add_ocdbt_write_options( kvstore_tspec: JsonSpec, target_data_file_size: int | None = None, - *, - store_ocdbt_metadata_and_values_separately: bool = False, ) -> None: """Adds write-specific options to a TensorStore OCDBT KVStore spec.""" if target_data_file_size is None: @@ -403,13 +468,6 @@ def _add_ocdbt_write_options( ) kvstore_tspec['target_data_file_size'] = target_data_file_size - if store_ocdbt_metadata_and_values_separately: - kvstore_tspec['value_data_prefix'] = _OCDBT_SPLIT_VALUE_DATA_PREFIX - kvstore_tspec['btree_node_data_prefix'] = _OCDBT_SPLIT_META_DATA_PREFIX - kvstore_tspec['version_tree_node_data_prefix'] = ( - _OCDBT_SPLIT_META_DATA_PREFIX - ) - kvstore_tspec['config'] = { # Store .zarray metadata inline but not large chunks. # If separate storage for OCDBT metadata is enabled, this will mean that @@ -635,6 +693,9 @@ def __init__( replica_separate_folder: bool = False, ext_metadata: ExtMetadata | None = None, store_ocdbt_metadata_and_values_separately: bool = False, + ocdbt_temporary_metadata_context: ( + OcdbtTemporaryMetadataContext | None + ) = None, ): """Builds a TensorStore spec for writing an array.""" # Construct the underlying KvStore spec. @@ -656,6 +717,7 @@ def __init__( store_ocdbt_metadata_and_values_separately ), ), + ocdbt_temporary_metadata_context=ocdbt_temporary_metadata_context, ) # Construct the top-level array spec. tspec = {