adutra commented on code in PR #8857:
URL: https://github.com/apache/iceberg/pull/8857#discussion_r1369932165
##########
nessie/src/main/java/org/apache/iceberg/nessie/NessieIcebergClient.java:
##########
@@ -181,133 +185,223 @@ public IcebergTable table(TableIdentifier
tableIdentifier) {
}
public void createNamespace(Namespace namespace, Map<String, String>
metadata) {
+ getRef().checkMutable();
+
+ if (namespace.isEmpty()) {
+ throw new IllegalArgumentException("Creating empty namespaces is not
supported");
+ }
+
+ ContentKey key = ContentKey.of(namespace.levels());
+ org.projectnessie.model.Namespace content =
+ org.projectnessie.model.Namespace.of(key.getElements(), metadata);
+
try {
- getRef().checkMutable();
- withReference(
- getApi()
- .createNamespace()
-
.namespace(org.projectnessie.model.Namespace.of(namespace.levels()))
- .properties(metadata))
- .create();
- refresh();
- } catch (NessieNamespaceAlreadyExistsException e) {
- throw new AlreadyExistsException(e, "Namespace already exists: %s",
namespace);
+
+ Map<ContentKey, Content> contentMap =
+ api.getContent().reference(getReference()).key(key).get();
+ Content existing = contentMap.get(key);
+ if (existing != null) {
+ throw namespaceAlreadyExists(key, existing, null);
+ }
+
+ try {
+
+ commitRetry("create namespace " + key, Operation.Put.of(key, content));
+
+ } catch (NessieReferenceConflictException e) {
+
+ NessieConflictHandler.handleSingle(
+ e,
+ (conflictType, contentKey) -> {
+ switch (conflictType) {
+ case KEY_EXISTS:
+ Content conflicting =
+
withReference(api.getContent()).key(contentKey).get().get(contentKey);
+ throw namespaceAlreadyExists(contentKey, conflicting, e);
+ case NAMESPACE_ABSENT:
+ throw new NoSuchNamespaceException(
+ e,
+ "Cannot create Namespace '%s': parent namespace '%s'
does not exist",
+ namespace,
+ contentKey);
+ default:
+ return false;
+ }
+ });
+ throw new RuntimeException(
+ String.format("Cannot create Namespace '%s': %s", namespace,
e.getMessage()));
+ }
+
} catch (NessieNotFoundException e) {
throw new RuntimeException(
String.format(
- "Cannot create Namespace '%s': " + "ref '%s' is no longer
valid.",
+ "Cannot create Namespace '%s': ref '%s' is no longer valid.",
namespace, getRef().getName()),
e);
+ } catch (BaseNessieClientServerException e) {
+ throw new RuntimeException(
+ String.format("Cannot create Namespace '%s': %s", namespace,
e.getMessage()), e);
}
}
public List<Namespace> listNamespaces(Namespace namespace) throws
NoSuchNamespaceException {
+
try {
- GetNamespacesResponse response =
- withReference(
- getApi()
- .getMultipleNamespaces()
-
.namespace(org.projectnessie.model.Namespace.of(namespace.levels())))
- .get();
- return response.getNamespaces().stream()
- .map(ns -> Namespace.of(ns.getElements().toArray(new String[0])))
- .filter(ns -> ns.length() == namespace.length() + 1)
+
+ org.projectnessie.model.Namespace root =
+ org.projectnessie.model.Namespace.of(namespace.levels());
+
+ String filter =
+ namespace.isEmpty()
+ ? "size(entry.keyElements) == 1"
+ : String.format(
+ "size(entry.keyElements) == %d &&
entry.encodedKey.startsWith('%s.')",
+ root.getElementCount() + 1, root.name());
+
+ List<ContentKey> entries =
+ withReference(api.getEntries()).filter(filter).stream()
+ .filter(e -> Content.Type.NAMESPACE.equals(e.getType()))
+ .map(EntriesResponse.Entry::getName)
+ .collect(Collectors.toList());
+
+ if (entries.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ GetContentBuilder getContent = withReference(api.getContent());
+ entries.forEach(getContent::key);
+
+ return getContent.get().values().stream()
+ .map(v -> v.unwrap(org.projectnessie.model.Namespace.class))
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .map(v -> Namespace.of(v.getElements().toArray(new String[0])))
+ .filter(v -> v.length() == namespace.length() + 1) // only direct
children
.collect(Collectors.toList());
- } catch (NessieReferenceNotFoundException e) {
+
+ } catch (NessieNotFoundException e) {
throw new RuntimeException(
String.format(
- "Cannot list Namespaces starting from '%s': " + "ref '%s' is no
longer valid.",
+ "Cannot list Namespaces starting from '%s': ref '%s' is no
longer valid.",
namespace, getRef().getName()),
e);
}
}
public boolean dropNamespace(Namespace namespace) throws
NamespaceNotEmptyException {
+
+ getRef().checkMutable();
+ ContentKey key = ContentKey.of(namespace.levels());
+
try {
- getRef().checkMutable();
- withReference(
- getApi()
- .deleteNamespace()
-
.namespace(org.projectnessie.model.Namespace.of(namespace.levels())))
- .delete();
- refresh();
+
+ commitRetry("drop namespace " + key, Operation.Delete.of(key));
return true;
- } catch (NessieNamespaceNotFoundException e) {
- return false;
+
+ } catch (NessieReferenceConflictException e) {
+
+ if (!NessieConflictHandler.handleSingle(
+ e,
+ (conflictType, contentKey) -> {
+ switch (conflictType) {
+ case KEY_DOES_NOT_EXIST:
+ return true; // OK, continue
+ case NAMESPACE_NOT_EMPTY:
+ throw new NamespaceNotEmptyException(e, "Namespace '%s' is not
empty.", namespace);
+ }
+ return false;
+ })) {
+ throw new RuntimeException(
+ String.format("Cannot drop Namespace '%s': %s", namespace,
e.getMessage()));
+ }
+
} catch (NessieNotFoundException e) {
LOG.error(
"Cannot drop Namespace '{}': ref '{}' is no longer valid.",
namespace,
getRef().getName(),
e);
- return false;
- } catch (NessieNamespaceNotEmptyException e) {
- throw new NamespaceNotEmptyException(
- e, "Namespace '%s' is not empty. One or more tables exist.",
namespace);
+ } catch (BaseNessieClientServerException e) {
+ throw new RuntimeException(
+ String.format("Cannot drop Namespace '%s': %s", namespace,
e.getMessage()), e);
}
+ return false;
}
public Map<String, String> loadNamespaceMetadata(Namespace namespace)
throws NoSuchNamespaceException {
+ ContentKey key = ContentKey.of(namespace.levels());
try {
- return withReference(
- getApi()
- .getNamespace()
-
.namespace(org.projectnessie.model.Namespace.of(namespace.levels())))
- .get()
- .getProperties();
- } catch (NessieNamespaceNotFoundException e) {
- throw new NoSuchNamespaceException(e, "Namespace does not exist: %s",
namespace);
- } catch (NessieReferenceNotFoundException e) {
+ Map<ContentKey, Content> contentMap =
withReference(api.getContent()).key(key).get();
+ Content existing = contentMap.get(key);
+ if (!(existing instanceof org.projectnessie.model.Namespace)) {
+ throw new NoSuchNamespaceException("Namespace does not exist: %s",
namespace);
+ }
+ return ((org.projectnessie.model.Namespace) existing).getProperties();
+ } catch (NessieNotFoundException e) {
throw new RuntimeException(
String.format(
- "Cannot load Namespace '%s': " + "ref '%s' is no longer valid.",
+ "Cannot load Namespace '%s': ref '%s' is no longer valid.",
namespace, getRef().getName()),
e);
}
}
public boolean setProperties(Namespace namespace, Map<String, String>
properties) {
- try {
- withReference(
- getApi()
- .updateProperties()
-
.namespace(org.projectnessie.model.Namespace.of(namespace.levels()))
- .updateProperties(properties))
- .update();
- refresh();
- // always successful, otherwise an exception is thrown
- return true;
- } catch (NessieNamespaceNotFoundException e) {
- throw new NoSuchNamespaceException(e, "Namespace does not exist: %s",
namespace);
- } catch (NessieNotFoundException e) {
- throw new RuntimeException(
- String.format(
- "Cannot update properties on Namespace '%s': ref '%s' is no
longer valid.",
- namespace, getRef().getName()),
- e);
- }
+ return updateProperties(namespace, props -> props.putAll(properties));
}
public boolean removeProperties(Namespace namespace, Set<String> properties)
{
+ return updateProperties(namespace, props ->
props.keySet().removeAll(properties));
+ }
+
+ private boolean updateProperties(Namespace namespace, Consumer<Map<String,
String>> action) {
+ getRef().checkMutable();
+ ContentKey key = ContentKey.of(namespace.levels());
+
try {
- withReference(
- getApi()
- .updateProperties()
-
.namespace(org.projectnessie.model.Namespace.of(namespace.levels()))
- .removeProperties(properties))
- .update();
- refresh();
+
+ Map<ContentKey, Content> contentMap =
+ api.getContent().reference(getReference()).key(key).get();
Review Comment:
The code here is _updating_ the properties, not _loading_ them.
In `loadNamespaceMetadata` there is indeed a call to `withReference` that
will omit the current hash if the ref is mutable, thus resulting in loading the
latest data from the HEAD of that ref.
Here, however, I think it's important to update the content by specifying
the latest hash we know about, so that any conflict that might happen due to
more recent commits will surface as `NessieConflictException` – not
`NessieBadRequestException`.
##########
nessie/src/main/java/org/apache/iceberg/nessie/NessieIcebergClient.java:
##########
@@ -181,133 +185,223 @@ public IcebergTable table(TableIdentifier
tableIdentifier) {
}
public void createNamespace(Namespace namespace, Map<String, String>
metadata) {
+ getRef().checkMutable();
+
+ if (namespace.isEmpty()) {
+ throw new IllegalArgumentException("Creating empty namespaces is not
supported");
+ }
+
+ ContentKey key = ContentKey.of(namespace.levels());
+ org.projectnessie.model.Namespace content =
+ org.projectnessie.model.Namespace.of(key.getElements(), metadata);
+
try {
- getRef().checkMutable();
- withReference(
- getApi()
- .createNamespace()
-
.namespace(org.projectnessie.model.Namespace.of(namespace.levels()))
- .properties(metadata))
- .create();
- refresh();
- } catch (NessieNamespaceAlreadyExistsException e) {
- throw new AlreadyExistsException(e, "Namespace already exists: %s",
namespace);
+
+ Map<ContentKey, Content> contentMap =
+ api.getContent().reference(getReference()).key(key).get();
+ Content existing = contentMap.get(key);
+ if (existing != null) {
+ throw namespaceAlreadyExists(key, existing, null);
+ }
+
+ try {
+
+ commitRetry("create namespace " + key, Operation.Put.of(key, content));
+
+ } catch (NessieReferenceConflictException e) {
+
+ NessieConflictHandler.handleSingle(
+ e,
+ (conflictType, contentKey) -> {
+ switch (conflictType) {
+ case KEY_EXISTS:
+ Content conflicting =
+
withReference(api.getContent()).key(contentKey).get().get(contentKey);
+ throw namespaceAlreadyExists(contentKey, conflicting, e);
+ case NAMESPACE_ABSENT:
+ throw new NoSuchNamespaceException(
+ e,
+ "Cannot create Namespace '%s': parent namespace '%s'
does not exist",
+ namespace,
+ contentKey);
+ default:
+ return false;
+ }
+ });
+ throw new RuntimeException(
+ String.format("Cannot create Namespace '%s': %s", namespace,
e.getMessage()));
+ }
+
} catch (NessieNotFoundException e) {
throw new RuntimeException(
String.format(
- "Cannot create Namespace '%s': " + "ref '%s' is no longer
valid.",
+ "Cannot create Namespace '%s': ref '%s' is no longer valid.",
namespace, getRef().getName()),
e);
+ } catch (BaseNessieClientServerException e) {
+ throw new RuntimeException(
+ String.format("Cannot create Namespace '%s': %s", namespace,
e.getMessage()), e);
}
}
public List<Namespace> listNamespaces(Namespace namespace) throws
NoSuchNamespaceException {
+
try {
- GetNamespacesResponse response =
- withReference(
- getApi()
- .getMultipleNamespaces()
-
.namespace(org.projectnessie.model.Namespace.of(namespace.levels())))
- .get();
- return response.getNamespaces().stream()
- .map(ns -> Namespace.of(ns.getElements().toArray(new String[0])))
- .filter(ns -> ns.length() == namespace.length() + 1)
+
+ org.projectnessie.model.Namespace root =
+ org.projectnessie.model.Namespace.of(namespace.levels());
+
+ String filter =
+ namespace.isEmpty()
+ ? "size(entry.keyElements) == 1"
+ : String.format(
+ "size(entry.keyElements) == %d &&
entry.encodedKey.startsWith('%s.')",
+ root.getElementCount() + 1, root.name());
+
+ List<ContentKey> entries =
+ withReference(api.getEntries()).filter(filter).stream()
+ .filter(e -> Content.Type.NAMESPACE.equals(e.getType()))
+ .map(EntriesResponse.Entry::getName)
+ .collect(Collectors.toList());
+
+ if (entries.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ GetContentBuilder getContent = withReference(api.getContent());
+ entries.forEach(getContent::key);
+
+ return getContent.get().values().stream()
+ .map(v -> v.unwrap(org.projectnessie.model.Namespace.class))
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .map(v -> Namespace.of(v.getElements().toArray(new String[0])))
+ .filter(v -> v.length() == namespace.length() + 1) // only direct
children
.collect(Collectors.toList());
- } catch (NessieReferenceNotFoundException e) {
+
+ } catch (NessieNotFoundException e) {
throw new RuntimeException(
String.format(
- "Cannot list Namespaces starting from '%s': " + "ref '%s' is no
longer valid.",
+ "Cannot list Namespaces starting from '%s': ref '%s' is no
longer valid.",
namespace, getRef().getName()),
e);
}
}
public boolean dropNamespace(Namespace namespace) throws
NamespaceNotEmptyException {
+
+ getRef().checkMutable();
+ ContentKey key = ContentKey.of(namespace.levels());
+
try {
- getRef().checkMutable();
- withReference(
- getApi()
- .deleteNamespace()
-
.namespace(org.projectnessie.model.Namespace.of(namespace.levels())))
- .delete();
- refresh();
+
+ commitRetry("drop namespace " + key, Operation.Delete.of(key));
return true;
- } catch (NessieNamespaceNotFoundException e) {
- return false;
+
+ } catch (NessieReferenceConflictException e) {
+
+ if (!NessieConflictHandler.handleSingle(
+ e,
+ (conflictType, contentKey) -> {
+ switch (conflictType) {
+ case KEY_DOES_NOT_EXIST:
+ return true; // OK, continue
+ case NAMESPACE_NOT_EMPTY:
+ throw new NamespaceNotEmptyException(e, "Namespace '%s' is not
empty.", namespace);
+ }
+ return false;
+ })) {
+ throw new RuntimeException(
+ String.format("Cannot drop Namespace '%s': %s", namespace,
e.getMessage()));
+ }
+
} catch (NessieNotFoundException e) {
LOG.error(
"Cannot drop Namespace '{}': ref '{}' is no longer valid.",
namespace,
getRef().getName(),
e);
- return false;
- } catch (NessieNamespaceNotEmptyException e) {
- throw new NamespaceNotEmptyException(
- e, "Namespace '%s' is not empty. One or more tables exist.",
namespace);
+ } catch (BaseNessieClientServerException e) {
+ throw new RuntimeException(
+ String.format("Cannot drop Namespace '%s': %s", namespace,
e.getMessage()), e);
}
+ return false;
}
public Map<String, String> loadNamespaceMetadata(Namespace namespace)
throws NoSuchNamespaceException {
+ ContentKey key = ContentKey.of(namespace.levels());
try {
- return withReference(
- getApi()
- .getNamespace()
-
.namespace(org.projectnessie.model.Namespace.of(namespace.levels())))
- .get()
- .getProperties();
- } catch (NessieNamespaceNotFoundException e) {
- throw new NoSuchNamespaceException(e, "Namespace does not exist: %s",
namespace);
- } catch (NessieReferenceNotFoundException e) {
+ Map<ContentKey, Content> contentMap =
withReference(api.getContent()).key(key).get();
+ Content existing = contentMap.get(key);
+ if (!(existing instanceof org.projectnessie.model.Namespace)) {
+ throw new NoSuchNamespaceException("Namespace does not exist: %s",
namespace);
+ }
+ return ((org.projectnessie.model.Namespace) existing).getProperties();
+ } catch (NessieNotFoundException e) {
throw new RuntimeException(
String.format(
- "Cannot load Namespace '%s': " + "ref '%s' is no longer valid.",
+ "Cannot load Namespace '%s': ref '%s' is no longer valid.",
namespace, getRef().getName()),
e);
}
}
public boolean setProperties(Namespace namespace, Map<String, String>
properties) {
- try {
- withReference(
- getApi()
- .updateProperties()
-
.namespace(org.projectnessie.model.Namespace.of(namespace.levels()))
- .updateProperties(properties))
- .update();
- refresh();
- // always successful, otherwise an exception is thrown
- return true;
- } catch (NessieNamespaceNotFoundException e) {
- throw new NoSuchNamespaceException(e, "Namespace does not exist: %s",
namespace);
- } catch (NessieNotFoundException e) {
- throw new RuntimeException(
- String.format(
- "Cannot update properties on Namespace '%s': ref '%s' is no
longer valid.",
- namespace, getRef().getName()),
- e);
- }
+ return updateProperties(namespace, props -> props.putAll(properties));
}
public boolean removeProperties(Namespace namespace, Set<String> properties)
{
+ return updateProperties(namespace, props ->
props.keySet().removeAll(properties));
+ }
+
+ private boolean updateProperties(Namespace namespace, Consumer<Map<String,
String>> action) {
+ getRef().checkMutable();
+ ContentKey key = ContentKey.of(namespace.levels());
+
try {
- withReference(
- getApi()
- .updateProperties()
-
.namespace(org.projectnessie.model.Namespace.of(namespace.levels()))
- .removeProperties(properties))
- .update();
- refresh();
+
+ Map<ContentKey, Content> contentMap =
+ api.getContent().reference(getReference()).key(key).get();
Review Comment:
The code here is _updating_ the properties, not _loading_ them.
In `loadNamespaceMetadata` there is indeed a call to `withReference` that
will omit the current hash if the ref is mutable, thus resulting in loading the
latest data from the HEAD of that ref.
Here, however, I think it's important to update the content by specifying
the latest hash we know about, so that any conflict that might happen due to
more recent commits will surface as `NessieConflictException` – not
`NessieBadRequestException`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
