Skip to content

Commit f9ffe1f

Browse files
author
lijinglun
committed
feat: add hms list namespaces
1 parent 89d5236 commit f9ffe1f

19 files changed

Lines changed: 2615 additions & 5 deletions

File tree

‎java/checkstyle.xml‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -172,10 +172,5 @@
172172
<property name="illegalPkgs" value="org.apache.log4j" />
173173
<property name="illegalPkgs" value="org.apache.commons.lang" />
174174
</module>
175-
<!-- support structured logging -->
176-
<module name="RegexpSinglelineJava">
177-
<property name="format" value="org\.slf4j\.(Logger|LoggerFactory)" />
178-
<property name="message" value="Please use org.apache.spark.internal.(SparkLogger|SparkLoggerFactory) instead." />
179-
</module>
180175
</module>
181176
</module>

‎java/lance-namespace-core/src/main/java/com/lancedb/lance/namespace/ObjectIdentifier.java‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
/*
2+
* Licensed under the Apache License, Version 2.0 (the "License");
3+
* you may not use this file except in compliance with the License.
4+
* You may obtain a copy of the License at
5+
*
6+
* http://www.apache.org/licenses/LICENSE-2.0
7+
*
8+
* Unless required by applicable law or agreed to in writing, software
9+
* distributed under the License is distributed on an "AS IS" BASIS,
10+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11+
* See the License for the specific language governing permissions and
12+
* limitations under the License.
13+
*/
14+
package com.lancedb.lance.namespace;
15+
16+
import java.util.List;
17+
import java.util.stream.Collectors;
18+
19+
public class ObjectIdentifier {
20+
private String[] levels;
21+
22+
private ObjectIdentifier(String[] levels) {
23+
this.levels = levels;
24+
}
25+
26+
public static ObjectIdentifier of(List<String> levels) {
27+
List<String> normalizedLevels =
28+
levels.stream()
29+
.filter(level -> level != null && !level.isEmpty())
30+
.collect(Collectors.toList());
31+
return new ObjectIdentifier(normalizedLevels.toArray(new String[0]));
32+
}
33+
34+
public String level(int pos) {
35+
return levels[pos];
36+
}
37+
38+
public int size() {
39+
return levels == null ? 0 : levels.length;
40+
}
41+
42+
public boolean empty() {
43+
return size() == 0;
44+
}
45+
}

‎java/lance-namespace-hms/pom.xml‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
3+
<project xmlns="http://maven.apache.org/POM/4.0.0"
4+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
5+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
6+
<modelVersion>4.0.0</modelVersion>
7+
8+
<parent>
9+
<groupId>com.lancedb</groupId>
10+
<artifactId>lance-namespace-root</artifactId>
11+
<version>0.0.1</version>
12+
<relativePath>../pom.xml</relativePath>
13+
</parent>
14+
15+
16+
<artifactId>lance-namespace-hms</artifactId>
17+
<name>${project.artifactId}</name>
18+
<description>Lance Namespace HMS Impl</description>
19+
<packaging>jar</packaging>
20+
21+
<dependencies>
22+
<dependency>
23+
<groupId>com.lancedb</groupId>
24+
<artifactId>lance-namespace-core</artifactId>
25+
<version>${lance-namespace.version}</version>
26+
</dependency>
27+
<dependency>
28+
<groupId>org.apache.hive</groupId>
29+
<artifactId>hive-standalone-metastore</artifactId>
30+
<version>3.1.3</version>
31+
</dependency>
32+
<dependency>
33+
<groupId>org.apache.hive</groupId>
34+
<artifactId>hive-exec</artifactId>
35+
<version>3.1.3</version>
36+
<exclusions>
37+
<exclusion>
38+
<groupId>com.google.guava</groupId>
39+
<artifactId>guava</artifactId>
40+
</exclusion>
41+
<exclusion>
42+
<groupId>com.tdunning</groupId>
43+
<artifactId>json</artifactId>
44+
</exclusion>
45+
</exclusions>
46+
<classifier>core</classifier>
47+
</dependency>
48+
<dependency>
49+
<groupId>com.google.guava</groupId>
50+
<artifactId>guava</artifactId>
51+
<version>33.4.0-jre</version>
52+
</dependency>
53+
54+
<dependency>
55+
<groupId>org.apache.hive</groupId>
56+
<artifactId>hive-metastore</artifactId>
57+
<version>3.1.3</version>
58+
<scope>test</scope>
59+
</dependency>
60+
<dependency>
61+
<groupId>org.apache.hive</groupId>
62+
<artifactId>hive-service</artifactId>
63+
<version>3.1.3</version>
64+
<scope>test</scope>
65+
</dependency>
66+
<dependency>
67+
<groupId>org.apache.hive</groupId>
68+
<artifactId>hive-standalone-metastore</artifactId>
69+
<version>3.1.3</version>
70+
<type>test-jar</type>
71+
<scope>test</scope>
72+
</dependency>
73+
<dependency>
74+
<groupId>org.junit.jupiter</groupId>
75+
<artifactId>junit-jupiter</artifactId>
76+
<version>5.11.3</version>
77+
<scope>test</scope>
78+
</dependency>
79+
<dependency>
80+
<groupId>org.assertj</groupId>
81+
<artifactId>assertj-core</artifactId>
82+
<version>3.26.3</version>
83+
<scope>test</scope>
84+
</dependency>
85+
</dependencies>
86+
</project>

