You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/08/24 17:09:53 UTC
(texera) 01/01: refactor(file-service): make the file-tree builder resource-aware for model upload (#7762)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7762-84e2d7d6498cfccdc1dcef33e1a31dbfaf976f2c
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 9b3c63aa20b07060abb964b1ca2bfb925795575f
Author: Tanishq Gandhi <56...@users.noreply.github.com>
AuthorDate: Mon Aug 24 17:09:30 2026 +0000
refactor(file-service): make the file-tree builder resource-aware for model upload (#7762)
### What changes were proposed in this PR?
Refactoring groundwork so model upload can reuse the dataset file-tree
builder. No new
features and no change to dataset behavior.
`fromLakeFSRepositoryCommittedObjects` hardcoded the resource type as
the leading path
segment of the tree. `FileResolver` uses that segment to pick the
backing table, so a model
tree built by it would report `/dataset/...` paths and resolve against
the dataset table.
`resourceType` is now a required parameter, so each call site says which
resource it is
building. The three dataset call sites pass `ResourceType.Dataset`, the
value that was
hardcoded before, so datasets are unaffected.
Folded in from review:
- The class is renamed `DatasetFileNode` → `LakeFSFileNode`, with its
serializer and specs,
and moved out of `type/dataset/` since the package was already
`service.type`. This was
queued for the model PRs and pulled forward here, where the class
becomes
resource-neutral.
- File-vs-directory is now decided positionally rather than by value.
Previously a path
repeating its final segment (`model/model`) made the intermediate
directory the leaf, so
it took the object's size and the real file hung underneath it.
A promote-on-collision change was also tried here and then reverted: for
a version holding
both `model` and `model/weights.bin` it dropped the colliding object
from the tree and from
the reported size. The pre-existing behaviour is kept instead — two
siblings with the same
name, untidy but every object listed, sized and downloadable — and a
test now pins that
property. Representing both unambiguously, or rejecting such uploads, is
worth its own
issue.
The JSON contract is unchanged, so the frontend is unaffected: the
serializer writes
explicit field names, and the tree component sets `displayField: "name"`
with no `idField`,
so nodes carry generated ids and selection keys off node data rather
than name.
### Any related issues, documentation, discussions?
Part of #6494, groundwork for #6498. Based directly on `main` and
independent of the other
open PRs in the model stack.
### How was this PR tested?
`sbt "FileService/test"` — 323 tests, 13 suites, 0 failures. `sbt
"WorkflowCore/test"` —
787 tests, 0 failures. scalafmt and scalafix clean.
The dataset side is pinned by the existing tree test, unchanged, still
asserting
`/dataset/bob@texera.com/twitter/v1/b/1.csv` — that is what demonstrates
the no-op. Added:
- a model tree rooted at `/model/`, and the `model/model` case — each
verified to fail
without its fix;
- the object/directory collision, which characterizes current behaviour
rather than a fix,
so it also guards against re-introducing the promote;
- coverage for `retrieveDatasetVersionRootFileNodes` and
`retrieveLatestDatasetVersion`,
which had none, pinning the dataset prefix end-to-end.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5)
---------
Co-authored-by: Claude Opus 5 <no...@anthropic.com>
Co-authored-by: ali <al...@gmail.com>
---
.../org/apache/texera/service/FileService.scala | 8 +-
.../texera/service/resource/DatasetResource.scala | 17 +-
.../texera/service/type/LakeFSFileNode.scala | 183 ++++++++++++++++++
.../service/type/dataset/DatasetFileNode.scala | 170 -----------------
...rializer.java => LakeFSFileNodeSerializer.java} | 14 +-
.../service/resource/DatasetResourceSpec.scala | 33 ++++
.../texera/service/type/LakeFSFileNodeSpec.scala | 204 +++++++++++++++++++++
.../service/type/dataset/DatasetFileNodeSpec.scala | 131 -------------
...ec.scala => LakeFSFileNodeSerializerSpec.scala} | 28 +--
9 files changed, 455 insertions(+), 333 deletions(-)
diff --git a/file-service/src/main/scala/org/apache/texera/service/FileService.scala b/file-service/src/main/scala/org/apache/texera/service/FileService.scala
index 337ff2ad1f..9ed86b9207 100644
--- a/file-service/src/main/scala/org/apache/texera/service/FileService.scala
+++ b/file-service/src/main/scala/org/apache/texera/service/FileService.scala
@@ -30,8 +30,8 @@ import org.apache.texera.common.util.RetryUtil
import org.apache.texera.amber.core.storage.util.LakeFSStorageClient
import org.apache.texera.auth.{AuthFeatures, RequestLoggingFilter, RoleAnnotationEnforcer}
import org.apache.texera.dao.SqlServer
-import org.apache.texera.service.`type`.DatasetFileNode
-import org.apache.texera.service.`type`.serde.DatasetFileNodeSerializer
+import org.apache.texera.service.`type`.LakeFSFileNode
+import org.apache.texera.service.`type`.serde.LakeFSFileNodeSerializer
import org.apache.texera.service.resource.{
DatasetAccessResource,
DatasetResource,
@@ -57,9 +57,9 @@ class FileService extends Application[FileServiceConfiguration] with LazyLogging
// Register Scala module to Dropwizard default object mapper
bootstrap.getObjectMapper.registerModule(DefaultScalaModule)
- // register a new custom module just for DatasetFileNode serde/deserde
+ // register a new custom module just for LakeFSFileNode serde/deserde
val customSerializerModule = new SimpleModule("CustomSerializers")
- customSerializerModule.addSerializer(classOf[DatasetFileNode], new DatasetFileNodeSerializer())
+ customSerializerModule.addSerializer(classOf[LakeFSFileNode], new LakeFSFileNodeSerializer())
bootstrap.getObjectMapper.registerModule(customSerializerModule)
}
diff --git a/file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala b/file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala
index 4f59d87862..0f123a25e5 100644
--- a/file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala
+++ b/file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala
@@ -49,7 +49,7 @@ import org.apache.texera.dao.jooq.generated.tables.pojos.{
DatasetUserAccess,
DatasetVersion
}
-import org.apache.texera.service.`type`.DatasetFileNode
+import org.apache.texera.service.`type`.LakeFSFileNode
import org.apache.texera.service.resource.DatasetAccessResource._
import org.apache.texera.service.resource.ResourceTables.{Dataset => DATASET_RESOURCE}
import org.apache.texera.service.resource.DatasetResource.{context, _}
@@ -269,7 +269,7 @@ object DatasetResource {
case class DashboardDatasetVersion(
datasetVersion: DatasetVersion,
- fileNodes: List[DatasetFileNode]
+ fileNodes: List[LakeFSFileNode]
)
case class CreateDatasetRequest(
@@ -296,7 +296,7 @@ object DatasetResource {
case class DatasetNameModification(did: Integer, name: String)
case class DatasetVersionRootFileNodesResponse(
- fileNodes: List[DatasetFileNode],
+ fileNodes: List[LakeFSFileNode],
size: Long
)
@@ -517,8 +517,9 @@ class DatasetResource extends LazyLogging {
DashboardDatasetVersion(
insertedVersion,
- DatasetFileNode
+ LakeFSFileNode
.fromLakeFSRepositoryCommittedObjects(
+ resourceType,
Map((user.getEmail, datasetName, newVersionName) -> fileNodes)
)
)
@@ -1339,8 +1340,9 @@ class DatasetResource extends LazyLogging {
throw new NotFoundException(ERR_DATASET_VERSION_NOT_FOUND_MESSAGE)
)
- val datasetsNode = DatasetFileNode
+ val datasetsNode = LakeFSFileNode
.fromLakeFSRepositoryCommittedObjects(
+ resourceType,
Map(
(
getOwner(ctx, did).getEmail,
@@ -1550,8 +1552,9 @@ class DatasetResource extends LazyLogging {
val datasetName = dataset.dataset.getName
val repositoryName = dataset.dataset.getRepositoryName
- val datasetsNode = DatasetFileNode
+ val datasetsNode = LakeFSFileNode
.fromLakeFSRepositoryCommittedObjects(
+ resourceType,
Map(
(dataset.ownerEmail, datasetName, datasetVersion.getName) -> LakeFSStorageClient
.retrieveObjectsOfVersion(repositoryName, datasetVersion.getVersionHash)
@@ -1575,7 +1578,7 @@ class DatasetResource extends LazyLogging {
.head
.children
.get,
- DatasetFileNode.calculateTotalSize(List(datasetsNode))
+ LakeFSFileNode.calculateTotalSize(List(datasetsNode))
)
}
diff --git a/file-service/src/main/scala/org/apache/texera/service/type/LakeFSFileNode.scala b/file-service/src/main/scala/org/apache/texera/service/type/LakeFSFileNode.scala
new file mode 100644
index 0000000000..be58e4ec45
--- /dev/null
+++ b/file-service/src/main/scala/org/apache/texera/service/type/LakeFSFileNode.scala
@@ -0,0 +1,183 @@
+/*
+ * 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.
+ */
+
+package org.apache.texera.service.`type`
+
+import io.lakefs.clients.sdk.model.ObjectStats
+import org.apache.texera.amber.core.storage.ResourceType
+
+import scala.collection.mutable
+
+// LakeFSFileNode represents a unique file in a versioned resource (dataset or model).
+// Its full path is in the format of:
+// /<resourceType>/ownerEmail/resourceName/versionName/fileRelativePath
+// e.g. /dataset/bob@texera.com/twitterDataset/v1/california/irvine/tw1.csv
+// /model/bob@texera.com/sentimentModel/v1/model.pt
+class LakeFSFileNode(
+ val name: String, // direct name of this node
+ val nodeType: String, // "file" or "directory"
+ val parent: LakeFSFileNode, // the parent node
+ val ownerEmail: String,
+ val size: Option[Long] = None, // size of the file in bytes, None if directory
+ var children: Option[List[LakeFSFileNode]] = None // Only populated if 'type' is 'directory'
+) {
+
+ // Ensure the type is either "file" or "directory"
+ require(nodeType == "file" || nodeType == "directory", "type must be 'file' or 'directory'")
+
+ // Getters for the parameters
+ def getName: String = name
+
+ def getNodeType: String = nodeType
+
+ def getParent: LakeFSFileNode = parent
+
+ def getOwnerEmail: String = ownerEmail
+
+ def getSize: Option[Long] = size
+
+ def getChildren: List[LakeFSFileNode] = children.getOrElse(List())
+
+ // Method to get the full file path
+ def getFilePath: String = {
+ val pathComponents = new mutable.ArrayBuffer[String]()
+ var currentNode: LakeFSFileNode = this
+ while (currentNode != null) {
+ if (currentNode.parent != null) { // Skip the root node to avoid double slashes
+ pathComponents.prepend(currentNode.name)
+ }
+ currentNode = currentNode.parent
+ }
+ "/" + pathComponents.mkString("/")
+ }
+}
+
+object LakeFSFileNode {
+
+ /**
+ * Converts a map of LakeFS committed objects into a structured file node tree.
+ *
+ * The tree is rooted at the resource-type segment (`dataset` or `model`), which
+ * [[LakeFSFileNode.getFilePath]] emits as the first path component. That prefix is
+ * what `FileResolver` keys on to pick the backing table, so it must match the
+ * resource the objects actually came from.
+ *
+ * @param resourceType The resource type the objects belong to (dataset or model).
+ * @param map A mapping from `(ownerEmail, resourceName, versionName)` to a list of committed objects.
+ * @return A list of root-level file nodes.
+ */
+ def fromLakeFSRepositoryCommittedObjects(
+ resourceType: ResourceType.Value,
+ map: Map[(String, String, String), List[ObjectStats]]
+ ): List[LakeFSFileNode] = {
+ val rootNode = new LakeFSFileNode("/", "directory", null, "")
+
+ // Root the tree at the resource-type prefix node (a directory node named e.g. "dataset").
+ val resourceTypeNode =
+ new LakeFSFileNode(resourceType.toString, "directory", rootNode, "")
+ rootNode.children = Some(List(resourceTypeNode))
+
+ // Owner level nodes map
+ val ownerNodes = mutable.Map[String, LakeFSFileNode]()
+
+ map.foreach {
+ case ((ownerEmail, resourceName, versionName), objects) =>
+ val ownerNode = ownerNodes.getOrElseUpdate(
+ ownerEmail, {
+ val newNode = new LakeFSFileNode(ownerEmail, "directory", resourceTypeNode, ownerEmail)
+ resourceTypeNode.children = Some(resourceTypeNode.getChildren :+ newNode)
+ newNode
+ }
+ )
+
+ val resourceNode = ownerNode.getChildren.find(_.getName == resourceName).getOrElse {
+ val newNode = new LakeFSFileNode(resourceName, "directory", ownerNode, ownerEmail)
+ ownerNode.children = Some(ownerNode.getChildren :+ newNode)
+ newNode
+ }
+
+ val versionNode = resourceNode.getChildren.find(_.getName == versionName).getOrElse {
+ val newNode = new LakeFSFileNode(versionName, "directory", resourceNode, ownerEmail)
+ resourceNode.children = Some(resourceNode.getChildren :+ newNode)
+ newNode
+ }
+
+ // Directories only. Registering leaves would let "model" be reused as the parent of
+ // "model/weights.bin", dropping the object it stands for from the tree and the size.
+ val directoryMap = mutable.Map[String, LakeFSFileNode]()
+ directoryMap("") = versionNode // Root of the resource version
+
+ // Process each object (file or directory) from LakeFS
+ objects.foreach { obj =>
+ val pathParts = obj.getPath.split("/").toList
+ var currentPath = ""
+ var parentNode: LakeFSFileNode = versionNode
+ //TODO: To check the logic of promoting the leaf vs duplicating the leaf
+ pathParts.zipWithIndex.foreach {
+ case (part, idx) =>
+ currentPath = if (currentPath.isEmpty) part else s"$currentPath/$part"
+
+ // Positional, not by value: "model/model" would otherwise make the directory
+ // the leaf.
+ val isFile = idx == pathParts.length - 1
+ val nodeType = if (isFile) "file" else "directory"
+ val fileSize = if (isFile) Some(obj.getSizeBytes.longValue()) else None
+
+ val node = directoryMap.getOrElse(
+ currentPath, {
+ val newNode = new LakeFSFileNode(part, nodeType, parentNode, ownerEmail, fileSize)
+ parentNode.children = Some(parentNode.getChildren :+ newNode)
+ if (!isFile) directoryMap(currentPath) = newNode
+ newNode
+ }
+ )
+
+ parentNode = node // Move parent reference deeper for next iteration
+ }
+ }
+ }
+
+ // Sorting function to sort children of a node alphabetically in descending order
+ def sortChildren(node: LakeFSFileNode): Unit = {
+ node.children = Some(node.getChildren.sortBy(_.getName)(Ordering.String.reverse))
+ node.getChildren.foreach(sortChildren)
+ }
+
+ // Apply the sorting to the root node
+ sortChildren(rootNode)
+
+ rootNode.getChildren
+ }
+
+ /**
+ * Traverses a given list of LakeFSFileNode and returns the total size of all files.
+ *
+ * @param nodes List of root-level LakeFSFileNode.
+ * @return Total size in bytes.
+ */
+ def calculateTotalSize(nodes: List[LakeFSFileNode]): Long = {
+ def traverse(node: LakeFSFileNode): Long = {
+ val fileSize = node.getSize.getOrElse(0L)
+ val childrenSize = node.getChildren.map(traverse).sum
+ fileSize + childrenSize
+ }
+
+ nodes.map(traverse).sum
+ }
+}
diff --git a/file-service/src/main/scala/org/apache/texera/service/type/dataset/DatasetFileNode.scala b/file-service/src/main/scala/org/apache/texera/service/type/dataset/DatasetFileNode.scala
deleted file mode 100644
index 85709b8e5b..0000000000
--- a/file-service/src/main/scala/org/apache/texera/service/type/dataset/DatasetFileNode.scala
+++ /dev/null
@@ -1,170 +0,0 @@
-/*
- * 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.
- */
-
-package org.apache.texera.service.`type`
-
-import io.lakefs.clients.sdk.model.ObjectStats
-import org.apache.texera.amber.core.storage.ResourceType
-
-import scala.collection.mutable
-
-// DatasetFileNode represents a unique file in dataset, its full path is in the format of:
-// /dataset/ownerEmail/datasetName/versionName/fileRelativePath
-// e.g. /dataset/bob@texera.com/twitterDataset/v1/california/irvine/tw1.csv
-class DatasetFileNode(
- val name: String, // direct name of this node
- val nodeType: String, // "file" or "directory"
- val parent: DatasetFileNode, // the parent node
- val ownerEmail: String,
- val size: Option[Long] = None, // size of the file in bytes, None if directory
- var children: Option[List[DatasetFileNode]] = None // Only populated if 'type' is 'directory'
-) {
-
- // Ensure the type is either "file" or "directory"
- require(nodeType == "file" || nodeType == "directory", "type must be 'file' or 'directory'")
-
- // Getters for the parameters
- def getName: String = name
-
- def getNodeType: String = nodeType
-
- def getParent: DatasetFileNode = parent
-
- def getOwnerEmail: String = ownerEmail
-
- def getSize: Option[Long] = size
-
- def getChildren: List[DatasetFileNode] = children.getOrElse(List())
-
- // Method to get the full file path
- def getFilePath: String = {
- val pathComponents = new mutable.ArrayBuffer[String]()
- var currentNode: DatasetFileNode = this
- while (currentNode != null) {
- if (currentNode.parent != null) { // Skip the root node to avoid double slashes
- pathComponents.prepend(currentNode.name)
- }
- currentNode = currentNode.parent
- }
- "/" + pathComponents.mkString("/")
- }
-}
-
-object DatasetFileNode {
-
- /**
- * Converts a map of LakeFS committed objects into a structured dataset file node tree.
- *
- * @param map A mapping from `(ownerEmail, datasetName, versionName)` to a list of committed objects.
- * @return A list of root-level dataset file nodes.
- */
- def fromLakeFSRepositoryCommittedObjects(
- map: Map[(String, String, String), List[ObjectStats]]
- ): List[DatasetFileNode] = {
- val rootNode = new DatasetFileNode("/", "directory", null, "")
-
- // Root the tree at the dataset prefix node (a directory node named "dataset").
- val prefixNode =
- new DatasetFileNode(ResourceType.Dataset.toString, "directory", rootNode, "")
- rootNode.children = Some(List(prefixNode))
-
- // Owner level nodes map
- val ownerNodes = mutable.Map[String, DatasetFileNode]()
-
- map.foreach {
- case ((ownerEmail, datasetName, versionName), objects) =>
- val ownerNode = ownerNodes.getOrElseUpdate(
- ownerEmail, {
- val newNode = new DatasetFileNode(ownerEmail, "directory", prefixNode, ownerEmail)
- prefixNode.children = Some(prefixNode.getChildren :+ newNode)
- newNode
- }
- )
-
- val datasetNode = ownerNode.getChildren.find(_.getName == datasetName).getOrElse {
- val newNode = new DatasetFileNode(datasetName, "directory", ownerNode, ownerEmail)
- ownerNode.children = Some(ownerNode.getChildren :+ newNode)
- newNode
- }
-
- val versionNode = datasetNode.getChildren.find(_.getName == versionName).getOrElse {
- val newNode = new DatasetFileNode(versionName, "directory", datasetNode, ownerEmail)
- datasetNode.children = Some(datasetNode.getChildren :+ newNode)
- newNode
- }
-
- // Directory map for efficient lookups
- val directoryMap = mutable.Map[String, DatasetFileNode]()
- directoryMap("") = versionNode // Root of the dataset version
-
- // Process each object (file or directory) from LakeFS
- objects.foreach { obj =>
- val pathParts = obj.getPath.split("/").toList
- var currentPath = ""
- var parentNode: DatasetFileNode = versionNode
-
- pathParts.foreach { part =>
- currentPath = if (currentPath.isEmpty) part else s"$currentPath/$part"
-
- val isFile = pathParts.last == part
- val nodeType = if (isFile) "file" else "directory"
- val fileSize = if (isFile) Some(obj.getSizeBytes.longValue()) else None
-
- val existingNode = directoryMap.get(currentPath)
-
- val node = existingNode.getOrElse {
- val newNode = new DatasetFileNode(part, nodeType, parentNode, ownerEmail, fileSize)
- parentNode.children = Some(parentNode.getChildren :+ newNode)
- if (!isFile) directoryMap(currentPath) = newNode
- newNode
- }
-
- parentNode = node // Move parent reference deeper for next iteration
- }
- }
- }
-
- // Sorting function to sort children of a node alphabetically in descending order
- def sortChildren(node: DatasetFileNode): Unit = {
- node.children = Some(node.getChildren.sortBy(_.getName)(Ordering.String.reverse))
- node.getChildren.foreach(sortChildren)
- }
-
- // Apply the sorting to the root node
- sortChildren(rootNode)
-
- rootNode.getChildren
- }
-
- /**
- * Traverses a given list of DatasetFileNode and returns the total size of all files.
- *
- * @param nodes List of root-level DatasetFileNode.
- * @return Total size in bytes.
- */
- def calculateTotalSize(nodes: List[DatasetFileNode]): Long = {
- def traverse(node: DatasetFileNode): Long = {
- val fileSize = node.getSize.getOrElse(0L)
- val childrenSize = node.getChildren.map(traverse).sum
- fileSize + childrenSize
- }
-
- nodes.map(traverse).sum
- }
-}
diff --git a/file-service/src/main/scala/org/apache/texera/service/type/serde/DatasetFileNodeSerializer.java b/file-service/src/main/scala/org/apache/texera/service/type/serde/LakeFSFileNodeSerializer.java
similarity index 80%
rename from file-service/src/main/scala/org/apache/texera/service/type/serde/DatasetFileNodeSerializer.java
rename to file-service/src/main/scala/org/apache/texera/service/type/serde/LakeFSFileNodeSerializer.java
index 70b24270ea..f253df6b56 100644
--- a/file-service/src/main/scala/org/apache/texera/service/type/serde/DatasetFileNodeSerializer.java
+++ b/file-service/src/main/scala/org/apache/texera/service/type/serde/LakeFSFileNodeSerializer.java
@@ -22,25 +22,25 @@ package org.apache.texera.service.type.serde;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.SerializerProvider;
import com.fasterxml.jackson.databind.ser.std.StdSerializer;
-import org.apache.texera.service.type.DatasetFileNode;
+import org.apache.texera.service.type.LakeFSFileNode;
import scala.collection.immutable.List;
import scala.jdk.javaapi.CollectionConverters;
import java.io.IOException;
// this class is used to serialize the FileNode as JSON. So that FileNodes can be inspected by the frontend through JSON.
-public class DatasetFileNodeSerializer extends StdSerializer<DatasetFileNode> {
+public class LakeFSFileNodeSerializer extends StdSerializer<LakeFSFileNode> {
- public DatasetFileNodeSerializer() {
+ public LakeFSFileNodeSerializer() {
this(null);
}
- public DatasetFileNodeSerializer(Class<DatasetFileNode> t) {
+ public LakeFSFileNodeSerializer(Class<LakeFSFileNode> t) {
super(t);
}
@Override
- public void serialize(DatasetFileNode value, JsonGenerator gen, SerializerProvider provider) throws IOException {
+ public void serialize(LakeFSFileNode value, JsonGenerator gen, SerializerProvider provider) throws IOException {
gen.writeStartObject();
gen.writeStringField("name", value.getName());
gen.writeStringField("type", value.getNodeType());
@@ -52,8 +52,8 @@ public class DatasetFileNodeSerializer extends StdSerializer<DatasetFileNode> {
if (value.getNodeType().equals("directory")) {
gen.writeFieldName("children");
gen.writeStartArray();
- List<DatasetFileNode> children = value.getChildren();
- for (DatasetFileNode child : CollectionConverters.asJava(children)) {
+ List<LakeFSFileNode> children = value.getChildren();
+ for (LakeFSFileNode child : CollectionConverters.asJava(children)) {
serialize(child, gen, provider); // Recursively serialize children
}
gen.writeEndArray();
diff --git a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala
index 588971d8a7..0dfb39909a 100644
--- a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala
+++ b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala
@@ -3846,4 +3846,37 @@ class DatasetResourceSpec
response.fileNodes.map(_.getName).toSet shouldEqual Set("a.txt", "sub")
}
+
+ // Guards the resource-type prefix the file tree is rooted at. `FileResolver` dispatches on
+ // that leading segment to choose the backing table, so a dataset tree must stay under
+ // `/dataset/...` now that the builder takes the resource type as a parameter.
+ "retrieveDatasetVersionRootFileNodes" should "root a dataset version's tree at the dataset prefix" in {
+ val version = testDatasetVersion
+
+ val response = datasetResource.retrieveDatasetVersionRootFileNodes(
+ baseDataset.getDid,
+ version.getDvid,
+ sessionUser
+ )
+
+ response.fileNodes should not be empty
+ response.size should be > 0L
+
+ val file = response.fileNodes.find(_.getName == "test-cover.jpg").get
+ file.getFilePath shouldBe
+ s"/dataset/${ownerUser.getEmail}/${baseDataset.getName}/${version.getName}/test-cover.jpg"
+ }
+
+ "retrieveLatestDatasetVersion" should "return the latest version rooted at the dataset prefix" in {
+ testDatasetVersion
+
+ val latest = datasetResource.retrieveLatestDatasetVersion(baseDataset.getDid, sessionUser)
+
+ latest.datasetVersion.getName shouldBe "v1"
+ latest.fileNodes.map(_.getName) should contain("test-cover.jpg")
+ latest.fileNodes
+ .find(_.getName == "test-cover.jpg")
+ .get
+ .getFilePath should startWith("/dataset/")
+ }
}
diff --git a/file-service/src/test/scala/org/apache/texera/service/type/LakeFSFileNodeSpec.scala b/file-service/src/test/scala/org/apache/texera/service/type/LakeFSFileNodeSpec.scala
new file mode 100644
index 0000000000..5ccb21d322
--- /dev/null
+++ b/file-service/src/test/scala/org/apache/texera/service/type/LakeFSFileNodeSpec.scala
@@ -0,0 +1,204 @@
+/*
+ * 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.
+ */
+
+package org.apache.texera.service.`type`
+
+import io.lakefs.clients.sdk.model.ObjectStats
+import org.apache.texera.amber.core.storage.ResourceType
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+// Unit tests for LakeFSFileNode: the instance helpers getFilePath, the
+// constructor's nodeType require guard, and the companion helpers
+// calculateTotalSize / fromLakeFSRepositoryCommittedObjects. The LakeFS
+// factory is fixtured with lightweight ObjectStats POJOs.
+class LakeFSFileNodeSpec extends AnyFlatSpec with Matchers {
+
+ // -- constructor require guard ----------------------------------------------
+
+ "LakeFSFileNode constructor" should "accept nodeType 'file'" in {
+ val root = new LakeFSFileNode("/", "directory", null, "")
+ val node = new LakeFSFileNode("a.csv", "file", root, "alice", Some(1L))
+ node.getNodeType shouldBe "file"
+ }
+
+ it should "accept nodeType 'directory'" in {
+ val node = new LakeFSFileNode("/", "directory", null, "")
+ node.getNodeType shouldBe "directory"
+ }
+
+ it should "reject any other nodeType" in {
+ val ex = intercept[IllegalArgumentException] {
+ new LakeFSFileNode("weird", "symlink", null, "")
+ }
+ ex.getMessage should include("type must be 'file' or 'directory'")
+ }
+
+ // -- getFilePath ------------------------------------------------------------
+
+ "getFilePath" should "return just '/' for the root node" in {
+ val root = new LakeFSFileNode("/", "directory", null, "")
+ root.getFilePath shouldBe "/"
+ }
+
+ it should "walk parents to build an absolute path and skip the root" in {
+ val root = new LakeFSFileNode("/", "directory", null, "")
+ val a = new LakeFSFileNode("a", "directory", root, "owner")
+ val b = new LakeFSFileNode("b", "directory", a, "owner")
+ val c = new LakeFSFileNode("c.csv", "file", b, "owner", Some(10L))
+
+ a.getFilePath shouldBe "/a"
+ b.getFilePath shouldBe "/a/b"
+ c.getFilePath shouldBe "/a/b/c.csv"
+ }
+
+ // -- calculateTotalSize -----------------------------------------------------
+
+ "calculateTotalSize" should "return 0 for an empty list" in {
+ LakeFSFileNode.calculateTotalSize(List.empty) shouldBe 0L
+ }
+
+ it should "sum file sizes recursively across the tree, ignoring directories" in {
+ val root = new LakeFSFileNode("/", "directory", null, "")
+ val f1 = new LakeFSFileNode("f1", "file", root, "owner", Some(100L))
+ val dir = new LakeFSFileNode("dir", "directory", root, "owner")
+ val f2 = new LakeFSFileNode("f2", "file", dir, "owner", Some(50L))
+ val f3 = new LakeFSFileNode("f3", "file", dir, "owner", Some(25L))
+ dir.children = Some(List(f2, f3))
+
+ LakeFSFileNode.calculateTotalSize(List(f1, dir)) shouldBe 175L
+ }
+
+ // -- fromLakeFSRepositoryCommittedObjects -----------------------------------
+
+ private def objStats(path: String, size: Long): ObjectStats =
+ new ObjectStats().path(path).sizeBytes(size)
+
+ "fromLakeFSRepositoryCommittedObjects" should "build a sorted owner/dataset/version tree" in {
+ val objects = List(
+ objStats("a/x.csv", 1L),
+ objStats("b/1.csv", 2L),
+ objStats("b/2.csv", 3L)
+ )
+ val roots = LakeFSFileNode.fromLakeFSRepositoryCommittedObjects(
+ ResourceType.Dataset,
+ Map(("bob@texera.com", "twitter", "v1") -> objects)
+ )
+
+ // The tree is rooted at a single "dataset" prefix node; owners nest under it.
+ roots should have size 1
+ val prefixNode = roots.head
+ prefixNode.getName shouldBe "dataset"
+ prefixNode.getNodeType shouldBe "directory"
+
+ val ownerNode = prefixNode.getChildren.find(_.getName == "bob@texera.com").get
+ ownerNode.getNodeType shouldBe "directory"
+
+ val datasetNode = ownerNode.getChildren.find(_.getName == "twitter").get
+ val versionNode = datasetNode.getChildren.find(_.getName == "v1").get
+
+ // Top-level dirs under the version are sorted by name descending: b before a.
+ versionNode.getChildren.map(_.getName) shouldBe List("b", "a")
+
+ // Directory "b" is created once (dedup) and holds both files, sorted descending.
+ val bDir = versionNode.getChildren.find(_.getName == "b").get
+ bDir.getNodeType shouldBe "directory"
+ bDir.getChildren.map(_.getName) shouldBe List("2.csv", "1.csv")
+
+ // Leaf sizes and full paths are wired up correctly.
+ val file1 = bDir.getChildren.find(_.getName == "1.csv").get
+ file1.getNodeType shouldBe "file"
+ file1.getSize shouldBe Some(2L)
+ file1.getFilePath shouldBe "/dataset/bob@texera.com/twitter/v1/b/1.csv"
+
+ // Total size equals the sum of the three files.
+ LakeFSFileNode.calculateTotalSize(roots) shouldBe 6L
+ }
+
+ it should "root a model tree at the model prefix, not dataset" in {
+ // The prefix is not cosmetic: FileResolver keys on the leading path segment to
+ // choose the backing table, so a model file rooted at "dataset" would resolve
+ // against the dataset table.
+ val roots = LakeFSFileNode.fromLakeFSRepositoryCommittedObjects(
+ ResourceType.Model,
+ Map(("bob@texera.com", "sentiment", "v1") -> List(objStats("model.pt", 4L)))
+ )
+
+ roots should have size 1
+ roots.head.getName shouldBe "model"
+
+ val file = roots.head.getChildren.head.getChildren.head.getChildren.head.getChildren.head
+ file.getFilePath shouldBe "/model/bob@texera.com/sentiment/v1/model.pt"
+ }
+
+ it should "treat a repeated final path segment as one directory and one file" in {
+ // Deciding file-vs-directory by value made the intermediate "model" the leaf, so the
+ // size was double-counted and a "file" node carried children.
+ val roots = LakeFSFileNode.fromLakeFSRepositoryCommittedObjects(
+ ResourceType.Model,
+ Map(("bob@texera.com", "sentiment", "v1") -> List(objStats("model/model", 7L)))
+ )
+
+ val versionNode =
+ roots.head.getChildren.head.getChildren.head.getChildren.head
+ val dir = versionNode.getChildren.find(_.getName == "model").get
+ dir.getNodeType shouldBe "directory"
+ dir.getSize shouldBe None
+
+ val leaf = dir.getChildren.find(_.getName == "model").get
+ leaf.getNodeType shouldBe "file"
+ leaf.getSize shouldBe Some(7L)
+ leaf.getFilePath shouldBe "/model/bob@texera.com/sentiment/v1/model/model"
+
+ // counted once, not twice
+ LakeFSFileNode.calculateTotalSize(roots) shouldBe 7L
+ }
+
+ it should "keep every object when a name is both an object and a directory prefix" in {
+ // Both keys can exist in one version and a tree cannot name both, so they surface as
+ // two siblings called "model". Untidy, but nothing is dropped.
+ val roots = LakeFSFileNode.fromLakeFSRepositoryCommittedObjects(
+ ResourceType.Model,
+ Map(
+ ("bob@texera.com", "sentiment", "v1") -> List(
+ objStats("model", 3L),
+ objStats("model/weights.bin", 5L)
+ )
+ )
+ )
+
+ val versionNode = roots.head.getChildren.head.getChildren.head.getChildren.head
+ val named = versionNode.getChildren.filter(_.getName == "model")
+
+ // One stands for the object, the other holds the deeper object.
+ named.map(_.getNodeType) should contain allOf ("file", "directory")
+
+ val asFile = named.find(_.getNodeType == "file").get
+ asFile.getSize shouldBe Some(3L)
+ asFile.getFilePath shouldBe "/model/bob@texera.com/sentiment/v1/model"
+
+ val asDir = named.find(_.getNodeType == "directory").get
+ asDir.getChildren.map(_.getName) shouldBe List("weights.bin")
+ asDir.getChildren.head.getFilePath shouldBe
+ "/model/bob@texera.com/sentiment/v1/model/weights.bin"
+
+ // Both objects are counted: 3 + 5.
+ LakeFSFileNode.calculateTotalSize(roots) shouldBe 8L
+ }
+}
diff --git a/file-service/src/test/scala/org/apache/texera/service/type/dataset/DatasetFileNodeSpec.scala b/file-service/src/test/scala/org/apache/texera/service/type/dataset/DatasetFileNodeSpec.scala
deleted file mode 100644
index 12321d4180..0000000000
--- a/file-service/src/test/scala/org/apache/texera/service/type/dataset/DatasetFileNodeSpec.scala
+++ /dev/null
@@ -1,131 +0,0 @@
-/*
- * 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.
- */
-
-package org.apache.texera.service.`type`
-
-import io.lakefs.clients.sdk.model.ObjectStats
-import org.scalatest.flatspec.AnyFlatSpec
-import org.scalatest.matchers.should.Matchers
-
-// Unit tests for DatasetFileNode: the instance helpers getFilePath, the
-// constructor's nodeType require guard, and the companion helpers
-// calculateTotalSize / fromLakeFSRepositoryCommittedObjects. The LakeFS
-// factory is fixtured with lightweight ObjectStats POJOs.
-class DatasetFileNodeSpec extends AnyFlatSpec with Matchers {
-
- // -- constructor require guard ----------------------------------------------
-
- "DatasetFileNode constructor" should "accept nodeType 'file'" in {
- val root = new DatasetFileNode("/", "directory", null, "")
- val node = new DatasetFileNode("a.csv", "file", root, "alice", Some(1L))
- node.getNodeType shouldBe "file"
- }
-
- it should "accept nodeType 'directory'" in {
- val node = new DatasetFileNode("/", "directory", null, "")
- node.getNodeType shouldBe "directory"
- }
-
- it should "reject any other nodeType" in {
- val ex = intercept[IllegalArgumentException] {
- new DatasetFileNode("weird", "symlink", null, "")
- }
- ex.getMessage should include("type must be 'file' or 'directory'")
- }
-
- // -- getFilePath ------------------------------------------------------------
-
- "getFilePath" should "return just '/' for the root node" in {
- val root = new DatasetFileNode("/", "directory", null, "")
- root.getFilePath shouldBe "/"
- }
-
- it should "walk parents to build an absolute path and skip the root" in {
- val root = new DatasetFileNode("/", "directory", null, "")
- val a = new DatasetFileNode("a", "directory", root, "owner")
- val b = new DatasetFileNode("b", "directory", a, "owner")
- val c = new DatasetFileNode("c.csv", "file", b, "owner", Some(10L))
-
- a.getFilePath shouldBe "/a"
- b.getFilePath shouldBe "/a/b"
- c.getFilePath shouldBe "/a/b/c.csv"
- }
-
- // -- calculateTotalSize -----------------------------------------------------
-
- "calculateTotalSize" should "return 0 for an empty list" in {
- DatasetFileNode.calculateTotalSize(List.empty) shouldBe 0L
- }
-
- it should "sum file sizes recursively across the tree, ignoring directories" in {
- val root = new DatasetFileNode("/", "directory", null, "")
- val f1 = new DatasetFileNode("f1", "file", root, "owner", Some(100L))
- val dir = new DatasetFileNode("dir", "directory", root, "owner")
- val f2 = new DatasetFileNode("f2", "file", dir, "owner", Some(50L))
- val f3 = new DatasetFileNode("f3", "file", dir, "owner", Some(25L))
- dir.children = Some(List(f2, f3))
-
- DatasetFileNode.calculateTotalSize(List(f1, dir)) shouldBe 175L
- }
-
- // -- fromLakeFSRepositoryCommittedObjects -----------------------------------
-
- private def objStats(path: String, size: Long): ObjectStats =
- new ObjectStats().path(path).sizeBytes(size)
-
- "fromLakeFSRepositoryCommittedObjects" should "build a sorted owner/dataset/version tree" in {
- val objects = List(
- objStats("a/x.csv", 1L),
- objStats("b/1.csv", 2L),
- objStats("b/2.csv", 3L)
- )
- val roots = DatasetFileNode.fromLakeFSRepositoryCommittedObjects(
- Map(("bob@texera.com", "twitter", "v1") -> objects)
- )
-
- // The tree is rooted at a single "dataset" prefix node; owners nest under it.
- roots should have size 1
- val prefixNode = roots.head
- prefixNode.getName shouldBe "dataset"
- prefixNode.getNodeType shouldBe "directory"
-
- val ownerNode = prefixNode.getChildren.find(_.getName == "bob@texera.com").get
- ownerNode.getNodeType shouldBe "directory"
-
- val datasetNode = ownerNode.getChildren.find(_.getName == "twitter").get
- val versionNode = datasetNode.getChildren.find(_.getName == "v1").get
-
- // Top-level dirs under the version are sorted by name descending: b before a.
- versionNode.getChildren.map(_.getName) shouldBe List("b", "a")
-
- // Directory "b" is created once (dedup) and holds both files, sorted descending.
- val bDir = versionNode.getChildren.find(_.getName == "b").get
- bDir.getNodeType shouldBe "directory"
- bDir.getChildren.map(_.getName) shouldBe List("2.csv", "1.csv")
-
- // Leaf sizes and full paths are wired up correctly.
- val file1 = bDir.getChildren.find(_.getName == "1.csv").get
- file1.getNodeType shouldBe "file"
- file1.getSize shouldBe Some(2L)
- file1.getFilePath shouldBe "/dataset/bob@texera.com/twitter/v1/b/1.csv"
-
- // Total size equals the sum of the three files.
- DatasetFileNode.calculateTotalSize(roots) shouldBe 6L
- }
-}
diff --git a/file-service/src/test/scala/org/apache/texera/service/type/serde/DatasetFileNodeSerializerSpec.scala b/file-service/src/test/scala/org/apache/texera/service/type/serde/LakeFSFileNodeSerializerSpec.scala
similarity index 73%
rename from file-service/src/test/scala/org/apache/texera/service/type/serde/DatasetFileNodeSerializerSpec.scala
rename to file-service/src/test/scala/org/apache/texera/service/type/serde/LakeFSFileNodeSerializerSpec.scala
index 4a74ad0143..31c8b12ea8 100644
--- a/file-service/src/test/scala/org/apache/texera/service/type/serde/DatasetFileNodeSerializerSpec.scala
+++ b/file-service/src/test/scala/org/apache/texera/service/type/serde/LakeFSFileNodeSerializerSpec.scala
@@ -22,35 +22,35 @@ package org.apache.texera.service.`type`.serde
import com.fasterxml.jackson.databind.module.SimpleModule
import com.fasterxml.jackson.databind.{JsonNode, ObjectMapper}
import com.fasterxml.jackson.module.scala.DefaultScalaModule
-import org.apache.texera.service.`type`.DatasetFileNode
+import org.apache.texera.service.`type`.LakeFSFileNode
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
-class DatasetFileNodeSerializerSpec extends AnyFlatSpec with Matchers {
+class LakeFSFileNodeSerializerSpec extends AnyFlatSpec with Matchers {
private val mapper: ObjectMapper = {
val m = new ObjectMapper()
// DefaultScalaModule lets Jackson unwrap scala.Option for the "size" field.
m.registerModule(DefaultScalaModule)
val module = new SimpleModule()
- module.addSerializer(classOf[DatasetFileNode], new DatasetFileNodeSerializer())
+ module.addSerializer(classOf[LakeFSFileNode], new LakeFSFileNodeSerializer())
m.registerModule(module)
m
}
- private def asJson(node: DatasetFileNode): JsonNode =
+ private def asJson(node: LakeFSFileNode): JsonNode =
mapper.readTree(mapper.writeValueAsString(node))
// The serializer dereferences value.getParent().getFilePath(), so every node it
// sees needs a non-null parent. Tests build a tree rooted at "/" and serialize
// its descendants.
- private def rootDir: DatasetFileNode =
- new DatasetFileNode("/", "directory", null, "")
+ private def rootDir: LakeFSFileNode =
+ new LakeFSFileNode("/", "directory", null, "")
- "DatasetFileNodeSerializer" should "serialize a file node with size and no children field" in {
+ "LakeFSFileNodeSerializer" should "serialize a file node with size and no children field" in {
val root = rootDir
- val owner = new DatasetFileNode("alice@example.com", "directory", root, "alice@example.com")
- val file = new DatasetFileNode("data.csv", "file", owner, "alice@example.com", Some(100L))
+ val owner = new LakeFSFileNode("alice@example.com", "directory", root, "alice@example.com")
+ val file = new LakeFSFileNode("data.csv", "file", owner, "alice@example.com", Some(100L))
val json = asJson(file)
@@ -64,10 +64,10 @@ class DatasetFileNodeSerializerSpec extends AnyFlatSpec with Matchers {
it should "recursively serialize a directory and its children" in {
val root = rootDir
- val owner = new DatasetFileNode("alice@example.com", "directory", root, "alice@example.com")
- val file = new DatasetFileNode("data.csv", "file", owner, "alice@example.com", Some(100L))
- val subdir = new DatasetFileNode("subdir", "directory", owner, "alice@example.com")
- val nested = new DatasetFileNode("nested.txt", "file", subdir, "alice@example.com", Some(200L))
+ val owner = new LakeFSFileNode("alice@example.com", "directory", root, "alice@example.com")
+ val file = new LakeFSFileNode("data.csv", "file", owner, "alice@example.com", Some(100L))
+ val subdir = new LakeFSFileNode("subdir", "directory", owner, "alice@example.com")
+ val nested = new LakeFSFileNode("nested.txt", "file", subdir, "alice@example.com", Some(200L))
subdir.children = Some(List(nested))
owner.children = Some(List(file, subdir))
@@ -88,7 +88,7 @@ class DatasetFileNodeSerializerSpec extends AnyFlatSpec with Matchers {
it should "emit an empty children array for a directory with no children" in {
val root = rootDir
- val empty = new DatasetFileNode("empty", "directory", root, "alice@example.com")
+ val empty = new LakeFSFileNode("empty", "directory", root, "alice@example.com")
val json = asJson(empty)