Refactor rivulet manifest files to use deltacat #464

mcember · 2025-01-24T20:48:27Z

Summary

See milestone https://github.com/ray-project/deltacat/milestone/5

Remove json manifest format in favor of deltacat delta (which wraps manifest)
Begin refactoring some internal classes

Note that this PR uses a metastore format like
$root/
$metadata/

$data/

<delta_id_n>/
/rev

This is NOT the final structure for using the deltacat metastore. In part this is waiting on the storage interface from #455

Discussion points:

How is a deltcat writer expected to assign stream position
does deltacat have opinions about where data files are written?

Testing

rivulet unit tests

…deltacat-metastore

deltacat/storage/rivulet/fs/fs_utils.py

…deltacat-metastore

mcember · 2025-01-29T16:18:17Z

Makefile

@@ -35,6 +35,9 @@ lint: install
 test: install
 	venv/bin/pytest -m "not integration"

+unit-test: install


I just added this to run tests more rapidly. Note that some of the unit tests for compaction take a while so we should consider moving them to benchmarks

mcember · 2025-01-29T16:33:16Z

deltacat/storage/model/manifest.py

@@ -135,7 +135,7 @@ def of(
                        f"'{entry_content_type}'"
                    )
                    raise ValueError(msg)
-                entry_content_encoding = meta["content_encoding"]
+                entry_content_encoding = meta.get("content_encoding", None)


This was raising a ValueError for manifests with no content encoding (like in Rivulet...). We don't populate lots of fields currently for rivulet deltas/manifests so we will have to update code to populate them (see: #476)

anshumankomawar · 2025-01-29T23:37:58Z

deltacat/storage/rivulet/reader/dataset_metastore.py

+            looks for the latest revision
+        """
+        rev_directory = os.path.join(delta_dir, "rev")
+        revisions = filesystem.get_file_info(fs.FileSelector(rev_directory))


nit: should we use the fs utils method here? I believe it adds some error handling around get_file_info.

I can include it as part of my changes as well.

anshumankomawar · 2025-01-29T23:40:14Z

deltacat/storage/rivulet/reader/dataset_metastore.py

+        root_path, filesystem = resolve_path_and_filesystem(self.delta_root_uri)
+        root_children = filesystem.get_file_info(fs.FileSelector(root_path))
+        delta_directories = [
+            child
+            for child in root_children
+            if not child.is_file and child.base_name not in excluded_dir_names
+        ]


Will have to change how we do this once I introduce the namespace -> table -> table_version -> stream -> partition hierarchy. But their probably isn't a better solution until we start filling out the storage interface.

thesalus

How is a deltcat writer expected to assign stream position

Should the writer be providing a stream position in all scenarios? Is it valid for the writer to ask to append a transaction (and simply take whatever (valid) stream position that gets generated)?

thesalus · 2025-01-30T22:44:53Z

deltacat/tests/storage/rivulet/writer/test_dataset_writer.py

    validate_with_full_scan,
-    assert_data_file_extension,
    create_dataset_for_method,
+    assert_data_file_extension,


Aside: What happened here? Is there a generally-accepted linting rule/convention to apply a consistent ordering to imports?

hmm not sure how this happened, if it was the linter or IDE or what

thesalus · 2025-01-30T22:51:04Z

deltacat/storage/rivulet/metastore/delta.py

+                    # TODO have rivulet writer populate these values
+                    # see: https://github.com/ray-project/deltacat/issues/476
+                    meta=ManifestMeta.of(
+                        record_count=None,  # or known


Aside: What does this comment mean?

it means that e.g. when writing data rivulet writer will know the record count and propagate that to manifests. Not updating code since I think the comment/linked ticket are sufficient

thesalus · 2025-01-30T22:52:35Z

deltacat/storage/rivulet/metastore/delta.py

+            # Using microsecond precision timestamp as stream position
+            # TODO consider having storage interface auto assign stream position
+            stream_position=time.time_ns(),


Nitpick: this appears to be nanosecond precision.

Yea. Not updating for now since will move automatic stream position generation behind storage/catalog interface

thesalus · 2025-01-30T22:58:20Z

deltacat/storage/rivulet/metastore/delta.py

+        assert (
+            len(paths) == 1
+        ), "expected delta commit transaction to write exactly 1 metafile"


At this point, the transaction has been committed. What (w|sh)ould be the right way to handle this AssertionError? (Is there a way for the caller to recover from such a scenario?)

This is more like a sanity check type exception - we only added one metafile and if the transaction committed more than one than something is seriously wrong

We are abstracting the low level Tx handling behind storage interface in near future

thesalus · 2025-01-30T23:01:58Z

deltacat/storage/rivulet/reader/dataset_metastore.py

-            yield ManifestAccessor(manifest, self.file_provider, self.sst_reader)
+        # Rivulet data and SST files written to /data and /metadata
+        # Deltacat transactions written to /txn
+        excluded_dir_names = ["data", "metadata", "txn"]


Minor: wouldn't it be more resilient to only target the manifest directory (until the deltacat storage API is available)?

thesalus · 2025-01-30T23:02:51Z

deltacat/storage/rivulet/reader/dataset_metastore.py

+    def _get_delta(self, delta_dir: str, filesystem: FileSystem) -> RivuletDelta:
+        """
+        Given a DeltaCat delta directory, find latest delta file