‎java/lance-namespace-hms/src/main/java/com/lancedb/lance/namespace/ClientPool.java‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
/*
2+
* Licensed under the Apache License, Version 2.0 (the "License");
3+
* you may not use this file except in compliance with the License.
4+
* You may obtain a copy of the License at
5+
*
6+
* http://www.apache.org/licenses/LICENSE-2.0
7+
*
8+
* Unless required by applicable law or agreed to in writing, software
9+
* distributed under the License is distributed on an "AS IS" BASIS,
10+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11+
* See the License for the specific language governing permissions and
12+
* limitations under the License.
13+
*/
14+
package com.lancedb.lance.namespace;
15+
16+
// Copied from apache iceberg.
17+
public interface ClientPool<C, E extends Exception> {
18+
interface Action<R, C, E extends Exception> {
19+
R run(C client) throws E;
20+
}
21+
22+
<R> R run(Action<R, C, E> action) throws E, InterruptedException;
23+
24+
<R> R run(Action<R, C, E> action, boolean retry) throws E, InterruptedException;
25+
}

‎java/lance-namespace-hms/src/main/java/com/lancedb/lance/namespace/ClientPoolImpl.java‎

Lines changed: 174 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,174 @@
1+
/*
2+
* Licensed under the Apache License, Version 2.0 (the "License");
3+
* you may not use this file except in compliance with the License.
4+
* You may obtain a copy of the License at
5+
*
6+
* http://www.apache.org/licenses/LICENSE-2.0
7+
*
8+
* Unless required by applicable law or agreed to in writing, software
9+
* distributed under the License is distributed on an "AS IS" BASIS,
10+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11+
* See the License for the specific language governing permissions and
12+
* limitations under the License.
13+
*/
14+
package com.lancedb.lance.namespace;
15+
16+
import com.google.common.annotations.VisibleForTesting;
17+
import com.google.common.base.Preconditions;
18+
import org.slf4j.Logger;
19+
import org.slf4j.LoggerFactory;
20+
21+
import java.io.Closeable;
22+
import java.util.ArrayDeque;
23+
import java.util.Deque;
24+
25+
// Copied from apache iceberg.
26+
public abstract class ClientPoolImpl<C, E extends Exception>
27+
implements Closeable, ClientPool<C, E> {
28+
private static final Logger LOG = LoggerFactory.getLogger(ClientPoolImpl.class);
29+
30+
private final int poolSize;
31+
private final Deque<C> clients;
32+
private final Class<? extends E> reconnectExc;
33+
private final Object signal = new Object();
34+
private final boolean retryByDefault;
35+
private final int maxRetries;
36+
37+
private volatile int currentSize;
38+
private boolean closed;
39+
40+
private static final int CONNECTION_RETRY_WAIT_PERIOD_MS = 1000;
41+
42+
public ClientPoolImpl(int poolSize, Class<? extends E> reconnectExc, boolean retryByDefault) {
43+
this(poolSize, reconnectExc, retryByDefault, 1);
44+
}
45+
46+
public ClientPoolImpl(
47+
int poolSize,
48+
Class<? extends E> reconnectExc,
49+
boolean retryByDefault,
50+
int maxConnectionRetries) {
51+
this.poolSize = poolSize;
52+
this.reconnectExc = reconnectExc;
53+
this.clients = new ArrayDeque<>(poolSize);
54+
this.currentSize = 0;
55+
this.closed = false;
56+
this.retryByDefault = retryByDefault;
57+
this.maxRetries = maxConnectionRetries;
58+
}
59+
60+
@Override
61+
public <R> R run(Action<R, C, E> action) throws E, InterruptedException {
62+
return run(action, retryByDefault);
63+
}
64+
65+
@Override
66+
public <R> R run(Action<R, C, E> action, boolean retry) throws E, InterruptedException {
67+
C client = get();
68+
try {
69+
return action.run(client);
70+
} catch (Exception exc) {
71+
if (retry && isConnectionException(exc)) {
72+
int retryAttempts = 0;
73+
while (retryAttempts < maxRetries) {
74+
try {
75+
client = reconnect(client);
76+
return action.run(client);
77+
} catch (Exception e) {
78+
if (isConnectionException(e)) {
79+
retryAttempts++;
80+
Thread.sleep(CONNECTION_RETRY_WAIT_PERIOD_MS);
81+
} else {
82+
throw reconnectExc.cast(exc);
83+
}
84+
}
85+
}
86+
}
87+
88+
throw exc;
89+
} finally {
90+
release(client);
91+
}
92+
}
93+
94+
protected abstract C newClient();
95+
96+
protected abstract C reconnect(C client);
97+
98+
protected boolean isConnectionException(Exception exc) {
99+
return reconnectExc.isInstance(exc);
100+
}
101+
102+
protected abstract void close(C client);
103+
104+
@Override
105+
public void close() {
106+
this.closed = true;
107+
try {
108+
while (currentSize > 0) {
109+
if (!clients.isEmpty()) {
110+
synchronized (this) {
111+
if (!clients.isEmpty()) {
112+
C client = clients.removeFirst();
113+
close(client);
114+
currentSize -= 1;
115+
}
116+
}
117+
}
118+
if (clients.isEmpty() && currentSize > 0) {
119+
// wake every second in case this missed the signal
120+
synchronized (signal) {
121+
signal.wait(1000);
122+
}
123+
}
124+
}
125+
126+
} catch (InterruptedException e) {
127+
Thread.currentThread().interrupt();
128+
LOG.warn("Interrupted while shutting down pool. Some clients may not be closed.", e);
129+
}
130+
}
131+
132+
private C get() throws InterruptedException {
133+
Preconditions.checkState(!closed, "Cannot get a client from a closed pool");
134+
while (true) {
135+
if (!clients.isEmpty() || currentSize < poolSize) {
136+
synchronized (this) {
137+
if (!clients.isEmpty()) {
138+
return clients.removeFirst();
139+
} else if (currentSize < poolSize) {
140+
C client = newClient();
141+
currentSize += 1;
142+
return client;
143+
}
144+
}
145+
}
146+
synchronized (signal) {
147+
// wake every second in case this missed the signal
148+
signal.wait(1000);
149+
}
150+
}
151+
}
152+
153+
private void release(C client) {
154+
synchronized (this) {
155+
clients.addFirst(client);
156+
}
157+
synchronized (signal) {
158+
signal.notify();
159+
}
160+
}
161+
162+
@VisibleForTesting
163+
Deque<C> clients() {
164+
return clients;
165+
}
166+
167+
public int poolSize() {
168+
return poolSize;
169+
}
170+
171+
public boolean isClosed() {
172+
return closed;
173+
}
174+
}

0 commit comments

Comments
 (0)