Minor: might I suggest _get_latest_delta?

thesalus · 2025-01-30T23:04:09Z

deltacat/storage/rivulet/reader/dataset_metastore.py

+        latest_revision = None
+        for revision in revisions:
+            latest_revision = (
+                revision if not latest_revision else max(latest_revision, revision.path)


Shouldn't this be

Suggested change

revision if not latest_revision else max(latest_revision, revision.path)

revision.path if not latest_revision else max(latest_revision, revision.path)

Otherwise you wouldn't be comparing the paths themselves.

revision in this case is a path since it is generated by filesystem list

mattcember added 5 commits January 22, 2025 15:12

remove datafile from manifest. begin initial deltacat manifest io

eff47c5

Merge remote-tracking branch 'origin/2.0' into dev/rivulet/integrate-…

58d0541

…deltacat-metastore

start big refactor of manifests

e639434

get manifest reading from metastore working

1a1df31

get all tests to pass

7830701

mcember requested review from thesalus, pdames and flliver January 24, 2025 20:48

mcember commented Jan 24, 2025

View reviewed changes

deltacat/storage/rivulet/fs/fs_utils.py Outdated Show resolved Hide resolved

run linter

6737e0b

This was referenced Jan 24, 2025

Wrap rivulet manifest in Deltacat delta #463

Open

Reconcile data file / sst files between deltacat and rivulet #469

Open

mcember added this to the Integrate Deltacat and Rivulet Metastores milestone Jan 24, 2025

mattcember added 8 commits January 28, 2025 14:58

Merge remote-tracking branch 'origin/2.0' into dev/rivulet/integrate-…

beecce0

…deltacat-metastore

fix tests

60df7c3

misc test fixes

3166252

run linter

f9224e0

Merge remote-tracking branch 'origin/2.0' into dev/rivulet/integrate-…

6d04f77

…deltacat-metastore

move to mainline filesystem utils. revert bad merge of dataset

872f8fe

fix merge issue and move all temp dirs to conftest temp dir

2e49359

run linter and fix bug in manifest

1e7f9af

mcember commented Jan 29, 2025

View reviewed changes

self review, tweak TODOs with ticket links

868ed5d

anshumankomawar approved these changes Jan 29, 2025

View reviewed changes

thesalus approved these changes Jan 30, 2025

View reviewed changes

mcember merged commit 250fa11 into 2.0 Feb 4, 2025
3 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Refactor rivulet manifest files to use deltacat #464

Refactor rivulet manifest files to use deltacat #464

mcember commented Jan 24, 2025

mcember Jan 29, 2025

mcember Jan 29, 2025

anshumankomawar Jan 29, 2025

anshumankomawar Jan 29, 2025

thesalus left a comment

thesalus Jan 30, 2025

mcember Feb 4, 2025

thesalus Jan 30, 2025

mcember Feb 4, 2025

thesalus Jan 30, 2025

mcember Feb 4, 2025

thesalus Jan 30, 2025

mcember Feb 4, 2025

thesalus Jan 30, 2025

thesalus Jan 30, 2025

thesalus Jan 30, 2025

mcember Feb 4, 2025

	revision if not latest_revision else max(latest_revision, revision.path)
	revision.path if not latest_revision else max(latest_revision, revision.path)

Refactor rivulet manifest files to use deltacat #464

Refactor rivulet manifest files to use deltacat #464

Conversation

mcember commented Jan 24, 2025

Summary

Testing

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

thesalus left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment